summaryrefslogtreecommitdiffhomepage
path: root/packages/kernel/src
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-29 00:56:07 +0900
committerAdam Malczewski <[email protected]>2026-06-29 00:56:07 +0900
commitebb29b1e5b929493315469fcbd5eca6f13bd1c32 (patch)
tree3be897eda7c2ab3c41395b76e7ba9fb24d30c463 /packages/kernel/src
parenta5855cad41b2e4b0a196e52cd4c1ca5d4c0ca25f (diff)
downloaddispatch-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/src')
-rw-r--r--packages/kernel/src/contracts/runtime.ts18
-rw-r--r--packages/kernel/src/runtime/run-turn.test.ts47
-rw-r--r--packages/kernel/src/runtime/run-turn.ts6
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);
}