import { randomUUID } from "node:crypto";

import type {
  CheckpointControlPlaneSessionOptions,
  ControlPlanePort,
} from "./contracts/control-plane-port.js";
import type {
  NativeExecutionInput,
  NativeSessionExecutionResult,
} from "./contracts/native-execution.js";
import {
  buildNativeModelEnvelope,
  parseNativeExecutionInput,
} from "./contracts/native-execution.js";
import type {
  NativeSession,
  NativeSessionBackend,
} from "./contracts/native-session-backend.js";
import type { PersistedNativeSession } from "./contracts/native-session-backend.js";
import {
  validatePrpStructuredRunResult,
  type PrpEvent,
  type PrpStructuredRunResult,
  type PrpTerminalState,
} from "./protocol/replay-contract.js";
import { parsePaperclipQuestionSet } from "./contracts/question-set.js";

export const DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS = 120_000;
export const DEFAULT_NATIVE_SEMANTIC_RESULT_TERMINAL_GRACE_MS = 5_000;
const OPTIONAL_SESSION_CANCELLATION_GRACE_MS = 100;
const FAILED_OPERATION_SETTLEMENT_GRACE_MS = 100;
const DEFAULT_NATIVE_CHECKPOINT_TIMEOUT_MS = 30_000;
type NativeSessionCleanupDomain = string;

const failedSessionCleanupOwners = new Map<
  Promise<void>,
  NativeSessionCleanupDomain
>();
const FAILED_SESSION_CLOSE_RETRY_MS = 1_000;
const MAX_FAILED_SESSION_CLOSE_RETRIES = 3;
const MAX_QUARANTINED_SESSION_CLOSE_RETRIES = 3;
const MAX_QUARANTINED_SESSION_AUTOMATIC_CLOSE_ATTEMPTS = 9;
const QUARANTINED_SESSION_CLOSE_RETRY_MS = 60_000;
// Admission can inherit the initial close plus all three retained retries.
// Keep the wait finite while covering every bounded close attempt and all
// three production retry delays instead of timing out midway through recovery.
const QUARANTINED_SESSION_CLOSE_ATTEMPT_BOUND_MS = 7_000;
const QUARANTINED_SESSION_ADMISSION_GRACE_MS =
  (MAX_FAILED_SESSION_CLOSE_RETRIES + 1) *
    QUARANTINED_SESSION_CLOSE_ATTEMPT_BOUND_MS +
  MAX_FAILED_SESSION_CLOSE_RETRIES * FAILED_SESSION_CLOSE_RETRY_MS;
// Admission can observe a retained close recovery and then the one quarantine
// recovery that retained owner installs when it exhausts its retries. Give
// each owner generation a complete finite grace, while rejecting any
// unexpected third generation instead of permitting an unbounded owner chain.
const MAX_QUARANTINED_SESSION_ADMISSION_OWNER_PHASES = 2;

interface QuarantinedSessionCleanup {
  session: NativeSession;
  domain: NativeSessionCleanupDomain;
  automaticAttempts: number;
  attempt: Promise<void> | null;
  recovery: Promise<void> | null;
  recoveryMaxAttempts: number | null;
  timer: ReturnType<typeof setTimeout> | null;
}

const quarantinedSessionCleanups = new Set<QuarantinedSessionCleanup>();

export interface ExecuteNativeSessionOptions {
  input: NativeExecutionInput;
  backend: NativeSessionBackend;
  controlPlane: ControlPlanePort;
  runnerInstanceId: string;
  controlPlaneInstanceId: string;
  timeoutMs?: number;
  /** Abort admission while waiting for prior cleanup in the same domain. */
  signal?: AbortSignal;
  /** Internal test seam; production bounds checkpoint persistence to 30 seconds. */
  checkpointTimeoutMs?: number;
  /** Internal test seam; production uses the fixed 120-second platform policy. */
  runtimeInputLiveWindowMs?: number;
  /** Internal test seam; production gives the provider five seconds to end after a result. */
  semanticResultTerminalGraceMs?: number;
  onSession?: (session: NativeSession | null) => void;
  existingSession?: NativeSession;
  persistedSession?: PersistedNativeSession | null;
  keepSessionOpen?: boolean;
  onCheckpoint?: (
    snapshot: PersistedNativeSession,
    options?: CheckpointControlPlaneSessionOptions,
  ) => Promise<void> | void;
  /** Called when exact provider recovery failed and policy opened a new provider session. */
  onContinuityBreak?: (input: {
    reason: string;
    previousDriverSessionId: string;
    previousProviderSessionId: string | null;
    replacementDriverSessionId: string;
    replacementProviderSessionId: string | null;
  }) => Promise<void> | void;
  /**
   * Control-plane policy seam for a provider turn that completed after
   * durably creating a governed wait, but did not emit a semantic finish
   * result. The runner package cannot inspect server-owned interactions, so
   * it asks the embedding control plane whether that missing result is an
   * intentional yield before treating it as provider failure.
   */
  resolveMissingResult?: (input: {
    turnId: string | null;
    terminalEvent: PrpEvent;
  }) => Promise<PrpStructuredRunResult | null>;
  /**
   * Detect a durable server-owned wait as soon as its provider tool event is
   * committed. Models are not trusted to stop or avoid polling after creating
   * a question/review interaction; the control plane may park the turn here.
   * This boundary is deliberately synchronous and observational: asynchronous
   * mutation authority cannot be revoked safely after a failed execution.
   */
  resolveGovernedWait?: (input: {
    turnId: string | null;
    event: PrpEvent;
  }) => PrpStructuredRunResult | null;
}

function isTurnTerminal(event: PrpEvent): boolean {
  return [
    "turn.completed",
    "turn.failed",
    "turn.interrupted",
    "turn.cancelled",
  ].includes(event.eventType);
}

function terminalFromEvent(
  event: PrpEvent,
  disposition: PrpTerminalState["reportedWorkDisposition"],
): PrpTerminalState {
  const states =
    event.eventType === "turn.completed"
      ? {
          turnTerminalState: "completed" as const,
          runTerminalState: "succeeded" as const,
        }
      : event.eventType === "turn.failed"
        ? {
            turnTerminalState: "failed" as const,
            runTerminalState: "failed" as const,
          }
        : event.eventType === "turn.interrupted"
          ? {
              turnTerminalState: "interrupted" as const,
              runTerminalState: "cancelled" as const,
            }
          : {
              turnTerminalState: "cancelled" as const,
              runTerminalState: "cancelled" as const,
            };
  return {
    schema: "paperclip.prp.terminal.v1",
    ...states,
    reportedWorkDisposition: disposition,
  };
}

async function attemptOptionalSessionCancellation(
  session: NativeSession,
  reason: string,
): Promise<{ settlement: Promise<PromiseSettledResult<void>[]> } | null> {
  if (session.cancel === undefined) return null;
  const cancellationAbort = new AbortController();
  let cleanup: Promise<void>;
  try {
    // The session commits cancellation before returning. Only provider
    // cleanup remains asynchronous, so bounded failure settlement cannot
    // leave an operation with accepted-output or mutation authority.
    cleanup = session.cancel({
      reason,
      signal: cancellationAbort.signal,
    }).cleanup;
  } catch {
    return null;
  }
  const attempts = Promise.allSettled([cleanup]);
  if (await settlesWithin(attempts, OPTIONAL_SESSION_CANCELLATION_GRACE_MS)) {
    return null;
  }
  cancellationAbort.abort(
    new Error("native session cancellation grace expired"),
  );
  return { settlement: attempts };
}

async function settlesWithin(
  operation: Promise<unknown>,
  timeoutMs: number,
  signal?: AbortSignal,
): Promise<boolean> {
  signal?.throwIfAborted();
  let graceTimer: ReturnType<typeof setTimeout> | undefined;
  let removeAbort = () => {};
  try {
    const aborted = signal
      ? new Promise<never>((_resolve, reject) => {
          const onAbort = () =>
            reject(
              signal.reason ??
                new Error("native session cleanup admission aborted"),
            );
          if (signal.aborted) {
            onAbort();
          } else {
            signal.addEventListener("abort", onAbort, { once: true });
            removeAbort = () => signal.removeEventListener("abort", onAbort);
          }
        })
      : new Promise<never>(() => undefined);
    return await Promise.race([
      operation.then(() => true),
      new Promise<false>((resolve) => {
        graceTimer = setTimeout(() => resolve(false), timeoutMs);
        graceTimer.unref?.();
      }),
      aborted,
    ]);
  } finally {
    if (graceTimer !== undefined) clearTimeout(graceTimer);
    removeAbort();
  }
}

async function runAbortableOperationWithin<T>(input: {
  timeoutMs: number;
  timeoutMessage: string;
  timeoutError?: () => Error;
  operation: (signal: AbortSignal) => Promise<T>;
  onLateResolution?: (value: T) => Promise<void> | void;
}): Promise<T> {
  const operationAbort = new AbortController();
  let timer: ReturnType<typeof setTimeout> | undefined;
  let timedOut = false;
  let timeoutError: Error | undefined;
  const operation = input.operation(operationAbort.signal);
  // The timeout path deliberately stops awaiting an uncooperative adapter.
  // Keep its eventual settlement observed after mutation authority is revoked.
  void operation.catch(() => undefined);
  if (input.onLateResolution !== undefined) {
    void operation.then(
      (value) => {
        if (!timedOut) return;
        void Promise.resolve()
          .then(() => input.onLateResolution!(value))
          .catch(() => undefined);
      },
      () => undefined,
    );
  }
  try {
    const value = await Promise.race([
      operation,
      new Promise<never>((_resolve, reject) => {
        timer = setTimeout(() => {
          const error =
            input.timeoutError?.() ?? new Error(input.timeoutMessage);
          timedOut = true;
          timeoutError = error;
          reject(error);
          operationAbort.abort(error);
        }, input.timeoutMs);
        timer.unref?.();
      }),
    ]);
    // An abort listener can resolve synchronously before the timeout promise's
    // rejection wins the race. The deadline still owns that boundary: the late
    // result is disposed above and must never be admitted by the caller.
    if (timedOut) throw timeoutError;
    return value;
  } finally {
    if (timer !== undefined) clearTimeout(timer);
  }
}

async function disposeUnadmittedSession(
  session: NativeSession,
  reason: string,
  cleanupDomain: NativeSessionCleanupDomain,
): Promise<void> {
  // A provider may ignore abort and return a session after its caller has
  // already timed out. That session was never published through onSession, so
  // close it without clearing ownership that a later execution may establish.
  // Retain slow or failed cleanup in the same admission gate as an admitted
  // session: the absence of an onSession publication does not mean provider
  // resources have already been released.
  const cleanup = retainUnadmittedSessionCleanup(
    session,
    reason,
    cleanupDomain,
  );
  await settlesWithin(
    Promise.allSettled([cleanup]),
    FAILED_OPERATION_SETTLEMENT_GRACE_MS,
  );
}

function retainUnadmittedSessionCleanup(
  session: NativeSession,
  reason: string,
  cleanupDomain: NativeSessionCleanupDomain,
): Promise<void> {
  const cleanup = (async () => {
    let attempt = Promise.resolve().then(() => session.close({ reason }));
    let retryCount = 0;
    while (true) {
      try {
        await attempt;
        return;
      } catch (error) {
        if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) {
          quarantineSessionCleanup(session, cleanupDomain);
          throw error;
        }
        retryCount += 1;
        await waitForSessionCloseRetry();
        attempt = Promise.resolve().then(() =>
          session.close({
            reason: `${reason} cleanup recovery (${retryCount})`,
          }),
        );
      }
    }
  })();
  retainFailedSessionCleanupOwner(cleanup, cleanupDomain);
  return cleanup;
}

class NativeSessionFinalizationTimeoutError extends Error {
  constructor(timeoutMs: number) {
    super(`native session finalization timed out after ${timeoutMs}ms`);
    this.name = "NativeSessionFinalizationTimeoutError";
  }
}

async function finalizeWithin<T>(input: {
  timeoutMs: number;
  operation: (signal: AbortSignal) => Promise<T>;
}): Promise<T> {
  return runAbortableOperationWithin({
    ...input,
    timeoutMessage: `native session finalization timed out after ${input.timeoutMs}ms`,
    timeoutError: () =>
      new NativeSessionFinalizationTimeoutError(input.timeoutMs),
  });
}

async function finalizeIdempotentControlPlaneWithin<T>(input: {
  timeoutMs: number;
  operation: (signal: AbortSignal) => Promise<T>;
}): Promise<T> {
  try {
    return await finalizeWithin(input);
  } catch (error) {
    if (!(error instanceof NativeSessionFinalizationTimeoutError)) throw error;
    // Only the deterministic control-plane transaction enters this retry.
    // Its event ids and completion dedupe key make the second settlement an
    // authoritative acknowledgement of any first-attempt commit. Provider
    // result resolution and checkpointing are deliberately outside this
    // boundary and are never started a second time.
    return finalizeWithin(input);
  }
}

async function quarantineRetainedSession(
  session: NativeSession,
  onSession: ExecuteNativeSessionOptions["onSession"],
  reason: string,
  cleanupDomain: NativeSessionCleanupDomain,
): Promise<void> {
  // Eviction and provider cleanup are independent obligations. Keep both
  // observed so a throwing owner callback cannot prevent close from starting,
  // and a broken provider cannot keep the failed execution pending forever.
  const quarantineSettlement = Promise.allSettled([
    Promise.resolve().then(() => onSession?.(null)),
    retainUnadmittedSessionCleanup(session, reason, cleanupDomain),
  ]);
  await settlesWithin(
    quarantineSettlement,
    FAILED_OPERATION_SETTLEMENT_GRACE_MS,
  );
}

async function persistCheckpointWithin(input: {
  snapshot: PersistedNativeSession;
  controlPlane: ControlPlanePort;
  onCheckpoint: ExecuteNativeSessionOptions["onCheckpoint"];
  timeoutMs: number;
  externalSignal?: AbortSignal;
}): Promise<void> {
  const checkpointAbort = new AbortController();
  const externalSignal = input.externalSignal;
  let timer: ReturnType<typeof setTimeout> | undefined;
  let removeExternalAbort = () => {};
  const externalAbortFailure = externalSignal
    ? new Promise<never>((_resolve, reject) => {
        const abort = () => {
          const reason =
            externalSignal.reason ??
            new Error("native session checkpoint aborted");
          checkpointAbort.abort(reason);
          reject(reason);
        };
        if (externalSignal.aborted) {
          abort();
        } else {
          externalSignal.addEventListener("abort", abort, { once: true });
          removeExternalAbort = () =>
            externalSignal.removeEventListener("abort", abort);
        }
      })
    : new Promise<never>(() => undefined);
  const checkpointing = (async () => {
    const checkpointOptions = { signal: checkpointAbort.signal };
    await input.controlPlane.checkpointSession?.(
      input.snapshot,
      checkpointOptions,
    );
    if (checkpointAbort.signal.aborted) {
      throw (
        checkpointAbort.signal.reason ??
        new Error("native session checkpoint aborted")
      );
    }
    await input.onCheckpoint?.(input.snapshot, checkpointOptions);
  })();
  // The timeout path intentionally stops awaiting an uncooperative adapter.
  // Keep its eventual rejection observed after execution has quarantined and
  // closed the provider session.
  void checkpointing.catch(() => undefined);
  try {
    await Promise.race([
      checkpointing,
      new Promise<never>((_resolve, reject) => {
        timer = setTimeout(() => {
          const error = new Error(
            `native session checkpoint timed out after ${input.timeoutMs}ms`,
          );
          checkpointAbort.abort(error);
          reject(error);
        }, input.timeoutMs);
      }),
      externalAbortFailure,
    ]);
  } finally {
    if (timer !== undefined) clearTimeout(timer);
    removeExternalAbort();
  }
}

function waitForSessionCloseRetry(): Promise<void> {
  return new Promise((resolve) => {
    const timer = setTimeout(resolve, FAILED_SESSION_CLOSE_RETRY_MS);
    timer.unref?.();
  });
}
function retainFailedSessionCleanupOwner(
  cleanup: Promise<void>,
  cleanupDomain: NativeSessionCleanupDomain,
): void {
  failedSessionCleanupOwners.set(cleanup, cleanupDomain);
  void cleanup
    .finally(() => failedSessionCleanupOwners.delete(cleanup))
    .catch(() => undefined);
}

function quarantineSessionCleanup(
  session: NativeSession,
  cleanupDomain: NativeSessionCleanupDomain,
): void {
  if (
    [...quarantinedSessionCleanups].some((entry) => entry.session === session)
  ) {
    return;
  }
  const cleanup: QuarantinedSessionCleanup = {
    session,
    domain: cleanupDomain,
    automaticAttempts: 0,
    attempt: null,
    recovery: null,
    recoveryMaxAttempts: null,
    timer: null,
  };
  quarantinedSessionCleanups.add(cleanup);
  startQuarantinedSessionCleanupRecovery(
    cleanup,
    MAX_QUARANTINED_SESSION_CLOSE_RETRIES,
    "native session quarantined cleanup recovery",
  );
}

function startQuarantinedSessionCleanupRecovery(
  cleanup: QuarantinedSessionCleanup,
  maxAttempts: number,
  reason: string,
): Promise<void> {
  if (cleanup.recovery) return cleanup.recovery;
  const remainingAttempts =
    MAX_QUARANTINED_SESSION_AUTOMATIC_CLOSE_ATTEMPTS -
    cleanup.automaticAttempts;
  const boundedMaxAttempts = Math.min(maxAttempts, remainingAttempts);
  if (boundedMaxAttempts <= 0) return Promise.resolve();
  const recovery = (async () => {
    for (
      let attemptCount = 0;
      attemptCount < boundedMaxAttempts &&
      quarantinedSessionCleanups.has(cleanup);
      attemptCount += 1
    ) {
      // The first retry starts immediately so an admission-triggered recovery
      // receives the complete close grace. Later attempts retain the bounded
      // delay that prevents a hot retry loop.
      if (attemptCount > 0) await waitForSessionCloseRetry();
      cleanup.automaticAttempts += 1;
      const attempt = Promise.resolve().then(() =>
        cleanup.session.close({
          reason,
        }),
      );
      cleanup.attempt = attempt;
      try {
        await attempt;
        quarantinedSessionCleanups.delete(cleanup);
      } catch {
        // Retain the quarantine after this finite, sequential retry batch.
      } finally {
        if (cleanup.attempt === attempt) cleanup.attempt = null;
      }
    }
  })();
  cleanup.recovery = recovery;
  cleanup.recoveryMaxAttempts = boundedMaxAttempts;
  failedSessionCleanupOwners.set(recovery, cleanup.domain);
  void recovery
    .finally(() => {
      failedSessionCleanupOwners.delete(recovery);
      if (cleanup.recovery === recovery) {
        cleanup.recovery = null;
        cleanup.recoveryMaxAttempts = null;
      }
      scheduleQuarantinedSessionCleanup(cleanup);
    })
    .catch(() => undefined);
  return recovery;
}

function scheduleQuarantinedSessionCleanup(
  cleanup: QuarantinedSessionCleanup,
): void {
  if (
    !quarantinedSessionCleanups.has(cleanup) ||
    cleanup.recovery ||
    cleanup.attempt ||
    cleanup.timer ||
    cleanup.automaticAttempts >=
      MAX_QUARANTINED_SESSION_AUTOMATIC_CLOSE_ATTEMPTS
  ) {
    return;
  }
  // Keep cleanup live without a hot retry loop: one unref'd timer and one
  // sequential close are the maximum background work owned by each entry.
  // A later admission may accelerate, but never overlap, the scheduled try.
  cleanup.timer = setTimeout(() => {
    cleanup.timer = null;
    if (!quarantinedSessionCleanups.has(cleanup)) return;
    startQuarantinedSessionCleanupRecovery(
      cleanup,
      1,
      "native session scheduled quarantined cleanup recovery",
    );
  }, QUARANTINED_SESSION_CLOSE_RETRY_MS);
  cleanup.timer.unref?.();
}

async function retryQuarantinedSessionCleanups(
  cleanupDomain: NativeSessionCleanupDomain,
  signal?: AbortSignal,
): Promise<void> {
  // A close attempt becomes admission-visible as soon as the runtime retains
  // its exact cleanup owner. It may not have rejected yet, so it may not have
  // entered the retry quarantine below. Observe both states through the same
  // finite gate to prevent a later execution from opening a second provider
  // session while the first close still owns provider resources.
  const observedOwners = new Set<Promise<void>>();
  const acceleratedCleanups = new Set<QuarantinedSessionCleanup>();
  let observedOwnerPhases = 0;
  while (true) {
    signal?.throwIfAborted();
    const cleanupOwners = new Set<Promise<void>>(
      [...failedSessionCleanupOwners]
        .filter(([, domain]) => domain === cleanupDomain)
        .map(([owner]) => owner),
    );
    for (const cleanup of quarantinedSessionCleanups) {
      if (cleanup.domain !== cleanupDomain) continue;
      if (cleanup.timer) {
        clearTimeout(cleanup.timer);
        cleanup.timer = null;
      }
      if (cleanup.recovery) {
        // An autonomous scheduled owner receives one attempt. Admission must
        // observe it without mistaking it for the complete three-attempt
        // admission batch; if that attempt fails, the next bounded owner
        // phase accelerates one full batch. Existing full recoveries already
        // consumed that allowance and are never duplicated.
        if (
          (cleanup.recoveryMaxAttempts ?? 0) >=
          MAX_QUARANTINED_SESSION_CLOSE_RETRIES
        ) {
          acceleratedCleanups.add(cleanup);
        }
        cleanupOwners.add(cleanup.recovery);
      } else if (!acceleratedCleanups.has(cleanup)) {
        acceleratedCleanups.add(cleanup);
        if (
          cleanup.automaticAttempts <
          MAX_QUARANTINED_SESSION_AUTOMATIC_CLOSE_ATTEMPTS
        ) {
          cleanupOwners.add(
            startQuarantinedSessionCleanupRecovery(
              cleanup,
              MAX_QUARANTINED_SESSION_CLOSE_RETRIES,
              "native session quarantined admission recovery",
            ),
          );
        }
      }
    }
    const replacementOwners = [...cleanupOwners].filter(
      (owner) => !observedOwners.has(owner),
    );
    if (replacementOwners.length === 0) break;
    replacementOwners.forEach((owner) => observedOwners.add(owner));
    observedOwnerPhases += 1;
    if (
      observedOwnerPhases > MAX_QUARANTINED_SESSION_ADMISSION_OWNER_PHASES ||
      !(await settlesWithin(
        Promise.all(
          replacementOwners.map((owner) => owner.catch(() => undefined)),
        ),
        QUARANTINED_SESSION_ADMISSION_GRACE_MS,
        signal,
      ))
    ) {
      throw new Error(
        "native_session_cleanup_quarantined: prior session cleanup exceeded the admission grace",
      );
    }
  }
  if (
    [...failedSessionCleanupOwners.values()].some(
      (domain) => domain === cleanupDomain,
    ) ||
    [...quarantinedSessionCleanups].some(
      (cleanup) => cleanup.domain === cleanupDomain,
    )
  ) {
    // Admission may have pulled a scheduled retry forward and observed a
    // complete recovery batch that still failed. Restore autonomous ownership
    // before rejecting so cleanup cannot remain dormant until another run.
    for (const cleanup of quarantinedSessionCleanups) {
      if (cleanup.domain !== cleanupDomain) continue;
      scheduleQuarantinedSessionCleanup(cleanup);
    }
    throw new Error(
      "native_session_cleanup_quarantined: prior session cleanup remains incomplete",
    );
  }
}
async function consumeTurn(
  session: NativeSession,
  controlPlane: ControlPlanePort,
  timeoutMs: number,
  runtimeInputLiveWindowMs: number,
  semanticResultTerminalGraceMs: number,
  closeFailedSession: () => Promise<void>,
  quarantineSession: () => void,
  resolveGovernedWait?: ExecuteNativeSessionOptions["resolveGovernedWait"],
  externalSignal?: AbortSignal,
) {
  let timer: ReturnType<typeof setTimeout> | undefined;
  let semanticResultTimer: ReturnType<typeof setTimeout> | undefined;
  const semanticResultGraceExpired = Symbol("semantic_result_grace_expired");
  const appendAbort = new AbortController();
  const governedCleanupOperations = new Set<Promise<unknown>>();
  let governedCancellationCommitted = false;
  let deferredGovernedCleanupSettlement: Promise<unknown> | null = null;
  let deferredSessionCancellationSettlement: Promise<unknown> | null = null;
  const inputTimers = new Map<string, ReturnType<typeof setTimeout>>();
  const handoffCleanupOperations = new Set<Promise<unknown>>();
  const eventIterator = session.events()[Symbol.asyncIterator]();
  let stopConsumer = false;
  let rejectHandoff: ((error: unknown) => void) | null = null;
  const handoffFailure = new Promise<never>((_resolve, reject) => {
    rejectHandoff = reject;
  });
  let removeExternalAbort = () => {};
  const externalAbortFailure = externalSignal
    ? new Promise<never>((_resolve, reject) => {
        const abort = () => {
          const reason =
            externalSignal.reason ??
            new Error("native event consumption aborted");
          stopConsumer = true;
          appendAbort.abort(reason);
          reject(reason);
        };
        if (externalSignal.aborted) {
          abort();
        } else {
          externalSignal.addEventListener("abort", abort, { once: true });
          removeExternalAbort = () =>
            externalSignal.removeEventListener("abort", abort);
        }
      })
    : new Promise<never>(() => undefined);
  const clearInputTimer = (requestId: string) => {
    const inputTimer = inputTimers.get(requestId);
    if (inputTimer !== undefined) clearTimeout(inputTimer);
    inputTimers.delete(requestId);
  };
  const consumer = (async () => {
    let eventCount = 0;
    let highestContiguousSourceSeq = 0;
    let governedResult: PrpStructuredRunResult | null = null;
    let resultSource: "semantic_result" | "governed_wait" | null = null;
    let semanticResultEvent: PrpEvent | null = null;
    let semanticResultDeadline: Promise<
      typeof semanticResultGraceExpired
    > | null = null;
    let pendingNext: ReturnType<typeof eventIterator.next> | null = null;
    const settleDurableResult = (
      event: PrpEvent,
      result: PrpStructuredRunResult,
      reason: string,
    ) => {
      if (session.cancel === undefined) {
        throw new Error("native_governed_wait_cancellation_unavailable");
      }
      const cancellation = session.cancel({
        reason,
        signal: appendAbort.signal,
      });
      governedCancellationCommitted = true;
      const cleanup = cancellation.cleanup;
      governedCleanupOperations.add(cleanup);
      void cleanup
        .catch(() => quarantineSession())
        .finally(() => governedCleanupOperations.delete(cleanup));
      return {
        event,
        eventCount,
        highestContiguousSourceSeq,
        governedResult: result,
      };
    };
    while (true) {
      pendingNext ??= eventIterator.next();
      const next = semanticResultDeadline === null
        ? await pendingNext
        : await Promise.race([pendingNext, semanticResultDeadline]);
      if (next === semanticResultGraceExpired) {
        void pendingNext.catch(() => undefined);
        if (semanticResultEvent === null || governedResult === null) {
          throw new Error("native_semantic_result_grace_lost_result");
        }
        return settleDurableResult(
          semanticResultEvent,
          governedResult,
          "Paperclip accepted the durable semantic result.",
        );
      }
      pendingNext = null;
      if (stopConsumer) throw new Error("native event consumer stopped");
      if (next.done) {
        if (
          resultSource === "semantic_result" &&
          semanticResultEvent !== null &&
          governedResult !== null
        ) {
          return {
            event: semanticResultEvent,
            eventCount,
            highestContiguousSourceSeq,
            governedResult,
          };
        }
        throw new Error(
          "native event stream closed before a turn terminal fact",
        );
      }
      const event = next.value;
      const payload = event.payload as Record<string, unknown>;
      const settlingRequestId =
        [
          "runtime_request.resolved",
          "runtime_request.cancelled",
          "runtime_request.expired",
        ].includes(event.eventType) && typeof payload.requestId === "string"
          ? payload.requestId
          : null;
      // Beginning settlement revokes the expiry timer's handoff authority.
      // appendEvent may remain pending across the live-window deadline; if
      // the timer stayed live until the receipt returned, both settlement
      // and a durable handoff could commit for the same request.
      if (settlingRequestId !== null) clearInputTimer(settlingRequestId);
      const receipt = await controlPlane.appendEvent(event, {
        signal: appendAbort.signal,
      });
      if (stopConsumer) throw new Error("native event consumer stopped");
      eventCount += receipt.disposition === "committed" ? 1 : 0;
      highestContiguousSourceSeq = Math.max(
        highestContiguousSourceSeq,
        receipt.highestContiguousSourceSeq,
      );
      const request =
        payload.request &&
        typeof payload.request === "object" &&
        !Array.isArray(payload.request)
          ? (payload.request as Record<string, unknown>)
          : null;
      if (
        receipt.disposition === "committed" &&
        event.eventType === "runtime_request.created" &&
        request?.schema === "paperclip.runtime_request.v2" &&
        request.type === "input" &&
        typeof request.requestId === "string" &&
        typeof request.turnId === "string"
      ) {
        try {
          parsePaperclipQuestionSet(request.input);
          const requestId = request.requestId;
          const turnId = request.turnId;
          clearInputTimer(requestId);
          const inputTimer = setTimeout(() => {
            // Clearing a timeout does not revoke a callback that is already
            // queued. The map entry is the per-request authority token.
            if (inputTimers.get(requestId) !== inputTimer) return;
            inputTimers.delete(requestId);
            // A timer callback can already be queued when teardown clears
            // its handle. Re-check the live-turn authority inside the
            // callback before starting or registering durable work.
            if (stopConsumer || appendAbort.signal.aborted) return;
            if (session.handoffRuntimeRequest === undefined) {
              rejectHandoff?.(
                new Error("native_runtime_request_handoff_unavailable"),
              );
              return;
            }
            let handoffCleanup: Promise<unknown>;
            try {
              const handoff = session.handoffRuntimeRequest({
                requestId,
                turnId,
                reason: "durable_handoff",
                signal: appendAbort.signal,
              });
              // Durable handoff mutation is synchronous. The returned
              // promise owns provider interruption only, so it can remain
              // observed without acquiring authority to delay or reverse
              // a provider terminal fact.
              handoffCleanup = handoff.cleanup;
            } catch (error) {
              rejectHandoff?.(error);
              return;
            }
            handoffCleanupOperations.add(handoffCleanup);
            void handoffCleanup
              .catch((error) => {
                if (!stopConsumer && !appendAbort.signal.aborted) {
                  rejectHandoff?.(error);
                }
              })
              .finally(() => handoffCleanupOperations.delete(handoffCleanup));
          }, runtimeInputLiveWindowMs);
          inputTimer.unref?.();
          inputTimers.set(requestId, inputTimer);
        } catch {
          // Invalid structured inputs remain rejected by the driver and never become durable questions.
        }
      }
      if (governedResult === null && event.eventType === "run.result.proposed") {
        const validation = validatePrpStructuredRunResult(event.payload);
        if (!validation.ok) {
          throw new Error("native_semantic_result_invalid");
        }
        governedResult = validation.result;
        resultSource = "semantic_result";
        semanticResultEvent = event;
        if (session.cancel !== undefined) {
          semanticResultDeadline = new Promise((resolve) => {
            semanticResultTimer = setTimeout(
              () => resolve(semanticResultGraceExpired),
              semanticResultTerminalGraceMs,
            );
            semanticResultTimer.unref?.();
          });
        }
      }
      if (governedResult === null && resolveGovernedWait) {
        if (appendAbort.signal.aborted) {
          throw (
            appendAbort.signal.reason ??
            new Error("native event consumption aborted")
          );
        }
        governedResult = resolveGovernedWait({
          turnId: event.turnId ?? null,
          event,
        });
        if (governedResult !== null) resultSource = "governed_wait";
      }
      if (governedResult !== null && !isTurnTerminal(event)) {
        if (resultSource === "semantic_result") {
          // Give the provider a short grace to publish its final assistant
          // message and terminal after the semantic tool returns. If no
          // terminal arrives, the deadline above finalizes the durable result.
          continue;
        }
        return settleDurableResult(
          event,
          governedResult,
          "Paperclip parked this turn on a durable governed interaction.",
        );
      }
      if (isTurnTerminal(event)) {
        return {
          event,
          eventCount,
          highestContiguousSourceSeq,
          governedResult,
        };
      }
    }
  })();
  // A timeout can win the race while an iterator is still waiting for data.
  // Observe any later consumer rejection so it cannot become process-fatal.
  void consumer.catch(() => undefined);
  let consumptionFailed = false;
  try {
    return await Promise.race([
      consumer,
      new Promise<never>((_, reject) => {
        timer = setTimeout(() => {
          reject(new Error(`native session timed out after ${timeoutMs}ms`));
        }, timeoutMs);
      }),
      handoffFailure,
      externalAbortFailure,
    ]);
  } catch (error) {
    consumptionFailed = true;
    stopConsumer = true;
    appendAbort.abort(error);
    for (const inputTimer of inputTimers.values()) clearTimeout(inputTimer);
    inputTimers.clear();
    // Governed-wait discovery is synchronous and observational, and provider
    // cancellation has already committed synchronously. Bound only the
    // authority-free provider cleanup while keeping its outcome observed.
    if (governedCleanupOperations.size > 0) {
      const governedCleanupSettlement = Promise.allSettled([
        ...governedCleanupOperations,
      ]);
      if (
        !(await settlesWithin(
          governedCleanupSettlement,
          FAILED_OPERATION_SETTLEMENT_GRACE_MS,
        ))
      ) {
        deferredGovernedCleanupSettlement = governedCleanupSettlement;
      }
    }
    if (!governedCancellationCommitted) {
      const deferredCancellation = await attemptOptionalSessionCancellation(
        session,
        "Native session event consumption failed.",
      );
      deferredSessionCancellationSettlement =
        deferredCancellation?.settlement ?? null;
    }
    throw error;
  } finally {
    stopConsumer = true;
    for (const inputTimer of inputTimers.values()) clearTimeout(inputTimer);
    inputTimers.clear();
    const activeHandoffCleanupSettlement =
      handoffCleanupOperations.size > 0
        ? Promise.allSettled([...handoffCleanupOperations])
        : null;
    const activeGovernedCleanupSettlement =
      deferredGovernedCleanupSettlement === null &&
      governedCleanupOperations.size > 0
        ? Promise.allSettled([...governedCleanupOperations])
        : null;
    if (!consumptionFailed && !appendAbort.signal.aborted) {
      // A provider terminal or synchronous governed cancellation revokes
      // live-turn authority. Handoff state was already committed; abort only
      // tells provider cleanup that it must not begin any new work.
      appendAbort.abort(new Error("native turn reached a terminal state"));
    }
    // Do not let failure escape while the provider iterator still owns a live
    // subscription. Cancellation above is responsible for releasing a blocked
    // `next()`; awaiting `return()` then synchronizes the iterator's `finally`
    // teardown before the session can be closed or reused.
    const iteratorTeardown = eventIterator.return?.().catch(() => undefined);
    // The consumer may already be past `next()` and awaiting a durable append.
    // Abort is a control-plane durability boundary: appendEvent must settle
    // without committing when its signal is aborted. Handoff and cancellation
    // promises below own provider cleanup only; their durable transitions were
    // synchronous, so a slow cleanup cannot reverse terminal completion.
    const passiveTeardownSettlement = Promise.allSettled([
      iteratorTeardown,
      consumer,
      ...(activeHandoffCleanupSettlement
        ? [activeHandoffCleanupSettlement]
        : []),
      ...(activeGovernedCleanupSettlement
        ? [activeGovernedCleanupSettlement]
        : []),
      ...(deferredGovernedCleanupSettlement
        ? [deferredGovernedCleanupSettlement]
        : []),
      ...(deferredSessionCancellationSettlement
        ? [deferredSessionCancellationSettlement]
        : []),
    ]);
    if (consumptionFailed) {
      // Start provider close immediately so a cooperative implementation can
      // release a blocked iterator or provider operation. Abort has already
      // revoked every control-plane mutation capability and closeSession has
      // removed this session from the caller, so an implementation that
      // violates its cancellation contract is quarantined rather than allowed
      // to defeat the execution deadline. Promise.allSettled keeps every late
      // rejection observed after the bounded wait expires.
      const cleanupSettlement = Promise.allSettled([
        passiveTeardownSettlement,
        Promise.resolve().then(closeFailedSession),
      ]);
      await settlesWithin(
        cleanupSettlement,
        FAILED_OPERATION_SETTLEMENT_GRACE_MS,
      );
    } else {
      // Iterator and provider cleanup own no control-plane mutation authority.
      // A slow subscription or cleanup remains observed and is released by the
      // normal session close, but it cannot erase an already committed
      // terminal fact or prevent result retrieval and durable finalization.
      const teardownSettled = await settlesWithin(
        passiveTeardownSettlement,
        FAILED_OPERATION_SETTLEMENT_GRACE_MS,
      );
      if (!teardownSettled) quarantineSession();
    }
    if (timer !== undefined) clearTimeout(timer);
    if (semanticResultTimer !== undefined) clearTimeout(semanticResultTimer);
    removeExternalAbort();
  }
}

function checkpointCursor(cursor: string | null | undefined): number {
  if (cursor === undefined || cursor === null || cursor === "") return 0;
  const parsed = Number(cursor);
  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : 0;
}

async function reconcileRecoveryCursor(input: {
  controlPlane: ControlPlanePort;
  checkpoint: PersistedNativeSession;
  runId: string;
  sourceInstanceId: string;
  signal: AbortSignal;
}): Promise<PersistedNativeSession> {
  const checkpointHighWater = checkpointCursor(input.checkpoint.cursor);
  let afterSourceSeq = checkpointHighWater;
  let persistedHighWater = checkpointHighWater;
  while (true) {
    const replay = await input.controlPlane.replayEvents(
      {
        runId: input.runId,
        sourceInstanceId: input.sourceInstanceId,
        afterSourceSeq,
        limit: 1_000,
      },
      { signal: input.signal },
    );
    input.signal.throwIfAborted();
    if (replay.events.length === 0) break;
    const pageHighWater = replay.events.reduce(
      (highest, event) => Math.max(highest, event.sourceSeq),
      afterSourceSeq,
    );
    if (pageHighWater <= afterSourceSeq) {
      throw new Error("native_recovery_replay_did_not_advance");
    }
    persistedHighWater = Math.max(persistedHighWater, pageHighWater);
    afterSourceSeq = pageHighWater;
  }
  if (
    persistedHighWater === checkpointHighWater &&
    input.checkpoint.cursor === String(checkpointHighWater)
  ) {
    return input.checkpoint;
  }
  return { ...input.checkpoint, cursor: String(persistedHighWater) };
}

async function replayCheckpointedTurnTerminal(input: {
  controlPlane: ControlPlanePort;
  runId: string;
  sourceInstanceId: string;
  priorTerminalTurnIds: readonly string[];
  expectedTurnId?: string | null;
}): Promise<{ terminal: PrpEvent; hasPriorResultProposal: boolean } | null> {
  let afterSourceSeq = 0;
  const terminals: PrpEvent[] = [];
  const latestResultProposalByTurn = new Map<string, number>();
  const priorTerminalTurnIds = new Set(input.priorTerminalTurnIds);
  while (true) {
    const replay = await input.controlPlane.replayEvents({
      runId: input.runId,
      sourceInstanceId: input.sourceInstanceId,
      afterSourceSeq,
      limit: 1_000,
    });
    if (replay.events.length === 0) {
      // Disposition recovery owns the newest durable provider terminal. An
      // older task turn may also have a valid proposal, but selecting it would
      // finalize stale work and strand the actual recovery terminal.
      const terminal =
        [...terminals].sort(
          (left, right) => right.sourceSeq - left.sourceSeq,
        )[0] ?? null;
      const proposalSequence =
        latestResultProposalByTurn.get(terminal?.turnId ?? "") ?? 0;
      return terminal === null
        ? null
        : {
            terminal,
            hasPriorResultProposal:
              proposalSequence > 0 && proposalSequence < terminal.sourceSeq,
          };
    }
    for (const event of replay.events) {
      if (event.turnId && event.eventType === "run.result.proposed") {
        latestResultProposalByTurn.set(
          event.turnId,
          Math.max(
            latestResultProposalByTurn.get(event.turnId) ?? 0,
            event.sourceSeq,
          ),
        );
      }
      if (
        event.turnId &&
        isTurnTerminal(event) &&
        (input.expectedTurnId !== undefined && input.expectedTurnId !== null
          ? event.turnId === input.expectedTurnId
          : !priorTerminalTurnIds.has(event.turnId))
      ) {
        terminals.push(structuredClone(event));
      }
    }
    const pageHighWater = replay.events.reduce(
      (highest, event) => Math.max(highest, event.sourceSeq),
      afterSourceSeq,
    );
    if (pageHighWater <= afterSourceSeq) {
      throw new Error("native_recovery_replay_did_not_advance");
    }
    afterSourceSeq = pageHighWater;
  }
}

function checkpointedResultlessDispositionFallback(input: {
  persisted: PersistedNativeSession;
  recovered: PersistedNativeSession;
  controlPlaneInstanceId: string;
}): PrpEvent | null {
  const turnId = input.persisted.dispositionOnlyRecoveryTurnId;
  if (
    !input.persisted.dispositionOnlyRecoveryConsumed ||
    input.persisted.semanticResult ||
    input.persisted.activeTurnId ||
    typeof turnId !== "string" ||
    turnId.length === 0
  )
    return null;
  const persistedTerminal =
    input.persisted.terminalTurns?.filter(
      (terminal) => terminal.turnId === turnId,
    ) ?? [];
  if (
    persistedTerminal.length !== 1 ||
    persistedTerminal[0]!.fingerprint.length === 0
  ) {
    return null;
  }
  const recoveredTurnId = input.recovered.dispositionOnlyRecoveryTurnId;
  const recoveredTerminal =
    input.recovered.terminalTurns?.filter(
      (terminal) => terminal.turnId === recoveredTurnId,
    ) ?? [];
  if (
    !input.recovered.dispositionOnlyRecoveryConsumed ||
    input.recovered.semanticResult ||
    input.recovered.activeTurnId ||
    recoveredTurnId !== turnId ||
    recoveredTerminal.length !== 1 ||
    recoveredTerminal[0]!.fingerprint !== persistedTerminal[0]!.fingerprint ||
    canonicalJson(input.recovered.identity) !==
      canonicalJson(input.persisted.identity)
  ) {
    throw new Error("native_disposition_recovery_checkpoint_conflict");
  }
  return {
    schema: "paperclip.prp.event.v1",
    sourceEventId: `${input.controlPlaneInstanceId}:${input.persisted.identity.runId}:checkpointed-disposition-terminal`,
    sourceSeq: 1,
    sourceInstanceId: input.controlPlaneInstanceId,
    sourceKind: "control_plane",
    runId: input.persisted.identity.runId,
    normalizedSessionId: input.persisted.identity.sessionId,
    turnId,
    eventType: "turn.completed",
    schemaVersion: 1,
    priority: 0,
    emittedAt: new Date().toISOString(),
    payload: {
      recovery: "checkpointed_resultless_disposition",
      terminalFingerprint: persistedTerminal[0]!.fingerprint,
    },
  };
}

/**
 * Package-owned normalized session loop. Paperclip supplies persistence and
 * authority through ControlPlanePort; provider/session behavior stays here.
 */
export async function executeNativeSession(
  options: ExecuteNativeSessionOptions,
): Promise<NativeSessionExecutionResult> {
  const input = parseNativeExecutionInput(options.input);
  const descriptor = await options.backend.descriptor();
  const cleanupDomain = JSON.stringify([
    input.binding.companyId,
    descriptor.kind,
    descriptor.name,
  ]);
  await retryQuarantinedSessionCleanups(cleanupDomain, options.signal);
  if ("runtimeContext" in input) {
    const capabilities = descriptor.runtimeContextCapabilities;
    const unsupported = (["instructions", "skills", "mcp"] as const).filter(
      (key) => capabilities?.[key] !== "native",
    );
    if (unsupported.length)
      throw new Error(
        `native_runtime_context_unsupported: ${descriptor.name} does not natively realize ${unsupported.join(", ")}`,
      );
  }
  let persistedSession = options.existingSession
    ? null
    : (options.persistedSession ??
      (await options.controlPlane.loadSessionCheckpoint?.()) ??
      null);
  if (
    persistedSession &&
    (persistedSession.identity.runId !== input.binding.runId ||
      persistedSession.identity.companyId !== input.binding.companyId ||
      persistedSession.identity.issueId !== input.binding.issueId ||
      persistedSession.identity.agentId !== input.binding.agentId ||
      input.session.normalizedSessionId === null ||
      persistedSession.identity.sessionId !== input.session.normalizedSessionId)
  )
    throw new Error("native_session_checkpoint_binding_mismatch");
  const existingIdentity = options.existingSession?.identity() ?? null;
  if (
    existingIdentity &&
    (existingIdentity.companyId !== input.binding.companyId ||
      existingIdentity.issueId !== input.binding.issueId ||
      existingIdentity.agentId !== input.binding.agentId ||
      input.session.normalizedSessionId === null ||
      existingIdentity.sessionId !== input.session.normalizedSessionId)
  )
    throw new Error("native_session_attach_binding_mismatch");
  const normalizedSessionId =
    persistedSession?.identity.sessionId ??
    existingIdentity?.sessionId ??
    input.session.normalizedSessionId ??
    randomUUID();
  const identity = {
    runId: input.binding.runId,
    sessionId: normalizedSessionId,
    companyId: input.binding.companyId,
    issueId: input.binding.issueId,
    agentId: input.binding.agentId,
  };
  let recovered = false;
  let session: NativeSession | null = null;
  let continuityBreak: {
    reason: string;
    previousDriverSessionId: string;
    previousProviderSessionId: string | null;
  } | null = null;
  let reconciledRecoveryCheckpoint: PersistedNativeSession | null = null;
  if (options.existingSession) {
    if (options.existingSession.attachRun === undefined) {
      throw new Error("native_session_multi_run_unavailable");
    }
    // Attaching can fail even after the retained session's identity passes the
    // static binding check (for example, when the provider lost multi-run
    // state). Prove the provider attachment before opening durable
    // control-plane state because ControlPlanePort has no rollback operation.
    try {
      await options.existingSession.attachRun({ identity });
    } catch (error) {
      // attachRun has no transactional guarantee: a provider may bind the new
      // run before reporting a later failure. Conservatively quarantine the
      // session so neither the old nor partially attached run can reuse it.
      await quarantineRetainedSession(
        options.existingSession,
        options.onSession,
        "native session attachment failed",
        cleanupDomain,
      );
      throw error;
    }
    session = options.existingSession;
    recovered = true;
  } else if (persistedSession) {
    // A persisted checkpoint proves that this is recovery of an existing
    // durable run. Reconcile its cursor and prove provider continuity before
    // re-opening that run in the control plane: ControlPlanePort has no
    // rollback operation if same-session recovery fails.
    const recoveryCheckpoint = persistedSession;
    const recoveryTimeoutMs = options.timeoutMs ?? 900_000;
    persistedSession = await runAbortableOperationWithin({
      timeoutMs: recoveryTimeoutMs,
      timeoutMessage: `native session recovery replay timed out after ${recoveryTimeoutMs}ms`,
      operation: (signal) =>
        reconcileRecoveryCursor({
          controlPlane: options.controlPlane,
          checkpoint: recoveryCheckpoint,
          runId: input.binding.runId,
          sourceInstanceId: options.runnerInstanceId,
          signal,
        }),
    });
    reconciledRecoveryCheckpoint = persistedSession;
    const providerRecoveryCheckpoint = persistedSession;

    const replacementAllowed =
      providerRecoveryCheckpoint.providerRecoveryPolicy ===
      "allow_replacement_after_resume_failure";
    const failedProviderSession =
      providerRecoveryCheckpoint.terminal?.runTerminalState === "failed" &&
      providerRecoveryCheckpoint.semanticResult === null;
    const recovery = failedProviderSession
      ? {
          recovered: false as const,
          reason: "provider session ended with a failed terminal",
        }
      : options.backend.recoverSession
        ? await runAbortableOperationWithin({
            timeoutMs: recoveryTimeoutMs,
            timeoutMessage: `native session provider recovery timed out after ${recoveryTimeoutMs}ms`,
            operation: (signal) =>
              options.backend.recoverSession!(providerRecoveryCheckpoint, {
                signal,
              }),
            onLateResolution: async (lateRecovery) => {
              if (lateRecovery.session) {
                await disposeUnadmittedSession(
                  lateRecovery.session,
                  "native session provider recovery timed out",
                  cleanupDomain,
                );
              }
            },
          })
        : {
            recovered: false as const,
            reason: "driver does not support recovery",
          };
    if (!recovery.recovered || !recovery.session) {
      if (!replacementAllowed) {
        throw new Error(
          `native_session_recovery_failed: ${recovery.reason ?? "unknown"}`,
        );
      }
      continuityBreak = {
        reason: recovery.reason ?? "provider session is no longer recoverable",
        previousDriverSessionId: providerRecoveryCheckpoint.sessionId,
        previousProviderSessionId:
          providerRecoveryCheckpoint.providerSessionId ?? null,
      };
      const replacementInput = {
        identity,
        workingDirectory: input.workspace.cwd,
      };
      session = await runAbortableOperationWithin({
        timeoutMs: recoveryTimeoutMs,
        timeoutMessage: `native session replacement bootstrap timed out after ${recoveryTimeoutMs}ms`,
        operation: (signal) => {
          const abortableReplacementInput = { ...replacementInput, signal };
          return options.backend.openReplacementSession
            ? options.backend.openReplacementSession(
                abortableReplacementInput,
                providerRecoveryCheckpoint,
              )
            : options.backend.openSession(abortableReplacementInput);
        },
        onLateResolution: (lateSession) =>
          disposeUnadmittedSession(
            lateSession,
            "native session replacement bootstrap timed out",
            cleanupDomain,
          ),
      });
    } else {
      session = recovery.session;
      recovered = true;
    }
  }
  if (session === null) {
    // A fresh provider session is part of admission. Prove that it exists
    // before opening durable run state because ControlPlanePort intentionally
    // exposes no rollback for an admitted run.
    const bootstrapTimeoutMs = options.timeoutMs ?? 900_000;
    const bootstrapInput = {
      identity,
      workingDirectory: input.workspace.cwd,
    };
    session = await runAbortableOperationWithin({
      timeoutMs: bootstrapTimeoutMs,
      timeoutMessage: `native session bootstrap timed out after ${bootstrapTimeoutMs}ms`,
      operation: (signal) =>
        options.backend.openSession({
          ...bootstrapInput,
          signal,
        }),
      onLateResolution: (lateSession) =>
        disposeUnadmittedSession(
          lateSession,
          "native session bootstrap timed out",
          cleanupDomain,
        ),
    });
  }
  try {
    await options.controlPlane.openRun({
      identity,
      backendKind: descriptor.kind,
      sourceInstanceId: options.runnerInstanceId,
    });
  } catch (error) {
    if (session) {
      // Attachment and recovery both establish provider-side authority before
      // durable admission. If admission then fails, the prepared session must
      // not remain available for reuse.
      await quarantineRetainedSession(
        session,
        options.onSession,
        "native control-plane run admission failed",
        cleanupDomain,
      );
    }
    throw error;
  }
  let sessionClosePromise: Promise<void> | null = null;
  let sessionQuarantined = false;
  const quarantineSession = () => {
    if (sessionQuarantined) return;
    sessionQuarantined = true;
    try {
      options.onSession?.(null);
    } catch {
      // Owner notification cannot prevent provider cleanup.
    }
  };
  let failedCleanupDeferred = false;
  let sessionCloseRecoveryPromise: Promise<void> | null = null;
  const retainFailedCleanup = (cleanup: Promise<void>) => {
    failedCleanupDeferred = true;
    quarantineSession();
    retainFailedSessionCleanupOwner(cleanup, cleanupDomain);
  };
  const startSessionClose = (reason: string) => {
    const attempt = session.close({ reason });
    sessionClosePromise = attempt;
    return attempt;
  };
  const closeSession = () => {
    if (sessionClosePromise === null) {
      quarantineSession();
      const firstAttempt = startSessionClose(
        "native session execution complete",
      );
      // Preserve the first close outcome for its caller. If an exact attempt
      // fails, retain one ordered, delay-bounded production recovery within a
      // finite retry budget. A still-pending attempt is never overlapped or
      // replaced, and repeated terminal failure cannot create an immortal loop.
      const recovery = (async () => {
        let attempt = firstAttempt;
        let retryCount = 0;
        while (true) {
          try {
            await attempt;
            return;
          } catch (error) {
            if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) {
              quarantineSessionCleanup(session, cleanupDomain);
              throw error;
            }
            retryCount += 1;
            await waitForSessionCloseRetry();
            if (sessionClosePromise === attempt) {
              sessionClosePromise = null;
            }
            attempt = startSessionClose(
              `native session cleanup recovery after close failure (${retryCount})`,
            );
          }
        }
      })();
      sessionCloseRecoveryPromise = recovery;
      retainFailedCleanup(recovery);
      void recovery
        .finally(() => {
          if (sessionCloseRecoveryPromise === recovery) {
            sessionCloseRecoveryPromise = null;
          }
        })
        .catch(() => undefined);
    }
    const activeClose = sessionClosePromise;
    if (activeClose === null) {
      throw new Error("native session cleanup lost its active close attempt");
    }
    return activeClose;
  };
  let executionSucceeded = false;
  try {
    // Ownership publication is part of the execution-owned lifetime. If the
    // callback fails, the finally block below still quarantines and closes the
    // provider session.
    options.onSession?.(session);
    const checkpointTimeoutMs =
      options.checkpointTimeoutMs ?? DEFAULT_NATIVE_CHECKPOINT_TIMEOUT_MS;
    const persistCheckpoint = (
      snapshot: PersistedNativeSession,
      externalSignal?: AbortSignal,
    ) =>
      persistCheckpointWithin({
        snapshot,
        controlPlane: options.controlPlane,
        onCheckpoint: options.onCheckpoint,
        timeoutMs: checkpointTimeoutMs,
        externalSignal,
      });
    // Cursor reconciliation is provisional until both provider recovery and
    // durable run admission succeed. Persist it only after those boundaries;
    // the execution-owned finally below quarantines the recovered session if
    // checkpoint persistence itself fails.
    if (reconciledRecoveryCheckpoint !== null) {
      await persistCheckpoint(reconciledRecoveryCheckpoint);
    }
    const checkpoint = async (signal?: AbortSignal) => {
      const snapshot = await session.snapshot(signal ? { signal } : undefined);
      signal?.throwIfAborted();
      await persistCheckpoint(snapshot, signal);
    };
    const recoveredSnapshot = await session.snapshot();
    const recoveredActiveTurnId = recovered
      ? (recoveredSnapshot.activeTurnId ?? null)
      : (persistedSession?.activeTurnId ?? null);
    const adoptedDispositionTerminal = Boolean(
      recovered &&
      recoveredSnapshot.dispositionOnlyRecoveryConsumed &&
      !recoveredActiveTurnId &&
      (recoveredSnapshot.terminalTurns?.length ?? 0) >
        (persistedSession?.terminalTurns?.length ?? 0),
    );
    if (continuityBreak) {
      await options.onContinuityBreak?.({
        ...continuityBreak,
        replacementDriverSessionId: recoveredSnapshot.sessionId,
        replacementProviderSessionId:
          recoveredSnapshot.providerSessionId ?? null,
      });
    }
    // recoverSession may adopt a provider terminal and enqueue its normalized
    // event before returning. Do not checkpoint that terminal fingerprint
    // until consumeTurn has durably appended the event: if this process dies
    // first, retaining the older checkpoint lets the next recovery adopt and
    // emit the same provider terminal again instead of reconstructing a closed
    // session with no event to finalize.
    if (!adoptedDispositionTerminal) {
      await persistCheckpoint(recoveredSnapshot);
    }

    let consumed = {
      event: null as PrpEvent | null,
      eventCount: 0,
      highestContiguousSourceSeq: 0,
      governedResult: null as PrpStructuredRunResult | null,
    };
    const completionSnapshot =
      recoveredSnapshot.semanticResult && recoveredSnapshot.terminal
        ? recoveredSnapshot
        : persistedSession;
    let completed =
      completionSnapshot?.semanticResult && completionSnapshot.terminal
        ? {
            result: completionSnapshot.semanticResult,
            terminal: completionSnapshot.terminal,
            turnId:
              completionSnapshot.activeTurnId ??
              completedSemanticResultTurnId(completionSnapshot),
          }
        : null;
    if (!completed) {
      // A recovered driver is authoritative about whether a provider turn is
      // still active. In particular, drivers normalize the checkpoint race
      // where a terminal fingerprint was persisted before activeTurnId was
      // cleared. Falling back to the older control-plane checkpoint here
      // resurrects that terminal turn and waits forever for an event that was
      // already consumed.
      const dispositionRecoveryWasSubmitted = Boolean(
        recovered &&
        persistedSession?.dispositionOnlyRecoveryConsumed &&
        !recoveredSnapshot.semanticResult &&
        !recoveredActiveTurnId,
      );
      const dispositionRecoveryTurnId =
        persistedSession?.dispositionOnlyRecoveryTurnId ??
        recoveredSnapshot.dispositionOnlyRecoveryTurnId ??
        null;
      const recoveredDispositionTurnObserved = Boolean(
        dispositionRecoveryTurnId &&
        recoveredSnapshot.terminalTurns?.some(
          (terminal) => terminal.turnId === dispositionRecoveryTurnId,
        ),
      );
      const dispositionRecoveryStillOwned = Boolean(
        dispositionRecoveryWasSubmitted &&
        recoveredSnapshot.dispositionOnlyRecoveryConsumed &&
        dispositionRecoveryTurnId !== null &&
        recoveredDispositionTurnObserved,
      );
      const replayedDisposition =
        dispositionRecoveryWasSubmitted && dispositionRecoveryTurnId !== null
          ? await replayCheckpointedTurnTerminal({
              controlPlane: options.controlPlane,
              runId: input.binding.runId,
              sourceInstanceId: options.runnerInstanceId,
              priorTerminalTurnIds: (persistedSession?.terminalTurns ?? []).map(
                (terminal) => terminal.turnId,
              ),
              expectedTurnId: dispositionRecoveryTurnId,
            })
          : null;
      // Durable replay remains authoritative. If it has no terminal, an exact
      // consumed marker bound to the same terminal fingerprint in both
      // checkpoints proves provider completion without reconstructing provider
      // output. Give only that non-provider fact to control-plane policy; a
      // mismatch fails closed and a null policy result fails finalization.
      const dispositionFallback =
        dispositionRecoveryWasSubmitted &&
        replayedDisposition === null &&
        persistedSession
          ? checkpointedResultlessDispositionFallback({
              persisted: persistedSession,
              recovered: recoveredSnapshot,
              controlPlaneInstanceId: options.controlPlaneInstanceId,
            })
          : null;
      const checkpointedDispositionTerminal =
        replayedDisposition !== null || dispositionFallback !== null;
      const recoveryTerminal =
        replayedDisposition?.terminal ?? dispositionFallback;
      const consumptionAbort = new AbortController();
      const consuming =
        recoveryTerminal === null
          ? consumeTurn(
              session,
              options.controlPlane,
              options.timeoutMs ?? 900_000,
              options.runtimeInputLiveWindowMs ??
                DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS,
              options.semanticResultTerminalGraceMs ??
                DEFAULT_NATIVE_SEMANTIC_RESULT_TERMINAL_GRACE_MS,
              closeSession,
              quarantineSession,
              options.resolveGovernedWait,
              consumptionAbort.signal,
            )
          : Promise.resolve({
              event: recoveryTerminal,
              eventCount: 0,
              highestContiguousSourceSeq:
                replayedDisposition === null ? 0 : recoveryTerminal.sourceSeq,
              governedResult: null,
            });
      // Event consumption must begin before startTurn so an eager provider cannot
      // outrun us. Observe its rejection immediately, though: if startTurn or
      // checkpointing fails first, the outer finally closes the session and the
      // abandoned consumer will reject when its stream closes. Without a handler
      // that later rejection becomes process-fatal under Node's strict policy.
      void consuming.catch(() => undefined);
      try {
        if (
          !recovered ||
          (!recoveredActiveTurnId &&
            !adoptedDispositionTerminal &&
            !checkpointedDispositionTerminal &&
            !dispositionRecoveryStillOwned)
        ) {
          const modelEnvelope = buildNativeModelEnvelope(input);
          const dispositionOnlyRecovery = Boolean(
            recovered &&
            !recoveredSnapshot.semanticResult &&
            (persistedSession?.terminalTurns?.length ?? 0) > 0 &&
            !recoveredActiveTurnId,
          );
          if (dispositionOnlyRecovery) {
            modelEnvelope.task.prompt = [
              "Paperclip semantic-result recovery for a prior completed provider turn.",
              "The prior turn already performed the work and its user-facing final answer is recorded.",
              "Do not repeat implementation, tests, research, or the final answer.",
              "Use the existing session context to invoke exactly one paperclip_finish or paperclip_block with the accurate current disposition, then stop without additional user-facing prose.",
            ].join("\n");
          }
          await session.startTurn({
            message: { role: "user", text: JSON.stringify(modelEnvelope) },
            requestedCollaborationMode:
              "executionMode" in input ? input.executionMode : "default",
          });
          await checkpoint();
        }
      } catch (error) {
        // Consumption starts before provider launch so eager events cannot be
        // lost. If launch or its checkpoint fails, abort any in-flight append
        // before joining cleanup so the failed turn cannot commit late or
        // strand execution on a never-settling durability call.
        consumptionAbort.abort(error);
        await consuming.catch(() => undefined);
        throw error;
      }
      const terminalEvent = await consuming;
      consumed = terminalEvent;
    }
    const finalizationTimeoutMs = options.timeoutMs ?? 900_000;
    const preparedFinalization = await finalizeWithin({
      timeoutMs: finalizationTimeoutMs,
      operation: async (signal) => {
        let settledCompletion = completed;
        if (settledCompletion === null) {
          const terminalEvent = consumed.event;
          if (terminalEvent === null) {
            throw new Error(
              "native_finalization_missing: session returned no terminal event",
            );
          }
          settledCompletion =
            consumed.governedResult === null
              ? await session.result()
              : {
                  result: consumed.governedResult,
                  terminal: {
                    schema: "paperclip.prp.terminal.v1",
                    turnTerminalState: "completed",
                    runTerminalState: "succeeded",
                    reportedWorkDisposition:
                      consumed.governedResult.reportedWorkDisposition,
                  },
                  turnId: terminalEvent.turnId ?? null,
                };
          signal.throwIfAborted();
          if (settledCompletion === null && options.resolveMissingResult) {
            const recoveredResult = await options.resolveMissingResult({
              turnId: terminalEvent.turnId ?? null,
              terminalEvent,
            });
            signal.throwIfAborted();
            if (recoveredResult !== null) {
              settledCompletion = {
                result: recoveredResult,
                terminal: terminalFromEvent(
                  terminalEvent,
                  recoveredResult.reportedWorkDisposition,
                ),
                turnId: terminalEvent.turnId ?? null,
              };
            }
          }
          // Do not publish the resolved result to the retryable phase until its
          // checkpoint has settled. A timed-out checkpoint therefore cannot be
          // skipped by a second attempt.
          await checkpoint(signal);
          signal.throwIfAborted();
          completed = settledCompletion;
        }
        if (settledCompletion === null) {
          throw new Error(
            "native_finalization_missing: session returned no semantic result",
          );
        }
        let terminal: PrpTerminalState;
        if (consumed.governedResult !== null) {
          terminal = settledCompletion.terminal;
        } else if (
          completionSnapshot?.semanticResult &&
          completionSnapshot.terminal
        ) {
          terminal = completionSnapshot.terminal;
        } else {
          terminal = terminalFromEvent(
            consumed.event!,
            settledCompletion.result.reportedWorkDisposition,
          );
        }
        const eventTurnId =
          settledCompletion.turnId ??
          persistedSession?.activeTurnId ??
          persistedSession?.terminalTurns?.at(-1)?.turnId ??
          consumed.event?.turnId;
        const controlEvent = (
          sourceSeq: number,
          eventType: PrpEvent["eventType"],
          payload: Record<string, unknown>,
        ): PrpEvent => ({
          schema: "paperclip.prp.event.v1",
          sourceEventId: `${options.controlPlaneInstanceId}:${input.binding.runId}:${sourceSeq}`,
          sourceSeq,
          sourceInstanceId: options.controlPlaneInstanceId,
          sourceKind: "control_plane",
          runId: input.binding.runId,
          normalizedSessionId,
          ...(eventTurnId ? { turnId: eventTurnId } : {}),
          eventType,
          schemaVersion: 1,
          priority: 0,
          emittedAt: new Date().toISOString(),
          payload,
        });
        return {
          completed: settledCompletion,
          terminal,
          expectedControlEvents: [
            controlEvent(1, "run.result.accepted", {
              result: settledCompletion.result,
            }),
            controlEvent(
              2,
              "run.terminal",
              terminal as unknown as Record<string, unknown>,
            ),
          ],
        };
      },
    });
    const baselineControlEventSequences = new Set<number>();
    const accountedControlEventSequences = new Set<number>();
    let baselineControlReplayCaptured = false;
    const durableExecutionResult = await finalizeIdempotentControlPlaneWithin({
      timeoutMs: finalizationTimeoutMs,
      operation: async (signal) => {
        const controlReplay = await options.controlPlane.replayEvents(
          {
            runId: input.binding.runId,
            sourceInstanceId: options.controlPlaneInstanceId,
            afterSourceSeq: 0,
            limit: 10,
          },
          { signal },
        );
        signal.throwIfAborted();
        const replayBySequence = new Map(
          controlReplay.events.map((event) => [event.sourceSeq, event]),
        );
        for (const existing of controlReplay.events) {
          const expected =
            preparedFinalization.expectedControlEvents[existing.sourceSeq - 1];
          if (
            expected === undefined ||
            existing.eventType !== expected.eventType ||
            canonicalJson(existing.payload) !== canonicalJson(expected.payload)
          ) {
            throw new Error(
              `native_control_event_replay_conflict:${existing.sourceSeq}`,
            );
          }
        }
        if (!baselineControlReplayCaptured) {
          for (const existing of controlReplay.events) {
            baselineControlEventSequences.add(existing.sourceSeq);
          }
          baselineControlReplayCaptured = true;
        } else {
          for (const existing of controlReplay.events) {
            if (
              !baselineControlEventSequences.has(existing.sourceSeq) &&
              !accountedControlEventSequences.has(existing.sourceSeq)
            ) {
              accountedControlEventSequences.add(existing.sourceSeq);
              consumed.eventCount += 1;
            }
          }
        }
        consumed.highestContiguousSourceSeq = Math.max(
          consumed.highestContiguousSourceSeq,
          controlReplay.highestContiguousSourceSeq,
        );
        for (const event of preparedFinalization.expectedControlEvents) {
          if (replayBySequence.has(event.sourceSeq)) continue;
          const receipt = await options.controlPlane.appendEvent(event, {
            signal,
          });
          signal.throwIfAborted();
          if (
            receipt.disposition === "committed" &&
            !accountedControlEventSequences.has(event.sourceSeq)
          ) {
            accountedControlEventSequences.add(event.sourceSeq);
            consumed.eventCount += 1;
          }
          consumed.highestContiguousSourceSeq = Math.max(
            consumed.highestContiguousSourceSeq,
            receipt.highestContiguousSourceSeq,
          );
        }
        await options.controlPlane.completeRun(
          {
            result: preparedFinalization.completed.result,
            terminal: preparedFinalization.terminal,
            turnId: preparedFinalization.completed.turnId,
            callerResultId: `${options.runnerInstanceId}:${input.binding.runId}:result`,
            callerDedupeKey: `${input.binding.runId}:${input.completionContract.sha256}`,
          },
          { signal },
        );
        return {
          result: preparedFinalization.completed.result,
          terminal: preparedFinalization.terminal,
          turnId: preparedFinalization.completed.turnId,
          normalizedSessionId,
          driverKind: descriptor.name,
          nativeEventCount: consumed.eventCount,
          highestContiguousSourceSeq: consumed.highestContiguousSourceSeq,
        };
      },
    });
    let enrichment: {
      providerSessionId: string | null;
      driverVersion: string;
      usage: Record<string, unknown> | null;
    };
    try {
      enrichment = await finalizeWithin({
        timeoutMs: options.timeoutMs ?? 900_000,
        operation: async (signal) => {
          const snapshot = await session.snapshot({ signal });
          signal.throwIfAborted();
          const completedSnapshot = {
            ...snapshot,
            semanticResult: durableExecutionResult.result,
            terminal: durableExecutionResult.terminal,
          };
          await persistCheckpoint(completedSnapshot, signal);
          signal.throwIfAborted();
          const usage = (await session.usage?.()) ?? null;
          signal.throwIfAborted();
          return {
            providerSessionId: snapshot.providerSessionId ?? null,
            driverVersion:
              typeof usage?.driverVersion === "string"
                ? usage.driverVersion
                : descriptor.version,
            usage,
          };
        },
      });
    } catch {
      // completeRun is the durable commit boundary. Snapshot/checkpoint/usage
      // enrichment cannot revoke that success, but a session whose final
      // checkpoint is unknown must not remain available for reuse.
      void closeSession().catch(() => undefined);
      enrichment = {
        providerSessionId: recoveredSnapshot.providerSessionId ?? null,
        driverVersion: descriptor.version,
        usage: null,
      };
    }
    executionSucceeded = true;
    return { ...durableExecutionResult, ...enrichment };
  } finally {
    if (
      (!options.keepSessionOpen || !executionSucceeded || sessionQuarantined) &&
      !failedCleanupDeferred
    ) {
      // A provider that ignores close must not keep execution pending forever.
      // closeSession removes it from the caller before invoking the backend;
      // retain observation of the promise, but bound the final join. Provider
      // cleanup cannot reverse a result the control plane already committed;
      // after that durable boundary the session remains unavailable for reuse
      // and late close rejection stays observed without contradicting success.
      const closeSettlement = Promise.allSettled([closeSession()]);
      await settlesWithin(
        closeSettlement,
        FAILED_OPERATION_SETTLEMENT_GRACE_MS,
      );
    }
  }
}

function canonicalJson(value: unknown): string {
  if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`;
  if (typeof value === "object" && value !== null) {
    const record = value as Record<string, unknown>;
    return `{${Object.keys(record)
      .sort()
      .map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`)
      .join(",")}}`;
  }
  return JSON.stringify(value) ?? "undefined";
}

function completedSemanticResultTurnId(
  snapshot: PersistedNativeSession,
): string | null {
  if (snapshot.semanticResult === undefined || snapshot.semanticResult === null) {
    return null;
  }
  const semanticFingerprint = canonicalJson(snapshot.semanticResult);
  for (const terminal of [...(snapshot.terminalTurns ?? [])].reverse()) {
    try {
      const value: unknown = JSON.parse(terminal.fingerprint);
      if (
        typeof value === "object"
        && value !== null
        && !Array.isArray(value)
        && (value as Record<string, unknown>).status === "completed"
        && (value as Record<string, unknown>).semanticResult
          === semanticFingerprint
      ) {
        return terminal.turnId;
      }
    } catch {
      // Legacy terminal fingerprints cannot prove semantic-result ownership.
    }
  }
  return null;
}