From 73adf537a7bf44517a63e782051ce7f5663535f8 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Fri, 18 Sep 2026 11:41:44 -0700 Subject: [PATCH] Support Codex native async and synchronous user questions --- ...ThreadPendingInteractionBanner.stories.tsx | 4 +- apps/app/src/hooks/queries/thread-queries.ts | 11 +- apps/host-daemon/src/app.test.ts | 1 + apps/host-daemon/src/app.ts | 16 +- .../src/interactive-request-registry.test.ts | 38 +++++ .../src/interactive-request-registry.ts | 23 ++- apps/host-daemon/src/runtime-manager.ts | 1 + apps/host-daemon/src/server-client.ts | 2 + apps/server/src/internal/events.ts | 54 +++++- .../src/internal/interactive-requests.ts | 1 + .../internal/session-owner-side-effects.ts | 1 + .../server/src/routes/threads/interactions.ts | 33 +++- .../interactions/async-user-questions.ts | 75 +++++++++ .../interactions/pending-interactions.ts | 40 +++-- .../internal-event-append-ownership.test.ts | 118 +++++++++++++ .../public/native-async-questions.test.ts | 157 ++++++++++++++++++ .../src/permission-matrix.test.ts | 4 +- .../src/runtime-provider-requests.ts | 32 +++- .../src/runtime-tool-cancellation.test.ts | 6 +- .../src/runtime.interactive-requests.test.ts | 5 +- .../src/runtime.tool-calls.test.ts | 6 +- packages/agent-runtime/src/runtime.ts | 17 +- packages/agent-runtime/src/types.ts | 1 + packages/db/src/data/pending-interactions.ts | 12 +- packages/domain/src/pending-interactions.ts | 1 + packages/domain/src/provider-event.ts | 7 + packages/host-daemon-contract/src/protocol.ts | 2 +- packages/host-daemon-contract/src/session.ts | 1 + .../src/assembler/delta-assembler.ts | 18 +- .../src/thread-delta.ts | 12 +- .../src/assistant-event-projection.ts | 7 + .../test/build-thread-timeline.test.ts | 44 +++++ .../skills/codex-provider/SKILL.md | 9 + .../src/bridge/app-server-connection.ts | 6 + .../bridge/bridge.native-questions.test.ts | 136 +++++++++++++++ plugins/provider-codex/src/bridge/bridge.ts | 76 ++++++++- .../src/bridge/fake-codex-app-server.mjs | 16 +- .../src/delta-translation.test.ts | 59 +++++++ .../provider-codex/src/delta-translation.ts | 36 +++- .../src/interactive-requests.test.ts | 101 +++++++++++ .../src/interactive-requests.ts | 52 ++++++ plugins/provider-codex/src/schemas.ts | 32 ++++ .../provider-codex/src/session-params.test.ts | 4 +- plugins/provider-codex/src/session-params.ts | 2 +- 44 files changed, 1199 insertions(+), 80 deletions(-) create mode 100644 apps/server/src/services/interactions/async-user-questions.ts create mode 100644 apps/server/test/public/native-async-questions.test.ts create mode 100644 plugins/provider-codex/src/bridge/bridge.native-questions.test.ts diff --git a/apps/app/src/components/thread/pending-interactions/ThreadPendingInteractionBanner.stories.tsx b/apps/app/src/components/thread/pending-interactions/ThreadPendingInteractionBanner.stories.tsx index b4e5eda182b..1abc2bd15a2 100644 --- a/apps/app/src/components/thread/pending-interactions/ThreadPendingInteractionBanner.stories.tsx +++ b/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 { diff --git a/apps/app/src/hooks/queries/thread-queries.ts b/apps/app/src/hooks/queries/thread-queries.ts index 0922143f951..6bda73d626c 100644 --- a/apps/app/src/hooks/queries/thread-queries.ts +++ b/apps/app/src/hooks/queries/thread-queries.ts @@ -1047,11 +1047,12 @@ export function getLatestPendingInteraction( } const [firstInteraction, ...restInteractions] = interactions; - return restInteractions.reduce( - (latest, interaction) => - interaction.createdAt > latest.createdAt ? interaction : latest, - firstInteraction, - ); + return restInteractions.reduce((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( diff --git a/apps/host-daemon/src/app.test.ts b/apps/host-daemon/src/app.test.ts index a2a8ce6e7f4..d74479e22c7 100644 --- a/apps/host-daemon/src/app.test.ts +++ b/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', }); diff --git a/apps/host-daemon/src/app.ts b/apps/host-daemon/src/app.ts index 7d25061bb5e..d908d2e49e2 100644 --- a/apps/host-daemon/src/app.ts +++ b/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 && diff --git a/apps/host-daemon/src/interactive-request-registry.test.ts b/apps/host-daemon/src/interactive-request-registry.test.ts index 660e3a7e8c4..37ab474fa20 100644 --- a/apps/host-daemon/src/interactive-request-registry.test.ts +++ b/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(); + 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"); + }, + ); +}); diff --git a/apps/host-daemon/src/interactive-request-registry.ts b/apps/host-daemon/src/interactive-request-registry.ts index c353bc710fc..c9d51d5931c 100644 --- a/apps/host-daemon/src/interactive-request-registry.ts +++ b/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 { + 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( diff --git a/apps/host-daemon/src/runtime-manager.ts b/apps/host-daemon/src/runtime-manager.ts index 30ff47a3d90..5aa491f846c 100644 --- a/apps/host-daemon/src/runtime-manager.ts +++ b/apps/host-daemon/src/runtime-manager.ts @@ -186,6 +186,7 @@ export interface RuntimeManagerOptions { }) => void; onInteractiveRequest?: ( request: PendingInteractionCreate, + signal?: AbortSignal, ) => Promise; onToolCall?: AgentRuntimeOptions["onToolCall"]; onStderr?: AgentRuntimeOptions["onStderr"]; diff --git a/apps/host-daemon/src/server-client.ts b/apps/host-daemon/src/server-client.ts index 838817332df..cccc7768a1e 100644 --- a/apps/host-daemon/src/server-client.ts +++ b/apps/host-daemon/src/server-client.ts @@ -217,6 +217,7 @@ export interface ServerClient { request: PendingInteractionCreate, ): Promise; 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"), diff --git a/apps/server/src/internal/events.ts b/apps/server/src/internal/events.ts index 707caaf86c4..dd5a104d97f 100644 --- a/apps/server/src/internal/events.ts +++ b/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, - 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) diff --git a/apps/server/src/internal/interactive-requests.ts b/apps/server/src/internal/interactive-requests.ts index 2911a4b0717..4b52cf6dde4 100644 --- a/apps/server/src/internal/interactive-requests.ts +++ b/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, }); diff --git a/apps/server/src/internal/session-owner-side-effects.ts b/apps/server/src/internal/session-owner-side-effects.ts index d14bf0864c0..26b09f17b6b 100644 --- a/apps/server/src/internal/session-owner-side-effects.ts +++ b/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, }); } diff --git a/apps/server/src/routes/threads/interactions.ts b/apps/server/src/routes/threads/interactions.ts index 2305972ef2c..ac27b017342 100644 --- a/apps/server/src/routes/threads/interactions.ts +++ b/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, }), ); diff --git a/apps/server/src/services/interactions/async-user-questions.ts b/apps/server/src/services/interactions/async-user-questions.ts new file mode 100644 index 00000000000..be4cd5579a5 --- /dev/null +++ b/apps/server/src/services/interactions/async-user-questions.ts @@ -0,0 +1,75 @@ +import type { + Thread, + PendingInteraction, + UserQuestionPendingInteraction, + UserQuestionPendingInteractionResolution, +} from "@bb/domain"; +import type { AppDeps } from "../../types.js"; +import { ApiError } from "../../errors.js"; +import { requireThreadCommandEnvironment } from "../threads/thread-command-environment.js"; +import { sendThreadMessage } from "../threads/thread-send.js"; +import { + pendingInteractionResolutionEquals, + validatePendingInteractionResolution, +} from "./pending-interaction-validation.js"; + +export async function resolveAsyncUserQuestion( + deps: AppDeps, + args: { + thread: Thread; + interaction: UserQuestionPendingInteraction; + resolution: UserQuestionPendingInteractionResolution; + }, +): Promise { + const { interaction, resolution } = args; + if ( + interaction.status === "resolved" && + pendingInteractionResolutionEquals(interaction.resolution, resolution) + ) + return interaction; + if (interaction.status !== "pending") + throw new ApiError(409, "invalid_request", "Question is no longer pending"); + validatePendingInteractionResolution(interaction, resolution); + const text = interaction.payload.questions + .map((question) => { + const answer = resolution.answers[question.id]; + return `${question.prompt}\n${answer ? [...answer.selected, ...(answer.freeText ? [answer.freeText] : [])].join("\n") : "No answer provided"}`; + }) + .join("\n\n"); + const environment = await requireThreadCommandEnvironment(deps, { + thread: args.thread, + }); + await sendThreadMessage(deps, { + thread: args.thread, + environment, + payload: { input: [{ type: "text", text, mentions: [] }], mode: "auto" }, + trigger: "user", + beforeAppendInTransaction: ({ tx }) => { + const current = deps.pendingInteractions.getThreadInteraction({ + threadId: interaction.threadId, + interactionId: interaction.id, + }); + if (current.status !== "pending") + throw new ApiError( + 409, + "invalid_request", + "Question is no longer pending", + ); + const completed = + deps.pendingInteractions.completeResolvingInteractionInTransaction( + { db: tx, hub: deps.hub }, + { interactionId: interaction.id, resolution }, + ); + if (!completed) + throw new ApiError( + 409, + "invalid_request", + "Question is no longer pending", + ); + }, + }); + return deps.pendingInteractions.getThreadInteraction({ + threadId: interaction.threadId, + interactionId: interaction.id, + }); +} diff --git a/apps/server/src/services/interactions/pending-interactions.ts b/apps/server/src/services/interactions/pending-interactions.ts index 801bd5e14e6..d109e4ff143 100644 --- a/apps/server/src/services/interactions/pending-interactions.ts +++ b/apps/server/src/services/interactions/pending-interactions.ts @@ -26,6 +26,7 @@ import { parseExtensionKind, pluginInteractionDescriptionSchema, type JsonValue, + type UserQuestionPendingInteractionPayload, type PendingInteraction, type PluginInteractionDescription, type ThreadEventItemPresentation, @@ -80,7 +81,12 @@ type RegisterPendingInteractionResult = }; interface RegisterPendingInteractionArgs { - interaction: PendingInteractionCreate; + interaction: + | PendingInteractionCreate + | (Omit & { + turnId: null; + payload: UserQuestionPendingInteractionPayload; + }); } interface ResolvePendingInteractionArgs { @@ -229,12 +235,14 @@ interface NotifyInteractionChangedArgs { } interface InterruptPendingInteractionsForThreadsLifecycleArgs { + providerRequestId?: string | null; providerId: string; reason: string; threadIds: readonly string[]; } interface InterruptPendingInteractionsForThreadIdsLifecycleArgs { + preserveAsyncQuestions?: boolean; reason: string; threadIds: readonly string[]; } @@ -255,7 +263,7 @@ export interface PendingInteractionPluginDirectory { } function getUnsupportedPendingInteractionReason( - interaction: PendingInteractionCreate, + interaction: RegisterPendingInteractionArgs["interaction"], plugins: PendingInteractionPluginDirectory | null, ): string | null { if (isPluginExtensionInteractionRequestPayload(interaction.payload)) { @@ -400,14 +408,12 @@ export class PendingInteractionLifecycle { return interaction; } - /** - * Whether a pending interaction holds this thread's turn. A provider's - * question or approval does: the provider is blocked on it, so nothing - * else can be sent until it settles. A plugin's card has no turn and never - * blocks a send; whatever answers it later steers or starts a turn. - */ hasTurnBoundPendingThreadInteraction(threadId: string): boolean { - const active = getActivePendingInteractionForThread(this.deps.db, threadId); + const active = getActivePendingInteractionForThread( + this.deps.db, + threadId, + true, + ); return active !== null && active.turnId !== null; } @@ -469,8 +475,9 @@ export class PendingInteractionLifecycle { const pendingForThread = getActivePendingInteractionForThread( tx, interaction.threadId, + true, ); - if (pendingForThread) { + if (interaction.turnId !== null && pendingForThread) { return { outcome: "rejected" as const, reason: `Thread ${interaction.threadId} is already awaiting user interaction`, @@ -852,6 +859,7 @@ export class PendingInteractionLifecycle { return this.settleInterruptedRows( interruptPendingInteractionsForThreads(this.deps.db, { providerId: args.providerId, + providerRequestId: args.providerRequestId, threadIds: args.threadIds, statusReason: args.reason, }), @@ -864,6 +872,7 @@ export class PendingInteractionLifecycle { return this.settleInterruptedRows( interruptPendingInteractionsForThreadIds(this.deps.db, { threadIds: args.threadIds, + preserveAsyncQuestions: args.preserveAsyncQuestions, statusReason: args.reason, }), ); @@ -877,6 +886,7 @@ export class PendingInteractionLifecycle { deps, interruptPendingInteractionsForThreadIds(deps.db, { threadIds: args.threadIds, + preserveAsyncQuestions: args.preserveAsyncQuestions, statusReason: args.reason, }), ); @@ -1005,7 +1015,11 @@ export class PendingInteractionLifecycle { appendPendingInteractionTimelineEvent(this.deps, interaction); notifyInteractionChanged({ deps: this.deps, - hasPendingInteraction: false, + hasPendingInteraction: + getActivePendingInteractionForThread( + this.deps.db, + interaction.threadId, + ) !== null, threadId: interaction.threadId, }); this.notifyInteractionSettled(interaction.threadId); @@ -1018,7 +1032,9 @@ export class PendingInteractionLifecycle { appendPendingInteractionTimelineEventInTransaction(deps, interaction); notifyInteractionChanged({ deps, - hasPendingInteraction: false, + hasPendingInteraction: + getActivePendingInteractionForThread(deps.db, interaction.threadId) !== + null, threadId: interaction.threadId, }); this.notifyInteractionSettled(interaction.threadId); diff --git a/apps/server/test/internal/internal-event-append-ownership.test.ts b/apps/server/test/internal/internal-event-append-ownership.test.ts index aac45093c67..dd02e731e02 100644 --- a/apps/server/test/internal/internal-event-append-ownership.test.ts +++ b/apps/server/test/internal/internal-event-append-ownership.test.ts @@ -1243,3 +1243,121 @@ describe("interaction lifecycle records from the daemon", () => { } }); }); + +it("persists native async questions once across event replay and never reopens a stopped question", async () => { + const { harness, session, thread } = await setupEventRoute(); + try { + seedTurnStarted(harness.deps, { + threadId: thread.id, + turnId: "native-question-turn", + sequence: 1, + }); + const questionEvent: HostDaemonEventEnvelope = { + threadId: thread.id, + event: { + type: "item/completed", + threadId: thread.id, + providerThreadId: "native-provider-thread", + scope: turnScope("native-question-turn"), + item: { + type: "agentMessage", + id: "assembled-question-1", + text: "Which target?", + asyncQuestion: { + id: "native-call-1", + payload: createUserQuestionPayload(), + }, + }, + }, + }; + const post = () => + postEventBatch({ + harness, + sessionId: session.id, + events: [questionEvent], + }); + expect((await post()).status).toBe(200); + expect((await post()).status).toBe(200); + const pending = + harness.deps.pendingInteractions.listPendingThreadInteractions(thread.id); + expect(pending).toHaveLength(1); + expect(pending[0]).toMatchObject({ + turnId: null, + providerRequestId: "message:native-call-1", + status: "pending", + }); + harness.deps.pendingInteractions.interruptPendingInteractionsForThreadIds({ + threadIds: [thread.id], + reason: "thread-stopped", + }); + expect((await post()).status).toBe(200); + expect( + harness.deps.pendingInteractions.listPendingThreadInteractions(thread.id), + ).toEqual([]); + expect( + harness.deps.pendingInteractions.getThreadInteraction({ + threadId: thread.id, + interactionId: pending[0]!.id, + }).status, + ).toBe("interrupted"); + } finally { + await harness.cleanup(); + } +}); + +it.each([ + { reason: "manual-stop", expected: 0 }, + { reason: "host-daemon-restarted", expected: 1 }, +] as const)( + "handles late async questions after $reason", + async ({ reason, expected }) => { + const { harness, session, thread, environment } = await setupEventRoute(); + try { + seedTurnStarted(harness.deps, { + threadId: thread.id, + turnId: "late-question-turn", + sequence: 1, + }); + seedEvent(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + sequence: 2, + type: "system/thread/interrupted", + scope: threadScope(), + data: { reason }, + }); + const response = await postEventBatch({ + harness, + sessionId: session.id, + events: [ + { + threadId: thread.id, + event: { + type: "item/completed", + threadId: thread.id, + providerThreadId: "native-provider-thread", + scope: turnScope("late-question-turn"), + item: { + type: "agentMessage", + id: "late-question", + text: "Which target?", + asyncQuestion: { + id: "late-native-call", + payload: createUserQuestionPayload(), + }, + }, + }, + }, + ], + }); + expect(response.status).toBe(200); + expect( + harness.deps.pendingInteractions.listPendingThreadInteractions( + thread.id, + ), + ).toHaveLength(expected); + } finally { + await harness.cleanup(); + } + }, +); diff --git a/apps/server/test/public/native-async-questions.test.ts b/apps/server/test/public/native-async-questions.test.ts new file mode 100644 index 00000000000..b580e0582ae --- /dev/null +++ b/apps/server/test/public/native-async-questions.test.ts @@ -0,0 +1,157 @@ +import { describe, expect, it } from "vitest"; +import { withTestHarness } from "../helpers/test-app.js"; +import { + seedThreadFixture, + seedThreadRuntimeState, + seedTurnStarted, +} from "../helpers/seed.js"; +import { createUserQuestionPayload } from "../helpers/pending-interactions.js"; +import { readJson } from "../helpers/json.js"; +import { waitForQueuedCommand } from "../helpers/commands.js"; + +describe("native async questions", () => { + it.each(["idle", "active"] as const)( + "sends an answer as a new user message while %s, once", + async (status) => { + await withTestHarness(async (harness) => { + const { thread, environment } = seedThreadFixture(harness, { + thread: { status }, + }); + seedThreadRuntimeState(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + providerThreadId: "native-thread", + }); + if (status === "active") + seedTurnStarted(harness.deps, { + threadId: thread.id, + turnId: "native-turn", + providerThreadId: "native-thread", + }); + const request = { + threadId: thread.id, + turnId: null, + providerId: thread.providerId, + providerThreadId: "native-thread", + providerRequestId: "message:native-question", + payload: createUserQuestionPayload(), + }; + const registered = + harness.deps.pendingInteractions.registerPendingInteraction({ + interaction: request, + }); + if (registered.outcome === "rejected") + throw new Error(registered.reason); + expect( + harness.deps.pendingInteractions.registerPendingInteraction({ + interaction: request, + }).outcome, + ).toBe("existing"); + expect( + harness.deps.pendingInteractions.hasTurnBoundPendingThreadInteraction( + thread.id, + ), + ).toBe(false); + const question = request.payload.questions[0]!; + const answer = { + kind: "user_answer", + answers: { + [question.id]: { + selected: [], + freeText: "Use the staging environment", + }, + }, + }; + const url = `/api/v1/threads/${thread.id}/interactions/${registered.interaction.id}/resolve`; + const post = () => + harness.app.request(url, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(answer), + }); + const response = await post(); + expect( + response.status, + JSON.stringify(await response.clone().json()), + ).toBe(200); + await expect(readJson(response)).resolves.toMatchObject({ + status: "resolved", + resolution: answer, + }); + const command = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "turn.submit" && command.threadId === thread.id, + ); + expect(JSON.stringify(command.command)).toContain( + "Use the staging environment", + ); + expect(command.command.type).toBe("turn.submit"); + const duplicate = await post(); + expect(duplicate.status).toBe(200); + const changed = await harness.app.request(url, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + ...answer, + answers: { [question.id]: { selected: [], freeText: "Different" } }, + }), + }); + expect(changed.status).toBe(409); + }); + }, + ); + + it("preserves async questions across a disconnect but rejects answers after explicit stop", async () => { + await withTestHarness(async (harness) => { + const { thread } = seedThreadFixture(harness); + const registered = + harness.deps.pendingInteractions.registerPendingInteraction({ + interaction: { + threadId: thread.id, + turnId: null, + providerId: thread.providerId, + providerThreadId: "native-thread", + providerRequestId: "message:persisted-question", + payload: createUserQuestionPayload(), + }, + }); + if (registered.outcome === "rejected") throw new Error(registered.reason); + harness.deps.pendingInteractions.interruptPendingInteractionsForThreadIds( + { + threadIds: [thread.id], + reason: "Host disconnected", + preserveAsyncQuestions: true, + }, + ); + expect( + harness.deps.pendingInteractions.listPendingThreadInteractions( + thread.id, + ), + ).toHaveLength(1); + harness.deps.pendingInteractions.start(); + expect( + harness.deps.pendingInteractions.listPendingThreadInteractions( + thread.id, + ), + ).toHaveLength(1); + harness.deps.pendingInteractions.interruptPendingInteractionsForThreadIds( + { threadIds: [thread.id], reason: "thread-stopped" }, + ); + const response = await harness.app.request( + `/api/v1/threads/${thread.id}/interactions/${registered.interaction.id}/resolve`, + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + kind: "user_answer", + answers: { + "question-1": { selected: [], freeText: "Late answer" }, + }, + }), + }, + ); + expect(response.status).toBe(409); + }); + }); +}); diff --git a/packages/agent-runtime/src/permission-matrix.test.ts b/packages/agent-runtime/src/permission-matrix.test.ts index 24c73e01fb7..9d7ade76c79 100644 --- a/packages/agent-runtime/src/permission-matrix.test.ts +++ b/packages/agent-runtime/src/permission-matrix.test.ts @@ -1,4 +1,4 @@ -import { RuntimeToolCalls } from "./runtime-provider-requests.js"; +import { RuntimeRequestLifetimes } from "./runtime-provider-requests.js"; import { spawn, type ChildProcess } from "node:child_process"; import readline from "node:readline"; import { @@ -395,7 +395,7 @@ async function runCell( }; const rawRequest = interactionRequest(requestId, payload); handleRuntimeProviderRequest({ - toolCalls: new RuntimeToolCalls(), + requestLifetimes: new RuntimeRequestLifetimes(), getActiveTurnId: () => "turn-1", getThreadExecutionOptions: () => executionOptions, onInteractiveRequest, diff --git a/packages/agent-runtime/src/runtime-provider-requests.ts b/packages/agent-runtime/src/runtime-provider-requests.ts index b7ef2c47bd6..b78bfc8754b 100644 --- a/packages/agent-runtime/src/runtime-provider-requests.ts +++ b/packages/agent-runtime/src/runtime-provider-requests.ts @@ -20,7 +20,7 @@ import { } from "@bb/provider-bridge-protocol/bridge-kit"; import { shouldAutoDenyInteractiveRequest } from "@bb/provider-bridge-protocol/bridge-kit"; -export class RuntimeToolCalls { +export class RuntimeRequestLifetimes { private readonly pending = new Map< string, Map< @@ -33,7 +33,10 @@ export class RuntimeToolCalls { > >(); - start(scope: string, request: ToolCallRequest): AbortController | null { + start( + scope: string, + request: Pick, + ): AbortController | null { let calls = this.pending.get(scope); if (!calls) { calls = new Map(); @@ -110,7 +113,7 @@ interface HandleRuntimeProviderRequestArgs extends RuntimeProviderRequestArgs { ) => AgentRuntimeExecutionOptions | undefined; onInteractiveRequest: AgentRuntimeOptions["onInteractiveRequest"]; onToolCall: AgentRuntimeOptions["onToolCall"]; - toolCalls: RuntimeToolCalls; + requestLifetimes: RuntimeRequestLifetimes; resolveThreadId: ( args: ResolveRuntimeProviderRequestThreadIdArgs, ) => string | null; @@ -208,7 +211,7 @@ function handleToolCallProviderRequest( : {}), }; const scope = args.providerProcess.interactiveRequestScope; - const controller = args.toolCalls.start(scope, scopedToolCallReq); + const controller = args.requestLifetimes.start(scope, scopedToolCallReq); if (!controller) return true; void Promise.resolve() .then(() => { @@ -232,7 +235,11 @@ function handleToolCallProviderRequest( }); }) .finally(() => - args.toolCalls.finish(scope, scopedToolCallReq.requestId, controller), + args.requestLifetimes.finish( + scope, + scopedToolCallReq.requestId, + controller, + ), ); return true; } @@ -342,9 +349,17 @@ function handleInteractiveProviderRequest( return true; } + const scope = args.providerProcess.interactiveRequestScope; + const controller = args.requestLifetimes.start(scope, { + requestId: args.parsedId, + threadId: resolvedThreadId, + turnId: resolvedTurnId, + }); + if (!controller) return true; void args - .onInteractiveRequest(scopedInteractiveReq) + .onInteractiveRequest(scopedInteractiveReq, controller.signal) .then((resolution) => { + controller.signal.throwIfAborted(); const result = buildInteractiveResponse({ request: resolvedInteractiveReq, resolution, @@ -370,7 +385,10 @@ function handleInteractiveProviderRequest( id: args.parsedId, message: err instanceof Error ? err.message : String(err), }); - }); + }) + .finally(() => + args.requestLifetimes.finish(scope, args.parsedId, controller), + ); return true; } diff --git a/packages/agent-runtime/src/runtime-tool-cancellation.test.ts b/packages/agent-runtime/src/runtime-tool-cancellation.test.ts index bf83ee8ad2e..23b198e76d3 100644 --- a/packages/agent-runtime/src/runtime-tool-cancellation.test.ts +++ b/packages/agent-runtime/src/runtime-tool-cancellation.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { RuntimeToolCalls } from "./runtime-provider-requests.js"; +import { RuntimeRequestLifetimes } from "./runtime-provider-requests.js"; const request = { requestId: 1, @@ -12,7 +12,7 @@ const request = { describe("runtime tool cancellation ownership", () => { it("scopes cancellation to the requesting process and preserves JSON-RPC id types", () => { - const calls = new RuntimeToolCalls(); + const calls = new RuntimeRequestLifetimes(); const first = calls.start("process-a", request)!; const otherProcess = calls.start("process-b", request)!; const stringId = calls.start("process-a", { ...request, requestId: "1" })!; @@ -28,7 +28,7 @@ describe("runtime tool cancellation ownership", () => { }); it("cancels only the finishing turn and cancels all requests on thread detach", () => { - const calls = new RuntimeToolCalls(); + const calls = new RuntimeRequestLifetimes(); const old = calls.start("process", request)!; const next = calls.start("process", { ...request, diff --git a/packages/agent-runtime/src/runtime.interactive-requests.test.ts b/packages/agent-runtime/src/runtime.interactive-requests.test.ts index 00c418ee2e3..7185cfa663c 100644 --- a/packages/agent-runtime/src/runtime.interactive-requests.test.ts +++ b/packages/agent-runtime/src/runtime.interactive-requests.test.ts @@ -1,4 +1,4 @@ -import { RuntimeToolCalls } from "./runtime-provider-requests.js"; +import { RuntimeRequestLifetimes } from "./runtime-provider-requests.js"; import { spawn, type ChildProcess } from "node:child_process"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -105,7 +105,7 @@ async function answerDirectRequest(args: { } try { handleRuntimeProviderRequest({ - toolCalls: new RuntimeToolCalls(), + requestLifetimes: new RuntimeRequestLifetimes(), getActiveTurnId: args.getActiveTurnId ?? (() => "bb-turn-1"), getThreadExecutionOptions: args.getThreadExecutionOptions ?? (() => undefined), @@ -489,6 +489,7 @@ describe("createAgentRuntime interactive requests", () => { expect.objectContaining({ payload: expect.objectContaining({ kind: "secrets/secret-request" }), }), + expect.any(AbortSignal), ); expect(answer).toMatchObject({ jsonrpc: "2.0", diff --git a/packages/agent-runtime/src/runtime.tool-calls.test.ts b/packages/agent-runtime/src/runtime.tool-calls.test.ts index 5588e873708..f03e227ebc8 100644 --- a/packages/agent-runtime/src/runtime.tool-calls.test.ts +++ b/packages/agent-runtime/src/runtime.tool-calls.test.ts @@ -7,7 +7,7 @@ import type { ThreadEvent, ToolCallResponse } from "@bb/domain"; import { createProviderForId } from "./provider-registry.js"; import { handleRuntimeProviderRequest, - RuntimeToolCalls, + RuntimeRequestLifetimes, } from "./runtime-provider-requests.js"; import { parseJsonRpcLine, @@ -246,7 +246,7 @@ describe("createAgentRuntime tool calls", () => { try { handleRuntimeProviderRequest({ - toolCalls: new RuntimeToolCalls(), + requestLifetimes: new RuntimeRequestLifetimes(), getActiveTurnId: () => null, getThreadExecutionOptions: () => undefined, onInteractiveRequest: async () => ({ @@ -310,7 +310,7 @@ describe("createAgentRuntime tool calls", () => { try { handleRuntimeProviderRequest({ - toolCalls: new RuntimeToolCalls(), + requestLifetimes: new RuntimeRequestLifetimes(), getActiveTurnId: () => "turn-1", getThreadExecutionOptions: () => undefined, onInteractiveRequest: async () => ({ diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index eee218c351b..6c8a90cf943 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -38,7 +38,7 @@ import { } from "./execution-options.js"; import { handleRuntimeProviderRequest, - RuntimeToolCalls, + RuntimeRequestLifetimes, type ResolveRuntimeProviderRequestThreadIdArgs, type RuntimeProviderRequestKind, } from "./runtime-provider-requests.js"; @@ -303,7 +303,7 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { const suppressedThreadEventIds = new Set(); const threadGoalState = new RuntimeThreadGoalState(); const turnState = new RuntimeTurnState(); - const toolCalls = new RuntimeToolCalls(); + const requestLifetimes = new RuntimeRequestLifetimes(); const backgroundWorkState = new RuntimeBackgroundWorkState(); const threadEventGrammar = new ThreadEventGrammar(); const bridgeNodeEnv = defaultBridgeNodeEnv(); @@ -328,7 +328,7 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { handleStdoutLine(args.line, args.providerProcess), onProcessExit: options.onProcessExit, onProviderThreadDetached: (threadId) => { - toolCalls.cancelThread(threadId); + requestLifetimes.cancelThread(threadId); threadIdentityRegistry.clearThread(threadId); clearThreadRuntimeConfig(threadId); turnState.clearThread(threadId); @@ -1241,7 +1241,10 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { normalizedEvent.type === "turn/completed" && normalizedEvent.scope.kind === "turn" ) { - toolCalls.cancelThread(targetThreadId, normalizedEvent.scope.turnId); + requestLifetimes.cancelThread( + targetThreadId, + normalizedEvent.scope.turnId, + ); } turnState.observe(normalizedEvent); backgroundWorkState.observe(normalizedEvent); @@ -1257,7 +1260,7 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { args.parsed.params, ); if (cancellation.success) { - toolCalls.cancel( + requestLifetimes.cancel( args.proc.interactiveRequestScope, cancellation.data.requestId, ); @@ -1322,7 +1325,7 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { threadRuntimeConfigs.get(threadId)?.options, onInteractiveRequest: options.onInteractiveRequest, onToolCall: options.onToolCall, - toolCalls, + requestLifetimes, parsedId: parsedLine.parsedId, parsedMethod: parsedLine.parsedMethod, providerProcess: proc, @@ -2155,7 +2158,7 @@ export function createAgentRuntime(options: AgentRuntimeOptions): AgentRuntime { }, async stopThread({ threadId }) { - toolCalls.cancelThread(threadId); + requestLifetimes.cancelThread(threadId); return runThreadOperation({ threadId, work: async () => { diff --git a/packages/agent-runtime/src/types.ts b/packages/agent-runtime/src/types.ts index a7e465a13bd..1e3eeed39e5 100644 --- a/packages/agent-runtime/src/types.ts +++ b/packages/agent-runtime/src/types.ts @@ -81,6 +81,7 @@ export interface AgentRuntimeOptions { onInteractiveRequest?: ( request: PendingInteractionCreate, + signal?: AbortSignal, ) => Promise; onStderr?: (line: string, threadId?: string) => void; diff --git a/packages/db/src/data/pending-interactions.ts b/packages/db/src/data/pending-interactions.ts index b55509197f6..b662f6bcab7 100644 --- a/packages/db/src/data/pending-interactions.ts +++ b/packages/db/src/data/pending-interactions.ts @@ -1,4 +1,4 @@ -import { and, desc, eq, inArray } from "drizzle-orm"; +import { and, desc, eq, inArray, isNotNull, or } from "drizzle-orm"; import type { SQL } from "drizzle-orm"; import type { PendingInteractionStatus } from "@bb/domain"; import type { DbConnection, DbTransaction } from "../connection.js"; @@ -22,7 +22,7 @@ export type CreatePendingInteractionInput = providerId: string; providerRequestId: string; providerThreadId: string; - turnId: string; + turnId: string | null; }) | (CreatePendingInteractionInputBase & { originKind: "plugin"; @@ -58,6 +58,7 @@ export interface SetPendingInteractionResolvingArgs { } export interface InterruptPendingInteractionsForThreadsArgs { + providerRequestId?: string | null; providerId: string; resolvedAt?: number; statusReason: string; @@ -65,6 +66,7 @@ export interface InterruptPendingInteractionsForThreadsArgs { } export interface InterruptPendingInteractionsForThreadIdsArgs { + preserveAsyncQuestions?: boolean; resolvedAt?: number; statusReason: string; threadIds: readonly string[]; @@ -208,6 +210,7 @@ export function listActivePluginPendingInteractions( export function getActivePendingInteractionForThread( db: PendingInteractionReadConnection, threadId: string, + turnBoundOnly = false, ): PendingInteractionRow | null { return ( db @@ -216,6 +219,7 @@ export function getActivePendingInteractionForThread( .where( and( eq(pendingInteractions.threadId, threadId), + turnBoundOnly ? isNotNull(pendingInteractions.turnId) : undefined, inArray(pendingInteractions.status, ["pending", "resolving"]), ), ) @@ -351,6 +355,8 @@ export function interruptPendingInteractionsForThreads( extraConditions: [ eq(pendingInteractions.originKind, "provider"), eq(pendingInteractions.providerId, args.providerId), + isNotNull(pendingInteractions.turnId), + ...(args.providerRequestId ? [eq(pendingInteractions.providerRequestId, args.providerRequestId)] : []), ], resolvedAt: args.resolvedAt, statusReason: args.statusReason, @@ -387,7 +393,7 @@ export function interruptPendingInteractionsForThreadIds( args: InterruptPendingInteractionsForThreadIdsArgs, ): PendingInteractionRow[] { return interruptPendingInteractionsBatched(db, { - extraConditions: [], + extraConditions: args.preserveAsyncQuestions ? [or(eq(pendingInteractions.originKind, "plugin"), isNotNull(pendingInteractions.turnId))!] : [], resolvedAt: args.resolvedAt, statusReason: args.statusReason, threadIds: args.threadIds, diff --git a/packages/domain/src/pending-interactions.ts b/packages/domain/src/pending-interactions.ts index 92e95b07c6a..78d49c8f8ae 100644 --- a/packages/domain/src/pending-interactions.ts +++ b/packages/domain/src/pending-interactions.ts @@ -618,6 +618,7 @@ export type ApprovalPendingInteraction = z.infer< const userQuestionPendingInteractionSchema = providerPendingInteractionBaseSchema.extend({ + turnId: z.string().min(1).nullable(), payload: userQuestionPendingInteractionPayloadSchema, resolution: userQuestionPendingInteractionResolutionSchema.nullable(), }); diff --git a/packages/domain/src/provider-event.ts b/packages/domain/src/provider-event.ts index 008114b9777..20a49e7a91f 100644 --- a/packages/domain/src/provider-event.ts +++ b/packages/domain/src/provider-event.ts @@ -1,3 +1,4 @@ +import { userQuestionPendingInteractionPayloadSchema } from "./pending-interactions.js"; import { contextSnapshotSchema } from "./context-snapshot.js"; import { z } from "zod"; import { @@ -388,6 +389,12 @@ export const threadEventItemSchema = z.discriminatedUnion("type", [ type: z.literal("agentMessage"), id: z.string(), text: z.string(), + asyncQuestion: z + .object({ + id: z.string().min(1), + payload: userQuestionPendingInteractionPayloadSchema, + }) + .optional(), ...itemPresentationField, parentToolCallId: z.string().optional(), }), diff --git a/packages/host-daemon-contract/src/protocol.ts b/packages/host-daemon-contract/src/protocol.ts index c3d0dd1ed1d..263316935ab 100644 --- a/packages/host-daemon-contract/src/protocol.ts +++ b/packages/host-daemon-contract/src/protocol.ts @@ -1,3 +1,3 @@ -export const HOST_DAEMON_PROTOCOL_VERSION = 213 as const; +export const HOST_DAEMON_PROTOCOL_VERSION = 214 as const; export const HOST_ARTIFACT_MAX_BYTES = 256 * 1024 * 1024; diff --git a/packages/host-daemon-contract/src/session.ts b/packages/host-daemon-contract/src/session.ts index 6f07a68221a..0080258f670 100644 --- a/packages/host-daemon-contract/src/session.ts +++ b/packages/host-daemon-contract/src/session.ts @@ -824,6 +824,7 @@ export type HostDaemonInteractiveRequestResponse = z.infer< >; export const hostDaemonInteractiveInterruptRequestSchema = z.object({ + providerRequestId: z.string().min(1).nullable().default(null), sessionId: z.string().min(1), providerId: z.string().min(1), threadIds: z.array(z.string().min(1)).min(1), diff --git a/packages/provider-bridge-protocol/src/assembler/delta-assembler.ts b/packages/provider-bridge-protocol/src/assembler/delta-assembler.ts index 2db5465b062..1079bdc3e86 100644 --- a/packages/provider-bridge-protocol/src/assembler/delta-assembler.ts +++ b/packages/provider-bridge-protocol/src/assembler/delta-assembler.ts @@ -762,7 +762,14 @@ export function createDeltaAssembler( ); case "agentMessage": return withParentToolCallId( - { type: "agentMessage", id: bbItemId, text: shape.text }, + { + type: "agentMessage", + id: bbItemId, + text: shape.text, + ...(shape.asyncQuestion + ? { asyncQuestion: shape.asyncQuestion } + : {}), + }, parentToolCallId, ); case "reasoning": @@ -1161,7 +1168,14 @@ export function createDeltaAssembler( case "agentMessage": return withPresentation( withParentToolCallId( - { type: "agentMessage", id: open.bbItemId, text }, + { + type: "agentMessage", + id: open.bbItemId, + text, + ...(open.item.asyncQuestion + ? { asyncQuestion: open.item.asyncQuestion } + : {}), + }, open.item.parentToolCallId, ), open.item.presentation, diff --git a/packages/provider-bridge-protocol/src/thread-delta.ts b/packages/provider-bridge-protocol/src/thread-delta.ts index 6c91529d946..bfb624b28e5 100644 --- a/packages/provider-bridge-protocol/src/thread-delta.ts +++ b/packages/provider-bridge-protocol/src/thread-delta.ts @@ -1,5 +1,6 @@ import { contextSnapshotSchema, + userQuestionPendingInteractionPayloadSchema, backgroundTaskStatusSchema, backgroundTaskUsageSchema, clientTurnRequestIdSchema, @@ -135,7 +136,16 @@ export const deltaItemShapeSchema = z.discriminatedUnion("type", [ durationMs: z.number().optional(), }), z.object({ type: z.literal("compaction") }), - z.object({ type: z.literal("agentMessage"), text: z.string() }), + z.object({ + type: z.literal("agentMessage"), + text: z.string(), + asyncQuestion: z + .object({ + id: z.string().min(1), + payload: userQuestionPendingInteractionPayloadSchema, + }) + .optional(), + }), z.object({ type: z.literal("reasoning"), summary: z.array(z.string()), diff --git a/packages/thread-view/src/assistant-event-projection.ts b/packages/thread-view/src/assistant-event-projection.ts index eff748c35aa..c50cdd7425d 100644 --- a/packages/thread-view/src/assistant-event-projection.ts +++ b/packages/thread-view/src/assistant-event-projection.ts @@ -60,6 +60,13 @@ function createAssistantTextMessage( export function projectAssistantAndReasoningEvent( args: ProjectAssistantAndReasoningEventArgs, ): boolean { + if ( + (args.decoded.type === "item/started" || + args.decoded.type === "item/completed") && + args.decoded.item.type === "agentMessage" && + args.decoded.item.asyncQuestion !== undefined + ) + return true; const assistantIdentity = resolveBufferedTextIdentity({ decoded: args.decoded, kind: "assistant", diff --git a/packages/thread-view/test/build-thread-timeline.test.ts b/packages/thread-view/test/build-thread-timeline.test.ts index 3fe50427633..a2ebda45331 100644 --- a/packages/thread-view/test/build-thread-timeline.test.ts +++ b/packages/thread-view/test/build-thread-timeline.test.ts @@ -3333,3 +3333,47 @@ it("keeps a canonical disclosure ID when completed reasoning gains a delegation }); expect(completed?.id).not.toBe(live.activeThinking?.id); }); + +it("renders an async native question once through its persisted interaction", () => { + const lifecycle = userQuestionLifecycleEvent({ seq: 3 }); + if ( + lifecycle.event.type !== "system/interaction/lifecycle" || + lifecycle.event.interaction.payload.kind !== "user_question" + ) + throw new Error("Question fixture expected"); + const item = { + type: "agentMessage" as const, + id: "native-question", + text: "DUPLICATE_NATIVE_QUESTION_TEXT", + asyncQuestion: { + id: "native-call", + payload: lifecycle.event.interaction.payload, + }, + }; + const rows = buildTimelineRows([ + turnStartedEvent({ seq: 0 }), + { + event: { + type: "item/started", + threadId: "thread-1", + providerThreadId: "native-thread", + scope: turnScope("turn-1"), + item, + }, + meta: { id: "question-start", seq: 1, createdAt: 1 }, + }, + { + event: { + type: "item/completed", + threadId: "thread-1", + providerThreadId: "native-thread", + scope: turnScope("turn-1"), + item, + }, + meta: { id: "question-end", seq: 2, createdAt: 2 }, + }, + { ...lifecycle, event: { ...lifecycle.event, scope: threadScope() } }, + ]); + expect(collectQuestionRows(rows)).toHaveLength(1); + expect(JSON.stringify(rows)).not.toContain("DUPLICATE_NATIVE_QUESTION_TEXT"); +}); diff --git a/plugins/provider-codex/skills/codex-provider/SKILL.md b/plugins/provider-codex/skills/codex-provider/SKILL.md index c3fdc2cc452..dc12d696d1f 100644 --- a/plugins/provider-codex/skills/codex-provider/SKILL.md +++ b/plugins/provider-codex/skills/codex-provider/SKILL.md @@ -16,3 +16,12 @@ account access. Inspect models on the actual execution host with Use the core CLI skill for command syntax and official Codex guidance for upstream product behavior. + +Native async questions remain answerable after the turn ends. Answers use a new +user message, steering active work or starting an idle thread. Native synchronous +questions are enabled in Default mode and keep their original request open until +answered, closed by Codex, or stopped; BB does not apply the TUI countdown. +Use the question card or `bb thread interactions answer`; the SDK uses +`threads.interactions.resolve`. AskUserQuestion remains available for models and +versions without native tools. Secret-marked native requests are rejected; use +a secure credential input tool. diff --git a/plugins/provider-codex/src/bridge/app-server-connection.ts b/plugins/provider-codex/src/bridge/app-server-connection.ts index 1b46d1ebed5..4f751571b8b 100644 --- a/plugins/provider-codex/src/bridge/app-server-connection.ts +++ b/plugins/provider-codex/src/bridge/app-server-connection.ts @@ -10,6 +10,8 @@ const KILL_ESCALATION_MS = 4_000; const CLOSED_STDIN_ERROR_CODES = new Set(["EPIPE", "ERR_STREAM_DESTROYED"]); export interface CodexAppServerRequestResponder { + requestId: string | number; + dismiss(): void; result(value: unknown): void; error(code: number, message: string): void; } @@ -285,6 +287,10 @@ export function createCodexAppServerConnection( if (typeof id === "string" || typeof id === "number") { let settled = false; options.onRequest(message.method, message.params, { + requestId: id, + dismiss() { + settled = true; + }, result(value) { if (settled || finalized) return; settled = true; diff --git a/plugins/provider-codex/src/bridge/bridge.native-questions.test.ts b/plugins/provider-codex/src/bridge/bridge.native-questions.test.ts new file mode 100644 index 00000000000..1ae29f228a7 --- /dev/null +++ b/plugins/provider-codex/src/bridge/bridge.native-questions.test.ts @@ -0,0 +1,136 @@ +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it, vi } from "vitest"; +import { experimental_createBridgeJsonRpcTestHarness as createHarness } from "@get-bb/plugin-sdk/provider-bridge/testing"; +import { handleLine } from "./bridge.js"; +import { + FULL_ACCESS_SESSION_OPTIONS, + stubFakeCodexAppServer, +} from "./fake-codex-app-server-harness.js"; + +afterEach(() => vi.unstubAllEnvs()); + +it("correlates provider resolution to the original RPC and ignores its late answer", async () => { + const workspace = mkdtempSync(join(tmpdir(), "bb-native-questions-")); + const scriptPath = join(workspace, "script.json"); + const responseLogPath = join(workspace, "responses.jsonl"); + const turn = { + id: "native-turn", + items: [], + status: "inProgress", + error: null, + }; + const request = { + kind: "request", + method: "item/tool/requestUserInput", + params: { + threadId: "native-thread", + turnId: "native-turn", + itemId: "native-item", + isBlocking: false, + questions: [ + { + id: "q", + header: "Choice", + question: "Which color?", + options: [{ label: "Blue", description: "Blue fixture" }], + isOther: true, + isSecret: false, + }, + ], + }, + }; + writeFileSync( + scriptPath, + JSON.stringify({ + responseLogPath, + turns: [ + [ + { + method: "turn/started", + params: { threadId: "native-thread", turn }, + }, + { ...request, resolveAfterMs: 40 }, + request, + { + method: "turn/completed", + params: { + threadId: "native-thread", + turn: { ...turn, status: "completed" }, + }, + }, + ], + ], + }), + ); + stubFakeCodexAppServer(scriptPath); + const bridge = createHarness(handleLine); + try { + bridge.sendRequest(1, "thread/start", { + threadId: "native-bb-thread", + cwd: workspace, + instructionMode: "append", + options: FULL_ACCESS_SESSION_OPTIONS, + }); + await bridge.waitForResponse(1); + bridge.sendRequest(2, "turn/start", { + threadId: "native-bb-thread", + providerThreadId: "native-thread", + input: [{ type: "text", text: "Ask", mentions: [] }], + clientRequestId: "creq_native2345", + options: FULL_ACCESS_SESSION_OPTIONS, + }); + await bridge.waitForResponse(2); + await vi.waitFor(() => { + expect( + bridge.messages.filter( + (message) => message.method === "interaction/request", + ), + ).toHaveLength(2); + }); + const requests = bridge.messages.filter( + (message) => message.method === "interaction/request", + ); + expect(bridge.messages).toContainEqual( + expect.objectContaining({ + method: "notifications/cancelled", + params: { requestId: requests[0]!.id }, + }), + ); + for (const request of requests) + handleLine( + JSON.stringify({ + jsonrpc: "2.0", + id: request.id, + result: { + kind: "user_answer", + answers: { q: { selected: ["Blue"], freeText: "Pale" } }, + }, + }), + ); + await vi.waitFor(() => { + const responses = readFileSync(responseLogPath, "utf8") + .trim() + .split("\n") + .map((line) => JSON.parse(line)); + expect(responses).toEqual([ + { + jsonrpc: "2.0", + id: "fx-req-2", + result: { answers: { q: { answers: ["Blue", "Pale"] } } }, + }, + ]); + }); + } finally { + bridge.sendRequest(999, "thread/stop", { + threadId: "native-bb-thread", + providerThreadId: "native-thread", + intent: "release", + activeTurnId: null, + }); + await bridge.waitForResponse(999); + bridge.restore(); + rmSync(workspace, { recursive: true, force: true }); + } +}); diff --git a/plugins/provider-codex/src/bridge/bridge.ts b/plugins/provider-codex/src/bridge/bridge.ts index c9ff4e4ca67..b469f0e364a 100644 --- a/plugins/provider-codex/src/bridge/bridge.ts +++ b/plugins/provider-codex/src/bridge/bridge.ts @@ -4,6 +4,7 @@ import { createHash } from "node:crypto"; import { isStandaloneBuiltinCompactCommand, approvalInteractionOutcomeSchema, + userQuestionInteractionOutcomeSchema, type DynamicTool, type PromptInput, type ThreadDelta, @@ -54,6 +55,7 @@ import { } from "../extension-kinds.js"; import { buildCodexInteractiveResponse, + buildCodexUserInputResponse, decodeCodexInteractiveRequest, extractCodexMacOsPermissionRequest, type CodexMacOsPermissionRequest, @@ -249,6 +251,7 @@ let runtimeRequestIdCounter = 0; function sendRuntimeRequest( method: string, params: Record, + onCreated?: (requestId: number) => void, ): Promise { runtimeRequestIdCounter += 1; const requestId = runtimeRequestIdCounter; @@ -265,6 +268,7 @@ function sendRuntimeRequest( }); }, ); + onCreated?.(requestId); send({ jsonrpc: "2.0", id: requestId, method, params }); return responsePromise; } @@ -458,6 +462,10 @@ interface CodexBridgeSession { awaitingReplayedUsage: boolean; identityAnnounced: boolean; pendingPreIdentityDeltas: ThreadDelta[]; + interactiveRequests: Map< + string | number, + { runtimeRequestId: number; responder: CodexAppServerRequestResponder } + >; rebuildBeforeNextTurnReason: string | null; closing: boolean; previousChildExit: Promise | null; @@ -669,6 +677,25 @@ function handleChildNotification( if (!session) { return; } + if (method === "serverRequest/resolved") { + const parsed = z + .object({ + threadId: z.string(), + requestId: z.union([z.string(), z.number()]), + }) + .safeParse(params); + if (parsed.success && parsed.data.threadId === session.codexThreadId) { + const request = session.interactiveRequests.get(parsed.data.requestId); + if (request) { + request.responder.dismiss(); + session.interactiveRequests.delete(parsed.data.requestId); + sendNotification("notifications/cancelled", { + requestId: request.runtimeRequestId, + }); + } + } + return; + } if (method === "thread/started") { const parsed = codexThreadStartedNotificationSchema.safeParse(params); if (parsed.success) { @@ -781,7 +808,11 @@ function handleChildRequest( let decoded: DecodedInteractiveRequest | null; try { - decoded = decodeCodexInteractiveRequest({ id: 0, method, params }); + decoded = decodeCodexInteractiveRequest({ + id: responder.requestId, + method, + params, + }); } catch (error) { responder.error( BRIDGE_JSON_RPC_ERRORS.INVALID_PARAMS, @@ -798,14 +829,34 @@ function handleChildRequest( } const request = decoded; - void sendRuntimeRequest(BRIDGE_INBOUND_REQUEST_METHODS.interactionRequest, { - providerThreadId: session.codexThreadId ?? request.providerThreadId, - threadId: session.bbThreadId, - turnId: request.turnId, - payload: request.payload, - providerNativeIds: true, - }) + void sendRuntimeRequest( + BRIDGE_INBOUND_REQUEST_METHODS.interactionRequest, + { + providerThreadId: session.codexThreadId ?? request.providerThreadId, + threadId: session.bbThreadId, + turnId: request.turnId, + payload: request.payload, + providerNativeIds: true, + }, + (runtimeRequestId) => { + session.interactiveRequests.set(responder.requestId, { + runtimeRequestId, + responder, + }); + }, + ) .then((result) => { + if (request.payload.kind === "user_question") { + responder.result( + buildCodexUserInputResponse( + userQuestionInteractionOutcomeSchema.parse({ + payload: request.payload, + resolution: result, + }), + ), + ); + return; + } const outcome = approvalInteractionOutcomeSchema.parse({ payload: request.payload, resolution: result, @@ -817,6 +868,13 @@ function handleChildRequest( BRIDGE_JSON_RPC_ERRORS.BRIDGE_ERROR, error instanceof Error ? error.message : String(error), ); + }) + .finally(() => { + if ( + session.interactiveRequests.get(responder.requestId)?.responder === + responder + ) + session.interactiveRequests.delete(responder.requestId); }); } @@ -1023,6 +1081,7 @@ async function constructThreadSession( awaitingReplayedUsage: args.request.kind !== "start", identityAnnounced: false, pendingPreIdentityDeltas: [], + interactiveRequests: new Map(), rebuildBeforeNextTurnReason: null, closing: false, previousChildExit: null, @@ -1183,6 +1242,7 @@ function registerResumableSession(session: CodexBridgeSession): void { awaitingReplayedUsage: true, identityAnnounced: session.identityAnnounced, pendingPreIdentityDeltas: [], + interactiveRequests: new Map(), rebuildBeforeNextTurnReason: null, closing: false, previousChildExit: null, diff --git a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs index d8e63b7cb61..c4b588fde3a 100644 --- a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs +++ b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs @@ -357,12 +357,22 @@ function withThreadId(value, threadId) { let outboundRequestCounter = 0; const pendingOutboundRequests = new Map(); -function requestFromClient(method, params) { +function requestFromClient(method, params, resolveAfterMs) { outboundRequestCounter += 1; const id = `fx-req-${outboundRequestCounter}`; return new Promise((resolve) => { pendingOutboundRequests.set(id, resolve); send({ jsonrpc: "2.0", id, method, params }); + if (resolveAfterMs !== undefined) + setTimeout(() => { + if (!pendingOutboundRequests.has(id)) return; + pendingOutboundRequests.delete(id); + notify("serverRequest/resolved", { + threadId: params.threadId, + requestId: id, + }); + resolve(null); + }, resolveAfterMs); }); } @@ -392,7 +402,7 @@ async function runScriptFileTurn(threadId) { for (const entry of turn) { const params = withThreadId(entry.params ?? {}, threadId); if (entry.kind === "request") { - await requestFromClient(entry.method, params); + await requestFromClient(entry.method, params, entry.resolveAfterMs); continue; } if (entry.method === "turn/started") { @@ -771,6 +781,8 @@ stdinLines.on("line", (line) => { } if (parsed.id !== undefined) { // A response to a request this process originated (an approval answer). + if (script?.responseLogPath) + appendFileSync(script.responseLogPath, `${JSON.stringify(parsed)}\n`); const resolve = pendingOutboundRequests.get(parsed.id); if (resolve) { pendingOutboundRequests.delete(parsed.id); diff --git a/plugins/provider-codex/src/delta-translation.test.ts b/plugins/provider-codex/src/delta-translation.test.ts index f029daf8451..ffbd2465731 100644 --- a/plugins/provider-codex/src/delta-translation.test.ts +++ b/plugins/provider-codex/src/delta-translation.test.ts @@ -2735,3 +2735,62 @@ describe("codex ignored notifications", () => { expect(events).toEqual([]); }); }); + +describe("native async questions", () => { + it("retains structured questions through translation and assembly without ending the turn", () => { + const harness = createHarness(); + const events = harness.translate({ + jsonrpc: "2.0", + method: "item/completed", + params: { + threadId: "t1", + turnId: "turn-async", + item: { + type: "agentMessage", + id: "call-async", + text: "Which database?", + phase: "final_answer", + delivery: "async", + questions: [ + { title: "Which database?", options: ["SQLite", "Postgres"] }, + { title: "Any constraints?" }, + ], + }, + }, + }); + expect(events).toContainEqual( + expect.objectContaining({ + type: "item/completed", + item: expect.objectContaining({ + type: "agentMessage", + presentation: expect.objectContaining({ suppress: true }), + asyncQuestion: { + id: "call-async", + payload: { + kind: "user_question", + questions: [ + { + id: "0", + prompt: "Which database?", + multiSelect: false, + allowFreeText: true, + options: [ + { value: "SQLite", label: "SQLite" }, + { value: "Postgres", label: "Postgres" }, + ], + }, + { + id: "1", + prompt: "Any constraints?", + multiSelect: false, + allowFreeText: true, + }, + ], + }, + }, + }), + }), + ); + expect(events.some((event) => event.type === "turn/completed")).toBe(false); + }); +}); diff --git a/plugins/provider-codex/src/delta-translation.ts b/plugins/provider-codex/src/delta-translation.ts index 8b6b37dbb8a..d0b17e007a7 100644 --- a/plugins/provider-codex/src/delta-translation.ts +++ b/plugins/provider-codex/src/delta-translation.ts @@ -773,8 +773,40 @@ function translateCodexItemShape( case "agentMessage": return { kind: "translated", - shape: { type: "agentMessage", text: parsedItem.text }, - presentation: AGENT_MESSAGE_PRESENTATION, + shape: { + type: "agentMessage", + text: parsedItem.text, + ...(parsedItem.delivery === "async" && parsedItem.questions?.length + ? { + asyncQuestion: { + id: parsedItem.id, + payload: { + kind: "user_question", + questions: parsedItem.questions.map((question, index) => ({ + id: String(index), + prompt: question.title, + multiSelect: false, + allowFreeText: true, + ...(question.options + ? { + options: question.options.map((label) => ({ + value: label, + label, + })), + } + : {}), + })), + }, + }, + } + : {}), + }, + presentation: { + ...AGENT_MESSAGE_PRESENTATION, + ...(parsedItem.delivery === "async" && parsedItem.questions?.length + ? { suppress: true } + : {}), + }, status: "completed", approvalDenied: false, }; diff --git a/plugins/provider-codex/src/interactive-requests.test.ts b/plugins/provider-codex/src/interactive-requests.test.ts index eed15e54e57..5ee6062e23a 100644 --- a/plugins/provider-codex/src/interactive-requests.test.ts +++ b/plugins/provider-codex/src/interactive-requests.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { buildCodexInteractiveResponse, + buildCodexUserInputResponse, decodeCodexInteractiveRequest, extractCodexMacOsPermissionRequest, } from "./interactive-requests.js"; @@ -582,3 +583,103 @@ describe("buildCodexInteractiveResponse", () => { }); }); }); + +describe("native user questions", () => { + const params = { + threadId: "thread-native", + turnId: "turn-native", + itemId: "call-native", + isBlocking: false, + questions: [ + { + id: "database", + header: "Database", + question: "Which database?", + isOther: true, + isSecret: false, + options: [ + { label: "SQLite", description: "Local storage" }, + { label: "Postgres", description: "" }, + ], + }, + ], + }; + + it.each([true, false])( + "keeps isBlocking=%s requests on their original turn", + (isBlocking) => { + const request = decodeCodexInteractiveRequest({ + id: 42, + method: "item/tool/requestUserInput", + params: { ...params, isBlocking }, + }); + expect(request).toMatchObject({ + requestId: 42, + turnId: "turn-native", + payload: { + kind: "user_question", + questions: [ + { + id: "database", + multiSelect: false, + allowFreeText: true, + options: [ + { + value: "SQLite", + label: "SQLite", + description: "Local storage", + }, + { value: "Postgres", label: "Postgres" }, + ], + }, + ], + }, + }); + expect( + buildCodexUserInputResponse({ + payload: { kind: "user_question", questions: [] }, + resolution: { + kind: "user_answer", + answers: { + database: { selected: ["SQLite"], freeText: "Keep it local" }, + }, + }, + }), + ).toEqual({ + answers: { database: { answers: ["SQLite", "Keep it local"] } }, + }); + }, + ); + + it("accepts older requests without blocking metadata and free-text-only questions", () => { + expect( + decodeCodexInteractiveRequest({ + id: "old", + method: "item/tool/requestUserInput", + params: { + threadId: "t", + turnId: "turn", + itemId: "i", + questions: [ + { id: "q", header: "", question: "Details?", options: null }, + ], + }, + }), + ).toMatchObject({ + payload: { questions: [{ id: "q", allowFreeText: true }] }, + }); + }); + + it("rejects secret questions before creating a persisted interaction", () => { + expect(() => + decodeCodexInteractiveRequest({ + id: 42, + method: "item/tool/requestUserInput", + params: { + ...params, + questions: [{ ...params.questions[0], isSecret: true }], + }, + }), + ).toThrow("secure credential input"); + }); +}); diff --git a/plugins/provider-codex/src/interactive-requests.ts b/plugins/provider-codex/src/interactive-requests.ts index fc36d49efe3..20d5c822d35 100644 --- a/plugins/provider-codex/src/interactive-requests.ts +++ b/plugins/provider-codex/src/interactive-requests.ts @@ -2,6 +2,8 @@ import { ProviderRequestDecodeError as ProviderRequestDecodeErrorValue, ProviderResponseEncodeError, type ApprovalInteractionOutcome, + type UserQuestionInteractionOutcome, + userQuestionInteractionOutcomeSchema, type DecodedInteractiveRequest, type ProviderInboundRequest, type PendingInteractionApprovalDecision, @@ -15,6 +17,7 @@ import type { CommandExecutionRequestApprovalResponse } from "./generated/codex- import type { FileChangeRequestApprovalResponse } from "./generated/codex-app-server/schema/v2/FileChangeRequestApprovalResponse.js"; import type { PermissionsRequestApprovalResponse } from "./generated/codex-app-server/schema/v2/PermissionsRequestApprovalResponse.js"; import { + codexUserInputRequestSchema, codexCommandExecutionRequestApprovalParamsSchema, codexFileChangeRequestApprovalParamsSchema, codexPermissionsRequestApprovalParamsSchema, @@ -87,6 +90,37 @@ export function decodeCodexInteractiveRequest( } switch (request.method) { + case "item/tool/requestUserInput": { + const parsed = codexUserInputRequestSchema.parse(request.params); + if (parsed.questions.some((question) => question.isSecret)) { + throw new ProviderRequestDecodeErrorValue( + "Secret questions require a secure credential input tool", + ); + } + return { + requestId: request.id, + method: request.method, + providerThreadId: parsed.threadId, + turnId: parsed.turnId, + payload: userQuestionInteractionOutcomeSchema.shape.payload.parse({ + kind: "user_question", + questions: parsed.questions.map((question) => ({ + id: question.id, + prompt: question.question, + ...(question.header.trim() ? { shortLabel: question.header } : {}), + multiSelect: false, + allowFreeText: true, + options: question.options?.map((option) => ({ + value: option.label, + label: option.label, + ...(option.description.trim() + ? { description: option.description } + : {}), + })), + })), + }), + }; + } case "item/commandExecution/requestApproval": { const parsed = codexCommandExecutionRequestApprovalParamsSchema.safeParse( request.params, @@ -423,3 +457,21 @@ function parseCodexAvailableDecisions( } return uniqueDecisions; } + +export function buildCodexUserInputResponse( + args: UserQuestionInteractionOutcome, +) { + return { + answers: Object.fromEntries( + Object.entries(args.resolution.answers).map(([id, answer]) => [ + id, + { + answers: [ + ...answer.selected, + ...(answer.freeText ? [answer.freeText] : []), + ], + }, + ]), + ), + }; +} diff --git a/plugins/provider-codex/src/schemas.ts b/plugins/provider-codex/src/schemas.ts index ffcf4fc1592..3778f21cf09 100644 --- a/plugins/provider-codex/src/schemas.ts +++ b/plugins/provider-codex/src/schemas.ts @@ -349,6 +349,15 @@ export const codexHandledThreadItemSchema = z.discriminatedUnion("type", [ type: z.literal("agentMessage"), id: z.string(), text: z.string(), + delivery: z.enum(["sync", "async"]).nullish(), + questions: z + .array( + z.object({ + title: z.string().min(1), + options: z.array(z.string().min(1)).nullish(), + }), + ) + .nullish(), }) .passthrough(), z @@ -1045,3 +1054,26 @@ export function isHandledCodexMethod( ): method is HandledCodexMethod { return handledCodexMethodSet.has(method); } + +export const codexUserInputRequestSchema = z.object({ + threadId: z.string().min(1), + turnId: z.string().min(1), + itemId: z.string().min(1), + isBlocking: z.boolean().optional(), + questions: z + .array( + z.object({ + id: z.string().min(1), + header: z.string(), + question: z.string().min(1), + isOther: z.boolean().default(false), + isSecret: z.boolean().default(false), + options: z + .array( + z.object({ label: z.string().min(1), description: z.string() }), + ) + .nullish(), + }), + ) + .min(1), +}); diff --git a/plugins/provider-codex/src/session-params.test.ts b/plugins/provider-codex/src/session-params.test.ts index ec0f03fad83..9cd2a4fbbc7 100644 --- a/plugins/provider-codex/src/session-params.test.ts +++ b/plugins/provider-codex/src/session-params.test.ts @@ -607,11 +607,11 @@ function configFor(options: CodexSessionOptions) { } describe("buildCodexConfig", () => { - it("disables provider user-input requests without overriding web search", () => { + it("enables native Default-mode questions without overriding web search", () => { const config = configFor(FULL_OPTIONS); expect(config).toMatchObject({ - "features.default_mode_request_user_input": false, + "features.default_mode_request_user_input": true, }); expect(JSON.stringify(config)).not.toContain("tools.web_search"); }); diff --git a/plugins/provider-codex/src/session-params.ts b/plugins/provider-codex/src/session-params.ts index 6f7eae84437..5b892739b5a 100644 --- a/plugins/provider-codex/src/session-params.ts +++ b/plugins/provider-codex/src/session-params.ts @@ -610,7 +610,7 @@ export function buildCodexConfig( args.options.reasoningLevel, ); } - config["features.default_mode_request_user_input"] = false; + config["features.default_mode_request_user_input"] = true; if (args.options?.providerSubagentsEnabled === false) { config["features.multi_agent"] = false; config["features.multi_agent_v2.max_concurrent_threads_per_session"] = 1;