fix(gateway): preserve consumed inputs when send becomes steer

docs/completion/2026-09-11-steer-consumption-recovery.md

@@ -0,0 +1,55 @@
+# Preserve consumption acknowledgements for concurrent sends
+
+When a concurrent `chat.send` joins an active turn as a steer, its background
+handler returns before the owning stream receives the user-message consumption
+acknowledgement. Its unconditional cleanup previously removed the session's
+pending user rows. An input the AgentBox had consumed could consequently remain
+unsequenced and be excluded when conversation history was reconstructed.
+
+Only a handler that acquired the turn lock and did not steer into an existing turn
+now clears the pending queue. Both Runtime-busy and AgentBox-409 fallback paths
+retain their entries. The owning turn still discards inputs it did not consume
+when it ends. RPC shapes and persistence schemas are unchanged.
+
+## Automated validation
+
+Validated on official `main` at `d9505a52` on 2026-09-14:
+
+- The real `chat.send` handlers, with the AgentBox and persistence calls mocked,
+  exercise a long-running reply, concurrent input, the original stream's delayed
+  consumption callback, and the real `loadFullHistory` filter. A separate case
+  covers the AgentBox-409 fallback; owner cleanup is also checked.
+- Both new regression cases fail on the original implementation and pass with
+  the fix.
+- Five targeted Runtime test files passed, covering 159 cases.
+- Full suite: 377 files and 7,607 tests passed.
+- `npx tsc --noEmit`, `npm run build`, and `git diff --check` passed.
+
+The review follow-up clarified that 409 retention depends on pi replaying an
+unconsumed steer on a later prompt and removed fixed sleeps from the two cases.
+Both cases still failed with an unconditional sweep; all 42 cases in the two
+affected Runtime test files passed with the guard. TypeScript and diff checks
+passed, and transpilation without comments confirmed unchanged Runtime code.
+Full-suite, build, and deployed Runtime validation were completed before this
+comment and test-only follow-up.
+
+## Deployed Runtime validation
+
+The Runtime behavior was also checked on a test deployment through an integration
+client. This validates the concurrent-send contract rather than claiming every
+client permits concurrent sends to an active conversation.
+
+Before the fix, Runtime logs confirmed an input was steered into the active turn
+and the model acknowledged it, but its persisted user row remained
+`seq_sequenced = 0`.
+
+After the fix, the same path retained the input until consumption. Its user row
+was sequenced between the original answer and the acknowledgement. After the
+AgentBox was automatically recycled, a new instance with empty transient session
+storage fetched history from the control plane. Its log confirmed reconstruction,
+and inspection of the new transcript found the original user message exactly
+once. A subsequent model response reproduced the complete message without tools.
+
+The check did not modify stored consumption flags or seed a session cache. It
+confirms preservation of consumed inputs through actual database-backed history
+reconstruction. Previously incorrect historical flags are not backfilled.

src/gateway/server-chat-abort.test.ts

@@ -12,8 +12,9 @@
  */
 import { describe, it, expect, afterEach, vi } from "vitest";
 
-const bindMessageTraceIdMock = vi.hoisted(() => vi.fn(async () => {}));
+const bindMessageTraceIdMock = vi.hoisted(() => vi.fn(async (..._args: unknown[]) => {}));
 const updateMessageMock = vi.hoisted(() => vi.fn(async () => {}));
+const sequenceMessageMock = vi.hoisted(() => vi.fn(async (_id: string, _sessionId: string) => {}));
 const steerSessionMock = vi.hoisted(() =>
   vi.fn(async () => ({ ok: true, traceId: "fedcba9876543210fedcba9876543210" })),
 );
@@ -23,6 +24,7 @@ vi.mock("./chat-repo.js", async (importOriginal) => ({
   appendMessage: vi.fn(async () => "msg-id"),
   bindMessageTraceId: bindMessageTraceIdMock,
   updateMessage: updateMessageMock,
+  sequenceMessage: sequenceMessageMock,
   incrementMessageCount: vi.fn(async () => {}),
   warnTraceBindFailure: vi.fn(),
   // The real validator: the delegated-turn ledger gates the box ack through it.
@@ -38,14 +40,16 @@ vi.mock("./output-redactor.js", () => ({
 // that is mid-tool when the user hits Stop. capturedSignal lets the test observe
 // whether chat.abort actually aborted it.
 let capturedSignal: AbortSignal | undefined;
+let consumeUserMessage: ((text: string) => Promise<void>) | undefined;
 // A real consumer notices its abort only when the NEXT event arrives, so a turn can
 // stay live across two supervisor passes. Set this to model that.
 let consumerIgnoresAbort = false;
 // Settles the mocked consumer on demand, to model a turn finishing NORMALLY mid-drain.
 let settleConsumer: (() => void) | undefined;
 vi.mock("./sse-consumer.js", () => ({
-  consumeAgentSse: vi.fn((opts: { signal?: AbortSignal }) => {
+  consumeAgentSse: vi.fn((opts: { signal?: AbortSignal; onUserMessageStarted?: (text: string) => Promise<void> }) => {
     capturedSignal = opts.signal;
+    consumeUserMessage = opts.onUserMessageStarted;
     return new Promise((resolve) => {
       const done = () =>
         resolve({ resultText: "", taskReportText: "", errorMessage: "", eventCount: 0, durationMs: 0 });
@@ -103,6 +107,8 @@ vi.mock("./agentbox/client.js", () => ({
 }));
 
 const { startRuntime } = await import("./server.js");
+const { pendingUserRows } = await import("./pending-user-rows.js");
+const { loadFullHistory } = await import("./session-history-api.js");
 
 function fakeFrontendClient() {
   return {
@@ -147,6 +153,8 @@ afterEach(async () => {
   if (server) await server.close();
   server = undefined;
   capturedSignal = undefined;
+  consumeUserMessage = undefined;
+  pendingUserRows.clear("S");
   consumerIgnoresAbort = false;
   settleConsumer = undefined;
   abortSessionCalls.length = 0;
@@ -997,3 +1005,52 @@ describe("startRuntime — turnId is derived from the dispatch", () => {
     settleConsumer?.();
   });
 });
+
+
+describe("concurrent sends preserve consumption acknowledgements", () => {
+  it("retains a busy steer until the owning stream consumes it, then restores it as history", async () => {
+    const manager = fakeAgentBoxManager();
+    manager.getOrCreate.mockResolvedValue({ boxId: "box-a", endpoint: "https://fake.internal" });
+    server = await bootRuntime(manager);
+    const send = server.rpcMethods.get("chat.send")!;
+    const ctx = { sendEvent: vi.fn() };
+    await send({ agentId: "a", userId: "u", sessionId: "S", text: "inspect", skipInitialPersistence: true, persistedInput: true, userMessageId: "q1" }, ctx);
+    await waitFor(() => consumeUserMessage !== undefined);
+    await consumeUserMessage!("inspect");
+    expect(sequenceMessageMock).toHaveBeenCalledWith("q1", "S");
+
+    // The first stream is waiting for a long tool. The second RPC returns after
+    // steering, but does not own either the stream or its pending input queue.
+    await send({ agentId: "a", userId: "u", sessionId: "S", text: "also inspect logs", skipInitialPersistence: true, persistedInput: true, userMessageId: "q2" }, { sendEvent: vi.fn() });
+    // After invoking trace binding, the handler returns through finally without
+    // another await, so observing this call also observes its cleanup.
+    await waitFor(() => bindMessageTraceIdMock.mock.calls.some(([id]) => id === "q2"), 8000);
+    expect(pendingUserRows.size("S")).toBe(1);
+    expect(sequenceMessageMock).not.toHaveBeenCalledWith("q2", "S");
+
+    await consumeUserMessage!("also inspect logs");
+    expect(sequenceMessageMock).toHaveBeenCalledWith("q2", "S");
+    expect(pendingUserRows.size("S")).toBe(0);
+    const sequenced = sequenceMessageMock.mock.calls.some(([id]) => id === "q2");
+    const history = await loadFullHistory("S", async () => [{
+      id: "q2", sessionId: "S", role: "user", content: "also inspect logs", seqSequenced: sequenced,
+      toolName: null, toolset: null, toolInput: null, metadata: { kind: "steer" }, outcome: null, durationMs: null,
+      fromAgentId: null, parentSessionId: null, delegationId: null, targetAgentId: null, createdAt: new Date(),
+    }]);
+    expect(history).toHaveLength(1);
+    expect(history[0].content).toBe("also inspect logs");
+
+    pendingUserRows.push("S", "not-consumed", "too late");
+    settleConsumer?.();
+    await waitFor(() => pendingUserRows.size("S") === 0);
+  }, 15000);
+
+  it("keeps the AgentBox 409 fallback queued after its RPC returns", async () => {
+    promptError = new Error("Session is already running");
+    server = await bootRuntime();
+    await server.rpcMethods.get("chat.send")!({ agentId: "a", userId: "u", text: "queued via 409", sessionId: "S", skipInitialPersistence: true, persistedInput: true, userMessageId: "q409" }, { sendEvent: vi.fn() });
+    // This branch also reaches finally synchronously after invoking trace binding.
+    await waitFor(() => bindMessageTraceIdMock.mock.calls.some(([id]) => id === "q409"));
+    expect(pendingUserRows.claim("S", "queued via 409")).toBe("q409");
+  });
+});

src/gateway/server.ts

@@ -916,6 +916,7 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise<RuntimeSe
       // dispatched to two boxes and both would run — two writers on one transcript.
       // Released in the finally below so a throw cannot wedge the session.
       let releaseTurn: (() => void) | undefined;
+      let steeredIntoExistingTurn = false;
       try {
         // One turn at a time for this session, across every box (see session-turn-lock.ts).
         // If the session is already running, fall back to the SAME steer the AgentBox's own
@@ -936,6 +937,7 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise<RuntimeSe
             .catch((e) => { console.warn(`[runtime] steer into ${running.boxId} failed session=${sessionId}:`, e); return undefined; })
             : undefined;
           if (steered) {
+            steeredIntoExistingTurn = true;
             console.log(`[runtime] session=${sessionId} busy; steered into the turn on ${running!.boxId}`);
             // The input WAS delivered (it rides the in-flight turn), so this
             // dispatch happened: confirm it rather than leaving it pending.
@@ -1048,9 +1050,14 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise<RuntimeSe
           if (err instanceof Error && err.message.includes("Session is already running")) {
             if (requiredResultToolName) throw err;
             const steerResult = await client.steerSession(sessionId, text, { images, files });
+            steeredIntoExistingTurn = true;
             // Delivered as a steer onto the running turn — the dispatch did
             // take effect, so it stays recorded (see the busy-degrade path).
             confirmDispatch(dispatchKey);
+            // We hold the Runtime lock, so this box turn has no consumer here.
+            // Keep the row for an unconsumed steer that pi can replay on the next
+            // prompt, where onUserMessageStarted can sequence it. This depends
+            // on pi retaining its queue; the AgentBox abort path clears that queue.
             if (promptMessageId) pendingUserRows.push(sessionId, promptMessageId, text);
             // chat.send persisted this row before it knew the active session would
             // reject a fresh prompt. Once the fallback steer is accepted, label the
@@ -1209,9 +1216,11 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise<RuntimeSe
           context.sendEvent("chat.event", { sessionId, turnId, event: { type: "prompt_done" } });
         }
       } finally {
-        // Anything still queued was never consumed — a steer the user sent into a turn
-        // that finished first. It must not be claimed by the next turn's first echo.
-        pendingUserRows.clear(sessionId);
+        // Only the owning turn may discard its unconsumed inputs. A concurrent
+        // send that degraded to steer returns before the active stream consumes
+        // that input; clearing here would lose its sequence acknowledgement and
+        // cause cold restore to filter out a question the box actually answered.
+        if (releaseTurn && !steeredIntoExistingTurn) pendingUserRows.clear(sessionId);
         unregisterPendingStart(sessionId, turnAbort);
         dropLiveTurn(sessionId, turnId);
         releaseSandboxTurn();