From 7a63a856d9bfc1235b5cd014f3fed82e0a38b0f4 Mon Sep 17 00:00:00 2001 From: Jason Date: Fri, 14 Aug 2026 22:26:13 +0900 Subject: [PATCH 1/5] 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 30876ae9aa1..6e15ba5b1e4 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 db996cf7016..6aee02ae808 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 66cd3cb0c16..ce9950b7f24 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 97e01c937ae..bb069e7fd7a 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 63aaa7d2659..9e87173d9e0 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 dc8001e3ec5..cfde0092ba2 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 d112ff38431..a7fabf53540 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 205cbe45cec..b3f14d97869 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 0b17fe8c5bf..21f94e5ba84 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/5] 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 6e15ba5b1e4..19aaa5d75c1 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 6aee02ae808..c24266d1577 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 a7fabf53540..8170652f171 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 b3f14d97869..1bdba693a4d 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. From ddc95e700666a2ed71ff537d770fb8be6c0ffad1 Mon Sep 17 00:00:00 2001 From: Joe Shull <42043763+joeshull@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:58:59 -0600 Subject: [PATCH 3/5] fix(omp): bound the post-agent_end completion gate After an assistant-bearing agent_end, the OMP provider polled get_state every 10 ms waiting for a non-streaming, non-compacting state, swallowed every error, and had no deadline. A stale or unreachable state path left the turn showing as running until the user cancelled, and startTurn throws while a turn is active, so the session was wedged for new prompts. The only signal was a debug log, and the daemon runs at info. Give the gate budgets. waitForRetry now backs off 10 ms to 1 s and returns a decision: retry, or abandon because the 60 s wall-clock wait ran out or three consecutive get_state calls failed. Wall clock rather than a retry count, because each get_state carries the JSONL-RPC request timeout and counting attempts would stretch the real wait to tens of minutes. On abandon the turn fails with omp_provider_idle_timeout or omp_provider_state_unavailable, and the diagnostic carries both the last observed state and the last RPC error, so a stale provider is distinguishable from a lost state path. turn_failed already carries code and diagnostic, so no protocol change. The turn is cleared, so recovery is an ordinary prompt. Hold one gate per turn. Every agent_end opened another concurrent loop, and two loops that both observed idle completed the same turn twice. Autonomous cycles carry no turn ID and share a key of their own; they poll like any other gate. The gate holds the terminal payload and the newest agent_end overwrites it, so an error reported by a later cycle is not dropped in favour of the first snapshot. Terminalize in-flight work when the gate gives up. A stall leaves OMP's state unknown, so a tool call or subagent still marked running has no turn left to finish it. Refs getpaseo/paseo#3654 --- .../server/agent/providers/omp/agent.test.ts | 246 +++++++++++++++++- .../src/server/agent/providers/omp/agent.ts | 184 +++++++++++-- .../providers/omp/test-utils/omp-harness.ts | 15 ++ 3 files changed, 424 insertions(+), 21 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 19aaa5d75c1..26182d6b88d 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -2,7 +2,13 @@ import { describe, expect, test } from "vitest"; import { setImmediate as waitForImmediate } from "node:timers/promises"; import type { PaseoToolCatalog } from "../../tools/types.js"; -import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js"; +import type { + OmpNoTurnScheduler, + OmpProviderIdleAttempt, + OmpProviderIdleDecision, + OmpProviderIdleScheduler, +} from "./agent.js"; +import { createOmpProviderIdleScheduler } from "./agent.js"; import type { OmpUsagePollScheduler } from "./usage-poller.js"; import { OmpHarness } from "./test-utils/omp-harness.js"; @@ -17,15 +23,38 @@ function lastToolCallStatus(omp: OmpHarness, callId: string): string | undefined class ManualIdleScheduler implements OmpProviderIdleScheduler { private readonly retries: Array<() => void> = []; private readonly waiters: Array<{ count: number; resolve: () => void }> = []; + private readonly seen: OmpProviderIdleAttempt[] = []; private waitCount = 0; + private abandoned = false; - waitForRetry(): Promise { + constructor( + private readonly decide: (attempt: OmpProviderIdleAttempt) => OmpProviderIdleDecision = () => ({ + retry: true, + }), + ) {} + + waitForRetry(attempt: OmpProviderIdleAttempt): Promise { this.waitCount += 1; + this.seen.push(attempt); for (const waiter of this.waiters.splice(0)) { if (this.waitCount >= waiter.count) waiter.resolve(); else this.waiters.push(waiter); } - return new Promise((resolve) => this.retries.push(resolve)); + const decision = this.decide(attempt); + if (this.abandoned) { + // Without this the caller spins with no timer and exhausts memory before + // the test reports anything. + throw new Error("OMP kept polling after the idle scheduler abandoned the gate"); + } + if (!decision.retry) { + this.abandoned = true; + return Promise.resolve(decision); + } + return new Promise((resolve) => this.retries.push(() => resolve(decision))); + } + + attempts(): OmpProviderIdleAttempt[] { + return this.seen; } waitForWaits(count: number): Promise { @@ -38,6 +67,10 @@ class ManualIdleScheduler implements OmpProviderIdleScheduler { if (!resolve) throw new Error("OMP has not requested an idle-state retry"); resolve(); } + + retryAll(): void { + for (const resolve of this.retries.splice(0)) resolve(); + } } class ManualNoTurnScheduler implements OmpNoTurnScheduler { @@ -337,6 +370,213 @@ describe("OMP agent client and session", () => { await expect(completion).resolves.toMatchObject({ finalText: "first done" }); }); + test("completes once when OMP repeats agent_end for the same turn", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await omp.waitForProviderStateChecks(2); + await scheduler.waitForWaits(1); + + // OMP can end a second cycle for the same prompt; the terminal assistant + // message it already streamed would otherwise open a second gate. + omp.runtime().finishTurn(); + for (let flush = 0; flush < 5; flush += 1) await waitForImmediate(); + expect(scheduler.attempts()).toHaveLength(1); + + omp.reportProviderState({ isStreaming: false, isCompacting: false }); + scheduler.retryAll(); + await expect(completion).resolves.toMatchObject({ finalText: "first done" }); + await waitForImmediate(); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("fails the turn when OMP never reports idle", async () => { + const scheduler = new ManualIdleScheduler((attempt) => + attempt.attempt < 2 ? { retry: true } : { retry: false, reason: "wait_budget" }, + ); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + scheduler.retryAll(); + + await expect(completion).rejects.toThrow(/idle/i); + expect(omp.completedTurnCount()).toBe(0); + expect(omp.failedTurns()).toMatchObject([ + { + code: "omp_provider_idle_timeout", + diagnostic: expect.stringContaining("isStreaming=true"), + }, + ]); + }); + + test("fails the turn when OMP state checks keep failing", async () => { + const scheduler = new ManualIdleScheduler((attempt) => + attempt.consecutiveFailures < 2 + ? { retry: true } + : { retry: false, reason: "failure_budget" }, + ); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + omp.failProviderStateChecks(new Error("state unavailable")); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + scheduler.retryAll(); + + await expect(completion).rejects.toThrow(/state/i); + expect(omp.completedTurnCount()).toBe(0); + expect(omp.failedTurns()).toMatchObject([ + { + code: "omp_provider_state_unavailable", + diagnostic: expect.stringContaining("state unavailable"), + }, + ]); + }); + + test("accepts a new prompt after a stalled turn fails", async () => { + const scheduler = new ManualIdleScheduler((attempt) => + attempt.attempt < 2 ? { retry: true } : { retry: false, reason: "wait_budget" }, + ); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + scheduler.retryAll(); + await expect(completion).rejects.toThrow(/idle/i); + + omp.reportProviderState({ isStreaming: false, isCompacting: false }); + await expect(omp.runPrompt("second", "second done")).resolves.toMatchObject({ + finalText: "second done", + }); + }); + + test("completes an autonomous turn once when OMP repeats agent_end", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + omp.startAutonomousTurnUntilProviderIdle("autonomous done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + + omp.runtime().finishTurn(); + for (let flush = 0; flush < 5; flush += 1) await waitForImmediate(); + expect(scheduler.attempts()).toHaveLength(1); + + omp.reportProviderState({ isStreaming: false, isCompacting: false }); + scheduler.retryAll(); + for (let flush = 0; flush < 5; flush += 1) await waitForImmediate(); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("reports the last OMP state when the idle wait budget expires after a failed check", async () => { + const scheduler = new ManualIdleScheduler((attempt) => + attempt.attempt < 3 ? { retry: true } : { retry: false, reason: "wait_budget" }, + ); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + omp.failProviderStateChecks(new Error("state unavailable")); + scheduler.retryAll(); + await scheduler.waitForWaits(2); + scheduler.retryAll(); + + await expect(completion).rejects.toThrow(/idle/i); + expect(omp.failedTurns()).toMatchObject([ + { + code: "omp_provider_idle_timeout", + diagnostic: expect.stringContaining("isStreaming=true"), + }, + ]); + // The failing check is still evidence; it must not be dropped. + expect(omp.failedTurns()[0]?.diagnostic).toContain("state unavailable"); + }); + + test("fails the turn with the newest agent_end error", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + + omp.runtime().finishTurn({ role: "assistant", content: [], errorMessage: "provider exploded" }); + for (let flush = 0; flush < 5; flush += 1) await waitForImmediate(); + + omp.reportProviderState({ isStreaming: false, isCompacting: false }); + scheduler.retryAll(); + + await expect(completion).rejects.toThrow(/provider exploded/); + expect(omp.completedTurnCount()).toBe(0); + }); + + test("the default idle scheduler stops on its wait and failure budgets", async () => { + const scheduler = createOmpProviderIdleScheduler(); + + await expect( + scheduler.waitForRetry({ attempt: 1, consecutiveFailures: 0, elapsedMs: 0 }), + ).resolves.toEqual({ retry: true }); + await expect( + scheduler.waitForRetry({ attempt: 200, consecutiveFailures: 0, elapsedMs: 60_000 }), + ).resolves.toEqual({ retry: false, reason: "wait_budget" }); + await expect( + scheduler.waitForRetry({ attempt: 3, consecutiveFailures: 3, elapsedMs: 100 }), + ).resolves.toEqual({ retry: false, reason: "failure_budget" }); + }); + + test("cancels in-flight tool calls when a turn stalls", async () => { + const scheduler = new ManualIdleScheduler((attempt) => + attempt.attempt < 2 ? { retry: true } : { retry: false, reason: "wait_budget" }, + ); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + omp.runtime().emit({ + type: "tool_execution_start", + toolCallId: "tool-1", + toolName: "bash", + args: { command: "sleep 30" }, + }); + expect(omp.runningToolCallIds()).toEqual(["tool-1"]); + + await scheduler.waitForWaits(1); + scheduler.retryAll(); + await expect(completion).rejects.toThrow(/idle/i); + + expect(omp.runningToolCallIds()).toEqual([]); + }); + // #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. diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index c24266d1577..7d4c77109cc 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -139,8 +139,26 @@ export interface OmpAgentClientOptions { usagePollScheduler?: OmpUsagePollScheduler; } +export interface OmpProviderIdleAttempt { + /** State checks already made for this completion gate, 1-based. */ + attempt: number; + /** Consecutive `get_state` rejections; any successful response resets it. */ + consecutiveFailures: number; + /** Wall-clock time since the gate opened, covering waits and `get_state`. */ + elapsedMs: number; +} + +/** + * `reason` names the exhausted budget so the turn reports why it stopped rather + * than inferring it from whichever check happened to be last. + */ +export type OmpProviderIdleDecision = + | { retry: true } + | { retry: false; reason: "wait_budget" | "failure_budget" }; + export interface OmpProviderIdleScheduler { - waitForRetry(): Promise; + /** Wait before the next state check, or abandon the gate so the turn fails. */ + waitForRetry(attempt: OmpProviderIdleAttempt): Promise; } export interface OmpNoTurnScheduler { @@ -194,10 +212,39 @@ interface OmpAgentSessionOptions { live?: boolean; } -function createOmpProviderIdleScheduler(): OmpProviderIdleScheduler { +// OMP processes a state request only once its RPC loop is promptable again, so +// the first checks after agent_end usually miss. Poll fast at first, then back +// off: a stalled provider otherwise costs 100 RPCs a second for as long as the +// stall lasts. +// Autonomous OMP cycles carry no turn ID; they still need a gate key of their own. +const OMP_AUTONOMOUS_GATE_KEY = "autonomous"; + +const OMP_PROVIDER_IDLE_MIN_RETRY_MS = 10; +const OMP_PROVIDER_IDLE_MAX_RETRY_MS = 1_000; +// Wall clock, not a retry count: each get_state carries the JSONL-RPC request +// timeout, so counting attempts would let a slow-but-answering state path stretch +// the wait to tens of minutes. +const OMP_PROVIDER_IDLE_BUDGET_MS = 60_000; +// A get_state rejection already costs a full RPC timeout, so a few in a row are +// enough evidence that the state path is gone. +const OMP_PROVIDER_IDLE_FAILURE_BUDGET = 3; + +function ompProviderIdleRetryDelayMs(attempt: number): number { + const backoff = OMP_PROVIDER_IDLE_MIN_RETRY_MS * 2 ** (attempt - 1); + return Math.min(OMP_PROVIDER_IDLE_MAX_RETRY_MS, backoff); +} + +export function createOmpProviderIdleScheduler(): OmpProviderIdleScheduler { return { - waitForRetry: async () => { - await new Promise((resolve) => setTimeout(resolve, 10)); + waitForRetry: async ({ attempt, consecutiveFailures, elapsedMs }) => { + if (consecutiveFailures >= OMP_PROVIDER_IDLE_FAILURE_BUDGET) { + return { retry: false, reason: "failure_budget" }; + } + if (elapsedMs >= OMP_PROVIDER_IDLE_BUDGET_MS) { + return { retry: false, reason: "wait_budget" }; + } + await delay(ompProviderIdleRetryDelayMs(attempt)); + return { retry: true }; }, }; } @@ -855,6 +902,7 @@ export class OmpAgentSession implements AgentSession { private activeAssistantMessageId: string | null = null; private activeTurnTerminalAssistantMessage: OmpAgentMessage | null = null; private activeTurnStarted = false; + private providerIdleGate: { key: string; messages: OmpAgentMessage[] } | null = null; private activeTurnHasUserMessage = false; private activeNoTurnPromptText: string | null = null; private readonly pendingNoTurnOutputs: Array<{ turnId: string; message: string }> = []; @@ -1161,6 +1209,7 @@ export class OmpAgentSession implements AgentSession { return; } this.closed = true; + this.providerIdleGate = null; this.usagePoller.close(); this.cancelNoTurnPromptCompletion(); try { @@ -1894,7 +1943,11 @@ export class OmpAgentSession implements AgentSession { } // A state request is processed after OMP's RPC loop becomes promptable, // so do not advertise Paseo idle until it reports that transition. - void this.completeTurnAfterProviderIdle(turnId, terminalMessages); + void this.completeTurnAfterProviderIdle(turnId, terminalMessages).catch( + (error: unknown) => { + this.logger.warn({ err: error }, "OMP provider idle gate failed"); + }, + ); return; } default: @@ -2169,8 +2222,7 @@ export class OmpAgentSession implements AgentSession { }); } - private completeTurn(turnId: string | undefined, messages: OmpAgentMessage[]): void { - this.forceSettleDeferredTaskCalls(); + private resetActiveTurn(): void { this.activeTurnId = null; this.activeClientMessageId = null; this.activeAssistantMessageId = null; @@ -2178,6 +2230,11 @@ export class OmpAgentSession implements AgentSession { this.activeTurnStarted = false; this.activeTurnHasUserMessage = false; this.clearNoTurnBuffers(); + } + + private completeTurn(turnId: string | undefined, messages: OmpAgentMessage[]): void { + this.forceSettleDeferredTaskCalls(); + this.resetActiveTurn(); const errorMessage = latestOmpErrorMessage(messages); if (typeof errorMessage === "string" && errorMessage.length > 0) { this.usagePoller.stopTurn(); @@ -2202,20 +2259,65 @@ export class OmpAgentSession implements AgentSession { turnId: string | undefined, messages: OmpAgentMessage[], ): Promise { - while (!this.closed && this.activeTurnStarted && this.currentTurnIdForEvent() === turnId) { - try { - const state = await this.runtimeSession.getState(); - this.state = state; - // 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); + // OMP can end more than one cycle for a single prompt. Without this gate a + // second loop races the first and completes the same turn twice. Autonomous + // cycles carry no turn ID, so they share one key: their loop polls too. + const key = turnId ?? OMP_AUTONOMOUS_GATE_KEY; + if (this.providerIdleGate?.key === key) { + // The newest agent_end owns the terminal payload; the running loop reads it + // at completion time so a late provider error is not dropped. + this.providerIdleGate.messages = messages; + return; + } + const gate = { key, messages }; + this.providerIdleGate = gate; + try { + const startedAt = Date.now(); + let attempt = 0; + let consecutiveFailures = 0; + let lastState: OmpSessionState | null = null; + let lastError: unknown = null; + while (!this.closed && this.activeTurnStarted && this.currentTurnIdForEvent() === turnId) { + attempt += 1; + try { + const state = await this.runtimeSession.getState(); + this.state = state; + lastState = state; + consecutiveFailures = 0; + // 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, gate.messages); + return; + } + } catch (error) { + lastError = error; + consecutiveFailures += 1; + this.logger.debug( + { err: error }, + "OMP state unavailable while waiting for provider idle", + ); + } + const decision = await this.providerIdleScheduler.waitForRetry({ + attempt, + consecutiveFailures, + elapsedMs: Date.now() - startedAt, + }); + if (!decision.retry) { + this.failStalledTurn(turnId, { + reason: decision.reason, + attempt, + consecutiveFailures, + lastState, + lastError, + }); return; } - } catch (error) { - this.logger.debug({ err: error }, "OMP state unavailable while waiting for provider idle"); } - await this.providerIdleScheduler.waitForRetry(); + } finally { + if (this.providerIdleGate === gate) { + this.providerIdleGate = null; + } } } @@ -2232,6 +2334,52 @@ export class OmpAgentSession implements AgentSession { return this.subagentIndex.hasRunning(this.runtimeSession); } + private failStalledTurn( + turnId: string | undefined, + context: { + reason: "wait_budget" | "failure_budget"; + attempt: number; + consecutiveFailures: number; + lastState: OmpSessionState | null; + lastError: unknown; + }, + ): void { + const stateUnavailable = context.reason === "failure_budget"; + // Both observations go into every diagnostic: which budget ran out says what + // Paseo did, the last state and the last error say what OMP was doing. + const details = [`state checks: ${context.attempt}`]; + if (context.lastState) { + details.push( + `last OMP state: isStreaming=${context.lastState.isStreaming}, isCompacting=${context.lastState.isCompacting}`, + ); + } + if (context.lastError) { + details.push( + `last get_state error after ${context.consecutiveFailures} consecutive failures: ${toDiagnosticErrorMessage(context.lastError)}`, + ); + } + const diagnostic = details.join("; "); + this.logger.warn( + { turnId, reason: context.reason, diagnostic }, + "OMP never reported an idle state after ending its response", + ); + // A stall leaves OMP's state unknown, so anything still marked running has no + // turn left to finish it. + this.terminalizeActiveWork(); + this.resetActiveTurn(); + this.usagePoller.stopTurn(); + this.emit({ + type: "turn_failed", + provider: this.provider, + turnId, + error: stateUnavailable + ? "OMP finished its response but its state is unavailable, so Paseo cannot confirm the turn ended." + : "OMP finished its response but never reported an idle state, so Paseo stopped waiting for the turn to end.", + code: stateUnavailable ? "omp_provider_state_unavailable" : "omp_provider_idle_timeout", + diagnostic, + }); + } + private async refreshState(): Promise { this.state = await this.runtimeSession.getState(); } diff --git a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts index 30d75bb8f4b..8512ba62476 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts @@ -247,6 +247,17 @@ export class OmpHarness { return { completion: run }; } + startAutonomousTurnUntilProviderIdle( + output: string, + providerState: { isStreaming: boolean; isCompacting: boolean }, + ): void { + const runtime = this.omp.latestSession(); + runtime.beginTurn(); + runtime.streamAssistantText(output); + runtime.state = { ...runtime.state, ...providerState }; + runtime.finishTurn(); + } + waitForProviderStateChecks(count: number): Promise { return this.omp.latestSession().waitForStateRequests(count); } @@ -406,6 +417,10 @@ export class OmpHarness { return items; } + failedTurns(): Array> { + return this.events.flatMap((event) => (event.type === "turn_failed" ? [event] : [])); + } + completedTurnCount(): number { return this.events.filter((event) => event.type === "turn_completed").length; } From 870c0c7e4a6012f23e41b00536cde09e746c7294 Mon Sep 17 00:00:00 2001 From: Joe Shull <42043763+joeshull@users.noreply.github.com> Date: Fri, 21 Aug 2026 15:12:45 -0600 Subject: [PATCH 4/5] fix(omp): keep the idle gate from failing a turn it no longer owns The gate re-checked ownership at the top of each poll, but the abandon path ran after two awaits without re-checking. A get_state carries the 30 s RPC timeout, so the window is wide: cancel a stuck turn, send another one, and the stale gate could fail the cancelled turn, cancel the new turn's tool calls, and null its turn ID. Events for the live turn then went out with no turn ID, which the manager back-fills with whatever turn is active. Check ownership before failing, and before completing for the same reason. Give compaction its own budget. The gate waits for isCompacting to clear, and compacting a large context is a model call that outlasts 60 s routinely, so a healthy provider could be failed mid-compaction. The observed compaction state now reaches the scheduler, which allows ten minutes while OMP reports it. Report an unavailable state path whenever no state was ever observed. A hanging get_state burns the 60 s wait budget before three rejections can land, so the failure budget only fired on fast rejections and the hanging case was labelled a timeout. Measure the budget on performance.now(). Date.now() steps with NTP and across suspend, either deferring the deadline or failing a healthy turn on wake. Keep an error the gate already holds when a later agent_end reports none: the fallback payload is a single assistant message, so newest-wins could drop the error the previous cycle reported. Refs getpaseo/paseo#3654 --- .../server/agent/providers/omp/agent.test.ts | 146 +++++++++++++++++- .../src/server/agent/providers/omp/agent.ts | 77 +++++++-- 2 files changed, 204 insertions(+), 19 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 26182d6b88d..9f1ab633c27 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "vitest"; -import { setImmediate as waitForImmediate } from "node:timers/promises"; +import { setImmediate as waitForImmediate, setTimeout as delay } from "node:timers/promises"; import type { PaseoToolCatalog } from "../../tools/types.js"; import type { @@ -541,13 +541,28 @@ describe("OMP agent client and session", () => { const scheduler = createOmpProviderIdleScheduler(); await expect( - scheduler.waitForRetry({ attempt: 1, consecutiveFailures: 0, elapsedMs: 0 }), + scheduler.waitForRetry({ + attempt: 1, + consecutiveFailures: 0, + elapsedMs: 0, + isCompacting: false, + }), ).resolves.toEqual({ retry: true }); await expect( - scheduler.waitForRetry({ attempt: 200, consecutiveFailures: 0, elapsedMs: 60_000 }), + scheduler.waitForRetry({ + attempt: 200, + consecutiveFailures: 0, + elapsedMs: 60_000, + isCompacting: false, + }), ).resolves.toEqual({ retry: false, reason: "wait_budget" }); await expect( - scheduler.waitForRetry({ attempt: 3, consecutiveFailures: 3, elapsedMs: 100 }), + scheduler.waitForRetry({ + attempt: 3, + consecutiveFailures: 3, + elapsedMs: 100, + isCompacting: false, + }), ).resolves.toEqual({ retry: false, reason: "failure_budget" }); }); @@ -577,6 +592,129 @@ describe("OMP agent client and session", () => { expect(omp.runningToolCallIds()).toEqual([]); }); + test("does not fail a turn that started after the gate was abandoned", async () => { + // The deny lands after ownership has moved on: get_state and waitForRetry are + // both awaits, so the turn can change under a parked gate. + let releaseDeny!: () => void; + let gateParked!: () => void; + const parked = new Promise((resolve) => { + gateParked = resolve; + }); + const scheduler: OmpProviderIdleScheduler = { + waitForRetry: () => { + gateParked(); + return new Promise((resolve) => { + releaseDeny = () => resolve({ retry: false, reason: "wait_budget" }); + }); + }, + }; + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await parked; + + // The user gives up on the stuck turn and sends another one. + await omp.interrupt(); + await completion; + omp.reportProviderState({ isStreaming: false, isCompacting: false }); + await omp.requireStartTurn("second"); + + releaseDeny(); + for (let flush = 0; flush < 5; flush += 1) await waitForImmediate(); + + expect(omp.failedTurns()).toEqual([]); + const runtime = omp.runtime(); + runtime.beginTurn(); + runtime.streamAssistantText("second done"); + runtime.finishTurn(); + for (let flush = 0; flush < 5; flush += 1) await waitForImmediate(); + expect(omp.completedTurnCount()).toBe(1); + }); + + test("keeps polling past the idle budget while OMP reports compacting", async () => { + const scheduler = createOmpProviderIdleScheduler(); + + await expect( + scheduler.waitForRetry({ + attempt: 1, + consecutiveFailures: 0, + elapsedMs: 120_000, + isCompacting: true, + }), + ).resolves.toEqual({ retry: true }); + await expect( + scheduler.waitForRetry({ + attempt: 1, + consecutiveFailures: 0, + elapsedMs: 120_000, + isCompacting: false, + }), + ).resolves.toEqual({ retry: false, reason: "wait_budget" }); + }); + + test("reports the observed compaction state and elapsed time to the scheduler", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: false, + isCompacting: true, + }); + await scheduler.waitForWaits(1); + expect(scheduler.attempts()[0]?.isCompacting).toBe(true); + + await delay(5); + scheduler.retryAll(); + await scheduler.waitForWaits(2); + expect(scheduler.attempts()[1]?.elapsedMs).toBeGreaterThan(0); + }); + + test("keeps an earlier agent_end error when a later cycle reports none", async () => { + const scheduler = new ManualIdleScheduler(); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + + omp.runtime().finishTurn({ role: "assistant", content: [], errorMessage: "provider exploded" }); + omp.runtime().finishTurn(); + for (let flush = 0; flush < 5; flush += 1) await waitForImmediate(); + + omp.reportProviderState({ isStreaming: false, isCompacting: false }); + scheduler.retryAll(); + + await expect(completion).rejects.toThrow(/provider exploded/); + }); + + test("reports an unavailable state path when no OMP state was ever observed", async () => { + const scheduler = new ManualIdleScheduler((attempt) => + attempt.attempt < 2 ? { retry: true } : { retry: false, reason: "wait_budget" }, + ); + const omp = new OmpHarness({ providerIdleScheduler: scheduler }); + await omp.start(); + omp.failProviderStateChecks(new Error("state unavailable")); + + const { completion } = await omp.startPromptUntilProviderIdle("first", "first done", { + isStreaming: true, + isCompacting: false, + }); + await scheduler.waitForWaits(1); + scheduler.retryAll(); + + await expect(completion).rejects.toThrow(/state/i); + expect(omp.failedTurns()).toMatchObject([{ code: "omp_provider_state_unavailable" }]); + expect(omp.failedTurns()[0]?.diagnostic).not.toContain("after 0 consecutive failures"); + }); + // #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. diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 7d4c77109cc..fdd749e41ec 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -144,8 +144,10 @@ export interface OmpProviderIdleAttempt { attempt: number; /** Consecutive `get_state` rejections; any successful response resets it. */ consecutiveFailures: number; - /** Wall-clock time since the gate opened, covering waits and `get_state`. */ + /** Monotonic time since the gate opened, covering waits and `get_state`. */ elapsedMs: number; + /** Whether the last observed state reported compaction in progress. */ + isCompacting: boolean; } /** @@ -212,19 +214,23 @@ interface OmpAgentSessionOptions { live?: boolean; } +// Autonomous OMP cycles carry no turn ID; they still need a gate key of their own. +const OMP_AUTONOMOUS_GATE_KEY = "autonomous"; + // OMP processes a state request only once its RPC loop is promptable again, so // the first checks after agent_end usually miss. Poll fast at first, then back // off: a stalled provider otherwise costs 100 RPCs a second for as long as the // stall lasts. -// Autonomous OMP cycles carry no turn ID; they still need a gate key of their own. -const OMP_AUTONOMOUS_GATE_KEY = "autonomous"; - const OMP_PROVIDER_IDLE_MIN_RETRY_MS = 10; const OMP_PROVIDER_IDLE_MAX_RETRY_MS = 1_000; // Wall clock, not a retry count: each get_state carries the JSONL-RPC request // timeout, so counting attempts would let a slow-but-answering state path stretch // the wait to tens of minutes. const OMP_PROVIDER_IDLE_BUDGET_MS = 60_000; +// Compaction is a model call over the whole context, so it routinely outlasts the +// idle budget. Waiting on a reported compaction is not the stall this gate guards +// against, but it still needs an end. +const OMP_PROVIDER_COMPACTING_BUDGET_MS = 600_000; // A get_state rejection already costs a full RPC timeout, so a few in a row are // enough evidence that the state path is gone. const OMP_PROVIDER_IDLE_FAILURE_BUDGET = 3; @@ -236,11 +242,14 @@ function ompProviderIdleRetryDelayMs(attempt: number): number { export function createOmpProviderIdleScheduler(): OmpProviderIdleScheduler { return { - waitForRetry: async ({ attempt, consecutiveFailures, elapsedMs }) => { + waitForRetry: async ({ attempt, consecutiveFailures, elapsedMs, isCompacting }) => { if (consecutiveFailures >= OMP_PROVIDER_IDLE_FAILURE_BUDGET) { return { retry: false, reason: "failure_budget" }; } - if (elapsedMs >= OMP_PROVIDER_IDLE_BUDGET_MS) { + const budgetMs = isCompacting + ? OMP_PROVIDER_COMPACTING_BUDGET_MS + : OMP_PROVIDER_IDLE_BUDGET_MS; + if (elapsedMs >= budgetMs) { return { retry: false, reason: "wait_budget" }; } await delay(ompProviderIdleRetryDelayMs(attempt)); @@ -573,6 +582,17 @@ function latestOmpErrorMessage(messages: OmpAgentMessage[]): string | null { return formatOmpErrorMessage(latestAssistant); } +/** Newest agent_end wins, unless it would discard an error the gate already holds. */ +function preferErroredOmpPayload( + current: OmpAgentMessage[], + incoming: OmpAgentMessage[], +): OmpAgentMessage[] { + if (latestOmpErrorMessage(incoming)) { + return incoming; + } + return latestOmpErrorMessage(current) ? current : incoming; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -2255,6 +2275,19 @@ export class OmpAgentSession implements AgentSession { void this.refreshAfterTurn(finalUsage); } + /** Ownership can change across an await; completing a turn this gate no longer + * represents would clear the turn that replaced it. */ + private completeTurnIfGateOwned(turnId: string | undefined, messages: OmpAgentMessage[]): void { + if (!this.ownsProviderIdleGate(turnId)) { + return; + } + this.completeTurn(turnId, messages); + } + + private ownsProviderIdleGate(turnId: string | undefined): boolean { + return !this.closed && this.activeTurnStarted && this.currentTurnIdForEvent() === turnId; + } + private async completeTurnAfterProviderIdle( turnId: string | undefined, messages: OmpAgentMessage[], @@ -2264,20 +2297,23 @@ export class OmpAgentSession implements AgentSession { // cycles carry no turn ID, so they share one key: their loop polls too. const key = turnId ?? OMP_AUTONOMOUS_GATE_KEY; if (this.providerIdleGate?.key === key) { - // The newest agent_end owns the terminal payload; the running loop reads it - // at completion time so a late provider error is not dropped. - this.providerIdleGate.messages = messages; + // The newest agent_end owns the terminal payload so a later provider error + // is not dropped, except when it would drop one the gate already holds. + this.providerIdleGate.messages = preferErroredOmpPayload( + this.providerIdleGate.messages, + messages, + ); return; } const gate = { key, messages }; this.providerIdleGate = gate; try { - const startedAt = Date.now(); + const startedAt = performance.now(); let attempt = 0; let consecutiveFailures = 0; let lastState: OmpSessionState | null = null; let lastError: unknown = null; - while (!this.closed && this.activeTurnStarted && this.currentTurnIdForEvent() === turnId) { + while (this.ownsProviderIdleGate(turnId)) { attempt += 1; try { const state = await this.runtimeSession.getState(); @@ -2287,7 +2323,7 @@ export class OmpAgentSession implements AgentSession { // 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, gate.messages); + this.completeTurnIfGateOwned(turnId, gate.messages); return; } } catch (error) { @@ -2301,9 +2337,13 @@ export class OmpAgentSession implements AgentSession { const decision = await this.providerIdleScheduler.waitForRetry({ attempt, consecutiveFailures, - elapsedMs: Date.now() - startedAt, + elapsedMs: performance.now() - startedAt, + isCompacting: lastState?.isCompacting === true, }); if (!decision.retry) { + if (!this.ownsProviderIdleGate(turnId)) { + return; + } this.failStalledTurn(turnId, { reason: decision.reason, attempt, @@ -2344,7 +2384,10 @@ export class OmpAgentSession implements AgentSession { lastError: unknown; }, ): void { - const stateUnavailable = context.reason === "failure_budget"; + // A hanging state path burns the wait budget before the failure budget, so a + // gate that never saw a state is an unavailable state path whichever ran out. + const stateUnavailable = + context.reason === "failure_budget" || (!context.lastState && context.lastError !== null); // Both observations go into every diagnostic: which budget ran out says what // Paseo did, the last state and the last error say what OMP was doing. const details = [`state checks: ${context.attempt}`]; @@ -2354,8 +2397,12 @@ export class OmpAgentSession implements AgentSession { ); } if (context.lastError) { + const failureCount = + context.consecutiveFailures > 0 + ? ` after ${context.consecutiveFailures} consecutive failures` + : ""; details.push( - `last get_state error after ${context.consecutiveFailures} consecutive failures: ${toDiagnosticErrorMessage(context.lastError)}`, + `last get_state error${failureCount}: ${toDiagnosticErrorMessage(context.lastError)}`, ); } const diagnostic = details.join("; "); From 14585e303a425c70104e2cbcc8771e0466dd0488 Mon Sep 17 00:00:00 2001 From: Joe Shull <42043763+joeshull@users.noreply.github.com> Date: Fri, 21 Aug 2026 15:46:16 -0600 Subject: [PATCH 5/5] fix(omp): let the idle budget cover the subagent wait #3371 adds a third reason the gate can wait: OMP-internal task children still running. That wait had the same shape as the two this branch bounded, so the budget now covers it, with one difference. A get_subagents reply that lists running children is positive evidence OMP is working, and a fan-out has no bounded length, so the wait budget does not apply while that evidence is current. It resumes the moment get_subagents stops answering, which is the stuck-parent risk that closed #2245: an index left holding children it can no longer confirm no longer holds the turn open forever. The failure budget for get_state applies throughout, and the diagnostic records that subagents were still reported running. Refs getpaseo/paseo#3654, getpaseo/paseo#2232 --- .../server/agent/providers/omp/agent.test.ts | 46 +++++++++++++++++++ .../src/server/agent/providers/omp/agent.ts | 39 ++++++++++++++-- 2 files changed, 81 insertions(+), 4 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 9f1ab633c27..498d5d433d2 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -546,6 +546,7 @@ describe("OMP agent client and session", () => { consecutiveFailures: 0, elapsedMs: 0, isCompacting: false, + isWaitingOnSubagents: false, }), ).resolves.toEqual({ retry: true }); await expect( @@ -554,6 +555,7 @@ describe("OMP agent client and session", () => { consecutiveFailures: 0, elapsedMs: 60_000, isCompacting: false, + isWaitingOnSubagents: false, }), ).resolves.toEqual({ retry: false, reason: "wait_budget" }); await expect( @@ -562,6 +564,7 @@ describe("OMP agent client and session", () => { consecutiveFailures: 3, elapsedMs: 100, isCompacting: false, + isWaitingOnSubagents: false, }), ).resolves.toEqual({ retry: false, reason: "failure_budget" }); }); @@ -644,6 +647,7 @@ describe("OMP agent client and session", () => { consecutiveFailures: 0, elapsedMs: 120_000, isCompacting: true, + isWaitingOnSubagents: false, }), ).resolves.toEqual({ retry: true }); await expect( @@ -652,6 +656,7 @@ describe("OMP agent client and session", () => { consecutiveFailures: 0, elapsedMs: 120_000, isCompacting: false, + isWaitingOnSubagents: false, }), ).resolves.toEqual({ retry: false, reason: "wait_budget" }); }); @@ -1039,6 +1044,47 @@ describe("OMP agent client and session", () => { expect(omp.runningToolCallIds()).toEqual([]); }); + test("waits past the idle budget while OMP reports running subagents", async () => { + const scheduler = createOmpProviderIdleScheduler(); + + await expect( + scheduler.waitForRetry({ + attempt: 1, + consecutiveFailures: 0, + elapsedMs: 120_000, + isCompacting: false, + isWaitingOnSubagents: true, + }), + ).resolves.toEqual({ retry: true }); + }); + + test("stops trusting a subagent wait once get_subagents stops answering", 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-fan"); + runtime.streamAssistantText("dispatching"); + runtime.emit({ + type: "subagent_lifecycle", + payload: { id: "child-1", agent: "worker", status: "started", index: 0 }, + }); + runtime.state = { ...runtime.state, isStreaming: false, isCompacting: false }; + runtime.finishTurn(); + + await scheduler.waitForWaits(1); + expect(scheduler.attempts()[0]?.isWaitingOnSubagents).toBe(true); + + // A snapshot that cannot be fetched is not evidence that work continues. + runtime.getSubagentsError = new Error("subagents unavailable"); + scheduler.retryAll(); + await scheduler.waitForWaits(2); + expect(scheduler.attempts()[1]?.isWaitingOnSubagents).toBe(false); + }); + 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 fdd749e41ec..7858ea8b664 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -148,6 +148,8 @@ export interface OmpProviderIdleAttempt { elapsedMs: number; /** Whether the last observed state reported compaction in progress. */ isCompacting: boolean; + /** Whether a current get_subagents reply still reports running children. */ + isWaitingOnSubagents: boolean; } /** @@ -242,10 +244,23 @@ function ompProviderIdleRetryDelayMs(attempt: number): number { export function createOmpProviderIdleScheduler(): OmpProviderIdleScheduler { return { - waitForRetry: async ({ attempt, consecutiveFailures, elapsedMs, isCompacting }) => { + waitForRetry: async ({ + attempt, + consecutiveFailures, + elapsedMs, + isCompacting, + isWaitingOnSubagents, + }) => { if (consecutiveFailures >= OMP_PROVIDER_IDLE_FAILURE_BUDGET) { return { retry: false, reason: "failure_budget" }; } + // A current snapshot reporting running children is positive evidence that + // OMP is working, and a fan-out has no bounded length. Stale evidence is + // not: the budget resumes as soon as get_subagents stops answering. + if (isWaitingOnSubagents) { + await delay(ompProviderIdleRetryDelayMs(attempt)); + return { retry: true }; + } const budgetMs = isCompacting ? OMP_PROVIDER_COMPACTING_BUDGET_MS : OMP_PROVIDER_IDLE_BUDGET_MS; @@ -2313,6 +2328,7 @@ export class OmpAgentSession implements AgentSession { let consecutiveFailures = 0; let lastState: OmpSessionState | null = null; let lastError: unknown = null; + let waitingOnSubagents = false; while (this.ownsProviderIdleGate(turnId)) { attempt += 1; try { @@ -2322,7 +2338,10 @@ export class OmpAgentSession implements AgentSession { consecutiveFailures = 0; // 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())) { + const modelBusy = state.isStreaming || state.isCompacting; + const subagents = modelBusy ? null : await this.pollOmpSubagents(); + waitingOnSubagents = subagents ? subagents.running && subagents.fresh : false; + if (subagents && !subagents.running) { this.completeTurnIfGateOwned(turnId, gate.messages); return; } @@ -2339,6 +2358,7 @@ export class OmpAgentSession implements AgentSession { consecutiveFailures, elapsedMs: performance.now() - startedAt, isCompacting: lastState?.isCompacting === true, + isWaitingOnSubagents: waitingOnSubagents, }); if (!decision.retry) { if (!this.ownsProviderIdleGate(turnId)) { @@ -2350,6 +2370,7 @@ export class OmpAgentSession implements AgentSession { consecutiveFailures, lastState, lastError, + waitingOnSubagents, }); return; } @@ -2361,17 +2382,23 @@ export class OmpAgentSession implements AgentSession { } } - private async hasRunningOmpSubagents(): Promise { + /** + * `fresh` reports whether this answer came from a snapshot OMP actually + * returned. The idle budget trusts running children only while it does. + */ + private async pollOmpSubagents(): Promise<{ running: boolean; fresh: boolean }> { + let fresh = true; try { const snapshots = await this.runtimeSession.getSubagents(); for (const event of this.subagentIndex.reconcileSnapshots(this.runtimeSession, snapshots)) { this.emit(event); } } catch (error) { + fresh = false; this.logger.debug({ err: error }, "OMP get_subagents unavailable during idle gate"); } this.settleDeferredTaskCalls(); - return this.subagentIndex.hasRunning(this.runtimeSession); + return { running: this.subagentIndex.hasRunning(this.runtimeSession), fresh }; } private failStalledTurn( @@ -2382,6 +2409,7 @@ export class OmpAgentSession implements AgentSession { consecutiveFailures: number; lastState: OmpSessionState | null; lastError: unknown; + waitingOnSubagents: boolean; }, ): void { // A hanging state path burns the wait budget before the failure budget, so a @@ -2405,6 +2433,9 @@ export class OmpAgentSession implements AgentSession { `last get_state error${failureCount}: ${toDiagnosticErrorMessage(context.lastError)}`, ); } + if (context.waitingOnSubagents) { + details.push("OMP still reported running subagents"); + } const diagnostic = details.join("; "); this.logger.warn( { turnId, reason: context.reason, diagnostic },