From 1ecc693e704b39f1bba35cbe2509758e820901f5 Mon Sep 17 00:00:00 2001 From: Jason Date: Mon, 20 Jul 2026 14:11:36 +0900 Subject: [PATCH] fix(omp): keep parent non-idle while internal task subagents run OMP can end the parent model loop (agent_end + isStreaming=false) while task children still write under the session stem directory. Gate turn completion on running subagents, reconcile via get_subagents, and recover orphan child history when the parent .jsonl is missing. Fixes #2232 --- .../server/agent/providers/omp/agent.test.ts | 105 ++++++++++++++++++ .../src/server/agent/providers/omp/agent.ts | 24 +++- .../server/agent/providers/omp/cli-runtime.ts | 7 ++ .../providers/omp/history-mapper.test.ts | 53 +++++++++ .../src/server/agent/providers/omp/history.ts | 40 ++++++- .../server/agent/providers/omp/rpc-types.ts | 33 +++--- .../src/server/agent/providers/omp/runtime.ts | 3 + .../providers/omp/subagent-index.test.ts | 48 ++++++++ .../agent/providers/omp/subagent-index.ts | 50 +++++++++ .../providers/omp/test-utils/fake-omp.ts | 19 +--- 10 files changed, 352 insertions(+), 30 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 4e01109bea1..08a392f5f21 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -245,6 +245,111 @@ describe("OMP agent client and session", () => { await expect(completion).resolves.toMatchObject({ finalText: "first done" }); }); + // Repro/regression for https://github.com/getpaseo/paseo/issues/2232: + // OMP can end the parent model turn (agent_end + isStreaming=false) while + // internal `task` subagents are still writing. Paseo must not complete the + // turn / flip idle until those children finish. + 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(); + + await omp.requireStartTurn("critically audit the entire repo"); + const runtime = omp.runtime(); + runtime.beginTurn(); + runtime.acceptPrompt("critically audit the entire repo", "user-audit"); + runtime.streamAssistantText("spawning fan-out workers"); + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "ApiBudgetAudit", + agent: "ApiBudgetAudit", + description: "audit API budget", + status: "started", + parentToolCallId: "task-1", + index: 0, + }, + }); + // Parent model loop is idle; child work continues (issue #2232 shape). + runtime.state = { ...runtime.state, isStreaming: false, isCompacting: false }; + runtime.finishTurn({ + role: "assistant", + content: [{ type: "text", text: "spawning fan-out workers" }], + }); + + await omp.waitForProviderStateChecks(1); + await scheduler.waitForWaits(1); + expect(omp.completedTurnCount()).toBe(0); + expect(omp.subagentUpserts()).toContainEqual({ id: "ApiBudgetAudit", status: "running" }); + + // Still busy after another idle poll while the child is running. + scheduler.retry(); + await omp.waitForProviderStateChecks(2); + await scheduler.waitForWaits(2); + expect(omp.completedTurnCount()).toBe(0); + + runtime.emit({ + type: "subagent_lifecycle", + payload: { + id: "ApiBudgetAudit", + agent: "ApiBudgetAudit", + status: "completed", + parentToolCallId: "task-1", + index: 0, + }, + }); + scheduler.retry(); + await omp.waitForProviderStateChecks(3); + // getState waiters resolve before completeTurn runs on the same tick. + await new Promise((resolve) => setImmediate(resolve)); + expect(omp.completedTurnCount()).toBe(1); + expect(omp.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 = [ + { + id: "PipelineFeedAudit", + index: 0, + agent: "PipelineFeedAudit", + status: "completed", + parentToolCallId: "task-2", + }, + ]; + scheduler.retry(); + await omp.waitForProviderStateChecks(2); + await new Promise((resolve) => setImmediate(resolve)); + expect(omp.completedTurnCount()).toBe(1); + }); + test("stays active when OMP state checks fail", async () => { const scheduler = new ManualIdleScheduler(); const omp = new OmpHarness({ providerIdleScheduler: scheduler }); diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 01ad77a36cf..5f1697c38dc 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -2166,7 +2166,11 @@ export class OmpAgentSession implements AgentSession { try { const state = await this.runtimeSession.getState(); this.state = state; - if (!state.isStreaming && !state.isCompacting) { + // Parent model loop idle is not enough: OMP-internal `task` children can + // keep running after agent_end / isStreaming=false (hub waits, fan-out). + // Completing the Paseo turn here marks the agent idle while work continues + // (https://github.com/getpaseo/paseo/issues/2232). + if (!state.isStreaming && !state.isCompacting && !(await this.hasRunningOmpSubagents())) { this.completeTurn(turnId, messages); return; } @@ -2177,6 +2181,24 @@ export class OmpAgentSession implements AgentSession { } } + /** + * Keep the event index in sync with OMP's `get_subagents` while waiting for + * idle. Events alone can miss frames; the index alone can go stale without a + * terminal lifecycle event. Reconcile every poll so either path can settle. + */ + 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) { + // Older OMP binaries may not expose get_subagents; event index is best-effort. + 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.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.ts index b080a541b67..3984c29f0c7 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.ts @@ -23,6 +23,7 @@ import { OmpRuntimeEventSchema, OmpSessionStateSchema, OmpSessionStatsSchema, + OmpSubagentsResultSchema, type OmpThinkingLevel, type OmpAgentMessage, type OmpModel, @@ -35,6 +36,7 @@ import { type OmpRuntimeEvent, type OmpSessionState, type OmpSessionStats, + type OmpSubagentSnapshot, type OmpSubagentSubscriptionLevel, } from "./rpc-types.js"; @@ -202,6 +204,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/history-mapper.test.ts b/packages/server/src/server/agent/providers/omp/history-mapper.test.ts index db67f0cbe3a..660d3fd3623 100644 --- a/packages/server/src/server/agent/providers/omp/history-mapper.test.ts +++ b/packages/server/src/server/agent/providers/omp/history-mapper.test.ts @@ -644,4 +644,57 @@ describe("OMP history mapper", () => { ]), ); }); + + test("discovers child transcripts when parent session file is missing", async () => { + const dir = mkdtempSync(join(tmpdir(), "omp-orphan-parent-")); + const parentFile = join(dir, "parent.jsonl"); + const parentStem = parentFile.slice(0, -".jsonl".length); + const childFile = join(parentStem, "ApiBudgetAudit.jsonl"); + mkdirSync(parentStem, { recursive: true }); + writeFileSync( + childFile, + [ + { type: "session", id: "child-root", parentId: null, timestamp: "2026-07-19T20:00:00Z" }, + { + type: "message", + id: "child-answer", + parentId: "child-root", + timestamp: "2026-07-19T20:00:01Z", + message: { role: "assistant", content: [{ type: "text", text: "audit ok" }] }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n"), + ); + + const events: AgentStreamEvent[] = []; + for await (const event of streamOmpHistory({ sessionFile: parentFile, provider: "omp" })) { + events.push(event); + } + const subagentEvents = events.flatMap((event) => + event.type === "provider_subagent" ? [event.event] : [], + ); + expect(subagentEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "upsert", + id: "ApiBudgetAudit", + status: "running", + }), + expect.objectContaining({ + type: "timeline", + id: "ApiBudgetAudit", + item: expect.objectContaining({ + type: "assistant_message", + text: "audit ok", + }), + }), + expect.objectContaining({ + type: "upsert", + id: "ApiBudgetAudit", + status: "completed", + }), + ]), + ); + }); }); diff --git a/packages/server/src/server/agent/providers/omp/history.ts b/packages/server/src/server/agent/providers/omp/history.ts index 7147e35e596..08f3573403c 100644 --- a/packages/server/src/server/agent/providers/omp/history.ts +++ b/packages/server/src/server/agent/providers/omp/history.ts @@ -1,4 +1,4 @@ -import { readFile } from "node:fs/promises"; +import { readdir, readFile, stat } from "node:fs/promises"; import { basename, extname, join } from "node:path"; import type { AgentProvider, AgentStreamEvent } from "../../agent-sdk-types.js"; import { normalizeProviderReplayTimestamp } from "../../provider-history-timestamps.js"; @@ -63,6 +63,12 @@ export async function* streamOmpHistory(input: { ); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") { + // Parent `.jsonl` can be missing while OMP still keeps child transcripts under + // the stem artifacts directory (issue #2232). Surface those children instead + // of returning an empty history. + for (const transcript of await discoverOrphanSubagentTranscripts(input.sessionFile)) { + yield* replaySubagentTranscript(transcript, input.provider, visitedSessionFiles); + } return; } throw error; @@ -290,6 +296,38 @@ function stripExtension(filePath: string): string { return extension ? filePath.slice(0, -extension.length) : filePath; } +/** + * When the parent session file is gone but `/` still holds child + * `*.jsonl` files, treat those as completed provider subagent transcripts. + */ +async function discoverOrphanSubagentTranscripts( + parentSessionFile: string, +): Promise { + const artifactsDir = stripExtension(parentSessionFile); + let names: string[]; + try { + const dirStat = await stat(artifactsDir); + if (!dirStat.isDirectory()) { + return []; + } + names = await readdir(artifactsDir); + } catch { + return []; + } + return names + .filter((name) => name.endsWith(".jsonl") && !name.startsWith(".")) + .map((name) => { + const id = basename(name, ".jsonl"); + return { + id, + title: id, + toolCallId: "", + sessionFile: join(artifactsDir, name), + status: "completed" as const, + }; + }); +} + export async function readActiveOmpEntryChain( sessionFile: string, activeEntryId?: string, 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 e6e53031747..e7f4617190a 100644 --- a/packages/server/src/server/agent/providers/omp/rpc-types.ts +++ b/packages/server/src/server/agent/providers/omp/rpc-types.ts @@ -531,6 +531,7 @@ export const OmpRpcCommandSchema = z.discriminatedUnion("type", [ type: z.literal("set_subagent_subscription"), level: OmpSubagentSubscriptionLevelSchema, }), + z.object({ ...OmpCommandBase, type: z.literal("get_subagents") }), z.object({ ...OmpCommandBase, type: z.literal("set_host_tools"), @@ -561,6 +562,24 @@ 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().optional(), + 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(), + detached: z.boolean().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(); @@ -589,6 +608,7 @@ export type OmpRpcHostToolUpdate = z.infer; export type OmpRpcHostToolResult = z.infer; export type OmpSubagentSubscriptionLevel = z.infer; export type OmpSubagentStatus = z.infer; +export type OmpSubagentSnapshot = z.infer; export type OmpSubagentLifecyclePayload = z.infer; export type OmpSubagentProgressPayload = z.infer; export type OmpSubagentEventPayload = z.infer; @@ -612,19 +632,6 @@ 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 interface OmpSubagentMessagesResult { sessionFile: string; fromByte: number; diff --git a/packages/server/src/server/agent/providers/omp/runtime.ts b/packages/server/src/server/agent/providers/omp/runtime.ts index 3a8d90c61c2..bf88e48c6e9 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"; @@ -58,6 +59,8 @@ export interface OmpRuntimeSession { getSessionStats(): Promise; getCommands(): Promise; setSubagentSubscription(level: OmpSubagentSubscriptionLevel): Promise; + /** OMP source-of-truth list of internal task children (may throw on old binaries). */ + getSubagents(): Promise; setHostTools(tools: OmpRpcHostToolDefinition[]): Promise; sendHostToolResult(result: OmpRpcHostToolResult): void; sendHostToolUpdate(update: OmpRpcHostToolUpdate): void; 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..d795e714525 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("hasRunning tracks in-flight OMP task children", () => { + const index = new OmpSubagentIndex(); + const parent = {}; + expect(index.hasRunning(parent)).toBe(false); + + index.handleLifecycle(parent, { + id: "child-1", + agent: "explore", + status: "started", + index: 0, + }); + expect(index.hasRunning(parent)).toBe(true); + + index.handleLifecycle(parent, { + id: "child-1", + agent: "explore", + status: "completed", + index: 0, + }); + expect(index.hasRunning(parent)).toBe(false); + }); + + test("reconcileSnapshots brings missed get_subagents state into the index", () => { + const index = new OmpSubagentIndex(); + const parent = {}; + expect( + index.reconcileSnapshots(parent, [ + { + id: "child-2", + agent: "audit", + status: "running", + parentToolCallId: "task-9", + }, + ]), + ).toMatchObject([{ event: { id: "child-2", status: "running" } }]); + expect(index.hasRunning(parent)).toBe(true); + + index.reconcileSnapshots(parent, [ + { + id: "child-2", + agent: "audit", + status: "completed", + parentToolCallId: "task-9", + }, + ]); + expect(index.hasRunning(parent)).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 205cbe45cec..575c293c2f5 100644 --- a/packages/server/src/server/agent/providers/omp/subagent-index.ts +++ b/packages/server/src/server/agent/providers/omp/subagent-index.ts @@ -80,6 +80,49 @@ export class OmpSubagentIndex { return events; } + /** + * True while any OMP-internal task child is still in-flight for this parent. + * Used to keep the Paseo turn non-idle after parent `agent_end` while tasks run. + */ + 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 `get_subagents` snapshot into local state so missed event frames + * do not leave idle detection blind. + */ + reconcileSnapshots( + parent: object, + snapshots: ReadonlyArray<{ + id: string; + agent: string; + description?: string; + status: "pending" | "running" | "completed" | "failed" | "aborted"; + parentToolCallId?: string; + }>, + ): AgentStreamEvent[] { + const events: AgentStreamEvent[] = []; + for (const snapshot of snapshots) { + const state = this.stateFor(parent, snapshot.id, snapshot.agent); + state.title = snapshot.agent || state.title; + state.description = snapshot.description ?? state.description; + state.toolCallId = snapshot.parentToolCallId ?? state.toolCallId; + state.status = mapSnapshotStatus(snapshot.status); + events.push(this.upsert(snapshot.id, state.status, state)); + } + return events; + } + clear(parent: object): void { this.statesByParent.delete(parent); } @@ -139,3 +182,10 @@ function mapProgressStatus( if (status === "completed" || status === "failed") return status; return status === "aborted" ? "canceled" : "running"; } + +function mapSnapshotStatus( + status: "pending" | "running" | "completed" | "failed" | "aborted", +): "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 31cd6fd0fb8..5bbb343da9e 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 @@ -15,25 +15,14 @@ import type { OmpRuntimeEvent, OmpSessionState, OmpSessionStats, + OmpSubagentSnapshot, OmpThinkingLevel, } from "../rpc-types.js"; import { buildOmpLaunch } from "../runtime.js"; type FakeOmpSubagentSubscriptionLevel = "off" | "progress" | "events"; -type FakeOmpSubagentStatus = "pending" | "running" | "completed" | "failed" | "aborted"; - -export interface FakeOmpSubagentSnapshot { - id: string; - index: number; - agent: string; - description?: string; - status: FakeOmpSubagentStatus; - task?: string; - assignment?: string; - sessionFile?: string; - parentToolCallId?: string; - lastUpdate?: number; -} + +export type FakeOmpSubagentSnapshot = OmpSubagentSnapshot; export interface FakeOmpSubagentMessagesSelector { subagentId?: string; @@ -352,7 +341,7 @@ export class FakeOmpSession implements OmpRuntimeSession { return new Promise((resolve) => this.hostToolResultWaiters.push(resolve)); } - async getSubagents(): Promise { + async getSubagents(): Promise { return this.subagents; }