From 738e15a09b27bb6f7dd3dc8d29238ade9c8795b4 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sun, 23 Aug 2026 10:24:29 +0200 Subject: [PATCH] =?UTF-8?q?fix(web):=20le=20client=20WS=20r=C3=A9attache?= =?UTF-8?q?=20les=20sessions=20chat=20structur=C3=A9es=20apr=C3=A8s=20reco?= =?UTF-8?q?nnexion=20=E2=80=94=20#241?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit needsReconnect() couvre désormais les chatSinks et le cycle de reconnexion renvoie un chat.attach par session active, un échec isolé n'empêchant pas les autres abonnements de reprendre. Tests ciblés ajoutés. --- .../adapters/http/wsLiveClient.chat.test.ts | 117 +++++++++++++++++- frontend/src/adapters/http/wsLiveClient.ts | 17 ++- 2 files changed, 131 insertions(+), 3 deletions(-) diff --git a/frontend/src/adapters/http/wsLiveClient.chat.test.ts b/frontend/src/adapters/http/wsLiveClient.chat.test.ts index 868c7a2a..a26dfbbc 100644 --- a/frontend/src/adapters/http/wsLiveClient.chat.test.ts +++ b/frontend/src/adapters/http/wsLiveClient.chat.test.ts @@ -1,7 +1,8 @@ -import { describe, it, expect } from "vitest"; +import { describe, it, expect, vi } from "vitest"; import type { WebSocketLike } from "./wsLiveClient"; import { WsLiveClient } from "./wsLiveClient"; +import { bytesToBase64 } from "./frames"; class FakeSocket implements WebSocketLike { sent: string[] = []; @@ -33,6 +34,32 @@ function connectedClient(): { client: WsLiveClient; socket: FakeSocket } { return { client, socket }; } +function attachedAck(id: string, sessionId: string) { + return { + kind: "terminal.attached", + replyTo: id, + payload: { + session: { sessionId, rows: 30, cols: 120 }, + scrollback: [], + nextSeq: 0, + status: "running", + gap: false, + }, + }; +} + +async function attachTerminal( + client: WsLiveClient, + sockets: FakeSocket[], + sessionId: string, +): Promise { + const attached = client.attachTerminal({ sessionId, onData: () => {} }); + await vi.waitFor(() => expect(sockets[0]?.sent.length ?? 0).toBeGreaterThan(0)); + const sent = JSON.parse(sockets[0].sent[0]); + sockets[0].receive(attachedAck(sent.id, sessionId)); + await attached; +} + describe("WsLiveClient structured chat routing", () => { it("routes chat.output to the per-session chat sink", async () => { const { client, socket } = connectedClient(); @@ -80,4 +107,92 @@ describe("WsLiveClient structured chat routing", () => { }); expect(received).toEqual([]); }); + + it("re-attaches chat sinks after reconnect while PTY logs keep the socket alive", async () => { + const sockets: FakeSocket[] = []; + let timerFn: (() => void) | null = null; + const client = new WsLiveClient({ + wsUrl: "wss://host", + reconnectBaseMs: 1, + socketFactory: () => { + const socket = new FakeSocket(); + sockets.push(socket); + queueMicrotask(() => socket.onopen?.()); + return socket; + }, + setTimeoutImpl: (fn) => { + timerFn = fn; + return 1; + }, + clearTimeoutImpl: () => { + timerFn = null; + }, + }); + + await attachTerminal(client, sockets, "pty-1"); + const received: unknown[] = []; + client.setChatSink("chat-1", (chunk) => received.push(chunk)); + + sockets[0].close(); + expect(client.getConnectionState()).toBe("reconnecting"); + expect(client.needsReconnect()).toBe(true); + + const fire = timerFn; + timerFn = null; + fire?.(); + await vi.waitFor(() => expect(sockets.length).toBe(2)); + await vi.waitFor(() => expect(sockets[1].sent.length).toBeGreaterThan(0)); + + const terminalAttach = JSON.parse(sockets[1].sent[0]); + expect(terminalAttach.kind).toBe("terminal.attach"); + sockets[1].receive(attachedAck(terminalAttach.id, "pty-1")); + + await vi.waitFor(() => expect(sockets[1].sent.length).toBeGreaterThan(1)); + const chatAttach = JSON.parse(sockets[1].sent[1]); + expect(chatAttach.kind).toBe("chat.attach"); + expect(chatAttach.payload).toEqual({ sessionId: "chat-1" }); + sockets[1].receive({ + kind: "chat.attached", + replyTo: chatAttach.id, + payload: { sessionId: "chat-1", cellKind: "chat" }, + }); + + sockets[1].receive({ + kind: "terminal.output", + payload: { + sessionId: "pty-1", + seq: 1, + bytesBase64: bytesToBase64(new Uint8Array([108, 111, 103])), + }, + }); + sockets[1].receive({ + kind: "chat.output", + payload: { + sessionId: "chat-1", + chunk: { + kind: "progress", + progress: { + source: "ideaLocal", + kind: "mcp", + stage: "started", + label: "idea_ticket_read", + text: "reading ticket #241", + }, + }, + }, + }); + + expect(received).toEqual([ + { + kind: "progress", + progress: { + source: "ideaLocal", + kind: "mcp", + stage: "started", + label: "idea_ticket_read", + text: "reading ticket #241", + }, + }, + ]); + }); }); diff --git a/frontend/src/adapters/http/wsLiveClient.ts b/frontend/src/adapters/http/wsLiveClient.ts index b2194037..961f26b2 100644 --- a/frontend/src/adapters/http/wsLiveClient.ts +++ b/frontend/src/adapters/http/wsLiveClient.ts @@ -294,7 +294,8 @@ export class WsLiveClient { } this.setConnState("reconnecting"); // Reconnect while anything still needs the socket: a live terminal to - // re-attach (F3) OR an active domain-event subscription (F5 live surfaces). + // re-attach (F3), an active structured chat subscription, OR an active + // domain-event subscription (F5 live surfaces). // Otherwise stay idle until the next explicit use. if (this.terminals.size > 0) { for (const sub of this.terminals.values()) sub.onData(notice("déconnecté — reconnexion…")); @@ -304,7 +305,11 @@ export class WsLiveClient { /** Whether the socket should auto-reconnect (a terminal or a live subscription). */ needsReconnect(): boolean { - return this.terminals.size > 0 || this.domainEventHandler !== null; + return ( + this.terminals.size > 0 || + this.chatSinks.size > 0 || + this.domainEventHandler !== null + ); } private scheduleReconnect(): void { @@ -354,6 +359,14 @@ export class WsLiveClient { // block the others; leave it tracked for the next reconnect cycle. } } + for (const sessionId of this.chatSinks.keys()) { + try { + await this.send("chat.attach", { sessionId }); + } catch { + // Same policy as PTY re-attach: one stale/temporarily unavailable chat + // session must not prevent the other live subscriptions from resuming. + } + } } // -------------------------------------------------------------------------