From be4fecdad502c3eeaf13b0f7f7ea5d415e4bc381 Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Fri, 19 Jun 2026 19:45:29 -0400 Subject: [PATCH 1/6] COD-133 fix header WS status indicator + typing lag from v1.1.15 merge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The upstream v1.1.15 merge spliced upstream's transport-object indicator body onto local's _connectionStatus-based _updateConnectionIndicator() without defining `transport`, so every transport.* reference threw ReferenceError on any queued state. That hid the "WS" status and, because _reliableSend() updates the indicator before _drainSession(), made every keystroke skip immediate delivery (input flushed only on the 2s sweep = typing lag). - Rewrite _updateConnectionIndicator() to show the terminal WebSocket transport from _wsState (WS / HTTP / WS… / Offline), falling back to the SSE _connectionStatus only on the idle dashboard. - Only annotate a backlog (· N queued) above 4 bytes so normal typing no longer flickers "sending 1B" on each key press. - test/connection-indicator.test.ts (new): transport labels, the >4B threshold, an exhaustive never-throws guard for the ReferenceError, and the _reliableSend -> _drainSession invariant (typing-lag guard). - test/input-send-order.test.ts: reconcile to local's durable input layer (the prior coalescing-fallback tests had been failing since 1255e28). --- src/web/public/app.js | 78 ++++++++---- test/connection-indicator.test.ts | 204 ++++++++++++++++++++++++++++++ test/input-send-order.test.ts | 159 +++++++++++++++++++++++ 3 files changed, 419 insertions(+), 22 deletions(-) create mode 100644 test/connection-indicator.test.ts create mode 100644 test/input-send-order.test.ts diff --git a/src/web/public/app.js b/src/web/public/app.js index c63f12be..169bfb19 100644 --- a/src/web/public/app.js +++ b/src/web/public/app.js @@ -2358,34 +2358,68 @@ class CodemanApp { const text = this.$('connectionText'); if (!indicator || !dot || !text) return; - const status = this._connectionStatus; - - // While the connection is healthy, never surface the input queue. With the - // reliable-delivery layer every keystroke is briefly "pending" until its ACK - // lands a few ms later — showing that flashed "Sending 1B…" on every single - // character. The indicator is only meaningful for an actual connection - // problem (reconnecting / offline), where the queued byte count reassures - // the user their typing is safely buffered and will be sent. - if (status === 'connected' || status === 'connecting') { - indicator.style.display = 'none'; + const { bytes: totalBytes, count } = this._pendingBytes(); + const hasQueue = count > 0; + // Only surface a backlog once it's more than a few bytes. A single keystroke + // (1B) ACKs in milliseconds, so without this the label flickered "sending 1B" + // on every key press. Above this threshold means input is genuinely backing up. + const BACKLOG_HINT_BYTES = 4; + const showBacklog = totalBytes > BACKLOG_HINT_BYTES; + const formatBytes = (b) => (b < 1024 ? `${b}B` : `${(b / 1024).toFixed(1)}KB`); + const queuedSuffix = showBacklog ? ` · ${formatBytes(totalBytes)} queued` : ''; + + // Hard offline (browser reports no network) dominates everything. + if (!this.isOnline || this._connectionStatus === 'offline') { + indicator.style.display = 'flex'; + dot.className = 'connection-dot offline'; + text.textContent = showBacklog ? `Offline (${formatBytes(totalBytes)} queued)` : 'Offline'; + indicator.title = 'No network connection'; return; } - const { bytes: totalBytes, count } = this._pendingBytes(); - const hasQueue = count > 0; - indicator.style.display = 'flex'; - dot.className = 'connection-dot'; + // With an active terminal, show its transport (WebSocket vs HTTP fallback). + if (this.activeSessionId) { + let cls, label, detail; + switch (this._wsState) { + case 'connected': + cls = 'connected'; label = 'WS'; detail = 'Terminal connected over WebSocket'; + break; + case 'fallback': + cls = 'fallback'; label = 'HTTP'; detail = 'WebSocket unavailable — input sent over HTTP'; + break; + case 'reconnecting': + cls = 'reconnecting'; label = 'WS…'; detail = 'Reconnecting WebSocket'; + break; + case 'connecting': + default: + cls = 'reconnecting'; label = 'WS…'; detail = 'Connecting WebSocket'; + break; + } + indicator.style.display = 'flex'; + dot.className = `connection-dot ${cls}`; + text.textContent = `${label}${queuedSuffix}`; + indicator.title = detail; + return; + } - const formatBytes = (b) => (b < 1024 ? `${b}B` : `${(b / 1024).toFixed(1)}KB`); + // No active terminal — reflect the SSE event stream only when it needs attention. + if (this._connectionStatus === 'reconnecting' || this._connectionStatus === 'disconnected') { + indicator.style.display = 'flex'; + dot.className = 'connection-dot reconnecting'; + text.textContent = showBacklog ? `Reconnecting (${formatBytes(totalBytes)} queued)` : 'Reconnecting...'; + indicator.title = 'Reconnecting to server'; + return; + } - if (status === 'reconnecting') { - dot.classList.add('reconnecting'); - text.textContent = hasQueue ? `Reconnecting (${formatBytes(totalBytes)} queued)` : 'Reconnecting...'; - } else { - // Offline or disconnected - dot.classList.add('offline'); - text.textContent = hasQueue ? `Offline (${formatBytes(totalBytes)} queued)` : 'Offline'; + // Idle dashboard, healthy stream — hide unless input is genuinely queued. + if (!hasQueue) { + indicator.style.display = 'none'; + return; } + indicator.style.display = 'flex'; + dot.className = 'connection-dot draining'; + text.textContent = showBacklog ? `Sending ${formatBytes(totalBytes)}...` : 'Sending...'; + indicator.title = 'Delivering queued input'; } setupOnlineDetection() { diff --git a/test/connection-indicator.test.ts b/test/connection-indicator.test.ts new file mode 100644 index 00000000..3f8bcc66 --- /dev/null +++ b/test/connection-indicator.test.ts @@ -0,0 +1,204 @@ +/** + * @fileoverview Regression tests for the header connection indicator + * (`CodemanApp._updateConnectionIndicator`) and the invariant that updating it + * never aborts durable input delivery. + * + * Guards two regressions from the upstream v1.1.15 merge (fixed in COD-133): + * 1. The indicator body referenced an undefined `transport` object, throwing + * `ReferenceError: transport is not defined` on every queued state. Because + * `_reliableSend()` calls `_updateConnectionIndicator()` *before* + * `_drainSession()`, the throw skipped immediate delivery on every keystroke + * → input only flushed on the 2s sweep (large typing lag) and the indicator + * never rendered (missing "WS" status). + * 2. The restored body only read the SSE `_connectionStatus`, so it never + * surfaced the terminal WebSocket transport ("WS" / "HTTP"), and it flashed + * "sending 1B" on every single keystroke. + * + * Loaded via `vm` with a stubbed context (no jsdom — see input-send-order.test.ts). + */ +import { readFileSync } from 'node:fs'; +import { performance } from 'node:perf_hooks'; +import { resolve } from 'node:path'; +import vm from 'node:vm'; +import { describe, expect, it, vi } from 'vitest'; + +function loadCodemanAppClass() { + const constants = readFileSync(resolve(import.meta.dirname, '../src/web/public/constants.js'), 'utf8'); + const source = readFileSync(resolve(import.meta.dirname, '../src/web/public/app.js'), 'utf8'); + const context = vm.createContext({ + console, + performance, + setInterval: vi.fn(), + clearInterval: vi.fn(), + setTimeout, + clearTimeout, + requestAnimationFrame: vi.fn(), + HTMLCanvasElement: class HTMLCanvasElement {}, + fetch: (...args: Parameters) => global.fetch(...args), + document: { addEventListener: vi.fn() }, + localStorage: { + length: 0, + key: vi.fn(), + getItem: vi.fn(), + setItem: vi.fn(), + removeItem: vi.fn(), + }, + window: { addEventListener: vi.fn(), removeEventListener: vi.fn() }, + MobileDetection: {}, + }); + vm.runInContext(`${constants}\n${source}\nglobalThis.__CodemanApp = CodemanApp;`, context); + return (context as { __CodemanApp: new () => unknown }).__CodemanApp; +} + +const CodemanApp = loadCodemanAppClass(); + +function fakeElement() { + return { style: { display: '' }, title: '', textContent: '', className: '' }; +} + +type Indicator = { + $: (id: string) => unknown; + _pendingDeliveries: Map>; + _connectionStatus: string; + _wsState: string; + activeSessionId: string | null; + isOnline: boolean; + _updateConnectionIndicator: () => void; +}; + +function makeApp(overrides: Partial & { queuedBytes?: number } = {}) { + const app = Object.create((CodemanApp as { prototype: object }).prototype) as Indicator & { + queuedBytes?: number; + }; + const els: Record> = { + connectionIndicator: fakeElement(), + connectionDot: fakeElement(), + connectionText: fakeElement(), + }; + app.$ = (id: string) => els[id]; + app._pendingDeliveries = new Map(); + app._connectionStatus = 'connected'; + app._wsState = 'disconnected'; + app.activeSessionId = null; + app.isOnline = true; + Object.assign(app, overrides); + const queued = overrides.queuedBytes ?? 0; + if (queued > 0) { + app._pendingDeliveries.set('s1', [{ seq: 1, data: 'x'.repeat(queued) }]); + } + return { app, els }; +} + +describe('connection indicator — transport display', () => { + it('shows "WS" with a connected dot when the terminal WebSocket is open', () => { + const { app, els } = makeApp({ activeSessionId: 's1', _wsState: 'connected' }); + app._updateConnectionIndicator(); + expect(els.connectionIndicator.style.display).toBe('flex'); + expect(els.connectionText.textContent).toBe('WS'); + expect(els.connectionDot.className).toContain('connected'); + }); + + it('shows "HTTP" with a fallback dot when the socket dropped to HTTP POST', () => { + const { app, els } = makeApp({ activeSessionId: 's1', _wsState: 'fallback' }); + app._updateConnectionIndicator(); + expect(els.connectionText.textContent).toBe('HTTP'); + expect(els.connectionDot.className).toContain('fallback'); + }); + + it('shows "WS…" while connecting or reconnecting the socket', () => { + for (const state of ['connecting', 'reconnecting']) { + const { app, els } = makeApp({ activeSessionId: 's1', _wsState: state }); + app._updateConnectionIndicator(); + expect(els.connectionText.textContent).toBe('WS…'); + expect(els.connectionDot.className).toContain('reconnecting'); + } + }); + + it('shows "Offline" when the browser reports no network', () => { + const { app, els } = makeApp({ activeSessionId: 's1', _wsState: 'connected', isOnline: false }); + app._updateConnectionIndicator(); + expect(els.connectionText.textContent).toBe('Offline'); + expect(els.connectionDot.className).toContain('offline'); + }); + + it('hides on an idle dashboard (no active session, healthy stream, no queue)', () => { + const { app, els } = makeApp({ activeSessionId: null, _connectionStatus: 'connected' }); + app._updateConnectionIndicator(); + expect(els.connectionIndicator.style.display).toBe('none'); + }); + + it('surfaces SSE reconnecting on the dashboard when there is no active terminal', () => { + const { app, els } = makeApp({ activeSessionId: null, _connectionStatus: 'reconnecting' }); + app._updateConnectionIndicator(); + expect(els.connectionText.textContent).toContain('Reconnecting'); + expect(els.connectionDot.className).toContain('reconnecting'); + }); +}); + +describe('connection indicator — keystroke backlog threshold', () => { + it('does NOT annotate a single-keystroke (1B) queue — no "sending 1B" flicker', () => { + const { app, els } = makeApp({ activeSessionId: 's1', _wsState: 'connected', queuedBytes: 1 }); + app._updateConnectionIndicator(); + expect(els.connectionText.textContent).toBe('WS'); + expect(els.connectionText.textContent).not.toMatch(/queued/); + }); + + it('does NOT annotate at the 4B threshold boundary', () => { + const { app, els } = makeApp({ activeSessionId: 's1', _wsState: 'connected', queuedBytes: 4 }); + app._updateConnectionIndicator(); + expect(els.connectionText.textContent).toBe('WS'); + }); + + it('annotates a genuine backlog (>4B) with a queued byte count', () => { + const { app, els } = makeApp({ activeSessionId: 's1', _wsState: 'connected', queuedBytes: 40 }); + app._updateConnectionIndicator(); + expect(els.connectionText.textContent).toMatch(/^WS · 40B queued$/); + }); +}); + +describe('connection indicator — never throws (the ReferenceError regression)', () => { + it('renders every transport × stream × queue combination without throwing', () => { + const wsStates = ['disconnected', 'connecting', 'connected', 'reconnecting', 'fallback']; + const sseStates = ['connected', 'connecting', 'reconnecting', 'disconnected', 'offline']; + for (const ws of wsStates) { + for (const sse of sseStates) { + for (const active of ['s1', null] as const) { + for (const queuedBytes of [0, 1, 4, 200]) { + for (const isOnline of [true, false]) { + const { app } = makeApp({ + activeSessionId: active, + _wsState: ws, + _connectionStatus: sse, + isOnline, + queuedBytes, + }); + expect(() => app._updateConnectionIndicator()).not.toThrow(); + } + } + } + } + } + }); +}); + +describe('durable input delivery is not aborted by the indicator (typing-lag regression)', () => { + it('_reliableSend reaches _drainSession after updating the indicator', () => { + const { app } = makeApp({ activeSessionId: 's1', _wsState: 'connected' }); + const a = app as unknown as { + _seqCounters: Map; + _persistReliableState: () => void; + _drainSession: (id: string) => void; + _reliableSend: (id: string, data: string, useMux: boolean) => void; + }; + a._seqCounters = new Map(); + a._persistReliableState = vi.fn(); + const drain = vi.fn(); + a._drainSession = drain; + + // A single keystroke. The indicator runs first; if it throws, drain is skipped. + a._reliableSend('s1', 'x', false); + + expect(drain).toHaveBeenCalledWith('s1'); + expect(app._pendingDeliveries.get('s1')).toHaveLength(1); + }); +}); diff --git a/test/input-send-order.test.ts b/test/input-send-order.test.ts new file mode 100644 index 00000000..cc7c6294 --- /dev/null +++ b/test/input-send-order.test.ts @@ -0,0 +1,159 @@ +/** + * @fileoverview Input dispatch ordering for the durable, acknowledged delivery + * layer (`CodemanApp._sendInputAsync` → `_reliableSend` → `_drainSession`). + * + * Local replaced the upstream best-effort "coalescing fallback queue" with the + * durable per-(clientId, seq) layer in commit 1255e28 (docs/reliable-input- + * delivery.md). This suite verifies the client-side ordering guarantees of that + * layer: each input is a distinct seq-tagged frame, delivered in order over the + * WebSocket when open, serialized over HTTP POST when not, and only dropped on a + * server ACK (HTTP 2xx). Exactly-once application is covered server-side in + * test/reliable-input-dedup.test.ts; the header transport indicator ("WS"/"HTTP") + * is covered in test/connection-indicator.test.ts. + * + * Loaded via `vm` with a stubbed context (no jsdom). + */ +import { readFileSync } from 'node:fs'; +import { performance } from 'node:perf_hooks'; +import { resolve } from 'node:path'; +import vm from 'node:vm'; +import { describe, expect, it, vi } from 'vitest'; + +function loadCodemanAppClass() { + const constants = readFileSync(resolve(import.meta.dirname, '../src/web/public/constants.js'), 'utf8'); + const source = readFileSync(resolve(import.meta.dirname, '../src/web/public/app.js'), 'utf8'); + const context = vm.createContext({ + console, + performance, + setInterval: vi.fn(), + clearInterval: vi.fn(), + setTimeout, + clearTimeout, + requestAnimationFrame: vi.fn(), + HTMLCanvasElement: class HTMLCanvasElement {}, + WebSocket: { OPEN: 1 }, + fetch: (...args: Parameters) => global.fetch(...args), + document: { addEventListener: vi.fn() }, + localStorage: { + length: 0, + key: vi.fn(), + getItem: vi.fn(), + setItem: vi.fn(), + removeItem: vi.fn(), + }, + window: { addEventListener: vi.fn(), removeEventListener: vi.fn() }, + MobileDetection: {}, + }); + vm.runInContext(`${constants}\n${source}\nglobalThis.__CodemanApp = CodemanApp;`, context); + return (context as { __CodemanApp: new () => unknown }).__CodemanApp; +} + +const CodemanApp = loadCodemanAppClass(); + +async function waitForCalls(calls: unknown[], count: number) { + for (let i = 0; i < 50; i++) { + if (calls.length >= count) return; + await new Promise((r) => setTimeout(r, 0)); + } +} + +type Frame = { t: string; d: string; seq: number; cid: string }; +type PostBody = { input: string; seq: number; clientId: string }; + +type App = { + _sendInputAsync: (sessionId: string, input: string, opts?: { useMux?: boolean }) => void; + _pendingDeliveries: Map>; + _ws: { readyState: number; send: (data: string) => void } | null; + _wsSessionId: string | null; + activeSessionId: string | null; +}; + +function makeApp(): App { + const app = Object.create((CodemanApp as { prototype: object }).prototype) as App & Record; + app._clientId = 'c-test'; + app._seqCounters = new Map(); + app._pendingDeliveries = new Map(); + app._postDraining = new Set(); + app._persistReliableState = vi.fn(); + app._persistReliableNow = vi.fn(); + app._updateConnectionIndicator = vi.fn(); + app.clearPendingHooks = vi.fn(); + app.activeSessionId = 'session-1'; + app.isOnline = true; + app._connectionStatus = 'connected'; + app._ws = null; + app._wsSessionId = null; + return app as unknown as App; +} + +describe('durable input delivery — send ordering', () => { + it('delivers rapid input as distinct ordered seq frames over an open WebSocket (no coalescing)', () => { + const app = makeApp(); + const frames: Frame[] = []; + app._ws = { readyState: 1, send: (d: string) => frames.push(JSON.parse(d)) }; + app._wsSessionId = 'session-1'; + + app._sendInputAsync('session-1', 'a'); + app._sendInputAsync('session-1', 'b'); + app._sendInputAsync('session-1', 'c'); + + // Each keystroke is its own frame, in seq order — never merged into "abc". + expect(frames.map((f) => f.d)).toEqual(['a', 'b', 'c']); + expect(frames.map((f) => f.seq)).toEqual([1, 2, 3]); + expect(frames.every((f) => f.t === 'i' && f.cid === 'c-test')).toBe(true); + }); + + it('POSTs queued input one frame at a time in seq order when no socket is open', async () => { + const app = makeApp(); + const calls: PostBody[] = []; + const completions: Array<() => void> = []; + global.fetch = vi.fn(async (_url, init) => { + calls.push(JSON.parse(String(init?.body)) as PostBody); + await new Promise((r) => completions.push(r)); + return new Response('{}', { status: 200 }); + }); + + app._sendInputAsync('session-1', 'a'); + app._sendInputAsync('session-1', 'b'); + + // Serialized: only the first frame is in flight until its 2xx ACK lands. + await waitForCalls(calls, 1); + expect(calls.map((c) => c.input)).toEqual(['a']); + + completions.shift()?.(); // ACK 'a' + await waitForCalls(calls, 2); + expect(calls.map((c) => c.input)).toEqual(['a', 'b']); + expect(calls.map((c) => c.seq)).toEqual([1, 2]); + + completions.shift()?.(); + await waitForCalls(calls, 2); + }); + + it('leaves a frame queued (unacked) when HTTP delivery fails', async () => { + const app = makeApp(); + const calls: PostBody[] = []; + global.fetch = vi.fn(async (_url, init) => { + calls.push(JSON.parse(String(init?.body)) as PostBody); + return new Response('busy', { status: 503 }); + }); + + app._sendInputAsync('session-1', 'a'); + await waitForCalls(calls, 1); + await new Promise((r) => setTimeout(r, 0)); + + // 5xx is not an ACK — the frame must survive for the sweep/reconnect to retry. + expect(app._pendingDeliveries.get('session-1')).toHaveLength(1); + expect(app._pendingDeliveries.get('session-1')?.[0].data).toBe('a'); + }); + + it('drops a frame addressed to a vanished session (404) instead of retrying forever', async () => { + const app = makeApp(); + global.fetch = vi.fn(async () => new Response('gone', { status: 404 })); + + app._sendInputAsync('session-1', 'a'); + await new Promise((r) => setTimeout(r, 0)); + await new Promise((r) => setTimeout(r, 0)); + + expect(app._pendingDeliveries.get('session-1')).toBeUndefined(); + }); +}); From 68fd6e8962de160950da46b8d08bf6d5df112117 Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Fri, 19 Jun 2026 21:32:42 -0400 Subject: [PATCH 2/6] COD-134 fix WS flap loop (undefined onopen call) + reconnect resilience + logging Root cause of the WS->HTTP->WS flapping: the v1.1.15 input-delivery merge left a call to the now-undefined _flushHttpFallbackQueuesViaWs() in ws.onopen, so every (re)connect threw a TypeError BEFORE _onWsReady() ran -- durable input was never re-flushed over the fresh socket, the 2s redeliver sweep then saw stale unacked frames and force-closed the socket, reconnect, throw again: a self-sustaining flap loop. Remove the dead call (_onWsReady, 10 lines below, is its replacement). Resilience + observability: - Pure CodemanWsReconnect.plan(code, attempt) (constants.js, TDD, 6 tests): <4004 -> fast reconnect (immediate jittered first retry, faster backoff); 4008/unknown->=4004 -> bounded retry-fallback (HTTP no longer sticks until a tab switch); 4004/4009 -> give up (session gone). Wired into onclose. - Redeliver sweep force-closes only a SILENT socket (no recent recv), not one actively delivering output/ACKs -- stops self-inflicted flaps while typing. - Client logs WS close code/reason to crash-diag; server logs [ws] open/close/terminate/4008 (console -> journald; Fastify runs logger:false). Verified: 6/6 unit, tsc 0, frontend-syntax + prettier clean, build; beta WS reaches connected with zero console errors (onopen TypeError gone), _wsLastRecvAt tracked, server [ws] lines emit. --- src/web/public/app.js | 76 ++++++++++++++++++++++++++------ src/web/public/constants.js | 62 ++++++++++++++++++++++++++ src/web/routes/ws-routes.ts | 15 +++++-- test/ws-reconnect-plan.test.ts | 80 ++++++++++++++++++++++++++++++++++ 4 files changed, 217 insertions(+), 16 deletions(-) create mode 100644 test/ws-reconnect-plan.test.ts diff --git a/src/web/public/app.js b/src/web/public/app.js index 169bfb19..2f1472f3 100644 --- a/src/web/public/app.js +++ b/src/web/public/app.js @@ -435,6 +435,11 @@ class CodemanApp { this._ws = null; // WebSocket instance for active session this._wsSessionId = null; // Session ID the WS is connected to this._wsReady = false; // True when WS is open and ready for I/O + this._wsState = 'disconnected'; // connecting | connected | reconnecting | fallback | disconnected + this._wsLastClose = null; // { code, reason, at } for transport diagnostics + this._wsLastRecvAt = 0; // ms timestamp of the last frame received on the active WS + this._wsInputSendCount = 0; + this._httpFallbackSendCount = 0; // Terminal write batching with DEC 2026 sync support this.pendingWrites = []; @@ -1976,6 +1981,7 @@ class CodemanApp { if (this._ws === ws) { this._wsReady = true; this._wsReconnectAttempts = 0; + this._updateConnectionIndicator(); // Send a typed resize over the fresh socket: syncs PTY dims after // (re)connects AND registers the desktop sizing claim server-side — // selectSession's earlier resizes ran before this WS existed, so they @@ -1990,6 +1996,9 @@ class CodemanApp { ws.onmessage = (event) => { if (this._ws !== ws) return; + // Mark the socket as alive on every received frame (output, ACK, etc.) so + // the redeliver sweep only force-closes a genuinely silent connection. + this._wsLastRecvAt = Date.now(); try { const msg = JSON.parse(event.data); if (msg.t === 'o') { @@ -2016,18 +2025,53 @@ class CodemanApp { this._wsReady = false; this._stopMobileResizeRetry(); - // Reconnect on unexpected close (server restart, network blip, ping timeout). - // Don't reconnect if we intentionally disconnected (_disconnectWs nulls onclose) - // or if the server rejected the session (4004=not found, 4008=too many, 4009=terminated). - if (event.code < 4004 && this.activeSessionId === sessionId) { - const delay = Math.min(1000 * Math.pow(2, this._wsReconnectAttempts || 0), 10000); - this._wsReconnectAttempts = (this._wsReconnectAttempts || 0) + 1; - this._wsReconnectTimer = setTimeout(() => { - this._wsReconnectTimer = null; - if (this.activeSessionId === sessionId) { - this._connectWs(sessionId); - } - }, delay); + // Decide what to do next from the close code + how many consecutive + // reconnects we've already made (pure policy in constants.js): + // reconnect → transient (server restart, network blip, ping timeout); + // schedule a backoff retry while this session stays active. + // retry-fallback → too-many-connections / unknown rejection; show the HTTP + // fallback but keep retrying so we return to WS when it clears. + // give-up → 4004 (not found) / 4009 (terminated); the session is gone. + // _disconnectWs() nulls onclose for intentional disconnects, so we never land here for those. + const plan = window.CodemanWsReconnect.plan(event.code, this._wsReconnectAttempts || 0); + _crashDiag.log( + `WS CLOSE code=${event.code} reason=${event.reason || ''} action=${plan.action} attempts=${this._wsReconnectAttempts || 0}` + ); + + const stillActive = this.activeSessionId === sessionId; + if (plan.action === 'give-up') { + this._wsState = stillActive ? 'fallback' : 'disconnected'; + this._updateConnectionIndicator(); + } else if (plan.action === 'reconnect') { + if (stillActive) { + this._wsState = 'reconnecting'; + this._updateConnectionIndicator(); + const delay = plan.delayMs + Math.floor(Math.random() * 250); // jitter to de-sync herds + this._wsReconnectAttempts = (this._wsReconnectAttempts || 0) + 1; + this._wsReconnectTimer = setTimeout(() => { + this._wsReconnectTimer = null; + if (this.activeSessionId === sessionId) { + this._connectWs(sessionId); + } + }, delay); + } else { + this._wsState = 'disconnected'; + this._updateConnectionIndicator(); + } + } else { + // retry-fallback: surface the HTTP fallback, but keep trying on a bounded + // timer so the transport returns to WS once the transient condition clears. + this._wsState = stillActive ? 'fallback' : 'disconnected'; + this._updateConnectionIndicator(); + if (stillActive) { + this._wsReconnectAttempts = (this._wsReconnectAttempts || 0) + 1; + this._wsReconnectTimer = setTimeout(() => { + this._wsReconnectTimer = null; + if (this.activeSessionId === sessionId) { + this._connectWs(sessionId); + } + }, plan.delayMs); + } } }; @@ -2234,7 +2278,13 @@ class CodemanApp { this._ws && this._ws.readyState === WebSocket.OPEN && this._wsSessionId === sessionId; if (isActiveWs) { const oldest = list[0]; - if (oldest && oldest.sentAt && Date.now() - oldest.sentAt > this._reliableAckTimeoutMs) { + // Only tear the socket down when the oldest unacked frame is stale AND the + // socket has been silent for the timeout: a connection still delivering + // output/ACKs is alive (the ACK is just behind), so force-closing it would + // cause needless WS↔HTTP flapping. A truly half-open socket goes quiet. + const stale = oldest && oldest.sentAt && Date.now() - oldest.sentAt > this._reliableAckTimeoutMs; + const silent = Date.now() - this._wsLastRecvAt > this._reliableAckTimeoutMs; + if (stale && silent) { try { this._ws.close(); // half-open: never recovers on its own — force reconnect } catch { diff --git a/src/web/public/constants.js b/src/web/public/constants.js index 33b74246..be01ac6a 100644 --- a/src/web/public/constants.js +++ b/src/web/public/constants.js @@ -126,12 +126,74 @@ function shouldAutoWrapTabs(input) { return scrollWidth > clientWidth + 1; } +// COD-122: Monitor "Tmux Sessions" row label policy. A row carries up to three +// identifiers — the tab name (user-facing L1 label), the tmux session name +// (e.g. "codeman-1fac7304"), and the agent (Claude/Codex) session id. The tab name +// must win as the primary label; the other two are secondary metadata shown only +// when known and not redundant with the primary, so the row degrades gracefully +// when identifiers are missing. +function resolveMonitorRowLabels(input) { + const clean = (v) => (typeof v === 'string' ? v.trim() : ''); + const tabName = clean(input && input.tabName); + const storedName = clean(input && input.storedName); + const muxName = clean(input && input.muxName); + const agentId = clean(input && input.agentSessionId); + const sessionId = clean(input && input.sessionId); + + // Tab name is the L1 label; fall back through the persisted mux name → tmux name + // → a generic placeholder so the primary is never blank. + const primary = tabName || storedName || muxName || 'session'; + + const secondaries = []; + // tmux name — skip when it's already serving as the primary fallback (no dup). + if (muxName && muxName !== primary) { + secondaries.push({ kind: 'tmux', label: muxName }); + } + // agent session id — show only when genuinely known: non-empty, not the codeman + // sessionId placeholder (claudeSessionId is seeded with the session id until a + // real agent message arrives), and not redundant with the primary label. + if (agentId && agentId !== sessionId && agentId !== primary) { + secondaries.push({ kind: 'agent', label: agentId }); + } + return { primary, secondaries }; +} + +// COD-134 — Terminal WebSocket reconnect policy. +// +// Decide what to do after a terminal WebSocket closes, given the close `code` +// and `attempt` (0-based count of consecutive reconnects already made): +// - transient closes (code < 4004: 1000/1001/1005/1006/etc.) → 'reconnect' +// with exponential backoff (0 on the first attempt; the caller adds jitter), +// 250ms → 500 → 1000 → ... capped at 10s. +// - 4004 (session not found) / 4009 (session terminated) → 'give-up': the +// session is gone, retrying only wastes connections. +// - 4008 (too many connections) and any other code >= 4004 → 'retry-fallback': +// show the HTTP fallback but keep retrying on a bounded 5s timer so the +// transport returns to WS once the transient condition clears (un-stick). +// Pure: no DOM, no side effects. +function planWsReconnect(code, attempt) { + if (code === 4004 || code === 4009) { + return { action: 'give-up', delayMs: 0 }; + } + if (code >= 4004) { + return { action: 'retry-fallback', delayMs: 5000 }; + } + const delayMs = attempt <= 0 ? 0 : Math.min(250 * Math.pow(2, attempt - 1), 10000); + return { action: 'reconnect', delayMs }; +} + if (typeof window !== 'undefined') { window.WEBGL_FALLBACK = WEBGL_FALLBACK; window.evaluateWebGLLongTaskTrip = evaluateWebGLLongTaskTrip; window.CodemanTabOverflow = { shouldAutoWrapTabs, }; + window.CodemanMonitorLabels = { + resolveMonitorRowLabels, + }; + window.CodemanWsReconnect = { + plan: planWsReconnect, + }; } // Scheduler API — prioritize terminal writes over background UI updates. diff --git a/src/web/routes/ws-routes.ts b/src/web/routes/ws-routes.ts index 8aaa1a21..17d7ee3e 100644 --- a/src/web/routes/ws-routes.ts +++ b/src/web/routes/ws-routes.ts @@ -83,13 +83,19 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost return; } + // Structured transport logging — surfaces WS open/close/timeout churn so the + // tunnel-flap behavior (COD-134) is observable in the server logs. Fastify is + // configured logger:false, so we log via console (→ journald under systemd). + // Enforce per-session connection limit const currentCount = sessionWsCount.get(id) ?? 0; if (currentCount >= MAX_WS_PER_SESSION) { + console.warn('[ws] terminal rejected: too many connections', { sessionId: id, wsCount: currentCount }); socket.close(4008, 'Too many connections'); return; } sessionWsCount.set(id, currentCount + 1); + console.info('[ws] terminal open', { sessionId: id, wsCount: currentCount + 1 }); // Swallow socket errors — cleanup happens in 'close' socket.on('error', () => {}); @@ -228,11 +234,12 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost if (socket.readyState !== 1) return; socket.ping(); pongTimeout = setTimeout(() => { + console.warn('[ws] terminal ping timeout — terminating', { sessionId: id }); socket.terminate(); }, WS_PONG_TIMEOUT_MS); }, WS_PING_INTERVAL_MS); - socket.on('close', () => { + socket.on('close', (code: number, reason: Buffer) => { clearInterval(pingInterval); if (pongTimeout) clearTimeout(pongTimeout); if (batchTimer) clearTimeout(batchTimer); @@ -245,11 +252,13 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost // Decrement per-session connection count const count = sessionWsCount.get(id) ?? 1; - if (count <= 1) { + const remaining = count <= 1 ? 0 : count - 1; + if (remaining === 0) { sessionWsCount.delete(id); } else { - sessionWsCount.set(id, count - 1); + sessionWsCount.set(id, remaining); } + console.info('[ws] terminal close', { sessionId: id, code, reason: String(reason), wsCount: remaining }); }); }); } diff --git a/test/ws-reconnect-plan.test.ts b/test/ws-reconnect-plan.test.ts new file mode 100644 index 00000000..dbbfd37f --- /dev/null +++ b/test/ws-reconnect-plan.test.ts @@ -0,0 +1,80 @@ +/** + * COD-134 — Terminal WebSocket reconnect policy. + * + * `CodemanWsReconnect.plan(code, attempt)` is the pure decision behind the + * client WS `onclose` handler in app.js: given a WebSocket close code and the + * number of consecutive reconnects already attempted, it returns the action to + * take (`reconnect` | `retry-fallback` | `give-up`) and a backoff delay. It is + * exposed on `window.CodemanWsReconnect` and tested here in a plain node VM + * context (no jsdom — jsdom env setup is broken on some hosts), mirroring the + * `CodemanMonitorLabels` harness in test/monitor-row-labels.test.ts. + */ +import { readFileSync } from 'node:fs'; +import { resolve } from 'node:path'; +import vm from 'node:vm'; +import { describe, expect, it } from 'vitest'; + +type Plan = { action: 'reconnect' | 'retry-fallback' | 'give-up'; delayMs: number }; + +function loadHelper() { + const context = vm.createContext({ window: {}, globalThis: {} }); + const source = readFileSync(resolve(import.meta.dirname, '../src/web/public/constants.js'), 'utf8'); + vm.runInContext(source, context, { filename: 'constants.js' }); + return (context.window as { CodemanWsReconnect: { plan: (code: number, attempt: number) => Plan } }) + .CodemanWsReconnect; +} + +describe('COD-134 WS reconnect plan policy', () => { + it('reconnects immediately on the first attempt for a transient close', () => { + const { plan } = loadHelper(); + expect(plan(1006, 0)).toEqual({ action: 'reconnect', delayMs: 0 }); + expect(plan(1000, 0).action).toBe('reconnect'); + expect(plan(1001, 0).delayMs).toBe(0); + expect(plan(1005, 0).delayMs).toBe(0); + }); + + it('grows the backoff exponentially with a 10s cap for transient closes', () => { + const { plan } = loadHelper(); + expect(plan(1006, 0).delayMs).toBe(0); + expect(plan(1006, 1).delayMs).toBe(250); + expect(plan(1006, 2).delayMs).toBe(500); + expect(plan(1006, 3).delayMs).toBe(1000); + expect(plan(1006, 4).delayMs).toBe(2000); + expect(plan(1006, 5).delayMs).toBe(4000); + expect(plan(1006, 6).delayMs).toBe(8000); + expect(plan(1006, 7).delayMs).toBe(10000); // 16000 capped to 10000 + expect(plan(1006, 8).delayMs).toBe(10000); + expect(plan(1006, 50).delayMs).toBe(10000); // stays capped no matter how many attempts + // every transient attempt is still a reconnect + for (let attempt = 0; attempt < 12; attempt++) { + expect(plan(1006, attempt).action).toBe('reconnect'); + } + }); + + it('auto-retries the fallback on a too-many-connections (4008) close', () => { + const { plan } = loadHelper(); + expect(plan(4008, 0)).toEqual({ action: 'retry-fallback', delayMs: 5000 }); + expect(plan(4008, 3)).toEqual({ action: 'retry-fallback', delayMs: 5000 }); + }); + + it('gives up on session-not-found (4004) and session-terminated (4009)', () => { + const { plan } = loadHelper(); + expect(plan(4004, 0)).toEqual({ action: 'give-up', delayMs: 0 }); + expect(plan(4004, 5)).toEqual({ action: 'give-up', delayMs: 0 }); + expect(plan(4009, 0)).toEqual({ action: 'give-up', delayMs: 0 }); + expect(plan(4009, 5)).toEqual({ action: 'give-up', delayMs: 0 }); + }); + + it('auto-retries the fallback for an unknown >=4004 code (e.g. 4010, 4005)', () => { + const { plan } = loadHelper(); + expect(plan(4010, 0)).toEqual({ action: 'retry-fallback', delayMs: 5000 }); + expect(plan(4005, 2)).toEqual({ action: 'retry-fallback', delayMs: 5000 }); + expect(plan(4500, 0)).toEqual({ action: 'retry-fallback', delayMs: 5000 }); + }); + + it('treats a sub-4004 close (e.g. 4003 Forbidden) as a transient reconnect', () => { + const { plan } = loadHelper(); + // 4003 is < 4004, so it is NOT a give-up; it follows the transient backoff. + expect(plan(4003, 0)).toEqual({ action: 'reconnect', delayMs: 0 }); + }); +}); From 20cb42d202d341c86c4289ae96e6af8d5eda0c4e Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Sat, 20 Jun 2026 09:31:14 -0400 Subject: [PATCH 3/6] COD-135 re-drive lost input ACK on a live WebSocket (durable-delivery gap) A reliable-input frame could be stranded forever if its server ACK ({t:'ia',seq}) was lost while the WebSocket kept delivering other output. _drainSession's WS fast path skips records with sentAt!==0, and after COD-134 the sweep only force-closes a *silent* socket -- so a lost ACK on an otherwise-live socket (stale && !silent) was never re-sent. _redeliverSweep now, for an active-WS session whose oldest unacked frame is stale but the socket is NOT silent, resets sentAt=0 on every stale unacked frame and lets the existing _drainSession re-drive them over the live socket (server dedups by seq). The stale && silent force-close remains the fallback for a genuinely half-open socket. Restores the exactly-once recovery guarantee without reintroducing the flap. Tests: new failing-first COD-135 cases in test/input-send-order.test.ts (re-drive on live socket; leave not-yet-stale alone; keep stale+silent force-close). 18/18 across input-send-order + reliable-input-dedup + ws-reconnect-plan; tsc 0, frontend-syntax, build all clean. --- src/web/public/app.js | 12 ++++++ test/input-send-order.test.ts | 75 +++++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+) diff --git a/src/web/public/app.js b/src/web/public/app.js index 2f1472f3..ccec5458 100644 --- a/src/web/public/app.js +++ b/src/web/public/app.js @@ -2292,6 +2292,18 @@ class CodemanApp { } continue; } + if (stale) { + // Stale but the socket is still delivering output: the ACK was lost, + // not the connection. Force-closing isn't warranted (the link is fine), + // but the fast path skips anything with sentAt!==0, so the stranded + // frame would never re-send. Reset sentAt=0 on every stale unacked + // frame so the _drainSession below re-drives them over the live socket + // (server dedups by seq, so a re-sent lost-ACK frame is harmless). + // Frames sent recently (not yet stale) are left untouched. + for (const rec of list) { + if (rec.sentAt && Date.now() - rec.sentAt > this._reliableAckTimeoutMs) rec.sentAt = 0; + } + } } this._drainSession(sessionId); } diff --git a/test/input-send-order.test.ts b/test/input-send-order.test.ts index cc7c6294..1644f37d 100644 --- a/test/input-send-order.test.ts +++ b/test/input-send-order.test.ts @@ -157,3 +157,78 @@ describe('durable input delivery — send ordering', () => { expect(app._pendingDeliveries.get('session-1')).toBeUndefined(); }); }); + +// COD-135 — durable redelivery sweep when an ACK is lost. +type RedriveApp = App & { + _redeliverSweep: () => void; + _reliableAckTimeoutMs: number; + _wsLastRecvAt: number; +}; + +describe('durable input delivery — _redeliverSweep ACK-loss recovery (COD-135)', () => { + it('re-drives a stale unacked frame over a STILL-LIVE socket (lost ACK, not silent)', () => { + const app = makeApp() as RedriveApp; + const frames: Frame[] = []; + const close = vi.fn(); + app._ws = { readyState: 1, send: (d: string) => frames.push(JSON.parse(d)), close } as never; + app._wsSessionId = 'session-1'; + app._reliableAckTimeoutMs = 4000; + + // Frame sent once over the open socket; ACK never arrives. + app._sendInputAsync('session-1', 'a'); + expect(frames.map((f) => f.d)).toEqual(['a']); + + // ACK is lost, but the socket KEEPS receiving output → it is NOT silent. + // Backdate the send so the frame is stale; keep recv timestamp fresh. + const list = app._pendingDeliveries.get('session-1')!; + list[0].sentAt = Date.now() - (app._reliableAckTimeoutMs + 1000); + app._wsLastRecvAt = Date.now(); + + app._redeliverSweep(); + + // The stale frame must be re-sent over the live socket (a second send), + // and the socket must NOT be force-closed (it's alive, just the ACK was lost). + expect(frames.map((f) => f.d)).toEqual(['a', 'a']); + expect(close).not.toHaveBeenCalled(); + expect(app._pendingDeliveries.get('session-1')).toHaveLength(1); + }); + + it('does NOT re-drive a not-yet-stale frame (sent recently)', () => { + const app = makeApp() as RedriveApp; + const frames: Frame[] = []; + const close = vi.fn(); + app._ws = { readyState: 1, send: (d: string) => frames.push(JSON.parse(d)), close } as never; + app._wsSessionId = 'session-1'; + app._reliableAckTimeoutMs = 4000; + + app._sendInputAsync('session-1', 'a'); + app._wsLastRecvAt = Date.now(); // not silent + + // sentAt is fresh (just sent) → below the stale threshold → leave it alone. + app._redeliverSweep(); + + expect(frames.map((f) => f.d)).toEqual(['a']); // no second send + expect(close).not.toHaveBeenCalled(); + }); + + it('force-closes the socket when stale AND silent (half-open — COD-134 fallback preserved)', () => { + const app = makeApp() as RedriveApp; + const frames: Frame[] = []; + const close = vi.fn(); + app._ws = { readyState: 1, send: (d: string) => frames.push(JSON.parse(d)), close } as never; + app._wsSessionId = 'session-1'; + app._reliableAckTimeoutMs = 4000; + + app._sendInputAsync('session-1', 'a'); + const list = app._pendingDeliveries.get('session-1')!; + list[0].sentAt = Date.now() - (app._reliableAckTimeoutMs + 1000); // stale + app._wsLastRecvAt = Date.now() - (app._reliableAckTimeoutMs + 1000); // silent + + app._redeliverSweep(); + + // Half-open socket never recovers on its own → force-close to reconnect. + // It must NOT have re-sent over the dead socket. + expect(close).toHaveBeenCalledTimes(1); + expect(frames.map((f) => f.d)).toEqual(['a']); + }); +}); From 4ab89f9a4e9cf3afd2b785e2660c3cffbd891ff2 Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Sat, 20 Jun 2026 09:53:52 -0400 Subject: [PATCH 4/6] COD-137 scope WS per-session limit by clientId (fix spurious 4008 on reconnect) MAX_WS_PER_SESSION was gated by a bare Map counter, incremented on upgrade and decremented only on the old socket's async close. A client that dropped and immediately reconnected could land its new upgrade before the old socket's close fired, briefly over-counting and tripping a spurious 4008 (-> HTTP fallback). The limit also counted raw sockets, so a reconnecting client consumed a new slot instead of its own. Replace the counter with WsConnectionRegistry (new pure, unit-tested module) that tracks live sockets per session keyed by clientId. A same-cid upgrade SUPERSEDES its own socket (evicts the stale one with close 4010, reuses the slot, no net count change) -> a reconnect can never be rejected by the cap. The reliable-input protocol (shouldApplyInput(cid,seq)) already assumes one logical client per cid per session, so same-cid eviction is principled, not a regression of multi-tab (which already collides on seq). Slots are freed EAGERLY on error/terminate, not just async close; close is identity-matched so a superseded socket's late close is a no-op. cid-less upgrades are admitted anonymously up to the cap and never evict (backward-compat). Client sends cid on the WS upgrade URL (?cid=, encoded, omitted if absent). Tests: ws-connection-registry.test.ts (reconnect-reclaim at cap, rejects N+1th distinct, eager-terminate frees slot, cid-less up-to-limit + no-evict, late-close-no-evict, per-session isolation) + route integration in ws-routes.test.ts (real upgrade through the cap). 45/45 across registry + ws-routes + input-send-order + ws-reconnect-plan; tsc 0, build, prettier, frontend-syntax clean. --- src/web/public/app.js | 8 +- src/web/routes/ws-routes.ts | 383 +++++++++++++++------------- src/web/ws-connection-registry.ts | 120 +++++++++ test/routes/ws-routes.test.ts | 23 ++ test/ws-connection-registry.test.ts | 112 ++++++++ 5 files changed, 471 insertions(+), 175 deletions(-) create mode 100644 src/web/ws-connection-registry.ts create mode 100644 test/ws-connection-registry.test.ts diff --git a/src/web/public/app.js b/src/web/public/app.js index ccec5458..608bd900 100644 --- a/src/web/public/app.js +++ b/src/web/public/app.js @@ -1971,7 +1971,13 @@ class CodemanApp { this._disconnectWs(); const proto = location.protocol === 'https:' ? 'wss:' : 'ws:'; - const url = `${proto}//${location.host}/ws/sessions/${sessionId}/terminal`; + // Pass the stable per-browser clientId on the upgrade URL so the server's + // connection registry scopes the per-session limit by client (COD-137): a + // reconnect supersedes its own socket instead of consuming a new slot and + // tripping a spurious 4008. Omitted if clientId is unavailable (server then + // treats the upgrade as anonymous — still admitted up to the limit). + const cidQuery = this._clientId ? `?cid=${encodeURIComponent(this._clientId)}` : ''; + const url = `${proto}//${location.host}/ws/sessions/${sessionId}/terminal${cidQuery}`; const ws = new WebSocket(url); this._ws = ws; this._wsSessionId = sessionId; diff --git a/src/web/routes/ws-routes.ts b/src/web/routes/ws-routes.ts index 17d7ee3e..f71a1ed4 100644 --- a/src/web/routes/ws-routes.ts +++ b/src/web/routes/ws-routes.ts @@ -34,6 +34,7 @@ import type { WebSocket } from 'ws'; import type { SessionPort } from '../ports/session-port.js'; import { MAX_INPUT_LENGTH } from '../../config/terminal-limits.js'; import { isAllowedRequestHost, isAllowedRequestOrigin, type HostPolicy } from '../network-auth-policy.js'; +import { WsConnectionRegistry } from '../ws-connection-registry.js'; /** Micro-batch interval for terminal output (ms). Short enough for low latency, * long enough to group Ink's rapid cursor-up redraw sequences into single frames. */ @@ -59,206 +60,240 @@ const DEC_2026_END = '\x1b[?2026l'; /** Max concurrent WS connections per session. Prevents listener/bandwidth multiplication. */ const MAX_WS_PER_SESSION = 5; -/** Track active WS connections per session for connection limiting. */ -const sessionWsCount = new Map(); +/** + * Track live WS connections per session, keyed by clientId (COD-137). + * Replaces a bare counter that over-counted across the async-close gap on + * reconnect (spurious 4008). A same-`cid` reconnect supersedes its own socket + * (reclaims the slot) instead of consuming a new one; cid-less upgrades are + * admitted anonymously up to the cap. See ws-connection-registry.ts. + */ +const sessionWsRegistry = new WsConnectionRegistry(MAX_WS_PER_SESSION); export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHostPolicy: () => HostPolicy): void { - app.get<{ Params: { id: string } }>('/ws/sessions/:id/terminal', { websocket: true }, (socket: WebSocket, req) => { - // Reject cross-site WebSocket hijacking (CSWSH) and DNS-rebinding before doing - // anything: the upgrade must come from an allowed Host and (when the browser - // sends one — it always does for WS) a same-site Origin. Writing to this socket - // injects keystrokes into a --dangerously-skip-permissions agent, so this gate - // matters even on the default no-password install. See security review H5. - const policy = getHostPolicy(); - if (!isAllowedRequestHost(req.headers.host, policy) || !isAllowedRequestOrigin(req.headers.origin, policy)) { - socket.close(4003, 'Forbidden'); - return; - } + app.get<{ Params: { id: string }; Querystring: { cid?: string } }>( + '/ws/sessions/:id/terminal', + { websocket: true }, + (socket: WebSocket, req) => { + // Reject cross-site WebSocket hijacking (CSWSH) and DNS-rebinding before doing + // anything: the upgrade must come from an allowed Host and (when the browser + // sends one — it always does for WS) a same-site Origin. Writing to this socket + // injects keystrokes into a --dangerously-skip-permissions agent, so this gate + // matters even on the default no-password install. See security review H5. + const policy = getHostPolicy(); + if (!isAllowedRequestHost(req.headers.host, policy) || !isAllowedRequestOrigin(req.headers.origin, policy)) { + socket.close(4003, 'Forbidden'); + return; + } - const { id } = req.params; - const session = ctx.sessions.get(id); + const { id } = req.params; + const session = ctx.sessions.get(id); - if (!session) { - socket.close(4004, 'Session not found'); - return; - } + if (!session) { + socket.close(4004, 'Session not found'); + return; + } - // Structured transport logging — surfaces WS open/close/timeout churn so the - // tunnel-flap behavior (COD-134) is observable in the server logs. Fastify is - // configured logger:false, so we log via console (→ journald under systemd). + // Structured transport logging — surfaces WS open/close/timeout churn so the + // tunnel-flap behavior (COD-134) is observable in the server logs. Fastify is + // configured logger:false, so we log via console (→ journald under systemd). - // Enforce per-session connection limit - const currentCount = sessionWsCount.get(id) ?? 0; - if (currentCount >= MAX_WS_PER_SESSION) { - console.warn('[ws] terminal rejected: too many connections', { sessionId: id, wsCount: currentCount }); - socket.close(4008, 'Too many connections'); - return; - } - sessionWsCount.set(id, currentCount + 1); - console.info('[ws] terminal open', { sessionId: id, wsCount: currentCount + 1 }); + // Enforce per-session connection limit, scoped by clientId. A same-cid + // reconnect reclaims its own slot (registry evicts the stale socket), so a + // drop+reconnect burst can no longer over-count across the async-close gap + // and trip a spurious 4008. cid-less upgrades are admitted anonymously. + const cid = typeof req.query?.cid === 'string' && req.query.cid.length > 0 ? req.query.cid : null; + const { admitted, evictedSocket } = sessionWsRegistry.register(id, cid, socket); + if (!admitted) { + console.warn('[ws] terminal rejected: too many connections', { + sessionId: id, + wsCount: sessionWsRegistry.liveCount(id), + }); + socket.close(4008, 'Too many connections'); + return; + } + if (evictedSocket) { + // Same client reconnected; retire the stale socket so it doesn't linger. + console.info('[ws] terminal superseded by reconnect', { sessionId: id }); + try { + evictedSocket.close(4010, 'Superseded by reconnect'); + } catch { + /* socket may already be closing */ + } + } + console.info('[ws] terminal open', { sessionId: id, wsCount: sessionWsRegistry.liveCount(id) }); - // Swallow socket errors — cleanup happens in 'close' - socket.on('error', () => {}); + // Eagerly free the slot on error/terminate — don't wait for the async + // 'close' (idempotent with the 'close' handler below). This is what kills + // the reconnect over-count: the slot is released the instant the socket dies. + socket.on('error', () => { + sessionWsRegistry.unregister(id, socket); + }); - // Per-connection micro-batch state - let batchChunks: string[] = []; - let batchSize = 0; - let batchTimer: ReturnType | null = null; + // Per-connection micro-batch state + let batchChunks: string[] = []; + let batchSize = 0; + let batchTimer: ReturnType | null = null; - const flushBatch = () => { - batchTimer = null; - if (batchChunks.length === 0 || socket.readyState !== 1) { + const flushBatch = () => { + batchTimer = null; + if (batchChunks.length === 0 || socket.readyState !== 1) { + batchChunks = []; + batchSize = 0; + return; + } + const data = batchChunks.join(''); batchChunks = []; batchSize = 0; - return; - } - const data = batchChunks.join(''); - batchChunks = []; - batchSize = 0; - socket.send(`{"t":"o","d":${JSON.stringify(DEC_2026_START + data + DEC_2026_END)}}`); - }; + socket.send(`{"t":"o","d":${JSON.stringify(DEC_2026_START + data + DEC_2026_END)}}`); + }; - // Per-connection desktop sizing claim — registered on the first - // desktop-typed resize and released on socket close, so Session.resize() - // can ignore small-viewport resizes only while a desktop is actually - // connected (see Session._desktopSizeClaims). - const sizingToken = Symbol('ws-desktop-sizing'); - let holdsDesktopClaim = false; + // Per-connection desktop sizing claim — registered on the first + // desktop-typed resize and released on socket close, so Session.resize() + // can ignore small-viewport resizes only while a desktop is actually + // connected (see Session._desktopSizeClaims). + const sizingToken = Symbol('ws-desktop-sizing'); + let holdsDesktopClaim = false; - // Attach message handler synchronously BEFORE any async work - // (@fastify/websocket requirement to avoid dropped messages). - socket.on('message', (raw) => { - try { - const msg = JSON.parse(String(raw)); - if (msg.t === 'i' && typeof msg.d === 'string') { - if (msg.d.length > MAX_INPUT_LENGTH) return; - // Reliable delivery: when the frame carries a clientId + seq, apply it - // exactly once (skip a duplicate redelivery) but ACK it regardless so - // the client can drop it from its durable queue. Frames without seq - // (legacy/other tools) are applied as-is — no behavior change. - const cid = typeof msg.cid === 'string' ? msg.cid : null; - const seq = Number.isInteger(msg.seq) ? (msg.seq as number) : null; - const apply = cid && seq !== null ? session.shouldApplyInput(cid, seq) : true; - if (apply) { - // Typed input from a claim-holding desktop keeps the claim "hot" - // and re-asserts the desktop layout after a mobile override. - if (holdsDesktopClaim) session.noteDesktopActivity(); - session.write(msg.d); - } - if (seq !== null && socket.readyState === 1) { - socket.send(`{"t":"ia","seq":${seq}}`); + // Attach message handler synchronously BEFORE any async work + // (@fastify/websocket requirement to avoid dropped messages). + socket.on('message', (raw) => { + try { + const msg = JSON.parse(String(raw)); + if (msg.t === 'i' && typeof msg.d === 'string') { + if (msg.d.length > MAX_INPUT_LENGTH) return; + // Reliable delivery: when the frame carries a clientId + seq, apply it + // exactly once (skip a duplicate redelivery) but ACK it regardless so + // the client can drop it from its durable queue. Frames without seq + // (legacy/other tools) are applied as-is — no behavior change. + const cid = typeof msg.cid === 'string' ? msg.cid : null; + const seq = Number.isInteger(msg.seq) ? (msg.seq as number) : null; + const apply = cid && seq !== null ? session.shouldApplyInput(cid, seq) : true; + if (apply) { + // Typed input from a claim-holding desktop keeps the claim "hot" + // and re-asserts the desktop layout after a mobile override. + if (holdsDesktopClaim) session.noteDesktopActivity(); + session.write(msg.d); + } + if (seq !== null && socket.readyState === 1) { + socket.send(`{"t":"ia","seq":${seq}}`); + } + } else if ( + msg.t === 'z' && + Number.isInteger(msg.c) && + Number.isInteger(msg.r) && + msg.c >= 1 && + msg.c <= 500 && + msg.r >= 1 && + msg.r <= 200 + ) { + const viewportType = msg.v === 'mobile' || msg.v === 'tablet' || msg.v === 'desktop' ? msg.v : undefined; + if (viewportType === 'desktop') { + session.claimDesktopSizing(sizingToken); + holdsDesktopClaim = true; + } else if (viewportType) { + // The connection's viewport can change (e.g. browser window + // narrowed past the tablet breakpoint) — drop a stale claim. + session.releaseDesktopSizing(sizingToken); + holdsDesktopClaim = false; + } + const force = msg.f === true; + session.resize(msg.c, msg.r, { viewportType, force }); } - } else if ( - msg.t === 'z' && - Number.isInteger(msg.c) && - Number.isInteger(msg.r) && - msg.c >= 1 && - msg.c <= 500 && - msg.r >= 1 && - msg.r <= 200 - ) { - const viewportType = msg.v === 'mobile' || msg.v === 'tablet' || msg.v === 'desktop' ? msg.v : undefined; - if (viewportType === 'desktop') { - session.claimDesktopSizing(sizingToken); - holdsDesktopClaim = true; - } else if (viewportType) { - // The connection's viewport can change (e.g. browser window - // narrowed past the tablet breakpoint) — drop a stale claim. - session.releaseDesktopSizing(sizingToken); - holdsDesktopClaim = false; - } - const force = msg.f === true; - session.resize(msg.c, msg.r, { viewportType, force }); + } catch { + // Ignore malformed messages } - } catch { - // Ignore malformed messages - } - }); + }); - // Terminal output -> micro-batched WS send - const onTerminal = (data: string) => { - if (socket.readyState !== 1) return; - batchChunks.push(data); - batchSize += data.length; + // Terminal output -> micro-batched WS send + const onTerminal = (data: string) => { + if (socket.readyState !== 1) return; + batchChunks.push(data); + batchSize += data.length; - // Flush immediately for large batches (responsiveness during bulk output) - if (batchSize > WS_BATCH_FLUSH_THRESHOLD) { - if (batchTimer) { - clearTimeout(batchTimer); + // Flush immediately for large batches (responsiveness during bulk output) + if (batchSize > WS_BATCH_FLUSH_THRESHOLD) { + if (batchTimer) { + clearTimeout(batchTimer); + } + flushBatch(); + return; } - flushBatch(); - return; - } - // Start timer if not already running - if (!batchTimer) { - batchTimer = setTimeout(flushBatch, WS_BATCH_INTERVAL_MS); - } - }; + // Start timer if not already running + if (!batchTimer) { + batchTimer = setTimeout(flushBatch, WS_BATCH_INTERVAL_MS); + } + }; - const onClearTerminal = () => { - if (socket.readyState === 1) { - socket.send('{"t":"c"}'); - } - }; + const onClearTerminal = () => { + if (socket.readyState === 1) { + socket.send('{"t":"c"}'); + } + }; - const onNeedsRefresh = () => { - if (socket.readyState === 1) { - socket.send('{"t":"r"}'); - } - }; + const onNeedsRefresh = () => { + if (socket.readyState === 1) { + socket.send('{"t":"r"}'); + } + }; - // Close WS when session exits (deleted, respawned, or crashed) — prevents - // orphaned listeners and stale writes to a dead PTY. - const onSessionExit = () => { - socket.close(4009, 'Session terminated'); - }; + // Close WS when session exits (deleted, respawned, or crashed) — prevents + // orphaned listeners and stale writes to a dead PTY. + const onSessionExit = () => { + socket.close(4009, 'Session terminated'); + }; - session.on('terminal', onTerminal); - session.on('clearTerminal', onClearTerminal); - session.on('needsRefresh', onNeedsRefresh); - session.on('exit', onSessionExit); + session.on('terminal', onTerminal); + session.on('clearTerminal', onClearTerminal); + session.on('needsRefresh', onNeedsRefresh); + session.on('exit', onSessionExit); - // Heartbeat: detect stale connections (especially through tunnels where - // TCP RST can take minutes to propagate). - let pongTimeout: ReturnType | null = null; + // Heartbeat: detect stale connections (especially through tunnels where + // TCP RST can take minutes to propagate). + let pongTimeout: ReturnType | null = null; - socket.on('pong', () => { - if (pongTimeout) { - clearTimeout(pongTimeout); - pongTimeout = null; - } - }); + socket.on('pong', () => { + if (pongTimeout) { + clearTimeout(pongTimeout); + pongTimeout = null; + } + }); - const pingInterval = setInterval(() => { - if (socket.readyState !== 1) return; - socket.ping(); - pongTimeout = setTimeout(() => { - console.warn('[ws] terminal ping timeout — terminating', { sessionId: id }); - socket.terminate(); - }, WS_PONG_TIMEOUT_MS); - }, WS_PING_INTERVAL_MS); + const pingInterval = setInterval(() => { + if (socket.readyState !== 1) return; + socket.ping(); + pongTimeout = setTimeout(() => { + console.warn('[ws] terminal ping timeout — terminating', { sessionId: id }); + // Free the slot eagerly — terminate()'s 'close' may lag, and a client + // reconnecting after a stale-connection drop must not be over-counted. + sessionWsRegistry.unregister(id, socket); + socket.terminate(); + }, WS_PONG_TIMEOUT_MS); + }, WS_PING_INTERVAL_MS); - socket.on('close', (code: number, reason: Buffer) => { - clearInterval(pingInterval); - if (pongTimeout) clearTimeout(pongTimeout); - if (batchTimer) clearTimeout(batchTimer); - batchChunks = []; - session.off('terminal', onTerminal); - session.off('clearTerminal', onClearTerminal); - session.off('needsRefresh', onNeedsRefresh); - session.off('exit', onSessionExit); - session.releaseDesktopSizing(sizingToken); + socket.on('close', (code: number, reason: Buffer) => { + clearInterval(pingInterval); + if (pongTimeout) clearTimeout(pongTimeout); + if (batchTimer) clearTimeout(batchTimer); + batchChunks = []; + session.off('terminal', onTerminal); + session.off('clearTerminal', onClearTerminal); + session.off('needsRefresh', onNeedsRefresh); + session.off('exit', onSessionExit); + session.releaseDesktopSizing(sizingToken); - // Decrement per-session connection count - const count = sessionWsCount.get(id) ?? 1; - const remaining = count <= 1 ? 0 : count - 1; - if (remaining === 0) { - sessionWsCount.delete(id); - } else { - sessionWsCount.set(id, remaining); - } - console.info('[ws] terminal close', { sessionId: id, code, reason: String(reason), wsCount: remaining }); - }); - }); + // Release this socket's slot. Idempotent and identity-matched: if this + // socket was already superseded (a same-cid reconnect took its slot) or + // eagerly unregistered on terminate/error, this is a no-op and the + // reconnected socket keeps the slot. + sessionWsRegistry.unregister(id, socket); + console.info('[ws] terminal close', { + sessionId: id, + code, + reason: String(reason), + wsCount: sessionWsRegistry.liveCount(id), + }); + }); + } + ); } diff --git a/src/web/ws-connection-registry.ts b/src/web/ws-connection-registry.ts new file mode 100644 index 00000000..d0c64cda --- /dev/null +++ b/src/web/ws-connection-registry.ts @@ -0,0 +1,120 @@ +/** + * @fileoverview Per-session WebSocket connection registry (COD-137). + * + * Replaces the bare `Map` counter that previously gated + * `MAX_WS_PER_SESSION`. That counter had two defects: + * + * 1. Transient over-count on reconnect: a client that drops and immediately + * reconnects could land its new upgrade BEFORE the old socket's async + * `close` fired, so the count briefly exceeded the live connection number. + * A reconnect burst could hit the cap and the next upgrade was rejected + * with 4008 → the client fell back to HTTP. (The real spurious-4008 defect.) + * 2. No clientId scoping: the limit counted raw sockets, so a reconnecting + * client consumed a NEW slot instead of replacing its own. + * + * This registry tracks the live socket(s) per session keyed by clientId (`cid`, + * parsed from the upgrade URL query). The reliable-input protocol + * (`session.shouldApplyInput(cid, seq)`) already assumes ONE logical client per + * `cid` per session, so a new upgrade for a `cid` that already holds a socket is + * a SUPERSEDE — the registry evicts the stale socket and reuses its slot, which + * makes a reconnect reclaim rather than double-count (fixes #1 and #2). + * + * Backward-compat: an upgrade with NO `cid` (legacy clients, other tools) is + * admitted anonymously — it counts toward the limit but never evicts another + * client, and several anonymous sockets can coexist up to the cap. + * + * The class is pure (no `ws`/Fastify imports) and generic over a minimal socket + * shape so it can be unit-tested with plain fakes. The route owns the actual + * socket close/terminate; the registry only decides admit/evict and tracks slots. + */ + +/** Minimal socket shape the registry needs — satisfied by `ws` WebSocket. */ +export interface RegistrableSocket { + /** Identity comparison only; never dereferenced beyond `===`. */ + readonly readyState?: number; +} + +export interface RegisterResult { + /** Whether the new socket was admitted (false → caller should reject with 4008). */ + admitted: boolean; + /** + * A stale socket whose slot the new socket reclaimed (same `cid`). The caller + * should close it. Present only on a keyed supersede; never set for anonymous + * upgrades or fresh slots. + */ + evictedSocket?: S; +} + +/** A single live entry: the socket plus its clientId (null = anonymous). */ +interface Entry { + socket: S; + cid: string | null; +} + +export class WsConnectionRegistry { + /** sessionId → live entries (keyed + anonymous). */ + private readonly bySession = new Map[]>(); + + constructor(private readonly maxPerSession: number) {} + + /** + * Attempt to register a new socket for `(sessionId, cid)`. + * + * - cid present and already holds a socket → SUPERSEDE: evict the old one, + * reuse its slot, always admit. + * - otherwise → admit iff distinct-entry count < maxPerSession. + * + * A null/empty `cid` is anonymous: it never matches an existing entry and so + * never evicts; it just consumes a slot. + */ + register(sessionId: string, cid: string | null, socket: S): RegisterResult { + const entries = this.bySession.get(sessionId) ?? []; + + if (cid) { + const existingIdx = entries.findIndex((e) => e.cid === cid); + if (existingIdx !== -1) { + const evicted = entries[existingIdx].socket; + // Reuse the slot in place — no net change to the live count, so a + // reconnect can never be rejected by the cap. + entries[existingIdx] = { socket, cid }; + this.bySession.set(sessionId, entries); + return { admitted: true, evictedSocket: evicted === socket ? undefined : evicted }; + } + } + + if (entries.length >= this.maxPerSession) { + return { admitted: false }; + } + + entries.push({ socket, cid: cid || null }); + this.bySession.set(sessionId, entries); + return { admitted: true }; + } + + /** + * Remove a socket from its session. Idempotent — safe to call on `close`, + * `error`, AND eagerly on `terminate()` (the over-count fix relies on eager + * removal freeing the slot before the async `close` fires). + * + * Matches by socket identity, so a socket that was already superseded + * (replaced in-slot by a same-cid reconnect) is NOT removed by its late + * `close` — the new socket keeps the slot. + */ + unregister(sessionId: string, socket: S): void { + const entries = this.bySession.get(sessionId); + if (!entries) return; + const idx = entries.findIndex((e) => e.socket === socket); + if (idx === -1) return; + entries.splice(idx, 1); + if (entries.length === 0) { + this.bySession.delete(sessionId); + } else { + this.bySession.set(sessionId, entries); + } + } + + /** Number of live entries for a session (0 if none). */ + liveCount(sessionId: string): number { + return this.bySession.get(sessionId)?.length ?? 0; + } +} diff --git a/test/routes/ws-routes.test.ts b/test/routes/ws-routes.test.ts index 93314dd7..5b722036 100644 --- a/test/routes/ws-routes.test.ts +++ b/test/routes/ws-routes.test.ts @@ -491,6 +491,29 @@ describe('ws-routes', () => { for (const ws of connections) ws.close(); } }); + + it('reconnecting client (same cid) is admitted at the cap instead of 4008 (COD-137)', async () => { + const connections: WebSocket[] = []; + try { + // Fill all 5 slots with DISTINCT clients, one of which is "alice". + for (const c of ['alice', 'b', 'c', 'd', 'e']) { + connections.push(await connectWs(`/ws/sessions/ws-test-session/terminal?cid=${c}`)); + } + + // Alice reconnects WHILE her old socket is still registered (the + // over-count window). This must reclaim her slot, not hit the cap. + const aliceNew = await connectWs('/ws/sessions/ws-test-session/terminal?cid=alice'); + connections.push(aliceNew); + + // Sanity: the reconnected socket is live and usable. + ctx._session.emit('terminal', 'reconnected-ok'); + const msg = (await nextMessage(aliceNew)) as { t: string; d: string }; + expect(msg.t).toBe('o'); + expect(msg.d).toContain('reconnected-ok'); + } finally { + for (const ws of connections) ws.close(); + } + }); }); // ========== Heartbeat ========== diff --git a/test/ws-connection-registry.test.ts b/test/ws-connection-registry.test.ts new file mode 100644 index 00000000..35105878 --- /dev/null +++ b/test/ws-connection-registry.test.ts @@ -0,0 +1,112 @@ +/** + * @fileoverview Unit tests for WsConnectionRegistry (COD-137). + * + * The registry is the pure decision unit extracted out of ws-routes.ts so the + * connection-limit / clientId-eviction logic is testable without driving real + * WebSocket upgrades. Uses plain fake sockets (identity only). + * + * @dependency src/web/ws-connection-registry.ts + */ + +import { describe, it, expect } from 'vitest'; +import { WsConnectionRegistry } from '../src/web/ws-connection-registry.js'; + +/** Fake socket — registry only compares identity, so any object works. */ +const sock = (label: string) => ({ readyState: 1, label }); + +describe('WsConnectionRegistry', () => { + it('reconnecting client (same cid) reclaims its slot instead of being rejected at the limit', () => { + const reg = new WsConnectionRegistry(5); + // Fill all 5 slots with distinct clients, one of which is "alice". + for (const c of ['alice', 'b', 'c', 'd', 'e']) { + expect(reg.register('s1', c, sock(c)).admitted).toBe(true); + } + expect(reg.liveCount('s1')).toBe(5); + + // Alice's new upgrade lands BEFORE her old socket's async close fires. + const aliceNew = sock('alice-new'); + const res = reg.register('s1', 'alice', aliceNew); + + expect(res.admitted).toBe(true); // NOT a spurious 4008 + expect(res.evictedSocket).toBeDefined(); // old alice socket handed back to close + expect(reg.liveCount('s1')).toBe(5); // slot reused, not double-counted + }); + + it('still rejects a genuine (N+1)th DISTINCT client', () => { + const reg = new WsConnectionRegistry(5); + for (const c of ['a', 'b', 'c', 'd', 'e']) { + expect(reg.register('s1', c, sock(c)).admitted).toBe(true); + } + const sixth = reg.register('s1', 'f', sock('f')); + expect(sixth.admitted).toBe(false); + expect(sixth.evictedSocket).toBeUndefined(); + expect(reg.liveCount('s1')).toBe(5); + }); + + it('eager removal on terminate frees a slot immediately', () => { + const reg = new WsConnectionRegistry(5); + const sockets = ['a', 'b', 'c', 'd', 'e'].map((c) => { + const s = sock(c); + reg.register('s1', c, s); + return [c, s] as const; + }); + expect(reg.register('s1', 'f', sock('f')).admitted).toBe(false); + + // Eagerly unregister one (simulating terminate/error, not async close). + reg.unregister('s1', sockets[0][1]); + expect(reg.liveCount('s1')).toBe(4); + + // Now a brand-new distinct client is admitted. + expect(reg.register('s1', 'f', sock('f')).admitted).toBe(true); + expect(reg.liveCount('s1')).toBe(5); + }); + + it('cid-less upgrades are admitted up to the limit and never evict a keyed client', () => { + const reg = new WsConnectionRegistry(5); + const keyed = sock('keyed'); + reg.register('s1', 'keyed', keyed); + + // Four anonymous upgrades fill the rest of the cap. + for (let i = 0; i < 4; i++) { + const res = reg.register('s1', null, sock(`anon${i}`)); + expect(res.admitted).toBe(true); + expect(res.evictedSocket).toBeUndefined(); // never evicts the keyed client + } + expect(reg.liveCount('s1')).toBe(5); + + // 6th anonymous is rejected — anonymous sockets count toward the cap. + expect(reg.register('s1', null, sock('anon-extra')).admitted).toBe(false); + + // The keyed client is untouched: a same-cid reconnect still reclaims. + const keyedNew = sock('keyed-new'); + const res = reg.register('s1', 'keyed', keyedNew); + expect(res.admitted).toBe(true); + expect(res.evictedSocket).toBe(keyed); + }); + + it('late close of a superseded socket does not evict the reconnected one', () => { + const reg = new WsConnectionRegistry(5); + const old = sock('old'); + reg.register('s1', 'alice', old); + const fresh = sock('fresh'); + reg.register('s1', 'alice', fresh); // supersede + + // The stale socket's async close arrives late — must NOT remove fresh. + reg.unregister('s1', old); + expect(reg.liveCount('s1')).toBe(1); + + // Fresh is still the live entry: another reconnect evicts fresh, not old. + const fresher = sock('fresher'); + expect(reg.register('s1', 'alice', fresher).evictedSocket).toBe(fresh); + }); + + it('isolates counts per session', () => { + const reg = new WsConnectionRegistry(2); + reg.register('s1', 'a', sock('a')); + reg.register('s1', 'b', sock('b')); + expect(reg.register('s1', 'c', sock('c')).admitted).toBe(false); + // s2 has its own budget. + expect(reg.register('s2', 'a', sock('a2')).admitted).toBe(true); + expect(reg.liveCount('s2')).toBe(1); + }); +}); From b86b132af59513a8ca023323009bd26869b3c614 Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Sat, 20 Jun 2026 10:03:25 -0400 Subject: [PATCH 5/6] COD-136 skip redundant connection-indicator DOM writes on hot input path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _updateConnectionIndicator() ran on every keystroke (_reliableSend) and every ACK (_ackDelivery), unconditionally writing display/className/ textContent/title. During fast typing the rendered output is usually identical between calls, so those were wasted main-thread DOM writes. Extracted the branch logic into a pure DOM-free _computeConnectionDescriptor() returning { display, dotClass, text, title } (every branch/string preserved verbatim; hidden state normalizes the three non-display fields to '' so the compare is well-defined). _updateConnectionIndicator() now computes the descriptor, compares all four fields against a cached _lastIndicatorDescriptor, and early-returns when unchanged — otherwise caches and writes the DOM exactly as before (display always; dotClass/text/title only when shown). First call renders (cache starts null). Perf only, no behavior change. Tests: test/connection-indicator.test.ts — 9 descriptor cases pinning the exact strings per state + 4 skip cases (first call writes; two identical calls write DOM once via counting setters; state change and hidden->shown re-render). 31/31 with input-send-order regression; build, frontend-syntax, prettier clean. --- src/web/public/app.js | 89 +++++++---- test/connection-indicator.test.ts | 241 ++++++++++++++++++++++++++++++ 2 files changed, 304 insertions(+), 26 deletions(-) diff --git a/src/web/public/app.js b/src/web/public/app.js index 608bd900..fa35d98c 100644 --- a/src/web/public/app.js +++ b/src/web/public/app.js @@ -483,6 +483,9 @@ class CodemanApp { this._clientId = ''; this._seqCounters = new Map(); // sessionId -> last issued seq this._pendingDeliveries = new Map(); // sessionId -> [{seq,data,useMux,ts,tries,sentAt}] + // Last rendered connection-indicator tuple; the hot input path skips DOM + // writes when the freshly computed descriptor is identical (COD-136). + this._lastIndicatorDescriptor = null; this._postDraining = new Set(); // sessionIds with an in-flight POST drainer this._persistReliableTimer = null; this._reliableAckTimeoutMs = 4000; // unacked WS frame older than this ⇒ socket likely dead @@ -2420,12 +2423,12 @@ class CodemanApp { } } - _updateConnectionIndicator() { - const indicator = this.$('connectionIndicator'); - const dot = this.$('connectionDot'); - const text = this.$('connectionText'); - if (!indicator || !dot || !text) return; - + // Pure render of the header connection indicator: reads only `this.*` state, + // touches NO DOM. Returns the exact { display, dotClass, text, title } tuple the + // writer applies. When hidden (display:'none') the other three are normalized to + // '' so the cache compare in _updateConnectionIndicator() is well-defined. + // Every branch/string here must stay byte-identical to what's rendered today. + _computeConnectionDescriptor() { const { bytes: totalBytes, count } = this._pendingBytes(); const hasQueue = count > 0; // Only surface a backlog once it's more than a few bytes. A single keystroke @@ -2438,11 +2441,12 @@ class CodemanApp { // Hard offline (browser reports no network) dominates everything. if (!this.isOnline || this._connectionStatus === 'offline') { - indicator.style.display = 'flex'; - dot.className = 'connection-dot offline'; - text.textContent = showBacklog ? `Offline (${formatBytes(totalBytes)} queued)` : 'Offline'; - indicator.title = 'No network connection'; - return; + return { + display: 'flex', + dotClass: 'connection-dot offline', + text: showBacklog ? `Offline (${formatBytes(totalBytes)} queued)` : 'Offline', + title: 'No network connection', + }; } // With an active terminal, show its transport (WebSocket vs HTTP fallback). @@ -2463,31 +2467,64 @@ class CodemanApp { cls = 'reconnecting'; label = 'WS…'; detail = 'Connecting WebSocket'; break; } - indicator.style.display = 'flex'; - dot.className = `connection-dot ${cls}`; - text.textContent = `${label}${queuedSuffix}`; - indicator.title = detail; - return; + return { + display: 'flex', + dotClass: `connection-dot ${cls}`, + text: `${label}${queuedSuffix}`, + title: detail, + }; } // No active terminal — reflect the SSE event stream only when it needs attention. if (this._connectionStatus === 'reconnecting' || this._connectionStatus === 'disconnected') { - indicator.style.display = 'flex'; - dot.className = 'connection-dot reconnecting'; - text.textContent = showBacklog ? `Reconnecting (${formatBytes(totalBytes)} queued)` : 'Reconnecting...'; - indicator.title = 'Reconnecting to server'; - return; + return { + display: 'flex', + dotClass: 'connection-dot reconnecting', + text: showBacklog ? `Reconnecting (${formatBytes(totalBytes)} queued)` : 'Reconnecting...', + title: 'Reconnecting to server', + }; } // Idle dashboard, healthy stream — hide unless input is genuinely queued. if (!hasQueue) { - indicator.style.display = 'none'; + return { display: 'none', dotClass: '', text: '', title: '' }; + } + return { + display: 'flex', + dotClass: 'connection-dot draining', + text: showBacklog ? `Sending ${formatBytes(totalBytes)}...` : 'Sending...', + title: 'Delivering queued input', + }; + } + + _updateConnectionIndicator() { + const indicator = this.$('connectionIndicator'); + const dot = this.$('connectionDot'); + const text = this.$('connectionText'); + if (!indicator || !dot || !text) return; + + // Called on EVERY keystroke (_reliableSend) and EVERY ACK (_ackDelivery). + // During fast typing the rendered tuple is usually identical, so skip the DOM + // writes when nothing changed (COD-136) — the compute above is DOM-free. + const next = this._computeConnectionDescriptor(); + const prev = this._lastIndicatorDescriptor; + if ( + prev && + prev.display === next.display && + prev.dotClass === next.dotClass && + prev.text === next.text && + prev.title === next.title + ) { return; } - indicator.style.display = 'flex'; - dot.className = 'connection-dot draining'; - text.textContent = showBacklog ? `Sending ${formatBytes(totalBytes)}...` : 'Sending...'; - indicator.title = 'Delivering queued input'; + this._lastIndicatorDescriptor = next; + + indicator.style.display = next.display; + if (next.display !== 'none') { + dot.className = next.dotClass; + text.textContent = next.text; + indicator.title = next.title; + } } setupOnlineDetection() { diff --git a/test/connection-indicator.test.ts b/test/connection-indicator.test.ts index 3f8bcc66..1c8a91ec 100644 --- a/test/connection-indicator.test.ts +++ b/test/connection-indicator.test.ts @@ -14,6 +14,15 @@ * surfaced the terminal WebSocket transport ("WS" / "HTTP"), and it flashed * "sending 1B" on every single keystroke. * + * COD-136 (perf, no behavior change) extracts the pure render into + * `_computeConnectionDescriptor()` and makes `_updateConnectionIndicator()` + * early-return when that descriptor is byte-identical to the last render — so + * fast typing stops doing redundant DOM writes on the hot input path. The + * `_computeConnectionDescriptor` block below pins the exact rendered strings per + * state (so a future refactor can't silently relabel), and the unchanged-skip + * block asserts the DOM is written once across two identical calls and re-written + * when state changes. + * * Loaded via `vm` with a stubbed context (no jsdom — see input-send-order.test.ts). */ import { readFileSync } from 'node:fs'; @@ -202,3 +211,235 @@ describe('durable input delivery is not aborted by the indicator (typing-lag reg expect(app._pendingDeliveries.get('s1')).toHaveLength(1); }); }); + +// ---- COD-136: pure descriptor + cache-skip on the hot input path -------------- + +type Descriptor = { display: string; dotClass: string; text: string; title: string }; + +type DescriptorApp = Indicator & { + _computeConnectionDescriptor: () => Descriptor; + _lastIndicatorDescriptor: Descriptor | null; +}; + +function computeApp(overrides: Partial & { queuedBytes?: number } = {}) { + const { app } = makeApp(overrides); + return app as unknown as DescriptorApp; +} + +describe('_computeConnectionDescriptor — pure render per state (COD-136)', () => { + it('offline dominates everything (even an active connected terminal)', () => { + const app = computeApp({ activeSessionId: 's1', _wsState: 'connected', isOnline: false }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot offline', + text: 'Offline', + title: 'No network connection', + }); + }); + + it('active terminal — connected → WS', () => { + const app = computeApp({ activeSessionId: 's1', _wsState: 'connected' }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot connected', + text: 'WS', + title: 'Terminal connected over WebSocket', + }); + }); + + it('active terminal — fallback → HTTP', () => { + const app = computeApp({ activeSessionId: 's1', _wsState: 'fallback' }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot fallback', + text: 'HTTP', + title: 'WebSocket unavailable — input sent over HTTP', + }); + }); + + it('active terminal — reconnecting → WS…', () => { + const app = computeApp({ activeSessionId: 's1', _wsState: 'reconnecting' }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot reconnecting', + text: 'WS…', + title: 'Reconnecting WebSocket', + }); + }); + + it('active terminal — connecting (default branch) → WS…', () => { + const app = computeApp({ activeSessionId: 's1', _wsState: 'connecting' }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot reconnecting', + text: 'WS…', + title: 'Connecting WebSocket', + }); + }); + + it('active terminal — a queued backlog (>4B) adds the " · …KB queued" suffix', () => { + const app = computeApp({ activeSessionId: 's1', _wsState: 'connected', queuedBytes: 2048 }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot connected', + text: 'WS · 2.0KB queued', + title: 'Terminal connected over WebSocket', + }); + }); + + it('no terminal, SSE reconnecting → Reconnecting...', () => { + const app = computeApp({ activeSessionId: null, _connectionStatus: 'reconnecting' }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot reconnecting', + text: 'Reconnecting...', + title: 'Reconnecting to server', + }); + }); + + it('idle dashboard, healthy stream, no queue → hidden (display:none, others normalized to "")', () => { + const app = computeApp({ activeSessionId: null, _connectionStatus: 'connected' }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'none', + dotClass: '', + text: '', + title: '', + }); + }); + + it('idle dashboard with a small queue (≤4B) → draining "Sending..."', () => { + const app = computeApp({ activeSessionId: null, _connectionStatus: 'connected', queuedBytes: 2 }); + expect(app._computeConnectionDescriptor()).toEqual({ + display: 'flex', + dotClass: 'connection-dot draining', + text: 'Sending...', + title: 'Delivering queued input', + }); + }); +}); + +/** A DOM element fake that COUNTS each property write — used to detect the skip. */ +function countingElement() { + const writes = { display: 0, className: 0, textContent: 0, title: 0 }; + let _display = ''; + let _className = ''; + let _textContent = ''; + let _title = ''; + return { + writes, + style: { + get display() { + return _display; + }, + set display(v: string) { + _display = v; + writes.display++; + }, + }, + get className() { + return _className; + }, + set className(v: string) { + _className = v; + writes.className++; + }, + get textContent() { + return _textContent; + }, + set textContent(v: string) { + _textContent = v; + writes.textContent++; + }, + get title() { + return _title; + }, + set title(v: string) { + _title = v; + writes.title++; + }, + }; +} + +function makeCountingApp(overrides: Partial & { queuedBytes?: number } = {}) { + const app = Object.create((CodemanApp as { prototype: object }).prototype) as DescriptorApp & { + queuedBytes?: number; + }; + const els = { + connectionIndicator: countingElement(), + connectionDot: countingElement(), + connectionText: countingElement(), + }; + app.$ = (id: string) => (els as Record>)[id]; + app._pendingDeliveries = new Map(); + app._connectionStatus = 'connected'; + app._wsState = 'disconnected'; + app.activeSessionId = null; + app.isOnline = true; + app._lastIndicatorDescriptor = null; + Object.assign(app, overrides); + const queued = overrides.queuedBytes ?? 0; + if (queued > 0) { + app._pendingDeliveries.set('s1', [{ seq: 1, data: 'x'.repeat(queued) }]); + } + return { app, els }; +} + +describe('_updateConnectionIndicator — COD-136 unchanged-skip', () => { + it('writes the DOM on the first call (cache starts null → renders)', () => { + const { app, els } = makeCountingApp({ activeSessionId: 's1', _wsState: 'connected' }); + app._updateConnectionIndicator(); + expect(els.connectionIndicator.style.display).toBe('flex'); + expect(els.connectionDot.className).toBe('connection-dot connected'); + expect(els.connectionText.textContent).toBe('WS'); + expect(els.connectionText.writes.textContent).toBe(1); + }); + + it('skips redundant DOM writes when the descriptor is unchanged across two calls', () => { + const { app, els } = makeCountingApp({ activeSessionId: 's1', _wsState: 'connected' }); + + app._updateConnectionIndicator(); // first render + const before = { + display: els.connectionIndicator.writes.display, + className: els.connectionDot.writes.className, + textContent: els.connectionText.writes.textContent, + title: els.connectionIndicator.writes.title, + }; + + app._updateConnectionIndicator(); // identical state → must early-return, no writes + + expect(els.connectionIndicator.writes.display).toBe(before.display); + expect(els.connectionDot.writes.className).toBe(before.className); + expect(els.connectionText.writes.textContent).toBe(before.textContent); + expect(els.connectionIndicator.writes.title).toBe(before.title); + }); + + it('re-renders when state changes between calls (WS → HTTP)', () => { + const { app, els } = makeCountingApp({ activeSessionId: 's1', _wsState: 'connected' }); + + app._updateConnectionIndicator(); // WS + const writesAfterFirst = els.connectionText.writes.textContent; + + app._wsState = 'fallback'; + app._updateConnectionIndicator(); // HTTP — must write again + + expect(els.connectionText.writes.textContent).toBe(writesAfterFirst + 1); + expect(els.connectionText.textContent).toBe('HTTP'); + expect(els.connectionDot.className).toBe('connection-dot fallback'); + }); + + it('re-renders display when the hidden→shown transition occurs (none → flex)', () => { + const { app, els } = makeCountingApp({ activeSessionId: null, _connectionStatus: 'connected' }); + + app._updateConnectionIndicator(); // hidden (display:none) + expect(els.connectionIndicator.style.display).toBe('none'); + const displayWrites = els.connectionIndicator.writes.display; + + app.activeSessionId = 's1'; + app._wsState = 'connected'; + app._updateConnectionIndicator(); // now shown + + expect(els.connectionIndicator.writes.display).toBe(displayWrites + 1); + expect(els.connectionIndicator.style.display).toBe('flex'); + expect(els.connectionText.textContent).toBe('WS'); + }); +}); From 584910f64533bc3aa079d0637990d210b675dfa2 Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Sun, 21 Jun 2026 14:47:21 -0400 Subject: [PATCH 6/6] COD-144 flush queued SSE output on empty buffer-load so new shells paint immediately A freshly created shell session rendered blank until a tab-switch. selectSession() fetches the terminal buffer, but for a just-started shell that fetch resolves before the PTY emits its prompt, so the buffer is empty; the prompt then arrives as a live SSE event queued during the load and _finishBufferLoad() discarded it. The discard is correct for an established session (its fetched buffer already contains that output), but harmful when the load painted nothing. _finishBufferLoad(owner, { flushQueued }) now REPLAYS the queued events through batchTerminalWrite (after _isLoadingBuffer is cleared, so they write through, not re-queue) instead of discarding. selectSession passes flushQueued only in the empty branch (no fresh buffer + no cache), so the established-session de-dup path is unchanged. TDD: test/terminal-buffer-flush.test.ts exercises the real begin/finish mixin (vm-harness, no jsdom). --- src/web/public/app.js | 14 ++- src/web/public/terminal-ui.js | 32 +++++- test/terminal-buffer-flush.test.ts | 172 +++++++++++++++++++++++++++++ 3 files changed, 209 insertions(+), 9 deletions(-) create mode 100644 test/terminal-buffer-flush.test.ts diff --git a/src/web/public/app.js b/src/web/public/app.js index fa35d98c..522b542b 100644 --- a/src/web/public/app.js +++ b/src/web/public/app.js @@ -3733,6 +3733,9 @@ class CodemanApp { // the buffer write, causing 70KB+ single-frame flushes that stall WebGL. // chunkedTerminalWrite also sets this, but we need it before the fetch too. const bufferLoadOwner = this._beginBufferLoad(selectGen); + // COD-144: track whether the load painted nothing (empty fetch + no cache). + // For that just-created-session case we flush (not discard) queued SSE events. + let bufferWasEmpty = false; try { // Fit terminal to container BEFORE writing any buffer data. // If the browser was resized while viewing another session, the terminal @@ -3889,13 +3892,16 @@ class CodemanApp { } else if (!cachedBuffer) { // No fresh buffer and no cache — clear any stale content this._resetTerminalForReplay(); + bufferWasEmpty = true; } - // Buffer load complete — unblock live SSE writes (queued events are discarded - // to prevent duplicate content). chunkedTerminalWrite calls _finishBufferLoad - // internally, but if we skipped the write (cache hit or empty), call it here. + // Buffer load complete — unblock live SSE writes. chunkedTerminalWrite calls + // _finishBufferLoad internally (discarding queued events to prevent duplicate + // content); if we skipped the write (cache hit or empty), call it here. + // COD-144: when the load painted nothing, FLUSH the queued events instead of + // discarding — a new session's prompt arrives only as a queued SSE event. if (this._isLoadingBuffer) { - this._finishBufferLoad(bufferLoadOwner); + this._finishBufferLoad(bufferLoadOwner, { flushQueued: bufferWasEmpty }); } // Drop the guard so user input clears state normally this._restoringFlushedState = false; diff --git a/src/web/public/terminal-ui.js b/src/web/public/terminal-ui.js index d44e5dbf..d38b7049 100644 --- a/src/web/public/terminal-ui.js +++ b/src/web/public/terminal-ui.js @@ -1925,11 +1925,25 @@ Object.assign(CodemanApp.prototype, { * Complete a buffer load: unblock live SSE writes. * Called when chunkedTerminalWrite finishes (or is skipped for empty buffers). * - * Queued SSE events are DISCARDED, not flushed. The loaded buffer from the API - * is the source of truth up to the response timestamp. SSE events queued during - * the fetch+write overlap with the buffer — flushing them writes duplicate data - * (especially Ink cursor-up redraws), corrupting the terminal display. + * By default queued SSE events are DISCARDED, not flushed. For an established + * session the loaded buffer from the API is the source of truth up to the + * response timestamp; SSE events queued during the fetch+write overlap already + * appear in that buffer, so flushing them writes duplicate data (especially Ink + * cursor-up redraws), corrupting the terminal display. + * + * COD-144: a brand-new session is the exception. Its terminal fetch can resolve + * BEFORE the PTY emits its first prompt, so the fetched buffer is empty and the + * prompt arrives only as a queued SSE event. Discarding it leaves the terminal + * blank until a tab-switch re-fetches a now-populated buffer. When the caller + * knows the load painted nothing (empty fetch + no cache), it passes + * `{ flushQueued: true }` so the queued events are REPLAYED through + * `batchTerminalWrite()` instead of dropped. Replay runs after `_isLoadingBuffer` + * is cleared, so the events write through normally and are not re-queued. + * * After unblocking, new SSE/WS events deliver subsequent output normally. + * + * @param {string} [owner] Load token from `_beginBufferLoad`; a stale owner is a no-op. + * @param {{ flushQueued?: boolean }} [opts] When `flushQueued` is true, replay any queued events. */ _beginBufferLoad(owner) { if (this._bufferLoadSeq === undefined) this._bufferLoadSeq = 0; @@ -1940,13 +1954,21 @@ Object.assign(CodemanApp.prototype, { return loadOwner; }, - _finishBufferLoad(owner) { + _finishBufferLoad(owner, opts) { if (owner !== undefined && this._bufferLoadOwner !== owner) { return false; } + const queued = this._loadBufferQueue; this._isLoadingBuffer = false; this._loadBufferQueue = null; this._bufferLoadOwner = null; + // COD-144: replay (rather than discard) queued live events when the load + // painted nothing — the queued prompt is the only content a new session has. + if (opts?.flushQueued && queued && queued.length) { + for (const data of queued) { + this.batchTerminalWrite(data); + } + } return true; }, diff --git a/test/terminal-buffer-flush.test.ts b/test/terminal-buffer-flush.test.ts new file mode 100644 index 00000000..54267beb --- /dev/null +++ b/test/terminal-buffer-flush.test.ts @@ -0,0 +1,172 @@ +/** + * @fileoverview Regression tests for the buffer-load flush path (COD-144). + * + * Bug: newly launched Shell sessions rendered BLANK until a tab-switch. The + * buffer-load path (`selectSession` → `_beginBufferLoad`/`_finishBufferLoad`) + * QUEUES live SSE terminal events while `_isLoadingBuffer` is true, then on + * completion DISCARDS the queue (`_loadBufferQueue = null`). That de-dup is + * correct for an established session (the fetched buffer already contains the + * queued output, so replaying it would duplicate Ink redraws). But for a + * brand-new shell the fetch resolves BEFORE the PTY emits its prompt — the + * fetched buffer is empty and the prompt arrives only as a queued event, which + * then gets discarded → blank terminal. + * + * Fix: `_finishBufferLoad(owner, { flushQueued })` REPLAYS the queued events + * through `batchTerminalWrite()` (after `_isLoadingBuffer` is cleared, so they + * write through normally) ONLY when the load painted nothing. The default path + * (no opts) still discards, preserving de-dup for established sessions. + * + * Loaded via `vm` with a stubbed context (no jsdom — jsdom is broken on this + * box; see connection-indicator.test.ts). We extract the REAL + * `_beginBufferLoad`/`_finishBufferLoad` mixin methods from terminal-ui.js by + * running it against a fake `CodemanApp` and capturing `CodemanApp.prototype`, + * then copy them onto a minimal stub whose `batchTerminalWrite` is a spy. This + * exercises the real flush/discard logic without a full xterm fake. + */ +import { readFileSync } from 'node:fs'; +import { performance } from 'node:perf_hooks'; +import { resolve } from 'node:path'; +import vm from 'node:vm'; +import { describe, expect, it, vi } from 'vitest'; + +/** Run terminal-ui.js in a vm against a fake CodemanApp and return the captured prototype mixin. */ +function loadTerminalMixin(): Record { + const source = readFileSync(resolve(import.meta.dirname, '../src/web/public/terminal-ui.js'), 'utf8'); + const FakeCodemanApp = function () {} as unknown as { prototype: Record }; + const context = vm.createContext({ + console, + performance, + setTimeout, + clearTimeout, + setInterval: vi.fn(), + clearInterval: vi.fn(), + requestAnimationFrame: vi.fn(), + CodemanApp: FakeCodemanApp, + // terminal-ui.js IIFE is invoked with `window`; it reads/writes a few globals. + window: { addEventListener: vi.fn(), removeEventListener: vi.fn() }, + document: { addEventListener: vi.fn() }, + }); + vm.runInContext(source, context); + return FakeCodemanApp.prototype; +} + +const mixin = loadTerminalMixin(); + +type BufferLoadApp = { + _bufferLoadSeq: number; + _bufferLoadOwner: string | null; + _isLoadingBuffer: boolean; + _loadBufferQueue: string[] | null; + batchTerminalWrite: (data: string) => void; + _beginBufferLoad: (owner?: string) => string; + _finishBufferLoad: (owner?: string, opts?: { flushQueued?: boolean }) => boolean; +}; + +/** + * Minimal stub carrying the buffer-load state plus the REAL begin/finish methods. + * `batchTerminalWrite` is a spy so flushed events are observable without a real + * xterm terminal. The real `batchTerminalWrite` would queue while loading, but + * the flush runs AFTER `_isLoadingBuffer` is cleared, so a spy is faithful here. + */ +function makeApp() { + const writes: string[] = []; + const app: BufferLoadApp = { + _bufferLoadSeq: 0, + _bufferLoadOwner: null, + _isLoadingBuffer: false, + _loadBufferQueue: null, + batchTerminalWrite: vi.fn((data: string) => { + writes.push(data); + }), + _beginBufferLoad: mixin._beginBufferLoad as BufferLoadApp['_beginBufferLoad'], + _finishBufferLoad: mixin._finishBufferLoad as BufferLoadApp['_finishBufferLoad'], + }; + return { app, writes }; +} + +/** Simulate live SSE events arriving while a buffer load is in progress (the queue path). */ +function pushWhileLoading(app: BufferLoadApp, data: string) { + // Mirrors batchTerminalWrite's queue branch: if loading, push to the queue. + if (app._isLoadingBuffer && app._loadBufferQueue) app._loadBufferQueue.push(data); +} + +describe('buffer-load flush (COD-144)', () => { + it('finish WITHOUT flushQueued discards the queue (de-dup preserved for established sessions)', () => { + const { app, writes } = makeApp(); + const owner = app._beginBufferLoad('load-1'); + pushWhileLoading(app, 'chunk-a'); + pushWhileLoading(app, 'chunk-b'); + + const ok = app._finishBufferLoad(owner); // default: discard + expect(ok).toBe(true); + expect(app._isLoadingBuffer).toBe(false); + expect(app._loadBufferQueue).toBeNull(); + // Queued events were NOT replayed. + expect(app.batchTerminalWrite).not.toHaveBeenCalled(); + expect(writes).toEqual([]); + }); + + it('finish WITH { flushQueued: true } replays queued events in order, exactly once each', () => { + const { app, writes } = makeApp(); + const owner = app._beginBufferLoad('load-2'); + pushWhileLoading(app, 'prompt-1'); + pushWhileLoading(app, 'prompt-2'); + + const ok = app._finishBufferLoad(owner, { flushQueued: true }); + expect(ok).toBe(true); + expect(app._isLoadingBuffer).toBe(false); + expect(app._loadBufferQueue).toBeNull(); + // Both chunks replayed, IN ORDER, exactly once each. + expect(writes).toEqual(['prompt-1', 'prompt-2']); + expect(app.batchTerminalWrite).toHaveBeenCalledTimes(2); + expect(app.batchTerminalWrite).toHaveBeenNthCalledWith(1, 'prompt-1'); + expect(app.batchTerminalWrite).toHaveBeenNthCalledWith(2, 'prompt-2'); + }); + + it('flushed events are not re-queued (the queue is null when batchTerminalWrite runs)', () => { + const { app } = makeApp(); + const owner = app._beginBufferLoad('load-3'); + pushWhileLoading(app, 'only'); + + // Spy that, like the real method, would re-queue if loading were still active. + let reQueued = false; + app.batchTerminalWrite = vi.fn((data: string) => { + if (app._isLoadingBuffer && app._loadBufferQueue) { + app._loadBufferQueue.push(data); + reQueued = true; + } + }); + + app._finishBufferLoad(owner, { flushQueued: true }); + expect(reQueued).toBe(false); + expect(app._isLoadingBuffer).toBe(false); + expect(app._loadBufferQueue).toBeNull(); + }); + + it('owner mismatch returns false and does NOT flush or clear state', () => { + const { app, writes } = makeApp(); + app._beginBufferLoad('real-owner'); + pushWhileLoading(app, 'queued'); + + const ok = app._finishBufferLoad('wrong-owner', { flushQueued: true }); + expect(ok).toBe(false); + // State untouched — still loading, queue intact, nothing replayed. + expect(app._isLoadingBuffer).toBe(true); + expect(app._bufferLoadOwner).toBe('real-owner'); + expect(app._loadBufferQueue).toEqual(['queued']); + expect(app.batchTerminalWrite).not.toHaveBeenCalled(); + expect(writes).toEqual([]); + }); + + it('empty queue + flushQueued is a no-op (no throw, no writes)', () => { + const { app, writes } = makeApp(); + const owner = app._beginBufferLoad('load-empty'); + // No events queued. + + expect(() => app._finishBufferLoad(owner, { flushQueued: true })).not.toThrow(); + expect(app._isLoadingBuffer).toBe(false); + expect(app._loadBufferQueue).toBeNull(); + expect(app.batchTerminalWrite).not.toHaveBeenCalled(); + expect(writes).toEqual([]); + }); +});