diff options
| -rw-r--r-- | packages/kernel/src/contracts/runtime.ts | 27 | ||||
| -rw-r--r-- | packages/kernel/src/runtime/run-turn.test.ts | 329 | ||||
| -rw-r--r-- | packages/kernel/src/runtime/run-turn.ts | 21 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/orchestrator.test.ts | 371 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/orchestrator.ts | 409 |
5 files changed, 1028 insertions, 129 deletions
diff --git a/packages/kernel/src/contracts/runtime.ts b/packages/kernel/src/contracts/runtime.ts index 71d2211..2bab47a 100644 --- a/packages/kernel/src/contracts/runtime.ts +++ b/packages/kernel/src/contracts/runtime.ts @@ -142,6 +142,33 @@ export interface RunTurnInput { readonly onStepComplete?: (messages: readonly ChatMessage[]) => Promise<void> | void; /** + * Optional. Called by the runtime at each tool-result boundary — after a + * step that produced tool calls has been finalized (`onStepComplete` has run + * and any `drainSteering` messages have been appended), before the next step + * begins — giving the caller a chance to REPLACE the running message + * history. The caller returns a new message list which the runtime adopts + * as its working history for all subsequent steps, or `undefined`/an empty + * array to keep the history unchanged. + * + * The runtime receives the step's token `usage` (so the caller can check a + * threshold against the model's context window) and the current `messages` + * (the full prompt the next step would otherwise see). Generic and + * feature-agnostic: the runtime calls it and adopts whatever message list it + * returns — it names no feature, owns no threshold policy, and performs no + * I/O, mirroring `drainSteering`. The shell uses this to compact the history + * in-flight when the context window nears capacity, so a long-running turn + * (e.g. left overnight) does not run out of context mid-turn: it summarizes + * the old history and continues with the summary + recent messages. Only + * invoked when a step PRODUCED tool calls (there is a "next step" to compact + * before); a step that ends without tool calls ends the turn, so there is no + * boundary to compact at. Injected (not ambient) so the kernel stays pure. + */ + readonly onStepBoundary?: (ctx: { + readonly stepUsage: Usage; + readonly messages: readonly ChatMessage[]; + }) => Promise<readonly ChatMessage[] | undefined> | (readonly ChatMessage[] | undefined); + + /** * Optional injected retry strategy for retryable provider errors (e.g. HTTP * 429 / 5xx "overloaded"). When omitted, a retryable error ends the step * exactly as before (backward-compatible). When provided, the runtime wraps diff --git a/packages/kernel/src/runtime/run-turn.test.ts b/packages/kernel/src/runtime/run-turn.test.ts index 8d20975..90357b1 100644 --- a/packages/kernel/src/runtime/run-turn.test.ts +++ b/packages/kernel/src/runtime/run-turn.test.ts @@ -2899,6 +2899,335 @@ describe("runTurn", () => { }); }); + // ── onStepBoundary (in-flight history replacement) ────────────────────── + // + // PURE tests: a capturing provider + a fake `onStepBoundary` that returns a + // replacement message list (or void). The kernel must adopt the returned + // list as its working history for subsequent steps. ZERO mocks of + // `@dispatch/*` — the hook is injected like `drainSteering`. + + describe("onStepBoundary", () => { + it("returns a replacement → next step's provider input is the replaced history (old messages dropped)", async () => { + const compactedHistory: ChatMessage[] = [ + { role: "system", chunks: [{ type: "text", text: "SUMMARY" }] }, + { role: "user", chunks: [{ type: "text", text: "recent" }] }, + ]; + let boundaryCallCount = 0; + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 999, outputTokens: 1 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: ({ messages }) => { + boundaryCallCount++; + // Sanity: the kernel passes the full running history at the boundary + // (user, assistant tool-call, tool result) — 3 messages. + expect(messages.length).toBe(3); + return compactedHistory; + }, + }); + + expect(boundaryCallCount).toBe(1); + expect(capturedMessages).toHaveLength(2); + // The second step must see ONLY the replaced history — the original + // 3 messages are gone, replaced by [SUMMARY, recent]. + const secondStepMessages = capturedMessages[1] ?? []; + expect(secondStepMessages).toHaveLength(2); + expect(secondStepMessages[0]).toEqual(compactedHistory[0]); + expect(secondStepMessages[1]).toEqual(compactedHistory[1]); + }); + + it("returns undefined → history unchanged (byte-identical to omitting the hook)", async () => { + let boundaryCallCount = 0; + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: () => { + boundaryCallCount++; + return undefined; + }, + }); + + expect(boundaryCallCount).toBe(1); + const secondStepMessages = capturedMessages[1] ?? []; + // user, assistant(tool-call), tool-result — unchanged. + expect(secondStepMessages).toHaveLength(3); + expect(secondStepMessages[0]?.role).toBe("user"); + expect(secondStepMessages[1]?.role).toBe("assistant"); + expect(secondStepMessages[2]?.role).toBe("tool"); + }); + + it("returns an empty array → treated as no replacement (history unchanged)", async () => { + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: () => [], + }); + + const secondStepMessages = capturedMessages[1] ?? []; + // Empty replacement is ignored — history unchanged. + expect(secondStepMessages).toHaveLength(3); + }); + + it("NOT called when a step has no tool calls (text-only turn) — there is no next step", async () => { + let boundaryCallCount = 0; + const provider = createFakeProvider([ + [ + { type: "text-delta", delta: "hello" }, + { type: "finish", reason: "stop" }, + ], + ]); + + await runTurn({ + provider, + messages: [userMessage], + tools: [], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: () => { + boundaryCallCount++; + return undefined; + }, + }); + + expect(boundaryCallCount).toBe(0); + }); + + it("multiple tool-call steps → called once per tool-call step, AFTER drainSteering (steering present in messages)", async () => { + let boundaryCallCount = 0; + const steeringMessage: ChatMessage = { + role: "user", + chunks: [{ type: "text", text: "steer!" }], + }; + let sawSteeringAtBoundary = false; + + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "tool-call", toolCallId: "tc2", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + drainSteering: () => [steeringMessage], + onStepBoundary: ({ messages }) => { + boundaryCallCount++; + // drainSteering runs BEFORE onStepBoundary, so the steering message + // must already be present in the boundary's `messages`. + if (messages.some((m) => m === steeringMessage)) { + sawSteeringAtBoundary = true; + } + return undefined; + }, + }); + + // Steps 0 and 1 produced tool calls → boundary once each. + // Step 2 (text-only) ends the turn → no boundary. Total = 2. + expect(boundaryCallCount).toBe(2); + expect(sawSteeringAtBoundary).toBe(true); + expect(capturedMessages).toHaveLength(3); + }); + + it("receives the step's usage as stepUsage", async () => { + const seenUsages: Array<{ inputTokens: number; outputTokens: number }> = []; + const { provider } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 4242, outputTokens: 17 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + onStepBoundary: ({ stepUsage }) => { + seenUsages.push({ + inputTokens: stepUsage.inputTokens, + outputTokens: stepUsage.outputTokens, + }); + return undefined; + }, + }); + + expect(seenUsages).toEqual([{ inputTokens: 4242, outputTokens: 17 }]); + }); + + it("omitted → strict no-op (byte-identical to before)", async () => { + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + const result = await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + // onStepBoundary omitted — must be a strict no-op. + }); + + const secondStepMessages = capturedMessages[1] ?? []; + expect(secondStepMessages).toHaveLength(3); + expect(result.messages).toHaveLength(3); + }); + + it("replacement adopted across multiple subsequent steps (history stays compacted)", async () => { + const compactedHistory: ChatMessage[] = [ + { role: "system", chunks: [{ type: "text", text: "SUMMARY" }] }, + ]; + let boundaryCallCount = 0; + const { provider, capturedMessages } = createCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "tool-call", toolCallId: "tc2", toolName: "echo", input: {} }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const tool = createFakeTool("echo", async () => ({ content: "echoed" })); + + await runTurn({ + provider, + messages: [userMessage], + tools: [tool], + dispatch: { maxConcurrent: 1, eager: false }, + conversationId: "conv-1", + turnId: "turn-1", + emit: () => {}, + // Compact only at the FIRST boundary; subsequent boundaries return void. + onStepBoundary: ({ messages }) => { + boundaryCallCount++; + if (boundaryCallCount === 1) { + // Pre-replacement: user + assistant(tool-call) + tool result = 3. + expect(messages.length).toBe(3); + return compactedHistory; + } + return undefined; + }, + }); + + expect(boundaryCallCount).toBe(2); + // After the FIRST boundary compacts to [SUMMARY], step 1's provider call + // (captured[1]) sees ONLY [SUMMARY] — the original user message is GONE, + // proving the replacement took effect before the next step. + const secondStepMessages = capturedMessages[1] ?? []; + expect(secondStepMessages).toHaveLength(1); + expect(secondStepMessages[0]).toEqual(compactedHistory[0]); + // Step 1 then produced tc2 + a tool result (pushed onto messages), so + // step 2's provider call (captured[2]) sees [SUMMARY, assistant(tc2), + // tool-result] = 3 — the compaction PERSISTED across step 1→2 (the + // SUMMARY is still heading the history; the original user message + // never came back). + const thirdStepMessages = capturedMessages[2] ?? []; + expect(thirdStepMessages).toHaveLength(3); + expect(thirdStepMessages[0]).toEqual(compactedHistory[0]); + expect(thirdStepMessages[1]?.role).toBe("assistant"); + expect(thirdStepMessages[2]?.role).toBe("tool"); + }); + }); + // ── Retry with backoff ────────────────────────────────────────────────── // // PURE tests: a fake `sleep` (records calls, resolves instantly, can abort diff --git a/packages/kernel/src/runtime/run-turn.ts b/packages/kernel/src/runtime/run-turn.ts index 3460033..76e2edf 100644 --- a/packages/kernel/src/runtime/run-turn.ts +++ b/packages/kernel/src/runtime/run-turn.ts @@ -725,6 +725,27 @@ export async function runTurn(input: RunTurnInput): Promise<RunTurnResult> { for (const msg of steering) { messages.push(msg); } + + // In-flight history replacement boundary: give the caller a chance + // to swap the running message history before the next step (e.g. to + // compact it when the context window nears capacity). The kernel names + // no feature and performs no I/O — it calls the callback and adopts + // whatever message list it returns, mirroring `drainSteering`. When the + // caller returns a list, the runtime replaces its working history with + // it (in place); `undefined`/empty leaves it unchanged. Only reached + // when there IS a next step (tool calls were produced). + if (input.onStepBoundary !== undefined) { + const replacement = await input.onStepBoundary({ + stepUsage: stepResult.usage, + messages, + }); + if (replacement !== undefined && replacement.length > 0) { + messages.length = 0; + for (const msg of replacement) { + messages.push(msg); + } + } + } } } } finally { diff --git a/packages/session-orchestrator/src/orchestrator.test.ts b/packages/session-orchestrator/src/orchestrator.test.ts index 400ca0b..f61ff21 100644 --- a/packages/session-orchestrator/src/orchestrator.test.ts +++ b/packages/session-orchestrator/src/orchestrator.test.ts @@ -22,8 +22,10 @@ import { createLogger, runTurn } from "@dispatch/kernel"; import type { SystemPromptService } from "@dispatch/system-prompt"; import { describe, expect, it } from "vitest"; import { + type ConversationCompactedPayload, type ConversationOpenedPayload, type ConversationStatusChangedPayload, + conversationCompacted, createCompactionService, createSessionOrchestrator, createWarmService, @@ -49,6 +51,7 @@ function createInMemoryStore(): ConversationStore & { const effortData = new Map<string, ReasoningEffort>(); const modelData = new Map<string, string>(); const workspaceIdData = new Map<string, string>(); + const compactPercentData = new Map<string, number>(); // Track conversations that have a meta row. In the real store, append, // setWorkspaceId, setConversationStatus, setConversationTitle, and // setCompactedFrom all create a minimal meta row on first contact. @@ -158,10 +161,12 @@ function createInMemoryStore(): ConversationStore & { knownConversations.add(conversationId); data.set(conversationId, [...messages]); }, - async getCompactPercent() { - return null; + async getCompactPercent(conversationId) { + return compactPercentData.get(conversationId) ?? null; + }, + async setCompactPercent(conversationId, percent) { + compactPercentData.set(conversationId, percent); }, - async setCompactPercent() {}, async forkHistory(_sourceId, targetId) { knownConversations.add(targetId); }, @@ -3985,6 +3990,366 @@ describe("system prompt: compaction flow", () => { }); }); +describe("in-flight compaction", () => { + // Seeds a conversation with `count` alternating user/assistant text messages + // so the history is long enough to compact (> DEFAULT_KEEP_LAST_N = 10). + function seedHistory( + store: ReturnType<typeof createInMemoryStore>, + conversationId: string, + count: number, + ): void { + const messages: ChatMessage[] = []; + for (let i = 0; i < count; i++) { + messages.push({ + role: i % 2 === 0 ? "user" : "assistant", + chunks: [{ type: "text", text: `seed message ${i}` }], + }); + } + store.data.set(conversationId, messages); + } + + // A provider whose `stream` serves a SCRIPT of per-call event lists, in + // order. Captures the messages passed to each call so a test can assert what + // the model saw at each step (incl. after in-flight compaction replaced it). + function createScriptedCapturingProvider(script: ProviderEvent[][]): { + provider: ProviderContract; + capturedMessages: ChatMessage[][]; + } { + const capturedMessages: ChatMessage[][] = []; + let callIndex = 0; + const provider: ProviderContract = { + id: "fake", + stream(messages) { + capturedMessages.push([...messages]); + const events = script[callIndex] ?? []; + callIndex++; + return (async function* () { + for (const event of events) { + yield event; + } + })(); + }, + }; + return { provider, capturedMessages }; + } + + function echoTool(): ToolContract { + return { + name: "echo", + description: "echo", + parameters: { type: "object" }, + execute: async () => ({ content: "echoed" }), + }; + } + + it("triggers when a step's usage exceeds the threshold: history is compacted mid-turn and the prompt continues with the summary", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-inflight", 15); // > keepLastN(10) → compactable + + // contextWindow 1000, default percent 85 → threshold 850. + // Step 0 emits a tool call + usage(inputTokens 900) → 910 > 850 → trigger. + // Then the compaction summary call, then step 1 ends the turn. + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 900, outputTokens: 10 } }, + { type: "finish", reason: "tool-calls" }, + ], + // Compaction summary call (performCompaction): + [ + { type: "text-delta", delta: "COMPACTED SUMMARY" }, + { type: "finish", reason: "stop" }, + ], + // Step 1 (post-compaction) — the turn CONTINUES: + [ + { type: "text-delta", delta: "all done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + const { events, onEvent } = collectEvents(); + + await orchestrator.handleMessage({ + conversationId: "conv-inflight", + text: "keep working overnight", + onEvent, + modelName: "test/model", + }); + + // 1) The conversationCompacted event fired mid-turn. + expect(compactedEvents).toHaveLength(1); + expect(compactedEvents[0]?.conversationId).toBe("conv-inflight"); + expect(compactedEvents[0]?.messagesSummarized).toBeGreaterThan(0); + expect(compactedEvents[0]?.messagesKept).toBe(10); + + // 2) The store history was replaced: it now begins with the system summary + // message, and the OLDEST seed messages are gone (summarized). The most + // recent messages are retained (keepLastN = 10), so some later seed + // messages may survive — that is correct. + const stored = store.data.get("conv-inflight") ?? []; + expect(stored.length).toBeGreaterThan(0); + expect(stored[0]?.role).toBe("system"); + expect(stored[0]?.chunks[0]).toMatchObject({ type: "text" }); + const firstText = (stored[0]?.chunks[0] as { text: string } | undefined)?.text ?? ""; + expect(firstText).toContain("COMPACTED SUMMARY"); + // The earliest seed messages were summarized away (not retained). + expect( + stored.some((m) => m.chunks.some((c) => c.type === "text" && c.text === "seed message 0")), + ).toBe(false); + + // 3) The turn CONTINUED after compaction: 3 provider calls happened + // (step 0, compaction summary, step 1) and the final assistant text + // was produced + persisted. + expect(capturedMessages).toHaveLength(3); + const step1Messages = capturedMessages[2] ?? []; + // Step 1 saw the COMPACTED history: it must start with the summary + // system message, NOT the original seed/user prefix. + expect(step1Messages[0]?.role).toBe("system"); + + const turnSealed = events.some((e) => e.type === "turn-sealed"); + expect(turnSealed).toBe(true); + }); + + it("does NOT trigger when the step usage is below the threshold (history unchanged, no event)", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-below", 15); + + // contextWindow 1000 → threshold 850. Step usage 100 < 850 → no trigger. + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 100, outputTokens: 5 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-below", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + // No compaction event, and only 2 provider calls (no summary call). + expect(compactedEvents).toHaveLength(0); + expect(capturedMessages).toHaveLength(2); + // The seed messages are still the start of history (uncompacted). + const stored = store.data.get("conv-below") ?? []; + expect(stored[0]?.chunks[0]).toMatchObject({ type: "text", text: "seed message 0" }); + }); + + it("does NOT trigger when auto-compact is disabled (compact percent = 0)", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-disabled", 15); + await store.setCompactPercent("conv-disabled", 0); + + // Usage would exceed the default threshold, but percent=0 disables it. + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 950, outputTokens: 10 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-disabled", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + expect(compactedEvents).toHaveLength(0); + expect(capturedMessages).toHaveLength(2); + }); + + it("does NOT trigger on a text-only turn (no tool calls → no next step → no boundary)", async () => { + const store = createInMemoryStore(); + seedHistory(store, "conv-textonly", 15); + + // Single text-only step with high usage — but no tool calls → the turn + // ends → there is no step boundary to compact at (post-seal handles it). + const { provider, capturedMessages } = createScriptedCapturingProvider([ + [ + { type: "text-delta", delta: "final answer" }, + { type: "usage", usage: { inputTokens: 950, outputTokens: 10 } }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-textonly", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + // No in-flight compaction (only 1 provider call — the turn ended). + expect(compactedEvents).toHaveLength(0); + expect(capturedMessages).toHaveLength(1); + }); + + it("the manual compaction SERVICE still refuses while a conversation is generating, but in-flight compaction runs anyway", async () => { + // This documents the two-path design: compact() (the service) guards on + // activeConversations and refuses mid-turn; the in-flight path bypasses + // that guard (it IS the mid-turn path) using performCompaction directly. + const store = createInMemoryStore(); + seedHistory(store, "conv-twopath", 15); + + const { provider } = createScriptedCapturingProvider([ + [ + { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} }, + { type: "usage", usage: { inputTokens: 900, outputTokens: 10 } }, + { type: "finish", reason: "tool-calls" }, + ], + [ + { type: "text-delta", delta: "SUMMARY" }, + { type: "finish", reason: "stop" }, + ], + [ + { type: "text-delta", delta: "done" }, + { type: "finish", reason: "stop" }, + ], + ]); + + const compactedEvents: ConversationCompactedPayload[] = []; + const activeConversations = new Set<string>(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [echoTool()], + applyToolsFilter: identityApplyToolsFilter, + resolveModel: () => ({ provider, model: "model" }), + resolveModelInfo: async () => ({ id: "test/model", contextWindow: 1000 }), + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + // The compaction SERVICE shares the orchestrator's activeConversations set; + // build it against the SAME set so the guard reflects reality. + const compactionService = createCompactionService( + { + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn, + emit: () => {}, + }, + activeConversations, + ); + + // Drive a turn that triggers in-flight compaction. We can't easily inspect + // activeConversations mid-turn, so we assert the observable contract: + // in-flight compaction produced an event (it ran WHILE active), and the + // store was compacted. + await orchestrator.handleMessage({ + conversationId: "conv-twopath", + text: "hi", + onEvent: () => {}, + modelName: "test/model", + }); + + expect(compactedEvents).toHaveLength(1); + const stored = store.data.get("conv-twopath") ?? []; + expect(stored[0]?.role).toBe("system"); + + // After the turn settles (idle), the manual service CAN compact (no longer + // active) — and it succeeds (history is compactable again only if long + // enough; here it is short post-compaction, so it reports too-short, which + // proves the service path is reachable and its guard is the ONLY reason it + // would have refused mid-turn). + const manual = await compactionService.compact("conv-twopath"); + // Post-compaction the history is short (summary + ~10 + turn tail) → the + // service reports an error (too short / threshold), NOT "generating". + expect("error" in manual).toBe(true); + if ("error" in manual) { + expect(manual.error).not.toBe("conversation is generating"); + } + }); +}); + describe("per-turn memory telemetry", () => { function capturingLogger(): { logger: Logger; records: LogRecord[] } { let id = 0; diff --git a/packages/session-orchestrator/src/orchestrator.ts b/packages/session-orchestrator/src/orchestrator.ts index aaf418a..6e5cfca 100644 --- a/packages/session-orchestrator/src/orchestrator.ts +++ b/packages/session-orchestrator/src/orchestrator.ts @@ -899,6 +899,79 @@ export function createSessionOrchestrator( ...(effectiveComputerId !== undefined ? { computerId: effectiveComputerId } : {}), ...(deps.now !== undefined ? { now: deps.now } : {}), ...(drainSteering !== undefined ? { drainSteering } : {}), + // In-flight compaction: at every tool-result boundary the kernel + // calls this with the step's usage + the running messages. When the + // context size exceeds the compact-percent threshold (percent of the + // model's context window), the old history is summarized and + // replaced with [summary, ...recent] — mid-turn, without stopping — + // so a long-running turn (e.g. left overnight) does not run out of + // context. This is DISTINCT from the post-seal auto-compact below: + // that one runs AFTER the turn ends (preparing the next turn) and + // refuses while the conversation is active; this one runs DURING the + // turn (saving the running turn) and uses the live step usage (not + // persisted metrics, which are only written at turn end). When the + // threshold is not exceeded, compaction is disabled (percent 0), or + // the model's context window is unknown, it returns void and the + // kernel keeps its history unchanged (a strict no-op). + onStepBoundary: async ({ stepUsage, messages }) => { + const stored = await deps.conversationStore.getCompactPercent(conversationId); + const percent = stored ?? DEFAULT_COMPACT_PERCENT; + if (percent <= 0) return; // auto-compact disabled + // contextSize mirrors the persisted definition: this step's + // inputTokens + outputTokens (the prompt the NEXT step would + // inherit, grown by this step's output). + const contextSize = stepUsage.inputTokens + stepUsage.outputTokens; + if (effectiveModelName === undefined || deps.resolveModelInfo === undefined) return; + const info = await deps.resolveModelInfo(effectiveModelName); + if (info?.contextWindow === undefined) return; + const threshold = Math.floor(info.contextWindow * (percent / 100)); + if (contextSize < threshold) return; // threshold not exceeded + + const keepLastN = DEFAULT_KEEP_LAST_N; + turnLogger?.info("compaction:in-flight", { + conversationId, + turnId, + contextSize, + threshold, + percent, + }); + const outcome = await performCompaction( + { + conversationStore: deps.conversationStore, + resolveProvider: deps.resolveProvider, + ...(deps.resolveModel !== undefined ? { resolveModel: deps.resolveModel } : {}), + ...(deps.resolveSystemPrompt !== undefined + ? { resolveSystemPrompt: deps.resolveSystemPrompt } + : {}), + ...(deps.resolveConcurrencyLimiter !== undefined + ? { resolveConcurrencyLimiter: deps.resolveConcurrencyLimiter } + : {}), + ...(deps.logger !== undefined ? { logger: deps.logger } : {}), + ...(deps.now !== undefined ? { now: deps.now } : {}), + // emit is required by performCompaction; fall back to a no-op + // when the orchestrator was constructed without one (tests). + emit: deps.emit ?? noopEmit, + }, + conversationId, + { keepLastN, modelName: effectiveModelName }, + ); + if ("error" in outcome) { + turnLogger?.warn("compaction:in-flight:skipped", { + conversationId, + turnId, + error: outcome.error, + }); + return; // too short / empty summary / unknown model → no replacement + } + // Replace the kernel's running history with the summary + the + // kernel's OWN most-recent N messages. Using the kernel's messages + // (not the store's) preserves mid-turn steering messages and the + // vision-transformed provider view the kernel already holds — no + // re-transcription needed. The STORE was updated by performCompaction + // with the canonical (un-transformed) recent messages. + const recent = messages.slice(messages.length - keepLastN); + return [outcome.summaryMessage, ...recent]; + }, }; // Persist the user message at turn start so it has a seq @@ -1296,6 +1369,16 @@ export function createWarmService( const DEFAULT_KEEP_LAST_N = 10; const DEFAULT_COMPACT_PERCENT = 85; +/** + * No-op emit used as a fallback when the orchestrator is constructed without an + * `emit` (some tests). `performCompaction` requires a non-optional `emit` (it + * emits `conversationCompacted`); the in-flight path degrades to emitting + * nothing rather than skipping compaction entirely. Generic-typed so it + * satisfies `PerformCompactionDeps["emit"]` for any hook payload type. + */ +const noopEmit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void = + () => {}; + const COMPACTION_SYSTEM_PROMPT = "You are a conversation summarizer. Summarize the following conversation concord concisely but comprehensively. " + "Focus on key decisions, context, file paths, and any unresolved questions. " + @@ -1317,6 +1400,192 @@ function formatMessagesForSummary(messages: readonly ChatMessage[]): string { .join("\n\n"); } +/** + * Deps for {@link performCompaction} — the subset of `SessionOrchestratorDeps` + * needed to summarize old history, fork it to an archive, and replace it with + * a summary + recent messages. Structural so both the compaction SERVICE + * (`compact`, manual + post-seal auto) and the IN-FLIGHT compaction path (the + * turn loop's `onStepBoundary`) can call the same shared core without duplicating + * the summarization/fork/replace/emit logic. The active-conversation guard and + * the threshold check are the CALLERS' policy (they differ between the two + * paths) and are NOT performed here. + */ +interface PerformCompactionDeps { + readonly conversationStore: ConversationStore; + readonly resolveProvider: () => ProviderContract; + readonly resolveModel?: ( + modelName: string, + ) => { provider: ProviderContract; model: string } | undefined; + readonly resolveSystemPrompt?: () => SystemPromptService | undefined; + readonly resolveConcurrencyLimiter?: () => ConcurrencyLimiter | undefined; + readonly logger?: Logger; + readonly now?: () => number; + readonly emit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void; +} + +/** Result of a successful {@link performCompaction}. */ +interface PerformCompactionResult { + readonly summary: string; + readonly newConversationId: string; + readonly messagesSummarized: number; + readonly messagesKept: number; + /** + * The system-role summary message that heads the compacted history + * (`[summaryMessage, ...recentKept]`). Returned so the in-flight caller can + * build the kernel's replacement history with the SAME summary object. + */ + readonly summaryMessage: ChatMessage; +} + +/** + * The shared compaction core: load the conversation history from the store, + * summarize the oldest `history.length - keepLastN` messages via a provider + * stream, fork the full pre-compaction history to an archive (non-destructive), + * and replace the live history with `[summaryMessage, ...recentKept]`. Emits + * `conversationCompacted`. Returns the result (incl. the `summaryMessage`) or an + * error object. + * + * Performs NO active-conversation guard and NO threshold check — those are the + * callers' policy. No-ops (returns an error) when the conversation is too + * short to compact (≤ keepLastN messages) or the summary is empty. + */ +async function performCompaction( + deps: PerformCompactionDeps, + conversationId: string, + opts: { readonly keepLastN?: number; readonly modelName?: string }, +): Promise<PerformCompactionResult | { readonly error: string }> { + const history = await deps.conversationStore.load(conversationId); + const keepLastN = opts?.keepLastN ?? DEFAULT_KEEP_LAST_N; + + if (history.length <= keepLastN) { + return { error: "conversation too short to compact" }; + } + + // Split: old messages to summarize + recent messages to keep. + const toSummarize = history.slice(0, history.length - keepLastN); + const toKeep = history.slice(history.length - keepLastN); + + // Resolve provider + let provider: ProviderContract; + let modelOverride: string | undefined; + if (opts?.modelName !== undefined && deps.resolveModel !== undefined) { + const resolved = deps.resolveModel(opts.modelName); + if (resolved === undefined) return { error: `unknown model: ${opts.modelName}` }; + provider = resolved.provider; + modelOverride = resolved.model; + } else { + provider = deps.resolveProvider(); + } + + // Wrap with concurrency limiting (same as the main turn path). + const compactionLimiter = deps.resolveConcurrencyLimiter?.(); + if (compactionLimiter !== undefined) { + const compactionWorkspaceId = await deps.conversationStore.getWorkspaceId(conversationId); + provider = wrapProviderWithConcurrency( + provider, + compactionLimiter, + conversationId, + compactionWorkspaceId, + deps.now?.() ?? Date.now(), + ); + } + + // Build the summarization request: system prompt + conversation text + instruction + const conversationText = formatMessagesForSummary(toSummarize); + const summaryRequest: ChatMessage = { + role: "user", + chunks: [ + { + type: "text", + text: `Please summarize the following conversation:\n\n${conversationText}`, + }, + ], + }; + + const providerOpts: ProviderStreamOptions = { + maxTokens: 2000, + ...(modelOverride !== undefined ? { model: modelOverride } : {}), + ...(deps.logger !== undefined + ? { logger: deps.logger.child({ conversationId, attrs: { compaction: true } }) } + : {}), + }; + + // Reconstruct the system prompt on compaction (fresh variable + // resolution — files/cwd/time may have changed since construction). + // The construct call also persists the result for future turns. When + // the system-prompt service is unavailable, fall back to the + // compaction-only system prompt (current behavior, no regression). + const systemPromptService = deps.resolveSystemPrompt?.(); + let compactionSystemPrompt: string; + if (systemPromptService !== undefined) { + const cwd = (await deps.conversationStore.getEffectiveCwd(conversationId)) ?? process.cwd(); + const workspaceId = await deps.conversationStore.getWorkspaceId(conversationId); + const computerId = await deps.conversationStore.getEffectiveComputer(conversationId); + const constructed = await systemPromptService.construct(conversationId, cwd, { + ...(opts?.modelName !== undefined ? { model: opts.modelName } : {}), + workspaceId, + ...(computerId !== null ? { computerId } : {}), + }); + compactionSystemPrompt = `${constructed}\n\n${COMPACTION_SYSTEM_PROMPT}`; + } else { + compactionSystemPrompt = COMPACTION_SYSTEM_PROMPT; + } + + // Call the provider and accumulate the summary + let summary = ""; + for await (const event of provider.stream([summaryRequest], [], { + ...providerOpts, + systemPrompt: compactionSystemPrompt, + })) { + if ((event as ProviderEvent).type === "text-delta") { + summary += (event as { delta: string }).delta; + } else if ((event as ProviderEvent).type === "error") { + return { error: (event as { message: string }).message }; + } + } + + if (summary.trim().length === 0) { + return { error: "model produced empty summary" }; + } + + // Non-destructive: fork the full pre-compaction history to a new + // archive conversation. The original conversation keeps its ID + // (so messaging between agents still works) and gets the compacted + // content. The archive inherits the original's compactedFrom, + // creating a chain: A → Y → X → ... + const archiveId = crypto.randomUUID(); + await deps.conversationStore.forkHistory(conversationId, archiveId); + + // Replace history: [system: summary] + recent messages + const summaryMessage: ChatMessage = { + role: "system", + chunks: [ + { + type: "text", + text: `The following is a summary of the previous conversation:\n\n${summary}`, + }, + ], + }; + + await deps.conversationStore.replaceHistory(conversationId, [summaryMessage, ...toKeep]); + await deps.conversationStore.setCompactedFrom(conversationId, archiveId); + + deps.emit(conversationCompacted, { + conversationId, + newConversationId: archiveId, + messagesSummarized: toSummarize.length, + messagesKept: toKeep.length, + }); + + return { + summary, + newConversationId: archiveId, + messagesSummarized: toSummarize.length, + messagesKept: toKeep.length, + summaryMessage, + }; +} + export function createCompactionService( deps: SessionOrchestratorDeps & { readonly emit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void; @@ -1329,14 +1598,9 @@ export function createCompactionService( return { error: "conversation is generating" }; } - const history = await deps.conversationStore.load(conversationId); - const keepLastN = opts?.keepLastN ?? DEFAULT_KEEP_LAST_N; - - if (history.length <= keepLastN) { - return { error: "conversation too short to compact" }; - } - // Auto mode: check if contextSize exceeds percent of contextWindow. + // The threshold check is the caller's policy (uses persisted turn + // metrics) and is NOT performed by the shared `performCompaction` core. if (opts?.auto === true) { const stored = await deps.conversationStore.getCompactPercent(conversationId); const percent = stored ?? DEFAULT_COMPACT_PERCENT; @@ -1360,129 +1624,22 @@ export function createCompactionService( if (contextSize < threshold) return { error: "threshold not exceeded" }; } - // Split: old messages to summarize + recent messages to keep. - const toSummarize = history.slice(0, history.length - keepLastN); - const toKeep = history.slice(history.length - keepLastN); - - // Resolve provider - let provider: ProviderContract; - let modelOverride: string | undefined; - if (opts?.modelName !== undefined && deps.resolveModel !== undefined) { - const resolved = deps.resolveModel(opts.modelName); - if (resolved === undefined) return { error: `unknown model: ${opts.modelName}` }; - provider = resolved.provider; - modelOverride = resolved.model; - } else { - provider = deps.resolveProvider(); - } - - // Wrap with concurrency limiting (same as the main turn path). - const compactionLimiter = deps.resolveConcurrencyLimiter?.(); - if (compactionLimiter !== undefined) { - const compactionWorkspaceId = await deps.conversationStore.getWorkspaceId(conversationId); - provider = wrapProviderWithConcurrency( - provider, - compactionLimiter, - conversationId, - compactionWorkspaceId, - deps.now?.() ?? Date.now(), - ); - } - - // Build the summarization request: system prompt + conversation text + instruction - const conversationText = formatMessagesForSummary(toSummarize); - const summaryRequest: ChatMessage = { - role: "user", - chunks: [ - { - type: "text", - text: `Please summarize the following conversation:\n\n${conversationText}`, - }, - ], - }; - - const providerOpts: ProviderStreamOptions = { - maxTokens: 2000, - ...(modelOverride !== undefined ? { model: modelOverride } : {}), - ...(deps.logger !== undefined - ? { logger: deps.logger.child({ conversationId, attrs: { compaction: true } }) } - : {}), - }; - - // Reconstruct the system prompt on compaction (fresh variable - // resolution — files/cwd/time may have changed since construction). - // The construct call also persists the result for future turns. When - // the system-prompt service is unavailable, fall back to the - // compaction-only system prompt (current behavior, no regression). - const systemPromptService = deps.resolveSystemPrompt?.(); - let compactionSystemPrompt: string; - if (systemPromptService !== undefined) { - const cwd = (await deps.conversationStore.getEffectiveCwd(conversationId)) ?? process.cwd(); - const workspaceId = await deps.conversationStore.getWorkspaceId(conversationId); - const computerId = await deps.conversationStore.getEffectiveComputer(conversationId); - const constructed = await systemPromptService.construct(conversationId, cwd, { - ...(opts?.modelName !== undefined ? { model: opts.modelName } : {}), - workspaceId, - ...(computerId !== null ? { computerId } : {}), - }); - compactionSystemPrompt = `${constructed}\n\n${COMPACTION_SYSTEM_PROMPT}`; - } else { - compactionSystemPrompt = COMPACTION_SYSTEM_PROMPT; - } - - // Call the provider and accumulate the summary - let summary = ""; - for await (const event of provider.stream([summaryRequest], [], { - ...providerOpts, - systemPrompt: compactionSystemPrompt, - })) { - if ((event as ProviderEvent).type === "text-delta") { - summary += (event as { delta: string }).delta; - } else if ((event as ProviderEvent).type === "error") { - return { error: (event as { message: string }).message }; - } - } - - if (summary.trim().length === 0) { - return { error: "model produced empty summary" }; - } - - // Non-destructive: fork the full pre-compaction history to a new - // archive conversation. The original conversation keeps its ID - // (so messaging between agents still works) and gets the compacted - // content. The archive inherits the original's compactedFrom, - // creating a chain: A → Y → X → ... - const archiveId = crypto.randomUUID(); - await deps.conversationStore.forkHistory(conversationId, archiveId); - - // Replace history: [system: summary] + recent messages - const summaryMessage: ChatMessage = { - role: "system", - chunks: [ - { - type: "text", - text: `The following is a summary of the previous conversation:\n\n${summary}`, - }, - ], - }; - - await deps.conversationStore.replaceHistory(conversationId, [summaryMessage, ...toKeep]); - await deps.conversationStore.setCompactedFrom(conversationId, archiveId); + // Shared summarize + fork + replace + emit core (no active guard, no + // threshold — those are this caller's policy above). The length check + // ("conversation too short to compact") lives inside the core. + const outcome = await performCompaction(deps, conversationId, { + ...(opts?.keepLastN !== undefined ? { keepLastN: opts.keepLastN } : {}), + ...(opts?.modelName !== undefined ? { modelName: opts.modelName } : {}), + }); + if ("error" in outcome) return { error: outcome.error }; + const { summary, newConversationId, messagesSummarized, messagesKept } = outcome; const result: CompactionResult = { summary, - newConversationId: archiveId, - messagesSummarized: toSummarize.length, - messagesKept: toKeep.length, + newConversationId, + messagesSummarized, + messagesKept, }; - - deps.emit(conversationCompacted, { - conversationId, - newConversationId: archiveId, - messagesSummarized: toSummarize.length, - messagesKept: toKeep.length, - }); - return result; }, }; |