blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(frontend): xterm.js sur WebSocket avec reconnexion et scrollback (#13)

Browse Source 
Lot F3 du chantier server/client mode : le client web branche xterm.js sur
le terminal distant via WebSocket, en face de l'endpoint PTY B5.

- wsLiveClient.ts : transport WebSocket du terminal avec reconnexion.
- streamGateways.ts : gateway terminal (open/attach/data/resize/close).
- frames.ts : frames PTY alignées sur le contrat serveur B5.
- Tests : terminalGateway.test.ts, wsLiveClientReconnect.test.ts.

Validé : frontend 743 tests verts (F3 ciblé 12), cohérence des frames
B5↔F3 confirmée, desktop non régressé.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-07-15 18:09:25 +02:00
parent 917be995a4
commit 7487902ddb
5 changed files with 685 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
frontend/src/adapters/http/frames.ts
View File

@ -94,7 +94,7 @@ export interface ServerFrame {
payload: Record<string, unknown>;
}
/** 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;

35
frontend/src/adapters/http/streamGateways.ts
View File

@ -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<void> {
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<TerminalHandle> {
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<ReattachResult> {
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<void> {
this.ws.removeOutputSink(sessionId);
await this.ws.send("terminal.close", { sessionId });
await this.ws.closeTerminalSession(sessionId);
}
}

144
frontend/src/adapters/http/terminalGateway.test.ts Normal file
View File

@ -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<Record<string, unknown>> {
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"),
);
});
});

378
frontend/src/adapters/http/wsLiveClient.ts
View File

@ -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<void> | 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<string, (bytes: Uint8Array) => void>();
/** Live terminal subscriptions tracked for reconnection replay + status. */
private readonly terminals = new Map<string, TerminalSubscription>();
/** 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<typeof setTimeout>));
}
/** 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<void> {
if (this.socket) return;
if (this.opening) return this.opening;
return this.connect();
}
private connect(): Promise<void> {
this.setConnState(this.everConnected ? "reconnecting" : "connecting");
this.opening = new Promise<void>((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<void> {
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<void> {
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<TerminalAttachResult> {
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<TerminalAttachResult> {
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<string, unknown>,
sub: TerminalSubscription,
knownSessionId?: string,
): Promise<TerminalAttachResult> {
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<void> {
this.removeTerminal(sessionId);
await this.sendFireAndForget("terminal.detach", { sessionId });
}
/** Kills the PTY (`terminal.close`) and stops tracking. */
async closeTerminalSession(sessionId: string): Promise<void> {
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);
}
}
}

170
frontend/src/adapters/http/wsLiveClientReconnect.test.ts Normal file
View File

@ -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<void> {
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);
});
});