fix(server): close Claude approvals when the CLI withdraws them

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

@@ -102,7 +102,7 @@ export interface EventSinkV2Shape {
       OrchestrationV2ProviderThread,
       "driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
     >;
-    readonly guardPendingUserInputCancellations?: boolean;
+    readonly guardPendingRuntimeRequestCancellations?: boolean;
     readonly commandId?: CommandId;
     readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
   }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
@@ -116,13 +116,13 @@ export interface EventSinkV2Shape {
       OrchestrationV2ProviderThread,
       "driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
     >;
-    readonly guardPendingUserInputCancellations?: boolean;
+    readonly guardPendingRuntimeRequestCancellations?: boolean;
     readonly commandId?: CommandId;
     readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
     readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
   }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
   readonly writeIfRunCurrent: (input: {
-    readonly guardPendingUserInputCancellations?: boolean;
+    readonly guardPendingRuntimeRequestCancellations?: boolean;
     readonly commandId?: CommandId;
     readonly threadId: ThreadId;
     readonly runId: RunId;
@@ -144,7 +144,7 @@ export interface EventSinkV2Shape {
   >;
   /** Commits only if the expected attempt and ordinal still own the provider thread, protecting newer attempts. */
   readonly writeIfProviderThreadOwner: (input: {
-    readonly guardPendingUserInputCancellations?: boolean;
+    readonly guardPendingRuntimeRequestCancellations?: boolean;
     readonly commandId?: CommandId;
     readonly providerThreadId: ProviderThreadId;
     readonly runId: RunId;
@@ -328,30 +328,55 @@ const layerBase: Layer.Layer<
           );
       });
 
-    // A user can answer after terminal normalization reads the pending request.
+    // A user can answer after terminal normalization reads the pending request, and an approval
+    // answer commits before its outbox delivery reaches a provider that may withdraw the request.
     // Recheck inside the write transaction so stale cleanup cannot erase answers.
-    const guardUserInputCancellations =
+    const guardRuntimeRequestCancellations =
       /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
       Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2DomainEvent>) {
         const staleRequests = new Set<RuntimeRequestId>();
         const staleNodes = new Set<NodeId>();
         for (const event of events) {
+          // Providers can withdraw the node and item in writes of their own, so check each against its request.
           if (
-            event.type !== "runtime-request.updated" ||
-            event.payload.kind !== "user_input" ||
-            event.payload.status !== "cancelled"
-          )
+            event.type === "node.updated" &&
+            event.payload.status === "cancelled" &&
+            event.payload.runtimeRequestId !== null
+          ) {
+            const current = yield* projectionStore.getRuntimeRequest(
+              event.threadId,
+              event.payload.runtimeRequestId,
+            );
+            if (current?.status === "resolved") staleNodes.add(event.payload.id);
+            continue;
+          }
+          if (
+            event.type === "turn-item.updated" &&
+            (event.payload.type === "user_input_request" ||
+              event.payload.type === "approval_request") &&
+            event.payload.status === "cancelled"
+          ) {
+            const current = yield* projectionStore.getRuntimeRequest(
+              event.threadId,
+              event.payload.requestId,
+            );
+            if (current?.status === "resolved") staleRequests.add(event.payload.requestId);
+            continue;
+          }
+          if (event.type !== "runtime-request.updated" || event.payload.status !== "cancelled")
             continue;
           const current = yield* projectionStore.getRuntimeRequest(
             event.threadId,
             event.payload.id,
           );
-          if (
-            current?.status !== "pending" ||
-            current.kind !== "user_input" ||
-            current.providerTurnId !== event.payload.providerTurnId ||
-            current.responseCapability.type === "message"
-          ) {
+          const stale =
+            event.payload.kind === "user_input"
+              ? current?.status !== "pending" ||
+                current.kind !== "user_input" ||
+                current.providerTurnId !== event.payload.providerTurnId ||
+                current.responseCapability.type === "message"
+              : current?.status === "resolved";
+          if (stale) {
             staleRequests.add(event.payload.id);
             staleNodes.add(event.payload.nodeId);
           }
@@ -364,7 +389,8 @@ const layerBase: Layer.Layer<
               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)
               );
@@ -485,8 +511,8 @@ const layerBase: Layer.Layer<
             ? yield* updateDelegatedCompletionOnly(input.events, () => true)
             : input.events;
           const normalized = yield* normalizeEvents(
-            input.guardPendingUserInputCancellations === true
-              ? yield* guardUserInputCancellations(events)
+            input.guardPendingRuntimeRequestCancellations === true
+              ? yield* guardRuntimeRequestCancellations(events)
               : events,
           );
           const committed = yield* eventStore.append({
@@ -562,8 +588,8 @@ const layerBase: Layer.Layer<
             }
 
             const normalized = yield* normalizeEvents(
-              input.guardPendingUserInputCancellations === true
-                ? yield* guardUserInputCancellations(input.events)
+              input.guardPendingRuntimeRequestCancellations === true
+                ? yield* guardRuntimeRequestCancellations(input.events)
                 : input.events,
             );
             const storedEvents = yield* eventStore.append({
@@ -627,8 +653,8 @@ const layerBase: Layer.Layer<
           }
 
           const normalized = yield* normalizeEvents(
-            input.guardPendingUserInputCancellations === true
-              ? yield* guardUserInputCancellations(input.events)
+            input.guardPendingRuntimeRequestCancellations === true
+              ? yield* guardRuntimeRequestCancellations(input.events)
               : input.events,
           );
           const storedEvents = yield* eventStore.append({

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

@@ -567,6 +567,145 @@ layer("ProviderEventIngestorV2", (it) => {
     }),
   );
 
+  // regression: #24690 an answer commits before its outbox delivery reaches the provider
+  it.effect.each([
+    { name: "keeps a committed decline", answer: { decision: "decline", feedback: "No." } },
+    { name: "keeps a committed accept", answer: { decision: "accept" } },
+    { name: "cancels an approval withdrawn before it was projected", answer: null },
+  ] as const)("$name when the provider withdraws the approval", ({ answer }) =>
+    Effect.gen(function* () {
+      const now = yield* DateTime.now;
+      const eventSink = yield* EventSink.EventSinkV2;
+      const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
+      const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
+      const idAllocator = yield* IdAllocator.IdAllocatorV2;
+      const threadEvent = yield* threadCreatedEvent(now);
+      const threadId = threadEvent.threadId;
+      yield* eventSink.write({ events: [threadEvent] });
+      const providerSessionId = yield* idAllocator.allocate.providerSession({
+        providerInstanceId: modelSelection.instanceId,
+        threadId,
+      });
+      const nodeId = NodeId.make(`${threadId}:approval`);
+      const request: OrchestrationV2RuntimeRequest = {
+        id: RuntimeRequestId.make(`${threadId}:approval`),
+        nodeId,
+        providerTurnId: null,
+        nativeRequestRef: null,
+        kind: "command",
+        status: "pending",
+        responseCapability: { type: "live", providerSessionId, supportsFeedback: true },
+        createdAt: now,
+        resolvedAt: null,
+      };
+      const node: OrchestrationV2ExecutionNode = {
+        id: nodeId,
+        threadId,
+        runId: null,
+        parentNodeId: null,
+        rootNodeId: nodeId,
+        kind: "approval_request",
+        status: "waiting",
+        countsForRun: false,
+        providerThreadId: null,
+        providerTurnId: null,
+        nativeItemRef: null,
+        runtimeRequestId: request.id,
+        checkpointScopeId: null,
+        startedAt: now,
+        completedAt: null,
+      };
+      const item: OrchestrationV2TurnItem = {
+        id: TurnItemId.make(`${threadId}:approval`),
+        threadId,
+        runId: null,
+        nodeId,
+        providerThreadId: null,
+        providerTurnId: null,
+        nativeItemRef: null,
+        parentItemId: null,
+        ordinal: 0,
+        status: "waiting",
+        title: null,
+        startedAt: now,
+        completedAt: null,
+        updatedAt: now,
+        type: "approval_request",
+        requestId: request.id,
+        requestKind: "command",
+      };
+      const ingest = (status: "waiting" | "cancelled") =>
+        Effect.forEach(
+          [
+            {
+              type: "node.updated" as const,
+              driver: CODEX_DRIVER,
+              node: status === "waiting" ? node : { ...node, status, completedAt: now },
+            },
+            {
+              type: "runtime_request.updated" as const,
+              driver: CODEX_DRIVER,
+              runtimeRequest:
+                status === "waiting" ? request : { ...request, status, resolvedAt: now },
+            },
+            {
+              type: "turn_item.updated" as const,
+              driver: CODEX_DRIVER,
+              turnItem: status === "waiting" ? item : { ...item, status, completedAt: now },
+            },
+          ],
+          (event) =>
+            ingestor.ingestNormalized({
+              providerSessionId,
+              providerInstanceId: modelSelection.instanceId,
+              threadId,
+              event,
+            }),
+          { concurrency: 1 },
+        );
+      const outcome = answer?.decision === "accept" ? "completed" : "cancelled";
+      if (answer !== null) {
+        yield* ingest("waiting");
+        // The orchestrator's response transaction; its outbox delivery never runs here.
+        const base = { threadId, nodeId, occurredAt: now };
+        yield* eventSink.write({
+          events: [
+            {
+              ...base,
+              id: yield* idAllocator.allocate.event({ threadId }),
+              type: "runtime-request.updated",
+              payload: { ...request, status: "resolved", resolvedAt: now, ...answer },
+            },
+            {
+              ...base,
+              id: yield* idAllocator.allocate.event({ threadId }),
+              type: "node.updated",
+              payload: { ...node, status: outcome, completedAt: now },
+            },
+            {
+              ...base,
+              id: yield* idAllocator.allocate.event({ threadId }),
+              type: "turn-item.updated",
+              payload: { ...item, status: outcome, completedAt: now },
+            },
+          ],
+        });
+      }
+
+      yield* ingest("cancelled");
+      const projection = yield* projectionStore.getThreadProjection(threadId);
+      const stored = projection.runtimeRequests.find((entry) => entry.id === request.id);
+      assert.equal(stored?.status, answer === null ? "cancelled" : "resolved");
+      assert.equal(stored?.decision, answer?.decision);
+      assert.equal(
+        stored?.feedback,
+        answer !== null && "feedback" in answer ? answer.feedback : undefined,
+      );
+      assert.equal(projection.nodes.find((entry) => entry.id === nodeId)?.status, outcome);
+      assert.equal(projection.turnItems.find((entry) => entry.id === item.id)?.status, outcome);
+    }),
+  );
+
   it.effect("rolls back runs a provider rewound off its active branch", () =>
     Effect.gen(function* () {
       const now = yield* DateTime.now;

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

@@ -715,7 +715,7 @@ export const layer: Layer.Layer<
             if (input.writeIfProviderThreadOwner !== undefined) {
               const ownerResult = yield* eventSink
                 .writeIfProviderThreadOwner({
-                  guardPendingUserInputCancellations: true,
+                  guardPendingRuntimeRequestCancellations: true,
                   ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
                   ...input.writeIfProviderThreadOwner,
                   events,
@@ -726,15 +726,15 @@ export const layer: Layer.Layer<
             if (input.writeIfRunCurrent === undefined) {
               return yield* eventSink
                 .write({
-                  guardPendingUserInputCancellations: true,
+                  guardPendingRuntimeRequestCancellations: true,
                   ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
                   events,
                 })
                 .pipe(retryLockTimeout, Effect.mapError(mapWriteError));
             }
             const result = yield* eventSink
               .writeIfRunCurrent({
-                guardPendingUserInputCancellations: true,
+                guardPendingRuntimeRequestCancellations: true,
                 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
                 threadId: input.threadId,
                 ...input.writeIfRunCurrent,