Skip to content

feat(coding-agent): resume coordinated waits for escalations #119

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Issacboy123 merged 4 commits into from
Aug 28, 2026
Merged

feat(coding-agent): resume coordinated waits for escalations #119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bc51340
feat(coding-agent): resume subagent coordination on escalation
Issacboy123 Aug 21, 2026
a780100
fix(coding-agent): coordinate new batch after consumption
Issacboy123 Aug 21, 2026
6dd8d84
fix(coding-agent): clear consumed escalation exit reason
Issacboy123 Aug 21, 2026
b0cca68
fix(coding-agent): settle explicitly consumed batches
Issacboy123 Aug 21, 2026
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
154 changes: 125 additions & 29 deletions packages/coding-agent/src/core/agent-session.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -381,6 +381,15 @@ interface PendingSubagentEscalationEnvelope {
messageCharacters: number;
}

interface SubagentCoordination {
readonly batch: SubagentDeliveryBatch;
readonly names: readonly string[];
readonly terminalNotices: Set<string>;
currentWaitAbortController?: AbortController;
exitReason?: "escalation" | "cancellation";
waitStarted: boolean;
}

function estimateMessagesTokens(messages: AgentMessage[]): number {
let tokens = 0;
for (const message of messages) {
Expand Down Expand Up @@ -483,7 +492,7 @@ export class AgentSession {
private readonly _telemetryContext: TelemetryContext;
private readonly _lifecycleObserver: AgentSessionLifecycleObserver | undefined;
private _harnessSwitchInProgress = false;
private _subagentWaitAbortController: AbortController | undefined;
private _subagentCoordination: SubagentCoordination | undefined;
private _pendingSubagentEscalationEnvelope: PendingSubagentEscalationEnvelope | undefined;
private readonly _pendingIdleSubagentEscalations = new Map<string, number>();
private _agentRunAborted = false;
Expand Down Expand Up @@ -803,6 +812,8 @@ export class AgentSession {
if (event.type === "message_start" && this._pendingSubagentEscalationEnvelope?.message === event.message) {
// The exact queued object is now entering parent context. Seal it so later reports form a new envelope.
this._pendingSubagentEscalationEnvelope = undefined;
const coordination = this._subagentCoordination;
if (coordination?.exitReason === "escalation") coordination.exitReason = undefined;
}
// When a user message starts, check if it's from either queue and remove it BEFORE emitting
// This ensures the UI sees the updated queue state
Expand Down Expand Up @@ -1043,6 +1054,7 @@ export class AgentSession {
rejectDispose = reject;
});
this._disposePromise = disposePromise;
this._exitSubagentCoordination("cancellation");
let childCleanup: Promise<void> | undefined;
try {
childCleanup = this._subagentController?.shutdown();
Expand Down Expand Up @@ -1564,7 +1576,18 @@ export class AgentSession {
if (firstRun) await this.agent.prompt(messages);
else await this.agent.continue();
firstRun = false;
if (await this._handlePostAgentRun()) continue;
if (await this._handlePostAgentRun()) {
if (this._agentRunAborted) {
const coordination = this._subagentCoordination;
if (coordination) {
this._exitSubagentCoordination("cancellation");
await this._waitForSubagentCoordination(coordination);
}
break;
}
continue;
}
if (this._agentRunAborted) break;

// No await is permitted between the inactive mark and final queue recheck.
this._isAgentRunActive = false;
Expand Down Expand Up @@ -1628,46 +1651,98 @@ export class AgentSession {
};
}

private async _waitForSubagentBatch(
batch: SubagentDeliveryBatch,
): Promise<{ cancelled: boolean; results: readonly SubagentBatchResult[] }> {
private _getOrCreateSubagentCoordination(): SubagentCoordination | undefined {
if (this._subagentCoordination) return this._subagentCoordination;
const batch = this._subagentController?.freezeUnconsumedBatch();
if (!batch) return undefined;
const coordination: SubagentCoordination = {
batch,
names: Object.freeze(batch.attempts.map((attempt) => attempt.name)),
terminalNotices: new Set(),
waitStarted: false,
};
this._subagentCoordination = coordination;
return coordination;
}

private _exitSubagentCoordination(reason: "escalation" | "cancellation"): void {
const coordination = this._subagentCoordination;
if (!coordination) return;
if (reason === "cancellation" || coordination.exitReason === undefined) {
coordination.exitReason = reason;
}
coordination.currentWaitAbortController?.abort(new Error(`Subagent coordination ${reason}`));
}

private _clearSubagentCoordination(coordination: SubagentCoordination, cancelled: boolean): void {
if (this._subagentCoordination !== coordination) return;
this._subagentCoordination = undefined;
coordination.currentWaitAbortController = undefined;
if (coordination.waitStarted) {
this._emit({ type: "subagent_wait_end", names: coordination.names, cancelled });
}
}

private async _waitForSubagentCoordination(
coordination: SubagentCoordination,
): Promise<
| { reason: "terminal"; results: readonly SubagentBatchResult[] }
| { reason: "escalation" }
| { reason: "cancellation" }
> {
const controller = this._subagentController;
if (!controller) return { cancelled: false, results: Object.freeze([]) };
if (!controller) return { reason: "terminal", results: Object.freeze([]) };

const names = Object.freeze(batch.attempts.map((attempt) => attempt.name));
const noticed = new Set<string>();
const emitNotice = (result: SubagentBatchResult): void => {
if (noticed.has(result.name)) return;
noticed.add(result.name);
if (coordination.terminalNotices.has(result.name)) return;
coordination.terminalNotices.add(result.name);
this._emit(this._createSubagentNotice(result));
};
const waitAbortController = new AbortController();
this._subagentWaitAbortController = waitAbortController;
if (this._agentRunAborted) {
waitAbortController.abort(new Error("Subagent coordination cancelled"));
coordination.currentWaitAbortController = waitAbortController;
if (this._agentRunAborted || coordination.exitReason === "cancellation") {
coordination.exitReason = "cancellation";
waitAbortController.abort(new Error("Subagent coordination cancellation"));
} else if (coordination.exitReason === "escalation") {
waitAbortController.abort(new Error("Subagent coordination escalation"));
}
if (!coordination.waitStarted) {
coordination.waitStarted = true;
this._emit({ type: "subagent_wait_start", names: coordination.names });
}
this._emit({ type: "subagent_wait_start", names });

let cancelled = false;
try {
try {
await controller.waitForBatch(batch, {
await controller.waitForBatch(coordination.batch, {
signal: waitAbortController.signal,
onTerminal: emitNotice,
});
} catch (error) {
if (!waitAbortController.signal.aborted) throw error;
cancelled = true;
await controller.interruptBatch(batch);
await controller.waitForBatch(batch, { onTerminal: emitNotice });
const exitReason =
coordination.exitReason === "cancellation" || this._agentRunAborted
? "cancellation"
: coordination.exitReason;
if (!exitReason) throw error;
if (exitReason === "escalation") {
coordination.exitReason = undefined;
return { reason: "escalation" };
}
if (!this._disposePromise) {
await controller.interruptBatch(coordination.batch);
await controller.waitForBatch(coordination.batch, { onTerminal: emitNotice });
controller.consumeBatch(coordination.batch);
}
this._clearSubagentCoordination(coordination, true);
return { reason: "cancellation" };
}
const results = controller.consumeBatch(batch);
return { cancelled, results };
const results = controller.consumeBatch(coordination.batch);
this._clearSubagentCoordination(coordination, false);
return { reason: "terminal", results };
} finally {
if (this._subagentWaitAbortController === waitAbortController) {
this._subagentWaitAbortController = undefined;
if (coordination.currentWaitAbortController === waitAbortController) {
coordination.currentWaitAbortController = undefined;
}
this._emit({ type: "subagent_wait_end", names, cancelled });
}
}

Expand Down Expand Up @@ -1696,15 +1771,33 @@ export class AgentSession {
return true;
}

const batch = this._subagentController?.freezeUnconsumedBatch();
if (batch) {
const delivery = await this._waitForSubagentBatch(batch);
if (delivery.cancelled) return false;
// Reports already queued by this run must enter context before a new batch is frozen
// or an escalation-suspended coordination resumes waiting.
if (this.agent.hasQueuedMessages()) return true;

while (true) {
const coordination = this._getOrCreateSubagentCoordination();
if (!coordination) break;
if (coordination.exitReason === "cancellation" || this._agentRunAborted) {
this._exitSubagentCoordination("cancellation");
}
const delivery = await this._waitForSubagentCoordination(coordination);
if (delivery.reason === "cancellation") return false;
if (delivery.reason === "escalation") {
if (coordination.exitReason === "cancellation" || this._agentRunAborted) {
this._exitSubagentCoordination("cancellation");
await this._waitForSubagentCoordination(coordination);
return false;
}
return true;
}
if (delivery.results.length > 0) {
const message = this._createSubagentResultMessage(delivery.results);
await this.sendCustomMessage(message, { deliverAs: "steer" });
return true;
}
// Explicit agents consumption can empty this batch while the same continuation
// starts another coordinated child. Freeze that pending run before settling.
}

// The agent loop drains both queues before emitting agent_end. Any messages
Expand Down Expand Up @@ -2161,6 +2254,7 @@ export class AgentSession {
}
throw error;
}
this._exitSubagentCoordination("escalation");
return;
}

Expand Down Expand Up @@ -2289,13 +2383,15 @@ export class AgentSession {
* Abort current operation and wait for agent to become idle.
*/
async abort(): Promise<void> {
if (this._isAgentRunActive) this._agentRunAborted = true;
this._exitSubagentCoordination("cancellation");
return this._traceOperation("abort", () => this._abort());
}

private async _abort(): Promise<void> {
this.abortRetry();
if (this._isAgentRunActive) this._agentRunAborted = true;
this._subagentWaitAbortController?.abort(new Error("Subagent coordination cancelled"));
this._exitSubagentCoordination("cancellation");
this.agent.abort();
await this.waitForIdle();
}
Expand Down
11 changes: 11 additions & 0 deletions packages/coding-agent/src/core/subagents/controller.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -483,6 +483,14 @@ export class SubagentController {
let snapshot = this.snapshot();
while (true) {
if (options.signal?.aborted) throw options.signal.reason;
for (const attempt of batch.attempts) {
const run = this.#records.get(attempt.name)?.coordinatedRun;
if (!run || run.id !== attempt.runId || run.delivery === "consumed") {
// Explicit query consumption can release a frozen attempt before AgentSession
// resumes this wait, and a same-child retry can then replace its run identity.
notified.add(attempt.name);
}
}
const terminal = batch.attempts
.flatMap((attempt) => {
if (notified.has(attempt.name)) return [];
Expand Down Expand Up @@ -1227,6 +1235,9 @@ export class SubagentController {
await record.run;
this.#assertOpen();
}
if (wake && isTerminalStatus(record.status) && record.coordinatedRun?.delivery === "frozen") {
throw new Error(`Cannot wake subagent while its coordinated result is pending delivery: ${name}`);
}

if (record.status === "queued") {
throw new Error(`Cannot message queued subagent: ${name}`);
Expand Down
Loading
Loading