PaperClipAI/packages/paperclip-runner/src/native-session-runtime.ts

import { randomUUID } from "node:crypto";
import { nativeSystemInstructions } from "./backends/runtime-context.js";

import type {
  CheckpointControlPlaneSessionOptions,
  ControlPlanePort,
} from "./contracts/control-plane-port.js";
import type {
  NativeExecutionInput,
  NativeSessionExecutionResult,
} from "./contracts/native-execution.js";
import { readProviderFailure } from "./contracts/provider-failure.generated.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 {
  PROVIDER_ATTACH_REFUSED,
  PROVIDER_SIDE_EFFECT_PENDING,
  PROVIDER_STATE_INCOMPATIBLE,
  type HarnessThreadGoal,
} from "./contracts/harness-driver.js";
import {
  NativeProviderSideEffectPendingError,
  NativeProviderStateIncompatibleError,
  NativeProviderTerminalFailure,
  NativeSessionCloseUnrecoverableError,
  NativeSessionCleanupQuarantinedError,
  NativeSessionProtocolIntegrityError,
  isNativeRestartInterruption,
  nativeRestartContinuationInFlight,
  nativeRestartInterruptedTurnId,
} 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";
import {
  retainedRunnerdCleanupProofIsCurrent,
  type RetainedRunnerdCleanupProof,
} from "./live/runnerd-codex-transport.js";

export const DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS = 120_000;
const OPTIONAL_SESSION_CANCELLATION_GRACE_MS = 100;
const FAILED_OPERATION_SETTLEMENT_GRACE_MS = 100;
// Retaining a remote provider requires terminal subscription teardown to be
// acknowledged before the session is handed to another run. Daytona command
// round trips routinely exceed the generic failed-operation grace, but remain
// bounded by the transport. Other iterator and handoff cleanup keeps the short
// fail-closed quarantine boundary below.
const REUSABLE_SESSION_CANCELLATION_SETTLEMENT_GRACE_MS = 10_000;
// Governed cancellation can traverse the ACPX turn cancellation (2s), host
// cancellation (2s), protocol/TERM/KILL shutdown (7s), and final status read
// (1s). Allow that bounded settlement plus transport scheduling; never treat
// elapsed time or a cancellation acknowledgement as usage/terminal evidence.
const GOVERNED_ACCOUNTING_DRAIN_MS = 15_000;
const MAX_GOVERNED_ACCOUNTING_DRAIN_MS = 30_000;
const DEFAULT_NATIVE_CHECKPOINT_TIMEOUT_MS = 30_000;
/**
 * A retained cleanup owner can only still hold resources it was bound to.
 * `key` names the backend (company, backend kind/name, optional remote
 * resource). `resources` names the task and provider conversation the owner
 * may still be writing: its issue and normalized session. Admission is gated
 * only when both the backend and one of those resources match, so one
 * session's unrecoverable close cannot block unrelated tasks in the company.
 */
interface NativeSessionCleanupDomain {
  key: string;
  resources: readonly string[];
}

function nativeSessionCleanupScopeBlocks(
  owner: NativeSessionCleanupDomain,
  admission: NativeSessionCleanupDomain,
): boolean {
  return (
    owner === admission ||
    (owner.key === admission.key &&
      owner.resources.some((resource) => admission.resources.includes(resource)))
  );
}

const failedSessionCleanupOwners = new Map<
  Promise<void>,
  NativeSessionCleanupDomain
>();

type NativeSessionCleanupSettledListener = (scope: {
  companyId: string;
  issueIds: string[];
}) => void;
const cleanupSettledListeners = new Set<NativeSessionCleanupSettledListener>();

function cleanupDomainScope(domain: NativeSessionCleanupDomain): {
  companyId: string;
  issueIds: string[];
} {
  const [companyId] = JSON.parse(domain.key) as string[];
  const issueIds = domain.resources.flatMap((resource) => {
    const [kind, id] = JSON.parse(resource) as string[];
    return kind === "issue" && typeof id === "string" ? [id] : [];
  });
  return { companyId: companyId ?? "", issueIds };
}

/** A retained cleanup owner or quarantine entry was released. Observers can
 * re-evaluate work that waited on it; this grants no execution authority. */
function announceCleanupSettled(domain: NativeSessionCleanupDomain): void {
  const scope = cleanupDomainScope(domain);
  for (const listener of cleanupSettledListeners) {
    try {
      listener(structuredClone(scope));
    } catch {
      // Observers cannot change cleanup ownership.
    }
  }
}

/** Subscribe to cleanup releases. Returns the unsubscribe function. */
export function onNativeSessionCleanupSettled(
  listener: NativeSessionCleanupSettledListener,
): () => void {
  cleanupSettledListeners.add(listener);
  return () => cleanupSettledListeners.delete(listener);
}

/**
 * Read-only view of the in-process admission gate for one task. "pending":
 * a retained close or bounded automatic close recovery still owns the task's
 * provider resources, and admission would be refused now. "operator_required":
 * the close cannot settle automatically. Never grants or retires authority.
 */
export function nativeSessionCleanupBlock(input: {
  companyId: string;
  issueId: string;
}): "operator_required" | "pending" | null {
  const issueResource = JSON.stringify(["issue", input.issueId]);
  const matches = (domain: NativeSessionCleanupDomain) =>
    domain.resources.includes(issueResource) &&
    cleanupDomainScope(domain).companyId === input.companyId;
  let block: "operator_required" | "pending" | null = null;
  for (const entry of quarantinedSessionCleanups) {
    if (!matches(entry.domain)) continue;
    // An entry whose bounded automatic close budget is spent has no further
    // scheduled attempt. Waiting would never end, so it needs the same
    // stopped-process recovery as an unrecoverable close.
    const automaticRecoveryExhausted =
      !entry.attempt && !entry.recovery && !entry.timer &&
      entry.automaticAttempts >= MAX_QUARANTINED_SESSION_AUTOMATIC_CLOSE_ATTEMPTS;
    if (entry.operatorRecoveryRequired || automaticRecoveryExhausted)
      return "operator_required";
    block = "pending";
  }
  for (const domain of failedSessionCleanupOwners.values()) {
    if (matches(domain)) block = "pending";
  }
  return block;
}
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;
  operatorRecoveryRequired: boolean;
}

const quarantinedSessionCleanups = new Set<QuarantinedSessionCleanup>();
const sessionOriginRunnerInstances = new WeakMap<NativeSession, string>();

/** Retire only the exact owner whose separate authenticated cleanup completed.
 * The rejected close promise remains rejected; this neither resets a session
 * nor authorizes an execution. Other quarantined owners remain admission gates. */
export function completeRetainedNativeSessionCleanup(
  proof: RetainedRunnerdCleanupProof,
): number {
  if (!retainedRunnerdCleanupProofIsCurrent(proof))
    throw new NativeSessionCleanupQuarantinedError();
  const domain = JSON.stringify([
    proof.binding.companyId,
    proof.backend.kind,
    proof.backend.name,
  ]);
  const matches = [...quarantinedSessionCleanups].filter((entry) => {
    const identity = entry.session.identity();
    return (
      entry.domain.key === domain &&
      Object.entries(proof.binding).every(
        ([key, value]) => identity[key as keyof typeof identity] === value,
      )
    );
  });
  if (
    matches.length > 1 ||
    matches.some(
      (entry) =>
        sessionOriginRunnerInstances.get(entry.session) !==
          proof.identity.runnerInstanceId ||
        !entry.operatorRecoveryRequired ||
        entry.attempt ||
        entry.recovery ||
        entry.timer,
    )
  ) {
    throw new NativeSessionCleanupQuarantinedError();
  }
  for (const entry of matches) {
    quarantinedSessionCleanups.delete(entry);
    announceCleanupSettled(entry.domain);
  }
  return matches.length;
}

/** Control-plane-only cleanup boundary after the environment provider confirmed
 * termination of the exact remote resource for this run. This retires process
 * ownership, not checkpoints, action outcomes, or authorization to run again.
 * An in-flight close must settle first: it must never reach a reused sandbox.
 */
export function completeTerminatedRemoteNativeSessionCleanup(binding: {
  companyId: string;
  runId: string;
  remoteCleanupScope: string;
}): boolean {
  if (!binding.remoteCleanupScope) return false;
  const matches = [...quarantinedSessionCleanups].filter(({ session, domain }) => {
    const identity = session.identity();
    // Domains are created internally from company, backend kind/name, and the
    // optional remote resource. Local domains have no fourth element.
    const [, , , remoteCleanupScope] = JSON.parse(domain.key) as string[];
    return identity.companyId === binding.companyId && identity.runId === binding.runId &&
      remoteCleanupScope === binding.remoteCleanupScope;
  });
  if (matches.some(entry => entry.attempt || entry.recovery)) return false;
  for (const entry of matches) {
    if (entry.timer) clearTimeout(entry.timer);
    entry.timer = null;
    quarantinedSessionCleanups.delete(entry);
    announceCleanupSettled(entry.domain);
  }
  return true;
}

/** Host-only counterpart of remote resource termination. The caller has verified
 * both exact local process identities and durably fenced their run. This removes
 * only cleanup ownership; the retained checkpoint is never made resumable.
 */
export function completeTerminatedLocalNativeSessionCleanup(binding: {
  companyId: string;
  runId: string;
  runnerInstanceId: string;
}): boolean {
  const matches = [...quarantinedSessionCleanups].filter(({ session, domain }) => {
    const identity = session.identity();
    const parts = JSON.parse(domain.key) as string[];
    return parts.length === 3 && identity.companyId === binding.companyId && identity.runId === binding.runId;
  });
  if (matches.some(entry => entry.attempt || entry.recovery ||
      sessionOriginRunnerInstances.get(entry.session) !== binding.runnerInstanceId)) return false;
  for (const entry of matches) {
    if (entry.timer) clearTimeout(entry.timer);
    quarantinedSessionCleanups.delete(entry);
    announceCleanupSettled(entry.domain);
  }
  return true;
}

export interface NativeSessionGoalControl {
  requestId: string;
  action: "create" | "edit" | "replace" | "pause" | "resume" | "clear";
  objective?: string;
  tokenBudget?: number | null;
}

export interface ExecuteNativeSessionOptions {
  /** The control plane proved the old local runner stopped and charged a recovery attempt. */
  resumeInterruptedTurn?: boolean;
  /** Read bounded task history only after recovery requires a fresh conversation. */
  getFreshSessionHandoff?: () => Promise<string | null>;
  /**
   * Seals this run's first-turn input immediately before `turn.start` and
   * returns user messages that arrived while the run was preparing. Repeated
   * calls for the same run return the same messages.
   */
  getFirstTurnQueuedInput?: () => Promise<string | null>;
  /**
   * Steered messages the provider acknowledged but the model never read, read
   * when a restart continuation starts. The provider lost them with the
   * interrupted turn, so the continuation delivers them.
   */
  getUnreadSteeredMessages?: () => Promise<Array<{ id: string; text: string }>>;
  /** Durable launch intent, after cleanup admission and before provider calls. */
  onSessionAdmission?: () => Promise<void>;
  input: NativeExecutionInput;
  backend: NativeSessionBackend;
  controlPlane: ControlPlanePort;
  runnerInstanceId: string;
  controlPlaneInstanceId: string;
  /** Trusted provider resource identity: independent remote sandboxes must not
   * inherit each other's process-cleanup gates. Omit for local backends. */
  remoteCleanupScope?: string;
  /** Operation bound; explicit values also preserve the legacy turn bound. */
  timeoutMs?: number;
  /** Total turn duration. Zero or no configured bound allows long-running work. */
  turnTimeoutMs?: number;
  /** Abort admission while waiting for prior cleanup in the same domain. */
  signal?: AbortSignal;
  startup?: { deadlineAtMs: number; 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;
  onSession?: (session: NativeSession | null) => void;
  /** Observes why a retained session was removed from warm reuse. */
  onSessionQuarantined?: (reason: string) => Promise<void> | void;
  existingSession?: NativeSession;
  persistedSession?: PersistedNativeSession | null;
  keepSessionOpen?: boolean;
  /** Apply a structured session-goal control instead of starting an ordinary turn. */
  sessionGoalControl?: NativeSessionGoalControl | null;
  /** Resume the active durable session goal after a bounded heartbeat rollover. */
  resumeSessionGoalHeartbeat?: boolean;
  /**
   * Compact the resumed provider session's context instead of starting an
   * ordinary turn. The run settles with a synthesized result once the
   * provider's compaction turn completes; it does no task work.
   */
  sessionCompaction?: { requestId: string; contextTokensBefore?: number } | null;
  /**
   * Wait for the backend's close contract before returning a durable result.
   * Use this only for backends whose close path is internally bounded and
   * carries required persistence (for example, a remote runner checkpoint).
   */
  requireSessionCloseBeforeReturn?: boolean;
  /** Trusted cleanup observer, called only after the owned close completes.
   * Requires requireSessionCloseBeforeReturn; never called for a live warm session. */
  onSessionClosed?: () => Promise<void>;
  onCheckpoint?: (
    snapshot: PersistedNativeSession,
    options?: CheckpointControlPlaneSessionOptions,
  ) => Promise<void> | void;
  /**
   * Observes a post-result failure after completeRun has committed. Checkpoint
   * failures quarantine the retained session; usage failures only omit
   * optional accounting enrichment.
   */
  onPostCompletionEnrichmentFailure?: (input: {
    stage: "checkpoint" | "usage";
    error: unknown;
  }) => 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;
  signal?: AbortSignal;
}): Promise<T> {
  input.signal?.throwIfAborted();
  if (input.timeoutMs <= 0) throw input.timeoutError?.() ?? new Error(input.timeoutMessage);
  const operationAbort = new AbortController();
  let timer: ReturnType<typeof setTimeout> | undefined;
  let timedOut = false;
  let timeoutError: Error | undefined;
  let removeAbort = () => {};
  const cancelled = input.signal ? new Promise<never>((_resolve, reject) => {
    const abort = () => {
      timedOut = true;
      timeoutError = input.signal!.reason instanceof Error ? input.signal!.reason : new Error("native session startup cancelled");
      reject(timeoutError);
      operationAbort.abort(timeoutError);
    };
    input.signal!.addEventListener("abort", abort, { once: true });
    removeAbort = () => input.signal!.removeEventListener("abort", abort);
  }) : undefined;
  const operation = Promise.resolve().then(() => {
    input.signal?.throwIfAborted();
    return 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,
      ...(cancelled ? [cancelled] : []),
      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);
    removeAbort();
  }
}

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 (error instanceof NativeSessionCloseUnrecoverableError) {
          quarantineSessionCleanup(session, cleanupDomain, true, error);
          throw error;
        }
        if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) {
          quarantineSessionCleanup(session, cleanupDomain, false, error);
          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);
      announceCleanupSettled(cleanupDomain);
    })
    .catch(() => undefined);
}

export interface NativeSessionCleanupQuarantine {
  companyId: string;
  runId: string;
  issueId: string;
  sessionId: string;
  operatorRecoveryRequired: boolean;
  /** Bounded error name/code/message of the close failure that caused it. */
  cause: string;
  /** Bounded settlement facts of an unrecoverable close, when it reported them. */
  settlement: Record<string, unknown> | null;
}
type NativeSessionCleanupQuarantineListener = (
  quarantine: NativeSessionCleanupQuarantine,
) => void;
const cleanupQuarantineListeners = new Set<NativeSessionCleanupQuarantineListener>();

/** Observe every cleanup quarantine when it is created or escalates to
 * operator recovery, with its cause. Observation grants no authority. */
export function onNativeSessionCleanupQuarantined(
  listener: NativeSessionCleanupQuarantineListener,
): () => void {
  cleanupQuarantineListeners.add(listener);
  return () => cleanupQuarantineListeners.delete(listener);
}

function boundedSettlement(cause: unknown): Record<string, unknown> | null {
  if (!(cause instanceof NativeSessionCloseUnrecoverableError) || !cause.settlement) return null;
  try {
    const text = JSON.stringify(cause.settlement);
    return text.length <= 4_000 ? (JSON.parse(text) as Record<string, unknown>) : { truncated: true };
  } catch {
    return { unreadable: true };
  }
}

function announceCleanupQuarantine(
  session: NativeSession,
  operatorRecoveryRequired: boolean,
  cause: unknown,
): void {
  if (cleanupQuarantineListeners.size === 0) return;
  let identity: ReturnType<NativeSession["identity"]> | null = null;
  try {
    identity = session.identity();
  } catch {
    // Observation must not fail quarantine.
  }
  const error = cause instanceof Error ? cause : null;
  const code = error && "code" in error && typeof error.code === "string" ? error.code : null;
  const quarantine: NativeSessionCleanupQuarantine = {
    companyId: identity?.companyId ?? "",
    runId: identity?.runId ?? "",
    issueId: identity?.issueId ?? "",
    sessionId: identity?.sessionId ?? "",
    operatorRecoveryRequired,
    cause: [error?.name, code, error?.message ?? (cause === undefined ? null : String(cause))]
      .filter((part): part is string => Boolean(part)).join(": ").slice(0, 1_000) || "unknown",
    settlement: boundedSettlement(cause),
  };
  for (const listener of cleanupQuarantineListeners) {
    try {
      listener(structuredClone(quarantine));
    } catch {
      // Observers cannot change cleanup ownership.
    }
  }
}

function quarantineSessionCleanup(
  session: NativeSession,
  cleanupDomain: NativeSessionCleanupDomain,
  operatorRecoveryRequired = false,
  cause?: unknown,
): void {
  const existing = [...quarantinedSessionCleanups].find(
    (entry) => entry.session === session,
  );
  if (existing) {
    if (operatorRecoveryRequired && !existing.operatorRecoveryRequired)
      announceCleanupQuarantine(session, true, cause);
    if (operatorRecoveryRequired) {
      existing.operatorRecoveryRequired = true;
      if (existing.timer) clearTimeout(existing.timer);
      existing.timer = null;
    }
    return;
  }
  const cleanup: QuarantinedSessionCleanup = {
    session,
    domain: cleanupDomain,
    automaticAttempts: 0,
    attempt: null,
    recovery: null,
    recoveryMaxAttempts: null,
    timer: null,
    operatorRecoveryRequired,
  };
  quarantinedSessionCleanups.add(cleanup);
  announceCleanupQuarantine(session, operatorRecoveryRequired, cause);
  if (operatorRecoveryRequired) return;
  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;
  if (cleanup.operatorRecoveryRequired) return Promise.resolve();
  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) &&
      !cleanup.operatorRecoveryRequired;
      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);
        announceCleanupSettled(cleanup.domain);
      } catch (error) {
        if (error instanceof NativeSessionCloseUnrecoverableError) {
          if (!cleanup.operatorRecoveryRequired)
            announceCleanupQuarantine(cleanup.session, true, error);
          cleanup.operatorRecoveryRequired = true;
        }
        // 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);
      announceCleanupSettled(cleanup.domain);
      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.operatorRecoveryRequired ||
    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();
    if (
      [...quarantinedSessionCleanups].some(
        (cleanup) =>
          nativeSessionCleanupScopeBlocks(cleanup.domain, cleanupDomain) &&
          cleanup.operatorRecoveryRequired,
      )
    ) {
      throw new NativeSessionCleanupQuarantinedError();
    }
    const cleanupOwners = new Set<Promise<void>>(
      [...failedSessionCleanupOwners]
        .filter(([, domain]) =>
          nativeSessionCleanupScopeBlocks(domain, cleanupDomain),
        )
        .map(([owner]) => owner),
    );
    for (const cleanup of quarantinedSessionCleanups) {
      if (!nativeSessionCleanupScopeBlocks(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) =>
      nativeSessionCleanupScopeBlocks(domain, cleanupDomain),
    ) ||
    [...quarantinedSessionCleanups].some((cleanup) =>
      nativeSessionCleanupScopeBlocks(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 (!nativeSessionCleanupScopeBlocks(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,
  input: NativeExecutionInput,
  timeoutMs: number,
  runtimeInputLiveWindowMs: number,
  reusableSessionCancellationGraceMs: number,
  closeFailedSession: () => Promise<void>,
  quarantineSession: (reason: string) => void,
  resolveGovernedWait?: ExecuteNativeSessionOptions["resolveGovernedWait"],
  externalSignal?: AbortSignal,
  sessionGoal?: {
    input: NativeExecutionInput;
    requestId: string;
  },
  initialGoal?: HarnessThreadGoal | null,
  resumeInterruptedTurn?: (event: PrpEvent, signal: AbortSignal) => Promise<boolean>,
  checkpointGovernedWait?: (wait: NonNullable<PersistedNativeSession["governedWait"]>, signal: AbortSignal) => Promise<void>,
  initialGovernedWait?: PersistedNativeSession["governedWait"],
  /** Settles a compaction-only turn, which reports no semantic result. */
  compactionResult?: PrpStructuredRunResult,
  supersededTerminal?: (event: PrpEvent, signal: AbortSignal) => Promise<boolean>,
) {
  let timer: ReturnType<typeof setTimeout> | undefined;
  const appendAbort = new AbortController();
  const governedCleanupOperations = new Set<Promise<unknown>>();
  let governedCancellationCommitted = false;
  let semanticResultObserved = false;
  let governedProviderTerminalObserved = 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 semanticResultProposal: PrpStructuredRunResult | null = null;
    let sessionGoalObserved = sessionGoal !== undefined;
    let previousGoal = initialGoal ?? null;
    let goalControlObserved = false;
    let latestSessionGoal: HarnessThreadGoal | null = null;
    let resultSource: "semantic_result" | "governed_wait" | null = null;
    let providerFailure: NativeProviderTerminalFailure | null = null;
    const settleDurableResult = async (
      event: PrpEvent,
      result: PrpStructuredRunResult,
      reason: string,
    ) => {
      if (session.cancel === undefined) {
        throw new Error("native_governed_wait_cancellation_unavailable");
      }
      // Revoke new publication before any asynchronous journal write. Persist
      // the disposition before provider interruption, retaining usage and terminal
      // evidence even when the interaction is answered during the restart.
      session.revokeTurnPublication?.();
      await checkpointGovernedWait?.({ sourceEvent: event, result }, appendAbort.signal);
      appendAbort.signal.throwIfAborted();
      const cancellation = session.cancel({ reason, signal: appendAbort.signal });
      governedCancellationCommitted = true;
      const cleanup = cancellation.cleanup;
      governedCleanupOperations.add(cleanup);
      void cleanup
        .catch(() => quarantineSession("governed_cleanup_failed"))
        .finally(() => governedCleanupOperations.delete(cleanup));
      // Cancellation revokes provider work immediately, but shutdown can still
      // supply its final usage. Keep a bounded accounting-only drain before
      // acknowledging the run; otherwise the next wake sees permanent debt.
      // Never let a stuck stream or cleanup revoke the durable governed wait.
      const drainAbort = new AbortController();
      const stopDrain = () => drainAbort.abort(appendAbort.signal.reason);
      appendAbort.signal.addEventListener("abort", stopDrain, { once: true });
      const requestedDrainMs = cancellation.terminalDrainTimeoutMs;
      const terminalDrainMs = typeof requestedDrainMs === "number" && Number.isSafeInteger(requestedDrainMs)
        && requestedDrainMs > 0 && requestedDrainMs <= MAX_GOVERNED_ACCOUNTING_DRAIN_MS
        ? requestedDrainMs : GOVERNED_ACCOUNTING_DRAIN_MS;
      const drainMs = Math.min(terminalDrainMs, timeoutMs > 0 ? Math.max(1, Math.floor(timeoutMs / 4)) : terminalDrainMs);
      let drainTimer: ReturnType<typeof setTimeout> | undefined;
      const drained = (async () => {
        while (!drainAbort.signal.aborted) {
          const next = await eventIterator.next();
          if (next.done || drainAbort.signal.aborted) return;
          const trailing = next.value;
          const terminal = isTurnTerminal(trailing);
          const usage = trailing.eventType === "item.completed" && trailing.payload.kind === "usage";
          // These already-numbered cancellation facts carry no new work
          // authority. Retain them so later receipts have contiguous replay.
          const cancellation = trailing.eventType === "runtime_request.cancelled"
            || trailing.eventType === "runtime_request.expired"
            || (trailing.eventType === "item.completed" && trailing.payload.kind === "interrupt_acknowledgement")
            || (trailing.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(trailing.payload.runTerminalState)));
          if (trailing.turnId === event.turnId && (usage || terminal || cancellation)) {
            const receipt = await controlPlane.appendEvent(trailing, {
              signal: drainAbort.signal,
              ...(terminal && trailing.turnId && trailing.eventType !== "turn.failed" && trailing.payload.providerTerminalObserved !== false
                ? { deferPostCommitObservers: true as const } : {}),
            });
            if (drainAbort.signal.aborted) return;
            eventCount += receipt.disposition === "committed" ? 1 : 0;
            highestContiguousSourceSeq = Math.max(highestContiguousSourceSeq, receipt.highestContiguousSourceSeq);
            if (terminal) {
              if (trailing.eventType === "turn.failed" || trailing.payload.providerTerminalObserved === false) {
                throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false);
              }
              governedProviderTerminalObserved = true;
              return;
            }
          }
        }
      })();
      await Promise.race([
        drained.catch(error => {
          quarantineSession("governed_accounting_drain_failed");
          if (error instanceof NativeProviderTerminalFailure) throw error;
        }),
        new Promise<void>(resolve => { drainTimer = setTimeout(() => {
          drainAbort.abort(new Error("governed accounting drain timed out"));
          quarantineSession("governed_accounting_drain_timed_out");
          resolve();
        }, drainMs); }),
      ]).finally(() => {
        clearTimeout(drainTimer);
        appendAbort.signal.removeEventListener("abort", stopDrain);
        drainAbort.abort();
      });
      // A durable wait and complete usage cannot prove that the provider
      // stopped. Leave the journaled wait recoverable, without a successful
      // result or an accounting-completion boundary, if shutdown is unproven.
      if (!governedProviderTerminalObserved) {
        throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false);
      }
      return {
        event,
        eventCount,
        highestContiguousSourceSeq,
        governedResult: result,
      };
    };
    if (initialGovernedWait) {
      return settleDurableResult(initialGovernedWait.sourceEvent, initialGovernedWait.result,
        "Paperclip resumed settlement of this durable governed interaction.");
    }
    while (true) {
      const next = await eventIterator.next().catch((error) => {
        throw providerFailure ?? error;
      });
      if (stopConsumer) throw new Error("native event consumer stopped");
      if (next.done) {
        if (providerFailure) throw providerFailure;
        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 eventGoal = goalFromEvent(event);
      const goalChanged = eventGoal !== undefined &&
        goalLifecycleFingerprint(eventGoal) !== goalLifecycleFingerprint(previousGoal);
      if (eventGoal !== undefined) previousGoal = eventGoal;
      if (
        eventGoal !== undefined && eventGoal !== null &&
        (sessionGoalObserved || goalStatus(eventGoal) === "active" ||
          (event.eventType === "session.goal.updated" && goalChanged))
      ) {
        // An inactive resume snapshot describes the previous goal, not the
        // work requested by a new ordinary prompt. Only an explicit goal
        // control, an active goal, or a new lifecycle change owns this run's
        // lifetime. Resume can replay unchanged updates as well as snapshots;
        // append both for the UI without mistaking them for new goal work.
        sessionGoalObserved = true;
        latestSessionGoal = eventGoal;
        // A harness-created goal may arrive after a semantic result proposal.
        // The goal owns the durable lifetime instead of the completion proposal.
        if (resultSource === "semantic_result") {
          governedResult = null;
          resultSource = null;
        }
        if (sessionGoal === undefined) goalControlObserved = true;
      }
      if (event.eventType === "run.result.proposed") {
        const validation = validatePrpStructuredRunResult(event.payload);
        if (validation.ok) semanticResultProposal = validation.result;
      }
      if (event.eventType === "session.failed") {
        const failure = payload.error && typeof payload.error === "object"
          ? payload.error as Record<string, unknown> : payload;
        providerFailure = new NativeProviderTerminalFailure(
          typeof failure.code === "string" ? failure.code : "provider_session_failed",
          failure.recoverable === true || payload.recoverable === true,
          typeof failure.message === "string" ? failure.message : undefined,
          readProviderFailure(failure.providerFailure),
        );
        if ((receipt.disposition === "committed" || receipt.disposition === "duplicate") && !providerFailure.recoverable && failure.code === "provider_exited"
          && failure.expected === false && failure.completedByExitedProcess === false
          && failure.previousTurnCompleted === false && failure.provider === input.provider.kind
          && event.runId === input.binding.runId && event.normalizedSessionId === session.identity().sessionId) {
          throw providerFailure;
        }
      }
      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 &&
        !sessionGoalObserved &&
        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";
        semanticResultObserved = true;
      }
      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) && !sessionGoalObserved) {
        if (resultSource === "semantic_result") {
          // A completion tool reports task disposition, not provider termination.
          // Persist the final answer and authoritative terminal before finalizing.
          continue;
        }
        return settleDurableResult(
          event,
          governedResult,
          "Paperclip parked this turn on a durable governed interaction.",
        );
      }
      if (sessionGoalObserved) {
        if (isTurnTerminal(event) && providerFailure && !providerFailure.recoverable) {
          throw providerFailure;
        }
        if (sessionGoal && payload.requestId === sessionGoal.requestId) {
          goalControlObserved = true;
        }
        if (
          goalControlObserved &&
          (providerFailure === null || providerFailure.recoverable) &&
          eventGoal !== undefined &&
          goalStatus(eventGoal) !== "active" &&
          payload.workingNow !== true
        ) {
          return {
            event,
            eventCount,
            highestContiguousSourceSeq,
            governedResult:
              governedResult ??
              (goalStatus(eventGoal) === "complete" ? semanticResultProposal : null) ??
              sessionGoalResult(
                input,
                eventGoal,
                `Provider session goal settled as ${goalStatus(eventGoal) ?? "cleared"}.`,
              ),
          };
        }
        if (isTurnTerminal(event)) {
          if (
            latestSessionGoal !== null &&
            goalStatus(latestSessionGoal) !== "active"
          ) {
            return {
              event,
              eventCount,
              highestContiguousSourceSeq,
              governedResult:
                governedResult ??
                (goalStatus(latestSessionGoal) === "complete" ? semanticResultProposal : null) ??
                sessionGoalResult(
                  input,
                  latestSessionGoal,
                  `Provider session goal settled as ${goalStatus(latestSessionGoal) ?? "cleared"}.`,
                ),
            };
          }
          if (!session.goal) throw new Error("native_session_goal_unavailable");
          const authoritativeGoal = await session.goal({ action: "get" });
          // `goal(get)` emits an authoritative goal snapshot. Keep consuming
          // until that snapshot is durably appended so the server projection
          // cannot lag behind the synthetic heartbeat result.
          if (goalStatus(authoritativeGoal) === "active") continue;
          goalControlObserved = true;
          latestSessionGoal = authoritativeGoal;
          continue;
        }
      }
      if (isTurnTerminal(event)) {
        if (await supersededTerminal?.(event, appendAbort.signal)) continue;
        if (providerFailure) throw providerFailure;
        if (compactionResult && event.eventType === "turn.completed") {
          return {
            event,
            eventCount,
            highestContiguousSourceSeq,
            governedResult: compactionResult,
          };
        }
        if (semanticResultProposal === null && governedResult === null &&
            inputTimers.size === 0 && await resumeInterruptedTurn?.(event, appendAbort.signal)) continue;
        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) => {
        if (timeoutMs <= 0) return;
        // Node timers overflow above ~24.8 days. Keep explicit long deadlines
        // in bounded chunks instead of accidentally firing them immediately.
        const deadline = Date.now() + timeoutMs;
        const checkDeadline = () => {
          const remaining = deadline - Date.now();
          if (remaining <= 0) {
            reject(new Error(`native session timed out after ${timeoutMs}ms`));
          } else {
            timer = setTimeout(checkDeadline, Math.min(remaining, 2_147_483_647));
          }
        };
        checkDeadline();
      }),
      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 reusable session needs a bounded window for its remote terminal
      // subscription to release. A governed wait qualifies only after its
      // actual provider terminal was retained during the accounting drain.
      // Missing terminal evidence and unrelated stalled cleanup retain the
      // short fail-closed boundary.
      const teardownSettled = await settlesWithin(
        passiveTeardownSettlement,
        semanticResultObserved || governedProviderTerminalObserved
          ? reusableSessionCancellationGraceMs
          : FAILED_OPERATION_SETTLEMENT_GRACE_MS,
      );
      if (!teardownSettled)
        quarantineSession("provider_event_teardown_timed_out");
    }
    if (timer !== undefined) clearTimeout(timer);
    removeExternalAbort();
  }
}

function goalLifecycleFingerprint(goal: HarnessThreadGoal | null): string {
  if (goal === null) return "cleared";
  // Provider checkpoints can use seconds while normalized events use ISO
  // timestamps (parsed as milliseconds). Usage/timing updates alone must not
  // turn a completed or paused goal into ownership of an ordinary chat run.
  const createdAt = goal.createdAt < 10_000_000_000
    ? goal.createdAt * 1_000 : goal.createdAt;
  return canonicalJson({ objective: goal.objective, status: goal.status, createdAt });
}

function goalStatus(goal: HarnessThreadGoal | null):
  | "active"
  | "paused"
  | "blocked"
  | "limited"
  | "usage_limited"
  | "budget_limited"
  | "complete"
  | null {
  if (!goal) return null;
  return goal.status === "usageLimited"
    ? "usage_limited"
    : goal.status === "budgetLimited"
      ? "budget_limited"
      : goal.status;
}

function goalFromEvent(event: PrpEvent): HarnessThreadGoal | null | undefined {
  if (
    event.eventType !== "session.goal.snapshot" &&
    event.eventType !== "session.goal.updated" &&
    event.eventType !== "session.goal.cleared"
  ) return undefined;
  if (event.eventType === "session.goal.cleared") return null;
  const value = event.payload.goal;
  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
  const record = value as Record<string, unknown>;
  if (typeof record.objective !== "string" || typeof record.status !== "string") return null;
  const providerStatus = record.status === "usage_limited"
    ? "usageLimited"
    : record.status === "budget_limited"
      ? "budgetLimited"
      : record.status;
  if (!["active", "paused", "blocked", "limited", "usageLimited", "budgetLimited", "complete"].includes(providerStatus)) {
    return null;
  }
  const epoch = (value: unknown): number => {
    if (typeof value !== "string") return 0;
    const parsed = Date.parse(value);
    return Number.isFinite(parsed) ? parsed : 0;
  };
  return {
    threadId: "normalized",
    objective: record.objective,
    status: providerStatus as HarnessThreadGoal["status"],
    tokenBudget: typeof record.tokenBudget === "number" ? record.tokenBudget : null,
    tokensUsed: typeof record.tokensUsed === "number" ? record.tokensUsed : 0,
    timeUsedSeconds: typeof record.elapsedSeconds === "number" ? record.elapsedSeconds : 0,
    createdAt: epoch(record.createdAt),
    updatedAt: epoch(record.updatedAt),
  };
}

export async function applyNativeSessionGoalControl(
  session: NativeSession,
  control: NativeSessionGoalControl,
): Promise<HarnessThreadGoal | null> {
  if (!session.goal) throw new Error("native_session_goal_unavailable");
  if (control.action === "clear") {
    return session.goal({ action: "clear", requestId: control.requestId });
  }
  if (control.action === "pause" || control.action === "resume") {
    return session.goal({ action: control.action, requestId: control.requestId });
  }
  const objective = control.objective?.trim();
  if (!objective) throw new Error(`native_session_goal_${control.action}_objective_required`);
  if (control.action === "replace") {
    await session.goal({ action: "clear", requestId: control.requestId });
  }
  let status: HarnessThreadGoal["status"] = "active";
  if (control.action === "edit") {
    const current = await session.goal({ action: "get" });
    if (!current) throw new Error("native_session_goal_not_found");
    status = current.status === "complete" ? "active" : current.status;
  }
  return session.goal({
    action: "set",
    objective,
    status,
    requestId: control.requestId,
    ...(control.tokenBudget !== undefined
      ? { tokenBudget: control.tokenBudget }
      : {}),
  });
}

/**
 * Result for a compaction-only run. Compaction does no task work, so the
 * result yields back to the same agent; the server's status arbiter preserves
 * the task status and schedules nothing for a compaction run.
 */
export function sessionCompactionResult(
  input: NativeExecutionInput,
  requestId: string,
): PrpStructuredRunResult {
  const summary = "Context compacted on request.";
  return {
    schema: "paperclip.run_result.v1",
    reportedWorkDisposition: "yielded",
    summary,
    completionClaim: {
      contractRevision: input.completionContract.contract.revision,
      objectiveSatisfied: false,
      criteria: input.completionContract.contract.criteria.map((criterion) => ({
        criterionId: criterion.id,
        status: "unknown",
        evidenceRefs: [],
        explanation: "A context compaction run does no task work.",
      })),
      remainingWork: [],
    },
    evidence: [],
    verification: [],
    attentionRequests: [],
    artifacts: [],
    continuation: {
      kind: "same_agent",
      summary: "The agent's context was compacted; the task continues unchanged.",
      idempotencyKey: `context-compaction:${input.binding.issueId}:${requestId}`,
    },
  };
}

function sessionGoalResult(
  input: NativeExecutionInput,
  goal: HarnessThreadGoal | null,
  reason: string,
): PrpStructuredRunResult {
  const status = goalStatus(goal);
  const objective = goal?.objective ?? input.completionContract.contract.objective;
  const disposition = status === "complete"
    ? "done"
    : status === "blocked"
      ? "blocked"
      : "yielded";
  const result: PrpStructuredRunResult = {
    schema: "paperclip.run_result.v1",
    reportedWorkDisposition: disposition,
    summary: status === "complete"
      ? `Session goal completed: ${objective}`
      : status === "blocked"
        ? `Session goal blocked: ${objective}`
        : `Session goal yielded (${status ?? "cleared"}): ${objective}`,
    completionClaim: {
      contractRevision: input.completionContract.contract.revision,
      objectiveSatisfied: status === "complete",
      criteria: input.completionContract.contract.criteria.map((criterion) => ({
        criterionId: criterion.id,
        status: status === "complete" ? "satisfied" : "unknown",
        evidenceRefs: status === "complete" ? ["session-goal:complete"] : [],
        explanation: `Provider session goal status: ${status ?? "cleared"}.`,
      })),
      remainingWork: status === "complete"
        ? []
        : [{ description: reason, blocksCompletion: true }],
    },
    evidence: status === "complete"
      ? [{ kind: "session_goal_status", ref: "session-goal:complete" }]
      : [],
    verification: [],
    attentionRequests: [],
    artifacts: [],
    ...(disposition === "blocked"
      ? {
          blocker: {
            reasonCode: "session_goal_blocked",
            owner: { kind: "agent", name: "session goal provider" },
            unblockAction: reason,
            scope: "current_track" as const,
          },
        }
      : {}),
    ...(disposition === "yielded"
      ? {
          continuation: {
            kind: "same_agent" as const,
            summary: reason,
            idempotencyKey: `session-goal:${input.binding.issueId}:${status ?? "cleared"}`,
          },
        }
      : {}),
  };
  return result;
}

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 reconcileRecoveryCheckpoint(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;
  let proposalTurnId = input.checkpoint.activeTurnId ?? null;
  let recoveredResult: PrpStructuredRunResult | null = null;
  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;
    for (const event of replay.events) {
      if (event.eventType === "turn.started" && event.turnId !== proposalTurnId) {
        proposalTurnId = null;
        recoveredResult = null;
      }
      if (
        proposalTurnId &&
        event.turnId === proposalTurnId &&
        event.eventType === "run.result.proposed"
      ) {
        const validation = validatePrpStructuredRunResult(event.payload);
        if (
          !validation.ok ||
          event.runId !== input.runId ||
          event.normalizedSessionId !== input.checkpoint.identity.sessionId ||
          event.sourceInstanceId !== input.sourceInstanceId
        ) {
          throw new NativeSessionProtocolIntegrityError("source_event_replay_conflict");
        }
        const previous = recoveredResult ?? input.checkpoint.semanticResult;
        const normalize = (result: PrpStructuredRunResult) =>
          canonicalJson(input.controlPlane.normalizeReplayPayload?.(result) ?? result);
        if (previous && normalize(previous) !== normalize(validation.result)) {
          throw new NativeSessionProtocolIntegrityError("source_event_replay_conflict");
        }
        recoveredResult = validation.result;
      }
    }
    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) &&
    recoveredResult === null
  ) {
    return input.checkpoint;
  }
  return {
    ...input.checkpoint,
    cursor: String(persistedHighWater),
    ...(recoveredResult === null ? {} : { semanticResult: recoveredResult }),
  };
}

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;
  }
}

const EFFECT_FREE_ACPX_USAGE_COUNTERS = [
  "inputTokens",
  "outputTokens",
  "activeSeconds",
  "providerCostUsd",
  "cacheReadTokens",
  "cacheWriteTokens",
] as const;

function objectRecord(value: unknown): Record<string, unknown> | null {
  return typeof value === "object" && value !== null && !Array.isArray(value)
    ? (value as Record<string, unknown>)
    : null;
}

function isZeroWorkAcpxUsage(payload: Record<string, unknown>): boolean {
  if (payload.kind !== "usage") return false;
  const usage = objectRecord(payload.usage);
  if (
    usage === null ||
    (Object.hasOwn(usage, "runDeltaComplete") && usage.runDeltaComplete !== true) ||
    Object.keys(usage).some((key) => key !== "total" && key !== "runDelta" && key !== "runDeltaComplete")
  ) {
    return false;
  }
  return ["total", "runDelta"].every((sectionName) => {
    const section = objectRecord(usage[sectionName]);
    if (
      section === null ||
      section.requests !== 1 ||
      Object.keys(section).some(
        (key) =>
          key !== "requests" &&
          !EFFECT_FREE_ACPX_USAGE_COUNTERS.includes(
            key as (typeof EFFECT_FREE_ACPX_USAGE_COUNTERS)[number],
          ),
      )
    ) {
      return false;
    }
    return EFFECT_FREE_ACPX_USAGE_COUNTERS.every(
      (counter) => section[counter] === 0,
    );
  });
}

async function replayProvesEffectFreeInitialAcpxTurn(input: {
  controlPlane: ControlPlanePort;
  checkpoint: PersistedNativeSession;
  runId: string;
  sourceInstanceId: string;
}): Promise<boolean> {
  const terminalTurns = input.checkpoint.terminalTurns ?? [];
  if (
    input.checkpoint.driverKind !== "acpx_runtime" ||
    input.checkpoint.semanticResult ||
    input.checkpoint.activeTurnId ||
    terminalTurns.length !== 1 ||
    terminalTurns[0]!.turnId.length === 0
  ) {
    return false;
  }
  const targetTurnId = terminalTurns[0]!.turnId;
  const events: PrpEvent[] = [];
  let afterSourceSeq = 0;
  try {
    while (true) {
      const replay = await input.controlPlane.replayEvents({
        runId: input.runId,
        sourceInstanceId: input.sourceInstanceId,
        afterSourceSeq,
        limit: 1_000,
      });
      if (replay.events.length === 0) break;
      for (const event of replay.events) {
        // An incomplete or reordered replay cannot prove absence of work.
        if (event.sourceSeq !== afterSourceSeq + 1) return false;
        events.push(structuredClone(event));
        afterSourceSeq = event.sourceSeq;
        if (events.length > 10_000) return false;
      }
    }
  } catch {
    return false;
  }

  const submittedIndexes = events.flatMap((event, index) =>
    event.eventType === "turn.submitted" ? [index] : [],
  );
  const startedIndexes = events.flatMap((event, index) =>
    event.turnId === targetTurnId && event.eventType === "turn.started"
      ? [index]
      : [],
  );
  const acceptedIndexes = events.flatMap((event, index) =>
    event.turnId === targetTurnId && event.eventType === "turn.accepted"
      ? [index]
      : [],
  );
  const completedIndexes = events.flatMap((event, index) =>
    event.turnId === targetTurnId && event.eventType === "turn.completed"
      ? [index]
      : [],
  );
  if (
    submittedIndexes.length !== 1 ||
    startedIndexes.length !== 1 ||
    acceptedIndexes.length !== 1 ||
    completedIndexes.length !== 1
  ) {
    return false;
  }
  const submittedIndex = submittedIndexes[0]!;
  const startedIndex = startedIndexes[0]!;
  const acceptedIndex = acceptedIndexes[0]!;
  const completedIndex = completedIndexes[0]!;
  const completedPayload = objectRecord(events[completedIndex]!.payload);
  if (
    startedIndex !== submittedIndex + 1 ||
    acceptedIndex !== startedIndex + 1 ||
    completedIndex <= acceptedIndex ||
    events[submittedIndex]!.turnId != null ||
    completedPayload?.status !== "completed" ||
    (completedPayload?.error !== undefined && completedPayload?.error !== null)
  ) {
    return false;
  }
  if (
    events.some(
      (event, index) =>
        event.turnId === targetTurnId &&
        (index < startedIndex || index > completedIndex),
    )
  ) {
    return false;
  }
  return events
    .slice(acceptedIndex + 1, completedIndex)
    .every(
      (event) =>
        event.turnId === targetTurnId &&
        event.eventType === "item.completed" &&
        isZeroWorkAcpxUsage(event.payload),
    );
}

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: NativeSessionCleanupDomain = {
    key: JSON.stringify([
      input.binding.companyId,
      descriptor.kind,
      descriptor.name,
      ...(options.remoteCleanupScope ? [options.remoteCleanupScope] : []),
    ]),
    // A failed close can leave its provider owning this task's workspace and
    // this provider conversation. Nothing else is shared between sessions of
    // the same backend, so other tasks are not gated by it.
    resources: [
      JSON.stringify(["issue", input.binding.issueId]),
      JSON.stringify([
        "session",
        input.session.normalizedSessionId ?? `run:${input.binding.runId}`,
      ]),
    ],
  };
  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" && (input.provider.kind !== "openai_dot"
        || key === "instructions" || (key === "skills" ? input.runtimeContext.skills.length > 0 : input.runtimeContext.mcp.bindingId !== null)),
    );
    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 governedWait = persistedSession?.governedWait;
  const governedReplay: PrpEvent[] = [];
  if (governedWait) {
    const event = governedWait.sourceEvent;
    const validation = validatePrpStructuredRunResult(governedWait.result);
    if (!event?.turnId || event.runId !== identity.runId || event.normalizedSessionId !== identity.sessionId
      || event.sourceInstanceId !== options.runnerInstanceId || !Number.isSafeInteger(event.sourceSeq) || event.sourceSeq < 1
      || !validation.ok || validation.result.reportedWorkDisposition !== "yielded"
      || validation.result.completionClaim?.contractRevision !== input.completionContract.contract.revision) {
      throw new Error("native_governed_wait_checkpoint_mismatch");
    }
    await runAbortableOperationWithin({
      timeoutMs: options.timeoutMs ?? 900_000,
      timeoutMessage: "native governed-wait recovery replay timed out",
      operation: async signal => {
        let afterSourceSeq = event.sourceSeq - 1;
        while (true) {
          const page = await options.controlPlane.replayEvents({ runId: identity.runId,
            sourceInstanceId: options.runnerInstanceId, afterSourceSeq, limit: 1_000 }, { signal });
          signal.throwIfAborted();
          if (!page.events.length) break;
          for (const entry of page.events) {
            if (entry.runId !== identity.runId || entry.normalizedSessionId !== identity.sessionId
              || entry.sourceInstanceId !== event.sourceInstanceId || entry.sourceSeq !== afterSourceSeq + 1
              || governedReplay.length >= 10_000) throw new Error("native_governed_wait_replay_mismatch");
            governedReplay.push(entry);
            afterSourceSeq = entry.sourceSeq;
          }
        }
      },
    });
    if (canonicalJson(governedReplay[0]) !== canonicalJson(event)) throw new Error("native_governed_wait_trigger_unproven");
    if (governedReplay.some(entry => entry.eventType === "turn.started" && entry.turnId !== event.turnId)) {
      throw new Error("native_governed_wait_turn_mismatch");
    }
  }
  let recovered = false;
  let session: NativeSession | null = null;
  let continuityBreak: {
    reason: string;
    previousDriverSessionId: string;
    previousProviderSessionId: string | null;
  } | null = null;
  let reconciledRecoveryCheckpoint: PersistedNativeSession | null = null;
  const checkpointedInterruption = persistedSession && nativeRestartInterruptedTurnId(persistedSession);
  if (persistedSession && (checkpointedInterruption || nativeRestartContinuationInFlight(persistedSession))) {
    // The turn failed because its process vanished; the conversation is still
    // usable. Keep its terminal fingerprint so replay cannot execute it twice.
    persistedSession = { ...persistedSession, terminal: null };
  }
  await options.onSessionAdmission?.();
  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 {
      const attach = async (signal?: AbortSignal) => {
        signal?.throwIfAborted();
        await options.existingSession!.attachRun!({ identity,
        completionContract: {
          revision: input.completionContract.contract.revision,
          criterionIds: input.completionContract.contract.criteria.map((criterion) => criterion.id),
        },
        ...(input.provider.kind === "acpx" && input.provider.agent === "copilot"
          ? { currentRunGrant: { runtimeContext: "runtimeContext" in input ? input.runtimeContext : null,
              instructions: nativeSystemInstructions(input) } }
          : {}),
        });
        signal?.throwIfAborted();
      };
      if (options.startup) await runAbortableOperationWithin({
        timeoutMs: Math.max(0, options.startup.deadlineAtMs - Date.now()),
        timeoutMessage: "native_execution_startup_expired",
        signal: options.startup.signal,
        operation: attach,
      });
      else await attach();
    } 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) =>
        reconcileRecoveryCheckpoint({
          controlPlane: options.controlPlane,
          checkpoint: recoveryCheckpoint,
          runId: input.binding.runId,
          sourceInstanceId: options.runnerInstanceId,
          signal,
        }),
    });
    reconciledRecoveryCheckpoint = persistedSession;
    const providerRecoveryCheckpoint = persistedSession;

    const replacementAllowed =
      !("homeMode" in input && input.homeMode === "preconfigured") &&
      providerRecoveryCheckpoint.providerRecoveryPolicy ===
      "allow_replacement_after_resume_failure";
    const failedProviderSession =
      providerRecoveryCheckpoint.terminal?.runTerminalState === "failed" &&
      providerRecoveryCheckpoint.semanticResult === null;
    if (failedProviderSession && !replacementAllowed) {
      throw new NativeProviderTerminalFailure("provider_checkpoint_failed_terminal", false);
    }
    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,
                normalizeReplayPayload: options.controlPlane.normalizeReplayPayload?.bind(options.controlPlane),
              }),
            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) {
      // A deterministic refusal left the saved state unchanged and repeats on
      // every retry. A replacement would archive that state and, with a pending
      // side-effecting call, could repeat or lose the external action. Stop here
      // for an operator regardless of the checkpoint's replacement policy.
      if (recovery.failureClass === PROVIDER_SIDE_EFFECT_PENDING) {
        throw new NativeProviderSideEffectPendingError(recovery.reason ?? "unknown");
      }
      if (recovery.failureClass === PROVIDER_ATTACH_REFUSED) {
        throw new NativeProviderStateIncompatibleError(recovery.reason ?? "unknown");
      }
      if (!replacementAllowed) {
        if (recovery.failureClass === PROVIDER_STATE_INCOMPATIBLE) {
          throw new NativeProviderStateIncompatibleError(recovery.reason ?? "unknown");
        }
        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 ?? undefined,
      };
      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.startup ? Math.max(0, options.startup.deadlineAtMs - Date.now()) : options.timeoutMs ?? 900_000;
    const bootstrapInput = {
      identity,
      workingDirectory: input.workspace.cwd ?? undefined,
    };
    session = await runAbortableOperationWithin({
      timeoutMs: bootstrapTimeoutMs,
      timeoutMessage: `native session bootstrap timed out after ${bootstrapTimeoutMs}ms`,
      signal: options.startup?.signal,
      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;
  }
  sessionOriginRunnerInstances.set(session, options.runnerInstanceId);
  let sessionClosePromise: Promise<void> | null = null;
  let sessionQuarantined = false;
  const quarantineSession = (reason: string) => {
    if (sessionQuarantined) return;
    sessionQuarantined = true;
    try {
      void options.onSessionQuarantined?.(reason);
    } catch {
      // Diagnostics cannot prevent provider cleanup.
    }
    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("session_close_recovery_retained");
    retainFailedSessionCleanupOwner(cleanup, cleanupDomain);
  };
  const startSessionClose = (reason: string) => {
    const attempt = session.close({ reason });
    sessionClosePromise = attempt;
    return attempt;
  };
  const closeSession = (quarantineReason = "session_close_started") => {
    if (sessionClosePromise === null) {
      quarantineSession(quarantineReason);
      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 (error instanceof NativeSessionCloseUnrecoverableError) {
              quarantineSessionCleanup(session, cleanupDomain, true, error);
              throw error;
            }
            if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) {
              quarantineSessionCleanup(session, cleanupDomain, false, error);
              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;
  let goalCheckpointRequiresSuspension = Boolean(
    options.sessionGoalControl || options.resumeSessionGoalHeartbeat || persistedSession?.goal,
  );
  let executionFailure: { error: unknown } | null = null;
  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.
    await options.onSession?.(session);
    const checkpointTimeoutMs =
      options.checkpointTimeoutMs ?? DEFAULT_NATIVE_CHECKPOINT_TIMEOUT_MS;
    const persistCheckpoint = (
      snapshot: PersistedNativeSession,
      externalSignal?: AbortSignal,
    ) =>
      persistCheckpointWithin({
        snapshot: governedWait ? { ...snapshot, governedWait } : 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);
      return snapshot;
    };
    const recoveredSnapshot = await session.snapshot();
    const restartContinuation = (steeredMessages: Array<{ id: string; text: string }>) => ({
      message: { role: "user" as const, text: JSON.stringify({
        schema: "paperclip.native-continuation.v1",
        events: [
          "This is a resumed session after a Paperclip control-plane restart. Paperclip restored this same conversation and workspace; the restart interrupted the previous turn.",
          "Continue the current user request from the recorded progress and preserved workspace.",
          ...(steeredMessages.length > 0 ? [
            "During the interrupted turn the user sent the messages in steeredMessages. The restart happened before you read them. Handle each one in this turn, oldest first, together with the current request.",
          ] : []),
          "Shell commands and background processes started in earlier turns may no longer be running, and their tool sessions may be gone. Do not block only because such a process or its tool session cannot be found or inspected.",
          "Reconcile any unfinished tool or command before proceeding: check the workspace for the outputs or artifacts it should have produced. If the expected output is present, use it. If the expected output is missing, re-run the command.",
          "Do not repeat completed work. Before you repeat an action with external effects, such as a push, a message or a deployment, check whether it already took effect. If you cannot determine that, report the specific blocker.",
          "Finish in this turn. Nothing resumes this conversation after the turn ends, so do not end it to wait for a background command or a notification; wait for or check that work first.",
          "Then invoke exactly one paperclip_finish or paperclip_block with the accurate disposition.",
        ].join("\n"),
        ...(steeredMessages.length > 0 ? { steeredMessages: steeredMessages.map(message => message.text) } : {}),
        completion: {
          revision: input.completionContract.contract.revision,
          criterionIds: input.completionContract.contract.criteria.map(criterion => criterion.id),
        },
      }) },
      continuation: true as const,
      requestedCollaborationMode: "executionMode" in input ? input.executionMode : "default" as const,
    });
    // A restart continuation borrows the driver's one-shot disposition
    // marker so a crash during it can adopt the accepted turn. A driver that
    // hands the marker back when the continuation completes without a result
    // still owes that one disposition request; ask it in this same attempt so
    // a restart-recovery budget already spent cannot block it.
    const continuationDispositionTurn = () => {
      const envelope = buildNativeModelEnvelope(input);
      envelope.completionContract = structuredClone(input.completionContract.contract);
      if ("requestedSkills" in envelope) envelope.requestedSkills = [];
      envelope.task.prompt = [
        "Paperclip semantic-result recovery after a restart continuation.",
        "The previous turn ended without paperclip_finish or paperclip_block, so Paperclip cannot settle this task.",
        "Do not repeat completed work. If a command you started is still running, wait for or check it in this turn; nothing resumes this conversation after the turn ends.",
        "Then invoke exactly one paperclip_finish or paperclip_block with the accurate current disposition, and stop.",
      ].join("\n");
      if ("workspace" in envelope && envelope.workspace && recoveredSnapshot.workingDirectory) {
        envelope.workspace.cwd = recoveredSnapshot.workingDirectory;
      }
      return {
        message: { role: "user" as const, text: JSON.stringify(envelope) },
        requestedCollaborationMode: "executionMode" in input ? input.executionMode : "default" as const,
      };
    };
    let restartContinuationStarted = false;
    let borrowedContinuationTurnId: string | null = null;
    let restartContinuationTurnId: string | null = null;
    let settleStartupDecision: () => void = () => undefined;
    const startupDecision = new Promise<void>(resolve => { settleStartupDecision = resolve; });
    let continuationDispositionRequested = false;
    // The consumer may already be running when a continuation starts, and the
    // continuation can complete before its start checkpoint settles. A
    // terminal waits for this setup so it neither misses the borrowed turn nor
    // persists its own checkpoint ahead of the continuation's.
    let continuationSetup: Promise<void> | null = null;
    const startRestartContinuation = (signal?: AbortSignal) => {
      restartContinuationStarted = true;
      continuationSetup = (async () => {
        // The driver reports the borrowed marker with the accepted turn; a
        // later snapshot may already show the marker returned.
        const steeredMessages = await options.getUnreadSteeredMessages?.() ?? [];
        signal?.throwIfAborted();
        const started = await session.startTurn(restartContinuation(steeredMessages));
        restartContinuationTurnId = started.turnId;
        if (started.borrowedDispositionMarker === true) borrowedContinuationTurnId = started.turnId;
        await checkpoint(signal);
      })();
      return continuationSetup;
    };
    const requestContinuationDisposition = async (event: PrpEvent, signal: AbortSignal) => {
      if (continuationDispositionRequested || event.eventType !== "turn.completed") return false;
      await continuationSetup?.catch(() => undefined);
      signal.throwIfAborted();
      if (borrowedContinuationTurnId === null || event.turnId !== borrowedContinuationTurnId) return false;
      const snapshot = await session.snapshot();
      signal.throwIfAborted();
      if (snapshot.semanticResult || snapshot.activeTurnId || snapshot.goal || snapshot.pendingRuntimeRequests?.length ||
          snapshot.dispositionOnlyRecoveryConsumed !== false ||
          !snapshot.terminalTurns?.some(terminal => terminal.turnId === borrowedContinuationTurnId)) return false;
      continuationDispositionRequested = true;
      await persistCheckpoint(snapshot, signal);
      signal.throwIfAborted();
      await session.startTurn(continuationDispositionTurn());
      await checkpoint(signal);
      return true;
    };
    const supersededByRestartContinuation = async (event: PrpEvent, signal: AbortSignal) => {
      if (!recovered || event.eventType !== "turn.failed" ||
          !isNativeRestartInterruption(event.payload.error) || !event.turnId) return false;
      if (checkpointedInterruption) {
        await startupDecision;
        signal.throwIfAborted();
      }
      if (!restartContinuationStarted) return false;
      await continuationSetup?.catch(() => undefined);
      signal.throwIfAborted();
      return restartContinuationTurnId !== null && event.turnId !== restartContinuationTurnId;
    };
    const resumeInterruptedTurn = async (event: PrpEvent, signal: AbortSignal) => {
      if (await requestContinuationDisposition(event, signal)) return true;
      if (restartContinuationStarted || !recovered || !options.resumeInterruptedTurn ||
          // A compaction-only run never continues task work after a restart.
          // Runnerd records a restart-lost turn only for these providers, after
          // proving the old process ended and restoring the same conversation.
          !["codex", "acpx", "opencode"].includes(input.provider.kind) || options.sessionCompaction || options.sessionGoalControl || options.resumeSessionGoalHeartbeat ||
          recoveredSnapshot.goal || persistedSession?.semanticResult || persistedSession?.pendingRuntimeRequests?.length ||
          event.eventType !== "turn.failed" || !isNativeRestartInterruption(event.payload.error) ||
          event.turnId !== persistedSession?.activeTurnId) return false;
      const snapshot = await session.snapshot();
      signal.throwIfAborted();
      if (nativeRestartInterruptedTurnId(snapshot) !== event.turnId || snapshot.pendingRuntimeRequests?.length) return false;
      restartContinuationStarted = true;
      // Persist the old terminal before launching. Recovery inspects provider
      // history to adopt an accepted continuation if its checkpoint was lost.
      await persistCheckpoint(snapshot, signal);
      signal.throwIfAborted();
      await startRestartContinuation(signal);
      return true;
    };
    const recoveredActiveTurnId = recovered
      ? (recoveredSnapshot.activeTurnId ?? null)
      : (persistedSession?.activeTurnId ?? null);
    const adoptedProviderTerminal = Boolean(
      recovered &&
      !recoveredActiveTurnId &&
      recoveredSnapshot.terminalTurns?.some(
        (terminal) =>
          !persistedSession?.terminalTurns?.some(
            (persistedTerminal) => persistedTerminal.turnId === terminal.turnId,
          ) &&
          (terminal.turnId === persistedSession?.activeTurnId ||
            recoveredSnapshot.dispositionOnlyRecoveryConsumed),
      ),
    );
    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 (!adoptedProviderTerminal) {
      await persistCheckpoint(recoveredSnapshot);
    }

    const governedTerminal = governedWait ? governedReplay.find(event =>
      event.turnId === governedWait!.sourceEvent.turnId && isTurnTerminal(event)) : undefined;
    if (governedTerminal?.eventType === "turn.failed" || governedTerminal?.payload.providerTerminalObserved === false) {
      throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false);
    }
    if (governedWait && recoveredActiveTurnId && recoveredActiveTurnId !== governedWait.sourceEvent.turnId) {
      throw new Error("native_governed_wait_turn_mismatch");
    }
    if (governedWait) {
      for (const event of governedReplay) {
        if (event.turnId === governedWait.sourceEvent.turnId && (event.payload.kind === "usage" || isTurnTerminal(event))) {
          await options.controlPlane.appendEvent(event);
        }
      }
    }
    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 =
        governedTerminal && governedWait
          ? Promise.resolve({ event: governedWait.sourceEvent, governedResult: governedWait.result,
              eventCount: 0, highestContiguousSourceSeq: governedTerminal.sourceSeq })
          : recoveryTerminal === null
          ? consumeTurn(
              session,
              options.controlPlane,
              input,
              options.turnTimeoutMs ?? options.timeoutMs ?? 0,
              options.runtimeInputLiveWindowMs ??
                DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS,
              options.keepSessionOpen
                ? REUSABLE_SESSION_CANCELLATION_SETTLEMENT_GRACE_MS
                : FAILED_OPERATION_SETTLEMENT_GRACE_MS,
              closeSession,
              quarantineSession,
              options.resolveGovernedWait,
              consumptionAbort.signal,
              options.sessionGoalControl || options.resumeSessionGoalHeartbeat
                ? {
                    input,
                    requestId: options.sessionGoalControl?.requestId ?? `recovery_${input.binding.runId}`,
                  }
                : undefined,
              recoveredSnapshot.goal,
              resumeInterruptedTurn,
              async (wait, signal) => {
                governedWait = structuredClone(wait);
                await checkpoint(signal);
              },
              governedWait,
              options.sessionCompaction
                ? sessionCompactionResult(input, options.sessionCompaction.requestId)
                : undefined,
              supersededByRestartContinuation,
            )
          : Promise.resolve({
              event: recoveryTerminal,
              eventCount: 0,
              highestContiguousSourceSeq:
                replayedDisposition === null ? 0 : recoveryTerminal.sourceSeq,
              // A restarted compaction run adopts its completed compaction turn.
              governedResult: options.sessionCompaction && recoveryTerminal.eventType === "turn.completed"
                ? sessionCompactionResult(input, options.sessionCompaction.requestId)
                : 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 {
        const shouldStartFreshTurn = !governedWait && (
          !recovered ||
          (!recoveredActiveTurnId &&
            !adoptedProviderTerminal &&
            !checkpointedDispositionTerminal &&
            !dispositionRecoveryStillOwned));
        if (options.sessionCompaction) {
          // Compaction summarizes existing provider context. A new session or
          // one that is still running a turn has nothing safe to compact.
          // A restart of this same compaction run keeps waiting on (or adopts)
          // the compaction turn it already started instead of starting another.
          const resumingThisCompaction = recovered && !shouldStartFreshTurn &&
            persistedSession?.identity.runId === input.binding.runId;
          if (!recovered || (!shouldStartFreshTurn && !resumingThisCompaction)) {
            throw new Error(
              "native_context_compaction_unavailable: the agent has no idle resumed session to compact",
            );
          }
          if (shouldStartFreshTurn) await session.startTurn({
            message: { role: "user", text: "" },
            continuation: true,
            compact: true,
            ...(options.sessionCompaction.contextTokensBefore === undefined
              ? {}
              : { contextTokensBefore: options.sessionCompaction.contextTokensBefore }),
          });
          await checkpoint();
        } else if (options.sessionGoalControl) {
          // Explicit controls remain authoritative after controller loss. In
          // particular, pause/clear/edit must reach a recovered session even
          // while its provider turn is still active.
          await applyNativeSessionGoalControl(session, options.sessionGoalControl);
          await checkpoint();
        } else if (options.resumeSessionGoalHeartbeat && shouldStartFreshTurn) {
          const requestId = `recovery_${input.binding.runId}`;
          if (recoveredSnapshot.goal?.status === "active") {
            await applyNativeSessionGoalControl(session, { requestId, action: "resume" });
          } else {
            // Reconcile completed/paused/cleared state without changing it.
            // An already-delivered outbox control must not be replayed as a
            // new resume just because the old heartbeat is being recovered.
            if (!session.goal) throw new Error("native_session_goal_unavailable");
            await session.goal({ action: "get", requestId });
          }
          await checkpoint();
        } else if (shouldStartFreshTurn && recovered && checkpointedInterruption) {
          await startRestartContinuation();
        } else if (shouldStartFreshTurn) {
          let modelEnvelope = recovered
            ? buildNativeModelEnvelope(input, { resumedSession: true })
            : buildNativeModelEnvelope(input);
          if (!recovered && options.getFreshSessionHandoff && "task" in modelEnvelope) {
            const handoff = await options.getFreshSessionHandoff();
            if (handoff) modelEnvelope.task.prompt = `${handoff}\n\n${modelEnvelope.task.prompt}`;
          }
          const dispositionOnlyRecovery = Boolean(
            recovered &&
            !recoveredSnapshot.semanticResult &&
            (persistedSession?.terminalTurns?.length ?? 0) > 0 &&
            !recoveredActiveTurnId,
          );
          const effectFreeInitialAcpxTurn =
            dispositionOnlyRecovery && persistedSession
              ? await replayProvesEffectFreeInitialAcpxTurn({
                  controlPlane: options.controlPlane,
                  checkpoint: persistedSession,
                  runId: input.binding.runId,
                  sourceInstanceId: options.runnerInstanceId,
                })
              : false;
          if (dispositionOnlyRecovery && !effectFreeInitialAcpxTurn) {
            modelEnvelope = buildNativeModelEnvelope(input);
            // Source references cannot resolve after replacing the task prompt.
            modelEnvelope.completionContract = structuredClone(input.completionContract.contract);
            if ("requestedSkills" in modelEnvelope) modelEnvelope.requestedSkills = [];
            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");
          }
          if (!(dispositionOnlyRecovery && !effectFreeInitialAcpxTurn) && "requestedSkills" in modelEnvelope && options.backend.preparedTaskConstraints) {
            modelEnvelope.constraints = [...options.backend.preparedTaskConstraints];
          }
          if ("workspace" in modelEnvelope && modelEnvelope.workspace && recoveredSnapshot.workingDirectory) {
            modelEnvelope.workspace.cwd = recoveredSnapshot.workingDirectory;
          }
          if (!(dispositionOnlyRecovery && !effectFreeInitialAcpxTurn) && options.getFirstTurnQueuedInput) {
            const queuedInput = await options.getFirstTurnQueuedInput();
            if (queuedInput && "task" in modelEnvelope) {
              modelEnvelope.task.prompt = `${modelEnvelope.task.prompt}\n\n${queuedInput}`;
            } else if (queuedInput && "events" in modelEnvelope) {
              modelEnvelope.events = `${modelEnvelope.events}\n\n${queuedInput}`;
            }
          }
          await session.startTurn({
            message: { role: "user", text: JSON.stringify(modelEnvelope) },
            ...(recovered && modelEnvelope.schema === "paperclip.native-continuation.v1"
              ? { continuation: true as const } : {}),
            requestedCollaborationMode:
              // Paperclip planning remains in the envelope and tool authority.
              // Cloud's qualified harness exposes only default collaboration.
              input.provider.kind === "cursor_cloud"
                ? "default"
                : "executionMode" in input ? input.executionMode : "default",
          });
          await checkpoint();
        }
        settleStartupDecision();
      } catch (error) {
        settleStartupDecision();
        // 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);
        let startupFailure = error;
        await consuming.catch((consumptionError) => {
          // A failed iterator may have already initiated close while start or
          // checkpoint was pending. Retain the authenticated integrity fault,
          // not the resulting transport/cleanup error from that race.
          if (consumptionError instanceof NativeSessionProtocolIntegrityError) {
            startupFailure = consumptionError;
          }
        });
        throw startupFailure;
      }
      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: isTurnTerminal(terminalEvent)
                    ? terminalFromEvent(
                        terminalEvent,
                        consumed.governedResult.reportedWorkDisposition,
                      )
                    : {
                        schema: "paperclip.prp.terminal.v1",
                        turnTerminalState: "completed",
                        runTerminalState: "succeeded",
                        reportedWorkDisposition: consumed.governedResult.reportedWorkDisposition,
                      },
                  turnId: terminalEvent.turnId ?? null,
                };
          signal.throwIfAborted();
          if (
            settledCompletion === null &&
            terminalEvent.eventType === "turn.failed"
          ) {
            await checkpoint(signal);
            const payload = terminalEvent.payload as Record<string, unknown>;
            const failure =
              payload.error && typeof payload.error === "object"
                ? (payload.error as Record<string, unknown>)
                : payload;
            const message =
              typeof failure.message === "string"
                ? failure.message.slice(0, 2_000)
                : "Provider turn failed";
            const recoverable =
              failure.recoverable === true || payload.recoverable === true;
            // Retain the older consumer's permanent-model classification while
            // preserving structured provider metadata. A provider explicitly
            // permitting retry must not become permanent merely from its text.
            // A driver's closed provider-failure category also marks a model
            // rejection, without putting provider text in the failure message.
            const providerFailure = readProviderFailure(failure.providerFailure);
            const modelRejected =
              !recoverable &&
              (providerFailure?.category === "model_unavailable" ||
                /issue with the selected model|model_not_found|invalid model|model[^\n]*(?:does not exist|not found|not supported)/i.test(
                  message,
                ));
            throw new NativeProviderTerminalFailure(
              typeof failure.code === "string"
                ? failure.code
                : "provider_turn_failed",
              recoverable,
              modelRejected
                ? `native_provider_model_rejected: ${message}`
                : message,
              providerFailure,
            );
          }
          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;
    let completionAdmissionStarted = 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(options.controlPlane.normalizeReplayPayload?.(existing.payload) ?? existing.payload)
              !== canonicalJson(options.controlPlane.normalizeReplayPayload?.(expected.payload) ?? 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,
          );
        }
        if (!completionAdmissionStarted) {
          // Replay/appends can yield after the prepared result's last snapshot.
          // Observe the driver's latched integrity fault once more before
          // admitting completion; Codex snapshots inspect local state only.
          try {
            await session.snapshot({ signal });
          } catch (error) {
            if (error instanceof NativeSessionProtocolIntegrityError) throw error;
            // Generic snapshot failures remain checkpoint enrichment failures,
            // not evidence that the already validated result is invalid.
          }
          signal.throwIfAborted();
          // Invocation is the local admission boundary, not an atomic fence
          // with the remote commit. A lost acknowledgement can mean committed
          // success, so later faults must not veto its idempotent confirmation.
          completionAdmissionStarted = true;
        }
        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,
        };
      },
    });
    const observeEnrichmentFailure = async (
      stage: "checkpoint" | "usage",
      error: unknown,
    ) => {
      try {
        await options.onPostCompletionEnrichmentFailure?.({ stage, error });
      } catch {
        // Diagnostics cannot revoke or delay a durably committed result.
      }
    };
    let completedSnapshot: PersistedNativeSession;
    try {
      completedSnapshot = await finalizeWithin({
        timeoutMs: options.timeoutMs ?? 900_000,
        operation: async (signal) => {
          const snapshot = await session.snapshot({ signal });
          signal.throwIfAborted();
          // A settled goal remains resumable after this controller exits. A
          // merely idle warm runner is owned only by the in-memory supervisor;
          // master correctly refuses that authority after a server restart.
          // Suspend at this quiescent boundary, including active-goal rollover
          // and clear, before returning the heartbeat result to the host. Clear
          // deliberately leaves no goal snapshot, but its session still needs
          // a durable handoff before the next run can create a new goal.
          goalCheckpointRequiresSuspension ||= snapshot.goal != null;
          const completedSnapshot = {
            ...snapshot,
            semanticResult: durableExecutionResult.result,
            terminal: durableExecutionResult.terminal,
          };
          await persistCheckpoint(completedSnapshot, signal);
          signal.throwIfAborted();
          return completedSnapshot;
        },
      });
    } catch (error) {
      await observeEnrichmentFailure("checkpoint", error);
      // completeRun is the durable commit boundary. A final snapshot or
      // checkpoint failure cannot revoke that success, but a session whose
      // exact checkpoint is unknown must not remain available for reuse.
      void closeSession("post_completion_checkpoint_failed").catch(
        () => undefined,
      );
      executionSucceeded = true;
      return {
        ...durableExecutionResult,
        providerSessionId: recoveredSnapshot.providerSessionId ?? null,
        driverVersion: descriptor.version,
        usage: null,
      };
    }

    let usage: Record<string, unknown> | null = null;
    try {
      usage = await finalizeWithin({
        timeoutMs: options.timeoutMs ?? 900_000,
        operation: async (signal) => {
          const current = (await session.usage?.()) ?? null;
          signal.throwIfAborted();
          return current;
        },
      });
    } catch (error) {
      // Usage is optional accounting enrichment. Once the exact completed
      // checkpoint is durable, an unavailable usage capability must not tear
      // down an otherwise reusable warm provider session.
      await observeEnrichmentFailure("usage", error);
    }
    const enrichment = {
      providerSessionId: completedSnapshot.providerSessionId ?? null,
      driverVersion:
        typeof usage?.driverVersion === "string"
          ? usage.driverVersion
          : descriptor.version,
      usage,
    };
    executionSucceeded = true;
    return { ...durableExecutionResult, ...enrichment };
  } catch (error) {
    executionFailure = { error };
    throw error;
  } finally {
    const shouldClose =
      !options.keepSessionOpen || !executionSucceeded || sessionQuarantined || goalCheckpointRequiresSuspension;
    if (shouldClose && options.requireSessionCloseBeforeReturn) {
      if (!failedCleanupDeferred) {
        closeSession(
          `execution_finally_close:keep=${options.keepSessionOpen === true}:succeeded=${executionSucceeded}:quarantined=${sessionQuarantined}`,
        );
      }
      // A remote runner close owns its suspension and verified checkpoint.
      // Its implementation is finite, and the host must not release the
      // environment until the complete close/retry owner has settled.
      const requiredClose = sessionCloseRecoveryPromise ?? sessionClosePromise;
      if (requiredClose !== null) {
        // Unlike ordinary provider cleanup, this close owns required remote
        // checkpoint persistence. Exhausting its bounded recovery must fail
        // the execution instead of converting the rejection into success.
        try {
          await requiredClose;
        } catch (closeError) {
          // The exact cleanup owner remains retained/quarantined above. Its
          // rejection must not turn permanent integrity failure into a
          // generic retryable transport failure at the control-plane boundary.
          if (executionFailure !== null) {
            // Keep the original identity/classification (including arbitrary
            // thrown values). Cleanup remains inspectable without replacing
            // the initiating failure with a generic shutdown timeout.
            if (executionFailure.error instanceof Error) {
              try {
                Object.defineProperty(executionFailure.error, "cleanupError", {
                  value: closeError, configurable: true,
                });
              } catch { /* Frozen errors still retain their original identity. */ }
            }
            throw executionFailure.error;
          }
          throw closeError;
        }
        await options.onSessionClosed?.();
      }
    } else if (shouldClose && !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.
      await settlesWithin(
        Promise.allSettled([closeSession()]),
        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;
}