import type {

ServerNotification

} from "./app-server";

import type {

SessionFailure,

SessionFailureAction,

SessionFailureCategory,

SessionState,

} from "./CodexAcpServer";

import {type Error as AcpError, type PlanEntry, RequestError} from "@agentclientprotocol/sdk/unstable";

import {ACPSessionConnection, type AcpClientConnection, type UpdateSessionEvent} from "./ACPSessionConnection";

import {normalizeSessionTitle} from "./SessionTitle";

import type {

AccountRateLimitsUpdatedNotification,

AccountUpdatedNotification,

AgentMessageDeltaNotification,

CodexErrorInfo,

ConfigWarningNotification,

DeprecationNoticeNotification,

ErrorNotification,

ItemCompletedNotification,

ItemStartedNotification,

ThreadItem,

ModelReroutedNotification,

ReasoningSummaryPartAddedNotification,

ReasoningSummaryTextDeltaNotification,

ReasoningTextDeltaNotification,

ThreadGoalClearedNotification,

ThreadGoalUpdatedNotification,

ThreadTokenUsageUpdatedNotification,

Turn,

TurnDiffUpdatedNotification,

TurnPlanUpdatedNotification,

WarningNotification

} from "./app-server/v2";

import {toTokenCount} from "./TokenCount";

import { stripShellPrefix } from "./CommandUtils";

import {AcpToolCallRenderer} from "./tool-calls/AcpToolCallRenderer";

import {commandUsesTerminal} from "./tool-calls/reporters/CommandReporter";

import {CompactionReporter} from "./tool-calls/reporters/CompactionReporter";

import {DynamicToolReporter} from "./tool-calls/reporters/DynamicToolReporter";

import {FileChangeReporter} from "./tool-calls/reporters/FileChangeReporter";

import {FileCitationStream, renderFileCitations} from "./CodexFileCitations";

import {CodexItemStreamState} from "./CodexItemStreamState";

import {PLAIN_TOOL_CALL_ITEM_TYPES} from "./CodexSessionToolCalls";

import {ImageGenerationReporter} from "./tool-calls/reporters/ImageGenerationReporter";

import {ImageViewReporter} from "./tool-calls/reporters/ImageViewReporter";

import {McpToolReporter} from "./tool-calls/reporters/McpToolReporter";

import {WebSearchReporter} from "./tool-calls/reporters/WebSearchReporter";

import {CodexPlanStream, structuredPlanUpdate} from "./CodexPlanStream";

import {isWorkspaceRoutingUnavailableMessage} from "./CodexThreadErrors";

import {

createMessagePhaseMeta,

createAgentTextMessageChunk,

createAgentTextThoughtChunk,

} from "./ContentChunks";

import {goalSessionInfoUpdate, sameThreadGoalSnapshot, toThreadGoalSnapshot} from "./ThreadGoalSnapshot";

import {logger} from "./Logger";

import {recordOrNull} from "./permissions/json";

import {

AIR_EXTENSION_VERSION,

AIR_EXTENSION_VERSION_KEY,

AIR_META_KEY,

AIR_SESSION_FAILURE_KEY,

JETBRAINS_META_KEY,

withAirMetadata,

} from "./AirExtension";

import type {SubagentEventRouter} from "./subagents/SubagentEventRouter";

import {StandardSubagentEventRouter} from "./subagents/StandardSubagentEventRouter";

import type {SubagentState} from "./subagents/AcpSubagents";

import {mergeRateLimitSnapshot} from "./RateLimitsMap";

import {AGENT_FILE_CHANGE_REPORT_MAX_DIFF_BYTES} from "./AgentFileChangeReport";

import {createSessionNotice} from "./SessionNotice";

import {isStaleAutomaticTitleEcho} from "./SessionIndexTitles";

import {readableServiceErrorMessage} from "./ServiceErrorMessage";

export { stripShellPrefix };

export type CompletedPlan = {

itemId: string;

text: string;

};

type CodexFailureKind =

| "transport_lost" | "auth_required" | "rate_limited" | "quota_exhausted" | "overloaded"

| "context_exhausted" | "budget_exhausted" | "policy_denied" | "bad_request"

| "provider_error" | "internal_error";

type SessionFailurePolicy = {

category: SessionFailureCategory;

actions: SessionFailureAction[];

};

const MAX_SESSION_FAILURE_TITLE_LENGTH = 240;

const SESSION_FAILURE_POLICY: Record<CodexFailureKind, SessionFailurePolicy> = {

transport_lost: {

category: "connection",

actions: ["retry", "new_session"],

},

auth_required: {

category: "access",

actions: ["login"],

},

rate_limited: {

category: "limit",

actions: ["retry"],

},

quota_exhausted: {

category: "limit",

actions: [],

},

overloaded: {

category: "service",

actions: ["retry"],

},

context_exhausted: {

category: "limit",

actions: ["new_session"],

},

budget_exhausted: {

category: "limit",

actions: ["new_session"],

},

policy_denied: {

category: "request",

actions: [],

},

bad_request: {

category: "request", actions: [],

},

provider_error: {

category: "service",

actions: ["retry"],

},

internal_error: {

category: "service",

actions: ["retry", "new_session"],

},

};

const SYNTHETIC_FAILURE_TITLE: Record<"transport_lost" | "internal_error", string> = {

transport_lost: "Connection to Codex was lost.",

internal_error: "Codex encountered an internal error.",

};

/**

* Records sharing an id form one logical banner whose revisions must increase; a new id restarts at 1.

*/

function nextSessionFailureRevision(previous: SessionFailure | undefined, id: string): number {

return previous?.id === id ? previous.revision + 1 : 1;

}

type StringCodexErrorInfo = Extract<CodexErrorInfo, string>;

type StructuredCodexErrorInfo = Exclude<CodexErrorInfo, string>;

type KeysOfUnion<T> = T extends unknown ? keyof T : never;

type StructuredCodexErrorKind = KeysOfUnion<StructuredCodexErrorInfo>;

/**

* Known error codes map to ACP categories; extensible upstream codes fall back to provider_error.

*/

const STRING_CODEX_ERROR_CATEGORIES: Partial<Record<StringCodexErrorInfo, CodexFailureKind>> = {

contextWindowExceeded: "context_exhausted",

sessionBudgetExceeded: "budget_exhausted",

usageLimitExceeded: "quota_exhausted",

rateLimitExceeded: "rate_limited",

flexUnavailable: "provider_error",

serverOverloaded: "overloaded",

cyberPolicy: "policy_denied",

misalignmentPolicyViolation: "policy_denied",

tooManyDenials: "provider_error",

internalServerError: "internal_error",

unauthorized: "auth_required",

badRequest: "bad_request",

threadRollbackFailed: "provider_error",

sandboxError: "provider_error",

other: "provider_error",

} satisfies Record<StringCodexErrorInfo, CodexFailureKind>;

const STRUCTURED_CODEX_ERROR_CATEGORIES: Partial<Record<StructuredCodexErrorKind, CodexFailureKind>> = {

httpConnectionFailed: "transport_lost",

responseStreamConnectionFailed: "transport_lost",

responseStreamDisconnected: "transport_lost",

responseTooManyFailedAttempts: "transport_lost",

activeTurnNotSteerable: "provider_error",

} satisfies Record<StructuredCodexErrorKind, CodexFailureKind>;

/** Prompt failures whose text the client has already received as an agent message. */

const failuresShownAsMessages = new WeakSet<RequestError>();

/** Whether a prompt failure's text has already been sent to the client as an agent message. */

export function failureWasShownAsMessage(error: unknown): boolean {

return error instanceof RequestError && failuresShownAsMessages.has(error);

}

export class CodexEventHandler {

private readonly sessionState: SessionState;

private readonly supportsTypedSessionFailures: boolean;

private readonly sessionFailureEpoch: string;

private readonly pendingErrors: ErrorNotification[] = [];

private readonly failuresById = new Map<string, SessionFailure>();

private readonly activeFailureIdByScope = new Map<string, string>();

private readonly failureTurnIdById = new Map<string, string | undefined>();

private readonly allocatedFailureScopes = new Set<string>();

private lastSessionNotice: {key: string; failure: SessionFailure} | undefined;

private nextNoticeId = 1;

private failure: RequestError | null = null;

private terminalTurnError: {turnId: string; error: ErrorNotification["error"]} | null = null;

private completedPlan: CompletedPlan | null = null;

private readonly session: ACPSessionConnection;

private readonly renderer: AcpToolCallRenderer;

private readonly itemStreams = new CodexItemStreamState();

private readonly plans: CodexPlanStream;

private disposed = false;

private readonly turnDiffs = new Map<string, string>();

private readonly oversizedTurnDiffs = new Set<string>();

/** Latest diff of each subagent turn, keyed by child thread and turn; `null` marks an oversized diff. */

private readonly childTurnDiffs = new Map<string, {rootTurnId: string; diff: string | null}>();

/** The last root turn that ended, for subagent turns that outlive it. */

private latestRootTurnId: string | null = null;

private readonly collectTurnDiffs: boolean;

private readonly subagents: SubagentEventRouter;

/** Connection-level `authStatus` sink; the app-server account push feeds it. */

private readonly onAccountUpdated: ((notification: AccountUpdatedNotification) => void) | undefined;

constructor(

connection: AcpClientConnection,

sessionState: SessionState,

supportsTypedSessionFailures: boolean,

sessionFailureEpoch: string,

subagents: SubagentEventRouter,

onAccountUpdated: ((notification: AccountUpdatedNotification) => void) | undefined,

collectTurnDiffs: boolean,

private readonly supportsCompaction: boolean,

private readonly supportsNotices: boolean,

) {

this.onAccountUpdated = onAccountUpdated;

this.sessionState = sessionState;

this.supportsTypedSessionFailures = supportsTypedSessionFailures;

this.sessionFailureEpoch = sessionFailureEpoch;

this.session = new ACPSessionConnection(connection, sessionState.sessionId);

this.renderer = new AcpToolCallRenderer(sessionState.clientCapabilities);

this.plans = new CodexPlanStream(this.session, sessionState.clientCapabilities);

this.subagents = subagents;

this.collectTurnDiffs = collectTurnDiffs;

if (sessionState.sessionFailure !== undefined) {

this.failuresById.set(sessionState.sessionFailure.id, sessionState.sessionFailure);

}

}

getFailure(): RequestError | null {

return this.failure;

}

/** Runtime outcomes, not AIR capability negotiation, determine standard v2 failures. */

getTurnFailure(turn: Turn): AcpError | null {

if (turn.status !== "failed") return null;

const error = turn.error ?? (this.terminalTurnError?.turnId === turn.id ? this.terminalTurnError.error : null);

if (this.isAuthenticationRequiredError(error?.codexErrorInfo ?? null)) {

return {code: -32000, message: "Authentication required"};

}

return {code: -32603, message: readableServiceErrorMessage(error?.message ?? "Codex turn failed.")};

}

private recordTerminalTurnError(notification: ServerNotification): void {

if (notification.method === "error" && notification.params.threadId === this.sessionState.sessionId

&& !notification.params.willRetry) {

this.terminalTurnError = {turnId: notification.params.turnId, error: notification.params.error};

}

}

/** The root turn's diff followed by the diffs of the subagent turns that ran during it. */

getTurnDiff(turnId: string): string {

const diffs = [this.turnDiffs.get(turnId) ?? ""];

for (const child of this.childTurnDiffs.values()) {

if (child.rootTurnId === turnId && child.diff !== null) diffs.push(child.diff);

}

return diffs

.filter(diff => diff.trim() !== "")

.reduce((all, diff) => all === "" || all.endsWith("\n") ? all + diff : `${all}\n${diff}`, "");

}

isTurnDiffOversized(turnId: string): boolean {

return this.oversizedTurnDiffs.has(turnId)

|| [...this.childTurnDiffs.values()].some(child => child.rootTurnId === turnId && child.diff === null);

}

getTerminalSessionFailureMeta(

turnId: string | null,

allowUnattributed = false,

): Record<string, unknown> | null {

const failure = this.sessionState.sessionFailure;

if (!this.supportsTypedSessionFailures

|| failure === undefined

|| (this.failureTurnIdById.get(failure.id) === undefined

? !allowUnattributed

: turnId === null || this.failureTurnIdById.get(failure.id) !== turnId)) {

return null;

}

return this.createSessionFailureMeta(failure);

}

recordSyntheticTerminalFailure(kind: "transport_lost" | "internal_error", turnId: string | null): void {

this.recordSessionFailure(kind, turnId ?? undefined, "error", SYNTHETIC_FAILURE_TITLE[kind]);

}

/**

* Handles notifications after the prompt-local handler has been disposed. The app-server subscription

* remains installed until the ACP session closes, so terminal errors need a durable session-level path

* instead of entering a turn buffer that will never be flushed.

*/

async handleSessionScopedNotification(notification: ServerNotification): Promise<void> {

const threadId = (notification.params as {threadId?: unknown}).threadId;

if (this.subagents instanceof StandardSubagentEventRouter && typeof threadId === "string"

&& threadId !== this.sessionState.sessionId) {

await this.handleNotification(notification);

return;

}

this.recordTerminalTurnError(notification);

if (notification.method !== "error") {

await this.handleNotification(notification);

return;

}

if (!this.supportsTypedSessionFailures) {

// Preserve the legacy behavior for clients that did not negotiate typed failures.

await this.handleNotification(notification);

return;

}

await this.finishCompactionsForNotification(notification);

if (this.isAuthenticationRequiredError(notification.params.error.codexErrorInfo)) return;

if (notification.params.willRetry) {

await this.publishUpdate(this.createSessionFailureUpdate(this.recordRetryWarning(notification.params, false)));

return;

}

const failure = this.recordSessionFailure(

this.sessionFailureKind(notification.params.error),

notification.params.turnId,

"error",

notification.params.error.message,

undefined,

false,

);

await this.publishUpdate(this.createSessionFailureUpdate(failure));

}

async flushPendingErrors(): Promise<void> {

if (this.sessionState.currentTurnId === null || this.pendingErrors.length === 0) {

return;

}

const errors = this.pendingErrors.splice(0);

for (const error of errors) {

const update = await this.createErrorEvent(error);

if (update) {

await this.publishUpdate(update);

}

}

}

async flushPendingErrorsAsSessionScoped(): Promise<void> {

if (!this.supportsTypedSessionFailures || this.pendingErrors.length === 0) {

return;

}

const errors = this.pendingErrors.splice(0);

for (const error of errors) {

await this.handleSessionScopedNotification({method: "error", params: error});

}

}

async clearSessionFailure(): Promise<void> {

const failure = this.sessionState.sessionFailure;

delete this.sessionState.sessionFailure;

if (!failure || this.session.protocolVersion !== 2) return;

await this.session.patchSessionMetadata(metadata => {

const jetbrains = recordOrNull(metadata[JETBRAINS_META_KEY]);

const air = recordOrNull(jetbrains?.[AIR_META_KEY]);

const published = recordOrNull(air?.[AIR_SESSION_FAILURE_KEY]);

if (!air || published?.["id"] !== failure.id || published?.["revision"] !== failure.revision) return null;

const {[AIR_SESSION_FAILURE_KEY]: _failure, ...currentAir} = air;

return {[JETBRAINS_META_KEY]: {...jetbrains, [AIR_META_KEY]: currentAir}};

});

}

async completeSuccessfulTurn(turnId: string | null): Promise<void> {

this.lastSessionNotice = undefined;

if (!this.supportsTypedSessionFailures || turnId === null) return;

const active = this.sessionState.sessionFailure;

if (active?.id !== this.activeFailureIdByScope.get(turnId) || active?.severity !== "warning") return;

this.activeFailureIdByScope.delete(turnId);

await this.clearSessionFailure();

}

async handleFailedTurn(turn: Turn): Promise<void> {

if (turn.status === "failed" && this.isAuthenticationRequiredError(turn.error?.codexErrorInfo ?? null)) {

this.failure = RequestError.authRequired();

return;

}

const activeFailure = this.sessionState.sessionFailure;

if (!this.supportsTypedSessionFailures

|| turn.status !== "failed"

|| this.failure !== null

|| this.failureTurnIdById.get(activeFailure?.id ?? "") === turn.id && activeFailure?.severity === "error") {

return;

}

const error = turn.error ?? {

message: "Turn failed",

codexErrorInfo: null,

additionalDetails: null,

misalignment: null,

};

this.recordTypedSessionFailure({

threadId: this.sessionState.sessionId,

turnId: turn.id,

willRetry: false,

error,

});

}

takeCompletedPlan(): CompletedPlan | null {

const plan = this.completedPlan;

this.completedPlan = null;

return plan;

}

async handleNotification(notification: ServerNotification) {

const threadId = (notification.params as {threadId?: unknown}).threadId;

if (this.subagents instanceof StandardSubagentEventRouter && this.subagents.isRetired()

&& typeof threadId === "string" && threadId !== this.sessionState.sessionId) return;

this.recordTerminalTurnError(notification);

if (notification.method === "item/started"

&& notification.params.threadId === this.sessionState.sessionId

&& isModelOutputItem(notification.params.item)) {

this.sessionState.promptTokenUsage?.observeModelOutput();

}

// Before subagent routing, which drops updates of ended children and of spawns that never materialized.

if (notification.method === "turn/diff/updated" && notification.params.threadId !== this.sessionState.sessionId) {

this.recordChildTurnDiff(notification.params);

}

if (this.subagents instanceof StandardSubagentEventRouter && !await this.subagents.prepareNotification(notification)) {

for (const event of this.subagents.takeBufferedNotifications()) await this.handleNotification(event);

return;

}

await this.flushPendingErrors();

await this.finishCompactionsForNotification(notification);

const closingChildren = this.subagents.closingChildSessions(notification);

for (const child of closingChildren) {

if (this.subagents instanceof StandardSubagentEventRouter) {

const turnId = notification.method === "turn/completed" ? notification.params.turn.id : undefined;

for (const update of this.subagents.openToolCalls(child.threadId)?.finishOutstanding(turnId) ?? []) {

await this.session.update(update, child.sessionId);

}

}

await this.finishOutstandingCompactions(

child.state === "cancelled" ? "cancelled" : "failed",

child.sessionId,

);

await this.sessionState.asyncTasks.reconcile(child.threadId, child.sessionId);

}

const handledBySubagents = await this.subagents.handle(notification);

for (const buffered of this.subagents.takeBufferedNotifications()) {

await this.handleNotification(buffered);

}

const ignoredBySubagents = !handledBySubagents && this.subagents.shouldIgnore(notification);

let updateEvent: UpdateSessionEvent | null | undefined;

if (!handledBySubagents

&& !ignoredBySubagents

&& notification.method === "item/started"

&& notification.params.item.type === "commandExecution") {

updateEvent = await this.createUpdateEvent(notification);

}

if (!handledBySubagents) {

await this.sessionState.asyncTasks.handleNotification(

notification,

this.subagents.notificationSessionId(notification),

toolCallTitle(updateEvent),

);

}

if (handledBySubagents) return;

if (ignoredBySubagents) return;

if (updateEvent === undefined) updateEvent = await this.createUpdateEvent(notification);

if (updateEvent) {

await this.publishUpdate(updateEvent, this.subagents.notificationSessionId(notification));

}

}

async waitForNativeSubagentSession(childThreadId: string): Promise<string | null> {

return await this.subagents.waitForMaterializedSession(childThreadId);

}

async waitForNativeSubagents(signal: AbortSignal): Promise<void> {

if (await this.subagents.wait(signal) === "timed_out") {

await this.finishOutstandingNativeSubagents("failed");

}

}

async finishOutstandingNativeSubagents(state: SubagentState): Promise<void> {

await this.finishOutstandingCompactions(state === "cancelled" ? "cancelled" : "failed");

await this.subagents.finishOutstanding(state);

}

async finishOutstandingCompactions(status: "failed" | "cancelled", sessionId?: string): Promise<void> {

if (!this.supportsCompaction) return;

for (const {sessionId: targetSessionId, update} of this.sessionState.compactions.finishOutstanding(status, sessionId)) {

await this.session.update(update, targetSessionId);

}

}

private async finishCompactionsForNotification(notification: ServerNotification): Promise<void> {

if (!this.supportsCompaction) return;

let updates: UpdateSessionEvent[];

const sessionId = this.subagents.notificationSessionId(notification);

if (notification.method === "turn/completed") {

const turn = notification.params.turn;

if (turn.status === "inProgress") return;

updates = this.sessionState.compactions.finishTurn(

sessionId,

turn.id,

turn.status === "interrupted" ? "cancelled" : "failed",

turn.error?.message ?? "Codex ended the turn before compaction completed.",

);

} else if (notification.method === "error" && !notification.params.willRetry) {

updates = this.sessionState.compactions.finishTurn(

sessionId,

notification.params.turnId,

"failed",

notification.params.error.message,

);

} else {

return;

}

for (const update of updates) await this.session.update(update, sessionId);

}

async flushPendingPlanUpdates(): Promise<void> {

await this.plans.flush();

}

async dispose(): Promise<void> {

if (this.disposed) return;

await this.plans.dispose();

if (this.pendingErrors.length > 0) {

logger.log("Discarding app-server errors that arrived before a turn started", {

sessionId: this.sessionState.sessionId,

count: this.pendingErrors.length,

turnIds: this.pendingErrors.map(error => error.turnId),

});

this.pendingErrors.splice(0);

}

this.disposed = true;

this.turnDiffs.clear();

this.oversizedTurnDiffs.clear();

this.childTurnDiffs.clear();

}

/** Subagent threads report their own turn diffs; the root turn's file-change report must include them. */

private recordChildTurnDiff(params: TurnDiffUpdatedNotification): void {

if (this.disposed || !this.collectTurnDiffs) return;

const key = `${params.threadId}\n${params.turnId}`;

// A child turn stays with the root turn it started under, even if its diff updates arrive later.

const rootTurnId = this.childTurnDiffs.get(key)?.rootTurnId

?? this.sessionState.currentTurnId

?? this.latestRootTurnId;

if (rootTurnId === null) return;

const oversized = Buffer.byteLength(params.diff, "utf8") > AGENT_FILE_CHANGE_REPORT_MAX_DIFF_BYTES;

this.childTurnDiffs.set(key, {rootTurnId, diff: oversized ? null : params.diff});

}

private async createUpdateEvent(notification: ServerNotification): Promise<UpdateSessionEvent | null> {

/*

TODO split UpdateSessionEvent to improve completion

createUpdateEvent({

sessionUpdate: "" , <- completion of UpdateSessionEvent["sessionUpdate"]

params: {}, <- quickfix to generate required fields (rest of)

});

*/

switch (notification.method) {

case "item/agentMessage/delta":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

return await this.createTextEvent(notification.params);

case "item/plan/delta":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

return this.plans.delta(notification.params.itemId, notification.params.delta);

case "item/started": {

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

const update = await this.createItemEvent(notification.params);

this.trackOpenToolCall(notification.params.item, notification.params.threadId, notification.params.turnId);

return update;

}

case "item/completed":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

this.openToolCalls(notification.params.threadId).complete(notification.params.item.id);

return await this.completeItemEvent(notification.params);

case "turn/plan/updated":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

return await this.updatePlan(notification.params);

case "turn/diff/updated":

if (notification.params.threadId === this.sessionState.sessionId) {

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

if (!this.disposed && this.collectTurnDiffs) {

if (Buffer.byteLength(notification.params.diff, "utf8") > AGENT_FILE_CHANGE_REPORT_MAX_DIFF_BYTES) {

this.turnDiffs.delete(notification.params.turnId);

this.oversizedTurnDiffs.add(notification.params.turnId);

} else {

// Codex 0.154 emits an empty snapshot when its tracker transitions

// from a non-empty aggregate to no diff, which clears stale state here.

this.oversizedTurnDiffs.delete(notification.params.turnId);

this.turnDiffs.set(notification.params.turnId, notification.params.diff);

}

}

}

return null;

case "error":

return await this.createErrorEvent(notification.params);

case "turn/started":

this.sessionState.interruptTurnId = notification.params.turn.id;

// A review's child turn starts under its own id, while the review's items, errors and

// completion keep the parent id already set from the `review/start` response.

this.sessionState.currentTurnId ??= notification.params.turn.id;

await this.flushPendingErrors();

return null;

case "turn/completed":

await this.plans.flush();

this.plans.clearTurn();

// Subagents can keep running after the root turn ends; their turns still belong to it.

this.latestRootTurnId = this.sessionState.currentTurnId ?? this.latestRootTurnId;

this.sessionState.currentTurnId = null;

this.sessionState.interruptTurnId = null;

if (this.session.protocolVersion === 2 && notification.params.turn.status === "failed") {

for (const update of this.sessionState.openToolCalls.finishOutstanding()) {

await this.session.update(update, this.subagents.notificationSessionId(notification));

}

}

this.sessionState.toolCallReports.releaseOpen(this.subagents.notificationSessionId(notification));

// An item that missed its item/completed (e.g. dropped by a provider restart

// mid-turn) must not linger past its own turn and get failed by a later restart.

this.sessionState.openToolCalls.clear();

return null;

case "thread/tokenUsage/updated":

return this.createUsageUpdate(notification.params);

case "thread/name/updated":

if (isStaleAutomaticTitleEcho(this.sessionState, normalizeSessionTitle(notification.params.threadName))) {

this.sessionState.titleGen?.observeRename();

return null;

}

this.sessionState.sessionTitle = normalizeSessionTitle(notification.params.threadName);

this.sessionState.sessionTitleSource = notification.params.threadName == null

? "unset"

: "explicit";

this.sessionState.titleGen?.observeRename();

return {

sessionUpdate: "session_info_update",

title: this.sessionState.sessionTitle,

};

case "thread/status/changed":

return this.createCodexSessionInfoUpdate({

threadStatus: notification.params.status,

});

case "thread/archived":

return this.createCodexSessionInfoUpdate({

archived: true,

});

case "thread/unarchived":

return this.createCodexSessionInfoUpdate({

archived: false,

});

case "thread/closed":

return this.createCodexSessionInfoUpdate({

closed: true,

});

case "item/commandExecution/outputDelta":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

return this.renderer.render(this.streamsFor(notification.params.threadId).commands.outputDelta(notification.params.itemId, notification.params.delta));

case "item/mcpToolCall/progress":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

// AIR does not show MCP progress.

if (this.renderer.capabilities.airToolCallContract) return null;

return this.renderer.render(McpToolReporter.progress(

notification.params.itemId,

notification.params.message,

));

case "account/rateLimits/updated":

this.handleRateLimitsUpdated(notification.params);

return null;

case "account/updated":

this.onAccountUpdated?.(notification.params);

return null;

case "configWarning":

return await this.createConfigWarningEvent(notification.params);

case "warning":

return this.createWarningEvent(notification.params);

case "guardianWarning":

return null;

case "deprecationNotice":

return this.createDeprecationNoticeEvent(notification.params);

case "item/autoApprovalReview/started":

return this.renderer.render(this.streamsFor(notification.params.threadId).guardianReviews.started(notification.params));

case "item/autoApprovalReview/completed":

return this.renderer.render(this.streamsFor(notification.params.threadId).guardianReviews.completed(notification.params));

case "thread/compacted":

return this.supportsCompaction

? this.sessionState.compactions.completeLegacy(

this.subagents.notificationSessionId(notification), notification.params.turnId,

)

: this.createContextCompactedEvent();

case "item/reasoning/summaryTextDelta":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

return this.createReasoningDeltaEvent(notification.params);

case "item/reasoning/textDelta":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

return this.createReasoningDeltaEvent(notification.params);

case "item/reasoning/summaryPartAdded":

this.completeRetryIncidentOnTurnProgress(notification.params.threadId);

return this.createReasoningSectionBreakEvent(notification.params);

case "model/rerouted":

return this.createModelReroutedEvent(notification.params);

case "fuzzyFileSearch/sessionUpdated":

return this.renderer.render(this.itemStreams.fuzzySearches.updated(notification.params));

case "fuzzyFileSearch/sessionCompleted":

return this.renderer.render(this.itemStreams.fuzzySearches.completed(notification.params));

case "thread/goal/updated":

return this.createThreadGoalUpdatedEvent(notification.params);

case "thread/goal/cleared":

return this.createThreadGoalClearedEvent(notification.params);

case "item/commandExecution/terminalInteraction":

return this.renderer.render(this.streamsFor(notification.params.threadId).commands.terminalInput(

notification.params.itemId,

notification.params.stdin,

));

case "thread/attachment/updated":

// Persisted attachment metadata has no ACP session update counterpart.

return null;

case "account/gatewayOAuth/changed":

// Gateway login state is account-scoped and has no ACP session update counterpart.

return null;

case "skills/changed":

// The server lists the skills again and publishes the commands. The change itself is no update.

return null;

// ignored events

// Read markers and predictions have no ACP session update counterpart.

case "thread/readState/changed":

case "thread/prediction/updated":

case "thread/deleted":

case "thread/reverted":

case "thread/queue/changed":

case "project/changed":

case "thread/project/updated":

case "thread/environment/connected":

case "thread/environment/disconnected":

case "command/exec/outputDelta":

case "hook/started":

case "hook/completed":

case "turn/moderationMetadata":

case "item/fileChange/outputDelta":

case "item/fileChange/patchUpdated":

case "fs/changed":

case "mcpServer/startupStatus/updated":

case "mcpServer/event/stream/notification":

case "serverRequest/resolved":

case "model/verification":

case "modelProvider/authRecoveryStarted":

case "modelProvider/authRecoveryCompleted":

case "model/safetyBuffering/updated":

case "windows/worldWritableWarning":

case "thread/realtime/started":

case "thread/realtime/itemAdded":

case "thread/realtime/item/started":

case "thread/realtime/item/transcript/delta":

case "thread/realtime/item/completed":

case "thread/realtime/transcript/delta":

case "thread/realtime/transcript/done":

case "thread/realtime/outputAudio/delta":

case "thread/realtime/sdp":

case "thread/realtime/error":

case "thread/realtime/closed":

case "windowsSandbox/setupCompleted":

case "account/login/completed":

case "mcpServer/oauthLogin/completed":

case "externalAgentConfig/import/completed":

case "rawResponseItem/completed":

case "rawResponse/completed":

case "thread/started":

case "remoteControl/status/changed":

case "app/list/updated":

case "thread/settings/updated":

case "externalAgentConfig/import/progress":

case "process/outputDelta":

case "process/exited":

case "autoApprovalReview/strictReviewRequired":

return null;

}

}

/** These fact builders produce partial owned namespaces, unlike raw metadata patches. */

private publishUpdate(update: UpdateSessionEvent, sessionId = this.sessionState.sessionId): Promise<void> {

if (this.session.protocolVersion !== 2 || update.sessionUpdate !== "session_info_update" || update._meta == null) {

return this.session.update(update, sessionId);

}

const {_meta, sessionUpdate: _tag, ...fields} = update;

return this.session.forSession(sessionId).patchSessionMetadata(current => {

const metadata = {..._meta};

const codex = recordOrNull(_meta["codex"]);

if (codex) metadata["codex"] = {...recordOrNull(current["codex"]), ...codex};

const air = recordOrNull(recordOrNull(_meta[JETBRAINS_META_KEY])?.[AIR_META_KEY]);

if (air) metadata[JETBRAINS_META_KEY] = withAirMetadata(current, air)[JETBRAINS_META_KEY];

return metadata;

}, fields);

}

private createCodexSessionInfoUpdate(codexMetadata: Record<string, unknown>): UpdateSessionEvent {

return {

sessionUpdate: "session_info_update",

_meta: {

codex: codexMetadata,

},

};

}

private async createTextEvent(event: AgentMessageDeltaNotification): Promise<UpdateSessionEvent | null> {

const streams = this.streamsFor(event.threadId);

let citations = streams.agentMessageCitations.get(event.itemId);

if (citations === undefined) {

citations = new FileCitationStream(this.sessionState.cwd);

streams.agentMessageCitations.set(event.itemId, citations);

}

return this.createAgentMessageText(citations.push(event.delta), event.itemId, event.threadId);

}

/**

* Sends the text a file citation held back, once the message is complete. A message that

* arrives whole, without deltas (an async question, for one), is sent in full.

*/

private flushAgentMessageText(item: ThreadItem & { type: "agentMessage" }, threadId: string): UpdateSessionEvent | null {

const streams = this.streamsFor(threadId);

const citations = streams.agentMessageCitations.get(item.id);

streams.agentMessageCitations.delete(item.id);

const text = citations ? citations.flush() : renderFileCitations(item.text, this.sessionState.cwd);

return this.createAgentMessageText(text, item.id, threadId);

}

private createAgentMessageText(text: string, itemId: string, threadId: string): UpdateSessionEvent | null {

if (text.length === 0) return null;

const phase = this.streamsFor(threadId).agentMessagePhases.get(itemId) ?? null;

const meta = createMessagePhaseMeta(phase, this.sessionState.clientCapabilities.airClient);

return createAgentTextMessageChunk(text, itemId, meta);

}

private async createConfigWarningEvent(event: ConfigWarningNotification): Promise<UpdateSessionEvent> {

if (this.supportsNotices) {

return createSessionNotice("warning", event.summary.trim() || "Configuration warning", event.details);

}

if (this.supportsTypedSessionFailures) {

return this.createSessionFailureUpdate(this.recordSessionNotice(...this.sessionNoticeContent(event.summary, event.details)));

}

const text = event.details ? `${event.summary}\n\n${event.details}` : event.summary;

return createAgentTextMessageChunk(`Config warning: ${text}\n\n`);

}

private createDeprecationNoticeEvent(event: DeprecationNoticeNotification): UpdateSessionEvent | null {

if (this.supportsNotices) {

return createSessionNotice("warning", event.summary.trim() || "Deprecated configuration", event.details);

}

// Legacy clients without typed failures have never received deprecation notices.

if (!this.supportsTypedSessionFailures) return null;

return this.createSessionFailureUpdate(

this.recordSessionNotice(...this.sessionNoticeContent(event.summary, event.details)),

);

}

private createWarningEvent(event: WarningNotification): UpdateSessionEvent {

if (this.supportsNotices) {

return createSessionNotice("warning", event.message.trim() || "Codex warning");

}

if (this.supportsTypedSessionFailures) {

return this.createSessionFailureUpdate(this.recordSessionNotice(event.message));

}

return createAgentTextMessageChunk(`Warning: ${event.message}\n\n`);

}

private createModelReroutedEvent(event: ModelReroutedNotification): UpdateSessionEvent {

if (!this.supportsNotices) {

return createAgentTextThoughtChunk(`Model rerouted from ${event.fromModel} to ${event.toModel} (${event.reason}).\n\n`);

}

return createSessionNotice(

"info",

"Model rerouted",

`Switched from ${event.fromModel} to ${event.toModel} (${event.reason}).`,

);

}

private createThreadGoalUpdatedEvent(event: ThreadGoalUpdatedNotification): UpdateSessionEvent | null {

this.sessionState.goalRevision += 1;

const goalSnapshot = toThreadGoalSnapshot(event.goal);

if (sameThreadGoalSnapshot(this.sessionState.currentGoal, goalSnapshot)) {

return null;

}

this.sessionState.currentGoal = goalSnapshot;

return goalSessionInfoUpdate(goalSnapshot, this.sessionState.clientCapabilities.airClient);

}

private createThreadGoalClearedEvent(_event: ThreadGoalClearedNotification): UpdateSessionEvent | null {

this.sessionState.goalRevision += 1;

if (this.sessionState.currentGoal === null) {

return null;

}

this.sessionState.currentGoal = null;

return goalSessionInfoUpdate(null, this.sessionState.clientCapabilities.airClient);

}

private createReasoningDeltaEvent(

event: ReasoningSummaryTextDeltaNotification | ReasoningTextDeltaNotification

): UpdateSessionEvent {

this.streamsFor(event.threadId).seenReasoningDeltaItemIds.add(event.itemId);

return this.createAgentThoughtEvent(event.delta, event.itemId);

}

private createReasoningSectionBreakEvent(event: ReasoningSummaryPartAddedNotification): UpdateSessionEvent {

this.streamsFor(event.threadId).seenReasoningDeltaItemIds.add(event.itemId);

return this.createAgentThoughtEvent("\n\n", event.itemId);

}

private createAgentThoughtEvent(text: string, messageId: string): UpdateSessionEvent {

return createAgentTextThoughtChunk(text, messageId);

}

private async createItemEvent(event: ItemStartedNotification): Promise<UpdateSessionEvent | null> {

switch (event.item.type) {

case "fileChange":

return this.renderer.render(FileChangeReporter.started(

event.item,

this.sessionState.clientCapabilities.diffPatchFormat,

));

case "commandExecution":

return this.renderer.render(this.streamsFor(event.threadId).commands.started(event.item));

case "mcpToolCall":

return this.renderer.render(McpToolReporter.started(event.item));

case "dynamicToolCall":

return this.renderer.render(DynamicToolReporter.started(event.item));

case "webSearch":

return this.renderer.render(WebSearchReporter.started(event.item));

case "imageView":

this.streamsFor(event.threadId).emittedImageViewItems.add(event.item.id);

return this.renderer.render(ImageViewReporter.viewed(event.item));

case "imageGeneration":

this.streamsFor(event.threadId).activeImageGenerationItems.add(event.item.id);

return this.renderer.render(ImageGenerationReporter.started(event.item));

case "collabAgentToolCall":

return this.renderer.render(this.subagents.legacyCollaborationStarted(event.item));

case "agentMessage":

this.rememberAgentMessagePhase(event.item, event.threadId);

return null;

case "contextCompaction":

return this.supportsCompaction

? this.sessionState.compactions.start(

this.subagents.notificationSessionId({method: "item/started", params: event}),

event.turnId, event.item.id,

)

: this.renderer.render(CompactionReporter.started(event.item));

case "subAgentActivity":

return this.renderer.render(this.subagents.legacyActivityStarted(event.item));

case "sleep":

case "functionCallOutput":

case "userMessage":

case "hookPrompt":

case "reasoning":

case "enteredReviewMode":

case "exitedReviewMode":

case "plan":

return null;

}

}

private async completeItemEvent(event: ItemCompletedNotification): Promise<UpdateSessionEvent | null> {

switch (event.item.type) {

case "fileChange":

return this.renderer.render(FileChangeReporter.completed(event.item));

case "dynamicToolCall":

return this.renderer.render(DynamicToolReporter.completed(event.item));

case "mcpToolCall":

return this.renderer.render(McpToolReporter.completed(event.item));

case "commandExecution":

return this.renderer.render(this.streamsFor(event.threadId).commands.completed(event.item, true));

case "imageView":

if (this.streamsFor(event.threadId).emittedImageViewItems.delete(event.item.id)) {

return null;

}

return this.renderer.render(ImageViewReporter.viewed(event.item));

case "imageGeneration":

return this.renderer.render(this.streamsFor(event.threadId).activeImageGenerationItems.delete(event.item.id)

? ImageGenerationReporter.completed(event.item)

: ImageGenerationReporter.whole(event.item));

case "reasoning":

if (this.streamsFor(event.threadId).seenReasoningDeltaItemIds.delete(event.item.id)) {

return null;

}

return this.createCompletedReasoningEvent(event.item);

case "webSearch":

return this.renderer.render(WebSearchReporter.completed(event.item));

case "collabAgentToolCall":