fix(cli): daemon restart no longer blacks out the mesh (1.37.1)

Daemon side of the 2026-09-08 incident (spec Addenda 3, 5, 6):

- ws-lifecycle: createWsLifecycle() returns the handle synchronously with
  ready exposed separately. A first hello-ack timeout used to REJECT the
  connect promise while the helper kept reconnecting on its own, so
  DaemonBrokerClient.lifecycle stayed null forever and every RPC gated on
  it (listPeers, send, ...) short-circuited: peer list answered [] on all 9
  meshes for as long as the daemon lived. Ack timeout 5 s → 15 s (a timeout
  is now just a reconnect). 1000 session_replaced is terminal for a
  lifecycle (two daemons no longer fight over the same sessions).
- drain: 24 h outbox TTL (expired), session_gone after 10 min, attempt cap on
  the not-ready path, drain_session_ws_not_ready logged on attempt 1 and
  every 60th (was every tick: 4 rows × ~130 k attempts = a 1 GB log).
- Rotating log sink (20 MB × 5, total budget) behind an explicit --log-sink
  flag that the launchd/systemd unit and the detached spawn pass; units
  point StandardOut/Err at daemon.launchd.log and pin ThrottleInterval 10.
- Exit forensics: daemon_exit / daemon_uncaught on every exit path; SIGHUP
  is a clean shutdown.
- Singleton hardening: the CLI stale-probe no longer unlinks a live
  daemon's pid/socket on a slow probe (only ECONNREFUSED/ENOENT or a dead
  pid is stale); the daemon probes the socket for another live daemon
  before taking the lock and re-asserts its pid file every 30 s.
- MCP SSE: resubscribe on every termination event with a 45 s idle
  watchdog; the daemon ends open SSE streams before exiting.
- Operator signal: GET /v1/diagnostics; daemon status shows per-mesh and
  per-session WS state; new daemon restart; doctor adds version match,
  member/session WS open, ghost detector (every registered session must
  appear in the broker peer list), log size + stale unit; peer list warns
  when the daemon is not connected to the broker for a mesh.

Tests: ws-lifecycle (timeout → same handle, close mid-connect,
session_replaced terminal), daemon-broker-lost-handle (Addendum 3
regression), drain-expiry, log-sink, sse-resubscribe (socket destroyed
without end), daemon-probe-classify. Local drills against the fixed broker:
SIGKILL + respawn → sessions visible at t+1 s, second daemon refused,
pid file self-healed.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Alejandro Gutiérrez
2026-09-08 22:22:58 +01:00
parent 0bf7ee9c84
commit dd351a7846
24 changed files with 2236 additions and 213 deletions

View File

@@ -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 `<channel>` 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-<pid>.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

View File

@@ -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",

View File

@@ -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 <id> 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<number> {
return 0;
}
interface DiagnosticsBody {
pid: number;
version: string;
uptime_s: number;
log: { path: string; bytes: number; rotating: boolean };
meshes: Record<string, { status: string; isOpen: boolean; lastAckAt: string | null; reconnects: number }>;
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<number> {
const pid = readRunningPid();
if (!pid) {
@@ -300,13 +332,54 @@ async function runStatus(opts: DaemonOptions): Promise<number> {
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<DiagnosticsBody | { error?: string }>({ 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<string, string> }>({ 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<number> {
}
}
/**
* 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<number> {
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<void>((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<number> {
try {
const res = await ipc<Record<string, unknown>>({ path: "/v1/version" });
@@ -385,7 +508,7 @@ async function spawnDetachedDaemon(opts: DaemonOptions): Promise<number> {
// .../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");

View File

@@ -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<Check> {
}
}
// ── 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<string, DiagMesh>;
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<Diagnostics | { error?: string }>({ 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<Check[]> {
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<Check> {
const byMesh = new Map<string, DiagSession[]>();
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<string>();
try {
const res = await ipc<{ peers?: Array<{ pubkey?: string }>; brokers?: Record<string, string> }>({
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<void> {
const useColor =
!process.env.NO_COLOR && process.env.TERM !== "dumb" && process.stdout.isTTY;
@@ -258,6 +481,7 @@ export async function runDoctor(): Promise<void> {
checkKeypairs(),
await checkBrokerWs(),
await checkNpmLatest(),
...(await daemonChecks()),
];
for (const c of checks) {

View File

@@ -111,7 +111,15 @@ function projectFields(record: PeerRecord, fields: string[]): Record<string, unk
return out;
}
async function listPeersForMesh(slug: string): Promise<PeerRecord[]> {
/** 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<PeersResult> {
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<PeerRecord[]> {
// `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<PeerRecord[]> {
annotateSelf(p, selfMemberPubkey, selfSessionPubkey),
);
});
return result;
return { peers: result, brokerStatus: null };
}
/**
@@ -235,11 +246,24 @@ export async function runPeers(flags: PeersFlags): Promise<void> {
: 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<void> {
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;
}

View File

@@ -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<string, unknown>) => void;
log?: (level: "info" | "warn" | "error", msg: string, meta?: Record<string, unknown>) => 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<void> {
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<string, unknown>): 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;
}

View File

@@ -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<string, unknown>) => 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<DrainOptions["log"]>): Promise<void> {
const now = Date.now();
/** Exported for tests: one drain pass. */
export async function drainOnce(opts: DrainOptions, log: NonNullable<DrainOptions["log"]> = opts.log ?? defaultLog): Promise<void> {
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<DrainOptions["log"
// Session-attributed row. Require an open session-WS — see comment
// above on why we don't fall back to the daemon-WS.
if (!sessionBroker || !sessionBroker.isOpen()) {
log("info", "drain_session_ws_not_ready", {
id: row.id, session_pubkey: row.sender_session_pubkey.slice(0, 12),
});
backoffPending(opts.db, row.id, row.attempts + 1, "session_ws_not_open", "session_ws_not_open");
const attempts = row.attempts + 1;
// No registered session owns this pubkey at all (not merely a
// reconnecting one) and the row has waited long enough: the
// session is gone, nothing will ever send it. Dead, not retry.
if (!sessionBroker && now - row.enqueued_at > 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<DrainOptions["log"
}
} catch (e) {
log("warn", "drain_send_threw", { id: row.id, err: String(e) });
backoffPending(opts.db, row.id, row.attempts + 1, "exception", String(e));
backoffPending(opts.db, row.id, row.attempts + 1, "exception", String(e), now);
continue;
}
@@ -208,7 +257,7 @@ async function drainOnce(opts: DrainOptions, log: NonNullable<DrainOptions["log"
log("warn", "drain_max_attempts", { id: row.id, err: res.error });
markDead(opts.db, row.id, `max_attempts: ${res.error}`);
} else {
backoffPending(opts.db, row.id, row.attempts + 1, "retry", res.error);
backoffPending(opts.db, row.id, row.attempts + 1, "retry", res.error, now);
}
}
}
@@ -233,9 +282,9 @@ function markDead(db: SqliteDb, id: string, err: string) {
db.prepare(`UPDATE outbox SET status = 'dead', last_error = ? WHERE id = ?`).run(err, id);
}
function backoffPending(db: SqliteDb, id: string, attempts: number, _kind: string, err: string) {
function backoffPending(db: SqliteDb, id: string, attempts: number, _kind: string, err: string, now: number = Date.now()) {
const wait = Math.min(BACKOFF_CAP_MS, BACKOFF_BASE_MS * (2 ** Math.min(attempts, 12)));
const next = Date.now() + wait;
const next = now + wait;
db.prepare(`
UPDATE outbox
SET status = 'pending', attempts = ?, next_attempt_at = ?, last_error = ?

View File

@@ -10,6 +10,9 @@ import { listOutbox, requeueDeadOrPending, type OutboxStatus } from "../db/outbo
import { randomUUID } from "node:crypto";
import { bindSseStream, type EventBus } from "../events.js";
import type { DaemonBrokerClient } from "../broker.js";
import type { SessionBrokerClient } from "../session-broker.js";
import type { LogSinkHandle } from "../log-sink.js";
import { statSync } from "node:fs";
import {
registerSession, deregisterByToken, resolveToken, listSessions, startReaper,
type SessionInfo,
@@ -48,6 +51,14 @@ export interface IpcServerOptions {
meshConfigs?: Map<string, { slug: string; pubkey: string; secretKey: string }>;
/** 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<string, SessionBrokerClient>;
/** 1.37.1: per-session WS clients keyed by hex session pubkey. */
sessionBrokersByPubkey?: Map<string, SessionBrokerClient>;
/** 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<void>((res) => uds.close(() => res())),
tcp ? new Promise<void>((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<string, unknown>) {
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<string, DaemonBrokerClient>;
meshConfigs?: Map<string, { slug: string; pubkey: string; secretKey: string }>;
onPendingInserted?: () => void;
sessionBrokers?: Map<string, SessionBrokerClient>;
sessionBrokersByPubkey?: Map<string, SessionBrokerClient>;
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<string, { status: string; isOpen: boolean; lastAckAt: string | null; reconnects: number }> = {};
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=<slug> filter narrows the set server-side.
const all: Array<Record<string, unknown> & { 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<string, string> = {};
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<string, unknown>), 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) });
}

View File

@@ -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.<keep>`, 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();
},
};
}

View File

@@ -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

View File

@@ -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<number | null> {
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<string, unknown>): 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<number> {
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<number> {
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<number> {
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<number> {
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<number> {
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<number> {
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<number>(() => { /* never resolves; signals call process.exit */ });

View File

@@ -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 {
"<string>daemon</string>",
"<string>up</string>",
"<string>--foreground</string>",
// 1.37.1: in-process rotating log sink (launchd never rotates).
"<string>--log-sink</string>",
...(args.meshSlug
? ["<string>--mesh</string>", `<string>${escapeXml(args.meshSlug)}</string>`]
: []),
@@ -129,6 +143,8 @@ function installDarwin(args: InstallArgs): InstallResult {
<true/>
<key>KeepAlive</key>
<true/>
<key>ThrottleInterval</key>
<integer>${LAUNCHD_THROTTLE_SECONDS}</integer>
<key>StandardOutPath</key>
<string>${escapeXml(log)}</string>
<key>StandardErrorPath</key>
@@ -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(`<string>${escapeXml(legacyLog)}</string>`)) {
return { stale: true, reason: "StandardOutPath/StandardErrorPath point at daemon.log (no rotation)", path: unit.path };
}
if (!c.includes("<key>ThrottleInterval</key>")) {
return { stale: true, reason: "missing ThrottleInterval", path: unit.path };
}
if (!c.includes("<string>--log-sink</string>")) {
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();

View File

@@ -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<string, unknown>) => void;
log?: (level: "info" | "warn" | "error", msg: string, meta?: Record<string, unknown>) => 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<void> {
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

View File

@@ -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<WsLifecycle>` 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<string, unknown>) => 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<void>;
/** 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<void>;
}
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<WsLifecycle> {
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<WsLifecy
let status: WsStatus = "closed";
let closed = false;
let reconnectAttempt = 0;
let reconnects = 0;
let reconnectTimer: NodeJS.Timeout | null = null;
let helloTimer: NodeJS.Timeout | null = null;
let lastAckAt: number | null = null;
let replaced = false;
let readyResolve!: () => void;
let readyReject!: (err: Error) => void;
let readySettled = false;
const ready = new Promise<void>((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): Promise<WsLifecy
opts.onStatusChange?.(s);
};
/**
* Open one WS attempt. Returns a promise that resolves on hello ack
* or rejects if the socket closes before we get one. Used by both the
* initial connect and the close-handler backoff timer (which awaits
* but ignores the rejection — by then the close handler has already
* scheduled its own reconnect).
*/
// Liveness watchdog: same cadence (30s) as the broker's outbound
// ping. Two jobs per tick:
// 1. If we haven't heard from the broker in >75s (2.5x the ping
@@ -158,8 +199,14 @@ export function connectWsWithBackoff(opts: WsLifecycleOptions): Promise<WsLifecy
let lastActivity = Date.now();
let watchdogTimer: NodeJS.Timeout | null = null;
const openOnce = (): Promise<void> => {
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<WsLifecy
ws = sock;
lastActivity = Date.now();
return new Promise<void>((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<string, unknown>;
try { msg = JSON.parse(raw.toString()) as Record<string, unknown>; }
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<string, unknown>;
try { msg = JSON.parse(raw.toString()) as Record<string, unknown>; }
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;
}

View File

@@ -504,6 +504,10 @@ async function main(): Promise<void> {
// 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);

View File

@@ -123,32 +123,117 @@ function daemonMarkSeen(ids: string[], sessionToken?: string | null): Promise<vo
interface DaemonEvent { kind: string; ts: string; data: Record<string, any> }
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<string, unknown>) => 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 `<channel>`
* 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<typeof httpRequest> | 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<string, string> = { 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<void> {
});
} catch { /* best effort */ }
}
}, { sessionToken: sessionTokenForSeen });
}, { sessionToken: sessionTokenForSeen, log: mcpLog });
// 1.34.6 — Welcome: single emit on oninitialized + 3s grace.
//

View File

@@ -36,20 +36,40 @@ async function daemonReachable(): Promise<boolean> {
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<string, string>;
}
/** 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<DaemonPeersResult | null> {
if (!(await daemonReachable())) return null;
try {
const res = await ipc<{ peers?: unknown[]; brokers?: Record<string, string> }>({
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<unknown[] | null> {
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;
}
/**

View File

@@ -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<EnsureDaemonResu
// Step 1 — probe.
const probe = await probeDaemon();
if (probe === "up") return { state: "up", durationMs: Date.now() - t0 };
// "busy": a live daemon owns the socket but did not answer in time. Do
// NOT clean up, do NOT spawn — report it as up and let the caller's own
// IPC timeout decide (it falls back to the cold path on its own).
if (probe === "busy") return { state: "up", durationMs: Date.now() - t0, reason: "daemon busy (probe timed out; files left intact)" };
// Step 2 — service-managed shortcut. When launchd / systemd manages
// the daemon and KeepAlive is set, the OS will restart a crashed
@@ -171,17 +176,34 @@ function isServiceManaged(): boolean {
return false;
}
async function probeDaemon(): Promise<"up" | "absent" | "stale"> {
/**
* 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);
}
}

View File

@@ -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<void>((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<string, unknown>;
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<void>((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);
});

View File

@@ -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");
});
});

View File

@@ -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<Parameters<typeof insertPending>[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<string, unknown> }>;
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);
});
});

View File

@@ -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);
});
});

View File

@@ -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<void>((r) => server!.close(() => r()));
server = null;
rmSync(dir, { recursive: true, force: true });
});
interface Fake {
connections: number;
responses: ServerResponse[];
}
function startFakeDaemon(opts: { keepalive?: boolean } = {}): Promise<Fake> {
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<void> => {
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);
});
});

View File

@@ -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<void>;
}
/** Minimal broker: `onHello(sock, attemptIndex)` decides whether/when to ack. */
function startFakeBroker(onHello: (sock: WebSocket, attempt: number) => void): Promise<FakeBroker> {
return new Promise((resolve) => {
const wss = new WebSocketServer({ port: 0 });
const state: FakeBroker = {
url: "",
connections: 0,
hellos: 0,
sockets: [],
close: () => new Promise<void>((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<Parameters<typeof createWsLifecycle>[0]> = {}) {
const received: Array<Record<string, unknown>> = [];
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();
});
});