From 27f97e5e4ee45e92119e831ff9240c9bf5e80829 Mon Sep 17 00:00:00 2001 From: probe Date: Sat, 24 Oct 2026 13:20:42 +0800 Subject: [PATCH] fix(webui): reconcile a turn that outlives the client's own response stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A response stream that ends says the runtime closed it — not that it finished the turn. The runtime may close the stream early, and the queue may move on to the next turn while this one is ending, and the client had no way to tell the two apart: it took the end of its own stream for the end of the turn, showed the turn as done, and never followed the turn that was still running. The answer to "is a turn still live?" was available the whole time — the runtime's active-turn read — and the command set already used it for a `session.start` it never saw. A local send simply never asked. A send now reconciles the session it landed on once its own stream is done: the same owner the event path uses asks the runtime and re-attaches to whatever is still running. The session to reconcile is the one the send landed on, not the one the command set was built for — a first message creates the session it is sent to and the record moves onto it — so the application resolves it from the completion. A queued submission is left alone. It joined somebody else's turn and never had a stream of its own to end, and the turn it joined is somebody else's stream, which the queue's own events speak for. The runtime's terminal frame is not the shortcut it looks like. It says that *this* turn is over, and it says nothing about the next one: measured against the real runtime, a send whose turn ended while the queue moved on still carried its own `session_status:finished` frame, and the runtime reported the *following* turn as live. So the frame settles this turn and nothing else, and the client's question — is anything still running on this session — is only ever answered by the runtime's read. The four cases that pin this are `packages/webui/test/unit/conversation-turn-commands.test.ts` (the send hands over the session it landed on; a first message hands over the session it created rather than the home slot; a queued submission hands over nothing) and `packages/webui/test/unit/webui-application.test.ts` (the whole shape end to end: the send's own stream ends, the runtime says a turn is still running, and the client attaches to it). The three that assert the reconciliation fail on the trunk — the owner is never asked and the runtime's read is never made — and the queued one holds on the trunk and holds here. --- .../client/application/create-application.ts | 6 ++ .../turn/application/turn-commands.ts | 36 ++++++++- .../unit/conversation-turn-commands.test.ts | 81 +++++++++++++++++++ .../webui/test/unit/webui-application.test.ts | 34 ++++++++ 4 files changed, 155 insertions(+), 2 deletions(-) diff --git a/packages/webui/src/client/application/create-application.ts b/packages/webui/src/client/application/create-application.ts index cddb04ffd..40e0f58ed 100644 --- a/packages/webui/src/client/application/create-application.ts +++ b/packages/webui/src/client/application/create-application.ts @@ -528,6 +528,12 @@ export function createWebuiApplication( refreshQueue: () => { if (sessionId) composerController?.refreshQueue(sessionId); }, + // A turn that outlives the client's own response stream is reconciled + // through the same owner the event path uses, so a local send and a + // missed `session.start` are repaired the same way — and the session to + // reconcile is the one the send landed on, which is not always the one + // this command set was built for. + reconcileTurn: (target) => turnCommandsFor(target).recover(), // No draft writer crosses into Turn: the composer consumes the input it // captured, on the admission the turn owner returns, and a link installed here // would clear a slot without knowing whether it was still the submitted diff --git a/packages/webui/src/client/features/turn/application/turn-commands.ts b/packages/webui/src/client/features/turn/application/turn-commands.ts index 40d20c51c..ba07c1210 100644 --- a/packages/webui/src/client/features/turn/application/turn-commands.ts +++ b/packages/webui/src/client/features/turn/application/turn-commands.ts @@ -419,6 +419,17 @@ export interface WebuiTurnCommandContext { readonly taskSessions?: WebuiTaskSessionSubmitter; /** Refresh the mounted queue projection after an attached turn settles. */ readonly refreshQueue?: () => void; + /** + * Reconcile a session whose turn may still be running after this client's own + * response stream ended. + * + * The application installs this rather than the command running the probe + * itself, because the session to reconcile is not always the one this owner + * was built for: a first message creates the session it is sent to, and the + * record moves onto it — so the target is named by the completion, not by the + * context. + */ + readonly reconcileTurn?: (sessionId: string) => void; } /** What a surface supplies to send: intent and data, never a writer or a sink. */ @@ -440,7 +451,7 @@ export interface WebuiTurnCommandContext { export function createWebuiConversationTurnCommands( context: WebuiTurnCommandContext, ): WebuiConversationTurnCommands { - const { store, sessionId, capabilities, taskSessions, refreshQueue } = context; + const { store, sessionId, capabilities, taskSessions, refreshQueue, reconcileTurn } = context; /** * The record writer for this turn: the session's, or the home slot's before a @@ -632,7 +643,28 @@ export function createWebuiConversationTurnCommands( // The delivery hold covers the attempt, not the decision: releasing it the // moment the ticket exists would let a second submission through while the // first is still dispatching. - void ticket?.completion.finally(hold.release); + void ticket?.completion + .then((completion) => { + hold.release(); + // The response stream is over; the turn is not necessarily over. + // + // A stream that ends says the runtime closed it — not that it finished + // the turn. The runtime may have closed the stream early, or the queue + // may have moved on to the next turn while this one was ending, and + // either way the turn this client was following is not the only + // candidate for "still running". Only the runtime's active-turn read + // says which is true, so the client asks once its own stream is done + // and re-attaches to whatever is still running. + // + // A queued submission has no stream of its own to end, so there is + // nothing here to reconcile: the turn it joined is somebody else's + // stream, and the queue's own events say when it starts. + if (completion.kind !== "dispatched") return; + reconcileTurn?.(completion.sessionId); + }) + .catch(() => { + hold.release(); + }); return ticket; }, attach: (turnId) => { diff --git a/packages/webui/test/unit/conversation-turn-commands.test.ts b/packages/webui/test/unit/conversation-turn-commands.test.ts index eb367c254..21ebd8b32 100644 --- a/packages/webui/test/unit/conversation-turn-commands.test.ts +++ b/packages/webui/test/unit/conversation-turn-commands.test.ts @@ -366,6 +366,87 @@ describe("recheck and gap recovery are the command set's, not the view's", () => }); }); +describe("a response stream that ends is not the end of the turn", () => { + /** A send whose own stream ends immediately: the runtime closed it. */ + const endTheStream = async ( + _request: unknown, + onFrame: (frame: WebuiStreamFrame) => void, + ): Promise => { + onFrame({ cursor: "c1", dataJson: "[DONE]" }); + }; + + it("asks the owner to reconcile the session the send landed on", async () => { + // The client's own stream ending says the runtime closed it — not that it + // finished the turn. The runtime may have closed the stream early, or the + // queue may have moved on to the next turn while this one was ending, and + // only the runtime's active-turn read says which is true. So the send hands + // the session to the owner, which asks that read and re-attaches to + // whatever is still running. + const store = createWebuiSessionStore(); + const reconcileTurn = vi.fn(); + const commands = createWebuiConversationTurnCommands({ + store, + sessionId: "s1", + capabilities: { send: endTheStream as never }, + reconcileTurn, + }); + + await commands.send({ message: "hello" })?.completion; + + expect(reconcileTurn).toHaveBeenCalledWith("s1"); + }); + + it("reconciles the session a first message created, not the home slot", async () => { + // The turn landed on the session the first message created, so that is the + // session whose turn may still be running — and the home slot is no longer + // where the turn is. + const store = createWebuiSessionStore(); + const reconcileTurn = vi.fn(); + const commands = createWebuiConversationTurnCommands({ + store, + sessionId: undefined, + capabilities: { + createSession: (async () => ({ sessionId: "s-created" })) as never, + send: endTheStream as never, + }, + taskSessions: { + creationPending: () => false, + submit: async () => { + store.transferSession(WEBUI_HOME_SESSION_KEY, "s-created", { + expectSourceIncarnation: store.readHomeIncarnation(), + }); + return { kind: "created" as const, sessionId: "s-created", selected: true }; + }, + }, + reconcileTurn, + }); + + await commands.send({ message: "hello" })?.completion; + + expect(reconcileTurn).toHaveBeenCalledWith("s-created"); + }); + + it("does not reconcile a queued submission: it had no stream of its own to end", async () => { + // A queued submission joined somebody else's turn. Its own stream never + // opened, so there is no stream of its own that ended — and the turn it + // joined is somebody else's stream, which the queue's own events speak for. + const store = createWebuiSessionStore(); + const reconcileTurn = vi.fn(); + const enqueueMessage = vi.fn(async () => undefined); + store.createSessionWriter({ kind: "session", sessionId: "s1" }).setSending(true); + const commands = createWebuiConversationTurnCommands({ + store, + sessionId: "s1", + capabilities: { enqueue: enqueueMessage as never }, + reconcileTurn, + }); + + await commands.send({ message: "queued" })?.completion; + + expect(reconcileTurn).not.toHaveBeenCalled(); + }); +}); + describe("an accepted attempt leases its session", () => { it("keeps the session interesting across a navigation and releases it when the attempt settles", async () => { const store = createWebuiSessionStore(); diff --git a/packages/webui/test/unit/webui-application.test.ts b/packages/webui/test/unit/webui-application.test.ts index 896a99b62..a6a481460 100644 --- a/packages/webui/test/unit/webui-application.test.ts +++ b/packages/webui/test/unit/webui-application.test.ts @@ -283,3 +283,37 @@ describe("stage 4 — the four call sites' handling through the coordinator", () expect(ready).toHaveBeenCalledTimes(1); }); }); + +describe("a local send's response stream ending is not the end of the turn", () => { + it("re-attaches to the turn the runtime says is still running", async () => { + // The whole shape, end to end: the client's own stream ends, the runtime + // still has a turn running, and the client asks the runtime instead of + // concluding from its own stream that nothing is running. The answer comes + // back with the turn to follow, and the client follows it — so the output + // of a turn that outlived its stream still reaches the transcript. + const getActiveTurn = vi.fn(async () => ({ + turnId: "turn-live", + busyReason: "turn" as const, + locallyOwned: false, + })); + // A resume that never settles: the attachment stays claimed, which is what + // lets the test read which turn the client attached to. + const resumeSession = vi.fn(async () => new Promise(() => undefined)); + const sendMessage = vi.fn(async () => {}); + const { application } = makeApplication({ + turnCapabilities: { + sendMessage, + resumeSession, + getActiveTurn, + } as never, + }); + + await application.turnCommandsFor("s1").send({ message: "hello" })?.completion; + await new Promise((resolve) => setTimeout(resolve, 0)); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(getActiveTurn).toHaveBeenCalledWith({ id: "s1" }); + expect(resumeSession).toHaveBeenCalled(); + expect(application.store.readSession("s1").stream.subscription?.turnId).toBe("turn-live"); + }); +});