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
62 changes: 33 additions & 29 deletions packages/coding-agent/src/core/agent-session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1244,20 +1244,23 @@ export class AgentSession {
);
}

const selected = await this._harnessStateStore.selectVersion(version);
await this._activateHarnessBundle(selected);
await this._activateHarnessBundle(target, () => this._harnessStateStore!.selectVersion(version));
} finally {
this._harnessSwitchInProgress = false;
}
}

private async _activateHarnessBundle(bundle: HarnessVersionBundle): Promise<void> {
private async _activateHarnessBundle(
bundle: HarnessVersionBundle,
selectBundle?: () => Promise<HarnessVersionBundle>,
): Promise<void> {
await this._subagentController?.reset();
this._harnessVersionBundle = bundle;
const activatedBundle = selectBundle ? await selectBundle() : bundle;
this._harnessVersionBundle = activatedBundle;
this._baseSystemPrompt = this._rebuildSystemPrompt(this.getActiveToolNames());
this._systemPromptOverride = undefined;
this.agent.state.systemPrompt = this._baseSystemPrompt;
await this._notifyHarnessActivated(bundle.manifest.version);
await this._notifyHarnessActivated(activatedBundle.manifest.version);
}

private async _refreshSelectedHarnessAtPromptBoundary(): Promise<void> {
Expand Down Expand Up @@ -3768,32 +3771,33 @@ export class AgentSession {
// Switch leaf (with or without summary)
// Summary is attached at the navigation target position (newLeafId), not the old branch
let summaryEntry: BranchSummaryEntry | undefined;
if (summaryText) {
// Create summary at target position (can be null for root)
const summaryId = this.sessionManager.branchWithSummary(
newLeafId,
summaryText,
summaryDetails,
fromExtension,
summaryUsage,
);
summaryEntry = this.sessionManager.getEntry(summaryId) as BranchSummaryEntry;

// Attach label to the summary entry
if (label) {
this.sessionManager.appendLabelChange(summaryId, label);
const mutateLeaf = (): void => {
if (summaryText) {
const summaryId = this.sessionManager.branchWithSummary(
newLeafId,
summaryText,
summaryDetails,
fromExtension,
summaryUsage,
);
summaryEntry = this.sessionManager.getEntry(summaryId) as BranchSummaryEntry;
if (label) this.sessionManager.appendLabelChange(summaryId, label);
} else if (newLeafId === null) {
this.sessionManager.resetLeaf();
} else {
this.sessionManager.branch(newLeafId);
}
} else if (newLeafId === null) {
// No summary, navigating to root - reset leaf
this.sessionManager.resetLeaf();
} else {
// No summary, navigating to non-root
this.sessionManager.branch(newLeafId);
}

// Attach label to target entry when not summarizing (no summary entry to label)
if (label && !summaryText) {
this.sessionManager.appendLabelChange(targetId, label);
if (label && !summaryText) this.sessionManager.appendLabelChange(targetId, label);
};

try {
if (this._subagentController) await this._subagentController.reconcileBranch(mutateLeaf);
else mutateLeaf();
} catch (error) {
if (oldLeafId === null) this.sessionManager.resetLeaf();
else this.sessionManager.branch(oldLeafId);
throw error;
}

// Update agent state
Expand Down
1 change: 1 addition & 0 deletions packages/coding-agent/src/core/sdk.ts
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,7 @@ function createRootSubagentController(
return new SubagentController({
maxConcurrent,
telemetryContext,
restoreRoster: (controller, parent) => restoreRootSubagents(controller, parent, agentDir, parentSessionManager),
captureSession: (parent, childName, origin) => {
const config = captureSubagentSessionConfig(parent, agentDir);
const sessionManager = sessionManagerFactory.create(
Expand Down
144 changes: 104 additions & 40 deletions packages/coding-agent/src/core/subagents/controller.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import type { ToolDefinition } from "../extensions/types.ts";
import { startCodingAgentSpan } from "../telemetry.ts";
import { createAgentActionTool, createAgentsTool, createSpawnAgentTool } from "../tools/subagents.ts";
import type { ChildSnapshotV1 } from "./persistence.ts";
import { reportSubagentRuntimeWarning } from "./runtime-diagnostics.ts";
import type {
CapturedSubagentSession,
CaptureSubagentSession,
Expand Down Expand Up @@ -165,6 +166,7 @@ export class SubagentController {
readonly #maxConcurrent: number;
readonly #captureSession: CaptureSubagentSession | undefined;
readonly #persistRoster: ((agents: readonly ChildSnapshotV1[], admission: boolean) => void) | undefined;
readonly #restoreRoster: ((controller: SubagentController, parent: AgentSession) => void) | undefined;
readonly #telemetryContext: TelemetryContext;
readonly #records = new Map<string, SubagentRecord>();
readonly #listeners = new Set<SubagentLifecycleListener>();
Expand All @@ -187,6 +189,7 @@ export class SubagentController {
captureSession?: CaptureSubagentSession;
telemetryContext?: TelemetryContext;
persistRoster?: (agents: readonly ChildSnapshotV1[], admission: boolean) => void;
restoreRoster?: (controller: SubagentController, parent: AgentSession) => void;
}) {
if (!Number.isInteger(options.maxConcurrent) || options.maxConcurrent <= 0) {
throw new RangeError("maxConcurrent must be a positive integer");
Expand All @@ -195,6 +198,7 @@ export class SubagentController {
this.#captureSession = options.captureSession;
this.#telemetryContext = options.telemetryContext ?? NOOP_TELEMETRY_CONTEXT;
this.#persistRoster = options.persistRoster;
this.#restoreRoster = options.restoreRoster;
}

createToolDefinitions(): ToolDefinition[] {
Expand Down Expand Up @@ -730,29 +734,108 @@ export class SubagentController {
});
this.#resetPromise = resetPromise;

void this.#resetContext().then(resolveReset, rejectReset);
return resetPromise;
}

reconcileBranch(mutateLeaf: () => void): Promise<void> {
if (this.#closed) return Promise.reject(this.#closedError());
if (this.#resetting) return Promise.reject(this.#resetError());
return this.#runContextBoundary(async () => {
await this.#quiesceActiveRecords();
if (this.#closed) throw this.#closedError();
mutateLeaf();
await this.#replaceRecordsFromSelectedBranch();
});
}

async #resetContext(): Promise<void> {
try {
this.#resetting = true;
const resetError = this.#resetError();
this.#rejectPermitQueue(resetError);
const records = this.#detachRecords(resetError);
const cleanup = this.#cleanupRecords(records);
this.#resetCleanupPromise = cleanup;
this.#publish(
(revision) => ({
type: "subagent_controller_reset",
revision,
timestamp: Date.now(),
}),
"reset",
);
void this.#finishReset(cleanup).then(resolveReset, rejectReset);
} catch (error) {
if (!this.#closed) this.#resetting = false;
this.#resetCleanupPromise = undefined;
await this.#runContextBoundary(async () => {
await this.#quiesceActiveRecords();
if (this.#closed) throw this.#closedError();
this.#persistRoster?.([], true);
await this.#clearRecords();
if (this.#closed) throw this.#closedError();
});
} finally {
this.#resetPromise = undefined;
rejectReset(error);
}
return resetPromise;
}

async #runContextBoundary(operation: () => Promise<void>): Promise<void> {
this.#resetting = true;
this.#rejectPermitQueue(this.#resetError());
try {
await operation();
} finally {
if (!this.#closed) {
this.#resetting = false;
this.#drainPermitQueue();
}
}
}

async #quiesceActiveRecords(): Promise<void> {
const activeRecords = [...this.#records.values()].filter(
(record) => record.status === "queued" || record.status === "running",
);
const results = await Promise.allSettled(activeRecords.map((record) => this.#quiesceRecord(record)));
const failures = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
if (failures.length > 0) throw new AggregateError(failures, "Failed to quiesce subagent controller");
}

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 record.run;
if (record.status === "queued" || record.status === "running") this.markInterrupted(record.name);
}

async #clearRecords(): Promise<void> {
const records = this.#detachRecords(this.#resetError());
const cleanup = this.#cleanupRecords(records);
this.#resetCleanupPromise = cleanup;
this.#publish(
(revision) => ({
type: "subagent_controller_reset",
revision,
timestamp: Date.now(),
}),
"reset",
);
const failures = await cleanup;
if (this.#resetCleanupPromise === cleanup) this.#resetCleanupPromise = undefined;
this.#reportCleanupFailures(failures);
}

async #replaceRecordsFromSelectedBranch(): Promise<void> {
await this.#clearRecords();
if (this.#closed) return;
this.#resetting = false;
try {
this.#restoreRoster?.(this, this.#getBoundParent());
} catch (error) {
reportSubagentRuntimeWarning(
this.#getBoundParent().sessionManager,
`Subagent roster reconciliation failed: ${normalizeError(error)}`,
);
} finally {
if (!this.#closed) this.#resetting = true;
}
}

#reportCleanupFailures(failures: readonly unknown[]): void {
if (failures.length === 0 || !this.#parent) return;
reportSubagentRuntimeWarning(
this.#parent.sessionManager,
`Subagent context cleanup failed after reset: ${failures.map(normalizeError).join("; ")}`,
);
}

#transition(name: string, status: SubagentStatus, fields: TransitionFields = {}): void {
Expand Down Expand Up @@ -824,7 +907,7 @@ export class SubagentController {
}

#getRecordForUpdate(name: string): SubagentRecord | undefined {
if (this.#closed || this.#resetting) return undefined;
if (this.#closed) return undefined;
const record = this.#records.get(name);
if (!record) {
throw new Error(`Unknown subagent: ${name}`);
Expand Down Expand Up @@ -1411,25 +1494,6 @@ export class SubagentController {
return failures;
}

async #finishReset(cleanup: Promise<unknown[]>): Promise<void> {
const failures = await cleanup;
if (!this.#closed) {
this.#resetting = false;
this.#drainPermitQueue();
}
if (this.#resetCleanupPromise === cleanup) {
this.#resetCleanupPromise = undefined;
}
this.#resetPromise = undefined;

if (this.#closed) {
throw this.#closedError();
}
if (failures.length > 0) {
throw new AggregateError(failures, "Failed to reset subagent controller");
}
}

async #finishShutdown(shutdownCleanup: Promise<unknown[]>, resetCleanup?: Promise<unknown[]>): Promise<void> {
const [resetFailures, shutdownFailures] = await Promise.all([
resetCleanup ?? Promise.resolve([]),
Expand Down
45 changes: 34 additions & 11 deletions packages/coding-agent/test/subagent-controller.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -428,29 +428,37 @@ describe("SubagentController", () => {
const firstReset = controller.reset();
const secondReset = controller.reset();
expect(firstReset).toBe(secondReset);
expect(controller.snapshot().agents).toEqual([]);
expect(controller.snapshot().agents).toMatchObject([{ name: "worker", status: "running" }]);
expect(() => controller.reserve("later", "task")).toThrow("Subagent controller is resetting");
expect(() => controller.message("worker", "message")).toThrow("Subagent controller is resetting");
await expect(controller.queryAgents({ timeoutMs: 0 })).rejects.toThrow("Subagent controller is resetting");
await expect(controller.acquirePermit()).rejects.toThrow("Subagent controller is resetting");
await expect(queuedPermit).rejects.toThrow("Subagent controller is resetting");
await expect(waiter).resolves.toEqual({ reason: "reset" });
await expect(waiter).resolves.toMatchObject({
reason: "changed",
snapshot: { agents: [{ name: "worker", status: "interrupted" }] },
});
await expect(controller.waitForChange(0, { timeoutMs: 100 })).resolves.toEqual({ reason: "reset" });
activePermit.release();
await disposeStarted.promise;
expect(controller.snapshot().agents).toEqual([]);
expect(await isPending(firstReset)).toBe(true);
disposeGate.resolve();
await firstReset;

expect(dispose).toHaveBeenCalledOnce();
expect(events).toHaveLength(1);
expect(events[0]).toMatchObject({ type: "subagent_controller_reset" });
expect(JSON.stringify(events[0])).not.toContain("private task");
expect(events).toHaveLength(2);
expect(events.map((event) => event.type)).toEqual(["subagent_status_changed", "subagent_controller_reset"]);
expect(JSON.stringify(events)).not.toContain("private task");
expect(() => controller.reserve("worker", "reused task")).not.toThrow();
expect(events.map((event) => event.type)).toEqual(["subagent_controller_reset", "subagent_reserved"]);
expect(events.map((event) => event.type)).toEqual([
"subagent_status_changed",
"subagent_controller_reset",
"subagent_reserved",
]);
});

it("reopens after aggregating reset cleanup failures", async () => {
it("reopens after treating reset cleanup failures as best effort", async () => {
const controller = new SubagentController({ maxConcurrent: 2 });
const first = await harness();
const second = await harness();
Expand All @@ -462,19 +470,34 @@ describe("SubagentController", () => {
controller.reserve("first", "task");
controller.attachSession("first", first.session);
controller.markRunning("first");
controller.markIdle("first");
controller.reserve("second", "task");
controller.attachSession("second", second.session);
controller.markRunning("second");
controller.markIdle("second");

const error = await controller.reset().catch((reason: unknown) => reason);
expect(error).toBeInstanceOf(AggregateError);
expect((error as AggregateError).errors).toEqual([abortFailure, disposeFailure]);
await expect(controller.reset()).resolves.toBeUndefined();
expect(first.session.dispose).toHaveBeenCalledOnce();
expect(secondDispose).toHaveBeenCalledOnce();
expect(controller.snapshot().agents).toEqual([]);
expect(() => controller.reserve("first", "reused")).not.toThrow();
});

it("reopens admission without clearing the roster when quiescing fails", async () => {
const controller = new SubagentController({ maxConcurrent: 1 });
const child = await harness();
vi.spyOn(child.session, "abort").mockRejectedValue(new Error("quiesce failed"));
const dispose = vi.spyOn(child.session, "dispose").mockResolvedValue();
controller.reserve("worker", "task");
controller.attachSession("worker", child.session);
controller.markRunning("worker");

await expect(controller.reset()).rejects.toThrow("Failed to quiesce subagent controller");
expect(controller.snapshot().agents).toMatchObject([{ name: "worker", status: "running" }]);
expect(dispose).not.toHaveBeenCalled();
expect(() => controller.reserve("later", "task")).not.toThrow();
});

it("keeps shutdown permanent when it races reset", async () => {
const controller = new SubagentController({ maxConcurrent: 1 });
const child = await harness();
Expand Down Expand Up @@ -510,8 +533,8 @@ describe("SubagentController", () => {
});

const reset = controller.reset();
expect(nestedReset).toBe(reset);
await reset;
expect(nestedReset).toBe(reset);
const shutdown = controller.shutdown();
expect(nestedShutdown).toBe(shutdown);
await shutdown;
Expand Down
Loading
Loading