import {
  deserializeToolExecuteResult,
  isSerializedToolExecuteResult,
  isToolExecuteResult,
  serializeToolExecuteResult,
} from './tool-result';
import { assertNoSecretTaint } from './secret-capability';
import type { RuntimeStepReceipt } from './ctx-types';
import type { DurableReceiptRecoverySource } from './tool-execution-outcome';
import {
  isPlayExecutionSuspendedError,
  isPlayRowExecutionSuspendedError,
} from './suspension';
import type { WorkReceiptFailureKind } from './work-receipts';
import {
  PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
  runtimeLeaseHeartbeatIntervalFromExpiry,
  runtimeLeaseHeartbeatIntervalMs,
} from './lease-policy';
import { createRuntimeReceiptHeartbeatSupervisor } from './receipt-heartbeat-supervisor';
import {
  deserializeToolExecutionFailure,
  serializeToolExecutionFailure,
  TOOL_EXECUTION_ERROR_SCHEMA_VERSION,
} from '../tool-execution-error';
import type {
  ToolExecutionErrorSchemaVersion,
  ToolExecutionFailureV1,
} from '../tool-execution-error';

export const DURABLE_RECEIPT_WAIT_MAX_ATTEMPTS = 240;
export const DURABLE_RECEIPT_WAIT_DELAY_MS = 250;
const TOOL_RECEIPT_DEFAULT_WAIT_MS = 300_000;
const TOOL_RECEIPT_COMPLETION_BUFFER_MS = 30_000;
const TOOL_RECEIPT_MAX_WAIT_MS = 30 * 60_000;
const DANGEROUS_SIDE_EFFECT_LOCK_WAIT_MS = 15_000;
export const COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID =
  '__deepline_completed_cache_insert__';

export class RuntimeReceiptWaitTimeoutError extends Error {
  constructor(key: string) {
    super(`Timed out waiting for durable receipt ${key}.`);
    this.name = 'RuntimeReceiptWaitTimeoutError';
  }
}

export class RuntimeReceiptLeaseLostError extends Error {
  constructor(input: { receiptKey: string; runId: string; leaseId: string }) {
    super(
      `Lost durable receipt lease for ${input.receiptKey}. runId=${input.runId} leaseId=${input.leaseId}`,
    );
    this.name = 'RuntimeReceiptLeaseLostError';
  }
}

function isInFlightRuntimeReceipt(
  receipt: RuntimeStepReceipt | null | undefined,
): receipt is RuntimeStepReceipt & {
  status: 'queued' | 'pending' | 'running';
} {
  return (
    receipt?.status === 'queued' ||
    receipt?.status === 'pending' ||
    receipt?.status === 'running'
  );
}

export function runtimeReceiptFailureKindForError(
  _error: unknown,
): WorkReceiptFailureKind {
  // Current-run retry and cross-run repair are separate contracts. The tool
  // policy still decides whether this call retries inside the owning run. Once
  // that run stores a failure, however, a later explicit run may always try the
  // semantic call again. The claim state machine keeps the owning run blocked.
  return 'repairable';
}

export function resolveRuntimeToolReceiptWaitTimeoutMs(
  requestInput: Record<string, unknown>,
): number {
  const explicitTimeoutCandidate =
    requestInput.timeoutMs ??
    requestInput.timeout_ms ??
    requestInput.max_wait_ms;
  const explicitTimeoutMs =
    typeof explicitTimeoutCandidate === 'number' &&
    isFinite(explicitTimeoutCandidate) &&
    explicitTimeoutCandidate > 0
      ? explicitTimeoutCandidate
      : undefined;
  const toolTimeoutMs = explicitTimeoutMs ?? TOOL_RECEIPT_DEFAULT_WAIT_MS;
  return Math.min(
    TOOL_RECEIPT_MAX_WAIT_MS,
    Math.max(60_000, toolTimeoutMs + TOOL_RECEIPT_COMPLETION_BUFFER_MS),
  );
}

export function resolveRuntimeToolReceiptWaitMaxAttempts(
  requestInput: Record<string, unknown>,
): number {
  return Math.ceil(
    resolveRuntimeToolReceiptWaitTimeoutMs(requestInput) /
      DURABLE_RECEIPT_WAIT_DELAY_MS,
  );
}

function throwIfReceiptWaitAborted(signal?: AbortSignal): void {
  if (!signal?.aborted) return;
  throw signal.reason instanceof Error
    ? signal.reason
    : new Error(
        typeof signal.reason === 'string' && signal.reason.trim()
          ? signal.reason
          : 'Durable receipt wait aborted.',
      );
}

async function sleepReceiptWait(
  delayMs: number,
  signal?: AbortSignal,
): Promise<void> {
  throwIfReceiptWaitAborted(signal);
  if (delayMs <= 0) return;
  await new Promise<void>((resolve, reject) => {
    const timeout = setTimeout(finish, delayMs);
    const abort = () => {
      clearTimeout(timeout);
      reject(
        signal?.reason instanceof Error
          ? signal.reason
          : new Error(
              typeof signal?.reason === 'string' && signal.reason.trim()
                ? signal.reason
                : 'Durable receipt wait aborted.',
            ),
      );
    };
    function finish() {
      signal?.removeEventListener('abort', abort);
      resolve();
    }
    signal?.addEventListener('abort', abort, { once: true });
    if (signal?.aborted) abort();
  });
  throwIfReceiptWaitAborted(signal);
}

export type DurableReceiptOperation = 'step' | 'tool' | 'fetch' | 'runPlay';

export type DurableReceiptExecutionStore = {
  enabled: boolean;
  get(receiptKey: string): Promise<RuntimeStepReceipt | null>;
  getMany(receiptKeys: string[]): Promise<Map<string, RuntimeStepReceipt>>;
  claim(
    receiptKey: string,
    runId: string,
    reclaimRunning?: boolean,
    forceRefresh?: boolean,
    forceFailedRefresh?: boolean,
  ): Promise<RuntimeStepReceipt | null>;
  markRunning?(
    receiptKey: string,
    runId: string,
    leaseId?: string | null,
  ): Promise<RuntimeStepReceipt | null>;
  complete(
    receiptKey: string,
    runId: string,
    output: unknown | null,
    leaseId?: string | null,
  ): Promise<RuntimeStepReceipt | null>;
  release(
    receiptKey: string,
    runId: string,
    leaseId?: string | null,
  ): Promise<RuntimeStepReceipt | null>;
  heartbeat?(
    receiptKey: string,
    runId: string,
    leaseId: string,
  ): Promise<RuntimeStepReceipt | null>;
  fail(
    receiptKey: string,
    runId: string,
    error: string,
    leaseId?: string | null,
    failureKind?: WorkReceiptFailureKind | null,
    errorPayload?: ToolExecutionFailureV1 | null,
  ): Promise<RuntimeStepReceipt | null>;
  canPersistFailure: boolean;
  canPersistCompletion: boolean;
  acquireExecutionLock?: (input: {
    receiptKey: string;
    ownerExecutionId: string;
    ttlMs: number;
  }) => Promise<{ ownerExecutionId: string; expiresAt: string } | null>;
  releaseExecutionLock?: (input: {
    receiptKey: string;
    ownerExecutionId: string;
  }) => Promise<boolean>;
};

async function executeWithDurableRuntimeReceiptHeartbeat<T>(input: {
  receiptKey: string;
  runId: string;
  leaseId: string;
  heartbeatIntervalMs?: number;
  store: DurableReceiptExecutionStore;
  onTerminalHeartbeat?: (receipt: RuntimeStepReceipt) => void;
  execute: () => Promise<T>;
}): Promise<T> {
  if (!input.store.heartbeat) {
    return await input.execute();
  }
  const heartbeatIntervalMs =
    input.heartbeatIntervalMs ??
    runtimeLeaseHeartbeatIntervalMs(PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS);
  if (!Number.isFinite(heartbeatIntervalMs) || heartbeatIntervalMs <= 0) {
    throw new Error('Runtime receipt heartbeat interval must be positive.');
  }

  let rejectLeaseLost!: (error: RuntimeReceiptLeaseLostError) => void;
  const leaseLost = new Promise<never>((_, reject) => {
    rejectLeaseLost = reject;
  });
  const heartbeatOnce = async (): Promise<'active' | 'terminal'> => {
    const receipt = await input.store.heartbeat!(
      input.receiptKey,
      input.runId,
      input.leaseId,
    );
    const isTerminal =
      receipt?.status === 'completed' || receipt?.status === 'skipped';
    if (receipt && isTerminal) {
      input.onTerminalHeartbeat?.(receipt);
      return 'terminal';
    }
    const stillOwnsReceipt =
      receipt &&
      (receipt.status === 'running' || receipt.status === 'queued') &&
      receipt.leaseId === input.leaseId;
    if (!stillOwnsReceipt) {
      const latest = await input.store.get(input.receiptKey);
      if (
        latest &&
        (latest.status === 'completed' || latest.status === 'skipped')
      ) {
        input.onTerminalHeartbeat?.(latest);
        return 'terminal';
      }
    }
    if (!stillOwnsReceipt) {
      throw new RuntimeReceiptLeaseLostError(input);
    }
    return 'active';
  };
  const supervisor = createRuntimeReceiptHeartbeatSupervisor({
    intervalMs: heartbeatIntervalMs,
    heartbeat: heartbeatOnce,
    isLeaseLost: (error) => error instanceof RuntimeReceiptLeaseLostError,
    onLeaseLost: (error) =>
      rejectLeaseLost(error as RuntimeReceiptLeaseLostError),
  });

  // The claim is the authoritative pre-dispatch ownership fence. Renewal is
  // only needed when the lease approaches expiry; an immediate read/renewal
  // adds a round trip to every short tool call without extending ownership.
  // Completion remains conditional on this lease, and long work still renews.
  supervisor.start();
  try {
    return await Promise.race([input.execute(), leaseLost]);
  } finally {
    supervisor.stop();
  }
}

export function runtimeReceiptOutput<T>(receipt: RuntimeStepReceipt): T {
  return isSerializedToolExecuteResult(receipt.output)
    ? (deserializeToolExecuteResult(receipt.output) as T)
    : (receipt.output as T);
}

export function runtimeReceiptFailureError(
  receipt: RuntimeStepReceipt,
  legacyPrefix: string,
  _toolErrorSchemaVersion: ToolExecutionErrorSchemaVersion,
): Error {
  const message = receipt.error ?? 'unknown error';
  return (
    deserializeToolExecutionFailure(
      message,
      receipt.errorPayload,
      TOOL_EXECUTION_ERROR_SCHEMA_VERSION,
    ) ?? new Error(`${legacyPrefix}: ${message}`)
  );
}

export async function waitForCompletedRuntimeReceipt(input: {
  receiptKey: string;
  store: Pick<DurableReceiptExecutionStore, 'getMany'>;
  maxAttempts?: number;
  delayMs?: number;
  abortSignal?: AbortSignal;
  toolErrorSchemaVersion: ToolExecutionErrorSchemaVersion;
}): Promise<RuntimeStepReceipt> {
  const maxAttempts = input.maxAttempts ?? DURABLE_RECEIPT_WAIT_MAX_ATTEMPTS;
  const delayMs = input.delayMs ?? DURABLE_RECEIPT_WAIT_DELAY_MS;
  for (let attempt = 0; attempt < maxAttempts; attempt += 1) {
    throwIfReceiptWaitAborted(input.abortSignal);
    if (attempt > 0) {
      await sleepReceiptWait(delayMs, input.abortSignal);
    }
    const receipt = (await input.store.getMany([input.receiptKey])).get(
      input.receiptKey,
    );
    if (receipt?.status === 'completed' || receipt?.status === 'skipped') {
      return receipt;
    }
    if (receipt?.status === 'failed') {
      throw runtimeReceiptFailureError(
        receipt,
        `Durable tool call ${input.receiptKey} failed`,
        input.toolErrorSchemaVersion,
      );
    }
  }
  throw new RuntimeReceiptWaitTimeoutError(input.receiptKey);
}

export async function waitForCompletedRuntimeReceipts(input: {
  receiptKeys: string[];
  store: Pick<DurableReceiptExecutionStore, 'getMany'>;
  maxAttempts?: number;
  delayMs?: number;
  abortSignal?: AbortSignal;
  log?: (message: string) => void;
  toolErrorSchemaVersion: ToolExecutionErrorSchemaVersion;
}): Promise<{
  completed: Map<string, RuntimeStepReceipt>;
  failed: Map<string, Error>;
  timedOut: Set<string>;
}> {
  const uniqueKeys = [
    ...new Set(input.receiptKeys.map((key) => key.trim()).filter(Boolean)),
  ];
  const pending = new Set(uniqueKeys);
  const completed = new Map<string, RuntimeStepReceipt>();
  const failed = new Map<string, Error>();
  const maxAttempts = input.maxAttempts ?? DURABLE_RECEIPT_WAIT_MAX_ATTEMPTS;
  const delayMs = input.delayMs ?? DURABLE_RECEIPT_WAIT_DELAY_MS;

  for (
    let attempt = 0;
    attempt < maxAttempts && pending.size > 0;
    attempt += 1
  ) {
    throwIfReceiptWaitAborted(input.abortSignal);
    if (attempt > 0) {
      await sleepReceiptWait(delayMs, input.abortSignal);
    }
    const receipts = await input.store.getMany([...pending]);
    const statuses = [...receipts.values()].reduce<Record<string, number>>(
      (counts, receipt) => {
        counts[receipt.status] = (counts[receipt.status] ?? 0) + 1;
        return counts;
      },
      {},
    );
    input.log?.(
      `[perf] runtime-receipt-wait attempt=${attempt} requested=${pending.size} indexed=${receipts.size} statuses=${
        Object.entries(statuses)
          .map(([status, count]) => `${status}:${count}`)
          .join(',') || 'none'
      }`,
    );
    for (const key of [...pending]) {
      const receipt = receipts.get(key);
      if (receipt?.status === 'completed' || receipt?.status === 'skipped') {
        completed.set(key, receipt);
        pending.delete(key);
        continue;
      }
      if (receipt?.status === 'failed') {
        failed.set(
          key,
          runtimeReceiptFailureError(
            receipt,
            `Durable tool call ${key} failed`,
            input.toolErrorSchemaVersion,
          ),
        );
        pending.delete(key);
      }
    }
    input.log?.(
      `[perf] runtime-receipt-wait attempt=${attempt} remaining=${pending.size}`,
    );
  }

  return { completed, failed, timedOut: pending };
}

export async function executeWithDurableRuntimeReceipt<T>(input: {
  operation: DurableReceiptOperation;
  id: string;
  runId: string;
  receiptKey: string;
  store: DurableReceiptExecutionStore;
  force?: boolean;
  repairRunningReceiptForSameRun?: boolean;
  repairRunningReceiptForSameRunAfterWaitTimeout?: boolean;
  runningReceiptWaitMaxAttempts?: number;
  runningReceiptWaitDelayMs?: number;
  reclaimRunning?: boolean;
  markSkipped?: (output: T) => Promise<void> | void;
  onRecovered?: (
    output: T,
    receipt: RuntimeStepReceipt,
    source: DurableReceiptRecoverySource,
  ) => T;
  onClaimedResult?: (output: T, receiptKey: string) => T;
  shouldPersistFailure?: (error: unknown) => boolean;
  heartbeatIntervalMs?: number;
  markRunningBeforeExecute?: boolean;
  /** New receipt-cache path: read completed facts, execute without creating an
   * in-flight receipt, then atomically insert the completed fact. */
  completedCacheOnly?: boolean;
  withCompletedReceiptHydration?: <Result>(
    hydrate: () => Promise<Result>,
  ) => Promise<Result>;
  requiresExecutionLock?: boolean;
  executionLockTtlMs?: number;
  toolErrorSchemaVersion: ToolExecutionErrorSchemaVersion;
  formatError: (error: unknown) => string;
  log: (message: string) => void;
  execute: (context: {
    leaseId: string | null;
    signal?: AbortSignal;
  }) => Promise<T>;
}): Promise<T> {
  const lifecycleStartedAt = Date.now();
  const logPhase = (phase: string, startedAt: number): void => {
    // Keep this at the semantic receipt layer. Transport timing below this
    // function explains HTTP cost; these markers expose gaps before/after the
    // store calls and prove whether user code or receipt orchestration owns the
    // elapsed time. Do not log receipt keys because they encode tool inputs.
    input.log(
      `[perf] durable receipt operation=${input.operation} id=${input.id} phase=${phase} elapsed_ms=${Date.now() - startedAt} lifecycle_ms=${Date.now() - lifecycleStartedAt}`,
    );
  };
  if (!input.store.enabled) {
    return await input.execute({ leaseId: null });
  }

  const recoverCompletedReceipt = async (
    receipt: RuntimeStepReceipt,
    source: DurableReceiptRecoverySource = 'cache',
  ): Promise<T> => {
    input.log(
      source === 'owner'
        ? `ctx.${input.operation}(${input.id}): executed and converged through immutable publication`
        : `ctx.${input.operation}(${input.id}): reused completed work`,
    );
    if (receipt.output === undefined) {
      return receipt.output as T;
    }
    const output = runtimeReceiptOutput<T>(receipt);
    const recovered = input.onRecovered
      ? input.onRecovered(output, receipt, source)
      : output;
    if (input.markSkipped) {
      await input.markSkipped(recovered);
    }
    return recovered;
  };

  if (input.completedCacheOnly === true) {
    const withCompletedReceiptHydration = async <Result>(
      hydrate: () => Promise<Result>,
    ): Promise<Result> =>
      input.withCompletedReceiptHydration
        ? await input.withCompletedReceiptHydration(hydrate)
        : await hydrate();
    const inspectCompletedReceipt = async (
      read: () => Promise<RuntimeStepReceipt | null>,
      source: DurableReceiptRecoverySource = 'cache',
    ): Promise<{ kind: 'recovered'; output: T } | { kind: 'unresolved' }> => {
      const inspect = async () => {
        const receipt = await read();
        if (receipt?.status === 'completed' || receipt?.status === 'skipped') {
          return {
            kind: 'recovered' as const,
            output: await recoverCompletedReceipt(receipt, source),
          };
        }
        return { kind: 'unresolved' as const };
      };
      return await withCompletedReceiptHydration(inspect);
    };
    if (input.force !== true) {
      const cached = await inspectCompletedReceipt(() =>
        input.store.get(input.receiptKey),
      );
      if (cached.kind === 'recovered') return cached.output;
    }
    const needsLock = input.requiresExecutionLock === true;
    const ownerExecutionId = `${input.runId}:${crypto.randomUUID()}`;
    let ownsLock = false;
    let lockSupervisor: ReturnType<
      typeof createRuntimeReceiptHeartbeatSupervisor
    > | null = null;
    const executionAbortController = needsLock ? new AbortController() : null;
    if (needsLock) {
      if (
        !input.store.acquireExecutionLock ||
        !input.store.releaseExecutionLock
      ) {
        throw new Error(
          `ctx.${input.operation}(${input.id}): non-idempotent provider requires the receipt execution-lock backend.`,
        );
      }
      const deadline = Date.now() + DANGEROUS_SIDE_EFFECT_LOCK_WAIT_MS;
      while (!ownsLock) {
        const lock = await input.store.acquireExecutionLock({
          receiptKey: input.receiptKey,
          ownerExecutionId,
          ttlMs: input.executionLockTtlMs ?? TOOL_RECEIPT_DEFAULT_WAIT_MS,
        });
        ownsLock = lock?.ownerExecutionId === ownerExecutionId;
        if (ownsLock) break;
        if (input.force !== true) {
          const winner = await inspectCompletedReceipt(
            () => input.store.get(input.receiptKey),
            'in_flight',
          );
          if (winner.kind === 'recovered') return winner.output;
        }
        if (Date.now() >= deadline) {
          throw new RuntimeReceiptWaitTimeoutError(input.receiptKey);
        }
        await sleepReceiptWait(DURABLE_RECEIPT_WAIT_DELAY_MS);
      }
      if (input.force !== true) {
        const afterLock = await inspectCompletedReceipt(
          () => input.store.get(input.receiptKey),
          'in_flight',
        );
        if (afterLock.kind === 'recovered') {
          await input.store.releaseExecutionLock({
            receiptKey: input.receiptKey,
            ownerExecutionId,
          });
          return afterLock.output;
        }
      }
      const lockTtlMs =
        input.executionLockTtlMs ?? TOOL_RECEIPT_DEFAULT_WAIT_MS;
      lockSupervisor = createRuntimeReceiptHeartbeatSupervisor({
        intervalMs:
          input.heartbeatIntervalMs ??
          Math.max(1_000, Math.floor(lockTtlMs / 3)),
        heartbeat: async () => {
          const renewed = await input.store.acquireExecutionLock!({
            receiptKey: input.receiptKey,
            ownerExecutionId,
            ttlMs: lockTtlMs,
          });
          if (renewed?.ownerExecutionId !== ownerExecutionId) {
            throw new RuntimeReceiptLeaseLostError({
              receiptKey: input.receiptKey,
              runId: input.runId,
              leaseId: ownerExecutionId,
            });
          }
          return 'active';
        },
        isLeaseLost: (error) => error instanceof RuntimeReceiptLeaseLostError,
        onLeaseLost: (error) => {
          input.log(
            `ctx.${input.operation}(${input.id}): dangerous side-effect execution fence lost: ${error instanceof Error ? error.message : String(error)}`,
          );
          executionAbortController?.abort(error);
        },
      });
      lockSupervisor.start();
    }
    const releaseOwnLock = async (): Promise<void> => {
      lockSupervisor?.stop();
      lockSupervisor = null;
      if (ownsLock) {
        ownsLock = false;
        await input.store.releaseExecutionLock!({
          receiptKey: input.receiptKey,
          ownerExecutionId,
        });
      }
    };
    let executed: T;
    try {
      executed = await input.execute({
        leaseId: null,
        signal: executionAbortController?.signal,
      });
      executionAbortController?.signal.throwIfAborted();
    } catch (error) {
      await releaseOwnLock();
      throw error;
    }
    try {
      const result = input.onClaimedResult
        ? input.onClaimedResult(executed, input.receiptKey)
        : executed;
      assertNoSecretTaint(result, `ctx.${input.operation} result`);
      // Force bypasses the immutable cache. It does not grant permission to
      // replace the completed fact already published under this semantic key.
      if (input.force === true) {
        return result;
      }
      // Completion normally returns a compact acknowledgement. Keep the
      // mutation and response decode inside the hydration turn anyway because
      // a concurrent winner may return a full persisted payload.
      const published = await withCompletedReceiptHydration(async () => {
        const completed = await input.store.complete(
          input.receiptKey,
          input.runId,
          isToolExecuteResult(result)
            ? serializeToolExecuteResult(result)
            : result,
          COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID,
        );
        if (
          completed?.status === 'completed' ||
          completed?.status === 'skipped'
        ) {
          return {
            kind: 'completed' as const,
            output:
              completed.output === undefined
                ? result
                : await recoverCompletedReceipt(completed, 'owner'),
          };
        }
        return { kind: 'unresolved' as const };
      });
      if (published.kind === 'completed') return published.output;
      const winner = await inspectCompletedReceipt(
        () => input.store.get(input.receiptKey),
        'owner',
      );
      if (winner.kind === 'recovered') return winner.output;
      throw new Error(
        `ctx.${input.operation}(${input.id}): completed receipt cache insert failed.`,
      );
    } finally {
      await releaseOwnLock();
    }
  }

  let ownedLeaseId: string | null = null;
  let ownedLeaseExpiresAt: string | null = null;

  const reclaimReceipt = async (): Promise<
    { kind: 'recovered'; output: T } | { kind: 'claimed' }
  > => {
    const reclaimed = await input.store.claim(
      input.receiptKey,
      input.runId,
      true,
    );
    if (reclaimed?.status === 'completed' || reclaimed?.status === 'skipped') {
      return {
        kind: 'recovered',
        output: await recoverCompletedReceipt(reclaimed, 'cache'),
      };
    }
    if (reclaimed?.status === 'failed') {
      throw runtimeReceiptFailureError(
        reclaimed,
        `ctx.${input.operation}(${input.id}): previous execution failed and cannot be reused`,
        input.toolErrorSchemaVersion,
      );
    }
    if (
      isInFlightRuntimeReceipt(reclaimed) &&
      reclaimed.claimState === 'existing'
    ) {
      throw new RuntimeReceiptWaitTimeoutError(input.receiptKey);
    }
    ownedLeaseId = reclaimed?.leaseId ?? null;
    ownedLeaseExpiresAt = reclaimed?.leaseExpiresAt ?? null;
    return { kind: 'claimed' };
  };
  const shouldRepairSameRunRunningReceipt = (
    receipt: RuntimeStepReceipt,
  ): boolean =>
    input.repairRunningReceiptForSameRun === true &&
    isInFlightRuntimeReceipt(receipt) &&
    typeof receipt.runId === 'string' &&
    receipt.runId.trim() === input.runId;
  const waitForRunningReceipt = async (): Promise<{
    kind: 'recovered';
    output: T;
  }> => ({
    kind: 'recovered',
    output: await recoverCompletedReceipt(
      await waitForCompletedRuntimeReceipt({
        receiptKey: input.receiptKey,
        store: input.store,
        maxAttempts: input.runningReceiptWaitMaxAttempts,
        delayMs: input.runningReceiptWaitDelayMs,
        toolErrorSchemaVersion: input.toolErrorSchemaVersion,
      }),
      'in_flight',
    ),
  });
  const waitForRunningReceiptOrTimeout = async (): Promise<{
    kind: 'recovered';
    output: T;
  }> => waitForRunningReceipt();
  const repairOrWaitForRunningReceipt = async (
    receipt: RuntimeStepReceipt,
  ): Promise<{ kind: 'recovered'; output: T } | { kind: 'claimed' }> => {
    if (shouldRepairSameRunRunningReceipt(receipt)) {
      const recovered = await reclaimReceipt();
      if (recovered.kind === 'recovered') return recovered;
      return { kind: 'claimed' };
    }
    try {
      return await waitForRunningReceiptOrTimeout();
    } catch (error) {
      if (error instanceof RuntimeReceiptWaitTimeoutError) {
        const recovered = await reclaimReceipt();
        if (recovered.kind === 'recovered') return recovered;
        return { kind: 'claimed' };
      }
      throw error;
    }
  };
  const repairFailedReceipt = async (
    receipt: RuntimeStepReceipt,
  ): Promise<never> => {
    throw runtimeReceiptFailureError(
      receipt,
      `ctx.${input.operation}(${input.id}): previous execution failed and cannot be reused`,
      input.toolErrorSchemaVersion,
    );
  };

  const claimStartedAt = Date.now();
  const claimed = await input.store.claim(
    input.receiptKey,
    input.runId,
    input.reclaimRunning === true || input.force === true,
    input.force === true,
  );
  logPhase('claim', claimStartedAt);
  if (
    input.force !== true &&
    (claimed?.status === 'completed' || claimed?.status === 'skipped')
  ) {
    return await recoverCompletedReceipt(claimed);
  }
  if (!claimed) {
    const latest = await input.store.get(input.receiptKey);
    if (latest?.status === 'completed' || latest?.status === 'skipped') {
      return await recoverCompletedReceipt(latest);
    }
    if (isInFlightRuntimeReceipt(latest)) {
      const recovered = await repairOrWaitForRunningReceipt(latest);
      if (recovered.kind === 'recovered') return recovered.output;
    } else if (latest?.status === 'failed') {
      await repairFailedReceipt(latest);
    } else {
      throw new Error(
        `ctx.${input.operation}(${input.id}): receipt claim did not return execution ownership.`,
      );
    }
  } else if (isInFlightRuntimeReceipt(claimed)) {
    if (claimed.claimState === 'existing') {
      const recovered = await repairOrWaitForRunningReceipt(claimed);
      if (recovered.kind === 'recovered') return recovered.output;
    } else {
      ownedLeaseId = claimed.leaseId ?? null;
      ownedLeaseExpiresAt = claimed.leaseExpiresAt ?? null;
    }
  } else if (claimed.status === 'failed') {
    await repairFailedReceipt(claimed);
  }

  let result: T;
  try {
    // New runners treat a claimed, nonterminal receipt as active. Keep the
    // mark-running transition as an explicit compatibility option for older
    // stores during their migration window, but do not pay the extra durable
    // mutation on the normal claim -> terminal path.
    if (input.markRunningBeforeExecute === true && input.store.markRunning) {
      const markRunningStartedAt = Date.now();
      const running = await input.store.markRunning(
        input.receiptKey,
        input.runId,
        ownedLeaseId,
      );
      logPhase('mark_running', markRunningStartedAt);
      if (!running || running.status !== 'running') {
        throw new RuntimeReceiptLeaseLostError({
          receiptKey: input.receiptKey,
          runId: input.runId,
          leaseId: ownedLeaseId ?? '',
        });
      }
      ownedLeaseId = running.leaseId ?? ownedLeaseId;
      ownedLeaseExpiresAt = running.leaseExpiresAt ?? ownedLeaseExpiresAt;
    }
    const executeStartedAt = Date.now();
    const executed =
      ownedLeaseId && input.store.heartbeat
        ? await executeWithDurableRuntimeReceiptHeartbeat({
            receiptKey: input.receiptKey,
            runId: input.runId,
            leaseId: ownedLeaseId,
            heartbeatIntervalMs:
              input.heartbeatIntervalMs ??
              runtimeLeaseHeartbeatIntervalFromExpiry({
                leaseExpiresAt: ownedLeaseExpiresAt,
                fallbackTtlMs: PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
              }),
            store: input.store,
            onTerminalHeartbeat: (receipt) => {
              input.log(
                `ctx.${input.operation}(${input.id}): receipt heartbeat observed terminal ${receipt.status}; treating inner gateway completion as ownership handoff`,
              );
            },
            execute: () => input.execute({ leaseId: ownedLeaseId }),
          })
        : await input.execute({ leaseId: ownedLeaseId });
    logPhase('execute', executeStartedAt);
    result = input.onClaimedResult
      ? input.onClaimedResult(executed, input.receiptKey)
      : executed;
    assertNoSecretTaint(result, `ctx.${input.operation} result`);
  } catch (error) {
    if (input.shouldPersistFailure?.(error) === false) {
      throw error;
    }
    // The ownership is uncertain, so neither `fail` nor `release` is safe.
    // Let the attempt-level cleanup and the receipt lease fence reconcile this
    // on the next Play Run attempt. In particular, never replace a possibly
    // completed invocation with a terminal error from the old owner.
    if (error instanceof RuntimeReceiptLeaseLostError) {
      throw error;
    }
    if (
      isPlayExecutionSuspendedError(error) ||
      isPlayRowExecutionSuspendedError(error)
    ) {
      const released = await input.store.release(
        input.receiptKey,
        input.runId,
        ownedLeaseId,
      );
      if (
        !released ||
        (released.status !== 'queued' && released.status !== 'pending')
      ) {
        throw new Error(
          `ctx.${input.operation}(${input.id}): receipt suspended but ownership could not be released: ${input.formatError(error)}.`,
        );
      }
      throw error;
    }
    const failed = await input.store.fail(
      input.receiptKey,
      input.runId,
      input.formatError(error),
      ownedLeaseId,
      runtimeReceiptFailureKindForError(error),
      serializeToolExecutionFailure(error),
    );
    if (!failed && input.store.canPersistFailure) {
      throw new Error(
        `ctx.${input.operation}(${input.id}): execution failed and failed receipt could not be persisted: ${input.formatError(error)}`,
      );
    }
    throw error;
  }

  const completeStartedAt = Date.now();
  const completed = await input.store.complete(
    input.receiptKey,
    input.runId,
    isToolExecuteResult(result) ? serializeToolExecuteResult(result) : result,
    ownedLeaseId,
  );
  logPhase('complete', completeStartedAt);
  if (
    !completed ||
    (completed.status !== 'completed' && completed.status !== 'skipped')
  ) {
    const latest = await input.store.get(input.receiptKey);
    if (
      latest &&
      (latest.status === 'completed' || latest.status === 'skipped')
    ) {
      input.log(
        `ctx.${input.operation}(${input.id}): receipt completion response reconciled from durable store read-back`,
      );
      return await recoverCompletedReceipt(latest, 'owner');
    }
  }
  if (!completed && input.store.canPersistCompletion) {
    throw new Error(
      `ctx.${input.operation}(${input.id}): lost durable receipt ownership before completion.`,
    );
  }
  if (
    completed &&
    (completed.status === 'completed' || completed.status === 'skipped')
  ) {
    logPhase('total', lifecycleStartedAt);
    // Completion endpoints acknowledge ownership and status only. The caller
    // already owns the live value; replay reads are the only path that needs
    // the receipt payload returned over the transport.
    if (completed.output === undefined) return result;
    return await recoverCompletedReceipt(completed, 'owner');
  }
  if (input.store.canPersistCompletion) {
    throw new Error(
      `ctx.${input.operation}(${input.id}): lost durable receipt ownership before completion.`,
    );
  }
  return result;
}