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 new file mode 100644 index 00000000..6de1cbd7 --- /dev/null +++ b/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 + +[{"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 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. 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": { diff --git a/readme-dev.md b/readme-dev.md index bc147807..cf4f1568 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 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. diff --git a/src/AcpExtensions.ts b/src/AcpExtensions.ts index b450c8bd..29d26f77 100644 --- a/src/AcpExtensions.ts +++ b/src/AcpExtensions.ts @@ -139,3 +139,9 @@ export async function steerSessionWithFallback( ): Promise { return await connection.request(SESSION_STEERING_METHOD, params); } + +export { + ASYNC_QUESTION_REQUEST_METHOD, + type AsyncQuestionRequest, + type AsyncQuestionResponse, +} from "./AsyncQuestionExtension"; 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 new file mode 100644 index 00000000..c85fcf65 --- /dev/null +++ b/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"}; diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 4dc15e01..3525f652 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -1,3 +1,4 @@ +import {CodexAsyncQuestionHandler} from "./CodexAsyncQuestionHandler"; import * as acp from "@agentclientprotocol/sdk"; import {RequestError, type SessionId, type SessionModeState} from "@agentclientprotocol/sdk"; import {CodexEventHandler, type CompletedPlan} from "./CodexEventHandler"; @@ -131,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, @@ -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; @@ -397,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, ], }, }, @@ -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,14 +1543,16 @@ 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); + const injected = await this.injectSteerIntoActiveTurn(params, turnId); if (injected) { logger.log("Steering session injected", {sessionId: params.sessionId, turnId}); return {outcome: "injected"}; } } - return await this.startNewTurnFromSteering(params); + signal?.throwIfAborted(); + return await this.startNewTurnFromSteering(params, signal); } /** @@ -1557,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. @@ -1568,7 +1579,6 @@ export class CodexAcpServer { private async injectSteerIntoActiveTurn( params: SessionSteerRequest, turnId: string, - sessionState: SessionState, ): Promise { try { await this.runWithProcessCheck(() => this.codexAcpClient.steerTurn({ @@ -1579,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; @@ -1599,8 +1608,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 +1642,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 +1658,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 @@ -2595,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) { @@ -2753,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) { @@ -2806,6 +2821,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 +3346,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..a7bbcb32 --- /dev/null +++ b/src/CodexAsyncQuestionHandler.ts @@ -0,0 +1,118 @@ +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, + type AsyncQuestionRequest, + type AsyncQuestionResponse, +} from "./AsyncQuestionExtension"; +import {AIR_ASYNC_QUESTIONS_KEY, clientSupportsAirCapability} from "./AirExtension"; +import {logger} from "./Logger"; + +type QuestionSession = { + acceptingRequests: boolean; + 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, + ) {} + + 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; + + 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); + 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; + session.acceptingRequests = false; + 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)!, + })); + // 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${body}\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..87a14eaa --- /dev/null +++ b/src/__tests__/CodexACPAgent/async-questions.test.ts @@ -0,0 +1,288 @@ +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} 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}; +} + +function airCapabilities(version: unknown = 1, capabilities: unknown = ["asyncQuestions"]): acp.ClientCapabilities { + return {_meta: {jetbrains: {air: {version, capabilities}}}}; +} + +async function setup(clientCapabilities: acp.ClientCapabilities = airCapabilities(), signal?: AbortSignal) { + 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, + }); + 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"}]}, signal); + 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(turnId = "turn-1") { + fixture.sendServerNotification({method: "item/completed", params: {threadId: session.sessionId, turnId, 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.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(); + 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._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"), + 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([{}, 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" + && e.args[0].update.content?.text === "Есть номер YouTrack-задачи?\nWhich scope?")).toBe(true); + 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: { + 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("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(); + 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..84a4320b 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -78,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 new file mode 100644 index 00000000..eeebf9b8 --- /dev/null +++ b/src/__tests__/CodexACPAgent/snapshots/async-questions-active.json @@ -0,0 +1,88 @@ +{ + "capability": { + "air": { + "version": 1, + "capabilities": [ + "sessionFailure", + "agentFileChangeReport", + "nativeSubagentSessions", + "asyncTasks", + "asyncQuestions" + ] + } + }, + "request": [ + "_session/async_question/request", + { + "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-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 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);