diff --git a/docs/air-extensions.md b/docs/air-extensions.md index c22a0d1b1..88820932b 100644 --- a/docs/air-extensions.md +++ b/docs/air-extensions.md @@ -784,6 +784,7 @@ Without the capability, the config options keep their old shape and carry no rec Codex app-server owns shell commands that keep running after their tool call. The adapter exposes them as async tasks when the client declares `asyncTasks`. Without the capability, the adapter sends no async task update. +ACP v1 and v2 use the same `async_task_spawned`, `async_task_progress`, and `async_task_state_update` names and payloads. ### Lifecycle @@ -1000,6 +1001,7 @@ Recovery never removes the transcript record. The adapter implements the draft [ACP subagent RFD](https://github.com/agentclientprotocol/agent-client-protocol/pull/1992). [Subagent sessions](subagent-sessions.md) describes the lifecycle. This section covers only the AIR bridge. +AIR receives `subagent_spawned` and `subagent_state_update` with the same names and payloads on ACP v1 and v2. - The canonical client field is `clientCapabilities.subagents: {}`. - Released ACP SDKs can strip that draft field. diff --git a/docs/async-questions.md b/docs/async-questions.md index c48450c4c..3761c4838 100644 --- a/docs/async-questions.md +++ b/docs/async-questions.md @@ -17,14 +17,13 @@ the questions were accepted, and the turn goes on. ## 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. +- An accepted answer joins the current turn through `turn/steer`. If the session is idle, + the adapter starts one new turn with the reply. The form summary shows the answer, without a second 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. +- The form stays open after its turn ends. Closing the session or losing its app-server cancels it. +- A rejected answer produces a warning notice where the client supports notices. +- On ACP v2, request metadata lists Air's companion field IDs under + `_meta.jetbrains.air.customAnswerFields`. The note stays part of its choice. ## Reply format and replay diff --git a/docs/steering.md b/docs/steering.md index a308c4e67..c179cc1f4 100644 --- a/docs/steering.md +++ b/docs/steering.md @@ -6,6 +6,19 @@ mid-task instead of queueing until the turn ends or cancelling it. The method is standardization proposal is [agent-client-protocol PR #1261](https://github.com/agentclientprotocol/agent-client-protocol/pull/1261). +## Native v2 prompts + +An ordinary v2 `session/prompt` steers a running turn. An idle session starts a new turn. +Slash commands wait for their own execution slot. The adapter returns `messageId` only after +Codex records the message. It emits the message once with that ID and keeps the foreground +state unchanged during steering. + +If the turn ends before it can take the input, the adapter starts a new turn once. If Codex +accepts the steer but never records its input, the prompt fails without a retry. +A session cancellation or close interrupts the work. An individual request cancellation can +stop an input before dispatch. After dispatch, insertion decides the response; cancelling +that request does not interrupt the existing turn. + ## Capability The adapter advertises steering in the `initialize` response as diff --git a/readme-dev.md b/readme-dev.md index f26698b78..e28a1793d 100644 --- a/readme-dev.md +++ b/readme-dev.md @@ -226,7 +226,7 @@ schema alone: - Custom session updates are renamed on v2: `subagent_spawned`/`subagent_state_update` become a single `_subagent_update` (gated on the AIR `nativeSubagentSessions` capability; this contract is retained independently of standard subagent sessions), and - `async_task_spawned`/`async_task_state_update` become `_async_task_spawned`/`_async_task_state_update` + `async_task_spawned`/`async_task_progress`/`async_task_state_update` gain the `_` prefix (gated on the AIR `asyncTasks` capability). Payload shapes are unchanged. - Foreground state is independent of prompt acceptance. `running` can precede the user-message echo, including while Codex runs hooks. A successful prompt response still waits for insertion. diff --git a/src/AcpV2Elicitation.ts b/src/AcpV2Elicitation.ts index a846ae221..387228165 100644 --- a/src/AcpV2Elicitation.ts +++ b/src/AcpV2Elicitation.ts @@ -1,15 +1,19 @@ import * as acp from "@agentclientprotocol/sdk/unstable"; import type * as acpV2 from "@agentclientprotocol/sdk/experimental/v2/unstable"; +import {AIR_CUSTOM_ANSWER_FIELDS_KEY, isAirCustomAnswer, withAirMeta} from "./AirExtension"; /** - * `elicitation/create`'s request/response, and `elicitation/complete`'s notification, are - * share field and method names between v1 and v2 -- unlike - * `session/request_permission`, there is no shape to render, only the type import differs per - * version. v2 responses allow arbitrary JSON content values. Keep them intact through the - * v1-typed view and validate with the v2 response guards in shared consumers. + * Keeps the shared elicitation wire shape and copies Air's companion IDs to v2 request metadata. + * V2 responses retain arbitrary JSON values for validation by their shared consumers. */ export function toV2CreateElicitationRequest(request: acp.CreateElicitationRequest): acpV2.CreateElicitationRequest { - return request as unknown as acpV2.CreateElicitationRequest; + if (!acp.CreateElicitationRequest.isForm(request)) return request as unknown as acpV2.CreateElicitationRequest; + const fields = Object.entries(request.requestedSchema.properties ?? {}) + .filter(([, property]) => isAirCustomAnswer(property._meta)) + .map(([id]) => id); + if (fields.length === 0) return request as unknown as acpV2.CreateElicitationRequest; + // AIR's Kotlin v2 property models omit metadata. Keep the companion IDs on the request. + return {...request, _meta: withAirMeta(request._meta, AIR_CUSTOM_ANSWER_FIELDS_KEY, fields)} as unknown as acpV2.CreateElicitationRequest; } export function toV1CreateElicitationResponse(response: acpV2.CreateElicitationResponse): acp.CreateElicitationResponse { diff --git a/src/AcpV2Prompt.ts b/src/AcpV2Prompt.ts index 4e5b6277f..9fdec0a22 100644 --- a/src/AcpV2Prompt.ts +++ b/src/AcpV2Prompt.ts @@ -13,13 +13,6 @@ export type UserMessageInsertion = { /** Sent as `clientUserMessageId` on `turn/start`; Codex echoes it as `userMessage.clientId`. */ clientUserMessageId: string; onInserted: () => Promise; - /** - * Shows a synthetic prompt codex-acp starts itself (the plan-implementation follow-up turn, - * a `/goal` continuation) as a live `user_message`, once its own minted `clientUserMessageId` - * is echoed back. These turns run inside the original prompt's running…idle pair, so this - * only emits the chunks: no response is resolved and no `state_update` is sent. - */ - onSyntheticInserted: (clientUserMessageId: string, prompt: acp.ContentBlock[]) => Promise; /** * Fired synchronously once, from the turn-start response, when Codex steers this prompt into * a turn that was already running and unowned (its `running` went out before this prompt @@ -53,6 +46,12 @@ export function isInsertedUserMessage( && event.params.item.clientId === clientUserMessageId; } +/** The identity and cancellation of a native v2 prompt offered to a running turn. */ +export interface V2SteeringPrompt { + messageId: string; + signal: AbortSignal; +} + const SUPPORTED_PROMPT_CONTENT_TYPES = new Set(["text", "image", "resource_link", "resource"]); export function assertPromptContentSupported(params: {prompt: Array<{type: string}>}): void { diff --git a/src/AcpV2SessionUpdate.ts b/src/AcpV2SessionUpdate.ts index 4153a3e86..70fab4ac4 100644 --- a/src/AcpV2SessionUpdate.ts +++ b/src/AcpV2SessionUpdate.ts @@ -122,39 +122,13 @@ export function toV2SessionUpdate(update: AcpSessionUpdate): acpV2.SessionUpdate } case "subagent_update": return update; - // AIR's private contract is selected independently upstream. - case "subagent_spawned": { - const {subagentSessionId, name, task, _meta} = update; - return { - sessionUpdate: "_subagent_update", - subagentSessionId, - name, - task, - capabilities: {}, - ...(_meta != null ? {_meta} : {}), - }; - } - case "subagent_state_update": { - const {subagentSessionId, state, _meta} = update; - return { - sessionUpdate: "_subagent_update", - subagentSessionId, - state, - ...(_meta != null ? {_meta} : {}), - }; - } - // No upstream v2 design exists for async tasks (AIR-private extension); same payload as v1 - // under a `_`-prefixed tag. Gated on AIR `asyncTasks` upstream, same as above. + case "subagent_spawned": + case "subagent_state_update": + return {...update, sessionUpdate: "_subagent_update"}; case "async_task_spawned": - return {...update, sessionUpdate: "_async_task_spawned"}; - case "async_task_state_update": - return {...update, sessionUpdate: "_async_task_state_update"}; - // Declared as a type but never emitted; no v2 rendering to give it yet. case "async_task_progress": - throw acp.RequestError.internalError( - undefined, - `'${update.sessionUpdate}' session update is not supported on an ACP v2 connection yet`, - ); + case "async_task_state_update": + return {...update, sessionUpdate: `_${update.sessionUpdate}` as const}; } } diff --git a/src/AirExtension.ts b/src/AirExtension.ts index a09e0b146..bf0a488ad 100644 --- a/src/AirExtension.ts +++ b/src/AirExtension.ts @@ -35,6 +35,7 @@ export const AIR_CODEX_HOOKS_KEY = "codexHooks"; export const AIR_CONTENT_DELTA_KEY = "contentDelta"; export const AIR_SUBAGENT_KEY = "subagent"; export const AIR_CUSTOM_ANSWER_KEY = "customAnswer"; +export const AIR_CUSTOM_ANSWER_FIELDS_KEY = "customAnswerFields"; /** * The root `_meta` spelling of {@link AIR_CUSTOM_ANSWER_KEY} that released AIR versions read. * Remove it when the ACP registry no longer serves AIR versions that read only this spelling. @@ -131,3 +132,12 @@ function asRecord(value: unknown): Record { ? value as Record : {}; } + +/** Whether property metadata marks a text companion to a choice. */ +export function isAirCustomAnswer(meta: unknown): boolean { + const root = asRecord(meta); + const air = asRecord(asRecord(root[JETBRAINS_META_KEY])[AIR_META_KEY]); + const marker = air[AIR_CUSTOM_ANSWER_KEY] === undefined + ? root[LEGACY_AIR_CUSTOM_ANSWER_KEY] : air[AIR_CUSTOM_ANSWER_KEY]; + return marker === true || Object.keys(asRecord(marker)).length > 0; +} diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 87a438872..bff78f69d 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -66,6 +66,7 @@ import { toV2Failure, toV2IdleState, type UserMessageInsertion, + type V2SteeringPrompt, } from "./AcpV2Prompt"; import type {InputModality, ReasoningEffort, ServerNotification} from "./app-server"; import type { @@ -104,8 +105,9 @@ import { } from "./ModelConfigOption"; import type {TokenCount} from "./TokenCount"; import {PromptTokenUsage, toPromptUsage, ZERO_TOKEN_COUNT} from "./TokenCount"; -import {CodexCommands} from "./CodexCommands"; +import {CodexCommands, type PromptKind} from "./CodexCommands"; import {SteeringQueue} from "./SteeringQueue"; +import type {TurnStartReservation} from "./TurnStartReservation"; import type {QuotaMeta} from "./QuotaMeta"; import {logger} from "./Logger"; import {settledWithin} from "./StdUtils"; @@ -158,6 +160,7 @@ import { resolveFastServiceTier, } from "./FastModeConfig"; import packageJson from "../package.json"; +import {SteerLanding} from "./SteerLanding"; import {isJetBrains2026_1Client} from "./JBUtils"; import {clientSupportsNotices} from "./SessionNotice"; import {AsyncQuestionClient, AsyncQuestionElicitations, renderAsyncQuestionReply} from "./CodexAsyncQuestions"; @@ -335,6 +338,9 @@ export interface SessionFailure { */ const TITLE_GENERATION_SETTLE_TIMEOUT_MS = 10_000; +// Codex persists this identity in userMessage.clientId, so history can omit the internal command. +const PLAN_IMPLEMENTATION_MESSAGE_ID_PREFIX = "codex-acp:plan-implementation:"; + /** * Backoff for re-sending `turn/interrupt` when Codex reports the turn is not * interruptible yet. Covers the sub-second window between a turn's first @@ -433,25 +439,8 @@ interface PendingTurnStart { resolve: (turnId: string | null) => void; } -/** - * One session's place in the shared per-session turn-start FIFO (see - * `CodexAcpServer.acquireTurnStartReservation`). `wait` resolves once every earlier reservation - * on the session has released; `release` must be called exactly once, by whoever ends up owning - * the turn this reservation was taken for, so the next queued starter can proceed. - * - * `needsWait` is false when there was nothing to wait for (a fresh session, or the previous - * holder already released). Callers should skip `await wait` in that case: awaiting an - * already-resolved promise still costs a microtask tick, which is enough to reorder synchronous - * setup (event-subscription registration, etc.) against a caller that fires a prompt without - * awaiting it and immediately does other synchronous work. - */ -interface TurnStartReservation { - wait: Promise; - needsWait: boolean; - release: () => void; -} - interface ActivePrompt { + inputKind: PromptKind; completion: Promise; closeSignal: Promise; cancelSignal: Promise; @@ -537,12 +526,10 @@ export class CodexAcpServer { 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 - * into an already-running turn lands (there is no `prompt()` call to hook into for that - * path); v1 has nothing to emit. A steer that starts a new turn instead goes through - * `prompt()`'s own `UserMessageInsertion` tracking and never enters this map. + * passed to `turn/steer`. Native v2 prompts also wait for this insertion before answering. + * A steer that starts a new turn uses `prompt()` insertion tracking instead. */ - private readonly pendingSteerLandings: Map; + private readonly pendingSteerLandings: Map; /** Sessions with a v2 prompt that has not finished yet, including before its turn starts. */ private readonly v2PromptsInFlight = new Set(); private readonly instructionUpdates = new Set(); @@ -1791,6 +1778,7 @@ export class CodexAcpServer { if (this.v2PromptsInFlight.has(sessionId) && state.state === "idle" && !this.sessionIsClosing(sessionId)) { return; } + if (state.state === "idle") this.finishSteerLandings(sessionId, state.stopReason === "cancelled"); const session = new ACPSessionConnection(this.connection, sessionId); if (session.protocolVersion !== 2) { return; @@ -1828,11 +1816,7 @@ export class CodexAcpServer { */ private async trackSteerLanding(sessionState: SessionState, event: ServerNotification): Promise { if (event.method === "turn/completed" && event.params.threadId === sessionState.sessionId) { - for (const [clientUserMessageId, entry] of this.pendingSteerLandings) { - if (entry.sessionId === sessionState.sessionId) { - this.pendingSteerLandings.delete(clientUserMessageId); - } - } + this.finishSteerLandings(sessionState.sessionId, event.params.turn.status === "interrupted", event.params.turn.id); return; } for (const [clientUserMessageId, entry] of this.pendingSteerLandings) { @@ -1840,16 +1824,26 @@ export class CodexAcpServer { && isInsertedUserMessage(event, sessionState.sessionId, clientUserMessageId)) { this.pendingSteerLandings.delete(clientUserMessageId); await this.emitLiveSteerUserMessage(sessionState.sessionId, clientUserMessageId, entry.prompt); + entry.inserted(); return; } } } + private finishSteerLandings(sessionId: string, cancelled: boolean, turnId?: string): void { + for (const [messageId, entry] of this.pendingSteerLandings) { + if (entry.sessionId === sessionId && (turnId === undefined || entry.turnId === turnId)) { + this.pendingSteerLandings.delete(messageId); + entry.ended(cancelled); + } + } + } + /** - * Shows a landed steer as a live-only `user_message` (v2 only; the steering response itself - * carries no `messageId`, per user decision). + * Shows landed steering input on v2. An elicitation reply stays in its form summary. */ private async emitLiveSteerUserMessage(sessionId: string, messageId: string, prompt: acp.ContentBlock[]): Promise { + if (prompt.length === 1 && prompt[0]?.type === "text" && renderAsyncQuestionReply(prompt[0].text) !== null) return; const session = new ACPSessionConnection(this.connection, sessionId); if (session.protocolVersion !== 2) { return; @@ -3047,14 +3041,18 @@ export class CodexAcpServer { * ("startedNewTurn"), or could not be applied ("failed"); see * {@link performSteeringRequest}. */ - async executeOrQueueSteeringRequest(params: SessionSteerRequest): Promise { + async executeOrQueueSteeringRequest(params: SessionSteerRequest, insertion?: V2SteeringPrompt): Promise { this.assertInstructionsReady(params.sessionId); + const reservation = insertion === undefined && params._meta?.steering?.idleBehavior !== "promptRequired" + ? this.acquireTurnStartReservation(params.sessionId) : undefined; const queue = this.getSteeringQueue(params.sessionId); try { - return await queue.enqueue(params); + return await queue.enqueue(params, insertion, reservation); } catch (error) { - if (error instanceof RequestError) { - throw error; + if (error instanceof RequestError) throw error; + if (insertion !== undefined) { + const failure = toV2Failure(error); + throw new RequestError(failure.code, failure.message); } logger.error(`Steering request for session ${params.sessionId} failed`, error); return {outcome: "failed"}; @@ -3075,7 +3073,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, insertion, reservation) => this.performSteeringRequest(params, insertion, reservation)); this.steeringQueues.set(sessionId, queue); } return queue; @@ -3091,32 +3089,49 @@ export class CodexAcpServer { * "promptRequired" when the host requested an owned idle fallback, or * the outcome of starting a legacy fallback turn. */ - private async performSteeringRequest(params: SessionSteerRequest): Promise { + private async performSteeringRequest( + params: SessionSteerRequest, + insertion?: V2SteeringPrompt, + reservation?: TurnStartReservation, + ): Promise { + const clientUserMessageId = insertion?.messageId ?? randomUUID(); + try { + const result = await this.tryInjectSteeringRequest(params, clientUserMessageId, insertion); + if (result !== null) return result; + const startReservation = reservation; + reservation = undefined; + return await this.startNewTurnFromSteering(params, clientUserMessageId, startReservation); + } finally { + const unusedReservation = reservation; + if (unusedReservation !== undefined) { + void unusedReservation.wait.then(() => unusedReservation.release()); + } + } + } + + private async tryInjectSteeringRequest( + params: SessionSteerRequest, + clientUserMessageId: string, + insertion?: V2SteeringPrompt, + ): Promise { assertPromptContentSupported(params); - logger.log("Steering session requested", { - sessionId: params.sessionId, - prompt: params.prompt, - }); + logger.log("Steering session requested", {sessionId: params.sessionId, prompt: params.prompt}); const sessionState = this.getSessionState(params.sessionId); this.assertSteerInputSupported(params, sessionState); - - // Minted fresh for every steer (both protocol versions), passed to Codex as - // `TurnSteerParams.clientUserMessageId`/`TurnStartParams.clientUserMessageId`. It is a - // Codex-side param only; on v2 it doubles as the `messageId` of the live `user_message` - // shown once the steer lands (the steering response itself carries no id). - const clientUserMessageId = randomUUID(); - const turnId = await this.getSteerableTurnId(sessionState); + const activePrompt = this.activePrompts.get(params.sessionId); + const mustWait = activePrompt !== undefined && (activePrompt.cancelRequested + || activePrompt.inputKind.kind === "codexTurnCommand" && activePrompt.inputKind.name === "compact"); + const turnId = mustWait ? null : await this.getSteerableTurnId(sessionState); + if (insertion?.signal.aborted) throw RequestError.requestCancelled(); if (turnId) { - const injected = await this.injectSteerIntoActiveTurn(params, turnId, sessionState, clientUserMessageId); + const injected = await this.injectSteerIntoActiveTurn(params, turnId, sessionState, clientUserMessageId, insertion !== undefined); if (injected) { logger.log("Steering session injected", {sessionId: params.sessionId, turnId}); return {outcome: "injected"}; } } - if (params._meta?.steering?.idleBehavior === "promptRequired") { - return {outcome: "promptRequired", reason: "noRunningTurn"}; - } - return await this.startNewTurnFromSteering(params, clientUserMessageId); + return params._meta?.steering?.idleBehavior === "promptRequired" + ? {outcome: "promptRequired", reason: "noRunningTurn"} : null; } /** @@ -3146,26 +3161,47 @@ export class CodexAcpServer { turnId: string, sessionState: SessionState, clientUserMessageId: string, + waitForInsertion = false, + ): Promise { + const landing = new SteerLanding(params.sessionId, turnId, params.prompt); + this.pendingSteerLandings.set(clientUserMessageId, landing); + const delivery = this.sendSteerToActiveTurn(params, turnId, sessionState, clientUserMessageId, landing); + if (!waitForInsertion) return await delivery; + const first = await Promise.race([ + landing.result, + delivery.then(delivered => ({kind: "delivery" as const, delivered})), + ]); + if (first.kind === "inserted") return true; + const delivered = first.kind === "delivery" ? first.delivered : await delivery; + if (!delivered) return false; + const result = first.kind === "ended" ? first : await landing.result; + if (result.kind === "inserted") return true; + if (result.cancelled) throw RequestError.requestCancelled(); + throw RequestError.internalError(undefined, "The turn ended before Codex recorded the user message"); + } + + private async sendSteerToActiveTurn( + params: SessionSteerRequest, + turnId: string, + sessionState: SessionState, + clientUserMessageId: string, + landing: SteerLanding, ): Promise { - // Registered before the call goes out (not after `steerTurn` resolves), so the landing - // matcher catches a userMessage that arrives immediately after acceptance. - this.pendingSteerLandings.set(clientUserMessageId, {sessionId: params.sessionId, prompt: params.prompt}); try { await this.runWithProcessCheck(() => this.codexAcpClient.steerTurn({ - threadId: params.sessionId, - turnId, - prompt: params.prompt, - clientUserMessageId, + threadId: params.sessionId, turnId, prompt: params.prompt, clientUserMessageId, })); return true; - } catch (err) { - this.pendingSteerLandings.delete(clientUserMessageId); + } catch (error) { await this.codexAcpClient.waitForSessionNotifications(params.sessionId); + if (this.pendingSteerLandings.get(clientUserMessageId) !== landing) { + if ((await landing.result).kind === "inserted") return true; + } + this.pendingSteerLandings.delete(clientUserMessageId); + landing.ended(false); const turnStillActive = sessionState.currentTurnId !== null && codexRunningTurnId(sessionState, sessionState.currentTurnId) === turnId; - if (turnStillActive && !this.isNoActiveTurnToSteerError(err)) { - throw err; - } + if (turnStillActive && !this.isNoActiveTurnToSteerError(error)) throw error; return false; } } @@ -3185,14 +3221,14 @@ export class CodexAcpServer { private async startNewTurnFromSteering( params: SessionSteerRequest, clientUserMessageId: string, + reservation?: TurnStartReservation, ): Promise { await this.startNewTurnFromExternalPrompt(params, "Steering", undefined, { clientUserMessageId, onInserted: async () => { await this.emitLiveSteerUserMessage(params.sessionId, clientUserMessageId, params.prompt); }, - onSyntheticInserted: async () => {}, - }); + }, reservation); return {outcome: "startedNewTurn"}; } @@ -3201,13 +3237,14 @@ export class CodexAcpServer { source: string, canStart: () => Promise = async () => true, insertion?: UserMessageInsertion, + turnReservation?: TurnStartReservation, ): Promise { // Takes this session's place in the shared turn-start FIFO before anything else runs, so // no other starter can begin between this check and the turn actually starting. This // hands the reservation to `prompt()` below rather than letting it self-acquire one, so // it releases only once `prompt()` truly finishes (not when this function's own steer // promise resolves early, on the "a turn was started" success path). - const reservation = this.acquireTurnStartReservation(params.sessionId); + const reservation = turnReservation ?? this.acquireTurnStartReservation(params.sessionId); if (reservation.needsWait) { await reservation.wait; } @@ -3229,8 +3266,7 @@ export class CodexAcpServer { // steer immediately ("a turn was started") and let prompt() finish the // turn in the background. resolve(true); - }, insertion, reservation); - void promptDone.finally(() => reservation.release()); + }, insertion, reservation).finally(() => reservation.release()); promptDone.then( (response) => { if (!turnStarted && response.stopReason === "cancelled") { @@ -4140,6 +4176,7 @@ export class CodexAcpServer { } private createUserMessageUpdates(item: ThreadItem & { type: "userMessage" }): UpdateSessionEvent[] { + if (item.clientId?.startsWith(PLAN_IMPLEMENTATION_MESSAGE_ID_PREFIX)) return []; const updates: UpdateSessionEvent[] = []; // On v2, a message inserted via `session/prompt` is replayed under the id the client // provided then (`clientId`), so it round-trips as the same message on reconnect; v1 has @@ -4285,7 +4322,7 @@ export class CodexAcpServer { return []; } - private trackActivePrompt(sessionId: string): ActivePrompt { + private trackActivePrompt(sessionId: string, inputKind: PromptKind): ActivePrompt { let resolveCompletion: () => void = () => {}; const completion = new Promise((resolve) => { resolveCompletion = resolve; @@ -4308,6 +4345,7 @@ export class CodexAcpServer { let completed = false; let closeRequested = false; const activePrompt: ActivePrompt = { + inputKind, completion, closeSignal, cancelSignal, @@ -4622,6 +4660,37 @@ export class CodexAcpServer { return turnId; } + /** Offers ordinary input to the active turn; commands wait for their own execution slot. */ + private async steerOrWaitForV2Prompt( + request: acp.PromptRequest, + messageId: string, + reservation: TurnStartReservation, + canSteer: boolean, + signal?: AbortSignal, + ): Promise { + const cancelled = new AbortController(); + const unregister = this.registerQueuedV2PromptCanceller(request.sessionId, () => cancelled.abort()); + const cancellation = signal ? AbortSignal.any([signal, cancelled.signal]) : cancelled.signal; + let onAbort: () => void = () => {}; + const aborted = new Promise(resolve => { onAbort = resolve; }); + cancellation.addEventListener("abort", onAbort, {once: true}); + try { + if (cancellation.aborted) throw RequestError.requestCancelled(); + if (canSteer) { + const result = await this.executeOrQueueSteeringRequest({ + ...request, _meta: {...request._meta, steering: {idleBehavior: "promptRequired"}}, + }, {messageId, signal: cancellation}); + if (result.outcome === "injected") return true; + } + await Promise.race([reservation.wait, aborted]); + if (cancellation.aborted) throw RequestError.requestCancelled(); + return false; + } finally { + unregister(); + cancellation.removeEventListener("abort", onAbort); + } + } + /** * v2 `session/prompt`: answers `{messageId}` once the user message is inserted and lets the * turn run on in the background. A Codex prompt is inserted when Codex records its user @@ -4635,43 +4704,20 @@ export class CodexAcpServer { if (this.sessionIsClosing(sessionId)) { throw RequestError.invalidRequest(`Session ${sessionId} is closing`); } - // A prompt overlapping a running one is queued behind it rather than rejected: take this - // session's place in the shared turn-start FIFO now (before any await), then wait. Nothing - // observable (response, user message, states) happens until this prompt reaches the front. const reservation = this.acquireTurnStartReservation(sessionId); + const promptKind = this.availableCommands.classifyPrompt(request.prompt); + const messageId = randomUUID(); if (reservation.needsWait) { - // `session/cancel`/`session/close` drop this prompt while it waits here: race the - // FIFO wait against a cancellation signal so the client sees `-32800` right away, - // instead of only once the running turn ahead of it actually finishes. A - // `$/cancel_request` for this specific request goes through the same canceller, so it - // only drops this prompt and leaves the rest of the queue untouched. - let cancelled = false; - let markCancelled: () => void = () => { cancelled = true; }; - const cancelSignal = new Promise((resolve) => { - markCancelled = () => { cancelled = true; resolve(); }; - }); - const unregister = this.registerQueuedV2PromptCanceller(sessionId, markCancelled); - const onRequestCancelled = () => markCancelled(); - if (signal) { - if (signal.aborted) { - onRequestCancelled(); - } else { - signal.addEventListener("abort", onRequestCancelled, {once: true}); + try { + if (await this.steerOrWaitForV2Prompt(request, messageId, reservation, promptKind.kind === "prompt", signal)) { + void reservation.wait.then(() => reservation.release()); + return {messageId}; } - } - await Promise.race([reservation.wait, cancelSignal]); - unregister(); - signal?.removeEventListener("abort", onRequestCancelled); - if (cancelled) { - // Still release in the FIFO's own order once it is actually this prompt's turn, - // so anything queued behind it does not start while the current turn is still - // being interrupted. + } catch (error) { void reservation.wait.then(() => reservation.release()); - throw RequestError.requestCancelled(undefined, `Session ${sessionId} was cancelled before the prompt was inserted`); + throw error; } } - const promptKind = this.availableCommands.classifyPrompt(request.prompt); - const messageId = randomUUID(); const session = new ACPSessionConnection(this.connection, sessionId); let failure: acpV2.Error | null = null; const generation = this.getSessionGeneration(sessionId); @@ -4694,6 +4740,7 @@ export class CodexAcpServer { }; const sendIdle = async (response: acp.PromptResponse) => { if (this.getSessionGeneration(sessionId) !== generation) return; + this.finishSteerLandings(sessionId, response.stopReason === "cancelled"); if (failure !== null && response.stopReason !== "cancelled") { try { await this.finishOutstandingToolCalls(sessionState); @@ -4759,15 +4806,6 @@ export class CodexAcpServer { return await this.prompt(request, undefined, undefined, { clientUserMessageId: messageId, onInserted, - onSyntheticInserted: async (syntheticId, prompt) => { - try { - for (const block of prompt) { - await session.update(createUserMessageChunk(block, syntheticId)); - } - } catch (error) { - logger.error(`Failed to send the synthetic user message for session ${sessionId}`, error); - } - }, onTurnAdopted: () => { turnWasAdopted = true; }, @@ -4906,7 +4944,7 @@ export class CodexAcpServer { // A prompt that starts no model work, such as /status or a cancellation before the turn, used no tokens. sessionState.promptTokenUsage = null; sessionState.interruptTurnId = null; - const activePrompt = this.trackActivePrompt(params.sessionId); + const activePrompt = this.trackActivePrompt(params.sessionId, this.availableCommands.classifyPrompt(params.prompt)); let pendingTurnStart: PendingTurnStart | null = null; const ensurePendingTurnStart = (): PendingTurnStart => { if (pendingTurnStart === null) { @@ -4936,16 +4974,6 @@ export class CodexAcpServer { ? activePrompt.signal : AbortSignal.any([activePrompt.signal, appServerLost.signal]); let pendingInsertion = insertion; - // Synthetic turns codex-acp starts itself inside this same prompt (the plan-implementation - // follow-up, a `/goal` continuation) each mint their own id and register here, so their - // userMessage can be told apart from the original prompt's once it lands. - const pendingSyntheticInsertions = new Map Promise>(); - const registerSyntheticInsertion = (clientUserMessageId: string, prompt: acp.ContentBlock[]): void => { - if (insertion === undefined) { - return; - } - pendingSyntheticInsertions.set(clientUserMessageId, () => insertion.onSyntheticInserted(clientUserMessageId, prompt)); - }; const clearRecoveredSessionFailure = async (handler: CodexEventHandler): Promise => { await handler.completeSuccessfulTurn(sessionState.currentTurnId); const current = sessionState.sessionFailure; @@ -5043,14 +5071,6 @@ export class CodexAcpServer { if (pendingInsertion !== undefined && isInsertedUserMessage(event, params.sessionId, pendingInsertion.clientUserMessageId)) { await resolvePendingInsertion(); - } else { - for (const [clientUserMessageId, resolveSynthetic] of pendingSyntheticInsertions) { - if (isInsertedUserMessage(event, params.sessionId, clientUserMessageId)) { - pendingSyntheticInsertions.delete(clientUserMessageId); - await resolveSynthetic(); - break; - } - } } await observeInteraction(event); if (!promptNotificationsActive) { @@ -5314,12 +5334,7 @@ export class CodexAcpServer { activePrompt.currentTurn = null; sessionState.currentTurnId = null; sessionState.interruptTurnId = null; - // This second turn stays inside the original prompt's running…idle pair, so it - // gets its own minted id rather than reusing the first turn's. - const implementationClientUserMessageId = insertion !== undefined ? randomUUID() : undefined; - if (implementationClientUserMessageId !== undefined) { - registerSyntheticInsertion(implementationClientUserMessageId, implementationRequest.prompt); - } + const implementationClientUserMessageId = `${PLAN_IMPLEMENTATION_MESSAGE_ID_PREFIX}${randomUUID()}`; const implementationPromise = this.runOnAppServer(promptGeneration, promptClient, () => promptClient.sendPrompt( implementationRequest, @@ -5580,7 +5595,7 @@ export class CodexAcpServer { let questions = this.asyncQuestions.get(sessionId); if (questions === undefined) { questions = new AsyncQuestionElicitations(sessionId, this.clientCapabilities, new AsyncQuestionClient( - sessionId, this.connection, this.clientCapabilities, () => this.codexAcpClient, + sessionId, this.connection, this.clientCapabilities, this, )); this.asyncQuestions.set(sessionId, questions); } diff --git a/src/CodexAsyncQuestions.ts b/src/CodexAsyncQuestions.ts index 23f795539..61b9ac1c4 100644 --- a/src/CodexAsyncQuestions.ts +++ b/src/CodexAsyncQuestions.ts @@ -7,20 +7,18 @@ * [{"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 adapter keeps the form open until the user answers or the session closes. The answer steers the running + * turn, or starts a new turn when the session is idle. On replay, the reply envelope shows the questions and 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 type {CodexAcpServer} from "./CodexAcpServer"; import {clientSupportsFormElicitation} from "./ElicitationCapabilities"; import {logger} from "./Logger"; import {clientSupportsNotices, createSessionNotice} from "./SessionNotice"; @@ -96,13 +94,13 @@ 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. + * @param server Owns steering and the queue for starting turns. */ constructor( private readonly sessionId: string, private readonly connection: AcpClientConnection, private readonly clientCapabilities: acp.ClientCapabilities | null, - private readonly codexAcpClient: () => CodexAcpClient, + private readonly server: CodexAcpServer, ) {} /** Sends the elicitation without marking the session as waiting for the user. */ @@ -113,14 +111,13 @@ export class AsyncQuestionClient { : 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, + /** Sends the reply through the session's shared steering and turn-start queue. */ + async sendReply(text: string): Promise { + const response = await this.server.executeOrQueueSteeringRequest({ + sessionId: this.sessionId, prompt: [{type: "text", text}], - clientUserMessageId: randomUUID(), }); + if (response.outcome === "failed") throw new Error("Codex did not accept the answer"); } /** Tells the user that an answer was not sent, if the client supports notices. */ @@ -131,16 +128,15 @@ export class AsyncQuestionClient { update: createSessionNotice( "warning", "Answer not sent", - "Codex finished before your answer arrived. Send it as a new message if it still matters.", + "The answer could not be sent. Send it as a new message to continue.", ), }); } } -/** Shows the async questions of one session as elicitations, for as long as the turn that asked them runs. */ +/** Keeps the session's async questions answerable between turns. */ export class AsyncQuestionElicitations { - // The running turns that asked a question, each with the signal that cancels its questions. - private readonly turns = new Map(); + private readonly requests = new Map(); constructor( private readonly sessionId: string, @@ -148,22 +144,17 @@ export class AsyncQuestionElicitations { private readonly client: AsyncQuestionClient, ) {} - /** Asks the questions of a completed async message and cancels them when their turn ends; never waits. */ + /** Opens the questions of a completed async message without waiting for an answer. */ handleNotification(notification: ServerNotification): void { switch (notification.method) { case "item/completed": { - const {threadId, turnId, item} = notification.params; + const {threadId, item} = notification.params; if (threadId === this.sessionId && item.type === "agentMessage" && item.delivery === "async" && (item.questions?.length ?? 0) > 0) { - this.ask(turnId, item); + this.ask(item); } return; } - case "turn/completed": - if (notification.params.threadId === this.sessionId) { - this.endTurn(notification.params.turn.id); - } - return; default: return; } @@ -171,41 +162,33 @@ export class AsyncQuestionElicitations { /** 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); - } + for (const request of this.requests.values()) request.abort(); + this.requests.clear(); } - 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); - } + private ask(message: AsyncQuestionMessage): void { + if (!clientSupportsFormElicitation(this.clientCapabilities) || this.requests.has(message.id)) return; + const pending = new AbortController(); + this.requests.set(message.id, pending); 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.", + message: "Codex keeps working while you answer. Your answer continues the conversation.", // 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) + void this.client.createElicitation(request, pending.signal).then( + response => this.answer(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}), - ); + error => logger.log("Async question closed without an answer", {sessionId: this.sessionId, error}), + ).finally(() => { + if (this.requests.get(message.id) === pending) this.requests.delete(message.id); + }); } private async answer( - turnId: string, message: AsyncQuestionMessage, questions: readonly ToolRequestUserInputQuestion[], response: acp.CreateElicitationResponse, @@ -220,15 +203,15 @@ export class AsyncQuestionElicitations { }]; }); if (replies.length === 0) return; - if (this.turns.has(turnId)) { + if (this.requests.has(message.id)) { try { - await this.client.steer(turnId, formatAsyncQuestionReply(replies)); + await this.client.sendReply(formatAsyncQuestionReply(replies)); return; } catch (error) { - logger.log("Async question answer was not steered into its turn", {sessionId: this.sessionId, turnId, error}); + logger.log("Async question answer was not delivered", {sessionId: this.sessionId, error}); } } - logger.log("Dropped an async question answer: its turn has ended", {sessionId: this.sessionId, turnId}); + logger.log("Dropped an async question answer", {sessionId: this.sessionId}); await this.client.reportDroppedAnswer(); } } diff --git a/src/SteerLanding.ts b/src/SteerLanding.ts new file mode 100644 index 000000000..6f5bb4f39 --- /dev/null +++ b/src/SteerLanding.ts @@ -0,0 +1,16 @@ +import type * as acp from "@agentclientprotocol/sdk/unstable"; + +export type SteerLandingResult = {kind: "inserted"} | {kind: "ended"; cancelled: boolean}; + +/** Tracks insertion independently of the turn/steer response. */ +export class SteerLanding { + readonly result: Promise; + private resolve: (result: SteerLandingResult) => void = () => {}; + + constructor(readonly sessionId: string, readonly turnId: string, readonly prompt: acp.ContentBlock[]) { + this.result = new Promise(resolve => { this.resolve = resolve; }); + } + + inserted(): void { this.resolve({kind: "inserted"}); } + ended(cancelled: boolean): void { this.resolve({kind: "ended", cancelled}); } +} diff --git a/src/SteeringQueue.ts b/src/SteeringQueue.ts index c1f553b1d..8615f9547 100644 --- a/src/SteeringQueue.ts +++ b/src/SteeringQueue.ts @@ -1,7 +1,11 @@ +import type {V2SteeringPrompt} from "./AcpV2Prompt"; import type {SessionSteerRequest, SessionSteeringResponse} from "./AcpExtensions"; +import type {TurnStartReservation} from "./TurnStartReservation"; interface QueuedSteering { params: SessionSteerRequest; + insertion: V2SteeringPrompt | undefined; + reservation: TurnStartReservation | undefined; resolve: (response: SessionSteeringResponse) => void; reject: (error: unknown) => void; } @@ -16,12 +20,12 @@ export class SteeringQueue { private processing = false; constructor( - private readonly handle: (params: SessionSteerRequest) => Promise, + private readonly handle: (params: SessionSteerRequest, insertion?: V2SteeringPrompt, reservation?: TurnStartReservation) => Promise, ) {} - enqueue(params: SessionSteerRequest): Promise { + enqueue(params: SessionSteerRequest, insertion?: V2SteeringPrompt, reservation?: TurnStartReservation): Promise { return new Promise((resolve, reject) => { - this.pending.push({params, resolve, reject}); + this.pending.push({params, insertion, reservation, resolve, reject}); this.startConsumer(); }); } @@ -44,7 +48,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.insertion, next.reservation)); } catch (error) { next.reject(error); // one failed steer must not stall the rest } diff --git a/src/TurnStartReservation.ts b/src/TurnStartReservation.ts new file mode 100644 index 000000000..ce2ad3cdc --- /dev/null +++ b/src/TurnStartReservation.ts @@ -0,0 +1,9 @@ +/** + * A place in the session's turn-start queue. Release it once the turn ends. + * Skip `await wait` when `needsWait` is false to preserve synchronous prompt setup. + */ +export interface TurnStartReservation { + wait: Promise; + needsWait: boolean; + release: () => void; +} diff --git a/src/__tests__/AcpV2Elicitation.test.ts b/src/__tests__/AcpV2Elicitation.test.ts new file mode 100644 index 000000000..db1ccdea4 --- /dev/null +++ b/src/__tests__/AcpV2Elicitation.test.ts @@ -0,0 +1,40 @@ +import {describe, expect, it} from "vitest"; +import type * as acp from "@agentclientprotocol/sdk/unstable"; +import {toV2CreateElicitationRequest} from "../AcpV2Elicitation"; + +function form(meta: unknown): acp.CreateElicitationRequest { + return {sessionId: "session", mode: "form", message: "Choose", _meta: {codex: {delivery: "async"}}, + requestedSchema: {type: "object", properties: { + choice: {type: "string", enum: ["Tea", "Coffee"]}, + note: {type: "string", _meta: meta}, + }}} as acp.CreateElicitationRequest; +} + +describe("v2 elicitation companions", () => { + it.each([ + {jetbrains: {air: {customAnswer: true}}}, + {jetbrains: {air: {customAnswer: {label: "Note"}}}}, + {_askUserQuestionCustomAnswer: true}, + ])("copies companion IDs to request metadata and keeps other namespaces: %j", meta => { + const request = form(meta); + const converted = toV2CreateElicitationRequest(request); + expect(converted).toMatchObject({requestedSchema: {properties: {note: {_meta: meta}}}}); + expect(converted._meta).toEqual({ + codex: {delivery: "async"}, jetbrains: {air: {version: 1, customAnswerFields: ["note"]}}, + }); + expect(request._meta).toEqual({codex: {delivery: "async"}}); + }); + + it.each([null, {}, {customAnswer: true}, {_askUserQuestionCustomAnswer: false}, + {jetbrains: {air: {customAnswer: false}}, _askUserQuestionCustomAnswer: true}, + ])("keeps unmarked forms unchanged: %j", meta => { + const request = form(meta); + expect(toV2CreateElicitationRequest(request)).toBe(request); + }); + + it("keeps URL requests unchanged", () => { + const request: acp.CreateElicitationRequest = {sessionId: "session", mode: "url", message: "Sign in", + url: "https://example.com", elicitationId: "login"}; + expect(toV2CreateElicitationRequest(request)).toBe(request); + }); +}); diff --git a/src/__tests__/CodexACPAgent/async-questions.test.ts b/src/__tests__/CodexACPAgent/async-questions.test.ts index 92bd24b8a..a57eb78e7 100644 --- a/src/__tests__/CodexACPAgent/async-questions.test.ts +++ b/src/__tests__/CodexACPAgent/async-questions.test.ts @@ -1,15 +1,19 @@ import {afterEach, describe, expect, it, vi} from 'vitest'; +import {RequestError} from '@agentclientprotocol/sdk/unstable'; import type * as acpV2 from '@agentclientprotocol/sdk/experimental/v2/unstable'; import type {ThreadItem} from '../../app-server/v2'; import { connectSession, + createTurn, dump, itemCompleted, itemStarted, type PromptSession, settle, stateUpdates, + sessionId, turnCompleted, + turnFinished, turnStarted, userMessageItem, } from './v2-prompt-harness'; @@ -108,10 +112,69 @@ describe.each([1, 2] as const)('async Codex questions over ACP v%i', (protocolVe expect(notices(client)).toEqual([]); }); - it('cancels the questions when their turn ends, and reports an answer that arrives after that', async () => { + it('keeps the form open after completion and starts one turn for its answer', async () => { + let answer!: (response: acpV2.CreateElicitationResponse) => void; + let signal!: AbortSignal; + const {client, steers} = await connect((_, pending) => { + signal = pending; + return new Promise(resolve => { answer = resolve; }); + }); + await askDuringTurn(client); + await vi.waitFor(() => expect(elicitationRequests(client)).toHaveLength(1)); + await finishTurn(client); + expect(signal.aborted).toBe(false); + answer({action: "accept", content: {question_1: "After completion"}}); + await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); + const params = client.turnStartParams[1]!; + client.emit(turnStarted("turn-2")); + client.emit(itemCompleted(userMessageItem(params["clientUserMessageId"] as string), "turn-2")); + client.emit(turnCompleted("turn-2")); + await client.promptRunFinished(1); + await settle(); + expect(steers).toEqual([]); + expect(notices(client)).toEqual([]); + if (protocolVersion === 2) { + const users = client.transcript.filter(entry => "sessionUpdate" in entry + && entry.sessionUpdate.sessionUpdate === "user_message_chunk"); + expect(users).toHaveLength(1); + } + await expect(dump({input: params["input"], states: stateUpdates(client.transcript)})) + .toMatchFileSnapshot(`data/async-questions-v${protocolVersion}-continued.json`); + }); + + if (protocolVersion === 2) { + it('delivers an async answer queued between cancelled follow-ups', async () => { + let answer!: (response: acpV2.CreateElicitationResponse) => void; + const {client, steers} = await connect(() => new Promise(resolve => { answer = resolve; })); + await askDuringTurn(client); + await vi.waitFor(() => expect(elicitationRequests(client)).toHaveLength(1)); + const pending = client.sendPrompt([{type: "text", text: "Pending"}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + const enqueue = vi.spyOn(client.agent, "executeOrQueueSteeringRequest"); + answer({action: "accept", content: {question_1: "Async answer"}}); + await vi.waitFor(() => expect(enqueue).toHaveBeenCalledTimes(1)); + const later = client.sendPrompt([{type: "text", text: "Later"}]); + await vi.waitFor(() => expect(enqueue).toHaveBeenCalledTimes(2)); + await client.cancel(); + await vi.waitFor(() => expect(client.turnInterruptCalls()).toHaveLength(1)); + client.emit(turnFinished("interrupted")); + await expect(pending).rejects.toMatchObject({code: -32800}); + await expect(later).rejects.toMatchObject({code: -32800}); + await client.promptRunFinished(); + await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); + const params = client.turnStartParams[1]!; + expect(params["input"]).toEqual([expect.objectContaining({text: expect.stringContaining("Async answer")})]); + client.emit(turnStarted("turn-2")); + client.emit(itemCompleted(userMessageItem(params["clientUserMessageId"] as string), "turn-2")); + client.emit(turnCompleted("turn-2")); + await client.promptRunFinished(1); + expect(notices(client)).toEqual([]); + }); + } + + it('cancels an open form when its session closes and drops a stale answer', 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"}}); @@ -120,22 +183,44 @@ describe.each([1, 2] as const)('async Codex questions over ACP v%i', (protocolVe await askDuringTurn(client); await vi.waitFor(() => expect(elicitationRequests(client)).toHaveLength(1)); await finishTurn(client); + await client.agent.closeSession({sessionId}); + await vi.waitFor(() => expect(cancelled).toBe(true)); + await settle(); + expect(steers).toEqual([]); + expect(client.turnStartParams).toHaveLength(1); + }); + + it('reports a rejected continuation and releases the next prompt', async () => { + let answer!: (response: acpV2.CreateElicitationResponse) => void; + const {client} = await connect(() => new Promise(resolve => { answer = resolve; })); + await askDuringTurn(client); + await vi.waitFor(() => expect(elicitationRequests(client)).toHaveLength(1)); + await finishTurn(client); + client.setTurnStart(async () => { throw RequestError.internalError(undefined, "Cannot start turn"); }); + answer({action: "accept", content: {question_1: "After completion"}}); await vi.waitFor(() => expect(notices(client)).toHaveLength(1)); + await client.promptRunFinished(1); + await expect(dump(notices(client))).toMatchFileSnapshot(`data/async-questions-v${protocolVersion}-rejected.json`); - expect(cancelled).toBe(true); - expect(steers).toEqual([]); - await expect(dump(notices(client))).toMatchFileSnapshot(`data/async-questions-v${protocolVersion}-dropped.json`); + client.setTurnStart(async () => ({turn: createTurn("inProgress", "turn-3")})); + const response = client.sendPrompt([{type: "text", text: "Try again"}]); + await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(3)); + const params = client.turnStartParams[2]!; + client.emit(turnStarted("turn-3")); + client.emit(itemCompleted(userMessageItem(params["clientUserMessageId"] as string), "turn-3")); + client.emit(turnCompleted("turn-3")); + await response; + await client.promptRunFinished(2); }); - it('reports an answer that Codex rejects because its turn has ended', async () => { + it('reports an answer that Codex rejects while its turn still runs', async () => { const {client, steers} = await connect( - async () => ({action: "accept", content: {question_1: "Late"}}), - async () => { throw new Error("no active turn to steer"); }, + async () => ({action: "accept", content: {question_1: "Rejected"}}), + async () => { throw new Error("Invalid answer"); }, ); 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`); }); diff --git a/src/__tests__/CodexACPAgent/async-task-stop-v2.test.ts b/src/__tests__/CodexACPAgent/async-task-stop-v2.test.ts index e546b01ab..a8cd499f7 100644 --- a/src/__tests__/CodexACPAgent/async-task-stop-v2.test.ts +++ b/src/__tests__/CodexACPAgent/async-task-stop-v2.test.ts @@ -59,7 +59,7 @@ describe('_session/async_task/stop on ACP v2', () => { expectConformingV2SessionUpdates(); }); - it('stops a spawned async task on v2, returning {stopped: true} and rendering _async_task_state_update', async () => { + it('stops a spawned async task on v2, returning {stopped: true} and rendering async_task_state_update', async () => { const client = await connectSession(2, {clientCapabilities: asyncTasksClientCapabilities}); closeClient = () => client.connection.close(); client.setCodexResponse("thread/backgroundTerminals/list", async () => ({ @@ -83,12 +83,8 @@ describe('_session/async_task/stop on ACP v2', () => { expect(stateUpdates).toEqual([{sessionUpdate: "_async_task_state_update", asyncTaskId: "command-1", state: "stopped", toolCallId: "command-1"}]); }); - // v1 renders the same lifecycle as bare `async_task_spawned`/`async_task_state_update` tags - // (pre-`_`-prefix convention), which the real v1 SDK client schema does not model -- see - // `async-tasks.test.ts` for the v1 rendering pin, using a raw mock connection instead of the - // real wire-validating client. This file's v1 iterations below pin the *router-level* - // extension-method behavior (params, unknown-session/task handling, errors), which is - // unaffected by that. + // AIR uses the same lifecycle tags in both versions. The v1 SDK schema omits these extensions, + // so async-tasks.test.ts captures the v1 lifecycle through a raw connection. it('stopping an unknown task id returns {stopped: false} on both v1 and v2, matching an unknown session', async () => { for (const protocolVersion of [1, 2] as const) { diff --git a/src/__tests__/CodexACPAgent/cancel-v2.test.ts b/src/__tests__/CodexACPAgent/cancel-v2.test.ts index e6459f5f5..2918a0f2a 100644 --- a/src/__tests__/CodexACPAgent/cancel-v2.test.ts +++ b/src/__tests__/CodexACPAgent/cancel-v2.test.ts @@ -94,8 +94,8 @@ describe('session/cancel and session/close over ACP v2', () => { await first; // B and C queue behind the running turn: nothing about them is observable yet. - const second = client.sendPrompt([{type: "text", text: "Two"}]); - const third = client.sendPrompt([{type: "text", text: "Three"}]); + const second = client.sendPrompt([{type: "text", text: "/status"}]); + const third = client.sendPrompt([{type: "text", text: "/plan"}]); await settle(); expect(client.turnStartParams).toHaveLength(1); @@ -107,7 +107,7 @@ describe('session/cancel and session/close over ACP v2', () => { await client.promptRunFinished(); await settle(); - // Neither B ("Two") nor C ("Three") ever reached `turn/start` or sent a user message. + // Neither queued command reaches Codex or sends a user message. expect(client.turnStartParams).toHaveLength(1); const userMessageTexts = client.transcript.flatMap(entry => { if (!("sessionUpdate" in entry) || entry.sessionUpdate.sessionUpdate !== "user_message_chunk") { @@ -116,7 +116,7 @@ describe('session/cancel and session/close over ACP v2', () => { const content = entry.sessionUpdate.content as {type: string, text?: string}; return content.type === "text" ? [content.text] : []; }); - expect(userMessageTexts).not.toEqual(expect.arrayContaining(["Two", "Three"])); + expect(userMessageTexts).not.toEqual(expect.arrayContaining(["/status", "/plan"])); expect(stateUpdates(client.transcript)).toEqual([{state: "running"}, {state: "idle", stopReason: "cancelled"}]); // The queue is empty afterwards: a fresh prompt runs like normal, not as if still queued. @@ -158,7 +158,7 @@ describe('session/cancel and session/close over ACP v2', () => { client.emit(turnStarted()); client.emit(itemCompleted(userMessageItem(client.turnStartParams[0]!["clientUserMessageId"] as string))); await first; - const beforeCancel = client.sendPrompt([{type: "text", text: "Before cancellation"}]); + const beforeCancel = client.sendPrompt([{type: "text", text: "/status"}]); await settle(); await client.cancel(); await interrupting.promise; @@ -196,7 +196,7 @@ describe('session/cancel and session/close over ACP v2', () => { client.emit(itemCompleted(userMessageItem(client.turnStartParams[0]!["clientUserMessageId"] as string))); await first; - const second = client.sendPrompt([{type: "text", text: "Two"}]); + const second = client.sendPrompt([{type: "text", text: "/status"}]); await settle(); expect(client.turnStartParams).toHaveLength(1); @@ -251,8 +251,8 @@ describe('session/cancel and session/close over ACP v2', () => { // B and C both queue behind the running turn. const controllerB = new AbortController(); - const second = client.sendPrompt([{type: "text", text: "Two"}], controllerB.signal); - const third = client.sendPrompt([{type: "text", text: "Three"}]); + const second = client.sendPrompt([{type: "text", text: "/status"}], controllerB.signal); + const third = client.sendPrompt([{type: "text", text: "/plan"}]); await settle(); expect(client.turnStartParams).toHaveLength(1); @@ -266,15 +266,10 @@ describe('session/cancel and session/close over ACP v2', () => { await client.promptRunFinished(); await settle(); - // C still runs once A finishes: dropping B did not disturb its FIFO slot. - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); - client.emit(turnStarted("turn-2")); - client.emit(itemCompleted(userMessageItem(client.turnStartParams[1]!["clientUserMessageId"] as string, "Three"), "turn-2")); - const {messageId} = await third; - expect(messageId).toBe(client.turnStartParams[1]!["clientUserMessageId"]); - client.emit({method: "turn/completed", params: {threadId: "thread-1", turn: createTurn("completed", "turn-2")}}); + await third; await client.promptRunFinished(1); await settle(); + expect(client.turnStartParams).toHaveLength(1); const userMessageTexts = client.transcript.flatMap(entry => { if (!("sessionUpdate" in entry) || entry.sessionUpdate.sessionUpdate !== "user_message_chunk") { @@ -283,8 +278,8 @@ describe('session/cancel and session/close over ACP v2', () => { const content = entry.sessionUpdate.content as {type: string, text?: string}; return content.type === "text" ? [content.text] : []; }); - expect(userMessageTexts).not.toEqual(expect.arrayContaining(["Two"])); - expect(userMessageTexts).toEqual(expect.arrayContaining(["One", "Three"])); + expect(userMessageTexts).not.toEqual(expect.arrayContaining(["/status"])); + expect(userMessageTexts).toEqual(expect.arrayContaining(["One", "/plan"])); expect(stateUpdates(client.transcript)).toEqual([ {state: "running"}, {state: "idle", stopReason: "end_turn"}, {state: "running"}, {state: "idle", stopReason: "end_turn"}, diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v1-answered.json b/src/__tests__/CodexACPAgent/data/async-questions-v1-answered.json index f79255dc8..4b9cf0ef4 100644 --- a/src/__tests__/CodexACPAgent/data/async-questions-v1-answered.json +++ b/src/__tests__/CodexACPAgent/data/async-questions-v1-answered.json @@ -54,7 +54,7 @@ } }, "mode": "form", - "message": "Codex keeps working while you answer. An answer sent after it finishes is not delivered.", + "message": "Codex keeps working while you answer. Your answer continues the conversation.", "_meta": { "codex": { "delivery": "async" diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v1-continued.json b/src/__tests__/CodexACPAgent/data/async-questions-v1-continued.json new file mode 100644 index 000000000..5519d71ea --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v1-continued.json @@ -0,0 +1,10 @@ +{ + "input": [ + { + "type": "text", + "text": "\n[{\"answer\":\"After completion\",\"question\":\"Anything else?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",1]\"}]\n", + "text_elements": [] + } + ], + "states": [] +} diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v1-dropped.json b/src/__tests__/CodexACPAgent/data/async-questions-v1-dropped.json deleted file mode 100644 index 7057306d8..000000000 --- a/src/__tests__/CodexACPAgent/data/async-questions-v1-dropped.json +++ /dev/null @@ -1,8 +0,0 @@ -[ - { - "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 index 7057306d8..ae7e72cd1 100644 --- a/src/__tests__/CodexACPAgent/data/async-questions-v1-rejected.json +++ b/src/__tests__/CodexACPAgent/data/async-questions-v1-rejected.json @@ -2,7 +2,7 @@ { "severity": "warning", "title": "Answer not sent", - "description": "Codex finished before your answer arrived. Send it as a new message if it still matters.", + "description": "The answer could not be sent. Send it as a new message to continue.", "sessionUpdate": "notice" } ] diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v2-answered.json b/src/__tests__/CodexACPAgent/data/async-questions-v2-answered.json index f79255dc8..4b9cf0ef4 100644 --- a/src/__tests__/CodexACPAgent/data/async-questions-v2-answered.json +++ b/src/__tests__/CodexACPAgent/data/async-questions-v2-answered.json @@ -54,7 +54,7 @@ } }, "mode": "form", - "message": "Codex keeps working while you answer. An answer sent after it finishes is not delivered.", + "message": "Codex keeps working while you answer. Your answer continues the conversation.", "_meta": { "codex": { "delivery": "async" diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v2-continued.json b/src/__tests__/CodexACPAgent/data/async-questions-v2-continued.json new file mode 100644 index 000000000..943869bca --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/async-questions-v2-continued.json @@ -0,0 +1,25 @@ +{ + "input": [ + { + "type": "text", + "text": "\n[{\"answer\":\"After completion\",\"question\":\"Anything else?\",\"questionItemId\":\"[\\\"request_user_input_async\\\",\\\"call-question\\\",1]\"}]\n", + "text_elements": [] + } + ], + "states": [ + { + "state": "running" + }, + { + "state": "idle", + "stopReason": "end_turn" + }, + { + "state": "running" + }, + { + "state": "idle", + "stopReason": "end_turn" + } + ] +} diff --git a/src/__tests__/CodexACPAgent/data/async-questions-v2-dropped.json b/src/__tests__/CodexACPAgent/data/async-questions-v2-dropped.json deleted file mode 100644 index 7057306d8..000000000 --- a/src/__tests__/CodexACPAgent/data/async-questions-v2-dropped.json +++ /dev/null @@ -1,8 +0,0 @@ -[ - { - "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 index 7057306d8..ae7e72cd1 100644 --- a/src/__tests__/CodexACPAgent/data/async-questions-v2-rejected.json +++ b/src/__tests__/CodexACPAgent/data/async-questions-v2-rejected.json @@ -2,7 +2,7 @@ { "severity": "warning", "title": "Answer not sent", - "description": "Codex finished before your answer arrived. Send it as a new message if it still matters.", + "description": "The answer could not be sent. Send it as a new message to continue.", "sessionUpdate": "notice" } ] diff --git a/src/__tests__/CodexACPAgent/data/live-core-v1-terminal.json b/src/__tests__/CodexACPAgent/data/live-core-v1-terminal.json index 24f5ac0b6..0d2d72677 100644 --- a/src/__tests__/CodexACPAgent/data/live-core-v1-terminal.json +++ b/src/__tests__/CodexACPAgent/data/live-core-v1-terminal.json @@ -63,34 +63,6 @@ "sessionUpdate": "subagent_update" } }, - { - "sessionId": "thread-1:subagent:worker", - "update": { - "toolCallId": "terminal", - "_meta": { - "terminal_output_delta": { - "data": "Still streaming", - "terminal_id": "terminal" - } - }, - "sessionUpdate": "tool_call_update" - } - }, - { - "sessionId": "thread-1:subagent:worker", - "update": { - "toolCallId": "terminal", - "status": "completed", - "_meta": { - "terminal_exit": { - "exit_code": 0, - "signal": null, - "terminal_id": "terminal" - } - }, - "sessionUpdate": "tool_call_update" - } - }, { "sessionId": "thread-1", "update": { @@ -118,5 +90,33 @@ "cwd": "/workspace", "sessionUpdate": "subagent_update" } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "terminal", + "_meta": { + "terminal_output_delta": { + "data": "Still streaming", + "terminal_id": "terminal" + } + }, + "sessionUpdate": "tool_call_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "terminal", + "status": "completed", + "_meta": { + "terminal_exit": { + "exit_code": 0, + "signal": null, + "terminal_id": "terminal" + } + }, + "sessionUpdate": "tool_call_update" + } } ] diff --git a/src/__tests__/CodexACPAgent/data/live-core-v2-ownership-owned.json b/src/__tests__/CodexACPAgent/data/live-core-v2-ownership-owned.json index 4d4641c5a..7e74d5b3f 100644 --- a/src/__tests__/CodexACPAgent/data/live-core-v2-ownership-owned.json +++ b/src/__tests__/CodexACPAgent/data/live-core-v2-ownership-owned.json @@ -25,16 +25,6 @@ "sessionUpdate": "subagent_update" } }, - { - "sessionId": "thread-1", - "update": { - "sessionId": "thread-1:subagent:worker", - "capabilities": { - "cancel": {} - }, - "sessionUpdate": "subagent_update" - } - }, { "sessionId": "thread-1", "update": { @@ -60,6 +50,16 @@ "sessionUpdate": "subagent_update" } }, + { + "sessionId": "thread-1", + "update": { + "sessionId": "thread-1:subagent:worker", + "capabilities": { + "cancel": {} + }, + "sessionUpdate": "subagent_update" + } + }, { "sessionId": "thread-1:subagent:worker", "update": { diff --git a/src/__tests__/CodexACPAgent/data/live-core-v2-terminal.json b/src/__tests__/CodexACPAgent/data/live-core-v2-terminal.json index 877262e6d..093a28fe4 100644 --- a/src/__tests__/CodexACPAgent/data/live-core-v2-terminal.json +++ b/src/__tests__/CodexACPAgent/data/live-core-v2-terminal.json @@ -73,33 +73,6 @@ "sessionUpdate": "subagent_update" } }, - { - "sessionId": "thread-1:subagent:worker", - "update": { - "terminalId": "terminal", - "data": "U3RpbGwgc3RyZWFtaW5n", - "sessionUpdate": "terminal_output_chunk" - } - }, - { - "sessionId": "thread-1:subagent:worker", - "update": { - "terminalId": "terminal", - "exitStatus": { - "exitCode": 0, - "signal": null - }, - "sessionUpdate": "terminal_update" - } - }, - { - "sessionId": "thread-1:subagent:worker", - "update": { - "toolCallId": "terminal", - "status": "completed", - "sessionUpdate": "tool_call_update" - } - }, { "sessionId": "thread-1:subagent:worker", "update": { @@ -134,5 +107,32 @@ "cwd": "/workspace", "sessionUpdate": "subagent_update" } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "terminalId": "terminal", + "data": "U3RpbGwgc3RyZWFtaW5n", + "sessionUpdate": "terminal_output_chunk" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "terminalId": "terminal", + "exitStatus": { + "exitCode": 0, + "signal": null + }, + "sessionUpdate": "terminal_update" + } + }, + { + "sessionId": "thread-1:subagent:worker", + "update": { + "toolCallId": "terminal", + "status": "completed", + "sessionUpdate": "tool_call_update" + } } ] diff --git a/src/__tests__/CodexACPAgent/data/permissions-v2-plan-implementation.json b/src/__tests__/CodexACPAgent/data/permissions-v2-plan-implementation.json index 2986a2175..89791cf87 100644 --- a/src/__tests__/CodexACPAgent/data/permissions-v2-plan-implementation.json +++ b/src/__tests__/CodexACPAgent/data/permissions-v2-plan-implementation.json @@ -246,16 +246,6 @@ "sessionUpdate": "state_update" } }, - { - "sessionUpdate": { - "messageId": "", - "content": { - "text": "Implement the approved plan.", - "type": "text" - }, - "sessionUpdate": "user_message_chunk" - } - }, { "sessionUpdate": { "title": "Hello", diff --git a/src/__tests__/CodexACPAgent/data/prompt-v1-plan-implementation-history.json b/src/__tests__/CodexACPAgent/data/prompt-v1-plan-implementation-history.json new file mode 100644 index 000000000..3cf5de8e9 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/prompt-v1-plan-implementation-history.json @@ -0,0 +1,26 @@ +[ + { + "content": { + "text": "Implement the approved plan.", + "type": "text" + }, + "messageId": "item-user", + "sessionUpdate": "user_message_chunk" + }, + { + "content": { + "text": "Implement the approved plan.", + "type": "text" + }, + "messageId": "external-user", + "sessionUpdate": "user_message_chunk" + }, + { + "content": { + "text": "Done.", + "type": "text" + }, + "messageId": "result", + "sessionUpdate": "agent_message_chunk" + } +] diff --git a/src/__tests__/CodexACPAgent/data/prompt-v1-plan-implementation-turn.json b/src/__tests__/CodexACPAgent/data/prompt-v1-plan-implementation-turn.json new file mode 100644 index 000000000..8cb11a943 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/prompt-v1-plan-implementation-turn.json @@ -0,0 +1,199 @@ +[ + { + "codexRequest": "turn/start", + "params": { + "threadId": "thread-1", + "input": [ + { + "type": "text", + "text": "Implement the approved plan.", + "text_elements": [] + } + ], + "approvalPolicy": "on-request", + "approvalsReviewer": "auto_review", + "sandboxPolicy": { + "type": "workspaceWrite", + "writableRoots": [], + "networkAccess": false, + "excludeTmpdirEnvVar": false, + "excludeSlashTmp": false + }, + "summary": "auto", + "effort": "medium", + "model": "gpt-5", + "serviceTier": null + } + }, + { + "codexResponse": "turn/start" + }, + { + "codexNotification": "turn/started" + }, + { + "codexNotification": "item/completed" + }, + { + "codexNotification": "item/completed" + }, + { + "codexNotification": "turn/completed" + }, + { + "sessionUpdate": { + "content": { + "text": "1. Do the change.", + "type": "text" + }, + "messageId": "plan-item", + "sessionUpdate": "agent_message_chunk" + } + }, + { + "sessionUpdate": { + "configOptions": [ + { + "currentValue": "agent", + "options": [ + { + "value": "read-only", + "name": "Read-only", + "description": "Requires approval to edit files and access the internet." + }, + { + "value": "workspace-write", + "name": "Workspace access", + "description": "Edit workspace files; ask before writing outside the workspace or accessing the network." + }, + { + "value": "agent", + "name": "Auto review", + "description": "Only ask for actions detected as potentially unsafe" + }, + { + "value": "agent-full-access", + "name": "Full access", + "description": "Unrestricted access to the internet and any file on your computer" + } + ], + "type": "select", + "id": "mode", + "name": "Mode", + "description": "Approval and sandboxing preset for the session", + "category": "mode" + }, + { + "currentValue": "default", + "options": [ + { + "value": "default", + "name": "Default" + }, + { + "value": "plan", + "name": "Plan", + "description": "Plan before making changes" + } + ], + "type": "select", + "id": "collaboration_mode", + "name": "Collaboration mode", + "description": "How Codex collaborates for subsequent turns", + "category": "collaboration_mode" + }, + { + "currentValue": "gpt-5", + "options": [ + { + "value": "gpt-5", + "name": "5", + "description": "gpt-5 model" + } + ], + "type": "select", + "id": "model", + "name": "Model", + "description": "Model Codex uses for the session", + "category": "model" + }, + { + "currentValue": "medium", + "options": [ + { + "value": "medium", + "name": "Medium", + "description": "Balanced" + } + ], + "type": "select", + "id": "reasoning_effort", + "name": "Reasoning effort", + "description": "How much reasoning effort the model should use", + "category": "thought_level" + } + ], + "sessionUpdate": "config_option_update" + } + }, + { + "codexRequest": "turn/start", + "params": { + "threadId": "thread-1", + "input": [ + { + "type": "text", + "text": "Implement the approved plan.", + "text_elements": [] + } + ], + "approvalPolicy": "on-request", + "approvalsReviewer": "auto_review", + "sandboxPolicy": { + "type": "workspaceWrite", + "writableRoots": [], + "networkAccess": false, + "excludeTmpdirEnvVar": false, + "excludeSlashTmp": false + }, + "summary": "auto", + "effort": "medium", + "model": "gpt-5", + "serviceTier": null, + "clientUserMessageId": "" + } + }, + { + "codexResponse": "turn/start" + }, + { + "codexNotification": "turn/started" + }, + { + "codexNotification": "item/started" + }, + { + "codexNotification": "item/completed" + }, + { + "codexNotification": "turn/completed" + }, + { + "sessionUpdate": { + "title": "Implement the approved plan.", + "sessionUpdate": "session_info_update" + } + }, + { + "promptResponse": { + "stopReason": "end_turn", + "usage": null, + "_meta": { + "quota": { + "token_count": null, + "model_usage": [] + } + } + } + } +] diff --git a/src/__tests__/CodexACPAgent/data/prompt-v2-plan-implementation-history.json b/src/__tests__/CodexACPAgent/data/prompt-v2-plan-implementation-history.json new file mode 100644 index 000000000..d388ed64f --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/prompt-v2-plan-implementation-history.json @@ -0,0 +1,26 @@ +[ + { + "messageId": "", + "content": { + "text": "Implement the approved plan.", + "type": "text" + }, + "sessionUpdate": "user_message_chunk" + }, + { + "messageId": "external-user", + "content": { + "text": "Implement the approved plan.", + "type": "text" + }, + "sessionUpdate": "user_message_chunk" + }, + { + "messageId": "result", + "content": { + "text": "Done.", + "type": "text" + }, + "sessionUpdate": "agent_message_chunk" + } +] diff --git a/src/__tests__/CodexACPAgent/data/prompt-v2-plan-implementation-turn.json b/src/__tests__/CodexACPAgent/data/prompt-v2-plan-implementation-turn.json index ea226dc28..bb650c86e 100644 --- a/src/__tests__/CodexACPAgent/data/prompt-v2-plan-implementation-turn.json +++ b/src/__tests__/CodexACPAgent/data/prompt-v2-plan-implementation-turn.json @@ -6,7 +6,7 @@ "input": [ { "type": "text", - "text": "Hello", + "text": "Implement the approved plan.", "text_elements": [] } ], @@ -45,7 +45,7 @@ "sessionUpdate": { "messageId": "", "content": { - "text": "Hello", + "text": "Implement the approved plan.", "type": "text" }, "sessionUpdate": "user_message_chunk" @@ -192,24 +192,17 @@ "codexNotification": "turn/started" }, { - "codexNotification": "item/completed" + "codexNotification": "item/started" }, { - "sessionUpdate": { - "messageId": "", - "content": { - "text": "Implement the approved plan.", - "type": "text" - }, - "sessionUpdate": "user_message_chunk" - } + "codexNotification": "item/completed" }, { "codexNotification": "turn/completed" }, { "sessionUpdate": { - "title": "Hello", + "title": "Implement the approved plan.", "sessionUpdate": "session_info_update" } }, diff --git a/src/__tests__/CodexACPAgent/data/session-update-v2-subagent-and-async-task.json b/src/__tests__/CodexACPAgent/data/session-update-v2-subagent-and-async-task.json index f84ecf4f5..774d9829f 100644 --- a/src/__tests__/CodexACPAgent/data/session-update-v2-subagent-and-async-task.json +++ b/src/__tests__/CodexACPAgent/data/session-update-v2-subagent-and-async-task.json @@ -6,7 +6,12 @@ "subagentSessionId": "child-1", "name": "Researcher", "task": "Look into flaky tests", - "capabilities": {} + "capabilities": { + "cancel": true + }, + "_meta": { + "source": "delegation" + } } }, { @@ -29,6 +34,15 @@ "toolCallId": "call-1" } }, + { + "sessionId": "session-1", + "update": { + "sessionUpdate": "_async_task_progress", + "asyncTaskId": "task-1", + "summary": "Tests are running", + "toolCallId": "call-1" + } + }, { "sessionId": "session-1", "update": { diff --git a/src/__tests__/CodexACPAgent/prompt-v2-steering.test.ts b/src/__tests__/CodexACPAgent/prompt-v2-steering.test.ts new file mode 100644 index 000000000..1a27810e3 --- /dev/null +++ b/src/__tests__/CodexACPAgent/prompt-v2-steering.test.ts @@ -0,0 +1,260 @@ +import {afterEach, describe, expect, it, vi} from 'vitest'; +import {ResponseError} from 'vscode-jsonrpc'; +import {SESSION_STEERING_METHOD} from '../../AcpExtensions'; +import { + sessionId, connectSession, userMessageItem, itemStarted, itemCompleted, + turnStarted, turnCompleted, turnFinished, settle, stateUpdates, + type PromptSession, +} from './v2-prompt-harness'; +import {expectConformingV2SessionUpdates} from './v2-session-update-guard'; + +type SteerParams = {threadId: string; expectedTurnId: string; clientUserMessageId: string}; + +async function startPrompt(client: PromptSession, version: 1 | 2 = 2): Promise { + const response = client.sendPrompt([{type: 'text', text: 'Write a long paragraph.'}]); + await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(1)); + client.emit(turnStarted()); + client.emit(itemCompleted(userMessageItem(client.turnStartParams[0]!['clientUserMessageId'] as string))); + if (version === 2) await response; +} + +function userMessages(client: PromptSession) { + return client.transcript.flatMap(entry => 'sessionUpdate' in entry ? [entry.sessionUpdate] : []) + .flatMap(update => update.sessionUpdate === 'user_message_chunk' ? [{messageId: update.messageId, content: update.content}] : []); +} + +describe('native v2 prompt steering', () => { + let client: PromptSession; + let steers: SteerParams[]; + + afterEach(() => { + client?.connection.close(); + vi.clearAllMocks(); + expectConformingV2SessionUpdates(); + }); + + async function connect(version: 1 | 2 = 2): Promise { + client = await connectSession(version); + steers = []; + client.setCodexResponse('turn/steer', async params => { + steers.push(params as SteerParams); + return {}; + }); + await startPrompt(client, version); + } + + async function insertPromptInTurn(turnNumber: number, text: string): Promise { + await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(turnNumber)); + const turnId = `turn-${turnNumber}`; + const messageId = client.turnStartParams[turnNumber - 1]!['clientUserMessageId'] as string; + client.emit(turnStarted(turnId)); + client.emit(itemCompleted(userMessageItem(messageId, text), turnId)); + return messageId; + } + + it.each([1, 2] as const)('delivers a follow-up during generation in v%s without restarting the turn', async version => { + await connect(version); + const response = version === 1 + ? client.request(SESSION_STEERING_METHOD, {sessionId, prompt: [{type: 'text', text: 'Answer briefly instead.'}]}) + : client.sendPrompt([{type: 'text', text: 'Answer briefly instead.'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + await settle(); + if (version === 2) { + expect(client.transcript.filter(entry => 'promptResponse' in entry)).toHaveLength(1); + client.emit(itemCompleted(userMessageItem('other-client', 'Unrelated'))); + await settle(); + expect(userMessages(client)).toHaveLength(1); + } + const messageId = steers[0]!.clientUserMessageId; + client.emit(itemStarted(userMessageItem(messageId, 'Answer briefly instead.'))); + client.emit(itemCompleted(userMessageItem(messageId, 'Answer briefly instead.'))); + expect(await response).toEqual(version === 1 ? {outcome: 'injected'} : {messageId}); + expect(client.turnStartParams).toHaveLength(1); + expect(client.turnInterruptCalls()).toEqual([]); + if (version === 2) { + expect(stateUpdates(client.transcript)).toEqual([{state: 'running'}]); + expect(userMessages(client)).toHaveLength(2); + } + client.emit(turnCompleted()); + await client.promptRunFinished(); + await settle(); + if (version === 2) expect(stateUpdates(client.transcript)) + .toEqual([{state: 'running'}, {state: 'idle', stopReason: 'end_turn'}]); + }); + + it('inserts several follow-ups in order within one running turn', async () => { + await connect(); + const second = client.sendPrompt([{type: 'text', text: 'Two'}]); + const third = client.sendPrompt([{type: 'text', text: 'Three'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + client.emit(itemCompleted(userMessageItem(steers[0]!.clientUserMessageId, 'Two'))); + await second; + await vi.waitFor(() => expect(steers).toHaveLength(2)); + client.emit(itemCompleted(userMessageItem(steers[1]!.clientUserMessageId, 'Three'))); + await third; + expect(steers[1]!.clientUserMessageId).not.toBe(steers[0]!.clientUserMessageId); + expect(userMessages(client).map(update => update.content)).toEqual([ + {type: 'text', text: 'Write a long paragraph.'}, + {type: 'text', text: 'Two'}, {type: 'text', text: 'Three'}, + ]); + expect(client.turnStartParams).toHaveLength(1); + client.emit(turnCompleted()); + await client.promptRunFinished(); + await settle(); + expect(stateUpdates(client.transcript)).toEqual([{state: 'running'}, {state: 'idle', stopReason: 'end_turn'}]); + }); + + it.each(['resolve', 'reject'] as const)('accepts a recorded input before a late steer response can %s', async outcome => { + await connect(); + let finish!: () => void; + const pending = new Promise((resolve, reject) => { + finish = outcome === 'resolve' ? resolve : () => reject(new Error('Late transport failure')); + }); + client.setCodexResponse('turn/steer', async params => { + steers.push(params as SteerParams); + await pending; + return {}; + }); + const response = client.sendPrompt([{type: 'text', text: 'Next'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + const messageId = steers[0]!.clientUserMessageId; + client.emit(itemCompleted(userMessageItem(messageId, 'Next'))); + expect(await response).toEqual({messageId}); + finish(); + await settle(); + expect(client.transcript.filter(entry => 'promptError' in entry)).toEqual([]); + client.emit(turnCompleted()); + await client.promptRunFinished(); + }); + + it('starts one new turn when the active turn ends before steering can accept the input', async () => { + await connect(); + let rejectSteer!: (error: unknown) => void; + client.setCodexResponse('turn/steer', async params => { + steers.push(params as SteerParams); + return await new Promise((_, reject) => { rejectSteer = reject; }); + }); + const response = client.sendPrompt([{type: 'text', text: 'Next'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + client.emit(turnCompleted()); + await client.promptRunFinished(); + rejectSteer(new ResponseError(-32600, 'No active turn to steer')); + await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); + const messageId = client.turnStartParams[1]!['clientUserMessageId'] as string; + expect(messageId).toBe(steers[0]!.clientUserMessageId); + client.emit(turnStarted('turn-2')); + client.emit(itemCompleted(userMessageItem(messageId, 'Next'), 'turn-2')); + expect(await response).toEqual({messageId}); + client.emit(turnCompleted('turn-2')); + await client.promptRunFinished(1); + await settle(); + expect(userMessages(client)).toHaveLength(2); + expect(stateUpdates(client.transcript)).toEqual([ + {state: 'running'}, {state: 'idle', stopReason: 'end_turn'}, + {state: 'running'}, {state: 'idle', stopReason: 'end_turn'}, + ]); + }); + + it.each(['completed', 'interrupted'] as const)('rejects input that a %s turn never recorded without retrying', async status => { + await connect(); + const response = client.sendPrompt([{type: 'text', text: 'Never inserted'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + client.emit(turnFinished(status)); + await expect(response).rejects.toMatchObject({code: status === 'interrupted' ? -32800 : -32603}); + await client.promptRunFinished(); + await settle(); + expect(client.turnStartParams).toHaveLength(1); + expect(userMessages(client)).toHaveLength(1); + }); + + it.each(['cancel', 'close'] as const)('settles an uninserted follow-up when session/%s stops its turn', async method => { + await connect(); + const response = client.sendPrompt([{type: 'text', text: 'Pending'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + const stopping = method === 'cancel' ? client.cancel() : client.request('session/close', {sessionId}); + await vi.waitFor(() => expect(client.turnInterruptCalls()).toHaveLength(1)); + client.emit(turnFinished('interrupted')); + await stopping; + await expect(response).rejects.toMatchObject({code: -32800}); + await client.promptRunFinished(); + await settle(); + expect(userMessages(client)).toHaveLength(1); + expect(stateUpdates(client.transcript)).toEqual([{state: 'running'}, {state: 'idle', stopReason: 'cancelled'}]); + }); + + it('keeps fallback steering ahead of later prompts after cancellation', async () => { + await connect(); + const pending = client.sendPrompt([{type: 'text', text: 'Pending'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + const enqueue = vi.spyOn(client.agent, 'executeOrQueueSteeringRequest'); + const steering = client.request(SESSION_STEERING_METHOD, { + sessionId, prompt: [{type: 'text', text: 'Fallback'}], + }); + steering.catch(() => {}); + await vi.waitFor(() => expect(enqueue).toHaveBeenCalledTimes(1)); + const later = client.sendPrompt([{type: 'text', text: 'Later'}]); + await vi.waitFor(() => expect(enqueue).toHaveBeenCalledTimes(2)); + await client.cancel(); + await vi.waitFor(() => expect(client.turnInterruptCalls()).toHaveLength(1)); + client.emit(turnFinished('interrupted')); + await expect(pending).rejects.toMatchObject({code: -32800}); + await expect(later).rejects.toMatchObject({code: -32800}); + await client.promptRunFinished(); + await insertPromptInTurn(2, 'Fallback'); + await expect(steering).resolves.toEqual({outcome: 'startedNewTurn'}); + client.emit(turnCompleted('turn-2')); + await client.promptRunFinished(1); + const next = client.sendPrompt([{type: 'text', text: 'After cancellation'}]); + const messageId = await insertPromptInTurn(3, 'After cancellation'); + await expect(next).resolves.toEqual({messageId}); + client.emit(turnCompleted('turn-3')); + await client.promptRunFinished(2); + }); + + it('reports a rejected steer without stopping the active turn', async () => { + await connect(); + client.setCodexResponse('turn/steer', async () => { + throw new ResponseError(-32602, 'Invalid input'); + }); + await expect(client.sendPrompt([{type: 'text', text: 'Rejected'}])) + .rejects.toMatchObject({code: -32602}); + expect(userMessages(client)).toHaveLength(1); + expect(stateUpdates(client.transcript)).toEqual([{state: 'running'}]); + expect(client.turnInterruptCalls()).toEqual([]); + client.emit(turnCompleted()); + await client.promptRunFinished(); + }); + + it('drops a cancelled follow-up before its steering dispatch and keeps the earlier input', async () => { + await connect(); + const second = client.sendPrompt([{type: 'text', text: 'Two'}]); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + const controller = new AbortController(); + const third = client.sendPrompt([{type: 'text', text: 'Three'}], controller.signal); + await settle(); + controller.abort(); + client.emit(itemCompleted(userMessageItem(steers[0]!.clientUserMessageId, 'Two'))); + await second; + await expect(third).rejects.toMatchObject({code: -32800}); + expect(steers).toHaveLength(1); + expect(userMessages(client)).toHaveLength(2); + expect(client.turnInterruptCalls()).toEqual([]); + client.emit(turnCompleted()); + await client.promptRunFinished(); + }); + + it('does not interrupt the active turn when an in-flight follow-up request is cancelled', async () => { + await connect(); + const controller = new AbortController(); + const response = client.sendPrompt([{type: 'text', text: 'Pending'}], controller.signal); + await vi.waitFor(() => expect(steers).toHaveLength(1)); + controller.abort(); + await settle(); + expect(client.turnInterruptCalls()).toEqual([]); + const messageId = steers[0]!.clientUserMessageId; + client.emit(itemCompleted(userMessageItem(messageId, 'Pending'))); + expect(await response).toEqual({messageId}); + client.emit(turnCompleted()); + await client.promptRunFinished(); + }); +}); diff --git a/src/__tests__/CodexACPAgent/prompt-v2.test.ts b/src/__tests__/CodexACPAgent/prompt-v2.test.ts index cd24ec2d4..9eccb7b31 100644 --- a/src/__tests__/CodexACPAgent/prompt-v2.test.ts +++ b/src/__tests__/CodexACPAgent/prompt-v2.test.ts @@ -5,6 +5,7 @@ import { turnId, connectSession, createTurn, + createThread, userMessageItem, itemStarted, itemCompleted, @@ -400,92 +401,6 @@ describe('session/prompt over ACP v2', () => { await expect(dump(client.transcript, clientUserMessageId)).toMatchFileSnapshot('data/prompt-v2-not-inserted.json'); }); - it('queues a prompt that overlaps a running one instead of rejecting it', async () => { - const client = await connectSession(); - closeClient = () => client.connection.close(); - - const first = client.sendPrompt([{type: "text", text: "Hello"}]); - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(1)); - client.emit(turnStarted()); - client.emit(itemCompleted(userMessageItem(client.turnStartParams[0]!["clientUserMessageId"] as string))); - await first; - - // B overlaps A: it is queued, not rejected. Nothing about it is observable yet. - const second = client.sendPrompt([{type: "text", text: "Again"}]); - await settle(); - expect(client.turnStartParams).toHaveLength(1); - expect(client.transcript.filter(entry => "promptResponse" in entry || "promptError" in entry)).toHaveLength(1); - expect(stateUpdates(client.transcript)).toEqual([{state: "running"}]); - - // A ends: idle(A) goes out, then B's turn starts. - client.emit(turnCompleted()); - await client.promptRunFinished(); - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); - expect(stateUpdates(client.transcript)).toEqual([ - {state: "running"}, {state: "idle", stopReason: "end_turn"}, {state: "running"}, - ]); - - const secondClientUserMessageId = client.turnStartParams[1]!["clientUserMessageId"] as string; - client.emit(turnStarted("turn-2")); - client.emit(itemCompleted(userMessageItem(secondClientUserMessageId, "Again"), "turn-2")); - const {messageId} = await second; - expect(messageId).toBe(secondClientUserMessageId); - - client.emit(turnCompleted("turn-2")); - await client.promptRunFinished(1); - await settle(); - - expect(stateUpdates(client.transcript)).toEqual([ - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - ]); - }); - - it('queues multiple overlapping prompts and runs them in FIFO order', async () => { - const client = await connectSession(); - closeClient = () => client.connection.close(); - - const first = client.sendPrompt([{type: "text", text: "One"}]); - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(1)); - client.emit(turnStarted()); - client.emit(itemCompleted(userMessageItem(client.turnStartParams[0]!["clientUserMessageId"] as string))); - await first; - - const second = client.sendPrompt([{type: "text", text: "Two"}]); - const third = client.sendPrompt([{type: "text", text: "Three"}]); - await settle(); - expect(client.turnStartParams).toHaveLength(1); - - client.emit(turnCompleted()); - await client.promptRunFinished(); - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); - // C stays queued behind B; it must not start alongside or ahead of it. - await settle(); - expect(client.turnStartParams).toHaveLength(2); - - const secondClientUserMessageId = client.turnStartParams[1]!["clientUserMessageId"] as string; - client.emit(turnStarted("turn-2")); - client.emit(itemCompleted(userMessageItem(secondClientUserMessageId, "Two"), "turn-2")); - await second; - client.emit(turnCompleted("turn-2")); - await client.promptRunFinished(1); - - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(3)); - const thirdClientUserMessageId = client.turnStartParams[2]!["clientUserMessageId"] as string; - client.emit(turnStarted("turn-3")); - client.emit(itemCompleted(userMessageItem(thirdClientUserMessageId, "Three"), "turn-3")); - await third; - client.emit(turnCompleted("turn-3")); - await client.promptRunFinished(2); - await settle(); - - expect(stateUpdates(client.transcript)).toEqual([ - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - ]); - }); - it('queues a local slash command behind a running prompt', async () => { const client = await connectSession(); closeClient = () => client.connection.close(); @@ -517,70 +432,28 @@ describe('session/prompt over ACP v2', () => { ]); }); - it('runs a queued prompt after the one ahead of it fails', async () => { + it('starts one turn for two prompts sent before either response is awaited', async () => { const client = await connectSession(); closeClient = () => client.connection.close(); - - await insertPrompt(client, 0); - const second = client.sendPrompt([{type: "text", text: "Again"}]); - await settle(); - expect(client.turnStartParams).toHaveLength(1); - - client.emit(turnError("usageLimitExceeded", "You've hit your usage limit.")); - client.emit(turnFinished("failed")); - await client.promptRunFinished(); - await settle(); - expect(stateUpdates(client.transcript)).toEqual([ - {state: "running"}, {state: "idle", stopReason: "error"}, {state: "running"}, - ]); - - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); - const secondClientUserMessageId = client.turnStartParams[1]!["clientUserMessageId"] as string; - client.emit(turnStarted("turn-2")); - client.emit(itemCompleted(userMessageItem(secondClientUserMessageId, "Again"), "turn-2")); - await second; - client.emit(turnCompleted("turn-2")); - await client.promptRunFinished(1); - await settle(); - - expect(stateUpdates(client.transcript)).toEqual([ - {state: "running"}, {state: "idle", stopReason: "error"}, - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - ]); - }); - - it('takes the turn-start reservation synchronously, so two prompts sent back-to-back never race', async () => { - const client = await connectSession(); - closeClient = () => client.connection.close(); - - // Neither request is awaited before the next is sent: this exercises the reservation - // being taken before either prompt's handler has had a chance to run any async work. + const steers: string[] = []; + client.setCodexResponse("turn/steer", async params => { + steers.push((params as {clientUserMessageId: string}).clientUserMessageId); + return {}; + }); const first = client.sendPrompt([{type: "text", text: "One"}]); const second = client.sendPrompt([{type: "text", text: "Two"}]); - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(1)); - await settle(); - expect(client.turnStartParams).toHaveLength(1); - client.emit(turnStarted()); client.emit(itemCompleted(userMessageItem(client.turnStartParams[0]!["clientUserMessageId"] as string))); await first; + await vi.waitFor(() => expect(steers).toHaveLength(1)); + client.emit(itemCompleted(userMessageItem(steers[0]!, "Two"))); + expect(await second).toEqual({messageId: steers[0]}); + expect(client.turnStartParams).toHaveLength(1); client.emit(turnCompleted()); await client.promptRunFinished(); - - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); - const secondClientUserMessageId = client.turnStartParams[1]!["clientUserMessageId"] as string; - client.emit(turnStarted("turn-2")); - client.emit(itemCompleted(userMessageItem(secondClientUserMessageId, "Two"), "turn-2")); - await second; - client.emit(turnCompleted("turn-2")); - await client.promptRunFinished(1); await settle(); - - expect(stateUpdates(client.transcript)).toEqual([ - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - {state: "running"}, {state: "idle", stopReason: "end_turn"}, - ]); + expect(stateUpdates(client.transcript)).toEqual([{state: "running"}, {state: "idle", stopReason: "end_turn"}]); }); it('keeps the turn going across a rendered tool call', async () => { @@ -881,54 +754,77 @@ describe('session/prompt over ACP v2', () => { await expect(dump(client.transcript, messageId)).toMatchFileSnapshot('data/prompt-v2-goal-resume-no-continuation.json'); }, 10000); - it('shows the plan-implementation follow-up turn as its own live user message, with its own minted id', async () => { - const client = await connectSession(); + it.each([1, 2] as const)('keeps the approved plan command internal in live and restored v%s chats', async (version) => { + const client = await connectSession(version); closeClient = () => client.connection.close(); vi.spyOn(CodexAcpServer.prototype as any, "requestPlanImplementationPermission").mockResolvedValue(true); - // Switch into plan mode first; `/plan` is a local command and starts no turn. await client.sendPrompt([{type: "text", text: "/plan"}]); await client.promptRunFinished(0); await settle(); expect(client.turnStartParams).toEqual([]); const start = client.transcript.length; - const response = client.sendPrompt([{type: "text", text: "Hello"}]); + // A real user can send the same text as the internal command. + const text = "Implement the approved plan."; + const response = client.sendPrompt([{type: "text", text}]); await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(1)); - const clientUserMessageId = client.turnStartParams[0]!["clientUserMessageId"] as string; + const clientUserMessageId = client.turnStartParams[0]!["clientUserMessageId"] as string | undefined; client.emit(turnStarted()); - client.emit(itemCompleted(userMessageItem(clientUserMessageId))); - const {messageId} = await response; + client.emit(itemCompleted(userMessageItem(clientUserMessageId ?? null, text))); + const messageId = version === 2 ? (await response).messageId as string : ""; client.emit(itemCompleted({type: "plan", id: "plan-item", text: "1. Do the change."})); client.emit(turnCompleted()); - // Approval is mocked, so codex-acp starts the "Implement the approved plan." turn itself, - // still inside this same prompt. await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); const implementationId = client.turnStartParams[1]!["clientUserMessageId"] as string; expect(implementationId).toEqual(expect.any(String)); expect(implementationId).not.toBe(messageId); client.emit(turnStarted("turn-2")); - client.emit(itemCompleted(userMessageItem(implementationId, "Implement the approved plan."), "turn-2")); - await settle(); - const userMessages = sessionUpdates(client.transcript.slice(start)).filter(update => update.sessionUpdate === "user_message_chunk"); - expect(userMessages).toEqual([ - {sessionUpdate: "user_message_chunk", messageId, content: {type: "text", text: "Hello"}}, - { - sessionUpdate: "user_message_chunk", - messageId: implementationId, - content: {type: "text", text: "Implement the approved plan."}, - }, - ]); - + const implementationItem = {...userMessageItem(implementationId, text), id: "implementation-item"}; + client.emit(itemStarted(implementationItem, "turn-2")); + client.emit(itemCompleted(implementationItem, "turn-2")); client.emit(turnCompleted("turn-2")); + await response; await client.promptRunFinished(1); await settle(); - // One running…idle pair for the whole exchange, despite the two turns. - expect(stateUpdates(client.transcript.slice(start))).toEqual([{state: "running"}, {state: "idle", stopReason: "end_turn"}]); + const userMessages = sessionUpdates(client.transcript.slice(start)).filter(update => update.sessionUpdate === "user_message_chunk"); + expect(userMessages).toEqual(version === 2 + ? [{sessionUpdate: "user_message_chunk", messageId, content: {type: "text", text}}] + : []); + expect(stateUpdates(client.transcript.slice(start))).toEqual(version === 2 + ? [{state: "running"}, {state: "idle", stopReason: "end_turn"}] + : []); await expect(dump(client.transcript.slice(start), messageId).replaceAll(implementationId, "")) - .toMatchFileSnapshot('data/prompt-v2-plan-implementation-turn.json'); + .toMatchFileSnapshot(`data/prompt-v${version}-plan-implementation-turn.json`); + client.connection.close(); + + // A fresh adapter must use the recorded identity, without a live insertion cache. + const restored = await connectSession(version); + closeClient = () => restored.connection.close(); + const thread = {...createThread(), turns: [{ + ...createTurn("completed"), itemsView: "full" as const, items: [ + userMessageItem(clientUserMessageId ?? null, text), + implementationItem, + {...userMessageItem(null, text), id: "external-user"}, + {type: "agentMessage" as const, id: "result", text: "Done.", phase: "final_answer" as const, + memoryCitation: null, delivery: null, questions: null}, + ], + }]}; + restored.setCodexResponse("thread/resume", async () => ({thread, model: "gpt-5", modelProvider: "openai", + reasoningEffort: "medium", serviceTier: null, turnsBackwardsCursor: null})); + restored.setCodexResponse("thread/read", async () => ({thread})); + if (version === 2) { + await restored.request("session/resume", {sessionId, cwd: "/workspace", replayFrom: {type: "start"}}); + } else { + await restored.request("session/load", {sessionId, cwd: "/workspace", mcpServers: []}); + } + await settle(); + const messages = sessionUpdates(restored.transcript).filter(update => update.sessionUpdate === "user_message_chunk" + || update.sessionUpdate === "agent_message_chunk"); + await expect(dump(messages, messageId)).toMatchFileSnapshot(`data/prompt-v${version}-plan-implementation-history.json`); + expect(messages.filter(update => update.sessionUpdate === "user_message_chunk")).toHaveLength(2); }); it('leaves v1 session/prompt unchanged: no clientUserMessageId, answered at turn end', async () => { diff --git a/src/__tests__/CodexACPAgent/session-index.test.ts b/src/__tests__/CodexACPAgent/session-index.test.ts index b30b1e362..f489fc01a 100644 --- a/src/__tests__/CodexACPAgent/session-index.test.ts +++ b/src/__tests__/CodexACPAgent/session-index.test.ts @@ -876,7 +876,7 @@ describe("_session/list/subscribe", () => { expect(listChanges(fixture).flatMap(changes => changes.sessions.map((row: acp.SessionInfo) => row.cwd))).toEqual([dir]); agent.sessionIndex.dispose(); - fs.rmSync(link); + fs.unlinkSync(link); } finally { fs.rmSync(dir, {recursive: true, force: true}); } diff --git a/src/__tests__/CodexACPAgent/session-update-v2.test.ts b/src/__tests__/CodexACPAgent/session-update-v2.test.ts index bb744df48..6749486f6 100644 --- a/src/__tests__/CodexACPAgent/session-update-v2.test.ts +++ b/src/__tests__/CodexACPAgent/session-update-v2.test.ts @@ -55,7 +55,7 @@ describe('ACPSessionConnection - session/update over ACP v2', () => { expectConformingV2SessionUpdates(); }); - async function connectV2Client() { + async function connectV2Client(capabilities?: acpV2.ClientCapabilities) { const {clientStream, agentConnection} = connectRouter(); const received: unknown[] = []; const connection = acpV2.client({name: "test-client"}) @@ -68,6 +68,7 @@ describe('ACPSessionConnection - session/update over ACP v2', () => { await connection.agent.request(acpV2.methods.agent.initialize, { protocolVersion: 2, info: {name: "test-client", version: "1.0.0"}, + ...(capabilities ? {capabilities} : {}), }); const handle = agentConnection(); if (!(handle instanceof AcpV2Connection)) { @@ -107,8 +108,10 @@ describe('ACPSessionConnection - session/update over ACP v2', () => { await expect(dump(received)).toMatchFileSnapshot('data/session-update-v2-tool-calls-and-chunks.json'); }); - it('renders the subagent RFD upsert and async task updates under `_`-prefixed tags', async () => { - const {received, view} = await connectV2Client(); + it('preserves the negotiated AIR lifecycle extensions on the v2 wire', async () => { + const {received, view} = await connectV2Client({ + _meta: {jetbrains: {air: {version: 1, capabilities: ["nativeSubagentSessions", "asyncTasks"]}}}, + }); const session = new ACPSessionConnection(view, "session-1"); await session.update({ @@ -116,7 +119,8 @@ describe('ACPSessionConnection - session/update over ACP v2', () => { subagentSessionId: "child-1", name: "Researcher", task: "Look into flaky tests", - capabilities: {}, + capabilities: {cancel: true}, + _meta: {source: "delegation"}, }); await session.update({sessionUpdate: "subagent_state_update", subagentSessionId: "child-1", state: "completed"}); await session.update({ @@ -128,6 +132,12 @@ describe('ACPSessionConnection - session/update over ACP v2', () => { canStop: true, toolCallId: "call-1", }); + await session.update({ + sessionUpdate: "async_task_progress", + asyncTaskId: "task-1", + summary: "Tests are running", + toolCallId: "call-1", + }); await session.update({ sessionUpdate: "async_task_state_update", asyncTaskId: "task-1", @@ -135,7 +145,7 @@ describe('ACPSessionConnection - session/update over ACP v2', () => { toolCallId: "call-1", }); - await vi.waitFor(() => expect(received).toHaveLength(4)); + await vi.waitFor(() => expect(received).toHaveLength(5)); await expect(dump(received)).toMatchFileSnapshot('data/session-update-v2-subagent-and-async-task.json'); }); diff --git a/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts b/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts index 32494f35c..91038201b 100644 --- a/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts +++ b/src/__tests__/CodexACPAgent/standard-subagent-live.test.ts @@ -63,6 +63,12 @@ describe.each([1, 2] as const)("ACP v%s live standard subagents", version => { }); const childUpdates = () => h.scopedUpdates.filter(entry => entry.sessionId.includes(":subagent:") || entry.update.sessionUpdate === "subagent_update"); + async function waitForChildDirectory() { + await vi.waitFor(() => expect(childUpdates()).toContainEqual({ + sessionId, update: {sessionUpdate: "subagent_update", sessionId: childId, cwd}, + })); + await flush(h); + } async function snapshot(name: string) { await flush(h); await expect(JSON.stringify(childUpdates(), null, 2) + "\n").toMatchFileSnapshot(`data/live-core-v${version}-${name}.json`); @@ -220,6 +226,7 @@ describe.each([1, 2] as const)("ACP v%s live standard subagents", version => { await load(); await flush(h); expect(resumes).toEqual(parent === sessionId ? [sessionId, worker] : [sessionId]); + if (parent === sessionId) await waitForChildDirectory(); h.emit(onWorker(turnStarted("child-turn"))); await flush(h); if (parent === sessionId) expect(await approval("restored")).toEqual({decision: "accept"}); @@ -235,7 +242,9 @@ describe.each([1, 2] as const)("ACP v%s live standard subagents", version => { await flush(h); history(h); await load(); + await waitForChildDirectory(); h.emit({method: "item/commandExecution/outputDelta", params: {threadId: worker, turnId: "child-turn", itemId: "terminal", delta: "Still streaming"}}); + await flush(h); h.emit({method: "item/completed", params: {threadId: worker, turnId: "child-turn", completedAtMs: 0, item: {...commandExecutionItem("terminal", "completed"), exitCode: 0, aggregatedOutput: "Still streaming"}}}); await snapshot("terminal"); diff --git a/src/__tests__/CodexACPAgent/steering-v2.test.ts b/src/__tests__/CodexACPAgent/steering-v2.test.ts index 96c8e5c16..b5b8d1397 100644 --- a/src/__tests__/CodexACPAgent/steering-v2.test.ts +++ b/src/__tests__/CodexACPAgent/steering-v2.test.ts @@ -183,44 +183,27 @@ describe('_session/steering over ACP v2', () => { await expect(dump(client.transcript, clientUserMessageId)).toMatchFileSnapshot('data/steering-v2-fallback-new-turn.json'); }); - it('queues a v2 prompt behind the steering fallback\'s new turn', async () => { + it('steers a v2 prompt into the turn that an extension fallback started', async () => { const client = await connectSession(); closeClient = () => client.connection.close(); - + const steerClientId = captureSteerClientId(client); const steerPromise = client.request(SESSION_STEERING_METHOD, { - sessionId, - prompt: [{type: "text", text: "start fresh"}], + sessionId, prompt: [{type: "text", text: "start fresh"}], }); await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(1)); - const steerClientUserMessageId = client.turnStartParams[0]!["clientUserMessageId"] as string; + const initialId = client.turnStartParams[0]!["clientUserMessageId"] as string; client.emit(turnStarted()); await steerPromise; - - // A v2 prompt arriving while the fallback turn runs must queue, not start its own turn/start. - const bResponse = client.sendPrompt([{type: "text", text: "B"}]); - await settle(); + client.emit(itemCompleted(userMessageItem(initialId, "start fresh"))); + const response = client.sendPrompt([{type: "text", text: "B"}]); + await vi.waitFor(() => expect(steerClientId.captured()).toEqual(expect.any(String))); + const messageId = steerClientId.captured(); + client.emit(itemCompleted(userMessageItem(messageId, "B"))); + expect(await response).toEqual({messageId}); expect(client.turnStartParams).toHaveLength(1); - - client.emit(itemCompleted(userMessageItem(steerClientUserMessageId, "start fresh"))); client.emit(turnCompleted()); - await client.promptRunFinished(0); - await settle(); - - await vi.waitFor(() => expect(client.turnStartParams).toHaveLength(2)); - const bClientUserMessageId = client.turnStartParams[1]!["clientUserMessageId"] as string; - client.emit(turnStarted("turn-2")); - client.emit(itemCompleted(userMessageItem(bClientUserMessageId, "B"), "turn-2")); - const {messageId: bMessageId} = await bResponse; - expect(bMessageId).toBe(bClientUserMessageId); - client.emit(turnCompleted("turn-2")); - await client.promptRunFinished(1); + await client.promptRunFinished(); await settle(); - - expect(stateUpdates(client.transcript)).toEqual([ - {state: "running"}, - {state: "idle", stopReason: "end_turn"}, - {state: "running"}, - {state: "idle", stopReason: "end_turn"}, - ]); + expect(stateUpdates(client.transcript)).toEqual([{state: "running"}, {state: "idle", stopReason: "end_turn"}]); }); }); diff --git a/src/__tests__/CodexACPAgent/subagent-and-async-task-updates-v2.test.ts b/src/__tests__/CodexACPAgent/subagent-and-async-task-updates-v2.test.ts index d1595fe71..7e1a55472 100644 --- a/src/__tests__/CodexACPAgent/subagent-and-async-task-updates-v2.test.ts +++ b/src/__tests__/CodexACPAgent/subagent-and-async-task-updates-v2.test.ts @@ -7,28 +7,28 @@ import {CodexAppServerClient} from '../../CodexAppServerClient'; import type {ThreadBackgroundTerminalsListResponse} from '../../async-tasks/BackgroundTerminalApi'; import {checkV2SessionUpdate, expectConformingV2SessionUpdates} from './v2-session-update-guard'; -/** A v2-wrapped `ACPSessionConnection`, so `session.update()` renders through `toV2SessionUpdate`. */ -function v2Session(sessionId: string): {session: ACPSessionConnection; updates: acpV2.SessionUpdate[]} { - const updates: acpV2.SessionUpdate[] = []; - const v2Client = { +/** Sends the negotiated AIR lifecycle through the selected protocol boundary. */ +function sessionForVersion(version: 1 | 2, sessionId: string): {session: ACPSessionConnection; updates: unknown[]} { + const updates: unknown[] = []; + const connection = { notify: vi.fn(async (_method: string, params: unknown) => { const update = (params as {update: acpV2.SessionUpdate}).update; - checkV2SessionUpdate(update); + if (version === 2) checkV2SessionUpdate(update); updates.push(update); }), request: vi.fn(), }; - const view = new AcpV2Connection(v2Client as any).extensionOnlyV1View(); + const view = version === 1 ? connection : new AcpV2Connection(connection as any).extensionOnlyV1View(); return {session: new ACPSessionConnection(view, sessionId), updates}; } -describe('subagent and async task updates on ACP v2', () => { +describe.each([1, 2] as const)('AIR subagent and async task updates on ACP v%s', version => { afterEach(() => { expectConformingV2SessionUpdates(); }); - it('renders the subagent lifecycle as `_subagent_update` when native subagents are enabled', async () => { - const {session, updates} = v2Session("root"); + it('preserves the AIR spawn and terminal lifecycle when native subagents are enabled', async () => { + const {session, updates} = sessionForVersion(version, "root"); const router = new CodexSubagentEventRouter("root", true, session, () => {}, () => ({currentModel: null, availableModels: []})); await router.handle({ @@ -68,22 +68,22 @@ describe('subagent and async task updates on ACP v2', () => { expect(updates).toEqual([ { - sessionUpdate: "_subagent_update", + sessionUpdate: version === 1 ? "subagent_spawned" : "_subagent_update", subagentSessionId: "child-1", name: "Researcher", task: "Delegated task for Researcher", capabilities: {}, }, { - sessionUpdate: "_subagent_update", + sessionUpdate: version === 1 ? "subagent_state_update" : "_subagent_update", subagentSessionId: "child-1", state: "cancelled", }, ]); }); - it('renders a follow-up turn of a finished subagent as a new `_subagent_update` generation', async () => { - const {session, updates} = v2Session("root"); + it('announces a new AIR generation for a follow-up turn of a finished subagent', async () => { + const {session, updates} = sessionForVersion(version, "root"); const router = new CodexSubagentEventRouter("root", true, session, () => {}, () => ({currentModel: null, availableModels: []})); const childTurn = (id: string, status: "completed" | "inProgress") => router.handle({ method: status === "completed" ? "turn/completed" : "turn/started", @@ -126,26 +126,26 @@ describe('subagent and async task updates on ACP v2', () => { expect(updates).toEqual([ { - sessionUpdate: "_subagent_update", + sessionUpdate: version === 1 ? "subagent_spawned" : "_subagent_update", subagentSessionId: "child-1", name: "Researcher", task: "Delegated task for Researcher", capabilities: {}, }, - {sessionUpdate: "_subagent_update", subagentSessionId: "child-1", state: "completed"}, + {sessionUpdate: version === 1 ? "subagent_state_update" : "_subagent_update", subagentSessionId: "child-1", state: "completed"}, { - sessionUpdate: "_subagent_update", + sessionUpdate: version === 1 ? "subagent_spawned" : "_subagent_update", subagentSessionId: "child-1:generation:2", name: "Researcher", task: "Delegated task for Researcher", capabilities: {}, }, - {sessionUpdate: "_subagent_update", subagentSessionId: "child-1:generation:2", state: "completed"}, + {sessionUpdate: version === 1 ? "subagent_state_update" : "_subagent_update", subagentSessionId: "child-1:generation:2", state: "completed"}, ]); }); - it('never emits `_subagent_update` (or `subagent_spawned`) when native subagents are disabled', async () => { - const {session, updates} = v2Session("root"); + it('emits no native lifecycle when native subagents are disabled', async () => { + const {session, updates} = sessionForVersion(version, "root"); const router = new CodexSubagentEventRouter("root", false, session, () => {}, () => ({currentModel: null, availableModels: []})); const handled = await router.handle({ @@ -173,7 +173,7 @@ describe('subagent and async task updates on ACP v2', () => { }); function createAsyncTasks(enabled: boolean) { - const {session, updates} = v2Session("root"); + const {session, updates} = sessionForVersion(version, "root"); const list = vi.fn<() => Promise>(); const appServer = {threadBackgroundTerminalsList: list, backgroundTerminalPages: CodexAppServerClient.prototype.backgroundTerminalPages} as unknown as CodexAppServerClient; @@ -181,7 +181,7 @@ describe('subagent and async task updates on ACP v2', () => { return {tasks, list, updates}; } - it('renders async task updates as `_async_task_*` when async tasks are enabled', async () => { + it('preserves the AIR async task lifecycle when async tasks are enabled', async () => { const {tasks, list, updates} = createAsyncTasks(true); list.mockResolvedValue({data: [{itemId: "command-1", processId: "42", command: "python -m http.server"}], nextCursor: null}); @@ -217,7 +217,7 @@ describe('subagent and async task updates on ACP v2', () => { _meta: {jetbrains: {air: {asyncTasks: {backgrounded: true}}}}, }, { - sessionUpdate: "_async_task_spawned", + sessionUpdate: version === 1 ? "async_task_spawned" : "_async_task_spawned", asyncTaskId: "command-1", name: "python -m http.server", taskType: "shell", @@ -228,7 +228,7 @@ describe('subagent and async task updates on ACP v2', () => { ]); }); - it('never emits `_async_task_*` (or `async_task_*`) when async tasks are disabled', async () => { + it('emits no async task lifecycle when async tasks are disabled', async () => { const {tasks, list, updates} = createAsyncTasks(false); list.mockResolvedValue({data: [], nextCursor: null}); 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 b9f38910e..675cca627 100644 --- a/src/__tests__/CodexACPAgent/tool-calls-and-messages-v2.test.ts +++ b/src/__tests__/CodexACPAgent/tool-calls-and-messages-v2.test.ts @@ -190,7 +190,7 @@ describe('tool calls and messages over ACP v2', () => { it('flags unknown unprefixed session update tags and v1-only shapes', () => { const violations = [ {sessionUpdate: "_vendor/custom", value: 1}, - {sessionUpdate: "subagent_spawned", subagentSessionId: "child"}, + {sessionUpdate: "vendor_unknown", value: 1}, {sessionUpdate: "tool_call", toolCallId: "call-1", title: "Run tests"}, {sessionUpdate: "agent_message_chunk", content: {type: "text", text: "no id"}}, {sessionUpdate: "agent_message_chunk", messageId: "", content: {type: "text", text: "empty id"}}, @@ -204,7 +204,7 @@ describe('tool calls and messages over ACP v2', () => { expect(violations).toEqual([ ["_vendor/custom", null], - ["subagent_spawned", "unknown unprefixed sessionUpdate 'subagent_spawned'"], + ["vendor_unknown", "unknown unprefixed sessionUpdate 'vendor_unknown'"], ["tool_call", "unknown unprefixed sessionUpdate 'tool_call'"], ["agent_message_chunk", expect.stringContaining("'agent_message_chunk' does not match its v2 schema")], ["agent_message_chunk", "'agent_message_chunk' has an empty messageId"], diff --git a/src/__tests__/CodexACPAgent/v2-resume-replay-subagents.test.ts b/src/__tests__/CodexACPAgent/v2-resume-replay-subagents.test.ts index 9d6b42771..30222df63 100644 --- a/src/__tests__/CodexACPAgent/v2-resume-replay-subagents.test.ts +++ b/src/__tests__/CodexACPAgent/v2-resume-replay-subagents.test.ts @@ -215,7 +215,7 @@ describe('ACP v2 resume replay: native subagent history (nativeSubagentSessions ]); // Async tasks aren't gated on in this capability set, so nothing async-task-shaped renders. - expect(updates.some(({update}) => update.sessionUpdate.startsWith("_async_task_"))).toBe(false); + expect(updates.some(({update}) => update.sessionUpdate.startsWith("async_task_"))).toBe(false); }); it('also emits the child subagent history (agent reply) alongside the lifecycle updates', async () => { diff --git a/src/__tests__/CodexACPAgent/v2-session-update-guard.ts b/src/__tests__/CodexACPAgent/v2-session-update-guard.ts index 7e2822271..5e63133be 100644 --- a/src/__tests__/CodexACPAgent/v2-session-update-guard.ts +++ b/src/__tests__/CodexACPAgent/v2-session-update-guard.ts @@ -25,8 +25,9 @@ const messageTags = new Set([ * Why a `session/update` payload is not a valid ACP v2 update, or `null` if it is. * * Neither the SDK nor the TS types stop a v1-only tag (e.g. `tool_call`) or a custom unprefixed - * tag from reaching a v2 client, so tests check the frames themselves: custom tags must start with - * `_`, and standard ones must match their v2 schema. Terminal bytes must be base64, `cwd` and diff + * tag from reaching a v2 client, so tests check the frames themselves. + * Custom tags must start with `_`. Standard updates must match their v2 schema. + * Terminal bytes must be base64, `cwd` and diff * paths absolute, and v1's private terminal and diff `_meta` keys must not reach the client. */ export function v2SessionUpdateViolation(update: acpV2.SessionUpdate): string | null {