From dd351a7846012609e3ca5b72a747677dc5d1231b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alejandro=20Guti=C3=A9rrez?= <35082514+alezmad@users.noreply.github.com> Date: Tue, 8 Sep 2026 22:22:58 +0100 Subject: [PATCH] fix(cli): daemon restart no longer blacks out the mesh (1.37.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Daemon side of the 2026-09-08 incident (spec Addenda 3, 5, 6): - ws-lifecycle: createWsLifecycle() returns the handle synchronously with ready exposed separately. A first hello-ack timeout used to REJECT the connect promise while the helper kept reconnecting on its own, so DaemonBrokerClient.lifecycle stayed null forever and every RPC gated on it (listPeers, send, ...) short-circuited: peer list answered [] on all 9 meshes for as long as the daemon lived. Ack timeout 5 s → 15 s (a timeout is now just a reconnect). 1000 session_replaced is terminal for a lifecycle (two daemons no longer fight over the same sessions). - drain: 24 h outbox TTL (expired), session_gone after 10 min, attempt cap on the not-ready path, drain_session_ws_not_ready logged on attempt 1 and every 60th (was every tick: 4 rows × ~130 k attempts = a 1 GB log). - Rotating log sink (20 MB × 5, total budget) behind an explicit --log-sink flag that the launchd/systemd unit and the detached spawn pass; units point StandardOut/Err at daemon.launchd.log and pin ThrottleInterval 10. - Exit forensics: daemon_exit / daemon_uncaught on every exit path; SIGHUP is a clean shutdown. - Singleton hardening: the CLI stale-probe no longer unlinks a live daemon's pid/socket on a slow probe (only ECONNREFUSED/ENOENT or a dead pid is stale); the daemon probes the socket for another live daemon before taking the lock and re-asserts its pid file every 30 s. - MCP SSE: resubscribe on every termination event with a 45 s idle watchdog; the daemon ends open SSE streams before exiting. - Operator signal: GET /v1/diagnostics; daemon status shows per-mesh and per-session WS state; new daemon restart; doctor adds version match, member/session WS open, ghost detector (every registered session must appear in the broker peer list), log size + stale unit; peer list warns when the daemon is not connected to the broker for a mesh. Tests: ws-lifecycle (timeout → same handle, close mid-connect, session_replaced terminal), daemon-broker-lost-handle (Addendum 3 regression), drain-expiry, log-sink, sse-resubscribe (socket destroyed without end), daemon-probe-classify. Local drills against the fixed broker: SIGKILL + respawn → sessions visible at t+1 s, second daemon refused, pid file self-healed. Co-Authored-By: Claude Fable 5.1 --- apps/cli/CHANGELOG.md | 53 +++ apps/cli/package.json | 2 +- apps/cli/src/commands/daemon.ts | 129 ++++++- apps/cli/src/commands/doctor.ts | 224 ++++++++++++ apps/cli/src/commands/peers.ts | 40 ++- apps/cli/src/daemon/broker.ts | 35 +- apps/cli/src/daemon/drain.ts | 73 +++- apps/cli/src/daemon/ipc/server.ts | 110 +++++- apps/cli/src/daemon/log-sink.ts | 226 ++++++++++++ apps/cli/src/daemon/paths.ts | 9 + apps/cli/src/daemon/run.ts | 140 +++++++- apps/cli/src/daemon/service-install.ts | 60 +++- apps/cli/src/daemon/session-broker.ts | 30 +- apps/cli/src/daemon/ws-lifecycle.ts | 335 +++++++++++------- apps/cli/src/entrypoints/cli.ts | 4 + apps/cli/src/mcp/server.ts | 146 ++++++-- apps/cli/src/services/bridge/daemon-route.ts | 40 ++- apps/cli/src/services/daemon/lifecycle.ts | 34 +- .../unit/daemon-broker-lost-handle.test.ts | 110 ++++++ .../tests/unit/daemon-probe-classify.test.ts | 30 ++ apps/cli/tests/unit/drain-expiry.test.ts | 139 ++++++++ apps/cli/tests/unit/log-sink.test.ts | 142 ++++++++ apps/cli/tests/unit/sse-resubscribe.test.ts | 149 ++++++++ apps/cli/tests/unit/ws-lifecycle.test.ts | 189 ++++++++++ 24 files changed, 2236 insertions(+), 213 deletions(-) create mode 100644 apps/cli/src/daemon/log-sink.ts create mode 100644 apps/cli/tests/unit/daemon-broker-lost-handle.test.ts create mode 100644 apps/cli/tests/unit/daemon-probe-classify.test.ts create mode 100644 apps/cli/tests/unit/drain-expiry.test.ts create mode 100644 apps/cli/tests/unit/log-sink.test.ts create mode 100644 apps/cli/tests/unit/sse-resubscribe.test.ts create mode 100644 apps/cli/tests/unit/ws-lifecycle.test.ts diff --git a/apps/cli/CHANGELOG.md b/apps/cli/CHANGELOG.md index b3199f6..e98257e 100644 --- a/apps/cli/CHANGELOG.md +++ b/apps/cli/CHANGELOG.md @@ -1,5 +1,58 @@ # Changelog +## 1.37.1 (2026-09-08) — daemon restart no longer blacks out the mesh + +Incident: after the local daemon restarted (launchd respawn), `peer list` +returned 0 peers on every mesh for >40 min while all sessions were alive, +sends reported `no connected peer`, and the MCP push channel went silent. +Spec + evidence: `.artifacts/specs/2026-09-08-daemon-restart-mesh-blackout.md`. + +### Root causes (daemon side; the broker side ships separately) + +- `DaemonBrokerClient.connect()` awaited `connectWsWithBackoff()`, which + REJECTED when the first hello ack timed out (5 s; the broker took 4–5 s + under the reconnect storm) while its internal loop kept reconnecting the + same socket. `lifecycle` stayed `null` forever, `_status` flipped to + `open`, and every RPC gated on `this.lifecycle` (`listPeers`, `send`, + …) short-circuited — `/v1/peers` answered `[]` in 30 ms on all 9 meshes. +- The outbox drain retried rows bound to long-gone session pubkeys ~130 k + times each and logged `drain_session_ws_not_ready` on every tick; + `daemon.log` reached 1.0 GB with no rotation (launchd `StandardOutPath`). +- The daemon that died left no log line (no crash/exit forensics). +- The MCP plugin's `/v1/events` SSE subscription only resubscribed on + `res 'end'`; a killed daemon produces `close`/`aborted` instead, so + mid-turn `` injection stayed dead until Claude Code restarted. + +### Fixes + +- `ws-lifecycle.ts`: `createWsLifecycle()` returns the handle synchronously + with `ready` exposed separately; a hello-ack timeout is just a backoff + reconnect (never rejects, never orphans); `close()` works mid-connect. + Ack timeout 5 s → 15 s. Both broker clients store the handle before + awaiting `ready` and expose `diagnostics()`. +- `drain.ts`: 24 h outbox TTL (`expired`), `session_gone` after 10 min for + rows whose session is no longer registered, attempt cap on the + not-ready path, `drain_session_ws_not_ready` logged on attempt 1 and + every 60th. +- Rotating log sink (20 MB × 5, total budget enforced) takes over + stdout/stderr when the daemon runs under launchd/systemd; the service + unit now points StandardOut/Err at `daemon.launchd.log` and pins + `ThrottleInterval 10`. Exit forensics: `daemon_exit` / + `daemon_uncaught` lines on every exit path; SIGHUP is a clean shutdown. +- MCP SSE: resubscribe on every termination event, 45 s idle watchdog + (daemon keepalive is 15 s), reconnects logged to `mcp-.log`; the + daemon ends open SSE streams before exiting. +- Operator signal: `GET /v1/diagnostics`; `claudemesh daemon status` + shows per-mesh and per-session WS state; new `claudemesh daemon + restart`; `claudemesh doctor` adds version match, member/session WS + open, a ghost detector (every registered session must appear in the + broker peer list), log size + stale service unit. `peer list` warns + when the daemon is not connected to the broker for a mesh instead of + printing a silent empty list. + +After upgrading run `claudemesh daemon install-service` (new unit) and +`claudemesh daemon restart`. + ## 1.34.17 (2026-05-20) — host fingerprint v2: stable across Mac reboots Fixes a long-standing class of false-positive `host_fingerprint diff --git a/apps/cli/package.json b/apps/cli/package.json index f893c30..fc75043 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -1,6 +1,6 @@ { "name": "claudemesh-cli", - "version": "1.37.0", + "version": "1.37.1", "description": "Peer mesh for Claude Code sessions — CLI + MCP server.", "keywords": [ "claude-code", diff --git a/apps/cli/src/commands/daemon.ts b/apps/cli/src/commands/daemon.ts index b14f165..f265d22 100644 --- a/apps/cli/src/commands/daemon.ts +++ b/apps/cli/src/commands/daemon.ts @@ -22,6 +22,10 @@ export interface DaemonOptions { * auto-spawn path always pass --foreground because their parents * (launchd / the launch helper) own the lifecycle. */ foreground?: boolean; + /** 1.37.1: route stdout/stderr into the rotating daemon.log (20 MB × 5). + * Set by the launchd/systemd unit and by the detached spawn; a human + * running `up --foreground` in a terminal leaves it off. */ + logSink?: boolean; /** outbox-list status filter, set from boolean flags --failed/--pending/etc. */ outboxStatus?: "pending" | "inflight" | "done" | "dead" | "aborted"; /** outbox requeue: optional id to mint a fresh client_message_id with. */ @@ -73,6 +77,9 @@ export async function runDaemonCommand( return runDaemon({ tcpEnabled: !opts.noTcp, publicHealthCheck: opts.publicHealth, + // 1.37.1: under launchd/systemd (no TTY) own the log file with a + // rotating sink; in a terminal keep streaming to the console. + logSink: !!opts.logSink, }); case "help": @@ -90,6 +97,9 @@ export async function runDaemonCommand( case "stop": return runStop(opts); + case "restart": + return runRestart(opts); + case "accept-host": return runAcceptHost(opts); @@ -117,9 +127,10 @@ USAGE COMMANDS up | start start the daemon (detached by default) - status show running pid + IPC health + status pid, per-mesh broker WS, per-session WS, log size (1.37.1) version ipc + schema version of the running daemon down | stop stop the running daemon (SIGTERM, then wait) + restart stop, then relaunch via launchd/systemd (or detached) (1.37.1) accept-host pin the current host fingerprint outbox list list local outbox rows (newest first) outbox requeue re-enqueue an aborted / dead outbox row @@ -128,6 +139,7 @@ COMMANDS OPTIONS --foreground keep daemon attached to terminal, JSON logs to stdout (1.34.12+) + --log-sink write logs to ~/.claudemesh/daemon/daemon.log with rotation (20 MB × 5); set by service units (1.37.1+) --no-tcp disable the loopback TCP listener (UDS only) --public-health expose /v1/health unauthenticated on TCP --json machine-readable output where supported @@ -293,6 +305,26 @@ async function runAcceptHost(opts: DaemonOptions): Promise { return 0; } +interface DiagnosticsBody { + pid: number; + version: string; + uptime_s: number; + log: { path: string; bytes: number; rotating: boolean }; + meshes: Record; + sessions: Array<{ + sessionId: string; mesh: string; displayName: string; pid: number; cwd: string | null; + sessionPubkey: string | null; sessionPubkeyPrefix: string | null; registeredAt: string; + ws: { status: string; isOpen: boolean; lastAckAt: string | null; reconnects: number }; + }>; +} + +function fmtBytes(n: number): string { + if (n >= 1024 * 1024 * 1024) return `${(n / (1024 * 1024 * 1024)).toFixed(1)} GB`; + if (n >= 1024 * 1024) return `${(n / (1024 * 1024)).toFixed(1)} MB`; + if (n >= 1024) return `${(n / 1024).toFixed(0)} KB`; + return `${n} B`; +} + async function runStatus(opts: DaemonOptions): Promise { const pid = readRunningPid(); if (!pid) { @@ -300,13 +332,54 @@ async function runStatus(opts: DaemonOptions): Promise { else process.stdout.write("daemon: not running\n"); return 1; } + // 1.37.1: prefer /v1/diagnostics (per-mesh + per-session WS state). + // Older daemons 404 it → fall back to /v1/health. try { - const res = await ipc<{ ok: boolean; pid: number }>({ path: "/v1/health" }); + const diag = await ipc({ path: "/v1/diagnostics" }); + if (diag.status === 200 && diag.body && "meshes" in diag.body) { + const d = diag.body as DiagnosticsBody; + if (opts.json) { + process.stdout.write(JSON.stringify({ running: true, pid, diagnostics: d }) + "\n"); + return 0; + } + const mark = (ok: boolean) => (ok ? "✔" : "✘"); + process.stdout.write(`daemon: running (pid ${pid}, v${d.version}, up ${d.uptime_s}s)\n`); + process.stdout.write(`socket: ${DAEMON_PATHS.SOCK_FILE}\n`); + process.stdout.write(`log: ${d.log.path} (${fmtBytes(d.log.bytes)}${d.log.rotating ? ", rotating" : ", NOT rotating — reinstall service"})\n`); + process.stdout.write(`meshes (member WS):\n`); + const meshSlugs = Object.keys(d.meshes); + if (meshSlugs.length === 0) process.stdout.write(` (none)\n`); + for (const slug of meshSlugs) { + const m = d.meshes[slug]!; + const extra = m.reconnects ? `, reconnects=${m.reconnects}` : ""; + process.stdout.write(` ${mark(m.status === "open" && m.isOpen)} ${slug.padEnd(16)} ${m.status}${extra}\n`); + } + process.stdout.write(`sessions (session WS):\n`); + if (d.sessions.length === 0) process.stdout.write(` (none registered)\n`); + for (const sess of d.sessions) { + const ok = sess.ws.status === "open" && sess.ws.isOpen; + process.stdout.write( + ` ${mark(ok)} ${sess.displayName.padEnd(24)} ${sess.mesh.padEnd(12)} ${sess.ws.status.padEnd(12)} ` + + `${sess.sessionPubkeyPrefix ?? "(no pubkey)"}… pid ${sess.pid}\n`, + ); + } + const degraded = meshSlugs.some((m) => d.meshes[m]!.status !== "open") + || d.sessions.some((x) => x.ws.status !== "open"); + if (degraded) process.stdout.write(`\n⚠ degraded — run \`claudemesh doctor\` (fix is usually \`claudemesh daemon restart\`)\n`); + return degraded ? 1 : 0; + } + } catch { /* fall through to /v1/health */ } + try { + const res = await ipc<{ ok: boolean; pid: number; brokers?: Record }>({ path: "/v1/health" }); if (opts.json) { process.stdout.write(JSON.stringify({ running: true, pid, health: res.body }) + "\n"); } else { process.stdout.write(`daemon: running (pid ${pid})\n`); process.stdout.write(`socket: ${DAEMON_PATHS.SOCK_FILE}\n`); + if (res.body.brokers) { + for (const [slug, st] of Object.entries(res.body.brokers)) process.stdout.write(` ${slug}: ${st}\n`); + } + process.stdout.write(`(daemon predates /v1/diagnostics — run \`claudemesh daemon restart\` after upgrading)\n`); } return 0; } catch (err) { @@ -316,6 +389,56 @@ async function runStatus(opts: DaemonOptions): Promise { } } +/** + * 1.37.1: `claudemesh daemon restart` — the one-command fix `doctor` + * points at. Stops the running daemon, then relaunches through the + * service manager when a unit is installed (so launchd/systemd keep + * owning the lifecycle) or as a detached process otherwise. + */ +async function runRestart(opts: DaemonOptions): Promise { + const { readInstalledUnit } = await import("~/daemon/service-install.js"); + const unit = readInstalledUnit(); + const stopCode = await runStop({ ...opts, json: false }); + if (stopCode !== 0 && readRunningPid()) { + process.stderr.write(`restart: previous daemon did not stop; aborting\n`); + return 1; + } + const { execSync } = await import("node:child_process"); + if (unit.platform === "darwin" && unit.path) { + try { + // kickstart -k: kill if running, then start. Idempotent whether or + // not the unit is currently loaded (bootstrap first if needed). + try { execSync(`launchctl bootstrap gui/$(id -u) '${unit.path}'`, { stdio: "ignore" }); } catch { /* already loaded */ } + execSync(`launchctl kickstart -k gui/$(id -u)/com.claudemesh.daemon`, { stdio: "ignore" }); + } catch (err) { + process.stderr.write(`launchctl kickstart failed: ${String(err)}\n`); + return 1; + } + } else if (unit.platform === "linux" && unit.path) { + try { execSync(`systemctl --user restart claudemesh-daemon.service`, { stdio: "ignore" }); } + catch (err) { process.stderr.write(`systemctl restart failed: ${String(err)}\n`); return 1; } + } else { + return spawnDetachedDaemon(opts); + } + // Wait for the new daemon's socket + a live IPC answer. + const startedAt = Date.now(); + while (Date.now() - startedAt < 8_000) { + if (existsSync(DAEMON_PATHS.SOCK_FILE)) { + try { + const res = await ipc<{ ok: boolean; pid: number }>({ path: "/v1/health", timeoutMs: 1_000 }); + if (res.status === 200) { + if (opts.json) process.stdout.write(JSON.stringify({ restarted: true, pid: res.body.pid, via: unit.platform }) + "\n"); + else process.stdout.write(`daemon: restarted via ${unit.platform} (pid ${res.body.pid})\n`); + return 0; + } + } catch { /* not up yet */ } + } + await new Promise((r) => setTimeout(r, 200)); + } + process.stderr.write(`restart: service did not come back within 8s — check ${DAEMON_PATHS.LAUNCHD_LOG_FILE} and ${DAEMON_PATHS.LOG_FILE}\n`); + return 1; +} + async function runVersion(opts: DaemonOptions): Promise { try { const res = await ipc>({ path: "/v1/version" }); @@ -385,7 +508,7 @@ async function spawnDetachedDaemon(opts: DaemonOptions): Promise { // .../bin/claudemesh, for `bun run` dev that's the local dist file. // Either way it's the right thing to re-exec. const binary = process.argv[1] ?? "claudemesh"; - const args = ["daemon", "up", "--foreground"]; + const args = ["daemon", "up", "--foreground", "--log-sink"]; if (opts.noTcp) args.push("--no-tcp"); if (opts.publicHealth) args.push("--public-health"); diff --git a/apps/cli/src/commands/doctor.ts b/apps/cli/src/commands/doctor.ts index d630b0a..ad72d77 100644 --- a/apps/cli/src/commands/doctor.ts +++ b/apps/cli/src/commands/doctor.ts @@ -12,6 +12,8 @@ import { join } from "node:path"; import { spawnSync } from "node:child_process"; import { readConfig, getConfigPath } from "~/services/config/facade.js"; import { VERSION, URLS } from "~/constants/urls.js"; +import { DAEMON_PATHS } from "~/daemon/paths.js"; +import { isInstalledUnitStale } from "~/daemon/service-install.js"; interface Check { name: string; @@ -239,6 +241,227 @@ async function checkNpmLatest(): Promise { } } + +// ── 1.37.1 daemon / mesh liveness checks ─────────────────────────────── +// +// Added after the 2026-09-08 blackout: 8 sessions registered with the +// daemon, every WS "hello_acked", and yet `peer list` returned 0 on all +// meshes for 40 minutes. Nothing told the user. These checks cross-check +// what the daemon believes (its own WS state) against what the broker +// reports (the peer list), and always name the one-command fix. + +const RESTART_FIX = "claudemesh daemon restart"; +const REINSTALL_FIX = "claudemesh daemon install-service && launchctl kickstart -k gui/$(id -u)/com.claudemesh.daemon"; +const LOG_SIZE_LIMIT = 200 * 1024 * 1024; + +interface DiagMesh { status: string; isOpen: boolean; lastAckAt: string | null; reconnects: number } +interface DiagSession { + sessionId: string; mesh: string; displayName: string; pid: number; + sessionPubkey: string | null; sessionPubkeyPrefix: string | null; + ws: { status: string; isOpen: boolean }; +} +interface Diagnostics { + pid: number; version: string; uptime_s: number; + log: { path: string; bytes: number; rotating: boolean }; + meshes: Record; + sessions: DiagSession[]; +} + +function fmtBytes(n: number): string { + if (n >= 1024 * 1024 * 1024) return `${(n / (1024 * 1024 * 1024)).toFixed(1)} GB`; + if (n >= 1024 * 1024) return `${(n / (1024 * 1024)).toFixed(1)} MB`; + return `${Math.round(n / 1024)} KB`; +} + +/** + * Fetch the daemon's self-report. Returns `{ state: "down" }` when no + * daemon runs, `{ state: "old" }` when it predates `/v1/diagnostics`. + */ +async function fetchDiagnostics(): Promise< + | { state: "ok"; diag: Diagnostics } + | { state: "down" } + | { state: "old"; version: string | null } + | { state: "error"; detail: string } +> { + const { readRunningPid } = await import("~/daemon/lock.js"); + if (!readRunningPid()) return { state: "down" }; + try { + const { ipc } = await import("~/daemon/ipc/client.js"); + const res = await ipc({ path: "/v1/diagnostics", timeoutMs: 3_000 }); + if (res.status === 200 && res.body && "meshes" in res.body) return { state: "ok", diag: res.body as Diagnostics }; + let version: string | null = null; + try { + const v = await ipc<{ daemon_version?: string }>({ path: "/v1/version", timeoutMs: 2_000 }); + version = v.body?.daemon_version ?? null; + } catch { /* ignore */ } + return { state: "old", version }; + } catch (e) { + return { state: "error", detail: e instanceof Error ? e.message : String(e) }; + } +} + +async function daemonChecks(): Promise { + const skipped = (name: string, detail: string): Check => ({ name, pass: true, detail }); + const r = await fetchDiagnostics(); + const names = { + version: "Daemon running on the installed CLI version", + meshWs: "Daemon connected to the broker on every mesh", + sessionWs: "Every registered session has an open broker WS", + ghosts: "Every registered session is visible in the broker peer list", + log: "daemon.log bounded + service unit current", + }; + + if (r.state === "down") { + return [ + { name: names.version, pass: false, detail: "daemon not running", fix: "claudemesh daemon up (or: claudemesh daemon install-service)" }, + skipped(names.meshWs, "daemon not running — skipped"), + skipped(names.sessionWs, "daemon not running — skipped"), + skipped(names.ghosts, "daemon not running — skipped"), + logAndUnitCheck(null), + ]; + } + if (r.state === "error") { + return [ + { name: names.version, pass: false, detail: `daemon IPC unreachable: ${r.detail}`, fix: RESTART_FIX }, + skipped(names.meshWs, "IPC unreachable — skipped"), + skipped(names.sessionWs, "IPC unreachable — skipped"), + skipped(names.ghosts, "IPC unreachable — skipped"), + logAndUnitCheck(null), + ]; + } + if (r.state === "old") { + return [ + { + name: names.version, pass: false, + detail: `running daemon ${r.version ?? "(unknown)"} predates /v1/diagnostics; CLI is ${VERSION}`, + fix: RESTART_FIX, + }, + skipped(names.meshWs, "old daemon — skipped"), + skipped(names.sessionWs, "old daemon — skipped"), + skipped(names.ghosts, "old daemon — skipped"), + logAndUnitCheck(null), + ]; + } + + const d = r.diag; + const checks: Check[] = []; + + // (a) version parity — an upgraded CLI with a stale daemon is how a + // fix "ships" but never runs. + const sameVersion = d.version === VERSION; + checks.push({ + name: names.version, + pass: sameVersion, + detail: sameVersion ? `v${d.version}, pid ${d.pid}, up ${d.uptime_s}s` : `daemon v${d.version} ≠ CLI v${VERSION}`, + fix: sameVersion ? undefined : RESTART_FIX, + }); + + // (b) member WS per mesh. + const badMeshes = Object.entries(d.meshes).filter(([, m]) => m.status !== "open" || !m.isOpen); + checks.push({ + name: names.meshWs, + pass: badMeshes.length === 0, + detail: badMeshes.length === 0 + ? `${Object.keys(d.meshes).length} mesh(es) open` + : badMeshes.map(([slug, m]) => `${slug}: ${m.status}`).join(", "), + fix: badMeshes.length === 0 ? undefined : RESTART_FIX, + }); + + // (c) session WS per registered session. + const badSessions = d.sessions.filter((s) => s.ws.status !== "open" || !s.ws.isOpen); + checks.push({ + name: names.sessionWs, + pass: badSessions.length === 0, + detail: badSessions.length === 0 + ? `${d.sessions.length} session(s)` + : badSessions.map((s) => `${s.displayName}@${s.mesh}: ${s.ws.status}`).join(", "), + fix: badSessions.length === 0 ? undefined : RESTART_FIX, + }); + + // (d) GHOST DETECTOR — the exact 2026-09-08 signature: the daemon says + // the session WS is open, the broker has no presence for it. Ask the + // broker (via the daemon's member WS) for each mesh's peer list and + // require every registered session pubkey to be in it. + checks.push(await ghostCheck(d, names.ghosts)); + + // (e) log size + service unit. + checks.push(logAndUnitCheck(d)); + return checks; +} + +async function ghostCheck(d: Diagnostics, name: string): Promise { + const byMesh = new Map(); + for (const s of d.sessions) { + if (!s.sessionPubkey) continue; + const arr = byMesh.get(s.mesh) ?? []; + arr.push(s); + byMesh.set(s.mesh, arr); + } + if (byMesh.size === 0) return { name, pass: true, detail: "no registered sessions" }; + const { ipc } = await import("~/daemon/ipc/client.js"); + const missing: string[] = []; + const unreachable: string[] = []; + let checked = 0; + for (const [mesh, sessions] of byMesh) { + let pubkeys = new Set(); + try { + const res = await ipc<{ peers?: Array<{ pubkey?: string }>; brokers?: Record }>({ + path: `/v1/peers?mesh=${encodeURIComponent(mesh)}`, timeoutMs: 6_000, + }); + if (res.status !== 200) { unreachable.push(`${mesh}: http ${res.status}`); continue; } + if (res.body.brokers && res.body.brokers[mesh] && res.body.brokers[mesh] !== "open") { + unreachable.push(`${mesh}: member WS ${res.body.brokers[mesh]}`); + continue; + } + pubkeys = new Set((res.body.peers ?? []).map((p) => String(p.pubkey ?? "").toLowerCase())); + } catch (e) { + unreachable.push(`${mesh}: ${e instanceof Error ? e.message : String(e)}`); + continue; + } + for (const s of sessions) { + checked++; + if (!pubkeys.has(s.sessionPubkey!.toLowerCase())) { + missing.push(`${s.displayName}@${mesh} (${s.sessionPubkeyPrefix}…)`); + } + } + } + if (missing.length === 0 && unreachable.length === 0) { + return { name, pass: true, detail: `${checked} session(s) visible on the mesh` }; + } + const parts: string[] = []; + if (missing.length) parts.push(`GHOST — registered but not in peer list: ${missing.join(", ")}`); + if (unreachable.length) parts.push(`could not verify: ${unreachable.join(", ")}`); + return { name, pass: false, detail: parts.join(" · "), fix: RESTART_FIX }; +} + +function logAndUnitCheck(d: Diagnostics | null): Check { + const name = "daemon.log bounded + service unit current"; + const problems: string[] = []; + const fixes: string[] = []; + let bytes = d?.log.bytes ?? 0; + if (!d) { try { bytes = statSync(DAEMON_PATHS.LOG_FILE).size; } catch { bytes = 0; } } + if (bytes >= LOG_SIZE_LIMIT) { + problems.push(`daemon.log is ${fmtBytes(bytes)} (limit ${fmtBytes(LOG_SIZE_LIMIT)})`); + fixes.push(`${RESTART_FIX} (1.37.1+ daemon rotates it at 20 MB)`); + } + if (d && !d.log.rotating) { + problems.push("running daemon is not rotating its log"); + } + try { + const st = isInstalledUnitStale(); + if (st.stale) { + problems.push(`service unit stale: ${st.reason}`); + fixes.push(REINSTALL_FIX); + } + } catch { /* no unit / unsupported platform */ } + return { + name, + pass: problems.length === 0, + detail: problems.length === 0 ? `daemon.log ${fmtBytes(bytes)}${d?.log.rotating ? ", rotating" : ""}` : problems.join(" · "), + fix: problems.length === 0 ? undefined : [...new Set(fixes)].join(" ; "), + }; +} + export async function runDoctor(): Promise { const useColor = !process.env.NO_COLOR && process.env.TERM !== "dumb" && process.stdout.isTTY; @@ -258,6 +481,7 @@ export async function runDoctor(): Promise { checkKeypairs(), await checkBrokerWs(), await checkNpmLatest(), + ...(await daemonChecks()), ]; for (const c of checks) { diff --git a/apps/cli/src/commands/peers.ts b/apps/cli/src/commands/peers.ts index 541cfaa..85e20e0 100644 --- a/apps/cli/src/commands/peers.ts +++ b/apps/cli/src/commands/peers.ts @@ -111,7 +111,15 @@ function projectFields(record: PeerRecord, fields: string[]): Record { +/** 1.37.1: what listPeersForMesh learned about the daemon's broker link. */ +interface PeersResult { + peers: PeerRecord[]; + /** Daemon-reported member-WS status for this mesh; null on the cold + * path or with a pre-1.37.1 daemon. */ + brokerStatus: string | null; +} + +async function listPeersForMesh(slug: string): Promise { const config = readConfig(); const joined = config.meshes.find((m) => m.slug === slug); const selfMemberPubkey = joined?.pubkey ?? null; @@ -144,10 +152,13 @@ async function listPeersForMesh(slug: string): Promise { // `meshFromCtx` already does the right scoping when the slug is // passed; the CLI just wasn't passing it. try { - const { tryListPeersViaDaemon } = await import("~/services/bridge/daemon-route.js"); - const dr = await tryListPeersViaDaemon(slug); + const { tryListPeersViaDaemonDetailed } = await import("~/services/bridge/daemon-route.js"); + const dr = await tryListPeersViaDaemonDetailed(slug); if (dr !== null) { - return dr.map((p) => annotateSelf(p as PeerRecord, selfMemberPubkey, selfSessionPubkey)); + return { + peers: dr.peers.map((p) => annotateSelf(p as PeerRecord, selfMemberPubkey, selfSessionPubkey)), + brokerStatus: typeof dr.brokers[slug] === "string" ? dr.brokers[slug]! : null, + }; } } catch { /* daemon route helper not available; fall through */ } @@ -161,7 +172,7 @@ async function listPeersForMesh(slug: string): Promise { annotateSelf(p, selfMemberPubkey, selfSessionPubkey), ); }); - return result; + return { peers: result, brokerStatus: null }; } /** @@ -235,11 +246,24 @@ export async function runPeers(flags: PeersFlags): Promise { : null; const wantsJson = flags.json !== undefined && flags.json !== false; - const allJson: Array<{ mesh: string; peers: unknown[] }> = []; + const allJson: Array<{ mesh: string; peers: unknown[]; brokerStatus?: string | null }> = []; for (const slug of slugs) { try { - const peers = await listPeersForMesh(slug); + const { peers, brokerStatus } = await listPeersForMesh(slug); + + // 1.37.1: a daemon whose member-WS for this mesh is not open cannot + // ask the broker, so its peer list is empty by construction — not + // because the mesh is empty. Say so instead of printing a silent + // empty list (the 2026-09-08 blackout looked exactly like "0 peers + // on every mesh" for 40 min with 8 live sessions). + const brokerDegraded = brokerStatus !== null && brokerStatus !== "open"; + if (brokerDegraded && !wantsJson) { + process.stderr.write( + `⚠ daemon is not connected to the broker for ${slug} (status: ${brokerStatus}) — ` + + `peer list may be incomplete; run \`claudemesh doctor\`\n`, + ); + } // Hide control-plane rows by default — they're infrastructure // (daemon-WS member-keyed presence), not interactive peers, and @@ -262,7 +286,7 @@ export async function runPeers(flags: PeersFlags): Promise { const projected = fieldList ? visible.map((p) => projectFields(p, fieldList)) : visible; - allJson.push({ mesh: slug, peers: projected }); + allJson.push({ mesh: slug, peers: projected, ...(brokerStatus !== null ? { brokerStatus } : {}) }); continue; } diff --git a/apps/cli/src/daemon/broker.ts b/apps/cli/src/daemon/broker.ts index b968741..d8c0822 100644 --- a/apps/cli/src/daemon/broker.ts +++ b/apps/cli/src/daemon/broker.ts @@ -17,7 +17,7 @@ import type { JoinedMesh } from "~/services/config/facade.js"; import { signHello } from "~/services/broker/hello-sig.js"; -import { connectWsWithBackoff, type WsLifecycle, type WsStatus } from "./ws-lifecycle.js"; +import { createWsLifecycle, type WsLifecycle, type WsStatus } from "./ws-lifecycle.js"; export type ConnStatus = WsStatus; @@ -99,6 +99,8 @@ export interface DaemonBrokerOptions { onStatusChange?: (s: ConnStatus) => void; onPush?: (msg: Record) => void; log?: (level: "info" | "warn" | "error", msg: string, meta?: Record) => void; + /** Lifecycle timing overrides (tests / tuning). Defaults live in ws-lifecycle.ts. */ + lifecycle?: { helloAckTimeoutMs?: number; backoffCapsMs?: readonly number[] }; } export class DaemonBrokerClient { @@ -126,12 +128,22 @@ export class DaemonBrokerClient { (this.opts.log ?? defaultLog)(level, msg, { mesh: this.mesh.slug, ...meta }); }; - /** Open the WS, run the hello handshake, resolve once the broker accepts. */ + /** + * Open the WS, run the hello handshake, resolve once the broker accepts. + * + * 1.37.1: the lifecycle handle is stored SYNCHRONOUSLY before awaiting + * readiness. Pre-1.37.1 `this.lifecycle = await connectWsWithBackoff(…)` + * left `lifecycle === null` forever when the first hello ack timed out + * (the helper kept reconnecting on its own and flipped `_status` to + * "open", but every RPC gated on `this.lifecycle` returned []). See + * spec 2026-09-08-daemon-restart-mesh-blackout addendum 3. Idempotent: + * a second call while a lifecycle exists just awaits the same `ready`. + */ async connect(): Promise { if (this.closed) throw new Error("client_closed"); - if (this._status === "connecting" || this._status === "open") return; + if (this.lifecycle) return this.lifecycle.ready; - this.lifecycle = await connectWsWithBackoff({ + this.lifecycle = createWsLifecycle({ url: this.mesh.brokerUrl, buildHello: async () => { const { timestamp, signature } = await signHello( @@ -172,7 +184,20 @@ export class DaemonBrokerClient { }, onBeforeReconnect: (code) => this.failPendingAcks(`broker_disconnected_${code}`), log: (level, msg, meta) => this.log(level, `broker_${msg}`, meta), + ...(this.opts.lifecycle ?? {}), }); + await this.lifecycle.ready; + } + + /** Operator diagnostics — surfaced by `/v1/diagnostics` + `claudemesh doctor`. */ + diagnostics(): { status: ConnStatus; isOpen: boolean; lastAckAt: number | null; reconnects: number; replaced: boolean } { + return { + status: this._status, + isOpen: this.isOpen(), + lastAckAt: this.lifecycle?.lastAckAt ?? null, + reconnects: this.lifecycle?.reconnects ?? 0, + replaced: this.lifecycle?.replaced ?? false, + }; } private handleMessage(msg: Record): void { @@ -276,7 +301,7 @@ export class DaemonBrokerClient { } /** True when underlying socket is OPEN-ready for direct sends. */ - private isOpen(): boolean { + isOpen(): boolean { const sock = this.lifecycle?.ws; return !!sock && sock.readyState === sock.OPEN; } diff --git a/apps/cli/src/daemon/drain.ts b/apps/cli/src/daemon/drain.ts index a7239bb..f27af8a 100644 --- a/apps/cli/src/daemon/drain.ts +++ b/apps/cli/src/daemon/drain.ts @@ -17,9 +17,25 @@ import type { SessionBrokerClient } from "./session-broker.js"; import type { OutboxStatus } from "./db/outbox.js"; const POLL_INTERVAL_MS = 500; -const MAX_ATTEMPTS_PER_ROW = 25; +export const MAX_ATTEMPTS_PER_ROW = 25; const BACKOFF_BASE_MS = 500; const BACKOFF_CAP_MS = 30_000; +/** + * 1.37.1 — outbox hygiene (spec 2026-09-08-daemon-restart-mesh-blackout §8). + * + * Before these caps, a row bound to a session WS that never came back + * (stale session pubkey from a previous day) was retried forever: the + * not-ready path neither counted attempts nor expired, and it logged + * `drain_session_ws_not_ready` on EVERY tick. Four such rows reached + * ~130k attempts each and grew daemon.log to 1 GB (disk at 95%). + */ +/** Any pending/inflight row older than this is dead (`expired`). */ +export const OUTBOX_TTL_MS = 24 * 60 * 60 * 1000; +/** A session-attributed row whose session is no longer registered at all + * (not merely reconnecting) is dead after this age (`session_gone`). */ +export const SESSION_GONE_TTL_MS = 10 * 60 * 1000; +/** `drain_session_ws_not_ready` is logged on attempt 1 and every Nth after. */ +export const NOT_READY_LOG_EVERY = 60; interface PendingRow { id: string; @@ -27,6 +43,7 @@ interface PendingRow { request_fingerprint: Uint8Array; payload: Uint8Array; attempts: number; + enqueued_at: number; /** Sprint 4 routing fields. NULL on legacy v0.9.0 rows → broadcast fallback. */ target_spec: string | null; nonce: string | null; @@ -60,6 +77,8 @@ export interface DrainOptions { */ getSessionBrokerByPubkey?: (sessionPubkey: string) => SessionBrokerClient | undefined; log?: (level: "info" | "warn" | "error", msg: string, meta?: Record) => void; + /** Clock override for tests. Defaults to Date.now. */ + nowFn?: () => number; } export interface DrainHandle { @@ -103,10 +122,21 @@ export function startDrainWorker(opts: DrainOptions): DrainHandle { }; } -async function drainOnce(opts: DrainOptions, log: NonNullable): Promise { - const now = Date.now(); +/** Exported for tests: one drain pass. */ +export async function drainOnce(opts: DrainOptions, log: NonNullable = opts.log ?? defaultLog): Promise { + const now = (opts.nowFn ?? Date.now)(); + + // TTL sweep: anything still undelivered after OUTBOX_TTL_MS is dead. + // One UPDATE per tick; cheap on the (status, next_attempt_at) index. + const expired = Number(opts.db.prepare(` + UPDATE outbox + SET status = 'dead', last_error = 'expired' + WHERE status IN ('pending', 'inflight') AND enqueued_at < ? + `).run(now - OUTBOX_TTL_MS).changes); + if (expired > 0) log("warn", "drain_expired_rows", { count: expired, ttl_ms: OUTBOX_TTL_MS }); + const rows = opts.db.prepare(` - SELECT id, client_message_id, request_fingerprint, payload, attempts, + SELECT id, client_message_id, request_fingerprint, payload, attempts, enqueued_at, target_spec, nonce, ciphertext, priority, mesh, sender_session_pubkey FROM outbox @@ -183,10 +213,29 @@ async function drainOnce(opts: DrainOptions, log: NonNullable SESSION_GONE_TTL_MS) { + log("warn", "drain_session_gone", { + id: row.id, session_pubkey: row.sender_session_pubkey.slice(0, 12), attempts, + }); + markDead(opts.db, row.id, "session_gone"); + continue; + } + if (attempts === 1 || attempts % NOT_READY_LOG_EVERY === 0) { + log("info", "drain_session_ws_not_ready", { + id: row.id, session_pubkey: row.sender_session_pubkey.slice(0, 12), attempts, + registered: !!sessionBroker, + }); + } + if (attempts >= MAX_ATTEMPTS_PER_ROW) { + log("warn", "drain_max_attempts", { id: row.id, err: "session_ws_not_open" }); + markDead(opts.db, row.id, "max_attempts: session_ws_not_open"); + continue; + } + backoffPending(opts.db, row.id, attempts, "session_ws_not_open", "session_ws_not_open", now); continue; } res = await sessionBroker.send(sendArgs); @@ -195,7 +244,7 @@ async function drainOnce(opts: DrainOptions, log: NonNullable; /** Notify when a new outbox row was inserted (drains can wake). */ onPendingInserted?: () => void; + /** 1.37.1: per-session WS clients keyed by session token (for /v1/diagnostics). */ + sessionBrokers?: Map; + /** 1.37.1: per-session WS clients keyed by hex session pubkey. */ + sessionBrokersByPubkey?: Map; + /** 1.37.1: rotating log sink handle (reports the active log size). */ + logSink?: LogSinkHandle | null; + /** 1.37.1: daemon start time (ms) for uptime reporting. */ + startedAt?: number; } export interface IpcServerHandle { @@ -61,6 +72,11 @@ export interface IpcServerHandle { export function startIpcServer(opts: IpcServerOptions): IpcServerHandle { const log = opts.log ?? defaultLogger; + // 1.37.1: live SSE streams. `close()` ends each one BEFORE the servers + // shut down so an MCP plugin sees a clean `end` (→ resubscribe) rather + // than a hard socket drop when the daemon exits. + const sseCleanups = new Set<() => void>(); + const handler = makeHandler({ localToken: opts.localToken, publicHealthCheck: !!opts.publicHealthCheck, @@ -71,6 +87,11 @@ export function startIpcServer(opts: IpcServerOptions): IpcServerHandle { brokers: opts.brokers, meshConfigs: opts.meshConfigs, onPendingInserted: opts.onPendingInserted, + sessionBrokers: opts.sessionBrokers, + sessionBrokersByPubkey: opts.sessionBrokersByPubkey, + logSink: opts.logSink ?? null, + startedAt: opts.startedAt ?? Date.now(), + sseCleanups, }); // --- UDS listener ------------------------------------------------------- @@ -106,6 +127,10 @@ export function startIpcServer(opts: IpcServerOptions): IpcServerHandle { tcp, ready: Promise.all([udsReady, tcpReady]).then(() => undefined), close: async () => { + for (const cleanup of [...sseCleanups]) { + try { cleanup(); } catch { /* ignore */ } + } + sseCleanups.clear(); await Promise.allSettled([ new Promise((res) => uds.close(() => res())), tcp ? new Promise((res) => tcp!.close(() => res())) : Promise.resolve(), @@ -115,6 +140,31 @@ export function startIpcServer(opts: IpcServerOptions): IpcServerHandle { }; } +/** + * 1.37.1: normalize a broker client's self-report. Prefers the client's + * `diagnostics()` (added alongside the lifecycle fix) and degrades to the + * `status` / `isOpen()` surface when running against an older client. + */ +function brokerDiagnostics( + client: { status: string; isOpen?: () => boolean; diagnostics?: () => { status: string; isOpen: boolean; lastAckAt?: number | string | null; reconnects?: number } }, +): { status: string; isOpen: boolean; lastAckAt: string | null; reconnects: number } { + try { + if (typeof client.diagnostics === "function") { + const d = client.diagnostics(); + const ack = d.lastAckAt; + return { + status: d.status, + isOpen: !!d.isOpen, + lastAckAt: typeof ack === "number" ? new Date(ack).toISOString() : (ack ?? null), + reconnects: d.reconnects ?? 0, + }; + } + } catch { /* fall through */ } + let isOpen = false; + try { isOpen = typeof client.isOpen === "function" ? client.isOpen() : client.status === "open"; } catch { isOpen = false; } + return { status: client.status, isOpen, lastAckAt: null, reconnects: 0 }; +} + function defaultLogger(level: "info" | "warn" | "error", msg: string, meta?: Record) { const line = JSON.stringify({ level, msg, ...meta, ts: new Date().toISOString() }); if (level === "info") process.stdout.write(line + "\n"); @@ -131,6 +181,11 @@ function makeHandler(opts: { brokers?: Map; meshConfigs?: Map; onPendingInserted?: () => void; + sessionBrokers?: Map; + sessionBrokersByPubkey?: Map; + logSink: LogSinkHandle | null; + startedAt: number; + sseCleanups: Set<() => void>; }) { const tokenBytes = Buffer.from(opts.localToken, "utf8"); @@ -197,7 +252,7 @@ function makeHandler(opts: { respond(res, 200, { daemon_version: VERSION, ipc_api: "v1", - ipc_features: ["version", "health", "send", "inbox", "events", "peers", "profile", "skills", "state", "memory", "sessions"], + ipc_features: ["version", "health", "send", "inbox", "events", "peers", "profile", "skills", "state", "memory", "sessions", "diagnostics"], schema_version: 1, }); return; @@ -218,6 +273,45 @@ function makeHandler(opts: { return; } + if (req.method === "GET" && url.pathname === "/v1/diagnostics") { + // 1.37.1: the operator signal behind `claudemesh daemon status` and + // `claudemesh doctor`. Reports what the daemon believes about every + // socket it owns so a "registered + hello_acked but invisible on + // the mesh" state (the 2026-09-08 blackout) can be detected by + // cross-checking against `/v1/peers`. + const meshes: Record = {}; + if (opts.brokers) { + for (const [slug, client] of opts.brokers) meshes[slug] = brokerDiagnostics(client); + } + const sessions = listSessions().map((s) => { + const pk = s.presence?.sessionPubkey ?? null; + const client = pk ? opts.sessionBrokersByPubkey?.get(pk) : undefined; + const d = client ? brokerDiagnostics(client) : null; + return { + sessionId: s.sessionId, + mesh: s.mesh, + displayName: s.displayName, + pid: s.pid, + cwd: s.cwd ?? null, + sessionPubkey: pk, + sessionPubkeyPrefix: pk ? pk.slice(0, 12) : null, + registeredAt: new Date(s.registeredAt).toISOString(), + ws: d ?? { status: "none", isOpen: false, lastAckAt: null, reconnects: 0 }, + }; + }); + let logBytes = 0; + try { logBytes = statSync(DAEMON_PATHS.LOG_FILE).size; } catch { /* missing */ } + respond(res, 200, { + pid: process.pid, + version: VERSION, + uptime_s: Math.round((Date.now() - opts.startedAt) / 1000), + log: { path: DAEMON_PATHS.LOG_FILE, bytes: logBytes, rotating: !!opts.logSink }, + meshes, + sessions, + }); + return; + } + // Session registry routes (1.29.0) if (req.method === "POST" && url.pathname === "/v1/sessions/register") { try { @@ -333,7 +427,9 @@ function makeHandler(opts: { const meshCfg = opts.meshConfigs?.get(session.mesh); if (meshCfg?.pubkey) filter.memberPubkey = meshCfg.pubkey; } - bindSseStream(res, opts.bus, filter); + const cleanup = bindSseStream(res, opts.bus, filter); + opts.sseCleanups.add(cleanup); + res.on("close", () => opts.sseCleanups.delete(cleanup)); return; } @@ -348,8 +444,16 @@ function makeHandler(opts: { // `mesh` field so the caller can scope client-side. A single // ?mesh= filter narrows the set server-side. const all: Array & { mesh: string }> = []; + // 1.37.1: report the member-WS status per iterated mesh so the + // CLI can tell "no peers" from "daemon not connected to the + // broker" (the latter used to print as a silent empty list). + const brokerStatus: Record = {}; for (const [slug, b] of opts.brokers.entries()) { if (filterMesh && filterMesh !== slug) continue; + brokerStatus[slug] = b.status; + if (b.status !== "open") { + opts.log("warn", "ipc_peers_broker_not_open", { mesh: slug, status: b.status }); + } try { const peers = await b.listPeers(); for (const p of peers) all.push({ ...(p as unknown as Record), mesh: slug }); @@ -357,7 +461,7 @@ function makeHandler(opts: { opts.log("warn", "ipc_peers_broker_failed", { mesh: slug, err: String(e) }); } } - respond(res, 200, { peers: all }); + respond(res, 200, { peers: all, brokers: brokerStatus }); } catch (e) { respond(res, 502, { error: "broker_unreachable", detail: String(e) }); } diff --git a/apps/cli/src/daemon/log-sink.ts b/apps/cli/src/daemon/log-sink.ts new file mode 100644 index 0000000..72e7b4d --- /dev/null +++ b/apps/cli/src/daemon/log-sink.ts @@ -0,0 +1,226 @@ +/** + * Rotating log sink for the daemon (1.37.1). + * + * Why this exists: the daemon writes JSON lines to `process.stdout` / + * `process.stderr` from ~20 call sites, and under launchd / systemd those + * streams were redirected straight into `~/.claudemesh/daemon/daemon.log` + * by the service unit (`StandardOutPath`). launchd does not rotate — the + * 2026-09-08 incident found a 1.0 GB `daemon.log` (disk at 95 %), most + * plausibly the reason the daemon died silently. + * + * The sink takes ownership of both streams in-process: every write is + * appended synchronously to `path` through our own fd, and the file is + * rotated when it reaches `maxBytes` (`path.1` … `path.`, oldest + * deleted). A total-size budget (`keep × maxBytes`) is enforced across + * the rotated archives so a pre-existing oversized file cannot pin the + * disk after upgrade. + * + * Failure policy: a throw anywhere inside the sink must NEVER take the + * daemon down. On any error the sink falls back to the original + * `write` for that call (which goes wherever launchd pointed stdio — + * `daemon.launchd.log` on 1.37.1+ units, the old path on stale units). + * + * Spec: .artifacts/specs/2026-09-08-daemon-restart-mesh-blackout.md, + * Addendum 5 item 9. + */ + +import { + closeSync, + existsSync, + fstatSync, + mkdirSync, + openSync, + renameSync, + statSync, + unlinkSync, + writeSync, +} from "node:fs"; +import { dirname } from "node:path"; + +export interface LogSinkOptions { + /** Active log file (e.g. DAEMON_PATHS.LOG_FILE). */ + path: string; + /** Rotate when the active file reaches this size. Default 20 MB. */ + maxBytes?: number; + /** Rotated archives to keep (`path.1` … `path.keep`). Default 5. */ + keep?: number; + /** Also mirror lines to the original stdout/stderr (dev use). Default false. */ + tee?: boolean; +} + +export interface LogSinkHandle { + /** Append one line (a trailing newline is added when missing). */ + writeLine(line: string): void; + /** Force a rotation now (used by tests + `daemon rotate-log`). */ + rotateNow(): void; + /** Bytes currently in the active file (0 when unknown). */ + currentSize(): number; + /** Restore the original process.stdout/stderr writes and close the fd. */ + uninstall(): void; + readonly path: string; +} + +export const DEFAULT_LOG_MAX_BYTES = 20 * 1024 * 1024; +export const DEFAULT_LOG_KEEP = 5; + +type WriteFn = typeof process.stdout.write; + +/** + * Rotate `path` → `path.1`, shifting existing archives up by one and + * deleting anything beyond `keep`. Pure filesystem operation; safe to + * call when `path` does not exist (no-op). + */ +export function rotateLogFiles(path: string, keep: number): void { + if (!existsSync(path)) return; + // Drop the oldest slot if occupied. + const oldest = `${path}.${keep}`; + if (existsSync(oldest)) { try { unlinkSync(oldest); } catch { /* ignore */ } } + for (let n = keep - 1; n >= 1; n--) { + const from = `${path}.${n}`; + if (existsSync(from)) { try { renameSync(from, `${path}.${n + 1}`); } catch { /* ignore */ } } + } + renameSync(path, `${path}.1`); +} + +/** Size of a file in bytes, 0 when missing/unreadable. */ +function sizeOf(p: string): number { + try { return statSync(p).size; } catch { return 0; } +} + +/** + * Enforce the archive budget: while the total size of `path.1..path.keep` + * exceeds `keep × maxBytes`, delete the OLDEST archive (highest index) + * first. Returns the list of deleted paths. + * + * This is what shrinks a 1 GB pre-1.37.1 log after upgrade: the install + * step rotates it to `path.1`, and since `path.1` alone exceeds the whole + * budget it is deleted right away. Evidence worth keeping must be + * extracted BEFORE upgrade (the 2026-09-08 excerpts live in + * `.artifacts/test-runs/`). + */ +export function purgeArchivesOverBudget(path: string, keep: number, maxBytes: number): string[] { + const deleted: string[] = []; + const budget = keep * maxBytes; + const total = () => { + let t = 0; + for (let n = 1; n <= keep; n++) t += sizeOf(`${path}.${n}`); + return t; + }; + for (let n = keep; n >= 1 && total() > budget; n--) { + const p = `${path}.${n}`; + if (!existsSync(p)) continue; + try { unlinkSync(p); deleted.push(p); } catch { /* ignore */ } + } + return deleted; +} + +/** + * Take over process.stdout/stderr and route every write into a rotating + * file. Returns a handle; call `uninstall()` to restore the originals. + */ +export function installRotatingLogSink(opts: LogSinkOptions): LogSinkHandle { + const path = opts.path; + const maxBytes = opts.maxBytes ?? DEFAULT_LOG_MAX_BYTES; + const keep = Math.max(1, opts.keep ?? DEFAULT_LOG_KEEP); + const tee = !!opts.tee; + + const origOut: WriteFn = process.stdout.write.bind(process.stdout) as WriteFn; + const origErr: WriteFn = process.stderr.write.bind(process.stderr) as WriteFn; + + let fd: number | null = null; + let size = 0; + let installed = false; + let inSink = false; // re-entrancy guard (a throw → fallback write must not recurse) + + const open = (): void => { + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + fd = openSync(path, "a", 0o600); + try { size = fstatSync(fd).size; } catch { size = 0; } + }; + + const closeFd = (): void => { + if (fd !== null) { try { closeSync(fd); } catch { /* ignore */ } fd = null; } + }; + + const rotate = (): void => { + closeFd(); + rotateLogFiles(path, keep); + purgeArchivesOverBudget(path, keep, maxBytes); + open(); + }; + + // Install-time hygiene: an oversized existing file (pre-1.37.1 launchd + // log) is rotated immediately so the active file starts small, then + // the archive budget is enforced. + if (sizeOf(path) >= maxBytes) { + try { rotateLogFiles(path, keep); } catch { /* ignore */ } + } + try { purgeArchivesOverBudget(path, keep, maxBytes); } catch { /* ignore */ } + open(); + + const append = (chunk: string | Uint8Array): boolean => { + if (fd === null) return false; + const buf = typeof chunk === "string" ? Buffer.from(chunk, "utf8") : Buffer.from(chunk); + if (size + buf.length >= maxBytes && size > 0) rotate(); + let off = 0; + while (off < buf.length) { + off += writeSync(fd!, buf, off, buf.length - off); + } + size += buf.length; + return true; + }; + + const makeWrite = (orig: WriteFn): WriteFn => { + const w = function (this: unknown, chunk: any, encodingOrCb?: any, cb?: any): boolean { + const callback = typeof encodingOrCb === "function" ? encodingOrCb : cb; + if (!inSink) { + inSink = true; + try { + const str = typeof chunk === "string" + ? chunk + : Buffer.isBuffer(chunk) || chunk instanceof Uint8Array + ? Buffer.from(chunk).toString(typeof encodingOrCb === "string" ? (encodingOrCb as BufferEncoding) : "utf8") + : String(chunk); + const ok = append(str); + inSink = false; + if (ok) { + if (tee) { try { orig(str); } catch { /* ignore */ } } + if (typeof callback === "function") { try { callback(); } catch { /* ignore */ } } + return true; + } + } catch { + inSink = false; + // fall through to the original write below + } + } + // Fallback: original stream (launchd/systemd redirect target). + try { return orig(chunk, encodingOrCb, cb); } catch { return false; } + }; + return w as unknown as WriteFn; + }; + + const sinkOut = makeWrite(origOut); + const sinkErr = makeWrite(origErr); + (process.stdout as { write: WriteFn }).write = sinkOut; + (process.stderr as { write: WriteFn }).write = sinkErr; + installed = true; + + return { + path, + writeLine(line: string) { + const l = line.endsWith("\n") ? line : line + "\n"; + try { if (!append(l)) origErr(l); } catch { try { origErr(l); } catch { /* ignore */ } } + }, + rotateNow() { + try { rotate(); } catch { /* ignore */ } + }, + currentSize() { return size; }, + uninstall() { + if (!installed) return; + installed = false; + (process.stdout as { write: WriteFn }).write = origOut; + (process.stderr as { write: WriteFn }).write = origErr; + closeFd(); + }, + }; +} diff --git a/apps/cli/src/daemon/paths.ts b/apps/cli/src/daemon/paths.ts index 2a56a61..05fbe97 100644 --- a/apps/cli/src/daemon/paths.ts +++ b/apps/cli/src/daemon/paths.ts @@ -31,6 +31,15 @@ export const DAEMON_PATHS = { get OUTBOX_DB() { return join(this.DAEMON_DIR, "outbox.db"); }, get INBOX_DB() { return join(this.DAEMON_DIR, "inbox.db"); }, get LOG_FILE() { return join(this.DAEMON_DIR, "daemon.log"); }, + /** 1.37.1: launchd/systemd StandardOut/Err target. Only receives + * writes from BEFORE the in-process rotating sink is installed + * (boot crashes, `node` startup errors) — the daemon's own log + * lines go to LOG_FILE via `installRotatingLogSink`, which rotates. + * Pre-1.37.1 units pointed StandardOutPath at LOG_FILE directly, + * which is how daemon.log reached 1 GB with no rotation. */ + get LAUNCHD_LOG_FILE() { return join(this.DAEMON_DIR, "daemon.launchd.log"); }, + /** Rotated archive name for LOG_FILE: daemon.log.1 … daemon.log.N. */ + rotatedLogFile(n: number): string { return `${join(this.DAEMON_DIR, "daemon.log")}.${n}`; }, /** Persisted session→mesh bindings. Rehydrated on daemon restart so a * restart never orphans a live session's mesh context (the bug where * a peer looked "disconnected" after the daemon bounced). Holds no diff --git a/apps/cli/src/daemon/run.ts b/apps/cli/src/daemon/run.ts index c4aa776..606d864 100644 --- a/apps/cli/src/daemon/run.ts +++ b/apps/cli/src/daemon/run.ts @@ -1,4 +1,4 @@ -import { existsSync, mkdirSync, readFileSync } from "node:fs"; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { DAEMON_PATHS } from "./paths.js"; import { acquireSingletonLock, releaseSingletonLock } from "./lock.js"; @@ -15,8 +15,10 @@ import { startInboxPruner, type InboxPrunerHandle } from "./inbox-pruner.js"; import { handleBrokerPush } from "./inbound.js"; import { EventBus } from "./events.js"; import { checkFingerprint, type ClonePolicy } from "./identity.js"; +import { isPidAlive } from "./process-info.js"; import { readConfig } from "~/services/config/facade.js"; import { VERSION } from "~/constants/urls.js"; +import { installRotatingLogSink, type LogSinkHandle } from "./log-sink.js"; export interface RunDaemonOptions { /** Disable TCP loopback (UDS-only). Defaults true in container envs. */ @@ -24,6 +26,62 @@ export interface RunDaemonOptions { publicHealthCheck?: boolean; /** Behavior on host_fingerprint mismatch. Defaults 'refuse'. */ clonePolicy?: ClonePolicy; + /** + * 1.37.1: route stdout/stderr into a rotating `daemon.log` (20 MB × 5) + * instead of relying on the service unit's StandardOutPath, which + * never rotates. `daemon up --foreground` sets this when stdout is + * not a TTY (launchd / systemd); a human running it in a terminal + * keeps the stream on the console. + */ + logSink?: boolean; +} + +/** + * 1.37.1: ask whoever is listening on the daemon socket for its pid. + * Returns null when nothing answers (absent socket, refused, timeout). + */ +async function probeLiveDaemonOnSocket(): Promise { + if (!existsSync(DAEMON_PATHS.SOCK_FILE)) return null; + try { + const { ipc } = await import("./ipc/client.js"); + const res = await ipc<{ pid?: number }>({ path: "/v1/health", timeoutMs: 1_500 }); + const pid = res.status === 200 && typeof res.body?.pid === "number" ? res.body.pid : null; + return pid !== null && isPidAlive(pid) ? pid : null; + } catch { + return null; + } +} + +/** + * 1.37.1: keep the pid file truthful for the daemon's whole life. It used + * to be written once at boot; anything that unlinked it afterwards (the + * CLI stale-probe, a user `rm`) silently disabled the singleton lock and + * `daemon status`/`restart`. Every 30 s: missing or pointing at a dead + * pid → rewrite ours; pointing at ANOTHER live pid → log the conflict + * (two daemons) and leave it — the operator needs to see that. + */ +function startPidFileGuard(): NodeJS.Timeout { + const tick = () => { + try { + let raw: string | null = null; + try { raw = readFileSync(DAEMON_PATHS.PID_FILE, "utf8").trim(); } catch { raw = null; } + if (raw === String(process.pid)) return; + const other = raw ? Number.parseInt(raw, 10) : Number.NaN; + if (Number.isFinite(other) && other > 0 && other !== process.pid && isPidAlive(other)) { + process.stderr.write(JSON.stringify({ + level: "warn", msg: "daemon_pidfile_conflict", other_pid: other, ts: new Date().toISOString(), + }) + "\n"); + return; + } + writeFileSync(DAEMON_PATHS.PID_FILE, String(process.pid), { mode: 0o600 }); + process.stderr.write(JSON.stringify({ + level: "warn", msg: "daemon_pidfile_restored", was: raw, ts: new Date().toISOString(), + }) + "\n"); + } catch { /* never let the guard throw */ } + }; + const t = setInterval(tick, 30_000); + t.unref(); + return t; } /** Detect a few common container environments to pick UDS-only by default. */ @@ -38,8 +96,75 @@ function detectContainer(): boolean { return false; } +/** Write one JSON line straight to the log, bypassing anything async. + * Used by the exit/crash forensics — by the time these fire, the event + * loop may be gone, so only synchronous writes are guaranteed. */ +function forensicLine(sink: LogSinkHandle | null, obj: Record): void { + const line = JSON.stringify({ ...obj, pid: process.pid, ts: new Date().toISOString() }) + "\n"; + try { + if (sink) sink.writeLine(line); + else process.stderr.write(line); + } catch { /* nothing left to do */ } +} + +/** + * 1.37.1 forensics: the 2026-09-08 daemon (pid 85205) died after 3 days + * without a single log line, and its respawn died within 6 s — also + * silently. `shutdown()` only logs on SIGINT/SIGTERM, so an uncaught + * exception, an unhandled rejection or a plain `process.exit` from a + * library left no trace. These handlers make every exit path leave a + * `daemon_exit` line (and `daemon_uncaught` with the stack when the + * cause is a throw). + */ +function installExitForensics(sink: LogSinkHandle | null, startedAt: number): void { + const uptime = () => Math.round((Date.now() - startedAt) / 1000); + process.on("uncaughtException", (err, origin) => { + forensicLine(sink, { + level: "error", msg: "daemon_uncaught", kind: origin, + err: err instanceof Error ? `${err.name}: ${err.message}` : String(err), + stack: err instanceof Error ? err.stack : undefined, + uptime_s: uptime(), + }); + process.exit(1); + }); + process.on("unhandledRejection", (reason) => { + forensicLine(sink, { + level: "error", msg: "daemon_uncaught", kind: "unhandledRejection", + err: reason instanceof Error ? `${reason.name}: ${reason.message}` : String(reason), + stack: reason instanceof Error ? reason.stack : undefined, + uptime_s: uptime(), + }); + process.exit(1); + }); + process.on("exit", (code) => { + forensicLine(sink, { level: "info", msg: "daemon_exit", code, uptime_s: uptime() }); + }); +} + export async function runDaemon(opts: RunDaemonOptions = {}): Promise { mkdirSync(DAEMON_PATHS.DAEMON_DIR, { recursive: true, mode: 0o700 }); + const startedAt = Date.now(); + + // 1.37.1: take over stdio BEFORE anything logs so the rotating sink + // sees every line (boot included). Must never throw — fall back to + // plain stdio if the file can't be opened. + let logSink: LogSinkHandle | null = null; + if (opts.logSink) { + try { logSink = installRotatingLogSink({ path: DAEMON_PATHS.LOG_FILE }); } + catch (e) { process.stderr.write(`log sink install failed, logging to stdio: ${String(e)}\n`); } + } + installExitForensics(logSink, startedAt); + + // 1.37.1: belt-and-braces singleton. The pid file alone is not enough — + // a CLI stale-probe used to unlink it under a live daemon (fixed in + // services/daemon/lifecycle.ts), which let a second daemon start and + // fight the first over every session. If the socket answers /v1/health + // with a different live pid, that daemon owns this directory. + const livePeer = await probeLiveDaemonOnSocket(); + if (livePeer !== null && livePeer !== process.pid) { + process.stderr.write(`daemon already running (pid ${livePeer}, answering on ${DAEMON_PATHS.SOCK_FILE})\n`); + return 1; + } const lock = acquireSingletonLock(); if (lock.result === "already-running") { @@ -49,6 +174,7 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { if (lock.result === "stale") { process.stderr.write(`recovered stale pid file; starting fresh\n`); } + const pidGuard = startPidFileGuard(); // Accidental-clone detection (spec §2.2). Default policy: refuse. const fpCheck = checkFingerprint(); @@ -393,6 +519,12 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { brokers, meshConfigs, onPendingInserted: () => drain?.wake(), + // 1.37.1: per-session WS clients for `/v1/diagnostics` (the ghost + // detector in `claudemesh doctor` reads their status). + sessionBrokers, + sessionBrokersByPubkey, + logSink, + startedAt, }); try { @@ -420,6 +552,7 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { const shutdown = async (sig: string) => { if (shuttingDown) return; shuttingDown = true; + clearInterval(pidGuard); process.stdout.write(JSON.stringify({ msg: "daemon_shutdown", signal: sig, ts: new Date().toISOString() }) + "\n"); inboxPruner.stop(); if (drain) await drain.close(); @@ -430,6 +563,8 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { try { await b.close(); } catch { /* ignore */ } } sessionBrokers.clear(); + // ipc.close() ends every open SSE response (MCP plugins see a clean + // `end` and resubscribe to the next daemon) before the listeners go. await ipc.close(); try { outboxDb.close(); } catch { /* ignore */ } try { inboxDb.close(); } catch { /* ignore */ } @@ -439,6 +574,9 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { process.on("SIGINT", () => shutdown("SIGINT")); process.on("SIGTERM", () => shutdown("SIGTERM")); + // 1.37.1: SIGHUP (terminal closed, launchd reload) is a clean shutdown + // too — previously it was Node's default (exit 129, no log line). + process.on("SIGHUP", () => shutdown("SIGHUP")); // Hold the event loop open until a signal arrives. return new Promise(() => { /* never resolves; signals call process.exit */ }); diff --git a/apps/cli/src/daemon/service-install.ts b/apps/cli/src/daemon/service-install.ts index dd23548..cb9ca1b 100644 --- a/apps/cli/src/daemon/service-install.ts +++ b/apps/cli/src/daemon/service-install.ts @@ -2,8 +2,14 @@ // - macOS: ~/Library/LaunchAgents/com.claudemesh.daemon.plist (launchctl bootstrap) // - Linux: ~/.config/systemd/user/claudemesh-daemon.service (systemctl --user enable) // -// Both run as the invoking user, redirect stdout/stderr to ~/.claudemesh/ -// daemon/daemon.log, restart on crash, and start at login. CI envs are +// Both run as the invoking user, restart on crash, and start at login. +// 1.37.1: stdout/stderr are redirected to ~/.claudemesh/daemon/ +// daemon.launchd.log — a small file that only receives writes from +// BEFORE the daemon installs its in-process rotating sink (boot +// crashes, node startup errors). The daemon's own log lines go to +// daemon.log via `installRotatingLogSink` (20 MB × 5). Pre-1.37.1 units +// pointed StandardOutPath at daemon.log directly, and launchd never +// rotates — that is how the 2026-09-08 incident found a 1 GB log. CI envs are // refused unless --allow-ci-persistent is passed (spec §9 / §16.3). import { existsSync, mkdirSync, writeFileSync, unlinkSync, readFileSync } from "node:fs"; @@ -23,6 +29,12 @@ export interface InstallResult { const SERVICE_LABEL = "com.claudemesh.daemon"; const SYSTEMD_UNIT = "claudemesh-daemon.service"; +/** 1.37.1: minimum seconds launchd waits between respawns. The + * 2026-09-08 incident had two daemons (one lived 6 s) racing the broker + * inside the same 90 s lease-grace window, doubling the reconnect + * storm. launchd's default is already 10 s; pinning it makes the + * contract explicit and lets `doctor` detect pre-1.37.1 units. */ +const LAUNCHD_THROTTLE_SECONDS = 10; export function detectPlatform(): ServicePlatform | null { if (process.platform === "darwin") return "darwin"; @@ -91,7 +103,7 @@ function darwinPlistPath(): string { function installDarwin(args: InstallArgs): InstallResult { const plist = darwinPlistPath(); mkdirSync(dirname(plist), { recursive: true }); - const log = DAEMON_PATHS.LOG_FILE; + const log = DAEMON_PATHS.LAUNCHD_LOG_FILE; // Resolve `node` explicitly. The bin script in node_modules/.bin starts // with `#!/usr/bin/env node`; under launchd's restricted PATH that would // resolve `node` to a system Node (often the wrong major) instead of the @@ -108,6 +120,8 @@ function installDarwin(args: InstallArgs): InstallResult { "daemon", "up", "--foreground", + // 1.37.1: in-process rotating log sink (launchd never rotates). + "--log-sink", ...(args.meshSlug ? ["--mesh", `${escapeXml(args.meshSlug)}`] : []), @@ -129,6 +143,8 @@ function installDarwin(args: InstallArgs): InstallResult { KeepAlive + ThrottleInterval + ${LAUNCHD_THROTTLE_SECONDS} StandardOutPath ${escapeXml(log)} StandardErrorPath @@ -190,7 +206,7 @@ function installLinux(args: InstallArgs): InstallResult { // and stdio capture; we don't want the child to double-fork into a // detached grandchild systemd can't track. const execArgs = [ - "daemon", "up", "--foreground", + "daemon", "up", "--foreground", "--log-sink", ...(args.meshSlug ? ["--mesh", args.meshSlug] : []), ...(args.displayName ? ["--name", args.displayName] : []), ].map(shellQuote).join(" "); @@ -205,8 +221,8 @@ Type=simple ExecStart=${shellQuote(nodeBin)} ${shellQuote(args.binaryPath)} ${execArgs} Restart=always RestartSec=3 -StandardOutput=append:${DAEMON_PATHS.LOG_FILE} -StandardError=append:${DAEMON_PATHS.LOG_FILE} +StandardOutput=append:${DAEMON_PATHS.LAUNCHD_LOG_FILE} +StandardError=append:${DAEMON_PATHS.LAUNCHD_LOG_FILE} Environment=PATH=/usr/local/bin:/usr/bin:/bin [Install] @@ -248,6 +264,38 @@ function shellQuote(s: string): string { return "'" + s.replace(/'/g, "'\"'\"'") + "'"; } +/** + * 1.37.1: is the installed unit from before the rotating-log-sink change? + * Stale when it still redirects stdio straight into `daemon.log` (no + * rotation → unbounded growth) or lacks the launchd ThrottleInterval. + * `claudemesh doctor` surfaces this with the reinstall command. + */ +export function isInstalledUnitStale(): { stale: boolean; reason?: string; path?: string } { + const unit = readInstalledUnit(); + if (!unit.platform || !unit.path || unit.content === null) return { stale: false }; + const c = unit.content; + const legacyLog = DAEMON_PATHS.LOG_FILE; + if (unit.platform === "darwin") { + if (c.includes(`${escapeXml(legacyLog)}`)) { + return { stale: true, reason: "StandardOutPath/StandardErrorPath point at daemon.log (no rotation)", path: unit.path }; + } + if (!c.includes("ThrottleInterval")) { + return { stale: true, reason: "missing ThrottleInterval", path: unit.path }; + } + if (!c.includes("--log-sink")) { + return { stale: true, reason: "missing --log-sink (daemon.log would not rotate)", path: unit.path }; + } + return { stale: false, path: unit.path }; + } + if (c.includes(`append:${legacyLog}\n`) || c.trimEnd().endsWith(`append:${legacyLog}`)) { + return { stale: true, reason: "StandardOutput/StandardError append to daemon.log (no rotation)", path: unit.path }; + } + if (!c.includes("--log-sink")) { + return { stale: true, reason: "missing --log-sink (daemon.log would not rotate)", path: unit.path }; + } + return { stale: false, path: unit.path }; +} + /** Diagnostic helper: dump current install status for `claudemesh daemon status --json`. */ export function readInstalledUnit(): { platform: ServicePlatform | null; path: string | null; content: string | null } { const platform = detectPlatform(); diff --git a/apps/cli/src/daemon/session-broker.ts b/apps/cli/src/daemon/session-broker.ts index ed1158a..75b617a 100644 --- a/apps/cli/src/daemon/session-broker.ts +++ b/apps/cli/src/daemon/session-broker.ts @@ -44,7 +44,7 @@ import { hostname as osHostname } from "node:os"; import type { JoinedMesh } from "~/services/config/facade.js"; import { signSessionHello, signParentAttestation } from "~/services/broker/session-hello-sig.js"; -import { connectWsWithBackoff, type WsLifecycle, type WsStatus } from "./ws-lifecycle.js"; +import { createWsLifecycle, type WsLifecycle, type WsStatus } from "./ws-lifecycle.js"; import type { BrokerSendArgs, BrokerSendResult } from "./broker.js"; export type SessionBrokerStatus = WsStatus; @@ -103,6 +103,8 @@ export interface SessionBrokerOptions { */ onPush?: (msg: Record) => void; log?: (level: "info" | "warn" | "error", msg: string, meta?: Record) => void; + /** Lifecycle timing overrides (tests / tuning). Defaults live in ws-lifecycle.ts. */ + lifecycle?: { helloAckTimeoutMs?: number; backoffCapsMs?: readonly number[] }; } export class SessionBrokerClient { @@ -135,12 +137,19 @@ export class SessionBrokerClient { }); }; - /** Open the WS, run session_hello, resolve once the broker accepts. */ + /** + * Open the WS, run session_hello, resolve once the broker accepts. + * + * 1.37.1: handle stored synchronously, then `ready` awaited — so a + * first-attempt hello-ack timeout (broker slow under a reconnect + * storm) can never orphan a live reconnecting socket, and `close()` + * called mid-connect always reaches the socket. Idempotent. + */ async connect(): Promise { if (this.closed) throw new Error("client_closed"); - if (this._status === "connecting" || this._status === "open") return; + if (this.lifecycle) return this.lifecycle.ready; - this.lifecycle = await connectWsWithBackoff({ + this.lifecycle = createWsLifecycle({ url: this.opts.mesh.brokerUrl, buildHello: async () => { const { timestamp, signature } = await signSessionHello({ @@ -275,7 +284,20 @@ export class SessionBrokerClient { } }, log: (level, msg, meta) => this.log(level, `session_broker_${msg}`, meta), + ...(this.opts.lifecycle ?? {}), }); + await this.lifecycle.ready; + } + + /** Operator diagnostics — surfaced by `/v1/diagnostics` + `claudemesh doctor`. */ + diagnostics(): { status: SessionBrokerStatus; isOpen: boolean; lastAckAt: number | null; reconnects: number; replaced: boolean } { + return { + status: this._status, + isOpen: this.isOpen(), + lastAckAt: this.lifecycle?.lastAckAt ?? null, + reconnects: this.lifecycle?.reconnects ?? 0, + replaced: this.lifecycle?.replaced ?? false, + }; } /** v2 agentic-comms (M1): send `client_ack` back to the broker after diff --git a/apps/cli/src/daemon/ws-lifecycle.ts b/apps/cli/src/daemon/ws-lifecycle.ts index db87c12..7b0b53e 100644 --- a/apps/cli/src/daemon/ws-lifecycle.ts +++ b/apps/cli/src/daemon/ws-lifecycle.ts @@ -13,19 +13,34 @@ * - new WebSocket(url), wire up open/message/close/error * - on open → call buildHello() and send the result * - start an ack-timeout timer; if it fires before the hello ack - * arrives, close the socket and reject the connect promise + * arrives, close the socket — the close handler then schedules a + * backoff reconnect exactly like any other disconnect * - on message, gate on isHelloAck(); when true, flip status to - * "open", clear the ack timer, resolve. All other messages are - * forwarded to onMessage() + * "open", clear the ack timer, resolve `ready` (first time only). + * All other messages are forwarded to onMessage() * - on close, schedule a backoff reconnect (unless explicitly closed) * + * 1.37.1 — the handle is returned SYNCHRONOUSLY (`createWsLifecycle`), + * with readiness exposed separately as `handle.ready`. Before this, the + * helper returned `Promise` that resolved on the FIRST ack + * and REJECTED on a first-attempt hello-ack timeout — while its internal + * close handler kept reconnecting the same socket under the caller's + * feet. The caller (`DaemonBrokerClient.connect`) then never received + * the handle: `lifecycle === null` forever while `onStatusChange` + * flipped `_status` to "open" on the silent reconnect. Every RPC gated + * on `this.lifecycle` (listPeers, send, …) short-circuited → `peer list` + * returned [] on every mesh for as long as the daemon lived + * (2026-09-08 incident, spec addendum 3). With a synchronous handle the + * caller always owns the socket it started, and a hello-ack timeout is + * just another reconnect — never an orphan. + * * Each client keeps its own concerns: DaemonBrokerClient still owns * pendingAcks / peerListResolvers / etc; SessionBrokerClient still owns * its onPush callback. The helper just hands them an open WS and a * stable status field, and reconnects under their feet on disconnect. * * Composition over inheritance — callers receive a `WsLifecycle` handle - * with `send` / `close` / `status`, NOT a subclass. + * with `send` / `close` / `status` / `ready`, NOT a subclass. */ import WebSocket from "ws"; @@ -61,8 +76,12 @@ export interface WsLifecycleOptions { onMessage: (msg: Record) => void; onStatusChange?: (s: WsStatus) => void; /** - * How long to wait for the broker's hello ack before giving up and - * forcing a close. Defaults 5s — same as both pre-refactor clients. + * How long to wait for the broker's hello ack before giving up on THIS + * attempt and forcing a close (→ backoff reconnect). Defaults 15s. + * Was 5s until 1.37.1; a broker under a reconnect storm took 4–5s per + * hello (2026-09-08), which tripped the old timeout on healthy sockets. + * Because a timeout is now just a reconnect, a generous value costs + * nothing on the happy path and avoids churning a slow-but-alive broker. */ helloAckTimeoutMs?: number; /** @@ -88,15 +107,29 @@ export interface WsLifecycle { /** Underlying socket. Exposed for callers that need OPEN-state checks * before sending (mirrors the pre-refactor `this.ws.readyState` checks). */ readonly ws: WebSocket | null; + /** + * Resolves once the broker has accepted a hello for the first time + * (any attempt — a first-attempt ack timeout does NOT reject). Rejects + * only if `close()` is called before any ack ever arrived. + */ + readonly ready: Promise; + /** Unix ms of the most recent hello ack, null before the first. Diagnostics. */ + readonly lastAckAt: number | null; + /** Number of reconnect attempts scheduled since creation. Diagnostics. */ + readonly reconnects: number; + /** True once the broker closed us with `1000 session_replaced` — another + * socket owns this identity now and this lifecycle will not reconnect. */ + readonly replaced: boolean; /** Send a JSON payload over the open WS. Throws if not open — callers * that need queue-while-disconnected semantics should layer that * themselves (DaemonBrokerClient does, via its `opens` deferred-fn array). */ send(payload: unknown): void; - /** Close the WS and stop reconnecting. Idempotent. */ + /** Close the WS and stop reconnecting. Idempotent. Effective at any + * point in the lifecycle, including mid-connect before the first ack. */ close(): Promise; } -const DEFAULT_HELLO_ACK_TIMEOUT_MS = 5_000; +export const DEFAULT_HELLO_ACK_TIMEOUT_MS = 15_000; const DEFAULT_BACKOFF_CAPS_MS: readonly number[] = [1_000, 2_000, 4_000, 8_000, 16_000, 30_000]; const defaultLog: WsLog = (level, msg, meta) => { @@ -106,15 +139,16 @@ const defaultLog: WsLog = (level, msg, meta) => { }; /** - * Connect a WebSocket with hello-handshake, ack-timeout, and reconnect - * with exponential backoff. Resolves once the broker accepts the hello; - * rejects if the first connect closes before the ack lands. + * Create a WebSocket lifecycle with hello-handshake, ack-timeout, and + * reconnect with exponential backoff. Returns the handle synchronously; + * the first connection attempt starts immediately. Await `handle.ready` + * to know when the broker has accepted a hello. * - * Subsequent automatic reconnects are silent — they fire on the close - * handler's backoff timer and surface only via onStatusChange (and any - * caller-installed log). + * Reconnects (after a close, or after a hello-ack timeout) are silent — + * they fire on the close handler's backoff timer and surface only via + * onStatusChange (and any caller-installed log). */ -export function connectWsWithBackoff(opts: WsLifecycleOptions): Promise { +export function createWsLifecycle(opts: WsLifecycleOptions): WsLifecycle { const helloAckTimeoutMs = opts.helloAckTimeoutMs ?? DEFAULT_HELLO_ACK_TIMEOUT_MS; const backoffCapsMs = opts.backoffCapsMs ?? DEFAULT_BACKOFF_CAPS_MS; const log: WsLog = opts.log ?? defaultLog; @@ -123,8 +157,22 @@ export function connectWsWithBackoff(opts: WsLifecycleOptions): Promise void; + let readyReject!: (err: Error) => void; + let readySettled = false; + const ready = new Promise((resolve, reject) => { + readyResolve = resolve; + readyReject = reject; + }); + // A rejected `ready` nobody awaits must not surface as an unhandled + // rejection (callers that fire-and-forget connect() are legitimate). + ready.catch(() => { /* observed via status */ }); const setStatus = (s: WsStatus) => { if (status === s) return; @@ -132,13 +180,6 @@ export function connectWsWithBackoff(opts: WsLifecycleOptions): Promise75s (2.5x the ping @@ -158,8 +199,14 @@ export function connectWsWithBackoff(opts: WsLifecycleOptions): Promise => { - if (closed) return Promise.reject(new Error("client_closed")); + const clearTimers = () => { + if (helloTimer) { clearTimeout(helloTimer); helloTimer = null; } + if (watchdogTimer) { clearInterval(watchdogTimer); watchdogTimer = null; } + }; + + /** Open one WS attempt. Every attempt — initial or reconnect — goes through here. */ + const openOnce = (): void => { + if (closed) return; setStatus("connecting"); log("info", "ws_open_attempt", { url: opts.url }); @@ -167,108 +214,148 @@ export function connectWsWithBackoff(opts: WsLifecycleOptions): Promise((resolve, reject) => { - sock.on("open", () => { - log("info", "ws_open_ok", { url: opts.url }); - // Build and send the hello inside a microtask so any sync - // throws from buildHello() reject this connect attempt cleanly. - (async () => { - try { - const hello = await opts.buildHello(); - sock.send(JSON.stringify(hello)); - log("info", "ws_hello_sent", { url: opts.url }); - helloTimer = setTimeout(() => { - log("warn", "hello_ack_timeout", { url: opts.url }); - try { sock.close(); } catch { /* ignore */ } - reject(new Error("hello_ack_timeout")); - }, helloAckTimeoutMs); - } catch (e) { - log("warn", "ws_build_hello_threw", { err: String(e) }); - reject(e instanceof Error ? e : new Error(String(e))); - } - })(); - }); - - sock.on("message", (raw) => { - lastActivity = Date.now(); - let msg: Record; - try { msg = JSON.parse(raw.toString()) as Record; } - catch { return; } - - if (opts.isHelloAck(msg)) { - if (helloTimer) { clearTimeout(helloTimer); helloTimer = null; } - setStatus("open"); - reconnectAttempt = 0; - log("info", "ws_hello_acked", { url: opts.url }); - // Start liveness watchdog only after a successful handshake. - if (watchdogTimer) clearInterval(watchdogTimer); - watchdogTimer = setInterval(() => { - if (sock.readyState !== sock.OPEN) return; - const idle = Date.now() - lastActivity; - if (idle > STALE_THRESHOLD_MS) { - log("warn", "ws_stale_terminate", { url: opts.url, idle_ms: idle }); - try { sock.terminate(); } catch { /* socket already gone */ } - return; - } - try { sock.ping(); } catch { /* ignore */ } - }, PING_INTERVAL_MS); - resolve(); - return; + sock.on("open", () => { + if (closed) { try { sock.close(); } catch { /* ignore */ } return; } + log("info", "ws_open_ok", { url: opts.url }); + // Build and send the hello inside a microtask so any sync throws + // from buildHello() are caught and turned into a reconnect. + (async () => { + try { + const hello = await opts.buildHello(); + if (closed || ws !== sock) return; + sock.send(JSON.stringify(hello)); + log("info", "ws_hello_sent", { url: opts.url }); + helloTimer = setTimeout(() => { + helloTimer = null; + // Not a failure of the lifecycle — just this attempt. Closing + // the socket routes through the close handler, which schedules + // the backoff reconnect. `ready` stays pending until an ack + // lands on some later attempt. + log("warn", "hello_ack_timeout", { url: opts.url, timeout_ms: helloAckTimeoutMs }); + // terminate(), not close(): a broker that never acked our hello + // may also never ack a close frame, and `ws` would then wait its + // 30 s closeTimeout before emitting `close` (→ reconnect). A hard + // teardown makes the retry start on the backoff schedule instead. + try { sock.terminate(); } catch { /* ignore */ } + }, helloAckTimeoutMs); + } catch (e) { + log("warn", "ws_build_hello_threw", { err: String(e) }); + try { sock.close(); } catch { /* ignore */ } } - - opts.onMessage(msg); - }); - - sock.on("ping", () => { lastActivity = Date.now(); }); - sock.on("pong", () => { lastActivity = Date.now(); }); - - sock.on("close", (code, reason) => { - if (helloTimer) { clearTimeout(helloTimer); helloTimer = null; } - if (watchdogTimer) { clearInterval(watchdogTimer); watchdogTimer = null; } - const reasonStr = reason.toString("utf8"); - log("warn", "ws_closed", { url: opts.url, code, reason: reasonStr, status }); - opts.onBeforeReconnect?.(code, reasonStr); - - if (closed) { - setStatus("closed"); - return; - } - setStatus("reconnecting"); - const wait = backoffCapsMs[Math.min(reconnectAttempt, backoffCapsMs.length - 1)] ?? 30_000; - reconnectAttempt++; - log("info", "ws_reconnect_scheduled", { url: opts.url, wait_ms: wait, code, reason: reasonStr }); - reconnectTimer = setTimeout( - () => openOnce().catch((err) => log("warn", "ws_reconnect_failed", { url: opts.url, err: String(err) })), - wait, - ); - if (status === "connecting" || status === "reconnecting") { - reject(new Error(`closed_before_hello_${code}`)); - } - }); - - sock.on("error", (err) => log("warn", "ws_error", { url: opts.url, err: err.message })); + })(); }); + + sock.on("message", (raw) => { + if (ws !== sock) return; // stale socket after a swap — ignore + lastActivity = Date.now(); + let msg: Record; + try { msg = JSON.parse(raw.toString()) as Record; } + catch { return; } + + if (opts.isHelloAck(msg)) { + if (helloTimer) { clearTimeout(helloTimer); helloTimer = null; } + lastAckAt = Date.now(); + setStatus("open"); + reconnectAttempt = 0; + log("info", "ws_hello_acked", { url: opts.url }); + // Start liveness watchdog only after a successful handshake. + if (watchdogTimer) clearInterval(watchdogTimer); + watchdogTimer = setInterval(() => { + if (sock.readyState !== sock.OPEN) return; + const idle = Date.now() - lastActivity; + if (idle > STALE_THRESHOLD_MS) { + log("warn", "ws_stale_terminate", { url: opts.url, idle_ms: idle }); + try { sock.terminate(); } catch { /* socket already gone */ } + return; + } + try { sock.ping(); } catch { /* ignore */ } + }, PING_INTERVAL_MS); + if (!readySettled) { readySettled = true; readyResolve(); } + return; + } + + opts.onMessage(msg); + }); + + sock.on("ping", () => { lastActivity = Date.now(); }); + sock.on("pong", () => { lastActivity = Date.now(); }); + + sock.on("close", (code, reason) => { + if (ws !== sock) return; // an older socket closing after a swap — nothing to do + clearTimers(); + const reasonStr = reason.toString("utf8"); + log("warn", "ws_closed", { url: opts.url, code, reason: reasonStr, status }); + opts.onBeforeReconnect?.(code, reasonStr); + + if (closed) { + setStatus("closed"); + return; + } + // 1.37.1: `1000 session_replaced` means the broker attached this + // identity (member or session pubkey) to ANOTHER live socket that + // hello'd after us. Reconnecting would replace it right back — two + // daemons holding the same sessions did exactly that at a 1 s + // cadence in the 2026-09-08 drill (each hello kicked the other's + // socket, forever). Our own reconnects never trigger this: the + // socket the broker kicks is the one this lifecycle already left + // (`ws !== sock` above). So treat it as terminal for this lifecycle: + // whoever replaced us owns the identity now. + if (code === 1000 && reasonStr === "session_replaced") { + replaced = true; + closed = true; + log("warn", "ws_replaced_terminal", { url: opts.url }); + setStatus("closed"); + if (!readySettled) { readySettled = true; readyReject(new Error("session_replaced")); } + return; + } + setStatus("reconnecting"); + const wait = backoffCapsMs[Math.min(reconnectAttempt, backoffCapsMs.length - 1)] ?? 30_000; + reconnectAttempt++; + reconnects++; + log("info", "ws_reconnect_scheduled", { url: opts.url, wait_ms: wait, code, reason: reasonStr }); + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + openOnce(); + }, wait); + }); + + sock.on("error", (err) => log("warn", "ws_error", { url: opts.url, err: err.message })); }; - return openOnce().then(() => { - const handle: WsLifecycle = { - get status() { return status; }, - get ws() { return ws; }, - send(payload: unknown) { - if (!ws || ws.readyState !== ws.OPEN) { - throw new Error("ws_not_open"); - } - ws.send(JSON.stringify(payload)); - }, - async close() { - closed = true; - if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } - if (helloTimer) { clearTimeout(helloTimer); helloTimer = null; } - if (watchdogTimer) { clearInterval(watchdogTimer); watchdogTimer = null; } - try { ws?.close(); } catch { /* ignore */ } - setStatus("closed"); - }, - }; - return handle; - }); + const handle: WsLifecycle = { + get status() { return status; }, + get ws() { return ws; }, + ready, + get lastAckAt() { return lastAckAt; }, + get reconnects() { return reconnects; }, + get replaced() { return replaced; }, + send(payload: unknown) { + if (!ws || ws.readyState !== ws.OPEN) { + throw new Error("ws_not_open"); + } + ws.send(JSON.stringify(payload)); + }, + async close() { + closed = true; + if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } + clearTimers(); + const sock = ws; + if (sock) { + // A socket still in CONNECTING can't be close()d cleanly; terminate + // tears the TCP attempt down without waiting for the handshake. + try { + if (sock.readyState === sock.CONNECTING) sock.terminate(); + else sock.close(); + } catch { /* ignore */ } + } + setStatus("closed"); + if (!readySettled) { + readySettled = true; + readyReject(new Error("client_closed")); + } + }, + }; + + openOnce(); + return handle; } diff --git a/apps/cli/src/entrypoints/cli.ts b/apps/cli/src/entrypoints/cli.ts index 3b50065..cc04afd 100644 --- a/apps/cli/src/entrypoints/cli.ts +++ b/apps/cli/src/entrypoints/cli.ts @@ -504,6 +504,10 @@ async function main(): Promise { // path always run with --foreground so their parents (launchd / // the launch helper) own lifecycle and stdio redirection. foreground: !!flags.foreground, + // 1.37.1: explicit opt-in to the in-process rotating log sink. + // Service units pass it; a human running `up --foreground` keeps + // stdio (even when piped). + logSink: !!flags["log-sink"], outboxStatus, newClientId: flags["new-client-id"] as string | undefined, }, rest); diff --git a/apps/cli/src/mcp/server.ts b/apps/cli/src/mcp/server.ts index b731ea4..13aac6f 100644 --- a/apps/cli/src/mcp/server.ts +++ b/apps/cli/src/mcp/server.ts @@ -123,32 +123,117 @@ function daemonMarkSeen(ids: string[], sessionToken?: string | null): Promise } -function subscribeEvents(onEvent: (e: DaemonEvent) => void, opts: { sessionToken?: string | null } = {}): { close: () => void } { +export interface SubscribeEventsOptions { + sessionToken?: string | null; + /** + * 1.37.1: idle watchdog. The daemon writes `: keepalive` every 15 s on + * every SSE stream; if NOTHING arrives for this long the socket is + * presumed half-dead (daemon killed with SIGKILL, machine slept, …) + * and is destroyed so the reconnect path runs. Default 45 s (3 missed + * keepalives). Tests lower it. + */ + idleTimeoutMs?: number; + /** Reconnect backoff caps (ms). Default [1s, 2s, 5s, 10s]. */ + backoffMs?: readonly number[]; + /** Structured log sink (`mcpLog` in startMcpServer). Optional. */ + log?: (msg: string, meta?: Record) => void; + /** Override the daemon UDS path (tests). Default DAEMON_PATHS.SOCK_FILE. */ + socketPath?: string; +} + +const SSE_DEFAULT_IDLE_MS = 45_000; +const SSE_DEFAULT_BACKOFF_MS: readonly number[] = [1_000, 2_000, 5_000, 10_000]; + +/** + * Subscribe to the daemon's `/v1/events` SSE stream and hand every parsed + * event to `onEvent`. Reconnects forever (with backoff) until `close()`. + * + * 1.37.1 rewrite — the pre-1.37.1 version only resubscribed on + * `res 'end'` / `req 'error'`. A daemon that dies hard (SIGKILL, OOM, + * `process.exit` from a crash handler) never sends `end`: the response + * emits `aborted`/`close`/`error` instead, which were logged to stderr + * and otherwise ignored. After the 2026-09-08 daemon restart, `lsof` + * showed 0 of 10 live MCP plugins connected to the new daemon and 0 + * `sse_event_received` for the rest of the day — mid-turn `` + * injection was dead until each Claude Code session was restarted. + * + * Now every termination signal (`res end/close/aborted/error`, + * `req close/error/timeout`) funnels into ONE `scheduleReconnect(reason)` + * per attempt (deduped — a socket teardown fires several of these), and + * an idle watchdog catches the case where nothing fires at all. + */ +export function subscribeEvents( + onEvent: (e: DaemonEvent) => void, + opts: SubscribeEventsOptions = {}, +): { close: () => void } { + const idleTimeoutMs = opts.idleTimeoutMs ?? SSE_DEFAULT_IDLE_MS; + const backoff = opts.backoffMs ?? SSE_DEFAULT_BACKOFF_MS; + const log = opts.log ?? (() => { /* silent */ }); + const socketPath = opts.socketPath ?? DAEMON_PATHS.SOCK_FILE; + let active = true; let req: ReturnType | null = null; + let reconnectTimer: NodeJS.Timeout | null = null; + let attempt = 0; // consecutive failed/aborted attempts (resets on a live stream) + let generation = 0; // bumps per connect(); stale socket callbacks are ignored const connect = (): void => { if (!active) return; - // 1.34.13: forward the session token on the SSE subscription so the - // daemon's `/v1/events` route can scope the stream to this session - // via the SseFilterOptions demux added in 1.34.10. Without this - // header, `session` resolves to null in the IPC handler, the filter - // is empty, and every MCP receives every event — manifests as - // session A rendering DMs that arrived on B's session-WS. The - // launch helper sets CLAUDEMESH_IPC_TOKEN_FILE in the child env; - // readSessionTokenFromEnv() picks it up at MCP boot time. + const gen = ++generation; + let scheduled = false; + let idleTimer: NodeJS.Timeout | null = null; + let sawBytes = false; + + const clearIdle = () => { if (idleTimer) { clearTimeout(idleTimer); idleTimer = null; } }; + const armIdle = () => { + clearIdle(); + if (idleTimeoutMs <= 0) return; + idleTimer = setTimeout(() => { + if (gen !== generation || !active) return; + log("sse_idle_timeout", { idle_ms: idleTimeoutMs }); + process.stderr.write(`[claudemesh-mcp] sse idle for ${idleTimeoutMs}ms; reconnecting\n`); + try { thisReq.destroy(new Error("sse_idle_timeout")); } catch { /* ignore */ } + scheduleReconnect("idle_timeout"); + }, idleTimeoutMs); + }; + + /** One reconnect per attempt, whatever fires first. */ + const scheduleReconnect = (reason: string): void => { + if (gen !== generation) return; // an older socket's late event + if (scheduled) return; + scheduled = true; + clearIdle(); + if (!active) return; + const wait = backoff[Math.min(attempt, backoff.length - 1)] ?? 10_000; + attempt++; + log("sse_reconnect", { reason, attempt, wait_ms: wait }); + process.stderr.write(`[claudemesh-mcp] sse ${reason}; reconnecting in ${wait}ms (attempt ${attempt})\n`); + if (reconnectTimer) clearTimeout(reconnectTimer); + reconnectTimer = setTimeout(() => { reconnectTimer = null; connect(); }, wait); + }; + + // 1.34.13: forward the session token so the daemon scopes the stream + // to this session (SseFilterOptions demux). Without it every MCP + // receives every event. const headers: Record = { Accept: "text/event-stream" }; if (opts.sessionToken) headers.Authorization = `ClaudeMesh-Session ${opts.sessionToken}`; - req = httpRequest({ - socketPath: DAEMON_PATHS.SOCK_FILE, - path: "/v1/events", - method: "GET", - headers, - }); + const thisReq = httpRequest({ socketPath, path: "/v1/events", method: "GET", headers }); + req = thisReq; let buffer = ""; - req.on("response", (res: IncomingMessage) => { + + thisReq.on("response", (res: IncomingMessage) => { + if (res.statusCode && res.statusCode >= 400) { + log("sse_subscribe_rejected", { status: res.statusCode }); + res.resume(); + scheduleReconnect(`http_${res.statusCode}`); + return; + } + log("sse_subscribed", { attempt }); + armIdle(); res.setEncoding("utf8"); res.on("data", (chunk: string) => { + if (!sawBytes) { sawBytes = true; attempt = 0; } // live stream → reset backoff + armIdle(); buffer += chunk; let idx; while ((idx = buffer.indexOf("\n\n")) >= 0) { @@ -169,24 +254,25 @@ function subscribeEvents(onEvent: (e: DaemonEvent) => void, opts: { sessionToken } catch { /* malformed event; skip */ } } }); - res.on("end", () => { - if (active) { - process.stderr.write("[claudemesh-mcp] sse stream ended; reconnecting in 1s\n"); - setTimeout(connect, 1_000); - } - }); - res.on("error", (err) => process.stderr.write(`[claudemesh-mcp] sse error: ${err.message}\n`)); + res.on("end", () => scheduleReconnect("stream_ended")); + res.on("close", () => scheduleReconnect("stream_closed")); + res.on("aborted", () => scheduleReconnect("stream_aborted")); + res.on("error", (err) => scheduleReconnect(`stream_error:${err.message}`)); }); - req.on("error", (err) => { - process.stderr.write(`[claudemesh-mcp] sse connect error: ${err.message}\n`); - if (active) setTimeout(connect, 2_000); - }); - req.end(); + thisReq.on("error", (err) => scheduleReconnect(`connect_error:${err.message}`)); + thisReq.on("close", () => scheduleReconnect("request_closed")); + thisReq.on("timeout", () => { try { thisReq.destroy(new Error("sse_connect_timeout")); } catch { /* ignore */ } }); + thisReq.end(); }; connect(); return { - close: () => { active = false; try { req?.destroy(); } catch { /* ignore */ } }, + close: () => { + active = false; + generation++; + if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } + try { req?.destroy(); } catch { /* ignore */ } + }, }; } @@ -445,7 +531,7 @@ export async function startMcpServer(): Promise { }); } catch { /* best effort */ } } - }, { sessionToken: sessionTokenForSeen }); + }, { sessionToken: sessionTokenForSeen, log: mcpLog }); // 1.34.6 — Welcome: single emit on oninitialized + 3s grace. // diff --git a/apps/cli/src/services/bridge/daemon-route.ts b/apps/cli/src/services/bridge/daemon-route.ts index bcc1711..626e9c3 100644 --- a/apps/cli/src/services/bridge/daemon-route.ts +++ b/apps/cli/src/services/bridge/daemon-route.ts @@ -36,20 +36,40 @@ async function daemonReachable(): Promise { return res.state === "up" || res.state === "started"; } +export interface DaemonPeersResult { + peers: unknown[]; + /** 1.37.1: per-mesh member-WS status as reported by the daemon + * (`open` | `connecting` | `reconnecting` | `closed`). Missing on + * pre-1.37.1 daemons. Lets callers tell "no peers" from "daemon not + * connected to the broker" — the latter used to render as a silent + * empty list (2026-09-08 blackout). */ + brokers: Record; +} + +/** 1.37.1: peer list + broker status through the daemon. Returns null + * when the daemon is unreachable so the caller can fall back. */ +export async function tryListPeersViaDaemonDetailed(mesh?: string): Promise { + if (!(await daemonReachable())) return null; + try { + const res = await ipc<{ peers?: unknown[]; brokers?: Record }>({ + path: `/v1/peers${meshQuery(mesh)}`, timeoutMs: 3_000, + }); + if (res.status !== 200) return null; + return { + peers: Array.isArray(res.body.peers) ? res.body.peers : [], + brokers: res.body.brokers && typeof res.body.brokers === "object" ? res.body.brokers : {}, + }; + } catch { + return null; + } +} + /** Try fetching the peer list through the daemon (~1ms warm IPC). * Returns null when the daemon socket isn't present so the caller can * fall back to bridge / cold paths. */ export async function tryListPeersViaDaemon(mesh?: string): Promise { - if (!(await daemonReachable())) return null; - try { - const res = await ipc<{ peers?: unknown[] }>({ path: `/v1/peers${meshQuery(mesh)}`, timeoutMs: 3_000 }); - if (res.status !== 200) return null; - return Array.isArray(res.body.peers) ? res.body.peers : []; - } catch (err) { - const msg = String(err); - if (/ENOENT|ECONNREFUSED|ipc_timeout/.test(msg)) return null; - return null; - } + const r = await tryListPeersViaDaemonDetailed(mesh); + return r ? r.peers : null; } /** diff --git a/apps/cli/src/services/daemon/lifecycle.ts b/apps/cli/src/services/daemon/lifecycle.ts index 259df55..8802d18 100644 --- a/apps/cli/src/services/daemon/lifecycle.ts +++ b/apps/cli/src/services/daemon/lifecycle.ts @@ -34,6 +34,7 @@ import { homedir } from "node:os"; import { join } from "node:path"; import { ipc, IpcError } from "~/daemon/ipc/client.js"; +import { readRunningPid } from "~/daemon/lock.js"; import { DAEMON_PATHS } from "~/daemon/paths.js"; export type DaemonReadyState = @@ -105,6 +106,10 @@ async function runEnsureDaemon(opts: EnsureDaemonOpts): Promise { +/** + * 1.37.1: classify a failed probe. Exported for tests. + * + * Before this, EVERY probe failure — including a plain timeout against a + * daemon that was merely busy (rehydrating sessions, signing attestations, + * a slow broker hello) — was "stale", and `cleanupStaleFiles()` then + * unlinked the LIVE daemon's socket + pid file. The daemon kept running, + * unreachable; the next CLI call auto-spawned a second daemon; the two + * fought over the same sessions with `session_replaced` at a 1 s cadence + * (2026-09-08 drill). Only "nobody is listening" is stale: the socket + * path is gone or the connection is refused, or the pid on record is + * dead. Anything else from a live pid is "busy" — leave its files alone. + */ +export function classifyProbeFailure(errMessage: string, pidAlive: boolean): "stale" | "busy" { + if (/ENOENT|ECONNREFUSED/.test(errMessage)) return "stale"; + return pidAlive ? "busy" : "stale"; +} + +async function probeDaemon(): Promise<"up" | "absent" | "stale" | "busy"> { if (!existsSync(DAEMON_PATHS.SOCK_FILE)) return "absent"; try { const res = await ipc<{ version?: string }>({ path: "/v1/version", timeoutMs: PROBE_TIMEOUT_MS }); if (res.status === 200) return "up"; - return "stale"; + // Answered, but not 200 — a daemon IS listening. Never stale. + return "busy"; } catch (err) { - if (err instanceof IpcError) return "stale"; - const msg = String(err); - if (/ENOENT|ECONNREFUSED|ipc_timeout|EPIPE|ECONNRESET/.test(msg)) return "stale"; - return "stale"; + const msg = err instanceof IpcError ? `ipc_error_${err.status}` : String(err); + return classifyProbeFailure(msg, readRunningPid() !== null); } } diff --git a/apps/cli/tests/unit/daemon-broker-lost-handle.test.ts b/apps/cli/tests/unit/daemon-broker-lost-handle.test.ts new file mode 100644 index 0000000..3393370 --- /dev/null +++ b/apps/cli/tests/unit/daemon-broker-lost-handle.test.ts @@ -0,0 +1,110 @@ +/** + * DaemonBrokerClient — Addendum 3 regression (2026-09-08 mesh blackout). + * + * The broker, busy with a reconnect storm, took >5s to ack the daemon's + * member `hello`. The old `connect()` did `this.lifecycle = await + * connectWsWithBackoff(...)`; the timeout REJECTED that promise so + * `lifecycle` stayed null forever, while the helper's own reconnect flipped + * `_status` to "open". `listPeers()` then short-circuited on + * `!this.lifecycle` and returned [] in ~30ms on every mesh — `peer list` + * showed 0 peers for as long as the daemon lived. + * + * Now: first hello times out, second is acked, and `listPeers()` returns + * the broker's list. + */ + +import { afterEach, describe, expect, it } from "vitest"; +import { WebSocketServer, type WebSocket } from "ws"; +import type { AddressInfo } from "node:net"; +import sodium from "libsodium-wrappers"; + +import { DaemonBrokerClient } from "~/daemon/broker.js"; +import type { JoinedMesh } from "~/services/config/schemas.js"; + +const silent = () => { /* quiet */ }; + +describe("DaemonBrokerClient keeps its lifecycle handle across a first-ack timeout", () => { + let wss: WebSocketServer | null = null; + let client: DaemonBrokerClient | null = null; + const sockets: WebSocket[] = []; + + afterEach(async () => { + await client?.close(); + client = null; + for (const s of sockets) { try { s.terminate(); } catch { /* ignore */ } } + sockets.length = 0; + await new Promise((r) => (wss ? wss.close(() => r()) : r())); + wss = null; + }); + + it("listPeers() returns the broker's peers after the second hello is acked", async () => { + await sodium.ready; + const kp = sodium.crypto_sign_keypair(); + + let hellos = 0; + let listPeersSeen = 0; + wss = new WebSocketServer({ port: 0 }); + wss.on("connection", (sock) => { + sockets.push(sock); + sock.on("message", (raw) => { + const msg = JSON.parse(raw.toString()) as Record; + if (msg.type === "hello") { + hellos++; + if (hellos === 1) return; // simulate a broker stalled under load + sock.send(JSON.stringify({ type: "hello_ack", presenceId: "p1", memberDisplayName: "d" })); + return; + } + if (msg.type === "list_peers") { + listPeersSeen++; + sock.send(JSON.stringify({ + type: "peers_list", + _reqId: msg._reqId, + peers: [ + { pubkey: "a".repeat(64), displayName: "alpha", status: "idle", summary: null, groups: [], sessionId: "s1", connectedAt: new Date().toISOString(), peerRole: "session" }, + { pubkey: "b".repeat(64), displayName: "beta", status: "idle", summary: null, groups: [], sessionId: "s2", connectedAt: new Date().toISOString(), peerRole: "session" }, + ], + })); + } + }); + }); + await new Promise((r) => wss!.on("listening", () => r())); + const { port } = wss.address() as AddressInfo; + + const mesh: JoinedMesh = { + meshId: "mesh_test", + memberId: "member_test", + slug: "test", + name: "test", + pubkey: sodium.to_hex(kp.publicKey), + secretKey: sodium.to_hex(kp.privateKey), + brokerUrl: `ws://127.0.0.1:${port}`, + joinedAt: new Date().toISOString(), + }; + + const statuses: string[] = []; + client = new DaemonBrokerClient(mesh, { + onStatusChange: (s) => statuses.push(s), + log: silent, + // Real client-side hello-ack timeout path, shortened for the test. + lifecycle: { helloAckTimeoutMs: 150, backoffCapsMs: [50] }, + }); + + await client.connect(); // resolves once the SECOND hello is acked + + expect(statuses).toEqual(["connecting", "reconnecting", "connecting", "open"]); + + expect(hellos).toBe(2); + expect(client.status).toBe("open"); + expect(client.isOpen()).toBe(true); + + const peers = await client.listPeers(2_000); + expect(listPeersSeen).toBe(1); + expect(peers.map((p) => p.displayName)).toEqual(["alpha", "beta"]); + + const d = client.diagnostics(); + expect(d.status).toBe("open"); + expect(d.isOpen).toBe(true); + expect(d.reconnects).toBe(1); + expect(d.lastAckAt).not.toBeNull(); + }, 10_000); +}); diff --git a/apps/cli/tests/unit/daemon-probe-classify.test.ts b/apps/cli/tests/unit/daemon-probe-classify.test.ts new file mode 100644 index 0000000..55e65d3 --- /dev/null +++ b/apps/cli/tests/unit/daemon-probe-classify.test.ts @@ -0,0 +1,30 @@ +/** + * 1.37.1 — `classifyProbeFailure`: a slow-but-live daemon must never be + * classified "stale" (which unlinks its socket + pid file and lets the + * CLI auto-spawn a second daemon). Spec Addendum 6. + */ +import { describe, expect, test } from "vitest"; +import { classifyProbeFailure } from "~/services/daemon/lifecycle.js"; + +describe("classifyProbeFailure", () => { + test("connection refused → stale regardless of pid", () => { + expect(classifyProbeFailure("Error: connect ECONNREFUSED", true)).toBe("stale"); + expect(classifyProbeFailure("Error: connect ECONNREFUSED", false)).toBe("stale"); + }); + test("socket path vanished (ENOENT) → stale", () => { + expect(classifyProbeFailure("Error: connect ENOENT /x/daemon.sock", true)).toBe("stale"); + }); + test("timeout with a live pid → busy (files must be left alone)", () => { + expect(classifyProbeFailure("ipc_timeout", true)).toBe("busy"); + }); + test("timeout with a dead/missing pid → stale", () => { + expect(classifyProbeFailure("ipc_timeout", false)).toBe("stale"); + }); + test("non-200 answer from a live daemon → busy", () => { + expect(classifyProbeFailure("ipc_error_500", true)).toBe("busy"); + }); + test("EPIPE/ECONNRESET with live pid → busy, without → stale", () => { + expect(classifyProbeFailure("read ECONNRESET", true)).toBe("busy"); + expect(classifyProbeFailure("write EPIPE", false)).toBe("stale"); + }); +}); diff --git a/apps/cli/tests/unit/drain-expiry.test.ts b/apps/cli/tests/unit/drain-expiry.test.ts new file mode 100644 index 0000000..071fb4d --- /dev/null +++ b/apps/cli/tests/unit/drain-expiry.test.ts @@ -0,0 +1,139 @@ +/** + * Outbox drain hygiene (1.37.1, spec 2026-09-08-daemon-restart-mesh-blackout §8). + * + * Before: a session-attributed row whose session WS never came back was + * retried forever (~130k attempts/row observed), never counted against the + * attempt cap, never expired, and logged `drain_session_ws_not_ready` on + * EVERY tick — the 1 GB daemon.log. These tests pin the new caps. + */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { openSqlite, type SqliteDb } from "~/daemon/db/sqlite.js"; +import { insertPending, migrateOutbox, findById } from "~/daemon/db/outbox.js"; +import { + drainOnce, + MAX_ATTEMPTS_PER_ROW, + NOT_READY_LOG_EVERY, + OUTBOX_TTL_MS, + SESSION_GONE_TTL_MS, + type DrainOptions, +} from "~/daemon/drain.js"; +import type { DaemonBrokerClient } from "~/daemon/broker.js"; +import type { SessionBrokerClient } from "~/daemon/session-broker.js"; + +const SESSION_PK = "ab".repeat(32); + +function row(db: SqliteDb, id: string, now: number, extra: Partial[1]> = {}) { + insertPending(db, { + id, + client_message_id: `cid-${id}`, + request_fingerprint: new Uint8Array(32), + payload: new Uint8Array([1]), + now, + mesh: "test", + target_spec: "peer", + nonce: "n", + ciphertext: "c", + priority: "next", + ...extra, + }); +} + +/** Advance a pending row's next_attempt_at so the next tick picks it up. */ +function makeDue(db: SqliteDb, id: string, now: number) { + db.prepare(`UPDATE outbox SET next_attempt_at = ? WHERE id = ?`).run(now, id); +} + +describe("drain hygiene", () => { + let dir: string; + let db: SqliteDb; + let logs: Array<{ level: string; msg: string; meta?: Record }>; + let clock: number; + /** A "daemon broker" that is never open — rows routed to it back off. */ + const fakeDaemonBroker = { + send: async () => ({ ok: false as const, error: "broker_not_open", permanent: false }), + } as unknown as DaemonBrokerClient; + + const opts = (getSession: DrainOptions["getSessionBrokerByPubkey"]): DrainOptions => ({ + db, + brokers: new Map([["test", fakeDaemonBroker]]), + getSessionBrokerByPubkey: getSession, + log: (level, msg, meta) => logs.push({ level, msg, meta }), + nowFn: () => clock, + }); + + beforeEach(async () => { + dir = mkdtempSync(join(tmpdir(), "cmh-drain-")); + db = await openSqlite(join(dir, "outbox.db")); + migrateOutbox(db); + logs = []; + clock = 1_700_000_000_000; + }); + + afterEach(() => { + try { db.close(); } catch { /* ignore */ } + rmSync(dir, { recursive: true, force: true }); + }); + + it("pending rows older than OUTBOX_TTL_MS are marked dead/expired in one sweep", async () => { + row(db, "old", clock - OUTBOX_TTL_MS - 1); + row(db, "fresh", clock); + db.prepare(`UPDATE outbox SET status = 'inflight' WHERE id = 'old'`).run(); + + await drainOnce(opts(() => undefined)); + + expect(findById(db, "old")?.status).toBe("dead"); + expect(findById(db, "old")?.last_error).toBe("expired"); + expect(findById(db, "fresh")?.status).not.toBe("dead"); + const warn = logs.find((l) => l.msg === "drain_expired_rows"); + expect(warn?.meta?.count).toBe(1); + }); + + it("session-attributed row with NO registered session → dead session_gone after SESSION_GONE_TTL_MS", async () => { + row(db, "gone", clock, { sender_session_pubkey: SESSION_PK }); + + // Young row: still retried (backoff), not dead. + await drainOnce(opts(() => undefined)); + expect(findById(db, "gone")?.status).toBe("pending"); + expect(findById(db, "gone")?.last_error).toBe("session_ws_not_open"); + + // Past the grace: dead. + clock += SESSION_GONE_TTL_MS + 1; + makeDue(db, "gone", clock); + await drainOnce(opts(() => undefined)); + expect(findById(db, "gone")?.status).toBe("dead"); + expect(findById(db, "gone")?.last_error).toBe("session_gone"); + expect(logs.some((l) => l.msg === "drain_session_gone")).toBe(true); + }); + + it("registered-but-reconnecting session: not-ready is rate-limited and capped by MAX_ATTEMPTS", async () => { + row(db, "wait", clock, { sender_session_pubkey: SESSION_PK }); + const reconnecting = { isOpen: () => false } as unknown as SessionBrokerClient; + const o = opts(() => reconnecting); + + for (let i = 0; i < MAX_ATTEMPTS_PER_ROW; i++) { + makeDue(db, "wait", clock); + await drainOnce(o); + // A registered session is never "gone", however long it reconnects. + expect(findById(db, "wait")?.last_error).not.toBe("session_gone"); + } + + const r = findById(db, "wait")!; + expect(r.status).toBe("dead"); + expect(r.last_error).toBe("max_attempts: session_ws_not_open"); + expect(r.attempts).toBe(MAX_ATTEMPTS_PER_ROW); + + // Logged on attempt 1 only (MAX_ATTEMPTS_PER_ROW < NOT_READY_LOG_EVERY), + // not once per tick. + const notReady = logs.filter((l) => l.msg === "drain_session_ws_not_ready"); + expect(MAX_ATTEMPTS_PER_ROW).toBeLessThan(NOT_READY_LOG_EVERY); + expect(notReady).toHaveLength(1); + expect(notReady[0]!.meta?.attempts).toBe(1); + expect(notReady[0]!.meta?.registered).toBe(true); + expect(logs.filter((l) => l.msg === "drain_max_attempts")).toHaveLength(1); + }); +}); diff --git a/apps/cli/tests/unit/log-sink.test.ts b/apps/cli/tests/unit/log-sink.test.ts new file mode 100644 index 0000000..25cbdaa --- /dev/null +++ b/apps/cli/tests/unit/log-sink.test.ts @@ -0,0 +1,142 @@ +/** + * Rotating log sink (1.37.1) — regression tests for the 1 GB daemon.log + * found in the 2026-09-08 blackout. Covers: rotation at the size cap, + * `keep` bound, total-budget purge, oversized pre-existing file rotated + * at install, and writes surviving a rotation. + */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { existsSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { + installRotatingLogSink, + purgeArchivesOverBudget, + rotateLogFiles, + type LogSinkHandle, +} from "~/daemon/log-sink.js"; + +let dir: string; +let sink: LogSinkHandle | null = null; + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "cmh-logsink-")); +}); +afterEach(() => { + sink?.uninstall(); + sink = null; + rmSync(dir, { recursive: true, force: true }); +}); + +const size = (p: string) => (existsSync(p) ? statSync(p).size : -1); + +describe("rotateLogFiles", () => { + it("shifts archives and drops the oldest beyond keep", () => { + const p = join(dir, "d.log"); + writeFileSync(p, "current"); + writeFileSync(`${p}.1`, "one"); + writeFileSync(`${p}.2`, "two"); + rotateLogFiles(p, 2); + expect(existsSync(p)).toBe(false); + expect(readFileSync(`${p}.1`, "utf8")).toBe("current"); + expect(readFileSync(`${p}.2`, "utf8")).toBe("one"); + expect(existsSync(`${p}.3`)).toBe(false); + }); + + it("is a no-op when the active file is missing", () => { + const p = join(dir, "missing.log"); + expect(() => rotateLogFiles(p, 3)).not.toThrow(); + expect(existsSync(`${p}.1`)).toBe(false); + }); +}); + +describe("purgeArchivesOverBudget", () => { + it("deletes oldest archives until total ≤ keep × maxBytes", () => { + const p = join(dir, "d.log"); + writeFileSync(`${p}.1`, "x".repeat(50)); + writeFileSync(`${p}.2`, "x".repeat(50)); + writeFileSync(`${p}.3`, "x".repeat(50)); + // budget = 3 × 40 = 120 < 150 → delete .3 (oldest) → 100 ≤ 120 + const deleted = purgeArchivesOverBudget(p, 3, 40); + expect(deleted).toEqual([`${p}.3`]); + expect(existsSync(`${p}.2`)).toBe(true); + }); + + it("deletes a single archive that alone exceeds the whole budget (the 1 GB case)", () => { + const p = join(dir, "d.log"); + writeFileSync(`${p}.1`, "x".repeat(1000)); + const deleted = purgeArchivesOverBudget(p, 5, 100); + expect(deleted).toEqual([`${p}.1`]); + }); +}); + +describe("installRotatingLogSink", () => { + it("captures process.stdout/stderr writes into the file and rotates at the cap", () => { + const p = join(dir, "daemon.log"); + sink = installRotatingLogSink({ path: p, maxBytes: 100, keep: 3 }); + const line = "x".repeat(39) + "\n"; // 40 bytes + process.stdout.write(line); // 40 + process.stderr.write(line); // 80 + process.stdout.write(line); // 120 ≥ 100 → rotate BEFORE writing → active=40, .1=80 + expect(size(p)).toBe(40); + expect(size(`${p}.1`)).toBe(80); + expect(readFileSync(`${p}.1`, "utf8").split("\n").filter(Boolean)).toHaveLength(2); + // Writes keep landing after the rotation (fd reopened). + process.stdout.write("after\n"); + expect(readFileSync(p, "utf8")).toContain("after"); + }); + + it("never keeps more than `keep` archives", () => { + const p = join(dir, "daemon.log"); + sink = installRotatingLogSink({ path: p, maxBytes: 10, keep: 2 }); + // 10-byte lines at a 10-byte cap → rotate on every write after the + // first; two 10-byte archives sit exactly at the 2 × 10 budget. + for (let i = 0; i < 20; i++) process.stdout.write("012345678\n"); + expect(existsSync(`${p}.1`)).toBe(true); + expect(existsSync(`${p}.2`)).toBe(true); + expect(existsSync(`${p}.3`)).toBe(false); + }); + + it("rotates an oversized pre-existing file at install and enforces the budget", () => { + const p = join(dir, "daemon.log"); + writeFileSync(p, "y".repeat(5000)); // way over 100-byte cap and over 3×100 budget + sink = installRotatingLogSink({ path: p, maxBytes: 100, keep: 3 }); + // Active file starts fresh… + expect(size(p)).toBe(0); + // …and the huge archive alone exceeded the 300-byte budget → purged. + expect(existsSync(`${p}.1`)).toBe(false); + process.stdout.write("hello\n"); + expect(readFileSync(p, "utf8")).toBe("hello\n"); + }); + + it("keeps a modest pre-existing file as .1 when within budget", () => { + const p = join(dir, "daemon.log"); + writeFileSync(p, "y".repeat(150)); // over the 100 cap, under the 500 budget + sink = installRotatingLogSink({ path: p, maxBytes: 100, keep: 5 }); + expect(size(p)).toBe(0); + expect(size(`${p}.1`)).toBe(150); + }); + + it("writeLine appends a newline and rotateNow forces a rotation", () => { + const p = join(dir, "daemon.log"); + sink = installRotatingLogSink({ path: p, maxBytes: 10_000, keep: 2 }); + sink.writeLine("{\"msg\":\"a\"}"); + expect(readFileSync(p, "utf8")).toBe("{\"msg\":\"a\"}\n"); + expect(sink.currentSize()).toBe(12); + sink.rotateNow(); + expect(size(p)).toBe(0); + expect(readFileSync(`${p}.1`, "utf8")).toBe("{\"msg\":\"a\"}\n"); + }); + + it("uninstall restores the original streams", () => { + const p = join(dir, "daemon.log"); + const before = process.stdout.write; + sink = installRotatingLogSink({ path: p, maxBytes: 1000, keep: 1 }); + expect(process.stdout.write).not.toBe(before); + sink.uninstall(); + sink = null; + // Restored function is the bound original — writing must not touch the file. + expect(size(p)).toBe(0); + }); +}); diff --git a/apps/cli/tests/unit/sse-resubscribe.test.ts b/apps/cli/tests/unit/sse-resubscribe.test.ts new file mode 100644 index 0000000..196e17c --- /dev/null +++ b/apps/cli/tests/unit/sse-resubscribe.test.ts @@ -0,0 +1,149 @@ +/** + * MCP SSE resubscribe (1.37.1) — regression for the 2026-09-08 finding + * that a daemon dying hard (SIGKILL / process.exit) left every MCP + * plugin's `/v1/events` subscription dead: the response emitted + * `aborted`/`close` (never `end`), and the pre-1.37.1 client only + * resubscribed on `end`. Also covers the idle watchdog (no bytes at all, + * e.g. a half-open socket after sleep). + */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { createServer, type Server, type ServerResponse } from "node:http"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { subscribeEvents } from "~/mcp/server.js"; + +let dir: string; +let server: Server | null = null; +let sockPath: string; + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "cmh-sse-")); + sockPath = join(dir, "d.sock"); +}); +afterEach(async () => { + if (server) await new Promise((r) => server!.close(() => r())); + server = null; + rmSync(dir, { recursive: true, force: true }); +}); + +interface Fake { + connections: number; + responses: ServerResponse[]; +} + +function startFakeDaemon(opts: { keepalive?: boolean } = {}): Promise { + const fake: Fake = { connections: 0, responses: [] }; + server = createServer((req, res) => { + if (req.url !== "/v1/events") { res.statusCode = 404; res.end(); return; } + fake.connections++; + fake.responses.push(res); + res.statusCode = 200; + res.setHeader("Content-Type", "text/event-stream"); + res.write(": connected\n\n"); + if (opts.keepalive !== false) { + const t = setInterval(() => { try { res.write(": keepalive\n\n"); } catch { /* gone */ } }, 100); + res.on("close", () => clearInterval(t)); + } + }); + return new Promise((resolve) => server!.listen(sockPath, () => resolve(fake))); +} + +const waitFor = async (pred: () => boolean, ms = 3_000): Promise => { + const start = Date.now(); + while (!pred()) { + if (Date.now() - start > ms) throw new Error("timeout waiting for condition"); + await new Promise((r) => setTimeout(r, 20)); + } +}; + +describe("subscribeEvents resubscribe", () => { + it("reconnects when the daemon destroys the socket without ending the response (kill -9)", async () => { + const fake = await startFakeDaemon(); + const events: string[] = []; + const logs: string[] = []; + const sub = subscribeEvents((e) => events.push(e.kind), { + socketPath: sockPath, + backoffMs: [50, 50], + idleTimeoutMs: 0, + log: (m) => logs.push(m), + }); + try { + await waitFor(() => fake.connections === 1); + // Simulate a hard daemon death: tear the TCP/UDS socket down with no + // `end` frame. The client sees close/aborted/error — never `end`. + fake.responses[0]!.socket?.destroy(); + await waitFor(() => fake.connections >= 2, 3_000); + expect(fake.connections).toBeGreaterThanOrEqual(2); + expect(logs).toContain("sse_reconnect"); + expect(logs.filter((l) => l === "sse_subscribed").length).toBeGreaterThanOrEqual(2); + // The new stream is live: an event pushed on it reaches onEvent. + const res = fake.responses[fake.responses.length - 1]!; + res.write(`event: message\ndata: ${JSON.stringify({ ts: "t", body: "hi" })}\n\n`); + await waitFor(() => events.includes("message")); + } finally { + sub.close(); + } + }); + + it("reconnects when the daemon is down at first and comes up later (connect error path)", async () => { + const events: string[] = []; + const sub = subscribeEvents((e) => events.push(e.kind), { + socketPath: sockPath, backoffMs: [50, 50], idleTimeoutMs: 0, + }); + try { + await new Promise((r) => setTimeout(r, 120)); // a couple of ECONNREFUSED/ENOENT attempts + const fake = await startFakeDaemon(); + await waitFor(() => fake.connections >= 1, 3_000); + } finally { + sub.close(); + } + }); + + it("idle watchdog: destroys a silent stream and resubscribes", async () => { + const fake = await startFakeDaemon({ keepalive: false }); + const logs: string[] = []; + const sub = subscribeEvents(() => { /* ignore */ }, { + socketPath: sockPath, + backoffMs: [30], + idleTimeoutMs: 300, + log: (m) => logs.push(m), + }); + try { + await waitFor(() => fake.connections === 1); + await waitFor(() => fake.connections >= 2, 3_000); + expect(logs).toContain("sse_idle_timeout"); + } finally { + sub.close(); + } + }); + + it("a live keepalive stream is NOT torn down by the watchdog", async () => { + const fake = await startFakeDaemon({ keepalive: true }); // keepalive every 100 ms + const logs: string[] = []; + const sub = subscribeEvents(() => { /* ignore */ }, { + socketPath: sockPath, backoffMs: [30], idleTimeoutMs: 400, log: (m) => logs.push(m), + }); + try { + await waitFor(() => fake.connections === 1); + await new Promise((r) => setTimeout(r, 900)); + expect(fake.connections).toBe(1); + expect(logs).not.toContain("sse_idle_timeout"); + } finally { + sub.close(); + } + }); + + it("close() stops reconnecting", async () => { + const fake = await startFakeDaemon(); + const sub = subscribeEvents(() => { /* ignore */ }, { + socketPath: sockPath, backoffMs: [30], idleTimeoutMs: 0, + }); + await waitFor(() => fake.connections === 1); + sub.close(); + await new Promise((r) => setTimeout(r, 300)); + expect(fake.connections).toBe(1); + }); +}); diff --git a/apps/cli/tests/unit/ws-lifecycle.test.ts b/apps/cli/tests/unit/ws-lifecycle.test.ts new file mode 100644 index 0000000..02c0b40 --- /dev/null +++ b/apps/cli/tests/unit/ws-lifecycle.test.ts @@ -0,0 +1,189 @@ +/** + * ws-lifecycle: the 1.37.1 synchronous-handle contract. + * + * Regression for the 2026-09-08 mesh blackout (spec addendum 3): a + * first-attempt hello-ack timeout used to REJECT the connect promise while + * the helper kept reconnecting on its own — the caller never got the + * handle. Now the handle is returned synchronously, a timeout only forces a + * reconnect, and `ready` resolves on whichever attempt the broker acks. + */ + +import { afterEach, describe, expect, it } from "vitest"; +import { WebSocketServer, type WebSocket } from "ws"; +import type { AddressInfo } from "node:net"; + +import { createWsLifecycle, type WsLifecycle } from "~/daemon/ws-lifecycle.js"; + +interface FakeBroker { + url: string; + connections: number; + hellos: number; + sockets: WebSocket[]; + close(): Promise; +} + +/** Minimal broker: `onHello(sock, attemptIndex)` decides whether/when to ack. */ +function startFakeBroker(onHello: (sock: WebSocket, attempt: number) => void): Promise { + return new Promise((resolve) => { + const wss = new WebSocketServer({ port: 0 }); + const state: FakeBroker = { + url: "", + connections: 0, + hellos: 0, + sockets: [], + close: () => new Promise((r) => { + for (const s of state.sockets) { try { s.terminate(); } catch { /* ignore */ } } + wss.close(() => r()); + }), + }; + wss.on("connection", (sock) => { + state.connections++; + state.sockets.push(sock); + sock.on("message", (raw) => { + const msg = JSON.parse(raw.toString()) as { type: string }; + if (msg.type === "hello") { + const attempt = state.hellos++; + onHello(sock, attempt); + } else if (msg.type === "echo") { + sock.send(JSON.stringify({ type: "echoed" })); + } + }); + }); + wss.on("listening", () => { + const { port } = wss.address() as AddressInfo; + state.url = `ws://127.0.0.1:${port}`; + resolve(state); + }); + }); +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +const silent = () => { /* quiet logs in tests */ }; + +function makeClient(url: string, extra: Partial[0]> = {}) { + const received: Array> = []; + const statuses: string[] = []; + const handle: WsLifecycle = createWsLifecycle({ + url, + buildHello: async () => ({ type: "hello" }), + isHelloAck: (m) => m.type === "hello_ack", + onMessage: (m) => received.push(m), + onStatusChange: (s) => statuses.push(s), + helloAckTimeoutMs: 150, + backoffCapsMs: [50], + log: silent, + ...extra, + }); + return { handle, received, statuses }; +} + +describe("createWsLifecycle", () => { + let broker: FakeBroker | null = null; + let handle: WsLifecycle | null = null; + + afterEach(async () => { + await handle?.close().catch(() => { /* ignore */ }); + handle = null; + await broker?.close(); + broker = null; + }); + + it("first hello_ack times out, second attempt acked → same handle, ready resolves, send works", async () => { + broker = await startFakeBroker((sock, attempt) => { + if (attempt === 0) return; // never ack the first attempt → client times out at 150ms + sock.send(JSON.stringify({ type: "hello_ack" })); + }); + const c = makeClient(broker.url); + handle = c.handle; + + // Handle is available synchronously, before any network round trip. + expect(handle.status).toBe("connecting"); + expect(handle.lastAckAt).toBeNull(); + + await handle.ready; + + expect(handle.status).toBe("open"); + expect(handle.reconnects).toBe(1); + expect(handle.lastAckAt).not.toBeNull(); + expect(broker.hellos).toBe(2); + expect(c.statuses).toEqual(["connecting", "reconnecting", "connecting", "open"]); + + handle.send({ type: "echo" }); + await sleep(50); + expect(c.received).toEqual([{ type: "echoed" }]); + }); + + it("close() before the first ack stops reconnecting and rejects ready", async () => { + broker = await startFakeBroker(() => { /* never ack */ }); + const c = makeClient(broker.url, { helloAckTimeoutMs: 60, backoffCapsMs: [20] }); + handle = c.handle; + + // Let the first attempt connect + send hello, then close mid-handshake. + await sleep(30); + const connectionsAtClose = broker.connections; + await handle.close(); + + await expect(handle.ready).rejects.toThrow("client_closed"); + expect(handle.status).toBe("closed"); + + await sleep(300); + expect(broker.connections).toBe(connectionsAtClose); // no further attempts + }); + + it("server-initiated close → client reconnects and re-hellos", async () => { + broker = await startFakeBroker((sock) => sock.send(JSON.stringify({ type: "hello_ack" }))); + const c = makeClient(broker.url); + handle = c.handle; + await handle.ready; + expect(broker.hellos).toBe(1); + + // Broker drops the socket (e.g. presence_evicted 4004). + broker.sockets[0]!.close(4004, "presence_evicted"); + await sleep(250); + + expect(broker.hellos).toBe(2); + expect(handle.status).toBe("open"); + expect(handle.reconnects).toBe(1); + expect(c.statuses.slice(-2)).toEqual(["connecting", "open"]); + }); +}); + +describe("createWsLifecycle — session_replaced is terminal (spec addendum 6)", () => { + let broker: FakeBroker | null = null; + afterEach(async () => { if (broker) { await broker.close(); broker = null; } }); + + it("stops reconnecting when the broker closes the current socket with 1000 session_replaced", async () => { + broker = await startFakeBroker((sock) => { + sock.send(JSON.stringify({ type: "hello_ack" })); + }); + const { handle, statuses } = makeClient(broker.url); + await handle.ready; + expect(handle.status).toBe("open"); + expect(handle.replaced).toBe(false); + + // Another socket took over this identity: broker kicks us. + broker.sockets[0]!.close(1000, "session_replaced"); + await sleep(300); + + expect(handle.replaced).toBe(true); + expect(handle.status).toBe("closed"); + expect(statuses.at(-1)).toBe("closed"); + // No reconnect attempt was made (backoff is 50 ms, we waited 300 ms). + expect(broker.connections).toBe(1); + expect(handle.reconnects).toBe(0); + }); + + it("an ordinary 1006 close still reconnects", async () => { + broker = await startFakeBroker((sock) => { + sock.send(JSON.stringify({ type: "hello_ack" })); + }); + const { handle } = makeClient(broker.url); + await handle.ready; + broker.sockets[0]!.terminate(); + await sleep(400); + expect(handle.replaced).toBe(false); + expect(broker.connections).toBe(2); + expect(handle.status).toBe("open"); + await handle.close(); + }); +});