diff --git a/src/AgentFileChangeReport.ts b/src/AgentFileChangeReport.ts new file mode 100644 index 00000000..aa036778 --- /dev/null +++ b/src/AgentFileChangeReport.ts @@ -0,0 +1,422 @@ +import fs from "node:fs"; +import path from "node:path"; +import type {JsonValue} from "./app-server/serde_json/JsonValue"; +import type {Turn} from "./app-server/v2"; +import { + AIR_AGENT_FILE_CHANGE_REPORT_REQUEST_KEY, + AIR_META_KEY, + JETBRAINS_META_KEY, +} from "./AirExtension"; + +export const AGENT_FILE_CHANGE_REPORT_VERSION = 1; +export const AGENT_FILE_CHANGE_REPORT_TIMEOUT_MS = 30_000; +export const AGENT_FILE_CHANGE_REPORT_MAX_PATHS = 1_024; +export const AGENT_FILE_CHANGE_REPORT_MAX_PATH_LENGTH = 4_096; +export const AGENT_FILE_CHANGE_REPORT_MAX_TOTAL_BYTES = 256 * 1_024; +export const AGENT_FILE_CHANGE_REPORT_MAX_UNCERTAINTY_LENGTH = 2_000; + +export interface AgentFileChangeReportRequest { + version: typeof AGENT_FILE_CHANGE_REPORT_VERSION; + requestId: string; +} + +export interface AgentFileChangeWorkspace { + cwd: string; + additionalDirectories: string[]; +} + +interface ModelFileChangeReport { + paths: string[]; + complete: boolean; + uncertainty?: string; +} + +export interface ReportedAgentFileChangeReport { + version: typeof AGENT_FILE_CHANGE_REPORT_VERSION; + requestId: string; + status: "reported"; + paths: string[]; + declaredComplete: boolean; + truncated: boolean; + uncertainty?: string; +} + +export type AgentFileChangeReportUnavailableReason = + | "cancelled" + | "timeout" + | "invalidOutput" + | "notReported" + | "providerError"; + +export interface UnavailableAgentFileChangeReport { + version: typeof AGENT_FILE_CHANGE_REPORT_VERSION; + requestId: string; + status: "unavailable"; + reason: AgentFileChangeReportUnavailableReason; +} + +export type AgentFileChangeReport = ReportedAgentFileChangeReport | UnavailableAgentFileChangeReport; + +export const AGENT_FILE_CHANGE_REPORT_OUTPUT_SCHEMA: JsonValue = { + type: "object", + additionalProperties: false, + required: ["paths", "complete"], + properties: { + paths: { + type: "array", + items: {type: "string"}, + }, + complete: {type: "boolean"}, + uncertainty: { + type: "string", + maxLength: AGENT_FILE_CHANGE_REPORT_MAX_UNCERTAINTY_LENGTH, + }, + }, +}; + +export const AGENT_FILE_CHANGE_REPORT_DEVELOPER_INSTRUCTIONS = `You are running an internal, read-only file-change audit for the immediately preceding turn. +Do not modify files, run commands that can modify files, or ask the user questions. +Report files that the preceding turn causally created, modified, deleted, or moved, including changes made by shell commands, version-control commands, generators, and child processes. +Do not report files that were only read or inspected. +You may use read-only inspection when needed. If the list may be incomplete, set complete to false and briefly explain why in uncertainty.`; + +export function createAgentFileChangeReportPrompt(workspace: AgentFileChangeWorkspace): string { + return `List the paths changed by the immediately preceding turn. +Return only the structured result required by the output schema. +Relative paths are resolved against the working directory. +Working directory: ${JSON.stringify(workspace.cwd)} +Additional allowed directories: ${JSON.stringify(workspace.additionalDirectories)}`; +} + +export function parseAgentFileChangeReportRequest( + meta: Record | null | undefined, +): AgentFileChangeReportRequest | null { + const jetbrains = asRecord(meta?.[JETBRAINS_META_KEY]); + const air = asRecord(jetbrains?.[AIR_META_KEY]); + const request = asRecord(air?.[AIR_AGENT_FILE_CHANGE_REPORT_REQUEST_KEY]); + if (request === null + || !hasOnlyKeys(request, ["version", "requestId"]) + || request["version"] !== AGENT_FILE_CHANGE_REPORT_VERSION) { + return null; + } + const requestId = request["requestId"]; + if (typeof requestId !== "string" || !/^[A-Za-z0-9._:-]{1,128}$/.test(requestId)) { + return null; + } + return {version: AGENT_FILE_CHANGE_REPORT_VERSION, requestId}; +} + +export function createReportedAgentFileChangeReport( + requestId: string, + turn: Turn, + workspace: AgentFileChangeWorkspace, +): ReportedAgentFileChangeReport { + const modelReport = parseModelFileChangeReport(turn); + const normalized = normalizeModelFileChangeReport(modelReport, workspace); + return fitReportedAgentFileChangeReport({ + version: AGENT_FILE_CHANGE_REPORT_VERSION, + requestId, + status: "reported", + ...normalized, + }); +} + +export function createUnavailableAgentFileChangeReport( + requestId: string, + reason: AgentFileChangeReportUnavailableReason, +): UnavailableAgentFileChangeReport { + return { + version: AGENT_FILE_CHANGE_REPORT_VERSION, + requestId, + status: "unavailable", + reason, + }; +} + +export class AgentFileChangeReportError extends Error { + readonly reason: AgentFileChangeReportUnavailableReason; + + constructor(reason: AgentFileChangeReportUnavailableReason, message: string) { + super(message); + this.name = "AgentFileChangeReportError"; + this.reason = reason; + } +} + +function parseModelFileChangeReport(turn: Turn): ModelFileChangeReport { + switch (turn.status) { + case "interrupted": + throw new AgentFileChangeReportError("cancelled", "The audit turn was interrupted"); + case "failed": + throw new AgentFileChangeReportError("providerError", "The audit turn failed"); + case "inProgress": + throw new AgentFileChangeReportError("notReported", "The audit turn did not complete"); + case "completed": + break; + } + + let text: string | null = null; + for (let index = turn.items.length - 1; index >= 0; index -= 1) { + const item = turn.items[index]; + if (item?.type === "agentMessage") { + text = item.text; + break; + } + } + if (text === null) { + throw new AgentFileChangeReportError("notReported", "The audit turn returned no agent message"); + } + + let value: unknown; + try { + value = JSON.parse(text); + } catch { + throw new AgentFileChangeReportError("invalidOutput", "The audit turn returned invalid JSON"); + } + const report = asRecord(value); + if (report === null || !hasOnlyKeys(report, ["paths", "complete", "uncertainty"])) { + throw new AgentFileChangeReportError("invalidOutput", "The audit turn returned an invalid object"); + } + const paths = report["paths"]; + const complete = report["complete"]; + const uncertainty = report["uncertainty"]; + if (!Array.isArray(paths) + || !paths.every((item): item is string => typeof item === "string") + || typeof complete !== "boolean" + || (uncertainty !== undefined && typeof uncertainty !== "string")) { + throw new AgentFileChangeReportError("invalidOutput", "The audit turn returned invalid fields"); + } + const normalizedUncertainty = uncertainty?.trim(); + if (normalizedUncertainty !== undefined + && normalizedUncertainty.length > AGENT_FILE_CHANGE_REPORT_MAX_UNCERTAINTY_LENGTH) { + throw new AgentFileChangeReportError("invalidOutput", "The audit turn returned oversized uncertainty"); + } + return { + paths, + complete, + ...(normalizedUncertainty ? {uncertainty: normalizedUncertainty} : {}), + }; +} + +/** + * AIR applies its 256 KiB limit to the serialized report object, not only to + * the raw path strings. Account for JSON quotes, escaping, separators, and + * fixed fields before publishing so a boundary-sized report remains decodable. + */ +function fitReportedAgentFileChangeReport( + report: ReportedAgentFileChangeReport, +): ReportedAgentFileChangeReport { + const paths = [...report.paths]; + let truncated = report.truncated; + let fitted = report; + while (Buffer.byteLength(JSON.stringify(fitted), "utf8") > AGENT_FILE_CHANGE_REPORT_MAX_TOTAL_BYTES) { + if (paths.length === 0) { + throw new AgentFileChangeReportError("invalidOutput", "The audit report exceeds the wire limit"); + } + paths.pop(); + truncated = true; + fitted = { + ...report, + paths: [...paths], + declaredComplete: false, + truncated, + }; + } + return fitted; +} + +function normalizeModelFileChangeReport( + report: ModelFileChangeReport, + workspace: AgentFileChangeWorkspace, +): Omit { + const cwd = normalizeWorkspaceRoot(workspace.cwd); + if (cwd === null) { + throw new AgentFileChangeReportError("providerError", "The session working directory is not absolute"); + } + const roots = [cwd, ...workspace.additionalDirectories.flatMap(directory => { + const root = normalizeWorkspaceRoot(directory); + return root === null || root.flavor !== cwd.flavor ? [] : [root]; + })]; + const paths: string[] = []; + const seen = new Set(); + let totalBytes = 0; + let truncated = false; + + for (const reportedPath of report.paths) { + if (reportedPath.length > AGENT_FILE_CHANGE_REPORT_MAX_PATH_LENGTH) { + truncated = true; + continue; + } + const normalized = normalizeReportedPath(reportedPath, cwd, roots); + if (normalized === null || normalized.value.length > AGENT_FILE_CHANGE_REPORT_MAX_PATH_LENGTH) { + truncated = true; + continue; + } + const key = normalized.flavor === "windows" ? normalized.value.toLowerCase() : normalized.value; + if (seen.has(key)) { + continue; + } + const bytes = Buffer.byteLength(normalized.value, "utf8"); + if (paths.length >= AGENT_FILE_CHANGE_REPORT_MAX_PATHS + || totalBytes + bytes > AGENT_FILE_CHANGE_REPORT_MAX_TOTAL_BYTES) { + truncated = true; + continue; + } + seen.add(key); + paths.push(normalized.value); + totalBytes += bytes; + } + + return { + paths, + declaredComplete: report.complete && !truncated, + truncated, + ...(report.uncertainty ? {uncertainty: report.uncertainty} : {}), + }; +} + +type PathFlavor = "posix" | "windows"; + +interface NormalizedPath { + value: string; + flavor: PathFlavor; +} + +function normalizeWorkspaceRoot(value: string): NormalizedPath | null { + const trimmed = value.trim(); + if (!isValidPathText(trimmed)) { + return null; + } + if (isWindowsAbsolutePath(trimmed)) { + return canonicalizeWorkspaceRoot({ + value: path.win32.normalize(trimmed.replace(/\//g, "\\")), + flavor: "windows", + }); + } + if (path.posix.isAbsolute(trimmed)) { + return canonicalizeWorkspaceRoot({ + value: path.posix.normalize(trimmed.replace(/\\/g, "/")), + flavor: "posix", + }); + } + return null; +} + +function normalizeReportedPath( + value: string, + cwd: NormalizedPath, + roots: NormalizedPath[], +): NormalizedPath | null { + const trimmed = value.trim(); + if (!isValidPathText(trimmed) + || /^[A-Za-z]:[^\\/]/.test(trimmed) + || /^\\\\[?.]\\/.test(trimmed) + || (/^(?:\\\\|\/\/)/.test(trimmed) && !isWindowsAbsolutePath(trimmed)) + || (cwd.flavor === "windows" && /^\\(?!\\)/.test(trimmed)) + || /^[A-Za-z][A-Za-z0-9+.-]*:\/\//.test(trimmed)) { + return null; + } + + let candidate: NormalizedPath; + if (isWindowsAbsolutePath(trimmed)) { + candidate = {value: path.win32.normalize(trimmed.replace(/\//g, "\\")), flavor: "windows"}; + } else if (path.posix.isAbsolute(trimmed)) { + candidate = {value: path.posix.normalize(trimmed.replace(/\\/g, "/")), flavor: "posix"}; + } else if (cwd.flavor === "windows") { + candidate = { + value: path.win32.resolve(cwd.value, trimmed.replace(/\//g, "\\")), + flavor: "windows", + }; + } else { + candidate = { + value: path.posix.resolve(cwd.value, trimmed.replace(/\\/g, "/")), + flavor: "posix", + }; + } + + candidate = canonicalizeReportedPath(candidate); + + return roots.some(root => pathIsStrictlyInside(root, candidate)) ? candidate : null; +} + +/** Resolve native filesystem aliases such as macOS' /tmp -> /private/tmp. */ +function canonicalizeWorkspaceRoot(root: NormalizedPath): NormalizedPath { + if (!isNativePathFlavor(root.flavor)) { + return root; + } + return {...root, value: canonicalizeFromExistingAncestor(root.value)}; +} + +/** + * Canonicalize the parent but not the leaf itself. A changed path may be + * deleted, or it may be a symlink whose node (rather than target) changed. + */ +function canonicalizeReportedPath(candidate: NormalizedPath): NormalizedPath { + if (!isNativePathFlavor(candidate.flavor)) { + return candidate; + } + const parent = canonicalizeFromExistingAncestor(path.dirname(candidate.value)); + return {...candidate, value: path.resolve(parent, path.basename(candidate.value))}; +} + +/** Resolve the nearest existing ancestor and retain any missing suffix. */ +function canonicalizeFromExistingAncestor(value: string): string { + const original = path.resolve(value); + let current = original; + const missingSegments: string[] = []; + + while (true) { + try { + const canonical = fs.realpathSync.native(current); + return path.resolve(canonical, ...missingSegments.reverse()); + } catch (error) { + if (!isMissingPathError(error)) return original; + } + + const parent = path.dirname(current); + if (parent === current) return original; + missingSegments.push(path.basename(current)); + current = parent; + } +} + +function isMissingPathError(error: unknown): boolean { + if (typeof error !== "object" || error === null || !("code" in error)) return false; + return error.code === "ENOENT" || error.code === "ENOTDIR"; +} + +function isNativePathFlavor(flavor: PathFlavor): boolean { + return process.platform === "win32" ? flavor === "windows" : flavor === "posix"; +} + +function pathIsStrictlyInside(root: NormalizedPath, candidate: NormalizedPath): boolean { + if (root.flavor !== candidate.flavor) { + return false; + } + const pathImplementation = root.flavor === "windows" ? path.win32 : path.posix; + const relative = pathImplementation.relative(root.value, candidate.value); + return relative.length > 0 + && !pathImplementation.isAbsolute(relative) + && relative !== ".." + && !relative.startsWith(`..${pathImplementation.sep}`); +} + +function isWindowsAbsolutePath(value: string): boolean { + const portable = value.replace(/\\/g, "/"); + return /^[A-Za-z]:\//.test(portable) || /^\/\/[^/]+\/[^/]+(?:\/|$)/.test(portable); +} + +function isValidPathText(value: string): boolean { + return value.length > 0 && !/[\u0000-\u001F\u007F-\u009F]/.test(value); +} + +function asRecord(value: unknown): Record | null { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : null; +} + +function hasOnlyKeys(value: Record, allowed: string[]): boolean { + const allowedKeys = new Set(allowed); + return Object.keys(value).every(key => allowedKeys.has(key)); +} diff --git a/src/AirExtension.ts b/src/AirExtension.ts index c16df898..cf97b512 100644 --- a/src/AirExtension.ts +++ b/src/AirExtension.ts @@ -11,4 +11,6 @@ export const AIR_META_KEY = "air"; export const AIR_EXTENSION_VERSION_KEY = "version"; export const AIR_EXTENSION_CAPABILITIES_KEY = "capabilities"; export const AIR_SESSION_FAILURE_KEY = "sessionFailure"; +export const AIR_AGENT_FILE_CHANGE_REPORT_KEY = "agentFileChangeReport"; +export const AIR_AGENT_FILE_CHANGE_REPORT_REQUEST_KEY = "agentFileChangeReportRequest"; export const AIR_EXTENSION_VERSION = 1; diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 5cee51ff..5fd4dbef 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -52,6 +52,17 @@ import type {AuthenticationStatusResponse} from "./AcpExtensions"; import {createCodexCollaborationMode} from "./CollaborationModeConfig"; import type {ModeKind} from "./app-server/ModeKind"; import {arePathBasenamesEqual, arePathsEqual, isAbsolutePathLike} from "./PathUtils"; +import { + AGENT_FILE_CHANGE_REPORT_DEVELOPER_INSTRUCTIONS, + AGENT_FILE_CHANGE_REPORT_OUTPUT_SCHEMA, + AGENT_FILE_CHANGE_REPORT_TIMEOUT_MS, + type AgentFileChangeReport, + AgentFileChangeReportError, + type AgentFileChangeWorkspace, + createAgentFileChangeReportPrompt, + createReportedAgentFileChangeReport, + createUnavailableAgentFileChangeReport, +} from "./AgentFileChangeReport"; /** * Well-known provider id for the client-configurable custom LLM gateway. @@ -799,6 +810,131 @@ export class CodexAcpClient { }, onTurnStarted); } + async runAgentFileChangeReport(params: { + sessionId: string; + turnId: string; + requestId: string; + workspace: AgentFileChangeWorkspace; + signal?: AbortSignal; + }): Promise { + if (params.signal?.aborted) { + return createUnavailableAgentFileChangeReport(params.requestId, "cancelled"); + } + + const budget = new AgentFileChangeReportBudget(params.signal); + let forkThreadId: string | null = null; + let auditTurnId: string | null = null; + let auditTurnCompleted = false; + let lateStopReason: "cancelled" | "timeout" | null = null; + try { + const forkPromise = this.codexClient.threadFork({ + threadId: params.sessionId, + lastTurnId: params.turnId, + cwd: params.workspace.cwd, + approvalPolicy: "never", + sandbox: "read-only", + developerInstructions: AGENT_FILE_CHANGE_REPORT_DEVELOPER_INSTRUCTIONS, + ephemeral: true, + }); + void forkPromise.then(fork => { + if (lateStopReason !== null && forkThreadId === null) { + void this.unsubscribeAgentFileChangeReportThread(fork.thread.id, budget); + } + }, () => {}); + const fork = await budget.wait(forkPromise); + forkThreadId = fork.thread.id; + + const turnPromise = this.codexClient.runTurn({ + threadId: forkThreadId, + input: [{ + type: "text", + text: createAgentFileChangeReportPrompt(params.workspace), + text_elements: [], + }], + cwd: params.workspace.cwd, + approvalPolicy: "never", + sandboxPolicy: {type: "readOnly", networkAccess: false}, + summary: "none", + outputSchema: AGENT_FILE_CHANGE_REPORT_OUTPUT_SCHEMA, + }, (turnId) => { + auditTurnId = turnId; + if (lateStopReason !== null && forkThreadId !== null) { + void this.interruptAgentFileChangeReport(forkThreadId, turnId, lateStopReason, budget); + } + }); + const outcome = await budget.wait(turnPromise); + auditTurnCompleted = true; + const thread = await budget.wait(this.codexClient.threadRead({ + threadId: forkThreadId, + includeTurns: true, + })); + const completedTurn = thread.thread.turns.find( + turn => turn.id === outcome.turn.id, + ); + if (completedTurn === undefined) { + throw new AgentFileChangeReportError( + "notReported", + "The completed audit turn was not present in thread history", + ); + } + return createReportedAgentFileChangeReport( + params.requestId, + completedTurn, + params.workspace, + ); + } catch (error) { + if (error instanceof AgentFileChangeReportBudgetError) { + lateStopReason = error.reason; + if (!auditTurnCompleted && forkThreadId !== null && auditTurnId !== null) { + await this.interruptAgentFileChangeReport( + forkThreadId, + auditTurnId, + error.reason, + budget, + ); + } + return createUnavailableAgentFileChangeReport(params.requestId, error.reason); + } + if (error instanceof AgentFileChangeReportError) { + logger.log("Agent file-change report unavailable", {reason: error.reason}); + return createUnavailableAgentFileChangeReport(params.requestId, error.reason); + } + logger.error("Agent file-change report failed", error); + return createUnavailableAgentFileChangeReport(params.requestId, "providerError"); + } finally { + if (forkThreadId !== null) { + await this.unsubscribeAgentFileChangeReportThread(forkThreadId, budget); + } + } + } + + private async interruptAgentFileChangeReport( + threadId: string, + turnId: string, + reason: "cancelled" | "timeout", + budget: AgentFileChangeReportBudget, + ): Promise { + this.codexClient.markTurnStale(threadId, turnId); + try { + await budget.wait(this.codexClient.turnInterrupt({threadId, turnId})); + } catch (error) { + logger.error(`Failed to interrupt ${reason} agent file-change report`, error); + } finally { + this.codexClient.resolveTurnInterrupted(threadId, turnId); + } + } + + private async unsubscribeAgentFileChangeReportThread( + threadId: string, + budget: AgentFileChangeReportBudget, + ): Promise { + try { + await budget.wait(this.codexClient.threadUnsubscribe({threadId})); + } catch (error) { + logger.error("Failed to unsubscribe the agent file-change report thread", error); + } + } + async setCollaborationMode(sessionId: string, mode: ModeKind, currentModelId: string): Promise { await this.codexClient.threadSettingsUpdate({ threadId: sessionId, @@ -977,6 +1113,62 @@ export class CodexAcpClient { } +class AgentFileChangeReportBudgetError extends Error { + constructor(readonly reason: "cancelled" | "timeout") { + super(`Agent file-change report ${reason}`); + this.name = "AgentFileChangeReportBudgetError"; + } +} + +/** One wall-clock budget shared by fork, turn, read, interruption, and cleanup. */ +class AgentFileChangeReportBudget { + private readonly deadline = Date.now() + AGENT_FILE_CHANGE_REPORT_TIMEOUT_MS; + + constructor(private readonly signal?: AbortSignal) {} + + async wait(operation: Promise): Promise { + // A stage can outlive the race at the transport layer. Attach a handler + // before the immediate budget checks so a late rejection is never + // unhandled even when no time remains to await it. + void operation.catch(() => {}); + const immediateReason = this.stopReason(); + if (immediateReason !== null) { + throw new AgentFileChangeReportBudgetError(immediateReason); + } + + return await new Promise((resolve, reject) => { + let settled = false; + const finish = (action: () => void): void => { + if (settled) return; + settled = true; + clearTimeout(timeout); + this.signal?.removeEventListener("abort", onAbort); + action(); + }; + const onAbort = (): void => finish(() => reject(new AgentFileChangeReportBudgetError("cancelled"))); + const timeout = setTimeout( + () => finish(() => reject(new AgentFileChangeReportBudgetError("timeout"))), + Math.max(1, this.deadline - Date.now()), + ); + timeout.unref(); + this.signal?.addEventListener("abort", onAbort, {once: true}); + if (this.signal?.aborted) { + onAbort(); + } + void operation.then( + value => finish(() => resolve(value)), + error => finish(() => reject(error)), + ); + }); + } + + private stopReason(): "cancelled" | "timeout" | null { + if (this.signal?.aborted) return "cancelled"; + if (Date.now() >= this.deadline) return "timeout"; + return null; + } +} + export type JsonObject = { [key in string]?: JsonValue } export type SessionMetadata = { diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index db3c6e34..47e9373a 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -93,6 +93,7 @@ import { import {sameThreadGoalSnapshot, type ThreadGoalSnapshot, toThreadGoalSnapshot,} from "./ThreadGoalSnapshot"; import {randomUUID} from "node:crypto"; import { + AIR_AGENT_FILE_CHANGE_REPORT_KEY, AIR_EXTENSION_CAPABILITIES_KEY, AIR_EXTENSION_VERSION, AIR_EXTENSION_VERSION_KEY, @@ -100,6 +101,13 @@ import { AIR_SESSION_FAILURE_KEY, JETBRAINS_META_KEY, } from "./AirExtension"; +import { + type AgentFileChangeReport, + type AgentFileChangeReportRequest, + type AgentFileChangeReportUnavailableReason, + createUnavailableAgentFileChangeReport, + parseAgentFileChangeReportRequest, +} from "./AgentFileChangeReport"; const IMPLEMENT_PLAN_OPTION_ID = "implement_plan"; const REVISE_PLAN_OPTION_ID = "revise_plan"; @@ -156,7 +164,10 @@ export interface SessionFailure { const CODEX_PROCESS_EXITED_ERROR_CODE = 1001; -function clientSupportsTypedSessionFailures(capabilities: acp.ClientCapabilities | null): boolean { +function clientSupportsAirCapability( + capabilities: acp.ClientCapabilities | null, + capability: string, +): boolean { const jetbrains = capabilities?._meta?.[JETBRAINS_META_KEY] as Record | undefined; const air = jetbrains?.[AIR_META_KEY] as Record | undefined; const version = air?.[AIR_EXTENSION_VERSION_KEY]; @@ -165,7 +176,15 @@ function clientSupportsTypedSessionFailures(capabilities: acp.ClientCapabilities && Number.isInteger(version) && version >= AIR_EXTENSION_VERSION && Array.isArray(supported) - && supported.includes(AIR_SESSION_FAILURE_KEY); + && supported.includes(capability); +} + +function clientSupportsTypedSessionFailures(capabilities: acp.ClientCapabilities | null): boolean { + return clientSupportsAirCapability(capabilities, AIR_SESSION_FAILURE_KEY); +} + +function clientSupportsAgentFileChangeReports(capabilities: acp.ClientCapabilities | null): boolean { + return clientSupportsAirCapability(capabilities, AIR_AGENT_FILE_CHANGE_REPORT_KEY); } interface ActiveAuthState { @@ -309,7 +328,10 @@ export class CodexAcpServer { [JETBRAINS_META_KEY]: { [AIR_META_KEY]: { [AIR_EXTENSION_VERSION_KEY]: AIR_EXTENSION_VERSION, - [AIR_EXTENSION_CAPABILITIES_KEY]: [AIR_SESSION_FAILURE_KEY], + [AIR_EXTENSION_CAPABILITIES_KEY]: [ + AIR_SESSION_FAILURE_KEY, + AIR_AGENT_FILE_CHANGE_REPORT_KEY, + ], }, }, }, @@ -1566,6 +1588,51 @@ export class CodexAcpServer { }); } + private async publishAgentFileChangeReport( + sessionState: SessionState, + turnId: string | null, + request: AgentFileChangeReportRequest, + unavailableReason: AgentFileChangeReportUnavailableReason, + signal: AbortSignal, + ): Promise { + let report: AgentFileChangeReport; + try { + report = turnId === null + ? createUnavailableAgentFileChangeReport(request.requestId, unavailableReason) + : await this.codexAcpClient.runAgentFileChangeReport({ + sessionId: sessionState.sessionId, + turnId, + // The client owns request-id correlation and duplicate suppression. The wrapper + // stays stateless so a retried ACP prompt still receives a terminal report. + requestId: request.requestId, + workspace: { + cwd: sessionState.cwd, + additionalDirectories: sessionState.additionalDirectories, + }, + signal, + }); + } catch (error) { + logger.error("Agent file-change report failed unexpectedly", error); + report = createUnavailableAgentFileChangeReport(request.requestId, "providerError"); + } + try { + const session = new ACPSessionConnection(this.connection, sessionState.sessionId); + await session.update({ + sessionUpdate: "session_info_update", + _meta: { + [JETBRAINS_META_KEY]: { + [AIR_META_KEY]: { + [AIR_EXTENSION_VERSION_KEY]: AIR_EXTENSION_VERSION, + [AIR_AGENT_FILE_CHANGE_REPORT_KEY]: report, + }, + }, + }, + }); + } catch (error) { + logger.error("Failed to publish agent file-change report", error); + } + } + private createPromptFallbackTitle(prompt: acp.ContentBlock[]): string | null { return this.normalizeSessionTitle(prompt .filter((block): block is Extract => block.type === "text") @@ -2030,6 +2097,11 @@ export class CodexAcpServer { prompt: params.prompt, }); const sessionState = this.getSessionState(params.sessionId); + const agentFileChangeReportRequest = clientSupportsAgentFileChangeReports(this.clientCapabilities) + ? parseAgentFileChangeReportRequest(params._meta) + : null; + let agentFileChangeReportTurnId: string | null = null; + let agentFileChangeReportUnavailableReason: AgentFileChangeReportUnavailableReason = "providerError"; let recoverableSessionFailure = sessionState.sessionFailure; sessionState.currentTurnId = null; sessionState.lastTokenUsage = null; @@ -2055,6 +2127,11 @@ export class CodexAcpServer { await handler.clearSessionFailure(); } }; + const cancelledPromptResponse = (): acp.PromptResponse => { + agentFileChangeReportTurnId = null; + agentFileChangeReportUnavailableReason = "cancelled"; + return this.cancelledPromptResponse(sessionState); + }; try { const promptEventHandler = new CodexEventHandler( @@ -2092,7 +2169,7 @@ export class CodexAcpServer { elicitationHandler); if (activePrompt.signal.aborted) { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } const commandPromise = this.availableCommands.tryHandleCommand(params.prompt, sessionState, { @@ -2134,7 +2211,7 @@ export class CodexAcpServer { this.cancelBeforeTurnStarted(activePrompt), ]); if (commandResult === null) { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } if (commandResult.handled) { promptNotificationsActive = false; @@ -2146,7 +2223,7 @@ export class CodexAcpServer { await eventHandler.handleFailedTurn(commandResult.turnCompleted.turn); } if (commandResult.turnCompleted?.turn.status === "interrupted") { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } const error = eventHandler.getFailure(); if (error) { @@ -2161,6 +2238,11 @@ export class CodexAcpServer { if (terminalFailure) { return terminalFailure; } + if (commandResult.turnCompleted?.turn.status === "completed") { + agentFileChangeReportTurnId = commandResult.turnCompleted.turn.id; + } else if (commandResult.turnCompleted === undefined) { + agentFileChangeReportUnavailableReason = "notReported"; + } await clearRecoveredSessionFailure(eventHandler); return { stopReason: "end_turn", @@ -2174,7 +2256,7 @@ export class CodexAcpServer { : {...params, prompt: commandResult.prompt}; if (this.sessionIsClosing(params.sessionId)) { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } const modelId = ModelId.fromString(sessionState.currentModelId); @@ -2232,7 +2314,7 @@ export class CodexAcpServer { ]); if (turnCompleted === null) { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } await this.codexAcpClient.waitForSessionNotifications(params.sessionId); @@ -2242,7 +2324,7 @@ export class CodexAcpServer { if (turnCompleted.turn.status === "interrupted") { await eventHandler.flushPendingPlanUpdates(); - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } const error = eventHandler.getFailure(); @@ -2272,7 +2354,7 @@ export class CodexAcpServer { activePrompt.signal, ); if (this.promptShouldStop(params.sessionId, activePrompt)) { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } if (approved && !this.promptShouldStop(params.sessionId, activePrompt)) { await this.applyCollaborationModeChange(sessionState, DEFAULT_COLLABORATION_MODE); @@ -2325,7 +2407,7 @@ export class CodexAcpServer { ]); if (turnCompleted === null) { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } await this.codexAcpClient.waitForSessionNotifications(params.sessionId); @@ -2334,7 +2416,7 @@ export class CodexAcpServer { promptNotificationsActive = false; if (turnCompleted.turn.status === "interrupted") { await eventHandler.flushPendingPlanUpdates(); - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } const implementationError = eventHandler.getFailure(); @@ -2351,6 +2433,9 @@ export class CodexAcpServer { } } } + if (turnCompleted.turn.status === "completed") { + agentFileChangeReportTurnId = turnCompleted.turn.id; + } await clearRecoveredSessionFailure(eventHandler); @@ -2367,8 +2452,10 @@ export class CodexAcpServer { } catch (err) { logger.error(`Prompt for session ${params.sessionId} failed`, err); if (activePrompt.signal.aborted || this.sessionIsClosing(params.sessionId)) { - return this.cancelledPromptResponse(sessionState); + return cancelledPromptResponse(); } + agentFileChangeReportTurnId = null; + agentFileChangeReportUnavailableReason = "providerError"; const isProcessExit = err instanceof RequestError && err.code === CODEX_PROCESS_EXITED_ERROR_CODE; const isUnexpectedFailure = !(err instanceof RequestError); @@ -2394,6 +2481,15 @@ export class CodexAcpServer { // The app-server subscription is session-scoped and outlives this prompt. Flip routing before // awaiting disposal so queued late notifications cannot enter prompt-local buffers. promptNotificationsActive = false; + if (agentFileChangeReportRequest !== null) { + await this.publishAgentFileChangeReport( + sessionState, + agentFileChangeReportTurnId, + agentFileChangeReportRequest, + agentFileChangeReportUnavailableReason, + activePrompt.signal, + ); + } logger.log("Prompt completed", {sessionId: params.sessionId}); await eventHandler?.dispose(); disposePromptRequestCancellation(); diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index 50a95e42..1e741eba 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -43,6 +43,8 @@ import type { ThreadGoalGetResponse, ThreadGoalSetParams, ThreadGoalSetResponse, + ThreadForkParams, + ThreadForkResponse, ThreadLoadedListParams, ThreadLoadedListResponse, ThreadListParams, @@ -530,6 +532,10 @@ export class CodexAppServerClient { return await this.sendRequest({ method: "thread/resume", params: params }); } + async threadFork(params: ThreadForkParams): Promise { + return await this.sendRequest({ method: "thread/fork", params: params }); + } + getThreadSettings(threadId: string): ThreadSettings | undefined { return this.threadSettings.get(threadId); } diff --git a/src/__tests__/AgentFileChangeReport.test.ts b/src/__tests__/AgentFileChangeReport.test.ts new file mode 100644 index 00000000..264b88b7 --- /dev/null +++ b/src/__tests__/AgentFileChangeReport.test.ts @@ -0,0 +1,216 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import {describe, expect, it} from "vitest"; +import { + AGENT_FILE_CHANGE_REPORT_MAX_PATH_LENGTH, + AGENT_FILE_CHANGE_REPORT_MAX_TOTAL_BYTES, + AgentFileChangeReportError, + createReportedAgentFileChangeReport, + parseAgentFileChangeReportRequest, +} from "../AgentFileChangeReport"; +import type {Turn} from "../app-server/v2"; + +function completedReport(value: unknown): Turn { + return { + id: "audit-turn-id", + items: [{ + type: "agentMessage", + id: "audit-message-id", + text: JSON.stringify(value), + phase: "final_answer", + memoryCitation: null, + }], + itemsView: "full", + status: "completed", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; +} + +describe("agent file-change report", () => { + it("accepts only a versioned request with a bounded opaque id", () => { + expect(parseAgentFileChangeReportRequest({ + jetbrains: {air: {agentFileChangeReportRequest: {version: 1, requestId: "turn.42:audit-1"}}}, + })).toEqual({version: 1, requestId: "turn.42:audit-1"}); + + for (const request of [ + {version: 2, requestId: "request-id"}, + {version: 1, requestId: "contains spaces"}, + {version: 1, requestId: "x".repeat(129)}, + {version: 1, requestId: "request-id", extra: true}, + true, + ]) { + expect(parseAgentFileChangeReportRequest({ + jetbrains: {air: {agentFileChangeReportRequest: request}}, + })).toBeNull(); + } + expect(parseAgentFileChangeReportRequest({jetbrains: {air: {}}})).toBeNull(); + }); + + it("normalizes POSIX paths, keeps additional roots, and marks rejected paths", () => { + const report = createReportedAgentFileChangeReport( + "request-id", + completedReport({ + paths: [ + "src/A.kt", + "/repo/src/A.kt", + "../outside.kt", + "/repository/not-inside.kt", + "/generated/out.txt", + ".", + "bad\u0085path.kt", + ], + complete: true, + uncertainty: " generator output may be incomplete ", + }), + {cwd: "/repo", additionalDirectories: ["/generated"]}, + ); + + expect(report).toEqual({ + version: 1, + requestId: "request-id", + status: "reported", + paths: ["/repo/src/A.kt", "/generated/out.txt"], + declaredComplete: false, + truncated: true, + uncertainty: "generator output may be incomplete", + }); + }); + + it("deduplicates Windows paths case-insensitively on a non-Windows host", () => { + const report = createReportedAgentFileChangeReport( + "request-id", + completedReport({ + paths: ["src\\A.kt", "c:/work/repo/SRC/a.KT", "D:/Generated/out.bin"], + complete: true, + }), + {cwd: "C:\\Work\\Repo", additionalDirectories: ["D:\\Generated"]}, + ); + + expect(report.paths).toEqual([ + "C:\\Work\\Repo\\src\\A.kt", + "D:\\Generated\\out.bin", + ]); + expect(report.declaredComplete).toBe(true); + expect(report.truncated).toBe(false); + }); + + it("supports UNC roots and rejects share-prefix collisions", () => { + const report = createReportedAgentFileChangeReport( + "request-id", + completedReport({ + paths: [ + "folder/file.txt", + "//SERVER/Share/folder/FILE.txt", + "//server/share-other/out.txt", + "\\\\server", + "\\\\?\\C:\\unsafe.txt", + ], + complete: true, + }), + {cwd: "\\\\server\\share", additionalDirectories: []}, + ); + + expect(report.paths).toEqual(["\\\\server\\share\\folder\\file.txt"]); + expect(report.declaredComplete).toBe(false); + expect(report.truncated).toBe(true); + }); + + it("accepts canonical paths under a symlinked workspace root", () => { + if (process.platform === "win32") return; + + const realRoot = fs.mkdtempSync(path.join(os.tmpdir(), "file-audit-real-")); + const linkedRoot = `${realRoot}-link`; + fs.symlinkSync(realRoot, linkedRoot, "dir"); + try { + const canonicalPath = path.join(fs.realpathSync.native(realRoot), "generated.ts"); + const report = createReportedAgentFileChangeReport( + "request-symlink-root", + completedReport({ + paths: [canonicalPath, path.join(linkedRoot, "generated.ts")], + complete: true, + }), + {cwd: linkedRoot, additionalDirectories: []}, + ); + + expect(report).toMatchObject({ + paths: [canonicalPath], + declaredComplete: true, + truncated: false, + }); + } finally { + fs.unlinkSync(linkedRoot); + fs.rmSync(realRoot, {recursive: true, force: true}); + } + }); + + it("keeps valid paths when another path exceeds the per-path cap", () => { + const report = createReportedAgentFileChangeReport( + "request-id", + completedReport({ + paths: ["valid.txt", "x".repeat(AGENT_FILE_CHANGE_REPORT_MAX_PATH_LENGTH + 1)], + complete: true, + }), + {cwd: "/repo", additionalDirectories: []}, + ); + + expect(report.paths).toEqual(["/repo/valid.txt"]); + expect(report.declaredComplete).toBe(false); + expect(report.truncated).toBe(true); + }); + + it("applies the per-path cap after resolving a relative path", () => { + const longCwd = `/${"x".repeat(AGENT_FILE_CHANGE_REPORT_MAX_PATH_LENGTH - 2)}`; + const report = createReportedAgentFileChangeReport( + "request-id", + completedReport({paths: ["file.txt"], complete: true}), + {cwd: longCwd, additionalDirectories: []}, + ); + + expect(report.paths).toEqual([]); + expect(report.declaredComplete).toBe(false); + expect(report.truncated).toBe(true); + }); + + it("caps the serialized report rather than only the raw path bytes", () => { + const report = createReportedAgentFileChangeReport( + "request-id", + completedReport({ + paths: Array.from( + {length: 1_024}, + (_, index) => `generated/${index}-${"x".repeat(240)}.txt`, + ), + complete: true, + }), + {cwd: "/repo", additionalDirectories: []}, + ); + + expect(Buffer.byteLength(JSON.stringify(report), "utf8")) + .toBeLessThanOrEqual(AGENT_FILE_CHANGE_REPORT_MAX_TOTAL_BYTES); + expect(report.declaredComplete).toBe(false); + expect(report.truncated).toBe(true); + }); + + it("classifies malformed model output without exposing it on the wire", () => { + const turn = completedReport({paths: ["a.txt"], complete: "yes"}); + + expect(() => createReportedAgentFileChangeReport( + "request-id", + turn, + {cwd: "/repo", additionalDirectories: []}, + )).toThrow(AgentFileChangeReportError); + + try { + createReportedAgentFileChangeReport( + "request-id", + turn, + {cwd: "/repo", additionalDirectories: []}, + ); + } catch (error) { + expect(error).toMatchObject({reason: "invalidOutput"}); + } + }); +}); diff --git a/src/__tests__/CodexACPAgent/agent-file-change-report.test.ts b/src/__tests__/CodexACPAgent/agent-file-change-report.test.ts new file mode 100644 index 00000000..abc1e0f5 --- /dev/null +++ b/src/__tests__/CodexACPAgent/agent-file-change-report.test.ts @@ -0,0 +1,555 @@ +import {afterEach, beforeEach, describe, expect, it, vi} from "vitest"; +import * as acp from "@agentclientprotocol/sdk"; +import { + createCodexMockTestFixture, + createTestSessionState, + type CodexMockTestFixture, +} from "../acp-test-utils"; +import type {SessionState} from "../../CodexAcpServer"; +import { + AGENT_FILE_CHANGE_REPORT_OUTPUT_SCHEMA, + AGENT_FILE_CHANGE_REPORT_TIMEOUT_MS, +} from "../../AgentFileChangeReport"; +import {CodexCommands} from "../../CodexCommands"; +import type { + ThreadForkResponse, + ThreadReadResponse, + Turn, + TurnCompletedNotification, +} from "../../app-server/v2"; + +function createTurn( + id: string, + status: Turn["status"], + items: Turn["items"] = [], + itemsView: Turn["itemsView"] = "notLoaded", +): Turn { + return { + id, + items, + itemsView, + status, + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; +} + +function createForkResponse(threadId: string): ThreadForkResponse { + return {thread: {id: threadId}} as ThreadForkResponse; +} + +function createThreadReadResponse(threadId: string, turns: Turn[]): ThreadReadResponse { + return {thread: {id: threadId, turns}} as ThreadReadResponse; +} + +function promptWithFileChangeReport( + sessionId: string, + requestId: string, + text = "make the change", +): acp.PromptRequest { + return { + sessionId, + prompt: [{type: "text", text}], + _meta: { + jetbrains: { + air: { + agentFileChangeReportRequest: {version: 1, requestId}, + }, + }, + }, + }; +} + +function reportedUpdates(fixture: CodexMockTestFixture): unknown[] { + return fixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update) + .filter(update => update.sessionUpdate === "session_info_update" + && update._meta?.jetbrains?.air?.agentFileChangeReport !== undefined); +} + +const FILE_CHANGE_REPORT_CLIENT_CAPABILITIES: acp.ClientCapabilities = { + _meta: { + jetbrains: { + air: { + version: 1, + capabilities: ["agentFileChangeReport"], + }, + }, + }, +}; + +async function setupMainPrompt(negotiateCapability = true): Promise<{ + fixture: CodexMockTestFixture; + sessionState: SessionState; + turnStart: ReturnType; + awaitTurnCompleted: ReturnType; +}> { + const fixture = createCodexMockTestFixture(); + await fixture.getCodexAcpAgent().initialize({ + protocolVersion: acp.PROTOCOL_VERSION, + ...(negotiateCapability ? {clientCapabilities: FILE_CHANGE_REPORT_CLIENT_CAPABILITIES} : {}), + }); + const sessionState = createTestSessionState({ + cwd: "/workspace", + additionalDirectories: ["/generated"], + }); + vi.spyOn(fixture.getCodexAcpAgent(), "getSessionState").mockReturnValue(sessionState); + const turnStart = vi.spyOn(fixture.getCodexAppServerClient(), "turnStart") + .mockResolvedValueOnce({turn: createTurn("main-turn", "inProgress")}); + const awaitTurnCompleted = vi.spyOn(fixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockResolvedValueOnce({ + threadId: sessionState.sessionId, + turn: createTurn("main-turn", "completed"), + }); + return {fixture, sessionState, turnStart, awaitTurnCompleted}; +} + +describe("agent file-change report lifecycle", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("advertises the AIR capability", async () => { + const fixture = createCodexMockTestFixture(); + + const response = await fixture.getCodexAcpAgent().initialize({ + protocolVersion: acp.PROTOCOL_VERSION, + }); + + expect(response._meta).toMatchObject({ + jetbrains: { + air: { + version: 1, + capabilities: expect.arrayContaining(["agentFileChangeReport"]), + }, + }, + }); + }); + + it("runs a hidden read-only fork and publishes one correlated report", async () => { + const {fixture, sessionState, turnStart, awaitTurnCompleted} = await setupMainPrompt(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "threadFork").mockResolvedValue(createForkResponse("audit-thread")); + turnStart.mockResolvedValueOnce({turn: createTurn("audit-turn", "inProgress")}); + awaitTurnCompleted.mockImplementationOnce(async (): Promise => { + fixture.sendServerNotification({ + method: "item/agentMessage/delta", + params: { + threadId: "audit-thread", + turnId: "audit-turn", + itemId: "audit-message", + delta: "this must stay hidden", + }, + }); + return { + threadId: "audit-thread", + turn: createTurn("audit-turn", "completed"), + }; + }); + vi.spyOn(appServer, "threadRead").mockResolvedValue(createThreadReadResponse("audit-thread", [ + createTurn("audit-turn", "completed", [{ + type: "agentMessage", + id: "audit-message", + text: JSON.stringify({ + paths: ["src/Main.kt", "/generated/output.txt"], + complete: true, + }), + phase: "final_answer", + memoryCitation: null, + }], "full"), + ])); + const unsubscribe = vi.spyOn(appServer, "threadUnsubscribe") + .mockResolvedValue({status: "unsubscribed"}); + + await expect(fixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "make the change"}], + _meta: { + jetbrains: { + air: { + agentFileChangeReportRequest: {version: 1, requestId: "request-42"}, + }, + }, + }, + })).resolves.toMatchObject({stopReason: "end_turn"}); + + expect(appServer.threadFork).toHaveBeenCalledWith({ + threadId: sessionState.sessionId, + lastTurnId: "main-turn", + cwd: "/workspace", + approvalPolicy: "never", + sandbox: "read-only", + developerInstructions: expect.any(String), + ephemeral: true, + }); + expect(turnStart).toHaveBeenNthCalledWith(2, { + threadId: "audit-thread", + input: [{type: "text", text: expect.any(String), text_elements: []}], + cwd: "/workspace", + approvalPolicy: "never", + sandboxPolicy: {type: "readOnly", networkAccess: false}, + summary: "none", + outputSchema: AGENT_FILE_CHANGE_REPORT_OUTPUT_SCHEMA, + }); + expect(appServer.threadRead).toHaveBeenCalledWith({ + threadId: "audit-thread", + includeTurns: true, + }); + expect(unsubscribe).toHaveBeenCalledWith({threadId: "audit-thread"}); + + const acpEvents = fixture.getAcpConnectionEvents([]); + expect(acpEvents).toEqual([{ + method: "sessionUpdate", + args: [{ + sessionId: sessionState.sessionId, + update: { + sessionUpdate: "session_info_update", + _meta: { + jetbrains: { + air: { + version: 1, + agentFileChangeReport: { + version: 1, + requestId: "request-42", + status: "reported", + paths: ["/workspace/src/Main.kt", "/generated/output.txt"], + declaredComplete: true, + truncated: false, + }, + }, + }, + }, + }, + }], + }]); + }); + + it("does not fork for absent or malformed opt-in metadata", async () => { + const {fixture, sessionState} = await setupMainPrompt(); + const fork = vi.spyOn(fixture.getCodexAppServerClient(), "threadFork"); + + await expect(fixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "ordinary prompt"}], + _meta: { + jetbrains: { + air: { + agentFileChangeReportRequest: {version: 1, requestId: "invalid id"}, + }, + }, + }, + })).resolves.toMatchObject({stopReason: "end_turn"}); + + expect(fork).not.toHaveBeenCalled(); + expect(fixture.getAcpConnectionEvents([])).toEqual([]); + }); + + it("ignores a valid request when the client did not negotiate the capability", async () => { + const {fixture, sessionState} = await setupMainPrompt(false); + const fork = vi.spyOn(fixture.getCodexAppServerClient(), "threadFork"); + + await expect(fixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "ordinary prompt"}], + _meta: { + jetbrains: { + air: { + agentFileChangeReportRequest: {version: 1, requestId: "request-44"}, + }, + }, + }, + })).resolves.toMatchObject({stopReason: "end_turn"}); + + expect(fork).not.toHaveBeenCalled(); + expect(fixture.getAcpConnectionEvents([])).toEqual([]); + }); + + it("keeps the completed main prompt successful when the hidden audit is cancelled", async () => { + const {fixture, sessionState, turnStart, awaitTurnCompleted} = await setupMainPrompt(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "threadFork").mockResolvedValue(createForkResponse("audit-thread")); + turnStart.mockResolvedValueOnce({turn: createTurn("audit-turn", "inProgress")}); + awaitTurnCompleted.mockReturnValueOnce(new Promise(() => {})); + vi.spyOn(appServer, "turnInterrupt").mockResolvedValue({}); + vi.spyOn(appServer, "threadUnsubscribe").mockResolvedValue({status: "unsubscribed"}); + const cancellation = new AbortController(); + + const prompt = fixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "make the change"}], + _meta: { + jetbrains: { + air: { + agentFileChangeReportRequest: {version: 1, requestId: "request-cancelled"}, + }, + }, + }, + }, cancellation.signal); + await vi.waitFor(() => expect(turnStart).toHaveBeenCalledTimes(2)); + cancellation.abort(); + + await expect(prompt).resolves.toMatchObject({stopReason: "end_turn"}); + expect(fixture.getAcpConnectionEvents([])).toEqual([{ + method: "sessionUpdate", + args: [{ + sessionId: sessionState.sessionId, + update: { + sessionUpdate: "session_info_update", + _meta: { + jetbrains: { + air: { + version: 1, + agentFileChangeReport: { + version: 1, + requestId: "request-cancelled", + status: "unavailable", + reason: "cancelled", + }, + }, + }, + }, + }, + }], + }]); + }); + + it("publishes cancelled once when the prompt ends before a turn starts", async () => { + const {fixture, sessionState} = await setupMainPrompt(); + const appServer = fixture.getCodexAppServerClient(); + const fork = vi.spyOn(appServer, "threadFork"); + const cancellation = new AbortController(); + cancellation.abort(); + + await expect(fixture.getCodexAcpAgent().prompt( + promptWithFileChangeReport(sessionState.sessionId, "request-early-cancel"), + cancellation.signal, + )).resolves.toMatchObject({stopReason: "cancelled"}); + + expect(fork).not.toHaveBeenCalled(); + expect(reportedUpdates(fixture)).toEqual([{ + sessionUpdate: "session_info_update", + _meta: { + jetbrains: { + air: { + version: 1, + agentFileChangeReport: { + version: 1, + requestId: "request-early-cancel", + status: "unavailable", + reason: "cancelled", + }, + }, + }, + }, + }]); + }); + + it("publishes notReported once for a local command without a provider turn", async () => { + const {fixture, sessionState} = await setupMainPrompt(); + const command = vi.spyOn(CodexCommands.prototype, "tryHandleCommand") + .mockResolvedValue({handled: true}); + const fork = vi.spyOn(fixture.getCodexAppServerClient(), "threadFork"); + try { + await expect(fixture.getCodexAcpAgent().prompt( + promptWithFileChangeReport(sessionState.sessionId, "request-local", "/status"), + )).resolves.toMatchObject({stopReason: "end_turn"}); + } finally { + command.mockRestore(); + } + + expect(fork).not.toHaveBeenCalled(); + expect(reportedUpdates(fixture)).toEqual([{ + sessionUpdate: "session_info_update", + _meta: { + jetbrains: { + air: { + version: 1, + agentFileChangeReport: { + version: 1, + requestId: "request-local", + status: "unavailable", + reason: "notReported", + }, + }, + }, + }, + }]); + }); + + it("publishes providerError once when the provider turn fails", async () => { + const {fixture, sessionState, turnStart} = await setupMainPrompt(); + turnStart.mockReset(); + turnStart.mockRejectedValue(new Error("provider failed")); + const fork = vi.spyOn(fixture.getCodexAppServerClient(), "threadFork"); + + await expect(fixture.getCodexAcpAgent().prompt( + promptWithFileChangeReport(sessionState.sessionId, "request-provider-error"), + )).rejects.toThrow("provider failed"); + + expect(fork).not.toHaveBeenCalled(); + expect(reportedUpdates(fixture)).toEqual([{ + sessionUpdate: "session_info_update", + _meta: { + jetbrains: { + air: { + version: 1, + agentFileChangeReport: { + version: 1, + requestId: "request-provider-error", + status: "unavailable", + reason: "providerError", + }, + }, + }, + }, + }]); + }); + + it("bounds a stuck fork with the shared audit deadline", async () => { + vi.useFakeTimers(); + const {fixture, sessionState} = await setupMainPrompt(); + const appServer = fixture.getCodexAppServerClient(); + const fork = vi.spyOn(appServer, "threadFork").mockReturnValue(new Promise(() => {})); + + const prompt = fixture.getCodexAcpAgent().prompt( + promptWithFileChangeReport(sessionState.sessionId, "request-fork-timeout"), + ); + await vi.advanceTimersByTimeAsync(0); + expect(fork).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(AGENT_FILE_CHANGE_REPORT_TIMEOUT_MS); + + await expect(prompt).resolves.toMatchObject({stopReason: "end_turn"}); + expect(reportedUpdates(fixture)).toHaveLength(1); + expect(reportedUpdates(fixture)[0]).toMatchObject({ + _meta: {jetbrains: {air: {agentFileChangeReport: { + requestId: "request-fork-timeout", + status: "unavailable", + reason: "timeout", + }}}}, + }); + }); + + it("does not wait past the shared deadline for interrupt or unsubscribe", async () => { + vi.useFakeTimers(); + const {fixture, sessionState, turnStart, awaitTurnCompleted} = await setupMainPrompt(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "threadFork").mockResolvedValue(createForkResponse("audit-thread")); + turnStart.mockResolvedValueOnce({turn: createTurn("audit-turn", "inProgress")}); + awaitTurnCompleted.mockReturnValueOnce(new Promise(() => {})); + const interrupt = vi.spyOn(appServer, "turnInterrupt").mockReturnValue(new Promise(() => {})); + const unsubscribe = vi.spyOn(appServer, "threadUnsubscribe").mockReturnValue(new Promise(() => {})); + + const prompt = fixture.getCodexAcpAgent().prompt( + promptWithFileChangeReport(sessionState.sessionId, "request-cleanup-timeout"), + ); + await vi.advanceTimersByTimeAsync(0); + expect(turnStart).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(AGENT_FILE_CHANGE_REPORT_TIMEOUT_MS); + + await expect(prompt).resolves.toMatchObject({stopReason: "end_turn"}); + expect(interrupt).toHaveBeenCalledWith({threadId: "audit-thread", turnId: "audit-turn"}); + expect(unsubscribe).toHaveBeenCalledWith({threadId: "audit-thread"}); + expect(reportedUpdates(fixture)).toHaveLength(1); + expect(reportedUpdates(fixture)[0]).toMatchObject({ + _meta: {jetbrains: {air: {agentFileChangeReport: { + requestId: "request-cleanup-timeout", + status: "unavailable", + reason: "timeout", + }}}}, + }); + }); + + it("bounds a stuck thread read with the same audit deadline", async () => { + vi.useFakeTimers(); + const {fixture, sessionState, turnStart, awaitTurnCompleted} = await setupMainPrompt(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "threadFork").mockResolvedValue(createForkResponse("audit-thread")); + turnStart.mockResolvedValueOnce({turn: createTurn("audit-turn", "inProgress")}); + awaitTurnCompleted.mockResolvedValueOnce({ + threadId: "audit-thread", + turn: createTurn("audit-turn", "completed"), + }); + const read = vi.spyOn(appServer, "threadRead").mockReturnValue(new Promise(() => {})); + vi.spyOn(appServer, "threadUnsubscribe").mockReturnValue(new Promise(() => {})); + + const prompt = fixture.getCodexAcpAgent().prompt( + promptWithFileChangeReport(sessionState.sessionId, "request-read-timeout"), + ); + await vi.advanceTimersByTimeAsync(0); + expect(read).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(AGENT_FILE_CHANGE_REPORT_TIMEOUT_MS); + + await expect(prompt).resolves.toMatchObject({stopReason: "end_turn"}); + expect(reportedUpdates(fixture)).toHaveLength(1); + expect(reportedUpdates(fixture)[0]).toMatchObject({ + _meta: {jetbrains: {air: {agentFileChangeReport: { + requestId: "request-read-timeout", + status: "unavailable", + reason: "timeout", + }}}}, + }); + }); + + it("reports invalid audit output as unavailable without failing the prompt", async () => { + const {fixture, sessionState, turnStart, awaitTurnCompleted} = await setupMainPrompt(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "threadFork").mockResolvedValue(createForkResponse("audit-thread")); + turnStart.mockResolvedValueOnce({turn: createTurn("audit-turn", "inProgress")}); + awaitTurnCompleted.mockResolvedValueOnce({ + threadId: "audit-thread", + turn: createTurn("audit-turn", "completed"), + }); + vi.spyOn(appServer, "threadRead").mockResolvedValue(createThreadReadResponse("audit-thread", [ + createTurn("audit-turn", "completed", [{ + type: "agentMessage", + id: "audit-message", + text: "not JSON", + phase: "final_answer", + memoryCitation: null, + }], "full"), + ])); + vi.spyOn(appServer, "threadUnsubscribe").mockResolvedValue({status: "unsubscribed"}); + + await expect(fixture.getCodexAcpAgent().prompt({ + sessionId: sessionState.sessionId, + prompt: [{type: "text", text: "make the change"}], + _meta: { + jetbrains: { + air: { + agentFileChangeReportRequest: {version: 1, requestId: "request-43"}, + }, + }, + }, + })).resolves.toMatchObject({stopReason: "end_turn"}); + + expect(fixture.getAcpConnectionEvents([])).toEqual([{ + method: "sessionUpdate", + args: [{ + sessionId: sessionState.sessionId, + update: { + sessionUpdate: "session_info_update", + _meta: { + jetbrains: { + air: { + version: 1, + agentFileChangeReport: { + version: 1, + requestId: "request-43", + status: "unavailable", + reason: "invalidOutput", + }, + }, + }, + }, + }, + }], + }]); + }); +}); diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 811cef28..739c4028 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -73,7 +73,7 @@ describe('CodexACPAgent - initialize', () => { jetbrains: { air: { version: 1, - capabilities: ["sessionFailure"], + capabilities: ["sessionFailure", "agentFileChangeReport"], }, }, },