From 055afd86d9a96fa46aff6d764c00658aac8d58a8 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Wed, 19 Aug 2026 15:11:45 -0400 Subject: [PATCH 1/3] feat(dev): supervise every runtime in project dev MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit project dev without --agent now runs all of the project's runtimes at once: a DevSupervisor owns per-agent lifecycle (sequential port resolution — a concurrent race would put two agents on one port), merges every runner's output into one agent-attributed stream ([name] prefixes; an agent field in NDJSON), and keeps the session alive when one agent crashes. Selecting a single runtime (--agent, or a one-runtime project) keeps the direct path where a crash still fails the command. --- src/core/dev/supervisor.test.ts | 237 +++++++++++++++++++++++ src/core/dev/supervisor.ts | 251 +++++++++++++++++++++++++ src/handlers/project/dev/index.test.ts | 96 +++++++++- src/handlers/project/dev/index.ts | 160 +++++++++++----- 4 files changed, 689 insertions(+), 55 deletions(-) create mode 100644 src/core/dev/supervisor.test.ts create mode 100644 src/core/dev/supervisor.ts diff --git a/src/core/dev/supervisor.test.ts b/src/core/dev/supervisor.test.ts new file mode 100644 index 000000000..bcb67ba57 --- /dev/null +++ b/src/core/dev/supervisor.test.ts @@ -0,0 +1,237 @@ +import { describe, expect, test } from "bun:test"; +import type { DevEvent, DevRunner, DevServerInput } from "../../handlers/project/dev/types"; +import type { ProjectRuntime } from "../../projectSchemas/runtime"; +import { DevSupervisor, type SupervisedEvent } from "./supervisor"; + +function runtime(name: string, build: ProjectRuntime["build"] = "CodeZip"): ProjectRuntime { + return { + name, + build, + protocol: "HTTP", + entrypoint: "main.py", + codeLocation: `app/${name}`, + } as ProjectRuntime; +} + +/** A runner that emits `events`, then stays alive until its signal aborts (like a real server). */ +function serverRunner(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 }; +} + +/** A runner whose process dies immediately (optionally with an error). */ +function dyingRunner(failure?: Error) { + const runner: DevRunner = { + run: async function* () { + yield { type: "status", message: "starting" }; + if (failure) throw failure; + }, + }; + return { runner }; +} + +type HarnessOptions = { + runtimes?: ProjectRuntime[]; + codeZip?: { runner: DevRunner }; + container?: { runner: DevRunner }; + ready?: (port: number, signal: AbortSignal) => Promise; +}; + +function harness(options: HarnessOptions = {}) { + const controller = new AbortController(); + const codeZip = options.codeZip ?? serverRunner(); + const container = options.container ?? serverRunner(); + let nextPort = 9100; + const supervisor = new DevSupervisor({ + runtimes: options.runtimes ?? [runtime("orders"), runtime("billing", "Container")], + projectRoot: "/workspace/project", + runners: { CodeZip: codeZip.runner, Container: container.runner }, + environment: async (agentRuntime) => ({ AGENT: agentRuntime.name }), + resolvePort: async () => nextPort++, + waitReady: options.ready ?? (async () => {}), + signal: controller.signal, + }); + return { supervisor, controller, codeZip, container }; +} + +async function drain( + supervisor: DevSupervisor, + controller: AbortController, +): Promise { + controller.abort(); + const collected: SupervisedEvent[] = []; + for await (const event of supervisor.events()) collected.push(event); + return collected; +} + +describe("DevSupervisor", () => { + test("agents are idle until started, then report running with their port", async () => { + const { supervisor, controller } = harness(); + expect(supervisor.snapshot()).toMatchObject([ + { name: "orders", phase: "idle", buildType: "CodeZip", protocol: "HTTP" }, + { name: "billing", phase: "idle", buildType: "Container" }, + ]); + + const started = await supervisor.start("orders"); + expect(started).toEqual({ name: "orders", port: 9100 }); + expect(supervisor.snapshot()[0]).toMatchObject({ + name: "orders", + phase: "running", + port: 9100, + }); + expect(supervisor.running("orders")).toEqual({ port: 9100, protocol: "HTTP" }); + expect(supervisor.running("billing")).toBeUndefined(); + controller.abort(); + }); + + test("passes environment, project root, and port to the runner", async () => { + const codeZip = serverRunner(); + const { supervisor, controller } = harness({ codeZip }); + await supervisor.start("orders"); + + expect(codeZip.inputs[0]).toMatchObject({ + projectRoot: "/workspace/project", + port: 9100, + env: { AGENT: "orders" }, + runtime: { name: "orders" }, + }); + controller.abort(); + }); + + test("concurrent starts of the same agent share one attempt", async () => { + const codeZip = serverRunner(); + let readiness!: () => void; + const { supervisor, controller } = harness({ + codeZip, + ready: () => new Promise((resolve) => (readiness = resolve)), + }); + + const [first, second] = [supervisor.start("orders"), supervisor.start("orders")]; + expect(supervisor.snapshot()[0]!.phase).toBe("starting"); + await Bun.sleep(1); // let the launch reach its readiness wait + readiness(); + expect(await first).toEqual(await second); + expect(codeZip.inputs).toHaveLength(1); + + // A start after running returns the existing port without a new attempt. + expect(await supervisor.start("orders")).toEqual({ name: "orders", port: 9100 }); + expect(codeZip.inputs).toHaveLength(1); + controller.abort(); + }); + + test("an agent that exits before readiness fails the start and can be retried", async () => { + const { supervisor, controller } = harness({ + codeZip: dyingRunner(new Error("boom")), + ready: () => new Promise(() => {}), + }); + + expect(supervisor.start("orders")).rejects.toThrow("boom"); + await supervisor.start("orders").catch(() => {}); + expect(supervisor.snapshot()[0]).toMatchObject({ + name: "orders", + phase: "failed", + error: "boom", + }); + + // Retry hits the runner again rather than being stuck. + await supervisor.start("orders").catch(() => {}); + controller.abort(); + }); + + test("unknown agents are rejected with the available names", () => { + const { supervisor, controller } = harness(); + expect(() => supervisor.start("missing")).toThrow("Available agents: orders, billing"); + controller.abort(); + }); + + test("merges attributed events from several agents into one stream", async () => { + const codeZip = serverRunner([{ type: "stdout", line: "orders out" }]); + const container = serverRunner([{ type: "stderr", line: "billing err" }]); + const { supervisor, controller } = harness({ codeZip, container }); + + await supervisor.start("orders"); + await supervisor.start("billing"); + const events = await drain(supervisor, controller); + + expect(events).toContainEqual({ + agent: "orders", + event: { type: "stdout", line: "orders out" }, + }); + expect(events).toContainEqual({ + agent: "billing", + event: { type: "stderr", line: "billing err" }, + }); + expect(events).toContainEqual({ + agent: "orders", + event: { type: "status", message: "Agent 'orders' is running on port 9100." }, + }); + }); + + test("a running agent that crashes reports failed and leaves the stream alive", async () => { + let fail!: () => void; + const crashing: DevRunner = { + run: async function* () { + yield { type: "status", message: "up" }; + await new Promise((resolve) => (fail = resolve)); + throw new Error("segfault"); + }, + }; + const { supervisor, controller } = harness({ codeZip: { runner: crashing } }); + + await supervisor.start("orders"); + fail(); + await Bun.sleep(5); + + expect(supervisor.snapshot()[0]).toMatchObject({ + name: "orders", + phase: "failed", + error: "segfault", + }); + expect(supervisor.running("orders")).toBeUndefined(); + const events = await drain(supervisor, controller); + expect(events).toContainEqual({ + agent: "orders", + event: { type: "status", message: "Agent 'orders' crashed: segfault" }, + }); + }); + + test("setRuntimes adds, updates, and drops agents without touching running ones", async () => { + const { supervisor, controller } = harness(); + await supervisor.start("orders"); + + supervisor.setRuntimes([runtime("orders"), runtime("payments")]); + expect(supervisor.snapshot().map(({ name, phase }) => ({ name, phase }))).toEqual([ + { name: "orders", phase: "running" }, + { name: "payments", phase: "idle" }, + ]); + + // A running agent survives removal from the config until it stops. + supervisor.setRuntimes([runtime("payments")]); + expect(supervisor.snapshot().map(({ name }) => name)).toEqual(["orders", "payments"]); + controller.abort(); + }); + + test("aborting the parent signal stops running agents and ends the stream", async () => { + const codeZip = serverRunner(); + const { supervisor, controller } = harness({ codeZip }); + await supervisor.start("orders"); + + const events: SupervisedEvent[] = []; + const consuming = (async () => { + for await (const event of supervisor.events()) events.push(event); + })(); + controller.abort(); + await consuming; + + expect(codeZip.inputs[0]!.signal.aborted).toBe(true); + }); +}); diff --git a/src/core/dev/supervisor.ts b/src/core/dev/supervisor.ts new file mode 100644 index 000000000..386262b4f --- /dev/null +++ b/src/core/dev/supervisor.ts @@ -0,0 +1,251 @@ +import { connect } from "node:net"; +import { ResourceNotFoundError } from "../../errors"; +import type { DevEvent, DevRunner } from "../../handlers/project/dev/types"; +import type { ProjectRuntime } from "../../projectSchemas/runtime"; + +export type AgentPhase = "idle" | "starting" | "running" | "failed"; + +export interface AgentStatus { + name: string; + buildType: ProjectRuntime["build"]; + protocol: NonNullable; + phase: AgentPhase; + port?: number; + error?: string; +} + +/** A dev event attributed to the agent that produced it. */ +export interface SupervisedEvent { + agent: string; + event: DevEvent; +} + +export type SupervisorConfig = { + runtimes: ProjectRuntime[]; + projectRoot: string; + runners: { CodeZip: DevRunner; Container: DevRunner }; + /** Resolves the full child environment for a runtime (dev env + OTEL vars). */ + environment: (runtime: ProjectRuntime) => Promise>; + /** Resolves the port a runtime should serve on. */ + resolvePort: (runtime: ProjectRuntime) => Promise; + /** Resolves once a started agent accepts connections on its port. */ + waitReady?: (port: number, signal: AbortSignal) => Promise; + signal: AbortSignal; +}; + +type AgentEntry = { + runtime: ProjectRuntime; + phase: AgentPhase; + port?: number; + error?: string; + starting?: Promise<{ name: string; port: number }>; +}; + +/** + * Owns the lifecycle of every dev-able runtime for the Inspector: agents start + * lazily (triggered from the browser), each in its own abort scope chained off + * the command's signal, and every runner's events merge into one attributed + * stream the dev handler renders. Restart-on-edit stays inside the child + * (uvicorn --reload / tsx watch) — the supervisor never restarts processes. + */ +export class DevSupervisor { + private readonly agents = new Map(); + private readonly queue: SupervisedEvent[] = []; + private wake: (() => void) | undefined; + private readonly waitReady: (port: number, signal: AbortSignal) => Promise; + + constructor(private readonly config: SupervisorConfig) { + for (const runtime of config.runtimes) { + this.agents.set(runtime.name, { runtime, phase: "idle" }); + } + this.waitReady = config.waitReady ?? waitForPort; + config.signal.addEventListener("abort", () => this.wake?.(), { once: true }); + } + + /** + * Replace the managed runtime set after a config change: new runtimes join + * idle, edited definitions apply on the next start, and removed runtimes + * drop unless they are currently starting or running. + */ + public setRuntimes(runtimes: ProjectRuntime[]): void { + 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" }); + } + for (const [name, entry] of this.agents) { + if (!names.has(name) && entry.phase !== "running" && entry.phase !== "starting") { + this.agents.delete(name); + } + } + } + + /** Current phase, port, and last error of every managed agent. */ + public snapshot(): AgentStatus[] { + return [...this.agents.values()].map(({ runtime, phase, port, error }) => ({ + name: runtime.name, + buildType: runtime.build, + protocol: runtime.protocol ?? "HTTP", + phase, + port, + error, + })); + } + + /** The port and protocol of a running agent, for proxying requests to it. */ + public running( + name: string, + ): { port: number; protocol: NonNullable } | undefined { + const entry = this.agents.get(name); + if (entry?.phase !== "running" || entry.port === undefined) return undefined; + return { port: entry.port, protocol: entry.runtime.protocol ?? "HTTP" }; + } + + /** + * Start an agent by name, resolving once it accepts connections. Concurrent + * and repeated starts of the same agent share one attempt; a previously + * failed agent may be started again. + */ + public start(name: string): Promise<{ name: string; port: number }> { + const entry = this.agents.get(name); + if (!entry) { + const available = [...this.agents.keys()].join(", "); + throw new ResourceNotFoundError( + `Agent '${name}' was not found. Available agents: ${available}.`, + ); + } + if (entry.phase === "running" && entry.port !== undefined) { + return Promise.resolve({ name, port: entry.port }); + } + if (entry.starting) return entry.starting; + + entry.starting = this.launch(entry).finally(() => { + entry.starting = undefined; + }); + return entry.starting; + } + + /** + * The merged event stream of every agent this supervisor has started. Ends + * when the supervisor's signal aborts and all pending events are drained. + */ + public async *events(): AsyncGenerator { + while (true) { + for (const event of this.queue.splice(0)) yield event; + if (this.config.signal.aborted) return; + await new Promise((resolve) => { + this.wake = resolve; + }); + this.wake = undefined; + } + } + + private push(agent: string, event: DevEvent): void { + this.queue.push({ agent, event }); + this.wake?.(); + } + + private async launch(entry: AgentEntry): Promise<{ name: string; port: number }> { + const name = entry.runtime.name; + entry.phase = "starting"; + entry.error = undefined; + + const controller = new AbortController(); + const onParentAbort = () => controller.abort(); + // Chained for the agent's whole lifetime (not just startup): the command's + // Ctrl-C must tear down every running child. The pump removes it on exit. + this.config.signal.addEventListener("abort", onParentAbort, { once: true }); + const unchain = () => this.config.signal.removeEventListener("abort", onParentAbort); + + try { + const port = await this.config.resolvePort(entry.runtime); + const env = await this.config.environment(entry.runtime); + const runner = this.config.runners[entry.runtime.build]; + + let ready = false; + const readiness = this.waitReady(port, controller.signal).then(() => { + ready = true; + }); + const earlyExit = this.pump(entry, runner, { port, env, signal: controller.signal }) + .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(() => {}); + earlyExit.catch(() => {}); + await Promise.race([readiness, earlyExit]); + + entry.phase = "running"; + entry.port = port; + this.push(name, { type: "status", message: `Agent '${name}' is running on port ${port}.` }); + return { name, port }; + } catch (error) { + controller.abort(); + entry.phase = "failed"; + entry.error = error instanceof Error ? error.message : String(error); + this.push(name, { + type: "status", + message: `Agent '${name}' failed to start: ${entry.error}`, + }); + throw error; + } + } + + /** Drives one runner generator, attributing its events; resolves when the runner ends. */ + private async pump( + entry: AgentEntry, + runner: DevRunner, + input: { port: number; env: Record; signal: AbortSignal }, + ): Promise { + const name = entry.runtime.name; + try { + for await (const event of runner.run({ + runtime: entry.runtime, + projectRoot: this.config.projectRoot, + port: input.port, + env: input.env, + signal: input.signal, + })) { + this.push(name, event); + } + if (entry.phase === "running") { + entry.phase = "idle"; + entry.port = undefined; + this.push(name, { type: "status", message: `Agent '${name}' stopped.` }); + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + entry.error = message; + if (entry.phase === "running") { + entry.phase = "failed"; + entry.port = undefined; + this.push(name, { type: "status", message: `Agent '${name}' crashed: ${message}` }); + } + } + } +} + +/** Poll until a loopback TCP connection to `port` succeeds, or the signal aborts. */ +function waitForPort(port: number, signal: AbortSignal, intervalMs = 250): Promise { + return new Promise((resolve, reject) => { + const attempt = () => { + if (signal.aborted) { + reject(new Error("Aborted while waiting for the agent to become ready.")); + return; + } + const socket = connect({ port, host: "127.0.0.1" }, () => { + socket.destroy(); + resolve(); + }); + socket.on("error", () => { + socket.destroy(); + setTimeout(attempt, intervalMs); + }); + }; + attempt(); + }); +} diff --git a/src/handlers/project/dev/index.test.ts b/src/handlers/project/dev/index.test.ts index 21c1b87f7..6b4c6209d 100644 --- a/src/handlers/project/dev/index.test.ts +++ b/src/handlers/project/dev/index.test.ts @@ -1,4 +1,5 @@ import { describe, expect, test } from "bun:test"; +import { createServer, type Server } from "node:net"; import { join } from "node:path"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { @@ -6,7 +7,7 @@ import { ResourceNotFoundError, UserCancellationError, } from "../../../errors"; -import type { PortChecker } from "../../../io"; +import { checkPort, type PortChecker } from "../../../io"; import { ProjectKey, ValueContext } from "../../../router"; import { testIO } from "../../../testing"; import { JsonRendererKey } from "../../../tui"; @@ -116,12 +117,6 @@ function harness(options: HarnessOptions = {}) { describe("project dev selection and dispatch", () => { test.each([ [project(), {}, "This project has no runtimes", InputValidationError], - [ - project(runtime("orders"), runtime("support", "Container")), - {}, - "Use --agent to select one. Available runtimes: orders, support", - InputValidationError, - ], [ project(runtime("orders"), runtime("support", "Container")), { agent: "missing" }, @@ -180,6 +175,93 @@ describe("project dev selection and dispatch", () => { }); }); +/** + * A runner that binds a real TCP listener on its assigned port (so the + * supervisor's genuine readiness probe passes) and stays alive until aborted. + */ +function listeningRunner(events: DevEvent[] = []) { + const inputs: DevServerInput[] = []; + const runner: DevRunner = { + run: async function* (input) { + inputs.push(input); + const server: Server = createServer(); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(input.port, "127.0.0.1", resolve); + }); + try { + yield* events; + await new Promise((resolve) => + input.signal.addEventListener("abort", () => resolve(), { once: true }), + ); + } finally { + server.close(); + } + }, + }; + return { runner, inputs }; +} + +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 = listeningRunner([{ type: "stdout", line: "orders says hi" }]); + const container = listeningRunner(); + const subject = harness({ + project: twoRuntimes(), + codeZip, + container, + checkPort, // the real checker: resolved ports reflect this machine + }); + const pending = subject.run(); + pending.catch(() => undefined); + await Bun.sleep(300); // both agents bind and pass the real readiness probe + + 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 container = listeningRunner(); + const subject = harness({ + project: twoRuntimes(), + codeZip: captureRunner([{ type: "status", message: "dying" }]), // ends immediately: never ready + container, + checkPort, + }); + const pending = subject.run(); + pending.catch(() => undefined); + await Bun.sleep(300); + + expect(subject.io.stderr()).toContain("[orders] Agent 'orders' failed to start"); + expect(subject.io.stderr()).toContain("Agent 'support' is running on port"); + + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); + }); + + 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", + ); + }); +}); + describe("project dev trace collection", () => { test("starts the collector, announces it, and points a CodeZip agent at loopback", async () => { const subject = harness(); diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index b37ae7b7a..727e5dbb4 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -2,6 +2,7 @@ import { join } from "node:path"; import z from "zod"; import { rewriteOtelEndpointForContainer } from "../../../core/dev/otel/collector"; import { resolveDevPort } from "../../../core/dev/port"; +import { DevSupervisor } from "../../../core/dev/supervisor"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { InputValidationError, @@ -33,36 +34,31 @@ function otelEnvForRuntime( return runtime.build === "Container" ? rewriteOtelEndpointForContainer(env) : env; } -function selectRuntime(project: Project, name?: string): ProjectRuntime { +function selectRuntimes(project: Project, name?: string): ProjectRuntime[] { if (project.spec.runtimes.length === 0) { throw new InputValidationError( "This project has no runtimes. Add a runtime to agentcore/agentcore.json and retry.", ); } - const available = project.spec.runtimes.map(({ name }) => name).join(", "); + if (!name) return project.spec.runtimes; - if (name) { - const runtime = project.spec.runtimes.find((candidate) => candidate.name === name); - if (runtime) return runtime; - throw new ResourceNotFoundError( - `Runtime '${name}' was not found. Available runtimes: ${available}.`, - ); - } - - if (project.spec.runtimes.length === 1) return project.spec.runtimes[0]!; - throw new InputValidationError( - `Multiple runtimes found. Use --agent to select one. Available runtimes: ${available}.`, + const runtime = project.spec.runtimes.find((candidate) => candidate.name === name); + if (runtime) return [runtime]; + const available = project.spec.runtimes.map((candidate) => candidate.name).join(", "); + throw new ResourceNotFoundError( + `Runtime '${name}' was not found. Available runtimes: ${available}.`, ); } -function renderEvent(io: AppIO, event: DevEvent, json?: JsonRenderer): void { +function renderEvent(io: AppIO, event: DevEvent, json?: JsonRenderer, agent?: string): void { if (json) { - json.renderJsonLine(event); + json.renderJsonLine(agent === undefined ? event : { agent, ...event }); return; } const output = event.type === "stdout" ? io.stdout : io.stderr; - output.write(`${event.type === "status" ? event.message : event.line}\n`); + const line = event.type === "status" ? event.message : event.line; + output.write(agent === undefined ? `${line}\n` : `[${agent}] ${line}\n`); } export const createDevProjectHandler = (config: DevProjectHandlerConfig) => @@ -92,33 +88,18 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => let collector: DevTraceCollector | undefined; try { const project = ctx.require(ProjectKey); - const runtime = selectRuntime(project, flags.agent); - const devPort = await resolveDevPort( - runtime.protocol, - flags.port, - config.checkPort, - controller.signal, - ); - if (devPort.port !== devPort.requestedPort) { - renderEvent( - config.io, - { - type: "status", - message: `Port ${devPort.requestedPort} is in use; using ${devPort.port}.`, - }, - json, + 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.", ); } - const { env } = await config.loadDevEnvironment({ - projectRoot: project.rootPath, - runtime, - region: ctx.require(RegionKey), - }); - controller.signal.throwIfAborted(); - - let otelEnv: Record = {}; - if (flags.traces && (runtime.instrumentation?.enableOtel ?? true)) { + 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({ @@ -142,7 +123,6 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => ); }, }); - otelEnv = otelEnvForRuntime(collector, runtime); renderEvent( config.io, { @@ -154,16 +134,54 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => } controller.signal.throwIfAborted(); - const runner = config.runners[runtime.build]; - for await (const event of runner.run({ - runtime, + const environment = 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 }; + }; + + if (runtimes.length === 1) { + await runSingleRuntime( + config, + runtimes[0]!, + project, + flags.port, + environment, + controller, + json, + ); + return; + } + + // Several runtimes: supervise them all, streaming agent-attributed output. + const supervisor = new DevSupervisor({ + runtimes, projectRoot: project.rootPath, - port: devPort.port, - env: { ...env, ...otelEnv }, + runners: config.runners, + environment, + resolvePort: async (runtime) => + (await resolveDevPort(runtime.protocol, undefined, config.checkPort, controller.signal)) + .port, signal: controller.signal, - })) { - renderEvent(config.io, event, json); + }); + // Sequential starts: concurrent port resolution would race two agents + // onto the same port. Failed starts surface as attributed status events. + for (const runtime of runtimes) { + await supervisor.start(runtime.name).catch(() => {}); + } + controller.signal.throwIfAborted(); + + for await (const { agent, event } of supervisor.events()) { + renderEvent(config.io, event, json, agent); } + controller.signal.throwIfAborted(); } catch (error) { controller.signal.throwIfAborted(); throw error; @@ -175,3 +193,49 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => } }, }); + +/** + * Run one runtime directly, streaming its output unattributed. Unlike the + * supervised multi-agent path, a crash here fails the command (scripts and CI + * rely on the non-zero exit). + */ +async function runSingleRuntime( + 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) { + renderEvent( + config.io, + { + type: "status", + message: `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, + })) { + renderEvent(config.io, event, json); + } +} From 4ad026cbc34883a55f51553fae301cf63da3a98b Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Wed, 19 Aug 2026 15:51:26 -0400 Subject: [PATCH 2/3] fix(dev): bound readiness polling; unchain abort listeners on failed setup A child that stays alive without ever binding its port previously blocked every later runtime (starts are sequential) until interrupted; readiness now gives up after 120s and fails that start. Setup failures before the pump exists (port resolution, environment) now remove their parent-abort listener like every other exit path, so Inspector retries of a failing agent cannot accumulate listeners. --- src/core/dev/supervisor.test.ts | 53 ++++++++++++++++++++++++++++++++- src/core/dev/supervisor.ts | 26 ++++++++++++++-- 2 files changed, 76 insertions(+), 3 deletions(-) diff --git a/src/core/dev/supervisor.test.ts b/src/core/dev/supervisor.test.ts index bcb67ba57..8d0946366 100644 --- a/src/core/dev/supervisor.test.ts +++ b/src/core/dev/supervisor.test.ts @@ -1,7 +1,8 @@ import { describe, expect, test } from "bun:test"; import type { DevEvent, DevRunner, DevServerInput } from "../../handlers/project/dev/types"; import type { ProjectRuntime } from "../../projectSchemas/runtime"; -import { DevSupervisor, type SupervisedEvent } from "./supervisor"; +import { DevSupervisor, waitForPort, type SupervisedEvent } from "./supervisor"; +import { createServer } from "node:net"; function runtime(name: string, build: ProjectRuntime["build"] = "CodeZip"): ProjectRuntime { return { @@ -220,6 +221,56 @@ describe("DevSupervisor", () => { controller.abort(); }); + test("readiness polling gives up at its deadline instead of blocking forever", async () => { + const signal = new AbortController().signal; + // Nothing listens on this port; a bounded poll must reject, not hang. + await expect(waitForPort(1, signal, 10, 100)).rejects.toThrow( + "did not accept connections on port 1 within 0.1s", + ); + + const server = createServer(); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const port = (server.address() as { port: number }).port; + await waitForPort(port, signal, 10, 1000); // resolves against a live listener + server.close(); + }); + + test("failed setup does not leak parent abort listeners across retries", async () => { + const adds: string[] = []; + const removes: string[] = []; + const controller = new AbortController(); + const countingSignal = { + aborted: false, + addEventListener: (type: string, listener: () => void, options?: unknown) => { + adds.push(type); + controller.signal.addEventListener(type as "abort", listener, options as undefined); + }, + removeEventListener: (type: string, listener: () => void) => { + removes.push(type); + controller.signal.removeEventListener(type as "abort", listener); + }, + } as unknown as AbortSignal; + + const supervisor = new DevSupervisor({ + runtimes: [runtime("orders")], + projectRoot: "/workspace/project", + runners: { CodeZip: serverRunner().runner, Container: serverRunner().runner }, + environment: async () => ({}), + resolvePort: async () => { + throw new Error("no ports for you"); + }, + waitReady: async () => {}, + signal: countingSignal, + }); + + for (let attempt = 0; attempt < 3; attempt++) { + await supervisor.start("orders").catch(() => {}); + } + // One constructor wake listener stays; every per-launch listener must be removed. + expect(adds.length - removes.length).toBe(1); + controller.abort(); + }); + test("aborting the parent signal stops running agents and ends the stream", async () => { const codeZip = serverRunner(); const { supervisor, controller } = harness({ codeZip }); diff --git a/src/core/dev/supervisor.ts b/src/core/dev/supervisor.ts index 386262b4f..4a54eb8da 100644 --- a/src/core/dev/supervisor.ts +++ b/src/core/dev/supervisor.ts @@ -185,6 +185,7 @@ export class DevSupervisor { return { name, port }; } catch (error) { 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); this.push(name, { @@ -229,14 +230,35 @@ export class DevSupervisor { } } -/** Poll until a loopback TCP connection to `port` succeeds, or the signal aborts. */ -function waitForPort(port: number, signal: AbortSignal, intervalMs = 250): Promise { +/** Generous enough for a cold dependency install before the server first binds. */ +const READY_TIMEOUT_MS = 120_000; + +/** + * Poll until a loopback TCP connection to `port` succeeds, the signal aborts, + * or the deadline passes — a child that stays alive without ever binding must + * fail its start instead of blocking every later runtime. + */ +export function waitForPort( + port: number, + signal: AbortSignal, + intervalMs = 250, + timeoutMs = READY_TIMEOUT_MS, +): Promise { + const deadline = Date.now() + timeoutMs; return new Promise((resolve, reject) => { const attempt = () => { if (signal.aborted) { reject(new Error("Aborted while waiting for the agent to become ready.")); return; } + if (Date.now() > deadline) { + reject( + new Error( + `Agent did not accept connections on port ${port} within ${timeoutMs / 1000}s.`, + ), + ); + return; + } const socket = connect({ port, host: "127.0.0.1" }, () => { socket.destroy(); resolve(); From 7b4b60ba716d4aa6d677f56d6e426b9a3618adcb Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 17:43:32 -0400 Subject: [PATCH 3/3] fix(dev): bind collector to 0.0.0.0 when any runtime is a container The #1980 rebase carried a single-runtime host check (runtime.build) into the multi-agent dev handler, where the variable is the runtimes array. Bind all interfaces when any selected runtime runs in a container. --- src/handlers/project/dev/index.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index 727e5dbb4..7e6e11ee9 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -105,8 +105,11 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => collector = await config.startTraceCollector({ tracesDirectory, // A container reaches the collector over the host bridge, which a - // 127.0.0.1 bind refuses, so the container path binds all interfaces. - host: runtime.build === "Container" ? "0.0.0.0" : "127.0.0.1", + // 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) => {