fix(server): withdraw approvals when provider turns fail

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

@@ -103,7 +103,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>;
@@ -117,13 +117,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;
@@ -145,7 +145,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;
@@ -342,25 +342,20 @@ const layerBase: Layer.Layer<
 
     // A user can answer after terminal normalization reads the pending 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) {
-          if (
-            event.type !== "runtime-request.updated" ||
-            event.payload.kind !== "user_input" ||
-            event.payload.status !== "cancelled"
-          )
+          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"
           ) {
@@ -376,7 +371,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)
               );
@@ -497,8 +493,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({
@@ -574,8 +570,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({
@@ -639,8 +635,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/ProjectionStore.ts

@@ -330,7 +330,7 @@ export type ProjectionRuntimeRecoveryState = Pick<
   | "turnItems"
 >;
 
-export type ProjectionPendingUserInputs = Pick<
+export type ProjectionPendingRuntimeRequests = Pick<
   OrchestrationV2ThreadProjection,
   "runtimeRequests" | "nodes" | "turnItems"
 >;
@@ -684,10 +684,11 @@ export interface ProjectionStoreV2Shape {
   readonly getRuntimeRecoveryProjection: (
     threadId: ThreadId,
   ) => Effect.Effect<ProjectionRuntimeRecoveryState, ProjectionStoreV2Error>;
-  readonly getPendingNativeUserInputs: (
+  readonly getPendingNativeRuntimeRequests: (
     threadId: ThreadId,
     providerTurnId: ProviderTurnId,
-  ) => Effect.Effect<ProjectionPendingUserInputs, ProjectionStoreV2Error>;
+    includeApprovals?: boolean,
+  ) => Effect.Effect<ProjectionPendingRuntimeRequests, ProjectionStoreV2Error>;
   readonly getPlan: (
     threadId: ThreadId,
     planId: PlanId,
@@ -5587,46 +5588,51 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         ? cause
         : new ProjectionStoreReadError({ threadId, cause });
 
-    const getPendingNativeUserInputs: ProjectionStoreV2Shape["getPendingNativeUserInputs"] = (
-      threadId,
-      providerTurnId,
-    ) =>
-      sql
-        .withTransaction(
-          Effect.gen(function* () {
-            const rows = yield* sql<PayloadRow>`
+    const getPendingNativeRuntimeRequests: ProjectionStoreV2Shape["getPendingNativeRuntimeRequests"] =
+      (threadId, providerTurnId, includeApprovals = false) =>
+        sql
+          .withTransaction(
+            Effect.gen(function* () {
+              const rows = yield* sql<PayloadRow>`
           SELECT payload_json FROM orchestration_v2_projection_runtime_requests
           WHERE thread_id = ${threadId} AND provider_turn_id = ${providerTurnId}
-            AND kind = 'user_input' AND status = 'pending'
+            AND status = 'pending'
         `;
-            const runtimeRequests = (yield* decodeRows(
-              decodeRuntimeRequestPayload,
-              threadId,
-            )(rows)).filter((request) => request.responseCapability.type !== "message");
-            if (runtimeRequests.length === 0) return { runtimeRequests, nodes: [], turnItems: [] };
-            const nodeRows = yield* sql<PayloadRow>`
+              const runtimeRequests = (yield* decodeRows(
+                decodeRuntimeRequestPayload,
+                threadId,
+              )(rows)).filter(
+                (request) =>
+                  request.responseCapability.type !== "message" &&
+                  (includeApprovals || request.kind === "user_input"),
+              );
+              if (runtimeRequests.length === 0)
+                return { runtimeRequests, nodes: [], turnItems: [] };
+              const nodeRows = yield* sql<PayloadRow>`
           SELECT payload_json FROM orchestration_v2_projection_nodes
           WHERE thread_id = ${threadId} AND ${sql.in(
             "node_id",
             runtimeRequests.map((request) => request.nodeId),
           )}
         `;
-            const itemRows = yield* sql<PayloadRow>`
+              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 = 'user_input_request'
+            AND type IN ('user_input_request', 'approval_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),
-              ),
-            };
-          }),
-        )
-        .pipe(Effect.mapError(controlReadError(threadId)));
+              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" || item.type === "approval_request") &&
+                    requestIds.has(item.requestId),
+                ),
+              };
+            }),
+          )
+          .pipe(Effect.mapError(controlReadError(threadId)));
 
     const getPlan: ProjectionStoreV2Shape["getPlan"] = (threadId, planId) =>
       sql
@@ -9127,7 +9133,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       getRun: readOnReader(getRun),
       canStartQueuedRun: readOnReader(canStartQueuedRun),
       getHeldQueuedRuns: readOnReader(getHeldQueuedRuns),
-      getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs),
+      getPendingNativeRuntimeRequests: readOnReader(getPendingNativeRuntimeRequests),
       hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest),
       isNodeStored: readOnReader(isNodeStored),
       getMessageCount: readOnReader(getMessageCount),

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

@@ -1209,8 +1209,8 @@ layer("ProviderEventIngestorV2", (it) => {
     }),
   );
 
-  it.effect.each(["interrupted", "control"] as const)(
-    "dismisses only native questions when a provider turn ends with %s",
+  it.effect.each(["interrupted", "failed", "completed", "cancelled", "control"] as const)(
+    "withdraws approvals only when explicitly abandoned (%s)",
     (terminal) =>
       Effect.gen(function* () {
         const now = yield* DateTime.now;
@@ -1325,6 +1325,7 @@ layer("ProviderEventIngestorV2", (it) => {
           providerSessionId,
           providerInstanceId: modelSelection.instanceId,
           threadId,
+          withdrawApprovals: terminal === "control",
           event:
             terminal === "control"
               ? {
@@ -1356,7 +1357,10 @@ layer("ProviderEventIngestorV2", (it) => {
         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" && terminal === "control");
           const request = projection.runtimeRequests.find(
             (item) => item.id === fixture.request.id,
           )!;
@@ -1377,10 +1381,10 @@ layer("ProviderEventIngestorV2", (it) => {
         }
         assert.equal(
           stored.filter((entry) => entry.event.type === "runtime-request.updated").length,
-          2,
+          terminal === "control" ? 3 : 2,
         );
         assert.isEmpty(
-          (yield* projectionStore.getPendingNativeUserInputs(threadId, providerTurnId))
+          (yield* projectionStore.getPendingNativeRuntimeRequests(threadId, providerTurnId))
             .runtimeRequests,
         );
         const replayed = yield* eventStore.read({ threadId }).pipe(Stream.runCollect);
@@ -1390,16 +1394,16 @@ layer("ProviderEventIngestorV2", (it) => {
               entry.event.type === "runtime-request.updated" &&
               entry.event.payload.status === "cancelled",
           ).length,
-          2,
+          terminal === "control" ? 3 : 2,
         );
         const repeated = yield* ingestor.ingestNormalized(input);
         assert.isFalse(repeated.some((entry) => entry.event.type === "runtime-request.updated"));
       }),
   );
 
-  it.effect(
-    "preserves an answer committed after terminal normalization reads a pending question",
-    () =>
+  it.effect.each(["question", "approval"] as const)(
+    "preserves an answer committed after terminal normalization reads a pending %s",
+    (kind) =>
       Effect.gen(function* () {
         const eventSink = yield* EventSink.EventSinkV2;
         const eventStore = yield* EventStore.EventStoreV2;
@@ -1426,7 +1430,7 @@ layer("ProviderEventIngestorV2", (it) => {
           nodeId,
           providerTurnId,
           nativeRequestRef: null,
-          kind: "user_input",
+          kind: kind === "approval" ? "command" : "user_input",
           status: "pending",
           responseCapability: { type: "live", providerSessionId },
           createdAt: now,
@@ -1438,7 +1442,7 @@ layer("ProviderEventIngestorV2", (it) => {
           runId: null,
           parentNodeId: null,
           rootNodeId: nodeId,
-          kind: "user_input_request",
+          kind: kind === "approval" ? "approval_request" : "user_input_request",
           status: "waiting",
           countsForRun: false,
           providerThreadId,
@@ -1464,9 +1468,9 @@ layer("ProviderEventIngestorV2", (it) => {
           startedAt: now,
           completedAt: null,
           updatedAt: now,
-          type: "user_input_request",
-          requestId: request.id,
-          questions: [],
+          ...(kind === "approval"
+            ? { type: "approval_request", requestId: request.id, requestKind: "command" }
+            : { type: "user_input_request", requestId: request.id, questions: [] }),
         };
         const seedEvents: Array<OrchestrationV2DomainEvent> = [threadEvent];
         for (const payload of [
@@ -1500,6 +1504,7 @@ layer("ProviderEventIngestorV2", (it) => {
             providerSessionId,
             providerInstanceId: modelSelection.instanceId,
             threadId,
+            withdrawApprovals: kind === "approval",
             event: {
               type: "provider_turn.updated",
               driver: CODEX_DRIVER,
@@ -1519,11 +1524,12 @@ layer("ProviderEventIngestorV2", (it) => {
           .pipe(Effect.forkScoped);
         yield* Deferred.await(normalized);
         const answers = { decision: "Use the existing workspace" };
+        const response = kind === "approval" ? { decision: "accept" as const } : { answers };
         const responseEvents: Array<OrchestrationV2DomainEvent> = [];
         for (const payload of [
           {
             type: "runtime-request.updated" as const,
-            payload: { ...request, status: "resolved" as const, answers, resolvedAt: now },
+            payload: { ...request, status: "resolved" as const, ...response, resolvedAt: now },
           },
           {
             type: "node.updated" as const,
@@ -1550,7 +1556,8 @@ layer("ProviderEventIngestorV2", (it) => {
         const projection = yield* projections.getThreadProjection(threadId);
         const answered = projection.runtimeRequests.find((entry) => entry.id === request.id)!;
         assert.equal(answered.status, "resolved");
-        assert.deepEqual(answered.answers, answers);
+        if (kind === "approval") assert.equal(answered.decision, "accept");
+        else assert.deepEqual(answered.answers, answers);
         assert.equal(projection.nodes.find((entry) => entry.id === nodeId)!.status, "completed");
         assert.equal(
           projection.turnItems.find((entry) => entry.id === item.id)!.status,