From bc51340fc2904988650c35311711883558be58cc Mon Sep 17 00:00:00 2001 From: issacboy123 Date: Fri, 21 Aug 2026 02:18:09 -0700 Subject: [PATCH 1/4] feat(coding-agent): resume subagent coordination on escalation --- .../coding-agent/src/core/agent-session.ts | 149 ++++++++-- .../src/core/subagents/controller.ts | 3 + .../suite/subagent-parent-messaging.test.ts | 262 +++++++++++++++++- 3 files changed, 384 insertions(+), 30 deletions(-) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index d4e027df5..66e09564c 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -381,6 +381,15 @@ interface PendingSubagentEscalationEnvelope { messageCharacters: number; } +interface SubagentCoordination { + readonly batch: SubagentDeliveryBatch; + readonly names: readonly string[]; + readonly terminalNotices: Set; + currentWaitAbortController?: AbortController; + exitReason?: "escalation" | "cancellation"; + waitStarted: boolean; +} + function estimateMessagesTokens(messages: AgentMessage[]): number { let tokens = 0; for (const message of messages) { @@ -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(); private _agentRunAborted = false; @@ -1043,6 +1052,7 @@ export class AgentSession { rejectDispose = reject; }); this._disposePromise = disposePromise; + this._exitSubagentCoordination("cancellation"); let childCleanup: Promise | undefined; try { childCleanup = this._subagentController?.shutdown(); @@ -1564,7 +1574,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; @@ -1628,46 +1649,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(); 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 }); } } @@ -1696,10 +1769,25 @@ 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; + + const coordination = this._getOrCreateSubagentCoordination(); + if (coordination) { + 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" }); @@ -2161,6 +2249,7 @@ export class AgentSession { } throw error; } + this._exitSubagentCoordination("escalation"); return; } @@ -2289,13 +2378,15 @@ export class AgentSession { * Abort current operation and wait for agent to become idle. */ async abort(): Promise { + if (this._isAgentRunActive) this._agentRunAborted = true; + this._exitSubagentCoordination("cancellation"); return this._traceOperation("abort", () => this._abort()); } private async _abort(): Promise { 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(); } diff --git a/packages/coding-agent/src/core/subagents/controller.ts b/packages/coding-agent/src/core/subagents/controller.ts index 46f35ea7b..629c3e1ec 100644 --- a/packages/coding-agent/src/core/subagents/controller.ts +++ b/packages/coding-agent/src/core/subagents/controller.ts @@ -1227,6 +1227,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}`); diff --git a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts index 4ec56ceb2..c737e4bab 100644 --- a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts +++ b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts @@ -1,7 +1,7 @@ import { mkdirSync } from "node:fs"; import { join } from "node:path"; import type { AgentTool } from "@earendil-works/diffex-agent-core"; -import { fauxAssistantMessage, fauxToolCall } from "@earendil-works/diffex-ai"; +import { type Context, fauxAssistantMessage, fauxToolCall } from "@earendil-works/diffex-ai"; import { Type } from "typebox"; import { Check } from "typebox/value"; import { afterEach, describe, expect, it, vi } from "vitest"; @@ -39,6 +39,14 @@ function testTool(name: string): ToolDefinition { }; } +function hasTool(context: Context, name: string): boolean { + return context.tools?.some((tool) => tool.name === name) ?? false; +} + +function hasToolResult(context: Context, name: string): boolean { + return context.messages.some((message) => message.role === "toolResult" && message.toolName === name); +} + describe("subagent parent messaging foundation", () => { const harnesses: Harness[] = []; const sessions: AgentSession[] = []; @@ -193,6 +201,13 @@ describe("subagent parent messaging foundation", () => { await started.promise; await waitForStatus(controller, name, "running"); expect(woken?.getActiveToolNames()).toContain("escalate_to_parent"); + await getTool(woken!, "escalate_to_parent").execute("restored-report", { + message: `${name} live report`, + }); + expect(session.messages.at(-1)).toMatchObject({ + customType: "subagent_escalations", + details: { names: [name] }, + }); await controller.interrupt(name); await controller.reset(); } @@ -325,4 +340,249 @@ describe("subagent parent messaging foundation", () => { await session.navigateTree(branchTarget.id); await expect(accept("worker", "capacity reopened after branch change")).resolves.toBeUndefined(); }); + + describe("coordination", () => { + it("resumes one frozen wait for escalation actions and delivers terminal results exactly once", async () => { + const { harness, session } = await setupRoot(); + const waitStarted = deferred(); + const childBlocked = deferred(); + const lifecycle: Array<{ type: "start" | "end"; cancelled?: boolean }> = []; + session.subscribe((event) => { + if (event.type === "subagent_wait_start") { + lifecycle.push({ type: "start" }); + waitStarted.resolve(); + } + if (event.type === "subagent_wait_end") lifecycle.push({ type: "end", cancelled: event.cancelled }); + }); + let rootProviderCalls = 0; + harness.setResponses( + Array.from({ length: 12 }, () => async (context: Context, options) => { + if (hasTool(context, "escalate_to_parent")) { + if (!hasToolResult(context, "escalate_to_parent")) { + await waitStarted.promise; + return fauxAssistantMessage( + fauxToolCall("escalate_to_parent", { message: "redirect before completion" }), + { stopReason: "toolUse" }, + ); + } + childBlocked.resolve(); + return new Promise((resolve) => { + const finish = (): void => + resolve(fauxAssistantMessage("child stopped", { stopReason: "aborted" })); + if (options?.signal?.aborted) finish(); + else options?.signal?.addEventListener("abort", finish, { once: true }); + }); + } + + rootProviderCalls += 1; + const serialized = JSON.stringify(context.messages); + if (serialized.includes("subagent_batch_results")) return fauxAssistantMessage("terminal synthesis"); + if (serialized.includes("subagent_escalations") && !hasToolResult(context, "agent_action")) { + return fauxAssistantMessage(fauxToolCall("agent_action", { target: "worker", action: "interrupt" }), { + stopReason: "toolUse", + }); + } + if (hasToolResult(context, "spawn_agent")) return fauxAssistantMessage("wait for delegated work"); + return fauxAssistantMessage( + fauxToolCall("spawn_agent", { name: "worker", task: "long delegated work" }), + { stopReason: "toolUse" }, + ); + }), + ); + + await session.prompt("delegate and react to material reports"); + await childBlocked.promise; + + expect(rootProviderCalls).toBe(5); + expect(lifecycle).toEqual([{ type: "start" }, { type: "end", cancelled: false }]); + expect( + session.messages.filter( + (message) => message.role === "custom" && message.customType === "subagent_escalations", + ), + ).toHaveLength(1); + expect( + session.messages.filter( + (message) => message.role === "custom" && message.customType === "subagent_results", + ), + ).toHaveLength(1); + expect(session.getLastAssistantText()).toBe("terminal synthesis"); + }); + + it("gates terminal frozen wake until explicit agents consumption", async () => { + const { harness, services, session } = await setupRoot(); + const controller = (session as unknown as { _subagentController: SubagentController })._subagentController; + const config = captureSubagentSessionConfig(session, services.agentDir); + harness.setResponses([fauxAssistantMessage("initial result"), fauxAssistantMessage("woken result")]); + controller.spawn( + "worker", + "initial task", + (signal, escalateToParent) => createSubagentSession(config, { signal, escalateToParent }), + { + type: "agent_tool_call", + toolCallId: "spawn", + assistantEntryId: "assistant", + contextEntryId: null, + }, + ); + await waitForStatus(controller, "worker", "idle"); + expect(controller.freezeUnconsumedBatch()).toBeDefined(); + + await expect(controller.message("worker", "replace frozen run", true)).rejects.toThrow( + "coordinated result is pending delivery", + ); + await getTool(session, "agents").execute("consume", { + timeout_ms: 0, + targets: ["worker"], + until: "all", + }); + await expect(controller.message("worker", "wake after consumption", true)).resolves.toMatchObject({ + status: "queued", + delivery: "prompt", + }); + await waitForStatus(controller, "worker", "idle"); + }); + + it("gives cancellation precedence over an admitted escalation continuation", async () => { + const { harness, session } = await setupRoot(); + const accept = (session as unknown as EscalationAcceptor)._acceptSubagentEscalation.bind(session); + const waitStarted = deferred(); + const lifecycle: Array<{ type: "start" | "end"; cancelled?: boolean }> = []; + session.subscribe((event) => { + if (event.type === "subagent_wait_start") { + lifecycle.push({ type: "start" }); + waitStarted.resolve(); + } + if (event.type === "subagent_wait_end") lifecycle.push({ type: "end", cancelled: event.cancelled }); + }); + let rootProviderCalls = 0; + const rootContexts: string[] = []; + harness.setResponses( + Array.from({ length: 6 }, () => async (context: Context, options) => { + if (!hasTool(context, "spawn_agent")) { + return new Promise((resolve) => { + const finish = (): void => + resolve(fauxAssistantMessage("child cancelled", { stopReason: "aborted" })); + if (options?.signal?.aborted) finish(); + else options?.signal?.addEventListener("abort", finish, { once: true }); + }); + } + rootProviderCalls += 1; + rootContexts.push(JSON.stringify(context.messages)); + if (hasToolResult(context, "spawn_agent")) return fauxAssistantMessage("wait for child"); + return fauxAssistantMessage(fauxToolCall("spawn_agent", { name: "worker", task: "block" }), { + stopReason: "toolUse", + }); + }), + ); + + const prompt = session.prompt("cancel coordinated work"); + await waitStarted.promise; + const accepted = accept("worker", "do not continue after cancellation"); + const cancellation = session.abort(); + expect( + (session as unknown as { _agentRunAborted: boolean; _subagentCoordination?: { exitReason?: string } }) + ._agentRunAborted, + ).toBe(true); + expect( + (session as unknown as { _subagentCoordination?: { exitReason?: string } })._subagentCoordination + ?.exitReason, + ).toBe("cancellation"); + await Promise.all([accepted, cancellation, prompt]); + + expect(rootContexts.filter((context) => context.includes("subagent_escalations"))).toEqual([]); + expect(rootProviderCalls).toBe(2); + expect(lifecycle).toEqual([{ type: "start" }, { type: "end", cancelled: true }]); + expect( + session.messages.some((message) => message.role === "custom" && message.customType === "subagent_results"), + ).toBe(false); + }); + }); + + describe("persistent integration", () => { + it("invalidates reconciled capabilities while retaining ordinary hidden history", async () => { + const { harness, services, session } = await setupRoot(); + harness.setResponses([fauxAssistantMessage("seed")]); + await session.prompt("seed history"); + const target = session.sessionManager.getBranch()[0]!; + const controller = (session as unknown as { _subagentController: SubagentController })._subagentController; + const config = captureSubagentSessionConfig(session, services.agentDir); + let child: AgentSession | undefined; + const started = deferred(); + harness.setResponses([ + fauxAssistantMessage(fauxToolCall("escalate_to_parent", { message: "persisted hidden report" }), { + stopReason: "toolUse", + }), + (_context, options) => + new Promise((resolve) => { + started.resolve(); + options?.signal?.addEventListener( + "abort", + () => resolve(fauxAssistantMessage("reconciled", { stopReason: "aborted" })), + { once: true }, + ); + }), + ]); + controller.spawn("worker", "work", async (signal, escalateToParent) => { + child = await createSubagentSession(config, { signal, escalateToParent }); + return child; + }); + await started.promise; + await waitForStatus(controller, "worker", "running"); + const tool = getTool(child!, "escalate_to_parent"); + expect(JSON.stringify(session.sessionManager.getEntries())).toContain("persisted hidden report"); + expect(JSON.stringify(child!.sessionManager.getEntries())).toContain("persisted hidden report"); + expect(JSON.stringify(controller.snapshot())).not.toContain("persisted hidden report"); + + await session.navigateTree(target.id); + await expect(tool.execute("stale", { message: "after reconciliation" })).rejects.toThrow(); + expect(JSON.stringify(controller.snapshot())).not.toContain("persisted hidden report"); + }); + + it("disposes a suspended coordination without continuation, result injection, or queued capacity", async () => { + const { harness, session } = await setupRoot(); + const accept = (session as unknown as EscalationAcceptor)._acceptSubagentEscalation.bind(session); + const waitStarted = deferred(); + const lifecycle: Array<{ type: "start" | "end"; cancelled?: boolean }> = []; + session.subscribe((event) => { + if (event.type === "subagent_wait_start") { + lifecycle.push({ type: "start" }); + waitStarted.resolve(); + } + if (event.type === "subagent_wait_end") lifecycle.push({ type: "end", cancelled: event.cancelled }); + }); + let rootProviderCalls = 0; + harness.setResponses( + Array.from({ length: 6 }, () => async (context: Context, options) => { + if (!hasTool(context, "spawn_agent")) { + return new Promise((resolve) => { + const finish = (): void => + resolve(fauxAssistantMessage("disposed child", { stopReason: "aborted" })); + if (options?.signal?.aborted) finish(); + else options?.signal?.addEventListener("abort", finish, { once: true }); + }); + } + rootProviderCalls += 1; + if (hasToolResult(context, "spawn_agent")) return fauxAssistantMessage("wait for child"); + return fauxAssistantMessage(fauxToolCall("spawn_agent", { name: "worker", task: "block" }), { + stopReason: "toolUse", + }); + }), + ); + + const prompt = session.prompt("dispose coordinated work"); + await waitStarted.promise; + const accepted = accept("worker", "discard on disposal"); + const disposal = session.dispose(); + await accepted; + await Promise.allSettled([prompt, disposal]); + + expect(rootProviderCalls).toBe(2); + expect(lifecycle).toEqual([{ type: "start" }, { type: "end", cancelled: true }]); + expect(session.pendingMessageCount).toBe(0); + expect(session.getSubagents()).toEqual([]); + expect( + session.messages.some((message) => message.role === "custom" && message.customType === "subagent_results"), + ).toBe(false); + }); + }); }); From a7801005947d23760760c96b94435078aabd8a3b Mon Sep 17 00:00:00 2001 From: issacboy123 Date: Fri, 21 Aug 2026 02:28:14 -0700 Subject: [PATCH 2/4] fix(coding-agent): coordinate new batch after consumption --- .../coding-agent/src/core/agent-session.ts | 7 +- .../suite/subagent-parent-messaging.test.ts | 83 +++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 66e09564c..fe255cfc3 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -1773,8 +1773,9 @@ export class AgentSession { // or an escalation-suspended coordination resumes waiting. if (this.agent.hasQueuedMessages()) return true; - const coordination = this._getOrCreateSubagentCoordination(); - if (coordination) { + while (true) { + const coordination = this._getOrCreateSubagentCoordination(); + if (!coordination) break; if (coordination.exitReason === "cancellation" || this._agentRunAborted) { this._exitSubagentCoordination("cancellation"); } @@ -1793,6 +1794,8 @@ export class AgentSession { 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 diff --git a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts index c737e4bab..1058836b6 100644 --- a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts +++ b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts @@ -442,6 +442,89 @@ describe("subagent parent messaging foundation", () => { await waitForStatus(controller, "worker", "idle"); }); + it("coordinates a new child after explicit consumption empties the frozen batch", async () => { + const { harness, session } = await setupRoot(); + const controller = (session as unknown as { _subagentController: SubagentController })._subagentController; + const firstWaitStarted = deferred(); + const firstFinished = deferred(); + const secondStarted = deferred(); + const releaseSecond = deferred(); + const secondWaitStarted = deferred(); + let waitStarts = 0; + session.subscribe((event) => { + if (event.type !== "subagent_wait_start") return; + waitStarts += 1; + if (waitStarts === 1) firstWaitStarted.resolve(); + if (waitStarts === 2) secondWaitStarted.resolve(); + }); + let rootProviderCalls = 0; + harness.setResponses( + Array.from({ length: 16 }, () => async (context: Context) => { + if (hasTool(context, "escalate_to_parent")) { + if (getLastUserText(context).includes("second delegated work")) { + secondStarted.resolve(); + await releaseSecond.promise; + return fauxAssistantMessage("second result"); + } + if (!hasToolResult(context, "escalate_to_parent")) { + await firstWaitStarted.promise; + return fauxAssistantMessage( + fauxToolCall("escalate_to_parent", { message: "consume first, start second" }), + { stopReason: "toolUse" }, + ); + } + firstFinished.resolve(); + return fauxAssistantMessage("first result"); + } + + rootProviderCalls += 1; + if (rootProviderCalls === 1) { + return fauxAssistantMessage( + fauxToolCall("spawn_agent", { name: "first", task: "first delegated work" }), + { stopReason: "toolUse" }, + ); + } + if (rootProviderCalls === 2) return fauxAssistantMessage("wait for first"); + if (rootProviderCalls === 3) { + await firstFinished.promise; + await waitForStatus(controller, "first", "idle"); + return fauxAssistantMessage( + fauxToolCall("agents", { timeout_ms: 0, targets: ["first"], until: "all" }), + { stopReason: "toolUse" }, + ); + } + if (rootProviderCalls === 4) { + return fauxAssistantMessage( + fauxToolCall("spawn_agent", { name: "second", task: "second delegated work" }), + { stopReason: "toolUse" }, + ); + } + if (rootProviderCalls === 5) return fauxAssistantMessage("wait for second"); + return fauxAssistantMessage("second synthesis"); + }), + ); + + const prompt = session.prompt("consume then delegate again"); + await secondStarted.promise; + await secondWaitStarted.promise; + let settled = false; + void prompt.finally(() => { + settled = true; + }); + await Promise.resolve(); + expect(settled).toBe(false); + + releaseSecond.resolve(); + await prompt; + expect(rootProviderCalls).toBe(6); + expect(session.getLastAssistantText()).toBe("second synthesis"); + expect( + session.messages.filter( + (message) => message.role === "custom" && message.customType === "subagent_results", + ), + ).toHaveLength(1); + }); + it("gives cancellation precedence over an admitted escalation continuation", async () => { const { harness, session } = await setupRoot(); const accept = (session as unknown as EscalationAcceptor)._acceptSubagentEscalation.bind(session); From 6dd8d84e6329474b2e2c1974a5c1182f704f3b50 Mon Sep 17 00:00:00 2001 From: issacboy123 Date: Fri, 21 Aug 2026 02:40:26 -0700 Subject: [PATCH 3/4] fix(coding-agent): clear consumed escalation exit reason --- .../coding-agent/src/core/agent-session.ts | 2 + .../suite/subagent-parent-messaging.test.ts | 97 +++++++++++++++++++ 2 files changed, 99 insertions(+) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index fe255cfc3..ec4a13be9 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -812,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 diff --git a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts index 1058836b6..76c510b10 100644 --- a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts +++ b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts @@ -408,6 +408,103 @@ describe("subagent parent messaging foundation", () => { expect(session.getLastAssistantText()).toBe("terminal synthesis"); }); + it("reblocks after successive reports without an empty provider continuation", async () => { + const { harness, session } = await setupRoot(); + const firstWaitStarted = deferred(); + const firstRootContinuationStarted = deferred(); + const secondReportAccepted = deferred(); + const secondRootContinuationFinished = deferred(); + const releaseChild = deferred(); + const lifecycle: Array<{ type: "start" | "end"; cancelled?: boolean }> = []; + session.subscribe((event) => { + if (event.type === "subagent_wait_start") { + lifecycle.push({ type: "start" }); + firstWaitStarted.resolve(); + } + if (event.type === "subagent_wait_end") lifecycle.push({ type: "end", cancelled: event.cancelled }); + }); + let rootProviderCalls = 0; + let terminalResultContexts = 0; + harness.setResponses( + Array.from({ length: 12 }, () => async (context: Context) => { + if (hasTool(context, "escalate_to_parent")) { + const reportResults = context.messages.filter( + (message) => message.role === "toolResult" && message.toolName === "escalate_to_parent", + ).length; + if (reportResults === 0) { + await firstWaitStarted.promise; + return fauxAssistantMessage(fauxToolCall("escalate_to_parent", { message: "first report" }), { + stopReason: "toolUse", + }); + } + if (reportResults === 1) { + await firstRootContinuationStarted.promise; + return fauxAssistantMessage(fauxToolCall("escalate_to_parent", { message: "second report" }), { + stopReason: "toolUse", + }); + } + secondReportAccepted.resolve(); + await releaseChild.promise; + return fauxAssistantMessage("child result"); + } + + rootProviderCalls += 1; + const serialized = JSON.stringify(context.messages); + if (rootProviderCalls === 1) { + return fauxAssistantMessage( + fauxToolCall("spawn_agent", { name: "worker", task: "report twice before finishing" }), + { stopReason: "toolUse" }, + ); + } + if (rootProviderCalls === 2) return fauxAssistantMessage("wait for reports"); + if (rootProviderCalls === 3) { + expect(serialized).toContain("first report"); + expect(serialized).not.toContain("second report"); + firstRootContinuationStarted.resolve(); + await secondReportAccepted.promise; + return fauxAssistantMessage("handled first report"); + } + if (rootProviderCalls === 4) { + expect(serialized).toContain("second report"); + secondRootContinuationFinished.resolve(); + return fauxAssistantMessage("handled second report"); + } + if (serialized.includes("subagent_batch_results")) { + terminalResultContexts += 1; + return fauxAssistantMessage("terminal synthesis"); + } + return fauxAssistantMessage("unexpected empty continuation"); + }), + ); + + const prompt = session.prompt("delegate and handle successive reports"); + await secondRootContinuationFinished.promise; + await vi.waitFor(() => { + expect( + ( + session as unknown as { + _subagentCoordination?: { currentWaitAbortController?: AbortController }; + } + )._subagentCoordination?.currentWaitAbortController, + ).toBeDefined(); + }); + expect(rootProviderCalls).toBe(4); + + releaseChild.resolve(); + await prompt; + expect(rootProviderCalls).toBe(5); + expect(terminalResultContexts).toBe(1); + expect(lifecycle).toEqual([{ type: "start" }, { type: "end", cancelled: false }]); + const terminalMessages = session.messages.flatMap((message) => + message.role === "custom" && message.customType === "subagent_results" ? [message] : [], + ); + expect(terminalMessages).toHaveLength(1); + expect(JSON.parse(terminalMessages[0]!.content as string).agents).toEqual([ + expect.objectContaining({ name: "worker", result: "child result" }), + ]); + expect(session.getLastAssistantText()).toBe("terminal synthesis"); + }); + it("gates terminal frozen wake until explicit agents consumption", async () => { const { harness, services, session } = await setupRoot(); const controller = (session as unknown as { _subagentController: SubagentController })._subagentController; From b0cca6823dec05b8e8dda1f769ccdfd956f0d6f8 Mon Sep 17 00:00:00 2001 From: issacboy123 Date: Fri, 21 Aug 2026 02:48:38 -0700 Subject: [PATCH 4/4] fix(coding-agent): settle explicitly consumed batches --- .../src/core/subagents/controller.ts | 8 ++ .../suite/subagent-parent-messaging.test.ts | 91 +++++++++++++++++++ 2 files changed, 99 insertions(+) diff --git a/packages/coding-agent/src/core/subagents/controller.ts b/packages/coding-agent/src/core/subagents/controller.ts index 629c3e1ec..3002d697a 100644 --- a/packages/coding-agent/src/core/subagents/controller.ts +++ b/packages/coding-agent/src/core/subagents/controller.ts @@ -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 []; diff --git a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts index 76c510b10..4b60324a6 100644 --- a/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts +++ b/packages/coding-agent/test/suite/subagent-parent-messaging.test.ts @@ -622,6 +622,97 @@ describe("subagent parent messaging foundation", () => { ).toHaveLength(1); }); + it("coordinates a same-child retry after explicit consumption releases the prior batch", async () => { + const { harness, session } = await setupRoot(); + const firstWaitStarted = deferred(); + const retryStarted = deferred(); + const releaseRetry = deferred(); + const secondWaitStarted = deferred(); + const lifecycle: Array<{ type: "start" | "end"; cancelled?: boolean }> = []; + session.subscribe((event) => { + if (event.type === "subagent_wait_start") { + lifecycle.push({ type: "start" }); + if (lifecycle.filter((entry) => entry.type === "start").length === 1) firstWaitStarted.resolve(); + else secondWaitStarted.resolve(); + } + if (event.type === "subagent_wait_end") lifecycle.push({ type: "end", cancelled: event.cancelled }); + }); + let rootProviderCalls = 0; + harness.setResponses( + Array.from({ length: 16 }, () => async (context: Context) => { + if (hasTool(context, "escalate_to_parent")) { + if (getLastUserText(context).includes("retry with correction")) { + retryStarted.resolve(); + await releaseRetry.promise; + return fauxAssistantMessage("retry result"); + } + if (!hasToolResult(context, "escalate_to_parent")) { + await firstWaitStarted.promise; + return fauxAssistantMessage( + fauxToolCall("escalate_to_parent", { message: "retry after inspection" }), + { stopReason: "toolUse" }, + ); + } + return fauxAssistantMessage("failed attempt", { + stopReason: "error", + errorMessage: "needs correction", + }); + } + + rootProviderCalls += 1; + const serialized = JSON.stringify(context.messages); + if (rootProviderCalls === 1) { + return fauxAssistantMessage( + fauxToolCall("spawn_agent", { name: "worker", task: "attempt delegated work" }), + { stopReason: "toolUse" }, + ); + } + if (rootProviderCalls === 2) return fauxAssistantMessage("wait for first attempt"); + if (rootProviderCalls === 3) { + expect(serialized).toContain("retry after inspection"); + return fauxAssistantMessage( + fauxToolCall("agents", { timeout_ms: 1_000, targets: ["worker"], until: "all" }), + { stopReason: "toolUse" }, + ); + } + if (rootProviderCalls === 4) { + return fauxAssistantMessage( + fauxToolCall("agent_action", { + target: "worker", + action: "message", + message: "retry with correction", + wake: true, + }), + { stopReason: "toolUse" }, + ); + } + if (rootProviderCalls === 5) return fauxAssistantMessage("wait for retry"); + if (serialized.includes("subagent_batch_results")) return fauxAssistantMessage("retry synthesis"); + return fauxAssistantMessage("unexpected continuation"); + }), + ); + + const prompt = session.prompt("consume failed result and coordinate its retry"); + await retryStarted.promise; + await secondWaitStarted.promise; + releaseRetry.resolve(); + await prompt; + + expect(rootProviderCalls).toBe(6); + expect(lifecycle).toEqual([ + { type: "start" }, + { type: "end", cancelled: false }, + { type: "start" }, + { type: "end", cancelled: false }, + ]); + expect(session.getLastAssistantText()).toBe("retry synthesis"); + expect( + session.messages.filter( + (message) => message.role === "custom" && message.customType === "subagent_results", + ), + ).toHaveLength(1); + }); + it("gives cancellation precedence over an admitted escalation continuation", async () => { const { harness, session } = await setupRoot(); const accept = (session as unknown as EscalationAcceptor)._acceptSubagentEscalation.bind(session);