From a4e1641d131a9ca3d0f068967dc1ef577da82b15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=B4=AB=E9=9B=A8?= Date: Fri, 11 Sep 2026 18:16:58 +0800 Subject: [PATCH] fix(gateway): retain pending inputs when send becomes steer --- .../2026-09-11-steer-consumption-recovery.md | 55 +++++++++++++++++ src/gateway/server-chat-abort.test.ts | 61 ++++++++++++++++++- src/gateway/server.ts | 15 ++++- 3 files changed, 126 insertions(+), 5 deletions(-) create mode 100644 docs/completion/2026-09-11-steer-consumption-recovery.md diff --git a/docs/completion/2026-09-11-steer-consumption-recovery.md b/docs/completion/2026-09-11-steer-consumption-recovery.md new file mode 100644 index 000000000..079785929 --- /dev/null +++ b/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. diff --git a/src/gateway/server-chat-abort.test.ts b/src/gateway/server-chat-abort.test.ts index fec971af3..5c3f35c5c 100644 --- a/src/gateway/server-chat-abort.test.ts +++ b/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) | 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 }) => { 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"); + }); +}); diff --git a/src/gateway/server.ts b/src/gateway/server.ts index c1e1be419..d73ce4178 100644 --- a/src/gateway/server.ts +++ b/src/gateway/server.ts @@ -916,6 +916,7 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise 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 { 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