Merge PR #149 from aakhter/cod-165-ws-resilience: WebSocket durable-delivery resilience

Includes review fixes: real _wsState lifecycle (connecting/connected/disconnected), per-tab supersede identity (multi-tab coexistence), preserved reconnect backoff, connection-dot CSS for connected/fallback states.
This commit is contained in:
Codeman maintainer
2026-07-12 20:07:55 +02:00
13 changed files with 1994 additions and 214 deletions
+204 -43
View File
@@ -455,6 +455,14 @@ class CodemanApp {
this._clientId = (typeof crypto !== 'undefined' && crypto.randomUUID)
? crypto.randomUUID()
: 'c-' + Math.random().toString(36).slice(2) + Date.now().toString(36);
// Per-TAB nonce for the WS registry key (COD-137). _loadReliableState()
// later replaces _clientId with the browser-wide localStorage identity
// (shared by every tab/window of this profile), so the WS upgrade sends
// `clientId:nonce` instead — a same-tab reconnect still supersedes its own
// socket, but two tabs on one session coexist instead of evicting each
// other in a 4010 ping-pong. Input frames keep the bare clientId for seq
// dedup.
this._wsTabNonce = this._clientId;
this.terminal = null;
this.fitAddon = null;
this.activeSessionId = null;
@@ -589,6 +597,8 @@ 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._wsLastRecvAt = 0; // ms timestamp of the last frame received on the active WS
// Terminal write batching with DEC 2026 sync support
this.pendingWrites = [];
@@ -632,6 +642,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
@@ -2142,9 +2155,21 @@ class CodemanApp {
*/
_connectWs(sessionId) {
this._disconnectWs();
this._wsState = 'connecting';
this._updateConnectionIndicator();
const proto = location.protocol === 'https:' ? 'wss:' : 'ws:';
const url = `${proto}//${location.host}/ws/sessions/${sessionId}/terminal`;
// Pass a per-TAB identity on the upgrade URL so the server's connection
// registry scopes the per-session limit by connection (COD-137): a same-tab
// reconnect supersedes its own socket instead of consuming a new slot and
// tripping a spurious 4008, while two tabs of the same browser (which share
// the localStorage clientId) each keep their own socket. The bare clientId
// still rides the input frames for seq dedup. Omitted if clientId is
// unavailable (server then treats the upgrade as anonymous — still admitted
// up to the limit).
const cid = this._clientId ? `${this._clientId}:${this._wsTabNonce}` : '';
const cidQuery = cid ? `?cid=${encodeURIComponent(cid)}` : '';
const url = `${proto}//${location.host}/ws/sessions/${sessionId}/terminal${cidQuery}`;
const ws = new WebSocket(url);
this._ws = ws;
this._wsSessionId = sessionId;
@@ -2153,7 +2178,9 @@ class CodemanApp {
// Only mark ready if this is still the intended session
if (this._ws === ws) {
this._wsReady = true;
this._wsState = 'connected';
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
@@ -2168,6 +2195,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') {
@@ -2194,18 +2224,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);
}
}
};
@@ -2217,7 +2282,11 @@ class CodemanApp {
/** Close the active WebSocket connection (if any). */
_disconnectWs() {
this._clearTimer('_wsReconnectTimer');
this._wsReconnectAttempts = 0;
// Deliberately do NOT reset _wsReconnectAttempts here: _connectWs() calls
// this first, so a reset would restart the exponential backoff ladder at
// attempt 0 on every retry (≈0ms tight reconnect loop during an outage).
// ws.onopen zeroes the counter once a connection actually succeeds.
this._wsState = 'disconnected';
this._stopMobileResizeRetry();
if (this._ws) {
this._ws.onclose = null; // Prevent re-entrant cleanup
@@ -2412,7 +2481,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 {
@@ -2420,6 +2495,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);
}
@@ -2530,39 +2617,107 @@ class CodemanApp {
}
}
// 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
// (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') {
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).
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;
}
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') {
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) {
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;
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';
// 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;
}
this._lastIndicatorDescriptor = next;
const { bytes: totalBytes, count } = this._pendingBytes();
const hasQueue = count > 0;
indicator.style.display = 'flex';
dot.className = 'connection-dot';
const formatBytes = (b) => (b < 1024 ? `${b}B` : `${(b / 1024).toFixed(1)}KB`);
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';
indicator.style.display = next.display;
if (next.display !== 'none') {
dot.className = next.dotClass;
text.textContent = next.text;
indicator.title = next.title;
}
}
@@ -3801,6 +3956,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
@@ -3966,13 +4124,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;
+27
View File
@@ -156,6 +156,30 @@ function shouldAutoWrapTabs(input) {
return scrollWidth > clientWidth + 1;
}
// 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;
@@ -163,6 +187,9 @@ if (typeof window !== 'undefined') {
window.CodemanTabOverflow = {
shouldAutoWrapTabs,
};
window.CodemanWsReconnect = {
plan: planWsReconnect,
};
}
// Scheduler API — prioritize terminal writes over background UI updates.
+9
View File
@@ -626,6 +626,15 @@ body {
flex-shrink: 0;
}
.connection-dot.connected {
background: var(--green);
}
.connection-dot.fallback {
background: var(--yellow);
box-shadow: 0 0 6px var(--yellow);
}
.connection-dot.offline {
background: var(--red);
box-shadow: 0 0 6px var(--red);
+27 -5
View File
@@ -2046,11 +2046,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;
@@ -2061,13 +2075,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;
},
+210 -166
View File
@@ -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,197 +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<string, number>();
/**
* 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<WebSocket>(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;
}
// Enforce per-session connection limit
const currentCount = sessionWsCount.get(id) ?? 0;
if (currentCount >= MAX_WS_PER_SESSION) {
socket.close(4008, 'Too many connections');
return;
}
sessionWsCount.set(id, currentCount + 1);
// 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).
// Swallow socket errors — cleanup happens in 'close'
socket.on('error', () => {});
// 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) });
// Per-connection micro-batch state
let batchChunks: string[] = [];
let batchSize = 0;
let batchTimer: ReturnType<typeof setTimeout> | null = null;
// 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);
});
const flushBatch = () => {
batchTimer = null;
if (batchChunks.length === 0 || socket.readyState !== 1) {
// Per-connection micro-batch state
let batchChunks: string[] = [];
let batchSize = 0;
let batchTimer: ReturnType<typeof setTimeout> | null = null;
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);
// 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 });
}
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 });
} 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<typeof setTimeout> | null = null;
// Heartbeat: detect stale connections (especially through tunnels where
// TCP RST can take minutes to propagate).
let pongTimeout: ReturnType<typeof setTimeout> | 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(() => {
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', () => {
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;
if (count <= 1) {
sessionWsCount.delete(id);
} else {
sessionWsCount.set(id, count - 1);
}
});
});
// 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),
});
});
}
);
}
+123
View File
@@ -0,0 +1,123 @@
/**
* @fileoverview Per-session WebSocket connection registry (COD-137).
*
* Replaces the bare `Map<sessionId, number>` 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 a per-TAB
* connection identity (`cid`, parsed from the upgrade URL query). The browser
* sends `clientId:tabNonce`, NOT the bare localStorage clientId — that one is
* shared by every tab/window of a profile, so keying on it would make two tabs
* on one session evict each other in a 4010 ping-pong. 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). The cid is opaque here; input-frame
* dedup uses the bare clientId separately (`session.shouldApplyInput`).
*
* 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<S> {
/** 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<S> {
socket: S;
cid: string | null;
}
export class WsConnectionRegistry<S extends RegistrableSocket = RegistrableSocket> {
/** sessionId → live entries (keyed + anonymous). */
private readonly bySession = new Map<string, Entry<S>[]>();
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<S> {
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;
}
}
+460
View File
@@ -0,0 +1,460 @@
/**
* @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.
*
* 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';
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<typeof fetch>) => 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<string, Array<{ seq: number; data: string }>>;
_connectionStatus: string;
_wsState: string;
activeSessionId: string | null;
isOnline: boolean;
_updateConnectionIndicator: () => void;
};
function makeApp(overrides: Partial<Indicator> & { queuedBytes?: number } = {}) {
const app = Object.create((CodemanApp as { prototype: object }).prototype) as Indicator & {
queuedBytes?: number;
};
const els: Record<string, ReturnType<typeof fakeElement>> = {
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<string, number>;
_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);
});
});
// ---- 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<Indicator> & { 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',
});
});
});
describe('connection-dot CSS — every emitted dot class has a styles.css rule', () => {
// The descriptor emits these dot variants; each needs a visible rule or the
// 8px dot renders as an invisible blob (the base .connection-dot rule has no
// background). 'connected' and 'fallback' were missing when this PR shipped.
const DOT_CLASSES = ['connected', 'fallback', 'offline', 'reconnecting', 'draining'];
const css = readFileSync(resolve(import.meta.dirname, '../src/web/public/styles.css'), 'utf8');
for (const cls of DOT_CLASSES) {
it(`.connection-dot.${cls} is styled`, () => {
const rule = new RegExp(`\\.connection-dot\\.${cls}\\s*\\{[^}]*background`, 'm');
expect(css).toMatch(rule);
});
}
});
/** 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<Indicator> & { 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<string, ReturnType<typeof countingElement>>)[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');
});
});
+234
View File
@@ -0,0 +1,234 @@
/**
* @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<typeof fetch>) => 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<string, Array<{ seq: number; data: string; sentAt: number }>>;
_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<string, unknown>;
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<void>((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();
});
});
// 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']);
});
});
+23
View File
@@ -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 ==========
+172
View File
@@ -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<string, unknown> {
const source = readFileSync(resolve(import.meta.dirname, '../src/web/public/terminal-ui.js'), 'utf8');
const FakeCodemanApp = function () {} as unknown as { prototype: Record<string, unknown> };
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([]);
});
});
+133
View File
@@ -0,0 +1,133 @@
/**
* @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('two tabs of the same browser (shared clientId, distinct tab nonce) coexist without eviction', () => {
// The client keys the upgrade by `clientId:tabNonce`, NOT the bare
// browser-wide clientId — otherwise two windows on one session would
// supersede each other in a perpetual 4010/5s reconnect ping-pong.
const reg = new WsConnectionRegistry(5);
const tabA = sock('tab-a');
const tabB = sock('tab-b');
expect(reg.register('s1', 'c-browser:tab-A', tabA).evictedSocket).toBeUndefined();
const resB = reg.register('s1', 'c-browser:tab-B', tabB);
expect(resB.admitted).toBe(true);
expect(resB.evictedSocket).toBeUndefined(); // tab A keeps its socket
expect(reg.liveCount('s1')).toBe(2);
// A genuine same-tab reconnect still supersedes only its own socket.
const tabANew = sock('tab-a-new');
const res = reg.register('s1', 'c-browser:tab-A', tabANew);
expect(res.evictedSocket).toBe(tabA);
expect(reg.liveCount('s1')).toBe(2);
});
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);
});
});
+79
View File
@@ -0,0 +1,79 @@
/**
* 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).
*/
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 });
});
});
+293
View File
@@ -0,0 +1,293 @@
/**
* @fileoverview Terminal WebSocket state-machine lifecycle tests
* (`CodemanApp._connectWs` / `ws.onopen` / `ws.onclose` / `_disconnectWs`).
*
* Unlike test/connection-indicator.test.ts (which pins the pure descriptor per
* pre-seeded `_wsState`), this suite drives the REAL transitions through a fake
* `WebSocket` class so the production assignments are covered:
*
* 1. `_connectWs()` → 'connecting', a real `onopen` → 'connected' (the chip
* renders "WS"), `_disconnectWs()` → 'disconnected'. Regression guard for
* the PR-review blocker where `_wsState` was only ever written in
* `onclose`, leaving the chip stuck on "WS…"/"HTTP" forever.
* 2. Exponential backoff really escalates across the onclose → timer →
* `_connectWs` cycle: `_disconnectWs()` (called first by `_connectWs`)
* must NOT zero `_wsReconnectAttempts`, or every retry replans at
* attempt 0 (a ~0ms tight reconnect loop during an outage). Only a
* successful `onopen` resets the counter.
* 3. The upgrade URL carries the per-TAB `cid` (`clientId:tabNonce`), not the
* browser-wide clientId — two tabs of one profile must register distinct
* registry keys so they coexist instead of 4010-evicting each other.
*
* 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';
type FakeTimer = { id: number; fn: () => void; delay: number; cleared: boolean };
class FakeWebSocket {
static OPEN = 1;
url: string;
readyState = 0;
closed = false;
onopen: (() => void) | null = null;
onmessage: ((e: unknown) => void) | null = null;
onclose: ((e: { code: number; reason: string }) => void) | null = null;
onerror: (() => void) | null = null;
constructor(url: string) {
this.url = url;
FakeWebSocket.instances.push(this);
}
send(): void {}
close(): void {
this.closed = true;
this.readyState = 3;
}
static instances: FakeWebSocket[] = [];
}
function loadHarness() {
FakeWebSocket.instances = [];
const timers: FakeTimer[] = [];
let nextTimerId = 1;
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: (fn: () => void, delay: number) => {
const id = nextTimerId++;
timers.push({ id, fn, delay, cleared: false });
return id;
},
clearTimeout: (id: number) => {
const t = timers.find((x) => x.id === id);
if (t) t.cleared = true;
},
requestAnimationFrame: vi.fn(),
HTMLCanvasElement: class HTMLCanvasElement {},
WebSocket: FakeWebSocket,
location: { protocol: 'https:', host: 'test.local' },
fetch: (...args: Parameters<typeof fetch>) => 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);
const CodemanApp = (context as { __CodemanApp: new () => unknown }).__CodemanApp;
return { CodemanApp, timers };
}
function fakeElement() {
return { style: { display: '' }, title: '', textContent: '', className: '' };
}
type LifecycleApp = {
_connectWs: (id: string) => void;
_disconnectWs: () => void;
_wsState: string;
_wsReady: boolean;
_wsReconnectAttempts: number | undefined;
_ws: FakeWebSocket | null;
activeSessionId: string | null;
};
function makeApp(
CodemanApp: new () => unknown,
overrides: Record<string, unknown> = {}
): { app: LifecycleApp; els: Record<string, ReturnType<typeof fakeElement>> } {
const app = Object.create((CodemanApp as { prototype: object }).prototype) as LifecycleApp & Record<string, unknown>;
const els: Record<string, ReturnType<typeof fakeElement>> = {
connectionIndicator: fakeElement(),
connectionDot: fakeElement(),
connectionText: fakeElement(),
};
app.$ = (id: string) => els[id];
app._clientId = 'c-browser';
app._wsTabNonce = 'tab-1';
app._ws = null;
app._wsSessionId = null;
app._wsReady = false;
app._wsState = 'disconnected';
app._wsLastRecvAt = 0;
app._lastIndicatorDescriptor = null;
app._pendingDeliveries = new Map();
app._connectionStatus = 'connected';
app.activeSessionId = 's1';
app.isOnline = true;
app.sendResize = vi.fn();
app._onWsReady = vi.fn();
Object.assign(app, overrides);
return { app: app as LifecycleApp, els };
}
/** Run the oldest pending (not-cleared, not-yet-fired) reconnect timer. */
function fireNextTimer(timers: FakeTimer[]): FakeTimer {
const t = timers.find((x) => !x.cleared);
if (!t) throw new Error('no pending timer');
t.cleared = true; // mark consumed so the next fire picks the following one
t.fn();
return t;
}
describe('WS state lifecycle — real _connectWs/onopen/onclose transitions', () => {
it("_connectWs sets 'connecting', a real onopen sets 'connected' and renders 'WS'", () => {
const { CodemanApp } = loadHarness();
const { app, els } = makeApp(CodemanApp);
app._connectWs('s1');
expect(app._wsState).toBe('connecting');
expect(els.connectionText.textContent).toBe('WS…');
const ws = FakeWebSocket.instances[0];
ws.readyState = 1;
ws.onopen?.();
expect(app._wsState).toBe('connected');
expect(app._wsReady).toBe(true);
expect(app._wsReconnectAttempts).toBe(0);
// The chip must show the healthy transport from the REAL open path — the
// 'connected' branch was dead code when only onclose wrote _wsState.
expect(els.connectionText.textContent).toBe('WS');
expect(els.connectionDot.className).toBe('connection-dot connected');
});
it("_disconnectWs resets the state machine to 'disconnected' and closes the socket", () => {
const { CodemanApp } = loadHarness();
const { app } = makeApp(CodemanApp);
app._connectWs('s1');
const ws = FakeWebSocket.instances[0];
ws.readyState = 1;
ws.onopen?.();
expect(app._wsState).toBe('connected');
app._disconnectWs();
expect(app._wsState).toBe('disconnected');
expect(app._wsReady).toBe(false);
expect(app._ws).toBeNull();
expect(ws.closed).toBe(true);
});
it("a retry-fallback close (4010) shows 'HTTP', and the successful retry returns the chip to 'WS'", () => {
const { CodemanApp, timers } = loadHarness();
const { app, els } = makeApp(CodemanApp);
app._connectWs('s1');
const ws1 = FakeWebSocket.instances[0];
ws1.readyState = 1;
ws1.onopen?.();
expect(els.connectionText.textContent).toBe('WS');
ws1.onclose?.({ code: 4010, reason: 'Superseded by reconnect' });
expect(app._wsState).toBe('fallback');
expect(els.connectionText.textContent).toBe('HTTP');
// The bounded 5s retry succeeds → the chip must NOT stay stuck on "HTTP".
const timer = fireNextTimer(timers);
expect(timer.delay).toBe(5000);
const ws2 = FakeWebSocket.instances[1];
ws2.readyState = 1;
ws2.onopen?.();
expect(app._wsState).toBe('connected');
expect(els.connectionText.textContent).toBe('WS');
});
});
describe('WS reconnect backoff — attempts survive the _connectWs → _disconnectWs call', () => {
it('escalates the transient-close delay ladder instead of replanning at attempt 0', () => {
const { CodemanApp, timers } = loadHarness();
const { app } = makeApp(CodemanApp);
app._connectWs('s1');
// Attempt 0: transient close plans 0ms (+ <250ms jitter).
FakeWebSocket.instances[0].onclose?.({ code: 1006, reason: '' });
expect(app._wsReconnectAttempts).toBe(1);
const t1 = fireNextTimer(timers);
expect(t1.delay).toBeLessThan(250);
// Attempt 1: the retry's _connectWs ran _disconnectWs first — the counter
// must survive it, so this close plans 250ms (+ jitter), not 0ms again.
FakeWebSocket.instances[1].onclose?.({ code: 1006, reason: '' });
expect(app._wsReconnectAttempts).toBe(2);
const t2 = fireNextTimer(timers);
expect(t2.delay).toBeGreaterThanOrEqual(250);
expect(t2.delay).toBeLessThan(500);
// Attempt 2 → 500ms rung.
FakeWebSocket.instances[2].onclose?.({ code: 1006, reason: '' });
expect(app._wsReconnectAttempts).toBe(3);
const t3 = fireNextTimer(timers);
expect(t3.delay).toBeGreaterThanOrEqual(500);
expect(t3.delay).toBeLessThan(750);
});
it('a successful onopen (not an intentional disconnect) is what resets the counter', () => {
const { CodemanApp, timers } = loadHarness();
const { app } = makeApp(CodemanApp);
app._connectWs('s1');
FakeWebSocket.instances[0].onclose?.({ code: 1006, reason: '' });
FakeWebSocket.instances[0].closed = true;
fireNextTimer(timers);
expect(app._wsReconnectAttempts).toBe(1);
const ws2 = FakeWebSocket.instances[1];
ws2.readyState = 1;
ws2.onopen?.();
expect(app._wsReconnectAttempts).toBe(0);
expect(app._wsState).toBe('connected');
});
});
describe('WS upgrade cid — per-TAB identity (clientId:tabNonce)', () => {
it('sends the composite cid on the upgrade URL, keeping the bare clientId for input frames', () => {
const { CodemanApp } = loadHarness();
const { app } = makeApp(CodemanApp);
app._connectWs('s1');
const url = new URL(FakeWebSocket.instances[0].url);
expect(url.searchParams.get('cid')).toBe('c-browser:tab-1');
});
it('two tabs sharing the browser clientId register DIFFERENT registry keys', () => {
const { CodemanApp } = loadHarness();
const { app: tabA } = makeApp(CodemanApp, { _wsTabNonce: 'tab-A' });
const { app: tabB } = makeApp(CodemanApp, { _wsTabNonce: 'tab-B' });
tabA._connectWs('s1');
tabB._connectWs('s1');
const cidA = new URL(FakeWebSocket.instances[0].url).searchParams.get('cid');
const cidB = new URL(FakeWebSocket.instances[1].url).searchParams.get('cid');
expect(cidA).toBe('c-browser:tab-A');
expect(cidB).toBe('c-browser:tab-B');
// Distinct keys → the server registry admits both instead of supersede-evicting.
expect(cidA).not.toBe(cidB);
});
it('omits the cid query entirely when no clientId is available', () => {
const { CodemanApp } = loadHarness();
const { app } = makeApp(CodemanApp, { _clientId: '' });
app._connectWs('s1');
expect(FakeWebSocket.instances[0].url).not.toContain('cid=');
});
});