From 4977f59bae629908e05a1cca7d843ed49d477cd2 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Sun, 6 Sep 2026 13:03:48 +0400 Subject: [PATCH 1/5] feat: support asynchronous questions through an ACP extension --- docs/ASYNC_QUESTIONS_PROPOSAL.md | 183 +++++++++++++++ readme-dev.md | 4 + src/AcpExtensions.ts | 8 + src/AsyncQuestionExtension.ts | 27 +++ src/CodexAcpServer.ts | 37 +++- src/CodexAsyncQuestionHandler.ts | 109 +++++++++ src/CodexEventHandler.ts | 8 + src/SteeringQueue.ts | 9 +- .../CodexACPAgent/async-questions.test.ts | 208 ++++++++++++++++++ .../CodexACPAgent/initialize.test.ts | 1 + .../snapshots/async-questions-active.json | 80 +++++++ .../snapshots/async-questions-late-input.json | 7 + src/__tests__/acp-test-utils.ts | 6 + 13 files changed, 675 insertions(+), 12 deletions(-) create mode 100644 docs/ASYNC_QUESTIONS_PROPOSAL.md create mode 100644 src/AsyncQuestionExtension.ts create mode 100644 src/CodexAsyncQuestionHandler.ts create mode 100644 src/__tests__/CodexACPAgent/async-questions.test.ts create mode 100644 src/__tests__/CodexACPAgent/snapshots/async-questions-active.json create mode 100644 src/__tests__/CodexACPAgent/snapshots/async-questions-late-input.json diff --git a/docs/ASYNC_QUESTIONS_PROPOSAL.md b/docs/ASYNC_QUESTIONS_PROPOSAL.md new file mode 100644 index 00000000..e6dcf3a9 --- /dev/null +++ b/docs/ASYNC_QUESTIONS_PROPOSAL.md @@ -0,0 +1,183 @@ +# Proposal: asynchronous user questions over ACP + +Status: experimental provider extension, version 1, implemented by `codex-acp`. +This document proposes a client contract; it does not add a standard ACP method. + +## Problem and intended behavior + +Codex can ask a question and continue working before the user answers. For example, +it asks for a YouTrack issue number while investigating a fix. The user answers +“create an issue” in a question form, and Codex receives that answer as new user input. + +The ACP client receives one ordinary request/response RPC for the question. The RPC +waits for the user, but neither the session notification queue nor the running Codex +turn waits for it. Answering after the original turn finishes can start another turn. + +This is separate from synchronous Codex `item/tool/requestUserInput`, which continues +to use standard ACP elicitation and returns a tool response to the waiting Codex call. + +## Capability negotiation + +A client opts in through `initialize.params.clientCapabilities._meta`: + +```json +{ + "clientCapabilities": { + "_meta": { + "codex.asyncQuestions": { "version": 1 } + } + } +} +``` + +The provider advertises its implementation in +`initialize.result.agentCapabilities._meta`: + +```json +{ + "codex.asyncQuestions": { + "version": 1, + "requestMethod": "_codex/requestUserInput" + } +} +``` + +Version must be the number `1`. Missing, malformed, or unsupported versions receive +ordinary question text without an extension request. Standard form elicitation +support does not implicitly enable this feature. No environment setting is needed. + +## Provider-to-client question request + +```json +{ + "jsonrpc": "2.0", + "id": 42, + "method": "_codex/requestUserInput", + "params": { + "sessionId": "session-1", + "turnId": "turn-1", + "itemId": "call-1", + "questions": [ + { + "id": "[\"request_user_input_async\",\"call-1\",0]", + "title": "Is there a YouTrack issue for this fix?" + }, + { + "id": "[\"request_user_input_async\",\"call-1\",1]", + "title": "Which component?", + "options": ["Platform", "Plugin"] + } + ] + } +} +``` + +The client should: + +- Present all questions together, associated with the specified session and item. +- Always allow free text. `options` are suggestions, not an enum restricting answers. +- Omit automatic submission or selection. Preserve the user's submitted text. +- Keep rendering session updates and accepting other input while the RPC is pending. +- Keep the form available after the originating `session/prompt` completes. +- Associate the form with the ordinary `agent_message_chunk` whose `messageId` equals + `itemId`, so the transcript and form do not appear to be unrelated questions. + +Question IDs are opaque strings. Return them unchanged. `turnId` identifies the +origin of the question; it does not require the answer to reach that same turn. + +## Client response + +On submission, return exactly one nonblank string answer for each question: + +```json +{ + "jsonrpc": "2.0", + "id": 42, + "result": { + "status": "answered", + "answers": [ + { "id": "[\"request_user_input_async\",\"call-1\",0]", "answer": "Create an issue" }, + { "id": "[\"request_user_input_async\",\"call-1\",1]", "answer": "Platform" } + ] + } +} +``` + +Order is not significant. Unknown or duplicate IDs, missing answers, non-string +values, and blank answers invalidate the entire response; nothing is submitted. +The client may instead close the form without sending user input: + +```json +{ "jsonrpc": "2.0", "id": 42, "result": { "status": "dismissed" } } +``` + +The client records the submitted answer in its UI. It must not also send +`session/prompt` or `_session/steering` for that answer: the provider handles delivery. +The RPC result is not an acknowledgement that Codex has consumed the answer. + +## Codex mapping and input delivery + +The adapter observes live `item/completed` notifications with an `agentMessage` +whose `delivery` is `"async"` and whose `questions` array is nonempty. Each question +is mapped to the custom request above. There is no additional app-server request +to register: this is a completed message event, not `item/tool/requestUserInput`. + +Codex's observed `request_user_input_async` call returns `{"accepted":true}` immediately. +The later answer is a user message with this payload: + +```text + +[{"questionItemId":"[\"request_user_input_async\",\"call-1\",0]","question":"Is there a YouTrack issue for this fix?","answer":"Create an issue"},{"questionItemId":"[\"request_user_input_async\",\"call-1\",1]","question":"Which component?","answer":"Platform"}] + +``` + +The adapter constructs this payload from the original request and validated answers. +Clients do not construct it. This wrapper follows the observed Codex desktop format; +it is a Codex-specific compatibility detail, not a portable ACP standard. + +The existing per-session steering queue delivers the input: + +1. An active turn receives `turn/steer` with its current `expectedTurnId`. +2. If the turn has finished, the adapter waits for prompt cleanup and uses `turn/start`. +3. A “no active turn” race follows the existing steering fallback to a new turn. + +Concurrent answers and other steering requests share the same queue. Output from a +new turn streams through ordinary ACP session updates even though no client +`session/prompt` request is outstanding. Clients opting in must support that lifecycle. + +## Cancellation, history, and failure + +- A pending question survives normal turn completion. There is no answer timeout. +- `session/cancel`, session close/delete, and provider replacement cancel outstanding + question RPCs through ACP `$/cancel_request`. The client should close the form and + settle its RPC. A late response to a cancelled question is ignored. +- Cancellation also prevents an answer still queued for delivery from starting work. + Input already accepted by Codex cannot be retracted by dismissing the form. +- Repeated live events with the same item ID create at most one request per loaded + session. Different sessions have independent question IDs and pending requests. +- Loading or forking history displays text only; it does not reopen historical forms. + Pending forms are not persisted across adapter restart or session close/reopen. +- Without the capability, a completed async message is rendered as ordinary text, + including messages that arrive without text deltas. The user can reply in chat. +- RPC errors, malformed responses, or failed answer delivery are logged and produce + a visible request to send the answer in chat. There is no automatic retry that could + duplicate an answer after an uncertain transport outcome. + +## Implementation and validation + +`AsyncQuestionExtension.ts` defines the wire types. `CodexAsyncQuestionHandler.ts` +owns pending questions across prompt boundaries. `CodexAcpServer.ts` connects live +events, session cancellation, and the existing `SteeringQueue`. `CodexEventHandler.ts` +provides ordinary text rendering for completed async messages without deltas. + +Behavior tests in `src/__tests__/CodexACPAgent/async-questions.test.ts` cover capability +negotiation, nonblocking progress, deduplication, multiple questions and free text, +active-turn delivery, late-answer turn creation, cancellation, and invalid responses. +File snapshots record the ACP request and exact Codex input payload. + +## Future standardization + +A standard ACP proposal could generalize this request without exposing Codex-specific +IDs or the input wrapper. Version 1 deliberately leaves durable pending-question +recovery and explicit delivery acknowledgements for a future revision. Client-side +form rendering must be implemented by each ACP client before advertising support. diff --git a/readme-dev.md b/readme-dev.md index bc147807..edd8774d 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -3,6 +3,10 @@ Set `CODEX_PATH` to run a different Codex binary; versions other than the one sp ### Runtime environment +For the opt-in client extension that displays asynchronous Codex questions and sends +answers back as user input, see [Async questions proposal](docs/ASYNC_QUESTIONS_PROPOSAL.md). +It is negotiated through ACP capabilities and requires no environment setting. + - `CODEX_API_KEY` - API key used when the API-key auth method is selected. Takes precedence over `OPENAI_API_KEY`. - `OPENAI_API_KEY` - fallback API key used when the API-key auth method is selected. - `CODEX_PATH` - run a specific Codex executable instead of the bundled package dependency. diff --git a/src/AcpExtensions.ts b/src/AcpExtensions.ts index b450c8bd..0b6cd6de 100644 --- a/src/AcpExtensions.ts +++ b/src/AcpExtensions.ts @@ -139,3 +139,11 @@ export async function steerSessionWithFallback( ): Promise { return await connection.request(SESSION_STEERING_METHOD, params); } + +export { + ASYNC_QUESTIONS_CAPABILITY, + ASYNC_QUESTIONS_VERSION, + ASYNC_QUESTION_REQUEST_METHOD, + type AsyncQuestionRequest, + type AsyncQuestionResponse, +} from "./AsyncQuestionExtension"; diff --git a/src/AsyncQuestionExtension.ts b/src/AsyncQuestionExtension.ts new file mode 100644 index 00000000..c8f862f4 --- /dev/null +++ b/src/AsyncQuestionExtension.ts @@ -0,0 +1,27 @@ +import type {ClientCapabilities} from "@agentclientprotocol/sdk"; + +/** Experimental, versioned provider extension; not a standard ACP method. */ +export const ASYNC_QUESTIONS_CAPABILITY = "codex.asyncQuestions"; +export const ASYNC_QUESTIONS_VERSION = 1; +export const ASYNC_QUESTION_REQUEST_METHOD = "_codex/requestUserInput"; + +export type AsyncQuestionRequest = { + sessionId: string; + turnId: string; + itemId: string; + questions: Array<{id: string; title: string; options?: string[]}>; +}; + +export type AsyncQuestionResponse = + | {status: "answered"; answers: Array<{id: string; answer: string}>} + | {status: "dismissed"}; + +export function asyncQuestionsCapability() { + return {version: ASYNC_QUESTIONS_VERSION, requestMethod: ASYNC_QUESTION_REQUEST_METHOD}; +} + +export function clientSupportsAsyncQuestions(capabilities: ClientCapabilities | null): boolean { + const value = capabilities?._meta?.[ASYNC_QUESTIONS_CAPABILITY]; + return typeof value === "object" && value !== null + && "version" in value && value.version === ASYNC_QUESTIONS_VERSION; +} diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 4dc15e01..510d9a86 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -1,3 +1,5 @@ +import {CodexAsyncQuestionHandler} from "./CodexAsyncQuestionHandler"; +import {ASYNC_QUESTIONS_CAPABILITY, asyncQuestionsCapability} from "./AsyncQuestionExtension"; import * as acp from "@agentclientprotocol/sdk"; import {RequestError, type SessionId, type SessionModeState} from "@agentclientprotocol/sdk"; import {CodexEventHandler, type CompletedPlan} from "./CodexEventHandler"; @@ -276,6 +278,7 @@ export class CodexAcpServer { private readonly pendingTurnStarts: Map; private readonly activePrompts: Map; private readonly steeringQueues: Map; + private readonly asyncQuestions: CodexAsyncQuestionHandler; private readonly closingSessions: Map; private readonly sessionGenerations: Map; private readonly sessionOpenGenerations: Map; @@ -305,6 +308,8 @@ export class CodexAcpServer { this.goalControlGenerations = new Map(); this.permissionLifecycleContexts = new WeakMap(); this.connection = connection; + this.asyncQuestions = new CodexAsyncQuestionHandler(connection, (request, signal) => + this.executeOrQueueSteeringRequest(request, signal)); this.codexAcpClient = codexAcpClient; this.defaultAuthRequest = defaultAuthRequest ?? null; this.codexProcessState = codexProcessState ?? null; @@ -377,6 +382,7 @@ export class CodexAcpServer { // Presence means "this agent pushes `_auth/status_update`". It // never carries a payload, and the client never asks for one. [AUTH_STATUS_META_KEY]: authStatusCapability(), + [ASYNC_QUESTIONS_CAPABILITY]: asyncQuestionsCapability(), }, }, authMethods: getCodexAuthMethods(_params.clientCapabilities), @@ -861,6 +867,7 @@ export class CodexAcpServer { async closeSession(params: acp.CloseSessionRequest): Promise { logger.log("Closing session...", {sessionId: params.sessionId}); + this.asyncQuestions.closeSession(params.sessionId); const closeGeneration = this.bumpSessionGeneration(params.sessionId); const sessionState = this.sessions.get(params.sessionId); this.beginSessionCloseFence(params.sessionId); @@ -1036,6 +1043,7 @@ export class CodexAcpServer { logger.log("Restarting Codex app-server for provider update", {sessionCount: this.sessions.size}); for (const session of this.sessions.values()) { + this.asyncQuestions.cancelSession(session.sessionId); session.asyncTasks.prepareForAppServerReplacement(); } await this.finishAllAsyncTasks("stopped", "before the provider restart"); @@ -1097,6 +1105,7 @@ export class CodexAcpServer { const generation = ++this.codexProcessGeneration; process.once("exit", () => { if (generation !== this.codexProcessGeneration) return; + this.asyncQuestions.cancelAll(); void this.finishAllAsyncTasks("failed", "after the Codex process exited"); }); } @@ -1483,10 +1492,10 @@ export class CodexAcpServer { * new one ("startedNewTurn"), or could not be applied ("failed"); see * {@link performSteeringRequest}. */ - async executeOrQueueSteeringRequest(params: SessionSteerRequest): Promise { + async executeOrQueueSteeringRequest(params: SessionSteerRequest, signal?: AbortSignal): Promise { const queue = this.getSteeringQueue(params.sessionId); try { - return await queue.enqueue(params); + return await queue.enqueue(params, signal); } catch (error) { if (error instanceof RequestError) { throw error; @@ -1510,7 +1519,7 @@ export class CodexAcpServer { private getSteeringQueue(sessionId: string): SteeringQueue { let queue = this.steeringQueues.get(sessionId); if (!queue) { - queue = new SteeringQueue((params) => this.performSteeringRequest(params)); + queue = new SteeringQueue((params, signal) => this.performSteeringRequest(params, signal)); this.steeringQueues.set(sessionId, queue); } return queue; @@ -1524,7 +1533,8 @@ export class CodexAcpServer { * @returns "injected" when the prompt joined an existing turn, otherwise the * outcome of starting a new turn. */ - private async performSteeringRequest(params: SessionSteerRequest): Promise { + private async performSteeringRequest(params: SessionSteerRequest, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); logger.log("Steering session requested", { sessionId: params.sessionId, prompt: params.prompt, @@ -1533,6 +1543,7 @@ export class CodexAcpServer { this.assertSteerInputSupported(params, sessionState); const turnId = await this.getSteerableTurnId(sessionState); + signal?.throwIfAborted(); if (turnId) { const injected = await this.injectSteerIntoActiveTurn(params, turnId, sessionState); if (injected) { @@ -1540,7 +1551,8 @@ export class CodexAcpServer { return {outcome: "injected"}; } } - return await this.startNewTurnFromSteering(params); + signal?.throwIfAborted(); + return await this.startNewTurnFromSteering(params, signal); } /** @@ -1599,8 +1611,11 @@ export class CodexAcpServer { * @returns "startedNewTurn" once the turn is running; throws if the prompt * fails or is cancelled before the turn starts. */ - private async startNewTurnFromSteering(params: SessionSteerRequest): Promise { - await this.startNewTurnFromExternalPrompt(params, "Steering"); + private async startNewTurnFromSteering(params: SessionSteerRequest, signal?: AbortSignal): Promise { + await this.startNewTurnFromExternalPrompt(params, "Steering", async () => { + signal?.throwIfAborted(); + return true; + }, signal); return {outcome: "startedNewTurn"}; } @@ -1630,6 +1645,7 @@ export class CodexAcpServer { params: acp.PromptRequest, source: string, canStart: () => Promise = async () => true, + signal?: AbortSignal, ): Promise { // A prompt can outlive its turn while post-turn cleanup runs. Starting a // control-triggered turn during that window would run two prompts on the @@ -1645,7 +1661,7 @@ export class CodexAcpServer { return await new Promise((resolve, reject) => { let turnStarted = false; - const promptDone = this.prompt(params, undefined, () => { + const promptDone = this.prompt(params, signal, () => { turnStarted = true; logger.log(`${source} started a new turn`, {sessionId: params.sessionId}); // The new turn is now running. This is the success path: answer the @@ -2806,6 +2822,10 @@ export class CodexAcpServer { activePrompt.signal, ); const observeInteraction = async (event: ServerNotification): Promise => { + if (!activePrompt.signal.aborted && !this.sessionIsClosing(params.sessionId) + && "threadId" in event.params && event.params.threadId === params.sessionId) { + this.asyncQuestions.handleNotification(event, this.clientCapabilities); + } permissionContext.handleNotification(event); await elicitationHandler.handleNotification(event); }; @@ -3327,6 +3347,7 @@ export class CodexAcpServer { } async cancel(params: acp.CancelNotification): Promise { + this.asyncQuestions.cancelSession(params.sessionId); const sessionState = this.sessions.get(params.sessionId); if (!sessionState) { logger.log("Cancel request rejected: session not found", {sessionId: params.sessionId}); diff --git a/src/CodexAsyncQuestionHandler.ts b/src/CodexAsyncQuestionHandler.ts new file mode 100644 index 00000000..2927e81e --- /dev/null +++ b/src/CodexAsyncQuestionHandler.ts @@ -0,0 +1,109 @@ +import type {ClientCapabilities} from "@agentclientprotocol/sdk"; +import {ACPSessionConnection, type AcpClientConnection} from "./ACPSessionConnection"; +import type {ServerNotification} from "./app-server"; +import type {SessionSteerRequest, SessionSteeringResponse} from "./AcpExtensions"; +import { + ASYNC_QUESTION_REQUEST_METHOD, + clientSupportsAsyncQuestions, + type AsyncQuestionRequest, + type AsyncQuestionResponse, +} from "./AsyncQuestionExtension"; +import {logger} from "./Logger"; + +type QuestionSession = { + seen: Set; + pending: Set; +}; + +/** Owns questions across prompt boundaries. Never await user interaction on the notification queue. */ +export class CodexAsyncQuestionHandler { + private readonly sessions = new Map(); + + constructor( + private readonly connection: AcpClientConnection, + private readonly deliver: (request: SessionSteerRequest, signal: AbortSignal) => Promise, + ) {} + + handleNotification(notification: ServerNotification, capabilities: ClientCapabilities | null): void { + if (notification.method !== "item/completed" || !clientSupportsAsyncQuestions(capabilities)) return; + const {threadId, turnId, item} = notification.params; + if (item.type !== "agentMessage" || item.delivery !== "async" || !item.questions?.length) return; + + let session = this.sessions.get(threadId); + if (!session) { + session = {seen: new Set(), pending: new Set()}; + this.sessions.set(threadId, session); + } + if (session.seen.has(item.id)) return; + session.seen.add(item.id); + const controller = new AbortController(); + session.pending.add(controller); + const request: AsyncQuestionRequest = { + sessionId: threadId, + turnId, + itemId: item.id, + questions: item.questions.map((question, index) => ({ + id: JSON.stringify(["request_user_input_async", item.id, index]), + title: question.title, + ...(question.options ? {options: question.options} : {}), + })), + }; + void this.ask(request, controller.signal).catch(async error => { + if (controller.signal.aborted) return; + logger.error("Async question request or answer delivery failed", error); + await new ACPSessionConnection(this.connection, threadId).update({ + sessionUpdate: "agent_message_chunk", + content: {type: "text", text: "Could not complete the question interaction. Please send your answer in chat."}, + }); + }).catch(error => logger.error("Failed to report async question error", error)) + .finally(() => session.pending.delete(controller)); + } + + cancelSession(sessionId: string): void { + const session = this.sessions.get(sessionId); + if (!session) return; + for (const controller of session.pending) controller.abort(); + session.pending.clear(); + } + + cancelAll(): void { + for (const sessionId of this.sessions.keys()) this.cancelSession(sessionId); + } + + closeSession(sessionId: string): void { + this.cancelSession(sessionId); + this.sessions.delete(sessionId); + } + + private async ask(request: AsyncQuestionRequest, signal: AbortSignal): Promise { + const response = await this.connection.request( + ASYNC_QUESTION_REQUEST_METHOD, request, {cancellationSignal: signal}, + ); + if (signal.aborted) return; + // Extension responses are untrusted wire data, even with a typed SDK call. + if (response?.status === "dismissed") return; + if (response?.status !== "answered" || !Array.isArray(response.answers) + || response.answers.length !== request.questions.length) { + throw new Error("Invalid async question response"); + } + const answers = new Map(); + for (const answer of response.answers) { + if (!answer || typeof answer.id !== "string" || typeof answer.answer !== "string" + || !answer.answer.trim() || answers.has(answer.id) + || !request.questions.some(question => question.id === answer.id)) { + throw new Error("Invalid async question answer"); + } + answers.set(answer.id, answer.answer); + } + const replies = request.questions.map(question => ({ + questionItemId: question.id, + question: question.title, + answer: answers.get(question.id)!, + })); + const result = await this.deliver({ + sessionId: request.sessionId, + prompt: [{type: "text", text: `\n${JSON.stringify(replies)}\n`}], + }, signal); + if (!signal.aborted && result.outcome === "failed") throw new Error("Could not deliver async question answer"); + } +} diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 7b567541..0a086195 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -225,6 +225,7 @@ export class CodexEventHandler { private readonly seenReasoningDeltaItemIds = new Set(); private readonly terminalCommandIds = new Set(); private readonly terminalCommandOutputIds = new Set(); + private readonly emittedAgentMessageIds = new Set(); private readonly agentMessagePhases = new Map(); private readonly subagents: CodexSubagentEventRouter; /** Connection-level `authStatus` sink; the app-server account push feeds it. */ @@ -460,6 +461,7 @@ export class CodexEventHandler { */ switch (notification.method) { case "item/agentMessage/delta": + this.emittedAgentMessageIds.add(notification.params.itemId); this.completeRetryIncidentOnTurnProgress(); return await this.createTextEvent(notification.params); case "item/plan/delta": @@ -802,6 +804,12 @@ export class CodexEventHandler { return this.subagents.legacyCollaborationCompleted(event.item); case "agentMessage": this.rememberAgentMessagePhase(event.item); + // Async questions can arrive as a completed item without any text deltas. + if (event.item.delivery === "async" && !this.emittedAgentMessageIds.has(event.item.id)) { + this.emittedAgentMessageIds.add(event.item.id); + return createAgentTextMessageChunk(event.item.text, event.item.id, + createCodexMessagePhaseMeta(event.item.phase)); + } return null; case "plan": { const deltaText = this.planDeltaTextByItemId.get(event.item.id) ?? ""; diff --git a/src/SteeringQueue.ts b/src/SteeringQueue.ts index c1f553b1..75877e63 100644 --- a/src/SteeringQueue.ts +++ b/src/SteeringQueue.ts @@ -2,6 +2,7 @@ import type {SessionSteerRequest, SessionSteeringResponse} from "./AcpExtensions interface QueuedSteering { params: SessionSteerRequest; + signal: AbortSignal | undefined; resolve: (response: SessionSteeringResponse) => void; reject: (error: unknown) => void; } @@ -16,12 +17,12 @@ export class SteeringQueue { private processing = false; constructor( - private readonly handle: (params: SessionSteerRequest) => Promise, + private readonly handle: (params: SessionSteerRequest, signal?: AbortSignal) => Promise, ) {} - enqueue(params: SessionSteerRequest): Promise { + enqueue(params: SessionSteerRequest, signal?: AbortSignal): Promise { return new Promise((resolve, reject) => { - this.pending.push({params, resolve, reject}); + this.pending.push({params, signal, resolve, reject}); this.startConsumer(); }); } @@ -44,7 +45,7 @@ export class SteeringQueue { while (this.pending.length > 0) { const next = this.pending.shift()!; try { - next.resolve(await this.handle(next.params)); + next.resolve(await this.handle(next.params, next.signal)); } catch (error) { next.reject(error); // one failed steer must not stall the rest } diff --git a/src/__tests__/CodexACPAgent/async-questions.test.ts b/src/__tests__/CodexACPAgent/async-questions.test.ts new file mode 100644 index 00000000..0aa1bcaa --- /dev/null +++ b/src/__tests__/CodexACPAgent/async-questions.test.ts @@ -0,0 +1,208 @@ +import {describe, expect, it, vi} from "vitest"; +import * as acp from "@agentclientprotocol/sdk"; +import {createCodexMockTestFixture, createTestSessionState} from "../acp-test-utils"; +import {ASYNC_QUESTION_REQUEST_METHOD, ASYNC_QUESTIONS_CAPABILITY} from "../../AsyncQuestionExtension"; +import type {AsyncQuestionRequest, AsyncQuestionResponse} from "../../AsyncQuestionExtension"; +import type {Turn, TurnCompletedNotification} from "../../app-server/v2"; + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise(done => { resolve = done; }); + return {promise, resolve}; +} + +function turn(id: string, status: Turn["status"]): Turn { + return {id, status, items: [], itemsView: "notLoaded", error: null, startedAt: null, completedAt: null, durationMs: null}; +} + +async function setup(version: unknown = 1) { + const fixture = createCodexMockTestFixture(); + const agent = fixture.getCodexAcpAgent(); + const appServer = fixture.getCodexAppServerClient(); + const session = createTestSessionState({sessionId: "session-id"}); + vi.spyOn(agent, "getSessionState").mockReturnValue(session); + const initialized = await agent.initialize({protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {_meta: {[ASYNC_QUESTIONS_CAPABILITY]: {version}}}, + }); + const completion = deferred(); + const nextCompletion = deferred(); + const start = vi.spyOn(appServer, "turnStart") + .mockResolvedValueOnce({turn: turn("turn-1", "inProgress")}) + .mockResolvedValue({turn: turn("turn-2", "inProgress")}); + vi.spyOn(appServer, "awaitTurnCompleted").mockReturnValueOnce(completion.promise).mockReturnValue(nextCompletion.promise); + const steer = vi.spyOn(appServer, "turnSteer").mockResolvedValue({turnId: "turn-1"}); + const requestSignals: AbortSignal[] = []; + fixture.onAcpConnectionEvent(event => { + if (event.method === "request" && event.args[0] === ASYNC_QUESTION_REQUEST_METHOD) { + requestSignals.push(event.args[2].cancellationSignal); + } + }); + const response = deferred(); + fixture.setExtensionResponse(ASYNC_QUESTION_REQUEST_METHOD, response.promise); + const prompt = agent.prompt({sessionId: session.sessionId, prompt: [{type: "text", text: "Do some work"}]}); + await vi.waitFor(() => expect(session.currentTurnId).toBe("turn-1")); + const questions = [ + {title: "Есть номер YouTrack-задачи?", options: null}, + {title: "Which scope?", options: ["Platform", "Plugin"]}, + ]; + const item = {type: "agentMessage", id: "question-call", text: questions.map(q => q.title).join("\n"), + phase: "final_answer", memoryCitation: null, delivery: "async", questions}; + async function sendQuestion() { + fixture.sendServerNotification({method: "item/completed", params: {threadId: session.sessionId, turnId: "turn-1", item}}); + await fixture.getCodexAcpClient().waitForSessionNotifications(session.sessionId); + } + function requests() { + return fixture.getAcpConnectionEvents([]).filter(e => e.method === "request" && e.args[0] === ASYNC_QUESTION_REQUEST_METHOD); + } + function answer() { + const request = requests()[0]!.args[1] as AsyncQuestionRequest; + response.resolve({status: "answered", answers: request.questions.map((q, i) => ({id: q.id, + answer: i === 0 ? "давай создай задачу" : "A custom scope"}))}); + } + async function finish() { + completion.resolve({threadId: session.sessionId, turn: turn("turn-1", "completed")}); + await prompt; + } + return {fixture, agent, session, initialized, response, requestSignals, start, steer, sendQuestion, requests, answer, finish, nextCompletion}; +} + +describe("asynchronous user questions", () => { + it("negotiates the extension, keeps streaming, deduplicates questions, and steers the answer", async () => { + const f = await setup(); + await f.sendQuestion(); + await f.sendQuestion(); + f.fixture.sendServerNotification({method: "item/agentMessage/delta", params: { + threadId: "session-id", turnId: "turn-1", itemId: "progress", delta: "Working while the question is open", + }}); + await f.fixture.getCodexAcpClient().waitForSessionNotifications("session-id"); + expect(f.requests()).toHaveLength(1); + expect(f.steer).not.toHaveBeenCalled(); + f.answer(); + await vi.waitFor(() => expect(f.steer).toHaveBeenCalledTimes(1)); + await expect(JSON.stringify({ + capability: f.initialized.agentCapabilities?._meta?.[ASYNC_QUESTIONS_CAPABILITY], + request: f.requests()[0]!.args.slice(0, 2), + updates: f.fixture.getAcpConnectionEvents([]).filter(e => e.method === "sessionUpdate" + && e.args[0].update.sessionUpdate === "agent_message_chunk"), + steer: f.steer.mock.calls[0], + }, null, 2)).toMatchFileSnapshot("./snapshots/async-questions-active.json"); + await f.finish(); + }); + + it("keeps the request alive after completion and starts a new turn with the answer", async () => { + const f = await setup(); + await f.sendQuestion(); + await f.finish(); + expect(f.requestSignals[0]!.aborted).toBe(false); + f.answer(); + await vi.waitFor(() => expect(f.start).toHaveBeenCalledTimes(2)); + expect(f.steer).not.toHaveBeenCalled(); + await expect(JSON.stringify(f.start.mock.calls[1]![0].input, null, 2)) + .toMatchFileSnapshot("./snapshots/async-questions-late-input.json"); + f.nextCompletion.resolve({threadId: "session-id", turn: turn("turn-2", "completed")}); + await vi.waitFor(() => expect(f.session.currentTurnId).toBeNull()); + }); + + it.each([undefined, 0, 2, "1"])("falls back to text when version %s is not negotiated", async version => { + const f = await setup(version === undefined ? null : version); + await f.sendQuestion(); + expect(f.requests()).toHaveLength(0); + expect(f.fixture.getAcpConnectionEvents([]).some(e => e.method === "sessionUpdate" + && e.args[0].update.content?.text === "Есть номер YouTrack-задачи?\nWhich scope?")).toBe(true); + await f.finish(); + }); + + it("does not repeat question text that already arrived as a delta", async () => { + const f = await setup(); + f.fixture.sendServerNotification({method: "item/agentMessage/delta", params: { + threadId: "session-id", turnId: "turn-1", itemId: "question-call", delta: "Already streamed question", + }}); + await f.sendQuestion(); + const textEvents = f.fixture.getAcpConnectionEvents([]).filter(e => e.method === "sessionUpdate" + && e.args[0].update.messageId === "question-call"); + expect(textEvents).toHaveLength(1); + expect(textEvents[0]!.args[0].update.content.text).toBe("Already streamed question"); + expect(f.requests()).toHaveLength(1); + f.response.resolve({status: "dismissed"}); + await f.finish(); + }); + + it("uses a new turn if the active turn finishes during answer delivery", async () => { + const f = await setup(); + f.steer.mockImplementationOnce(async () => { + await f.finish(); + throw new Error("no active turn to steer"); + }); + await f.sendQuestion(); + f.answer(); + await vi.waitFor(() => expect(f.start).toHaveBeenCalledTimes(2)); + expect(f.steer).toHaveBeenCalledTimes(1); + expect(f.start.mock.calls[1]![0].input).toEqual(f.steer.mock.calls[0]![0].input); + f.nextCompletion.resolve({threadId: "session-id", turn: turn("turn-2", "completed")}); + await vi.waitFor(() => expect(f.session.currentTurnId).toBeNull()); + }); + + it("cancels an answer waiting behind another steering request", async () => { + const f = await setup(); + const blocked = deferred<{turnId: string}>(); + f.steer.mockReturnValueOnce(blocked.promise); + const first = f.agent.executeOrQueueSteeringRequest({sessionId: "session-id", + prompt: [{type: "text", text: "Other input"}]}); + await vi.waitFor(() => expect(f.steer).toHaveBeenCalledTimes(1)); + await f.sendQuestion(); + const enqueued = vi.spyOn(f.agent, "executeOrQueueSteeringRequest"); + f.answer(); + await vi.waitFor(() => expect(enqueued).toHaveBeenCalledTimes(1)); + await f.agent.cancel({sessionId: "session-id"}); + blocked.resolve({turnId: "turn-1"}); + await first; + await enqueued.mock.results[0]!.value; + expect(f.steer).toHaveBeenCalledTimes(1); + expect(f.start).toHaveBeenCalledTimes(1); + await f.finish(); + }); + + it("reports a delivery failure without retrying the answer", async () => { + const f = await setup(); + f.steer.mockRejectedValue(new Error("Transport failure")); + await f.sendQuestion(); + f.answer(); + await vi.waitFor(() => expect(f.fixture.getAcpConnectionEvents([]).some(e => e.method === "sessionUpdate" + && e.args[0].update.content?.text?.includes("Please send your answer in chat"))).toBe(true)); + expect(f.steer).toHaveBeenCalledTimes(1); + expect(f.start).toHaveBeenCalledTimes(1); + await f.finish(); + }); + + it.each(["dismiss", "cancel", "close"])("does not submit input after %s", async action => { + const f = await setup(); + await f.sendQuestion(); + await f.finish(); + if (action === "dismiss") { + f.response.resolve({status: "dismissed"}); + } else { + if (action === "cancel") await f.agent.cancel({sessionId: "session-id"}); + else await f.agent.closeSession({sessionId: "session-id"}); + expect(f.requestSignals[0]!.aborted).toBe(true); + // A client may ignore RPC cancellation and still return a result. + f.answer(); + } + await new Promise(resolve => setImmediate(resolve)); + expect(f.steer).not.toHaveBeenCalled(); + expect(f.start).toHaveBeenCalledTimes(1); + }); + + it.each([ + {status: "answered", answers: []}, + {status: "answered", answers: [{id: "unknown", answer: "a"}, {id: "unknown", answer: "b"}]}, + {status: "unexpected"}, + ])("reports invalid responses without submitting input: %j", async response => { + const f = await setup(); + await f.sendQuestion(); + f.response.resolve(response as AsyncQuestionResponse); + await vi.waitFor(() => expect(f.fixture.getAcpConnectionEvents([]).some(e => e.method === "sessionUpdate" + && e.args[0].update.content?.text?.includes("Please send your answer in chat"))).toBe(true)); + expect(f.steer).not.toHaveBeenCalled(); + await f.finish(); + }); +}); diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 7fff7721..c63863ce 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -63,6 +63,7 @@ describe('CodexACPAgent - initialize', () => { }, _meta: { authStatus: {}, + "codex.asyncQuestions": {version: 1, requestMethod: "_codex/requestUserInput"}, }, }, authMethods: getCodexAuthMethods(), diff --git a/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json b/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json new file mode 100644 index 00000000..5364da21 --- /dev/null +++ b/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json @@ -0,0 +1,80 @@ +{ + "capability": { + "version": 1, + "requestMethod": "_codex/requestUserInput" + }, + "request": [ + "_codex/requestUserInput", + { + "sessionId": "session-id", + "turnId": "turn-1", + "itemId": "question-call", + "questions": [ + { + "id": "[\"request_user_input_async\",\"question-call\",0]", + "title": "Есть номер YouTrack-задачи?" + }, + { + "id": "[\"request_user_input_async\",\"question-call\",1]", + "title": "Which scope?", + "options": [ + "Platform", + "Plugin" + ] + } + ] + } + ], + "updates": [ + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "session-id", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "question-call", + "content": { + "type": "text", + "text": "Есть номер YouTrack-задачи?\nWhich scope?" + }, + "_meta": { + "codex": { + "phase": "final_answer" + } + } + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "session-id", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "progress", + "content": { + "type": "text", + "text": "Working while the question is open" + } + } + } + ] + } + ], + "steer": [ + { + "threadId": "session-id", + "expectedTurnId": "turn-1", + "input": [ + { + "type": "text", + "text": "\n[{\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"question-call\\\",0]\",\"question\":\"Есть номер YouTrack-задачи?\",\"answer\":\"давай создай задачу\"},{\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"question-call\\\",1]\",\"question\":\"Which scope?\",\"answer\":\"A custom scope\"}]\n", + "text_elements": [] + } + ] + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/snapshots/async-questions-late-input.json b/src/__tests__/CodexACPAgent/snapshots/async-questions-late-input.json new file mode 100644 index 00000000..568896d9 --- /dev/null +++ b/src/__tests__/CodexACPAgent/snapshots/async-questions-late-input.json @@ -0,0 +1,7 @@ +[ + { + "type": "text", + "text": "\n[{\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"question-call\\\",0]\",\"question\":\"Есть номер YouTrack-задачи?\",\"answer\":\"давай создай задачу\"},{\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"question-call\\\",1]\",\"question\":\"Which scope?\",\"answer\":\"A custom scope\"}]\n", + "text_elements": [] + } +] \ No newline at end of file diff --git a/src/__tests__/acp-test-utils.ts b/src/__tests__/acp-test-utils.ts index f358664c..4a77a657 100644 --- a/src/__tests__/acp-test-utils.ts +++ b/src/__tests__/acp-test-utils.ts @@ -259,6 +259,7 @@ export function removeDirectoryWithRetry(directory: string): void { export interface CodexMockTestFixture extends TestFixture { sendServerNotification(notification: ServerNotification | Record): void, sendServerRequest(method: string, params: unknown): Promise, + setExtensionResponse(method: string, response: unknown): void, setPermissionResponse(response: RequestPermissionResponse | Promise): void, setElicitationResponse(response: CreateElicitationResponse | Promise): void, } @@ -276,6 +277,7 @@ export function createCodexMockTestFixture( ): CodexMockTestFixture { let unhandledNotificationHandler: ((notification: any) => void) | null = null; const requestHandlers = new Map Promise>(); + const extensionResponses = new Map(); // State for controlling permission responses const permissionState: { response: RequestPermissionResponse | Promise } = { @@ -302,6 +304,7 @@ export function createCodexMockTestFixture( const acpEventHandlers: ((event: MethodCallEvent) => void)[] = []; const returnValues = new Map any>(); returnValues.set('request', (args) => { + if (extensionResponses.has(args[0])) return extensionResponses.get(args[0]); if (args[0] === acp.methods.client.session.requestPermission) { return permissionState.response; } @@ -341,6 +344,9 @@ export function createCodexMockTestFixture( return { ...baseFixture, + setExtensionResponse(method: string, response: unknown): void { + extensionResponses.set(method, response); + }, sendServerNotification(notification: ServerNotification | Record): void { if (unhandledNotificationHandler) { unhandledNotificationHandler(notification); From f03f6985e7c8e47d9d8923e1a16fd438dda52424 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Sun, 6 Sep 2026 13:15:54 +0400 Subject: [PATCH 2/5] refactor: use AIR extension for asynchronous questions --- docs/ASYNC_QUESTIONS_PROPOSAL.md | 183 ------------------ docs/async-questions.md | 107 ++++++++++ readme-dev.md | 4 +- src/AcpExtensions.ts | 2 - src/AirExtension.ts | 1 + src/AsyncQuestionExtension.ts | 18 +- src/CodexAcpServer.ts | 4 +- src/CodexAsyncQuestionHandler.ts | 4 +- .../CodexACPAgent/async-questions.test.ts | 26 ++- .../CodexACPAgent/initialize.test.ts | 3 +- .../snapshots/async-questions-active.json | 14 +- 11 files changed, 148 insertions(+), 218 deletions(-) delete mode 100644 docs/ASYNC_QUESTIONS_PROPOSAL.md create mode 100644 docs/async-questions.md diff --git a/docs/ASYNC_QUESTIONS_PROPOSAL.md b/docs/ASYNC_QUESTIONS_PROPOSAL.md deleted file mode 100644 index e6dcf3a9..00000000 --- a/docs/ASYNC_QUESTIONS_PROPOSAL.md +++ /dev/null @@ -1,183 +0,0 @@ -# Proposal: asynchronous user questions over ACP - -Status: experimental provider extension, version 1, implemented by `codex-acp`. -This document proposes a client contract; it does not add a standard ACP method. - -## Problem and intended behavior - -Codex can ask a question and continue working before the user answers. For example, -it asks for a YouTrack issue number while investigating a fix. The user answers -“create an issue” in a question form, and Codex receives that answer as new user input. - -The ACP client receives one ordinary request/response RPC for the question. The RPC -waits for the user, but neither the session notification queue nor the running Codex -turn waits for it. Answering after the original turn finishes can start another turn. - -This is separate from synchronous Codex `item/tool/requestUserInput`, which continues -to use standard ACP elicitation and returns a tool response to the waiting Codex call. - -## Capability negotiation - -A client opts in through `initialize.params.clientCapabilities._meta`: - -```json -{ - "clientCapabilities": { - "_meta": { - "codex.asyncQuestions": { "version": 1 } - } - } -} -``` - -The provider advertises its implementation in -`initialize.result.agentCapabilities._meta`: - -```json -{ - "codex.asyncQuestions": { - "version": 1, - "requestMethod": "_codex/requestUserInput" - } -} -``` - -Version must be the number `1`. Missing, malformed, or unsupported versions receive -ordinary question text without an extension request. Standard form elicitation -support does not implicitly enable this feature. No environment setting is needed. - -## Provider-to-client question request - -```json -{ - "jsonrpc": "2.0", - "id": 42, - "method": "_codex/requestUserInput", - "params": { - "sessionId": "session-1", - "turnId": "turn-1", - "itemId": "call-1", - "questions": [ - { - "id": "[\"request_user_input_async\",\"call-1\",0]", - "title": "Is there a YouTrack issue for this fix?" - }, - { - "id": "[\"request_user_input_async\",\"call-1\",1]", - "title": "Which component?", - "options": ["Platform", "Plugin"] - } - ] - } -} -``` - -The client should: - -- Present all questions together, associated with the specified session and item. -- Always allow free text. `options` are suggestions, not an enum restricting answers. -- Omit automatic submission or selection. Preserve the user's submitted text. -- Keep rendering session updates and accepting other input while the RPC is pending. -- Keep the form available after the originating `session/prompt` completes. -- Associate the form with the ordinary `agent_message_chunk` whose `messageId` equals - `itemId`, so the transcript and form do not appear to be unrelated questions. - -Question IDs are opaque strings. Return them unchanged. `turnId` identifies the -origin of the question; it does not require the answer to reach that same turn. - -## Client response - -On submission, return exactly one nonblank string answer for each question: - -```json -{ - "jsonrpc": "2.0", - "id": 42, - "result": { - "status": "answered", - "answers": [ - { "id": "[\"request_user_input_async\",\"call-1\",0]", "answer": "Create an issue" }, - { "id": "[\"request_user_input_async\",\"call-1\",1]", "answer": "Platform" } - ] - } -} -``` - -Order is not significant. Unknown or duplicate IDs, missing answers, non-string -values, and blank answers invalidate the entire response; nothing is submitted. -The client may instead close the form without sending user input: - -```json -{ "jsonrpc": "2.0", "id": 42, "result": { "status": "dismissed" } } -``` - -The client records the submitted answer in its UI. It must not also send -`session/prompt` or `_session/steering` for that answer: the provider handles delivery. -The RPC result is not an acknowledgement that Codex has consumed the answer. - -## Codex mapping and input delivery - -The adapter observes live `item/completed` notifications with an `agentMessage` -whose `delivery` is `"async"` and whose `questions` array is nonempty. Each question -is mapped to the custom request above. There is no additional app-server request -to register: this is a completed message event, not `item/tool/requestUserInput`. - -Codex's observed `request_user_input_async` call returns `{"accepted":true}` immediately. -The later answer is a user message with this payload: - -```text - -[{"questionItemId":"[\"request_user_input_async\",\"call-1\",0]","question":"Is there a YouTrack issue for this fix?","answer":"Create an issue"},{"questionItemId":"[\"request_user_input_async\",\"call-1\",1]","question":"Which component?","answer":"Platform"}] - -``` - -The adapter constructs this payload from the original request and validated answers. -Clients do not construct it. This wrapper follows the observed Codex desktop format; -it is a Codex-specific compatibility detail, not a portable ACP standard. - -The existing per-session steering queue delivers the input: - -1. An active turn receives `turn/steer` with its current `expectedTurnId`. -2. If the turn has finished, the adapter waits for prompt cleanup and uses `turn/start`. -3. A “no active turn” race follows the existing steering fallback to a new turn. - -Concurrent answers and other steering requests share the same queue. Output from a -new turn streams through ordinary ACP session updates even though no client -`session/prompt` request is outstanding. Clients opting in must support that lifecycle. - -## Cancellation, history, and failure - -- A pending question survives normal turn completion. There is no answer timeout. -- `session/cancel`, session close/delete, and provider replacement cancel outstanding - question RPCs through ACP `$/cancel_request`. The client should close the form and - settle its RPC. A late response to a cancelled question is ignored. -- Cancellation also prevents an answer still queued for delivery from starting work. - Input already accepted by Codex cannot be retracted by dismissing the form. -- Repeated live events with the same item ID create at most one request per loaded - session. Different sessions have independent question IDs and pending requests. -- Loading or forking history displays text only; it does not reopen historical forms. - Pending forms are not persisted across adapter restart or session close/reopen. -- Without the capability, a completed async message is rendered as ordinary text, - including messages that arrive without text deltas. The user can reply in chat. -- RPC errors, malformed responses, or failed answer delivery are logged and produce - a visible request to send the answer in chat. There is no automatic retry that could - duplicate an answer after an uncertain transport outcome. - -## Implementation and validation - -`AsyncQuestionExtension.ts` defines the wire types. `CodexAsyncQuestionHandler.ts` -owns pending questions across prompt boundaries. `CodexAcpServer.ts` connects live -events, session cancellation, and the existing `SteeringQueue`. `CodexEventHandler.ts` -provides ordinary text rendering for completed async messages without deltas. - -Behavior tests in `src/__tests__/CodexACPAgent/async-questions.test.ts` cover capability -negotiation, nonblocking progress, deduplication, multiple questions and free text, -active-turn delivery, late-answer turn creation, cancellation, and invalid responses. -File snapshots record the ACP request and exact Codex input payload. - -## Future standardization - -A standard ACP proposal could generalize this request without exposing Codex-specific -IDs or the input wrapper. Version 1 deliberately leaves durable pending-question -recovery and explicit delivery acknowledgements for a future revision. Client-side -form rendering must be implemented by each ACP client before advertising support. diff --git a/docs/async-questions.md b/docs/async-questions.md new file mode 100644 index 00000000..15751ab7 --- /dev/null +++ b/docs/async-questions.md @@ -0,0 +1,107 @@ +# Asynchronous user questions + +Codex can ask a question and continue working before the user answers. The adapter exposes these questions through the AIR `asyncQuestions` extension. + +The client receives a request that waits for the user's answer. The running turn and session updates continue while that request is pending. The adapter sends the answer to Codex as new user input. + +## Negotiation + +The client adds `asyncQuestions` to `clientCapabilities._meta.jetbrains.air.capabilities` during `initialize`: + +```json +{ + "clientCapabilities": { + "_meta": { + "jetbrains": { + "air": { + "version": 1, + "capabilities": ["asyncQuestions"] + } + } + } + } +} +``` + +The adapter advertises the same capability in `initialize.result._meta.jetbrains.air.capabilities`. This uses the shared AIR extension version and capability check. + +Without negotiation, the adapter displays the question as ordinary text. The user can answer in chat. Standard ACP elicitation support does not enable this extension. + +## Question request + +The adapter sends `_session/async_question/request` to the client: + +```json +{ + "sessionId": "thread-id", + "turnId": "turn-id", + "itemId": "call-id", + "questions": [ + { + "id": "[\"request_user_input_async\",\"call-id\",0]", + "title": "Is there a YouTrack issue for this fix?" + }, + { + "id": "[\"request_user_input_async\",\"call-id\",1]", + "title": "Which component?", + "options": ["Platform", "Plugin"] + } + ] +} +``` + +The client displays all questions together. It always allows free text; `options` are suggestions. It must not submit a preselected option automatically. + +The client associates the form with the ordinary `agent_message_chunk` whose `messageId` equals `itemId`. It keeps processing session updates and other input while the request waits. Normal turn completion does not close the form. + +Question IDs are opaque strings. The client returns them unchanged. `turnId` identifies the originating turn, not necessarily the turn that receives the answer. + +## Answer response + +The client returns one nonblank string answer for each question: + +```json +{ + "status": "answered", + "answers": [ + {"id": "[\"request_user_input_async\",\"call-id\",0]", "answer": "Create an issue"}, + {"id": "[\"request_user_input_async\",\"call-id\",1]", "answer": "Platform"} + ] +} +``` + +Answer order is not significant. Missing answers, unknown or duplicate IDs, non-string values, and blank answers invalidate the whole response. Closing the form returns `{ "status": "dismissed" }` and sends no input. + +The client records the submitted answer in its UI. It must not also send `session/prompt` or `_session/steering` for that answer. The adapter owns delivery; the question RPC response does not acknowledge that Codex consumed the answer. + +## Input delivery + +The adapter reads live `item/completed` events with `agentMessage.delivery: "async"` and a nonempty `questions` array. It sends the client request without blocking the event queue. + +After a valid response, it constructs a user message in the observed Codex desktop format: + +```text + +[{"questionItemId":"[\"request_user_input_async\",\"call-id\",0]","question":"Is there a YouTrack issue for this fix?","answer":"Create an issue"},{"questionItemId":"[\"request_user_input_async\",\"call-id\",1]","question":"Which component?","answer":"Platform"}] + +``` + +This wrapper is a Codex compatibility detail. The client does not construct it. + +The existing steering queue sends the message through `turn/steer` when a turn is active. Otherwise it waits for prompt cleanup and starts a new turn. If the active turn finishes during delivery, the existing steering fallback starts a new turn. + +Answers share the queue with other steering requests. A new turn streams ordinary ACP updates even when no client `session/prompt` request is outstanding. Clients advertising this extension must support that lifecycle. + +Synchronous Codex `item/tool/requestUserInput` still uses standard ACP elicitation and returns its answer to the waiting tool call. + +## Cancellation and failure + +There is no answer timeout. Session cancellation, close/delete, provider replacement, and Codex process exit cancel pending question RPCs through ACP `$/cancel_request`. The client closes the form and settles its request. Late responses are ignored, and cancelled answers waiting in the steering queue cannot start work. Input already accepted by Codex cannot be retracted by dismissing the form. + +Request errors, invalid responses, and failed delivery produce a visible message asking the user to answer in chat. The adapter does not automatically retry an uncertain delivery. + +## Session load + +Repeated live events with the same item ID create at most one request per loaded session. Sessions track their questions independently. + +Loading or forking history displays question text without reopening forms. Pending forms are not restored after adapter restart or session close/reopen. Durable recovery and delivery acknowledgements are outside this version of the extension. diff --git a/readme-dev.md b/readme-dev.md index edd8774d..cf4f1568 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -3,8 +3,8 @@ Set `CODEX_PATH` to run a different Codex binary; versions other than the one sp ### Runtime environment -For the opt-in client extension that displays asynchronous Codex questions and sends -answers back as user input, see [Async questions proposal](docs/ASYNC_QUESTIONS_PROPOSAL.md). +For the AIR extension that displays asynchronous Codex questions and sends +answers back as user input, see [Asynchronous user questions](docs/async-questions.md). It is negotiated through ACP capabilities and requires no environment setting. - `CODEX_API_KEY` - API key used when the API-key auth method is selected. Takes precedence over `OPENAI_API_KEY`. diff --git a/src/AcpExtensions.ts b/src/AcpExtensions.ts index 0b6cd6de..29d26f77 100644 --- a/src/AcpExtensions.ts +++ b/src/AcpExtensions.ts @@ -141,8 +141,6 @@ export async function steerSessionWithFallback( } export { - ASYNC_QUESTIONS_CAPABILITY, - ASYNC_QUESTIONS_VERSION, ASYNC_QUESTION_REQUEST_METHOD, type AsyncQuestionRequest, type AsyncQuestionResponse, diff --git a/src/AirExtension.ts b/src/AirExtension.ts index 03b12749..84d90460 100644 --- a/src/AirExtension.ts +++ b/src/AirExtension.ts @@ -15,6 +15,7 @@ export const AIR_EXTENSION_CAPABILITIES_KEY = "capabilities"; export const AIR_SESSION_FAILURE_KEY = "sessionFailure"; export const AIR_AGENT_FILE_CHANGE_REPORT_KEY = "agentFileChangeReport"; export const AIR_NATIVE_SUBAGENT_SESSIONS_KEY = "nativeSubagentSessions"; +export const AIR_ASYNC_QUESTIONS_KEY = "asyncQuestions"; export const AIR_ASYNC_TASKS_KEY = "asyncTasks"; export const AIR_ASYNC_TASKS_BACKGROUNDED_KEY = "backgrounded"; export const AIR_AGENT_FILE_CHANGE_REPORT_REQUEST_KEY = "agentFileChangeReportRequest"; diff --git a/src/AsyncQuestionExtension.ts b/src/AsyncQuestionExtension.ts index c8f862f4..c85fcf65 100644 --- a/src/AsyncQuestionExtension.ts +++ b/src/AsyncQuestionExtension.ts @@ -1,9 +1,5 @@ -import type {ClientCapabilities} from "@agentclientprotocol/sdk"; - -/** Experimental, versioned provider extension; not a standard ACP method. */ -export const ASYNC_QUESTIONS_CAPABILITY = "codex.asyncQuestions"; -export const ASYNC_QUESTIONS_VERSION = 1; -export const ASYNC_QUESTION_REQUEST_METHOD = "_codex/requestUserInput"; +/** Request/response contract for the AIR asyncQuestions capability. */ +export const ASYNC_QUESTION_REQUEST_METHOD = "_session/async_question/request"; export type AsyncQuestionRequest = { sessionId: string; @@ -15,13 +11,3 @@ export type AsyncQuestionRequest = { export type AsyncQuestionResponse = | {status: "answered"; answers: Array<{id: string; answer: string}>} | {status: "dismissed"}; - -export function asyncQuestionsCapability() { - return {version: ASYNC_QUESTIONS_VERSION, requestMethod: ASYNC_QUESTION_REQUEST_METHOD}; -} - -export function clientSupportsAsyncQuestions(capabilities: ClientCapabilities | null): boolean { - const value = capabilities?._meta?.[ASYNC_QUESTIONS_CAPABILITY]; - return typeof value === "object" && value !== null - && "version" in value && value.version === ASYNC_QUESTIONS_VERSION; -} diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 510d9a86..cda2457f 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -1,5 +1,4 @@ import {CodexAsyncQuestionHandler} from "./CodexAsyncQuestionHandler"; -import {ASYNC_QUESTIONS_CAPABILITY, asyncQuestionsCapability} from "./AsyncQuestionExtension"; import * as acp from "@agentclientprotocol/sdk"; import {RequestError, type SessionId, type SessionModeState} from "@agentclientprotocol/sdk"; import {CodexEventHandler, type CompletedPlan} from "./CodexEventHandler"; @@ -133,6 +132,7 @@ import {once} from "node:events"; import { AIR_AGENT_FILE_CHANGE_REPORT_KEY, AIR_ASYNC_TASKS_KEY, + AIR_ASYNC_QUESTIONS_KEY, AIR_NATIVE_SUBAGENT_SESSIONS_KEY, AIR_EXTENSION_CAPABILITIES_KEY, AIR_EXTENSION_VERSION, @@ -382,7 +382,6 @@ export class CodexAcpServer { // Presence means "this agent pushes `_auth/status_update`". It // never carries a payload, and the client never asks for one. [AUTH_STATUS_META_KEY]: authStatusCapability(), - [ASYNC_QUESTIONS_CAPABILITY]: asyncQuestionsCapability(), }, }, authMethods: getCodexAuthMethods(_params.clientCapabilities), @@ -403,6 +402,7 @@ export class CodexAcpServer { AIR_AGENT_FILE_CHANGE_REPORT_KEY, AIR_NATIVE_SUBAGENT_SESSIONS_KEY, AIR_ASYNC_TASKS_KEY, + AIR_ASYNC_QUESTIONS_KEY, ], }, }, diff --git a/src/CodexAsyncQuestionHandler.ts b/src/CodexAsyncQuestionHandler.ts index 2927e81e..d8638b4a 100644 --- a/src/CodexAsyncQuestionHandler.ts +++ b/src/CodexAsyncQuestionHandler.ts @@ -4,10 +4,10 @@ import type {ServerNotification} from "./app-server"; import type {SessionSteerRequest, SessionSteeringResponse} from "./AcpExtensions"; import { ASYNC_QUESTION_REQUEST_METHOD, - clientSupportsAsyncQuestions, type AsyncQuestionRequest, type AsyncQuestionResponse, } from "./AsyncQuestionExtension"; +import {AIR_ASYNC_QUESTIONS_KEY, clientSupportsAirCapability} from "./AirExtension"; import {logger} from "./Logger"; type QuestionSession = { @@ -25,7 +25,7 @@ export class CodexAsyncQuestionHandler { ) {} handleNotification(notification: ServerNotification, capabilities: ClientCapabilities | null): void { - if (notification.method !== "item/completed" || !clientSupportsAsyncQuestions(capabilities)) return; + if (notification.method !== "item/completed" || !clientSupportsAirCapability(capabilities, AIR_ASYNC_QUESTIONS_KEY)) return; const {threadId, turnId, item} = notification.params; if (item.type !== "agentMessage" || item.delivery !== "async" || !item.questions?.length) return; diff --git a/src/__tests__/CodexACPAgent/async-questions.test.ts b/src/__tests__/CodexACPAgent/async-questions.test.ts index 0aa1bcaa..5b70471a 100644 --- a/src/__tests__/CodexACPAgent/async-questions.test.ts +++ b/src/__tests__/CodexACPAgent/async-questions.test.ts @@ -1,7 +1,7 @@ import {describe, expect, it, vi} from "vitest"; import * as acp from "@agentclientprotocol/sdk"; import {createCodexMockTestFixture, createTestSessionState} from "../acp-test-utils"; -import {ASYNC_QUESTION_REQUEST_METHOD, ASYNC_QUESTIONS_CAPABILITY} from "../../AsyncQuestionExtension"; +import {ASYNC_QUESTION_REQUEST_METHOD} from "../../AsyncQuestionExtension"; import type {AsyncQuestionRequest, AsyncQuestionResponse} from "../../AsyncQuestionExtension"; import type {Turn, TurnCompletedNotification} from "../../app-server/v2"; @@ -15,14 +15,18 @@ function turn(id: string, status: Turn["status"]): Turn { return {id, status, items: [], itemsView: "notLoaded", error: null, startedAt: null, completedAt: null, durationMs: null}; } -async function setup(version: unknown = 1) { +function airCapabilities(version: unknown = 1, capabilities: unknown = ["asyncQuestions"]): acp.ClientCapabilities { + return {_meta: {jetbrains: {air: {version, capabilities}}}}; +} + +async function setup(clientCapabilities: acp.ClientCapabilities = airCapabilities()) { const fixture = createCodexMockTestFixture(); const agent = fixture.getCodexAcpAgent(); const appServer = fixture.getCodexAppServerClient(); const session = createTestSessionState({sessionId: "session-id"}); vi.spyOn(agent, "getSessionState").mockReturnValue(session); const initialized = await agent.initialize({protocolVersion: acp.PROTOCOL_VERSION, - clientCapabilities: {_meta: {[ASYNC_QUESTIONS_CAPABILITY]: {version}}}, + clientCapabilities, }); const completion = deferred(); const nextCompletion = deferred(); @@ -80,7 +84,7 @@ describe("asynchronous user questions", () => { f.answer(); await vi.waitFor(() => expect(f.steer).toHaveBeenCalledTimes(1)); await expect(JSON.stringify({ - capability: f.initialized.agentCapabilities?._meta?.[ASYNC_QUESTIONS_CAPABILITY], + capability: f.initialized._meta?.["jetbrains"], request: f.requests()[0]!.args.slice(0, 2), updates: f.fixture.getAcpConnectionEvents([]).filter(e => e.method === "sessionUpdate" && e.args[0].update.sessionUpdate === "agent_message_chunk"), @@ -103,8 +107,10 @@ describe("asynchronous user questions", () => { await vi.waitFor(() => expect(f.session.currentTurnId).toBeNull()); }); - it.each([undefined, 0, 2, "1"])("falls back to text when version %s is not negotiated", async version => { - const f = await setup(version === undefined ? null : version); + it.each([{}, airCapabilities(0), airCapabilities("1"), airCapabilities(1, []), + {_meta: {"codex.asyncQuestions": {version: 1}}}, + ])("falls back to text without AIR negotiation: %j", async capabilities => { + const f = await setup(capabilities); await f.sendQuestion(); expect(f.requests()).toHaveLength(0); expect(f.fixture.getAcpConnectionEvents([]).some(e => e.method === "sessionUpdate" @@ -112,6 +118,14 @@ describe("asynchronous user questions", () => { await f.finish(); }); + it("uses the shared AIR version compatibility rule", async () => { + const f = await setup(airCapabilities(2)); + await f.sendQuestion(); + expect(f.requests()).toHaveLength(1); + f.response.resolve({status: "dismissed"}); + await f.finish(); + }); + it("does not repeat question text that already arrived as a delta", async () => { const f = await setup(); f.fixture.sendServerNotification({method: "item/agentMessage/delta", params: { diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index c63863ce..84a4320b 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -63,7 +63,6 @@ describe('CodexACPAgent - initialize', () => { }, _meta: { authStatus: {}, - "codex.asyncQuestions": {version: 1, requestMethod: "_codex/requestUserInput"}, }, }, authMethods: getCodexAuthMethods(), @@ -79,7 +78,7 @@ describe('CodexACPAgent - initialize', () => { jetbrains: { air: { version: 1, - capabilities: ["sessionFailure", "agentFileChangeReport", "nativeSubagentSessions", "asyncTasks"], + capabilities: ["sessionFailure", "agentFileChangeReport", "nativeSubagentSessions", "asyncTasks", "asyncQuestions"], }, }, }, diff --git a/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json b/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json index 5364da21..eeebf9b8 100644 --- a/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json +++ b/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json @@ -1,10 +1,18 @@ { "capability": { - "version": 1, - "requestMethod": "_codex/requestUserInput" + "air": { + "version": 1, + "capabilities": [ + "sessionFailure", + "agentFileChangeReport", + "nativeSubagentSessions", + "asyncTasks", + "asyncQuestions" + ] + } }, "request": [ - "_codex/requestUserInput", + "_session/async_question/request", { "sessionId": "session-id", "turnId": "turn-1", From db14ae495e39c0e5aa89c446d63126d48bfa2190 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Sun, 6 Sep 2026 13:21:16 +0400 Subject: [PATCH 3/5] fix: prevent cancelled and duplicate async question input --- docs/async-questions.md | 2 +- src/CodexAcpServer.ts | 15 +++--- src/CodexAsyncQuestionHandler.ts | 19 ++++--- .../CodexACPAgent/async-questions.test.ts | 50 +++++++++++++++++-- 4 files changed, 67 insertions(+), 19 deletions(-) diff --git a/docs/async-questions.md b/docs/async-questions.md index 15751ab7..c80df412 100644 --- a/docs/async-questions.md +++ b/docs/async-questions.md @@ -96,7 +96,7 @@ Synchronous Codex `item/tool/requestUserInput` still uses standard ACP elicitati ## Cancellation and failure -There is no answer timeout. Session cancellation, close/delete, provider replacement, and Codex process exit cancel pending question RPCs through ACP `$/cancel_request`. The client closes the form and settles its request. Late responses are ignored, and cancelled answers waiting in the steering queue cannot start work. Input already accepted by Codex cannot be retracted by dismissing the form. +There is no answer timeout. Prompt RPC cancellation, session cancellation, close/delete, provider replacement, and Codex process exit cancel pending question RPCs through ACP `$/cancel_request`. The client closes the form and settles its request. Late responses are ignored, and cancelled answers waiting in the steering queue cannot start work. After cancellation, new question events are ignored until another prompt begins. Input already accepted by Codex cannot be retracted by dismissing the form. Request errors, invalid responses, and failed delivery produce a visible message asking the user to answer in chat. The adapter does not automatically retry an uncertain delivery. diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index cda2457f..3525f652 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -1545,7 +1545,7 @@ export class CodexAcpServer { const turnId = await this.getSteerableTurnId(sessionState); signal?.throwIfAborted(); if (turnId) { - const injected = await this.injectSteerIntoActiveTurn(params, turnId, sessionState); + const injected = await this.injectSteerIntoActiveTurn(params, turnId); if (injected) { logger.log("Steering session injected", {sessionId: params.sessionId, turnId}); return {outcome: "injected"}; @@ -1569,10 +1569,9 @@ export class CodexAcpServer { /** * Attempts to inject the prompt into the given running turn. * - * A failed injection is fatal only when the turn is still the session's - * current turn and Codex reported something other than "no active turn to - * steer". Otherwise the turn has already ended underneath us and the caller - * should start a new turn instead. + * Only an explicit "no active turn to steer" rejection permits a new turn. + * A transport failure may occur after Codex accepted the input, even if the + * tracked turn has since completed; retrying it could duplicate user input. * * @returns true when the prompt was injected; false when the caller should * fall back to starting a new turn. @@ -1580,7 +1579,6 @@ export class CodexAcpServer { private async injectSteerIntoActiveTurn( params: SessionSteerRequest, turnId: string, - sessionState: SessionState, ): Promise { try { await this.runWithProcessCheck(() => this.codexAcpClient.steerTurn({ @@ -1591,8 +1589,7 @@ export class CodexAcpServer { return true; } catch (err) { await this.codexAcpClient.waitForSessionNotifications(params.sessionId); - const turnStillActive = sessionState.currentTurnId === turnId; - if (turnStillActive && !this.isNoActiveTurnToSteerError(err)) { + if (!this.isNoActiveTurnToSteerError(err)) { throw err; } return false; @@ -2611,6 +2608,7 @@ export class CodexAcpServer { return; } logger.log("Prompt request cancelled", {sessionId: sessionState.sessionId}); + this.asyncQuestions.cancelSession(sessionState.sessionId); activePrompt.requestCancel(); const turn = activePrompt.currentTurn; if (!turn) { @@ -2769,6 +2767,7 @@ export class CodexAcpServer { let recoverableSessionFailure = sessionState.sessionFailure; sessionState.currentTurnId = null; const activePrompt = this.trackActivePrompt(params.sessionId); + this.asyncQuestions.beginPrompt(params.sessionId); let pendingTurnStart: PendingTurnStart | null = null; const ensurePendingTurnStart = (): PendingTurnStart => { if (pendingTurnStart === null) { diff --git a/src/CodexAsyncQuestionHandler.ts b/src/CodexAsyncQuestionHandler.ts index d8638b4a..9553b335 100644 --- a/src/CodexAsyncQuestionHandler.ts +++ b/src/CodexAsyncQuestionHandler.ts @@ -11,6 +11,7 @@ import {AIR_ASYNC_QUESTIONS_KEY, clientSupportsAirCapability} from "./AirExtensi import {logger} from "./Logger"; type QuestionSession = { + acceptingRequests: boolean; seen: Set; pending: Set; }; @@ -24,17 +25,22 @@ export class CodexAsyncQuestionHandler { private readonly deliver: (request: SessionSteerRequest, signal: AbortSignal) => Promise, ) {} + beginPrompt(sessionId: string): void { + const session = this.sessions.get(sessionId); + if (session) { + session.acceptingRequests = true; + } else { + this.sessions.set(sessionId, {acceptingRequests: true, seen: new Set(), pending: new Set()}); + } + } + handleNotification(notification: ServerNotification, capabilities: ClientCapabilities | null): void { if (notification.method !== "item/completed" || !clientSupportsAirCapability(capabilities, AIR_ASYNC_QUESTIONS_KEY)) return; const {threadId, turnId, item} = notification.params; if (item.type !== "agentMessage" || item.delivery !== "async" || !item.questions?.length) return; - let session = this.sessions.get(threadId); - if (!session) { - session = {seen: new Set(), pending: new Set()}; - this.sessions.set(threadId, session); - } - if (session.seen.has(item.id)) return; + const session = this.sessions.get(threadId); + if (!session?.acceptingRequests || session.seen.has(item.id)) return; session.seen.add(item.id); const controller = new AbortController(); session.pending.add(controller); @@ -62,6 +68,7 @@ export class CodexAsyncQuestionHandler { cancelSession(sessionId: string): void { const session = this.sessions.get(sessionId); if (!session) return; + session.acceptingRequests = false; for (const controller of session.pending) controller.abort(); session.pending.clear(); } diff --git a/src/__tests__/CodexACPAgent/async-questions.test.ts b/src/__tests__/CodexACPAgent/async-questions.test.ts index 5b70471a..f253ad54 100644 --- a/src/__tests__/CodexACPAgent/async-questions.test.ts +++ b/src/__tests__/CodexACPAgent/async-questions.test.ts @@ -19,7 +19,7 @@ function airCapabilities(version: unknown = 1, capabilities: unknown = ["asyncQu return {_meta: {jetbrains: {air: {version, capabilities}}}}; } -async function setup(clientCapabilities: acp.ClientCapabilities = airCapabilities()) { +async function setup(clientCapabilities: acp.ClientCapabilities = airCapabilities(), signal?: AbortSignal) { const fixture = createCodexMockTestFixture(); const agent = fixture.getCodexAcpAgent(); const appServer = fixture.getCodexAppServerClient(); @@ -43,7 +43,7 @@ async function setup(clientCapabilities: acp.ClientCapabilities = airCapabilitie }); const response = deferred(); fixture.setExtensionResponse(ASYNC_QUESTION_REQUEST_METHOD, response.promise); - const prompt = agent.prompt({sessionId: session.sessionId, prompt: [{type: "text", text: "Do some work"}]}); + const prompt = agent.prompt({sessionId: session.sessionId, prompt: [{type: "text", text: "Do some work"}]}, signal); await vi.waitFor(() => expect(session.currentTurnId).toBe("turn-1")); const questions = [ {title: "Есть номер YouTrack-задачи?", options: null}, @@ -51,8 +51,8 @@ async function setup(clientCapabilities: acp.ClientCapabilities = airCapabilitie ]; const item = {type: "agentMessage", id: "question-call", text: questions.map(q => q.title).join("\n"), phase: "final_answer", memoryCitation: null, delivery: "async", questions}; - async function sendQuestion() { - fixture.sendServerNotification({method: "item/completed", params: {threadId: session.sessionId, turnId: "turn-1", item}}); + async function sendQuestion(turnId = "turn-1") { + fixture.sendServerNotification({method: "item/completed", params: {threadId: session.sessionId, turnId, item}}); await fixture.getCodexAcpClient().waitForSessionNotifications(session.sessionId); } function requests() { @@ -188,6 +188,48 @@ describe("asynchronous user questions", () => { await f.finish(); }); + it("cancels pending questions when the prompt RPC is cancelled", async () => { + const controller = new AbortController(); + const f = await setup(airCapabilities(), controller.signal); + await f.sendQuestion(); + controller.abort(); + expect(f.requestSignals[0]!.aborted).toBe(true); + await f.finish(); + f.answer(); + await new Promise(resolve => setImmediate(resolve)); + expect(f.steer).not.toHaveBeenCalled(); + expect(f.start).toHaveBeenCalledTimes(1); + }); + + it("does not open a late question after session cancellation", async () => { + const f = await setup(); + await f.finish(); + await f.agent.cancel({sessionId: "session-id"}); + await f.sendQuestion(); + expect(f.requests()).toHaveLength(0); + const nextPrompt = f.agent.prompt({sessionId: "session-id", prompt: [{type: "text", text: "Continue"}]}); + await vi.waitFor(() => expect(f.session.currentTurnId).toBe("turn-2")); + await f.sendQuestion("turn-2"); + expect(f.requests()).toHaveLength(1); + f.response.resolve({status: "dismissed"}); + f.nextCompletion.resolve({threadId: "session-id", turn: turn("turn-2", "completed")}); + await nextPrompt; + }); + + it("does not resend an uncertain answer when its turn completed during a transport failure", async () => { + const f = await setup(); + f.steer.mockImplementationOnce(async () => { + await f.finish(); + throw new Error("Transport disconnected after sending input"); + }); + await f.sendQuestion(); + f.answer(); + await vi.waitFor(() => expect(f.fixture.getAcpConnectionEvents([]).some(e => e.method === "sessionUpdate" + && e.args[0].update.content?.text?.includes("Please send your answer in chat"))).toBe(true)); + expect(f.steer).toHaveBeenCalledTimes(1); + expect(f.start).toHaveBeenCalledTimes(1); + }); + it.each(["dismiss", "cancel", "close"])("does not submit input after %s", async action => { const f = await setup(); await f.sendQuestion(); From 73801a0f76c774c67af9864837c83324f146a52b Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Sun, 6 Sep 2026 13:29:05 +0400 Subject: [PATCH 4/5] fix: escape delimiters in async question replies --- docs/async-questions.md | 2 ++ src/CodexAsyncQuestionHandler.ts | 4 +++- .../CodexACPAgent/async-questions.test.ts | 24 +++++++++++++++++++ .../async-questions-escaped-input.txt | 3 +++ 4 files changed, 32 insertions(+), 1 deletion(-) create mode 100644 src/__tests__/CodexACPAgent/snapshots/async-questions-escaped-input.txt diff --git a/docs/async-questions.md b/docs/async-questions.md index c80df412..833c4da0 100644 --- a/docs/async-questions.md +++ b/docs/async-questions.md @@ -88,6 +88,8 @@ After a valid response, it constructs a user message in the observed Codex deskt This wrapper is a Codex compatibility detail. The client does not construct it. +The adapter escapes `<` and `>` inside the JSON body as `\u003c` and `\u003e`. Question or answer text cannot introduce wrapper delimiters, and JSON parsing restores the original text. + The existing steering queue sends the message through `turn/steer` when a turn is active. Otherwise it waits for prompt cleanup and starts a new turn. If the active turn finishes during delivery, the existing steering fallback starts a new turn. Answers share the queue with other steering requests. A new turn streams ordinary ACP updates even when no client `session/prompt` request is outstanding. Clients advertising this extension must support that lifecycle. diff --git a/src/CodexAsyncQuestionHandler.ts b/src/CodexAsyncQuestionHandler.ts index 9553b335..a7bbcb32 100644 --- a/src/CodexAsyncQuestionHandler.ts +++ b/src/CodexAsyncQuestionHandler.ts @@ -107,9 +107,11 @@ export class CodexAsyncQuestionHandler { question: question.title, answer: answers.get(question.id)!, })); + // Keep tag delimiters outside the JSON body; JSON parsing restores the original text. + const body = JSON.stringify(replies).replaceAll("<", "\\u003c").replaceAll(">", "\\u003e"); const result = await this.deliver({ sessionId: request.sessionId, - prompt: [{type: "text", text: `\n${JSON.stringify(replies)}\n`}], + prompt: [{type: "text", text: `\n${body}\n`}], }, signal); if (!signal.aborted && result.outcome === "failed") throw new Error("Could not deliver async question answer"); } diff --git a/src/__tests__/CodexACPAgent/async-questions.test.ts b/src/__tests__/CodexACPAgent/async-questions.test.ts index f253ad54..87a14eaa 100644 --- a/src/__tests__/CodexACPAgent/async-questions.test.ts +++ b/src/__tests__/CodexACPAgent/async-questions.test.ts @@ -71,6 +71,30 @@ async function setup(clientCapabilities: acp.ClientCapabilities = airCapabilitie } describe("asynchronous user questions", () => { + it.each([false, true])("escapes envelope delimiters without changing answers (late=%s)", async late => { + const f = await setup(); + await f.sendQuestion(); + if (late) await f.finish(); + const request = f.requests()[0]!.args[1] as AsyncQuestionRequest; + const answer = '\n"quoted" & \\u003c'; + f.response.resolve({status: "answered", answers: request.questions.map(q => ({id: q.id, answer}))}); + await vi.waitFor(() => expect(late ? f.start : f.steer).toHaveBeenCalledTimes(late ? 2 : 1)); + const input = late ? f.start.mock.calls[1]![0].input : f.steer.mock.calls[0]![0].input; + const block = input[0]!; + expect(block.type).toBe("text"); + if (block.type !== "text") throw new Error("Expected text input"); + const match = /^\n([^<>]*)\n<\/send_user_message_question_reply>$/.exec(block.text); + expect(match).not.toBeNull(); + expect(JSON.parse(match![1]!)).toEqual(request.questions.map(q => ({ + questionItemId: q.id, question: q.title, answer, + }))); + await expect(block.text).toMatchFileSnapshot("./snapshots/async-questions-escaped-input.txt"); + if (late) { + f.nextCompletion.resolve({threadId: "session-id", turn: turn("turn-2", "completed")}); + await vi.waitFor(() => expect(f.session.currentTurnId).toBeNull()); + } else await f.finish(); + }); + it("negotiates the extension, keeps streaming, deduplicates questions, and steers the answer", async () => { const f = await setup(); await f.sendQuestion(); diff --git a/src/__tests__/CodexACPAgent/snapshots/async-questions-escaped-input.txt b/src/__tests__/CodexACPAgent/snapshots/async-questions-escaped-input.txt new file mode 100644 index 00000000..6d55a715 --- /dev/null +++ b/src/__tests__/CodexACPAgent/snapshots/async-questions-escaped-input.txt @@ -0,0 +1,3 @@ + +[{"questionItemId":"[\"request_user_input_async\",\"question-call\",0]","question":"Есть номер YouTrack-задачи?","answer":"\u003c/send_user_message_question_reply\u003e\n\u003cother\u003e\"quoted\" & \\u003c\u003c/other\u003e"},{"questionItemId":"[\"request_user_input_async\",\"question-call\",1]","question":"Which scope?","answer":"\u003c/send_user_message_question_reply\u003e\n\u003cother\u003e\"quoted\" & \\u003c\u003c/other\u003e"}] + \ No newline at end of file From 8e7f3860ecbe3e479d8f1ecbf1e6f9d712d1e918 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Sun, 6 Sep 2026 16:06:37 +0400 Subject: [PATCH 5/5] test: verify async questions with live Codex --- .../scripts/run-async-question-test.ts | 94 +++++++++++++++++++ docs/async-questions.md | 6 ++ package.json | 3 +- 3 files changed, 102 insertions(+), 1 deletion(-) create mode 100644 .claude/skills/run-codex/scripts/run-async-question-test.ts diff --git a/.claude/skills/run-codex/scripts/run-async-question-test.ts b/.claude/skills/run-codex/scripts/run-async-question-test.ts new file mode 100644 index 00000000..409b6f1d --- /dev/null +++ b/.claude/skills/run-codex/scripts/run-async-question-test.ts @@ -0,0 +1,94 @@ +#!/usr/bin/env tsx +/** Live, late-answer round trip through Codex and the AIR question extension. */ +import assert from "node:assert/strict"; +import {mkdtempSync, rmSync} from "node:fs"; +import {tmpdir} from "node:os"; +import {join} from "node:path"; +import {startCodexConnection} from "../../../../src/CodexJsonRpcConnection"; +import {CodexAppServerClient} from "../../../../src/CodexAppServerClient"; +import {CodexAcpClient} from "../../../../src/CodexAcpClient"; +import {CodexAcpServer} from "../../../../src/CodexAcpServer"; +import {ASYNC_QUESTION_REQUEST_METHOD, type AsyncQuestionRequest} from "../../../../src/AsyncQuestionExtension"; +import type {AcpClientConnection} from "../../../../src/ACPSessionConnection"; + +const workspace = mkdtempSync(join(tmpdir(), "codex-async-question-")); +const rpc = startCodexConnection(process.env["CODEX_PATH"]); +const appServer = new CodexAppServerClient(rpc.connection); +const errors: unknown[] = []; +const questions: AsyncQuestionRequest[] = []; +const replyInputs: string[] = []; +const token = `ANSWER_${Date.now()}`; +let sessionId: string | undefined; +let output = ""; +let release!: () => void; +const answerGate = new Promise(done => { release = done; }); +let complete!: () => void; +const followUpCompleted = new Promise(done => { complete = done; }); + +appServer.onClientTransportEvent(event => { + if (event.eventType === "request" && event.method === "turn/start" && event.params.threadId === sessionId) { + for (const input of event.params.input) { + if (input.type === "text" && input.text.startsWith("")) replyInputs.push(input.text); + } + } + if (event.eventType !== "notification" || !("threadId" in event.params) || event.params.threadId !== sessionId) return; + if (event.method === "error" && !event.params.willRetry) errors.push(event.params); + if (event.method === "turn/completed") { + if (event.params.turn.error) errors.push(event.params.turn.error); + if (replyInputs.length > 0) complete(); + } +}); + +const connection: AcpClientConnection = { + async notify(_method: string, params: unknown) { + const event = params as {sessionId?: string; update?: {sessionUpdate?: string; content?: {text?: string}}}; + if (event.sessionId === sessionId && event.update?.sessionUpdate === "agent_message_chunk") { + output += event.update.content?.text ?? ""; + } + }, + async request(method: string, params?: Params): Promise { + assert.equal(method, ASYNC_QUESTION_REQUEST_METHOD, "Unexpected client request"); + const question = params as AsyncQuestionRequest; + assert.equal(question.sessionId, sessionId); + questions.push(question); + console.log("AIR question:", JSON.stringify(question)); + await answerGate; + return {status: "answered", answers: question.questions.map(q => ({id: q.id, answer: token}))} as Response; + }, +}; +const client = new CodexAcpClient(appServer); +const agent = new CodexAcpServer(connection, client, undefined, () => rpc.process.exitCode); +let timeout: ReturnType; +const deadline = new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error("Async question smoke test timed out after 90 seconds")), 90_000); +}); + +async function run() { + await agent.initialize({protocolVersion: 1, clientCapabilities: {_meta: {jetbrains: {air: {version: 1, capabilities: ["asyncQuestions"]}}}}}); + const session = await agent.newSession({cwd: workspace, mcpServers: []}); + sessionId = session.sessionId; + console.log("Session:", sessionId, "model:", session.models?.currentModelId); + await agent.prompt({sessionId, prompt: [{type: "text", text: "Protocol smoke test. Do not read or change files, run commands, or call external services. Call request_user_input_async once to ask 'What is the test token?' with no suggested answers, then finish with QUESTION_SENT without waiting for an answer. When my answer arrives later, reply with its exact token and nothing else. If request_user_input_async is not available, reply ASYNC_TOOL_UNAVAILABLE and stop."}]}); + assert.deepEqual(errors, [], "Initial Codex turn failed"); + assert.equal(questions.length, 1, `Expected one real async-question RPC. Model output: ${output}`); + assert.ok(output.includes("QUESTION_SENT"), "Original turn must complete while the question remains unanswered"); + output = ""; + release(); + await followUpCompleted; + await client.waitForSessionNotifications(sessionId); + assert.deepEqual(errors, [], "Follow-up Codex turn failed"); + assert.equal(replyInputs.length, 1, "Expected exactly one new turn carrying the answer"); + const body = replyInputs[0]!.split("\n")[1]!; + assert.deepEqual(JSON.parse(body), questions[0]!.questions.map(q => ({questionItemId: q.id, question: q.title, answer: token}))); + assert.ok(output.includes(token), `Model did not confirm the submitted token. Output: ${output}`); + console.log("PASS: Codex async question -> AIR RPC -> late answer -> new turn input -> model confirmation"); +} + +try { + await Promise.race([run(), deadline]); +} finally { + clearTimeout(timeout!); + rpc.connection.end(); + rpc.process.kill(); + rmSync(workspace, {recursive: true, force: true}); +} diff --git a/docs/async-questions.md b/docs/async-questions.md index 833c4da0..6de1cbd7 100644 --- a/docs/async-questions.md +++ b/docs/async-questions.md @@ -107,3 +107,9 @@ Request errors, invalid responses, and failed delivery produce a visible message Repeated live events with the same item ID create at most one request per loaded session. Sessions track their questions independently. Loading or forking history displays question text without reopening forms. Pending forms are not restored after adapter restart or session close/reopen. Durable recovery and delivery acknowledgements are outside this version of the extension. + +## Live validation + +Run `npm ci` to install the locked Codex version, then `npm run codex-test:async-questions` with an authenticated Codex account. The test uses the configured model and a temporary workspace. `CODEX_PATH` can select another CLI. + +The test asks real Codex to emit an asynchronous question, waits for the original prompt to finish, and answers the AIR request with a generated token. It verifies that exactly one new turn receives the reply envelope and that the model returns the token through ACP text updates. It fails on unavailable tools, turn errors, or a 90-second timeout. diff --git a/package.json b/package.json index b54d80af..7b515239 100644 --- a/package.json +++ b/package.json @@ -42,7 +42,8 @@ "test:e2e": "npm run build && RUN_E2E_TESTS=true vitest run --no-file-parallelism --retry=2 src/__tests__/CodexACPAgent/e2e", "test:watch": "vitest", "typecheck": "tsc --noEmit && tsc --noEmit -p examples/tsconfig.json", - "codex-test": "tsx .claude/skills/run-codex/scripts/run-codex-test.ts" + "codex-test": "tsx .claude/skills/run-codex/scripts/run-codex-test.ts", + "codex-test:async-questions": "tsx .claude/skills/run-codex/scripts/run-async-question-test.ts" }, "homepage": "https://github.com/agentclientprotocol/codex-acp#readme", "bugs": {