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.
This commit is contained in:
nikita-ashihmin
2026-08-10 14:15:44 +04:00
committed by GitHub
parent 3df1ede89f
commit 8157ee113e
2 changed files with 1333 additions and 17 deletions

View File

@@ -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<string, unknown> | undefined;
const air = jetbrains?.[AIR_META_KEY] as Record<string, unknown> | 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<string, number>();
const sessionFailureEpoch = randomUUID();
const activeSessionFailures = new Map<string, PublishedSessionFailure>();
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<boolean> => {
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<PublishedSessionFailure | undefined> => {
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<SDKMessage, void>;
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);
}
}

View File

@@ -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<any>();
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<any>();
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<any>();
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<any>();
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<void>((resolve) => (reachedQuietTail = resolve));
let closeStream!: () => void;
const streamClose = new Promise<void>((resolve) => (closeStream = resolve));
const input = new Pushable<any>();
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<void>((resolve) => (reachedQuietTail = resolve));
let closeOldStream!: () => void;
const oldStreamClose = new Promise<void>((resolve) => (closeOldStream = resolve));
const input = new Pushable<any>();
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<any>();
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 () => {