fix(webchat): reattach a reloaded browser to its in-flight turn

docs/designs/webchat-multi-agents.md

@@ -564,6 +564,28 @@ Two scope rules:
   - new `payload.op: 'context'` carrying a `WebchatPost` (transcript-only, no
     ack beyond transport, deduplicated by `postId`);
   - `resume` and `cancel` carry `agentId` as above;
+  - new `payload.op: 'attach' { agentId? }` (daemon capability
+    `webchat-attach-v1`): cold-load stream discovery for a browser that
+    reloaded mid-turn and lost its local turn state. Read-only — the ack names
+    the live stream for (conversation, agent) (`turnId` plus its current
+    resume generation) and the browser follows with an ordinary from-scratch
+    `resume`; the browser envelope is `{type:'attach'}` answered by
+    `{type:'attached'}`. The relay refuses the probe locally
+    (`attached {reason:'unsupported'}`) for a daemon without the capability,
+    and an idle conversation answers `stream_not_found` — both are quiet
+    misses, never errors. This is what restores the typing indicator and the
+    partially streamed reply after a page refresh, instead of both being lost
+    until the turn completes and persists. A cold-attached replay carries no
+    local prompt step, so the console's turn-shaped live/persisted
+    reconciliation has nothing to anchor on — it takes the canonical reply
+    `postId` off the turn's `rd/webchat-post` instead (whichever of that frame
+    and the lane's `done` lands second stamps it onto the replayed steps), and
+    the ordinary exact-`postId` arm retires them when the transcript tail
+    persists the reply. Because that post is now a reconciliation anchor and not
+    just a record, a browser turn's reply post is fanned out to EVERY relay the
+    daemon holds rather than back down the socket that admitted the turn — a
+    reload may land on a different relay instance, and only the one owning the
+    conversation acts on the post;
   - the `set_*` runtime ops are unchanged and carry no `agentId`: multi-agent
     conversations expose no runtime override (section 9.3), so these ops
     occur only in single-agent conversations.

packages/daemon/src/cp/relay-client.ts

@@ -17,6 +17,7 @@ import {
   RD_HEADLESS_AGENT_DELIVERY_V1,
   RD_AGENT_IMPLICIT_ROUTING_V1,
   RD_GITHUB_THREAD_WORKTREE_CLEANUP_V2,
+  RD_WEBCHAT_ATTACH_V1,
   type RelayDaemonFrame,
   type RdHelloOk,
   type RdMsg,
@@ -46,6 +47,8 @@ const DAEMON_RD_CAPABILITIES: readonly string[] = [
   RD_HEADLESS_AGENT_DELIVERY_V1,
   RD_AGENT_IMPLICIT_ROUTING_V1,
   RD_GITHUB_THREAD_WORKTREE_CLEANUP_V2,
+  // The relay refuses the webchat `attach` probe for daemons without this.
+  RD_WEBCHAT_ATTACH_V1,
   // The relay gates gitlab rd/msg dispatch on this capability.
   GITLAB_COM_V1_FEATURE,
   // §24.4: and gates a SELF-MANAGED gitlab delivery on this one, per delivery attempt.
@@ -69,23 +72,8 @@ export interface RelayClientDeps {
   log: Logger
   /** Backoff jitter in [0,1); defaults to Math.random. Injected as `() => 0` in tests. */
   jitter?: () => number
-  /**
-   * Handle one inbound item from the relay (`rd/msg` — a webchat op or a hook
-   * fire): dispatch it (explicit-agent, same engine as the retired CP path) and
-   * return the `rd/ack` verdict. Hook admission may be asynchronous because
-   * its ACK is a durability barrier; webchat/IM handlers remain synchronous.
-   * The `chat` callback streams a
-   * webchat reply back over THIS relay's socket (`rd/chat`); for a hook fire it
-   * is a no-op — the turn's outcome goes to the CP as `hook/report` instead.
-   * The optional `post` callback (webchat only) sends a completed reply as a
-   * canonical conversation post (`rd/webchat-post`) over the same socket so the
-   * relay can fan it to the other participants' daemons as context.
-   */
-  onRelayMsg: (
-    msg: RdMsg,
-    chat: (event: RdChatEvent) => void,
-    post?: (p: RdWebchatPost) => void
-  ) => RdAck | Promise<RdAck>
+  /** Admit one relay delivery; chat streams over this socket, while completed posts fan out through RelayManager. */
+  onRelayMsg: (msg: RdMsg, chat: (event: RdChatEvent) => void) => RdAck | Promise<RdAck>
   /**
    * Handle a forwarded cross-daemon agent-call (`rd/agentmsg/fwd`, agent-collaboration
    * P2): the relay validated the caller and minted a TRUSTED claim. The daemon
@@ -244,8 +232,7 @@ export class RelayClient {
   private async handleMsg(reqId: string, msg: RdMsg): Promise<void> {
     const chat =
       msg.source === 'webchat' ? (event: RdChatEvent) => this.sendChat(msg.chatId, event) : (): void => undefined
-    const post = msg.source === 'webchat' ? (p: RdWebchatPost) => this.sendWebchatPost(p) : undefined
-    const ack = await this.deps.onRelayMsg(msg, chat, post)
+    const ack = await this.deps.onRelayMsg(msg, chat)
     this.transport?.send(JSON.stringify(buildRelayDaemonFrame('rd/ack', ack, { corr: reqId })))
   }
 
@@ -261,12 +248,7 @@ export class RelayClient {
     }
     this.transport?.send(JSON.stringify(buildRelayDaemonFrame('rd/agentmsg/ack', ack, { corr: reqId })))
   }
-  /** One completed conversation post (fire-and-forget EVT) — either on the socket the
-   *  triggering turn arrived on, or (agent-initiated turns, #753) broadcast by
-   *  {@link RelayManager.sendWebchatPost} to every relay this daemon holds, since none
-   *  of them is "the" socket for a wake with no browser turn of its own. A dead
-   *  transport drops it — bounded loss, the transcript row already exists on the
-   *  authoring daemon. */
+  /** Send one completed post on this relay; RelayManager owns daemon-wide fan-out. */
   sendWebchatPost(post: RdWebchatPost): void {
     this.transport?.send(JSON.stringify(buildRelayDaemonFrame('rd/webchat-post', post)))
   }

packages/daemon/src/daemon.ts

@@ -415,7 +415,6 @@ import type {
   SessionKey,
   SessionActivity,
   Ack,
-  RdWebchatPost,
   RdMsg,
   RdMsgHook,
   RdMsgWebchat,
@@ -6047,11 +6046,7 @@ export class Daemon {
    * replay the original ack (so the relay settles) without re-dispatching. For hooks
    * the same replay absorbs a GitHub/manual REDELIVERY of the same deliveryKey.
    */
-  private handleRelayMsg(
-    msg: RdMsg,
-    chat: (event: RdChatEvent) => void,
-    post?: (p: RdWebchatPost) => void
-  ): RdAck | Promise<RdAck> {
+  private handleRelayMsg(msg: RdMsg, chat: (event: RdChatEvent) => void): RdAck | Promise<RdAck> {
     const dedupKey = `${msg.source === 'im' ? `${msg.botId}:` : ''}${msg.sessionKey}:${msg.msgId}`
     const prior = this.relayMsgAcks.get(dedupKey)
     if (prior) {
@@ -6070,7 +6065,7 @@ export class Daemon {
     if (this.dutyCoordinator.dutyEnforced() && !this.duties.holdsAgent(msg.agentId)) {
       const task = this.dutyCoordinator.claimDutyForTrigger(msg.agentId).then((claimed) => {
         this.pendingRelayMsgAcks.delete(dedupKey)
-        if (claimed.granted) return this.handleRelayMsg(msg, chat, post)
+        if (claimed.granted) return this.handleRelayMsg(msg, chat)
         return {
           msgId: msg.msgId,
           accepted: false,
@@ -6084,7 +6079,7 @@ export class Daemon {
 
     const ack =
       msg.source === 'webchat'
-        ? this.dispatchRelayOp(msg, chat, post)
+        ? this.dispatchRelayOp(msg, chat)
         : msg.source === 'platform_action'
           ? this.handleRelayPlatformAction(msg)
           : this.handleRelayIm(msg)
@@ -6968,11 +6963,7 @@ export class Daemon {
   }
 
   /** The op-switch behind {@link handleRelayMsg} (dedup handled by the caller). */
-  private async dispatchRelayOp(
-    msg: RdMsgWebchat,
-    chat: (event: RdChatEvent) => void,
-    post?: (p: RdWebchatPost) => void
-  ): Promise<RdAck> {
+  private async dispatchRelayOp(msg: RdMsgWebchat, chat: (event: RdChatEvent) => void): Promise<RdAck> {
     const sink: WebchatSink = {
       output: (o) => chat({ kind: 'output', output: o }),
       done: (d) => chat({ kind: 'done', done: d })
@@ -6982,9 +6973,9 @@ export class Daemon {
     // Session-targeted continuation: `turn` dispatches onto the target session's
     // own coordinates; runtime-set ops are refused (this ingress adds human
     // input, never session-global administration); a context copy is a no-op
-    // (the roster is fixed at one). resume/cancel/close keep their ordinary
-    // shape — resume is keyed by (turnId, agentId), cancel by the conversation's
-    // own webchat-attached turns.
+    // (the roster is fixed at one). resume/attach/cancel/close keep their ordinary
+    // shape — resume is keyed by (turnId, agentId), attach by (conversation,
+    // agentId), cancel by the conversation's own webchat-attached turns.
     if (msg.targetSessionId !== undefined) {
       switch (op.op) {
         case 'turn':
@@ -7030,7 +7021,6 @@ export class Daemon {
           msg.remoteMcp,
           op.mentions,
           op.post,
-          post,
           op.worktree
         )
         return {
@@ -7074,6 +7064,18 @@ export class Daemon {
           ...(resumed.reason ? { reason: resumed.reason } : {})
         }
       }
+      case 'attach': {
+        // Read-only probe: an accepted verdict names the live stream (turnId +
+        // current generation) and the browser follows with an ordinary resume.
+        const probed = this.webchatTransport.probeWebchatStream(msg.agentId, msg.chatId)
+        return {
+          msgId: msg.msgId,
+          accepted: probed.accepted,
+          ...(probed.turnId ? { turnId: probed.turnId } : {}),
+          ...(probed.generation !== undefined ? { generation: probed.generation } : {}),
+          ...(probed.reason ? { reason: probed.reason } : {})
+        }
+      }
       case 'set_model':
         return (await this.commands.setModelByKey(key(), op.model))
           ? { msgId: msg.msgId, accepted: true }
@@ -15743,7 +15745,7 @@ export class Daemon {
         }),
       log: this.log,
       // Bridge an inbound relay webchat op onto the shared turn engine (webchat, PR 3).
-      onRelayMsg: (msg, chat, post) => this.handleRelayMsg(msg, chat, post),
+      onRelayMsg: (msg, chat) => this.handleRelayMsg(msg, chat),
       // A forwarded cross-daemon agent-call — terminal-verify + dispatch (P2).
       onRelayAgentMsg: (msg) => this.handleRelayAgentMsg(msg)
     })

packages/daemon/src/webchat/transport.ts

@@ -169,7 +169,6 @@ export class WebchatTransport {
     remoteMcp?: WebchatRemoteMcpEntitlement,
     mentions?: string[],
     post?: { postId: string; at: number },
-    postSink?: (p: RdWebchatPost) => void,
     requestedWorktree?: boolean
   ): Promise<WebchatAck> {
     const turnId = requestedTurnId ?? randomUUID()
@@ -275,7 +274,8 @@ export class WebchatTransport {
       remoteMcp,
       requestedWorktree
     )
-    if (postSink) stream.postSink = postSink
+    // Broadcast the reconciliation post daemon-wide so a cold attach through another relay receives it.
+    stream.postSink = (p) => this.host.sendWebchatPost(p)
     // Observed-inbound analogue for webchat (turn-final refresh, §5.4): record the
     // user message at ADMISSION — not only when its turn eventually runs — so a
     // generation already in flight for this agent can see it at the final fence
@@ -630,6 +630,28 @@ export class WebchatTransport {
     else sink.done(event.done)
   }
 
+  /** Cold-load discovery (`attach`, webchat-attach-v1): name the live stream for
+   *  (conversation, agent) so a browser that reloaded mid-turn can resume it from
+   *  scratch. Read-only — the follow-up `resume` does the rebind + replay. */
+  probeWebchatStream(
+    agentId: string,
+    conversationId: string
+  ): { accepted: boolean; turnId?: string; generation?: number; reason?: string } {
+    this.pruneWebchatStreams()
+    let match: WebchatTurnStream | undefined
+    for (const stream of this.webchatStreams.values()) {
+      if (stream.agentId !== agentId || stream.conversationId !== conversationId) continue
+      // A completed turn is already in the transcript; nothing live to reattach.
+      if (stream.completedAt !== undefined || stream.replayDisabled) continue
+      match = stream // insertion order: the last match is the newest admitted turn
+    }
+    if (!match) return { accepted: false, reason: 'stream_not_found' }
+    // A trimmed replay head cannot rebuild the reply from scratch — refuse; the
+    // transcript covers it at turn end.
+    if (match.replayFloor > 0) return { accepted: false, turnId: match.turnId, reason: 'stream_gap' }
+    return { accepted: true, turnId: match.turnId, generation: match.resumeGeneration }
+  }
+
   resumeWebchatStream(
     agentId: string,
     conversationId: string,

packages/daemon/test/cp/relay-client.test.ts

@@ -6,6 +6,7 @@ import {
   RD_HEADLESS_AGENT_DELIVERY_V1,
   RD_AGENT_IMPLICIT_ROUTING_V1,
   RD_GITHUB_THREAD_WORKTREE_CLEANUP_V2,
+  RD_WEBCHAT_ATTACH_V1,
   RELAY_DAEMON_SUBPROTOCOL,
   type RelayDaemonFrame,
   type RdMsg,
@@ -114,6 +115,7 @@ describe('RelayClient (daemon → one relay)', () => {
         RD_HEADLESS_AGENT_DELIVERY_V1,
         RD_AGENT_IMPLICIT_ROUTING_V1,
         RD_GITHUB_THREAD_WORKTREE_CLEANUP_V2,
+        RD_WEBCHAT_ATTACH_V1,
         GITLAB_COM_V1_FEATURE,
         GITLAB_INSTANCE_V1_FEATURE
       ]

packages/daemon/test/daemon-webchat-continuation.test.ts

@@ -75,8 +75,7 @@ describe('webchat multi-agent continuation (#549 parity)', () => {
         },
         { agentId: P1, msgId: 'turn-p1' }
       ),
-      (_e: RdChatEvent) => {},
-      fanOut
+      (_e: RdChatEvent) => {}
     )
     expect(ack).toMatchObject({ accepted: true })
     for (const [peer, msgId] of [
@@ -184,8 +183,7 @@ describe('webchat multi-agent continuation (#549 parity)', () => {
         },
         { agentId: P1, msgId: 'turn-p1' }
       ),
-      () => {},
-      fanOut
+      () => {}
     )
 
     // Posts at depths 0..MAX-1; the wake that would run at depth MAX is refused.
@@ -316,8 +314,7 @@ describe('webchat multi-agent continuation (#549 parity)', () => {
         { op: 'turn', text: 'go', user: 'owner', turnId: KICKOFF_TURN, post: { postId: KICKOFF_TURN, at: 1_000 } },
         { agentId: P1, msgId: 'turn-p1' }
       ),
-      () => {},
-      fanOut
+      () => {}
     )
     await vi.waitFor(() => expect(prompts.get(P1)).toHaveLength(1), WAIT)
     await settle()