import { createWriteStream } from 'node:fs';
import { randomUUID } from 'node:crypto';
import { stat } from 'node:fs/promises';
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import type { PlaySheetContract } from '@shared_libs/plays/static-pipeline';
import type { PlayRowUpdate } from '@shared_libs/play-runtime/ctx-types';
import type { PlayRunTimelineEntry } from '@shared_libs/play-runtime/live-events';
import type { PlayDatasetBornFrom } from '@shared_libs/play-runtime/cell-provenance';
import {
  routePlayActivityObservation,
  type PlayActivityObservation,
} from '@shared_libs/play-runtime/activity-observation';
import {
  buildPlayRunLedgerEventsFromStatusPatch,
  buildTerminalLogReplayEvents,
  createEmptyPlayRunLedgerSnapshot,
  reducePlayRunLedgerEvents,
  slicePositionalLogLines,
  type PlayRunLedgerEvent,
  type PlayRunLedgerSnapshot,
} from '@shared_libs/play-runtime/run-ledger';
import type {
  ComputeBillingItem,
  PlayStagedFileRef,
  PlayVisualNodeProgressMap,
} from '@shared_libs/play-runtime/worker-api-types';
import type { RuntimeStepReceipt } from '@shared_libs/play-runtime/ctx-types';
import type {
  CreateDbSessionRequest,
  CreateDbSessionResponse,
} from '@shared_libs/play-runtime/db-session';
import type { PlayBundleArtifact } from '@shared_libs/plays/artifact-types';
import type {
  PlayRunnerRateStateAcquireInput,
  PlayRunnerRateStateAcquireResult,
  PlayRunnerBudgetChargeInput,
  PlayRunnerBudgetChargeResult,
  PlayRunnerRateStatePenalizeInput,
  PlayRunnerRateStateReleaseInput,
} from '@shared_libs/play-runtime/protocol';
import {
  PLAY_RUNTIME_TRANSPORT_ATTEMPT_HEADER,
  PLAY_RUNTIME_CONTRACT,
  PLAY_RUNTIME_CONTRACT_HEADER,
} from '@shared_libs/play-runtime/runtime-contract';
import { PLAY_RUNTIME_API_COMPAT_PATH } from '@shared_libs/play-runtime/runtime-api-paths';
import {
  PLAY_RUNTIME_TEST_FAULT_HEADER,
  recognizedRuntimeTestFaultCount,
} from '@shared_libs/play-runtime/test-runtime-seams';
import { vercelProtectionBypassHeaders } from '@shared_libs/play-runtime/vercel-protection';
import type { RuntimeReceiptAction } from '@shared_libs/play-runtime/runtime-actions';
import { RUNTIME_CAPACITY_POLICY } from '@shared_libs/play-runtime/runtime-capacity-policy';
import {
  DEFAULT_RUNTIME_TRAFFIC_POLICY,
  isRuntimeTrafficPolicy,
  type RuntimeTrafficPolicy,
} from '@shared_libs/play-runtime/runtime-traffic-policy';

export type StoredPlayArtifactPayload = {
  sourceCode: string;
  codeFormat: 'function' | 'cjs_module' | 'esm_module';
  artifact: PlayBundleArtifact;
};

export type RuntimeStatusUpdate = {
  playId: string;
  status: string;
  error?: string;
  runtimeBackend?: string | null;
  artifactHash?: string | null;
  graphHash?: string | null;
  waitKind?: string | null;
  waitUntil?: number | null;
  activeBoundaryId?: string | null;
  lastCheckpointAt?: number | null;
  liveLogs?: string[];
  /**
   * Monotonic count of log lines the producer ever emitted on this run's
   * channel. `liveLogs` is a rotating tail of those lines; the total lets the
   * positional cursor stay correct once the buffer has rotated.
   */
  liveLogTotalCount?: number;
  /**
   * Monotonic scheduler attempt that owns the positional log channel. A new
   * attempt starts at offset zero without colliding with earlier attempts.
   */
  liveLogProducerAttempt?: number;
  datasetLifecycleEvents?: Array<{
    datasetId: string;
    path: string;
    tableNamespace: string;
    phase: 'registered' | 'available' | 'failed';
    persistedRows?: number;
    succeededRows?: number;
    failedRows?: number;
    complete?: boolean;
    /** Dataset-grain row birth, stated once per dataset (ADR 0019). */
    bornFrom?: PlayDatasetBornFrom;
    at: number;
  }>;
  /** Sparse typed activity facts. Routing/deduplication happens server-side. */
  activityObservations?: PlayActivityObservation[];
  /**
   * Explicit terminal-output replay for final runner logs when the caller must
   * keep `status` nonterminal. Direct-worker/Postgres progress finalizers do this so
   * scheduler projection remains the only terminal run writer, while the Run
   * Log Stream still gets terminal transport recovery semantics.
   */
  terminalLogReplay?: {
    lines: string[];
    totalCount?: number;
    dedupeMode?: 'semantic' | 'exact';
    producerAttempt?: number;
  } | null;
  liveTimeline?: PlayRunTimelineEntry[];
  liveNodeProgress?: PlayVisualNodeProgressMap;
  result?: Record<string, unknown> | null;
};

export type RuntimeSaveResults = {
  playId: string;
  result: {
    success: boolean;
    error?: string;
    publicResult?: Record<string, unknown> | null;
    maxCreditsPerRun?: number | null;
  };
  userId?: string | null;
};

type RuntimeApiRequest =
  | {
      action: 'create_signed_artifact_url';
      storageKey: string;
    }
  | {
      action: 'create_signed_staged_file_url';
      file: Pick<PlayStagedFileRef, 'storageKey'>;
    }
  | ({
      action: 'create_db_session';
    } & CreateDbSessionRequest)
  | {
      action: 'append_run_events';
      playId: string;
      events: PlayRunLedgerEvent[];
      idempotencyKey?: string;
    }
  | {
      action: 'start_run';
      idempotencyKey?: string;
      playName: string;
      runId: string;
      artifactStorageKey?: string | null;
      artifactHash?: string | null;
      graphHash?: string | null;
      runtimeBackend?: string | null;
      schedulerBackend?: string | null;
      schedulerSchema?: string | null;
      executionProfile?: string | null;
      maxCreditsPerRun?: number | null;
      staticPipeline?: unknown;
      source?: 'published' | 'ad_hoc' | 'draft';
      inputFileId?: string;
      inputBytes?: number;
      inputSha256?: string;
      replayedFromRunId?: string | null;
    }
  | ({
      action: 'save_results';
    } & RuntimeSaveResults)
  | {
      action: 'apply_row_updates';
      playName: string;
      tableNamespace: string;
      sheetContract?: PlaySheetContract | null;
      contractSnapshot?: unknown;
      runId: string;
      userEmail: string;
      updates: Array<Omit<PlayRowUpdate, 'rowId'> & { runId?: string }>;
    }
  | {
      action: 'compute_billing_upsert';
      sessionId: string;
      orgId: string;
      userId?: string | null;
      operation: string;
      workflowId?: string;
      runId?: string;
    }
  | {
      action: 'compute_billing_record_item';
      sessionId: string;
      orgId: string;
      userId?: string | null;
      operation: string;
      item: ComputeBillingItem;
    }
  | {
      action: 'compute_billing_finalize';
      sessionId: string;
      orgId: string;
      userId?: string | null;
      operation: string;
      status: 'completed' | 'error';
      workflowId?: string;
      runId?: string;
      maxCreditsPerRun?: number | null;
      finalItem?: ComputeBillingItem;
    }
  | ({
      action: 'rate_state_acquire';
    } & PlayRunnerRateStateAcquireInput)
  | ({
      action: 'rate_state_release';
    } & PlayRunnerRateStateReleaseInput)
  | ({
      action: 'rate_state_penalize';
    } & PlayRunnerRateStatePenalizeInput)
  | ({
      action: 'governor_budget_charge';
    } & PlayRunnerBudgetChargeInput)
  | { action: 'get_runtime_traffic_policy' }
  | RuntimeReceiptAction;

export type WorkerRuntimeApiContext = {
  baseUrl: string;
  executorToken: string;
  boundary?: 'app_runtime' | 'receipt_gateway';
  integrationMode?: 'live' | 'eval_stub' | 'fixture' | null;
  vercelProtectionBypassToken?: string | null;
  runtimeTestFaultHeader?: string | null;
  fetch?: typeof fetch;
  requestTimeoutMs?: number | null;
  /** A higher-level writer owns retries and preserves the exact request body. */
  retryPolicy?: 'default' | 'none';
  signal?: AbortSignal;
};

const APP_RUNTIME_API_RETRY_DELAYS_MS = [100, 250, 500, 1_000] as const;
const APP_RUNTIME_API_RECEIPT_RETRY_DELAYS_MS = [
  250, 500, 1_000, 2_000, 4_000, 8_000, 12_000,
] as const;
const APP_RUNTIME_API_APPEND_RETRY_DELAYS_MS = [
  100, 250, 500, 1_000, 2_000, 4_000, 8_000,
] as const;
// Full jitter (AWS "equal jitter" variant): sleep a random amount in
// [delay/2, delay] so a thundering herd of sibling settlements that all lost the
// same OCC race do not re-collide in lockstep on the next attempt.
function applyRetryJitter(delayMs: number): number {
  if (delayMs <= 0) {
    return 0;
  }
  const half = delayMs / 2;
  return Math.round(half + Math.random() * half);
}
const APP_RUNTIME_API_DEFAULT_REQUEST_TIMEOUT_MS =
  RUNTIME_CAPACITY_POLICY.receiptGateway.requestTimeoutMs;
const APP_RUNTIME_RECEIPT_RETRY_TELEMETRY_TAG =
  '[perf][worker.receipt_api.transport]';
const RUN_STATUS_LEDGER_SNAPSHOT_CACHE_LIMIT = 1_000;
const runStatusLedgerSnapshots = new Map<string, PlayRunLedgerSnapshot>();
// Positional log cursor per run attempt: count of liveLogs lines already
// forwarded from this process. A fresh process re-sends from offset 0; durable
// ingestion skips overlap within that same attempt.
const runLogChannelSentCounts = new Map<string, number>();
const runStatusUpdateChains = new Map<string, Promise<void>>();
const receiptClaimResponseTimeoutFaults = new Map<string, number>();
const receiptClaimQueryHoldFaults = new Set<string>();
const RECEIPT_CLAIM_RESPONSE_TIMEOUT_FAULT_LIMIT = 1_000;
const APP_RUNTIME_LOG_EVENT_MAX_JSON_BYTES = 128 * 1024;
const APP_RUNTIME_EVENT_BATCH_MAX_JSON_BYTES = 512 * 1024;
const appRuntimeTextEncoder = new TextEncoder();

function shouldSuppressBulkClaimResponse(input: {
  body: RuntimeApiRequest;
  runtimeTestFaultHeader?: string | null;
}): boolean {
  if (
    input.body.action !== 'claim_runtime_step_receipts' ||
    input.body.keys.length < 128
  ) {
    return false;
  }
  const requested = recognizedRuntimeTestFaultCount(
    input.runtimeTestFaultHeader,
    'receipt_claim_response_timeout',
  );
  if (requested <= 0) return false;
  const key = `${input.body.runId}:${input.body.runAttempt ?? 0}`;
  const consumed = receiptClaimResponseTimeoutFaults.get(key) ?? 0;
  if (consumed >= requested) return false;
  receiptClaimResponseTimeoutFaults.set(key, consumed + 1);
  while (
    receiptClaimResponseTimeoutFaults.size >
    RECEIPT_CLAIM_RESPONSE_TIMEOUT_FAULT_LIMIT
  ) {
    const oldest = receiptClaimResponseTimeoutFaults.keys().next().value;
    if (typeof oldest !== 'string') break;
    receiptClaimResponseTimeoutFaults.delete(oldest);
  }
  return true;
}

function runtimeTestFaultHeaderForRequest(input: {
  body: RuntimeApiRequest;
  runtimeTestFaultHeader?: string | null;
}): string | null {
  const header = input.runtimeTestFaultHeader?.trim();
  if (!header) return null;
  if (
    input.body.action !== 'claim_runtime_step_receipts' ||
    input.body.keys.length < 128 ||
    recognizedRuntimeTestFaultCount(
      header,
      'receipt_claim_query_hold_once_ms',
    ) <= 0
  ) {
    return header;
  }

  const key = `${input.body.runId}:${input.body.runAttempt ?? 0}`;
  if (!receiptClaimQueryHoldFaults.has(key)) {
    receiptClaimQueryHoldFaults.add(key);
    while (
      receiptClaimQueryHoldFaults.size >
      RECEIPT_CLAIM_RESPONSE_TIMEOUT_FAULT_LIMIT
    ) {
      const oldest = receiptClaimQueryHoldFaults.values().next().value;
      if (typeof oldest !== 'string') break;
      receiptClaimQueryHoldFaults.delete(oldest);
    }
    return header;
  }

  const filtered = header
    .split(',')
    .map((part) => part.trim())
    .filter(
      (part) =>
        part &&
        part.split(':', 1)[0]?.trim() !== 'receipt_claim_query_hold_once_ms',
    )
    .join(',');
  return filtered || null;
}

async function suppressBulkClaimResponseUntilTimeout(
  signal: AbortSignal,
): Promise<never> {
  if (signal.aborted) {
    throw signal.reason ?? new Error('Runtime receipt request was aborted.');
  }
  return await new Promise<never>((_resolve, reject) => {
    const onAbort = () => {
      signal.removeEventListener('abort', onAbort);
      reject(
        signal.reason ?? new Error('Runtime receipt request was aborted.'),
      );
    };
    signal.addEventListener('abort', onAbort, { once: true });
  });
}

function splitRunEventForAppRuntime(
  event: PlayRunLedgerEvent,
): PlayRunLedgerEvent[] {
  if (event.type !== 'log.appended' || event.lines.length === 0) return [event];
  const split: PlayRunLedgerEvent[] = [];
  let index = 0;
  while (index < event.lines.length) {
    const start = index;
    let bytes = 2;
    while (index < event.lines.length) {
      const lineBytes = appRuntimeTextEncoder.encode(
        JSON.stringify(event.lines[index]!),
      ).length;
      if (
        index > start &&
        bytes + lineBytes + 1 > APP_RUNTIME_LOG_EVENT_MAX_JSON_BYTES
      ) {
        break;
      }
      bytes += lineBytes + 1;
      index += 1;
    }
    split.push({
      ...event,
      lines: event.lines.slice(start, index),
      ...(typeof event.channelOffset === 'number'
        ? { channelOffset: event.channelOffset + start }
        : {}),
    });
  }
  return split;
}

export function partitionRunEventsForAppRuntime(
  events: readonly PlayRunLedgerEvent[],
): PlayRunLedgerEvent[][] {
  const batches: PlayRunLedgerEvent[][] = [];
  let current: PlayRunLedgerEvent[] = [];
  let currentBytes = 2;
  for (const event of events.flatMap(splitRunEventForAppRuntime)) {
    const eventBytes = appRuntimeTextEncoder.encode(
      JSON.stringify(event),
    ).length;
    if (
      current.length > 0 &&
      currentBytes + eventBytes + 1 > APP_RUNTIME_EVENT_BATCH_MAX_JSON_BYTES
    ) {
      batches.push(current);
      current = [];
      currentBytes = 2;
    }
    current.push(event);
    currentBytes += eventBytes + 1;
  }
  if (current.length > 0) batches.push(current);
  return batches;
}

async function appendRunEventBatchesViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: { playId: string; events: readonly PlayRunLedgerEvent[] },
): Promise<void> {
  for (const events of partitionRunEventsForAppRuntime(input.events)) {
    await appendRunEventsViaAppRuntimeRaw(context, {
      playId: input.playId,
      events,
    });
  }
}

function isTerminalRuntimeStatus(status: string): boolean {
  return (
    status === 'completed' ||
    status === 'failed' ||
    status === 'cancelled' ||
    status === 'terminated' ||
    status === 'timed_out'
  );
}

function runLogChannelKey(runId: string, producerAttempt?: number): string {
  return `${runId}:${producerAttempt ?? 'legacy'}`;
}

function rememberRunStatusLedgerSnapshot(
  runId: string,
  snapshot: PlayRunLedgerSnapshot,
) {
  runStatusLedgerSnapshots.delete(runId);
  runStatusLedgerSnapshots.set(runId, snapshot);
  while (
    runStatusLedgerSnapshots.size > RUN_STATUS_LEDGER_SNAPSHOT_CACHE_LIMIT
  ) {
    const oldestRunId = runStatusLedgerSnapshots.keys().next().value;
    if (!oldestRunId) break;
    runStatusLedgerSnapshots.delete(oldestRunId);
    for (const key of runLogChannelSentCounts.keys()) {
      if (key.startsWith(`${oldestRunId}:`)) {
        runLogChannelSentCounts.delete(key);
      }
    }
    runStatusUpdateChains.delete(oldestRunId);
  }
}

function isRetryableAppRuntimeResponse(input: {
  action: RuntimeApiRequest['action'];
  status: number;
  body: string;
}): boolean {
  if (!isRetryableAppRuntimeAction(input.action)) {
    return false;
  }
  // The app runtime may explicitly classify an otherwise-client-error status
  // as transient. Keep the structured delivery contract authoritative rather
  // than reducing every 4xx to permanent at the worker boundary.
  const explicitRetryable = appRuntimeExplicitRetryable(input.body);
  if (explicitRetryable !== null) return explicitRetryable;
  if (
    input.action === 'append_run_events' &&
    input.status === 500 &&
    /Play run .+ not found/i.test(input.body)
  ) {
    return true;
  }
  if (input.status === 500 && isTransientAppRuntimeFailureBody(input.body)) {
    return true;
  }
  return isRetryableAppRuntimeResponseStatus(input.status);
}

function appRuntimeExplicitRetryable(body: string): boolean | null {
  try {
    const parsed = JSON.parse(body) as unknown;
    if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
      return null;
    }
    const retryable = (parsed as { retryable?: unknown }).retryable;
    return typeof retryable === 'boolean' ? retryable : null;
  } catch {
    return null;
  }
}

function isRetryableAppRuntimeResponseStatus(status: number): boolean {
  return (
    status === 429 ||
    (status >= 520 && status <= 524) ||
    status === 502 ||
    status === 503 ||
    status === 504
  );
}

function summarizeAppRuntimeErrorBody(body: string): string {
  const trimmed = body.trim();
  if (!trimmed) {
    return 'empty response body';
  }
  if (/<!doctype\s+html|<html[\s>]/i.test(trimmed)) {
    const title = trimmed.match(/<title[^>]*>([\s\S]*?)<\/title>/i)?.[1];
    const normalizedTitle = title
      ?.replace(/<[^>]+>/g, ' ')
      .replace(/\s+/g, ' ')
      .trim();
    return normalizedTitle
      ? `HTML error page: ${normalizedTitle}`
      : 'HTML error page';
  }
  try {
    const parsed = JSON.parse(trimmed) as unknown;
    if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
      const record = parsed as Record<string, unknown>;
      const message =
        typeof record.debug_error === 'string'
          ? record.debug_error
          : typeof record.details === 'string'
            ? record.details
            : typeof record.detail === 'string'
              ? record.detail
              : typeof record.error === 'string'
                ? record.error
                : typeof record.message === 'string'
                  ? record.message
                  : null;
      if (message?.trim()) {
        return message.trim();
      }
    }
  } catch {}
  if (/<[A-Za-z][\s\S]*>/.test(trimmed)) {
    return 'HTML error page';
  }
  return trimmed.length > 500 ? `${trimmed.slice(0, 500)}...` : trimmed;
}

function appRuntimeErrorCode(response: Response, body: string): string | null {
  const header = response.headers.get('x-deepline-error-code')?.trim();
  if (header) return header;
  try {
    const parsed = JSON.parse(body) as { code?: unknown };
    return typeof parsed?.code === 'string' && parsed.code.trim()
      ? parsed.code.trim()
      : null;
  } catch {
    return null;
  }
}

export function isTransientAppRuntimeFailureBody(body: string): boolean {
  return /WorkerOverloaded|"\s*code\s*"\s*:\s*"\s*InternalServerError\s*"|Your request couldn't be completed\. Try again later\.|timeout exceeded when trying to connect|timed out|fetch failed|ECONNRESET|ECONNREFUSED|UND_ERR_CONNECT_TIMEOUT|tuple concurrently updated|RUNTIME_SCHEDULER_(?:SATURATED|UNAVAILABLE)|Runtime scheduler DB pool is saturated|Runtime scheduler DB circuit breaker is open|OptimisticConcurrencyControlFailure|changed while this mutation was being run|Documents read from or written to the .* table changed while this mutation/i.test(
    body,
  );
}

function isRetryableAppRuntimeFetchError(input: {
  action: RuntimeApiRequest['action'];
  error: unknown;
}): boolean {
  if (!isRetryableAppRuntimeAction(input.action)) {
    return false;
  }
  // The only signal on the request is AbortSignal.timeout(requestTimeoutMs), so
  // any abort here is our own per-request deadline firing against a slow app
  // runtime API (common under heavy map load) — always safe to retry across the
  // backoff ladder rather than failing the tool call on the first slow write.
  if (isRequestTimeoutAbort(input.error)) {
    return true;
  }
  const message =
    input.error instanceof Error ? input.error.message : String(input.error);
  return /fetch failed|connection (terminated|timeout|timed out|closed|reset)|econnreset|etimedout|econnrefused/i.test(
    message,
  );
}

function isRequestTimeoutAbort(error: unknown): boolean {
  if (!error || typeof error !== 'object') {
    return false;
  }
  // AbortSignal.timeout() — the only signal on the request — rejects with a
  // DOMException named TimeoutError. Match that (and its message across
  // runtimes) but NOT a bare AbortError / "the operation was aborted", so a
  // genuine run cancellation still propagates instead of being retried.
  if ((error as { name?: unknown }).name === 'TimeoutError') {
    return true;
  }
  const message =
    error instanceof Error
      ? error.message
      : String((error as { message?: unknown }).message ?? '');
  return /operation was aborted due to timeout|signal timed out/i.test(message);
}

function isRetryableAppRuntimeAction(
  action: RuntimeApiRequest['action'],
): boolean {
  return (
    action === 'append_run_events' ||
    action === 'apply_row_updates' ||
    action === 'compute_billing_finalize' ||
    action === 'compute_billing_record_item' ||
    action === 'compute_billing_upsert' ||
    action === 'create_db_session' ||
    action === 'create_signed_artifact_url' ||
    action === 'create_signed_staged_file_url' ||
    action === 'get_runtime_step_receipt' ||
    action === 'get_runtime_step_receipts' ||
    action === 'heartbeat_runtime_step_receipts' ||
    action === 'claim_runtime_step_receipt' ||
    action === 'claim_runtime_step_receipts' ||
    action === 'mark_runtime_step_receipt_running' ||
    action === 'mark_runtime_step_receipts_running' ||
    action === 'mark_runtime_step_receipts_queued' ||
    action === 'complete_runtime_step_receipt' ||
    action === 'complete_runtime_step_receipts' ||
    action === 'fail_runtime_step_receipt' ||
    action === 'fail_runtime_step_receipts' ||
    action === 'rate_state_acquire' ||
    action === 'rate_state_release' ||
    action === 'rate_state_penalize' ||
    action === 'release_runtime_step_receipt' ||
    action === 'save_results' ||
    action === 'skip_runtime_step_receipt' ||
    action === 'start_run'
  );
}

function isRuntimeStepReceiptAction(
  action: RuntimeApiRequest['action'],
): boolean {
  return (
    action === 'get_runtime_step_receipt' ||
    action === 'get_runtime_step_receipts' ||
    action === 'heartbeat_runtime_step_receipts' ||
    action === 'claim_runtime_step_receipt' ||
    action === 'claim_runtime_step_receipts' ||
    action === 'mark_runtime_step_receipt_running' ||
    action === 'mark_runtime_step_receipts_running' ||
    action === 'mark_runtime_step_receipts_queued' ||
    action === 'complete_runtime_step_receipt' ||
    action === 'complete_runtime_step_receipts' ||
    action === 'fail_runtime_step_receipt' ||
    action === 'fail_runtime_step_receipts' ||
    action === 'release_runtime_step_receipt' ||
    action === 'skip_runtime_step_receipt'
  );
}

function appRuntimeRetryDelaysForAction(
  action: RuntimeApiRequest['action'],
): readonly number[] {
  if (isRuntimeStepReceiptAction(action)) {
    return APP_RUNTIME_API_RECEIPT_RETRY_DELAYS_MS;
  }
  return action === 'append_run_events'
    ? APP_RUNTIME_API_APPEND_RETRY_DELAYS_MS
    : APP_RUNTIME_API_RETRY_DELAYS_MS;
}

function appRuntimeMaxAttempts(action: RuntimeApiRequest['action']): number {
  return appRuntimeRetryDelaysForAction(action).length + 1;
}

function appRuntimeRetryDelayMs(
  action: RuntimeApiRequest['action'],
  attempt: number,
): number {
  return appRuntimeRetryDelaysForAction(action)[attempt - 1] ?? 1_000;
}

function appRuntimeRetryAfterMs(
  response: Response,
  body: string,
): number | null {
  try {
    const parsed = JSON.parse(body) as unknown;
    if (
      parsed &&
      typeof parsed === 'object' &&
      !Array.isArray(parsed) &&
      typeof (parsed as Record<string, unknown>).retry_after_ms === 'number'
    ) {
      return Math.max(
        0,
        Math.min((parsed as Record<string, number>).retry_after_ms, 30_000),
      );
    }
  } catch {}

  const retryAfterSeconds = Number(response.headers.get('retry-after'));
  return Number.isFinite(retryAfterSeconds)
    ? Math.max(0, Math.min(retryAfterSeconds * 1_000, 30_000))
    : null;
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

type AppRuntimeRetryFailureKind =
  | 'connection_refused'
  | 'connection_reset'
  | 'connection_timeout'
  | 'dns'
  | 'fetch_failed'
  | 'http_error'
  | 'request_timeout'
  | 'response_body_error'
  | 'response_body_timeout'
  | 'retryable_http'
  | 'transport_error';

type AppRuntimeRetryFailure = {
  attempt: number;
  transportAttemptId: string;
  failureKind: AppRuntimeRetryFailureKind;
  attemptElapsedMs: number;
  retryDelayMs: number;
  httpStatus?: number;
};

type AppRuntimeRetryTelemetry = {
  enabled: boolean;
  startedAt: number;
  retryCount: number;
  retrySleepMs: number;
  failures: AppRuntimeRetryFailure[];
};

function appRuntimeTransportFailureKind(
  error: unknown,
): AppRuntimeRetryFailureKind {
  if (isRequestTimeoutAbort(error)) return 'request_timeout';
  const message = error instanceof Error ? error.message : String(error ?? '');
  if (/\b(?:ENOTFOUND|EAI_AGAIN)\b|getaddrinfo|\bdns\b/i.test(message)) {
    return 'dns';
  }
  if (/\bECONNRESET\b|connection reset/i.test(message)) {
    return 'connection_reset';
  }
  if (/\bECONNREFUSED\b|connection refused/i.test(message)) {
    return 'connection_refused';
  }
  if (
    /\bETIMEDOUT\b|UND_ERR_CONNECT_TIMEOUT|connection (?:timeout|timed out)/i.test(
      message,
    )
  ) {
    return 'connection_timeout';
  }
  return /fetch failed/i.test(message) ? 'fetch_failed' : 'transport_error';
}

function recordAppRuntimeRetryFailure(input: {
  telemetry: AppRuntimeRetryTelemetry;
  attempt: number;
  transportAttemptId: string;
  failureKind: AppRuntimeRetryFailureKind;
  attemptStartedAt: number;
  retryDelayMs: number;
  httpStatus?: number;
}): void {
  if (!input.telemetry.enabled) return;
  // Receipt actions have at most eight attempts. Keep the terminal eighth
  // failure instead of truncating the causally important final observation.
  if (input.telemetry.failures.length < 8) {
    input.telemetry.failures.push({
      attempt: input.attempt,
      transportAttemptId: input.transportAttemptId,
      failureKind: input.failureKind,
      attemptElapsedMs: Date.now() - input.attemptStartedAt,
      retryDelayMs: input.retryDelayMs,
      ...(input.httpStatus === undefined
        ? {}
        : { httpStatus: input.httpStatus }),
    });
  }
  if (input.retryDelayMs > 0) {
    input.telemetry.retryCount += 1;
    input.telemetry.retrySleepMs += input.retryDelayMs;
  }
}

function emitAppRuntimeRetrySummary(input: {
  telemetry: AppRuntimeRetryTelemetry;
  action: RuntimeApiRequest['action'];
  outcome: 'failed_after_retry' | 'recovered_after_retry';
  attempts: number;
  maxAttempts: number;
  finalAttemptStartedAt: number;
}): void {
  if (!input.telemetry.enabled || input.telemetry.retryCount === 0) return;
  const payload = {
    action: input.action,
    outcome: input.outcome,
    attempts: input.attempts,
    maxAttempts: input.maxAttempts,
    totalElapsedMs: Date.now() - input.telemetry.startedAt,
    retrySleepMs: input.telemetry.retrySleepMs,
    finalAttemptElapsedMs: Date.now() - input.finalAttemptStartedAt,
    failures: input.telemetry.failures,
  };
  // Observability must never alter receipt execution. The payload contains
  // primitives from a fixed taxonomy only: no URL, token, body, raw error,
  // run/play/org id, receipt key, input, or output.
  try {
    const serialized = JSON.stringify(payload);
    if (input.outcome === 'failed_after_retry') {
      console.warn(APP_RUNTIME_RECEIPT_RETRY_TELEMETRY_TAG, serialized);
    } else {
      console.info(APP_RUNTIME_RECEIPT_RETRY_TELEMETRY_TAG, serialized);
    }
  } catch {}
}

async function retryAppRuntimeBodyTimeoutOrThrow(input: {
  action: RuntimeApiRequest['action'];
  attempt: number;
  transportAttemptId: string;
  maxAttempts: number;
  error: unknown;
  telemetry: AppRuntimeRetryTelemetry;
  attemptStartedAt: number;
  httpStatus?: number;
  retryAfterMs?: number | null;
  boundaryLabel: string;
}): Promise<void> {
  if (!isRequestTimeoutAbort(input.error)) {
    recordAppRuntimeRetryFailure({
      telemetry: input.telemetry,
      attempt: input.attempt,
      transportAttemptId: input.transportAttemptId,
      failureKind: 'response_body_error',
      attemptStartedAt: input.attemptStartedAt,
      retryDelayMs: 0,
      httpStatus: input.httpStatus,
    });
    emitAppRuntimeRetrySummary({
      telemetry: input.telemetry,
      action: input.action,
      outcome: 'failed_after_retry',
      attempts: input.attempt,
      maxAttempts: input.maxAttempts,
      finalAttemptStartedAt: input.attemptStartedAt,
    });
    throw input.error;
  }
  if (
    input.attempt < input.maxAttempts &&
    isRetryableAppRuntimeFetchError({
      action: input.action,
      error: input.error,
    })
  ) {
    const retryDelayMs = Math.max(
      applyRetryJitter(appRuntimeRetryDelayMs(input.action, input.attempt)),
      input.retryAfterMs ?? 0,
    );
    recordAppRuntimeRetryFailure({
      telemetry: input.telemetry,
      attempt: input.attempt,
      transportAttemptId: input.transportAttemptId,
      failureKind: 'response_body_timeout',
      attemptStartedAt: input.attemptStartedAt,
      retryDelayMs,
      httpStatus: input.httpStatus,
    });
    await sleep(retryDelayMs);
    return;
  }
  recordAppRuntimeRetryFailure({
    telemetry: input.telemetry,
    attempt: input.attempt,
    transportAttemptId: input.transportAttemptId,
    failureKind: 'response_body_timeout',
    attemptStartedAt: input.attemptStartedAt,
    retryDelayMs: 0,
    httpStatus: input.httpStatus,
  });
  emitAppRuntimeRetrySummary({
    telemetry: input.telemetry,
    action: input.action,
    outcome: 'failed_after_retry',
    attempts: input.attempt,
    maxAttempts: input.maxAttempts,
    finalAttemptStartedAt: input.attemptStartedAt,
  });
  throw new AppRuntimeApiTransportError({
    action: input.action,
    attempts: input.attempt,
    transportAttemptId: input.transportAttemptId,
    cause: input.error,
    boundaryLabel: input.boundaryLabel,
  });
}

function runtimeApiBoundaryLabel(
  context: Pick<WorkerRuntimeApiContext, 'boundary'>,
): string {
  return context.boundary === 'receipt_gateway'
    ? 'Runtime receipt gateway'
    : 'App runtime API';
}

export class AppRuntimeApiTransportError extends Error {
  readonly action: RuntimeApiRequest['action'];
  readonly attempts: number;
  readonly transportAttemptId: string | null;

  constructor(input: {
    action: RuntimeApiRequest['action'];
    attempts: number;
    transportAttemptId?: string | null;
    cause: unknown;
    boundaryLabel?: string;
  }) {
    const causeMessage =
      input.cause instanceof Error ? input.cause.message : String(input.cause);
    super(
      `${input.boundaryLabel ?? 'App runtime API'} transport exhausted action=${input.action} attempts=${input.attempts}: ${causeMessage}` +
        (input.transportAttemptId
          ? ` (transport_attempt=${input.transportAttemptId})`
          : ''),
      { cause: input.cause },
    );
    this.name = 'AppRuntimeApiTransportError';
    this.action = input.action;
    this.attempts = input.attempts;
    this.transportAttemptId = input.transportAttemptId ?? null;
  }
}

/**
 * A non-2xx response from the app runtime boundary.
 *
 * Keep the response contract structured here so durable callers can decide
 * whether to retry or block a delivery without parsing an error message.
 */
export class AppRuntimeApiResponseError extends Error {
  readonly action: RuntimeApiRequest['action'];
  readonly status: number;
  readonly code: string | null;
  readonly requestId: string | null;
  readonly retryable: boolean;
  readonly detail: string;

  constructor(input: {
    action: RuntimeApiRequest['action'];
    status: number;
    code?: string | null;
    requestId?: string | null;
    retryable: boolean;
    detail: string;
    boundaryLabel?: string;
  }) {
    super(
      `${input.boundaryLabel ?? 'App runtime API'} ${input.action} failed with status ${input.status}` +
        `${input.code ? ` code=${input.code}` : ''}` +
        `${input.requestId ? ` request_id=${input.requestId}` : ''}: ` +
        input.detail,
    );
    this.name = 'AppRuntimeApiResponseError';
    this.action = input.action;
    this.status = input.status;
    this.code = input.code?.trim() || null;
    this.requestId = input.requestId?.trim() || null;
    this.retryable = input.retryable;
    this.detail = input.detail;
  }
}

const APP_RUNTIME_CAPACITY_ERROR_CODES = new Set([
  'receipt_db_admission_backpressure',
  'runtime_receipt_claim_timeout',
  'scheduler_capacity',
  'runtime_postgres_admission_backpressure',
  'RUNTIME_SCHEDULER_SATURATED',
]);

/**
 * The runtime persistence plane is temporarily full. This is an admission
 * signal, not a transport failure: callers must park the same operation rather
 * than spending the ordinary request retry ladder.
 */
export class AppRuntimeApiCapacityError extends AppRuntimeApiResponseError {
  readonly retryAfterMs: number;

  constructor(input: {
    action: RuntimeApiRequest['action'];
    status: number;
    code: string;
    requestId?: string | null;
    detail: string;
    retryAfterMs: number;
  }) {
    super({
      action: input.action,
      status: input.status,
      code: input.code,
      requestId: input.requestId,
      retryable: true,
      detail: input.detail,
    });
    this.name = 'AppRuntimeApiCapacityError';
    this.retryAfterMs = Math.max(0, Math.floor(input.retryAfterMs));
  }
}

export function isAppRuntimeApiCapacityError(
  error: unknown,
): error is AppRuntimeApiCapacityError {
  if (error instanceof AppRuntimeApiCapacityError) return true;
  if (!error || typeof error !== 'object') return false;
  const candidate = error as {
    name?: unknown;
    code?: unknown;
    retryAfterMs?: unknown;
  };
  return (
    candidate.name === 'AppRuntimeApiCapacityError' &&
    typeof candidate.code === 'string' &&
    APP_RUNTIME_CAPACITY_ERROR_CODES.has(candidate.code) &&
    typeof candidate.retryAfterMs === 'number'
  );
}

function resolveAppRuntimeApiUrl(context: WorkerRuntimeApiContext): string {
  const baseUrl = context.baseUrl.trim().replace(/\/$/, '');
  return `${baseUrl}${PLAY_RUNTIME_API_COMPAT_PATH}`;
}

async function postAppRuntimeApi<TResponse>(
  context: WorkerRuntimeApiContext,
  body: RuntimeApiRequest,
): Promise<TResponse> {
  const baseUrl = context.baseUrl.trim().replace(/\/$/, '');
  const token = context.executorToken.trim();
  if (!baseUrl) {
    throw new Error('Worker runtime API requires baseUrl.');
  }
  if (!token) {
    throw new Error('Worker runtime API requires executorToken.');
  }
  const runtimeFetch = context.fetch ?? fetch;
  const boundaryLabel = runtimeApiBoundaryLabel(context);
  const vercelHeaders = await vercelProtectionBypassHeaders({
    baseUrl,
    token: context.vercelProtectionBypassToken,
    fetchImpl: runtimeFetch,
    signal: context.signal,
  });

  const maxAttempts =
    context.retryPolicy === 'none' ? 1 : appRuntimeMaxAttempts(body.action);
  const retryTelemetry: AppRuntimeRetryTelemetry = {
    enabled: isRuntimeStepReceiptAction(body.action),
    startedAt: Date.now(),
    retryCount: 0,
    retrySleepMs: 0,
    failures: [],
  };
  const requestTimeoutMs = Math.max(
    1,
    Math.floor(
      context.requestTimeoutMs ?? APP_RUNTIME_API_DEFAULT_REQUEST_TIMEOUT_MS,
    ),
  );

  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
    const attemptStartedAt = Date.now();
    const transportAttemptId = randomUUID();
    const requestSignal = context.signal
      ? AbortSignal.any([context.signal, AbortSignal.timeout(requestTimeoutMs)])
      : AbortSignal.timeout(requestTimeoutMs);
    let response: Response;
    const runtimeTestFaultHeader = runtimeTestFaultHeaderForRequest({
      body,
      runtimeTestFaultHeader: context.runtimeTestFaultHeader,
    });
    try {
      response = await runtimeFetch(resolveAppRuntimeApiUrl(context), {
        method: 'POST',
        headers: {
          'content-type': 'application/json',
          authorization: `Bearer ${token}`,
          [PLAY_RUNTIME_CONTRACT_HEADER]: String(PLAY_RUNTIME_CONTRACT),
          [PLAY_RUNTIME_TRANSPORT_ATTEMPT_HEADER]: transportAttemptId,
          ...vercelHeaders,
          ...(runtimeTestFaultHeader
            ? {
                [PLAY_RUNTIME_TEST_FAULT_HEADER]: runtimeTestFaultHeader,
              }
            : {}),
        },
        body: JSON.stringify(body),
        signal: requestSignal,
      });
    } catch (error) {
      if (context.signal?.aborted) {
        throw context.signal.reason ?? error;
      }
      if (
        attempt < maxAttempts &&
        isRetryableAppRuntimeFetchError({ action: body.action, error })
      ) {
        const retryDelayMs = applyRetryJitter(
          appRuntimeRetryDelayMs(body.action, attempt),
        );
        recordAppRuntimeRetryFailure({
          telemetry: retryTelemetry,
          attempt,
          transportAttemptId,
          failureKind: appRuntimeTransportFailureKind(error),
          attemptStartedAt,
          retryDelayMs,
        });
        await sleep(retryDelayMs);
        continue;
      }
      recordAppRuntimeRetryFailure({
        telemetry: retryTelemetry,
        attempt,
        transportAttemptId,
        failureKind: appRuntimeTransportFailureKind(error),
        attemptStartedAt,
        retryDelayMs: 0,
      });
      emitAppRuntimeRetrySummary({
        telemetry: retryTelemetry,
        action: body.action,
        outcome: 'failed_after_retry',
        attempts: attempt,
        maxAttempts,
        finalAttemptStartedAt: attemptStartedAt,
      });
      throw new AppRuntimeApiTransportError({
        action: body.action,
        attempts: attempt,
        transportAttemptId,
        cause: error,
        boundaryLabel,
      });
    }
    if (response.ok) {
      try {
        const parsed = (await response.json()) as TResponse;
        if (
          shouldSuppressBulkClaimResponse({
            body,
            runtimeTestFaultHeader: context.runtimeTestFaultHeader,
          })
        ) {
          await suppressBulkClaimResponseUntilTimeout(requestSignal);
        }
        emitAppRuntimeRetrySummary({
          telemetry: retryTelemetry,
          action: body.action,
          outcome: 'recovered_after_retry',
          attempts: attempt,
          maxAttempts,
          finalAttemptStartedAt: attemptStartedAt,
        });
        return parsed;
      } catch (error) {
        await retryAppRuntimeBodyTimeoutOrThrow({
          action: body.action,
          attempt,
          transportAttemptId,
          maxAttempts,
          error,
          telemetry: retryTelemetry,
          attemptStartedAt,
          httpStatus: response.status,
          boundaryLabel,
        });
        continue;
      }
    }
    let responseText: string;
    try {
      responseText = await response.text();
    } catch (error) {
      if (
        isRequestTimeoutAbort(error) &&
        !isRetryableAppRuntimeResponseStatus(response.status)
      ) {
        recordAppRuntimeRetryFailure({
          telemetry: retryTelemetry,
          attempt,
          transportAttemptId,
          failureKind: 'response_body_timeout',
          attemptStartedAt,
          retryDelayMs: 0,
          httpStatus: response.status,
        });
        emitAppRuntimeRetrySummary({
          telemetry: retryTelemetry,
          action: body.action,
          outcome: 'failed_after_retry',
          attempts: attempt,
          maxAttempts,
          finalAttemptStartedAt: attemptStartedAt,
        });
        throw new Error(
          `${boundaryLabel} ${body.action} failed with status ${response.status}: response body timed out`,
          { cause: error },
        );
      }
      await retryAppRuntimeBodyTimeoutOrThrow({
        action: body.action,
        attempt,
        transportAttemptId,
        maxAttempts,
        error,
        telemetry: retryTelemetry,
        attemptStartedAt,
        httpStatus: response.status,
        retryAfterMs: appRuntimeRetryAfterMs(response, ''),
        boundaryLabel,
      });
      continue;
    }
    const responseCode = appRuntimeErrorCode(response, responseText);
    if (
      response.status === 503 &&
      responseCode &&
      APP_RUNTIME_CAPACITY_ERROR_CODES.has(responseCode)
    ) {
      throw new AppRuntimeApiCapacityError({
        action: body.action,
        status: response.status,
        code: responseCode,
        requestId: response.headers.get('x-deepline-request-id')?.trim(),
        detail: summarizeAppRuntimeErrorBody(responseText),
        retryAfterMs: appRuntimeRetryAfterMs(response, responseText) ?? 1_000,
      });
    }
    if (
      attempt < maxAttempts &&
      isRetryableAppRuntimeResponse({
        action: body.action,
        status: response.status,
        body: responseText,
      })
    ) {
      const retryDelayMs = Math.max(
        applyRetryJitter(appRuntimeRetryDelayMs(body.action, attempt)),
        appRuntimeRetryAfterMs(response, responseText) ?? 0,
      );
      recordAppRuntimeRetryFailure({
        telemetry: retryTelemetry,
        attempt,
        transportAttemptId,
        failureKind: 'retryable_http',
        attemptStartedAt,
        retryDelayMs,
        httpStatus: response.status,
      });
      await sleep(retryDelayMs);
      continue;
    }
    recordAppRuntimeRetryFailure({
      telemetry: retryTelemetry,
      attempt,
      transportAttemptId,
      failureKind: isRetryableAppRuntimeResponse({
        action: body.action,
        status: response.status,
        body: responseText,
      })
        ? 'retryable_http'
        : 'http_error',
      attemptStartedAt,
      retryDelayMs: 0,
      httpStatus: response.status,
    });
    emitAppRuntimeRetrySummary({
      telemetry: retryTelemetry,
      action: body.action,
      outcome: 'failed_after_retry',
      attempts: attempt,
      maxAttempts,
      finalAttemptStartedAt: attemptStartedAt,
    });
    const code = responseCode;
    const requestId = response.headers.get('x-deepline-request-id')?.trim();
    throw new AppRuntimeApiResponseError({
      action: body.action,
      status: response.status,
      code,
      requestId,
      // The structured response is authoritative even for a 5xx. Retrying an
      // explicitly deterministic database failure here turns one rejected SQL
      // statement into an unbounded outer-writer loop.
      retryable: isRetryableAppRuntimeResponse({
        action: body.action,
        status: response.status,
        body: responseText,
      }),
      detail: summarizeAppRuntimeErrorBody(responseText),
      boundaryLabel,
    });
  }

  throw new Error(`${boundaryLabel} ${body.action} failed after retries.`);
}

type SignedR2ReadUrlResponse = {
  url: string;
  storageKey: string;
  expiresAt: string;
};

function logSignedR2FetchPerf(input: {
  kind: 'artifact' | 'staged_file';
  storageKey: string;
  startedAt: number;
  bytes: number;
}) {
  console.info('[perf][worker.r2.signed_fetch]', {
    kind: input.kind,
    storageKey: input.storageKey,
    bytes: input.bytes,
    ms: Date.now() - input.startedAt,
  });
}

async function fetchSignedR2Buffer(input: {
  kind: 'artifact' | 'staged_file';
  signed: SignedR2ReadUrlResponse;
}): Promise<Buffer> {
  const startedAt = Date.now();
  const response = await fetch(input.signed.url);
  if (!response.ok) {
    throw new Error(
      `Signed R2 ${input.kind} fetch failed for ${input.signed.storageKey} with status ${response.status}: ${await response.text()}`,
    );
  }
  const buffer = Buffer.from(await response.arrayBuffer());
  logSignedR2FetchPerf({
    kind: input.kind,
    storageKey: input.signed.storageKey,
    startedAt,
    bytes: buffer.byteLength,
  });
  return buffer;
}

async function fetchSignedR2ToFile(input: {
  kind: 'staged_file';
  signed: SignedR2ReadUrlResponse;
  targetPath: string;
}): Promise<void> {
  const startedAt = Date.now();
  const response = await fetch(input.signed.url);
  if (!response.ok) {
    throw new Error(
      `Signed R2 ${input.kind} fetch failed for ${input.signed.storageKey} with status ${response.status}: ${await response.text()}`,
    );
  }
  if (!response.body) {
    throw new Error(
      `Signed R2 ${input.kind} fetch returned an empty response body for ${input.signed.storageKey}.`,
    );
  }
  await pipeline(
    Readable.fromWeb(response.body as Parameters<typeof Readable.fromWeb>[0]),
    createWriteStream(input.targetPath),
  );
  const written = await stat(input.targetPath);
  logSignedR2FetchPerf({
    kind: input.kind,
    storageKey: input.signed.storageKey,
    startedAt,
    bytes: written.size,
  });
}

export async function loadArtifactFromAppRuntime(
  context: WorkerRuntimeApiContext,
  storageKey: string,
): Promise<StoredPlayArtifactPayload> {
  const signed = await postAppRuntimeApi<SignedR2ReadUrlResponse>(context, {
    action: 'create_signed_artifact_url',
    storageKey,
  });
  const buffer = await fetchSignedR2Buffer({
    kind: 'artifact',
    signed,
  });
  return JSON.parse(buffer.toString('utf-8')) as StoredPlayArtifactPayload;
}

export async function loadStagedFileFromAppRuntime(
  context: WorkerRuntimeApiContext,
  file: Pick<PlayStagedFileRef, 'storageKey'>,
): Promise<Buffer> {
  const signed = await postAppRuntimeApi<SignedR2ReadUrlResponse>(context, {
    action: 'create_signed_staged_file_url',
    file,
  });
  return fetchSignedR2Buffer({
    kind: 'staged_file',
    signed,
  });
}

export async function createSignedStagedFileReadUrl(
  context: WorkerRuntimeApiContext,
  file: Pick<PlayStagedFileRef, 'storageKey'>,
): Promise<SignedR2ReadUrlResponse> {
  return await postAppRuntimeApi<SignedR2ReadUrlResponse>(context, {
    action: 'create_signed_staged_file_url',
    file,
  });
}

export async function acquireRateStateViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: PlayRunnerRateStateAcquireInput,
): Promise<PlayRunnerRateStateAcquireResult> {
  return await postAppRuntimeApi<PlayRunnerRateStateAcquireResult>(context, {
    action: 'rate_state_acquire',
    ...input,
  });
}

export async function releaseRateStateViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: PlayRunnerRateStateReleaseInput,
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'rate_state_release',
    ...input,
  });
}

export async function penalizeRateStateViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: PlayRunnerRateStatePenalizeInput,
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'rate_state_penalize',
    ...input,
  });
}

export async function chargeGovernorBudgetViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: PlayRunnerBudgetChargeInput,
): Promise<PlayRunnerBudgetChargeResult> {
  return await postAppRuntimeApi<PlayRunnerBudgetChargeResult>(context, {
    action: 'governor_budget_charge',
    ...input,
  });
}

export async function getRuntimeTrafficPolicyViaAppRuntime(
  context: WorkerRuntimeApiContext,
): Promise<RuntimeTrafficPolicy> {
  const response = await postAppRuntimeApi<unknown>(context, {
    action: 'get_runtime_traffic_policy',
  });
  if (isRuntimeTrafficPolicy(response)) return response;
  // A 200 response is not enough to activate an incident control. A rollout
  // mismatch or proxy body must behave as a control-plane outage (and use the
  // caller's visible compiled-default fallback), never as an implicit limit.
  throw new AppRuntimeApiResponseError({
    action: 'get_runtime_traffic_policy',
    status: 502,
    code: 'runtime_traffic_policy_invalid_response',
    retryable: true,
    detail:
      'Runtime traffic policy response did not match the versioned contract.',
  });
}

/**
 * The incident policy is a load-reduction overlay, never an execution
 * prerequisite. During an app/Convex rolling deploy or a control-plane outage
 * it is safer to use the compiled normal policy than to strand every new Play
 * before sandbox creation. Callers must emit the returned diagnostic; this is
 * intentionally a visible fail-open, not a silent fallback.
 */
export function fallbackRuntimeTrafficPolicyForUnavailableControlPlane(
  error: unknown,
): RuntimeTrafficPolicy | null {
  if (
    error instanceof AppRuntimeApiTransportError ||
    (error instanceof AppRuntimeApiResponseError && error.status >= 500)
  ) {
    return DEFAULT_RUNTIME_TRAFFIC_POLICY;
  }
  // A new worker is intentionally canaried against the currently active app
  // before the staged app is promoted. Older apps predate this non-critical
  // traffic-policy action and reject the unknown action before capability
  // evaluation with this exact response. Treat only that known cross-version
  // contract as unavailable; broad 403 fallback would mask a genuinely
  // missing or revoked capability.
  if (isLegacyRuntimeTrafficPolicyControlPlaneResponse(error)) {
    return DEFAULT_RUNTIME_TRAFFIC_POLICY;
  }
  return null;
}

/**
 * Identifies the precise response an app predating runtime traffic policy
 * returns during a rolling worker-before-app deployment. Keep this separate
 * from normal control-plane unavailability so on-call can distinguish a
 * rollout bridge from a service outage.
 */
export function isLegacyRuntimeTrafficPolicyControlPlaneResponse(
  error: unknown,
): error is AppRuntimeApiResponseError {
  return (
    error instanceof AppRuntimeApiResponseError &&
    error.action === 'get_runtime_traffic_policy' &&
    error.status === 403 &&
    error.code === null &&
    error.detail === 'Unsupported runtime action capability scope.'
  );
}

export async function writeStagedFileFromAppRuntime(
  context: WorkerRuntimeApiContext,
  file: Pick<PlayStagedFileRef, 'storageKey'>,
  targetPath: string,
): Promise<void> {
  const signed = await postAppRuntimeApi<SignedR2ReadUrlResponse>(context, {
    action: 'create_signed_staged_file_url',
    file,
  });
  await fetchSignedR2ToFile({
    kind: 'staged_file',
    signed,
    targetPath,
  });
}

export async function updateRunStatusViaAppRuntime(
  context: WorkerRuntimeApiContext,
  update: RuntimeStatusUpdate,
): Promise<void> {
  const previous =
    runStatusUpdateChains.get(update.playId) ?? Promise.resolve();
  const queued = previous
    .catch(() => {})
    .then(() => updateRunStatusViaAppRuntimeUnlocked(context, update));
  const tracked = queued.finally(() => {
    if (runStatusUpdateChains.get(update.playId) === tracked) {
      runStatusUpdateChains.delete(update.playId);
    }
  });
  runStatusUpdateChains.set(update.playId, tracked);
  await tracked;
}

async function updateRunStatusViaAppRuntimeUnlocked(
  context: WorkerRuntimeApiContext,
  update: RuntimeStatusUpdate,
): Promise<void> {
  const previousSnapshot =
    runStatusLedgerSnapshots.get(update.playId) ??
    createEmptyPlayRunLedgerSnapshot({
      runId: update.playId,
      status: 'queued',
    });
  // Positional cursor over this producer's log buffer. Running updates forward
  // only the unsent suffix; terminal updates resend the retained buffer so the
  // durable stream can recover any progress flush that was lost in transit.
  const sentCount = isTerminalRuntimeStatus(update.status)
    ? 0
    : (runLogChannelSentCounts.get(
        runLogChannelKey(update.playId, update.liveLogProducerAttempt),
      ) ?? 0);
  const logSlice = Array.isArray(update.liveLogs)
    ? slicePositionalLogLines({
        bufferLines: update.liveLogs,
        bufferTotalCount: update.liveLogTotalCount ?? update.liveLogs.length,
        sentCount,
      })
    : null;
  const terminalLogReplay =
    update.terminalLogReplay && update.terminalLogReplay.lines.length > 0
      ? update.terminalLogReplay
      : isTerminalRuntimeStatus(update.status) && Array.isArray(update.liveLogs)
        ? {
            lines: update.liveLogs,
            totalCount: update.liveLogTotalCount,
            producerAttempt: update.liveLogProducerAttempt,
          }
        : null;
  const statusPatchLogSlice = terminalLogReplay ? null : logSlice;
  const events = buildPlayRunLedgerEventsFromStatusPatch({
    patch: {
      runId: update.playId,
      status: update.status,
      error: update.error ?? null,
      runtimeBackend: update.runtimeBackend ?? null,
      lastCheckpointAt: update.lastCheckpointAt ?? null,
      liveLogs: statusPatchLogSlice?.lines ?? null,
      liveLogsChannelOffset: statusPatchLogSlice?.channelOffset ?? null,
      liveLogsProducerAttempt: update.liveLogProducerAttempt ?? null,
      liveNodeProgress: update.liveNodeProgress ?? null,
      result: update.result,
    },
    previousSnapshot,
    now: update.lastCheckpointAt ?? Date.now(),
    source: 'worker',
  });
  for (const event of update.datasetLifecycleEvents ?? []) {
    const current = previousSnapshot.datasetsById[event.datasetId];
    const changesSnapshot =
      !current ||
      current.path !== event.path ||
      current.tableNamespace !== event.tableNamespace ||
      current.phase !== event.phase ||
      (event.persistedRows ?? 0) > current.persistedRows ||
      (event.succeededRows ?? 0) > current.succeededRows ||
      (event.failedRows ?? 0) > current.failedRows ||
      (event.complete === true && current.complete !== true) ||
      // A birth record the snapshot has not seen — or a larger admitted-row
      // count from a later page — is itself a change, or the dataset would keep
      // an unattributed or stale birth for the whole run.
      (event.bornFrom !== undefined &&
        (!current.bornFrom ||
          event.bornFrom.rowCountIn > current.bornFrom.rowCountIn));
    if (!changesSnapshot) continue;
    events.push({
      type: 'dataset.lifecycle',
      runId: update.playId,
      source: 'worker',
      occurredAt: event.at,
      datasetId: event.datasetId,
      path: event.path,
      tableNamespace: event.tableNamespace,
      phase: event.phase,
      ...(event.persistedRows === undefined
        ? {}
        : { persistedRows: event.persistedRows }),
      ...(event.succeededRows === undefined
        ? {}
        : { succeededRows: event.succeededRows }),
      ...(event.failedRows === undefined
        ? {}
        : { failedRows: event.failedRows }),
      ...(event.complete === undefined ? {} : { complete: event.complete }),
      ...(event.bornFrom === undefined ? {} : { bornFrom: event.bornFrom }),
    });
  }
  const latestActivities = new Map(
    Object.entries(previousSnapshot.activitiesById),
  );
  for (const observation of update.activityObservations ?? []) {
    const previous = latestActivities.get(observation.activityId) ?? null;
    const lane = routePlayActivityObservation({
      previous,
      next: observation,
    });
    if (lane === 'drop' || lane === 'liveness') continue;
    events.push({
      type: 'activity.observed',
      runId: update.playId,
      source: 'worker',
      occurredAt: observation.observedAt,
      observation,
    });
    latestActivities.set(observation.activityId, observation);
  }
  const terminalLogEvents =
    terminalLogReplay && terminalLogReplay.lines.length > 0
      ? buildTerminalLogReplayEvents({
          runId: update.playId,
          source: 'worker',
          occurredAt: update.lastCheckpointAt ?? Date.now(),
          lines: terminalLogReplay.lines,
          liveLogTotalCount: terminalLogReplay.totalCount,
          dedupeMode: terminalLogReplay.dedupeMode,
          producerAttempt: terminalLogReplay.producerAttempt,
        })
      : [];
  if (events.length === 0 && terminalLogEvents.length === 0) {
    return;
  }
  const combinedEvents = [...events, ...terminalLogEvents];
  if (partitionRunEventsForAppRuntime(combinedEvents).length === 1) {
    // Preserve the single-mutation fast path for ordinary bounded updates.
    await appendRunEventsViaAppRuntime(context, {
      playId: update.playId,
      events: combinedEvents,
    });
  } else {
    // Persist terminal lifecycle first so an oversized or interrupted
    // best-effort log heal can never prevent settlement. Each bounded log
    // batch is a separate mutation; retries are safe because ingestion owns
    // positional reconciliation.
    await appendRunEventBatchesViaAppRuntime(context, {
      playId: update.playId,
      events,
    });
    await appendRunEventBatchesViaAppRuntime(context, {
      playId: update.playId,
      events: terminalLogEvents,
    });
  }
  if (statusPatchLogSlice) {
    runLogChannelSentCounts.set(
      runLogChannelKey(update.playId, update.liveLogProducerAttempt),
      statusPatchLogSlice.channelOffset + statusPatchLogSlice.lines.length,
    );
  }
  rememberRunStatusLedgerSnapshot(
    update.playId,
    reducePlayRunLedgerEvents(previousSnapshot, [
      ...events,
      ...terminalLogEvents,
    ]),
  );
}

async function appendRunEventsViaAppRuntimeRaw(
  context: WorkerRuntimeApiContext,
  input: {
    playId: string;
    events: PlayRunLedgerEvent[];
    idempotencyKey?: string;
  },
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'append_run_events',
    ...input,
  });
}

/**
 * Append Run Ledger events through the bounded transport used by every runtime
 * producer. Scheduler terminal outbox events can carry a complete log replay,
 * so this boundary must never forward an arbitrarily large event array or log
 * event directly to Convex.
 */
export async function appendRunEventsViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: {
    playId: string;
    events: PlayRunLedgerEvent[];
    idempotencyKey?: string;
  },
): Promise<void> {
  const batches = partitionRunEventsForAppRuntime(input.events);
  if (batches.length <= 1) {
    await appendRunEventsViaAppRuntimeRaw(context, input);
    return;
  }

  // Terminal idempotency keys record a completed delivery. Sending the same
  // key on an earlier log-only batch would make Convex drop the terminal batch,
  // so attach it to the batch that contains the final terminal event only.
  let terminalBatchIndex = -1;
  for (let index = batches.length - 1; index >= 0; index -= 1) {
    if (
      batches[index]!.some(
        (event) =>
          event.type === 'run.completed' ||
          event.type === 'run.failed' ||
          event.type === 'run.cancelled',
      )
    ) {
      terminalBatchIndex = index;
      break;
    }
  }
  const idempotencyBatchIndex =
    terminalBatchIndex >= 0 ? terminalBatchIndex : batches.length - 1;

  for (let index = 0; index < batches.length; index += 1) {
    await appendRunEventsViaAppRuntimeRaw(context, {
      playId: input.playId,
      events: batches[index]!,
      ...(input.idempotencyKey && index === idempotencyBatchIndex
        ? { idempotencyKey: input.idempotencyKey }
        : {}),
    });
  }
}

export async function startRunViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    artifactStorageKey?: string | null;
    artifactHash?: string | null;
    graphHash?: string | null;
    runtimeBackend?: string | null;
    schedulerBackend?: string | null;
    schedulerSchema?: string | null;
    executionProfile?: string | null;
    maxCreditsPerRun?: number | null;
    staticPipeline?: unknown;
    source?: 'published' | 'ad_hoc' | 'draft';
    inputFileId?: string;
    inputBytes?: number;
    inputSha256?: string;
    replayedFromRunId?: string | null;
    idempotencyKey?: string;
  },
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'start_run',
    ...input,
  });
}

export async function saveResultsViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: RuntimeSaveResults,
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'save_results',
    ...input,
  });
}

export async function applyRowUpdatesViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Extract<
    RuntimeApiRequest,
    { action: 'apply_row_updates' }
  > extends infer T
    ? Omit<T & { action: 'apply_row_updates' }, 'action'>
    : never,
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'apply_row_updates',
    ...input,
  });
}

export async function createDbSessionViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: CreateDbSessionRequest,
): Promise<CreateDbSessionResponse> {
  return await postAppRuntimeApi<CreateDbSessionResponse>(context, {
    action: 'create_db_session',
    ...input,
  });
}

export async function upsertComputeBillingSessionViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'compute_billing_upsert' }>,
    'action'
  >,
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'compute_billing_upsert',
    ...input,
  });
}

export async function recordComputeBillingItemViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'compute_billing_record_item' }>,
    'action'
  >,
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'compute_billing_record_item',
    ...input,
  });
}

export async function finalizeComputeBillingSessionViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'compute_billing_finalize' }>,
    'action'
  >,
): Promise<void> {
  await postAppRuntimeApi<{ ok: true }>(context, {
    action: 'compute_billing_finalize',
    ...input,
  });
}

export async function getRuntimeStepReceiptViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'get_runtime_step_receipt' }>,
    'action'
  >,
): Promise<RuntimeStepReceipt | null> {
  return await postAppRuntimeApi<RuntimeStepReceipt | null>(context, {
    action: 'get_runtime_step_receipt',
    ...input,
  });
}

export async function acquireRuntimeReceiptExecutionLockViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<
      RuntimeApiRequest,
      { action: 'acquire_runtime_receipt_execution_lock' }
    >,
    'action'
  >,
): Promise<{ ownerExecutionId: string; expiresAt: string } | null> {
  return await postAppRuntimeApi(context, {
    action: 'acquire_runtime_receipt_execution_lock',
    ...input,
  });
}

export async function releaseRuntimeReceiptExecutionLockViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<
      RuntimeApiRequest,
      { action: 'release_runtime_receipt_execution_lock' }
    >,
    'action'
  >,
): Promise<boolean> {
  return await postAppRuntimeApi(context, {
    action: 'release_runtime_receipt_execution_lock',
    ...input,
  });
}

export async function getRuntimeStepReceiptsViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'get_runtime_step_receipts' }>,
    'action'
  >,
): Promise<Array<RuntimeStepReceipt | null>> {
  return await postAppRuntimeApi<Array<RuntimeStepReceipt | null>>(context, {
    action: 'get_runtime_step_receipts',
    ...input,
  });
}

export async function claimRuntimeStepReceiptViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'claim_runtime_step_receipt' }>,
    'action'
  >,
): Promise<RuntimeStepReceipt | null> {
  return await postAppRuntimeApi<RuntimeStepReceipt | null>(context, {
    action: 'claim_runtime_step_receipt',
    ...input,
  });
}

export async function claimRuntimeStepReceiptsViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'claim_runtime_step_receipts' }>,
    'action'
  >,
): Promise<Array<RuntimeStepReceipt | null>> {
  return await postAppRuntimeApi<Array<RuntimeStepReceipt | null>>(context, {
    action: 'claim_runtime_step_receipts',
    ...input,
  });
}

export async function markRuntimeStepReceiptRunningViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'mark_runtime_step_receipt_running' }>,
    'action'
  >,
): Promise<RuntimeStepReceipt | null> {
  return await postAppRuntimeApi<RuntimeStepReceipt | null>(context, {
    action: 'mark_runtime_step_receipt_running',
    ...input,
  });
}

export async function markRuntimeStepReceiptsRunningViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<
      RuntimeApiRequest,
      { action: 'mark_runtime_step_receipts_running' }
    >,
    'action'
  >,
): Promise<Array<RuntimeStepReceipt | null>> {
  return await postAppRuntimeApi<Array<RuntimeStepReceipt | null>>(context, {
    action: 'mark_runtime_step_receipts_running',
    ...input,
  });
}

export async function markRuntimeStepReceiptsQueuedViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'mark_runtime_step_receipts_queued' }>,
    'action'
  >,
): Promise<Array<RuntimeStepReceipt | null>> {
  return await postAppRuntimeApi<Array<RuntimeStepReceipt | null>>(context, {
    action: 'mark_runtime_step_receipts_queued',
    ...input,
  });
}

export async function completeRuntimeStepReceiptViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'complete_runtime_step_receipt' }>,
    'action'
  >,
): Promise<RuntimeStepReceipt | null> {
  return await postAppRuntimeApi<RuntimeStepReceipt | null>(context, {
    action: 'complete_runtime_step_receipt',
    ...input,
  });
}

export async function releaseRuntimeStepReceiptViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'release_runtime_step_receipt' }>,
    'action'
  >,
): Promise<RuntimeStepReceipt | null> {
  return await postAppRuntimeApi<RuntimeStepReceipt | null>(context, {
    action: 'release_runtime_step_receipt',
    ...input,
  });
}

export async function completeRuntimeStepReceiptsViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'complete_runtime_step_receipts' }>,
    'action'
  >,
): Promise<Array<RuntimeStepReceipt | null>> {
  return await postAppRuntimeApi<Array<RuntimeStepReceipt | null>>(context, {
    action: 'complete_runtime_step_receipts',
    ...input,
  });
}

export async function failRuntimeStepReceiptViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'fail_runtime_step_receipt' }>,
    'action'
  >,
): Promise<RuntimeStepReceipt | null> {
  return await postAppRuntimeApi<RuntimeStepReceipt | null>(context, {
    action: 'fail_runtime_step_receipt',
    ...input,
  });
}

export async function failRuntimeStepReceiptsViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'fail_runtime_step_receipts' }>,
    'action'
  >,
): Promise<Array<RuntimeStepReceipt | null>> {
  return await postAppRuntimeApi<Array<RuntimeStepReceipt | null>>(context, {
    action: 'fail_runtime_step_receipts',
    ...input,
  });
}

export async function heartbeatRuntimeStepReceiptsViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'heartbeat_runtime_step_receipts' }>,
    'action'
  >,
): Promise<Array<RuntimeStepReceipt | null>> {
  return await postAppRuntimeApi<Array<RuntimeStepReceipt | null>>(context, {
    action: 'heartbeat_runtime_step_receipts',
    ...input,
  });
}

export async function skipRuntimeStepReceiptViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'skip_runtime_step_receipt' }>,
    'action'
  >,
): Promise<RuntimeStepReceipt | null> {
  return await postAppRuntimeApi<RuntimeStepReceipt | null>(context, {
    action: 'skip_runtime_step_receipt',
    ...input,
  });
}

/**
 * Post one durable docflow observation to the receipt gateway (ADR 0016 rule
 * 2). Callers invoke this fire-and-forget; it resolves on the gateway ack and
 * surfaces transport failures to its caller (the runner's observation sink),
 * which swallows-and-logs so the play body is never affected.
 */
export async function observeDocflowNodeViaAppRuntime(
  context: WorkerRuntimeApiContext,
  input: Omit<
    Extract<RuntimeApiRequest, { action: 'observe_docflow_node' }>,
    'action'
  >,
): Promise<void> {
  await postAppRuntimeApi<{ ok?: boolean }>(context, {
    action: 'observe_docflow_node',
    ...input,
  });
}