From ebb29b1e5b929493315469fcbd5eca6f13bd1c32 Mon Sep 17 00:00:00 2001
From: Adam Malczewski <github@tradam.dev>
Date: Mon, 29 Jun 2026 00:56:07 +0900
Subject: fix(in-flight-compaction): persist steering messages + use live
 messages array for compaction
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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).
---
 packages/kernel/src/runtime/run-turn.test.ts | 47 ++++++++++++++++++++++++++++
 1 file changed, 47 insertions(+)

(limited to 'packages/kernel/src/runtime/run-turn.test.ts')

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
-- 
cgit v1.2.3