import { pipe } from "effect/Function";
import * as DocumentText from "../../provider/DocumentText.ts";
import { revertCodexThread } from "../../provider/CodexThreadRevert.ts";
import { historyResponseItems } from "@t3tools/provider-core/server/handoffBudget";
import { makeProviderTextDeltaCoalescer } from "./ProviderTextDeltaCoalescer.ts";
import {
  mcpToolPresentation,
  type McpToolPresentation,
} from "../../provider/CodexToolPresentation.ts";
import {
  makeCodexTurnTokenUsageState,
  getCodexTurnAccumulator,
  accumulateCodexTurnTokenUsage,
  completeCodexTurnTokenUsage,
  type CodexTurnTokenUsageState,
} from "../../provider/CodexTurnTokenUsage.ts";
import type { ManagedServerProvider } from "@t3tools/provider-core/server/snapshot";
import type { CodexEffectiveRuntime } from "../../provider/CodexManagedRuntime.ts";
import { classifyCodexManagedError } from "../../provider/CodexManagedErrors.ts";
import {
  buildCodexInitializeParams,
  codexAppServerEnvironment,
  codexDefaultReasoningEffortOverride,
} from "../../provider/CodexProvider.ts";
import {
  codexRateLimitsToUpdate,
  mergeCodexRateLimits,
  codexUsageLimitResetAt,
  type CodexRateLimitSnapshot,
} from "../../provider/codexUsageLimits.ts";
import {
  CodexSettings,
  defaultInstanceIdForDriver,
  isOrchestrationV2WorkActive,
  type OrchestrationV2GoalCommand,
  OrchestrationV2ProviderGoal,
  parseGoalCommand,
  ProviderDriverKind,
  ProviderSetupError,
  type ServerProviderCompatibilityAdvisory,
  type ServerProviderModel,
  isDelegatedAppThread,
  ThreadId,
} from "@t3tools/contracts";
import { SKILL_MENTION_PATTERN } from "@t3tools/shared/composerInlineTokens";
import { AgentScope } from "@t3tools/shared/AgentScope";
import * as HostProcess from "@t3tools/shared/HostProcess";
import {
  COMPUTER_USE_TOOL_SOURCE,
  dynamicToolSurface,
  dynamicToolTitle,
} from "@t3tools/shared/toolActivity";
import { getModelSelectionStringOptionValue, modelSelectionsEqual } from "@t3tools/shared/model";
import { appendGeneratedImageEmbeds } from "@t3tools/shared/generatedImageMarkdown";
import { resolveSpawnCommand } from "@t3tools/shared/shell";
import { mcpToolResultOutput } from "@t3tools/shared/toolOutput";
import type {
  ChatAttachment,
  OrchestrationV2AppThread,
  OrchestrationV2ConversationMessage,
  OrchestrationV2ExecutionNode,
  ModelSelection,
  OrchestrationV2PlanArtifact,
  OrchestrationV2ProviderCapabilities,
  OrchestrationV2ProviderFailure,
  OrchestrationV2ProviderRetry,
  OrchestrationV2ProviderSession,
  OrchestrationV2ProviderThread,
  OrchestrationV2ProviderTurn,
  OrchestrationV2PlanStep,
  OrchestrationV2RuntimeRequest,
  OrchestrationV2Subagent,
  OrchestrationV2TurnItem,
  ProviderUserInputAnswers,
  ProviderApprovalDecision,
  ProviderApprovalOption,
  ProviderRequestKind,
  ProviderThreadId,
  ProviderTurnId,
  ProviderInstanceId,
  RuntimeMode,
} from "@t3tools/contracts";
import * as CodexClient from "effect-codex-app-server/client";
import * as CodexErrors from "effect-codex-app-server/errors";
import * as CodexSchema from "effect-codex-app-server/schema";
import * as Cause from "effect/Cause";
import * as Context from "effect/Context";
import * as Digest from "effect/crypto/Digest";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as FileSystem from "effect/FileSystem";
import * as Function from "effect/Function";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as Queue from "effect/Queue";
import * as Ref from "effect/Ref";
import * as Schema from "effect/Schema";
import * as Scope from "effect/Scope";
import * as Semaphore from "effect/Semaphore";
import * as Stream from "effect/Stream";
import { ChildProcess, ChildProcessSpawner } from "effect/process";

import { resolveAttachmentPath, resolveAttachmentPathById } from "../../attachmentStore.ts";
import * as PortScanner from "../../preview/PortScanner.ts";
import { getCodexServiceTierOptionValue } from "../../codexModelOptions.ts";
import { ServerConfig } from "../../config.ts";
import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion";
import { buildCodexAdditionalContext } from "../../provider/CodexDeveloperInstructions.ts";
import {
  describeMcpElicitation,
  toMcpElicitationResponse,
} from "../../provider/CodexMcpElicitation.ts";
import {
  materializeCodexShadowHome,
  resolveCodexHomeLayout,
} from "../../provider/Drivers/CodexHomeLayout.ts";
import {
  boundProviderEventForLogging,
  shouldPersistProviderEvent,
} from "../../provider/EventNdjsonLogger.ts";
import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
import { codexAppServerArgs, resolveCodexLaunchArgs } from "../../provider/codexLaunchArgs.ts";
import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment";
import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession";
import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
import {
  MCP_APP_EXTENSION_ID,
  MCP_APP_MIME_TYPE,
  MCP_APP_OUTPUT_KEY,
  MCP_APP_RESOURCE_SCHEME,
} from "@t3tools/shared/mcpApp";
import { snapshotMcpApp } from "../../mcpApps/McpAppSnapshot.ts";
import {
  ProviderAdapterDriverCreateError,
  type ProviderAdapterDriverCreateInput,
} from "@t3tools/provider-core/server/adapterDriver";
import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
import { backgroundWorkNotification } from "@t3tools/provider-core/server/notification";
import {
  makeProviderFailure,
  makeProviderFailureTurnItem,
  makeProviderRetryTurnItem,
} from "@t3tools/provider-core/server/failure";
import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition";
import {
  isProviderNativeImageAttachment,
  providerMessageTextWithAttachmentPaths,
} from "@t3tools/provider-core/server/attachmentPrompt";
import {
  makeSubagentChildThread,
  makeSubagentConversationArtifacts,
  subagentThreadTitle,
} from "@t3tools/provider-core/server/subagentProjection";
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";

const CODEX_PROVIDER = ProviderDriverKind.make("codex");
const CODEX_CAPACITY_MAX_RETRIES = 5;
// How long Stop waits for Codex to end the turn before the user is offered Force stop.
const CODEX_STOP_CONFIRM_TIMEOUT_MS = 10_000;
// Codex can announce a retry and never resume. Recovered retries resume model output within ~75s.
const CODEX_PROVIDER_RETRY_STALL_TIMEOUT = "10 minutes";
export const CODEX_DRIVER_KIND = CODEX_PROVIDER;
export const CODEX_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(CODEX_DRIVER_KIND);

type CodexFileChangeKinds = Readonly<
  Record<string, CodexSchema.V2ItemStartedNotification__PatchChangeKind>
>;

function codexFileChangeApprovalPrompt(input: {
  readonly reason?: string | null;
  readonly grantRoot?: string | null;
  readonly fileChanges?: CodexFileChangeKinds;
}): string | undefined {
  const reason = input.reason?.trim();
  if (reason) return reason;
  const entries = Object.entries(input.fileChanges ?? {}).toSorted(([left], [right]) =>
    left.localeCompare(right),
  );
  if (entries.length > 0) {
    const described = entries.slice(0, 20).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");
  }
  return input.grantRoot?.trim() || undefined;
}

export function normalizeCodexThreadLoadModelSelection(selection: ModelSelection): ModelSelection {
  // Codex applies effort on turn/start; resume discards its pinned request effort.
  const { options, ...rest } = selection;
  const loadOptions = options?.filter((option) => option.id !== "reasoningEffort");
  return { ...rest, ...(loadOptions?.length ? { options: loadOptions } : {}) };
}

// Reasoning-effort changes preserve the tokenizer and context window; other changes need fresh telemetry.
export function canReuseCodexContextUsage(previous: ModelSelection, next: ModelSelection): boolean {
  return modelSelectionsEqual(
    {
      ...previous,
      options: (previous.options ?? []).filter((option) => option.id !== "reasoningEffort"),
    },
    { ...next, options: (next.options ?? []).filter((option) => option.id !== "reasoningEffort") },
  );
}

export function codexProviderTurnTokenUsage(
  tokenUsage: CodexSchema.V2ThreadTokenUsageUpdatedNotification["tokenUsage"],
  updatedAt: string,
) {
  return {
    usedTokens: Math.max(0, tokenUsage.last.totalTokens),
    maxTokens: tokenUsage.modelContextWindow ?? null,
    inputTokens: Math.max(0, tokenUsage.last.inputTokens),
    cachedInputTokens: Math.max(0, tokenUsage.last.cachedInputTokens),
    outputTokens: Math.max(0, tokenUsage.last.outputTokens),
    reasoningOutputTokens: Math.max(0, tokenUsage.last.reasoningOutputTokens),
    updatedAt,
  };
}
const CODEX_ASSISTANT_DELTA_FLUSH_INTERVAL_MS = 50;
// Plan flushes store three copies; use the client's assistant-text cadence.
const CODEX_PLAN_DELTA_FLUSH_INTERVAL_MS = 400;
const CodexBackgroundTerminalTerminateResponse = Schema.Struct({
  terminated: Schema.Boolean,
});
const CodexBackgroundTerminalsListResponse = Schema.Struct({
  data: Schema.Array(
    Schema.Struct({
      processId: Schema.String,
    }),
  ),
  nextCursor: Schema.NullOr(Schema.String),
});
type CodexBackgroundTerminalsListPage = typeof CodexBackgroundTerminalsListResponse.Type;
const decodeCodexBackgroundTerminalTerminateResponse = Schema.decodeUnknownEffect(
  CodexBackgroundTerminalTerminateResponse,
);
const decodeCodexBackgroundTerminalsListResponse = Schema.decodeUnknownEffect(
  CodexBackgroundTerminalsListResponse,
);
const CodexThreadSampleResponse = Schema.Struct({ text: Schema.NullOr(Schema.String) });
const decodeCodexThreadSampleResponse = Schema.decodeUnknownEffect(CodexThreadSampleResponse);
export const CODEX_PROMPT_SUGGESTION_PROMPT =
  "Predict the next message the user will send in this conversation. Reply with only that message, written as the user would type it, in under 20 words. Do not use tools. If there is no clear next step, reply with nothing.";
const CODEX_CLIENT_CAPABILITIES = {
  experimentalApi: true,
  optOutNotificationMethods: ["turn/diff/updated"],
  extensions: { [MCP_APP_EXTENSION_ID]: { mimeTypes: [MCP_APP_MIME_TYPE] } },
} as const;

export const CodexProviderCapabilitiesV2 = {
  sessions: {
    supportsMultipleProviderThreadsPerSession: true,
    supportsModelSwitchInSession: true,
    supportsProviderSwitchingViaHandoff: true,
    supportsRuntimeModeSwitchInSession: true,
    pendingRequestsSurviveRestart: false,
  },
  threads: {
    canCreateEmptyThread: true,
    canReadThreadSnapshot: true,
    canRollbackThread: true,
    canForkThread: true,
    canForkFromTurn: true,
    canForkFromSubagentThread: true,
    exposesNativeThreadId: true,
  },
  turns: {
    exposesNativeTurnId: true,
    emitsTurnStarted: true,
    emitsTurnCompleted: true,
    supportsInterrupt: true,
    supportsActiveSteering: true,
    activeSteeringDuringCompaction: true,
    supportsSteeringByInterruptRestart: true,
    supportsQueuedMessages: true,
    terminalStatusQuality: "strong",
  },
  streaming: {
    streamsAssistantText: true,
    streamsReasoning: true,
    streamsToolOutput: true,
    streamsPlanText: true,
    emitsMessageCompleted: true,
  },
  tools: {
    exposesToolItemIds: true,
    emitsToolStarted: true,
    emitsToolCompleted: true,
    emitsToolOutput: true,
    supportsMcpTools: true,
    supportsDynamicToolCallbacks: true,
  },
  approvals: {
    supportsCommandApproval: true,
    supportsFileReadApproval: true,
    supportsFileChangeApproval: true,
    supportsApplyPatchApproval: true,
    approvalsHaveNativeRequestIds: true,
    approvalCallbacksAreLiveOnly: true,
    approvalsCanOriginateFromSubagents: true,
  },
  planning: {
    emitsPlanUpdated: true,
    emitsTodoList: true,
    emitsProposedPlan: true,
    supportsStructuredQuestions: true,
    planDeltasHaveItemIds: true,
  },
  subagents: {
    supportsSubagents: true,
    exposesSubagentThreadIds: true,
    emitsSubagentLifecycle: true,
    canWaitForSubagents: true,
    canCloseSubagents: true,
    canForkSubagentThread: true,
  },
  context: {
    acceptsSystemContext: true,
    acceptsDeveloperContext: true,
    acceptsSyntheticUserContext: true,
    canGenerateSummaries: true,
    canConsumeHandoffSummaries: true,
    supportsDeltaHandoff: true,
    supportsFullThreadHandoff: true,
    maxRecommendedHandoffChars: null,
  },
  checkpointing: {
    appCanCheckpointFilesystem: true,
    supportsNestedCheckpointScopes: true,
    providerCanRollbackConversation: true,
    providerRollbackReturnsSnapshot: true,
    providerCanReadConversationSnapshot: true,
  },
  identity: {
    nativeThreadIds: "strong",
    nativeTurnIds: "strong",
    nativeItemIds: "strong",
    nativeRequestIds: "strong",
  },
  runtimePolicy: {
    enforcement: "native",
  },
} satisfies OrchestrationV2ProviderCapabilities;

function toProtocolError(
  detail: string,
  payload?: unknown,
): ProviderAdapter.ProviderAdapterProtocolError {
  return new ProviderAdapter.ProviderAdapterProtocolError({
    driver: CODEX_PROVIDER,
    detail,
    ...(payload === undefined ? {} : { payload }),
  });
}

const CODEX_THREAD_IN_USE_MESSAGE =
  "This conversation is open in another Codex client, such as the Codex app, the Codex CLI, or an IDE extension. Codex lets one client write to a conversation at a time. Quit that client, then send your message again.";

const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError);

// Codex holds a per-thread writer lock for as long as a thread stays loaded in
// any app-server sharing the same CODEX_HOME, and reports a conflict this way.
function isCodexActiveWriterError(error: unknown): boolean {
  return (
    isCodexAppServerRequestError(error) &&
    error.code === -32600 &&
    error.errorMessage.includes("already has an active writer")
  );
}

function codexTimestamp(seconds: number | null | undefined): DateTime.Utc {
  return seconds === null || seconds === undefined
    ? DateTime.nowUnsafe()
    : DateTime.makeUnsafe(seconds * 1000);
}

function codexUserMessageText(
  content: ReadonlyArray<CodexSchema.V2ItemCompletedNotification__UserInput>,
): string {
  return content
    .flatMap((item) => (item.type === "text" ? [item.text] : []))
    .join("\n")
    .trim();
}

function mapCodexTurnStatus(
  status: CodexSchema.V2TurnCompletedNotification__TurnStatus,
): OrchestrationV2ProviderTurn["status"] {
  switch (status) {
    case "completed":
      return "completed";
    case "interrupted":
      return "interrupted";
    case "failed":
      return "failed";
    case "inProgress":
      return "running";
  }
}

function providerTurnStatusToTerminal(
  status: OrchestrationV2ProviderTurn["status"],
): Extract<ProviderAdapter.ProviderAdapterV2Event, { type: "turn.terminal" }>["status"] {
  switch (status) {
    case "completed":
      return "completed";
    case "interrupted":
      return "interrupted";
    case "failed":
      return "failed";
    case "cancelled":
      return "cancelled";
    case "pending":
    case "running":
      return "failed";
  }
}

function codexItemStatus(status: "inProgress" | "completed" | "failed" | "declined"): {
  readonly node: OrchestrationV2ExecutionNode["status"];
  readonly turnItem: OrchestrationV2TurnItem["status"];
  readonly completed: boolean;
} {
  switch (status) {
    case "inProgress":
      return { node: "running", turnItem: "running", completed: false };
    case "completed":
      return {
        node: "completed",
        turnItem: "completed",
        completed: true,
      };
    case "failed":
      return { node: "failed", turnItem: "failed", completed: true };
    case "declined":
      return {
        node: "cancelled",
        turnItem: "cancelled",
        completed: true,
      };
  }
}

/**
 * The composers let a skill be typed with any currency sigil (`€review`), but
 * Codex only parses `$name` as a skill mention.
 */
function codexSkillMentionText(text: string): string {
  return text.replace(SKILL_MENTION_PATTERN, "$1$$$2");
}

const BACKGROUND_COMMAND_DETAIL_COMMAND_MAX_LENGTH = 200;
const BACKGROUND_COMMAND_DETAIL_OUTPUT_TAIL_MAX_LENGTH = 1_000;

function codexBackgroundCommandDetail(item: {
  readonly command: string;
  readonly exitCode?: number | null | undefined;
  readonly aggregatedOutput?: string | null | undefined;
}): string {
  // Codex rejects a turn/start with a lone surrogate and never answers it, so cut on code points.
  let commandEnd = BACKGROUND_COMMAND_DETAIL_COMMAND_MAX_LENGTH;
  const finalCommandUnit = item.command.charCodeAt(commandEnd - 1);
  if (finalCommandUnit >= 0xd800 && finalCommandUnit <= 0xdbff) commandEnd -= 1;
  const command =
    item.command.length > BACKGROUND_COMMAND_DETAIL_COMMAND_MAX_LENGTH
      ? `${item.command.slice(0, commandEnd)}...`
      : item.command;
  const exit =
    item.exitCode === null || item.exitCode === undefined ? "" : ` (exit ${item.exitCode})`;
  const output = (item.aggregatedOutput ?? "").trimEnd();
  let outputStart = output.length - BACKGROUND_COMMAND_DETAIL_OUTPUT_TAIL_MAX_LENGTH;
  const firstOutputUnit = output.charCodeAt(outputStart);
  if (firstOutputUnit >= 0xdc00 && firstOutputUnit <= 0xdfff) outputStart += 1;
  const outputTail = outputStart > 0 ? `...${output.slice(outputStart)}` : output;
  const header = `Background command completed${exit}: ${command}`;
  return outputTail.length === 0 ? header : `${header}\n\nOutput tail:\n${outputTail}`;
}

// Codex spawns can omit the prompt, which registers as an empty string.
function codexNativeSubagentLabel(task: OrchestrationV2Subagent): string | undefined {
  return task.title?.trim() || task.prompt.trim() || undefined;
}

function codexNativeSubagentDetail(task: OrchestrationV2Subagent): string {
  const header = `Subagent ${task.status}: ${codexNativeSubagentLabel(task) ?? task.id}`;
  return task.result == null || task.result.length === 0
    ? header
    : `${header}\n\nResult:\n${task.result}`;
}

export interface CodexDynamicToolProjection extends McpToolPresentation {
  readonly toolName: string;
  readonly title?: string;
  readonly input: unknown;
  readonly output?: unknown;
  readonly status: OrchestrationV2TurnItem["status"];
}

function codexMcpToolOutput(
  item: Extract<CodexDynamicToolItem, { readonly type: "mcpToolCall" }>,
): unknown | undefined {
  const resultOutput =
    item.result === null || item.result === undefined
      ? undefined
      : mcpToolResultOutput(item.result);

  if (item.error === null || item.error === undefined) {
    return resultOutput;
  }
  return resultOutput === undefined
    ? { error: item.error.message }
    : { error: item.error.message, result: resultOutput };
}

// A hung MCP resource read must not hold run ingestion open.
const MCP_APP_CAPTURE_TIMEOUT = "20 seconds";

function codexMcpAppResourceUri(item: CodexDynamicToolItem): string | undefined {
  if (item.type !== "mcpToolCall") return undefined;
  const uri = item.mcpAppUi?.resourceUri ?? item.mcpAppResourceUri ?? item.appContext?.resourceUri;
  return typeof uri === "string" && uri.startsWith(MCP_APP_RESOURCE_SCHEME) ? uri : undefined;
}

const CODEX_IMAGE_DATA_URL = /^data:(image\/[a-z0-9.+-]+);base64,([A-Za-z0-9+/=]+)$/i;

export function codexDynamicToolOutput(
  item: Extract<CodexDynamicToolItem, { readonly type: "dynamicToolCall" }>,
): unknown | undefined {
  if (item.contentItems !== null && item.contentItems !== undefined) {
    return item.contentItems.map((content) => {
      if (content.type !== "inputImage") return content;
      const match = CODEX_IMAGE_DATA_URL.exec(content.imageUrl);
      return match?.[1] && match[2]
        ? { type: "image", mimeType: match[1].toLowerCase(), data: match[2] }
        : content;
    });
  }
  return item.success === false ? { success: false } : undefined;
}

const projectCodexDynamicToolItem = Effect.fn("CodexAdapterV2.projectDynamicToolItem")(function* (
  item: CodexDynamicToolItem,
): Effect.fn.Return<CodexDynamicToolProjection, never, Digest.Digest> {
  const output =
    item.type === "mcpToolCall" ? codexMcpToolOutput(item) : codexDynamicToolOutput(item);
  const toolName =
    item.type === "mcpToolCall"
      ? `${item.server}.${item.tool}`
      : [trimText(item.namespace), item.tool].filter(Boolean).join(".");
  const presentation: McpToolPresentation =
    item.type === "mcpToolCall" ? yield* mcpToolPresentation(item) : {};
  const title =
    dynamicToolTitle(toolName, item.arguments, presentation.title) ?? presentation.title;
  const toolSurface = presentation.toolSurface ?? dynamicToolSurface(toolName, item.arguments);
  const toolSource =
    toolSurface === "computer" && presentation.toolSource?.kind !== "computer"
      ? COMPUTER_USE_TOOL_SOURCE
      : presentation.toolSource;
  const projection: CodexDynamicToolProjection = {
    ...presentation,
    ...(toolSurface ? { toolSurface } : {}),
    ...(toolSource ? { toolSource } : {}),
    toolName,
    ...(title ? { title } : {}),
    input: item.arguments,
    status: codexItemStatus(item.status).turnItem,
  };
  return output === undefined ? projection : { ...projection, output };
});

function codexNativeItemRef(nativeItemId: string) {
  return {
    driver: CODEX_PROVIDER,
    nativeId: nativeItemId,
    strength: "strong" as const,
  };
}

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

function nonEmptyText(value: string | null | undefined, fallback: string): string {
  return trimText(value) ?? fallback;
}

function codexPlanStepStatus(
  status: CodexSchema.V2TurnPlanUpdatedNotification__TurnPlanStepStatus,
): OrchestrationV2PlanStep["status"] {
  switch (status) {
    case "completed":
      return "completed";
    case "inProgress":
      return "running";
    case "pending":
      return "pending";
  }
}

function approvalDecisionToLegacyReviewDecision(
  decision: ProviderApprovalDecision,
): CodexSchema.ExecCommandApprovalResponse__ReviewDecision {
  switch (decision) {
    case "accept":
      return "approved";
    case "acceptForSession":
    case "acceptAlways":
      return "approved_for_session";
    case "decline":
      return { denied: { rejection: "User declined the request." } };
    case "cancel":
      return "abort";
  }
}

function providerRequestKindFromPermissions(
  permissions: CodexSchema.PermissionsRequestApprovalParams["permissions"],
): ProviderRequestKind {
  // Codex leaves the legacy `read`/`write` lists out when a path is a glob or
  // special path, and is replacing them with `entries`.
  const requests = (access: "read" | "write") =>
    (permissions.fileSystem?.[access]?.length ?? 0) > 0 ||
    permissions.fileSystem?.entries?.some((entry) => entry.access === access);
  if (requests("write")) {
    return "file-change";
  }
  if (requests("read")) {
    return "file-read";
  }
  return "command";
}

function permissionsResponseFromDecision(input: {
  readonly decision: ProviderApprovalDecision;
  readonly permissions: CodexSchema.PermissionsRequestApprovalParams["permissions"];
}): CodexSchema.PermissionsRequestApprovalResponse {
  if (input.decision !== "accept" && input.decision !== "acceptForSession") {
    return { permissions: {}, scope: "turn" };
  }

  return {
    permissions: input.permissions,
    scope: input.decision === "acceptForSession" ? "session" : "turn",
  };
}

function answerValueToStrings(value: unknown): ReadonlyArray<string> {
  if (Array.isArray(value)) {
    return value.map((item) => String(item));
  }
  if (typeof value === "string") {
    return [value];
  }
  if (value === null || value === undefined) {
    return [];
  }
  if (typeof value === "number" || typeof value === "boolean") {
    return [String(value)];
  }
  return [JSON.stringify(value)];
}

function toCodexUserInputAnswers(
  answers: ProviderUserInputAnswers,
  nativeQuestionIds: ReadonlyMap<string, string>,
): CodexSchema.ToolRequestUserInputResponse["answers"] {
  return Object.fromEntries(
    Object.entries(answers).flatMap(([questionId, value]) => {
      const nativeId = nativeQuestionIds.get(questionId);
      return nativeId === undefined
        ? []
        : [[nativeId, { answers: [...answerValueToStrings(value)] }]];
    }),
  );
}

function compactStrings(values: ReadonlyArray<string | null | undefined>): ReadonlyArray<string> {
  return values.flatMap((value) => {
    const trimmed = trimText(value);
    return trimmed === undefined ? [] : [trimmed];
  });
}

function webSearchPatterns(item: CodexWebSearchItem): ReadonlyArray<string> {
  if (item.action === null || item.action === undefined) {
    return compactStrings([item.query]);
  }

  switch (item.action.type) {
    case "search":
      return compactStrings([...(item.action.queries ?? []), item.action.query, item.query]);
    case "openPage":
      return compactStrings([item.action.url, item.query]);
    case "findInPage":
      return compactStrings([item.action.pattern, item.action.url, item.query]);
    case "other":
      return compactStrings([item.query]);
  }
}

const decodeTurnApprovalPolicy = Schema.decodeUnknownEffect(
  Schema.Union([CodexSchema.V2TurnStartParams__AskForApproval, Schema.Null]),
);
const decodeTurnSandboxPolicy = Schema.decodeUnknownEffect(
  Schema.Union([CodexSchema.V2TurnStartParams__SandboxPolicy, Schema.Null]),
);
const decodeTurnReasoningEffort = Schema.decodeUnknownEffect(
  Schema.Union([CodexSchema.V2TurnStartParams__ReasoningEffort, Schema.Null]),
);

type CodexTurnStartParams = typeof CodexClient.V2TurnStartParams.Type;
const decodeCodexTurnStartParams = Schema.decodeUnknownEffect(CodexClient.V2TurnStartParams);
const isProviderAdapterRuntimeRequestResponseError = Schema.is(
  ProviderAdapter.ProviderAdapterRuntimeRequestResponseError,
);

function codexRuntimeModeTurnDefaults(runtimeMode: RuntimeMode): {
  readonly approvalPolicy: CodexSchema.V2TurnStartParams__AskForApproval;
  readonly approvalsReviewer: CodexSchema.V2TurnStartParams__ApprovalsReviewer;
  readonly sandboxPolicy: CodexSchema.V2TurnStartParams__SandboxPolicy;
} {
  switch (runtimeMode) {
    case "approval-required":
      return {
        approvalPolicy: "untrusted",
        approvalsReviewer: "user",
        sandboxPolicy: {
          type: "readOnly",
        },
      };
    case "auto-accept-edits":
      return {
        approvalPolicy: "on-request",
        approvalsReviewer: "user",
        sandboxPolicy: {
          type: "workspaceWrite",
        },
      };
    case "auto":
      return {
        approvalPolicy: "on-request",
        approvalsReviewer: "auto_review",
        sandboxPolicy: {
          type: "workspaceWrite",
        },
      };
    case "full-access":
      return {
        approvalPolicy: "never",
        approvalsReviewer: "user",
        sandboxPolicy: {
          type: "dangerFullAccess",
        },
      };
  }
}

const isCodexCyberAccessProgram = Schema.is(CodexSchema.V2TurnStartParams__CyberAccessProgram);

function codexCyberAccess(
  models: ReadonlyArray<ServerProviderModel>,
  model: string,
): { readonly defaultProgram: string | undefined } | undefined {
  const descriptor = models
    .find((candidate) => candidate.slug === model)
    ?.capabilities?.optionDescriptors?.find((candidate) => candidate.id === "cyberAccessProgram");
  return descriptor?.type === "select" ? { defaultProgram: descriptor.currentValue } : undefined;
}

export const buildCodexTurnStartParams = Effect.fn("CodexAdapterV2.buildCodexTurnStartParams")(
  function* (input: {
    readonly nativeThreadId: string;
    readonly codexInput: ReadonlyArray<CodexSchema.V2TurnStartParams__UserInput>;
    readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy;
    readonly modelSelection: ModelSelection;
    readonly hasT3Mcp?: boolean;
    readonly browserToolsAvailable?: boolean;
    readonly deviceToolsAvailable?: boolean;
    /** ChatGPT token sharing does not accept service tiers. */
    readonly omitServiceTier?: boolean;
    readonly appContext?: ProviderAdapter.ProviderAdapterV2TurnInput["appContext"];
    /** Set only for a model that offers Cyber Access; custom models never do. */
    readonly cyberAccess?: { readonly defaultProgram: string | undefined } | undefined;
    /** The user's `model_reasoning_summary` from the Codex config, when set. */
    readonly configuredReasoningSummary?: CodexSchema.V2TurnStartParams__ReasoningSummary;
  }) {
    const runtimeModeDefaults = codexRuntimeModeTurnDefaults(input.runtimePolicy.runtimeMode);
    const approvalPolicy =
      input.runtimePolicy.approvalPolicy === undefined
        ? runtimeModeDefaults.approvalPolicy
        : yield* decodeTurnApprovalPolicy(input.runtimePolicy.approvalPolicy);
    const sandboxPolicy =
      input.runtimePolicy.sandboxPolicy === undefined
        ? runtimeModeDefaults.sandboxPolicy
        : yield* decodeTurnSandboxPolicy(input.runtimePolicy.sandboxPolicy);
    const selectedEffort =
      getModelSelectionStringOptionValue(input.modelSelection, "reasoningEffort") ??
      codexDefaultReasoningEffortOverride(input.modelSelection.model);
    const effort =
      selectedEffort === undefined ? undefined : yield* decodeTurnReasoningEffort(selectedEffort);
    const serviceTier =
      input.omitServiceTier === true
        ? undefined
        : getCodexServiceTierOptionValue(input.modelSelection);
    // A chosen program goes as is, even one the model lacks, so Codex rejects it visibly.
    const cyberAccessProgram =
      input.cyberAccess === undefined
        ? undefined
        : [
            getModelSelectionStringOptionValue(input.modelSelection, "cyberAccessProgram"),
            input.cyberAccess.defaultProgram,
          ].find(isCodexCyberAccessProgram);
    // MCP-written app context must be quoted user input, not developer instructions.
    const appContext = Object.fromEntries(
      (input.appContext ?? []).map((entry) => [
        entry.key,
        { kind: "untrusted" as const, value: entry.text },
      ]),
    );
    const t3Context =
      input.hasT3Mcp === true
        ? buildCodexAdditionalContext(
            { model: input.modelSelection.model, reasoningEffort: effort ?? undefined },
            {
              browser: input.browserToolsAvailable ?? true,
              device: input.deviceToolsAvailable ?? false,
            },
          )
        : undefined;
    const additionalContext =
      t3Context === undefined && Object.keys(appContext).length === 0
        ? undefined
        : { ...t3Context, ...appContext };
    return yield* decodeCodexTurnStartParams({
      threadId: input.nativeThreadId,
      input: input.codexInput,
      ...(additionalContext ? { additionalContext } : {}),
      cwd: input.runtimePolicy.cwd,
      model: input.modelSelection.model,
      // Catalogues can default summaries to "none", and turn/start overrides the user's config,
      // so pass theirs through or `auto`: some providers (Amazon Bedrock) reject `detailed`.
      summary: input.configuredReasoningSummary ?? "auto",
      // Omitting the reviewer on resume preserves the previous Auto reviewer.
      approvalsReviewer: runtimeModeDefaults.approvalsReviewer,
      ...(approvalPolicy === undefined ? {} : { approvalPolicy }),
      ...(sandboxPolicy === undefined ? {} : { sandboxPolicy }),
      ...(effort === undefined ? {} : { effort }),
      ...(serviceTier === undefined ? {} : { serviceTier }),
      // Codex keeps no per-thread choice for this, so every turn must carry it.
      ...(cyberAccessProgram === undefined ? {} : { cyberAccessProgram }),
      // Codex supplies its own mode instructions; send both modes to clear Plan on exit.
      collaborationMode: {
        mode: input.runtimePolicy.interactionMode,
        settings: {
          model: input.modelSelection.model,
          ...(effort === undefined ? {} : { reasoning_effort: effort }),
        },
      },
    });
  },
);

function providerSession(input: {
  readonly providerSessionId: OrchestrationV2ProviderSession["id"];
  readonly providerInstanceId: ProviderInstanceId;
  readonly cwd: string | null;
  readonly model: string;
  readonly now: DateTime.Utc;
}): OrchestrationV2ProviderSession {
  return {
    id: input.providerSessionId,
    driver: CODEX_PROVIDER,
    providerInstanceId: input.providerInstanceId,
    status: "ready",
    cwd: input.cwd ?? process.cwd(),
    model: input.model,
    capabilities: CodexProviderCapabilitiesV2,
    createdAt: input.now,
    updatedAt: input.now,
    lastError: null,
  };
}

const getNativeThreadId = Effect.fn("CodexAdapterV2.getNativeThreadId")(function* (
  providerThread: OrchestrationV2ProviderThread,
) {
  const nativeThreadId = providerThread.nativeThreadRef?.nativeId;
  if (nativeThreadId === undefined || nativeThreadId === null) {
    return yield* toProtocolError(
      `Provider thread ${providerThread.id} is missing a native Codex thread id.`,
    );
  }
  return nativeThreadId;
});

function providerThreadFromCodexThread(input: {
  readonly appThreadId: ThreadId | null;
  readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
  readonly ownerNodeId: OrchestrationV2ProviderThread["ownerNodeId"];
  readonly providerSessionId: OrchestrationV2ProviderThread["providerSessionId"];
  readonly providerInstanceId: ProviderInstanceId;
  readonly thread: {
    readonly createdAt: number;
    readonly forkedFromId?: string | null;
    readonly id: string;
    readonly updatedAt: number;
  };
  readonly forkedFrom?: OrchestrationV2ProviderThread["forkedFrom"];
}): OrchestrationV2ProviderThread {
  return {
    id: input.idAllocator.derive.providerThread({
      driver: CODEX_PROVIDER,
      nativeThreadId: input.thread.id,
    }),
    driver: CODEX_PROVIDER,
    providerInstanceId: input.providerInstanceId,
    providerSessionId: input.providerSessionId,
    appThreadId: input.appThreadId,
    ownerNodeId: input.ownerNodeId,
    nativeThreadRef: {
      driver: CODEX_PROVIDER,
      nativeId: input.thread.id,
      strength: "strong" as const,
    },
    nativeConversationHeadRef: null,
    status: "idle",
    firstRunOrdinal: null,
    lastRunOrdinal: null,
    handoffIds: [],
    forkedFrom: input.forkedFrom ?? null,
    createdAt: codexTimestamp(input.thread.createdAt),
    updatedAt: codexTimestamp(input.thread.updatedAt),
  };
}

// /goal control turns never reach Codex and must not count toward native rollback.
const isTerminalProviderTurn = (turn: OrchestrationV2ProviderTurn): boolean =>
  turn.nativeTurnRef !== null &&
  (turn.status === "completed" ||
    turn.status === "interrupted" ||
    turn.status === "failed" ||
    turn.status === "cancelled");

class CodexGoalObjectiveTooLongError extends Schema.TaggedError<CodexGoalObjectiveTooLongError>()(
  "CodexGoalObjectiveTooLongError",
  { length: Schema.Number, limit: Schema.Number },
) {
  override get message() {
    return `The goal objective is ${this.length} characters; Codex supports at most ${this.limit}. Shorten it and send /goal again. The current goal was not changed.`;
  }
}

function validateCodexGoalObjective(objective: string) {
  let length = 0;
  for (const _ of objective) length++;
  return length > CodexSchema.MAX_THREAD_GOAL_OBJECTIVE_CHARS
    ? Effect.fail(
        new CodexGoalObjectiveTooLongError({
          length,
          limit: CodexSchema.MAX_THREAD_GOAL_OBJECTIVE_CHARS,
        }),
      )
    : Effect.void;
}

type CodexThreadGoal = CodexSchema.V2ThreadGoalUpdatedNotification["goal"];

const CODEX_GOAL_STATUSES = {
  active: "active",
  paused: "paused",
  blocked: "blocked",
  usageLimited: "usage_limited",
  budgetLimited: "budget_limited",
  complete: "complete",
} as const satisfies Record<CodexThreadGoal["status"], OrchestrationV2ProviderGoal["status"]>;

function providerGoalFromCodex(goal: CodexThreadGoal): OrchestrationV2ProviderGoal | null {
  const objective = goal.objective.trim();
  if (objective.length === 0) return null;
  return {
    objective,
    status: CODEX_GOAL_STATUSES[goal.status],
    tokensUsed: Math.max(0, goal.tokensUsed),
    tokenBudget: goal.tokenBudget ?? null,
    timeUsedSeconds: Math.max(0, goal.timeUsedSeconds),
  };
}

const providerGoalsEqual = Schema.toEquivalence(Schema.NullOr(OrchestrationV2ProviderGoal));

function describeCodexGoal(goal: OrchestrationV2ProviderGoal): string {
  return `Goal ${goal.status.replace("_", " ")}: ${goal.objective}`;
}

// Codex starts the next goal turn milliseconds after the last one completes.
const CODEX_GOAL_CONTINUATION_GRACE = "5 seconds";

const providerTurnsForThread = (
  providerTurns: ReadonlyArray<OrchestrationV2ProviderTurn>,
  providerThread: OrchestrationV2ProviderThread,
): ReadonlyArray<OrchestrationV2ProviderTurn> =>
  providerTurns.filter((turn) => turn.providerThreadId === providerThread.id);

const countTerminalTurnsAfterBoundary = (
  providerTurns: ReadonlyArray<OrchestrationV2ProviderTurn>,
  providerTurnId: ProviderTurnId,
): number | null => {
  const boundaryTurn = providerTurns.find((turn) => turn.id === providerTurnId);
  if (boundaryTurn === undefined) {
    return null;
  }

  return providerTurns.filter(
    (turn) => turn.ordinal > boundaryTurn.ordinal && isTerminalProviderTurn(turn),
  ).length;
};

const resolveCodexForkRollbackTurnCount = Effect.fn("CodexAdapterV2.resolveForkRollbackTurnCount")(
  function* (input: ProviderAdapter.ProviderAdapterV2ForkThreadInput) {
    if (input.providerTurnId === undefined || input.sourceProviderTurns === undefined) {
      return 0;
    }

    const rollbackTurnCount = countTerminalTurnsAfterBoundary(
      providerTurnsForThread(input.sourceProviderTurns, input.sourceProviderThread),
      input.providerTurnId,
    );
    if (rollbackTurnCount === null) {
      return yield* new ProviderAdapter.ProviderAdapterForkThreadError({
        driver: CODEX_PROVIDER,
        providerThreadId: input.sourceProviderThread.id,
        cause: `Cannot fork Codex thread from provider turn ${input.providerTurnId}: source turn was not found in provider thread ${input.sourceProviderThread.id}.`,
      });
    }

    return rollbackTurnCount;
  },
);

/** Codex lastTurnId is inclusive, atomically omitting later turns even in paginated threads. */
export const resolveCodexForkBoundary = Effect.fn("CodexAdapterV2.resolveForkBoundary")(function* (
  input: ProviderAdapter.ProviderAdapterV2ForkThreadInput,
) {
  const rollbackTurnCount = yield* resolveCodexForkRollbackTurnCount(input);
  if (input.providerTurnId === undefined || input.sourceProviderTurns === undefined) {
    return { lastTurnId: undefined, rollbackTurnCount };
  }

  const boundaryTurn = providerTurnsForThread(
    input.sourceProviderTurns,
    input.sourceProviderThread,
  ).find((turn) => turn.id === input.providerTurnId);
  const nativeTurnId = boundaryTurn?.nativeTurnRef?.nativeId;
  if (nativeTurnId === null || nativeTurnId === undefined) {
    return { lastTurnId: undefined, rollbackTurnCount };
  }

  return { lastTurnId: nativeTurnId, rollbackTurnCount: 0 };
});

/** Generated thread/read schemas omit historyMode; probe through the raw request channel. */
const CodexThreadHistoryMetadata = Schema.Struct({
  thread: Schema.Struct({
    historyMode: Schema.optionalKey(Schema.Literals(["legacy", "paginated"])),
    status: Schema.optionalKey(Schema.Struct({ type: Schema.String })),
  }),
});
const decodeCodexThreadHistoryMetadata = Schema.decodeUnknownEffect(CodexThreadHistoryMetadata);

const readCodexThreadHistoryMetadata = Effect.fn("CodexAdapterV2.readThreadHistoryMetadata")(
  function* (
    raw: Pick<CodexClient.CodexAppServerClient["Service"]["raw"], "request">,
    threadId: string,
  ) {
    const response = yield* raw.request("thread/read", { threadId, includeTurns: false });
    const metadata = yield* decodeCodexThreadHistoryMetadata(response).pipe(
      Effect.mapError((error) =>
        CodexErrors.CodexAppServerRequestError.invalidPayload(
          "thread/read",
          "decode-payload",
          error,
        ),
      ),
    );
    return {
      historyMode: metadata.thread.historyMode,
      loaded: metadata.thread.status?.type !== "notLoaded",
    };
  },
);

export const resolveCodexRollbackTurnCount = Effect.fn("CodexAdapterV2.resolveRollbackTurnCount")(
  function* (input: ProviderAdapter.ProviderAdapterV2RollbackThreadInput) {
    const providerTurns = input.providerThreadTurns;
    switch (input.target.type) {
      case "thread_start":
        return providerTurns.filter(isTerminalProviderTurn).length;
      case "provider_turn": {
        if (input.target.providerTurn.providerThreadId !== input.providerThread.id) {
          return yield* new ProviderAdapter.ProviderAdapterRollbackThreadError({
            driver: CODEX_PROVIDER,
            providerThreadId: input.providerThread.id,
            cause: `Cannot roll back Codex thread ${input.providerThread.id} to provider turn ${input.target.providerTurn.id}: target turn belongs to provider thread ${input.target.providerTurn.providerThreadId}.`,
          });
        }

        const rollbackTurnCount = countTerminalTurnsAfterBoundary(
          providerTurns,
          input.target.providerTurn.id,
        );
        if (rollbackTurnCount === null) {
          return yield* new ProviderAdapter.ProviderAdapterRollbackThreadError({
            driver: CODEX_PROVIDER,
            providerThreadId: input.providerThread.id,
            cause: `Cannot roll back Codex thread ${input.providerThread.id} to provider turn ${input.target.providerTurn.id}: target turn was not found in durable provider turn history.`,
          });
        }

        return rollbackTurnCount;
      }
    }
  },
);

function parseCodexRetryProgress(
  message: string,
): Pick<OrchestrationV2ProviderRetry, "attempt" | "maxAttempts"> | null {
  const match = /\b(\d+)\s*\/\s*(\d+)\b/u.exec(message);
  if (match?.[1] === undefined || match[2] === undefined) {
    return null;
  }
  const attempt = Number.parseInt(match[1], 10);
  const maxAttempts = Number.parseInt(match[2], 10);
  if (attempt < 1 || maxAttempts < 1) {
    return null;
  }
  return { attempt, maxAttempts };
}

function codexErrorInfoCode(value: unknown): string | null {
  if (typeof value === "string") {
    return value;
  }
  if (typeof value !== "object" || value === null) {
    return null;
  }
  return Object.keys(value)[0] ?? null;
}

type CodexErrorInfo = CodexSchema.ServerNotification__CodexErrorInfo;

/**
 * Retry exhaustion only says why it failed through the last attempt's HTTP
 * status. Exhaustion on 429 is a usage or rate limit, so it keeps the Limited
 * state and resume controls; other statuses stay provider or transport errors.
 */
function codexExhaustionHttpStatus(errorInfo: CodexErrorInfo | null | undefined): number | null {
  return typeof errorInfo === "object" &&
    errorInfo !== null &&
    "responseTooManyFailedAttempts" in errorInfo
    ? (errorInfo.responseTooManyFailedAttempts.httpStatusCode ?? null)
    : null;
}

function codexErrorHttpStatus(errorInfo: CodexErrorInfo | null | undefined): number | null {
  const detail =
    typeof errorInfo === "object" && errorInfo !== null ? Object.values(errorInfo)[0] : undefined;
  return typeof detail === "object" &&
    detail !== null &&
    "httpStatusCode" in detail &&
    typeof detail.httpStatusCode === "number"
    ? detail.httpStatusCode
    : null;
}

function codexUsageLimitError(
  code: string | null,
  errorInfo: CodexErrorInfo | null | undefined,
): boolean {
  return (
    code === "usageLimitExceeded" ||
    code === "rateLimitExceeded" ||
    codexExhaustionHttpStatus(errorInfo) === 429
  );
}

/**
 * ChatGPT sharing reports its own reason inside Codex's generic turn error, so
 * the code it names replaces Codex's, and clients pick a recovery from it. Its
 * allowance limits get the usage-limit recovery; other sharing failures keep their class.
 */
function withCodexManagedFailure(
  failure: OrchestrationV2ProviderFailure,
  nativeError: unknown,
): OrchestrationV2ProviderFailure {
  const managed = classifyCodexManagedError(nativeError);
  return managed === undefined
    ? failure
    : {
        ...failure,
        message: managed.message,
        code: managed.code,
        ...(managed.usageLimit ? { class: "usage_limit" } : {}),
      };
}

function failureResetAt(
  failure: OrchestrationV2ProviderFailure,
  snapshot: CodexRateLimitSnapshot | undefined,
) {
  // Sharing caps have no reset timestamp in the ordinary Codex allowance snapshot.
  return classifyCodexManagedError(failure.code)?.usageLimit
    ? null
    : codexUsageLimitResetAt(snapshot);
}

interface ActiveCodexTurnContext {
  hookStopReason?: string;
  capacityRetryAttempt?: number;
  capacityRetryPending?: Deferred.Deferred<ActiveCodexTurnContext | undefined>;
  closedCommandItemIds?: ReadonlySet<string>;
  latestProviderFailure?: {
    readonly nativeMessage: string;
    readonly nativeCode: string | null;
    readonly failure: OrchestrationV2ProviderFailure;
  };
  readonly nativeStartReady?: Deferred.Deferred<void>;
  readonly input: ProviderAdapter.ProviderAdapterV2TurnInput;
  readonly projectionAppThread: OrchestrationV2AppThread;
  readonly projectionThreadId: ThreadId;
  readonly projectionRunId: ProviderAdapter.ProviderAdapterV2TurnInput["runId"] | null;
  readonly nativeTurnId: string;
  readonly providerThread: OrchestrationV2ProviderThread;
  readonly providerTurnId: ProviderTurnId;
  readonly providerTurnOrdinal: number;
  readonly providerNodeId: OrchestrationV2ExecutionNode["id"];
  readonly providerNodeStartedAt: DateTime.Utc | null;
  readonly itemParentNodeId: OrchestrationV2ExecutionNode["id"];
  readonly rootNodeId: OrchestrationV2ExecutionNode["id"];
  readonly subagent: CodexSubagentThreadContext | null;
  readonly startedAt: DateTime.Utc;
  readonly runningCompactionItemIds: Set<string>;
  // Modern approval requests omit paths; keep the kinds without retaining patch bodies.
  readonly fileChangeKindsByItem: Map<string, CodexFileChangeKinds>;
  readonly messagePhases: Map<string, "commentary" | "partial_answer" | "final_answer">;
  readonly itemPositions: Map<
    string,
    { readonly ordinal: number; readonly startedAt: DateTime.Utc }
  >;
}

interface ActiveCodexProviderRetry {
  readonly context?: ActiveCodexTurnContext;
  readonly nativeMessage: string;
  readonly nativeCode: string | null;
  readonly retry: OrchestrationV2ProviderRetry;
  readonly failure: OrchestrationV2ProviderFailure;
  readonly startedAt: DateTime.Utc;
  readonly itemOrdinal: number;
}

function codexRetryTurnItem(
  activeContext: ActiveCodexTurnContext,
  state: ActiveCodexProviderRetry,
  status: Parameters<typeof makeProviderRetryTurnItem>[0]["status"],
  updatedAt: DateTime.Utc,
) {
  // A continuation keeps the retry row attached to the native turn that failed.
  const context = state.context ?? activeContext;
  return makeProviderRetryTurnItem({
    driver: CODEX_PROVIDER,
    threadId: context.projectionThreadId,
    runId: context.projectionRunId,
    nodeId: context.providerNodeId,
    providerThreadId: context.providerThread.id,
    providerTurnId: context.providerTurnId,
    itemOrdinal: state.itemOrdinal,
    failure: state.failure,
    retry: state.retry,
    status,
    startedAt: state.startedAt,
    updatedAt,
  });
}

interface TrackedRunningCommandItem {
  readonly id: string;
  readonly command: string;
  readonly aggregatedOutput?: string;
  readonly processId?: string;
}

function isPersistentCodexDynamicTool(item: CodexDynamicToolItem): boolean {
  const input = item.arguments;
  if (input === null || typeof input !== "object" || Array.isArray(input)) {
    return false;
  }
  return Reflect.get(input, "persistent") === true;
}

type CodexRootTerminalEvent = Extract<
  ProviderAdapter.ProviderAdapterV2Event,
  { readonly type: "turn.terminal" }
>;

interface CodexGoalHold {
  readonly context: ActiveCodexTurnContext;
  readonly event: Extract<CodexRootTerminalEvent, { readonly failure: null }>;
  readonly completedTurn: OrchestrationV2ProviderTurn;
  readonly next: Deferred.Deferred<ActiveCodexTurnContext | undefined>;
}

interface DeferredCodexRootTerminal {
  readonly context: ActiveCodexTurnContext;
  readonly event: CodexRootTerminalEvent;
}

interface CodexSubagentThreadContext {
  parentContext: ActiveCodexTurnContext;
  readonly providerThread: OrchestrationV2ProviderThread;
  readonly childThread: OrchestrationV2AppThread;
  readonly subagentNodeId: OrchestrationV2ExecutionNode["id"];
  readonly childRootNodeId: OrchestrationV2ExecutionNode["id"];
  readonly childThreadId: ThreadId;
  readonly nativeTurnIds: Set<string>;
  startedAt: DateTime.Utc;
  readonly turnItemId: OrchestrationV2TurnItem["id"];
  readonly turnItemOrdinal: number;
  task: OrchestrationV2Subagent;
}

const approvalOwnerCodexTurn = (
  context: ActiveCodexTurnContext,
): {
  readonly owner: ActiveCodexTurnContext;
  readonly subagent: CodexSubagentThreadContext | null;
} => {
  let owner = context;
  let subagent: CodexSubagentThreadContext | null = null;
  while (owner.subagent !== null) {
    subagent = owner.subagent;
    owner = owner.subagent.parentContext;
  }
  return { owner, subagent };
};

const isDescendantCodexTurn = (
  candidate: ActiveCodexTurnContext,
  ancestor: ActiveCodexTurnContext,
): boolean => {
  let parent = candidate.subagent?.parentContext;
  while (parent !== undefined) {
    if (parent === ancestor) {
      return true;
    }
    parent = parent.subagent?.parentContext;
  }
  return false;
};

interface PendingCodexSubagentTurnStarted {
  readonly nativeTurnId: string;
  readonly startedAt: DateTime.Utc;
}

type PendingCodexRuntimeRequest =
  | {
      readonly type: "approval";
      readonly requestKind: ProviderRequestKind;
      readonly decision: Deferred.Deferred<ProviderApprovalDecision, never>;
    }
  | {
      readonly type: "user_input";
      readonly answers: Deferred.Deferred<ProviderUserInputAnswers, never>;
    };

type CodexWebSearchItem = {
  readonly id: string;
  readonly type: "webSearch";
  readonly query?: string | null;
  readonly action?:
    | CodexSchema.V2ItemStartedNotification__WebSearchAction
    | CodexSchema.V2ItemCompletedNotification__WebSearchAction
    | null;
};

export type CodexDynamicToolItem = Extract<
  | CodexSchema.V2ItemStartedNotification__ThreadItem
  | CodexSchema.V2ItemCompletedNotification__ThreadItem,
  { readonly type: "mcpToolCall" | "dynamicToolCall" }
>;

type CodexCollabAgentToolCallItem = Extract<
  CodexSchema.V2ItemCompletedNotification__ThreadItem,
  { readonly type: "collabAgentToolCall" }
>;

type CodexSubAgentActivityItem = Extract<
  | CodexSchema.V2ItemStartedNotification__ThreadItem
  | CodexSchema.V2ItemCompletedNotification__ThreadItem,
  { readonly type: "subAgentActivity" }
>;

export interface CodexAppServerClientFactoryShape {
  readonly open: (input: {
    readonly instanceId: ProviderInstanceId;
    readonly threadId: ThreadId;
    readonly providerSessionId: OrchestrationV2ProviderSession["id"];
    readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy;
    readonly settings: CodexSettings;
    readonly environment: NodeJS.ProcessEnv;
  }) => Effect.Effect<
    CodexClient.CodexAppServerClient["Service"],
    ProviderAdapter.ProviderAdapterOpenSessionError,
    Scope.Scope
  >;
}

export class CodexAppServerClientFactory extends Context.Service<
  CodexAppServerClientFactory,
  CodexAppServerClientFactoryShape
>()("t3/orchestration-v2/Adapters/CodexAdapterV2/CodexAppServerClientFactory") {}

/** MCP startup grace in milliseconds; the 1 s default excludes cold npx/uvx servers from the first model request. */
export const CODEX_THREAD_CONFIG = {
  "tools.update_plan.enabled": true,
  mcp_optional_startup_grace_ms: 10_000,
  // Tool calls stop waiting on the detailed reasoning summary. Codex applies it only to its
  // built-in OpenAI provider and sends each summary section whole, not as token deltas.
  "features.concurrent_reasoning_summaries": true,
} as const;

// T3 waits can block for an hour; keep Codex's MCP timeout above that.
const CODEX_T3_MCP_TOOL_TIMEOUT_SEC = 65 * 60;
// Holds T3 tools Codex must call directly, never from Code Mode. Calls arrive as
// `t3-code-direct.<tool>`; t3McpToolPresentation.ts must keep resolving that form.
const CODEX_T3_DIRECT_MCP_SERVER = "t3-code-direct";
const CODEX_DIRECT_ONLY_T3_TOOL = "t3_thread_wait";

function codexThreadRuntimeParams(input: {
  readonly threadId: ThreadId | null;
  readonly mcpSession: McpProviderSession.McpProviderSessionConfig | undefined;
  readonly modelSelection?: { readonly model: string };
  readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy;
}): {
  readonly cwd?: string;
  readonly model?: string;
  readonly config: Readonly<Record<string, Schema.Json>>;
} {
  const { mcpSession } = input;
  const t3Server =
    mcpSession === undefined
      ? undefined
      : {
          url: mcpSession.endpoint,
          tool_timeout_sec: CODEX_T3_MCP_TOOL_TIMEOUT_SEC,
          http_headers: { Authorization: mcpSession.authorizationHeader },
        };
  return {
    ...(input.runtimePolicy?.cwd == null ? {} : { cwd: input.runtimePolicy.cwd }),
    ...(input.modelSelection === undefined ? {} : { model: input.modelSelection.model }),
    config: {
      ...CODEX_THREAD_CONFIG,
      // One app-server serves many threads, so the id goes in per-thread shell config, not its env.
      ...(input.threadId === null
        ? {}
        : { [`shell_environment_policy.set.${PortScanner.AGENT_THREAD_ENV}`]: input.threadId }),
      ...(t3Server === undefined
        ? {}
        : {
            // Override only T3's servers so launch-argument MCP servers survive.
            "mcp_servers.t3-code": { ...t3Server, disabled_tools: [CODEX_DIRECT_ONLY_T3_TOOL] },
            // Code Mode runs a blocking wait as a nested call, and the model keeps working
            // around it. A second, direct-only registration makes it a plain tool call.
            [`mcp_servers.${CODEX_T3_DIRECT_MCP_SERVER}`]: {
              ...t3Server,
              enabled_tools: [CODEX_DIRECT_ONLY_T3_TOOL],
              omit_tools_from: ["code_mode", "deferred"],
            },
          }),
    },
  };
}

const decodeCodexResumeMetadata = Schema.decodeUnknownEffect(
  Schema.Struct({ thread: Schema.Struct({ id: Schema.String, updatedAt: Schema.Number }) }),
);

const CodexChildSelection = Schema.Struct({
  thread: Schema.Struct({ id: Schema.String }),
  model: Schema.NullOr(Schema.String),
  reasoningEffort: Schema.optional(Schema.NullOr(Schema.String)),
  serviceTier: Schema.optional(Schema.NullOr(Schema.String)),
});
const decodeCodexChildModel = Schema.decodeUnknownEffect(CodexChildSelection);

const decodeCodexChildThread = Schema.decodeUnknownEffect(
  Schema.Struct({
    thread: Schema.Struct({
      id: Schema.String,
      model: Schema.optional(Schema.NullOr(Schema.String)),
      reasoningEffort: Schema.optional(Schema.NullOr(Schema.String)),
    }),
  }),
);

// Without a deadline the spawner waits indefinitely for a process that ignores SIGTERM.
const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds";

export const makeCodexAppServerSpawnCommand = Effect.fn(
  "CodexAdapterV2.makeCodexAppServerSpawnCommand",
)(function* (input: {
  readonly command: string;
  readonly args: ReadonlyArray<string>;
  readonly cwd?: string | undefined;
  readonly env?: NodeJS.ProcessEnv | undefined;
  readonly extendEnv?: boolean | undefined;
  readonly threadId?: string | undefined;
}) {
  const launch = yield* (yield* AgentScope).wrap({
    command: input.command,
    args: input.args,
    name: "codex",
    threadId: input.threadId,
    env: input.env,
    shared: true,
  });
  const spawnCommand = yield* resolveSpawnCommand(launch.command, launch.args, {
    ...(input.env === undefined ? {} : { env: input.env }),
    ...(input.extendEnv === undefined ? {} : { extendEnv: input.extendEnv }),
  });
  return ChildProcess.make(spawnCommand.command, spawnCommand.args, {
    ...(input.cwd === undefined ? {} : { cwd: input.cwd }),
    ...(input.env === undefined ? {} : { env: input.env }),
    ...(input.extendEnv === undefined ? {} : { extendEnv: input.extendEnv }),
    shell: spawnCommand.shell,
    forceKillAfter: CODEX_APP_SERVER_FORCE_KILL_AFTER,
  });
});

function readCodexProtocolPath(value: unknown, ...path: ReadonlyArray<string>): unknown {
  let current = value;
  for (const key of path) {
    if (current === null || typeof current !== "object") return undefined;
    current = Reflect.get(current, key);
  }
  return current;
}

const CODEX_THREAD_ENV_CONFIG_KEY = `shell_environment_policy.set.${PortScanner.AGENT_THREAD_ENV}`;

/**
 * One app-server serves every Codex thread on an instance, so each frame is routed to the
 * T3 thread it belongs to. Frames tied to no thread, such as `initialize`, go to the global log.
 */
function makeCodexProtocolThreadRouter() {
  const threadIdsByNativeId = new Map<string, ThreadId>();
  const threadIdsByRequest = new Map<string, ThreadId>();
  return (direction: "incoming" | "outgoing", frame: unknown): ThreadId | null => {
    const id = readCodexProtocolPath(frame, "id");
    const method = readCodexProtocolPath(frame, "method");
    const isRequest = typeof method === "string";
    const requestKey =
      typeof id === "string" || typeof id === "number"
        ? `${isRequest === (direction === "outgoing") ? "client" : "server"}:${id}`
        : undefined;
    if (!isRequest) {
      if (requestKey === undefined) return null;
      const threadId = threadIdsByRequest.get(requestKey);
      threadIdsByRequest.delete(requestKey);
      const nativeThreadId = readCodexProtocolPath(frame, "result", "thread", "id");
      if (threadId !== undefined && typeof nativeThreadId === "string") {
        threadIdsByNativeId.set(nativeThreadId, threadId);
      }
      return threadId ?? null;
    }

    const params = readCodexProtocolPath(frame, "params");
    const configuredThreadId = readCodexProtocolPath(params, "config", CODEX_THREAD_ENV_CONFIG_KEY);
    const nativeThreadId =
      readCodexProtocolPath(params, "threadId") ?? readCodexProtocolPath(params, "thread", "id");
    let threadId =
      typeof configuredThreadId === "string"
        ? ThreadId.make(configuredThreadId)
        : typeof nativeThreadId === "string"
          ? threadIdsByNativeId.get(nativeThreadId)
          : undefined;
    if (threadId === undefined && typeof nativeThreadId === "string") {
      const parentNativeThreadId = readCodexProtocolPath(
        params,
        "thread",
        "source",
        "subAgent",
        "thread_spawn",
        "parent_thread_id",
      );
      threadId =
        typeof parentNativeThreadId === "string"
          ? threadIdsByNativeId.get(parentNativeThreadId)
          : undefined;
      if (threadId !== undefined) threadIdsByNativeId.set(nativeThreadId, threadId);
    }
    if (threadId !== undefined && requestKey !== undefined) {
      threadIdsByRequest.set(requestKey, threadId);
    }
    return threadId ?? null;
  };
}

export function makeCodexAppServerProtocolLogger(input: {
  readonly nativeEventLogger: ProviderEventLoggers.EventNdjsonLogger | undefined;
  readonly providerSessionId: OrchestrationV2ProviderSession["id"];
}): CodexClient.CodexAppServerClientOptions["logger"] | undefined {
  const { nativeEventLogger } = input;
  if (nativeEventLogger === undefined) {
    return undefined;
  }

  const routeThread = makeCodexProtocolThreadRouter();
  return (event) => {
    if (!shouldPersistProviderEvent("native", event)) return Effect.void;
    const threadId = event.stage === "decoded" ? routeThread(event.direction, event.payload) : null;
    return nativeEventLogger
      .write(
        {
          provider: CODEX_PROVIDER,
          protocol: "codex.app-server",
          kind: "protocol",
          providerSessionId: input.providerSessionId,
          event: redactCodexProtocolValue(boundProviderEventForLogging(event)),
        },
        threadId,
      )
      .pipe(Effect.ignore);
  };
}

function redactCodexProtocolValue(value: unknown): unknown {
  if (Array.isArray(value)) {
    return value.map(redactCodexProtocolValue);
  }
  if (value === null || typeof value !== "object") {
    if (typeof value !== "string") {
      return value;
    }
    if (/^Bearer\s+/i.test(value)) {
      return "[REDACTED]";
    }
    const trimmed = value.trim();
    if (
      !(
        (trimmed.startsWith("{") && trimmed.endsWith("}")) ||
        (trimmed.startsWith("[") && trimmed.endsWith("]"))
      )
    ) {
      return value;
    }
    try {
      return pipe(
        JSON.parse(trimmed) as unknown,
        boundProviderEventForLogging,
        redactCodexProtocolValue,
        JSON.stringify,
      );
    } catch {
      return value;
    }
  }
  return Object.fromEntries(
    Object.entries(value).map(([key, nested]) => [
      key,
      isSensitiveCodexProtocolKey(key) ? "[REDACTED]" : redactCodexProtocolValue(nested),
    ]),
  );
}

function isSensitiveCodexProtocolKey(key: string): boolean {
  const normalized = key.replace(/[^a-z0-9]/gi, "").toLowerCase();
  return (
    normalized.endsWith("authorization") ||
    normalized.endsWith("apikey") ||
    normalized.endsWith("token") ||
    normalized.endsWith("password") ||
    normalized.endsWith("secret")
  );
}

export const layerAppServerClientFactory: Layer.Layer<
  CodexAppServerClientFactory,
  never,
  | ChildProcessSpawner.ChildProcessSpawner
  | PortScanner.PortDiscovery
  | ProviderEventLoggers.ProviderEventLoggers
> = Layer.effect(
  CodexAppServerClientFactory,
  Effect.gen(function* () {
    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
    const { native: nativeEventLogger } = yield* ProviderEventLoggers.ProviderEventLoggers;
    const agentScope = yield* AgentScope;
    const portDiscovery = yield* PortScanner.PortDiscovery;

    return CodexAppServerClientFactory.of({
      open: Effect.fn("CodexAdapterV2.open")(function* (input) {
        const scope = yield* Scope.Scope;
        const environment = {
          ...codexAppServerEnvironment(input.environment),
          ...(input.settings.homePath ? { CODEX_HOME: input.settings.homePath } : {}),
        };
        const command = yield* makeCodexAppServerSpawnCommand({
          command: input.settings.binaryPath || "codex",
          args: codexAppServerArgs(
            resolveCodexLaunchArgs(input.settings.launchArgs, input.environment),
          ),
          env: environment,
          threadId: input.threadId,
        }).pipe(Effect.provideService(AgentScope, agentScope));
        const handle = yield* spawner.spawn(command).pipe(
          Effect.provideService(Scope.Scope, scope),
          Effect.mapError(
            (cause) =>
              new ProviderAdapter.ProviderAdapterOpenSessionError({
                driver: CODEX_PROVIDER,
                providerSessionId: input.providerSessionId,
                cause,
              }),
          ),
        );
        // Its shells carry each thread's T3CODE_THREAD_ID, so port discovery reads listeners' threads.
        yield* Effect.acquireRelease(Effect.sync(portDiscovery.trackAgent), (release) =>
          Effect.sync(release),
        );
        const protocolLogger = makeCodexAppServerProtocolLogger({
          nativeEventLogger,
          providerSessionId: input.providerSessionId,
        });
        const clientOptions: CodexClient.CodexAppServerClientOptions =
          protocolLogger === undefined
            ? {}
            : {
                logIncoming: true,
                logOutgoing: true,
                logger: protocolLogger,
              };
        const context = yield* Layer.build(CodexClient.layerChildProcess(handle, clientOptions));
        return yield* Effect.service(CodexClient.CodexAppServerClient).pipe(
          Effect.provide(context),
        );
      }),
    });
  }),
);

export type CodexAdapterV2DriverEnv =
  | CodexAppServerClientFactory
  | Digest.Digest
  | FileSystem.FileSystem
  | IdAllocator.IdAllocatorV2
  | McpProviderSessions.McpProviderSessions
  | Path.Path
  | ServerConfig;

export const createCodexAdapterV2 = Effect.fn("CodexAdapterV2.createCodexAdapterV2")(function* (
  { instanceId, environment, enabled, config }: ProviderAdapterDriverCreateInput<CodexSettings>,
  hooks: Pick<
    CodexAdapterV2Options,
    "onUsageLimits" | "resolveRuntime" | "compatibility" | "models"
  > = {},
) {
  const clientFactory = yield* CodexAppServerClientFactory;
  const continuationRequests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
  const digest = yield* Digest.Digest;
  const fileSystem = yield* FileSystem.FileSystem;
  const hostEnvironment = yield* HostProcess.Environment;
  const idAllocator = yield* IdAllocator.IdAllocatorV2;
  const serverConfig = yield* ServerConfig;
  const homeLayout = yield* resolveCodexHomeLayout(config);

  yield* materializeCodexShadowHome(homeLayout).pipe(
    Effect.mapError(
      (cause) =>
        new ProviderAdapterDriverCreateError({
          driver: CODEX_DRIVER_KIND,
          instanceId,
          detail: "Failed to materialize the Codex shadow home.",
          cause,
        }),
    ),
  );

  const settings = {
    ...config,
    enabled,
    binaryPath: expandHomePath(config.binaryPath, yield* HostProcess.HomeDirectory),
    homePath: homeLayout.effectiveHomePath ?? "",
  } satisfies CodexSettings;

  return yield* makeCodexAdapterV2({
    instanceId,
    settings,
    environment: yield* mergeProviderInstanceEnvironment(environment, hostEnvironment),
    clientFactory,
    digest,
    fileSystem,
    idAllocator,
    serverConfig,
    continuationRequests,
    ...hooks,
  });
});

export interface CodexAdapterV2Options {
  readonly instanceId: ProviderInstanceId;
  readonly settings: CodexSettings;
  readonly environment: NodeJS.ProcessEnv;
  readonly clientFactory: CodexAppServerClientFactoryShape;
  readonly onUsageLimits?: ManagedServerProvider["applyUsageLimits"];
  /** The provider's published models, read per turn for catalog defaults. */
  readonly models?: Effect.Effect<ReadonlyArray<ServerProviderModel>>;
  readonly resolveRuntime?: Effect.Effect<CodexEffectiveRuntime, ProviderSetupError, Scope.Scope>;
  readonly compatibility?: (
    version: string,
  ) => Effect.Effect<ServerProviderCompatibilityAdvisory | undefined>;
  readonly digest: Digest.Digest["Service"];
  readonly fileSystem: FileSystem.FileSystem;
  readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
  readonly serverConfig: ServerConfig["Service"];
  /** Omitting this leaves late completions projection-only. */
  readonly continuationRequests?: {
    readonly offer: (
      request: ProviderContinuationRequests.ProviderContinuationRequest,
    ) => Effect.Effect<void>;
  };
}

export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
  adapterOptions: CodexAdapterV2Options,
) {
  const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
  const readMcpSession = (threadId: ThreadId | null) =>
    threadId === null ? Effect.undefined : mcpSessions.read(threadId);
  const { clientFactory, digest, fileSystem, idAllocator, serverConfig } = adapterOptions;
  const continuationRequests = adapterOptions.continuationRequests;

  return ProviderAdapter.ProviderAdapterV2.of({
    instanceId: adapterOptions.instanceId,
    driver: CODEX_PROVIDER,
    getCapabilities: Effect.fn("CodexAdapterV2.getCapabilities")(function () {
      return Effect.succeed(CodexProviderCapabilitiesV2);
    }),
    planSelectionTransition: Effect.fn("CodexAdapterV2.planSelectionTransition")(function () {
      return Effect.succeed(turnScopedSelectionTransition());
    }),
    openSession: Effect.fn("CodexAdapterV2.openSession")(
      function* (input) {
        const scope = yield* Scope.Scope;
        const resolvedRuntime =
          adapterOptions.resolveRuntime === undefined
            ? undefined
            : yield* adapterOptions.resolveRuntime.pipe(
                Effect.mapError(
                  (cause) =>
                    new ProviderAdapter.ProviderAdapterOpenSessionError({
                      driver: CODEX_PROVIDER,
                      providerSessionId: input.providerSessionId,
                      cause,
                    }),
                ),
              );
        const client = yield* clientFactory.open({
          instanceId: adapterOptions.instanceId,
          threadId: input.threadId,
          providerSessionId: input.providerSessionId,
          runtimePolicy: input.runtimePolicy,
          settings: resolvedRuntime?.config ?? adapterOptions.settings,
          environment: resolvedRuntime?.environment ?? adapterOptions.environment,
        });
        const additionalContextByThread = yield* Ref.make(
          new Map<string, NonNullable<CodexTurnStartParams["additionalContext"]>>(),
        );
        const threadsAwaitingContextRestore = yield* Ref.make(new Set<string>());
        // Codex drops client developer messages during compaction but only resends additionalContext
        // when it changes. Restore after turn/completed: items injected mid-turn make the turn sample again.
        const restoreAdditionalContext = Effect.fn("CodexAdapterV2.restoreAdditionalContext")(
          function* (threadId: string) {
            const context = (yield* Ref.get(additionalContextByThread)).get(threadId);
            if (!context) return;
            yield* client.request("thread/inject_items", {
              threadId,
              items: Object.entries(context).map(([key, entry]) =>
                entry.kind === "untrusted"
                  ? {
                      type: "message",
                      role: "user",
                      content: [
                        {
                          type: "input_text",
                          text: `<external_${key}>${entry.value}</external_${key}>`,
                        },
                      ],
                    }
                  : {
                      type: "message",
                      role: "developer",
                      content: [{ type: "input_text", text: `<${key}>${entry.value}</${key}>` }],
                    },
              ),
            });
          },
          Effect.timeout("10 seconds"),
          Effect.catch((cause) =>
            Effect.logWarning("Failed to restore Codex additional context after compaction.", {
              cause,
            }),
          ),
        );
        const initialized = yield* Ref.make(false);
        // Codex rejects a second initialize on the shared app-server.
        const initializePermit = yield* Semaphore.make(1);
        const ensureInitialized = initializePermit.withPermit(
          Effect.gen(function* () {
            const alreadyInitialized = yield* Ref.get(initialized);
            if (alreadyInitialized) {
              return;
            }

            const userAgent = yield* client
              .request("initialize", {
                clientInfo: buildCodexInitializeParams().clientInfo,
                capabilities: CODEX_CLIENT_CAPABILITIES,
              })
              .pipe(
                Effect.map((response) => response.userAgent),
                Effect.catchTags({
                  // Interruption can leave Codex initialized before the local flag is set.
                  CodexAppServerRequestError: (error) =>
                    error.code === -32600 && error.errorMessage === "Already initialized"
                      ? Effect.succeed(null)
                      : Effect.fail(error),
                }),
              );
            const version = userAgent && /\/(\S+)/u.exec(userAgent)?.[1];
            const advisory =
              version && adapterOptions.compatibility
                ? yield* adapterOptions.compatibility(version)
                : undefined;
            if (advisory?.status === "broken") {
              const target =
                advisory.recommendedVersion ?? advisory.recommendedRange ?? "a newer version";
              return yield* new ProviderSetupError({
                instanceId: adapterOptions.instanceId,
                operation: "runtime",
                detail: `Codex ${version} does not work with this T3 Code release. Update Codex to ${target}, then retry.`,
              });
            }
            yield* client.notify("initialized", undefined);
            yield* Ref.set(initialized, true);
          }),
        );
        // A failed open is discarded, so the next turn spawns the updated binary.
        if (adapterOptions.compatibility !== undefined) yield* ensureInitialized;
        const now = yield* DateTime.now;
        const session = providerSession({
          providerSessionId: input.providerSessionId,
          providerInstanceId: adapterOptions.instanceId,
          cwd: input.runtimePolicy.cwd,
          model: input.modelSelection.model,
          now,
        });
        const events = yield* Queue.unbounded<
          ProviderAdapter.ProviderAdapterV2Event,
          ProviderAdapter.ProviderAdapterEventStreamError
        >();
        // Drain final notifications before the session manager retires the dead process.
        // Queue.fail lets readers take items already buffered before they see the failure.
        yield* client.awaitTermination.pipe(
          Effect.catch((cause) =>
            Queue.fail(
              events,
              new ProviderAdapter.ProviderAdapterEventStreamError({
                driver: CODEX_PROVIDER,
                providerSessionId: input.providerSessionId,
                cause,
              }),
            ),
          ),
          Effect.forkScoped,
        );
        const rateLimitSnapshot = yield* Ref.make<CodexRateLimitSnapshot | undefined>(undefined);
        const limitedTurnItems = yield* Ref.make(
          new Map<ProviderThreadId, Extract<OrchestrationV2TurnItem, { type: "error" }>>(),
        );
        const activeTurns = yield* Ref.make(new Map<string, ActiveCodexTurnContext>());
        // Codex reports MCP startup on thread start, usually before the turn context exists.
        const mcpStartupFailures = new Map<string, Map<string, string>>();
        const turnTokenUsageByThread = new Map<string, CodexTurnTokenUsageState>();
        const usageStateForThread = (nativeThreadId: string) => {
          let state = turnTokenUsageByThread.get(nativeThreadId);
          if (!state) {
            state = makeCodexTurnTokenUsageState();
            turnTokenUsageByThread.set(nativeThreadId, state);
          }
          return state;
        };
        const beginTurnTokenUsage = (context: ActiveCodexTurnContext) => {
          const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId;
          if (!nativeThreadId) return;
          const state = usageStateForThread(nativeThreadId);
          if (state.activeTurnId !== context.nativeTurnId) {
            state.byTurnId.clear();
            state.activeTurnId = context.nativeTurnId;
            getCodexTurnAccumulator(state, context.nativeTurnId);
          }
        };
        const markSubagentUsage = (context: ActiveCodexTurnContext) => {
          const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId;
          if (!nativeThreadId) return;
          getCodexTurnAccumulator(
            usageStateForThread(nativeThreadId),
            context.nativeTurnId,
          ).hasSubagents = true;
        };
        const pendingRootTurns = yield* Ref.make(
          new Map<string, ProviderAdapter.ProviderAdapterV2TurnInput>(),
        );
        const turnWaiters = yield* Ref.make(new Map<string, Deferred.Deferred<void, never>>());
        const subagentThreads = yield* Ref.make(new Map<string, CodexSubagentThreadContext>());
        const subagentSelections = new Map<string, Omit<ModelSelection, "instanceId">>();
        // A reroute names one turn; until new settings arrive, the thread's model is unknown.
        const reroutedTurnModels = new Map<string, string>();
        const reroutedThreads = new Set<string>();
        const reportedTurnModel = (nativeThreadId: string, nativeTurnId: string) => {
          const rerouted = reroutedTurnModels.get(nativeTurnId);
          reroutedTurnModels.delete(nativeTurnId);
          if (rerouted !== undefined) return rerouted;
          return reroutedThreads.has(nativeThreadId)
            ? undefined
            : subagentSelections.get(nativeThreadId)?.model;
        };
        const pendingSubagentTurns = yield* Ref.make(
          new Map<string, ReadonlyArray<PendingCodexSubagentTurnStarted>>(),
        );
        const nextProviderTurnOrdinals = yield* Ref.make(new Map<string, number>());
        const providerRetries = yield* Ref.make(
          new Map<ProviderTurnId, ActiveCodexProviderRetry>(),
        );
        const planIds = yield* Ref.make(new Map<string, OrchestrationV2PlanArtifact["id"]>());
        const pendingRuntimeRequests = yield* Ref.make(
          new Map<string, PendingCodexRuntimeRequest>(),
        );
        /** Completed by thread/closed after this connection unsubscribes from the native thread. */
        const threadClosings = new Map<string, Deferred.Deferred<void>>();
        /** Codex acceptForSession covers only the exact command or patch, not future commands. */
        const sessionApprovals = new Map<
          string,
          { readonly runtimeMode: RuntimeMode; readonly kinds: Set<ProviderRequestKind> }
        >();
        /** Bind answers to the request's approval set so mode changes invalidate late answers. */
        const endSessionApprovalsOnModeChange = (context: ActiveCodexTurnContext) => {
          const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId;
          if (
            nativeThreadId != null &&
            sessionApprovals.get(nativeThreadId)?.runtimeMode !==
              context.input.runtimePolicy.runtimeMode
          ) {
            sessionApprovals.delete(nativeThreadId);
          }
        };
        const sessionAllowance = (
          nativeThreadId: string,
          context: ActiveCodexTurnContext,
        ): Set<ProviderRequestKind> => {
          const runtimeMode = context.input.runtimePolicy.runtimeMode;
          const allowed = sessionApprovals.get(nativeThreadId);
          if (allowed?.runtimeMode === runtimeMode) return allowed.kinds;
          if (allowed !== undefined) return new Set();
          const kinds = new Set<ProviderRequestKind>();
          sessionApprovals.set(nativeThreadId, { runtimeMode, kinds });
          return kinds;
        };
        /** Retain settled contexts while background commands run so late item events still project. */
        const settledTurns = yield* Ref.make(new Map<string, ActiveCodexTurnContext>());
        const runningCommandItemsByTurn = yield* Ref.make(
          new Map<string, Map<string, TrackedRunningCommandItem>>(),
        );
        const runningDynamicToolsByTurn = yield* Ref.make(
          new Map<string, Map<string, CodexDynamicToolItem>>(),
        );
        const interruptingNativeTurns = yield* Ref.make(new Set<string>());
        const terminalizedNonCompletedNativeTurns = yield* Ref.make(new Set<string>());
        // Keep the stream open until descendants settle to avoid stranding child projections.
        const deferredRootTerminals = yield* Ref.make(new Map<string, DeferredCodexRootTerminal>());
        const offeredContinuationItemsByTurn = yield* Ref.make(new Map<string, Set<string>>());
        const finalAnswerItemIdsByTurn = yield* Ref.make(new Map<string, Set<string>>());
        const completedFinalAnswerTextsByTurn = yield* Ref.make(new Map<string, Set<string>>());
        const imageTurns = new Map<
          string,
          {
            readonly running: Map<
              string,
              {
                readonly revisedPrompt: string | null;
                readonly transparentBackground: boolean | null;
              }
            >;
            readonly saved: Map<string, string>;
            pending: boolean;
            answer?: { readonly id: string; readonly text: string };
          }
        >();
        const imageTurn = (nativeTurnId: string) => {
          let state = imageTurns.get(nativeTurnId);
          if (state === undefined) {
            state = { running: new Map(), saved: new Map(), pending: false };
            imageTurns.set(nativeTurnId, state);
          }
          return state;
        };
        // Serialize native completion and interrupt timeout so only one publishes terminal events.
        const turnTerminalizationPermit = yield* Semaphore.make(1);
        const goalsByNativeThread = new Map<string, OrchestrationV2ProviderGoal | null>();
        const rootProviderThreads = new Map<string, OrchestrationV2ProviderThread>();
        const goalHolds = new Map<string, CodexGoalHold>();
        const continuedRuns = new Map<ProviderTurnId, Array<ActiveCodexTurnContext>>();
        const capacityContinuations = new Map<
          string,
          { readonly context: ActiveCodexTurnContext; readonly retry: ActiveCodexProviderRetry }
        >();
        const goalActivations = new Map<string, { stopped: boolean }>();
        // Held runs must not complete while Stop resolves its target.
        const goalStops = new Map<string, number>();
        // Goals paused because a delegated task will wake the thread; the next ordinary turn resumes them.
        const goalsPausedForWake = new Set<string>();
        const latestRunTurnId = (providerTurnId: ProviderTurnId) =>
          continuedRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId;
        const isContinuedRunDescendant = (
          candidate: ActiveCodexTurnContext,
          context: ActiveCodexTurnContext,
        ) =>
          (continuedRuns.get(context.providerTurnId) ?? [context]).some((ancestor) =>
            isDescendantCodexTurn(candidate, ancestor),
          );

        const emitProviderEvent = (event: ProviderAdapter.ProviderAdapterV2Event) =>
          Queue.offer(events, event).pipe(Effect.asVoid);

        const promptSuggestionsEnabled = input.promptSuggestions ?? Effect.succeed(false);
        // Best effort: older Codex lacks thread/sample, and a sample fails once the user sends again.
        // Delegated threads share this session but have no one typing, so they never sample.
        const samplePromptSuggestion = Effect.fn("CodexAdapterV2.samplePromptSuggestion")(
          function* (
            context: ActiveCodexTurnContext,
            nativeThreadId: string,
            nativeTurnId: string,
          ) {
            if (
              isDelegatedAppThread(context.input.appThread) ||
              !(yield* promptSuggestionsEnabled)
            ) {
              return;
            }
            const response = yield* client.raw.request("thread/sample", {
              threadId: nativeThreadId,
              expectedTurnId: nativeTurnId,
              prompt: CODEX_PROMPT_SUGGESTION_PROMPT,
              effort: "low",
            });
            const suggestion =
              (yield* decodeCodexThreadSampleResponse(response)).text?.trim() ?? "";
            if (suggestion.length === 0) return;
            yield* emitProviderEvent({
              type: "prompt_suggestion.updated",
              driver: CODEX_PROVIDER,
              threadId: context.projectionThreadId,
              runId: context.input.runId,
              text: suggestion,
            });
          },
          Effect.catchCause((cause) =>
            Effect.logDebug("orchestration-v2.codex-prompt-suggestion-failed", { cause }),
          ),
        );

        const emitImageWorkState =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (context: ActiveCodexTurnContext, pending: boolean) {
            const state = imageTurns.get(context.nativeTurnId);
            if (state === undefined || state.pending === pending) return;
            state.pending = pending;
            yield* emitProviderEvent({
              type: "background_work.updated",
              driver: CODEX_PROVIDER,
              providerTurnId: context.providerTurnId,
              pending,
            });
          });

        const clearImageWork = Effect.fn("CodexAdapterV2.clearImageWork")(function* (
          context: ActiveCodexTurnContext,
        ) {
          for (const [id, input] of imageTurns.get(context.nativeTurnId)?.running ?? []) {
            yield* emitImageGenerationItem(
              context,
              {
                type: "imageGeneration",
                id,
                ...input,
                status: "cancelled",
                result: "",
              },
              true,
            );
          }
          yield* emitImageWorkState(context, false);
          imageTurns.delete(context.nativeTurnId);
        });

        // Setting a goal passes through cleared and paused states; publish only the result.
        const goalUpdatesHeld = new Set<string>();
        const emitGoalUpdate = Effect.fnUntraced(function* (nativeThreadId: string) {
          const providerThread = rootProviderThreads.get(nativeThreadId);
          if (
            providerThread === undefined ||
            !goalsByNativeThread.has(nativeThreadId) ||
            goalUpdatesHeld.has(nativeThreadId)
          ) {
            return;
          }
          const goal = goalsByNativeThread.get(nativeThreadId) ?? null;
          if (providerGoalsEqual(providerThread.goal ?? null, goal)) return;
          const updated = { ...providerThread, goal, updatedAt: yield* DateTime.now };
          rootProviderThreads.set(nativeThreadId, updated);
          yield* emitProviderEvent({
            type: "provider_thread.updated",
            driver: CODEX_PROVIDER,
            providerThread: updated,
          });
        });

        const rememberRootProviderThread = Effect.fnUntraced(function* (
          providerThread: OrchestrationV2ProviderThread,
        ) {
          const nativeThreadId = providerThread.nativeThreadRef?.nativeId;
          if (nativeThreadId == null) return;
          rootProviderThreads.set(nativeThreadId, providerThread);
          yield* emitGoalUpdate(nativeThreadId);
        });

        const emitMcpStartupFailures = Effect.fn("CodexAdapterV2.emitMcpStartupFailures")(
          function* (context: ActiveCodexTurnContext) {
            const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId;
            if (nativeThreadId == null) return;
            const failures = mcpStartupFailures.get(nativeThreadId);
            if (failures === undefined) return;
            mcpStartupFailures.delete(nativeThreadId);
            for (const [name, message] of failures) {
              const nativeItemId = `mcp-startup-failed:${context.providerTurnId}:${name}`;
              const { ordinal, startedAt } = yield* resolveItemPosition(context, nativeItemId);
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: {
                  id: idAllocator.derive.turnItemFromProviderItem({
                    driver: CODEX_PROVIDER,
                    nativeItemId,
                  }),
                  threadId: context.projectionThreadId,
                  runId: context.projectionRunId,
                  nodeId: context.providerNodeId,
                  providerThreadId: context.providerThread.id,
                  providerTurnId: context.providerTurnId,
                  nativeItemRef: { ...codexNativeItemRef(nativeItemId), strength: "weak" },
                  parentItemId: null,
                  ordinal,
                  type: "system_notice",
                  status: "completed",
                  title: `MCP server ${name} failed to start`,
                  message,
                  startedAt,
                  completedAt: startedAt,
                  updatedAt: startedAt,
                },
              });
            }
          },
        );

        const releaseGoalUpdates = (nativeThreadId: string) =>
          Effect.suspend(() => {
            goalUpdatesHeld.delete(nativeThreadId);
            return emitGoalUpdate(nativeThreadId);
          });

        // Only new model output proves recovery; local item/completed can arrive during upstream retries.
        // Runs on every model output frame and is almost always a no-op; only a pending retry gets a span.
        const completeProviderRetry = Effect.fnUntraced(function* (
          context: ActiveCodexTurnContext,
          updatedAt: DateTime.Utc,
        ) {
          const providerRetry = yield* Ref.modify(providerRetries, (current) => {
            const retry = current.get(context.providerTurnId);
            if (retry === undefined) {
              return [undefined, current] as const;
            }
            const updated = new Map(current);
            updated.delete(context.providerTurnId);
            return [retry, updated] as const;
          });
          if (providerRetry === undefined) {
            return;
          }
          yield* emitProviderEvent({
            type: "turn_item.updated",
            driver: CODEX_PROVIDER,
            turnItem: codexRetryTurnItem(context, providerRetry, "completed", updatedAt),
          }).pipe(Effect.withSpan("CodexAdapterV2.completeProviderRetry"));
        });

        const registerRootTurn = Effect.fn("CodexAdapterV2.registerRootTurn")(function* (input: {
          readonly turnInput: ProviderAdapter.ProviderAdapterV2TurnInput;
          readonly nativeTurnId: string;
          readonly startedAt: DateTime.Utc;
          readonly waitForNativeStart?: boolean;
        }) {
          const existing = (yield* Ref.get(activeTurns)).get(input.nativeTurnId);
          if (existing !== undefined) {
            return existing;
          }
          const providerTurnId = idAllocator.derive.providerTurn({
            driver: CODEX_PROVIDER,
            nativeTurnId: input.nativeTurnId,
          });
          const context: ActiveCodexTurnContext = {
            ...(input.waitForNativeStart ? { nativeStartReady: yield* Deferred.make<void>() } : {}),
            input: input.turnInput,
            projectionAppThread: input.turnInput.appThread,
            projectionThreadId: input.turnInput.threadId,
            projectionRunId: input.turnInput.runId,
            nativeTurnId: input.nativeTurnId,
            providerThread: input.turnInput.providerThread,
            providerTurnId,
            providerTurnOrdinal: input.turnInput.providerTurnOrdinal,
            providerNodeId: input.turnInput.rootNodeId,
            providerNodeStartedAt: input.startedAt,
            itemParentNodeId: input.turnInput.rootNodeId,
            rootNodeId: input.turnInput.rootNodeId,
            subagent: null,
            startedAt: input.startedAt,
            itemPositions: new Map(),
            fileChangeKindsByItem: new Map(),
            messagePhases: new Map(),
            runningCompactionItemIds: new Set(),
          };
          const nativeThreadId = input.turnInput.providerThread.nativeThreadRef?.nativeId;
          const continuation =
            nativeThreadId == null ? undefined : capacityContinuations.get(nativeThreadId);
          if (continuation !== undefined) {
            context.capacityRetryAttempt = continuation.context.capacityRetryAttempt ?? 0;
            yield* Ref.update(providerRetries, (current) =>
              new Map(current).set(providerTurnId, continuation.retry),
            );
            const turns = continuedRuns.get(continuation.context.providerTurnId) ?? [
              continuation.context,
            ];
            turns.push(context);
            continuedRuns.set(continuation.context.providerTurnId, turns);
            continuedRuns.set(providerTurnId, turns);
            if (continuation.context.capacityRetryPending !== undefined) {
              yield* Deferred.succeed(continuation.context.capacityRetryPending, context);
            }
          }
          yield* Ref.update(limitedTurnItems, (current) => {
            const next = new Map(current);
            next.delete(context.providerThread.id);
            return next;
          });
          beginTurnTokenUsage(context);
          endSessionApprovalsOnModeChange(context);
          yield* Ref.update(activeTurns, (current) => {
            const updated = new Map(current);
            updated.set(input.nativeTurnId, context);
            return updated;
          });
          if (continuation !== undefined && nativeThreadId != null) {
            capacityContinuations.delete(nativeThreadId);
          }
          yield* emitProviderEvent({
            type: "provider_turn.updated",
            driver: CODEX_PROVIDER,
            threadId: input.turnInput.threadId,
            providerTurn: {
              id: providerTurnId,
              providerThreadId: input.turnInput.providerThread.id,
              nodeId: input.turnInput.rootNodeId,
              runAttemptId: input.turnInput.attemptId,
              nativeTurnRef: {
                driver: CODEX_PROVIDER,
                nativeId: input.nativeTurnId,
                strength: "strong",
              },
              ordinal: input.turnInput.providerTurnOrdinal,
              status: "running",
              startedAt: input.startedAt,
              completedAt: null,
            },
          });
          yield* rememberRootProviderThread(input.turnInput.providerThread);
          yield* emitMcpStartupFailures(context);
          return context;
        });

        const findActiveTurnByNativeThreadId =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (nativeThreadId: string) {
            const turns = Array.from((yield* Ref.get(activeTurns)).values());
            return turns.find(
              (context) => context.providerThread.nativeThreadRef?.nativeId === nativeThreadId,
            );
          });

        // Streaming updates repeatedly await this context, including retries before registration.
        const awaitActiveTurn = Effect.fnUntraced(function* (
          nativeTurnId: string,
          attemptsRemaining = 1_000,
        ): Effect.fn.Return<ActiveCodexTurnContext | undefined> {
          const context = (yield* Ref.get(activeTurns)).get(nativeTurnId);
          if (context !== undefined || attemptsRemaining <= 0) {
            return context;
          }
          yield* Effect.yieldNow;
          return yield* awaitActiveTurn(nativeTurnId, attemptsRemaining - 1);
        });

        const resolveItemEventContext =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (nativeTurnId: string) {
            const settled = (yield* Ref.get(settledTurns)).get(nativeTurnId);
            if (settled !== undefined) {
              return { context: settled, settled: true } as const;
            }
            const context = yield* awaitActiveTurn(nativeTurnId);
            return context === undefined ? undefined : ({ context, settled: false } as const);
          });

        const trackRunningCommandItem = (nativeTurnId: string, item: TrackedRunningCommandItem) =>
          Ref.update(runningCommandItemsByTurn, (current) => {
            const updated = new Map(current);
            const items = new Map(updated.get(nativeTurnId) ?? []);
            items.set(item.id, item);
            updated.set(nativeTurnId, items);
            return updated;
          });

        const clearRunningCommandItem = (nativeTurnId: string, nativeItemId: string) =>
          Ref.modify(runningCommandItemsByTurn, (current) => {
            const items = current.get(nativeTurnId);
            if (items === undefined || !items.has(nativeItemId)) {
              return [items === undefined || items.size === 0, current] as const;
            }
            const remaining = new Map(items);
            remaining.delete(nativeItemId);
            const updated = new Map(current);
            if (remaining.size === 0) {
              updated.delete(nativeTurnId);
            } else {
              updated.set(nativeTurnId, remaining);
            }
            return [remaining.size === 0, updated] as const;
          });

        interface PendingMcpAppCapture {
          readonly item: Extract<CodexDynamicToolItem, { readonly type: "mcpToolCall" }>;
          readonly fiber: Fiber.Fiber<void>;
        }
        const pendingMcpAppCaptures = yield* Ref.make(
          new Map<string, ReadonlyMap<string, PendingMcpAppCapture>>(),
        );
        const takePendingMcpAppCapture = (nativeTurnId: string, nativeItemId: string) =>
          Ref.modify(pendingMcpAppCaptures, (current) => {
            const captures = current.get(nativeTurnId);
            if (captures === undefined || !captures.has(nativeItemId)) {
              return [false, current] as const;
            }
            const remaining = new Map(captures);
            remaining.delete(nativeItemId);
            const updated = new Map(current);
            if (remaining.size === 0) updated.delete(nativeTurnId);
            else updated.set(nativeTurnId, remaining);
            return [true, updated] as const;
          });

        const trackRunningDynamicTool = (nativeTurnId: string, item: CodexDynamicToolItem) =>
          Ref.update(runningDynamicToolsByTurn, (current) => {
            const updated = new Map(current);
            const items = new Map(updated.get(nativeTurnId) ?? []);
            items.set(item.id, item);
            updated.set(nativeTurnId, items);
            return updated;
          });

        const clearRunningDynamicTool = (nativeTurnId: string, nativeItemId: string) =>
          Ref.update(runningDynamicToolsByTurn, (current) => {
            const items = current.get(nativeTurnId);
            if (items === undefined || !items.has(nativeItemId)) {
              return current;
            }
            const remaining = new Map(items);
            remaining.delete(nativeItemId);
            const updated = new Map(current);
            if (remaining.size === 0) {
              updated.delete(nativeTurnId);
            } else {
              updated.set(nativeTurnId, remaining);
            }
            return updated;
          });

        const turnHasRetainedBackgroundWork =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (nativeTurnId: string) {
            const commands = (yield* Ref.get(runningCommandItemsByTurn)).get(nativeTurnId);
            if (commands !== undefined && commands.size > 0) {
              return true;
            }
            if ((yield* Ref.get(pendingMcpAppCaptures)).has(nativeTurnId)) {
              return true;
            }
            const images = imageTurns.get(nativeTurnId);
            if (images !== undefined && (images.running.size > 0 || images.saved.size > 0)) {
              return true;
            }
            const tools = (yield* Ref.get(runningDynamicToolsByTurn)).get(nativeTurnId);
            if (tools === undefined) {
              return false;
            }
            for (const item of tools.values()) {
              if (isPersistentCodexDynamicTool(item)) {
                return true;
              }
            }
            return false;
          });

        const releaseSettledTurnIfIdle =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (nativeTurnId: string) {
            if (yield* turnHasRetainedBackgroundWork(nativeTurnId)) {
              return;
            }
            yield* Ref.update(settledTurns, (current) => {
              if (!current.has(nativeTurnId)) {
                return current;
              }
              const updated = new Map(current);
              updated.delete(nativeTurnId);
              return updated;
            });
            yield* Ref.update(offeredContinuationItemsByTurn, (current) => {
              if (!current.has(nativeTurnId)) {
                return current;
              }
              const updated = new Map(current);
              updated.delete(nativeTurnId);
              return updated;
            });
            yield* Ref.update(completedFinalAnswerTextsByTurn, (current) => {
              if (!current.has(nativeTurnId)) {
                return current;
              }
              const updated = new Map(current);
              updated.delete(nativeTurnId);
              return updated;
            });
            imageTurns.delete(nativeTurnId);
            yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
              if (!current.has(nativeTurnId)) {
                return current;
              }
              const updated = new Map(current);
              updated.delete(nativeTurnId);
              return updated;
            });
          });

        /**
         * When a turn is interrupted or failed, Codex often leaves commandExecution
         * items mid-flight (no item/completed). Emit terminal turn items before
         * turn.terminal so the projection never keeps a forever-running command card.
         */
        const terminalizeRunningCommandItems = Effect.fn(
          "CodexAdapterV2.terminalizeRunningCommandItems",
        )(function* (
          context: ActiveCodexTurnContext,
          nativeTurnId: string,
          status: "interrupted" | "failed",
          completedAt: DateTime.Utc,
        ) {
          const items = (yield* Ref.get(runningCommandItemsByTurn)).get(nativeTurnId);
          if (items === undefined || items.size === 0) {
            return;
          }
          for (const tracked of items.values()) {
            const nodeId = idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: tracked.id,
            });
            const turnItemId = idAllocator.derive.turnItemFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: tracked.id,
            });
            const { ordinal, startedAt } = yield* resolveItemPosition(context, tracked.id);
            const node: OrchestrationV2ExecutionNode = {
              id: nodeId,
              threadId: context.projectionThreadId,
              runId: context.projectionRunId,
              parentNodeId: context.itemParentNodeId,
              rootNodeId: context.rootNodeId,
              kind: "tool_call",
              status,
              countsForRun: false,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(tracked.id),
              runtimeRequestId: null,
              checkpointScopeId: null,
              startedAt,
              completedAt,
            };
            const turnItem: OrchestrationV2TurnItem = {
              id: turnItemId,
              threadId: context.projectionThreadId,
              runId: context.projectionRunId,
              nodeId,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(tracked.id),
              parentItemId: null,
              ordinal,
              status,
              title: null,
              startedAt,
              completedAt,
              updatedAt: completedAt,
              type: "command_execution",
              input: tracked.command,
              ...(tracked.aggregatedOutput === undefined
                ? {}
                : { output: tracked.aggregatedOutput }),
            };
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem,
            });
          }
        });

        const terminateBackgroundTerminal = Effect.fn("CodexAdapterV2.terminateBackgroundTerminal")(
          function* (nativeThreadId: string, processId: string) {
            const response = yield* client.raw
              .request("thread/backgroundTerminals/terminate", {
                threadId: nativeThreadId,
                processId,
              })
              .pipe(
                Effect.catchTags({
                  CodexAppServerProcessExitedError: () => Effect.succeed({ terminated: true }),
                  CodexAppServerInputStreamEndedError: () => Effect.succeed({ terminated: true }),
                  // Unloading kills terminals, so "thread not found" means nothing remains to stop.
                  CodexAppServerRequestError: (error) =>
                    error.code === -32600 && error.errorMessage.startsWith("thread not found:")
                      ? Effect.succeed({ terminated: true })
                      : Effect.fail(error),
                }),
              );
            const result = yield* decodeCodexBackgroundTerminalTerminateResponse(response);
            if (result.terminated) return;
            let cursor: string | null = null;
            while (true) {
              const response: unknown = yield* client.raw.request(
                "thread/backgroundTerminals/list",
                {
                  threadId: nativeThreadId,
                  ...(cursor === null ? {} : { cursor }),
                },
              );
              const page: CodexBackgroundTerminalsListPage =
                yield* decodeCodexBackgroundTerminalsListResponse(response);
              if (page.data.some((terminal) => terminal.processId === processId)) {
                return yield* toProtocolError(
                  `Codex background terminal ${processId} remained active after termination.`,
                );
              }
              if (page.nextCursor === null) return;
              cursor = page.nextCursor;
            }
          },
          Effect.timeout("10 seconds"),
        );

        /**
         * Yielded exec cells can start MCP / dynamic tools that never receive
         * item/completed when functions.wait terminates the cell. Persistent tools
         * can outlive the root turn and must stay running.
         */
        const terminalizeRunningDynamicTools = Effect.fn(
          "CodexAdapterV2.terminalizeRunningDynamicTools",
        )(function* (
          context: ActiveCodexTurnContext,
          nativeTurnId: string,
          status: "cancelled" | "interrupted" | "failed",
          completedAt: DateTime.Utc,
          includePersistent: boolean,
        ) {
          const items = (yield* Ref.get(runningDynamicToolsByTurn)).get(nativeTurnId);
          if (items === undefined || items.size === 0) {
            return;
          }
          for (const tracked of items.values()) {
            if (!includePersistent && isPersistentCodexDynamicTool(tracked)) {
              continue;
            }
            const artifacts = yield* buildDynamicToolArtifacts(context, tracked);
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: {
                ...artifacts.node,
                status,
                completedAt,
              },
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: {
                ...artifacts.turnItem,
                status,
                completedAt,
                updatedAt: completedAt,
              },
            });
          }
          yield* Ref.update(runningDynamicToolsByTurn, (current) => {
            if (!current.has(nativeTurnId)) {
              return current;
            }
            const remaining = new Map(current.get(nativeTurnId) ?? []);
            for (const tracked of remaining.values()) {
              if (includePersistent || !isPersistentCodexDynamicTool(tracked)) {
                remaining.delete(tracked.id);
              }
            }
            const updated = new Map(current);
            if (remaining.size === 0) {
              updated.delete(nativeTurnId);
            } else {
              updated.set(nativeTurnId, remaining);
            }
            return updated;
          });
        });

        const resolveItemPosition =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (
            context: ActiveCodexTurnContext,
            nativeItemId: string,
            startedAt?: DateTime.Utc,
          ) {
            // Read the clock first so the lookup and the insert run without yielding.
            const now = startedAt ?? (yield* DateTime.now);
            const existing = context.itemPositions.get(nativeItemId);
            if (existing !== undefined) {
              return existing;
            }
            const position = {
              ordinal: context.providerTurnOrdinal * 100 + context.itemPositions.size + 1,
              startedAt: now,
            };
            context.itemPositions.set(nativeItemId, position);
            return position;
          });

        const resolveItemOrdinal = (context: ActiveCodexTurnContext, nativeItemId: string) =>
          resolveItemPosition(context, nativeItemId).pipe(
            Effect.map((position) => position.ordinal),
          );

        const nextProviderTurnOrdinal = (
          providerThreadId: OrchestrationV2ProviderThread["id"],
          minimum: number,
        ) =>
          Ref.modify(nextProviderTurnOrdinals, (current) => {
            const previous = current.get(String(providerThreadId));
            const next = previous === undefined ? minimum : Math.max(previous + 1, minimum);
            const updated = new Map(current);
            updated.set(String(providerThreadId), next);
            return [next, updated];
          });

        const emitSubagentTaskUpdate =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (input: {
            readonly subagent: CodexSubagentThreadContext;
            readonly status: OrchestrationV2Subagent["status"];
            readonly result?: string | null;
            readonly completedAt?: DateTime.Utc | null;
            readonly startedAt?: DateTime.Utc;
            readonly reopen?: boolean;
            readonly progress?: string;
          }) {
            const now = yield* DateTime.now;
            const prior = input.reopen
              ? (({ progress: _staleProgress, ...rest }) => rest)(input.subagent.task)
              : input.subagent.task;
            const settled = !isOrchestrationV2WorkActive(prior.status);
            // Late snapshots must not reopen completed children; only new turns may.
            if (settled && isOrchestrationV2WorkActive(input.status) && !input.reopen) return;
            const status = settled && !input.reopen ? prior.status : input.status;
            const completedAt = isOrchestrationV2WorkActive(status)
              ? null
              : (prior.completedAt ?? input.completedAt ?? now);
            const task = {
              ...prior,
              status,
              ...(input.reopen
                ? {
                    runId: input.subagent.parentContext.projectionRunId,
                    parentNodeId: input.subagent.parentContext.itemParentNodeId,
                  }
                : {}),
              startedAt: input.startedAt ?? prior.startedAt,
              ...(input.progress === undefined ? {} : { progress: input.progress }),
              result: input.result === undefined ? input.subagent.task.result : input.result,
              completedAt,
              updatedAt: now,
            } satisfies OrchestrationV2Subagent;
            input.subagent.task = task;
            if (input.startedAt !== undefined) input.subagent.startedAt = input.startedAt;

            yield* emitProviderEvent({
              type: "subagent.updated",
              driver: CODEX_PROVIDER,
              subagent: task,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: {
                id: input.subagent.turnItemId,
                threadId: task.threadId,
                runId: task.runId,
                nodeId: task.id,
                providerThreadId: task.providerThreadId,
                providerTurnId: input.subagent.parentContext.providerTurnId,
                nativeItemRef: task.nativeTaskRef,
                parentItemId: null,
                ordinal: input.subagent.turnItemOrdinal,
                status: task.status,
                title: task.title,
                startedAt: task.startedAt,
                completedAt: task.completedAt,
                updatedAt: task.updatedAt,
                type: "subagent",
                subagentId: task.id,
                origin: task.origin,
                driver: task.driver,
                providerInstanceId: task.providerInstanceId,
                childThreadId: task.childThreadId,
                prompt: task.prompt,
                ...(task.progress === undefined ? {} : { progress: task.progress }),
                result: task.result,
              },
            });
          });

        // A native child's running item is the line its parent's subagent row shows.
        const reportSubagentProgress =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (context: ActiveCodexTurnContext, progress: string | null) {
            const subagent = context.subagent;
            const line = progress?.trim();
            if (subagent === null || !line) return;
            if (!isOrchestrationV2WorkActive(subagent.task.status)) return;
            if (subagent.task.progress === line) return;
            yield* emitSubagentTaskUpdate({
              subagent,
              status: subagent.task.status,
              progress: line,
            });
          });

        const emitSubagentProviderTurnStarted =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (
            subagent: CodexSubagentThreadContext,
            turn: PendingCodexSubagentTurnStarted,
          ) {
            if (subagent.nativeTurnIds.has(turn.nativeTurnId)) return;
            subagent.nativeTurnIds.add(turn.nativeTurnId);
            const parentNativeThreadId = yield* getNativeThreadId(
              subagent.parentContext.providerThread,
            );
            const currentParent = yield* findActiveTurnByNativeThreadId(parentNativeThreadId);
            if (currentParent !== undefined) subagent.parentContext = currentParent;
            const terminalizedNativeTurns = yield* Ref.get(terminalizedNonCompletedNativeTurns);
            let ancestor: ActiveCodexTurnContext | undefined = subagent.parentContext;
            while (ancestor !== undefined) {
              if (terminalizedNativeTurns.has(ancestor.nativeTurnId)) {
                const nativeThreadId = yield* getNativeThreadId(subagent.providerThread);
                // Fork requests off the notification reader so it can read their responses.
                yield* client
                  .request("turn/interrupt", {
                    threadId: nativeThreadId,
                    turnId: turn.nativeTurnId,
                  })
                  .pipe(
                    Effect.catch((cause) =>
                      Effect.logWarning("orchestration-v2.codex-late-subagent-interrupt-failed", {
                        nativeThreadId,
                        nativeTurnId: turn.nativeTurnId,
                        cause,
                      }),
                    ),
                    Effect.timeoutOrElse({
                      duration: "10 seconds",
                      orElse: () =>
                        Effect.logWarning(
                          "orchestration-v2.codex-late-subagent-interrupt-timeout",
                          {
                            nativeThreadId,
                            nativeTurnId: turn.nativeTurnId,
                          },
                        ),
                    }),
                    Effect.forkIn(scope),
                  );
                return;
              }
              ancestor = ancestor.subagent?.parentContext;
            }
            const providerTurnId = idAllocator.derive.providerTurn({
              driver: CODEX_PROVIDER,
              nativeTurnId: turn.nativeTurnId,
            });
            const providerTurnOrdinal = yield* nextProviderTurnOrdinal(
              subagent.providerThread.id,
              1,
            );
            const providerNodeId =
              providerTurnOrdinal === 1
                ? subagent.childRootNodeId
                : idAllocator.derive.nodeFromProviderItem({
                    driver: CODEX_PROVIDER,
                    nativeItemId: `${turn.nativeTurnId}:thread-root`,
                  });
            const activeContext: ActiveCodexTurnContext = {
              input: subagent.parentContext.input,
              projectionAppThread: subagent.childThread,
              projectionThreadId: subagent.childThreadId,
              projectionRunId: null,
              nativeTurnId: turn.nativeTurnId,
              providerThread: subagent.providerThread,
              providerTurnId,
              providerTurnOrdinal,
              providerNodeId,
              providerNodeStartedAt: turn.startedAt,
              itemParentNodeId: providerNodeId,
              rootNodeId: providerNodeId,
              subagent,
              startedAt: turn.startedAt,
              itemPositions: new Map(),
              fileChangeKindsByItem: new Map(),
              messagePhases: new Map(),
              runningCompactionItemIds: new Set(),
            };
            beginTurnTokenUsage(activeContext);
            endSessionApprovalsOnModeChange(activeContext);
            yield* Ref.update(activeTurns, (current) => {
              const updated = new Map(current);
              updated.set(turn.nativeTurnId, activeContext);
              return updated;
            });
            yield* emitSubagentTaskUpdate({
              subagent,
              status: "running",
              startedAt: turn.startedAt,
              ...(providerTurnOrdinal > 1 ? { reopen: true, result: null } : {}),
            });
            const now = yield* DateTime.now;
            yield* emitProviderEvent({
              type: "provider_thread.updated",
              driver: CODEX_PROVIDER,
              providerThread: {
                ...subagent.providerThread,
                status: "active",
                updatedAt: now,
              },
            });
            yield* emitProviderEvent({
              type: "provider_turn.updated",
              driver: CODEX_PROVIDER,
              threadId: subagent.childThreadId,
              providerTurn: {
                id: providerTurnId,
                providerThreadId: subagent.providerThread.id,
                nodeId: providerNodeId,
                runAttemptId: null,
                nativeTurnRef: {
                  driver: CODEX_PROVIDER,
                  nativeId: turn.nativeTurnId,
                  strength: "strong",
                },
                ordinal: activeContext.providerTurnOrdinal,
                status: "running",
                startedAt: turn.startedAt,
                completedAt: null,
              },
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: {
                id: providerNodeId,
                threadId: subagent.childThreadId,
                runId: null,
                parentNodeId: null,
                rootNodeId: providerNodeId,
                kind: "root_turn",
                status: "running",
                countsForRun: false,
                providerThreadId: subagent.providerThread.id,
                providerTurnId,
                nativeItemRef: subagent.task.nativeTaskRef,
                runtimeRequestId: null,
                checkpointScopeId: null,
                startedAt: turn.startedAt,
                completedAt: null,
              },
            });
          });

        const rememberSubagentTurnStarted =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (input: {
            readonly nativeThreadId: string;
            readonly nativeTurnId: string;
            readonly startedAt: DateTime.Utc;
          }) {
            const subagent = (yield* Ref.get(subagentThreads)).get(input.nativeThreadId);
            if (subagent !== undefined) {
              yield* emitSubagentProviderTurnStarted(subagent, input);
              return;
            }
            yield* Ref.update(pendingSubagentTurns, (current) => {
              const updated = new Map(current);
              updated.set(input.nativeThreadId, [
                ...(updated.get(input.nativeThreadId) ?? []),
                { nativeTurnId: input.nativeTurnId, startedAt: input.startedAt },
              ]);
              return updated;
            });
          });

        const updateSubagentSelection = Effect.fnUntraced(function* (
          nativeThreadId: string,
          value: string | null,
          effort?: string | null,
          tier?: string | null,
        ) {
          const model = value?.trim();
          if (!model) return;
          const previous = subagentSelections.get(nativeThreadId);
          const previousOptions = previous?.options;
          const options =
            effort === undefined && tier === undefined
              ? previousOptions
              : [
                  ...(previousOptions ?? []).filter(({ id }) =>
                    id === "reasoningEffort"
                      ? effort === undefined
                      : id === "serviceTier" && tier === undefined,
                  ),
                  ...[
                    { id: "reasoningEffort", value: effort },
                    { id: "serviceTier", value: tier },
                  ].flatMap(({ id, value }) =>
                    value?.trim() ? [{ id, value: value.trim() }] : [],
                  ),
                ];
          const selection = {
            model,
            ...(options === undefined ? {} : { options }),
          };
          subagentSelections.set(nativeThreadId, selection);
          const subagent = (yield* Ref.get(subagentThreads)).get(nativeThreadId);
          if (subagent === undefined) return;
          const modelSelection = { instanceId: subagent.task.providerInstanceId, ...selection };
          if (
            subagent.task.modelSelection &&
            modelSelectionsEqual(subagent.task.modelSelection, modelSelection)
          )
            return;
          subagent.task = {
            ...subagent.task,
            model,
            modelSelection,
            updatedAt: yield* DateTime.now,
          };
          yield* emitProviderEvent({
            type: "subagent.updated",
            driver: CODEX_PROVIDER,
            subagent: subagent.task,
          });
        });

        const registerSubagentThread =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (input: {
            readonly context: ActiveCodexTurnContext;
            readonly nativeThreadId: string;
            readonly nativeItemId: string;
            readonly nativeToolCallId: string;
            readonly prompt: string;
            readonly title: string | null;
            readonly model: string | null;
            readonly reasoningEffort?: string | null | undefined;
            readonly ordinal: number;
            readonly emitInitialPrompt: boolean;
          }) {
            const registeredSubagents = yield* Ref.get(subagentThreads);
            if (input.model && !subagentSelections.has(input.nativeThreadId)) {
              yield* updateSubagentSelection(
                input.nativeThreadId,
                input.model,
                input.reasoningEffort,
              );
            }
            if (registeredSubagents.has(input.nativeThreadId)) {
              return;
            }

            const now = yield* DateTime.now;
            const subagentNodeId = idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: input.nativeItemId,
            });
            const childRootNodeId = idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: `${input.nativeItemId}:thread-root`,
            });
            const childThreadId = idAllocator.derive.subagentThread({
              driver: CODEX_PROVIDER,
              nativeThreadId: input.nativeThreadId,
            });
            const turnItemOrdinal = yield* resolveItemOrdinal(input.context, input.nativeItemId);
            const providerThread = {
              id: idAllocator.derive.providerThread({
                driver: CODEX_PROVIDER,
                nativeThreadId: input.nativeThreadId,
              }),
              driver: CODEX_PROVIDER,
              providerInstanceId: input.context.input.modelSelection.instanceId,
              providerSessionId: input.context.providerThread.providerSessionId,
              appThreadId: childThreadId,
              ownerNodeId: null,
              nativeThreadRef: {
                driver: CODEX_PROVIDER,
                nativeId: input.nativeThreadId,
                strength: "strong" as const,
              },
              nativeConversationHeadRef: null,
              status: "idle" as const,
              firstRunOrdinal: null,
              lastRunOrdinal: null,
              handoffIds: [],
              forkedFrom: {
                providerThreadId: input.context.providerThread.id,
                providerTurnId: input.context.providerTurnId,
              },
              createdAt: now,
              updatedAt: now,
            } satisfies OrchestrationV2ProviderThread;
            const reportedSelection = subagentSelections.get(input.nativeThreadId);
            const task = {
              id: subagentNodeId,
              threadId: input.context.projectionThreadId,
              runId: input.context.projectionRunId,
              parentNodeId: input.context.itemParentNodeId,
              origin: "provider_native",
              createdBy: "agent",
              driver: CODEX_PROVIDER,
              providerInstanceId: input.context.input.modelSelection.instanceId,
              providerThreadId: providerThread.id,
              childThreadId,
              nativeTaskRef: codexNativeItemRef(input.nativeItemId),
              prompt: input.prompt,
              title: input.title,
              model: reportedSelection?.model ?? input.model,
              modelSelection: reportedSelection
                ? {
                    instanceId: input.context.input.modelSelection.instanceId,
                    ...reportedSelection,
                  }
                : undefined,
              status: "running",
              result: null,
              startedAt: now,
              completedAt: null,
              updatedAt: now,
            } satisfies OrchestrationV2Subagent;
            const childThread = makeSubagentChildThread({
              parentThread: input.context.projectionAppThread,
              childThreadId,
              parentNodeId: subagentNodeId,
              activeProviderThreadId: providerThread.id,
              providerInstanceId: input.context.input.modelSelection.instanceId,
              // A child's effort can differ from the parent's. Until Codex reports it, show none.
              modelSelection: task.modelSelection ?? {
                ...input.context.input.modelSelection,
                model: task.model ?? input.context.input.modelSelection.model,
                options: (input.context.input.modelSelection.options ?? []).filter(
                  ({ id }) => id !== "reasoningEffort",
                ),
              },
              title: subagentThreadTitle({
                parentTitle: input.context.projectionAppThread.title,
                prompt: task.prompt,
                title: task.title,
                ordinal: input.ordinal,
              }),
              now,
              createdBy: "agent",
              creationSource: "provider",
            });
            const subagent = {
              parentContext: input.context,
              providerThread,
              childThread,
              subagentNodeId,
              childRootNodeId,
              childThreadId,
              nativeTurnIds: new Set<string>(),
              startedAt: now,
              turnItemId: idAllocator.derive.turnItemFromProviderItem({
                driver: CODEX_PROVIDER,
                nativeItemId: input.nativeItemId,
              }),
              turnItemOrdinal,
              task,
            } satisfies CodexSubagentThreadContext;

            yield* Ref.update(subagentThreads, (current) => {
              const updated = new Map(current);
              updated.set(input.nativeThreadId, subagent);
              return updated;
            });
            yield* emitProviderEvent({
              type: "app_thread.created",
              driver: CODEX_PROVIDER,
              appThread: childThread,
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: {
                id: subagentNodeId,
                threadId: input.context.projectionThreadId,
                runId: input.context.projectionRunId,
                parentNodeId: input.context.itemParentNodeId,
                rootNodeId: input.context.rootNodeId,
                kind: "subagent",
                status: "running",
                countsForRun: false,
                providerThreadId: providerThread.id,
                providerTurnId: input.context.providerTurnId,
                nativeItemRef: codexNativeItemRef(input.nativeToolCallId),
                runtimeRequestId: null,
                checkpointScopeId: null,
                startedAt: now,
                completedAt: null,
              },
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: {
                id: childRootNodeId,
                threadId: childThreadId,
                runId: null,
                parentNodeId: null,
                rootNodeId: childRootNodeId,
                kind: "root_turn",
                status: "running",
                countsForRun: false,
                providerThreadId: providerThread.id,
                providerTurnId: null,
                nativeItemRef: codexNativeItemRef(input.nativeItemId),
                runtimeRequestId: null,
                checkpointScopeId: null,
                startedAt: now,
                completedAt: null,
              },
            });
            yield* emitProviderEvent({
              type: "provider_thread.updated",
              driver: CODEX_PROVIDER,
              providerThread,
            });
            if (input.emitInitialPrompt && input.prompt.length > 0) {
              const promptNativeItemId = `${input.nativeItemId}:prompt`;
              const promptArtifacts = makeSubagentConversationArtifacts({
                senderThreadId: input.context.projectionThreadId,
                messageId: idAllocator.derive.messageFromProviderItem({
                  driver: CODEX_PROVIDER,
                  nativeItemId: promptNativeItemId,
                }),
                turnItemId: idAllocator.derive.turnItemFromProviderItem({
                  driver: CODEX_PROVIDER,
                  nativeItemId: promptNativeItemId,
                }),
                threadId: childThreadId,
                rootNodeId: childRootNodeId,
                providerThreadId: providerThread.id,
                providerTurnId: null,
                nativeItemRef: codexNativeItemRef(promptNativeItemId),
                role: "user",
                text: input.prompt,
                ordinal: 100,
                now,
              });
              yield* emitProviderEvent({
                type: "message.updated",
                driver: CODEX_PROVIDER,
                message: promptArtifacts.message,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: promptArtifacts.turnItem,
              });
            }
            yield* emitSubagentTaskUpdate({
              subagent,
              status: "running",
            });

            const pendingTurns = yield* Ref.modify(pendingSubagentTurns, (current) => {
              const pending = current.get(input.nativeThreadId) ?? [];
              const updated = new Map(current);
              updated.delete(input.nativeThreadId);
              return [pending, updated];
            });
            for (const pendingTurn of pendingTurns) {
              yield* emitSubagentProviderTurnStarted(subagent, pendingTurn);
            }
            if (task.model === null) {
              yield* client.raw
                .request("thread/read", { threadId: input.nativeThreadId, includeTurns: false })
                .pipe(
                  Effect.flatMap(decodeCodexChildThread),
                  Effect.map((response): typeof CodexChildSelection.Type | null =>
                    response.thread.id === input.nativeThreadId && response.thread.model?.trim()
                      ? {
                          thread: response.thread,
                          model: response.thread.model,
                          reasoningEffort: response.thread.reasoningEffort,
                        }
                      : null,
                  ),
                  Effect.orElseSucceed(() => null),
                  Effect.filterOrElse(
                    (response) => !(response === null),
                    (_response) =>
                      client.raw
                        .request("thread/resume", {
                          threadId: input.nativeThreadId,
                          excludeTurns: true,
                        })
                        .pipe(Effect.flatMap(decodeCodexChildModel)),
                  ),
                  Effect.timeout("5 seconds"),
                  Effect.flatMap((response) =>
                    response.thread.id === input.nativeThreadId &&
                    !subagentSelections.has(input.nativeThreadId)
                      ? updateSubagentSelection(
                          input.nativeThreadId,
                          response.model,
                          response.reasoningEffort,
                          response.serviceTier,
                        )
                      : Effect.void,
                  ),
                  Effect.ignore,
                  Effect.forkIn(scope),
                );
            }
          });

        const registerSubagentThreads =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (input: {
            readonly context: ActiveCodexTurnContext;
            readonly item: CodexCollabAgentToolCallItem;
          }) {
            if (input.item.tool !== "spawnAgent" || input.item.receiverThreadIds.length === 0) {
              return;
            }

            for (const [index, nativeThreadId] of input.item.receiverThreadIds.entries()) {
              const model =
                typeof input.item.model === "string" && input.item.model.length > 0
                  ? input.item.model
                  : null;
              yield* registerSubagentThread({
                context: input.context,
                nativeThreadId,
                nativeItemId: `${input.item.id}:${nativeThreadId}`,
                nativeToolCallId: input.item.id,
                prompt: input.item.prompt ?? "",
                title: null,
                model,
                reasoningEffort: input.item.reasoningEffort,
                ordinal: index + 1,
                emitInitialPrompt: true,
              });
            }
          });

        const registerSubagentActivity =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (input: {
            readonly context?: ActiveCodexTurnContext;
            readonly item: CodexSubAgentActivityItem;
          }) {
            if (input.item.kind === "started") {
              const context = input.context;
              if (context === undefined) return;
              const registeredSubagents = yield* Ref.get(subagentThreads);
              const ordinal =
                Array.from(registeredSubagents.values()).filter(
                  (subagent) => subagent.parentContext.rootNodeId === context.rootNodeId,
                ).length + 1;
              yield* registerSubagentThread({
                context,
                nativeThreadId: input.item.agentThreadId,
                nativeItemId: `${input.item.id}:${input.item.agentThreadId}`,
                nativeToolCallId: input.item.id,
                prompt: "",
                title: input.item.agentPath,
                model: null,
                ordinal,
                emitInitialPrompt: false,
              });
              return;
            }

            const subagent = (yield* Ref.get(subagentThreads)).get(input.item.agentThreadId);
            if (subagent === undefined) {
              return;
            }

            if (input.item.kind === "interrupted" || input.item.kind === "completed") {
              yield* emitSubagentTaskUpdate({
                subagent,
                status: input.item.kind,
              });
            }
          });

        const updateSubagentStates =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (input: { readonly item: CodexCollabAgentToolCallItem }) {
            const subagents = yield* Ref.get(subagentThreads);
            for (const [nativeThreadId, state] of Object.entries(input.item.agentsStates)) {
              const subagent = subagents.get(nativeThreadId);
              if (subagent === undefined) {
                continue;
              }
              const statuses = {
                pendingInit: "pending",
                running: "running",
                interrupted: "interrupted",
                completed: "completed",
                errored: "failed",
                shutdown: "cancelled",
                notFound: "failed",
              } as const satisfies Record<
                CodexSchema.ServerNotification__CollabAgentStatus,
                OrchestrationV2Subagent["status"]
              >;
              const status = statuses[state.status];
              yield* emitSubagentTaskUpdate({
                subagent,
                status,
                ...(state.message === null ? {} : { result: state.message }),
              });
            }
          });

        const resolvePlanId =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (context: ActiveCodexTurnContext, planKey: string) {
            const existing = (yield* Ref.get(planIds)).get(planKey);
            if (existing !== undefined) {
              return existing;
            }
            const planId = yield* idAllocator.allocate.plan({
              threadId: context.projectionThreadId,
              ...(context.projectionRunId === null ? {} : { runId: context.projectionRunId }),
              driver: CODEX_PROVIDER,
            });
            yield* Ref.update(planIds, (current) => {
              const updated = new Map(current);
              updated.set(planKey, planId);
              return updated;
            });
            return planId;
          });

        const resolveCodexAttachment = Effect.fn("CodexAdapterV2.resolveCodexAttachment")(
          function* (attachment: ChatAttachment) {
            const attachmentPath = resolveAttachmentPath({
              attachmentsDir: serverConfig.attachmentsDir,
              attachment,
            });
            if (attachmentPath === null) {
              return yield* toProtocolError(`Invalid attachment id '${attachment.id}'`);
            }
            const bytes = yield* fileSystem.readFile(attachmentPath).pipe(
              Effect.mapError(
                (cause) =>
                  new ProviderAdapter.ProviderAdapterProtocolError({
                    driver: CODEX_PROVIDER,
                    detail: `Failed to read attachment '${attachment.id}'.`,
                    reason: "attachment_unreadable",
                    payload: cause,
                  }),
              ),
            );
            return {
              type: "image" as const,
              url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`,
            } satisfies CodexSchema.V2TurnStartParams__UserInput;
          },
        );

        const toCodexInput = Effect.fn("CodexAdapterV2.toCodexInput")(function* (
          turnInput: Pick<
            | ProviderAdapter.ProviderAdapterV2TurnInput
            | ProviderAdapter.ProviderAdapterV2SteerInput,
            "message"
          >,
        ) {
          const inputItems: Array<CodexSchema.V2TurnStartParams__UserInput> = [];
          const text = yield* providerMessageTextWithAttachmentPaths({
            text: codexSkillMentionText(turnInput.message.text),
            attachments: turnInput.message.attachments,
            resolveExtractedTextPath: (path) =>
              DocumentText.resolveExtractedTextPath(path).pipe(
                Effect.provideService(FileSystem.FileSystem, fileSystem),
              ),
            resolveAttachmentPath: (attachment) =>
              resolveAttachmentPath({ attachmentsDir: serverConfig.attachmentsDir, attachment }),
          });
          if (text.length > 0) {
            inputItems.push({
              type: "text",
              text,
            });
          }
          const attachmentItems = yield* Effect.forEach(
            turnInput.message.attachments.filter(isProviderNativeImageAttachment),
            resolveCodexAttachment,
            { concurrency: 1 },
          );
          inputItems.push(...attachmentItems);
          if (inputItems.length === 0) {
            return yield* toProtocolError("Turn requires non-empty text or attachments.");
          }
          return inputItems;
        });

        // Reasoning and assistant-text coalescers build these artifacts on every streamed update.
        const buildAgentMessageArtifacts = Effect.fnUntraced(function* (
          context: ActiveCodexTurnContext,
          item: { readonly id: string; readonly text: string },
          completed: boolean,
        ) {
          const updatedAt = yield* DateTime.now;
          const completedAt = completed ? updatedAt : null;
          const nodeId = idAllocator.derive.nodeFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: item.id,
          });
          const { ordinal, startedAt } = yield* resolveItemPosition(context, item.id);
          const messageId = idAllocator.derive.messageFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: item.id,
          });
          const turnItemId = idAllocator.derive.turnItemFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: item.id,
          });
          const node: OrchestrationV2ExecutionNode = {
            id: nodeId,
            threadId: context.projectionThreadId,
            runId: context.projectionRunId,
            parentNodeId: context.itemParentNodeId,
            rootNodeId: context.rootNodeId,
            kind: "assistant_message",
            status: completed ? "completed" : "running",
            countsForRun: false,
            providerThreadId: context.providerThread.id,
            providerTurnId: context.providerTurnId,
            nativeItemRef: codexNativeItemRef(item.id),
            runtimeRequestId: null,
            checkpointScopeId: null,
            startedAt,
            completedAt,
          };
          const message: OrchestrationV2ConversationMessage = {
            createdBy: "agent",
            creationSource: "provider",
            id: messageId,
            threadId: context.projectionThreadId,
            runId: context.projectionRunId,
            nodeId,
            role: "assistant",
            text: item.text,
            attachments: [],
            streaming: !completed,
            createdAt: startedAt,
            updatedAt,
          };
          const turnItem: OrchestrationV2TurnItem = {
            id: turnItemId,
            threadId: context.projectionThreadId,
            runId: context.projectionRunId,
            nodeId,
            providerThreadId: context.providerThread.id,
            providerTurnId: context.providerTurnId,
            nativeItemRef: codexNativeItemRef(item.id),
            parentItemId: null,
            ordinal,
            status: completed ? "completed" : "running",
            title: null,
            startedAt,
            completedAt,
            updatedAt,
            type: "assistant_message",
            ...(context.messagePhases.has(item.id)
              ? { phase: context.messagePhases.get(item.id) }
              : {}),
            messageId,
            text: item.text,
            streaming: !completed,
          };
          return { node, message, turnItem };
        });

        // Coalesce token bursts to avoid a database write per token.
        const reasoningParts = new Map<
          string,
          { turnId: string; nativeItemId: string; stream: "summary" | "content"; index: number }
        >();
        const reasoningDeltas = yield* makeProviderTextDeltaCoalescer({
          flushIntervalMs: CODEX_ASSISTANT_DELTA_FLUSH_INTERVAL_MS,
          emit:
            /* Provider events include streaming deltas; avoid a span for each update. */
            Effect.fnUntraced(function* (update) {
              const part = reasoningParts.get(update.itemId);
              if (part === undefined) return;
              const context = yield* awaitActiveTurn(update.turnId);
              if (context === undefined) return;
              const artifacts = yield* buildAgentMessageArtifacts(
                context,
                { id: update.itemId, text: update.text },
                update.completed,
              );
              const interrupted = (yield* Ref.get(terminalizedNonCompletedNativeTurns)).has(
                update.turnId,
              );
              const { messageId: _messageId, ...item } = artifacts.turnItem;
              const nativeItemRef = codexNativeItemRef(part.nativeItemId);
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: {
                  ...artifacts.node,
                  kind: "reasoning",
                  nativeItemRef,
                  ...(interrupted ? { status: "interrupted" as const } : {}),
                },
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: {
                  ...item,
                  type: "reasoning",
                  nativeItemRef,
                  ...(interrupted ? { status: "interrupted" as const } : {}),
                },
              });
            }),
        });
        const reasoningPartKey = (
          turnId: string,
          nativeItemId: string,
          stream: "summary" | "content",
          index: number,
        ) => {
          const key = JSON.stringify([turnId, nativeItemId, stream, index]);
          reasoningParts.set(key, { turnId, nativeItemId, stream, index });
          return key;
        };
        const appendReasoning = Effect.fn("CodexAdapterV2.appendReasoning")(function* (
          payload: { turnId: string; itemId: string; delta: string },
          stream: "summary" | "content",
          index: number,
        ) {
          const context = yield* awaitActiveTurn(payload.turnId);
          if (context === undefined || payload.delta.length === 0) return;
          yield* completeProviderRetry(context, yield* DateTime.now);
          const itemId = reasoningPartKey(payload.turnId, payload.itemId, stream, index);
          // Reserve positions before delayed flushes so later tools stay after them.
          yield* resolveItemOrdinal(context, itemId);
          yield* reasoningDeltas.append({ turnId: payload.turnId, itemId, delta: payload.delta });
        });
        const completeReasoning = Effect.fn("CodexAdapterV2.completeReasoning")(function* (
          turnId: string,
          item: { id: string; summary?: ReadonlyArray<string>; content?: ReadonlyArray<string> },
        ) {
          for (const stream of ["summary", "content"] as const) {
            for (const [index] of (item[stream] ?? []).entries()) {
              reasoningPartKey(turnId, item.id, stream, index);
            }
          }
          for (const [key, part] of reasoningParts) {
            if (part.turnId !== turnId || part.nativeItemId !== item.id) continue;
            const finalText = item[part.stream]?.[part.index];
            yield* reasoningDeltas.complete({
              turnId,
              itemId: key,
              ...(finalText ? { finalText } : {}),
              emitEmpty: false,
            });
            reasoningParts.delete(key);
          }
        });

        const agentMessageDeltas = yield* makeProviderTextDeltaCoalescer({
          flushIntervalMs: CODEX_ASSISTANT_DELTA_FLUSH_INTERVAL_MS,
          emit:
            /* Provider events include streaming deltas; avoid a span for each update. */
            Effect.fnUntraced(function* (update) {
              const context = yield* awaitActiveTurn(update.turnId);
              if (context === undefined) {
                return;
              }
              const finalAnswerItem = (yield* Ref.get(finalAnswerItemIdsByTurn))
                .get(update.turnId)
                ?.has(update.itemId);
              if (finalAnswerItem) {
                const finalAnswerItemIds = (yield* Ref.get(finalAnswerItemIdsByTurn)).get(
                  update.turnId,
                );
                const completedTexts =
                  (yield* Ref.get(completedFinalAnswerTextsByTurn)).get(update.turnId) ??
                  new Set<string>();
                const firstFinalAnswerItemId = finalAnswerItemIds?.values().next().value;
                const duplicateCompletion =
                  update.completed &&
                  completedTexts.size > 0 &&
                  (update.text.length === 0 || completedTexts.has(update.text));
                const deferredStreamingUpdate =
                  !update.completed &&
                  (completedTexts.size > 0 || firstFinalAnswerItemId !== update.itemId);
                if (deferredStreamingUpdate || duplicateCompletion) {
                  return;
                }
              }
              const artifacts = yield* buildAgentMessageArtifacts(
                context,
                { id: update.itemId, text: update.text },
                update.completed,
              );
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              if (update.completed) {
                yield* emitProviderEvent({
                  type: "message.updated",
                  driver: CODEX_PROVIDER,
                  message: artifacts.message,
                });
              }
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              if (finalAnswerItem && update.completed) {
                imageTurn(update.turnId).answer = { id: update.itemId, text: update.text };
                yield* Ref.update(completedFinalAnswerTextsByTurn, (current) => {
                  const updated = new Map(current);
                  const texts = new Set(updated.get(update.turnId) ?? []);
                  texts.add(update.text);
                  updated.set(update.turnId, texts);
                  return updated;
                });
              }
            }),
        });

        const emitSubagentUserMessage =
          /* Provider events include streaming deltas; avoid a span for each update. */
          Effect.fnUntraced(function* (
            context: ActiveCodexTurnContext,
            item: Extract<
              | CodexSchema.V2ItemStartedNotification__ThreadItem
              | CodexSchema.V2ItemCompletedNotification__ThreadItem,
              { type: "userMessage" }
            >,
            nativeStartedAt?: DateTime.Utc,
          ) {
            if (context.subagent === null || context.providerTurnOrdinal === 1) {
              return false;
            }
            const text = codexUserMessageText(item.content);
            if (text.length === 0) {
              return false;
            }
            const { ordinal, startedAt } = yield* resolveItemPosition(
              context,
              item.id,
              nativeStartedAt,
            );
            const artifacts = makeSubagentConversationArtifacts({
              senderThreadId: context.subagent.parentContext.projectionThreadId,
              messageId: idAllocator.derive.messageFromProviderItem({
                driver: CODEX_PROVIDER,
                nativeItemId: item.id,
              }),
              turnItemId: idAllocator.derive.turnItemFromProviderItem({
                driver: CODEX_PROVIDER,
                nativeItemId: item.id,
              }),
              threadId: context.projectionThreadId,
              rootNodeId: context.rootNodeId,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(item.id),
              role: "user",
              text,
              ordinal,
              now: startedAt,
            });
            yield* emitProviderEvent({
              type: "message.updated",
              driver: CODEX_PROVIDER,
              message: artifacts.message,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            return true;
          });

        const buildCommandExecutionArtifacts = Effect.fn(
          "CodexAdapterV2.buildCommandExecutionArtifacts",
        )(function* (
          context: ActiveCodexTurnContext,
          item: Extract<
            | CodexSchema.V2ItemStartedNotification__ThreadItem
            | CodexSchema.V2ItemCompletedNotification__ThreadItem,
            { type: "commandExecution" }
          >,
        ) {
          const updatedAt = yield* DateTime.now;
          const status = codexItemStatus(item.status);
          const completedAt = status.completed ? updatedAt : null;
          const nodeId = idAllocator.derive.nodeFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: item.id,
          });
          const turnItemId = idAllocator.derive.turnItemFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: item.id,
          });
          const { ordinal, startedAt } = yield* resolveItemPosition(context, item.id);
          const node: OrchestrationV2ExecutionNode = {
            id: nodeId,
            threadId: context.projectionThreadId,
            runId: context.projectionRunId,
            parentNodeId: context.itemParentNodeId,
            rootNodeId: context.rootNodeId,
            kind: "tool_call",
            status: status.node,
            countsForRun: false,
            providerThreadId: context.providerThread.id,
            providerTurnId: context.providerTurnId,
            nativeItemRef: codexNativeItemRef(item.id),
            runtimeRequestId: null,
            checkpointScopeId: null,
            startedAt,
            completedAt,
          };
          const turnItem: OrchestrationV2TurnItem = {
            id: turnItemId,
            threadId: context.projectionThreadId,
            runId: context.projectionRunId,
            nodeId,
            providerThreadId: context.providerThread.id,
            providerTurnId: context.providerTurnId,
            nativeItemRef: codexNativeItemRef(item.id),
            parentItemId: null,
            ordinal,
            status: status.turnItem,
            title: null,
            startedAt,
            completedAt,
            updatedAt,
            type: "command_execution",
            input: item.command,
            ...(item.aggregatedOutput === null || item.aggregatedOutput === undefined
              ? {}
              : { output: item.aggregatedOutput }),
            ...(item.exitCode === null || item.exitCode === undefined
              ? {}
              : { exitCode: item.exitCode }),
            // Matches the continuation offered when it completes after the turn.
            ...(context.subagent === null && continuationRequests !== undefined
              ? { completionWake: "always" as const }
              : {}),
          };
          return { node, turnItem };
        });

        const buildFileChangeArtifacts = Effect.fn("CodexAdapterV2.buildFileChangeArtifacts")(
          function* (
            context: ActiveCodexTurnContext,
            item: Extract<
              CodexSchema.V2ItemCompletedNotification__ThreadItem,
              { type: "fileChange" }
            >,
          ) {
            const firstChange = item.changes[0];
            if (firstChange === undefined) {
              return null;
            }

            const updatedAt = yield* DateTime.now;
            const status = codexItemStatus(item.status);
            const completedAt = status.completed ? updatedAt : null;
            const nodeId = idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: item.id,
            });
            const turnItemId = idAllocator.derive.turnItemFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: item.id,
            });
            const { ordinal, startedAt } = yield* resolveItemPosition(context, item.id);
            const node: OrchestrationV2ExecutionNode = {
              id: nodeId,
              threadId: context.projectionThreadId,
              runId: context.projectionRunId,
              parentNodeId: context.itemParentNodeId,
              rootNodeId: context.rootNodeId,
              kind: "tool_call",
              status: status.node,
              countsForRun: false,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(item.id),
              runtimeRequestId: null,
              checkpointScopeId: null,
              startedAt,
              completedAt,
            };
            const turnItem: OrchestrationV2TurnItem = {
              id: turnItemId,
              threadId: context.projectionThreadId,
              runId: context.projectionRunId,
              nodeId,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(item.id),
              parentItemId: null,
              ordinal,
              status: status.turnItem,
              title: null,
              startedAt,
              completedAt,
              updatedAt,
              type: "file_change",
              fileName: firstChange.path,
              diffStr: firstChange.diff,
              changes: item.changes.map((change) => {
                const movePath = change.kind.type === "update" ? change.kind.move_path : undefined;
                return movePath
                  ? { operation: change.kind.type, path: movePath, oldPath: change.path }
                  : { operation: change.kind.type, path: change.path };
              }),
            };
            return { node, turnItem };
          },
        );

        const buildWebSearchArtifacts = Effect.fn("CodexAdapterV2.buildWebSearchArtifacts")(
          function* (input: {
            readonly context: ActiveCodexTurnContext;
            readonly item: CodexWebSearchItem;
            readonly completed: boolean;
          }) {
            const updatedAt = yield* DateTime.now;
            const completedAt = input.completed ? updatedAt : null;
            const nodeId = idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: input.item.id,
            });
            const turnItemId = idAllocator.derive.turnItemFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: input.item.id,
            });
            const { ordinal, startedAt } = yield* resolveItemPosition(input.context, input.item.id);
            const patterns = webSearchPatterns(input.item);
            const status = input.completed ? "completed" : "running";
            const node: OrchestrationV2ExecutionNode = {
              id: nodeId,
              threadId: input.context.projectionThreadId,
              runId: input.context.projectionRunId,
              parentNodeId: input.context.itemParentNodeId,
              rootNodeId: input.context.rootNodeId,
              kind: "tool_call",
              status,
              countsForRun: false,
              providerThreadId: input.context.providerThread.id,
              providerTurnId: input.context.providerTurnId,
              nativeItemRef: codexNativeItemRef(input.item.id),
              runtimeRequestId: null,
              checkpointScopeId: null,
              startedAt,
              completedAt,
            };
            const turnItem: OrchestrationV2TurnItem = {
              id: turnItemId,
              threadId: input.context.projectionThreadId,
              runId: input.context.projectionRunId,
              nodeId,
              providerThreadId: input.context.providerThread.id,
              providerTurnId: input.context.providerTurnId,
              nativeItemRef: codexNativeItemRef(input.item.id),
              parentItemId: null,
              ordinal,
              status,
              title: null,
              startedAt,
              completedAt,
              updatedAt,
              type: "web_search",
              ...(patterns.length === 0 ? {} : { patterns: [...patterns] }),
            };
            return { node, turnItem };
          },
        );

        const buildDynamicToolArtifacts = Effect.fn("CodexAdapterV2.buildDynamicToolArtifacts")(
          function* (context: ActiveCodexTurnContext, item: CodexDynamicToolItem) {
            const updatedAt = yield* DateTime.now;
            const status = codexItemStatus(item.status);
            const completedAt = status.completed ? updatedAt : null;
            const nodeId = idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: item.id,
            });
            const turnItemId = idAllocator.derive.turnItemFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: item.id,
            });
            const { ordinal, startedAt } = yield* resolveItemPosition(context, item.id);
            const projection = yield* projectCodexDynamicToolItem(item).pipe(
              Effect.provideService(Digest.Digest, digest),
            );
            const node: OrchestrationV2ExecutionNode = {
              id: nodeId,
              threadId: context.projectionThreadId,
              runId: context.projectionRunId,
              parentNodeId: context.itemParentNodeId,
              rootNodeId: context.rootNodeId,
              kind: "tool_call",
              status: status.node,
              countsForRun: false,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(item.id),
              runtimeRequestId: null,
              checkpointScopeId: null,
              startedAt,
              completedAt,
            };
            const turnItem: OrchestrationV2TurnItem = {
              id: turnItemId,
              threadId: context.projectionThreadId,
              runId: context.projectionRunId,
              nodeId,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(item.id),
              parentItemId: null,
              ordinal,
              title: projection.title ?? null,
              startedAt,
              completedAt,
              updatedAt,
              type: "dynamic_tool",
              ...projection,
            };
            return { node, turnItem };
          },
        );

        const emitImageGenerationItem = Effect.fn("CodexAdapterV2.emitImageGenerationItem")(
          function* (
            context: ActiveCodexTurnContext,
            item: Extract<
              CodexSchema.V2ItemCompletedNotification__ThreadItem,
              { type: "imageGeneration" }
            >,
            completed: boolean,
          ) {
            const cancelled = completed && item.status === "cancelled";
            const failed = item.status === "failed" || item.failure != null;
            const artifacts = yield* buildDynamicToolArtifacts(context, {
              type: "dynamicToolCall",
              id: item.id,
              namespace: "image_gen",
              tool: "imagegen",
              status: completed ? (failed ? "failed" : "completed") : "inProgress",
              arguments: {
                ...(item.revisedPrompt == null ? {} : { prompt: item.revisedPrompt }),
                ...(item.transparentBackground == null
                  ? {}
                  : { transparent_background: item.transparentBackground }),
              },
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: { ...artifacts.node, ...(cancelled ? { status: "cancelled" } : {}) },
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: {
                ...artifacts.turnItem,
                title: "Generate image",
                ...(cancelled ? { status: "cancelled" } : {}),
                ...(completed
                  ? {
                      output: {
                        status: item.status,
                        ...(item.savedPath == null ? {} : { savedPath: item.savedPath }),
                        ...(item.failure == null ? {} : { failure: item.failure }),
                      },
                    }
                  : {}),
              },
            });
          },
        );

        /**
         * The item stays running until then, which keeps the run's event
         * ingestion open past turn/completed; it runs off the notification
         * reader so a slow server cannot stall other events.
         */
        const captureMcpApp = Effect.fn("CodexAdapterV2.captureMcpApp")(function* (
          context: ActiveCodexTurnContext,
          nativeTurnId: string,
          item: Extract<CodexDynamicToolItem, { readonly type: "mcpToolCall" }>,
          resourceUri: string,
        ) {
          // Registered before it can run, so a turn ending right away sees it.
          const started = yield* Deferred.make<void>();
          const fiber = yield* Deferred.await(started).pipe(
            Effect.andThen(captureMcpAppItem(context, nativeTurnId, item, resourceUri)),
            Effect.forkIn(scope, { startImmediately: true }),
          );
          yield* Ref.update(pendingMcpAppCaptures, (current) => {
            const updated = new Map(current);
            updated.set(
              nativeTurnId,
              new Map(current.get(nativeTurnId) ?? []).set(item.id, { item, fiber }),
            );
            return updated;
          });
          yield* Deferred.succeed(started, undefined);
        });

        const captureMcpAppItem = (
          context: ActiveCodexTurnContext,
          nativeTurnId: string,
          item: Extract<CodexDynamicToolItem, { readonly type: "mcpToolCall" }>,
          resourceUri: string,
        ) =>
          // Only reads and storage are cancellable; remove partial writes. Once stored, attach the app or delete it if the item settled.
          Effect.uninterruptibleMask((restore) =>
            Effect.gen(function* () {
              const reference =
                item.result != null && item.error == null
                  ? yield* restore(
                      getNativeThreadId(context.providerThread).pipe(
                        Effect.flatMap((threadId) =>
                          client.request("mcpServer/resource/read", {
                            threadId,
                            server: item.server,
                            uri: resourceUri,
                          }),
                        ),
                        Effect.timeout(MCP_APP_CAPTURE_TIMEOUT),
                        Effect.flatMap((response) =>
                          snapshotMcpApp({
                            attachmentsDir: serverConfig.attachmentsDir,
                            threadId: context.projectionThreadId,
                            server: item.server,
                            tool: item.tool,
                            resourceUri,
                            contents: response.contents,
                          }),
                        ),
                        Effect.provideService(FileSystem.FileSystem, fileSystem),
                        // Convert failures and defects to plain tool rows; only interruption ends the capture fiber.
                        Effect.catchCauseIf(
                          (cause) => !Cause.hasInterruptsOnly(cause),
                          (cause) =>
                            Effect.logWarning(
                              "Failed to capture an MCP app; showing a plain tool row.",
                              { server: item.server, tool: item.tool, resourceUri, cause },
                            ).pipe(Effect.as(undefined)),
                        ),
                        Effect.orDie,
                      ),
                    )
                  : undefined;
              const artifacts = yield* buildDynamicToolArtifacts(context, item);
              // Clear before emitting; ingestion rechecks pending work after each event.
              if (!(yield* takePendingMcpAppCapture(nativeTurnId, item.id))) {
                if (reference !== undefined) {
                  const filePath = resolveAttachmentPathById({
                    attachmentsDir: serverConfig.attachmentsDir,
                    attachmentId: reference.attachmentId,
                  });
                  if (filePath !== null) {
                    yield* fileSystem.remove(filePath, { force: true }).pipe(Effect.ignore);
                  }
                }
                return;
              }
              const turnItem =
                reference === undefined || artifacts.turnItem.type !== "dynamic_tool"
                  ? artifacts.turnItem
                  : {
                      ...artifacts.turnItem,
                      output: {
                        [MCP_APP_OUTPUT_KEY]: reference,
                        result:
                          item.status === "failed"
                            ? { ...item.result, isError: true }
                            : item.result,
                      },
                    };
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem,
              });
              if ((yield* Ref.get(settledTurns)).has(nativeTurnId)) {
                yield* releaseSettledTurnIfIdle(nativeTurnId);
              }
            }),
          );

        const terminalizeMcpAppCaptures = Effect.fn("CodexAdapterV2.terminalizeMcpAppCaptures")(
          function* (context: ActiveCodexTurnContext, nativeTurnId: string) {
            const captures = (yield* Ref.get(pendingMcpAppCaptures)).get(nativeTurnId);
            if (captures === undefined) return;
            for (const { item, fiber } of captures.values()) {
              if (!(yield* takePendingMcpAppCapture(nativeTurnId, item.id))) continue;
              yield* Fiber.interrupt(fiber);
              const artifacts = yield* buildDynamicToolArtifacts(context, item);
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
            }
          },
        );

        // The plan coalescer builds these artifacts on every streamed update.
        const buildProposedPlanArtifacts = Effect.fnUntraced(function* (input: {
          readonly context: ActiveCodexTurnContext;
          readonly nativeItemId: string;
          readonly status: OrchestrationV2PlanArtifact["status"];
          readonly markdown: string;
          readonly completed?: boolean;
        }) {
          const updatedAt = yield* DateTime.now;
          const completedAt = input.completed === true ? updatedAt : null;
          const planId = yield* resolvePlanId(input.context, input.nativeItemId);
          const nodeId = idAllocator.derive.nodeFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: input.nativeItemId,
          });
          const turnItemId = idAllocator.derive.turnItemFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: input.nativeItemId,
          });
          const { ordinal, startedAt } = yield* resolveItemPosition(
            input.context,
            input.nativeItemId,
          );
          const plan: OrchestrationV2PlanArtifact = {
            id: planId,
            threadId: input.context.projectionThreadId,
            runId: input.context.projectionRunId,
            nodeId,
            kind: "proposed_plan",
            status: input.status,
            markdown: input.markdown,
          };
          const node: OrchestrationV2ExecutionNode = {
            id: nodeId,
            threadId: input.context.projectionThreadId,
            runId: input.context.projectionRunId,
            parentNodeId: input.context.itemParentNodeId,
            rootNodeId: input.context.rootNodeId,
            kind: "plan",
            status: input.completed === true ? "completed" : "running",
            countsForRun: false,
            providerThreadId: input.context.providerThread.id,
            providerTurnId: input.context.providerTurnId,
            nativeItemRef: codexNativeItemRef(input.nativeItemId),
            runtimeRequestId: null,
            checkpointScopeId: null,
            startedAt,
            completedAt,
          };
          const turnItem: OrchestrationV2TurnItem = {
            id: turnItemId,
            threadId: input.context.projectionThreadId,
            runId: input.context.projectionRunId,
            nodeId,
            providerThreadId: input.context.providerThread.id,
            providerTurnId: input.context.providerTurnId,
            nativeItemRef: codexNativeItemRef(input.nativeItemId),
            parentItemId: null,
            ordinal,
            status: input.completed === true ? "completed" : "running",
            title: null,
            startedAt,
            completedAt,
            updatedAt,
            type: "proposed_plan",
            planId,
            markdown: input.markdown,
            streaming: input.completed !== true,
          };
          return { node, plan, turnItem };
        });

        const planDeltas = yield* makeProviderTextDeltaCoalescer({
          flushIntervalMs: CODEX_PLAN_DELTA_FLUSH_INTERVAL_MS,
          emit:
            /* Provider events include streaming deltas; avoid a span for each update. */
            Effect.fnUntraced(function* (update) {
              const resolved = yield* resolveItemEventContext(update.turnId);
              if (resolved === undefined) return;
              const interrupted = (yield* Ref.get(terminalizedNonCompletedNativeTurns)).has(
                update.turnId,
              );
              const artifacts = yield* buildProposedPlanArtifacts({
                context: resolved.context,
                nativeItemId: update.itemId,
                // A finished proposal stays active until Implement consumes it.
                status: "active",
                markdown: update.text,
                completed: update.completed && !interrupted,
              });
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "plan.updated",
                driver: CODEX_PROVIDER,
                plan: artifacts.plan,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
            }, Effect.orDie),
        });

        const buildTodoListArtifacts = Effect.fn("CodexAdapterV2.buildTodoListArtifacts")(
          function* (input: {
            readonly context: ActiveCodexTurnContext;
            readonly nativeItemId: string;
            readonly status: OrchestrationV2PlanArtifact["status"];
            readonly steps: ReadonlyArray<OrchestrationV2PlanStep>;
            readonly explanation?: string;
            readonly completed?: boolean;
          }) {
            const updatedAt = yield* DateTime.now;
            const completedAt = input.completed === true ? updatedAt : null;
            const planId = yield* resolvePlanId(input.context, input.nativeItemId);
            const nodeId = idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: input.nativeItemId,
            });
            const turnItemId = idAllocator.derive.turnItemFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: input.nativeItemId,
            });
            const { ordinal, startedAt } = yield* resolveItemPosition(
              input.context,
              input.nativeItemId,
            );
            const plan: OrchestrationV2PlanArtifact = {
              id: planId,
              threadId: input.context.projectionThreadId,
              runId: input.context.projectionRunId,
              nodeId,
              kind: "todo_list",
              status: input.status,
              steps: [...input.steps],
              ...(input.explanation === undefined ? {} : { explanation: input.explanation }),
            };
            const node: OrchestrationV2ExecutionNode = {
              id: nodeId,
              threadId: input.context.projectionThreadId,
              runId: input.context.projectionRunId,
              parentNodeId: input.context.itemParentNodeId,
              rootNodeId: input.context.rootNodeId,
              kind: "todo_list",
              status: input.completed === true ? "completed" : "running",
              countsForRun: false,
              providerThreadId: input.context.providerThread.id,
              providerTurnId: input.context.providerTurnId,
              nativeItemRef: codexNativeItemRef(input.nativeItemId),
              runtimeRequestId: null,
              checkpointScopeId: null,
              startedAt,
              completedAt,
            };
            const turnItem: OrchestrationV2TurnItem = {
              id: turnItemId,
              threadId: input.context.projectionThreadId,
              runId: input.context.projectionRunId,
              nodeId,
              providerThreadId: input.context.providerThread.id,
              providerTurnId: input.context.providerTurnId,
              nativeItemRef: codexNativeItemRef(input.nativeItemId),
              parentItemId: null,
              ordinal,
              status: input.completed === true ? "completed" : "running",
              title: null,
              startedAt,
              completedAt,
              updatedAt,
              type: "todo_list",
              planId,
              steps: [...input.steps],
              ...(input.explanation === undefined ? {} : { explanation: input.explanation }),
            };
            return { node, plan, turnItem };
          },
        );

        const buildApprovalRequestArtifacts = Effect.fn(
          "CodexAdapterV2.buildApprovalRequestArtifacts",
        )(function* (input: {
          readonly context: ActiveCodexTurnContext;
          readonly nativeItemId: string;
          readonly nativeRequestId: string;
          readonly requestKind: ProviderRequestKind;
          readonly prompt?: string | null;
          readonly appName?: string;
          readonly options?: ReadonlyArray<ProviderApprovalOption>;
        }) {
          const createdAt = yield* DateTime.now;
          const { owner, subagent } = approvalOwnerCodexTurn(input.context);
          const parentNodeId =
            subagent?.subagentNodeId ??
            idAllocator.derive.nodeFromProviderItem({
              driver: CODEX_PROVIDER,
              nativeItemId: input.nativeItemId,
            });
          const ordinal = yield* resolveItemOrdinal(
            owner,
            `${input.nativeItemId}:approval:${input.nativeRequestId}`,
          );
          const requestId = yield* idAllocator.allocate.runtimeRequest({
            driver: CODEX_PROVIDER,
            providerTurnId: owner.providerTurnId,
            nativeRequestId: input.nativeRequestId,
          });
          const nodeId = idAllocator.derive.approvalNode({ requestId });
          const providerSessionId = owner.input.providerThread.providerSessionId;
          if (providerSessionId === null) {
            return yield* toProtocolError(
              `Provider thread ${owner.providerThread.id} is missing a provider session id.`,
            );
          }
          const node: OrchestrationV2ExecutionNode = {
            id: nodeId,
            threadId: owner.projectionThreadId,
            runId: owner.projectionRunId,
            parentNodeId,
            rootNodeId: owner.rootNodeId,
            kind: "approval_request",
            status: "waiting",
            countsForRun: false,
            providerThreadId: owner.providerThread.id,
            providerTurnId: owner.providerTurnId,
            nativeItemRef: codexNativeItemRef(input.nativeItemId),
            runtimeRequestId: requestId,
            checkpointScopeId: null,
            startedAt: createdAt,
            completedAt: null,
          };
          const request: OrchestrationV2RuntimeRequest = {
            id: requestId,
            nodeId,
            providerTurnId: owner.providerTurnId,
            nativeRequestRef: {
              driver: CODEX_PROVIDER,
              nativeId: input.nativeRequestId,
              strength: "strong",
            },
            kind: input.requestKind,
            status: "pending",
            responseCapability: {
              type: "live",
              providerSessionId,
            },
            createdAt,
            resolvedAt: null,
          };
          const turnItem: OrchestrationV2TurnItem = {
            id: idAllocator.derive.approvalTurnItem({ requestId }),
            threadId: owner.projectionThreadId,
            runId: owner.projectionRunId,
            nodeId,
            providerThreadId: owner.providerThread.id,
            providerTurnId: owner.providerTurnId,
            nativeItemRef: codexNativeItemRef(input.nativeItemId),
            parentItemId: null,
            ordinal,
            status: "waiting",
            title: null,
            startedAt: createdAt,
            completedAt: null,
            updatedAt: createdAt,
            type: "approval_request",
            requestId,
            requestKind: input.requestKind,
            ...(input.prompt === null || input.prompt === undefined
              ? {}
              : { prompt: input.prompt }),
            ...(input.appName === undefined ? {} : { appName: input.appName }),
            ...(input.options === undefined ? {} : { options: input.options }),
          };
          return { node, request, turnItem };
        });

        const buildUserInputRequestArtifacts = Effect.fn(
          "CodexAdapterV2.buildUserInputRequestArtifacts",
        )(function* (input: {
          readonly context: ActiveCodexTurnContext;
          readonly nativeItemId: string;
          readonly nativeRequestId: string;
          readonly questions: ReadonlyArray<CodexSchema.ToolRequestUserInputParams__ToolRequestUserInputQuestion>;
          readonly responseMode?: "message";
        }) {
          const createdAt = yield* DateTime.now;
          const requestId = yield* idAllocator.allocate.runtimeRequest({
            driver: CODEX_PROVIDER,
            providerTurnId: input.context.providerTurnId,
            nativeRequestId: input.nativeRequestId,
          });
          const providerSessionId = input.context.input.providerThread.providerSessionId;
          if (providerSessionId === null) {
            return yield* toProtocolError(
              `Provider thread ${input.context.providerThread.id} is missing a provider session id.`,
            );
          }
          const nativeQuestionIds = new Map(
            input.questions.map((question, index) => [
              nonEmptyText(question.id, `question-${index + 1}`),
              question.id,
            ]),
          );
          const questions = input.questions.map((question, index) => ({
            id: nonEmptyText(question.id, `question-${index + 1}`),
            header: nonEmptyText(question.header, "Question"),
            question: nonEmptyText(question.question, "Choose an answer."),
            options:
              question.options?.map((option, optionIndex) => ({
                label: nonEmptyText(option.label, `Option ${optionIndex + 1}`),
                description: nonEmptyText(option.description, option.label),
              })) ?? [],
          }));
          const nodeId = idAllocator.derive.nodeFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: input.nativeItemId,
          });
          const turnItemId = idAllocator.derive.turnItemFromProviderItem({
            driver: CODEX_PROVIDER,
            nativeItemId: input.nativeItemId,
          });
          const ordinal = yield* resolveItemOrdinal(input.context, input.nativeItemId);
          const node: OrchestrationV2ExecutionNode = {
            id: nodeId,
            threadId: input.context.projectionThreadId,
            runId: input.context.projectionRunId,
            parentNodeId: input.context.itemParentNodeId,
            rootNodeId: input.context.rootNodeId,
            kind: "user_input_request",
            status: "waiting",
            countsForRun: false,
            providerThreadId: input.context.providerThread.id,
            providerTurnId: input.context.providerTurnId,
            nativeItemRef: codexNativeItemRef(input.nativeItemId),
            runtimeRequestId: requestId,
            checkpointScopeId: null,
            startedAt: createdAt,
            completedAt: null,
          };
          const request: OrchestrationV2RuntimeRequest = {
            id: requestId,
            nodeId,
            providerTurnId: input.context.providerTurnId,
            nativeRequestRef: {
              driver: CODEX_PROVIDER,
              nativeId: input.nativeRequestId,
              strength: "strong",
            },
            kind: "user_input",
            status: "pending",
            responseCapability:
              input.responseMode === "message"
                ? { type: "message" }
                : { type: "live", providerSessionId },
            createdAt,
            resolvedAt: null,
          };
          const turnItem: OrchestrationV2TurnItem = {
            id: turnItemId,
            threadId: input.context.projectionThreadId,
            runId: input.context.projectionRunId,
            nodeId,
            providerThreadId: input.context.providerThread.id,
            providerTurnId: input.context.providerTurnId,
            nativeItemRef: codexNativeItemRef(input.nativeItemId),
            parentItemId: null,
            ordinal,
            status: "waiting",
            title: null,
            startedAt: createdAt,
            completedAt: null,
            updatedAt: createdAt,
            type: "user_input_request",
            requestId,
            questions,
            ...(input.responseMode === undefined ? {} : { responseMode: input.responseMode }),
          };
          return { node, request, turnItem, nativeQuestionIds };
        });

        yield* client.handleServerNotification("item/agentMessage/delta", (payload) =>
          Effect.gen(function* () {
            const context = (yield* Ref.get(activeTurns)).get(payload.turnId);
            if (context !== undefined) {
              yield* completeProviderRetry(context, yield* DateTime.now);
            }
            yield* agentMessageDeltas.append({
              turnId: payload.turnId,
              itemId: payload.itemId,
              delta: payload.delta,
            });
          }),
        );

        yield* client.handleServerNotification("item/commandExecution/outputDelta", (payload) =>
          Effect.gen(function* () {
            // Background commands emit deltas after settlement; item/started already registered their turn.
            const context =
              (yield* Ref.get(activeTurns)).get(payload.turnId) ??
              (yield* Ref.get(settledTurns)).get(payload.turnId);
            if (context === undefined) return;
            yield* emitProviderEvent({
              type: "command_output.delta",
              driver: CODEX_PROVIDER,
              threadId: context.projectionThreadId,
              itemId: idAllocator.derive.turnItemFromProviderItem({
                driver: CODEX_PROVIDER,
                nativeItemId: payload.itemId,
              }),
              chunk: payload.delta,
            });
          }),
        );

        yield* client.handleServerNotification("item/reasoning/summaryTextDelta", (payload) =>
          appendReasoning(payload, "summary", payload.summaryIndex),
        );
        yield* client.handleServerNotification("item/reasoning/textDelta", (payload) =>
          appendReasoning(payload, "content", payload.contentIndex),
        );

        yield* client.handleServerNotification("item/plan/delta", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return;
            }
            yield* completeProviderRetry(context, yield* DateTime.now);
            yield* planDeltas.append({
              turnId: payload.turnId,
              itemId: payload.itemId,
              delta: payload.delta,
            });
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerNotification("turn/plan/updated", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return;
            }
            yield* completeProviderRetry(context, yield* DateTime.now);
            const steps = payload.plan.map((step, index) => ({
              id: `step-${index + 1}`,
              text: nonEmptyText(step.step, `Step ${index + 1}`),
              status: codexPlanStepStatus(step.status),
            }));
            const explanation = trimText(payload.explanation);
            const artifacts = yield* buildTodoListArtifacts({
              context,
              nativeItemId: `turn-plan:${payload.turnId}`,
              status: "active",
              ...(explanation === undefined ? {} : { explanation }),
              steps,
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "plan.updated",
              driver: CODEX_PROVIDER,
              plan: artifacts.plan,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerNotification("account/rateLimits/updated", (payload) =>
          Effect.gen(function* () {
            const previous = yield* Ref.getAndUpdate(rateLimitSnapshot, (previous) =>
              mergeCodexRateLimits(previous, payload.rateLimits),
            );
            const snapshot = yield* Ref.get(rateLimitSnapshot);
            for (const item of (yield* Ref.get(limitedTurnItems)).values()) {
              const resetAt = failureResetAt(item.failure, snapshot);
              if (resetAt === null || item.failure.resetAt != null) continue;
              const updated = {
                ...item,
                updatedAt: yield* DateTime.now,
                failure: { ...item.failure, resetAt },
              };
              yield* Ref.update(limitedTurnItems, (current) =>
                new Map(current).set(item.providerThreadId!, updated),
              );
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: updated,
              });
            }
            const update = codexRateLimitsToUpdate(payload.rateLimits, previous);
            if (update && adapterOptions.onUsageLimits) {
              const checkedAt = DateTime.formatIso(yield* DateTime.now);
              yield* adapterOptions.onUsageLimits({ ...update, checkedAt });
            }
          }).pipe(Effect.orDie),
        );

        // A run's first request can 429 before Codex pushes a codex snapshot.
        const readUsageLimitReset = Effect.gen(function* () {
          const sent = yield* Ref.get(rateLimitSnapshot);
          const response = yield* client
            .request("account/rateLimits/read", null)
            .pipe(Effect.timeout("10 seconds"));
          const read = response.rateLimitsByLimitId?.codex ?? response.rateLimits;
          // A notification that landed during the read is newer than it.
          yield* Ref.update(rateLimitSnapshot, (current) =>
            current === sent ? mergeCodexRateLimits(current, read) : current,
          );
        }).pipe(
          // The terminal waits on this read, so no failure may skip it; interruption ends both.
          Effect.catchCauseIf(
            (cause) => !Cause.hasInterrupts(cause),
            (cause) =>
              Effect.logWarning("orchestration-v2.codex-usage-limit-reset-read-failed", { cause }),
          ),
        );

        yield* client.handleServerNotification("thread/tokenUsage/updated", (payload) =>
          Effect.gen(function* () {
            accumulateCodexTurnTokenUsage(
              usageStateForThread(payload.threadId),
              payload.turnId,
              payload.tokenUsage,
            );
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return;
            }
            const now = yield* DateTime.now;
            yield* emitProviderEvent({
              type: "provider_turn.updated",
              driver: CODEX_PROVIDER,
              threadId: context.projectionThreadId,
              providerTurn: {
                id: context.providerTurnId,
                providerThreadId: context.providerThread.id,
                nodeId: context.providerNodeId,
                runAttemptId: context.subagent === null ? context.input.attemptId : null,
                nativeTurnRef: {
                  driver: CODEX_PROVIDER,
                  nativeId: payload.turnId,
                  strength: "strong",
                },
                ordinal: context.providerTurnOrdinal,
                status: "running",
                startedAt: context.startedAt,
                completedAt: null,
                tokenUsage: codexProviderTurnTokenUsage(
                  payload.tokenUsage,
                  DateTime.formatIso(now),
                ),
              },
            });
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerNotification("thread/settings/updated", (payload) =>
          Effect.sync(() => reroutedThreads.delete(payload.threadId)).pipe(
            Effect.andThen(
              updateSubagentSelection(
                payload.threadId,
                payload.threadSettings.model,
                payload.threadSettings.effort,
                payload.threadSettings.serviceTier,
              ),
            ),
          ),
        );
        yield* client.handleServerNotification("model/rerouted", (payload) =>
          Effect.sync(() => {
            reroutedTurnModels.set(payload.turnId, payload.toModel);
            reroutedThreads.add(payload.threadId);
          }).pipe(Effect.andThen(updateSubagentSelection(payload.threadId, payload.toModel))),
        );

        yield* client.handleServerNotification("mcpServer/startupStatus/updated", (payload) =>
          Effect.gen(function* () {
            if (payload.threadId == null) return;
            const failures = mcpStartupFailures.get(payload.threadId) ?? new Map<string, string>();
            if (payload.status === "failed") {
              failures.set(
                payload.name,
                payload.error?.trim() || `MCP server ${payload.name} failed to start.`,
              );
              mcpStartupFailures.set(payload.threadId, failures);
            } else {
              failures.delete(payload.name);
              if (failures.size === 0) mcpStartupFailures.delete(payload.threadId);
              return;
            }
            const context = yield* findActiveTurnByNativeThreadId(payload.threadId);
            if (context !== undefined) yield* emitMcpStartupFailures(context);
          }),
        );

        yield* client.handleServerNotification("hook/completed", (payload) =>
          Effect.gen(function* () {
            if (payload.turnId == null || payload.run.status !== "stopped") return;
            const context = (yield* Ref.get(activeTurns)).get(payload.turnId);
            if (context?.providerThread.nativeThreadRef?.nativeId !== payload.threadId) return;
            const reason = payload.run.entries
              .filter((entry) => entry.kind === "stop")
              .map((entry) => entry.text.trim())
              .filter(Boolean)
              .join("\n");
            if (!reason) return;
            const source = payload.run.source === undefined ? "" : ` (${payload.run.source})`;
            context.hookStopReason = `${payload.run.eventName} hook${source}: ${reason}`;
          }),
        );

        yield* client.handleServerNotification("turn/started", (payload) =>
          Effect.gen(function* () {
            const context = (yield* Ref.get(activeTurns)).get(payload.turn.id);
            if (context !== undefined) {
              if (context.nativeStartReady !== undefined) {
                yield* Deferred.succeed(context.nativeStartReady, undefined);
              }
              return;
            }
            const pendingRootTurn = (yield* Ref.get(pendingRootTurns)).get(payload.threadId);
            if (pendingRootTurn !== undefined) {
              yield* registerRootTurn({
                turnInput: pendingRootTurn,
                nativeTurnId: payload.turn.id,
                startedAt: codexTimestamp(payload.turn.startedAt),
              });
              yield* Ref.update(pendingRootTurns, (current) => {
                const updated = new Map(current);
                updated.delete(payload.threadId);
                return updated;
              });
              return;
            }
            const goalHold = goalHolds.get(payload.threadId);
            if (goalHold !== undefined) {
              goalHolds.delete(payload.threadId);
              yield* emitProviderEvent({
                type: "provider_turn.updated",
                driver: CODEX_PROVIDER,
                threadId: goalHold.context.projectionThreadId,
                providerTurn: goalHold.completedTurn,
              });
              const next = yield* registerRootTurn({
                turnInput: {
                  ...goalHold.context.input,
                  providerTurnOrdinal: goalHold.context.providerTurnOrdinal + 1,
                },
                nativeTurnId: payload.turn.id,
                startedAt: codexTimestamp(payload.turn.startedAt),
              });
              const goalRun = continuedRuns.get(goalHold.context.providerTurnId) ?? [
                goalHold.context,
              ];
              goalRun.push(next);
              continuedRuns.set(goalHold.context.providerTurnId, goalRun);
              continuedRuns.set(next.providerTurnId, goalRun);
              yield* Deferred.succeed(goalHold.next, next);
              return;
            }
            // Pause orphan goal turns so they cannot work without an owning run.
            if (
              rootProviderThreads.has(payload.threadId) &&
              (goalsByNativeThread.get(payload.threadId)?.status === "active" ||
                goalStops.has(payload.threadId))
            ) {
              yield* Effect.logWarning("orchestration-v2.codex-goal-turn-without-run", {
                nativeThreadId: payload.threadId,
                nativeTurnId: payload.turn.id,
              });
              yield* client
                .request("thread/goal/set", { threadId: payload.threadId, status: "paused" })
                .pipe(
                  Effect.catch((cause) =>
                    Effect.logWarning("orchestration-v2.codex-goal-pause-failed", { cause }),
                  ),
                  Effect.andThen(
                    client.request("turn/interrupt", {
                      threadId: payload.threadId,
                      turnId: payload.turn.id,
                    }),
                  ),
                  Effect.catch((cause) =>
                    Effect.logWarning("orchestration-v2.codex-goal-turn-stop-failed", { cause }),
                  ),
                  Effect.forkIn(scope),
                );
              return;
            }
            yield* rememberSubagentTurnStarted({
              nativeThreadId: payload.threadId,
              nativeTurnId: payload.turn.id,
              startedAt: codexTimestamp(payload.turn.startedAt),
            });
          }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)),
        );

        yield* client.handleServerNotification("thread/goal/updated", (payload) =>
          Effect.gen(function* () {
            goalsByNativeThread.set(payload.threadId, providerGoalFromCodex(payload.goal));
            yield* emitGoalUpdate(payload.threadId);
            if (payload.goal.status !== "active") yield* releaseGoalHold(payload.threadId);
          }).pipe(turnTerminalizationPermit.withPermits(1)),
        );
        yield* client.handleServerNotification("thread/goal/cleared", (payload) =>
          Effect.gen(function* () {
            goalsByNativeThread.set(payload.threadId, null);
            yield* emitGoalUpdate(payload.threadId);
            yield* releaseGoalHold(payload.threadId);
          }).pipe(turnTerminalizationPermit.withPermits(1)),
        );

        yield* client.handleServerNotification("error", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return;
            }
            const notificationCode = codexErrorInfoCode(payload.error.codexErrorInfo);
            if (!payload.willRetry) {
              context.latestProviderFailure = {
                nativeMessage: payload.error.message,
                nativeCode: notificationCode,
                failure: withCodexManagedFailure(
                  makeProviderFailure({
                    message: payload.error.additionalDetails?.trim() || payload.error.message,
                    code: notificationCode,
                    class: codexUsageLimitError(notificationCode, payload.error.codexErrorInfo)
                      ? "usage_limit"
                      : "provider_error",
                  }),
                  payload.error,
                ),
              };
              return;
            }
            const updatedAt = yield* DateTime.now;
            const previous = (yield* Ref.get(providerRetries)).get(context.providerTurnId);
            const progress = parseCodexRetryProgress(payload.error.message);
            const retry: OrchestrationV2ProviderRetry = {
              attempt: progress?.attempt ?? (previous?.retry.attempt ?? 0) + 1,
              // Codex drops the count once it starts waiting for the network, which has no limit.
              maxAttempts: progress?.maxAttempts ?? null,
              retryDelayMs: null,
            };
            const code = codexErrorInfoCode(payload.error.codexErrorInfo);
            const additionalDetails = payload.error.additionalDetails?.trim();
            const failure = withCodexManagedFailure(
              makeProviderFailure({
                message:
                  additionalDetails === undefined || additionalDetails.length === 0
                    ? payload.error.message
                    : additionalDetails,
                code,
                class: codexUsageLimitError(code, payload.error.codexErrorInfo)
                  ? "usage_limit"
                  : code?.startsWith("http") === true || code?.startsWith("responseStream") === true
                    ? "transport_error"
                    : "provider_error",
                retryable: true,
              }),
              payload.error,
            );
            const itemOrdinal =
              previous?.itemOrdinal ??
              (yield* resolveItemOrdinal(context, `terminal-failure:${context.providerTurnId}`));
            const state: ActiveCodexProviderRetry = {
              ...(previous?.context === undefined ? {} : { context: previous.context }),
              nativeMessage: payload.error.message,
              nativeCode: code,
              retry,
              failure,
              startedAt: previous?.startedAt ?? updatedAt,
              itemOrdinal,
            };
            yield* Ref.update(providerRetries, (current) => {
              const updated = new Map(current);
              updated.set(context.providerTurnId, state);
              return updated;
            });
            yield* failStalledProviderRetry(context, state, payload.error.codexErrorInfo).pipe(
              Effect.delay(CODEX_PROVIDER_RETRY_STALL_TIMEOUT),
              Effect.forkIn(scope, { startImmediately: true }),
            );
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: codexRetryTurnItem(context, state, "running", updatedAt),
            });
          }).pipe(Effect.orDie),
        );

        const emitCompactionItem = Effect.fn("CodexAdapterV2.emitCompactionItem")(function* (
          context: ActiveCodexTurnContext,
          nativeItemId: string,
          status: "running" | "completed" | "failed" | "interrupted" | "cancelled",
          updatedAt: DateTime.Utc,
        ) {
          const { ordinal, startedAt } = yield* resolveItemPosition(
            context,
            nativeItemId,
            updatedAt,
          );
          if (status === "running") context.runningCompactionItemIds.add(nativeItemId);
          else context.runningCompactionItemIds.delete(nativeItemId);
          yield* emitProviderEvent({
            type: "turn_item.updated",
            driver: CODEX_PROVIDER,
            turnItem: {
              id: idAllocator.derive.turnItemFromProviderItem({
                driver: CODEX_PROVIDER,
                nativeItemId,
              }),
              threadId: context.projectionThreadId,
              runId: context.projectionRunId,
              nodeId: context.providerNodeId,
              providerThreadId: context.providerThread.id,
              providerTurnId: context.providerTurnId,
              nativeItemRef: codexNativeItemRef(nativeItemId),
              parentItemId: null,
              ordinal,
              type: "compaction",
              driver: CODEX_PROVIDER,
              status,
              title:
                status === "running"
                  ? "Compacting context"
                  : status === "completed"
                    ? "Context compacted"
                    : status === "failed"
                      ? "Context compaction failed"
                      : "Context compaction stopped",
              startedAt,
              completedAt: status === "running" ? null : updatedAt,
              updatedAt,
            },
          });
        });

        // Unloading an idle thread kills its terminals without completing
        // their command items, while the shared provider session stays alive.
        yield* client.handleServerNotification("thread/closed", (payload) =>
          Effect.gen(function* () {
            const closing = threadClosings.get(payload.threadId);
            threadClosings.delete(payload.threadId);
            if (closing !== undefined) yield* Deferred.succeed(closing, undefined);
            const completedAt = yield* DateTime.now;
            for (const context of (yield* Ref.get(settledTurns)).values()) {
              if (context.providerThread.nativeThreadRef?.nativeId !== payload.threadId) continue;
              const commands = (yield* Ref.get(runningCommandItemsByTurn)).get(
                context.nativeTurnId,
              );
              if (commands !== undefined) {
                context.closedCommandItemIds = new Set(commands.keys());
              }
              yield* terminalizeRunningCommandItems(
                context,
                context.nativeTurnId,
                "interrupted",
                completedAt,
              );
              yield* Ref.update(runningCommandItemsByTurn, (current) => {
                const updated = new Map(current);
                updated.delete(context.nativeTurnId);
                return updated;
              });
              yield* releaseSettledTurnIfIdle(context.nativeTurnId);
            }
          }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)),
        );

        yield* client.handleServerNotification("item/started", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return;
            }

            if (payload.item.type === "contextCompaction") {
              yield* emitCompactionItem(
                context,
                payload.item.id,
                "running",
                DateTime.makeUnsafe(payload.startedAtMs),
              );
              return;
            }

            if (payload.item.type === "userMessage") {
              yield* emitSubagentUserMessage(
                context,
                payload.item,
                DateTime.makeUnsafe(payload.startedAtMs),
              );
              return;
            }

            if (payload.item.type === "subAgentActivity") {
              markSubagentUsage(context);
              yield* registerSubagentActivity({
                context,
                item: payload.item,
              });
              return;
            }

            yield* completeProviderRetry(context, yield* DateTime.now);
            if (payload.item.type !== "reasoning") {
              yield* resolveItemPosition(
                context,
                payload.item.id,
                DateTime.makeUnsafe(payload.startedAtMs),
              );
            }

            if (payload.item.type === "fileChange") {
              context.fileChangeKindsByItem.set(
                payload.item.id,
                Object.fromEntries(payload.item.changes.map(({ path, kind }) => [path, kind])),
              );
              return;
            }

            if (payload.item.type === "agentMessage") {
              if (payload.item.phase != null)
                context.messagePhases.set(payload.item.id, payload.item.phase);
              if (payload.item.phase == null || payload.item.phase === "final_answer") {
                yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
                  const updated = new Map(current);
                  const itemIds = new Set(updated.get(payload.turnId) ?? []);
                  itemIds.add(payload.item.id);
                  updated.set(payload.turnId, itemIds);
                  return updated;
                });
              }
              return;
            }

            if (payload.item.type === "imageGeneration") {
              if (!(yield* Ref.get(interruptingNativeTurns)).has(payload.turnId)) {
                imageTurn(payload.turnId).running.set(payload.item.id, {
                  revisedPrompt: payload.item.revisedPrompt ?? null,
                  transparentBackground: payload.item.transparentBackground ?? null,
                });
                yield* emitImageGenerationItem(context, payload.item, false);
                yield* emitImageWorkState(context, true);
              }
              return;
            }

            if (payload.item.type === "commandExecution") {
              if (!codexItemStatus(payload.item.status).completed) {
                yield* trackRunningCommandItem(payload.turnId, {
                  id: payload.item.id,
                  command: payload.item.command,
                  ...(payload.item.aggregatedOutput === null ||
                  payload.item.aggregatedOutput === undefined
                    ? {}
                    : { aggregatedOutput: payload.item.aggregatedOutput }),
                  ...(typeof payload.item.processId === "string"
                    ? { processId: payload.item.processId }
                    : {}),
                });
              }
              const artifacts = yield* buildCommandExecutionArtifacts(context, payload.item);
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              yield* reportSubagentProgress(context, payload.item.command);
              return;
            }

            if (payload.item.type === "mcpToolCall" || payload.item.type === "dynamicToolCall") {
              if (!codexItemStatus(payload.item.status).completed) {
                yield* trackRunningDynamicTool(payload.turnId, payload.item);
              }
              const artifacts = yield* buildDynamicToolArtifacts(context, payload.item);
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              if (artifacts.turnItem.type === "dynamic_tool") {
                yield* reportSubagentProgress(
                  context,
                  artifacts.turnItem.title ?? artifacts.turnItem.toolName,
                );
              }
              return;
            }

            if (payload.item.type !== "webSearch") {
              return;
            }

            const artifacts = yield* buildWebSearchArtifacts({
              context,
              item: payload.item,
              completed: false,
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
          }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)),
        );

        yield* client.handleServerNotification("item/completed", (payload) =>
          Effect.gen(function* () {
            if (payload.item.type === "contextCompaction")
              yield* Ref.update(threadsAwaitingContextRestore, (current) =>
                new Set(current).add(payload.threadId),
              );
            if (
              (payload.item.type === "subAgentActivity" ||
                payload.item.type === "collabAgentToolCall") &&
              (yield* Ref.get(terminalizedNonCompletedNativeTurns)).has(payload.turnId)
            )
              return;
            // Child lifecycle notifications outlive the parent's turn context.
            if (payload.item.type === "subAgentActivity" && payload.item.kind !== "started") {
              yield* registerSubagentActivity({ item: payload.item });
              return;
            }
            if (payload.item.type === "collabAgentToolCall" && payload.item.tool !== "spawnAgent") {
              yield* updateSubagentStates({ item: payload.item });
              return;
            }
            const resolved = yield* resolveItemEventContext(payload.turnId);
            if (resolved === undefined) {
              return;
            }
            const { context, settled } = resolved;

            if (payload.item.type === "reasoning") {
              yield* completeReasoning(payload.turnId, payload.item);
              return;
            }

            if (payload.item.type === "contextCompaction") {
              yield* emitCompactionItem(
                context,
                payload.item.id,
                "completed",
                DateTime.makeUnsafe(payload.completedAtMs),
              );
              return;
            }

            if (payload.item.type === "userMessage") {
              if (yield* emitSubagentUserMessage(context, payload.item)) {
                return;
              }
            }

            if (payload.item.type === "commandExecution") {
              // Other tools can retain the context after thread closure.
              if (context.closedCommandItemIds?.has(payload.item.id)) return;
              const turnDrained = yield* clearRunningCommandItem(payload.turnId, payload.item.id);
              const artifacts = yield* buildCommandExecutionArtifacts(context, payload.item);
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              if (settled) {
                if (
                  context.subagent === null &&
                  continuationRequests !== undefined &&
                  !(yield* Ref.get(interruptingNativeTurns)).has(payload.turnId)
                ) {
                  const alreadyOffered = yield* Ref.modify(
                    offeredContinuationItemsByTurn,
                    (current) => {
                      const items = current.get(payload.turnId);
                      if (items !== undefined && items.has(payload.item.id)) {
                        return [true, current] as const;
                      }
                      const updated = new Map(current);
                      const updatedItems = new Set(items ?? []);
                      updatedItems.add(payload.item.id);
                      updated.set(payload.turnId, updatedItems);
                      return [false, updated] as const;
                    },
                  );
                  if (!alreadyOffered) {
                    yield* continuationRequests.offer({
                      threadId: context.projectionThreadId,
                      providerThreadId: context.providerThread.id,
                      driver: CODEX_PROVIDER,
                      // Codex starts a new turn from this text, so nothing is buffered.
                      delivery: "message_text",
                      detail: codexBackgroundCommandDetail(payload.item),
                      notification: {
                        ...backgroundWorkNotification([
                          {
                            kind: "command",
                            label: payload.item.command,
                            outcome:
                              payload.item.exitCode === 0
                                ? "completed"
                                : payload.item.exitCode == null
                                  ? "unknown"
                                  : "failed",
                            exitCode: payload.item.exitCode ?? undefined,
                          },
                        ]),
                        detail: payload.item.command,
                      },
                    });
                  }
                }
                if (turnDrained) {
                  yield* releaseSettledTurnIfIdle(payload.turnId);
                }
              }
              return;
            }

            if (payload.item.type === "mcpToolCall" || payload.item.type === "dynamicToolCall") {
              yield* clearRunningDynamicTool(payload.turnId, payload.item.id);
              const appResourceUri = codexMcpAppResourceUri(payload.item);
              if (appResourceUri !== undefined && payload.item.type === "mcpToolCall") {
                yield* captureMcpApp(context, payload.turnId, payload.item, appResourceUri);
                return;
              }
              const artifacts = yield* buildDynamicToolArtifacts(context, payload.item);
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              if (settled) {
                yield* releaseSettledTurnIfIdle(payload.turnId);
              }
              return;
            }

            if (payload.item.type === "fileChange") {
              context.fileChangeKindsByItem.delete(payload.item.id);
              const artifacts = yield* buildFileChangeArtifacts(context, payload.item);
              if (artifacts === null) {
                return;
              }
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              return;
            }

            if (payload.item.type === "webSearch") {
              const artifacts = yield* buildWebSearchArtifacts({
                context,
                item: payload.item,
                completed: true,
              });
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              return;
            }

            if (payload.item.type === "plan") {
              yield* planDeltas.complete({
                turnId: payload.turnId,
                itemId: payload.item.id,
                ...(payload.item.text.length > 0 ? { finalText: payload.item.text } : {}),
              });
              return;
            }

            if (payload.item.type === "collabAgentToolCall") {
              if (payload.item.tool === "spawnAgent") markSubagentUsage(context);
              yield* registerSubagentThreads({
                context,
                item: payload.item,
              });
              yield* updateSubagentStates({
                item: payload.item,
              });
              return;
            }

            if (payload.item.type === "subAgentActivity") {
              markSubagentUsage(context);
              yield* registerSubagentActivity({
                context,
                item: payload.item,
              });
              return;
            }

            if (payload.item.type === "imageGeneration") {
              if ((yield* Ref.get(interruptingNativeTurns)).has(payload.turnId)) return;
              yield* emitImageGenerationItem(context, payload.item, true);
              const { id, savedPath, status, failure } = payload.item;
              const state = imageTurn(payload.turnId);
              if (savedPath?.trim() && status === "completed" && failure == null) {
                state.saved.set(id, savedPath);
              }
              // Idle release runs concurrently, so keep this work pending through projection.
              if (settled) {
                yield* projectGeneratedImages(context, payload.turnId);
              }
              state.running.delete(id);
              yield* emitImageWorkState(context, state.running.size > 0 || state.saved.size > 0);
              if (settled) {
                yield* releaseSettledTurnIfIdle(payload.turnId);
              }
              return;
            }

            if (payload.item.type === "imageView") {
              const artifacts = yield* buildDynamicToolArtifacts(context, {
                type: "dynamicToolCall",
                id: payload.item.id,
                tool: "view_image",
                arguments: { path: payload.item.path },
                status: "completed",
              });
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: { ...artifacts.turnItem, viewedImagePath: payload.item.path },
              });
              return;
            }

            if (payload.item.type !== "agentMessage") {
              return;
            }

            if (payload.item.delivery === "async" && payload.item.questions?.length) {
              const artifacts = yield* buildUserInputRequestArtifacts({
                context,
                nativeItemId: payload.item.id,
                nativeRequestId: `async:${payload.item.id}`,
                responseMode: "message",
                questions: payload.item.questions.map((question, index) => ({
                  id: String(index),
                  header: "Question",
                  question: question.title,
                  options: (question.options ?? []).map((label) => ({ label, description: "" })),
                })),
              });
              yield* emitProviderEvent({
                type: "node.updated",
                driver: CODEX_PROVIDER,
                node: artifacts.node,
              });
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: artifacts.turnItem,
              });
              yield* emitProviderEvent({
                type: "runtime_request.updated",
                driver: CODEX_PROVIDER,
                threadId: artifacts.node.threadId,
                runtimeRequest: artifacts.request,
              });
              yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
                const ids = current.get(payload.turnId);
                if (!ids?.has(payload.item.id)) return current;
                const next = new Map(current);
                const remaining = new Set(ids);
                remaining.delete(payload.item.id);
                if (remaining.size === 0) next.delete(payload.turnId);
                else next.set(payload.turnId, remaining);
                return next;
              });
              return;
            }

            if (payload.item.phase != null)
              context.messagePhases.set(payload.item.id, payload.item.phase);
            const finalAnswer = payload.item.phase == null || payload.item.phase === "final_answer";
            if (finalAnswer) {
              yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
                const updated = new Map(current);
                const itemIds = new Set(updated.get(payload.turnId) ?? []);
                itemIds.add(payload.item.id);
                updated.set(payload.turnId, itemIds);
                return updated;
              });
            }
            const completedTextsBefore =
              (yield* Ref.get(completedFinalAnswerTextsByTurn)).get(payload.turnId) ??
              new Set<string>();
            const text = yield* agentMessageDeltas.complete({
              turnId: payload.turnId,
              itemId: payload.item.id,
              finalText: payload.item.text,
            });
            yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
              const itemIds = current.get(payload.turnId);
              if (itemIds === undefined || !itemIds.has(payload.item.id)) {
                return current;
              }
              const updated = new Map(current);
              const remainingItemIds = new Set(itemIds);
              remainingItemIds.delete(payload.item.id);
              if (remainingItemIds.size === 0) {
                updated.delete(payload.turnId);
              } else {
                updated.set(payload.turnId, remainingItemIds);
              }
              return updated;
            });
            const emitted =
              !finalAnswer ||
              completedTextsBefore.size === 0 ||
              (text.length > 0 && !completedTextsBefore.has(text));
            if (emitted && context.subagent !== null && finalAnswer) {
              yield* emitSubagentTaskUpdate({
                subagent: context.subagent,
                status: context.subagent.task.status,
                result: text,
              });
            }
          }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)),
        );

        yield* client.handleServerRequest("item/commandExecution/requestApproval", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return yield* toProtocolError(
                `No active Codex turn context for approval turn ${payload.turnId}.`,
                payload,
              );
            }
            const allowance = sessionAllowance(payload.threadId, context);
            if (allowance.has("command")) {
              return {
                decision: "acceptForSession",
              } satisfies CodexSchema.CommandExecutionRequestApprovalResponse;
            }

            const nativeRequestId = payload.approvalId ?? payload.itemId;
            const artifacts = yield* buildApprovalRequestArtifacts({
              context,
              nativeItemId: payload.itemId,
              nativeRequestId,
              requestKind: "command",
              ...((payload.reason ?? payload.command) === undefined
                ? {}
                : { prompt: payload.reason ?? payload.command }),
            });
            const decision = yield* Deferred.make<ProviderApprovalDecision, never>();
            yield* Ref.update(pendingRuntimeRequests, (current) => {
              const updated = new Map(current);
              updated.set(String(artifacts.request.id), {
                type: "approval",
                requestKind: "command",
                decision,
              });
              return updated;
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            yield* emitProviderEvent({
              type: "runtime_request.updated",
              driver: CODEX_PROVIDER,
              threadId: artifacts.node.threadId,
              runtimeRequest: artifacts.request,
            });

            const resolved = yield* Deferred.await(decision).pipe(
              Effect.ensuring(
                Ref.update(pendingRuntimeRequests, (current) => {
                  const updated = new Map(current);
                  updated.delete(String(artifacts.request.id));
                  return updated;
                }),
              ),
            );
            if (resolved === "acceptForSession" || resolved === "acceptAlways") {
              allowance.add("command");
            }
            return {
              decision: resolved === "acceptAlways" ? "acceptForSession" : resolved,
            } satisfies CodexSchema.CommandExecutionRequestApprovalResponse;
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerRequest("item/fileChange/requestApproval", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return yield* toProtocolError(
                `No active Codex turn context for file change approval turn ${payload.turnId}.`,
                payload,
              );
            }
            const allowance = sessionAllowance(payload.threadId, context);
            if (allowance.has("file-change")) {
              return {
                decision: "acceptForSession",
              } satisfies CodexSchema.FileChangeRequestApprovalResponse;
            }

            const artifacts = yield* buildApprovalRequestArtifacts({
              context,
              nativeItemId: payload.itemId,
              nativeRequestId: payload.itemId,
              requestKind: "file-change",
              prompt:
                codexFileChangeApprovalPrompt({
                  ...payload,
                  fileChanges: context.fileChangeKindsByItem.get(payload.itemId) ?? {},
                }) ?? null,
            });
            const decision = yield* Deferred.make<ProviderApprovalDecision, never>();
            yield* Ref.update(pendingRuntimeRequests, (current) => {
              const updated = new Map(current);
              updated.set(String(artifacts.request.id), {
                type: "approval",
                requestKind: "file-change",
                decision,
              });
              return updated;
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            yield* emitProviderEvent({
              type: "runtime_request.updated",
              driver: CODEX_PROVIDER,
              threadId: artifacts.node.threadId,
              runtimeRequest: artifacts.request,
            });

            const resolved = yield* Deferred.await(decision).pipe(
              Effect.ensuring(
                Ref.update(pendingRuntimeRequests, (current) => {
                  const updated = new Map(current);
                  updated.delete(String(artifacts.request.id));
                  return updated;
                }),
              ),
            );
            if (resolved === "acceptForSession" || resolved === "acceptAlways") {
              allowance.add("file-change");
            }
            return {
              decision: resolved === "acceptAlways" ? "acceptForSession" : resolved,
            } satisfies CodexSchema.FileChangeRequestApprovalResponse;
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerRequest("item/permissions/requestApproval", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return yield* toProtocolError(
                `No active Codex turn context for permissions approval turn ${payload.turnId}.`,
                payload,
              );
            }

            const requestKind = providerRequestKindFromPermissions(payload.permissions);
            const artifacts = yield* buildApprovalRequestArtifacts({
              context,
              nativeItemId: payload.itemId,
              nativeRequestId: payload.itemId,
              requestKind,
              ...(payload.reason === undefined ? {} : { prompt: payload.reason }),
            });
            const decision = yield* Deferred.make<ProviderApprovalDecision, never>();
            yield* Ref.update(pendingRuntimeRequests, (current) => {
              const updated = new Map(current);
              updated.set(String(artifacts.request.id), {
                type: "approval",
                requestKind,
                decision,
              });
              return updated;
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            yield* emitProviderEvent({
              type: "runtime_request.updated",
              driver: CODEX_PROVIDER,
              threadId: artifacts.node.threadId,
              runtimeRequest: artifacts.request,
            });

            const resolved = yield* Deferred.await(decision).pipe(
              Effect.ensuring(
                Ref.update(pendingRuntimeRequests, (current) => {
                  const updated = new Map(current);
                  updated.delete(String(artifacts.request.id));
                  return updated;
                }),
              ),
            );
            return permissionsResponseFromDecision({
              decision: resolved,
              permissions: payload.permissions,
            });
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerRequest("mcpServer/elicitation/request", (payload) =>
          Effect.gen(function* () {
            if (toMcpElicitationResponse(payload, "accept").action !== "accept") {
              yield* Effect.logWarning("Declined an unsupported MCP elicitation.", {
                serverName: payload.serverName,
                mode: payload.mode,
              });
              return {
                action: "decline",
              } satisfies CodexSchema.McpServerElicitationRequestResponse;
            }
            const context =
              payload.turnId === undefined || payload.turnId === null
                ? undefined
                : yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              yield* Effect.logWarning(
                "Declined an MCP elicitation without an active Codex turn context.",
                { serverName: payload.serverName },
              );
              return {
                action: "decline",
              } satisfies CodexSchema.McpServerElicitationRequestResponse;
            }

            const nativeRequestId =
              payload.mode === "url"
                ? payload.elicitationId
                : `mcp-elicitation:${payload.serverName}`;
            const described = describeMcpElicitation(payload);
            const artifacts = yield* buildApprovalRequestArtifacts({
              context,
              nativeItemId: nativeRequestId,
              nativeRequestId,
              requestKind: "mcp-elicitation",
              prompt: payload.message,
              appName: described.appName,
              options: described.options,
            });
            const decision = yield* Deferred.make<ProviderApprovalDecision, never>();
            yield* Ref.update(pendingRuntimeRequests, (current) => {
              const updated = new Map(current);
              updated.set(String(artifacts.request.id), {
                type: "approval",
                requestKind: "mcp-elicitation",
                decision,
              });
              return updated;
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            yield* emitProviderEvent({
              type: "runtime_request.updated",
              driver: CODEX_PROVIDER,
              threadId: artifacts.node.threadId,
              runtimeRequest: artifacts.request,
            });

            const resolved = yield* Deferred.await(decision).pipe(
              Effect.ensuring(
                Ref.update(pendingRuntimeRequests, (current) => {
                  const updated = new Map(current);
                  updated.delete(String(artifacts.request.id));
                  return updated;
                }),
              ),
            );
            return toMcpElicitationResponse(payload, resolved);
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerRequest("execCommandApproval", (payload) =>
          Effect.gen(function* () {
            const context = yield* findActiveTurnByNativeThreadId(payload.conversationId);
            if (context === undefined) {
              return yield* toProtocolError(
                `No active Codex turn context for exec approval thread ${payload.conversationId}.`,
                payload,
              );
            }
            const allowance = sessionAllowance(payload.conversationId, context);
            if (allowance.has("command")) {
              return {
                decision: "approved_for_session",
              } satisfies CodexSchema.ExecCommandApprovalResponse;
            }

            const nativeRequestId = payload.approvalId ?? payload.callId;
            const artifacts = yield* buildApprovalRequestArtifacts({
              context,
              nativeItemId: payload.callId,
              nativeRequestId,
              requestKind: "command",
              prompt: payload.reason ?? payload.command.join(" "),
            });
            const decision = yield* Deferred.make<ProviderApprovalDecision, never>();
            yield* Ref.update(pendingRuntimeRequests, (current) => {
              const updated = new Map(current);
              updated.set(String(artifacts.request.id), {
                type: "approval",
                requestKind: "command",
                decision,
              });
              return updated;
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            yield* emitProviderEvent({
              type: "runtime_request.updated",
              driver: CODEX_PROVIDER,
              threadId: artifacts.node.threadId,
              runtimeRequest: artifacts.request,
            });

            const resolved = yield* Deferred.await(decision).pipe(
              Effect.ensuring(
                Ref.update(pendingRuntimeRequests, (current) => {
                  const updated = new Map(current);
                  updated.delete(String(artifacts.request.id));
                  return updated;
                }),
              ),
            );
            if (resolved === "acceptForSession" || resolved === "acceptAlways") {
              allowance.add("command");
            }
            return {
              decision: approvalDecisionToLegacyReviewDecision(resolved),
            } satisfies CodexSchema.ExecCommandApprovalResponse;
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerRequest("applyPatchApproval", (payload) =>
          Effect.gen(function* () {
            const context = yield* findActiveTurnByNativeThreadId(payload.conversationId);
            if (context === undefined) {
              return yield* toProtocolError(
                `No active Codex turn context for apply patch approval thread ${payload.conversationId}.`,
                payload,
              );
            }
            const allowance = sessionAllowance(payload.conversationId, context);
            if (allowance.has("file-change")) {
              return {
                decision: "approved_for_session",
              } satisfies CodexSchema.ApplyPatchApprovalResponse;
            }

            const artifacts = yield* buildApprovalRequestArtifacts({
              context,
              nativeItemId: payload.callId,
              nativeRequestId: payload.callId,
              requestKind: "file-change",
              prompt: codexFileChangeApprovalPrompt(payload) ?? null,
            });
            const decision = yield* Deferred.make<ProviderApprovalDecision, never>();
            yield* Ref.update(pendingRuntimeRequests, (current) => {
              const updated = new Map(current);
              updated.set(String(artifacts.request.id), {
                type: "approval",
                requestKind: "file-change",
                decision,
              });
              return updated;
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            yield* emitProviderEvent({
              type: "runtime_request.updated",
              driver: CODEX_PROVIDER,
              threadId: artifacts.node.threadId,
              runtimeRequest: artifacts.request,
            });

            const resolved = yield* Deferred.await(decision).pipe(
              Effect.ensuring(
                Ref.update(pendingRuntimeRequests, (current) => {
                  const updated = new Map(current);
                  updated.delete(String(artifacts.request.id));
                  return updated;
                }),
              ),
            );
            if (resolved === "acceptForSession" || resolved === "acceptAlways") {
              allowance.add("file-change");
            }
            return {
              decision: approvalDecisionToLegacyReviewDecision(resolved),
            } satisfies CodexSchema.ApplyPatchApprovalResponse;
          }).pipe(Effect.orDie),
        );

        yield* client.handleServerRequest("item/tool/requestUserInput", (payload) =>
          Effect.gen(function* () {
            const context = yield* awaitActiveTurn(payload.turnId);
            if (context === undefined) {
              return yield* toProtocolError(
                `No active Codex turn context for user input request turn ${payload.turnId}.`,
                payload,
              );
            }

            const artifacts = yield* buildUserInputRequestArtifacts({
              context,
              nativeItemId: payload.itemId,
              nativeRequestId: payload.itemId,
              questions: payload.questions,
            });
            const answers = yield* Deferred.make<ProviderUserInputAnswers, never>();
            yield* Ref.update(pendingRuntimeRequests, (current) => {
              const updated = new Map(current);
              updated.set(String(artifacts.request.id), {
                type: "user_input",
                answers,
              });
              return updated;
            });
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            yield* emitProviderEvent({
              type: "runtime_request.updated",
              driver: CODEX_PROVIDER,
              threadId: artifacts.node.threadId,
              runtimeRequest: artifacts.request,
            });

            const resolved = yield* Deferred.await(answers).pipe(
              Effect.ensuring(
                Ref.update(pendingRuntimeRequests, (current) => {
                  const updated = new Map(current);
                  updated.delete(String(artifacts.request.id));
                  return updated;
                }),
              ),
            );
            return {
              answers: toCodexUserInputAnswers(resolved, artifacts.nativeQuestionIds),
            } satisfies CodexSchema.ToolRequestUserInputResponse;
          }).pipe(Effect.orDie),
        );

        const makeRootTerminalEvent = Effect.fn("CodexAdapterV2.makeRootTerminalEvent")(
          function* (input: {
            readonly context: ActiveCodexTurnContext;
            readonly status: OrchestrationV2ProviderTurn["status"];
            readonly failureMessage?: string;
            readonly failureErrorInfo?: CodexErrorInfo;
            readonly providerRetry?: ActiveCodexProviderRetry;
          }): Effect.fn.Return<CodexRootTerminalEvent> {
            const terminalStatus = providerTurnStatusToTerminal(input.status);
            if (terminalStatus === "failed") {
              const previousFailure = input.context.latestProviderFailure ?? input.providerRetry;
              const failureCode =
                input.failureErrorInfo === undefined
                  ? undefined
                  : codexErrorInfoCode(input.failureErrorInfo);
              const failureClass = codexUsageLimitError(failureCode ?? null, input.failureErrorInfo)
                ? "usage_limit"
                : "provider_error";
              // The notification carries the richer message, so it wins when the
              // completion repeats it. An explicit exhaustion status on the
              // completion still decides the class; a missing one is not evidence.
              const exhaustionStatus = codexExhaustionHttpStatus(input.failureErrorInfo);
              const failure =
                previousFailure !== undefined &&
                (input.failureMessage === undefined ||
                  input.failureMessage === previousFailure.nativeMessage) &&
                (failureCode === undefined || failureCode === previousFailure.nativeCode) &&
                (exhaustionStatus === null || previousFailure.failure.class === failureClass)
                  ? previousFailure.failure
                  : withCodexManagedFailure(
                      makeProviderFailure({
                        message: input.failureMessage,
                        code: failureCode,
                        class: failureClass,
                      }),
                      input.failureMessage,
                    );
              return {
                type: "turn.terminal",
                driver: CODEX_PROVIDER,
                providerThreadId: input.context.providerThread.id,
                providerTurnId: input.context.providerTurnId,
                runOrdinal: input.context.input.runOrdinal,
                failureItemOrdinal: yield* resolveItemOrdinal(
                  input.context,
                  `terminal-failure:${input.context.providerTurnId}`,
                ),
                status: terminalStatus,
                failure:
                  failure.class === "usage_limit"
                    ? {
                        ...failure,
                        resetAt: failureResetAt(failure, yield* Ref.get(rateLimitSnapshot)),
                      }
                    : failure,
                ...(input.providerRetry === undefined
                  ? {}
                  : {
                      retry: input.providerRetry.retry,
                      retryStartedAt: input.providerRetry.startedAt,
                    }),
                threadDisposition: "reusable",
              };
            }
            return {
              type: "turn.terminal",
              driver: CODEX_PROVIDER,
              providerThreadId: input.context.providerThread.id,
              providerTurnId: input.context.providerTurnId,
              runOrdinal: input.context.input.runOrdinal,
              status: terminalStatus,
              failure: null,
              ...(terminalStatus === "interrupted" && input.context.hookStopReason !== undefined
                ? { reason: input.context.hookStopReason }
                : {}),
              threadDisposition: "reusable",
            };
          },
        );

        const publishRootTerminal = Effect.fnUntraced(function* (
          context: ActiveCodexTurnContext,
          event: CodexRootTerminalEvent,
        ) {
          const current =
            event.status === "failed" && event.failure.class === "usage_limit"
              ? {
                  ...event,
                  failure: {
                    ...event.failure,
                    resetAt:
                      event.failure.resetAt ??
                      failureResetAt(event.failure, yield* Ref.get(rateLimitSnapshot)),
                  },
                }
              : event;
          yield* emitProviderEvent(current);
          if (current.status === "failed" && current.failure.class === "usage_limit") {
            const item = makeProviderFailureTurnItem({
              driver: CODEX_PROVIDER,
              threadId: context.input.threadId,
              runId: context.input.runId,
              nodeId: context.input.rootNodeId,
              providerThreadId: current.providerThreadId,
              providerTurnId: current.providerTurnId,
              itemOrdinal: current.failureItemOrdinal,
              failure: current.failure,
              occurredAt: yield* DateTime.now,
            });
            yield* Ref.update(limitedTurnItems, (items) =>
              new Map(items).set(context.providerThread.id, item),
            );
          }
        });

        const emitRootTerminal = Effect.fnUntraced(function* (
          context: ActiveCodexTurnContext,
          event: CodexRootTerminalEvent,
        ) {
          const awaitsReset =
            event.status === "failed" &&
            event.failure.class === "usage_limit" &&
            event.failure.resetAt == null &&
            // The account read only reports the ordinary allowance, never a sharing cap's reset.
            !classifyCodexManagedError(event.failure.code)?.usageLimit &&
            codexUsageLimitResetAt(yield* Ref.get(rateLimitSnapshot)) === null;
          if (!awaitsReset) yield* publishRootTerminal(context, event);
          const continued = continuedRuns.get(context.providerTurnId) ?? [context];
          if (event.status !== "completed") {
            yield* Ref.update(
              terminalizedNonCompletedNativeTurns,
              (nativeTurns) =>
                new Set([...nativeTurns, ...continued.map((turn) => turn.nativeTurnId)]),
            );
          }
          for (const turn of continued) {
            continuedRuns.delete(turn.providerTurnId);
          }
          const nativeThreadId = context.providerThread.nativeThreadRef?.nativeId;
          const rootProviderThread =
            nativeThreadId == null ? undefined : rootProviderThreads.get(nativeThreadId);
          if (nativeThreadId != null && rootProviderThread !== undefined) {
            rootProviderThreads.set(nativeThreadId, { ...rootProviderThread, status: "idle" });
          }
          // Notification handlers block response routing, so the read and terminal get a fiber.
          if (awaitsReset) {
            yield* readUsageLimitReset.pipe(
              Effect.andThen(
                turnTerminalizationPermit.withPermits(1)(publishRootTerminal(context, event)),
              ),
              Effect.forkIn(scope, { startImmediately: true }),
            );
          }
        });

        const emitOrDeferRootTerminal = Effect.fn("CodexAdapterV2.emitOrDeferRootTerminal")(
          function* (input: {
            readonly context: ActiveCodexTurnContext;
            readonly nativeTurnId: string;
            readonly status: OrchestrationV2ProviderTurn["status"];
            readonly failureMessage?: string;
            readonly failureErrorInfo?: CodexErrorInfo;
            readonly providerRetry?: ActiveCodexProviderRetry;
            readonly goalHoldTurn?: OrchestrationV2ProviderTurn;
          }) {
            const event = yield* makeRootTerminalEvent(input);
            const hasActiveDescendants = Array.from((yield* Ref.get(activeTurns)).values()).some(
              (candidate) => isContinuedRunDescendant(candidate, input.context),
            );
            // Failed roots leave native children working and must settle without them.
            // Interrupted roots wait for descendants because Stop interrupts the whole lineage.
            if (event.status === "failed" && hasActiveDescendants) {
              yield* emitRootTerminal(input.context, { ...event, childTurnsContinue: true });
              return;
            }
            if (event.status !== "completed" && hasActiveDescendants) {
              yield* Ref.update(deferredRootTerminals, (current) => {
                const updated = new Map(current);
                updated.set(input.nativeTurnId, { context: input.context, event });
                return updated;
              });
              return;
            }
            const nativeThreadId = input.context.providerThread.nativeThreadRef?.nativeId;
            if (
              input.goalHoldTurn !== undefined &&
              event.status === "completed" &&
              nativeThreadId != null
            ) {
              const hasPendingWake = (yield* mcpSessions.read(input.context.input.threadId))
                ?.hasPendingWake;
              // Decided under the permit before the hold exists, so no continuation can adopt it first.
              const waitsForWake = hasPendingWake === undefined ? false : yield* hasPendingWake;
              goalHolds.set(nativeThreadId, {
                context: input.context,
                event,
                completedTurn: input.goalHoldTurn,
                next: yield* Deferred.make<ActiveCodexTurnContext | undefined>(),
              });
              // A continuation turn would only report that the delegated task is still running.
              const claimed = waitsForWake ? claimGoalHold(nativeThreadId, event) : undefined;
              if (claimed !== undefined) {
                yield* pauseClaimedGoal(nativeThreadId, claimed, true).pipe(Effect.forkIn(scope));
                return;
              }
              yield* Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE).pipe(
                Effect.andThen(
                  Effect.suspend(() => {
                    const hold = claimGoalHold(nativeThreadId, event);
                    return hold === undefined
                      ? Effect.void
                      : pauseClaimedGoal(nativeThreadId, hold, false);
                  }),
                ),
                Effect.andThen(
                  turnTerminalizationPermit.withPermits(1)(releaseGoalHold(nativeThreadId, event)),
                ),
                // The grace window starts when the turn completes, not when the fiber is scheduled.
                Effect.forkIn(scope, { startImmediately: true }),
              );
              return;
            }
            yield* emitRootTerminal(input.context, event);
          },
        );

        // A goal must not read as active once its run ends. Claiming the hold first
        // makes a turn Codex starts during the pause an orphan, which is stopped.
        const claimGoalHold = (nativeThreadId: string, expected: CodexRootTerminalEvent) => {
          const hold = goalHolds.get(nativeThreadId);
          if (
            hold?.event !== expected ||
            goalStops.has(nativeThreadId) ||
            goalsByNativeThread.get(nativeThreadId)?.status !== "active"
          ) {
            return undefined;
          }
          goalHolds.delete(nativeThreadId);
          // Keeps a turn Codex starts before settlement an orphan once the goal reads paused.
          goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1);
          return hold;
        };

        const pauseClaimedGoal = Effect.fnUntraced(function* (
          nativeThreadId: string,
          hold: CodexGoalHold,
          waitsForWake: boolean,
        ) {
          if (!waitsForWake) {
            yield* Effect.logWarning("orchestration-v2.codex-goal-continuation-expired", {
              nativeThreadId,
            });
          }
          yield* client
            .request("thread/goal/set", { threadId: nativeThreadId, status: "paused" })
            .pipe(
              // The hold is already claimed, so an unanswered pause must not keep the run open.
              Effect.timeout("10 seconds"),
              Effect.flatMap(({ goal }) => {
                goalsByNativeThread.set(nativeThreadId, providerGoalFromCodex(goal));
                if (waitsForWake && goal.status === "paused")
                  goalsPausedForWake.add(nativeThreadId);
                return emitGoalUpdate(nativeThreadId);
              }),
              Effect.catch((cause) =>
                Effect.logWarning("orchestration-v2.codex-goal-pause-failed", { cause }),
              ),
              Effect.ensuring(
                turnTerminalizationPermit.withPermits(1)(settleGoalHold(hold, "completed")),
              ),
              Effect.ensuring(
                Effect.sync(() => {
                  const stops = (goalStops.get(nativeThreadId) ?? 1) - 1;
                  if (stops > 0) goalStops.set(nativeThreadId, stops);
                  else goalStops.delete(nativeThreadId);
                }),
              ),
            );
        });

        /** Hold the terminalization permit and remove the goalHolds entry before calling. */
        const settleGoalHold = Effect.fnUntraced(function* (
          hold: CodexGoalHold,
          status: "completed" | "interrupted",
        ) {
          yield* emitProviderEvent({
            type: "provider_turn.updated",
            driver: CODEX_PROVIDER,
            threadId: hold.context.projectionThreadId,
            providerTurn: hold.completedTurn,
          });
          yield* emitRootTerminal(hold.context, { ...hold.event, status });
          yield* Deferred.succeed(hold.next, undefined);
          const nativeThreadId = hold.context.providerThread.nativeThreadRef?.nativeId;
          if (status === "completed" && nativeThreadId != null) {
            yield* samplePromptSuggestion(
              hold.context,
              nativeThreadId,
              hold.context.nativeTurnId,
            ).pipe(Effect.forkIn(scope));
          }
        });

        const releaseGoalHold = Effect.fnUntraced(function* (
          nativeThreadId: string,
          expected?: CodexRootTerminalEvent,
        ) {
          const hold = goalHolds.get(nativeThreadId);
          if (hold === undefined || goalStops.has(nativeThreadId)) return;
          if (expected !== undefined && hold.event !== expected) return;
          goalHolds.delete(nativeThreadId);
          yield* settleGoalHold(hold, "completed");
        });

        const flushReadyRootTerminals = Effect.fn("CodexAdapterV2.flushReadyRootTerminals")(
          function* () {
            const activeTurnContexts = Array.from((yield* Ref.get(activeTurns)).values());
            const readyEvents = yield* Ref.modify(deferredRootTerminals, (current) => {
              const updated = new Map(current);
              const ready: Array<{
                context: ActiveCodexTurnContext;
                event: CodexRootTerminalEvent;
              }> = [];
              for (const [nativeTurnId, deferred] of current) {
                if (
                  !activeTurnContexts.some((candidate) =>
                    isContinuedRunDescendant(candidate, deferred.context),
                  )
                ) {
                  updated.delete(nativeTurnId);
                  ready.push(deferred);
                }
              }
              return [ready, updated] as const;
            });
            for (const ready of readyEvents) {
              yield* emitRootTerminal(ready.context, ready.event);
            }
          },
        );

        const projectGeneratedImages = Effect.fn("CodexAdapterV2.projectGeneratedImages")(
          function* (context: ActiveCodexTurnContext, nativeTurnId: string) {
            const state = imageTurns.get(nativeTurnId);
            const firstImageItemId = state?.saved.keys().next().value;
            if (state === undefined || firstImageItemId === undefined) return;
            const finalAnswer = state.answer;
            const text = appendGeneratedImageEmbeds(finalAnswer?.text ?? "", state.saved.values());
            if (text === (finalAnswer?.text ?? "")) {
              state.saved.clear();
              return;
            }
            const message = {
              id: finalAnswer?.id ?? `generated-image-message:${firstImageItemId}`,
              text,
            };
            state.answer = message;
            const artifacts = yield* buildAgentMessageArtifacts(context, message, true);
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "message.updated",
              driver: CODEX_PROVIDER,
              message: artifacts.message,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
            state.saved.clear();
          },
        );

        const finalizeCodexTurn = (input: {
          readonly context: ActiveCodexTurnContext;
          readonly nativeTurnId: string;
          readonly status: OrchestrationV2ProviderTurn["status"];
          readonly completedAt: DateTime.Utc;
          readonly failureMessage?: string;
          readonly failureErrorInfo?: CodexErrorInfo;
          readonly continueRun?: boolean;
          readonly stalledRetry?: {
            readonly state: ActiveCodexProviderRetry;
            readonly failure: NonNullable<ActiveCodexTurnContext["latestProviderFailure"]>;
          };
        }) =>
          // Capacity continuation already holds the permit across finalization and restart.
          (input.continueRun ? Function.identity : turnTerminalizationPermit.withPermits(1))(
            Effect.gen(function* () {
              const current = (yield* Ref.get(activeTurns)).get(input.nativeTurnId);
              if (current !== input.context) {
                return false;
              }
              if (input.stalledRetry !== undefined) {
                // Checked under the permit so item/started cannot clear the retry mid-finalize.
                const retries = yield* Ref.get(providerRetries);
                if (
                  retries.get(input.context.providerTurnId) !== input.stalledRetry.state ||
                  (yield* Ref.get(interruptingNativeTurns)).has(input.nativeTurnId)
                )
                  return false;
                input.context.latestProviderFailure ??= input.stalledRetry.failure;
              }
              input.context.fileChangeKindsByItem.clear();
              const providerRetry = yield* Ref.modify(providerRetries, (current) => {
                const retry = current.get(input.context.providerTurnId);
                if (retry === undefined) {
                  return [undefined, current] as const;
                }
                const updated = new Map(current);
                updated.delete(input.context.providerTurnId);
                return [retry, updated] as const;
              });
              if (
                providerRetry !== undefined &&
                !input.continueRun &&
                (providerRetry.context !== undefined ||
                  input.status !== "failed" ||
                  input.context.subagent !== null)
              ) {
                yield* emitProviderEvent({
                  type: "turn_item.updated",
                  driver: CODEX_PROVIDER,
                  turnItem: codexRetryTurnItem(
                    input.context,
                    providerRetry,
                    input.status === "failed" &&
                      providerRetry.context !== input.context &&
                      providerRetry.context !== undefined
                      ? "interrupted"
                      : providerTurnStatusToTerminal(input.status),
                    input.completedAt,
                  ),
                });
              }
              if (input.status !== "completed" && !input.continueRun) {
                yield* Ref.update(terminalizedNonCompletedNativeTurns, (current) => {
                  const updated = new Set(current);
                  for (const turn of continuedRuns.get(input.context.providerTurnId) ?? [
                    input.context,
                  ])
                    updated.add(turn.nativeTurnId);
                  return updated;
                });
              }
              yield* agentMessageDeltas.flushTurn(input.nativeTurnId);
              yield* reasoningDeltas.flushTurn(input.nativeTurnId);
              yield* planDeltas.flushTurn(input.nativeTurnId);
              for (const [key, part] of reasoningParts) {
                if (part.turnId === input.nativeTurnId) reasoningParts.delete(key);
              }
              // Failed compaction can end the turn without item/completed.
              for (const nativeItemId of input.context.runningCompactionItemIds) {
                yield* emitCompactionItem(
                  input.context,
                  nativeItemId,
                  providerTurnStatusToTerminal(input.status),
                  input.completedAt,
                );
              }
              yield* projectGeneratedImages(input.context, input.nativeTurnId);
              if (input.status !== "completed") {
                yield* clearImageWork(input.context);
              } else {
                const images = imageTurns.get(input.nativeTurnId);
                if (images !== undefined) {
                  yield* emitImageWorkState(input.context, images.running.size > 0);
                }
              }
              const nativeThreadId = input.context.providerThread.nativeThreadRef?.nativeId;
              const activation =
                nativeThreadId == null ? undefined : goalActivations.get(nativeThreadId);
              const holdsForGoal =
                input.context.subagent === null &&
                input.status === "completed" &&
                nativeThreadId != null &&
                (goalsByNativeThread.get(nativeThreadId)?.status === "active" ||
                  (activation !== undefined && !activation.stopped));
              const executedModel = reportedTurnModel(nativeThreadId ?? "", input.nativeTurnId);
              const completedTurn: OrchestrationV2ProviderTurn = {
                id: input.context.providerTurnId,
                providerThreadId: input.context.providerThread.id,
                nodeId: input.context.providerNodeId,
                runAttemptId:
                  input.context.subagent === null ? input.context.input.attemptId : null,
                nativeTurnRef: {
                  driver: CODEX_PROVIDER,
                  nativeId: input.nativeTurnId,
                  strength: "strong",
                },
                ordinal: input.context.providerTurnOrdinal,
                status: input.status,
                startedAt: input.context.startedAt,
                completedAt: input.completedAt,
                turnTokenUsage: completeCodexTurnTokenUsage(
                  usageStateForThread(
                    input.context.providerThread.nativeThreadRef?.nativeId ??
                      String(input.context.providerThread.id),
                  ),
                  input.nativeTurnId,
                  input.status === "completed",
                  executedModel,
                ),
                ...(executedModel === undefined ? {} : { model: executedModel }),
              };
              if (!holdsForGoal) {
                yield* emitProviderEvent({
                  type: "provider_turn.updated",
                  driver: CODEX_PROVIDER,
                  threadId: input.context.projectionThreadId,
                  providerTurn: completedTurn,
                });
              }
              if (input.context.subagent !== null) {
                yield* emitProviderEvent({
                  type: "node.updated",
                  driver: CODEX_PROVIDER,
                  node: {
                    id: input.context.providerNodeId,
                    threadId: input.context.projectionThreadId,
                    runId: null,
                    parentNodeId: null,
                    rootNodeId: input.context.rootNodeId,
                    kind: "root_turn",
                    status: input.status,
                    countsForRun: false,
                    providerThreadId: input.context.providerThread.id,
                    providerTurnId: input.context.providerTurnId,
                    nativeItemRef: input.context.subagent.task.nativeTaskRef,
                    runtimeRequestId: null,
                    checkpointScopeId: null,
                    startedAt: input.context.providerNodeStartedAt,
                    completedAt: input.completedAt,
                  },
                });
                yield* emitProviderEvent({
                  type: "provider_thread.updated",
                  driver: CODEX_PROVIDER,
                  providerThread: {
                    ...input.context.providerThread,
                    status: "idle",
                    updatedAt: input.completedAt,
                  },
                });
                yield* emitProviderEvent({
                  type: "node.updated",
                  driver: CODEX_PROVIDER,
                  node: {
                    id: input.context.subagent.subagentNodeId,
                    threadId: input.context.subagent.parentContext.projectionThreadId,
                    runId: input.context.subagent.parentContext.projectionRunId,
                    parentNodeId: input.context.subagent.parentContext.itemParentNodeId,
                    rootNodeId: input.context.subagent.parentContext.rootNodeId,
                    kind: "subagent",
                    status: input.status,
                    countsForRun: false,
                    providerThreadId: input.context.providerThread.id,
                    providerTurnId: input.context.subagent.parentContext.providerTurnId,
                    nativeItemRef: input.context.subagent.task.nativeTaskRef,
                    runtimeRequestId: null,
                    checkpointScopeId: null,
                    startedAt: input.context.subagent.startedAt,
                    completedAt: input.completedAt,
                  },
                });
                yield* emitSubagentTaskUpdate({
                  subagent: input.context.subagent,
                  status: input.status,
                  completedAt: input.completedAt,
                });
                // Codex does not wake a completed parent turn, leaving child results unread.
                // Only direct children wake an idle root turn; grandchildren report to their
                // own parent, and running root turns receive results natively.
                const parent = input.context.subagent.parentContext;
                const parentNativeThreadId = parent.providerThread.nativeThreadRef?.nativeId;
                if (
                  continuationRequests !== undefined &&
                  parent.subagent === null &&
                  (input.status === "completed" || input.status === "failed") &&
                  !(yield* Ref.get(terminalizedNonCompletedNativeTurns)).has(parent.nativeTurnId) &&
                  (parentNativeThreadId == null ||
                    !capacityContinuations.has(parentNativeThreadId)) &&
                  ![...(yield* Ref.get(activeTurns)).values()].some(
                    (active) =>
                      active.subagent === null &&
                      active.providerThread.id === parent.providerThread.id,
                  )
                ) {
                  const task = input.context.subagent.task;
                  yield* continuationRequests.offer({
                    threadId: parent.projectionThreadId,
                    providerThreadId: parent.providerThread.id,
                    driver: CODEX_PROVIDER,
                    detail: codexNativeSubagentDetail(task),
                    // The detail is the whole wake; Codex buffers no output for it.
                    delivery: "message_text",
                    notification: backgroundWorkNotification([
                      {
                        kind: "subagent",
                        label: codexNativeSubagentLabel(task),
                        outcome: input.status,
                        childThreadId: input.context.subagent.childThreadId,
                      },
                    ]),
                  });
                }
              }
              if (input.status === "interrupted" || input.status === "failed") {
                yield* terminalizeRunningCommandItems(
                  input.context,
                  input.nativeTurnId,
                  input.status,
                  input.completedAt,
                );
                yield* terminalizeMcpAppCaptures(input.context, input.nativeTurnId);
              }
              const dynamicToolStatus: "cancelled" | "interrupted" | "failed" =
                input.status === "interrupted" || input.status === "failed"
                  ? input.status
                  : "cancelled";
              yield* terminalizeRunningDynamicTools(
                input.context,
                input.nativeTurnId,
                dynamicToolStatus,
                input.completedAt,
                input.status === "interrupted" || input.status === "failed",
              );
              if (input.context.subagent === null && !input.continueRun) {
                yield* emitOrDeferRootTerminal({
                  ...input,
                  ...(providerRetry === undefined ? {} : { providerRetry }),
                  ...(holdsForGoal ? { goalHoldTurn: completedTurn } : {}),
                });
              }
              const waiter = (yield* Ref.get(turnWaiters)).get(input.nativeTurnId);
              if (waiter !== undefined) {
                yield* Deferred.succeed(waiter, undefined);
              }
              const interruptInProgress = (yield* Ref.get(interruptingNativeTurns)).has(
                input.nativeTurnId,
              );
              const retainSettledContext =
                input.status === "completed" &&
                (yield* turnHasRetainedBackgroundWork(input.nativeTurnId));
              // A held goal predicts when settleGoalHold completes it.
              if (
                input.status === "completed" &&
                input.context.subagent === null &&
                !input.continueRun &&
                !holdsForGoal &&
                nativeThreadId != null
              ) {
                yield* samplePromptSuggestion(
                  input.context,
                  nativeThreadId,
                  input.nativeTurnId,
                ).pipe(Effect.forkIn(scope));
              }
              if (retainSettledContext) {
                yield* Ref.update(settledTurns, (current) => {
                  const updated = new Map(current);
                  updated.set(input.nativeTurnId, input.context);
                  return updated;
                });
                // Recheck after recording settlement; background work may have finished before a settled turn existed to release.
                yield* releaseSettledTurnIfIdle(input.nativeTurnId);
              }
              yield* Ref.update(activeTurns, (current) => {
                const updated = new Map(current);
                updated.delete(input.nativeTurnId);
                return updated;
              });
              if (input.context.nativeStartReady !== undefined) {
                yield* Deferred.succeed(input.context.nativeStartReady, undefined);
              }
              if (!input.continueRun && input.context.capacityRetryPending !== undefined) {
                yield* Deferred.succeed(input.context.capacityRetryPending, undefined);
              }
              yield* flushReadyRootTerminals();
              if (!retainSettledContext && !interruptInProgress) {
                yield* Ref.update(runningCommandItemsByTurn, (current) => {
                  if (!current.has(input.nativeTurnId)) {
                    return current;
                  }
                  const updated = new Map(current);
                  updated.delete(input.nativeTurnId);
                  return updated;
                });
                yield* Ref.update(runningDynamicToolsByTurn, (current) => {
                  if (!current.has(input.nativeTurnId)) {
                    return current;
                  }
                  const updated = new Map(current);
                  updated.delete(input.nativeTurnId);
                  return updated;
                });
              }
              if (!retainSettledContext) {
                yield* Ref.update(completedFinalAnswerTextsByTurn, (current) => {
                  if (!current.has(input.nativeTurnId)) {
                    return current;
                  }
                  const updated = new Map(current);
                  updated.delete(input.nativeTurnId);
                  return updated;
                });
                imageTurns.delete(input.nativeTurnId);
                yield* Ref.update(finalAnswerItemIdsByTurn, (current) => {
                  if (!current.has(input.nativeTurnId)) {
                    return current;
                  }
                  const updated = new Map(current);
                  updated.delete(input.nativeTurnId);
                  return updated;
                });
              }
              return true;
            }),
          );

        // A retry that ends on 429 is the same exhaustion responseTooManyFailedAttempts reports.
        const failStalledProviderRetry = Effect.fn("CodexAdapterV2.failStalledProviderRetry")(
          function* (
            context: ActiveCodexTurnContext,
            state: ActiveCodexProviderRetry,
            errorInfo: CodexErrorInfo | null | undefined,
          ) {
            const finalized = yield* finalizeCodexTurn({
              context,
              nativeTurnId: context.nativeTurnId,
              status: "failed",
              completedAt: yield* DateTime.now,
              stalledRetry: {
                state,
                failure: {
                  nativeMessage: state.nativeMessage,
                  nativeCode: state.nativeCode,
                  failure: {
                    ...state.failure,
                    retryable: false,
                    ...(codexErrorHttpStatus(errorInfo) === 429 ? { class: "usage_limit" } : {}),
                  },
                },
              },
            });
            if (!finalized) return;
            const nativeThreadId = yield* getNativeThreadId(context.providerThread);
            yield* client
              .request("turn/interrupt", { threadId: nativeThreadId, turnId: context.nativeTurnId })
              .pipe(
                Effect.timeout("10 seconds"),
                Effect.catch((cause) =>
                  Effect.logWarning("orchestration-v2.codex-stalled-retry-interrupt-failed", {
                    nativeThreadId,
                    nativeTurnId: context.nativeTurnId,
                    cause,
                  }),
                ),
              );
          },
        );

        // Codex can end an overloaded turn without retry advice. Continue its
        // persisted history with empty input instead of resending the user message.
        const retryCapacityFailure = Effect.fn("CodexAdapterV2.retryCapacityFailure")(function* (
          terminal: Parameters<typeof finalizeCodexTurn>[0],
        ) {
          const context = terminal.context;
          const code =
            terminal.failureErrorInfo == null
              ? context.latestProviderFailure?.nativeCode
              : codexErrorInfoCode(terminal.failureErrorInfo);
          if (
            terminal.status !== "failed" ||
            code !== "serverOverloaded" ||
            context.subagent !== null ||
            (context.input.message.attachments.length === 0 &&
              context.input.message.text.trim().toLowerCase() === "/compact")
          )
            return false;

          return yield* turnTerminalizationPermit.withPermits(1)(
            Effect.gen(function* () {
              if ((yield* Ref.get(activeTurns)).get(terminal.nativeTurnId) !== context) return true;
              if (context.capacityRetryPending !== undefined) return true;
              if (
                (context.capacityRetryAttempt ?? 0) >= CODEX_CAPACITY_MAX_RETRIES ||
                (yield* Ref.get(interruptingNativeTurns)).has(terminal.nativeTurnId)
              )
                return false;

              const attempt = (context.capacityRetryAttempt ?? 0) + 1;
              context.capacityRetryAttempt = attempt;
              const pending = yield* Deferred.make<ActiveCodexTurnContext | undefined>();
              context.capacityRetryPending = pending;
              const retryDelayMs = Math.min(2_000 * 2 ** (attempt - 1), 30_000);
              const now = yield* DateTime.now;
              const previous = (yield* Ref.get(providerRetries)).get(context.providerTurnId);
              const state: ActiveCodexProviderRetry = {
                context: previous?.context ?? context,
                nativeMessage:
                  terminal.failureMessage ??
                  context.latestProviderFailure?.nativeMessage ??
                  "Selected model is at capacity.",
                nativeCode: code,
                failure: makeProviderFailure({
                  message:
                    context.latestProviderFailure?.failure.message ?? terminal.failureMessage,
                  code,
                  class: "provider_error",
                  retryable: true,
                }),
                retry: { attempt, maxAttempts: CODEX_CAPACITY_MAX_RETRIES, retryDelayMs },
                itemOrdinal:
                  previous?.itemOrdinal ??
                  (yield* resolveItemOrdinal(
                    context,
                    `terminal-failure:${context.providerTurnId}`,
                  )),
                startedAt: previous?.startedAt ?? now,
              };
              yield* Ref.update(providerRetries, (current) =>
                new Map(current).set(context.providerTurnId, state),
              );
              yield* emitProviderEvent({
                type: "turn_item.updated",
                driver: CODEX_PROVIDER,
                turnItem: codexRetryTurnItem(context, state, "running", now),
              });
              yield* Effect.sleep(retryDelayMs).pipe(
                Effect.andThen(
                  Effect.gen(function* () {
                    const nativeThreadId = yield* getNativeThreadId(context.providerThread);
                    const claimed = yield* turnTerminalizationPermit.withPermits(1)(
                      Effect.gen(function* () {
                        if (
                          (yield* Ref.get(activeTurns)).get(terminal.nativeTurnId) !== context ||
                          goalStops.has(nativeThreadId) ||
                          (yield* Ref.get(interruptingNativeTurns)).has(terminal.nativeTurnId)
                        )
                          return false;
                        capacityContinuations.set(nativeThreadId, { context, retry: state });
                        yield* finalizeCodexTurn({ ...terminal, continueRun: true });
                        return true;
                      }),
                    );
                    if (!claimed) return;
                    // Notifications before the RPC response also take the terminalization permit.
                    yield* startNativeTurn(
                      {
                        ...context.input,
                        providerTurnOrdinal: context.providerTurnOrdinal + 1,
                      },
                      [],
                    ).pipe(
                      Effect.catch((cause) =>
                        Effect.gen(function* () {
                          if (yield* Deferred.isDone(pending)) return;
                          yield* Effect.logWarning(
                            "orchestration-v2.codex-capacity-retry-start-failed",
                            {
                              providerTurnId: context.providerTurnId,
                              cause,
                            },
                          );
                          yield* emitProviderEvent({
                            type: "turn_item.updated",
                            driver: CODEX_PROVIDER,
                            turnItem: codexRetryTurnItem(
                              context,
                              state,
                              "failed",
                              yield* DateTime.now,
                            ),
                          });
                          yield* Deferred.succeed(pending, undefined);
                          yield* turnTerminalizationPermit.withPermits(1)(
                            emitOrDeferRootTerminal({ ...terminal, providerRetry: state }),
                          );
                        }),
                      ),
                      Effect.ensuring(
                        Effect.sync(() => {
                          if (capacityContinuations.get(nativeThreadId)?.context === context)
                            capacityContinuations.delete(nativeThreadId);
                        }),
                      ),
                    );
                  }),
                ),
                Effect.forkIn(scope, { startImmediately: true }),
              );
              return true;
            }),
          );
        });

        yield* client.handleServerNotification("turn/completed", (payload) =>
          Effect.gen(function* () {
            const compacted = yield* Ref.modify(threadsAwaitingContextRestore, (current) => {
              if (!current.has(payload.threadId)) return [false, current] as const;
              const next = new Set(current);
              next.delete(payload.threadId);
              return [true, next] as const;
            });
            if (compacted)
              // Fork requests off the notification reader so it can read their responses.
              yield* restoreAdditionalContext(payload.threadId).pipe(
                Effect.forkIn(scope, { startImmediately: true }),
              );
            const context = (yield* Ref.get(activeTurns)).get(payload.turn.id);
            if (context === undefined) {
              return;
            }
            const nativeStatus = mapCodexTurnStatus(payload.turn.status);
            const status =
              nativeStatus === "completed" &&
              (yield* Ref.get(interruptingNativeTurns)).has(payload.turn.id)
                ? "interrupted"
                : nativeStatus;
            const terminal = {
              context,
              nativeTurnId: payload.turn.id,
              status,
              completedAt: codexTimestamp(payload.turn.completedAt),
              ...(payload.turn.error?.message === undefined
                ? {}
                : {
                    failureMessage: payload.turn.error.message,
                    ...(payload.turn.error.codexErrorInfo == null
                      ? {}
                      : { failureErrorInfo: payload.turn.error.codexErrorInfo }),
                  }),
            };
            if (!(yield* retryCapacityFailure(terminal))) yield* finalizeCodexTurn(terminal);
          }),
        );

        /** Settles a /goal turn that never reaches Codex. State changes reply through the goal row. */
        const completeGoalCommandTurn = Effect.fnUntraced(function* (
          turnInput: ProviderAdapter.ProviderAdapterV2TurnInput,
          reply: string | null,
        ) {
          const now = yield* DateTime.now;
          const nativeId = `goal-command:${turnInput.attemptId}`;
          const context: ActiveCodexTurnContext = {
            input: turnInput,
            projectionAppThread: turnInput.appThread,
            projectionThreadId: turnInput.threadId,
            projectionRunId: turnInput.runId,
            nativeTurnId: nativeId,
            providerThread: turnInput.providerThread,
            providerTurnId: idAllocator.derive.providerTurn({
              driver: CODEX_PROVIDER,
              nativeTurnId: nativeId,
            }),
            providerTurnOrdinal: turnInput.providerTurnOrdinal,
            providerNodeId: turnInput.rootNodeId,
            providerNodeStartedAt: now,
            itemParentNodeId: turnInput.rootNodeId,
            rootNodeId: turnInput.rootNodeId,
            subagent: null,
            startedAt: now,
            itemPositions: new Map(),
            fileChangeKindsByItem: new Map(),
            messagePhases: new Map(),
            runningCompactionItemIds: new Set(),
          };
          const providerTurn = {
            id: context.providerTurnId,
            providerThreadId: turnInput.providerThread.id,
            nodeId: turnInput.rootNodeId,
            runAttemptId: turnInput.attemptId,
            nativeTurnRef: null,
            ordinal: turnInput.providerTurnOrdinal,
            status: "running",
            startedAt: now,
            completedAt: null,
          } satisfies OrchestrationV2ProviderTurn;
          yield* emitProviderEvent({
            type: "provider_turn.updated",
            driver: CODEX_PROVIDER,
            threadId: turnInput.threadId,
            providerTurn,
          });
          yield* rememberRootProviderThread(turnInput.providerThread);
          if (reply !== null) {
            const artifacts = yield* buildAgentMessageArtifacts(
              context,
              { id: nativeId, text: reply },
              true,
            );
            yield* emitProviderEvent({
              type: "node.updated",
              driver: CODEX_PROVIDER,
              node: artifacts.node,
            });
            yield* emitProviderEvent({
              type: "message.updated",
              driver: CODEX_PROVIDER,
              message: artifacts.message,
            });
            yield* emitProviderEvent({
              type: "turn_item.updated",
              driver: CODEX_PROVIDER,
              turnItem: artifacts.turnItem,
            });
          }
          yield* emitProviderEvent({
            type: "provider_turn.updated",
            driver: CODEX_PROVIDER,
            threadId: turnInput.threadId,
            providerTurn: { ...providerTurn, status: "completed", completedAt: now },
          });
          yield* emitRootTerminal(context, {
            type: "turn.terminal",
            driver: CODEX_PROVIDER,
            providerThreadId: turnInput.providerThread.id,
            providerTurnId: context.providerTurnId,
            runOrdinal: turnInput.runOrdinal,
            status: "completed",
            failure: null,
            threadDisposition: "reusable",
          });
        });

        const heldGoalTurn = (
          providerThread: OrchestrationV2ProviderThread,
          providerTurnId: ProviderTurnId,
        ) => {
          const nativeThreadId = providerThread.nativeThreadRef?.nativeId;
          const hold = nativeThreadId == null ? undefined : goalHolds.get(nativeThreadId);
          return hold?.context.providerTurnId === providerTurnId && nativeThreadId != null
            ? { nativeThreadId, hold }
            : undefined;
        };

        /** Resolve Stop after pausing; Codex can settle or continue to a newer turn while pause is in flight. */
        const resolveGoalStopTarget = Effect.fnUntraced(function* (
          turnInput: ProviderAdapter.ProviderAdapterV2InterruptInput,
        ) {
          const goalThreadId = turnInput.providerThread.nativeThreadRef?.nativeId;
          if (goalThreadId == null) {
            return { turnInput, goalTurns: [] as ReadonlyArray<ActiveCodexTurnContext> };
          }
          goalStops.set(goalThreadId, (goalStops.get(goalThreadId) ?? 0) + 1);
          return yield* Effect.gen(function* () {
            // The run can settle while the pause is in flight, which drops its record.
            const requestedTurnId = latestRunTurnId(turnInput.providerTurnId);
            const heldBefore = goalHolds.get(goalThreadId)?.context;
            const turnsBefore = [
              ...(continuedRuns.get(requestedTurnId) ?? []),
              ...Array.from((yield* Ref.get(activeTurns)).values()).filter(
                (context) => context.providerTurnId === requestedTurnId,
              ),
              ...(heldBefore?.providerTurnId === requestedTurnId ? [heldBefore] : []),
            ];
            const activation = goalActivations.get(goalThreadId);
            if (activation !== undefined) activation.stopped = true;
            if (goalsByNativeThread.get(goalThreadId)?.status === "active") {
              yield* client
                .request("thread/goal/set", { threadId: goalThreadId, status: "paused" })
                .pipe(
                  Effect.tap(({ goal }) =>
                    Effect.sync(() =>
                      goalsByNativeThread.set(goalThreadId, providerGoalFromCodex(goal)),
                    ),
                  ),
                  Effect.catch((cause) =>
                    Effect.logWarning("orchestration-v2.codex-goal-pause-failed", { cause }),
                  ),
                );
            }
            return yield* turnTerminalizationPermit.withPermits(1)(
              Effect.gen(function* () {
                const providerTurnId = latestRunTurnId(turnInput.providerTurnId);
                const hold = goalHolds.get(goalThreadId);
                const held = hold !== undefined && hold.context.providerTurnId === providerTurnId;
                const goalTurns: ReadonlyArray<ActiveCodexTurnContext> = Array.from(
                  new Set([
                    ...turnsBefore,
                    ...(continuedRuns.get(providerTurnId) ?? []),
                    ...(held ? [hold.context] : []),
                  ]),
                );
                if (!held) return { turnInput: { ...turnInput, providerTurnId }, goalTurns };
                goalHolds.delete(goalThreadId);
                yield* settleGoalHold(hold, "interrupted");
                return {
                  turnInput: { ...turnInput, providerTurnId, requestRuntimeRestart: true },
                  goalTurns,
                };
              }),
            );
          }).pipe(
            Effect.ensuring(
              Effect.sync(() => {
                const stops = (goalStops.get(goalThreadId) ?? 1) - 1;
                if (stops > 0) goalStops.set(goalThreadId, stops);
                else goalStops.delete(goalThreadId);
              }),
            ),
          );
        });

        // Codex refuses to resume a session archived elsewhere until it is unarchived, and
        // one it is still unloading after thread/unsubscribe until it emits thread/closed.
        const resumeNativeThread = Effect.fn("CodexAdapterV2.resumeNativeThread")(function* (
          threadId: string,
          params: Record<string, unknown>,
        ) {
          const resume = client.raw.request("thread/resume", { threadId, ...params });
          return yield* resume.pipe(
            Effect.catchTags({
              CodexAppServerRequestError: (cause) => {
                if (/\bsession \S+ is archived\b|\bcodex unarchive\b/i.test(cause.errorMessage)) {
                  return client.raw
                    .request("thread/unarchive", { threadId })
                    .pipe(Effect.andThen(resume));
                }
                if (/\bthread \S+ is closing\b/i.test(cause.errorMessage)) {
                  const closing = threadClosings.get(threadId);
                  return (
                    closing === undefined
                      ? Effect.void
                      : Deferred.await(closing).pipe(Effect.timeoutOption("10 seconds"))
                  ).pipe(Effect.andThen(resume));
                }
                return Effect.fail(cause);
              },
            }),
          );
        });

        const startNativeTurn = Effect.fn("CodexAdapterV2.startNativeTurn")(
          function* (
            turnInput: ProviderAdapter.ProviderAdapterV2TurnInput,
            codexInput: ReadonlyArray<CodexSchema.V2TurnStartParams__UserInput>,
          ) {
            const threadId = yield* getNativeThreadId(turnInput.providerThread);
            const mcpSession = yield* mcpSessions.read(turnInput.threadId);
            const models = adapterOptions.models === undefined ? [] : yield* adapterOptions.models;
            const configuredReasoningSummary = yield* client
              .request("config/read", { cwd: turnInput.runtimePolicy.cwd, includeLayers: false })
              .pipe(
                Effect.map((response) => response.config.model_reasoning_summary ?? undefined),
                Effect.catch(() =>
                  Effect.logDebug("Codex config read failed.").pipe(Effect.as(undefined)),
                ),
              );
            const turnStartParams = yield* buildCodexTurnStartParams({
              cyberAccess: codexCyberAccess(models, turnInput.modelSelection.model),
              nativeThreadId: threadId,
              codexInput,
              runtimePolicy: turnInput.runtimePolicy,
              modelSelection: turnInput.modelSelection,
              hasT3Mcp: mcpSession !== undefined,
              browserToolsAvailable: mcpSession?.browserToolsAvailable ?? true,
              deviceToolsAvailable: mcpSession?.capabilities?.has("device") ?? false,
              omitServiceTier: adapterOptions.resolveRuntime !== undefined,
              ...(turnInput.appContext === undefined ? {} : { appContext: turnInput.appContext }),
              ...(configuredReasoningSummary === undefined ? {} : { configuredReasoningSummary }),
            });
            yield* Ref.update(pendingRootTurns, (current) => {
              const updated = new Map(current);
              updated.set(threadId, turnInput);
              return updated;
            });
            yield* Ref.update(additionalContextByThread, (current) => {
              const next = new Map(current);
              if (turnStartParams.additionalContext)
                next.set(threadId, turnStartParams.additionalContext);
              else next.delete(threadId);
              return next;
            });
            // Completion can remove the old context while turn/start is in flight.
            const previousTurns = new Map([
              ...(yield* Ref.get(settledTurns)),
              ...(yield* Ref.get(activeTurns)),
            ]);
            const started = yield* client.request("turn/start", turnStartParams);
            const previousTurn =
              previousTurns.get(started.turn.id) ??
              (yield* Ref.get(activeTurns)).get(started.turn.id);
            if (
              previousTurn !== undefined &&
              previousTurn.input.attemptId !== turnInput.attemptId
            ) {
              return yield* new ProviderAdapter.ProviderAdapterProtocolError({
                driver: CODEX_PROVIDER,
                detail:
                  "Codex returned a turn belonging to an earlier run. Stop the existing turn before retrying this message.",
              });
            }
            // A turn/started notification may have registered and completed this turn already.
            if ((yield* Ref.get(pendingRootTurns)).get(threadId) !== turnInput) return;
            yield* registerRootTurn({
              turnInput,
              nativeTurnId: started.turn.id,
              startedAt: codexTimestamp(started.turn.startedAt),
              waitForNativeStart: started.turn.startedAt === null,
            });
          },
          (effect, turnInput) =>
            effect.pipe(
              Effect.ensuring(
                Effect.flatMap(getNativeThreadId(turnInput.providerThread), (threadId) =>
                  Ref.update(pendingRootTurns, (current) => {
                    if (current.get(threadId) !== turnInput) return current;
                    const updated = new Map(current);
                    updated.delete(threadId);
                    return updated;
                  }),
                ).pipe(Effect.ignore),
              ),
            ),
        );

        /** Starts a turn and then activates its paused goal; Stop cancels activation. */
        const activateGoal = Effect.fnUntraced(function* (
          turnInput: ProviderAdapter.ProviderAdapterV2TurnInput,
          threadId: string,
          codexInput: ReturnType<typeof toCodexInput>,
        ) {
          goalsPausedForWake.delete(threadId);
          // Hold fast first turns until activation.
          const activation = { stopped: false };
          goalActivations.set(threadId, activation);
          yield* Effect.gen(function* () {
            yield* startNativeTurn(turnInput, yield* codexInput);
            if (activation.stopped) return;
            const { goal } = yield* client.request("thread/goal/set", {
              threadId,
              status: "active",
            });
            goalsByNativeThread.set(threadId, providerGoalFromCodex(goal));
            if (activation.stopped) {
              const paused = yield* client.request("thread/goal/set", {
                threadId,
                status: "paused",
              });
              goalsByNativeThread.set(threadId, providerGoalFromCodex(paused.goal));
            }
          }).pipe(
            Effect.ensuring(
              turnTerminalizationPermit.withPermits(1)(
                Effect.gen(function* () {
                  if (goalActivations.get(threadId) === activation) {
                    goalActivations.delete(threadId);
                  }
                  if (goalsByNativeThread.get(threadId)?.status !== "active") {
                    yield* releaseGoalHold(threadId);
                  }
                }),
              ),
            ),
          );
        });

        /**
         * Setting or resuming a goal starts the first goal turn with this run's
         * turn configuration (approvals, sandbox, model, plan mode), since a
         * turn Codex starts on its own reuses the last turn's settings.
         */
        const runGoalCommand = Effect.fn("CodexAdapterV2.runGoalCommand")(
          function* (
            turnInput: ProviderAdapter.ProviderAdapterV2TurnInput,
            command: OrchestrationV2GoalCommand,
          ) {
            if (command.type === "set") yield* validateCodexGoalObjective(command.objective);
            const threadId = yield* getNativeThreadId(turnInput.providerThread);
            rootProviderThreads.set(threadId, turnInput.providerThread);
            const readGoal = client
              .request("thread/goal/get", { threadId })
              .pipe(
                Effect.map((response) =>
                  response.goal == null ? null : providerGoalFromCodex(response.goal),
                ),
              );
            const current = yield* readGoal;
            goalsByNativeThread.set(threadId, current);
            if (command.type === "show") {
              return yield* completeGoalCommandTurn(
                turnInput,
                current === null ? "No goal is set." : describeCodexGoal(current),
              );
            }
            goalsPausedForWake.delete(threadId);
            if (command.type === "clear") {
              const { cleared } = yield* client.request("thread/goal/clear", { threadId });
              goalsByNativeThread.set(threadId, null);
              return yield* completeGoalCommandTurn(turnInput, cleared ? null : "No goal is set.");
            }
            if (current === null && command.type !== "set") {
              return yield* completeGoalCommandTurn(turnInput, "No goal is set.");
            }
            if (command.type === "pause") {
              const { goal } = yield* client.request("thread/goal/set", {
                threadId,
                status: "paused",
              });
              goalsByNativeThread.set(threadId, providerGoalFromCodex(goal));
              return yield* completeGoalCommandTurn(turnInput, null);
            }
            if (command.type === "set") {
              goalUpdatesHeld.add(threadId);
              if (current !== null) yield* client.request("thread/goal/clear", { threadId });
              // Keep paused until our turn starts so Codex cannot start one first.
              const { goal } = yield* client.request("thread/goal/set", {
                threadId,
                objective: command.objective,
                status: "paused",
              });
              goalsByNativeThread.set(threadId, providerGoalFromCodex(goal));
            }
            // A bare objective gets echoed back, and "start this goal" gets a
            // create_goal call that Codex rejects because the goal already exists.
            yield* activateGoal(
              turnInput,
              threadId,
              command.type === "set"
                ? toCodexInput({
                    ...turnInput,
                    message: {
                      ...turnInput.message,
                      text: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`,
                    },
                  })
                : Effect.succeed([]),
            );
          },
          (effect, turnInput) =>
            effect.pipe(
              Effect.ensuring(
                Effect.suspend(() => {
                  const threadId = turnInput.providerThread.nativeThreadRef?.nativeId;
                  return threadId == null || !goalUpdatesHeld.has(threadId)
                    ? Effect.void
                    : releaseGoalUpdates(threadId);
                }),
              ),
              Effect.mapError(
                (cause) =>
                  new ProviderAdapter.ProviderAdapterTurnStartError({
                    driver: CODEX_PROVIDER,
                    threadId: turnInput.threadId,
                    providerThreadId: turnInput.providerThread.id,
                    runId: turnInput.runId,
                    cause,
                  }),
              ),
            ),
        );

        const mcpAppsError = (detail: string) => (cause: unknown) =>
          new ProviderAdapter.ProviderAdapterProtocolError({
            driver: CODEX_PROVIDER,
            detail,
            payload: cause,
          });
        const mcpApps: ProviderAdapter.ProviderAdapterV2McpApps = {
          listTools: Effect.fn("CodexAdapterV2.listTools")(
            function* (toolsInput) {
              const threadId = yield* getNativeThreadId(toolsInput.providerThread);
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(
                  client.request("mcpServerStatus/list", {
                    threadId,
                    serverName: toolsInput.server,
                    detail: "toolsAndAuthOnly",
                  }),
                ),
              );
              const server = response.data.find((entry) => entry.name === toolsInput.server);
              return Object.values(server?.tools ?? {});
            },
            Effect.mapError(mcpAppsError("Failed to list Codex MCP tools.")),
          ),
          callTool: Effect.fn("CodexAdapterV2.callTool")(
            function* (callInput) {
              const threadId = yield* getNativeThreadId(callInput.providerThread);
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(
                  client.request("mcpServer/tool/call", {
                    threadId,
                    server: callInput.server,
                    tool: callInput.tool,
                    arguments: callInput.arguments as Schema.Json,
                  }),
                ),
              );
              return {
                content: response.content,
                ...(response.structuredContent === undefined
                  ? {}
                  : { structuredContent: response.structuredContent }),
                ...(response.isError === true ? { isError: true } : {}),
                ...(response._meta === undefined ? {} : { _meta: response._meta }),
              };
            },
            Effect.mapError(mcpAppsError("Codex MCP tool call failed.")),
          ),
          readResource: Effect.fn("CodexAdapterV2.readResource")(
            function* (readInput) {
              const threadId = yield* getNativeThreadId(readInput.providerThread);
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(
                  client.request("mcpServer/resource/read", {
                    threadId,
                    server: readInput.server,
                    uri: readInput.uri,
                  }),
                ),
              );
              return { contents: response.contents };
            },
            Effect.mapError(mcpAppsError("Codex MCP resource read failed.")),
          ),
        };

        const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = {
          instanceId: adapterOptions.instanceId,
          driver: CODEX_PROVIDER,
          providerSessionId: input.providerSessionId,
          providerSession: session,
          events: Stream.fromEffectRepeat(Queue.take(events)),
          canReuseContextUsage: canReuseCodexContextUsage,
          normalizeThreadLoadModelSelection: normalizeCodexThreadLoadModelSelection,
          // Idle release can beat delayed subagent resumes: Codex reports them completed between turns with no resume-expected signal.
          hasPendingBackgroundWork: Effect.gen(function* () {
            for (const items of (yield* Ref.get(runningCommandItemsByTurn)).values()) {
              if (items.size > 0) {
                return true;
              }
            }
            for (const items of (yield* Ref.get(runningDynamicToolsByTurn)).values()) {
              if (items.size > 0) {
                return true;
              }
            }
            if ((yield* Ref.get(pendingMcpAppCaptures)).size > 0) {
              return true;
            }
            for (const state of imageTurns.values()) {
              if (state.running.size > 0 || state.saved.size > 0) return true;
            }
            for (const subagent of (yield* Ref.get(subagentThreads)).values()) {
              if (subagent.task.status === "running") {
                return true;
              }
            }
            return false;
          }),
          // Untraced: the run's event stream calls this once per provider event.
          hasPendingBackgroundWorkForThread: Effect.fnUntraced(function* (providerThread) {
            const contexts = [
              ...(yield* Ref.get(activeTurns)).values(),
              ...(yield* Ref.get(settledTurns)).values(),
            ];
            const roots = contexts.filter(
              (context) => context.providerThread.id === providerThread.id,
            );
            for (const context of contexts) {
              if (!roots.some((root) => context === root || isDescendantCodexTurn(context, root)))
                continue;
              if (yield* turnHasRetainedBackgroundWork(context.nativeTurnId)) return true;
              if (
                context.subagent !== null &&
                (yield* Ref.get(activeTurns)).has(context.nativeTurnId)
              )
                return true;
            }
            return false;
          }),
          ensureThread: (threadInput) =>
            ensureInitialized.pipe(
              Effect.andThen(mcpSessions.read(threadInput.threadId)),
              Effect.flatMap((mcpSession) =>
                client.request(
                  "thread/start",
                  codexThreadRuntimeParams({
                    threadId: threadInput.threadId,
                    mcpSession,
                    modelSelection: threadInput.modelSelection,
                    runtimePolicy: threadInput.runtimePolicy,
                  }),
                ),
              ),
              Effect.map((response): OrchestrationV2ProviderThread =>
                providerThreadFromCodexThread({
                  appThreadId: threadInput.threadId,
                  idAllocator,
                  ownerNodeId: null,
                  providerSessionId: input.providerSessionId,
                  providerInstanceId: adapterOptions.instanceId,
                  thread: response.thread,
                }),
              ),
              Effect.mapError(
                (cause) =>
                  new ProviderAdapter.ProviderAdapterEnsureThreadError({
                    driver: CODEX_PROVIDER,
                    threadId: threadInput.threadId,
                    cause,
                  }),
              ),
            ),
          resumeThread: Effect.fn("CodexAdapterV2.resumeThread")(
            function* (threadInput) {
              const nativeThreadId = yield* getNativeThreadId(threadInput.providerThread);
              const appThreadId = threadInput.threadId ?? threadInput.providerThread.appThreadId;
              const resume = resumeNativeThread(nativeThreadId, {
                excludeTurns: true,
                ...codexThreadRuntimeParams({
                  threadId: appThreadId,
                  mcpSession: yield* readMcpSession(appThreadId),
                  ...(threadInput.modelSelection === undefined
                    ? {}
                    : { modelSelection: threadInput.modelSelection }),
                  ...(threadInput.runtimePolicy === undefined
                    ? {}
                    : { runtimePolicy: threadInput.runtimePolicy }),
                }),
              });
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(resume),
                Effect.flatMap(decodeCodexResumeMetadata),
              );
              return {
                ...threadInput.providerThread,
                providerSessionId: input.providerSessionId,
                providerInstanceId: adapterOptions.instanceId,
                status: "idle",
                nativeThreadRef: {
                  driver: CODEX_PROVIDER,
                  nativeId: response.thread.id,
                  strength: "strong",
                },
                nativeConversationHeadRef: threadInput.providerThread.nativeConversationHeadRef,
                updatedAt: codexTimestamp(response.thread.updatedAt),
              } satisfies OrchestrationV2ProviderThread;
            },
            (effect, threadInput) =>
              effect.pipe(
                Effect.mapError((cause) =>
                  isCodexActiveWriterError(cause)
                    ? new ProviderAdapter.ProviderAdapterThreadInUseError({
                        driver: CODEX_PROVIDER,
                        providerThreadId: threadInput.providerThread.id,
                        detail: CODEX_THREAD_IN_USE_MESSAGE,
                        cause,
                      })
                    : new ProviderAdapter.ProviderAdapterResumeThreadError({
                        driver: CODEX_PROVIDER,
                        providerSessionId: input.providerSessionId,
                        providerThreadId: threadInput.providerThread.id,
                        cause,
                      }),
                ),
              ),
          ),
          compactThread: Effect.fn("CodexAdapterV2.compactThread")(
            function* (turnInput) {
              const threadId = yield* getNativeThreadId(turnInput.providerThread);
              yield* Ref.update(pendingRootTurns, (current) =>
                new Map(current).set(threadId, turnInput),
              );
              // Clear on interruption; a late turn/started must not be adopted after an interrupted start.
              yield* client.request("thread/compact/start", { threadId }).pipe(
                Effect.onError(() =>
                  Ref.update(pendingRootTurns, (current) => {
                    const next = new Map(current);
                    next.delete(threadId);
                    return next;
                  }),
                ),
              );
            },
            (effect, turnInput) =>
              effect.pipe(
                Effect.mapError(
                  (cause) =>
                    new ProviderAdapter.ProviderAdapterTurnStartError({
                      driver: CODEX_PROVIDER,
                      threadId: turnInput.threadId,
                      providerThreadId: turnInput.providerThread.id,
                      runId: turnInput.runId,
                      cause,
                    }),
                ),
              ),
          ),
          injectHistory: Effect.fn("CodexAdapterV2.injectHistory")(
            function* (input) {
              const threadId = yield* getNativeThreadId(input.providerThread);
              return yield* client
                .request("thread/inject_items", {
                  threadId,
                  items: historyResponseItems(input.messages, input.context),
                })
                .pipe(
                  Effect.as(true),
                  // Retry only unknown-method errors, which precede history mutation. Transport or payload errors leave delivery ambiguous.
                  Effect.catchTags({
                    CodexAppServerRequestError: (error) =>
                      error.code === -32601 ? Effect.succeed(false) : Effect.fail(error),
                  }),
                );
            },
            Effect.mapError(
              (cause) =>
                new ProviderAdapter.ProviderAdapterProtocolError({
                  driver: CODEX_PROVIDER,
                  detail: "Failed to inject historical context",
                  cause,
                }),
            ),
          ),
          startTurn: (turnInput) => {
            const goalCommand =
              turnInput.restartContinuationOfRunId === undefined
                ? parseGoalCommand(turnInput.message)
                : null;
            if (goalCommand !== null) return runGoalCommand(turnInput, goalCommand);
            const nativeThreadId = turnInput.providerThread.nativeThreadRef?.nativeId;
            return Effect.gen(function* () {
              if (
                turnInput.restartContinuationOfRunId === undefined &&
                nativeThreadId != null &&
                goalsPausedForWake.has(nativeThreadId) &&
                goalsByNativeThread.get(nativeThreadId)?.status === "paused"
              ) {
                return yield* activateGoal(turnInput, nativeThreadId, toCodexInput(turnInput));
              }
              yield* startNativeTurn(
                turnInput,
                turnInput.restartContinuationOfRunId === undefined
                  ? yield* toCodexInput(turnInput)
                  : [],
              );
            }).pipe(
              Effect.mapError(
                (cause) =>
                  new ProviderAdapter.ProviderAdapterTurnStartError({
                    driver: CODEX_PROVIDER,
                    threadId: turnInput.threadId,
                    providerThreadId: turnInput.providerThread.id,
                    runId: turnInput.runId,
                    cause,
                  }),
              ),
            );
          },
          steerTurn: Effect.fn("CodexAdapterV2.steerTurn")(
            function* (turnInput) {
              const threadId = yield* getNativeThreadId(turnInput.providerThread);
              const latestTurnId = latestRunTurnId(turnInput.providerTurnId);
              const held = heldGoalTurn(turnInput.providerThread, latestTurnId);
              const providerTurnId =
                held === undefined
                  ? latestTurnId
                  : ((yield* Deferred.await(held.hold.next))?.providerTurnId ?? latestTurnId);
              let activeTurn = Array.from((yield* Ref.get(activeTurns)).values()).find(
                (candidate) => candidate.providerTurnId === providerTurnId,
              );
              const startingRetry = capacityContinuations.get(threadId)?.context;
              if (activeTurn === undefined && startingRetry?.providerTurnId === providerTurnId) {
                activeTurn = startingRetry;
              }
              while (activeTurn?.capacityRetryPending !== undefined) {
                activeTurn = yield* Deferred.await(activeTurn.capacityRetryPending);
              }
              if (activeTurn === undefined) {
                return yield* toProtocolError(
                  `Provider turn ${turnInput.providerTurnId} is not active and cannot be steered.`,
                );
              }

              const codexInput = yield* toCodexInput(turnInput);
              yield* client.request("turn/steer", {
                expectedTurnId: activeTurn.nativeTurnId,
                input: codexInput,
                threadId,
              });
            },
            (effect, turnInput) =>
              effect.pipe(
                Effect.mapError(
                  (cause) =>
                    new ProviderAdapter.ProviderAdapterSteerRunError({
                      driver: CODEX_PROVIDER,
                      providerThreadId: turnInput.providerThread.id,
                      providerTurnId: turnInput.providerTurnId,
                      cause,
                    }),
                ),
              ),
          ),
          // Dropping this connection's subscription lets the shared app-server
          // shut the native thread (and its MCP servers) down once it is idle.
          unloadThread: Effect.fn("CodexAdapterV2.unloadThread")(
            function* (unloadInput) {
              const nativeThreadId = yield* getNativeThreadId(unloadInput.providerThread);
              sessionApprovals.delete(nativeThreadId);
              // Archive also unloads, inline and without thread/closed; resumeNativeThread
              // unarchives it on the next turn.
              if (unloadInput.archive === true) {
                yield* client.request("thread/archive", { threadId: nativeThreadId });
              } else {
                // A resume may already wait on the pending entry; replacing it would strand that wait.
                if (!threadClosings.has(nativeThreadId)) {
                  threadClosings.set(nativeThreadId, yield* Deferred.make<void>());
                }
                yield* client.request("thread/unsubscribe", { threadId: nativeThreadId });
              }
              // No turn/completed arrives for an unloaded thread, so a Stop Codex never
              // confirmed would otherwise keep its turn open for the life of the session.
              const interrupting = yield* Ref.get(interruptingNativeTurns);
              const completedAt = yield* DateTime.now;
              for (const context of (yield* Ref.get(activeTurns)).values()) {
                if (
                  context.providerThread.id !== unloadInput.providerThread.id ||
                  !interrupting.has(context.nativeTurnId)
                )
                  continue;
                yield* finalizeCodexTurn({
                  context,
                  nativeTurnId: context.nativeTurnId,
                  status: "interrupted",
                  completedAt,
                });
              }
            },
            (effect, unloadInput) =>
              effect.pipe(
                Effect.mapError((cause) =>
                  cause._tag === "ProviderAdapterProtocolError"
                    ? cause
                    : new ProviderAdapter.ProviderAdapterProtocolError({
                        driver: CODEX_PROVIDER,
                        detail: `Failed to unload Codex thread for provider thread ${unloadInput.providerThread.id}`,
                        cause,
                      }),
                ),
              ),
          ),
          interruptTurn: Effect.fn("CodexAdapterV2.interruptTurn")(
            function* (requestedInput) {
              const nativeThreadId = requestedInput.providerThread.nativeThreadRef?.nativeId;
              const capacityRetry =
                nativeThreadId == null
                  ? undefined
                  : capacityContinuations.get(nativeThreadId)?.context;
              const startingRetry =
                capacityRetry?.providerTurnId === latestRunTurnId(requestedInput.providerTurnId)
                  ? capacityRetry
                  : undefined;
              if (startingRetry?.capacityRetryPending !== undefined) {
                const started = yield* Deferred.await(startingRetry.capacityRetryPending).pipe(
                  Effect.timeoutOption("10 seconds"),
                );
                if (Option.isNone(started))
                  return yield* toProtocolError(
                    "Codex did not start the capacity retry within 10 seconds; Stop could not be delivered.",
                  );
              }
              const { turnInput, goalTurns } = yield* resolveGoalStopTarget(requestedInput);
              const [activeTurnContexts, settledTurnContexts] =
                yield* turnTerminalizationPermit.withPermits(1)(
                  Effect.gen(function* () {
                    return [
                      Array.from((yield* Ref.get(activeTurns)).values()),
                      Array.from((yield* Ref.get(settledTurns)).values()),
                    ] as const;
                  }),
                );
              const activeTurn =
                activeTurnContexts.find(
                  (candidate) => candidate.providerTurnId === turnInput.providerTurnId,
                ) ??
                (turnInput.requestRuntimeRestart === true
                  ? settledTurnContexts.find(
                      (candidate) => candidate.providerThread.id === turnInput.providerThread.id,
                    )
                  : undefined);
              const lineageRoots = Array.from(
                new Set([
                  ...(activeTurn === undefined ? [] : [activeTurn]),
                  ...(startingRetry === undefined ? [] : [startingRetry]),
                  ...goalTurns,
                ]),
              );
              const inLineage = (context: ActiveCodexTurnContext) =>
                lineageRoots.some(
                  (root) => context === root || isDescendantCodexTurn(context, root),
                );
              const interruptTargetContexts = [
                ...(activeTurn === undefined ? [] : [activeTurn]),
                ...activeTurnContexts.filter(
                  (candidate) => candidate !== activeTurn && inLineage(candidate),
                ),
                ...settledTurnContexts.filter(
                  (candidate) =>
                    candidate !== activeTurn &&
                    ((turnInput.requestRuntimeRestart === true &&
                      candidate.providerThread.id === turnInput.providerThread.id) ||
                      inLineage(candidate)),
                ),
              ];
              if (interruptTargetContexts.length === 0) {
                if (turnInput.requestRuntimeRestart === true) return;
                return yield* toProtocolError(
                  `Provider turn ${turnInput.providerTurnId} is not active and cannot be interrupted.`,
                );
              }
              const interruptTargets: Array<{
                readonly context: ActiveCodexTurnContext;
                readonly completion: Deferred.Deferred<void, never>;
              }> = [];
              const trackedInterruptNativeTurnIds = new Set<string>();
              // Untraced: runs once per lineage turn inside the interruptTurn span.
              const trackInterruptTarget = Effect.fnUntraced(function* (
                context: ActiveCodexTurnContext,
              ) {
                const fresh = yield* Deferred.make<void>();
                // An earlier Stop may still wait on this turn; replacing its waiter would strand it.
                const completion = yield* Ref.modify(turnWaiters, (current) => {
                  const pending = current.get(context.nativeTurnId);
                  return pending === undefined
                    ? [fresh, new Map(current).set(context.nativeTurnId, fresh)]
                    : [pending, current];
                });
                yield* Ref.update(interruptingNativeTurns, (current) =>
                  new Set(current).add(context.nativeTurnId),
                );
                interruptTargets.push({ context, completion });
                trackedInterruptNativeTurnIds.add(context.nativeTurnId);
              });
              for (const context of interruptTargetContexts) {
                yield* trackInterruptTarget(context);
              }
              const registeredActiveTurns = yield* Ref.get(activeTurns);
              for (const target of interruptTargets) {
                if (registeredActiveTurns.get(target.context.nativeTurnId) !== target.context) {
                  yield* Deferred.succeed(target.completion, undefined);
                }
              }

              const cleanupInterruptState = Effect.gen(function* () {
                yield* Ref.update(turnWaiters, (current) => {
                  const updated = new Map(current);
                  for (const target of interruptTargets) {
                    updated.delete(target.context.nativeTurnId);
                  }
                  return updated;
                });
                yield* Ref.update(interruptingNativeTurns, (current) => {
                  const updated = new Set(current);
                  for (const target of interruptTargets) {
                    updated.delete(target.context.nativeTurnId);
                  }
                  return updated;
                });
                yield* Ref.update(runningCommandItemsByTurn, (current) => {
                  const updated = new Map(current);
                  for (const target of interruptTargets) {
                    if (settledTurnContexts.includes(target.context)) continue;
                    updated.delete(target.context.nativeTurnId);
                  }
                  return updated;
                });
              });

              const finalizeRemainingInterruptLineage = Effect.gen(function* () {
                while (true) {
                  const newlyDiscovered: Array<ActiveCodexTurnContext> = [];
                  const activeLineage = yield* turnTerminalizationPermit.withPermits(1)(
                    Effect.gen(function* () {
                      const lineage = Array.from((yield* Ref.get(activeTurns)).values()).filter(
                        inLineage,
                      );
                      for (const context of lineage) {
                        if (trackedInterruptNativeTurnIds.has(context.nativeTurnId)) {
                          continue;
                        }
                        yield* trackInterruptTarget(context);
                        newlyDiscovered.push(context);
                      }
                      return lineage;
                    }),
                  );
                  if (activeLineage.length === 0) {
                    return;
                  }
                  if (newlyDiscovered.length > 0) {
                    const completed = yield* Effect.forEach(
                      newlyDiscovered,
                      (context) =>
                        Effect.flatMap(getNativeThreadId(context.providerThread), (threadId) =>
                          client.request("turn/interrupt", {
                            threadId,
                            turnId: context.nativeTurnId,
                          }),
                        ).pipe(
                          Effect.catch((cause) =>
                            Effect.logWarning(
                              "orchestration-v2.codex-remaining-lineage-interrupt-failed",
                              {
                                nativeTurnId: context.nativeTurnId,
                                providerSessionId: input.providerSessionId,
                                cause,
                              },
                            ),
                          ),
                        ),
                      { concurrency: "unbounded", discard: true },
                    ).pipe(Effect.timeoutOption("10 seconds"));
                    if (Option.isNone(completed)) {
                      yield* Effect.logWarning(
                        "orchestration-v2.codex-remaining-lineage-interrupt-timeout",
                        {
                          nativeTurnIds: newlyDiscovered.map((context) => context.nativeTurnId),
                          providerSessionId: input.providerSessionId,
                        },
                      );
                    }
                  }
                  const completedAt = yield* DateTime.now;
                  for (const context of activeLineage) {
                    yield* finalizeCodexTurn({
                      context,
                      nativeTurnId: context.nativeTurnId,
                      status: "interrupted",
                      completedAt,
                    });
                  }
                }
              });

              const interruptLateDescendants = Effect.gen(function* () {
                const activeLineage = yield* turnTerminalizationPermit.withPermits(1)(
                  Effect.gen(function* () {
                    const lineage = Array.from((yield* Ref.get(activeTurns)).values()).filter(
                      (context) =>
                        !lineageRoots.includes(context) &&
                        inLineage(context) &&
                        !trackedInterruptNativeTurnIds.has(context.nativeTurnId),
                    );
                    for (const context of lineage) {
                      yield* trackInterruptTarget(context);
                    }
                    return lineage;
                  }),
                );
                const completed = yield* Effect.forEach(
                  activeLineage,
                  (context) =>
                    Effect.flatMap(getNativeThreadId(context.providerThread), (threadId) =>
                      client.request("turn/interrupt", {
                        threadId,
                        turnId: context.nativeTurnId,
                      }),
                    ),
                  { concurrency: "unbounded", discard: true },
                ).pipe(Effect.timeoutOption("10 seconds"));
                if (Option.isNone(completed)) {
                  yield* Effect.logWarning(
                    "orchestration-v2.codex-late-descendant-interrupt-timeout",
                    {
                      nativeTurnIds: activeLineage.map((context) => context.nativeTurnId),
                      providerSessionId: input.providerSessionId,
                    },
                  );
                }
              });

              const containedTerminalKeys = new Set<string>();
              const attemptedTerminalKeys = new Set<string>();
              const collectTrackedTerminals = Effect.fn("CodexAdapterV2.collectTrackedTerminals")(
                function* (targets: typeof interruptTargets) {
                  const trackedTerminals = new Map<
                    string,
                    { readonly nativeThreadId: string; readonly processId: string }
                  >();
                  const runningItems = yield* Ref.get(runningCommandItemsByTurn);
                  for (const target of targets) {
                    const nativeThreadId = yield* getNativeThreadId(target.context.providerThread);
                    const items = runningItems.get(target.context.nativeTurnId);
                    for (const item of items?.values() ?? []) {
                      if (item.processId !== undefined) {
                        trackedTerminals.set(`${nativeThreadId}:${item.processId}`, {
                          nativeThreadId,
                          processId: item.processId,
                        });
                      }
                    }
                  }
                  return trackedTerminals;
                },
              );
              const terminateTrackedTerminals = Effect.fn(
                "CodexAdapterV2.terminateTrackedTerminals",
              )(function* (targets: typeof interruptTargets) {
                const trackedTerminals = yield* collectTrackedTerminals(targets);
                const pendingTerminals = Array.from(trackedTerminals.entries())
                  .filter(([key]) => !containedTerminalKeys.has(key))
                  .sort(
                    ([leftKey], [rightKey]) =>
                      Number(attemptedTerminalKeys.has(leftKey)) -
                      Number(attemptedTerminalKeys.has(rightKey)),
                  );
                return yield* Effect.forEach(
                  pendingTerminals,
                  ([key, { nativeThreadId, processId }]) =>
                    Effect.gen(function* () {
                      attemptedTerminalKeys.add(key);
                      yield* terminateBackgroundTerminal(nativeThreadId, processId);
                      containedTerminalKeys.add(key);
                    }).pipe(
                      Effect.as({ success: true as const }),
                      Effect.catch((error) => Effect.succeed({ success: false as const, error })),
                    ),
                );
              });
              const settleInterruptedLineage = Effect.gen(function* () {
                yield* interruptLateDescendants;
                yield* finalizeRemainingInterruptLineage;

                const terminationResults = yield* terminateTrackedTerminals(interruptTargets);
                const failedTermination = terminationResults.find((result) => !result.success);
                if (failedTermination !== undefined && !failedTermination.success) {
                  return yield* failedTermination.error;
                }
                yield* turnTerminalizationPermit.withPermits(1)(
                  Effect.gen(function* () {
                    const completedAt = yield* DateTime.now;
                    for (const target of interruptTargets) {
                      const context = target.context;
                      if ((yield* Ref.get(settledTurns)).get(context.nativeTurnId) !== context)
                        continue;
                      yield* terminalizeRunningCommandItems(
                        context,
                        context.nativeTurnId,
                        "interrupted",
                        completedAt,
                      );
                      yield* terminalizeRunningDynamicTools(
                        context,
                        context.nativeTurnId,
                        "interrupted",
                        completedAt,
                        true,
                      );
                      yield* terminalizeMcpAppCaptures(context, context.nativeTurnId);
                      yield* Ref.update(runningCommandItemsByTurn, (current) => {
                        const updated = new Map(current);
                        updated.delete(context.nativeTurnId);
                        return updated;
                      });
                      yield* clearImageWork(context);
                      yield* releaseSettledTurnIfIdle(context.nativeTurnId);
                    }
                  }),
                );
              });
              // Codex may still be working, so a root stays open and marked interrupting.
              // Whatever ends it later (Codex, Force stop) settles the rest of this Stop.
              const deferToActiveRoots = Effect.gen(function* () {
                const registered = yield* Ref.get(activeTurns);
                const roots = interruptTargets.filter(
                  ({ context }) =>
                    lineageRoots.includes(context) &&
                    registered.get(context.nativeTurnId) === context,
                );
                if (roots.length === 0) return false;
                for (const { context } of roots) {
                  yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", {
                    providerSessionId: input.providerSessionId,
                    providerThreadId: context.providerThread.id,
                    providerTurnId: context.providerTurnId,
                    nativeTurnId: context.nativeTurnId,
                  });
                }
                yield* Effect.forEach(roots, (target) => Deferred.await(target.completion), {
                  concurrency: "unbounded",
                  discard: true,
                }).pipe(
                  Effect.andThen(settleInterruptedLineage),
                  Effect.catchCause((cause) =>
                    Cause.hasInterruptsOnly(cause)
                      ? Effect.void
                      : Effect.logWarning("orchestration-v2.codex-late-stop-settle-failed", {
                          providerSessionId: input.providerSessionId,
                          cause,
                        }),
                  ),
                  Effect.ensuring(cleanupInterruptState),
                  Effect.forkIn(scope),
                );
                return true;
              });

              const unconfirmed = yield* Effect.gen(function* () {
                for (const target of interruptTargets) {
                  const context = target.context;
                  if (context.capacityRetryPending !== undefined) {
                    yield* finalizeCodexTurn({
                      context,
                      nativeTurnId: context.nativeTurnId,
                      status: "interrupted",
                      completedAt: yield* DateTime.now,
                    });
                    continue;
                  }
                  // Null start timestamps ACK queued turns; interruption must wait for turn/started.
                  if (context.nativeStartReady !== undefined) {
                    const ready = yield* Deferred.await(context.nativeStartReady).pipe(
                      Effect.timeoutOption("10 seconds"),
                    );
                    if (Option.isNone(ready)) {
                      return yield* toProtocolError(
                        "Codex did not start the queued turn within 10 seconds; Stop could not be delivered.",
                      );
                    }
                  }
                  if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) {
                    continue;
                  }
                  yield* client.request("turn/interrupt", {
                    threadId: yield* getNativeThreadId(context.providerThread),
                    turnId: context.nativeTurnId,
                  });
                }
                const [, completed] = yield* Effect.all(
                  [
                    terminateTrackedTerminals(interruptTargets),
                    Effect.forEach(
                      interruptTargets,
                      (target) => Deferred.await(target.completion),
                      { concurrency: "unbounded", discard: true },
                    ).pipe(Effect.timeoutOption(CODEX_STOP_CONFIRM_TIMEOUT_MS)),
                  ],
                  { concurrency: "unbounded" },
                );
                if (Option.isSome(completed) || !(yield* deferToActiveRoots)) {
                  yield* settleInterruptedLineage;
                  return false;
                }
                return true;
              }).pipe(
                Effect.onExit((exit) => {
                  if (Exit.isSuccess(exit)) return exit.value ? Effect.void : cleanupInterruptState;
                  const abandon = finalizeRemainingInterruptLineage.pipe(
                    Effect.ensuring(cleanupInterruptState),
                  );
                  // A caller's deadline cutting this short confirms nothing about the turn either.
                  return Cause.hasInterruptsOnly(exit.cause)
                    ? Effect.flatMap(deferToActiveRoots, (deferred) =>
                        deferred ? Effect.void : abandon,
                      )
                    : abandon;
                }),
              );
              if (unconfirmed) {
                return yield* new ProviderAdapter.ProviderAdapterInterruptTimeoutError({
                  driver: CODEX_PROVIDER,
                  providerThreadId: turnInput.providerThread.id,
                  providerTurnId: turnInput.providerTurnId,
                  timeoutMs: CODEX_STOP_CONFIRM_TIMEOUT_MS,
                });
              }
            },
            (effect, requestedInput) =>
              Effect.acquireUseRelease(
                Effect.sync(() => {
                  const nativeThreadId = requestedInput.providerThread.nativeThreadRef?.nativeId;
                  if (nativeThreadId != null)
                    goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1);
                  return nativeThreadId;
                }),
                () => effect,
                (nativeThreadId) =>
                  Effect.sync(() => {
                    if (nativeThreadId == null) return;
                    const stops = (goalStops.get(nativeThreadId) ?? 1) - 1;
                    if (stops > 0) goalStops.set(nativeThreadId, stops);
                    else goalStops.delete(nativeThreadId);
                  }),
              ).pipe(
                Effect.mapError((cause) =>
                  cause._tag === "ProviderAdapterInterruptTimeoutError"
                    ? cause
                    : new ProviderAdapter.ProviderAdapterInterruptError({
                        driver: CODEX_PROVIDER,
                        providerThreadId: requestedInput.providerThread.id,
                        providerTurnId: requestedInput.providerTurnId,
                        cause,
                      }),
                ),
              ),
          ),
          respondToRuntimeRequest: Effect.fn("CodexAdapterV2.respondToRuntimeRequest")(
            function* (requestInput) {
              const pending = (yield* Ref.get(pendingRuntimeRequests)).get(
                String(requestInput.requestId),
              );
              if (pending === undefined) {
                return yield* new ProviderAdapter.ProviderAdapterRuntimeRequestResponseError({
                  driver: CODEX_PROVIDER,
                  requestId: requestInput.requestId,
                  cause: toProtocolError(
                    `No pending Codex runtime request ${requestInput.requestId}.`,
                  ),
                });
              }
              if (pending.type === "user_input") {
                if (requestInput.answers === undefined) {
                  return yield* new ProviderAdapter.ProviderAdapterRuntimeRequestResponseError({
                    driver: CODEX_PROVIDER,
                    requestId: requestInput.requestId,
                    cause: toProtocolError(
                      `Codex user input request ${requestInput.requestId} requires answers.`,
                    ),
                  });
                }
                yield* Deferred.succeed(pending.answers, requestInput.answers);
                return;
              }
              if (requestInput.decision === undefined) {
                return yield* new ProviderAdapter.ProviderAdapterRuntimeRequestResponseError({
                  driver: CODEX_PROVIDER,
                  requestId: requestInput.requestId,
                  cause: toProtocolError(
                    `Codex ${pending.requestKind} request ${requestInput.requestId} requires an approval decision.`,
                  ),
                });
              }
              yield* Deferred.succeed(pending.decision, requestInput.decision);
            },
            (effect, requestInput) =>
              effect.pipe(
                Effect.mapError((cause) =>
                  isProviderAdapterRuntimeRequestResponseError(cause)
                    ? cause
                    : new ProviderAdapter.ProviderAdapterRuntimeRequestResponseError({
                        driver: CODEX_PROVIDER,
                        requestId: requestInput.requestId,
                        cause,
                      }),
                ),
              ),
          ),
          uploadFeedback: Effect.fn("CodexAdapterV2.uploadFeedback")(
            function* (feedbackInput) {
              const threadId = yield* getNativeThreadId(feedbackInput.providerThread);
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(
                  client.request("feedback/upload", {
                    classification: "bug",
                    includeLogs: true,
                    ...(feedbackInput.reason ? { reason: feedbackInput.reason } : {}),
                    threadId,
                  }),
                ),
              );
              return { feedbackId: response.threadId };
            },
            Effect.mapError(
              (cause) =>
                new ProviderAdapter.ProviderAdapterProtocolError({
                  driver: CODEX_PROVIDER,
                  detail: "Failed to upload Codex thread feedback.",
                  payload: cause,
                }),
            ),
          ),
          mcpApps,
          readThreadSnapshot: Effect.fn("CodexAdapterV2.readThreadSnapshot")(
            function* (threadInput) {
              const threadId = yield* getNativeThreadId(threadInput.providerThread);
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(client.request("thread/read", { threadId, includeTurns: true })),
              );
              return {
                providerThread: {
                  ...threadInput.providerThread,
                  nativeThreadRef: {
                    driver: CODEX_PROVIDER,
                    nativeId: response.thread.id,
                    strength: "strong" as const,
                  },
                  nativeConversationHeadRef: threadInput.providerThread.nativeConversationHeadRef,
                  updatedAt: codexTimestamp(response.thread.updatedAt),
                },
                providerTurns: [],
                messages: [],
                runtimeRequests: [],
                providerPayload: response.thread,
              };
            },
            (effect, threadInput) =>
              effect.pipe(
                Effect.mapError(
                  (cause) =>
                    new ProviderAdapter.ProviderAdapterReadThreadSnapshotError({
                      driver: CODEX_PROVIDER,
                      providerThreadId: threadInput.providerThread.id,
                      cause,
                    }),
                ),
              ),
          ),
          rollbackThread: Effect.fn("CodexAdapterV2.rollbackThread")(
            function* (threadInput) {
              const threadId = yield* getNativeThreadId(threadInput.providerThread);
              const numTurns = yield* resolveCodexRollbackTurnCount(threadInput);
              const nativeConversationHeadRef =
                threadInput.target.type === "provider_turn"
                  ? threadInput.target.providerTurn.nativeTurnRef
                  : null;
              if (numTurns === 0) {
                return {
                  providerThread: {
                    ...threadInput.providerThread,
                    nativeConversationHeadRef,
                    status: "idle" as const,
                  },
                  providerTurns: [],
                  messages: [],
                  runtimeRequests: [],
                };
              }
              const { historyMode, loaded } = yield* ensureInitialized.pipe(
                Effect.andThen(readCodexThreadHistoryMetadata(client.raw, threadId)),
              );
              if (historyMode !== "paginated") {
                return yield* new ProviderAdapter.ProviderAdapterRollbackThreadError({
                  driver: CODEX_PROVIDER,
                  providerThreadId: threadInput.providerThread.id,
                  cause: `Cannot roll back Codex thread ${threadId}: the thread uses legacy history, which Codex 0.156 cannot revert.`,
                });
              }
              // thread/revert requires the thread loaded in this app-server; resume after restart or idle release.
              if (!loaded) {
                yield* resumeNativeThread(threadId, {
                  excludeTurns: true,
                  ...codexThreadRuntimeParams({
                    threadId: threadInput.providerThread.appThreadId,
                    mcpSession: yield* readMcpSession(threadInput.providerThread.appThreadId),
                    modelSelection: input.modelSelection,
                    runtimePolicy: input.runtimePolicy,
                  }),
                });
              }
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(revertCodexThread(client, threadId, numTurns)),
              );
              turnTokenUsageByThread.delete(threadId);
              return {
                providerThread: {
                  ...threadInput.providerThread,
                  nativeThreadRef: {
                    driver: CODEX_PROVIDER,
                    nativeId: response.thread.id,
                    strength: "strong" as const,
                  },
                  nativeConversationHeadRef,
                  status: "idle" as const,
                  updatedAt: codexTimestamp(response.thread.updatedAt),
                },
                providerTurns: [],
                messages: [],
                runtimeRequests: [],
                providerPayload: response.thread,
              };
            },
            (effect, threadInput) =>
              effect.pipe(
                Effect.mapError(
                  (cause) =>
                    new ProviderAdapter.ProviderAdapterRollbackThreadError({
                      driver: CODEX_PROVIDER,
                      providerThreadId: threadInput.providerThread.id,
                      cause,
                    }),
                ),
              ),
          ),
          forkThread: Effect.fn("CodexAdapterV2.forkThread")(
            function* (threadInput) {
              const threadId = yield* getNativeThreadId(threadInput.sourceProviderThread);
              const boundary = yield* resolveCodexForkBoundary(threadInput);
              const fork = client.request("thread/fork", {
                threadId,
                excludeTurns: true,
                ...(boundary.lastTurnId === undefined ? {} : { lastTurnId: boundary.lastTurnId }),
                ...codexThreadRuntimeParams({
                  threadId: threadInput.targetThreadId,
                  mcpSession: yield* mcpSessions.read(threadInput.targetThreadId),
                  ...(threadInput.modelSelection === undefined
                    ? {}
                    : { modelSelection: threadInput.modelSelection }),
                  ...(threadInput.runtimePolicy === undefined
                    ? {}
                    : { runtimePolicy: threadInput.runtimePolicy }),
                }),
              });
              const response = yield* ensureInitialized.pipe(
                Effect.andThen(
                  fork.pipe(
                    Effect.catchTags({
                      CodexAppServerRequestError: (cause) =>
                        /\bsession \S+ is archived\b|\bcodex unarchive\b/i.test(cause.errorMessage)
                          ? client.raw
                              .request("thread/unarchive", { threadId })
                              .pipe(Effect.andThen(fork))
                          : Effect.fail(cause),
                    }),
                  ),
                ),
                Effect.mapError(
                  (cause) =>
                    new ProviderAdapter.ProviderAdapterForkThreadError({
                      driver: CODEX_PROVIDER,
                      providerThreadId: threadInput.sourceProviderThread.id,
                      cause,
                    }),
                ),
              );
              let forkedThread = response.thread;
              if (boundary.rollbackTurnCount > 0) {
                const { historyMode } = yield* ensureInitialized.pipe(
                  Effect.andThen(readCodexThreadHistoryMetadata(client.raw, response.thread.id)),
                );
                if (historyMode !== "paginated") {
                  return yield* new ProviderAdapter.ProviderAdapterForkThreadError({
                    driver: CODEX_PROVIDER,
                    providerThreadId: threadInput.sourceProviderThread.id,
                    cause: `Cannot fork Codex thread ${threadId} at provider turn ${threadInput.providerTurnId}: the source turn has no native Codex turn reference, and the forked thread uses legacy history which Codex 0.156 cannot revert.`,
                  });
                }
                forkedThread = (yield* ensureInitialized.pipe(
                  Effect.andThen(
                    revertCodexThread(client, response.thread.id, boundary.rollbackTurnCount),
                  ),
                  Effect.mapError(
                    (cause) =>
                      new ProviderAdapter.ProviderAdapterForkThreadError({
                        driver: CODEX_PROVIDER,
                        providerThreadId: threadInput.sourceProviderThread.id,
                        cause,
                      }),
                  ),
                )).thread;
              }
              return providerThreadFromCodexThread({
                appThreadId: threadInput.targetThreadId,
                idAllocator,
                ownerNodeId: threadInput.ownerNodeId ?? null,
                providerSessionId: input.providerSessionId,
                providerInstanceId: adapterOptions.instanceId,
                thread: forkedThread,
                forkedFrom: {
                  providerThreadId: threadInput.sourceProviderThread.id,
                  ...(threadInput.providerTurnId === undefined
                    ? {}
                    : { providerTurnId: threadInput.providerTurnId }),
                },
              });
            },
            (effect, threadInput) =>
              effect.pipe(
                Effect.mapError(
                  (cause) =>
                    new ProviderAdapter.ProviderAdapterForkThreadError({
                      driver: CODEX_PROVIDER,
                      providerThreadId: threadInput.sourceProviderThread.id,
                      cause,
                    }),
                ),
              ),
          ),
        };
        return runtime;
      },
      (effect, input) =>
        effect.pipe(
          Effect.mapError(
            (cause) =>
              new ProviderAdapter.ProviderAdapterOpenSessionError({
                driver: CODEX_PROVIDER,
                providerSessionId: input.providerSessionId,
                cause,
              }),
          ),
        ),
    ),
  });
});