Skip to content
Merged
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
69 changes: 43 additions & 26 deletions packages/coding-agent/src/core/subagents/controller.ts
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ interface SubagentRecord {
run?: Promise<void>;
createSession?: CreateSubagentSession;
runController?: AbortController;
runCancellation?: Promise<void>;
interruptionRequested?: boolean;
actionTail?: Promise<void>;
cleanupFailures?: unknown[];
Expand Down Expand Up @@ -810,11 +811,7 @@ export class SubagentController {
async #quiesceRecord(record: SubagentRecord): Promise<void> {
record.session?.clearQueue();
record.interruptionRequested = true;
if (record.status === "queued") {
record.runController?.abort(new Error(`Subagent interrupted: ${record.name}`));
} else if (record.session) {
await record.session.abort();
}
await this.#requestRunCancellation(record, new Error(`Subagent interrupted: ${record.name}`));
await record.run;
if (record.status === "queued" || record.status === "running") this.markInterrupted(record.name);
}
Expand Down Expand Up @@ -1318,14 +1315,10 @@ export class SubagentController {

record.interruptionRequested = true;
const run = record.run;
if (record.status === "queued") {
record.runController?.abort(new Error(`Subagent interrupted: ${name}`));
} else if (record.session) {
try {
await record.session.abort();
} catch {
// The captured run still determines the settled controller state.
}
try {
await this.#requestRunCancellation(record, new Error(`Subagent interrupted: ${name}`));
} catch {
// The captured run still determines the settled controller state.
}
await run;
if (record.status === "queued" || record.status === "running") {
Expand All @@ -1342,6 +1335,7 @@ export class SubagentController {
record.interruptionRequested = false;
const runController = new AbortController();
record.runController = runController;
record.runCancellation = undefined;
const run = this.#runPrompt(record, prompt, runController);
record.run = run;
void run.catch(() => {});
Expand Down Expand Up @@ -1423,7 +1417,7 @@ export class SubagentController {
assistantMessages.push(event.message);
}
});
await activeSession.prompt(prompt);
await activeSession.prompt(prompt, { signal: runController.signal });
if (this.#closed) return;

const terminalMessage = assistantMessages.at(-1);
Expand Down Expand Up @@ -1483,31 +1477,54 @@ export class SubagentController {
#detachRecords(reason: Error): SubagentRecord[] {
const records = [...this.#records.values()];
for (const record of records) {
record.runController?.abort(reason);
void this.#requestRunCancellation(record, reason);
}
this.#records.clear();
return records;
}

#requestRunCancellation(record: SubagentRecord, reason: Error): Promise<void> {
if (record.runCancellation) return record.runCancellation;

let cancellation: Promise<void>;
// AgentSession marks streaming before awaiting agent_start hooks, while this record can still be queued.
if (record.runController && !record.session?.isStreaming) {
record.runController.abort(reason);
cancellation = Promise.resolve();
} else if (record.session) {
try {
cancellation = record.session.abort();
} catch (error) {
cancellation = Promise.reject(error);
}
} else {
record.runController?.abort(reason);
cancellation = Promise.resolve();
}

record.runCancellation = cancellation;
void cancellation.catch(() => {
if (record.runCancellation === cancellation) {
record.runCancellation = undefined;
}
});
return cancellation;
}

async #cleanupRecords(records: SubagentRecord[]): Promise<unknown[]> {
const failures: unknown[] = [];
for (const record of records) record.leafUnsubscribe?.();
const abortResults = await Promise.allSettled(
records.flatMap((record) => {
const session = record.session;
if (!session) return [];
for (const record of records) {
const session = record.session;
if (session) {
try {
session.clearQueue();
} catch (error) {
failures.push(error);
}
try {
return [session.abort()];
} catch (error) {
return [Promise.reject(error)];
}
}),
);
}
}
const abortResults = await Promise.allSettled(records.map((record) => record.runCancellation));
for (const result of abortResults) {
if (result.status === "rejected") failures.push(result.reason);
}
Expand Down
14 changes: 9 additions & 5 deletions packages/coding-agent/test/suite/subagent-persistence.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -618,15 +618,16 @@ describe("durable subagent lifecycle", () => {
expect(root.session.getSubagents()).toEqual([{ name: "target", status: "idle", task: "Inspect the change" }]);
});

it("retains the old leaf and roster when child quiescing fails", async () => {
it("retains the old leaf and roster when child quiescing fails and retries cancellation on shutdown", async () => {
const childStarted = deferred<void>();
const releaseChild = deferred<void>();
const root = await createRoot();
const targetId = createRosterBranch(root, []);
root.harness.setResponses([
async () => {
async (_context, request) => {
childStarted.resolve();
await releaseChild.promise;
await new Promise<void>((resolvePromise) => {
request?.signal?.addEventListener("abort", () => resolvePromise(), { once: true });
});
return fauxAssistantMessage("finished later");
},
]);
Expand All @@ -640,7 +641,10 @@ describe("durable subagent lifecycle", () => {
expect(root.sessionManager.getLeafId()).toBe(currentLeafId);
expect(root.session.getSubagents()[0]).toMatchObject({ name: "worker", status: "running" });
abort.mockRestore();
releaseChild.resolve();
const shutdownAbort = vi.spyOn(AgentSession.prototype, "abort");

await expect(root.session.dispose()).resolves.toBeUndefined();
expect(shutdownAbort).toHaveBeenCalledOnce();
});

it("does not block a successful leaf change when a quiesce status snapshot fails", async () => {
Expand Down
81 changes: 81 additions & 0 deletions packages/coding-agent/test/suite/subagent-startup.test.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import type { AgentEvent } from "@earendil-works/diffex-agent-core";
import type { AssistantMessage, Context } from "@earendil-works/diffex-ai";
import { fauxAssistantMessage } from "@earendil-works/diffex-ai";
import { afterEach, describe, expect, it, vi } from "vitest";
Expand All @@ -14,6 +15,10 @@ import {
waitForStatus,
} from "./subagent-harness.ts";

interface AgentSessionWithExtensionEmitHook {
_emitExtensionEvent: (event: AgentEvent) => Promise<void>;
}

describe("asynchronous subagent startup", () => {
const harnesses: SubagentTestHarness[] = [];

Expand Down Expand Up @@ -188,6 +193,82 @@ describe("asynchronous subagent startup", () => {
expect(events.map((event) => event.type)).toEqual(["subagent_reserved", "subagent_controller_disposed"]);
});

it("interrupts an attached child during prompt preflight without starting the provider", async () => {
const { harness, controller, createSession } = await setup();
const authStarted = deferred<void>();
let child: AgentSession | undefined;
harness.setResponses([fauxAssistantMessage("must not run")]);
vi.spyOn(harness.session.modelRuntime, "hasConfiguredAuth").mockReturnValue(false);
vi.spyOn(harness.session.modelRuntime, "checkAuth").mockImplementation(async (_providerId, options) => {
authStarted.resolve();
const signal = options?.signal;
if (!signal) throw new Error("Expected the controller run signal during prompt preflight");
return new Promise<never>((_resolve, reject) => {
if (signal.aborted) {
reject(signal.reason);
return;
}
signal.addEventListener("abort", () => reject(signal.reason), { once: true });
});
});
controller.spawn("worker", "task", async (signal) => {
child = await createSession(signal);
return child;
});
await authStarted.promise;

expect(controller.snapshot().agents).toMatchObject([{ name: "worker", status: "queued" }]);
await expect(controller.interrupt("worker")).resolves.toMatchObject({ status: "interrupted" });
expect(controller.snapshot().agents).toMatchObject([{ name: "worker", status: "interrupted" }]);
expect(child?.messages).toEqual([]);
expect(harness.faux.state.callCount).toBe(0);
});

it("serializes shutdown cancellation during agent_start publication", async () => {
const { harness, controller, createSession } = await setup();
const childReady = deferred<void>();
const allowChildReturn = deferred<void>();
const agentStartHandling = deferred<void>();
const allowAgentStartHandling = deferred<void>();
const events: SubagentLifecycleEvent[] = [];
let child: AgentSession | undefined;
harness.setResponses([fauxAssistantMessage("must not run")]);
controller.subscribe((event) => events.push(event));
controller.spawn("worker", "task", async (signal) => {
child = await createSession(signal);
childReady.resolve();
await allowChildReturn.promise;
return child;
});
await childReady.promise;

const activeChild = child!;
const sessionWithHook = activeChild as unknown as AgentSessionWithExtensionEmitHook;
const emitExtensionEvent = sessionWithHook._emitExtensionEvent.bind(sessionWithHook);
sessionWithHook._emitExtensionEvent = async (event) => {
if (event.type === "agent_start") {
agentStartHandling.resolve();
await allowAgentStartHandling.promise;
}
await emitExtensionEvent(event);
};
const abort = vi.spyOn(activeChild, "abort");
const dispose = vi.spyOn(activeChild, "dispose");
allowChildReturn.resolve();
await agentStartHandling.promise;

expect(controller.snapshot().agents).toMatchObject([{ name: "worker", status: "queued" }]);
const shutdown = controller.shutdown();
expect(abort).toHaveBeenCalledOnce();
allowAgentStartHandling.resolve();
await shutdown;

expect(abort).toHaveBeenCalledOnce();
expect(dispose).toHaveBeenCalledOnce();
expect(harness.faux.state.callCount).toBe(0);
expect(events.map((event) => event.type)).toEqual(["subagent_reserved", "subagent_controller_disposed"]);
});

it("aborts and disposes an executing child once during shutdown", async () => {
const { harness, controller, createSession } = await setup();
const response = deferred<AssistantMessage>();
Expand Down
Loading