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

6 files changed, 266 insertions, 30 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);
         }
diff --git a/packages/session-orchestrator/src/orchestrator.test.ts b/packages/session-orchestrator/src/orchestrator.test.ts
index f61ff21..e67d1b7 100644
--- a/packages/session-orchestrator/src/orchestrator.test.ts
+++ b/packages/session-orchestrator/src/orchestrator.test.ts
@@ -19,6 +19,7 @@ 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 {
@@ -4348,6 +4349,124 @@ describe("in-flight compaction", () => {
       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", () => {
diff --git a/packages/session-orchestrator/src/orchestrator.ts b/packages/session-orchestrator/src/orchestrator.ts
index 85a31ff..ffc5d58 100644
--- a/packages/session-orchestrator/src/orchestrator.ts
+++ b/packages/session-orchestrator/src/orchestrator.ts
@@ -870,17 +870,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
@@ -979,7 +995,12 @@ export function createSessionOrchestrator(
                 emit: deps.emit ?? noopEmit,
               },
               conversationId,
-              { keepLastN, modelName: effectiveModelName },
+              // 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", {
@@ -989,14 +1010,11 @@ export function createSessionOrchestrator(
               });
               return; // too short / empty summary / unknown model → no replacement
             }
-            // Replace the kernel's running history with the summary + the
-            // kernel's OWN most-recent N messages. Using the kernel's messages
-            // (not the store's) preserves mid-turn steering messages and the
-            // vision-transformed provider view the kernel already holds — no
-            // re-transcription needed. The STORE was updated by performCompaction
-            // with the canonical (un-transformed) recent messages.
-            const recent = messages.slice(messages.length - keepLastN);
-            return [outcome.summaryMessage, ...recent];
+            // 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;
           },
         };
 
@@ -1474,15 +1492,33 @@ interface PerformCompactionResult {
    * 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: load the conversation history from the store,
- * 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`) or an
- * error object.
+ * 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
@@ -1491,9 +1527,16 @@ interface PerformCompactionResult {
 async function performCompaction(
   deps: PerformCompactionDeps,
   conversationId: string,
-  opts: { readonly keepLastN?: number; readonly modelName?: 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 }> {
-  const history = await deps.conversationStore.load(conversationId);
+  // 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) {
@@ -1595,7 +1638,10 @@ async function performCompaction(
   const archiveId = crypto.randomUUID();
   await deps.conversationStore.forkHistory(conversationId, archiveId);
 
-  // Replace history: [system: summary] + recent messages
+  // 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: [
@@ -1606,7 +1652,8 @@ async function performCompaction(
     ],
   };
 
-  await deps.conversationStore.replaceHistory(conversationId, [summaryMessage, ...toKeep]);
+  const compactedMessages: readonly ChatMessage[] = [summaryMessage, ...toKeep];
+  await deps.conversationStore.replaceHistory(conversationId, compactedMessages);
   await deps.conversationStore.setCompactedFrom(conversationId, archiveId);
 
   deps.emit(conversationCompacted, {
@@ -1622,6 +1669,7 @@ async function performCompaction(
     messagesSummarized: toSummarize.length,
     messagesKept: toKeep.length,
     summaryMessage,
+    compactedMessages,
   };
 }
 
diff --git a/packages/session-orchestrator/src/queue.test.ts b/packages/session-orchestrator/src/queue.test.ts
index 65cc06f..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 () => {