From d1fc77ba32bb0581da4742c74da78412c75407e9 Mon Sep 17 00:00:00 2001 From: Dazhan Date: Wed, 26 Aug 2026 17:17:51 +0800 Subject: [PATCH 1/3] fix(webchat): reattach a reloaded browser to its in-flight turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A page refresh mid-turn lost the typing indicator and the streamed reply: all live-turn state (busy flag, stream-lane cursors, streamed steps) lived only in browser memory, the persisted transcript gets the reply only at turn end, and nothing let a cold-loaded client rediscover the stream — resume requires the exact turnId the reload just wiped. Add cold-load stream discovery on the existing replay machinery: - protocol: new RelayWebchatOp `attach { agentId? }` (read-only probe), RdAck gains optional `generation`, capability `webchat-attach-v1` - daemon: `probeWebchatStream` names the live stream for a (conversation, agent) — turnId plus current resume generation; refuses trimmed-replay (`stream_gap`) and completed/idle streams - relay: parses `{type:'attach'}`, forwards it per participant gated on the daemon capability, answers `{type:'attached'}`; every miss is a quiet per-agent refusal, never an error frame - web: opening a webchat session detail probes the roster over the conversation socket; a hit recreates the lane (generation seeded from the daemon), restores the busy indicator, and replays the reply from scratch through the ordinary resume path. Idle `error` frames no longer push a warning step (an older relay answers the probe that way). Co-Authored-By: Claude Fable 5 --- docs/designs/webchat-multi-agents.md | 12 ++++ packages/daemon/src/cp/relay-client.ts | 3 + packages/daemon/src/daemon.ts | 18 +++++- packages/daemon/src/webchat/transport.ts | 22 +++++++ packages/daemon/test/daemon-webchat.test.ts | 62 +++++++++++++++++++ .../protocol/src/frames/relay-daemon.test.ts | 8 +++ packages/protocol/src/frames/relay-daemon.ts | 17 +++++ .../src/relay-browser-connection.test.ts | 45 +++++++++++++- .../relay/src/relay-browser-connection.ts | 42 ++++++++++--- .../components/console/PlaygroundProvider.tsx | 58 ++++++++++++++++- .../console/views/SessionDetailView.tsx | 10 +++ .../views/SessionDetailView.viewer.test.tsx | 1 + 12 files changed, 284 insertions(+), 14 deletions(-) diff --git a/docs/designs/webchat-multi-agents.md b/docs/designs/webchat-multi-agents.md index d8d25be8d..9425c3190 100644 --- a/docs/designs/webchat-multi-agents.md +++ b/docs/designs/webchat-multi-agents.md @@ -564,6 +564,18 @@ 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; - 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. diff --git a/packages/daemon/src/cp/relay-client.ts b/packages/daemon/src/cp/relay-client.ts index 89c23cad7..1eb767411 100644 --- a/packages/daemon/src/cp/relay-client.ts +++ b/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. diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index fc8bcc471..712fd0a96 100644 --- a/packages/daemon/src/daemon.ts +++ b/packages/daemon/src/daemon.ts @@ -6831,9 +6831,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': @@ -6923,6 +6923,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 } diff --git a/packages/daemon/src/webchat/transport.ts b/packages/daemon/src/webchat/transport.ts index 06ef5b2e9..8fa37419d 100644 --- a/packages/daemon/src/webchat/transport.ts +++ b/packages/daemon/src/webchat/transport.ts @@ -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, diff --git a/packages/daemon/test/daemon-webchat.test.ts b/packages/daemon/test/daemon-webchat.test.ts index bf8f75aee..1a8633b94 100644 --- a/packages/daemon/test/daemon-webchat.test.ts +++ b/packages/daemon/test/daemon-webchat.test.ts @@ -1730,6 +1730,68 @@ describe('Daemon handleRelayMsg (rd/msg op dispatch — the relay data plane)', () => {} ) ).toMatchObject({ accepted: false, turnId, reason: 'stream_gap' }) + + // A from-scratch reattach cannot rebuild past the trimmed head either — the + // cold-load probe refuses the same way instead of naming an unresumable turn. + expect( + await (daemon as any).handleRelayMsg(rd({ op: 'attach' }, { msgId: 'attach-overflow' }), () => {}) + ).toMatchObject({ accepted: false, turnId, reason: 'stream_gap' }) + await daemon.stop() + }) + + it('attach names the live stream so a reloaded browser can resume it from scratch', async () => { + const { factory } = streamingHost([]) + const daemon = new Daemon({ root: scaffold(), hostFactory: factory }) + await daemon.start() + + const turnId = '77777777-7777-4777-8777-777777777777' + // Idle conversation: the probe answers the quiet not-found, never an error. + expect( + await (daemon as any).handleRelayMsg(rd({ op: 'attach' }, { msgId: 'attach-idle' }), () => {}) + ).toMatchObject({ accepted: false, reason: 'stream_not_found' }) + + const pre: RdChatEvent[] = [] + const stream = (daemon as any).webchatTransport.createWebchatTurnStream(AGENT_ID, CONV, turnId, { + output: (output: WebchatOutput) => pre.push({ kind: 'output', output }), + done: (done: WebchatDone) => pre.push({ kind: 'done', done }) + }) + stream.sink.output({ conversationId: CONV, turnId, index: 0, event: { kind: 'message', text: 'partial' } }) + + // The probe is read-only: it names the turn + its current resume generation and + // replays nothing itself — the follow-up resume does the rebind + replay. + const probeEvents: RdChatEvent[] = [] + expect( + await (daemon as any).handleRelayMsg(rd({ op: 'attach' }, { msgId: 'attach-live' }), (event: RdChatEvent) => + probeEvents.push(event) + ) + ).toMatchObject({ accepted: true, turnId, generation: 0 }) + expect(probeEvents).toEqual([]) + + const replayed: RdChatEvent[] = [] + expect( + await (daemon as any).handleRelayMsg( + rd({ op: 'resume', turnId, generation: 1, afterIndex: -1 }, { msgId: 'attach-resume' }), + (event: RdChatEvent) => replayed.push(event) + ) + ).toMatchObject({ accepted: true, turnId }) + expect(replayed).toEqual([ + { + kind: 'output', + output: { + conversationId: CONV, + turnId, + agentId: AGENT_ID, + index: 0, + event: { kind: 'message', text: 'partial' } + } + } + ]) + + // A completed turn is transcript-owned — the probe stops naming it. + stream.sink.done({ conversationId: CONV, turnId, stopReason: 'end_turn' }) + expect( + await (daemon as any).handleRelayMsg(rd({ op: 'attach' }, { msgId: 'attach-done' }), () => {}) + ).toMatchObject({ accepted: false, reason: 'stream_not_found' }) await daemon.stop() }) diff --git a/packages/protocol/src/frames/relay-daemon.test.ts b/packages/protocol/src/frames/relay-daemon.test.ts index 1e6b34e21..57fa729da 100644 --- a/packages/protocol/src/frames/relay-daemon.test.ts +++ b/packages/protocol/src/frames/relay-daemon.test.ts @@ -188,6 +188,8 @@ describe('relay↔daemon wire — skeleton frame codec (shared-bot-relay.md §7. runtime: { model: 'gpt-5.6-sol', effort: 'xhigh', permissionMode: 'full-access', fastMode: true } }, { op: 'resume', turnId: TURN_ID, generation: 2, afterIndex: 3 }, + { op: 'attach' }, + { op: 'attach', agentId: AGENT_ID }, { op: 'set_model', model: 'opus-4.8' }, { op: 'set_effort', effort: 'high' }, { op: 'set_permission_mode', permissionMode: 'plan' }, @@ -213,6 +215,12 @@ describe('relay↔daemon wire — skeleton frame codec (shared-bot-relay.md §7. expect(RelayWebchatOp.safeParse({ op: 'resume', turnId: TURN_ID, generation: 0, afterIndex: 0 }).success).toBe( false ) + expect(RelayWebchatOp.safeParse({ op: 'attach', agentId: 'not-a-uuid' }).success).toBe(false) + }) + + it('rd/ack carries the attach probe verdict (turnId + resume generation)', () => { + expect(RdAck.safeParse({ msgId: 'm-1', accepted: true, turnId: TURN_ID, generation: 0 }).success).toBe(true) + expect(RdAck.safeParse({ msgId: 'm-1', accepted: true, turnId: TURN_ID, generation: -1 }).success).toBe(false) }) it('carries one bounded inline image on a webchat turn', () => { diff --git a/packages/protocol/src/frames/relay-daemon.ts b/packages/protocol/src/frames/relay-daemon.ts index 912828c1f..9b83ecba4 100644 --- a/packages/protocol/src/frames/relay-daemon.ts +++ b/packages/protocol/src/frames/relay-daemon.ts @@ -86,6 +86,15 @@ export const RD_AGENT_IMPLICIT_ROUTING_V1 = 'agent-implicit-routing-v1' */ export const RD_GITHUB_THREAD_WORKTREE_CLEANUP_V2 = 'github-thread-worktree-cleanup-v2' +/** + * `webchat-attach-v1`: this daemon answers the webchat `attach` probe — naming the + * live reply stream for (conversation, agent) so a browser that reloaded mid-turn + * can rediscover and resume it. The relay refuses the probe locally + * (`attached {reason:'unsupported'}`) for a daemon without it, instead of + * forwarding an op an older daemon cannot parse. + */ +export const RD_WEBCHAT_ATTACH_V1 = 'webchat-attach-v1' + // D→R REQ → rd/hello/ok. The daemon presents the same credential it uses on the CP // socket — an API key, or an in-cluster daemon's projected ServiceAccount token. The // relay holds no database, so it delegates either to the CP via `rc/verify` and caches @@ -184,6 +193,11 @@ export const RelayWebchatOp = z.discriminatedUnion('op', [ generation: z.number().int().min(1).max(Number.MAX_SAFE_INTEGER), afterIndex: z.number().int().min(-1) }), + // Cold-load stream discovery (a page reload lost the browser's turn state): name + // the LIVE stream for (conversation, agent). Read-only — the accepted ack carries + // the stream's turnId + current resume generation, and the browser follows with an + // ordinary `resume` from scratch; nothing rebinds until that resume lands. + z.object({ op: z.literal('attach'), agentId: z.string().uuid().optional() }), z.object({ op: z.literal('set_model'), model: z.string() }), z.object({ op: z.literal('set_effort'), effort: z.string() }), z.object({ op: z.literal('set_permission_mode'), permissionMode: z.string() }), @@ -443,6 +457,9 @@ export const RdAck = z.object({ reason: z.string().optional(), /** Bounded human-readable cause for a refusal the browser should explain (see WebchatAck.detail). */ detail: z.string().max(240).optional(), + /** `attach` verdicts only: the named stream's current resume generation — the browser + * seeds its cursor from it so the follow-up `resume` outruns pre-reload generations. */ + generation: z.number().int().min(0).optional(), /** Set with `reason: 'not_holder'`: the member that holds the duty now, as the * losing claimant learned it from the CP. Absent when even the CP could not * name one — the router then retries rather than re-routing. */ diff --git a/packages/relay/src/relay-browser-connection.test.ts b/packages/relay/src/relay-browser-connection.test.ts index 4be61bb1e..422261af4 100644 --- a/packages/relay/src/relay-browser-connection.test.ts +++ b/packages/relay/src/relay-browser-connection.test.ts @@ -51,6 +51,7 @@ function build( log?: Logger participants?: Array<{ agentId: string; daemonId?: string; primary?: boolean }> targetSessionId?: string + supports?: boolean } = {} ) { const sent: RdMsgWebchat[] = [] @@ -58,7 +59,10 @@ function build( sent.push(m) return over.ack ?? { msgId: m.msgId, accepted: true } }) - const daemon = 'daemon' in over ? over.daemon : ({ sendMsg } as unknown as RelayDaemonConnection) + const daemon = + 'daemon' in over + ? over.daemon + : ({ sendMsg, supports: () => over.supports !== false } as unknown as RelayDaemonConnection) const register = vi.fn() const unregister = vi.fn() const transport = new FakeBrowserTransport() @@ -152,6 +156,11 @@ describe('parseBrowserFrame', () => { op: { op: 'set_fast', fastMode: true } }) expect(parseBrowserFrame({ type: 'cancel' }, USER)).toEqual({ op: { op: 'cancel' } }) + expect(parseBrowserFrame({ type: 'attach' }, USER)).toEqual({ op: { op: 'attach' } }) + expect(parseBrowserFrame({ type: 'attach', agentId: AGENT }, USER)).toEqual({ + op: { op: 'attach', agentId: AGENT } + }) + expect(parseBrowserFrame({ type: 'attach', agentId: 'not-a-uuid' }, USER)).toBeNull() }) it('preserves structured mentions on the turn op and surfaces targets separately', () => { const PEER = '22222222-2222-4222-8222-222222222222' @@ -322,6 +331,40 @@ describe('RelayBrowserConnection', () => { }) }) + it('forwards an attach probe and surfaces the named stream on {type:"attached"}', async () => { + const turnId = '22222222-2222-4222-8222-222222222222' + const { transport, sent } = build({ ack: { msgId: 'attach-1', accepted: true, turnId, generation: 2 } }) + transport.feed({ type: 'attach', agentId: AGENT }) + await tick() + expect(sent[0]).toMatchObject({ agentId: AGENT, payload: { op: 'attach', agentId: AGENT } }) + expect(transport.last('attached')).toEqual({ + type: 'attached', + ack: { accepted: true, turnId, agentId: AGENT, generation: 2 } + }) + }) + + it('refuses the attach probe locally for a daemon without webchat-attach-v1', async () => { + const { transport, sent } = build({ supports: false }) + transport.feed({ type: 'attach', agentId: AGENT }) + await tick() + expect(sent).toHaveLength(0) + expect(transport.last('attached')).toEqual({ + type: 'attached', + ack: { accepted: false, agentId: AGENT, reason: 'unsupported' } + }) + }) + + it('answers the attach probe with a quiet per-agent refusal when the daemon is offline', async () => { + const { transport } = build({ daemon: undefined }) + transport.feed({ type: 'attach' }) + await tick() + expect(transport.last('attached')).toEqual({ + type: 'attached', + ack: { accepted: false, agentId: AGENT, reason: 'no_agent' } + }) + expect(transport.last('error')).toBeUndefined() + }) + // A stream that "never came back" is diagnosed from these lines alone: who joined/left the // conversation (with the close code), and which daemon refused a resume, and why. it('logs the browser join/leave and every refused resume with the daemon and reason', async () => { diff --git a/packages/relay/src/relay-browser-connection.ts b/packages/relay/src/relay-browser-connection.ts index e242e5716..a1309b8e3 100644 --- a/packages/relay/src/relay-browser-connection.ts +++ b/packages/relay/src/relay-browser-connection.ts @@ -6,8 +6,8 @@ * * It speaks the browser-facing, type-tagged webchat envelope. The browser sends * `{text, turnId, mentions?, targets?, attachments?, runtime?}` (a turn) or - * `{type:'resume'|'set_model'|'set_effort'|'set_permission_mode'|'set_fast'|'cancel'}`, - * and the relay sends `{type:'ready'|'output'|'done'|'ack'|'resumed'|'post'|'error'}`. + * `{type:'resume'|'attach'|'set_model'|'set_effort'|'set_permission_mode'|'set_fast'|'cancel'}`, + * and the relay sends `{type:'ready'|'output'|'done'|'ack'|'resumed'|'attached'|'post'|'error'}`. * A turn fans out as one pre-addressed `rd/msg(webchat)` per targeted agent's * daemon (targets are validated against the verified roster) plus a transcript-only * `context` copy to every other participant's daemon; each `rd/chat` chunk a daemon @@ -23,6 +23,7 @@ import { ErrorCode, RelayWebchatOp, RD_ACK_NOT_HOLDER, + RD_WEBCHAT_ATTACH_V1, type RdChat, type RdMsgWebchat, type RdWebchatPost, @@ -163,6 +164,12 @@ export function parseBrowserFrame(msg: unknown, user: string, userId?: string): } } : null + case 'attach': + // Cold-load probe: name the live stream for (conversation, agent) so a reloaded + // browser can resume it. `agentId` defaults to the primary at dispatch. + return m.agentId === undefined || (typeof m.agentId === 'string' && UUID_RE.test(m.agentId)) + ? { op: { op: 'attach', ...(typeof m.agentId === 'string' ? { agentId: m.agentId.toLowerCase() } : {}) } } + : null case 'set_model': return typeof m.model === 'string' ? { op: { op: 'set_model', model: m.model } } : null case 'set_effort': @@ -286,10 +293,13 @@ export class RelayBrowserConnection implements ChatSink { for (const p of this.byAgentId.values()) void this.sendToParticipant(p.agentId, { op: 'cancel' }, 'cancel') return } - // Single-daemon ops: resume/cancel go to the named participant, everything + // Single-daemon ops: resume/attach/cancel go to the named participant, everything // else (set_*) to the primary — multi-agent conversations expose no runtime // override (webchat-multi-agents.md §9.3), so set_* only occurs single-agent. - const targetAgent = op.op === 'resume' || op.op === 'cancel' ? (op.agentId ?? this.deps.agentId) : this.deps.agentId + const targetAgent = + op.op === 'resume' || op.op === 'attach' || op.op === 'cancel' + ? (op.agentId ?? this.deps.agentId) + : this.deps.agentId await this.sendToParticipant(targetAgent, op, op.op) } @@ -356,6 +366,12 @@ export class RelayBrowserConnection implements ChatSink { } } if (!daemon) { + // The attach probe is background discovery — always a quiet per-agent + // refusal, never the legacy error frame. + if (kind === 'attach') { + this.send({ type: 'attached', ack: { accepted: false, agentId, reason: 'no_agent' } }) + return + } // A single-participant conversation keeps the legacy error frame; a // multi-agent one degrades per agent so the other targets still run. if (this.byAgentId.size === 1) { @@ -376,6 +392,12 @@ export class RelayBrowserConnection implements ChatSink { } return } + // Fail closed on an older daemon that cannot parse the probe op — refusing + // here degrades to the pre-attach behavior (the reload recovers at turn end). + if (kind === 'attach' && !daemon.supports(RD_WEBCHAT_ATTACH_V1)) { + this.send({ type: 'attached', ack: { accepted: false, agentId, reason: 'unsupported' } }) + return + } const rdMsg: RdMsgWebchat = { source: 'webchat', agentId, @@ -410,9 +432,11 @@ export class RelayBrowserConnection implements ChatSink { ...(ack.turnId ? { turnId: ack.turnId } : {}), agentId, ...(ack.reason ? { reason: ack.reason } : {}), - ...(ack.detail ? { detail: ack.detail } : {}) + ...(ack.detail ? { detail: ack.detail } : {}), + ...(ack.generation !== undefined ? { generation: ack.generation } : {}) } - // A refusal is the whole story of a stream that "never came back" — name it, and who refused. + // A refusal is the whole story of a stream that "never came back" — name it, and who + // refused. Attach misses are the normal idle answer, not worth a line. if (!ack.accepted && (op.op === 'turn' || op.op === 'resume')) { this.deps.log.info( `webchat: ${op.op} ${op.turnId ?? '?'} for ${agentId} in ${this.deps.chatId} refused by ${daemonId}: ${ack.reason ?? 'unspecified'}` @@ -422,12 +446,16 @@ export class RelayBrowserConnection implements ChatSink { this.send({ type: 'ack', ack: browserAck }) } else if (kind === 'resume') { this.send({ type: 'resumed', ack: browserAck }) + } else if (kind === 'attach') { + this.send({ type: 'attached', ack: browserAck }) } } catch (error) { // Lower layers may include the outbound frame in an error. Do not let the opaque // remote-MCP entitlement become log content. this.deps.log.warn(`relay: webchat op delivery failed ${deliveryFailureDiagnostic(error)}`) - if (this.byAgentId.size > 1 && kind === 'turn') { + if (kind === 'attach') { + this.send({ type: 'attached', ack: { accepted: false, agentId, reason: 'no_agent' } }) + } else if (this.byAgentId.size > 1 && kind === 'turn') { this.send({ type: 'ack', ack: { diff --git a/packages/web/src/components/console/PlaygroundProvider.tsx b/packages/web/src/components/console/PlaygroundProvider.tsx index 3207b3a61..1bcd30e24 100644 --- a/packages/web/src/components/console/PlaygroundProvider.tsx +++ b/packages/web/src/components/console/PlaygroundProvider.tsx @@ -111,6 +111,11 @@ interface PlaygroundData { * agent whose runtime has the skill instead of waking the roster to decline. */ commandPick?: { agentId: string; name: string } ) => boolean + /** Reattach a webchat session after a cold page load: probe the conversation's + * daemons for a turn still streaming (the reload wiped the busy flag, lanes, + * and streamed reply) and, on a hit, recreate the lane, restore the typing + * indicator, and replay the reply from the start. No-op while already busy. */ + pgAttach: (id: string, agentId: string, conversationId: string) => void /** Mark `id` (a CP session id) as a session-targeted continuation: the socket * mints through the session-target token route and the daemon dispatches * turns onto that session's own platform coordinates @@ -803,7 +808,7 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { * existing conversation (adopted webchat sessions); omit it for a fresh playground * turn (the CP mints the id). */ const connect = useCallback( - (id: string, agentId: string, conversationId?: string, resumeStream = false): Conn => { + (id: string, agentId: string, conversationId?: string, resumeStream = false, probeOnReady = false): Conn => { const resumeId = conversationId ?? conversationIds.current.get(id) if (resumeId) conversationIds.current.set(id, resumeId) const existing = conns.current.get(id) @@ -975,7 +980,14 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { participants?: WebchatParticipant[] output?: WebchatOutput done?: WebchatDone - ack?: { accepted?: boolean; reason?: string; detail?: string; turnId?: string; agentId?: string } + ack?: { + accepted?: boolean + reason?: string + detail?: string + turnId?: string + agentId?: string + generation?: number + } post?: WebchatPost initiator?: string } @@ -1016,6 +1028,12 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { } if (resumeStream && busyRef.current[id]) { sendResume(ws) + } else if (probeOnReady && !busyRef.current[id] && lanesOf(id).length === 0) { + // Cold-load discovery: ask each verified participant's daemon + // whether a turn is still streaming here (see the 'attached' + // handler below). Idle daemons answer with a quiet refusal. + const probeIds = m.participants?.length ? m.participants.map((p) => p.agentId) : [agentId] + for (const probeId of probeIds) ws.send(JSON.stringify({ type: 'attach', agentId: probeId })) } } else if (m.type === 'output') { if (m.output) receiveOutput(id, m.output) @@ -1067,6 +1085,25 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { } const cursor = key ? streamCursors.current.get(key) : undefined if (cursor && m.ack?.turnId) bindWebchatTurn(cursor, m.ack.turnId) + } else if (m.type === 'attached') { + // Cold-load probe verdict: a hit names the in-flight turn — recreate + // its lane, seed the cursor's generation from the daemon's (so our + // resume outruns pre-reload generations), restore busy, and pull the + // stream from the start through the ordinary resume path. A miss is + // the normal idle answer and stays silent. + const a = m.ack + if (a?.accepted === true && typeof a.turnId === 'string' && typeof a.agentId === 'string') { + const key = laneKey(id, a.agentId) + if (!streamCursors.current.has(key) && !finishedFor(id, a.turnId)?.has(a.agentId)) { + const cursor = createWebchatCursor(a.turnId) + bindWebchatTurn(cursor, a.turnId) + if (typeof a.generation === 'number') cursor.resumeGeneration = a.generation + streamCursors.current.set(key, cursor) + syncBusyLanes(id) + setBusy(id, true) + sendLaneResume(ws, key) + } + } } else if (m.type === 'resumed' && m.ack?.accepted !== false) { const key = cursorKeyFor(id, m.ack?.agentId) const cursor = key ? streamCursors.current.get(key) : undefined @@ -1112,7 +1149,10 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { } else if (m.type === 'ack' && m.ack?.accepted === false) { rejectLane(m.ack.agentId, m.ack.turnId, m.ack.reason, m.ack.detail) } else if (m.type === 'error') { - failStream(id, 'Connection error.') + // An error frame with nothing in flight (e.g. an older relay answering + // the attach probe with 'unrecognized frame') must not push a warning + // step into an idle transcript. + if (busyRef.current[id]) failStream(id, 'Connection error.') } } if (conns.current.get(id) === conn) conn.ws = ws @@ -1536,6 +1576,16 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { sessionTargets.current.add(id) }, []) + /** See PlaygroundData.pgAttach. Doubles as socket warming: a reused live conn + * skips the probe (its ready already passed — nothing was lost in a reload). */ + const pgAttach = useCallback( + (id: string, agentId: string, conversationId: string): void => { + if (!conversationId || busyRef.current[id]) return + connect(id, agentId, conversationId, false, true).ready.catch(() => {}) + }, + [connect] + ) + const value = useMemo( () => ({ getPgInput, @@ -1549,6 +1599,7 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { openPlayground, pgAddAgent, pgSend, + pgAttach, markSessionTarget, getPgQueue, pgCancelQueued, @@ -1575,6 +1626,7 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { openPlayground, pgAddAgent, pgSend, + pgAttach, markSessionTarget, getPgQueue, pgCancelQueued, diff --git a/packages/web/src/components/console/views/SessionDetailView.tsx b/packages/web/src/components/console/views/SessionDetailView.tsx index 53b3734ef..23ee2a018 100644 --- a/packages/web/src/components/console/views/SessionDetailView.tsx +++ b/packages/web/src/components/console/views/SessionDetailView.tsx @@ -1537,6 +1537,7 @@ export default function SessionDetailView() { isPgBusy, setPgImage, pgSend, + pgAttach, markSessionTarget, getPgQueue, pgCancelQueued, @@ -2324,6 +2325,15 @@ export default function SessionDetailView() { void refreshTail() }, [visibleTailReady, sessionBusy, sessionActivityVersion, sessionStreamGeneration, refreshTail]) + // A reload mid-turn wipes the provider's live state (busy flag, stream lanes, + // streamed reply). Probe the conversation's daemons for a still-streaming turn + // and reattach, restoring the typing indicator and the live reply stream. + useEffect(() => { + if (!sid || !session || session.platform !== 'webchat') return + if (!session.channelId || !session.agentId) return + pgAttach(sid, session.agentId, session.channelId) + }, [sid, session?.platform, session?.channelId, session?.agentId, pgAttach]) + // Everything above only APPENDS rows; nothing ever moved the viewport, so a // live session's newest output landed below the fold. Follow it — but only for // a reader who is already at the bottom (see lib/stick-to-bottom). diff --git a/packages/web/src/components/console/views/SessionDetailView.viewer.test.tsx b/packages/web/src/components/console/views/SessionDetailView.viewer.test.tsx index 6f9aa87dc..163690122 100644 --- a/packages/web/src/components/console/views/SessionDetailView.viewer.test.tsx +++ b/packages/web/src/components/console/views/SessionDetailView.viewer.test.tsx @@ -247,6 +247,7 @@ vi.mock('@/components/console/PlaygroundProvider', () => ({ setPgImage: () => {}, openPlayground: () => 'pg_new', pgSend: () => {}, + pgAttach: () => {}, getPgQueue: () => [], pgCancelQueued: () => {}, pgAddAgent: () => {}, From 061125355e267499273349b1a40ac49a5a930a57 Mon Sep 17 00:00:00 2001 From: Dazhan Date: Thu, 27 Aug 2026 15:19:59 +0800 Subject: [PATCH 2/3] fix(web): anchor a cold-attached webchat replay to its reply post MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cold `attach` replays the live turn's agent output with no local `kind:'msg'` prompt step and no `postId`, so `reconcilePersistedLiveSteps` finds neither of its anchors: the replayed reply survived the transcript-tail refresh and the session rendered the completed answer twice until another reload. Take the canonical reply `postId` off the turn's own `rd/webchat-post` frame (routed by conversation, so a reloaded socket receives it) and stamp it onto the replayed steps — the same anchor #753 gave agent-initiated posts. Whichever of that frame and the lane's `done` lands second does the stamping: the success path posts before `done`, the failure path sends `done` first. An agent-initiated post is a different turn's reply and never anchors the attached one. The existing exact-`postId` arm then retires the replay when the persisted row arrives. Also add the new `webchat-attach-v1` capability to the `rd/hello` expectation in relay-client.test.ts, which asserts the advertised list exactly. Co-Authored-By: Claude Opus 5 --- docs/designs/webchat-multi-agents.md | 8 +- packages/daemon/test/cp/relay-client.test.ts | 2 + .../components/console/PlaygroundProvider.tsx | 102 ++++++++++++---- .../console/PlaygroundSend.test.tsx | 109 ++++++++++++++++++ 4 files changed, 195 insertions(+), 26 deletions(-) diff --git a/docs/designs/webchat-multi-agents.md b/docs/designs/webchat-multi-agents.md index 9425c3190..2dcd04fd9 100644 --- a/docs/designs/webchat-multi-agents.md +++ b/docs/designs/webchat-multi-agents.md @@ -575,7 +575,13 @@ Two scope rules: 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; + 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; - 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. diff --git a/packages/daemon/test/cp/relay-client.test.ts b/packages/daemon/test/cp/relay-client.test.ts index 4b9f6162a..6a1227d87 100644 --- a/packages/daemon/test/cp/relay-client.test.ts +++ b/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 ] diff --git a/packages/web/src/components/console/PlaygroundProvider.tsx b/packages/web/src/components/console/PlaygroundProvider.tsx index 1bcd30e24..493b12aa4 100644 --- a/packages/web/src/components/console/PlaygroundProvider.tsx +++ b/packages/web/src/components/console/PlaygroundProvider.tsx @@ -342,6 +342,13 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { // never receive another terminal frame and wedge the busy state. Reset on // each send (one in-flight turn per session). const finishedTurnLanes = useRef }>>(new Map()) + // Cold-attached lanes (the `attach` probe reattached a reload to a live turn), + // keyed by lane. Such a replay has no local prompt step, so the turn-shaped arm of + // `reconcilePersistedLiveSteps` cannot retire it and the reply would render twice + // once the transcript tail persists it. The reply post frame carries the canonical + // postId — stamp it onto the replayed steps so the exact-postId arm retires them + // (the same anchor #753 gave agent-initiated posts). + const coldAttached = useRef>(new Map()) // Participant display names per session id, mirrored in a ref: the socket's // message handlers are closures captured when the socket opened — often the // same tick openPlayground staged the session — so state-based lookups there @@ -699,7 +706,10 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { // Preserve text already received before surfacing a terminal connection // error; otherwise the final sub-frame would disappear from the transcript. deltaBuffer.flushSession(id) - for (const key of lanesOf(id)) streamCursors.current.delete(key) + for (const key of lanesOf(id)) { + streamCursors.current.delete(key) + coldAttached.current.delete(key) + } syncBusyLanes(id) } @@ -713,6 +723,27 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { [pushStep, setBusy] ) + /** Retire a cold-attached turn's replayed steps by stamping the reply's canonical + * postId on them (see `coldAttached`), once both the postId and the lane's `done` + * are in — either order: the failure path sends `done` before the reply post. */ + const anchorColdTurn = useCallback( + (id: string, cursorKey: string): void => { + const cold = coldAttached.current.get(cursorKey) + const postId = cold?.postId + if (!cold || !postId || !cold.done) return + coldAttached.current.delete(cursorKey) + const agentId = laneAgentId(cursorKey) + mutateSteps(id, (steps) => + steps.map((step) => + step.turnId === cold.turnId && (step.agentId ?? undefined) === agentId && !step.postId + ? { ...step, postId } + : step + ) + ) + }, + [mutateSteps] + ) + const applyStreamResult = useCallback( (id: string, cursorKey: string, result: OrderedWebchatResult): void => { const agentId = laneAgentId(cursorKey) @@ -739,6 +770,13 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { // `done` is a hard fence and must not clear busy state before the last // buffered reply text has committed. deltaBuffer.flush(cursorKey) + // Every replayed step of a cold-attached turn is now committed — anchor it. Only + // after the flush: the final chunk becomes a step here. + const cold = coldAttached.current.get(cursorKey) + if (cold) { + cold.done = true + anchorColdTurn(id, cursorKey) + } reconnectAttempts.current.delete(id) streamCursors.current.delete(cursorKey) syncBusyLanes(id) @@ -760,7 +798,7 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { // The turn stays busy until every targeted participant's lane finished. if (lanesOf(id).length === 0) setBusy(id, false) }, - [applyEvent, applyStatus, applyTitle, deltaBuffer, failStream, participantName, pushStep, setBusy] + [anchorColdTurn, applyEvent, applyStatus, applyTitle, deltaBuffer, failStream, participantName, pushStep, setBusy] ) const receiveOutput = useCallback( @@ -1039,33 +1077,44 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { if (m.output) receiveOutput(id, m.output) } else if (m.type === 'done') { if (m.done) receiveDone(id, m.done) - } else if (m.type === 'post' && m.initiator === 'agent' && m.post?.author.kind === 'agent') { - // Agent-initiated turn (another participant's sendMessage/lineage-reply - // wake, #753): it never streamed output/done to this socket, so the - // completed post IS its first and only rendering here. + } else if (m.type === 'post' && m.post?.author.kind === 'agent') { const agentId = m.post.author.agentId const post = m.post + // This lane's own reply post while cold-attached: keep the canonical + // postId as that turn's retirement anchor. An agent-initiated post is + // a DIFFERENT turn's reply and must not anchor the attached one. + const coldKey = laneKey(id, agentId) + const cold = m.initiator === 'agent' ? undefined : coldAttached.current.get(coldKey) + if (cold) { + cold.postId = post.postId + anchorColdTurn(id, coldKey) + } + // Agent-initiated turn (another participant's sendMessage/lineage-reply + // wake, #753): it never streamed output/done to this socket, so the + // completed post IS its first and only rendering here. A human-initiated + // turn already streamed it and only needs the anchor above. // Keyed by postId so a daemon re-broadcast (inbox replay, relay fan-out // echo) upserts instead of duplicating the step. - mutateSteps(id, (steps) => - steps.some((step) => step.postId === post.postId) - ? steps - : [ - ...steps, - stampStep({ - kind: 'done', - turnId: post.postId, - // The daemon persists this reply before the post frame ever arrives, so - // `postId` is what lets `reconcilePersistedLiveSteps` drop this step once - // the canonical row lands in a later transcript refresh (#753) — text/time - // matching (the prompt-turn heuristic) has nothing to anchor on here. - postId: post.postId, - agentId, - ...(participantName(id, agentId) ? { who: participantName(id, agentId) } : {}), - text: post.text - }) - ] - ) + if (m.initiator === 'agent') + mutateSteps(id, (steps) => + steps.some((step) => step.postId === post.postId) + ? steps + : [ + ...steps, + stampStep({ + kind: 'done', + turnId: post.postId, + // The daemon persists this reply before the post frame ever arrives, so + // `postId` is what lets `reconcilePersistedLiveSteps` drop this step once + // the canonical row lands in a later transcript refresh (#753) — text/time + // matching (the prompt-turn heuristic) has nothing to anchor on here. + postId: post.postId, + agentId, + ...(participantName(id, agentId) ? { who: participantName(id, agentId) } : {}), + text: post.text + }) + ] + ) } else if (m.type === 'ack' && m.ack?.accepted !== false) { let key = cursorKeyFor(id, m.ack?.agentId) // The relay may target participants the client did not lane (a @@ -1099,6 +1148,9 @@ export function PlaygroundProvider({ children }: { children: ReactNode }) { bindWebchatTurn(cursor, a.turnId) if (typeof a.generation === 'number') cursor.resumeGeneration = a.generation streamCursors.current.set(key, cursor) + // No local prompt step to reconcile against — the reply post frame + // will name this turn's retirement anchor (see `coldAttached`). + coldAttached.current.set(key, { turnId: a.turnId }) syncBusyLanes(id) setBusy(id, true) sendLaneResume(ws, key) diff --git a/packages/web/src/components/console/PlaygroundSend.test.tsx b/packages/web/src/components/console/PlaygroundSend.test.tsx index 5c7c94e60..08f8174df 100644 --- a/packages/web/src/components/console/PlaygroundSend.test.tsx +++ b/packages/web/src/components/console/PlaygroundSend.test.tsx @@ -53,6 +53,7 @@ let openPlayground: ReturnType['openPlayground'] let getPgQueue: ReturnType['getPgQueue'] let pgCancelQueued: ReturnType['pgCancelQueued'] let getLiveSteps: ReturnType['getLiveSteps'] +let pgAttach: ReturnType['pgAttach'] function Probe() { const pg = usePlayground() @@ -61,6 +62,7 @@ function Probe() { getPgQueue = pg.getPgQueue pgCancelQueued = pg.pgCancelQueued getLiveSteps = pg.getLiveSteps + pgAttach = pg.pgAttach return null } @@ -550,3 +552,110 @@ describe('reconnect after an unacked turn', () => { expect(sent.at(-1)).toMatchObject({ type: 'resume', turnId: turn.turnId }) }) }) + +// A cold attach replays a live turn with NO local prompt step, so the turn-shaped arm +// of `reconcilePersistedLiveSteps` cannot retire it. The reply post's canonical postId +// is the anchor that keeps the transcript tail from rendering the answer twice. +describe('cold-attach retirement anchor', () => { + class AttachSocket extends StubSocket { + static instances: AttachSocket[] = [] + onopen?: () => void + onmessage?: (e: { data: string }) => void + onerror?: (e: unknown) => void + onclose?: () => void + constructor() { + super() + AttachSocket.instances.push(this) + } + } + + async function coldAttach() { + AttachSocket.instances = [] + Reflect.set(globalThis, 'WebSocket', AttachSocket) + const api = await import('@/lib/api') + vi.mocked(api.webchatWsUrl).mockResolvedValue('wss://relay.test/ws') + await act(async () => { + pgAttach('s-cold', 'agent-1', 'c-cold') + }) + const socket = AttachSocket.instances[0]! + await act(async () => { + socket.readyState = 1 + socket.onopen?.() + socket.onmessage?.({ + data: JSON.stringify({ type: 'ready', conversationId: 'c-cold', participants: [{ agentId: 'agent-1' }] }) + }) + }) + return socket + } + + it('probes on ready and stamps the reply postId on the replayed steps', async () => { + const socket = await coldAttach() + expect(socket.send.mock.calls.map((c) => JSON.parse(String(c[0])))).toContainEqual({ + type: 'attach', + agentId: 'agent-1' + }) + await act(async () => { + socket.onmessage?.({ + data: JSON.stringify({ + type: 'attached', + ack: { accepted: true, turnId: 'turn-cold', agentId: 'agent-1', generation: 4 } + }) + }) + socket.onmessage?.({ + data: JSON.stringify({ + type: 'output', + output: { + turnId: 'turn-cold', + agentId: 'agent-1', + index: 0, + event: { kind: 'tool_call', toolCallId: 't1', title: 'Read file', status: 'completed' } + } + }) + }) + }) + expect(getLiveSteps('s-cold').filter((s) => s.turnId === 'turn-cold')).toHaveLength(1) + // Human-initiated: no `initiator`, so the frame only carries the anchor. + await act(async () => { + socket.onmessage?.({ + data: JSON.stringify({ + type: 'post', + post: { postId: 'post-cold', author: { kind: 'agent', agentId: 'agent-1' }, text: 'done' } + }) + }) + socket.onmessage?.({ data: JSON.stringify({ type: 'done', done: { turnId: 'turn-cold', agentId: 'agent-1' } }) }) + }) + const replayed = getLiveSteps('s-cold').filter((s) => s.turnId === 'turn-cold') + expect(replayed.length).toBeGreaterThan(0) + expect(replayed.every((s) => s.postId === 'post-cold')).toBe(true) + }) + + it('anchors when the reply post arrives after done (the failure path order)', async () => { + const socket = await coldAttach() + await act(async () => { + socket.onmessage?.({ + data: JSON.stringify({ type: 'attached', ack: { accepted: true, turnId: 'turn-late', agentId: 'agent-1' } }) + }) + socket.onmessage?.({ + data: JSON.stringify({ + type: 'output', + output: { + turnId: 'turn-late', + agentId: 'agent-1', + index: 0, + event: { kind: 'tool_call', toolCallId: 't1', title: 'Read file', status: 'completed' } + } + }) + }) + socket.onmessage?.({ data: JSON.stringify({ type: 'done', done: { turnId: 'turn-late', agentId: 'agent-1' } }) }) + socket.onmessage?.({ + data: JSON.stringify({ + type: 'post', + post: { postId: 'post-late', author: { kind: 'agent', agentId: 'agent-1' }, text: 'partial' } + }) + }) + }) + const replayed = getLiveSteps('s-cold').filter((s) => s.turnId === 'turn-late') + expect(replayed.length).toBeGreaterThan(0) + expect(replayed.every((s) => s.postId === 'post-late')).toBe(true) + }) +}) From c02fc885be4f528fc8abc5dbd434b959c774a487 Mon Sep 17 00:00:00 2001 From: Dazhan Date: Thu, 27 Aug 2026 16:50:24 +0800 Subject: [PATCH 3/3] fix(webchat): fan reply posts across relays --- docs/designs/webchat-multi-agents.md | 6 +++- packages/daemon/src/cp/relay-client.ts | 29 +++------------- packages/daemon/src/daemon.ts | 20 +++-------- packages/daemon/src/webchat/transport.ts | 4 +-- .../test/daemon-webchat-continuation.test.ts | 9 ++--- packages/daemon/test/daemon-webchat.test.ts | 33 ++++++++++++++++--- .../daemon/test/webchat-turn-refresh.test.ts | 26 ++++++++------- 7 files changed, 62 insertions(+), 65 deletions(-) diff --git a/docs/designs/webchat-multi-agents.md b/docs/designs/webchat-multi-agents.md index 2dcd04fd9..987f9c8f8 100644 --- a/docs/designs/webchat-multi-agents.md +++ b/docs/designs/webchat-multi-agents.md @@ -581,7 +581,11 @@ Two scope rules: `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; + 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. diff --git a/packages/daemon/src/cp/relay-client.ts b/packages/daemon/src/cp/relay-client.ts index 1eb767411..f4b118840 100644 --- a/packages/daemon/src/cp/relay-client.ts +++ b/packages/daemon/src/cp/relay-client.ts @@ -72,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 + /** 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 /** * 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 @@ -247,8 +232,7 @@ export class RelayClient { private async handleMsg(reqId: string, msg: RdMsg): Promise { 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 }))) } @@ -264,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))) } diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index 712fd0a96..f7385a63f 100644 --- a/packages/daemon/src/daemon.ts +++ b/packages/daemon/src/daemon.ts @@ -403,7 +403,6 @@ import type { SessionKey, SessionActivity, Ack, - RdWebchatPost, RdMsg, RdMsgHook, RdMsgWebchat, @@ -5896,11 +5895,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 { + private handleRelayMsg(msg: RdMsg, chat: (event: RdChatEvent) => void): RdAck | Promise { const dedupKey = `${msg.source === 'im' ? `${msg.botId}:` : ''}${msg.sessionKey}:${msg.msgId}` const prior = this.relayMsgAcks.get(dedupKey) if (prior) { @@ -5919,7 +5914,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, @@ -5933,7 +5928,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) @@ -6817,11 +6812,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 { + private async dispatchRelayOp(msg: RdMsgWebchat, chat: (event: RdChatEvent) => void): Promise { const sink: WebchatSink = { output: (o) => chat({ kind: 'output', output: o }), done: (d) => chat({ kind: 'done', done: d }) @@ -6879,7 +6870,6 @@ export class Daemon { msg.remoteMcp, op.mentions, op.post, - post, op.worktree ) return { @@ -15575,7 +15565,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) }) diff --git a/packages/daemon/src/webchat/transport.ts b/packages/daemon/src/webchat/transport.ts index 8fa37419d..cb6588246 100644 --- a/packages/daemon/src/webchat/transport.ts +++ b/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 { 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 diff --git a/packages/daemon/test/daemon-webchat-continuation.test.ts b/packages/daemon/test/daemon-webchat-continuation.test.ts index f0f20ccbe..fabe2043c 100644 --- a/packages/daemon/test/daemon-webchat-continuation.test.ts +++ b/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() diff --git a/packages/daemon/test/daemon-webchat.test.ts b/packages/daemon/test/daemon-webchat.test.ts index 1a8633b94..b88f9fcaf 100644 --- a/packages/daemon/test/daemon-webchat.test.ts +++ b/packages/daemon/test/daemon-webchat.test.ts @@ -1513,11 +1513,11 @@ describe('Daemon handleRelayMsg (rd/msg op dispatch — the relay data plane)', const turnId = '77777777-7777-4777-8777-777777777777' const events: RdChatEvent[] = [] - const posts: unknown[] = [] + const sendWebchatPost = vi.fn() + ;(daemon as any).relays = { stop: vi.fn(async () => {}), sendWebchatPost } const ack = await (daemon as any).handleRelayMsg( rd({ op: 'turn', text: 'anyone?', user: 'owner', turnId, post: { postId: turnId, at: 1_000 } }), - (event: RdChatEvent) => events.push(event), - (post: unknown) => posts.push(post) + (event: RdChatEvent) => events.push(event) ) expect(ack).toMatchObject({ accepted: true }) await vi.waitFor(() => expect(events.some((e) => e.kind === 'done')).toBe(true), WAIT) @@ -1526,7 +1526,7 @@ describe('Daemon handleRelayMsg (rd/msg op dispatch — the relay data plane)', e.kind === 'output' && e.output.event?.kind === 'message' ? [e.output.event.text] : [] ) expect(messages).toEqual([]) // the sentinel was held and dropped - expect(posts).toEqual([]) // no canonical post fan-out + expect(sendWebchatPost).not.toHaveBeenCalled() // no canonical post fan-out const replies = (await (daemon as any).store.transcriptSince(`${CONV}`, `webchat:${CONV}`, null)).filter( (row: { sender: string }) => row.sender === AGENT_ID ) @@ -2049,6 +2049,31 @@ describe('Daemon handleRelayMsg (rd/msg op dispatch — the relay data plane)', await daemon.stop() }, 15_000) + // Reply posts use daemon-wide relay fan-out so a cold attach through another relay can reconcile. + it('fans a browser turn reply post out daemon-wide, not down the admitting relay socket', async () => { + const { factory } = streamingHost([text('the answer')]) + const daemon = new Daemon({ root: scaffold(), hostFactory: factory }) + await daemon.start() + ;(daemon as any).cpClient = fakeCpClient() + const sendWebchatPost = vi.fn() + ;(daemon as any).relays = { stop: vi.fn(async () => {}), sendWebchatPost } + + const turnId = '77777777-7777-4777-8777-777777777777' + const events: RdChatEvent[] = [] + const ack = await (daemon as any).handleRelayMsg( + rd({ op: 'turn', text: 'ask', user: 'owner', turnId, post: { postId: turnId, at: 1_000 } }), + (event: RdChatEvent) => events.push(event) + ) + expect(ack).toMatchObject({ accepted: true }) + await vi.waitFor(() => expect(sendWebchatPost).toHaveBeenCalledTimes(1), WAIT) + expect(sendWebchatPost.mock.calls[0]![0]).toMatchObject({ + conversationId: CONV, + agentId: AGENT_ID, + post: { conversationId: CONV, text: 'the answer', author: { kind: 'agent', agentId: AGENT_ID } } + }) + await daemon.stop() + }, 15_000) + it('rejects a turn while draining (accepted:false, reason draining) — no turn dispatched', async () => { const { factory } = streamingHost([]) const daemon = new Daemon({ root: scaffold(), hostFactory: factory }) diff --git a/packages/daemon/test/webchat-turn-refresh.test.ts b/packages/daemon/test/webchat-turn-refresh.test.ts index 46a17b334..eb2987b46 100644 --- a/packages/daemon/test/webchat-turn-refresh.test.ts +++ b/packages/daemon/test/webchat-turn-refresh.test.ts @@ -107,10 +107,11 @@ describe('webchat turn-final context refresh', () => { const events: RdChatEvent[] = [] const posts: RdWebchatPost[] = [] + // Completed posts use daemon-wide relay fan-out. + ;(daemon as any).relays = { stop: vi.fn(async () => {}), sendWebchatPost: (p: RdWebchatPost) => posts.push(p) } const ack = await (daemon as any).handleRelayMsg( rd({ op: 'turn', text: 'original request', user: 'owner', turnId: TURN, post: { postId: TURN, at: 1_000 } }), - (event: RdChatEvent) => events.push(event), - (post: RdWebchatPost) => posts.push(post) + (event: RdChatEvent) => events.push(event) ) expect(ack).toMatchObject({ accepted: true, turnId: TURN }) await vi.waitFor(() => expect(host.prompt).toHaveBeenCalledTimes(1), WAIT) @@ -200,10 +201,11 @@ describe('webchat turn-final context refresh', () => { const first: RdChatEvent[] = [] const second: RdChatEvent[] = [] const posts: RdWebchatPost[] = [] + // Completed posts use daemon-wide relay fan-out. + ;(daemon as any).relays = { stop: vi.fn(async () => {}), sendWebchatPost: (p: RdWebchatPost) => posts.push(p) } await (daemon as any).handleRelayMsg( rd({ op: 'turn', text: 'original request', user: 'owner', turnId: TURN, post: { postId: TURN, at: 1_000 } }), - (event: RdChatEvent) => first.push(event), - (post: RdWebchatPost) => posts.push(post) + (event: RdChatEvent) => first.push(event) ) await vi.waitFor(() => expect(host.prompt).toHaveBeenCalledTimes(1), WAIT) @@ -221,8 +223,7 @@ describe('webchat turn-final context refresh', () => { }, { msgId: 'm-2' } ), - (event: RdChatEvent) => second.push(event), - (post: RdWebchatPost) => posts.push(post) + (event: RdChatEvent) => second.push(event) ) expect(ack2).toMatchObject({ accepted: true, turnId: secondTurn }) releaseFirst() @@ -278,10 +279,11 @@ describe('webchat turn-final context refresh', () => { const first: RdChatEvent[] = [] const second: RdChatEvent[] = [] const posts: RdWebchatPost[] = [] + // Completed posts use daemon-wide relay fan-out. + ;(daemon as any).relays = { stop: vi.fn(async () => {}), sendWebchatPost: (p: RdWebchatPost) => posts.push(p) } await (daemon as any).handleRelayMsg( rd({ op: 'turn', text: 'original request', user: 'owner', turnId: TURN, post: { postId: TURN, at: 1_000 } }), - (event: RdChatEvent) => first.push(event), - (post: RdWebchatPost) => posts.push(post) + (event: RdChatEvent) => first.push(event) ) await vi.waitFor(() => expect(host.prompt).toHaveBeenCalledTimes(1), WAIT) @@ -309,8 +311,7 @@ describe('webchat turn-final context refresh', () => { }, { msgId: 'm-2' } ), - (event: RdChatEvent) => second.push(event), - (post: RdWebchatPost) => posts.push(post) + (event: RdChatEvent) => second.push(event) ) releaseFirst() @@ -368,10 +369,11 @@ describe('webchat turn-final context refresh', () => { const events: RdChatEvent[] = [] const posts: RdWebchatPost[] = [] + // Completed posts use daemon-wide relay fan-out. + ;(daemon as any).relays = { stop: vi.fn(async () => {}), sendWebchatPost: (p: RdWebchatPost) => posts.push(p) } await (daemon as any).handleRelayMsg( rd({ op: 'turn', text: 'original request', user: 'owner', turnId: TURN, post: { postId: TURN, at: 1_000 } }), - (event: RdChatEvent) => events.push(event), - (post: RdWebchatPost) => posts.push(post) + (event: RdChatEvent) => events.push(event) ) await vi.waitFor(() => expect(events.some((e) => e.kind === 'done')).toBe(true), WAIT)