diff --git a/CHANGELOG.md b/CHANGELOG.md index d2c71cdd5..478643fd2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased - Patch] +### Added + +- `agent-relay node agent list --status` shows each agent's inbound delivery mode, pending-queue contents, and a derived "stuck" flag alongside last activity, reporting unavailable delivery reads as `unknown`. + ### Fixed - Live broker workers missing from the Relaycast reconnect inventory are now restored from their existing agent identity, so a node-control reconnect no longer makes a still-running terminal permanently unreachable. diff --git a/packages/cli/README.md b/packages/cli/README.md index 243af92c9..98b463b69 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -39,6 +39,7 @@ agent-relay node agent new claude # spawn + attach agent-relay node agent new codex --runtime native agent-relay node agent spawn opencode --runtime pty agent-relay node agent list +agent-relay node agent list --status # + inbound delivery mode and pending-queue contents per agent agent-relay node agent attach --mode view agent-relay node agent release ``` diff --git a/packages/cli/src/cli/commands/local-agent.test.ts b/packages/cli/src/cli/commands/local-agent.test.ts index c063b4fb8..eb4f8959c 100644 --- a/packages/cli/src/cli/commands/local-agent.test.ts +++ b/packages/cli/src/cli/commands/local-agent.test.ts @@ -11,7 +11,9 @@ vi.mock('@agent-relay/harness-driver', () => ({ import { formatPrettyAgentList, + formatPrettyAgentStatusList, registerLocalAgentCommands, + withDeliveryStatus, type LocalAgentDependencies, } from './local-agent.js'; @@ -497,6 +499,156 @@ describe('local agent subtree', () => { expect(row).not.toContain('\x1b'); }); + it('list --status enriches each agent with delivery mode and pending contents (relay#1387)', async () => { + const getInboundDeliveryMode = vi.fn(async () => 'manual_flush'); + const getPending = vi.fn(async () => [ + { from: 'a', body: 'hi', target: 'b', priority: 0, mode: 'wait', queued_at_ms: 1 }, + ]); + const { program, log } = harness({ + connect: vi.fn( + async () => + ({ + listAgents: vi.fn(async () => [{ name: 'lead' }]), + getInboundDeliveryMode, + getPending, + }) as never + ), + }); + + await program.parseAsync(['local', 'agent', 'list', '--status'], { from: 'user' }); + + expect(getInboundDeliveryMode).toHaveBeenCalledWith('lead'); + expect(getPending).toHaveBeenCalledWith('lead'); + const parsed = JSON.parse(log.mock.calls[0]![0] as string); + expect(parsed).toEqual([ + { + name: 'lead', + delivery_mode: 'manual_flush', + pending: [{ from: 'a', body: 'hi', target: 'b', priority: 0, mode: 'wait', queued_at_ms: 1 }], + }, + ]); + }); + + it('list --status --pretty marks manual_flush + non-empty queue as stuck, and auto_inject + empty queue as not stuck', () => { + const now = new Date('2026-08-16T18:00:00.000Z'); + const stuckAgent = { + name: 'stuck-agent', + runtime: 'pty' as const, + channels: [], + last_activity_at: '2026-08-16T17:00:00.000Z', + delivery_mode: 'manual_flush' as const, + pending: [ + { from: 'a', body: 'hi', target: 'stuck-agent', priority: 0, mode: 'wait' as const, queued_at_ms: 1 }, + ], + }; + const healthyAgent = { + name: 'healthy-agent', + runtime: 'pty' as const, + channels: [], + last_activity_at: '2026-08-16T17:59:00.000Z', + delivery_mode: 'auto_inject' as const, + pending: [], + }; + + const [, , stuckRow, healthyRow] = formatPrettyAgentStatusList([stuckAgent, healthyAgent], now).split( + '\n' + ); + + expect(stuckRow).toMatch(/stuck-agent\s+manual_flush\s+1\s+yes/); + expect(healthyRow).toMatch(/healthy-agent\s+auto_inject\s+0\s+no/); + }); + + it('list --status --pretty renders a failed pending lookup as unknown, distinct from empty and stuck', async () => { + const now = new Date('2026-08-16T18:00:00.000Z'); + const client = { + getInboundDeliveryMode: vi.fn(async () => 'manual_flush'), + getPending: vi.fn(async (name: string) => { + if (name === 'unknown-agent') throw new Error('broker unavailable'); + return name === 'stuck-agent' + ? [{ from: 'a', body: 'hi', target: name, priority: 0, mode: 'wait' as const, queued_at_ms: 1 }] + : []; + }), + }; + + const result = await withDeliveryStatus(client as never, [ + { + name: 'stuck-agent', + runtime: 'pty', + channels: [], + last_activity_at: '2026-08-16T17:00:00.000Z', + } as never, + { + name: 'empty-agent', + runtime: 'pty', + channels: [], + last_activity_at: '2026-08-16T17:00:00.000Z', + } as never, + { + name: 'unknown-agent', + runtime: 'pty', + channels: [], + last_activity_at: '2026-08-16T17:00:00.000Z', + } as never, + ]); + const [, , stuckRow, emptyRow, unknownRow] = formatPrettyAgentStatusList(result, now).split('\n'); + + expect(stuckRow).toMatch(/stuck-agent\s+manual_flush\s+1\s+yes/); + expect(emptyRow).toMatch(/empty-agent\s+manual_flush\s+0\s+no/); + expect(unknownRow).toMatch(/unknown-agent\s+manual_flush\s+-\s+unknown/); + }); + + it('withDeliveryStatus bounds concurrent broker reads while preserving input order', async () => { + let activeReads = 0; + let peakReads = 0; + let releaseReads: () => void; + const readsReleased = new Promise((resolve) => { + releaseReads = resolve; + }); + const read = (value: T) => + vi.fn(async () => { + activeReads += 1; + peakReads = Math.max(peakReads, activeReads); + try { + await readsReleased; + return value; + } finally { + activeReads -= 1; + } + }); + const client = { + getInboundDeliveryMode: read('auto_inject'), + getPending: read([]), + }; + const agents = Array.from({ length: 5 }, (_, index) => ({ name: `agent-${index}` }) as never); + + const result = withDeliveryStatus(client as never, agents); + + expect(activeReads).toBe(8); + releaseReads!(); + + await expect(result).resolves.toMatchObject(agents); + expect(peakReads).toBe(8); + expect(client.getInboundDeliveryMode).toHaveBeenCalledTimes(5); + expect(client.getPending).toHaveBeenCalledTimes(5); + }); + + it('withDeliveryStatus tolerates a per-agent delivery mode fetch failure', async () => { + const client = { + getInboundDeliveryMode: vi.fn(async (name: string) => + name === 'broken' ? Promise.reject(new Error('gone')) : 'auto_inject' + ), + getPending: vi.fn(async () => []), + }; + + const result = await withDeliveryStatus(client as never, [ + { name: 'ok' } as never, + { name: 'broken' } as never, + ]); + + expect(result[0]).toMatchObject({ name: 'ok', delivery_mode: 'auto_inject', pending: [] }); + expect(result[1]).toMatchObject({ name: 'broken', delivery_mode: undefined }); + }); + it('spawn forwards task-exit lifecycle options', async () => { const { program, client } = harness(); await program.parseAsync( diff --git a/packages/cli/src/cli/commands/local-agent.ts b/packages/cli/src/cli/commands/local-agent.ts index 2338d60a4..f145e9004 100644 --- a/packages/cli/src/cli/commands/local-agent.ts +++ b/packages/cli/src/cli/commands/local-agent.ts @@ -1,7 +1,7 @@ import type { Command } from 'commander'; import { HarnessDriverClient } from '@agent-relay/harness-driver'; -import type { ListAgent } from '@agent-relay/harness-driver'; +import type { InboundDeliveryMode, ListAgent, PendingRelayMessage } from '@agent-relay/harness-driver'; import type { HarnessRuntime } from '@agent-relay/harnesses'; import { stripAnsiFast } from '@agent-relay/utils'; @@ -338,6 +338,27 @@ function sanitizeTerminalCell(value: string): string { return stripAnsiFast(value).replace(/[\u0000-\u001f\u007f-\u009f\u202a-\u202e\u2066-\u2069]/g, '�'); } +/** Fixed-width text table shared by the plain and delivery-status pretty views. */ +function renderTable(columns: { header: string; values: string[] }[]): string { + const widths = columns.map((column) => + Math.max(column.header.length, ...column.values.map((value) => value.length)) + ); + const formatRow = (values: string[]) => + values + .map((value, index) => value.padEnd(widths[index]!)) + .join(' ') + .trimEnd(); + const rowCount = columns[0]?.values.length ?? 0; + + return [ + formatRow(columns.map((column) => column.header)), + formatRow(columns.map((_, index) => '-'.repeat(widths[index]!))), + ...Array.from({ length: rowCount }, (_, rowIndex) => + formatRow(columns.map((column) => column.values[rowIndex]!)) + ), + ].join('\n'); +} + /** Render a compact terminal view while retaining JSON as the script-friendly default. */ export function formatPrettyAgentList(agents: ListAgent[], now: Date): string { if (agents.length === 0) return 'No agents running.'; @@ -354,27 +375,98 @@ export function formatPrettyAgentList(agents: ListAgent[], now: Date): string { lastActive: sanitizeTerminalCell(formatRelativeTime(agent.last_activity_at, now)), }; }); - const columns = [ + + return renderTable([ { header: 'NAME', values: rows.map((row) => row.name) }, { header: 'CLI / MODEL', values: rows.map((row) => row.cliModel) }, { header: 'STATE', values: rows.map((row) => row.state) }, { header: 'PENDING', values: rows.map((row) => row.pending) }, { header: 'LAST ACTIVE', values: rows.map((row) => row.lastActive) }, - ]; - const widths = columns.map((column) => - Math.max(column.header.length, ...column.values.map((value) => value.length)) - ); - const formatRow = (values: string[]) => - values - .map((value, index) => value.padEnd(widths[index]!)) - .join(' ') - .trimEnd(); + ]); +} - return [ - formatRow(columns.map((column) => column.header)), - formatRow(columns.map((_, index) => '-'.repeat(widths[index]!))), - ...rows.map((row) => formatRow([row.name, row.cliModel, row.state, row.pending, row.lastActive])), - ].join('\n'); +/** + * A `ListAgent` enriched with the two delivery-observability fields that + * `/api/spawned` does not carry: the worker's current `InboundDeliveryMode` + * and its raw pending-queue contents (relay#1387). `delivery_mode`/`pending` + * are `undefined` when the per-agent fetch failed (e.g. the worker vanished + * mid-request), so callers can tell "unknown" apart from "empty". + */ +export interface AgentDeliveryStatus extends ListAgent { + delivery_mode?: InboundDeliveryMode; + pending?: PendingRelayMessage[]; +} + +/** Four agents at once means no more than eight concurrent broker reads. */ +const DELIVERY_STATUS_AGENT_CONCURRENCY = 4; + +/** Map in input order without allowing a fleet status read to overwhelm its broker. */ +async function mapWithConcurrency( + values: T[], + concurrency: number, + mapper: (value: T) => Promise +): Promise { + const results = new Array(values.length); + let nextIndex = 0; + + async function worker(): Promise { + while (true) { + const index = nextIndex; + nextIndex += 1; + if (index >= values.length) return; + results[index] = await mapper(values[index]!); + } + } + + await Promise.all(Array.from({ length: Math.min(concurrency, values.length) }, () => worker())); + return results; +} + +/** + * Fetch delivery mode + pending-queue contents for every listed agent. A + * status read costs one `/api/spawned` call plus two calls per agent, but we + * limit enrichment to four agents at once so inspecting a busy fleet cannot + * turn into an unbounded broker request fan-out. + */ +export async function withDeliveryStatus( + client: Pick, + agents: ListAgent[] +): Promise { + return mapWithConcurrency(agents, DELIVERY_STATUS_AGENT_CONCURRENCY, async (agent) => { + const [delivery_mode, pending] = await Promise.all([ + client.getInboundDeliveryMode(agent.name).catch(() => undefined), + client.getPending(agent.name).catch(() => undefined), + ]); + return { ...agent, delivery_mode, pending }; + }); +} + +/** A worker holding messages it cannot currently act on: parked in `manual_flush` with a non-empty queue. */ +function stuckStatus(agent: AgentDeliveryStatus): 'yes' | 'no' | 'unknown' { + // A failed status read is observability data, not proof that the queue is empty. + if (agent.delivery_mode === undefined || agent.pending === undefined) return 'unknown'; + return agent.delivery_mode === 'manual_flush' && agent.pending.length > 0 ? 'yes' : 'no'; +} + +/** Render the `--status` pretty view: mode, pending depth, and the derived stuck signal, alongside last-activity. */ +export function formatPrettyAgentStatusList(agents: AgentDeliveryStatus[], now: Date): string { + if (agents.length === 0) return 'No agents running.'; + + const rows = agents.map((agent) => ({ + name: sanitizeTerminalCell(agent.name), + mode: sanitizeTerminalCell(agent.delivery_mode ?? 'unknown'), + pending: agent.pending ? String(agent.pending.length) : '-', + stuck: stuckStatus(agent), + lastActive: sanitizeTerminalCell(formatRelativeTime(agent.last_activity_at, now)), + })); + + return renderTable([ + { header: 'NAME', values: rows.map((row) => row.name) }, + { header: 'MODE', values: rows.map((row) => row.mode) }, + { header: 'PENDING', values: rows.map((row) => row.pending) }, + { header: 'STUCK', values: rows.map((row) => row.stuck) }, + { header: 'LAST ACTIVE', values: rows.map((row) => row.lastActive) }, + ]); } async function run( @@ -500,9 +592,19 @@ export function registerLocalAgentCommands( .command('list') .description('List agents running on the local broker') .option('--pretty', 'Show a compact human-readable list') - .action(async (opts: { pretty?: boolean }) => { + .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) => { const agents = await client.listAgents(); + if (opts.status) { + const withStatus = await withDeliveryStatus(client, agents); + deps.log( + opts.pretty + ? formatPrettyAgentStatusList(withStatus, deps.now()) + : JSON.stringify(withStatus, null, 2) + ); + return; + } deps.log(opts.pretty ? formatPrettyAgentList(agents, deps.now()) : JSON.stringify(agents, null, 2)); }); });