diff --git a/README.md b/README.md index 6192319e7..ab5ce5c92 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ Use [OpenAI Codex](https://github.com/openai/codex) from [Agent Client Protocol] - For AIR, one tool call shape with each fact in one field, as the [tool call contract](docs/air-extensions.md#tool-call-contract) defines. Other clients keep the earlier fields. - [Native ACP subagent sessions](docs/subagent-sessions.md) (after capability negotiation) with separate child histories and root-routed permissions; a legacy tool-call fallback otherwise. - Mid-turn input through the [_session/steering extension](docs/steering.md), with an opt-in host-owned fallback when no turn can accept it. +- [Async Codex questions](docs/async-questions.md) as non-blocking form elicitations, with answers steered into the turn that asked them. - [Background terminal tasks](docs/air-extensions.md#async-tasks) in AIR, with task status and targeted stop support after capability negotiation. - Session-scoped long-running goals in AIR through the [goal extension](docs/air-extensions.md#goal). - Typed warnings and errors through the opt-in [AIR session failure extension](docs/air-extensions.md#session-failure). diff --git a/docs/async-questions.md b/docs/async-questions.md new file mode 100644 index 000000000..c48450c4c --- /dev/null +++ b/docs/async-questions.md @@ -0,0 +1,50 @@ +# Async questions + +Codex's `request_user_input_async` tool lets the agent ask the user questions without waiting for +an answer. Unlike the blocking `request_user_input` of Plan mode, it is not an app-server request: +Codex emits an `agentMessage` item with `delivery: "async"` and `questions`, tells the model that +the questions were accepted, and the turn goes on. + +## What the client sees + +- The message text (the questions and their options) is sent as an ordinary agent message. +- If the client supports form elicitation, the adapter also sends a session-scoped + `elicitation/create` with one field per question. A question with options offers them, plus an + own answer the way Plan mode questions do. Fields are optional, since each question can be + answered on its own. The request carries `_meta.codex.delivery: "async"`. +- The adapter does not wait for the elicitation. On ACP v2 it does not count as a user action, so + the session state stays `running` while the form is open. + +## Answers + +- An accepted answer is steered into the turn that asked the question (`turn/steer` with that + turn as `expectedTurnId`). It is not echoed as a live user message. +- Declining or cancelling sends nothing; the agent keeps its own default. +- When the turn ends, its open elicitations are cancelled with `$/cancel_request`. The same happens + when the session closes or its app-server is lost. +- An answer that arrives after its turn ended is dropped, as in the Codex TUI. This includes an + answer to a cancelled elicitation, and one that Codex rejects because a new turn has started. The + adapter then sends a `warning` notice, where the client supports notices. + +## Reply format and replay + +Answers use the envelope that the Codex TUI and desktop app send, so Codex reads them the same way: + +``` + +[{"answer":"Staging","question":"Which environment?","questionItemId":"[\"request_user_input_async\",\"call_1\",0]"}] + +``` + +Codex stores this text as an ordinary user message. On `session/load` and `session/resume`, a user +message that is exactly one such envelope is replayed as each question in a quote followed by its +answer: + +``` +> Which environment? + +Staging +``` + +The format follows Codex's `codex-rs/context-fragments/src/answered_question.rs`, and its parsing +follows `codex-rs/tui/src/async_question_reply.rs`. diff --git a/src/ACPSessionConnection.ts b/src/ACPSessionConnection.ts index 6341e7fa8..164efe6be 100644 --- a/src/ACPSessionConnection.ts +++ b/src/ACPSessionConnection.ts @@ -138,11 +138,23 @@ export class AcpV2Connection { }, options?.cancellationSignal); } - /** Request-scoped login elicitations and idle background elicitations never change state. */ + /** + * Sends `elicitation/create`. Request-scoped login elicitations and idle background elicitations never change + * the session state. + * + * @param request The elicitation to show. + * @param options Request options, such as the signal that cancels the request. + * @param interaction.userAction Whether the agent waits for the answer, `true` by default. While such an + * elicitation is open in a running turn, the session reports `requires_action` whenever none of its items + * is in progress. Pass `false` for a question the agent keeps working through, so the session state is left + * as it is. + */ async createElicitation( request: acp.CreateElicitationRequest, options?: acp.SendRequestOptions, + interaction: {userAction?: boolean} = {}, ): Promise { + const userAction = interaction.userAction ?? true; const sendRequest = async () => { const response = await this.client.request( acpV2.methods.client.elicitation.create, @@ -152,7 +164,7 @@ export class AcpV2Connection { return toV1CreateElicitationResponse(response); }; const sessionId = elicitationSessionId(request); - if (sessionId === undefined) { + if (sessionId === undefined || !userAction) { return sendRequest(); } const toolCallId = "toolCallId" in request && typeof request.toolCallId === "string" ? request.toolCallId : undefined; diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 33dbdce85..9a7abb2de 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -157,6 +157,7 @@ import { import packageJson from "../package.json"; import {isJetBrains2026_1Client} from "./JBUtils"; import {clientSupportsNotices} from "./SessionNotice"; +import {AsyncQuestionClient, AsyncQuestionElicitations, renderAsyncQuestionReply} from "./CodexAsyncQuestions"; import { createAgentTextMessageChunk, createAgentTextThoughtChunk, @@ -520,6 +521,8 @@ export class CodexAcpServer { /** Tail of the per-session turn-start FIFO; see `acquireTurnStartReservation`. */ private readonly turnStartQueueTail: Map; settled: boolean}>; private readonly steeringQueues: Map; + /** The async questions of each session, shown while the turn that asked them runs. */ + private readonly asyncQuestions = new Map(); /** * Steers awaiting their `userMessage` landing, keyed by the minted `clientUserMessageId` * passed to `turn/steer`. Used only to show a v2 live `user_message` when a steer injected @@ -2016,6 +2019,8 @@ export class CodexAcpServer { } finally { if (this.getSessionGeneration(params.sessionId) === closeGeneration) { this.sessions.delete(params.sessionId); + this.asyncQuestions.get(params.sessionId)?.dispose(); + this.asyncQuestions.delete(params.sessionId); if (!this.v2PromptsInFlight.has(params.sessionId)) { AcpV2Connection.of(this.connection)?.forgetSession(params.sessionId); } @@ -4040,9 +4045,12 @@ export class CodexAcpServer { private userInputToContentBlocks(input: UserInput): acp.ContentBlock[] { switch (input.type) { - case "text": + case "text": { + const asyncQuestionReply = renderAsyncQuestionReply(input.text); + if (asyncQuestionReply !== null) return [{type: "text", text: asyncQuestionReply}]; return desktopAttachmentHistory(input.text) ?? (input.text.length > 0 ? [{ type: "text", text: input.text }] : []); + } case "image": return [{ type: "text", @@ -4867,6 +4875,7 @@ export class CodexAcpServer { await observeInteraction(event); if (!promptNotificationsActive) { await promptEventHandler.handleSessionScopedNotification(event); + this.asyncQuestionsFor(sessionState).handleNotification(event); await this.trackCodexTurnCompletion(sessionState, event, promptEventHandler); return; } @@ -4874,6 +4883,8 @@ export class CodexAcpServer { && event.params.threadId === sessionState.sessionId && event.params.turn.id === sessionState.currentTurnId; await promptEventHandler.handleNotification(event); + // After the question text is shown. + this.asyncQuestionsFor(sessionState).handleNotification(event); if (completesActiveTurn) { // The prompt may remain open for plan approval after its turn has ended. Switch at // the causal boundary so a queued late error cannot enter the completed turn's buffer. @@ -5388,6 +5399,19 @@ export class CodexAcpServer { throw this.recovery.errorForGeneration(generation); } + /** The async questions of a session, created on first use. */ + private asyncQuestionsFor(sessionState: SessionState): AsyncQuestionElicitations { + const sessionId = sessionState.sessionId; + let questions = this.asyncQuestions.get(sessionId); + if (questions === undefined) { + questions = new AsyncQuestionElicitations(sessionId, this.clientCapabilities, new AsyncQuestionClient( + sessionId, this.connection, this.clientCapabilities, () => this.codexAcpClient, + )); + this.asyncQuestions.set(sessionId, questions); + } + return questions; + } + /** * Ends what a prompt showed as running when its app-server died: open tool calls of the session and of its native * subagent sessions, the subagent sessions themselves and compactions. Nothing will ever complete them. @@ -5400,6 +5424,7 @@ export class CodexAcpServer { state: "failed" | "cancelled", ): Promise { try { + this.asyncQuestions.get(sessionState.sessionId)?.dispose(); await elicitations?.completeAllUrlElicitations(); const sessionIds = new Set([sessionState.sessionId, ...sessionState.subagents.childSessionIds()]); await this.failUnfinishedToolCalls(sessionIds); diff --git a/src/CodexAsyncQuestions.ts b/src/CodexAsyncQuestions.ts new file mode 100644 index 000000000..23f795539 --- /dev/null +++ b/src/CodexAsyncQuestions.ts @@ -0,0 +1,249 @@ +/** + * Codex's `request_user_input_async` tool asks the user questions without waiting for them: it emits an + * `agentMessage` item with `delivery: "async"` and `questions`, and the turn goes on. Codex clients answer with + * an ordinary user message that wraps the answers in a reply envelope: + * + * + * [{"answer":"Staging","question":"Which environment?","questionItemId":"[\"request_user_input_async\",\"call_1\",0]"}] + * + * + * The adapter shows the questions as a form elicitation while their turn runs and steers the answers into that + * turn in the same envelope. An answer that arrives after the turn ended is dropped, as in the Codex TUI. On + * replay, a user message that holds an envelope is shown as the questions and their answers. + * + * The envelope follows Codex's `codex-rs/context-fragments/src/answered_question.rs`, and its parsing follows + * `codex-rs/tui/src/async_question_reply.rs`. + */ +import * as acp from "@agentclientprotocol/sdk/unstable"; +import {randomUUID} from "node:crypto"; +import {type AcpClientConnection, AcpV2Connection} from "./ACPSessionConnection"; +import type {ServerNotification} from "./app-server"; +import type {ThreadItem, ToolRequestUserInputQuestion} from "./app-server/v2"; +import {buildUserInputSchema, userInputAnswers, userInputAnswerText} from "./CodexElicitationHandler"; +import type {CodexAcpClient} from "./CodexAcpClient"; +import {clientSupportsFormElicitation} from "./ElicitationCapabilities"; +import {logger} from "./Logger"; +import {clientSupportsNotices, createSessionNotice} from "./SessionNotice"; + +const TOOL_NAME = "request_user_input_async"; +const REPLY_START = ""; +const REPLY_END = ""; +// Codex bounds the model-authored parts of a reply, and falls back to plain text for a longer question id. +const MAX_REPLY_FIELD_LENGTH = 512; + +export type AsyncQuestionReply = {questionItemId: string, question: string, answer: string}; + +type AsyncQuestionMessage = ThreadItem & {type: "agentMessage"}; + +/** The id Codex clients give a question: the tool name, the id of the message that asked it, and its index. */ +export function asyncQuestionItemId(messageId: string, index: number): string { + return JSON.stringify([TOOL_NAME, messageId, index]); +} + +/** The user message text that answers async questions. */ +export function formatAsyncQuestionReply(replies: readonly AsyncQuestionReply[]): string { + const bounded = replies.map(reply => ({ + ...reply, + question: [...reply.question].slice(0, MAX_REPLY_FIELD_LENGTH).join("").replace(/[\r\n]/g, " "), + })); + if (bounded.some(reply => reply.questionItemId.length > MAX_REPLY_FIELD_LENGTH)) { + return displayReplies(bounded); + } + // Codex writes the keys in this order. + const body = bounded.map(({answer, question, questionItemId}) => ({answer, question, questionItemId})); + return `${REPLY_START}\n${JSON.stringify(body)}\n${REPLY_END}`; +} + +/** + * The replies in a user message that is exactly one reply envelope with one reply or an array of them, or + * `null`. Text around the envelope makes it an ordinary message. + */ +export function parseAsyncQuestionReply(text: string): AsyncQuestionReply[] | null { + const trimmed = text.trim(); + if (!trimmed.startsWith(REPLY_START) || !trimmed.endsWith(REPLY_END)) return null; + let body: unknown; + try { + body = JSON.parse(trimmed.slice(REPLY_START.length, trimmed.length - REPLY_END.length)); + } catch { + return null; + } + const replies = Array.isArray(body) ? body : [body]; + if (replies.length === 0 || !replies.every(isReply)) return null; + return replies.map(({questionItemId, question, answer}) => ({questionItemId, question, answer})); +} + +/** A user message that answers async questions, as each question in a quote followed by its answer. */ +export function renderAsyncQuestionReply(text: string): string | null { + const replies = parseAsyncQuestionReply(text); + return replies === null ? null : displayReplies(replies); +} + +function displayReplies(replies: readonly AsyncQuestionReply[]): string { + return replies.map(reply => `> ${reply.question}\n\n${reply.answer}`).join("\n\n"); +} + +function isReply(value: unknown): value is AsyncQuestionReply { + if (typeof value !== "object" || value === null) return false; + const reply = value as Record; + return typeof reply["questionItemId"] === "string" + && typeof reply["question"] === "string" + && typeof reply["answer"] === "string"; +} + +/** Reaches the ACP client and Codex on behalf of the async questions of one session. */ +export class AsyncQuestionClient { + /** + * @param sessionId The session that the questions belong to. + * @param connection The connection to the ACP client. + * @param clientCapabilities The capabilities that the ACP client declared. + * @param codexAcpClient Returns the current Codex client, which changes when the app-server restarts. + */ + constructor( + private readonly sessionId: string, + private readonly connection: AcpClientConnection, + private readonly clientCapabilities: acp.ClientCapabilities | null, + private readonly codexAcpClient: () => CodexAcpClient, + ) {} + + /** Sends the elicitation without marking the session as waiting for the user. */ + createElicitation(request: acp.CreateElicitationRequest, signal: AbortSignal): Promise { + const v2 = AcpV2Connection.of(this.connection); + return v2 !== null + ? v2.createElicitation(request, {cancellationSignal: signal}, {userAction: false}) + : this.connection.request(acp.methods.client.elicitation.create, request, {cancellationSignal: signal}); + } + + /** Steers the text into the turn; rejects if that turn is no longer running. */ + async steer(turnId: string, text: string): Promise { + await this.codexAcpClient().steerTurn({ + threadId: this.sessionId, + turnId, + prompt: [{type: "text", text}], + clientUserMessageId: randomUUID(), + }); + } + + /** Tells the user that an answer was not sent, if the client supports notices. */ + async reportDroppedAnswer(): Promise { + if (!clientSupportsNotices(this.clientCapabilities)) return; + await this.connection.notify(acp.methods.client.session.update, { + sessionId: this.sessionId, + update: createSessionNotice( + "warning", + "Answer not sent", + "Codex finished before your answer arrived. Send it as a new message if it still matters.", + ), + }); + } +} + +/** Shows the async questions of one session as elicitations, for as long as the turn that asked them runs. */ +export class AsyncQuestionElicitations { + // The running turns that asked a question, each with the signal that cancels its questions. + private readonly turns = new Map(); + + constructor( + private readonly sessionId: string, + private readonly clientCapabilities: acp.ClientCapabilities | null, + private readonly client: AsyncQuestionClient, + ) {} + + /** Asks the questions of a completed async message and cancels them when their turn ends; never waits. */ + handleNotification(notification: ServerNotification): void { + switch (notification.method) { + case "item/completed": { + const {threadId, turnId, item} = notification.params; + if (threadId === this.sessionId && item.type === "agentMessage" && item.delivery === "async" + && (item.questions?.length ?? 0) > 0) { + this.ask(turnId, item); + } + return; + } + case "turn/completed": + if (notification.params.threadId === this.sessionId) { + this.endTurn(notification.params.turn.id); + } + return; + default: + return; + } + } + + /** Cancels every open question, when the session closes or its app-server is gone. */ + dispose(): void { + for (const turnId of [...this.turns.keys()]) { + this.endTurn(turnId); + } + } + + private endTurn(turnId: string): void { + this.turns.get(turnId)?.abort(); + this.turns.delete(turnId); + } + + private ask(turnId: string, message: AsyncQuestionMessage): void { + if (!clientSupportsFormElicitation(this.clientCapabilities)) return; + let turn = this.turns.get(turnId); + if (turn === undefined) { + turn = new AbortController(); + this.turns.set(turnId, turn); + } + const questions = formQuestions(message); + const request: acp.CreateElicitationRequest = { + sessionId: this.sessionId, + mode: "form", + message: "Codex keeps working while you answer. An answer sent after it finishes is not delivered.", + // Questions are answered independently, as in Codex. + requestedSchema: buildUserInputSchema(questions, this.clientCapabilities, false), + _meta: {codex: {delivery: "async"}}, + }; + void this.client.createElicitation(request, turn.signal).then( + response => this.answer(turnId, message, questions, response) + .catch(error => logger.error("Failed to deliver an async question answer", error)), + error => logger.log("Async question closed without an answer", {sessionId: this.sessionId, turnId, error}), + ); + } + + private async answer( + turnId: string, + message: AsyncQuestionMessage, + questions: readonly ToolRequestUserInputQuestion[], + response: acp.CreateElicitationResponse, + ): Promise { + const answers = userInputAnswers(response, questions, this.clientCapabilities); + const replies = questions.flatMap((question, index): AsyncQuestionReply[] => { + const answer = userInputAnswerText(answers[question.id]?.answers ?? []); + return answer.length === 0 ? [] : [{ + questionItemId: asyncQuestionItemId(message.id, index), + question: question.question, + answer, + }]; + }); + if (replies.length === 0) return; + if (this.turns.has(turnId)) { + try { + await this.client.steer(turnId, formatAsyncQuestionReply(replies)); + return; + } catch (error) { + logger.log("Async question answer was not steered into its turn", {sessionId: this.sessionId, turnId, error}); + } + } + logger.log("Dropped an async question answer: its turn has ended", {sessionId: this.sessionId, turnId}); + await this.client.reportDroppedAnswer(); + } +} + +function formQuestions(message: AsyncQuestionMessage): ToolRequestUserInputQuestion[] { + return (message.questions ?? []).map((question, index) => { + const options = question.options ?? []; + return { + id: `question_${index}`, + header: "", + question: question.title, + // Codex always accepts an own answer. + isOther: options.length > 0, + isSecret: false, + options: options.length > 0 ? options.map(label => ({label, description: ""})) : null, + }; + }); +} diff --git a/src/CodexElicitationHandler.ts b/src/CodexElicitationHandler.ts index 09bfc7137..533fb4414 100644 --- a/src/CodexElicitationHandler.ts +++ b/src/CodexElicitationHandler.ts @@ -128,7 +128,7 @@ function userInputNoteFieldId(questionId: string, questionIds: ReadonlySet = {}; + const required: string[] = []; + const questionIds = new Set(questions.map(question => question.id)); + const airClient = isAirClient(clientCapabilities); + + for (const question of questions) { + const options = question.options ?? []; + const hasOptions = options.length > 0; + const hasOtherAnswer = question.isOther && hasOptions; + const base = { + title: question.question || question.header || question.id, + ...(question.header ? { description: question.header } : {}), + _meta: { + codex: { + isOther: question.isOther, + isSecret: question.isSecret, + }, + }, + }; + if (answersRequired) { + required.push(question.id); + } + properties[question.id] = hasOptions + ? { + ...base, + type: "string", + oneOf: [ + ...options.map(option => ({ + const: option.label, + title: option.label, + ...(option.description ? { description: option.description } : {}), + })), + ...(hasOtherAnswer && !airClient && !options.some(option => option.label === USER_INPUT_OTHER_OPTION) ? [{ + const: USER_INPUT_OTHER_OPTION, + title: USER_INPUT_OTHER_OPTION, + description: "Provide a different answer in the note field.", + }] : []), + ], + } + : { + ...base, + type: "string", + }; + if (hasOtherAnswer) { + const noteMeta = { + codex: { + questionId: question.id, + role: "user_note", + isSecret: question.isSecret, + }, + }; + properties[userInputNoteFieldId(question.id, questionIds)] = { + type: "string", + title: "Additional answer or note", + _meta: airClient + ? withAirMeta({...noteMeta, [LEGACY_AIR_CUSTOM_ANSWER_KEY]: true}, AIR_CUSTOM_ANSWER_KEY, true) + : noteMeta, + }; + } + } + + return { + type: "object", + properties, + ...(answersRequired ? {required} : {}), + }; +} + +function buildUserInputRequest( + params: ToolRequestUserInputParams, + clientCapabilities: acp.ClientCapabilities | null, +): acp.CreateElicitationRequest { + return { + sessionId: params.threadId, + toolCallId: params.itemId, + mode: "form", + message: "Codex needs your input to continue.", + requestedSchema: buildUserInputSchema(params.questions, clientCapabilities, true), + _meta: { + codex: { + autoResolutionMs: params.autoResolutionMs, + }, + }, + }; +} + +/** + * The answers in a response to the form of {@link buildUserInputSchema}, in Codex's `request_user_input` shape: + * a chosen option, "None of the above" for an own answer, and notes prefixed with `user_note: `. + */ +export function userInputAnswers( + response: acp.CreateElicitationResponse, + questions: readonly UserInputQuestion[], + clientCapabilities: acp.ClientCapabilities | null, +): ToolRequestUserInputResponse["answers"] { + if (!acp.CreateElicitationResponse.isAccept(response)) { + return {}; + } + + const answers: ToolRequestUserInputResponse["answers"] = {}; + const content = contentRecord(response.content); + const questionIds = new Set(questions.map(question => question.id)); + const airClient = isAirClient(clientCapabilities); + for (const question of questions) { + const answerValues: string[] = []; + const hasOtherAnswer = question.isOther && question.options != null && question.options.length > 0; + const value = userInputResponseValue(content, question.id); + const typedAnswer = airClient && hasOtherAnswer ? typedChoiceAnswer(value, question) : undefined; + if (typedAnswer !== undefined) { + answerValues.push(USER_INPUT_OTHER_OPTION, `${USER_INPUT_NOTE_PREFIX}${typedAnswer.trim()}`); + } else if (value !== undefined) { + answerValues.push(...(Array.isArray(value) ? value.map(String) : [String(value)])); + } + if (hasOtherAnswer) { + const note = userInputResponseValue(content, userInputNoteFieldId(question.id, questionIds)); + if (note !== undefined) { + const notes = Array.isArray(note) ? note : [note]; + answerValues.push(...notes.map(item => `${USER_INPUT_NOTE_PREFIX}${String(item).trim()}`)); + } + } + if (answerValues.length === 0) { + continue; + } + answers[question.id] = { + answers: answerValues, + }; + } + return answers; +} + +/** One question's {@link userInputAnswers} as plain text: the chosen option and notes, without the markers. */ +export function userInputAnswerText(answers: readonly string[]): string { + return answers + .filter(answer => answer !== USER_INPUT_OTHER_OPTION) + .map(answer => answer.startsWith(USER_INPUT_NOTE_PREFIX) ? answer.slice(USER_INPUT_NOTE_PREFIX.length) : answer) + .filter(answer => answer.trim().length > 0) + .join("\n\n"); +} + export class CodexElicitationHandler implements ElicitationHandler { private readonly connection: AcpClientConnection; private readonly renderer: AcpToolCallRenderer; @@ -273,7 +423,7 @@ export class CodexElicitationHandler implements ElicitationHandler { if (response === null) { return { answers: {} }; } - return this.convertUserInputResponse(response, params); + return {answers: userInputAnswers(response, params.questions, this.clientCapabilities)}; } catch (error) { logger.error("Error handling Codex user input request", error); return { answers: {} }; @@ -289,7 +439,7 @@ export class CodexElicitationHandler implements ElicitationHandler { private async requestUserInputElicitation( params: ToolRequestUserInputParams ): Promise { - const request = this.buildUserInputRequest(params); + const request = buildUserInputRequest(params, this.clientCapabilities); const signal = subagentInteractionSignal(params.threadId, this.cancellationSignal); if (params.autoResolutionMs === null) { return await withSubagentUserAction(params.threadId, () => this.connection.request( @@ -406,84 +556,6 @@ export class CodexElicitationHandler implements ElicitationHandler { } } - private buildUserInputRequest(params: ToolRequestUserInputParams): acp.CreateElicitationRequest { - const properties: Record = {}; - const required: string[] = []; - const questionIds = new Set(params.questions.map(question => question.id)); - const airClient = isAirClient(this.clientCapabilities); - - for (const question of params.questions) { - const options = question.options ?? []; - const hasOptions = options.length > 0; - const hasOtherAnswer = question.isOther && hasOptions; - const base = { - title: question.question || question.header || question.id, - ...(question.header ? { description: question.header } : {}), - _meta: { - codex: { - isOther: question.isOther, - isSecret: question.isSecret, - }, - }, - }; - required.push(question.id); - properties[question.id] = hasOptions - ? { - ...base, - type: "string", - oneOf: [ - ...options.map(option => ({ - const: option.label, - title: option.label, - ...(option.description ? { description: option.description } : {}), - })), - ...(hasOtherAnswer && !airClient && !options.some(option => option.label === USER_INPUT_OTHER_OPTION) ? [{ - const: USER_INPUT_OTHER_OPTION, - title: USER_INPUT_OTHER_OPTION, - description: "Provide a different answer in the note field.", - }] : []), - ], - } - : { - ...base, - type: "string", - }; - if (hasOtherAnswer) { - const noteMeta = { - codex: { - questionId: question.id, - role: "user_note", - isSecret: question.isSecret, - }, - }; - properties[userInputNoteFieldId(question.id, questionIds)] = { - type: "string", - title: "Additional answer or note", - _meta: airClient - ? withAirMeta({...noteMeta, [LEGACY_AIR_CUSTOM_ANSWER_KEY]: true}, AIR_CUSTOM_ANSWER_KEY, true) - : noteMeta, - }; - } - } - - return { - sessionId: params.threadId, - toolCallId: params.itemId, - mode: "form", - message: "Codex needs your input to continue.", - requestedSchema: { - type: "object", - properties, - required, - }, - _meta: { - codex: { - autoResolutionMs: params.autoResolutionMs, - }, - }, - }; - } - private convertElicitationResponse( response: acp.CreateElicitationResponse, context: McpElicitationContext @@ -532,45 +604,6 @@ export class CodexElicitationHandler implements ElicitationHandler { || (persist === "always" && persistOptions.has("always")); } - private convertUserInputResponse( - response: acp.CreateElicitationResponse, - params: ToolRequestUserInputParams - ): ToolRequestUserInputResponse { - if (!acp.CreateElicitationResponse.isAccept(response)) { - return { answers: {} }; - } - - const answers: ToolRequestUserInputResponse["answers"] = {}; - const content = contentRecord(response.content); - const questionIds = new Set(params.questions.map(question => question.id)); - const airClient = isAirClient(this.clientCapabilities); - for (const question of params.questions) { - const answerValues: string[] = []; - const hasOtherAnswer = question.isOther && question.options != null && question.options.length > 0; - const value = userInputResponseValue(content, question.id); - const typedAnswer = airClient && hasOtherAnswer ? typedChoiceAnswer(value, question) : undefined; - if (typedAnswer !== undefined) { - answerValues.push(USER_INPUT_OTHER_OPTION, `${USER_INPUT_NOTE_PREFIX}${typedAnswer.trim()}`); - } else if (value !== undefined) { - answerValues.push(...(Array.isArray(value) ? value.map(String) : [String(value)])); - } - if (hasOtherAnswer) { - const note = userInputResponseValue(content, userInputNoteFieldId(question.id, questionIds)); - if (note !== undefined) { - const notes = Array.isArray(note) ? note : [note]; - answerValues.push(...notes.map(item => `${USER_INPUT_NOTE_PREFIX}${String(item).trim()}`)); - } - } - if (answerValues.length === 0) { - continue; - } - answers[question.id] = { - answers: answerValues, - }; - } - return { answers }; - } - private async publishAcceptedMcpToolApproval( sessionId: string, context: McpElicitationContext, diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 8c6f9b039..ae9566eeb 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -40,7 +40,7 @@ import {commandUsesTerminal} from "./tool-calls/reporters/CommandReporter"; import {CompactionReporter} from "./tool-calls/reporters/CompactionReporter"; import {DynamicToolReporter} from "./tool-calls/reporters/DynamicToolReporter"; import {FileChangeReporter} from "./tool-calls/reporters/FileChangeReporter"; -import {FileCitationStream} from "./CodexFileCitations"; +import {FileCitationStream, renderFileCitations} from "./CodexFileCitations"; import {CodexItemStreamState} from "./CodexItemStreamState"; import {PLAIN_TOOL_CALL_ITEM_TYPES} from "./CodexSessionToolCalls"; import {ImageGenerationReporter} from "./tool-calls/reporters/ImageGenerationReporter"; @@ -797,12 +797,16 @@ export class CodexEventHandler { return this.createAgentMessageText(citations.push(event.delta), event.itemId, event.threadId); } - /** Sends the text a file citation held back, once the message is complete. */ - private flushAgentMessageText(itemId: string, threadId: string): UpdateSessionEvent | null { + /** + * Sends the text a file citation held back, once the message is complete. A message that + * arrives whole, without deltas (an async question, for one), is sent in full. + */ + private flushAgentMessageText(item: ThreadItem & { type: "agentMessage" }, threadId: string): UpdateSessionEvent | null { const streams = this.streamsFor(threadId); - const rest = streams.agentMessageCitations.get(itemId)?.flush() ?? ""; - streams.agentMessageCitations.delete(itemId); - return this.createAgentMessageText(rest, itemId, threadId); + const citations = streams.agentMessageCitations.get(item.id); + streams.agentMessageCitations.delete(item.id); + const text = citations ? citations.flush() : renderFileCitations(item.text, this.sessionState.cwd); + return this.createAgentMessageText(text, item.id, threadId); } private createAgentMessageText(text: string, itemId: string, threadId: string): UpdateSessionEvent | null { @@ -969,7 +973,7 @@ export class CodexEventHandler { return this.renderer.render(this.subagents.legacyCollaborationCompleted(event.item)); case "agentMessage": this.rememberAgentMessagePhase(event.item, event.threadId); - return this.flushAgentMessageText(event.item.id, event.threadId); + return this.flushAgentMessageText(event.item, event.threadId); case "plan": return await this.createCompletedPlanEvent(event.item); case "exitedReviewMode": diff --git a/src/__tests__/CodexACPAgent/agent-message-events.test.ts b/src/__tests__/CodexACPAgent/agent-message-events.test.ts index 879784bb8..3e510a5ba 100644 --- a/src/__tests__/CodexACPAgent/agent-message-events.test.ts +++ b/src/__tests__/CodexACPAgent/agent-message-events.test.ts @@ -123,4 +123,26 @@ describe("CodexEventHandler - agent message events", () => { "data/agent-message-file-citations.json" ); }); + + it("sends an agent message that arrives whole, without deltas", async () => { + const item = { + type: "agentMessage" as const, + id: "call-question", + text: "Which environment? See :codex-file-citation{path=\"docs/envs.md\"}\n- Staging\n- Production", + phase: "final_answer" as const, + memoryCitation: null, + delivery: "async" as const, + questions: [{title: "Which environment? See :codex-file-citation{path=\"docs/envs.md\"}", options: ["Staging", "Production"]}], + }; + const notifications: ServerNotification[] = [ + {method: "item/started", params: {threadId: sessionId, turnId: "turn-1", startedAtMs: 0, item}}, + {method: "item/completed", params: {threadId: sessionId, turnId: "turn-1", completedAtMs: 0, item}}, + ]; + + await setupPromptAndSendNotifications(mockFixture, sessionId, sessionState, notifications); + + await expect(mockFixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/agent-message-without-deltas.json" + ); + }); }); diff --git a/src/__tests__/CodexACPAgent/async-questions.test.ts b/src/__tests__/CodexACPAgent/async-questions.test.ts new file mode 100644 index 000000000..3acb31607 --- /dev/null +++ b/src/__tests__/CodexACPAgent/async-questions.test.ts @@ -0,0 +1,156 @@ +import {afterEach, describe, expect, it, vi} from 'vitest'; +import type * as acpV2 from '@agentclientprotocol/sdk/experimental/v2/unstable'; +import type {ThreadItem} from '../../app-server/v2'; +import { + connectSession, + dump, + itemCompleted, + itemStarted, + type PromptSession, + settle, + stateUpdates, + turnCompleted, + turnStarted, + userMessageItem, +} from './v2-prompt-harness'; + +const questionItem: ThreadItem = { + type: "agentMessage", + id: "call-question", + text: "Which environment?\n- Staging\n- Production\n\nAnything else?", + phase: "final_answer", + memoryCitation: null, + delivery: "async", + questions: [ + {title: "Which environment?", options: ["Staging", "Production"]}, + {title: "Anything else?", options: null}, + ], +}; + +type Answer = (request: acpV2.CreateElicitationRequest, signal: AbortSignal) => Promise; + +describe.each([1, 2] as const)('async Codex questions over ACP v%i', (protocolVersion) => { + let closeClient: (() => void) | null = null; + + afterEach(() => { + closeClient?.(); + closeClient = null; + vi.clearAllMocks(); + }); + + async function connect(onElicitation: Answer, steer: (params: unknown) => Promise = async () => ({turnId: "turn-1"})) { + const capabilities = {elicitation: {form: {}}}; + const client = await connectSession(protocolVersion, { + clientCapabilities: capabilities, + v1ClientCapabilities: {...capabilities, session: {notices: {}}}, + onElicitation, + }); + closeClient = () => client.connection.close(); + const steers: unknown[] = []; + client.setCodexResponse("turn/steer", async params => { + steers.push({...(params as object), clientUserMessageId: ""}); + return await steer(params); + }); + return {client, steers}; + } + + /** Sends a prompt and lets Codex start its turn and ask the questions. */ + async function askDuringTurn(client: PromptSession) { + const response = client.sendPrompt([{type: "text", text: "Hello"}]); + await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(1)); + const clientUserMessageId = client.turnStartParams[0]!["clientUserMessageId"] as string; + client.emit(turnStarted()); + client.emit(itemCompleted(userMessageItem(clientUserMessageId))); + // A v1 prompt responds only when its turn ends. + if (protocolVersion === 2) await response; + client.emit(itemStarted(questionItem)); + client.emit(itemCompleted(questionItem)); + } + + async function finishTurn(client: PromptSession) { + client.emit(turnCompleted()); + await client.promptRunFinished(); + await settle(); + } + + function elicitationRequests(client: PromptSession) { + return client.transcript.flatMap(entry => "elicitationRequest" in entry ? [entry.elicitationRequest] : []); + } + + function notices(client: PromptSession) { + return client.transcript.flatMap(entry => "sessionUpdate" in entry && entry.sessionUpdate.sessionUpdate === "notice" + ? [entry.sessionUpdate] : []); + } + + it('asks the questions and steers the answers into the turn that asked them', async () => { + const {client, steers} = await connect(async () => ({ + action: "accept", + content: {question_0: "Staging", question_0_note: "but only for now", question_1: "Use the new API"}, + })); + await askDuringTurn(client); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + await finishTurn(client); + + await expect(dump({ + elicitationRequests: elicitationRequests(client), + steers, + notices: notices(client), + })).toMatchFileSnapshot(`data/async-questions-v${protocolVersion}-answered.json`); + }); + + it('sends nothing to Codex when the user declines', async () => { + const {client, steers} = await connect(async () => ({action: "decline"})); + await askDuringTurn(client); + await vi.waitFor(() => expect(elicitationRequests(client)).toHaveLength(1)); + await finishTurn(client); + + expect(steers).toEqual([]); + expect(notices(client)).toEqual([]); + }); + + it('cancels the questions when their turn ends, and reports an answer that arrives after that', async () => { + let cancelled = false; + const {client, steers} = await connect((_, signal) => new Promise(resolve => { + // A client that still answers after the cancellation. + signal.addEventListener("abort", () => { + cancelled = true; + resolve({action: "accept", content: {question_1: "Too late"}}); + }); + })); + await askDuringTurn(client); + await vi.waitFor(() => expect(elicitationRequests(client)).toHaveLength(1)); + await finishTurn(client); + await vi.waitFor(() => expect(notices(client)).toHaveLength(1)); + + expect(cancelled).toBe(true); + expect(steers).toEqual([]); + await expect(dump(notices(client))).toMatchFileSnapshot(`data/async-questions-v${protocolVersion}-dropped.json`); + }); + + it('reports an answer that Codex rejects because its turn has ended', async () => { + const {client, steers} = await connect( + async () => ({action: "accept", content: {question_1: "Late"}}), + async () => { throw new Error("no active turn to steer"); }, + ); + await askDuringTurn(client); + await vi.waitFor(() => expect(notices(client)).toHaveLength(1)); + await finishTurn(client); + + expect(steers).toHaveLength(1); + await expect(dump(notices(client))).toMatchFileSnapshot(`data/async-questions-v${protocolVersion}-rejected.json`); + }); + + if (protocolVersion === 2) { + it('keeps the session running while the questions are open', async () => { + const {client} = await connect((_, signal) => new Promise(resolve => { + signal.addEventListener("abort", () => resolve({action: "cancel"})); + })); + await askDuringTurn(client); + await vi.waitFor(() => expect(elicitationRequests(client)).toHaveLength(1)); + await settle(); + await finishTurn(client); + + expect(stateUpdates(client.transcript).map(update => update.state)).not.toContain("requires_action"); + }); + } +}); diff --git a/src/__tests__/CodexACPAgent/data/agent-message-without-deltas.json b/src/__tests__/CodexACPAgent/data/agent-message-without-deltas.json new file mode 100644 index 000000000..0baacbbc5 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/agent-message-without-deltas.json @@ -0,0 +1,24 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "test-session-id", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "call-question", + "content": { + "type": "text", + "text": "Which environment? See [docs/envs.md](/test/cwd/docs/envs.md)\n- Staging\n- Production" + }, + "_meta": { + "jetbrains": { + "air": { + "version": 1, + "phase": "final_answer" + } + } + } + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v1-answered.json b/src/__tests__/CodexACPAgent/data/async-questions-v1-answered.json new file mode 100644 index 000000000..f79255dc8 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v1-answered.json @@ -0,0 +1,80 @@ +{ + "elicitationRequests": [ + { + "sessionId": "thread-1", + "requestedSchema": { + "type": "object", + "properties": { + "question_0": { + "title": "Which environment?", + "oneOf": [ + { + "const": "Staging", + "title": "Staging" + }, + { + "const": "Production", + "title": "Production" + }, + { + "const": "None of the above", + "title": "None of the above", + "description": "Provide a different answer in the note field." + } + ], + "_meta": { + "codex": { + "isOther": true, + "isSecret": false + } + }, + "type": "string" + }, + "question_0_note": { + "title": "Additional answer or note", + "_meta": { + "codex": { + "questionId": "question_0", + "role": "user_note", + "isSecret": false + } + }, + "type": "string" + }, + "question_1": { + "title": "Anything else?", + "_meta": { + "codex": { + "isOther": false, + "isSecret": false + } + }, + "type": "string" + } + } + }, + "mode": "form", + "message": "Codex keeps working while you answer. An answer sent after it finishes is not delivered.", + "_meta": { + "codex": { + "delivery": "async" + } + } + } + ], + "steers": [ + { + "threadId": "thread-1", + "expectedTurnId": "turn-1", + "input": [ + { + "type": "text", + "text": "\n[{\"answer\":\"Staging\\n\\nbut only for now\",\"question\":\"Which environment?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",0]\"},{\"answer\":\"Use the new API\",\"question\":\"Anything else?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",1]\"}]\n", + "text_elements": [] + } + ], + "clientUserMessageId": "" + } + ], + "notices": [] +} diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v1-dropped.json b/src/__tests__/CodexACPAgent/data/async-questions-v1-dropped.json new file mode 100644 index 000000000..7057306d8 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v1-dropped.json @@ -0,0 +1,8 @@ +[ + { + "severity": "warning", + "title": "Answer not sent", + "description": "Codex finished before your answer arrived. Send it as a new message if it still matters.", + "sessionUpdate": "notice" + } +] diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v1-rejected.json b/src/__tests__/CodexACPAgent/data/async-questions-v1-rejected.json new file mode 100644 index 000000000..7057306d8 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v1-rejected.json @@ -0,0 +1,8 @@ +[ + { + "severity": "warning", + "title": "Answer not sent", + "description": "Codex finished before your answer arrived. Send it as a new message if it still matters.", + "sessionUpdate": "notice" + } +] diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v2-answered.json b/src/__tests__/CodexACPAgent/data/async-questions-v2-answered.json new file mode 100644 index 000000000..f79255dc8 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v2-answered.json @@ -0,0 +1,80 @@ +{ + "elicitationRequests": [ + { + "sessionId": "thread-1", + "requestedSchema": { + "type": "object", + "properties": { + "question_0": { + "title": "Which environment?", + "oneOf": [ + { + "const": "Staging", + "title": "Staging" + }, + { + "const": "Production", + "title": "Production" + }, + { + "const": "None of the above", + "title": "None of the above", + "description": "Provide a different answer in the note field." + } + ], + "_meta": { + "codex": { + "isOther": true, + "isSecret": false + } + }, + "type": "string" + }, + "question_0_note": { + "title": "Additional answer or note", + "_meta": { + "codex": { + "questionId": "question_0", + "role": "user_note", + "isSecret": false + } + }, + "type": "string" + }, + "question_1": { + "title": "Anything else?", + "_meta": { + "codex": { + "isOther": false, + "isSecret": false + } + }, + "type": "string" + } + } + }, + "mode": "form", + "message": "Codex keeps working while you answer. An answer sent after it finishes is not delivered.", + "_meta": { + "codex": { + "delivery": "async" + } + } + } + ], + "steers": [ + { + "threadId": "thread-1", + "expectedTurnId": "turn-1", + "input": [ + { + "type": "text", + "text": "\n[{\"answer\":\"Staging\\n\\nbut only for now\",\"question\":\"Which environment?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",0]\"},{\"answer\":\"Use the new API\",\"question\":\"Anything else?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",1]\"}]\n", + "text_elements": [] + } + ], + "clientUserMessageId": "" + } + ], + "notices": [] +} diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v2-dropped.json b/src/__tests__/CodexACPAgent/data/async-questions-v2-dropped.json new file mode 100644 index 000000000..7057306d8 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v2-dropped.json @@ -0,0 +1,8 @@ +[ + { + "severity": "warning", + "title": "Answer not sent", + "description": "Codex finished before your answer arrived. Send it as a new message if it still matters.", + "sessionUpdate": "notice" + } +] diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v2-rejected.json b/src/__tests__/CodexACPAgent/data/async-questions-v2-rejected.json new file mode 100644 index 000000000..7057306d8 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v2-rejected.json @@ -0,0 +1,8 @@ +[ + { + "severity": "warning", + "title": "Answer not sent", + "description": "Codex finished before your answer arrived. Send it as a new message if it still matters.", + "sessionUpdate": "notice" + } +] diff --git a/src/__tests__/CodexACPAgent/data/tool-calls-and-messages-v2-without-deltas.json b/src/__tests__/CodexACPAgent/data/tool-calls-and-messages-v2-without-deltas.json new file mode 100644 index 000000000..70162eb9d --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/tool-calls-and-messages-v2-without-deltas.json @@ -0,0 +1,10 @@ +[ + { + "messageId": "call-question", + "content": { + "text": "Which environment?\n- Staging\n- Production", + "type": "text" + }, + "sessionUpdate": "agent_message_chunk" + } +] diff --git a/src/__tests__/CodexACPAgent/data/v1-load-session-async-question-answers.json b/src/__tests__/CodexACPAgent/data/v1-load-session-async-question-answers.json new file mode 100644 index 000000000..ca1bff49f --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/v1-load-session-async-question-answers.json @@ -0,0 +1,18 @@ +[ + { + "sessionUpdate": "user_message_chunk", + "messageId": "answer", + "content": { + "type": "text", + "text": "> Which environment?\n\nStaging\n\nbut only for now" + } + }, + { + "sessionUpdate": "user_message_chunk", + "messageId": "quoted", + "content": { + "type": "text", + "text": "Quoted: \n[{\"answer\":\"Staging\\n\\nbut only for now\",\"question\":\"Which environment?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",0]\"}]\n" + } + } +] diff --git a/src/__tests__/CodexACPAgent/data/v2-resume-replay-async-question-answers.json b/src/__tests__/CodexACPAgent/data/v2-resume-replay-async-question-answers.json new file mode 100644 index 000000000..edaf39551 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/v2-resume-replay-async-question-answers.json @@ -0,0 +1,24 @@ +[ + { + "sessionId": "thread-1", + "update": { + "messageId": "answer", + "content": { + "text": "> Which environment?\n\nStaging\n\nbut only for now", + "type": "text" + }, + "sessionUpdate": "user_message_chunk" + } + }, + { + "sessionId": "thread-1", + "update": { + "messageId": "quoted", + "content": { + "text": "Quoted: \n[{\"answer\":\"Staging\\n\\nbut only for now\",\"question\":\"Which environment?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",0]\"}]\n", + "type": "text" + }, + "sessionUpdate": "user_message_chunk" + } + } +] diff --git a/src/__tests__/CodexACPAgent/tool-calls-and-messages-v2.test.ts b/src/__tests__/CodexACPAgent/tool-calls-and-messages-v2.test.ts index c99be6e94..b9f38910e 100644 --- a/src/__tests__/CodexACPAgent/tool-calls-and-messages-v2.test.ts +++ b/src/__tests__/CodexACPAgent/tool-calls-and-messages-v2.test.ts @@ -130,6 +130,28 @@ describe('tool calls and messages over ACP v2', () => { .toMatchFileSnapshot('data/tool-calls-and-messages-v2-file-citations.json'); }); + it('sends an agent message that arrives whole, without deltas', async () => { + const client = await connectSession(); + closeClient = () => client.connection.close(); + await startTurn(client); + + const item: ThreadItem = { + type: "agentMessage", + id: "call-question", + text: "Which environment?\n- Staging\n- Production", + phase: "final_answer", + memoryCitation: null, + delivery: "async", + questions: [{title: "Which environment?", options: ["Staging", "Production"]}], + }; + client.emit(itemStarted(item)); + client.emit(itemCompleted(item)); + await finishTurn(client); + + await expect(dump(updatesTagged(client, "agent_message_chunk"))) + .toMatchFileSnapshot('data/tool-calls-and-messages-v2-without-deltas.json'); + }); + it('sends a tool call create as a tool_call_update upsert and passes later updates through', async () => { const client = await connectSession(); closeClient = () => client.connection.close(); diff --git a/src/__tests__/CodexACPAgent/v2-resume-replay.test.ts b/src/__tests__/CodexACPAgent/v2-resume-replay.test.ts index e655eb622..cf34d1216 100644 --- a/src/__tests__/CodexACPAgent/v2-resume-replay.test.ts +++ b/src/__tests__/CodexACPAgent/v2-resume-replay.test.ts @@ -518,3 +518,73 @@ describe('resume replay of Codex file citations', () => { await expect(dump(chunks)).toMatchFileSnapshot('data/v1-load-session-file-citations.json'); }); }); + +describe('resume replay of async question answers', () => { + const reply = '\n[{"answer":"Staging\\n\\nbut only for now","question":"Which environment?","questionItemId":"[\\"request_user_input_async\\",\\"call-question\\",0]"}]\n'; + const answerTurns: Thread["turns"] = [{ + id: T2_ID, + rootTurnId: null, + itemsView: "full", + status: "completed", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + items: [ + {type: "userMessage", id: "answer", clientId: null, content: [{type: "text", text: reply, text_elements: []}]}, + {type: "userMessage", id: "quoted", clientId: null, content: [{type: "text", text: `Quoted: ${reply}`, text_elements: []}]}, + ], + }]; + + afterEach(() => { + vi.clearAllMocks(); + expectConformingV2SessionUpdates(); + }); + + it('replays answers as the questions and their answers on v2', async () => { + const {connection, updates} = await connectV2Client({replayTurns: answerTurns}); + try { + await connection.agent.request(acpV2.methods.agent.session.resume, { + sessionId, + cwd, + replayFrom: {type: "start"}, + }); + } finally { + connection.close(); + } + + const chunks = updates.filter(({update}) => update.sessionUpdate === "user_message_chunk"); + await expect(dump(chunks)).toMatchFileSnapshot('data/v2-resume-replay-async-question-answers.json'); + }); + + it('replays answers as the questions and their answers on v1', async () => { + const fixture = createCodexMockTestFixture(); + const agent = fixture.getCodexAcpAgent(); + const codexAcpClient = fixture.getCodexAcpClient(); + const appServer = fixture.getCodexAppServerClient(); + codexAcpClient.readAuthRequirement = vi.fn().mockResolvedValue({required: false, account: null}); + codexAcpClient.getAccount = vi.fn().mockResolvedValue({account: null, requiresOpenaiAuth: false}); + codexAcpClient.listSkills = vi.fn().mockResolvedValue({data: []}); + const model = createTestModel(); + appServer.listModels = vi.fn().mockResolvedValue({data: [model], nextCursor: null}); + appServer.threadResume = vi.fn().mockResolvedValue({ + thread: createThread(), + model: model.id, + modelProvider: "openai", + cwd, + approvalPolicy: "never", + sandbox: {type: "dangerFullAccess"}, + reasoningEffort: model.defaultReasoningEffort, + }); + appServer.threadReadWithHistory = vi.fn().mockResolvedValue({thread: createThread({turns: answerTurns})}); + + await agent.initialize({protocolVersion: 1}); + await agent.loadSession({sessionId, cwd, mcpServers: []}); + + const chunks = fixture.getAcpConnectionEvents([]) + .filter((event) => event.method === "sessionUpdate") + .map((event) => event.args[0].update) + .filter((update: {sessionUpdate: string}) => update.sessionUpdate === "user_message_chunk"); + await expect(dump(chunks)).toMatchFileSnapshot('data/v1-load-session-async-question-answers.json'); + }); +}); diff --git a/src/__tests__/CodexAsyncQuestions.test.ts b/src/__tests__/CodexAsyncQuestions.test.ts new file mode 100644 index 000000000..3456afcd5 --- /dev/null +++ b/src/__tests__/CodexAsyncQuestions.test.ts @@ -0,0 +1,55 @@ +import {describe, expect, it} from "vitest"; +import { + asyncQuestionItemId, + formatAsyncQuestionReply, + parseAsyncQuestionReply, + renderAsyncQuestionReply, +} from "../CodexAsyncQuestions"; + +const reply = {questionItemId: asyncQuestionItemId("item", 0), question: "Which environment?", answer: "Staging"}; + +describe("async question replies", () => { + it("formats answers in Codex's reply envelope", () => { + expect(formatAsyncQuestionReply([reply])).toBe( + "\n" + + '[{"answer":"Staging","question":"Which environment?","questionItemId":"[\\"request_user_input_async\\",\\"item\\",0]"}]\n' + + "", + ); + }); + + it("keeps a question on one line and bounds its length", () => { + const [parsed] = parseAsyncQuestionReply(formatAsyncQuestionReply([ + {...reply, question: `First line\nsecond line ${"é".repeat(600)}`}, + ]))!; + expect(parsed!.question.startsWith("First line second line é")).toBe(true); + expect([...parsed!.question]).toHaveLength(512); + }); + + it("falls back to plain text for an overlong question id", () => { + expect(formatAsyncQuestionReply([{...reply, questionItemId: "x".repeat(513)}])) + .toBe("> Which environment?\n\nStaging"); + }); + + it("renders single and batched envelopes as questions and answers", () => { + const one = JSON.stringify(reply); + const two = JSON.stringify([reply, {...reply, question: "Second?", answer: "No"}]); + expect(renderAsyncQuestionReply(` \n\n${one}\n\n `)) + .toBe("> Which environment?\n\nStaging"); + expect(renderAsyncQuestionReply(`${two}`)) + .toBe("> Which environment?\n\nStaging\n\n> Second?\n\nNo"); + }); + + it("keeps malformed or embedded envelopes as ordinary text", () => { + const one = JSON.stringify(reply); + for (const text of [ + "> Which environment?\n\nStaging", + "[]", + `[${one},null]`, + '{"question":"First?","answer":"Yes"}', + `Quoted: ${one}`, + `${one} trailing text`, + ]) { + expect(renderAsyncQuestionReply(text), text).toBeNull(); + } + }); +});