diff --git a/CHANGELOG.md b/CHANGELOG.md index bfd4a2491..91f6d29a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,10 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename impersonating Reject; Escape still denies. - After context compaction, ChatGPT Codex requests keep the operating prompt as instructions. +- Compaction spacers stamp a reserved harness producer id and a visible + sentinel. Spacer-only replies are incomplete and stay on open-task and + workflow rails. Frozen-prefix matching ignores model-emitted copies of the + marker. ## [0.3.18] - 2026-09-08 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3681d2f36..8ef72ff31 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -162,7 +162,7 @@ The agent maintains an optional **`manage_tasks`** list (create/update via the h #### Context compaction (the compaction governor) -When a cycle's input tokens cross a threshold, the director compacts the inference-facing history (the full run is always retained in the context store). The threshold is **model-aware** — roughly 60% of the active model's real context window — so small-window models compact early enough to avoid provider context-overflow while large-window models do not compact prematurely. The compacted prefix is **append-only across passes**: the existing compacted user turn stays byte-identical; new folds become later summary turns with an assistant spacer between them so the prompt head can remain in the provider KV cache. The governor covers three cases: +When a cycle's input tokens cross a threshold, the director compacts the inference-facing history (the full run is always retained in the context store). The threshold is **model-aware** — roughly 60% of the active model's real context window — so small-window models compact early enough to avoid provider context-overflow while large-window models do not compact prematurely. The compacted prefix is **append-only across passes**: the existing compacted user turn stays byte-identical; new folds become later summary turns with a harness-inserted assistant spacer (identified by reserved `model: "harness"`, plus a visible sentinel; persisted `[compaction]` tokens without a producer still freeze) between them so the prompt head can remain in the provider KV cache. Model-emitted copies of the spacer are not frozen. Spacer-only model replies are incomplete: ChatDirector nudges, then falls through loop-protection, workflow-idle, and open-task rails rather than empty-settling with work still open. The governor covers three cases: - **Threshold at a tool pause** — Once over threshold, the follow-up `infer` after a tool batch is swapped for a `compact` cycle, and inference resumes via a host continuation message. After a compact that remains over the high watermark, the governor uses **growth hysteresis** (wait for usage to grow by ~10% of the window) instead of re-arming on every cycle; dropping under 60% is not required. - **Idle (end-of-turn)** — An interactive turn can end with a reply and then sit idle with no tool batch to intercept; the governor requests a continuation at that pause and compacts when it arrives. An operator message that races the continuation still compacts first, then re-enters inference to answer it. diff --git a/src/agent/compaction.test.ts b/src/agent/compaction.test.ts index ac4f70145..8a35a790e 100644 --- a/src/agent/compaction.test.ts +++ b/src/agent/compaction.test.ts @@ -8,7 +8,12 @@ import type { } from "@intx/types/runtime"; import { createCompactionGovernor } from "./compaction.js"; import { compactionResumeDeltaFor, compactionThresholdFor } from "../provider/context-window.js"; -import { COMPACTOR_KEEP_RECENT_TURNS, compactorNoOpFloor } from "../session/compactor.js"; +import { + COMPACTOR_KEEP_RECENT_TURNS, + COMPACT_SPACER_TEXT, + LEGACY_COMPACT_SPACER_TEXT, + compactorNoOpFloor, +} from "../session/compactor.js"; const capabilities = { infer: (options?: unknown) => ({ type: "infer", ...(options !== undefined ? { options } : {}) }), @@ -37,10 +42,30 @@ function turnsOfLength(count: number, textLength: number): ConversationTurn[] { })) as unknown as ConversationTurn[]; } -function inferenceDone(input: number): Extract { +function inferenceDone( + input: number, + text = "", +): Extract { return { type: "inference.done", - turn: { role: "assistant", content: [] }, + turn: { + role: "assistant", + content: text.length > 0 ? [{ type: "text", text }] : [], + }, + usage: usage(input), + source: { sourceId: "s", provider: "p", model: "m" }, + } as unknown as Extract; +} + +function inferenceDoneWithTools( + input: number, +): Extract { + return { + type: "inference.done", + turn: { + role: "assistant", + content: [{ type: "tool_call", id: "c1", name: "read_file", arguments: { path: "a.ts" } }], + }, usage: usage(input), source: { sourceId: "s", provider: "p", model: "m" }, } as unknown as Extract; @@ -443,4 +468,52 @@ describe("compaction governor", () => { expect(actions).not.toBeNull(); expect(actions?.some((a) => a.type === "compact")).toBe(true); }); + + test("consecutive threshold and idle compacts are bounded until occupancy", () => { + const governor = createCompactionGovernor(() => {}); + const echo = LEGACY_COMPACT_SPACER_TEXT; + governor.noteInferenceDone(inferenceDone(overThreshold, echo), tenTurns); + expect(governor.interceptActions(toolDone(), inferAction, capabilities)).not.toBeNull(); + + governor.noteInferenceDone(inferenceDone(overThreshold, echo), tenTurns); + governor.noteInferenceDone(inferenceDone(overThreshold + resumeDelta, echo), tenTurns); + expect(governor.interceptActions(toolDone(), inferAction, capabilities)).not.toBeNull(); + + governor.noteInferenceDone(inferenceDone(overThreshold + resumeDelta, echo), tenTurns); + governor.noteInferenceDone(inferenceDone(overThreshold + 2 * resumeDelta, echo), tenTurns); + expect(governor.interceptActions(toolDone(), inferAction, capabilities)).toBeNull(); + governor.noteIdleTurn(inferenceDone(overThreshold + 2 * resumeDelta, echo), [ + { type: "reply", content: "done" }, + ]); + expect(governor.interceptIdleContinuation(emptyMessage(), capabilities)).toBeNull(); + + governor.noteInferenceDone( + inferenceDone(overThreshold + 3 * resumeDelta, "real work"), + tenTurns, + ); + expect(governor.interceptActions(toolDone(), inferAction, capabilities)).toBeNull(); + + governor.noteInferenceDone(inferenceDoneWithTools(overThreshold + 4 * resumeDelta), tenTurns); + expect(governor.interceptActions(toolDone(), inferAction, capabilities)).not.toBeNull(); + }); + + test("spacer-echo terminal does not arm idle compact", () => { + let continuations = 0; + const governor = createCompactionGovernor(() => continuations++); + governor.noteInferenceDone(inferenceDone(overThreshold, LEGACY_COMPACT_SPACER_TEXT), tenTurns); + governor.noteIdleTurn(inferenceDone(overThreshold, LEGACY_COMPACT_SPACER_TEXT), [ + { type: "reply", content: LEGACY_COMPACT_SPACER_TEXT }, + ]); + expect(continuations).toBe(0); + governor.noteIdleTurn(inferenceDone(overThreshold, COMPACT_SPACER_TEXT), [ + { type: "reply", content: COMPACT_SPACER_TEXT }, + ]); + expect(continuations).toBe(0); + expect(governor.interceptIdleContinuation(emptyMessage(), capabilities)).toBeNull(); + + governor.noteIdleTurn(inferenceDone(overThreshold, "done"), [ + { type: "reply", content: "done" }, + ]); + expect(continuations).toBe(1); + }); }); diff --git a/src/agent/compaction.ts b/src/agent/compaction.ts index 4a348feb8..720ae5124 100644 --- a/src/agent/compaction.ts +++ b/src/agent/compaction.ts @@ -10,7 +10,12 @@ import { compactionThresholdFor, contextTokensFromUsage, } from "../provider/context-window.js"; -import { COMPACTOR_KEEP_RECENT_TURNS, compactorNoOpFloor } from "../session/compactor.js"; +import { + COMPACTOR_KEEP_RECENT_TURNS, + assistantTextIsCompactSpacerEcho, + compactorNoOpFloor, + isCompactSpacerEchoTurn, +} from "../session/compactor.js"; import { createContextEstimate, estimateOverheadTokens } from "./context-estimate.js"; import { onTurnBoundary } from "./reactor-events.js"; @@ -22,6 +27,12 @@ const COMPACTOR_NAME = "pruning-compactor"; // would spend a reactor cycle that shrinks nothing. const MIN_TURNS_TO_COMPACT = compactorNoOpFloor(COMPACTOR_KEEP_RECENT_TURNS); const MAX_OVERFLOW_RECOVERIES = 2; +// Last-ditch bound on compact→infer→compact when the post-compact infer never +// occupies the loop. Reset on tool-call occupancy or when a post-compact +// measurement lands at or under the high watermark (that infer is not itself +// a compact). Do not reset merely because assistant text ≠ spacer. Overflow +// recoveries (above) reset on any successful inference.done instead. +const MAX_CONSECUTIVE_THRESHOLD_COMPACTS = 2; // A compact action runs in its own reactor cycle, after which the reactor // idles until the next inbound event. Worker loops (sub-agents, the coding @@ -45,6 +56,7 @@ export function createCompactionGovernor( // operator question to answer). Distinct from postCompactInfer. let postCompactMeter = false; let overflowRecoveries = 0; + let consecutiveThresholdCompacts = 0; // Set whenever the arming decision fell back to the local estimate because // the provider omitted usage or reported zero, so callers rendering a meter // can flag the number as approximate instead of implying provider-grade @@ -90,11 +102,31 @@ export function createCompactionGovernor( awaitingPostCompactMeasurement = true; } + function atThresholdCompactCap(): boolean { + return consecutiveThresholdCompacts >= MAX_CONSECUTIVE_THRESHOLD_COMPACTS; + } + + function issueThresholdCompact(): void { + consecutiveThresholdCompacts++; + noteCompactIssued(); + } + + function isSpacerEchoTerminal(event: ReactorInboundEvent, actions: ReactorAction[]): boolean { + // Fail-closed only. ChatDirector owns spacer-echo completeness (nudge, then + // loop-protection / workflow / open-task rails). This just refuses to treat + // that incomplete wait or reply as an idle-compact pause. + if (event.type === "inference.done" && isCompactSpacerEchoTurn(event.turn)) return true; + return actions.some((a) => a.type === "reply" && assistantTextIsCompactSpacerEcho(a.content)); + } + function noteInferenceDone( event: Extract, turns: readonly ConversationTurn[], ): void { overflowRecoveries = 0; + if (event.turn.content.some((block) => block.type === "tool_call")) { + consecutiveThresholdCompacts = 0; + } if (requestContinuation === undefined) return; syncFromTurns(turns); lastModel = event.source?.model; @@ -109,6 +141,7 @@ export function createCompactionGovernor( } if (contextTokens <= compactionThresholdFor(lastModel)) { tokensAtLastCompact = undefined; + consecutiveThresholdCompacts = 0; } // Assign, don't OR: an under-threshold follow-up must disarm a sticky // pending left from an earlier over-threshold turn (e.g. after the @@ -138,9 +171,10 @@ export function createCompactionGovernor( if (event.type !== "tool.done") return null; if (!pending && !(usingEstimate && isOverThreshold(estimate.tokens))) return null; if (!actions.some((a) => a.type === "infer")) return null; + if (atThresholdCompactCap()) return null; pending = false; postCompactInfer = true; - noteCompactIssued(); + issueThresholdCompact(); requestContinuation?.(); return [ ...actions.filter((a) => a.type !== "infer"), @@ -154,7 +188,9 @@ export function createCompactionGovernor( // compact when it (or the operator's next message) arrives. function noteIdleTurn(event: ReactorInboundEvent, actions: ReactorAction[]): void { if (!pending || idlePending || requestContinuation === undefined) return; + if (atThresholdCompactCap()) return; if (!onTurnBoundary(event)) return; + if (isSpacerEchoTerminal(event, actions)) return; const terminal = actions.some((a) => a.type === "reply" || a.type === "wait") && !actions.some((a) => a.type === "infer" || a.type === "execute_tools"); @@ -168,6 +204,10 @@ export function createCompactionGovernor( capabilities: ReactorCapabilities, ): ReactorAction[] | null { if (!idlePending || event.type !== "message.received") return null; + if (atThresholdCompactCap()) { + idlePending = false; + return null; + } idlePending = false; pending = false; const content = typeof event.message.content === "string" ? event.message.content : ""; @@ -180,7 +220,7 @@ export function createCompactionGovernor( } else { postCompactMeter = true; } - noteCompactIssued(); + issueThresholdCompact(); requestContinuation?.(); return [capabilities.compact(COMPACTOR_NAME, "context-threshold")]; } diff --git a/src/agent/director.ts b/src/agent/director.ts index f5f2b4aae..95036d8f9 100644 --- a/src/agent/director.ts +++ b/src/agent/director.ts @@ -10,7 +10,11 @@ import type { ConversationTurn, RetryPolicy, } from "@intx/types/runtime"; -import { type SessionMetadata, type TaskBoundary } from "../session/compactor.js"; +import { + type SessionMetadata, + type TaskBoundary, + isCompactSpacerEchoTurn, +} from "../session/compactor.js"; import type { WorkflowCoordinator } from "../workflows/coordinator.js"; import { createCompactionGovernor, type CompactionGovernor } from "./compaction.js"; import { onTurnBoundary } from "./reactor-events.js"; @@ -72,11 +76,13 @@ function inferWithNudge( // // Assumes a terminal bare wait always means the turn is over. That holds for // every current wait path: DefaultDirector in conversational mode (the only -// mode ChatDirector uses) yields one only on an empty model turn, and its halt -// path already carries a reply; the compaction, workflow, and open-task -// rewrites either keep those terminals or replace them with an infer. -// A future wait that pauses mid-turn while expecting more work must not be -// settled here. +// mode ChatDirector uses) yields one only on an empty model turn; exhausted +// spacer-echo incompleteness uses the same wait so loop-protection, workflow, +// and open-task rails can rewrite it to infer first. This helper only settles +// a leftover wait into an empty reply. The halt path already carries a reply; +// compaction, workflow, and open-task rewrites either keep those terminals or +// replace them with an infer. A future wait that pauses mid-turn while +// expecting more work must not be settled here. function ensureCycleSettlesWithReply( actions: ReactorAction | ReactorAction[], capabilities: ReactorCapabilities, @@ -98,6 +104,9 @@ function ensureCycleSettlesWithReply( const MAX_OPEN_TASK_NUDGES = 3; const MAX_DECLINED_OPEN_TASK_NUDGES = 2; const MAX_INFERENCE_RECOVERIES = 2; +const MAX_SPACER_ECHO_NUDGES = 2; + +const SPACER_ECHO_NUDGE = "Continue the task. Do not repeat internal markers."; const IDLE_OPEN_TASK_NUDGE = "\n\nYou are ending your turn while tasks are still open (todo/doing). " + @@ -394,6 +403,7 @@ class ChatDirectorImpl extends DefaultDirector { private idleTerminationNudges = 0; private declinedTerminationNudges = 0; private inferenceRecoveries = 0; + private spacerEchoNudges = 0; private lastInferenceTurnHadContent = false; private operatorJustResponded = false; private tasks: Task[] = []; @@ -633,6 +643,7 @@ class ChatDirectorImpl extends DefaultDirector { this.idleTerminationNudges = 0; this.declinedTerminationNudges = 0; this.inferenceRecoveries = 0; + this.spacerEchoNudges = 0; this.toolOnlyStreak = 0; this.toolOnlyNudgeFired = false; this.pendingToolOnlyNudge = false; @@ -682,9 +693,10 @@ class ChatDirectorImpl extends DefaultDirector { if (onTurnBoundary(event)) { this.turnCount++; const hasToolCalls = event.turn.content.some((b) => b.type === "tool_call"); - const hasText = event.turn.content.some( - (b) => b.type === "text" && typeof b.text === "string" && b.text.length > 0, - ); + const hasText = + event.turn.content.some( + (b) => b.type === "text" && typeof b.text === "string" && b.text.length > 0, + ) && !isCompactSpacerEchoTurn(event.turn); this.lastInferenceTurnHadContent = hasToolCalls || hasText; // toolOnlyStreak is narration-sensitive: any turn with text clears it @@ -710,7 +722,12 @@ class ChatDirectorImpl extends DefaultDirector { if (hasToolCalls) { this.workflowIdleTurns = 0; } else { - this.workflowIdleTurns++; + // Echo-nudge cycles are incompleteness, not a contentful idle beat. + // Count them only after the echo budget is spent so the step-nudge + // rail still has its three turns before the stuck reply. + const spacerEchoStillNudging = + isCompactSpacerEchoTurn(event.turn) && this.spacerEchoNudges < MAX_SPACER_ECHO_NUDGES; + if (!spacerEchoStillNudging) this.workflowIdleTurns++; } } for (const block of event.turn.content) { @@ -795,8 +812,20 @@ class ChatDirectorImpl extends DefaultDirector { this.compaction.noteInferenceDone(event, turns); } + if (event.type === "inference.done" && isCompactSpacerEchoTurn(event.turn)) { + if (this.spacerEchoNudges < MAX_SPACER_ECHO_NUDGES) { + this.spacerEchoNudges++; + return inferWithNudge(capabilities, SPACER_ECHO_NUDGE); + } + } + const base = await super.decide(event, state, capabilities); - const baseActions = Array.isArray(base) ? base : [base]; + let baseActions = Array.isArray(base) ? base : [base]; + const spacerEchoExhausted = + event.type === "inference.done" && isCompactSpacerEchoTurn(event.turn); + if (spacerEchoExhausted) { + baseActions = baseActions.map((a) => (a.type === "reply" ? capabilities.wait() : a)); + } this.compaction.noteIdleTurn(event, baseActions); const compacted = this.compaction.interceptActions(event, baseActions, capabilities); @@ -816,12 +845,11 @@ class ChatDirectorImpl extends DefaultDirector { const coordinator = this.workflowCoordinator; if (coordinator?.isActive() && !coordinator.currentStepIsGate()) { - const actions = Array.isArray(base) ? base : [base]; - const hasTerminal = actions.some((a) => a.type === "wait" || a.type === "reply"); - if (hasTerminal && this.lastInferenceTurnHadContent) { + const hasTerminal = baseActions.some((a) => a.type === "wait" || a.type === "reply"); + if (hasTerminal && (this.lastInferenceTurnHadContent || spacerEchoExhausted)) { if (this.operatorJustResponded) { this.operatorJustResponded = false; - return base; + return baseActions; } if (this.workflowIdleTurns >= 3) { if (hasActiveTasks(this.tasks)) this.logTerminationWithOpenTasks("workflow-idle-stall"); @@ -840,7 +868,7 @@ class ChatDirectorImpl extends DefaultDirector { `\n\nYou have not yet completed this workflow step. ` + `If this step is complete, ${stepClause}. ` + `Otherwise continue working with tools.`; - const passThrough = actions.filter( + const passThrough = baseActions.filter( (a): a is Exclude => a.type !== "wait" && a.type !== "reply", ); @@ -872,7 +900,7 @@ class ChatDirectorImpl extends DefaultDirector { } } - return base; + return baseActions; } } diff --git a/src/context-compactor.test.ts b/src/context-compactor.test.ts index 0e1f0a327..e28c3e19b 100644 --- a/src/context-compactor.test.ts +++ b/src/context-compactor.test.ts @@ -8,6 +8,9 @@ import { buildLLMTurnSummary, COMPACTED_PREFIX, COMPACT_SPACER_TEXT, + LEGACY_COMPACT_SPACER_TEXT, + HARNESS_COMPACT_SPACER_MODEL, + isHarnessCompactSpacer, type SessionMetadata, } from "./session/compactor.js"; import { createModelSummarizer } from "./session/summarizer.js"; @@ -627,6 +630,94 @@ describe("createPruningCompactor — prefix-stable summaries (CL-6914)", () => { ).toBe(true); }); + test("harness spacer is stamped with the reserved producer id and a visible sentinel", async () => { + const compactor = createPruningCompactor({ keepRecentTurns: 2, summaryMaxChars: 500 }); + const output1 = (await compactor.apply(grow([], 16, "round1"), mockStrategyCtx)).output; + const output2 = (await compactor.apply(grow(output1, 16, "round2"), mockStrategyCtx)).output; + const spacer = output2.find(isHarnessCompactSpacer); + expect(spacer).toBeDefined(); + expect(spacer!.model).toBe(HARNESS_COMPACT_SPACER_MODEL); + expect(firstText(spacer!)).toBe(COMPACT_SPACER_TEXT); + expect(firstText(spacer!)).not.toBe(LEGACY_COMPACT_SPACER_TEXT); + expect(COMPACT_SPACER_TEXT).not.toBe(LEGACY_COMPACT_SPACER_TEXT); + }); + + test("frozen prefix does not absorb a model-emitted spacer", async () => { + const compactor = createPruningCompactor({ keepRecentTurns: 2, summaryMaxChars: 500 }); + const output1 = (await compactor.apply(grow([], 16, "round1"), mockStrategyCtx)).output; + const summary = output1.find((t) => firstText(t).startsWith(COMPACTED_PREFIX)); + expect(summary).toBeDefined(); + const echo = makeTurn({ + role: "assistant", + model: "omen-alpha", + content: [{ type: "text", text: LEGACY_COMPACT_SPACER_TEXT }], + }); + const output2 = (await compactor.apply(grow([summary!, echo], 16, "round2"), mockStrategyCtx)) + .output; + + let frozenLen = 0; + while ( + frozenLen < output2.length && + firstText(output2[frozenLen]!).startsWith(COMPACTED_PREFIX) + ) { + frozenLen++; + if (frozenLen < output2.length && isHarnessCompactSpacer(output2[frozenLen]!)) frozenLen++; + } + expect(output2.slice(0, frozenLen)).not.toContain(echo); + expect(isHarnessCompactSpacer(echo)).toBe(false); + const harness = output2.find(isHarnessCompactSpacer); + expect(harness).toBeDefined(); + expect(harness!.model).toBe(HARNESS_COMPACT_SPACER_MODEL); + expect(firstText(harness!)).toBe(COMPACT_SPACER_TEXT); + expect(harness).not.toBe(echo); + }); + + test("frozen prefix does not absorb a model-stamped new-sentinel echo", async () => { + const compactor = createPruningCompactor({ keepRecentTurns: 2, summaryMaxChars: 500 }); + const output1 = (await compactor.apply(grow([], 16, "round1"), mockStrategyCtx)).output; + const summary = output1.find((t) => firstText(t).startsWith(COMPACTED_PREFIX)); + expect(summary).toBeDefined(); + const echo = makeTurn({ + role: "assistant", + model: "omen-alpha", + content: [{ type: "text", text: COMPACT_SPACER_TEXT }], + }); + const output2 = (await compactor.apply(grow([summary!, echo], 16, "round2"), mockStrategyCtx)) + .output; + + let frozenLen = 0; + while ( + frozenLen < output2.length && + firstText(output2[frozenLen]!).startsWith(COMPACTED_PREFIX) + ) { + frozenLen++; + if (frozenLen < output2.length && isHarnessCompactSpacer(output2[frozenLen]!)) frozenLen++; + } + expect(output2.slice(0, frozenLen)).not.toContain(echo); + expect(isHarnessCompactSpacer(echo)).toBe(false); + const harness = output2.find(isHarnessCompactSpacer); + expect(harness).toBeDefined(); + expect(harness!.model).toBe(HARNESS_COMPACT_SPACER_MODEL); + expect(firstText(harness!)).toBe(COMPACT_SPACER_TEXT); + expect(harness).not.toBe(echo); + }); + + test("legacy harness spacer without model still freezes", async () => { + const compactor = createPruningCompactor({ keepRecentTurns: 2, summaryMaxChars: 500 }); + const output1 = (await compactor.apply(grow([], 16, "round1"), mockStrategyCtx)).output; + const summary = output1.find((t) => firstText(t).startsWith(COMPACTED_PREFIX)); + expect(summary).toBeDefined(); + const legacySpacer = makeTurn({ + role: "assistant", + content: [{ type: "text", text: LEGACY_COMPACT_SPACER_TEXT }], + }); + const grown = grow([summary!, legacySpacer], 16, "round2"); + const output2 = (await compactor.apply(grown, mockStrategyCtx)).output; + expect(output2[0]).toBe(summary); + expect(output2[1]).toBe(legacySpacer); + expect(isHarnessCompactSpacer(legacySpacer)).toBe(true); + }); + test("empty-fold keep-set returns the input unchanged", async () => { const compactor = createPruningCompactor({ keepRecentTurns: 1, diff --git a/src/director.test.ts b/src/director.test.ts index 54557d18c..5d62fa7a2 100644 --- a/src/director.test.ts +++ b/src/director.test.ts @@ -3,7 +3,12 @@ import { createChatDirector, askOperatorDefinition } from "./agent/director.js"; import { createAgentToolset } from "./agent/tools.js"; import { advertisedTools, createActivatedToolTracker } from "./agent/tool-search.js"; import { createPermissionGate } from "./permission/gate.js"; -import { COMPACTOR_KEEP_RECENT_TURNS, compactorNoOpFloor } from "./session/compactor.js"; +import { + COMPACTOR_KEEP_RECENT_TURNS, + COMPACT_SPACER_TEXT, + LEGACY_COMPACT_SPACER_TEXT, + compactorNoOpFloor, +} from "./session/compactor.js"; import type { SessionMetadata, TaskBoundary } from "./session/compactor.js"; import { validateActions, type ExtendedInferenceOptions } from "@intx/inference"; import type { @@ -1068,3 +1073,155 @@ describe("transient nudges", () => { expect(options?.systemPrompt).toBe("stable-base"); }); }); + +describe("chatDirector spacer echo", () => { + const longState = { + turns: Array.from({ length: compactorNoOpFloor(COMPACTOR_KEEP_RECENT_TURNS) + 1 }, () => ({ + role: "user", + content: [], + timestamp: 0, + })), + } as unknown as ReactorState; + + function spacerInferenceDone(text: string): ReactorInboundEvent { + return { + type: "inference.done", + turn: { + role: "assistant", + model: "omen-alpha", + timestamp: 0, + content: [{ type: "text", text }], + }, + usage: { input: 999_999, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 }, + source: { model: "omen-alpha" }, + } as unknown as ReactorInboundEvent; + } + + function messageReceived(content: string): ReactorInboundEvent { + return { + type: "message.received", + message: { role: "user", content }, + } as unknown as ReactorInboundEvent; + } + + test("spacer-only assistant reply is not a finished turn", async () => { + for (const text of [LEGACY_COMPACT_SPACER_TEXT, COMPACT_SPACER_TEXT]) { + const director = createChatDirector("base", [], { onTasksChange: () => {} }); + const actions = actionsArray( + await director.decide(spacerInferenceDone(text), mockState, mockCapabilities), + ); + expect(actions.some((a) => a.type === "infer")).toBe(true); + expect(actions.some((a) => a.type === "reply" && "content" in a && a.content === text)).toBe( + false, + ); + } + }); + + test("spacer-echo does not arm idle compact, including after the nudge cap", async () => { + let continuations = 0; + const director = createChatDirector("base", [], { + onTasksChange: () => {}, + requestContinuation: () => { + continuations++; + }, + }); + for (let i = 0; i < 2; i++) { + const nudged = actionsArray( + await director.decide( + spacerInferenceDone(LEGACY_COMPACT_SPACER_TEXT), + longState, + mockCapabilities, + ), + ); + expect(nudged.some((a) => a.type === "infer")).toBe(true); + expect(continuations).toBe(0); + } + const settled = actionsArray( + await director.decide(spacerInferenceDone(COMPACT_SPACER_TEXT), longState, mockCapabilities), + ); + expect(settled.some((a) => a.type === "infer")).toBe(false); + expect(settled.some((a) => a.type === "reply" && "content" in a && a.content === "")).toBe( + true, + ); + expect(continuations).toBe(0); + }); + + test("echo-nudge cap is two then empty settle, and resets on message.received", async () => { + const director = createChatDirector("base", [], { onTasksChange: () => {} }); + for (let i = 0; i < 2; i++) { + const nudged = actionsArray( + await director.decide( + spacerInferenceDone(LEGACY_COMPACT_SPACER_TEXT), + mockState, + mockCapabilities, + ), + ); + expect(nudged.some((a) => a.type === "infer")).toBe(true); + expect(nudged.some((a) => a.type === "reply")).toBe(false); + } + const exhausted = actionsArray( + await director.decide(spacerInferenceDone(COMPACT_SPACER_TEXT), mockState, mockCapabilities), + ); + expect(exhausted.some((a) => a.type === "infer")).toBe(false); + expect(exhausted.some((a) => a.type === "reply" && "content" in a && a.content === "")).toBe( + true, + ); + + await director.decide(messageReceived("keep going"), mockState, mockCapabilities); + const afterReset = actionsArray( + await director.decide( + spacerInferenceDone(LEGACY_COMPACT_SPACER_TEXT), + mockState, + mockCapabilities, + ), + ); + expect(afterReset.some((a) => a.type === "infer")).toBe(true); + }); + + test("after echo-cap with open tasks, falls through to open-task rails", async () => { + const director = createChatDirector("base", [], { onTasksChange: () => {} }); + await director.decide( + makeInferenceDoneEvent([ + { + id: "mt", + name: "manage_tasks", + args: { action: "create", tasks: [{ id: "t1", title: "work", status: "doing" }] }, + }, + ]), + mockState, + mockCapabilities, + ); + for (let i = 0; i < 2; i++) { + const nudged = actionsArray( + await director.decide( + spacerInferenceDone(LEGACY_COMPACT_SPACER_TEXT), + mockState, + mockCapabilities, + ), + ); + expect(nudged.some((a) => a.type === "infer")).toBe(true); + } + const afterCap = actionsArray( + await director.decide(spacerInferenceDone(COMPACT_SPACER_TEXT), mockState, mockCapabilities), + ); + expect(afterCap.some((a) => a.type === "infer")).toBe(true); + expect(afterCap.some((a) => a.type === "reply" && "content" in a && a.content === "")).toBe( + false, + ); + for (let i = 0; i < 2; i++) { + const nudged = actionsArray( + await director.decide( + spacerInferenceDone(COMPACT_SPACER_TEXT), + mockState, + mockCapabilities, + ), + ); + expect(nudged.some((a) => a.type === "infer")).toBe(true); + } + const exhausted = actionsArray( + await director.decide(spacerInferenceDone(COMPACT_SPACER_TEXT), mockState, mockCapabilities), + ); + expect(exhausted.some((a) => a.type === "infer")).toBe(false); + expect(exhausted.some((a) => a.type === "reply")).toBe(true); + }); +}); diff --git a/src/provider/replay-sanitizer.test.ts b/src/provider/replay-sanitizer.test.ts index 0d3f13aef..cd92f40a5 100644 --- a/src/provider/replay-sanitizer.test.ts +++ b/src/provider/replay-sanitizer.test.ts @@ -14,6 +14,12 @@ import { THINKING_ONLY_OMITTED, withReplaySanitizer, } from "./replay-sanitizer.js"; +import { + COMPACT_SPACER_TEXT, + COMPACTED_PREFIX, + HARNESS_COMPACT_SPACER_MODEL, + isHarnessCompactSpacer, +} from "../session/compactor.js"; const GROK_SIGNATURE = "grok-opaque-signature-blob"; @@ -400,4 +406,32 @@ describe("withReplaySanitizer", () => { expect(body.input.some((item) => item.type === "reasoning")).toBe(true); expect(body.input.some((item) => item.type === "function_call")).toBe(true); }); + + it("keeps a harness compact spacer as a role-alternating assistant turn", () => { + const spacer: ConversationTurn = { + role: "assistant", + model: HARNESS_COMPACT_SPACER_MODEL, + content: [{ type: "text", text: COMPACT_SPACER_TEXT }], + timestamp: 2, + }; + const turns = sanitizeReplayTurns( + [ + { + role: "user", + content: [{ type: "text", text: `${COMPACTED_PREFIX} earlier` }], + timestamp: 1, + }, + spacer, + { + role: "user", + content: [{ type: "text", text: `${COMPACTED_PREFIX} later` }], + timestamp: 3, + }, + ], + "claude-opus-4", + ); + expect(turns.map((t) => t.role)).toEqual(["user", "assistant", "user"]); + expect(turns[1]?.content).toEqual([{ type: "text", text: COMPACT_SPACER_TEXT }]); + expect(isHarnessCompactSpacer(turns[1]!)).toBe(true); + }); }); diff --git a/src/session/compactor.ts b/src/session/compactor.ts index 32036dc0d..7605aaa23 100644 --- a/src/session/compactor.ts +++ b/src/session/compactor.ts @@ -221,7 +221,12 @@ export const COMPACTED_PREFIX = "[Compacted prior context]"; // Inserted between a frozen prefix that ends on a user summary and a newly // appended user summary so the assembled history stays role-alternating. -export const COMPACT_SPACER_TEXT = "[compaction]"; +// Visible, non-format (not Unicode Cf) sentinel so Chat Completions adapters +// keep a non-empty assistant turn. Identity is the reserved producer id on +// `compactSpacerTurn`, not this text and not a missing `model` field. +export const COMPACT_SPACER_TEXT = "[compact]"; +export const LEGACY_COMPACT_SPACER_TEXT = "[compaction]"; +export const HARNESS_COMPACT_SPACER_MODEL = "harness"; const DEFAULT_COMPACTOR_CONFIG: CompactorConfig = { keepRecentTurns: COMPACTOR_KEEP_RECENT_TURNS, @@ -699,15 +704,45 @@ function firstTextBlock(turn: ConversationTurn): string | undefined { return undefined; } +function joinedTextBlocks(turn: ConversationTurn): string { + let out = ""; + for (const block of turn.content) { + if (block.type === "text") out += block.text; + } + return out; +} + +function isCompactSpacerSentinel(text: string): boolean { + return text === COMPACT_SPACER_TEXT || text === LEGACY_COMPACT_SPACER_TEXT; +} + +export function assistantTextIsCompactSpacerEcho(text: string): boolean { + return isCompactSpacerSentinel(text.trim()); +} + +export function isCompactSpacerEchoTurn(turn: ConversationTurn): boolean { + for (const block of turn.content) { + if (block.type === "tool_call") return false; + } + return assistantTextIsCompactSpacerEcho(joinedTextBlocks(turn)); +} + function isCompactedSummaryTurn(turn: ConversationTurn): boolean { if (turn.role !== "user") return false; const text = firstTextBlock(turn); return text !== undefined && text.startsWith(COMPACTED_PREFIX); } -function isCompactSpacerTurn(turn: ConversationTurn): boolean { +// Harness spacers stamp `model: "harness"`. Missing `model` is unattributed +// (replay sanitizer), except persisted `[compaction]` spacers from before +// producer-id stamping, which still freeze. Model-produced copies always +// carry a real model id and must not enter the frozen prefix. +export function isHarnessCompactSpacer(turn: ConversationTurn): boolean { if (turn.role !== "assistant") return false; - return firstTextBlock(turn) === COMPACT_SPACER_TEXT; + const text = firstTextBlock(turn); + if (text === undefined || !isCompactSpacerSentinel(text)) return false; + if (turn.model === HARNESS_COMPACT_SPACER_MODEL) return true; + return turn.model === undefined && text === LEGACY_COMPACT_SPACER_TEXT; } // Leading run of prior summaries plus the spacers between them. Walks from @@ -718,7 +753,7 @@ function frozenPrefixLength(turns: readonly ConversationTurn[]): number { let i = 0; while (i < turns.length && isCompactedSummaryTurn(turns[i]!)) { i++; - if (i < turns.length && isCompactSpacerTurn(turns[i]!)) i++; + if (i < turns.length && isHarnessCompactSpacer(turns[i]!)) i++; } return i; } @@ -728,6 +763,7 @@ function compactSpacerTurn(timestamp: number): ConversationTurn { role: "assistant", content: [{ type: "text", text: COMPACT_SPACER_TEXT }], timestamp, + model: HARNESS_COMPACT_SPACER_MODEL, }; } @@ -736,7 +772,7 @@ export function createPruningCompactor(config: Partial = {}): C return { name: "pruning-compactor", - version: "1.4.0", + version: "1.4.1", async apply( turns: ConversationTurn[], _ctx: StrategyContext, diff --git a/tests/unit/workflows-director.test.ts b/tests/unit/workflows-director.test.ts index 534aa8d0e..a8d6c42b1 100644 --- a/tests/unit/workflows-director.test.ts +++ b/tests/unit/workflows-director.test.ts @@ -12,6 +12,7 @@ import { WorkflowRuntime } from "../../src/workflows/runtime.js"; import { WorkflowCoordinator } from "../../src/workflows/coordinator.js"; import type { CapabilityMap } from "../../src/workflows/capabilities.js"; import type { Workflow } from "../../src/workflows/types.js"; +import { COMPACT_SPACER_TEXT, LEGACY_COMPACT_SPACER_TEXT } from "../../src/session/compactor.js"; const usage: TokenUsage = { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 }; @@ -372,3 +373,24 @@ test("auto-continuation falls back after 3 consecutive text-only turns", async ( expect(hasInfer(result)).toBe(false); expect(actions.some((a) => a.type === "wait" || a.type === "reply")).toBe(true); }); + +test("after spacer echo-cap a non-gate workflow step does not empty-settle", async () => { + const runtime = new WorkflowRuntime(emptyCaps, (n) => (n === "flow" ? flow : undefined)); + runtime.start(flow); + const coordinator = new WorkflowCoordinator(runtime); + const director = createChatDirector("BASE", [], { + onTasksChange: () => {}, + workflowCoordinator: coordinator, + }); + const caps = makeCapabilities(); + + for (let i = 0; i < 2; i++) { + const nudged = await director.decide(textTurn(LEGACY_COMPACT_SPACER_TEXT), state, caps); + expect(hasInfer(nudged)).toBe(true); + } + const afterCap = await director.decide(textTurn(COMPACT_SPACER_TEXT), state, caps); + const actions = Array.isArray(afterCap) ? afterCap : [afterCap]; + expect(actions.some((a) => a.type === "reply" && "content" in a && a.content === "")).toBe(false); + expect(hasInfer(afterCap)).toBe(true); + expect(ephemeralNudgeText(afterCap)).toContain("workflow step"); +});