From 791f92d48d9699a3b1ad3ab2387352bad83654f5 Mon Sep 17 00:00:00 2001 From: pjhsion Date: Wed, 7 Oct 2026 17:44:11 +0900 Subject: [PATCH 1/4] Add isolated websocket test harness foundation --- bunfig.websocket.toml | 2 + package.json | 2 + .../create-server-harness.test.ts | 100 +++++++++ .../websocket-test/create-server-harness.ts | 179 ++++++++++++++++ .../create-server.smoke.integration.test.ts | 196 ++++++++++++++++++ .../websocket-test/fake-herdr-client.test.ts | 40 ++++ server/websocket-test/fake-herdr-client.ts | 92 ++++++++ .../websocket-test/fake-pty-session.test.ts | 68 ++++++ server/websocket-test/fake-pty-session.ts | 101 +++++++++ 9 files changed, 780 insertions(+) create mode 100644 bunfig.websocket.toml create mode 100644 server/websocket-test/create-server-harness.test.ts create mode 100644 server/websocket-test/create-server-harness.ts create mode 100644 server/websocket-test/create-server.smoke.integration.test.ts create mode 100644 server/websocket-test/fake-herdr-client.test.ts create mode 100644 server/websocket-test/fake-herdr-client.ts create mode 100644 server/websocket-test/fake-pty-session.test.ts create mode 100644 server/websocket-test/fake-pty-session.ts diff --git a/bunfig.websocket.toml b/bunfig.websocket.toml new file mode 100644 index 0000000000..2bd13cacd7 --- /dev/null +++ b/bunfig.websocket.toml @@ -0,0 +1,2 @@ +[test] +preload = [] diff --git a/package.json b/package.json index 81c14f4382..715ea10451 100644 --- a/package.json +++ b/package.json @@ -16,6 +16,8 @@ "start": "bun run build && bun run server/managed.ts", "typecheck": "tsc --noEmit", "test": "bun test", + "test:websocket": "bun --config=bunfig.websocket.toml test --isolate server/output-window.test.ts server/websocket-test/fake-herdr-client.test.ts server/websocket-test/fake-pty-session.test.ts server/websocket-test/create-server-harness.test.ts server/websocket-test/create-server.smoke.integration.test.ts", + "test:websocket:mutations": "bun --config=bunfig.websocket.toml test --isolate server/websocket-test/mutation-source.test.ts server/websocket-test/mutation-controls.integration.test.ts", "test:ui": "bun run build && bun scripts/ui-regression.ts", "generate:types": "bun run scripts/generate-protocol-types.ts", "build:remote": "bun run scripts/build-remote-bundle.ts", diff --git a/server/websocket-test/create-server-harness.test.ts b/server/websocket-test/create-server-harness.test.ts new file mode 100644 index 0000000000..b6f4a310d8 --- /dev/null +++ b/server/websocket-test/create-server-harness.test.ts @@ -0,0 +1,100 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { access, mkdir, readFile, writeFile } from "node:fs/promises"; +import { basename, dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { createServerHarness, defaultServerIndex, evidenceDirectory, resolveContainedPath, websocketRunRoot } from "./create-server-harness.ts"; +import * as fakeHerdrClient from "./fake-herdr-client.ts"; + +const checkoutRoot = resolve(fileURLToPath(new URL("../..", import.meta.url))); +const runRoot = websocketRunRoot; +const defaultPath = resolve(checkoutRoot, "server/index.ts"); +const evidencePath = resolve(evidenceDirectory(), "create-server-smoke.jsonl"); +const originalIndexOverride = process.env.SODAM_WS_SERVER_INDEX; +const originalEvidenceOverride = process.env.SODAM_WS_EVIDENCE_DIR; +const originalSocket = process.env.HERDR_SOCKET; + +afterEach(() => { + if (originalIndexOverride === undefined) delete process.env.SODAM_WS_SERVER_INDEX; + else process.env.SODAM_WS_SERVER_INDEX = originalIndexOverride; + if (originalEvidenceOverride === undefined) delete process.env.SODAM_WS_EVIDENCE_DIR; + else process.env.SODAM_WS_EVIDENCE_DIR = originalEvidenceOverride; + if (originalSocket === undefined) delete process.env.HERDR_SOCKET; + else process.env.HERDR_SOCKET = originalSocket; +}); + +describe("original createServer harness boundary", () => { + test("uses the original source by default and contains allowed overrides", () => { + delete process.env.SODAM_WS_SERVER_INDEX; + delete process.env.SODAM_WS_EVIDENCE_DIR; + expect(defaultServerIndex()).toBe(resolve(defaultPath)); + expect(evidenceDirectory()).toBe(resolve(runRoot, "evidence")); + process.env.SODAM_WS_SERVER_INDEX = resolve(runRoot, "alternate/index.ts"); + process.env.SODAM_WS_EVIDENCE_DIR = resolve(runRoot, "alternate/evidence"); + expect(defaultServerIndex()).toBe(resolve(runRoot, "alternate/index.ts")); + expect(evidenceDirectory()).toBe(resolve(runRoot, "alternate/evidence")); + }); + + test("rejects an override outside the run root", () => { + process.env.SODAM_WS_SERVER_INDEX = defaultPath; + expect(() => defaultServerIndex()).toThrow("path must resolve under"); + const siblingRunRoot = resolve(runRoot, "..", `${basename(runRoot)}-attacker`); + const siblingStateDir = resolve(siblingRunRoot, "state-1"); + expect(() => resolveContainedPath(siblingStateDir, siblingStateDir)).toThrow("path must resolve under"); + }); + + test("registers fake modules before import and cleans all observed resources", async () => { + delete process.env.SODAM_WS_SERVER_INDEX; + delete process.env.SODAM_WS_EVIDENCE_DIR; + fakeHerdrClient.resetFakeHerdrCounters(); + const harness = await createServerHarness(); + expect(harness.serverIndex).toBe(resolve(defaultPath)); + const baseUrl = new URL(harness.baseUrl); + expect(baseUrl.protocol).toBe("ws:"); + expect(baseUrl.hostname).toBe("127.0.0.1"); + expect(Number(baseUrl.port)).toBeGreaterThan(0); + expect(harness.stateDir.startsWith(`${resolve(runRoot)}${process.platform === "win32" ? "\\" : "/"}state-`)).toBe(true); + const opening = harness.openSocket(); + expect(opening.readyState).toBe(WebSocket.CONNECTING); + const openEvent = harness.prearmSocketEvent("open"); + const snapshot = new Promise>((resolvePromise, rejectPromise) => { + const timer = setTimeout(() => rejectPromise(new Error("timed out waiting for initial snapshot")), 5000); + opening.addEventListener("message", (event: MessageEvent) => { + clearTimeout(timer); + resolvePromise(event); + }, { once: true }); + }); + const [opened, snapshotEvent] = await Promise.all([openEvent, snapshot]); + expect(opened.socket).toBe(opening); + expect(snapshotEvent.data).toContain('"type":"snapshot"'); + const cleanup = await harness.cleanup(); + let stateAbsent = false; + try { + await access(harness.stateDir); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") stateAbsent = true; + else throw error; + } + expect(stateAbsent).toBe(true); + expect(harness.instances).toHaveLength(0); + expect(cleanup.subscriptionCloseDelta).toBeGreaterThanOrEqual(0); + const evidence = { + kind: "create-server-smoke", + testCommand: "bun --config=bunfig.websocket.toml test --isolate server/websocket-test/create-server-harness.test.ts", + serverIndex: harness.serverIndex, + baseUrl: harness.baseUrl, + port: Number(baseUrl.port), + endpointHost: baseUrl.hostname, + stateDirAbsentAfterCleanup: stateAbsent, + stateDir: harness.stateDir, + subscriptionCloseDelta: cleanup.subscriptionCloseDelta, + fakePtyInstancesAtCleanup: harness.instances.length, + herdrSocketRestored: process.env.HERDR_SOCKET === originalSocket, + bunVersion: Bun.version, + externalHerdrCalls: 0, + }; + await mkdir(dirname(evidencePath), { recursive: true }); + await writeFile(evidencePath, `${JSON.stringify(evidence)}\n`); + expect((await readFile(evidencePath, "utf8")).trim()).toBe(JSON.stringify(evidence)); + expect(cleanup.subscriptionCloseDelta).toBeGreaterThan(0); + }); +}); diff --git a/server/websocket-test/create-server-harness.ts b/server/websocket-test/create-server-harness.ts new file mode 100644 index 0000000000..f1d07648f2 --- /dev/null +++ b/server/websocket-test/create-server-harness.ts @@ -0,0 +1,179 @@ +import { mkdir, rm } from "node:fs/promises"; +import { fileURLToPath } from "node:url"; +import { isAbsolute, join, relative, resolve, sep } from "node:path"; +import { tmpdir } from "node:os"; +import { mock } from "bun:test"; +import { FakePtySession, onPtyEvent, queueInitialOutput, type PtyEvent, type PtyEventType } from "./fake-pty-session.ts"; +import * as fakeHerdrClient from "./fake-herdr-client.ts"; + +const checkoutRoot = resolve(fileURLToPath(new URL("../..", import.meta.url))); +export const websocketRunRoot = resolve(process.env.SODAM_WS_RUN_ROOT ?? join(tmpdir(), `herdr-websocket-${crypto.randomUUID()}`)); +const originalIndex = resolve(checkoutRoot, "server/index.ts"); +const clientPath = resolve(checkoutRoot, "server/herdr/client.ts"); +const ptyPath = resolve(checkoutRoot, "server/pty/session.ts"); +const fakePane = "fake-pane"; +const timeoutMs = 5000; + +export function resolveContainedPath(value: string, fallback: string): string { + const candidate = resolve(value || fallback); + const rel = relative(websocketRunRoot, candidate); + if (rel === ".." || rel.startsWith(`..${sep}`) || isAbsolute(rel)) { + throw new Error(`path must resolve under ${websocketRunRoot}: ${candidate}`); + } + return candidate; +} + +export function defaultServerIndex(): string { + const override = process.env.SODAM_WS_SERVER_INDEX; + return override === undefined ? originalIndex : resolveContainedPath(override, originalIndex); +} + +export function evidenceDirectory(): string { + return resolveContainedPath(process.env.SODAM_WS_EVIDENCE_DIR ?? join(websocketRunRoot, "evidence"), join(websocketRunRoot, "evidence")); +} + +export interface PtyEventWaiter { + readonly promise: Promise; + readonly cancel: () => void; +} + +export function prearmPtyEvent(type: PtyEventType, timeout = timeoutMs): PtyEventWaiter { + let unsubscribe: () => void = () => {}; + let timer: ReturnType | undefined; + let cancel: () => void = () => {}; + const promise = new Promise((resolvePromise, rejectPromise) => { + unsubscribe = onPtyEvent((event) => { + if (event.type !== type) return; + unsubscribe(); + if (timer) clearTimeout(timer); + resolvePromise(event); + }); + timer = setTimeout(() => { + unsubscribe(); + rejectPromise(new Error(`timed out waiting for PTY ${type}`)); + }, timeout); + cancel = () => { + unsubscribe(); + if (timer) clearTimeout(timer); + }; + }); + return { promise, cancel }; +} + +export async function createServerHarness(): Promise<{ + readonly serverIndex: string; + readonly baseUrl: string; + readonly stateDir: string; + readonly paneId: string; + readonly instances: FakePtySession[]; + readonly openSocket: () => WebSocket; + readonly connect: () => Promise; + readonly prearmSocketEvent: (type: "open" | "close" | "error", timeout?: number) => Promise<{ socket: WebSocket; event: Event }>; + readonly waitForPtyEvent: (paneId: string, type: PtyEventType) => PtyEventWaiter; + readonly queueInitialPtyOutput: (...chunks: string[]) => void; + readonly emitPtyData: (paneId: string, data: string) => void; + readonly queuePtyOutputOnResume: (paneId: string, data: string) => void; + readonly cleanup: () => Promise<{ readonly subscriptionCloseDelta: number }>; +}> { + const serverIndex = defaultServerIndex(); + const evidenceDir = evidenceDirectory(); + await mkdir(evidenceDir, { recursive: true }); + const stateDir = resolve(websocketRunRoot, `state-${crypto.randomUUID()}`); + await mkdir(stateDir, { recursive: true }); + const priorSocket = process.env.HERDR_SOCKET; + const socketPath = resolve(websocketRunRoot, `missing-herdr-${crypto.randomUUID()}.sock`); + const restoreSocket = () => { + if (priorSocket === undefined) delete process.env.HERDR_SOCKET; + else process.env.HERDR_SOCKET = priorSocket; + }; + process.env.HERDR_SOCKET = socketPath; + const instances: FakePtySession[] = []; + let server: { port: number; hostname: string; stop: () => void } | undefined; + try { + mock.module(clientPath, () => ({ + ...fakeHerdrClient, + herdrRpc: fakeHerdrClient.unexpectedRpc, + })); + mock.module(ptyPath, () => ({ + PtySession: class extends FakePtySession { + constructor(options: ConstructorParameters[0]) { + const mapped = { ...options, onExit: (code: number | null) => options.onExit(code) }; + super(mapped); + instances.push(this); + } + }, + })); + const imported = await import(serverIndex); + server = imported.createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir, machines: false, registerBridge: false }); + } catch (error) { + server?.stop(); + for (const instance of instances) instance.kill(); + await Promise.all(instances.map((instance) => instance.exited)); + await rm(stateDir, { recursive: true, force: true }); + restoreSocket(); + throw error; + } + if (server === undefined) throw new Error("createServer did not return a running server"); + const runningServer = server; + + const sockets = new Set(); + const closeWaiters = new Map>(); + const ptyWaiters = new Set(); + const prearmSocketEvent = (type: "open" | "close" | "error", timeout = timeoutMs) => { + const socket = [...sockets].at(-1); + if (!socket) throw new Error("connect a socket before pre-arming its event"); + let timer: ReturnType; + const eventPromise = new Promise((resolvePromise, rejectPromise) => { + socket.addEventListener(type, (event) => { clearTimeout(timer); resolvePromise(event); }, { once: true }); + timer = setTimeout(() => rejectPromise(new Error(`timed out waiting for socket ${type}`)), timeout); + }); + const result = eventPromise.then((event) => ({ socket, event })); + if (type === "close") closeWaiters.set(socket, eventPromise); + return result; + }; + const openSocket = (): WebSocket => { + const opening = new WebSocket(`ws://${runningServer.hostname}:${runningServer.port}/ws`); + sockets.add(opening); + prearmSocketEvent("close"); + return opening; + }; + const connect = async (): Promise => { + const opening = openSocket(); + await new Promise((resolvePromise, rejectPromise) => { + const timer = setTimeout(() => rejectPromise(new Error("timed out opening websocket")), timeoutMs); + opening.addEventListener("open", () => { clearTimeout(timer); resolvePromise(); }, { once: true }); + opening.addEventListener("error", () => { clearTimeout(timer); rejectPromise(new Error("websocket failed to open")); }, { once: true }); + }); + return opening; + }; + const waitForPtyEvent = (paneId: string, type: PtyEventType): PtyEventWaiter => { + if (paneId !== fakePane) throw new Error(`unknown fake pane: ${paneId}`); + const waiter = prearmPtyEvent(type); + ptyWaiters.add(waiter); + return waiter; + }; + + return { + serverIndex, baseUrl: `ws://${runningServer.hostname}:${runningServer.port}`, stateDir, paneId: fakePane, instances, openSocket, connect, prearmSocketEvent, waitForPtyEvent, + queueInitialPtyOutput: (...chunks) => queueInitialOutput(...chunks), + emitPtyData: (paneId, data) => { if (paneId !== fakePane) throw new Error(`unknown fake pane: ${paneId}`); instances.at(-1)?.emitData(data); }, + queuePtyOutputOnResume: (paneId, data) => { if (paneId !== fakePane) throw new Error(`unknown fake pane: ${paneId}`); instances.at(-1)?.queueOutputOnResume(data); }, + async cleanup() { + const before = fakeHerdrClient.getFakeHerdrCounters().subscriptionCloseCount; + for (const waiter of ptyWaiters) waiter.cancel(); + for (const socket of sockets) socket.close(); + await Promise.all([...closeWaiters.values()]); + runningServer.stop(); + for (const instance of instances) instance.kill(); + await Promise.all(instances.map((instance) => instance.exited)); + try { + const subscriptionCloseDelta = fakeHerdrClient.getFakeHerdrCounters().subscriptionCloseCount - before; + await rm(stateDir, { recursive: true, force: true }); + return { subscriptionCloseDelta }; + } finally { + restoreSocket(); + sockets.clear(); closeWaiters.clear(); instances.length = 0; ptyWaiters.clear(); + } + }, + }; +} diff --git a/server/websocket-test/create-server.smoke.integration.test.ts b/server/websocket-test/create-server.smoke.integration.test.ts new file mode 100644 index 0000000000..40ead22033 --- /dev/null +++ b/server/websocket-test/create-server.smoke.integration.test.ts @@ -0,0 +1,196 @@ +import { access, mkdir, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { expect, test } from "bun:test"; +import { createServerHarness, resolveContainedPath, websocketRunRoot } from "./create-server-harness.ts"; +import { onPtyEvent } from "./fake-pty-session.ts"; + +const evidencePath = join(websocketRunRoot, "evidence", "smoke-ws.jsonl"); +const cleanupPath = join(websocketRunRoot, "evidence", "smoke-ws-cleanup.json"); +const attachMessage = '{"type":"attach","pane_id":"fake-pane","cols":100,"rows":30,"flow_control":"ack"}'; +const noOutputDiagnostic = "timed out waiting for WebSocket message after attach (no-output guard)"; +const records: string[] = []; +const cleanupCases: { readonly case: string; readonly socketsClosed: number; readonly fakeExitedResolved: boolean; readonly subscriptionsClosed: boolean; readonly stateDirAbsent: boolean; readonly stoppedPortHandshakeFailed: boolean }[] = []; + +await mkdir(dirname(evidencePath), { recursive: true }); +await writeFile(evidencePath, "", "utf8"); +await writeFile(cleanupPath, "", "utf8"); + +interface WsFrame { readonly type: string; readonly [key: string]: unknown } + +function boundedMessage(socket: WebSocket, timeoutMs: number, diagnostic: string): Promise { + return new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const cleanup = () => { + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + socket.removeEventListener("close", onClose); + socket.removeEventListener("error", onError); + }; + const onMessage = (event: MessageEvent) => { + cleanup(); + const parsed: unknown = JSON.parse(String(event.data)); + if (typeof parsed !== "object" || parsed === null || !("type" in parsed) || typeof parsed.type !== "string") { + rejectPromise(new Error("WebSocket frame has no string type")); + return; + } + resolvePromise(parsed as WsFrame); + }; + const onClose = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket closed`)); }; + const onError = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket errored`)); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error(diagnostic)); }, timeoutMs); + socket.addEventListener("message", onMessage); + socket.addEventListener("close", onClose); + socket.addEventListener("error", onError); + }); +} + +function boundedNoOutputMessage(socket: WebSocket): Promise { + return boundedMessage(socket, 1000, noOutputDiagnostic); +} + +async function finishCase( + harness: Awaited>, + socket: WebSocket | undefined, + caseName: string, + fakeExited: Promise | undefined, + exitEvent: Promise | undefined, + unsubscribeExit: (() => void) | undefined, + socketWaits: Promise[], +): Promise { + const stateDir = harness.stateDir; + const sessionExit = fakeExited ?? harness.instances.at(-1)?.exited; + const settledSocketWaits = Promise.allSettled(socketWaits); + let cleanupResult: Awaited> | undefined; + let fakeExitedResolved = false; + try { + cleanupResult = await harness.cleanup(); + await settledSocketWaits; + if (sessionExit) { + await sessionExit; + fakeExitedResolved = true; + if (exitEvent) await exitEvent; + } + unsubscribeExit?.(); + let stateDirAbsent = false; + try { + await access(stateDir); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") stateDirAbsent = true; + else throw error; + } + const stoppedPortHandshakeFailed = await new Promise((resolvePromise, rejectPromise) => { + const stopped = new WebSocket(new URL("/ws", harness.baseUrl)); + const timer = setTimeout(() => { stopped.close(); rejectPromise(new Error("stopped-port handshake did not settle")); }, 1000); + const settle = (failed: boolean) => { + clearTimeout(timer); + stopped.removeEventListener("error", onError); + stopped.removeEventListener("open", onOpen); + if (failed) resolvePromise(true); + else { stopped.close(); resolvePromise(false); } + }; + const onError = () => settle(true); + const onOpen = () => settle(false); + stopped.addEventListener("error", onError); + stopped.addEventListener("open", onOpen); + }); + const subscriptionsClosed = cleanupResult.subscriptionCloseDelta > 0; + const stateDirUnderRunRoot = resolveContainedPath(stateDir, stateDir) === stateDir; + const cleanupCase = { case: caseName, socketsClosed: socket?.readyState === WebSocket.CLOSED ? 1 : 0, fakeExitedResolved, subscriptionsClosed, stateDirAbsent, stoppedPortHandshakeFailed }; + cleanupCases.push(cleanupCase); + const priorJsonl = await Bun.file(evidencePath).text(); + await writeFile(evidencePath, `${priorJsonl}${records.splice(0).join("\n")}\n`, "utf8"); + await writeFile(cleanupPath, `${JSON.stringify({ cases: cleanupCases }, null, 2)}\n`, "utf8"); + expect(stateDirUnderRunRoot).toBe(true); + expect(cleanupCase.socketsClosed).toBe(socket ? 1 : 0); + expect(cleanupCase.fakeExitedResolved).toBe(true); + expect(cleanupCase.subscriptionsClosed).toBe(true); + expect(cleanupCase.stateDirAbsent).toBe(true); + expect(cleanupCase.stoppedPortHandshakeFailed).toBe(true); + } finally { + if (cleanupResult === undefined) { + await harness.cleanup(); + unsubscribeExit?.(); + } + } +} + +test("real localhost attach replays queued PTY output", async () => { + const harness = await createServerHarness(); + let socket: WebSocket | undefined; + let fakeExited: Promise | undefined; + let exitEvent: Promise | undefined; + let unsubscribeExit: (() => void) | undefined; + const socketWaits: Promise[] = []; + try { + expect(resolveContainedPath(harness.stateDir, harness.stateDir)).toBe(harness.stateDir); + harness.queueInitialPtyOutput("SMOKE"); + socket = harness.openSocket(); + const openWait = harness.prearmSocketEvent("open"); + const snapshotWait = boundedMessage(socket, 5000, "timed out waiting for initial snapshot"); + socketWaits.push(openWait); + socketWaits.push(snapshotWait); + const [openEvent, snapshotFrame] = await Promise.all([openWait, snapshotWait]); + expect(openEvent.socket).toBe(socket); + expect(snapshotFrame.type).toBe("snapshot"); + const frameWait = boundedMessage(socket, 5000, "timed out waiting for PTY output after attach"); + socketWaits.push(frameWait); + exitEvent = new Promise((resolvePromise) => { + const unsubscribe = onPtyEvent((event) => { + if (event.type !== "exit") return; + unsubscribe(); + resolvePromise(); + }); + unsubscribeExit = unsubscribe; + }); + socket.send(attachMessage); + fakeExited = harness.instances.at(-1)?.exited; + const frame = await frameWait; + expect(frame.type).toBe("pty-data"); + expect(frame.data).toContain("SMOKE"); + const data = String(frame.data); + const flow = frame.flow; + expect(typeof flow).toBe("object"); + expect(flow).not.toBeNull(); + if (typeof flow !== "object" || flow === null || !("offset" in flow)) throw new Error("pty-data frame omitted flow offset"); + const utf8ByteOffset = Number(flow.offset); + expect(utf8ByteOffset).toBe(new TextEncoder().encode(data.slice(0, data.indexOf("SMOKE") + "SMOKE".length)).byteLength); + records.push(JSON.stringify({ case: "attach-with-output", snapshot: true, ptyData: true, data, utf8ByteOffset })); + } finally { + await finishCase(harness, socket, "attach-with-output", fakeExited, exitEvent, unsubscribeExit, socketWaits); + } +}, 15000); + +test("no-output attach reports bounded diagnostic and still cleans up", async () => { + const harness = await createServerHarness(); + let socket: WebSocket | undefined; + let fakeExited: Promise | undefined; + let exitEvent: Promise | undefined; + let unsubscribeExit: (() => void) | undefined; + const socketWaits: Promise[] = []; + try { + socket = harness.openSocket(); + const openWait = harness.prearmSocketEvent("open"); + const snapshotWait = boundedMessage(socket, 5000, "timed out waiting for initial snapshot"); + socketWaits.push(openWait); + socketWaits.push(snapshotWait); + const [openEvent, snapshot] = await Promise.all([openWait, snapshotWait]); + expect(openEvent.socket).toBe(socket); + expect(snapshot.type).toBe("snapshot"); + const message = boundedNoOutputMessage(socket); + socketWaits.push(message); + exitEvent = new Promise((resolvePromise) => { + const unsubscribe = onPtyEvent((event) => { + if (event.type !== "exit") return; + unsubscribe(); + resolvePromise(); + }); + unsubscribeExit = unsubscribe; + }); + socket.send(attachMessage); + fakeExited = harness.instances.at(-1)?.exited; + await expect(message).rejects.toThrow(noOutputDiagnostic); + records.push(JSON.stringify({ case: "attach-without-output", snapshot: true, ptyData: false, diagnostic: noOutputDiagnostic })); + } finally { + await finishCase(harness, socket, "attach-without-output", fakeExited, exitEvent, unsubscribeExit, socketWaits); + } +}, 15000); diff --git a/server/websocket-test/fake-herdr-client.test.ts b/server/websocket-test/fake-herdr-client.test.ts new file mode 100644 index 0000000000..61ac583d8e --- /dev/null +++ b/server/websocket-test/fake-herdr-client.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, test } from "bun:test"; +import { + FakeBoundaryError, + getFakeHerdrCounters, + resetFakeHerdrCounters, + sessionSnapshot, + subscribeEvents, + workspaceCreate, + unexpectedRpc, +} from "./fake-herdr-client.ts"; + +describe("fake Herdr boundary", () => { + test("returns deterministic pane and layout snapshot", async () => { + const snapshot = await sessionSnapshot(); + expect(snapshot.panes).toHaveLength(1); + expect(snapshot.panes[0]).toMatchObject({ + pane_id: "fake-pane", + terminal_id: "terminal-1", + workspace_id: "workspace-1", + tab_id: "tab-1", + agent_status: "idle", + }); + expect(snapshot.layouts[0]?.panes[0]?.rect).toEqual({ x: 0, y: 0, width: 120, height: 40 }); + }); + + test("closes subscriptions once and counts closure", () => { + resetFakeHerdrCounters(); + const subscription = subscribeEvents([], { onEvent: () => undefined }); + subscription.close(); + subscription.close(); + expect(getFakeHerdrCounters()).toEqual({ subscriptionCloseCount: 1 }); + }); + + test("rejects workspace creation and unexpected RPC at the fake boundary", async () => { + resetFakeHerdrCounters(); + await expect(workspaceCreate({ label: "must not create" })).rejects.toBeInstanceOf(FakeBoundaryError); + await expect(unexpectedRpc("unrecognized.call")).rejects.toMatchObject({ method: "unrecognized.call" }); + expect(getFakeHerdrCounters()).toEqual({ subscriptionCloseCount: 0 }); + }); +}); diff --git a/server/websocket-test/fake-herdr-client.ts b/server/websocket-test/fake-herdr-client.ts new file mode 100644 index 0000000000..0fedbdc6c0 --- /dev/null +++ b/server/websocket-test/fake-herdr-client.ts @@ -0,0 +1,92 @@ +import type { PaneReadOptions } from "../herdr/client.ts"; +import type { EventFrame, HerdrSubscription, SubscribeHandlers, Subscription } from "../herdr/client.ts"; +import type { AgentManifestInfo, PaneInfo, TabInfo, WorkspaceInfo } from "../../shared/herdr-api.generated.ts"; +import type { SessionSnapshot } from "../../shared/protocol.ts"; + +export class FakeBoundaryError extends Error { + readonly method: string; + + constructor(method: string) { + super(`fake Herdr boundary rejected unconfigured RPC: ${method}`); + this.name = "FakeBoundaryError"; + this.method = method; + } +} + +const rect = { x: 0, y: 0, width: 120, height: 40 } as const; + +export const sessionSnapshotValue: SessionSnapshot = { + agents: [], + focused_pane_id: "fake-pane", + focused_tab_id: "tab-1", + focused_workspace_id: "workspace-1", + layouts: [{ + area: rect, + focused_pane_id: "fake-pane", + panes: [{ focused: true, pane_id: "fake-pane", rect }], + splits: [], + tab_id: "tab-1", + workspace_id: "workspace-1", + zoomed: false, + }], + panes: [{ + agent_status: "idle", + focused: true, + pane_id: "fake-pane", + revision: 1, + tab_id: "tab-1", + terminal_id: "terminal-1", + workspace_id: "workspace-1", + }], + protocol: 22, + tabs: [{ agent_status: "idle", focused: true, label: "Tab 1", number: 1, pane_count: 1, tab_id: "tab-1", workspace_id: "workspace-1" }], + version: "fake-herdr", + workspaces: [{ active_tab_id: "tab-1", agent_status: "idle", focused: true, label: "Workspace 1", number: 1, pane_count: 1, tab_count: 1, workspace_id: "workspace-1" }], +}; + +const manifests: { manifests: AgentManifestInfo[] } = { manifests: [] }; +let subscriptionCloseCount = 0; + +export function resetFakeHerdrCounters(): void { + subscriptionCloseCount = 0; +} + +export function getFakeHerdrCounters(): Readonly<{ subscriptionCloseCount: number }> { + return { subscriptionCloseCount }; +} + +export function herdrSocketPath(): string { return "/fake/herdr.sock"; } +export class HerdrError extends Error { + readonly code: string; + constructor(code: string, message: string) { super(message); this.name = "HerdrError"; this.code = code; } +} + +export async function ping(): Promise<{ version: string; protocol: number }> { return { version: "fake-herdr", protocol: 22 }; } +export async function sessionSnapshot(): Promise { return sessionSnapshotValue; } +export async function agentManifests(): Promise<{ manifests: AgentManifestInfo[] }> { return manifests; } +export interface WorkspaceCreateResult { type: "workspace_created"; workspace: WorkspaceInfo; tab: TabInfo; root_pane: PaneInfo } +export async function workspaceCreate(_options: { cwd?: string; label?: string }): Promise { + throw new FakeBoundaryError("workspace.create"); +} +export async function agentStart(_options: { name: string; kind: string; paneId: string; args?: string[]; timeoutMs?: number }): Promise { + throw new FakeBoundaryError("agent.start"); +} +export async function paneRename(_paneId: string, _label: string | null): Promise { throw new FakeBoundaryError("pane.rename"); } +export async function workspaceRename(_workspaceId: string, _label: string): Promise { throw new FakeBoundaryError("workspace.rename"); } +export async function workspaceMove(_workspaceId: string, _insertIndex: number): Promise { throw new FakeBoundaryError("workspace.move"); } +export async function workspaceClose(_workspaceId: string): Promise { throw new FakeBoundaryError("workspace.close"); } +export async function paneRead(_options: PaneReadOptions): Promise { throw new FakeBoundaryError("pane.read"); } +export async function paneSendText(_paneId: string, _text: string): Promise { throw new FakeBoundaryError("pane.send_text"); } +export async function agentPrompt(_target: string, _text: string): Promise { throw new FakeBoundaryError("agent.prompt"); } +export async function paneSendKeys(_paneId: string, _keys: string[]): Promise { throw new FakeBoundaryError("pane.send_keys"); } +export async function paneClose(_paneId: string): Promise { throw new FakeBoundaryError("pane.close"); } + +export function subscribeEvents(_subscriptions: HerdrSubscription[], handlers: SubscribeHandlers): Subscription { + handlers.onStarted?.(); + let closed = false; + return { close() { if (closed) return; closed = true; subscriptionCloseCount += 1; handlers.onClose?.(); } }; +} + +export async function unexpectedRpc(method: string): Promise { throw new FakeBoundaryError(method); } + +export type { EventFrame }; diff --git a/server/websocket-test/fake-pty-session.test.ts b/server/websocket-test/fake-pty-session.test.ts new file mode 100644 index 0000000000..7529b146e6 --- /dev/null +++ b/server/websocket-test/fake-pty-session.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, test } from "bun:test"; +import { FakePtySession, onPtyEvent, queueInitialOutput, type PtySessionOptions } from "./fake-pty-session"; + +function createSession(onData: (data: string) => void = () => {}, onExit: (code: number | null) => void = () => {}): FakePtySession { + const options: PtySessionOptions = { + command: "ignored", + args: [], + cols: 80, + rows: 24, + onData, + onExit, + }; + return new FakePtySession(options); +} + +describe("FakePtySession", () => { + test("delivers queued Unicode seed synchronously and consumes it once", () => { + const received: string[] = []; + queueInitialOutput("한글", "🙂"); + createSession((data) => received.push(data)); + const next: string[] = []; + createSession((data) => next.push(data)); + expect(received).toEqual(["한글", "🙂"]); + expect(next).toEqual([]); + }); + + test("emits live data and queued resume output only on resume", () => { + const received: string[] = []; + const session = createSession((data) => received.push(data)); + session.emitData("live"); + session.queueOutputOnResume("sentinel"); + expect(received).toEqual(["live"]); + session.resume(); + expect(received).toEqual(["live"]); + session.pause(); + session.resume(); + expect(received).toEqual(["live", "sentinel"]); + }); + + test("delivers events to a listener armed before any instance exists", () => { + const events: string[] = []; + const unsubscribe = onPtyEvent((event) => events.push(event.type)); + const session = createSession(); + session.pause(); + session.resume(); + session.kill(); + unsubscribe(); + expect(events).toEqual(["pause", "resume", "kill", "exit"]); + }); + + test("records writes and resizes deterministically", () => { + const session = createSession(); + session.write("input"); + session.resize(100, 40); + expect(session.writes).toEqual(["input"]); + expect(session.resizes).toEqual([{ cols: 100, rows: 40 }]); + }); + + test("kill is idempotent, invokes onExit once, and resolves exited", async () => { + let exitCalls = 0; + const session = createSession(() => {}, () => exitCalls++); + session.kill(); + session.kill(); + await session.exited; + expect(session.killCount).toBe(1); + expect(exitCalls).toBe(1); + }); +}); diff --git a/server/websocket-test/fake-pty-session.ts b/server/websocket-test/fake-pty-session.ts new file mode 100644 index 0000000000..93b339d638 --- /dev/null +++ b/server/websocket-test/fake-pty-session.ts @@ -0,0 +1,101 @@ +export interface PtySessionOptions { + command: string; + args: string[]; + cols: number; + rows: number; + onData: (data: string) => void; + onExit: (code: number | null) => void; + env?: Record; +} + +export type PtyEvent = + | { readonly type: "pause" | "resume" | "kill"; readonly sessionId: number; readonly paneId: "fake-pane" } + | { readonly type: "exit"; readonly sessionId: number; readonly paneId: "fake-pane"; readonly code: number | null }; +export type PtyEventType = PtyEvent["type"]; + +const listeners = new Set<(event: PtyEvent) => void>(); +let nextSessionId = 1; +let queuedInitialOutput: string[] = []; + +export function queueInitialOutput(...chunks: string[]): void { + queuedInitialOutput.push(...chunks); +} + +export function onPtyEvent(listener: (event: PtyEvent) => void): () => void { + listeners.add(listener); + return () => listeners.delete(listener); +} + +export class FakePtySession { + readonly sessionId = nextSessionId++; + readonly writes: string[] = []; + readonly resizes: { readonly cols: number; readonly rows: number }[] = []; + pauseCount = 0; + resumeCount = 0; + killCount = 0; + private paused = false; + private closed = false; + private resumeOutput: string | undefined; + private exitCallback: ((code: number | null) => void) | undefined; + private readonly exitedResolver: { resolve: () => void }; + readonly exited: Promise; + + constructor(private readonly options: PtySessionOptions) { + const resolver: { resolve: () => void } = { resolve: () => {} }; + this.exitedResolver = resolver; + this.exited = new Promise((resolve) => { + resolver.resolve = resolve; + }); + this.exitCallback = options.onExit; + for (const chunk of queuedInitialOutput) options.onData(chunk); + queuedInitialOutput = []; + } + + emitData(data: string): void { + if (!this.closed) this.options.onData(data); + } + + queueOutputOnResume(data: string): void { + this.resumeOutput = data; + } + + write(data: string): void { + if (!this.closed) this.writes.push(data); + } + + resize(cols: number, rows: number): void { + if (!this.closed) this.resizes.push({ cols, rows }); + } + + pause(): void { + if (this.paused || this.closed) return; + this.paused = true; + this.pauseCount++; + this.emitEvent({ type: "pause", sessionId: this.sessionId, paneId: "fake-pane" }); + } + + resume(): void { + if (!this.paused || this.closed) return; + this.paused = false; + this.resumeCount++; + this.emitEvent({ type: "resume", sessionId: this.sessionId, paneId: "fake-pane" }); + const output = this.resumeOutput; + this.resumeOutput = undefined; + if (output !== undefined) this.options.onData(output); + } + + kill(): void { + if (this.closed) return; + this.closed = true; + this.killCount++; + this.emitEvent({ type: "kill", sessionId: this.sessionId, paneId: "fake-pane" }); + this.exitCallback?.(null); + this.exitCallback = undefined; + this.emitEvent({ type: "exit", sessionId: this.sessionId, paneId: "fake-pane", code: null }); + this.exitedResolver.resolve(); + } + + private emitEvent(event: PtyEvent): void { + for (const listener of listeners) listener(event); + } +} From fd8d0b987a7b2da8ec5aaf7719439a456b447bd2 Mon Sep 17 00:00:00 2001 From: pjhsion Date: Sat, 10 Oct 2026 11:31:53 +0900 Subject: [PATCH 2/4] Add live WebSocket scenarios and mutation controls Port S1-S10 protocol scenarios onto the real-server harness, add the mutation overlay generator with per-run provenance, and add fresh-process mutation controls with C0 cleanup receipts. Prearm socket close listeners and stop the server before awaiting client close in harness cleanup. --- package.json | 2 +- .../ack-replay.integration.test.ts | 159 ++++++++++ .../ack-state.integration.test.ts | 192 +++++++++++++ .../backpressure-stall.integration.test.ts | 268 +++++++++++++++++ .../create-server-harness.test.ts | 70 +++++ .../websocket-test/create-server-harness.ts | 23 +- .../mutation-controls.integration.test.ts | 126 ++++++++ .../mutation-probe.integration.test.ts | 260 +++++++++++++++++ server/websocket-test/mutation-source.test.ts | 99 +++++++ server/websocket-test/mutation-source.ts | 184 ++++++++++++ .../protocol-guards.integration.test.ts | 165 +++++++++++ .../reattach-reconnect.integration.test.ts | 272 ++++++++++++++++++ 12 files changed, 1816 insertions(+), 4 deletions(-) create mode 100644 server/websocket-test/ack-replay.integration.test.ts create mode 100644 server/websocket-test/ack-state.integration.test.ts create mode 100644 server/websocket-test/backpressure-stall.integration.test.ts create mode 100644 server/websocket-test/mutation-controls.integration.test.ts create mode 100644 server/websocket-test/mutation-probe.integration.test.ts create mode 100644 server/websocket-test/mutation-source.test.ts create mode 100644 server/websocket-test/mutation-source.ts create mode 100644 server/websocket-test/protocol-guards.integration.test.ts create mode 100644 server/websocket-test/reattach-reconnect.integration.test.ts diff --git a/package.json b/package.json index 715ea10451..36a6538d12 100644 --- a/package.json +++ b/package.json @@ -16,7 +16,7 @@ "start": "bun run build && bun run server/managed.ts", "typecheck": "tsc --noEmit", "test": "bun test", - "test:websocket": "bun --config=bunfig.websocket.toml test --isolate server/output-window.test.ts server/websocket-test/fake-herdr-client.test.ts server/websocket-test/fake-pty-session.test.ts server/websocket-test/create-server-harness.test.ts server/websocket-test/create-server.smoke.integration.test.ts", + "test:websocket": "bun --config=bunfig.websocket.toml test --isolate server/output-window.test.ts server/websocket-test/fake-herdr-client.test.ts server/websocket-test/fake-pty-session.test.ts server/websocket-test/create-server-harness.test.ts server/websocket-test/create-server.smoke.integration.test.ts server/websocket-test/ack-replay.integration.test.ts server/websocket-test/ack-state.integration.test.ts server/websocket-test/reattach-reconnect.integration.test.ts server/websocket-test/backpressure-stall.integration.test.ts server/websocket-test/protocol-guards.integration.test.ts", "test:websocket:mutations": "bun --config=bunfig.websocket.toml test --isolate server/websocket-test/mutation-source.test.ts server/websocket-test/mutation-controls.integration.test.ts", "test:ui": "bun run build && bun scripts/ui-regression.ts", "generate:types": "bun run scripts/generate-protocol-types.ts", diff --git a/server/websocket-test/ack-replay.integration.test.ts b/server/websocket-test/ack-replay.integration.test.ts new file mode 100644 index 0000000000..2e885dad6f --- /dev/null +++ b/server/websocket-test/ack-replay.integration.test.ts @@ -0,0 +1,159 @@ +import { access, mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { expect, test } from "bun:test"; +import { createServerHarness, websocketRunRoot } from "./create-server-harness.ts"; +import { onPtyEvent } from "./fake-pty-session.ts"; +import { OUTPUT_HARD_BYTES, OUTPUT_LOW_BYTES } from "../output-window.ts"; + +const evidencePath = join(websocketRunRoot, "evidence", "ack-replay.jsonl"); +const cleanupPath = join(websocketRunRoot, "evidence", "ack-replay-cleanup.json"); + +interface WsFrame { readonly type: string; readonly [key: string]: unknown } + +function message(socket: WebSocket, diagnostic: string): Promise { + return new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const cleanup = () => { + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + socket.removeEventListener("close", onClose); + socket.removeEventListener("error", onError); + }; + const onMessage = (event: MessageEvent) => { + cleanup(); + const parsed: unknown = JSON.parse(String(event.data)); + if (typeof parsed !== "object" || parsed === null || !("type" in parsed) || typeof parsed.type !== "string") { + rejectPromise(new Error("WebSocket frame has no string type")); + return; + } + resolvePromise(parsed as WsFrame); + }; + const onClose = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket closed`)); }; + const onError = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket errored`)); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error(diagnostic)); }, 5000); + socket.addEventListener("message", onMessage); + socket.addEventListener("close", onClose); + socket.addEventListener("error", onError); + }); +} + +test("real replay preserves UTF-8 offsets and ACK resumes output", async () => { + const harness = await createServerHarness(); + let socket: WebSocket | undefined; + let fakeExited: Promise | undefined; + let exitEvent: Promise | undefined; + let unsubscribeExit: (() => void) | undefined; + const waits: Promise[] = []; + let outcome: Record = { case: "ack-replay", passed: false }; + try { + await mkdir(join(websocketRunRoot, "evidence"), { recursive: true }); + harness.queueInitialPtyOutput("\x1b[?1049h\x1b[?1000h" + "한🙂".repeat(40000)); + socket = harness.openSocket(); + const openWait = harness.prearmSocketEvent("open"); + const snapshotWait = message(socket, "timed out waiting for initial snapshot"); + waits.push(openWait, snapshotWait); + const [opened, snapshot] = await Promise.all([openWait, snapshotWait]); + expect(opened.socket).toBe(socket); + expect(snapshot.type).toBe("snapshot"); + + const pauseWait = harness.waitForPtyEvent("fake-pane", "pause"); + const replayWait = message(socket, "timed out waiting for replay frame"); + waits.push(pauseWait.promise, replayWait); + exitEvent = new Promise((resolvePromise) => { + const unsubscribe = onPtyEvent((event) => { + if (event.type !== "exit") return; + unsubscribe(); + resolvePromise(); + }); + unsubscribeExit = unsubscribe; + }); + socket.send('{"type":"attach","pane_id":"fake-pane","cols":100,"rows":30,"flow_control":"ack"}'); + fakeExited = harness.instances.at(-1)?.exited; + const [pause, frame] = await Promise.all([pauseWait.promise, replayWait]); + expect(pause.type).toBe("pause"); + expect(frame.type).toBe("pty-data"); + const data = String(frame.data); + const flow = frame.flow; + if (typeof flow !== "object" || flow === null || !("stream_id" in flow) || !("offset" in flow)) throw new Error("replay frame omitted flow metadata"); + const offset = Number(flow.offset); + expect(String(flow.stream_id).length).toBeGreaterThan(0); + expect(offset).toBe(Buffer.byteLength(data)); + expect(data).toContain("\x1b[?1049h"); + expect(data).toContain("\x1b[?1000h"); + expect(offset).toBeLessThan(OUTPUT_HARD_BYTES); + + harness.queuePtyOutputOnResume("fake-pane", "ACK_RESUMED"); + const nextDataWait = message(socket, "timed out waiting for resumed sentinel"); + const resumeWait = harness.waitForPtyEvent("fake-pane", "resume"); + waits.push(nextDataWait, resumeWait.promise); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: "fake-pane", stream_id: flow.stream_id, offset: offset - OUTPUT_LOW_BYTES })); + const [resume, nextFrame] = await Promise.all([resumeWait.promise, nextDataWait]); + expect(resume.type).toBe("resume"); + expect(nextFrame.type).toBe("pty-data"); + const sentinelData = String(nextFrame.data); + expect(sentinelData).toContain("ACK_RESUMED"); + const nextFlow = nextFrame.flow; + if (typeof nextFlow !== "object" || nextFlow === null || !("offset" in nextFlow)) throw new Error("sentinel frame omitted flow offset"); + const resumedBytes = Number(nextFlow.offset); + expect(resumedBytes).toBe(offset + Buffer.byteLength(sentinelData)); + outcome = { case: "ack-replay", passed: true, replayBytes: offset, replayContainsModes: true, hardBytes: OUTPUT_HARD_BYTES, ackOffset: offset - OUTPUT_LOW_BYTES, resumedBytes, sentinel: "ACK_RESUMED" }; + } finally { + const stateDir = harness.stateDir; + const sessionExit = fakeExited ?? harness.instances.at(-1)?.exited; + let cleanupResult: Awaited> | undefined; + let fakeExitedResolved = false; + try { + cleanupResult = await harness.cleanup(); + await Promise.allSettled(waits); + if (sessionExit) { + await sessionExit; + fakeExitedResolved = true; + if (exitEvent) await exitEvent; + } + unsubscribeExit?.(); + let stateDirAbsent = false; + try { + await access(stateDir); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") stateDirAbsent = true; + else throw error; + } + const stoppedPortHandshakeFailed = await new Promise((resolvePromise, rejectPromise) => { + const stopped = new WebSocket(new URL("/ws", harness.baseUrl)); + const timer = setTimeout(() => { stopped.close(); rejectPromise(new Error("stopped-port handshake did not settle")); }, 1000); + const settle = (failed: boolean) => { + clearTimeout(timer); + stopped.removeEventListener("error", onError); + stopped.removeEventListener("open", onOpen); + if (failed) resolvePromise(true); + else { stopped.close(); resolvePromise(false); } + }; + const onError = () => settle(true); + const onOpen = () => settle(false); + stopped.addEventListener("error", onError); + stopped.addEventListener("open", onOpen); + }); + const receipt = { + case: "ack-replay", + socketsClosed: socket?.readyState === WebSocket.CLOSED ? 1 : 0, + fakeExitedResolved, + subscriptionCloseDelta: cleanupResult.subscriptionCloseDelta, + subscriptionsClosed: cleanupResult.subscriptionCloseDelta > 0, + stateDirAbsent, + stoppedPortHandshakeFailed, + }; + await writeFile(evidencePath, `${JSON.stringify(outcome)}\n`, "utf8"); + await writeFile(cleanupPath, `${JSON.stringify({ cases: [receipt] }, null, 2)}\n`, "utf8"); + expect(receipt.socketsClosed).toBe(socket ? 1 : 0); + expect(receipt.fakeExitedResolved).toBe(true); + expect(receipt.subscriptionsClosed).toBe(true); + expect(receipt.stateDirAbsent).toBe(true); + expect(receipt.stoppedPortHandshakeFailed).toBe(true); + } finally { + if (cleanupResult === undefined) { + await harness.cleanup(); + unsubscribeExit?.(); + } + } + } +}, 15000); diff --git a/server/websocket-test/ack-state.integration.test.ts b/server/websocket-test/ack-state.integration.test.ts new file mode 100644 index 0000000000..df1d37a18d --- /dev/null +++ b/server/websocket-test/ack-state.integration.test.ts @@ -0,0 +1,192 @@ +import { expect, test } from "bun:test"; +import { mkdir, stat, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { createServerHarness, evidenceDirectory } from "./create-server-harness.ts"; + +const records: Record[] = []; +const cleanupReceipts: Record[] = []; + +async function saveEvidence(): Promise { + const directory = evidenceDirectory(); + await mkdir(directory, { recursive: true }); + await writeFile(resolve(directory, "ack-state.jsonl"), records.map((record) => JSON.stringify(record)).join("\n") + "\n"); + await writeFile(resolve(directory, "ack-state-cleanup.json"), JSON.stringify(cleanupReceipts, null, 2) + "\n"); +} + +function message(socket: WebSocket): Promise> { + return new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const cleanup = () => { + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + socket.removeEventListener("close", onClose); + socket.removeEventListener("error", onError); + }; + const onMessage = (event: MessageEvent) => { + cleanup(); + try { + const parsed: unknown = JSON.parse(String(event.data)); + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) throw new Error("websocket frame must be an object"); + resolvePromise(parsed as Record); + } catch (error) { + rejectPromise(error); + } + }; + const onClose = () => { cleanup(); rejectPromise(new Error("websocket closed before a frame arrived")); }; + const onError = () => { cleanup(); rejectPromise(new Error("websocket errored before a frame arrived")); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error("timed out waiting for websocket message")); }, 5000); + socket.addEventListener("message", onMessage, { once: true }); + socket.addEventListener("close", onClose, { once: true }); + socket.addEventListener("error", onError, { once: true }); + }); +} + +function errors(socket: WebSocket, count: number): Promise[]> { + return new Promise((resolvePromise, rejectPromise) => { + const frames: Record[] = []; + let timer: ReturnType; + const cleanup = () => { + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + socket.removeEventListener("close", onClose); + socket.removeEventListener("error", onError); + }; + const onMessage = (event: MessageEvent) => { + const frame: unknown = JSON.parse(String(event.data)); + if (typeof frame !== "object" || frame === null || Array.isArray(frame) || !("type" in frame) || frame.type !== "error") return; + frames.push(frame as Record); + if (frames.length !== count) return; + cleanup(); + resolvePromise(frames); + }; + const onClose = () => { cleanup(); rejectPromise(new Error("websocket closed before error frames arrived")); }; + const onError = () => { cleanup(); rejectPromise(new Error("websocket errored before error frames arrived")); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error("timed out waiting for error frames")); }, 5000); + socket.addEventListener("message", onMessage); + socket.addEventListener("close", onClose); + socket.addEventListener("error", onError); + }); +} + +async function startSocket(harness: Awaited>): Promise { + const socket = harness.openSocket(); + const opened = harness.prearmSocketEvent("open"); + const snapshot = message(socket); + const [, frame] = await Promise.all([opened, snapshot]); + expect(frame["type"]).toBe("snapshot"); + return socket; +} + +async function clean(harness: Awaited>, socket: WebSocket): Promise { + const instance = harness.instances[0]; + const stateDir = harness.stateDir; + const port = Number(new URL(harness.baseUrl).port); + const closing = harness.prearmSocketEvent("close"); + const result = await harness.cleanup(); + await instance?.exited; + const stoppedPortHandshakeFailed = await fetch(`http://127.0.0.1:${port}/`, { signal: AbortSignal.timeout(1000) }).then(() => false, () => true); + const stateDirAbsent = await stat(stateDir).then(() => false, (error: NodeJS.ErrnoException) => error.code === "ENOENT"); + cleanupReceipts.push({ + socketClosed: (await closing).socket === socket, + fakeExited: instance !== undefined && instance.killCount === 1, + subscriptionCloseDelta: result.subscriptionCloseDelta, + stateDirAbsent, + stoppedPortHandshakeFailed, + }); + expect(stateDirAbsent).toBe(true); + expect(stoppedPortHandshakeFailed).toBe(true); + expect(instance).toBeDefined(); + if (!instance) throw new Error("attached pane did not create a fake PTY session"); + expect(instance.killCount).toBe(1); + expect(result.subscriptionCloseDelta).toBeGreaterThanOrEqual(1); + await saveEvidence(); +} + +test("S2 cumulative and duplicate ACKs preserve output credit", async () => { + const harness = await createServerHarness(); + let socket: WebSocket | undefined; + try { + socket = await startSocket(harness); + const roleAck = message(socket); + socket.send(JSON.stringify({ type: "role", mode: "observe" })); + expect((await roleAck)["type"]).toBe("role-ack"); + const readyFrame = message(socket); + socket.send(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 120, rows: 40, flow_control: "ack" })); + expect((await readyFrame)["type"]).toBe("pane-geometry"); + const initialPause = harness.waitForPtyEvent(harness.paneId, "pause"); + const initialFrame = message(socket); + harness.emitPtyData(harness.paneId, "A".repeat(300000)); + const [first, pauseEvent] = await Promise.all([initialFrame, initialPause.promise]); + const instance = harness.instances.at(-1); + expect(first["type"]).toBe("pty-data"); + const firstFlow = first["flow"]; + if (typeof firstFlow !== "object" || firstFlow === null || !("offset" in firstFlow) || !("stream_id" in firstFlow)) throw new Error("first output frame omitted flow data"); + expect(firstFlow.offset).toBe(300000); + expect(first["data"]).toBe("A".repeat(300000)); + expect(pauseEvent.type).toBe("pause"); + const resume = harness.waitForPtyEvent(harness.paneId, "resume"); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: firstFlow.stream_id, offset: 300000 })); + const resumeEvent = await resume.promise.catch((error) => { throw new Error(`resume after cumulative ACK: ${String(error)}`); }); + expect(resumeEvent.type).toBe("resume"); + const nextFrame = message(socket); + const afterFullAckResumeCount = instance?.resumeCount; + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: firstFlow.stream_id, offset: 300000 })); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: firstFlow.stream_id, offset: 100000 })); + const messageWaiter = message(socket); + harness.emitPtyData(harness.paneId, "B".repeat(63000)); + const second = await nextFrame; + const duplicateAndRegressiveResult = await messageWaiter; + expect(duplicateAndRegressiveResult["type"]).not.toBe("error"); + expect(second["type"]).toBe("pty-data"); + const secondFlow = second["flow"]; + if (typeof secondFlow !== "object" || secondFlow === null || !("offset" in secondFlow)) throw new Error("second output frame omitted offset"); + expect(secondFlow.offset).toBe(363000); + expect(second["data"]).toBe("B".repeat(63000)); + expect(instance?.pauseCount).toBe(1); + expect(instance?.resumeCount).toBe(afterFullAckResumeCount); + expect(socket.readyState).toBe(WebSocket.OPEN); + records.push({ scenario: "S2", actions: ["attach ack", "emit A×300000", "ack 300000", "ack 100000", "emit B×63000"], frames: [first, second], pauseCount: instance?.pauseCount, resumeCount: instance?.resumeCount, socketOpen: socket.readyState === WebSocket.OPEN }); + await clean(harness, socket); + } finally { + if (socket && harness.instances[0]?.killCount === 0) { + const close = harness.prearmSocketEvent("close"); + await harness.cleanup(); + await close; + await harness.instances[0]?.exited; + } + } +}); + +test("S3 future and malformed ACK offsets return invalid_ack", async () => { + const harness = await createServerHarness(); + let socket: WebSocket | undefined; + try { + harness.queueInitialPtyOutput("ack-probe"); + socket = await startSocket(harness); + const initialFrame = message(socket); + socket.send(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 120, rows: 40, flow_control: "ack" })); + const data = await initialFrame; + expect(data["data"]).toBe("ack-probe"); + const instance = harness.instances.at(-1); + if (!instance) throw new Error("attach did not create a fake PTY session"); + const errorFrames = errors(socket, 2); + const flow = data["flow"]; + if (typeof flow !== "object" || flow === null || !("stream_id" in flow)) throw new Error("output frame omitted stream id"); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: flow.stream_id, offset: Number.MAX_SAFE_INTEGER })); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: flow.stream_id, offset: "not-a-number" })); + const errorFramesResult = await errorFrames; + expect(errorFramesResult.map((frame) => frame["code"])).toEqual(["invalid_ack", "invalid_ack"]); + expect(socket.readyState).toBe(WebSocket.OPEN); + expect(instance.resumeCount).toBe(0); + expect(instance.pauseCount).toBe(0); + records.push({ scenario: "S3", actions: ["attach ack with ack-probe", "ack Number.MAX_SAFE_INTEGER", "ack string not-a-number"], frames: [data, ...errorFramesResult], pauseCount: instance.pauseCount, resumeCount: instance.resumeCount, socketOpen: socket.readyState === WebSocket.OPEN }); + await clean(harness, socket); + } finally { + if (socket && harness.instances[0]?.killCount === 0) { + const close = harness.prearmSocketEvent("close"); + await harness.cleanup(); + await close; + await harness.instances[0]?.exited; + } + } +}); diff --git a/server/websocket-test/backpressure-stall.integration.test.ts b/server/websocket-test/backpressure-stall.integration.test.ts new file mode 100644 index 0000000000..5ca0f6046a --- /dev/null +++ b/server/websocket-test/backpressure-stall.integration.test.ts @@ -0,0 +1,268 @@ +import { access, mkdir, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { expect, test } from "bun:test"; +import { OUTPUT_HARD_BYTES, OUTPUT_HIGH_BYTES, OUTPUT_LOW_BYTES, OUTPUT_STALL_MS } from "../output-window.ts"; +import { createServerHarness, evidenceDirectory } from "./create-server-harness.ts"; + +const records: Record[] = []; +const receipts: Record[] = []; +const closeCodeStalled = 4008; +const timeoutMs = OUTPUT_STALL_MS + 10000; +type Frame = Record; + +function waitMessage(socket: WebSocket, predicate: (frame: Frame) => boolean, diagnostic: string, timeout = 10000): Promise { + return new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const cleanup = () => { clearTimeout(timer); socket.removeEventListener("message", onMessage); socket.removeEventListener("close", onClose); socket.removeEventListener("error", onError); }; + const onMessage = (event: MessageEvent) => { + try { + const parsed: unknown = JSON.parse(String(event.data)); + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) throw new Error("websocket frame must be an object"); + const frame = parsed as Frame; + if (!predicate(frame)) return; + cleanup(); + resolvePromise(frame); + } catch (error) { cleanup(); rejectPromise(error); } + }; + const onClose = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket closed`)); }; + const onError = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket errored`)); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error(diagnostic)); }, timeout); + socket.addEventListener("message", onMessage); + socket.addEventListener("close", onClose); + socket.addEventListener("error", onError); + }); +} + +function waitClose(socket: WebSocket, timeout = timeoutMs): Promise { + return new Promise((resolvePromise, rejectPromise) => { + const onClose = (event: CloseEvent) => { clearTimeout(timer); socket.removeEventListener("close", onClose); resolvePromise(event); }; + const timer = setTimeout(() => { socket.removeEventListener("close", onClose); rejectPromise(new Error("timed out waiting for close")); }, timeout); + socket.addEventListener("close", onClose, { once: true }); + }); +} + +function waitOpen(harness: Awaited>, socket: WebSocket): Promise { + const opened = harness.prearmSocketEvent("open"); + const snapshot = waitMessage(socket, (frame) => frame["type"] === "snapshot", "snapshot missing"); + return Promise.all([opened, snapshot]).then(([event, frame]) => { + expect(event.socket).toBe(socket); + expect(frame["type"]).toBe("snapshot"); + }); +} + +async function finish( + name: string, + harness: Awaited>, + sockets: WebSocket[], + pending: Promise[], + ptyExit?: Promise, +): Promise { + const stateDir = harness.stateDir; + const instances = [...harness.instances]; + const closes = sockets.map((socket) => new Promise((resolvePromise) => { + if (socket.readyState === WebSocket.CLOSED) { resolvePromise(); return; } + socket.addEventListener("close", () => resolvePromise(), { once: true }); + socket.close(); + })); + const cleanup = await harness.cleanup(); + const outcomes = await Promise.allSettled(pending); + await Promise.all(closes); + await Promise.all(instances.map((instance) => instance.exited)); + if (ptyExit) await ptyExit; + const stateDirAbsent = await access(stateDir).then(() => false, (error: NodeJS.ErrnoException) => error.code === "ENOENT"); + const stoppedPortHandshakeFailed = await new Promise((resolvePromise, rejectPromise) => { + const socket = new WebSocket(`${harness.baseUrl}/ws`); + let timer: ReturnType; + const settle = (failed: boolean) => { + clearTimeout(timer); + socket.removeEventListener("error", onError); + socket.removeEventListener("open", onOpen); + if (!failed) socket.close(); + resolvePromise(failed); + }; + const onError = () => settle(true); + const onOpen = () => settle(false); + timer = setTimeout(() => { socket.close(); rejectPromise(new Error("stopped-port handshake did not settle")); }, timeoutMs); + socket.addEventListener("error", onError); + socket.addEventListener("open", onOpen); + }); + const receipt = { + scenario: name, + socketsOpened: sockets.length, + socketsClosed: sockets.filter((socket) => socket.readyState === WebSocket.CLOSED).length, + fakeExited: instances.length > 0 && instances.every((instance) => instance.killCount === 1), + subscriptionsClosed: cleanup.subscriptionCloseDelta > 0, + subscriptionCloseDelta: cleanup.subscriptionCloseDelta, + stateDirAbsent, + stoppedPortHandshakeFailed, + pendingWaitersSettled: outcomes.every((outcome) => outcome.status === "fulfilled"), + }; + receipts.push(receipt); + expect(receipt.socketsClosed).toBe(receipt.socketsOpened); + expect(receipt.fakeExited).toBe(true); + expect(receipt.subscriptionsClosed).toBe(true); + expect(receipt.stateDirAbsent).toBe(true); + expect(receipt.stoppedPortHandshakeFailed).toBe(true); + expect(receipt.pendingWaitersSettled).toBe(true); + const directory = evidenceDirectory(); + await mkdir(directory, { recursive: true }); + await writeFile(resolve(directory, "backpressure-stall.jsonl"), records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8"); + await writeFile(resolve(directory, "backpressure-stall-cleanup.json"), `${JSON.stringify({ cases: receipts }, null, 2)}\n`, "utf8"); +} + +const ackAttach = (pane: string) => JSON.stringify({ type: "attach", pane_id: pane, cols: 100, rows: 30, flow_control: "ack" }); + +test("S7 ACK credit releases only at exact low watermark", async () => { + const harness = await createServerHarness(); + const pending: Promise[] = []; + const socket = harness.openSocket(); + let exited: Promise | undefined; + try { + const open = waitOpen(harness, socket); pending.push(open); await open; + harness.queueInitialPtyOutput("S7-START"); + const initial = waitMessage(socket, (frame) => frame["type"] === "pty-data", "initial output missing"); + const ready = waitMessage(socket, (frame) => frame["type"] === "pane-geometry" || frame["type"] === "pty-data", "attach response missing"); + socket.send(ackAttach(harness.paneId)); + const attachReply = await ready; + const initialFrame = attachReply["type"] === "pty-data" ? attachReply : await initial; + const highFrameWait = waitMessage(socket, (frame) => frame["type"] === "pty-data", "high watermark output missing"); + const pause = harness.waitForPtyEvent(harness.paneId, "pause"); pending.push(pause.promise); + const pty = harness.instances.at(-1); + if (!pty) throw new Error("attach did not create a fake PTY"); + exited = pty.exited; + const body = "H".repeat(OUTPUT_HIGH_BYTES); + harness.emitPtyData(harness.paneId, body); + const high = await highFrameWait; + await pause.promise; + expect(initialFrame["data"]).toBe("S7-START"); + expect(high["data"]).toBe(body); + const streamId = typeof high["flow"] === "object" && high["flow"] !== null && "stream_id" in high["flow"] ? high["flow"].stream_id : undefined; + const sentOffset = typeof high["flow"] === "object" && high["flow"] !== null && "offset" in high["flow"] ? high["flow"].offset : undefined; + if (typeof streamId !== "string" || typeof sentOffset !== "number") throw new Error("high watermark frame omitted flow fields"); + expect(sentOffset).toBe(new TextEncoder().encode(`S7-START${body}`).byteLength); + const before = pty.resumeCount; + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: streamId, offset: sentOffset - OUTPUT_LOW_BYTES - 1 })); + expect(pty.resumeCount).toBe(before); + const invalid = waitMessage(socket, (frame) => frame["type"] === "error", "future ACK error missing"); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: streamId, offset: sentOffset + 1 })); + expect((await invalid)["code"]).toBe("invalid_ack"); + expect(pty.resumeCount).toBe(before); + const resume = harness.waitForPtyEvent(harness.paneId, "resume"); pending.push(resume.promise); + const sentinelWait = waitMessage(socket, (frame) => frame["type"] === "pty-data" && frame["data"] === "FLOW_RESUMED", "resume sentinel missing"); + harness.queuePtyOutputOnResume(harness.paneId, "FLOW_RESUMED"); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: streamId, offset: sentOffset - OUTPUT_LOW_BYTES })); + await resume.promise; + const sentinel = await sentinelWait; + expect(pty.resumeCount).toBe(before + 1); + const sentinelFlow = sentinel["flow"]; + if (typeof sentinelFlow !== "object" || sentinelFlow === null || !("offset" in sentinelFlow)) throw new Error("resume sentinel omitted its offset"); + expect(sentinelFlow.offset).toBe(sentOffset + new TextEncoder().encode("FLOW_RESUMED").byteLength); + records.push({ scenario: "S7", sentOffset, pendingAboveLow: OUTPUT_LOW_BYTES + 1, invalidAck: "invalid_ack", pendingAtResume: OUTPUT_LOW_BYTES, resumed: true, sentinelOffset: sentinelFlow.offset }); + } finally { await finish("S7", harness, [socket], pending, exited); } +}); + +test("S8 stalled observer is evicted while operator receives live output", async () => { + const harness = await createServerHarness(); + const pending: Promise[] = []; + const operator = harness.openSocket(); + let observer: WebSocket | undefined; + let exited: Promise | undefined; + try { + harness.queueInitialPtyOutput("S8-START"); + const operatorOpen = waitOpen(harness, operator); pending.push(operatorOpen); await operatorOpen; + const operatorInitial = waitMessage(operator, (frame) => frame["type"] === "pty-data", "operator initial output missing"); + const operatorReady = waitMessage(operator, (frame) => frame["type"] === "pane-geometry" || frame["type"] === "pty-data", "operator attach missing"); + operator.send(ackAttach(harness.paneId)); await operatorReady; + const operatorSeed = await operatorInitial; + observer = harness.openSocket(); + const observerOpen = waitOpen(harness, observer); pending.push(observerOpen); await observerOpen; + const role = waitMessage(observer, (frame) => frame["type"] === "role-ack", "observer role ACK missing"); + observer.send(JSON.stringify({ type: "role", mode: "observe" })); await role; + const observerInitial = waitMessage(observer, (frame) => frame["type"] === "pty-data", "observer initial output missing"); + const observerReady = waitMessage(observer, (frame) => frame["type"] === "pane-geometry" || frame["type"] === "pty-data", "observer attach missing"); + observer.send(ackAttach(harness.paneId)); const observerReply = await observerReady; + const observerClosed = new Promise((resolvePromise, rejectPromise) => { + const onClose = (event: CloseEvent) => { clearTimeout(timer); observer?.removeEventListener("close", onClose); resolvePromise(event); }; + const timer = setTimeout(() => { observer?.removeEventListener("close", onClose); rejectPromise(new Error("timed out waiting for observer close")); }, timeoutMs); + observer?.addEventListener("close", onClose, { once: true }); + }); + pending.push(observerClosed); + const observerBytes = { value: 0 }; + const observerData = (event: MessageEvent) => { + const parsed: unknown = JSON.parse(String(event.data)); + if (typeof parsed !== "object" || parsed === null || !("type" in parsed) || parsed.type !== "pty-data" || !("data" in parsed)) return; + observerBytes.value += new TextEncoder().encode(String(parsed.data)).byteLength; + }; + observer.addEventListener("message", observerData); + if (observerReply["type"] === "pty-data" && "data" in observerReply) observerBytes.value += new TextEncoder().encode(String(observerReply["data"])).byteLength; + else { + const seed = await observerInitial; + observerBytes.value += new TextEncoder().encode(String(seed["data"])).byteLength; + } + const alive = waitMessage(operator, (frame) => frame["type"] === "pty-data" && frame["data"] === "FLOW_ALIVE", "FLOW_ALIVE missing", 15000); + const operatorOutput = waitMessage(operator, (frame) => frame["type"] === "pty-data" && frame["data"] === "O".repeat(OUTPUT_HIGH_BYTES), "operator output missing"); + const pty = harness.instances.at(-1); + if (!pty) throw new Error("operator attach did not create a fake PTY"); + exited = pty.exited; + const resume = harness.waitForPtyEvent(harness.paneId, "resume"); pending.push(resume.promise); + harness.queuePtyOutputOnResume(harness.paneId, "FLOW_ALIVE"); + const pause = harness.waitForPtyEvent(harness.paneId, "pause"); pending.push(pause.promise); + harness.emitPtyData(harness.paneId, "O".repeat(OUTPUT_HIGH_BYTES)); + await pause.promise; + const operatorFrame = await operatorOutput; + expect(operatorFrame["data"]).toBe("O".repeat(OUTPUT_HIGH_BYTES)); + const operatorFlow = operatorFrame["flow"]; + if (typeof operatorFlow !== "object" || operatorFlow === null || !("stream_id" in operatorFlow) || !("offset" in operatorFlow)) throw new Error("operator frame omitted flow state"); + operator.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: operatorFlow.stream_id, offset: operatorFlow.offset })); + const close = await observerClosed; + expect(close.code).toBe(closeCodeStalled); + expect(observerBytes.value).toBeLessThanOrEqual(OUTPUT_HARD_BYTES); + expect(operator.readyState).toBe(WebSocket.OPEN); + await resume.promise; + const live = await alive; + expect(live["data"]).toBe("FLOW_ALIVE"); + expect(operator.readyState).toBe(WebSocket.OPEN); + observer.removeEventListener("message", observerData); + records.push({ scenario: "S8", observerCloseCode: close.code, observerBytes: observerBytes.value, hardLimit: OUTPUT_HARD_BYTES, operatorOpen: true, flowAlive: live["data"], operatorSeed: operatorSeed["data"] }); + } finally { + await finish("S8", harness, observer ? [operator, observer] : [operator], pending, exited); + } +}, OUTPUT_STALL_MS + 12000); + +test("S9 output beyond hard limit closes before frame and kills PTY once", async () => { + const harness = await createServerHarness(); + const pending: Promise[] = []; + const socket = harness.openSocket(); + let exited: Promise | undefined; + try { + const open = waitOpen(harness, socket); pending.push(open); await open; + harness.queueInitialPtyOutput("S9-START"); + const ready = waitMessage(socket, (frame) => frame["type"] === "pane-geometry" || frame["type"] === "pty-data", "attach response missing"); + const initial = waitMessage(socket, (frame) => frame["type"] === "pty-data", "initial PTY data missing"); + socket.send(ackAttach(harness.paneId)); const attachReply = await ready; + if (attachReply["type"] !== "pty-data") await initial; + const closeWait = waitClose(socket); pending.push(closeWait); + const pty = harness.instances.at(-1); + if (!pty) throw new Error("attach did not create a fake PTY"); + exited = pty.exited; + const kill = harness.waitForPtyEvent(harness.paneId, "kill"); pending.push(kill.promise); + let ptyDataDelivered = false; + let firstPtyDataBeforeClose: boolean | undefined; + const onPtyData = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)) as Frame; + if (frame["type"] === "pty-data") { + ptyDataDelivered = true; + firstPtyDataBeforeClose ??= socket.readyState !== WebSocket.CLOSED; + } + }; + socket.addEventListener("message", onPtyData); + harness.emitPtyData(harness.paneId, "X".repeat(OUTPUT_HARD_BYTES + 1)); + const [close] = await Promise.all([closeWait, kill.promise, exited]); + socket.removeEventListener("message", onPtyData); + expect(close.code).toBe(closeCodeStalled); + expect(ptyDataDelivered).toBe(false); + expect(firstPtyDataBeforeClose).toBeUndefined(); + expect(pty.killCount).toBe(1); + records.push({ scenario: "S9", emittedBytes: OUTPUT_HARD_BYTES + 1, oversizedFrameDelivered: false, closeCode: close.code, killCount: pty.killCount }); + } finally { await finish("S9", harness, [socket], pending, exited); } +}); diff --git a/server/websocket-test/create-server-harness.test.ts b/server/websocket-test/create-server-harness.test.ts index b6f4a310d8..892bc7fb21 100644 --- a/server/websocket-test/create-server-harness.test.ts +++ b/server/websocket-test/create-server-harness.test.ts @@ -97,4 +97,74 @@ describe("original createServer harness boundary", () => { expect((await readFile(evidencePath, "utf8")).trim()).toBe(JSON.stringify(evidence)); expect(cleanup.subscriptionCloseDelta).toBeGreaterThan(0); }); + + test("cleanup closes a paused ACK socket and still removes its state", async () => { + delete process.env.SODAM_WS_SERVER_INDEX; + delete process.env.SODAM_WS_EVIDENCE_DIR; + const harness = await createServerHarness(); + const socket = harness.openSocket(); + const open = harness.prearmSocketEvent("open"); + const snapshot = new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const onMessage = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)) as { type?: string }; + if (frame.type !== "snapshot") return; + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + resolvePromise(); + }; + socket.addEventListener("message", onMessage); + timer = setTimeout(() => { + socket.removeEventListener("message", onMessage); + rejectPromise(new Error("timed out waiting for paused cleanup test snapshot")); + }, 5000); + }); + await Promise.all([open, snapshot]); + const pause = harness.waitForPtyEvent(harness.paneId, "pause"); + const role = new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const onMessage = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)) as { type?: string; mode?: string }; + if (frame.type !== "role-ack" || frame.mode !== "observe") return; + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + resolvePromise(); + }; + socket.addEventListener("message", onMessage); + timer = setTimeout(() => { + socket.removeEventListener("message", onMessage); + rejectPromise(new Error("timed out waiting for observer role acknowledgement")); + }, 5000); + }); + socket.send(JSON.stringify({ type: "role", mode: "observe" })); + await role; + const geometry = new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const onMessage = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)) as { type?: string }; + if (frame.type !== "pane-geometry") return; + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + resolvePromise(); + }; + socket.addEventListener("message", onMessage); + timer = setTimeout(() => { + socket.removeEventListener("message", onMessage); + rejectPromise(new Error("timed out waiting for paused cleanup attach")); + }, 5000); + }); + socket.send(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 100, rows: 30, flow_control: "ack" })); + await geometry; + const pty = harness.instances.at(-1); + if (!pty) throw new Error("attach did not create a fake PTY session"); + harness.emitPtyData(harness.paneId, "P".repeat(256 * 1024)); + await pause.promise; + const stateDir = harness.stateDir; + const cleanup = await harness.cleanup(); + const stateDirAbsent = await access(stateDir).then(() => false, (error: NodeJS.ErrnoException) => error.code === "ENOENT"); + expect(socket.readyState).toBe(WebSocket.CLOSED); + expect(pty.killCount).toBe(1); + expect(cleanup.subscriptionCloseDelta).toBeGreaterThan(0); + expect(stateDirAbsent).toBe(true); + }, 12000); }); diff --git a/server/websocket-test/create-server-harness.ts b/server/websocket-test/create-server-harness.ts index f1d07648f2..307db3a8dd 100644 --- a/server/websocket-test/create-server-harness.ts +++ b/server/websocket-test/create-server-harness.ts @@ -134,7 +134,6 @@ export async function createServerHarness(): Promise<{ const openSocket = (): WebSocket => { const opening = new WebSocket(`ws://${runningServer.hostname}:${runningServer.port}/ws`); sockets.add(opening); - prearmSocketEvent("close"); return opening; }; const connect = async (): Promise => { @@ -161,9 +160,27 @@ export async function createServerHarness(): Promise<{ async cleanup() { const before = fakeHerdrClient.getFakeHerdrCounters().subscriptionCloseCount; for (const waiter of ptyWaiters) waiter.cancel(); - for (const socket of sockets) socket.close(); - await Promise.all([...closeWaiters.values()]); + const closing = [...sockets].flatMap((socket) => { + if (socket.readyState === WebSocket.CLOSED) return []; + let timer: ReturnType; + const closed = new Promise((resolvePromise, rejectPromise) => { + const onClose = (event: Event) => { + clearTimeout(timer); + socket.removeEventListener("close", onClose); + resolvePromise(event); + }; + timer = setTimeout(() => { + socket.removeEventListener("close", onClose); + rejectPromise(new Error("timed out waiting for socket close during cleanup")); + }, timeoutMs); + socket.addEventListener("close", onClose, { once: true }); + }); + closeWaiters.set(socket, closed); + return [closed]; + }); runningServer.stop(); + for (const socket of sockets) if (socket.readyState !== WebSocket.CLOSED) socket.close(); + await Promise.all(closing); for (const instance of instances) instance.kill(); await Promise.all(instances.map((instance) => instance.exited)); try { diff --git a/server/websocket-test/mutation-controls.integration.test.ts b/server/websocket-test/mutation-controls.integration.test.ts new file mode 100644 index 0000000000..42c315638d --- /dev/null +++ b/server/websocket-test/mutation-controls.integration.test.ts @@ -0,0 +1,126 @@ +import { expect, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { websocketRunRoot } from "./create-server-harness.ts"; +import { generateAllMutants } from "./mutation-source.ts"; + +const sourceRoot = resolve(fileURLToPath(new URL("../..", import.meta.url))); +const evidenceRoot = resolve(websocketRunRoot, "evidence/mutants"); +const manifestPath = resolve(websocketRunRoot, "evidence/mutant-manifest.json"); +const logPath = resolve(websocketRunRoot, "evidence/mutation-controls.jsonl"); +const cleanupPath = resolve(websocketRunRoot, "evidence/mutation-controls-cleanup.json"); +const cases = [ + { name: "utf8-byte-count", scenario: "S1", marker: "MUTATION_ASSERTION_S1_UTF8_BYTE_COUNT" }, + { name: "cumulative-ack", scenario: "S2", marker: "MUTATION_ASSERTION_S2_CUMULATIVE_ACK" }, + { name: "stale-stream-id", scenario: "S4", marker: "MUTATION_ASSERTION_S4_STALE_STREAM_ID" }, + { name: "pause-resume", scenario: "S7", marker: "MUTATION_ASSERTION_S7_PAUSE_RESUME" }, +] as const; +const childTimeoutMs = 45000; + +interface ChildResult { + readonly name: (typeof cases)[number]["name"]; + readonly variant: "mutant" | "clean"; + readonly evidenceDir: string; + readonly exitCode: number; + readonly log: string; + readonly scenario: string; + readonly cleanup: unknown; +} + +function c0Complete(value: unknown, scenario: string): boolean { + const receipts = Array.isArray(value) ? value : typeof value === "object" && value !== null && "cases" in value ? value.cases : undefined; + if (!Array.isArray(receipts) || receipts.length !== 1) return false; + const receipt = receipts[0]; + if (typeof receipt !== "object" || receipt === null || Array.isArray(receipt)) return false; + const r = receipt as Record; + return r["scenario"] === scenario && r["cleanupError"] === null && r["stateDirAbsent"] === true && r["stoppedPortHandshakeFailed"] === true && + typeof r["subscriptionCloseDelta"] === "number" && r["subscriptionCloseDelta"] > 0 && r["subscriptionsClosed"] === true && + r["fakeExitedResolved"] === true && typeof r["socketsOpened"] === "number" && r["socketsOpened"] > 0 && + r["socketsClosed"] === r["socketsOpened"]; +} + +async function runChild(name: ChildResult["name"], variant: ChildResult["variant"], serverIndex?: string): Promise { + const evidenceDir = resolve(evidenceRoot, `${name}-${variant}-${randomUUID()}`); + await mkdir(evidenceDir, { recursive: true }); + const env: NodeJS.ProcessEnv = { + ...process.env, + SODAM_WS_MUTATION_CASE: name, + SODAM_WS_RUN_ROOT: websocketRunRoot, + SODAM_WS_EVIDENCE_DIR: evidenceDir, + }; + if (serverIndex === undefined) delete env["SODAM_WS_SERVER_INDEX"]; + else env["SODAM_WS_SERVER_INDEX"] = serverIndex; + const child = Bun.spawn(["bun", "--config=bunfig.websocket.toml", "test", "--isolate", "server/websocket-test/mutation-probe.integration.test.ts"], { + cwd: sourceRoot, + env, + stdout: "pipe", + stderr: "pipe", + }); + let timedOut = false; + const watchdog = setTimeout(() => { timedOut = true; child.kill(); }, childTimeoutMs); + const [exitCode, stdout, stderr] = await Promise.all([ + child.exited, + new Response(child.stdout).text(), + new Response(child.stderr).text(), + ]); + clearTimeout(watchdog); + const log = `${stdout}${stderr}`; + await writeFile(resolve(evidenceDir, "child.log"), log, "utf8"); + const scenario = await readFile(resolve(evidenceDir, "scenario.jsonl"), "utf8").catch(() => ""); + const cleanup = await readFile(resolve(evidenceDir, "scenario-cleanup.json"), "utf8").then((raw) => JSON.parse(raw) as unknown, () => null); + return { name, variant, evidenceDir, exitCode: timedOut ? -1 : exitCode, log: timedOut ? `${log}\nCHILD_WATCHDOG_TIMEOUT` : log, scenario, cleanup }; +} + +test("four websocket mutations fail their named assertions and clean controls pass", async () => { + await mkdir(dirname(manifestPath), { recursive: true }); + await generateAllMutants(); + const manifest = JSON.parse(await readFile(manifestPath, "utf8")) as { mutants: { name: string; outputPath: string }[] }; + expect(manifest.mutants.map((item) => item.name)).toEqual(cases.map((item) => item.name)); + for (const item of manifest.mutants) expect(resolve(item.outputPath)).toBe(resolve(websocketRunRoot, "mutants", item.name)); + const results: ChildResult[] = []; + for (const item of cases) { + const overlay = manifest.mutants.find((entry) => entry.name === item.name); + if (!overlay) throw new Error(`missing mutant overlay ${item.name}`); + results.push(await runChild(item.name, "mutant", resolve(overlay.outputPath, "server/index.ts"))); + results.push(await runChild(item.name, "clean")); + } + + const lines = results.map((result) => { + const item = cases.find((candidate) => candidate.name === result.name); + if (!item) throw new Error(`missing mutation case ${result.name}`); + const mutantPass = result.variant === "mutant" && result.exitCode !== 0 && result.log.includes(item.marker) && result.scenario.includes(item.marker) && c0Complete(result.cleanup, item.scenario); + const cleanPass = result.variant === "clean" && result.exitCode === 0 && !result.log.includes("MUTATION_ASSERTION_") && c0Complete(result.cleanup, item.scenario); + return JSON.stringify({ ...result, mutantPass, cleanPass }); + }); + await mkdir(dirname(logPath), { recursive: true }); + await writeFile(logPath, `${lines.join("\n")}\n`, "utf8"); + await writeFile(cleanupPath, `${JSON.stringify({ receipts: results.map(({ name, variant, evidenceDir, cleanup }) => { + const item = cases.find((candidate) => candidate.name === name); + if (!item) throw new Error(`missing mutation case ${name}`); + return { name, variant, evidenceDir, cleanup, c0Complete: c0Complete(cleanup, item.scenario) }; + }) }, null, 2)}\n`, "utf8"); + + expect(results).toHaveLength(8); + for (const result of results) { + const item = cases.find((candidate) => candidate.name === result.name); + if (!item) throw new Error(`missing mutation case ${result.name}`); + expect(c0Complete(result.cleanup, item.scenario)).toBe(true); + expect(result.scenario).not.toBe(""); + expect(result.cleanup).not.toBeNull(); + expect(result.scenario).toContain(`"scenario":"${item.scenario}"`); + expect(resolve(result.evidenceDir).startsWith(`${resolve(evidenceRoot)}${process.platform === "win32" ? "\\" : "/"}`)).toBe(true); + expect(result.evidenceDir).toContain(`${result.name}-${result.variant}-`); + } + for (const item of cases) { + const mutant = results.find((result) => result.name === item.name && result.variant === "mutant"); + const clean = results.find((result) => result.name === item.name && result.variant === "clean"); + if (!mutant || !clean) throw new Error(`missing mutant/clean result pair for ${item.name}`); + expect(mutant.exitCode).not.toBe(0); + expect(mutant.log).toContain(item.marker); + expect(mutant.scenario).toContain(item.marker); + expect(clean.exitCode).toBe(0); + expect(clean.log).not.toContain("MUTATION_ASSERTION_"); + } +}, 600000); diff --git a/server/websocket-test/mutation-probe.integration.test.ts b/server/websocket-test/mutation-probe.integration.test.ts new file mode 100644 index 0000000000..4f21f151e8 --- /dev/null +++ b/server/websocket-test/mutation-probe.integration.test.ts @@ -0,0 +1,260 @@ +import { access, mkdir, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { expect, test } from "bun:test"; +import { createServerHarness, evidenceDirectory } from "./create-server-harness.ts"; +import { OUTPUT_HIGH_BYTES, OUTPUT_LOW_BYTES } from "../output-window.ts"; + +const caseName = process.env.SODAM_WS_MUTATION_CASE; +const probes = { + "utf8-byte-count": { scenario: "S1", marker: "MUTATION_ASSERTION_S1_UTF8_BYTE_COUNT" }, + "cumulative-ack": { scenario: "S2", marker: "MUTATION_ASSERTION_S2_CUMULATIVE_ACK" }, + "stale-stream-id": { scenario: "S4", marker: "MUTATION_ASSERTION_S4_STALE_STREAM_ID" }, + "pause-resume": { scenario: "S7", marker: "MUTATION_ASSERTION_S7_PAUSE_RESUME" }, +} as const; +type ProbeName = keyof typeof probes; + +interface Frame { readonly type: string; readonly [key: string]: unknown } + +function requireCase(value: string | undefined): ProbeName { + if (value === "utf8-byte-count" || value === "cumulative-ack" || value === "stale-stream-id" || value === "pause-resume") return value; + throw new Error(`unknown or missing SODAM_WS_MUTATION_CASE: ${String(value)}`); +} + +function message(socket: WebSocket, predicate: (frame: Frame) => boolean, diagnostic: string): Promise { + return new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const cleanup = () => { + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + socket.removeEventListener("close", onClose); + socket.removeEventListener("error", onError); + }; + const onMessage = (event: MessageEvent) => { + try { + const value: unknown = JSON.parse(String(event.data)); + if (typeof value !== "object" || value === null || !("type" in value) || typeof value.type !== "string") throw new Error("invalid WebSocket frame"); + const frame = value as Frame; + if (!predicate(frame)) return; + cleanup(); + resolvePromise(frame); + } catch (error) { cleanup(); rejectPromise(error); } + }; + const onClose = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket closed`)); }; + const onError = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket errored`)); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error(diagnostic)); }, 5000); + socket.addEventListener("message", onMessage); + socket.addEventListener("close", onClose); + socket.addEventListener("error", onError); + }); +} + +function flow(frame: Frame): { stream_id: string; offset: number } { + const value: unknown = frame["flow"]; + if (typeof value !== "object" || value === null || !("stream_id" in value) || !("offset" in value)) throw new Error("pty-data omitted flow metadata"); + return { stream_id: String(value.stream_id), offset: Number(value.offset) }; +} + +function marker(name: ProbeName, detail: string): never { + throw new Error(`${probes[name].marker}: ${detail}`); +} + +async function c0( + harness: Awaited>, + sockets: WebSocket[], + name: ProbeName, + outcome: Record, +): Promise { + const stateDir = harness.stateDir; + const instances = [...harness.instances]; + let cleanupResult: Awaited> | undefined; + let cleanupError: unknown; + try { cleanupResult = await harness.cleanup(); } + catch (error) { cleanupError = error; } + const socketsClosed = sockets.every((socket) => socket.readyState === WebSocket.CLOSED); + const fakeExitedResolved = instances.length > 0 && instances.every((instance) => instance.killCount === 1); + const stateDirAbsent = await access(stateDir).then(() => false, (error: NodeJS.ErrnoException) => error.code === "ENOENT"); + const stoppedPortHandshakeFailed = await new Promise((resolvePromise) => { + const stopped = new WebSocket(`${harness.baseUrl}/ws`); + const timer = setTimeout(() => { stopped.close(); resolvePromise(false); }, 5000); + const onError = () => { clearTimeout(timer); stopped.removeEventListener("open", onOpen); resolvePromise(true); }; + const onOpen = () => { clearTimeout(timer); stopped.removeEventListener("error", onError); stopped.close(); resolvePromise(false); }; + stopped.addEventListener("error", onError, { once: true }); + stopped.addEventListener("open", onOpen, { once: true }); + }); + const receipt = { + scenario: probes[name].scenario, + socketsOpened: sockets.length, + socketsClosed: socketsClosed ? sockets.length : 0, + fakeExitedResolved, + subscriptionCloseDelta: cleanupResult?.subscriptionCloseDelta ?? 0, + subscriptionsClosed: (cleanupResult?.subscriptionCloseDelta ?? 0) > 0, + stateDirAbsent, + stoppedPortHandshakeFailed, + cleanupError: cleanupError === undefined ? null : String(cleanupError), + }; + if (cleanupError !== undefined) throw cleanupError; + expect(receipt.socketsClosed).toBe(sockets.length); + expect(receipt.fakeExitedResolved).toBe(true); + expect(receipt.subscriptionsClosed).toBe(true); + expect(receipt.stateDirAbsent).toBe(true); + expect(receipt.stoppedPortHandshakeFailed).toBe(true); + const directory = evidenceDirectory(); + await mkdir(directory, { recursive: true }); + await writeFile(resolve(directory, "scenario.jsonl"), `${JSON.stringify({ ...outcome, scenario: probes[name].scenario })}\n`, "utf8"); + await writeFile(resolve(directory, "scenario-cleanup.json"), `${JSON.stringify({ cases: [receipt] }, null, 2)}\n`, "utf8"); +} + +async function runProbe(name: ProbeName): Promise { + const harness = await createServerHarness(); + const sockets: WebSocket[] = []; + let outcome: Record = { passed: false }; + let assertionError: unknown; + try { + if (name === "utf8-byte-count") harness.queueInitialPtyOutput("\u001b[?1000h", "\u001b[?1006h", "éREPLAY"); + else harness.queueInitialPtyOutput("BASE"); + const socket = harness.openSocket(); sockets.push(socket); + await Promise.all([ + harness.prearmSocketEvent("open"), + message(socket, (frame) => frame.type === "snapshot", "snapshot timeout"), + ]); + let current: { stream_id: string; offset: number }; + let firstData: Frame | undefined; + if (name === "pause-resume") { + const readyMessage = message(socket, (frame) => frame.type === "pane-geometry" || frame.type === "pty-data", "S7 attach timeout"); + const initialWait = message(socket, (frame) => frame.type === "pty-data", "S7 initial data timeout"); + socket.send(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 100, rows: 30, flow_control: "ack" })); + const [initial, initialData] = await Promise.all([readyMessage, initialWait]); + const initialFrame = initial.type === "pty-data" ? initial : initialData; + current = flow(initialFrame); + } else { + const attachReply = message(socket, (frame) => frame.type === "pane-geometry" || frame.type === "pty-data", "attach reply timeout"); + socket.send(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 100, rows: 30, flow_control: "ack" })); + const first = await attachReply; + const currentFrame = first.type === "pty-data" ? first : await message(socket, (frame) => frame.type === "pty-data", "initial data timeout"); + firstData = currentFrame; + current = flow(currentFrame); + } + const ptySession = harness.instances.at(-1); + if (!ptySession) throw new Error("attach did not create a fake PTY session"); + + if (name === "utf8-byte-count") { + const replay = firstData; + if (!replay) throw new Error("initial replay was missing"); + const replayFlow = flow(replay); + const replayExpected = Buffer.byteLength(String(replay["data"]), "utf8"); + if (replayFlow.offset !== replayExpected) marker(name, `replay offset ${replayFlow.offset} != UTF-8 byteLength offset ${replayExpected}`); + current = replayFlow; + const liveWait = message(socket, (frame) => frame.type === "pty-data", "live data timeout"); + harness.emitPtyData(harness.paneId, "éLIVE"); + const live = await liveWait; + const liveFlow = flow(live); + const expected = current.offset + Buffer.byteLength(String(live["data"]), "utf8"); + if (liveFlow.offset !== expected) marker(name, `live offset ${liveFlow.offset} != UTF-8 byteLength offset ${expected}`); + outcome = { passed: true, replayData: replay["data"], replayOffset: replayFlow.offset, replayExpected, data: live["data"], offset: liveFlow.offset, expected }; + } else if (name === "cumulative-ack") { + const pauseWait = harness.waitForPtyEvent(harness.paneId, "pause"); + const burstWait = message(socket, (frame) => frame.type === "pty-data", "A burst timeout"); + harness.emitPtyData(harness.paneId, "A".repeat(300000)); + const firstBurst = await burstWait; await pauseWait.promise; + current = flow(firstBurst); + const resumeWait = harness.waitForPtyEvent(harness.paneId, "resume"); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: current.stream_id, offset: current.offset })); + await resumeWait.promise; + const roleAck = message(socket, (frame) => frame.type === "role-ack" && frame.mode === "observe", "role-ack barrier timeout"); + socket.send(JSON.stringify({ type: "role", mode: "observe" })); + await roleAck; + const regressiveAckBarrier = message(socket, (frame) => frame.type === "role-ack" && frame.mode === "observe", "regressive ACK barrier timeout"); + const secondWait = message(socket, (frame) => frame.type === "pty-data", "B burst timeout"); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: current.stream_id, offset: 100000 })); + socket.send(JSON.stringify({ type: "role", mode: "observe" })); + await regressiveAckBarrier; + harness.emitPtyData(harness.paneId, "B".repeat(63000)); + const second = await secondWait.catch((error) => { + if (error instanceof Error && error.message.includes("timeout")) marker(name, `regressive ACK blocked the expected B frame: ${error.message}`); + throw error; + }); + if (second.type !== "pty-data" || ptySession.pauseCount !== 1) marker(name, `regressive ACK restored credit: pauseCount=${ptySession.pauseCount}, B offset=${flow(second).offset}`); + outcome = { passed: true, firstOffset: current.offset, secondOffset: flow(second).offset, pauseCount: ptySession.pauseCount }; + } else if (name === "stale-stream-id") { + const oldStream = current.stream_id; + const initialClosed = message(socket, () => false, "initial websocket close timeout"); + socket.close(); + await initialClosed.catch((error) => { + if (!(error instanceof Error) || !error.message.includes("socket closed")) throw error; + }); + harness.queueInitialPtyOutput("NEW"); + const reopened = harness.openSocket(); sockets.push(reopened); + await Promise.all([ + harness.prearmSocketEvent("open"), + message(reopened, (frame) => frame.type === "snapshot", "reconnect snapshot timeout"), + ]); + const replayWait = message(reopened, (frame) => frame.type === "pty-data", "new replay timeout"); + reopened.send(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 100, rows: 30, flow_control: "ack" })); + const replay = await replayWait; + current = flow(replay); + const pauseWait = harness.waitForPtyEvent(harness.paneId, "pause"); + const burstWait = message(reopened, (frame) => frame.type === "pty-data", "current burst timeout"); + harness.emitPtyData(harness.paneId, "x".repeat(256 * 1024)); + const currentBurst = await burstWait; + await pauseWait.promise; + const currentPty = harness.instances.at(-1); + if (!currentPty) throw new Error("reconnected attach did not create a fake PTY session"); + const resumeCount = currentPty.resumeCount; + const barrier = message(reopened, (frame) => frame.type === "role-ack" && frame.mode === "observe", "stale ACK ordering barrier timeout"); + const errorCodes: string[] = []; + const collect = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)) as Frame; + if (frame.type === "error") errorCodes.push(String(frame.code)); + }; + reopened.addEventListener("message", collect); + reopened.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: oldStream, offset: Number.MAX_SAFE_INTEGER })); + reopened.send(JSON.stringify({ type: "role", mode: "observe" })); + await barrier; + reopened.removeEventListener("message", collect); + const currentPauseHeld = currentPty.resumeCount === resumeCount && currentPty.pauseCount === 1 && !errorCodes.includes("invalid_ack"); + if (!currentPauseHeld) marker(name, `old stream ACK affected current credit: errors=${errorCodes.join(",")}, resumeCount=${currentPty.resumeCount}`); + harness.queuePtyOutputOnResume(harness.paneId, "FLOW_RESUMED"); + const validResume = message(reopened, (frame) => frame.type === "pty-data" && frame.data === "FLOW_RESUMED", "valid current stream did not resume"); + const resumeEvent = harness.waitForPtyEvent(harness.paneId, "resume"); + reopened.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: current.stream_id, offset: flow(currentBurst).offset })); + await resumeEvent.promise; + const sentinel = await validResume; + outcome = { passed: true, oldStream, currentStream: current.stream_id, currentPauseHeld, errors: errorCodes, sentinel: sentinel["data"] }; + } else { + const highWait = message(socket, (frame) => frame.type === "pty-data", "high output timeout"); + const pause = harness.waitForPtyEvent(harness.paneId, "pause"); + harness.emitPtyData(harness.paneId, "H".repeat(OUTPUT_HIGH_BYTES)); + const high = await highWait; await pause.promise; + const highFlow = flow(high); + const before = ptySession.resumeCount; + const resume = harness.waitForPtyEvent(harness.paneId, "resume"); + const sentinel = message(socket, (frame) => frame.type === "pty-data" && frame.data === "FLOW_RESUMED", "resume sentinel missing"); + void sentinel.catch(() => undefined); + harness.queuePtyOutputOnResume(harness.paneId, "FLOW_RESUMED"); + socket.send(JSON.stringify({ type: "pty-ack", pane_id: harness.paneId, stream_id: highFlow.stream_id, offset: highFlow.offset - OUTPUT_LOW_BYTES })); + try { + await resume.promise; + } catch (error) { + if (ptySession.resumeCount === before) marker(name, `resume event missing after low-water ACK: ${String(error)}`); + throw error; + } + const resumed = await sentinel; + if (ptySession.resumeCount !== before + 1 || !String(resumed["data"]).includes("FLOW_RESUMED")) marker(name, `resume count/sentinel mismatch after barrier: ${ptySession.resumeCount - before}`); + outcome = { passed: true, initialOffset: current.offset, resumeDelta: ptySession.resumeCount - before, sentinel: resumed["data"] }; + } + } catch (error) { + assertionError = error; + if (error instanceof Error && error.message.includes(probes[name].marker)) outcome = { ...outcome, marker: probes[name].marker, detail: error.message }; + } + + try { await c0(harness, sockets, name, outcome); } + catch (error) { + if (assertionError === undefined) assertionError = error; + else if (error instanceof Error) assertionError = new AggregateError([assertionError, error], "probe assertion and C0 cleanup failed"); + } + if (assertionError !== undefined) throw assertionError; + expect(outcome["passed"]).toBe(true); +} + +const selected = requireCase(caseName); +test(`targeted websocket mutation probe: ${selected}`, () => runProbe(selected), 20000); diff --git a/server/websocket-test/mutation-source.test.ts b/server/websocket-test/mutation-source.test.ts new file mode 100644 index 0000000000..df7c8706be --- /dev/null +++ b/server/websocket-test/mutation-source.test.ts @@ -0,0 +1,99 @@ +import { createHash } from "node:crypto"; +import { access, mkdir, mkdtemp, readFile, readdir, rm } from "node:fs/promises"; +import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, test } from "bun:test"; +import ts from "typescript"; +import { generateAllMutants, generateMutant, mutationAnchors } from "./mutation-source.ts"; +import { websocketRunRoot } from "./create-server-harness.ts"; + +const sourceRoot = resolve(fileURLToPath(new URL("../..", import.meta.url))); +const names = ["utf8-byte-count", "cumulative-ack", "stale-stream-id", "pause-resume"] as const; +const indexPath = join(sourceRoot, "server/index.ts"); +const windowPath = join(sourceRoot, "server/output-window.ts"); + +function assertUnder(root: string, candidate: string): void { + const rel = relative(resolve(root), resolve(candidate)); + if (!rel || rel === "." || rel === ".." || rel.startsWith(`..${sep}`) || isAbsolute(rel)) { + throw new Error(`path is not contained under ${root}: ${candidate}`); + } +} + +test("generates isolated mutation overlays with current checkout provenance", async () => { + await mkdir(websocketRunRoot, { recursive: true }); + const fixtureRoot = await mkdtemp(join(websocketRunRoot, "mutation-source-fixture-")); + const mutantsRoot = join(fixtureRoot, "mutants"); + const manifestPath = join(fixtureRoot, "evidence", "mutant-manifest.json"); + try { + await generateAllMutants({ outputRoot: mutantsRoot, manifestPath }); + const manifest = JSON.parse(await readFile(manifestPath, "utf8")) as { + sourceCommit: string; + sourceRoot: string; + sourceHashes: Record; + mutants: Array<{ name: string; sourceHash: string; anchor: string; changedCopyPaths: string[]; outputPath: string }>; + }; + const proc = Bun.spawn(["git", "rev-parse", "HEAD"], { cwd: sourceRoot, stdout: "pipe", stderr: "pipe" }); + const [currentCommit, gitError, gitExit] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text(), proc.exited]); + expect(gitExit).toBe(0); + expect(gitError).toBe(""); + expect(manifest.sourceCommit).toBe(currentCommit.trim()); + expect(resolve(manifest.sourceRoot)).toBe(sourceRoot); + expect(manifest.sourceHashes["server/index.ts"]).toBe(createHash("sha256").update(await readFile(indexPath)).digest("hex")); + expect(manifest.sourceHashes["server/output-window.ts"]).toBe(createHash("sha256").update(await readFile(windowPath)).digest("hex")); + expect(manifest.mutants.map((entry) => entry.name)).toEqual([...names]); + for (const entry of manifest.mutants) { + expect(entry.sourceHash).toMatch(/^[a-f0-9]{64}$/); + expect(entry.anchor).toBe(mutationAnchors[entry.name as keyof typeof mutationAnchors]); + assertUnder(mutantsRoot, entry.outputPath); + for (const copyPath of entry.changedCopyPaths) assertUnder(entry.outputPath, copyPath); + const files = await readdir(join(entry.outputPath, "server")); + expect(files.sort()).toEqual(["index.ts", "output-window.ts"]); + const indexText = await readFile(join(entry.outputPath, "server/index.ts"), "utf8"); + const parsed = ts.createSourceFile("index.ts", indexText, ts.ScriptTarget.Latest, true, ts.ScriptKind.TS); + const imports: string[] = []; + for (const statement of parsed.statements) { + if (!ts.isImportDeclaration(statement) && !ts.isExportDeclaration(statement)) continue; + const moduleSpecifier = statement.moduleSpecifier; + if (!moduleSpecifier || !ts.isStringLiteral(moduleSpecifier)) continue; + const specifier = moduleSpecifier.text; + imports.push(specifier); + expect(specifier.startsWith(".")).toBe(false); + if (specifier.startsWith("node:") || specifier === "bun") continue; + if (!isAbsolute(specifier)) continue; + const resolved = resolve(specifier); + const rel = relative(sourceRoot, resolved); + const contained = rel !== "" && rel !== "." && rel !== ".." && !rel.startsWith(`..${sep}`) && !isAbsolute(rel); + const expected = resolved === resolve(join(entry.outputPath, "server/output-window.ts")) || contained; + expect(expected).toBe(true); + } + expect(imports.some((specifier) => resolve(specifier) === resolve(join(entry.outputPath, "server/output-window.ts")))).toBe(true); + expect(dirname(join(entry.outputPath, "server/index.ts"))).toBe(join(entry.outputPath, "server")); + } + } finally { + await rm(fixtureRoot, { recursive: true, force: true }); + } +}); + +test("unknown, absent, duplicate and escaping mutation inputs reject before writes", async () => { + await mkdir(websocketRunRoot, { recursive: true }); + const fixtureRoot = await mkdtemp(join(websocketRunRoot, "mutation-anchor-fixture-")); + const outputRoot = join(fixtureRoot, "mutants"); + const variantDirectory = join(outputRoot, "utf8-byte-count"); + const indexSource = await readFile(indexPath, "utf8"); + const missingAnchorSource = indexSource.replace(mutationAnchors["utf8-byte-count"], "const bytes = Buffer.byteLength(data) + 1;"); + const duplicateAnchorSource = `${indexSource}\n${mutationAnchors["utf8-byte-count"]}\n`; + try { + await expect(generateMutant("unknown-mutant", { outputRoot })).rejects.toThrow("unknown mutant"); + await expect(access(join(outputRoot, "unknown-mutant"))).rejects.toMatchObject({ code: "ENOENT" }); + await expect(generateMutant("utf8-byte-count", { sourceText: missingAnchorSource, outputRoot })).rejects.toThrow("expected exactly one source anchor"); + await expect(access(variantDirectory)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(generateMutant("utf8-byte-count", { sourceText: duplicateAnchorSource, outputRoot })).rejects.toThrow("expected exactly one source anchor"); + await expect(access(variantDirectory)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(generateMutant("cumulative-ack", { sourceText: indexSource, outputRoot })).rejects.toThrow("expected exactly one source anchor"); + await expect(access(join(outputRoot, "cumulative-ack"))).rejects.toMatchObject({ code: "ENOENT" }); + await expect(generateMutant("utf8-byte-count", { outputRoot: resolve(websocketRunRoot, "..", `${websocketRunRoot.split(/[\\/]/).at(-1)}-attacker`) })).rejects.toThrow("path is not contained"); + await expect(access(outputRoot)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { + await rm(fixtureRoot, { recursive: true, force: true }); + } +}); diff --git a/server/websocket-test/mutation-source.ts b/server/websocket-test/mutation-source.ts new file mode 100644 index 0000000000..4bcf586c35 --- /dev/null +++ b/server/websocket-test/mutation-source.ts @@ -0,0 +1,184 @@ +import { createHash } from "node:crypto"; +import { access, mkdir, readFile, readdir, writeFile } from "node:fs/promises"; +import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; +import { fileURLToPath } from "node:url"; +import ts from "typescript"; +import { websocketRunRoot } from "./create-server-harness.ts"; + +const sourceRoot = resolve(fileURLToPath(new URL("../..", import.meta.url))); +const sourceIndex = join(sourceRoot, "server/index.ts"); +const sourceWindow = join(sourceRoot, "server/output-window.ts"); +const mutantsRoot = join(websocketRunRoot, "mutants"); +const manifestPath = join(websocketRunRoot, "evidence/mutant-manifest.json"); + +export const mutationAnchors = { + "utf8-byte-count": "const bytes = Buffer.byteLength(data);", + "cumulative-ack": "this.acknowledged = Math.max(this.acknowledged, offset);", + "stale-stream-id": "if (window && window.id === message.stream_id && !window.acknowledge(message.stream_id, message.offset)) {", + "pause-resume": "else attachment.pty.resume();", +} as const; + +export type MutantName = keyof typeof mutationAnchors; + +interface Mutation { + readonly name: MutantName; + readonly sourcePath: string; + readonly anchor: string; + readonly replacement: string; +} + +const mutations: readonly Mutation[] = [ + { name: "utf8-byte-count", sourcePath: sourceIndex, anchor: mutationAnchors["utf8-byte-count"], replacement: "const bytes = data.length;" }, + { name: "cumulative-ack", sourcePath: sourceWindow, anchor: mutationAnchors["cumulative-ack"], replacement: "this.acknowledged = offset;" }, + { name: "stale-stream-id", sourcePath: sourceIndex, anchor: mutationAnchors["stale-stream-id"], replacement: "if (window && !window.acknowledge(message.stream_id, message.offset)) {" }, + { name: "pause-resume", sourcePath: sourceIndex, anchor: mutationAnchors["pause-resume"], replacement: "else { /* mutation: resume disabled */ }" }, +]; + +function replaceUniqueAnchor(source: string, anchor: string, replacement: string, label: string): string { + const first = source.indexOf(anchor); + if (first < 0 || source.indexOf(anchor, first + anchor.length) >= 0) { + throw new Error(`${label}: expected exactly one source anchor`); + } + return source.slice(0, first) + replacement + source.slice(first + anchor.length); +} + +function assertUniqueAnchor(source: string, anchor: string, label: string): void { + const first = source.indexOf(anchor); + if (first < 0 || source.indexOf(anchor, first + anchor.length) >= 0) { + throw new Error(`${label}: expected exactly one source anchor`); + } +} + +function assertUnder(root: string, candidate: string): void { + const rel = relative(resolve(root), resolve(candidate)); + if (!rel || rel === "." || rel === ".." || rel.startsWith(`..${sep}`) || isAbsolute(rel)) { + throw new Error(`path is not contained under ${root}: ${candidate}`); + } +} + +async function assertDoesNotExist(path: string, label: string): Promise { + try { + await access(path); + throw new Error(`${label} already exists: ${path}`); + } catch (error) { + if (error instanceof Error && error.message.startsWith(`${label} already exists:`)) throw error; + if (!(error instanceof Error && "code" in error && error.code === "ENOENT")) throw error; + } +} + +function transformedIndex(source: string, overlayIndex: string, overlayWindow: string): string { + const file = ts.createSourceFile(sourceIndex, source, ts.ScriptTarget.Latest, true, ts.ScriptKind.TS); + const edits: Array<{ start: number; end: number; value: string }> = []; + const sourceDir = dirname(sourceIndex); + const paired = resolve(overlayWindow); + + for (const statement of file.statements) { + if (!ts.isImportDeclaration(statement) && !ts.isExportDeclaration(statement)) continue; + const moduleSpecifier = statement.moduleSpecifier; + if (!moduleSpecifier || !ts.isStringLiteral(moduleSpecifier)) continue; + const specifier = moduleSpecifier.text; + if (!specifier.startsWith(".")) continue; + const cleanTarget = resolve(sourceDir, specifier); + const resolvedTarget = cleanTarget === resolve(sourceWindow) ? paired : cleanTarget; + if (cleanTarget !== resolve(sourceWindow)) assertUnder(sourceRoot, cleanTarget); + if (resolvedTarget === paired) assertUnder(dirname(overlayIndex), resolvedTarget); + const quoted = JSON.stringify(resolvedTarget).replaceAll("\\\\", "/"); + edits.push({ start: moduleSpecifier.getStart(file), end: moduleSpecifier.end, value: quoted }); + } + + let text = source; + for (const edit of edits.sort((a, b) => b.start - a.start)) text = text.slice(0, edit.start) + edit.value + text.slice(edit.end); + return text; +} + +async function ensureFreshVariant(name: MutantName, root: string): Promise { + const outputDir = resolve(root, name); + assertUnder(root, outputDir); + try { + await readdir(outputDir); + throw new Error(`variant output already exists: ${outputDir}`); + } catch (error) { + if (error instanceof Error && error.message.startsWith("variant output already exists:")) throw error; + if (!(error instanceof Error && "code" in error && error.code === "ENOENT")) throw error; + } + return outputDir; +} + +export async function generateMutant(name: string, options: { sourceText?: string; outputRoot?: string } = {}): Promise<{ name: MutantName; sourceHash: string; anchor: string; changedCopyPaths: string[]; outputPath: string }> { + if (!Object.hasOwn(mutationAnchors, name)) throw new Error(`unknown mutant: ${name}`); + const outputRoot = resolve(options.outputRoot ?? mutantsRoot); + assertUnder(websocketRunRoot, outputRoot); + const mutation = mutations.find((item) => item.name === name); + if (!mutation) throw new Error(`mutation definition missing: ${name}`); + const source = options.sourceText ?? await readFile(mutation.sourcePath, "utf8"); + assertUniqueAnchor(source, mutation.anchor, name); + const variantDir = await ensureFreshVariant(mutation.name, outputRoot); + const serverDir = join(variantDir, "server"); + const overlayIndex = join(serverDir, "index.ts"); + const overlayWindow = join(serverDir, "output-window.ts"); + const indexSource = mutation.sourcePath === sourceIndex ? source : await readFile(sourceIndex, "utf8"); + const windowSource = mutation.sourcePath === sourceWindow ? source : await readFile(sourceWindow, "utf8"); + const cleanIndex = mutation.sourcePath === sourceIndex ? replaceUniqueAnchor(indexSource, mutation.anchor, mutation.replacement, name) : indexSource; + const cleanWindow = mutation.sourcePath === sourceWindow ? replaceUniqueAnchor(windowSource, mutation.anchor, mutation.replacement, name) : windowSource; + const overlayIndexText = transformedIndex(cleanIndex, overlayIndex, overlayWindow); + const windowText = cleanWindow; + const importAudit = ts.createSourceFile(overlayIndex, overlayIndexText, ts.ScriptTarget.Latest, true, ts.ScriptKind.TS); + for (const statement of importAudit.statements) { + if (!ts.isImportDeclaration(statement) && !ts.isExportDeclaration(statement)) continue; + const moduleSpecifier = statement.moduleSpecifier; + if (!moduleSpecifier || !ts.isStringLiteral(moduleSpecifier)) continue; + const specifier = moduleSpecifier.text; + if (specifier.startsWith(".")) throw new Error(`relative import remains in overlay: ${specifier}`); + if (specifier.startsWith("node:") || specifier === "bun" || !isAbsolute(specifier)) continue; + if (resolve(specifier) !== resolve(overlayWindow) && !resolve(specifier).startsWith(`${sourceRoot}${sep}`)) { + throw new Error(`unexpected rewritten import target: ${specifier}`); + } + } + + await mkdir(serverDir, { recursive: true }); + await writeFile(overlayIndex, overlayIndexText, "utf8"); + await writeFile(overlayWindow, windowText, "utf8"); + return { + name: mutation.name, + sourceHash: createHash("sha256").update(source).digest("hex"), + anchor: mutation.anchor, + changedCopyPaths: [overlayIndex, ...(mutation.sourcePath === sourceIndex ? [] : [overlayWindow])], + outputPath: variantDir, + }; +} + +async function currentSourceCommit(): Promise { + const proc = Bun.spawn(["git", "rev-parse", "HEAD"], { cwd: sourceRoot, stdout: "pipe", stderr: "pipe" }); + const [commit, stderr, exitCode] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text(), proc.exited]); + if (exitCode !== 0) throw new Error(`git rev-parse HEAD failed: ${stderr.trim()}`); + return commit.trim(); +} + +export async function generateAllMutants(options: { outputRoot?: string; manifestPath?: string } = {}): Promise { + const outputRoot = resolve(options.outputRoot ?? mutantsRoot); + const outputManifest = resolve(options.manifestPath ?? manifestPath); + assertUnder(websocketRunRoot, outputRoot); + assertUnder(websocketRunRoot, outputManifest); + await assertDoesNotExist(outputManifest, "manifest"); + for (const mutation of mutations) await assertDoesNotExist(resolve(outputRoot, mutation.name), "variant output"); + + const [indexText, windowText, sourceCommit] = await Promise.all([ + readFile(sourceIndex, "utf8"), + readFile(sourceWindow, "utf8"), + currentSourceCommit(), + ]); + const entries = []; + for (const mutation of mutations) entries.push(await generateMutant(mutation.name, { outputRoot })); + await mkdir(dirname(outputManifest), { recursive: true }); + await writeFile(outputManifest, `${JSON.stringify({ + sourceCommit, + sourceRoot, + sourceHashes: { + "server/index.ts": createHash("sha256").update(indexText).digest("hex"), + "server/output-window.ts": createHash("sha256").update(windowText).digest("hex"), + }, + mutants: entries, + }, null, 2)}\n`, "utf8"); +} + +if (import.meta.main) await generateAllMutants(); diff --git a/server/websocket-test/protocol-guards.integration.test.ts b/server/websocket-test/protocol-guards.integration.test.ts new file mode 100644 index 0000000000..3b174f9654 --- /dev/null +++ b/server/websocket-test/protocol-guards.integration.test.ts @@ -0,0 +1,165 @@ +import { access, mkdir, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { expect, test } from "bun:test"; +import { createServerHarness, evidenceDirectory } from "./create-server-harness.ts"; +import { onPtyEvent } from "./fake-pty-session.ts"; +import { getFakeHerdrCounters } from "./fake-herdr-client.ts"; + +const evidencePath = resolve(evidenceDirectory(), "protocol-guards.jsonl"); +const cleanupPath = resolve(evidenceDirectory(), "protocol-guards-cleanup.json"); +const timeoutMs = 5000; + +interface WsFrame { readonly type: string; readonly [key: string]: unknown } + +function messageWait(socket: WebSocket, diagnostic: string): Promise { + return new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const cleanup = () => { + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + socket.removeEventListener("close", onClose); + socket.removeEventListener("error", onError); + }; + const onMessage = (event: MessageEvent) => { + cleanup(); + try { + const frame: unknown = JSON.parse(String(event.data)); + if (typeof frame !== "object" || frame === null || !("type" in frame) || typeof frame.type !== "string") throw new Error("frame has no type"); + resolvePromise(frame as WsFrame); + } catch (error) { rejectPromise(error); } + }; + const onClose = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket closed`)); }; + const onError = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket errored`)); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error(diagnostic)); }, timeoutMs); + socket.addEventListener("message", onMessage); + socket.addEventListener("close", onClose); + socket.addEventListener("error", onError); + }); +} + +test("real server rejects malformed protocol and observer mutations", async () => { + await mkdir(evidenceDirectory(), { recursive: true }); + const harness = await createServerHarness(); + const socket = harness.openSocket(); + const openWait = harness.prearmSocketEvent("open"); + const snapshotWait = messageWait(socket, "timed out waiting for initial snapshot"); + const waits: Promise[] = [openWait, snapshotWait]; + let exitEvent: Promise | undefined; + let unsubscribeExit: (() => void) | undefined; + let fakeExited: Promise | undefined; + let cleanupReceipt: Record | undefined; + try { + const [opened, snapshot] = await Promise.all([openWait, snapshotWait]); + expect(opened.socket).toBe(socket); + expect(snapshot.type).toBe("snapshot"); + + const observedFrames: WsFrame[] = []; + const protocolOrder: string[] = []; + const expectedMessages: Record = { + invalid_json: "message must be JSON", + invalid_flow_control: "flow_control must be ack", + invalid_geometry: "cols and rows must be integers in 1..1000", + read_only: "this connection is in observe mode", + }; + const sendAndExpect = async (raw: string, code: string): Promise => { + const responseWait = messageWait(socket, `timed out waiting for ${code}`); + waits.push(responseWait); + socket.send(raw); + const frame = await responseWait; + expect(frame).toEqual({ type: "error", code, message: expectedMessages[code] }); + observedFrames.push(frame); + protocolOrder.push(code); + return frame; + }; + + await sendAndExpect("{", "invalid_json"); + await sendAndExpect(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 120, rows: 40, flow_control: "bogus" }), "invalid_flow_control"); + await sendAndExpect(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 0, rows: 40 }), "invalid_geometry"); + + const roleWait = messageWait(socket, "timed out waiting for observe role acknowledgement"); + waits.push(roleWait); + socket.send(JSON.stringify({ type: "role", mode: "observe" })); + const roleAck = await roleWait; + expect(roleAck).toEqual({ type: "role-ack", mode: "observe" }); + protocolOrder.push("role-ack"); + + const attachWait = messageWait(socket, "timed out waiting for valid observer attach"); + waits.push(attachWait); + socket.send(JSON.stringify({ type: "attach", pane_id: harness.paneId, cols: 120, rows: 40, flow_control: "ack" })); + const attached = await attachWait; + expect(["pane-geometry", "pty-data"]).toContain(attached.type); + if (attached.type === "pty-data") { + const geometryWait = messageWait(socket, "timed out waiting for observer geometry"); + waits.push(geometryWait); + const geometry = await geometryWait; + expect(geometry.type).toBe("pane-geometry"); + } + + const inputErrorWait = messageWait(socket, "timed out waiting for read_only input error"); + waits.push(inputErrorWait); + socket.send(JSON.stringify({ type: "input", pane_id: harness.paneId, text: "must-not-write" })); + const inputError = await inputErrorWait; + expect(inputError).toEqual({ type: "error", code: "read_only", message: expectedMessages["read_only"] }); + observedFrames.push(inputError); + protocolOrder.push("read_only"); + + const inputErrorIndex = observedFrames.length - 1; + await sendAndExpect(JSON.stringify({ type: "resize", pane_id: harness.paneId, cols: 120, rows: 40 }), "read_only"); + expect(observedFrames.map((frame) => frame.code)).toEqual(["invalid_json", "invalid_flow_control", "invalid_geometry", "read_only", "read_only"]); + expect(inputErrorIndex).toBe(3); + expect(protocolOrder).toEqual(["invalid_json", "invalid_flow_control", "invalid_geometry", "role-ack", "read_only", "read_only"]); + const fake = harness.instances.at(-1); + expect(fake).toBeDefined(); + expect(fake?.writes).toEqual([]); + expect(fake?.resizes).toEqual([]); + const subscriptionCountBeforeCleanup = getFakeHerdrCounters().subscriptionCloseCount; + exitEvent = new Promise((resolvePromise) => { + const unsubscribe = onPtyEvent((event) => { + if (event.type !== "exit") return; + unsubscribe(); + resolvePromise(); + }); + unsubscribeExit = unsubscribe; + }); + fakeExited = fake?.exited; + await writeFile(evidencePath, `${JSON.stringify({ scenario: "S10", snapshot, actions: ["raw {", "attach flow_control bogus", "attach cols 0", "role observe", "attach valid ACK flow", "observer input must-not-write", "observer resize 120x40"], frames: observedFrames, protocolOrder, roleAck, counters: { writes: fake?.writes.length ?? 0, resizes: fake?.resizes.length ?? 0, subscriptionCloseCountBeforeCleanup: subscriptionCountBeforeCleanup } })}\n`, "utf8"); + } finally { + const stateDir = harness.stateDir; + const sessionExit = fakeExited ?? harness.instances.at(-1)?.exited; + const cleanup = await harness.cleanup(); + await Promise.allSettled(waits); + let fakeExitedResolved = false; + if (sessionExit) { await sessionExit; fakeExitedResolved = true; } + if (exitEvent) await exitEvent; + unsubscribeExit?.(); + const stateDirAbsent = await access(stateDir).then(() => false, (error: NodeJS.ErrnoException) => error.code === "ENOENT"); + const stoppedPortHandshakeFailed = await new Promise((resolvePromise, rejectPromise) => { + const stopped = new WebSocket(new URL("/ws", harness.baseUrl)); + let timer: ReturnType; + const settle = (failed: boolean) => { + clearTimeout(timer); + stopped.removeEventListener("error", onError); + stopped.removeEventListener("open", onOpen); + if (failed) resolvePromise(true); + else { stopped.close(); resolvePromise(false); } + }; + const onError = () => settle(true); + const onOpen = () => settle(false); + timer = setTimeout(() => { stopped.close(); rejectPromise(new Error("stopped-port handshake did not settle")); }, 1000); + stopped.addEventListener("error", onError); + stopped.addEventListener("open", onOpen); + }); + cleanupReceipt = { + scenario: "S10", + socketsClosed: socket.readyState === WebSocket.CLOSED, + fakeExitedResolved, + subscriptionCloseDelta: cleanup.subscriptionCloseDelta, + stateDirAbsent, + stoppedPortHandshakeFailed, + }; + await writeFile(cleanupPath, `${JSON.stringify({ cases: [cleanupReceipt] }, null, 2)}\n`, "utf8"); + unsubscribeExit?.(); + } + expect(cleanupReceipt).toMatchObject({ socketsClosed: true, fakeExitedResolved: true, stateDirAbsent: true, stoppedPortHandshakeFailed: true }); + expect(Number(cleanupReceipt?.subscriptionCloseDelta)).toBeGreaterThan(0); +}, 15000); diff --git a/server/websocket-test/reattach-reconnect.integration.test.ts b/server/websocket-test/reattach-reconnect.integration.test.ts new file mode 100644 index 0000000000..a06c300c68 --- /dev/null +++ b/server/websocket-test/reattach-reconnect.integration.test.ts @@ -0,0 +1,272 @@ +import { access, mkdir, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { expect, test } from "bun:test"; +import { createServerHarness, evidenceDirectory } from "./create-server-harness.ts"; + +const evidencePath = resolve(evidenceDirectory(), "reattach-reconnect.jsonl"); +const cleanupPath = resolve(evidenceDirectory(), "reattach-reconnect-cleanup.json"); +const paneId = "fake-pane"; +const timeoutMs = 5000; +const records: Record[] = []; +const cleanupCases: Record[] = []; + +interface Frame { readonly type: string; readonly [key: string]: unknown } + +function message(socket: WebSocket, diagnostic: string, predicate: (frame: Frame) => boolean = () => true): Promise { + return new Promise((resolvePromise, rejectPromise) => { + let timer: ReturnType; + const cleanup = () => { + clearTimeout(timer); + socket.removeEventListener("message", onMessage); + socket.removeEventListener("close", onClose); + socket.removeEventListener("error", onError); + }; + const onMessage = (event: MessageEvent) => { + try { + const parsed: unknown = JSON.parse(String(event.data)); + if (typeof parsed !== "object" || parsed === null || !("type" in parsed) || typeof parsed.type !== "string") throw new Error("frame has no string type"); + const frame = parsed as Frame; + if (!predicate(frame)) return; + cleanup(); + resolvePromise(frame); + } catch (error) { + cleanup(); + rejectPromise(error); + } + }; + const onClose = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket closed`)); }; + const onError = () => { cleanup(); rejectPromise(new Error(`${diagnostic}: socket errored`)); }; + timer = setTimeout(() => { cleanup(); rejectPromise(new Error(diagnostic)); }, timeoutMs); + socket.addEventListener("message", onMessage); + socket.addEventListener("close", onClose); + socket.addEventListener("error", onError); + }); +} + +const ptyMessage = (socket: WebSocket, diagnostic: string) => message(socket, diagnostic, (frame) => frame.type === "pty-data"); +const dataMessage = (socket: WebSocket, data: string, diagnostic: string) => message(socket, diagnostic, (frame) => frame.type === "pty-data" && frame.data === data); + +function flow(frame: Frame): { stream_id: string; offset: number } { + const value = frame.flow; + if (typeof value !== "object" || value === null || !("stream_id" in value) || !("offset" in value)) throw new Error("pty-data omitted flow metadata"); + return { stream_id: String(value.stream_id), offset: Number(value.offset) }; +} + +async function openedWithSnapshot(harness: Awaited>, socket: WebSocket): Promise { + const opened = harness.prearmSocketEvent("open"); + const snapshot = message(socket, "timed out waiting for initial snapshot", (frame) => frame.type === "snapshot"); + const [openResult, snapshotFrame] = await Promise.all([opened, snapshot]); + expect(openResult.socket).toBe(socket); + expect(snapshotFrame.type).toBe("snapshot"); +} + +async function finishCase(harness: Awaited>, sockets: WebSocket[], name: string, expectedExit?: Promise): Promise { + const stateDir = harness.stateDir; + const instances = [...harness.instances]; + const exits = instances.map((instance) => instance.exited); + const closePromises = sockets.map(async (socket) => { + if (socket.readyState === WebSocket.CLOSED) return; + const closed = harness.prearmSocketEvent("close"); + if (socket.readyState !== WebSocket.CLOSED) socket.close(); + await closed; + }); + await Promise.all(closePromises); + const cleanup = await harness.cleanup(); + await Promise.all(exits); + if (expectedExit) await expectedExit; + const stateDirAbsent = await access(stateDir).then(() => false, (error: NodeJS.ErrnoException) => error.code === "ENOENT"); + const stoppedPortHandshakeFailed = await new Promise((resolvePromise, rejectPromise) => { + const stopped = new WebSocket(`${harness.baseUrl}/ws`); + let timer: ReturnType; + const settle = (failed: boolean) => { + clearTimeout(timer); + stopped.removeEventListener("error", onError); + stopped.removeEventListener("open", onOpen); + if (!failed && stopped.readyState !== WebSocket.CLOSED) stopped.close(); + resolvePromise(failed); + }; + const onError = () => settle(true); + const onOpen = () => settle(false); + timer = setTimeout(() => { stopped.close(); rejectPromise(new Error("stopped-port handshake did not settle")); }, timeoutMs); + stopped.addEventListener("error", onError); + stopped.addEventListener("open", onOpen); + }); + const receipt = { + scenario: name, + socketsClosed: sockets.every((socket) => socket.readyState === WebSocket.CLOSED), + fakeExited: instances.length > 0 && instances.every((instance) => instance.killCount === 1), + subscriptionCloseDelta: cleanup.subscriptionCloseDelta, + stateDirAbsent, + stoppedPortHandshakeFailed, + }; + cleanupCases.push(receipt); + await mkdir(evidenceDirectory(), { recursive: true }); + await writeFile(evidencePath, `${records.map((record) => JSON.stringify(record)).join("\n")}\n`, "utf8"); + await writeFile(cleanupPath, `${JSON.stringify({ cases: cleanupCases }, null, 2)}\n`, "utf8"); + expect(receipt.socketsClosed).toBe(true); + expect(receipt.fakeExited).toBe(true); + expect(receipt.subscriptionCloseDelta).toBeGreaterThan(0); + expect(receipt.stateDirAbsent).toBe(true); + expect(receipt.stoppedPortHandshakeFailed).toBe(true); +} + +function attach(socket: WebSocket): void { socket.send(JSON.stringify({ type: "attach", pane_id: paneId, cols: 100, rows: 30, flow_control: "ack" })); } +function roleBarrier(socket: WebSocket): Promise { + const ack = message(socket, "timed out waiting for interact role-ack", (frame) => frame.type === "role-ack" && frame.mode === "interact"); + socket.send(JSON.stringify({ type: "role", mode: "interact" })); + return ack; +} + +test("S4 reattach isolates stale stream ACKs and resumes current credit", async () => { + const harness = await createServerHarness(); + const sockets: WebSocket[] = []; + let expectedExit: Promise | undefined; + try { + harness.queueInitialPtyOutput("OLD"); + const socket = harness.openSocket(); sockets.push(socket); + await openedWithSnapshot(harness, socket); + const oldReplayWait = ptyMessage(socket, "timed out waiting for OLD replay"); + attach(socket); + const oldReplay = await oldReplayWait; + const oldFlow = flow(oldReplay); + const oldSession = harness.instances.at(-1); + if (!oldSession) throw new Error("OLD attach did not create a PTY session"); + expectedExit = oldSession.exited; + const roleAck = roleBarrier(socket); + const detached = message(socket, "timed out waiting for detach acknowledgement", (frame) => frame.type === "detach-ack" || frame.type === "role-ack"); + socket.send(JSON.stringify({ type: "detach", pane_id: paneId })); + await Promise.all([roleAck, detached]); + const closed = harness.prearmSocketEvent("close"); + socket.close(); + await closed; + await oldSession.exited; + + harness.queueInitialPtyOutput("NEW"); + const reopened = harness.openSocket(); sockets.push(reopened); + await openedWithSnapshot(harness, reopened); + await roleBarrier(reopened); + const newReplayWait = ptyMessage(reopened, "timed out waiting for NEW replay"); + attach(reopened); + const current = await newReplayWait; + const currentFlow = flow(current); + expect(current.data).toBe("NEW"); + expect(currentFlow.stream_id).not.toBe(oldFlow.stream_id); + const currentSession = harness.instances.at(-1); + if (!currentSession) throw new Error("NEW attach did not create a PTY session"); + expectedExit = currentSession.exited; + + const paused = harness.waitForPtyEvent(paneId, "pause"); + const burstWait = ptyMessage(reopened, "timed out waiting for current-stream burst"); + harness.emitPtyData(paneId, "x".repeat(256 * 1024)); + const burst = await burstWait; + await paused.promise; + const burstFlow = flow(burst); + expect(burstFlow.offset).toBe(currentFlow.offset + 256 * 1024); + + const resumeCountBeforeStaleAck = currentSession.resumeCount; + const staleAckBarrier = message(reopened, "timed out waiting for stale-ACK barrier", (frame) => frame.type === "role-ack" && frame.mode === "interact"); + const observedErrorCodes: unknown[] = []; + const collectErrors = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)) as Frame; + if (frame.type === "error") observedErrorCodes.push(frame.code); + }; + reopened.addEventListener("message", collectErrors); + reopened.send(JSON.stringify({ type: "pty-ack", pane_id: paneId, stream_id: oldFlow.stream_id, offset: Number.MAX_SAFE_INTEGER })); + reopened.send(JSON.stringify({ type: "role", mode: "interact" })); + await staleAckBarrier; + expect(currentSession.resumeCount).toBe(resumeCountBeforeStaleAck); + const currentPauseStillHeld = currentSession.pauseCount === 1 && currentSession.resumeCount === resumeCountBeforeStaleAck; + expect(currentPauseStillHeld).toBe(true); + const staleAckReleasedNoCredit = currentSession.resumeCount === resumeCountBeforeStaleAck; + reopened.removeEventListener("message", collectErrors); + expect(observedErrorCodes).not.toContain("invalid_ack"); + + const sentinelWait = dataMessage(reopened, "FLOW_RESUMED", "timed out waiting for FLOW_RESUMED sentinel"); + const resumed = harness.waitForPtyEvent(paneId, "resume"); + harness.queuePtyOutputOnResume(paneId, "FLOW_RESUMED"); + reopened.send(JSON.stringify({ type: "pty-ack", pane_id: paneId, stream_id: currentFlow.stream_id, offset: burstFlow.offset })); + await resumed.promise; + expect(currentSession.resumeCount).toBe(resumeCountBeforeStaleAck + 1); + const sentinel = await sentinelWait; + records.push({ case: "S4", oldStreamId: oldFlow.stream_id, currentStreamId: currentFlow.stream_id, staleAckOffset: Number.MAX_SAFE_INTEGER, staleAckReleasedNoCredit, invalidAck: false, resumedSentinel: sentinel.data, sentinelOffset: flow(sentinel).offset }); + } finally { await finishCase(harness, sockets, "S4", expectedExit); } +}, 20000); + +test("S5 duplicate attach is idempotent and live output advances UTF-8 offset", async () => { + const harness = await createServerHarness(); + const sockets: WebSocket[] = []; + let expectedExit: Promise | undefined; + try { + harness.queueInitialPtyOutput("BASE"); + const socket = harness.openSocket(); sockets.push(socket); + await openedWithSnapshot(harness, socket); + const firstWait = ptyMessage(socket, "timed out waiting for initial replay"); + attach(socket); + const first = await firstWait; + const originalFlow = flow(first); + const session = harness.instances.at(-1); + if (!session) throw new Error("initial attach did not create a PTY session"); + expectedExit = session.exited; + + let duplicateReplayObserved = false; + const collectDuplicateReplay = (event: MessageEvent) => { + const frame = JSON.parse(String(event.data)) as Frame; + if (frame.type === "pty-data") duplicateReplayObserved = true; + }; + socket.addEventListener("message", collectDuplicateReplay); + attach(socket); + await roleBarrier(socket); + socket.removeEventListener("message", collectDuplicateReplay); + expect(duplicateReplayObserved).toBe(false); + + const liveWait = ptyMessage(socket, "timed out waiting for LIVE frame"); + harness.emitPtyData(paneId, "éLIVE"); + const live = await liveWait; + const liveFlow = flow(live); + expect(live.data).toBe("éLIVE"); + expect(liveFlow.stream_id).toBe(originalFlow.stream_id); + expect(liveFlow.offset).toBe(originalFlow.offset + new TextEncoder().encode("éLIVE").byteLength); + records.push({ case: "S5", duplicateAttachAcknowledged: true, secondReplay: false, originalStreamId: originalFlow.stream_id, liveData: live.data, liveOffsetDelta: liveFlow.offset - originalFlow.offset, utf8Bytes: new TextEncoder().encode("éLIVE").byteLength }); + } finally { await finishCase(harness, sockets, "S5", expectedExit); } +}, 15000); + +test("S6 fresh reconnect receives independent stream replay", async () => { + const harness = await createServerHarness(); + const sockets: WebSocket[] = []; + let expectedExit: Promise | undefined; + try { + harness.queueInitialPtyOutput("FIRST"); + const firstSocket = harness.openSocket(); sockets.push(firstSocket); + await openedWithSnapshot(harness, firstSocket); + const firstReplayWait = ptyMessage(firstSocket, "timed out waiting for FIRST replay"); + attach(firstSocket); + const firstReplay = await firstReplayWait; + const firstFlow = flow(firstReplay); + const firstSession = harness.instances.at(-1); + if (!firstSession) throw new Error("FIRST attach did not create a PTY session"); + expectedExit = firstSession.exited; + const close = harness.prearmSocketEvent("close"); + firstSocket.close(); + await close; + await firstSession.exited; + + harness.queueInitialPtyOutput("SECOND-é"); + const secondSocket = harness.openSocket(); sockets.push(secondSocket); + const opened = harness.prearmSocketEvent("open"); + const snapshot = message(secondSocket, "timed out waiting for reconnect snapshot", (frame) => frame.type === "snapshot"); + const [openResult, snapshotFrame] = await Promise.all([opened, snapshot]); + expect(openResult.socket).toBe(secondSocket); + expect(snapshotFrame.type).toBe("snapshot"); + const replayWait = ptyMessage(secondSocket, "timed out waiting for SECOND replay"); + attach(secondSocket); + const replay = await replayWait; + const secondFlow = flow(replay); + expect(replay.data).toBe("SECOND-é"); + expect(secondFlow.offset).toBe(new TextEncoder().encode("SECOND-é").byteLength); + expect(secondFlow.stream_id).not.toBe(firstFlow.stream_id); + const secondSession = harness.instances.at(-1); + if (!secondSession) throw new Error("SECOND attach did not create a PTY session"); + expectedExit = secondSession.exited; + records.push({ case: "S6", firstStreamId: firstFlow.stream_id, secondStreamId: secondFlow.stream_id, distinctStreamIds: true, replayData: replay.data, replayOffset: secondFlow.offset, replayBytes: new TextEncoder().encode("SECOND-é").byteLength, firstSocketClosedBeforeReattach: firstSocket.readyState === WebSocket.CLOSED, firstSocketReceivedNoLaterFrame: true }); + } finally { await finishCase(harness, sockets, "S6", expectedExit); } +}, 15000); From d5a6f30c0e12cc66be8a5b98bac4a9bf7670e2f6 Mon Sep 17 00:00:00 2001 From: pjhsion Date: Sat, 10 Oct 2026 11:44:08 +0900 Subject: [PATCH 3/4] Add pull-request CI workflow and WebSocket test docs --- .github/workflows/ci.yml | 35 +++++++++++++++++++++++++++++++++++ README.md | 15 +++++++++++++++ 2 files changed, 50 insertions(+) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000000..3d4e53b061 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,35 @@ +name: CI +on: + pull_request: + branches: [main] + push: + branches: [main] +permissions: + contents: read +jobs: + ci: + runs-on: ubuntu-24.04 + env: + HERDR_SOCKET: ${{ runner.temp }}/herdr-websocket-no-service.sock + SODAM_WS_RUN_ROOT: ${{ runner.temp }}/herdr-websocket-${{ github.run_id }}-${{ github.run_attempt }} + steps: + - uses: actions/checkout@v4 + - uses: oven-sh/setup-bun@v2 + with: + bun-version: 1.4.2 + - run: bun install --frozen-lockfile + - run: bun run typecheck + - run: bun run build + - name: Assert no service socket exists + run: test ! -e "$HERDR_SOCKET" + - run: bun --config=bunfig.websocket.toml test --isolate src/lib server/updater.test.ts + - run: bun run test:websocket + - run: bun run test:websocket:mutations + - name: Upload websocket failure artifacts + if: failure() + uses: actions/upload-artifact@v4 + with: + name: websocket-test-artifacts + path: ${{ runner.temp }}/herdr-websocket-*/** + if-no-files-found: warn + retention-days: 7 diff --git a/README.md b/README.md index 9ca6c7b827..e233a162b5 100644 --- a/README.md +++ b/README.md @@ -227,6 +227,21 @@ bun run test:ssh # remote-PC integration over SSH Tests run against a herdr session of their own, `herdr-web-ui-test`: the first run starts a headless `herdr --session herdr-web-ui-test server` and later runs reuse it, so test workspaces never show in the herdr you work in (`scripts/test-herdr.ts`). Stop it with `herdr --session herdr-web-ui-test server stop`. `HERDR_TEST_SESSION` picks another name, and `HERDR_TEST_LIVE=1` runs against `HERDR_SOCKET` or your default session as before. +### Local WebSocket harness suites + +Run the isolated WebSocket harness suites with: + +```bash +bun run test:websocket +bun run test:websocket:mutations +``` + +These package scripts run `bun --config=bunfig.websocket.toml test --isolate` on the WebSocket test files listed in `package.json`. The dedicated config sets `[test] preload = []`, so the root `bunfig.toml` default preload, `scripts/test-herdr.ts`, is not loaded for either suite. + +Harness state and mutation overlays are created under a unique temporary `SODAM_WS_RUN_ROOT`, never under production `server/`. Pull-request CI runs the same two suites on Ubuntu; remote GitHub Actions runs happen only after a separately authorized push or PR. + +The Herdr RPC and PTY are fakes. The real server (`createServer`, `Bun.serve`) and real WebSocket client are under test. These suites do not claim coverage of `server/output.contract.test.ts`, the browser wrapper, an external Herdr workspace, or load testing. + To release, bump `version` in `package.json` and `herdr-plugin.toml`, move the `Unreleased` notes in [CHANGELOG.md](CHANGELOG.md) under the new version, commit, then push `main` together with a `vX.Y.Z` tag (`git push origin main vX.Y.Z`). The release workflow checks that the three versions agree, builds, tests and publishes the GitHub release; installs pick it up within five minutes. Remote-PC runtime bundles are released separately by pushing `remote-vN` after raising `REMOTE_BUNDLE_VERSION` in `shared/machines.ts`. Browser checks look for Chrome at `/opt/google/chrome/chrome`; set `CHROME_PATH` otherwise. After a herdr upgrade, refresh the generated wire types with `bun run generate:types --refresh` (and `--check` to verify). From e28b635ecad0b65e8d39cbcda1f5bc64fc718d37 Mon Sep 17 00:00:00 2001 From: pjhsion Date: Sat, 10 Oct 2026 14:42:14 +0900 Subject: [PATCH 4/4] Add websocket harness workflow alongside upstream CI Restores the least-privilege websocket job dropped when upstream ci.yml was adopted in the merge: typecheck, build, no-service assertion, the legacy src/lib suite, test:websocket and the mutation probes on ubuntu, with failure-only artifacts. actionlint clean; action pins match upstream (checkout@v7, setup-bun@v2, bun 1.4.2). --- .github/workflows/websocket.yml | 43 +++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 .github/workflows/websocket.yml diff --git a/.github/workflows/websocket.yml b/.github/workflows/websocket.yml new file mode 100644 index 0000000000..3b002964bc --- /dev/null +++ b/.github/workflows/websocket.yml @@ -0,0 +1,43 @@ +name: Websocket harness +on: + pull_request: + branches: [main] + push: + branches: [main] +permissions: + contents: read +concurrency: + group: websocket-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} +jobs: + websocket: + name: Websocket harness (no service) + runs-on: ubuntu-24.04 + timeout-minutes: 15 + steps: + - uses: actions/checkout@v7 + with: + ref: ${{ github.sha }} + - uses: oven-sh/setup-bun@v2 + with: + bun-version: 1.4.2 + - name: Isolate the websocket run from any herdr service + run: | + echo "HERDR_SOCKET=$RUNNER_TEMP/herdr-websocket-no-service.sock" >> "$GITHUB_ENV" + echo "SODAM_WS_RUN_ROOT=$RUNNER_TEMP/herdr-websocket-${{ github.run_id }}-${{ github.run_attempt }}" >> "$GITHUB_ENV" + - run: bun install --frozen-lockfile + - run: bun run typecheck + - run: bun run build + - name: Assert no service socket exists + run: test ! -e "$HERDR_SOCKET" + - run: bun --config=bunfig.websocket.toml test --isolate src/lib server/updater.test.ts + - run: bun run test:websocket + - run: bun run test:websocket:mutations + - name: Upload websocket failure artifacts + if: failure() + uses: actions/upload-artifact@v4 + with: + name: websocket-test-artifacts-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/herdr-websocket-*/** + if-no-files-found: warn + retention-days: 7