diff --git a/src/core/dev/inspector/invocations.ts b/src/core/dev/inspector/invocations.ts index b3056dcbe..313fe6a08 100644 --- a/src/core/dev/inspector/invocations.ts +++ b/src/core/dev/inspector/invocations.ts @@ -181,7 +181,7 @@ async function* transformA2aSse( for await (const data of sseData(stream)) { try { const event = JSON.parse(data) as Record; - const { text, kind } = extractSseEventText(event, streamedFromStatus); + const { text, kind } = extractA2aEventText(event, streamedFromStatus); if (text) { if (kind === "status-update") streamedFromStatus = true; yield sseEvent(text); @@ -193,7 +193,7 @@ async function* transformA2aSse( } // When streamedFromStatus is set, artifact-update text is skipped because status-update already streamed it. -function extractSseEventText( +function extractA2aEventText( event: Record, streamedFromStatus: boolean, ): { text: string | null; kind: string | undefined } { diff --git a/src/core/dev/otel/collector.test.ts b/src/core/dev/otel/collector.test.ts index b9ae3fa66..9153424fd 100644 --- a/src/core/dev/otel/collector.test.ts +++ b/src/core/dev/otel/collector.test.ts @@ -90,7 +90,7 @@ describe("startOtelCollector", () => { const response = await post("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/v1/traces", protobufTracePayload()); expect(response.status).toBe(200); - const traces = await collector.store.list(); + const traces = await collector.traces.list(); expect(traces).toHaveLength(1); expect(traces[0]!.traceId).toBe(TRACE_ID_HEX); }); @@ -100,7 +100,7 @@ describe("startOtelCollector", () => { const response = await post("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/v1/logs", protobufLogsPayload()); expect(response.status).toBe(200); - const detail = await collector.store.get(TRACE_ID_HEX); + const detail = await collector.traces.get(TRACE_ID_HEX); expect(detail?.resourceLogs).toBeDefined(); }); @@ -129,13 +129,13 @@ describe("startOtelCollector", () => { }); const response = await post("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/v1/traces", body, "application/json"); expect(response.status).toBe(200); - expect((await collector.store.list()).map((trace) => trace.traceId)).toEqual([TRACE_ID_HEX]); + expect((await collector.traces.list()).map((trace) => trace.traceId)).toEqual([TRACE_ID_HEX]); }); test("rejects malformed payloads with 400", async () => { expect((await post("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/v1/traces", "not json", "application/json")).status).toBe(400); expect((await post("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/v1/traces", Buffer.from([0xff, 0xff, 0xff]))).status).toBe(400); - expect(await collector.store.list()).toEqual([]); + expect(await collector.traces.list()).toEqual([]); }); test.each(["null", "[]", "42", '{"resourceSpans":5}'])( @@ -154,7 +154,7 @@ describe("startOtelCollector", () => { }); expect(response.status).toBe(400); expect(errors).toEqual([]); - expect(await strict.store.list()).toEqual([]); + expect(await strict.traces.list()).toEqual([]); } finally { await strict.close(); } diff --git a/src/core/dev/otel/collector.ts b/src/core/dev/otel/collector.ts index 8775aeca1..673dccb55 100644 --- a/src/core/dev/otel/collector.ts +++ b/src/core/dev/otel/collector.ts @@ -36,7 +36,7 @@ export interface OtelCollector { /** The port the OTLP/HTTP receiver listens on. */ port: number; /** Reads the traces this collector persists. */ - store: TraceStore; + traces: TraceStore; /** Environment variables that point an agent's OTEL SDK at this collector. */ envVars: Record; /** Stops the receiver. Also invoked by the start signal, if one was given. */ @@ -68,7 +68,12 @@ export async function startOtelCollector( signal: options.signal, }); - return { port: server.port, store, envVars: otelEnvVars(server.port), close: server.close }; + return { + port: server.port, + traces: store, + envVars: otelEnvVars(server.port), + close: server.close, + }; } async function route( diff --git a/src/core/dev/port.ts b/src/core/dev/port.ts index 1b88425ec..231959479 100644 --- a/src/core/dev/port.ts +++ b/src/core/dev/port.ts @@ -26,7 +26,19 @@ export async function resolveDevPort( checkPort: PortChecker, signal: AbortSignal, ): Promise { - const defaultPort = DEV_PORTS[protocol ?? "HTTP"]; + return findFreePort(DEV_PORTS[protocol ?? "HTTP"], explicitPort, checkPort, signal); +} + +/** + * Resolve a free port from `defaultPort`. An explicit port must be free or the + * call fails; otherwise the next free port from the default up is taken. + */ +export async function findFreePort( + defaultPort: number, + explicitPort: number | undefined, + checkPort: PortChecker, + signal: AbortSignal, +): Promise { const requestedPort = explicitPort ?? defaultPort; if (await checkPort(requestedPort, signal)) { diff --git a/src/core/dev/supervisor.test.ts b/src/core/dev/supervisor.test.ts index a41797f61..7df6ac9d1 100644 --- a/src/core/dev/supervisor.test.ts +++ b/src/core/dev/supervisor.test.ts @@ -22,6 +22,7 @@ function serverRunner(events: DevEvent[] = []) { run: async function* (input) { inputs.push(input); yield* events; + if (input.signal.aborted) return; await new Promise((resolve) => input.signal.addEventListener("abort", () => resolve(), { once: true }), ); @@ -360,5 +361,22 @@ describe("DevSupervisor", () => { await consuming; expect(codeZip.inputs[0]!.signal.aborted).toBe(true); + // events() waits for the child's pump before ending, so the agent's final + // "stopped" event is drained rather than lost when the collector closes. + expect(events).toContainEqual({ + agentName: "orders", + event: { type: "status", message: "Agent 'orders' stopped." }, + }); + }); + + test("an edit to a running agent applies on its next start, not live", async () => { + const { supervisor, controller } = harness(); + await supervisor.start("orders"); + + supervisor.setRuntimes([{ ...runtime("orders"), protocol: "A2A" } as ProjectRuntime]); + // The live process keeps its protocol, so the Inspector never proxies it with + // metadata that no longer matches the running child. + expect(supervisor.running("orders")).toEqual({ port: 9100, protocol: "HTTP" }); + controller.abort(); }); }); diff --git a/src/core/dev/supervisor.ts b/src/core/dev/supervisor.ts index d39ab3f57..eed5dded6 100644 --- a/src/core/dev/supervisor.ts +++ b/src/core/dev/supervisor.ts @@ -43,6 +43,10 @@ type AgentEntry = { port?: number; error?: string; starting?: Promise<{ name: string; port: number }>; + /** The running child's pump, so shutdown can await its final spans. */ + running?: Promise; + /** A reloaded definition held for a live agent, applied on its next start. */ + pendingRuntime?: ProjectRuntime; }; /** @@ -79,8 +83,15 @@ export class DevSupervisor { const names = new Set(runtimes.map((runtime) => runtime.name)); for (const runtime of runtimes) { const existing = this.agents.get(runtime.name); - if (existing) existing.runtime = runtime; - else this.agents.set(runtime.name, { runtime, phase: "idle" }); + if (!existing) { + this.agents.set(runtime.name, { runtime, phase: "idle" }); + } else if (existing.phase === "running" || existing.phase === "starting") { + // A live process keeps its current definition; the edit applies on next + // start, so the Inspector never proxies a running agent with stale metadata. + existing.pendingRuntime = runtime; + } else { + existing.runtime = runtime; + } } for (const [name, entry] of this.agents) { if (!names.has(name) && entry.phase !== "running" && entry.phase !== "starting") { @@ -128,6 +139,10 @@ export class DevSupervisor { } if (entry.starting) return entry.starting; + if (entry.pendingRuntime) { + entry.runtime = entry.pendingRuntime; + entry.pendingRuntime = undefined; + } entry.starting = this.launch(entry).finally(() => { entry.starting = undefined; }); @@ -141,7 +156,14 @@ export class DevSupervisor { public async *events(): AsyncGenerator { while (true) { for (const event of this.queue.splice(0)) yield event; - if (this.config.signal.aborted) return; + 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); + for (const event of this.queue.splice(0)) yield event; + return; + } await new Promise((resolve) => { this.wake = resolve; // A push during the yields above ran while wake was undefined, so its @@ -180,14 +202,14 @@ export class DevSupervisor { const readiness = this.waitReady(port, controller.signal, () => activity.at).then(() => { ready = true; }); - const earlyExit = this.pump(entry, runner, { port, env, signal: controller.signal }, () => { + const pump = this.pump(entry, runner, { port, env, signal: controller.signal }, () => { activity.at = Date.now(); - }) - .finally(unchain) - .then(() => { - if (!ready) - throw new Error(entry.error ?? `Agent '${name}' exited before it became ready.`); - }); + }); + entry.running = pump; + const earlyExit = pump.finally(unchain).then(() => { + if (!ready) + throw new Error(entry.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(() => {}); diff --git a/src/core/project/fsUtils.ts b/src/core/project/fsUtils.ts index 15019f424..a9e6a4a15 100644 --- a/src/core/project/fsUtils.ts +++ b/src/core/project/fsUtils.ts @@ -1,10 +1,18 @@ import { existsSync } from "node:fs"; import { dirname, join } from "node:path"; +/** The project spec, relative to the project root. */ +export const PROJECT_SPEC_RELATIVE_PATH = join("agentcore", "agentcore.json"); + +/** The project spec's absolute path under `rootPath`. */ +export function projectSpecPath(rootPath: string): string { + return join(rootPath, PROJECT_SPEC_RELATIVE_PATH); +} + /** Walks up from directory looking for the agentcore/agentcore.json project marker. */ export function enclosingProjectRoot(directory: string): string | undefined { for (let current = directory; ; current = dirname(current)) { - if (existsSync(join(current, "agentcore", "agentcore.json"))) { + if (existsSync(join(current, PROJECT_SPEC_RELATIVE_PATH))) { return current; } if (dirname(current) === current) { diff --git a/src/core/project/manager.tsx b/src/core/project/manager.tsx index d34a07980..e53f9ccf0 100644 --- a/src/core/project/manager.tsx +++ b/src/core/project/manager.tsx @@ -32,7 +32,7 @@ import { CredentialSchema } from "../../projectSchemas/credential"; import { MemorySchema } from "../../projectSchemas/memory"; import { OnlineEvalConfigSchema } from "../../projectSchemas/online-eval-config"; import { PolicyEngineSchema, PolicySchema } from "../../projectSchemas/policy"; -import { enclosingProjectRoot } from "./fsUtils"; +import { enclosingProjectRoot, projectSpecPath } from "./fsUtils"; import { AgentCoreCLIError, InputValidationError, @@ -92,7 +92,7 @@ export class FsProjectManager implements ProjectManager { const rootPath = enclosingProjectRoot(input.filePath); if (!rootPath) return undefined; - const configPath = join(rootPath, "agentcore", "agentcore.json"); + const configPath = projectSpecPath(rootPath); const spec = await this.json.read(configPath, ProjectSpecSchema); return { name: spec.name, @@ -344,7 +344,7 @@ export class FsProjectManager implements ProjectManager { } private getProjectSpecPath(project: Project): string { - return join(project.rootPath, "agentcore", "agentcore.json"); + return projectSpecPath(project.rootPath); } public async removeResource(project: Project, input: RemoveResourceInput): Promise { diff --git a/src/handlers/project/dev/index.test.ts b/src/handlers/project/dev/index.test.ts index 9bb4979dc..461d83ffe 100644 --- a/src/handlers/project/dev/index.test.ts +++ b/src/handlers/project/dev/index.test.ts @@ -4,10 +4,9 @@ import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { InputValidationError, ResourceNotFoundError, - SilentCLIError, UserCancellationError, } from "../../../errors"; -import type { PortChecker } from "../../../io"; +import type { HttpRequestHandler, PortChecker } from "../../../io"; import { ProjectKey, ValueContext } from "../../../router"; import { testIO } from "../../../testing"; import { JsonRendererKey } from "../../../tui"; @@ -35,7 +34,6 @@ function project(...runtimes: ProjectRuntime[]): Project { }; } -/** A runner that emits `events` then exits, so the supervisor marks it failed to start. */ function captureRunner(events: DevEvent[] = []) { const inputs: DevServerInput[] = []; const runner: DevRunner = { @@ -47,21 +45,6 @@ function captureRunner(events: DevEvent[] = []) { return { runner, inputs }; } -/** A runner that emits `events` then stays alive until aborted, like a real dev server. */ -function stayingRunner(events: DevEvent[] = []) { - const inputs: DevServerInput[] = []; - const runner: DevRunner = { - run: async function* (input) { - inputs.push(input); - yield* events; - await new Promise((resolve) => - input.signal.addEventListener("abort", () => resolve(), { once: true }), - ); - }, - }; - return { runner, inputs }; -} - function fakeCollector() { const starts: Parameters[0][] = []; const state = { closed: 0 }; @@ -72,6 +55,7 @@ function fakeCollector() { OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "http://127.0.0.1:43180/v1/traces", OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf", }, + traces: { list: async () => [], get: async () => undefined }, close: async () => { state.closed++; }, @@ -85,6 +69,8 @@ function fakeCollector() { type HarnessOptions = { project?: Project; + tty?: boolean; + reloadedRuntimes?: ProjectRuntime[]; codeZip?: ReturnType; container?: ReturnType; checkPort?: PortChecker; @@ -94,8 +80,11 @@ type HarnessOptions = { function harness(options: HarnessOptions = {}) { const io = testIO(); - const codeZip = options.codeZip ?? stayingRunner(); - const container = options.container ?? stayingRunner(); + const ui = { starts: [] as { port?: number }[], opened: [] as string[], closed: 0 }; + const watchers: { path: string; onChange: () => void }[] = []; + let capturedHandler: HttpRequestHandler | undefined; + const codeZip = options.codeZip ?? captureRunner(); + const container = options.container ?? captureRunner(); const collector = fakeCollector(); const environmentInputs: DevEnvironmentInput[] = []; const handler = createDevProjectHandler({ @@ -109,10 +98,27 @@ function harness(options: HarnessOptions = {}) { }), checkPort: options.checkPort ?? (async () => true), startTraceCollector: collector.start, - // A staying agent is ready after this delay; one that exits sooner loses the - // race and is reported failed, so tests never bind a real port. - waitReady: async () => { - await Bun.sleep(20); + startServer: async (requestHandler, serverOptions) => { + capturedHandler = requestHandler; + ui.starts.push({ port: serverOptions?.port }); + return { + port: serverOptions?.port ?? 8081, + close: async () => { + ui.closed++; + }, + }; + }, + openBrowser: async (url) => { + ui.opened.push(url); + }, + inspectorAssets: { read: async () => undefined }, + isInteractive: () => options.tty ?? false, + watchFile: (path, onChange) => { + watchers.push({ path, onChange }); + }, + projectManager: { + resolve: async () => + options.reloadedRuntimes ? project(...options.reloadedRuntimes) : undefined, }, }); const ctx = ValueContext.EmptyContext() @@ -130,33 +136,49 @@ function harness(options: HarnessOptions = {}) { collector, environmentInputs, io, - run: (flags: { agent?: string; port?: number; traces?: boolean } = {}) => - handler.handle(ctx, { traces: true, ...flags }, {}), + ui, + watchers, + inspectorHandler: () => capturedHandler, + run: ( + flags: { + agent?: string; + port?: number; + traces?: boolean; + mode?: "browser" | "headless" | "tui"; + "ui-port"?: number; + } = {}, + ) => handler.handle(ctx, { traces: true, mode: "headless", ...flags }, {}), }; } -/** - * Start a supervised run and let its agents reach "running". Returns the pending - * promise wrapped, so awaiting this helper does not flatten into the run itself. - */ -async function supervised( - subject: ReturnType, - flags = {}, -): Promise<{ pending: Promise }> { - const pending = subject.run(flags); - pending.catch(() => undefined); - await Bun.sleep(50); - return { pending }; -} - -async function interrupt(pending: Promise): Promise { - process.emit("SIGINT", "SIGINT"); - await pending.catch(() => undefined); +/** Ask the captured Inspector handler for the current agent status. */ +async function inspectorStatus(subject: ReturnType): Promise<{ name: string }[]> { + const response = await subject.inspectorHandler()!({ + method: "GET", + url: "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/api/status", + headers: { host: "127.0.0.1:8081" }, + body: Buffer.alloc(0), + signal: new AbortController().signal, + }); + const status = JSON.parse(String(response.body)) as { agents: { name: string }[] }; + return status.agents; } describe("project dev selection and dispatch", () => { test.each([ [project(), {}, "This project has no runtimes", InputValidationError], + [ + project(runtime("orders")), + {}, + "--mode headless runs a single agent in the terminal. Pass --agent to choose which one. Available: orders", + InputValidationError, + ], + [ + project(runtime("orders"), runtime("support", "Container")), + {}, + "--mode headless runs a single agent in the terminal. Pass --agent to choose which one. Available: orders, support", + InputValidationError, + ], [ project(runtime("orders"), runtime("support", "Container")), { agent: "missing" }, @@ -176,7 +198,7 @@ describe("project dev selection and dispatch", () => { const subject = harness({ project: project(runtime("orders"), runtime("support", "Container")), }); - const { pending } = await supervised(subject, { agent: "support", port: 4567 }); + await subject.run({ agent: "support", port: 4567 }); expect(subject.codeZip.inputs).toHaveLength(0); expect(subject.environmentInputs).toEqual([ @@ -197,10 +219,9 @@ describe("project dev selection and dispatch", () => { }, runtime: { name: "support", build: "Container" }, }); - await interrupt(pending); }); - test("resolves and announces an automatically selected port", async () => { + test("announces an automatically selected port", async () => { const checked: number[] = []; const subject = harness({ checkPort: async (port) => { @@ -208,66 +229,18 @@ describe("project dev selection and dispatch", () => { return port === 8081; }, }); - const { pending } = await supervised(subject); + await subject.run({ agent: "orders" }); expect(checked).toEqual([8080, 8081]); expect(subject.codeZip.inputs[0]?.port).toBe(8081); - expect(subject.io.stderr()).toContain("Agent 'orders' is running on port 8081"); - await interrupt(pending); - }); -}); - -describe("project dev multi-agent supervision", () => { - const twoRuntimes = () => project(runtime("orders"), runtime("support", "Container")); - - test("supervises every runtime with attributed output and per-runtime env", async () => { - const codeZip = stayingRunner([{ type: "stdout", line: "orders says hi" }]); - const container = stayingRunner(); - const subject = harness({ project: twoRuntimes(), codeZip, container }); - const { pending } = await supervised(subject); - - expect(codeZip.inputs).toHaveLength(1); - expect(container.inputs).toHaveLength(1); - expect(codeZip.inputs[0]!.env).toMatchObject({ - OTEL_EXPORTER_OTLP_ENDPOINT: "http://127.0.0.1:43180", - OTEL_SERVICE_NAME: "orders", - }); - expect(container.inputs[0]!.env).toMatchObject({ - OTEL_EXPORTER_OTLP_ENDPOINT: "http://host.docker.internal:43180", - OTEL_SERVICE_NAME: "support", - }); - expect(subject.io.stdout()).toContain("[orders] orders says hi"); - expect(subject.io.stderr()).toContain("Agent 'orders' is running on port"); - - process.emit("SIGINT", "SIGINT"); - await expect(pending).rejects.toMatchObject({ exitCode: 130 }); - expect(subject.collector.state.closed).toBe(1); - }); - - test("one agent failing to start does not stop the others", async () => { - const subject = harness({ - project: twoRuntimes(), - codeZip: captureRunner([{ type: "status", message: "dying" }]), // exits: never ready - container: stayingRunner(), - }); - const { pending } = await supervised(subject); - - expect(subject.io.stderr()).toContain("[orders] Agent 'orders' failed to start"); - expect(subject.io.stderr()).toContain("Agent 'support' is running on port"); - await interrupt(pending); - }); - - test("--port without --agent is rejected when several runtimes exist", async () => { - await expect(harness({ project: twoRuntimes() }).run({ port: 4567 })).rejects.toThrow( - "--port applies to a single runtime", - ); + expect(subject.io.stderr()).toContain("Port 8080 is in use; using 8081."); }); }); describe("project dev trace collection", () => { test("starts the collector, announces it, and points a CodeZip agent at loopback", async () => { const subject = harness(); - const { pending } = await supervised(subject); + await subject.run({ agent: "orders" }); expect(subject.collector.starts).toEqual([ { @@ -281,21 +254,19 @@ describe("project dev trace collection", () => { OTEL_EXPORTER_OTLP_ENDPOINT: "http://127.0.0.1:43180", OTEL_SERVICE_NAME: "orders", }); - await interrupt(pending); expect(subject.collector.state.closed).toBe(1); }); test("binds the collector to all interfaces so a container can reach it", async () => { const subject = harness({ project: project(runtime("support", "Container")) }); - const { pending } = await supervised(subject); + await subject.run({ agent: "support" }); expect(subject.collector.starts[0]?.host).toBe("0.0.0.0"); - await interrupt(pending); }); test("reports a trace-persistence failure once, not per failed export", async () => { const subject = harness(); - const { pending } = await supervised(subject); + await subject.run({ agent: "orders" }); const onError = subject.collector.starts[0]?.onError; onError?.(new Error("disk full")); @@ -305,29 +276,26 @@ describe("project dev trace collection", () => { expect(stderr).toContain("failed to persist traces"); expect(stderr).toContain("disk full"); expect(stderr.match(/failed to persist traces/g)).toHaveLength(1); - await interrupt(pending); }); test("--no-traces skips the collector entirely", async () => { const subject = harness(); - const { pending } = await supervised(subject, { traces: false }); + await subject.run({ agent: "orders", traces: false }); expect(subject.collector.starts).toHaveLength(0); expect(subject.codeZip.inputs[0]?.env).toEqual({ FROM_LOADER: "yes" }); - await interrupt(pending); }); test("a runtime with instrumentation disabled skips the collector", async () => { const disabled = { ...runtime(), instrumentation: { enableOtel: false } } as ProjectRuntime; const subject = harness({ project: project(disabled) }); - const { pending } = await supervised(subject); + await subject.run({ agent: "orders" }); expect(subject.collector.starts).toHaveLength(0); expect(subject.codeZip.inputs[0]?.env).toEqual({ FROM_LOADER: "yes" }); - await interrupt(pending); }); - test("a failed agent exits non-zero and still closes the collector", async () => { + test("the collector is closed when the runner fails", async () => { const codeZip = captureRunner(); codeZip.runner.run = async function* () { yield* []; @@ -335,9 +303,108 @@ describe("project dev trace collection", () => { }; const subject = harness({ codeZip }); - await expect(subject.run()).rejects.toBeInstanceOf(SilentCLIError); + await expect(subject.run({ agent: "orders" })).rejects.toThrow("runner failed"); + expect(subject.collector.state.closed).toBe(1); + }); +}); + +describe("project dev Inspector UI mode", () => { + // Wraps the run promise so awaiting this helper does not flatten it into + // "wait for the whole dev command to exit". + async function runUi(subject: ReturnType, flags: Record = {}) { + const pending = subject.run({ mode: "browser", ...flags }); + pending.catch(() => undefined); + await Bun.sleep(5); // let the handler start the UI server and block on events + return { pending }; + } + + test("starts the Inspector, prints the URL, and opens the browser on a TTY", async () => { + const subject = harness({ tty: true }); + const { pending } = await runUi(subject); + + expect(subject.ui.starts).toEqual([{ port: 8081 }]); + expect(subject.io.stderr()).toContain("Agent Inspector running at http://127.0.0.1:8081"); + expect(subject.ui.opened).toEqual(["http://127.0.0.1:8081"]); + + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); expect(subject.collector.state.closed).toBe(1); }); + + test.each([{}, { tty: true, json: true }] as const)( + "never opens a browser without a TTY or in JSON mode (%o)", + async (options) => { + const subject = harness(options); + const { pending } = await runUi(subject); + expect(subject.ui.opened).toEqual([]); + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); + }, + ); + + test("serves the Inspector API: status lists every runtime, none started", async () => { + const subject = harness({ + project: project(runtime("orders"), runtime("support", "Container")), + }); + const { pending } = await runUi(subject); + + expect((await inspectorStatus(subject)).map((agent) => agent.name)).toEqual([ + "orders", + "support", + ]); + expect(subject.codeZip.inputs).toHaveLength(0); + + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); + }); + + test("agentcore.json edits reload the supervised agents", async () => { + const subject = harness({ reloadedRuntimes: [runtime("orders"), runtime("payments")] }); + const { pending } = await runUi(subject); + + expect(subject.watchers[0]?.path).toBe( + join("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/workspace/project", "agentcore", "agentcore.json"), + ); + subject.watchers[0]!.onChange(); + await Bun.sleep(5); + + expect((await inspectorStatus(subject)).map((agent) => agent.name)).toEqual([ + "orders", + "payments", + ]); + expect(subject.io.stderr()).toContain("Reloaded agents from agentcore.json."); + + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); + }); + + test("--agent narrows the supervised set", async () => { + const subject = harness({ + project: project(runtime("orders"), runtime("support", "Container")), + }); + const { pending } = await runUi(subject, { agent: "support" }); + + expect((await inspectorStatus(subject)).map((agent) => agent.name)).toEqual(["support"]); + + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); + }); + + test("an explicit --ui-port that is taken fails fast", async () => { + const subject = harness({ checkPort: async () => false }); + await expect(subject.run({ mode: "browser", "ui-port": 9999 })).rejects.toThrow( + "Port 9999 is already in use", + ); + }); + + test("--port with several runtimes is rejected", async () => { + const subject = harness({ + project: project(runtime("orders"), runtime("support", "Container")), + }); + await expect(subject.run({ mode: "browser", port: 4567 })).rejects.toThrow( + "--port applies to a single runtime", + ); + }); }); test("project dev renders attributed human and NDJSON output", async () => { @@ -349,16 +416,13 @@ test("project dev renders attributed human and NDJSON output", async () => { for (const json of [false, true]) { const subject = harness({ codeZip: captureRunner(events), json }); - await subject.run({ traces: false }).catch(() => undefined); - if (json) { - expect(subject.io.stdout()).toContain( - JSON.stringify({ agent: "orders", type: "stdout", line: "agent output" }), - ); - } else { - expect(subject.io.stdout()).toContain("[orders] agent output"); - expect(subject.io.stderr()).toContain("[orders] Starting"); - expect(subject.io.stderr()).toContain("[orders] agent warning"); - } + await subject.run({ agent: "orders", traces: false }); + expect(subject.io.stdout()).toBe( + json + ? events.map((event) => JSON.stringify({ agent: "orders", ...event })).join("\n") + : "[orders] agent output", + ); + expect(subject.io.stderr()).toBe(json ? "" : "[orders] Starting\n[orders] agent warning"); } }); @@ -384,7 +448,7 @@ describe("project dev interruption", () => { const codeZip = heldRunner(); const before = process.listenerCount(signal); const subject = harness({ codeZip }); - const pending = subject.run(); + const pending = subject.run({ agent: "orders" }); const input = await codeZip.started; process.emit(signal, signal); @@ -401,4 +465,15 @@ describe("project dev interruption", () => { expect(process.listenerCount(signal)).toBe(before); }, ); + + test("preserves an ordinary runner failure", async () => { + const failure = new InputValidationError("runner failed"); + const codeZip = captureRunner(); + codeZip.runner.run = async function* () { + yield* []; + throw failure; + }; + + await expect(harness({ codeZip }).run({ agent: "orders" })).rejects.toBe(failure); + }); }); diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index 952c0bfed..c21d1a2cb 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -1,29 +1,43 @@ import { join } from "node:path"; import z from "zod"; +import { createInspectorHandler } from "../../../core/dev/inspector/server"; +import type { InspectorDeps } from "../../../core/dev/inspector/types"; import { rewriteOtelEndpointForContainer } from "../../../core/dev/otel/collector"; -import { resolveDevPort } from "../../../core/dev/port"; +import { findFreePort, resolveDevPort } from "../../../core/dev/port"; +import { projectSpecPath } from "../../../core/project/fsUtils"; import { DevSupervisor, type SupervisorConfig } from "../../../core/dev/supervisor"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { InputValidationError, ResourceNotFoundError, - SilentCLIError, UserCancellationError, } from "../../../errors"; -import type { AppIO, PortChecker } from "../../../io"; +import type { AppIO, BrowserOpener, FileWatcher, PortChecker, startHttpServer } from "../../../io"; import { createHandler, flag, ProjectKey } from "../../../router"; import { JsonRendererKey, type JsonRenderer } from "../../../tui"; import { JsonKey, RegionKey } from "../../keys"; -import type { Project } from "../types"; +import type { Project, ProjectManager } from "../types"; import type { DevEnvironmentLoader } from "./environment"; import type { DevEvent, DevRunner, DevTraceCollector, DevTraceCollectorStarter } from "./types"; +/** The Inspector UI binds 8081 or, when that is taken, the next free port. */ +const UI_DEFAULT_PORT = 8081; + export type DevProjectHandlerConfig = { io: AppIO; runners: { CodeZip: DevRunner; Container: DevRunner }; loadDevEnvironment: DevEnvironmentLoader; checkPort: PortChecker; startTraceCollector: DevTraceCollectorStarter; + startServer: typeof startHttpServer; + openBrowser: BrowserOpener; + inspectorAssets: InspectorDeps["assets"]; + /** Whether the command runs on an interactive terminal (gates browser auto-open). */ + isInteractive: () => boolean; + /** Watches agentcore.json so the Inspector reflects config edits live. */ + watchFile: FileWatcher; + /** Re-resolves the project after a config change to pick up runtime edits. */ + projectManager: Pick; /** Overrides how the supervisor decides an agent is ready (defaults to a real TCP poll). */ waitReady?: SupervisorConfig["waitReady"]; }; @@ -86,6 +100,16 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => z.coerce.number().int().min(1).max(65535).optional(), ), flag("traces", "disable local OTEL trace collection", z.boolean().default(true)), + flag( + "mode", + "how to run: browser (Agent Inspector web UI), headless (one agent in the terminal), or tui", + z.enum(["browser", "headless", "tui"]).default("browser"), + ), + flag( + "ui-port", + "port for the Agent Inspector web UI (browser mode)", + z.coerce.number().int().min(1).max(65535).optional(), + ), ], handle: async (ctx, flags) => { const controller = new AbortController(); @@ -102,7 +126,18 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => try { const project = ctx.require(ProjectKey); const region = ctx.require(RegionKey); + if (flags.mode === "tui") { + throw new InputValidationError( + "TUI mode is not available yet. Use --mode browser (default) or --mode headless.", + ); + } const runtimes = selectRuntimes(project, flags.agent); + if (flags.mode === "headless" && !flags.agent) { + const available = runtimes.map((runtime) => runtime.name).join(", "); + throw new InputValidationError( + `--mode headless runs a single agent in the terminal. Pass --agent to choose which one. Available: ${available}.`, + ); + } if (runtimes.length > 1 && flags.port !== undefined) { throw new InputValidationError( "--port applies to a single runtime. Use --agent to select one.", @@ -159,11 +194,26 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => return { ...env, ...otel }; }; + if (flags.mode === "headless") { + await runWithoutUi( + config, + runtimes[0]!, + project, + flags.port, + getDevEnvVarsForRuntime, + controller, + json, + ); + return; + } + 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) => ( await resolveDevPort( @@ -177,21 +227,47 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => signal: controller.signal, }); - const starts = Promise.allSettled( - runtimes.map((runtime) => supervisor.start(runtime.name)), + 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 onConfigChange = async () => { + try { + const reloaded = await config.projectManager.resolve({ filePath: project.rootPath }); + if (!reloaded) return; + const runtimes = reloaded.spec.runtimes; + 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(), + 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); + for await (const { agentName, event } of supervisor.events()) { renderAgentEvent(config.io, event, agentName, json); - if (controller.signal.aborted) break; - 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(); - await starts; } catch (error) { controller.signal.throwIfAborted(); throw error; @@ -203,3 +279,45 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => } }, }); + +/** + * 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/handlers/project/dev/types.ts b/src/handlers/project/dev/types.ts index 15b51cb96..98175248d 100644 --- a/src/handlers/project/dev/types.ts +++ b/src/handlers/project/dev/types.ts @@ -1,3 +1,4 @@ +import type { InspectorTraces } from "../../../core/dev/inspector/types"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; export type DevEvent = @@ -22,6 +23,8 @@ export interface DevTraceCollector { port: number; /** Environment variables that point an agent's OTEL SDK at the receiver. */ envVars: Record; + /** The trace reads the Inspector serves, without exposing the store itself. */ + traces: InspectorTraces; close(): Promise; } diff --git a/src/handlers/project/index.ts b/src/handlers/project/index.ts index 2eb42eeaa..774a2ca39 100644 --- a/src/handlers/project/index.ts +++ b/src/handlers/project/index.ts @@ -1,7 +1,8 @@ import { Router } from "../../router"; -import { checkPort, type AppIO } from "../../io"; +import { checkPort, openBrowser, startHttpServer, watchFile, type AppIO } from "../../io"; import { CodeZipDevRunner } from "../../core/dev/codezip"; import { ContainerDevRunner } from "../../core/dev/container"; +import { InspectorAssets } from "../../core/dev/inspectorAssets"; import { startOtelCollector } from "../../core/dev/otel/collector"; import { withProject } from "../../middleware"; import { createCreateProjectHandler } from "./create"; @@ -42,6 +43,12 @@ export function createProjectHandler(config: ProjectHandlerConfig): Router { loadDevEnvironment, checkPort, startTraceCollector: startOtelCollector, + startServer: startHttpServer, + openBrowser, + inspectorAssets: new InspectorAssets(), + isInteractive: () => process.stdout.isTTY === true, + watchFile, + projectManager: config.projectManager, }), ), ); diff --git a/src/io/index.ts b/src/io/index.ts index 770ac030d..f30dd7206 100644 --- a/src/io/index.ts +++ b/src/io/index.ts @@ -45,3 +45,5 @@ export { type HttpResponse, type HttpServerHandle, } from "./httpServer"; +export { openBrowser, type BrowserOpener } from "./openBrowser"; +export { watchFile, type FileWatcher } from "./watchFile"; diff --git a/src/io/openBrowser.ts b/src/io/openBrowser.ts new file mode 100644 index 000000000..1d6bfbe68 --- /dev/null +++ b/src/io/openBrowser.ts @@ -0,0 +1,26 @@ +import { spawn } from "node:child_process"; + +export type BrowserOpener = (url: string) => Promise; + +/** + * Open a URL in the user's default browser, best-effort: failures resolve + * quietly because the URL is always printed as well, so a machine without a + * browser association must not fail `project dev`. + */ +export const openBrowser: BrowserOpener = (url) => { + const [command, args] = + process.platform === "darwin" + ? ["open", [url]] + : process.platform === "win32" + ? ["cmd", ["/c", "start", "", url]] + : ["xdg-open", [url]]; + + return new Promise((resolve) => { + const child = spawn(command!, args, { stdio: "ignore", detached: true }); + child.on("error", () => resolve()); + child.on("spawn", () => { + child.unref(); + resolve(); + }); + }); +}; diff --git a/src/io/watchFile.test.ts b/src/io/watchFile.test.ts new file mode 100644 index 000000000..59dc25e48 --- /dev/null +++ b/src/io/watchFile.test.ts @@ -0,0 +1,45 @@ +import { afterEach, expect, test } from "bun:test"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { watchFile } from "./watchFile"; + +let dir: string | undefined; +afterEach(() => { + if (dir) rmSync(dir, { recursive: true, force: true }); + dir = undefined; +}); + +function tempFile(): string { + dir = mkdtempSync(join(tmpdir(), "watch-")); + const path = join(dir, "agentcore.json"); + writeFileSync(path, "{}"); + return path; +} + +test("debounces a burst of edits into a single callback and stops on abort", async () => { + const path = tempFile(); + const controller = new AbortController(); + let calls = 0; + watchFile(path, () => calls++, controller.signal); + + writeFileSync(path, '{"a":1}'); + writeFileSync(path, '{"a":2}'); + await Bun.sleep(250); + expect(calls).toBe(1); + + controller.abort(); + writeFileSync(path, '{"a":3}'); + await Bun.sleep(250); + expect(calls).toBe(1); +}); + +test("a missing file fails quietly rather than throwing", () => { + expect(() => + watchFile( + join(tmpdir(), "does-not-exist-agentcore.json"), + () => {}, + new AbortController().signal, + ), + ).not.toThrow(); +}); diff --git a/src/io/watchFile.ts b/src/io/watchFile.ts new file mode 100644 index 000000000..f46390726 --- /dev/null +++ b/src/io/watchFile.ts @@ -0,0 +1,30 @@ +import { watch } from "node:fs"; + +export type FileWatcher = (path: string, onChange: () => void, signal: AbortSignal) => void; + +/** + * Watch one file, debouncing the editor's burst of change events into a single + * callback. Watching stops when the signal aborts; a missing file or an + * unsupported platform fails quietly, because watching is a convenience and + * never load-bearing. + */ +export const watchFile: FileWatcher = (path, onChange, signal, debounceMs = 150) => { + let timer: ReturnType | undefined; + try { + const watcher = watch(path, () => { + clearTimeout(timer); + timer = setTimeout(onChange, debounceMs); + }); + watcher.on("error", () => watcher.close()); + signal.addEventListener( + "abort", + () => { + clearTimeout(timer); + watcher.close(); + }, + { once: true }, + ); + } catch { + // Watching is best-effort. + } +};