diff --git a/CHANGELOG.md b/CHANGELOG.md
index af9a22bc2..07558b008 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,7 +5,19 @@ 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]
+## [Unreleased - Minor]
+
+### Added
+
+- `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/`.
+- 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.
+- `node status` and `node down` name a missing broker identity record and how to create one.
## [13.1.0] - 2026-10-03
diff --git a/packages/cli/README.md b/packages/cli/README.md
index e6ef31c1e..afa86e8e9 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
```
+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`.
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 082726db6..a81a3b8bb 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,213 @@ 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 --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('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 --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 --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 --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;
+ 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 +2432,57 @@ 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 --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 --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/commands/local-agent.test.ts b/packages/cli/src/cli/commands/local-agent.test.ts
index ba3395213..5faf20bbc 100644
--- a/packages/cli/src/cli/commands/local-agent.test.ts
+++ b/packages/cli/src/cli/commands/local-agent.test.ts
@@ -1201,3 +1201,142 @@ 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.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();
+ 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).toHaveBeenCalledWith('/tmp/project', {
+ brokerUrl: undefined,
+ apiKey: undefined,
+ stateDir,
+ });
+ 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')} 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 });
+ }
+ });
+
+ 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 fe6cf2506..eeb23cbd7 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,35 @@ export function formatPrettyAgentStatusList(agents: AgentDeliveryStatus[], now:
]);
}
+/**
+ * Whether the caller chose a broker explicitly (`--broker-url` / `--api-key` /
+ * `--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(
+ deps: Pick,
+ options: LocalAgentMessageBrokerOptions
+): boolean {
+ return Boolean(
+ options.brokerUrl?.trim() ||
+ options.apiKey?.trim() ||
+ options.stateDir?.trim() ||
+ deps.env.RELAY_BROKER_URL?.trim() ||
+ deps.env.RELAY_BROKER_API_KEY?.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 +814,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 +835,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 +866,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 +928,7 @@ export function registerLocalAgentCommands(
})
)
return;
+ const brokerOptions = brokerOptionsFromOpts(options);
const baseName = (options.name as string | undefined) ?? provider;
const resolved = resolveAutoSpawn(
provider,
@@ -895,52 +936,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 +1195,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 1b548cac5..668c211df 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 7a00d2e47..736db5a9a 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 a9528f074..b602da0b1 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 557f9db09..92d73327e 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 c4752a610..1fae79034 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 f05cfd52a..ff43ed68a 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 6587a640c..e14bd3fcf 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,127 @@ 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 = {
+ [nodeStateDir]: { 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: nodeStateDir }, 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: path.resolve('/srv/missing') }, deps)).toBeNull();
+ });
+
+ it('still lets --api-key override the connection file key', () => {
+ const deps = makeDeps({ readConnectionFile });
+ 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: nodeDir }, 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('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', () => {
+ // 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', () => {
+ 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(
+ `${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');
+ });
+});
+
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 45cb66e57..839106d23 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,75 @@ 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) {
+ // The caller named this broker's state directory: its connection file is
+ // 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(
+ explicitStateDir,
+ (dir) => fs.existsSync(path.join(dir, 'connection.json')) || deps.readConnectionFile(dir) !== null
+ );
+ const connectionFile = deps.readConnectionFile(stateDir);
+ 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 = options.stateDir ? path.resolve(options.stateDir) : deps.getDefaultStateDir();
- const connectionFile = deps.readConnectionFile(stateDir);
+ const connectionFile = deps.readConnectionFile(deps.getDefaultStateDir());
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);
+ 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 ${exact} or ${nested} (from --state-dir). ` +
+ '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 fc7dff1ef..d0142a84c 100644
--- a/packages/cli/src/cli/lib/broker-lifecycle.ts
+++ b/packages/cli/src/cli/lib/broker-lifecycle.ts
@@ -9,7 +9,8 @@ import { redactCredentialValues } from '@agent-relay/cloud/redact';
import type { CoreDependencies, CoreProjectPaths, CoreRelay, SpawnedProcess } from '../commands/core.js';
import {
- brokerIdentityPath,
+ brokerIdentityDirectory,
+ locateBrokerIdentityFile,
matchesBrokerIdentity,
persistBrokerIdentity,
readBrokerIdentities,
@@ -1077,7 +1078,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 };
}
@@ -1093,7 +1094,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;
@@ -1114,7 +1115,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;
}
@@ -1186,7 +1187,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 +2027,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 +2074,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 +2096,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 +2511,74 @@ 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 the nested one holds the broker (relay#1575).
+ *
+ * Evidence is ranked, checking the exact directory before `state/` at each
+ * 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) => hasVerifiedBrokerIdentity(dir, deps),
+ (dir) => hasLiveConnection(dir, deps),
+ (dir) => deps.fs.existsSync(path.join(dir, CONNECTION_FILENAME)),
+ (dir) => hasBrokerStateFiles(dir, deps),
+ ];
+ for (const matches of ranks) {
+ for (const candidate of candidates) {
+ if (await matches(candidate)) return candidate;
+ }
+ }
+ return exact;
+}
+
+function brokerStateFiles(dir: string, deps: CoreDependencies): string[] {
+ try {
+ return deps.fs
+ .readdirSync(dir)
+ .filter((file) => file.startsWith('broker-') && (file.endsWith('.lock') || file.endsWith('.json')));
+ } catch {
+ return [];
+ }
+}
+
+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.
+ */
+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 = path.resolve(options.stateDir);
+ paths.dataDir = await resolveExistingStateDir(options.stateDir, deps);
}
const timeout = Number.parseInt(options.timeout ?? '5000', 10) || 5000;
@@ -2612,9 +2676,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 +2758,7 @@ export async function runStatusCommand(
): Promise {
const paths = deps.getProjectPaths();
if (options?.stateDir) {
- paths.dataDir = path.resolve(options.stateDir);
+ paths.dataDir = await resolveExistingStateDir(options.stateDir, deps);
}
const waitMs = parseWaitForMs(options?.waitFor, deps);
if (waitMs === null) {
@@ -2727,6 +2801,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 54b3813c4..bca48eecd 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 5757f021e..fec7fd14f 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,47 @@ 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;
+}
+
+/** 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,
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 +187,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(
@@ -187,6 +252,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.
@@ -199,6 +265,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,
@@ -278,11 +367,16 @@ export function removeBrokerIdentity(
identity: BrokerProcessIdentity,
deps: IdentityDependencies
): void {
- const current = readBrokerIdentity(paths, deps, identity.brokerName);
- if (!current || JSON.stringify(current) !== JSON.stringify(identity)) return;
- try {
- deps.fs.unlinkSync(brokerIdentityPath(paths, deps, identity.brokerName));
- } 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/scripts/verify-features/targeted-relayflow-case.mjs b/scripts/verify-features/targeted-relayflow-case.mjs
index ed195e10b..68e93d771 100644
--- a/scripts/verify-features/targeted-relayflow-case.mjs
+++ b/scripts/verify-features/targeted-relayflow-case.mjs
@@ -35,6 +35,64 @@ async function gitHead(repoRoot) {
return result.stdout.trim();
}
+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), 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;
+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) {
+ 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.
+ *
+ * `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, 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) * 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', '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()}`
+ );
+ }
+}
+
async function main() {
const caseId = requiredOption('--case');
const timeoutSeconds = Number(requiredOption('--timeout-seconds'));
@@ -60,6 +118,8 @@ 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,
@@ -96,8 +156,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 });
+ try {
+ 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;
+ console.error(restoreError instanceof Error ? restoreError.message : String(restoreError));
+ }
}
}
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 000000000..ea6895998
--- /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 000000000..e68b891b0
--- /dev/null
+++ b/tests/relayflows/cases/1446-state-dir-broker-selection/run.mjs
@@ -0,0 +1,208 @@
+#!/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.
+ *
+ * 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, 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';
+
+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');
+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 cliEntry = path.join(targetDir, 'packages/cli/dist/cli/index.js');
+
+run('npm', ['ci', '--ignore-scripts'], targetDir, 'workspace dependency installation');
+run('npm', ['run', 'build:core'], targetDir, 'workspace package build');
+
+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');
+ 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,
+ })
+ );
+ // The real CLI binary, from an unrelated checkout, with ambient env naming a decoy broker.
+ const env = {
+ 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',
+ };
+ 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 = { list, release, requests };
+ console.log('State-dir broker selection observation:', JSON.stringify(observation));
+
+ 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.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');
+
+ let outcome;
+ let signature;
+ let details;
+ if (baseObserved) {
+ outcome = 'bug';
+ signature = 'node_agent_state_dir_rejected';
+ details =
+ "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 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)}.`);
+ }
+
+ 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 {
+ 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));
+ // 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);
+ resolve({ code, stdout: stdout.slice(-4000), stderr: stderr.slice(-4000) });
+ });
+ });
+}
+
+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))
+ );
+}
+
+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)}`
+ );
+ }
+}