4 Commits

Author SHA1 Message Date
Alejandro Gutiérrez
a9b735a183 fix(broker): balance per-mesh connection counter — stop capacity-leak that bricks meshes
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
`connectionsPerMesh` (the in-memory counter enforcing MAX_CONNECTIONS_PER_MESH=100)
was incremented on every successful hello (incMeshCount at member + session paths)
but decremented ONLY inside evictPresenceFully. Every other removal path —
session-id dedup on reconnect (the common one), kick, and ban — deleted the entry
from `connections` without decrementing, leaking +1 each time. Because the counter
is in-memory and only resets on broker restart, it crept up to 100 over hours/days
of normal reconnect churn (network blips, sleep/wake, relaunches) until the mesh
hit capacity and rejected ALL new connections with `1008 "capacity"` — bricking it
until the broker process was restarted. A user with <10 sessions saw "mesh at
connection capacity" because the 100 were leaked phantoms, not live connections.

Fix: route every non-evict removal through a new dropConnection() helper that
deletes from `connections` AND decMeshCount()s, so the counter tracks map
membership exactly. The replaced socket's own close handler then no-ops (entry
already gone, guarded by `if (!conn) return`), so the decrement happens exactly
once — no double-count. evictPresenceFully keeps its existing balanced delete+dec.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 14:32:33 +01:00
Alejandro Gutiérrez
213a6b37cc feat(cli): self-renewing session attestation + honest send + hide daemons (1.37.0)
Fixes the "peers stop receiving after a while / like expiring" class of bugs.

- Session attestation self-renewal (root cause of the expiry). The daemon
  minted a 12h-TTL parent attestation once at `claudemesh launch` and replayed
  the same stale token on every WS reconnect. Past launch+12h the broker
  rejected session_hello with `expired`, after which the daemon reconnect-
  looped forever with the dead token and the session silently fell off the
  mesh — its ephemeral pubkey lingering in rosters, undeliverable. Trigger was
  any reconnect past 12h: a network blip, a sleep/wake, or (most reliably) a
  broker redeploy that drops every WS at once, killing all sessions >12h old
  together. buildHello now re-mints a fresh attestation per (re)connect from
  the in-memory mesh.secretKey (already used at daemon rehydration), so
  presence is self-healing across reconnects/redeploys. The 12h security bound
  is preserved — live tokens stay short-lived, just refreshed on use. This is
  what lets sessions stay on the mesh for days; the existing keepalive
  (30s ping) + auto-reconnect (exp backoff) + 90s broker grace were already
  sound — the stale attestation was the single point of failure defeating them.

- Honest send status. The daemon outbox path returned `queued` optimistically
  and the drain retried failures (incl. "no connected peer") async forever, so
  a bare `✔ sent` for an offline or stale-session-key target was misleading.
  Direct sends now pre-check the live roster: offline/unknown key → `⚠ queued
  — no connected peer matches this key` + ephemeral-key explanation; online →
  `✔ sent to <name> (online)`. Applies to both daemon and cold paths; JSON
  gains `recipientOnline` + `status`.

- Hide control-plane daemons from peer list + target resolution. The
  control-plane filter was human-output-only; `peer list --json` still leaked
  the daemon's row, making it look like an addressable peer. Now filtered from
  JSON too (--all still shows it). Name/prefix resolution and the --self
  fan-out filter now exclude control-plane rows by peerRole (the reliable
  marker) rather than the channel string.

- --self no-op warning. --self only governs own-member-key fan-out; passed with
  a session pubkey it was silently ignored. It now warns it had no effect and
  sends normally (messaging a specific session pubkey needs no flag).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 12:51:31 +01:00
Alejandro Gutiérrez
c747040e0d feat(cli): durable session→mesh binding + cross-mesh send (1.36.0)
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
Fixes the 'live peer looks disconnected' class of bugs. Two layers:

ROOT CAUSE — involuntary mesh context loss:
The session→mesh binding lived only in the daemon's in-memory registry,
so a daemon restart (e.g. `daemon down && up`) wiped it. Every live
session then lost its mesh, and CLI commands fell back to an arbitrary
default mesh — a peer that never moved looked offline.

Fix: persist session bindings to ~/.claudemesh/daemon/sessions.json
(secret-free — keypairs reload from the per-session keypair store). On
boot the daemon rehydrates each binding whose pid is still alive (with a
start-time PID-reuse guard), reloads its keypair, re-signs a parent
attestation, and re-registers it — which reconnects its SessionBroker
WS. Restarts are now transparent; sessions keep their mesh.

DEFENSIVE LAYER — cross-mesh send resolution:
`send` without --mesh and several joined meshes returned mesh_required;
a prefix under --mesh X resolved against the default mesh's roster, not
X's (only the full 64-char pubkey worked). Now a name/prefix is resolved
across all joined meshes (or scoped to --mesh): unique match auto-selects
its mesh, multi-mesh match asks for --mesh, none gives a clear error.
Kills mesh_required for peers on a non-default mesh and fixes P3.

Maps to field-report P1/P2/P3. P4 (shared member) left as-is (by design).
New: 5 persistence unit tests. Full suite 119/119. Daemon boot verified.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-02 20:38:37 +01:00
Alejandro Gutiérrez
71401b1d50 fix(cli): session config.json written 0600, not 0644 (1.35.1)
Some checks failed
CI / Lint (push) Has been cancelled
CI / Typecheck (push) Has been cancelled
CI / Broker tests (Postgres) (push) Has been cancelled
CI / Docker build (linux/amd64) (push) Has been cancelled
The per-session config written to the launch tmpdir embeds the mesh
keypair (secret key) but was created without a mode → 0644
(world/group-readable), which `claudemesh status` flags as
"perms 0644 — expected 0600". The enclosing mkdtemp dir is 0700, but
lock the file down too so the secret is never world-readable. File is
freshly created in a new tmpdir, so the mode applies on create.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-02 18:30:47 +01:00
10 changed files with 471 additions and 74 deletions

View File

@@ -1635,6 +1635,30 @@ function decMeshCount(meshId: string): void {
metrics.connectionsActive.set(connections.size); metrics.connectionsActive.set(connections.size);
} }
/**
* Remove a connection from the in-memory map AND keep the per-mesh
* connection counter balanced.
*
* THE counter (`connectionsPerMesh`) is incremented once per successful
* hello (incMeshCount). Before this helper, it was decremented ONLY in
* evictPresenceFully — so every other removal path (session-id dedup on
* reconnect, kick, ban) deleted the entry from `connections` without
* decrementing, leaking +1 each time. Over many reconnects the counter
* crept up to MAX_CONNECTIONS_PER_MESH and the mesh got bricked with
* `capacity` rejections until the broker process restarted (the counter
* is in-memory). Routing every non-evict removal through here makes the
* counter track `connections` membership exactly. The replaced socket's
* own close handler then no-ops (entry already gone), so the decrement
* happens exactly once. Returns true if an entry was actually removed.
*/
function dropConnection(presenceId: string): boolean {
const c = connections.get(presenceId);
if (!c) return false; // already removed (e.g. evicted) — don't double-decrement
connections.delete(presenceId);
decMeshCount(c.meshId);
return true;
}
function sendError( function sendError(
ws: WebSocket, ws: WebSocket,
code: string, code: string,
@@ -1954,7 +1978,7 @@ async function handleHello(
if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) { if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) {
log.info("hello dedup", { old_presence: oldPid, session_id: hello.sessionId }); log.info("hello dedup", { old_presence: oldPid, session_id: hello.sessionId });
try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ }
connections.delete(oldPid); dropConnection(oldPid);
void disconnectPresence(oldPid); void disconnectPresence(oldPid);
} }
} }
@@ -2241,7 +2265,7 @@ async function handleSessionHello(
if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) { if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) {
log.info("session_hello dedup", { old_presence: oldPid, session_id: hello.sessionId }); log.info("session_hello dedup", { old_presence: oldPid, session_id: hello.sessionId });
try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ }
connections.delete(oldPid); dropConnection(oldPid);
void disconnectPresence(oldPid); void disconnectPresence(oldPid);
} }
} }
@@ -4932,7 +4956,7 @@ function handleConnection(ws: WebSocket): void {
continue; continue;
} }
try { peer.ws.close(closeCode, closeReason); } catch {} try { peer.ws.close(closeCode, closeReason); } catch {}
connections.delete(pid); dropConnection(pid);
void disconnectPresence(pid); void disconnectPresence(pid);
affected.push(peer.displayName || pid); affected.push(peer.displayName || pid);
} }
@@ -4947,7 +4971,7 @@ function handleConnection(ws: WebSocket): void {
continue; continue;
} }
try { peer.ws.close(closeCode, `${closeReason}_stale`); } catch {} try { peer.ws.close(closeCode, `${closeReason}_stale`); } catch {}
connections.delete(pid); dropConnection(pid);
void disconnectPresence(pid); void disconnectPresence(pid);
affected.push(peer.displayName || pid); affected.push(peer.displayName || pid);
} }
@@ -4961,7 +4985,7 @@ function handleConnection(ws: WebSocket): void {
continue; continue;
} }
try { peer.ws.close(closeCode, closeReason); } catch {} try { peer.ws.close(closeCode, closeReason); } catch {}
connections.delete(pid); dropConnection(pid);
void disconnectPresence(pid); void disconnectPresence(pid);
affected.push(peer.displayName || pid); affected.push(peer.displayName || pid);
} }
@@ -5017,7 +5041,7 @@ function handleConnection(ws: WebSocket): void {
for (const [pid, peer] of connections) { for (const [pid, peer] of connections) {
if (peer.meshId === conn.meshId && peer.memberPubkey === targetMember.peerPubkey) { if (peer.meshId === conn.meshId && peer.memberPubkey === targetMember.peerPubkey) {
try { peer.ws.close(4002, "banned"); } catch {} try { peer.ws.close(4002, "banned"); } catch {}
connections.delete(pid); dropConnection(pid);
void disconnectPresence(pid); void disconnectPresence(pid);
} }
} }

View File

@@ -1,6 +1,6 @@
{ {
"name": "claudemesh-cli", "name": "claudemesh-cli",
"version": "1.35.0", "version": "1.37.0",
"description": "Peer mesh for Claude Code sessions — CLI + MCP server.", "description": "Peer mesh for Claude Code sessions — CLI + MCP server.",
"keywords": [ "keywords": [
"claude-code", "claude-code",

View File

@@ -763,10 +763,16 @@ export async function runLaunch(flags: LaunchFlags, rawArgs: string[]): Promise<
...(parsedGroups.length > 0 ? { groups: parsedGroups } : {}), ...(parsedGroups.length > 0 ? { groups: parsedGroups } : {}),
messageMode, messageMode,
}; };
// mode 0600: this config embeds the mesh keypair (secret key). Written
// without a mode it lands at 0644 (world/group-readable) — which
// `claudemesh status` flags as "perms 0644 — expected 0600". The
// enclosing tmpDir is already 0700, but lock the file down too so the
// secret is never world-readable even for a moment. The file is freshly
// created in a new mkdtemp dir, so the mode applies on create.
writeFileSync( writeFileSync(
join(tmpDir, "config.json"), join(tmpDir, "config.json"),
JSON.stringify(sessionConfig, null, 2) + "\n", JSON.stringify(sessionConfig, null, 2) + "\n",
"utf-8", { encoding: "utf-8", mode: 0o600 },
); );
// 4b. Mint a per-session IPC token, persist it under tmpDir, and // 4b. Mint a per-session IPC token, persist it under tmpDir, and

View File

@@ -241,14 +241,6 @@ export async function runPeers(flags: PeersFlags): Promise<void> {
try { try {
const peers = await listPeersForMesh(slug); const peers = await listPeersForMesh(slug);
if (wantsJson) {
const projected = fieldList
? peers.map((p) => projectFields(p, fieldList))
: peers;
allJson.push({ mesh: slug, peers: projected });
continue;
}
// Hide control-plane rows by default — they're infrastructure // Hide control-plane rows by default — they're infrastructure
// (daemon-WS member-keyed presence), not interactive peers, and // (daemon-WS member-keyed presence), not interactive peers, and
// they confused users into thinking the daemon counted as a // they confused users into thinking the daemon counted as a
@@ -258,10 +250,22 @@ export async function runPeers(flags: PeersFlags): Promise<void> {
// 2026-05-04). annotateSelf() filled in 'session' for older // 2026-05-04). annotateSelf() filled in 'session' for older
// brokers that don't emit peerRole yet, so this filter is // brokers that don't emit peerRole yet, so this filter is
// backwards-compatible by construction — legacy rows show up. // backwards-compatible by construction — legacy rows show up.
//
// Applied to JSON too (was human-output-only): `peer list --json`
// leaking the daemon's control-plane row is what made the daemon
// look like an addressable peer and sent DMs into a black hole.
const visible = flags.all const visible = flags.all
? peers ? peers
: peers.filter((p) => p.peerRole !== "control-plane"); : peers.filter((p) => p.peerRole !== "control-plane");
if (wantsJson) {
const projected = fieldList
? visible.map((p) => projectFields(p, fieldList))
: visible;
allJson.push({ mesh: slug, peers: projected });
continue;
}
// Sort: this-session first, then your-other-sessions, then real // Sort: this-session first, then your-other-sessions, then real
// peers. Within each group, idle/working ahead of dnd. Inside the // peers. Within each group, idle/working ahead of dnd. Inside the
// groups, leave broker order. The point is: when you run peer // groups, leave broker order. The point is: when you run peer

View File

@@ -40,63 +40,98 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
: flags.priority === "low" ? "low" : flags.priority === "low" ? "low"
: "next"; : "next";
// Resolve which mesh to use. With --mesh, target it directly. // Resolve which mesh to use. With --mesh, target it directly. Without,
// Without, use first joined mesh — same default as withMesh. // use the only joined mesh, else leave null and let target resolution
// below discover the right mesh from where the peer actually lives.
const config = readConfig(); const config = readConfig();
const meshSlug = let meshSlug =
flags.mesh ?? flags.mesh ??
(config.meshes.length === 1 ? config.meshes[0]!.slug : null); (config.meshes.length === 1 ? config.meshes[0]!.slug : null);
// 1.31.6: hex-prefix resolution. If `to` looks like hex but isn't a // Cross-mesh target resolution (1.36.0). A direct send to a hex prefix
// full 64-char pubkey, resolve it against the peer list and replace // or display name is resolved against the peer rosters so the CLI:
// it with the matching full pubkey. The broker stores `targetSpec` // - expands a prefix/name to the full session pubkey (the broker's
// verbatim and the drain query at apps/broker/src/broker.ts:2408 // drain matches only full pubkeys — a bare prefix would queue but
// matches only on full pubkeys, so a 16-hex prefix would queue // never fetch: sender saw "sent", recipient saw nothing);
// successfully but never fetch — sender saw "sent", recipient saw // - DISCOVERS which joined mesh the target is on when no --mesh was
// nothing. Resolving here makes the CLI's prefix UX work end-to-end // given and several meshes are joined. Previously this returned
// and surfaces ambiguous / unmatched prefixes with a clear error // `mesh_required` and a live peer on a non-default mesh looked
// instead of a silent drop. // "disconnected". We now scan every joined mesh's roster and, if
if ( // the target resolves in exactly one, auto-select that mesh.
!to.startsWith("@") && // With --mesh (or a single joined mesh) the scan is scoped to that one
!to.startsWith("#") && // mesh, so `send --mesh X <prefix>` resolves against X's roster — not
to !== "*" && // the default mesh (the bug where only the full 64-char pubkey worked).
/^[0-9a-f]{4,63}$/i.test(to) const isDirect = !to.startsWith("@") && !to.startsWith("#") && to !== "*";
) { const isFullPubkey = /^[0-9a-f]{64}$/i.test(to);
try { const isPrefix = /^[0-9a-f]{4,63}$/i.test(to);
const isName = isDirect && !isFullPubkey && !isPrefix;
if (isDirect && (isPrefix || isName || (isFullPubkey && !meshSlug))) {
const { tryListPeersViaDaemon } = await import("~/services/bridge/daemon-route.js"); const { tryListPeersViaDaemon } = await import("~/services/bridge/daemon-route.js");
const peers = (await tryListPeersViaDaemon()) ?? []; const searchSlugs = meshSlug ? [meshSlug] : config.meshes.map((m) => m.slug);
const lower = to.toLowerCase(); const lower = to.toLowerCase();
const matches = peers.filter((p) => { let daemonReachable = false;
const pk = (p as { pubkey?: string }).pubkey ?? ""; type Hit = { slug: string; pubkey: string; displayName: string };
const mpk = (p as { memberPubkey?: string }).memberPubkey ?? ""; const matches: Hit[] = [];
return pk.toLowerCase().startsWith(lower) || mpk.toLowerCase().startsWith(lower); for (const slug of searchSlugs) {
}); const peers = await tryListPeersViaDaemon(slug);
if (matches.length === 0) { if (peers === null) continue; // daemon unreachable for this query
render.err(`No peer matches hex prefix "${to}".`); daemonReachable = true;
const names = peers for (const p of peers) {
.map((p) => (p as { displayName?: string }).displayName) // Never resolve a name/prefix to a control-plane daemon row — it's
.filter(Boolean) // infrastructure, not an addressable peer, and matching it sends a
.join(", "); // DM that the daemon swallows. (peerRole is the reliable marker;
if (names) render.hint(`online: ${names}`); // the daemon's own row is control-plane.)
if ((p as { peerRole?: string }).peerRole === "control-plane") continue;
const pk = ((p as { pubkey?: string }).pubkey ?? "").toLowerCase();
const mpk = ((p as { memberPubkey?: string }).memberPubkey ?? "").toLowerCase();
const dn = (p as { displayName?: string }).displayName ?? "?";
const hit = isName
? dn.toLowerCase() === lower
: pk.startsWith(lower) || mpk.startsWith(lower);
if (hit) matches.push({ slug, pubkey: (p as { pubkey?: string }).pubkey ?? "", displayName: dn });
}
}
// Only act on a reachable daemon. If it was down for every query, fall
// through to the cold path, which opens its own WS and resolves names.
if (daemonReachable) {
const byPubkey = new Map<string, Hit>();
for (const m of matches) if (!byPubkey.has(m.pubkey)) byPubkey.set(m.pubkey, m);
const uniq = [...byPubkey.values()];
const meshesHit = [...new Set(uniq.map((m) => m.slug))];
if (uniq.length === 0) {
// For a full pubkey we couldn't locate, keep going — the user gave
// a complete key and the daemon send will surface a clear error.
if (!isFullPubkey) {
render.err(`No peer matches "${to}"${flags.mesh ? ` on mesh "${flags.mesh}"` : " on any joined mesh"}.`);
render.hint("Check `claudemesh peer list` (add --mesh <slug> to scope).");
process.exit(1); process.exit(1);
} }
if (matches.length > 1) { } else if (uniq.length > 1) {
const candidates = matches if (meshesHit.length > 1 && !meshSlug) {
.map((p) => { // Target lives on several meshes — disambiguate by mesh, not prefix.
const pk = (p as { pubkey?: string }).pubkey ?? ""; const where = uniq
const dn = (p as { displayName?: string }).displayName ?? "?"; .map((m) => `${m.displayName} ${m.pubkey.slice(0, 12)}… @${m.slug}`)
return `${dn} ${pk.slice(0, 16)}`;
})
.join(", "); .join(", ");
render.err(`Ambiguous hex prefix "${to}" matches ${matches.length} peers.`); render.err(`"${to}" matches peers on ${meshesHit.length} meshes — pick one with --mesh <slug>.`);
render.hint(`candidates: ${where}`);
process.exit(1);
}
const candidates = uniq
.map((m) => `${m.displayName} ${m.pubkey.slice(0, 16)}`)
.join(", ");
render.err(`Ambiguous ${isName ? "name" : "prefix"} "${to}" — matches ${uniq.length} peers.`);
render.hint(`candidates: ${candidates}`); render.hint(`candidates: ${candidates}`);
render.hint("Use a longer prefix or paste the full 64-char pubkey."); render.hint("Use a longer prefix or paste the full 64-char pubkey.");
process.exit(1); process.exit(1);
} else {
// Exactly one match — adopt its mesh (P1: kills mesh_required for
// peers on a non-default mesh) and its full pubkey (prefix/name).
meshSlug = uniq[0]!.slug;
if (!isFullPubkey) to = uniq[0]!.pubkey;
} }
to = (matches[0] as { pubkey?: string }).pubkey ?? to;
} catch {
// Daemon unreachable — fall through; cold path will try a name
// lookup and surface its own error if that also fails.
} }
} }
@@ -135,10 +170,14 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
const session = await getSessionInfo(); const session = await getSessionInfo();
const ownSessionPk = session?.presence?.sessionPubkey?.toLowerCase(); const ownSessionPk = session?.presence?.sessionPubkey?.toLowerCase();
const siblings = peers.filter((p) => { const siblings = peers.filter((p) => {
const r = p as { memberPubkey?: string; pubkey?: string; channel?: string }; const r = p as { memberPubkey?: string; pubkey?: string; channel?: string; peerRole?: string };
if (!r.pubkey) return false; if (!r.pubkey) return false;
if (ownSessionPk && r.pubkey.toLowerCase() === ownSessionPk) return false; if (ownSessionPk && r.pubkey.toLowerCase() === ownSessionPk) return false;
if (r.channel === "claudemesh-daemon") return false; // Exclude the daemon's own control-plane presence row. peerRole is
// the reliable marker (the live daemon row is control-plane even
// when its channel reads "claudemesh-session"); keep the channel
// check too for older brokers that don't emit peerRole.
if (r.peerRole === "control-plane" || r.channel === "claudemesh-daemon") return false;
return r.memberPubkey?.toLowerCase() === to.toLowerCase(); return r.memberPubkey?.toLowerCase() === to.toLowerCase();
}); });
if (siblings.length === 0) { if (siblings.length === 0) {
@@ -184,6 +223,45 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
} }
} }
// --self only governs the own-member-key fan-out above, which returns
// early. Reaching here with --self still set means the target was NOT
// your own member pubkey, so the flag did nothing. Say so rather than
// ignoring it silently — the old behavior made `send --self <session-
// pubkey>` look like it controlled routing when it was inert. Messaging
// a specific session pubkey (including one of your own sibling sessions)
// needs no flag and just works.
if (flags.self) {
render.warn("--self had no effect: it only applies when the target is your own member pubkey (fan-out to your sibling sessions). Sending to this specific pubkey directly.");
}
// Honest-delivery pre-check (direct sends only). The daemon path below
// queues into the local outbox and returns `queued` optimistically; the
// drain then delivers async and retries failures (incl. "no connected
// peer") forever. So a bare "sent" line was misleading — a DM to an
// offline or stale-session-key target looked delivered but never was.
// Resolve the live roster once to learn whether `to` is addressable
// right now; this only shapes the confirmation wording (the send still
// queues regardless, preserving store-and-forward for genuinely-offline
// peers). null = unknown (not a direct DM, or daemon unreachable).
let recipientOnline: boolean | null = null;
let recipientName: string | undefined;
if (isDirect && meshSlug) {
const { tryListPeersViaDaemon } = await import("~/services/bridge/daemon-route.js");
const peers = await tryListPeersViaDaemon(meshSlug);
if (peers !== null) {
const lower = to.toLowerCase();
const match = peers.find((p) => {
const r = p as { pubkey?: string; memberPubkey?: string; peerRole?: string };
if (r.peerRole === "control-plane") return false;
return r.pubkey?.toLowerCase() === lower || r.memberPubkey?.toLowerCase() === lower;
});
recipientOnline = !!match;
recipientName = match ? (match as { displayName?: string }).displayName : undefined;
}
}
const offlineHint =
"Session pubkeys are ephemeral — a key from an ended session never reconnects, so the message can't be delivered. Re-fetch a live target with `claudemesh peer list --json`.";
// Daemon path — preferred when a long-lived daemon is local. UDS at // Daemon path — preferred when a long-lived daemon is local. UDS at
// ~/.claudemesh/daemon/daemon.sock; ~1ms round-trip; persists outbox // ~/.claudemesh/daemon/daemon.sock; ~1ms round-trip; persists outbox
// across CLI invocations so a `claudemesh send` survives a daemon // across CLI invocations so a `claudemesh send` survives a daemon
@@ -192,8 +270,18 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
const dr = await trySendViaDaemon({ to, message, priority, expectedMesh: meshSlug ?? undefined }); const dr = await trySendViaDaemon({ to, message, priority, expectedMesh: meshSlug ?? undefined });
if (dr !== null) { if (dr !== null) {
if (dr.ok) { if (dr.ok) {
if (flags.json) console.log(JSON.stringify({ ok: true, messageId: dr.messageId, target: to, via: "daemon", duplicate: !!dr.duplicate })); if (flags.json) {
else render.ok(`sent to ${to} (daemon)`, dr.messageId ? dim(dr.messageId.slice(0, 8)) : undefined); console.log(JSON.stringify({ ok: true, messageId: dr.messageId, target: to, via: "daemon", duplicate: !!dr.duplicate, status: dr.status, recipientOnline }));
} else if (recipientOnline === false) {
render.warn(`queued for ${recipientName ?? to.slice(0, 16) + "…"} — no connected peer matches this key on "${meshSlug}".`);
render.hint(offlineHint);
} else {
const who = recipientName ? `${recipientName} (${to.slice(0, 16)}…)` : to;
// recipientOnline === true → peer is present, delivery imminent.
// null → daemon couldn't tell (e.g. roster query failed); keep
// the neutral "(daemon)" transport tag rather than overclaiming.
render.ok(`sent to ${who}${recipientOnline === true ? " (online)" : " (daemon)"}`, dr.messageId ? dim(dr.messageId.slice(0, 8)) : undefined);
}
return; return;
} }
// Daemon answered but rejected (409 idempotency, 400 schema). Surface; do not fall through. // Daemon answered but rejected (409 idempotency, 400 schema). Surface; do not fall through.
@@ -206,8 +294,10 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
// was removed in 1.28.0. // was removed in 1.28.0.
} }
// Cold path — open our own WS, encrypt locally, fire envelope. // Cold path — open our own WS, encrypt locally, fire envelope. Use the
await withMesh({ meshSlug: flags.mesh ?? null }, async (client) => { // resolved meshSlug (may have been discovered above) so a name/prefix
// that lives on a non-default mesh still targets the right one.
await withMesh({ meshSlug: meshSlug ?? flags.mesh ?? null }, async (client) => {
let targetSpec = to; let targetSpec = to;
if (to.startsWith("#") && !/^#[0-9a-z_-]{20,}$/i.test(to)) { if (to.startsWith("#") && !/^#[0-9a-z_-]{20,}$/i.test(to)) {
// Topic by name → resolve to "#<topicId>" via topicList. The broker // Topic by name → resolve to "#<topicId>" via topicList. The broker
@@ -237,9 +327,13 @@ export async function runSend(flags: SendFlags, to: string, message: string): Pr
const result = await client.send(targetSpec, message, priority); const result = await client.send(targetSpec, message, priority);
if (result.ok) { if (result.ok) {
if (flags.json) { if (flags.json) {
console.log(JSON.stringify({ ok: true, messageId: result.messageId, target: to })); console.log(JSON.stringify({ ok: true, messageId: result.messageId, target: to, recipientOnline }));
} else if (recipientOnline === false) {
render.warn(`queued for ${recipientName ?? to} — no connected peer matches this key on "${meshSlug ?? flags.mesh ?? "default"}".`);
render.hint(offlineHint);
} else { } else {
render.ok(`sent to ${to}`, result.messageId ? dim(result.messageId.slice(0, 8)) : undefined); const who = recipientName ? `${recipientName} (${to.slice(0, 16)}…)` : to;
render.ok(`sent to ${who}${recipientOnline === true ? " (online)" : ""}`, result.messageId ? dim(result.messageId.slice(0, 8)) : undefined);
} }
} else { } else {
if (flags.json) { if (flags.json) {

View File

@@ -31,6 +31,11 @@ export const DAEMON_PATHS = {
get OUTBOX_DB() { return join(this.DAEMON_DIR, "outbox.db"); }, get OUTBOX_DB() { return join(this.DAEMON_DIR, "outbox.db"); },
get INBOX_DB() { return join(this.DAEMON_DIR, "inbox.db"); }, get INBOX_DB() { return join(this.DAEMON_DIR, "inbox.db"); },
get LOG_FILE() { return join(this.DAEMON_DIR, "daemon.log"); }, get LOG_FILE() { return join(this.DAEMON_DIR, "daemon.log"); },
/** 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
* secrets — keypairs are reloaded from the per-session keypair store. */
get SESSIONS_FILE() { return join(this.DAEMON_DIR, "sessions.json"); },
} as const; } as const;
export const DAEMON_TCP_HOST = "127.0.0.1"; export const DAEMON_TCP_HOST = "127.0.0.1";

View File

@@ -4,7 +4,7 @@ import { DAEMON_PATHS } from "./paths.js";
import { acquireSingletonLock, releaseSingletonLock } from "./lock.js"; import { acquireSingletonLock, releaseSingletonLock } from "./lock.js";
import { ensureLocalToken } from "./local-token.js"; import { ensureLocalToken } from "./local-token.js";
import { startIpcServer } from "./ipc/server.js"; import { startIpcServer } from "./ipc/server.js";
import { setRegistryHooks, startReaper, type SessionInfo } from "./session-registry.js"; import { setRegistryHooks, startReaper, registerSession, readPersistedSessions, setRegistryPersistence, type SessionInfo } from "./session-registry.js";
import { openSqlite, type SqliteDb } from "./db/sqlite.js"; import { openSqlite, type SqliteDb } from "./db/sqlite.js";
import { migrateOutbox } from "./db/outbox.js"; import { migrateOutbox } from "./db/outbox.js";
import { migrateInbox } from "./db/inbox.js"; import { migrateInbox } from "./db/inbox.js";
@@ -308,6 +308,81 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise<number> {
startReaper(); startReaper();
// Rehydrate persisted session bindings (1.36.0). A daemon restart used
// to wipe the in-memory registry, so every live session lost its mesh
// context and CLI commands fell back to an arbitrary default mesh — a
// live peer then looked "disconnected" though nothing had moved. We now
// reload each persisted binding, validate the pid is still alive (with
// a start-time PID-reuse guard), reload its keypair from the per-session
// store, re-sign a fresh parent attestation, and re-register it — which
// fires onRegister and reconnects its SessionBrokerClient on the broker.
try {
const persisted = readPersistedSessions(DAEMON_PATHS.SESSIONS_FILE);
if (persisted.length > 0) {
const { loadOrCreateSessionKeypair } = await import("~/services/session/keypair-store.js");
const { signParentAttestation } = await import("~/services/broker/session-hello-sig.js");
const { isPidAlive, getProcessStartTimes } = await import("./process-info.js");
const liveStartTimes = await getProcessStartTimes(persisted.map((p) => p.pid)).catch(() => new Map<number, string>());
let revived = 0;
for (const s of persisted) {
if (!isPidAlive(s.pid)) continue;
if (s.startTime !== undefined) {
const live = liveStartTimes.get(s.pid);
if (live !== undefined && live !== s.startTime) continue; // PID reused
}
const meshConfig = meshConfigs.get(s.mesh);
if (!meshConfig) continue; // mesh no longer joined
try {
const kp = await loadOrCreateSessionKeypair(meshConfig.slug, s.sessionId);
const att = await signParentAttestation({
parentMemberPubkey: meshConfig.pubkey,
parentSecretKey: meshConfig.secretKey,
sessionPubkey: kp.publicKey,
});
registerSession({
token: s.token,
sessionId: s.sessionId,
mesh: s.mesh,
displayName: s.displayName,
pid: s.pid,
...(s.cwd ? { cwd: s.cwd } : {}),
...(s.role ? { role: s.role } : {}),
...(s.groups ? { groups: s.groups } : {}),
...(s.startTime ? { startTime: s.startTime } : {}),
presence: {
sessionPubkey: kp.publicKey,
sessionSecretKey: kp.secretKey,
parentAttestation: {
sessionPubkey: att.sessionPubkey,
parentMemberPubkey: att.parentMemberPubkey,
expiresAt: att.expiresAt,
signature: att.signature,
},
},
});
revived++;
} catch (err) {
process.stderr.write(JSON.stringify({
level: "warn", msg: "session_rehydrate_failed",
token: s.token.slice(0, 8), mesh: s.mesh, err: String(err),
ts: new Date().toISOString(),
}) + "\n");
}
}
process.stderr.write(JSON.stringify({
level: "info", msg: "sessions_rehydrated",
revived, persisted: persisted.length, ts: new Date().toISOString(),
}) + "\n");
}
} catch (err) {
process.stderr.write(JSON.stringify({
level: "warn", msg: "session_rehydrate_scan_failed", err: String(err),
ts: new Date().toISOString(),
}) + "\n");
}
// Enable ongoing persistence now that rehydration has read the old file.
setRegistryPersistence(DAEMON_PATHS.SESSIONS_FILE);
const ipc = startIpcServer({ const ipc = startIpcServer({
localToken, localToken,
tcpEnabled, tcpEnabled,

View File

@@ -43,7 +43,7 @@
import { hostname as osHostname } from "node:os"; import { hostname as osHostname } from "node:os";
import type { JoinedMesh } from "~/services/config/facade.js"; import type { JoinedMesh } from "~/services/config/facade.js";
import { signSessionHello } from "~/services/broker/session-hello-sig.js"; import { signSessionHello, signParentAttestation } from "~/services/broker/session-hello-sig.js";
import { connectWsWithBackoff, type WsLifecycle, type WsStatus } from "./ws-lifecycle.js"; import { connectWsWithBackoff, type WsLifecycle, type WsStatus } from "./ws-lifecycle.js";
import type { BrokerSendArgs, BrokerSendResult } from "./broker.js"; import type { BrokerSendArgs, BrokerSendResult } from "./broker.js";
@@ -149,13 +149,35 @@ export class SessionBrokerClient {
sessionPubkey: this.opts.sessionPubkey, sessionPubkey: this.opts.sessionPubkey,
sessionSecretKey: this.opts.sessionSecretKey, sessionSecretKey: this.opts.sessionSecretKey,
}); });
// Re-mint the parent attestation fresh on every (re)connect rather
// than reusing the one signed at `claudemesh launch`. The minted
// attestation has a 12h TTL; reusing the stored instance meant any
// reconnect past launch+12h — a network blip, a sleep/wake, or
// (most commonly) a broker redeploy that drops every WS at once —
// was rejected by the broker with `expired`, after which the daemon
// reconnect-looped forever with the same dead token and the session
// silently fell off the mesh (its ephemeral pubkey lingering in
// peer rosters, undeliverable). The member secret key is in memory
// (`mesh.secretKey`, already used at daemon rehydration), so the
// daemon can self-renew: fresh-minting keeps live attestations
// short-lived AND makes presence self-healing across reconnects.
let parentAttestation = this.opts.parentAttestation;
try {
parentAttestation = await signParentAttestation({
parentMemberPubkey: this.opts.mesh.pubkey,
parentSecretKey: this.opts.mesh.secretKey,
sessionPubkey: this.opts.sessionPubkey,
});
} catch (e) {
this.log("warn", "parent attestation re-mint failed; reusing stored token (may be expired)", { err: String(e) });
}
return { return {
type: "session_hello", type: "session_hello",
meshId: this.opts.mesh.meshId, meshId: this.opts.mesh.meshId,
parentMemberId: this.opts.mesh.memberId, parentMemberId: this.opts.mesh.memberId,
parentMemberPubkey: this.opts.mesh.pubkey, parentMemberPubkey: this.opts.mesh.pubkey,
sessionPubkey: this.opts.sessionPubkey, sessionPubkey: this.opts.sessionPubkey,
parentAttestation: this.opts.parentAttestation, parentAttestation,
displayName: this.opts.displayName, displayName: this.opts.displayName,
sessionId: this.opts.sessionId, sessionId: this.opts.sessionId,
pid: this.opts.pid, pid: this.opts.pid,

View File

@@ -22,6 +22,10 @@
* session have no token to begin with. * session have no token to begin with.
*/ */
import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
import { dirname } from "node:path";
import { randomBytes } from "node:crypto";
import { getProcessStartTime, getProcessStartTimes, isPidAlive } from "./process-info.js"; import { getProcessStartTime, getProcessStartTimes, isPidAlive } from "./process-info.js";
/** /**
@@ -83,6 +87,65 @@ const hooks: RegistryHooks = {};
let reaperHandle: NodeJS.Timeout | null = null; let reaperHandle: NodeJS.Timeout | null = null;
/** When set, registry mutations are mirrored to this file so a daemon
* restart can rehydrate live sessions. Holds NO secret material — the
* session keypair is reloaded from the per-session keypair store on
* rehydrate. null (default) disables persistence, which keeps unit
* tests from touching disk unless they opt in. */
let persistPath: string | null = null;
/** Slim, secret-free projection persisted to disk. */
export interface PersistedSession {
token: string;
sessionId: string;
mesh: string;
displayName: string;
pid: number;
cwd?: string;
role?: string;
groups?: string[];
startTime?: string;
registeredAt: number;
}
function toPersisted(info: SessionInfo): PersistedSession {
// Drop `presence` (carries the session secret key) — never to disk here.
const { presence: _presence, ...rest } = info;
return rest;
}
/** Enable on-disk persistence of session bindings (called at daemon boot
* with DAEMON_PATHS.SESSIONS_FILE). Pass null to disable. */
export function setRegistryPersistence(path: string | null): void {
persistPath = path;
}
function persist(): void {
if (!persistPath) return;
try {
mkdirSync(dirname(persistPath), { recursive: true, mode: 0o700 });
const rows = [...byToken.values()].map(toPersisted);
const tmp = `${persistPath}.${randomBytes(6).toString("hex")}.tmp`;
writeFileSync(tmp, JSON.stringify({ version: 1, sessions: rows }), { mode: 0o600 });
renameSync(tmp, persistPath);
} catch {
// Best-effort: a persistence failure must never throttle the registry.
}
}
/** Read persisted session bindings from disk (pure — no registration, no
* liveness check). Returns [] when the file is absent or unreadable.
* The daemon's boot rehydration validates liveness and re-registers. */
export function readPersistedSessions(path: string): PersistedSession[] {
try {
if (!existsSync(path)) return [];
const parsed = JSON.parse(readFileSync(path, "utf8")) as { sessions?: PersistedSession[] };
return Array.isArray(parsed.sessions) ? parsed.sessions : [];
} catch {
return [];
}
}
export function startReaper(): void { export function startReaper(): void {
if (reaperHandle) return; if (reaperHandle) return;
// The sweep is async (batched ps) — wrap in `void` so setInterval // The sweep is async (batched ps) — wrap in `void` so setInterval
@@ -125,6 +188,7 @@ export function registerSession(info: Omit<SessionInfo, "registeredAt">): Sessio
const stored: SessionInfo = { ...info, registeredAt: Date.now() }; const stored: SessionInfo = { ...info, registeredAt: Date.now() };
byToken.set(info.token, stored); byToken.set(info.token, stored);
bySessionId.set(info.sessionId, info.token); bySessionId.set(info.sessionId, info.token);
persist();
try { hooks.onRegister?.(stored); } catch { /* see above */ } try { hooks.onRegister?.(stored); } catch { /* see above */ }
if (stored.startTime === undefined) { if (stored.startTime === undefined) {
void captureStartTimeAsync(info.token, info.pid); void captureStartTimeAsync(info.token, info.pid);
@@ -138,6 +202,7 @@ async function captureStartTimeAsync(token: string, pid: number): Promise<void>
const entry = byToken.get(token); const entry = byToken.get(token);
if (!entry || entry.pid !== pid) return; // entry was replaced; skip if (!entry || entry.pid !== pid) return; // entry was replaced; skip
entry.startTime = lstart; entry.startTime = lstart;
persist(); // capture start-time on disk so restart can PID-reuse-guard
} }
export function deregisterByToken(token: string): boolean { export function deregisterByToken(token: string): boolean {
@@ -145,6 +210,7 @@ export function deregisterByToken(token: string): boolean {
if (!entry) return false; if (!entry) return false;
byToken.delete(token); byToken.delete(token);
if (bySessionId.get(entry.sessionId) === token) bySessionId.delete(entry.sessionId); if (bySessionId.get(entry.sessionId) === token) bySessionId.delete(entry.sessionId);
persist();
try { hooks.onDeregister?.(entry); } catch { /* see above */ } try { hooks.onDeregister?.(entry); } catch { /* see above */ }
return true; return true;
} }
@@ -217,4 +283,5 @@ export function _resetRegistry(): void {
bySessionId.clear(); bySessionId.clear();
hooks.onRegister = undefined; hooks.onRegister = undefined;
hooks.onDeregister = undefined; hooks.onDeregister = undefined;
persistPath = null;
} }

View File

@@ -0,0 +1,100 @@
/**
* Session-registry persistence (1.36.0) — durable session→mesh bindings.
*
* A daemon restart used to wipe the in-memory registry, orphaning every
* live session's mesh context. Persistence lets the daemon rehydrate on
* boot. Verifies:
* - register writes a slim record to disk; readPersistedSessions reads it;
* - the session SECRET KEY is never written to disk;
* - deregister removes the record;
* - persistence is off by default (no disk writes until enabled).
*/
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import {
_resetRegistry,
deregisterByToken,
readPersistedSessions,
registerSession,
setRegistryPersistence,
} from "../../src/daemon/session-registry.js";
const SECRET = "b".repeat(128);
const PRESENCE = {
sessionPubkey: "a".repeat(64),
sessionSecretKey: SECRET,
parentAttestation: {
sessionPubkey: "a".repeat(64),
parentMemberPubkey: "c".repeat(64),
expiresAt: 9_999_999_999,
signature: "d".repeat(128),
},
};
let dir: string;
let file: string;
beforeEach(() => {
_resetRegistry();
dir = mkdtempSync(join(tmpdir(), "cm-reg-"));
file = join(dir, "sessions.json");
});
afterEach(() => {
_resetRegistry();
rmSync(dir, { recursive: true, force: true });
});
describe("registry persistence", () => {
test("off by default — no disk writes until enabled", () => {
registerSession({ token: "t1", sessionId: "s1", mesh: "flexicar", displayName: "a", pid: process.pid, startTime: "x" });
expect(existsSync(file)).toBe(false);
});
test("register persists a slim record; readPersistedSessions round-trips", () => {
setRegistryPersistence(file);
registerSession({
token: "t1", sessionId: "11111111-2222-3333-4444-555555555555",
mesh: "flexicar", displayName: "intra-back", pid: process.pid,
cwd: "/tmp/x", role: "dev", startTime: "x", presence: PRESENCE,
});
const rows = readPersistedSessions(file);
expect(rows).toHaveLength(1);
expect(rows[0]).toMatchObject({
token: "t1", mesh: "flexicar", displayName: "intra-back", cwd: "/tmp/x", role: "dev",
});
});
test("session secret key is NEVER written to disk", () => {
setRegistryPersistence(file);
registerSession({ token: "t1", sessionId: "s1", mesh: "flexicar", displayName: "a", pid: process.pid, startTime: "x", presence: PRESENCE });
const raw = readFileSync(file, "utf8");
expect(raw).not.toContain(SECRET);
expect(raw).not.toContain("sessionSecretKey");
expect(raw).not.toContain("parentAttestation");
// And the parsed record carries no presence material.
expect(readPersistedSessions(file)[0]).not.toHaveProperty("presence");
});
test("deregister removes the record from disk", () => {
setRegistryPersistence(file);
registerSession({ token: "t1", sessionId: "s1", mesh: "flexicar", displayName: "a", pid: process.pid, startTime: "x" });
registerSession({ token: "t2", sessionId: "s2", mesh: "nedas", displayName: "b", pid: process.pid, startTime: "x" });
expect(readPersistedSessions(file)).toHaveLength(2);
deregisterByToken("t1");
const rows = readPersistedSessions(file);
expect(rows).toHaveLength(1);
expect(rows[0]!.token).toBe("t2");
});
test("readPersistedSessions tolerates a missing/corrupt file", () => {
expect(readPersistedSessions(join(dir, "nope.json"))).toEqual([]);
const bad = join(dir, "bad.json");
writeFileSync(bad, "{not json");
expect(readPersistedSessions(bad)).toEqual([]);
});
});