Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/core/dev/inspector/server.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ describe("GET /api/status", () => {
buildType: "Container",
protocol: "MCP",
phase: "failed",
error: "boom",
error: new Error("boom"),
},
],
});
Expand Down
6 changes: 3 additions & 3 deletions src/core/dev/inspector/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -86,9 +86,9 @@ function handleStatus(deps: InspectorDeps): HttpResponse {
running: snapshot
.filter((agent) => agent.phase === "running" && agent.port !== undefined)
.map(({ name, port }) => ({ name, port: port! })),
errors: snapshot
.filter((agent) => agent.error !== undefined)
.map(({ name, error }) => ({ name, message: error! })),
errors: snapshot.flatMap(({ name, error }) =>
error ? [{ name, message: error.message }] : [],
),
selectedAgent: deps.selectedAgent,
};
return json(200, status);
Expand Down
2 changes: 1 addition & 1 deletion src/core/dev/inspector/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ export interface InspectorAgentStatus {
protocol: string;
phase: "idle" | "starting" | "running" | "failed";
port?: number;
error?: string;
error?: Error;
}

/** The slice of the dev supervisor the Inspector routes need. */
Expand Down
4 changes: 2 additions & 2 deletions src/core/dev/supervisor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@ describe("DevSupervisor", () => {
expect(supervisor.snapshot()[0]).toMatchObject({
name: "orders",
phase: "failed",
error: "boom",
error: { message: "boom" },
});

// Retry hits the runner again rather than being stuck.
Expand Down Expand Up @@ -262,7 +262,7 @@ describe("DevSupervisor", () => {
expect(supervisor.snapshot()[0]).toMatchObject({
name: "orders",
phase: "failed",
error: "segfault",
error: { message: "segfault" },
});
expect(supervisor.running("orders")).toBeUndefined();
const events = await drain(supervisor, controller);
Expand Down
57 changes: 45 additions & 12 deletions src/core/dev/supervisor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ export interface AgentStatus {
protocol: NonNullable<ProjectRuntime["protocol"]>;
phase: AgentPhase;
port?: number;
error?: string;
error?: Error;
}

/** A dev event tagged with the name of the runtime that produced it. */
Expand Down Expand Up @@ -41,7 +41,8 @@ type AgentEntry = {
runtime: ProjectRuntime;
phase: AgentPhase;
port?: number;
error?: string;
error?: Error;
completed?: boolean;
starting?: Promise<{ name: string; port: number }>;
/** The running child's pump, so shutdown can await its final spans. */
running?: Promise<void>;
Expand Down Expand Up @@ -152,18 +153,30 @@ export class DevSupervisor {
/**
* The merged event stream of every agent this supervisor has started. Ends
* when the supervisor's signal aborts and all pending events are drained.
* Headless consumers can also finish after every agent stops.
*/
public async *events(): AsyncGenerator<SupervisedEvent, void> {
public async *events(
options: { untilStopped?: boolean } = {},
): AsyncGenerator<SupervisedEvent, void> {
while (true) {
for (const event of this.queue.splice(0)) yield event;
if (this.config.signal.aborted) {
// Let every live child finish shutting down so its final spans reach the
// collector before the caller closes it, then drain what they emitted.
const running = [...this.agents.values()].map((entry) => entry.running).filter(Boolean);
await Promise.allSettled(running);
const pending = [...this.agents.values()]
.flatMap((entry) => [entry.starting, entry.running])
.filter(Boolean);
await Promise.allSettled(pending);
for (const event of this.queue.splice(0)) yield event;
return;
}
if (
options.untilStopped &&
[...this.agents.values()].every(({ phase }) => phase !== "starting" && phase !== "running")
) {
if (this.queue.length > 0) continue;
return;
}
await new Promise<void>((resolve) => {
this.wake = resolve;
// A push during the yields above ran while wake was undefined, so its
Expand All @@ -184,6 +197,7 @@ export class DevSupervisor {
const name = entry.runtime.name;
entry.phase = "starting";
entry.error = undefined;
entry.completed = false;

const controller = new AbortController();
const onParentAbort = () => controller.abort(this.config.signal.reason);
Expand All @@ -195,6 +209,7 @@ export class DevSupervisor {
try {
const port = await this.config.resolvePort(entry.runtime);
const env = await this.config.getDevEnvVarsForRuntime(entry.runtime);
controller.signal.throwIfAborted();
const runner = this.config.runners[entry.runtime.build];

let ready = false;
Expand All @@ -207,14 +222,24 @@ export class DevSupervisor {
});
entry.running = pump;
const earlyExit = pump.finally(unchain).then(() => {
if (!ready)
throw new Error(entry.error ?? `Agent '${name}' exited before it became ready.`);
if (!ready) {
if (entry.error) throw entry.error;
throw new Error(`Agent '${name}' exited before it became ready.`);
}
});
// Both branches outlive the race (the pump runs for the agent's lifetime);
// swallow their late rejections so losing branches never become unhandled.
readiness.catch(() => {});
earlyExit.catch(() => {});
await Promise.race([readiness, earlyExit]);
if (entry.error) throw entry.error;

if (entry.completed) {
entry.phase = "idle";
entry.port = undefined;
this.push(name, { type: "status", message: `Agent '${name}' stopped.` });
return { name, port };
}

entry.phase = "running";
entry.port = port;
Expand All @@ -224,10 +249,12 @@ export class DevSupervisor {
controller.abort();
unchain(); // idempotent alongside the pump's cleanup; covers setup failures before the pump exists
entry.phase = "failed";
entry.error = error instanceof Error ? error.message : String(error);
const agentError =
error instanceof Error ? error : new Error(String(error), { cause: error });
entry.error = agentError;
this.push(name, {
type: "status",
message: `Agent '${name}' failed to start: ${entry.error}`,
message: `Agent '${name}' failed to start: ${agentError.message}`,
});
throw error;
}
Expand Down Expand Up @@ -256,6 +283,8 @@ export class DevSupervisor {
entry.phase = "idle";
entry.port = undefined;
this.push(name, { type: "status", message: `Agent '${name}' stopped.` });
} else if (entry.phase === "starting") {
entry.completed = true;
}
} catch (error) {
/** The runner rejects with the abort reason on teardown, which is a stop, not a crash. */
Expand All @@ -267,12 +296,16 @@ export class DevSupervisor {
}
return;
}
const message = error instanceof Error ? error.message : String(error);
entry.error = message;
const agentError =
error instanceof Error ? error : new Error(String(error), { cause: error });
entry.error = agentError;
if (entry.phase === "running") {
entry.phase = "failed";
entry.port = undefined;
this.push(name, { type: "status", message: `Agent '${name}' crashed: ${message}` });
this.push(name, {
type: "status",
message: `Agent '${name}' crashed: ${agentError.message}`,
});
}
}
}
Expand Down
Loading
Loading