import {
  ApprovalRequestId,
  DEFAULT_MODEL,
  EventId,
  ProviderDriverKind,
  ProviderItemId,
  type ProviderInstanceId,
  type ProviderApprovalDecision,
  type ProviderApprovalOption,
  type ProviderEvent,
  type ProviderInteractionMode,
  type ProviderRequestKind,
  type ProviderSession,
  type ProviderTurnStartResult,
  type ProviderUserInputAnswers,
  RuntimeMode,
  ThreadId,
  TurnId,
} from "@t3tools/contracts";
import { resolveSpawnCommand } from "@t3tools/shared/shell";
import { normalizeModelSlug } from "@t3tools/shared/model";
import * as Crypto from "effect/Crypto";
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 Layer from "effect/Layer";
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 Stream from "effect/Stream";
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
import * as CodexClient from "effect-codex-app-server/client";
import * as CodexErrors from "effect-codex-app-server/errors";
import * as CodexRpc from "effect-codex-app-server/rpc";
import * as EffectCodexSchema from "effect-codex-app-server/schema";

import { buildCodexInitializeParams } from "./CodexProvider.ts";
import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts";
import { expandHomePath } from "../../pathExpansion.ts";
import {
  buildCodexDeveloperInstructions,
  type T3CodeToolAvailability,
} from "../CodexDeveloperInstructions.ts";
const decodeV2TurnStartResponse = Schema.decodeUnknownEffect(EffectCodexSchema.V2TurnStartResponse);

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

const ANSI_ESCAPE_CHAR = String.fromCharCode(27);
const ANSI_ESCAPE_REGEX = new RegExp(`${ANSI_ESCAPE_CHAR}\\[[0-9;]*m`, "g");
const CODEX_STDERR_LOG_REGEX =
  /^\d{4}-\d{2}-\d{2}T\S+\s+(TRACE|DEBUG|INFO|WARN|ERROR)\s+\S+:\s+(.*)$/;
const BENIGN_ERROR_LOG_SNIPPETS = [
  "state db missing rollout path for thread",
  "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back",
];
const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const;
const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [
  "not found",
  "missing thread",
  "no such thread",
  "unknown thread",
  "does not exist",
  "no rollout found",
];

export function hasConfiguredMcpServer(appServerArgs: ReadonlyArray<string> | undefined): boolean {
  return appServerArgs?.some((argument) => argument.includes("mcp_servers.")) === true;
}

function configuredMcpToolAvailability(
  appServerArgs: ReadonlyArray<string> | undefined,
  mcpCapabilities: ReadonlySet<string> | undefined,
): T3CodeToolAvailability {
  if (!hasConfiguredMcpServer(appServerArgs)) return { browser: false, device: false };
  // Callers predating the capability set attached the browser toolkit only.
  if (mcpCapabilities === undefined) return { browser: true, device: false };
  return { browser: mcpCapabilities.has("preview"), device: mcpCapabilities.has("device") };
}

export const CodexResumeCursorSchema = Schema.Struct({
  threadId: Schema.String,
});
const CodexUserInputAnswerObject = Schema.Struct({
  answers: Schema.Array(Schema.String),
});
const isCodexResumeCursorSchema = Schema.is(CodexResumeCursorSchema);
const isCodexUserInputAnswerObject = Schema.is(CodexUserInputAnswerObject);
const NullableMcpElicitationString = Schema.NullOr(Schema.String);
const McpElicitationMetadata = Schema.Struct({
  app: Schema.optionalKey(NullableMcpElicitationString),
  app_name: Schema.optionalKey(NullableMcpElicitationString),
  appName: Schema.optionalKey(NullableMcpElicitationString),
  connector_name: Schema.optionalKey(NullableMcpElicitationString),
  connectorName: Schema.optionalKey(NullableMcpElicitationString),
  allowPersistentApproval: Schema.optionalKey(Schema.NullOr(Schema.Boolean)),
  persist: Schema.optionalKey(
    Schema.NullOr(Schema.Union([Schema.String, Schema.Array(Schema.String)])),
  ),
  target: Schema.optionalKey(
    Schema.NullOr(
      Schema.Struct({
        app: Schema.optionalKey(NullableMcpElicitationString),
        name: Schema.optionalKey(NullableMcpElicitationString),
      }),
    ),
  ),
  tool_params: Schema.optionalKey(
    Schema.NullOr(
      Schema.Struct({
        app: Schema.optionalKey(NullableMcpElicitationString),
        app_name: Schema.optionalKey(NullableMcpElicitationString),
      }),
    ),
  ),
});
const McpElicitationFormField = Schema.Struct({
  type: Schema.optionalKey(NullableMcpElicitationString),
  title: Schema.optionalKey(NullableMcpElicitationString),
  description: Schema.optionalKey(NullableMcpElicitationString),
  default: Schema.optionalKey(Schema.Unknown),
  enum: Schema.optionalKey(Schema.NullOr(Schema.Array(Schema.String))),
  enumNames: Schema.optionalKey(Schema.NullOr(Schema.Array(Schema.String))),
  oneOf: Schema.optionalKey(
    Schema.NullOr(
      Schema.Array(
        Schema.Struct({
          const: Schema.String,
          title: Schema.optionalKey(NullableMcpElicitationString),
        }),
      ),
    ),
  ),
});
const McpElicitationForm = Schema.Struct({
  properties: Schema.optionalKey(Schema.Record(Schema.String, McpElicitationFormField)),
  required: Schema.optionalKey(Schema.NullOr(Schema.Array(Schema.String))),
});
const isMcpElicitationMetadata = Schema.is(McpElicitationMetadata);
const isMcpElicitationForm = Schema.is(McpElicitationForm);

// TODO: Verify `packages/effect-codex-app-server/scripts/generate.ts` so the generated
// `V2TurnStartParams` schema includes `collaborationMode` directly.
const CodexTurnStartParamsWithCollaborationMode = EffectCodexSchema.V2TurnStartParams.pipe(
  Schema.fieldsAssign({
    collaborationMode: Schema.optionalKey(EffectCodexSchema.V2TurnStartParams__CollaborationMode),
  }),
);
const decodeCodexTurnStartParamsWithCollaborationMode = Schema.decodeUnknownEffect(
  CodexTurnStartParamsWithCollaborationMode,
);
const CodexChildResumeMetadata = Schema.Struct({
  thread: Schema.Struct({ id: Schema.String }),
  model: Schema.String,
  reasoningEffort: Schema.optionalKey(Schema.NullOr(Schema.String)),
});
const decodeCodexChildResumeMetadata = Schema.decodeUnknownEffect(CodexChildResumeMetadata);

export type CodexTurnStartParamsWithCollaborationMode =
  typeof CodexTurnStartParamsWithCollaborationMode.Type;

export type CodexResumeCursor = typeof CodexResumeCursorSchema.Type;
type CodexServiceTier = NonNullable<EffectCodexSchema.V2ThreadStartParams["serviceTier"]>;
type CodexThreadItem =
  | EffectCodexSchema.V2ThreadReadResponse["thread"]["turns"][number]["items"][number]
  | EffectCodexSchema.V2ThreadRollbackResponse["thread"]["turns"][number]["items"][number];

export interface CodexSessionRuntimeOptions {
  readonly threadId: ThreadId;
  readonly providerInstanceId?: ProviderInstanceId;
  readonly binaryPath: string;
  readonly homePath?: string;
  readonly launchArgs?: string;
  readonly environment?: NodeJS.ProcessEnv;
  readonly cwd: string;
  readonly runtimeMode: RuntimeMode;
  readonly model?: string;
  readonly serviceTier?: CodexServiceTier | undefined;
  readonly resumeCursor?: CodexResumeCursor;
  readonly appServerArgs?: ReadonlyArray<string>;
  /** Capabilities the session's `t3-code` MCP credential grants; drives the prompt blocks. */
  readonly mcpCapabilities?: ReadonlySet<string>;
}

export interface CodexSessionRuntimeSendTurnInput {
  readonly input?: string;
  readonly attachments?: ReadonlyArray<{
    readonly type: "image";
    readonly url: string;
  }>;
  readonly model?: string;
  readonly serviceTier?: CodexServiceTier | undefined;
  readonly effort?: EffectCodexSchema.V2TurnStartParams__ReasoningEffort | undefined;
  readonly interactionMode?: ProviderInteractionMode;
}

export interface CodexThreadTurnSnapshot {
  readonly id: TurnId;
  readonly items: ReadonlyArray<CodexThreadItem>;
}

export interface CodexThreadSnapshot {
  readonly threadId: string;
  readonly turns: ReadonlyArray<CodexThreadTurnSnapshot>;
}

export interface CodexSessionRuntimeShape {
  readonly start: () => Effect.Effect<ProviderSession, CodexSessionRuntimeError>;
  readonly getSession: Effect.Effect<ProviderSession>;
  readonly sendTurn: (
    input: CodexSessionRuntimeSendTurnInput,
  ) => Effect.Effect<ProviderTurnStartResult, CodexSessionRuntimeError>;
  readonly compactThread: Effect.Effect<void, CodexSessionRuntimeError>;
  readonly interruptTurn: (turnId?: TurnId) => Effect.Effect<void, CodexSessionRuntimeError>;
  readonly readThread: Effect.Effect<CodexThreadSnapshot, CodexSessionRuntimeError>;
  readonly rollbackThread: (
    numTurns: number,
  ) => Effect.Effect<CodexThreadSnapshot, CodexSessionRuntimeError>;
  readonly uploadFeedback: (
    reason?: string,
  ) => Effect.Effect<EffectCodexSchema.V2FeedbackUploadResponse, CodexSessionRuntimeError>;
  readonly respondToRequest: (
    requestId: ApprovalRequestId,
    decision: ProviderApprovalDecision,
  ) => Effect.Effect<void, CodexSessionRuntimeError>;
  readonly respondToUserInput: (
    requestId: ApprovalRequestId,
    answers: ProviderUserInputAnswers,
  ) => Effect.Effect<void, CodexSessionRuntimeError>;
  readonly events: Stream.Stream<ProviderEvent, never>;
  readonly close: Effect.Effect<void>;
}

export type CodexSessionRuntimeError =
  | CodexErrors.CodexAppServerError
  | CodexSessionRuntimePendingApprovalNotFoundError
  | CodexSessionRuntimePendingUserInputNotFoundError
  | CodexSessionRuntimeInvalidUserInputAnswersError
  | CodexSessionRuntimeThreadIdMissingError;

export class CodexSessionRuntimePendingApprovalNotFoundError extends Schema.TaggedError<CodexSessionRuntimePendingApprovalNotFoundError>()(
  "CodexSessionRuntimePendingApprovalNotFoundError",
  {
    requestId: Schema.String,
  },
) {
  override get message(): string {
    return `Unknown pending Codex approval request: ${this.requestId}`;
  }
}

export class CodexSessionRuntimePendingUserInputNotFoundError extends Schema.TaggedError<CodexSessionRuntimePendingUserInputNotFoundError>()(
  "CodexSessionRuntimePendingUserInputNotFoundError",
  {
    requestId: Schema.String,
  },
) {
  override get message(): string {
    return `Unknown pending Codex user input request: ${this.requestId}`;
  }
}

export class CodexSessionRuntimeInvalidUserInputAnswersError extends Schema.TaggedError<CodexSessionRuntimeInvalidUserInputAnswersError>()(
  "CodexSessionRuntimeInvalidUserInputAnswersError",
  {
    questionId: Schema.String,
  },
) {
  override get message(): string {
    return `Invalid Codex user input answers for question '${this.questionId}'`;
  }
}

export class CodexSessionRuntimeThreadIdMissingError extends Schema.TaggedError<CodexSessionRuntimeThreadIdMissingError>()(
  "CodexSessionRuntimeThreadIdMissingError",
  {
    threadId: Schema.String,
  },
) {
  override get message(): string {
    return `Codex session is missing a provider thread id for ${this.threadId}`;
  }
}

interface PendingApproval {
  readonly requestId: ApprovalRequestId;
  readonly jsonRpcId: string;
  readonly requestKind: ProviderRequestKind;
  readonly turnId: TurnId | undefined;
  readonly itemId: ProviderItemId | undefined;
  readonly decision: Deferred.Deferred<ProviderApprovalDecision>;
}

interface ApprovalCorrelation {
  readonly requestId: ApprovalRequestId;
  readonly requestKind: ProviderRequestKind;
  readonly turnId: TurnId | undefined;
  readonly itemId: ProviderItemId | undefined;
}

interface PendingUserInput {
  readonly requestId: ApprovalRequestId;
  readonly turnId: TurnId | undefined;
  readonly itemId: ProviderItemId | undefined;
  readonly answers: Deferred.Deferred<ProviderUserInputAnswers>;
}

type McpElicitationPersistenceDecision = Extract<
  ProviderApprovalDecision,
  "acceptForSession" | "acceptAlways"
>;

function mcpElicitationPersistenceDecision(
  value: string,
): McpElicitationPersistenceDecision | null {
  const normalized = value.toLowerCase();
  if (normalized.includes("session")) return "acceptForSession";
  if (
    normalized.includes("always") ||
    normalized.includes("permanent") ||
    normalized.includes("forever") ||
    normalized.includes("persistent")
  ) {
    return "acceptAlways";
  }
  return null;
}

function mcpElicitationFormFields(payload: EffectCodexSchema.McpServerElicitationRequestParams) {
  if (payload.mode === "url" || !isMcpElicitationForm(payload.requestedSchema)) {
    return undefined;
  }
  return payload.requestedSchema;
}

function mcpElicitationFieldOptions(field: typeof McpElicitationFormField.Type) {
  if (field.oneOf) {
    return field.oneOf.map((option) => ({ value: option.const, label: option.title }));
  }
  return (field.enum ?? []).map((value, index) => ({
    value,
    label: field.enumNames?.[index],
  }));
}

function isMcpElicitationPersistenceField(
  key: string,
  field: typeof McpElicitationFormField.Type,
): boolean {
  return (
    mcpElicitationPersistenceDecision(key) !== null ||
    key.toLowerCase() === "persist" ||
    mcpElicitationPersistenceDecision(field.title ?? "") !== null ||
    mcpElicitationPersistenceDecision(field.description ?? "") !== null
  );
}

/** Returns the app and approval choices advertised by an MCP elicitation. */
export function describeMcpElicitation(
  payload: EffectCodexSchema.McpServerElicitationRequestParams,
): { readonly appName: string; readonly options: ReadonlyArray<ProviderApprovalOption> } {
  const metadata = isMcpElicitationMetadata(payload._meta) ? payload._meta : undefined;
  const appName =
    metadata?.app_name ??
    metadata?.appName ??
    metadata?.app ??
    metadata?.target?.app ??
    metadata?.target?.name ??
    metadata?.tool_params?.app_name ??
    metadata?.tool_params?.app ??
    payload.message.match(/^Allow ChatGPT to use (.+?)\?$/i)?.[1] ??
    metadata?.connector_name ??
    metadata?.connectorName ??
    payload.serverName;
  const persistenceOptions = new Map<McpElicitationPersistenceDecision, string>();
  const persist = metadata?.persist;
  for (const value of typeof persist === "string" ? [persist] : (persist ?? [])) {
    const decision = mcpElicitationPersistenceDecision(value);
    if (decision) persistenceOptions.set(decision, "");
  }
  if (metadata?.allowPersistentApproval) {
    persistenceOptions.set("acceptAlways", "");
  }

  const form = mcpElicitationFormFields(payload);
  for (const [key, field] of Object.entries(form?.properties ?? {})) {
    for (const option of mcpElicitationFieldOptions(field)) {
      const decision = mcpElicitationPersistenceDecision(option.value);
      if (decision) persistenceOptions.set(decision, option.label ?? "");
    }
    if (field.type === "boolean" && isMcpElicitationPersistenceField(key, field)) {
      persistenceOptions.set("acceptAlways", field.title ?? "");
    }
  }

  return {
    appName,
    options: [
      { decision: "cancel", label: "Cancel" },
      { decision: "decline", label: "Decline" },
      ...(persistenceOptions.has("acceptForSession") &&
      toMcpElicitationResponse(payload, "acceptForSession").action === "accept"
        ? [
            {
              decision: "acceptForSession" as const,
              label: persistenceOptions.get("acceptForSession") || "Always allow this session",
            },
          ]
        : []),
      ...(persistenceOptions.has("acceptAlways") &&
      toMcpElicitationResponse(payload, "acceptAlways").action === "accept"
        ? [
            {
              decision: "acceptAlways" as const,
              label: persistenceOptions.get("acceptAlways") || "Always allow",
            },
          ]
        : []),
      { decision: "accept", label: "Approve" },
    ],
  };
}

/** Converts a T3 approval decision into the MCP elicitation wire response. */
export function toMcpElicitationResponse(
  payload: EffectCodexSchema.McpServerElicitationRequestParams,
  decision: ProviderApprovalDecision,
): EffectCodexSchema.McpServerElicitationRequestResponse {
  if (decision === "decline" || decision === "cancel") {
    return { action: decision };
  }

  if (payload.mode === "url") {
    return { action: "decline" };
  }

  const persist =
    decision === "acceptForSession"
      ? "session"
      : decision === "acceptAlways"
        ? "always"
        : undefined;
  const form = mcpElicitationFormFields(payload);
  const content: Record<string, unknown> = {};

  for (const [key, field] of Object.entries(form?.properties ?? {})) {
    const options = mcpElicitationFieldOptions(field);
    const chosenOption = options.find((option) =>
      persist
        ? mcpElicitationPersistenceDecision(option.value) === decision
        : /once|accept|approve|allow/i.test(option.value) &&
          mcpElicitationPersistenceDecision(option.value) === null,
    );
    if (chosenOption) {
      content[key] = chosenOption.value;
    } else if (field.type === "boolean" && isMcpElicitationPersistenceField(key, field)) {
      content[key] = decision === "acceptAlways";
    } else if (field.default !== undefined && field.default !== null) {
      content[key] = field.default;
    }
  }

  if (form?.required?.some((key) => !Object.hasOwn(content, key))) {
    return { action: "decline" };
  }

  return {
    action: "accept",
    ...(persist ? { _meta: { persist } } : {}),
    ...(form ? { content } : {}),
  };
}

type CodexServerNotification = {
  readonly [M in CodexRpc.ServerNotificationMethod]: {
    readonly method: M;
    readonly params: CodexRpc.ServerNotificationParamsByMethod[M];
  };
}[CodexRpc.ServerNotificationMethod];

function makeCodexServerNotification<M extends CodexRpc.ServerNotificationMethod>(
  method: M,
  params: CodexRpc.ServerNotificationParamsByMethod[M],
): CodexServerNotification {
  return { method, params } as CodexServerNotification;
}

function normalizeCodexModelSlug(
  model: string | undefined | null,
  preferredId?: string,
): string | undefined {
  const normalized = normalizeModelSlug(model);
  if (!normalized) {
    return undefined;
  }
  if (preferredId?.endsWith("-codex") && preferredId !== normalized) {
    return preferredId;
  }
  return normalized;
}

function readResumeCursorThreadId(
  resumeCursor: ProviderSession["resumeCursor"],
): string | undefined {
  return isCodexResumeCursorSchema(resumeCursor) ? resumeCursor.threadId : undefined;
}

function runtimeModeToThreadConfig(input: RuntimeMode): {
  readonly approvalPolicy: EffectCodexSchema.V2ThreadStartParams__AskForApproval;
  readonly sandbox: EffectCodexSchema.V2ThreadStartParams__SandboxMode;
  // Always explicit: omitting the field on resume keeps the thread's previous
  // reviewer, which would leave auto_review sticky after switching modes.
  readonly approvalsReviewer: EffectCodexSchema.V2ThreadStartParams__ApprovalsReviewer;
} {
  switch (input) {
    case "approval-required":
      return {
        approvalPolicy: "untrusted",
        sandbox: "read-only",
        approvalsReviewer: "user",
      };
    case "auto-accept-edits":
      return {
        approvalPolicy: "on-request",
        sandbox: "workspace-write",
        approvalsReviewer: "user",
      };
    case "auto":
      return {
        approvalPolicy: "on-request",
        sandbox: "workspace-write",
        approvalsReviewer: "auto_review",
      };
    case "full-access":
    default:
      return {
        approvalPolicy: "never",
        sandbox: "danger-full-access",
        approvalsReviewer: "user",
      };
  }
}

function buildThreadStartParams(input: {
  readonly cwd: string;
  readonly runtimeMode: RuntimeMode;
  readonly model: string | undefined;
  readonly serviceTier: CodexServiceTier | undefined;
}): EffectCodexSchema.V2ThreadStartParams {
  const config = runtimeModeToThreadConfig(input.runtimeMode);
  return {
    cwd: input.cwd,
    approvalPolicy: config.approvalPolicy,
    sandbox: config.sandbox,
    approvalsReviewer: config.approvalsReviewer,
    ...(input.model ? { model: input.model } : {}),
    ...(input.serviceTier ? { serviceTier: input.serviceTier } : {}),
  };
}

function runtimeModeToTurnSandboxPolicy(
  input: RuntimeMode,
): EffectCodexSchema.V2TurnStartParams__SandboxPolicy {
  switch (input) {
    case "approval-required":
      return {
        type: "readOnly",
      };
    case "auto-accept-edits":
    case "auto":
      return {
        type: "workspaceWrite",
      };
    case "full-access":
    default:
      return {
        type: "dangerFullAccess",
      };
  }
}

function buildCodexCollaborationMode(input: {
  readonly interactionMode?: ProviderInteractionMode;
  readonly model?: string;
  readonly effort?: EffectCodexSchema.V2TurnStartParams__ReasoningEffort;
  readonly browserToolsAvailable?: boolean | T3CodeToolAvailability;
}): EffectCodexSchema.V2TurnStartParams__CollaborationMode | undefined {
  if (input.interactionMode === undefined) {
    return undefined;
  }
  const model = normalizeCodexModelSlug(input.model) ?? DEFAULT_MODEL;
  const reasoningEffort = input.effort ?? "medium";
  return {
    mode: input.interactionMode,
    settings: {
      model,
      reasoning_effort: reasoningEffort,
      developer_instructions: buildCodexDeveloperInstructions(
        input.interactionMode,
        { model, reasoningEffort },
        input.browserToolsAvailable ?? true,
      ),
    },
  };
}

export function buildTurnStartParams(input: {
  readonly threadId: string;
  readonly runtimeMode: RuntimeMode;
  readonly prompt?: string;
  readonly attachments?: ReadonlyArray<{
    readonly type: "image";
    readonly url: string;
  }>;
  readonly model?: string;
  readonly serviceTier?: CodexServiceTier;
  readonly effort?: EffectCodexSchema.V2TurnStartParams__ReasoningEffort;
  readonly interactionMode?: ProviderInteractionMode;
  /** Defaults to true so callers that predate the agent-access gate are unchanged. */
  readonly browserToolsAvailable?: boolean | T3CodeToolAvailability;
}): Effect.Effect<
  CodexTurnStartParamsWithCollaborationMode,
  CodexErrors.CodexAppServerProtocolParseError
> {
  const turnInput: Array<EffectCodexSchema.V2TurnStartParams__UserInput> = [];
  if (input.prompt) {
    turnInput.push({
      type: "text",
      text: input.prompt,
    });
  }
  for (const attachment of input.attachments ?? []) {
    turnInput.push(attachment);
  }

  const config = runtimeModeToThreadConfig(input.runtimeMode);
  const collaborationMode = buildCodexCollaborationMode({
    ...(input.interactionMode ? { interactionMode: input.interactionMode } : {}),
    ...(input.model ? { model: input.model } : {}),
    ...(input.effort ? { effort: input.effort } : {}),
    browserToolsAvailable: input.browserToolsAvailable ?? true,
  });

  return decodeCodexTurnStartParamsWithCollaborationMode({
    threadId: input.threadId,
    input: turnInput,
    approvalPolicy: config.approvalPolicy,
    approvalsReviewer: config.approvalsReviewer,
    sandboxPolicy: runtimeModeToTurnSandboxPolicy(input.runtimeMode),
    ...(input.model ? { model: input.model } : {}),
    ...(input.serviceTier ? { serviceTier: input.serviceTier } : {}),
    ...(input.effort ? { effort: input.effort } : {}),
    ...(collaborationMode ? { collaborationMode } : {}),
  }).pipe(
    Effect.mapError((cause) =>
      CodexErrors.CodexAppServerProtocolParseError.fromSchemaError(
        "decode-request-payload",
        cause,
        { method: "turn/start" },
      ),
    ),
  );
}

function classifyCodexStderrLine(rawLine: string): { readonly message: string } | null {
  const line = rawLine.replaceAll(ANSI_ESCAPE_REGEX, "").trim();
  if (!line) {
    return null;
  }

  const match = line.match(CODEX_STDERR_LOG_REGEX);
  if (match) {
    const level = match[1];
    if (level && level !== "ERROR") {
      return null;
    }
    if (BENIGN_ERROR_LOG_SNIPPETS.some((snippet) => line.includes(snippet))) {
      return null;
    }
  }

  return { message: line };
}

export function isRecoverableThreadResumeError(error: unknown): boolean {
  const message = (error instanceof Error ? error.message : String(error)).toLowerCase();
  if (!message.includes("thread")) {
    return false;
  }
  return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet));
}

const CodexThreadResumeMetadata = Schema.Struct({
  cwd: Schema.String,
  model: Schema.String,
  thread: Schema.Struct({ id: Schema.String }),
});
const decodeCodexThreadResumeMetadata = Schema.decodeUnknownEffect(CodexThreadResumeMetadata);

interface CodexThreadOpenClient {
  readonly raw: {
    readonly request: (
      method: "thread/resume",
      payload: CodexRpc.ClientRequestParamsByMethod["thread/resume"] & {
        readonly excludeTurns?: boolean;
      },
    ) => Effect.Effect<unknown, CodexErrors.CodexAppServerError>;
  };
  readonly request: (
    method: "thread/start",
    payload: CodexRpc.ClientRequestParamsByMethod["thread/start"],
  ) => Effect.Effect<
    CodexRpc.ClientRequestResponsesByMethod["thread/start"],
    CodexErrors.CodexAppServerError
  >;
}

export const openCodexThread = (input: {
  readonly client: CodexThreadOpenClient;
  readonly threadId: ThreadId;
  readonly runtimeMode: RuntimeMode;
  readonly cwd: string;
  readonly requestedModel: string | undefined;
  readonly serviceTier: CodexServiceTier | undefined;
  readonly resumeThreadId: string | undefined;
}): Effect.Effect<typeof CodexThreadResumeMetadata.Type, CodexErrors.CodexAppServerError> => {
  const resumeThreadId = input.resumeThreadId;
  const startParams = buildThreadStartParams({
    cwd: input.cwd,
    runtimeMode: input.runtimeMode,
    model: input.requestedModel,
    serviceTier: input.serviceTier,
  });

  if (resumeThreadId === undefined) {
    return input.client.request("thread/start", startParams);
  }

  // Older providers may still return history despite excludeTurns. Only the
  // session metadata is needed here, so unrelated historical items cannot
  // prevent resuming a valid provider thread.
  return input.client.raw
    .request("thread/resume", {
      threadId: resumeThreadId,
      ...startParams,
      excludeTurns: true,
    })
    .pipe(
      Effect.flatMap((response) =>
        decodeCodexThreadResumeMetadata(response).pipe(
          Effect.mapError((error) =>
            CodexErrors.CodexAppServerRequestError.invalidPayload(
              "thread/resume",
              "decode-payload",
              error,
            ),
          ),
        ),
      ),
      Effect.catchIf(isRecoverableThreadResumeError, (error) =>
        Effect.logWarning("codex app-server thread resume fell back to fresh start", {
          threadId: input.threadId,
          requestedRuntimeMode: input.runtimeMode,
          resumeThreadId,
          recoverable: true,
          cause: error,
        }).pipe(Effect.andThen(input.client.request("thread/start", startParams))),
      ),
    );
};

function readNotificationThreadId(notification: CodexServerNotification): string | undefined {
  switch (notification.method) {
    case "thread/started":
      return notification.params.thread.id;
    case "error":
    case "thread/status/changed":
    case "thread/archived":
    case "thread/unarchived":
    case "thread/closed":
    case "thread/name/updated":
    case "thread/settings/updated":
    case "thread/tokenUsage/updated":
    case "model/rerouted":
    case "turn/started":
    case "hook/started":
    case "turn/completed":
    case "hook/completed":
    case "turn/diff/updated":
    case "turn/plan/updated":
    case "item/started":
    case "item/autoApprovalReview/started":
    case "item/autoApprovalReview/completed":
    case "item/completed":
    case "rawResponseItem/completed":
    case "item/agentMessage/delta":
    case "item/plan/delta":
    case "item/commandExecution/outputDelta":
    case "item/commandExecution/terminalInteraction":
    case "item/fileChange/outputDelta":
    case "item/fileChange/patchUpdated":
    case "serverRequest/resolved":
    case "item/mcpToolCall/progress":
    case "item/reasoning/summaryTextDelta":
    case "item/reasoning/summaryPartAdded":
    case "item/reasoning/textDelta":
    case "thread/compacted":
    case "thread/realtime/started":
    case "thread/realtime/itemAdded":
    case "thread/realtime/transcript/delta":
    case "thread/realtime/transcript/done":
    case "thread/realtime/outputAudio/delta":
    case "thread/realtime/sdp":
    case "thread/realtime/error":
    case "thread/realtime/closed":
      return notification.params.threadId;
    default:
      return undefined;
  }
}

export function makeMemoryConsolidationNotificationFilter(): (
  notification: CodexServerNotification,
) => boolean {
  const threadIds = new Set<string>();

  return (notification) => {
    if (notification.method === "thread/started") {
      const thread = notification.params.thread;
      const source = thread.source;
      if (
        thread.threadSource === "memory_consolidation" ||
        (typeof source === "object" &&
          source !== null &&
          "subAgent" in source &&
          source.subAgent === "memory_consolidation")
      ) {
        threadIds.add(thread.id);
        return true;
      }
    }

    const params = notification.params;
    const threadId =
      notification.method === "thread/started"
        ? notification.params.thread.id
        : "threadId" in params && typeof params.threadId === "string"
          ? params.threadId
          : undefined;
    if (!threadId || !threadIds.has(threadId)) {
      return false;
    }

    if (notification.method === "serverRequest/resolved") {
      return false;
    }

    if (notification.method === "thread/closed") {
      threadIds.delete(threadId);
    }
    return true;
  };
}

function readRouteFields(notification: CodexServerNotification): {
  readonly turnId: TurnId | undefined;
  readonly itemId: ProviderItemId | undefined;
} {
  switch (notification.method) {
    case "thread/started":
      return {
        turnId: undefined,
        itemId: undefined,
      };
    case "turn/started":
    case "turn/completed":
      return {
        turnId: TurnId.make(notification.params.turn.id),
        itemId: undefined,
      };
    case "error":
      return {
        turnId: TurnId.make(notification.params.turnId),
        itemId: undefined,
      };
    case "turn/diff/updated":
    case "turn/plan/updated":
      return {
        turnId: TurnId.make(notification.params.turnId),
        itemId: undefined,
      };
    case "serverRequest/resolved":
      return {
        turnId: undefined,
        itemId: undefined,
      };
    case "item/started":
    case "item/completed":
      return {
        turnId: TurnId.make(notification.params.turnId),
        itemId: ProviderItemId.make(notification.params.item.id),
      };
    case "item/agentMessage/delta":
    case "item/plan/delta":
    case "item/commandExecution/outputDelta":
    case "item/commandExecution/terminalInteraction":
    case "item/fileChange/outputDelta":
    case "item/fileChange/patchUpdated":
    case "item/reasoning/summaryTextDelta":
    case "item/reasoning/summaryPartAdded":
    case "item/reasoning/textDelta":
      return {
        turnId: TurnId.make(notification.params.turnId),
        itemId: ProviderItemId.make(notification.params.itemId),
      };
    default:
      return {
        turnId: undefined,
        itemId: undefined,
      };
  }
}

/**
 * Native collab child-agent tracking (multi-agent v2). Under v2 subagents are
 * full app-server threads: identity arrives on `thread/started` with
 * source.subAgent.thread_spawn, lifecycle on `subAgentActivity` items and the
 * child thread's own turn/status/tokenUsage notifications. The runtime
 * registers children from those explicit signals, intercepts their
 * notifications before parent-timeline mapping, and re-emits them as
 * synthetic `collabAgent/*` provider events the adapter turns into task.*
 * runtime events (timelineBypass keeps them out of the parent chat).
 *
 * WIP, probe-gated: registration is deliberately explicit-signals-only. The
 * spec's "provisionally treat unknown foreign thread ids as v2 children" rule
 * needs a live wire capture of the packaged binary before it lands — blind
 * capture risks eating unrelated traffic. Until then a child whose first
 * notification precedes registration passes through as today (no regression
 * vs main, which passes everything through).
 */
interface CollabChildAgentState {
  readonly agentThreadId: string;
  readonly nickname: string | undefined;
  readonly role: string | undefined;
  readonly agentPath: string | undefined;
  readonly depth: number | undefined;
  readonly parentThreadId: string | undefined;
  /**
   * Parent canonical turn active when the child registered. Stamped on every
   * synthetic collabAgent/* event so clients can batch a fleet by its spawn
   * turn — without it, separate fleets in one thread collapsed into a single
   * "direct:no-turn" CTA (review finding).
   */
  readonly spawnTurnId: TurnId | undefined;
}

interface CollabChildMetadataState {
  readonly model: string | undefined;
  readonly effort: string | undefined;
  readonly lookupStarted: boolean;
  readonly closed: boolean;
}

function collabChildIdentity(
  child: CollabChildAgentState,
  metadata: CollabChildMetadataState | undefined,
) {
  return {
    agentThreadId: child.agentThreadId,
    ...(child.nickname ? { nickname: child.nickname } : {}),
    ...(child.role ? { role: child.role } : {}),
    ...(child.agentPath ? { agentPath: child.agentPath } : {}),
    ...(metadata?.model ? { model: metadata.model } : {}),
    ...(metadata?.effort ? { effort: metadata.effort } : {}),
  };
}

function nonEmptyMetadataValue(value: unknown): string | undefined {
  if (typeof value !== "string") {
    return undefined;
  }
  const trimmed = value.trim();
  return trimmed.length > 0 ? trimmed : undefined;
}

function readThreadSpawnSource(thread: { readonly source: unknown }):
  | {
      nickname: string | undefined;
      role: string | undefined;
      agentPath: string | undefined;
      depth: number | undefined;
      parentThreadId: string | undefined;
    }
  | undefined {
  const source = thread.source;
  if (typeof source !== "object" || source === null || !("subAgent" in source)) {
    return undefined;
  }
  const subAgent = (source as { subAgent: unknown }).subAgent;
  if (typeof subAgent !== "object" || subAgent === null || !("thread_spawn" in subAgent)) {
    return undefined;
  }
  const spawn = (subAgent as { thread_spawn: unknown }).thread_spawn;
  if (typeof spawn !== "object" || spawn === null) {
    return undefined;
  }
  const record = spawn as Record<string, unknown>;
  return {
    nickname: typeof record.agent_nickname === "string" ? record.agent_nickname : undefined,
    role: typeof record.agent_role === "string" ? record.agent_role : undefined,
    agentPath: typeof record.agent_path === "string" ? record.agent_path : undefined,
    depth: typeof record.depth === "number" ? record.depth : undefined,
    parentThreadId:
      typeof record.parent_thread_id === "string" ? record.parent_thread_id : undefined,
  };
}

function rememberCollabReceiverTurns(
  collabReceiverTurns: Map<string, TurnId>,
  notification: CodexServerNotification,
  parentTurnId: TurnId | undefined,
): void {
  if (!parentTurnId) {
    return;
  }

  if (notification.method !== "item/started" && notification.method !== "item/completed") {
    return;
  }

  if (notification.params.item.type !== "collabAgentToolCall") {
    return;
  }

  for (const receiverThreadId of notification.params.item.receiverThreadIds) {
    collabReceiverTurns.set(receiverThreadId, parentTurnId);
  }
}

function shouldSuppressChildConversationNotification(
  method: CodexRpc.ServerNotificationMethod,
): boolean {
  return (
    method === "thread/started" ||
    method === "thread/status/changed" ||
    method === "thread/archived" ||
    method === "thread/unarchived" ||
    method === "thread/closed" ||
    method === "thread/compacted" ||
    method === "thread/name/updated" ||
    method === "thread/settings/updated" ||
    method === "thread/tokenUsage/updated" ||
    method === "model/rerouted" ||
    method === "turn/started" ||
    method === "turn/completed" ||
    method === "turn/plan/updated" ||
    method === "item/plan/delta"
  );
}

/**
 * How a notification addressed to a REGISTERED child thread is handled.
 *
 * Exported and pure so the routing table can be asserted against captured
 * wire traces (see codexMultiAgentWire.json) rather than only read.
 *
 * - "agent-event": map to a synthetic collabAgent/* event (Agents surface).
 * - "parent": pass through to the parent path — it carries state the parent
 *   still owns (approval correlation cleanup).
 * - "drop": genuine child chatter with no parent meaning (deltas, name and
 *   plan updates).
 *
 * Default is "drop" ONLY for the enumerated chatter; anything unrecognized
 * routes to "parent" so new wire methods surface instead of vanishing
 * (two shipped bugs came from a catch-all that swallowed everything).
 */
export type CodexChildNotificationRoute = "agent-event" | "parent" | "drop";

const CHILD_AGENT_EVENT_METHODS: ReadonlySet<string> = new Set([
  "turn/started",
  "turn/completed",
  "thread/status/changed",
  "thread/tokenUsage/updated",
  "thread/settings/updated",
  "model/rerouted",
  "item/started",
  "item/completed",
  "thread/closed",
  "error",
]);

const CHILD_CHATTER_METHODS: ReadonlySet<string> = new Set([
  "item/agentMessage/delta",
  "item/reasoning/textDelta",
  "item/reasoning/summaryTextDelta",
  "item/reasoning/summaryPartAdded",
  "item/commandExecution/outputDelta",
  "item/fileChange/outputDelta",
  "item/fileChange/patchUpdated",
  "item/plan/delta",
  "turn/plan/updated",
  "turn/diff/updated",
  "thread/name/updated",
  "rawResponseItem/completed",
  // Child-owned thread lifecycle: the parent adapter maps these onto the
  // PARENT thread (archived/compacted state), so a child compacting would
  // rewrite the parent. Mirrors the v1 suppressor list — dropping them is
  // the pre-existing behavior for collab children (review finding).
  "thread/archived",
  "thread/unarchived",
  "thread/compacted",
  // Registration path 1 handles a child's first thread/started; a repeat
  // must not reach the parent (it would restart the parent's thread state).
  "thread/started",
]);

export function routeCodexChildNotification(method: string): CodexChildNotificationRoute {
  if (CHILD_AGENT_EVENT_METHODS.has(method)) {
    return "agent-event";
  }
  if (CHILD_CHATTER_METHODS.has(method)) {
    return "drop";
  }
  // Unknown or parent-owned (serverRequest/resolved, approvals, …).
  return "parent";
}

function toCodexUserInputAnswer(
  questionId: string,
  value: ProviderUserInputAnswers[string],
): Effect.Effect<
  EffectCodexSchema.ToolRequestUserInputResponse__ToolRequestUserInputAnswer,
  CodexSessionRuntimeInvalidUserInputAnswersError
> {
  if (typeof value === "string") {
    return Effect.succeed({ answers: [value] });
  }
  if (Array.isArray(value)) {
    const answers = value.filter((entry): entry is string => typeof entry === "string");
    return Effect.succeed({ answers });
  }
  if (isCodexUserInputAnswerObject(value)) {
    return Effect.succeed({ answers: value.answers });
  }
  return Effect.fail(new CodexSessionRuntimeInvalidUserInputAnswersError({ questionId }));
}

function toCodexUserInputAnswers(
  answers: ProviderUserInputAnswers,
): Effect.Effect<
  EffectCodexSchema.ToolRequestUserInputResponse["answers"],
  CodexSessionRuntimeInvalidUserInputAnswersError
> {
  return Effect.forEach(
    Object.entries(answers),
    ([questionId, value]) =>
      toCodexUserInputAnswer(questionId, value).pipe(
        Effect.map((answer) => [questionId, answer] as const),
      ),
    { concurrency: 1 },
  ).pipe(Effect.map((entries) => Object.fromEntries(entries)));
}

function currentProviderThreadId(session: ProviderSession): string | undefined {
  return readResumeCursorThreadId(session.resumeCursor);
}

function updateSession(
  sessionRef: Ref.Ref<ProviderSession>,
  updates: Partial<ProviderSession> | ((session: ProviderSession) => Partial<ProviderSession>),
): Effect.Effect<void> {
  return Effect.gen(function* () {
    const updatedAt = DateTime.formatIso(yield* DateTime.now);
    yield* Ref.update(sessionRef, (session) => ({
      ...session,
      ...(typeof updates === "function" ? updates(session) : updates),
      updatedAt,
    }));
  });
}

function parseThreadSnapshot(
  response: EffectCodexSchema.V2ThreadReadResponse | EffectCodexSchema.V2ThreadRollbackResponse,
): CodexThreadSnapshot {
  return {
    threadId: response.thread.id,
    turns: response.thread.turns.map((turn) => ({
      id: TurnId.make(turn.id),
      items: turn.items,
    })),
  };
}

export const makeCodexSessionRuntime = (
  options: CodexSessionRuntimeOptions,
): Effect.Effect<
  CodexSessionRuntimeShape,
  CodexErrors.CodexAppServerError,
  ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope
> =>
  Effect.gen(function* () {
    const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
    const runtimeScope = yield* Scope.Scope;
    const crypto = yield* Crypto.Crypto;
    const events = yield* Queue.unbounded<ProviderEvent>();
    const pendingApprovalsRef = yield* Ref.make(new Map<ApprovalRequestId, PendingApproval>());
    const approvalCorrelationsRef = yield* Ref.make(new Map<string, ApprovalCorrelation>());
    const pendingUserInputsRef = yield* Ref.make(new Map<ApprovalRequestId, PendingUserInput>());
    const collabReceiverTurnsRef = yield* Ref.make(new Map<string, TurnId>());
    const collabChildAgentsRef = yield* Ref.make(new Map<string, CollabChildAgentState>());
    const collabChildMetadataRef = yield* Ref.make(new Map<string, CollabChildMetadataState>());
    /** Child provider-thread id → its currently running provider turn id. */
    const collabChildLiveTurnsRef = yield* Ref.make(new Map<string, string>());
    const suppressMemoryConsolidationNotification = makeMemoryConsolidationNotificationFilter();
    const closedRef = yield* Ref.make(false);

    // `~` is not shell-expanded when env vars are set via
    // `child_process.spawn`; `expandHomePath` lets a configured
    // `CODEX_HOME=~/.codex_work` reach codex as an absolute path.
    const resolvedHomePath = options.homePath ? expandHomePath(options.homePath) : undefined;
    const env = {
      ...options.environment,
      ...(resolvedHomePath ? { CODEX_HOME: resolvedHomePath } : {}),
    };
    const extendEnv = options.environment === undefined;
    const appServerArgs = codexSessionAppServerArgs(options.appServerArgs, options.launchArgs);
    const spawnCommand = yield* resolveSpawnCommand(options.binaryPath, appServerArgs, {
      env,
      extendEnv,
    });
    const child = yield* spawner
      .spawn(
        ChildProcess.make(spawnCommand.command, spawnCommand.args, {
          cwd: options.cwd,
          env,
          extendEnv,
          forceKillAfter: CODEX_APP_SERVER_FORCE_KILL_AFTER,
          shell: spawnCommand.shell,
        }),
      )
      .pipe(
        Effect.provideService(Scope.Scope, runtimeScope),
        Effect.mapError(
          (cause) =>
            new CodexErrors.CodexAppServerSpawnError({
              command: `${options.binaryPath} app-server`,
              cause,
            }),
        ),
      );

    const clientContext = yield* CodexClient.layerChildProcess(child).pipe(
      Layer.build,
      Effect.provideService(Scope.Scope, runtimeScope),
    );
    const client = yield* Effect.service(CodexClient.CodexAppServerClient).pipe(
      Effect.provide(clientContext),
    );
    const serverNotifications = yield* Queue.unbounded<CodexServerNotification>();
    const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
    const randomUUIDv4 = (purpose: CodexErrors.CodexAppServerIdentifierPurpose) =>
      crypto.randomUUIDv4.pipe(
        Effect.mapError(
          (cause) =>
            new CodexErrors.CodexAppServerIdentifierGenerationError({
              purpose,
              cause,
            }),
        ),
      );

    const sessionCreatedAt = yield* nowIso;
    const initialSession = {
      provider: PROVIDER,
      ...(options.providerInstanceId ? { providerInstanceId: options.providerInstanceId } : {}),
      status: "connecting",
      runtimeMode: options.runtimeMode,
      cwd: options.cwd,
      ...(options.model ? { model: options.model } : {}),
      threadId: options.threadId,
      ...(options.resumeCursor !== undefined ? { resumeCursor: options.resumeCursor } : {}),
      createdAt: sessionCreatedAt,
      updatedAt: sessionCreatedAt,
    } satisfies ProviderSession;
    const sessionRef = yield* Ref.make<ProviderSession>(initialSession);
    const offerEvent = (event: ProviderEvent) => Queue.offer(events, event).pipe(Effect.asVoid);

    const emitEvent = (event: Omit<ProviderEvent, "id" | "provider" | "createdAt">) =>
      Effect.gen(function* () {
        const id = yield* randomUUIDv4("provider-event");
        return yield* offerEvent({
          id: EventId.make(id),
          provider: PROVIDER,
          ...(options.providerInstanceId ? { providerInstanceId: options.providerInstanceId } : {}),
          createdAt: yield* nowIso,
          ...event,
        });
      });
    const emitSessionEvent = (method: string, message: string) =>
      emitEvent({
        kind: "session",
        threadId: options.threadId,
        method,
        message,
      });

    const updateCollabChildMetadata = (
      agentThreadId: string,
      update: { readonly model?: string; readonly effort?: string },
      overwriteKnown: boolean,
    ) =>
      Ref.modify(collabChildMetadataRef, (current) => {
        const previous = current.get(agentThreadId) ?? {
          model: undefined,
          effort: undefined,
          lookupStarted: false,
          closed: false,
        };
        const model =
          update.model && (overwriteKnown || !previous.model) ? update.model : previous.model;
        const effort =
          update.effort && (overwriteKnown || !previous.effort) ? update.effort : previous.effort;
        const changed = model !== previous.model || effort !== previous.effort;
        if (!changed) {
          return [false, current] as const;
        }
        const next = new Map(current);
        next.set(agentThreadId, { ...previous, model, effort });
        return [true, next] as const;
      });

    const markCollabChildClosed = (agentThreadId: string) =>
      Ref.update(collabChildMetadataRef, (current) => {
        const previous = current.get(agentThreadId) ?? {
          model: undefined,
          effort: undefined,
          lookupStarted: false,
          closed: false,
        };
        if (previous.closed) {
          return current;
        }
        const next = new Map(current);
        next.set(agentThreadId, { ...previous, closed: true });
        return next;
      });

    const markCollabChildOpen = (agentThreadId: string) =>
      Ref.update(collabChildMetadataRef, (current) => {
        const previous = current.get(agentThreadId);
        if (!previous?.closed) {
          return current;
        }
        const next = new Map(current);
        next.set(agentThreadId, { ...previous, closed: false });
        return next;
      });

    const emitCollabChildMetadataUpdated = Effect.fn(
      "CodexSessionRuntime.emitCollabChildMetadataUpdated",
    )(function* (agentThreadId: string) {
      const child = (yield* Ref.get(collabChildAgentsRef)).get(agentThreadId);
      const metadata = (yield* Ref.get(collabChildMetadataRef)).get(agentThreadId);
      if (!child || metadata?.closed) {
        return;
      }
      yield* emitEvent({
        kind: "notification",
        threadId: options.threadId,
        ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
        method: "collabAgent/metadataUpdated",
        payload: collabChildIdentity(child, metadata),
      });
    });

    const startCollabChildMetadataLookup = Effect.fn(
      "CodexSessionRuntime.startCollabChildMetadataLookup",
    )(function* (agentThreadId: string) {
      const shouldStart = yield* Ref.modify(collabChildMetadataRef, (current) => {
        const previous = current.get(agentThreadId) ?? {
          model: undefined,
          effort: undefined,
          lookupStarted: false,
          closed: false,
        };
        if (previous.lookupStarted || previous.closed) {
          return [false, current] as const;
        }
        const next = new Map(current);
        next.set(agentThreadId, { ...previous, lookupStarted: true });
        return [true, next] as const;
      });
      if (!shouldStart) {
        return;
      }

      // The child is already loaded. This rejoins it without starting a turn,
      // and excludeTurns avoids loading or replaying its history.
      yield* client.raw
        .request("thread/resume", { threadId: agentThreadId, excludeTurns: true })
        .pipe(
          Effect.flatMap(decodeCodexChildResumeMetadata),
          Effect.timeout("5 seconds"),
          Effect.flatMap((response) =>
            Effect.gen(function* () {
              if (response.thread.id !== agentThreadId) {
                return;
              }
              const child = (yield* Ref.get(collabChildAgentsRef)).get(agentThreadId);
              const metadata = (yield* Ref.get(collabChildMetadataRef)).get(agentThreadId);
              if (!child || metadata?.closed) {
                return;
              }
              const model = nonEmptyMetadataValue(response.model);
              const effort = nonEmptyMetadataValue(response.reasoningEffort);
              const changed = yield* updateCollabChildMetadata(
                agentThreadId,
                {
                  ...(model ? { model } : {}),
                  ...(effort ? { effort } : {}),
                },
                false,
              );
              if (changed) {
                yield* emitCollabChildMetadataUpdated(agentThreadId);
              }
            }),
          ),
          Effect.catch(() => Effect.void),
          Effect.forkIn(runtimeScope),
        );
    });

    const settlePendingApprovals = (decision: ProviderApprovalDecision) =>
      Ref.get(pendingApprovalsRef).pipe(
        Effect.flatMap((pendingApprovals) =>
          Effect.forEach(
            Array.from(pendingApprovals.values()),
            (pendingApproval) =>
              Deferred.succeed(pendingApproval.decision, decision).pipe(Effect.ignore),
            { discard: true },
          ),
        ),
      );

    const settlePendingUserInputs = (answers: ProviderUserInputAnswers) =>
      Ref.get(pendingUserInputsRef).pipe(
        Effect.flatMap((pendingUserInputs) =>
          Effect.forEach(
            Array.from(pendingUserInputs.values()),
            (pendingUserInput) =>
              Deferred.succeed(pendingUserInput.answers, answers).pipe(Effect.ignore),
            { discard: true },
          ),
        ),
      );

    /**
     * Registers v2 collab children and re-emits their notifications as
     * synthetic `collabAgent/*` events for the adapter's task.* synthesis.
     * Returns true when the notification was fully handled (must not reach
     * parent-timeline mapping).
     */
    const interceptCollabChildNotification = (notification: CodexServerNotification) =>
      Effect.gen(function* () {
        // Registration path 1: child thread announces itself with a
        // subAgent thread_spawn source.
        if (notification.method === "thread/started") {
          const thread = notification.params.thread;
          const spawn = readThreadSpawnSource(thread);
          if (!spawn) {
            return false;
          }
          const rootProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));
          if (thread.id === rootProviderThreadId) {
            return false;
          }
          // Merge with any subAgentActivity registration that got here
          // first. spawnTurnId is REGISTRATION-time-only on both paths: for
          // an already-known child we keep its value (set or unset) — a
          // later thread/started during an unrelated parent turn must not
          // backfill that turn as the spawn batch, which would stamp an old
          // child onto a new fleet's CTA (review finding). Only a genuinely
          // new registration captures the current turn.
          const existingChild = (yield* Ref.get(collabChildAgentsRef)).get(thread.id);
          const spawnTurnId = existingChild
            ? existingChild.spawnTurnId
            : ((yield* Ref.get(sessionRef)).activeTurnId ?? undefined);
          const state: CollabChildAgentState = {
            agentThreadId: thread.id,
            nickname: spawn.nickname ?? thread.agentNickname ?? existingChild?.nickname,
            role: spawn.role ?? thread.agentRole ?? existingChild?.role,
            agentPath: spawn.agentPath ?? existingChild?.agentPath,
            depth: spawn.depth ?? existingChild?.depth,
            parentThreadId:
              spawn.parentThreadId ?? thread.parentThreadId ?? existingChild?.parentThreadId,
            spawnTurnId,
          };
          yield* Ref.update(collabChildAgentsRef, (current) => {
            const next = new Map(current);
            next.set(thread.id, state);
            return next;
          });
          const metadata = (yield* Ref.get(collabChildMetadataRef)).get(thread.id);
          yield* emitEvent({
            kind: "notification",
            threadId: options.threadId,
            method: "collabAgent/started",
            ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}),
            payload: {
              ...collabChildIdentity(state, metadata),
              ...(state.depth !== undefined ? { depth: state.depth } : {}),
              ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}),
            },
          });
          yield* startCollabChildMetadataLookup(thread.id);
          return true;
        }

        // Registration path 2: parent-side subAgentActivity item names the
        // child thread (may arrive before or after thread/started).
        if (
          (notification.method === "item/started" || notification.method === "item/completed") &&
          notification.params.item.type === "subAgentActivity"
        ) {
          const item = notification.params.item;
          // Never register the session's ROOT thread as its own child. The
          // wire emits subAgentActivity {agentPath: "/root", interacted}
          // about the root during collab runs; registering it intercepted
          // every subsequent root notification — including the final
          // assistant message and turn/completed — so the thread hung
          // "working" after all subagents finished (live-probe finding).
          const rootProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));
          if (
            item.agentThreadId === rootProviderThreadId ||
            item.agentPath === "/root" ||
            item.agentPath === "/"
          ) {
            return false;
          }
          const activitySpawnTurnId = (yield* Ref.get(sessionRef)).activeTurnId ?? undefined;
          yield* Ref.update(collabChildAgentsRef, (current) => {
            const existing = current.get(item.agentThreadId);
            const next = new Map(current);
            // Merge-late semantics: when thread/started registered first, a
            // later subAgentActivity still carries the real agentPath (and a
            // derived nickname) — fill missing fields, never clobber known
            // ones. spawnTurnId is registration-time-only: for an already
            // registered child, a later activity during an UNRELATED turn
            // must not backfill that turn as the spawn batch (review
            // finding); an unset spawn turn stays unset.
            next.set(item.agentThreadId, {
              agentThreadId: item.agentThreadId,
              nickname:
                existing?.nickname ??
                item.agentPath.split("/").findLast((segment) => segment.length > 0),
              role: existing?.role,
              agentPath: existing?.agentPath ?? item.agentPath,
              depth: existing?.depth,
              parentThreadId: existing?.parentThreadId,
              spawnTurnId: existing ? existing.spawnTurnId : activitySpawnTurnId,
            });
            return next;
          });
          const registeredChild = (yield* Ref.get(collabChildAgentsRef)).get(item.agentThreadId);
          const metadata = (yield* Ref.get(collabChildMetadataRef)).get(item.agentThreadId);
          yield* emitEvent({
            kind: "notification",
            threadId: options.threadId,
            method: "collabAgent/activity",
            ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}),
            payload: {
              ...(registeredChild
                ? collabChildIdentity(registeredChild, metadata)
                : { agentThreadId: item.agentThreadId, agentPath: item.agentPath }),
              activityKind: item.kind,
            },
          });
          if (item.kind === "started") {
            yield* startCollabChildMetadataLookup(item.agentThreadId);
          }
          return true;
        }

        // Interception: notifications addressed to a registered child thread
        // become agent-scoped synthetic events instead of parent chatter.
        const providerConversationId = readNotificationThreadId(notification);
        if (!providerConversationId) {
          return false;
        }
        // Belt-and-braces: the root thread's traffic must never be
        // intercepted, whatever the registry says.
        const interceptRootId = currentProviderThreadId(yield* Ref.get(sessionRef));
        if (providerConversationId === interceptRootId) {
          return false;
        }

        if (
          interceptRootId !== undefined &&
          (notification.method === "thread/settings/updated" ||
            notification.method === "model/rerouted")
        ) {
          const model = nonEmptyMetadataValue(
            notification.method === "thread/settings/updated"
              ? notification.params.threadSettings.model
              : notification.params.toModel,
          );
          const effort =
            notification.method === "thread/settings/updated"
              ? nonEmptyMetadataValue(notification.params.threadSettings.effort)
              : undefined;
          const changed = yield* updateCollabChildMetadata(
            providerConversationId,
            {
              ...(model ? { model } : {}),
              ...(effort ? { effort } : {}),
            },
            true,
          );
          if (changed && (yield* Ref.get(collabChildAgentsRef)).has(providerConversationId)) {
            yield* emitCollabChildMetadataUpdated(providerConversationId);
          }
          return true;
        }

        const children = yield* Ref.get(collabChildAgentsRef);
        const child = children.get(providerConversationId);
        if (!child) {
          return false;
        }
        const metadata = (yield* Ref.get(collabChildMetadataRef)).get(child.agentThreadId);
        const childIdentity = collabChildIdentity(child, metadata);
        switch (notification.method) {
          case "turn/started": {
            yield* markCollabChildOpen(child.agentThreadId);
            const childTurnId =
              typeof (notification.params as { turn?: { id?: unknown } }).turn?.id === "string"
                ? ((notification.params as { turn: { id: string } }).turn.id as string)
                : undefined;
            if (childTurnId) {
              yield* Ref.update(collabChildLiveTurnsRef, (current) => {
                const next = new Map(current);
                next.set(child.agentThreadId, childTurnId);
                return next;
              });
            }
            yield* emitEvent({
              kind: "notification",
              threadId: options.threadId,
              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
              method: "collabAgent/turnStarted",
              payload: childIdentity,
            });
            return true;
          }
          case "turn/completed":
            yield* Ref.update(collabChildLiveTurnsRef, (current) => {
              const next = new Map(current);
              next.delete(child.agentThreadId);
              return next;
            });
            yield* emitEvent({
              kind: "notification",
              threadId: options.threadId,
              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
              method: "collabAgent/turnCompleted",
              payload: {
                ...childIdentity,
                turn: notification.params.turn,
              },
            });
            return true;
          case "thread/status/changed":
            yield* emitEvent({
              kind: "notification",
              threadId: options.threadId,
              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
              method: "collabAgent/statusChanged",
              payload: {
                ...childIdentity,
                status: notification.params.status,
              },
            });
            return true;
          case "thread/tokenUsage/updated":
            yield* emitEvent({
              kind: "notification",
              threadId: options.threadId,
              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
              method: "collabAgent/tokenUsage",
              payload: {
                ...childIdentity,
                tokenUsage: notification.params.tokenUsage,
              },
            });
            return true;
          case "item/started":
          case "item/completed":
            yield* emitEvent({
              kind: "notification",
              threadId: options.threadId,
              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
              method: "collabAgent/item",
              payload: {
                ...childIdentity,
                item: notification.params.item,
              },
            });
            return true;
          case "thread/closed":
            // The child is gone: drop its live-turn entry so a later Stop
            // doesn't waste a turn/interrupt RPC on a closed thread before
            // reaching the parent (review finding).
            yield* Ref.update(collabChildLiveTurnsRef, (current) => {
              const next = new Map(current);
              next.delete(child.agentThreadId);
              return next;
            });
            yield* markCollabChildClosed(child.agentThreadId);
            yield* emitEvent({
              kind: "notification",
              threadId: options.threadId,
              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
              method: "collabAgent/closed",
              payload: childIdentity,
            });
            return true;
          case "error": {
            // A child error must surface as a failed agent, not vanish into
            // the default swallow (review finding: the child stayed
            // "running" forever). Retryable errors (willRetry) keep the
            // child RUNNING and interruptible — mirroring the root error
            // handler; settling it would orphan a still-live child from
            // Stop (review finding). Terminal errors clean up the live turn
            // like thread/closed and reuse the statusChanged systemError
            // path.
            const willRetry = (notification.params as { willRetry?: boolean }).willRetry === true;
            if (willRetry) {
              return true;
            }
            yield* Ref.update(collabChildLiveTurnsRef, (current) => {
              const next = new Map(current);
              next.delete(child.agentThreadId);
              return next;
            });
            yield* emitEvent({
              kind: "notification",
              threadId: options.threadId,
              ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}),
              method: "collabAgent/statusChanged",
              payload: {
                ...childIdentity,
                status: { type: "systemError" },
              },
            });
            return true;
          }
          default:
            // Routing table decides (single source of truth, asserted
            // against captured wire traces): enumerated chatter is dropped,
            // everything else — including methods this build has never seen
            // — falls through to the parent path rather than vanishing.
            return routeCodexChildNotification(notification.method) === "drop";
        }
      });

    const handleRawNotification = (notification: CodexServerNotification) =>
      Effect.gen(function* () {
        const isMemoryConsolidationNotification =
          suppressMemoryConsolidationNotification(notification);

        const payload = notification.params;
        const route = readRouteFields(notification);
        const collabReceiverTurns = yield* Ref.get(collabReceiverTurnsRef);
        const childParentTurnId = (() => {
          const providerConversationId = readNotificationThreadId(notification);
          return providerConversationId
            ? collabReceiverTurns.get(providerConversationId)
            : undefined;
        })();

        rememberCollabReceiverTurns(collabReceiverTurns, notification, route.turnId);
        // Interception FIRST: a registered v2 child is usually also in the
        // receiver-turn map (collabAgentToolCall.receiverThreadIds), and the
        // legacy suppressor below would drop its lifecycle before it could
        // become synthetic collabAgent events (review finding). The
        // suppressor still covers UNREGISTERED children.
        if (yield* interceptCollabChildNotification(notification)) {
          yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns);
          return;
        }

        // Suppression applies to receiver-map children (v1) AND to any
        // conversation that is not the root thread. The live capture
        // (codexMultiAgentWire.json) shows a child's thread/status/changed
        // arriving BEFORE anything registers the child — pre-registration
        // lifecycle must not reach the parent path, where the adapter maps
        // thread/* onto parent session state. Root-id-known guard keeps the
        // root's own early notifications flowing during session open.
        const suppressRootId = currentProviderThreadId(yield* Ref.get(sessionRef));
        const foreignConversation = (() => {
          const providerConversationId = readNotificationThreadId(notification);
          return (
            providerConversationId !== undefined &&
            suppressRootId !== undefined &&
            providerConversationId !== suppressRootId
          );
        })();
        if (
          (childParentTurnId !== undefined || foreignConversation) &&
          shouldSuppressChildConversationNotification(notification.method)
        ) {
          // Stop-everything must not depend on registration timing: a
          // child's turn/started can arrive before the subAgentActivity that
          // registers it (captured ordering), and suppressing it without
          // remembering the live turn would leave that child running after
          // Stop (review finding). Track live turns for ANY foreign
          // conversation; interrupts are best-effort per child, so a
          // false-positive entry costs one ignored RPC at worst.
          const foreignThreadId = readNotificationThreadId(notification);
          if (foreignThreadId !== undefined) {
            if (notification.method === "turn/started") {
              const foreignTurnId =
                typeof (notification.params as { turn?: { id?: unknown } }).turn?.id === "string"
                  ? (notification.params as { turn: { id: string } }).turn.id
                  : undefined;
              if (foreignTurnId) {
                yield* Ref.update(collabChildLiveTurnsRef, (current) => {
                  const next = new Map(current);
                  next.set(foreignThreadId, foreignTurnId);
                  return next;
                });
              }
            } else if (
              notification.method === "turn/completed" ||
              notification.method === "thread/closed"
            ) {
              yield* Ref.update(collabChildLiveTurnsRef, (current) => {
                const next = new Map(current);
                next.delete(foreignThreadId);
                return next;
              });
            }
          }
          yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns);
          return;
        }

        if (isMemoryConsolidationNotification) {
          return;
        }

        let requestId: ApprovalRequestId | undefined;
        let requestKind: ProviderRequestKind | undefined;
        let turnId = childParentTurnId ?? route.turnId;
        let itemId = route.itemId;

        if (notification.method === "serverRequest/resolved") {
          const rawRequestId =
            typeof notification.params.requestId === "string"
              ? notification.params.requestId
              : String(notification.params.requestId);
          const correlation = rawRequestId
            ? (yield* Ref.get(approvalCorrelationsRef)).get(rawRequestId)
            : undefined;
          if (correlation) {
            requestId = correlation.requestId;
            requestKind = correlation.requestKind;
            turnId = correlation.turnId ?? turnId;
            itemId = correlation.itemId ?? itemId;
            yield* Ref.update(approvalCorrelationsRef, (current) => {
              const next = new Map(current);
              next.delete(rawRequestId);
              return next;
            });
          }
        }

        yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns);
        yield* emitEvent({
          kind: "notification",
          threadId: options.threadId,
          method: notification.method,
          ...(turnId ? { turnId } : {}),
          ...(itemId ? { itemId } : {}),
          ...(requestId ? { requestId } : {}),
          ...(requestKind ? { requestKind } : {}),
          ...(notification.method === "item/agentMessage/delta"
            ? { textDelta: notification.params.delta }
            : {}),
          ...(payload !== undefined ? { payload } : {}),
        });
      });

    const currentSessionProviderThreadId = Effect.map(Ref.get(sessionRef), currentProviderThreadId);

    yield* client.handleServerNotification("thread/started", (payload) =>
      currentSessionProviderThreadId.pipe(
        Effect.flatMap((providerThreadId) => {
          if (providerThreadId && payload.thread.id !== providerThreadId) {
            return Effect.void;
          }
          return updateSession(sessionRef, {
            resumeCursor: { threadId: payload.thread.id },
          });
        }),
      ),
    );

    yield* client.handleServerNotification("turn/started", (payload) =>
      currentSessionProviderThreadId.pipe(
        Effect.flatMap((providerThreadId) => {
          if (providerThreadId && payload.threadId !== providerThreadId) {
            return Effect.void;
          }
          return updateSession(sessionRef, {
            status: "running",
            activeTurnId: TurnId.make(payload.turn.id),
          });
        }),
      ),
    );

    yield* client.handleServerNotification("turn/completed", (payload) =>
      currentSessionProviderThreadId.pipe(
        Effect.flatMap((providerThreadId) => {
          if (providerThreadId && payload.threadId !== providerThreadId) {
            return Effect.void;
          }
          const lastError =
            payload.turn.status === "failed" && "error" in payload.turn && payload.turn.error
              ? payload.turn.error.message
              : undefined;
          return updateSession(sessionRef, {
            status: payload.turn.status === "failed" ? "error" : "ready",
            activeTurnId: undefined,
            ...(lastError ? { lastError } : {}),
          });
        }),
      ),
    );

    yield* client.handleServerNotification("error", (payload) =>
      currentSessionProviderThreadId.pipe(
        Effect.flatMap((providerThreadId) => {
          const payloadThreadId = payload.threadId;
          if (providerThreadId && payloadThreadId && payloadThreadId !== providerThreadId) {
            return Effect.void;
          }
          const errorMessage = payload.error.message;
          const willRetry = payload.willRetry;
          return updateSession(sessionRef, {
            status: willRetry ? "running" : "error",
            ...(errorMessage ? { lastError: errorMessage } : {}),
          });
        }),
      ),
    );

    yield* client.handleServerRequest("item/commandExecution/requestApproval", (payload) =>
      Effect.gen(function* () {
        const requestId = ApprovalRequestId.make(yield* randomUUIDv4("command-approval-request"));
        const turnId = TurnId.make(payload.turnId);
        const itemId = ProviderItemId.make(payload.itemId);
        const decision = yield* Deferred.make<ProviderApprovalDecision>();

        yield* Ref.update(pendingApprovalsRef, (current) => {
          const next = new Map(current);
          next.set(requestId, {
            requestId,
            jsonRpcId: payload.approvalId ?? payload.itemId,
            requestKind: "command",
            turnId,
            itemId,
            decision,
          });
          return next;
        });
        yield* Ref.update(approvalCorrelationsRef, (current) => {
          const next = new Map(current);
          next.set(payload.approvalId ?? payload.itemId, {
            requestId,
            requestKind: "command",
            turnId,
            itemId,
          });
          return next;
        });

        yield* emitEvent({
          kind: "request",
          threadId: options.threadId,
          method: "item/commandExecution/requestApproval",
          requestId,
          requestKind: "command",
          ...(turnId ? { turnId } : {}),
          ...(itemId ? { itemId } : {}),
          payload,
        });

        const resolved = yield* Deferred.await(decision).pipe(
          Effect.ensuring(
            Ref.update(pendingApprovalsRef, (current) => {
              const next = new Map(current);
              next.delete(requestId);
              return next;
            }),
          ),
        );
        return {
          decision: resolved === "acceptAlways" ? "acceptForSession" : resolved,
        } satisfies EffectCodexSchema.CommandExecutionRequestApprovalResponse;
      }),
    );

    yield* client.handleServerRequest("item/fileChange/requestApproval", (payload) =>
      Effect.gen(function* () {
        const requestId = ApprovalRequestId.make(
          yield* randomUUIDv4("file-change-approval-request"),
        );
        const turnId = TurnId.make(payload.turnId);
        const itemId = ProviderItemId.make(payload.itemId);
        const decision = yield* Deferred.make<ProviderApprovalDecision>();

        yield* Ref.update(pendingApprovalsRef, (current) => {
          const next = new Map(current);
          next.set(requestId, {
            requestId,
            jsonRpcId: payload.itemId,
            requestKind: "file-change",
            turnId,
            itemId,
            decision,
          });
          return next;
        });
        yield* Ref.update(approvalCorrelationsRef, (current) => {
          const next = new Map(current);
          next.set(payload.itemId, {
            requestId,
            requestKind: "file-change",
            turnId,
            itemId,
          });
          return next;
        });

        yield* emitEvent({
          kind: "request",
          threadId: options.threadId,
          method: "item/fileChange/requestApproval",
          requestId,
          requestKind: "file-change",
          ...(turnId ? { turnId } : {}),
          ...(itemId ? { itemId } : {}),
          payload,
        });

        const resolved = yield* Deferred.await(decision).pipe(
          Effect.ensuring(
            Ref.update(pendingApprovalsRef, (current) => {
              const next = new Map(current);
              next.delete(requestId);
              return next;
            }),
          ),
        );
        return {
          decision: resolved === "acceptAlways" ? "acceptForSession" : resolved,
        } satisfies EffectCodexSchema.FileChangeRequestApprovalResponse;
      }),
    );

    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 EffectCodexSchema.McpServerElicitationRequestResponse;
        }

        const requestId = ApprovalRequestId.make(yield* randomUUIDv4("mcp-elicitation-request"));
        const turnId = payload.turnId
          ? TurnId.make(payload.turnId)
          : (yield* Ref.get(sessionRef)).activeTurnId;
        const jsonRpcId = payload.mode === "url" ? payload.elicitationId : requestId;
        const decision = yield* Deferred.make<ProviderApprovalDecision>();

        yield* Ref.update(pendingApprovalsRef, (current) => {
          const next = new Map(current);
          next.set(requestId, {
            requestId,
            jsonRpcId,
            requestKind: "mcp-elicitation",
            turnId,
            itemId: undefined,
            decision,
          });
          return next;
        });
        yield* Ref.update(approvalCorrelationsRef, (current) => {
          const next = new Map(current);
          next.set(jsonRpcId, {
            requestId,
            requestKind: "mcp-elicitation",
            turnId,
            itemId: undefined,
          });
          return next;
        });

        yield* emitEvent({
          kind: "request",
          threadId: options.threadId,
          method: "mcpServer/elicitation/request",
          requestId,
          requestKind: "mcp-elicitation",
          ...(turnId ? { turnId } : {}),
          payload,
        });

        const resolved = yield* Deferred.await(decision).pipe(
          Effect.ensuring(
            Ref.update(pendingApprovalsRef, (current) => {
              const next = new Map(current);
              next.delete(requestId);
              return next;
            }),
          ),
        );
        return toMcpElicitationResponse(payload, resolved);
      }),
    );

    yield* client.handleServerRequest("item/tool/requestUserInput", (payload) =>
      Effect.gen(function* () {
        const requestId = ApprovalRequestId.make(yield* randomUUIDv4("user-input-request"));
        const turnId = TurnId.make(payload.turnId);
        const itemId = ProviderItemId.make(payload.itemId);
        const answers = yield* Deferred.make<ProviderUserInputAnswers>();

        yield* Ref.update(pendingUserInputsRef, (current) => {
          const next = new Map(current);
          next.set(requestId, {
            requestId,
            turnId,
            itemId,
            answers,
          });
          return next;
        });

        yield* emitEvent({
          kind: "request",
          threadId: options.threadId,
          method: "item/tool/requestUserInput",
          requestId,
          ...(turnId ? { turnId } : {}),
          ...(itemId ? { itemId } : {}),
          payload,
        });

        const resolvedAnswers = yield* Deferred.await(answers).pipe(
          Effect.ensuring(
            Ref.update(pendingUserInputsRef, (current) => {
              const next = new Map(current);
              next.delete(requestId);
              return next;
            }),
          ),
        );

        return {
          answers: yield* toCodexUserInputAnswers(resolvedAnswers).pipe(
            Effect.mapError((error) =>
              CodexErrors.CodexAppServerRequestError.invalidParams(error.message, {
                questionId: error.questionId,
              }),
            ),
          ),
        } satisfies EffectCodexSchema.ToolRequestUserInputResponse;
      }),
    );

    yield* client.handleUnknownServerRequest((method) =>
      Effect.fail(CodexErrors.CodexAppServerRequestError.methodNotFound(method)),
    );

    const registerServerNotification = <M extends CodexRpc.ServerNotificationMethod>(method: M) =>
      client.handleServerNotification(method, (params) =>
        Queue.offer(serverNotifications, makeCodexServerNotification(method, params)).pipe(
          Effect.asVoid,
        ),
      );

    yield* Effect.forEach(
      Object.values(
        CodexRpc.SERVER_NOTIFICATION_METHODS,
      ) as ReadonlyArray<CodexRpc.ServerNotificationMethod>,
      registerServerNotification,
      { concurrency: 1, discard: true },
    );

    yield* Stream.fromQueue(serverNotifications).pipe(
      Stream.runForEach(handleRawNotification),
      Effect.forkIn(runtimeScope),
    );

    const stderrRemainderRef = yield* Ref.make("");
    yield* child.stderr.pipe(
      Stream.decodeText(),
      Stream.runForEach((chunk) =>
        Ref.modify(stderrRemainderRef, (current) => {
          const combined = current + chunk;
          const lines = combined.split("\n");
          const remainder = lines.pop() ?? "";
          return [lines.map((line) => line.replace(/\r$/, "")), remainder] as const;
        }).pipe(
          Effect.flatMap((lines) =>
            Effect.forEach(
              lines,
              (line) => {
                const classified = classifyCodexStderrLine(line);
                if (!classified) {
                  return Effect.void;
                }
                return emitEvent({
                  kind: "notification",
                  threadId: options.threadId,
                  method: "process/stderr",
                  message: classified.message,
                });
              },
              { discard: true },
            ),
          ),
        ),
      ),
      Effect.forkIn(runtimeScope),
    );

    yield* child.exitCode.pipe(
      Effect.flatMap((exitCode) =>
        Ref.get(closedRef).pipe(
          Effect.flatMap((closed) => {
            if (closed) {
              return Effect.void;
            }
            const nextStatus = exitCode === 0 ? "closed" : "error";
            return updateSession(sessionRef, {
              status: nextStatus,
              activeTurnId: undefined,
            }).pipe(
              Effect.andThen(
                emitSessionEvent(
                  "session/exited",
                  exitCode === 0
                    ? "Codex App Server exited."
                    : `Codex App Server exited with code ${exitCode}.`,
                ),
              ),
            );
          }),
        ),
      ),
      Effect.forkIn(runtimeScope),
    );

    const start = Effect.fn("CodexSessionRuntime.start")(function* () {
      yield* emitSessionEvent("session/connecting", "Starting Codex App Server session.");
      yield* client.request("initialize", buildCodexInitializeParams());
      yield* client.notify("initialized", undefined);

      const requestedModel = normalizeCodexModelSlug(options.model);

      const opened = yield* openCodexThread({
        client,
        threadId: options.threadId,
        runtimeMode: options.runtimeMode,
        cwd: options.cwd,
        requestedModel,
        serviceTier: options.serviceTier,
        resumeThreadId: readResumeCursorThreadId(options.resumeCursor),
      });

      const providerThreadId = opened.thread.id;
      const session = {
        ...(yield* Ref.get(sessionRef)),
        status: "ready",
        cwd: opened.cwd,
        model: opened.model,
        resumeCursor: { threadId: providerThreadId },
        updatedAt: yield* nowIso,
      } satisfies ProviderSession;
      yield* Ref.set(sessionRef, session);
      yield* emitSessionEvent("session/ready", "Codex App Server session ready.");
      return session;
    });

    const readProviderThreadId = Effect.gen(function* () {
      const providerThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));
      if (!providerThreadId) {
        return yield* new CodexSessionRuntimeThreadIdMissingError({
          threadId: options.threadId,
        });
      }
      return providerThreadId;
    });

    const close = Effect.gen(function* () {
      const alreadyClosed = yield* Ref.getAndSet(closedRef, true);
      if (alreadyClosed) {
        return;
      }
      yield* settlePendingApprovals("cancel");
      yield* settlePendingUserInputs({});
      yield* updateSession(sessionRef, {
        status: "closed",
        activeTurnId: undefined,
      });
      yield* emitSessionEvent("session/closed", "Session stopped").pipe(
        Effect.catch((cause) =>
          Effect.logError("Failed to emit Codex session closed event.", { cause }),
        ),
      );
      yield* Scope.close(runtimeScope, Exit.void);
      yield* Queue.shutdown(serverNotifications);
      yield* Queue.shutdown(events);
    });

    return {
      start,
      getSession: Ref.get(sessionRef),
      compactThread: Effect.gen(function* () {
        const providerThreadId = yield* readProviderThreadId;
        yield* client.request("thread/compact/start", { threadId: providerThreadId });
      }),
      sendTurn: (input) =>
        Effect.gen(function* () {
          const providerThreadId = yield* readProviderThreadId;
          if (hasConfiguredMcpServer(options.appServerArgs)) {
            yield* client.request("config/mcpServer/reload", undefined).pipe(
              Effect.catch((cause) =>
                Effect.logWarning("Failed to refresh Codex MCP tool catalog before turn.", {
                  cause,
                }),
              ),
            );
          }
          const normalizedModel = normalizeCodexModelSlug(
            input.model ?? (yield* Ref.get(sessionRef)).model,
          );
          const params = yield* buildTurnStartParams({
            threadId: providerThreadId,
            runtimeMode: options.runtimeMode,
            ...(input.input ? { prompt: input.input } : {}),
            ...(input.attachments ? { attachments: input.attachments } : {}),
            ...(normalizedModel ? { model: normalizedModel } : {}),
            ...(input.serviceTier ? { serviceTier: input.serviceTier } : {}),
            ...(input.effort ? { effort: input.effort } : {}),
            ...(input.interactionMode ? { interactionMode: input.interactionMode } : {}),
            // Derived from the session's own credential rather than the
            // setting, so the prompt describes the tools this turn actually
            // has even if the setting changed after the session started.
            browserToolsAvailable: configuredMcpToolAvailability(
              options.appServerArgs,
              options.mcpCapabilities,
            ),
          });
          const rawResponse = yield* client.raw.request("turn/start", params);
          const response = yield* decodeV2TurnStartResponse(rawResponse).pipe(
            Effect.mapError((error) =>
              CodexErrors.CodexAppServerProtocolParseError.fromSchemaError(
                "decode-response-payload",
                error,
                { method: "turn/start" },
              ),
            ),
          );
          const turnId = TurnId.make(response.turn.id);
          yield* updateSession(sessionRef, (session) => ({
            status: "running",
            // Codex accepts follow-ups while the current turn is still
            // running. The response contains the queued turn id, but
            // turn/interrupt only accepts the id that is active now.
            activeTurnId: session.activeTurnId ?? turnId,
            ...(normalizedModel ? { model: normalizedModel } : {}),
          }));
          const resumedProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));
          return {
            threadId: options.threadId,
            turnId,
            ...(resumedProviderThreadId
              ? { resumeCursor: { threadId: resumedProviderThreadId } }
              : {}),
          } satisfies ProviderTurnStartResult;
        }),
      interruptTurn: (turnId) =>
        Effect.gen(function* () {
          const providerThreadId = yield* readProviderThreadId;
          const session = yield* Ref.get(sessionRef);
          // Stop-everything: children are full threads with their own turns;
          // interrupting only the parent leaves the fleet running. Interrupt
          // each live child turn first, best-effort per child, BOUNDED: the
          // transport awaits an unbounded Deferred per request, so a wedged
          // child would otherwise block the parent interrupt forever —
          // exactly during the runaway fleet where Stop matters most
          // (review finding). Per-child and overall deadlines guarantee the
          // parent interrupt below always runs.
          const liveChildTurns = yield* Ref.get(collabChildLiveTurnsRef);
          yield* Effect.forEach(
            Array.from(liveChildTurns.entries()),
            ([childThreadId, childTurnId]) =>
              client
                .request("turn/interrupt", {
                  threadId: childThreadId,
                  turnId: childTurnId,
                })
                .pipe(Effect.timeoutOption("3 seconds"), Effect.ignore),
            { concurrency: 8, discard: true },
          ).pipe(Effect.timeoutOption("10 seconds"), Effect.ignore);
          const effectiveTurnId = turnId ?? session.activeTurnId;
          if (!effectiveTurnId) {
            return;
          }
          yield* client.request("turn/interrupt", {
            threadId: providerThreadId,
            turnId: effectiveTurnId,
          });
        }),
      readThread: Effect.gen(function* () {
        const providerThreadId = yield* readProviderThreadId;
        const response = yield* client.request("thread/read", {
          threadId: providerThreadId,
          includeTurns: true,
        });
        return parseThreadSnapshot(response);
      }),
      rollbackThread: (numTurns) =>
        Effect.gen(function* () {
          const providerThreadId = yield* readProviderThreadId;
          const response = yield* client.request("thread/rollback", {
            threadId: providerThreadId,
            numTurns,
          });
          yield* updateSession(sessionRef, {
            status: "ready",
            activeTurnId: undefined,
          });
          return parseThreadSnapshot(response);
        }),
      uploadFeedback: (reason) =>
        Effect.gen(function* () {
          const providerThreadId = yield* readProviderThreadId;
          return yield* client.request("feedback/upload", {
            classification: "bug",
            includeLogs: true,
            ...(reason ? { reason } : {}),
            threadId: providerThreadId,
          });
        }),
      respondToRequest: (requestId, decision) =>
        Effect.gen(function* () {
          const pending = (yield* Ref.get(pendingApprovalsRef)).get(requestId);
          if (!pending) {
            return yield* new CodexSessionRuntimePendingApprovalNotFoundError({
              requestId,
            });
          }
          yield* Ref.update(pendingApprovalsRef, (current) => {
            const next = new Map(current);
            next.delete(requestId);
            return next;
          });
          yield* Deferred.succeed(pending.decision, decision);
          yield* emitEvent({
            kind: "notification",
            threadId: options.threadId,
            method: "item/requestApproval/decision",
            requestId: pending.requestId,
            requestKind: pending.requestKind,
            ...(pending.turnId ? { turnId: pending.turnId } : {}),
            ...(pending.itemId ? { itemId: pending.itemId } : {}),
            payload: {
              requestId: pending.requestId,
              requestKind: pending.requestKind,
              decision,
            },
          });
        }),
      respondToUserInput: (requestId, answers) =>
        Effect.gen(function* () {
          const pending = (yield* Ref.get(pendingUserInputsRef)).get(requestId);
          if (!pending) {
            return yield* new CodexSessionRuntimePendingUserInputNotFoundError({
              requestId,
            });
          }
          const codexAnswers = yield* toCodexUserInputAnswers(answers);
          yield* Ref.update(pendingUserInputsRef, (current) => {
            const next = new Map(current);
            next.delete(requestId);
            return next;
          });
          yield* Deferred.succeed(pending.answers, answers);
          yield* emitEvent({
            kind: "notification",
            threadId: options.threadId,
            method: "item/tool/requestUserInput/answered",
            requestId: pending.requestId,
            ...(pending.turnId ? { turnId: pending.turnId } : {}),
            ...(pending.itemId ? { itemId: pending.itemId } : {}),
            payload: {
              answers: codexAnswers,
            },
          });
        }),
      events: Stream.fromQueue(events),
      close,
    } satisfies CodexSessionRuntimeShape;
  });