fix(server): Stop clears a finished delegated child's row held by a stale native subagent

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -1123,6 +1123,21 @@ function lastDeliveredRunForProviderThread(
   );
 }
 
+type DyingNativeSubagent = {
+  readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "subagents" | "turnItems">;
+  readonly subagent: Pick<
+    Extract<OrchestrationV2TurnItem, { readonly type: "subagent" }>,
+    "subagentId" | "origin" | "childThreadId" | "providerInstanceId"
+  >;
+};
+
+const nativeSubagentRef = (subagent: OrchestrationV2Subagent): DyingNativeSubagent["subagent"] => ({
+  subagentId: subagent.id,
+  origin: subagent.origin,
+  childThreadId: subagent.childThreadId,
+  providerInstanceId: subagent.providerInstanceId,
+});
+
 const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(function* () {
   const checkpointService = yield* CheckpointServiceV2;
   const commandPolicy = yield* CommandPolicyV2;
@@ -10830,6 +10845,78 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     }
   });
 
+  // Ends each native subagent's record, its own thread's runless root turns, and the
+  // subagents it started there, which a provider can record on that thread rather than its parent's.
+  const endNativeSubagentTrees = Effect.fn("Orchestrator.endNativeSubagentTrees")(
+    function* (input: {
+      readonly command: Extract<
+        OrchestrationV2ServerCommand,
+        { readonly type: "run.interrupt" | "thread.background-work.settle" | "thread.stop" }
+      >;
+      readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>;
+      readonly dying: ReadonlyArray<DyingNativeSubagent>;
+      readonly childThreadIds: Set<ThreadId>;
+      readonly now: DateTime.Utc;
+    }) {
+      const allocateEventId = () =>
+        idAllocator.allocate.event({
+          threadId: input.command.threadId,
+          commandId: input.command.commandId,
+        });
+      const dying = [...input.dying];
+      const { childThreadIds } = input;
+      for (const { projection, subagent } of dying) {
+        const ended = yield* endOrphanedNativeSubagent({
+          projection,
+          subagentId: subagent.subagentId,
+          status: "interrupted",
+          now: input.now,
+          emitted: yield* Ref.get(input.events),
+          allocateEventId,
+        }).pipe(mapDispatchError(input.command));
+        yield* Ref.update(input.events, (events) => [...events, ...ended]);
+        const childThreadId = subagent.childThreadId;
+        if (
+          subagent.origin !== "provider_native" ||
+          childThreadId === null ||
+          childThreadIds.has(childThreadId)
+        ) {
+          continue;
+        }
+        childThreadIds.add(childThreadId);
+        const child = yield* projectionStore
+          .getThreadRecords(childThreadId, ["nodes", "providerTurns", "subagents", "turnItems"], {
+            // Runless items are the subagent's; any run there is the user's own.
+            turnItemRunIds: [null],
+            turnItemStatuses: ["pending", "running", "waiting"],
+          })
+          .pipe(
+            Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }),
+            mapDispatchError(input.command),
+          );
+        if (child === null) continue;
+        const childEnded = yield* endRunlessRootTurns({
+          threadId: childThreadId,
+          providerInstanceId: subagent.providerInstanceId,
+          projection: child,
+          status: "interrupted",
+          now: input.now,
+          emitted: yield* Ref.get(input.events),
+          allocateEventId,
+        }).pipe(mapDispatchError(input.command));
+        yield* Ref.update(input.events, (events) => [...events, ...childEnded]);
+        for (const nested of child.turnItems) {
+          if (nested.type === "subagent") dying.push({ projection: child, subagent: nested });
+        }
+        for (const nested of child.subagents) {
+          if (nested.origin === "provider_native" && nested.runId === null) {
+            dying.push({ projection: child, subagent: nativeSubagentRef(nested) });
+          }
+        }
+      }
+    },
+  );
+
   const settleBackgroundWork = Effect.fn("Orchestrator.settleBackgroundWork")(function* (input: {
     readonly command: Extract<
       OrchestrationV2ServerCommand,
@@ -10846,11 +10933,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   }) {
     const emitEvent = emit(input.events, input.command);
     const runOrdinals = new Map(input.projection.runs.map((run) => [run.id, run.ordinal]));
-    const allocateEventId = () =>
-      idAllocator.allocate.event({
-        threadId: input.command.threadId,
-        commandId: input.command.commandId,
-      });
     const childThreadIds = new Set<ThreadId>();
     const liveness = new Map<string, boolean>();
     const hasLiveSession = Effect.fn("Orchestrator.hasLiveSession")(function* (
@@ -10915,76 +10997,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         },
       });
       if (item.type !== "subagent") continue;
-      // A native subagent died with the process too: end its record, its own
-      // thread's runless root turn, and the subagents it started there, which
-      // a provider can record on that thread rather than this one.
-      const dying: Array<{
-        readonly projection: Pick<
-          OrchestrationV2ThreadProjection,
-          "nodes" | "subagents" | "turnItems"
-        >;
-        readonly subagent: Pick<
-          Extract<OrchestrationV2TurnItem, { readonly type: "subagent" }>,
-          "subagentId" | "origin" | "childThreadId" | "providerInstanceId"
-        >;
-      }> = [{ projection: input.projection, subagent: item }];
-      for (const { projection, subagent } of dying) {
-        const ended = yield* endOrphanedNativeSubagent({
-          projection,
-          subagentId: subagent.subagentId,
-          status: "interrupted",
-          now: input.now,
-          emitted: yield* Ref.get(input.events),
-          allocateEventId,
-        }).pipe(mapDispatchError(input.command));
-        yield* Ref.update(input.events, (events) => [...events, ...ended]);
-        const childThreadId = subagent.childThreadId;
-        if (
-          subagent.origin !== "provider_native" ||
-          childThreadId === null ||
-          childThreadIds.has(childThreadId)
-        ) {
-          continue;
-        }
-        childThreadIds.add(childThreadId);
-        const child = yield* projectionStore
-          .getThreadRecords(childThreadId, ["nodes", "providerTurns", "subagents", "turnItems"], {
-            // Runless items are the subagent's; any run there is the user's own.
-            turnItemRunIds: [null],
-            turnItemStatuses: ["pending", "running", "waiting"],
-          })
-          .pipe(
-            Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }),
-            mapDispatchError(input.command),
-          );
-        if (child === null) continue;
-        const childEnded = yield* endRunlessRootTurns({
-          threadId: childThreadId,
-          providerInstanceId: subagent.providerInstanceId,
-          projection: child,
-          status: "interrupted",
-          now: input.now,
-          emitted: yield* Ref.get(input.events),
-          allocateEventId,
-        }).pipe(mapDispatchError(input.command));
-        yield* Ref.update(input.events, (events) => [...events, ...childEnded]);
-        for (const nested of child.turnItems) {
-          if (nested.type === "subagent") dying.push({ projection: child, subagent: nested });
-        }
-        for (const nested of child.subagents) {
-          if (nested.origin === "provider_native" && nested.runId === null) {
-            dying.push({
-              projection: child,
-              subagent: {
-                subagentId: nested.id,
-                origin: nested.origin,
-                childThreadId: nested.childThreadId,
-                providerInstanceId: nested.providerInstanceId,
-              },
-            });
-          }
-        }
-      }
+      // A native subagent died with the process too.
+      yield* endNativeSubagentTrees({
+        command: input.command,
+        events: input.events,
+        dying: [{ projection: input.projection, subagent: item }],
+        childThreadIds,
+        now: input.now,
+      });
     }
     for (const providerThread of input.projection.providerThreads) {
       if ((providerThread.pendingBackgroundTasks?.length ?? 0) === 0) continue;
@@ -12458,6 +12478,36 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     for (const run of reached.values()) {
       yield* markStoppedRun({ command, events, thread: projection.thread, run, now });
     }
+    // Record and item are written together, so an active record without a live item lost its end.
+    const staleNative = (target === undefined ? projection.subagents : []).filter(
+      (subagent) =>
+        subagent.origin === "provider_native" &&
+        subagent.runId !== null &&
+        isOrchestrationV2WorkActive(subagent.status) &&
+        !projection.turnItems.some(
+          (item) => item.type === "subagent" && item.subagentId === subagent.id,
+        ),
+    );
+    if (staleNative.length > 0) {
+      const { nodes } = yield* projectionStore
+        .getThreadRecords(command.threadId, ["nodes"])
+        .pipe(mapDispatchError(command));
+      const threadProjection = {
+        nodes,
+        subagents: projection.subagents,
+        turnItems: projection.turnItems,
+      };
+      yield* endNativeSubagentTrees({
+        command,
+        events,
+        dying: staleNative.map((subagent) => ({
+          projection: threadProjection,
+          subagent: nativeSubagentRef(subagent),
+        })),
+        childThreadIds: new Set(),
+        now,
+      });
+    }
     yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now });
     yield* cancelIdleOrphans({ command, events, projection });
     yield* stopDelegatedTasks;

apps/server/src/orchestration-v2/ThreadStop.test.ts

@@ -1742,3 +1742,151 @@ it.effect.each(["Stop", "Force stop"] as const)(
       );
     }).pipe(Effect.provide(layerTest)),
 );
+
+// A finished delegated child whose native subagent the provider never reported ending,
+// with no open turn item, and a nested native subagent on that agent's own thread.
+it.effect(
+  "Stop settles a delegated row whose finished child still records a running native subagent",
+  () =>
+    Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const projections = yield* ProjectionStore.ProjectionStoreV2;
+      const parentThreadId = ThreadId.make("thread:held-row");
+      yield* createWatchingThread(parentThreadId, 30);
+      yield* send(parentThreadId, "parent", "start_immediately");
+      const childThreadId = yield* delegate(parentThreadId, "held child");
+      const child = yield* orchestrator.getThreadProjection(childThreadId);
+      const childRun = child.runs[0]!;
+      const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!;
+      const now = yield* DateTime.now;
+      const nativeThreadId = ThreadId.make("thread:held-row-native");
+      const nestedThreadId = ThreadId.make("thread:held-row-nested");
+      yield* createWatchingThread(nativeThreadId, 31);
+      yield* createWatchingThread(nestedThreadId, 32);
+      const nativeProviderThreadId = ProviderThreadId.make("provider-thread:held-row-native");
+      const runningRoot = (threadId: ThreadId, id: NodeId) => ({
+        ...child.nodes.find((node) => node.id === childRun.rootNodeId)!,
+        id,
+        rootNodeId: id,
+        threadId,
+        runId: null,
+        parentNodeId: null,
+        status: "running" as const,
+        countsForRun: false,
+        completedAt: null,
+      });
+      const nativeRoot = runningRoot(nativeThreadId, NodeId.make("node:held-row-native-root"));
+      const nestedRoot = runningRoot(nestedThreadId, NodeId.make("node:held-row-nested-root"));
+      for (const root of [nativeRoot, nestedRoot]) {
+        yield* projections.apply({
+          id: EventId.make(`event:held-row:${root.id}`),
+          type: "node.updated",
+          threadId: root.threadId,
+          nodeId: root.id,
+          occurredAt: now,
+          payload: root,
+        });
+      }
+      yield* projections.apply({
+        id: EventId.make("event:held-row:native-turn"),
+        type: "provider-turn.updated",
+        threadId: nativeThreadId,
+        nodeId: nativeRoot.id,
+        occurredAt: now,
+        payload: {
+          id: ProviderTurnId.make("turn:held-row-native"),
+          providerThreadId: nativeProviderThreadId,
+          nodeId: nativeRoot.id,
+          runAttemptId: null,
+          nativeTurnRef: null,
+          ordinal: 1,
+          status: "running",
+          startedAt: now,
+          completedAt: null,
+        },
+      });
+      const native = {
+        ...task,
+        id: NodeId.make("node:held-row-native"),
+        threadId: childThreadId,
+        runId: childRun.id,
+        parentNodeId: childRun.rootNodeId!,
+        origin: "provider_native" as const,
+        providerThreadId: nativeProviderThreadId,
+        childThreadId: nativeThreadId,
+        status: "running" as const,
+        completionDelivery: undefined,
+      };
+      yield* projections.apply({
+        id: EventId.make("event:held-row:native"),
+        type: "subagent.updated",
+        threadId: childThreadId,
+        runId: childRun.id,
+        occurredAt: now,
+        payload: native,
+      });
+      yield* projections.apply({
+        id: EventId.make("event:held-row:nested"),
+        type: "subagent.updated",
+        threadId: nativeThreadId,
+        occurredAt: now,
+        payload: {
+          ...native,
+          id: NodeId.make("node:held-row-nested"),
+          threadId: nativeThreadId,
+          runId: null,
+          parentNodeId: nativeRoot.id,
+          childThreadId: nestedThreadId,
+        },
+      });
+      yield* projections.apply({
+        id: EventId.make("event:held-row:child-completed"),
+        type: "run.updated",
+        threadId: childThreadId,
+        runId: childRun.id,
+        occurredAt: now,
+        payload: { ...childRun, status: "completed", startedAt: now, completedAt: now },
+      });
+      yield* orchestrator.recoverDelegatedTasks;
+      assert.equal(
+        (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status,
+        "running",
+      );
+      const statuses = Effect.gen(function* () {
+        const nativeAfter = yield* orchestrator.getThreadProjection(nativeThreadId);
+        const nestedAfter = yield* orchestrator.getThreadProjection(nestedThreadId);
+        return {
+          row: (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]?.status,
+          native: (yield* orchestrator.getThreadProjection(childThreadId)).subagents.find(
+            (subagent) => subagent.id === native.id,
+          )?.status,
+          nativeRoot: nativeAfter.nodes.find((node) => node.id === nativeRoot.id)?.status,
+          nested: nativeAfter.subagents[0]?.status,
+          nestedRoot: nestedAfter.nodes.find((node) => node.id === nestedRoot.id)?.status,
+        };
+      });
+
+      // The row's Stop: the parent's task, then the child thread.
+      yield* orchestrator.dispatch({
+        type: "subagent.stop",
+        commandId: CommandId.make("held-row:stop-task"),
+        threadId: parentThreadId,
+        runId: task.runId!,
+        subagentId: task.id,
+      });
+      yield* orchestrator.dispatch({
+        type: "thread.stop",
+        commandId: CommandId.make("held-row:stop-child"),
+        threadId: childThreadId,
+      });
+      const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+      yield* worker.drain();
+      assert.deepEqual(yield* statuses, {
+        row: "completed",
+        native: "interrupted",
+        nativeRoot: "interrupted",
+        nested: "interrupted",
+        nestedRoot: "interrupted",
+      });
+    }).pipe(Effect.provide(layerTest)),
+);