From 1383200054ba70c7378c749c89f1b25b5d440522 Mon Sep 17 00:00:00 2001 From: Your Name Date: Tue, 7 Apr 2026 21:51:09 +0000 Subject: [PATCH] gcbackstop trigger --- packages/core/src/context/DESIGN.md | 12 ++--- packages/core/src/context/ir/projector.ts | 21 ++++++++- .../src/context/sidecar/SidecarLoader.test.ts | 1 - .../src/context/sidecar/orchestrator.test.ts | 1 - .../core/src/context/sidecar/orchestrator.ts | 45 ++++++++++++++++++- packages/core/src/context/sidecar/profiles.ts | 23 +++++----- packages/core/src/context/sidecar/schema.ts | 21 +-------- packages/core/src/context/sidecar/types.ts | 8 +--- .../system-tests/lifecycle.golden.test.ts | 1 - 9 files changed, 85 insertions(+), 48 deletions(-) diff --git a/packages/core/src/context/DESIGN.md b/packages/core/src/context/DESIGN.md index 8baaebc249..fc9baac5bd 100644 --- a/packages/core/src/context/DESIGN.md +++ b/packages/core/src/context/DESIGN.md @@ -21,6 +21,7 @@ The architecture is built upon seven core principles that distinguish it from th 5. **Pluggability:** `ContextProcessor`s are isolated plugins with typed schemas. They are registered via Dependency Injection and can be arranged into arbitrary pipelines. 6. **Debuggability:** A built-in `ContextTracer` tracks every step of the pipeline, providing full audit trails of exactly when, why, and how a message was altered. 7. **Testability:** Global state has been eliminated. The system uses strict Dependency Injection (`ProcessorRegistry`, `ContextEnvironment`, `ContextEventBus`), making every layer easily unit-testable. +8. **Orthogonality via Targets:** Processors do not implicitly scan the entire history graph. The `ContextManager` computes exact Deltas (e.g., new nodes just added, or specific nodes that just aged out of the retained buffer). Processors are sandboxed by the `EpisodeEditor` to only iterate over and mutate these specific `targetNodes`, ensuring surgical and highly efficient reductions. --- @@ -58,13 +59,14 @@ To understand how these pieces fit together, let's walk through the lifecycle of 4. **Registration:** The new `Episode` is added to the `ContextManager`'s pristine graph. ### Phase 2: Triggering the Pipelines -1. **Event Emission:** The `ContextManager` fires a `PristineHistoryUpdatedEvent` over the `ContextEventBus`. -2. **Orchestration:** The `PipelineOrchestrator` hears the event and evaluates its configured `PipelineDef`s. It finds a pipeline with the trigger `on_turn`. -3. **Execution:** The Orchestrator begins running the processors in that pipeline sequentially. If the pipeline is marked `execution: 'background'`, this happens asynchronously. +1. **Delta Generation:** The `ContextManager` receives the updated pristine graph. It diffs it against the previous state and extracts a Delta—the exact Set of new `IrNode` IDs. +2. **Event Emission:** The `ContextManager` fires a `ChunkReceivedEvent` (with the Delta targets) over the `ContextEventBus`. +3. **Orchestration:** The `PipelineOrchestrator` hears the event and evaluates its configured `PipelineDef`s. It finds a pipeline with the trigger `on_turn`. +4. **Execution:** The Orchestrator creates an `EpisodeEditor` heavily sandboxed to *only* allow access to the targeted Delta nodes, and begins running the processors in that pipeline sequentially. ### Phase 3: Processing & Safe Editing -1. **Processing:** A processor (e.g., `ToolMaskingProcessor`) receives the `EpisodeEditor`. It identifies a massive JSON payload in the tool execution. -2. **Editing:** Instead of deleting the JSON, it calls `editor.editEpisode()`. It creates a `MaskedVariant` containing a string summary of the JSON. +1. **Processing:** A processor (e.g., `ToolMaskingProcessor`) receives the `EpisodeEditor`. It iterates over `editor.targets` (ignoring the rest of the historical graph). It identifies a massive JSON payload in one of the new tool executions. +2. **Editing:** Instead of deleting the JSON, it calls `editor.editEpisode()`. It creates a `MaskedVariant` containing a string summary of the JSON. If it had attempted to edit a node outside its target Delta, the editor would have thrown an error. 3. **Auditing:** The editor automatically appends a record to the node's `IrMetadata.transformations` indicating that the `ToolMaskingProcessor` applied a `MASKED` action. ### Phase 4: Async Resolution diff --git a/packages/core/src/context/ir/projector.ts b/packages/core/src/context/ir/projector.ts index b98d494a90..2057a32d94 100644 --- a/packages/core/src/context/ir/projector.ts +++ b/packages/core/src/context/ir/projector.ts @@ -60,8 +60,24 @@ export class IrProjector { `Context Manager Synchronous Barrier triggered: View at ${currentTokens} tokens (limit: ${maxTokens}).`, ); - const processedEpisodes = await orchestrator.executePipeline( - 'Immediate Sanitization', + // Calculate exactly which nodes aged out of the retainedTokens budget to form our target delta + const agedOutNodes = new Set(); + let rollingTokens = 0; + // Start from newest and count backwards + for (let i = workingBuffer.length - 1; i >= 0; i--) { + const ep = workingBuffer[i]; + const epTokens = env.tokenCalculator.calculateEpisodeListTokens([ep]); + rollingTokens += epTokens; + if (rollingTokens > sidecar.budget.retainedTokens) { + agedOutNodes.add(ep.id); + agedOutNodes.add(ep.trigger.id); + for (const step of ep.steps) agedOutNodes.add(step.id); + if (ep.yield) agedOutNodes.add(ep.yield.id); + } + } + + const processedEpisodes = await orchestrator.executeTriggerSync( + 'gc_backstop', workingBuffer, { currentTokens, @@ -70,6 +86,7 @@ export class IrProjector { deficitTokens: Math.max(0, currentTokens - sidecar.budget.maxTokens), protectedEpisodeIds: protectedIds, isBudgetSatisfied: currentTokens <= sidecar.budget.maxTokens, + targetNodeIds: agedOutNodes, }, ); diff --git a/packages/core/src/context/sidecar/SidecarLoader.test.ts b/packages/core/src/context/sidecar/SidecarLoader.test.ts index 88add76c20..941400e127 100644 --- a/packages/core/src/context/sidecar/SidecarLoader.test.ts +++ b/packages/core/src/context/sidecar/SidecarLoader.test.ts @@ -47,7 +47,6 @@ describe('SidecarLoader (Fake FS)', () => { it('returns parsed config if file is valid', () => { const validConfig = { budget: { retainedTokens: 1000, maxTokens: 2000 }, - gcBackstop: { strategy: 'truncate', target: 'max' }, pipelines: [], }; fileSystem.setFile('/path/to/sidecar.json', JSON.stringify(validConfig)); diff --git a/packages/core/src/context/sidecar/orchestrator.test.ts b/packages/core/src/context/sidecar/orchestrator.test.ts index 9a3e2213a1..d0fb1acffb 100644 --- a/packages/core/src/context/sidecar/orchestrator.test.ts +++ b/packages/core/src/context/sidecar/orchestrator.test.ts @@ -111,7 +111,6 @@ describe('PipelineOrchestrator (Component)', () => { const createConfig = (pipelines: PipelineDef[]): SidecarConfig => ({ budget: { maxTokens: 100, retainedTokens: 50 }, - gcBackstop: { strategy: 'truncate', target: 'max' }, pipelines, }); diff --git a/packages/core/src/context/sidecar/orchestrator.ts b/packages/core/src/context/sidecar/orchestrator.ts index 89795fc2e2..3241952329 100644 --- a/packages/core/src/context/sidecar/orchestrator.ts +++ b/packages/core/src/context/sidecar/orchestrator.ts @@ -6,7 +6,7 @@ import type { Episode } from '../ir/types.js'; import type { ContextProcessor, ContextAccountingState } from '../pipeline.js'; -import type { SidecarConfig, PipelineDef } from './types.js'; +import type { SidecarConfig, PipelineDef, PipelineTrigger } from './types.js'; import type { ContextEnvironment, ContextEventBus, @@ -111,6 +111,49 @@ export class PipelineOrchestrator { * Executes a pipeline asynchronously in the background. This is the "Eventual Consistency" path. * When the pipeline resolves, it emits a VariantReady event to cache the new graph. */ + /** + * Synchronously executes all pipelines configured with a specific trigger. + * Forces them to run in a blocking manner regardless of their 'execution' setting + * (useful for emergency backstops right before sending a prompt). + */ + async executeTriggerSync( + trigger: PipelineTrigger, + episodes: Episode[], + state: ContextAccountingState, + ): Promise { + let currentEpisodes = [...episodes]; + const pipelines = this.config.pipelines.filter((p) => p.triggers.includes(trigger)); + + for (const pipeline of pipelines) { + this.tracer.logEvent( + 'Orchestrator', + `Triggering synchronous emergency pipeline via [${trigger}]: ${pipeline.name}`, + ); + + for (let i = 0; i < pipeline.processors.length; i++) { + const procDef = pipeline.processors[i]; + const processor = this.instantiatedProcessors.get(procDef.processorId); + if (!processor) continue; + + try { + this.tracer.logEvent( + 'Orchestrator', + `Executing processor: ${procDef.processorId}`, + ); + const editor = new EpisodeEditor(currentEpisodes, state.targetNodeIds); + await processor.process(editor, state); + currentEpisodes = editor.getFinalEpisodes(); + } catch (error) { + debugLogger.error( + `Pipeline ${pipeline.name} failed synchronously at ${procDef.processorId}:`, + error, + ); + } + } + } + return currentEpisodes; + } + /** * Executes a pipeline based on its configured execution strategy ('blocking' or 'background'). */ diff --git a/packages/core/src/context/sidecar/profiles.ts b/packages/core/src/context/sidecar/profiles.ts index f5012319f7..8a40e939b3 100644 --- a/packages/core/src/context/sidecar/profiles.ts +++ b/packages/core/src/context/sidecar/profiles.ts @@ -15,11 +15,6 @@ export const defaultSidecarProfile: SidecarConfig = { retainedTokens: 65000, maxTokens: 150000, }, - gcBackstop: { - strategy: 'truncate', - target: 'incremental', - freeTokensTarget: 10000, - }, pipelines: [ { name: 'Immediate Sanitization', @@ -31,11 +26,6 @@ export const defaultSidecarProfile: SidecarConfig = { options: { stringLengthThresholdTokens: 8000 }, }, { processorId: 'BlobDegradationProcessor', options: {} }, - { - processorId: 'SemanticCompressionProcessor', - options: { nodeThresholdTokens: 5000 }, - }, - { processorId: 'EmergencyTruncationProcessor', options: {} }, ], }, { @@ -47,8 +37,21 @@ export const defaultSidecarProfile: SidecarConfig = { processorId: 'HistorySquashingProcessor', options: { maxTokensPerNode: 3000 }, }, + { + processorId: 'SemanticCompressionProcessor', + options: { nodeThresholdTokens: 5000 }, + }, { processorId: 'StateSnapshotProcessor', options: {} }, ], }, + { + name: 'Emergency Backstop', + triggers: ['gc_backstop'], + execution: 'blocking', + processors: [ + { processorId: 'StateSnapshotProcessor', options: {} }, + { processorId: 'EmergencyTruncationProcessor', options: {} }, + ], + }, ], }; diff --git a/packages/core/src/context/sidecar/schema.ts b/packages/core/src/context/sidecar/schema.ts index 6f1efd57e7..a7f6f185ff 100644 --- a/packages/core/src/context/sidecar/schema.ts +++ b/packages/core/src/context/sidecar/schema.ts @@ -13,7 +13,7 @@ export function getSidecarConfigSchema(registry: ProcessorRegistry) { title: 'SidecarConfig', description: 'The Data-Driven Schema for the Context Manager.', type: 'object', - required: ['budget', 'gcBackstop', 'pipelines'], + required: ['budget', 'pipelines'], properties: { budget: { type: 'object', @@ -32,25 +32,6 @@ export function getSidecarConfigSchema(registry: ProcessorRegistry) { }, }, }, - gcBackstop: { - type: 'object', - description: - "Defines what happens when the pipeline fails to compress under 'maxTokens'", - required: ['strategy', 'target'], - properties: { - strategy: { - type: 'string', - enum: ['truncate', 'compress', 'rollingSummarizer'], - }, - target: { - type: 'string', - enum: ['incremental', 'freeNTokens', 'max'], - }, - freeTokensTarget: { - type: 'number', - }, - }, - }, pipelines: { type: 'array', description: 'The execution graphs for context manipulation.', diff --git a/packages/core/src/context/sidecar/types.ts b/packages/core/src/context/sidecar/types.ts index 19e7a4f74a..ab238bc863 100644 --- a/packages/core/src/context/sidecar/types.ts +++ b/packages/core/src/context/sidecar/types.ts @@ -36,6 +36,7 @@ export type PipelineTrigger = | 'on_turn' | 'post_turn' | 'budget_exceeded' + | 'gc_backstop' | { type: 'timer'; intervalMs: number }; export interface PipelineDef { @@ -55,13 +56,6 @@ export interface SidecarConfig { maxTokens: number; }; - /** Defines what happens when the pipeline fails to compress under 'maxTokens' */ - gcBackstop: { - strategy: 'truncate' | 'compress' | 'rollingSummarizer'; - target: 'incremental' | 'freeNTokens' | 'max'; - freeTokensTarget?: number; - }; - /** The execution graphs for context manipulation */ pipelines: PipelineDef[]; } diff --git a/packages/core/src/context/system-tests/lifecycle.golden.test.ts b/packages/core/src/context/system-tests/lifecycle.golden.test.ts index fb8dc5b28a..2efe55d061 100644 --- a/packages/core/src/context/system-tests/lifecycle.golden.test.ts +++ b/packages/core/src/context/system-tests/lifecycle.golden.test.ts @@ -33,7 +33,6 @@ describe('System Lifecycle Golden Tests', () => { const getAggressiveConfig = (): SidecarConfig => ({ budget: { maxTokens: 4000, retainedTokens: 2000 }, // Extremely tight limits - gcBackstop: { strategy: 'truncate', target: 'max' }, pipelines: [ { name: 'Pressure Relief', // Emits from eventBus 'budget_exceeded'