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(); + }); +});