MediaMetz/gemini-cli
1
0
Fork 0
mirror of https://github.com/google-gemini/gemini-cli.git synced 2026-07-16 04:50:29 -07:00
Code Issues Packages Projects Releases Wiki Activity

hmm speculative

Browse Source
This commit is contained in:
Your Name
2026-04-08 20:27:00 +00:00
parent 57f13a196e
commit 42022279bb
12 changed files with 321 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/context/sidecar/orchestrator.test.ts
+37 -21
View File
@@ -26,7 +26,13 @@ class DummySyncProcessor implements ContextProcessor {
readonly name = 'DummySync';
readonly id = 'DummySync';
readonly options = {};
async process() { return []; }
async process(args: any) {
const newTargets = [...args.targets];
if (newTargets.length > 0) {
newTargets[0] = { ...newTargets[0], dummyModified: true };
}
return newTargets;
}
}
class DummyAsyncProcessor implements ContextProcessor {
@@ -37,7 +43,13 @@ class DummyAsyncProcessor implements ContextProcessor {
readonly name = 'DummyAsync';
readonly id = 'DummyAsync';
readonly options = {};
async process() { return []; }
async process(args: any) {
const newTargets = [...args.targets];
if (newTargets.length > 0) {
newTargets[0] = { ...newTargets[0], dummyModified: true };
}
return newTargets;
}
}
class ThrowingProcessor implements ContextProcessor {
@@ -95,9 +107,9 @@ describe('PipelineOrchestrator (Component)', () => {
it('instantiates processors from the registry on initialization', () => {
const config = createConfig([
{
name: 'Sync',
name: 'ThrowPipe',
execution: 'blocking',
triggers: [],
triggers: ['new_message'],
processors: [
{ processorId: 'DummySyncProcessor' } as unknown as ProcessorConfig,
],
@@ -120,9 +132,9 @@ describe('PipelineOrchestrator (Component)', () => {
it('throws an error if a config requests an unknown processor', () => {
const config = createConfig([
{
name: 'Bad',
name: 'ThrowPipe',
execution: 'blocking',
triggers: [],
triggers: ['new_message'],
processors: [
{ processorId: 'DoesNotExist' } as unknown as ProcessorConfig,
],
@@ -140,7 +152,7 @@ describe('PipelineOrchestrator (Component)', () => {
{
name: 'SyncPipe',
execution: 'blocking',
triggers: [],
triggers: ['new_message'],
processors: [
{ processorId: 'DummySyncProcessor' } as unknown as ProcessorConfig,
],
@@ -154,19 +166,19 @@ describe('PipelineOrchestrator (Component)', () => {
registry,
);
const ship = [createDummyNode('1', 'USER_PROMPT')];
const state = createDummyState(false);
const ship = [createDummyNode('not-protected-ep', 'USER_PROMPT', 100, undefined, 'not-protected-id')];
const state = createDummyState(false, 0, new Set());
const result = await orchestrator.executeTriggerSync(
'new_message',
ship,
new Set(ship.map(s => s.id)),
new Set(ship.map((s) => s.id)),
state,
);
expect(result).toHaveLength(1);
expect(
(result[0] as unknown as { dummyModified: boolean }).dummyModified,
(result[0] as unknown as { dummyModified?: boolean }).dummyModified,
).toBe(true);
});
@@ -175,7 +187,7 @@ describe('PipelineOrchestrator (Component)', () => {
{
name: 'AsyncPipe',
execution: 'background',
triggers: [],
triggers: ['new_message'],
processors: [
{ processorId: 'DummyAsyncProcessor' } as unknown as ProcessorConfig,
],
@@ -189,14 +201,14 @@ describe('PipelineOrchestrator (Component)', () => {
registry,
);
const ship = [createDummyNode('1', 'USER_PROMPT')];
const state = createDummyState(false);
const ship = [createDummyNode('not-protected-ep', 'USER_PROMPT', 100, undefined, 'not-protected-id')];
const state = createDummyState(false, 0, new Set());
// This should resolve immediately with the UNMODIFIED array because execution is background
const result = await orchestrator.executeTriggerSync(
'new_message',
ship,
new Set(ship.map(s => s.id)),
new Set(ship.map((s) => s.id)),
state,
);
@@ -212,9 +224,9 @@ describe('PipelineOrchestrator (Component)', () => {
it('gracefully handles and swallows processor errors in synchronous pipelines', async () => {
const config = createConfig([
{
name: 'ThrowingPipe',
name: 'ThrowPipe',
execution: 'blocking',
triggers: [],
triggers: ['new_message'],
processors: [
{ processorId: 'ThrowingProcessor' } as unknown as ProcessorConfig,
],
@@ -228,14 +240,14 @@ describe('PipelineOrchestrator (Component)', () => {
registry,
);
const ship = [createDummyNode('1', 'USER_PROMPT')];
const state = createDummyState(false);
const ship = [createDummyNode('not-protected-ep', 'USER_PROMPT', 100, undefined, 'not-protected-id')];
const state = createDummyState(false, 0, new Set());
// It should not throw! It should swallow the error and return the unmodified array.
const result = await orchestrator.executeTriggerSync(
'new_message',
ship,
new Set(ship.map(s => s.id)),
new Set(ship.map((s) => s.id)),
state,
);
@@ -267,7 +279,11 @@ describe('PipelineOrchestrator (Component)', () => {
const ship = [createDummyNode('1', 'USER_PROMPT')];
// Emit the trigger
eventBus.emitConsolidationNeeded({ ship, targetDeficit: 100, targetNodeIds: new Set() });
eventBus.emitConsolidationNeeded({
ship,
targetDeficit: 100,
targetNodeIds: new Set(),
});
expect(executeSpy).toHaveBeenCalled();
});
packages/core/src/context/sidecar/orchestrator.ts
+86 -41
View File
@@ -5,7 +5,11 @@
*/
import type { ConcreteNode } from '../ir/types.js';
import type { ContextProcessor, ContextWorker, ContextAccountingState } from '../pipeline.js';
import type {
ContextProcessor,
ContextWorker,
ContextAccountingState,
} from '../pipeline.js';
import type { SidecarConfig, PipelineDef, PipelineTrigger } from './types.js';
import type {
ContextEnvironment,
@@ -33,12 +37,28 @@ export class PipelineOrchestrator {
this.setupTriggers();
}
private isNodeAllowed(
node: import('../ir/types.js').ConcreteNode,
triggerTargets: ReadonlySet<string>,
state: ContextAccountingState,
): boolean {
return (
triggerTargets.has(node.id) &&
!state.protectedLogicalIds.has(node.id) &&
(!node.logicalParentId ||
!state.protectedLogicalIds.has(node.logicalParentId))
);
}
private instantiateProcessors() {
for (const pipeline of this.config.pipelines) {
for (const procDef of pipeline.processors) {
if (!this.instantiatedProcessors.has(procDef.processorId)) {
const factory = this.registry.get(procDef.processorId);
const instance = factory.create(this.env, procDef.options || {}) as ContextProcessor;
const instance = factory.create(
this.env,
procDef.options || {},
) as ContextProcessor;
this.instantiatedProcessors.set(procDef.processorId, instance);
}
}
@@ -50,7 +70,10 @@ export class PipelineOrchestrator {
for (const workerDef of this.config.workers) {
if (!this.instantiatedWorkers.has(workerDef.workerId)) {
const factory = this.registry.get(workerDef.workerId);
const instance = factory.create(this.env, workerDef.options || {}) as unknown as ContextWorker;
const instance = factory.create(
this.env,
workerDef.options || {},
) as unknown as ContextWorker;
this.instantiatedWorkers.set(workerDef.workerId, instance);
}
}
@@ -75,7 +98,12 @@ export class PipelineOrchestrator {
deficitTokens: event.targetDeficit,
protectedLogicalIds: new Set(),
};
void this.executePipelineAsync(pipeline, [], event.targetNodeIds, state);
void this.executePipelineAsync(
pipeline,
[],
event.targetNodeIds,
state,
);
});
} else if (trigger === 'new_message') {
this.eventBus.onChunkReceived((event) => {
@@ -87,7 +115,12 @@ export class PipelineOrchestrator {
deficitTokens: 0,
protectedLogicalIds: new Set(),
};
void this.executePipelineAsync(pipeline, [], event.targetNodeIds, state);
void this.executePipelineAsync(
pipeline,
[],
event.targetNodeIds,
state,
);
});
}
}
@@ -95,16 +128,18 @@ export class PipelineOrchestrator {
// 2. Worker Triggers (onNodesAdded is roughly onChunkReceived for now)
this.eventBus.onChunkReceived((event) => {
// Fire all workers that care about new nodes
for (const worker of this.instantiatedWorkers.values()) {
if (worker.triggers.onNodesAdded) {
const inboxSnapshot = new InboxSnapshotImpl(this.env.inbox.getMessages() || []);
// Fire and forget
worker.execute({ targets: [], inbox: inboxSnapshot }).catch(e => {
debugLogger.error(`Worker ${worker.name} failed onNodesAdded:`, e);
});
}
}
// Fire all workers that care about new nodes
for (const worker of this.instantiatedWorkers.values()) {
if (worker.triggers.onNodesAdded) {
const inboxSnapshot = new InboxSnapshotImpl(
this.env.inbox?.getMessages() || [],
);
// Fire and forget
worker.execute({ targets: [], inbox: inboxSnapshot }).catch((e) => {
debugLogger.error(`Worker ${worker.name} failed onNodesAdded:`, e);
});
}
}
});
// We don't have a formal event bus for inbox publish yet, but we will soon.
@@ -123,8 +158,8 @@ export class PipelineOrchestrator {
returnedNodes: ReadonlyArray<ConcreteNode>,
): ReadonlyArray<ConcreteNode> {
const mutableShip = [...ship];
const targetSet = new Set(targets.map(n => n.id));
const returnedMap = new Map(returnedNodes.map(n => [n.id, n]));
const targetSet = new Set(targets.map((n) => n.id));
const returnedMap = new Map(returnedNodes.map((n) => [n.id, n]));
const removedIds = new Set<string>();
const newNodes: ConcreteNode[] = [];
@@ -146,7 +181,7 @@ export class PipelineOrchestrator {
}
if (removedIds.size === 0 && newNodes.length === 0) {
return ship;
return ship;
}
let earliestRemovalIdx = mutableShip.length;
@@ -174,10 +209,14 @@ export class PipelineOrchestrator {
state: ContextAccountingState,
): Promise<ReadonlyArray<ConcreteNode>> {
let currentShip = ship;
const pipelines = this.config.pipelines.filter((p) => p.triggers.includes(trigger));
const pipelines = this.config.pipelines.filter((p) =>
p.triggers.includes(trigger),
);
// Freeze the inbox for this pipeline run
const inboxSnapshot = new InboxSnapshotImpl(this.env.inbox.getMessages() || []);
const inboxSnapshot = new InboxSnapshotImpl(
this.env.inbox?.getMessages() || [],
);
for (const pipeline of pipelines) {
for (const procDef of pipeline.processors) {
@@ -190,20 +229,22 @@ export class PipelineOrchestrator {
`Executing processor synchronously: ${procDef.processorId}`,
);
const allowedTargets = currentShip.filter(n =>
triggerTargets.has(n.id) &&
(!n.logicalParentId || !state.protectedLogicalIds.has(n.logicalParentId))
const allowedTargets = currentShip.filter((n) =>
this.isNodeAllowed(n, triggerTargets, state),
);
const returnedNodes = await processor.process({
buffer: {} as any, // TODO: Implement ContextWorkingBuffer fully
targets: allowedTargets,
state,
inbox: inboxSnapshot,
});
currentShip = this.applyProcessorDiff(currentShip, allowedTargets, returnedNodes);
currentShip = this.applyProcessorDiff(
currentShip,
allowedTargets,
returnedNodes,
);
} catch (error) {
debugLogger.error(
`Synchronous processor ${procDef.processorId} failed:`,
@@ -214,7 +255,7 @@ export class PipelineOrchestrator {
}
// Success! Drain consumed messages
this.env.inbox.drainConsumed(inboxSnapshot.getConsumedIds());
this.env.inbox?.drainConsumed(inboxSnapshot.getConsumedIds());
return currentShip;
}
@@ -232,7 +273,9 @@ export class PipelineOrchestrator {
if (!ship || ship.length === 0) return;
let currentShip = ship;
const inboxSnapshot = new InboxSnapshotImpl(this.env.inbox.getMessages() || []);
const inboxSnapshot = new InboxSnapshotImpl(
this.env.inbox?.getMessages() || [],
);
for (const procDef of pipeline.processors) {
const processor = this.instantiatedProcessors.get(procDef.processorId);
@@ -244,29 +287,31 @@ export class PipelineOrchestrator {
`Executing processor: ${procDef.processorId} (async)`,
);
const allowedTargets = currentShip.filter(n =>
triggerTargets.has(n.id) &&
(!n.logicalParentId || !state.protectedLogicalIds.has(n.logicalParentId))
const allowedTargets = currentShip.filter((n) =>
this.isNodeAllowed(n, triggerTargets, state),
);
const returnedNodes = await processor.process({
buffer: {} as any,
targets: allowedTargets,
state,
inbox: inboxSnapshot,
buffer: {} as any,
targets: allowedTargets,
state,
inbox: inboxSnapshot,
});
currentShip = this.applyProcessorDiff(currentShip, allowedTargets, returnedNodes);
currentShip = this.applyProcessorDiff(
currentShip,
allowedTargets,
returnedNodes,
);
} catch (error) {
debugLogger.error(
`Pipeline ${pipeline.name} failed async at ${procDef.processorId}:`,
error,
);
return;
return;
}
}
this.env.inbox.drainConsumed(inboxSnapshot.getConsumedIds());
this.env.inbox?.drainConsumed(inboxSnapshot.getConsumedIds());
}
}
packages/core/src/context/sidecar/testProfile.ts
+18
View File
@@ -0,0 +1,18 @@
import type { SidecarConfig } from './types.js';
export const testTruncateProfile: SidecarConfig = {
budget: {
retainedTokens: 65000,
maxTokens: 150000,
},
pipelines: [
{
name: 'Emergency Backstop (Truncate Only)',
triggers: ['gc_backstop', 'retained_exceeded'],
execution: 'blocking',
processors: [
{ processorId: 'EmergencyTruncationProcessor', options: {} },
],
},
],
};