diff --git a/CHANGELOG.md b/CHANGELOG.md index c86799ccb..fd19b7743 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `agent-relay node up|down|status|metrics|tail`, `node agent …`, and `node workflow run|logs|sync` unify `local up` and `fleet serve` under one command group. `node up [--config ]` brings the current context's node online: the broker runs agents, and a project `agent-relay.{ts,…}` or `agent-relay.py` is served as a capability provider. A plain agent host needs no definition file. - **Node providers** — a node's actions are hosted by providers that connect directly to the engine and are invoked node-addressed (`POST /v1/nodes/:node/actions/:name/invoke`). Author TypeScript actions with `@agent-relay/fleet` (`defineNode`/`serveNode`); an action named `spawn:` wraps the broker's spawn so any language can adjust the command, env, or cwd before it runs. - Python `agent_relay.node.NodeProvider.from_enrollment()` serves node actions from Python, reading the node credentials from the enrollment. Install with the `node` extra. +- `agent-relay node up` gains `--log-file `, `--log-level `, and `--log-json`: a served node logs each capability it registers (`debug`) and every action that hits it (`info`, with a duration and `node`/`kind`/`invocationId` fields; failures at `warn`). Without a flag the node stays quiet apart from warnings; `--verbose` raises the level to `debug`. An invalid `--log-level` is now rejected instead of silently disabling logs. Serving programmatically, inject any sink via `serveNode({ logger })`. - Swift SDK (`AgentRelaySDK`, `packages/sdk-swift`): `AgentClient` gains `invokeAction(_:input:timeout:pollInterval:)` — invoke a relay action and await its output — plus `channelHistory(_:limit:before:)` and `dmHistory(with:limit:before:)` for reading channel and 1:1 DM message history as oldest-first `RelayChannelEvent`s. ### Changed diff --git a/packages/cli/src/cli/commands/core.ts b/packages/cli/src/cli/commands/core.ts index 033b7e029..2ad00a10b 100644 --- a/packages/cli/src/cli/commands/core.ts +++ b/packages/cli/src/cli/commands/core.ts @@ -4,7 +4,7 @@ import path from 'node:path'; import { fileURLToPath } from 'node:url'; import { exec, spawn as spawnProcess } from 'node:child_process'; import { promisify } from 'node:util'; -import { Command } from 'commander'; +import { Command, InvalidArgumentError } from 'commander'; import { getProjectPaths, loadTeamsConfig } from '@agent-relay/config'; import { HarnessDriverClient, type BrokerInitArgs } from '@agent-relay/harness-driver'; @@ -270,6 +270,9 @@ export interface UpCommandOptions { stateDir?: string; brokerName?: string; config?: string; + logFile?: string; + logLevel?: string; + logJson?: boolean; } /** @@ -287,7 +290,32 @@ export function addUpCommandOptions(command: Command): Command { '--state-dir ', 'Directory for broker state and connection files (default: .agentworkforce/relay/)' ) - .option('--broker-name ', 'Override the broker name (defaults to project directory basename)'); + .option('--broker-name ', 'Override the broker name (defaults to project directory basename)') + .option( + '--log-file ', + 'Write structured node logs (capabilities registered, actions invoked/completed) to a file' + ) + .option( + '--log-level ', + 'Node log verbosity: debug | info | warn | error (default: info)', + parseLogLevel + ) + .option('--log-json', 'Emit node logs as JSON lines instead of text'); +} + +const LOG_LEVELS = ['debug', 'info', 'warn', 'error'] as const; + +/** + * Validate `--log-level` at parse time (case-insensitive). Rejecting a typo here + * — rather than passing it through — avoids the silent failure where an + * unrecognized `AGENT_RELAY_LOG_LEVEL` drops every log line. + */ +function parseLogLevel(value: string): string { + const normalized = value.toLowerCase(); + if (!(LOG_LEVELS as readonly string[]).includes(normalized)) { + throw new InvalidArgumentError(`Expected one of: ${LOG_LEVELS.join(', ')}.`); + } + return normalized; } export function registerCoreCommands( diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index 37923ad7e..76837adad 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -2,6 +2,7 @@ import fs from 'node:fs'; import path from 'node:path'; import { HarnessDriverClient } from '@agent-relay/harness-driver'; import { startServeNode, type FleetNodeDefinition, type RunningNode } from '@agent-relay/fleet'; +import { createLogger } from '@agent-relay/utils'; import type { CoreDependencies, CoreProjectPaths, CoreRelay, SpawnedProcess } from '../commands/core.js'; import { track } from '../telemetry/index.js'; @@ -32,6 +33,12 @@ type UpOptions = { discoverConfig?: boolean; /** Registered node name override (e.g. from a persisted Cloud enrollment). */ nodeName?: string; + /** Write structured node logs (capabilities, action invocations) to this file. */ + logFile?: string; + /** Log verbosity floor: debug | info | warn | error. Defaults to info. */ + logLevel?: string; + /** Emit logs as JSON lines instead of human-readable text. */ + logJson?: boolean; }; type DownOptions = { @@ -136,6 +143,30 @@ function vlog(deps: CoreDependencies, verbose: boolean | undefined, message: str } } +/** True when any log flag (or `--verbose`) opts the node into structured logging. */ +function nodeLoggingEnabled(options: UpOptions): boolean { + return Boolean(options.logFile || options.logLevel || options.logJson || options.verbose); +} + +/** + * Translate the `--log-*` (and `--verbose`) flags into the `AGENT_RELAY_LOG_*` + * environment the shared `createLogger` reads. `--verbose` alone raises the + * floor to DEBUG so per-capability registration lines surface; an explicit + * `--log-level` always wins. Called before the fleet sidecar starts. + */ +function applyNodeLogEnv(options: UpOptions, deps: CoreDependencies): void { + if (options.logFile) { + deps.env.AGENT_RELAY_LOG_FILE = options.logFile; + } + const level = options.logLevel ?? (options.verbose ? 'debug' : undefined); + if (level) { + deps.env.AGENT_RELAY_LOG_LEVEL = level.toUpperCase(); + } + if (options.logJson) { + deps.env.AGENT_RELAY_LOG_JSON = '1'; + } +} + type ErrorWithCode = { code?: unknown }; function isRecord(value: unknown): value is Record { @@ -390,8 +421,14 @@ async function startNodeCapabilityProviders( providerName: nodeDefinition.name, ...(workspaceKey ? { triggers: createTriggerSyncClient({ workspaceKey, baseUrl }) } : {}), reconnect: true, - warn: (message) => deps.warn(message), - log: (message) => deps.log(message), + // With any --log-* flag (or --verbose), surface the node's full lifecycle + // — capabilities registered, every action invoked/completed — through the + // shared logger, which honors AGENT_RELAY_LOG_FILE/_LEVEL/_JSON. Without a + // flag, keep the prior behavior: the registration summary via log, warnings + // via warn. + ...(nodeLoggingEnabled(options) + ? { logger: createLogger('fleet') } + : { warn: (message) => deps.warn(message), log: (message) => deps.log(message) }), }) ); } catch (err) { @@ -1099,6 +1136,10 @@ export async function runUpCommand(options: UpOptions, deps: CoreDependencies): deps.env.RELAY_API_KEY = options.workspaceKey; } + // Point the shared logger at a file / level / format before the fleet + // sidecar (which reads this env when it builds its logger) starts. + applyNodeLogEnv(options, deps); + // Resolved BEFORE the broker starts so an explicit bad --config fails // fast instead of tearing down a broker that just came up. const nodeDefinition = await resolveNodeDefinitionForUp(paths, options, deps); diff --git a/packages/fleet/README.md b/packages/fleet/README.md index c6b468619..e1c62d840 100644 --- a/packages/fleet/README.md +++ b/packages/fleet/README.md @@ -73,6 +73,33 @@ await running.stop(); await serveNode({ definition, connection }); ``` +### Logging + +The node runtime emits structured events — each capability it registers and every +action that hits it (invoked / completed / failed, with a duration) — through a +`logger` you inject. The shape matches `@agent-relay/utils`' `createLogger`, and +every event carries a structured `extra` bag (`{ capability, action, invocationId, +ms, … }`) so file and JSON sinks can key on the fields: + +```ts +import { createLogger } from '@agent-relay/utils'; + +await serveNode({ definition, connection, logger: createLogger('fleet') }); +``` + +Via the CLI, `agent-relay node up` surfaces this without code: + +```bash +agent-relay node up --config ./builder.node.ts --log-file ./node.log +agent-relay node up --config ./builder.node.ts --log-level debug # include per-capability lines +agent-relay node up --config ./builder.node.ts --log-json # one JSON object per line +``` + +Capability registration logs at `debug`; action invocations at `info`; failures at +`warn`. With no `logger` the node is silent — pass a `logger` (or the older `log`/`warn` +callbacks) to receive events. The CLI wires a `warn`-only sink when no `--log-*` flag is +given, so `agent-relay node up` stays quiet apart from warnings until you opt in. + ## Concepts - **Node** — a named host registered with the workspace. `defineNode` validates the manifest diff --git a/packages/fleet/src/serve-node.test.ts b/packages/fleet/src/serve-node.test.ts index 00e84bf28..fe2800424 100644 --- a/packages/fleet/src/serve-node.test.ts +++ b/packages/fleet/src/serve-node.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { z } from 'zod'; import { action, defineNode, spawn } from './index.js'; -import { startServeNode, type FleetTriggerSyncClient } from './serve-node.js'; +import { startServeNode, type FleetLogger, type FleetTriggerSyncClient } from './serve-node.js'; /** * A fake node-ws server: captures frames the client sends and lets the test @@ -307,4 +307,104 @@ describe('serveNode', () => { ]); await running.stop(); }); + + describe('structured logging', () => { + function recordingLogger() { + const entries: Array<{ level: string; message: string; extra?: Record }> = []; + const record = (level: string) => (message: string, extra?: Record) => { + entries.push({ level, message, ...(extra ? { extra } : {}) }); + }; + const logger: FleetLogger = { + debug: record('debug'), + info: record('info'), + warn: record('warn'), + error: record('error'), + }; + return { logger, entries }; + } + + function loggingNode() { + return defineNode({ + name: 'test-node', + capabilities: { + echo: action({ input: z.object({ value: z.string() }) }, async (input) => input), + explode: action({}, async () => { + throw new Error('boom'); + }), + }, + }); + } + + it('logs each registered capability at debug and a registration summary at info', async () => { + const { logger, entries } = recordingLogger(); + const running = startServeNode({ definition: loggingNode(), connection, reconnect: false, logger }); + const sock = socket(); + sock.open(); + sock.emit(acceptAll(sock.lastRegister())); + await flush(); + + const capabilities = entries.filter((entry) => entry.message.startsWith('Capability')); + expect(capabilities.map((entry) => entry.extra?.capability)).toEqual(['echo', 'explode']); + expect(capabilities.every((entry) => entry.level === 'debug')).toBe(true); + expect(capabilities[0]?.extra).toMatchObject({ node: 'test-node', capability: 'echo', kind: 'action' }); + + const summary = entries.find((entry) => entry.message.includes('with 2 capabilities')); + expect(summary?.level).toBe('info'); + expect(summary?.extra).toMatchObject({ node: 'test-node', capabilities: 2 }); + + await running.stop(); + }); + + it('logs an action invocation and its completion with a duration', async () => { + const { logger, entries } = recordingLogger(); + const running = startServeNode({ definition: loggingNode(), connection, reconnect: false, logger }); + const sock = socket(); + sock.open(); + sock.emit(acceptAll(sock.lastRegister())); + await flush(); + + sock.emit({ + v: 1, + type: 'action.invoke', + invocation_id: 'inv-1', + action: 'echo', + input: { value: 'hi' }, + }); + await flush(); + + const invoked = entries.find((entry) => entry.message === 'Action "echo" invoked'); + expect(invoked?.level).toBe('info'); + expect(invoked?.extra).toMatchObject({ + node: 'test-node', + action: 'echo', + kind: 'action', + invocationId: 'inv-1', + }); + + const completed = entries.find((entry) => entry.message === 'Action "echo" completed'); + expect(completed?.level).toBe('info'); + expect(completed?.extra).toMatchObject({ node: 'test-node', action: 'echo', invocationId: 'inv-1' }); + expect(typeof completed?.extra?.ms).toBe('number'); + + await running.stop(); + }); + + it('logs a failed action as a warning carrying the error', async () => { + const { logger, entries } = recordingLogger(); + const running = startServeNode({ definition: loggingNode(), connection, reconnect: false, logger }); + const sock = socket(); + sock.open(); + sock.emit(acceptAll(sock.lastRegister())); + await flush(); + + sock.emit({ v: 1, type: 'action.invoke', invocation_id: 'inv-2', action: 'explode', input: {} }); + await flush(); + + const failed = entries.find((entry) => entry.message === 'Action "explode" failed'); + expect(failed?.level).toBe('warn'); + expect(failed?.extra).toMatchObject({ action: 'explode', invocationId: 'inv-2', error: 'boom' }); + + await running.stop(); + }); + }); }); diff --git a/packages/fleet/src/serve-node.ts b/packages/fleet/src/serve-node.ts index e20afd08d..b7af3a44b 100644 --- a/packages/fleet/src/serve-node.ts +++ b/packages/fleet/src/serve-node.ts @@ -65,6 +65,23 @@ export interface FleetTriggerSyncClient { delete(id: string): Promise; } +/** + * Structured logger the serve runtime writes to. The shape matches + * `@agent-relay/utils`' `createLogger` so the CLI can inject that logger + * directly, but fleet depends on nothing to keep the seam a plain interface: + * a host may supply any sink (stdout, a file, a JSON collector). + * + * Each method takes a human message plus an optional structured `extra` bag — + * `{ node, capability, action, invocationId, ms, ... }` — so file/JSON adapters + * can key on the fields rather than parse the message. + */ +export interface FleetLogger { + debug(message: string, extra?: Record): void; + info(message: string, extra?: Record): void; + warn(message: string, extra?: Record): void; + error(message: string, extra?: Record): void; +} + /** * Options for {@link serveNode} / {@link startServeNode}. */ @@ -86,11 +103,38 @@ export interface ServeNodeOptions { maxAgentsOverride?: number; reconnect?: boolean; signal?: AbortSignal; + /** + * Structured sink for lifecycle and per-invocation events. Preferred over + * `log`/`warn`: capability registration and every action hitting the node are + * emitted here with structured `extra` fields. When omitted, `log`/`warn` + * receive the same events as plain messages. + */ + logger?: FleetLogger; log?: (message: string) => void; warn?: (message: string) => void; onRegistered?: (info: ReturnType) => void; } +/** + * Resolve a {@link FleetLogger} from the serve options. An injected `logger` + * wins; otherwise the legacy `log`/`warn` callbacks are adapted (info/debug → + * `log`, warn/error → `warn`), with a no-op fallback so callers can pass + * neither. + */ +function resolveLogger(options: ServeNodeOptions): FleetLogger { + if (options.logger) { + return options.logger; + } + const info = options.log ?? (() => undefined); + const warn = options.warn ?? (() => undefined); + return { + debug: (message) => info(message), + info: (message) => info(message), + warn: (message) => warn(message), + error: (message) => warn(message), + }; +} + /** * Handle returned by {@link startServeNode} for stopping a running node. */ @@ -138,6 +182,7 @@ export async function serveNode(options: ServeNodeOptions): Promise { const providerName = options.providerName ?? options.definition.name; const maxAgents = options.maxAgentsOverride ?? options.definition.maxAgents; const reconnect = options.reconnect ?? true; + const logger = resolveLogger(options); const client = new NodeProviderClient({ ...(options.connection.baseUrl ? { baseUrl: options.connection.baseUrl } : {}), @@ -152,7 +197,10 @@ export async function serveNode(options: ServeNodeOptions): Promise { ...(reconnect ? {} : { maxReconnectAttempts: 0 }), onError: (error) => { if (!options.signal?.aborted) { - options.warn?.(`Fleet node error: ${errorMessage(error)}`); + logger.warn(`Fleet node error: ${errorMessage(error)}`, { + node: nodeName, + error: errorMessage(error), + }); } }, }); @@ -161,7 +209,13 @@ export async function serveNode(options: ServeNodeOptions): Promise { // Both `action` and `spawn` definitions materialize as invokable `action` // capabilities: a `spawn:` definition shadows the node's native // capacity, delegating through `ctx.spawnAgent`. - client.capability(name, { kind: 'action' }, adaptHandler(options, name)); + const kind = options.definition.capabilities[name]?.kind; + logger.debug(`Capability "${name}" registered`, { + node: nodeName, + capability: name, + ...(kind ? { kind } : {}), + }); + client.capability(name, { kind: 'action' }, adaptHandler(options, name, logger)); } const abort = () => { @@ -190,11 +244,11 @@ export async function serveNode(options: ServeNodeOptions): Promise { name: nodeName, ...(maxAgents !== undefined ? { maxAgents } : {}), }); - await syncTriggers(options); - options.log?.( - `Fleet node "${nodeName}" registered provider "${providerName}" with ${ - Object.keys(options.definition.capabilities).length - } capabilities.` + await syncTriggers(options, logger); + const capabilityCount = Object.keys(options.definition.capabilities).length; + logger.info( + `Fleet node "${nodeName}" registered provider "${providerName}" with ${capabilityCount} capabilities.`, + { node: nodeName, capabilities: capabilityCount } ); try { @@ -208,10 +262,37 @@ export async function serveNode(options: ServeNodeOptions): Promise { * Adapt a fleet capability handler to the engine node-provider handler contract: * validate the input against the capability schema (via {@link invokeNodeHandler}) * and expose the fleet action context built from the engine handler context. + * + * Every invocation is logged invoked → completed/failed with the elapsed ms and + * structured `{ node, action, kind, invocationId }` fields, so a file/JSON sink + * can group a node's activity by node and by capability kind (spawn vs action). */ -function adaptHandler(options: ServeNodeOptions, name: string): NodeCapabilityHandler { - return (input, nodeCtx) => - invokeNodeHandler(options.definition, name, input, makeContext(options, nodeCtx, name)); +function adaptHandler(options: ServeNodeOptions, name: string, logger: FleetLogger): NodeCapabilityHandler { + const node = options.nameOverride ?? options.definition.name; + const kind = options.definition.capabilities[name]?.kind; + const base = { node, action: name, ...(kind ? { kind } : {}) }; + return async (input, nodeCtx) => { + const context = { ...base, invocationId: nodeCtx.invocationId }; + logger.info(`Action "${name}" invoked`, context); + const startedAt = Date.now(); + try { + const output = await invokeNodeHandler( + options.definition, + name, + input, + makeContext(options, nodeCtx, name) + ); + logger.info(`Action "${name}" completed`, { ...context, ms: Date.now() - startedAt }); + return output; + } catch (error) { + logger.warn(`Action "${name}" failed`, { + ...context, + ms: Date.now() - startedAt, + error: errorMessage(error), + }); + throw error; + } + }; } // The engine handler context takes wire-JSON shapes; the fleet authoring API @@ -280,7 +361,7 @@ function buildSpawnInput( } as unknown as NodeSpawnInput; } -async function syncTriggers(options: ServeNodeOptions): Promise { +async function syncTriggers(options: ServeNodeOptions, logger: FleetLogger): Promise { const triggers = triggerSyncInputs(options.definition); if (triggers.length === 0 || !options.triggers) { return; @@ -327,7 +408,10 @@ async function syncTriggers(options: ServeNodeOptions): Promise { }) ); } catch (error) { - options.warn?.(`Fleet trigger sync skipped: ${errorMessage(error)}`); + logger.warn(`Fleet trigger sync skipped: ${errorMessage(error)}`, { + node: options.nameOverride ?? options.definition.name, + error: errorMessage(error), + }); } } @@ -350,7 +434,7 @@ function triggerSyncKey(trigger: { trigger.channel ?? '', trigger.pattern ?? '', String(normalizeTriggerMention(trigger.mention)), - ].join('\u001f'); + ].join(''); } function triggerEquals( diff --git a/packages/utils/src/logger.ts b/packages/utils/src/logger.ts index 6fda566a9..6457b71aa 100644 --- a/packages/utils/src/logger.ts +++ b/packages/utils/src/logger.ts @@ -28,7 +28,10 @@ function getLogFile(): string | undefined { } function getLogLevel(): LogLevel { - return (process.env.AGENT_RELAY_LOG_LEVEL ?? 'INFO').toUpperCase() as LogLevel; + const level = (process.env.AGENT_RELAY_LOG_LEVEL ?? 'INFO').toUpperCase(); + // An unrecognized level would make LEVEL_PRIORITY[level] undefined and every + // shouldLog() comparison false — silently dropping all logs. Fall back to INFO. + return (level in LEVEL_PRIORITY ? level : 'INFO') as LogLevel; } function isLogJson(): boolean {