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