From 3d9cb5bb8e36c9f05820efc62b9cefc5e5708001 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Wed, 19 Aug 2026 15:11:45 -0400 Subject: [PATCH 1/6] 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 1b79e92929def11ab3ae5f8f4353681ecfc2e8a4 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Wed, 19 Aug 2026 15:51:26 -0400 Subject: [PATCH 2/6] 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 a697e1531bb46539a9079eb3dbf6250e5116a927 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 17:43:32 -0400 Subject: [PATCH 3/6] 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) => { From abf8c232c3a24a0c77d95b436525f510ee75952a Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Mon, 24 Aug 2026 12:26:15 -0400 Subject: [PATCH 4/6] fix(dev): deliver events queued while the supervisor stream drains a batch events() spliced the queue, yielded each event, then installed its wake callback and blocked. A push landing during the yields ran while wake was undefined, so its wake was a no-op, and the newly installed waiter never noticed the queued event until the next push or shutdown. Re-check the queue inside the wait so a queued event resolves immediately. --- src/core/dev/supervisor.test.ts | 63 +++++++++++++++++++++++++++++++++ src/core/dev/supervisor.ts | 4 +++ 2 files changed, 67 insertions(+) diff --git a/src/core/dev/supervisor.test.ts b/src/core/dev/supervisor.test.ts index 8d0946366..3ee021ee2 100644 --- a/src/core/dev/supervisor.test.ts +++ b/src/core/dev/supervisor.test.ts @@ -40,6 +40,41 @@ function dyingRunner(failure?: Error) { return { runner }; } +/** A runner that stays alive and emits events on demand via `emit`, until its signal aborts. */ +function pushableRunner() { + const buffer: DevEvent[] = []; + let wake: (() => void) | undefined; + let done = false; + const runner: DevRunner = { + run: async function* (input) { + input.signal.addEventListener( + "abort", + () => { + done = true; + wake?.(); + }, + { once: true }, + ); + while (!done) { + while (buffer.length) yield buffer.shift()!; + if (done) break; + await new Promise((resolve) => { + wake = resolve; + }); + wake = undefined; + } + }, + }; + const emit = (event: DevEvent) => { + buffer.push(event); + wake?.(); + }; + return { runner, emit }; +} + +/** Let a runner emission propagate through the pump into the supervisor queue. */ +const flush = () => Bun.sleep(1); + type HarnessOptions = { runtimes?: ProjectRuntime[]; codeZip?: { runner: DevRunner }; @@ -177,6 +212,34 @@ describe("DevSupervisor", () => { }); }); + test("delivers an event queued while the consumer is draining a batch", async () => { + const orders = pushableRunner(); + const { supervisor, controller } = harness({ + runtimes: [runtime("orders")], + codeZip: { runner: orders.runner }, + }); + const events = supervisor.events(); + await supervisor.start("orders"); + + orders.emit({ type: "stdout", line: "one" }); + await flush(); + // Park the generator mid-batch (at a yield, so `wake` is unset), then queue + // another event into that window. A lost-wakeup would strand it. + await events.next(); + orders.emit({ type: "stdout", line: "two" }); + await flush(); + await events.next(); + + const next = await Promise.race([events.next(), Bun.sleep(200).then(() => "stalled" as const)]); + expect(next).not.toBe("stalled"); + expect((next as IteratorResult).value).toMatchObject({ + agent: "orders", + event: { type: "stdout", line: "two" }, + }); + controller.abort(); + await events.next(); + }); + test("a running agent that crashes reports failed and leaves the stream alive", async () => { let fail!: () => void; const crashing: DevRunner = { diff --git a/src/core/dev/supervisor.ts b/src/core/dev/supervisor.ts index 4a54eb8da..3a3e8bf13 100644 --- a/src/core/dev/supervisor.ts +++ b/src/core/dev/supervisor.ts @@ -136,6 +136,10 @@ export class DevSupervisor { if (this.config.signal.aborted) return; await new Promise((resolve) => { this.wake = resolve; + // A push during the yields above ran while wake was undefined, so its + // wake was a no-op. Re-check now that wake is installed, so a queued + // event resolves immediately instead of waiting for the next push. + if (this.queue.length > 0) resolve(); }); this.wake = undefined; } From cf625ca471189de0cbd1ca99cc6ada9107bb5ca0 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Tue, 25 Aug 2026 15:41:15 -0400 Subject: [PATCH 5/6] refactor(dev): route single runtime through the supervisor and address review Remove the runSingleRuntime special case so every runtime, one or many, runs through DevSupervisor. Start all agents concurrently and stream their output live rather than awaiting each readiness first. End the event loop when no agent is starting or running, and exit non-zero when none started. Make start() async so an unknown name rejects rather than throwing synchronously. Time readiness off the agent's last output (idle timeout) so a slow build that keeps logging is not killed. Move waitForPort into src/io/port.ts to keep socket IO in the IO layer. Rename the environment accessor to getDevEnvVarsForRuntime and SupervisedEvent.agent to agentName. --- src/core/dev/supervisor.test.ts | 37 ++++-- src/core/dev/supervisor.ts | 85 +++++-------- src/handlers/project/dev/index.test.ts | 165 ++++++++++++------------- src/handlers/project/dev/index.ts | 103 +++++---------- src/io/index.ts | 2 +- src/io/port.ts | 46 ++++++- 6 files changed, 210 insertions(+), 228 deletions(-) diff --git a/src/core/dev/supervisor.test.ts b/src/core/dev/supervisor.test.ts index 3ee021ee2..4c8bb38e2 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, waitForPort, type SupervisedEvent } from "./supervisor"; +import { DevSupervisor, type SupervisedEvent } from "./supervisor"; +import { waitForPort } from "../../io"; import { createServer } from "node:net"; function runtime(name: string, build: ProjectRuntime["build"] = "CodeZip"): ProjectRuntime { @@ -91,7 +92,7 @@ function harness(options: HarnessOptions = {}) { runtimes: options.runtimes ?? [runtime("orders"), runtime("billing", "Container")], projectRoot: "/workspace/project", runners: { CodeZip: codeZip.runner, Container: container.runner }, - environment: async (agentRuntime) => ({ AGENT: agentRuntime.name }), + getDevEnvVarsForRuntime: async (agentRuntime) => ({ AGENT: agentRuntime.name }), resolvePort: async () => nextPort++, waitReady: options.ready ?? (async () => {}), signal: controller.signal, @@ -183,9 +184,9 @@ describe("DevSupervisor", () => { controller.abort(); }); - test("unknown agents are rejected with the available names", () => { + test("unknown agents are rejected with the available names", async () => { const { supervisor, controller } = harness(); - expect(() => supervisor.start("missing")).toThrow("Available agents: orders, billing"); + await expect(supervisor.start("missing")).rejects.toThrow("Available agents: orders, billing"); controller.abort(); }); @@ -199,15 +200,15 @@ describe("DevSupervisor", () => { const events = await drain(supervisor, controller); expect(events).toContainEqual({ - agent: "orders", + agentName: "orders", event: { type: "stdout", line: "orders out" }, }); expect(events).toContainEqual({ - agent: "billing", + agentName: "billing", event: { type: "stderr", line: "billing err" }, }); expect(events).toContainEqual({ - agent: "orders", + agentName: "orders", event: { type: "status", message: "Agent 'orders' is running on port 9100." }, }); }); @@ -233,7 +234,7 @@ describe("DevSupervisor", () => { const next = await Promise.race([events.next(), Bun.sleep(200).then(() => "stalled" as const)]); expect(next).not.toBe("stalled"); expect((next as IteratorResult).value).toMatchObject({ - agent: "orders", + agentName: "orders", event: { type: "stdout", line: "two" }, }); controller.abort(); @@ -263,7 +264,7 @@ describe("DevSupervisor", () => { expect(supervisor.running("orders")).toBeUndefined(); const events = await drain(supervisor, controller); expect(events).toContainEqual({ - agent: "orders", + agentName: "orders", event: { type: "status", message: "Agent 'orders' crashed: segfault" }, }); }); @@ -287,17 +288,27 @@ describe("DevSupervisor", () => { 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", + await expect(waitForPort(1, signal, undefined, 10, 100)).rejects.toThrow( + "produced no output and 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 + await waitForPort(port, signal, undefined, 10, 1000); // resolves against a live listener server.close(); }); + test("recent activity keeps a silent port from timing out", async () => { + const controller = new AbortController(); + // Idle window is 20ms, but the agent keeps "producing output", so the poll + // must not give up on the deadline — only the abort below ends it. + const pending = waitForPort(1, controller.signal, () => Date.now(), 5, 20); + await Bun.sleep(60); + controller.abort(); + await expect(pending).rejects.toThrow("Aborted while waiting"); + }); + test("failed setup does not leak parent abort listeners across retries", async () => { const adds: string[] = []; const removes: string[] = []; @@ -318,7 +329,7 @@ describe("DevSupervisor", () => { runtimes: [runtime("orders")], projectRoot: "/workspace/project", runners: { CodeZip: serverRunner().runner, Container: serverRunner().runner }, - environment: async () => ({}), + getDevEnvVarsForRuntime: async () => ({}), resolvePort: async () => { throw new Error("no ports for you"); }, diff --git a/src/core/dev/supervisor.ts b/src/core/dev/supervisor.ts index 3a3e8bf13..d39ab3f57 100644 --- a/src/core/dev/supervisor.ts +++ b/src/core/dev/supervisor.ts @@ -1,5 +1,5 @@ -import { connect } from "node:net"; import { ResourceNotFoundError } from "../../errors"; +import { waitForPort } from "../../io"; import type { DevEvent, DevRunner } from "../../handlers/project/dev/types"; import type { ProjectRuntime } from "../../projectSchemas/runtime"; @@ -14,9 +14,9 @@ export interface AgentStatus { error?: string; } -/** A dev event attributed to the agent that produced it. */ +/** A dev event tagged with the name of the runtime that produced it. */ export interface SupervisedEvent { - agent: string; + agentName: string; event: DevEvent; } @@ -25,11 +25,15 @@ export type SupervisorConfig = { projectRoot: string; runners: { CodeZip: DevRunner; Container: DevRunner }; /** Resolves the full child environment for a runtime (dev env + OTEL vars). */ - environment: (runtime: ProjectRuntime) => Promise>; + getDevEnvVarsForRuntime: (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; + /** + * Resolves once a started agent accepts connections on its port. `lastActivityAt` + * returns the time of the agent's most recent output, so a build that keeps + * logging is not timed out mid-flight. + */ + waitReady?: (port: number, signal: AbortSignal, lastActivityAt: () => number) => Promise; signal: AbortSignal; }; @@ -52,7 +56,11 @@ export class DevSupervisor { private readonly agents = new Map(); private readonly queue: SupervisedEvent[] = []; private wake: (() => void) | undefined; - private readonly waitReady: (port: number, signal: AbortSignal) => Promise; + private readonly waitReady: ( + port: number, + signal: AbortSignal, + lastActivityAt: () => number, + ) => Promise; constructor(private readonly config: SupervisorConfig) { for (const runtime of config.runtimes) { @@ -107,7 +115,7 @@ export class DevSupervisor { * 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 }> { + public async start(name: string): Promise<{ name: string; port: number }> { const entry = this.agents.get(name); if (!entry) { const available = [...this.agents.keys()].join(", "); @@ -116,7 +124,7 @@ export class DevSupervisor { ); } if (entry.phase === "running" && entry.port !== undefined) { - return Promise.resolve({ name, port: entry.port }); + return { name, port: entry.port }; } if (entry.starting) return entry.starting; @@ -145,8 +153,8 @@ export class DevSupervisor { } } - private push(agent: string, event: DevEvent): void { - this.queue.push({ agent, event }); + private push(agentName: string, event: DevEvent): void { + this.queue.push({ agentName, event }); this.wake?.(); } @@ -156,7 +164,7 @@ export class DevSupervisor { entry.error = undefined; const controller = new AbortController(); - const onParentAbort = () => controller.abort(); + const onParentAbort = () => controller.abort(this.config.signal.reason); // 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 }); @@ -164,14 +172,17 @@ export class DevSupervisor { try { const port = await this.config.resolvePort(entry.runtime); - const env = await this.config.environment(entry.runtime); + const env = await this.config.getDevEnvVarsForRuntime(entry.runtime); const runner = this.config.runners[entry.runtime.build]; let ready = false; - const readiness = this.waitReady(port, controller.signal).then(() => { + const activity = { at: Date.now() }; + const readiness = this.waitReady(port, controller.signal, () => activity.at).then(() => { ready = true; }); - const earlyExit = this.pump(entry, runner, { port, env, signal: controller.signal }) + const earlyExit = this.pump(entry, runner, { port, env, signal: controller.signal }, () => { + activity.at = Date.now(); + }) .finally(unchain) .then(() => { if (!ready) @@ -205,6 +216,7 @@ export class DevSupervisor { entry: AgentEntry, runner: DevRunner, input: { port: number; env: Record; signal: AbortSignal }, + onActivity: () => void, ): Promise { const name = entry.runtime.name; try { @@ -215,6 +227,7 @@ export class DevSupervisor { env: input.env, signal: input.signal, })) { + onActivity(); this.push(name, event); } if (entry.phase === "running") { @@ -233,45 +246,3 @@ export class DevSupervisor { } } } - -/** 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(); - }); - 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 6b4c6209d..9bb4979dc 100644 --- a/src/handlers/project/dev/index.test.ts +++ b/src/handlers/project/dev/index.test.ts @@ -1,13 +1,13 @@ 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 { InputValidationError, ResourceNotFoundError, + SilentCLIError, UserCancellationError, } from "../../../errors"; -import { checkPort, type PortChecker } from "../../../io"; +import type { PortChecker } from "../../../io"; import { ProjectKey, ValueContext } from "../../../router"; import { testIO } from "../../../testing"; import { JsonRendererKey } from "../../../tui"; @@ -35,6 +35,7 @@ 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 = { @@ -46,6 +47,21 @@ 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 }; @@ -78,8 +94,8 @@ type HarnessOptions = { function harness(options: HarnessOptions = {}) { const io = testIO(); - const codeZip = options.codeZip ?? captureRunner(); - const container = options.container ?? captureRunner(); + const codeZip = options.codeZip ?? stayingRunner(); + const container = options.container ?? stayingRunner(); const collector = fakeCollector(); const environmentInputs: DevEnvironmentInput[] = []; const handler = createDevProjectHandler({ @@ -93,6 +109,11 @@ 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); + }, }); const ctx = ValueContext.EmptyContext() .withValue(ProjectKey, options.project ?? project(runtime())) @@ -114,6 +135,25 @@ function harness(options: HarnessOptions = {}) { }; } +/** + * 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); +} + describe("project dev selection and dispatch", () => { test.each([ [project(), {}, "This project has no runtimes", InputValidationError], @@ -136,7 +176,7 @@ describe("project dev selection and dispatch", () => { const subject = harness({ project: project(runtime("orders"), runtime("support", "Container")), }); - await subject.run({ agent: "support", port: 4567 }); + const { pending } = await supervised(subject, { agent: "support", port: 4567 }); expect(subject.codeZip.inputs).toHaveLength(0); expect(subject.environmentInputs).toEqual([ @@ -157,9 +197,10 @@ describe("project dev selection and dispatch", () => { }, runtime: { name: "support", build: "Container" }, }); + await interrupt(pending); }); - test("announces an automatically selected port", async () => { + test("resolves and announces an automatically selected port", async () => { const checked: number[] = []; const subject = harness({ checkPort: async (port) => { @@ -167,56 +208,23 @@ describe("project dev selection and dispatch", () => { return port === 8081; }, }); - await subject.run(); + const { pending } = await supervised(subject); expect(checked).toEqual([8080, 8081]); expect(subject.codeZip.inputs[0]?.port).toBe(8081); - expect(subject.io.stderr()).toContain("Port 8080 is in use; using 8081."); + expect(subject.io.stderr()).toContain("Agent 'orders' is running on port 8081"); + await interrupt(pending); }); }); -/** - * 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 + 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); @@ -237,22 +245,16 @@ describe("project dev multi-agent supervision", () => { }); 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, + codeZip: captureRunner([{ type: "status", message: "dying" }]), // exits: never ready + container: stayingRunner(), }); - const pending = subject.run(); - pending.catch(() => undefined); - await Bun.sleep(300); + 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"); - - process.emit("SIGINT", "SIGINT"); - await pending.catch(() => undefined); + await interrupt(pending); }); test("--port without --agent is rejected when several runtimes exist", async () => { @@ -265,7 +267,7 @@ describe("project dev multi-agent supervision", () => { describe("project dev trace collection", () => { test("starts the collector, announces it, and points a CodeZip agent at loopback", async () => { const subject = harness(); - await subject.run(); + const { pending } = await supervised(subject); expect(subject.collector.starts).toEqual([ { @@ -279,21 +281,21 @@ 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")), - }); - await subject.run(); + const subject = harness({ project: project(runtime("support", "Container")) }); + const { pending } = await supervised(subject); 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(); - await subject.run(); + const { pending } = await supervised(subject); const onError = subject.collector.starts[0]?.onError; onError?.(new Error("disk full")); @@ -303,26 +305,29 @@ 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(); - await subject.run({ traces: false }); + const { pending } = await supervised(subject, { 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) }); - await subject.run(); + const { pending } = await supervised(subject); expect(subject.collector.starts).toHaveLength(0); expect(subject.codeZip.inputs[0]?.env).toEqual({ FROM_LOADER: "yes" }); + await interrupt(pending); }); - test("the collector is closed when the runner fails", async () => { + test("a failed agent exits non-zero and still closes the collector", async () => { const codeZip = captureRunner(); codeZip.runner.run = async function* () { yield* []; @@ -330,12 +335,12 @@ describe("project dev trace collection", () => { }; const subject = harness({ codeZip }); - await expect(subject.run()).rejects.toThrow("runner failed"); + await expect(subject.run()).rejects.toBeInstanceOf(SilentCLIError); expect(subject.collector.state.closed).toBe(1); }); }); -test("project dev renders human and NDJSON output", async () => { +test("project dev renders attributed human and NDJSON output", async () => { const events: DevEvent[] = [ { type: "status", message: "Starting" }, { type: "stdout", line: "agent output" }, @@ -344,11 +349,16 @@ test("project dev renders human and NDJSON output", async () => { for (const json of [false, true]) { const subject = harness({ codeZip: captureRunner(events), json }); - await subject.run({ traces: false }); - expect(subject.io.stdout()).toBe( - json ? events.map((event) => JSON.stringify(event)).join("\n") : "agent output", - ); - expect(subject.io.stderr()).toBe(json ? "" : "Starting\nagent warning"); + 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"); + } } }); @@ -391,15 +401,4 @@ 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()).rejects.toBe(failure); - }); }); diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index 7e6e11ee9..6f6962953 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -2,11 +2,12 @@ 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 { 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"; @@ -23,6 +24,8 @@ export type DevProjectHandlerConfig = { loadDevEnvironment: DevEnvironmentLoader; checkPort: PortChecker; startTraceCollector: DevTraceCollectorStarter; + /** Overrides how the supervisor decides an agent is ready (defaults to a real TCP poll). */ + waitReady?: SupervisorConfig["waitReady"]; }; /** Env for a spawned agent so its OTEL SDK reports to the collector as this runtime. */ @@ -137,7 +140,9 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => } controller.signal.throwIfAborted(); - const environment = async (runtime: ProjectRuntime): Promise> => { + const getDevEnvVarsForRuntime = async ( + runtime: ProjectRuntime, + ): Promise> => { const { env } = await config.loadDevEnvironment({ projectRoot: project.rootPath, runtime, @@ -150,41 +155,39 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => 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, runners: config.runners, - environment, + getDevEnvVarsForRuntime, resolvePort: async (runtime) => - (await resolveDevPort(runtime.protocol, undefined, config.checkPort, controller.signal)) - .port, + ( + await resolveDevPort( + runtime.protocol, + flags.port, + config.checkPort, + controller.signal, + ) + ).port, + waitReady: config.waitReady, signal: controller.signal, }); - // 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); + const starts = Promise.allSettled( + runtimes.map((runtime) => supervisor.start(runtime.name)), + ); + + for await (const { agentName, event } of supervisor.events()) { + renderEvent(config.io, event, json, agentName); + 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; @@ -196,49 +199,3 @@ 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); - } -} diff --git a/src/io/index.ts b/src/io/index.ts index 23eb4a03d..c6684d5e2 100644 --- a/src/io/index.ts +++ b/src/io/index.ts @@ -36,7 +36,7 @@ export { } from "./streamingResponse"; export type { AppIO, ReadWriteJson } from "./types"; export { warn } from "./warn"; -export { checkPort, type PortChecker } from "./port"; +export { checkPort, waitForPort, type PortChecker } from "./port"; export { startHttpServer, type HttpRequest, diff --git a/src/io/port.ts b/src/io/port.ts index 897de2da5..79c90fe44 100644 --- a/src/io/port.ts +++ b/src/io/port.ts @@ -1,4 +1,4 @@ -import { createServer } from "node:net"; +import { connect, createServer } from "node:net"; export type PortChecker = (port: number, signal: AbortSignal) => Promise; @@ -25,3 +25,47 @@ export const checkPort: PortChecker = async (port, signal) => { signal.throwIfAborted(); return available; }; + +/** How long an agent may stay silent before its startup is abandoned. */ +const IDLE_TIMEOUT_MS = 120_000; + +/** + * Poll until a loopback TCP connection to `port` succeeds, the signal aborts, + * or the agent has been silent past `idleMs`. Timing off the last output rather + * than a fixed deadline lets a still-building container keep its start alive. + */ +export function waitForPort( + port: number, + signal: AbortSignal, + lastActivityAt?: () => number, + intervalMs = 250, + idleMs = IDLE_TIMEOUT_MS, +): Promise { + const startedAt = Date.now(); + const since = lastActivityAt ?? (() => startedAt); + 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() - since() > idleMs) { + reject( + new Error( + `Agent produced no output and did not accept connections on port ${port} within ${idleMs / 1000}s.`, + ), + ); + return; + } + const socket = connect({ port, host: "127.0.0.1" }, () => { + socket.destroy(); + resolve(); + }); + socket.on("error", () => { + socket.destroy(); + setTimeout(attempt, intervalMs); + }); + }; + attempt(); + }); +} From ec9377c9dadadc8463b25391e635be51bcfb533f Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Tue, 25 Aug 2026 16:14:21 -0400 Subject: [PATCH 6/6] refactor(dev): always attribute agent output to its runtime Split renderEvent into renderAgentEvent (agent name required) and renderStatus for command-level trace-collector lines, so agent output is always tagged with [name] and the agent-optional branches are gone. Addresses review feedback on #2041. --- src/handlers/project/dev/index.ts | 32 +++++++++++++++++-------------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index 6f6962953..952c0bfed 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -53,15 +53,25 @@ function selectRuntimes(project: Project, name?: string): ProjectRuntime[] { ); } -function renderEvent(io: AppIO, event: DevEvent, json?: JsonRenderer, agent?: string): void { +/** An agent's own output, always tagged with the agent that produced it. */ +function renderAgentEvent(io: AppIO, event: DevEvent, agent: string, json?: JsonRenderer): void { if (json) { - json.renderJsonLine(agent === undefined ? event : { agent, ...event }); + json.renderJsonLine({ agent, ...event }); return; } const output = event.type === "stdout" ? io.stdout : io.stderr; const line = event.type === "status" ? event.message : event.line; - output.write(agent === undefined ? `${line}\n` : `[${agent}] ${line}\n`); + output.write(`[${agent}] ${line}\n`); +} + +/** A command-level status line, not attributed to any agent. */ +function renderStatus(io: AppIO, message: string, json?: JsonRenderer): void { + if (json) { + json.renderJsonLine({ type: "status", message }); + return; + } + io.stderr.write(`${message}\n`); } export const createDevProjectHandler = (config: DevProjectHandlerConfig) => @@ -119,22 +129,16 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => if (tracePersistErrorReported) return; tracePersistErrorReported = true; const detail = error instanceof Error ? error.message : String(error); - renderEvent( + renderStatus( config.io, - { - type: "status", - message: `Warning: failed to persist traces to ${tracesDirectory} (${detail}); collected traces may be incomplete.`, - }, + `Warning: failed to persist traces to ${tracesDirectory} (${detail}); collected traces may be incomplete.`, json, ); }, }); - renderEvent( + renderStatus( config.io, - { - type: "status", - message: `OTEL collector listening on port ${collector.port}; traces persist to ${tracesDirectory}.`, - }, + `OTEL collector listening on port ${collector.port}; traces persist to ${tracesDirectory}.`, json, ); } @@ -178,7 +182,7 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => ); for await (const { agentName, event } of supervisor.events()) { - renderEvent(config.io, event, json, agentName); + renderAgentEvent(config.io, event, agentName, json); if (controller.signal.aborted) break; const phases = supervisor.snapshot(); if (phases.every(({ phase }) => phase !== "starting" && phase !== "running")) {