From acf9bfae97a1c0c4ba2775e2f02ee7bcfc45afbb Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:23:00 +0100 Subject: [PATCH 1/8] feat(errors): add AliasStartError for reply-alias worker startup failures A reply-alias worker process can exit, or crash before registering, before its parent observes a "started" acknowledgement; this gives that failure its own machine-readable error code, matching the existing CcPeerError taxonomy. --- src/errors.ts | 8 ++++++++ src/errors.unit.test.ts | 2 ++ 2 files changed, 10 insertions(+) diff --git a/src/errors.ts b/src/errors.ts index a3bc363..a42e153 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -57,3 +57,11 @@ export class ProtocolError extends CcPeerError { this.name = "ProtocolError"; } } + +/** A reply-alias worker process exited (or failed to reach the "started" acknowledgement) before it finished registering as a discoverable peer. */ +export class AliasStartError extends CcPeerError { + constructor(message: string) { + super("ALIAS_START_FAILED", message); + this.name = "AliasStartError"; + } +} diff --git a/src/errors.unit.test.ts b/src/errors.unit.test.ts index 56d7566..2b1d79d 100644 --- a/src/errors.unit.test.ts +++ b/src/errors.unit.test.ts @@ -8,6 +8,7 @@ import { UnvettedReplyTargetError, NotStartedError, ProtocolError, + AliasStartError, } from "./errors.js"; describe("error taxonomy", () => { @@ -20,6 +21,7 @@ describe("error taxonomy", () => { [new UnvettedReplyTargetError("v"), "UNVETTED_REPLY_TARGET"], [new NotStartedError("s"), "NOT_STARTED"], [new ProtocolError("p"), "PROTOCOL"], + [new AliasStartError("a"), "ALIAS_START_FAILED"], ] as const; for (const [error, code] of cases) { expect(error).toBeInstanceOf(CcPeerError); From 199a9083aa98e733b2aad6cb82c6a7d880de389e Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:24:56 +0100 Subject: [PATCH 2/8] feat(schemas): add IPC contract for reply-alias parent/child processes AliasPool's Node adapter and worker exchange start/stop commands and started/message events over child_process.fork()'s IPC channel; this schema is the single source of truth for that shape, independent of the UDS wire protocol schemas. --- src/schemas/alias-ipc.ts | 52 ++++++++++++ src/schemas/alias-ipc.unit.test.ts | 130 +++++++++++++++++++++++++++++ 2 files changed, 182 insertions(+) create mode 100644 src/schemas/alias-ipc.ts create mode 100644 src/schemas/alias-ipc.unit.test.ts diff --git a/src/schemas/alias-ipc.ts b/src/schemas/alias-ipc.ts new file mode 100644 index 0000000..89974f0 --- /dev/null +++ b/src/schemas/alias-ipc.ts @@ -0,0 +1,52 @@ +import { z } from "zod"; +import { defineSchema } from "./define-schema.js"; + +/** + * The parent-to-child and child-to-parent IPC contract `AliasPool`'s Node adapter (`adapters/node/forked-alias-process.ts`) and worker (`adapters/node/alias-worker.ts`) exchange over `child_process.fork()`'s built-in channel. This is deliberately independent of the wire protocol schemas in `wire.ts`: it never touches a socket, it only carries commands and events between a pool and the real OS process backing one reply alias. + */ +export const AliasStartCommandSchema = defineSchema( + z.object({ + type: z.literal("start"), + name: z.string().min(1), + homeDir: z.string().optional(), + socketDir: z.string().optional(), + sessionId: z.string().optional(), + }), +); +export type AliasStartCommand = z.infer; + +export const AliasStopCommandSchema = defineSchema( + z.object({ type: z.literal("stop") }), +); +export type AliasStopCommand = z.infer; + +export const AliasCommandSchema = defineSchema( + z.union([AliasStartCommandSchema, AliasStopCommandSchema]), +); +export type AliasCommand = z.infer; + +/** Sent once the worker's own CcPeer is listening and registered. */ +export const AliasStartedEventSchema = defineSchema( + z.object({ type: z.literal("started") }), +); +export type AliasStartedEvent = z.infer; + +/** Mirrors CcPeer's own InboundMessage shape, carried over IPC instead of an EventEmitter within one process. */ +export const AliasMessageEventSchema = defineSchema( + z.object({ + type: z.literal("message"), + from: z.string().optional(), + fromSession: z.string().optional(), + fromName: z.string().optional(), + fromMode: z.enum(["bypass", "prompting"]).optional(), + hopChain: z.array(z.string()).optional(), + body: z.string(), + msgId: z.string(), + }), +); +export type AliasMessageEvent = z.infer; + +export const AliasEventSchema = defineSchema( + z.union([AliasStartedEventSchema, AliasMessageEventSchema]), +); +export type AliasEvent = z.infer; diff --git a/src/schemas/alias-ipc.unit.test.ts b/src/schemas/alias-ipc.unit.test.ts new file mode 100644 index 0000000..9b336b8 --- /dev/null +++ b/src/schemas/alias-ipc.unit.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, test } from "vitest"; +import { + AliasStartCommandSchema, + AliasStopCommandSchema, + AliasCommandSchema, + AliasStartedEventSchema, + AliasMessageEventSchema, + AliasEventSchema, +} from "./alias-ipc.js"; + +describe("AliasStartCommandSchema", () => { + test("accepts a name-only start command", () => { + expect(AliasStartCommandSchema.is({ type: "start", name: "alice" })).toBe( + true, + ); + }); + + test("accepts optional homeDir, socketDir, and sessionId", () => { + expect( + AliasStartCommandSchema.is({ + type: "start", + name: "alice", + homeDir: "/tmp/home", + socketDir: "/tmp/socks", + sessionId: "sess-1", + }), + ).toBe(true); + }); + + test("rejects an empty name", () => { + expect(AliasStartCommandSchema.is({ type: "start", name: "" })).toBe(false); + }); + + test("rejects a wrong type discriminant", () => { + expect(AliasStartCommandSchema.is({ type: "stop", name: "alice" })).toBe( + false, + ); + }); +}); + +describe("AliasStopCommandSchema", () => { + test("accepts a bare stop command", () => { + expect(AliasStopCommandSchema.is({ type: "stop" })).toBe(true); + }); + + test("rejects a wrong type discriminant", () => { + expect(AliasStopCommandSchema.is({ type: "start" })).toBe(false); + }); +}); + +describe("AliasCommandSchema", () => { + test("accepts either command shape", () => { + expect(AliasCommandSchema.is({ type: "start", name: "alice" })).toBe(true); + expect(AliasCommandSchema.is({ type: "stop" })).toBe(true); + }); + + test("rejects an unrelated shape", () => { + expect(AliasCommandSchema.is({ type: "ping" })).toBe(false); + }); +}); + +describe("AliasStartedEventSchema", () => { + test("accepts the bare started event", () => { + expect(AliasStartedEventSchema.is({ type: "started" })).toBe(true); + }); + + test("rejects a wrong type discriminant", () => { + expect(AliasStartedEventSchema.is({ type: "message" })).toBe(false); + }); +}); + +const VALID_MESSAGE_EVENT = { + type: "message", + body: "hello", + msgId: "msg-1", +}; + +describe("AliasMessageEventSchema", () => { + test("accepts the minimal required shape", () => { + expect(AliasMessageEventSchema.is(VALID_MESSAGE_EVENT)).toBe(true); + }); + + test("accepts every optional attribution field", () => { + expect( + AliasMessageEventSchema.is({ + ...VALID_MESSAGE_EVENT, + from: "uds:/tmp/cc-socks/9.sock", + fromSession: "sess-9", + fromName: "hopper", + fromMode: "bypass", + hopChain: ["a".repeat(24)], + }), + ).toBe(true); + }); + + test("rejects a missing body", () => { + expect( + AliasMessageEventSchema.is({ type: "message", msgId: "msg-1" }), + ).toBe(false); + }); + + test("rejects an invalid fromMode value", () => { + expect( + AliasMessageEventSchema.is({ + ...VALID_MESSAGE_EVENT, + fromMode: "bogus", + }), + ).toBe(false); + }); + + test("accepts both fromMode enum members", () => { + expect( + AliasMessageEventSchema.is({ + ...VALID_MESSAGE_EVENT, + fromMode: "prompting", + }), + ).toBe(true); + }); +}); + +describe("AliasEventSchema", () => { + test("accepts either event shape", () => { + expect(AliasEventSchema.is({ type: "started" })).toBe(true); + expect(AliasEventSchema.is(VALID_MESSAGE_EVENT)).toBe(true); + }); + + test("rejects an unrelated shape", () => { + expect(AliasEventSchema.is({ type: "pong" })).toBe(false); + }); +}); From b119a5d581ac20c92da5ba9d1f260d017984c353 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:38:47 +0100 Subject: [PATCH 3/8] feat(alias): add the reply-alias worker process and its Node adapter Registering more than one natively-discoverable peer name per process is not possible: the registry is one file per real OS pid (~/.claude/sessions/.json) with a single optional name field each, so a process publishing a second name overwrites its own entry rather than adding to it. Each reply alias therefore needs its own real, distinct OS process. alias-worker.ts is that process's entry point: it registers a standalone CcPeer under a given name and relays every inbound reply back over its fork()'s own IPC channel. ForkedAliasProcess is the Node adapter that forks one, satisfying the new AliasProcess port. alias-worker.ts is deliberately its own dedicated tsdown entry (not left to be pulled in transitively), so it is emitted as a real, directly forkable file at a predictable path rather than folded into a shared internal chunk whose location is a bundler implementation detail. --- src/adapters/node/alias-worker.ts | 39 +++ .../forked-alias-process.integration.test.ts | 188 +++++++++++++++ src/adapters/node/forked-alias-process.ts | 106 ++++++++ .../node/forked-alias-process.unit.test.ts | 228 ++++++++++++++++++ src/ports/alias-process.ts | 16 ++ 5 files changed, 577 insertions(+) create mode 100644 src/adapters/node/alias-worker.ts create mode 100644 src/adapters/node/forked-alias-process.integration.test.ts create mode 100644 src/adapters/node/forked-alias-process.ts create mode 100644 src/adapters/node/forked-alias-process.unit.test.ts create mode 100644 src/ports/alias-process.ts diff --git a/src/adapters/node/alias-worker.ts b/src/adapters/node/alias-worker.ts new file mode 100644 index 0000000..c266be5 --- /dev/null +++ b/src/adapters/node/alias-worker.ts @@ -0,0 +1,39 @@ +/** + * Entry point for one reply alias's real OS process, forked by ForkedAliasProcess. Excluded from the coverage gate the same way src/bin/** is (see vitest.config.ts): pure process-lifecycle glue, exercised end to end by a real fork in forked-alias-process.integration.test.ts rather than in-process unit coverage. + */ +import process from "node:process"; + +import { CcPeer, type InboundMessage } from "../../cc-peer.js"; +import { AliasCommandSchema } from "../../schemas/alias-ipc.js"; + +let peer: CcPeer | undefined; + +process.on("message", (raw: unknown) => { + void handleCommand(raw); +}); + +async function handleCommand(raw: unknown): Promise { + if (!AliasCommandSchema.is(raw)) return; + if (raw.type === "stop") { + await peer?.stop(); + process.exit(0); + return; + } + let created: CcPeer; + try { + created = await CcPeer.create({ + name: raw.name, + ...(raw.homeDir !== undefined ? { homeDir: raw.homeDir } : {}), + ...(raw.socketDir !== undefined ? { socketDir: raw.socketDir } : {}), + ...(raw.sessionId !== undefined ? { sessionId: raw.sessionId } : {}), + }); + } catch { + process.exit(1); + return; + } + peer = created; + peer.on("message", (message: InboundMessage) => { + process.send?.({ type: "message", ...message }); + }); + process.send?.({ type: "started" }); +} diff --git a/src/adapters/node/forked-alias-process.integration.test.ts b/src/adapters/node/forked-alias-process.integration.test.ts new file mode 100644 index 0000000..d44b0bd --- /dev/null +++ b/src/adapters/node/forked-alias-process.integration.test.ts @@ -0,0 +1,188 @@ +import { describe, expect, test } from "vitest"; +import { fork, type ChildProcess } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { mkdtemp, mkdir, readdir, readFile, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { connect } from "node:net"; +import { once } from "node:events"; +import { randomUUID } from "node:crypto"; + +import { ForkedAliasProcess } from "./forked-alias-process.js"; +import { AliasStartError } from "../../errors.js"; +import { REAL_PROCESS_SPAWN_TEST_TIMEOUT_MS } from "../../test/timeouts.js"; +import type { RegistryEntry } from "../../schemas/registry.js"; +import type { PeerKeyFile } from "../../schemas/keyfile.js"; + +/** + * Runs the real alias-worker.ts source under tsx's Node loader hook (registered in-process via --import, never a nested subprocess of its own, so fork()'s IPC channel is unaffected) rather than the built .js sibling ForkedAliasProcess uses by default. That built file only exists once this package has actually been run through tsdown; forking the real TypeScript source directly here proves the worker's own behaviour without requiring a prior `pnpm build`, matching how the SEA binary's own smoke test is instead deferred to a separate, build-gated e2e tier. + */ +const REAL_WORKER_PATH = fileURLToPath( + new URL("./alias-worker.ts", import.meta.url), +); + +function forkViaTsx( + modulePath: string, + args: readonly string[] | undefined, + options: Readonly> | undefined, +): ChildProcess { + return fork(modulePath, args ?? [], { + ...options, + execArgv: ["--import", "tsx"], + }); +} + +function makeAliasProcess(): ForkedAliasProcess { + return new ForkedAliasProcess({ + fork: forkViaTsx as typeof fork, + workerPath: REAL_WORKER_PATH, + }); +} + +async function tempHome(): Promise { + return mkdtemp(join(tmpdir(), "cc-peer-alias-it-")); +} + +async function readAliasRegistryEntry( + homeDir: string, + name: string, +): Promise { + const sessionsDir = join(homeDir, ".claude", "sessions"); + const files = await readdir(sessionsDir); + for (const file of files) { + if (!/^\d+\.json$/.test(file)) continue; + const raw = await readFile(join(sessionsDir, file), "utf8"); + const entry = JSON.parse(raw) as RegistryEntry; + if (entry.name === name) return entry; + } + throw new Error(`no registry entry found for alias ${name}`); +} + +async function readAliasKey( + homeDir: string, + socketPath: string, +): Promise { + const sessionsDir = join(homeDir, ".claude", "sessions"); + const files = await readdir(sessionsDir); + const { createHash } = await import("node:crypto"); + const hash = createHash("sha256").update(socketPath).digest("hex"); + const match = files.find((f) => f.endsWith(`.${hash}.key`)); + if (match === undefined) { + throw new Error(`no key file found for socket ${socketPath}`); + } + const raw = await readFile(join(sessionsDir, match), "utf8"); + return JSON.parse(raw) as PeerKeyFile; +} + +/** Sends one raw wire frame to the alias's socket, mirroring the reference reproduction in docs/PROTOCOL.md. */ +async function sendReplyFrame( + socketPath: string, + token: string, + body: string, +): Promise { + const socket = connect(socketPath); + await once(socket, "connect"); + const envelope = `\n${body}\n`; + const frame = { + msgV: 1, + msg_id: randomUUID(), + type: "user", + message: { role: "user", content: envelope }, + priority: "next", + from: "uds:/tmp/cc-socks/9.sock", + }; + socket.write(`${JSON.stringify({ type: "auth", token })}\n`); + socket.write(`${JSON.stringify(frame)}\n`); + await new Promise((resolve) => { + const timer = setTimeout(resolve, 200); + timer.unref(); + }); + socket.destroy(); +} + +describe("ForkedAliasProcess default fork() fallback", () => { + test( + "with no injected fork function, uses the real node:child_process.fork and fails fast against a nonexistent path", + async () => { + // Proves the `deps.fork ?? fork` fallback (used whenever no fork is injected, i.e. every real production call) computes a real, invokable fork() call rather than merely typechecking. workerPath itself is always required now (see ForkedAliasProcessDeps's own doc comment) — AliasPool.create() is what computes the real default for that (proved by alias-pool.integration.test.ts's own "default wiring" case). + const proc = new ForkedAliasProcess({ + workerPath: "/nonexistent/alias-worker-path.js", + }); + await expect( + proc.start({ name: "unbuilt-default-test" }), + ).rejects.toThrow(AliasStartError); + }, + REAL_PROCESS_SPAWN_TEST_TIMEOUT_MS, + ); +}); + +describe("ForkedAliasProcess against the real alias-worker source", () => { + test( + "registers a discoverable peer and relays an inbound reply", + async () => { + const homeDir = await tempHome(); + const socketDir = join(homeDir, "socks"); + const proc = makeAliasProcess(); + await proc.start({ name: "alice-relay-test", homeDir, socketDir }); + const entry = await readAliasRegistryEntry(homeDir, "alice-relay-test"); + const key = await readAliasKey(homeDir, entry.messagingSocketPath); + const received: unknown[] = []; + proc.events.on("message", (m: unknown) => { + received.push(m); + }); + await sendReplyFrame( + entry.messagingSocketPath, + key.peerToken, + "reply from the relayed session", + ); + const deadline = Date.now() + 5_000; + while (received.length === 0 && Date.now() < deadline) { + await new Promise((resolve) => { + const timer = setTimeout(resolve, 50); + timer.unref(); + }); + } + expect(received).toEqual([ + expect.objectContaining({ body: "reply from the relayed session" }), + ]); + await proc.stop(); + }, + REAL_PROCESS_SPAWN_TEST_TIMEOUT_MS, + ); + + test( + "stop() removes the registry entry and the process exits", + async () => { + const homeDir = await tempHome(); + const socketDir = join(homeDir, "socks"); + const proc = makeAliasProcess(); + await proc.start({ name: "bob-stop-test", homeDir, socketDir }); + await readAliasRegistryEntry(homeDir, "bob-stop-test"); + await proc.stop(); + await expect( + readAliasRegistryEntry(homeDir, "bob-stop-test"), + ).rejects.toThrow(); + }, + REAL_PROCESS_SPAWN_TEST_TIMEOUT_MS, + ); + + test( + "rejects with AliasStartError when the worker fails to start", + async () => { + const homeDir = await tempHome(); + // A socketDir that is actually a regular file, not a directory, makes the worker's own CcPeer.start() throw on mkdir(dirname(socketPath)), so the worker process exits before ever sending "started". + const brokenSocketDir = join(homeDir, "not-a-directory"); + await mkdir(homeDir, { recursive: true }); + await writeFile(brokenSocketDir, "not a directory"); + const proc = makeAliasProcess(); + await expect( + proc.start({ + name: "carol-fail-test", + homeDir, + socketDir: brokenSocketDir, + }), + ).rejects.toThrow(AliasStartError); + }, + REAL_PROCESS_SPAWN_TEST_TIMEOUT_MS, + ); +}); diff --git a/src/adapters/node/forked-alias-process.ts b/src/adapters/node/forked-alias-process.ts new file mode 100644 index 0000000..5e6dcd9 --- /dev/null +++ b/src/adapters/node/forked-alias-process.ts @@ -0,0 +1,106 @@ +import { EventEmitter } from "node:events"; +import { fork, type ChildProcess } from "node:child_process"; + +import type { + AliasProcess, + AliasStartOptions, +} from "../../ports/alias-process.js"; +import { + AliasMessageEventSchema, + AliasStartedEventSchema, + type AliasMessageEvent, +} from "../../schemas/alias-ipc.js"; +import { AliasStartError } from "../../errors.js"; +import type { InboundMessage } from "../../cc-peer.js"; + +export interface ForkedAliasProcessDeps { + /** + * The alias-worker.ts entry's own built location. Deliberately required rather than computed from this module's own `import.meta.url`: tsdown folds this file's compiled code into a shared chunk alongside the other entries, so a sibling-relative URL computed here would resolve against that chunk's own (bundler-chosen, unstable) location rather than the source tree's real layout. alias-pool.ts computes the real path instead, from its own `import.meta.url` — a genuine dedicated entry, guaranteed to sit next to alias-worker.ts's own build output at a predictable relative path (see tsdown.config.ts). + */ + workerPath: string; + fork?: typeof fork; +} + +/** + * The real, Node-backed {@link AliasProcess}: forks a fresh OS process running `alias-worker.ts`, which registers a standalone CcPeer under the given name and relays every inbound reply back over the fork's own IPC channel. + */ +export class ForkedAliasProcess implements AliasProcess { + readonly events = new EventEmitter(); + private readonly forkFn: typeof fork; + private readonly workerPath: string; + private child: ChildProcess | undefined; + + constructor(deps: Readonly) { + this.forkFn = deps.fork ?? fork; + this.workerPath = deps.workerPath; + } + + async start(options: Readonly): Promise { + const child = this.forkFn(this.workerPath, [], {}); + this.child = child; + child.on("message", (raw: unknown) => { + if (AliasMessageEventSchema.is(raw)) { + this.events.emit("message", toInboundMessage(raw)); + } + }); + child.on("exit", () => { + this.events.emit("exit"); + }); + await new Promise((resolve, reject) => { + function onExitBeforeStart(code: number | null): void { + child.off("message", onStarted); + reject( + new AliasStartError( + `alias worker exited before starting (code ${code === null ? "null" : code.toString()})`, + ), + ); + } + function onStarted(raw: unknown): void { + if (AliasStartedEventSchema.is(raw)) { + child.off("message", onStarted); + child.off("exit", onExitBeforeStart); + resolve(); + } + } + child.on("message", onStarted); + child.once("exit", onExitBeforeStart); + child.send({ + type: "start", + name: options.name, + ...(options.homeDir !== undefined ? { homeDir: options.homeDir } : {}), + ...(options.socketDir !== undefined + ? { socketDir: options.socketDir } + : {}), + ...(options.sessionId !== undefined + ? { sessionId: options.sessionId } + : {}), + }); + }); + } + + async stop(): Promise { + const child = this.child; + if (child === undefined) return; + if (child.exitCode !== null || child.signalCode !== null) return; + await new Promise((resolve) => { + child.once("exit", () => { + resolve(); + }); + child.send({ type: "stop" }); + }); + } +} + +function toInboundMessage(event: Readonly): InboundMessage { + return { + ...(event.from !== undefined ? { from: event.from } : {}), + ...(event.fromSession !== undefined + ? { fromSession: event.fromSession } + : {}), + ...(event.fromName !== undefined ? { fromName: event.fromName } : {}), + ...(event.fromMode !== undefined ? { fromMode: event.fromMode } : {}), + ...(event.hopChain !== undefined ? { hopChain: event.hopChain } : {}), + body: event.body, + msgId: event.msgId, + }; +} diff --git a/src/adapters/node/forked-alias-process.unit.test.ts b/src/adapters/node/forked-alias-process.unit.test.ts new file mode 100644 index 0000000..e7b45e1 --- /dev/null +++ b/src/adapters/node/forked-alias-process.unit.test.ts @@ -0,0 +1,228 @@ +import { describe, expect, test, vi } from "vitest"; +import { EventEmitter } from "node:events"; +import type { ChildProcess } from "node:child_process"; + +import { ForkedAliasProcess } from "./forked-alias-process.js"; +import { AliasStartError } from "../../errors.js"; + +/** A fake ChildProcess: just enough of the EventEmitter + send() surface for the adapter's own protocol logic, driven manually by each test. */ +class FakeChild extends EventEmitter { + sent: unknown[] = []; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + + send(message: unknown): boolean { + this.sent.push(message); + return true; + } + + emitExit(code: number | null): void { + this.exitCode = code; + this.emit("exit", code); + } +} + +function makeForkedAliasProcess(child: FakeChild): { + process: ForkedAliasProcess; + forkCalls: unknown[][]; +} { + const forkCalls: unknown[][] = []; + const fork = vi.fn((modulePath: string, args: unknown, options: unknown) => { + forkCalls.push([modulePath, args, options]); + return child as unknown as ChildProcess; + }); + return { + process: new ForkedAliasProcess({ + fork: fork as never, + workerPath: "/fake/worker.js", + }), + forkCalls, + }; +} + +describe("ForkedAliasProcess.start", () => { + test("forks the configured worker path with no extra args", async () => { + const child = new FakeChild(); + const { process: proc, forkCalls } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { type: "started" }); + await pending; + expect(forkCalls).toEqual([["/fake/worker.js", [], {}]]); + }); + + test("sends a start command built from the given options", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ + name: "alice", + homeDir: "/tmp/home", + socketDir: "/tmp/socks", + sessionId: "sess-1", + }); + child.emit("message", { type: "started" }); + await pending; + expect(child.sent).toEqual([ + { + type: "start", + name: "alice", + homeDir: "/tmp/home", + socketDir: "/tmp/socks", + sessionId: "sess-1", + }, + ]); + }); + + test("resolves once the child sends a started acknowledgement", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + let resolved = false; + void pending.then(() => { + resolved = true; + }); + await Promise.resolve(); + expect(resolved).toBe(false); + child.emit("message", { type: "started" }); + await pending; + expect(resolved).toBe(true); + }); + + test("ignores a malformed inbound message while waiting to start", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { totally: "unrelated" }); + child.emit("message", { type: "started" }); + await expect(pending).resolves.toBeUndefined(); + }); + + test("rejects with AliasStartError if the child exits before starting", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emitExit(1); + await expect(pending).rejects.toThrow(AliasStartError); + await expect(pending).rejects.toThrow(/exited before starting/); + }); + + test("reports a null exit code as null in the rejection message", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emitExit(null); + await expect(pending).rejects.toThrow(/code null/); + }); +}); + +describe("ForkedAliasProcess message relay", () => { + test("relays a well-formed message event to its own events emitter", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { type: "started" }); + await pending; + const received: unknown[] = []; + proc.events.on("message", (m: unknown) => { + received.push(m); + }); + child.emit("message", { + type: "message", + body: "hi", + msgId: "m1", + fromName: "alice", + }); + expect(received).toEqual([{ body: "hi", msgId: "m1", fromName: "alice" }]); + }); + + test("carries every optional attribution field through untouched", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { type: "started" }); + await pending; + const received: unknown[] = []; + proc.events.on("message", (m: unknown) => { + received.push(m); + }); + child.emit("message", { + type: "message", + body: "hi", + msgId: "m1", + from: "uds:/tmp/cc-socks/9.sock", + fromSession: "sess-9", + fromName: "hopper", + fromMode: "bypass", + hopChain: ["a".repeat(24)], + }); + expect(received).toEqual([ + { + body: "hi", + msgId: "m1", + from: "uds:/tmp/cc-socks/9.sock", + fromSession: "sess-9", + fromName: "hopper", + fromMode: "bypass", + hopChain: ["a".repeat(24)], + }, + ]); + }); + + test("ignores a malformed message event after starting", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { type: "started" }); + await pending; + const received: unknown[] = []; + proc.events.on("message", (m: unknown) => { + received.push(m); + }); + child.emit("message", { type: "message" }); + expect(received).toEqual([]); + }); + + test("emits its own exit event once the backing process exits after starting", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { type: "started" }); + await pending; + const exits: unknown[] = []; + proc.events.on("exit", () => { + exits.push(true); + }); + child.emitExit(0); + expect(exits).toEqual([true]); + }); +}); + +describe("ForkedAliasProcess.stop", () => { + test("sends a stop command and resolves once the child exits", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { type: "started" }); + await pending; + const stopping = proc.stop(); + expect(child.sent).toContainEqual({ type: "stop" }); + child.emitExit(0); + await stopping; + }); + + test("stopping before start() has ever been called is a harmless no-op", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + await expect(proc.stop()).resolves.toBeUndefined(); + }); + + test("stopping an already-exited process is a harmless no-op", async () => { + const child = new FakeChild(); + const { process: proc } = makeForkedAliasProcess(child); + const pending = proc.start({ name: "alice" }); + child.emit("message", { type: "started" }); + await pending; + child.emitExit(0); + await expect(proc.stop()).resolves.toBeUndefined(); + expect(child.sent).not.toContainEqual({ type: "stop" }); + }); +}); diff --git a/src/ports/alias-process.ts b/src/ports/alias-process.ts new file mode 100644 index 0000000..8404121 --- /dev/null +++ b/src/ports/alias-process.ts @@ -0,0 +1,16 @@ +import type { EventEmitter } from "node:events"; +import type { PathConfig } from "../adapters/node/paths.js"; + +export interface AliasStartOptions extends PathConfig { + name: string; + sessionId?: string; +} + +/** + * A synthetic, natively-discoverable Claude Code peer identity backed by a real OS process. A real process is required, not a design choice: the registry is one file per real pid with a single optional name field each (see docs/PROTOCOL.md's "Reply aliases" section), so a single process can only ever publish one discoverable name at a time. `events` emits "message" (an InboundMessage-shaped object) for each reply the alias receives, and "exit" once the backing process has ended, whether from a deliberate stop() or an unexpected crash. + */ +export interface AliasProcess { + readonly events: EventEmitter; + start: (options: Readonly) => Promise; + stop: () => Promise; +} From 30511ad93f659522f66964c78876869ac496c84f Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:38:59 +0100 Subject: [PATCH 4/8] feat(alias): add AliasPool for lazily managing per-correspondent aliases Lazily materialises one ForkedAliasProcess per mesh correspondent name on ensure(), dedup'd against concurrent calls for the same name, and relays each alias's own "message" and "exit" events with the correspondent name attached. retire() waits for an in-flight ensure() of the same name to settle before stopping it, so a retire issued while an alias is still starting still stops it once (and if) it becomes active. --- src/alias-pool.integration.test.ts | 17 ++ src/alias-pool.ts | 122 +++++++++++++ src/alias-pool.unit.test.ts | 277 +++++++++++++++++++++++++++++ 3 files changed, 416 insertions(+) create mode 100644 src/alias-pool.integration.test.ts create mode 100644 src/alias-pool.ts create mode 100644 src/alias-pool.unit.test.ts diff --git a/src/alias-pool.integration.test.ts b/src/alias-pool.integration.test.ts new file mode 100644 index 0000000..e65a317 --- /dev/null +++ b/src/alias-pool.integration.test.ts @@ -0,0 +1,17 @@ +import { describe, expect, test } from "vitest"; + +import { AliasPool } from "./alias-pool.js"; +import { REAL_PROCESS_SPAWN_TEST_TIMEOUT_MS } from "./test/timeouts.js"; + +describe("AliasPool.create default wiring", () => { + test( + "ensure() drives a real ForkedAliasProcess, which fails fast against the unbuilt worker path in this dev tree", + async () => { + // Exercises AliasPool.create()'s own spawn wiring for real (a bare unit test with an injected fake spawn never calls the real ForkedAliasProcess constructor). In this repo's own dev/test tree that means a real fork() against the .js sibling worker path, which only exists once `pnpm build` has run (see forked-alias-process.integration.test.ts's own default-construction test) — so this fails fast rather than hanging, which is itself the behaviour worth pinning: a broken or missing worker build surfaces as a rejected ensure(), never a silent no-op. + const pool = AliasPool.create(); + await expect(pool.ensure("unbuilt-default-pool-test")).rejects.toThrow(); + expect(pool.activeAliases()).toEqual([]); + }, + REAL_PROCESS_SPAWN_TEST_TIMEOUT_MS, + ); +}); diff --git a/src/alias-pool.ts b/src/alias-pool.ts new file mode 100644 index 0000000..83de0c5 --- /dev/null +++ b/src/alias-pool.ts @@ -0,0 +1,122 @@ +import { EventEmitter } from "node:events"; +import { fileURLToPath } from "node:url"; + +import { ForkedAliasProcess } from "./adapters/node/forked-alias-process.js"; +import type { PathConfig } from "./adapters/node/paths.js"; +import type { AliasProcess } from "./ports/alias-process.js"; +import type { InboundMessage } from "./cc-peer.js"; + +export interface AliasPoolOptions extends PathConfig { + logger?: (message: string) => void; +} + +/** An inbound reply relayed from one alias, with the correspondent name it arrived for attached. */ +export interface AliasMessage extends InboundMessage { + alias: string; +} + +/** Default log sink: logs go nowhere unless a logger is provided, matching CcPeer's own default. */ +const sinkLog = (): void => undefined; + +/** + * alias-worker.ts's own built location, computed from this file's `import.meta.url` rather than ForkedAliasProcess's own: this file is itself a dedicated tsdown entry (see tsdown.config.ts), so its emitted location is stable and predictable, and alias-worker.ts (also a dedicated entry) is emitted at the same relative position to it as the two source files hold to each other in src/. The target extension mirrors this file's own real extension at runtime (`.mjs` or `.cjs`, tsdown's `fixedExtension: true`) rather than a hardcoded `.js`: tsdown builds matching-format sibling entries for every declared output, so alias-pool.mjs's own sibling is always alias-worker.mjs, never a plain `.js` neither format ever actually emits. + */ +const DEFAULT_WORKER_PATH = fileURLToPath( + new URL( + `./adapters/node/alias-worker${import.meta.url.endsWith(".cjs") ? ".cjs" : ".mjs"}`, + import.meta.url, + ), +); + +interface Deps { + spawn: () => AliasProcess; +} + +/** + * Lazily materialises one natively-discoverable Claude Code peer identity per mesh correspondent name, each backed by its own real OS process (see ports/alias-process.ts for why a real process is required). Emits "message" (an {@link AliasMessage}) for every reply an alias receives, and "exit" (an object with an `alias` name field) when an alias's backing process ends, whether from a deliberate retire() or an unexpected crash. + */ +export class AliasPool extends EventEmitter { + private readonly active = new Map(); + private readonly pending = new Map>(); + private readonly log: (message: string) => void; + + constructor( + private readonly deps: Readonly, + private readonly options: Readonly = {}, + ) { + super(); + this.log = options.logger ?? sinkLog; + } + + static create(options: Readonly = {}): AliasPool { + return new AliasPool( + { + spawn: () => + new ForkedAliasProcess({ workerPath: DEFAULT_WORKER_PATH }), + }, + options, + ); + } + + /** Idempotent: a no-op if the alias is already active, and dedup'd if another ensure() for the same name is already in flight. */ + async ensure(name: string): Promise { + if (this.active.has(name)) return; + const inFlight = this.pending.get(name); + if (inFlight !== undefined) { + await inFlight; + return; + } + const started = this.startAlias(name); + this.pending.set(name, started); + try { + await started; + } finally { + this.pending.delete(name); + } + } + + private async startAlias(name: string): Promise { + const proc = this.deps.spawn(); + proc.events.on("message", (message: InboundMessage) => { + this.emit("message", { alias: name, ...message } satisfies AliasMessage); + }); + proc.events.on("exit", () => { + this.active.delete(name); + this.emit("exit", { alias: name }); + }); + await proc.start({ + name, + ...(this.options.homeDir !== undefined + ? { homeDir: this.options.homeDir } + : {}), + ...(this.options.socketDir !== undefined + ? { socketDir: this.options.socketDir } + : {}), + }); + this.active.set(name, proc); + this.log(`alias ${name} active`); + } + + /** Waits for any in-flight ensure() of the same name to settle first, so a retire() issued while an alias is still starting stops it once (and if) it becomes active. A no-op for a name that is neither active nor pending. */ + async retire(name: string): Promise { + const inFlight = this.pending.get(name); + if (inFlight !== undefined) { + await inFlight.catch(() => undefined); + } + const proc = this.active.get(name); + if (proc === undefined) return; + this.active.delete(name); + await proc.stop(); + this.log(`alias ${name} retired`); + } + + async stopAll(): Promise { + await Promise.all( + [...this.active.keys()].map(async (name) => this.retire(name)), + ); + } + + activeAliases(): string[] { + return [...this.active.keys()]; + } +} diff --git a/src/alias-pool.unit.test.ts b/src/alias-pool.unit.test.ts new file mode 100644 index 0000000..0cd7008 --- /dev/null +++ b/src/alias-pool.unit.test.ts @@ -0,0 +1,277 @@ +import { describe, expect, test, vi } from "vitest"; +import { EventEmitter } from "node:events"; + +import { AliasPool } from "./alias-pool.js"; +import type { AliasProcess } from "./ports/alias-process.js"; + +/** A fake AliasProcess whose start()/stop() are externally controllable, so tests can assert ordering and concurrency without a real child process. */ +class FakeAliasProcess implements AliasProcess { + readonly events = new EventEmitter(); + startCalls: unknown[] = []; + stopCalls = 0; + private resolveStart: (() => void) | undefined; + private rejectStart: ((error: Error) => void) | undefined; + + async start(options: unknown): Promise { + this.startCalls.push(options); + return new Promise((resolve, reject) => { + this.resolveStart = resolve; + this.rejectStart = reject; + }); + } + + async stop(): Promise { + this.stopCalls += 1; + return Promise.resolve(); + } + + finishStart(): void { + this.resolveStart?.(); + } + + failStart(error: Error): void { + this.rejectStart?.(error); + } +} + +function makePool(procs: readonly FakeAliasProcess[]): { + pool: AliasPool; + spawnCount: () => number; +} { + let index = 0; + const spawn = vi.fn(() => { + const proc = procs[index]; + index += 1; + if (proc === undefined) { + throw new Error("makePool: not enough fake processes provided"); + } + return proc; + }); + return { + pool: new AliasPool({ spawn }), + spawnCount: () => spawn.mock.calls.length, + }; +} + +describe("AliasPool.ensure", () => { + test("spawns a process and passes the alias name through to start()", async () => { + const proc = new FakeAliasProcess(); + const { pool } = makePool([proc]); + const pending = pool.ensure("alice"); + proc.finishStart(); + await pending; + expect(proc.startCalls).toEqual([{ name: "alice" }]); + expect(pool.activeAliases()).toEqual(["alice"]); + }); + + test("passes pool-level homeDir/socketDir through to start()", async () => { + const proc = new FakeAliasProcess(); + let index = 0; + const spawn = vi.fn(() => { + index += 1; + return proc; + }); + const pool = new AliasPool( + { spawn }, + { homeDir: "/tmp/home", socketDir: "/tmp/socks" }, + ); + const pending = pool.ensure("alice"); + proc.finishStart(); + await pending; + expect(index).toBe(1); + expect(proc.startCalls).toEqual([ + { name: "alice", homeDir: "/tmp/home", socketDir: "/tmp/socks" }, + ]); + }); + + test("a second ensure() for an already-active alias is a no-op", async () => { + const proc = new FakeAliasProcess(); + const { pool, spawnCount } = makePool([proc]); + const first = pool.ensure("alice"); + proc.finishStart(); + await first; + await pool.ensure("alice"); + expect(spawnCount()).toBe(1); + }); + + test("concurrent ensure() calls for the same name spawn only once", async () => { + const proc = new FakeAliasProcess(); + const { pool, spawnCount } = makePool([proc]); + const first = pool.ensure("alice"); + const second = pool.ensure("alice"); + proc.finishStart(); + await Promise.all([first, second]); + expect(spawnCount()).toBe(1); + expect(pool.activeAliases()).toEqual(["alice"]); + }); + + test("ensure() for two different names spawns two processes", async () => { + const alice = new FakeAliasProcess(); + const bob = new FakeAliasProcess(); + const { pool, spawnCount } = makePool([alice, bob]); + const pending = Promise.all([pool.ensure("alice"), pool.ensure("bob")]); + alice.finishStart(); + bob.finishStart(); + await pending; + expect(spawnCount()).toBe(2); + expect(pool.activeAliases().sort()).toEqual(["alice", "bob"]); + }); + + test("a failed start() rejects ensure() and leaves the alias inactive", async () => { + const proc = new FakeAliasProcess(); + const { pool } = makePool([proc]); + const pending = pool.ensure("alice"); + proc.failStart(new Error("boom")); + await expect(pending).rejects.toThrow("boom"); + expect(pool.activeAliases()).toEqual([]); + }); + + test("ensure() can be retried after a failed start()", async () => { + const failing = new FakeAliasProcess(); + const retry = new FakeAliasProcess(); + const { pool, spawnCount } = makePool([failing, retry]); + const firstAttempt = pool.ensure("alice"); + failing.failStart(new Error("boom")); + await expect(firstAttempt).rejects.toThrow("boom"); + const secondAttempt = pool.ensure("alice"); + retry.finishStart(); + await secondAttempt; + expect(spawnCount()).toBe(2); + expect(pool.activeAliases()).toEqual(["alice"]); + }); +}); + +describe("AliasPool message and exit relay", () => { + test("relays a process's message event with the alias name attached", async () => { + const proc = new FakeAliasProcess(); + const { pool } = makePool([proc]); + const pending = pool.ensure("alice"); + proc.finishStart(); + await pending; + const received: unknown[] = []; + pool.on("message", (m: unknown) => { + received.push(m); + }); + proc.events.emit("message", { body: "hi", msgId: "m1" }); + expect(received).toEqual([{ alias: "alice", body: "hi", msgId: "m1" }]); + }); + + test("an unexpected process exit removes the alias and emits a pool exit event", async () => { + const proc = new FakeAliasProcess(); + const { pool } = makePool([proc]); + const pending = pool.ensure("alice"); + proc.finishStart(); + await pending; + const exits: unknown[] = []; + pool.on("exit", (e: unknown) => { + exits.push(e); + }); + proc.events.emit("exit"); + expect(exits).toEqual([{ alias: "alice" }]); + expect(pool.activeAliases()).toEqual([]); + }); +}); + +describe("AliasPool.retire", () => { + test("stops an active alias and removes it", async () => { + const proc = new FakeAliasProcess(); + const { pool } = makePool([proc]); + const pending = pool.ensure("alice"); + proc.finishStart(); + await pending; + await pool.retire("alice"); + expect(proc.stopCalls).toBe(1); + expect(pool.activeAliases()).toEqual([]); + }); + + test("retiring an unknown alias is a harmless no-op", async () => { + const { pool } = makePool([]); + await expect(pool.retire("ghost")).resolves.toBeUndefined(); + }); + + test("retire waits for an in-flight ensure() and then stops it", async () => { + const proc = new FakeAliasProcess(); + const { pool } = makePool([proc]); + const ensurePromise = pool.ensure("alice"); + const retirePromise = pool.retire("alice"); + proc.finishStart(); + await ensurePromise; + await retirePromise; + expect(proc.stopCalls).toBe(1); + expect(pool.activeAliases()).toEqual([]); + }); + + test("retire waits for an in-flight ensure() that fails without throwing", async () => { + const proc = new FakeAliasProcess(); + const { pool } = makePool([proc]); + const ensurePromise = pool.ensure("alice"); + const retirePromise = pool.retire("alice"); + proc.failStart(new Error("boom")); + await expect(ensurePromise).rejects.toThrow("boom"); + await expect(retirePromise).resolves.toBeUndefined(); + expect(proc.stopCalls).toBe(0); + }); +}); + +describe("AliasPool.stopAll", () => { + test("retires every active alias", async () => { + const alice = new FakeAliasProcess(); + const bob = new FakeAliasProcess(); + const { pool } = makePool([alice, bob]); + const pending = Promise.all([pool.ensure("alice"), pool.ensure("bob")]); + alice.finishStart(); + bob.finishStart(); + await pending; + await pool.stopAll(); + expect(alice.stopCalls).toBe(1); + expect(bob.stopCalls).toBe(1); + expect(pool.activeAliases()).toEqual([]); + }); + + test("on an empty pool is a harmless no-op", async () => { + const { pool } = makePool([]); + await expect(pool.stopAll()).resolves.toBeUndefined(); + }); +}); + +describe("AliasPool logging", () => { + test("logs when an alias becomes active and when it is retired", async () => { + const proc = new FakeAliasProcess(); + const spawn = vi.fn(() => proc); + const messages: string[] = []; + const pool = new AliasPool( + { spawn }, + { + logger: (m) => { + messages.push(m); + }, + }, + ); + const pending = pool.ensure("alice"); + proc.finishStart(); + await pending; + await pool.retire("alice"); + expect( + messages.some((m) => m.includes("alice") && m.includes("active")), + ).toBe(true); + expect( + messages.some((m) => m.includes("alice") && m.includes("retired")), + ).toBe(true); + }); + + test("create() without a logger starts cleanly (sink log path)", async () => { + const proc = new FakeAliasProcess(); + const spawn = vi.fn(() => proc); + const pool = new AliasPool({ spawn }); + const pending = pool.ensure("alice"); + proc.finishStart(); + await expect(pending).resolves.toBeUndefined(); + }); +}); + +describe("AliasPool.create", () => { + test("returns a pool with no active aliases and no logger required", () => { + const pool = AliasPool.create(); + expect(pool.activeAliases()).toEqual([]); + }); +}); From c06f74cbdb0dca853a07393e10d32ee4370836e2 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:39:10 +0100 Subject: [PATCH 5/8] build: publish AliasPool as its own cc-peer/alias-pool entry Adds a dedicated tsdown entry and package.json subpath export for alias-pool.ts, mirroring the existing pattern for the "." entry, rather than re-exporting it through cc-peer.ts (barrel re-exports are banned by this repo's own eslint config). Also excludes alias-worker.ts from the coverage gate, matching src/bin/** and sea-entry.ts: it is process lifecycle glue exercised by a real fork in an integration test, not by in-process unit coverage. --- package.json | 10 ++++++++++ tsdown.config.ts | 8 +++++++- vitest.config.ts | 3 ++- 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/package.json b/package.json index b669ee9..f00dddc 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,16 @@ "default": "./dist/cc-peer.cjs" } }, + "./alias-pool": { + "import": { + "types": "./dist/alias-pool.d.mts", + "default": "./dist/alias-pool.mjs" + }, + "require": { + "types": "./dist/alias-pool.d.cts", + "default": "./dist/alias-pool.cjs" + } + }, "./schemas/*.schema.json": "./schemas/*.schema.json", "./package.json": "./package.json" }, diff --git a/tsdown.config.ts b/tsdown.config.ts index 08fc4dd..71c7a3e 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -18,7 +18,13 @@ const seaEntryBuildConfig: UserConfig = { export default defineConfig([ { - entry: ["src/cc-peer.ts", "src/bin/cc-peer.ts"], + // alias-worker.ts is listed as its own dedicated entry, not left to be pulled in transitively by alias-pool.ts, specifically so it is emitted as a real, directly-forkable file at a predictable path (tsdown preserves each declared entry's own src/ directory structure in dist/, so dist/adapters/node/alias-worker.mjs sits at exactly the same relative position to dist/alias-pool.mjs as the two source files do to each other) rather than being folded into one of the shared internal chunks tsdown otherwise splits code across entries into, whose exact filenames and locations are an implementation detail child_process.fork() cannot rely on. + entry: [ + "src/cc-peer.ts", + "src/bin/cc-peer.ts", + "src/alias-pool.ts", + "src/adapters/node/alias-worker.ts", + ], outDir: "dist", format: ["esm", "cjs"], dts: true, diff --git a/vitest.config.ts b/vitest.config.ts index 2c34ae2..f797363 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -22,13 +22,14 @@ export default defineConfig({ coverage: { provider: "v8", include: ["src/**/*.ts"], - // Entry glue (bin firing wrappers, the SEA entry) is exercised by the spawn-based smoke and SEA smoke tests, not by in-process unit coverage; ports are type-only declarations with no runtime statements to cover. + // Entry glue (bin firing wrappers, the SEA entry, the reply-alias worker) is exercised by the spawn-based smoke tests, not by in-process unit coverage; ports are type-only declarations with no runtime statements to cover. exclude: [ "src/**/*.test.ts", "src/test/**", "src/bin/**", "src/sea-entry.ts", "src/ports/**", + "src/adapters/node/alias-worker.ts", ], reporter: ["text", "html", "json-summary", "json"], thresholds: { From 8f571101d15114a0eb2be6f29cc294020e64f427 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:39:17 +0100 Subject: [PATCH 6/8] docs: document session enumeration and reply-alias findings Records the empirical finding behind AliasPool: session enumeration is already fully native (the roster builder reads every registry file on disk, not just cc-peer's own), while reply aliases genuinely are not, because the registry is one file per real pid with a single name each. README documents cc-peer/alias-pool's own usage. --- README.md | 26 ++++++++++++++++++++++++++ docs/PROTOCOL.md | 9 +++++++++ 2 files changed, 35 insertions(+) diff --git a/README.md b/README.md index 1c72707..361625b 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,32 @@ Every release is also mirrored to the GitHub Packages registry as `@exadev/cc-pe The REST facade (`npx cc-peer`) serves `GET /sessions`, `POST /messages`, `POST /idle-subscriptions`, `GET /events` (SSE), and a self-describing `GET /openapi.json` on loopback with a bearer token. +### Session discovery and reply aliases + +`CcPeer.roster()` already lists every live local Claude Code session, not just ones `cc-peer` itself registered — the registry it reads (`~/.claude/sessions/*.json`) is written by every interactive session on startup. A relay/front application that wants to discover every session to attach to needs nothing beyond `roster()`. + +Giving a relayed session a name it can reply to natively for each of several correspondents is a different problem: the registry is one file per real OS pid with a single name each, so one process can only ever publish one discoverable name at a time (see [docs/PROTOCOL.md](docs/PROTOCOL.md#session-enumeration-and-reply-aliases-for-a-relayfront-building-on-this-sdk) for the empirical detail). `AliasPool`, exported from `cc-peer/alias-pool`, is the mechanism for this: it lazily forks one lightweight `CcPeer`-backed child process per correspondent name, and relays whatever that alias receives back to the parent. + +```ts +import { AliasPool } from "cc-peer/alias-pool"; + +const aliases = AliasPool.create(); + +aliases.on("message", (m) => { + // m.alias is the correspondent name the relayed session replied to; + // forward m.body to that correspondent's own channel. + console.log(`reply for ${m.alias}: ${m.body}`); +}); + +// Whenever a new correspondent messages the relayed session for the first +// time, give it a reply-able name (idempotent; a no-op if already active). +await aliases.ensure("alice"); + +// …later, once a correspondent is no longer relevant: +await aliases.retire("alice"); +await aliases.stopAll(); +``` + ## Limitations - **Same-process constraint**: receipts and idle notices only reach the process that owns the peer's listening socket (the protocol verifies return addresses via kernel peer-pids). Do not split `CcPeer` listening and sending across processes or differently-owned workers. diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index bb6e475..211eeec 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -76,6 +76,15 @@ The roster builder (`listLivePeerSessions`) reads all `~/.claude/sessions/. Verified chain: a Python peer registered this way appears in `ListAgents` within seconds and receives native `SendMessage` by bare name (`from-name` resolves from the sender's own registry entry). +### Session enumeration and reply aliases (for a relay/front building on this SDK) + +Two capabilities a message relay ("front") needs from this protocol, gated on what it actually supports rather than assumed: + +- **Session enumeration** — listing every live local Claude Code session, not just ones the relay itself registered — is already fully native. The roster builder above reads every `~/.claude/sessions/.json` file on disk, regardless of who wrote it; `cc-peer`'s own `CcPeer.roster()` (and the REST facade's `GET /sessions`) is exactly this roster builder, so a relay gets full session discovery for free, with no separate mechanism needed. +- **Reply aliases** — giving each correspondent that messages a relayed session its own natively-`SendMessage`-reachable name, so the session can reply to it directly by name — is **not** natively supported for more than one name per process. The registry is one file per real OS pid (`registryFilePath`: `.json`) and each entry carries a single optional `name` field; a process publishing a second name overwrites, rather than adds to, its own entry. This is directly observable in this SDK's own test suite: two `CcPeer` instances sharing one pid (unavoidable — both are the same OS process) leave only the last-registered name visible in the roster, because both wrote to the identical `.json` file. Native name resolution (`ListAgents`/`SendMessage(name=X)`) walks the registry directory exactly as it is on disk — it has no concept of "this one process answers to several names." + +The practical consequence: a relay that wants N correspondents to each get their own reply-able name needs N distinct, genuinely live OS processes — one real pid, one registry file, one name, per correspondent — not a lighter-weight in-process mapping. `cc-peer`'s own `AliasPool` (see the root README) implements exactly this: it lazily forks one lightweight child process per correspondent name, each running an ordinary `CcPeer` instance under that name, and relays whatever the relayed session replies with back to the parent process for translation into whatever channel the correspondent actually lives on. + ## Receipts and status `peer_message_status` is pushed from receiver to sender over a fresh connection to the sender's socket, authenticated with the sender's own peerToken: From 642bb6846a93c4fc1919118170aa2a6a5ee5808d Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:48:18 +0100 Subject: [PATCH 7/8] fix(alias): close a coverage gap and a Windows-only test failure Extracts the worker-path extension derivation into workerExtensionFor(), a pure function both branches can be exercised directly rather than only through this module's own import.meta.url, which in dev/test always ends in .ts and so could only ever reach the .mjs branch. Also fixes the "worker fails to start" integration test: pointing socketDir directly at a plain file (not a directory) forces mkdir() to throw on macOS and Linux, but recursive mkdir() against an already-existing path does not appear to verify it is actually a directory on Windows, so the worker started successfully there instead of failing. Nesting socketDir one level inside the broken file forces mkdir() to create a genuinely new directory entry inside a file, which has no valid resolution on any platform. --- .../forked-alias-process.integration.test.ts | 7 ++++--- src/alias-pool.ts | 11 +++++++++-- src/alias-pool.unit.test.ts | 16 +++++++++++++++- 3 files changed, 28 insertions(+), 6 deletions(-) diff --git a/src/adapters/node/forked-alias-process.integration.test.ts b/src/adapters/node/forked-alias-process.integration.test.ts index d44b0bd..6f47366 100644 --- a/src/adapters/node/forked-alias-process.integration.test.ts +++ b/src/adapters/node/forked-alias-process.integration.test.ts @@ -170,10 +170,11 @@ describe("ForkedAliasProcess against the real alias-worker source", () => { "rejects with AliasStartError when the worker fails to start", async () => { const homeDir = await tempHome(); - // A socketDir that is actually a regular file, not a directory, makes the worker's own CcPeer.start() throw on mkdir(dirname(socketPath)), so the worker process exits before ever sending "started". - const brokenSocketDir = join(homeDir, "not-a-directory"); + // A socketDir nested one level inside a plain regular file makes the worker's own CcPeer.start() throw on mkdir(dirname(socketPath)): creating a new directory entry inside a file (not merely re-stating an already-existing path) has no valid filesystem resolution on any platform. Nesting one level deeper than the broken file itself is load-bearing: pointing socketDir directly AT the broken file was tried first and passed on macOS/Linux but silently succeeded on Windows, where recursive mkdir() against an already-existing path does not appear to verify it is actually a directory. + const brokenFile = join(homeDir, "not-a-directory"); + const brokenSocketDir = join(brokenFile, "sub"); await mkdir(homeDir, { recursive: true }); - await writeFile(brokenSocketDir, "not a directory"); + await writeFile(brokenFile, "not a directory"); const proc = makeAliasProcess(); await expect( proc.start({ diff --git a/src/alias-pool.ts b/src/alias-pool.ts index 83de0c5..6ac0de3 100644 --- a/src/alias-pool.ts +++ b/src/alias-pool.ts @@ -19,11 +19,18 @@ export interface AliasMessage extends InboundMessage { const sinkLog = (): void => undefined; /** - * alias-worker.ts's own built location, computed from this file's `import.meta.url` rather than ForkedAliasProcess's own: this file is itself a dedicated tsdown entry (see tsdown.config.ts), so its emitted location is stable and predictable, and alias-worker.ts (also a dedicated entry) is emitted at the same relative position to it as the two source files hold to each other in src/. The target extension mirrors this file's own real extension at runtime (`.mjs` or `.cjs`, tsdown's `fixedExtension: true`) rather than a hardcoded `.js`: tsdown builds matching-format sibling entries for every declared output, so alias-pool.mjs's own sibling is always alias-worker.mjs, never a plain `.js` neither format ever actually emits. + * Mirrors this module's own real build extension (tsdown's `fixedExtension: true` always emits `.mjs` or `.cjs`, never a plain `.js`) so a computed sibling path names a file tsdown actually produces. Exported as a pure function, independent of `import.meta.url`, so both branches are directly unit-testable: in this repo's own dev/test tree `import.meta.url` always ends in `.ts`, so a test exercising this module's own `import.meta.url` can only ever observe the `.mjs` branch. + */ +export function workerExtensionFor(moduleUrl: string): ".mjs" | ".cjs" { + return moduleUrl.endsWith(".cjs") ? ".cjs" : ".mjs"; +} + +/** + * alias-worker.ts's own built location, computed from this file's `import.meta.url` rather than ForkedAliasProcess's own: this file is itself a dedicated tsdown entry (see tsdown.config.ts), so its emitted location is stable and predictable, and alias-worker.ts (also a dedicated entry) is emitted at the same relative position to it as the two source files hold to each other in src/. */ const DEFAULT_WORKER_PATH = fileURLToPath( new URL( - `./adapters/node/alias-worker${import.meta.url.endsWith(".cjs") ? ".cjs" : ".mjs"}`, + `./adapters/node/alias-worker${workerExtensionFor(import.meta.url)}`, import.meta.url, ), ); diff --git a/src/alias-pool.unit.test.ts b/src/alias-pool.unit.test.ts index 0cd7008..0cf1260 100644 --- a/src/alias-pool.unit.test.ts +++ b/src/alias-pool.unit.test.ts @@ -1,9 +1,23 @@ import { describe, expect, test, vi } from "vitest"; import { EventEmitter } from "node:events"; -import { AliasPool } from "./alias-pool.js"; +import { AliasPool, workerExtensionFor } from "./alias-pool.js"; import type { AliasProcess } from "./ports/alias-process.js"; +describe("workerExtensionFor", () => { + test("returns .cjs for a module URL ending in .cjs", () => { + expect(workerExtensionFor("file:///dist/alias-pool.cjs")).toBe(".cjs"); + }); + + test("returns .mjs for a module URL ending in .mjs", () => { + expect(workerExtensionFor("file:///dist/alias-pool.mjs")).toBe(".mjs"); + }); + + test("returns .mjs for any other extension (e.g. the .ts source in dev/test)", () => { + expect(workerExtensionFor("file:///src/alias-pool.ts")).toBe(".mjs"); + }); +}); + /** A fake AliasProcess whose start()/stop() are externally controllable, so tests can assert ordering and concurrency without a real child process. */ class FakeAliasProcess implements AliasProcess { readonly events = new EventEmitter(); From 49417bfd28354f2abeb95da42d65a58cf8503307 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:57:42 +0100 Subject: [PATCH 8/8] fix(alias): trigger the worker-start-failure test via homeDir, not socketDir CcPeer.start() skips its socketDir mkdir entirely on Windows (a named pipe has no filesystem directory of its own), so corrupting socketDir can never force a failure there regardless of nesting, confirmed by this test still failing on both Windows CI runners after the first nesting fix. keys.writeForSocket() and registry.write() both mkdir into sessionsDir(homeDir) unconditionally on every platform, so corrupting homeDir instead reaches a real, unconditional mkdir() everywhere. --- .../node/forked-alias-process.integration.test.ts | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/adapters/node/forked-alias-process.integration.test.ts b/src/adapters/node/forked-alias-process.integration.test.ts index 6f47366..ff5634c 100644 --- a/src/adapters/node/forked-alias-process.integration.test.ts +++ b/src/adapters/node/forked-alias-process.integration.test.ts @@ -169,18 +169,18 @@ describe("ForkedAliasProcess against the real alias-worker source", () => { test( "rejects with AliasStartError when the worker fails to start", async () => { - const homeDir = await tempHome(); - // A socketDir nested one level inside a plain regular file makes the worker's own CcPeer.start() throw on mkdir(dirname(socketPath)): creating a new directory entry inside a file (not merely re-stating an already-existing path) has no valid filesystem resolution on any platform. Nesting one level deeper than the broken file itself is load-bearing: pointing socketDir directly AT the broken file was tried first and passed on macOS/Linux but silently succeeded on Windows, where recursive mkdir() against an already-existing path does not appear to verify it is actually a directory. - const brokenFile = join(homeDir, "not-a-directory"); - const brokenSocketDir = join(brokenFile, "sub"); - await mkdir(homeDir, { recursive: true }); + const root = await tempHome(); + // homeDir is the trigger, not socketDir: CcPeer.start() skips its socketDir mkdir entirely on Windows (a named pipe has no filesystem directory of its own — see cc-peer.ts's own isWindows() guard), so corrupting socketDir can never fail there regardless of nesting (confirmed: this test failed on Windows CI with exactly that approach). keys.writeForSocket() and registry.write() both mkdir into sessionsDir(homeDir) unconditionally on every platform, so corrupting homeDir instead reaches a real, unconditional mkdir() everywhere. Nesting one level inside the broken file (not pointing homeDir directly at it) is still load-bearing for the same reason established earlier: creating a genuinely new directory entry inside a file has no valid resolution on any platform, whereas recursive mkdir() against an already-existing path does not appear to verify it is actually a directory on Windows. + const brokenFile = join(root, "not-a-directory"); + const brokenHomeDir = join(brokenFile, "home"); + await mkdir(root, { recursive: true }); await writeFile(brokenFile, "not a directory"); const proc = makeAliasProcess(); await expect( proc.start({ name: "carol-fail-test", - homeDir, - socketDir: brokenSocketDir, + homeDir: brokenHomeDir, + socketDir: join(brokenHomeDir, "socks"), }), ).rejects.toThrow(AliasStartError); },