From ba4f9df474fdd965bbd6293c152d1508bbd369bd Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:21:21 +0100 Subject: [PATCH 01/11] feat(core): add read-only slot probing and lock-free identity loading probeSlotOwner reads a slot's lock file without writing to it, and loadIdentityForFront loads or creates the slot's persisted identity without taking the exclusivity lock. Together these let a caller assume a not-yet-live (harness, cwd) slot's future identity while leaving the lock free for that slot's real owner to acquire normally later. --- src/core/identity-store.ts | 21 +++++++++++++ src/test/identity-store.test.ts | 56 +++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/src/core/identity-store.ts b/src/core/identity-store.ts index 4f705c6b..db47c030 100644 --- a/src/core/identity-store.ts +++ b/src/core/identity-store.ts @@ -279,6 +279,27 @@ function createIdentity(identityFile: string): PeerIdentity { return identity; } +/** + * Probes a slot's current lock holder without taking it -- the cc-peer front (agent-comms#157) uses this read-only check to decide whether a local Claude Code session already fronts itself (a live agent-comms bridge already holds the slot) before attaching, and to notice a session's own bridge taking the slot over later so the front can yield. Returns the live PID currently holding the slot's lock, or undefined when the slot is unlocked or its recorded holder is no longer alive. + */ +export function probeSlotOwner(slot: Readonly): number | undefined { + const { lockFile } = slotPaths(slot); + const heldBy = readLockPid(lockFile); + if (heldBy === undefined) return undefined; + return isPidAlive(heldBy) ? heldBy : undefined; +} + +/** + * Loads (creating on first use) the persisted identity for a slot without taking its exclusivity lock. The cc-peer front (agent-comms#157) uses this to assume a not-yet-live session's own future identity, so the device-id -- and with it every agent id, room membership, and pending delivery already addressed to it -- carries over unchanged the moment that session's own bridge starts and claims the slot for real. Callers must have already confirmed via probeSlotOwner that no live bridge currently holds the slot; re-probing atomically against a concurrent acquisition isn't possible across the two separate files (identity vs lock) this store keeps, so it's the front's own periodic re-probe, not this function, that detects and reacts to a real bridge taking over afterwards. + */ +export function loadIdentityForFront( + slot: Readonly, +): PeerIdentity { + const { dir, identityFile } = slotPaths(slot); + fs.mkdirSync(dir, { recursive: true, mode: 0o700 }); + return loadStoredIdentity(identityFile) ?? createIdentity(identityFile); +} + /** * Release the slot lock on graceful shutdown. A lock held by another PID (taken over after this process crashed and restarted) is left alone. */ diff --git a/src/test/identity-store.test.ts b/src/test/identity-store.test.ts index 4b41a058..f190b716 100644 --- a/src/test/identity-store.test.ts +++ b/src/test/identity-store.test.ts @@ -16,6 +16,8 @@ import { deleteGroupToken, loadGroupTokens, loadOrCreateIdentity, + loadIdentityForFront, + probeSlotOwner, oplogDirFor, releaseIdentityLock, saveGroupToken, @@ -188,6 +190,60 @@ test("a corrupt identity file is regenerated", () => { releaseIdentityLock(slot); }); +test("probeSlotOwner reports undefined for a never-touched slot", () => { + const { slot } = tempSlot("claude-code"); + expect(probeSlotOwner(slot)).toBeUndefined(); +}); + +test("probeSlotOwner reports the live PID holding a slot's lock", () => { + const { slot } = tempSlot("claude-code"); + loadOrCreateIdentity(slot); + expect(probeSlotOwner(slot)).toBe(process.pid); + releaseIdentityLock(slot); +}); + +test("probeSlotOwner reports undefined once the lock is released", () => { + const { slot } = tempSlot("claude-code"); + loadOrCreateIdentity(slot); + releaseIdentityLock(slot); + expect(probeSlotOwner(slot)).toBeUndefined(); +}); + +test("probeSlotOwner reports undefined for a lock left by a dead process", () => { + const { slot, dir } = tempSlot("claude-code"); + loadOrCreateIdentity(slot); + const lockFile = slotFile(dir, ".lock"); + releaseIdentityLock(slot); + // A pid that is exceedingly unlikely to be alive on any real machine, standing in for a crashed process's stale lock file (matching the "a stale lock from a dead process" test above, which relies on the same never-recycled-in-practice assumption). + const deadPid = 999_999; + fs.writeFileSync(lockFile, `${String(deadPid)}\n`); + expect(probeSlotOwner(slot)).toBeUndefined(); +}); + +test("loadIdentityForFront creates and persists an identity without taking the slot's lock", () => { + const { slot, dir } = tempSlot("claude-code"); + const identity = loadIdentityForFront(slot); + expect(fs.existsSync(slotFile(dir, ".json"))).toBe(true); + expect(fs.existsSync(path.join(dir, "identity-claude-code--_tmp_project.lock"))).toBe( + false, + ); + expect(probeSlotOwner(slot)).toBeUndefined(); + + const reloaded = loadIdentityForFront(slot); + expect(reloaded.fingerprint).toBe(identity.fingerprint); + expect(reloaded.deviceId).toEqual(identity.deviceId); +}); + +test("loadIdentityForFront's identity matches what loadOrCreateIdentity would later assume for the same slot", () => { + const { slot } = tempSlot("claude-code"); + const front = loadIdentityForFront(slot); + + const owned = loadOrCreateIdentity(slot); + expect(owned.fingerprint).toBe(front.fingerprint); + expect(owned.deviceId).toEqual(front.deviceId); + releaseIdentityLock(slot); +}); + test("oplogDirFor is a sibling directory of the identity file, distinct per (harness, cwd)", () => { const { slot, dir } = tempSlot("pi"); const oplogDir = oplogDirFor(slot); From 32a5b812f94075fcea3ea21a68deba569ef78e5b Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:25:01 +0100 Subject: [PATCH 02/11] style(core): satisfy prettier formatting for the new identity-store tests Wraps probeSlotOwner's own parameter list and a multi-line assertion in the identity-store test suite onto prettier's expected line breaks. --- src/core/identity-store.ts | 4 +++- src/test/identity-store.test.ts | 6 +++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/src/core/identity-store.ts b/src/core/identity-store.ts index db47c030..12008316 100644 --- a/src/core/identity-store.ts +++ b/src/core/identity-store.ts @@ -282,7 +282,9 @@ function createIdentity(identityFile: string): PeerIdentity { /** * Probes a slot's current lock holder without taking it -- the cc-peer front (agent-comms#157) uses this read-only check to decide whether a local Claude Code session already fronts itself (a live agent-comms bridge already holds the slot) before attaching, and to notice a session's own bridge taking the slot over later so the front can yield. Returns the live PID currently holding the slot's lock, or undefined when the slot is unlocked or its recorded holder is no longer alive. */ -export function probeSlotOwner(slot: Readonly): number | undefined { +export function probeSlotOwner( + slot: Readonly, +): number | undefined { const { lockFile } = slotPaths(slot); const heldBy = readLockPid(lockFile); if (heldBy === undefined) return undefined; diff --git a/src/test/identity-store.test.ts b/src/test/identity-store.test.ts index f190b716..2555e97e 100644 --- a/src/test/identity-store.test.ts +++ b/src/test/identity-store.test.ts @@ -224,9 +224,9 @@ test("loadIdentityForFront creates and persists an identity without taking the s const { slot, dir } = tempSlot("claude-code"); const identity = loadIdentityForFront(slot); expect(fs.existsSync(slotFile(dir, ".json"))).toBe(true); - expect(fs.existsSync(path.join(dir, "identity-claude-code--_tmp_project.lock"))).toBe( - false, - ); + expect( + fs.existsSync(path.join(dir, "identity-claude-code--_tmp_project.lock")), + ).toBe(false); expect(probeSlotOwner(slot)).toBeUndefined(); const reloaded = loadIdentityForFront(slot); From 461ff4d8308f13dcde245d0ecf01d27d58b8900f Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:25:03 +0100 Subject: [PATCH 03/11] refactor(core): split bridge-mesh construction from identity loading createBridgeMeshSyncFromIdentity and createBridgeMeshFromIdentity build a mesh from an already-loaded identity, letting createBridgeMeshSync/createBridgeMesh keep owning loadOrCreateIdentity's lock while the cc-peer front (agent-comms#157) builds a mesh identity from loadIdentityForFront's lock-free one instead. Also exports the identity-store slot helpers from core/index.ts so bridge code outside core/ can reach them the same way it already reaches everything else in this factory. --- src/core/bridge-mesh.ts | 42 +++++++++++++++++++++++++++++++++++- src/core/index.ts | 13 ++++++++++- src/test/bridge-mesh.test.ts | 24 ++++++++++++++++++++- 3 files changed, 76 insertions(+), 3 deletions(-) diff --git a/src/core/bridge-mesh.ts b/src/core/bridge-mesh.ts index b4ba7672..2aad5909 100644 --- a/src/core/bridge-mesh.ts +++ b/src/core/bridge-mesh.ts @@ -6,6 +6,8 @@ * Split into a synchronous half (wireBridgeMesh) and an async half (attachMintingIdentity, wrapping toIdentityPort's WebCrypto import) because deriving the IdentityPort MeshStore mints room-membership grants against is unavoidably async, but not every bridge entry point can await one inline -- a plugin loader that calls its extension's default export synchronously (e.g. pi's own) cannot. createBridgeMeshSync exposes both halves for that case, deferring attachIdentity() to wherever the bridge's own lifecycle first has an async context (its own session-start hook), which is always well before the bridge does anything identity-dependent like createRoom. createBridgeMesh remains the convenient all-in-one for every bridge whose own entry point is already async. * * Also starts this bridge's own VersionDriftChecker (agent-comms#166) and wires its result into the CommsTool it builds, so every real bridge gets npm release-drift reporting on whoami/update for free from this one construction point, with no per-bridge wiring. fetchLatestVersion is exposed purely for tests -- every real caller omits it and gets VersionDriftChecker's own default (a real npm registry lookup); a test that would otherwise trigger real network I/O on every createBridgeMesh call injects a fake resolver instead. + * + * createBridgeMeshSync/createBridgeMesh own loadOrCreateIdentity's slot lock on the caller's behalf; createBridgeMeshSyncFromIdentity/createBridgeMeshFromIdentity take an already-loaded identity instead and never touch the lock at all -- the cc-peer front (agent-comms#157) uses these directly, via loadIdentityForFront's lock-free load, to build a mesh identity for a not-yet-live session's slot while leaving that slot's own lock free for its real bridge to acquire normally later. */ import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; @@ -18,6 +20,7 @@ import { WireMeshTransport } from "./wire-mesh-transport.js"; import { loadOrCreateIdentity, oplogDirFor } from "./identity-store.js"; import type { IdentitySlot } from "./identity-store.js"; import { toIdentityPort } from "./wire-mesh-identity.js"; +import type { PeerIdentity } from "./identity.js"; import { VersionDriftChecker } from "./version-check.js"; import { getOwnPackageVersion } from "./package-version.js"; @@ -38,7 +41,25 @@ export function createBridgeMeshSync( hubUrl?: string, fetchLatestVersion?: () => Promise, ): BridgeMeshSync { - const identity = loadOrCreateIdentity(slot); + return createBridgeMeshSyncFromIdentity( + loadOrCreateIdentity(slot), + slot, + coordinatorPort, + hubUrl, + fetchLatestVersion, + ); +} + +/** + * The same synchronous construction as createBridgeMeshSync, but taking an already-loaded identity rather than calling loadOrCreateIdentity itself -- see this file's own header comment for who this is for. Every other caller should go through createBridgeMeshSync/createBridgeMesh above, which own the lock on the caller's behalf. + */ +export function createBridgeMeshSyncFromIdentity( + identity: PeerIdentity, + slot: Readonly, + coordinatorPort?: number, + hubUrl?: string, + fetchLatestVersion?: () => Promise, +): BridgeMeshSync { const store = new MeshStore(coordinatorPort, hubUrl); store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); // One shared dataStorage instance for both the transport's own data-domain frame responder and the store's own durable-send mint path (P5, agent-comms#50) -- oplogDirFor(slot) needs only the slot, not the async identity below, so this can be constructed synchronously right here. @@ -95,3 +116,22 @@ export async function createBridgeMesh( await attachIdentity(); return { store, tool }; } + +/** The async, already-loaded-identity counterpart to createBridgeMesh, mirroring createBridgeMeshSyncFromIdentity's relationship to createBridgeMeshSync. */ +export async function createBridgeMeshFromIdentity( + identity: PeerIdentity, + slot: Readonly, + coordinatorPort?: number, + hubUrl?: string, + fetchLatestVersion?: () => Promise, +): Promise { + const { store, tool, attachIdentity } = createBridgeMeshSyncFromIdentity( + identity, + slot, + coordinatorPort, + hubUrl, + fetchLatestVersion, + ); + await attachIdentity(); + return { store, tool }; +} diff --git a/src/core/index.ts b/src/core/index.ts index 708de2eb..53f9d290 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -9,8 +9,19 @@ export type { CommsStore } from "./comms-store.js"; export { FileStore, CommsError } from "./store.js"; export { MeshStore } from "./mesh-store.js"; export { CommsTool } from "./tool.js"; -export { createBridgeMesh, createBridgeMeshSync } from "./bridge-mesh.js"; +export { + createBridgeMesh, + createBridgeMeshSync, + createBridgeMeshFromIdentity, + createBridgeMeshSyncFromIdentity, +} from "./bridge-mesh.js"; export type { BridgeMesh, BridgeMeshSync } from "./bridge-mesh.js"; +export { + loadIdentityForFront, + probeSlotOwner, + releaseIdentityLock, +} from "./identity-store.js"; +export type { IdentitySlot } from "./identity-store.js"; export type { CommsContext, CommsResult } from "./tool.js"; export { buildAction, diff --git a/src/test/bridge-mesh.test.ts b/src/test/bridge-mesh.test.ts index 9ade7844..00caf4a0 100644 --- a/src/test/bridge-mesh.test.ts +++ b/src/test/bridge-mesh.test.ts @@ -7,9 +7,14 @@ import { tmpdir } from "node:os"; import path from "node:path"; import { test, expect } from "vitest"; import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; -import { createBridgeMesh } from "../core/bridge-mesh.js"; +import { + createBridgeMesh, + createBridgeMeshFromIdentity, +} from "../core/bridge-mesh.js"; import { loadOrCreateIdentity, + loadIdentityForFront, + probeSlotOwner, type IdentitySlot, } from "../core/identity-store.js"; import { waitFor } from "./test-transport.js"; @@ -84,6 +89,23 @@ test("createBridgeMesh passes an explicit coordinatorPort through to MeshStore, } }); +test("createBridgeMeshFromIdentity wires the given identity's own device-id as peerId, without taking the slot's lock", async () => { + const slot = tempSlot("cc-peer-front"); + const identity = loadIdentityForFront(slot); + expect(probeSlotOwner(slot)).toBeUndefined(); + + const { store } = await createBridgeMeshFromIdentity(identity, slot); + try { + expect(store.peerId).toBe( + deviceIdToHex(Uint8Array.from(identity.deviceId)), + ); + // Constructing a mesh from a lock-free identity must not itself take the lock -- the whole point is leaving it free for the slot's real owner to acquire normally later (agent-comms#157). + expect(probeSlotOwner(slot)).toBeUndefined(); + } finally { + await store.shutdown(); + } +}); + test("createBridgeMesh passes an explicit hubUrl through to MeshStore, dialled once the store becomes coordinator and dropped on shutdown", async () => { const hub = await realHubOverWs(); const slot = tempSlot("test-harness-hub"); From 1cdb4b4ee90769f53dd5531092978a5690f899df Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:26:38 +0100 Subject: [PATCH 04/11] feat(cc-peer): add pure roster-selection and inbound-routing logic for the default front selectSessionsToFront filters a cc-peer roster down to the real Claude Code sessions the front should attach to: excludes other cc-peer-library peers (identified by cc-peer's own literal "cc-peer" version marker) and any session whose own identity slot is already held by a live bridge. matchInboundMessageSession attributes an inbound cc-peer message to the fronted session it came from, by matching the envelope's uds: convention against each fronted session's own messagingSocketPath. Kept free of any real cc-peer/MeshStore construction so the decision logic is testable purely over data; front-runtime.ts (agent-comms#157, still to come) builds the real stateful controller on top of these functions. --- src/bridges/cc-peer/front.ts | 59 +++++++++++++ src/test/cc-peer-front.test.ts | 146 +++++++++++++++++++++++++++++++++ 2 files changed, 205 insertions(+) create mode 100644 src/bridges/cc-peer/front.ts create mode 100644 src/test/cc-peer-front.test.ts diff --git a/src/bridges/cc-peer/front.ts b/src/bridges/cc-peer/front.ts new file mode 100644 index 00000000..18eae005 --- /dev/null +++ b/src/bridges/cc-peer/front.ts @@ -0,0 +1,59 @@ +/** + * Pure decision logic for the default cc-peer front (agent-comms#157): a machine's coordinator bridge relays every local Claude Code session that doesn't already front itself, using the same (harness, cwd) identity slot that session's own agent-comms bridge would use, so addressing carries over unchanged the moment that session's own bridge appears. + * + * Kept free of any real cc-peer/MeshStore construction so it's testable without a real local Claude Code session, a real filesystem, or real sockets -- see front-runtime.ts for the stateful controller and the real wiring built on top of these functions. + */ + +import type { IdentitySlot } from "../../core/identity-store.js"; + +/** The slice of cc-peer's own RegistryEntry this module needs. Narrowed rather than importing cc-peer's type directly so this file has no compile-time dependency on the cc-peer package (only front-runtime.ts, which does the real construction, needs that). */ +export interface CcPeerRosterEntryLike { + pid: number; + cwd: string; + name?: string; + /** cc-peer's own registry entries for library-backed peers (this front itself, the one-shot `bridge cc-peer` command, an alias-pool worker) always carry the literal string "cc-peer" here -- see buildRegistryEntry in cc-peer's own source. A real interactive Claude Code session, which registers itself natively rather than through the cc-peer library, reports its own Claude Code version instead. */ + version: string; + messagingSocketPath: string; +} + +/** True for a roster entry that is itself a cc-peer-library-backed peer (this front's own shared identity, the one-shot bridge command, an alias-pool worker) rather than a real interactive Claude Code session -- see CcPeerRosterEntryLike.version's own doc comment for the mechanism. The front must never try to front one of these: it would compute a spurious claude-code identity slot for a process that isn't a Claude Code session at all, and relay cc-peer traffic into it that nothing there ever reads. */ +export function isCcPeerLibraryPeer( + entry: Readonly, +): boolean { + return entry.version === "cc-peer"; +} + +/** The identity slot a session's own agent-comms bridge would hold if it started right now -- the same (harness, cwd) pair every real claude-code bridge entry point (bridges/claude-code/channel.ts) already constructs. */ +export function computeFrontSlot(cwd: string): IdentitySlot { + return { harness: "claude-code", cwd }; +} + +/** + * Filters a cc-peer roster down to the real Claude Code sessions this front should attach to right now: excludes other cc-peer-library peers, and excludes any session whose own slot is already held by a live PID (that session already fronts itself, via its own agent-comms bridge). probeSlotOwner is injected rather than imported directly so this stays a pure function over its inputs -- the real probe (identity-store.ts's probeSlotOwner) reads the filesystem, which has no place in a decision function tested purely over data. + */ +export function selectSessionsToFront( + roster: readonly CcPeerRosterEntryLike[], + probeSlotOwner: (slot: Readonly) => number | undefined, +): CcPeerRosterEntryLike[] { + return roster.filter((entry) => { + if (isCcPeerLibraryPeer(entry)) return false; + const slot = computeFrontSlot(entry.cwd); + return probeSlotOwner(slot) === undefined; + }); +} + +/** + * Finds which currently-fronted session an inbound cc-peer message came from, by matching the envelope's own `from` field (always the literal string "uds:" followed by the sender's own listening socket path, per cc-peer's own send() -- see CcPeer.send in cc-peer's source) against each fronted session's messagingSocketPath. Generic over the fronted-session record type so front-runtime.ts's real records (which carry a live MeshStore/CommsTool alongside the roster entry) can be matched directly without this module needing to know their shape. + */ +export function matchInboundMessageSession< + T extends { readonly messagingSocketPath: string }, +>( + fronted: Readonly>, + message: Readonly<{ from?: string }>, +): T | undefined { + if (message.from === undefined) return undefined; + for (const record of fronted) { + if (message.from === `uds:${record.messagingSocketPath}`) return record; + } + return undefined; +} diff --git a/src/test/cc-peer-front.test.ts b/src/test/cc-peer-front.test.ts new file mode 100644 index 00000000..66335c0e --- /dev/null +++ b/src/test/cc-peer-front.test.ts @@ -0,0 +1,146 @@ +/** + * Unit tests for the cc-peer front's pure decision logic (bridges/cc-peer/front.ts) -- roster filtering, slot computation, and inbound message routing, all tested purely over data with no real cc-peer/MeshStore/filesystem involved. + */ +import { describe, expect, it } from "vitest"; +import { + isCcPeerLibraryPeer, + computeFrontSlot, + selectSessionsToFront, + matchInboundMessageSession, + type CcPeerRosterEntryLike, +} from "../bridges/cc-peer/front.js"; + +function rosterEntry( + overrides: Readonly> = {}, +): CcPeerRosterEntryLike { + return { + pid: 111, + cwd: "/tmp/project", + version: "2.1.269", + messagingSocketPath: "/tmp/sock-111", + ...overrides, + }; +} + +/** A pid standing in for a real live agent-comms bridge holding a slot's lock -- arbitrary beyond "a plausible process id", used only to prove selectSessionsToFront treats "probe returned a pid" as "already fronted", not to assert anything about the value itself. */ +const LIVE_LOCK_HOLDER_PID = 4242; + +describe("isCcPeerLibraryPeer", () => { + it("is true for an entry carrying cc-peer's own literal version marker", () => { + expect(isCcPeerLibraryPeer(rosterEntry({ version: "cc-peer" }))).toBe(true); + }); + + it("is false for an entry reporting a real Claude Code version", () => { + expect(isCcPeerLibraryPeer(rosterEntry({ version: "2.1.269" }))).toBe( + false, + ); + }); +}); + +describe("computeFrontSlot", () => { + it("uses the claude-code harness, matching the real claude-code bridge's own slot", () => { + expect(computeFrontSlot("/tmp/project")).toEqual({ + harness: "claude-code", + cwd: "/tmp/project", + }); + }); + + it("is distinct per cwd", () => { + expect(computeFrontSlot("/tmp/a")).not.toEqual(computeFrontSlot("/tmp/b")); + }); +}); + +describe("selectSessionsToFront", () => { + it("excludes cc-peer-library peers regardless of slot state", () => { + const libraryPeer = rosterEntry({ pid: 1, version: "cc-peer" }); + const selected = selectSessionsToFront([libraryPeer], () => undefined); + expect(selected).toEqual([]); + }); + + it("excludes a real session whose own slot is already held by a live bridge", () => { + const alreadyFronted = rosterEntry({ pid: 2, cwd: "/tmp/self-fronted" }); + const selected = selectSessionsToFront([alreadyFronted], (slot) => + slot.cwd === "/tmp/self-fronted" ? LIVE_LOCK_HOLDER_PID : undefined, + ); + expect(selected).toEqual([]); + }); + + it("selects a real session whose own slot is unheld", () => { + const unfronted = rosterEntry({ pid: 3, cwd: "/tmp/unfronted" }); + const selected = selectSessionsToFront([unfronted], () => undefined); + expect(selected).toEqual([unfronted]); + }); + + it("probes the slot computed from the entry's own cwd, not an unrelated one", () => { + const entry = rosterEntry({ pid: 4, cwd: "/tmp/watched" }); + let probedSlot: { harness: string; cwd: string } | undefined; + selectSessionsToFront([entry], (slot) => { + probedSlot = slot; + return undefined; + }); + expect(probedSlot).toEqual({ harness: "claude-code", cwd: "/tmp/watched" }); + }); + + it("handles a mixed roster, keeping only the unfronted real sessions", () => { + const libraryPeer = rosterEntry({ pid: 1, version: "cc-peer" }); + const selfFronted = rosterEntry({ pid: 2, cwd: "/tmp/self-fronted" }); + const unfronted = rosterEntry({ pid: 3, cwd: "/tmp/unfronted" }); + const selected = selectSessionsToFront( + [libraryPeer, selfFronted, unfronted], + (slot) => + slot.cwd === "/tmp/self-fronted" ? LIVE_LOCK_HOLDER_PID : undefined, + ); + expect(selected).toEqual([unfronted]); + }); +}); + +interface FrontedRecordStub { + messagingSocketPath: string; + label: string; +} + +describe("matchInboundMessageSession", () => { + it("matches a fronted session by the uds: envelope convention", () => { + const a: FrontedRecordStub = { + messagingSocketPath: "/tmp/sock-a", + label: "a", + }; + const b: FrontedRecordStub = { + messagingSocketPath: "/tmp/sock-b", + label: "b", + }; + const match = matchInboundMessageSession([a, b], { + from: "uds:/tmp/sock-b", + }); + expect(match).toBe(b); + }); + + it("returns undefined when no fronted session's socket matches", () => { + const a: FrontedRecordStub = { + messagingSocketPath: "/tmp/sock-a", + label: "a", + }; + const match = matchInboundMessageSession([a], { + from: "uds:/tmp/sock-unknown", + }); + expect(match).toBeUndefined(); + }); + + it("returns undefined when the message carries no from field at all", () => { + const a: FrontedRecordStub = { + messagingSocketPath: "/tmp/sock-a", + label: "a", + }; + expect(matchInboundMessageSession([a], {})).toBeUndefined(); + }); + + it("never matches a from value missing the uds: prefix, even with an otherwise identical path", () => { + const a: FrontedRecordStub = { + messagingSocketPath: "/tmp/sock-a", + label: "a", + }; + expect( + matchInboundMessageSession([a], { from: "/tmp/sock-a" }), + ).toBeUndefined(); + }); +}); From 1713816a06ad4d39c3db494d76686185d5ba49a7 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:30:31 +0100 Subject: [PATCH 05/11] feat(core): add an onCoordinatorRoleChanged hook to MeshStore Fires true right after this store becomes coordinator (a fresh bind in init(), or a takeover via PeerLifecycle.handleBecomeCoordinator) and false right before shutdown() drops the role, mirroring the existing onDelivery/onPatch/onError callback fields. Gives bridge-mesh.ts a place to start and stop a coordinator-only, Node/filesystem-specific capability -- the cc-peer front (agent-comms#157) -- without pulling that concern into this transport-agnostic core. --- src/core/mesh-store.ts | 11 +++++++ src/core/peer-lifecycle.ts | 3 ++ src/test/mesh-store-orchestration.test.ts | 38 +++++++++++++++++++++++ src/test/peer-lifecycle.test.ts | 17 ++++++++++ 4 files changed, 69 insertions(+) diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 4c197dbc..ff09532e 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -159,6 +159,12 @@ export class MeshStore implements CommsStore { */ onError: ((error: Error) => void) | undefined; + /** + * Fires whenever this store's own coordinator role changes: true right after becoming coordinator (a fresh bind in init(), or a takeover via PeerLifecycle.handleBecomeCoordinator), false right before shutdown() drops it. There is no live "lost the role to someone else while still running" case today -- CoordinatorGateway.onLostCoordinator is only ever called from shutdown(), so this callback mirrors that same lifecycle. Left undefined by default (matching onDelivery/onPatch/onError): a caller that wants to react to owning the coordinator role -- e.g. bridge-mesh.ts starting/stopping the cc-peer front (agent-comms#157), a Node/filesystem-specific capability that has no place in this transport-agnostic core -- sets it, exactly like those three. + */ + onCoordinatorRoleChanged: + ((isCoordinator: boolean) => void | Promise) | undefined; + /** Serialise the full mesh state for state_sync messages. */ serialise(): SerialisedState { return { @@ -299,6 +305,9 @@ export class MeshStore implements CommsStore { deliveryEngine: this.deliveryEngine, staleAgentChecker: this.staleAgentChecker, coordinatorGateway: this.coordinatorGateway, + onCoordinatorRoleChanged: async () => { + await this.onCoordinatorRoleChanged?.(true); + }, }); } @@ -370,6 +379,7 @@ export class MeshStore implements CommsStore { ); this.staleAgentChecker.start(); await this.coordinatorGateway.onBecameCoordinator(); + await this.onCoordinatorRoleChanged?.(true); connected = true; } catch (coordErr) { const msg = @@ -773,6 +783,7 @@ export class MeshStore implements CommsStore { this.staleAgentChecker.stop(); await this.coordinatorGateway.onLostCoordinator(); + await this.onCoordinatorRoleChanged?.(false); await this.requireTransport().shutdown(); } } diff --git a/src/core/peer-lifecycle.ts b/src/core/peer-lifecycle.ts index f2f1f2f3..6463d4d0 100644 --- a/src/core/peer-lifecycle.ts +++ b/src/core/peer-lifecycle.ts @@ -29,6 +29,8 @@ export interface PeerLifecycleDeps { staleAgentChecker: Pick; /** Dials the hub the moment this side takes over as coordinator (agent-comms#154) -- see CoordinatorGateway's own class doc. Narrowed to the one method handleBecomeCoordinator ever calls; onLostCoordinator is MeshStore.shutdown()'s own concern, not this class's. */ coordinatorGateway: Pick; + /** Fires alongside coordinatorGateway.onBecameCoordinator, right after this side takes over as coordinator -- MeshStore's own hook for starting a coordinator-only capability that doesn't belong in the transport-agnostic core itself (e.g. the cc-peer front, agent-comms#157). Optional and left unset by most callers, mirroring MeshStore's own onDelivery/onPatch/onError callback fields. */ + onCoordinatorRoleChanged?: (() => void | Promise) | undefined; } export class PeerLifecycle { @@ -118,6 +120,7 @@ export class PeerLifecycle { } this.deps.staleAgentChecker.start(); await this.deps.coordinatorGateway.onBecameCoordinator(); + await this.deps.onCoordinatorRoleChanged?.(); } handlePeerDisconnected(handle: Readonly): void { diff --git a/src/test/mesh-store-orchestration.test.ts b/src/test/mesh-store-orchestration.test.ts index 698ae01d..3ad4669f 100644 --- a/src/test/mesh-store-orchestration.test.ts +++ b/src/test/mesh-store-orchestration.test.ts @@ -215,6 +215,34 @@ describe("MeshStore — init()", () => { expect(transport.unref).toHaveBeenCalledTimes(1); }); + it("fires onCoordinatorRoleChanged(true) once this store becomes coordinator on a fresh bind", async () => { + const store = new MeshStore(); + const transport = fakeTransport(); + vi.mocked(transport.connectToCoordinator).mockRejectedValue( + new Error("ECONNREFUSED"), + ); + store.setTransport(transport); + const onCoordinatorRoleChanged = vi.fn<(isCoordinator: boolean) => void>(); + store.onCoordinatorRoleChanged = onCoordinatorRoleChanged; + + await store.init(); + + expect(onCoordinatorRoleChanged).toHaveBeenCalledTimes(1); + expect(onCoordinatorRoleChanged).toHaveBeenCalledWith(true); + }); + + it("never fires onCoordinatorRoleChanged when joining an existing coordinator rather than becoming one", async () => { + const store = new MeshStore(); + const transport = fakeTransport(); + store.setTransport(transport); + const onCoordinatorRoleChanged = vi.fn<(isCoordinator: boolean) => void>(); + store.onCoordinatorRoleChanged = onCoordinatorRoleChanged; + + await store.init(); + + expect(onCoordinatorRoleChanged).not.toHaveBeenCalled(); + }); + it("degrades gracefully (no throw, onError fires) when becomeCoordinator fails with EADDRINUSE", async () => { const store = new MeshStore(); const transport = fakeTransport(); @@ -661,6 +689,16 @@ describe("MeshStore — shutdown()", () => { expect(transport.shutdown).toHaveBeenCalledTimes(1); }); + it("fires onCoordinatorRoleChanged(false) on shutdown, when one is set", async () => { + const onCoordinatorRoleChanged = vi.fn<(isCoordinator: boolean) => void>(); + store.onCoordinatorRoleChanged = onCoordinatorRoleChanged; + + await store.shutdown(); + + expect(onCoordinatorRoleChanged).toHaveBeenCalledTimes(1); + expect(onCoordinatorRoleChanged).toHaveBeenCalledWith(false); + }); + it("broadcasts agent_offline for its own self agent when one is registered", async () => { const agent = await store.registerAgent({ name: "self", diff --git a/src/test/peer-lifecycle.test.ts b/src/test/peer-lifecycle.test.ts index d367a290..7c7c2e72 100644 --- a/src/test/peer-lifecycle.test.ts +++ b/src/test/peer-lifecycle.test.ts @@ -34,6 +34,7 @@ interface Harness { applyPatch: ReturnType; staleAgentCheckerStart: ReturnType; coordinatorGatewayOnBecameCoordinator: ReturnType; + onCoordinatorRoleChanged: ReturnType; } function makeHarness(): Harness { @@ -52,6 +53,9 @@ function makeHarness(): Harness { const coordinatorGatewayOnBecameCoordinator = vi .fn() .mockResolvedValue(undefined); + const onCoordinatorRoleChanged = vi + .fn>() + .mockResolvedValue(undefined); const deps: PeerLifecycleDeps = { peerInfo: new Map(), agents: new Map(), @@ -65,6 +69,7 @@ function makeHarness(): Harness { coordinatorGateway: { onBecameCoordinator: coordinatorGatewayOnBecameCoordinator, }, + onCoordinatorRoleChanged, }; return { deps, @@ -75,6 +80,7 @@ function makeHarness(): Harness { applyPatch, staleAgentCheckerStart, coordinatorGatewayOnBecameCoordinator, + onCoordinatorRoleChanged, }; } @@ -211,6 +217,17 @@ describe("PeerLifecycle — handleBecomeCoordinator", () => { ); expect(h.staleAgentCheckerStart).toHaveBeenCalledTimes(1); expect(h.coordinatorGatewayOnBecameCoordinator).toHaveBeenCalledTimes(1); + expect(h.onCoordinatorRoleChanged).toHaveBeenCalledTimes(1); + }); + + it("still becomes coordinator when onCoordinatorRoleChanged is left unset", async () => { + const h = makeHarness(); + h.deps.onCoordinatorRoleChanged = undefined; + + await expect( + h.lifecycle.handleBecomeCoordinator([peerInfo("a")]), + ).resolves.toBeUndefined(); + expect(h.transport.becomeCoordinator).toHaveBeenCalledTimes(1); }); }); From b80b93ddcc89b440f0936c5d968312c3af340c73 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:34:44 +0100 Subject: [PATCH 06/11] feat(cc-peer): add the coordinator-only periodic front controller CcPeerFront polls the injected roster on a fixed interval, attaching newly-selected sessions and detaching ones that exited or yielded (their own slot got claimed by a live bridge), diffing against what's already fronted so nothing is re-attached or double-torn-down across ticks. stop() detaches every still-fronted session, since a side that just lost the coordinator role has no business relaying on any session's behalf anymore. Mirrors StaleAgentChecker's own coordinator-only periodic-probe shape and stays free of any real cc-peer/MeshStore construction, matching front.ts's own testability discipline. --- src/bridges/cc-peer/front-controller.ts | 118 ++++++++++ src/test/cc-peer-front-controller.test.ts | 275 ++++++++++++++++++++++ 2 files changed, 393 insertions(+) create mode 100644 src/bridges/cc-peer/front-controller.ts create mode 100644 src/test/cc-peer-front-controller.test.ts diff --git a/src/bridges/cc-peer/front-controller.ts b/src/bridges/cc-peer/front-controller.ts new file mode 100644 index 00000000..14546a5c --- /dev/null +++ b/src/bridges/cc-peer/front-controller.ts @@ -0,0 +1,118 @@ +/** + * CcPeerFront — coordinator-only periodic controller for the default cc-peer front (agent-comms#157), attaching and detaching fronted sessions on each poll tick. Mirrors StaleAgentChecker's own "coordinator-only periodic probe" shape (owns its own interval timer exclusively, start/stop, no immediate first tick) since both are the same kind of thing: a background job that only ever runs on this machine's current mesh coordinator. + * + * Kept free of any real cc-peer/MeshStore construction, exactly like front.ts's pure decision logic this class is built on -- every real I/O (listRoster, probeSlotOwner, attach, detach) is injected, so the attach/detach/yield diffing across ticks is testable with fake timers and no real local Claude Code session. front-runtime.ts supplies the real dependencies. + */ + +import { + selectSessionsToFront, + matchInboundMessageSession, + type CcPeerRosterEntryLike, +} from "./front.js"; +import type { IdentitySlot } from "../../core/identity-store.js"; + +/** How often the front re-enumerates the local cc-peer roster and re-probes every currently-fronted session's own slot. */ +const DEFAULT_POLL_INTERVAL_MS = 5000; + +/** The minimum shape a fronted-session record must carry so CcPeerFront can track and route to it, regardless of whatever real MeshStore/CommsTool state a concrete implementation (front-runtime.ts) attaches alongside these fields. */ +export interface FrontedSessionRecord { + readonly pid: number; + readonly cwd: string; + readonly messagingSocketPath: string; + /** Delivers an inbound cc-peer message addressed to this fronted session into the mesh (the session-to-mesh direction) -- wired by attach() itself, exactly as wireCcPeerBridge's own peer.on("message") handler does for the one-shot bridge command. */ + readonly handleInbound: ( + message: Readonly<{ from?: string; fromName?: string; body: string }>, + ) => void; +} + +export interface CcPeerFrontDeps { + /** Enumerates the current local cc-peer roster. Rejects propagate to onError rather than throwing out of the poll timer. */ + listRoster: () => Promise; + /** Read-only probe of a slot's current lock holder -- identity-store.ts's probeSlotOwner in production. */ + probeSlotOwner: (slot: Readonly) => number | undefined; + /** Builds a fronted-session record for a newly-selected roster entry (mints/loads its identity, wires the mesh store and the inbound relay). Rejects propagate to onError; the entry is retried on the next tick. */ + attach: (entry: Readonly) => Promise; + /** Tears a fronted-session record down (marks its agent offline, shuts its mesh store down). Rejects propagate to onError. */ + detach: (record: TRecord) => Promise; + pollIntervalMs?: number; + onError?: (error: Error) => void; +} + +function toError(value: unknown): Error { + return value instanceof Error ? value : new Error(String(value)); +} + +export class CcPeerFront { + private readonly fronted = new Map(); + private timer: ReturnType | undefined; + + constructor(private readonly deps: Readonly>) {} + + /** Starts the periodic poll (coordinator-only). No-op if already running. Matches StaleAgentChecker's own convention of no immediate first tick -- the first attach happens on the first elapsed interval, not synchronously on start(). */ + start(): void { + if (this.timer) return; + this.timer = setInterval(() => { + void this.tick(); + }, this.deps.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS); + } + + /** Stops the periodic poll and detaches every currently-fronted session -- this side is no longer the coordinator, so it has no business still relaying on any session's behalf. */ + async stop(): Promise { + if (this.timer) { + clearInterval(this.timer); + this.timer = undefined; + } + const records = [...this.fronted.values()]; + this.fronted.clear(); + for (const record of records) { + await this.detachOne(record); + } + } + + /** Routes an inbound cc-peer message to whichever fronted session it came from, if any -- the shared front-wide CcPeer instance's own "message" listener calls this directly (front-runtime.ts), since only this class knows the current fronted set. A message matching no fronted session (already detached, or genuinely foreign) is silently dropped -- there's nowhere for it to go. */ + handleInboundMessage( + message: Readonly<{ from?: string; fromName?: string; body: string }>, + ): void { + const record = matchInboundMessageSession(this.fronted.values(), message); + record?.handleInbound(message); + } + + private async tick(): Promise { + let roster: readonly CcPeerRosterEntryLike[]; + try { + roster = await this.deps.listRoster(); + } catch (err) { + this.deps.onError?.(toError(err)); + return; + } + + const selected = selectSessionsToFront(roster, this.deps.probeSlotOwner); + const selectedPids = new Set(selected.map((entry) => entry.pid)); + const rosterPids = new Set(roster.map((entry) => entry.pid)); + + // Detach anything that either exited (gone from the roster entirely) or yielded (its own slot is now held by a live bridge, so it's no longer in the selected set even though the session process itself is still running). + for (const [pid, record] of [...this.fronted]) { + if (rosterPids.has(pid) && selectedPids.has(pid)) continue; + this.fronted.delete(pid); + await this.detachOne(record); + } + + for (const entry of selected) { + if (this.fronted.has(entry.pid)) continue; + try { + const record = await this.deps.attach(entry); + this.fronted.set(entry.pid, record); + } catch (err) { + this.deps.onError?.(toError(err)); + } + } + } + + private async detachOne(record: TRecord): Promise { + try { + await this.deps.detach(record); + } catch (err) { + this.deps.onError?.(toError(err)); + } + } +} diff --git a/src/test/cc-peer-front-controller.test.ts b/src/test/cc-peer-front-controller.test.ts new file mode 100644 index 00000000..0c432206 --- /dev/null +++ b/src/test/cc-peer-front-controller.test.ts @@ -0,0 +1,275 @@ +/** + * Unit tests for CcPeerFront -- the coordinator-only periodic controller (front-controller.ts) that attaches/detaches fronted sessions on each tick, driven by injected roster/probe/attach/detach/inbound-routing dependencies so it's testable with fake timers and no real cc-peer/MeshStore involved. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + CcPeerFront, + type CcPeerFrontDeps, + type FrontedSessionRecord, +} from "../bridges/cc-peer/front-controller.js"; +import type { CcPeerRosterEntryLike } from "../bridges/cc-peer/front.js"; + +const POLL_INTERVAL_MS = 5000; +/** How many poll intervals stop() must survive with zero further polling -- arbitrary beyond "more than one", chosen to make a lingering timer's own recurrence visible rather than a one-off fluke (matches stale-agent-checker.test.ts's own HALT_CHECK_INTERVAL_COUNT convention). */ +const HALT_CHECK_INTERVAL_COUNT = 3; +/** A pid standing in for a real live agent-comms bridge holding a slot's lock -- arbitrary beyond "a plausible process id", used only to prove the yield path treats "probe returned a pid" as "session now fronts itself". */ +const LIVE_LOCK_HOLDER_PID = 4242; + +function rosterEntry( + overrides: Readonly> = {}, +): CcPeerRosterEntryLike { + return { + pid: 111, + cwd: "/tmp/project", + version: "2.1.269", + messagingSocketPath: "/tmp/sock-111", + ...overrides, + }; +} + +interface StubRecord extends FrontedSessionRecord { + detached: boolean; +} + +/** Unwraps the resolved value of a mocked attach() call's Nth invocation, asserting it actually happened -- vi's own mock.results indexing returns undefined for a call that never occurred, which every call site here has already asserted against via toHaveBeenCalledTimes. */ +async function attachResult( + attach: ReturnType["attach"]>>, + callIndex = 0, +): Promise { + const value = await attach.mock.results[callIndex]?.value; + if (value === undefined) { + throw new Error(`attach() was never called at index ${String(callIndex)}`); + } + return value; +} + +function stubRecord(entry: Readonly): StubRecord { + return { + pid: entry.pid, + cwd: entry.cwd, + messagingSocketPath: entry.messagingSocketPath, + handleInbound: vi.fn(), + detached: false, + }; +} + +interface Harness { + front: CcPeerFront; + deps: CcPeerFrontDeps; + roster: CcPeerRosterEntryLike[]; + attach: ReturnType["attach"]>>; + detach: ReturnType["detach"]>>; + probeSlotOwner: ReturnType< + typeof vi.fn["probeSlotOwner"]> + >; + onError: ReturnType< + typeof vi.fn["onError"]>> + >; +} + +function harness(initialRoster: readonly CcPeerRosterEntryLike[]): Harness { + const roster = [...initialRoster]; + const attach = vi.fn["attach"]>(async (entry) => + Promise.resolve(stubRecord(entry)), + ); + const detach = vi.fn["detach"]>( + async (record) => { + record.detached = true; + return Promise.resolve(); + }, + ); + const probeSlotOwner = vi.fn["probeSlotOwner"]>( + () => undefined, + ); + const onError = vi.fn["onError"]>>(); + const deps: CcPeerFrontDeps = { + listRoster: async () => Promise.resolve([...roster]), + probeSlotOwner, + attach, + detach, + pollIntervalMs: POLL_INTERVAL_MS, + onError, + }; + return { + front: new CcPeerFront(deps), + deps, + roster, + attach, + detach, + probeSlotOwner, + onError, + }; +} + +describe("CcPeerFront — start/stop", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("does nothing until start() is called", async () => { + const h = harness([rosterEntry()]); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS * 2); + expect(h.attach).not.toHaveBeenCalled(); + h.front.stop(); + }); + + it("attaches a session on the first poll tick", async () => { + const h = harness([rosterEntry({ pid: 1 })]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(1); + await h.front.stop(); + }); + + it("calling start() twice does not create a second timer", async () => { + const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); + const h = harness([]); + h.front.start(); + h.front.start(); + expect(setIntervalSpy).toHaveBeenCalledTimes(1); + await h.front.stop(); + }); + + it("stop() halts further polling", async () => { + const h = harness([rosterEntry({ pid: 1 })]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(1); + await h.front.stop(); + await vi.advanceTimersByTimeAsync( + POLL_INTERVAL_MS * HALT_CHECK_INTERVAL_COUNT, + ); + expect(h.attach).toHaveBeenCalledTimes(1); + }); + + it("stop() detaches every currently-fronted session", async () => { + const h = harness([rosterEntry({ pid: 1 })]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(1); + const record = await attachResult(h.attach); + + await h.front.stop(); + + expect(h.detach).toHaveBeenCalledTimes(1); + expect(record.detached).toBe(true); + }); +}); + +describe("CcPeerFront — attach/detach diffing across ticks", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("does not re-attach a session already fronted from a previous tick", async () => { + const h = harness([rosterEntry({ pid: 1 })]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(1); + await h.front.stop(); + }); + + it("attaches a session that appears in the roster on a later tick", async () => { + const h = harness([]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).not.toHaveBeenCalled(); + + h.roster.push(rosterEntry({ pid: 2 })); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(1); + await h.front.stop(); + }); + + it("detaches a session that has exited (no longer in the roster)", async () => { + const h = harness([rosterEntry({ pid: 1 })]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(1); + + h.roster.length = 0; + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.detach).toHaveBeenCalledTimes(1); + await h.front.stop(); + }); + + it("yields a session the moment its own slot is claimed by a live bridge, without re-attaching it", async () => { + const h = harness([rosterEntry({ pid: 1, cwd: "/tmp/yields" })]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(1); + const record = await attachResult(h.attach); + + // A real agent-comms bridge has since taken the slot over -- the probe now reports its pid. + h.probeSlotOwner.mockReturnValue(LIVE_LOCK_HOLDER_PID); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + + expect(h.detach).toHaveBeenCalledTimes(1); + expect(record.detached).toBe(true); + expect(h.attach).toHaveBeenCalledTimes(1); + + // Even if the slot frees up again later, this front never re-attaches on its own within the same run -- the session is expected to keep its own bridge running from here. + h.probeSlotOwner.mockReturnValue(undefined); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.attach).toHaveBeenCalledTimes(2); + await h.front.stop(); + }); + + it("routes an inbound message to the fronted session it came from", async () => { + const h = harness([ + rosterEntry({ pid: 1, messagingSocketPath: "/tmp/sock-1" }), + ]); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + const record = await attachResult(h.attach); + + h.front.handleInboundMessage({ from: "uds:/tmp/sock-1", body: "hi" }); + + expect(record.handleInbound).toHaveBeenCalledWith({ + from: "uds:/tmp/sock-1", + body: "hi", + }); + await h.front.stop(); + }); + + it("silently ignores an inbound message matching no fronted session", async () => { + const h = harness([]); + expect(() => + h.front.handleInboundMessage({ from: "uds:/tmp/unknown", body: "hi" }), + ).not.toThrow(); + }); + + it("reports a roster failure via onError rather than throwing out of the poll timer", async () => { + const h = harness([]); + const failure = new Error("roster unavailable"); + h.deps.listRoster = async () => Promise.reject(failure); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + expect(h.onError).toHaveBeenCalledWith(failure); + await h.front.stop(); + }); + + it("reports an attach failure via onError without losing track of other sessions", async () => { + const h = harness([ + rosterEntry({ pid: 1, cwd: "/tmp/a" }), + rosterEntry({ pid: 2, cwd: "/tmp/b" }), + ]); + const failure = new Error("cc-peer send failed"); + h.attach.mockImplementationOnce(async () => Promise.reject(failure)); + h.front.start(); + await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS); + + expect(h.onError).toHaveBeenCalledWith(failure); + expect(h.attach).toHaveBeenCalledTimes(2); + await h.front.stop(); + }); +}); From fc46e0f97ef4c1cec4bd38f8d760dd0a37ed7109 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:37:12 +0100 Subject: [PATCH 07/11] feat(cc-peer): add DI-testable relay wiring for one fronted session buildFrontedSessionRecord wires both relay directions for a single fronted session against its already-constructed mesh store/tool and the front's shared cc-peer peer: mesh-to-session sends a formatted delivery to the session's own pid, session-to-mesh posts an inbound cc-peer message into the session's own project room. detachFrontedSession marks the agent offline before shutting its store down. Mirrors bridges/cc-peer/bridge.ts's own wireCcPeerBridge split -- kept free of any real CcPeer.create()/createBridgeMeshFromIdentity construction so it's testable against fakes, with front-runtime.ts (still to come) supplying the real ones. --- src/bridges/cc-peer/front-relay.ts | 91 +++++++++++ src/test/cc-peer-front-relay.test.ts | 217 +++++++++++++++++++++++++++ 2 files changed, 308 insertions(+) create mode 100644 src/bridges/cc-peer/front-relay.ts create mode 100644 src/test/cc-peer-front-relay.test.ts diff --git a/src/bridges/cc-peer/front-relay.ts b/src/bridges/cc-peer/front-relay.ts new file mode 100644 index 00000000..169a1c68 --- /dev/null +++ b/src/bridges/cc-peer/front-relay.ts @@ -0,0 +1,91 @@ +/** + * Pure, DI-testable wiring for one fronted session's own cc-peer-to-mesh relay (agent-comms#157) -- the "attach"/"detach" half of the default front, building a FrontedSessionRecord (front-controller.ts) from an already-constructed mesh and the front's shared cc-peer peer. Mirrors bridges/cc-peer/bridge.ts's own wireCcPeerBridge split: kept free of any real CcPeer.create()/createBridgeMeshFromIdentity construction so it's testable against fakes, with front-runtime.ts supplying the real ones. + * + * Reuses CcPeerRef/CcPeerInboundMessage from bridge.ts rather than redeclaring them -- the wire shape a shared front-wide peer speaks is identical to the one-shot bridge command's own peer, just addressed by pid instead of a fixed target. + */ + +import { buildAction, formatDeliveryEvent } from "../../core/bridge.js"; +import type { CommsTool } from "../../core/tool.js"; +import type { DeliveryEvent } from "../../core/types.js"; +import type { FrontedSessionRecord } from "./front-controller.js"; +import type { CcPeerRosterEntryLike } from "./front.js"; +import type { CcPeerInboundMessage, CcPeerRef } from "./bridge.js"; + +/** The narrow slice of MeshStore a fronted session's own record needs -- onDelivery to wire the mesh-to-session direction, setAgentOffline/shutdown for detachFrontedSession's own teardown. */ +export interface FrontRelayStore { + onDelivery: + | ((agentId: string, event: DeliveryEvent) => void | Promise) + | undefined; + setAgentOffline: (id: string) => Promise; + shutdown: () => Promise; +} + +/** The narrow slice of the front's shared CcPeer instance this relay needs -- just send(), since the front-wide "message" listener (front-runtime.ts) is what routes an inbound message to this record's own handleInbound in the first place, not this module. */ +export interface FrontRelayPeer { + send: ( + target: Readonly, + body: string, + ) => Promise<{ msgId: string }>; +} + +export interface FrontedRelayRecord extends FrontedSessionRecord { + agentId: string; + roomId: string; + store: FrontRelayStore; +} + +export interface BuildFrontedSessionRecordDeps { + entry: Readonly; + agentId: string; + roomId: string; + store: FrontRelayStore; + tool: Pick; + peer: FrontRelayPeer; +} + +/** + * Wires both relay directions for one fronted session and returns the record CcPeerFront tracks it under. Mesh-to-session: store.onDelivery sends the formatted event to this session's own pid via the shared peer. Session-to-mesh: the returned handleInbound (called by the front's shared "message" listener once it's matched this record by socket path) posts the message into this session's own project room, exactly as wireCcPeerBridge's own peer.on("message") handler does for the one-shot bridge command. + */ +export function buildFrontedSessionRecord( + deps: Readonly, +): FrontedRelayRecord { + const { entry, agentId, roomId, store, tool, peer } = deps; + + store.onDelivery = (_targetId, event) => { + void peer.send({ pid: entry.pid }, formatDeliveryEvent(event)); + }; + + return { + pid: entry.pid, + cwd: entry.cwd, + messagingSocketPath: entry.messagingSocketPath, + agentId, + roomId, + store, + handleInbound: (message: Readonly) => { + const sender = message.fromName ?? message.from ?? "unknown"; + const action = buildAction({ + action: "send", + room: roomId, + content: `${sender}: ${message.body}`, + }); + void tool.handle( + { + agentId, + harness: "claude-code", + cwd: entry.cwd, + pid: process.pid, + }, + action, + ); + }, + }; +} + +/** Tears a fronted session's record down: marks its agent offline (a courtesy to peers watching its presence), then shuts its mesh store down. There is no identity lock to release -- attach() never took one (loadIdentityForFront's whole point), so a real bridge for this slot can already have claimed it by the time this runs. */ +export async function detachFrontedSession( + record: Readonly>, +): Promise { + await record.store.setAgentOffline(record.agentId); + await record.store.shutdown(); +} diff --git a/src/test/cc-peer-front-relay.test.ts b/src/test/cc-peer-front-relay.test.ts new file mode 100644 index 00000000..8a581821 --- /dev/null +++ b/src/test/cc-peer-front-relay.test.ts @@ -0,0 +1,217 @@ +/** + * Direct, DI-based unit tests for buildFrontedSessionRecord/detachFrontedSession -- the pure relay wiring front-relay.ts exposes, tested against fake store/tool/peer objects rather than a real MeshStore or local Claude Code session, mirroring cc-peer-bridge.test.ts's own approach for the one-shot bridge command. + */ +import { describe, expect, it, vi } from "vitest"; +import { + buildFrontedSessionRecord, + detachFrontedSession, +} from "../bridges/cc-peer/front-relay.js"; +import type { + FrontRelayPeer, + FrontRelayStore, +} from "../bridges/cc-peer/front-relay.js"; +import type { CcPeerRosterEntryLike } from "../bridges/cc-peer/front.js"; +import type { CommsTool } from "../core/tool.js"; +import type { DeliveryEvent, RoomMessage } from "../core/types.js"; + +function rosterEntry( + overrides: Readonly> = {}, +): CcPeerRosterEntryLike { + return { + pid: 222, + cwd: "/tmp/project", + version: "2.1.269", + messagingSocketPath: "/tmp/sock-222", + ...overrides, + }; +} + +function fakeStore(): FrontRelayStore & { + setAgentOfflineCalls: string[]; + shutdownCalls: number; +} { + const setAgentOfflineCalls: string[] = []; + let shutdownCalls = 0; + return { + onDelivery: undefined, + setAgentOfflineCalls, + get shutdownCalls() { + return shutdownCalls; + }, + setAgentOffline: vi.fn(async (id: string) => { + setAgentOfflineCalls.push(id); + return Promise.resolve(); + }), + shutdown: vi.fn(async () => { + shutdownCalls += 1; + return Promise.resolve(); + }), + }; +} + +function fakePeer(): FrontRelayPeer & { + sendCalls: { target: unknown; body: string }[]; +} { + const sendCalls: { target: unknown; body: string }[] = []; + return { + sendCalls, + send: vi.fn(async (target: unknown, body: string) => { + sendCalls.push({ target, body }); + return Promise.resolve({ msgId: "msg-1" }); + }), + }; +} + +function fakeTool(): Pick & { handleCalls: unknown[] } { + const handleCalls: unknown[] = []; + return { + handleCalls, + handle: vi.fn(async (ctx: unknown, action: unknown) => { + handleCalls.push({ ctx, action }); + return Promise.resolve({ content: "ok", isError: false }); + }), + }; +} + +function roomMessage(overrides: Partial = {}): RoomMessage { + return { + id: "msg-1", + from: "peer-a", + room: "owner/project", + content: "hi from the mesh", + timestamp: "2026-01-01T00:00:00.000Z", + readBy: [], + ...overrides, + }; +} + +describe("buildFrontedSessionRecord — outbound (mesh -> cc-peer)", () => { + it("wires store.onDelivery to send the formatted event to the session's own pid", () => { + const store = fakeStore(); + const peer = fakePeer(); + const entry = rosterEntry({ pid: 333 }); + + buildFrontedSessionRecord({ + entry, + agentId: "agent-1", + roomId: "owner/project", + store, + tool: fakeTool(), + peer, + }); + + expect(store.onDelivery).toBeDefined(); + const event: DeliveryEvent = { + type: "room_message", + message: roomMessage(), + }; + void store.onDelivery?.("agent-1", event); + + expect(peer.sendCalls).toHaveLength(1); + expect(peer.sendCalls[0]?.target).toEqual({ pid: 333 }); + }); +}); + +describe("buildFrontedSessionRecord — inbound (cc-peer -> mesh)", () => { + it("returns a record whose handleInbound posts into the session's own project room", () => { + const tool = fakeTool(); + const entry = rosterEntry({ cwd: "/tmp/my-project" }); + + const record = buildFrontedSessionRecord({ + entry, + agentId: "agent-1", + roomId: "owner/my-project", + store: fakeStore(), + tool, + peer: fakePeer(), + }); + + record.handleInbound({ + from: "local-session", + fromName: "my-local-session", + body: "hello from cc-peer", + }); + + expect(tool.handleCalls).toHaveLength(1); + expect(tool.handleCalls[0]).toEqual({ + ctx: { + agentId: "agent-1", + harness: "claude-code", + cwd: "/tmp/my-project", + pid: process.pid, + }, + action: { + action: "send", + target: "owner/my-project", + content: "my-local-session: hello from cc-peer", + }, + }); + }); + + it("falls back to the raw from id when the session has no registered display name", () => { + const tool = fakeTool(); + const record = buildFrontedSessionRecord({ + entry: rosterEntry(), + agentId: "agent-1", + roomId: "owner/project", + store: fakeStore(), + tool, + peer: fakePeer(), + }); + + record.handleInbound({ from: "local-session", body: "hi" }); + + const call = tool.handleCalls[0] as { action: { content: string } }; + expect(call.action.content).toBe("local-session: hi"); + }); +}); + +describe("buildFrontedSessionRecord — record shape", () => { + it("carries the roster entry's own pid/cwd/messagingSocketPath, matching what CcPeerFront diffs and routes on", () => { + const entry = rosterEntry({ + pid: 444, + cwd: "/tmp/other", + messagingSocketPath: "/tmp/sock-444", + }); + const record = buildFrontedSessionRecord({ + entry, + agentId: "agent-1", + roomId: "owner/other", + store: fakeStore(), + tool: fakeTool(), + peer: fakePeer(), + }); + + expect(record.pid).toBe(entry.pid); + expect(record.cwd).toBe(entry.cwd); + expect(record.messagingSocketPath).toBe(entry.messagingSocketPath); + }); +}); + +describe("detachFrontedSession", () => { + it("marks the agent offline, then shuts its store down", async () => { + const store = fakeStore(); + + await detachFrontedSession({ agentId: "agent-1", store }); + + expect(store.setAgentOfflineCalls).toEqual(["agent-1"]); + expect(store.shutdownCalls).toBe(1); + }); + + it("shuts the store down only after setAgentOffline resolves, not concurrently with it", async () => { + const store = fakeStore(); + const order: string[] = []; + vi.mocked(store.setAgentOffline).mockImplementation(async () => { + order.push("offline"); + return Promise.resolve(); + }); + vi.mocked(store.shutdown).mockImplementation(async () => { + order.push("shutdown"); + return Promise.resolve(); + }); + + await detachFrontedSession({ agentId: "agent-1", store }); + + expect(order).toEqual(["offline", "shutdown"]); + }); +}); From 680282971aa2f4181c9f0dff9fbe94e774aaf3d7 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:39:13 +0100 Subject: [PATCH 08/11] feat(cc-peer): wire the default front's real cc-peer/MeshStore construction createDefaultCcPeerFront builds the real CcPeerFront: a shared, lazily-created CcPeer instance (created on the first poll tick that actually needs it, not at start()), and an attach function that loads a not-yet-fronted session's identity via loadIdentityForFront, builds its mesh via createBridgeMeshFromIdentity, registers and rooms it, then wires the relay via buildFrontedSessionRecord. A construction or attach failure reports through onError rather than throwing, so a machine with no local Claude Code sessions -- or one where cc-peer itself can't bind -- degrades to a clean no-op front instead of crashing the coordinator that owns it. Widens CcPeerRosterEntryLike.name and CcPeerFrontDeps.pollIntervalMs/onError to explicit `| undefined` so they type-check against cc-peer's own RegistryEntry and an options object's optional fields under this repo's exactOptionalPropertyTypes. Untested directly, matching run.ts's own established precedent for real CcPeer.create() construction -- front.ts, front-controller.ts, and front-relay.ts already carry the front's entire decision/diffing/relay logic under direct DI-based unit tests. --- src/bridges/cc-peer/front-controller.ts | 4 +- src/bridges/cc-peer/front-runtime.ts | 117 ++++++++++++++++++++++++ src/bridges/cc-peer/front.ts | 2 +- 3 files changed, 120 insertions(+), 3 deletions(-) create mode 100644 src/bridges/cc-peer/front-runtime.ts diff --git a/src/bridges/cc-peer/front-controller.ts b/src/bridges/cc-peer/front-controller.ts index 14546a5c..fda0ac70 100644 --- a/src/bridges/cc-peer/front-controller.ts +++ b/src/bridges/cc-peer/front-controller.ts @@ -34,8 +34,8 @@ export interface CcPeerFrontDeps { attach: (entry: Readonly) => Promise; /** Tears a fronted-session record down (marks its agent offline, shuts its mesh store down). Rejects propagate to onError. */ detach: (record: TRecord) => Promise; - pollIntervalMs?: number; - onError?: (error: Error) => void; + pollIntervalMs?: number | undefined; + onError?: ((error: Error) => void) | undefined; } function toError(value: unknown): Error { diff --git a/src/bridges/cc-peer/front-runtime.ts b/src/bridges/cc-peer/front-runtime.ts new file mode 100644 index 00000000..d7e8be5d --- /dev/null +++ b/src/bridges/cc-peer/front-runtime.ts @@ -0,0 +1,117 @@ +/** + * Real cc-peer/MeshStore construction for the default coordinator-run front (agent-comms#157) -- the counterpart to bridges/cc-peer/run.ts's own real CcPeer.create()/createBridgeMesh construction for the one-shot `bridge cc-peer` command. Wires front-controller.ts's CcPeerFront (attach/detach diffing across ticks) to front-relay.ts's buildFrontedSessionRecord/detachFrontedSession (one session's own relay) against one shared, front-wide CcPeer instance and a fresh per-session MeshStore built from loadIdentityForFront's lock-free identity. + * + * Untested directly, exactly like run.ts's own real construction (see cc-peer-bridge.test.ts's header comment): front.ts, front-controller.ts, and front-relay.ts already carry the front's entire decision/diffing/relay logic under direct DI-based unit tests, so this file's only remaining job -- calling the real cc-peer/core APIs in the right order -- is exercised by actually running a bridge as this machine's coordinator. + */ + +import { CcPeer } from "cc-peer"; +import type { InboundMessage as CcPeerInboundMessage } from "cc-peer"; +import { + createBridgeMeshFromIdentity, + ensureRegistered, + ensureProjectRoom, +} from "../../core/index.js"; +import { + loadIdentityForFront, + probeSlotOwner, +} from "../../core/identity-store.js"; +import { CcPeerFront } from "./front-controller.js"; +import { computeFrontSlot } from "./front.js"; +import type { CcPeerRosterEntryLike } from "./front.js"; +import { + buildFrontedSessionRecord, + detachFrontedSession, + type FrontedRelayRecord, +} from "./front-relay.js"; + +/** cc-peer's own registered display name for the front's shared peer -- distinct from any individual fronted session's own agent-comms display name (front-relay.ts's ensureRegistered call uses the session's own cc-peer name/pid for that). */ +const FRONT_PEER_NAME = "agent-comms-front"; + +export interface CreateDefaultCcPeerFrontOptions { + coordinatorPort?: number; + hubUrl?: string; + pollIntervalMs?: number; + onError?: (error: Error) => void; +} + +/** Falls back to "claude-code-" when a session has never picked its own cc-peer display name -- ensureRegistered requires a defaultName, and an unnamed session is still worth fronting under something stable and identifiable. */ +function defaultSessionName(entry: Readonly): string { + return entry.name ?? `claude-code-${String(entry.pid)}`; +} + +/** + * Builds the default cc-peer front. start()/stop() are safe to call from bridge-mesh.ts's onCoordinatorRoleChanged hook regardless of whether cc-peer itself is usable on this machine: the shared CcPeer instance is created lazily, on the first poll tick that actually needs it (listRoster), not at start() itself, and any construction or attach failure is reported via onError rather than thrown -- "absent sockets" (no local Claude Code sessions at all) is then a clean no-op front rather than a crash of the coordinator that owns it. + */ +export function createDefaultCcPeerFront( + options: Readonly = {}, +): Pick, "start" | "stop"> { + let sharedPeerPromise: Promise | undefined; + + const front = new CcPeerFront({ + listRoster: async () => { + const peer = await ensureSharedPeer(); + return peer.roster(); + }, + probeSlotOwner, + attach: async (entry) => attachSession(entry, await ensureSharedPeer()), + detach: detachFrontedSession, + pollIntervalMs: options.pollIntervalMs, + onError: options.onError, + }); + + return { + start: () => { + front.start(); + }, + stop: async () => { + await front.stop(); + const peer = await sharedPeerPromise?.catch(() => undefined); + sharedPeerPromise = undefined; + await peer?.stop(); + }, + }; + + async function ensureSharedPeer(): Promise { + sharedPeerPromise ??= CcPeer.create({ name: FRONT_PEER_NAME }).then( + (peer) => { + peer.on("message", (message: Readonly) => { + front.handleInboundMessage(message); + }); + return peer; + }, + ); + return sharedPeerPromise; + } + + async function attachSession( + entry: Readonly, + peer: CcPeer, + ): Promise { + const slot = computeFrontSlot(entry.cwd); + const identity = loadIdentityForFront(slot); + const { store, tool } = await createBridgeMeshFromIdentity( + identity, + slot, + options.coordinatorPort, + options.hubUrl, + ); + + const reg = await ensureRegistered({ + store, + harness: "claude-code", + cwd: entry.cwd, + defaultName: defaultSessionName(entry), + visibility: "visible", + }); + const roomId = await ensureProjectRoom(store, reg.agentId, entry.cwd); + + return buildFrontedSessionRecord({ + entry, + agentId: reg.agentId, + roomId, + store, + tool, + peer, + }); + } +} diff --git a/src/bridges/cc-peer/front.ts b/src/bridges/cc-peer/front.ts index 18eae005..1742037b 100644 --- a/src/bridges/cc-peer/front.ts +++ b/src/bridges/cc-peer/front.ts @@ -10,7 +10,7 @@ import type { IdentitySlot } from "../../core/identity-store.js"; export interface CcPeerRosterEntryLike { pid: number; cwd: string; - name?: string; + name?: string | undefined; /** cc-peer's own registry entries for library-backed peers (this front itself, the one-shot `bridge cc-peer` command, an alias-pool worker) always carry the literal string "cc-peer" here -- see buildRegistryEntry in cc-peer's own source. A real interactive Claude Code session, which registers itself natively rather than through the cc-peer library, reports its own Claude Code version instead. */ version: string; messagingSocketPath: string; From 90c9102c97f5ce3a185ed31dc98201c9f49c042b Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:42:50 +0100 Subject: [PATCH 09/11] feat(cc-peer): add the store.onCoordinatorRoleChanged glue for the default front wireDefaultCcPeerFront builds the callback a real bridge assigns to its own store.onCoordinatorRoleChanged: starts the front the moment that store becomes this machine's mesh coordinator, stops it the moment it loses that role. This is the piece that actually makes "a machine's coordinator bridge fronts local Claude Code sessions by default" true, rather than a capability every bridge builds but nothing switches on -- the real bridge entry points wire it next. Kept the only file outside bridges/cc-peer/ that needs to know cc-peer exists, matching this repo's own portable-runtime-boundary convention: core/mesh-store.ts's hook is a bare boolean callback, and core/bridge-mesh.ts's factories stay cc-peer-agnostic. --- src/bridges/cc-peer/default-front.ts | 49 ++++++++++++++ src/bridges/cc-peer/front-runtime.ts | 8 +-- src/test/cc-peer-default-front.test.ts | 93 ++++++++++++++++++++++++++ 3 files changed, 146 insertions(+), 4 deletions(-) create mode 100644 src/bridges/cc-peer/default-front.ts create mode 100644 src/test/cc-peer-default-front.test.ts diff --git a/src/bridges/cc-peer/default-front.ts b/src/bridges/cc-peer/default-front.ts new file mode 100644 index 00000000..cfe8f3ac --- /dev/null +++ b/src/bridges/cc-peer/default-front.ts @@ -0,0 +1,49 @@ +/** + * Wires the default cc-peer front (agent-comms#157) onto a real bridge's own store, right after construction -- the piece that actually makes "a machine's coordinator bridge fronts local Claude Code sessions by default" true, rather than a capability every bridge builds but nothing ever switches on. Every real bridge entry point (pi, claude-code, mcp, codex, opencode, user, cc-peer's own one-shot command) calls this once, immediately after createBridgeMesh/createBridgeMeshSync. + * + * Deliberately the only file outside bridges/cc-peer/ that ever needs to know cc-peer exists: core/mesh-store.ts's onCoordinatorRoleChanged hook is generic (a bare boolean callback), and core/bridge-mesh.ts's factories return a plain MeshStore with no cc-peer awareness at all -- keeping this Node/filesystem/cc-peer-specific capability out of the transport-agnostic core, per this repo's own portable-runtime-boundary convention. A bridge that never becomes this machine's coordinator never starts a front at all; one that does gets it started and stopped automatically as that role comes and goes. + */ + +import type { MeshStore } from "../../core/mesh-store.js"; +import { + createDefaultCcPeerFront, + type CreateDefaultCcPeerFrontOptions, +} from "./front-runtime.js"; +import type { CcPeerFront } from "./front-controller.js"; +import type { FrontedSessionRecord } from "./front-controller.js"; + +export interface WireDefaultCcPeerFrontOptions { + coordinatorPort?: number | undefined; + hubUrl?: string | undefined; + /** Builds the front itself -- defaults to the real createDefaultCcPeerFront. Overridable so this function's own start/stop wiring against store.onCoordinatorRoleChanged is testable without a real CcPeer/local Claude Code session. */ + createFront?: + | (( + options: Readonly, + ) => Pick, "start" | "stop">) + | undefined; +} + +/** + * Builds the store.onCoordinatorRoleChanged callback for a real bridge's own store. Returns the callback rather than assigning it directly, so the caller does the actual store.onCoordinatorRoleChanged = ... assignment itself -- this keeps the store parameter here read-only, matching every other pure-construction function in this file's neighbourhood, instead of this function reaching into the store to mutate it. + */ +export function wireDefaultCcPeerFront( + store: Readonly>, + options: Readonly = {}, +): (isCoordinator: boolean) => Promise { + const createFront = options.createFront ?? createDefaultCcPeerFront; + const front = createFront({ + coordinatorPort: options.coordinatorPort, + hubUrl: options.hubUrl, + onError: (error) => { + store.onError?.(error); + }, + }); + + return async (isCoordinator: boolean) => { + if (isCoordinator) { + front.start(); + } else { + await front.stop(); + } + }; +} diff --git a/src/bridges/cc-peer/front-runtime.ts b/src/bridges/cc-peer/front-runtime.ts index d7e8be5d..c5cfb974 100644 --- a/src/bridges/cc-peer/front-runtime.ts +++ b/src/bridges/cc-peer/front-runtime.ts @@ -28,10 +28,10 @@ import { const FRONT_PEER_NAME = "agent-comms-front"; export interface CreateDefaultCcPeerFrontOptions { - coordinatorPort?: number; - hubUrl?: string; - pollIntervalMs?: number; - onError?: (error: Error) => void; + coordinatorPort?: number | undefined; + hubUrl?: string | undefined; + pollIntervalMs?: number | undefined; + onError?: ((error: Error) => void) | undefined; } /** Falls back to "claude-code-" when a session has never picked its own cc-peer display name -- ensureRegistered requires a defaultName, and an unnamed session is still worth fronting under something stable and identifiable. */ diff --git a/src/test/cc-peer-default-front.test.ts b/src/test/cc-peer-default-front.test.ts new file mode 100644 index 00000000..a5334cb7 --- /dev/null +++ b/src/test/cc-peer-default-front.test.ts @@ -0,0 +1,93 @@ +/** + * Unit tests for wireDefaultCcPeerFront -- builds the store.onCoordinatorRoleChanged callback a real bridge assigns to its own store, tested with an injected fake front-builder rather than a real CcPeer/MeshStore. + */ +import { describe, expect, it, vi } from "vitest"; +import { wireDefaultCcPeerFront } from "../bridges/cc-peer/default-front.js"; +import type { WireDefaultCcPeerFrontOptions } from "../bridges/cc-peer/default-front.js"; +import type { MeshStore } from "../core/mesh-store.js"; + +function fakeStore(): Pick { + return { onError: undefined }; +} + +function fakeFront(): { + start: ReturnType void>>; + stop: ReturnType Promise>>; +} { + return { + start: vi.fn<() => void>(), + stop: vi.fn<() => Promise>(async () => Promise.resolve()), + }; +} + +describe("wireDefaultCcPeerFront", () => { + it("starts the front when told the store became coordinator", async () => { + const front = fakeFront(); + const onRoleChanged = wireDefaultCcPeerFront(fakeStore(), { + createFront: () => front, + }); + + await onRoleChanged(true); + + expect(front.start).toHaveBeenCalledTimes(1); + expect(front.stop).not.toHaveBeenCalled(); + }); + + it("stops the front when told the store lost the coordinator role", async () => { + const front = fakeFront(); + const onRoleChanged = wireDefaultCcPeerFront(fakeStore(), { + createFront: () => front, + }); + + await onRoleChanged(false); + + expect(front.stop).toHaveBeenCalledTimes(1); + expect(front.start).not.toHaveBeenCalled(); + }); + + it("builds the front once, up front, not on every role change", async () => { + const createFront = vi.fn(() => fakeFront()); + const onRoleChanged = wireDefaultCcPeerFront(fakeStore(), { createFront }); + + await onRoleChanged(true); + await onRoleChanged(false); + await onRoleChanged(true); + + expect(createFront).toHaveBeenCalledTimes(1); + }); + + it("passes coordinatorPort and hubUrl through to the front builder", () => { + const createFront = vi.fn(() => fakeFront()); + const options: WireDefaultCcPeerFrontOptions = { + coordinatorPort: 20123, + hubUrl: "wss://example.test/hub", + createFront, + }; + wireDefaultCcPeerFront(fakeStore(), options); + + expect(createFront).toHaveBeenCalledWith( + expect.objectContaining({ + coordinatorPort: 20123, + hubUrl: "wss://example.test/hub", + }), + ); + }); + + it("forwards a front error to the store's own onError, when one is set", () => { + const store = fakeStore(); + const onError = vi.fn<(error: Error) => void>(); + store.onError = onError; + let capturedOnError: ((error: Error) => void) | undefined; + wireDefaultCcPeerFront(store, { + createFront: (opts) => { + capturedOnError = opts.onError; + return fakeFront(); + }, + }); + + const error = new Error("cc-peer roster unavailable"); + capturedOnError?.(error); + + expect(onError).toHaveBeenCalledWith(error); + }); +}); From e1ba5b3ef1d539f67b525d8df6da4b896ad8a7d6 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:46:53 +0100 Subject: [PATCH 10/11] feat(bridges): wire the default cc-peer front into every real bridge entry point Every bridge entry point (pi, claude-code, mcp, codex, opencode, user, cc-peer's own one-shot command) now assigns store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(store) right after constructing its own mesh, before init() ever runs. This is what actually makes "a machine's coordinator bridge fronts local Claude Code sessions by default" true for a real deployment, rather than leaving the front's own machinery built but never switched on -- whichever bridge type happens to win this machine's coordinator election starts fronting, and stops the moment it loses that role. user/controller.ts's fromExisting() path (wrapping an already-built store shared with a UI, e.g. pi's own web server) deliberately does not re-wire this -- the store's owning bridge already did, and wiring it twice would silently orphan the first front instance without ever starting it. --- src/bridges/cc-peer/run.ts | 2 ++ src/bridges/claude-code/channel.ts | 2 ++ src/bridges/codex/tool.ts | 2 ++ src/bridges/mcp/server.ts | 2 ++ src/bridges/opencode/plugin.ts | 2 ++ src/bridges/pi/extension.ts | 2 ++ src/bridges/user/controller.ts | 4 ++++ 7 files changed, 16 insertions(+) diff --git a/src/bridges/cc-peer/run.ts b/src/bridges/cc-peer/run.ts index 737aa463..3aca01ec 100644 --- a/src/bridges/cc-peer/run.ts +++ b/src/bridges/cc-peer/run.ts @@ -15,6 +15,7 @@ import { import type { IdentitySlot } from "../../core/identity-store.js"; import { releaseIdentityLock } from "../../core/identity-store.js"; import { wireCcPeerBridge, type CcPeerRef } from "./bridge.js"; +import { wireDefaultCcPeerFront } from "./default-front.js"; const PID_FLAG_PREFIX = "--pid="; /** argv layout for `node cli.js bridge cc-peer `: index 0/1 are the node binary and script path, 2 is "bridge", 3 is the bridge id ("cc-peer") itself -- this bridge's own args start one past that. */ @@ -44,6 +45,7 @@ export async function run(): Promise { const identitySlot: IdentitySlot = { harness: "cc-peer", cwd: process.cwd() }; const { store, tool } = await createBridgeMesh(identitySlot); + store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(store); const reg = await ensureRegistered({ store, diff --git a/src/bridges/claude-code/channel.ts b/src/bridges/claude-code/channel.ts index cb518671..304b3646 100644 --- a/src/bridges/claude-code/channel.ts +++ b/src/bridges/claude-code/channel.ts @@ -33,6 +33,7 @@ import { releaseIdentityLock, type IdentitySlot, } from "../../core/identity-store.js"; +import { wireDefaultCcPeerFront } from "../cc-peer/default-front.js"; import { tryStartWebServer } from "../user/web/server.js"; import { nanoid } from "../../core/nanoid.js"; @@ -164,6 +165,7 @@ export async function run(): Promise { cwd: process.cwd(), }; const { store, tool } = await createBridgeMesh(identitySlot); + store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(store); let agentId: string | undefined; const claudeCodePid = findClaudeCodePid(); diff --git a/src/bridges/codex/tool.ts b/src/bridges/codex/tool.ts index dec3513e..18c5e6e0 100644 --- a/src/bridges/codex/tool.ts +++ b/src/bridges/codex/tool.ts @@ -19,6 +19,7 @@ import { MCP_TOOL_PARAMS, } from "../../core/index.js"; import type { IdentitySlot } from "../../core/identity-store.js"; +import { wireDefaultCcPeerFront } from "../cc-peer/default-front.js"; import { tryStartWebServer } from "../user/web/server.js"; import { nanoid } from "../../core/nanoid.js"; @@ -33,6 +34,7 @@ export async function run(): Promise { // Persistent identity for this slot: a stable device-id means the agent ID survives restarts, so peers can keep targeting us. The stdio server has no graceful shutdown hook; a stale lock self-heals via the pid probe. const identitySlot: IdentitySlot = { harness: "codex", cwd: process.cwd() }; const { store, tool } = await createBridgeMesh(identitySlot); + store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(store); let agentId: string | undefined; const mcp = new McpServer( diff --git a/src/bridges/mcp/server.ts b/src/bridges/mcp/server.ts index 04f5bcf9..6cff481c 100644 --- a/src/bridges/mcp/server.ts +++ b/src/bridges/mcp/server.ts @@ -19,6 +19,7 @@ import { MCP_TOOL_PARAMS, } from "../../core/index.js"; import type { IdentitySlot } from "../../core/identity-store.js"; +import { wireDefaultCcPeerFront } from "../cc-peer/default-front.js"; import { tryStartWebServer } from "../user/web/server.js"; import { nanoid } from "../../core/nanoid.js"; @@ -33,6 +34,7 @@ export async function run(): Promise { // Persistent identity for this slot: a stable device-id means the agent ID survives restarts, so peers can keep targeting us. The stdio server has no graceful shutdown hook; a stale lock self-heals via the pid probe. const identitySlot: IdentitySlot = { harness: "mcp", cwd: process.cwd() }; const { store, tool } = await createBridgeMesh(identitySlot); + store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(store); let agentId: string | undefined; const mcp = new McpServer( diff --git a/src/bridges/opencode/plugin.ts b/src/bridges/opencode/plugin.ts index 548b321f..92fc3707 100644 --- a/src/bridges/opencode/plugin.ts +++ b/src/bridges/opencode/plugin.ts @@ -14,6 +14,7 @@ import { formatDeliveryEvent, } from "../../core/index.js"; import type { IdentitySlot } from "../../core/identity-store.js"; +import { wireDefaultCcPeerFront } from "../cc-peer/default-front.js"; import { tryStartWebServer } from "../user/web/server.js"; import { nanoid } from "../../core/nanoid.js"; @@ -60,6 +61,7 @@ export const AgentCommsPlugin = async (opts: { const client = opts.client; const { store } = await createBridgeMesh(identitySlot); + store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(store); await store.init(); await tryStartWebServer(); diff --git a/src/bridges/pi/extension.ts b/src/bridges/pi/extension.ts index f367e3a3..7906643a 100644 --- a/src/bridges/pi/extension.ts +++ b/src/bridges/pi/extension.ts @@ -31,6 +31,7 @@ import { releaseIdentityLock, type IdentitySlot, } from "../../core/identity-store.js"; +import { wireDefaultCcPeerFront } from "../cc-peer/default-front.js"; import { tryStartWebServer, type WebServerHandle } from "../user/web/server.js"; import { ChatController } from "../user/controller.js"; import { nanoid } from "../../core/nanoid.js"; @@ -45,6 +46,7 @@ export default function (pi: ExtensionAPI) { // Persistent identity for this slot: a stable device-id means the agent ID survives restarts, so peers can keep targeting us const identitySlot: IdentitySlot = { harness: "pi", cwd: process.cwd() }; const { store, tool, attachIdentity } = createBridgeMeshSync(identitySlot); + store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(store); let agentId: string | undefined; let webHandle: WebServerHandle | undefined; diff --git a/src/bridges/user/controller.ts b/src/bridges/user/controller.ts index 14d5f217..f0115fa2 100644 --- a/src/bridges/user/controller.ts +++ b/src/bridges/user/controller.ts @@ -17,6 +17,7 @@ import { releaseIdentityLock, type IdentitySlot, } from "../../core/identity-store.js"; +import { wireDefaultCcPeerFront } from "../cc-peer/default-front.js"; import type { CommsContext, CommsResult } from "../../core/tool.js"; import type { AgentIdentity, @@ -85,6 +86,9 @@ export class ChatController extends EventEmitter { ); this.store = store; this.tool = tool; + this.store.onCoordinatorRoleChanged = wireDefaultCcPeerFront(this.store, { + coordinatorPort: this.coordinatorPort, + }); // Push delivery events to UIs this.store.onDelivery = (_agentId: string, event: DeliveryEvent) => { From 7b4459253d6b4b8b5f6938ef10341a42cd77bf5d Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:48:24 +0100 Subject: [PATCH 11/11] docs(readme): document the default cc-peer front Adds a subsection alongside the existing cc-peer (cross-machine Claude Code relay) docs explaining the coordinator-run default front: identity belongs to the (harness, cwd) slot rather than whichever process currently serves it, no configuration is needed, and an empty local roster degrades to a clean no-op. --- README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/README.md b/README.md index e76d128b..4dc0c62a 100644 --- a/README.md +++ b/README.md @@ -153,6 +153,10 @@ npx agent-comms bridge cc-peer --pid=12345 One bridge process relays for exactly one local Claude Code session, the same "one bridge process is one agent is one device" model every other bridge here follows. Inbound messages from that session are posted into this bridge's own project room; mesh deliveries addressed to this bridge's agent are relayed back to that same session via `cc-peer`'s own `send()`. +### Default cc-peer front + +A Claude Code session with no agent-comms bridge of its own is still reachable from the mesh: whichever bridge on the machine currently holds the mesh coordinator role fronts every local session it discovers via `cc-peer`'s own roster, using the same `(harness, cwd)` identity slot that session's own `claude-code` bridge would use if it started. Identity belongs to the slot, not to whichever process is currently serving it — the session's own bridge holds the slot when it's live; the front holds it otherwise, and yields the moment a real bridge for that slot appears, so addressing, room membership, and queued deliveries all carry over unchanged across the transition. No configuration is needed: every bridge in this repo wires the front to its own coordinator-role transitions automatically, and a machine with no local Claude Code sessions (or no `cc-peer` sockets at all) runs it as a clean no-op. + ## Adding a new harness A bridge is two things: