Merge branch 'feature/in-flight-compaction' into predev

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);
         }