From 7a63a856d9bfc1235b5cd014f3fed82e0a38b0f4 Mon Sep 17 00:00:00 2001 From: Jason Date: Fri, 14 Aug 2026 22:26:13 +0900 Subject: [PATCH 1/2] fix(omp): keep parent non-idle while internal task subagents run OMP can end the parent model loop while `task` children are still writing. Do not complete the Paseo turn until the subagent index reports no runners. Reconcile successful get_subagents replies so listed ids stay running, and treat absence of a previously listed id as completed. Never-listed lifecycle children stay running so an empty first snapshot cannot fake-idle the parent. Closes #2232 --- .../server/agent/providers/omp/agent.test.ts | 200 ++++++++++++++++++ .../src/server/agent/providers/omp/agent.ts | 16 +- .../agent/providers/omp/cli-runtime.test.ts | 5 + .../server/agent/providers/omp/cli-runtime.ts | 7 + .../server/agent/providers/omp/rpc-types.ts | 31 +-- .../src/server/agent/providers/omp/runtime.ts | 2 + .../providers/omp/subagent-index.test.ts | 48 +++++ .../agent/providers/omp/subagent-index.ts | 58 +++++ .../providers/omp/test-utils/fake-omp.ts | 4 + 9 files changed, 358 insertions(+), 13 deletions(-) diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index 30876ae9aa..6e15ba5b1e 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -329,6 +329,206 @@ describe("OMP agent client and session", () => { await expect(completion).resolves.toMatchObject({ finalText: "first done" }); }); + // #2232: parent model loop can go idle while OMP-internal `task` children + // keep writing. Wire order is tool_execution_end (dispatch ack) then + // subagent_lifecycle started — never the reverse. + test("stays active while OMP internal task subagents are still running", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const session = omp; + await session.requireStartTurn("critically audit the entire repo"); + const runtime = session.runtime(); + runtime.beginTurn(); + runtime.acceptPrompt("critically audit the entire repo", "user-audit"); + runtime.streamAssistantText("spawning fan-out workers"); + runtime.emit({ + type: "tool_execution_start", + toolCallId: "task-1", + toolName: "task", + args: { description: "audit API budget" }, + }); + runtime.emit({ + type: "tool_execution_end", + toolCallId: "task-1", + toolName: "task", + isError: false, + result: { text: "Spawned 1 background agent" }, + }); + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "ApiBudgetAudit", + agent: "ApiBudgetAudit", + description: "audit API budget", + status: "started", + parentToolCallId: "task-1", + index: 0, + }, + }); + runtime.state = { ...runtime.state, isStreaming: false, isCompacting: false }; + runtime.finishTurn({ + role: "assistant", + content: [{ type: "text", text: "spawning fan-out workers" }], + }); + + await session.waitForProviderStateChecks(1); + await scheduler.waitForWaits(1); + expect(session.completedTurnCount()).toBe(0); + expect(session.subagentUpserts()).toContainEqual({ id: "ApiBudgetAudit", status: "running" }); + + scheduler.retry(); + await session.waitForProviderStateChecks(2); + await scheduler.waitForWaits(2); + expect(session.completedTurnCount()).toBe(0); + + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "ApiBudgetAudit", + agent: "ApiBudgetAudit", + status: "completed", + parentToolCallId: "task-1", + index: 0, + }, + }); + scheduler.retry(); + await session.waitForProviderStateChecks(3); + await waitForImmediate(); + expect(session.completedTurnCount()).toBe(1); + expect(session.subagentUpserts()).toContainEqual({ + id: "ApiBudgetAudit", + status: "completed", + }); + }); + + test("stays active when get_subagents reports running children without prior events", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + await omp.requireStartTurn("fan out"); + const runtime = omp.runtime(); + runtime.beginTurn(); + runtime.acceptPrompt("fan out", "user-fanout"); + runtime.streamAssistantText("delegating"); + runtime.subagents = [ + { + id: "PipelineFeedAudit", + index: 0, + agent: "PipelineFeedAudit", + status: "running", + parentToolCallId: "task-2", + }, + ]; + runtime.state = { ...runtime.state, isStreaming: false, isCompacting: false }; + runtime.finishTurn({ + role: "assistant", + content: [{ type: "text", text: "delegating" }], + }); + + await omp.waitForProviderStateChecks(1); + await scheduler.waitForWaits(1); + expect(omp.completedTurnCount()).toBe(0); + + runtime.subagents = []; + scheduler.retry(); + await omp.waitForProviderStateChecks(2); + await waitForImmediate(); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("does not treat an empty get_subagents reply as completion for never-listed children", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + await omp.requireStartTurn("audit"); + const runtime = omp.runtime(); + runtime.beginTurn(); + runtime.acceptPrompt("audit", "user-audit"); + runtime.streamAssistantText("working"); + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "OnlyLifecycle", + agent: "OnlyLifecycle", + status: "started", + index: 0, + }, + }); + runtime.subagents = []; + runtime.state = { ...runtime.state, isStreaming: false, isCompacting: false }; + runtime.finishTurn({ + role: "assistant", + content: [{ type: "text", text: "working" }], + }); + + await omp.waitForProviderStateChecks(1); + await scheduler.waitForWaits(1); + expect(omp.completedTurnCount()).toBe(0); + + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "OnlyLifecycle", + agent: "OnlyLifecycle", + status: "completed", + index: 0, + }, + }); + scheduler.retry(); + await omp.waitForProviderStateChecks(2); + await waitForImmediate(); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("keeps the parent active when get_subagents is unavailable", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + await omp.requireStartTurn("legacy omp"); + const runtime = omp.runtime(); + runtime.getSubagentsError = new Error("unknown command get_subagents"); + runtime.beginTurn(); + runtime.acceptPrompt("legacy omp", "user-legacy"); + runtime.streamAssistantText("delegating"); + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "LegacyChild", + agent: "LegacyChild", + status: "started", + index: 0, + }, + }); + runtime.state = { ...runtime.state, isStreaming: false, isCompacting: false }; + runtime.finishTurn({ + role: "assistant", + content: [{ type: "text", text: "delegating" }], + }); + + await omp.waitForProviderStateChecks(1); + await scheduler.waitForWaits(1); + expect(omp.completedTurnCount()).toBe(0); + + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "LegacyChild", + agent: "LegacyChild", + status: "completed", + index: 0, + }, + }); + scheduler.retry(); + await omp.waitForProviderStateChecks(2); + await waitForImmediate(); + expect(omp.completedTurnCount()).toBe(1); + }); + test("does not complete on OMP's extension-notice agent_end", async () => { const omp = new OmpHarness(); await omp.start(); diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index db996cf701..6aee02ae80 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -2158,7 +2158,9 @@ export class OmpAgentSession implements AgentSession { try { const state = await this.runtimeSession.getState(); this.state = state; - if (!state.isStreaming && !state.isCompacting) { + // Parent model idle is not enough: OMP-internal `task` children keep + // writing after agent_end / isStreaming=false (#2232). + if (!state.isStreaming && !state.isCompacting && !(await this.hasRunningOmpSubagents())) { this.completeTurn(turnId, messages); return; } @@ -2169,6 +2171,18 @@ export class OmpAgentSession implements AgentSession { } } + private async hasRunningOmpSubagents(): Promise { + try { + const snapshots = await this.runtimeSession.getSubagents(); + for (const event of this.subagentIndex.reconcileSnapshots(this.runtimeSession, snapshots)) { + this.emit(event); + } + } catch (error) { + this.logger.debug({ err: error }, "OMP get_subagents unavailable during idle gate"); + } + return this.subagentIndex.hasRunning(this.runtimeSession); + } + private async refreshState(): Promise { this.state = await this.runtimeSession.getState(); } diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts index 66cd3cb0c1..ce9950b7f2 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts @@ -243,14 +243,19 @@ describe("OMP CLI runtime", () => { const commands: Record[] = []; replyToCommands(child, (command) => { commands.push(command); + if (command.type === "get_subagents") { + return { subagents: [] }; + } return undefined; }); const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); await session.setSubagentSubscription("events"); + await session.getSubagents(); expect(commands.map(withoutRequestId)).toEqual([ { type: "set_subagent_subscription", level: "events" }, + { type: "get_subagents" }, ]); }); diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.ts index 97e01c937a..bb069e7fd7 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.ts @@ -28,6 +28,7 @@ import { OmpRuntimeEventSchema, OmpSessionStateSchema, OmpSessionStatsSchema, + OmpSubagentsResultSchema, type OmpThinkingLevel, type OmpAgentMessage, type OmpModel, @@ -40,6 +41,7 @@ import { type OmpRuntimeEvent, type OmpSessionState, type OmpSessionStats, + type OmpSubagentSnapshot, type OmpSubagentSubscriptionLevel, } from "./rpc-types.js"; @@ -275,6 +277,11 @@ class OmpCliRuntimeSession implements OmpRuntimeSession { await this.request({ type: "set_subagent_subscription", level }); } + async getSubagents(): Promise { + const data = OmpSubagentsResultSchema.parse(await this.request({ type: "get_subagents" })); + return data.subagents ?? []; + } + async setHostTools(tools: OmpRpcHostToolDefinition[]): Promise { const data = OmpHostToolsResultSchema.parse( await this.request({ type: "set_host_tools", tools }), diff --git a/packages/server/src/server/agent/providers/omp/rpc-types.ts b/packages/server/src/server/agent/providers/omp/rpc-types.ts index 63aaa7d265..9e87173d9e 100644 --- a/packages/server/src/server/agent/providers/omp/rpc-types.ts +++ b/packages/server/src/server/agent/providers/omp/rpc-types.ts @@ -512,6 +512,7 @@ export const OmpRpcCommandSchema = z.discriminatedUnion("type", [ z.object({ ...OmpCommandBase, type: z.literal("set_auto_compaction"), enabled: z.boolean() }), z.object({ ...OmpCommandBase, type: z.literal("abort") }), z.object({ ...OmpCommandBase, type: z.literal("get_state") }), + z.object({ ...OmpCommandBase, type: z.literal("get_subagents") }), z.object({ ...OmpCommandBase, type: z.literal("get_messages") }), z.object({ ...OmpCommandBase, type: z.literal("get_available_models") }), z.object({ @@ -562,6 +563,23 @@ export const OmpCommandsResultSchema = z export const OmpHostToolsResultSchema = z .object({ toolNames: z.array(z.string()).optional() }) .passthrough(); +export const OmpSubagentSnapshotSchema = z + .object({ + id: z.string(), + index: z.number().int().nonnegative(), + agent: z.string(), + description: z.string().optional(), + status: OmpSubagentStatusSchema, + task: z.string().optional(), + assignment: z.string().optional(), + sessionFile: z.string().optional(), + parentToolCallId: z.string().optional(), + lastUpdate: z.number().optional(), + }) + .passthrough(); +export const OmpSubagentsResultSchema = z + .object({ subagents: z.array(OmpSubagentSnapshotSchema).optional() }) + .passthrough(); export const OmpBranchResultSchema = z .object({ text: z.string().optional(), cancelled: z.boolean().optional() }) .passthrough(); @@ -613,18 +631,7 @@ export type OmpAvailableCommandsUpdateEvent = z.infer; export type OmpPromptAck = z.infer & { requestId?: string }; -export interface OmpSubagentSnapshot { - id: string; - index: number; - agent: string; - description?: string; - status: OmpSubagentStatus; - task?: string; - assignment?: string; - sessionFile?: string; - parentToolCallId?: string; - lastUpdate?: number; -} +export type OmpSubagentSnapshot = z.infer; export interface OmpSubagentMessagesResult { sessionFile: string; diff --git a/packages/server/src/server/agent/providers/omp/runtime.ts b/packages/server/src/server/agent/providers/omp/runtime.ts index dc8001e3ec..cfde0092ba 100644 --- a/packages/server/src/server/agent/providers/omp/runtime.ts +++ b/packages/server/src/server/agent/providers/omp/runtime.ts @@ -9,6 +9,7 @@ import type { OmpRuntimeEvent, OmpSessionState, OmpSessionStats, + OmpSubagentSnapshot, OmpSubagentSubscriptionLevel, OmpThinkingLevel, } from "./rpc-types.js"; @@ -52,6 +53,7 @@ export interface OmpRuntimeSession { setAutoCompaction(enabled: boolean): Promise; abort(): Promise; getState(): Promise; + getSubagents(): Promise; getMessages(): Promise; getAvailableModels(timeoutMs?: number | null): Promise; setModel(provider: string, modelId: string): Promise; diff --git a/packages/server/src/server/agent/providers/omp/subagent-index.test.ts b/packages/server/src/server/agent/providers/omp/subagent-index.test.ts index d112ff3843..a7fabf5354 100644 --- a/packages/server/src/server/agent/providers/omp/subagent-index.test.ts +++ b/packages/server/src/server/agent/providers/omp/subagent-index.test.ts @@ -114,4 +114,52 @@ describe("OMP provider subagent mapper", () => { })[0], ).toMatchObject({ event: { id: "child-1", status: "canceled" } }); }); + + test("treats a listed snapshot as running and its later absence as completed", () => { + const index = new OmpSubagentIndex(); + const parent = {}; + + expect( + index.reconcileSnapshots(parent, [ + { + id: "child-1", + index: 0, + agent: "ApiBudgetAudit", + status: "running", + parentToolCallId: "task-1", + }, + ]), + ).toEqual([ + expect.objectContaining({ + event: { + type: "upsert", + id: "child-1", + title: "ApiBudgetAudit", + description: null, + status: "running", + toolCallId: "task-1", + }, + }), + ]); + expect(index.hasRunning(parent)).toBe(true); + + expect(index.reconcileSnapshots(parent, [])[0]).toMatchObject({ + event: { type: "upsert", id: "child-1", status: "completed" }, + }); + expect(index.hasRunning(parent)).toBe(false); + }); + + test("does not complete a lifecycle-only child from an empty snapshot", () => { + const index = new OmpSubagentIndex(); + const parent = {}; + index.handleLifecycle(parent, { + id: "child-1", + agent: "worker", + status: "started", + index: 0, + }); + + expect(index.reconcileSnapshots(parent, [])).toEqual([]); + expect(index.hasRunning(parent)).toBe(true); + }); }); diff --git a/packages/server/src/server/agent/providers/omp/subagent-index.ts b/packages/server/src/server/agent/providers/omp/subagent-index.ts index 205cbe45ce..b3f14d9786 100644 --- a/packages/server/src/server/agent/providers/omp/subagent-index.ts +++ b/packages/server/src/server/agent/providers/omp/subagent-index.ts @@ -7,6 +7,7 @@ import type { OmpSubagentEventPayload, OmpSubagentLifecyclePayload, OmpSubagentProgressPayload, + OmpSubagentSnapshot, } from "./rpc-types.js"; interface OmpSubagentState { @@ -15,6 +16,7 @@ interface OmpSubagentState { resolvedModel: string | null; toolCallId: string | null; status: "running" | "completed" | "failed" | "canceled"; + seenInSnapshot: boolean; mapper: OmpHistoryMapper; } @@ -64,6 +66,54 @@ export class OmpSubagentIndex { ); } + hasRunning(parent: object): boolean { + const states = this.statesByParent.get(parent); + if (!states) { + return false; + } + for (const state of states.values()) { + if (state.status === "running") { + return true; + } + } + return false; + } + + /** + * Merge a successful `get_subagents` reply. That RPC lists only still-running + * children: an id that previously appeared and is now missing is finished. + * Never-listed lifecycle children stay running so an empty first reply cannot + * kill a child whose started frame beat the first snapshot. + */ + reconcileSnapshots(parent: object, snapshots: OmpSubagentSnapshot[]): AgentStreamEvent[] { + const present = new Set(); + const events: AgentStreamEvent[] = []; + + for (const snapshot of snapshots) { + present.add(snapshot.id); + const state = this.stateFor(parent, snapshot.id, snapshot.agent); + state.seenInSnapshot = true; + state.title = snapshot.agent || state.title; + state.description = snapshot.description ?? snapshot.assignment ?? state.description; + state.toolCallId = snapshot.parentToolCallId ?? state.toolCallId; + state.status = mapSnapshotStatus(snapshot.status); + events.push(this.upsert(snapshot.id, state.status, state)); + } + + const states = this.statesByParent.get(parent); + if (!states) { + return events; + } + for (const [id, state] of states) { + if (state.status !== "running" || !state.seenInSnapshot || present.has(id)) { + continue; + } + state.status = "completed"; + events.push(this.upsert(id, state.status, state)); + } + return events; + } + terminalizeRunning(parent: object): AgentStreamEvent[] { const states = this.statesByParent.get(parent); if (!states) { @@ -94,6 +144,7 @@ export class OmpSubagentIndex { resolvedModel: null, toolCallId: null, status: "running", + seenInSnapshot: false, mapper: new OmpHistoryMapper("omp", [], OMP_HISTORY_MAPPER_HOOKS), }; states.set(id, state); @@ -139,3 +190,10 @@ function mapProgressStatus( if (status === "completed" || status === "failed") return status; return status === "aborted" ? "canceled" : "running"; } + +function mapSnapshotStatus( + status: OmpSubagentSnapshot["status"], +): "running" | "completed" | "failed" | "canceled" { + if (status === "completed" || status === "failed") return status; + return status === "aborted" ? "canceled" : "running"; +} diff --git a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts index 0b17fe8c5b..21f94e5ba8 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts @@ -131,6 +131,7 @@ export class FakeOmpSession implements OmpRuntimeSession { compactError: Error | null = null; emitCompactEnd = true; getStateError: Error | null = null; + getSubagentsError: Error | null = null; promptAck: OmpPromptAck = {}; branchResponse: { text?: string; cancelled?: boolean } = { text: "" }; branchMessages: Array<{ entryId: string; text: string }> = []; @@ -357,6 +358,9 @@ export class FakeOmpSession implements OmpRuntimeSession { } async getSubagents(): Promise { + if (this.getSubagentsError) { + throw this.getSubagentsError; + } return this.subagents; } From fa9fc5e6244edc3252851f3132c49b34c3f56a84 Mon Sep 17 00:00:00 2001 From: Jason Date: Fri, 14 Aug 2026 22:35:35 +0900 Subject: [PATCH 2/2] fix(omp): hold task cards open until linked children finish OMP emits tool_execution_end for `task` as a dispatch ack, then starts children. Keep the parent call running and settle it from the subagent index once a linked child exists and none remain running. completeTurn force-settles a task that never produced a child. Wire-order tests emit the result before subagent_lifecycle started. --- .../server/agent/providers/omp/agent.test.ts | 132 ++++++++++++++++++ .../src/server/agent/providers/omp/agent.ts | 51 ++++++- .../providers/omp/subagent-index.test.ts | 26 ++++ .../agent/providers/omp/subagent-index.ts | 26 ++++ 4 files changed, 234 insertions(+), 1 deletion(-) diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index 6e15ba5b1e..19aaa5d75c 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -6,6 +6,14 @@ import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js"; import type { OmpUsagePollScheduler } from "./usage-poller.js"; import { OmpHarness } from "./test-utils/omp-harness.js"; +function lastToolCallStatus(omp: OmpHarness, callId: string): string | undefined { + const items = omp + .timeline() + .filter((item) => item.type === "tool_call" && item.callId === callId); + const last = items[items.length - 1]; + return last?.type === "tool_call" ? last.status : undefined; +} + class ManualIdleScheduler implements OmpProviderIdleScheduler { private readonly retries: Array<() => void> = []; private readonly waiters: Array<{ count: number; resolve: () => void }> = []; @@ -529,6 +537,130 @@ describe("OMP agent client and session", () => { expect(omp.completedTurnCount()).toBe(1); }); + test("holds a task tool call open when its OMP children appear after the result", async () => { + const omp = new OmpHarness(); + await omp.start(); + + await omp.requireStartTurn("fan out"); + const runtime = omp.runtime(); + runtime.beginTurn(); + runtime.acceptPrompt("fan out", "user-fanout"); + runtime.streamAssistantText("delegating"); + runtime.emit({ + type: "tool_execution_start", + toolCallId: "task-1", + toolName: "task", + args: { description: "spawn workers" }, + }); + runtime.emit({ + type: "tool_execution_end", + toolCallId: "task-1", + toolName: "task", + isError: false, + result: { text: "Spawned 1 background agent" }, + }); + expect(lastToolCallStatus(omp, "task-1")).toBe("running"); + expect(omp.runningToolCallIds()).toEqual(["task-1"]); + + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "Worker", + agent: "Worker", + status: "started", + parentToolCallId: "task-1", + index: 0, + }, + }); + expect(lastToolCallStatus(omp, "task-1")).toBe("running"); + expect(omp.runningToolCallIds()).toEqual(["task-1"]); + + runtime.emit({ + type: "subagent_progress", + payload: { + index: 0, + agent: "Worker", + parentToolCallId: "task-1", + progress: { id: "Worker", status: "running", recentOutput: ["still working"] }, + }, + }); + expect(lastToolCallStatus(omp, "task-1")).toBe("running"); + + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "Worker", + agent: "Worker", + status: "completed", + parentToolCallId: "task-1", + index: 0, + }, + }); + expect(lastToolCallStatus(omp, "task-1")).toBe("completed"); + expect(omp.runningToolCallIds()).toEqual([]); + }); + + test("force-settles a task that never produced a child when the turn completes", async () => { + const omp = new OmpHarness(); + await omp.start(); + + const session = omp; + await session.requireStartTurn("no child"); + const runtime = session.runtime(); + runtime.beginTurn(); + runtime.acceptPrompt("no child", "user-orphan"); + runtime.streamAssistantText("done"); + runtime.emit({ + type: "tool_execution_start", + toolCallId: "task-orphan", + toolName: "task", + args: { description: "never spawned" }, + }); + runtime.emit({ + type: "tool_execution_end", + toolCallId: "task-orphan", + toolName: "task", + isError: false, + result: { text: "Spawned 0 background agents" }, + }); + expect(lastToolCallStatus(session, "task-orphan")).toBe("running"); + + runtime.state = { ...runtime.state, isStreaming: false, isCompacting: false }; + runtime.finishTurn({ + role: "assistant", + content: [{ type: "text", text: "done" }], + }); + await waitForImmediate(); + await waitForImmediate(); + expect(session.completedTurnCount()).toBe(1); + expect(lastToolCallStatus(session, "task-orphan")).toBe("completed"); + expect(session.runningToolCallIds()).toEqual([]); + }); + + test("completes a failed task immediately", async () => { + const omp = new OmpHarness(); + await omp.start(); + + await omp.requireStartTurn("task fails"); + const runtime = omp.runtime(); + runtime.beginTurn(); + runtime.emit({ + type: "tool_execution_start", + toolCallId: "task-fail", + toolName: "task", + args: { description: "boom" }, + }); + runtime.emit({ + type: "tool_execution_end", + toolCallId: "task-fail", + toolName: "task", + isError: true, + result: { text: "spawn failed" }, + }); + expect(lastToolCallStatus(omp, "task-fail")).toBe("failed"); + expect(omp.runningToolCallIds()).toEqual([]); + }); + test("does not complete on OMP's extension-notice agent_end", async () => { const omp = new OmpHarness(); await omp.start(); diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 6aee02ae80..c24266d157 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -843,6 +843,10 @@ export class OmpAgentSession implements AgentSession { private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); private readonly activeToolCalls = new Map(); + private readonly deferredTaskResults = new Map< + string, + { toolCall: OmpTrackedToolCall; result: OmpToolResult } + >(); private readonly pendingExtensionUiRequests = new Map(); private activeAskUserDialog: ActiveAskUserDialog | null = null; private pendingCombinedAskUserResponse: PendingCombinedAskUserResponse | null = null; @@ -1173,6 +1177,7 @@ export class OmpAgentSession implements AgentSession { private clearOmpTurnState(): void { clearOmpHostToolState(this.runtimeSession); + this.deferredTaskResults.clear(); this.subagentCardTracker.clear(); } @@ -1641,6 +1646,7 @@ export class OmpAgentSession implements AgentSession { for (const mapped of this.subagentIndex.handleLifecycle(this.runtimeSession, payload)) { this.emit(mapped); } + this.settleDeferredTaskCalls(); return true; } if (event.type === "subagent_progress") { @@ -1653,6 +1659,7 @@ export class OmpAgentSession implements AgentSession { for (const mapped of this.subagentIndex.handleProgress(this.runtimeSession, payload)) { this.emit(mapped); } + this.settleDeferredTaskCalls(); return true; } if (event.type === "subagent_event") { @@ -1901,7 +1908,6 @@ export class OmpAgentSession implements AgentSession { ): void { const toolCall = this.activeToolCalls.get(event.toolCallId) ?? parseToolArgs(event.toolName, null); - this.activeToolCalls.delete(event.toolCallId); if (event.toolName === "ask_user") { this.activeAskUserDialog = null; @@ -1909,6 +1915,17 @@ export class OmpAgentSession implements AgentSession { } const result = parseToolResult(event.result); + // `task` tool_execution_end is a dispatch ack. Children start later, so keep + // the call active until the index has a linked child and none are running. + if (event.toolName === "task" && !event.isError) { + this.activeToolCalls.set(event.toolCallId, toolCall); + this.deferredTaskResults.set(event.toolCallId, { toolCall, result }); + this.emitToolCallEvent(event.toolCallId, toolCall, "running", result, null); + this.settleDeferredTaskCalls(); + return; + } + + this.activeToolCalls.delete(event.toolCallId); const error = event.isError ? event.result : null; const status = event.isError ? "failed" : "completed"; this.emitToolCallEvent(event.toolCallId, toolCall, status, result, error); @@ -1925,6 +1942,36 @@ export class OmpAgentSession implements AgentSession { } } + private settleDeferredTaskCalls(): void { + const pendingIds = Array.from(this.deferredTaskResults.keys()); + for (const toolCallId of pendingIds) { + if ( + this.subagentIndex.hasLinkedChild(this.runtimeSession, toolCallId) && + !this.subagentIndex.hasRunningLinkedTo(this.runtimeSession, toolCallId) + ) { + this.finalizeDeferredTask(toolCallId); + } + } + } + + private forceSettleDeferredTaskCalls(): void { + const pendingIds = Array.from(this.deferredTaskResults.keys()); + for (const toolCallId of pendingIds) { + this.finalizeDeferredTask(toolCallId); + } + } + + private finalizeDeferredTask(toolCallId: string): void { + const pending = this.deferredTaskResults.get(toolCallId); + if (!pending) { + return; + } + this.deferredTaskResults.delete(toolCallId); + this.activeToolCalls.delete(toolCallId); + this.emitToolCallEvent(toolCallId, pending.toolCall, "completed", pending.result, null); + this.subagentCardTracker.delete(toolCallId); + } + private emitCompactionTimeline(input: { turnId: string | undefined; item: Extract["item"]; @@ -2123,6 +2170,7 @@ export class OmpAgentSession implements AgentSession { } private completeTurn(turnId: string | undefined, messages: OmpAgentMessage[]): void { + this.forceSettleDeferredTaskCalls(); this.activeTurnId = null; this.activeClientMessageId = null; this.activeAssistantMessageId = null; @@ -2180,6 +2228,7 @@ export class OmpAgentSession implements AgentSession { } catch (error) { this.logger.debug({ err: error }, "OMP get_subagents unavailable during idle gate"); } + this.settleDeferredTaskCalls(); return this.subagentIndex.hasRunning(this.runtimeSession); } diff --git a/packages/server/src/server/agent/providers/omp/subagent-index.test.ts b/packages/server/src/server/agent/providers/omp/subagent-index.test.ts index a7fabf5354..8170652f17 100644 --- a/packages/server/src/server/agent/providers/omp/subagent-index.test.ts +++ b/packages/server/src/server/agent/providers/omp/subagent-index.test.ts @@ -162,4 +162,30 @@ describe("OMP provider subagent mapper", () => { expect(index.reconcileSnapshots(parent, [])).toEqual([]); expect(index.hasRunning(parent)).toBe(true); }); + + test("links children to their parent task call for deferred card settlement", () => { + const index = new OmpSubagentIndex(); + const parent = {}; + index.handleLifecycle(parent, { + id: "child-1", + agent: "worker", + status: "started", + parentToolCallId: "task-1", + index: 0, + }); + + expect(index.hasLinkedChild(parent, "task-1")).toBe(true); + expect(index.hasLinkedChild(parent, "task-other")).toBe(false); + expect(index.hasRunningLinkedTo(parent, "task-1")).toBe(true); + + index.handleLifecycle(parent, { + id: "child-1", + agent: "worker", + status: "completed", + parentToolCallId: "task-1", + index: 0, + }); + expect(index.hasLinkedChild(parent, "task-1")).toBe(true); + expect(index.hasRunningLinkedTo(parent, "task-1")).toBe(false); + }); }); diff --git a/packages/server/src/server/agent/providers/omp/subagent-index.ts b/packages/server/src/server/agent/providers/omp/subagent-index.ts index b3f14d9786..1bdba693a4 100644 --- a/packages/server/src/server/agent/providers/omp/subagent-index.ts +++ b/packages/server/src/server/agent/providers/omp/subagent-index.ts @@ -79,6 +79,32 @@ export class OmpSubagentIndex { return false; } + hasLinkedChild(parent: object, toolCallId: string): boolean { + const states = this.statesByParent.get(parent); + if (!states) { + return false; + } + for (const state of states.values()) { + if (state.toolCallId === toolCallId) { + return true; + } + } + return false; + } + + hasRunningLinkedTo(parent: object, toolCallId: string): boolean { + const states = this.statesByParent.get(parent); + if (!states) { + return false; + } + for (const state of states.values()) { + if (state.toolCallId === toolCallId && state.status === "running") { + return true; + } + } + return false; + } + /** * Merge a successful `get_subagents` reply. That RPC lists only still-running * children: an id that previously appeared and is now missing is finished.