# Files changed · PR \#21724 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/21724/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[fix\(server\): settle durable native subagents from terminal receipts](https://git.cafe/pingdotgg/t3code/pull/21724.md)

Version: 8

Head commit: ac83029f2f56a593085120c2ccacc3cae79ba57e

Base commit: 2532e64099a2a14e18427783a79e7ede44f626ce

Comparison base: 2532e64099a2a14e18427783a79e7ede44f626ce

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

## apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts

Status: modified; +226; −0

```diff
@@ -25,6 +25,7 @@
   RunId,
   RuntimeRequestId,
   TurnItemId,
+  ThreadId,
 } from "@t3tools/contracts";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
@@ -2241,4 +2242,229 @@
       );
     }),
   );
+  it.effect("reconciles a native task after its provider instance was replaced", () =>
+    Effect.gen(function* () {
+      const now = yield* DateTime.now;
+      const sink = yield* EventSink.EventSinkV2;
+      const store = yield* ProjectionStore.ProjectionStoreV2;
+      const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
+      const root = yield* threadCreatedEvent(now);
+      yield* sink.write({ events: [root] });
+      const sessionId = (yield* IdAllocator.IdAllocatorV2).allocate.providerSession;
+      const allocatedSessionId = yield* sessionId({
+        providerInstanceId: modelSelection.instanceId,
+        threadId: root.threadId!,
+      });
+      const ingest = (
+        event: ProviderEventIngestor.ProviderEventIngestInput["event"],
+        instance = modelSelection.instanceId,
+      ) =>
+        ingestor.ingestNormalized({
+          providerSessionId: allocatedSessionId,
+          providerInstanceId: instance,
+          threadId: root.threadId!,
+          event,
+        });
+      const childThreadId = ThreadId.make("thread:old-native-child");
+      const appThread = yield* store.getThread(root.threadId!);
+      yield* ingest({
+        type: "app_thread.created",
+        driver: CODEX_DRIVER,
+        appThread: {
+          ...appThread,
+          id: childThreadId,
+          activeProviderThreadId: null,
+          lineage: {
+            parentThreadId: appThread.id,
+            relationshipToParent: "subagent",
+            rootThreadId: appThread.id,
+          },
+        },
+      });
+      const task = {
+        id: NodeId.make("node:old-native-task"),
+        threadId: root.threadId!,
+        runId: null,
+        parentNodeId: NodeId.make("node:root"),
+        origin: "provider_native" as const,
+        createdBy: "agent" as const,
+        driver: CODEX_DRIVER,
+        providerInstanceId: modelSelection.instanceId,
+        providerThreadId: null,
+        childThreadId,
+        nativeTaskRef: { driver: CODEX_DRIVER, nativeId: "old-task", strength: "strong" as const },
+        prompt: "Review",
+        title: "Review",
+        model: null,
+        status: "running" as const,
+        result: null,
+        startedAt: now,
+        completedAt: null,
+        updatedAt: now,
+      };
+      yield* ingest({ type: "subagent.updated", driver: CODEX_DRIVER, subagent: task });
+      const childRootId = NodeId.make("node:old-native-child-root");
+      const childRoot = {
+        id: childRootId,
+        threadId: childThreadId,
+        runId: null,
+        parentNodeId: null,
+        rootNodeId: childRootId,
+        kind: "root_turn",
+        status: "running",
+        countsForRun: false,
+        providerThreadId: null,
+        providerTurnId: null,
+        nativeItemRef: task.nativeTaskRef,
+        runtimeRequestId: null,
+        checkpointScopeId: null,
+        startedAt: now,
+        completedAt: null,
+      } satisfies OrchestrationV2ExecutionNode;
+      yield* ingest({ type: "node.updated", driver: CODEX_DRIVER, node: childRoot });
+      yield* ingest({
+        type: "node.updated",
+        driver: CODEX_DRIVER,
+        node: {
+          ...childRoot,
+          id: task.id,
+          threadId: task.threadId,
+          parentNodeId: task.parentNodeId,
+          rootNodeId: task.parentNodeId,
+          kind: "subagent",
+        },
+      });
+
+      const parentItemId = TurnItemId.make("turn-item:old-native-parent");
+      yield* ingest({
+        type: "turn_item.updated",
+        driver: CODEX_DRIVER,
+        turnItem: {
+          id: parentItemId,
+          threadId: root.threadId!,
+          runId: null,
+          nodeId: task.id,
+          providerThreadId: null,
+          providerTurnId: null,
+          nativeItemRef: { driver: CODEX_DRIVER, nativeId: "old-task", strength: "strong" },
+          parentItemId: null,
+          ordinal: 2,
+          status: "waiting",
+          title: "Review",
+          startedAt: now,
+          completedAt: null,
+          updatedAt: now,
+          type: "subagent",
+          subagentId: task.id,
+          origin: "provider_native",
+          driver: CODEX_DRIVER,
+          providerInstanceId: modelSelection.instanceId,
+          childThreadId,
+          prompt: "Review",
+          result: null,
+        },
+      });
+      const otherChildRootId = NodeId.make("node:old-native-other-child-root");
+      yield* ingest({
+        type: "node.updated",
+        driver: CODEX_DRIVER,
+        node: {
+          ...childRoot,
+          id: otherChildRootId,
+          rootNodeId: otherChildRootId,
+          nativeItemRef: { ...task.nativeTaskRef, nativeId: "other-task" },
+        },
+      });
+      const childItemId = TurnItemId.make("turn-item:old-native-child");
+      const childItem = {
+        id: childItemId,
+        threadId: childThreadId,
+        runId: null,
+        nodeId: childRootId,
+        providerThreadId: null,
+        providerTurnId: null,
+        nativeItemRef: { driver: CODEX_DRIVER, nativeId: "child-reasoning", strength: "strong" },
+        parentItemId: null,
+        ordinal: 1,
+        status: "waiting",
+        title: "Thinking",
+        startedAt: now,
+        completedAt: null,
+        updatedAt: now,
+        type: "reasoning",
+        text: "partial",
+        streaming: true,
+      } satisfies OrchestrationV2TurnItem;
+      yield* ingest({ type: "turn_item.updated", driver: CODEX_DRIVER, turnItem: childItem });
+      const otherChildItemId = TurnItemId.make("turn-item:old-native-other-child");
+      yield* ingest({
+        type: "turn_item.updated",
+        driver: CODEX_DRIVER,
+        turnItem: {
+          ...childItem,
+          id: otherChildItemId,
+          nodeId: otherChildRootId,
+          nativeItemRef: { ...task.nativeTaskRef, nativeId: "other-reasoning" },
+          status: "running",
+        },
+      });
+      yield* ingest({
+        type: "subagent.updated",
+        driver: CODEX_DRIVER,
+        subagent: {
+          ...task,
+          id: NodeId.make("node:other-task"),
+          nativeTaskRef: { ...task.nativeTaskRef, nativeId: "other-task" },
+        },
+      });
+      const ended = {
+        type: "subagent.native-task-ended" as const,
+        driver: CODEX_DRIVER,
+        threadId: root.threadId!,
+        nativeTaskId: "old-task",
+        status: "cancelled" as const,
+        result: "Previous session ended",
+      };
+      yield* ingest(ended, ProviderInstanceId.make("codex-replacement"));
+      const projection = yield* store.getThreadProjection(root.threadId!);
+      const updated = projection.subagents.find((candidate) => candidate.id === task.id);
+      assert.equal(updated?.status, "cancelled");
+      assert.equal(projection.nodes.find((node) => node.id === task.id)?.status, "cancelled");
+      assert.equal(updated?.providerInstanceId, modelSelection.instanceId);
+      assert.equal(updated?.result, "Previous session ended");
+      assert.equal(
+        projection.turnItems.find((item) => item.id === parentItemId)?.status,
+        "cancelled",
+      );
+
+      const childProjection = yield* store.getThreadProjection(childThreadId);
+      assert.equal(
+        childProjection.nodes.find((node) => node.id === childRootId)?.status,
+        "cancelled",
+      );
+      assert.equal(
+        childProjection.nodes.find((node) => node.id === otherChildRootId)?.status,
+        "running",
+      );
+      const settledChildItem = childProjection.turnItems.find((item) => item.id === childItemId);
+      assert.equal(settledChildItem?.status, "cancelled");
+      assert.isNotNull(settledChildItem?.completedAt);
+      if (settledChildItem?.type !== "reasoning") {
+        assert.fail("expected the reconciled child item to be reasoning");
+      }
+      assert.isFalse(settledChildItem.streaming);
+      assert.equal(
+        childProjection.turnItems.find((item) => item.id === otherChildItemId)?.status,
+        "running",
+      );
+
+      assert.equal(
+        projection.subagents.find((candidate) => candidate.id !== task.id)?.status,
+        "running",
+      );
+      assert.lengthOf(yield* ingest(ended, ProviderInstanceId.make("codex-replacement")), 0);
+      assert.deepEqual(yield* store.getThreadProjection(root.threadId!), projection);
+      assert.deepEqual(yield* store.getThreadProjection(childThreadId), childProjection);
+    }),
+  );
 });
```

## apps/server/src/orchestration-v2/ProviderEventIngestor.ts

Status: modified; +96; −0

```diff
@@ -19,6 +19,7 @@
   RunAttemptId,
   RunId,
   ThreadId,
+  isOrchestrationV2WorkActive,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
@@ -617,6 +618,101 @@
                 nodeId: input.event.subagent.id,
               }),
             ];
+          case "subagent.native-task-ended": {
+            const ended = input.event;
+            const projection = yield* projections.getThreadRecords(ended.threadId, [
+              "subagents",
+              "nodes",
+              "turnItems",
+            ]);
+            const task = projection.subagents.find(
+              (task) =>
+                task.origin === "provider_native" &&
+                task.driver === ended.driver &&
+                task.nativeTaskRef?.nativeId === ended.nativeTaskId &&
+                isOrchestrationV2WorkActive(task.status),
+            );
+            if (task === undefined) return [];
+            const now = yield* DateTime.now;
+            const owner = { ...input, providerInstanceId: task.providerInstanceId };
+            const events = [
+              yield* makeDomainEvent(owner, {
+                type: "subagent.updated",
+                threadId: task.threadId,
+                runId: task.runId,
+                nodeId: task.id,
+                payload: {
+                  ...task,
+                  status: ended.status,
+                  result: ended.result,
+                  completedAt: now,
+                  updatedAt: now,
+                },
+              }),
+            ];
+            const nodes = projection.nodes.filter(
+              (node) => node.id === task.id && isOrchestrationV2WorkActive(node.status),
+            );
+            const turnItems = projection.turnItems.filter(
+              (item) => item.nodeId === task.id && isOrchestrationV2WorkActive(item.status),
+            );
+            if (task.childThreadId !== null) {
+              const child = yield* projections.getThreadRecords(task.childThreadId, [
+                "nodes",
+                "turnItems",
+              ]);
+              // A child thread may contain other roots; the receipt owns only this native task.
+              const childRoots = child.nodes.filter(
+                (node) =>
+                  node.kind === "root_turn" &&
+                  isOrchestrationV2WorkActive(node.status) &&
+                  node.nativeItemRef?.driver === ended.driver &&
+                  node.nativeItemRef.nativeId === ended.nativeTaskId,
+              );
+              const childRootIds = new Set(childRoots.map((node) => node.id));
+              nodes.push(...childRoots);
+              turnItems.push(
+                ...child.turnItems.filter(
+                  (item) =>
+                    item.nodeId !== null &&
+                    childRootIds.has(item.nodeId) &&
+                    isOrchestrationV2WorkActive(item.status),
+                ),
+              );
+            }
+            for (const node of nodes) {
+              events.push(
+                yield* makeDomainEvent(owner, {
+                  type: "node.updated",
+                  threadId: node.threadId,
+                  runId: node.runId,
+                  nodeId: node.id,
+                  payload: { ...node, status: ended.status, completedAt: now },
+                }),
+              );
+            }
+            for (const item of turnItems) {
+              events.push(
+                yield* makeDomainEvent(owner, {
+                  type: "turn-item.updated",
+                  threadId: item.threadId,
+                  runId: item.runId,
+                  nodeId: item.nodeId,
+                  payload: {
+                    ...item,
+                    ...(item.type === "subagent" && item.nodeId === task.id
+                      ? { result: ended.result }
+                      : {}),
+                    ...("streaming" in item ? { streaming: false } : {}),
+                    status: ended.status,
+                    completedAt: now,
+                    updatedAt: now,
+                  },
+                }),
+              );
+            }
+            return events;
+          }
           case "message.updated":
             return [
               yield* makeDomainEvent(input, {
```

## apps/server/src/orchestration-v2/ProviderSessionManager.test.ts

Status: modified; +57; −1

```diff
@@ -4,6 +4,7 @@
 import { AgentScope } from "@t3tools/shared/AgentScope";
 import {
   EnvironmentId,
+  NodeId,
   type ModelSelection,
   type OrchestrationV2AppThread,
   type OrchestrationV2DomainEvent,
@@ -3999,7 +4000,7 @@
       yield* eventSink.write({
         events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
       });
-      yield* manager.open({
+      const runtime = yield* manager.open({
         threadId,
         providerSessionId,
         modelSelection,
@@ -4026,7 +4027,61 @@
       );
       const adapterEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
       assert.isDefined(adapterEvents);
+      const taskId = NodeId.make("node:idle-native-task");
+      yield* eventSink.write({
+        events: [
+          {
+            id: yield* idAllocator.allocate.event({ threadId }),
+            type: "subagent.updated",
+            threadId,
+            occurredAt: now,
+            payload: {
+              id: taskId,
+              threadId,
+              runId: null,
+              parentNodeId: NodeId.make("node:old-root"),
+              origin: "provider_native",
+              createdBy: "agent",
+              driver: CODEX_DRIVER,
+              providerInstanceId: modelSelection.instanceId,
+              providerThreadId: null,
+              childThreadId: null,
+              nativeTaskRef: { driver: CODEX_DRIVER, nativeId: "old-task", strength: "strong" },
+              prompt: "Review",
+              title: "Review",
+              model: null,
+              status: "running",
+              result: null,
+              startedAt: now,
+              completedAt: null,
+              updatedAt: now,
+            },
+          },
+        ],
+      });
+      const subscription = yield* runtime.subscribeEvents!;
+      yield* Queue.offer(adapterEvents!, {
+        type: "subagent.native-task-ended",
+        driver: CODEX_DRIVER,
+        threadId,
+        nativeTaskId: "old-task",
+        status: "cancelled",
+        result: "Previous session ended",
+      });
       yield* Queue.offerAll(adapterEvents!, pendingRequest.providerEvents);
+      yield* Queue.offer(adapterEvents!, {
+        type: "provider_thread.updated",
+        driver: CODEX_DRIVER,
+        providerThread,
+      });
+      const published = yield* subscription.events.pipe(
+        Stream.takeUntil((event) => event.type === "provider_thread.updated"),
+        Stream.runCollect,
+      );
+      assert.deepEqual(
+        published.map((event) => event.type),
+        ["subagent.native-task-ended", "provider_thread.updated"],
+      );
       const persisted = Array.from(yield* Fiber.join(persistedFiber));
 
       assert.sameMembers(
@@ -4034,6 +4089,7 @@
         ["runtime-request.updated", "node.updated", "turn-item.updated"],
       );
       const projection = yield* projectionStore.getThreadProjection(threadId);
+      assert.equal(projection.subagents.find((task) => task.id === taskId)?.status, "cancelled");
       const request = projection.runtimeRequests.find(
         (candidate) => candidate.id === pendingRequest.requestId,
       );
```

## apps/server/src/orchestration-v2/ProviderSessionManager.ts

Status: modified; +7; −7

```diff
@@ -248,10 +248,12 @@
   return String(providerSessionId);
 }
 
-function sessionScopedRuntimeRequestThreadId(
+function sessionScopedEventThreadId(
   event: ProviderAdapter.ProviderAdapterV2Event,
 ): ThreadId | undefined {
   switch (event.type) {
+    case "subagent.native-task-ended":
+      return event.threadId;
     case "runtime_request.updated":
       return event.runtimeRequest.providerTurnId === null ? event.threadId : undefined;
     case "node.updated":
@@ -1952,11 +1954,8 @@
               ),
               Effect.andThen(
                 Effect.gen(function* () {
-                  // Some providers can block before a run subscriber exists
-                  // (project trust, login, or session-switch hooks). Persist
-                  // their runless request artifacts directly so the normal T3
-                  // request UI can answer them and unblock session setup.
-                  const threadId = sessionScopedRuntimeRequestThreadId(event);
+                  // Requests and native task receipts can arrive without a run subscriber.
+                  const threadId = sessionScopedEventThreadId(event);
                   if (threadId !== undefined) {
                     yield* Effect.gen(function* () {
                       const current = (yield* Ref.get(sessions)).get(
@@ -1981,7 +1980,8 @@
                           ),
                         );
                     }).pipe(entry.requestEventPermit.withPermits(1));
+                    // Runs need receipts to release inherited work; other session-scoped events stay here.
+                    if (event.type !== "subagent.native-task-ended") return;
-                    return;
                   }
                   yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event });
                   yield* releaseLiveOutput;
```

## apps/server/src/orchestration-v2/RunExecutionService.test.ts

Status: modified; +72; −12

```diff
@@ -378,7 +378,7 @@
   };
   const initial = RunExecutionService.makeProviderEventRoutingState({
     identity,
-    inheritedBackgroundTurnItems: [{ id: itemId, runId: priorRunId }],
+    inheritedBackgroundTurnItems: [{ id: itemId, runId: priorRunId, nativeTaskId: null }],
     providerTurnId: null,
   });
   const inheritedRunning = {
@@ -559,9 +559,9 @@
   });
 
   assert.deepEqual(selected, [
-    { id: inheritedItemId, runId: interruptedRunId },
+    { id: inheritedItemId, runId: interruptedRunId, nativeTaskId: null },
+    { id: failedItemId, runId: failedRunId, nativeTaskId: null },
+    { id: cancelledItemId, runId: cancelledRunId, nativeTaskId: null },
-    { id: failedItemId, runId: failedRunId },
-    { id: cancelledItemId, runId: cancelledRunId },
   ]);
 });
 
@@ -1783,7 +1783,9 @@
       ],
       {
         loadInheritedBackgroundTurnItems: () =>
+          Effect.succeed([
+            { id: TurnItemId.make(`turn-item:${key}`), runId: priorRunId, nativeTaskId: null },
+          ]),
-          Effect.succeed([{ id: TurnItemId.make(`turn-item:${key}`), runId: priorRunId }]),
       },
     );
 
@@ -1804,7 +1806,9 @@
       ],
       {
         loadInheritedBackgroundTurnItems: () =>
+          Effect.succeed([
+            { id: TurnItemId.make(`turn-item:${key}`), runId: priorRunId, nativeTaskId: null },
-          Effect.succeed([{ id: TurnItemId.make(`turn-item:${key}`), runId: priorRunId }]),
+          ]),
       },
     );
 
@@ -1812,6 +1816,40 @@
   }),
 );
 
+it.effect("releases an inherited background item when its native task receipt settles it", () =>
+  Effect.gen(function* () {
+    const key = "inherited-background-receipt";
+    const priorRunId = RunId.make(`run:${key}:prior`);
+    const observed = yield* runBackgroundItemScenario(
+      key,
+      (ids) => [
+        rootTerminalEvent(ids, "completed"),
+        {
+          type: "subagent.native-task-ended",
+          driver,
+          threadId: ids.threadId,
+          nativeTaskId: "old-task",
+          status: "cancelled",
+          result: "Previous session ended",
+        },
+      ],
+      {
+        requireCloseAfterEvents: true,
+        loadInheritedBackgroundTurnItems: () =>
+          Effect.succeed([
+            {
+              id: TurnItemId.make(`turn-item:${key}`),
+              runId: priorRunId,
+              nativeTaskId: "old-task",
+            },
+          ]),
+      },
+    );
+
+    assert.deepEqual(observed, ["root-finalized"]);
+  }),
+);
+
 it.effect("does not hold the live stream open for a foreign provider's background item", () =>
   Effect.gen(function* () {
     const key = "inherited-background-foreign-provider";
@@ -4373,8 +4411,9 @@
   makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>,
   options?: {
     readonly keepEventStreamOpen?: boolean;
+    readonly requireCloseAfterEvents?: boolean;
     readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect<
-      ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }>
+      ReadonlyArray<RunExecutionService.InheritedBackgroundTurnItemRoute>
     >;
     readonly onSubscribe?: Effect.Effect<void>;
     readonly onFinalized?: (
@@ -4387,6 +4426,7 @@
     const providerInstanceId = ProviderInstanceId.make("codex");
     const observed = yield* Ref.make<ReadonlyArray<string>>([]);
     const ingestionDone = yield* Deferred.make<void>();
+    const eventsExhausted = yield* Deferred.make<void>();
     const layerTest = RunExecutionService.layer.pipe(
       Layer.provide(
         Layer.mergeAll(
@@ -4414,6 +4454,9 @@
             ingestNormalized: (input) =>
               Effect.gen(function* () {
                 const event = input.event;
+                if (event.type === "subagent.native-task-ended") {
+                  yield* Ref.update(observed, (current) => [...current, "receipt-ingested-twice"]);
+                }
                 if (event.type === "turn_item.updated") {
                   yield* Ref.update(observed, (current) => [
                     ...current,
@@ -4447,9 +4490,18 @@
             const events = Stream.fromIterable(makeEvents(ids));
             return {
               events:
-                options?.keepEventStreamOpen === true
+                options?.requireCloseAfterEvents === true
+                  ? events.pipe(
+                      Stream.concat(
-                  ? events.pipe(Stream.concat(Stream.never))
-                  : events,
+                        Stream.fromEffect(Deferred.succeed(eventsExhausted, undefined)).pipe(
+                          Stream.drain,
+                        ),
+                      ),
+                      Stream.concat(Stream.never),
+                    )
+                  : options?.keepEventStreamOpen === true
+                    ? events.pipe(Stream.concat(Stream.never))
+                    : events,
               close: Deferred.succeed(ingestionDone, undefined),
             };
           }),
@@ -4502,8 +4554,16 @@
       });
     }).pipe(Effect.provide(layerTest));
 
+    if (options?.requireCloseAfterEvents) {
+      const closed = yield* Effect.race(
+        Deferred.await(ingestionDone).pipe(Effect.as(true)),
+        Deferred.await(eventsExhausted).pipe(Effect.as(false)),
+      );
-    const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
-    assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish");
+      assert.isTrue(closed, "native receipt did not release the inherited stream");
+    } else {
+      const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
+      assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish");
+    }
     return yield* Ref.get(observed);
   });
 }
```

## apps/server/src/orchestration-v2/RunExecutionService.ts

Status: modified; +22; −2

```diff
@@ -75,6 +75,7 @@
 export interface InheritedBackgroundTurnItemRoute {
   readonly id: TurnItemId;
   readonly runId: OrchestrationV2Run["id"];
+  readonly nativeTaskId: string | null;
 }
 
 type ProviderTerminalEvent = Extract<
@@ -137,7 +138,14 @@
     settledPriorRunIds.has(turnItem.runId) &&
     backgroundCapableTurnItemTypes.has(turnItem.type) &&
     !isSettledTurnItemStatus(turnItem.status)
+      ? [
-      ? [{ id: turnItem.id, runId: turnItem.runId }]
+          {
+            id: turnItem.id,
+            runId: turnItem.runId,
+            nativeTaskId:
+              turnItem.type === "subagent" ? (turnItem.nativeItemRef?.nativeId ?? null) : null,
+          },
+        ]
       : [],
   );
 }
@@ -369,6 +377,8 @@
   });
 
   switch (event.type) {
+    case "subagent.native-task-ended":
+      return [event.threadId === input.threadId, state];
     case "provider_session.updated":
       return [false, state];
     case "app_thread.created": {
@@ -1164,6 +1174,15 @@
                   });
                 }
               }
+              if (event.type === "subagent.native-task-ended") {
+                yield* Ref.update(activeBackgroundTurnItems, (current) => {
+                  const next = new Set(current);
+                  for (const item of inheritedBackgroundTurnItems) {
+                    if (item.nativeTaskId === event.nativeTaskId) next.delete(item.id);
+                  }
+                  return next;
+                });
+              }
             });
           const shouldStopProviderEventIngestion = Effect.gen(function* () {
             if (!(yield* Ref.get(rootTerminalSeen))) {
@@ -1225,7 +1244,8 @@
                   event,
                   DateTime.toEpochMillis(yield* DateTime.now),
                 );
+                // Receipts are already persisted by the session pump; this run only releases tracking.
+                if (deliveredEvent && event.type !== "subagent.native-task-ended") {
-                if (deliveredEvent) {
                   const rootTerminalAlreadySeen = yield* Ref.get(rootTerminalSeen);
                   const isRootProviderThreadUpdate =
                     event.type === "provider_thread.updated" &&
```

## apps/web/src/components/files/AttachmentFilePreview.browser.test.tsx

Status: modified; +4; −3

```diff
@@ -70,9 +70,10 @@
     now = 61 * 60_000;
     await toggleMode("Show HTML source");
     expect(fetch).toHaveBeenCalledExactlyOnceWith(renewedUrl, expect.any(Object));
+    await act(async () => {
+      await import("./ReadOnlySourcePreview");
+    });
+    expect(container.querySelector("pre")?.textContent).toBe("<p>Captured HTML</p>");
-    await vi.waitFor(() =>
-      expect(container.querySelector("pre")?.textContent).toBe("<p>Captured HTML</p>"),
-    );
 
     await toggleMode("Show rendered page");
     expect(container.querySelector("iframe")?.getAttribute("src")).toBe(renewedUrl);
```

## packages/provider-core/src/server/ProviderAdapter.ts

Status: modified; +8; −0

```diff
@@ -110,6 +110,14 @@
     subagent: OrchestrationV2Subagent,
   }),
   Schema.Struct({
+    type: Schema.Literal("subagent.native-task-ended"),
+    driver: ProviderDriverKind,
+    threadId: ThreadId,
+    nativeTaskId: Schema.String,
+    status: Schema.Literals(["completed", "failed", "cancelled"]),
+    result: Schema.String,
+  }),
+  Schema.Struct({
     type: Schema.Literal("message.updated"),
     driver: ProviderDriverKind,
     message: OrchestrationV2ConversationMessage,
```