fix(server): withdraw a stopped turn's pending approvals

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

@@ -10,6 +10,7 @@ import {
   RuntimeRequestId,
   NodeId,
   type ProjectId,
+  ProviderRequestKind,
   ThreadId,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
@@ -291,8 +292,9 @@ const layerBase: Layer.Layer<
           );
       });
 
-    // A user can answer after terminal normalization reads the pending request.
+    // A user can answer a question or approval after terminal normalization reads it pending.
     // 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 @@ const layerBase: Layer.Layer<
         for (const event of events) {
           if (
             event.type !== "runtime-request.updated" ||
-            event.payload.kind !== "user_input" ||
+            (event.payload.kind !== "user_input" && !isApprovalRequestKind(event.payload.kind)) ||
             event.payload.status !== "cancelled"
           )
             continue;
@@ -311,7 +313,7 @@ const layerBase: Layer.Layer<
           );
           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 @@ 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)
               );

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

@@ -911,6 +911,129 @@ it.effect("SQL projection keeps background thread updates out of updatedAt", ()
   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

@@ -21,6 +21,7 @@ import type {
   OrchestrationV2ExecutionNode,
   OrchestrationV2PlanArtifact,
   OrchestrationV2PendingRuntimeRequestSummary,
+  OrchestrationV2RuntimeRequest,
   PlanId,
   OrchestrationV2ConversationMessage,
   OrchestrationV2DomainEvent,
@@ -67,6 +68,7 @@ import {
   TurnItemId,
   NodeId,
   ProviderInstanceId,
+  ProviderRequestKind,
 } from "@t3tools/contracts";
 import {
   createOrchestrationV2TurnItemVisibility,
@@ -300,11 +302,22 @@ export type ProjectionRuntimeRecoveryState = Pick<
   | "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 @@ export interface ProjectionStoreV2Shape {
   readonly getRuntimeRecoveryProjection: (
     threadId: ThreadId,
   ) => Effect.Effect<ProjectionRuntimeRecoveryState, ProjectionStoreV2Error>;
-  readonly getPendingNativeUserInputs: (
+  readonly getPendingNativeRequests: (
     threadId: ThreadId,
     providerTurnId: ProviderTurnId,
-  ) => Effect.Effect<ProjectionPendingUserInputs, ProjectionStoreV2Error>;
+    options: { readonly includeApprovals: boolean },
+  ) => Effect.Effect<ProjectionPendingNativeRequests, ProjectionStoreV2Error>;
   readonly getPlan: (
     threadId: ThreadId,
     planId: PlanId,
@@ -4647,22 +4661,23 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         ? cause
         : new ProjectionStoreReadError({ threadId, cause });
 
-    const getPendingNativeUserInputs: ProjectionStoreV2Shape["getPendingNativeUserInputs"] = (
+    const getPendingNativeRequests: ProjectionStoreV2Shape["getPendingNativeRequests"] = (
       threadId,
       providerTurnId,
+      { includeApprovals },
     ) =>
       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");
+            )(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 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
             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),
+                (item) =>
+                  (item.type === "user_input_request" || item.type === "approval_request") &&
+                  requestIds.has(item.requestId),
               ),
             };
           }),
@@ -6503,7 +6520,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       getCheckpointRollbackContext,
       getRunMessage,
       canStartQueuedRun,
-      getPendingNativeUserInputs,
+      getPendingNativeRequests,
       hasUnpairedRunInterruptRequest,
       isNodeStored,
       getMessageCount,
@@ -6743,28 +6760,31 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
           .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) =>
-              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),
-            ),
-          };
-        },
-      ),
+      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") &&
+              requestIds.has(item.requestId),
+          ),
+        };
+      }),
       getPlan: Effect.fn("ProjectionStore.getPlan")(function* (threadId, planId) {
         const projection = (yield* Ref.get(replayState)).projections.get(threadId);
         if (projection === undefined)