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

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

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

Visibility: public

[fix\(server\): withdraw a stopped turn's pending approvals](https://git.cafe/pingdotgg/t3code/pull/22023.md)

Version: 16

Head commit: b3dbe20998016d0755b2053369797fc8571d2fc3

Base commit: 3b95e7831b11b17627dff16843bc087bfca397ec

Comparison base: 3b95e7831b11b17627dff16843bc087bfca397ec

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

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

Status: modified; +7; −4

```diff
@@ -10,6 +10,7 @@
   RuntimeRequestId,
   NodeId,
   type ProjectId,
+  ProviderRequestKind,
   ThreadId,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
@@ -291,8 +292,9 @@
           );
       });
 
+    // A user can answer a question or approval after terminal normalization reads it pending.
-    // A user can answer after terminal normalization reads the pending request.
     // Recheck inside the write transaction so stale cleanup cannot erase answers.
+    const isApprovalRequestKind = Schema.is(ProviderRequestKind);
     const guardUserInputCancellations =
       /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
       Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2DomainEvent>) {
@@ -301,7 +303,7 @@
         for (const event of events) {
           if (
             event.type !== "runtime-request.updated" ||
+            (event.payload.kind !== "user_input" && !isApprovalRequestKind(event.payload.kind)) ||
-            event.payload.kind !== "user_input" ||
             event.payload.status !== "cancelled"
           )
             continue;
@@ -311,7 +313,7 @@
           );
           if (
             current?.status !== "pending" ||
-            current.kind !== "user_input" ||
+            current.kind !== event.payload.kind ||
             current.providerTurnId !== event.payload.providerTurnId ||
             current.responseCapability.type === "message"
           ) {
@@ -327,7 +329,8 @@
               return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
             case "turn-item.updated":
               return (
+                (event.payload.type !== "user_input_request" &&
-                event.payload.type !== "user_input_request" ||
+                  event.payload.type !== "approval_request") ||
                 event.payload.status !== "cancelled" ||
                 !staleRequests.has(event.payload.requestId)
               );
```

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

Status: modified; +123; −0

```diff
@@ -911,6 +911,129 @@
   backgroundThreadUpdatesKeepTimestamp.pipe(Effect.provide(layerTest)),
 );
 
+const pendingNativeRequestsStayWithTheirTurn = Effect.gen(function* () {
+  const store = yield* ProjectionStore.ProjectionStoreV2;
+  const now = yield* DateTime.now;
+  const suffix = "pending-native-requests";
+  const threadId = yield* addRolledBackRecoveryCandidate(suffix);
+  const providerThreadId = ProviderThreadId.make(`provider-thread:${suffix}`);
+  const turnId = ProviderTurnId.make(`turn:${suffix}`);
+  const otherTurnId = ProviderTurnId.make(`turn:${suffix}:other`);
+  const live = {
+    type: "live" as const,
+    providerSessionId: ProviderSessionId.make(`session:${suffix}`),
+  };
+  const requests = [
+    ["approval", turnId, "command", "pending", live],
+    ["other-turn-approval", otherTurnId, "command", "pending", live],
+    ["session-approval", null, "command", "pending", live],
+    ["resolved-approval", turnId, "command", "resolved", live],
+    ["dynamic-tool", turnId, "dynamic_tool_call", "pending", live],
+    ["question", turnId, "user_input", "pending", live],
+    ["message-question", turnId, "user_input", "pending", { type: "message" as const }],
+  ] as const;
+  for (const [
+    ordinal,
+    [name, providerTurnId, kind, status, responseCapability],
+  ] of requests.entries()) {
+    const requestId = RuntimeRequestId.make(`request:${suffix}:${name}`);
+    const nodeId = NodeId.make(`node:${suffix}:${name}`);
+    const approval = kind === "command";
+    yield* store.apply({
+      id: EventId.make(`event:${suffix}:${name}:request`),
+      type: "runtime-request.updated",
+      threadId,
+      nodeId,
+      occurredAt: now,
+      payload: {
+        id: requestId,
+        nodeId,
+        providerTurnId,
+        nativeRequestRef: null,
+        kind,
+        status,
+        responseCapability,
+        createdAt: now,
+        resolvedAt: status === "pending" ? null : now,
+      },
+    });
+    yield* store.apply({
+      id: EventId.make(`event:${suffix}:${name}:node`),
+      type: "node.updated",
+      threadId,
+      nodeId,
+      occurredAt: now,
+      payload: {
+        id: nodeId,
+        threadId,
+        runId: null,
+        parentNodeId: null,
+        rootNodeId: nodeId,
+        kind: approval ? "approval_request" : "user_input_request",
+        status: "waiting",
+        countsForRun: false,
+        providerThreadId,
+        providerTurnId,
+        nativeItemRef: null,
+        runtimeRequestId: requestId,
+        checkpointScopeId: null,
+        startedAt: now,
+        completedAt: null,
+      },
+    });
+    if (kind === "dynamic_tool_call") continue;
+    yield* store.apply({
+      id: EventId.make(`event:${suffix}:${name}:item`),
+      type: "turn-item.updated",
+      threadId,
+      nodeId,
+      occurredAt: now,
+      payload: {
+        id: TurnItemId.make(`item:${suffix}:${name}`),
+        threadId,
+        runId: null,
+        nodeId,
+        providerThreadId,
+        providerTurnId,
+        nativeItemRef: null,
+        parentItemId: null,
+        ordinal,
+        status: "waiting",
+        title: null,
+        startedAt: now,
+        completedAt: null,
+        updatedAt: now,
+        ...(approval
+          ? { type: "approval_request" as const, requestId, requestKind: "command" as const }
+          : { type: "user_input_request" as const, requestId, questions: [] }),
+      },
+    });
+  }
+
+  const names = (records: ReadonlyArray<{ readonly id: string }>) =>
+    records.map((record) => record.id.slice(record.id.lastIndexOf(":") + 1)).toSorted();
+  const withoutApprovals = yield* store.getPendingNativeRequests(threadId, turnId, {
+    includeApprovals: false,
+  });
+  assert.deepEqual(names(withoutApprovals.runtimeRequests), ["question"]);
+  assert.deepEqual(names(withoutApprovals.nodes), ["question"]);
+  assert.deepEqual(names(withoutApprovals.turnItems), ["question"]);
+  const withApprovals = yield* store.getPendingNativeRequests(threadId, turnId, {
+    includeApprovals: true,
+  });
+  assert.deepEqual(names(withApprovals.runtimeRequests), ["approval", "question"]);
+  assert.deepEqual(names(withApprovals.nodes), ["approval", "question"]);
+  assert.deepEqual(names(withApprovals.turnItems), ["approval", "question"]);
+});
+
+it.effect("memory pending native requests: approvals are opt-in and stay with their turn", () =>
+  pendingNativeRequestsStayWithTheirTurn.pipe(Effect.provide(ProjectionStore.layerMemory)),
+);
+
+it.effect("SQL pending native requests: approvals are opt-in and stay with their turn", () =>
+  pendingNativeRequestsStayWithTheirTurn.pipe(Effect.provide(layerTest)),
+);
+
 it.effect("memory recovery selection includes unfinished items from missing runs", () =>
   Effect.gen(function* () {
     const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
```

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

Status: modified; +51; −31

```diff
@@ -21,6 +21,7 @@
   OrchestrationV2ExecutionNode,
   OrchestrationV2PlanArtifact,
   OrchestrationV2PendingRuntimeRequestSummary,
+  OrchestrationV2RuntimeRequest,
   PlanId,
   OrchestrationV2ConversationMessage,
   OrchestrationV2DomainEvent,
@@ -67,6 +68,7 @@
   TurnItemId,
   NodeId,
   ProviderInstanceId,
+  ProviderRequestKind,
 } from "@t3tools/contracts";
 import {
   createOrchestrationV2TurnItemVisibility,
@@ -300,11 +302,22 @@
   | "turnItems"
 >;
 
-export type ProjectionPendingUserInputs = Pick<
+export type ProjectionPendingNativeRequests = Pick<
   OrchestrationV2ThreadProjection,
   "runtimeRequests" | "nodes" | "turnItems"
 >;
 
+const isApprovalRequestKind = Schema.is(ProviderRequestKind);
+
+function isPendingNativeRequest(
+  request: OrchestrationV2RuntimeRequest,
+  includeApprovals: boolean,
+): boolean {
+  return request.kind === "user_input"
+    ? request.responseCapability.type !== "message"
+    : includeApprovals && isApprovalRequestKind(request.kind);
+}
+
 export type ProjectionThreadProviderContext = Pick<
   OrchestrationV2ThreadProjection,
   "thread" | "providerSessions" | "providerThreads"
@@ -489,10 +502,11 @@
   readonly getRuntimeRecoveryProjection: (
     threadId: ThreadId,
   ) => Effect.Effect<ProjectionRuntimeRecoveryState, ProjectionStoreV2Error>;
-  readonly getPendingNativeUserInputs: (
+  readonly getPendingNativeRequests: (
     threadId: ThreadId,
     providerTurnId: ProviderTurnId,
+    options: { readonly includeApprovals: boolean },
+  ) => Effect.Effect<ProjectionPendingNativeRequests, ProjectionStoreV2Error>;
-  ) => Effect.Effect<ProjectionPendingUserInputs, ProjectionStoreV2Error>;
   readonly getPlan: (
     threadId: ThreadId,
     planId: PlanId,
@@ -4647,9 +4661,10 @@
         ? cause
         : new ProjectionStoreReadError({ threadId, cause });
 
+    const getPendingNativeRequests: ProjectionStoreV2Shape["getPendingNativeRequests"] = (
-    const getPendingNativeUserInputs: ProjectionStoreV2Shape["getPendingNativeUserInputs"] = (
       threadId,
       providerTurnId,
+      { includeApprovals },
     ) =>
       sql
         .withTransaction(
@@ -4657,12 +4672,12 @@
             const rows = yield* sql<PayloadRow>`
           SELECT payload_json FROM orchestration_v2_projection_runtime_requests
           WHERE thread_id = ${threadId} AND provider_turn_id = ${providerTurnId}
+            AND status = 'pending'
-            AND kind = 'user_input' AND status = 'pending'
         `;
             const runtimeRequests = (yield* decodeRows(
               decodeRuntimeRequestPayload,
               threadId,
-            )(rows)).filter((request) => request.responseCapability.type !== "message");
+            )(rows)).filter((request) => isPendingNativeRequest(request, includeApprovals));
             if (runtimeRequests.length === 0) return { runtimeRequests, nodes: [], turnItems: [] };
             const nodeRows = yield* sql<PayloadRow>`
           SELECT payload_json FROM orchestration_v2_projection_nodes
@@ -4674,14 +4689,16 @@
             const itemRows = yield* sql<PayloadRow>`
           SELECT payload_json FROM orchestration_v2_projection_turn_items
           WHERE thread_id = ${threadId} AND provider_turn_id = ${providerTurnId}
+            AND type IN ('user_input_request', 'approval_request')
-            AND type = 'user_input_request'
         `;
             const requestIds = new Set(runtimeRequests.map((request) => request.id));
             return {
               runtimeRequests,
               nodes: yield* decodeRows(decodeNodePayload, threadId)(nodeRows),
               turnItems: (yield* decodeRows(decodeTurnItemPayload, threadId)(itemRows)).filter(
-                (item) => item.type === "user_input_request" && requestIds.has(item.requestId),
+                (item) =>
+                  (item.type === "user_input_request" || item.type === "approval_request") &&
+                  requestIds.has(item.requestId),
               ),
             };
           }),
@@ -6503,7 +6520,7 @@
       getCheckpointRollbackContext,
       getRunMessage,
       canStartQueuedRun,
+      getPendingNativeRequests,
-      getPendingNativeUserInputs,
       hasUnpairedRunInterruptRequest,
       isNodeStored,
       getMessageCount,
@@ -6743,28 +6760,31 @@
           .map((projection) => projection.thread.id);
       }),
       getUnreadableThreadIds: () => Effect.succeed([]),
-      getPendingNativeUserInputs: Effect.fn("ProjectionStore.getPendingNativeUserInputs")(
-        function* (threadId, providerTurnId) {
-          const projection = (yield* Ref.get(replayState)).projections.get(threadId);
-          if (projection === undefined) return { runtimeRequests: [], nodes: [], turnItems: [] };
-          const runtimeRequests = projection.runtimeRequests.filter(
-            (request) =>
+      getPendingNativeRequests: Effect.fn("ProjectionStore.getPendingNativeRequests")(function* (
+        threadId,
+        providerTurnId,
+        { includeApprovals },
+      ) {
+        const projection = (yield* Ref.get(replayState)).projections.get(threadId);
+        if (projection === undefined) return { runtimeRequests: [], nodes: [], turnItems: [] };
+        const runtimeRequests = projection.runtimeRequests.filter(
+          (request) =>
+            request.providerTurnId === providerTurnId &&
+            request.status === "pending" &&
+            isPendingNativeRequest(request, includeApprovals),
+        );
+        const requestIds = new Set(runtimeRequests.map((request) => request.id));
+        const nodeIds = new Set(runtimeRequests.map((request) => request.nodeId));
+        return {
+          runtimeRequests,
+          nodes: projection.nodes.filter((node) => nodeIds.has(node.id)),
+          turnItems: projection.turnItems.filter(
+            (item) =>
+              (item.type === "user_input_request" || item.type === "approval_request") &&
-              request.providerTurnId === providerTurnId &&
-              request.kind === "user_input" &&
-              request.status === "pending" &&
-              request.responseCapability.type !== "message",
-          );
-          const requestIds = new Set(runtimeRequests.map((request) => request.id));
-          const nodeIds = new Set(runtimeRequests.map((request) => request.nodeId));
-          return {
-            runtimeRequests,
-            nodes: projection.nodes.filter((node) => nodeIds.has(node.id)),
-            turnItems: projection.turnItems.filter(
-              (item) => item.type === "user_input_request" && requestIds.has(item.requestId),
-            ),
-          };
-        },
+              requestIds.has(item.requestId),
+          ),
+        };
-      ),
+      }),
       getPlan: Effect.fn("ProjectionStore.getPlan")(function* (threadId, planId) {
         const projection = (yield* Ref.get(replayState)).projections.get(threadId);
         if (projection === undefined)
```

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

Status: modified; +201; −13

```diff
@@ -963,8 +963,8 @@
     }),
   );
 
-  it.effect.each(["interrupted", "control"] as const)(
-    "dismisses only native questions when a provider turn ends with %s",
+  it.effect.each(["interrupted", "control", "control-interrupted"] as const)(
+    "dismisses native questions, and approvals only of a stopped turn, when a provider turn ends with %s",
     (terminal) =>
       Effect.gen(function* () {
         const now = yield* DateTime.now;
@@ -1001,16 +1001,19 @@
           { key: "answered", responseCapability: { type: "live", providerSessionId } },
           { key: "other-turn", responseCapability: { type: "live", providerSessionId } },
           { key: "approval", responseCapability: { type: "live", providerSessionId } },
+          { key: "other-turn-approval", responseCapability: { type: "live", providerSessionId } },
         ] as const;
+        const stopped = terminal === "interrupted" || terminal === "control-interrupted";
         const fixtures = specs.map((spec, ordinal) => {
           const nodeId = NodeId.make(`${threadId}:${spec.key}`);
           const resolved = spec.key === "answered";
+          const approval = spec.key === "approval" || spec.key === "other-turn-approval";
           const request: OrchestrationV2RuntimeRequest = {
             id: RuntimeRequestId.make(`${threadId}:${spec.key}`),
             nodeId,
-            providerTurnId: spec.key === "other-turn" ? otherTurnId : providerTurnId,
+            providerTurnId: spec.key.startsWith("other-turn") ? otherTurnId : providerTurnId,
             nativeRequestRef: null,
-            kind: spec.key === "approval" ? "command" : "user_input",
+            kind: approval ? "command" : "user_input",
             status: resolved ? "resolved" : "pending",
             responseCapability: spec.responseCapability,
             createdAt: now,
@@ -1022,7 +1025,7 @@
             runId: null,
             parentNodeId: null,
             rootNodeId: nodeId,
-            kind: spec.key === "approval" ? "approval_request" : "user_input_request",
+            kind: approval ? "approval_request" : "user_input_request",
             status: resolved ? "completed" : "waiting",
             countsForRun: false,
             providerThreadId,
@@ -1048,7 +1051,7 @@
             startedAt: now,
             completedAt: resolved ? now : null,
             updatedAt: now,
+            ...(approval
-            ...(spec.key === "approval"
               ? { type: "approval_request", requestId: request.id, requestKind: "command" }
               : {
                   type: "user_input_request",
@@ -1080,14 +1083,15 @@
           providerInstanceId: modelSelection.instanceId,
           threadId,
           event:
-            terminal === "control"
+            terminal === "control" || terminal === "control-interrupted"
               ? {
                   type: "turn.terminal" as const,
                   driver: CODEX_DRIVER,
                   providerThreadId,
                   providerTurnId,
                   runOrdinal: 1,
+                  status:
-                  status: "completed" as const,
+                    terminal === "control" ? ("completed" as const) : ("interrupted" as const),
                   failure: null,
                   threadDisposition: "reusable" as const,
                 }
@@ -1110,7 +1114,10 @@
         const stored = yield* ingestor.ingestNormalized(input);
         const projection = yield* projectionStore.getThreadProjection(threadId);
         for (const fixture of fixtures) {
+          const closed =
+            fixture.key === "native" ||
+            fixture.key === "unavailable" ||
-          const closed = fixture.key === "native" || fixture.key === "unavailable";
+            (fixture.key === "approval" && stopped);
           const request = projection.runtimeRequests.find(
             (item) => item.id === fixture.request.id,
           )!;
@@ -1123,19 +1130,24 @@
             assert.isNotNull(node.completedAt);
             assert.equal(item.status, "cancelled");
             assert.isNotNull(item.completedAt);
+            if (fixture.key === "approval") {
+              assert.equal(request.responseCapability.type, "not_resumable");
+            }
           } else {
             assert.equal(request.status, fixture.request.status);
             assert.equal(node.status, fixture.node.status);
             assert.equal(item.status, fixture.item.status);
           }
         }
+        const dismissedCount = stopped ? 3 : 2;
         assert.equal(
           stored.filter((entry) => entry.event.type === "runtime-request.updated").length,
-          2,
+          dismissedCount,
         );
         assert.isEmpty(
-          (yield* projectionStore.getPendingNativeUserInputs(threadId, providerTurnId))
+          (yield* projectionStore.getPendingNativeRequests(threadId, providerTurnId, {
-            .runtimeRequests,
+            includeApprovals: stopped,
+          })).runtimeRequests,
         );
         const replayed = yield* eventStore.read({ threadId }).pipe(Stream.runCollect);
         assert.equal(
@@ -1144,7 +1156,7 @@
               entry.event.type === "runtime-request.updated" &&
               entry.event.payload.status === "cancelled",
           ).length,
-          2,
+          dismissedCount,
         );
         const repeated = yield* ingestor.ingestNormalized(input);
         assert.isFalse(repeated.some((entry) => entry.event.type === "runtime-request.updated"));
@@ -1323,6 +1335,182 @@
       ),
   );
 
+  it.effect.each(["interrupted", "cancelled"] as const)(
+    "preserves an approval resolved after terminal normalization reads it pending (%s)",
+    (status) =>
+      Effect.scoped(
+        Effect.gen(function* () {
+          const eventSink = yield* EventSink.EventSinkV2;
+          const eventStore = yield* EventStore.EventStoreV2;
+          const projections = yield* ProjectionStore.ProjectionStoreV2;
+          const idAllocator = yield* IdAllocator.IdAllocatorV2;
+          const now = yield* DateTime.now;
+          const threadEvent = yield* threadCreatedEvent(now);
+          const threadId = threadEvent.threadId;
+          const providerSessionId = yield* idAllocator.allocate.providerSession({
+            providerInstanceId: modelSelection.instanceId,
+            threadId,
+          });
+          const providerThreadId = idAllocator.derive.providerThread({
+            driver: CODEX_DRIVER,
+            nativeThreadId: `${threadId}:approval-race`,
+          });
+          const providerTurnId = idAllocator.derive.providerTurn({
+            driver: CODEX_DRIVER,
+            nativeTurnId: `${threadId}:approval-race`,
+          });
+          const nodeId = NodeId.make(`${threadId}:approval`);
+          const request: OrchestrationV2RuntimeRequest = {
+            id: RuntimeRequestId.make(`${threadId}:approval`),
+            nodeId,
+            providerTurnId,
+            nativeRequestRef: null,
+            kind: "command",
+            status: "pending",
+            responseCapability: { type: "live", providerSessionId },
+            createdAt: now,
+            resolvedAt: null,
+          };
+          const node: OrchestrationV2ExecutionNode = {
+            id: nodeId,
+            threadId,
+            runId: null,
+            parentNodeId: null,
+            rootNodeId: nodeId,
+            kind: "approval_request",
+            status: "waiting",
+            countsForRun: false,
+            providerThreadId,
+            providerTurnId,
+            nativeItemRef: null,
+            runtimeRequestId: request.id,
+            checkpointScopeId: null,
+            startedAt: now,
+            completedAt: null,
+          };
+          const item: OrchestrationV2TurnItem = {
+            id: TurnItemId.make(`${threadId}:approval`),
+            threadId,
+            runId: null,
+            nodeId,
+            providerThreadId,
+            providerTurnId,
+            nativeItemRef: null,
+            parentItemId: null,
+            ordinal: 1,
+            status: "waiting",
+            title: null,
+            startedAt: now,
+            completedAt: null,
+            updatedAt: now,
+            type: "approval_request",
+            requestId: request.id,
+            requestKind: "command",
+          };
+          const seedEvents: Array<OrchestrationV2DomainEvent> = [threadEvent];
+          for (const payload of [
+            { type: "runtime-request.updated" as const, payload: request },
+            { type: "node.updated" as const, payload: node },
+            { type: "turn-item.updated" as const, payload: item },
+          ])
+            seedEvents.push({
+              id: yield* idAllocator.allocate.event({ threadId }),
+              threadId,
+              occurredAt: now,
+              ...payload,
+            });
+          yield* eventSink.write({ events: seedEvents });
+          const normalized = yield* Deferred.make<void>();
+          const releaseTerminalWrite = yield* Deferred.make<void>();
+          const gatedSink = EventSink.EventSinkV2.of({
+            ...eventSink,
+            write: (input) =>
+              Deferred.succeed(normalized, undefined).pipe(
+                Effect.andThen(Deferred.await(releaseTerminalWrite)),
+                Effect.andThen(eventSink.write(input)),
+              ),
+          });
+          const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2.pipe(
+            Effect.provide(Layer.fresh(ProviderEventIngestor.layer)),
+            Effect.provideService(EventSink.EventSinkV2, gatedSink),
+          );
+          const terminal = yield* ingestor
+            .ingestNormalized({
+              providerSessionId,
+              providerInstanceId: modelSelection.instanceId,
+              threadId,
+              event: {
+                type: "provider_turn.updated",
+                driver: CODEX_DRIVER,
+                providerTurn: {
+                  id: providerTurnId,
+                  providerThreadId,
+                  nodeId,
+                  runAttemptId: null,
+                  nativeTurnRef: null,
+                  ordinal: 1,
+                  status,
+                  startedAt: now,
+                  completedAt: now,
+                },
+              },
+            })
+            .pipe(Effect.forkScoped);
+          yield* Deferred.await(normalized);
+          const responseEvents: Array<OrchestrationV2DomainEvent> = [];
+          for (const payload of [
+            {
+              type: "runtime-request.updated" as const,
+              payload: {
+                ...request,
+                status: "resolved" as const,
+                decision: "accept" as const,
+                resolvedAt: now,
+              },
+            },
+            {
+              type: "node.updated" as const,
+              payload: { ...node, status: "completed" as const, completedAt: now },
+            },
+            {
+              type: "turn-item.updated" as const,
+              payload: { ...item, status: "completed" as const, completedAt: now },
+            },
+          ])
+            responseEvents.push({
+              id: yield* idAllocator.allocate.event({ threadId }),
+              threadId,
+              occurredAt: now,
+              ...payload,
+            });
+          yield* eventSink.write({ events: responseEvents });
+          yield* Deferred.succeed(releaseTerminalWrite, undefined);
+          const committedTerminal = yield* Fiber.join(terminal);
+          const projection = yield* projections.getThreadProjection(threadId);
+          const approved = projection.runtimeRequests.find((entry) => entry.id === request.id)!;
+          assert.equal(approved.status, "resolved");
+          assert.equal(approved.decision, "accept");
+          assert.equal(projection.nodes.find((entry) => entry.id === nodeId)!.status, "completed");
+          assert.equal(
+            projection.turnItems.find((entry) => entry.id === item.id)!.status,
+            "completed",
+          );
+          assert.deepEqual(
+            committedTerminal.map((stored) => stored.event.type),
+            ["provider-turn.updated"],
+          );
+          const history = yield* eventStore.read({ threadId }).pipe(Stream.runCollect);
+          assert.isFalse(
+            history.some(
+              (stored) =>
+                stored.event.type === "runtime-request.updated" &&
+                stored.event.payload.status === "cancelled",
+            ),
+          );
+        }),
+      ),
+  );
+
   it.effect("persists a failed provider terminal as one expected error item", () =>
     Effect.gen(function* () {
       const now = yield* DateTime.now;
```

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

Status: modified; +27; −5

```diff
@@ -345,13 +345,18 @@
         );
       });
 
-    const dismissNativeUserInputs = Effect.fn("ProviderEventIngestor.dismissNativeUserInputs")(
+    const dismissNativeRequests = Effect.fn("ProviderEventIngestor.dismissNativeRequests")(
       function* (
         input: ProviderEventIngestInput,
         providerTurnId: ProviderTurnId,
+        turnStatus: OrchestrationV2ProviderTurn["status"],
         threadId = input.threadId,
       ) {
-        const pending = yield* projections.getPendingNativeUserInputs(threadId, providerTurnId);
+        // A stopped turn's approval callbacks are gone with it. A completed or failed root turn
+        // can leave subagent turns working whose approvals are filed under the root turn.
+        const pending = yield* projections.getPendingNativeRequests(threadId, providerTurnId, {
+          includeApprovals: turnStatus === "interrupted" || turnStatus === "cancelled",
+        });
         const now = yield* DateTime.now;
         const events: Array<OrchestrationV2DomainEvent> = [];
         for (const request of pending.runtimeRequests) {
@@ -360,7 +365,19 @@
               type: "runtime-request.updated",
               threadId,
               nodeId: request.nodeId,
-              payload: { ...request, status: "cancelled", resolvedAt: now },
+              payload: {
+                ...request,
+                status: "cancelled",
+                resolvedAt: now,
+                ...(request.kind === "user_input"
+                  ? {}
+                  : {
+                      responseCapability: {
+                        type: "not_resumable" as const,
+                        reason: "The provider turn ended before this request was resolved.",
+                      },
+                    }),
+              },
             }),
           );
         }
@@ -585,9 +602,10 @@
                 ...(["completed", "interrupted", "failed", "cancelled"].includes(
                   input.event.providerTurn.status,
                 )
-                  ? yield* dismissNativeUserInputs(
+                  ? yield* dismissNativeRequests(
                       input,
                       input.event.providerTurn.id,
+                      input.event.providerTurn.status,
                       input.event.threadId,
                     )
                   : []),
@@ -682,7 +700,11 @@
               // Live output is routed to the command output hub before ingestion.
               return [];
             case "turn.terminal":
-              const dismissed = yield* dismissNativeUserInputs(input, input.event.providerTurnId);
+              const dismissed = yield* dismissNativeRequests(
+                input,
+                input.event.providerTurnId,
+                input.event.status,
+              );
               if (input.event.status !== "failed") {
                 return dismissed;
               }
```

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

Status: modified; +1; −1

```diff
@@ -236,7 +236,7 @@
           getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"),
           getThreadProviderContext: () => Effect.die("unused getThreadProviderContext"),
           getRuntimeResponseContext: () => Effect.die("unused getRuntimeResponseContext"),
-          getPendingNativeUserInputs: () => Effect.die("unused getPendingNativeUserInputs"),
+          getPendingNativeRequests: () => Effect.die("unused getPendingNativeRequests"),
           getProviderControlContext: (_threadId, target) =>
             Ref.get(projection).pipe(
               Effect.map((current) => ({
```