Support Codex native async and synchronous user questions

apps/app/src/components/thread/pending-interactions/ThreadPendingInteractionBanner.stories.tsx

@@ -1,6 +1,6 @@
 import type {
   PendingInteraction,
-  ProviderPendingInteraction,
+  ApprovalPendingInteraction,
 } from "@bb/domain";
 import { ThreadPendingInteractionBanner } from "@/components/thread/pending-interactions/ThreadPendingInteractionBanner";
 import { ThreadPromptContextBanner } from "@/components/promptbox/banner/ThreadPromptContextBanner";
@@ -15,7 +15,7 @@ function PromptStage({ children }: { children: React.ReactNode }) {
 }
 
 function basePendingInteraction(): Omit<
-  ProviderPendingInteraction,
+  ApprovalPendingInteraction,
   "payload" | "resolution"
 > {
   return {

apps/app/src/hooks/queries/thread-queries.ts

@@ -1047,11 +1047,12 @@ export function getLatestPendingInteraction(
   }
 
   const [firstInteraction, ...restInteractions] = interactions;
-  return restInteractions.reduce<PendingInteraction>(
-    (latest, interaction) =>
-      interaction.createdAt > latest.createdAt ? interaction : latest,
-    firstInteraction,
-  );
+  return restInteractions.reduce<PendingInteraction>((latest, interaction) => {
+    if ((latest.turnId === null) !== (interaction.turnId === null)) {
+      return interaction.turnId !== null ? interaction : latest;
+    }
+    return interaction.createdAt > latest.createdAt ? interaction : latest;
+  }, firstInteraction);
 }
 
 export function isPendingInteractionStateUnknown(

apps/host-daemon/src/app.test.ts

@@ -1086,6 +1086,7 @@ describe("createHostDaemonApp", () => {
       expect(payload).toEqual({
         sessionId: "session-app-test",
         providerId: "codex",
+        providerRequestId: null,
         threadIds: [request.threadId],
         reason: 'Provider "codex" exited while awaiting user interaction',
       });

apps/host-daemon/src/app.ts

@@ -141,6 +141,7 @@ export interface HostDaemonApp {
 }
 
 interface PendingInteractiveInterruptRequest {
+  providerRequestId?: string;
   providerId: string;
   reason: string;
   threadIds: readonly string[];
@@ -300,6 +301,7 @@ export async function createHostDaemonApp(
     return [
       request.providerId,
       request.reason,
+      request.providerRequestId ?? "",
       [...request.threadIds].sort().join(","),
     ].join("|");
   }
@@ -391,6 +393,13 @@ export async function createHostDaemonApp(
   });
 
   const interactiveRequestRegistry = new InteractiveRequestRegistry({
+    onCancellation: (request) =>
+      enqueueInteractiveInterrupt({
+        providerId: request.providerId,
+        providerRequestId: request.providerRequestId,
+        threadIds: [request.threadId],
+        reason: "Provider request was closed",
+      }),
     registerRequest: (request) =>
       runSessionRequest({
         source: "registerInteractiveRequest",
@@ -551,9 +560,12 @@ export async function createHostDaemonApp(
         throw error;
       }
     },
-    onInteractiveRequest: async (request) => {
+    onInteractiveRequest: async (request, signal) => {
       try {
-        return await interactiveRequestRegistry.registerAndWait(request);
+        return await interactiveRequestRegistry.registerAndWait(
+          request,
+          signal,
+        );
       } catch (error) {
         if (
           error instanceof InteractiveRequestRegistryError &&

apps/host-daemon/src/interactive-request-registry.test.ts

@@ -203,3 +203,41 @@ describe("InteractiveRequestRegistry", () => {
     await expect(pending).rejects.toThrow("Provider exited");
   });
 });
+
+describe("provider request cancellation", () => {
+  it.each([true, false])(
+    "closes the exact request when abort occurs before registration=%s",
+    async (beforeRegistration) => {
+      const registered =
+        createDeferredPromise<HostDaemonInteractiveRequestResponse>();
+      const cancelled: string[] = [];
+      const registry = new InteractiveRequestRegistry({
+        registerRequest: () => registered.promise,
+        onCancellation: (request) => cancelled.push(request.providerRequestId),
+      });
+      const request = createCommandApprovalRequest();
+      const controller = new AbortController();
+      const pending = registry.registerAndWait(request, controller.signal);
+      const rejected = expect(pending).rejects.toThrow(
+        "Provider request was closed",
+      );
+      if (beforeRegistration) controller.abort();
+      registered.resolve({
+        outcome: "created",
+        interactionId: "pint_cancelled",
+        status: "pending",
+      });
+      await Promise.resolve();
+      if (!beforeRegistration) controller.abort();
+      await rejected;
+      expect(cancelled).toEqual([request.providerRequestId]);
+      expect(() =>
+        registry.resolve({
+          ...request,
+          interactionId: "pint_cancelled",
+          resolution: createCommandApprovalResolution(),
+        }),
+      ).toThrow("no longer awaiting");
+    },
+  );
+});

apps/host-daemon/src/interactive-request-registry.ts

@@ -22,6 +22,7 @@ interface InteractiveRequestRegistrationFailure {
 }
 
 interface InteractiveRequestRegistryOptions {
+  onCancellation?: (request: PendingInteractionCreate) => void;
   onRegistrationFailure?: (
     failure: InteractiveRequestRegistrationFailure,
   ) => void;
@@ -105,7 +106,9 @@ export class InteractiveRequestRegistry {
 
   async registerAndWait(
     request: PendingInteractionCreate,
+    signal?: AbortSignal,
   ): Promise<PendingInteractionResolution> {
+    signal?.throwIfAborted();
     const key = buildInteractiveRequestKey(request);
     const existing = this.pendingEntries.get(key);
     if (existing) {
@@ -122,11 +125,23 @@ export class InteractiveRequestRegistry {
         rejectEntry = reject;
       },
     );
+    const cancel = () => {
+      if (this.pendingEntries.get(key) !== entry) return;
+      this.pendingEntries.delete(key);
+      this.options.onCancellation?.(request);
+      entry.reject(new Error("Provider request was closed"));
+    };
     const entry: PendingInteractiveRequestEntry = {
       interactionId: null,
       promise,
-      reject: (error) => rejectEntry(error),
-      resolve: (resolution) => resolveEntry(resolution),
+      reject: (error) => {
+        signal?.removeEventListener("abort", cancel);
+        rejectEntry(error);
+      },
+      resolve: (resolution) => {
+        signal?.removeEventListener("abort", cancel);
+        resolveEntry(resolution);
+      },
       request,
     };
     this.pendingEntries.set(key, entry);
@@ -145,6 +160,10 @@ export class InteractiveRequestRegistry {
       }
 
       entry.interactionId = response.interactionId;
+      if (this.pendingEntries.get(key) === entry) {
+        signal?.addEventListener("abort", cancel, { once: true });
+        if (signal?.aborted) cancel();
+      }
       if (response.status !== "pending" && response.status !== "resolving") {
         this.pendingEntries.delete(key);
         entry.reject(

apps/host-daemon/src/runtime-manager.ts

@@ -186,6 +186,7 @@ export interface RuntimeManagerOptions {
   }) => void;
   onInteractiveRequest?: (
     request: PendingInteractionCreate,
+    signal?: AbortSignal,
   ) => Promise<PendingInteractionResolution>;
   onToolCall?: AgentRuntimeOptions["onToolCall"];
   onStderr?: AgentRuntimeOptions["onStderr"];

apps/host-daemon/src/server-client.ts

@@ -217,6 +217,7 @@ export interface ServerClient {
     request: PendingInteractionCreate,
   ): Promise<HostDaemonInteractiveRequestResponse>;
   interruptInteractiveRequests(args: {
+    providerRequestId?: string;
     providerId: string;
     reason: string;
     threadIds: readonly string[];
@@ -669,6 +670,7 @@ export function createServerClient(
         providerId: args.providerId,
         threadIds: [...args.threadIds],
         reason: args.reason,
+        providerRequestId: args.providerRequestId ?? null,
       };
       const response = await fetchFn(
         buildInternalUrl("/session/interactive-request/interrupt"),

apps/server/src/internal/events.ts

@@ -197,7 +197,8 @@ interface HasThreadCommandFailureSystemErrorForTurnArgs {
   turnId: string;
 }
 
-interface HasThreadStopBeforeTurnStartedArgs {
+interface HasThreadStopForTurnArgs {
+  beforeTurnStartedOnly: boolean;
   threadId: string;
   turnId: string;
 }
@@ -384,15 +385,52 @@ async function applyEventEffects(
   for (const entry of events) {
     try {
       const event = entry.event;
+      if (
+        event.type === "item/completed" &&
+        event.item.type === "agentMessage" &&
+        event.item.asyncQuestion
+      ) {
+        const thread = getThread(deps.db, entry.threadId);
+        if (!thread || thread.deletedAt !== null) continue;
+        if (
+          hasThreadStopForTurn(deps, {
+            threadId: entry.threadId,
+            turnId: requireThreadEventScopeTurnId({
+              type: event.type,
+              scope: event.scope,
+            }),
+            beforeTurnStartedOnly: false,
+          })
+        )
+          continue;
+        const registered = deps.pendingInteractions.registerPendingInteraction({
+          interaction: {
+            threadId: entry.threadId,
+            turnId: null,
+            providerId: thread.providerId,
+            providerThreadId: event.providerThreadId,
+            providerRequestId: `message:${event.item.asyncQuestion.id}`,
+            payload: event.item.asyncQuestion.payload,
+          },
+        });
+        if (registered.outcome === "rejected") {
+          deps.logger.warn(
+            { threadId: entry.threadId, reason: registered.reason },
+            "Could not register async question",
+          );
+        }
+        continue;
+      }
       if (event.type === "turn/started") {
         const turnId = requireThreadEventScopeTurnId({
           type: event.type,
           scope: event.scope,
         });
         if (
-          hasThreadStopBeforeTurnStarted(deps, {
+          hasThreadStopForTurn(deps, {
             threadId: entry.threadId,
             turnId,
+            beforeTurnStartedOnly: true,
           })
         ) {
           continue;
@@ -421,9 +459,10 @@ async function applyEventEffects(
         });
         if (
           event.status !== "interrupted" &&
-          hasThreadStopBeforeTurnStarted(deps, {
+          hasThreadStopForTurn(deps, {
             threadId: entry.threadId,
             turnId,
+            beforeTurnStartedOnly: true,
           })
         ) {
           continue;
@@ -474,6 +513,7 @@ async function applyEventEffects(
         }
         deps.pendingInteractions.interruptPendingInteractionsForThreadIds({
           threadIds: [entry.threadId],
+          preserveAsyncQuestions: true,
           reason:
             "Provider process exited while awaiting user interaction; retry the thread to continue",
         });
@@ -595,9 +635,9 @@ function hasThreadCommandFailureSystemErrorForTurn(
   );
 }
 
-function hasThreadStopBeforeTurnStarted(
+function hasThreadStopForTurn(
   deps: Pick<AppDeps, "db">,
-  args: HasThreadStopBeforeTurnStartedArgs,
+  args: HasThreadStopForTurnArgs,
 ): boolean {
   const turnStarted = deps.db
     .select({ sequence: storedEvents.sequence })
@@ -639,7 +679,9 @@ function hasThreadStopBeforeTurnStarted(
           eq(storedEvents.threadId, args.threadId),
           eq(storedEvents.type, "system/thread/interrupted"),
           gt(storedEvents.sequence, lowerSequence),
-          lt(storedEvents.sequence, turnStarted.sequence),
+          args.beforeTurnStartedOnly
+            ? lt(storedEvents.sequence, turnStarted.sequence)
+            : sql`json_extract(${storedEvents.data}, '$.reason') = 'manual-stop' AND json_extract(${storedEvents.data}, '$.cause') IS NULL`,
         ),
       )
       .limit(1)

apps/server/src/internal/interactive-requests.ts

@@ -244,6 +244,7 @@ export function registerInternalInteractiveRequestRoutes(
       const interrupted =
         deps.pendingInteractions.interruptPendingInteractionsForThreads({
           providerId: payload.providerId,
+          providerRequestId: payload.providerRequestId,
           threadIds: interruptibleThreadIds,
           reason: payload.reason,
         });

apps/server/src/internal/session-owner-side-effects.ts

@@ -300,6 +300,7 @@ function interruptPendingInteractionsForHostThreads(
 ): void {
   deps.pendingInteractions.interruptPendingInteractionsForThreadIds({
     threadIds: listHostThreadIds(deps.db, { hostId: args.hostId }),
+    preserveAsyncQuestions: true,
     reason: args.reason,
   });
 }

apps/server/src/routes/threads/interactions.ts

@@ -1,4 +1,8 @@
-import { PLUGIN_INTERACTION_MAX_PAYLOAD_BYTES } from "@bb/domain";
+import {
+  isUserQuestionPendingInteraction,
+  isUserQuestionPendingInteractionResolution,
+  PLUGIN_INTERACTION_MAX_PAYLOAD_BYTES,
+} from "@bb/domain";
 import {
   publicApiRoutes,
   typedRoutes,
@@ -7,6 +11,7 @@ import {
 import type { Hono } from "hono";
 import { z } from "zod";
 import type { AppDeps } from "../../types.js";
+import { resolveAsyncUserQuestion } from "../../services/interactions/async-user-questions.js";
 import { ApiError } from "../../errors.js";
 import { requirePublicThread } from "../../services/lib/entity-lookup.js";
 
@@ -55,14 +60,32 @@ export function registerThreadInteractionRoutes(
     );
   });
 
-  post(routes.resolveInteraction, (context, payload) => {
+  post(routes.resolveInteraction, async (context, payload) => {
     const thread = requirePublicThread(deps.db, context.req.param("id"));
+    const interactionId = parsePendingInteractionId(
+      context.req.param("interactionId"),
+    );
+    const interaction = deps.pendingInteractions.getThreadInteraction({
+      threadId: thread.id,
+      interactionId,
+    });
+    if (
+      isUserQuestionPendingInteraction(interaction) &&
+      interaction.turnId === null &&
+      isUserQuestionPendingInteractionResolution(payload)
+    ) {
+      return context.json(
+        await resolveAsyncUserQuestion(deps, {
+          thread,
+          interaction,
+          resolution: payload,
+        }),
+      );
+    }
     return context.json(
       deps.pendingInteractions.resolvePendingInteraction({
         threadId: thread.id,
-        interactionId: parsePendingInteractionId(
-          context.req.param("interactionId"),
-        ),
+        interactionId,
         resolution: payload,
       }),
     );