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

apps/server/src/mcp/threadAccess.ts

@@ -11,7 +11,9 @@ import {
 } from "@t3tools/contracts";
 import * as SecureRandom from "effect/crypto/SecureRandom";
 import * as Effect from "effect/Effect";
+import * as Schema from "effect/Schema";
 
+import { EventSinkRuntimeRequestNotPendingError } from "../orchestration-v2/EventSink.ts";
 import type { OrchestratorV2Error } from "../orchestration-v2/Orchestrator.ts";
 import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
 import { resolveInteractionMode, resolveRuntimeMode } from "./delegatedTaskModes.ts";
@@ -44,21 +46,26 @@ export const unavailable = () =>
     message: "The operation could not be completed.",
   });
 
+// Its message is built only from the request id and status, so it is safe to expose.
+const isRuntimeRequestNotPending = Schema.is(EventSinkRuntimeRequestNotPendingError);
+
 export const dispatchFailure = (
   error: OrchestratorV2Error | ThreadManagement.ThreadManagementProviderInstanceDisabledError,
 ) =>
   error._tag === "ThreadManagementProviderInstanceDisabledError" ||
   error._tag === "OrchestratorRuntimeModeAboveCeilingError"
     ? new OrchestratorMcpFailure({ code: "invalid_request", message: error.message })
-    : (error._tag === "OrchestratorDispatchError" ||
-          error._tag === "OrchestratorCommandRejectedError") &&
-        typeof error.cause === "string" &&
-        error.cause.length > 0
-      ? new OrchestratorMcpFailure({
-          code: "orchestration_error",
-          message: Array.from(error.cause).slice(0, 1000).join(""),
-        })
-      : unavailable();
+    : error._tag === "OrchestratorDispatchError" && isRuntimeRequestNotPending(error.cause)
+      ? new OrchestratorMcpFailure({ code: "orchestration_error", message: error.cause.message })
+      : (error._tag === "OrchestratorDispatchError" ||
+            error._tag === "OrchestratorCommandRejectedError") &&
+          typeof error.cause === "string" &&
+          error.cause.length > 0
+        ? new OrchestratorMcpFailure({
+            code: "orchestration_error",
+            message: Array.from(error.cause).slice(0, 1000).join(""),
+          })
+        : unavailable();
 
 export interface CallerLimits {
   readonly runtimeMode: RuntimeMode;

apps/server/src/mcp/toolkits/core.test.ts

@@ -36,6 +36,7 @@ import {
   OrchestratorDispatchError,
   OrchestratorProjectionError,
 } from "../../orchestration-v2/Orchestrator.ts";
+import { EventSinkRuntimeRequestNotPendingError } from "../../orchestration-v2/EventSink.ts";
 
 import * as ServerSecretStore from "../../auth/ServerSecretStore.ts";
 import * as ServerConfig from "../../config.ts";
@@ -227,6 +228,17 @@ it("bounds public command rejections and redacts internal dispatch causes", () =
       new OrchestratorCommandRejectedError({ ...command, cause: "Run is not queued." }),
     ).message,
   ).toBe("Run is not queued.");
+  expect(
+    dispatchFailure(
+      new OrchestratorDispatchError({
+        ...command,
+        cause: new EventSinkRuntimeRequestNotPendingError({
+          requestId: RuntimeRequestId.make("mcp-core-request"),
+          status: "cancelled",
+        }),
+      }),
+    ).message,
+  ).toBe("Runtime request mcp-core-request is cancelled.");
   for (const cause of [
     undefined,
     "",

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

@@ -10,6 +10,7 @@ import {
   RuntimeRequestId,
   NodeId,
   type ProjectId,
+  ProviderRequestKind,
   ThreadId,
 } from "@t3tools/contracts";
 import * as ProviderCommitBatch from "./ProviderCommitBatch.ts";
@@ -65,7 +66,27 @@ export class EventSinkStreamError extends Schema.TaggedError<EventSinkStreamErro
   }
 }
 
-export const EventSinkV2Error = Schema.Union([EventSinkWriteError, EventSinkStreamError]);
+// A respond command plans from a pending read outside the provider ingestor's
+// writes, so terminal normalization can cancel the request before it commits.
+export class EventSinkRuntimeRequestNotPendingError extends Schema.TaggedError<EventSinkRuntimeRequestNotPendingError>()(
+  "EventSinkRuntimeRequestNotPendingError",
+  {
+    requestId: RuntimeRequestId,
+    status: Schema.String,
+  },
+) {
+  override get message(): string {
+    return `Runtime request ${this.requestId} is ${this.status}.`;
+  }
+}
+
+const isRuntimeRequestNotPending = Schema.is(EventSinkRuntimeRequestNotPendingError);
+
+export const EventSinkV2Error = Schema.Union([
+  EventSinkWriteError,
+  EventSinkStreamError,
+  EventSinkRuntimeRequestNotPendingError,
+]);
 export type EventSinkV2Error = typeof EventSinkV2Error.Type;
 
 interface ProviderBatchPublication {
@@ -311,8 +332,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>) {
@@ -321,7 +343,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;
@@ -331,7 +353,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"
           ) {
@@ -347,7 +369,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)
               );
@@ -680,6 +703,24 @@ const layerBase: Layer.Layer<
             const existing = yield* existingCommandResult(input.commandId);
             return { ...existing, committed: false as const, cancelledEffectIds: [] };
           }
+          for (const event of input.events) {
+            if (event.type !== "runtime-request.updated" || event.payload.status !== "resolved")
+              continue;
+            const current = yield* projectionStore.getRuntimeRequest(
+              event.threadId,
+              event.payload.id,
+            );
+            if (
+              current?.status !== "pending" ||
+              current.kind !== event.payload.kind ||
+              current.providerTurnId !== event.payload.providerTurnId
+            ) {
+              return yield* new EventSinkRuntimeRequestNotPendingError({
+                requestId: event.payload.id,
+                status: current?.status ?? "missing",
+              });
+            }
+          }
 
           const runId = input.updateDelegatedCompletionOnlyRunId;
           const normalized = yield* normalizeEvents(
@@ -1048,13 +1089,14 @@ const layerBase: Layer.Layer<
         ),
       commitCommand: (input) =>
         commitCommandEffect(input).pipe(
-          Effect.mapError(
-            (cause) =>
-              new EventSinkWriteError({
-                commandId: input.commandId,
-                eventCount: input.events.length,
-                cause,
-              }),
+          Effect.mapError((cause) =>
+            isRuntimeRequestNotPending(cause)
+              ? cause
+              : new EventSinkWriteError({
+                  commandId: input.commandId,
+                  eventCount: input.events.length,
+                  cause,
+                }),
           ),
         ),
       commitRejectedCommand: (input) =>

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

@@ -1849,6 +1849,125 @@ 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("SQL pending native requests: approvals are opt-in and stay with their turn", () =>
+  pendingNativeRequestsStayWithTheirTurn.pipe(Effect.provide(layerTest)),
+);
+
 const runlessNativeSubagentNeedsRecovery = Effect.gen(function* () {
   const store = yield* ProjectionStore.ProjectionStoreV2;
   const threadId = yield* addRolledBackRecoveryCandidate("runless-native-recovery");