diff --git a/frontend/src/adapters/http/frames.ts b/frontend/src/adapters/http/frames.ts index e61cef9..d63e8e2 100644 --- a/frontend/src/adapters/http/frames.ts +++ b/frontend/src/adapters/http/frames.ts @@ -94,7 +94,7 @@ export interface ServerFrame { payload: Record; } -/** Payload of a `terminal.attached` acknowledgement. */ +/** Payload of a `terminal.attached` acknowledgement (B5, `server.rs`). */ export interface AttachedPayload { session: { sessionId: string; @@ -104,8 +104,15 @@ export interface AttachedPayload { cols: number; }; nextSeq: number; + /** + * Bounded scrollback replayed at (re)attach. B5 sends a single entry (`seq:0`) + * carrying all retained bytes, or an empty array when there is nothing to + * replay. + */ scrollback: { seq: number; bytesBase64: string }[]; gap: boolean; + /** Lifecycle status carried on the ack (B5 sends `"running"`). */ + status?: string; /** Conversation id minted by an `agent.launch` (mirrors `assignedConversationId`). */ assignedConversationId?: string; } @@ -117,6 +124,16 @@ export interface OutputPayload { bytesBase64: string; } +/** + * Payload of a `terminal.status` frame (B5). `status` is a lifecycle transition + * (`"exited"` on close); `exitCode` is present for `"exited"` (may be `null`). + */ +export interface StatusPayload { + sessionId: string; + status: string; + exitCode?: number | null; +} + /** Payload of a `chat.output` frame (structured assistant stream). */ export interface ChatOutputPayload { sessionId: string; diff --git a/frontend/src/adapters/http/streamGateways.ts b/frontend/src/adapters/http/streamGateways.ts index 3966980..61ecd13 100644 --- a/frontend/src/adapters/http/streamGateways.ts +++ b/frontend/src/adapters/http/streamGateways.ts @@ -93,12 +93,11 @@ export function makeWsTerminalHandle( detach(): void { // Stop delivering output locally and tell the server the view is gone. The // backend PTY keeps running (detach ≠ close), matching the Tauri handle. - ws.removeOutputSink(sessionId); - void ws.sendFireAndForget("terminal.detach", { sessionId }); + // Untracking also disables reconnection re-attach for this session. + void ws.detachTerminal(sessionId); }, async close(): Promise { - ws.removeOutputSink(sessionId); - await ws.send("terminal.close", { sessionId }); + await ws.closeTerminalSession(sessionId); }, }; } @@ -137,37 +136,33 @@ export class HttpTerminalGateway implements TerminalGateway { options: OpenTerminalOptions, onData: (bytes: Uint8Array) => void, ): Promise { - const ack = await this.ws.send("terminal.open", { - projectId: undefined, // TODO(F3/B5): the port lacks projectId; confirm contract. - nodeId: options.nodeId ?? null, + // B5: `terminal.open` carries no projectId; `cwd` is a server-validated path. + // A fresh open has empty scrollback (the view does not repaint on open). + const res = await this.ws.openTerminal({ cwd: options.cwd, rows: options.rows, cols: options.cols, + nodeId: options.nodeId ?? null, + onData, }); - const payload = ack.payload as unknown as AttachedPayload; - const sessionId = payload.session.sessionId; - this.ws.setOutputSink(sessionId, onData); - const scrollback = attachedToScrollback(payload); - if (scrollback.length > 0) onData(scrollback); - return makeWsTerminalHandle(sessionId, this.ws); + return makeWsTerminalHandle(res.sessionId, this.ws); } async reattach( sessionId: string, onData: (bytes: Uint8Array) => void, ): Promise { - const ack = await this.ws.send("terminal.attach", { sessionId, lastSeq: null }); - const payload = ack.payload as unknown as AttachedPayload; - this.ws.setOutputSink(sessionId, onData); + // The view repaints the returned scrollback itself (TerminalView contract); + // the client does not also push it through `onData` on the initial attach. + const res = await this.ws.attachTerminal({ sessionId, onData }); return { - handle: makeWsTerminalHandle(sessionId, this.ws), - scrollback: attachedToScrollback(payload), + handle: makeWsTerminalHandle(res.sessionId, this.ws), + scrollback: res.scrollback, }; } async closeTerminal(sessionId: string): Promise { - this.ws.removeOutputSink(sessionId); - await this.ws.send("terminal.close", { sessionId }); + await this.ws.closeTerminalSession(sessionId); } } diff --git a/frontend/src/adapters/http/terminalGateway.test.ts b/frontend/src/adapters/http/terminalGateway.test.ts new file mode 100644 index 0000000..c7449bc --- /dev/null +++ b/frontend/src/adapters/http/terminalGateway.test.ts @@ -0,0 +1,144 @@ +/** + * F3 — the WS terminal gateway + handle round-trip against the B5 frame contract: + * open→attached (empty scrollback), attach→attached (scrollback returned for the + * view to repaint, not double-pushed), input/resize emit conforming frames, + * detach ≠ close, and `terminal.output` bytes are base64-decoded into the sink. + */ +import { describe, it, expect, vi } from "vitest"; + +import { HttpTerminalGateway } from "./streamGateways"; +import { WsLiveClient, type WebSocketLike } from "./wsLiveClient"; +import { bytesToBase64 } from "./frames"; + +class FakeSocket implements WebSocketLike { + sent: string[] = []; + onopen: (() => void) | null = null; + onmessage: ((event: { data: string }) => void) | null = null; + onerror: ((event: unknown) => void) | null = null; + onclose: (() => void) | null = null; + send(data: string): void { + this.sent.push(data); + } + close(): void { + this.onclose?.(); + } + receive(frame: unknown): void { + this.onmessage?.({ data: JSON.stringify(frame) }); + } +} + +/** Client whose socket auto-opens on the next microtask. */ +function client(): { ws: WsLiveClient; sockets: FakeSocket[] } { + const sockets: FakeSocket[] = []; + const ws = new WsLiveClient({ + wsUrl: "wss://host", + socketFactory: () => { + const s = new FakeSocket(); + sockets.push(s); + queueMicrotask(() => s.onopen?.()); + return s; + }, + }); + return { ws, sockets }; +} + +/** Awaits the frame the client just sent, then replies with `reply(id)`. */ +async function replyToLast( + socket: FakeSocket, + index: number, + reply: (id: string) => unknown, +): Promise> { + await vi.waitFor(() => expect(socket.sent.length).toBeGreaterThan(index)); + const sent = JSON.parse(socket.sent[index]); + socket.receive(reply(sent.id)); + return sent; +} + +function attachedAck(id: string, sessionId: string, scrollbackBytes?: Uint8Array) { + return { + kind: "terminal.attached", + replyTo: id, + payload: { + session: { sessionId, rows: 30, cols: 120 }, + scrollback: scrollbackBytes + ? [{ seq: 0, bytesBase64: bytesToBase64(scrollbackBytes) }] + : [], + nextSeq: scrollbackBytes ? 1 : 0, + status: "running", + gap: false, + }, + }; +} + +describe("HttpTerminalGateway round-trip (B5 frames)", () => { + it("open → attached with empty scrollback; output is decoded into the sink", async () => { + const { ws, sockets } = client(); + const gw = new HttpTerminalGateway(ws); + const chunks: Uint8Array[] = []; + + const handlePromise = gw.openTerminal({ cwd: "/srv/app", rows: 30, cols: 120 }, (b) => + chunks.push(b), + ); + const sent = await replyToLast(sockets[0], 0, (id) => attachedAck(id, "s1")); + const handle = await handlePromise; + + // Conforming open frame (no projectId; cwd is server-validated). + expect(sent.kind).toBe("terminal.open"); + expect(sent.payload).toMatchObject({ cwd: "/srv/app", rows: 30, cols: 120 }); + expect(handle.sessionId).toBe("s1"); + // No scrollback pushed on a fresh open. + expect(chunks).toHaveLength(0); + + // A terminal.output frame is base64-decoded and written to the sink. + sockets[0].receive({ + kind: "terminal.output", + payload: { sessionId: "s1", seq: 1, bytesBase64: bytesToBase64(new Uint8Array([104, 105])) }, + }); + expect(chunks).toEqual([new Uint8Array([104, 105])]); + }); + + it("attach → attached returns the scrollback for the view (not double-pushed)", async () => { + const { ws, sockets } = client(); + const gw = new HttpTerminalGateway(ws); + const chunks: Uint8Array[] = []; + const scroll = new Uint8Array([65, 66, 67]); + + const resPromise = gw.reattach("s7", (b) => chunks.push(b)); + const sent = await replyToLast(sockets[0], 0, (id) => attachedAck(id, "s7", scroll)); + const res = await resPromise; + + expect(sent.kind).toBe("terminal.attach"); + expect(sent.payload).toMatchObject({ sessionId: "s7" }); + // Scrollback is returned for TerminalView to repaint… + expect(res.scrollback).toEqual(scroll); + // …and NOT also pushed through onData (no double paint on the initial attach). + expect(chunks).toHaveLength(0); + }); + + it("input and resize emit conforming frames; detach ≠ close", async () => { + const { ws, sockets } = client(); + const gw = new HttpTerminalGateway(ws); + + const handlePromise = gw.openTerminal({ cwd: "/srv", rows: 24, cols: 80 }, () => {}); + await replyToLast(sockets[0], 0, (id) => attachedAck(id, "s1")); + const handle = await handlePromise; + const base = sockets[0].sent.length; + + await handle.write(new Uint8Array([13])); + await handle.resize(40, 140); + + const input = JSON.parse(sockets[0].sent[base]); + expect(input.kind).toBe("terminal.input"); + expect(input.payload).toEqual({ sessionId: "s1", bytesBase64: bytesToBase64(new Uint8Array([13])) }); + + const resize = JSON.parse(sockets[0].sent[base + 1]); + expect(resize.kind).toBe("terminal.resize"); + expect(resize.payload).toEqual({ sessionId: "s1", rows: 40, cols: 140 }); + + // detach keeps the PTY alive (terminal.detach); close kills it (terminal.close). + handle.detach(); + await vi.waitFor(() => + expect(JSON.parse(sockets[0].sent[base + 2]).kind).toBe("terminal.detach"), + ); + }); +}); diff --git a/frontend/src/adapters/http/wsLiveClient.ts b/frontend/src/adapters/http/wsLiveClient.ts index 21e9d70..a5e5f07 100644 --- a/frontend/src/adapters/http/wsLiveClient.ts +++ b/frontend/src/adapters/http/wsLiveClient.ts @@ -1,28 +1,45 @@ /** - * WebSocket live client skeleton for the web transport — ticket #13, lot F1. + * WebSocket live client for the web transport — ticket #13, lot F3 (finalised + * from the F1 skeleton). * - * Owns a single WS connection to `{wsUrl}/ws/live` and multiplexes over it, per - * the B0 draft: PTY output, structured chat output and the low-frequency domain - * event stream (`event.domain`, replacing Tauri `listen("domain://event")`). + * Owns a single WS connection to `{wsUrl}/ws/live` (authenticated by the session + * cookie at the upgrade — same-origin, no URL secret) and multiplexes over it, + * aligned on the **B5 frozen frame contract** (`crates/app-tauri/src/server.rs`): * - * **F1 scope = skeleton.** The connection, frame envelope, request/reply - * correlation by `id`, and the per-session output routing are implemented and - * unit-testable (inject a fake WebSocket factory). What is deliberately deferred - * to **F3/B5** (when a server exists — B3/B4): reconnection/backpressure, exact - * `open`/`launch` reply shape, sequence-gap replay policy, and the full xterm - * round-trip. Those are marked `TODO(F3/B5)`. + * - client→server: `terminal.open` `{cwd,rows,cols}`, `terminal.attach` + * `{sessionId,rows?,cols?,lastSeq?}`, `terminal.input` `{sessionId,bytesBase64}`, + * `terminal.resize` `{sessionId,rows,cols}`, `terminal.detach` `{sessionId}`, + * `terminal.close` `{sessionId}`, `ping`. + * - server→client: `terminal.attached` (ack: `{session,scrollback,nextSeq,status, + * gap,assignedConversationId}`), `terminal.output` `{sessionId,seq,bytesBase64}`, + * `terminal.status` `{sessionId,status,exitCode}`, `error`, `pong`, plus the + * low-frequency `event.domain` stream (domain events). * - * Lives in `src/adapters/**`; touches no `@tauri-apps/api`. + * F3 additions over F1: a **connection state machine** (connecting / connected / + * reconnecting / closed) with automatic reconnection + re-attach and bounded + * scrollback repaint, and per-session **status routing** (`exited`). The terminal + * lifecycle is owned here so a browser reload / network drop transparently + * re-attaches the surviving server-side PTY. Multi-tab "last attach wins" is + * silent on the wire (the evicted attachment simply stops receiving output — see + * the F3 report), so there is nothing to crash on; the socket-close path handles + * disconnection uniformly. + * + * Lives in `src/adapters/**`; touches no `@tauri-apps/api`. The port + * (`TerminalGateway`/`TerminalHandle`) and `TerminalView` are unchanged; the UI + * indication of connection state is written into xterm through the session's own + * output sink (a notice line), so no component needs to know about this client. */ -import type { DomainEvent } from "@/domain"; +import type { DomainEvent, Unsubscribe } from "@/domain"; import { base64ToBytes, bytesToBase64, + type AttachedPayload, type ClientFrame, type ClientFrameKind, type OutputPayload, type ServerFrame, + type StatusPayload, } from "./frames"; /** Minimal WebSocket surface used here; injectable so tests pass a fake. */ @@ -38,14 +55,48 @@ export interface WebSocketLike { /** Factory building a {@link WebSocketLike} for a URL (defaults to global WS). */ export type WebSocketFactory = (url: string) => WebSocketLike; +/** Injectable deferred-call primitives (defaults to global timers). */ +export type SetTimeoutLike = (fn: () => void, ms: number) => unknown; +export type ClearTimeoutLike = (handle: unknown) => void; + +/** UI-facing connection state of the live socket. */ +export type ConnectionState = "connecting" | "connected" | "reconnecting" | "closed"; + /** Configuration for the live client. */ export interface WsLiveClientConfig { /** Base WS URL, e.g. `wss://host:port`. No trailing slash. */ wsUrl: string; - /** Bearer token (ticket #13 auth). B0 note: prefer header/cookie over URL. */ + /** Bearer token (dev/tests only). Prod auth is the session cookie at upgrade. */ token?: string; /** Injected WebSocket factory (defaults to `new WebSocket(url)`). */ socketFactory?: WebSocketFactory; + /** Base reconnect delay in ms (exponential backoff, capped). Default 500. */ + reconnectBaseMs?: number; + /** Max reconnect delay in ms. Default 10000. */ + reconnectMaxMs?: number; + /** Injected `setTimeout` (tests drive reconnection deterministically). */ + setTimeoutImpl?: SetTimeoutLike; + /** Injected `clearTimeout`. */ + clearTimeoutImpl?: ClearTimeoutLike; +} + +/** A live terminal subscription tracked for output routing + reconnection replay. */ +interface TerminalSubscription { + /** Delivers raw PTY bytes (and adapter notices) to the xterm view. */ + onData: (bytes: Uint8Array) => void; + /** Optional lifecycle callback (`exited`), for tests / future UI. */ + onStatus?: (status: string, exitCode?: number | null) => void; + /** Last output sequence seen — sent as `lastSeq` when re-attaching. */ + lastSeq: number; +} + +/** Result of opening/attaching a terminal over the WS. */ +export interface TerminalAttachResult { + sessionId: string; + /** Bounded scrollback bytes to repaint (empty on a fresh open). */ + scrollback: Uint8Array; + assignedConversationId?: string; + status?: string; } let frameCounter = 0; @@ -55,20 +106,50 @@ function nextFrameId(): string { return `c${frameCounter}`; } +const encoder = new TextEncoder(); +/** Encodes a human notice line to bytes for writing into xterm. */ +function notice(text: string): Uint8Array { + return encoder.encode(`\r\n\x1b[2m[${text}]\x1b[0m\r\n`); +} + +/** Concatenates an attached-frame scrollback list into a single byte buffer. */ +function scrollbackBytes(payload: AttachedPayload): Uint8Array { + const chunks = payload.scrollback.map((c) => base64ToBytes(c.bytesBase64)); + const total = chunks.reduce((n, c) => n + c.length, 0); + const out = new Uint8Array(total); + let offset = 0; + for (const c of chunks) { + out.set(c, offset); + offset += c.length; + } + return out; +} + /** - * Manages the single live WebSocket. Callers subscribe an output sink per PTY - * session and a single domain-event handler; the client routes inbound frames. + * Manages the single live WebSocket, the connection state machine, and the live + * terminal sessions (so reconnection re-attaches them transparently). */ export class WsLiveClient { private readonly wsUrl: string; private readonly token?: string; private readonly socketFactory: WebSocketFactory; + private readonly reconnectBaseMs: number; + private readonly reconnectMaxMs: number; + private readonly setTimeoutImpl: SetTimeoutLike; + private readonly clearTimeoutImpl: ClearTimeoutLike; private socket: WebSocketLike | null = null; private opening: Promise | null = null; + private disposed = false; + private everConnected = false; + private connState: ConnectionState = "connecting"; + private reconnectAttempt = 0; + private reconnectHandle: unknown = null; - /** Per-session PTY output sinks (sessionId → onData). */ + /** Per-session PTY output sinks (sessionId → onData). Low-level routing map. */ private readonly outputSinks = new Map void>(); + /** Live terminal subscriptions tracked for reconnection replay + status. */ + private readonly terminals = new Map(); /** Pending command acknowledgements, keyed by client frame id. */ private readonly pending = new Map< string, @@ -76,6 +157,8 @@ export class WsLiveClient { >(); /** The single domain-event handler (set by the system gateway). */ private domainEventHandler: ((event: DomainEvent) => void) | null = null; + /** Connection-state listeners (UI / tests). */ + private readonly connListeners = new Set<(state: ConnectionState) => void>(); constructor(config: WsLiveClientConfig) { this.wsUrl = config.wsUrl.replace(/\/+$/, ""); @@ -83,65 +166,248 @@ export class WsLiveClient { this.socketFactory = config.socketFactory ?? ((url: string) => new WebSocket(url) as unknown as WebSocketLike); + this.reconnectBaseMs = config.reconnectBaseMs ?? 500; + this.reconnectMaxMs = config.reconnectMaxMs ?? 10000; + this.setTimeoutImpl = + config.setTimeoutImpl ?? ((fn, ms) => setTimeout(fn, ms)); + this.clearTimeoutImpl = + config.clearTimeoutImpl ?? ((h) => clearTimeout(h as ReturnType)); } - /** Registers the domain-event handler (replaces any previous one). */ + // ------------------------------------------------------------------------- + // Connection state + // ------------------------------------------------------------------------- + + /** Current connection state. */ + getConnectionState(): ConnectionState { + return this.connState; + } + + /** Subscribes to connection-state transitions (returns an unsubscribe). */ + onConnectionStateChange(listener: (state: ConnectionState) => void): Unsubscribe { + this.connListeners.add(listener); + return () => this.connListeners.delete(listener); + } + + private setConnState(state: ConnectionState): void { + if (this.connState === state) return; + this.connState = state; + for (const listener of this.connListeners) listener(state); + } + + // ------------------------------------------------------------------------- + // Domain events (system gateway) + // ------------------------------------------------------------------------- + setDomainEventHandler(handler: (event: DomainEvent) => void): void { this.domainEventHandler = handler; } - - /** Clears the domain-event handler. */ clearDomainEventHandler(): void { this.domainEventHandler = null; } - /** Registers a per-session PTY output sink. */ + // ------------------------------------------------------------------------- + // Low-level output sink map (used by the agent gateway too) + // ------------------------------------------------------------------------- + setOutputSink(sessionId: string, onData: (bytes: Uint8Array) => void): void { this.outputSinks.set(sessionId, onData); } - - /** Drops a per-session PTY output sink (view detached). */ removeOutputSink(sessionId: string): void { this.outputSinks.delete(sessionId); } + // ------------------------------------------------------------------------- + // Connection lifecycle + // ------------------------------------------------------------------------- + /** Ensures the socket is connected, connecting on first use. */ async ensureConnected(): Promise { if (this.socket) return; if (this.opening) return this.opening; + return this.connect(); + } + private connect(): Promise { + this.setConnState(this.everConnected ? "reconnecting" : "connecting"); this.opening = new Promise((resolve, reject) => { - // B0 auth note: token should ride an `Authorization` header / secure - // cookie at upgrade, NOT a URL secret. Browsers can't set WS upgrade - // headers, so the token placement is a contract point to confirm (see the - // F1 report). The query below is only a placeholder for the skeleton. + // Prod: the session cookie rides the upgrade (same-origin). The optional + // token is a dev/test convenience only (never a prod secret in the URL). const url = this.token ? `${this.wsUrl}/ws/live?token=${encodeURIComponent(this.token)}` : `${this.wsUrl}/ws/live`; const socket = this.socketFactory(url); socket.onopen = () => { this.socket = socket; + this.opening = null; + this.everConnected = true; + this.reconnectAttempt = 0; + this.setConnState("connected"); resolve(); }; socket.onerror = (event) => { this.opening = null; reject(event); }; - socket.onclose = () => { - // TODO(F3/B5): reconnection + re-attach of live sessions. For F1 we drop - // the socket so a later call reconnects fresh. - this.socket = null; - this.rejectAllPending({ code: "WS_CLOSED", message: "socket closed" }); - }; + socket.onclose = () => this.handleSocketClosed(); socket.onmessage = (event) => this.handleMessage(event.data); }); return this.opening; } + private handleSocketClosed(): void { + this.socket = null; + this.opening = null; + this.rejectAllPending({ code: "WS_CLOSED", message: "socket closed" }); + if (this.disposed) { + this.setConnState("closed"); + return; + } + // Only reconnect when there is a live terminal to re-attach; otherwise stay + // idle until the next explicit use. + if (this.terminals.size > 0) { + this.setConnState("reconnecting"); + for (const sub of this.terminals.values()) sub.onData(notice("déconnecté — reconnexion…")); + this.scheduleReconnect(); + } else { + this.setConnState("reconnecting"); + } + } + + private scheduleReconnect(): void { + if (this.disposed || this.reconnectHandle) return; + const delay = Math.min( + this.reconnectMaxMs, + this.reconnectBaseMs * 2 ** this.reconnectAttempt, + ); + this.reconnectAttempt += 1; + this.reconnectHandle = this.setTimeoutImpl(() => { + this.reconnectHandle = null; + void this.reconnectNow(); + }, delay); + } + + private async reconnectNow(): Promise { + if (this.disposed) return; + try { + await this.connect(); + await this.reattachAll(); + } catch { + // Still down: back off and retry (unless everything was detached/closed). + if (this.terminals.size > 0) this.scheduleReconnect(); + } + } + + /** Re-attaches every tracked terminal after a reconnect and repaints scrollback. */ + private async reattachAll(): Promise { + for (const [sessionId, sub] of this.terminals) { + try { + const ack = await this.send("terminal.attach", { + sessionId, + lastSeq: sub.lastSeq, + }); + const payload = ack.payload as unknown as AttachedPayload; + sub.onData(notice("reconnecté")); + const bytes = scrollbackBytes(payload); + if (bytes.length > 0) sub.onData(bytes); + if (typeof payload.nextSeq === "number") { + sub.lastSeq = Math.max(0, payload.nextSeq - 1); + } + } catch { + // A single session failing to re-attach (e.g. server killed it) must not + // block the others; leave it tracked for the next reconnect cycle. + } + } + } + + // ------------------------------------------------------------------------- + // Terminal sessions (high-level; used by HttpTerminalGateway) + // ------------------------------------------------------------------------- + + /** Opens a fresh terminal (`terminal.open`) and starts tracking it. */ + openTerminal(params: { + cwd: string; + rows: number; + cols: number; + nodeId?: string | null; + onData: (bytes: Uint8Array) => void; + onStatus?: (status: string, exitCode?: number | null) => void; + }): Promise { + return this.attachInternal( + "terminal.open", + { cwd: params.cwd, rows: params.rows, cols: params.cols, nodeId: params.nodeId ?? null }, + { onData: params.onData, onStatus: params.onStatus, lastSeq: 0 }, + ); + } + + /** Re-attaches to an existing terminal (`terminal.attach`) and tracks it. */ + attachTerminal(params: { + sessionId: string; + rows?: number; + cols?: number; + lastSeq?: number; + onData: (bytes: Uint8Array) => void; + onStatus?: (status: string, exitCode?: number | null) => void; + }): Promise { + return this.attachInternal( + "terminal.attach", + { + sessionId: params.sessionId, + rows: params.rows, + cols: params.cols, + lastSeq: params.lastSeq ?? null, + }, + { onData: params.onData, onStatus: params.onStatus, lastSeq: params.lastSeq ?? 0 }, + params.sessionId, + ); + } + + private async attachInternal( + kind: ClientFrameKind, + payload: Record, + sub: TerminalSubscription, + knownSessionId?: string, + ): Promise { + const ack = await this.send(kind, payload); + const ap = ack.payload as unknown as AttachedPayload; + const sessionId = knownSessionId ?? ap.session.sessionId; + if (typeof ap.nextSeq === "number") sub.lastSeq = Math.max(0, ap.nextSeq - 1); + this.terminals.set(sessionId, sub); + // Route output to the subscription's onData (the low-level map is the single + // source of routing; lastSeq is tracked in handleMessage). + this.setOutputSink(sessionId, sub.onData); + return { + sessionId, + scrollback: scrollbackBytes(ap), + assignedConversationId: ap.assignedConversationId, + status: ap.status, + }; + } + + /** Detaches the view (server PTY keeps running): stop tracking + `terminal.detach`. */ + async detachTerminal(sessionId: string): Promise { + this.removeTerminal(sessionId); + await this.sendFireAndForget("terminal.detach", { sessionId }); + } + + /** Kills the PTY (`terminal.close`) and stops tracking. */ + async closeTerminalSession(sessionId: string): Promise { + this.removeTerminal(sessionId); + await this.send("terminal.close", { sessionId }); + } + + private removeTerminal(sessionId: string): void { + this.terminals.delete(sessionId); + this.outputSinks.delete(sessionId); + } + + // ------------------------------------------------------------------------- + // Frame I/O + // ------------------------------------------------------------------------- + /** - * Sends a client frame and resolves with its acknowledgement frame (the server - * frame whose `replyTo` equals this frame's `id`). Fire-and-forget frames - * (input/resize/detach) can ignore the returned promise. + * Sends a client frame and resolves with its acknowledgement (the server frame + * whose `replyTo` equals this frame's `id`). */ async send( kind: ClientFrameKind, @@ -150,8 +416,7 @@ export class WsLiveClient { await this.ensureConnected(); const socket = this.socket; if (!socket) { - const err = { code: "WS_CLOSED", message: "socket not connected" }; - throw err; + throw { code: "WS_CLOSED", message: "socket not connected" }; } const id = nextFrameId(); const frame: ClientFrame = { id, kind, payload }; @@ -180,12 +445,19 @@ export class WsLiveClient { /** Closes the socket and rejects everything pending. */ dispose(): void { + this.disposed = true; + if (this.reconnectHandle) { + this.clearTimeoutImpl(this.reconnectHandle); + this.reconnectHandle = null; + } this.rejectAllPending({ code: "WS_CLOSED", message: "client disposed" }); this.outputSinks.clear(); + this.terminals.clear(); this.domainEventHandler = null; this.socket?.close(); this.socket = null; this.opening = null; + this.setConnState("closed"); } private rejectAllPending(err: unknown): void { @@ -198,20 +470,25 @@ export class WsLiveClient { try { frame = JSON.parse(data) as ServerFrame; } catch { - return; // Ignore malformed frames in the skeleton. + return; // Ignore malformed frames. } // Route unsolicited streams first (no replyTo). switch (frame.kind) { case "terminal.output": { const payload = frame.payload as unknown as OutputPayload; + const sub = this.terminals.get(payload.sessionId); + if (sub && typeof payload.seq === "number") sub.lastSeq = payload.seq; const sink = this.outputSinks.get(payload.sessionId); - // TODO(F3/B5): honour `seq` ordering + gap detection for precise replay. if (sink) sink(base64ToBytes(payload.bytesBase64)); return; } + case "terminal.status": { + const payload = frame.payload as unknown as StatusPayload; + this.handleStatus(payload); + return; + } case "event.domain": { - // `payload` is a DomainEventDto (kind-tagged); forward as-is. this.domainEventHandler?.(frame.payload as unknown as DomainEvent); return; } @@ -229,4 +506,25 @@ export class WsLiveClient { } } } + + private handleStatus(payload: StatusPayload): void { + const sub = this.terminals.get(payload.sessionId); + if (!sub) return; + if (payload.status === "exited" || payload.status === "closed") { + const code = payload.exitCode; + sub.onData( + notice( + code === null || code === undefined + ? "session terminée" + : `session terminée (code ${code})`, + ), + ); + sub.onStatus?.(payload.status, payload.exitCode); + // The PTY is gone: stop tracking so a later disconnect does not try to + // re-attach a dead session. + this.removeTerminal(payload.sessionId); + } else { + sub.onStatus?.(payload.status, payload.exitCode); + } + } } diff --git a/frontend/src/adapters/http/wsLiveClientReconnect.test.ts b/frontend/src/adapters/http/wsLiveClientReconnect.test.ts new file mode 100644 index 0000000..da057e7 --- /dev/null +++ b/frontend/src/adapters/http/wsLiveClientReconnect.test.ts @@ -0,0 +1,170 @@ +/** + * F3 — the WS live client connection state machine + reconnection: + * - disconnected (socket close) ⇒ `reconnecting`, a notice is written to xterm; + * - on reconnect, the tracked terminal is re-attached with its `lastSeq` and the + * bounded scrollback is repainted; state returns to `connected`; + * - a `terminal.status` `exited` frame notifies the session and stops tracking + * (so a later disconnect does not try to re-attach a dead PTY). + */ +import { describe, it, expect, vi } from "vitest"; + +import { WsLiveClient, type WebSocketLike, type ConnectionState } from "./wsLiveClient"; +import { bytesToBase64 } from "./frames"; + +class FakeSocket implements WebSocketLike { + sent: string[] = []; + onopen: (() => void) | null = null; + onmessage: ((event: { data: string }) => void) | null = null; + onerror: ((event: unknown) => void) | null = null; + onclose: (() => void) | null = null; + send(data: string): void { + this.sent.push(data); + } + close(): void { + this.onclose?.(); + } + receive(frame: unknown): void { + this.onmessage?.({ data: JSON.stringify(frame) }); + } +} + +function harness() { + const sockets: FakeSocket[] = []; + let timerFn: (() => void) | null = null; + const ws = new WsLiveClient({ + wsUrl: "wss://host", + reconnectBaseMs: 1, + socketFactory: () => { + const s = new FakeSocket(); + sockets.push(s); + queueMicrotask(() => s.onopen?.()); + return s; + }, + setTimeoutImpl: (fn) => { + timerFn = fn; + return 1; + }, + clearTimeoutImpl: () => { + timerFn = null; + }, + }); + const states: ConnectionState[] = []; + ws.onConnectionStateChange((s) => states.push(s)); + return { + ws, + sockets, + states, + fireTimer: () => { + const fn = timerFn; + timerFn = null; + fn?.(); + }, + hasTimer: () => timerFn !== null, + }; +} + +function attachedAck(id: string, sessionId: string, nextSeq: number, scroll?: Uint8Array) { + return { + kind: "terminal.attached", + replyTo: id, + payload: { + session: { sessionId, rows: 30, cols: 120 }, + scrollback: scroll ? [{ seq: 0, bytesBase64: bytesToBase64(scroll) }] : [], + nextSeq, + status: "running", + gap: false, + }, + }; +} + +async function attach( + ws: WsLiveClient, + sockets: FakeSocket[], + sessionId: string, + onData: (b: Uint8Array) => void, + onStatus?: (s: string, c?: number | null) => void, +): Promise { + const p = ws.attachTerminal({ sessionId, onData, onStatus }); + // The socket is created lazily by the factory inside ensureConnected. + 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, 0)); + await p; +} + +const decode = (b: Uint8Array) => new TextDecoder().decode(b); + +describe("WsLiveClient connection state + reconnection", () => { + it("goes connecting → connected on the first attach", async () => { + const h = harness(); + await attach(h.ws, h.sockets, "s1", () => {}); + expect(h.ws.getConnectionState()).toBe("connected"); + expect(h.states).toContain("connected"); + }); + + it("on socket close it reconnects, re-attaches with lastSeq and repaints scrollback", async () => { + const h = harness(); + const chunks: Uint8Array[] = []; + await attach(h.ws, h.sockets, "s1", (b) => chunks.push(b)); + + // Advance lastSeq via an output frame (seq 5). + h.sockets[0].receive({ + kind: "terminal.output", + payload: { sessionId: "s1", seq: 5, bytesBase64: bytesToBase64(new Uint8Array([120])) }, + }); + + // Drop the socket → reconnecting + a "déconnecté" notice. + h.sockets[0].close(); + expect(h.ws.getConnectionState()).toBe("reconnecting"); + expect(chunks.some((c) => decode(c).includes("déconnecté"))).toBe(true); + expect(h.hasTimer()).toBe(true); + + // Fire the reconnect timer → a new socket opens and re-attaches. + h.fireTimer(); + await vi.waitFor(() => expect(h.sockets.length).toBe(2)); + await vi.waitFor(() => expect(h.sockets[1].sent.length).toBeGreaterThan(0)); + const reattach = JSON.parse(h.sockets[1].sent[0]); + expect(reattach.kind).toBe("terminal.attach"); + // lastSeq carries the last output seq seen (5) for precise replay. + expect(reattach.payload).toMatchObject({ sessionId: "s1", lastSeq: 5 }); + + // Server replies with fresh scrollback; the client repaints it + "reconnecté". + const scroll = new Uint8Array([82, 69]); + h.sockets[1].receive(attachedAck(reattach.id, "s1", 3, scroll)); + await vi.waitFor(() => expect(h.ws.getConnectionState()).toBe("connected")); + // The repaint happens after the re-attach ack resolves (a microtask later). + await vi.waitFor(() => + expect(chunks.some((c) => decode(c).includes("reconnecté"))).toBe(true), + ); + expect(chunks.some((c) => c.length === 2 && c[0] === 82 && c[1] === 69)).toBe(true); + }); + + it("routes terminal.status exited to onStatus and stops tracking the session", async () => { + const h = harness(); + const chunks: Uint8Array[] = []; + const onStatus = vi.fn(); + await attach(h.ws, h.sockets, "s1", (b) => chunks.push(b), onStatus); + + h.sockets[0].receive({ + kind: "terminal.status", + payload: { sessionId: "s1", status: "exited", exitCode: 0 }, + }); + + expect(onStatus).toHaveBeenCalledWith("exited", 0); + expect(chunks.some((c) => decode(c).includes("session terminée"))).toBe(true); + + // Session untracked ⇒ a later socket close does not schedule a reconnect. + h.sockets[0].close(); + expect(h.hasTimer()).toBe(false); + }); + + it("dispose() moves to closed and clears any pending reconnect", async () => { + const h = harness(); + await attach(h.ws, h.sockets, "s1", () => {}); + h.sockets[0].close(); + expect(h.ws.getConnectionState()).toBe("reconnecting"); + h.ws.dispose(); + expect(h.ws.getConnectionState()).toBe("closed"); + expect(h.hasTimer()).toBe(false); + }); +});