fix(webui): reconcile a turn that outlives the client's own response stream

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

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) => {

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<void> => {
+    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();

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