diff --git a/src/core/dev/inspector/server.test.ts b/src/core/dev/inspector/server.test.ts index b1f7335097..c6e97560e1 100644 --- a/src/core/dev/inspector/server.test.ts +++ b/src/core/dev/inspector/server.test.ts @@ -92,7 +92,7 @@ describe("GET /api/status", () => { buildType: "Container", protocol: "MCP", phase: "failed", - error: "boom", + error: new Error("boom"), }, ], }); diff --git a/src/core/dev/inspector/server.ts b/src/core/dev/inspector/server.ts index 28470b6d8e..ca4847c0f4 100644 --- a/src/core/dev/inspector/server.ts +++ b/src/core/dev/inspector/server.ts @@ -86,9 +86,9 @@ function handleStatus(deps: InspectorDeps): HttpResponse { running: snapshot .filter((agent) => agent.phase === "running" && agent.port !== undefined) .map(({ name, port }) => ({ name, port: port! })), - errors: snapshot - .filter((agent) => agent.error !== undefined) - .map(({ name, error }) => ({ name, message: error! })), + errors: snapshot.flatMap(({ name, error }) => + error ? [{ name, message: error.message }] : [], + ), selectedAgent: deps.selectedAgent, }; return json(200, status); diff --git a/src/core/dev/inspector/types.ts b/src/core/dev/inspector/types.ts index 3d8be57be5..e43f1ea0b5 100644 --- a/src/core/dev/inspector/types.ts +++ b/src/core/dev/inspector/types.ts @@ -14,7 +14,7 @@ export interface InspectorAgentStatus { protocol: string; phase: "idle" | "starting" | "running" | "failed"; port?: number; - error?: string; + error?: Error; } /** The slice of the dev supervisor the Inspector routes need. */ diff --git a/src/core/dev/supervisor.test.ts b/src/core/dev/supervisor.test.ts index 4c81d83b1c..53e4441c95 100644 --- a/src/core/dev/supervisor.test.ts +++ b/src/core/dev/supervisor.test.ts @@ -179,7 +179,7 @@ describe("DevSupervisor", () => { expect(supervisor.snapshot()[0]).toMatchObject({ name: "orders", phase: "failed", - error: "boom", + error: { message: "boom" }, }); // Retry hits the runner again rather than being stuck. @@ -262,7 +262,7 @@ describe("DevSupervisor", () => { expect(supervisor.snapshot()[0]).toMatchObject({ name: "orders", phase: "failed", - error: "segfault", + error: { message: "segfault" }, }); expect(supervisor.running("orders")).toBeUndefined(); const events = await drain(supervisor, controller); diff --git a/src/core/dev/supervisor.ts b/src/core/dev/supervisor.ts index 289846468c..53350a5f1d 100644 --- a/src/core/dev/supervisor.ts +++ b/src/core/dev/supervisor.ts @@ -11,7 +11,7 @@ export interface AgentStatus { protocol: NonNullable; phase: AgentPhase; port?: number; - error?: string; + error?: Error; } /** A dev event tagged with the name of the runtime that produced it. */ @@ -41,7 +41,8 @@ type AgentEntry = { runtime: ProjectRuntime; phase: AgentPhase; port?: number; - error?: string; + error?: Error; + completed?: boolean; starting?: Promise<{ name: string; port: number }>; /** The running child's pump, so shutdown can await its final spans. */ running?: Promise; @@ -152,18 +153,30 @@ export class DevSupervisor { /** * The merged event stream of every agent this supervisor has started. Ends * when the supervisor's signal aborts and all pending events are drained. + * Headless consumers can also finish after every agent stops. */ - public async *events(): AsyncGenerator { + public async *events( + options: { untilStopped?: boolean } = {}, + ): AsyncGenerator { while (true) { for (const event of this.queue.splice(0)) yield event; if (this.config.signal.aborted) { // Let every live child finish shutting down so its final spans reach the // collector before the caller closes it, then drain what they emitted. - const running = [...this.agents.values()].map((entry) => entry.running).filter(Boolean); - await Promise.allSettled(running); + const pending = [...this.agents.values()] + .flatMap((entry) => [entry.starting, entry.running]) + .filter(Boolean); + await Promise.allSettled(pending); for (const event of this.queue.splice(0)) yield event; return; } + if ( + options.untilStopped && + [...this.agents.values()].every(({ phase }) => phase !== "starting" && phase !== "running") + ) { + if (this.queue.length > 0) continue; + return; + } await new Promise((resolve) => { this.wake = resolve; // A push during the yields above ran while wake was undefined, so its @@ -184,6 +197,7 @@ export class DevSupervisor { const name = entry.runtime.name; entry.phase = "starting"; entry.error = undefined; + entry.completed = false; const controller = new AbortController(); const onParentAbort = () => controller.abort(this.config.signal.reason); @@ -195,6 +209,7 @@ export class DevSupervisor { try { const port = await this.config.resolvePort(entry.runtime); const env = await this.config.getDevEnvVarsForRuntime(entry.runtime); + controller.signal.throwIfAborted(); const runner = this.config.runners[entry.runtime.build]; let ready = false; @@ -207,14 +222,24 @@ export class DevSupervisor { }); entry.running = pump; const earlyExit = pump.finally(unchain).then(() => { - if (!ready) - throw new Error(entry.error ?? `Agent '${name}' exited before it became ready.`); + if (!ready) { + if (entry.error) throw entry.error; + throw new Error(`Agent '${name}' exited before it became ready.`); + } }); // Both branches outlive the race (the pump runs for the agent's lifetime); // swallow their late rejections so losing branches never become unhandled. readiness.catch(() => {}); earlyExit.catch(() => {}); await Promise.race([readiness, earlyExit]); + if (entry.error) throw entry.error; + + if (entry.completed) { + entry.phase = "idle"; + entry.port = undefined; + this.push(name, { type: "status", message: `Agent '${name}' stopped.` }); + return { name, port }; + } entry.phase = "running"; entry.port = port; @@ -224,10 +249,12 @@ export class DevSupervisor { controller.abort(); unchain(); // idempotent alongside the pump's cleanup; covers setup failures before the pump exists entry.phase = "failed"; - entry.error = error instanceof Error ? error.message : String(error); + const agentError = + error instanceof Error ? error : new Error(String(error), { cause: error }); + entry.error = agentError; this.push(name, { type: "status", - message: `Agent '${name}' failed to start: ${entry.error}`, + message: `Agent '${name}' failed to start: ${agentError.message}`, }); throw error; } @@ -256,6 +283,8 @@ export class DevSupervisor { entry.phase = "idle"; entry.port = undefined; this.push(name, { type: "status", message: `Agent '${name}' stopped.` }); + } else if (entry.phase === "starting") { + entry.completed = true; } } catch (error) { /** The runner rejects with the abort reason on teardown, which is a stop, not a crash. */ @@ -267,12 +296,16 @@ export class DevSupervisor { } return; } - const message = error instanceof Error ? error.message : String(error); - entry.error = message; + const agentError = + error instanceof Error ? error : new Error(String(error), { cause: error }); + entry.error = agentError; if (entry.phase === "running") { entry.phase = "failed"; entry.port = undefined; - this.push(name, { type: "status", message: `Agent '${name}' crashed: ${message}` }); + this.push(name, { + type: "status", + message: `Agent '${name}' crashed: ${agentError.message}`, + }); } } } diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index de3603cdd1..3a12d98cb5 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -8,15 +8,14 @@ import { projectSpecPath } from "../../../core/project/fsUtils"; import { DevSupervisor, type SupervisorConfig } from "../../../core/dev/supervisor"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { - AgentCoreCLIError, ERROR_SOURCE, InputValidationError, NotImplementedError, ResourceNotFoundError, SilentCLIError, - UserCancellationError, } from "../../../errors"; import type { AppIO, BrowserOpener, FileWatcher, PortChecker, startHttpServer } from "../../../io"; +import { withUserCancellation } from "../../../runnable"; import { createHandler, flag, ProjectKey, type Middleware } from "../../../router"; import { JsonRendererKey, type JsonRenderer } from "../../../tui"; import { JsonKey, RegionKey } from "../../keys"; @@ -133,238 +132,181 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => ), ], handle: async (ctx, flags) => { - const controller = new AbortController(); const json = ctx.require(JsonKey) ? ctx.require(JsonRendererKey) : undefined; - const interrupt = () => { - if (controller.signal.aborted) return; - config.io.stderr.write("Shutting down…\n"); - controller.abort(new UserCancellationError()); - }; - - const signals = ["SIGINT", "SIGTERM"] as const; - for (const signal of signals) process.on(signal, interrupt); - let collector: DevTraceCollector | undefined; - try { - const project = ctx.require(ProjectKey); - const region = ctx.require(RegionKey); - const runtimes = selectRuntimes(project, flags.agent); - if (runtimes.length > 1 && flags.port !== undefined) { - throw new InputValidationError( - "--port applies to a single runtime. Use --agent to select one.", - ); - } - if (!flags.agent) { - for (const runtime of project.spec.runtimes.filter( - (runtime) => !supportsLocalDev(runtime), - )) { - renderStatus( - config.io, - `Skipping runtime '${runtime.name}': local dev is not supported for ${BMA_TEMPLATE_NAME}.`, - json, - ); - } - } - - if ( - flags.traces && - runtimes.some((runtime) => runtime.instrumentation?.enableOtel ?? true) - ) { - const tracesDirectory = join(project.rootPath, "agentcore", ".cli", "traces", "otlp"); - let tracePersistErrorReported = false; - collector = await config.startTraceCollector({ - tracesDirectory, - // A container reaches the collector over the host bridge, which a - // 127.0.0.1 bind refuses, so bind all interfaces when any runtime - // is a container. - host: runtimes.some((runtime) => runtime.build === "Container") - ? "0.0.0.0" - : "127.0.0.1", - // Persistence can fail after startup (disk, permissions). Warn once — - // exports are still acked, so without this the loss would be silent. - onError: (error) => { - if (tracePersistErrorReported) return; - tracePersistErrorReported = true; - const detail = error instanceof Error ? error.message : String(error); + const project = ctx.require(ProjectKey); + const region = ctx.require(RegionKey); + await withUserCancellation( + async (signal) => { + let collector: DevTraceCollector | undefined; + try { + const runtimes = selectRuntimes(project, flags.agent); + if (runtimes.length > 1 && flags.port !== undefined) { + throw new InputValidationError( + "--port applies to a single runtime. Use --agent to select one.", + ); + } + if (!flags.agent) { + for (const runtime of project.spec.runtimes.filter( + (runtime) => !supportsLocalDev(runtime), + )) { + renderStatus( + config.io, + `Skipping runtime '${runtime.name}': local dev is not supported for ${BMA_TEMPLATE_NAME}.`, + json, + ); + } + } + if ( + flags.traces && + runtimes.some((runtime) => runtime.instrumentation?.enableOtel ?? true) + ) { + const tracesDirectory = join(project.rootPath, "agentcore", ".cli", "traces", "otlp"); + let tracePersistErrorReported = false; + collector = await config.startTraceCollector({ + tracesDirectory, + // A container reaches the collector over the host bridge, which a + // 127.0.0.1 bind refuses, so bind all interfaces when any runtime + // is a container. + host: runtimes.some((runtime) => runtime.build === "Container") + ? "0.0.0.0" + : "127.0.0.1", + // Persistence can fail after startup (disk, permissions). Warn once — + // exports are still acked, so without this the loss would be silent. + onError: (error) => { + if (tracePersistErrorReported) return; + tracePersistErrorReported = true; + const detail = error instanceof Error ? error.message : String(error); + renderStatus( + config.io, + `Warning: failed to persist traces to ${tracesDirectory} (${detail}); collected traces may be incomplete.`, + json, + ); + }, + }); renderStatus( config.io, - `Warning: failed to persist traces to ${tracesDirectory} (${detail}); collected traces may be incomplete.`, + `OTEL collector listening on port ${collector.port}; traces persist to ${tracesDirectory}.`, json, ); - }, - }); - renderStatus( - config.io, - `OTEL collector listening on port ${collector.port}; traces persist to ${tracesDirectory}.`, - json, - ); - } - controller.signal.throwIfAborted(); - - const getDevEnvVarsForRuntime = async ( - runtime: ProjectRuntime, - ): Promise> => { - const { env } = await config.loadDevEnvironment({ - projectRoot: project.rootPath, - runtime, - region, - }); - const otel = - collector && (runtime.instrumentation?.enableOtel ?? true) - ? otelEnvForRuntime(collector, runtime) - : {}; - return { ...env, ...otel }; - }; + } + signal.throwIfAborted(); - if (flags.mode === "headless" && flags.agent) { - await runWithoutUi( - config, - runtimes[0]!, - project, - flags.port, - getDevEnvVarsForRuntime, - controller, - json, - ); - return; - } + const assignedPorts = + flags.mode === "headless" && !flags.agent + ? await resolveDevPorts(runtimes, flags.port, config.checkPort, signal) + : undefined; + const supervisor = new DevSupervisor({ + runtimes, + projectRoot: project.rootPath, + runners: config.runners, + getDevEnvVarsForRuntime: async (runtime) => { + const { env } = await config.loadDevEnvironment({ + projectRoot: project.rootPath, + runtime, + region, + }); + return collector && (runtime.instrumentation?.enableOtel ?? true) + ? { ...env, ...otelEnvForRuntime(collector, runtime) } + : { ...env }; + }, + // Runtime selection rejects an explicit port for multiple runtimes. + resolvePort: async (runtime) => { + if (assignedPorts) return assignedPorts.get(runtime.name)!; + const resolved = await resolveDevPort( + runtime.protocol, + flags.port, + config.checkPort, + signal, + ); + if ( + flags.mode === "headless" && + flags.agent && + resolved.port !== resolved.requestedPort + ) { + renderStatus( + config.io, + `Port ${resolved.requestedPort} is in use; using ${resolved.port}.`, + json, + ); + } + return resolved.port; + }, + // Selected headless dev streams the runner without a readiness deadline. + waitReady: + flags.mode === "headless" && flags.agent ? async () => {} : config.waitReady, + signal, + }); - const assignedPorts = - flags.mode === "headless" - ? await resolveDevPorts(runtimes, flags.port, config.checkPort, controller.signal) - : undefined; - const supervisor = new DevSupervisor({ - runtimes, - projectRoot: project.rootPath, - runners: config.runners, - getDevEnvVarsForRuntime, - // The --port guard above rejects an explicit port with more than one - // runtime, so passing flags.port here only ever applies to a lone one. - resolvePort: async (runtime) => { - if (assignedPorts) { - const assignedPort = assignedPorts.get(runtime.name); - if (assignedPort === undefined) { - throw new AgentCoreCLIError(`No port was assigned to runtime '${runtime.name}'.`); + if (flags.mode === "headless") { + void Promise.allSettled(runtimes.map((runtime) => supervisor.start(runtime.name))); + for await (const { agentName, event } of supervisor.events({ untilStopped: true })) { + const message = event.type === "status" ? event.message : undefined; + const lifecycleStatus = + flags.agent !== undefined && + (message === `Agent '${agentName}' stopped.` || + message?.startsWith(`Agent '${agentName}' is running on port `) || + message?.startsWith(`Agent '${agentName}' failed to start: `) || + message?.startsWith(`Agent '${agentName}' crashed: `)); + if (!lifecycleStatus) renderAgentEvent(config.io, event, agentName, json); } - return assignedPort; + signal.throwIfAborted(); + const failure = supervisor.snapshot().find(({ phase }) => phase === "failed"); + if (failure) throw flags.agent ? failure.error : new SilentCLIError(); + return; } - return ( - await resolveDevPort( - runtime.protocol, - flags.port, - config.checkPort, - controller.signal, - ) - ).port; - }, - waitReady: config.waitReady, - signal: controller.signal, - }); - - if (flags.mode === "headless") { - void Promise.allSettled(runtimes.map((runtime) => supervisor.start(runtime.name))); - for await (const { agentName, event } of supervisor.events()) { - renderAgentEvent(config.io, event, agentName, json); - const phases = supervisor.snapshot(); - if (phases.every(({ phase }) => phase !== "starting" && phase !== "running")) { - if (phases.some(({ phase }) => phase === "failed")) throw new SilentCLIError(); - break; - } - } - controller.signal.throwIfAborted(); - return; - } - const uiPort = ( - await findFreePort(UI_DEFAULT_PORT, flags["ui-port"], config.checkPort, controller.signal) - ).port; - const server = await config.startServer( - createInspectorHandler({ - supervisor, - traces: collector?.traces, - assets: config.inspectorAssets, - project, - selectedAgent: flags.agent, - }), - { port: uiPort, signal: controller.signal }, - ); + const uiPort = ( + await findFreePort(UI_DEFAULT_PORT, flags["ui-port"], config.checkPort, signal) + ).port; + const server = await config.startServer( + createInspectorHandler({ + supervisor, + traces: collector?.traces, + assets: config.inspectorAssets, + project, + selectedAgent: flags.agent, + }), + { port: uiPort, signal }, + ); - const onConfigChange = async () => { - try { - const reloaded = await config.projectManager.resolve({ filePath: project.rootPath }); - if (!reloaded) return; - const runtimes = reloaded.spec.runtimes.filter(supportsLocalDev); - supervisor.setRuntimes( - flags.agent ? runtimes.filter((runtime) => runtime.name === flags.agent) : runtimes, + const onConfigChange = async () => { + try { + const reloaded = await config.projectManager.resolve({ + filePath: project.rootPath, + }); + if (!reloaded) return; + const runtimes = reloaded.spec.runtimes.filter(supportsLocalDev); + supervisor.setRuntimes( + flags.agent + ? runtimes.filter((runtime) => runtime.name === flags.agent) + : runtimes, + ); + renderStatus(config.io, "Reloaded agents from agentcore.json.", json); + } catch { + // A half-saved config parses on the next change event. + } + }; + config.watchFile( + projectSpecPath(project.rootPath), + () => void onConfigChange(), + signal, ); - renderStatus(config.io, "Reloaded agents from agentcore.json.", json); - } catch { - // A half-saved config parses on the next change event. - } - }; - config.watchFile( - projectSpecPath(project.rootPath), - () => void onConfigChange(), - controller.signal, - ); - const url = `http://127.0.0.1:${server.port}`; - renderStatus(config.io, `Agent Inspector running at ${url}`, json); - if (config.isInteractive() && !json) await config.openBrowser(url); + const url = `http://127.0.0.1:${server.port}`; + renderStatus(config.io, `Agent Inspector running at ${url}`, json); + if (config.isInteractive() && !json) await config.openBrowser(url); - for await (const { agentName, event } of supervisor.events()) { - renderAgentEvent(config.io, event, agentName, json); - } - controller.signal.throwIfAborted(); - } catch (error) { - controller.signal.throwIfAborted(); - throw error; - } finally { - for (const signal of signals) process.removeListener(signal, interrupt); - // Close only after the runner returns, which is after the child's own - // shutdown grace, so the agent's final spans still reach the collector. - await collector?.close(); - } + for await (const { agentName, event } of supervisor.events()) { + renderAgentEvent(config.io, event, agentName, json); + } + signal.throwIfAborted(); + } finally { + // Close only after the runner returns, which is after the child's own + // shutdown grace, so the agent's final spans still reach the collector. + await collector?.close(); + } + }, + { onCancel: () => config.io.stderr.write("Shutting down…\n") }, + ); }, }); - -/** - * Run one runtime directly. Unlike the supervised Inspector path, a crash here - * fails the command (scripts and CI rely on the non-zero exit). - */ -async function runWithoutUi( - config: DevProjectHandlerConfig, - runtime: ProjectRuntime, - project: Project, - explicitPort: number | undefined, - environment: (runtime: ProjectRuntime) => Promise>, - controller: AbortController, - json?: JsonRenderer, -): Promise { - const devPort = await resolveDevPort( - runtime.protocol, - explicitPort, - config.checkPort, - controller.signal, - ); - if (devPort.port !== devPort.requestedPort) { - renderStatus( - config.io, - `Port ${devPort.requestedPort} is in use; using ${devPort.port}.`, - json, - ); - } - - const env = await environment(runtime); - controller.signal.throwIfAborted(); - - const runner = config.runners[runtime.build]; - for await (const event of runner.run({ - runtime, - projectRoot: project.rootPath, - port: devPort.port, - env, - signal: controller.signal, - })) { - renderAgentEvent(config.io, event, runtime.name, json); - } -} diff --git a/src/runnable/index.tsx b/src/runnable/index.tsx index 1cd2401671..a9a16262a3 100644 --- a/src/runnable/index.tsx +++ b/src/runnable/index.tsx @@ -1,11 +1,19 @@ import { AgentCoreCLIError, ExitCode, SilentCLIError, UserCancellationError } from "../errors"; -/** Runs a headless operation with process SIGINT mapped to UserCancellationError. */ -export async function withUserCancellation(fn: (signal: AbortSignal) => Promise): Promise { +/** Runs a headless operation with process interrupts mapped to UserCancellationError. */ +export async function withUserCancellation( + fn: (signal: AbortSignal) => Promise, + options: { onCancel?: () => void } = {}, +): Promise { const controller = new AbortController(); - const interrupt = () => controller.abort(new UserCancellationError()); + const signals = ["SIGINT", "SIGTERM"] as const; + const interrupt = () => { + if (controller.signal.aborted) return; + options.onCancel?.(); + controller.abort(new UserCancellationError()); + }; // Ink's signal-exit handler must see our listener until cancellation cleanup settles. - process.on("SIGINT", interrupt); + for (const signal of signals) process.on(signal, interrupt); try { const result = await fn(controller.signal); controller.signal.throwIfAborted(); @@ -15,7 +23,7 @@ export async function withUserCancellation(fn: (signal: AbortSignal) => Promi throw error; } finally { controller.abort(); - process.off("SIGINT", interrupt); + for (const signal of signals) process.off(signal, interrupt); } }