CodexAdapter.ts

/**
 * CodexAdapterLive - Scoped live implementation for the Codex provider adapter.
 *
 * Wraps the typed Codex session runtime behind the `CodexAdapter` service
 * contract and maps runtime failures into the shared `ProviderAdapterError`
 * algebra.
 *
 * @module CodexAdapterLive
 */
import {
  EventId,
  type CanonicalItemType,
  type CanonicalRequestType,
  type CodexSettings,
  ProviderDriverKind,
  type ProviderEvent,
  ProviderInstanceId,
  type ProviderRuntimeEvent,
  type ProviderRequestKind,
  type ProviderSession,
  type ProviderTurnStartResult,
  type ThreadTokenUsageSnapshot,
  type ToolActivityIcon,
  type ToolActivityNativeAppReference,
  type ToolActivitySource,
  type ProviderUserInputAnswers,
  RuntimeItemId,
  RuntimeRequestId,
  RuntimeTaskId,
  type RuntimeTaskUsage,
  type TurnTokenUsage,
  ProviderApprovalDecision,
  ThreadId,
  type TurnId,
  ProviderSendTurnInput,
  DESKTOP_MCP_SERVER_NAME,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as NodeCrypto from "node:crypto";
import * as Crypto from "effect/Crypto";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as FileSystem from "effect/FileSystem";
import * as Queue from "effect/Queue";
import * as Schema from "effect/Schema";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
import { ChildProcessSpawner } from "effect/unstable/process";
import * as CodexErrors from "effect-codex-app-server/errors";
import * as EffectCodexSchema from "effect-codex-app-server/schema";

import { getModelSelectionStringOptionValue, normalizeModelSlug } from "@t3tools/shared/model";
import { getCodexServiceTierOptionValue } from "../../codexModelOptions.ts";
import { makeResolveEnabledDesktopMcp } from "../../desktopControl/desktopMcpLaunch.ts";
import * as McpProviderSession from "../../mcp/McpProviderSession.ts";

import {
  ProviderAdapterRequestError,
  ProviderAdapterProcessError,
  ProviderAdapterSessionClosedError,
  ProviderAdapterSessionNotFoundError,
  ProviderAdapterValidationError,
  type ProviderAdapterError,
} from "../Errors.ts";
import { type CodexAdapterShape } from "../Services/CodexAdapter.ts";
import { resolveAttachmentPath } from "../../attachmentStore.ts";
import { ServerConfig } from "../../config.ts";
import * as ComputerHistoryService from "../../computerHistory/service.ts";
import * as ServerSettings from "../../serverSettings.ts";
import {
  CodexResumeCursorSchema,
  CodexSessionRuntimeThreadIdMissingError,
  describeMcpElicitation,
  hasConfiguredMcpServerNamed,
  makeCodexSessionRuntime,
  type CodexSessionRuntimeError,
  type CodexSessionRuntimeOptions,
  type CodexSessionRuntimeSendTurnInput,
  type CodexSessionRuntimeShape,
  type CodexThreadSnapshot,
} from "./CodexSessionRuntime.ts";
import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
import { codexLaunchArgv, resolveCodexLaunchArgs } from "./codexLaunchArgs.ts";
import { codexRateLimitsToUpdate } from "./codexUsageLimits.ts";
const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError);
const isCodexAppServerTransportError = Schema.is(CodexErrors.CodexAppServerTransportError);
const isCodexSessionRuntimeThreadIdMissingError = Schema.is(
  CodexSessionRuntimeThreadIdMissingError,
);
const isCodexResumeCursorSchema = Schema.is(CodexResumeCursorSchema);

const PROVIDER = ProviderDriverKind.make("codex");

export interface CodexAdapterLiveOptions {
  readonly instanceId?: ProviderInstanceId;
  readonly environment?: NodeJS.ProcessEnv;
  readonly makeRuntime?: (
    options: CodexSessionRuntimeOptions,
  ) => Effect.Effect<
    CodexSessionRuntimeShape,
    CodexSessionRuntimeError,
    ChildProcessSpawner.ChildProcessSpawner | Scope.Scope
  >;
  readonly nativeEventLogPath?: string;
  readonly nativeEventLogger?: EventNdjsonLogger;
}

interface CodexPendingTurn {
  readonly input: CodexSessionRuntimeSendTurnInput;
  readonly restartInput: Parameters<CodexAdapterShape["startSession"]>[0];
  readonly retryCount: number;
}

interface CodexAdapterSessionContext {
  readonly threadId: ThreadId;
  readonly scope: Scope.Closeable;
  readonly runtime: CodexSessionRuntimeShape;
  readonly eventFiber: Fiber.Fiber<void, never>;
  readonly startInput: Parameters<CodexAdapterShape["startSession"]>[0];
  readonly providerSession: ProviderSession;
  readonly pendingTurns: Map<TurnId, CodexPendingTurn>;
  readonly deferredTurnCompletions: Map<TurnId, ProviderEvent>;
  readonly unsafeTurnIds: Set<TurnId>;
  readonly recoveryToken?: symbol;
  recovering: boolean;
  readonly turnTokenUsage: CodexTurnTokenUsageState;
  stopped: boolean;
}

interface CodexCollabUsageState {
  readonly baseline: RuntimeTaskUsage;
  readonly latest: RuntimeTaskUsage;
}

function isRetryableUsageLimitCompletion(event: ProviderEvent): boolean {
  if (event.method !== "turn/completed") {
    return false;
  }
  const payload = readPayload(EffectCodexSchema.V2TurnCompletedNotification, event.payload);
  return (
    payload?.turn.status === "failed" &&
    payload.turn.error?.codexErrorInfo === "usageLimitExceeded" &&
    payload.turn.items.every((item) => item.type === "userMessage")
  );
}

function isUnsafeUsageLimitRetryEvent(event: ProviderEvent): boolean {
  if (event.turnId === undefined) {
    return false;
  }
  if (event.method === "item/started") {
    const payload = readPayload(EffectCodexSchema.V2ItemStartedNotification, event.payload);
    return payload?.item.type !== "userMessage";
  }
  if (event.method === "item/completed") {
    const payload = readPayload(EffectCodexSchema.V2ItemCompletedNotification, event.payload);
    return payload?.item.type !== "userMessage";
  }
  return (
    event.itemId !== undefined ||
    event.requestId !== undefined ||
    event.method.startsWith("item/") ||
    event.method.startsWith("rawResponseItem/") ||
    event.method.startsWith("hook/") ||
    event.method.startsWith("turn/diff/") ||
    event.method.startsWith("turn/plan/") ||
    event.method.startsWith("collabAgent/")
  );
}

function requiresFailedTurnRollback(
  snapshot: CodexThreadSnapshot,
  failedTurnId: TurnId,
): boolean | undefined {
  const lastTurn = snapshot.turns.at(-1);
  if (!lastTurn || lastTurn.id !== failedTurnId) {
    return false;
  }
  return lastTurn.items.every((item) => item.type === "userMessage") ? true : undefined;
}

function resumedRequestedModel(
  requestedModel: string | undefined,
  resumedModel: string | undefined,
): boolean {
  const requested = normalizeModelSlug(requestedModel);
  if (!requested) {
    return true;
  }
  return requested === normalizeModelSlug(resumedModel);
}

type CodexCumulativeTokenUsage = {
  readonly inputTokens: number;
  readonly cachedInputTokens: number;
  readonly cacheCreationTokens?: number;
  readonly outputTokens: number;
  readonly reasoningTokens: number;
};

interface CodexTurnTokenUsageAccumulator {
  inputTokens: number;
  cachedInputTokens: number;
  cacheCreationTokens: number | undefined;
  outputTokens: number;
  reasoningTokens: number;
  observed: boolean;
  hasSubagents: boolean;
}

interface CodexTurnTokenUsageState {
  baseline: CodexCumulativeTokenUsage | undefined;
  activeTurnId: string | undefined;
  readonly byTurnId: Map<string, CodexTurnTokenUsageAccumulator>;
}

function mapCodexRuntimeError(
  threadId: ThreadId,
  method: string,
  error: CodexSessionRuntimeError,
): ProviderAdapterError {
  if (isCodexAppServerProcessExitedError(error) || isCodexAppServerTransportError(error)) {
    return new ProviderAdapterSessionClosedError({
      provider: PROVIDER,
      threadId,
      cause: error,
    });
  }

  if (isCodexSessionRuntimeThreadIdMissingError(error)) {
    return new ProviderAdapterSessionNotFoundError({
      provider: PROVIDER,
      threadId,
      cause: error,
    });
  }

  return new ProviderAdapterRequestError({
    provider: PROVIDER,
    method,
    detail: error.message,
    cause: error,
  });
}

type CodexLifecycleItem =
  | EffectCodexSchema.V2ItemStartedNotification["item"]
  | EffectCodexSchema.V2ItemCompletedNotification["item"];

type CodexToolUserInputQuestion =
  | EffectCodexSchema.ServerRequest__ToolRequestUserInputQuestion
  | EffectCodexSchema.ToolRequestUserInputParams__ToolRequestUserInputQuestion;

const ApprovalDecisionPayload = Schema.Struct({
  decision: ProviderApprovalDecision,
});

function readPayload<A>(
  schema: Schema.Schema<A>,
  payload: ProviderEvent["payload"],
): A | undefined {
  const isPayload = Schema.is(schema);
  return isPayload(payload) ? payload : undefined;
}

function trimText(value: string | undefined | null): string | undefined {
  const trimmed = value?.trim();
  return trimmed && trimmed.length > 0 ? trimmed : undefined;
}

function asUnknownRecord(value: unknown): Record<string, unknown> | undefined {
  return value && typeof value === "object" ? (value as Record<string, unknown>) : undefined;
}

function normalizeMcpIntentTitle(value: unknown): string | undefined {
  if (typeof value !== "string") return undefined;
  const normalized = value.trim().replace(/\s+/gu, " ");
  if (!normalized) return undefined;
  const characters = Array.from(normalized);
  return characters.length <= 80 ? normalized : `${characters.slice(0, 79).join("")}…`;
}

function normalizedHttpUrl(value: unknown): string | undefined {
  if (typeof value !== "string" || value.length > 4096) return undefined;
  try {
    const url = new URL(value);
    const href = url.href;
    return (url.protocol === "http:" || url.protocol === "https:") && href.length <= 4096
      ? href
      : undefined;
  } catch {
    return undefined;
  }
}

function normalizedImageUrl(value: unknown): string | undefined {
  if (typeof value !== "string" || value.length > 4096) return undefined;
  try {
    const url = new URL(value);
    return url.protocol === "http:" || url.protocol === "https:" || url.protocol === "data:"
      ? url.href
      : undefined;
  } catch {
    return undefined;
  }
}

function normalizedAppId(value: unknown): string | undefined {
  if (typeof value !== "string") return undefined;
  const appId = value.trim();
  return appId.length > 0 && appId.length <= 512 && /^[A-Za-z0-9._-]+$/u.test(appId)
    ? appId
    : undefined;
}

function normalizedDisplayName(value: unknown): string | undefined {
  if (typeof value !== "string") return undefined;
  const displayName = value.trim().replace(/\s+/gu, " ");
  return displayName && displayName.length <= 160 ? displayName : undefined;
}

function normalizedSourceKeyPart(value: string): string {
  return value.trim().toLowerCase();
}

function nativeAppSourceKey(appId: string): string {
  const key = `native-app:${appId.toLowerCase()}`;
  if (key.length <= 512) return key;
  const digest = NodeCrypto.createHash("sha256").update(key).digest("hex");
  return `${key.slice(0, 512 - digest.length - 1)}:${digest}`;
}

function browserDisplayName(value: unknown): string | undefined {
  const normalized = normalizedDisplayName(value)?.toLowerCase();
  if (!normalized) return undefined;
  if (normalized.includes("chrome") || normalized === "chromium") return "Chrome";
  if (normalized.includes("edge")) return "Microsoft Edge";
  if (normalized.includes("firefox")) return "Firefox";
  if (normalized.includes("safari")) return "Safari";
  if (normalized.includes("arc")) return "Arc";
  if (normalized === "iab" || normalized.includes("in-app")) return "Browser";
  return normalizedDisplayName(value);
}

function browserNativeAppReference(name: string): ToolActivityNativeAppReference | undefined {
  switch (name) {
    case "Chrome":
      return { _tag: "display-name", displayName: "Google Chrome" };
    case "Microsoft Edge":
    case "Firefox":
    case "Safari":
    case "Arc":
      return { _tag: "display-name", displayName: name };
    default:
      return undefined;
  }
}

function appDisplayNameFromId(appId: string): string | undefined {
  const knownNames: Readonly<Record<string, string>> = {
    "com.apple.finder": "Finder",
    "com.apple.safari": "Safari",
    "com.google.chrome": "Chrome",
    "com.microsoft.edgemac": "Microsoft Edge",
    "org.mozilla.firefox": "Firefox",
    "company.thebrowser.browser": "Arc",
  };
  return knownNames[appId.toLowerCase()];
}

function nativeAppReference(value: unknown): ToolActivityNativeAppReference | undefined {
  const app = asUnknownRecord(value);
  if (app?.kind === "appId") {
    const appId = normalizedAppId(app.appId);
    return appId ? { _tag: "app-id", appId } : undefined;
  }
  if (app?.kind === "displayName") {
    const displayName = normalizedDisplayName(app.displayName);
    return displayName ? { _tag: "display-name", displayName } : undefined;
  }
  return undefined;
}

function themedLogoIcon(
  ...records: ReadonlyArray<Record<string, unknown> | undefined>
): ToolActivityIcon | undefined {
  for (const record of records) {
    const logoUrl = normalizedImageUrl(record?.logoUrl);
    if (!logoUrl) continue;
    const logoUrlDark = normalizedImageUrl(record?.logoUrlDark ?? record?.logoDarkUrl);
    return {
      _tag: "themed-logo",
      logoUrl,
      ...(logoUrlDark ? { logoUrlDark } : {}),
    };
  }
  return undefined;
}

interface McpToolPresentation {
  readonly toolSurface?: "browser" | "computer";
  readonly toolIcon?: ToolActivityIcon;
  readonly toolSource?: ToolActivitySource;
}

function mcpToolPresentation(
  item: Extract<CodexLifecycleItem, { readonly type: "mcpToolCall" }>,
): McpToolPresentation {
  const result = asUnknownRecord(item.result);
  const metadata = asUnknownRecord(result?._meta);
  const surface = asUnknownRecord(metadata?.["codex/toolSurface"]);
  const sourceMetadata = asUnknownRecord(metadata?.source);
  const appContext = asUnknownRecord(item.appContext);
  const sourceLogo = themedLogoIcon(surface, sourceMetadata, appContext);
  if (surface?.kind === "browserUse") {
    const screenshot = asUnknownRecord(surface.screenshot);
    const browserUse = asUnknownRecord(metadata?.browser_use);
    const openTabs = Array.isArray(surface.openTabs) ? surface.openTabs : [];
    const latestOpenTab = openTabs
      .toReversed()
      .map(asUnknownRecord)
      .find((tab) => normalizedHttpUrl(tab?.url) !== undefined);
    const selectedPage = [
      { record: screenshot, url: screenshot?.pageUrl },
      { record: browserUse, url: browserUse?.url },
      { record: latestOpenTab, url: latestOpenTab?.url },
    ]
      .map((candidate) => ({ ...candidate, pageUrl: normalizedHttpUrl(candidate.url) }))
      .find((candidate) => candidate.pageUrl !== undefined);
    const pageUrl = selectedPage?.pageUrl;
    const faviconUrl = normalizedImageUrl(
      selectedPage?.record?.faviconUrl ?? selectedPage?.record?.favIconUrl,
    );
    const faviconUrlDark = normalizedImageUrl(
      selectedPage?.record?.faviconUrlDark ?? selectedPage?.record?.favIconUrlDark,
    );
    const name =
      browserDisplayName(appContext?.appName) ??
      browserDisplayName(surface.browserFamily) ??
      browserDisplayName(surface.backend) ??
      "Browser";
    const nativeBrowserIcon = browserNativeAppReference(name);
    const sourceIcon =
      sourceLogo ??
      (nativeBrowserIcon ? ({ _tag: "native-app", app: nativeBrowserIcon } as const) : undefined);
    const sourceKeyPart = normalizedSourceKeyPart(name) || "browser";
    return {
      toolSurface: "browser",
      ...(pageUrl
        ? {
            toolIcon: {
              _tag: "website",
              pageUrl,
              ...(faviconUrl ? { faviconUrl } : {}),
              ...(faviconUrlDark ? { faviconUrlDark } : {}),
            } as const,
          }
        : {}),
      toolSource: {
        key: `browser-use:${sourceKeyPart}`,
        name,
        kind: name === "Browser" ? "browser" : "integration",
        ...(sourceIcon ? { icon: sourceIcon } : {}),
      },
    };
  }
  if (surface?.kind === "computerUse") {
    const app = nativeAppReference(surface.app);
    const args = asUnknownRecord(item.arguments);
    const argumentAppName =
      normalizedDisplayName(args?.appName) ??
      normalizedDisplayName(args?.application) ??
      normalizedDisplayName(typeof args?.app === "string" ? args.app : undefined);
    const name =
      normalizedDisplayName(appContext?.appName) ??
      argumentAppName ??
      (app?._tag === "display-name" ? app.displayName : undefined) ??
      (app?._tag === "app-id" ? appDisplayNameFromId(app.appId) : undefined) ??
      "Computer Use";
    const sourceIcon = sourceLogo ?? (app ? ({ _tag: "native-app", app } as const) : undefined);
    const sourceKey = app
      ? app._tag === "app-id"
        ? nativeAppSourceKey(app.appId)
        : `native-app-name:${normalizedSourceKeyPart(app.displayName)}`
      : "computer-use";
    return {
      toolSurface: "computer",
      ...(app ? { toolIcon: { _tag: "native-app", app } as const } : {}),
      toolSource: {
        key: sourceKey,
        name,
        kind: "computer",
        ...(sourceIcon ? { icon: sourceIcon } : {}),
      },
    };
  }

  return {};
}

const FATAL_CODEX_STDERR_SNIPPETS = ["failed to connect to websocket"];

function isFatalCodexProcessStderrMessage(message: string): boolean {
  const normalized = message.toLowerCase();
  return FATAL_CODEX_STDERR_SNIPPETS.some((snippet) => normalized.includes(snippet));
}

function normalizeCodexTokenUsage(
  usage: EffectCodexSchema.V2ThreadTokenUsageUpdatedNotification["tokenUsage"],
): ThreadTokenUsageSnapshot | undefined {
  const totalProcessedTokens = usage.total.totalTokens;
  const usedTokens = usage.last.totalTokens;
  if (usedTokens === undefined || usedTokens <= 0) {
    return undefined;
  }

  const maxTokens = usage.modelContextWindow ?? undefined;
  const inputTokens = usage.last.inputTokens;
  const cachedInputTokens = usage.last.cachedInputTokens;
  const outputTokens = usage.last.outputTokens;
  const reasoningOutputTokens = usage.last.reasoningOutputTokens;

  return {
    usedTokens,
    ...(totalProcessedTokens !== undefined && totalProcessedTokens > usedTokens
      ? { totalProcessedTokens }
      : {}),
    ...(maxTokens !== undefined ? { maxTokens } : {}),
    ...(inputTokens !== undefined ? { inputTokens } : {}),
    ...(cachedInputTokens !== undefined ? { cachedInputTokens } : {}),
    ...(outputTokens !== undefined ? { outputTokens } : {}),
    ...(reasoningOutputTokens !== undefined ? { reasoningOutputTokens } : {}),
    ...(usedTokens !== undefined ? { lastUsedTokens: usedTokens } : {}),
    ...(inputTokens !== undefined ? { lastInputTokens: inputTokens } : {}),
    ...(cachedInputTokens !== undefined ? { lastCachedInputTokens: cachedInputTokens } : {}),
    ...(outputTokens !== undefined ? { lastOutputTokens: outputTokens } : {}),
    ...(reasoningOutputTokens !== undefined
      ? { lastReasoningOutputTokens: reasoningOutputTokens }
      : {}),
    compactsAutomatically: true,
  };
}

function codexTokenUsageBreakdown(
  usage: EffectCodexSchema.V2ThreadTokenUsageUpdatedNotification__TokenUsageBreakdown,
): CodexCumulativeTokenUsage {
  return {
    inputTokens: usage.inputTokens,
    cachedInputTokens: usage.cachedInputTokens,
    ...(usage.cacheWriteInputTokens !== undefined
      ? { cacheCreationTokens: usage.cacheWriteInputTokens }
      : {}),
    outputTokens: usage.outputTokens,
    reasoningTokens: usage.reasoningOutputTokens,
  };
}

function makeCodexTurnTokenUsageState(): CodexTurnTokenUsageState {
  return {
    baseline: undefined,
    activeTurnId: undefined,
    byTurnId: new Map(),
  };
}

function getCodexTurnAccumulator(
  state: CodexTurnTokenUsageState,
  turnId: string,
): CodexTurnTokenUsageAccumulator {
  const existing = state.byTurnId.get(turnId);
  if (existing) return existing;
  const created: CodexTurnTokenUsageAccumulator = {
    inputTokens: 0,
    cachedInputTokens: 0,
    cacheCreationTokens: 0,
    outputTokens: 0,
    reasoningTokens: 0,
    observed: false,
    hasSubagents: false,
  };
  state.byTurnId.set(turnId, created);
  return created;
}

/**
 * Usage added by one `thread/tokenUsage/updated` notification. Codex reports a
 * running `total` for the thread and `last`, the usage of the newest model
 * response. Within a turn the growth of `total` equals `last`. Without a prior
 * total (first update after resume or rollback), or when Codex reset the
 * running total, `last` is the delta.
 */
function codexTurnTokenUsageDelta(
  previous: CodexCumulativeTokenUsage | undefined,
  current: CodexCumulativeTokenUsage,
  last: CodexCumulativeTokenUsage,
): CodexCumulativeTokenUsage {
  if (
    previous === undefined ||
    current.inputTokens < previous.inputTokens ||
    current.cachedInputTokens < previous.cachedInputTokens ||
    current.outputTokens < previous.outputTokens ||
    current.reasoningTokens < previous.reasoningTokens
  ) {
    return last;
  }
  return {
    inputTokens: current.inputTokens - previous.inputTokens,
    cachedInputTokens: current.cachedInputTokens - previous.cachedInputTokens,
    ...(current.cacheCreationTokens !== undefined &&
    previous.cacheCreationTokens !== undefined &&
    current.cacheCreationTokens >= previous.cacheCreationTokens
      ? { cacheCreationTokens: current.cacheCreationTokens - previous.cacheCreationTokens }
      : {}),
    outputTokens: current.outputTokens - previous.outputTokens,
    reasoningTokens: current.reasoningTokens - previous.reasoningTokens,
  };
}

function accumulateCodexTurnTokenUsage(
  state: CodexTurnTokenUsageState,
  turnId: string,
  usage: EffectCodexSchema.V2ThreadTokenUsageUpdatedNotification["tokenUsage"],
): void {
  const current = codexTokenUsageBreakdown(usage.total);
  if (state.activeTurnId !== turnId) {
    // The total is thread-wide, so every update moves the baseline. A late
    // update for a finished turn is not counted toward the live turn.
    state.baseline = current;
    return;
  }

  const accumulator = getCodexTurnAccumulator(state, turnId);
  const delta = codexTurnTokenUsageDelta(
    state.baseline,
    current,
    codexTokenUsageBreakdown(usage.last),
  );
  state.baseline = current;

  if (
    delta.inputTokens > 0 ||
    delta.cachedInputTokens > 0 ||
    delta.outputTokens > 0 ||
    delta.reasoningTokens > 0
  ) {
    accumulator.observed = true;
  }
  accumulator.inputTokens += delta.inputTokens;
  accumulator.cachedInputTokens += delta.cachedInputTokens;
  accumulator.outputTokens += delta.outputTokens;
  accumulator.reasoningTokens += delta.reasoningTokens;
  if (delta.cacheCreationTokens === undefined) {
    accumulator.cacheCreationTokens = undefined;
  } else if (accumulator.cacheCreationTokens !== undefined) {
    accumulator.cacheCreationTokens += delta.cacheCreationTokens;
  }
}

function completeCodexTurnTokenUsage(
  state: CodexTurnTokenUsageState,
  turnId: string,
  completed: boolean,
): TurnTokenUsage {
  const usage = state.byTurnId.get(turnId);
  state.byTurnId.delete(turnId);
  if (state.activeTurnId === turnId) state.activeTurnId = undefined;
  if (!usage) {
    return {
      usageStatus: "unavailable",
      usageScope: "main_agent",
      hasSubagents: false,
    };
  }

  if (!usage.observed) {
    return {
      usageStatus: "unavailable",
      usageScope: "main_agent",
      hasSubagents: usage.hasSubagents,
    };
  }

  // Codex counts cache reads and writes inside inputTokens. Clamp the
  // subsets so the record keeps the documented relationships even if a
  // counter drifts.
  return {
    usageStatus: completed ? "complete" : "partial",
    usageScope: "main_agent",
    inputTokens: usage.inputTokens,
    cachedInputTokens: Math.min(usage.inputTokens, usage.cachedInputTokens),
    ...(usage.cacheCreationTokens !== undefined
      ? { cacheCreationTokens: Math.min(usage.inputTokens, usage.cacheCreationTokens) }
      : {}),
    outputTokens: usage.outputTokens,
    reasoningTokens: Math.min(usage.outputTokens, usage.reasoningTokens),
    hasSubagents: usage.hasSubagents,
  };
}

function toTurnStatus(
  value: EffectCodexSchema.V2TurnCompletedNotification["turn"]["status"] | "cancelled",
): "completed" | "failed" | "cancelled" | "interrupted" {
  switch (value) {
    case "completed":
    case "failed":
    case "cancelled":
    case "interrupted":
      return value;
    default:
      return "completed";
  }
}

function normalizeItemType(raw: string | undefined | null): string {
  const type = trimText(raw);
  if (!type) return "item";
  return type
    .replace(/([a-z0-9])([A-Z])/g, "$1 $2")
    .replace(/[._/-]/g, " ")
    .replace(/\s+/g, " ")
    .trim()
    .toLowerCase();
}

function toCanonicalItemType(raw: string | undefined | null): CanonicalItemType {
  const type = normalizeItemType(raw);
  if (type.includes("user")) return "user_message";
  if (type.includes("agent message") || type.includes("assistant")) return "assistant_message";
  if (type.includes("reasoning") || type.includes("thought")) return "reasoning";
  if (type.includes("plan") || type.includes("todo")) return "plan";
  if (type.includes("command")) return "command_execution";
  if (type.includes("file change") || type.includes("patch") || type.includes("edit"))
    return "file_change";
  if (type.includes("mcp")) return "mcp_tool_call";
  if (type.includes("dynamic tool")) return "dynamic_tool_call";
  if (type.includes("collab")) return "collab_agent_tool_call";
  if (type.includes("web search")) return "web_search";
  if (type.includes("image")) return "image_view";
  if (type.includes("review entered")) return "review_entered";
  if (type.includes("review exited")) return "review_exited";
  if (type.includes("compact")) return "context_compaction";
  if (type.includes("error")) return "error";
  return "unknown";
}

function boundedToolArgument(value: unknown): string | undefined {
  const normalized = typeof value === "string" ? value.trim().replace(/\s+/gu, " ") : "";
  if (!normalized) return undefined;
  return normalized.length <= 48 ? normalized : `${normalized.slice(0, 47)}…`;
}

function normalizedMcpToolName(value: string): string {
  return (
    value
      .split(/__|[./:]/u)
      .at(-1)
      ?.trim() ?? value.trim()
  );
}

function computerUseToolTitle(
  item: Extract<CodexLifecycleItem, { readonly type: "mcpToolCall" }>,
  presentation: McpToolPresentation,
): string | undefined {
  if (normalizeItemType(item.server) !== "computer use") return undefined;
  if (item.status === "failed") return undefined;
  const tool = normalizeItemType(normalizedMcpToolName(item.tool)).replace(/ /gu, "_");
  const inProgress = item.status === "inProgress";
  const args = asUnknownRecord(item.arguments);
  const appName =
    (presentation.toolSource?.kind === "computer" && presentation.toolSource.name !== "Computer Use"
      ? presentation.toolSource.name
      : undefined) ??
    normalizedDisplayName(args?.appName) ??
    normalizedDisplayName(args?.application) ??
    normalizedDisplayName(typeof args?.app === "string" ? args.app : undefined);
  const withApp = (label: string) => (appName ? `${label} in ${appName}` : label);
  switch (tool) {
    case "list_apps":
      return inProgress ? "Listing apps" : "Listed apps";
    case "click":
      return withApp(inProgress ? "Clicking" : "Clicked");
    case "drag":
      return withApp(inProgress ? "Dragging" : "Dragged");
    case "get_app_state":
    case "get_state":
      return appName
        ? `${inProgress ? "Looking at" : "Looked at"} ${appName}`
        : inProgress
          ? "Looking at the screen"
          : "Looked at the screen";
    case "perform_accessibility_action":
    case "perform_secondary_action":
      return inProgress ? "Performing accessibility action" : "Performed accessibility action";
    case "press_key":
      return withApp(inProgress ? "Pressing key" : "Pressed key");
    case "scroll": {
      const direction = boundedToolArgument(args?.direction)?.toLowerCase();
      return withApp(`${inProgress ? "Scrolling" : "Scrolled"}${direction ? ` ${direction}` : ""}`);
    }
    case "set_value":
      return withApp(inProgress ? "Setting value" : "Set value");
    case "type_text":
      return withApp(inProgress ? "Typing text" : "Typed text");
    default:
      return undefined;
  }
}

function itemTitle(
  itemType: CanonicalItemType,
  item?: CodexLifecycleItem,
  presentation: McpToolPresentation = {},
): string | undefined {
  if (itemType === "mcp_tool_call" && item?.type === "mcpToolCall") {
    if (normalizedMcpToolName(item.tool) === "js") {
      const intentTitle = normalizeMcpIntentTitle(asUnknownRecord(item.arguments)?.title);
      if (intentTitle) return intentTitle;
    }
    const computerUseTitle = computerUseToolTitle(item, presentation);
    if (computerUseTitle) return computerUseTitle;
    return `${item.server} · ${item.tool}`;
  }
  switch (itemType) {
    case "assistant_message":
      return "Assistant message";
    case "user_message":
      return "User message";
    case "reasoning":
      return "Reasoning";
    case "plan":
      return "Plan";
    case "command_execution":
      return "Ran command";
    case "file_change":
      return "File change";
    case "mcp_tool_call":
      return "MCP tool call";
    case "dynamic_tool_call":
      return "Tool call";
    case "web_search":
      return "Web search";
    case "image_view":
      return "Image view";
    case "error":
      return "Error";
    default:
      return undefined;
  }
}

function itemDetail(itemType: CanonicalItemType, item: CodexLifecycleItem): string | undefined {
  const itemRecord = item as Record<string, unknown>;
  const action = itemRecord.action as Record<string, unknown> | undefined;
  const actionQueries = Array.isArray(action?.queries) ? action.queries : [];
  const candidates = [
    ...(itemType === "web_search"
      ? [itemRecord.query, action?.query, ...actionQueries, action?.pattern, action?.url]
      : []),
    "command" in item ? item.command : undefined,
    "title" in item ? item.title : undefined,
    "summary" in item ? item.summary : undefined,
    "text" in item ? item.text : undefined,
    "path" in item ? item.path : undefined,
    "prompt" in item ? item.prompt : undefined,
  ];

  for (const candidate of candidates) {
    const trimmed = typeof candidate === "string" ? trimText(candidate) : undefined;
    if (!trimmed) continue;
    return trimmed;
  }
  return undefined;
}

// Codex sends `reason` only sometimes, and sends it blank rather than absent
// often enough to matter, so an empty one must not outrank the paths below.
function nonEmptyDetail(value: string | null | undefined): string | undefined {
  const trimmed = value?.trim();
  return trimmed === undefined || trimmed.length === 0 ? undefined : trimmed;
}

// Keeps one oversized patch from pushing a wall of paths through every consumer
// of the approval, while still saying how much it covers.
const MAX_DESCRIBED_FILE_CHANGES = 20;

// An apply-patch approval carries the edited paths as the keys of `fileChanges`.
// Without them the approval card has nothing to show but its own title — the
// command-execution branch already falls back to the command for the same reason.
function describeFileChanges(
  fileChanges: EffectCodexSchema.ServerRequest__ApplyPatchApprovalParams["fileChanges"] | undefined,
): string | undefined {
  if (fileChanges === undefined) return undefined;
  const entries = Object.entries(fileChanges).toSorted(([left], [right]) =>
    left.localeCompare(right),
  );
  if (entries.length === 0) return undefined;
  const described = entries.slice(0, MAX_DESCRIBED_FILE_CHANGES).map(([path, change]) => {
    const movePath = change.type === "update" ? change.move_path : undefined;
    return movePath ? `${change.type} ${path} -> ${movePath}` : `${change.type} ${path}`;
  });
  const remaining = entries.length - described.length;
  return remaining > 0 ? `${described.join("\n")}\n+${remaining} more` : described.join("\n");
}

function toRequestTypeFromMethod(method: string): CanonicalRequestType {
  switch (method) {
    case "item/commandExecution/requestApproval":
      return "command_execution_approval";
    case "item/fileRead/requestApproval":
      return "file_read_approval";
    case "item/fileChange/requestApproval":
      return "file_change_approval";
    case "item/permissions/requestApproval":
      return "permissions_approval";
    case "mcpServer/elicitation/request":
      return "mcp_elicitation_approval";
    case "applyPatchApproval":
      return "apply_patch_approval";
    case "execCommandApproval":
      return "exec_command_approval";
    case "item/tool/requestUserInput":
      return "tool_user_input";
    case "item/tool/call":
      return "dynamic_tool_call";
    case "account/chatgptAuthTokens/refresh":
      return "auth_tokens_refresh";
    default:
      return "unknown";
  }
}

function resolveOpenedRequestType(
  method: string,
  kind: ProviderRequestKind | undefined,
): CanonicalRequestType {
  // MT desktop permissions / generic tool approvals ride MCP elicitations; their kind is
  // the only thing that tells them apart from a plain elicitation.
  if (kind === "permissions" || kind === "tool") return toRequestTypeFromKind(kind);
  const byMethod = toRequestTypeFromMethod(method);
  return byMethod === "unknown" && kind ? toRequestTypeFromKind(kind) : byMethod;
}

function toRequestTypeFromKind(kind: ProviderRequestKind | undefined): CanonicalRequestType {
  switch (kind) {
    case "command":
      return "command_execution_approval";
    case "file-read":
      return "file_read_approval";
    case "file-change":
      return "file_change_approval";
    case "permissions":
      return "permissions_approval";
    case "tool":
      return "tool_approval";
    case "mcp-elicitation":
      return "mcp_elicitation_approval";
    default:
      return "unknown";
  }
}

function toCanonicalUserInputAnswers(
  answers: EffectCodexSchema.ToolRequestUserInputResponse["answers"],
): ProviderUserInputAnswers {
  return Object.fromEntries(
    Object.entries(answers).map(([questionId, value]) => {
      const normalizedAnswers = value.answers.length === 1 ? value.answers[0]! : [...value.answers];
      return [questionId, normalizedAnswers] as const;
    }),
  );
}

function toUserInputQuestions(questions: ReadonlyArray<CodexToolUserInputQuestion>) {
  const parsedQuestions = questions
    .map((question) => {
      const options =
        question.options
          ?.map((option) => {
            const label = trimText(option.label);
            const description = trimText(option.description);
            if (!label || !description) {
              return undefined;
            }
            return { label, description };
          })
          .filter((option) => option !== undefined) ?? [];

      const id = trimText(question.id);
      const header = trimText(question.header);
      const prompt = trimText(question.question);
      if (!id || !header || !prompt || options.length === 0) {
        return undefined;
      }
      return {
        id,
        header,
        question: prompt,
        options,
        multiSelect: false,
      };
    })
    .filter((question) => question !== undefined);

  return parsedQuestions.length > 0 ? parsedQuestions : undefined;
}

function toThreadState(
  status: EffectCodexSchema.V2ThreadStatusChangedNotification["status"],
): "active" | "idle" | "archived" | "closed" | "compacted" | "error" {
  switch (status.type) {
    case "idle":
      return "idle";
    case "systemError":
      return "error";
    default:
      return "active";
  }
}

function contentStreamKindFromMethod(
  method: string,
):
  | "assistant_text"
  | "reasoning_text"
  | "reasoning_summary_text"
  | "plan_text"
  | "command_output"
  | "file_change_output" {
  switch (method) {
    case "item/agentMessage/delta":
      return "assistant_text";
    case "item/reasoning/textDelta":
      return "reasoning_text";
    case "item/reasoning/summaryTextDelta":
      return "reasoning_summary_text";
    case "item/commandExecution/outputDelta":
      return "command_output";
    case "item/fileChange/outputDelta":
      return "file_change_output";
    default:
      return "assistant_text";
  }
}

function asRuntimeItemId(itemId: ProviderEvent["itemId"] & string): RuntimeItemId {
  return RuntimeItemId.make(itemId);
}

function asRuntimeRequestId(requestId: string): RuntimeRequestId {
  return RuntimeRequestId.make(requestId);
}

function eventRawSource(event: ProviderEvent): NonNullable<ProviderRuntimeEvent["raw"]>["source"] {
  return event.kind === "request" ? "codex.app-server.request" : "codex.app-server.notification";
}

function providerRefsFromEvent(
  event: ProviderEvent,
): ProviderRuntimeEvent["providerRefs"] | undefined {
  const refs: Record<string, string> = {};
  if (event.turnId) refs.providerTurnId = event.turnId;
  if (event.itemId) refs.providerItemId = event.itemId;
  if (event.requestId) refs.providerRequestId = event.requestId;

  return Object.keys(refs).length > 0 ? (refs as ProviderRuntimeEvent["providerRefs"]) : undefined;
}

function runtimeEventBase(
  event: ProviderEvent,
  canonicalThreadId: ThreadId,
): Omit<ProviderRuntimeEvent, "type" | "payload"> {
  const refs = providerRefsFromEvent(event);
  return {
    eventId: event.id,
    provider: event.provider,
    threadId: canonicalThreadId,
    createdAt: event.createdAt,
    ...(event.turnId ? { turnId: event.turnId } : {}),
    ...(event.itemId ? { itemId: asRuntimeItemId(event.itemId) } : {}),
    ...(event.requestId ? { requestId: asRuntimeRequestId(event.requestId) } : {}),
    ...(refs ? { providerRefs: refs } : {}),
    raw: {
      source: eventRawSource(event),
      method: event.method,
      payload: event.payload ?? {},
    },
  };
}

function mapItemLifecycle(
  event: ProviderEvent,
  canonicalThreadId: ThreadId,
  lifecycle: "item.started" | "item.updated" | "item.completed",
): ProviderRuntimeEvent | undefined {
  const payload =
    readPayload(EffectCodexSchema.V2ItemStartedNotification, event.payload) ??
    readPayload(EffectCodexSchema.V2ItemCompletedNotification, event.payload);
  const item = payload?.item;
  if (!item) {
    return undefined;
  }
  const itemType = toCanonicalItemType(item.type);
  if (itemType === "unknown" && lifecycle !== "item.updated") {
    return undefined;
  }

  const detail = itemDetail(itemType, item);
  const toolPresentation = item.type === "mcpToolCall" ? mcpToolPresentation(item) : {};
  const title = itemTitle(itemType, item, toolPresentation);
  const status =
    lifecycle === "item.started"
      ? "inProgress"
      : lifecycle === "item.completed"
        ? "status" in item && (item.status === "failed" || item.status === "declined")
          ? item.status
          : "completed"
        : undefined;

  return {
    ...runtimeEventBase(event, canonicalThreadId),
    type: lifecycle,
    payload: {
      itemType,
      ...(status ? { status } : {}),
      ...(title ? { title } : {}),
      ...(detail ? { detail } : {}),
      ...toolPresentation,
      ...(event.payload !== undefined ? { data: event.payload } : {}),
    },
  };
}

/**
 * Maps the session runtime's synthetic `collabAgent/*` events (native
 * multi-agent v2 child-thread signals) into the shared task.* lifecycle.
 * Agent identity = child thread id; nickname is the display title, role is
 * agentRole (fallback: last agentPath segment, then "general-purpose").
 * A completed child turn is idle (resumable), not terminal. timelineBypass
 * keeps these rows out of the parent chat.
 */
function mapCollabAgentEvent(
  event: ProviderEvent,
  canonicalThreadId: ThreadId,
  usageByAgent: Map<string, CodexCollabUsageState>,
): ReadonlyArray<ProviderRuntimeEvent> {
  const payload =
    typeof event.payload === "object" && event.payload !== null
      ? (event.payload as Record<string, unknown>)
      : undefined;
  const agentThreadId = typeof payload?.agentThreadId === "string" ? payload.agentThreadId : "";
  if (!payload || agentThreadId.length === 0) {
    return [];
  }
  const base = runtimeEventBase(event, canonicalThreadId);
  const taskId = RuntimeTaskId.make(agentThreadId);
  const agentPath = typeof payload.agentPath === "string" ? payload.agentPath : undefined;
  const pathLeaf = agentPath?.split("/").findLast((segment) => segment.length > 0);
  const nickname = typeof payload.nickname === "string" ? payload.nickname : undefined;
  const role =
    (typeof payload.role === "string" ? payload.role : undefined) ?? pathLeaf ?? "general-purpose";
  // A bare thread id is not a name. Omitting the title lets the client fold
  // keep the real one from task.started instead of clobbering it (probe
  // finding: progress rows renamed math_one to its UUID).
  const knownName = nickname ?? pathLeaf;
  const title = knownName ?? agentThreadId;
  const model = typeof payload.model === "string" ? payload.model.trim() : "";
  const effort = typeof payload.effort === "string" ? payload.effort.trim() : "";
  // Identity repeated on every status patch so rows are self-describing when
  // the start row ages out of activity retention (review finding: a
  // reconstructed agent had a UUID name and no role/path).
  const linkage = {
    role,
    ...(knownName ? { title: knownName } : {}),
    ...(model ? { model } : {}),
    ...(effort ? { effort } : {}),
    ...(agentPath ? { agentPath } : {}),
    timelineBypass: true,
  } as const;

  switch (event.method) {
    case "collabAgent/started":
      return [
        {
          ...base,
          type: "task.started",
          payload: {
            taskId,
            description: title,
            title,
            ...linkage,
            ...(typeof payload.parentThreadId === "string"
              ? { parentAgentId: payload.parentThreadId }
              : {}),
          },
        },
      ];
    case "collabAgent/metadataUpdated":
      return [
        {
          ...base,
          type: "task.updated",
          payload: { taskId, ...linkage },
        },
      ];
    case "collabAgent/activity": {
      const activityKind = typeof payload.activityKind === "string" ? payload.activityKind : "";
      if (activityKind === "interrupted") {
        return [
          {
            ...base,
            type: "task.updated",
            payload: { taskId, status: "interrupted", ...linkage },
          },
        ];
      }
      if (activityKind === "started") {
        // Wire-probe finding: children often register via subAgentActivity
        // alone (no thread/started with a spawn source), so this is the one
        // shot at a task.started with a real name — agentPath leaf beats a
        // bare thread-id title.
        return [
          {
            ...base,
            type: "task.started",
            payload: {
              taskId,
              description: title,
              title,
              ...linkage,
            },
          },
        ];
      }
      // Reading a child's result also emits "interacted" after its turn is idle.
      // Only the child's turn or thread lifecycle can prove it resumed work.
      return [];
    }
    case "collabAgent/turnStarted":
      return [
        {
          ...base,
          type: "task.updated",
          payload: { taskId, status: "running", ...linkage },
        },
      ];
    case "collabAgent/turnCompleted": {
      // Idle, not terminal: the identity is resumable via sendInput/resume.
      const turn =
        typeof payload.turn === "object" && payload.turn !== null
          ? (payload.turn as Record<string, unknown>)
          : undefined;
      const turnStatus = typeof turn?.status === "string" ? turn.status : undefined;
      const status =
        turnStatus === "failed"
          ? ("failed" as const)
          : turnStatus === "interrupted"
            ? ("interrupted" as const)
            : ("idle" as const);
      return [
        {
          ...base,
          type: "task.updated",
          payload: { taskId, status, ...linkage },
        },
      ];
    }
    case "collabAgent/statusChanged": {
      const status =
        typeof payload.status === "object" && payload.status !== null
          ? (payload.status as Record<string, unknown>)
          : undefined;
      const statusType = typeof status?.type === "string" ? status.type : undefined;
      if (statusType === "systemError") {
        // Silently dropping this once left children stuck running forever.
        return [
          {
            ...base,
            type: "task.updated",
            payload: { taskId, status: "failed", ...linkage },
          },
        ];
      }
      if (statusType === "active") {
        const flags = Array.isArray(status?.activeFlags) ? status.activeFlags : [];
        const waiting = flags.some(
          (flag) => flag === "waitingOnApproval" || flag === "waitingOnUserInput",
        );
        return [
          {
            ...base,
            type: "task.updated",
            payload: { taskId, status: waiting ? "waiting" : "running", ...linkage },
          },
        ];
      }
      if (statusType === "idle") {
        return [
          {
            ...base,
            type: "task.updated",
            payload: { taskId, status: "idle", ...linkage },
          },
        ];
      }
      return [];
    }
    case "collabAgent/tokenUsage": {
      // Calibrate the first cumulative `total` with `last` to remove copied
      // history. Later snapshots stay based on `total`; `last` resets per turn.
      const tokenUsage =
        typeof payload.tokenUsage === "object" && payload.tokenUsage !== null
          ? (payload.tokenUsage as Record<string, unknown>)
          : undefined;
      const total =
        typeof tokenUsage?.total === "object" && tokenUsage.total !== null
          ? (tokenUsage.total as Record<string, unknown>)
          : undefined;
      const last =
        typeof tokenUsage?.last === "object" && tokenUsage.last !== null
          ? (tokenUsage.last as Record<string, unknown>)
          : undefined;
      const count = (value: unknown): number | undefined =>
        typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined;
      // Same validation as every other field: RuntimeTaskUsage.totalTokens
      // is NonNegativeInt, so NaN/Infinity/negative wire values must miss.
      const totalTokens = count(total?.totalTokens);
      const lastTotalTokens = count(last?.totalTokens);
      if (totalTokens === undefined || lastTotalTokens === undefined) {
        return [];
      }

      const inputTokens = count(total?.inputTokens);
      const cachedInputTokens = count(total?.cachedInputTokens);
      const outputTokens = count(total?.outputTokens);
      const reasoningOutputTokens = count(total?.reasoningOutputTokens);
      const lastInputTokens = count(last?.inputTokens);
      const lastCachedInputTokens = count(last?.cachedInputTokens);
      const lastOutputTokens = count(last?.outputTokens);
      const lastReasoningOutputTokens = count(last?.reasoningOutputTokens);
      const existing = usageByAgent.get(agentThreadId);
      const baselineCount = (
        offset: number | undefined,
        cumulative: number | undefined,
        currentTurn: number | undefined,
      ): number | undefined =>
        offset ??
        (cumulative !== undefined && currentTurn !== undefined
          ? Math.max(cumulative - currentTurn, 0)
          : undefined);
      const baselineInputTokens = baselineCount(
        existing?.baseline.inputTokens,
        inputTokens,
        lastInputTokens,
      );
      const baselineCachedInputTokens = baselineCount(
        existing?.baseline.cachedInputTokens,
        cachedInputTokens,
        lastCachedInputTokens,
      );
      const baselineOutputTokens = baselineCount(
        existing?.baseline.outputTokens,
        outputTokens,
        lastOutputTokens,
      );
      const baselineReasoningOutputTokens = baselineCount(
        existing?.baseline.reasoningOutputTokens,
        reasoningOutputTokens,
        lastReasoningOutputTokens,
      );
      const baseline = {
        totalTokens: existing?.baseline.totalTokens ?? Math.max(totalTokens - lastTotalTokens, 0),
        ...(baselineInputTokens !== undefined ? { inputTokens: baselineInputTokens } : {}),
        ...(baselineCachedInputTokens !== undefined
          ? { cachedInputTokens: baselineCachedInputTokens }
          : {}),
        ...(baselineOutputTokens !== undefined ? { outputTokens: baselineOutputTokens } : {}),
        ...(baselineReasoningOutputTokens !== undefined
          ? { reasoningOutputTokens: baselineReasoningOutputTokens }
          : {}),
      } satisfies RuntimeTaskUsage;
      const normalizedCount = (
        cumulative: number | undefined,
        offset: number | undefined,
        latest: number | undefined,
      ): number | undefined =>
        cumulative !== undefined && offset !== undefined
          ? Math.max(cumulative - offset, latest ?? 0, 0)
          : undefined;
      const normalizedInputTokens = normalizedCount(
        inputTokens,
        baseline.inputTokens,
        existing?.latest.inputTokens,
      );
      const normalizedCachedInputTokens = normalizedCount(
        cachedInputTokens,
        baseline.cachedInputTokens,
        existing?.latest.cachedInputTokens,
      );
      const normalizedOutputTokens = normalizedCount(
        outputTokens,
        baseline.outputTokens,
        existing?.latest.outputTokens,
      );
      const normalizedReasoningOutputTokens = normalizedCount(
        reasoningOutputTokens,
        baseline.reasoningOutputTokens,
        existing?.latest.reasoningOutputTokens,
      );
      const typedUsage: RuntimeTaskUsage = {
        totalTokens: Math.max(
          totalTokens - baseline.totalTokens,
          existing?.latest.totalTokens ?? 0,
          0,
        ),
        ...(normalizedInputTokens !== undefined ? { inputTokens: normalizedInputTokens } : {}),
        ...(normalizedCachedInputTokens !== undefined
          ? { cachedInputTokens: normalizedCachedInputTokens }
          : {}),
        ...(normalizedOutputTokens !== undefined ? { outputTokens: normalizedOutputTokens } : {}),
        ...(normalizedReasoningOutputTokens !== undefined
          ? { reasoningOutputTokens: normalizedReasoningOutputTokens }
          : {}),
      };
      usageByAgent.set(agentThreadId, { baseline, latest: typedUsage });
      return [
        {
          ...base,
          type: "task.progress",
          payload: {
            taskId,
            description: title,
            ...linkage,
            typedUsage,
          },
        },
      ];
    }
    case "collabAgent/item": {
      const item =
        typeof payload.item === "object" && payload.item !== null
          ? (payload.item as Record<string, unknown>)
          : undefined;
      const itemTypeRaw = typeof item?.type === "string" ? item.type : undefined;
      if (!itemTypeRaw) {
        return [];
      }
      // A loose summary from the raw item: the child stream is untyped at
      // this boundary (synthetic event payload), so read best-effort fields
      // rather than force a schema decode.
      const looseSummary =
        (typeof item?.command === "string" ? item.command : undefined) ??
        (typeof item?.title === "string" ? item.title : undefined) ??
        (typeof item?.query === "string" ? item.query : undefined);
      const canonical = toCanonicalItemType(itemTypeRaw);
      const summary = looseSummary ?? canonical.replaceAll("_", " ");
      return [
        {
          ...base,
          type: "task.progress",
          payload: {
            taskId,
            description: title,
            ...linkage,
            summary,
          },
        },
      ];
    }
    case "collabAgent/closed":
      return [
        {
          ...base,
          type: "task.updated",
          payload: { taskId, status: "interrupted", ...linkage },
        },
      ];
    default:
      return [];
  }
}

function mapToRuntimeEvents(
  event: ProviderEvent,
  canonicalThreadId: ThreadId,
  collabUsageByAgent: Map<string, CodexCollabUsageState>,
): ReadonlyArray<ProviderRuntimeEvent> {
  if (event.kind === "notification" && event.method.startsWith("collabAgent/")) {
    return mapCollabAgentEvent(event, canonicalThreadId, collabUsageByAgent);
  }
  if (event.kind === "error") {
    if (!event.message) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "runtime.error",
        payload: {
          message: event.message,
          class: "provider_error",
          ...(event.payload !== undefined ? { detail: event.payload } : {}),
        },
      },
    ];
  }

  if (event.kind === "request") {
    if (event.method === "item/tool/requestUserInput") {
      const payload =
        readPayload(EffectCodexSchema.ServerRequest__ToolRequestUserInputParams, event.payload) ??
        readPayload(EffectCodexSchema.ToolRequestUserInputParams, event.payload);
      const questions = payload ? toUserInputQuestions(payload.questions) : undefined;
      if (!questions) {
        return [];
      }
      return [
        {
          ...runtimeEventBase(event, canonicalThreadId),
          type: "user-input.requested",
          payload: {
            questions,
          },
        },
      ];
    }

    const elicitation =
      event.method === "mcpServer/elicitation/request"
        ? readPayload(EffectCodexSchema.McpServerElicitationRequestParams, event.payload)
        : undefined;
    const elicitationApproval = elicitation ? describeMcpElicitation(elicitation) : undefined;
    const detail = (() => {
      switch (event.method) {
        case "item/commandExecution/requestApproval": {
          const payload = readPayload(
            EffectCodexSchema.ServerRequest__CommandExecutionRequestApprovalParams,
            event.payload,
          );
          return payload?.command ?? payload?.reason ?? undefined;
        }
        case "item/fileChange/requestApproval": {
          const payload = readPayload(
            EffectCodexSchema.ServerRequest__FileChangeRequestApprovalParams,
            event.payload,
          );
          // These params carry no path of their own, only the root the agent
          // wants to write under.
          return nonEmptyDetail(payload?.reason) ?? nonEmptyDetail(payload?.grantRoot);
        }
        case "item/permissions/requestApproval": {
          const payload = readPayload(
            EffectCodexSchema.ServerRequest__PermissionsRequestApprovalParams,
            event.payload,
          );
          return payload?.reason ?? undefined;
        }
        case "mcpServer/elicitation/request": {
          const payload = readPayload(
            EffectCodexSchema.ServerRequest__McpServerElicitationRequestParams,
            event.payload,
          );
          return payload?.message ?? undefined;
        }
        case "applyPatchApproval": {
          const payload = readPayload(
            EffectCodexSchema.ServerRequest__ApplyPatchApprovalParams,
            event.payload,
          );
          return (
            nonEmptyDetail(payload?.reason) ??
            describeFileChanges(payload?.fileChanges) ??
            nonEmptyDetail(payload?.grantRoot)
          );
        }
        case "execCommandApproval": {
          const payload = readPayload(
            EffectCodexSchema.ServerRequest__ExecCommandApprovalParams,
            event.payload,
          );
          return payload?.reason ?? payload?.command.join(" ");
        }
        case "item/tool/call": {
          const payload = readPayload(
            EffectCodexSchema.ServerRequest__DynamicToolCallParams,
            event.payload,
          );
          return payload?.tool ?? undefined;
        }
        default:
          return undefined;
      }
    })();

    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "request.opened",
        payload: {
          // Method first, as upstream does (legacy `applyPatchApproval` stays distinct);
          // the request kind only names methods the table does not know (MT desktop
          // permissions / tool approvals).
          requestType: resolveOpenedRequestType(event.method, event.requestKind),
          ...(detail ? { detail } : {}),
          ...(elicitationApproval
            ? {
                appName: elicitationApproval.appName,
                options: elicitationApproval.options,
              }
            : {}),
          ...(event.payload !== undefined ? { args: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "item/requestApproval/decision" && event.requestId) {
    const payload = readPayload(ApprovalDecisionPayload, event.payload);
    const requestType =
      event.requestKind !== undefined
        ? toRequestTypeFromKind(event.requestKind)
        : toRequestTypeFromMethod(event.method);
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "request.resolved",
        payload: {
          requestType,
          ...(payload ? { decision: payload.decision } : {}),
          ...(event.payload !== undefined ? { resolution: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "session/connecting") {
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "session.state.changed",
        payload: {
          state: "starting",
          ...(event.message ? { reason: event.message } : {}),
        },
      },
    ];
  }

  if (event.method === "session/ready") {
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "session.state.changed",
        payload: {
          state: "ready",
          ...(event.message ? { reason: event.message } : {}),
        },
      },
    ];
  }

  if (event.method === "session/started") {
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "session.started",
        payload: {
          ...(event.message ? { message: event.message } : {}),
          ...(event.payload !== undefined ? { resume: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "session/exited" || event.method === "session/closed") {
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "session.exited",
        payload: {
          ...(event.message ? { reason: event.message } : {}),
          ...(event.method === "session/closed" ? { exitKind: "graceful" } : {}),
        },
      },
    ];
  }

  if (event.method === "thread/started") {
    const payload = readPayload(EffectCodexSchema.V2ThreadStartedNotification, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "thread.started",
        payload: {
          providerThreadId: payload.thread.id,
        },
      },
    ];
  }

  if (
    event.method === "thread/status/changed" ||
    event.method === "thread/archived" ||
    event.method === "thread/unarchived" ||
    event.method === "thread/closed" ||
    event.method === "thread/compacted"
  ) {
    const payload =
      event.method === "thread/status/changed"
        ? readPayload(EffectCodexSchema.V2ThreadStatusChangedNotification, event.payload)
        : undefined;
    return [
      {
        type: "thread.state.changed",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          state:
            event.method === "thread/archived"
              ? "archived"
              : event.method === "thread/closed"
                ? "closed"
                : event.method === "thread/compacted"
                  ? "compacted"
                  : payload
                    ? toThreadState(payload.status)
                    : "active",
          ...(event.payload !== undefined ? { detail: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "thread/name/updated") {
    const payload = readPayload(EffectCodexSchema.V2ThreadNameUpdatedNotification, event.payload);
    return [
      {
        type: "thread.metadata.updated",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          ...(trimText(payload?.threadName) ? { name: trimText(payload?.threadName) } : {}),
          ...(payload
            ? {
                metadata: {
                  threadId: payload.threadId,
                  ...(payload.threadName !== undefined && payload.threadName !== null
                    ? { threadName: payload.threadName }
                    : {}),
                },
              }
            : {}),
        },
      },
    ];
  }

  if (event.method === "thread/tokenUsage/updated") {
    const payload = readPayload(
      EffectCodexSchema.V2ThreadTokenUsageUpdatedNotification,
      event.payload,
    );
    const normalizedUsage = payload ? normalizeCodexTokenUsage(payload.tokenUsage) : undefined;
    if (!normalizedUsage) {
      return [];
    }
    return [
      {
        type: "thread.token-usage.updated",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          usage: normalizedUsage,
        },
      },
    ];
  }

  if (event.method === "turn/started") {
    const turnId = event.turnId;
    if (!turnId) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        turnId,
        type: "turn.started",
        payload: {},
      },
    ];
  }

  if (event.method === "turn/completed") {
    const payload = readPayload(EffectCodexSchema.V2TurnCompletedNotification, event.payload);
    if (!payload) {
      return [];
    }
    const errorMessage = trimText(payload.turn.error?.message);
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "turn.completed",
        payload: {
          state: toTurnStatus(payload.turn.status),
          ...(errorMessage ? { errorMessage } : {}),
        },
      },
    ];
  }

  if (event.method === "turn/aborted") {
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "turn.aborted",
        payload: {
          reason: event.message ?? "Turn aborted",
        },
      },
    ];
  }

  if (event.method === "turn/plan/updated") {
    const payload = readPayload(EffectCodexSchema.V2TurnPlanUpdatedNotification, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "turn.plan.updated",
        payload: {
          ...(trimText(payload.explanation) ? { explanation: trimText(payload.explanation) } : {}),
          plan: payload.plan.map((step) => ({
            step: trimText(step.step) ?? "step",
            status:
              step.status === "completed" || step.status === "inProgress" ? step.status : "pending",
          })),
        },
      },
    ];
  }

  if (event.method === "turn/diff/updated") {
    const payload = readPayload(EffectCodexSchema.V2TurnDiffUpdatedNotification, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "turn.diff.updated",
        payload: {
          unifiedDiff: payload.diff,
        },
      },
    ];
  }

  if (event.method === "item/started") {
    const started = mapItemLifecycle(event, canonicalThreadId, "item.started");
    return started ? [started] : [];
  }

  if (event.method === "item/completed") {
    const payload = readPayload(EffectCodexSchema.V2ItemCompletedNotification, event.payload);
    const item = payload?.item;
    if (!item) {
      return [];
    }
    if (item.type === "agentMessage" && item.delivery === "async" && item.questions?.length) {
      return [
        {
          ...runtimeEventBase(event, canonicalThreadId),
          type: "user-input.requested",
          requestId: RuntimeRequestId.make(`codex-async:${canonicalThreadId}:${item.id}`),
          eventId: EventId.make(`codex-async:${canonicalThreadId}:${item.id}`),
          payload: {
            responseMode: "message",
            questions: item.questions.map((question, index) => ({
              id: String(index),
              header: "Question",
              question: question.title,
              options: (question.options ?? []).map((label) => ({ label, description: "" })),
              allowCustomAnswer: true,
              multiSelect: false,
            })),
          },
        },
      ];
    }
    const itemType = toCanonicalItemType(item.type);
    if (itemType === "plan") {
      const detail = itemDetail(itemType, item);
      if (!detail) {
        return [];
      }
      return [
        {
          ...runtimeEventBase(event, canonicalThreadId),
          type: "turn.proposed.completed",
          payload: {
            planMarkdown: detail,
          },
        },
      ];
    }
    const completed = mapItemLifecycle(event, canonicalThreadId, "item.completed");
    if (!completed || itemType !== "context_compaction") {
      return completed ? [completed] : [];
    }
    return [
      completed,
      {
        ...runtimeEventBase(event, canonicalThreadId),
        eventId: EventId.make(`${event.id}:thread-compacted`),
        type: "thread.state.changed",
        payload: { state: "compacted" },
      },
    ];
  }

  if (
    event.method === "item/reasoning/summaryPartAdded" ||
    event.method === "item/commandExecution/terminalInteraction"
  ) {
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "item.updated",
        payload: {
          itemType:
            event.method === "item/reasoning/summaryPartAdded" ? "reasoning" : "command_execution",
          ...(event.payload !== undefined ? { data: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "item/plan/delta") {
    const payload = readPayload(EffectCodexSchema.V2PlanDeltaNotification, event.payload);
    const delta = event.textDelta ?? payload?.delta;
    if (!delta || delta.length === 0) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "turn.proposed.delta",
        payload: {
          delta,
        },
      },
    ];
  }

  if (event.method === "item/agentMessage/delta") {
    const payload = readPayload(EffectCodexSchema.V2AgentMessageDeltaNotification, event.payload);
    const delta = event.textDelta ?? payload?.delta;
    if (!delta || delta.length === 0) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "content.delta",
        payload: {
          streamKind: contentStreamKindFromMethod(event.method),
          delta,
        },
      },
    ];
  }

  if (event.method === "item/commandExecution/outputDelta") {
    const payload = readPayload(
      EffectCodexSchema.V2CommandExecutionOutputDeltaNotification,
      event.payload,
    );
    const delta = event.textDelta ?? payload?.delta;
    if (!delta || delta.length === 0) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "content.delta",
        payload: {
          streamKind: "command_output",
          delta,
        },
      },
    ];
  }

  if (event.method === "item/fileChange/outputDelta") {
    const payload = readPayload(
      EffectCodexSchema.V2FileChangeOutputDeltaNotification,
      event.payload,
    );
    const delta = event.textDelta ?? payload?.delta;
    if (!delta || delta.length === 0) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "content.delta",
        payload: {
          streamKind: "file_change_output",
          delta,
        },
      },
    ];
  }

  if (event.method === "item/reasoning/summaryTextDelta") {
    const payload = readPayload(
      EffectCodexSchema.V2ReasoningSummaryTextDeltaNotification,
      event.payload,
    );
    const delta = event.textDelta ?? payload?.delta;
    if (!delta || delta.length === 0) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "content.delta",
        payload: {
          streamKind: "reasoning_summary_text",
          delta,
          ...(payload ? { summaryIndex: payload.summaryIndex } : {}),
        },
      },
    ];
  }

  if (event.method === "item/reasoning/textDelta") {
    const payload = readPayload(EffectCodexSchema.V2ReasoningTextDeltaNotification, event.payload);
    const delta = event.textDelta ?? payload?.delta;
    if (!delta || delta.length === 0) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "content.delta",
        payload: {
          streamKind: "reasoning_text",
          delta,
          ...(payload ? { contentIndex: payload.contentIndex } : {}),
        },
      },
    ];
  }

  if (event.method === "item/mcpToolCall/progress") {
    const payload = readPayload(EffectCodexSchema.V2McpToolCallProgressNotification, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "tool.progress",
        payload: {
          summary: payload.message,
        },
      },
    ];
  }

  if (event.method === "serverRequest/resolved") {
    const payload = readPayload(
      EffectCodexSchema.V2ServerRequestResolvedNotification,
      event.payload,
    );
    if (!payload) {
      return [];
    }
    const requestType = toRequestTypeFromKind(event.requestKind);
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "request.resolved",
        payload: {
          requestType,
          ...(event.payload !== undefined ? { resolution: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "item/tool/requestUserInput/answered") {
    const payload = readPayload(EffectCodexSchema.ToolRequestUserInputResponse, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        ...runtimeEventBase(event, canonicalThreadId),
        type: "user-input.resolved",
        payload: {
          answers: toCanonicalUserInputAnswers(payload.answers),
        },
      },
    ];
  }

  if (event.method === "model/rerouted") {
    const payload = readPayload(EffectCodexSchema.V2ModelReroutedNotification, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        type: "model.rerouted",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          fromModel: payload.fromModel,
          toModel: payload.toModel,
          reason: payload.reason,
        },
      },
    ];
  }

  if (event.method === "deprecationNotice") {
    const payload = readPayload(EffectCodexSchema.V2DeprecationNoticeNotification, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        type: "deprecation.notice",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          summary: payload.summary,
          ...(trimText(payload.details) ? { details: trimText(payload.details) } : {}),
        },
      },
    ];
  }

  if (event.method === "configWarning") {
    const payload = readPayload(EffectCodexSchema.V2ConfigWarningNotification, event.payload);
    if (!payload) {
      return [];
    }
    return [
      {
        type: "config.warning",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          summary: payload.summary,
          ...(trimText(payload.details) ? { details: trimText(payload.details) } : {}),
          ...(trimText(payload.path) ? { path: trimText(payload.path) } : {}),
          ...(payload.range !== undefined && payload.range !== null
            ? { range: payload.range }
            : {}),
        },
      },
    ];
  }

  if (event.method === "account/updated") {
    if (!readPayload(EffectCodexSchema.V2AccountUpdatedNotification, event.payload)) {
      return [];
    }
    return [
      {
        type: "account.updated",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          account: event.payload ?? {},
        },
      },
    ];
  }

  if (event.method === "account/rateLimits/updated") {
    const payload = readPayload(
      EffectCodexSchema.V2AccountRateLimitsUpdatedNotification,
      event.payload,
    );
    const limits = payload ? codexRateLimitsToUpdate(payload.rateLimits) : undefined;
    if (!limits) {
      return [];
    }
    return [
      {
        type: "account.rate-limits.updated",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: { limits },
      },
    ];
  }

  if (event.method === "mcpServer/oauthLogin/completed") {
    const payload = readPayload(
      EffectCodexSchema.V2McpServerOauthLoginCompletedNotification,
      event.payload,
    );
    if (!payload) {
      return [];
    }
    return [
      {
        type: "mcp.oauth.completed",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          success: payload.success,
          name: payload.name,
          ...(trimText(payload.error) ? { error: trimText(payload.error) } : {}),
        },
      },
    ];
  }

  if (event.method === "thread/realtime/started") {
    const payload = readPayload(
      EffectCodexSchema.V2ThreadRealtimeStartedNotification,
      event.payload,
    );
    if (!payload) {
      return [];
    }
    return [
      {
        type: "thread.realtime.started",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          realtimeSessionId: payload.realtimeSessionId ?? undefined,
        },
      },
    ];
  }

  if (event.method === "thread/realtime/itemAdded") {
    const payload = readPayload(
      EffectCodexSchema.V2ThreadRealtimeItemAddedNotification,
      event.payload,
    );
    if (!payload) {
      return [];
    }
    return [
      {
        type: "thread.realtime.item-added",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          item: payload.item,
        },
      },
    ];
  }

  if (event.method === "thread/realtime/outputAudio/delta") {
    const payload = readPayload(
      EffectCodexSchema.V2ThreadRealtimeOutputAudioDeltaNotification,
      event.payload,
    );
    if (!payload) {
      return [];
    }
    return [
      {
        type: "thread.realtime.audio.delta",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          audio: payload.audio,
        },
      },
    ];
  }

  if (event.method === "thread/realtime/error") {
    const payload = readPayload(EffectCodexSchema.V2ThreadRealtimeErrorNotification, event.payload);
    const message = payload?.message ?? event.message ?? "Realtime error";
    return [
      {
        type: "thread.realtime.error",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          message,
        },
      },
    ];
  }

  if (event.method === "thread/realtime/closed") {
    const payload = readPayload(
      EffectCodexSchema.V2ThreadRealtimeClosedNotification,
      event.payload,
    );
    return [
      {
        type: "thread.realtime.closed",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          reason: payload?.reason ?? event.message,
        },
      },
    ];
  }

  if (event.method === "error") {
    const payload = readPayload(EffectCodexSchema.V2ErrorNotification, event.payload);
    const message = payload?.error.message ?? event.message ?? "Provider runtime error";
    const willRetry = payload?.willRetry === true;
    return [
      {
        type: willRetry ? "runtime.warning" : "runtime.error",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          message,
          ...(!willRetry ? { class: "provider_error" as const } : {}),
          ...(event.payload !== undefined ? { detail: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "process/stderr") {
    const message = event.message ?? "Codex process stderr";
    const isFatal = isFatalCodexProcessStderrMessage(message);
    return [
      isFatal
        ? {
            type: "runtime.error",
            ...runtimeEventBase(event, canonicalThreadId),
            payload: {
              message,
              class: "provider_error" as const,
              ...(event.payload !== undefined ? { detail: event.payload } : {}),
            },
          }
        : {
            type: "runtime.warning",
            ...runtimeEventBase(event, canonicalThreadId),
            payload: {
              message,
              ...(event.payload !== undefined ? { detail: event.payload } : {}),
            },
          },
    ];
  }

  if (event.method === "windows/worldWritableWarning") {
    if (!readPayload(EffectCodexSchema.V2WindowsWorldWritableWarningNotification, event.payload)) {
      return [];
    }
    return [
      {
        type: "runtime.warning",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          message: event.message ?? "Windows world-writable warning",
          ...(event.payload !== undefined ? { detail: event.payload } : {}),
        },
      },
    ];
  }

  if (event.method === "windowsSandbox/setupCompleted") {
    const payload = readPayload(
      EffectCodexSchema.V2WindowsSandboxSetupCompletedNotification,
      event.payload,
    );
    if (!payload) {
      return [];
    }
    const successMessage = event.message ?? "Windows sandbox setup completed";
    const failureMessage = event.message ?? "Windows sandbox setup failed";

    return [
      {
        type: "session.state.changed",
        ...runtimeEventBase(event, canonicalThreadId),
        payload: {
          state: payload.success === false ? "error" : "ready",
          reason: payload.success === false ? failureMessage : successMessage,
          ...(event.payload !== undefined ? { detail: event.payload } : {}),
        },
      },
      ...(payload.success === false
        ? [
            {
              type: "runtime.warning" as const,
              ...runtimeEventBase(event, canonicalThreadId),
              payload: {
                message: failureMessage,
                ...(event.payload !== undefined ? { detail: event.payload } : {}),
              },
            },
          ]
        : []),
    ];
  }

  return [];
}

/**
 * Build a Codex provider adapter bound to a specific `CodexSettings` payload.
 *
 * The adapter is a captured closure over `codexConfig` — the `binaryPath` and
 * `homePath` are read from that payload, not from `ServerSettingsService`.
 * This is what makes multi-instance routing possible: each `ProviderInstance`
 * in the registry owns its own closure with its own config, so two Codex
 * instances with different `homePath`s cannot step on each other.
 */
export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
  codexConfig: CodexSettings,
  options?: CodexAdapterLiveOptions,
) {
  const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("codex");
  const resolveDesktopMcp = yield* makeResolveEnabledDesktopMcp();
  const fileSystem = yield* FileSystem.FileSystem;
  const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
  const crypto = yield* Crypto.Crypto;
  const serverConfig = yield* Effect.service(ServerConfig);
  const serverSettings = yield* ServerSettings.ServerSettingsService;
  const nativeEventLogger =
    options?.nativeEventLogger ??
    (options?.nativeEventLogPath !== undefined
      ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, {
          stream: "native",
        })
      : undefined);
  const managedNativeEventLogger =
    options?.nativeEventLogger === undefined ? nativeEventLogger : undefined;
  const adapterScope = yield* Scope.Scope;
  const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
  const sessions = new Map<ThreadId, CodexAdapterSessionContext>();
  const recoveryTokens = new Map<ThreadId, symbol>();

  const startSessionInternal: (
    input: Parameters<CodexAdapterShape["startSession"]>[0],
    internalOptions?: {
      readonly requireResume?: boolean;
      readonly recoveryToken?: symbol;
    },
  ) => Effect.Effect<ProviderSession, ProviderAdapterError> = (input, internalOptions) =>
    Effect.scoped(
      Effect.gen(function* () {
        if (input.provider !== undefined && input.provider !== PROVIDER) {
          return yield* new ProviderAdapterValidationError({
            provider: PROVIDER,
            operation: "startSession",
            issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`,
          });
        }

        const existing = sessions.get(input.threadId);
        if (existing && !existing.stopped) {
          yield* Effect.suspend(() => stopSessionInternal(existing));
        }

        const serviceTier =
          input.modelSelection?.instanceId === boundInstanceId
            ? getCodexServiceTierOptionValue(input.modelSelection)
            : undefined;
        const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId);
        const launchArgs = resolveCodexLaunchArgs(codexConfig.launchArgs, options?.environment);
        // Offer Computer Use tools whenever the desktop MCP resolves and is
        // enabled — same as Claude/Cursor/Grok. Wire via Codex `-c mcp_servers.*`.
        // User config wins: launch args that already configure the server name
        // skip injection, since session `-c` flags would override them.
        const desktopMcp = hasConfiguredMcpServerNamed(
          codexLaunchArgv(launchArgs),
          DESKTOP_MCP_SERVER_NAME,
        )
          ? undefined
          : yield* resolveDesktopMcp();
        const appServerArgs: string[] = [];
        if (mcpSession) {
          appServerArgs.push(
            "-c",
            `mcp_servers.t3-code.url=${mcpSession.endpoint}`,
            "-c",
            'mcp_servers.t3-code.bearer_token_env_var="T3_MCP_BEARER_TOKEN"',
          );
        }
        if (desktopMcp) {
          // Prefer TOML literal strings so Windows paths keep backslashes;
          // fall back to escaped basic strings when a value contains `'`.
          const quoteToml = (value: string): string => {
            if (!value.includes("'") && !value.includes("\n") && !value.includes("\r")) {
              return `'${value}'`;
            }
            return `"${value
              .replace(/\\/g, "\\\\")
              .replace(/"/g, '\\"')
              .replace(/\n/g, "\\n")
              .replace(/\r/g, "\\r")
              .replace(/\t/g, "\\t")}"`;
          };
          appServerArgs.push(
            "-c",
            `mcp_servers.${DESKTOP_MCP_SERVER_NAME}.command=${quoteToml(desktopMcp.path)}`,
          );
          for (const entry of desktopMcp.env) {
            appServerArgs.push(
              "-c",
              `mcp_servers.${DESKTOP_MCP_SERVER_NAME}.env.${entry.name}=${quoteToml(entry.value)}`,
            );
          }
        }
        const runtimeInput: CodexSessionRuntimeOptions = {
          threadId: input.threadId,
          providerInstanceId: boundInstanceId,
          cwd: input.cwd ?? process.cwd(),
          binaryPath: codexConfig.binaryPath,
          launchArgs,
          ...(options?.environment ? { environment: options.environment } : {}),
          ...(codexConfig.homePath ? { homePath: codexConfig.homePath } : {}),
          ...(isCodexResumeCursorSchema(input.resumeCursor)
            ? { resumeCursor: input.resumeCursor }
            : {}),
          ...(internalOptions?.requireResume === true ? { requireResume: true } : {}),
          runtimeMode: input.runtimeMode,
          ...(input.modelSelection?.instanceId === boundInstanceId
            ? { model: input.modelSelection.model }
            : {}),
          ...(serviceTier ? { serviceTier } : {}),
          ...(mcpSession || desktopMcp
            ? {
                environment: {
                  ...(options?.environment ?? process.env),
                  ...(mcpSession
                    ? {
                        T3_MCP_BEARER_TOKEN: mcpSession.authorizationHeader.replace(
                          /^Bearer\s+/,
                          "",
                        ),
                      }
                    : {}),
                },
                appServerArgs,
              }
            : {}),
        };
        const turnTokenUsage = makeCodexTurnTokenUsageState();
        const sessionScope = yield* Scope.make("sequential");
        let sessionScopeTransferred = false;
        yield* Effect.addFinalizer(() =>
          sessionScopeTransferred ? Effect.void : Scope.close(sessionScope, Exit.void),
        );
        const createRuntime = options?.makeRuntime ?? makeCodexSessionRuntime;
        const pendingTurns = new Map<TurnId, CodexPendingTurn>();
        const deferredTurnCompletions = new Map<TurnId, ProviderEvent>();
        const unsafeTurnIds = new Set<TurnId>();
        let sessionContext: CodexAdapterSessionContext | undefined;
        const runtime = yield* createRuntime(runtimeInput).pipe(
          Effect.provideService(Scope.Scope, sessionScope),
          Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner),
          Effect.provideService(Crypto.Crypto, crypto),
          Effect.mapError(
            (cause) =>
              new ProviderAdapterProcessError({
                provider: PROVIDER,
                threadId: input.threadId,
                detail: cause.message,
                cause,
              }),
          ),
        );

        // Fork into the session scope, not the calling fiber. `forkChild` makes
        // this a child of `startSession`, and Effect interrupts a fiber's
        // children when it completes, so the consumer died on return and every
        // runtime event the session emitted afterwards was dropped.
        const collabUsageByAgent = new Map<string, CodexCollabUsageState>();
        const eventFiber = yield* Stream.runForEach(runtime.events, (event) =>
          Effect.gen(function* () {
            yield* writeNativeEvent(event);
            if (event.turnId !== undefined && isUnsafeUsageLimitRetryEvent(event)) {
              unsafeTurnIds.add(event.turnId);
            }
            if (event.method === "turn/started" && event.turnId) {
              if (turnTokenUsage.activeTurnId !== event.turnId) {
                turnTokenUsage.byTurnId.clear();
                turnTokenUsage.activeTurnId = event.turnId;
                getCodexTurnAccumulator(turnTokenUsage, event.turnId);
              }
            } else if (event.method === "thread/tokenUsage/updated") {
              const payload = readPayload(
                EffectCodexSchema.V2ThreadTokenUsageUpdatedNotification,
                event.payload,
              );
              if (payload) {
                accumulateCodexTurnTokenUsage(turnTokenUsage, payload.turnId, payload.tokenUsage);
              }
            } else if (turnTokenUsage.activeTurnId) {
              const collabPayload =
                typeof event.payload === "object" && event.payload !== null
                  ? (event.payload as Record<string, unknown>)
                  : undefined;
              const isCollabSpawn =
                event.method === "collabAgent/started" ||
                (event.method === "collabAgent/activity" &&
                  collabPayload?.activityKind === "started");
              if (isCollabSpawn && event.turnId === turnTokenUsage.activeTurnId) {
                getCodexTurnAccumulator(turnTokenUsage, turnTokenUsage.activeTurnId).hasSubagents =
                  true;
              }
            }

            const runtimeEvents = mapToRuntimeEvents(event, event.threadId, collabUsageByAgent).map(
              (runtimeEvent) => {
                if (runtimeEvent.type === "turn.completed" && runtimeEvent.turnId) {
                  return {
                    ...runtimeEvent,
                    payload: {
                      ...runtimeEvent.payload,
                      tokenUsage: completeCodexTurnTokenUsage(
                        turnTokenUsage,
                        String(runtimeEvent.turnId),
                        runtimeEvent.payload.state === "completed",
                      ),
                    },
                  } satisfies ProviderRuntimeEvent;
                }
                if (runtimeEvent.type === "turn.aborted" && runtimeEvent.turnId) {
                  return {
                    ...runtimeEvent,
                    payload: {
                      ...runtimeEvent.payload,
                      tokenUsage: completeCodexTurnTokenUsage(
                        turnTokenUsage,
                        String(runtimeEvent.turnId),
                        false,
                      ),
                    },
                  } satisfies ProviderRuntimeEvent;
                }
                return runtimeEvent;
              },
            );
            if (runtimeEvents.length === 0) {
              yield* Effect.logDebug("ignoring unhandled Codex provider event", {
                method: event.method,
                threadId: event.threadId,
                turnId: event.turnId,
                itemId: event.itemId,
              });
            } else {
              yield* Queue.offerAll(runtimeEventQueue, runtimeEvents);
            }
            if (sessionContext !== undefined) {
              yield* scheduleUsageLimitRecovery(sessionContext, event);
            }
          }),
        ).pipe(Effect.forkIn(sessionScope));

        const started = yield* runtime.start().pipe(
          Effect.mapError(
            (cause) =>
              new ProviderAdapterProcessError({
                provider: PROVIDER,
                threadId: input.threadId,
                detail: cause.message,
                cause,
              }),
          ),
          Effect.onError(() =>
            runtime.close.pipe(
              Effect.andThen(Effect.ignore(Scope.close(sessionScope, Exit.void))),
              Effect.andThen(Fiber.interrupt(eventFiber)),
              Effect.ignore,
            ),
          ),
        );

        if (
          internalOptions?.recoveryToken !== undefined &&
          recoveryTokens.get(input.threadId) !== internalOptions.recoveryToken
        ) {
          sessionScopeTransferred = true;
          yield* runtime.close.pipe(
            Effect.andThen(Effect.ignore(Scope.close(sessionScope, Exit.void))),
            Effect.andThen(Fiber.interrupt(eventFiber)),
            Effect.ignore,
          );
          return started;
        }

        const context: CodexAdapterSessionContext = {
          threadId: input.threadId,
          scope: sessionScope,
          runtime,
          eventFiber,
          startInput: input,
          providerSession: started,
          pendingTurns,
          deferredTurnCompletions,
          unsafeTurnIds,
          ...(internalOptions?.recoveryToken !== undefined
            ? { recoveryToken: internalOptions.recoveryToken }
            : {}),
          recovering: false,
          turnTokenUsage,
          stopped: false,
        };
        sessionContext = context;
        sessions.set(input.threadId, context);
        sessionScopeTransferred = true;

        return started;
      }),
    );

  const startSession: CodexAdapterShape["startSession"] = (input) =>
    Effect.sync(() => recoveryTokens.delete(input.threadId)).pipe(
      Effect.andThen(startSessionInternal(input)),
    );

  const resolveAttachment = Effect.fn("resolveAttachment")(function* (
    input: ProviderSendTurnInput,
    attachment: NonNullable<ProviderSendTurnInput["attachments"]>[number],
  ) {
    const attachmentPath = resolveAttachmentPath({
      attachmentsDir: serverConfig.attachmentsDir,
      attachment,
    });
    if (!attachmentPath) {
      return yield* new ProviderAdapterRequestError({
        provider: PROVIDER,
        method: "turn/start",
        detail: `Invalid attachment id '${attachment.id}'.`,
      });
    }
    const bytes = yield* fileSystem.readFile(attachmentPath).pipe(
      Effect.mapError(
        (cause) =>
          new ProviderAdapterRequestError({
            provider: PROVIDER,
            method: "turn/start",
            detail: `Failed to read attachment file: ${cause.message}.`,
            cause,
          }),
      ),
    );
    return {
      type: "image" as const,
      url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`,
    };
  });

  const sendMappedTurn = Effect.fn("CodexAdapter.sendMappedTurn")(function* (
    session: CodexAdapterSessionContext,
    input: CodexSessionRuntimeSendTurnInput,
    restartInput: Parameters<CodexAdapterShape["startSession"]>[0],
    retryCount: number,
  ): Effect.fn.Return<ProviderTurnStartResult, ProviderAdapterError> {
    const result = yield* session.runtime
      .sendTurn(input)
      .pipe(
        Effect.mapError((cause) => mapCodexRuntimeError(session.threadId, "turn/start", cause)),
      );
    session.pendingTurns.set(result.turnId, {
      input,
      restartInput,
      retryCount,
    });
    const deferredCompletion = session.deferredTurnCompletions.get(result.turnId);
    if (deferredCompletion !== undefined) {
      yield* scheduleUsageLimitRecovery(session, deferredCompletion);
    }
    return result;
  });

  const recoverUsageLimitTurn = Effect.fn("CodexAdapter.recoverUsageLimitTurn")(function* (input: {
    readonly sourceSession: CodexAdapterSessionContext;
    readonly failedTurnId: TurnId;
    readonly pendingTurn: CodexPendingTurn;
    readonly token: symbol;
  }): Effect.fn.Return<void, ProviderAdapterError | CodexSessionRuntimeError> {
    const resumeCursor = input.sourceSession.providerSession.resumeCursor;
    if (!isCodexResumeCursorSchema(resumeCursor)) {
      yield* Effect.logWarning("codex usage-limit recovery skipped without a resume cursor", {
        threadId: input.sourceSession.threadId,
        failedTurnId: input.failedTurnId,
      });
      return;
    }

    const resumedSession = yield* startSessionInternal(
      {
        ...input.pendingTurn.restartInput,
        resumeCursor,
      },
      { requireResume: true, recoveryToken: input.token },
    );
    const recoveredContext = sessions.get(input.sourceSession.threadId);
    if (
      recoveryTokens.get(input.sourceSession.threadId) !== input.token ||
      recoveredContext === undefined ||
      recoveredContext.stopped
    ) {
      if (
        recoveredContext?.recoveryToken === input.token &&
        !recoveredContext.stopped &&
        sessions.get(input.sourceSession.threadId) === recoveredContext
      ) {
        yield* stopSessionInternal(recoveredContext);
      }
      return;
    }

    if (
      !isCodexResumeCursorSchema(resumedSession.resumeCursor) ||
      resumedSession.resumeCursor.threadId !== resumeCursor.threadId
    ) {
      yield* Effect.logWarning("codex usage-limit recovery resumed a different thread", {
        threadId: input.sourceSession.threadId,
        expectedProviderThreadId: resumeCursor.threadId,
      });
      yield* stopSessionInternal(recoveredContext);
      return;
    }

    if (!resumedRequestedModel(input.pendingTurn.input.model, resumedSession.model)) {
      yield* Effect.logWarning("codex usage-limit recovery resumed with a different model", {
        threadId: input.sourceSession.threadId,
        requestedModel: input.pendingTurn.input.model,
        resumedModel: resumedSession.model,
      });
      yield* stopSessionInternal(recoveredContext);
      return;
    }

    const snapshot = yield* recoveredContext.runtime.readThread;
    const rollbackFailedTurn = requiresFailedTurnRollback(snapshot, input.failedTurnId);
    if (rollbackFailedTurn === undefined) {
      yield* Effect.logWarning(
        "codex usage-limit recovery found provider activity in failed turn",
        {
          threadId: input.sourceSession.threadId,
          failedTurnId: input.failedTurnId,
        },
      );
      return;
    }
    if (rollbackFailedTurn) {
      yield* recoveredContext.runtime.rollbackThread(1);
    }

    if (recoveryTokens.get(input.sourceSession.threadId) !== input.token) {
      yield* stopSessionInternal(recoveredContext);
      return;
    }

    yield* sendMappedTurn(
      recoveredContext,
      input.pendingTurn.input,
      input.pendingTurn.restartInput,
      input.pendingTurn.retryCount + 1,
    );
  });

  const scheduleUsageLimitRecovery = Effect.fn("CodexAdapter.scheduleUsageLimitRecovery")(
    function* (session: CodexAdapterSessionContext, event: ProviderEvent): Effect.fn.Return<void> {
      if (event.method !== "turn/completed" || event.turnId === undefined) {
        return;
      }

      const failedTurnId = event.turnId;
      const pendingTurn = session.pendingTurns.get(failedTurnId);
      if (pendingTurn === undefined) {
        session.deferredTurnCompletions.set(failedTurnId, event);
        return;
      }

      if (!isRetryableUsageLimitCompletion(event)) {
        session.pendingTurns.delete(failedTurnId);
        session.deferredTurnCompletions.delete(failedTurnId);
        session.unsafeTurnIds.delete(failedTurnId);
        return;
      }

      const hadQueuedTurns = session.pendingTurns.size !== 1;
      const safeToRetry =
        pendingTurn.retryCount === 0 &&
        !hadQueuedTurns &&
        !session.unsafeTurnIds.has(failedTurnId) &&
        !session.recovering &&
        !session.stopped &&
        sessions.get(session.threadId) === session;
      session.pendingTurns.delete(failedTurnId);
      session.deferredTurnCompletions.delete(failedTurnId);
      session.unsafeTurnIds.delete(failedTurnId);
      if (hadQueuedTurns) {
        for (const queuedTurnId of session.pendingTurns.keys()) {
          session.unsafeTurnIds.add(queuedTurnId);
        }
      }
      if (!safeToRetry) {
        return;
      }

      const token = Symbol("codex-usage-limit-recovery");
      recoveryTokens.set(session.threadId, token);
      session.recovering = true;
      yield* recoverUsageLimitTurn({
        sourceSession: session,
        failedTurnId,
        pendingTurn,
        token,
      }).pipe(
        Effect.catchCause((cause) =>
          Effect.logWarning("codex usage-limit recovery failed", {
            threadId: session.threadId,
            failedTurnId,
            cause,
          }),
        ),
        Effect.ensuring(
          Effect.sync(() => {
            if (recoveryTokens.get(session.threadId) === token) {
              recoveryTokens.delete(session.threadId);
            }
            session.recovering = false;
          }),
        ),
        Effect.forkIn(adapterScope),
        Effect.asVoid,
      );
    },
  );

  const sendTurn: CodexAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) {
    // Codex ingests images only. Anything else would be base64-encoded as an
    // image and rejected or misread; generic files reach the agent through the
    // path line ProviderService puts in the prompt.
    const codexAttachments = yield* Effect.forEach(
      (input.attachments ?? []).filter((attachment) => attachment.type === "image"),
      (attachment) => resolveAttachment(input, attachment),
      { concurrency: 1 },
    );

    const session = yield* requireSession(input.threadId);
    if (session.recovering || recoveryTokens.has(input.threadId)) {
      return yield* new ProviderAdapterRequestError({
        provider: PROVIDER,
        method: "turn/start",
        detail: "Codex is recovering this conversation after reaching its usage limit.",
      });
    }
    const reasoningEffort =
      input.modelSelection?.instanceId === boundInstanceId
        ? getModelSelectionStringOptionValue(input.modelSelection, "reasoningEffort")
        : undefined;
    const serviceTier =
      input.modelSelection?.instanceId === boundInstanceId
        ? getCodexServiceTierOptionValue(input.modelSelection)
        : undefined;
    const computerHistoryContext = yield* ComputerHistoryService.loadComputerHistoryContextProvided(
      serverConfig,
      serverSettings,
    );
    const mappedInput: CodexSessionRuntimeSendTurnInput = {
      ...(input.input !== undefined ? { input: input.input } : {}),
      ...(input.modelSelection?.instanceId === boundInstanceId
        ? { model: input.modelSelection.model }
        : {}),
      ...(reasoningEffort
        ? {
            effort: reasoningEffort as EffectCodexSchema.V2TurnStartParams__ReasoningEffort,
          }
        : {}),
      ...(serviceTier ? { serviceTier } : {}),
      ...(input.interactionMode !== undefined ? { interactionMode: input.interactionMode } : {}),
      ...(codexAttachments.length > 0 ? { attachments: codexAttachments } : {}),
      ...(computerHistoryContext ? { computerHistoryContext } : {}),
    };
    const restartInput = {
      ...session.startInput,
      ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}),
    };
    return yield* sendMappedTurn(session, mappedInput, restartInput, 0);
  });

  const requireSession = Effect.fn("requireSession")(function* (threadId: ThreadId) {
    const session = sessions.get(threadId);
    if (!session || session.stopped) {
      return yield* new ProviderAdapterSessionNotFoundError({
        provider: PROVIDER,
        threadId,
      });
    }
    return session;
  });

  const interruptTurn: CodexAdapterShape["interruptTurn"] = (threadId, turnId) =>
    Effect.sync(() => recoveryTokens.delete(threadId)).pipe(
      Effect.andThen(requireSession(threadId)),
      Effect.flatMap((session) => session.runtime.interruptTurn(turnId)),
      Effect.mapError((cause) =>
        cause._tag === "ProviderAdapterSessionNotFoundError"
          ? cause
          : mapCodexRuntimeError(threadId, "turn/interrupt", cause),
      ),
    );

  const compactThread = Effect.fn("compactThread")(function* (threadId: ThreadId) {
    const session = yield* requireSession(threadId);
    yield* session.runtime.compactThread.pipe(
      Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/compact/start", cause)),
    );
  });

  const readThread: CodexAdapterShape["readThread"] = (threadId) =>
    requireSession(threadId).pipe(
      Effect.flatMap((session) => session.runtime.readThread),
      Effect.mapError((cause) =>
        cause._tag === "ProviderAdapterSessionNotFoundError"
          ? cause
          : mapCodexRuntimeError(threadId, "thread/read", cause),
      ),
      Effect.map((snapshot) => ({
        threadId,
        turns: snapshot.turns,
      })),
    );

  const rollbackThread: CodexAdapterShape["rollbackThread"] = (threadId, numTurns) => {
    if (!Number.isInteger(numTurns) || numTurns < 1) {
      return Effect.fail(
        new ProviderAdapterValidationError({
          provider: PROVIDER,
          operation: "rollbackThread",
          issue: "numTurns must be an integer >= 1.",
        }),
      );
    }

    return requireSession(threadId).pipe(
      Effect.flatMap((session) =>
        session.runtime.rollbackThread(numTurns).pipe(
          Effect.tap(() =>
            Effect.sync(() => {
              session.turnTokenUsage.baseline = undefined;
              session.turnTokenUsage.activeTurnId = undefined;
              session.turnTokenUsage.byTurnId.clear();
            }),
          ),
        ),
      ),
      Effect.mapError((cause) =>
        cause._tag === "ProviderAdapterSessionNotFoundError"
          ? cause
          : mapCodexRuntimeError(threadId, "thread/rollback", cause),
      ),
      Effect.map((snapshot) => ({
        threadId,
        turns: snapshot.turns,
      })),
    );
  };

  const uploadFeedback: CodexAdapterShape["uploadFeedback"] = (input) =>
    requireSession(input.threadId).pipe(
      Effect.flatMap((session) => session.runtime.uploadFeedback(input.reason)),
      Effect.map(({ threadId }) => ({ feedbackId: threadId })),
      Effect.mapError((cause) =>
        cause._tag === "ProviderAdapterSessionNotFoundError"
          ? cause
          : mapCodexRuntimeError(input.threadId, "feedback/upload", cause),
      ),
    );

  const respondToRequest: CodexAdapterShape["respondToRequest"] = (threadId, requestId, decision) =>
    requireSession(threadId).pipe(
      Effect.flatMap((session) => session.runtime.respondToRequest(requestId, decision)),
      Effect.mapError((cause) =>
        cause._tag === "ProviderAdapterSessionNotFoundError"
          ? cause
          : mapCodexRuntimeError(threadId, "item/requestApproval/decision", cause),
      ),
    );

  const respondToUserInput: CodexAdapterShape["respondToUserInput"] = (
    threadId,
    requestId,
    answers,
  ) =>
    requireSession(threadId).pipe(
      Effect.flatMap((session) => session.runtime.respondToUserInput(requestId, answers)),
      Effect.mapError((cause) =>
        cause._tag === "ProviderAdapterSessionNotFoundError"
          ? cause
          : mapCodexRuntimeError(threadId, "item/tool/requestUserInput", cause),
      ),
    );

  const writeNativeEvent = Effect.fnUntraced(function* (event: ProviderEvent) {
    if (!nativeEventLogger) {
      return;
    }
    yield* nativeEventLogger.write(event, event.threadId);
  });

  const stopSessionInternal = Effect.fn("stopSessionInternal")(function* (
    session: CodexAdapterSessionContext,
  ) {
    if (session.stopped) {
      return;
    }
    session.stopped = true;
    if (sessions.get(session.threadId) === session) {
      sessions.delete(session.threadId);
    }
    yield* session.runtime.close.pipe(Effect.ignore);
    yield* Effect.ignore(Scope.close(session.scope, Exit.void));
    yield* Fiber.interrupt(session.eventFiber).pipe(Effect.ignore);
  });

  const stopSession: CodexAdapterShape["stopSession"] = (threadId) =>
    Effect.gen(function* () {
      recoveryTokens.delete(threadId);
      const session = sessions.get(threadId);
      if (!session) {
        return;
      }
      yield* stopSessionInternal(session);
    });

  const listSessions: CodexAdapterShape["listSessions"] = () =>
    Effect.forEach(
      Array.from(sessions.values()).filter((session) => !session.stopped),
      (session) => session.runtime.getSession,
      { concurrency: 1 },
    );

  const hasSession: CodexAdapterShape["hasSession"] = (threadId) =>
    Effect.succeed(Boolean(sessions.get(threadId) && !sessions.get(threadId)?.stopped));

  const stopAll: CodexAdapterShape["stopAll"] = () =>
    Effect.sync(() => recoveryTokens.clear()).pipe(
      Effect.andThen(
        Effect.forEach(Array.from(sessions.values()), stopSessionInternal, {
          concurrency: 1,
          discard: true,
        }),
      ),
      Effect.asVoid,
    );

  yield* Effect.acquireRelease(Effect.void, () =>
    stopAll().pipe(
      Effect.andThen(Queue.shutdown(runtimeEventQueue)),
      Effect.andThen(managedNativeEventLogger?.close() ?? Effect.void),
      Effect.ignore,
    ),
  );

  return {
    provider: PROVIDER,
    capabilities: {
      sessionModelSwitch: "in-session",
      promptlessTurnContinuation: true,
    },
    startSession,
    sendTurn,
    compaction: { type: "native", start: compactThread },
    interruptTurn,
    readThread,
    rollbackThread,
    uploadFeedback,
    respondToRequest,
    respondToUserInput,
    stopSession,
    listSessions,
    hasSession,
    stopAll,
    get streamEvents() {
      return Stream.fromQueue(runtimeEventQueue);
    },
  } satisfies CodexAdapterShape;
});

// NOTE: the old `CodexAdapterLive` / `makeCodexAdapterLive` singleton Layer
// exports have been removed as part of the per-instance-driver refactor.
// `makeCodexAdapter(codexConfig, options?)` is now invoked directly by
// `CodexDriver.create()` for each configured instance; downstream consumers
// (server bootstrap, integration harness, this module's tests) will be
// migrated to the registry in a follow-up pass.