diff options
Diffstat (limited to 'packages/session-orchestrator/src')
| -rw-r--r-- | packages/session-orchestrator/src/index.ts | 1 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/orchestrator.test.ts | 678 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/orchestrator.ts | 546 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/queue.test.ts | 196 | ||||
| -rw-r--r-- | packages/session-orchestrator/src/tools-filter.ts | 10 |
5 files changed, 1297 insertions, 134 deletions
diff --git a/packages/session-orchestrator/src/index.ts b/packages/session-orchestrator/src/index.ts index 3d3e816..360e8d2 100644 --- a/packages/session-orchestrator/src/index.ts +++ b/packages/session-orchestrator/src/index.ts @@ -1,5 +1,6 @@ export { extension, manifest } from "./extension.js"; export { + type CancelQueuedMessageResult, type CompactionService, type ConversationClosedPayload, type ConversationCompactedPayload, diff --git a/packages/session-orchestrator/src/orchestrator.test.ts b/packages/session-orchestrator/src/orchestrator.test.ts index 400ca0b..c4be03c 100644 --- a/packages/session-orchestrator/src/orchestrator.test.ts +++ b/packages/session-orchestrator/src/orchestrator.test.ts @@ -19,11 +19,14 @@ import type { TurnMetrics, } from "@dispatch/kernel"; import { createLogger, runTurn } from "@dispatch/kernel"; +import { createMessageQueueService } from "@dispatch/message-queue"; 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 +52,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 +162,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); }, @@ -3873,6 +3879,194 @@ describe("system prompt: regular turn flow", () => { }); }); +describe("title (summon-title): deferred until after workspace initialization", () => { + // Regression: an earlier implementation set the title in the HTTP /chat + // route BEFORE the turn started, which pre-created the conversation meta + // and made the orchestrator's `meta === null` newness check falsely report + // an EXISTING conversation — so ensureWorkspace / setWorkspaceId / the + // first-turn system-prompt construct were ALL skipped. The fix defers the + // title set into workspaceSetupPromise, AFTER the newness check + workspace + // assignment, so a titled new conversation is still initialized correctly. + + /** Wrap the in-memory store to record the ORDER of init-relevant calls. */ + function createCallRecordingStore() { + const base = createInMemoryStore(); + const calls: string[] = []; + const titleCalls: { conversationId: string; title: string }[] = []; + return { + store: { + ...base, + async getConversationMeta(conversationId: string) { + calls.push(`getMeta:${conversationId}`); + return base.getConversationMeta(conversationId); + }, + async ensureWorkspace(id: string) { + calls.push(`ensureWorkspace:${id}`); + return base.ensureWorkspace(id); + }, + async setWorkspaceId(conversationId: string, workspaceId: string) { + calls.push(`setWorkspaceId:${workspaceId}`); + await base.setWorkspaceId(conversationId, workspaceId); + }, + async setConversationTitle(conversationId: string, title: string) { + calls.push(`setTitle:${title}`); + titleCalls.push({ conversationId, title }); + await base.setConversationTitle(conversationId, title); + }, + } as ConversationStore, + calls, + titleCalls, + }; + } + + it("titled new conversation: workspace assigned, system prompt constructed, title set", async () => { + const { store, calls, titleCalls } = createCallRecordingStore(); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + const constructCalls: string[] = []; + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + resolveSystemPrompt: () => + createFakeSystemPromptService(async (conversationId) => { + constructCalls.push(conversationId); + return "CONSTRUCTED_PROMPT"; + }), + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-new", + text: "hi", + onEvent: () => {}, + title: "My Task", + workspaceId: "my-workspace", + }); + + // The bug: workspace init was skipped. It must NOT be. + expect(calls).toContain("ensureWorkspace:my-workspace"); + expect(calls).toContain("setWorkspaceId:my-workspace"); + // First-turn system prompt construct runs (proves isNewConversation was + // true — the newness check was not fooled by a pre-created meta). + expect(constructCalls).toEqual(["conv-title-new"]); + // The title is persisted. + expect(titleCalls).toEqual([{ conversationId: "conv-title-new", title: "My Task" }]); + }); + + it("title is set AFTER the newness check + workspace assignment (order)", async () => { + const { store, calls } = createCallRecordingStore(); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-order", + text: "hi", + onEvent: () => {}, + title: "Ordered", + }); + + const getMetaIdx = calls.findIndex((c) => c.startsWith("getMeta:")); + const ensureIdx = calls.findIndex((c) => c.startsWith("ensureWorkspace:")); + const setWsIdx = calls.findIndex((c) => c.startsWith("setWorkspaceId:")); + const setTitleIdx = calls.findIndex((c) => c.startsWith("setTitle:")); + expect(getMetaIdx).toBeGreaterThanOrEqual(0); + expect(ensureIdx).toBeGreaterThan(getMetaIdx); + expect(setWsIdx).toBeGreaterThan(ensureIdx); + expect(setTitleIdx).toBeGreaterThan(setWsIdx); + }); + + it("no title: setConversationTitle is not called", async () => { + const { store, calls } = createCallRecordingStore(); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-no-title", + text: "hi", + onEvent: () => {}, + }); + + expect(calls.some((c) => c.startsWith("setTitle:"))).toBe(false); + }); + + it("existing conversation with a title: workspace NOT re-assigned, title still set", async () => { + const { store, calls, titleCalls } = createCallRecordingStore(); + // Seed an existing conversation (meta non-null, workspace already set). + await store.setWorkspaceId("conv-title-existing", "prior-workspace"); + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-existing", + text: "hi", + onEvent: () => {}, + title: "Renamed", + }); + + // Existing conversation: workspace init must not run again. + expect(calls.some((c) => c.startsWith("ensureWorkspace:"))).toBe(false); + // But the title is still applied (rename on an existing conversation). + expect(titleCalls).toEqual([{ conversationId: "conv-title-existing", title: "Renamed" }]); + }); + + it("turn still completes if setConversationTitle throws", async () => { + const base = createInMemoryStore(); + const store: ConversationStore = { + ...base, + async setConversationTitle() { + throw new Error("title store unavailable"); + }, + }; + const provider: ProviderContract = { id: "p", stream: async function* () {} }; + const { captured, captureRunTurn } = createCapturingRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => provider, + resolveTools: () => [], + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-title-throws", + text: "hi", + onEvent: () => {}, + title: "Resilient", + }); + + // The turn ran despite the title-set failure. + expect(captured).toHaveLength(1); + }); +}); + describe("system prompt: compaction flow", () => { function seedHistory( store: ReturnType<typeof createInMemoryStore>, @@ -3985,6 +4179,484 @@ 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"); + } + }); + + it("steering messages are persisted and survive in-flight compaction — the store and the LLM's context stay aligned (Bug A + B)", async () => { + // Regression test for the two critical bugs: + // A) drainSteering injected steering into the kernel's in-memory messages + // but never persisted it → the user could never see it, and + // compaction (loading the store) scrubbed it. + // B) compaction sliced the store and the kernel's messages independently; + // the unpersisted steering offset the slices → DB and LLM dropped + // DIFFERENT messages (structural divergence). + // Fix: drainSteering persists (awaited); compaction uses the kernel's LIVE + // messages array, so the store write and the kernel's replacement use the + // SAME recent slice → aligned, and the steering is retained. + const store = createInMemoryStore(); + seedHistory(store, "conv-align", 15); // > keepLastN(10) → compactable + const queue = createMessageQueueService({ + id: () => `q-${Math.random().toString(36).slice(2, 8)}`, + now: () => 1000, + notify: () => {}, + }); + queue.enqueue("conv-align", "STEER MID-TURN"); // drained at step 0's boundary + + // contextWindow 1000, percent 85 → threshold 850. Step 0 usage 900 → fire. + 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: + [ + { type: "text-delta", delta: "ALIGN SUMMARY" }, + { type: "finish", reason: "stop" }, + ], + // Step 1 (post-compaction) — the turn continues: + [ + { 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 }), + resolveQueue: () => queue, + runTurn, + emit: (hook, payload) => { + if (hook === conversationCompacted) { + compactedEvents.push(payload as ConversationCompactedPayload); + } + }, + }); + + await orchestrator.handleMessage({ + conversationId: "conv-align", + text: "keep working overnight", + onEvent: () => {}, + modelName: "test/model", + }); + + // Bug A: the steering message was PERSISTED to the store (the user CAN see + // it). It is either retained in the kept recent slice or captured in the + // summary; either way it must be present in the store, not lost. + expect(compactedEvents).toHaveLength(1); + const stored = store.data.get("conv-align") ?? []; + // The compacted history begins with the summary system message. + expect(stored[0]?.role).toBe("system"); + expect((stored[0]?.chunks[0] as { text?: string } | undefined)?.text).toContain( + "ALIGN SUMMARY", + ); + // The steering message survived in the kept recent slice (it was the most + // recent user message before compaction, so it is within keepLastN=10). + const storedSteering = stored.find( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + expect(storedSteering).toBeDefined(); + + // Bug B: the store and the LLM's context are ALIGNED. The provider's + // post-compaction call (captured[2]) is the kernel's working history AFTER + // compaction replaced it. The store was written with the SAME compacted + // history, then step 1's assistant output was appended on top. So the + // kernel's view (captured[2]) must be an exact PREFIX of the store — same + // summary, same recent slice, same steering at the same index. (Before the + // fix, the store dropped the steering while the kernel kept it, so the two + // diverged structurally.) + const step1Messages = capturedMessages[2] ?? []; + expect(step1Messages[0]?.role).toBe("system"); // summary heads both + const kernelSteering = step1Messages.find( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + expect(kernelSteering).toBeDefined(); + // The kernel's post-compaction history is an exact PREFIX of the store + // (the store then has step 1's appended assistant output after it). Same + // length, same roles in order, same steering index => structural alignment. + expect(stored.length).toBeGreaterThanOrEqual(step1Messages.length); + const storePrefix = stored.slice(0, step1Messages.length); + expect(storePrefix).toHaveLength(step1Messages.length); + for (let i = 0; i < step1Messages.length; i++) { + expect(storePrefix[i]?.role).toBe(step1Messages[i]?.role); + } + const storeSteerIdx = storePrefix.findIndex( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + const kernelSteerIdx = step1Messages.findIndex( + (m) => + m.role === "user" && m.chunks.some((c) => c.type === "text" && c.text === "STEER MID-TURN"), + ); + expect(kernelSteerIdx).toBe(storeSteerIdx); + expect(kernelSteerIdx).toBeGreaterThanOrEqual(0); + }); +}); + 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..a2e141a 100644 --- a/packages/session-orchestrator/src/orchestrator.ts +++ b/packages/session-orchestrator/src/orchestrator.ts @@ -131,6 +131,16 @@ export interface StartTurnInput { * system-prompt service when loaded). */ readonly systemPrompt?: string; + /** + * A human-readable title for the conversation tab. When provided, it is + * persisted via `setConversationTitle` AFTER the new-conversation workspace + * setup resolves (so the `meta === null` newness detection still fires and + * `ensureWorkspace` / `setWorkspaceId` / first-turn system-prompt + * construction are NOT skipped) and BEFORE the first message append (so the + * append's auto-title does not overwrite it). Omit to keep the auto-derived + * title. The caller is responsible for trimming/validation. + */ + readonly title?: string; } export type StartTurnResult = @@ -170,6 +180,20 @@ export interface EnqueueResult { readonly queue: readonly QueuedMessage[]; } +/** + * Result of `SessionOrchestrator.cancelQueuedMessage`. `cancelled` is true when + * a message with the given id was found in the conversation's queue and removed + * (it will never run — never delivered as steering, never carried into a new + * turn). `cancelled` is false when the message was not in the queue (already + * drained/delivered, never existed, unknown conversation) OR when the + * message-queue extension isn't loaded (degraded — feature off). `queue` is the + * post-cancel snapshot (empty when no queue extension is loaded). + */ +export interface CancelQueuedMessageResult { + readonly cancelled: boolean; + readonly queue: readonly QueuedMessage[]; +} + export type TurnEventListener = (event: AgentEvent) => void; interface ActiveTurn { @@ -331,6 +355,18 @@ export interface SessionOrchestrator { * returned snapshot is empty (degraded — feature off). */ enqueue(input: EnqueueInput): EnqueueResult; + /** + * Cancel (remove) a SINGLE queued message by id so it never runs. The single + * entry transports call to cancel a queued steering message. Resolves the + * message-queue service lazily (same as `enqueue`); when the extension isn't + * loaded the call degrades to `{ cancelled: false, queue: [] }`. Idempotent — + * cancelling a message that is no longer queued (already drained/delivered) + * returns `{ cancelled: false, ... }` without error. + */ + cancelQueuedMessage(input: { + readonly conversationId: string; + readonly messageId: string; + }): CancelQueuedMessageResult; subscribe(conversationId: string, listener: TurnEventListener): () => void; isActive(conversationId: string): boolean; /** @@ -370,6 +406,8 @@ export interface SessionOrchestrator { systemPrompt?: string; /** Images attached to this turn — see {@link StartTurnInput.images}. */ images?: readonly ImageInput[]; + /** Conversation tab title — see {@link StartTurnInput.title}. */ + title?: string; }): Promise<void>; } @@ -548,6 +586,7 @@ export function createSessionOrchestrator( workspaceId: string, systemPromptOverride: string | undefined, images: readonly ImageInput[] | undefined, + title: string | undefined, ): void { const turnId = generateTurnId(); const promptStartedAt = deps.now?.() ?? Date.now(); @@ -567,14 +606,34 @@ export function createSessionOrchestrator( // The newness flag is also reused to decide whether to construct // (first turn) or get (subsequent turn) the system prompt — see the // providerOpts assembly below. + // + // An explicit `title` (e.g. the CLI `--title` flag) is persisted HERE, + // AFTER the workspace setup resolves — deliberately NOT before the turn. + // Setting it earlier (e.g. in the HTTP route) would pre-create the meta + // row, make `meta !== null`, and fool this newness check into skipping + // `ensureWorkspace` / `setWorkspaceId` / first-turn system-prompt + // construction. By deferring it to here, the title lands after the + // workspace is assigned but BEFORE the first message append (so the + // append's auto-title sees a non-"Untitled" title and preserves it). const workspaceSetupPromise = (async (): Promise<boolean> => { const meta = await deps.conversationStore.getConversationMeta(conversationId); if (meta === null) { await deps.conversationStore.ensureWorkspace(workspaceId); await deps.conversationStore.setWorkspaceId(conversationId, workspaceId); - return true; } - return false; + if (title !== undefined) { + // Best-effort: a title-set failure must NOT break the turn (the + // workspace setup above already succeeded). Log and continue — the + // append's auto-derived title applies instead. + try { + await deps.conversationStore.setConversationTitle(conversationId, title); + } catch (err) { + deps.logger?.child({ conversationId }).warn("orchestrator: title set failure", { + error: err instanceof Error ? err.message : String(err), + }); + } + } + return meta === null; })(); // ALWAYS resolve the effective cwd through getEffectiveCwd, passing the @@ -760,6 +819,11 @@ export function createSessionOrchestrator( conversationId, ...(effectiveCwd !== undefined ? { cwd: effectiveCwd } : {}), ...(effectiveComputerId !== undefined ? { computerId: effectiveComputerId } : {}), + // Thread the turn's abort signal into the filter chain so a filter + // awaiting slow I/O (the MCP tools filter connecting to MCP servers) + // can be interrupted by POST /conversations/:id/stop instead of + // blocking the turn until its own timeout fires. + signal: controller.signal, }); const dispatch = deps.resolveDispatch?.() ?? defaultDispatchPolicy(); const turnLogger = deps.logger?.child({ conversationId, turnId }); @@ -844,17 +908,33 @@ export function createSessionOrchestrator( const drainSteering = queue === undefined ? undefined - : (): readonly ChatMessage[] => { + : async (): Promise<readonly ChatMessage[]> => { const queued = queue.drain(conversationId); if (queued.length === 0) return []; const steerText = queued.map((q) => q.text).join("\n\n"); + const steeringMessage: ChatMessage = { + role: "user", + chunks: [{ type: "text", text: steerText }], + }; + // Persist the injected steering message to the store as part + // of the SAME critical section as the injection, so it is + // never lost. Without this, the message would live only in + // the kernel's in-memory messages array (never persisted), + // so a user could never see it — and in-flight compaction + // (which loads the store) would scrub it. A fire-and-forget + // append would race with the next step's `onStepComplete` + // append and collide on the store's seq counter, so we + // `await` it (the kernel awaits drainSteering). Errors + // propagate (a DB failure ends the turn, matching + // `onStepComplete`'s behavior). + await deps.conversationStore.append(conversationId, [steeringMessage]); emitToHub(conversationId, { type: "steering", conversationId, turnId, text: steerText, }); - return [{ role: "user", chunks: [{ type: "text", text: steerText }] }]; + return [steeringMessage]; }; // Vision handoff: transform the message list for the provider. When the @@ -899,6 +979,81 @@ 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, + // Pass the kernel's LIVE messages array (not a store reload): + // it includes mid-turn steering messages (now persisted by + // drainSteering) and is the authoritative prompt state. Using it + // for the split keeps the store write and the kernel's + // replacement aligned (same recent slice) — no DB↔LLM divergence. + { keepLastN, modelName: effectiveModelName, messages }, + ); + if ("error" in outcome) { + turnLogger?.warn("compaction:in-flight:skipped", { + conversationId, + turnId, + error: outcome.error, + }); + return; // too short / empty summary / unknown model → no replacement + } + // Return the compacted history the kernel should adopt. This is + // EXACTLY what performCompaction wrote to the store + // ([summary, ...recent-from-live]), so the kernel's working + // history and the store stay byte-aligned. + return outcome.compactedMessages; + }, }; // Persist the user message at turn start so it has a seq @@ -1021,6 +1176,7 @@ export function createSessionOrchestrator( workspaceId, systemPrompt, images, + title, }) { if (activeTurns.has(conversationId)) { return { started: false, reason: "already-active" }; @@ -1035,6 +1191,7 @@ export function createSessionOrchestrator( workspaceId ?? "default", systemPrompt, images, + title, ); const turn = activeTurns.get(conversationId); const turnId = turn !== undefined ? turn.turnId : ""; @@ -1060,6 +1217,19 @@ export function createSessionOrchestrator( return { startedTurn: false, queue: snapshot }; }, + cancelQueuedMessage({ conversationId, messageId }) { + // When the message-queue extension isn't loaded this degrades: nothing to + // cancel, empty snapshot (feature off). Mirrors `enqueue`'s degraded path. + const queue = deps.resolveQueue?.(); + if (queue === undefined) { + return { cancelled: false, queue: [] }; + } + const beforeLen = queue.getQueue(conversationId).length; + const snapshot = queue.cancel(conversationId, messageId); + const cancelled = snapshot.length < beforeLen; + return { cancelled, queue: snapshot }; + }, + subscribe(conversationId, listener) { let listeners = subscribers.get(conversationId); if (listeners === undefined) { @@ -1140,6 +1310,7 @@ export function createSessionOrchestrator( workspaceId, systemPrompt, images, + title, }) { const turnInput: StartTurnInput = { conversationId, @@ -1151,6 +1322,7 @@ export function createSessionOrchestrator( ...(workspaceId !== undefined ? { workspaceId } : {}), ...(systemPrompt !== undefined ? { systemPrompt } : {}), ...(images !== undefined ? { images } : {}), + ...(title !== undefined ? { title } : {}), }; const result = orchestrator.startTurn(turnInput); if (!result.started) { @@ -1296,6 +1468,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 +1499,222 @@ 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 full compacted history `[summaryMessage, ...recentKept]` exactly as + * written to the store. The in-flight caller returns this to the kernel so + * the kernel's working history and the store stay byte-aligned (the same + * `recentKept` slice — taken from the caller-supplied live `messages` — is + * used for BOTH the store write and this return value). + */ + readonly compactedMessages: readonly ChatMessage[]; +} + +/** + * The shared compaction core: 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` + the `compactedMessages`) or an error. + * + * History source: when `opts.messages` is provided (the in-flight path), it is + * used as the authoritative history — this is the kernel's LIVE messages array, + * which includes mid-turn steering messages (and the vision-transformed + * provider view) that a store reload could miss (the steering persist may not + * have completed, or — before this fix — was never done at all). Using the live + * array keeps the store write and the kernel's replacement aligned (same + * `recentKept` slice), avoiding the DB↔LLM structural divergence where + * independent slices dropped different messages. When `opts.messages` is + * omitted (the post-seal/manual `compact()` path — the turn has ended, so the + * store is stable), the history is loaded from the store. + * + * 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; + /** The kernel's live messages array (in-flight path). Omit to load the store (post-seal/manual). */ + readonly messages?: readonly ChatMessage[]; + }, +): Promise<PerformCompactionResult | { readonly error: string }> { + // Use the caller-supplied live messages (in-flight) or load the store + // (post-seal/manual — the store is stable once the turn has ended). + const history = opts.messages ?? (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] + the recent kept messages. `toKeep` + // is sliced from the caller-supplied live `messages` (in-flight) — the SAME + // slice returned below as `compactedMessages` — so the store and the kernel's + // working history stay byte-aligned (same messages kept/dropped). + const summaryMessage: ChatMessage = { + role: "system", + chunks: [ + { + type: "text", + text: `The following is a summary of the previous conversation:\n\n${summary}`, + }, + ], + }; + + const compactedMessages: readonly ChatMessage[] = [summaryMessage, ...toKeep]; + await deps.conversationStore.replaceHistory(conversationId, compactedMessages); + 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, + compactedMessages, + }; +} + export function createCompactionService( deps: SessionOrchestratorDeps & { readonly emit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void; @@ -1329,14 +1727,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 +1753,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; }, }; diff --git a/packages/session-orchestrator/src/queue.test.ts b/packages/session-orchestrator/src/queue.test.ts index a09a441..e5746c3 100644 --- a/packages/session-orchestrator/src/queue.test.ts +++ b/packages/session-orchestrator/src/queue.test.ts @@ -217,7 +217,9 @@ function createDrainingCaptureRunTurn(): { captured.push(input); if (input.drainSteering !== undefined) { drainCalled = true; - const drained = input.drainSteering(); + // The kernel awaits drainSteering (it may return a Promise that + // persists the injected messages); the fake mirrors that. + const drained = await input.drainSteering(); drainedMessages.push(...drained); } return { @@ -332,6 +334,18 @@ describe("drainSteering", () => { expect(steering?.conversationId).toBe("conv-drain"); expect(steering?.text).toBe("first\n\nsecond"); expect(steering?.turnId).toMatch(/^turn-/); + + // The steering message was PERSISTED to the store (not just injected into + // the kernel's in-memory array). Without this, a user could never see the + // steering message, and in-flight compaction (which uses the live messages) + // would be the only thing keeping it — but only if it fired. + const stored = store.data.get("conv-drain") ?? []; + const storedSteering = stored.find( + (m) => + m.role === "user" && + m.chunks.some((c) => c.type === "text" && c.text === "first\n\nsecond"), + ); + expect(storedSteering).toBeDefined(); }); it("drainSteering on an empty queue returns [] and emits nothing", async () => { @@ -593,3 +607,183 @@ describe("enqueue", () => { await sealed; }); }); + +// --- cancelQueuedMessage facade (remove a single queued message by id) --- + +describe("cancelQueuedMessage", () => { + it("removes a queued message by id → cancelled:true + post-cancel snapshot", () => { + const store = createInMemoryStore(); + const queue = createTestQueue(); + queue.enqueue("conv-cancel", "a"); + queue.enqueue("conv-cancel", "b"); + const second = queue.getQueue("conv-cancel")[1]; + if (second === undefined) throw new Error("expected a second enqueued message"); + const secondId = second.id; + expect(queue.getQueue("conv-cancel")).toHaveLength(2); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => simpleProvider(), + resolveTools: noTools, + applyToolsFilter: identityApplyToolsFilter, + runTurn, + resolveQueue: () => queue, + }); + + const result = orchestrator.cancelQueuedMessage({ + conversationId: "conv-cancel", + messageId: secondId, + }); + expect(result.cancelled).toBe(true); + expect(result.queue.map((m) => m.id)).not.toContain(secondId); + expect(result.queue).toHaveLength(1); + expect(queue.getQueue("conv-cancel").map((m) => m.id)).not.toContain(secondId); + }); + + it("cancel of the only message empties the queue (cancelled:true)", () => { + const store = createInMemoryStore(); + const queue = createTestQueue(); + queue.enqueue("conv-only", "solo"); + const onlyId = queue.getQueue("conv-only")[0]?.id; + if (onlyId === undefined) throw new Error("expected an enqueued message"); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => simpleProvider(), + resolveTools: noTools, + applyToolsFilter: identityApplyToolsFilter, + runTurn, + resolveQueue: () => queue, + }); + + const result = orchestrator.cancelQueuedMessage({ + conversationId: "conv-only", + messageId: onlyId, + }); + expect(result.cancelled).toBe(true); + expect(result.queue).toEqual([]); + expect(queue.getQueue("conv-only")).toEqual([]); + }); + + it("cancel of a missing id → cancelled:false, queue unchanged (idempotent)", () => { + const store = createInMemoryStore(); + const queue = createTestQueue(); + queue.enqueue("conv-miss", "a"); + queue.enqueue("conv-miss", "b"); + const before = queue.getQueue("conv-miss"); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => simpleProvider(), + resolveTools: noTools, + applyToolsFilter: identityApplyToolsFilter, + runTurn, + resolveQueue: () => queue, + }); + + const result = orchestrator.cancelQueuedMessage({ + conversationId: "conv-miss", + messageId: "does-not-exist", + }); + expect(result.cancelled).toBe(false); + expect(result.queue.map((m) => m.id)).toEqual(before.map((m) => m.id)); + // live state unchanged + expect(queue.getQueue("conv-miss").map((m) => m.id)).toEqual(before.map((m) => m.id)); + }); + + it("cancel on an unknown conversation → cancelled:false, empty queue", () => { + const store = createInMemoryStore(); + const queue = createTestQueue(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => simpleProvider(), + resolveTools: noTools, + applyToolsFilter: identityApplyToolsFilter, + runTurn, + resolveQueue: () => queue, + }); + + const result = orchestrator.cancelQueuedMessage({ + conversationId: "never-existed", + messageId: "anything", + }); + expect(result.cancelled).toBe(false); + expect(result.queue).toEqual([]); + }); + + it("no queue ext (resolveQueue undefined) → cancelled:false, empty queue (degraded)", () => { + const store = createInMemoryStore(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => simpleProvider(), + resolveTools: noTools, + applyToolsFilter: identityApplyToolsFilter, + runTurn, + // resolveQueue intentionally omitted — feature degrades off + }); + + const result = orchestrator.cancelQueuedMessage({ + conversationId: "conv-noqueue", + messageId: "whatever", + }); + expect(result.cancelled).toBe(false); + expect(result.queue).toEqual([]); + }); + + it("cancelled message is NOT delivered as steering (never runs)", async () => { + const store = createInMemoryStore(); + const queue = createTestQueue(); + queue.enqueue("conv-steer", "keep-me"); + const cancelId = queue.enqueue("conv-steer", "cancel-me")[1]?.id; + if (cancelId === undefined) throw new Error("expected a second enqueued message"); + // the first message id (kept) + const keepId = queue.getQueue("conv-steer")[0]?.id; + if (keepId === undefined) throw new Error("expected a kept message"); + + const { captured, drainedMessages, runTurn: captureRunTurn } = createDrainingCaptureRunTurn(); + + const { orchestrator } = createSessionOrchestrator({ + conversationStore: store, + resolveProvider: () => ({ id: "p", stream: async function* () {} }), + resolveTools: noTools, + applyToolsFilter: identityApplyToolsFilter, + runTurn: captureRunTurn, + resolveQueue: () => queue, + }); + + // Cancel the second message BEFORE the turn drains. + const cancelResult = orchestrator.cancelQueuedMessage({ + conversationId: "conv-steer", + messageId: cancelId, + }); + expect(cancelResult.cancelled).toBe(true); + expect(cancelResult.queue.map((m) => m.id)).toEqual([keepId]); + + const events: AgentEvent[] = []; + const unsub = orchestrator.subscribe("conv-steer", (e) => events.push(e)); + + orchestrator.startTurn({ conversationId: "conv-steer", text: "go" }); + await waitForSealed(orchestrator, "conv-steer"); + unsub(); + + // The steering drain combined ONLY the kept message — the cancelled one + // is absent from the drained text. + expect(drainedMessages).toHaveLength(1); + const steerMsg = drainedMessages[0]; + if (steerMsg === undefined) throw new Error("expected a drained message"); + const chunk = steerMsg.chunks[0]; + if (chunk === undefined || chunk.type !== "text") throw new Error("expected text chunk"); + expect(chunk.text).toBe("keep-me"); + expect(chunk.text).not.toContain("cancel-me"); + + // drainSteering was wired + the queue is now empty (the kept one drained). + expect(captured[0]?.drainSteering).toBeDefined(); + expect(queue.getQueue("conv-steer")).toHaveLength(0); + + // The steering event carries only the kept text. + const steering = events.find(isSteering); + expect(steering?.text).toBe("keep-me"); + }); +}); diff --git a/packages/session-orchestrator/src/tools-filter.ts b/packages/session-orchestrator/src/tools-filter.ts index 28e82bf..22bc5cd 100644 --- a/packages/session-orchestrator/src/tools-filter.ts +++ b/packages/session-orchestrator/src/tools-filter.ts @@ -15,6 +15,15 @@ export interface ToolAssembly { readonly computerId?: string; /** The conversation this turn belongs to. */ readonly conversationId: string; + /** + * The turn's abort signal, threaded through the filter chain so a filter that + * awaits slow I/O (e.g. the MCP tools filter connecting to MCP servers) can be + * interrupted by `POST /conversations/:id/stop` instead of blocking the turn + * until its own timeout fires. Optional: omitted by paths that have no turn + * controller (e.g. the cache-warm probe), in which case filters fall back to + * their own timeouts. Filters that return a fresh assembly MUST preserve it. + */ + readonly signal?: AbortSignal; } /** Filter chain run once per turn to transform the tool set before it reaches runTurn. */ @@ -55,5 +64,6 @@ export function filterRemoteIncompatibleTools(assembly: ToolAssembly): ToolAssem ...(assembly.cwd !== undefined ? { cwd: assembly.cwd } : {}), ...(assembly.computerId !== undefined ? { computerId: assembly.computerId } : {}), conversationId: assembly.conversationId, + ...(assembly.signal !== undefined ? { signal: assembly.signal } : {}), }; } |