fix(omp): bound the post-agent_end completion gate (stacked on #3371)

packages/server/src/server/agent/providers/omp/cli-runtime.test.ts

@@ -243,14 +243,19 @@ describe("OMP CLI runtime", () => {
     const commands: Record<string, unknown>[] = [];
     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" },
     ]);
   });
 

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<OmpSubagentSnapshot[]> {
+    const data = OmpSubagentsResultSchema.parse(await this.request({ type: "get_subagents" }));
+    return data.subagents ?? [];
+  }
+
   async setHostTools(tools: OmpRpcHostToolDefinition[]): Promise<string[]> {
     const data = OmpHostToolsResultSchema.parse(
       await this.request({ type: "set_host_tools", tools }),

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<typeof OmpAvailableCommand
 export type OmpRpcCommand = z.infer<typeof OmpRpcCommandSchema>;
 export type OmpPromptAck = z.infer<typeof OmpPromptAckSchema> & { 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<typeof OmpSubagentSnapshotSchema>;
 
 export interface OmpSubagentMessagesResult {
   sessionFile: string;

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<void>;
   abort(): Promise<void>;
   getState(): Promise<OmpSessionState>;
+  getSubagents(): Promise<OmpSubagentSnapshot[]>;
   getMessages(): Promise<OmpAgentMessage[]>;
   getAvailableModels(timeoutMs?: number | null): Promise<OmpModel[]>;
   setModel(provider: string, modelId: string): Promise<OmpModel>;

packages/server/src/server/agent/providers/omp/subagent-index.test.ts

@@ -114,4 +114,78 @@ 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);
+  });
+
+  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);
+  });
 });

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,80 @@ 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;
+  }
+
+  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.
+   * 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<string>();
+    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 +170,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 +216,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";
+}

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<FakeOmpSubagentSnapshot[]> {
+    if (this.getSubagentsError) {
+      throw this.getSubagentsError;
+    }
     return this.subagents;
   }
 

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<void> {
     return this.omp.latestSession().waitForStateRequests(count);
   }
@@ -406,6 +417,10 @@ export class OmpHarness {
     return items;
   }
 
+  failedTurns(): Array<Extract<AgentStreamEvent, { type: "turn_failed" }>> {
+    return this.events.flatMap((event) => (event.type === "turn_failed" ? [event] : []));
+  }
+
   completedTurnCount(): number {
     return this.events.filter((event) => event.type === "turn_completed").length;
   }