diff options
| author | Adam Malczewski <[email protected]> | 2026-06-29 00:56:07 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-29 00:56:07 +0900 |
| commit | ebb29b1e5b929493315469fcbd5eca6f13bd1c32 (patch) | |
| tree | 3be897eda7c2ab3c41395b76e7ba9fb24d30c463 /packages/kernel | |
| parent | a5855cad41b2e4b0a196e52cd4c1ca5d4c0ca25f (diff) | |
| download | dispatch-ebb29b1e5b929493315469fcbd5eca6f13bd1c32.tar.gz dispatch-ebb29b1e5b929493315469fcbd5eca6f13bd1c32.zip | |
fix(in-flight-compaction): persist steering messages + use live messages array for compaction
Fix two critical bugs found in code review:
Bug A — Permanent loss of mid-turn steering messages:
drainSteering injected queued messages into the kernel's in-memory messages
array but never persisted them, so a user could never see them and in-flight
compaction (which loaded the store) scrubbed them. Fix: drainSteering now
persists the steering message to the store as part of the same critical section
as the injection (await store.append). This required making drainSteering async
— the kernel now awaits it (contract: return type allows Promise; backward-
compatible with sync callbacks). Fire-and-forget was unsafe: the conversation-
store append reads the seq counter then writes chunks across multiple awaits,
so a concurrent steering append + next-step onStepComplete append would both
read the same seq counter and collide (the msgIdx-collision class of bug).
Bug B — Index misalignment (DB <-> LLM divergence):
keepLastN was sliced independently from the store's array (no steering) and
the kernel's array (with steering), so the slices dropped DIFFERENT messages.
Fix (follows from A): performCompaction accepts the kernel's LIVE messages
array instead of reloading the stale store; the SAME recentKept slice is used
for both the store write (replaceHistory) and the value returned to the kernel
(compactedMessages), so they stay byte-aligned by construction. The post-seal/
manual compact() path still loads the store (the turn has ended, so it is
stable).
Tests: kernel async-drainSteering-await contract test; orchestrator steering-
persisted + store/LLM-alignment regression test; queue.test.ts asserts the
steering is persisted; its fake runTurn now awaits drainSteering.
Verification: typecheck clean; 2014 tests pass (was 2012; +2 new + 1 assertion);
biome 0 errors (12 pre-existing warnings in untouched files).
Diffstat (limited to 'packages/kernel')
| -rw-r--r-- | packages/kernel/src/contracts/runtime.ts | 18 | ||||
| -rw-r--r-- | packages/kernel/src/runtime/run-turn.test.ts | 47 | ||||
| -rw-r--r-- | packages/kernel/src/runtime/run-turn.ts | 6 |
3 files changed, 63 insertions, 8 deletions
diff --git a/packages/kernel/src/contracts/runtime.ts b/packages/kernel/src/contracts/runtime.ts index 2bab47a..deae126 100644 --- a/packages/kernel/src/contracts/runtime.ts +++ b/packages/kernel/src/contracts/runtime.ts @@ -121,13 +121,19 @@ export interface RunTurnInput { * results. When omitted or returning an empty array, no injection happens * (the runtime is unchanged). * - * Injected (not ambient) so the kernel stays pure: it owns no queue and - * names no feature — it just calls the callback and appends what it gets. - * Only invoked when a step PRODUCED tool calls (the tool-result boundary); - * a step that ends without tool calls does not drain (the caller decides - * what to do with any pending messages after the turn ends). + * May return a Promise (the runtime `await`s it): the shell uses this to + * PERSIST the injected messages to the store as part of the same critical + * section as the injection, so they are never lost (a fire-and-forget + * persist would race with the next step's `onStepComplete` append and + * collide on the store's seq counter). A sync return is still supported + * (backward-compatible). Injected (not ambient) so the kernel stays pure: + * it owns no queue and names no feature — it just calls the callback, + * awaits it, and appends what it gets. Only invoked when a step PRODUCED + * tool calls (the tool-result boundary); a step that ends without tool + * calls does not drain (the caller decides what to do with any pending + * messages after the turn ends). */ - readonly drainSteering?: () => readonly ChatMessage[]; + readonly drainSteering?: () => readonly ChatMessage[] | Promise<readonly ChatMessage[]>; /** * Optional. Called by the runtime after each step's messages are finalized diff --git a/packages/kernel/src/runtime/run-turn.test.ts b/packages/kernel/src/runtime/run-turn.test.ts index 90357b1..452e162 100644 --- a/packages/kernel/src/runtime/run-turn.test.ts +++ b/packages/kernel/src/runtime/run-turn.test.ts @@ -2853,6 +2853,53 @@ describe("runTurn", () => { expect(drainCallCount).toBe(2); }); + it("async drainSteering (returns a Promise) is awaited — its messages reach the next step (the shell persists before returning)", async () => { + // The shell's drainSteering is async so it can persist the injected + // messages before returning. The kernel must `await` it (a sync call + // would get a Promise, not the array). This test pins that contract. + let drainCallCount = 0; + const steeringMessage: ChatMessage = { + role: "user", + chunks: [{ type: "text", text: "async steer!" }], + }; + 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: () => {}, + // Async drainSteering — resolves on a microtask, like a real persist. + drainSteering: () => + new Promise((resolve) => { + drainCallCount++; + // Defer the resolve so the kernel MUST await to get the array. + queueMicrotask(() => resolve([steeringMessage])); + }), + }); + + expect(drainCallCount).toBe(1); + const secondStepMessages = capturedMessages[1] ?? []; + // The async-returned steering message was awaited and appended AFTER the + // tool result, before the next step — proving the kernel awaited it. + expect(secondStepMessages).toHaveLength(4); + expect(secondStepMessages[3]).toEqual(steeringMessage); + }); + it("MAX_STEPS=0 (unlimited): turn runs past the old 50-step limit and drains at every tool-result boundary until the model stops naturally", async () => { let drainCallCount = 0; // 100 tool-call steps (past the old MAX_STEPS=50) + 1 text-only step diff --git a/packages/kernel/src/runtime/run-turn.ts b/packages/kernel/src/runtime/run-turn.ts index 76e2edf..8f68865 100644 --- a/packages/kernel/src/runtime/run-turn.ts +++ b/packages/kernel/src/runtime/run-turn.ts @@ -720,8 +720,10 @@ export async function runTurn(input: RunTurnInput): Promise<RunTurnResult> { // and append them after the tool results, before the next call. // The kernel owns no queue and names no feature — it just calls // the callback and appends. Emits nothing (caller emits the - // `steering` AgentEvent in its own wrapper). - const steering = input.drainSteering?.() ?? []; + // `steering` AgentEvent in its own wrapper). The callback MAY + // return a Promise (the shell persists the injected messages + // before returning) — `await` handles both sync and async. + const steering = (await input.drainSteering?.()) ?? []; for (const msg of steering) { messages.push(msg); } |
