From e601563652b9c5df520a85e72c7887478d813d5f Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 6 Apr 2026 18:46:21 +0000 Subject: [PATCH] working in progress --- packages/core/src/context/contextManager.ts | 269 +++--------------- packages/core/src/context/historyObserver.ts | 75 +++++ packages/core/src/context/ir/graphUtils.ts | 160 +++++++++++ packages/core/src/context/ir/projector.ts | 36 +++ .../emergencyTruncationProcessor.ts | 48 ++++ packages/core/src/context/sidecar/profiles.ts | 3 +- packages/core/src/context/sidecar/registry.ts | 1 + .../context/utils/contextTokenCalculator.ts | 13 + 8 files changed, 368 insertions(+), 237 deletions(-) create mode 100644 packages/core/src/context/historyObserver.ts create mode 100644 packages/core/src/context/ir/graphUtils.ts create mode 100644 packages/core/src/context/ir/projector.ts create mode 100644 packages/core/src/context/processors/emergencyTruncationProcessor.ts diff --git a/packages/core/src/context/contextManager.ts b/packages/core/src/context/contextManager.ts index 5a005469c7..2627e63174 100644 --- a/packages/core/src/context/contextManager.ts +++ b/packages/core/src/context/contextManager.ts @@ -8,7 +8,6 @@ import type { Content } from '@google/genai'; import type { AgentChatHistory } from '../core/agentChatHistory.js'; import { debugLogger } from '../utils/debugLogger.js'; -import { IrMapper } from './ir/mapper.js'; import type { Episode } from './ir/types.js'; import { ContextEventBus } from './eventBus.js'; @@ -21,6 +20,9 @@ import type { ContextEnvironment } from './sidecar/environment.js'; import type { SidecarConfig } from './sidecar/types.js'; import { ProcessorRegistry } from './sidecar/registry.js'; import { PipelineOrchestrator } from './sidecar/orchestrator.js'; +import { HistoryObserver } from './historyObserver.js'; +import { calculateEpisodeListTokens } from './utils/contextTokenCalculator.js'; +import { generateWorkingBufferView } from './ir/graphUtils.js'; import { ToolMaskingProcessor } from './processors/toolMaskingProcessor.js'; @@ -28,6 +30,9 @@ import { BlobDegradationProcessor } from './processors/blobDegradationProcessor. import { SemanticCompressionProcessor } from './processors/semanticCompressionProcessor.js'; import { HistorySquashingProcessor } from './processors/historySquashingProcessor.js'; import { StateSnapshotProcessor } from './processors/stateSnapshotProcessor.js'; +import { EmergencyTruncationProcessor } from './processors/emergencyTruncationProcessor.js'; + +import { IrProjector } from './ir/projector.js'; export class ContextManager { @@ -35,13 +40,13 @@ export class ContextManager { // The stateful, pristine Episodic Intermediate Representation graph. // This allows the agent to remember and summarize continuously without losing data across turns. private pristineEpisodes: Episode[] = []; - private unsubscribeHistory?: () => void; private readonly eventBus: ContextEventBus; // Internal sub-components // Synchronous processors are instantiated but effectively used as singletons within this class private orchestrator: PipelineOrchestrator; + private historyObserver?: HistoryObserver; @@ -59,6 +64,7 @@ export class ContextManager { ProcessorRegistry.register({ id: 'SemanticCompressionProcessor', create: (env, opts) => new SemanticCompressionProcessor(env, opts as any) }); ProcessorRegistry.register({ id: 'HistorySquashingProcessor', create: (env, opts) => new HistorySquashingProcessor(env, opts as any) }); ProcessorRegistry.register({ id: 'StateSnapshotProcessor', create: (env, opts) => StateSnapshotProcessor.create(env, opts as any) }); + ProcessorRegistry.register({ id: 'EmergencyTruncationProcessor', create: (env, opts) => EmergencyTruncationProcessor.create(env, opts as any) }); this.orchestrator = new PipelineOrchestrator(this.sidecar, this.env, this.eventBus, this.tracer); @@ -86,8 +92,8 @@ export class ContextManager { */ shutdown() { this.orchestrator.shutdown(); - if (this.unsubscribeHistory) { - this.unsubscribeHistory(); + if (this.historyObserver) { + this.historyObserver.stop(); } } @@ -96,49 +102,21 @@ export class ContextManager { * converting them seamlessly into pristine Episodes. */ subscribeToHistory(chatHistory: AgentChatHistory) { - if (this.unsubscribeHistory) { - this.unsubscribeHistory(); + if (this.historyObserver) { + this.historyObserver.stop(); } - this.unsubscribeHistory = chatHistory.subscribe((event) => { - // Rebuild the pristine IR graph from the full source history on every change. - // We must map the FULL array at once because IrMapper groups adjacent - // function calls and responses into unified Episodes. Pushing messages - // individually would shatter these episodic boundaries. - this.pristineEpisodes = IrMapper.toIr(chatHistory.get()); - this.tracer.logEvent('ContextManager', 'Rebuilt pristine graph from chat history update', { episodeCount: this.pristineEpisodes.length }); - this.checkTriggers(); - }); - } + this.historyObserver = new HistoryObserver( + chatHistory, + this.eventBus, + this.tracer, + this.sidecar, + (episodes) => { this.pristineEpisodes = episodes; }, + () => this.getWorkingBufferView(), + (episodes) => calculateEpisodeListTokens(episodes) + ); - private checkTriggers() { - if (!this.sidecar.budget) return; - - const mngConfig = this.sidecar; - - // Calculate tokens based on the *Working Buffer View*, not the raw pristine log. - // This solves Bug 2: The View shrinks when variants are applied, preventing infinite GC loops. - const workingBuffer = this.getWorkingBufferView(); - const currentTokens = this.calculateIrTokens(workingBuffer); - - this.tracer.logEvent('ContextManager', 'Evaluated triggers', { currentTokens, retainedTokens: mngConfig.budget.retainedTokens }); - - // 1. Eager Compute Trigger (Continuous Streaming) - // Broadcast the full pristine log to the async workers so they can proactively summarize partial massive files. - this.eventBus.emitChunkReceived({ episodes: this.pristineEpisodes }); - - // 2. The Ship of Theseus Trigger (retainedTokens crossed) - // If we exceed 65k, tell the background processors to opportunistically synthesize the oldest nodes. - if (currentTokens > mngConfig.budget.retainedTokens) { - const deficit = currentTokens - mngConfig.budget.retainedTokens; - this.tracer.logEvent('ContextManager', 'Budget crossed. Emitting ConsolidationNeeded', { deficit }); - console.log('EMITTING CONSOLIDATION. Buffer:', workingBuffer.length, 'Deficit:', deficit); - this.eventBus.emitConsolidationNeeded({ - - episodes: workingBuffer, // Pass the working buffer so they know what still needs compression - targetDeficit: deficit, - }); - } + this.historyObserver.start(); } /** @@ -148,163 +126,12 @@ export class ContextManager { * (snapshot > summary > masked) instead of the raw text. * Handles N-to-1 variant skipping automatically. */ - /** - * Applies the data-driven Sidecar configuration graphs. - * Splits the episodes into the 'retained' and 'normal' ranges, - * runs their respective processor pipelines sequentially, and recombines them. - */ - private async applyProcessorGraphs(episodes: Episode[]): Promise { - const mngConfig = this.sidecar; - - const protectedIds = new Set(); - // We must protect the System Episode, which is always index 0 of pristineEpisodes. - if (this.pristineEpisodes.length > 0) { - protectedIds.add(this.pristineEpisodes[0].id); // Structural invariant - } - - let currentTokens = this.calculateIrTokens(episodes); - - return this.orchestrator.executePipelineForking('Immediate Sanitization', episodes, { - currentTokens: currentTokens, - maxTokens: mngConfig.budget.maxTokens, - retainedTokens: mngConfig.budget.retainedTokens, - deficitTokens: Math.max(0, currentTokens - mngConfig.budget.maxTokens), - protectedEpisodeIds: protectedIds, - isBudgetSatisfied: currentTokens <= mngConfig.budget.maxTokens, - }); - } - public getWorkingBufferView(): Episode[] { - const mngConfig = this.sidecar; - const retainedTokens = mngConfig.budget.retainedTokens; - - let currentEpisodes: Episode[] = []; - let rollingTokens = 0; - const skippedIds = new Set(); - this.tracer.logEvent('ViewGenerator', 'Generating Working Buffer View'); - - for (let i = this.pristineEpisodes.length - 1; i >= 0; i--) { - const ep = this.pristineEpisodes[i]; - - // If this episode was already replaced by an N-to-1 Snapshot injected earlier in the sweep, skip it entirely! - // This solves Bug 1 (Duplicate Projection). - if (skippedIds.has(ep.id)) { - this.tracer.logEvent('ViewGenerator', `Skipping episode [${ep.id}] due to N-to-1 replacement.`); - continue; - } - - let projectedEp = { - ...ep, - trigger: { - ...ep.trigger, - metadata: { - ...ep.trigger.metadata, - transformations: [...ep.trigger.metadata.transformations], - }, - semanticParts: - ep.trigger.type === 'USER_PROMPT' - ? [...ep.trigger.semanticParts.map((sp) => ({ ...sp }))] - : undefined, - } as any, - steps: ep.steps.map( - (step) => - ({ - ...step, - metadata: { - ...step.metadata, - transformations: [...step.metadata.transformations], - }, - }) as any, - ), - yield: ep.yield - ? { - ...ep.yield, - metadata: { - ...ep.yield.metadata, - transformations: [...ep.yield.metadata.transformations], - }, - } - : undefined, - }; - - const epTokens = this.calculateIrTokens([projectedEp]); - - if (ep.variants) { console.log('Checking variants for', ep.id, 'rollingTokens:', rollingTokens, 'retained:', retainedTokens); } - if (rollingTokens > retainedTokens && ep.variants) { - console.log('EVALUATING VARIANTS FOR', ep.id); - const snapshot = ep.variants['snapshot']; - const summary = ep.variants['summary']; - const masked = ep.variants['masked']; - - if ( - snapshot && - snapshot.status === 'ready' && - snapshot.type === 'snapshot' - ) { - projectedEp = snapshot.episode as any; - // Mark all the episodes this snapshot covers to be skipped by the backwards sweep. - for (const id of snapshot.replacedEpisodeIds) { - skippedIds.add(id); - } - this.tracer.logEvent('ViewGenerator', `Episode [${ep.id}] has SnapshotVariant. Selecting variant over raw text. Added [${snapshot.replacedEpisodeIds.join(',')}] to skippedIds.`); - debugLogger.log( - `Opportunistically swapped Episodes [${snapshot.replacedEpisodeIds.join(', ')}] for pre-computed Snapshot variant.`, - ); - } else if ( - summary && - summary.status === 'ready' && - summary.type === 'summary' - ) { - projectedEp.steps = [ - { - id: ep.id + '-summary', - type: 'AGENT_THOUGHT', - text: summary.text, - metadata: { - originalTokens: epTokens, - currentTokens: summary.recoveredTokens || 50, - transformations: [ - { - processorName: 'AsyncSemanticCompressor', - action: 'SUMMARIZED', - timestamp: Date.now(), - }, - ], - }, - }, - ] as any; - projectedEp.yield = undefined; - this.tracer.logEvent('ViewGenerator', `Episode [${ep.id}] has SummaryVariant. Selecting variant over raw text.`); - debugLogger.log( - `Opportunistically swapped Episode ${ep.id} for pre-computed Summary variant.`, - ); - } else if ( - masked && - masked.status === 'ready' && - masked.type === 'masked' - ) { - if ( - projectedEp.trigger.type === 'USER_PROMPT' && - projectedEp.trigger.semanticParts.length > 0 - ) { - projectedEp.trigger.semanticParts[0].presentation = { - text: masked.text, - tokens: masked.recoveredTokens || 10, - }; - } - this.tracer.logEvent('ViewGenerator', `Episode [${ep.id}] has MaskedVariant. Selecting variant over raw text.`); - debugLogger.log( - `Opportunistically swapped Episode ${ep.id} for pre-computed Masked variant.`, - ); - } - } - - currentEpisodes.unshift(projectedEp); - rollingTokens += this.calculateIrTokens([projectedEp]); - } - - - return currentEpisodes; + return generateWorkingBufferView( + this.pristineEpisodes, + this.sidecar.budget.retainedTokens, + this.tracer + ); } /** @@ -313,7 +140,7 @@ export class ContextManager { */ async projectCompressedHistory(): Promise { if (!this.sidecar.budget) { - return this._projectAndDump(IrMapper.fromIr(this.pristineEpisodes)); + return IrProjector.projectAndDump(this.pristineEpisodes, this.env); } const mngConfig = this.sidecar; @@ -323,14 +150,12 @@ export class ContextManager { // Get the dynamically computed Working Buffer View let currentEpisodes = this.getWorkingBufferView(); - currentEpisodes = await this.applyProcessorGraphs(currentEpisodes); - - let currentTokens = this.calculateIrTokens(currentEpisodes); + let currentTokens = calculateEpisodeListTokens(currentEpisodes); if (currentTokens <= maxTokens) { this.tracer.logEvent('ContextManager', `View is within maxTokens (${currentTokens} <= ${maxTokens}). Returning view.`); - return this._projectAndDump(IrMapper.fromIr(currentEpisodes)); + return IrProjector.projectAndDump(currentEpisodes, this.env); } this.tracer.logEvent('ContextManager', `View exceeds maxTokens (${currentTokens} > ${maxTokens}). Hitting Synchronous Pressure Barrier. Strategy: ${mngConfig.gcBackstop.strategy}`); @@ -365,7 +190,7 @@ export class ContextManager { for (const ep of currentEpisodes) { - const epTokens = this.calculateIrTokens([ep]); + const epTokens = calculateEpisodeListTokens([ep]); if (remainingTokens > targetTokens && ep.id !== protectedEpisodeId) { console.log('DROPPING EPISODE:', ep.id, 'rem:', remainingTokens, 'tgt:', targetTokens); @@ -389,40 +214,12 @@ export class ContextManager { } currentEpisodes = truncated; - const finalTokens = this.calculateIrTokens(currentEpisodes); + const finalTokens = calculateEpisodeListTokens(currentEpisodes); this.tracer.logEvent('ContextManager', `Finished projection. Final token count: ${finalTokens}.`); debugLogger.log( `Context Manager finished. Final actual token count: ${finalTokens}.`, ); - return this._projectAndDump(IrMapper.fromIr(currentEpisodes)); - } - - private async _projectAndDump(contents: Content[]): Promise { - if (process.env['GEMINI_DUMP_CONTEXT'] === 'true') { - try { - const fs = await import('node:fs/promises'); - const path = await import('node:path'); - const dumpPath = path.join(this.env.getTraceDir(), '.gemini', 'projected_context.json'); - await fs.mkdir(path.dirname(dumpPath), { recursive: true }); - await fs.writeFile(dumpPath, JSON.stringify(contents, null, 2), 'utf-8'); - debugLogger.log(`[Observability] Context successfully dumped to ${dumpPath}`); - } catch (e) { - debugLogger.error(`Failed to dump context: ${e}`); - } - } - return contents; - } - - private calculateIrTokens(episodes: Episode[]): number { - let tokens = 0; - for (const ep of episodes) { - if (ep.trigger) tokens += ep.trigger.metadata.currentTokens; - for (const step of ep.steps) { - tokens += step.metadata.currentTokens; - } - if (ep.yield) tokens += ep.yield.metadata.currentTokens; - } - return tokens; + return IrProjector.projectAndDump(currentEpisodes, this.env); } } diff --git a/packages/core/src/context/historyObserver.ts b/packages/core/src/context/historyObserver.ts new file mode 100644 index 0000000000..560dd11ebe --- /dev/null +++ b/packages/core/src/context/historyObserver.ts @@ -0,0 +1,75 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { AgentChatHistory, HistoryEvent } from '../core/agentChatHistory.js'; +import { IrMapper } from './ir/mapper.js'; +import type { ContextEventBus } from './eventBus.js'; +import type { ContextTracer } from './tracer.js'; +import type { SidecarConfig } from './sidecar/types.js'; +import type { Episode } from './ir/types.js'; + +/** + * Connects the raw AgentChatHistory to the ContextManager. + * It maps raw messages into Episodic Intermediate Representation (IR) + * and evaluates background triggers whenever history changes. + */ +export class HistoryObserver { + private unsubscribeHistory?: () => void; + + constructor( + private readonly chatHistory: AgentChatHistory, + private readonly eventBus: ContextEventBus, + private readonly tracer: ContextTracer, + private readonly sidecar: SidecarConfig, + private readonly onIrRebuilt: (episodes: Episode[]) => void, + private readonly computeWorkingBuffer: () => Episode[], + private readonly calculateIrTokens: (episodes: Episode[]) => number, + ) {} + + start() { + if (this.unsubscribeHistory) { + this.unsubscribeHistory(); + } + + this.unsubscribeHistory = this.chatHistory.subscribe((_event: HistoryEvent) => { + // Rebuild the pristine IR graph from the full source history on every change. + const pristineEpisodes = IrMapper.toIr(this.chatHistory.get()); + this.tracer.logEvent('HistoryObserver', 'Rebuilt pristine graph from chat history update', { episodeCount: pristineEpisodes.length }); + + this.onIrRebuilt(pristineEpisodes); + this.checkTriggers(pristineEpisodes); + }); + } + + stop() { + if (this.unsubscribeHistory) { + this.unsubscribeHistory(); + this.unsubscribeHistory = undefined; + } + } + + private checkTriggers(pristineEpisodes: Episode[]) { + if (!this.sidecar.budget) return; + + const workingBuffer = this.computeWorkingBuffer(); + const currentTokens = this.calculateIrTokens(workingBuffer); + + this.tracer.logEvent('HistoryObserver', 'Evaluated triggers', { currentTokens, retainedTokens: this.sidecar.budget.retainedTokens }); + + // 1. Eager Compute Trigger + this.eventBus.emitChunkReceived({ episodes: pristineEpisodes }); + + // 2. Budget Crossed Trigger + if (currentTokens > this.sidecar.budget.retainedTokens) { + const deficit = currentTokens - this.sidecar.budget.retainedTokens; + this.tracer.logEvent('HistoryObserver', 'Budget crossed. Emitting ConsolidationNeeded', { deficit }); + this.eventBus.emitConsolidationNeeded({ + episodes: workingBuffer, + targetDeficit: deficit, + }); + } + } +} diff --git a/packages/core/src/context/ir/graphUtils.ts b/packages/core/src/context/ir/graphUtils.ts new file mode 100644 index 0000000000..1f3edf707f --- /dev/null +++ b/packages/core/src/context/ir/graphUtils.ts @@ -0,0 +1,160 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { Episode } from './types.js'; +import type { ContextTracer } from '../tracer.js'; +import { debugLogger } from '../../utils/debugLogger.js'; +import { calculateEpisodeListTokens } from '../utils/contextTokenCalculator.js'; + +/** + * Generates a computed view of the pristine log. + * Sweeps backwards (newest to oldest), tracking rolling tokens. + * When rollingTokens > retainedTokens, it injects the "best" available ready variant + * (snapshot > summary > masked) instead of the raw text. + * Handles N-to-1 variant skipping automatically. + */ +export function generateWorkingBufferView( + pristineEpisodes: Episode[], + retainedTokens: number, + tracer: ContextTracer, +): Episode[] { + let currentEpisodes: Episode[] = []; + let rollingTokens = 0; + const skippedIds = new Set(); + tracer.logEvent('ViewGenerator', 'Generating Working Buffer View'); + + for (let i = pristineEpisodes.length - 1; i >= 0; i--) { + const ep = pristineEpisodes[i]; + + // If this episode was already replaced by an N-to-1 Snapshot injected earlier in the sweep, skip it entirely! + if (skippedIds.has(ep.id)) { + tracer.logEvent( + 'ViewGenerator', + `Skipping episode [${ep.id}] due to N-to-1 replacement.`, + ); + continue; + } + + let projectedEp = { + ...ep, + trigger: { + ...ep.trigger, + metadata: { + ...ep.trigger?.metadata, + transformations: [...(ep.trigger?.metadata?.transformations || [])], + }, + semanticParts: + ep.trigger?.type === 'USER_PROMPT' + ? [...(ep.trigger.semanticParts || []).map((sp) => ({ ...sp }))] + : undefined, + } as any, + steps: ep.steps.map( + (step) => + ({ + ...step, + metadata: { + ...step.metadata, + transformations: [...(step.metadata?.transformations || [])], + }, + }) as any, + ), + yield: ep.yield + ? { + ...ep.yield, + metadata: { + ...ep.yield.metadata, + transformations: [...(ep.yield.metadata?.transformations || [])], + }, + } + : undefined, + }; + + const epTokens = calculateEpisodeListTokens([projectedEp]); + + if (rollingTokens > retainedTokens && ep.variants) { + const snapshot = ep.variants['snapshot']; + const summary = ep.variants['summary']; + const masked = ep.variants['masked']; + + if ( + snapshot && + snapshot.status === 'ready' && + snapshot.type === 'snapshot' + ) { + projectedEp = snapshot.episode as any; + // Mark all the episodes this snapshot covers to be skipped by the backwards sweep. + for (const id of snapshot.replacedEpisodeIds) { + skippedIds.add(id); + } + tracer.logEvent( + 'ViewGenerator', + `Episode [${ep.id}] has SnapshotVariant. Selecting variant over raw text. Added [${snapshot.replacedEpisodeIds.join(',')}] to skippedIds.`, + ); + debugLogger.log( + `Opportunistically swapped Episodes [${snapshot.replacedEpisodeIds.join(', ')}] for pre-computed Snapshot variant.`, + ); + } else if ( + summary && + summary.status === 'ready' && + summary.type === 'summary' + ) { + projectedEp.steps = [ + { + id: ep.id + '-summary', + type: 'AGENT_THOUGHT', + text: summary.text, + metadata: { + originalTokens: epTokens, + currentTokens: summary.recoveredTokens || 50, + transformations: [ + { + processorName: 'AsyncSemanticCompressor', + action: 'SUMMARIZED', + timestamp: Date.now(), + }, + ], + }, + }, + ] as any; + projectedEp.yield = undefined; + tracer.logEvent( + 'ViewGenerator', + `Episode [${ep.id}] has SummaryVariant. Selecting variant over raw text.`, + ); + debugLogger.log( + `Opportunistically swapped Episode ${ep.id} for pre-computed Summary variant.`, + ); + } else if ( + masked && + masked.status === 'ready' && + masked.type === 'masked' + ) { + if ( + projectedEp.trigger.type === 'USER_PROMPT' && + projectedEp.trigger.semanticParts && + projectedEp.trigger.semanticParts.length > 0 + ) { + projectedEp.trigger.semanticParts[0].presentation = { + text: masked.text, + tokens: masked.recoveredTokens || 10, + }; + } + tracer.logEvent( + 'ViewGenerator', + `Episode [${ep.id}] has MaskedVariant. Selecting variant over raw text.`, + ); + debugLogger.log( + `Opportunistically swapped Episode ${ep.id} for pre-computed Masked variant.`, + ); + } + } + + currentEpisodes.unshift(projectedEp); + rollingTokens += calculateEpisodeListTokens([projectedEp]); + } + + return currentEpisodes; +} diff --git a/packages/core/src/context/ir/projector.ts b/packages/core/src/context/ir/projector.ts new file mode 100644 index 0000000000..1f777a95a3 --- /dev/null +++ b/packages/core/src/context/ir/projector.ts @@ -0,0 +1,36 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { Content } from '@google/genai'; +import { IrMapper } from './mapper.js'; +import type { Episode } from './types.js'; +import { debugLogger } from '../../utils/debugLogger.js'; +import type { ContextEnvironment } from '../sidecar/environment.js'; + +export class IrProjector { + /** + * Converts the internal IR graph into a flat Content[] array for the LLM. + * If tracing is enabled via environment variables, dumps the payload to disk. + */ + static async projectAndDump(episodes: Episode[], env: ContextEnvironment): Promise { + const contents = IrMapper.fromIr(episodes); + + if (process.env['GEMINI_DUMP_CONTEXT'] === 'true') { + try { + const fs = await import('node:fs/promises'); + const path = await import('node:path'); + const dumpPath = path.join(env.getTraceDir(), '.gemini', 'projected_context.json'); + await fs.mkdir(path.dirname(dumpPath), { recursive: true }); + await fs.writeFile(dumpPath, JSON.stringify(contents, null, 2), 'utf-8'); + debugLogger.log(`[Observability] Context successfully dumped to ${dumpPath}`); + } catch (e) { + debugLogger.error(`Failed to dump context: ${e}`); + } + } + + return contents; + } +} diff --git a/packages/core/src/context/processors/emergencyTruncationProcessor.ts b/packages/core/src/context/processors/emergencyTruncationProcessor.ts new file mode 100644 index 0000000000..69e077fdeb --- /dev/null +++ b/packages/core/src/context/processors/emergencyTruncationProcessor.ts @@ -0,0 +1,48 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { ContextProcessor, ContextAccountingState } from '../pipeline.js'; +import type { Episode } from '../ir/types.js'; +import type { ContextEnvironment } from '../sidecar/environment.js'; +import { estimateContextTokenCountSync } from '../utils/contextTokenCalculator.js'; + +export interface EmergencyTruncationProcessorOptions {} + +export class EmergencyTruncationProcessor implements ContextProcessor { + static create(env: ContextEnvironment, options: EmergencyTruncationProcessorOptions): EmergencyTruncationProcessor { + return new EmergencyTruncationProcessor(env, options); + } + + readonly id = 'EmergencyTruncationProcessor'; + readonly name = 'EmergencyTruncationProcessor'; + readonly options: EmergencyTruncationProcessorOptions; + constructor(_env: ContextEnvironment, options: EmergencyTruncationProcessorOptions) { + this.options = options; + } + + async process(episodes: Episode[], state: ContextAccountingState): Promise { + if (state.currentTokens <= state.maxTokens) return episodes; + + let remainingTokens = state.currentTokens; + const targetTokens = state.maxTokens; + const truncated: Episode[] = []; + + // We respect the global protected Episode IDs (like the system prompt at index 0) + for (const ep of episodes) { + // Calculate individual episode tokens efficiently (assume metadata is accurate if present) + const epTokens = ep.yield?.metadata?.currentTokens ?? estimateContextTokenCountSync([{ text: ep.yield?.text ?? '' }]); + + if (remainingTokens > targetTokens && !state.protectedEpisodeIds.has(ep.id)) { + remainingTokens -= epTokens; + // Dropped! We do not add it to the truncated array. + } else { + truncated.push(ep); + } + } + + return truncated; + } +} diff --git a/packages/core/src/context/sidecar/profiles.ts b/packages/core/src/context/sidecar/profiles.ts index 7610979438..8d4fea60c5 100644 --- a/packages/core/src/context/sidecar/profiles.ts +++ b/packages/core/src/context/sidecar/profiles.ts @@ -27,7 +27,8 @@ export const defaultSidecarProfile: SidecarConfig = { processors: [ { processorId: 'ToolMaskingProcessor', options: { stringLengthThresholdTokens: 8000 } }, { processorId: 'BlobDegradationProcessor', options: {} }, - { processorId: 'SemanticCompressionProcessor', options: { nodeThresholdTokens: 5000, contextWindowPercentage: 0.2 } } + { processorId: 'SemanticCompressionProcessor', options: { nodeThresholdTokens: 5000, contextWindowPercentage: 0.2 } }, + { processorId: 'EmergencyTruncationProcessor', options: {} } ] }, { diff --git a/packages/core/src/context/sidecar/registry.ts b/packages/core/src/context/sidecar/registry.ts index 867612e108..ac8edd8da4 100644 --- a/packages/core/src/context/sidecar/registry.ts +++ b/packages/core/src/context/sidecar/registry.ts @@ -7,6 +7,7 @@ import type { ContextProcessor } from '../pipeline.js'; import type { ContextEnvironment } from './environment.js'; + export interface ContextProcessorDef< TOptions extends Record = any, > { diff --git a/packages/core/src/context/utils/contextTokenCalculator.ts b/packages/core/src/context/utils/contextTokenCalculator.ts index d120e8fc63..1ed5fb452e 100644 --- a/packages/core/src/context/utils/contextTokenCalculator.ts +++ b/packages/core/src/context/utils/contextTokenCalculator.ts @@ -6,6 +6,19 @@ import type { Part } from '@google/genai'; import { estimateTokenCountSync as baseEstimate } from '../../utils/tokenCalculation.js'; +import type { Episode } from '../ir/types.js'; + +export function calculateEpisodeListTokens(episodes: Episode[]): number { + let tokens = 0; + for (const ep of episodes) { + if (ep.trigger) tokens += ep.trigger.metadata.currentTokens; + for (const step of ep.steps) { + tokens += step.metadata.currentTokens; + } + if (ep.yield) tokens += ep.yield.metadata.currentTokens; + } + return tokens; +} export function estimateContextTokenCountSync( parts: Part[],