From 9262f2b49acbc1f2587fceb27abbfc8ebf9a45f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 22 Jul 2026 13:46:19 +0200 Subject: [PATCH] fix(cli): recover remote session connection that silently dies and never reconnects (#12393) * fix(cli): bound token and connection attempts with a single fenced retry owner * fix(cli): bounded heartbeat gather with freshness-fenced attach * chore: changeset for CLI live reconnect fix * fix(cli): harden heartbeat gather slot release and connection-close signalling * fix(cli): ignore late token continuation for a settled connection generation open() only bailed on `closed` after awaiting getToken(); if the connect- attempt deadline settled the generation while the token was still pending, the late continuation would construct and assign a WebSocket for an expired generation, clobbering the newer generation's live socket. Bail on `g.settled` too so a settled generation never builds a socket. Also make the AC2a token-rejection test genuinely exercise rejection (flush the rejection before the deadline can fire) and add AC3g covering the late- token continuation. * fix(cli): fence attach announcements on heartbeat session-id containment An attach announcement resolved on any fresh heartbeat, even one whose gathered session list omitted the announced id (getSessions() drops a session whose per-id lookup fails via Effect.orElseSucceed while the gather still succeeds). That falsely reported a session as attached when the relay never received it. Thread an optional requireSessionId through Connection.heartbeat: an id-gated waiter now resolves only when a fresh heartbeat whose payload contains that id is actually sent, and is otherwise requeued for the next fresh cycle (rejecting on connection shutdown). announce(id) forwards the id; presence heartbeats remain id-agnostic. Adds AC4a (degraded heartbeat preserves last known-good sessions), AC6d (id-containment fence), and AC6e (in-flight-cycle waiter rejects on permanent close). * ci: retrigger checks after GitHub Actions incident recovery --- .changeset/cli-live-reconnect.md | 5 + .../src/kilo-sessions/attached-state.ts | 14 +- .../src/kilo-sessions/kilo-sessions.ts | 6 +- .../opencode/src/kilo-sessions/remote-ws.ts | 488 ++++-- .../kilocode/sessions/attached-state.test.ts | 25 + .../test/kilocode/sessions/remote-ws.test.ts | 1458 ++++++++++++++++- 6 files changed, 1856 insertions(+), 140 deletions(-) create mode 100644 .changeset/cli-live-reconnect.md diff --git a/.changeset/cli-live-reconnect.md b/.changeset/cli-live-reconnect.md new file mode 100644 index 0000000000..5f2536c91e --- /dev/null +++ b/.changeset/cli-live-reconnect.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": patch +--- + +Remote CLI sessions no longer appear frozen on mobile when the connection to the session relay stalls; they now recover on their own instead of staying read-only until the CLI is restarted. Token acquisition and connection attempts are bounded by deadlines with a single fenced retry owner, and heartbeat session gathers are bounded so one stuck gather can no longer silently kill every future heartbeat. diff --git a/packages/opencode/src/kilo-sessions/attached-state.ts b/packages/opencode/src/kilo-sessions/attached-state.ts index a9af1638e2..3afad1cd05 100644 --- a/packages/opencode/src/kilo-sessions/attached-state.ts +++ b/packages/opencode/src/kilo-sessions/attached-state.ts @@ -30,8 +30,13 @@ export namespace AttachedState { export type Options = { /** Fires the relay heartbeat. May be fire-and-forget or awaited. Must * reject (not resolve) when no relay connection is available so that - * `announce` cannot silently mark a session as attached. */ - heartbeat: () => Promise + * `announce` cannot silently mark a session as attached. + * + * `opts.requireSessionId` is forwarded by `announce(id)` so the relay + * only resolves the attach promise when a fresh heartbeat whose + * payload contains that id was actually sent. Presence fire-and- + * forget heartbeats call without an id and resolve on any fresh send. */ + heartbeat: (opts?: { requireSessionId?: string }) => Promise log?: { warn: (msg: string, meta?: unknown) => void } } @@ -141,7 +146,10 @@ export namespace AttachedState { const owned = (async () => { pending.add(id) try { - await options.heartbeat() + // kilocode_change - forward the announced id so the relay only + // resolves the attach once a fresh heartbeat whose payload + // contains this id was actually sent (id-containment fence). + await options.heartbeat({ requireSessionId: id }) } catch (err) { // Roll back only the entry this call added. If presence adopted // the id while the heartbeat was in flight, presence is the diff --git a/packages/opencode/src/kilo-sessions/kilo-sessions.ts b/packages/opencode/src/kilo-sessions/kilo-sessions.ts index b1cbd36b16..fc981f2b12 100644 --- a/packages/opencode/src/kilo-sessions/kilo-sessions.ts +++ b/packages/opencode/src/kilo-sessions/kilo-sessions.ts @@ -221,8 +221,10 @@ export namespace KiloSessions { // create_session's catch block turns that into the sanitized failure // response and the user retries manually. const attachedState = AttachedState.create({ - heartbeat: () => - remote ? remote.conn.heartbeat() : Promise.reject(new Error("attachRemoteSession: no remote connection")), + heartbeat: (opts) => + remote + ? remote.conn.heartbeat(opts) + : Promise.reject(new Error("attachRemoteSession: no remote connection")), log: attachedLog, }) // kilocode_change end diff --git a/packages/opencode/src/kilo-sessions/remote-ws.ts b/packages/opencode/src/kilo-sessions/remote-ws.ts index d755ea27f0..0e8c6187c6 100644 --- a/packages/opencode/src/kilo-sessions/remote-ws.ts +++ b/packages/opencode/src/kilo-sessions/remote-ws.ts @@ -4,6 +4,13 @@ import { InstallationVersion } from "@opencode-ai/core/installation/version" export namespace RemoteWS { export type SessionInfo = RemoteProtocol.SessionInfo + export type Timers = { + setTimeout: (fn: () => void, ms?: number) => unknown + clearTimeout: (t: unknown) => void + setInterval: (fn: () => void, ms?: number) => unknown + clearInterval: (t: unknown) => void + } + export type Options = { url: string getToken: () => Promise @@ -23,79 +30,272 @@ export namespace RemoteWS { onClose?: (code: number, reason: string) => void /** Inactivity timeout in ms — force-close if no inbound message within this window */ timeout?: number + /** Injectable timer primitives for deterministic testing. Defaults to globals. */ + timers?: Timers + /** Injectable clock for deterministic testing. Defaults to Date.now. */ + now?: () => number + /** Token-acquisition deadline in ms. Defaults to 15_000. */ + tokenTimeout?: number + /** Connection-attempt deadline (token acquisition through onopen) in ms. Defaults to 30_000. */ + connectTimeout?: number + /** Session-gather deadline in ms. Defaults to 15_000. */ + gatherTimeout?: number + /** Max unresolved gather operations before cycles send degraded heartbeats. Defaults to 4. */ + maxOutstandingGathers?: number } export type Connection = { readonly connectionId: string send(msg: RemoteProtocol.Outbound): void - heartbeat(): Promise + /** + * Resolves when a heartbeat built from a FRESH gather has actually been + * sent over a live socket. Degraded sends and non-live buffered sends + * leave the returned promise pending; `close()` rejects it. + * + * When `opts.requireSessionId` is provided, the promise only resolves + * when the sent fresh payload's session list contains that id. This + * fences attach-announce waiters so a fresh heartbeat that legitimately + * omits the announced id (e.g. the gather's `Effect.orElseSucceed` + * filtered it out) does not falsely report the session as attached. + */ + heartbeat(opts?: { requireSessionId?: string }): Promise close(): void readonly connected: boolean } - type Timer = ReturnType + const defaultTimers: Timers = { + setTimeout: (fn, ms) => setTimeout(fn, ms), + clearTimeout: (t) => clearTimeout(t as ReturnType), + setInterval: (fn, ms) => setInterval(fn, ms), + clearInterval: (t) => clearInterval(t as ReturnType), + } + + type Gen = { id: number; settled: boolean; opened: boolean } export function connect(options: Options): Connection { const interval = options.heartbeat ?? 10_000 const connectionId = crypto.randomUUID() const withContext = options.withContext ?? ((fn) => fn()) + const timers = options.timers ?? defaultTimers + const now = options.now ?? Date.now + const tokenTimeout = options.tokenTimeout ?? 15_000 + const connectTimeout = options.connectTimeout ?? 30_000 let ws: WebSocket | undefined let backoff = 1000 - let timer: Timer | undefined - let beat: Timer | undefined + let timer: unknown + let beat: unknown let closed = false const buffer: string[] = [] let beating: Promise | undefined let queued = false + // --- Bounded heartbeat gather with freshness-fenced attach (Path D fix) --- + // A single never-settling getSessions() must not permanently kill heartbeats. + // Each cycle bounds the gather with a deadline; on failure it sends a + // "degraded" heartbeat carrying the last known-good session list so + // server-side liveness is preserved even when metadata is stale. Callers + // awaiting heartbeat() (session attach announcements) resolve ONLY when a + // heartbeat built from a FRESH gather is actually sent over a live socket; + // degraded sends leave them pending, a transient reconnect keeps them + // pending (they resolve on the next fresh send over the new socket), and + // Connection.close() rejects them. + // + // Degraded-mode limitation (deliberate, bounded by recovery): while gathers + // keep failing, session membership/status/title/gitUrl/gitBranch can be + // stale indefinitely, and sessions created or closed during degradation are + // reflected only after the first fresh gather succeeds. + // + // Attach-announce fencing (AC6): a waiter registered with + // `requireSessionId` stays pending until a fresh heartbeat whose + // payload contains that id is sent over a live socket. A fresh + // gather whose session list omits the required id (e.g. the + // upstream `get(id)` was filtered by `Effect.orElseSucceed`) does + // NOT resolve the waiter — it is requeued and re-evaluated on the + // next fresh cycle. The periodic interval keeps calling + // `requestCycle`, so recovery is automatic once a fresh gather + // includes the id. Permanent close (Connection.close) rejects the + // waiter. + const gatherTimeout = options.gatherTimeout ?? 15_000 + const maxOutstandingGathers = options.maxOutstandingGathers ?? 4 + let lastGood: SessionInfo[] | undefined + let outstanding = 0 + let degradedCount = 0 + type Waiter = { resolve: () => void; reject: (err: unknown) => void; requireSessionId?: string } + let waiters: Waiter[] = [] - function heartbeat(): Promise { - queued = true - if (beating) return beating + function makeWaiter(): { promise: Promise; waiter: Waiter } { + let resolve!: () => void + let reject!: (err: unknown) => void + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) + return { promise, waiter: { resolve, reject } } + } - const current = Promise.resolve( - withContext(async () => { - while (queued) { - if (closed) return - queued = false - const sessions = await options.getSessions() - if (closed) return - send({ type: "heartbeat", protocolVersion: InstallationVersion, ...sessions }) - } - }), - ).finally(() => { - beating = undefined - if (!queued || closed) return - void heartbeat().catch((err) => { - options.log.error("remote-ws heartbeat failed", { error: String(err) }) + function rejectWaiters(list: Waiter[], err: unknown) { + for (const w of list) w.reject(err) + } + + // One bounded gather. Never throws. Returns the fresh session list, or + // undefined to signal a degraded cycle (caller sends last known-good). + async function gatherOnce(): Promise { + if (outstanding >= maxOutstandingGathers) { + degradedCount++ + options.log.warn("remote-ws heartbeat gather cap reached, degraded heartbeat", { + outstanding, + degraded: degradedCount, + }) + return undefined + } + outstanding++ + let released = false + const release = () => { + if (released) return + released = true + outstanding-- + } + // Free the slot on ANY settle — success, rejection, or a late settle after + // this cycle abandoned it on timeout. A late result is never read below, + // so an abandoned gather's eventual value is discarded, never emitted. + const normalized = Promise.resolve() + .then(() => options.getSessions()) + .then( + (r) => { + release() + return { ok: true as const, sessions: r.sessions } + }, + (err) => { + release() + return { ok: false as const, error: err } + }, + ) + const outcome = await new Promise< + { kind: "ok"; sessions: SessionInfo[] } | { kind: "err"; error: unknown } | { kind: "timeout" } + >((resolve) => { + let done = false + const t = timers.setTimeout(() => { + if (done) return + done = true + resolve({ kind: "timeout" }) + }, gatherTimeout) + void normalized.then((res) => { + if (done) return + done = true + timers.clearTimeout(t) + resolve(res.ok ? { kind: "ok", sessions: res.sessions } : { kind: "err", error: res.error }) }) }) - beating = current - return current + if (outcome.kind === "ok") return outcome.sessions + degradedCount++ + if (outcome.kind === "err") { + options.log.warn("remote-ws heartbeat gather rejected, degraded heartbeat", { + error: String(outcome.error), + degraded: degradedCount, + }) + } else { + options.log.warn("remote-ws heartbeat gather timeout, degraded heartbeat", { + outstanding, + degraded: degradedCount, + }) + } + return undefined + } + + function heartbeat(opts?: { requireSessionId?: string }): Promise { + if (closed) return Promise.reject(new Error("remote-ws connection closed")) + const { promise, waiter } = makeWaiter() + waiter.requireSessionId = opts?.requireSessionId + waiters.push(waiter) + requestCycle() + return promise + } + + // Interval-driven ticks call requestCycle directly so the periodic heartbeat + // never registers a waiter (no waiter accumulation during degradation). + function requestCycle() { + queued = true + runLoop() + } + + function runLoop() { + if (beating || closed) return + beating = Promise.resolve( + withContext(async () => { + while (queued && !closed) { + queued = false + const cycleWaiters = waiters + waiters = [] + const fresh = await gatherOnce() + if (closed) { + rejectWaiters(cycleWaiters, new Error("remote-ws connection closed")) + return + } + if (fresh !== undefined) { + lastGood = fresh + const sentLive = ws?.readyState === WebSocket.OPEN + send({ type: "heartbeat", protocolVersion: InstallationVersion, sessions: fresh }) + if (sentLive) { + // A waiter requiring a specific id is satisfied only when + // the sent payload contains that id. Unsatisfied waiters + // are requeued so the periodic interval keeps evaluating + // them; they resolve on a future fresh send whose payload + // includes their required id (or reject on close). + const satisfied: Waiter[] = [] + const unsatisfied: Waiter[] = [] + for (const w of cycleWaiters) { + if ( + w.requireSessionId === undefined || + fresh.some((s) => s.id === w.requireSessionId) + ) { + satisfied.push(w) + } else { + unsatisfied.push(w) + } + } + for (const w of satisfied) w.resolve() + if (unsatisfied.length > 0) { + waiters = unsatisfied.concat(waiters) + } + } else { + // Buffered because the socket is not open; resolve on the next + // fresh send over the (re)connected socket. + waiters = cycleWaiters.concat(waiters) + } + } else { + // Degraded: preserve liveness with the last known-good list (empty + // on cold start) and keep waiters pending for a future fresh send. + send({ type: "heartbeat", protocolVersion: InstallationVersion, sessions: lastGood ?? [] }) + waiters = cycleWaiters.concat(waiters) + } + } + }), + ) + .catch((err) => options.log.error("remote-ws heartbeat loop failed", { error: String(err) })) + .finally(() => { + beating = undefined + if (queued && !closed) runLoop() + }) } function startHeartbeat() { stopHeartbeat() - beat = setInterval(() => { - void heartbeat().catch((err) => { - options.log.error("remote-ws heartbeat failed", { error: String(err) }) - }) - }, interval) + beat = timers.setInterval(() => requestCycle(), interval) } function stopHeartbeat() { - if (beat) clearInterval(beat) + if (beat) timers.clearInterval(beat) beat = undefined } - let activity = Date.now() - let watchdog: Timer | undefined + let activity = now() + let watchdog: unknown const timeout = options.timeout ?? 30_000 function startWatchdog() { stopWatchdog() - watchdog = setInterval( + watchdog = timers.setInterval( () => { - if (Date.now() - activity > timeout) { + if (now() - activity > timeout) { options.log.warn("remote-ws activity timeout, forcing reconnect") stopWatchdog() ws?.close(4000, "activity timeout") @@ -106,88 +306,172 @@ export namespace RemoteWS { } function stopWatchdog() { - if (watchdog) clearInterval(watchdog) + if (watchdog) timers.clearInterval(watchdog) watchdog = undefined } + // Connect-attempt deadline (covers token acquisition through onopen). + let connectDeadline: unknown + let currentGen = 0 + + function startConnectDeadline(g: Gen) { + stopConnectDeadline() + connectDeadline = timers.setTimeout(() => { + connectDeadline = undefined + if (closed || g.settled) return + options.log.warn("remote-ws connect attempt deadline, will retry", { gen: g.id }) + if (ws) ws.close(4001, "connect timeout") + scheduleRetry(g) + }, connectTimeout) + } + + function stopConnectDeadline() { + if (connectDeadline) timers.clearTimeout(connectDeadline) + connectDeadline = undefined + } + + // Single fenced retry owner: exactly one of {token-failure, connect-deadline, + // onclose, sync-throw} may schedule a retry for a given generation. + function scheduleRetry(g: Gen) { + if (closed || g.settled) return + g.settled = true + schedule() + } + + function withTimeout(promise: Promise, ms: number, label: string): Promise { + return new Promise((resolve, reject) => { + let done = false + const t = timers.setTimeout(() => { + if (done) return + done = true + reject(new Error(label)) + }, ms) + promise.then( + (v) => { + if (done) return + done = true + timers.clearTimeout(t) + resolve(v) + }, + (err) => { + if (done) return + done = true + timers.clearTimeout(t) + reject(err) + }, + ) + }) + } + async function open() { if (closed) return - const token = await options.getToken() - if (closed) return - if (!token) { - options.log.warn("remote-ws no token, will retry") - schedule() - return - } - const endpoint = `${options.url}/api/user/cli?token=${encodeURIComponent(token)}&connectionId=${connectionId}` - options.log.info("remote-ws connecting", { connectionId, endpoint: endpoint.replace(/token=[^&]+/, "token=***") }) - const socket = new WebSocket(endpoint) - ws = socket - - socket.onopen = () => { - if (ws !== socket || closed) { - socket.close() - return - } - options.log.info("remote-ws connected", { buffered: buffer.length }) - void withContext(() => options.onOpen?.()) - backoff = 1000 - for (const msg of buffer) socket.send(msg) - buffer.length = 0 - activity = Date.now() - startHeartbeat() - startWatchdog() - } - - socket.onmessage = (event) => { - if (ws !== socket || closed) return - activity = Date.now() - const raw = String(event.data) - let json: unknown + const g: Gen = { id: ++currentGen, settled: false, opened: false } + startConnectDeadline(g) + try { + let token: string | undefined try { - json = JSON.parse(raw) - } catch { - options.log.warn("remote-ws invalid JSON", { bytes: raw.length }) + token = await withTimeout(options.getToken(), tokenTimeout, "remote-ws token timeout") + } catch (err) { + if (closed) return + options.log.warn("remote-ws getToken failed, will retry", { gen: g.id, error: String(err) }) + scheduleRetry(g) return } - const preview = RemoteProtocol.Preview.safeParse(json) - options.log.info("remote-ws received", { bytes: raw.length, ...preview.data }) - const parsed = RemoteProtocol.Inbound.safeParse(json) - if (!parsed.success) { - options.log.warn("remote-ws message parse failed", { error: parsed.error }) + if (closed || g.settled) return + if (!token) { + options.log.warn("remote-ws no token, will retry", { gen: g.id }) + scheduleRetry(g) return } - options.onMessage?.(parsed.data) - } + const endpoint = `${options.url}/api/user/cli?token=${encodeURIComponent(token)}&connectionId=${connectionId}` + options.log.info("remote-ws connecting", { connectionId, gen: g.id, endpoint: endpoint.replace(/token=[^&]+/, "token=***") }) + let socket: WebSocket + try { + socket = new WebSocket(endpoint) + } catch (err) { + if (closed) return + options.log.warn("remote-ws constructor threw, will retry", { gen: g.id, error: String(err) }) + scheduleRetry(g) + return + } + ws = socket - socket.onclose = (event) => { - if (ws !== socket) return - options.log.info("remote-ws closed", { code: event.code, reason: event.reason }) - ws = undefined - stopHeartbeat() - stopWatchdog() + socket.onopen = () => { + if (g.settled || ws !== socket || closed) { + socket.close() + return + } + g.opened = true + stopConnectDeadline() + options.log.info("remote-ws connected", { gen: g.id, buffered: buffer.length }) + void withContext(() => options.onOpen?.()) + backoff = 1000 + for (const msg of buffer) socket.send(msg) + buffer.length = 0 + activity = now() + startHeartbeat() + startWatchdog() + if (waiters.length > 0) requestCycle() + } + + socket.onmessage = (event) => { + if (g.settled || ws !== socket || closed) return + activity = now() + const raw = String(event.data) + let json: unknown + try { + json = JSON.parse(raw) + } catch { + options.log.warn("remote-ws invalid JSON", { bytes: raw.length }) + return + } + const preview = RemoteProtocol.Preview.safeParse(json) + options.log.info("remote-ws received", { bytes: raw.length, ...preview.data }) + const parsed = RemoteProtocol.Inbound.safeParse(json) + if (!parsed.success) { + options.log.warn("remote-ws message parse failed", { error: parsed.error }) + return + } + options.onMessage?.(parsed.data) + } + + socket.onclose = (event) => { + if (ws !== socket) return + stopConnectDeadline() + options.log.info("remote-ws closed", { code: event.code, reason: event.reason, gen: g.id }) + ws = undefined + stopHeartbeat() + stopWatchdog() + if (closed) return + if (event.code === 4401 || event.code === 4403 || event.code === 4409) { + options.log.warn("remote-ws closed permanently", { + code: event.code, + reason: event.reason, + }) + const pending = waiters + waiters = [] + rejectWaiters(pending, new Error("remote-ws connection permanently closed")) + void withContext(() => options.onClose?.(event.code, event.reason)) + return + } + if (g.opened) void withContext(() => options.onDisconnect?.()) + scheduleRetry(g) + } + + socket.onerror = (event) => { + if (g.settled || ws !== socket || closed) return + options.log.error("remote-ws error", { error: event }) + } + } catch (err) { if (closed) return - if (event.code === 4401 || event.code === 4403 || event.code === 4409) { - options.log.warn("remote-ws closed permanently", { - code: event.code, - reason: event.reason, - }) - void withContext(() => options.onClose?.(event.code, event.reason)) - return - } - void withContext(() => options.onDisconnect?.()) - schedule() - } - - socket.onerror = (event) => { - if (ws !== socket || closed) return - options.log.error("remote-ws error", { error: event }) + options.log.warn("remote-ws open threw, will retry", { gen: g.id, error: String(err) }) + scheduleRetry(g) } } function schedule() { if (closed) return - timer = setTimeout(() => open(), backoff) + timer = timers.setTimeout(() => open(), backoff) backoff = Math.min(backoff * 2, 60000) } @@ -206,8 +490,12 @@ export namespace RemoteWS { queued = false stopHeartbeat() stopWatchdog() - if (timer) clearTimeout(timer) + stopConnectDeadline() + if (timer) timers.clearTimeout(timer) if (ws) ws.close() + const pending = waiters + waiters = [] + rejectWaiters(pending, new Error("remote-ws connection closed")) } void open() diff --git a/packages/opencode/test/kilocode/sessions/attached-state.test.ts b/packages/opencode/test/kilocode/sessions/attached-state.test.ts index 676ba79b8c..77ff6ddb08 100644 --- a/packages/opencode/test/kilocode/sessions/attached-state.test.ts +++ b/packages/opencode/test/kilocode/sessions/attached-state.test.ts @@ -744,4 +744,29 @@ describe("AttachedState", () => { await replacement expect([...state.union()].sort()).toEqual(["ses_x"]) }) + + // AC6d: announce(id) must forward { requireSessionId: id } to the + // heartbeat callback so the relay only resolves the attach once a fresh + // heartbeat whose payload contains that id was actually sent. Presence + // fire-and-forget heartbeats (from setPresence) continue to call + // without an id and resolve on any fresh send. + test("announce(id) forwards { requireSessionId: id } to the heartbeat callback", async () => { + const calls: Array<{ requireSessionId?: string }> = [] + const state = AttachedState.create({ + heartbeat: (opts) => { + calls.push(opts ? { ...opts } : {}) + return Promise.resolve() + }, + log: nolog, + }) + + // setPresence fires a fire-and-forget heartbeat with NO id. + state.setPresence(["ses_a"]) + await Promise.resolve() + + // announce(id) forwards the id to the awaited heartbeat. + await state.announce("ses_b") + + expect(calls).toEqual([{}, { requireSessionId: "ses_b" }]) + }) }) diff --git a/packages/opencode/test/kilocode/sessions/remote-ws.test.ts b/packages/opencode/test/kilocode/sessions/remote-ws.test.ts index a4fe6769a0..f114abace6 100644 --- a/packages/opencode/test/kilocode/sessions/remote-ws.test.ts +++ b/packages/opencode/test/kilocode/sessions/remote-ws.test.ts @@ -22,6 +22,115 @@ function capture() { } } +class FakeClock { + now = 0 + private timers: { id: number; fireAt: number; fn: () => void; interval?: number }[] = [] + private nextId = 1 + + setTimeout(fn: () => void, ms = 0) { + const id = this.nextId++ + this.timers.push({ id, fireAt: this.now + ms, fn }) + this.timers.sort((a, b) => a.fireAt - b.fireAt || a.id - b.id) + return id + } + + clearTimeout(id: unknown) { + this.timers = this.timers.filter((t) => t.id !== id) + } + + setInterval(fn: () => void, ms = 0) { + const id = this.nextId++ + this.timers.push({ id, fireAt: this.now + ms, fn, interval: ms }) + this.timers.sort((a, b) => a.fireAt - b.fireAt || a.id - b.id) + return id + } + + clearInterval(id: unknown) { + this.timers = this.timers.filter((t) => t.id !== id) + } + + advance(ms: number) { + const end = this.now + ms + while (true) { + const due = this.timers.filter((t) => t.fireAt <= end) + if (due.length === 0) { + this.now = end + return + } + const next = due[0] + this.now = next.fireAt + this.timers = this.timers.filter((t) => t.id !== next.id) + if (next.interval !== undefined) { + next.fireAt = this.now + next.interval + this.timers.push(next) + this.timers.sort((a, b) => a.fireAt - b.fireAt || a.id - b.id) + } + next.fn() + } + } +} + +class FakeWebSocket { + static readonly OPEN = 1 + static readonly CONNECTING = 0 + static readonly CLOSED = 3 + static instances: FakeWebSocket[] = [] + + static reset() { + this.instances = [] + } + + readonly sent: string[] = [] + readyState = FakeWebSocket.CONNECTING + onopen: (() => void) | null = null + onmessage: ((event: { data: string }) => void) | null = null + onclose: ((event: { code: number; reason: string }) => void) | null = null + onerror: ((event: unknown) => void) | null = null + + constructor(readonly url: string) { + FakeWebSocket.instances.push(this) + } + + send(message: string) { + this.sent.push(message) + } + + close(code = 1000, reason = "closed") { + if (this.readyState === FakeWebSocket.CLOSED) return + this.readyState = FakeWebSocket.CLOSED + this.onclose?.({ code, reason }) + } + + open() { + if (this.readyState !== FakeWebSocket.CONNECTING) return + this.readyState = FakeWebSocket.OPEN + this.onopen?.() + } + + disconnect(code = 1000, reason = "closed") { + this.close(code, reason) + } + + receive(data: unknown) { + this.onmessage?.({ data: JSON.stringify(data) }) + } +} + +async function flush() { + // Flush a few microtask ticks to let async getToken / onopen chains settle. + await Promise.resolve() + await Promise.resolve() + await Promise.resolve() +} + +async function flushLong(iterations = 10) { + // Flush enough microtask ticks for a full gather cycle to settle: + // .then handler on getSessions → .then handler on normalized → resolve inner + // Promise → await resume in gatherOnce → await resume in while loop body → + // .finally on runLoop's beating Promise. + for (let i = 0; i < iterations; i++) await Promise.resolve() +} + function createServer() { const messages: string[] = [] const clients: ServerWebSocket[] = [] @@ -97,6 +206,12 @@ describe("RemoteWS", () => { server?.stop() }) + // Fire a heartbeat() and swallow its rejection (e.g. on close()) so the + // discarded promise cannot become an unhandled rejection. + function fireHeartbeat() { + void conn?.heartbeat().catch(() => {}) + } + test("connects and sends heartbeat", async () => { server = createServer() const connecting = server.waitForConnect() @@ -253,38 +368,7 @@ describe("RemoteWS", () => { const sockets: FakeWebSocket[] = [] const received: unknown[] = [] - class FakeWebSocket { - static readonly OPEN = 1 - readonly sent: string[] = [] - readyState = 0 - onopen: (() => void) | null = null - onmessage: ((event: { data: string }) => void) | null = null - onclose: ((event: { code: number; reason: string }) => void) | null = null - onerror: ((event: unknown) => void) | null = null - - constructor(readonly url: string) { - sockets.push(this) - } - - send(message: string) { - this.sent.push(message) - } - - close() { - this.readyState = 3 - } - - open() { - this.readyState = FakeWebSocket.OPEN - this.onopen?.() - } - - disconnect(code = 1000, reason = "closed") { - this.readyState = 3 - this.onclose?.({ code, reason }) - } - } - + FakeWebSocket.reset() Object.defineProperty(globalThis, "WebSocket", { value: FakeWebSocket, configurable: true, writable: true }) try { conn = RemoteWS.connect({ @@ -297,13 +381,13 @@ describe("RemoteWS", () => { }) await settled() - const first = sockets[0] + const first = FakeWebSocket.instances[0] expect(first).toBeDefined() first?.open() first?.disconnect() - await until(() => sockets.length >= 2) - const second = sockets[1] + await until(() => FakeWebSocket.instances.length >= 2) + const second = FakeWebSocket.instances[1] expect(second).toBeDefined() second?.open() @@ -517,4 +601,1308 @@ describe("RemoteWS", () => { await settled() expect(conn.connected).toBe(true) }) + + // ------------------------------------------------------------------------- + // Deterministic fake-clock tests (AC2: bounded token acquisition) + // ------------------------------------------------------------------------- + + async function withFakeWebSocket(fn: (clock: FakeClock) => T): Promise { + const OriginalWebSocket = globalThis.WebSocket + FakeWebSocket.reset() + Object.defineProperty(globalThis, "WebSocket", { value: FakeWebSocket, configurable: true, writable: true }) + try { + const clock = new FakeClock() + return await fn(clock) + } finally { + Object.defineProperty(globalThis, "WebSocket", { value: OriginalWebSocket, configurable: true, writable: true }) + } + } + + test("AC2a: getToken() rejection schedules a bounded retry and later succeeds", async () => { + await withFakeWebSocket(async (clock) => { + let attempt = 0 + const getToken = async () => { + attempt++ + if (attempt === 1) throw new Error("no token") + return "tok" + } + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken, + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + tokenTimeout: 15_000, + }) + + // The rejected getToken settles before the token deadline. + await flush() + expect(attempt).toBe(1) + expect(FakeWebSocket.instances.length).toBe(0) + + // Retry fires at the initial backoff (1000ms), well before tokenTimeout. + clock.advance(1000) + await flush() + expect(attempt).toBe(2) + expect(FakeWebSocket.instances.length).toBe(1) + + const socket = FakeWebSocket.instances[0] + socket.open() + expect(conn.connected).toBe(true) + }) + }) + + test("AC2b: getToken() that never settles triggers a bounded retry and later succeeds", async () => { + await withFakeWebSocket(async (clock) => { + let attempt = 0 + const getToken = async () => { + attempt++ + if (attempt === 1) return new Promise(() => {}) // never resolves + return "tok" + } + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken, + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + tokenTimeout: 1000, + }) + + // First token attempt times out. + clock.advance(1000) + await flush() + expect(attempt).toBe(1) + expect(FakeWebSocket.instances.length).toBe(0) + + // Retry fires after backoff. + clock.advance(1000) + await flush() + expect(attempt).toBe(2) + expect(FakeWebSocket.instances.length).toBe(1) + + const socket = FakeWebSocket.instances[0] + socket.open() + expect(conn.connected).toBe(true) + }) + }) + + test("AC2c: getToken() resolving undefined schedules a bounded retry and later succeeds", async () => { + await withFakeWebSocket(async (clock) => { + let attempt = 0 + const getToken = async () => { + attempt++ + if (attempt === 1) return undefined + return "tok" + } + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken, + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + tokenTimeout: 1000, + }) + + // First token attempt resolves to undefined before the deadline. + await flush() + expect(attempt).toBe(1) + expect(FakeWebSocket.instances.length).toBe(0) + + // The undefined result schedules a retry at the initial backoff. + clock.advance(1000) + await flush() + expect(attempt).toBe(2) + expect(FakeWebSocket.instances.length).toBe(1) + + const socket = FakeWebSocket.instances[0] + socket.open() + expect(conn.connected).toBe(true) + }) + }) + + // ------------------------------------------------------------------------- + // AC3: connection-attempt deadline with a single fenced retry owner + // ------------------------------------------------------------------------- + + test("AC3a: a socket stuck in CONNECTING is replaced by exactly one new attempt", async () => { + await withFakeWebSocket(async (clock) => { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + }) + + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + + // Connect deadline fires, scheduling a retry. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(1) // old socket not yet replaced + + // Retry fires after backoff; exactly one new socket is created. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + + // No further attempts appear. + clock.advance(60_000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + }) + }) + + test("AC3b: synchronous WebSocket constructor throw schedules exactly one retry", async () => { + await withFakeWebSocket(async (clock) => { + let attempts = 0 + class ThrowingWebSocket { + static readonly OPEN = 1 + constructor() { + attempts++ + throw new Error("constructor failed") + } + } + const OriginalWebSocket = globalThis.WebSocket + Object.defineProperty(globalThis, "WebSocket", { value: ThrowingWebSocket, configurable: true, writable: true }) + + try { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + }) + + await flush() + expect(attempts).toBe(1) + + // Retry fires after backoff. The connect deadline (also at 1000ms) observes + // the generation is already settled and does not schedule a second retry. + clock.advance(1000) + await flush() + expect(attempts).toBe(2) + } finally { + Object.defineProperty(globalThis, "WebSocket", { value: OriginalWebSocket, configurable: true, writable: true }) + } + }) + }) + + test("AC3c: connect deadline only schedules exactly one retry when onclose arrives late", async () => { + await withFakeWebSocket(async (clock) => { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + }) + + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + + // Connect deadline fires for the first generation, closing the socket and scheduling a retry. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + + // Retry fires after backoff; exactly one new socket is created. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + + // Open the replacement socket and confirm the connection is live. + const second = FakeWebSocket.instances[1] + second.open() + expect(conn.connected).toBe(true) + + // No further sockets are created. + clock.advance(60_000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + expect(conn.connected).toBe(true) + }) + }) + + test("AC3d: connect deadline is cleared after a successful open", async () => { + await withFakeWebSocket(async (clock) => { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + expect(conn.connected).toBe(true) + + // Advance well past connectTimeout; no deadline-driven reconnect should occur. + clock.advance(60_000) + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + expect(conn.connected).toBe(true) + }) + }) + + test("AC3d: connect deadline is cleared after onclose", async () => { + await withFakeWebSocket(async (clock) => { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + expect(conn.connected).toBe(true) + + socket.disconnect() + await flush() + expect(conn.connected).toBe(false) + + // Reconnect happens at the backoff time (1000ms), not at the connectTimeout. + clock.advance(999) + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + + clock.advance(1) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + + // No deadline-driven reconnect after the reconnect. + clock.advance(60_000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + }) + }) + + test("AC3d/e: connect deadline is cleared and no reconnect after Connection.close()", async () => { + await withFakeWebSocket(async (clock) => { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + expect(conn.connected).toBe(true) + + conn.close() + expect(conn.connected).toBe(false) + + // Advance past connectTimeout and backoff; no new attempts. + clock.advance(60_000) + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + }) + }) + + test("AC3f: connect-attempt deadline close does not invoke onDisconnect; post-open transient close does", async () => { + await withFakeWebSocket(async (clock) => { + let disconnects = 0 + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + onDisconnect: () => disconnects++, + }) + + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + + // Connect deadline fires, closing the socket before it opened. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + expect(disconnects).toBe(0) + + // Retry fires after backoff; a new socket is created. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + + // Open the replacement socket and then close it transiently. + const socket = FakeWebSocket.instances[1] + socket.open() + expect(conn.connected).toBe(true) + + socket.disconnect(1000, "transient") + await flush() + expect(disconnects).toBe(1) + expect(conn.connected).toBe(false) + }) + }) + + test("AC3g: a token that resolves after the connect deadline fired does not assign a stale socket", async () => { + await withFakeWebSocket(async (clock) => { + let attempt = 0 + let lateResolve!: (v: string) => void + const getToken = () => { + attempt++ + if (attempt === 1) { + return new Promise((r) => { + lateResolve = r + }) + } + return Promise.resolve("tok") + } + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken, + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + connectTimeout: 1000, + tokenTimeout: 15000, + }) + + // Gen1: getToken() is still pending; no socket has been constructed yet. + await flush() + expect(attempt).toBe(1) + expect(FakeWebSocket.instances.length).toBe(0) + + // The connect deadline fires while getToken() is pending, settling gen1 + // and scheduling a retry. ws is still undefined at this point, so the + // deadline's ws.close() is a no-op. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(0) + + // Gen2 fires after the initial backoff (1000ms). Its getToken() + // resolves immediately and a socket is constructed. + clock.advance(1000) + await flush() + expect(attempt).toBe(2) + expect(FakeWebSocket.instances.length).toBe(1) + const gen2Socket = FakeWebSocket.instances[0] + gen2Socket.open() + expect(conn.connected).toBe(true) + + // Now gen1's late token resolves. The guarded continuation must + // observe g.settled and return without constructing a stale socket + // and without clobbering the live ws pointer. + lateResolve("tok-late") + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + expect(conn.connected).toBe(true) + expect(FakeWebSocket.instances[0]).toBe(gen2Socket) + expect(gen2Socket.readyState).toBe(FakeWebSocket.OPEN) + }) + }) + + // ------------------------------------------------------------------------- + // AC5: regression guard for permanent close codes and backoff reset + // + // Existing coverage: + // - Initial backoff retry: "reconnects with backoff after server close". + // - 4401 permanent stop: "stops reconnecting on 4401". + // - Stale-generation fencing: "ignores callbacks from a stale WebSocket generation". + // - close() no-reconnect: "close() prevents further reconnection and stops heartbeat". + // - Activity timeout: "force-reconnects on activity timeout" / "resets activity timer...". + // Missing and added below: + // - 4403 and 4409 permanent close (no reconnect, onClose fired). + // - Pending heartbeat() waiters reject on permanent close. + // - Backoff resets to the initial value after a successful onopen. + // ------------------------------------------------------------------------- + + test("AC5: 4403 and 4409 are permanent close codes with no reconnect", async () => { + await withFakeWebSocket(async (clock) => { + for (const code of [4403, 4409]) { + conn?.close() + const codes: number[] = [] + FakeWebSocket.reset() + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + onClose: (c) => codes.push(c), + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + socket.disconnect(code, "permanent") + + await flush() + expect(codes).toEqual([code]) + expect(conn.connected).toBe(false) + + // Advance far past any backoff; no reconnect. + clock.advance(120_000) + await flush() + expect(FakeWebSocket.instances.length).toBe(1) + } + }) + }) + + test("AC5b: pending heartbeat() rejects on permanent close codes 4403 and 4409", async () => { + await withFakeWebSocket(async (clock) => { + for (const code of [4403, 4409]) { + conn?.close() + const closed: Array<{ code: number; reason: string }> = [] + FakeWebSocket.reset() + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: () => new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + onClose: (c, r) => closed.push({ code: c, reason: r }), + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Wedge the gather so heartbeat() stays pending. + const promise = conn.heartbeat() + let resolved = false + let rejected = false + let rejectionError: unknown + void promise.then( + () => { + resolved = true + }, + (err) => { + rejected = true + rejectionError = err + }, + ) + clock.advance(1000) + await flushLong() + expect(socket.sent.length).toBe(1) + expect(resolved).toBe(false) + expect(rejected).toBe(false) + + // Permanent close: pending waiter must reject, and onClose fires. + socket.disconnect(code, "permanent") + await flush() + expect(resolved).toBe(false) + expect(rejected).toBe(true) + expect(String(rejectionError)).toContain("remote-ws connection permanently closed") + expect(closed).toEqual([{ code, reason: "permanent" }]) + expect(conn.connected).toBe(false) + } + }) + }) + + test("AC5: backoff resets to the initial value after a successful open", async () => { + await withFakeWebSocket(async (clock) => { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions: async () => ({ sessions: [] }), + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + }) + + await flush() + const first = FakeWebSocket.instances[0] + first.open() + + // First transient close. schedule() uses 1000ms, then doubles to 2000ms. + first.disconnect(1000, "first") + await flush() + + // Reconnect at 1000ms. + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + const second = FakeWebSocket.instances[1] + second.open() + + // Second transient close. Because onopen reset backoff to 1000ms, the next + // reconnect should be at 1000ms, not 2000ms. + second.disconnect(1001, "second") + await flush() + + clock.advance(999) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + + clock.advance(1) + await flush() + expect(FakeWebSocket.instances.length).toBe(3) + }) + }) + + // ------------------------------------------------------------------------- + // AC4: bounded heartbeat gather with freshness-fenced attach (Path D fix) + // ------------------------------------------------------------------------- + + test("AC4a: never-settling gather sends a degraded heartbeat and keeps the connection live", async () => { + await withFakeWebSocket(async (clock) => { + const getSessions = () => new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}) + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + expect(conn.connected).toBe(true) + + const promise = conn.heartbeat() + clock.advance(1000) + await flushLong() + + // Degraded heartbeat was sent over the live socket (cold start → empty) + expect(socket.sent.length).toBe(1) + const parsed = JSON.parse(socket.sent[0]) + expect(parsed.type).toBe("heartbeat") + expect(parsed.sessions).toEqual([]) + + // Connection still live + expect(conn.connected).toBe(true) + + // Waiter stays pending (degraded sends do not resolve attach) + let resolved = false + void promise.then( + () => { + resolved = true + }, + () => {}, + ) + await flushLong() + expect(resolved).toBe(false) + }) + }) + + test("AC4a: degraded heartbeat preserves the last known-good non-empty session list", async () => { + await withFakeWebSocket(async (clock) => { + let mode: "fresh" | "wedge" = "fresh" + const knownGoodSessions = [ + { id: "s1", status: "active" as const, title: "One" }, + ] as RemoteWS.SessionInfo[] + const getSessions = () => + mode === "fresh" + ? Promise.resolve({ sessions: knownGoodSessions }) + : new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}) + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: fresh gather establishes a non-empty last-known-good list. + fireHeartbeat() + await flushLong() + expect(socket.sent.length).toBe(1) + const payload1 = JSON.parse(socket.sent[0]) + expect(payload1.type).toBe("heartbeat") + expect(payload1.sessions).toEqual(knownGoodSessions) + + // Cycle 2: wedged gather times out → degraded send carries the same list. + mode = "wedge" + fireHeartbeat() + clock.advance(1000) + await flushLong() + expect(socket.sent.length).toBe(2) + const payload2 = JSON.parse(socket.sent[1]) + expect(payload2.type).toBe("heartbeat") + expect(payload2.sessions).toEqual(knownGoodSessions) + + // Connection still live + expect(conn.connected).toBe(true) + }) + }) + + test("AC4b: a gather that settles after its deadline is discarded", async () => { + await withFakeWebSocket(async (clock) => { + let lateResolve!: (v: { sessions: RemoteWS.SessionInfo[] }) => void + const getSessions = () => + new Promise<{ sessions: RemoteWS.SessionInfo[] }>((r) => { + lateResolve = r + }) + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: wedge, then time out → degraded send (cold start → empty) + fireHeartbeat() + clock.advance(1000) + await flushLong() + expect(socket.sent.length).toBe(1) + expect(JSON.parse(socket.sent[0]).sessions).toEqual([]) + + // Late settle: the original getSessions promise eventually resolves. + // This must NOT cause any additional heartbeat to be sent — its result + // is discarded because the cycle already abandoned it on timeout. + lateResolve({ + sessions: [{ id: "late", status: "active", title: "Late" }] as RemoteWS.SessionInfo[], + }) + await flushLong() + + // No further heartbeat sent; the last payload is still the degraded/last-good list + expect(socket.sent.length).toBe(1) + expect(JSON.parse(socket.sent[0]).sessions).toEqual([]) + }) + }) + + test("AC4c: after a timed-out cycle, a later fresh-gather cycle sends fresh sessions", async () => { + await withFakeWebSocket(async (clock) => { + let mode: "wedge" | "fresh" = "wedge" + const freshSessions = [ + { id: "fresh", status: "active" as const, title: "Fresh" }, + ] as RemoteWS.SessionInfo[] + const getSessions = () => + mode === "wedge" + ? new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}) + : Promise.resolve({ sessions: freshSessions }) + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: wedge → degraded send + fireHeartbeat() + clock.advance(1000) + await flushLong() + expect(socket.sent.length).toBe(1) + expect(JSON.parse(socket.sent[0]).sessions).toEqual([]) + + // Cycle 2: switch to fresh, kick another cycle + mode = "fresh" + const promise2 = conn.heartbeat() + void promise2.catch(() => {}) + await flushLong() + expect(socket.sent.length).toBe(2) + const payload2 = JSON.parse(socket.sent[1]) + expect(payload2.type).toBe("heartbeat") + expect(payload2.sessions).toEqual(freshSessions) + // The fresh cycle sent over a live socket → promise resolved + await promise2 + }) + }) + + test("AC4d: maxOutstandingGathers cap blocks calls while wedged; settling one allows recovery", async () => { + await withFakeWebSocket(async (clock) => { + let calls = 0 + let mode: "wedge" | "fresh" = "wedge" + const wedgeResolvers: Array<(v: { sessions: RemoteWS.SessionInfo[] }) => void> = [] + const freshSessions = [ + { id: "fresh", status: "active" as const, title: "Fresh" }, + ] as RemoteWS.SessionInfo[] + const getSessions = () => { + calls++ + if (mode === "wedge") { + return new Promise<{ sessions: RemoteWS.SessionInfo[] }>((r) => { + wedgeResolvers.push(r) + }) + } + return Promise.resolve({ sessions: freshSessions }) + } + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + maxOutstandingGathers: 2, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: wedge → degraded. The wedge never settles, so its slot stays held. + fireHeartbeat() + clock.advance(1000) + await flushLong() + expect(calls).toBe(1) + expect(JSON.parse(socket.sent[0]).sessions).toEqual([]) + + // Cycle 2: wedge → degraded. Second wedge held, outstanding now at cap (2). + fireHeartbeat() + clock.advance(1000) + await flushLong() + expect(calls).toBe(2) + expect(JSON.parse(socket.sent[1]).sessions).toEqual([]) + + // Cycle 3: cap reached. getSessions MUST NOT be called again, but a + // degraded heartbeat must still be sent (liveness is preserved). + fireHeartbeat() + clock.advance(1000) + await flushLong() + expect(calls).toBe(2) + expect(socket.sent.length).toBe(3) + expect(JSON.parse(socket.sent[2]).sessions).toEqual([]) + + // Settle the first wedge → its slot is released, outstanding drops to 1. + wedgeResolvers.shift()!({ sessions: freshSessions }) + // Let the .then release handler run (microtask). + await flushLong() + // Switch to fresh so the next gather resolves promptly. + mode = "fresh" + // Cycle 4: outstanding=1 < cap=2, getSessions IS called, resolves fresh. + fireHeartbeat() + await flushLong() + expect(calls).toBe(3) + expect(socket.sent.length).toBe(4) + expect(JSON.parse(socket.sent[3]).sessions).toEqual(freshSessions) + }) + }) + + test("AC4f: promptly-rejecting getSessions → degraded, no slot consumed, recovers on resolve", async () => { + await withFakeWebSocket(async (clock) => { + let calls = 0 + let mode: "reject" | "fresh" = "reject" + const freshSessions = [ + { id: "fresh", status: "active" as const, title: "Fresh" }, + ] as RemoteWS.SessionInfo[] + const getSessions = () => { + calls++ + return mode === "reject" + ? Promise.reject(new Error("gather failed")) + : Promise.resolve({ sessions: freshSessions }) + } + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: reject → degraded (cold start → empty) + fireHeartbeat() + await flushLong() + expect(calls).toBe(1) + expect(JSON.parse(socket.sent[0]).sessions).toEqual([]) + + // Cycle 2: reject again — no slot was held, so getSessions is called again immediately + fireHeartbeat() + await flushLong() + expect(calls).toBe(2) + expect(socket.sent.length).toBe(2) + expect(JSON.parse(socket.sent[1]).sessions).toEqual([]) + + // Switch to fresh → next cycle is fresh + mode = "fresh" + fireHeartbeat() + await flushLong() + expect(calls).toBe(3) + expect(socket.sent.length).toBe(3) + expect(JSON.parse(socket.sent[2]).sessions).toEqual(freshSessions) + }) + }) + + test("AC4g: synchronously-throwing getSessions releases slot and recovers", async () => { + await withFakeWebSocket(async (clock) => { + let calls = 0 + let mode: "throw" | "fresh" = "throw" + const freshSessions = [ + { id: "fresh", status: "active" as const, title: "Fresh" }, + ] as RemoteWS.SessionInfo[] + const getSessions = () => { + calls++ + if (mode === "throw") { + throw new Error("gather sync throw") + } + return Promise.resolve({ sessions: freshSessions }) + } + + const unhandled: unknown[] = [] + const onUnhandled = (reason: unknown) => unhandled.push(reason) + process.on("unhandledRejection", onUnhandled) + + try { + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + maxOutstandingGathers: 1, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: synchronous throw → degraded (cold start → empty) + fireHeartbeat() + await flushLong() + expect(calls).toBe(1) + expect(socket.sent.length).toBe(1) + expect(JSON.parse(socket.sent[0]).sessions).toEqual([]) + expect(conn.connected).toBe(true) + + // Cycle 2: with maxOutstandingGathers=1, a leaked slot would hit the cap + // and skip getSessions. The call proves the slot was released. + fireHeartbeat() + await flushLong() + expect(calls).toBe(2) + expect(socket.sent.length).toBe(2) + expect(JSON.parse(socket.sent[1]).sessions).toEqual([]) + + // Switch to fresh → recovers to fresh payloads + mode = "fresh" + fireHeartbeat() + await flushLong() + expect(calls).toBe(3) + expect(socket.sent.length).toBe(3) + expect(JSON.parse(socket.sent[2]).sessions).toEqual(freshSessions) + + expect(unhandled).toEqual([]) + } finally { + process.off("unhandledRejection", onUnhandled) + } + }) + }) + + // ------------------------------------------------------------------------- + // AC6: freshness-fenced attach — heartbeat() resolves only on a fresh send + // over a live socket; survives transient reconnect; rejects on close() + // ------------------------------------------------------------------------- + + test("AC6a: heartbeat() does not resolve on degraded, resolves on the next fresh send", async () => { + await withFakeWebSocket(async (clock) => { + let mode: "wedge" | "fresh" = "wedge" + const freshSessions = [ + { id: "fresh", status: "active" as const, title: "Fresh" }, + ] as RemoteWS.SessionInfo[] + const getSessions = () => + mode === "wedge" + ? new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}) + : Promise.resolve({ sessions: freshSessions }) + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Degraded cycle: promise stays pending + const promise1 = conn.heartbeat() + let settled1 = false + void promise1.then( + () => { + settled1 = true + }, + () => {}, + ) + clock.advance(1000) + await flushLong() + expect(socket.sent.length).toBe(1) + expect(settled1).toBe(false) + + // Fresh cycle: both the deferred promise1 and the new promise2 resolve + mode = "fresh" + const promise2 = conn.heartbeat() + let settled2 = false + void promise2.then( + () => { + settled2 = true + }, + () => {}, + ) + await flushLong() + expect(socket.sent.length).toBe(2) + expect(JSON.parse(socket.sent[1]).sessions).toEqual(freshSessions) + expect(settled1).toBe(true) + expect(settled2).toBe(true) + }) + }) + + test("AC6b: pending heartbeat() survives disconnect+reconnect and resolves on fresh send over the new socket", async () => { + await withFakeWebSocket(async (clock) => { + let mode: "wedge" | "fresh" = "wedge" + const freshSessions = [ + { id: "fresh", status: "active" as const, title: "Fresh" }, + ] as RemoteWS.SessionInfo[] + const getSessions = () => + mode === "wedge" + ? new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}) + : Promise.resolve({ sessions: freshSessions }) + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket1 = FakeWebSocket.instances[0] + socket1.open() + + // Cycle 1 on socket1: wedge → degraded send, waiter pending + const promise = conn.heartbeat() + let settled = false + void promise.then( + () => { + settled = true + }, + () => {}, + ) + clock.advance(1000) + await flushLong() + expect(socket1.sent.length).toBe(1) + expect(settled).toBe(false) + + // Transient disconnect + socket1.disconnect(1000, "transient") + await flush() + expect(conn.connected).toBe(false) + + // Switch to fresh before reconnect so the next gather is fresh + mode = "fresh" + + // Reconnect after backoff + clock.advance(1000) + await flush() + expect(FakeWebSocket.instances.length).toBe(2) + const socket2 = FakeWebSocket.instances[1] + socket2.open() + await flushLong() + expect(conn.connected).toBe(true) + + // The onopen handler kicks a cycle because waiters > 0; that fresh + // cycle sends over socket2 and resolves the deferred waiter. + expect(socket2.sent.length).toBe(1) + const payload = JSON.parse(socket2.sent[0]) + expect(payload.type).toBe("heartbeat") + expect(payload.sessions).toEqual(freshSessions) + expect(settled).toBe(true) + }) + }) + + test("AC6c: pending heartbeat() rejects (does not hang) when close() is called", async () => { + await withFakeWebSocket(async (clock) => { + const getSessions = () => + new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}) // wedge + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: wedge → degraded send, waiter pending + const promise = conn.heartbeat() + let resolved = false + let rejected = false + let rejectionError: unknown + void promise.then( + () => { + resolved = true + }, + (err) => { + rejected = true + rejectionError = err + }, + ) + clock.advance(1000) + await flushLong() + expect(socket.sent.length).toBe(1) + expect(resolved).toBe(false) + expect(rejected).toBe(false) + + // Close: pending waiter must reject (not hang) + conn.close() + conn = undefined + await flush() + expect(resolved).toBe(false) + expect(rejected).toBe(true) + expect(String(rejectionError)).toContain("remote-ws connection closed") + }) + }) + + // AC6d: id-containment fence. A fresh heartbeat that LEGITIMATELY OMITS + // the announced id (e.g. an upstream `get(id)` was filtered by + // `Effect.orElseSucceed`) must NOT resolve an announce waiter that + // required that id. The waiter stays pending and is re-evaluated on + // the next fresh cycle whose payload actually contains the id. + test("AC6d: heartbeat({ requireSessionId }) stays pending on a fresh send that omits the id and resolves on the next fresh send containing it", async () => { + await withFakeWebSocket(async (clock) => { + let mode: "without" | "with" = "without" + const otherSession = { id: "other", status: "active" as const, title: "Other" } + const targetSession = { id: "target", status: "active" as const, title: "Target" } + const listWithout = [otherSession] as RemoteWS.SessionInfo[] + const listWith = [otherSession, targetSession] as RemoteWS.SessionInfo[] + const getSessions = () => + Promise.resolve({ sessions: mode === "without" ? listWithout : listWith }) + + conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle 1: fresh gather EXCLUDES "target". A heartbeat with + // requireSessionId="target" must stay pending even though a fresh + // heartbeat was sent over the live socket. + const idPromise = conn.heartbeat({ requireSessionId: "target" }) + let idResolved = false + let idRejected = false + void idPromise.then( + () => { + idResolved = true + }, + () => { + idRejected = true + }, + ) + await flushLong() + expect(socket.sent.length).toBe(1) + const firstPayload = JSON.parse(socket.sent[0]) + expect(firstPayload.type).toBe("heartbeat") + expect(firstPayload.sessions.map((s: { id: string }) => s.id)).toEqual(["other"]) + expect(idResolved).toBe(false) + expect(idRejected).toBe(false) + + // Cycle 2: switch the gather to INCLUDE "target" and drive another + // cycle. The id-gated waiter now resolves. + mode = "with" + // A no-id heartbeat() call drives the next cycle. It is allowed to + // resolve on any fresh send (it does not require a specific id) — + // the assertion below checks that this no-id call resolves, which + // guards against regressing AC6a. + const noIdPromise = conn.heartbeat() + let noIdResolved = false + void noIdPromise.then( + () => { + noIdResolved = true + }, + () => {}, + ) + await flushLong() + expect(socket.sent.length).toBe(2) + const secondPayload = JSON.parse(socket.sent[1]) + expect(secondPayload.type).toBe("heartbeat") + expect(secondPayload.sessions.map((s: { id: string }) => s.id).sort()).toEqual(["other", "target"]) + // The id-gated waiter resolved on this fresh send that includes the id. + expect(idResolved).toBe(true) + expect(idRejected).toBe(false) + // The no-id waiter also resolved (it is satisfied by any fresh send). + expect(noIdResolved).toBe(true) + }) + }) + + test("AC6e: pending heartbeat({ requireSessionId }) rejects when permanent close arrives during an in-flight gather cycle", async () => { + await withFakeWebSocket(async (clock) => { + const getSessions = () => new Promise<{ sessions: RemoteWS.SessionInfo[] }>(() => {}) + + let c: RemoteWS.Connection | undefined + c = conn = RemoteWS.connect({ + url: "ws://example.test", + getToken: async () => "tok", + getSessions, + log: nolog(), + heartbeat: 60_000, + timers: clock, + now: () => clock.now, + timeout: 300_000, + gatherTimeout: 1000, + onClose: () => { + c?.close() + }, + }) + + await flush() + const socket = FakeWebSocket.instances[0] + socket.open() + + // Cycle is in gatherOnce; the waiter has been moved into the in-flight + // cycleWaiters, so waiters is momentarily empty. + const promise = c.heartbeat({ requireSessionId: "target" }) + let resolved = false + let rejected = false + let rejectionError: unknown + void promise.then( + () => { + resolved = true + }, + (err) => { + rejected = true + rejectionError = err + }, + ) + await flush() + expect(resolved).toBe(false) + expect(rejected).toBe(false) + + // Permanent close mirrors production: onClose calls close(), which sets + // the terminal closed flag. The bounded gather then times out, and the + // in-flight cycleWaiters must be rejected (not orphaned). + socket.disconnect(4403, "permanent") + clock.advance(1000) + await flushLong() + expect(resolved).toBe(false) + expect(rejected).toBe(true) + expect(String(rejectionError)).toContain("remote-ws connection closed") + }) + }) })