feat(frontend): xterm.js sur WebSocket avec reconnexion et scrollback (#13)
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:
@ -94,7 +94,7 @@ export interface ServerFrame {
|
|||||||
payload: Record<string, unknown>;
|
payload: Record<string, unknown>;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Payload of a `terminal.attached` acknowledgement. */
|
/** Payload of a `terminal.attached` acknowledgement (B5, `server.rs`). */
|
||||||
export interface AttachedPayload {
|
export interface AttachedPayload {
|
||||||
session: {
|
session: {
|
||||||
sessionId: string;
|
sessionId: string;
|
||||||
@ -104,8 +104,15 @@ export interface AttachedPayload {
|
|||||||
cols: number;
|
cols: number;
|
||||||
};
|
};
|
||||||
nextSeq: 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 }[];
|
scrollback: { seq: number; bytesBase64: string }[];
|
||||||
gap: boolean;
|
gap: boolean;
|
||||||
|
/** Lifecycle status carried on the ack (B5 sends `"running"`). */
|
||||||
|
status?: string;
|
||||||
/** Conversation id minted by an `agent.launch` (mirrors `assignedConversationId`). */
|
/** Conversation id minted by an `agent.launch` (mirrors `assignedConversationId`). */
|
||||||
assignedConversationId?: string;
|
assignedConversationId?: string;
|
||||||
}
|
}
|
||||||
@ -117,6 +124,16 @@ export interface OutputPayload {
|
|||||||
bytesBase64: string;
|
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). */
|
/** Payload of a `chat.output` frame (structured assistant stream). */
|
||||||
export interface ChatOutputPayload {
|
export interface ChatOutputPayload {
|
||||||
sessionId: string;
|
sessionId: string;
|
||||||
|
|||||||
@ -93,12 +93,11 @@ export function makeWsTerminalHandle(
|
|||||||
detach(): void {
|
detach(): void {
|
||||||
// Stop delivering output locally and tell the server the view is gone. The
|
// Stop delivering output locally and tell the server the view is gone. The
|
||||||
// backend PTY keeps running (detach ≠ close), matching the Tauri handle.
|
// backend PTY keeps running (detach ≠ close), matching the Tauri handle.
|
||||||
ws.removeOutputSink(sessionId);
|
// Untracking also disables reconnection re-attach for this session.
|
||||||
void ws.sendFireAndForget("terminal.detach", { sessionId });
|
void ws.detachTerminal(sessionId);
|
||||||
},
|
},
|
||||||
async close(): Promise<void> {
|
async close(): Promise<void> {
|
||||||
ws.removeOutputSink(sessionId);
|
await ws.closeTerminalSession(sessionId);
|
||||||
await ws.send("terminal.close", { sessionId });
|
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
@ -137,37 +136,33 @@ export class HttpTerminalGateway implements TerminalGateway {
|
|||||||
options: OpenTerminalOptions,
|
options: OpenTerminalOptions,
|
||||||
onData: (bytes: Uint8Array) => void,
|
onData: (bytes: Uint8Array) => void,
|
||||||
): Promise<TerminalHandle> {
|
): Promise<TerminalHandle> {
|
||||||
const ack = await this.ws.send("terminal.open", {
|
// B5: `terminal.open` carries no projectId; `cwd` is a server-validated path.
|
||||||
projectId: undefined, // TODO(F3/B5): the port lacks projectId; confirm contract.
|
// A fresh open has empty scrollback (the view does not repaint on open).
|
||||||
nodeId: options.nodeId ?? null,
|
const res = await this.ws.openTerminal({
|
||||||
cwd: options.cwd,
|
cwd: options.cwd,
|
||||||
rows: options.rows,
|
rows: options.rows,
|
||||||
cols: options.cols,
|
cols: options.cols,
|
||||||
|
nodeId: options.nodeId ?? null,
|
||||||
|
onData,
|
||||||
});
|
});
|
||||||
const payload = ack.payload as unknown as AttachedPayload;
|
return makeWsTerminalHandle(res.sessionId, this.ws);
|
||||||
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);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async reattach(
|
async reattach(
|
||||||
sessionId: string,
|
sessionId: string,
|
||||||
onData: (bytes: Uint8Array) => void,
|
onData: (bytes: Uint8Array) => void,
|
||||||
): Promise<ReattachResult> {
|
): Promise<ReattachResult> {
|
||||||
const ack = await this.ws.send("terminal.attach", { sessionId, lastSeq: null });
|
// The view repaints the returned scrollback itself (TerminalView contract);
|
||||||
const payload = ack.payload as unknown as AttachedPayload;
|
// the client does not also push it through `onData` on the initial attach.
|
||||||
this.ws.setOutputSink(sessionId, onData);
|
const res = await this.ws.attachTerminal({ sessionId, onData });
|
||||||
return {
|
return {
|
||||||
handle: makeWsTerminalHandle(sessionId, this.ws),
|
handle: makeWsTerminalHandle(res.sessionId, this.ws),
|
||||||
scrollback: attachedToScrollback(payload),
|
scrollback: res.scrollback,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
async closeTerminal(sessionId: string): Promise<void> {
|
async closeTerminal(sessionId: string): Promise<void> {
|
||||||
this.ws.removeOutputSink(sessionId);
|
await this.ws.closeTerminalSession(sessionId);
|
||||||
await this.ws.send("terminal.close", { sessionId });
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
144
frontend/src/adapters/http/terminalGateway.test.ts
Normal file
144
frontend/src/adapters/http/terminalGateway.test.ts
Normal 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"),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
@ -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
|
* Owns a single WS connection to `{wsUrl}/ws/live` (authenticated by the session
|
||||||
* the B0 draft: PTY output, structured chat output and the low-frequency domain
|
* cookie at the upgrade — same-origin, no URL secret) and multiplexes over it,
|
||||||
* event stream (`event.domain`, replacing Tauri `listen("domain://event")`).
|
* aligned on the **B5 frozen frame contract** (`crates/app-tauri/src/server.rs`):
|
||||||
*
|
*
|
||||||
* **F1 scope = skeleton.** The connection, frame envelope, request/reply
|
* - client→server: `terminal.open` `{cwd,rows,cols}`, `terminal.attach`
|
||||||
* correlation by `id`, and the per-session output routing are implemented and
|
* `{sessionId,rows?,cols?,lastSeq?}`, `terminal.input` `{sessionId,bytesBase64}`,
|
||||||
* unit-testable (inject a fake WebSocket factory). What is deliberately deferred
|
* `terminal.resize` `{sessionId,rows,cols}`, `terminal.detach` `{sessionId}`,
|
||||||
* to **F3/B5** (when a server exists — B3/B4): reconnection/backpressure, exact
|
* `terminal.close` `{sessionId}`, `ping`.
|
||||||
* `open`/`launch` reply shape, sequence-gap replay policy, and the full xterm
|
* - server→client: `terminal.attached` (ack: `{session,scrollback,nextSeq,status,
|
||||||
* round-trip. Those are marked `TODO(F3/B5)`.
|
* 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 {
|
import {
|
||||||
base64ToBytes,
|
base64ToBytes,
|
||||||
bytesToBase64,
|
bytesToBase64,
|
||||||
|
type AttachedPayload,
|
||||||
type ClientFrame,
|
type ClientFrame,
|
||||||
type ClientFrameKind,
|
type ClientFrameKind,
|
||||||
type OutputPayload,
|
type OutputPayload,
|
||||||
type ServerFrame,
|
type ServerFrame,
|
||||||
|
type StatusPayload,
|
||||||
} from "./frames";
|
} from "./frames";
|
||||||
|
|
||||||
/** Minimal WebSocket surface used here; injectable so tests pass a fake. */
|
/** 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). */
|
/** Factory building a {@link WebSocketLike} for a URL (defaults to global WS). */
|
||||||
export type WebSocketFactory = (url: string) => WebSocketLike;
|
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. */
|
/** Configuration for the live client. */
|
||||||
export interface WsLiveClientConfig {
|
export interface WsLiveClientConfig {
|
||||||
/** Base WS URL, e.g. `wss://host:port`. No trailing slash. */
|
/** Base WS URL, e.g. `wss://host:port`. No trailing slash. */
|
||||||
wsUrl: string;
|
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;
|
token?: string;
|
||||||
/** Injected WebSocket factory (defaults to `new WebSocket(url)`). */
|
/** Injected WebSocket factory (defaults to `new WebSocket(url)`). */
|
||||||
socketFactory?: WebSocketFactory;
|
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;
|
let frameCounter = 0;
|
||||||
@ -55,20 +106,50 @@ function nextFrameId(): string {
|
|||||||
return `c${frameCounter}`;
|
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
|
* Manages the single live WebSocket, the connection state machine, and the live
|
||||||
* session and a single domain-event handler; the client routes inbound frames.
|
* terminal sessions (so reconnection re-attaches them transparently).
|
||||||
*/
|
*/
|
||||||
export class WsLiveClient {
|
export class WsLiveClient {
|
||||||
private readonly wsUrl: string;
|
private readonly wsUrl: string;
|
||||||
private readonly token?: string;
|
private readonly token?: string;
|
||||||
private readonly socketFactory: WebSocketFactory;
|
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 socket: WebSocketLike | null = null;
|
||||||
private opening: Promise<void> | 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>();
|
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. */
|
/** Pending command acknowledgements, keyed by client frame id. */
|
||||||
private readonly pending = new Map<
|
private readonly pending = new Map<
|
||||||
string,
|
string,
|
||||||
@ -76,6 +157,8 @@ export class WsLiveClient {
|
|||||||
>();
|
>();
|
||||||
/** The single domain-event handler (set by the system gateway). */
|
/** The single domain-event handler (set by the system gateway). */
|
||||||
private domainEventHandler: ((event: DomainEvent) => void) | null = null;
|
private domainEventHandler: ((event: DomainEvent) => void) | null = null;
|
||||||
|
/** Connection-state listeners (UI / tests). */
|
||||||
|
private readonly connListeners = new Set<(state: ConnectionState) => void>();
|
||||||
|
|
||||||
constructor(config: WsLiveClientConfig) {
|
constructor(config: WsLiveClientConfig) {
|
||||||
this.wsUrl = config.wsUrl.replace(/\/+$/, "");
|
this.wsUrl = config.wsUrl.replace(/\/+$/, "");
|
||||||
@ -83,65 +166,248 @@ export class WsLiveClient {
|
|||||||
this.socketFactory =
|
this.socketFactory =
|
||||||
config.socketFactory ??
|
config.socketFactory ??
|
||||||
((url: string) => new WebSocket(url) as unknown as WebSocketLike);
|
((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 {
|
setDomainEventHandler(handler: (event: DomainEvent) => void): void {
|
||||||
this.domainEventHandler = handler;
|
this.domainEventHandler = handler;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Clears the domain-event handler. */
|
|
||||||
clearDomainEventHandler(): void {
|
clearDomainEventHandler(): void {
|
||||||
this.domainEventHandler = null;
|
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 {
|
setOutputSink(sessionId: string, onData: (bytes: Uint8Array) => void): void {
|
||||||
this.outputSinks.set(sessionId, onData);
|
this.outputSinks.set(sessionId, onData);
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Drops a per-session PTY output sink (view detached). */
|
|
||||||
removeOutputSink(sessionId: string): void {
|
removeOutputSink(sessionId: string): void {
|
||||||
this.outputSinks.delete(sessionId);
|
this.outputSinks.delete(sessionId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// -------------------------------------------------------------------------
|
||||||
|
// Connection lifecycle
|
||||||
|
// -------------------------------------------------------------------------
|
||||||
|
|
||||||
/** Ensures the socket is connected, connecting on first use. */
|
/** Ensures the socket is connected, connecting on first use. */
|
||||||
async ensureConnected(): Promise<void> {
|
async ensureConnected(): Promise<void> {
|
||||||
if (this.socket) return;
|
if (this.socket) return;
|
||||||
if (this.opening) return this.opening;
|
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) => {
|
this.opening = new Promise<void>((resolve, reject) => {
|
||||||
// B0 auth note: token should ride an `Authorization` header / secure
|
// Prod: the session cookie rides the upgrade (same-origin). The optional
|
||||||
// cookie at upgrade, NOT a URL secret. Browsers can't set WS upgrade
|
// token is a dev/test convenience only (never a prod secret in the URL).
|
||||||
// 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.
|
|
||||||
const url = this.token
|
const url = this.token
|
||||||
? `${this.wsUrl}/ws/live?token=${encodeURIComponent(this.token)}`
|
? `${this.wsUrl}/ws/live?token=${encodeURIComponent(this.token)}`
|
||||||
: `${this.wsUrl}/ws/live`;
|
: `${this.wsUrl}/ws/live`;
|
||||||
const socket = this.socketFactory(url);
|
const socket = this.socketFactory(url);
|
||||||
socket.onopen = () => {
|
socket.onopen = () => {
|
||||||
this.socket = socket;
|
this.socket = socket;
|
||||||
|
this.opening = null;
|
||||||
|
this.everConnected = true;
|
||||||
|
this.reconnectAttempt = 0;
|
||||||
|
this.setConnState("connected");
|
||||||
resolve();
|
resolve();
|
||||||
};
|
};
|
||||||
socket.onerror = (event) => {
|
socket.onerror = (event) => {
|
||||||
this.opening = null;
|
this.opening = null;
|
||||||
reject(event);
|
reject(event);
|
||||||
};
|
};
|
||||||
socket.onclose = () => {
|
socket.onclose = () => this.handleSocketClosed();
|
||||||
// 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.onmessage = (event) => this.handleMessage(event.data);
|
socket.onmessage = (event) => this.handleMessage(event.data);
|
||||||
});
|
});
|
||||||
return this.opening;
|
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
|
* Sends a client frame and resolves with its acknowledgement (the server frame
|
||||||
* frame whose `replyTo` equals this frame's `id`). Fire-and-forget frames
|
* whose `replyTo` equals this frame's `id`).
|
||||||
* (input/resize/detach) can ignore the returned promise.
|
|
||||||
*/
|
*/
|
||||||
async send(
|
async send(
|
||||||
kind: ClientFrameKind,
|
kind: ClientFrameKind,
|
||||||
@ -150,8 +416,7 @@ export class WsLiveClient {
|
|||||||
await this.ensureConnected();
|
await this.ensureConnected();
|
||||||
const socket = this.socket;
|
const socket = this.socket;
|
||||||
if (!socket) {
|
if (!socket) {
|
||||||
const err = { code: "WS_CLOSED", message: "socket not connected" };
|
throw { code: "WS_CLOSED", message: "socket not connected" };
|
||||||
throw err;
|
|
||||||
}
|
}
|
||||||
const id = nextFrameId();
|
const id = nextFrameId();
|
||||||
const frame: ClientFrame = { id, kind, payload };
|
const frame: ClientFrame = { id, kind, payload };
|
||||||
@ -180,12 +445,19 @@ export class WsLiveClient {
|
|||||||
|
|
||||||
/** Closes the socket and rejects everything pending. */
|
/** Closes the socket and rejects everything pending. */
|
||||||
dispose(): void {
|
dispose(): void {
|
||||||
|
this.disposed = true;
|
||||||
|
if (this.reconnectHandle) {
|
||||||
|
this.clearTimeoutImpl(this.reconnectHandle);
|
||||||
|
this.reconnectHandle = null;
|
||||||
|
}
|
||||||
this.rejectAllPending({ code: "WS_CLOSED", message: "client disposed" });
|
this.rejectAllPending({ code: "WS_CLOSED", message: "client disposed" });
|
||||||
this.outputSinks.clear();
|
this.outputSinks.clear();
|
||||||
|
this.terminals.clear();
|
||||||
this.domainEventHandler = null;
|
this.domainEventHandler = null;
|
||||||
this.socket?.close();
|
this.socket?.close();
|
||||||
this.socket = null;
|
this.socket = null;
|
||||||
this.opening = null;
|
this.opening = null;
|
||||||
|
this.setConnState("closed");
|
||||||
}
|
}
|
||||||
|
|
||||||
private rejectAllPending(err: unknown): void {
|
private rejectAllPending(err: unknown): void {
|
||||||
@ -198,20 +470,25 @@ export class WsLiveClient {
|
|||||||
try {
|
try {
|
||||||
frame = JSON.parse(data) as ServerFrame;
|
frame = JSON.parse(data) as ServerFrame;
|
||||||
} catch {
|
} catch {
|
||||||
return; // Ignore malformed frames in the skeleton.
|
return; // Ignore malformed frames.
|
||||||
}
|
}
|
||||||
|
|
||||||
// Route unsolicited streams first (no replyTo).
|
// Route unsolicited streams first (no replyTo).
|
||||||
switch (frame.kind) {
|
switch (frame.kind) {
|
||||||
case "terminal.output": {
|
case "terminal.output": {
|
||||||
const payload = frame.payload as unknown as OutputPayload;
|
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);
|
const sink = this.outputSinks.get(payload.sessionId);
|
||||||
// TODO(F3/B5): honour `seq` ordering + gap detection for precise replay.
|
|
||||||
if (sink) sink(base64ToBytes(payload.bytesBase64));
|
if (sink) sink(base64ToBytes(payload.bytesBase64));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
case "terminal.status": {
|
||||||
|
const payload = frame.payload as unknown as StatusPayload;
|
||||||
|
this.handleStatus(payload);
|
||||||
|
return;
|
||||||
|
}
|
||||||
case "event.domain": {
|
case "event.domain": {
|
||||||
// `payload` is a DomainEventDto (kind-tagged); forward as-is.
|
|
||||||
this.domainEventHandler?.(frame.payload as unknown as DomainEvent);
|
this.domainEventHandler?.(frame.payload as unknown as DomainEvent);
|
||||||
return;
|
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
170
frontend/src/adapters/http/wsLiveClientReconnect.test.ts
Normal 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);
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user