feat: support asynchronous user questions through AIR

.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<void>(done => { release = done; });
+let complete!: () => void;
+const followUpCompleted = new Promise<void>(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("<send_user_message_question_reply>")) 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<Response, Params>(method: string, params?: Params): Promise<Response> {
+        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<typeof setTimeout>;
+const deadline = new Promise<never>((_, 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});
+}

docs/async-questions.md

@@ -0,0 +1,115 @@
+# 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
+<send_user_message_question_reply>
+[{"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"}]
+</send_user_message_question_reply>
+```
+
+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.
+
+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. 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.
+
+## 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.
+
+## 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.

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": {

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 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`.
 - `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.

src/AcpExtensions.ts

@@ -139,3 +139,9 @@ export async function steerSessionWithFallback(
 ): Promise<SessionSteeringResponse> {
     return await connection.request<SessionSteeringResponse, SessionSteerRequest>(SESSION_STEERING_METHOD, params);
 }
+
+export {
+    ASYNC_QUESTION_REQUEST_METHOD,
+    type AsyncQuestionRequest,
+    type AsyncQuestionResponse,
+} from "./AsyncQuestionExtension";

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";

src/AsyncQuestionExtension.ts

@@ -0,0 +1,13 @@
+/** Request/response contract for the AIR asyncQuestions capability. */
+export const ASYNC_QUESTION_REQUEST_METHOD = "_session/async_question/request";
+
+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"};