From df441bac1f328b359d053b837758420d30b2b4ee Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 06:11:35 +0000 Subject: [PATCH 01/14] fix(cli): honor --state-dir across node agent and node down/status Every node agent subcommand (list, spawn, new, release, set-model) and node tail now accept --state-dir / --broker-url / --api-key, and a flag-free RELAY_BROKER_URL selects the broker, so brokers started with node up --state-dir (fleet nodes) are manageable from any directory. - An explicit --state-dir reads only that broker's connection.json; ambient RELAY_BROKER_URL / RELAY_BROKER_API_KEY no longer override it. - --state-dir also accepts a fleet node directory whose broker state lives in state/ (node agent, node status, node down). - Missing-connection errors name the searched path and whether it was the project default or --state-dir. - Broker identity records now live in the broker state dir, so node down --force --state-dir verifies brokers started from another cwd; legacy project-dir records are still read. node status and node down name a missing identity record and the migration path. Fixes #1446, #1822, #1820, #1575, #1192. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- CHANGELOG.md | 10 +- packages/cli/README.md | 2 + packages/cli/src/cli/commands/core.test.ts | 91 +++++- .../cli/src/cli/commands/local-agent.test.ts | 134 +++++++++ packages/cli/src/cli/commands/local-agent.ts | 237 +++++++++------- packages/cli/src/cli/lib/attach-drive.test.ts | 2 +- packages/cli/src/cli/lib/attach-native.ts | 3 +- .../src/cli/lib/attach-passthrough.test.ts | 2 +- packages/cli/src/cli/lib/attach-view.test.ts | 2 +- packages/cli/src/cli/lib/attach-view.ts | 53 +--- packages/cli/src/cli/lib/attach.ts | 6 +- .../cli/src/cli/lib/broker-connection.test.ts | 85 +++++- packages/cli/src/cli/lib/broker-connection.ts | 90 +++++- packages/cli/src/cli/lib/broker-lifecycle.ts | 47 +++- .../cli/lib/broker-process-identity.test.ts | 3 +- .../src/cli/lib/broker-process-identity.ts | 138 +++++++--- .../1446-state-dir-broker-selection/case.json | 21 ++ .../1446-state-dir-broker-selection/run.mjs | 258 ++++++++++++++++++ 18 files changed, 969 insertions(+), 215 deletions(-) create mode 100644 tests/relayflows/cases/1446-state-dir-broker-selection/case.json create mode 100755 tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e9757323a..70afb5c795 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,10 +5,18 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased - Patch] +## [Unreleased - Minor] + +### Added + +- `node agent list|spawn|new|release|set-model` and `node tail` accept `--state-dir`, `--broker-url`, and `--api-key` (or `RELAY_BROKER_URL`), so brokers started with `node up --state-dir` — such as fleet nodes — can be managed from any directory. ### Fixed +- `--state-dir` on `node agent` commands, `node status`, and `node down` also accepts a fleet node directory whose broker state lives in `state/`, and an explicit `--state-dir` is no longer overridden by ambient `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`. +- "No broker connection" errors name the path searched and whether it was the project default or `--state-dir`, instead of implying a running broker is down. +- `node down --force --state-dir ` can verify and stop brokers started from another working directory: identity records now live in the broker state dir (records written by earlier releases are still read). `node status` and `node down` name a missing identity record and how to create one. + - Broker: Muse workers now reuse the host Muse login (`RELAY_MUSE_SHARED_AUTH_PATH`, the host's `MUSE_AUTH_PATH`, `$XDG_CONFIG_HOME/muse/auth.json`, or `$HOME/.config/muse/auth.json`) instead of prompting for provider auth under every new agent name; Relay MCP settings stay per-worker. Multi-tenant hosts should set `RELAY_MUSE_ISOLATED_AUTH=1` before upgrading to keep a separate login per worker. ## [13.0.0] - 2026-09-30 diff --git a/packages/cli/README.md b/packages/cli/README.md index e6ef31c1e6..e592462629 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -45,6 +45,8 @@ agent-relay node agent attach --mode view agent-relay node agent release ``` +Every `node agent` subcommand (and `node tail`) accepts `--state-dir `, `--broker-url `, and `--api-key ` to target a broker other than the current project's — for example a fleet node started with `node up --state-dir`. `--state-dir` may name the broker state dir or a fleet node directory whose broker state lives in `state/`. An explicit `--state-dir` reads only that broker's `connection.json`, ignoring `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`; without flags, a nonblank `RELAY_BROKER_URL` selects the broker, otherwise the enclosing project's broker is used. + `node agent spawn` and `node agent new` accept `--runtime auto|native|pty`. `auto` is the default and keeps experimental dual-runtime adapters on PTY. Claude Code, Codex, and OpenCode support explicit native or PTY selection; Pi and Deep Agents are experimental native-only harnesses and require `--runtime native`. For AI SDK native harnesses, attach renders structured activity, text, tools, approvals, files, usage, and lifecycle events. Add `--json` for NDJSON, `--reasoning` for reasoning events, or `--diagnostics` for sidecar diagnostics. Native harness `drive` is line-oriented and acknowledged; native harness `passthrough` is unsupported because no terminal stream exists. PTY attach behavior is unchanged. diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index 082726db6b..a2883e45de 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -1136,7 +1136,9 @@ describe('registerCoreCommands', () => { const paths = { ...deps.getProjectPaths(), dataDir: stateDir }; await persistBrokerIdentity(paths, 222, 'custom-node', deps); const filename = brokerIdentityPath(paths, deps, 'custom-node'); - expect(filename).toMatch(/^\/tmp\/project\/\.agentworkforce\/relay\/broker-identity-/); + // Records live in the broker state dir, so `down --state-dir` finds them from any cwd. + expect(nodePath.dirname(filename)).toBe(stateDir); + expect(nodePath.basename(filename)).toMatch(/^broker-identity-/); expect(deps.fs.existsSync(filename)).toBe(true); const stateArg = stateDir === nodePath.resolve('nested/candidate-state') ? 'nested/candidate-state' : stateDir; @@ -1616,6 +1618,77 @@ describe('registerCoreCommands', () => { expect(deps.killProcess).toHaveBeenCalledWith(222, 'SIGTERM'); }); + it.each([ + ['the state dir itself', '/srv/dogpatch-node/state'], + ['the fleet node directory (relay#1575)', '/srv/dogpatch-node'], + ])( + 'down --force verifies a broker started from another cwd via %s (relay#1820)', + async (_label, stateArg) => { + const stateDir = '/srv/dogpatch-node/state'; + let running = true; + const fs = createFsMock({ [`${stateDir}/connection.json`]: connectionFile(222) }); + const { program, deps } = createHarness({ + fs, + execCommand: identityCommand(222, `${stateDir}/broker-dogpatch-mini.lock`), + killImpl: vi.fn((_pid, signal) => { + if (signal === 'SIGTERM') running = false; + if (signal === 0 && !running) throw new Error('not running'); + }), + }); + // `node up` ran under launchd with a different working directory, so its + // project root differs from the operator's. + const launchPaths = { ...deps.getProjectPaths(), projectRoot: '/Users/op', dataDir: stateDir }; + expect(await persistBrokerIdentity(launchPaths, 222, 'dogpatch-mini', deps)).toBeDefined(); + expect(nodePath.dirname(brokerIdentityPath(launchPaths, deps, 'dogpatch-mini'))).toBe(stateDir); + + await runCommand(program, ['down', '--force', '--state-dir', stateArg]); + + expect(deps.killProcess).toHaveBeenCalledWith(222, 'SIGTERM'); + expect(deps.log).toHaveBeenCalledWith('Stopped'); + expect(fs.existsSync(brokerIdentityPath(launchPaths, deps, 'dogpatch-mini'))).toBe(false); + } + ); + + it('down still verifies records written under the project by earlier releases', async () => { + const stateDir = '/srv/legacy-node/state'; + let running = true; + const fs = createFsMock({ [`${stateDir}/connection.json`]: connectionFile(222) }); + const { program, deps } = createHarness({ + fs, + execCommand: identityCommand(222, `${stateDir}/broker-legacy.lock`), + killImpl: vi.fn((_pid, signal) => { + if (signal === 'SIGTERM') running = false; + if (signal === 0 && !running) throw new Error('not running'); + }), + }); + const paths = { ...deps.getProjectPaths(), dataDir: stateDir }; + await persistBrokerIdentity(paths, 222, 'legacy', deps); + // Move the record to where earlier releases wrote it. + const current = brokerIdentityPath(paths, deps, 'legacy'); + const legacy = nodePath.join(paths.projectRoot, '.agentworkforce', 'relay', nodePath.basename(current)); + fs.writeFileSync(legacy, fs.readFileSync(current, 'utf-8')); + fs.unlinkSync(current); + + await runCommand(program, ['down', '--force', '--state-dir', stateDir]); + + expect(deps.killProcess).toHaveBeenCalledWith(222, 'SIGTERM'); + expect(fs.existsSync(legacy)).toBe(false); + }); + + it('down names the missing identity record and the migration path', async () => { + const stateDir = '/srv/direct-init/state'; + const fs = createFsMock({ [`${stateDir}/connection.json`]: connectionFile(3030) }); + const { program, deps } = createHarness({ fs }); + expect(await runCommand(program, ['down', '--force', '--state-dir', stateDir])).toBe(1); + expect(vi.mocked(deps.killProcess).mock.calls.filter(([, signal]) => signal !== 0)).toEqual([]); + expect(deps.error).toHaveBeenCalledWith( + expect.stringContaining(`No identity record for the running broker (pid: 3030) in ${stateDir}`) + ); + expect(deps.error).toHaveBeenCalledWith( + expect.stringContaining(`agent-relay node up --state-dir ${stateDir}`) + ); + }); + it('different names retain independent records and forced cleanup verifies each', async () => { const running = new Set([222, 333]); const first = identityCommand(222, '/tmp/project/.agentworkforce/relay/broker-first.lock'); @@ -2223,6 +2296,22 @@ describe('registerCoreCommands', () => { expect(deps.log).toHaveBeenCalledWith('Not running'); }); + it('status --state-dir accepts a fleet node directory and flags a missing identity record', async () => { + const stateDir = '/srv/sf-mini-node/state'; + const fs = createFsMock({ [`${stateDir}/connection.json`]: connectionFile(4242) }); + sdkStatusClient.getStatus.mockResolvedValueOnce({ agent_count: 1, pending_delivery_count: 0 }); + sdkStatusClient.getSession.mockResolvedValueOnce({ workspace_key: 'rk_live_teststatus123' }); + const { program, deps } = createHarness({ fs }); + + expect(await runCommand(program, ['status', '--state-dir', '/srv/sf-mini-node'])).toBeUndefined(); + + expect(deps.log).toHaveBeenCalledWith('Status: RUNNING'); + expect(deps.log).toHaveBeenCalledWith('PID: 4242'); + expect(deps.warn).toHaveBeenCalledWith( + expect.stringContaining(`Restart it once with \`agent-relay node up --state-dir ${stateDir}\``) + ); + }); + it('status checks broker status and prints metrics', async () => { const connectionPath = '/tmp/project/.agentworkforce/relay/connection.json'; const fs = createFsMock({ diff --git a/packages/cli/src/cli/commands/local-agent.test.ts b/packages/cli/src/cli/commands/local-agent.test.ts index ba33952135..6a497b6b5a 100644 --- a/packages/cli/src/cli/commands/local-agent.test.ts +++ b/packages/cli/src/cli/commands/local-agent.test.ts @@ -1201,3 +1201,137 @@ describe('local agent subtree', () => { ); }); }); + +describe('explicit broker selection on every node agent subcommand (relay#1446, relay#1822)', () => { + const stateDir = '/Users/op/.agentworkforce/relay/finn-mini-node/state'; + + function selectionHarness(env: NodeJS.ProcessEnv = {}) { + const connect = vi.fn(async () => { + throw new Error('project broker must not be used'); + }); + const local = { + listAgents: vi.fn(async () => [{ name: 'worker' }]), + spawnPty: vi.fn(async () => undefined), + release: vi.fn(async () => undefined), + setModel: vi.fn(async () => ({ name: 'worker', model: 'opus', success: true })), + disconnect: vi.fn(), + }; + const connectLocal = vi.fn(async () => local as never); + const h = harness({ connect, connectLocal, env }); + return { ...h, connect, connectLocal, local }; + } + + it.each([ + [['list'], 'listAgents'], + [['spawn', 'codex', '--name', 'worker'], 'spawnPty'], + [['release', 'worker'], 'release'], + [['set-model', 'worker', 'opus'], 'setModel'], + ] as const)('%j --state-dir targets that broker, not the project default', async (args, method) => { + const { program, connect, connectLocal, local, exit } = selectionHarness(); + await program.parseAsync(['local', 'agent', ...args, '--state-dir', stateDir], { from: 'user' }); + expect(connect).not.toHaveBeenCalled(); + expect(connectLocal).toHaveBeenCalledWith('/tmp/project', { + brokerUrl: undefined, + apiKey: undefined, + stateDir, + }); + expect(local[method]).toHaveBeenCalled(); + expect(local.disconnect).toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + }); + + it('release accepts --broker-url / --api-key outside any broker project', async () => { + const { program, connect, connectLocal, local, log } = selectionHarness(); + await program.parseAsync( + ['local', 'agent', 'release', 'worker', '--broker-url', 'http://127.0.0.1:4100', '--api-key', 'k'], + { from: 'user' } + ); + expect(connect).not.toHaveBeenCalled(); + expect(connectLocal).toHaveBeenCalledWith('/tmp/project', { + brokerUrl: 'http://127.0.0.1:4100', + apiKey: 'k', + stateDir: undefined, + }); + expect(local.release).toHaveBeenCalledWith('worker'); + expect(log).toHaveBeenCalledWith('Released worker.'); + }); + + it('RELAY_BROKER_URL selects the broker for flag-free release', async () => { + const { program, connect, connectLocal, local } = selectionHarness({ + RELAY_BROKER_URL: 'http://127.0.0.1:4100', + }); + await program.parseAsync(['local', 'agent', 'release', 'worker'], { from: 'user' }); + expect(connect).not.toHaveBeenCalled(); + expect(connectLocal).toHaveBeenCalled(); + expect(local.release).toHaveBeenCalledWith('worker'); + }); + + it('new spawns and attaches on the same explicitly selected broker', async () => { + const { program, connectLocal, local, attach } = selectionHarness(); + await program.parseAsync( + ['local', 'agent', 'new', 'codex', '--name', 'worker', '--state-dir', stateDir], + { + from: 'user', + } + ); + expect(connectLocal).toHaveBeenCalled(); + expect(local.spawnPty).toHaveBeenCalled(); + expect(attach).toHaveBeenCalledWith( + 'worker', + 'drive', + expect.objectContaining({ stateDir, brokerUrl: undefined, apiKey: undefined }) + ); + }); + + it('flag-free commands keep using the enclosing project broker', async () => { + const connectLocal = vi.fn(); + const { program, client } = harness({ connectLocal, env: {} }); + await program.parseAsync(['local', 'agent', 'release', 'lead'], { from: 'user' }); + expect(connectLocal).not.toHaveBeenCalled(); + expect(client.release).toHaveBeenCalledWith('lead'); + }); + + it('a missing --state-dir connection names the searched path and ignores ambient env', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'relay-state-dir-miss-')); + const { program, error, exit } = harness({ + env: { RELAY_BROKER_URL: 'http://127.0.0.1:9', RELAY_BROKER_API_KEY: 'other-broker' }, + }); + try { + await program.parseAsync(['local', 'agent', 'list', '--state-dir', root], { from: 'user' }); + expect(exit).toHaveBeenCalledWith(1); + expect(error).toHaveBeenCalledWith( + `Error: no broker connection at ${path.join(root, 'connection.json')} (from --state-dir; also checked ${path.join(root, 'state', 'connection.json')}). Pass the same --state-dir the broker was started with.` + ); + } finally { + fs.rmSync(root, { recursive: true, force: true }); + } + }); + + it('flag-free "no broker" errors say the default path was searched and how to target a --state-dir broker', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'relay-no-broker-')); + fs.mkdirSync(path.join(root, '.git')); + harnessConnectMock.mockImplementationOnce(() => { + throw new Error( + `No running broker found (${root}/.agentworkforce/relay/connection.json does not exist).` + ); + }); + const error = vi.fn(); + const exit = vi.fn(); + const program = new Command(); + registerLocalAgentCommands(program.command('local'), { + cwd: () => root, + env: {}, + error, + exit: exit as never, + }); + try { + await program.parseAsync(['local', 'agent', 'list'], { from: 'user' }); + expect(exit).toHaveBeenCalledWith(1); + expect(error).toHaveBeenCalledWith( + expect.stringContaining('If the broker was started with --state-dir, pass the same --state-dir here') + ); + } finally { + fs.rmSync(root, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/cli/src/cli/commands/local-agent.ts b/packages/cli/src/cli/commands/local-agent.ts index fe6cf2506f..6d667a9141 100644 --- a/packages/cli/src/cli/commands/local-agent.ts +++ b/packages/cli/src/cli/commands/local-agent.ts @@ -19,6 +19,7 @@ import { attachView } from '../lib/attach-view.js'; import { createBackpressureAwareWriter } from '../lib/attach.js'; import { defaultStateDir, + describeMissingBrokerConnection, readConnectionFileFromDisk, resolveBrokerConnection, type BrokerConnectionOptions, @@ -210,7 +211,20 @@ export interface LocalAgentDependencies { function withDefaults(overrides: Partial = {}): LocalAgentDependencies { const deps = { - connect: async (cwd: string) => connectProjectBrokerClient(findProjectRoot(cwd)), + connect: async (cwd: string) => { + try { + return connectProjectBrokerClient(findProjectRoot(cwd)); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + if (!message.startsWith('No running broker found')) throw err; + // The project file is only the default; a broker started with + // --state-dir is alive elsewhere. Say so instead of implying it is down. + throw new Error( + `${message}\nThat is the project default location. If the broker was started with --state-dir, ` + + 'pass the same --state-dir here (or --broker-url / RELAY_BROKER_URL).' + ); + } + }, cwd: () => process.cwd(), readConnectionFile: readConnectionFileFromDisk, getDefaultStateDir: defaultStateDir, @@ -235,10 +249,7 @@ function withDefaults(overrides: Partial = {}): LocalAge env: deps.env, }); if (!connection) { - throw new Error( - 'Error: could not locate broker connection. Pass --broker-url, set RELAY_BROKER_URL, ' + - 'or run from a directory containing .agentworkforce/relay/connection.json.' - ); + throw new Error(describeMissingBrokerConnection(options, deps)); } return createBrokerClient(connection, deps.fetch); }; @@ -489,13 +500,33 @@ export function formatPrettyAgentStatusList(agents: AgentDeliveryStatus[], now: ]); } +/** + * Whether the caller chose a broker explicitly (`--broker-url` / `--api-key` / + * `--state-dir`, or `RELAY_BROKER_URL`). Without one, commands keep using the + * enclosing project's broker. + */ +function hasExplicitBrokerSelection( + deps: Pick, + options: LocalAgentMessageBrokerOptions +): boolean { + return Boolean( + options.brokerUrl?.trim() || + options.apiKey?.trim() || + options.stateDir?.trim() || + deps.env.RELAY_BROKER_URL?.trim() + ); +} + async function run( deps: LocalAgentDependencies, - fn: (client: HarnessDriverClient) => Promise + fn: (client: HarnessDriverClient) => Promise, + brokerOptions: LocalAgentMessageBrokerOptions = {} ): Promise { let client: HarnessDriverClient | undefined; try { - client = await deps.connect(deps.cwd()); + client = hasExplicitBrokerSelection(deps, brokerOptions) + ? await deps.connectLocal(deps.cwd(), brokerOptions) + : await deps.connect(deps.cwd()); await fn(client); } catch (err) { deps.error(err instanceof Error ? err.message : String(err)); @@ -781,13 +812,16 @@ export function registerLocalAgentCommands( const deps = withDefaults(overrides); const agent = group.command('agent').description('Inspect and manage broker-spawned agents'); - agent - .command('list') - .description('List agents running on the local broker') - .option('--pretty', 'Show a compact human-readable list') - .option('--status', 'Include each agent inbound delivery mode and pending-queue contents (relay#1387)') - .action(async (opts: { pretty?: boolean; status?: boolean }) => { - await run(deps, async (client) => { + addBrokerOptions( + agent + .command('list') + .description('List agents running on the local broker') + .option('--pretty', 'Show a compact human-readable list') + .option('--status', 'Include each agent inbound delivery mode and pending-queue contents (relay#1387)') + ).action(async (opts: { pretty?: boolean; status?: boolean } & Record) => { + await run( + deps, + async (client) => { const agents = await client.listAgents(); if (opts.status) { const withStatus = await withDeliveryStatus(client, agents); @@ -799,11 +833,12 @@ export function registerLocalAgentCommands( return; } deps.log(opts.pretty ? formatPrettyAgentList(agents, deps.now()) : JSON.stringify(agents, null, 2)); - }); - }); + }, + brokerOptionsFromOpts(opts) + ); + }); - agent - .command('spawn') + addBrokerOptions(agent.command('spawn')) .description('Spawn an agent with the given provider CLI') .argument( '', @@ -829,32 +864,35 @@ export function registerLocalAgentCommands( }) ) return; - await run(deps, async (client) => { - const baseName = (opts.name as string | undefined) ?? provider; - const resolved = resolveAutoSpawn( - provider, - baseName, - opts.task as string | undefined, - opts.model as string | undefined - ); - await spawnAgentWithClient(client, { - name: resolved.name, - cli: provider, - channels: (opts.channels as string[] | undefined) ?? ['general'], - task: resolved.task, - model: resolved.model, - cwd: (opts.cwd as string | undefined) ?? deps.cwd(), - spawnMode, - exitAfterTask: opts.exitAfterTask as boolean | undefined, - runtime: runtime.requested, - }); - const autoNote = opts.model === 'auto' ? ' (auto-routed)' : ''; - deps.log(`Spawned ${resolved.name} (${provider}, ${runtime.selected})${autoNote}.`); - }); + await run( + deps, + async (client) => { + const baseName = (opts.name as string | undefined) ?? provider; + const resolved = resolveAutoSpawn( + provider, + baseName, + opts.task as string | undefined, + opts.model as string | undefined + ); + await spawnAgentWithClient(client, { + name: resolved.name, + cli: provider, + channels: (opts.channels as string[] | undefined) ?? ['general'], + task: resolved.task, + model: resolved.model, + cwd: (opts.cwd as string | undefined) ?? deps.cwd(), + spawnMode, + exitAfterTask: opts.exitAfterTask as boolean | undefined, + runtime: runtime.requested, + }); + const autoNote = opts.model === 'auto' ? ' (auto-routed)' : ''; + deps.log(`Spawned ${resolved.name} (${provider}, ${runtime.selected})${autoNote}.`); + }, + brokerOptionsFromOpts(opts) + ); }); - agent - .command('new') + addBrokerOptions(agent.command('new')) .description('Spawn an agent and attach to it') .argument( '', @@ -888,6 +926,7 @@ export function registerLocalAgentCommands( }) ) return; + const brokerOptions = brokerOptionsFromOpts(options); const baseName = (options.name as string | undefined) ?? provider; const resolved = resolveAutoSpawn( provider, @@ -895,52 +934,61 @@ export function registerLocalAgentCommands( options.task as string | undefined, options.model as string | undefined ); - await run(deps, async (client) => { - await spawnAgentWithClient(client, { - name: resolved.name, - cli: provider, - channels: (options.channels as string[] | undefined) ?? ['general'], - task: resolved.task, - model: resolved.model, - cwd: (options.cwd as string | undefined) ?? deps.cwd(), - spawnMode, - exitAfterTask: options.exitAfterTask as boolean | undefined, - runtime: runtime.requested, - }); - const autoNote = options.model === 'auto' ? ' (auto-routed)' : ''; - deps.log( - `Spawned ${resolved.name} (${provider}, ${runtime.selected}). Attaching (${mode})${autoNote}…` - ); - }); - // `new` spawns and attaches on the same default local broker — broker - // override flags belong on the standalone `attach` command. - const code = await deps.attach(resolved.name, mode as AttachMode, {}); + await run( + deps, + async (client) => { + await spawnAgentWithClient(client, { + name: resolved.name, + cli: provider, + channels: (options.channels as string[] | undefined) ?? ['general'], + task: resolved.task, + model: resolved.model, + cwd: (options.cwd as string | undefined) ?? deps.cwd(), + spawnMode, + exitAfterTask: options.exitAfterTask as boolean | undefined, + runtime: runtime.requested, + }); + const autoNote = options.model === 'auto' ? ' (auto-routed)' : ''; + deps.log( + `Spawned ${resolved.name} (${provider}, ${runtime.selected}). Attaching (${mode})${autoNote}…` + ); + }, + brokerOptions + ); + // Attach to the same broker the agent was just spawned on. + const code = await deps.attach(resolved.name, mode as AttachMode, brokerOptions); if (code !== 0) { deps.exit(code); } }); - agent - .command('release') + addBrokerOptions(agent.command('release')) .description('Release an agent (graceful stop)') .argument('', 'Agent name') - .action(async (name: string) => { - await run(deps, async (client) => { - await client.release(name); - deps.log(`Released ${name}.`); - }); + .action(async (name: string, opts: Record) => { + await run( + deps, + async (client) => { + await client.release(name); + deps.log(`Released ${name}.`); + }, + brokerOptionsFromOpts(opts) + ); }); - agent - .command('set-model') + addBrokerOptions(agent.command('set-model')) .description("Switch a running agent's model (sends `/model` to its TUI; best-effort)") .argument('', 'Agent name') .argument('', 'Model identifier to switch to') - .action(async (name: string, model: string) => { - await run(deps, async (client) => { - await client.setModel(name, model); - deps.log(`Sent \`/model ${model}\` to ${name} (best-effort — the agent's TUI applies it).`); - }); + .action(async (name: string, model: string, opts: Record) => { + await run( + deps, + async (client) => { + await client.setModel(name, model); + deps.log(`Sent \`/model ${model}\` to ${name} (best-effort — the agent's TUI applies it).`); + }, + brokerOptionsFromOpts(opts) + ); }); agent @@ -1145,26 +1193,29 @@ export function registerLocalAgentCommands( if (result !== undefined) deps.log(JSON.stringify({ name, ...result }, null, 2)); }); - group - .command('tail') + addBrokerOptions(group.command('tail')) .description('Stream broker events (optionally filtered to one agent)') .option('--agent ', "Filter to a single agent's output stream") - .action(async (options: { agent?: string }) => { - await run(deps, async (client) => { - if (options.agent) { - for await (const chunk of client.subscribeWorkerStream(options.agent)) { - process.stdout.write(chunk); + .action(async (options: { agent?: string } & Record) => { + await run( + deps, + async (client) => { + if (options.agent) { + for await (const chunk of client.subscribeWorkerStream(options.agent)) { + process.stdout.write(chunk); + } + return; } - return; - } - client.connectEvents(); - await new Promise((resolve) => { - client.onEvent((event) => { - deps.log(JSON.stringify(event)); + client.connectEvents(); + await new Promise((resolve) => { + client.onEvent((event) => { + deps.log(JSON.stringify(event)); + }); + // Ctrl+C ends a streaming tail cleanly (exit 0, no error output). + process.once('SIGINT', () => resolve()); }); - // Ctrl+C ends a streaming tail cleanly (exit 0, no error output). - process.once('SIGINT', () => resolve()); - }); - }); + }, + brokerOptionsFromOpts(options) + ); }); } diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts index 1b548cac5a..668c211dfc 100644 --- a/packages/cli/src/cli/lib/attach-drive.test.ts +++ b/packages/cli/src/cli/lib/attach-drive.test.ts @@ -1442,7 +1442,7 @@ describe('runDriveSession', () => { deps.readConnectionFile = vi.fn(() => null); const code = await runDriveSession('Alice', {}, deps); expect(code).toBe(1); - expect(errors[0]?.[0]).toMatch(/could not locate broker connection/); + expect(errors[0]?.[0]).toMatch(/no broker connection at .*connection\.json \(project default\)/); }); // ---- API-key header propagation ---- diff --git a/packages/cli/src/cli/lib/attach-native.ts b/packages/cli/src/cli/lib/attach-native.ts index 7a00d2e47b..736db5a9ae 100644 --- a/packages/cli/src/cli/lib/attach-native.ts +++ b/packages/cli/src/cli/lib/attach-native.ts @@ -12,6 +12,7 @@ import { import { defaultStateDir, readConnectionFileFromDisk, + describeMissingBrokerConnection, resolveBrokerConnection, type BrokerConnection, } from './broker-connection.js'; @@ -134,7 +135,7 @@ export async function attachNative( }); if (!connection) { (overrides.output?.stderr ?? ((text: string) => process.stderr.write(text)))( - 'Error: could not locate broker connection.\n' + `${describeMissingBrokerConnection(options, { getDefaultStateDir: defaultStateDir })}\n` ); return 1; } diff --git a/packages/cli/src/cli/lib/attach-passthrough.test.ts b/packages/cli/src/cli/lib/attach-passthrough.test.ts index a9528f0741..b602da0b1b 100644 --- a/packages/cli/src/cli/lib/attach-passthrough.test.ts +++ b/packages/cli/src/cli/lib/attach-passthrough.test.ts @@ -839,7 +839,7 @@ describe('runPassthroughSession', () => { deps.readConnectionFile = vi.fn(() => null); const code = await runPassthroughSession('Alice', {}, deps); expect(code).toBe(1); - expect(errors[0]?.[0]).toMatch(/could not locate broker connection/); + expect(errors[0]?.[0]).toMatch(/no broker connection at .*connection\.json \(project default\)/); }); // ---- API-key header propagation ---- diff --git a/packages/cli/src/cli/lib/attach-view.test.ts b/packages/cli/src/cli/lib/attach-view.test.ts index 557f9db096..92d73327e0 100644 --- a/packages/cli/src/cli/lib/attach-view.test.ts +++ b/packages/cli/src/cli/lib/attach-view.test.ts @@ -636,7 +636,7 @@ describe('runViewSession', () => { const { deps, errors } = createHarness({ env: {}, connectionFile: null }); const code = await runViewSession('Alice', {}, deps); expect(code).toBe(1); - expect(errors[0]?.[0]).toMatch(/could not locate broker connection/); + expect(errors[0]?.[0]).toMatch(/no broker connection at .*connection\.json \(project default\)/); }); it('omits the X-API-Key header when no api key is available', async () => { diff --git a/packages/cli/src/cli/lib/attach-view.ts b/packages/cli/src/cli/lib/attach-view.ts index c4752a610f..1fae790344 100644 --- a/packages/cli/src/cli/lib/attach-view.ts +++ b/packages/cli/src/cli/lib/attach-view.ts @@ -31,6 +31,7 @@ import { type AttachSnapshotConnection, type AttachSnapshotDeps, } from '../lib/attach.js'; +import { describeMissingBrokerConnection, resolveBrokerConnection } from '../lib/broker-connection.js'; import { resolveFleetHint } from '../lib/fleet-hint.js'; import { defaultExit, runSignalHandler } from '../lib/exit.js'; @@ -178,55 +179,16 @@ function isStringObject(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value); } -function readString(obj: unknown, key: string): string | undefined { - if (!isStringObject(obj)) return undefined; - const value = obj[key]; - if (typeof value !== 'string') return undefined; - const trimmed = value.trim(); - return trimmed === '' ? undefined : trimmed; -} - /** - * Resolve the broker connection to use for `view`, in priority order: - * - * 1. `--broker-url` / `--api-key` CLI flags - * 2. `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY` environment variables - * 3. `/connection.json` (default `.agentworkforce/relay/connection.json`) - * - * Matches the resolution order used by `agent-relay-broker dump-pty` so users - * don't have to learn two patterns. + * Resolve the broker connection to use for `view`. Delegates to the shared + * {@link resolveBrokerConnection} so `view` honours `--state-dir` exactly like + * every other local broker verb. */ export function resolveViewBrokerConnection( options: { brokerUrl?: string; apiKey?: string; stateDir?: string; requestTimeoutMs?: number }, deps: ViewDependencies ): ViewBrokerConnection | null { - const explicitUrl = trimOrUndefined(options.brokerUrl); - const envUrl = trimOrUndefined(deps.env.RELAY_BROKER_URL); - const stateDir = options.stateDir ? path.resolve(options.stateDir) : deps.getDefaultStateDir(); - const connectionFile = deps.readConnectionFile(stateDir); - const fileUrl = readString(connectionFile, 'url'); - - const resolveApiKey = (): string | undefined => { - const explicit = trimOrUndefined(options.apiKey); - if (explicit) return explicit; - const fromEnv = trimOrUndefined(deps.env.RELAY_BROKER_API_KEY); - if (fromEnv) return fromEnv; - return readString(connectionFile, 'api_key'); - }; - - const url = explicitUrl ?? envUrl ?? fileUrl; - if (!url) return null; - - return { - url: url.replace(/\/+$/, ''), - apiKey: resolveApiKey(), - ...(options.requestTimeoutMs === undefined ? {} : { requestTimeoutMs: options.requestTimeoutMs }), - }; -} - -function trimOrUndefined(value: string | undefined): string | undefined { - const trimmed = value?.trim(); - return trimmed ? trimmed : undefined; + return resolveBrokerConnection(options, deps); } /** Convert an `http(s)://host:port` base URL to the matching `ws(s)://…/ws`. */ @@ -413,10 +375,7 @@ export async function runViewSession( const connection = resolveViewBrokerConnection(options, deps); if (!connection) { - deps.error( - 'Error: could not locate broker connection. Pass --broker-url, set RELAY_BROKER_URL, ' + - 'or run from a directory containing .agentworkforce/relay/connection.json.' - ); + deps.error(describeMissingBrokerConnection(options, deps)); return 1; } diff --git a/packages/cli/src/cli/lib/attach.ts b/packages/cli/src/cli/lib/attach.ts index f05cfd52a4..ff43ed68a9 100644 --- a/packages/cli/src/cli/lib/attach.ts +++ b/packages/cli/src/cli/lib/attach.ts @@ -14,6 +14,7 @@ import type { InboundDeliveryMode } from '@agent-relay/harness-driver'; import { + describeMissingBrokerConnection, resolveBrokerConnection, type BrokerConnection, type BrokerConnectionDeps, @@ -358,10 +359,7 @@ export function prepareAttachTarget( } const connection = resolveBrokerConnection(options, deps); if (!connection) { - deps.error( - 'Error: could not locate broker connection. Pass --broker-url, set RELAY_BROKER_URL, ' + - 'or run from a directory containing .agentworkforce/relay/connection.json.' - ); + deps.error(describeMissingBrokerConnection(options, deps)); return null; } return { name, connection }; diff --git a/packages/cli/src/cli/lib/broker-connection.test.ts b/packages/cli/src/cli/lib/broker-connection.test.ts index 6587a640c0..79dbba0ddc 100644 --- a/packages/cli/src/cli/lib/broker-connection.test.ts +++ b/packages/cli/src/cli/lib/broker-connection.test.ts @@ -1,6 +1,17 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; + import { describe, expect, it, vi } from 'vitest'; -import { resolveBrokerConnection, toWsUrl, type BrokerConnectionDeps } from './broker-connection.js'; +import { + describeMissingBrokerConnection, + readConnectionFileFromDisk, + resolveBrokerConnection, + resolveConnectionStateDir, + toWsUrl, + type BrokerConnectionDeps, +} from './broker-connection.js'; function makeDeps(overrides: Partial = {}): BrokerConnectionDeps { return { @@ -90,6 +101,78 @@ describe('resolveBrokerConnection', () => { }); }); +describe('explicit --state-dir selection', () => { + const files: Record = { + '/srv/node/state': { url: 'http://node-host:4100/', api_key: 'node-key' }, + }; + const readConnectionFile = vi.fn((dir: string) => files[dir] ?? null); + + it('wins over ambient RELAY_BROKER_URL / RELAY_BROKER_API_KEY (relay#1822)', () => { + const deps = makeDeps({ + env: { RELAY_BROKER_URL: 'http://other:1', RELAY_BROKER_API_KEY: 'other-key' }, + readConnectionFile, + }); + expect(resolveBrokerConnection({ stateDir: '/srv/node/state' }, deps)).toEqual({ + url: 'http://node-host:4100', + apiKey: 'node-key', + }); + }); + + it('never falls back to env when the named state dir has no broker', () => { + const deps = makeDeps({ env: { RELAY_BROKER_URL: 'http://other:1' }, readConnectionFile }); + expect(resolveBrokerConnection({ stateDir: '/srv/missing' }, deps)).toBeNull(); + }); + + it('still lets --api-key override the connection file key', () => { + const deps = makeDeps({ readConnectionFile }); + expect(resolveBrokerConnection({ stateDir: '/srv/node/state', apiKey: 'flag-key' }, deps)?.apiKey).toBe( + 'flag-key' + ); + }); + + it('accepts a fleet node directory whose broker state lives in state/ (relay#1575)', () => { + const deps = makeDeps({ readConnectionFile }); + expect(resolveBrokerConnection({ stateDir: '/srv/node' }, deps)?.url).toBe('http://node-host:4100'); + }); + + it('reads the nested state/connection.json from disk and prefers an exact match', () => { + const nodeDir = fs.mkdtempSync(path.join(os.tmpdir(), 'relay-node-dir-')); + try { + fs.mkdirSync(path.join(nodeDir, 'state')); + fs.writeFileSync( + path.join(nodeDir, 'state', 'connection.json'), + JSON.stringify({ url: 'http://127.0.0.1:4555', api_key: 'nested' }) + ); + const deps = makeDeps({ readConnectionFile: readConnectionFileFromDisk }); + expect(resolveConnectionStateDir(nodeDir)).toBe(path.join(nodeDir, 'state')); + expect(resolveBrokerConnection({ stateDir: nodeDir }, deps)?.apiKey).toBe('nested'); + + fs.writeFileSync( + path.join(nodeDir, 'connection.json'), + JSON.stringify({ url: 'http://127.0.0.1:4666', api_key: 'exact' }) + ); + expect(resolveConnectionStateDir(nodeDir)).toBe(nodeDir); + expect(resolveBrokerConnection({ stateDir: nodeDir }, deps)?.apiKey).toBe('exact'); + } finally { + fs.rmSync(nodeDir, { recursive: true, force: true }); + } + }); +}); + +describe('describeMissingBrokerConnection', () => { + it('names the --state-dir paths that were searched', () => { + expect(describeMissingBrokerConnection({ stateDir: '/srv/node' }, makeDeps())).toBe( + 'Error: no broker connection at /srv/node/connection.json (from --state-dir; also checked /srv/node/state/connection.json). Pass the same --state-dir the broker was started with.' + ); + }); + + it('labels the project default and points at --state-dir', () => { + const message = describeMissingBrokerConnection({}, makeDeps()); + expect(message).toContain('/tmp/fake/.agentworkforce/relay/connection.json (project default)'); + expect(message).toContain('If the broker was started with --state-dir, pass the same --state-dir here'); + }); +}); + describe('toWsUrl', () => { it('rewrites http://host:port to ws://host:port/ws', () => { expect(toWsUrl('http://localhost:3889')).toBe('ws://localhost:3889/ws'); diff --git a/packages/cli/src/cli/lib/broker-connection.ts b/packages/cli/src/cli/lib/broker-connection.ts index 45cb66e573..1f20643e97 100644 --- a/packages/cli/src/cli/lib/broker-connection.ts +++ b/packages/cli/src/cli/lib/broker-connection.ts @@ -2,12 +2,15 @@ * Shared broker-connection discovery for the attach-style CLI verbs * (`view`, `drive`, `passthrough`). * - * Resolution order matches `agent-relay-broker dump-pty` so users don't - * have to learn two patterns: + * Resolution order: * * 1. `--broker-url` / `--api-key` CLI flags - * 2. `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY` environment variables - * 3. `/connection.json` (default `.agentworkforce/relay/connection.json`) + * 2. `--state-dir ` — that broker's `connection.json`, and nothing else + * 3. `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY` environment variables + * 4. The project default `.agentworkforce/relay/connection.json` + * + * An explicit `--state-dir` names one broker on purpose, so ambient env vars + * must not silently redirect it to another one (relay#1822). */ import fs from 'node:fs'; @@ -50,6 +53,27 @@ export function readConnectionFileFromDisk(stateDir: string): unknown { } } +/** + * Locate the directory that actually holds a broker's `connection.json` for + * a caller-supplied `--state-dir`. + * + * The broker writes `/connection.json`, but fleet nodes keep their + * broker state in a `state/` child of the node directory + * (`~/.agentworkforce/relay/-node/state`), so operators routinely pass + * the node directory instead. Accept that parent form as a fallback so the + * same path a node was configured with reaches it (relay#1575). The exact + * directory always wins when it has a connection file. + */ +export function resolveConnectionStateDir( + stateDir: string, + hasConnectionFile: (dir: string) => boolean = (dir) => fs.existsSync(path.join(dir, 'connection.json')) +): string { + const resolved = path.resolve(stateDir); + if (hasConnectionFile(resolved)) return resolved; + const nested = path.join(resolved, 'state'); + return hasConnectionFile(nested) ? nested : resolved; +} + /** Default state-directory: `.agentworkforce/relay/` under the resolved project root. */ export function defaultStateDir(): string { const projectRoot = getProjectPaths().projectRoot; @@ -90,24 +114,64 @@ export function resolveBrokerConnection( deps: BrokerConnectionDeps ): BrokerConnection | null { const explicitUrl = trimOrUndefined(options.brokerUrl); + const explicitKey = trimOrUndefined(options.apiKey); + const envKey = trimOrUndefined(deps.env.RELAY_BROKER_API_KEY); + const finish = (url: string, apiKey: string | undefined): BrokerConnection => ({ + url: url.replace(/\/+$/, ''), + apiKey, + ...(options.requestTimeoutMs === undefined ? {} : { requestTimeoutMs: options.requestTimeoutMs }), + }); + + const explicitStateDir = trimOrUndefined(options.stateDir); + if (explicitStateDir && !explicitUrl) { + // The caller named this broker's state directory: its connection file is + // the only source for both URL and key, so env vars pointing at another + // broker cannot pair that broker's URL or key with this one. + const stateDir = resolveConnectionStateDir( + explicitStateDir, + (dir) => readString(deps.readConnectionFile(dir), 'url') !== undefined + ); + const connectionFile = deps.readConnectionFile(stateDir); + const fileUrl = readString(connectionFile, 'url'); + if (!fileUrl) return null; + return finish(fileUrl, explicitKey ?? readString(connectionFile, 'api_key')); + } + const envUrl = trimOrUndefined(deps.env.RELAY_BROKER_URL); - const stateDir = options.stateDir ? path.resolve(options.stateDir) : deps.getDefaultStateDir(); + const stateDir = explicitStateDir ? path.resolve(explicitStateDir) : deps.getDefaultStateDir(); const connectionFile = deps.readConnectionFile(stateDir); const fileUrl = readString(connectionFile, 'url'); const url = explicitUrl ?? envUrl ?? fileUrl; if (!url) return null; - const explicitKey = trimOrUndefined(options.apiKey); - const envKey = trimOrUndefined(deps.env.RELAY_BROKER_API_KEY); const fileKey = readString(connectionFile, 'api_key'); - const apiKey = explicitKey ?? envKey ?? fileKey; + return finish(url, explicitKey ?? envKey ?? fileKey); +} - return { - url: url.replace(/\/+$/, ''), - apiKey, - ...(options.requestTimeoutMs === undefined ? {} : { requestTimeoutMs: options.requestTimeoutMs }), - }; +/** + * Explain a failed {@link resolveBrokerConnection} lookup by naming the path + * that was searched and which input chose it. "No broker" and "looked in the + * wrong place" have different remedies — only one warrants a restart. + */ +export function describeMissingBrokerConnection( + options: Pick, + deps: Pick +): string { + const explicitStateDir = trimOrUndefined(options.stateDir); + if (explicitStateDir) { + const resolved = path.resolve(explicitStateDir); + return ( + `Error: no broker connection at ${path.join(resolved, 'connection.json')} (from --state-dir; ` + + `also checked ${path.join(resolved, 'state', 'connection.json')}). ` + + 'Pass the same --state-dir the broker was started with.' + ); + } + return ( + `Error: no broker connection at ${path.join(deps.getDefaultStateDir(), 'connection.json')} (project default). ` + + 'If the broker was started with --state-dir, pass the same --state-dir here, ' + + 'or pass --broker-url / set RELAY_BROKER_URL.' + ); } /** Convert an `http(s)://host:port` base URL to the matching `ws(s)://…/ws`. */ diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index fc7dff1efe..8921b08239 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -9,6 +9,7 @@ import { redactCredentialValues } from '@agent-relay/cloud/redact'; import type { CoreDependencies, CoreProjectPaths, CoreRelay, SpawnedProcess } from '../commands/core.js'; import { + brokerIdentityDirectory, brokerIdentityPath, matchesBrokerIdentity, persistBrokerIdentity, @@ -56,6 +57,7 @@ import { type NodeClaim, } from './node-claim.js'; import { maskSecret } from './redact.js'; +import { resolveConnectionStateDir } from './broker-connection.js'; import { startReflexCapture, type RunningReflexCapture } from './reflex-capture.js'; import { readProjectWorkspaceSession, @@ -1077,7 +1079,7 @@ async function killOrphanedBrokerProcesses( const identities = readBrokerIdentities(paths, deps); if (!identities) { deps.warn( - `Broker identities could not be read in ${path.join(paths.projectRoot, '.agentworkforce', 'relay')}. State retained; inspect the records and verify process ownership before manual recovery.` + `Broker identities could not be read in ${brokerIdentityDirectory(paths, deps)}. State retained; inspect the records and verify process ownership before manual recovery.` ); return { matchedCount: 1, killedCount: 0 }; } @@ -1186,7 +1188,7 @@ async function recoverHalfStartedBroker( if (!stopped) { deps.error( `Failed to stop half-started broker process (pid: ${readiness.conn.pid}). ` + - `Verify process ownership manually before stopping it; inspect identities in ${path.join(paths.projectRoot, '.agentworkforce', 'relay')}. Connection metadata alone cannot authorize a signal.` + `Verify process ownership manually before stopping it; inspect identities in ${brokerIdentityDirectory(paths, deps)}. Connection metadata alone cannot authorize a signal.` ); return 'blocked'; } @@ -2026,7 +2028,7 @@ export async function runUpCommand(options: UpOptions, deps: CoreDependencies): if (!stopped) { deps.error( `Failed to stop half-started broker process (pid: ${cleanupPid}). ` + - `Verify process ownership before stopping it manually; inspect identities in ${path.join(paths.projectRoot, '.agentworkforce', 'relay')}.` + `Verify process ownership before stopping it manually; inspect identities in ${brokerIdentityDirectory(paths, deps)}.` ); } } @@ -2073,7 +2075,7 @@ export async function runUpCommand(options: UpOptions, deps: CoreDependencies): allStopped = false; deps.error( `Failed to stop broker process after Cloud enrollment startup failed (pid: ${cleanupPid}). ` + - `Verify process ownership before stopping it manually; inspect identities in ${path.join(paths.projectRoot, '.agentworkforce', 'relay')}.` + `Verify process ownership before stopping it manually; inspect identities in ${brokerIdentityDirectory(paths, deps)}.` ); } } @@ -2095,7 +2097,7 @@ export async function runUpCommand(options: UpOptions, deps: CoreDependencies): } if (!identityReady) { deps.error( - `Broker API is ready but process identity was not confirmed (pid: ${readiness.conn.pid}). State retained; verify ownership manually and inspect identities in ${path.join(paths.projectRoot, '.agentworkforce', 'relay')}.` + `Broker API is ready but process identity was not confirmed (pid: ${readiness.conn.pid}). State retained; verify ownership manually and inspect identities in ${brokerIdentityDirectory(paths, deps)}.` ); deps.exit(1); return; @@ -2510,11 +2512,22 @@ async function reportNodeClaimsElsewhere(paths: CoreProjectPaths, deps: CoreDepe deps.log('Stop one with: agent-relay node down --state-dir '); } +/** + * Resolve `--state-dir` for commands that inspect an existing broker. A fleet + * node directory holds its broker state in `state/`, so accept the node + * directory too when only the nested connection file exists (relay#1575). + */ +function resolveExistingStateDir(stateDir: string, deps: CoreDependencies): string { + return resolveConnectionStateDir(stateDir, (dir) => + deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)) + ); +} + // eslint-disable-next-line complexity, max-depth export async function runDownCommand(options: DownOptions, deps: CoreDependencies): Promise { const paths = deps.getProjectPaths(); if (options.stateDir) { - paths.dataDir = path.resolve(options.stateDir); + paths.dataDir = resolveExistingStateDir(options.stateDir, deps); } const timeout = Number.parseInt(options.timeout ?? '5000', 10) || 5000; @@ -2612,9 +2625,19 @@ export async function runDownCommand(options: DownOptions, deps: CoreDependencie const identities = readBrokerIdentities(paths, deps); const identity = identities?.find((record) => record.pid === pid); + if (identities && !identity) { + deps.error( + `No identity record for the running broker (pid: ${pid}) in ${brokerIdentityDirectory(paths, deps)}; retained its state. ` + + 'It was started by an older CLI or directly with `agent-relay-broker init`, which record none. ' + + `Verify ownership and stop it manually once, then start it with \`agent-relay node up --state-dir ${paths.dataDir}\` ` + + 'so later `node down` calls can verify it. Connection metadata alone cannot authorize a signal.' + ); + deps.exit(1); + return; + } if (!identity || !(await matchesBrokerIdentity(identity, paths, deps))) { deps.error( - `Broker identity could not be verified (pid: ${pid}); retained its state. Verify ownership before stopping the process manually; inspect identities in ${path.join(paths.projectRoot, '.agentworkforce', 'relay')}. Connection metadata alone cannot authorize a signal.` + `Broker identity could not be verified (pid: ${pid}); retained its state. Verify ownership before stopping the process manually; inspect identities in ${brokerIdentityDirectory(paths, deps)}. Connection metadata alone cannot authorize a signal.` ); deps.exit(1); return; @@ -2684,7 +2707,7 @@ export async function runStatusCommand( ): Promise { const paths = deps.getProjectPaths(); if (options?.stateDir) { - paths.dataDir = path.resolve(options.stateDir); + paths.dataDir = resolveExistingStateDir(options.stateDir, deps); } const waitMs = parseWaitForMs(options?.waitFor, deps); if (waitMs === null) { @@ -2727,6 +2750,14 @@ export async function runStatusCommand( } deps.log(`PID: ${readiness.conn.pid}`); deps.log(`Project: ${paths.projectRoot}`); + const identities = readBrokerIdentities(paths, deps); + if (identities && !identities.some((record) => record.pid === readiness.conn.pid)) { + // Surface this before an outage: without a record `node down` refuses to signal. + deps.warn( + `Broker identity: not recorded in ${brokerIdentityDirectory(paths, deps)}; \`node down\` cannot verify this broker. ` + + `Restart it once with \`agent-relay node up --state-dir ${paths.dataDir}\` to record one.` + ); + } const source = workspaceBindingSource(readiness.conn.workspace_source); deps.log( source diff --git a/packages/cli/src/cli/lib/broker-process-identity.test.ts b/packages/cli/src/cli/lib/broker-process-identity.test.ts index 54b3813c41..bca48eecde 100644 --- a/packages/cli/src/cli/lib/broker-process-identity.test.ts +++ b/packages/cli/src/cli/lib/broker-process-identity.test.ts @@ -107,7 +107,8 @@ describe.skipIf(!['linux', 'darwin'].includes(process.platform))('native broker } const filename = brokerIdentityPath(paths, deps, 'custom-node'); expect(fs.existsSync(filename)).toBe(true); - expect(path.dirname(filename)).toBe(path.join(projectRoot, '.agentworkforce', 'relay')); + // Records live with the broker state so `down --state-dir` finds them from any cwd. + expect(path.dirname(filename)).toBe(fs.realpathSync(stateDir)); const recorded = fs.readFileSync(filename, 'utf8'); if (mode === 'reopened') { await new Promise((resolve) => setTimeout(resolve, 20)); diff --git a/packages/cli/src/cli/lib/broker-process-identity.ts b/packages/cli/src/cli/lib/broker-process-identity.ts index 5757f021e7..0611d23922 100644 --- a/packages/cli/src/cli/lib/broker-process-identity.ts +++ b/packages/cli/src/cli/lib/broker-process-identity.ts @@ -15,31 +15,60 @@ export type BrokerProcessIdentity = { type IdentityDependencies = Pick; -function stateDirectory(paths: CoreProjectPaths, deps?: IdentityDependencies): string { +function canonicalDirectory(directory: string, deps?: IdentityDependencies): string { try { - return (deps?.fs.realpathSync ?? fs.realpathSync)(paths.dataDir); + return (deps?.fs.realpathSync ?? fs.realpathSync)(directory); } catch { - return path.resolve(paths.dataDir); + return path.resolve(directory); } } +function stateDirectory(paths: CoreProjectPaths, deps?: IdentityDependencies): string { + return canonicalDirectory(paths.dataDir, deps); +} + function identityPrefix(paths: CoreProjectPaths, deps?: IdentityDependencies): string { const scope = createHash('sha256').update(stateDirectory(paths, deps)).digest('hex'); return `broker-identity-${scope}`; } +/** + * Directory holding identity records: the broker's own state directory, so + * `node down --state-dir ` finds the record from any working directory + * (relay#1820). For the default state dir this is the project data dir. + */ +export function brokerIdentityDirectory(paths: CoreProjectPaths, deps?: IdentityDependencies): string { + return stateDirectory(paths, deps); +} + +/** + * Directories searched for identity records, newest layout first. Earlier + * releases wrote every record under the invoking project's data dir even when + * `--state-dir` redirected broker state; keep reading there so brokers started + * by those releases stay verifiable. + */ +function identityDirectories(paths: CoreProjectPaths, deps?: IdentityDependencies): string[] { + const primary = brokerIdentityDirectory(paths, deps); + const legacy = canonicalDirectory(path.join(paths.projectRoot, '.agentworkforce', 'relay'), deps); + return primary === legacy ? [primary] : [primary, legacy]; +} + +function identityFilename( + directory: string, + paths: CoreProjectPaths, + deps: IdentityDependencies | undefined, + brokerName: string +): string { + const name = createHash('sha256').update(brokerName).digest('hex'); + return path.join(directory, `${identityPrefix(paths, deps)}-${name}.json`); +} + export function brokerIdentityPath( paths: CoreProjectPaths, deps?: IdentityDependencies, brokerName = path.basename(paths.projectRoot) || 'project' ): string { - const name = createHash('sha256').update(brokerName).digest('hex'); - return path.join( - paths.projectRoot, - '.agentworkforce', - 'relay', - `${identityPrefix(paths, deps)}-${name}.json` - ); + return identityFilename(brokerIdentityDirectory(paths, deps), paths, deps, brokerName); } function normalizedStart(value: string): string | null { @@ -86,15 +115,36 @@ export async function readBrokerProcessIdentity( } } +/** Locate the record for `brokerName`, preferring the state-dir location over the legacy one. */ +function findBrokerIdentityRecord( + paths: CoreProjectPaths, + deps: IdentityDependencies, + brokerName: string +): { identity: BrokerProcessIdentity; filename: string } | null { + for (const directory of identityDirectories(paths, deps)) { + const filename = identityFilename(directory, paths, deps, brokerName); + const identity = readBrokerIdentityFile(filename, paths, deps, brokerName); + if (identity) return { identity, filename }; + } + return null; +} + export function readBrokerIdentity( paths: CoreProjectPaths, deps: IdentityDependencies, brokerName = path.basename(paths.projectRoot) || 'project' +): BrokerProcessIdentity | null { + return findBrokerIdentityRecord(paths, deps, brokerName)?.identity ?? null; +} + +function readBrokerIdentityFile( + filename: string, + paths: CoreProjectPaths, + deps: IdentityDependencies, + brokerName: string ): BrokerProcessIdentity | null { try { - const value = JSON.parse( - deps.fs.readFileSync(brokerIdentityPath(paths, deps, brokerName), 'utf-8') - ) as BrokerProcessIdentity; + const value = JSON.parse(deps.fs.readFileSync(filename, 'utf-8')) as BrokerProcessIdentity; if ( !value || value.version !== 1 || @@ -126,34 +176,38 @@ export function readBrokerIdentities( paths: CoreProjectPaths, deps: IdentityDependencies ): BrokerProcessIdentity[] | null { - const directory = path.join(paths.projectRoot, '.agentworkforce', 'relay'); - let filenames: string[]; - try { - filenames = deps.fs.readdirSync(directory); - } catch (error) { - return (error as NodeJS.ErrnoException)?.code === 'ENOENT' ? [] : null; - } - try { - const prefix = identityPrefix(paths, deps); - const records: BrokerProcessIdentity[] = []; - for (const filename of filenames) { - if (!filename.startsWith(prefix)) continue; - if (!new RegExp(`^${prefix}-[a-f0-9]{64}\\.json$`).test(filename)) return null; - const raw = JSON.parse(deps.fs.readFileSync(path.join(directory, filename), 'utf-8')); - if (typeof raw?.brokerName !== 'string') return null; - const identity = readBrokerIdentity(paths, deps, raw.brokerName); - if ( - !identity || - brokerIdentityPath(paths, deps, identity.brokerName) !== path.join(directory, filename) - ) - return null; - records.push(identity); + const records: BrokerProcessIdentity[] = []; + const seen = new Set(); + for (const directory of identityDirectories(paths, deps)) { + let filenames: string[]; + try { + filenames = deps.fs.readdirSync(directory); + } catch (error) { + if ((error as NodeJS.ErrnoException)?.code === 'ENOENT') continue; + return null; + } + try { + const prefix = identityPrefix(paths, deps); + for (const filename of filenames) { + if (!filename.startsWith(prefix)) continue; + if (!new RegExp(`^${prefix}-[a-f0-9]{64}\\.json$`).test(filename)) return null; + const filepath = path.join(directory, filename); + const raw = JSON.parse(deps.fs.readFileSync(filepath, 'utf-8')); + if (typeof raw?.brokerName !== 'string') return null; + const identity = readBrokerIdentityFile(filepath, paths, deps, raw.brokerName); + if (!identity || identityFilename(directory, paths, deps, identity.brokerName) !== filepath) + return null; + // The state-dir record shadows a legacy record for the same broker name. + if (seen.has(identity.brokerName)) continue; + seen.add(identity.brokerName); + records.push(identity); + } + } catch { + // A listed record disappearing is an uncertain snapshot, not legacy state. + return null; } - return records; - } catch { - // A listed record disappearing is an uncertain snapshot, not legacy state. - return null; } + return records; } export async function persistBrokerIdentity( @@ -278,10 +332,10 @@ export function removeBrokerIdentity( identity: BrokerProcessIdentity, deps: IdentityDependencies ): void { - const current = readBrokerIdentity(paths, deps, identity.brokerName); - if (!current || JSON.stringify(current) !== JSON.stringify(identity)) return; + const current = findBrokerIdentityRecord(paths, deps, identity.brokerName); + if (!current || JSON.stringify(current.identity) !== JSON.stringify(identity)) return; try { - deps.fs.unlinkSync(brokerIdentityPath(paths, deps, identity.brokerName)); + deps.fs.unlinkSync(current.filename); } catch { /* Best effort. */ } diff --git a/tests/relayflows/cases/1446-state-dir-broker-selection/case.json b/tests/relayflows/cases/1446-state-dir-broker-selection/case.json new file mode 100644 index 0000000000..ea68959983 --- /dev/null +++ b/tests/relayflows/cases/1446-state-dir-broker-selection/case.json @@ -0,0 +1,21 @@ +{ + "version": 1, + "id": "1446-state-dir-broker-selection", + "kind": "bugfix", + "title": "Target a --state-dir broker from node agent list and release", + "runner": { + "command": ["node", "tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs"] + }, + "requirements": [], + "timeoutSeconds": 900, + "expected": { + "base": { + "outcome": "bug", + "signature": "node_agent_state_dir_rejected" + }, + "head": { + "outcome": "fixed", + "signature": "node_agent_state_dir_targets_broker" + } + } +} diff --git a/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs b/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs new file mode 100755 index 0000000000..120f344030 --- /dev/null +++ b/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs @@ -0,0 +1,258 @@ +#!/usr/bin/env node + +/** + * Red/green proof for relay#1446 / relay#1822 / relay#1575: `node agent` + * subcommands must reach a broker started with `--state-dir` from an unrelated + * working directory. + * + * The exact-head harness injects a Vitest probe into each exact target checkout. + * The probe drives the production Commander registration and the real + * HarnessDriverClient against a loopback fixture broker whose connection.json + * lives in a fleet-node style `/state/` directory. A decoy broker named by + * RELAY_BROKER_URL must never be contacted when --state-dir is explicit. + */ + +import { execFileSync, spawnSync } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import { lstat, mkdir, open, readFile, rename, rm, writeFile } from 'node:fs/promises'; +import path from 'node:path'; +import process from 'node:process'; +import { fileURLToPath } from 'node:url'; + +const CASE_ID = '1446-state-dir-broker-selection'; +const COMMAND_TIMEOUT_MS = 5 * 60 * 1000; + +const targetDir = requiredDirectory('RELAY_PR_PROOF_TARGET_DIR'); +const harnessDir = requiredDirectory('RELAY_PR_PROOF_HARNESS_DIR'); +const resultPath = path.resolve(requiredValue('RELAY_PR_PROOF_RESULT_PATH')); +const arm = requiredValue('RELAY_PR_PROOF_ARM'); +if (arm !== 'base' && arm !== 'head') { + throw new Error(`RELAY_PR_PROOF_ARM must be base or head, received ${JSON.stringify(arm)}.`); +} + +const expectedSha = + arm === 'base' ? process.env.RELAY_PR_PROOF_BASE_SHA : process.env.RELAY_PR_PROOF_HEAD_SHA; +if (!expectedSha) throw new Error(`Missing expected ${arm} SHA.`); +const targetSha = execFileSync('git', ['-C', targetDir, 'rev-parse', 'HEAD'], { + encoding: 'utf8', + timeout: COMMAND_TIMEOUT_MS, +}).trim(); +if (targetSha !== expectedSha) { + throw new Error(`Target checkout ${targetSha} does not match exact ${arm} SHA ${expectedSha}.`); +} + +const runnerPath = fileURLToPath(import.meta.url); +if (!isWithin(harnessDir, runnerPath)) { + throw new Error('The RelayFlow runner must execute from the exact-head harness checkout.'); +} + +const probePath = path.join(targetDir, 'packages/cli/src/cli/.relayflow-1446-state-dir.test.ts'); +const configPath = path.join(targetDir, '.relayflow-1446-state-dir.vitest.config.mjs'); +const observationPath = path.join(targetDir, '.relayflow-1446-state-dir-observation.json'); + +const probeSource = String.raw`import fs from 'node:fs'; +import { writeFile } from 'node:fs/promises'; +import http from 'node:http'; +import os from 'node:os'; +import path from 'node:path'; + +import { Command } from 'commander'; +import { test } from 'vitest'; + +import { registerLocalAgentCommands } from './commands/local-agent.js'; + +const observationPath = process.env.RELAY_PR1446_OBSERVATION_PATH; + +function startBroker(label, requests) { + const server = http.createServer((req, res) => { + requests.push({ broker: label, method: req.method, url: req.url, apiKey: req.headers['x-api-key'] ?? null }); + res.setHeader('content-type', 'application/json'); + if (req.method === 'GET' && req.url === '/api/spawned') { + res.end(JSON.stringify({ agents: [{ name: 'worker' }] })); + return; + } + if (req.method === 'DELETE' && req.url === '/api/spawned/worker') { + res.end(JSON.stringify({ name: 'worker' })); + return; + } + res.statusCode = 404; + res.end(JSON.stringify({ error: 'not found' })); + }); + return new Promise((resolve) => { + server.listen(0, '127.0.0.1', () => resolve(server)); + }); +} + +async function invoke(argv, cwd, env) { + const logs = []; + const errors = []; + const program = new Command(); + program.exitOverride(); + registerLocalAgentCommands(program.command('node'), { + cwd: () => cwd, + env, + log: (...args) => logs.push(args.join(' ')), + error: (...args) => errors.push(args.join(' ')), + exit: (code) => { + throw new Error('cli exit ' + code); + }, + }); + let error = null; + try { + await program.parseAsync(argv, { from: 'user' }); + } catch (caught) { + error = caught instanceof Error ? caught.message : String(caught); + } + return { logs, errors, error }; +} + +test('observe node agent --state-dir broker selection', async () => { + const requests = []; + const broker = await startBroker('node', requests); + const decoy = await startBroker('decoy', requests); + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'relayflow-1446-')); + const nodeDir = path.join(root, 'sf-mini-node'); + const stateDir = path.join(nodeDir, 'state'); + const unrelatedCwd = path.join(root, 'elsewhere'); + fs.mkdirSync(stateDir, { recursive: true }); + fs.mkdirSync(path.join(unrelatedCwd, '.git'), { recursive: true }); + fs.writeFileSync( + path.join(stateDir, 'connection.json'), + JSON.stringify({ url: 'http://127.0.0.1:' + broker.address().port, api_key: 'proof-node-key', pid: process.pid }) + ); + const env = { + RELAY_BROKER_URL: 'http://127.0.0.1:' + decoy.address().port, + RELAY_BROKER_API_KEY: 'decoy-key', + }; + try { + const list = await invoke(['node', 'agent', 'list', '--state-dir', stateDir], unrelatedCwd, env); + const release = await invoke(['node', 'agent', 'release', 'worker', '--state-dir', nodeDir], unrelatedCwd, env); + await writeFile(observationPath, JSON.stringify({ list, release, requests }), 'utf8'); + } finally { + broker.close(); + decoy.close(); + fs.rmSync(root, { recursive: true, force: true }); + } +}); +`; + +const configSource = `export default { + test: { + environment: 'node', + include: ['packages/cli/src/cli/.relayflow-1446-state-dir.test.ts'], + setupFiles: [], + }, +};\n`; + +try { + run('npm', ['ci', '--ignore-scripts'], targetDir, 'workspace dependency installation'); + run('npm', ['run', 'build:core'], targetDir, 'workspace package build'); + await writeGeneratedFile(probePath, probeSource); + await writeGeneratedFile(configPath, configSource); + run( + 'npm', + ['exec', '--', 'vitest', 'run', '--config', path.relative(targetDir, configPath)], + targetDir, + 'state-dir broker selection probe', + { RELAY_PR1446_OBSERVATION_PATH: observationPath } + ); + + const observation = JSON.parse(await readFile(observationPath, 'utf8')); + console.log('State-dir broker selection observation:', JSON.stringify(observation)); + const { list, release, requests } = observation; + const rejected = (result) => + typeof result?.error === 'string' && /unknown option '--state-dir'/.test(result.error); + const baseObserved = rejected(list) && rejected(release) && requests.length === 0; + const nodeRequests = requests.filter((request) => request.broker === 'node'); + const headObserved = + list?.error === null && + release?.error === null && + list.errors.length === 0 && + release.errors.length === 0 && + list.logs.some((line) => line.includes('"worker"')) && + release.logs.includes('Released worker.') && + requests.every((request) => request.broker === 'node' && request.apiKey === 'proof-node-key') && + nodeRequests.some((request) => request.method === 'GET' && request.url === '/api/spawned') && + nodeRequests.some((request) => request.method === 'DELETE' && request.url === '/api/spawned/worker'); + + let outcome; + let signature; + let details; + if (baseObserved) { + outcome = 'bug'; + signature = 'node_agent_state_dir_rejected'; + details = + "The exact base CLI rejects --state-dir on node agent list and release with unknown option '--state-dir', so a broker started with --state-dir cannot be managed from another directory."; + } else if (headObserved) { + outcome = 'fixed'; + signature = 'node_agent_state_dir_targets_broker'; + details = + 'The exact head CLI lists and releases through the broker named by --state-dir (both the state dir and its fleet node parent) from an unrelated cwd, authenticating with that broker key and never contacting the RELAY_BROKER_URL decoy.'; + } else { + throw new Error(`Unexpected state-dir broker selection observation: ${JSON.stringify(observation)}.`); + } + + await mkdir(path.dirname(resultPath), { recursive: true }); + await writeFile( + resultPath, + `${JSON.stringify({ version: 1, caseId: CASE_ID, arm, outcome, signature, details }, null, 2)}\n`, + 'utf8' + ); +} finally { + await rm(probePath, { force: true }); + await rm(configPath, { force: true }); + await rm(observationPath, { force: true }); +} + +function requiredValue(name) { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`Missing required environment variable ${name}.`); + return value; +} + +function requiredDirectory(name) { + return path.resolve(requiredValue(name)); +} + +function isWithin(directory, candidate) { + const relative = path.relative(directory, candidate); + return ( + relative === '' || + (!relative.startsWith(`..${path.sep}`) && relative !== '..' && !path.isAbsolute(relative)) + ); +} + +async function writeGeneratedFile(targetPath, source) { + try { + const existing = await lstat(targetPath); + if (!existing.isFile()) throw new Error(`Refusing to replace non-file ${targetPath}.`); + } catch (error) { + if (error?.code !== 'ENOENT') throw error; + } + + const temporaryPath = `${targetPath}.tmp-${process.pid}-${randomUUID()}`; + try { + const handle = await open(temporaryPath, 'wx', 0o600); + await handle.writeFile(source, 'utf8'); + await handle.close(); + await rename(temporaryPath, targetPath); + } finally { + await rm(temporaryPath, { force: true }); + } +} + +function run(command, args, cwd, label, extraEnv = {}) { + const result = spawnSync(command, args, { + cwd, + env: { ...process.env, ...extraEnv }, + encoding: 'utf8', + timeout: COMMAND_TIMEOUT_MS, + maxBuffer: 8 * 1024 * 1024, + }); + if (result.error) throw new Error(`${label} failed to start: ${result.error.message}`); + if (result.status !== 0) { + throw new Error( + `${label} exited ${result.status ?? 'null'}:\n${(result.stdout ?? '').slice(-2000)}\n${(result.stderr ?? '').slice(-2000)}` + ); + } +} From e801a48712d0e2e841533cf4c8d5a1f76e160a50 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 06:17:12 +0000 Subject: [PATCH 02/14] fix(cli): keep exact --state-dir authoritative and recover nested orphans - Select the exact --state-dir whenever its connection.json exists, so a malformed file fails instead of redirecting to a nested state/ broker. - node down/status treat lock and identity files as broker-state evidence, so down --force --state-dir still recovers a nested broker whose connection.json is gone. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/commands/core.test.ts | 22 +++++++++++++++++++ .../cli/src/cli/lib/broker-connection.test.ts | 18 +++++++++++++++ packages/cli/src/cli/lib/broker-connection.ts | 4 +++- packages/cli/src/cli/lib/broker-lifecycle.ts | 19 ++++++++++++---- 4 files changed, 58 insertions(+), 5 deletions(-) diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index a2883e45de..3d535ac794 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -1649,6 +1649,28 @@ describe('registerCoreCommands', () => { } ); + it('down --force via the node directory recovers a nested broker whose connection file is gone', async () => { + const stateDir = '/srv/orphan-node/state'; + const running = new Set([222, 333]); + const execCommand = identityCommand(222, `${stateDir}/broker-orphan.lock`); + const killImpl = vi.fn((pid: number, signal?: NodeJS.Signals | number) => { + if (signal === 0) { + if (!running.has(pid)) throw new Error('not running'); + return; + } + running.delete(pid); + }); + const { program, deps } = createHarness({ execCommand, killImpl }); + const paths = { ...deps.getProjectPaths(), dataDir: stateDir }; + await persistBrokerIdentity(paths, 222, 'orphan', deps); + expect(deps.fs.existsSync(`${stateDir}/connection.json`)).toBe(false); + + await runCommand(program, ['down', '--force', '--state-dir', '/srv/orphan-node']); + + expect(killImpl).toHaveBeenCalledWith(222, 'SIGTERM'); + expect([...running]).toEqual([333]); + }); + it('down still verifies records written under the project by earlier releases', async () => { const stateDir = '/srv/legacy-node/state'; let running = true; diff --git a/packages/cli/src/cli/lib/broker-connection.test.ts b/packages/cli/src/cli/lib/broker-connection.test.ts index 79dbba0ddc..4ac3e1f677 100644 --- a/packages/cli/src/cli/lib/broker-connection.test.ts +++ b/packages/cli/src/cli/lib/broker-connection.test.ts @@ -159,6 +159,24 @@ describe('explicit --state-dir selection', () => { }); }); +describe('malformed exact connection file', () => { + it('fails instead of falling back to a nested broker', () => { + const nodeDir = fs.mkdtempSync(path.join(os.tmpdir(), 'relay-node-dir-')); + try { + fs.mkdirSync(path.join(nodeDir, 'state')); + fs.writeFileSync(path.join(nodeDir, 'connection.json'), '{not json'); + fs.writeFileSync( + path.join(nodeDir, 'state', 'connection.json'), + JSON.stringify({ url: 'http://127.0.0.1:4555', api_key: 'nested' }) + ); + const deps = makeDeps({ readConnectionFile: readConnectionFileFromDisk }); + expect(resolveBrokerConnection({ stateDir: nodeDir }, deps)).toBeNull(); + } finally { + fs.rmSync(nodeDir, { recursive: true, force: true }); + } + }); +}); + describe('describeMissingBrokerConnection', () => { it('names the --state-dir paths that were searched', () => { expect(describeMissingBrokerConnection({ stateDir: '/srv/node' }, makeDeps())).toBe( diff --git a/packages/cli/src/cli/lib/broker-connection.ts b/packages/cli/src/cli/lib/broker-connection.ts index 1f20643e97..4b8a315f15 100644 --- a/packages/cli/src/cli/lib/broker-connection.ts +++ b/packages/cli/src/cli/lib/broker-connection.ts @@ -127,9 +127,11 @@ export function resolveBrokerConnection( // The caller named this broker's state directory: its connection file is // the only source for both URL and key, so env vars pointing at another // broker cannot pair that broker's URL or key with this one. + // Select by file presence, not validity: a malformed exact file must fail + // here rather than silently redirect to a nested broker. const stateDir = resolveConnectionStateDir( explicitStateDir, - (dir) => readString(deps.readConnectionFile(dir), 'url') !== undefined + (dir) => fs.existsSync(path.join(dir, 'connection.json')) || deps.readConnectionFile(dir) !== null ); const connectionFile = deps.readConnectionFile(stateDir); const fileUrl = readString(connectionFile, 'url'); diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 8921b08239..9c6ccc55a6 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -2515,12 +2515,23 @@ async function reportNodeClaimsElsewhere(paths: CoreProjectPaths, deps: CoreDepe /** * Resolve `--state-dir` for commands that inspect an existing broker. A fleet * node directory holds its broker state in `state/`, so accept the node - * directory too when only the nested connection file exists (relay#1575). + * directory too when only the nested one holds broker state (relay#1575). + * Lock and identity files count as evidence alongside connection.json so + * `down --force` can still recover a broker whose connection file is gone. */ function resolveExistingStateDir(stateDir: string, deps: CoreDependencies): string { - return resolveConnectionStateDir(stateDir, (dir) => - deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)) - ); + return resolveConnectionStateDir(stateDir, (dir) => hasBrokerStateEvidence(dir, deps)); +} + +function hasBrokerStateEvidence(dir: string, deps: CoreDependencies): boolean { + if (deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME))) return true; + try { + return deps.fs + .readdirSync(dir) + .some((file) => file.startsWith('broker-') && (file.endsWith('.lock') || file.endsWith('.json'))); + } catch { + return false; + } } // eslint-disable-next-line complexity, max-depth From fb7ba95af1a729caf99087fd5f228dbe5f52365c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 06:23:37 +0000 Subject: [PATCH 03/14] fix(cli): drop superseded legacy broker identity records on launch Once a launch is proven to hold the broker lock and its state-dir record is written, remove the same broker's record from the legacy project location so it cannot resurface after down and block cleanup. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/commands/core.test.ts | 28 +++++++++++++++++++ .../src/cli/lib/broker-process-identity.ts | 24 ++++++++++++++++ 2 files changed, 52 insertions(+) diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index 3d535ac794..e27b191f47 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -1671,6 +1671,34 @@ describe('registerCoreCommands', () => { expect([...running]).toEqual([333]); }); + it('a new launch supersedes a stale record left by an earlier release', async () => { + const stateDir = '/srv/upgraded-node/state'; + let running = true; + const fs = createFsMock({ [`${stateDir}/connection.json`]: connectionFile(222) }); + const { program, deps } = createHarness({ + fs, + execCommand: identityCommand(222, `${stateDir}/broker-upgraded.lock`), + killImpl: vi.fn((_pid, signal) => { + if (signal === 'SIGTERM') running = false; + if (signal === 0 && !running) throw new Error('not running'); + }), + }); + const paths = { ...deps.getProjectPaths(), dataDir: stateDir }; + await persistBrokerIdentity(paths, 222, 'upgraded', deps); + const current = brokerIdentityPath(paths, deps, 'upgraded'); + const legacy = nodePath.join(paths.projectRoot, '.agentworkforce', 'relay', nodePath.basename(current)); + // A record an earlier release left for a previous launch of this broker. + fs.writeFileSync(legacy, fs.readFileSync(current, 'utf-8').replace('"pid":222', '"pid":111')); + + await persistBrokerIdentity(paths, 222, 'upgraded', deps); + expect(fs.existsSync(legacy)).toBe(false); + + await runCommand(program, ['down', '--force', '--state-dir', stateDir]); + expect(deps.log).toHaveBeenCalledWith('Stopped'); + expect(readBrokerIdentities(paths, deps)).toEqual([]); + expect(fs.existsSync(`${stateDir}/connection.json`)).toBe(false); + }); + it('down still verifies records written under the project by earlier releases', async () => { const stateDir = '/srv/legacy-node/state'; let running = true; diff --git a/packages/cli/src/cli/lib/broker-process-identity.ts b/packages/cli/src/cli/lib/broker-process-identity.ts index 0611d23922..dd6ec22ca1 100644 --- a/packages/cli/src/cli/lib/broker-process-identity.ts +++ b/packages/cli/src/cli/lib/broker-process-identity.ts @@ -241,6 +241,7 @@ export async function persistBrokerIdentity( deps.fs.mkdirSync(path.dirname(filename), { recursive: true }); deps.fs.writeFileSync(temporary, `${JSON.stringify(record)}\n`, 'utf-8'); deps.fs.renameSync(temporary, filename); + removeSupersededIdentities(paths, filename, brokerName, deps); return record; } catch { // A failed write never grants fallback ownership of an unrecorded process. @@ -253,6 +254,29 @@ export async function persistBrokerIdentity( } } +/** + * The new record proves this launch holds the broker lock, so any record for + * the same state dir and name in an older location describes a dead launch. + * Left in place it would resurface once the new record is removed and keep + * cleanup from deleting the stopped broker's connection file. + */ +function removeSupersededIdentities( + paths: CoreProjectPaths, + current: string, + brokerName: string, + deps: IdentityDependencies +): void { + for (const directory of identityDirectories(paths, deps)) { + const filename = identityFilename(directory, paths, deps, brokerName); + if (filename === current) continue; + try { + deps.fs.unlinkSync(filename); + } catch { + /* Absent or unavailable. */ + } + } +} + export async function matchesBrokerIdentity( identity: BrokerProcessIdentity, paths: CoreProjectPaths, From fc96112e05d294234f9a9e4dc4b0dd44ff25d69c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 06:26:02 +0000 Subject: [PATCH 04/14] fix(cli): address review on --state-dir broker selection - --state-dir stays authoritative with --broker-url: the key comes from that state dir (nested state/ included), never from the env. - RELAY_BROKER_API_KEY alone selects the local broker, matching attach. - removeBrokerIdentity removes every location holding the exact record. - RelayFlow proof now spawns the real built agent-relay binary. - Platform-aware test fixtures; README and CHANGELOG clarifications. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- CHANGELOG.md | 8 +- packages/cli/README.md | 2 +- .../cli/src/cli/commands/local-agent.test.ts | 15 +- packages/cli/src/cli/commands/local-agent.ts | 6 +- .../cli/src/cli/lib/broker-connection.test.ts | 30 ++- packages/cli/src/cli/lib/broker-connection.ts | 16 +- .../src/cli/lib/broker-process-identity.ts | 17 +- .../1446-state-dir-broker-selection/run.mjs | 232 +++++++----------- 8 files changed, 151 insertions(+), 175 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 70afb5c795..a34f28ac1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,13 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- `node agent list|spawn|new|release|set-model` and `node tail` accept `--state-dir`, `--broker-url`, and `--api-key` (or `RELAY_BROKER_URL`), so brokers started with `node up --state-dir` — such as fleet nodes — can be managed from any directory. +- `node agent list|spawn|new|release|set-model` and `node tail` accept `--state-dir`, `--broker-url`, and `--api-key`, so brokers started with `node up --state-dir` — such as fleet nodes — can be managed from any directory. Without these flags, `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY` or the enclosing project's broker is used. ### Fixed -- `--state-dir` on `node agent` commands, `node status`, and `node down` also accepts a fleet node directory whose broker state lives in `state/`, and an explicit `--state-dir` is no longer overridden by ambient `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`. +- `--state-dir` on `node agent` commands, `node status`, and `node down` also accepts a fleet node directory whose broker state lives in `state/`. +- An explicit `--state-dir` on `node agent` commands is no longer overridden by `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`. - "No broker connection" errors name the path searched and whether it was the project default or `--state-dir`, instead of implying a running broker is down. -- `node down --force --state-dir ` can verify and stop brokers started from another working directory: identity records now live in the broker state dir (records written by earlier releases are still read). `node status` and `node down` name a missing identity record and how to create one. +- `node down --force --state-dir ` can verify and stop brokers started from another working directory. +- `node status` and `node down` name a missing broker identity record and how to create one. - Broker: Muse workers now reuse the host Muse login (`RELAY_MUSE_SHARED_AUTH_PATH`, the host's `MUSE_AUTH_PATH`, `$XDG_CONFIG_HOME/muse/auth.json`, or `$HOME/.config/muse/auth.json`) instead of prompting for provider auth under every new agent name; Relay MCP settings stay per-worker. Multi-tenant hosts should set `RELAY_MUSE_ISOLATED_AUTH=1` before upgrading to keep a separate login per worker. diff --git a/packages/cli/README.md b/packages/cli/README.md index e592462629..afa86e8e9c 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -45,7 +45,7 @@ agent-relay node agent attach --mode view agent-relay node agent release ``` -Every `node agent` subcommand (and `node tail`) accepts `--state-dir `, `--broker-url `, and `--api-key ` to target a broker other than the current project's — for example a fleet node started with `node up --state-dir`. `--state-dir` may name the broker state dir or a fleet node directory whose broker state lives in `state/`. An explicit `--state-dir` reads only that broker's `connection.json`, ignoring `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`; without flags, a nonblank `RELAY_BROKER_URL` selects the broker, otherwise the enclosing project's broker is used. +Local-broker `node agent` subcommands and `node tail` accept `--state-dir `, `--broker-url `, and `--api-key ` to target a broker other than the current project's — for example a fleet node started with `node up --state-dir`. `--state-dir` may name the broker state dir or a fleet node directory whose broker state lives in `state/`. An explicit `--state-dir` reads only that broker's `connection.json` (`--broker-url` may still override its URL), ignoring `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`. Without flags, a nonblank `RELAY_BROKER_URL` or `RELAY_BROKER_API_KEY` selects the local broker; otherwise the enclosing project's broker is used, except that flag-free `attach` and `message flush|hold|auto` may auto-route to a live fleet placement. Remote routes differ: `attach --node` rejects these flags, `attach --ssh-host` rejects `--broker-url` / `--api-key`, and `message … --node` ignores them. `node agent spawn` and `node agent new` accept `--runtime auto|native|pty`. `auto` is the default and keeps experimental dual-runtime adapters on PTY. Claude Code, Codex, and OpenCode support explicit native or PTY selection; Pi and Deep Agents are experimental native-only harnesses and require `--runtime native`. diff --git a/packages/cli/src/cli/commands/local-agent.test.ts b/packages/cli/src/cli/commands/local-agent.test.ts index 6a497b6b5a..943ce28abc 100644 --- a/packages/cli/src/cli/commands/local-agent.test.ts +++ b/packages/cli/src/cli/commands/local-agent.test.ts @@ -1256,10 +1256,11 @@ describe('explicit broker selection on every node agent subcommand (relay#1446, expect(log).toHaveBeenCalledWith('Released worker.'); }); - it('RELAY_BROKER_URL selects the broker for flag-free release', async () => { - const { program, connect, connectLocal, local } = selectionHarness({ - RELAY_BROKER_URL: 'http://127.0.0.1:4100', - }); + it.each([ + ['RELAY_BROKER_URL', 'http://127.0.0.1:4100'], + ['RELAY_BROKER_API_KEY', 'env-key'], + ])('%s selects the broker for flag-free release', async (name, value) => { + const { program, connect, connectLocal, local } = selectionHarness({ [name]: value }); await program.parseAsync(['local', 'agent', 'release', 'worker'], { from: 'user' }); expect(connect).not.toHaveBeenCalled(); expect(connectLocal).toHaveBeenCalled(); @@ -1274,7 +1275,11 @@ describe('explicit broker selection on every node agent subcommand (relay#1446, from: 'user', } ); - expect(connectLocal).toHaveBeenCalled(); + expect(connectLocal).toHaveBeenCalledWith('/tmp/project', { + brokerUrl: undefined, + apiKey: undefined, + stateDir, + }); expect(local.spawnPty).toHaveBeenCalled(); expect(attach).toHaveBeenCalledWith( 'worker', diff --git a/packages/cli/src/cli/commands/local-agent.ts b/packages/cli/src/cli/commands/local-agent.ts index 6d667a9141..eeb23cbd79 100644 --- a/packages/cli/src/cli/commands/local-agent.ts +++ b/packages/cli/src/cli/commands/local-agent.ts @@ -502,7 +502,8 @@ export function formatPrettyAgentStatusList(agents: AgentDeliveryStatus[], now: /** * Whether the caller chose a broker explicitly (`--broker-url` / `--api-key` / - * `--state-dir`, or `RELAY_BROKER_URL`). Without one, commands keep using the + * `--state-dir`, or `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`), matching the + * selection `attach` and `message` honour. Without one, commands keep using the * enclosing project's broker. */ function hasExplicitBrokerSelection( @@ -513,7 +514,8 @@ function hasExplicitBrokerSelection( options.brokerUrl?.trim() || options.apiKey?.trim() || options.stateDir?.trim() || - deps.env.RELAY_BROKER_URL?.trim() + deps.env.RELAY_BROKER_URL?.trim() || + deps.env.RELAY_BROKER_API_KEY?.trim() ); } diff --git a/packages/cli/src/cli/lib/broker-connection.test.ts b/packages/cli/src/cli/lib/broker-connection.test.ts index 4ac3e1f677..7657c9384d 100644 --- a/packages/cli/src/cli/lib/broker-connection.test.ts +++ b/packages/cli/src/cli/lib/broker-connection.test.ts @@ -102,8 +102,11 @@ describe('resolveBrokerConnection', () => { }); describe('explicit --state-dir selection', () => { + // Fixture keys go through path.resolve, as the resolver does, so they match on every platform. + const nodeDir = path.resolve('/srv/node'); + const nodeStateDir = path.join(nodeDir, 'state'); const files: Record = { - '/srv/node/state': { url: 'http://node-host:4100/', api_key: 'node-key' }, + [nodeStateDir]: { url: 'http://node-host:4100/', api_key: 'node-key' }, }; const readConnectionFile = vi.fn((dir: string) => files[dir] ?? null); @@ -112,7 +115,7 @@ describe('explicit --state-dir selection', () => { env: { RELAY_BROKER_URL: 'http://other:1', RELAY_BROKER_API_KEY: 'other-key' }, readConnectionFile, }); - expect(resolveBrokerConnection({ stateDir: '/srv/node/state' }, deps)).toEqual({ + expect(resolveBrokerConnection({ stateDir: nodeStateDir }, deps)).toEqual({ url: 'http://node-host:4100', apiKey: 'node-key', }); @@ -120,19 +123,27 @@ describe('explicit --state-dir selection', () => { it('never falls back to env when the named state dir has no broker', () => { const deps = makeDeps({ env: { RELAY_BROKER_URL: 'http://other:1' }, readConnectionFile }); - expect(resolveBrokerConnection({ stateDir: '/srv/missing' }, deps)).toBeNull(); + expect(resolveBrokerConnection({ stateDir: path.resolve('/srv/missing') }, deps)).toBeNull(); }); it('still lets --api-key override the connection file key', () => { const deps = makeDeps({ readConnectionFile }); - expect(resolveBrokerConnection({ stateDir: '/srv/node/state', apiKey: 'flag-key' }, deps)?.apiKey).toBe( + expect(resolveBrokerConnection({ stateDir: nodeStateDir, apiKey: 'flag-key' }, deps)?.apiKey).toBe( 'flag-key' ); }); + it('pairs --broker-url with the named state dir key, not the env key', () => { + const deps = makeDeps({ env: { RELAY_BROKER_API_KEY: 'other-key' }, readConnectionFile }); + expect(resolveBrokerConnection({ stateDir: nodeDir, brokerUrl: 'http://tunnel:9000' }, deps)).toEqual({ + url: 'http://tunnel:9000', + apiKey: 'node-key', + }); + }); + it('accepts a fleet node directory whose broker state lives in state/ (relay#1575)', () => { const deps = makeDeps({ readConnectionFile }); - expect(resolveBrokerConnection({ stateDir: '/srv/node' }, deps)?.url).toBe('http://node-host:4100'); + expect(resolveBrokerConnection({ stateDir: nodeDir }, deps)?.url).toBe('http://node-host:4100'); }); it('reads the nested state/connection.json from disk and prefers an exact match', () => { @@ -179,14 +190,17 @@ describe('malformed exact connection file', () => { describe('describeMissingBrokerConnection', () => { it('names the --state-dir paths that were searched', () => { - expect(describeMissingBrokerConnection({ stateDir: '/srv/node' }, makeDeps())).toBe( - 'Error: no broker connection at /srv/node/connection.json (from --state-dir; also checked /srv/node/state/connection.json). Pass the same --state-dir the broker was started with.' + const nodeDir = path.resolve('/srv/node'); + expect(describeMissingBrokerConnection({ stateDir: nodeDir }, makeDeps())).toBe( + `Error: no broker connection at ${path.join(nodeDir, 'connection.json')} (from --state-dir; also checked ${path.join(nodeDir, 'state', 'connection.json')}). Pass the same --state-dir the broker was started with.` ); }); it('labels the project default and points at --state-dir', () => { const message = describeMissingBrokerConnection({}, makeDeps()); - expect(message).toContain('/tmp/fake/.agentworkforce/relay/connection.json (project default)'); + expect(message).toContain( + `${path.join('/tmp/fake/.agentworkforce/relay', 'connection.json')} (project default)` + ); expect(message).toContain('If the broker was started with --state-dir, pass the same --state-dir here'); }); }); diff --git a/packages/cli/src/cli/lib/broker-connection.ts b/packages/cli/src/cli/lib/broker-connection.ts index 4b8a315f15..ef4c7a7400 100644 --- a/packages/cli/src/cli/lib/broker-connection.ts +++ b/packages/cli/src/cli/lib/broker-connection.ts @@ -123,10 +123,11 @@ export function resolveBrokerConnection( }); const explicitStateDir = trimOrUndefined(options.stateDir); - if (explicitStateDir && !explicitUrl) { + if (explicitStateDir) { // The caller named this broker's state directory: its connection file is - // the only source for both URL and key, so env vars pointing at another - // broker cannot pair that broker's URL or key with this one. + // the only source for the URL (unless --broker-url overrides it) and the + // key, so env vars pointing at another broker cannot pair that broker's + // URL or key with this one. // Select by file presence, not validity: a malformed exact file must fail // here rather than silently redirect to a nested broker. const stateDir = resolveConnectionStateDir( @@ -134,14 +135,13 @@ export function resolveBrokerConnection( (dir) => fs.existsSync(path.join(dir, 'connection.json')) || deps.readConnectionFile(dir) !== null ); const connectionFile = deps.readConnectionFile(stateDir); - const fileUrl = readString(connectionFile, 'url'); - if (!fileUrl) return null; - return finish(fileUrl, explicitKey ?? readString(connectionFile, 'api_key')); + const url = explicitUrl ?? readString(connectionFile, 'url'); + if (!url) return null; + return finish(url, explicitKey ?? readString(connectionFile, 'api_key')); } const envUrl = trimOrUndefined(deps.env.RELAY_BROKER_URL); - const stateDir = explicitStateDir ? path.resolve(explicitStateDir) : deps.getDefaultStateDir(); - const connectionFile = deps.readConnectionFile(stateDir); + const connectionFile = deps.readConnectionFile(deps.getDefaultStateDir()); const fileUrl = readString(connectionFile, 'url'); const url = explicitUrl ?? envUrl ?? fileUrl; diff --git a/packages/cli/src/cli/lib/broker-process-identity.ts b/packages/cli/src/cli/lib/broker-process-identity.ts index dd6ec22ca1..5cde495423 100644 --- a/packages/cli/src/cli/lib/broker-process-identity.ts +++ b/packages/cli/src/cli/lib/broker-process-identity.ts @@ -356,11 +356,16 @@ export function removeBrokerIdentity( identity: BrokerProcessIdentity, deps: IdentityDependencies ): void { - const current = findBrokerIdentityRecord(paths, deps, identity.brokerName); - if (!current || JSON.stringify(current.identity) !== JSON.stringify(identity)) return; - try { - deps.fs.unlinkSync(current.filename); - } catch { - /* Best effort. */ + // Remove every location holding this exact record so a legacy copy cannot + // outlive the process it describes. + for (const directory of identityDirectories(paths, deps)) { + const filename = identityFilename(directory, paths, deps, identity.brokerName); + const current = readBrokerIdentityFile(filename, paths, deps, identity.brokerName); + if (!current || JSON.stringify(current) !== JSON.stringify(identity)) continue; + try { + deps.fs.unlinkSync(filename); + } catch { + /* Best effort. */ + } } } diff --git a/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs b/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs index 120f344030..4321e2214c 100755 --- a/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs +++ b/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs @@ -5,16 +5,17 @@ * subcommands must reach a broker started with `--state-dir` from an unrelated * working directory. * - * The exact-head harness injects a Vitest probe into each exact target checkout. - * The probe drives the production Commander registration and the real - * HarnessDriverClient against a loopback fixture broker whose connection.json - * lives in a fleet-node style `/state/` directory. A decoy broker named by - * RELAY_BROKER_URL must never be contacted when --state-dir is explicit. + * Each exact target checkout is built, and its real `agent-relay` binary is run + * from an unrelated directory against a loopback fixture broker whose + * connection.json lives in a fleet-node style `/state/` directory. A + * decoy broker named by RELAY_BROKER_URL must never be contacted when + * --state-dir is explicit. */ -import { execFileSync, spawnSync } from 'node:child_process'; -import { randomUUID } from 'node:crypto'; -import { lstat, mkdir, open, readFile, rename, rm, writeFile } from 'node:fs/promises'; +import { execFileSync, spawn, spawnSync } from 'node:child_process'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import http from 'node:http'; +import os from 'node:os'; import path from 'node:path'; import process from 'node:process'; import { fileURLToPath } from 'node:url'; @@ -46,131 +47,57 @@ if (!isWithin(harnessDir, runnerPath)) { throw new Error('The RelayFlow runner must execute from the exact-head harness checkout.'); } -const probePath = path.join(targetDir, 'packages/cli/src/cli/.relayflow-1446-state-dir.test.ts'); -const configPath = path.join(targetDir, '.relayflow-1446-state-dir.vitest.config.mjs'); -const observationPath = path.join(targetDir, '.relayflow-1446-state-dir-observation.json'); - -const probeSource = String.raw`import fs from 'node:fs'; -import { writeFile } from 'node:fs/promises'; -import http from 'node:http'; -import os from 'node:os'; -import path from 'node:path'; - -import { Command } from 'commander'; -import { test } from 'vitest'; - -import { registerLocalAgentCommands } from './commands/local-agent.js'; - -const observationPath = process.env.RELAY_PR1446_OBSERVATION_PATH; - -function startBroker(label, requests) { - const server = http.createServer((req, res) => { - requests.push({ broker: label, method: req.method, url: req.url, apiKey: req.headers['x-api-key'] ?? null }); - res.setHeader('content-type', 'application/json'); - if (req.method === 'GET' && req.url === '/api/spawned') { - res.end(JSON.stringify({ agents: [{ name: 'worker' }] })); - return; - } - if (req.method === 'DELETE' && req.url === '/api/spawned/worker') { - res.end(JSON.stringify({ name: 'worker' })); - return; - } - res.statusCode = 404; - res.end(JSON.stringify({ error: 'not found' })); - }); - return new Promise((resolve) => { - server.listen(0, '127.0.0.1', () => resolve(server)); - }); -} +const cliEntry = path.join(targetDir, 'packages/cli/dist/cli/index.js'); -async function invoke(argv, cwd, env) { - const logs = []; - const errors = []; - const program = new Command(); - program.exitOverride(); - registerLocalAgentCommands(program.command('node'), { - cwd: () => cwd, - env, - log: (...args) => logs.push(args.join(' ')), - error: (...args) => errors.push(args.join(' ')), - exit: (code) => { - throw new Error('cli exit ' + code); - }, - }); - let error = null; - try { - await program.parseAsync(argv, { from: 'user' }); - } catch (caught) { - error = caught instanceof Error ? caught.message : String(caught); - } - return { logs, errors, error }; -} +run('npm', ['ci', '--ignore-scripts'], targetDir, 'workspace dependency installation'); +run('npm', ['run', 'build:core'], targetDir, 'workspace package build'); -test('observe node agent --state-dir broker selection', async () => { - const requests = []; - const broker = await startBroker('node', requests); - const decoy = await startBroker('decoy', requests); - const root = fs.mkdtempSync(path.join(os.tmpdir(), 'relayflow-1446-')); +const requests = []; +const broker = await startBroker('node', requests); +const decoy = await startBroker('decoy', requests); +const root = await mkdtemp(path.join(os.tmpdir(), 'relayflow-1446-')); +try { const nodeDir = path.join(root, 'sf-mini-node'); const stateDir = path.join(nodeDir, 'state'); const unrelatedCwd = path.join(root, 'elsewhere'); - fs.mkdirSync(stateDir, { recursive: true }); - fs.mkdirSync(path.join(unrelatedCwd, '.git'), { recursive: true }); - fs.writeFileSync( + const home = path.join(root, 'home'); + await mkdir(stateDir, { recursive: true }); + await mkdir(path.join(unrelatedCwd, '.git'), { recursive: true }); + await mkdir(home, { recursive: true }); + await writeFile( path.join(stateDir, 'connection.json'), - JSON.stringify({ url: 'http://127.0.0.1:' + broker.address().port, api_key: 'proof-node-key', pid: process.pid }) + JSON.stringify({ + url: `http://127.0.0.1:${broker.address().port}`, + api_key: 'proof-node-key', + pid: process.pid, + }) ); + // The real CLI binary, from an unrelated checkout, with ambient env naming a decoy broker. const env = { - RELAY_BROKER_URL: 'http://127.0.0.1:' + decoy.address().port, + PATH: process.env.PATH ?? '', + HOME: home, + AGENT_RELAY_TELEMETRY_DISABLED: '1', + DO_NOT_TRACK: '1', + RELAY_BROKER_URL: `http://127.0.0.1:${decoy.address().port}`, RELAY_BROKER_API_KEY: 'decoy-key', }; - try { - const list = await invoke(['node', 'agent', 'list', '--state-dir', stateDir], unrelatedCwd, env); - const release = await invoke(['node', 'agent', 'release', 'worker', '--state-dir', nodeDir], unrelatedCwd, env); - await writeFile(observationPath, JSON.stringify({ list, release, requests }), 'utf8'); - } finally { - broker.close(); - decoy.close(); - fs.rmSync(root, { recursive: true, force: true }); - } -}); -`; - -const configSource = `export default { - test: { - environment: 'node', - include: ['packages/cli/src/cli/.relayflow-1446-state-dir.test.ts'], - setupFiles: [], - }, -};\n`; - -try { - run('npm', ['ci', '--ignore-scripts'], targetDir, 'workspace dependency installation'); - run('npm', ['run', 'build:core'], targetDir, 'workspace package build'); - await writeGeneratedFile(probePath, probeSource); - await writeGeneratedFile(configPath, configSource); - run( - 'npm', - ['exec', '--', 'vitest', 'run', '--config', path.relative(targetDir, configPath)], - targetDir, - 'state-dir broker selection probe', - { RELAY_PR1446_OBSERVATION_PATH: observationPath } + const list = await runCli(['node', 'agent', 'list', '--state-dir', stateDir], unrelatedCwd, env); + const release = await runCli( + ['node', 'agent', 'release', 'worker', '--state-dir', nodeDir], + unrelatedCwd, + env ); - - const observation = JSON.parse(await readFile(observationPath, 'utf8')); + const observation = { list, release, requests }; console.log('State-dir broker selection observation:', JSON.stringify(observation)); - const { list, release, requests } = observation; - const rejected = (result) => - typeof result?.error === 'string' && /unknown option '--state-dir'/.test(result.error); + + const rejected = (result) => result.code !== 0 && /unknown option '--state-dir'/.test(result.stderr); const baseObserved = rejected(list) && rejected(release) && requests.length === 0; const nodeRequests = requests.filter((request) => request.broker === 'node'); const headObserved = - list?.error === null && - release?.error === null && - list.errors.length === 0 && - release.errors.length === 0 && - list.logs.some((line) => line.includes('"worker"')) && - release.logs.includes('Released worker.') && + list.code === 0 && + release.code === 0 && + list.stdout.includes('"worker"') && + release.stdout.includes('Released worker.') && requests.every((request) => request.broker === 'node' && request.apiKey === 'proof-node-key') && nodeRequests.some((request) => request.method === 'GET' && request.url === '/api/spawned') && nodeRequests.some((request) => request.method === 'DELETE' && request.url === '/api/spawned/worker'); @@ -182,12 +109,12 @@ try { outcome = 'bug'; signature = 'node_agent_state_dir_rejected'; details = - "The exact base CLI rejects --state-dir on node agent list and release with unknown option '--state-dir', so a broker started with --state-dir cannot be managed from another directory."; + "The exact base agent-relay binary rejects --state-dir on node agent list and release with unknown option '--state-dir', so a broker started with --state-dir cannot be managed from another directory."; } else if (headObserved) { outcome = 'fixed'; signature = 'node_agent_state_dir_targets_broker'; details = - 'The exact head CLI lists and releases through the broker named by --state-dir (both the state dir and its fleet node parent) from an unrelated cwd, authenticating with that broker key and never contacting the RELAY_BROKER_URL decoy.'; + 'The exact head agent-relay binary lists and releases through the broker named by --state-dir (both the state dir and its fleet node parent) from an unrelated cwd, authenticating with that broker key and never contacting the RELAY_BROKER_URL decoy.'; } else { throw new Error(`Unexpected state-dir broker selection observation: ${JSON.stringify(observation)}.`); } @@ -199,9 +126,49 @@ try { 'utf8' ); } finally { - await rm(probePath, { force: true }); - await rm(configPath, { force: true }); - await rm(observationPath, { force: true }); + broker.close(); + decoy.close(); + await rm(root, { recursive: true, force: true }); +} + +/** Loopback fixture broker answering the two endpoints list and release use. */ +function startBroker(label, log) { + const server = http.createServer((req, res) => { + log.push({ broker: label, method: req.method, url: req.url, apiKey: req.headers['x-api-key'] ?? null }); + res.setHeader('content-type', 'application/json'); + if (req.method === 'GET' && req.url === '/api/spawned') { + res.end(JSON.stringify({ agents: [{ name: 'worker' }] })); + return; + } + if (req.method === 'DELETE' && req.url === '/api/spawned/worker') { + res.end(JSON.stringify({ name: 'worker' })); + return; + } + res.statusCode = 404; + res.end(JSON.stringify({ error: 'not found' })); + }); + return new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve(server))); +} + +/** Run the built CLI asynchronously so the in-process fixture brokers can answer it. */ +function runCli(args, cwd, env) { + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, [cliEntry, ...args], { + cwd, + env, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let stdout = ''; + let stderr = ''; + child.stdout.on('data', (chunk) => (stdout += chunk)); + child.stderr.on('data', (chunk) => (stderr += chunk)); + const timer = setTimeout(() => child.kill('SIGKILL'), COMMAND_TIMEOUT_MS); + child.on('error', reject); + child.on('close', (code) => { + clearTimeout(timer); + resolve({ code, stdout: stdout.slice(-4000), stderr: stderr.slice(-4000) }); + }); + }); } function requiredValue(name) { @@ -222,25 +189,6 @@ function isWithin(directory, candidate) { ); } -async function writeGeneratedFile(targetPath, source) { - try { - const existing = await lstat(targetPath); - if (!existing.isFile()) throw new Error(`Refusing to replace non-file ${targetPath}.`); - } catch (error) { - if (error?.code !== 'ENOENT') throw error; - } - - const temporaryPath = `${targetPath}.tmp-${process.pid}-${randomUUID()}`; - try { - const handle = await open(temporaryPath, 'wx', 0o600); - await handle.writeFile(source, 'utf8'); - await handle.close(); - await rename(temporaryPath, targetPath); - } finally { - await rm(temporaryPath, { force: true }); - } -} - function run(command, args, cwd, label, extraEnv = {}) { const result = spawnSync(command, args, { cwd, From 0bad7940bf062bf3d69b91f070936cb8007c3b15 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 06:29:51 +0000 Subject: [PATCH 05/14] fix(cli): precise state-dir diagnostics and nested-broker precedence - Missing-connection errors name only the file resolution actually read (an existing malformed exact file is reported as unusable). - node down/status prefer a live connection.json (exact, then nested state/) before lock/identity evidence, so stale locks cannot shadow a running nested broker. - Stale-identity recovery messages name the record file actually found. - RelayFlow proof bounds each CLI invocation to 15s. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/commands/core.test.ts | 16 +++++++++++++ .../cli/src/cli/commands/local-agent.test.ts | 2 +- .../cli/src/cli/lib/broker-connection.test.ts | 16 +++++++++++-- packages/cli/src/cli/lib/broker-connection.ts | 13 +++++++++-- packages/cli/src/cli/lib/broker-lifecycle.ts | 23 ++++++++++++------- .../src/cli/lib/broker-process-identity.ts | 11 +++++++++ .../1446-state-dir-broker-selection/run.mjs | 4 +++- 7 files changed, 71 insertions(+), 14 deletions(-) diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index e27b191f47..ee1f6e6e26 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -2362,6 +2362,22 @@ describe('registerCoreCommands', () => { ); }); + it('status --state-dir prefers a live nested broker over stale lock files in the node directory', async () => { + const nodeDir = '/srv/stale-node'; + const fs = createFsMock({ + [`${nodeDir}/broker-old.lock`]: '', + [`${nodeDir}/state/connection.json`]: connectionFile(4242), + }); + sdkStatusClient.getStatus.mockResolvedValueOnce({ agent_count: 1, pending_delivery_count: 0 }); + sdkStatusClient.getSession.mockResolvedValueOnce({ workspace_key: 'rk_live_teststatus123' }); + const { program, deps } = createHarness({ fs }); + + expect(await runCommand(program, ['status', '--state-dir', nodeDir])).toBeUndefined(); + + expect(deps.log).toHaveBeenCalledWith('Status: RUNNING'); + expect(deps.log).toHaveBeenCalledWith('PID: 4242'); + }); + it('status checks broker status and prints metrics', async () => { const connectionPath = '/tmp/project/.agentworkforce/relay/connection.json'; const fs = createFsMock({ diff --git a/packages/cli/src/cli/commands/local-agent.test.ts b/packages/cli/src/cli/commands/local-agent.test.ts index 943ce28abc..5faf20bbc5 100644 --- a/packages/cli/src/cli/commands/local-agent.test.ts +++ b/packages/cli/src/cli/commands/local-agent.test.ts @@ -1305,7 +1305,7 @@ describe('explicit broker selection on every node agent subcommand (relay#1446, await program.parseAsync(['local', 'agent', 'list', '--state-dir', root], { from: 'user' }); expect(exit).toHaveBeenCalledWith(1); expect(error).toHaveBeenCalledWith( - `Error: no broker connection at ${path.join(root, 'connection.json')} (from --state-dir; also checked ${path.join(root, 'state', 'connection.json')}). Pass the same --state-dir the broker was started with.` + `Error: no broker connection at ${path.join(root, 'connection.json')} or ${path.join(root, 'state', 'connection.json')} (from --state-dir). Pass the same --state-dir the broker was started with.` ); } finally { fs.rmSync(root, { recursive: true, force: true }); diff --git a/packages/cli/src/cli/lib/broker-connection.test.ts b/packages/cli/src/cli/lib/broker-connection.test.ts index 7657c9384d..49d897215a 100644 --- a/packages/cli/src/cli/lib/broker-connection.test.ts +++ b/packages/cli/src/cli/lib/broker-connection.test.ts @@ -190,12 +190,24 @@ describe('malformed exact connection file', () => { describe('describeMissingBrokerConnection', () => { it('names the --state-dir paths that were searched', () => { - const nodeDir = path.resolve('/srv/node'); + const nodeDir = path.resolve('/srv/node-that-does-not-exist'); expect(describeMissingBrokerConnection({ stateDir: nodeDir }, makeDeps())).toBe( - `Error: no broker connection at ${path.join(nodeDir, 'connection.json')} (from --state-dir; also checked ${path.join(nodeDir, 'state', 'connection.json')}). Pass the same --state-dir the broker was started with.` + `Error: no broker connection at ${path.join(nodeDir, 'connection.json')} or ${path.join(nodeDir, 'state', 'connection.json')} (from --state-dir). Pass the same --state-dir the broker was started with.` ); }); + it('names the unusable exact file instead of claiming the nested one was read', () => { + const nodeDir = fs.mkdtempSync(path.join(os.tmpdir(), 'relay-node-dir-')); + try { + fs.writeFileSync(path.join(nodeDir, 'connection.json'), '{not json'); + expect(describeMissingBrokerConnection({ stateDir: nodeDir }, makeDeps())).toBe( + `Error: unusable broker connection file ${path.join(nodeDir, 'connection.json')} (from --state-dir): it is malformed or has no url.` + ); + } finally { + fs.rmSync(nodeDir, { recursive: true, force: true }); + } + }); + it('labels the project default and points at --state-dir', () => { const message = describeMissingBrokerConnection({}, makeDeps()); expect(message).toContain( diff --git a/packages/cli/src/cli/lib/broker-connection.ts b/packages/cli/src/cli/lib/broker-connection.ts index ef4c7a7400..839106d236 100644 --- a/packages/cli/src/cli/lib/broker-connection.ts +++ b/packages/cli/src/cli/lib/broker-connection.ts @@ -163,9 +163,18 @@ export function describeMissingBrokerConnection( const explicitStateDir = trimOrUndefined(options.stateDir); if (explicitStateDir) { const resolved = path.resolve(explicitStateDir); + const exact = path.join(resolved, 'connection.json'); + // Report only what resolution actually read: an existing exact file stops + // the nested fallback, so it is the file at fault. + if (fs.existsSync(exact)) { + return `Error: unusable broker connection file ${exact} (from --state-dir): it is malformed or has no url.`; + } + const nested = path.join(resolved, 'state', 'connection.json'); + if (fs.existsSync(nested)) { + return `Error: unusable broker connection file ${nested} (from --state-dir): it is malformed or has no url.`; + } return ( - `Error: no broker connection at ${path.join(resolved, 'connection.json')} (from --state-dir; ` + - `also checked ${path.join(resolved, 'state', 'connection.json')}). ` + + `Error: no broker connection at ${exact} or ${nested} (from --state-dir). ` + 'Pass the same --state-dir the broker was started with.' ); } diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 9c6ccc55a6..0d568afb53 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -10,7 +10,7 @@ import { redactCredentialValues } from '@agent-relay/cloud/redact'; import type { CoreDependencies, CoreProjectPaths, CoreRelay, SpawnedProcess } from '../commands/core.js'; import { brokerIdentityDirectory, - brokerIdentityPath, + locateBrokerIdentityFile, matchesBrokerIdentity, persistBrokerIdentity, readBrokerIdentities, @@ -1095,7 +1095,7 @@ async function killOrphanedBrokerProcesses( if (options?.matchBrokerName) { result.matchedCount++; deps.warn( - `Recorded broker has already exited; its identity was retained because no matched exit was observed. Verify ownership before removing ${brokerIdentityPath(paths, deps, identity.brokerName)} and restarting.` + `Recorded broker has already exited; its identity was retained because no matched exit was observed. Verify ownership before removing ${locateBrokerIdentityFile(paths, deps, identity.brokerName)} and restarting.` ); } continue; @@ -1116,7 +1116,7 @@ async function stopRecordedBroker( // escalation after a wait. Never rediscover ownership from rendered argv. if (!(await matchesBrokerIdentity(identity, paths, deps))) { deps.warn( - `Broker identity could not be verified (pid: ${identity.pid}). State retained; verify process ownership before stopping it manually. If the recorded broker has exited, remove its stale identity file: ${brokerIdentityPath(paths, deps, identity.brokerName)}` + `Broker identity could not be verified (pid: ${identity.pid}). State retained; verify process ownership before stopping it manually. If the recorded broker has exited, remove its stale identity file: ${locateBrokerIdentityFile(paths, deps, identity.brokerName)}` ); return false; } @@ -2516,15 +2516,22 @@ async function reportNodeClaimsElsewhere(paths: CoreProjectPaths, deps: CoreDepe * Resolve `--state-dir` for commands that inspect an existing broker. A fleet * node directory holds its broker state in `state/`, so accept the node * directory too when only the nested one holds broker state (relay#1575). - * Lock and identity files count as evidence alongside connection.json so - * `down --force` can still recover a broker whose connection file is gone. + * A live connection file is the strongest evidence and is checked in both + * places first, so leftover lock or identity files in the exact directory + * cannot shadow a running nested broker. Lock and identity files still count + * when neither has a connection file, so `down --force` can recover a broker + * whose connection file is gone. */ function resolveExistingStateDir(stateDir: string, deps: CoreDependencies): string { - return resolveConnectionStateDir(stateDir, (dir) => hasBrokerStateEvidence(dir, deps)); + const exact = path.resolve(stateDir); + const nested = path.join(exact, 'state'); + const hasConnection = (dir: string) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)); + if (hasConnection(exact)) return exact; + if (hasConnection(nested)) return nested; + return resolveConnectionStateDir(exact, (dir) => hasBrokerStateFiles(dir, deps)); } -function hasBrokerStateEvidence(dir: string, deps: CoreDependencies): boolean { - if (deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME))) return true; +function hasBrokerStateFiles(dir: string, deps: CoreDependencies): boolean { try { return deps.fs .readdirSync(dir) diff --git a/packages/cli/src/cli/lib/broker-process-identity.ts b/packages/cli/src/cli/lib/broker-process-identity.ts index 5cde495423..fec7fd14fd 100644 --- a/packages/cli/src/cli/lib/broker-process-identity.ts +++ b/packages/cli/src/cli/lib/broker-process-identity.ts @@ -129,6 +129,17 @@ function findBrokerIdentityRecord( return null; } +/** The file holding the record for `brokerName`, falling back to where a new record would be written. */ +export function locateBrokerIdentityFile( + paths: CoreProjectPaths, + deps: IdentityDependencies, + brokerName: string +): string { + return ( + findBrokerIdentityRecord(paths, deps, brokerName)?.filename ?? brokerIdentityPath(paths, deps, brokerName) + ); +} + export function readBrokerIdentity( paths: CoreProjectPaths, deps: IdentityDependencies, diff --git a/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs b/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs index 4321e2214c..e68b891b0c 100755 --- a/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs +++ b/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs @@ -22,6 +22,7 @@ import { fileURLToPath } from 'node:url'; const CASE_ID = '1446-state-dir-broker-selection'; const COMMAND_TIMEOUT_MS = 5 * 60 * 1000; +const CLI_OPERATION_TIMEOUT_MS = 15 * 1000; const targetDir = requiredDirectory('RELAY_PR_PROOF_TARGET_DIR'); const harnessDir = requiredDirectory('RELAY_PR_PROOF_HARNESS_DIR'); @@ -162,7 +163,8 @@ function runCli(args, cwd, env) { let stderr = ''; child.stdout.on('data', (chunk) => (stdout += chunk)); child.stderr.on('data', (chunk) => (stderr += chunk)); - const timer = setTimeout(() => child.kill('SIGKILL'), COMMAND_TIMEOUT_MS); + // Bounded like any normal RelayFlow operation; the fixture answers immediately. + const timer = setTimeout(() => child.kill('SIGKILL'), CLI_OPERATION_TIMEOUT_MS); child.on('error', reject); child.on('close', (code) => { clearTimeout(timer); From 6346e34989083022d3f89d1c98b049b25f47b6a7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 06:36:38 +0000 Subject: [PATCH 06/14] test(cli): isolate missing-connection message test in a temp dir Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/lib/broker-connection.test.ts | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/packages/cli/src/cli/lib/broker-connection.test.ts b/packages/cli/src/cli/lib/broker-connection.test.ts index 49d897215a..e14bd3fcfd 100644 --- a/packages/cli/src/cli/lib/broker-connection.test.ts +++ b/packages/cli/src/cli/lib/broker-connection.test.ts @@ -190,10 +190,15 @@ describe('malformed exact connection file', () => { describe('describeMissingBrokerConnection', () => { it('names the --state-dir paths that were searched', () => { - const nodeDir = path.resolve('/srv/node-that-does-not-exist'); - expect(describeMissingBrokerConnection({ stateDir: nodeDir }, makeDeps())).toBe( - `Error: no broker connection at ${path.join(nodeDir, 'connection.json')} or ${path.join(nodeDir, 'state', 'connection.json')} (from --state-dir). Pass the same --state-dir the broker was started with.` - ); + // An isolated empty directory: the message depends on what exists on disk. + const nodeDir = fs.mkdtempSync(path.join(os.tmpdir(), 'relay-empty-node-')); + try { + expect(describeMissingBrokerConnection({ stateDir: nodeDir }, makeDeps())).toBe( + `Error: no broker connection at ${path.join(nodeDir, 'connection.json')} or ${path.join(nodeDir, 'state', 'connection.json')} (from --state-dir). Pass the same --state-dir the broker was started with.` + ); + } finally { + fs.rmSync(nodeDir, { recursive: true, force: true }); + } }); it('names the unusable exact file instead of claiming the nested one was read', () => { From 84a29f9d6164443b5eba3be64f41715be625d6f5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 07:10:57 +0000 Subject: [PATCH 07/14] fix(verify): restore the shared install after a RelayFlow corpus case changes it Corpus cases share one checkout. The 1797 case installs with a cloud-workspace-only npm ci, pruning root dependencies such as commander, so later cases like 1801 failed with ERR_MODULE_NOT_FOUND whenever the full corpus ran (any PR that adds a case). The case harness now fingerprints node_modules/.package-lock.json and runs a full npm ci when a case changed the installed tree. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- .../targeted-relayflow-case.mjs | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/scripts/verify-features/targeted-relayflow-case.mjs b/scripts/verify-features/targeted-relayflow-case.mjs index ed195e10be..7cdfa571b8 100644 --- a/scripts/verify-features/targeted-relayflow-case.mjs +++ b/scripts/verify-features/targeted-relayflow-case.mjs @@ -35,6 +35,39 @@ async function gitHead(repoRoot) { return result.stdout.trim(); } +const INSTALL_FINGERPRINT = 'node_modules/.package-lock.json'; +const RESTORE_TIMEOUT_MS = 15 * 60 * 1_000; + +/** npm rewrites its hidden lockfile whenever an install changes the tree. */ +async function installFingerprint(repoRoot) { + try { + return await readFile(path.join(repoRoot, INSTALL_FINGERPRINT), 'utf8'); + } catch { + return null; + } +} + +/** + * Corpus cases share one checkout. A case that reinstalls a subset of the + * workspace (for example `npm ci --workspace …`) prunes dependencies every + * later case needs, so restore the full install whenever a case changed it. + */ +async function restoreInstallIfChanged(repoRoot, before, caseId) { + if ((await installFingerprint(repoRoot)) === before) return; + console.log(`TARGETED_RELAYFLOW_CASE_RESTORE case=${caseId} reason=dependency_tree_changed`); + const result = await runTargetedProcess(['npm', 'ci', '--ignore-scripts', '--no-audit', '--no-fund'], { + cwd: repoRoot, + env: process.env, + timeoutMs: RESTORE_TIMEOUT_MS, + maxOutputBytes: MAX_OUTPUT_BYTES, + }); + if (result.timedOut || result.aborted || result.outputLimitExceeded || result.exitCode !== 0) { + throw new Error( + `Restoring dependencies after RelayFlow case ${caseId} failed: ${(result.stderr ?? '').trim()}` + ); + } +} + async function main() { const caseId = requiredOption('--case'); const timeoutSeconds = Number(requiredOption('--timeout-seconds')); @@ -60,6 +93,7 @@ async function main() { const temporaryDirectory = await mkdtemp(path.join(os.tmpdir(), `relay-targeted-${caseId}-`)); const resultPath = path.join(temporaryDirectory, 'observation.json'); + const installBefore = await installFingerprint(repoRoot); try { const result = await runTargetedProcess(manifest.runner.command, { cwd: repoRoot, @@ -98,6 +132,7 @@ async function main() { ); } finally { await rm(temporaryDirectory, { recursive: true, force: true }); + await restoreInstallIfChanged(repoRoot, installBefore, caseId); } } From 782921dc5478d7c1304964b80caaebd451a23973 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 07:16:46 +0000 Subject: [PATCH 08/14] fix(verify): never let a failed install restore mask a corpus case failure Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- scripts/verify-features/targeted-relayflow-case.mjs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/scripts/verify-features/targeted-relayflow-case.mjs b/scripts/verify-features/targeted-relayflow-case.mjs index 7cdfa571b8..8e11c0b5a9 100644 --- a/scripts/verify-features/targeted-relayflow-case.mjs +++ b/scripts/verify-features/targeted-relayflow-case.mjs @@ -94,6 +94,7 @@ async function main() { const temporaryDirectory = await mkdtemp(path.join(os.tmpdir(), `relay-targeted-${caseId}-`)); const resultPath = path.join(temporaryDirectory, 'observation.json'); const installBefore = await installFingerprint(repoRoot); + let caseFailed = false; try { const result = await runTargetedProcess(manifest.runner.command, { cwd: repoRoot, @@ -130,9 +131,18 @@ async function main() { console.log( `TARGETED_RELAYFLOW_CASE_PASS case=${caseId} signature=${observation.signature} sha=${headSha}` ); + } catch (error) { + caseFailed = true; + throw error; } finally { await rm(temporaryDirectory, { recursive: true, force: true }); - await restoreInstallIfChanged(repoRoot, installBefore, caseId); + try { + await restoreInstallIfChanged(repoRoot, installBefore, caseId); + } catch (restoreError) { + // The case's own failure is the proof result; never let cleanup mask it. + if (!caseFailed) throw restoreError; + console.error(restoreError instanceof Error ? restoreError.message : String(restoreError)); + } } } From 19a8eeac4592366807e8d8f5c039f5e80d956108 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 07:20:36 +0000 Subject: [PATCH 09/14] fix(verify): restore corpus installs non-destructively within the command budget Use npm install --no-save --prefer-offline (no node_modules wipe, no package-lock rewrite) bounded by the time the parent plan command still allows, so a restore cut short can never leave a deleted tree behind. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- .../targeted-relayflow-case.mjs | 43 +++++++++++++++---- 1 file changed, 34 insertions(+), 9 deletions(-) diff --git a/scripts/verify-features/targeted-relayflow-case.mjs b/scripts/verify-features/targeted-relayflow-case.mjs index 8e11c0b5a9..d4d46144f8 100644 --- a/scripts/verify-features/targeted-relayflow-case.mjs +++ b/scripts/verify-features/targeted-relayflow-case.mjs @@ -36,7 +36,13 @@ async function gitHead(repoRoot) { } const INSTALL_FINGERPRINT = 'node_modules/.package-lock.json'; -const RESTORE_TIMEOUT_MS = 15 * 60 * 1_000; +// The targeted plan gives each corpus command its timeout plus 30s of slack, +// capped at 720s (targeted-pr-plan.mjs). The restore must finish inside that. +const PLAN_MAX_COMMAND_SECONDS = 720; +const PLAN_COMMAND_SLACK_SECONDS = 30; +const RESTORE_SAFETY_MARGIN_MS = 10_000; +const MIN_RESTORE_MS = 15_000; +const startedAt = Date.now(); /** npm rewrites its hidden lockfile whenever an install changes the tree. */ async function installFingerprint(repoRoot) { @@ -51,16 +57,35 @@ async function installFingerprint(repoRoot) { * Corpus cases share one checkout. A case that reinstalls a subset of the * workspace (for example `npm ci --workspace …`) prunes dependencies every * later case needs, so restore the full install whenever a case changed it. + * + * `npm install --no-save` reinstalls the lockfile's tree from the warm cache + * without first deleting node_modules or rewriting package-lock.json, so a + * restore cut short by the command budget leaves the tree no worse than the + * case left it. It runs with `--ignore-scripts`, matching how the job itself + * installs, and only within the time the parent command still allows. */ -async function restoreInstallIfChanged(repoRoot, before, caseId) { +async function restoreInstallIfChanged(repoRoot, before, caseId, timeoutSeconds) { if ((await installFingerprint(repoRoot)) === before) return; + const commandBudgetMs = + (Math.min(timeoutSeconds + PLAN_COMMAND_SLACK_SECONDS, PLAN_MAX_COMMAND_SECONDS) + + PLAN_COMMAND_SLACK_SECONDS) * + 1_000; + const remainingMs = commandBudgetMs - (Date.now() - startedAt) - RESTORE_SAFETY_MARGIN_MS; + if (remainingMs < MIN_RESTORE_MS) { + throw new Error( + `RelayFlow case ${caseId} changed the shared dependency tree and no command budget remains to restore it` + ); + } console.log(`TARGETED_RELAYFLOW_CASE_RESTORE case=${caseId} reason=dependency_tree_changed`); - const result = await runTargetedProcess(['npm', 'ci', '--ignore-scripts', '--no-audit', '--no-fund'], { - cwd: repoRoot, - env: process.env, - timeoutMs: RESTORE_TIMEOUT_MS, - maxOutputBytes: MAX_OUTPUT_BYTES, - }); + const result = await runTargetedProcess( + ['npm', 'install', '--no-save', '--ignore-scripts', '--prefer-offline', '--no-audit', '--no-fund'], + { + cwd: repoRoot, + env: process.env, + timeoutMs: remainingMs, + maxOutputBytes: MAX_OUTPUT_BYTES, + } + ); if (result.timedOut || result.aborted || result.outputLimitExceeded || result.exitCode !== 0) { throw new Error( `Restoring dependencies after RelayFlow case ${caseId} failed: ${(result.stderr ?? '').trim()}` @@ -137,7 +162,7 @@ async function main() { } finally { await rm(temporaryDirectory, { recursive: true, force: true }); try { - await restoreInstallIfChanged(repoRoot, installBefore, caseId); + await restoreInstallIfChanged(repoRoot, installBefore, caseId, timeoutSeconds); } catch (restoreError) { // The case's own failure is the proof result; never let cleanup mask it. if (!caseFailed) throw restoreError; From e3715dd507ae7d9a2c6831f4c103fda371b40475 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 07:27:11 +0000 Subject: [PATCH 10/14] fix(verify): bound the corpus restore by the command's real timeout targeted-command.mjs kills a corpus command at min(timeout + 30, 720)s; the restore budget no longer adds a second 30s of slack on top. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- scripts/verify-features/targeted-relayflow-case.mjs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/scripts/verify-features/targeted-relayflow-case.mjs b/scripts/verify-features/targeted-relayflow-case.mjs index d4d46144f8..68e93d7713 100644 --- a/scripts/verify-features/targeted-relayflow-case.mjs +++ b/scripts/verify-features/targeted-relayflow-case.mjs @@ -37,7 +37,8 @@ async function gitHead(repoRoot) { const INSTALL_FINGERPRINT = 'node_modules/.package-lock.json'; // The targeted plan gives each corpus command its timeout plus 30s of slack, -// capped at 720s (targeted-pr-plan.mjs). The restore must finish inside that. +// capped at 720s (targeted-pr-plan.mjs), and targeted-command.mjs enforces +// exactly that. The restore must finish inside it. const PLAN_MAX_COMMAND_SECONDS = 720; const PLAN_COMMAND_SLACK_SECONDS = 30; const RESTORE_SAFETY_MARGIN_MS = 10_000; @@ -66,10 +67,9 @@ async function installFingerprint(repoRoot) { */ async function restoreInstallIfChanged(repoRoot, before, caseId, timeoutSeconds) { if ((await installFingerprint(repoRoot)) === before) return; + // targeted-command.mjs kills the command at the plan's bounded timeout. const commandBudgetMs = - (Math.min(timeoutSeconds + PLAN_COMMAND_SLACK_SECONDS, PLAN_MAX_COMMAND_SECONDS) + - PLAN_COMMAND_SLACK_SECONDS) * - 1_000; + Math.min(timeoutSeconds + PLAN_COMMAND_SLACK_SECONDS, PLAN_MAX_COMMAND_SECONDS) * 1_000; const remainingMs = commandBudgetMs - (Date.now() - startedAt) - RESTORE_SAFETY_MARGIN_MS; if (remainingMs < MIN_RESTORE_MS) { throw new Error( From 435af9a2b80c9424926137e8910bf45aa433d8a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 13:24:12 +0000 Subject: [PATCH 11/14] fix(cli): rank state-dir evidence so stale node-dir files cannot shadow a live broker node status/down now pick the exact dir or its state/ child by ranked evidence: connection.json, then an identity record naming a live pid, then any broker lock/record. A leftover lock in a fleet node directory no longer hides a running nested broker from down --force. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/commands/core.test.ts | 23 +++++++++ packages/cli/src/cli/lib/broker-lifecycle.ts | 53 ++++++++++++++------ 2 files changed, 61 insertions(+), 15 deletions(-) diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index ee1f6e6e26..6b4464d2a6 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -1699,6 +1699,29 @@ describe('registerCoreCommands', () => { expect(fs.existsSync(`${stateDir}/connection.json`)).toBe(false); }); + it('down --force ignores stale lock files in the node directory when the nested broker is live', async () => { + const nodeDir = '/srv/stale-lock-node'; + const stateDir = `${nodeDir}/state`; + const running = new Set([222]); + const execCommand = identityCommand(222, `${stateDir}/broker-nested.lock`); + const killImpl = vi.fn((pid: number, signal?: NodeJS.Signals | number) => { + if (signal === 0) { + if (!running.has(pid)) throw new Error('not running'); + return; + } + running.delete(pid); + }); + const fs = createFsMock({ [`${nodeDir}/broker-old.lock`]: '' }); + const { program, deps } = createHarness({ fs, execCommand, killImpl }); + await persistBrokerIdentity({ ...deps.getProjectPaths(), dataDir: stateDir }, 222, 'nested', deps); + expect(fs.existsSync(`${stateDir}/connection.json`)).toBe(false); + + await runCommand(program, ['down', '--force', '--state-dir', nodeDir]); + + expect(killImpl).toHaveBeenCalledWith(222, 'SIGTERM'); + expect([...running]).toEqual([]); + }); + it('down still verifies records written under the project by earlier releases', async () => { const stateDir = '/srv/legacy-node/state'; let running = true; diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 0d568afb53..3c249c395f 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -57,7 +57,6 @@ import { type NodeClaim, } from './node-claim.js'; import { maskSecret } from './redact.js'; -import { resolveConnectionStateDir } from './broker-connection.js'; import { startReflexCapture, type RunningReflexCapture } from './reflex-capture.js'; import { readProjectWorkspaceSession, @@ -2515,32 +2514,56 @@ async function reportNodeClaimsElsewhere(paths: CoreProjectPaths, deps: CoreDepe /** * Resolve `--state-dir` for commands that inspect an existing broker. A fleet * node directory holds its broker state in `state/`, so accept the node - * directory too when only the nested one holds broker state (relay#1575). - * A live connection file is the strongest evidence and is checked in both - * places first, so leftover lock or identity files in the exact directory - * cannot shadow a running nested broker. Lock and identity files still count - * when neither has a connection file, so `down --force` can recover a broker - * whose connection file is gone. + * directory too when the nested one holds the broker (relay#1575). + * + * Evidence is ranked, checking the exact directory before `state/` at each + * rank: a `connection.json`, then an identity record naming a live process, + * then any broker lock or record. Leftover files in the node directory + * therefore cannot shadow a running nested broker, and `down --force` can + * still recover a broker whose connection file is gone. */ function resolveExistingStateDir(stateDir: string, deps: CoreDependencies): string { const exact = path.resolve(stateDir); - const nested = path.join(exact, 'state'); - const hasConnection = (dir: string) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)); - if (hasConnection(exact)) return exact; - if (hasConnection(nested)) return nested; - return resolveConnectionStateDir(exact, (dir) => hasBrokerStateFiles(dir, deps)); + const candidates = [exact, path.join(exact, 'state')]; + const ranks: Array<(dir: string) => boolean> = [ + (dir) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)), + (dir) => hasLiveBrokerIdentity(dir, deps), + (dir) => hasBrokerStateFiles(dir, deps), + ]; + for (const matches of ranks) { + const found = candidates.find(matches); + if (found) return found; + } + return exact; } -function hasBrokerStateFiles(dir: string, deps: CoreDependencies): boolean { +function brokerStateFiles(dir: string, deps: CoreDependencies): string[] { try { return deps.fs .readdirSync(dir) - .some((file) => file.startsWith('broker-') && (file.endsWith('.lock') || file.endsWith('.json'))); + .filter((file) => file.startsWith('broker-') && (file.endsWith('.lock') || file.endsWith('.json'))); } catch { - return false; + return []; } } +function hasBrokerStateFiles(dir: string, deps: CoreDependencies): boolean { + return brokerStateFiles(dir, deps).length > 0; +} + +/** An identity record whose pid is still running marks the directory of a live broker. */ +function hasLiveBrokerIdentity(dir: string, deps: CoreDependencies): boolean { + return brokerStateFiles(dir, deps).some((file) => { + if (!file.startsWith('broker-identity-')) return false; + try { + const record = JSON.parse(deps.fs.readFileSync(path.join(dir, file), 'utf-8')) as { pid?: unknown }; + return typeof record.pid === 'number' && record.pid > 0 && isProcessRunning(record.pid, deps); + } catch { + return false; + } + }); +} + // eslint-disable-next-line complexity, max-depth export async function runDownCommand(options: DownOptions, deps: CoreDependencies): Promise { const paths = deps.getProjectPaths(); From 64bbddef4a85b24397effcaf63b36511fc7ced5a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 13:30:51 +0000 Subject: [PATCH 12/14] fix(cli): verify identity records before using them to pick a state dir Directory selection for node status/down now requires an identity record that still matches its process (start time, executable and held lock), so a stale record whose pid was reused cannot outrank a live nested broker. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/commands/core.test.ts | 33 ++++++++++++++++ packages/cli/src/cli/lib/broker-lifecycle.ts | 41 ++++++++++---------- 2 files changed, 54 insertions(+), 20 deletions(-) diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index 6b4464d2a6..0f357d186a 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -1722,6 +1722,39 @@ describe('registerCoreCommands', () => { expect([...running]).toEqual([]); }); + it('down --force ignores a node-dir identity whose pid was reused by another process', async () => { + const nodeDir = '/srv/reused-pid-node'; + const stateDir = `${nodeDir}/state`; + const running = new Set([222, 333]); + const nested = identityCommand(222, `${stateDir}/broker-nested.lock`); + const old = identityCommand(333, `${nodeDir}/broker-old.lock`); + let pidReused = false; + const execCommand = vi.fn(async (command: string) => { + if (!command.includes('-p 333 ')) return nested(command); + // After the record was written, pid 333 belongs to a different process. + if (pidReused && command.includes('lstart')) + return { stdout: 'Fri Oct 2 09:00:00 2026\n', stderr: '' }; + return old(command); + }); + const killImpl = vi.fn((pid: number, signal?: NodeJS.Signals | number) => { + if (signal === 0) { + if (!running.has(pid)) throw new Error('not running'); + return; + } + running.delete(pid); + }); + const { program, deps } = createHarness({ execCommand, killImpl }); + await persistBrokerIdentity({ ...deps.getProjectPaths(), dataDir: nodeDir }, 333, 'old', deps); + await persistBrokerIdentity({ ...deps.getProjectPaths(), dataDir: stateDir }, 222, 'nested', deps); + pidReused = true; + + await runCommand(program, ['down', '--force', '--state-dir', nodeDir]); + + expect(killImpl).toHaveBeenCalledWith(222, 'SIGTERM'); + expect(killImpl).not.toHaveBeenCalledWith(333, 'SIGTERM'); + expect([...running]).toEqual([333]); + }); + it('down still verifies records written under the project by earlier releases', async () => { const stateDir = '/srv/legacy-node/state'; let running = true; diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 3c249c395f..7da792729a 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -2517,22 +2517,24 @@ async function reportNodeClaimsElsewhere(paths: CoreProjectPaths, deps: CoreDepe * directory too when the nested one holds the broker (relay#1575). * * Evidence is ranked, checking the exact directory before `state/` at each - * rank: a `connection.json`, then an identity record naming a live process, - * then any broker lock or record. Leftover files in the node directory + * rank: a `connection.json`, then an identity record verified against its + * running process (start time, executable and held lock, as `down` requires + * before signalling), then any broker lock or record. Leftover files in the node directory * therefore cannot shadow a running nested broker, and `down --force` can * still recover a broker whose connection file is gone. */ -function resolveExistingStateDir(stateDir: string, deps: CoreDependencies): string { +async function resolveExistingStateDir(stateDir: string, deps: CoreDependencies): Promise { const exact = path.resolve(stateDir); const candidates = [exact, path.join(exact, 'state')]; - const ranks: Array<(dir: string) => boolean> = [ + const ranks: Array<(dir: string) => boolean | Promise> = [ (dir) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)), - (dir) => hasLiveBrokerIdentity(dir, deps), + (dir) => hasVerifiedBrokerIdentity(dir, deps), (dir) => hasBrokerStateFiles(dir, deps), ]; for (const matches of ranks) { - const found = candidates.find(matches); - if (found) return found; + for (const candidate of candidates) { + if (await matches(candidate)) return candidate; + } } return exact; } @@ -2551,24 +2553,23 @@ function hasBrokerStateFiles(dir: string, deps: CoreDependencies): boolean { return brokerStateFiles(dir, deps).length > 0; } -/** An identity record whose pid is still running marks the directory of a live broker. */ -function hasLiveBrokerIdentity(dir: string, deps: CoreDependencies): boolean { - return brokerStateFiles(dir, deps).some((file) => { - if (!file.startsWith('broker-identity-')) return false; - try { - const record = JSON.parse(deps.fs.readFileSync(path.join(dir, file), 'utf-8')) as { pid?: unknown }; - return typeof record.pid === 'number' && record.pid > 0 && isProcessRunning(record.pid, deps); - } catch { - return false; - } - }); +/** + * Whether `dir` holds an identity record that still matches its process. A + * bare pid check would trust a stale record whose pid has been reused. + */ +async function hasVerifiedBrokerIdentity(dir: string, deps: CoreDependencies): Promise { + const paths = { ...deps.getProjectPaths(), dataDir: dir }; + for (const identity of readBrokerIdentities(paths, deps) ?? []) { + if (await matchesBrokerIdentity(identity, paths, deps)) return true; + } + return false; } // eslint-disable-next-line complexity, max-depth export async function runDownCommand(options: DownOptions, deps: CoreDependencies): Promise { const paths = deps.getProjectPaths(); if (options.stateDir) { - paths.dataDir = resolveExistingStateDir(options.stateDir, deps); + paths.dataDir = await resolveExistingStateDir(options.stateDir, deps); } const timeout = Number.parseInt(options.timeout ?? '5000', 10) || 5000; @@ -2748,7 +2749,7 @@ export async function runStatusCommand( ): Promise { const paths = deps.getProjectPaths(); if (options?.stateDir) { - paths.dataDir = resolveExistingStateDir(options.stateDir, deps); + paths.dataDir = await resolveExistingStateDir(options.stateDir, deps); } const waitMs = parseWaitForMs(options?.waitFor, deps); if (waitMs === null) { From 9377ba99472e2715ab2b6f2c5382be9271c56808 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 13:32:45 +0000 Subject: [PATCH 13/14] fix(cli): a stale connection file no longer outranks a live nested broker State-dir selection for node status/down counts connection.json as live evidence only while its pid is running; a dead file ranks below a verified identity record. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/commands/core.test.ts | 19 +++++++++++++++++++ packages/cli/src/cli/lib/broker-lifecycle.ts | 16 ++++++++++++---- 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index 0f357d186a..d3d2d8e1e0 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -2434,6 +2434,25 @@ describe('registerCoreCommands', () => { expect(deps.log).toHaveBeenCalledWith('PID: 4242'); }); + it('status --state-dir prefers a live nested broker over a stale connection file in the node directory', async () => { + const nodeDir = '/srv/stale-conn-node'; + const fs = createFsMock({ + [`${nodeDir}/connection.json`]: connectionFile(999), + [`${nodeDir}/state/connection.json`]: connectionFile(4242), + }); + sdkStatusClient.getStatus.mockResolvedValueOnce({ agent_count: 1, pending_delivery_count: 0 }); + sdkStatusClient.getSession.mockResolvedValueOnce({ workspace_key: 'rk_live_teststatus123' }); + const killImpl = vi.fn((pid: number, signal?: NodeJS.Signals | number) => { + if (signal === 0 && pid === 999) throw new Error('not running'); + }); + const { program, deps } = createHarness({ fs, killImpl }); + + expect(await runCommand(program, ['status', '--state-dir', nodeDir])).toBeUndefined(); + + expect(deps.log).toHaveBeenCalledWith('Status: RUNNING'); + expect(deps.log).toHaveBeenCalledWith('PID: 4242'); + }); + it('status checks broker status and prints metrics', async () => { const connectionPath = '/tmp/project/.agentworkforce/relay/connection.json'; const fs = createFsMock({ diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 7da792729a..8ca6d0359e 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -2517,9 +2517,10 @@ async function reportNodeClaimsElsewhere(paths: CoreProjectPaths, deps: CoreDepe * directory too when the nested one holds the broker (relay#1575). * * Evidence is ranked, checking the exact directory before `state/` at each - * rank: a `connection.json`, then an identity record verified against its - * running process (start time, executable and held lock, as `down` requires - * before signalling), then any broker lock or record. Leftover files in the node directory + * rank: a `connection.json` whose pid is running, then an identity record + * verified against its running process (start time, executable and held + * lock, as `down` requires before signalling), then any `connection.json`, + * then any broker lock or record. Leftover files in the node directory * therefore cannot shadow a running nested broker, and `down --force` can * still recover a broker whose connection file is gone. */ @@ -2527,8 +2528,9 @@ async function resolveExistingStateDir(stateDir: string, deps: CoreDependencies) const exact = path.resolve(stateDir); const candidates = [exact, path.join(exact, 'state')]; const ranks: Array<(dir: string) => boolean | Promise> = [ - (dir) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)), + (dir) => hasLiveConnection(dir, deps), (dir) => hasVerifiedBrokerIdentity(dir, deps), + (dir) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)), (dir) => hasBrokerStateFiles(dir, deps), ]; for (const matches of ranks) { @@ -2553,6 +2555,12 @@ function hasBrokerStateFiles(dir: string, deps: CoreDependencies): boolean { return brokerStateFiles(dir, deps).length > 0; } +/** A connection file only proves a live broker while its pid is running. */ +function hasLiveConnection(dir: string, deps: CoreDependencies): boolean { + const pid = readBrokerConnectionFromFs(deps.fs, dir)?.pid; + return typeof pid === 'number' && pid > 0 && isProcessRunning(pid, deps); +} + /** * Whether `dir` holds an identity record that still matches its process. A * bare pid check would trust a stale record whose pid has been reused. From aef361025bb5c42f316ab67d679e6b0e937a814c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 22:30:57 +0000 Subject: [PATCH 14/14] fix(cli): a reused-pid connection file no longer outranks a verified nested broker resolveExistingStateDir now checks for a verified identity record before a connection.json whose pid is merely running, so a stale node-dir file whose pid belongs to an unrelated process cannot redirect node status/down. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01J58Z89z3X6nT8XvtpR4Q5M --- packages/cli/src/cli/commands/core.test.ts | 30 ++++++++++++++++++++ packages/cli/src/cli/lib/broker-lifecycle.ts | 13 +++++---- 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/packages/cli/src/cli/commands/core.test.ts b/packages/cli/src/cli/commands/core.test.ts index d3d2d8e1e0..a81a3b8bba 100644 --- a/packages/cli/src/cli/commands/core.test.ts +++ b/packages/cli/src/cli/commands/core.test.ts @@ -1755,6 +1755,36 @@ describe('registerCoreCommands', () => { expect([...running]).toEqual([333]); }); + it('down --force ignores a node-dir connection file whose pid was reused by another process', async () => { + const nodeDir = '/srv/reused-conn-node'; + const stateDir = `${nodeDir}/state`; + const running = new Set([222, 333]); + const killImpl = vi.fn((pid: number, signal?: NodeJS.Signals | number) => { + if (signal === 0) { + if (!running.has(pid)) throw new Error('not running'); + return; + } + running.delete(pid); + }); + // The stale node-dir file names pid 333, which now belongs to an unrelated process. + const fs = createFsMock({ + [`${nodeDir}/connection.json`]: connectionFile(333), + [`${stateDir}/connection.json`]: connectionFile(222), + }); + const { program, deps } = createHarness({ + fs, + execCommand: identityCommand(222, `${stateDir}/broker-nested.lock`), + killImpl, + }); + await persistBrokerIdentity({ ...deps.getProjectPaths(), dataDir: stateDir }, 222, 'nested', deps); + + await runCommand(program, ['down', '--force', '--state-dir', nodeDir]); + + expect(killImpl).toHaveBeenCalledWith(222, 'SIGTERM'); + expect(killImpl).not.toHaveBeenCalledWith(333, 'SIGTERM'); + expect([...running]).toEqual([333]); + }); + it('down still verifies records written under the project by earlier releases', async () => { const stateDir = '/srv/legacy-node/state'; let running = true; diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 8ca6d0359e..d0142a84c1 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -2517,19 +2517,20 @@ async function reportNodeClaimsElsewhere(paths: CoreProjectPaths, deps: CoreDepe * directory too when the nested one holds the broker (relay#1575). * * Evidence is ranked, checking the exact directory before `state/` at each - * rank: a `connection.json` whose pid is running, then an identity record - * verified against its running process (start time, executable and held - * lock, as `down` requires before signalling), then any `connection.json`, - * then any broker lock or record. Leftover files in the node directory - * therefore cannot shadow a running nested broker, and `down --force` can + * rank: an identity record verified against its running process (start + * time, executable and held lock, as `down` requires before signalling), + * then a `connection.json` whose pid is running, then any `connection.json`, + * then any broker lock or record. Leftover files in the node directory — + * including a connection file whose pid was reused by another process — + * therefore cannot shadow a verified nested broker, and `down --force` can * still recover a broker whose connection file is gone. */ async function resolveExistingStateDir(stateDir: string, deps: CoreDependencies): Promise { const exact = path.resolve(stateDir); const candidates = [exact, path.join(exact, 'state')]; const ranks: Array<(dir: string) => boolean | Promise> = [ - (dir) => hasLiveConnection(dir, deps), (dir) => hasVerifiedBrokerIdentity(dir, deps), + (dir) => hasLiveConnection(dir, deps), (dir) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)), (dir) => hasBrokerStateFiles(dir, deps), ];