diff --git a/apps/sidecar/src/workflow-substrate-factory/compactors.test.ts b/apps/sidecar/src/workflow-substrate-factory/compactors.test.ts index 6d88b1963..fb4a51e6f 100644 --- a/apps/sidecar/src/workflow-substrate-factory/compactors.test.ts +++ b/apps/sidecar/src/workflow-substrate-factory/compactors.test.ts @@ -7,8 +7,11 @@ import { expect, test } from "bun:test"; import type { ConversationTurn, StrategyContext } from "@intx/types/runtime"; import { + SUMMARIZE_BUDGETED_TURNS_NAME, SUMMARIZE_OLDER_TURNS_NAME, + createBudgetedContextCompactor, createSummarizeOlderTurnsCompactor, + estimateTurnsChars, } from "./compactors"; function textTurn( @@ -118,3 +121,46 @@ test("carries the compactor's name and a version for the manifest record", () => expect(compactor.name).toBe(SUMMARIZE_OLDER_TURNS_NAME); expect(compactor.version).toBe("1"); }); + +test("createBudgetedContextCompactor: a short conversation under budget is untouched", async () => { + const turns = Array.from({ length: 6 }, (_, i) => + textTurn(i % 2 === 0 ? "user" : "assistant", `turn ${i}`, i), + ); + const compactor = createBudgetedContextCompactor( + estimateTurnsChars(turns) + 1_000, + ); + + const result = await compactor.apply(turns, makeCtx()); + + expect(result.output).toEqual(turns); + expect(result.record.reason).toBe("within-budget"); +}); + +test("createBudgetedContextCompactor: a conversation past the budget is folded rather than resent whole", async () => { + const turns = Array.from({ length: 40 }, (_, i) => + textTurn(i % 2 === 0 ? "user" : "assistant", `message number ${i}`, i), + ); + // A budget generous enough for only the newest handful of turns. + const budgetChars = estimateTurnsChars(turns.slice(-6)); + const compactor = createBudgetedContextCompactor(budgetChars); + + const result = await compactor.apply(turns, makeCtx()); + + expect(result.record.reason).toBe("folded-older-turns"); + expect(result.record.strategy).toBe(SUMMARIZE_BUDGETED_TURNS_NAME); + expect(result.output.length).toBeLessThan(turns.length); + const [summary] = result.output; + expect(summary?.role).toBe("system"); +}); + +test("createBudgetedContextCompactor: always keeps a minimum verbatim tail even under a near-zero budget", async () => { + const turns = Array.from({ length: 10 }, (_, i) => + textTurn(i % 2 === 0 ? "user" : "assistant", `message ${i}`, i), + ); + const compactor = createBudgetedContextCompactor(1); + + const result = await compactor.apply(turns, makeCtx()); + + // 1 synthetic summary turn + at least the floor of kept turns. + expect(result.output.length).toBeGreaterThanOrEqual(2); +}); diff --git a/apps/sidecar/src/workflow-substrate-factory/compactors.ts b/apps/sidecar/src/workflow-substrate-factory/compactors.ts index 88dea8a63..75deb3726 100644 --- a/apps/sidecar/src/workflow-substrate-factory/compactors.ts +++ b/apps/sidecar/src/workflow-substrate-factory/compactors.ts @@ -1,41 +1,29 @@ -// The `summarize-older-turns` compactor (CL-6204): a `Compactor` (per -// `@intx/types/runtime`'s `ContextStrategy`) the sidecar registers on `env.compactors` for -// long-lived channel agents (the assistant workflow's warm, durable- -// conversation step) so a director that names it via -// `caps.compact("summarize-older-turns", reason)` gets a bounded fold -// instead of the reactor growing history until the model window -// truncates it. +// Two `Compactor`s (per `@intx/types/runtime`'s `ContextStrategy< +// ConversationTurn[], ConversationTurn[]>`), both deterministic textual +// folds, not LLM calls: `ContextStrategy.apply` receives only `{ state, +// trigger }` (`StrategyContext`) -- no inference handle -- so an +// LLM-based summary is not cleanly available inside this seam today. // -// Deterministic textual fold, not an LLM call: `ContextStrategy.apply` -// receives only `{ state, trigger }` (`StrategyContext`) -- no inference -// handle -- so an LLM-based summary is not cleanly available inside this -// seam today. A future version could route through a director-supplied -// summarizer if the runtime grows one; until then this fold keeps the -// newest `keep` turns verbatim and replaces everything older with one -// synthetic `system` turn built from bounded, deterministic per-turn -// excerpts. +// - `summarize-older-turns`: fixed `keep` (default 12), registered on +// `env.compactors` for a director that names it via +// `caps.compact("summarize-older-turns", reason)`. +// - `summarize-budgeted-turns` (CL-6204): the one actually fired, by +// `WorkbenchDirector` (see `./workbench-director`) via the same +// `caps.compact` reactor action. `@intx/agent`'s `createAgent` does +// not thread a `ContextTransform` seam through `BaseEnv` at all (only +// `env.compactors` reaches `createReactorAssembly` -- confirmed +// against `@intx/agent`'s `agent.js`), so the reactor's `compact` +// action is the only seam actually available; `WorkbenchDirector` +// fires it only at the one point in the reactor's event flow where +// `compact`'s exclusion from `infer` in the same cycle cannot stall a +// reply -- see that file's header comment for exactly which point and +// why. // -// NB (the remaining CL-6204 gap): nothing calls `caps.compact` yet. The -// built-in `DefaultDirector` (`@intx/inference`'s -// `createDefaultDirector`, the only director our channel-host/assistant -// `AgentDefinition`s reference) never emits a `compact` action -- see -// `resolveDirector` in `@intx/agent/agent.ts`, which surfaces -// `compactorNames` to a director factory but the default factory's -// `decide()` never reads them. Firing compaction for real needs a -// director that decides to compact, which the reactor's control flow -// (`@intx/inference/src/reactor.ts` ~1375) only accepts as an -// action exclusive of `infer` in the same cycle, with no synthetic -// follow-up event afterward -- so a director cannot fire it on -// `message.received` without stalling that message's reply. Registering -// this compactor makes the name resolvable and the fold available; the -// trigger itself needs either a vendor-side chained action/event after -// `executeCompact`, or a definition-side custom director authored via -// `defineDirector`/`createWorkflowDirectorRegistry` that fires compact -// on a cycle that doesn't owe an immediate reply (e.g. `tool.done` -// before the batch's re-infer would still hit the same pairing -// restriction, so even that needs the vendor seam to grow a hook this -// registry alone cannot supply). +// Both folds are deliberately lossy -- a bounded recap of what was +// active in the folded turns, not a durable record. Anything a channel +// agent needs to recall past the working window is firm memory's job +// (`@corbits/memory-tools`'s `memory_search`/`memory_add`, already +// pinned for the assistant workflow), not this recap's. import type { Compactor, @@ -61,6 +49,52 @@ export type SummarizeOlderTurnsOptions = { maxExcerptCharsPerTurn?: number; }; +export const SUMMARIZE_BUDGETED_TURNS_NAME = "summarize-budgeted-turns"; +const SUMMARIZE_BUDGETED_TURNS_VERSION = "1"; +const MIN_KEPT_TURNS = 4; + +function turnChars(turn: ConversationTurn): number { + let total = 0; + for (const block of turn.content) { + total += excerptBlock(block).length; + } + return total; +} + +/** Total character length of a turn list -- the budget check's estimate. */ +export function estimateTurnsChars(turns: ConversationTurn[]): number { + let total = 0; + for (const turn of turns) { + total += turnChars(turn); + } + return total; +} + +/** + * Newest-first walk that keeps turns while their combined length stays + * under `budgetChars`, with a floor of `MIN_KEPT_TURNS` so the most + * recent exchange is never folded away even when it alone exceeds + * budget (the honest-overflow case a director checks separately). + */ +function countTurnsWithinBudget( + turns: ConversationTurn[], + budgetChars: number, +): number { + let runningChars = 0; + let kept = 0; + for (let i = turns.length - 1; i >= 0; i--) { + const turn = turns[i]; + if (turn === undefined) continue; + const chars = turnChars(turn); + if (kept >= MIN_KEPT_TURNS && runningChars + chars > budgetChars) { + break; + } + runningChars += chars; + kept += 1; + } + return kept; +} + function roleLabel(role: ConversationTurn["role"]): string { return role === "user" ? "User" @@ -139,41 +173,135 @@ export function createSummarizeOlderTurnsCompactor( }; } - const splitAt = turns.length - keep; - const older = turns.slice(0, splitAt); - const kept = turns.slice(splitAt); - - const lines = older.map( - (turn, i) => - `${String(i + 1)}. ${roleLabel(turn.role)}: ${excerptTurn(turn, maxExcerptCharsPerTurn)}`, - ); - let summaryText = - `[Summary of ${String(older.length)} earlier turns, folded to stay ` + - `within context]\n${lines.join("\n")}`; - if (summaryText.length > maxSummaryChars) { - summaryText = `${summaryText.slice(0, maxSummaryChars - 1)}…`; - } + return foldOlderTurns(turns, keep, { + strategy: SUMMARIZE_OLDER_TURNS_NAME, + version: SUMMARIZE_OLDER_TURNS_VERSION, + maxSummaryChars, + maxExcerptCharsPerTurn, + }); + }, + }; +} - const summaryTurn: ConversationTurn = { - role: "system", - content: [{ type: "text", text: summaryText }], - timestamp: older[0]?.timestamp ?? Date.now(), - }; - - return { - output: [summaryTurn, ...kept], - record: { - strategy: SUMMARIZE_OLDER_TURNS_NAME, - version: SUMMARIZE_OLDER_TURNS_VERSION, - parameters: { keep, maxSummaryChars, maxExcerptCharsPerTurn }, - reason: "folded-older-turns", - decisions: { - foldedCount: older.length, - keptCount: kept.length, - summaryChars: summaryText.length, +/** + * Shared fold: keeps the newest `keep` turns verbatim and replaces + * everything older with one synthetic `system` turn of numbered, + * per-turn excerpts, truncated to `maxSummaryChars`. The recap is + * deliberately lossy -- it is a bounded reminder of what was active in + * the folded turns, not a durable record. Anything a channel agent + * needs to recall past the working window belongs in firm memory + * (`@corbits/memory-tools`'s `memory_search`/`memory_add`), not in this + * recap. + */ +function foldOlderTurns( + turns: ConversationTurn[], + keep: number, + meta: { + strategy: string; + version: string; + maxSummaryChars: number; + maxExcerptCharsPerTurn: number; + }, +): StrategyResult { + const splitAt = turns.length - keep; + const older = turns.slice(0, splitAt); + const kept = turns.slice(splitAt); + + const lines = older.map( + (turn, i) => + `${String(i + 1)}. ${roleLabel(turn.role)}: ${excerptTurn(turn, meta.maxExcerptCharsPerTurn)}`, + ); + let summaryText = + `[Summary of ${String(older.length)} earlier turns, folded to stay ` + + `within context]\n${lines.join("\n")}`; + if (summaryText.length > meta.maxSummaryChars) { + summaryText = `${summaryText.slice(0, meta.maxSummaryChars - 1)}…`; + } + + const summaryTurn: ConversationTurn = { + role: "system", + content: [{ type: "text", text: summaryText }], + timestamp: older[0]?.timestamp ?? Date.now(), + }; + + return { + output: [summaryTurn, ...kept], + record: { + strategy: meta.strategy, + version: meta.version, + parameters: { + keep, + maxSummaryChars: meta.maxSummaryChars, + maxExcerptCharsPerTurn: meta.maxExcerptCharsPerTurn, + }, + reason: "folded-older-turns", + decisions: { + foldedCount: older.length, + keptCount: kept.length, + summaryChars: summaryText.length, + }, + }, + }; +} + +export type BudgetedContextCompactorOptions = { + /** Hard cap on the synthetic summary turn's total text length. */ + maxSummaryChars?: number; + /** Per-folded-turn excerpt cap, applied before the total cap. */ + maxExcerptCharsPerTurn?: number; +}; + +/** + * Builds a `summarize-budgeted-turns` `Compactor`: folds turns older + * than the newest ones that fit `budgetChars` (with a floor of + * `MIN_KEPT_TURNS` verbatim), instead of a fixed turn count. + * `budgetChars` should come from `resolveContextBudgetChars` so the + * fold point tracks the model's actual context window rather than one + * constant shared by every model. + * + * Registered on `env.compactors` (see `./step-env`) and fired by + * `WorkbenchDirector` via `caps.compact("summarize-budgeted-turns", …)` + * only at the one point in the reactor's event flow where doing so + * cannot stall a reply -- see `./workbench-director`'s header comment. + */ +export function createBudgetedContextCompactor( + budgetChars: number, + options: BudgetedContextCompactorOptions = {}, +): Compactor { + const maxSummaryChars = options.maxSummaryChars ?? DEFAULT_MAX_SUMMARY_CHARS; + const maxExcerptCharsPerTurn = + options.maxExcerptCharsPerTurn ?? DEFAULT_MAX_EXCERPT_CHARS_PER_TURN; + + return { + name: SUMMARIZE_BUDGETED_TURNS_NAME, + version: SUMMARIZE_BUDGETED_TURNS_VERSION, + async apply( + turns: ConversationTurn[], + _ctx: StrategyContext, + ): Promise> { + const keep = countTurnsWithinBudget(turns, budgetChars); + if (keep >= turns.length) { + return { + output: turns, + record: { + strategy: SUMMARIZE_BUDGETED_TURNS_NAME, + version: SUMMARIZE_BUDGETED_TURNS_VERSION, + parameters: { + budgetChars, + maxSummaryChars, + maxExcerptCharsPerTurn, + }, + reason: "within-budget", + decisions: { turnCount: turns.length }, }, - }, - }; + }; + } + return foldOlderTurns(turns, keep, { + strategy: SUMMARIZE_BUDGETED_TURNS_NAME, + version: SUMMARIZE_BUDGETED_TURNS_VERSION, + maxSummaryChars, + maxExcerptCharsPerTurn, + }); }, }; } diff --git a/apps/sidecar/src/workflow-substrate-factory/context-budget.test.ts b/apps/sidecar/src/workflow-substrate-factory/context-budget.test.ts new file mode 100644 index 000000000..9d1c4aa12 --- /dev/null +++ b/apps/sidecar/src/workflow-substrate-factory/context-budget.test.ts @@ -0,0 +1,57 @@ +// Tests for the CL-6204 context-budget resolver: reads a per-model +// `numCtx` hint off an `InferenceSource.quirks` bag shaped like +// `@corbits/ollama-adapter`'s `OllamaAdapterConfig`, falling back to a +// conservative default for any other shape. +import { expect, test } from "bun:test"; + +import { + readNumCtxHint, + resolveContextBudgetChars, + resolveHardContextLimitChars, +} from "./context-budget"; + +test("readNumCtxHint: reads a per-model override over the default", () => { + const quirks = { + default: { numCtx: 8_000 }, + perModel: { "gpt-oss:20b": { numCtx: 128_000 } }, + }; + + expect(readNumCtxHint(quirks, "gpt-oss:20b")).toBe(128_000); + expect(readNumCtxHint(quirks, "other-model")).toBe(8_000); +}); + +test("readNumCtxHint: unrecognized or absent quirks resolve to undefined", () => { + expect(readNumCtxHint(undefined, "claude")).toBeUndefined(); + expect(readNumCtxHint(null, "claude")).toBeUndefined(); + expect(readNumCtxHint({ unrelated: true }, "claude")).toBeUndefined(); + expect(readNumCtxHint("not-an-object", "claude")).toBeUndefined(); +}); + +test("resolveContextBudgetChars: a bigger numCtx yields a bigger budget", () => { + const small = resolveContextBudgetChars( + { default: { numCtx: 32_000 } }, + "qwen3", + ); + const large = resolveContextBudgetChars( + { default: { numCtx: 128_000 } }, + "gpt-oss:20b", + ); + + expect(large).toBeGreaterThan(small); +}); + +test("resolveContextBudgetChars: unknown model falls back to the conservative default", () => { + const withoutQuirks = resolveContextBudgetChars(undefined, "unknown-model"); + + expect(withoutQuirks).toBeGreaterThan(0); + expect(Number.isFinite(withoutQuirks)).toBe(true); +}); + +test("resolveHardContextLimitChars: sits above the headroomed budget for the same source", () => { + const quirks = { default: { numCtx: 32_000 } }; + + const budget = resolveContextBudgetChars(quirks, "m"); + const hardLimit = resolveHardContextLimitChars(quirks, "m"); + + expect(hardLimit).toBeGreaterThan(budget); +}); diff --git a/apps/sidecar/src/workflow-substrate-factory/context-budget.ts b/apps/sidecar/src/workflow-substrate-factory/context-budget.ts new file mode 100644 index 000000000..bfad27b0c --- /dev/null +++ b/apps/sidecar/src/workflow-substrate-factory/context-budget.ts @@ -0,0 +1,90 @@ +// Best-effort per-model context-window budget for the workbench +// director's compaction gate (CL-6204). +// +// Reads the same `numCtx` override shape `@corbits/ollama-adapter` +// resolves against a live request (`OllamaAdapterConfig`'s +// `default`/`perModel` bag) directly off `InferenceSource.quirks`, so a +// deployment that already pins a per-model `num_ctx` for Ollama gets +// that same number as its compaction budget instead of a second, +// drifting constant. `quirks` is `unknown` at this boundary (an +// operator-authored, per-source passthrough bag), so every read here is +// defensive rather than a validating parse: an unrecognized shape (every +// non-Ollama source today) falls back to a conservative default sized +// for the smallest models this repo deploys against, rather than +// throwing or silently trusting a shape the adapter itself doesn't own +// at this call site. +// +// Token counts are estimated from character counts (`CHARS_PER_TOKEN_ESTIMATE`) +// -- no tokenizer is available at this layer -- so the budget is +// deliberately conservative (see `COMPACTION_HEADROOM`), leaving room +// for the system prompt, tool definitions, and the model's own reply +// inside the same window. + +const DEFAULT_CONTEXT_BUDGET_TOKENS = 8_000; +const CHARS_PER_TOKEN_ESTIMATE = 4; +const COMPACTION_HEADROOM = 0.6; + +function readNumCtxFromBag(bag: Record): number | undefined { + const numCtx = bag.numCtx; + return typeof numCtx === "number" && numCtx > 0 ? numCtx : undefined; +} + +/** + * Best-effort read of a per-model `numCtx` off an `InferenceSource.quirks` + * bag shaped like `@corbits/ollama-adapter`'s `OllamaAdapterConfig` + * (`{ default?: { numCtx? }, perModel?: { [model]: { numCtx? } } }`). + * Returns `undefined` for any other shape rather than throwing -- + * `quirks` is provider-specific and most sources carry none of this. + */ +export function readNumCtxHint( + quirks: unknown, + model: string, +): number | undefined { + if (typeof quirks !== "object" || quirks === null) { + return undefined; + } + const bag = quirks as Record; + const perModel = bag.perModel; + if (typeof perModel === "object" && perModel !== null) { + const entry = (perModel as Record)[model]; + if (typeof entry === "object" && entry !== null) { + const fromPerModel = readNumCtxFromBag(entry as Record); + if (fromPerModel !== undefined) { + return fromPerModel; + } + } + } + const base = bag.default; + if (typeof base === "object" && base !== null) { + return readNumCtxFromBag(base as Record); + } + return undefined; +} + +/** + * Resolve the compaction budget, in characters, for one `InferenceSource`. + * Applies `COMPACTION_HEADROOM` on top of the resolved (or default) + * `numCtx` so compaction fires with room left in the window, not at its + * hard edge. + */ +export function resolveContextBudgetChars( + quirks: unknown, + model: string, +): number { + const numCtx = readNumCtxHint(quirks, model) ?? DEFAULT_CONTEXT_BUDGET_TOKENS; + return Math.floor(numCtx * CHARS_PER_TOKEN_ESTIMATE * COMPACTION_HEADROOM); +} + +/** + * The raw (no-headroom) character budget for one source -- the point + * past which content can no longer be assumed to fit the model's window + * at all, used to detect the unrecoverable case where even the turns a + * compactor must keep verbatim are already too large. + */ +export function resolveHardContextLimitChars( + quirks: unknown, + model: string, +): number { + const numCtx = readNumCtxHint(quirks, model) ?? DEFAULT_CONTEXT_BUDGET_TOKENS; + return Math.floor(numCtx * CHARS_PER_TOKEN_ESTIMATE); +} diff --git a/apps/sidecar/src/workflow-substrate-factory/step-env.test.ts b/apps/sidecar/src/workflow-substrate-factory/step-env.test.ts index 0ffc96c2b..004941e99 100644 --- a/apps/sidecar/src/workflow-substrate-factory/step-env.test.ts +++ b/apps/sidecar/src/workflow-substrate-factory/step-env.test.ts @@ -18,7 +18,10 @@ import type { SourcesSnapshotRef, } from "@intx/workflow-host"; -import { SUMMARIZE_OLDER_TURNS_NAME } from "./compactors"; +import { + SUMMARIZE_BUDGETED_TURNS_NAME, + SUMMARIZE_OLDER_TURNS_NAME, +} from "./compactors"; import { createSidecarStepBuildEnv } from "./step-env"; const tmpDirs: string[] = []; @@ -191,8 +194,13 @@ test("the built step env forwards the summarize-older-turns compactor (CL-6204) const compactors = ( env as unknown as { compactors: Record } ).compactors; - expect(Object.keys(compactors)).toEqual([SUMMARIZE_OLDER_TURNS_NAME]); + expect(Object.keys(compactors).sort()).toEqual( + [SUMMARIZE_OLDER_TURNS_NAME, SUMMARIZE_BUDGETED_TURNS_NAME].sort(), + ); expect(compactors[SUMMARIZE_OLDER_TURNS_NAME]?.name).toBe( SUMMARIZE_OLDER_TURNS_NAME, ); + expect(compactors[SUMMARIZE_BUDGETED_TURNS_NAME]?.name).toBe( + SUMMARIZE_BUDGETED_TURNS_NAME, + ); }); diff --git a/apps/sidecar/src/workflow-substrate-factory/step-env.ts b/apps/sidecar/src/workflow-substrate-factory/step-env.ts index ee11d0f7a..981fcf6f9 100644 --- a/apps/sidecar/src/workflow-substrate-factory/step-env.ts +++ b/apps/sidecar/src/workflow-substrate-factory/step-env.ts @@ -37,10 +37,16 @@ import { type StepToolMaterialization, } from "../step-agent-tools"; import { + SUMMARIZE_BUDGETED_TURNS_NAME, SUMMARIZE_OLDER_TURNS_NAME, + createBudgetedContextCompactor, createSummarizeOlderTurnsCompactor, } from "./compactors"; import { createStepInferenceSourceResolver } from "./config"; +import { + resolveContextBudgetChars, + resolveHardContextLimitChars, +} from "./context-budget"; import { stepStorageRoot, warmStepStorageRoot } from "./storage-paths"; import { createWorkbenchDirectorRegistry } from "./workbench-director"; @@ -48,19 +54,11 @@ import { createWorkbenchDirectorRegistry } from "./workbench-director"; // the compactor is a pure, stateless `Compactor` (see `./compactors`), so // one instance can serve every step's `env.compactors` map. Registering it // here (CL-6204) makes the name resolvable to any director that names it -// via `caps.compact("summarize-older-turns", reason)`; see `./compactors`'s -// header comment for the remaining gap -- no director in this codebase -// fires that action yet. +// via `caps.compact("summarize-older-turns", reason)`. const stepCompactors = { [SUMMARIZE_OLDER_TURNS_NAME]: createSummarizeOlderTurnsCompactor(), }; -// Registered once and reused across every step env this builder produces: -// the workbench director wraps DefaultDirector with empty-turn retry and -// is the sidecar default (see `./workbench-director`). `@intx/agent/default` -// stays resolvable for definitions that name it. -const stepDirectors = createWorkbenchDirectorRegistry(); - const isogitStorage = createIsogitStorage(createNodeIsogitRuntime()); export interface SidecarStepBuildEnvDeps { @@ -228,6 +226,30 @@ export function createSidecarStepBuildEnv( ); } + // Context-window budget for this step's active source (CL-6204): + // sized from `activeSource.quirks`/`activeSource.model` so a 128K + // model and a 32K model compact at different points rather than + // sharing one constant. Built per invocation (not module scope) + // because the budget depends on the step's own resolved source. + const contextBudgetChars = resolveContextBudgetChars( + activeSource.quirks, + activeSource.model, + ); + const contextHardLimitChars = resolveHardContextLimitChars( + activeSource.quirks, + activeSource.model, + ); + const stepDirectors = createWorkbenchDirectorRegistry({ + budgetChars: contextBudgetChars, + hardLimitChars: contextHardLimitChars, + compactorName: SUMMARIZE_BUDGETED_TURNS_NAME, + }); + const compactors = { + ...stepCompactors, + [SUMMARIZE_BUDGETED_TURNS_NAME]: + createBudgetedContextCompactor(contextBudgetChars), + }; + // Root the per-step scratch (workspace + tool tarball-cache + // apply-state). The cold (multi-step) path keys it per // run/step/attempt: each run rebuilds the agent and its scratch, and @@ -346,7 +368,7 @@ export function createSidecarStepBuildEnv( workdir, audit: storage, directors: stepDirectors, - compactors: stepCompactors, + compactors, // Resolve inference adapters through the child's boot-built // registry (built-ins + operator custom adapters), so a // custom-provider step source resolves in the child the same way diff --git a/apps/sidecar/src/workflow-substrate-factory/workbench-director.test.ts b/apps/sidecar/src/workflow-substrate-factory/workbench-director.test.ts index 43cbb2cde..5fd4142bb 100644 --- a/apps/sidecar/src/workflow-substrate-factory/workbench-director.test.ts +++ b/apps/sidecar/src/workflow-substrate-factory/workbench-director.test.ts @@ -6,6 +6,7 @@ import { defaultDirectorFactory } from "@intx/agent"; import { createCapabilities, createDefaultDirector } from "@intx/inference"; import type { AssistantTurn, + ConversationTurn, ReactorAction, ReactorDirector, ReactorInboundEvent, @@ -14,6 +15,7 @@ import type { } from "@intx/types/runtime"; import { + CONTEXT_OVERFLOW_MESSAGE, EMPTY_TURN_REPLY, WORKBENCH_DIRECTOR_ID, createWorkbenchDirector, @@ -68,6 +70,26 @@ function toolTurn(name: string): AssistantTurn { }; } +function twoCallToolTurn(): AssistantTurn { + return { + role: "assistant", + content: [ + { type: "tool_call", id: "call_1", name: "look_up", arguments: {} }, + { type: "tool_call", id: "call_2", name: "look_up", arguments: {} }, + ], + model: "test", + timestamp: 0, + }; +} + +function conversationTurn(text: string): ConversationTurn { + return { role: "user", content: [{ type: "text", text }], timestamp: 0 }; +} + +function stateWithTurns(turns: ConversationTurn[]): ReactorState { + return { ...state(), turns }; +} + function inferenceDone(turn: AssistantTurn): ReactorInboundEvent { return { type: "inference.done", turn, usage, source }; } @@ -191,6 +213,118 @@ test("a new message.received resets the empty-turn retry budget", async () => { expect(typesOf(actions)).toEqual(["checkpoint", "infer"]); }); +test("context budget: a short conversation under budget is untouched (infers normally)", async () => { + const director = createWorkbenchDirector( + "you are a test agent", + [], + {}, + { + budgetChars: 10_000, + hardLimitChars: 20_000, + compactorName: "summarize-budgeted-turns", + }, + ); + const shortState = stateWithTurns([conversationTurn("hi")]); + + const actions = await director.decide( + { type: "message.received", message: { id: "m1", content: "hi" } as never }, + shortState, + caps, + ); + + expect(typesOf(actions)).toEqual(["infer"]); +}); + +test("context budget: history past the hard limit replies with the honest overflow message instead of inferring", async () => { + const director = createWorkbenchDirector( + "you are a test agent", + [], + {}, + { + budgetChars: 10, + hardLimitChars: 20, + compactorName: "summarize-budgeted-turns", + }, + ); + const bigState = stateWithTurns([ + conversationTurn("a".repeat(1_000)), + conversationTurn("more content that pushes well past the hard limit"), + ]); + + const actions = await director.decide( + { type: "message.received", message: { id: "m1", content: "hi" } as never }, + bigState, + caps, + ); + + expect(typesOf(actions)).toEqual(["checkpoint", "reply"]); + expect(replyOf(actions)).toBe(CONTEXT_OVERFLOW_MESSAGE); +}); + +test("context budget: a non-final tool.done in a multi-call batch compacts instead of no-op when over budget", async () => { + const director = createWorkbenchDirector( + "you are a test agent", + [], + {}, + { + budgetChars: 10, + hardLimitChars: 1_000_000, + compactorName: "summarize-budgeted-turns", + }, + ); + await director.decide(inferenceDone(twoCallToolTurn()), state(), caps); + + const bigState = stateWithTurns([ + conversationTurn("a".repeat(200)), + conversationTurn("b".repeat(200)), + ]); + const firstDone: ReactorInboundEvent = { + type: "tool.done", + result: { callId: "call_1", content: "ok", isError: false }, + }; + + const actions = await director.decide(firstDone, bigState, caps); + + expect(typesOf(actions)).toEqual(["checkpoint", "compact"]); +}); + +test("context budget: a non-final tool.done under budget stays a no-op (regression)", async () => { + const director = createWorkbenchDirector( + "you are a test agent", + [], + {}, + { + budgetChars: 1_000_000, + hardLimitChars: 2_000_000, + compactorName: "summarize-budgeted-turns", + }, + ); + await director.decide(inferenceDone(twoCallToolTurn()), state(), caps); + + const smallState = stateWithTurns([conversationTurn("hi")]); + const firstDone: ReactorInboundEvent = { + type: "tool.done", + result: { callId: "call_1", content: "ok", isError: false }, + }; + + const actions = await director.decide(firstDone, smallState, caps); + + expect(typesOf(actions)).toEqual([]); +}); + +test("context budget: with no contextBudget configured, behavior is unchanged", async () => { + const director = createWorkbenchDirector("you are a test agent"); + const bigState = stateWithTurns([conversationTurn("a".repeat(100_000))]); + + const actions = await director.decide( + { type: "message.received", message: { id: "m1", content: "hi" } as never }, + bigState, + caps, + ); + + expect(typesOf(actions)).toEqual(["infer"]); +}); + test("the factory is namespaced and is the sidecar registry default", () => { expect(workbenchDirectorFactory.id).toBe(WORKBENCH_DIRECTOR_ID); const registry = createWorkbenchDirectorRegistry(); diff --git a/apps/sidecar/src/workflow-substrate-factory/workbench-director.ts b/apps/sidecar/src/workflow-substrate-factory/workbench-director.ts index 107bfe7e3..358c86a19 100644 --- a/apps/sidecar/src/workflow-substrate-factory/workbench-director.ts +++ b/apps/sidecar/src/workflow-substrate-factory/workbench-director.ts @@ -1,4 +1,5 @@ -// Workbench director: DefaultDirector plus an empty-turn retry. +// Workbench director: DefaultDirector plus an empty-turn retry and a +// context-budget gate (CL-6204). // // `@intx/inference`'s DefaultDirector checkpoints and waits when // inference.done has no text and no tool calls. The human then sits in a @@ -9,6 +10,35 @@ // outstanding batch is complete, including when the result is an error. // We compose that path rather than reimplement it. // +// Context budget (`contextBudget`, optional -- absent means unbudgeted, +// today's pre-CL-6204 behavior): before letting any inner decision that +// includes `infer` through, checks the turn history against the +// model's real context window (`resolveContextBudgetChars` / +// `resolveHardContextLimitChars` in `./context-budget`, sized from +// `InferenceSource.quirks`). +// +// - Over the hard limit (no headroom left at all): this is Ollama's +// silent-truncation case made honest -- reply with the same +// "exceeded the model's context limit" message hosted providers +// produce via `inference.error`'s `context_overflow` category +// (`@intx/inference`'s `default-director.js`), instead of sending a +// request that would truncate server-side with no error. +// - Over the (headroomed) budget but under the hard limit, on a +// non-final `tool.done` in a multi-call batch (DefaultDirector +// returns `[]` for every `tool.done` before the batch's last): +// fire `caps.compact` here instead of the no-op. This is the one +// point in the reactor's event flow where `compact` cannot stall a +// reply -- the batch's remaining `tool.done` events are already +// enqueued (`@intx/inference`'s `reactor.js` `executeTools` enqueues +// every result from one `Promise.all` before the reactor dequeues +// any of them) and will still drive the eventual re-infer. Firing +// compact on `message.received` or the batch's *last* `tool.done` +// would instead stall that message's reply waiting for a follow-up +// event the reactor never produces (`compactors.ts`'s header +// comment) -- this director never does that. +// - Otherwise: let the inner decision through unchanged. Compaction is +// deferred to the next safe point rather than forced here. +// // Registered as the sidecar step-env default via // `createWorkbenchDirectorRegistry` (see `./step-env`). Id is // `@workbench/sidecar/workbench`; `@intx/agent/default` stays resolvable @@ -38,8 +68,21 @@ import { type ToolDefinition, } from "@intx/types/runtime"; +import { estimateTurnsChars } from "./compactors"; + export const WORKBENCH_DIRECTOR_ID = "@workbench/sidecar/workbench"; export const EMPTY_TURN_REPLY = "I got an empty model turn"; +export const CONTEXT_OVERFLOW_MESSAGE = + "This agent could not complete your request because the conversation exceeded the model's context limit"; + +export type ContextBudgetOptions = { + /** Headroomed char budget past which a safe compaction point fires. */ + budgetChars: number; + /** Raw char limit past which sending would overflow the model's window. */ + hardLimitChars: number; + /** Name the compactor is registered under in `env.compactors`. */ + compactorName: string; +}; function extractToolCalls(turn: AssistantTurn): ToolCall[] { const calls: ToolCall[] = []; @@ -79,17 +122,20 @@ export class WorkbenchDirector implements ReactorDirector { private readonly systemPrompt: string; private readonly toolDefinitions: ToolDefinition[]; private readonly conversational: boolean; + private readonly contextBudget: ContextBudgetOptions | undefined; private emptyTurnRetried = false; constructor( systemPrompt: string, toolDefinitions: ToolDefinition[] = [], policy: DefaultDirectorPolicy = {}, + contextBudget?: ContextBudgetOptions, ) { this.inner = createDefaultDirector(systemPrompt, toolDefinitions, policy); this.systemPrompt = systemPrompt; this.toolDefinitions = toolDefinitions; this.conversational = policy.mode !== "reactive"; + this.contextBudget = contextBudget; } async decide( @@ -103,6 +149,16 @@ export class WorkbenchDirector implements ReactorDirector { const actions = await this.inner.decide(event, state, capabilities); + const budgeted = this.applyContextBudget( + event, + state, + capabilities, + actions, + ); + if (budgeted !== undefined) { + return budgeted; + } + if (event.type !== "inference.done") { return actions; } @@ -142,14 +198,66 @@ export class WorkbenchDirector implements ReactorDirector { extractTextContent(turn).length === 0 ); } + + /** + * Returns a replacement action set when the context budget overrides + * the inner director's decision, `undefined` to let it through + * unchanged. See this file's header comment for the two cases this + * covers (honest overflow, safe-point compaction) and why every other + * case passes through untouched. + */ + private applyContextBudget( + event: ReactorInboundEvent, + state: ReactorState, + capabilities: ReactorCapabilities, + actions: ReactorAction | ReactorAction[], + ): ReactorAction[] | undefined { + if (this.contextBudget === undefined) { + return undefined; + } + const list = Array.isArray(actions) ? actions : [actions]; + const chars = estimateTurnsChars(state.turns); + + if (list.some((action) => action.type === "infer")) { + if (chars > this.contextBudget.hardLimitChars) { + return [ + capabilities.checkpoint("context-overflow"), + capabilities.reply(CONTEXT_OVERFLOW_MESSAGE), + ]; + } + return undefined; + } + + if ( + event.type === "tool.done" && + list.length === 0 && + chars > this.contextBudget.budgetChars + ) { + return [ + capabilities.checkpoint("context-budget-compact"), + capabilities.compact( + this.contextBudget.compactorName, + "context-budget", + ), + ]; + } + + return undefined; + } } export function createWorkbenchDirector( systemPrompt: string, toolDefinitions: ToolDefinition[] = [], policy: DefaultDirectorPolicy = {}, + contextBudget?: ContextBudgetOptions, ): ReactorDirector { - return new WorkbenchDirector(systemPrompt, toolDefinitions, policy); + return new WorkbenchDirector( + systemPrompt, + toolDefinitions, + policy, + contextBudget, + ); } const WorkbenchDirectorConfigSchema = type({ @@ -160,6 +268,27 @@ export type WorkbenchDirectorConfig = { mode?: "conversational" | "reactive"; }; +function buildWorkbenchFactory( + contextBudget: ContextBudgetOptions | undefined, +) { + return defineDirector({ + id: WORKBENCH_DIRECTOR_ID, + configSchema: WorkbenchDirectorConfigSchema, + factory: (config, _env, agent) => { + const policy: DefaultDirectorPolicy = {}; + if (config.mode !== undefined) { + policy.mode = config.mode; + } + return createWorkbenchDirector( + agent.systemPrompt, + [...agent.toolDefinitions], + policy, + contextBudget, + ); + }, + }).factory; +} + const defined = defineDirector({ id: WORKBENCH_DIRECTOR_ID, configSchema: WorkbenchDirectorConfigSchema, @@ -183,10 +312,21 @@ export const buildWorkbenchDirectorRef = defined.build; * Sidecar step-env director registry: workbench is the default so * unspecified AgentDefinitions get empty-turn retry. The built-in * `@intx/agent/default` stays resolvable for definitions that name it. + * + * `contextBudget`, when supplied, bakes a per-step context-window budget + * (sized from the step's active `InferenceSource`, which varies per + * step/model) into the workbench factory this registry resolves -- + * `createSidecarStepBuildEnv` builds a fresh registry per step build + * rather than reusing one shared instance so each step's budget matches + * its own model. See `WorkbenchDirector`'s header comment for what the + * budget gates. */ -export function createWorkbenchDirectorRegistry(): DirectorRegistry { +export function createWorkbenchDirectorRegistry( + contextBudget?: ContextBudgetOptions, +): DirectorRegistry { + const workbenchFactory = buildWorkbenchFactory(contextBudget); return createDirectorRegistry({ - factories: [workbenchDirectorFactory, defaultDirectorFactory], - defaultId: workbenchDirectorFactory.id, + factories: [workbenchFactory, defaultDirectorFactory], + defaultId: workbenchFactory.id, }); }