From 8157ee113e705750be4eb6cc787bdd12f1db84ff Mon Sep 17 00:00:00 2001 From: nikita-ashihmin Date: Mon, 10 Aug 2026 14:15:44 +0400 Subject: [PATCH] feat: expose typed session failures for AIR (#979) ## Summary - Negotiate the JetBrains AIR `sessionFailure` extension when the client advertises an integer version `>= 1` and the capability name. - Return turn-terminal failures only on `PromptResponse._meta`; publish idle and asynchronous session health failures through `session/update`. - Preserve legacy ACP behavior for clients that do not negotiate the extension. - Emit bounded fixed safe messages without SDK/provider details, paths, credentials, or stack traces. ## Recovery contract Failures use `_meta.jetbrains.air.sessionFailure` with a stable id, monotonic revision, phase, category, source, safe message, retryability, and optional real turn attribution. Session-scoped ids include a consumer epoch, so replayed tombstones from a dead consumer cannot suppress failures after restart. Stale consumers cannot publish into a replacement session. AIR owns localized action presentation. Claude supplies typed category and retryability facts. Billing maps to quota exhaustion, max-output limits map to context exhaustion, and the irreversibly closed Claude Query reports non-retryable `transport_lost` so AIR offers New Session. A held-open turn keeps its already recorded successful `PromptResponse`, while worker shutdown or transport death also emits a separate session-scoped health failure without `turnId`; the next prompt is proven to fail because the Query is closed. Worker-shutdown remains armed through buffered trailing frames and is classified only when EOF proves the live tail. ## Validation - 750 passed, 25 skipped. - Build passed. - ESLint passed. - Changed files pass Prettier; the repository-wide check is blocked only by an unrelated untracked local proposal file. - Tests cover negotiation, taxonomy, sanitization, terminal/asynchronous carrier separation, stale consumers, worker replay/live-tail behavior, held turns, process/transport failure, restart-safe ids, clear revisions, and capability-off behavior. --- src/acp-agent.ts | 398 ++++++++++++++- src/tests/acp-agent.test.ts | 952 +++++++++++++++++++++++++++++++++++- 2 files changed, 1333 insertions(+), 17 deletions(-) diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 0636c263..2f4c65bd 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -916,11 +916,135 @@ export type ToolUpdateMeta = { }; const SUBAGENT_TRANSCRIPT_CAPABILITY = "subagent-transcript"; +// This is a protocol namespace, not user-facing branding: `jetbrains` owns the +// non-standard contract and `air` identifies the client defining its rendering +// semantics, without affecting unrelated JetBrains ACP clients. +const JETBRAINS_META_KEY = "jetbrains"; +const AIR_META_KEY = "air"; +const AIR_EXTENSION_VERSION_KEY = "version"; +const AIR_EXTENSION_CAPABILITIES_KEY = "capabilities"; +const AIR_SESSION_FAILURE_KEY = "sessionFailure"; +const AIR_EXTENSION_VERSION = 1; function supportsSubagentTranscript(capabilities?: ClientCapabilities | null): boolean { return capabilities?._meta?.[SUBAGENT_TRANSCRIPT_CAPABILITY] === true; } +type AirSessionFailureCategory = + | "auth_required" + | "bad_request" + | "budget_exhausted" + | "context_exhausted" + | "internal_error" + | "overloaded" + | "provider_error" + | "quota_exhausted" + | "rate_limited" + | "transport_lost" + | "worker_shutdown"; + +const AIR_FAILURE_PRESENTATION: Record< + AirSessionFailureCategory, + { message: string; retryable: boolean; actions: string[] } +> = { + auth_required: { + message: "Sign in to continue using Claude.", + retryable: false, + actions: ["login"], + }, + bad_request: { + message: "Claude could not process this request.", + retryable: false, + actions: ["new_turn"], + }, + budget_exhausted: { + message: "This Claude session reached its configured budget.", + retryable: false, + actions: ["new_session"], + }, + context_exhausted: { + message: "This Claude turn reached its configured limit.", + retryable: false, + actions: ["new_turn"], + }, + internal_error: { + message: "Claude Agent encountered an internal error.", + retryable: true, + actions: ["retry"], + }, + overloaded: { + message: "Claude is temporarily overloaded.", + retryable: true, + actions: ["retry"], + }, + provider_error: { + message: "The model provider reported an error.", + retryable: true, + actions: ["retry"], + }, + quota_exhausted: { + message: "The Claude account has no available quota.", + retryable: false, + actions: ["new_session"], + }, + rate_limited: { + message: "Claude is temporarily rate limited.", + retryable: true, + actions: ["retry"], + }, + transport_lost: { + message: "The connection to Claude was lost.", + retryable: false, + actions: ["new_session"], + }, + worker_shutdown: { + message: "The Claude worker is shutting down.", + retryable: false, + actions: ["new_session"], + }, +}; + +function supportsAirSessionFailures(capabilities?: ClientCapabilities): 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]; + const advertised = air?.[AIR_EXTENSION_CAPABILITIES_KEY]; + return ( + typeof version === "number" && + Number.isFinite(version) && + Number.isInteger(version) && + version >= AIR_EXTENSION_VERSION && + Array.isArray(advertised) && + advertised.some((capability) => capability === AIR_SESSION_FAILURE_KEY) + ); +} + +function providerFailureCategory(errorKind?: SDKAssistantMessageError): AirSessionFailureCategory { + switch (errorKind) { + case "authentication_failed": + case "oauth_org_not_allowed": + return "auth_required"; + case "billing_error": + return "quota_exhausted"; + case "rate_limit": + return "rate_limited"; + case "overloaded": + return "overloaded"; + case "invalid_request": + case "model_not_found": + return "bad_request"; + case "max_output_tokens": + return "context_exhausted"; + case "server_error": + case "unknown": + case undefined: + return "provider_error"; + default: + return "provider_error"; + } +} + function parentToolUseIdOf(message: { parent_tool_use_id?: unknown }): string | null { if (!("parent_tool_use_id" in message)) return null; return typeof message.parent_tool_use_id === "string" ? message.parent_tool_use_id : null; @@ -1623,6 +1747,12 @@ export class ClaudeAcpAgent { // steering extension contract: advertises the `_session/steering` request // so clients know they may inject a follow-up into a running turn. _meta: { + [JETBRAINS_META_KEY]: { + [AIR_META_KEY]: { + [AIR_EXTENSION_VERSION_KEY]: AIR_EXTENSION_VERSION, + [AIR_EXTENSION_CAPABILITIES_KEY]: [AIR_SESSION_FAILURE_KEY], + }, + }, steering: { supported: true, }, @@ -2186,6 +2316,127 @@ export class ClaudeAcpAgent { await this.client.sessionUpdate(notification); }; + type PublishedSessionFailure = { + id: string; + revision: number; + category: AirSessionFailureCategory; + turnId?: string; + }; + const failureRevisions = new Map(); + const sessionFailureEpoch = randomUUID(); + const activeSessionFailures = new Map(); + let pendingWorkerShutdown = false; + const isCurrentConsumer = () => this.sessions[params.sessionId] === session; + + const sessionFailureMeta = (failure: PublishedSessionFailure, phase: "active" | "cleared") => { + const presentation = AIR_FAILURE_PRESENTATION[failure.category]; + return { + [JETBRAINS_META_KEY]: { + [AIR_META_KEY]: { + [AIR_EXTENSION_VERSION_KEY]: AIR_EXTENSION_VERSION, + [AIR_SESSION_FAILURE_KEY]: { + id: failure.id, + revision: failure.revision, + phase, + category: failure.category, + source: "claude", + safeMessage: presentation.message, + retryable: presentation.retryable, + actions: presentation.actions, + ...(failure.turnId ? { turnId: failure.turnId } : {}), + }, + }, + }, + }; + }; + + const emitSessionFailure = async ( + failure: PublishedSessionFailure, + phase: "active" | "cleared", + ): Promise => { + if (!isCurrentConsumer()) { + this.logger.error( + `Session ${params.sessionId}: ignored AIR session failure from a stale query consumer`, + ); + return false; + } + try { + await sendUpdate({ + sessionId: params.sessionId, + update: { + sessionUpdate: "session_info_update", + _meta: sessionFailureMeta(failure, phase), + }, + }); + return true; + } catch (error) { + this.logger.error( + `Session ${params.sessionId}: failed to publish AIR session failure: ${error}`, + ); + return false; + } + }; + + const clearSessionFailure = async ( + shouldClear: (failure: PublishedSessionFailure) => boolean = () => true, + ) => { + if (!supportsAirSessionFailures(this.clientCapabilities)) return; + for (const failure of [...activeSessionFailures.values()]) { + if (!shouldClear(failure)) continue; + const cleared = { + ...failure, + revision: failure.revision + 1, + }; + if (await emitSessionFailure(cleared, "cleared")) { + failureRevisions.set(cleared.id, cleared.revision); + activeSessionFailures.delete(cleared.id); + } + } + }; + + const createSessionFailure = async ( + category: AirSessionFailureCategory, + options: { turnScoped?: boolean } = {}, + ): Promise => { + if (!supportsAirSessionFailures(this.clientCapabilities) || !isCurrentConsumer()) { + return undefined; + } + const turnId = options.turnScoped === false ? undefined : session.activeTurn?.promptUuid; + const id = turnId + ? `${turnId}:error` + : `${params.sessionId}:session-error:${sessionFailureEpoch}`; + await clearSessionFailure((failure) => failure.id !== id); + return { + id, + revision: (failureRevisions.get(id) ?? 0) + 1, + category, + ...(turnId ? { turnId } : {}), + }; + }; + + const publishSessionFailure = async ( + category: AirSessionFailureCategory, + options: { turnScoped?: boolean } = {}, + ) => { + const failure = await createSessionFailure(category, options); + if (!failure) return; + if (await emitSessionFailure(failure, "active")) { + failureRevisions.set(failure.id, failure.revision); + activeSessionFailures.set(failure.id, failure); + } + }; + + const clearFailuresFromEarlierTurns = async () => { + const activeTurnId = session.activeTurn?.promptUuid; + await clearSessionFailure((failure) => failure.turnId !== activeTurnId); + }; + + const internalErrorForClient = (data: unknown, rawDetail?: string) => + RequestError.internalError( + data, + supportsAirSessionFailures(this.clientCapabilities) ? undefined : rawDetail, + ); + const resetTurnScratch = () => { lastAssistantTotalUsage = null; lastAssistantUsage = null; @@ -2479,6 +2730,9 @@ export class ClaudeAcpAgent { disarmForceCancel(session); const turn = session.activeTurn; if (!turn || turn.settled) { + this.logger.error( + `Session ${params.sessionId}: cannot fail active turn because no unsettled active turn exists: ${error}`, + ); return; } turn.settled = true; @@ -2493,6 +2747,38 @@ export class ClaudeAcpAgent { turn.reject(error); }; + /** Complete a negotiated terminal failure on the prompt response itself, + * which is the canonical AIR carrier. Legacy clients keep the historical + * JSON-RPC rejection path. */ + const failActiveWithSessionFailure = async ( + category: AirSessionFailureCategory, + error: unknown, + ) => { + if (!supportsAirSessionFailures(this.clientCapabilities)) { + failActive(error); + return; + } + if (!session.activeTurn || session.activeTurn.settled) { + this.logger.error( + `Session ${params.sessionId}: cannot attach ${category} to a prompt response because no active turn exists; publishing a session-scoped failure`, + ); + await publishSessionFailure(category, { turnScoped: false }); + return; + } + const failure = await createSessionFailure(category); + if (!failure) { + failActive(error); + return; + } + failureRevisions.set(failure.id, failure.revision); + activeSessionFailures.set(failure.id, failure); + settleActive({ + stopReason: "end_turn", + usage: sessionUsage(session), + _meta: sessionFailureMeta(failure, "active"), + }); + }; + /** Reject every in-flight turn — used when the stream dies. */ const failAllTurns = (error: unknown) => { disarmForceCancel(session); @@ -2614,6 +2900,25 @@ export class ClaudeAcpAgent { const { value: message, done } = raced.result as IteratorResult; if (done || !message) { + if (pendingWorkerShutdown) { + pendingWorkerShutdown = false; + if (session.activeTurn) { + if (!isHeldOpen(session.activeTurn)) { + await failActiveWithSessionFailure( + "worker_shutdown", + internalErrorForClient({ errorKind: "worker_shutdown" }), + ); + } else { + // The held turn already has its authoritative terminal outcome, + // but EOF permanently closes the non-revivable Query. Preserve + // the turn result below and report the independent session-health + // failure without a turnId so AIR can offer a new session. + await publishSessionFailure("worker_shutdown", { turnScoped: false }); + } + } else { + await publishSessionFailure("worker_shutdown", { turnScoped: false }); + } + } // The stream ended. Settle the in-flight turns FIRST, then release the // stream resources — same order as the error paths (failAllTurns before // closeQueryStream). Settling is the user-facing contract; resource @@ -2970,7 +3275,8 @@ export class ClaudeAcpAgent { `Session ${params.sessionId}: SDK went idle without emitting a result ` + `for the active turn; failing the in-flight prompt (issue #825)`, ); - failActive( + await failActiveWithSessionFailure( + "internal_error", RequestError.internalError( errorKindData("no_result"), TURN_NO_RESULT_MESSAGE, @@ -3183,9 +3489,14 @@ export class ClaudeAcpAgent { } break; case "worker_shutting_down": - // A Remote Control worker announced a graceful teardown. This is a - // live-tail signal for remote clients to explain why a session went - // away; it's not meaningful for a local stdio ACP session. + // Defer until stream end. The announcement is durable and may be + // replayed before later frames, but those frames do not prove that + // a new worker epoch began: they can be buffered output from the + // shutting-down worker. Keep the signal armed until the transport + // actually ends. Deliberately do not add a quiet-period timer: the + // iterator has no replay/live boundary, so a timeout could publish + // while a slow replay is still in flight. + pendingWorkerShutdown = true; break; case "elicitation_complete": { // A url-mode MCP elicitation finished server-side. Let the client @@ -3506,6 +3817,7 @@ export class ClaudeAcpAgent { if (session.cancelled) { if (!isAutonomousResult) { + await clearFailuresFromEarlierTurns(); stopReason = "cancelled"; } break; @@ -3543,6 +3855,8 @@ export class ClaudeAcpAgent { break; } + await clearFailuresFromEarlierTurns(); + // A refusal can arrive on any result subtype (and may even set // is_error), so handle it before the subtype switch — otherwise the // is_error throw below would surface it as an internal error. The @@ -3583,10 +3897,17 @@ export class ClaudeAcpAgent { break; } + if (!message.is_error) { + await clearSessionFailure(); + } + switch (message.subtype) { case "success": { if (message.result.includes("Please run /login")) { - failActive(RequestError.authRequired()); + await failActiveWithSessionFailure( + "auth_required", + RequestError.authRequired(), + ); break; } if (message.stop_reason === "max_tokens") { @@ -3594,8 +3915,9 @@ export class ClaudeAcpAgent { break; } if (message.is_error) { - failActive( - RequestError.internalError(errorKindData(lastAssistantError), message.result), + await failActiveWithSessionFailure( + providerFailureCategory(lastAssistantError), + internalErrorForClient(errorKindData(lastAssistantError), message.result), ); break; } @@ -3641,8 +3963,9 @@ export class ClaudeAcpAgent { break; } if (message.is_error) { - failActive( - RequestError.internalError( + await failActiveWithSessionFailure( + providerFailureCategory(lastAssistantError), + internalErrorForClient( errorKindData(lastAssistantError), message.errors.join(", ") || message.subtype, ), @@ -3653,11 +3976,36 @@ export class ClaudeAcpAgent { break; } case "error_max_budget_usd": + if (message.is_error) { + await failActiveWithSessionFailure( + "budget_exhausted", + internalErrorForClient( + errorKindData(lastAssistantError), + message.errors.join(", ") || message.subtype, + ), + ); + break; + } + stopReason = "max_turn_requests"; + break; case "error_max_turns": + if (message.is_error) { + await failActiveWithSessionFailure( + "provider_error", + internalErrorForClient( + errorKindData(lastAssistantError), + message.errors.join(", ") || message.subtype, + ), + ); + break; + } + stopReason = "max_turn_requests"; + break; case "error_max_structured_output_retries": if (message.is_error) { - failActive( - RequestError.internalError( + await failActiveWithSessionFailure( + "provider_error", + internalErrorForClient( errorKindData(lastAssistantError), message.errors.join(", ") || message.subtype, ), @@ -4025,7 +4373,7 @@ export class ClaudeAcpAgent { } if (message.type === "assistant" && isSyntheticLoginMessage(message.message)) { - failActive(RequestError.authRequired()); + await failActiveWithSessionFailure("auth_required", RequestError.authRequired()); break; } @@ -4211,9 +4559,13 @@ export class ClaudeAcpAgent { break; } case "tool_use_summary": - case "auth_status": case "prompt_suggestion": break; + case "auth_status": + if (!message.isAuthenticating && message.error === undefined) { + await clearSessionFailure((failure) => failure.category === "auth_required"); + } + break; default: unreachable(message, this.logger); break; @@ -4234,6 +4586,20 @@ export class ClaudeAcpAgent { message.includes("process exited with") || message.includes("process terminated by signal") || message.includes("Failed to write to process stdin")); + if (supportsAirSessionFailures(this.clientCapabilities) && session.activeTurn) { + if (!isHeldOpen(session.activeTurn)) { + await failActiveWithSessionFailure( + "transport_lost", + internalErrorForClient({ errorKind: "transport_lost" }), + ); + } else { + // The held turn keeps its recorded PromptResponse outcome, while the + // exhausted Query makes the session independently unrecoverable. + await publishSessionFailure("transport_lost", { turnScoped: false }); + } + } else { + await publishSessionFailure("transport_lost", { turnScoped: false }); + } // Either way the query iterator is finished and the consumer is exiting, // so release its resources via closeQueryStream (idempotent). A process // death is unrecoverable, so additionally evict the session so the client @@ -4251,7 +4617,11 @@ export class ClaudeAcpAgent { delete this.sessions[params.sessionId]; } else { this.logger.error(`Session ${params.sessionId}: query stream error: ${message}`); - failAllTurns(error); + failAllTurns( + supportsAirSessionFailures(this.clientCapabilities) + ? internalErrorForClient({ errorKind: "transport_lost" }) + : error, + ); this.closeQueryStream(session); } } diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index bfb529fb..9fba24d6 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -3123,7 +3123,12 @@ describe("stop reason propagation", () => { } function createResultMessage(overrides: { - subtype: "success" | "error_during_execution"; + subtype: + | "success" + | "error_during_execution" + | "error_max_budget_usd" + | "error_max_turns" + | "error_max_structured_output_retries"; stop_reason: string | null; is_error: boolean; result?: string; @@ -3153,6 +3158,49 @@ describe("stop reason propagation", () => { }; } + function createAssistantError(error: SDKAssistantMessage["error"]): SDKAssistantMessage { + return { + type: "assistant", + parent_tool_use_id: null, + error, + uuid: randomUUID(), + session_id: "test-session", + message: { + id: `msg-${error}`, + type: "message", + role: "assistant", + container: null, + model: "claude-sonnet-4-20250514", + content: [], + stop_reason: "stop_sequence", + stop_sequence: null, + usage: { + input_tokens: 0, + output_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + server_tool_use: { web_search_requests: 0, web_fetch_requests: 0 }, + service_tier: null, + cache_creation: { + ephemeral_1h_input_tokens: 0, + ephemeral_5m_input_tokens: 0, + }, + } as any, + } as any, + }; + } + + const airSessionFailureCapabilities = { + _meta: { + jetbrains: { + air: { version: 1, capabilities: ["sessionFailure"] }, + }, + }, + }; + + const sessionFailureFromResponse = (response: PromptResponse) => + (response._meta as any)?.jetbrains?.air?.sessionFailure; + function injectSession(agent: ClaudeAcpAgent, messages: any[]) { const input = new Pushable(); async function* messageGenerator() { @@ -3934,6 +3982,783 @@ describe("stop reason propagation", () => { ).rejects.toThrow("Internal error"); }); + it.each([1, 2])( + "publishes a safe typed failure for compatible AIR version %i", + async (version) => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + await agent.initialize({ + protocolVersion: 1, + clientCapabilities: { + _meta: { jetbrains: { air: { version, capabilities: ["sessionFailure"] } } }, + }, + }); + injectSession(agent, [ + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "raw provider secret must not be shown", + }), + ]); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + expect(sessionFailureFromResponse(response)).toEqual( + expect.objectContaining({ + revision: 1, + phase: "active", + category: "provider_error", + source: "claude", + safeMessage: "The model provider reported an error.", + retryable: true, + actions: ["retry"], + }), + ); + expect(JSON.stringify(response)).not.toContain("raw provider secret"); + expect(JSON.stringify(updates)).not.toContain("sessionFailure"); + }, + ); + + it.each([ + ["missing envelope", {}], + ["missing capability", { _meta: { jetbrains: { air: { version: 1, capabilities: [] } } } }], + [ + "string version", + { _meta: { jetbrains: { air: { version: "1", capabilities: ["sessionFailure"] } } } }, + ], + [ + "fractional version", + { _meta: { jetbrains: { air: { version: 1.5, capabilities: ["sessionFailure"] } } } }, + ], + [ + "zero version", + { _meta: { jetbrains: { air: { version: 0, capabilities: ["sessionFailure"] } } } }, + ], + [ + "negative version", + { _meta: { jetbrains: { air: { version: -1, capabilities: ["sessionFailure"] } } } }, + ], + [ + "non-finite version", + { _meta: { jetbrains: { air: { version: Number.NaN, capabilities: ["sessionFailure"] } } } }, + ], + ["non-string capability", { _meta: { jetbrains: { air: { version: 1, capabilities: [1] } } } }], + [ + "legacy nested version", + { _meta: { jetbrains: { air: { sessionFailure: { version: 1 } } } } }, + ], + ])("does not publish typed failures for %s", async (_label, capabilities) => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + await agent.initialize({ + protocolVersion: 1, + clientCapabilities: capabilities, + }); + injectSession(agent, [ + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "provider detail", + }), + ]); + + await expect( + agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "test" }] }), + ).rejects.toThrow("provider detail"); + expect(JSON.stringify(updates)).not.toContain("sessionFailure"); + }); + + it.each([ + ["authentication_failed", "auth_required", false], + ["billing_error", "quota_exhausted", false], + ["overloaded", "overloaded", true], + ["invalid_request", "bad_request", false], + ["max_output_tokens", "context_exhausted", false], + ] as const)("maps SDK error %s to %s", async (sdkError, expectedCategory, retryable) => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + injectSession(agent, [ + createAssistantError(sdkError), + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "private provider detail", + }), + ]); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + expect(sessionFailureFromResponse(response)).toEqual( + expect.objectContaining({ category: expectedCategory, retryable }), + ); + expect(JSON.stringify(updates)).not.toContain("sessionFailure"); + }); + + it.each([ + ["error_max_budget_usd", "budget_exhausted"], + ["error_max_turns", "provider_error"], + ["error_max_structured_output_retries", "provider_error"], + ] as const)("maps terminal subtype %s to %s", async (subtype, expectedCategory) => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + injectSession(agent, [ + createResultMessage({ + subtype, + stop_reason: null, + is_error: true, + errors: ["private provider detail"], + }), + ]); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + expect(sessionFailureFromResponse(response)).toEqual( + expect.objectContaining({ category: expectedCategory }), + ); + expect(JSON.stringify(updates)).not.toContain("sessionFailure"); + }); + + it("clears auth_required after a successful auth_status without leaking auth output", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + injectSession(agent, [ + createAssistantError("authentication_failed"), + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "private auth detail", + }), + { + type: "auth_status", + isAuthenticating: false, + output: ["private login token"], + uuid: randomUUID(), + session_id: "test-session", + }, + ]); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + await agent.sessions["test-session"]?.consumer; + const active = sessionFailureFromResponse(response); + const cleared = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .find(Boolean); + expect(active).toEqual(expect.objectContaining({ category: "auth_required", revision: 1 })); + expect(cleared).toEqual( + expect.objectContaining({ id: active.id, revision: 2, phase: "cleared" }), + ); + expect(JSON.stringify({ response, updates })).not.toContain("private"); + }); + + it("keeps auth_required active when auth_status reports an error", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + injectSession(agent, [ + createAssistantError("authentication_failed"), + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "private auth detail", + }), + { + type: "auth_status", + isAuthenticating: false, + output: ["private login token"], + error: "private login error", + uuid: randomUUID(), + session_id: "test-session", + }, + ]); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + await agent.sessions["test-session"]?.consumer; + expect(sessionFailureFromResponse(response)).toEqual( + expect.objectContaining({ category: "auth_required", phase: "active" }), + ); + expect(JSON.stringify(updates)).not.toContain("sessionFailure"); + expect(JSON.stringify({ response, updates })).not.toContain("private"); + }); + + it("clears the active failure with the same id and a higher revision after recovery", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + const input = new Pushable(); + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const first = await iter.next(); + yield { + type: "user", + message: first.value.message, + parent_tool_use_id: null, + uuid: first.value.uuid, + session_id: "test-session", + isReplay: true, + }; + yield createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "first turn failed", + }); + const second = await iter.next(); + yield { + type: "user", + message: second.value.message, + parent_tool_use_id: null, + uuid: second.value.uuid, + session_id: "test-session", + isReplay: true, + }; + yield createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: false, + result: "recovered", + }); + } + agent.sessions["test-session"] = mockSessionState({ + query: wrapQuery(messageGenerator()), + input, + }); + + const failed = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await expect( + agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "second" }] }), + ).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" })); + + const failures = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .filter(Boolean); + const active = sessionFailureFromResponse(failed); + expect(active).toEqual(expect.objectContaining({ revision: 1, phase: "active" })); + expect(failures).toHaveLength(1); + expect(failures[0]).toEqual( + expect.objectContaining({ + id: active.id, + revision: 2, + phase: "cleared", + }), + ); + }); + + it("clears a previous turn failure when the next turn is refused", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + const input = new Pushable(); + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + for (const result of [ + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "first turn failed", + }), + createResultMessage({ + subtype: "success", + stop_reason: "refusal", + is_error: true, + }), + ]) { + const user = await iter.next(); + yield { + type: "user", + message: user.value.message, + parent_tool_use_id: null, + uuid: user.value.uuid, + session_id: "test-session", + isReplay: true, + }; + yield result; + } + } + agent.sessions["test-session"] = mockSessionState({ + query: wrapQuery(messageGenerator()), + input, + }); + + const failed = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await expect( + agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "second" }] }), + ).resolves.toEqual(expect.objectContaining({ stopReason: "refusal" })); + + const failures = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .filter(Boolean); + const active = sessionFailureFromResponse(failed); + expect(failures).toHaveLength(1); + expect(failures[0]).toEqual( + expect.objectContaining({ id: active.id, revision: 2, phase: "cleared" }), + ); + }); + + it("retries a clear whose first delivery failed", async () => { + const updates: SessionNotification[] = []; + let rejectFirstClear = true; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + const failure = (update.update._meta as any)?.jetbrains?.air?.sessionFailure; + if (failure?.phase === "cleared" && rejectFirstClear) { + rejectFirstClear = false; + throw new Error("temporary client write failure"); + } + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + const input = new Pushable(); + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const results = [ + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "first turn failed", + }), + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: false, + }), + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: false, + }), + ]; + for (const result of results) { + const user = await iter.next(); + yield { + type: "user", + message: user.value.message, + parent_tool_use_id: null, + uuid: user.value.uuid, + session_id: "test-session", + isReplay: true, + }; + yield result; + } + } + agent.sessions["test-session"] = mockSessionState({ + query: wrapQuery(messageGenerator()), + input, + }); + + const failed = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "first" }], + }); + await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "second" }] }); + await agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "third" }] }); + + const failures = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .filter(Boolean); + const active = sessionFailureFromResponse(failed); + expect(failures).toHaveLength(1); + expect(failures[0]).toEqual( + expect.objectContaining({ id: active.id, revision: 2, phase: "cleared" }), + ); + }); + + it("logs loudly when a failure has no active turn to reject", async () => { + const errors: string[] = []; + const agent = new ClaudeAcpAgent({ sessionUpdate: async () => {} } as unknown as AcpClient, { + log: () => {}, + error: (message) => errors.push(String(message)), + }); + injectSession(agent, [ + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "first failure", + }), + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: true, + result: "late duplicate failure", + }), + ]); + + const prompt = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + const consumer = agent.sessions["test-session"]?.consumer; + await expect(prompt).rejects.toThrow("first failure"); + await consumer; + + expect(errors).toContainEqual(expect.stringContaining("no unsettled active turn exists")); + expect(errors).toContainEqual(expect.stringContaining("late duplicate failure")); + }); + + it("retains a replayed worker shutdown through later buffered activity", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + injectSession(agent, [ + { + type: "system", + subtype: "worker_shutting_down", + reason: "historical_host_exit", + uuid: randomUUID(), + session_id: "test-session", + }, + createResultMessage({ + subtype: "success", + stop_reason: "end_turn", + is_error: false, + }), + ]); + + const prompt = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + const consumer = agent.sessions["test-session"]?.consumer; + const response = await prompt; + await consumer; + + expect(sessionFailureFromResponse(response)).toBeUndefined(); + const failure = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .find((candidate) => candidate?.category === "worker_shutdown"); + expect(failure).toEqual(expect.objectContaining({ phase: "active", revision: 1 })); + }); + + it("returns an active-turn worker shutdown on PromptResponse only at live tail", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + let reachedQuietTail!: () => void; + const quietTail = new Promise((resolve) => (reachedQuietTail = resolve)); + let closeStream!: () => void; + const streamClose = new Promise((resolve) => (closeStream = resolve)); + const input = new Pushable(); + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const user = await iter.next(); + yield { + type: "user", + message: user.value.message, + parent_tool_use_id: null, + uuid: user.value.uuid, + session_id: "test-session", + isReplay: true, + }; + yield { + type: "system", + subtype: "worker_shutting_down", + reason: "host_exit", + uuid: randomUUID(), + session_id: "test-session", + }; + reachedQuietTail(); + await streamClose; + } + agent.sessions["test-session"] = mockSessionState({ + query: wrapQuery(messageGenerator()), + input, + }); + + const prompt = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + await quietTail; + expect(JSON.stringify(updates)).not.toContain("worker_shutdown"); + closeStream(); + const response = await prompt; + const failure = sessionFailureFromResponse(response); + expect(failure).toEqual( + expect.objectContaining({ category: "worker_shutdown", phase: "active", revision: 1 }), + ); + expect(failure).toHaveProperty("turnId"); + expect(JSON.stringify(updates)).not.toContain("sessionFailure"); + }); + + it("publishes an idle worker shutdown as a session update only", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + injectSession(agent, [ + createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }), + { + type: "system", + subtype: "worker_shutting_down", + reason: "host_exit", + uuid: randomUUID(), + session_id: "test-session", + }, + ]); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + await agent.sessions["test-session"]?.consumer; + + expect(sessionFailureFromResponse(response)).toBeUndefined(); + const failure = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .find((candidate) => candidate?.category === "worker_shutdown"); + expect(failure).toEqual(expect.objectContaining({ phase: "active", revision: 1 })); + expect(failure).not.toHaveProperty("turnId"); + }); + + it("ignores a late worker shutdown from a replaced query consumer", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + let reachedQuietTail!: () => void; + const quietTail = new Promise((resolve) => (reachedQuietTail = resolve)); + let closeOldStream!: () => void; + const oldStreamClose = new Promise((resolve) => (closeOldStream = resolve)); + const input = new Pushable(); + async function* oldMessageGenerator() { + const iter = input[Symbol.asyncIterator](); + const user = await iter.next(); + yield { + type: "user", + message: user.value.message, + parent_tool_use_id: null, + uuid: user.value.uuid, + session_id: "test-session", + isReplay: true, + }; + yield createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }); + yield { + type: "system", + subtype: "worker_shutting_down", + reason: "old_host_exit", + uuid: randomUUID(), + session_id: "test-session", + }; + reachedQuietTail(); + await oldStreamClose; + } + const oldSession = mockSessionState({ query: wrapQuery(oldMessageGenerator()), input }); + agent.sessions["test-session"] = oldSession; + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + const oldConsumer = oldSession.consumer; + await quietTail; + agent.sessions["test-session"] = mockSessionState(); + closeOldStream(); + await oldConsumer; + + expect(response).toEqual(expect.objectContaining({ stopReason: "end_turn" })); + expect(JSON.stringify(updates)).not.toContain("worker_shutdown"); + }); + + it("uses a fresh session-scoped failure identity for each query-stream consumer", async () => { + const runWorkerTail = async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + injectSession(agent, [ + createResultMessage({ subtype: "success", stop_reason: "end_turn", is_error: false }), + { + type: "system", + subtype: "worker_shutting_down", + reason: "host_exit", + uuid: randomUUID(), + session_id: "test-session", + }, + ]); + const prompt = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + const consumer = agent.sessions["test-session"]?.consumer; + await prompt; + await consumer; + return updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .find((candidate) => candidate?.category === "worker_shutdown"); + }; + + const first = await runWorkerTail(); + const restarted = await runWorkerTail(); + expect(first).toEqual(expect.objectContaining({ revision: 1, phase: "active" })); + expect(restarted).toEqual(expect.objectContaining({ revision: 1, phase: "active" })); + expect(restarted.id).not.toBe(first.id); + }); + + it("publishes and returns a sanitized transport failure when the query iterator throws", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = airSessionFailureCapabilities; + const input = new Pushable(); + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const user = await iter.next(); + yield { + type: "user", + message: user.value.message, + parent_tool_use_id: null, + uuid: user.value.uuid, + session_id: "test-session", + isReplay: true, + }; + throw new Error("private transport URL and token"); + } + agent.sessions["test-session"] = mockSessionState({ + query: wrapQuery(messageGenerator()), + input, + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "test" }], + }); + expect(sessionFailureFromResponse(response)).toEqual( + expect.objectContaining({ + category: "transport_lost", + retryable: false, + actions: ["new_session"], + }), + ); + expect(JSON.stringify(response)).not.toContain("private transport URL and token"); + expect(JSON.stringify(updates)).not.toContain("sessionFailure"); + }); + it("forwards SDKAssistantMessage.error as structured data on internal errors", async () => { const agent = createMockAgent(); const assistantMessage: SDKAssistantMessage = { @@ -4404,6 +5229,38 @@ describe("logout", () => { }); expect(response.agentCapabilities?.auth?.logout).toEqual({}); }); + + it("advertises canonical AIR sessionFailure support during initialize", async () => { + const agent = createMockAgent(); + const response = await agent.initialize({ + protocolVersion: 1, + clientCapabilities: { + _meta: { + jetbrains: { + air: { version: 1, capabilities: ["sessionFailure"] }, + }, + }, + }, + }); + + expect((response._meta as any)?.jetbrains?.air).toEqual({ + version: 1, + capabilities: ["sessionFailure"], + }); + }); + + it("advertises AIR sessionFailure support even before client negotiation", async () => { + const agent = createMockAgent(); + const response = await agent.initialize({ + protocolVersion: 1, + clientCapabilities: {}, + }); + + expect((response._meta as any)?.jetbrains?.air).toEqual({ + version: 1, + capabilities: ["sessionFailure"], + }); + }); }); describe("session/close", () => { @@ -9639,10 +10496,21 @@ describe("deferred settlement for live background subagents (issues #864/#866)", await agent.sessions["test-session"]?.consumer; }); - it("resolves a held turn with its recorded outcome when the stream dies", async () => { + it("resolves a held turn and reports the dead session when the transport dies", async () => { // The held turn's answer already streamed; a stream death during the // post-answer hold is a background failure, not the turn's. - const agent = createMockAgent(); + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = { + _meta: { jetbrains: { air: { version: 1, capabilities: ["sessionFailure"] } } }, + }; injectGeneratorSession(agent, (input) => { async function* messageGenerator() { @@ -9664,6 +10532,84 @@ describe("deferred settlement for live background subagents (issues #864/#866)", }); expect(response.stopReason).toBe("end_turn"); await agent.sessions["test-session"]?.consumer; + const failure = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .find((candidate) => candidate?.category === "transport_lost"); + expect(failure).toEqual( + expect.objectContaining({ + phase: "active", + retryable: false, + actions: ["new_session"], + }), + ); + expect(failure).not.toHaveProperty("turnId"); + expect((response._meta as any)?.jetbrains?.air?.sessionFailure).toBeUndefined(); + await expect( + agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "continue" }], + }), + ).rejects.toThrow("session has ended"); + }); + + it("resolves a held turn and reports the dead session on worker-shutdown EOF", async () => { + const updates: SessionNotification[] = []; + const agent = new ClaudeAcpAgent( + { + sessionUpdate: async (update: SessionNotification) => { + updates.push(update); + }, + } as unknown as AcpClient, + { log: () => {}, error: () => {} }, + ); + (agent as any).clientCapabilities = { + _meta: { jetbrains: { air: { version: 1, capabilities: ["sessionFailure"] } } }, + }; + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const { value: userMessage } = await iter.next(); + yield userEcho(userMessage); + yield running(); + yield subagentStarted("agent-1"); + yield resultMessage(); // held + yield idle(); + yield { + type: "system", + subtype: "worker_shutting_down", + reason: "host_exit", + uuid: randomUUID(), + session_id: "test-session", + }; + } + return messageGenerator(); + }); + + const response = await agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "explore" }], + }); + expect(response.stopReason).toBe("end_turn"); + await agent.sessions["test-session"]?.consumer; + const failure = updates + .map((update) => (update.update._meta as any)?.jetbrains?.air?.sessionFailure) + .find((candidate) => candidate?.category === "worker_shutdown"); + expect(failure).toEqual( + expect.objectContaining({ + phase: "active", + retryable: false, + actions: ["new_session"], + }), + ); + expect(failure).not.toHaveProperty("turnId"); + expect((response._meta as any)?.jetbrains?.air?.sessionFailure).toBeUndefined(); + await expect( + agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "continue" }], + }), + ).rejects.toThrow("session has ended"); }); it("absorbs the interrupt's lagged trailer after cancelling a held turn mid-followup", async () => {