import { createHash, randomUUID } from 'node:crypto';
import {
  createRuntimePool,
  createRuntimeOneShotQueryClient,
  canReuseRuntimePostgresPoolsAcrossRequests,
  isRuntimeOneShotQueryFactoryRegistered,
  type RuntimePool,
  type RuntimePoolClient,
  type RuntimeOneShotQueryClient,
} from './runtime-pg-driver';
import { createDeferredPlayDataset, type PlayDataset } from '../plays/dataset';
import type {
  PlayStaticPipeline,
  PlaySheetContract,
} from '../plays/static-pipeline';
import type { PlayBundleArtifact } from '../plays/artifact-types';
import {
  augmentSheetContractWithDatasetFields,
  outputPhysicalSheetColumnProjections,
  physicalSheetColumnNames,
  physicalSheetColumnProjections,
  type PhysicalSheetColumnProjection,
} from '../play-data-plane/sheet-contract';
import type {
  CreateDbSessionResponse,
  DbLogicalTable,
  DbSessionLimits,
  DbSessionOperation,
  PreloadedRuntimeDbSession,
  RowsWriteResponse,
} from './db-session';
import {
  derivePlayRowIdentity,
  normalizePlayNameForSheet,
  normalizeTableNamespace,
} from '../plays/row-identity';
import { toSerializableCsvAliasedRow } from './csv-rename';
import {
  RUNTIME_WORK_RECEIPT_LOGICAL_TABLE,
  RUNTIME_WORK_RECEIPT_POSTGRES_TABLE,
  RUNTIME_WORK_RECEIPT_TABLE_NAMESPACE,
  createDbSessionResponseSchema,
  rowsWriteResponseSchema,
} from './db-session';
import {
  dbSessionPostgresUrlAad,
  decryptDbSessionPostgresUrl,
  decryptDbSessionPostgresUrlWithPrivateKey,
  generateDbSessionPostgresUrlDecryptionKey,
  type PostgresUrlDecryptionKey,
} from './db-session-crypto';
import { createRuntimeDatasetId } from './dataset-id';
import {
  scopedWorkReceiptKeyPrefix,
  isReusableWorkReceipt,
  type WorkReceipt,
  type WorkReceiptClaim,
  workReceiptFailureKindCodeForWrite,
  workReceiptFailureKindFromCode,
  type WorkReceiptFailureKind,
} from './work-receipts';
import type { ToolExecutionFailureV1 } from '../tool-execution-error';
import {
  sanitizePostgresJsonValue,
  stringifyPostgresJson,
} from './postgres-json';
import { RECEIPT_STATUS_CODE, receiptStatusFromCode } from './receipt-status';
import {
  workReceiptClaimableStatusCodes,
  workReceiptActiveOwnerPredicateSql,
  workReceiptClaimConflictPredicateSql,
  workReceiptHeartbeatPredicateSql,
  workReceiptReleaseOwnerPredicateSql,
} from './receipt-sql';
import {
  activeRuntimeSheetAttemptFenceSql,
  newerTerminalRuntimeSheetRowSql,
  releasableRuntimeSheetAttemptFenceSql,
  sameOwnerTerminalAttemptEpochSql,
} from './sheet-attempt-sql';
import type { MapRowOutcome } from './durability-store';
import { RUNTIME_CAPACITY_POLICY } from './runtime-capacity-policy';
import {
  normalizeRuntimeMapInputIndex,
  prepareRuntimeSheetRowsForJsonTransport,
  prepareRuntimeSheetRowTransitions,
  type RuntimePreparedCompletedRow,
  type RuntimePreparedFailedRow,
} from './runtime-sheet-row-transition';
import {
  MAP_ROW_OUTCOME_RUNTIME_FIELDS,
  mapRowOutcomeRuntimeFields,
  resolveMapRowOutcomeKey,
} from './map-row-outcome';
import { DEEPLINE_CELL_META_FIELD } from './cell-staleness';
import {
  PLAY_RUNTIME_CONTRACT,
  PLAY_RUNTIME_CONTRACT_HEADER,
} from './runtime-contract';
import { PLAY_RUNTIME_API_COMPAT_PATH } from './runtime-api-paths';
import {
  PLAY_RUNTIME_SHEET_ATTEMPT_LEASE_TTL_MS,
  PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
} from './lease-policy';
import {
  parseRuntimeTestFaultCounts,
  PLAY_RUNTIME_TEST_FAULT_HEADER,
} from './test-runtime-seams';
import type { RuntimeTestFaultName } from './runtime-incident-drills';
import { COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID } from './durable-receipt-execution';
import { vercelProtectionBypassHeader } from './vercel-protection';
import {
  isRuntimePostgresAdmissionError,
  RuntimePostgresAdmission,
  type RuntimePostgresAdmissionSnapshot,
  type RuntimePostgresLane,
} from './runtime-postgres-admission';

type RuntimeApiContext = {
  baseUrl?: string | null;
  executorToken?: string | null;
  orgId?: string | null;
  playName?: string | null;
  runId?: string | null;
  userEmail?: string | null;
  integrationMode?: 'live' | 'eval_stub' | 'fixture' | null;
  dbSessionStrategy?: 'preloaded' | 'gateway_only' | 'trusted_dynamic' | null;
  preloadedDbSessions?: PreloadedRuntimeDbSession[] | null;
  vercelProtectionBypassToken?: string | null;
  runtimeTestFaultHeader?: string | null;
  disablePostgresPoolCache?: boolean | null;
  postgresSessionUnwrapKey?: string | null;
  abortSignal?: AbortSignal | null;
};

const RUNTIME_RUN_DATASET_CATALOG_TABLE = '_deepline_run_datasets';

type DirectRuntimeTestFaultState = {
  headerValue: string;
  counts: Partial<Record<RuntimeTestFaultName, number>>;
};

const directRuntimeTestFaults = new Map<string, DirectRuntimeTestFaultState>();

function consumeDirectRuntimeTestFault(input: {
  context: RuntimeApiContext;
  runId: string;
  name: RuntimeTestFaultName;
}): boolean {
  const headerValue = input.context.runtimeTestFaultHeader?.trim();
  if (!headerValue) return false;
  const stateKey = input.runId.trim();
  if (!stateKey) return false;
  let state = directRuntimeTestFaults.get(stateKey);
  if (!state || state.headerValue !== headerValue) {
    const parsed = parseRuntimeTestFaultCounts({
      headerValue,
      syntheticRunHeader: '1',
    });
    if (parsed.ok === false) {
      throw new Error(parsed.error);
    }
    state = {
      headerValue,
      counts:
        (parsed.counts as Partial<Record<RuntimeTestFaultName, number>>) ?? {},
    };
    directRuntimeTestFaults.set(stateKey, state);
  }
  const count = state.counts[input.name] ?? 0;
  if (count <= 0) return false;
  if (count === 1) {
    delete state.counts[input.name];
  } else {
    state.counts[input.name] = count - 1;
  }
  state.headerValue = headerValue;
  directRuntimeTestFaults.set(stateKey, state);
  return true;
}

function runtimeTestFaultError(name: RuntimeTestFaultName): Error {
  const error = new Error(`Injected runtime test fault: ${name}.`);
  if (
    name === 'receipt_complete_write_fail' ||
    name === 'receipt_fail_write_fail'
  ) {
    error.name = 'RuntimeReceiptPersistenceError';
  }
  return error;
}

async function forceFailRuntimeWorkReceiptForRuntimeTestFault(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    error: string;
    failureKind: WorkReceiptFailureKind;
    runAttempt?: number | null;
  },
): Promise<WorkReceipt | null> {
  const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH failed AS (
          UPDATE ${workReceiptTable(session)}
             SET status = $2::smallint,
                 output = NULL,
                 error = $3,
                 error_payload = NULL,
                 failure_kind = $5::smallint,
                 lease_id = NULL,
                 lease_owner_run_id = NULL,
                 lease_owner_attempt = NULL,
                 lease_expires_at = NULL,
                 updated_at = now()
           WHERE k = decode($1, 'hex')
             AND run_id = $4
             AND COALESCE(lease_owner_attempt, 0) = $7::integer
             AND status <> $6::smallint
          RETURNING convert_from(k, 'UTF8') AS k,
                    status,
                    output,
                    error,
                    error_payload,
                    failure_kind,
                    run_id,
                    lease_id,
                    lease_owner_run_id,
                    lease_owner_attempt,
                    lease_expires_at,
                    updated_at
        )
        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at
          FROM failed
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_FAILED,
          input.error,
          input.runId,
          workReceiptFailureKindCodeForWrite(input.failureKind),
          RECEIPT_STATUS_COMPLETED,
          runAttempt,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

type DbSessionCacheEntry = {
  session: CreateDbSessionResponse;
};

type RuntimeQueryClient = Pick<RuntimeOneShotQueryClient, 'query'>;
type RuntimeDatasetRowEntry = {
  key: string;
  row: Record<string, unknown>;
  inputIndex: number;
};
type RuntimeSheetPrepareDisposition = 'pending' | 'completed' | 'blocked';
type RuntimeSheetPreparedRowDisposition = {
  key: string;
  disposition: RuntimeSheetPrepareDisposition;
};

// A failed connect retires one cached generation. It cannot end that pool
// until every request that already borrowed the generation has returned.
type CachedRuntimePostgresPool = {
  pool: RuntimePool;
  borrowers: number;
  retired: boolean;
  endPromise: Promise<void> | null;
};

type RuntimePostgresPoolLease = {
  pool: RuntimePool;
  cached: CachedRuntimePostgresPool | null;
  release(): Promise<void>;
};

type RuntimeSheetDatasetMode = 'upsert' | 'net_new';
const dbSessionCache = new Map<string, DbSessionCacheEntry>();
const dbSessionInFlight = new Map<string, Promise<CreateDbSessionResponse>>();
const postgresPools = new Map<string, CachedRuntimePostgresPool>();
const runtimePostgresAdmissions = new Map<string, RuntimePostgresAdmission>();
const runtimeWorkReceiptEnsureCache = new Map<string, Promise<void>>();
const runtimeSheetEnsureCache = new Map<
  string,
  { expiresAt: number; promise: Promise<void> }
>();
const DIRECT_POSTGRES_BATCH_SIZE = 10_000;
const RUNTIME_DB_SESSION_ROW_LIMIT_FLOOR = DIRECT_POSTGRES_BATCH_SIZE;
const APPEND_KEY_SUFFIX_LENGTH = 12;
// On-demand (create_db_session) mint TTL. Deliberately NOT aligned with
// DB_SESSION_DEFAULT_TTL_SECONDS (60 min): on-demand sessions are isolate-
// cached and re-minted through sessionHasRenewalWindow, never stored in the
// dedup DO, so they are not subject to the 410 expired-retention race the
// 60-min alignment fixed. A shorter TTL only causes periodic re-mints (cheap;
// the tenant runtime role is long-lived), not run failures.
const RUNTIME_DB_SESSION_TTL_SECONDS = 10 * 60;
const RUNTIME_WORK_RECEIPT_LEASE_COLUMNS = [
  'lease_id',
  'lease_owner_run_id',
  'lease_owner_attempt',
  'lease_expires_at',
] as const;
const RUNTIME_WORK_RECEIPT_FAILURE_KIND_COLUMN = 'failure_kind';
const RUNTIME_WORK_RECEIPT_ERROR_PAYLOAD_COLUMN = 'error_payload';
const RUNTIME_WORK_RECEIPT_SELF_HEAL_COLUMNS = [
  ...RUNTIME_WORK_RECEIPT_LEASE_COLUMNS,
  RUNTIME_WORK_RECEIPT_FAILURE_KIND_COLUMN,
  RUNTIME_WORK_RECEIPT_ERROR_PAYLOAD_COLUMN,
] as const;
const RUNTIME_SHEET_ATTEMPT_COLUMNS = [
  '_attempt_id',
  '_attempt_owner_run_id',
  '_attempt_seq',
  '_attempt_expires_at',
] as const;
const RUNTIME_SHEET_ENSURE_CACHE_TTL_MS = 10 * 60_000;
const RUNTIME_SHEET_READY_AFTER_ENSURE_RETRY_DELAYS_MS = [
  100, 250, 500, 1_000, 2_000,
] as const;
const RUNTIME_DB_SESSION_RENEWAL_WINDOW_MS = 60_000;
const RUNTIME_API_RETRY_DELAYS_MS = [
  250, 500, 1_000, 2_000, 4_000, 8_000, 8_000,
] as const;
const RUNTIME_API_RESOLVE_PLAY_RETRY_DELAYS_MS = [
  250, 500, 1_000, 2_000, 4_000, 8_000, 8_000, 15_000, 15_000, 30_000,
] as const;
const RUNTIME_API_DEFAULT_RETRY_AFTER_MS = 2_000;
const RUNTIME_API_REQUEST_TIMEOUT_MS = 30_000;
// A sheet admission can legitimately wait behind two large transactional
// writes. Keep the caller alive long enough for the FIFO admission queue to
// make progress instead of aborting a healthy queued start after 30 seconds.
const RUNTIME_SHEET_ADMISSION_REQUEST_TIMEOUT_MS = 10 * 60_000;
const RUNTIME_SHEET_COMPLETION_SERVER_OPERATION_TIMEOUT_MS = 75_000;
// Server admission is bounded to 30s and the single connection attempt to
// 10s. Keep the transport outside that envelope so it never retries while the
// first server-side write can still be active.
const RUNTIME_SHEET_COMPLETION_REQUEST_TIMEOUT_MS = 150_000;
const RUNTIME_SHEET_COMPLETION_TOTAL_TIMEOUT_MS = 180_000;
const RUNTIME_SHEET_COMPLETION_STATEMENT_TIMEOUT_MS = 60_000;
const RUNTIME_POSTGRES_SHEET_ADMISSION_TIMEOUT_MS = 30_000;
const RUNTIME_POSTGRES_PREWARM_MAX_ATTEMPTS = 4;
const RUNTIME_POSTGRES_PREWARM_RETRY_DELAYS_MS = [250, 750, 1_500] as const;
const RUNTIME_POSTGRES_CONNECT_MAX_ATTEMPTS = 4;
const RUNTIME_POSTGRES_CONNECT_RETRY_DELAYS_MS = [250, 750, 1_500] as const;
const RUNTIME_WORK_RECEIPT_QUERY_MAX_ATTEMPTS = 3;
const RUNTIME_WORK_RECEIPT_QUERY_RETRY_DELAYS_MS = [250, 750] as const;
// Runtime DB sessions are minted against the pooled tenant endpoint. A healthy
// connect is sub-second; spending minutes on one sandbox dial only hides a
// broken route and stalls the whole play. Keep retries bounded and loud.
const RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS = 10_000;
// Daytona executes one play per runner process. This is the bounded data-plane
// budget for that one run's sheet reads/writes, receipts, and terminal/export
// reads over the PgBouncer URL; it is not a scheduler/control-plane fan-out.
// One runner gets four tenant-gateway clients, split evenly so a large sheet
// flush cannot starve receipt claims (and receipt traffic cannot block the
// persistence barrier). The admission queues add a driver-independent bound:
// Neon Pool and node-postgres may queue internally, but runtime callers never
// enter those unobservable queues without first holding a lane permit.
const RUNTIME_POSTGRES_POOL_CONNECTIONS_PER_LANE = 2;
const RUNTIME_POSTGRES_RECEIPT_ADMISSION_MAX_QUEUED = 2_000;
// Sheet starts retain their parsed request body while waiting. Bound that
// memory independently from lightweight receipt requests; overflow is a
// retryable 503 and waits outside this long-lived gateway process.
const RUNTIME_POSTGRES_SHEET_ADMISSION_MAX_QUEUED = 4;
const RUNTIME_POSTGRES_RECEIPT_ADMISSION_TIMEOUT_MS =
  RUNTIME_CAPACITY_POLICY.receiptGateway.admissionTimeoutMs;
const RECEIPT_STATUS_QUEUED = RECEIPT_STATUS_CODE.queued;
const RECEIPT_STATUS_PENDING = RECEIPT_STATUS_CODE.pending;
const RECEIPT_STATUS_RUNNING = RECEIPT_STATUS_CODE.running;
const RECEIPT_STATUS_COMPLETED = RECEIPT_STATUS_CODE.completed;
const RECEIPT_STATUS_FAILED = RECEIPT_STATUS_CODE.failed;
const RECEIPT_STATUS_SKIPPED = RECEIPT_STATUS_CODE.skipped;

function runtimeSummaryTotalSql(input: {
  currentTotal: string;
  totalDelta: string;
  queued: string;
  running: string;
  completed: string;
  failed: string;
}): string {
  return `GREATEST(
    GREATEST(${input.currentTotal} + ${input.totalDelta}, 0),
    (${input.queued}) + (${input.running}) + (${input.completed}) + (${input.failed})
  )`;
}

export type ResolvedRuntimePlay = {
  playId: string;
  sourceCode?: string | null;
  artifact?: PlayBundleArtifact | null;
  codeFormat?: 'function' | 'cjs_module' | 'esm_module';
  /**
   * The runtime must receive the child contract with its artifact. Inline
   * composition rejects unresolved or dataset-backed children before running
   * their code, so dropping this field at the API boundary makes valid scalar
   * children indistinguishable from unknown ones.
   */
  staticPipeline?: PlayStaticPipeline | null;
  contractSnapshot?: Record<string, unknown> | null;
  /** Preview-only contract propagation evidence, emitted by the server and
   * logged by the runner while diagnosing a cross-runtime child resolution
   * failure. It intentionally contains no source, inputs, or credentials. */
  resolutionDiagnostics?: {
    serverPlayPipeline: boolean;
    manifestPipeline: boolean;
    responsePipeline: boolean;
  } | null;
};

export type PrepareRuntimeSheetResult = {
  inserted: number;
  skipped: number;
  pendingRows: Record<string, unknown>[];
  completedRows: Record<string, unknown>[];
  blockedRows: Record<string, unknown>[];
  tableNamespace: string;
  timings?: RuntimeSheetTiming[];
  attemptId?: string;
  attemptOwnerRunId?: string;
  attemptExpiresAt?: string;
  attemptSeq?: number;
  writeVersion?: number;
};

export type RuntimeSheetTiming = {
  phase: string;
  ms: number;
  rows?: number;
  chunks?: number;
  inserted?: number;
  skipped?: number;
  pending?: number;
  completed?: number;
  ready?: boolean;
  cached?: boolean;
  retried?: boolean;
  error?: string;
};

export type RuntimeSheetAttemptHeartbeatResult = {
  renewed: number;
  renewedKeys: string[];
  attemptExpiresAt: string | null;
};

export type RuntimeSheetAttemptReleaseResult = {
  released: number;
  releasedKeys: string[];
};

export type RuntimeWorkReceiptReleaseResult = {
  released: number;
  releasedKeys: string[];
};

type RuntimeApiActionRequest =
  | {
      action: 'resolve_play';
      playRef: string;
    }
  | {
      action: 'ensure_sheet';
      playName: string;
      runId?: string | null;
      tableNamespace: string;
      sheetContract?: PlaySheetContract | null;
      userEmail?: string | null;
    }
  | {
      action: 'create_db_session';
      playName: string;
      runId?: string | null;
      target: {
        tableNamespace: string;
        logicalTable: DbLogicalTable;
      };
      operations: DbSessionOperation[];
      limits?: {
        maxRows?: number;
        maxBytes?: number;
        maxRequests?: number;
      };
      sheetContract?: PlaySheetContract | null;
      ttlSeconds?: number;
      userEmail?: string | null;
      postgresUrlEncryption?: {
        alg: 'RSA-OAEP-256+A256GCM';
        publicKeyJwk: JsonWebKey;
      } | null;
    }
  | {
      action: 'repair_runtime_storage_grants';
      playName: string;
      runId?: string | null;
    }
  | {
      action: 'runtime_sheet_start';
      input: Parameters<typeof startRuntimeSheetDataset>[1];
    }
  | {
      action: 'runtime_sheet_complete_map_rows';
      input: Parameters<typeof completeRuntimeMapRows>[1];
    }
  | {
      action: 'runtime_sheet_read_rows';
      input: Parameters<typeof readRuntimeSheetDatasetRows>[1];
    }
  | {
      action: 'runtime_sheet_read_row_keys';
      input: Parameters<typeof readRuntimeSheetDatasetRowKeys>[1];
    }
  | {
      action: 'runtime_sheet_release_attempt';
      input: Parameters<typeof releaseRuntimeSheetAttempt>[1];
    }
  | {
      action: 'runtime_receipts_release_attempt';
      input: Parameters<typeof releaseRuntimeWorkReceipts>[1];
    };

/**
 * The Runtime Sheet persistence lane is temporarily full. This is an
 * admission signal, not a failed transport attempt: the runner must park the
 * same idempotent write and retry it after capacity becomes available.
 */
export class RuntimeApiCapacityError extends Error {
  readonly status: number;
  readonly code: 'runtime_postgres_admission_backpressure';
  readonly action: RuntimeApiActionRequest['action'];
  readonly requestId: string | null;
  readonly retryAfterMs: number;

  constructor(input: {
    status: number;
    action: RuntimeApiActionRequest['action'];
    requestId?: string | null;
    retryAfterMs: number;
    detail: string;
  }) {
    super(
      `Runtime API ${input.action} delayed by persistence capacity` +
        `${input.requestId ? ` request_id=${input.requestId}` : ''}: ${input.detail}`,
    );
    this.name = 'RuntimeApiCapacityError';
    this.status = input.status;
    this.code = 'runtime_postgres_admission_backpressure';
    this.action = input.action;
    this.requestId = input.requestId?.trim() || null;
    this.retryAfterMs = Math.max(0, Math.floor(input.retryAfterMs));
  }
}

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

export type RuntimeApiRowRecord = MapRowOutcome & {
  inputIndex?: number | null;
};

type RuntimePostgresSession = CreateDbSessionResponse & {
  postgresUrl: string;
  postgres: NonNullable<CreateDbSessionResponse['postgres']>;
};

function resolveRuntimeApiUrl(context: RuntimeApiContext): string {
  const baseUrl = context.baseUrl?.trim();
  if (!baseUrl) {
    throw new Error('Runner runtime API requires a baseUrl.');
  }
  return `${baseUrl.replace(/\/$/, '')}${PLAY_RUNTIME_API_COMPAT_PATH}`;
}

function resolveRuntimeApiHeaders(
  context: RuntimeApiContext,
  vercelHeaders: Record<string, string>,
): Record<string, string> {
  const token = context.executorToken?.trim();
  if (!token) {
    throw new Error('Runner runtime API requires an executorToken.');
  }
  return {
    'content-type': 'application/json',
    authorization: `Bearer ${token}`,
    [PLAY_RUNTIME_CONTRACT_HEADER]: String(PLAY_RUNTIME_CONTRACT),
    // A run has many callback HTTP attempts. Preserve the logical run
    // correlation in headers while Vercel keeps its own per-attempt request id.
    ...(context.runId?.trim()
      ? {
          'x-deepline-run-id': context.runId.trim(),
          'x-deepline-correlation-id': `run:${context.runId.trim()}`,
        }
      : {}),
    ...vercelHeaders,
    ...(context.runtimeTestFaultHeader?.trim()
      ? {
          [PLAY_RUNTIME_TEST_FAULT_HEADER]:
            context.runtimeTestFaultHeader.trim(),
        }
      : {}),
  };
}

function normalizeRuntimeUserEmail(
  value: string | null | undefined,
): string | null {
  const email = value?.trim();
  return email ? email : null;
}

async function postRuntimeApi<TResponse>(
  context: RuntimeApiContext,
  body: RuntimeApiActionRequest,
): Promise<TResponse> {
  const url = resolveRuntimeApiUrl(context);
  const vercelHeaders = vercelProtectionBypassHeader(
    context.vercelProtectionBypassToken,
  );
  const maxAttempts = runtimeApiMaxAttempts(body.action);
  const operationStartedAt = Date.now();
  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
    let response: Response | null = null;
    let parsed: Record<string, unknown> | null = null;
    const abortController = new AbortController();
    const operationRemainingMs = runtimeApiOperationRemainingMs(
      body.action,
      operationStartedAt,
    );
    if (operationRemainingMs <= 0) {
      throw runtimeApiTotalDeadlineError(body.action);
    }
    const timeout = setTimeout(
      () => abortController.abort(),
      Math.min(runtimeApiRequestTimeoutMs(body.action), operationRemainingMs),
    );
    try {
      response = await fetch(url, {
        method: 'POST',
        headers: resolveRuntimeApiHeaders(context, vercelHeaders),
        body: JSON.stringify(body),
        signal: context.abortSignal
          ? AbortSignal.any([context.abortSignal, abortController.signal])
          : abortController.signal,
      });
      let parsedValue: unknown;
      try {
        parsedValue = await response.json();
      } catch (error) {
        if (abortController.signal.aborted || context.abortSignal?.aborted) {
          throw error;
        }
        parsedValue = null;
      }
      parsed =
        parsedValue !== null &&
        typeof parsedValue === 'object' &&
        !Array.isArray(parsedValue)
          ? (parsedValue as Record<string, unknown>)
          : null;
    } catch (error) {
      clearTimeout(timeout);
      if (context.abortSignal?.aborted) {
        throw context.abortSignal.reason ?? error;
      }
      if (
        runtimeApiOperationRemainingMs(body.action, operationStartedAt) <= 0
      ) {
        throw runtimeApiTotalDeadlineError(body.action);
      }
      if (attempt < maxAttempts) {
        await sleepRuntimeApiRetry({
          action: body.action,
          attempt,
          operationStartedAt,
          abortSignal: context.abortSignal ?? undefined,
        });
        continue;
      }
      const message = error instanceof Error ? error.message : String(error);
      throw new Error(
        `Runtime API request to ${url} failed ${
          response
            ? 'while reading the response body'
            : 'before receiving a response'
        }: ${message}`,
      );
    } finally {
      clearTimeout(timeout);
    }

    if (response.ok) {
      if (parsed !== null) {
        return parsed as TResponse;
      }
      // A proxy can deliver the successful status line and then truncate the
      // response body. Treat that as a transport failure, not a typed success:
      // every runtime action returns a JSON object and runtime writes are
      // already retry-safe across the equivalent fetch-failed boundary.
      if (attempt < maxAttempts) {
        await sleepRuntimeApiRetry({
          action: body.action,
          attempt,
          operationStartedAt,
          abortSignal: context.abortSignal ?? undefined,
        });
        continue;
      }
      const requestId = response.headers.get('x-deepline-request-id');
      throw new Error(
        `Runtime API request to ${url} returned status ${response.status} ` +
          `without a valid JSON response body (action=${body.action}` +
          `${requestId ? `, request_id=${requestId}` : ''}).`,
      );
    }

    const retryAfterMs =
      typeof parsed?.retry_after_ms === 'number' &&
      Number.isFinite(parsed.retry_after_ms)
        ? parsed.retry_after_ms
        : RUNTIME_API_DEFAULT_RETRY_AFTER_MS;
    const details =
      typeof parsed?.detail === 'string'
        ? parsed.detail
        : typeof parsed?.details === 'string'
          ? parsed.details
          : typeof parsed?.debug_error === 'string'
            ? parsed.debug_error
            : null;
    if (
      response.status === 503 &&
      parsed?.code === 'runtime_postgres_admission_backpressure'
    ) {
      const requestId = response.headers.get('x-deepline-request-id');
      const errorMessage =
        typeof parsed.error === 'string'
          ? parsed.error
          : 'Runtime Postgres admission delayed.';
      throw new RuntimeApiCapacityError({
        status: response.status,
        action: body.action,
        requestId,
        retryAfterMs,
        detail: details ? `${errorMessage}: ${details}` : errorMessage,
      });
    }
    const shouldRetryRuntimeResponse =
      (response.status === 503 &&
        parsed?.code === 'ingestion_plane_not_ready') ||
      response.status === 408 ||
      response.status === 429 ||
      response.status === 502 ||
      response.status === 503 ||
      response.status === 504 ||
      (response.status === 500 &&
        isTransientRuntimeApiErrorMessage(
          `${typeof parsed?.error === 'string' ? parsed.error : ''}\n${details ?? ''}`,
        ));
    if (shouldRetryRuntimeResponse && attempt < maxAttempts) {
      await sleepRuntimeApiRetry({
        action: body.action,
        attempt,
        retryAfterMs,
        operationStartedAt,
        abortSignal: context.abortSignal ?? undefined,
      });
      continue;
    }

    const errorMessage =
      typeof parsed?.error === 'string'
        ? parsed.error
        : `Runtime API request failed with status ${response.status}.`;
    const errorCode = typeof parsed?.code === 'string' ? parsed.code : null;
    const errorAction =
      typeof parsed?.action === 'string' ? parsed.action : body.action;
    const requestId = response.headers.get('x-deepline-request-id');
    const diagnosticContext = [
      `status ${response.status}`,
      errorCode ? `code=${errorCode}` : null,
      errorAction ? `action=${errorAction}` : null,
      requestId ? `request_id=${requestId}` : null,
    ]
      .filter((value): value is string => Boolean(value))
      .join(', ');
    throw new Error(
      details
        ? `${errorMessage} (${diagnosticContext}): ${details}`
        : `${errorMessage} (${diagnosticContext}).`,
    );
  }

  throw new Error('Runtime API request failed after retries.');
}

function runtimeApiRetryDelayMs(
  action: RuntimeApiActionRequest['action'],
  attempt: number,
  retryAfterMs?: number,
): number {
  const configuredDelay =
    runtimeApiRetryDelaysForAction(action)[attempt - 1] ??
    RUNTIME_API_DEFAULT_RETRY_AFTER_MS;
  return retryAfterMs === undefined
    ? configuredDelay
    : Math.max(retryAfterMs, configuredDelay);
}

function runtimeApiRequestTimeoutMs(
  action: RuntimeApiActionRequest['action'],
): number {
  if (action === 'runtime_sheet_complete_map_rows') {
    return RUNTIME_SHEET_COMPLETION_REQUEST_TIMEOUT_MS;
  }
  return action === 'runtime_sheet_start'
    ? RUNTIME_SHEET_ADMISSION_REQUEST_TIMEOUT_MS
    : RUNTIME_API_REQUEST_TIMEOUT_MS;
}

function runtimeApiOperationRemainingMs(
  action: RuntimeApiActionRequest['action'],
  startedAt: number,
): number {
  return action === 'runtime_sheet_complete_map_rows'
    ? Math.max(
        0,
        RUNTIME_SHEET_COMPLETION_TOTAL_TIMEOUT_MS - (Date.now() - startedAt),
      )
    : Number.POSITIVE_INFINITY;
}

function runtimeApiTotalDeadlineError(
  action: RuntimeApiActionRequest['action'],
): Error {
  return new Error(
    `Runtime API ${action} exceeded its ` +
      `${RUNTIME_SHEET_COMPLETION_TOTAL_TIMEOUT_MS}ms total deadline.`,
  );
}

async function sleepRuntimeApiRetry(input: {
  action: RuntimeApiActionRequest['action'];
  attempt: number;
  retryAfterMs?: number;
  operationStartedAt: number;
  abortSignal?: AbortSignal;
}): Promise<void> {
  const requestedDelayMs = runtimeApiRetryDelayMs(
    input.action,
    input.attempt,
    input.retryAfterMs,
  );
  const operationRemainingMs = runtimeApiOperationRemainingMs(
    input.action,
    input.operationStartedAt,
  );
  if (operationRemainingMs <= 0) {
    throw runtimeApiTotalDeadlineError(input.action);
  }
  const delayMs = Math.min(requestedDelayMs, operationRemainingMs);
  await new Promise<void>((resolve, reject) => {
    const timeout = setTimeout(() => {
      input.abortSignal?.removeEventListener('abort', onAbort);
      resolve();
    }, delayMs);
    const onAbort = () => {
      clearTimeout(timeout);
      reject(input.abortSignal?.reason ?? new Error('Runtime API aborted.'));
    };
    if (input.abortSignal?.aborted) {
      onAbort();
      return;
    }
    input.abortSignal?.addEventListener('abort', onAbort, { once: true });
  });
  if (requestedDelayMs >= operationRemainingMs) {
    throw runtimeApiTotalDeadlineError(input.action);
  }
}

function runtimeApiRetryDelaysForAction(
  action: RuntimeApiActionRequest['action'],
): readonly number[] {
  return action === 'resolve_play'
    ? RUNTIME_API_RESOLVE_PLAY_RETRY_DELAYS_MS
    : RUNTIME_API_RETRY_DELAYS_MS;
}

function runtimeApiMaxAttempts(
  action: RuntimeApiActionRequest['action'],
): number {
  return runtimeApiRetryDelaysForAction(action).length + 1;
}

export function isTransientRuntimeApiErrorMessage(message: string): boolean {
  return /timeout exceeded when trying to connect|timed out|fetch failed|ECONNRESET|ECONNREFUSED|UND_ERR_CONNECT_TIMEOUT|requested endpoint could not be found, or you don't have access|RUNTIME_SCHEDULER_(?:SATURATED|UNAVAILABLE)|Runtime scheduler DB pool is saturated|Runtime scheduler DB circuit breaker is open/i.test(
    message,
  );
}

function getDbSessionCacheKey(input: {
  baseUrl?: string | null;
  executorToken?: string | null;
  playName: string;
  runId?: string | null;
  tableNamespace: string;
  logicalTable: DbLogicalTable;
  operations: DbSessionOperation[];
  limits?: DbSessionLimits;
  sheetContract?: PlaySheetContract | null;
  userEmail?: string | null;
}): string {
  // Worker processes are long-lived. Include a hash of the executor token plus
  // the full requested access shape so a pooled runtime cannot reuse a scoped
  // Postgres URL across orgs, runs, logical tables, or privilege sets.
  const tokenHash = createHash('sha256')
    .update(input.executorToken?.trim() ?? '')
    .digest('hex')
    .slice(0, 24);
  return [
    input.baseUrl?.trim() ?? '',
    tokenHash,
    input.playName,
    input.runId?.trim() ?? '',
    input.tableNamespace,
    input.logicalTable,
    [...input.operations].sort().join(','),
    JSON.stringify(input.limits ?? {}),
    JSON.stringify(input.sheetContract ?? null),
    input.userEmail?.trim() ?? '',
  ].join('::');
}

function getRuntimeSheetEnsureCacheKey(input: {
  baseUrl?: string | null;
  orgId?: string | null;
  playName: string;
  runId?: string | null;
  tableNamespace: string;
  sheetContract: PlaySheetContract;
  userEmail?: string | null;
}): string {
  const contractHash = createHash('sha256')
    .update(JSON.stringify(input.sheetContract))
    .digest('hex')
    .slice(0, 24);
  return [
    input.baseUrl?.trim() ?? '',
    input.orgId?.trim() ?? '',
    input.playName,
    input.runId?.trim() ?? '',
    normalizeTableNamespace(input.tableNamespace),
    contractHash,
    input.userEmail?.trim() ?? '',
  ].join('::');
}

async function isRuntimeSheetSchemaReady(
  session: RuntimePostgresSession,
  input: {
    sheetContract: PlaySheetContract;
  },
): Promise<boolean> {
  const physicalColumns = physicalSheetColumnNames(input.sheetContract);
  const selectList =
    physicalColumns.length > 0
      ? physicalColumns.map(quoteIdentifier).join(', ')
      : '1';
  const query = async (client: RuntimeQueryClient): Promise<boolean> => {
    await client.query(
      `SELECT ${selectList}
       FROM ${sheetTable(session)}
       LIMIT 0`,
    );
    const attemptColumns = await client.query<Record<string, unknown>>(
      `
        SELECT count(*)::int AS present_count
        FROM information_schema.columns
        WHERE table_schema = $1
          AND table_name = $2
          AND column_name = ANY($3::text[])
      `,
      [
        session.postgres.schema,
        session.postgres.sheetTable,
        [...RUNTIME_SHEET_ATTEMPT_COLUMNS],
      ],
    );
    return (
      Number(attemptColumns.rows[0]?.present_count ?? 0) ===
      RUNTIME_SHEET_ATTEMPT_COLUMNS.length
    );
  };
  try {
    if (isRuntimeOneShotQueryFactoryRegistered()) {
      return await withRuntimeOneShotPostgres(session, query);
    }
    return await withRuntimePostgres(session, query);
  } catch (error) {
    if (isMissingRelationError(error)) {
      return false;
    }
    throw error;
  }
}

/**
 * Report exactly what the scoped runtime role can see for the sheet table, from
 * the same connection the readiness probe uses. When ensure_sheet reports the
 * table as provisioned but the runner probe still returns not-ready, the cause
 * is almost always that the runtime role lacks a privilege on the freshly
 * created table (so information_schema.columns hides the attempt columns) or a
 * schema/table-name mismatch. This turns the previously blind "still not ready"
 * error into an actionable one that names the concrete missing piece.
 */
async function readRuntimeSheetSchemaDiagnostic(
  session: RuntimePostgresSession,
): Promise<Record<string, unknown>> {
  const query = async (
    client: RuntimeQueryClient,
  ): Promise<Record<string, unknown>> => {
    const visibleColumns = await client.query<{ column_name: string }>(
      `
        SELECT column_name
        FROM information_schema.columns
        WHERE table_schema = $1
          AND table_name = $2
        ORDER BY ordinal_position
      `,
      [session.postgres.schema, session.postgres.sheetTable],
    );
    const privileges = await client.query<Record<string, unknown>>(
      `
        SELECT
          current_user AS probe_role,
          to_regclass($3) IS NOT NULL AS regclass_visible,
          has_table_privilege(current_user, $3, 'SELECT') AS can_select,
          has_table_privilege(current_user, $3, 'INSERT') AS can_insert,
          has_table_privilege(current_user, $3, 'UPDATE') AS can_update
      `,
      [
        session.postgres.schema,
        session.postgres.sheetTable,
        `${quoteIdentifier(session.postgres.schema)}.${quoteIdentifier(session.postgres.sheetTable)}`,
      ],
    );
    const visible = new Set(visibleColumns.rows.map((row) => row.column_name));
    return {
      schema: session.postgres.schema,
      table: session.postgres.sheetTable,
      probeRole: privileges.rows[0]?.probe_role ?? null,
      regclassVisible: privileges.rows[0]?.regclass_visible ?? null,
      canSelect: privileges.rows[0]?.can_select ?? null,
      canInsert: privileges.rows[0]?.can_insert ?? null,
      canUpdate: privileges.rows[0]?.can_update ?? null,
      visibleColumnCount: visible.size,
      missingAttemptColumns: RUNTIME_SHEET_ATTEMPT_COLUMNS.filter(
        (column) => !visible.has(column),
      ),
    };
  };
  try {
    if (isRuntimeOneShotQueryFactoryRegistered()) {
      return await withRuntimeOneShotPostgres(session, query);
    }
    return await withRuntimePostgres(session, query);
  } catch (error) {
    return {
      schema: session.postgres.schema,
      table: session.postgres.sheetTable,
      diagnosticError: error instanceof Error ? error.message : String(error),
    };
  }
}

async function waitForRuntimeSheetSchemaReadyAfterEnsure(
  session: RuntimePostgresSession,
  input: {
    sheetContract: PlaySheetContract;
    timings?: RuntimeSheetTiming[];
  },
): Promise<boolean> {
  for (
    let attempt = 0;
    attempt <= RUNTIME_SHEET_READY_AFTER_ENSURE_RETRY_DELAYS_MS.length;
    attempt += 1
  ) {
    const recheckStartedAt = Date.now();
    const ready = await isRuntimeSheetSchemaReady(session, {
      sheetContract: input.sheetContract,
    });
    input.timings?.push({
      phase: 'schema_check_after_preloaded_session_ensure',
      ms: Date.now() - recheckStartedAt,
      ready,
      retried: attempt > 0,
    });
    if (ready) {
      return true;
    }
    const delay =
      RUNTIME_SHEET_READY_AFTER_ENSURE_RETRY_DELAYS_MS[attempt] ?? null;
    if (delay === null) {
      return false;
    }
    await sleep(delay);
  }
  return false;
}

async function ensureRuntimeSheetForPreloadedSession(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    sheetContract: PlaySheetContract;
    session: RuntimePostgresSession;
    timings?: RuntimeSheetTiming[];
  },
): Promise<void> {
  const cacheKey = getRuntimeSheetEnsureCacheKey({
    baseUrl: context.baseUrl,
    orgId: input.session.target.orgId,
    playName: context.playName,
    runId: context.runId,
    tableNamespace: input.tableNamespace,
    sheetContract: input.sheetContract,
    userEmail: normalizeRuntimeUserEmail(context.userEmail),
  });
  const now = Date.now();
  const cached = runtimeSheetEnsureCache.get(cacheKey);
  if (cached && cached.expiresAt > now) {
    input.timings?.push({
      phase: 'ensure_sheet_for_preloaded_session_cached',
      ms: 0,
      cached: true,
    });
    await cached.promise;
    return;
  }
  if (cached) {
    runtimeSheetEnsureCache.delete(cacheKey);
  }
  const checkStartedAt = Date.now();
  let ready: boolean;
  try {
    ready = await isRuntimeSheetSchemaReady(input.session, {
      sheetContract: input.sheetContract,
    });
  } catch (error) {
    if (!isPostgresPermissionDeniedError(error)) {
      throw error;
    }
    const repairStartedAt = Date.now();
    await repairRuntimeStorageGrants(context, {
      playName: context.playName,
    });
    input.timings?.push({
      phase: 'repair_storage_grants_after_schema_check_permission_error',
      ms: Date.now() - repairStartedAt,
      retried: true,
    });
    ready = await isRuntimeSheetSchemaReady(input.session, {
      sheetContract: input.sheetContract,
    });
  }
  input.timings?.push({
    phase: 'schema_check_for_preloaded_session',
    ms: Date.now() - checkStartedAt,
    ready,
  });
  if (ready) {
    runtimeSheetEnsureCache.set(cacheKey, {
      expiresAt: now + RUNTIME_SHEET_ENSURE_CACHE_TTL_MS,
      promise: Promise.resolve(),
    });
    return;
  }
  const ensureStartedAt = Date.now();
  const promise = (async () => {
    await ensureRuntimeSheet(context, {
      playName: context.playName,
      tableNamespace: input.tableNamespace,
      sheetContract: input.sheetContract,
    });
    input.timings?.push({
      phase: 'ensure_sheet_for_preloaded_session',
      ms: Date.now() - ensureStartedAt,
    });
    let readyAfterEnsure = await waitForRuntimeSheetSchemaReadyAfterEnsure(
      input.session,
      {
        sheetContract: input.sheetContract,
        timings: input.timings,
      },
    );
    if (!readyAfterEnsure) {
      // ensure_sheet provisions the sheet as the tenant admin role, but the
      // preloaded session probes as the scoped runtime role. When the tenant
      // role contract was only partially applied (e.g. role-membership grants
      // skipped on a data plane where the owner cannot administer roles), the
      // runtime role never inherits access to the freshly created table, so its
      // information_schema view hides the attempt columns and the probe stays
      // not-ready even though the table exists. Re-apply the storage grants
      // through the same self-heal the pre-ensure probe uses, then re-check
      // once. This is a grant repair, not a blind retry: if the runtime role
      // still cannot see the sheet, we fail loudly below with the concrete
      // runtime-role view.
      const repairStartedAt = Date.now();
      await repairRuntimeStorageGrants(context, {
        playName: context.playName,
      });
      input.timings?.push({
        phase: 'repair_storage_grants_after_preloaded_session_ensure',
        ms: Date.now() - repairStartedAt,
        retried: true,
      });
      readyAfterEnsure = await waitForRuntimeSheetSchemaReadyAfterEnsure(
        input.session,
        {
          sheetContract: input.sheetContract,
          timings: input.timings,
        },
      );
    }
    if (!readyAfterEnsure) {
      const diagnostic = await readRuntimeSheetSchemaDiagnostic(input.session);
      throw new Error(
        `Runtime sheet schema for ctx.dataset("${input.tableNamespace}") is still not ready after ensure_sheet. ` +
          `Runtime-role view: ${JSON.stringify(diagnostic)}`,
      );
    }
  })();
  runtimeSheetEnsureCache.set(cacheKey, {
    expiresAt: now + RUNTIME_SHEET_ENSURE_CACHE_TTL_MS,
    promise,
  });
  try {
    await promise;
  } catch (error) {
    if (runtimeSheetEnsureCache.get(cacheKey)?.promise === promise) {
      runtimeSheetEnsureCache.delete(cacheKey);
    }
    throw error;
  }
}

function operationsSatisfyRequest(
  candidate: readonly DbSessionOperation[],
  requested: readonly DbSessionOperation[],
): boolean {
  const candidateOperations = new Set(candidate);
  return requested.every((operation) => candidateOperations.has(operation));
}

function limitsSatisfyRequest(
  candidate: DbSessionLimits | undefined,
  requested: DbSessionLimits | undefined,
): boolean {
  const candidateLimits = candidate ?? {};
  const requestedLimits = requested ?? {};
  for (const key of ['maxRows', 'maxBytes', 'maxRequests'] as const) {
    const requestedValue = requestedLimits[key];
    if (requestedValue === undefined) {
      continue;
    }
    const candidateValue = candidateLimits[key];
    // A preloaded session with no advisory limit means the launch contract
    // intentionally authorized the whole run for this target. Runtime scope is
    // enforced by signed session metadata plus constructed SQL identifiers;
    // this fast path does not create per-session database roles or grants.
    if (candidateValue === undefined) {
      continue;
    }
    if (candidateValue < requestedValue) {
      return false;
    }
  }
  return true;
}

function requirePreloadedRuntimeDbSessionOrgId(
  context: Pick<RuntimeApiContext, 'orgId'>,
): string {
  const orgId = context.orgId?.trim();
  if (!orgId) {
    throw new Error(
      'Preloaded Runtime DB sessions require an orgId to validate session scope.',
    );
  }
  return orgId;
}

function shouldEnsureRuntimeSheetBeforeSession(input: {
  logicalTable: DbLogicalTable;
  operations: DbSessionOperation[];
  sheetContract?: PlaySheetContract | null;
}): input is typeof input & { sheetContract: PlaySheetContract } {
  return (
    input.logicalTable === 'sheet_rows' &&
    !!input.sheetContract &&
    input.operations.some((operation) => operation !== 'rows.read')
  );
}

function sessionHasRenewalWindow(session: CreateDbSessionResponse): boolean {
  const expiresAtMs = Date.parse(session.expiresAt);
  return (
    Number.isFinite(expiresAtMs) &&
    expiresAtMs - Date.now() > RUNTIME_DB_SESSION_RENEWAL_WINDOW_MS
  );
}

function preloadedSessionMatchesRequest(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    logicalTable: DbLogicalTable;
    operations: DbSessionOperation[];
    limits?: DbSessionLimits;
  },
  preloaded: PreloadedRuntimeDbSession,
  expectedOrgId: string,
): boolean {
  const session = preloaded.session;
  const requestedTableNamespace = normalizeTableNamespace(input.tableNamespace);
  return (
    sessionHasRenewalWindow(session) &&
    session.target.orgId === expectedOrgId &&
    session.playName === context.playName &&
    normalizeTableNamespace(preloaded.tableNamespace) ===
      requestedTableNamespace &&
    normalizeTableNamespace(session.target.tableNamespace) ===
      requestedTableNamespace &&
    preloaded.logicalTable === input.logicalTable &&
    session.target.logicalTable === input.logicalTable &&
    operationsSatisfyRequest(preloaded.operations, input.operations) &&
    operationsSatisfyRequest(session.operations, input.operations) &&
    limitsSatisfyRequest(preloaded.limits, input.limits) &&
    limitsSatisfyRequest(session.limits, input.limits)
  );
}

function findPreloadedRuntimeDbSession(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    logicalTable: DbLogicalTable;
    operations: DbSessionOperation[];
    limits?: DbSessionLimits;
  },
): CreateDbSessionResponse | null {
  const preloadedSessions = context.preloadedDbSessions ?? [];
  if (preloadedSessions.length === 0) {
    return null;
  }
  const expectedOrgId = requirePreloadedRuntimeDbSessionOrgId(context);
  for (const preloaded of preloadedSessions) {
    const parsed = createDbSessionResponseSchema.safeParse(preloaded.session);
    if (!parsed.success) {
      continue;
    }
    const candidate = {
      ...preloaded,
      session: parsed.data,
    };
    if (
      preloadedSessionMatchesRequest(context, input, candidate, expectedOrgId)
    ) {
      return candidate.session;
    }
  }
  return null;
}

async function unwrapRuntimeDbSession(
  context: RuntimeApiContext,
  session: CreateDbSessionResponse,
  decryptionKey?: PostgresUrlDecryptionKey | null,
): Promise<CreateDbSessionResponse> {
  if (session.postgresUrl) {
    if (context.postgresSessionUnwrapKey?.trim()) {
      throw new Error(
        'Harness preloaded Runtime DB sessions must carry an encryptedPostgresUrl, not a raw postgresUrl.',
      );
    }
    return session;
  }
  if (!session.encryptedPostgresUrl) {
    return session;
  }
  if (session.encryptedPostgresUrl.alg === 'RSA-OAEP-256+A256GCM') {
    if (!decryptionKey) {
      throw new Error(
        'Runtime DB session response used public-key encryption, but no private key was retained for unwrap.',
      );
    }
    const {
      encryptedPostgresUrl: _encryptedPostgresUrl,
      ...sessionWithoutUrl
    } = session;
    void _encryptedPostgresUrl;
    return {
      ...sessionWithoutUrl,
      postgresUrl: await decryptDbSessionPostgresUrlWithPrivateKey({
        encrypted: session.encryptedPostgresUrl,
        privateKey: decryptionKey.privateKey,
        aad: dbSessionPostgresUrlAad(sessionWithoutUrl),
      }),
    };
  }
  if (decryptionKey) {
    throw new Error(
      'Runtime DB session response used the shared-secret envelope after the runner requested public-key encryption.',
    );
  }
  const unwrapKey = context.postgresSessionUnwrapKey?.trim();
  if (!unwrapKey) {
    throw new Error(
      'Runtime DB session response is encrypted, but no harness unwrap key was provided.',
    );
  }
  const { encryptedPostgresUrl: _encryptedPostgresUrl, ...sessionWithoutUrl } =
    session;
  void _encryptedPostgresUrl;
  return {
    ...sessionWithoutUrl,
    postgresUrl: await decryptDbSessionPostgresUrl({
      encrypted: session.encryptedPostgresUrl,
      secret: unwrapKey,
      aad: dbSessionPostgresUrlAad(sessionWithoutUrl),
    }),
  };
}

async function getRuntimeDbSession(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    logicalTable: DbLogicalTable;
    operations: DbSessionOperation[];
    limits?: DbSessionLimits;
    sheetContract?: PlaySheetContract | null;
    timings?: RuntimeSheetTiming[];
  },
): Promise<CreateDbSessionResponse> {
  const userEmail = normalizeRuntimeUserEmail(context.userEmail);
  const cacheKey = getDbSessionCacheKey({
    baseUrl: context.baseUrl,
    executorToken: context.executorToken,
    playName: context.playName,
    runId: context.runId,
    tableNamespace: input.tableNamespace,
    logicalTable: input.logicalTable,
    operations: input.operations,
    limits: input.limits,
    sheetContract: input.sheetContract,
    userEmail,
  });
  const cached = dbSessionCache.get(cacheKey)?.session;
  if (cached) {
    if (sessionHasRenewalWindow(cached)) {
      return cached;
    }
    await deleteRuntimeDbSessionCacheEntry(cacheKey, cached);
  }

  const pending = dbSessionInFlight.get(cacheKey);
  if (pending) {
    return await pending;
  }

  const sessionPromise = (async (): Promise<CreateDbSessionResponse> => {
    const preloaded = findPreloadedRuntimeDbSession(context, input);
    if (preloaded) {
      const unwrappedPreloaded = await unwrapRuntimeDbSession(
        context,
        preloaded,
      );
      if (input.logicalTable === 'sheet_rows' && input.sheetContract) {
        await ensureRuntimeSheetForPreloadedSession(context, {
          tableNamespace: input.tableNamespace,
          sheetContract: input.sheetContract,
          session: requireRuntimePostgresSession(unwrappedPreloaded),
          timings: input.timings,
        });
      }
      dbSessionCache.set(cacheKey, { session: unwrappedPreloaded });
      return unwrappedPreloaded;
    }

    if (context.dbSessionStrategy !== 'trusted_dynamic') {
      throw new Error(
        `RUNTIME_DB_SESSION_PRELOAD_INCOMPLETE: no preloaded Runtime DB session matches play=${context.playName} table=${normalizeTableNamespace(input.tableNamespace)} logical_table=${input.logicalTable}. Dynamic database authority requires an explicit trusted control-plane context.`,
      );
    }

    if (shouldEnsureRuntimeSheetBeforeSession(input)) {
      await ensureRuntimeSheet(context, {
        playName: context.playName,
        tableNamespace: input.tableNamespace,
        sheetContract: input.sheetContract,
      });
    }

    const decryptionKey = await generateDbSessionPostgresUrlDecryptionKey();
    const response = await unwrapRuntimeDbSession(
      context,
      createDbSessionResponseSchema.parse(
        await postRuntimeApi<CreateDbSessionResponse>(context, {
          action: 'create_db_session',
          playName: context.playName,
          runId: context.runId ?? null,
          target: {
            tableNamespace: input.tableNamespace,
            logicalTable: input.logicalTable,
          },
          operations: input.operations,
          limits: input.limits,
          sheetContract: input.sheetContract ?? null,
          ttlSeconds: RUNTIME_DB_SESSION_TTL_SECONDS,
          userEmail,
          postgresUrlEncryption: decryptionKey.request,
        }),
      ),
      decryptionKey,
    );
    dbSessionCache.set(cacheKey, { session: response });
    return response;
  })();
  dbSessionInFlight.set(cacheKey, sessionPromise);
  try {
    return await sessionPromise;
  } finally {
    if (dbSessionInFlight.get(cacheKey) === sessionPromise) {
      dbSessionInFlight.delete(cacheKey);
    }
  }
}

async function deleteRuntimeDbSessionCacheEntry(
  cacheKey: string,
  session: CreateDbSessionResponse,
): Promise<void> {
  dbSessionCache.delete(cacheKey);
  if (session.postgresUrl) {
    await resetRuntimePostgresPool(session.postgresUrl);
  }
}

function requireRuntimePostgresSession(
  session: CreateDbSessionResponse,
): RuntimePostgresSession {
  if (!session.postgresUrl || !session.postgres) {
    throw new Error(
      'Runtime DB session did not include a scoped Postgres URL. Direct Postgres sheet IO is required.',
    );
  }
  return session as RuntimePostgresSession;
}

function validatePreloadedRuntimeDbSessionScope(
  context: RuntimeApiContext,
  session: CreateDbSessionResponse,
): void {
  const expectedPlayName = context.playName?.trim();
  if (!expectedPlayName) {
    throw new Error(
      'Preloaded Runtime DB sessions require a playName to validate session scope.',
    );
  }
  if (session.playName !== expectedPlayName) {
    throw new Error(
      'Preloaded Runtime DB session is outside the requested play scope.',
    );
  }
  const expectedOrgId = requirePreloadedRuntimeDbSessionOrgId(context);
  if (session.target.orgId !== expectedOrgId) {
    throw new Error(
      'Preloaded Runtime DB session is outside the requested org scope.',
    );
  }
}

function runtimeDbSessionRowLimit(rowCount: number): number {
  return Math.max(
    Math.max(1, Math.floor(rowCount)),
    RUNTIME_DB_SESSION_ROW_LIMIT_FLOOR,
  );
}

export async function prewarmRuntimePostgresSessions(
  context: RuntimeApiContext,
): Promise<void> {
  const sessions = context.preloadedDbSessions ?? [];
  if (sessions.length === 0) {
    return;
  }
  for (const preloaded of sessions) {
    const parsed = createDbSessionResponseSchema.parse(preloaded.session);
    validatePreloadedRuntimeDbSessionScope(context, parsed);
    const session = requireRuntimePostgresSession(
      await unwrapRuntimeDbSession(context, parsed),
    );
    if (isRuntimeOneShotQueryFactoryRegistered()) {
      await prewarmRuntimeOneShotPostgresSession(session);
    } else {
      await prewarmRuntimePostgresSession(session);
    }
  }
}

async function prewarmRuntimeOneShotPostgresSession(
  session: RuntimePostgresSession,
): Promise<void> {
  for (
    let attempt = 1;
    attempt <= RUNTIME_POSTGRES_PREWARM_MAX_ATTEMPTS;
    attempt += 1
  ) {
    try {
      await createRuntimeOneShotQueryClient({
        connectionString: session.postgresUrl,
      }).query('SELECT 1');
      return;
    } catch (error) {
      if (
        attempt >= RUNTIME_POSTGRES_PREWARM_MAX_ATTEMPTS ||
        !isTransientRuntimePostgresConnectionError(error)
      ) {
        throw error;
      }
      await sleep(
        RUNTIME_POSTGRES_PREWARM_RETRY_DELAYS_MS[attempt - 1] ??
          RUNTIME_POSTGRES_PREWARM_RETRY_DELAYS_MS[
            RUNTIME_POSTGRES_PREWARM_RETRY_DELAYS_MS.length - 1
          ],
      );
    }
  }
}

async function prewarmRuntimePostgresSession(
  session: RuntimePostgresSession,
): Promise<void> {
  for (
    let attempt = 1;
    attempt <= RUNTIME_POSTGRES_PREWARM_MAX_ATTEMPTS;
    attempt += 1
  ) {
    try {
      await withRuntimePostgres(
        session,
        async (client) => {
          await client.query('SELECT 1');
        },
        { maxConnectAttempts: 1 },
      );
      return;
    } catch (error) {
      if (
        attempt >= RUNTIME_POSTGRES_PREWARM_MAX_ATTEMPTS ||
        !isTransientRuntimePostgresConnectionError(error)
      ) {
        await resetRuntimePostgresPool(session.postgresUrl);
        throw error;
      }
      await resetRuntimePostgresPool(session.postgresUrl);
      await sleep(
        RUNTIME_POSTGRES_PREWARM_RETRY_DELAYS_MS[attempt - 1] ??
          RUNTIME_POSTGRES_PREWARM_RETRY_DELAYS_MS[
            RUNTIME_POSTGRES_PREWARM_RETRY_DELAYS_MS.length - 1
          ],
      );
    }
  }
}

function isTransientRuntimePostgresConnectionError(error: unknown): boolean {
  if (!error || typeof error !== 'object') {
    return false;
  }
  const nestedErrors = (error as { errors?: unknown }).errors;
  if (
    Array.isArray(nestedErrors) &&
    nestedErrors.some(isTransientRuntimePostgresConnectionError)
  ) {
    return true;
  }
  const code = 'code' in error ? String(error.code) : '';
  if (
    code === 'ECONNRESET' ||
    code === 'ETIMEDOUT' ||
    code === 'ECONNREFUSED' ||
    code === 'EAI_AGAIN' ||
    code === 'EAI_FAIL' ||
    code === 'ENOTFOUND' ||
    code === 'ESERVFAIL' ||
    code === '57P01'
  ) {
    return true;
  }
  const name = 'name' in error ? String(error.name) : '';
  const message = 'message' in error ? String(error.message) : '';
  return /connection (terminated|timeout|timed out|closed|reset)|ETIMEDOUT|ECONNRESET|ECONNREFUSED|UND_ERR_CONNECT_TIMEOUT|EAI_AGAIN|EAI_FAIL|ENOTFOUND|ESERVFAIL|getaddrinfo|\bdns\b/i.test(
    `${name} ${message} ${String(error)}`,
  );
}

function isTransientRuntimePostgresOperationError(error: unknown): boolean {
  if (isTransientRuntimePostgresConnectionError(error)) {
    return true;
  }
  if (!error || typeof error !== 'object') {
    return false;
  }
  const nestedErrors = (error as { errors?: unknown }).errors;
  if (
    Array.isArray(nestedErrors) &&
    nestedErrors.some(isTransientRuntimePostgresOperationError)
  ) {
    return true;
  }
  const name = 'name' in error ? String(error.name) : '';
  const message = 'message' in error ? String(error.message) : '';
  return /fetch failed|network error|socket hang up|connection (terminated|timeout|timed out|closed|reset)|ETIMEDOUT|ECONNRESET|ECONNREFUSED|UND_ERR_CONNECT_TIMEOUT/i.test(
    `${name} ${message} ${String(error)}`,
  );
}

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

async function connectRuntimePostgresPool(
  pool: RuntimePool,
  admission: RuntimePostgresAdmission,
  lane: RuntimePostgresLane,
  postgresUrl: string,
  signal?: AbortSignal | null,
): Promise<RuntimePoolClient> {
  const admissionStartedAt = Date.now();
  const releaseAdmission = await admission.acquire(lane, {
    signal: signal ?? undefined,
  });
  const admissionWaitMs = Math.max(0, Date.now() - admissionStartedAt);
  if (admissionWaitMs > 0) {
    const telemetry = admission.snapshot()[lane];
    console.info('[runtime-postgres-admission]', {
      event: 'acquired_after_wait',
      lane,
      waitMs: admissionWaitMs,
      active: telemetry.active,
      queued: telemetry.queued,
      poolKeyHash: createHash('sha256')
        .update(postgresUrl)
        .digest('hex')
        .slice(0, 12),
      phase: 'pool_connect',
    });
  }
  let timedOut = false;
  let timeout: ReturnType<typeof setTimeout> | null = null;
  const connectPromise = pool.connect();
  connectPromise
    .then((client) => {
      if (timedOut) {
        client.release();
      }
    })
    .catch(() => {});
  try {
    const client = await Promise.race([
      connectPromise,
      new Promise<never>((_, reject) => {
        timeout = setTimeout(() => {
          timedOut = true;
          // Do not make a timed-out driver connect pin an admission permit.
          // A late connect still releases its client in the handler above.
          releaseAdmission();
          reject(
            new Error(
              `Runtime Postgres connection timed out after ${RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS}ms.`,
            ),
          );
        }, RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS);
      }),
    ]);
    let released = false;
    return {
      query: (text, params) => client.query(text, params),
      release: (destroy = false) => {
        if (released) return;
        released = true;
        client.release(destroy);
        releaseAdmission();
      },
      destroy: (error) => {
        if (released) return;
        released = true;
        if (client.destroy) client.destroy(error);
        else client.release(true);
        releaseAdmission();
      },
    };
  } catch (error) {
    if (!timedOut) {
      releaseAdmission();
    }
    throw error;
  } finally {
    if (timeout) {
      clearTimeout(timeout);
    }
  }
}

function runtimePostgresLane(
  session: RuntimePostgresSession,
): RuntimePostgresLane {
  return session.target.logicalTable === RUNTIME_WORK_RECEIPT_LOGICAL_TABLE
    ? 'receipts'
    : 'sheets';
}

function runtimePostgresPoolKey(
  postgresUrl: string,
  lane: RuntimePostgresLane,
): string {
  return `${postgresUrl}::${lane}`;
}

function getRuntimePostgresAdmission(
  postgresUrl: string,
): RuntimePostgresAdmission {
  const existing = runtimePostgresAdmissions.get(postgresUrl);
  if (existing) return existing;
  const admission = new RuntimePostgresAdmission({
    maxActivePerLane: RUNTIME_POSTGRES_POOL_CONNECTIONS_PER_LANE,
    maxQueuedPerLane: {
      receipts: RUNTIME_POSTGRES_RECEIPT_ADMISSION_MAX_QUEUED,
      sheets: RUNTIME_POSTGRES_SHEET_ADMISSION_MAX_QUEUED,
    },
    // Capacity is backpressure, not a sheet-persistence failure. Bound both
    // queues so the gateway can return typed capacity before an HTTP caller
    // times out. The runner then parks the same frozen write and retries it;
    // it never leaves an invisible request queued behind a dead client.
    acquireTimeoutMs: {
      receipts: RUNTIME_POSTGRES_RECEIPT_ADMISSION_TIMEOUT_MS,
      sheets: RUNTIME_POSTGRES_SHEET_ADMISSION_TIMEOUT_MS,
    },
  });
  runtimePostgresAdmissions.set(postgresUrl, admission);
  return admission;
}

export function runtimePostgresAdmissionTelemetry(
  postgresUrl: string,
): RuntimePostgresAdmissionSnapshot | null {
  return runtimePostgresAdmissions.get(postgresUrl)?.snapshot() ?? null;
}

function endRetiredRuntimePostgresPool(
  cached: CachedRuntimePostgresPool,
): Promise<void> {
  if (!cached.retired || cached.borrowers > 0) {
    return Promise.resolve();
  }
  cached.endPromise ??= Promise.resolve()
    .then(() => cached.pool.end())
    .catch(() => {});
  return cached.endPromise;
}

async function retireRuntimePostgresPool(
  poolKey: string,
  cached: CachedRuntimePostgresPool,
): Promise<void> {
  if (postgresPools.get(poolKey) === cached) {
    postgresPools.delete(poolKey);
  }
  cached.retired = true;
  await endRetiredRuntimePostgresPool(cached);
}

function getPostgresPoolLease(
  postgresUrl: string,
  lane: RuntimePostgresLane,
  cachePool = true,
): RuntimePostgresPoolLease {
  if (!cachePool) {
    return {
      pool: createRuntimePool({
        connectionString: postgresUrl,
        maxConnections: RUNTIME_POSTGRES_POOL_CONNECTIONS_PER_LANE,
        idleTimeoutMs: 0,
        connectTimeoutMs: RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS,
      }),
      cached: null,
      release: async () => {},
    };
  }
  const poolKey = runtimePostgresPoolKey(postgresUrl, lane);
  let cached = postgresPools.get(poolKey);
  if (!cached) {
    cached = {
      pool: createRuntimePool({
        connectionString: postgresUrl,
        maxConnections: RUNTIME_POSTGRES_POOL_CONNECTIONS_PER_LANE,
        idleTimeoutMs: 15_000,
        connectTimeoutMs: RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS,
      }),
      borrowers: 0,
      retired: false,
      endPromise: null,
    };
    postgresPools.set(poolKey, cached);
  }
  cached.borrowers += 1;
  let released = false;
  return {
    pool: cached.pool,
    cached,
    release: async () => {
      if (released) return;
      released = true;
      cached.borrowers -= 1;
      await endRetiredRuntimePostgresPool(cached);
    },
  };
}

async function resetRuntimePostgresPool(postgresUrl: string): Promise<void> {
  const retirements = (['receipts', 'sheets'] as const)
    .map((lane) => runtimePostgresPoolKey(postgresUrl, lane))
    .map((key) => {
      const cached = postgresPools.get(key);
      if (!cached) return null;
      return retireRuntimePostgresPool(key, cached);
    })
    .filter((retirement): retirement is Promise<void> => retirement !== null);
  await Promise.all(retirements);
}

async function awaitRuntimePostgresOperation<T>(input: {
  client: RuntimePoolClient;
  deadlineAt: number | null;
  operation: Promise<T>;
  signal?: AbortSignal | null;
}): Promise<T> {
  if (input.deadlineAt === null && !input.signal) return await input.operation;
  const remainingMs =
    input.deadlineAt === null
      ? null
      : Math.max(0, input.deadlineAt - Date.now());
  if (remainingMs === 0 || input.signal?.aborted) {
    const error =
      input.signal?.reason ??
      new Error('Runtime Postgres operation exceeded its total deadline.');
    if (input.client.destroy) input.client.destroy(error);
    else input.client.release(true);
    throw error;
  }
  return await new Promise<T>((resolve, reject) => {
    let timeout: ReturnType<typeof setTimeout> | null = null;
    const cleanup = () => {
      if (timeout) clearTimeout(timeout);
      input.signal?.removeEventListener('abort', onAbort);
    };
    const rejectAndDestroy = (error: unknown) => {
      const normalizedError =
        error instanceof Error ? error : new Error(String(error));
      if (input.client.destroy) input.client.destroy(normalizedError);
      else input.client.release(true);
      cleanup();
      reject(error);
    };
    const onAbort = () =>
      rejectAndDestroy(
        input.signal?.reason ??
          new Error('Runtime Postgres operation aborted.'),
      );
    if (remainingMs !== null) {
      timeout = setTimeout(
        () =>
          rejectAndDestroy(
            new Error(
              `Runtime Postgres operation exceeded its ${remainingMs}ms remaining deadline.`,
            ),
          ),
        remainingMs,
      );
    }
    input.signal?.addEventListener('abort', onAbort, { once: true });
    input.operation.then(
      (value) => {
        cleanup();
        resolve(value);
      },
      (error) => {
        cleanup();
        reject(error);
      },
    );
  });
}

async function withRuntimePostgres<T>(
  session: RuntimePostgresSession,
  fn: (
    client: RuntimePoolClient,
    transaction: { active: boolean },
  ) => Promise<T>,
  options: {
    cachePool?: boolean;
    maxConnectAttempts?: number;
    operationTimeoutMs?: number;
    signal?: AbortSignal | null;
  } = {},
): Promise<T> {
  let client: RuntimePoolClient | null = null;
  let requestLocalPool: RuntimePool | null = null;
  let poolLease: RuntimePostgresPoolLease | null = null;
  let roleTransactionStarted = false;
  let operationDeadlineAt: number | null = null;
  const cachePool =
    (options.cachePool ?? true) && canReuseRuntimePostgresPoolsAcrossRequests();
  const maxConnectAttempts =
    options.maxConnectAttempts === undefined
      ? RUNTIME_POSTGRES_CONNECT_MAX_ATTEMPTS
      : Math.max(1, Math.floor(options.maxConnectAttempts));
  const lane = runtimePostgresLane(session);
  const admission = getRuntimePostgresAdmission(session.postgresUrl);
  for (let attempt = 1; attempt <= maxConnectAttempts; attempt += 1) {
    try {
      poolLease = getPostgresPoolLease(session.postgresUrl, lane, cachePool);
      const pool = poolLease.pool;
      if (!cachePool) {
        requestLocalPool = pool;
      }
      client = await connectRuntimePostgresPool(
        pool,
        admission,
        lane,
        session.postgresUrl,
        options.signal,
      );
      operationDeadlineAt =
        options.operationTimeoutMs === undefined
          ? null
          : Date.now() + Math.max(1, Math.floor(options.operationTimeoutMs));
      if (session.executionRole) {
        await awaitRuntimePostgresOperation({
          client,
          deadlineAt: operationDeadlineAt,
          operation: client.query('BEGIN'),
          signal: options.signal,
        });
        roleTransactionStarted = true;
        await awaitRuntimePostgresOperation({
          client,
          deadlineAt: operationDeadlineAt,
          operation: client.query(
            `SET LOCAL ROLE ${quoteIdentifier(session.executionRole)}`,
          ),
          signal: options.signal,
        });
      }
      break;
    } catch (error) {
      if (client) {
        if (roleTransactionStarted) {
          await awaitRuntimePostgresOperation({
            client,
            deadlineAt: operationDeadlineAt,
            operation: client.query('ROLLBACK'),
            signal: options.signal,
          }).catch(() => {});
          roleTransactionStarted = false;
        }
        client.release();
        client = null;
      }
      // Queue overflow and caller cancellation happen before pool.connect().
      // They describe admission state, not a broken shared pool. Ending that
      // pool would invalidate the active clients and FIFO waiters that caused
      // the backpressure in the first place.
      const preserveSharedPool =
        cachePool && isRuntimePostgresAdmissionError(error);
      if (cachePool && !preserveSharedPool && poolLease?.cached) {
        const poolKey = runtimePostgresPoolKey(session.postgresUrl, lane);
        await retireRuntimePostgresPool(poolKey, poolLease.cached);
      } else if (requestLocalPool) {
        await Promise.resolve(requestLocalPool.end()).catch(() => {});
        requestLocalPool = null;
      }
      await poolLease?.release();
      poolLease = null;
      if (
        attempt >= maxConnectAttempts ||
        !isTransientRuntimePostgresConnectionError(error)
      ) {
        throw error;
      }
      await sleep(
        RUNTIME_POSTGRES_CONNECT_RETRY_DELAYS_MS[attempt - 1] ??
          RUNTIME_POSTGRES_CONNECT_RETRY_DELAYS_MS[
            RUNTIME_POSTGRES_CONNECT_RETRY_DELAYS_MS.length - 1
          ],
      );
    }
  }
  if (!client) {
    throw new Error('Runtime Postgres connection was not acquired.');
  }
  try {
    const result = await awaitRuntimePostgresOperation({
      client,
      deadlineAt: operationDeadlineAt,
      operation: fn(client, { active: roleTransactionStarted }),
      signal: options.signal,
    });
    if (roleTransactionStarted) {
      await awaitRuntimePostgresOperation({
        client,
        deadlineAt: operationDeadlineAt,
        operation: client.query('COMMIT'),
        signal: options.signal,
      });
      roleTransactionStarted = false;
    }
    return result;
  } catch (error) {
    if (roleTransactionStarted) {
      await awaitRuntimePostgresOperation({
        client,
        deadlineAt: operationDeadlineAt,
        operation: client.query('ROLLBACK'),
        signal: options.signal,
      }).catch(() => {});
      roleTransactionStarted = false;
    }
    throw error;
  } finally {
    client.release();
    await poolLease?.release();
    if (requestLocalPool) {
      await Promise.resolve(requestLocalPool.end()).catch(() => {});
    }
  }
}

async function withRuntimeOneShotPostgres<T>(
  session: RuntimePostgresSession,
  operation: (client: RuntimeOneShotQueryClient) => Promise<T>,
): Promise<T> {
  for (
    let attempt = 1;
    attempt <= RUNTIME_POSTGRES_CONNECT_MAX_ATTEMPTS;
    attempt += 1
  ) {
    try {
      return await operation(
        createRuntimeOneShotQueryClient({
          connectionString: session.postgresUrl,
        }),
      );
    } catch (error) {
      if (
        attempt >= RUNTIME_POSTGRES_CONNECT_MAX_ATTEMPTS ||
        !isTransientRuntimePostgresConnectionError(error)
      ) {
        throw error;
      }
      await sleep(
        RUNTIME_POSTGRES_CONNECT_RETRY_DELAYS_MS[attempt - 1] ??
          RUNTIME_POSTGRES_CONNECT_RETRY_DELAYS_MS[
            RUNTIME_POSTGRES_CONNECT_RETRY_DELAYS_MS.length - 1
          ],
      );
    }
  }
  throw new Error('Runtime Postgres one-shot connection was not acquired.');
}

/**
 * True when a Postgres error means the backing table/column is not provisioned
 * in the session's *physical* database — `42P01` (undefined_table), `42703`
 * (undefined_column), or the equivalent messages.
 *
 * Both data planes (sheet rows and work receipts) treat this as "re-provision,
 * then retry once" rather than a hard failure, because a cached "already
 * ensured" entry can outlive the physical relation it stands for — a reset Neon
 * branch, or an ensure cache primed in one Worker isolate while the operation
 * runs against another. This is the single classifier that keeps both planes
 * self-healing; do not fork it.
 */
function isMissingRelationError(error: unknown): boolean {
  if (!error || typeof error !== 'object') {
    return false;
  }
  const code = 'code' in error ? String(error.code) : '';
  if (code === '42P01' || code === '42703') {
    return true;
  }
  const message = 'message' in error ? String(error.message) : '';
  return (
    /relation .* does not exist/i.test(message) ||
    /column .* does not exist/i.test(message)
  );
}

function isPostgresPermissionDeniedError(error: unknown): boolean {
  if (error === null || typeof error !== 'object') {
    return false;
  }
  if ('code' in error && String(error.code) === '42501') {
    return true;
  }
  const message =
    'message' in error && typeof error.message === 'string'
      ? error.message
      : '';
  // Postgres errors raised by a direct session retain SQLSTATE 42501. The
  // same failure crossing the runtime API boundary is intentionally rendered
  // as a safe Error message, so preserve the exact permission-denied shape as
  // the repair signal without treating arbitrary 500s as storage drift.
  return /permission denied for (?:database|schema|relation|table|sequence)/i.test(
    message,
  );
}

async function repairRuntimeStorageGrants(
  context: RuntimeApiContext,
  input: {
    playName: string;
  },
): Promise<void> {
  await postRuntimeApi<{ ok: true }>(context, {
    action: 'repair_runtime_storage_grants',
    playName: input.playName,
    runId: context.runId ?? null,
  });
}

async function withRuntimeSheetProvisioningRetry<T>(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    sheetContract: PlaySheetContract;
    timings?: RuntimeSheetTiming[];
  },
  operation: () => Promise<T>,
): Promise<T> {
  const firstAttemptStartedAt = Date.now();
  try {
    const result = await operation();
    input.timings?.push({
      phase: 'operation_attempt',
      ms: Date.now() - firstAttemptStartedAt,
      retried: false,
    });
    return result;
  } catch (error) {
    input.timings?.push({
      phase: 'operation_attempt',
      ms: Date.now() - firstAttemptStartedAt,
      retried: false,
      error: error instanceof Error ? error.message : String(error),
    });
    if (isMissingRelationError(error)) {
      const ensureStartedAt = Date.now();
      await ensureRuntimeSheet(context, input);
      input.timings?.push({
        phase: 'ensure_sheet_after_provisioning_error',
        ms: Date.now() - ensureStartedAt,
        retried: true,
      });
    } else if (isPostgresPermissionDeniedError(error)) {
      const repairStartedAt = Date.now();
      await repairRuntimeStorageGrants(context, {
        playName: input.playName,
      });
      input.timings?.push({
        phase: 'repair_storage_grants_after_permission_error',
        ms: Date.now() - repairStartedAt,
        retried: true,
      });
    } else {
      throw error;
    }
    const retryStartedAt = Date.now();
    const result = await operation();
    input.timings?.push({
      phase: 'operation_retry',
      ms: Date.now() - retryStartedAt,
      retried: true,
    });
    return result;
  }
}

async function withRuntimeSheetQueryClient<T>(
  context: RuntimeApiContext,
  session: RuntimePostgresSession,
  input: {
    playName: string;
    tableNamespace: string;
    sheetContract: PlaySheetContract;
    transactional: boolean;
    statementTimeoutMs?: number;
    timings?: RuntimeSheetTiming[];
  },
  operation: (client: RuntimeQueryClient) => Promise<T>,
): Promise<T> {
  const totalStartedAt = Date.now();
  const result = await withRuntimeSheetProvisioningRetry(
    context,
    input,
    async () => {
      if (
        !input.transactional &&
        input.statementTimeoutMs === undefined &&
        isRuntimeOneShotQueryFactoryRegistered()
      ) {
        const operationStartedAt = Date.now();
        return await withRuntimeOneShotPostgres(session, operation).finally(
          () => {
            input.timings?.push({
              phase: 'one_shot_operation',
              ms: Date.now() - operationStartedAt,
            });
          },
        );
      }

      return await withRuntimePostgres(
        session,
        async (client, transaction) => {
          const managesTransaction =
            (input.transactional || input.statementTimeoutMs !== undefined) &&
            !transaction.active;
          if (managesTransaction) await client.query('BEGIN');
          try {
            if (input.statementTimeoutMs !== undefined) {
              await client.query(
                `SET LOCAL statement_timeout TO ${Math.max(1, Math.floor(input.statementTimeoutMs))}`,
              );
            }
            const result = await operation(client);
            if (managesTransaction) await client.query('COMMIT');
            return result;
          } catch (error) {
            if (managesTransaction) {
              await client.query('ROLLBACK').catch(() => {});
            }
            throw error;
          }
        },
        {
          cachePool: !context.disablePostgresPoolCache,
          maxConnectAttempts:
            input.statementTimeoutMs === undefined ? undefined : 1,
          operationTimeoutMs:
            input.statementTimeoutMs === undefined
              ? undefined
              : RUNTIME_SHEET_COMPLETION_SERVER_OPERATION_TIMEOUT_MS,
          signal: context.abortSignal,
        },
      );
    },
  );
  input.timings?.push({
    phase: 'query_client_total',
    ms: Date.now() - totalStartedAt,
  });
  return result;
}

function quoteIdentifier(value: string): string {
  return `"${value.replace(/"/g, '""')}"`;
}

function quoteLiteral(value: string): string {
  return `'${value.replace(/'/g, "''")}'`;
}

function fqRuntimeTable(
  session: RuntimePostgresSession,
  table: string,
): string {
  return `${quoteIdentifier(session.postgres.schema)}.${quoteIdentifier(table)}`;
}

function sheetTable(session: RuntimePostgresSession): string {
  return fqRuntimeTable(session, session.postgres.sheetTable);
}

function summaryTable(session: RuntimePostgresSession): string {
  return fqRuntimeTable(session, session.postgres.summaryTable);
}

function columnSummaryTable(session: RuntimePostgresSession): string {
  return fqRuntimeTable(session, session.postgres.columnSummaryTable);
}

function runDatasetCatalogTable(session: RuntimePostgresSession): string {
  return fqRuntimeTable(session, RUNTIME_RUN_DATASET_CATALOG_TABLE);
}

async function registerRuntimeDataset(
  session: RuntimePostgresSession,
  input: { playName: string; tableNamespace: string; runId: string },
  signal?: AbortSignal | null,
): Promise<void> {
  const tableNamespace = normalizeTableNamespace(input.tableNamespace);
  const playName = normalizePlayNameForSheet(input.playName);
  const datasetId = createRuntimeDatasetId(playName, tableNamespace);
  await withRuntimePostgres(
    session,
    async (client) => {
      await client.query(
        `INSERT INTO ${runDatasetCatalogTable(session)} (
         run_id, dataset_id, play_name, table_namespace, public_path
       ) VALUES ($1, $2, $3, $4, $5)
       ON CONFLICT (run_id, dataset_id) DO UPDATE SET
         play_name = EXCLUDED.play_name,
         table_namespace = EXCLUDED.table_namespace,
         public_path = EXCLUDED.public_path,
         updated_at = now()`,
        [
          input.runId,
          datasetId,
          playName,
          tableNamespace,
          `datasets.${tableNamespace}`,
        ],
      );
    },
    { signal },
  );
}

function workReceiptTable(session: RuntimePostgresSession): string {
  return fqRuntimeTable(
    session,
    session.postgres.receiptTable ?? RUNTIME_WORK_RECEIPT_POSTGRES_TABLE,
  );
}

function receiptExecutionLockTable(session: RuntimePostgresSession): string {
  return fqRuntimeTable(session, '_deepline_receipt_execution_locks');
}

function postgresUrlCacheKey(value: string): string {
  return createHash('sha256').update(value).digest('hex').slice(0, 24);
}

function workReceiptKeyHex(key: string): string {
  return Array.from(new TextEncoder().encode(key), (byte) =>
    byte.toString(16).padStart(2, '0'),
  ).join('');
}

function newRuntimeWorkReceiptLeaseId(): string {
  return `receipt-lease:${randomUUID()}`;
}

function validateRuntimeWorkReceiptKeyScope(
  session: RuntimePostgresSession,
  input: { key: string },
): void {
  const orgId = session.target.orgId.trim();
  const playName = session.playName.trim();
  const scopedReceiptPrefix = scopedWorkReceiptKeyPrefix({ orgId, playName });
  const durableCtxPrefix = `ctx:${orgId}:`;
  const isScopedReceiptKey = input.key.startsWith(scopedReceiptPrefix);
  const isDurableCtxKey = input.key.startsWith(durableCtxPrefix);
  if (!orgId || !playName || (!isScopedReceiptKey && !isDurableCtxKey)) {
    throw new Error(
      'Runtime work receipt key is outside the scoped session scope.',
    );
  }
}

function mapRuntimeWorkReceiptRow(raw: Record<string, unknown>): WorkReceipt {
  const leaseExpiresAt = raw.lease_expires_at;
  return {
    key: String(raw.k ?? ''),
    status: receiptStatusFromCode(raw.status),
    output: raw.output == null ? null : raw.output,
    error: raw.error == null ? null : String(raw.error),
    errorPayload:
      raw.error_payload != null && typeof raw.error_payload === 'object'
        ? (raw.error_payload as WorkReceipt['errorPayload'])
        : null,
    failureKind: workReceiptFailureKindFromCode(raw.failure_kind),
    runId: raw.run_id == null ? null : String(raw.run_id),
    leaseId: raw.lease_id == null ? null : String(raw.lease_id),
    leaseOwnerRunId:
      raw.lease_owner_run_id == null ? null : String(raw.lease_owner_run_id),
    leaseOwnerAttempt:
      raw.lease_owner_attempt == null ||
      !Number.isFinite(Number(raw.lease_owner_attempt))
        ? null
        : Number(raw.lease_owner_attempt),
    leaseExpiresAt:
      leaseExpiresAt instanceof Date
        ? leaseExpiresAt.toISOString()
        : leaseExpiresAt == null
          ? null
          : String(leaseExpiresAt),
    updatedAt: raw.updated_at == null ? null : String(raw.updated_at),
  };
}

function isMissingRuntimeWorkReceiptSelfHealColumnError(
  error: unknown,
): boolean {
  if (!error || typeof error !== 'object') {
    return false;
  }
  const code = 'code' in error ? String(error.code) : '';
  const message = 'message' in error ? String(error.message) : '';
  if (code !== '42703' && !/column .* does not exist/i.test(message)) {
    return false;
  }
  return RUNTIME_WORK_RECEIPT_SELF_HEAL_COLUMNS.some((column) =>
    message.includes(column),
  );
}

function runtimeWorkReceiptEnsureCacheKey(
  session: RuntimePostgresSession,
): string {
  return `${postgresUrlCacheKey(session.postgresUrl)}::${session.postgres.schema}::${session.postgres.receiptTable ?? RUNTIME_WORK_RECEIPT_POSTGRES_TABLE}`;
}

async function missingRuntimeWorkReceiptSelfHealColumns(
  session: RuntimePostgresSession,
  client: RuntimeQueryClient,
): Promise<string[]> {
  const result = await client.query(
    `
      SELECT column_name
      FROM information_schema.columns
      WHERE table_schema = $1
        AND table_name = $2
        AND column_name = ANY($3::text[])
    `,
    [
      session.postgres.schema,
      session.postgres.receiptTable ?? RUNTIME_WORK_RECEIPT_POSTGRES_TABLE,
      [...RUNTIME_WORK_RECEIPT_SELF_HEAL_COLUMNS],
    ],
  );
  const present = new Set(
    result.rows.map((row) => String(row.column_name ?? '')),
  );
  return RUNTIME_WORK_RECEIPT_SELF_HEAL_COLUMNS.filter(
    (column) => !present.has(column),
  );
}

// COMPAT / SELF-HEAL — owner: runtime. Removal milestone: runtime-ledger
// cutover. The customer-Neon
// work-receipt table (lease columns + failure_kind) is the dying surface. The
// F2 rolling migration (scripts/migrate-play-runtime-storage.ts) backfills these
// columns; this lazy CREATE/ALTER self-heal is a compat-window safety net and is
// deleted together with the Neon receipt path at M1.
async function ensureRuntimeWorkReceiptTable(
  session: RuntimePostgresSession,
  client: RuntimeQueryClient,
): Promise<void> {
  const cacheKey = runtimeWorkReceiptEnsureCacheKey(session);
  const cached = runtimeWorkReceiptEnsureCache.get(cacheKey);
  if (cached) {
    await cached;
    return;
  }
  const promise = client
    .query(
      `
    CREATE TABLE IF NOT EXISTS ${workReceiptTable(session)} (
      k bytea PRIMARY KEY,
      status smallint NOT NULL DEFAULT 0,
      output jsonb,
      error text,
      error_payload jsonb,
      failure_kind smallint NOT NULL DEFAULT 0,
      run_id text,
      lease_id text,
      lease_owner_run_id text,
      lease_owner_attempt integer,
      lease_expires_at timestamptz,
      updated_at timestamptz NOT NULL DEFAULT now()
    )
  `,
    )
    .then(async () => {
      await client.query(`
        CREATE TABLE IF NOT EXISTS ${receiptExecutionLockTable(session)} (
          k bytea PRIMARY KEY,
          owner_execution_id text NOT NULL,
          expires_at timestamptz NOT NULL,
          updated_at timestamptz NOT NULL DEFAULT now()
        )
      `);
      const missingColumns = await missingRuntimeWorkReceiptSelfHealColumns(
        session,
        client,
      );
      if (missingColumns.length === 0) return;
      await client.query(`
        ALTER TABLE ${workReceiptTable(session)}
          ${missingColumns
            .map((column) => {
              const type =
                column === 'lease_expires_at'
                  ? 'timestamptz'
                  : column === 'lease_owner_attempt'
                    ? 'integer'
                    : column === RUNTIME_WORK_RECEIPT_FAILURE_KIND_COLUMN
                      ? 'smallint NOT NULL DEFAULT 0'
                      : column === RUNTIME_WORK_RECEIPT_ERROR_PAYLOAD_COLUMN
                        ? 'jsonb'
                        : 'text';
              return `ADD COLUMN IF NOT EXISTS ${column} ${type}`;
            })
            .join(',\n          ')}
      `);
    })
    .then(() => undefined);
  runtimeWorkReceiptEnsureCache.set(cacheKey, promise);
  try {
    await promise;
  } catch (error) {
    if (runtimeWorkReceiptEnsureCache.get(cacheKey) === promise) {
      runtimeWorkReceiptEnsureCache.delete(cacheKey);
    }
    throw error;
  }
}

export async function acquireRuntimeReceiptExecutionLock(
  context: RuntimeApiContext,
  input: {
    playName: string;
    key: string;
    ownerExecutionId: string;
    ttlMs: number;
  },
): Promise<{ ownerExecutionId: string; expiresAt: string } | null> {
  const ttlMs = Math.floor(input.ttlMs);
  if (!Number.isFinite(ttlMs) || ttlMs <= 0 || ttlMs > 10 * 60_000) {
    throw new Error('Runtime receipt execution lock TTL must be 1..600000ms.');
  }
  const ownerExecutionId = input.ownerExecutionId.trim();
  if (!ownerExecutionId) {
    throw new Error('Runtime receipt execution lock owner is required.');
  }
  const session = await getRuntimeWorkReceiptSession(context, input);
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
      INSERT INTO ${receiptExecutionLockTable(session)} AS target (
        k, owner_execution_id, expires_at, updated_at
      ) VALUES (
        decode($1, 'hex'), $2, now() + ($3::double precision * interval '1 millisecond'), now()
      )
      ON CONFLICT (k) DO UPDATE
        SET owner_execution_id = EXCLUDED.owner_execution_id,
            expires_at = EXCLUDED.expires_at,
            updated_at = now()
      WHERE target.expires_at <= now()
         OR target.owner_execution_id = EXCLUDED.owner_execution_id
      RETURNING owner_execution_id, expires_at
      `,
        [workReceiptKeyHex(input.key), ownerExecutionId, ttlMs],
      );
      const row = rows[0];
      return row
        ? {
            ownerExecutionId: String(row.owner_execution_id),
            expiresAt: new Date(String(row.expires_at)).toISOString(),
          }
        : null;
    },
  );
}

export async function releaseRuntimeReceiptExecutionLock(
  context: RuntimeApiContext,
  input: { playName: string; key: string; ownerExecutionId: string },
): Promise<boolean> {
  const session = await getRuntimeWorkReceiptSession(context, input);
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query(
        `DELETE FROM ${receiptExecutionLockTable(session)}
        WHERE k = decode($1, 'hex')
          AND owner_execution_id = $2
        RETURNING 1`,
        [workReceiptKeyHex(input.key), input.ownerExecutionId],
      );
      return rows.length > 0;
    },
  );
}

async function withRuntimeWorkReceiptClient<T>(
  context: RuntimeApiContext,
  session: RuntimePostgresSession,
  operation: (client: RuntimeQueryClient) => Promise<T>,
): Promise<T> {
  // Receipt tables are part of customer Postgres bootstrap, so the hot path
  // should not pay DDL on every fresh runtime isolate. Try the receipt query
  // first, then self-heal once on a missing relation/column. Permission loss
  // during an operational receipt query must fail loud: silently repairing it
  // can let provider work continue after durable receipt persistence is gone.
  // Only the schema-migration branch below may repair ownership before its
  // retry. A genuinely missing schema or second failure also rethrows loudly.
  const withClient = async <R>(
    run: (client: RuntimeQueryClient) => Promise<R>,
  ): Promise<R> =>
    isRuntimeOneShotQueryFactoryRegistered()
      ? await withRuntimeOneShotPostgres(session, run)
      : await withRuntimePostgres(session, run, {
          cachePool: !context.disablePostgresPoolCache,
          signal: context.abortSignal,
        });
  let selfHealAttempted = false;

  for (
    let attempt = 1;
    attempt <= RUNTIME_WORK_RECEIPT_QUERY_MAX_ATTEMPTS;
    attempt += 1
  ) {
    try {
      return await withClient(operation);
    } catch (error) {
      const missingStorage =
        isMissingRelationError(error) ||
        isMissingRuntimeWorkReceiptSelfHealColumnError(error);
      if (missingStorage && !selfHealAttempted) {
        selfHealAttempted = true;
        runtimeWorkReceiptEnsureCache.delete(
          runtimeWorkReceiptEnsureCacheKey(session),
        );
        try {
          await withClient((client) =>
            ensureRuntimeWorkReceiptTable(session, client),
          );
        } catch (ensureError) {
          if (!isPostgresPermissionDeniedError(ensureError)) {
            throw ensureError;
          }
          runtimeWorkReceiptEnsureCache.delete(
            runtimeWorkReceiptEnsureCacheKey(session),
          );
          await repairRuntimeStorageGrants(context, {
            playName: context.playName?.trim() || session.playName,
          });
          await withClient((client) =>
            ensureRuntimeWorkReceiptTable(session, client),
          );
        }
        continue;
      }
      if (
        attempt >= RUNTIME_WORK_RECEIPT_QUERY_MAX_ATTEMPTS ||
        !isTransientRuntimePostgresOperationError(error)
      ) {
        throw error;
      }
      if (
        !isRuntimeOneShotQueryFactoryRegistered() &&
        !context.disablePostgresPoolCache
      ) {
        await resetRuntimePostgresPool(session.postgresUrl);
      }
      await sleep(
        RUNTIME_WORK_RECEIPT_QUERY_RETRY_DELAYS_MS[attempt - 1] ??
          RUNTIME_WORK_RECEIPT_QUERY_RETRY_DELAYS_MS[
            RUNTIME_WORK_RECEIPT_QUERY_RETRY_DELAYS_MS.length - 1
          ],
      );
    }
  }
  throw new Error('Runtime work receipt query failed after retries.');
}

const PLAY_INTERNAL_SHEET_VERSION_SEQUENCE = '_deepline_sheet_version_seq';

function nextRuntimeSheetVersionExpression(
  session: RuntimePostgresSession,
): string {
  return `nextval(${quoteLiteral(`${session.postgres.schema}.${PLAY_INTERNAL_SHEET_VERSION_SEQUENCE}`)}::regclass)`;
}

function missingOutputCellSql(
  tableAlias: string,
  outputPhysicalColumns: readonly PhysicalSheetColumnProjection[],
): string {
  if (outputPhysicalColumns.length === 0) {
    return 'false';
  }
  return outputPhysicalColumns
    .map((column) => {
      const quoted = `${tableAlias}.${quoteIdentifier(column.sqlName)}`;
      const staleAt = `${tableAlias}._cell_meta -> ${quoteLiteral(column.fieldName)} -> 'staleAt'`;
      const staleAtMs = `(CASE WHEN jsonb_typeof(${staleAt}) = 'number' THEN (${staleAt})::text::double precision ELSE NULL END)`;
      return `(${quoted} IS NULL OR ${quoted} = 'null'::jsonb OR ${quoted} = '""'::jsonb OR (${staleAtMs} IS NOT NULL AND ${staleAtMs} <= extract(epoch from now()) * 1000))`;
    })
    .join(' OR ');
}

function changedPatchedCellSql(
  tableAlias: string,
  patchAlias: string,
  projections: readonly PhysicalSheetColumnProjection[],
): string {
  if (projections.length === 0) {
    return 'false';
  }
  return projections
    .map((column) => {
      const quoted = `${tableAlias}.${quoteIdentifier(column.sqlName)}`;
      return `(${patchAlias} ? ${quoteLiteral(column.fieldName)} AND ${quoted} IS DISTINCT FROM ${patchAlias} -> ${quoteLiteral(column.fieldName)})`;
    })
    .join(' OR ');
}

function isSystemSheetColumn(columnName: string): boolean {
  switch (columnName) {
    case '_key':
    case '_status':
    case '_run_id':
    case '_error':
    case '_stage':
    case '_provider':
    case '_input_index':
    case '_attempt_id':
    case '_attempt_owner_run_id':
    case '_attempt_expires_at':
    case '_attempt_seq':
    case '_created_at':
    case '_updated_at':
    case '_version':
    case '_cell_meta':
    case 'seq':
    case '__has_enriched':
    case '__has_failed':
    case '__deeplineCsvProjectedFields':
    case '__deeplineCsvProjectedValues':
    case '__deeplineSourceRowIndex':
    case '__deeplineOriginalSourceRowIndex':
      return true;
    default:
      return false;
  }
}

function normalizeRuntimeSheetAttemptId(
  value: string | null | undefined,
): string {
  const trimmed = value?.trim();
  return trimmed || `sheet-attempt:${randomUUID()}`;
}

function normalizeRuntimeSheetAttemptOwnerRunId(input: {
  attemptOwnerRunId?: string | null;
  runId: string;
}): string {
  return input.attemptOwnerRunId?.trim() || input.runId;
}

function normalizeRuntimeRunAttempt(value: number | null | undefined): number {
  if (typeof value !== 'number' || !Number.isFinite(value)) return 0;
  return Math.max(0, Math.floor(value));
}

function normalizeRuntimeLeaseTtlMs(
  value: number | null | undefined,
  fallback: number,
): number {
  if (
    typeof value !== 'number' ||
    !Number.isFinite(value) ||
    !Number.isInteger(value) ||
    value <= 0
  ) {
    return fallback;
  }
  return value;
}

async function mintRuntimeSheetAttemptExpiresAt(
  client: RuntimeQueryClient,
  leaseTtlMs?: number | null,
): Promise<string> {
  const ttlMs = normalizeRuntimeLeaseTtlMs(
    leaseTtlMs,
    PLAY_RUNTIME_SHEET_ATTEMPT_LEASE_TTL_MS,
  );
  const { rows } = await client.query<{ attempt_expires_at: Date | string }>(
    `
      SELECT now() + ($1::double precision * interval '1 millisecond') AS attempt_expires_at
    `,
    [ttlMs],
  );
  const value = rows[0]?.attempt_expires_at;
  if (value instanceof Date) {
    return value.toISOString();
  }
  if (typeof value === 'string' && value.trim()) {
    return new Date(value).toISOString();
  }
  throw new Error('Runtime sheet attempt expiry mint did not return a value.');
}

async function mintRuntimeSheetWriteVersion(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  requested?: number | null,
): Promise<number> {
  if (requested != null) {
    if (!Number.isSafeInteger(requested) || requested <= 0) {
      throw new Error(
        'Runtime sheet writeVersion must be a positive safe integer.',
      );
    }
    return requested;
  }
  const { rows } = await client.query<{ write_version: number | string }>(
    `SELECT ${nextRuntimeSheetVersionExpression(session)} AS write_version`,
  );
  const value = Number(rows[0]?.write_version);
  if (!Number.isSafeInteger(value) || value <= 0) {
    throw new Error(
      'Runtime sheet write version allocation did not return a safe integer.',
    );
  }
  return value;
}

function parseRuntimeCellMeta(value: unknown): Record<string, unknown> {
  if (value && typeof value === 'object' && !Array.isArray(value)) {
    return value as Record<string, unknown>;
  }
  if (typeof value !== 'string' || !value.trim()) {
    return {};
  }
  try {
    const parsed = JSON.parse(value) as unknown;
    return parsed && typeof parsed === 'object' && !Array.isArray(parsed)
      ? (parsed as Record<string, unknown>)
      : {};
  } catch {
    return {};
  }
}

function mapRuntimePostgresRow(input: {
  raw: Record<string, unknown>;
  sheetContract?: PlaySheetContract | null;
}): RuntimeApiRowRecord {
  const { raw, sheetContract } = input;
  const cellMeta = parseRuntimeCellMeta(raw._cell_meta);
  const projections = physicalSheetColumnProjections(sheetContract);
  const publicData =
    projections.length > 0
      ? Object.fromEntries(
          projections
            .filter((column) =>
              Object.prototype.hasOwnProperty.call(raw, column.sqlName),
            )
            .map((column) => [column.fieldName, raw[column.sqlName]]),
        )
      : Object.fromEntries(
          Object.entries(raw).filter(([key]) => !isSystemSheetColumn(key)),
        );
  const data =
    Object.keys(cellMeta).length > 0
      ? {
          ...publicData,
          [DEEPLINE_CELL_META_FIELD]: cellMeta,
        }
      : publicData;
  return {
    key: String(raw._key ?? ''),
    data,
    inputIndex: raw._input_index != null ? Number(raw._input_index) : undefined,
  };
}

function completedRuntimeCellMetaPatch(input: {
  runId: string;
  outputFields: readonly string[];
  rowPatch?: Record<string, unknown>;
}): Record<string, unknown> {
  const patch: Record<string, unknown> = {};
  const completedAt = Date.now();
  for (const field of input.outputFields) {
    const existing =
      input.rowPatch?.[field] &&
      typeof input.rowPatch[field] === 'object' &&
      !Array.isArray(input.rowPatch[field])
        ? (input.rowPatch[field] as Record<string, unknown>)
        : {};
    patch[field] = {
      status: 'completed',
      runId: input.runId,
      completedAt,
      ...existing,
    };
  }
  for (const [field, meta] of Object.entries(input.rowPatch ?? {})) {
    if (!Object.hasOwn(patch, field)) {
      patch[field] = meta;
    }
  }
  return patch;
}

function mergeRuntimeCellMetaPatchSql(
  targetExpression: string,
  patchExpression: string,
): string {
  return `coalesce(${targetExpression}, '{}'::jsonb) || (
    SELECT coalesce(
      jsonb_object_agg(patch.key, coalesce(${targetExpression} -> patch.key, '{}'::jsonb) || patch.value),
      '{}'::jsonb
    )
    FROM jsonb_each(coalesce(${patchExpression}, '{}'::jsonb)) AS patch(key, value)
  )`;
}

async function readRuntimeRowsByKey(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  keys: readonly string[],
  sheetContract?: PlaySheetContract | null,
): Promise<RuntimeApiRowRecord[]> {
  if (keys.length === 0) {
    return [];
  }
  const { rows } = await client.query<Record<string, unknown>>(
    `
      SELECT *
      FROM ${sheetTable(session)}
      WHERE _key = ANY($1::text[])
      ORDER BY _input_index ASC NULLS LAST, _created_at ASC, _key ASC
    `,
    [keys],
  );
  return rows.map((raw) => mapRuntimePostgresRow({ raw, sheetContract }));
}

function mergeRuntimeCompletedRow(input: {
  inputRow: Record<string, unknown>;
  completedData: Record<string, unknown>;
  sheetContract: PlaySheetContract;
}): Record<string, unknown> {
  const syntheticNullInputColumns = new Set(
    input.sheetContract.columns.flatMap((column) => {
      const field = column.field;
      if (
        column.source !== 'input' ||
        typeof field !== 'string' ||
        field in input.inputRow ||
        input.completedData[field] != null
      ) {
        return [];
      }
      return [field];
    }),
  );
  const cleanedCompletedData = Object.fromEntries(
    Object.entries(input.completedData).filter(
      ([key]) => !syntheticNullInputColumns.has(key),
    ),
  );
  return {
    ...input.inputRow,
    ...cleanedCompletedData,
  };
}

function buildAppendedRowKey(input: {
  row: Record<string, unknown>;
  tableNamespace: string;
  idempotencyKey: string;
  ordinal: number;
}): string {
  const baseKey = derivePlayRowIdentity(input.row, input.tableNamespace);
  const suffix = createHash('sha1')
    .update(`${input.idempotencyKey}:${input.ordinal}`)
    .digest('hex')
    .slice(0, APPEND_KEY_SUFFIX_LENGTH);
  return `${baseKey}:append:${suffix}`;
}

function chunkValues<T>(values: readonly T[], chunkSize: number): T[][] {
  const chunks: T[][] = [];
  for (let index = 0; index < values.length; index += chunkSize) {
    chunks.push(values.slice(index, index + chunkSize));
  }
  return chunks;
}

async function readRuntimeRows(
  session: RuntimePostgresSession,
  input: {
    limit: number;
    offset: number;
    runId?: string | null;
    rowMode?: 'output' | 'all';
    sheetContract?: PlaySheetContract | null;
  },
): Promise<RuntimeApiRowRecord[]> {
  return await withRuntimePostgres(session, async (client) => {
    if (input.runId) {
      if (input.rowMode === 'all') {
        const { rows } = await client.query(
          `SELECT *
           FROM ${sheetTable(session)}
           WHERE _run_id = $1::text
             AND _status IN ('enriched', 'failed')
           ORDER BY _input_index ASC NULLS LAST, _created_at ASC, _key ASC
           LIMIT $2 OFFSET $3`,
          [input.runId, input.limit, input.offset],
        );
        return rows.map((raw) =>
          mapRuntimePostgresRow({ raw, sheetContract: input.sheetContract }),
        );
      }
      const { rows } = await client.query(
        `WITH scoped AS (
           SELECT *,
                  bool_or(_status = 'enriched') OVER () AS __has_enriched,
                  bool_or(_status = 'failed') OVER () AS __has_failed
           FROM ${sheetTable(session)}
           WHERE _run_id = $1::text
         )
         SELECT *
         FROM scoped
         WHERE (__has_enriched AND _status = 'enriched')
            OR (NOT __has_enriched AND __has_failed AND _status = 'failed')
            OR (NOT __has_enriched AND NOT __has_failed)
         ORDER BY _input_index ASC NULLS LAST, _created_at ASC, _key ASC
         LIMIT $2 OFFSET $3`,
        [input.runId, input.limit, input.offset],
      );
      return rows.map((raw) =>
        mapRuntimePostgresRow({ raw, sheetContract: input.sheetContract }),
      );
    }
    const { rows } = await client.query(
      `SELECT *
       FROM ${sheetTable(session)}
       ORDER BY _input_index ASC NULLS LAST, _created_at ASC, _key ASC
       LIMIT $1 OFFSET $2`,
      [input.limit, input.offset],
    );
    return rows.map((raw) =>
      mapRuntimePostgresRow({ raw, sheetContract: input.sheetContract }),
    );
  });
}

async function readRuntimeSummary(
  session: RuntimePostgresSession,
): Promise<{ stats: { total: number } }> {
  const normalizedPlayName = normalizePlayNameForSheet(session.playName);
  const normalizedTableNamespace = normalizeTableNamespace(
    session.target.tableNamespace,
  );
  return await withRuntimePostgres(session, async (client) => {
    const { rows } = await client.query(
      `SELECT total, queued, running, completed, failed
       FROM ${summaryTable(session)}
       WHERE play_name = $1 AND table_namespace = $2
       LIMIT 1`,
      [normalizedPlayName, normalizedTableNamespace],
    );
    const row = rows[0];
    const total = Number(row?.total ?? 0);
    const partitionTotal =
      Number(row?.queued ?? 0) +
      Number(row?.running ?? 0) +
      Number(row?.completed ?? 0) +
      Number(row?.failed ?? 0);
    return { stats: { total: Math.max(total, partitionTotal) } };
  });
}

async function writeRuntimeRows(
  session: RuntimePostgresSession,
  input: {
    tableNamespace: string;
    rows: Record<string, unknown>[];
    runId: string;
    idempotencyKey: string;
    sheetContract: PlaySheetContract;
    mode: 'append' | 'upsert' | 'replace';
  },
): Promise<RowsWriteResponse> {
  const physicalColumnProjections = physicalSheetColumnProjections(
    input.sheetContract,
  );
  const physicalColumns = physicalColumnProjections.map(
    (column) => column.sqlName,
  );
  const physicalInsertColumnsSql =
    physicalColumns.length > 0
      ? `, ${physicalColumns.map(quoteIdentifier).join(', ')}`
      : '';
  const physicalInsertValuesSql =
    physicalColumns.length > 0
      ? `, ${physicalColumnProjections
          .map((column) => `payload -> ${quoteLiteral(column.fieldName)}`)
          .join(', ')}`
      : '';

  const rowsToWrite = input.rows.filter(
    (row) => row && typeof row === 'object' && !Array.isArray(row),
  );
  if (rowsToWrite.length === 0) {
    return { disposition: 'completed', writtenRows: 0 };
  }
  const normalizedPlayName = normalizePlayNameForSheet(session.playName);
  const normalizedTableNamespace = normalizeTableNamespace(
    input.tableNamespace,
  );

  // Build write entries first so we know chunk count up front. Single-chunk
  // writes skip the BEGIN/COMMIT pair entirely (the mega-CTE is atomic at
  // statement level), and the per-chunk mega-CTE folds: starting-index lookup,
  // sheet insert, and summary upsert into one round-trip.
  const rowEntries =
    input.mode === 'upsert'
      ? Array.from(
          rowsToWrite
            .reduce((uniqueRows, row) => {
              const key = derivePlayRowIdentity(row, input.tableNamespace);
              if (key && !uniqueRows.has(key)) {
                uniqueRows.set(key, row);
              }
              return uniqueRows;
            }, new Map<string, Record<string, unknown>>())
            .entries(),
        ).map(([key, row], inputIndex) => ({ key, row, inputIndex }))
      : rowsToWrite.map((row, index) => ({
          key: buildAppendedRowKey({
            row,
            tableNamespace: input.tableNamespace,
            idempotencyKey: input.idempotencyKey,
            ordinal: index,
          }),
          row,
          // For append/replace modes the starting offset is computed in SQL
          // as `coalesce(max(_input_index), -1) + 1 + (ord - 1)`; we only
          // pass the per-chunk ordinal here so the SQL can add it to the
          // dynamic starting index without an extra round-trip.
          inputIndex: index,
        }));

  const chunks = chunkValues(rowEntries, DIRECT_POSTGRES_BATCH_SIZE);
  const needsTransaction = input.mode === 'replace' || chunks.length > 1;

  return await withRuntimePostgres(session, async (client, transaction) => {
    const managesTransaction = needsTransaction && !transaction.active;
    if (managesTransaction) await client.query('BEGIN');
    try {
      if (input.mode === 'replace') {
        // Collapse 4 cleanup statements into one CTE-shaped query. Postgres
        // executes data-modifying CTEs against the same snapshot, but each
        // operates on a distinct table so ordering does not matter.
        await client.query(
          `WITH cleared_sheet AS (
             DELETE FROM ${sheetTable(session)} RETURNING 1
           ),
           reset_summary AS (
             UPDATE ${summaryTable(session)}
                SET total = 0, queued = 0, running = 0, completed = 0, failed = 0, _updated_at = now()
              WHERE play_name = $1 AND table_namespace = $2
              RETURNING 1
           ),
           cleared_col_summary AS (
             DELETE FROM ${columnSummaryTable(session)}
             WHERE play_name = $1 AND table_namespace = $2
             RETURNING 1
           )
           SELECT 1`,
          [normalizedPlayName, normalizedTableNamespace],
        );
      }

      // For append/replace: SQL computes _input_index from the live max so
      // we never need a separate SELECT round-trip. For upsert: the JS
      // ordinals (0..N) are passed straight through, matching prior shape.
      const computesStartingIndexInSql =
        input.mode === 'append' || input.mode === 'replace';

      let writtenRows = 0;
      for (const chunk of chunks) {
        const chunkKeys = chunk.map((entry) => entry.key);
        const chunkPayloads = chunk.map((entry) =>
          stringifyPostgresJson(entry.row),
        );
        const chunkInputIndexes = chunk.map((entry) => entry.inputIndex);

        const startingIndexCte = computesStartingIndexInSql
          ? `starting_index AS (
              SELECT coalesce(max(_input_index), -1)::bigint AS v FROM ${sheetTable(session)}
            ),`
          : '';
        const inputIndexExpr = computesStartingIndexInSql
          ? `(SELECT v FROM starting_index) + index_values._input_index::bigint + 1`
          : `index_values._input_index::bigint`;
        const insertedRowsCte =
          input.mode === 'upsert'
            ? `inserted_rows AS (
                INSERT INTO ${sheetTable(session)} (_key, _status, _run_id, _input_index${physicalInsertColumnsSql})
                SELECT _key, 'pending', $4::text, _input_index${physicalInsertValuesSql}
                FROM input_rows
                ON CONFLICT (_key) DO UPDATE SET
                  _status = 'pending',
                  _run_id = EXCLUDED._run_id,
                  _input_index = EXCLUDED._input_index,
                  _updated_at = now(),
                  _version = ${nextRuntimeSheetVersionExpression(session)}
                WHERE ${sheetTable(session)}._status = 'stale'
                RETURNING _key
              ),`
            : `inserted_rows AS (
                INSERT INTO ${sheetTable(session)} (_key, _status, _run_id, _input_index${physicalInsertColumnsSql})
                SELECT _key, 'pending', $4::text, _input_index${physicalInsertValuesSql}
                FROM input_rows
                RETURNING _key
              ),`;

        const sql = `
          WITH ${startingIndexCte}
          input_rows AS (
            SELECT DISTINCT ON (key_values._key)
                   key_values._key, payload_values.payload,
                   ${inputIndexExpr} AS _input_index
            FROM unnest($1::text[]) WITH ORDINALITY AS key_values(_key, ord)
            JOIN unnest($2::jsonb[]) WITH ORDINALITY AS payload_values(payload, ord)
              ON payload_values.ord = key_values.ord
            JOIN unnest($3::bigint[]) WITH ORDINALITY AS index_values(_input_index, ord)
              ON index_values.ord = key_values.ord
            ORDER BY key_values._key, key_values.ord
          ),
          ${insertedRowsCte}
          inserted_count_cte AS (
            SELECT count(*)::bigint AS c FROM inserted_rows
          ),
          summary_upsert AS (
            INSERT INTO ${summaryTable(session)} (play_name, table_namespace, total, queued, running, completed, failed)
            SELECT $5::text, $6::text, c, c, 0, 0, 0
              FROM inserted_count_cte
             WHERE c > 0
            ON CONFLICT (play_name, table_namespace) DO UPDATE SET
              queued = ${summaryTable(session)}.queued + EXCLUDED.queued,
              total = ${runtimeSummaryTotalSql({
                currentTotal: `${summaryTable(session)}.total`,
                totalDelta: 'EXCLUDED.total',
                queued: `${summaryTable(session)}.queued + EXCLUDED.queued`,
                running: `${summaryTable(session)}.running`,
                completed: `${summaryTable(session)}.completed`,
                failed: `${summaryTable(session)}.failed`,
              })},
              _updated_at = now()
            RETURNING 1
          )
          SELECT c::int AS inserted_count FROM inserted_count_cte
        `;

        const { rows } = await client.query(sql, [
          chunkKeys,
          chunkPayloads,
          chunkInputIndexes,
          input.runId,
          normalizedPlayName,
          normalizedTableNamespace,
        ]);
        writtenRows += Number(rows[0]?.inserted_count ?? 0);
      }

      if (managesTransaction) await client.query('COMMIT');
      return { disposition: 'completed', writtenRows };
    } catch (error) {
      if (managesTransaction) {
        await client.query('ROLLBACK').catch(() => {});
      }
      throw error;
    }
  });
}

export async function resolveRuntimeReferencedPlay(
  context: RuntimeApiContext,
  playRef: string,
): Promise<ResolvedRuntimePlay | null> {
  const response = await postRuntimeApi<{
    play: ResolvedRuntimePlay | null;
    manifest?: { staticPipeline?: PlayStaticPipeline | null } | null;
  }>(context, {
    action: 'resolve_play',
    playRef,
  });
  if (!response.play) {
    return null;
  }

  // The server builds the manifest and child artifact together. Its manifest
  // carries the canonical empty scalar contract when a published scalar's
  // revision snapshot has no pipeline. Keep that contract on the object the
  // runner passes to ctx.runPlay; otherwise a valid scalar leaf arrives as an
  // indistinguishable "missing static contract" at runtime.
  return {
    ...response.play,
    staticPipeline:
      response.play.staticPipeline ?? response.manifest?.staticPipeline ?? null,
  };
}

export async function ensureRuntimeSheet(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    sheetContract: PlaySheetContract;
  },
): Promise<void> {
  const request = {
    action: 'ensure_sheet' as const,
    ...input,
    runId: context.runId ?? null,
    userEmail: normalizeRuntimeUserEmail(context.userEmail),
  };
  try {
    await postRuntimeApi<{ ok: true }>(context, request);
  } catch (error) {
    if (!isPostgresPermissionDeniedError(error)) {
      throw error;
    }
    // Fresh tenant databases can become visible through the pooled endpoint
    // before their runtime-role grants are usable. Non-preloaded file runs hit
    // ensure_sheet before any direct Postgres operation, so the existing
    // operation-level self-heal cannot observe this boundary. Repair the
    // canonical storage contract once, then retry the exact ensure request.
    await repairRuntimeStorageGrants(context, {
      playName: input.playName,
    });
    await postRuntimeApi<{ ok: true }>(context, request);
  }
}

async function prepareRuntimeSheetDatasetRows(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    chunks: RuntimeDatasetRowEntry[][];
    runId: string;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
    attemptId: string;
    attemptOwnerRunId: string;
    attemptExpiresAt: string;
    attemptSeq: number;
    writeVersion: number;
    physicalInsertColumnsSql: string;
    physicalInsertValuesSql: string;
    physicalRefreshSetSql: string;
    physicalUpsertSetSql: string;
    outputPhysicalColumns: PhysicalSheetColumnProjection[];
    force?: boolean;
    mode?: RuntimeSheetDatasetMode;
  },
): Promise<{
  inserted: number;
  rowDispositions: RuntimeSheetPreparedRowDisposition[];
}> {
  let inserted = 0;
  const rowDispositions: RuntimeSheetPreparedRowDisposition[] = [];
  for (const chunk of input.chunks) {
    const chunkKeys = chunk.map((entry) => entry.key);
    const chunkPayloads = chunk.map((entry) =>
      stringifyPostgresJson(entry.row),
    );
    const chunkInputIndexes = chunk.map((entry) => entry.inputIndex);
    const existingMissingOutputSql = missingOutputCellSql(
      'existing',
      input.outputPhysicalColumns,
    );
    const targetMissingOutputSql = missingOutputCellSql(
      'target',
      input.outputPhysicalColumns,
    );
    const targetAttemptFenceSql = activeRuntimeSheetAttemptFenceSql(
      'target',
      '$7::text',
      '$8::text',
      '$9::timestamptz',
      '$10::integer',
    );
    const targetOlderForeignAttemptOwnerSql =
      'coalesce(target._attempt_owner_run_id, target._run_id) IS DISTINCT FROM $8::text AND COALESCE(target._attempt_seq, 0) < $10::integer';
    const existingAttemptFenceSql = activeRuntimeSheetAttemptFenceSql(
      'existing',
      '$7::text',
      '$8::text',
      '$9::timestamptz',
      '$10::integer',
    );
    const existingForeignAttemptOwnerSql =
      'coalesce(existing._attempt_owner_run_id, existing._run_id) IS DISTINCT FROM $8::text';
    const targetNewerTerminalRowSql = newerTerminalRuntimeSheetRowSql(
      'target',
      '$9::timestamptz',
      '$10::integer',
      '$7::text',
      '$8::text',
    );
    const existingNewerTerminalRowSql = newerTerminalRuntimeSheetRowSql(
      'existing',
      '$9::timestamptz',
      '$10::integer',
      '$7::text',
      '$8::text',
    );
    const existingWritableEnrichedAttemptSql =
      writableEnrichedRuntimeSheetAttemptSql(
        'existing',
        '$8::text',
        '$9::timestamptz',
        '$10',
      );
    const { rows } = await client.query(
      `
       WITH input_rows AS (
          SELECT DISTINCT ON (key_values._key)
                 key_values._key, payload_values.payload, index_values._input_index
          FROM unnest($1::text[]) WITH ORDINALITY AS key_values(_key, ord)
          JOIN unnest($2::jsonb[]) WITH ORDINALITY AS payload_values(payload, ord)
            ON payload_values.ord = key_values.ord
          JOIN unnest($3::bigint[]) WITH ORDINALITY AS index_values(_input_index, ord)
            ON index_values.ord = key_values.ord
          ORDER BY key_values._key, key_values.ord
       ),
       preexisting_rows AS (
          SELECT target._key
          FROM ${sheetTable(session)} AS target
          JOIN input_rows ON input_rows._key = target._key
       ),
       versioned_completed_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _run_id = $4::text,
              _writer_run_id = $4::text,
              _write_version = $12::bigint,
              _input_index = input_rows._input_index,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}${input.physicalRefreshSetSql}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._status = 'enriched'
            AND COALESCE(target._write_version, 0) < $12::bigint
          RETURNING target._key,
                    ($11::boolean OR (${targetMissingOutputSql})) AS needs_recompute
       ),
       versioned_failed_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _run_id = $4::text,
              _writer_run_id = $4::text,
              _write_version = $12::bigint,
              _input_index = input_rows._input_index,
              _attempt_id = $7::text,
              _attempt_owner_run_id = $8::text,
              _attempt_expires_at = NULL,
              _attempt_seq = $10::integer,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}${input.physicalRefreshSetSql}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._status = 'failed'
            AND COALESCE(target._write_version, 0) < $12::bigint
          RETURNING target._key
       ),
       existing_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _input_index = input_rows._input_index,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}${input.physicalRefreshSetSql}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._write_version = $12::bigint
            AND target._status <> 'stale'
            AND (
              NOT (${targetNewerTerminalRowSql})
              OR (
                $11::boolean
                AND target._status = 'enriched'
                AND COALESCE(target._attempt_seq, 0) <= $10::integer
              )
            )
            AND (
              ${targetAttemptFenceSql}
              OR (
                $11::boolean
                AND target._status = 'enriched'
              )
            )
            AND (
              $11::boolean
              OR target._input_index IS DISTINCT FROM input_rows._input_index
            )
          RETURNING target._key
        ),
        inserted_rows AS (
          INSERT INTO ${sheetTable(session)} (_key, _status, _run_id, _input_index, _attempt_id, _attempt_owner_run_id, _attempt_expires_at, _attempt_seq, _write_version, _writer_run_id${input.physicalInsertColumnsSql})
          SELECT _key, 'pending', $4::text, _input_index, $7::text, $8::text, $9::timestamptz, $10::integer, $12::bigint, $4::text${input.physicalInsertValuesSql}
          FROM input_rows
          ON CONFLICT (_key) DO UPDATE SET
            _status = 'pending',
            _run_id = EXCLUDED._run_id,
            _input_index = EXCLUDED._input_index,
            _attempt_id = EXCLUDED._attempt_id,
            _attempt_owner_run_id = EXCLUDED._attempt_owner_run_id,
            _attempt_expires_at = EXCLUDED._attempt_expires_at,
            _attempt_seq = EXCLUDED._attempt_seq,
            _write_version = EXCLUDED._write_version,
            _writer_run_id = EXCLUDED._writer_run_id,
            _updated_at = now(),
            _version = ${nextRuntimeSheetVersionExpression(session)}${input.physicalUpsertSetSql}
          WHERE ${sheetTable(session)}._status = 'stale'
          RETURNING _key
        ),
        missing_output_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _status = 'pending',
              _run_id = $4::text,
              _input_index = input_rows._input_index,
              _attempt_id = $7::text,
              _attempt_owner_run_id = $8::text,
              _attempt_expires_at = $9::timestamptz,
              _attempt_seq = $10::integer,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._status = 'enriched'
            AND target._write_version = $12::bigint
            AND (${targetMissingOutputSql})
            AND COALESCE(target._attempt_seq, 0) <= $10::integer
          RETURNING target._key
        ),
        superseded_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _run_id = $4::text,
              _input_index = input_rows._input_index,
              _attempt_id = $7::text,
              _attempt_owner_run_id = $8::text,
              _attempt_expires_at = NULL,
              _attempt_seq = $10::integer,
              _write_version = $12::bigint,
              _writer_run_id = $4::text,
              _error = NULL,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}${input.physicalRefreshSetSql}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._status IN ('pending', 'running')
            AND COALESCE(target._write_version, 0) < $12::bigint
          RETURNING target._key
        ),
        claimed_existing_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _run_id = $4::text,
              _input_index = input_rows._input_index,
              _attempt_id = $7::text,
              _attempt_owner_run_id = $8::text,
              _attempt_expires_at = $9::timestamptz,
              _attempt_seq = $10::integer,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._status IN ('pending', 'running', 'failed')
            AND target._write_version = $12::bigint
            AND NOT (${targetNewerTerminalRowSql})
            AND (
              ${targetAttemptFenceSql}
              OR ${targetOlderForeignAttemptOwnerSql}
              -- Failed rows are repairable projection state, never terminal
              -- ownership. This matches decideRuntimeSheetPrepare(): any
              -- attempt may reclaim them, even after a newer failed attempt.
              OR target._status = 'failed'
            )
            AND (
              target._run_id IS DISTINCT FROM $4::text
              OR target._input_index IS DISTINCT FROM input_rows._input_index
              OR target._attempt_id IS DISTINCT FROM $7::text
              OR COALESCE(target._attempt_seq, 0) IS DISTINCT FROM $10::integer
            )
          RETURNING target._key
        ),
        pending_rows AS (
          SELECT _key
          FROM inserted_rows
          UNION
          SELECT _key
          FROM superseded_rows
          UNION
          SELECT _key
          FROM missing_output_rows
          UNION
          SELECT _key
          FROM claimed_existing_rows
          UNION
          SELECT _key
          FROM existing_rows
          WHERE $11::boolean
          UNION
          SELECT _key
          FROM versioned_completed_rows
          WHERE needs_recompute
          UNION
          SELECT _key FROM versioned_failed_rows
          UNION
          SELECT existing._key
          FROM ${sheetTable(session)} AS existing
          JOIN input_rows ON input_rows._key = existing._key
          WHERE (
               existing._write_version = $12::bigint
               AND (
               existing._status IN ('pending', 'running', 'failed')
               AND NOT (${existingNewerTerminalRowSql})
               AND (
                 ${existingAttemptFenceSql}
                 -- See claimed_existing_rows: failed rows remain repairable
                 -- across attempt ordering, while enriched rows stay fenced.
                 OR existing._status = 'failed'
                 OR (
                   ${existingForeignAttemptOwnerSql}
                   AND COALESCE(existing._attempt_seq, 0) < $10::integer
                 )
               )
             ))
             OR (
               existing._write_version = $12::bigint
               AND
               existing._status = 'enriched'
               AND (${existingMissingOutputSql})
               AND (
                 ${existingAttemptFenceSql}
                 OR ${existingWritableEnrichedAttemptSql}
               )
             )
        ),
        completed_rows AS (
          SELECT existing._key
          FROM ${sheetTable(session)} AS existing
          JOIN input_rows ON input_rows._key = existing._key
          WHERE existing._status = 'enriched'
            AND NOT ($11::boolean)
            AND NOT (${existingMissingOutputSql})
            AND NOT EXISTS (
              SELECT 1 FROM pending_rows WHERE pending_rows._key = existing._key
            )
          UNION
          SELECT _key FROM versioned_completed_rows
          WHERE NOT needs_recompute
        ),
        blocked_rows AS (
          SELECT input_rows._key
          FROM input_rows
          WHERE NOT EXISTS (
            SELECT 1 FROM pending_rows WHERE pending_rows._key = input_rows._key
          )
            AND NOT EXISTS (
              SELECT 1 FROM completed_rows WHERE completed_rows._key = input_rows._key
            )
        ),
        row_dispositions AS (
          SELECT _key, 'pending'::text AS disposition
          FROM pending_rows
          UNION ALL
          SELECT _key, 'completed'::text AS disposition
          FROM completed_rows
          UNION ALL
          SELECT _key, 'blocked'::text AS disposition
          FROM blocked_rows
        ),
        inserted_count_cte AS (
          SELECT count(*)::bigint AS c
          FROM inserted_rows
          WHERE NOT EXISTS (
            SELECT 1 FROM preexisting_rows WHERE preexisting_rows._key = inserted_rows._key
          )
        ),
        missing_output_count_cte AS (
          SELECT count(*)::bigint AS c FROM missing_output_rows
        ),
        summary_counts AS (
          SELECT
            (SELECT c FROM inserted_count_cte) AS inserted_count,
            (SELECT c FROM missing_output_count_cte) AS missing_output_count
        ),
        summary_delta AS (
          INSERT INTO ${summaryTable(session)} AS target (play_name, table_namespace, total, queued, running, completed, failed)
          SELECT
            $5::text,
            $6::text,
            inserted_count::int,
            (inserted_count + missing_output_count)::int,
            0,
            (-missing_output_count)::int,
            0
          FROM summary_counts
          WHERE inserted_count > 0 OR missing_output_count > 0
          ON CONFLICT (play_name, table_namespace) DO UPDATE SET
            queued = GREATEST(target.queued + EXCLUDED.queued, 0),
            completed = GREATEST(target.completed + EXCLUDED.completed, 0),
            total = ${runtimeSummaryTotalSql({
              currentTotal: 'target.total',
              totalDelta: 'EXCLUDED.total',
              queued: 'GREATEST(target.queued + EXCLUDED.queued, 0)',
              running: 'target.running',
              completed: 'GREATEST(target.completed + EXCLUDED.completed, 0)',
              failed: 'target.failed',
            })},
            _updated_at = now()
          RETURNING 1
        )
        SELECT
          (SELECT c::int FROM inserted_count_cte) AS inserted_count,
          coalesce(
            (
              SELECT jsonb_agg(jsonb_build_object('key', _key, 'disposition', disposition))
              FROM row_dispositions
            ),
            '[]'::jsonb
          ) AS row_dispositions,
          (SELECT count(*)::int FROM existing_rows) AS reordered_count,
          (SELECT c::int FROM missing_output_count_cte) AS missing_output_count
      `,
      [
        chunkKeys,
        chunkPayloads,
        chunkInputIndexes,
        input.runId,
        input.normalizedPlayName,
        input.normalizedTableNamespace,
        input.attemptId,
        input.attemptOwnerRunId,
        input.attemptExpiresAt,
        input.attemptSeq,
        input.force === true,
        input.writeVersion,
      ],
    );
    inserted += Number(rows[0]?.inserted_count ?? 0);
    if (Array.isArray(rows[0]?.row_dispositions)) {
      for (const row of rows[0]
        ?.row_dispositions as RuntimeSheetPreparedRowDisposition[]) {
        if (
          row &&
          typeof row.key === 'string' &&
          (row.disposition === 'pending' ||
            row.disposition === 'completed' ||
            row.disposition === 'blocked')
        ) {
          rowDispositions.push(row);
        }
      }
    }
  }
  return { inserted, rowDispositions };
}

/**
 * Atomically admit source-table keys that have not completed. A completed row
 * is consumed forever; failed rows and rows owned by this logical run remain
 * recoverable. This is deliberately a separate insert-only path: deriving
 * "new" from a read before the normal insert would race a concurrent sourcing
 * run.
 */
async function prepareNetNewRuntimeSheetDatasetRows(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    chunks: RuntimeDatasetRowEntry[][];
    runId: string;
    attemptId: string;
    attemptOwnerRunId: string;
    attemptExpiresAt: string;
    attemptSeq: number;
    writeVersion: number;
    physicalInsertColumnsSql: string;
    physicalInsertValuesSql: string;
  },
): Promise<{
  inserted: number;
  rowDispositions: RuntimeSheetPreparedRowDisposition[];
}> {
  let inserted = 0;
  const rowDispositions: RuntimeSheetPreparedRowDisposition[] = [];
  for (const chunk of input.chunks) {
    const { rows } = await client.query<{
      inserted_count: number;
      row_dispositions: RuntimeSheetPreparedRowDisposition[];
    }>(
      `WITH input_rows AS (
         SELECT DISTINCT ON (key_values._key)
                key_values._key, payload_values.payload, index_values._input_index
         FROM unnest($1::text[]) WITH ORDINALITY AS key_values(_key, ord)
         JOIN unnest($2::jsonb[]) WITH ORDINALITY AS payload_values(payload, ord)
           ON payload_values.ord = key_values.ord
         JOIN unnest($3::bigint[]) WITH ORDINALITY AS index_values(_input_index, ord)
           ON index_values.ord = key_values.ord
         ORDER BY key_values._key, key_values.ord
       ),
       reclaimed_rows AS (
         UPDATE ${sheetTable(session)} AS target
            SET _status = 'pending',
                _run_id = $4::text,
                _input_index = input_rows._input_index,
                _attempt_id = $5::text,
                _attempt_owner_run_id = $6::text,
                _attempt_expires_at = $7::timestamptz,
                _attempt_seq = $8::integer,
                _write_version = $9::bigint,
                _writer_run_id = $4::text,
                _error = NULL,
                _updated_at = now(),
                _version = ${nextRuntimeSheetVersionExpression(session)}
           FROM input_rows
          WHERE target._key = input_rows._key
            AND (
              target._status = 'failed'
              OR (
                target._status IN ('pending', 'running')
                AND coalesce(target._attempt_owner_run_id, target._run_id) = $6::text
              )
            )
         RETURNING target._key
       ),
       inserted_rows AS (
         INSERT INTO ${sheetTable(session)}
         (_key, _status, _run_id, _input_index, _attempt_id, _attempt_owner_run_id,
          _attempt_expires_at, _attempt_seq, _write_version, _writer_run_id${input.physicalInsertColumnsSql})
       SELECT _key, 'pending', $4::text, _input_index, $5::text, $6::text,
              $7::timestamptz, $8::integer, $9::bigint, $4::text${input.physicalInsertValuesSql}
       FROM input_rows
       ON CONFLICT (_key) DO NOTHING
       RETURNING _key
       ),
       pending_rows AS (
         SELECT _key FROM inserted_rows
         UNION
         SELECT _key FROM reclaimed_rows
       )
       SELECT
         (SELECT count(*)::int FROM inserted_rows) AS inserted_count,
         coalesce(
           (SELECT jsonb_agg(jsonb_build_object('key', _key, 'disposition', 'pending')) FROM pending_rows),
           '[]'::jsonb
         ) AS row_dispositions`,
      [
        chunk.map((entry) => entry.key),
        chunk.map((entry) => stringifyPostgresJson(entry.row)),
        chunk.map((entry) => entry.inputIndex),
        input.runId,
        input.attemptId,
        input.attemptOwnerRunId,
        input.attemptExpiresAt,
        input.attemptSeq,
        input.writeVersion,
      ],
    );
    inserted += Number(rows[0]?.inserted_count ?? 0);
    if (Array.isArray(rows[0]?.row_dispositions)) {
      rowDispositions.push(
        ...rows[0].row_dispositions.filter(
          (row): row is RuntimeSheetPreparedRowDisposition =>
            row != null &&
            typeof row.key === 'string' &&
            row.disposition === 'pending',
        ),
      );
    }
  }
  return { inserted, rowDispositions };
}

async function buildRuntimeSheetDatasetStartResult(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    tableNamespace: string;
    sourceRowsLength: number;
    rowEntries: RuntimeDatasetRowEntry[];
    sheetContract: PlaySheetContract;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
    runId: string;
    attemptId: string;
    attemptOwnerRunId: string;
    attemptExpiresAt: string;
    attemptSeq: number;
    writeVersion: number;
    inserted: number;
    rowDispositions: RuntimeSheetPreparedRowDisposition[];
    force?: boolean;
    timings?: RuntimeSheetTiming[];
  },
): Promise<PrepareRuntimeSheetResult> {
  const datasetFields = input.sheetContract.columns.flatMap((column) =>
    column.source === 'datasetColumn' && typeof column.field === 'string'
      ? [column.field]
      : [],
  );
  const buildFreshRow = (entry: RuntimeDatasetRowEntry) => {
    const row: Record<string, unknown> = {
      ...sanitizePostgresJsonValue(entry.row),
      __deeplineRowKey: entry.key,
    };
    for (const field of datasetFields) {
      if (!Object.prototype.hasOwnProperty.call(row, field)) {
        row[field] = null;
      }
    }
    return row;
  };
  const buildFreshPendingRows = (entries = input.rowEntries) => {
    return entries.map(buildFreshRow);
  };

  if (input.inserted === input.rowEntries.length) {
    return {
      inserted: input.inserted,
      skipped: input.sourceRowsLength - input.rowEntries.length,
      pendingRows: buildFreshPendingRows(),
      completedRows: [],
      blockedRows: [],
      tableNamespace: input.tableNamespace,
      writeVersion: input.writeVersion,
    };
  }

  const dispositionsByKey = new Map(
    input.rowDispositions.map((row) => [row.key, row.disposition]),
  );
  const pendingKeys = new Set(
    input.rowDispositions.flatMap((row) =>
      row.disposition === 'pending' ? [row.key] : [],
    ),
  );
  const startedAt = Date.now();
  await markRuntimeRowsPendingForRecompute(client, session, {
    keys: [...pendingKeys],
    runId: input.runId,
    attemptId: input.attemptId,
    attemptOwnerRunId: input.attemptOwnerRunId,
    attemptExpiresAt: input.attemptExpiresAt,
    attemptSeq: input.attemptSeq,
    writeVersion: input.writeVersion,
    normalizedPlayName: input.normalizedPlayName,
    normalizedTableNamespace: input.normalizedTableNamespace,
    outputFields: input.force === true ? datasetFields : [],
    force: input.force === true,
  });
  if (pendingKeys.size > 0) {
    input.timings?.push({
      phase: 'mark_rows_pending_for_recompute',
      ms: Date.now() - startedAt,
      rows: pendingKeys.size,
    });
  }
  if (input.force === true) {
    const pendingRows: Record<string, unknown>[] = [];
    const blockedRows: Record<string, unknown>[] = [];
    for (const entry of input.rowEntries) {
      const disposition = dispositionsByKey.get(entry.key) ?? 'blocked';
      if (disposition === 'pending') {
        pendingRows.push(buildFreshRow(entry));
      } else if (disposition === 'blocked') {
        blockedRows.push(buildFreshRow(entry));
      }
    }
    return {
      inserted: input.inserted,
      skipped: input.sourceRowsLength - input.rowEntries.length,
      pendingRows,
      completedRows: [],
      blockedRows,
      tableNamespace: input.tableNamespace,
      writeVersion: input.writeVersion,
    };
  }
  const persistedRows = await readRuntimeRowsByKey(
    client,
    session,
    input.rowEntries.map((entry) => entry.key),
    input.sheetContract,
  );
  const persistedRowsByKey = new Map(
    persistedRows.map((row) => [row.key, row.data]),
  );
  const buildMergedRow = (entry: RuntimeDatasetRowEntry) => {
    const merged = { ...entry.row };
    for (const [field, value] of Object.entries(
      persistedRowsByKey.get(entry.key) ?? {},
    )) {
      if (
        value !== null ||
        !Object.prototype.hasOwnProperty.call(merged, field)
      ) {
        merged[field] = value;
      }
    }
    return {
      ...merged,
      __deeplineRowKey: entry.key,
    };
  };
  const pendingRows: Record<string, unknown>[] = [];
  const completedRows: Record<string, unknown>[] = [];
  const blockedRows: Record<string, unknown>[] = [];
  for (const entry of input.rowEntries) {
    const merged = buildMergedRow(entry);
    const disposition = dispositionsByKey.get(entry.key) ?? 'blocked';
    if (disposition === 'pending') {
      pendingRows.push(merged);
    } else if (disposition === 'completed') {
      completedRows.push(merged);
    } else {
      blockedRows.push(merged);
    }
  }
  return {
    inserted: input.inserted,
    skipped: input.sourceRowsLength - input.rowEntries.length,
    pendingRows,
    completedRows,
    blockedRows,
    tableNamespace: input.tableNamespace,
    writeVersion: input.writeVersion ?? undefined,
  };
}

async function markRuntimeRowsPendingForRecompute(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    keys: string[];
    runId: string;
    attemptId: string;
    attemptOwnerRunId: string;
    attemptExpiresAt: string;
    attemptSeq: number;
    writeVersion: number | null;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
    outputFields?: string[];
    force?: boolean;
  },
): Promise<void> {
  if (input.keys.length === 0) return;
  const outputFields = [...new Set(input.outputFields ?? [])];
  await client.query(
    `WITH target_rows AS (
       SELECT _key, _status, _cell_meta
         FROM ${sheetTable(session)}
        WHERE _key = ANY($1::text[])
          AND ($10::boolean OR NOT $10::boolean)
        FOR UPDATE
     ),
     updated AS (
       UPDATE ${sheetTable(session)} AS target
        SET _status = 'pending',
            _run_id = $2::text,
            _attempt_id = $5::text,
            _attempt_owner_run_id = $6::text,
            _attempt_expires_at = $7::timestamptz,
            _attempt_seq = $8::integer,
            _write_version = $11::bigint,
            _writer_run_id = $2::text,
            _updated_at = now(),
            _version = ${nextRuntimeSheetVersionExpression(session)}
        FROM target_rows
      WHERE target._key = target_rows._key
         AND target._write_version = $11::bigint
         AND (
           target._status <> 'pending'
           OR target._run_id IS DISTINCT FROM $2::text
           OR target._attempt_id IS DISTINCT FROM $5::text
           OR COALESCE(target._attempt_seq, 0) IS DISTINCT FROM $8::integer
         )
       RETURNING target_rows._status AS previous_status,
                 target_rows._cell_meta AS previous_cell_meta
     ),
     summary_counts AS (
       SELECT
         count(*) FILTER (WHERE previous_status = 'enriched')::int AS completed_to_pending,
         count(*) FILTER (WHERE previous_status = 'failed')::int AS failed_to_pending,
         count(*) FILTER (WHERE previous_status = 'running')::int AS running_to_pending
       FROM updated
     ),
     column_delta_counts AS (
       SELECT field_values.field,
              count(*) FILTER (
                WHERE previous_cell_meta -> field_values.field ->> 'status' = 'completed'
              )::int AS completed_to_pending,
              count(*) FILTER (
                WHERE previous_cell_meta -> field_values.field ->> 'status' = 'failed'
              )::int AS failed_to_pending
         FROM updated
         JOIN unnest($9::text[]) AS field_values(field) ON true
        GROUP BY field_values.field
     ),
     summary_delta AS (
       INSERT INTO ${summaryTable(session)} AS target (play_name, table_namespace, total, queued, running, completed, failed)
       SELECT
         $3::text,
         $4::text,
         0,
         completed_to_pending + failed_to_pending + running_to_pending,
         -running_to_pending,
         -completed_to_pending,
         -failed_to_pending
       FROM summary_counts
       WHERE completed_to_pending > 0 OR failed_to_pending > 0 OR running_to_pending > 0
       ON CONFLICT (play_name, table_namespace) DO UPDATE SET
         queued = GREATEST(target.queued + EXCLUDED.queued, 0),
         running = GREATEST(target.running + EXCLUDED.running, 0),
         completed = GREATEST(target.completed + EXCLUDED.completed, 0),
         failed = GREATEST(target.failed + EXCLUDED.failed, 0),
         total = ${runtimeSummaryTotalSql({
           currentTotal: 'target.total',
           totalDelta: 'EXCLUDED.total',
           queued: 'GREATEST(target.queued + EXCLUDED.queued, 0)',
           running: 'GREATEST(target.running + EXCLUDED.running, 0)',
           completed: 'GREATEST(target.completed + EXCLUDED.completed, 0)',
           failed: 'GREATEST(target.failed + EXCLUDED.failed, 0)',
         })},
         _updated_at = now()
       RETURNING 1
     ),
     column_summary_delta AS (
       INSERT INTO ${columnSummaryTable(session)} AS target (
         play_name,
         table_namespace,
         field,
         completed,
         failed
       )
       SELECT $3::text,
              $4::text,
              field,
              -completed_to_pending,
              -failed_to_pending
         FROM column_delta_counts
        WHERE completed_to_pending > 0 OR failed_to_pending > 0
       ON CONFLICT (play_name, table_namespace, field) DO UPDATE SET
         completed = GREATEST(target.completed + EXCLUDED.completed, 0),
         failed = GREATEST(target.failed + EXCLUDED.failed, 0),
         _updated_at = now()
       RETURNING 1
     )
     SELECT 1`,
    [
      input.keys,
      input.runId,
      input.normalizedPlayName,
      input.normalizedTableNamespace,
      input.attemptId,
      input.attemptOwnerRunId,
      input.attemptExpiresAt,
      input.attemptSeq,
      outputFields,
      input.force === true,
      input.writeVersion,
    ],
  );
}

async function getRuntimeWorkReceiptSession(
  context: RuntimeApiContext,
  input: {
    playName: string;
    key: string;
  },
): Promise<RuntimePostgresSession> {
  const playName = context.playName?.trim() || input.playName;
  if (!playName) {
    throw new Error('Runtime work receipts require a playName.');
  }
  const runtimeContext = {
    ...context,
    playName,
  };
  const preloaded = findPreloadedRuntimeDbSession(runtimeContext, {
    tableNamespace: RUNTIME_WORK_RECEIPT_TABLE_NAMESPACE,
    logicalTable: RUNTIME_WORK_RECEIPT_LOGICAL_TABLE,
    operations: ['rows.read', 'rows.upsert'],
  });
  const session = requireRuntimePostgresSession(
    preloaded
      ? await unwrapRuntimeDbSession(runtimeContext, preloaded)
      : await getRuntimeDbSession(runtimeContext, {
          tableNamespace: RUNTIME_WORK_RECEIPT_TABLE_NAMESPACE,
          logicalTable: RUNTIME_WORK_RECEIPT_LOGICAL_TABLE,
          operations: ['rows.read', 'rows.upsert'],
        }),
  );
  validateRuntimeWorkReceiptKeyScope(session, { key: input.key });
  return session;
}

/**
 * Work receipts for a play share a single table, so run-scoped operations (the
 * run-fatal lease release) need a session without a specific key to scope by.
 * This mirrors `getRuntimeWorkReceiptSession` minus the per-key scope check.
 */
async function getRuntimeWorkReceiptSessionForRun(
  context: RuntimeApiContext,
  input: {
    playName: string;
  },
): Promise<RuntimePostgresSession> {
  const playName = context.playName?.trim() || input.playName;
  if (!playName) {
    throw new Error('Runtime work receipts require a playName.');
  }
  const runtimeContext = {
    ...context,
    playName,
  };
  const preloaded = findPreloadedRuntimeDbSession(runtimeContext, {
    tableNamespace: RUNTIME_WORK_RECEIPT_TABLE_NAMESPACE,
    logicalTable: RUNTIME_WORK_RECEIPT_LOGICAL_TABLE,
    operations: ['rows.read', 'rows.upsert'],
  });
  return requireRuntimePostgresSession(
    preloaded
      ? await unwrapRuntimeDbSession(runtimeContext, preloaded)
      : await getRuntimeDbSession(runtimeContext, {
          tableNamespace: RUNTIME_WORK_RECEIPT_TABLE_NAMESPACE,
          logicalTable: RUNTIME_WORK_RECEIPT_LOGICAL_TABLE,
          operations: ['rows.read', 'rows.upsert'],
        }),
  );
}

async function getRuntimeWorkReceiptSessionForKeys(
  context: RuntimeApiContext,
  input: {
    playName: string;
    keys: string[];
  },
): Promise<RuntimePostgresSession> {
  const firstKey = input.keys.find((key) => key.trim());
  if (!firstKey) {
    throw new Error('Runtime work receipt batch requires at least one key.');
  }
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: firstKey,
  });
  for (const key of input.keys) {
    validateRuntimeWorkReceiptKeyScope(session, { key });
  }
  return session;
}

async function readRuntimeWorkReceipt(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  key: string,
): Promise<WorkReceipt | null> {
  const { rows } = await client.query<Record<string, unknown>>(
    `SELECT convert_from(k, 'UTF8') AS k,
            status,
            output,
            error,
            error_payload,
            failure_kind,
            run_id,
            lease_id,
            lease_owner_run_id,
            lease_owner_attempt,
            lease_expires_at,
            updated_at
       FROM ${workReceiptTable(session)}
      WHERE k = decode($1, 'hex')`,
    [workReceiptKeyHex(key)],
  );
  return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
}

export async function getRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    key: string;
  },
): Promise<WorkReceipt | null> {
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(context, session, async (client) =>
    readRuntimeWorkReceipt(client, session, input.key),
  );
}

export async function getRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    keys: string[];
  },
): Promise<Array<WorkReceipt | null>> {
  const keys = [
    ...new Set(input.keys.map((key) => key.trim()).filter(Boolean)),
  ];
  if (keys.length === 0) return [];
  const session = await getRuntimeWorkReceiptSessionForKeys(context, {
    playName: input.playName,
    keys,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
      WITH input_keys AS (
        SELECT key_values.key_hex, key_values.ord
        FROM unnest($1::text[]) WITH ORDINALITY AS key_values(key_hex, ord)
      )
      SELECT convert_from(receipts.k, 'UTF8') AS k,
             receipts.status,
             receipts.output,
             receipts.error,
             receipts.error_payload,
             receipts.failure_kind,
             receipts.run_id,
             receipts.lease_id,
             receipts.lease_owner_run_id,
             receipts.lease_owner_attempt,
             receipts.lease_expires_at,
             receipts.updated_at
        FROM input_keys
        LEFT JOIN ${workReceiptTable(session)} AS receipts
          ON receipts.k = decode(input_keys.key_hex, 'hex')
       ORDER BY input_keys.ord
    `,
        [keys.map(workReceiptKeyHex)],
      );
      return rows.map((row) =>
        row.k == null ? null : mapRuntimeWorkReceiptRow(row),
      );
    },
  );
}

export async function claimRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    leaseId?: string | null;
    runAttempt?: number | null;
    leaseAware?: boolean;
    reclaimRunning?: boolean;
    forceRefresh?: boolean;
    forceFailedRefresh?: boolean;
    leaseTtlMs?: number | null;
  },
): Promise<WorkReceiptClaim> {
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const leaseId =
        input.leaseId?.trim() ||
        (input.leaseAware === true ? newRuntimeWorkReceiptLeaseId() : null);
      const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
      const leaseTtlMs = normalizeRuntimeLeaseTtlMs(
        input.leaseTtlMs,
        PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
      );
      // A successful lease-bearing claim is the execution transition. Keep the
      // stored queued status for mixed-version readers; the runner does not add
      // a second queued -> running write on the provider hot path.
      const claimStatus =
        input.leaseAware === true
          ? RECEIPT_STATUS_QUEUED
          : RECEIPT_STATUS_RUNNING;
      const claimableStatuses = workReceiptClaimableStatusCodes({
        forceRefresh: input.forceRefresh,
        forceFailedRefresh: input.forceFailedRefresh,
        reclaimRunning: input.reclaimRunning,
      });
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH existing AS MATERIALIZED (
          SELECT status
          FROM ${workReceiptTable(session)}
          WHERE k = decode($1, 'hex')
          FOR UPDATE
        ),
        claimed AS (
          INSERT INTO ${workReceiptTable(session)} (
            k,
            status,
            run_id,
            lease_id,
            lease_owner_run_id,
            lease_owner_attempt,
            lease_expires_at,
            updated_at
          )
          VALUES (
            decode($1, 'hex'),
            $2::smallint,
            $3,
            $6,
            CASE WHEN $6::text IS NULL THEN NULL ELSE $3 END,
            CASE WHEN $6::text IS NULL THEN NULL ELSE $9::integer END,
            CASE
              WHEN $6::text IS NULL THEN NULL
              ELSE now() + ($7::double precision * interval '1 millisecond')
            END,
            now()
          )
          ON CONFLICT (k) DO UPDATE
          SET status = $4::smallint,
              output = NULL,
              run_id = $3,
              lease_id = $6,
              lease_owner_run_id = CASE WHEN $6::text IS NULL THEN NULL ELSE $3 END,
              lease_owner_attempt = CASE WHEN $6::text IS NULL THEN NULL ELSE $9::integer END,
              lease_expires_at = CASE
                WHEN $6::text IS NULL THEN NULL
                ELSE now() + ($7::double precision * interval '1 millisecond')
              END,
              error = NULL,
              error_payload = NULL,
              failure_kind = 0,
              updated_at = now()
          WHERE ${workReceiptClaimConflictPredicateSql({
            receiptTable: workReceiptTable(session),
            claimableStatusesSql: '$5',
            claimantRunIdSql: '$3',
            claimantRunAttemptSql: '$9',
            reclaimRunningSql: '$11',
            forceRefreshSql: '$8',
            forceFailedRefreshSql: '$10',
          })}
          RETURNING convert_from(k, 'UTF8') AS k,
                    status,
                    output,
                    error,
                    error_payload,
                    failure_kind,
                    run_id,
                    lease_id,
                    lease_owner_run_id,
                    lease_owner_attempt,
                    lease_expires_at,
                    updated_at
        )
        SELECT k,
               status,
               output,
               error,
               error_payload,
               failure_kind,
               run_id,
               lease_id,
               lease_owner_run_id,
               lease_owner_attempt,
               lease_expires_at,
               updated_at,
               EXISTS (
                 SELECT 1
                 FROM existing
                 WHERE status = $12::smallint
               ) AS refreshed_failed_receipt
        FROM claimed
      `,
        [
          workReceiptKeyHex(input.key),
          claimStatus,
          input.runId,
          claimStatus,
          claimableStatuses,
          leaseId,
          leaseTtlMs,
          input.forceRefresh === true,
          runAttempt,
          input.forceFailedRefresh === true,
          input.reclaimRunning === true,
          RECEIPT_STATUS_CODE.failed,
        ],
      );
      const claimedRow = rows[0] ?? null;
      const claimed = claimedRow ? mapRuntimeWorkReceiptRow(claimedRow) : null;
      if (claimed) {
        return {
          disposition: 'claimed',
          receipt: claimed,
          wasFailed:
            claimedRow.refreshed_failed_receipt === true ||
            claimedRow.refreshed_failed_receipt === 'true',
        };
      }

      const latest = await readRuntimeWorkReceipt(client, session, input.key);
      if (
        latest &&
        leaseId !== null &&
        latest.leaseId === leaseId &&
        (latest.leaseOwnerRunId ?? latest.runId) === input.runId &&
        (latest.leaseOwnerAttempt ?? 0) === runAttempt
      ) {
        return { disposition: 'claimed', receipt: latest };
      }
      if (
        latest &&
        isReusableWorkReceipt(latest) &&
        (input.forceRefresh !== true || latest.runId === input.runId)
      ) {
        return { disposition: 'reused', receipt: latest };
      }
      if (latest) {
        return runtimeWorkReceiptClaimDispositionForBlockedReceipt({
          receipt: latest,
          claimantRunId: input.runId,
          claimantRunAttempt: runAttempt,
        });
      }
      throw new Error(
        `Runtime receipt ${input.key} claim did not return execution ownership.`,
      );
    },
  );
}

function runtimeWorkReceiptClaimDispositionForBlockedReceipt(input: {
  receipt: WorkReceipt;
  claimantRunId: string;
  claimantRunAttempt?: number | null;
}): Exclude<WorkReceiptClaim, { disposition: 'claimed' | 'reused' }> {
  const receipt = input.receipt;
  if (receipt.status === 'failed') {
    return { disposition: 'failed', receipt };
  }
  if (
    receipt.status === 'queued' ||
    receipt.status === 'pending' ||
    receipt.status === 'running'
  ) {
    const ownerRunId = receipt.leaseOwnerRunId ?? receipt.runId ?? null;
    const leaseExpiresAtMs =
      typeof receipt.leaseExpiresAt === 'string'
        ? Date.parse(receipt.leaseExpiresAt)
        : NaN;
    const hasActiveForeignLease =
      ownerRunId !== null &&
      ownerRunId !== input.claimantRunId &&
      typeof receipt.leaseId === 'string' &&
      receipt.leaseId.length > 0 &&
      Number.isFinite(leaseExpiresAtMs) &&
      leaseExpiresAtMs > Date.now();
    return {
      disposition: hasActiveForeignLease ? 'blocked_active_lease' : 'running',
      receipt,
    };
  }
  return { disposition: 'running', receipt };
}

export async function claimRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    keys: string[];
    leaseIds?: string[];
    runAttempt?: number | null;
    leaseAware?: boolean;
    reclaimRunning?: boolean;
    forceRefresh?: boolean;
    forceFailedRefresh?: boolean;
    leaseTtlMs?: number | null;
  },
): Promise<WorkReceiptClaim[]> {
  if (
    input.leaseIds !== undefined &&
    input.leaseIds.length !== input.keys.length
  ) {
    throw new Error(
      `Runtime receipt bulk claim requires one lease ID per key. Received ${input.leaseIds.length} lease IDs for ${input.keys.length} keys.`,
    );
  }
  const positionalEntries = input.keys
    .map((key, originalIndex) => ({ key: key.trim(), originalIndex }))
    .filter((entry) => Boolean(entry.key));
  const positionalKeys = positionalEntries.map((entry) => entry.key);
  if (positionalKeys.length === 0) return [];
  const firstPositionByKey = new Map<string, number>();
  positionalEntries.forEach(({ key, originalIndex }) => {
    if (!firstPositionByKey.has(key)) {
      firstPositionByKey.set(key, originalIndex);
    }
  });
  const keys = [...firstPositionByKey.keys()];
  const session = await getRuntimeWorkReceiptSessionForKeys(context, {
    playName: input.playName,
    keys,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const keyHexes = keys.map(workReceiptKeyHex);
      const leaseIds = keys.map((key) => {
        const position = firstPositionByKey.get(key)!;
        const providedLeaseId = input.leaseIds?.[position]?.trim();
        return (
          providedLeaseId ||
          (input.leaseAware === true ? newRuntimeWorkReceiptLeaseId() : null)
        );
      });
      const claimStatus =
        input.leaseAware === true
          ? RECEIPT_STATUS_QUEUED
          : RECEIPT_STATUS_RUNNING;
      const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
      const leaseTtlMs = normalizeRuntimeLeaseTtlMs(
        input.leaseTtlMs,
        PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
      );
      const reclaimStatuses = workReceiptClaimableStatusCodes({
        forceRefresh: input.forceRefresh,
        forceFailedRefresh: input.forceFailedRefresh,
        reclaimRunning: input.reclaimRunning,
      });
      const { rows } = await client.query<Record<string, unknown>>(
        `
      WITH input_keys AS (
        SELECT key_values.key_hex, lease_values.lease_id, key_values.ord
        FROM unnest($1::text[]) WITH ORDINALITY AS key_values(key_hex, ord)
        JOIN unnest($2::text[]) WITH ORDINALITY AS lease_values(lease_id, ord)
          ON lease_values.ord = key_values.ord
      ),
      claimed AS (
        INSERT INTO ${workReceiptTable(session)} (
          k,
          status,
          run_id,
          lease_id,
          lease_owner_run_id,
          lease_owner_attempt,
          lease_expires_at,
          updated_at
        )
        SELECT decode(input_keys.key_hex, 'hex'),
               $3::smallint,
               $4::text,
               input_keys.lease_id,
               CASE
                 WHEN input_keys.lease_id IS NULL THEN NULL
                 ELSE $4::text
               END,
               CASE
                 WHEN input_keys.lease_id IS NULL THEN NULL
                 ELSE $9::integer
               END,
               CASE
                 WHEN input_keys.lease_id IS NULL THEN NULL
                 ELSE now() + ($7::double precision * interval '1 millisecond')
               END,
               now()
          FROM input_keys
          ON CONFLICT (k) DO UPDATE
          SET status = $5::smallint,
              output = NULL,
              run_id = $4::text,
            lease_id = EXCLUDED.lease_id,
            lease_owner_run_id = CASE
              WHEN EXCLUDED.lease_id IS NULL THEN NULL
              ELSE $4::text
            END,
            lease_owner_attempt = CASE
              WHEN EXCLUDED.lease_id IS NULL THEN NULL
              ELSE $9::integer
            END,
            lease_expires_at = EXCLUDED.lease_expires_at,
            error = NULL,
            error_payload = NULL,
            failure_kind = 0,
            updated_at = now()
        WHERE ${workReceiptClaimConflictPredicateSql({
          receiptTable: workReceiptTable(session),
          claimableStatusesSql: '$6',
          claimantRunIdSql: '$4::text',
          claimantRunAttemptSql: '$9',
          reclaimRunningSql: '$11',
          forceRefreshSql: '$8',
          forceFailedRefreshSql: '$10',
        })}
        RETURNING k,
                  status,
                  output,
                  error,
                  error_payload,
                  failure_kind,
                  run_id,
                  lease_id,
                  lease_owner_run_id,
                  lease_owner_attempt,
                  lease_expires_at,
                  updated_at
      ),
      latest AS (
        SELECT receipts.k,
               receipts.status,
               receipts.output,
               receipts.error,
               receipts.error_payload,
               receipts.failure_kind,
               receipts.run_id,
               receipts.lease_id,
               receipts.lease_owner_run_id,
               receipts.lease_owner_attempt,
               receipts.lease_expires_at,
               receipts.updated_at,
               input_keys.ord,
               receipts.lease_id = input_keys.lease_id
                 AND receipts.lease_owner_run_id = $4::text
                 AND COALESCE(receipts.lease_owner_attempt, 0) = $9::integer
                 AS claimed
          FROM input_keys
          JOIN ${workReceiptTable(session)} AS receipts
            ON receipts.k = decode(input_keys.key_hex, 'hex')
         WHERE NOT EXISTS (
           SELECT 1 FROM claimed WHERE claimed.k = receipts.k
         )
      ),
      returned AS (
        SELECT claimed.k,
               claimed.status,
               claimed.output,
               claimed.error,
               claimed.error_payload,
               claimed.failure_kind,
               claimed.run_id,
               claimed.lease_id,
               claimed.lease_owner_run_id,
               claimed.lease_owner_attempt,
               claimed.lease_expires_at,
               claimed.updated_at,
               input_keys.ord,
               true AS claimed
          FROM claimed
          JOIN input_keys ON claimed.k = decode(input_keys.key_hex, 'hex')
        UNION ALL
        SELECT k,
               status,
               output,
               error,
               error_payload,
               failure_kind,
               run_id,
               lease_id,
               lease_owner_run_id,
               lease_owner_attempt,
               lease_expires_at,
               updated_at,
               ord,
               claimed
          FROM latest
      )
      SELECT convert_from(k, 'UTF8') AS k,
             status,
             output,
             error,
             error_payload,
             failure_kind,
             run_id,
             lease_id,
             lease_owner_run_id,
             lease_owner_attempt,
             lease_expires_at,
             updated_at,
             claimed
        FROM returned
       ORDER BY ord
    `,
        [
          keyHexes,
          leaseIds,
          claimStatus,
          input.runId,
          claimStatus,
          reclaimStatuses,
          leaseTtlMs,
          input.forceRefresh === true,
          runAttempt,
          input.forceFailedRefresh === true,
          input.reclaimRunning === true,
        ],
      );
      const claimsByKey = new Map<string, WorkReceiptClaim>();
      for (const row of rows) {
        const receipt = mapRuntimeWorkReceiptRow(row);
        if (row.claimed === true) {
          claimsByKey.set(receipt.key, {
            disposition: 'claimed',
            receipt,
          });
          continue;
        }
        if (
          isReusableWorkReceipt(receipt) &&
          (input.forceRefresh !== true || receipt.runId === input.runId)
        ) {
          claimsByKey.set(receipt.key, {
            disposition: 'reused',
            receipt,
          });
          continue;
        }
        claimsByKey.set(
          receipt.key,
          runtimeWorkReceiptClaimDispositionForBlockedReceipt({
            receipt,
            claimantRunId: input.runId,
            claimantRunAttempt: runAttempt,
          }),
        );
      }
      return positionalKeys.map((key) => {
        const claim = claimsByKey.get(key);
        if (!claim) {
          throw new Error(
            `Runtime receipt ${key} bulk claim did not return a positional result.`,
          );
        }
        return claim;
      });
    },
  );
}

export async function markRuntimeWorkReceiptRunning(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    runAttempt?: number | null;
    leaseId?: string | null;
    leaseTtlMs?: number | null;
  },
): Promise<WorkReceipt | null> {
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const leaseId = input.leaseId?.trim() || null;
      const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
      const leaseTtlMs = normalizeRuntimeLeaseTtlMs(
        input.leaseTtlMs,
        PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
      );
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH running AS (
          UPDATE ${workReceiptTable(session)}
             SET status = $2::smallint,
                 error = NULL,
                 error_payload = NULL,
                 failure_kind = 0,
                 lease_expires_at = CASE
                   WHEN lease_id IS NULL THEN NULL
                   ELSE now() + ($6::double precision * interval '1 millisecond')
                 END,
                 updated_at = now()
           WHERE k = decode($1, 'hex')
             AND ${workReceiptActiveOwnerPredicateSql({
               receiptTable: workReceiptTable(session),
               ownerRunIdSql: '$3',
               ownerRunAttemptSql: '$5',
               leaseIdSql: '$4::text',
             })}
          RETURNING convert_from(k, 'UTF8') AS k,
                    status,
                    output,
                    error,
                    error_payload,
                    failure_kind,
                    run_id,
                    lease_id,
                    lease_owner_run_id,
                    lease_owner_attempt,
                    lease_expires_at,
                    updated_at
        )
        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at
          FROM running
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_RUNNING,
          input.runId,
          leaseId,
          runAttempt,
          leaseTtlMs,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

export async function markRuntimeWorkReceiptsRunning(
  context: RuntimeApiContext,
  input: {
    playName: string;
    receipts: Array<{
      runId: string;
      key: string;
      leaseId?: string | null;
      runAttempt?: number | null;
    }>;
    leaseTtlMs?: number | null;
  },
): Promise<Array<WorkReceipt | null>> {
  const receipts = input.receipts.filter((receipt) => receipt.key.trim());
  if (receipts.length === 0) return [];
  const leaseTtlMs = normalizeRuntimeLeaseTtlMs(
    input.leaseTtlMs,
    PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
  );
  const session = await getRuntimeWorkReceiptSessionForKeys(context, {
    playName: input.playName,
    keys: receipts.map((receipt) => receipt.key),
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
      WITH inputs AS (
        SELECT key_values.key_hex,
               run_values.run_id,
               lease_values.lease_id,
               attempt_values.run_attempt,
               key_values.ord
          FROM unnest($1::text[]) WITH ORDINALITY AS key_values(key_hex, ord)
          JOIN unnest($2::text[]) WITH ORDINALITY AS run_values(run_id, ord)
            ON run_values.ord = key_values.ord
          JOIN unnest($3::text[]) WITH ORDINALITY AS lease_values(lease_id, ord)
            ON lease_values.ord = key_values.ord
          JOIN unnest($4::integer[]) WITH ORDINALITY AS attempt_values(run_attempt, ord)
            ON attempt_values.ord = key_values.ord
      ),
      running AS (
        UPDATE ${workReceiptTable(session)} AS target
           SET status = $5::smallint,
               error = NULL,
               error_payload = NULL,
               failure_kind = 0,
               lease_expires_at = CASE
                 WHEN target.lease_id IS NULL THEN NULL
                 ELSE now() + ($6::double precision * interval '1 millisecond')
               END,
               updated_at = now()
          FROM inputs
         WHERE target.k = decode(inputs.key_hex, 'hex')
           AND ${workReceiptActiveOwnerPredicateSql({
             receiptTable: 'target',
             ownerRunIdSql: 'inputs.run_id',
             ownerRunAttemptSql: 'inputs.run_attempt',
             leaseIdSql: 'inputs.lease_id',
           })}
        RETURNING target.k,
                  target.status,
                  target.output,
                  target.error,
                  target.error_payload,
                  target.failure_kind,
                  target.run_id,
                  target.lease_id,
                  target.lease_owner_run_id,
                  target.lease_owner_attempt,
                  target.lease_expires_at,
                  target.updated_at,
                  inputs.ord
      ),
      returned AS (
        SELECT running.k,
               running.status,
               running.output,
               running.error,
               running.error_payload,
               running.failure_kind,
               running.run_id,
               running.lease_id,
               running.lease_owner_run_id,
               running.lease_owner_attempt,
               running.lease_expires_at,
               running.updated_at,
               inputs.ord
          FROM inputs
          LEFT JOIN running ON running.ord = inputs.ord
      )
      SELECT convert_from(returned.k, 'UTF8') AS k,
             returned.status,
             returned.output,
             returned.error,
             returned.error_payload,
             returned.failure_kind,
             returned.run_id,
             returned.lease_id,
             returned.lease_owner_run_id,
             returned.lease_owner_attempt,
             returned.lease_expires_at,
             returned.updated_at
        FROM returned
       ORDER BY returned.ord
    `,
        [
          receipts.map((receipt) => workReceiptKeyHex(receipt.key)),
          receipts.map((receipt) => receipt.runId),
          receipts.map((receipt) => receipt.leaseId?.trim() || null),
          receipts.map((receipt) =>
            normalizeRuntimeRunAttempt(receipt.runAttempt),
          ),
          RECEIPT_STATUS_RUNNING,
          leaseTtlMs,
        ],
      );
      return rows.map((row) =>
        row.k == null ? null : mapRuntimeWorkReceiptRow(row),
      );
    },
  );
}

export async function markRuntimeWorkReceiptsQueued(
  context: RuntimeApiContext,
  input: {
    playName: string;
    receipts: Array<{
      runId: string;
      key: string;
      leaseId?: string | null;
      runAttempt?: number | null;
    }>;
    leaseTtlMs?: number | null;
  },
): Promise<Array<WorkReceipt | null>> {
  const receipts = input.receipts.filter((receipt) => receipt.key.trim());
  if (receipts.length === 0) return [];
  const leaseTtlMs = normalizeRuntimeLeaseTtlMs(
    input.leaseTtlMs,
    PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
  );
  const session = await getRuntimeWorkReceiptSessionForKeys(context, {
    playName: input.playName,
    keys: receipts.map((receipt) => receipt.key),
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
      WITH inputs AS (
        SELECT key_values.key_hex,
               run_values.run_id,
               lease_values.lease_id,
               attempt_values.run_attempt,
               key_values.ord
          FROM unnest($1::text[]) WITH ORDINALITY AS key_values(key_hex, ord)
          JOIN unnest($2::text[]) WITH ORDINALITY AS run_values(run_id, ord)
            ON run_values.ord = key_values.ord
          JOIN unnest($3::text[]) WITH ORDINALITY AS lease_values(lease_id, ord)
            ON lease_values.ord = key_values.ord
          JOIN unnest($4::integer[]) WITH ORDINALITY AS attempt_values(run_attempt, ord)
            ON attempt_values.ord = key_values.ord
      ),
      queued AS (
        UPDATE ${workReceiptTable(session)} AS target
           SET status = $5::smallint,
               error = NULL,
               error_payload = NULL,
               failure_kind = 0,
               lease_expires_at = CASE
                 WHEN target.lease_id IS NULL THEN NULL
                 ELSE now() + ($6::double precision * interval '1 millisecond')
               END,
               updated_at = now()
          FROM inputs
         WHERE target.k = decode(inputs.key_hex, 'hex')
           AND ${workReceiptActiveOwnerPredicateSql({
             receiptTable: 'target',
             ownerRunIdSql: 'inputs.run_id',
             ownerRunAttemptSql: 'inputs.run_attempt',
             leaseIdSql: 'inputs.lease_id',
           })}
        RETURNING target.k,
                  target.status,
                  target.output,
                  target.error,
                  target.error_payload,
                  target.failure_kind,
                  target.run_id,
                  target.lease_id,
                  target.lease_owner_run_id,
                  target.lease_owner_attempt,
                  target.lease_expires_at,
                  target.updated_at,
                  inputs.ord
      ),
      returned AS (
        SELECT queued.k,
               queued.status,
               queued.output,
               queued.error,
               queued.error_payload,
               queued.failure_kind,
               queued.run_id,
               queued.lease_id,
               queued.lease_owner_run_id,
               queued.lease_owner_attempt,
               queued.lease_expires_at,
               queued.updated_at,
               inputs.ord
          FROM inputs
          LEFT JOIN queued ON queued.ord = inputs.ord
      )
      SELECT convert_from(returned.k, 'UTF8') AS k,
             returned.status,
             returned.output,
             returned.error,
             returned.error_payload,
             returned.failure_kind,
             returned.run_id,
             returned.lease_id,
             returned.lease_owner_run_id,
             returned.lease_owner_attempt,
             returned.lease_expires_at,
             returned.updated_at
        FROM returned
       ORDER BY returned.ord
    `,
        [
          receipts.map((receipt) => workReceiptKeyHex(receipt.key)),
          receipts.map((receipt) => receipt.runId),
          receipts.map((receipt) => receipt.leaseId?.trim() || null),
          receipts.map((receipt) =>
            normalizeRuntimeRunAttempt(receipt.runAttempt),
          ),
          RECEIPT_STATUS_QUEUED,
          leaseTtlMs,
        ],
      );
      return rows.map((row) =>
        row.k == null ? null : mapRuntimeWorkReceiptRow(row),
      );
    },
  );
}

/**
 * Replay-idempotency fence for a receipt whose active complete/fail UPDATE
 * matched 0 rows. This fires when the durable write committed but the HTTP
 * response was lost and the SAME caller retries: the row is already terminal,
 * so the active-owner predicate (which requires status IN (queued, running))
 * can never match again. We surface the already-terminal row as success ONLY
 * when its run identity matches the retrying caller, so the retry does not
 * spuriously fail a receipt whose provider spend and durable write both
 * succeeded.
 *
 * The predicate mirrors the run-identity half of {@link
 * workReceiptActiveOwnerPredicateSql}: `COALESCE(lease_owner_run_id, run_id)`
 * must equal the caller's run id. complete/fail null the lease-owner and
 * attempt columns as part of the terminal write, so run id is the only durable
 * ownership marker left on a terminal row — a different run (the stale-owner
 * rejection case) never matches and still resolves to null exactly as before.
 * This is replay-idempotency only; it never lets a foreign run read or claim a
 * receipt it did not complete.
 */
function workReceiptTerminalReplayPredicateSql(input: {
  receiptTable: string;
  terminalStatusSql: string;
  ownerRunIdSql: string;
}): string {
  const table = input.receiptTable;
  return `${table}.status = ${input.terminalStatusSql}::smallint
            /* replay-idempotency: same-run retry after a lost response; run identity is the only durable owner marker on a terminal row */
            AND COALESCE(${table}.lease_owner_run_id, ${table}.run_id) = ${input.ownerRunIdSql}`;
}

export async function completeRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    runAttempt?: number | null;
    leaseId?: string | null;
    output: unknown;
    returnOutput?: boolean;
  },
): Promise<WorkReceipt | null> {
  if (
    consumeDirectRuntimeTestFault({
      context,
      runId: input.runId,
      name: 'receipt_complete_write_fail',
    })
  ) {
    const error = runtimeTestFaultError('receipt_complete_write_fail');
    // Tool receipts are immutable completed facts. A failed publication is an
    // execution failure, never a cached failed receipt.
    if (input.leaseId === COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID) {
      throw error;
    }
    const failed = await failRuntimeWorkReceipt(context, {
      playName: input.playName,
      runId: input.runId,
      key: input.key,
      leaseId: input.leaseId,
      error: error.message,
      failureKind: 'repairable',
      runAttempt: input.runAttempt,
    });
    if (!failed) {
      await forceFailRuntimeWorkReceiptForRuntimeTestFault(context, {
        playName: input.playName,
        runId: input.runId,
        key: input.key,
        error: error.message,
        failureKind: 'repairable',
        runAttempt: input.runAttempt,
      });
    }
    throw error;
  }
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const immutableCacheInsert =
        input.leaseId === COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID;
      const leaseId = immutableCacheInsert
        ? null
        : input.leaseId?.trim() || null;
      if (immutableCacheInsert) {
        const { rows: insertedRows } = await client.query<
          Record<string, unknown>
        >(
          `
          INSERT INTO ${workReceiptTable(session)} AS receipt (
            k, status, output, error, error_payload, failure_kind, run_id, updated_at
          ) VALUES (
            decode($1, 'hex'), $2::smallint, $3::jsonb, NULL, NULL, 0, $4, now()
          )
          ON CONFLICT (k) DO UPDATE
          SET status = EXCLUDED.status,
              output = EXCLUDED.output,
              error = NULL,
              error_payload = NULL,
              failure_kind = 0,
              run_id = EXCLUDED.run_id,
              lease_id = NULL,
              lease_owner_run_id = NULL,
              lease_owner_attempt = NULL,
              lease_expires_at = NULL,
              updated_at = now()
          WHERE receipt.status NOT IN (
            ${RECEIPT_STATUS_COMPLETED}::smallint,
            ${RECEIPT_STATUS_SKIPPED}::smallint
          )
          RETURNING convert_from(k, 'UTF8') AS k, status, output, error, error_payload,
                    failure_kind, run_id, lease_id, lease_owner_run_id,
                    lease_owner_attempt, lease_expires_at, updated_at
          `,
          [
            workReceiptKeyHex(input.key),
            RECEIPT_STATUS_COMPLETED,
            input.output === null ? null : stringifyPostgresJson(input.output),
            input.runId,
          ],
        );
        if (insertedRows[0]) return mapRuntimeWorkReceiptRow(insertedRows[0]);
        const { rows: winnerRows } = await client.query<
          Record<string, unknown>
        >(
          `SELECT convert_from(k, 'UTF8') AS k, status, output, error, error_payload,
                  failure_kind, run_id, lease_id, lease_owner_run_id,
                  lease_owner_attempt, lease_expires_at, updated_at
             FROM ${workReceiptTable(session)}
            WHERE k = decode($1, 'hex')
              AND status IN ($2::smallint, ${RECEIPT_STATUS_SKIPPED}::smallint)
            LIMIT 1`,
          [workReceiptKeyHex(input.key), RECEIPT_STATUS_COMPLETED],
        );
        return winnerRows[0] ? mapRuntimeWorkReceiptRow(winnerRows[0]) : null;
      }
      const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH completed AS (
          UPDATE ${workReceiptTable(session)}
          SET status = $2::smallint,
              output = $3::jsonb,
              error = NULL,
              error_payload = NULL,
              failure_kind = 0,
              run_id = $4,
              lease_id = NULL,
              lease_owner_run_id = NULL,
              lease_owner_attempt = NULL,
              lease_expires_at = NULL,
              updated_at = now()
          WHERE k = decode($1, 'hex')
            AND ${workReceiptActiveOwnerPredicateSql({
              receiptTable: workReceiptTable(session),
              ownerRunIdSql: '$4',
              ownerRunAttemptSql: '$6',
              leaseIdSql: '$5::text',
            })}
	          RETURNING convert_from(k, 'UTF8') AS k,
	                    status,
	                    CASE WHEN $7::boolean THEN output ELSE NULL::jsonb END AS output,
	                    error,
	                    error_payload,
	                    failure_kind,
	                    run_id,
	                    lease_id,
	                    lease_owner_run_id,
	                    lease_owner_attempt,
	                    lease_expires_at,
	                    updated_at
	        ),
	        replay AS (
	          SELECT convert_from(k, 'UTF8') AS k,
	                 status,
	                 CASE WHEN $7::boolean THEN output ELSE NULL::jsonb END AS output,
	                 error,
	                 error_payload,
	                 failure_kind,
	                 run_id,
	                 lease_id,
	                 lease_owner_run_id,
	                 lease_owner_attempt,
	                 lease_expires_at,
	                 updated_at
	            FROM ${workReceiptTable(session)}
	           WHERE k = decode($1, 'hex')
	             AND NOT EXISTS (SELECT 1 FROM completed)
	             AND ${workReceiptTerminalReplayPredicateSql({
                 receiptTable: workReceiptTable(session),
                 terminalStatusSql: '$2',
                 ownerRunIdSql: '$4',
               })}
	        )
	        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM completed
	        UNION ALL
	        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM replay
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_COMPLETED,
          input.output === null ? null : stringifyPostgresJson(input.output),
          input.runId,
          leaseId,
          runAttempt,
          input.returnOutput !== false,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

export async function completeRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    receipts: Array<{
      runId: string;
      key: string;
      output: unknown;
      leaseId?: string | null;
      runAttempt?: number | null;
    }>;
    returnOutput?: boolean;
  },
): Promise<Array<WorkReceipt | null>> {
  const receipts = input.receipts.filter((receipt) => receipt.key.trim());
  if (receipts.length === 0) return [];
  if (
    consumeDirectRuntimeTestFault({
      context,
      runId: receipts[0]?.runId ?? '',
      name: 'receipt_complete_write_fail',
    })
  ) {
    const faultedIndex = receipts.findIndex(
      (receipt) => receipt.leaseId !== COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID,
    );
    if (faultedIndex < 0) {
      throw runtimeTestFaultError('receipt_complete_write_fail');
    }
    const faultedReceipt = receipts[faultedIndex];
    const siblingReceipts = receipts.filter(
      (_receipt, index) => index !== faultedIndex,
    );
    let siblingError: unknown = null;
    if (siblingReceipts.length > 0) {
      try {
        await completeRuntimeWorkReceipts(context, {
          playName: input.playName,
          receipts: siblingReceipts,
          returnOutput: input.returnOutput,
        });
      } catch (error) {
        siblingError = error;
      }
    }
    const error = runtimeTestFaultError('receipt_complete_write_fail');
    if (faultedReceipt) {
      const failed = await failRuntimeWorkReceipt(context, {
        playName: input.playName,
        runId: faultedReceipt.runId,
        key: faultedReceipt.key,
        leaseId: faultedReceipt.leaseId,
        error: error.message,
        failureKind: 'repairable',
        runAttempt: faultedReceipt.runAttempt,
      });
      if (!failed) {
        await forceFailRuntimeWorkReceiptForRuntimeTestFault(context, {
          playName: input.playName,
          runId: faultedReceipt.runId,
          key: faultedReceipt.key,
          error: error.message,
          failureKind: 'repairable',
          runAttempt: faultedReceipt.runAttempt,
        });
      }
    }
    if (siblingError) {
      throw siblingError;
    }
    throw error;
  }
  const immutableEntries = receipts
    .map((receipt, index) => ({ receipt, index }))
    .filter(
      ({ receipt }) =>
        receipt.leaseId === COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID,
    );
  if (immutableEntries.length > 0) {
    const leasedEntries = receipts
      .map((receipt, index) => ({ receipt, index }))
      .filter(
        ({ receipt }) =>
          receipt.leaseId !== COMPLETED_RECEIPT_CACHE_INSERT_LEASE_ID,
      );
    const resolved = Array<WorkReceipt | null>(receipts.length).fill(null);
    const [immutableResults, leasedResults] = await Promise.all([
      Promise.all(
        immutableEntries.map(({ receipt }) =>
          completeRuntimeWorkReceipt(context, {
            playName: input.playName,
            ...receipt,
          }),
        ),
      ),
      leasedEntries.length > 0
        ? completeRuntimeWorkReceipts(context, {
            playName: input.playName,
            receipts: leasedEntries.map(({ receipt }) => receipt),
          })
        : Promise.resolve([]),
    ]);
    immutableEntries.forEach(({ index }, resultIndex) => {
      resolved[index] = immutableResults[resultIndex] ?? null;
    });
    leasedEntries.forEach(({ index }, resultIndex) => {
      resolved[index] = leasedResults[resultIndex] ?? null;
    });
    return resolved;
  }
  const session = await getRuntimeWorkReceiptSessionForKeys(context, {
    playName: input.playName,
    keys: receipts.map((receipt) => receipt.key),
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
      WITH inputs AS (
        SELECT key_values.key_hex,
               run_values.run_id,
               lease_values.lease_id,
               attempt_values.run_attempt,
               output_values.output,
               key_values.ord
          FROM unnest($1::text[]) WITH ORDINALITY AS key_values(key_hex, ord)
          JOIN unnest($2::text[]) WITH ORDINALITY AS run_values(run_id, ord)
            ON run_values.ord = key_values.ord
          JOIN unnest($3::text[]) WITH ORDINALITY AS lease_values(lease_id, ord)
            ON lease_values.ord = key_values.ord
          JOIN unnest($4::integer[]) WITH ORDINALITY AS attempt_values(run_attempt, ord)
            ON attempt_values.ord = key_values.ord
          JOIN jsonb_array_elements($5::jsonb) WITH ORDINALITY AS output_values(output, ord)
            ON output_values.ord = key_values.ord
      ),
      completed AS (
        UPDATE ${workReceiptTable(session)} AS target
           SET status = $6::smallint,
		               output = inputs.output,
	               error = NULL,
	               error_payload = NULL,
	               failure_kind = 0,
	               run_id = inputs.run_id,
	               lease_id = NULL,
	               lease_owner_run_id = NULL,
	               lease_owner_attempt = NULL,
	               lease_expires_at = NULL,
               updated_at = now()
          FROM inputs
         WHERE target.k = decode(inputs.key_hex, 'hex')
           AND ${workReceiptActiveOwnerPredicateSql({
             receiptTable: 'target',
             ownerRunIdSql: 'inputs.run_id',
             ownerRunAttemptSql: 'inputs.run_attempt',
             leaseIdSql: 'inputs.lease_id',
           })}
        RETURNING target.k,
                  target.status,
	                  CASE WHEN $7::boolean THEN target.output ELSE NULL::jsonb END AS output,
	                  target.error,
	                  target.error_payload,
	                  target.failure_kind,
	                  target.run_id,
                  target.lease_id,
                  target.lease_owner_run_id,
                  target.lease_owner_attempt,
                  target.lease_expires_at,
                  target.updated_at,
                  inputs.ord
      ),
      replay AS (
        SELECT target.k,
               target.status,
               CASE WHEN $7::boolean THEN target.output ELSE NULL::jsonb END AS output,
               target.error,
               target.error_payload,
               target.failure_kind,
               target.run_id,
               target.lease_id,
               target.lease_owner_run_id,
               target.lease_owner_attempt,
               target.lease_expires_at,
               target.updated_at,
               inputs.ord
          FROM inputs
          JOIN ${workReceiptTable(session)} AS target
            ON target.k = decode(inputs.key_hex, 'hex')
         WHERE NOT EXISTS (SELECT 1 FROM completed WHERE completed.ord = inputs.ord)
           AND ${workReceiptTerminalReplayPredicateSql({
             receiptTable: 'target',
             terminalStatusSql: '$6',
             ownerRunIdSql: 'inputs.run_id',
           })}
      ),
      resolved AS (
        SELECT ord, k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM completed
        UNION ALL
        SELECT ord, k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM replay
      ),
      returned AS (
        SELECT resolved.k,
               resolved.status,
	               resolved.output,
	               resolved.error,
	               resolved.error_payload,
	               resolved.failure_kind,
	               resolved.run_id,
               resolved.lease_id,
               resolved.lease_owner_run_id,
               resolved.lease_owner_attempt,
               resolved.lease_expires_at,
               resolved.updated_at,
               inputs.ord
          FROM inputs
          LEFT JOIN resolved ON resolved.ord = inputs.ord
      )
      SELECT convert_from(returned.k, 'UTF8') AS k,
             returned.status,
	             returned.output,
	             returned.error,
	             returned.error_payload,
	             returned.failure_kind,
		             returned.run_id,
             returned.lease_id,
             returned.lease_owner_run_id,
             returned.lease_owner_attempt,
             returned.lease_expires_at,
             returned.updated_at
        FROM returned
       ORDER BY returned.ord
    `,
        [
          receipts.map((receipt) => workReceiptKeyHex(receipt.key)),
          receipts.map((receipt) => receipt.runId),
          receipts.map((receipt) => receipt.leaseId?.trim() || null),
          receipts.map((receipt) =>
            normalizeRuntimeRunAttempt(receipt.runAttempt),
          ),
          stringifyPostgresJson(
            receipts.map((receipt) =>
              receipt.output === null ? null : receipt.output,
            ),
          ),
          RECEIPT_STATUS_COMPLETED,
          input.returnOutput !== false,
        ],
      );
      return rows.map((row) =>
        row.k == null ? null : mapRuntimeWorkReceiptRow(row),
      );
    },
  );
}

export async function failRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    runAttempt?: number | null;
    leaseId?: string | null;
    error: string;
    errorPayload?: ToolExecutionFailureV1 | null;
    failureKind?: WorkReceiptFailureKind | null;
  },
): Promise<WorkReceipt | null> {
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const leaseId = input.leaseId?.trim() || null;
      const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH failed AS (
          UPDATE ${workReceiptTable(session)}
	          SET status = $2::smallint,
	              output = NULL,
	              error = $3,
	              error_payload = $8::jsonb,
	              failure_kind = $6::smallint,
              run_id = $4,
              lease_id = NULL,
              lease_owner_run_id = NULL,
              lease_owner_attempt = NULL,
              lease_expires_at = NULL,
              updated_at = now()
          WHERE k = decode($1, 'hex')
            AND ${workReceiptActiveOwnerPredicateSql({
              receiptTable: workReceiptTable(session),
              ownerRunIdSql: '$4',
              ownerRunAttemptSql: '$7',
              leaseIdSql: '$5::text',
            })}
	          RETURNING convert_from(k, 'UTF8') AS k,
	                    status,
	                    output,
	                    error,
	                    error_payload,
	                    failure_kind,
	                    run_id,
	                    lease_id,
	                    lease_owner_run_id,
	                    lease_owner_attempt,
	                    lease_expires_at,
	                    updated_at
	        ),
	        replay AS (
	          SELECT convert_from(k, 'UTF8') AS k,
	                 status,
	                 output,
	                 error,
	                 error_payload,
	                 failure_kind,
	                 run_id,
	                 lease_id,
	                 lease_owner_run_id,
	                 lease_owner_attempt,
	                 lease_expires_at,
	                 updated_at
	            FROM ${workReceiptTable(session)}
	           WHERE k = decode($1, 'hex')
	             AND NOT EXISTS (SELECT 1 FROM failed)
	             AND ${workReceiptTerminalReplayPredicateSql({
                 receiptTable: workReceiptTable(session),
                 terminalStatusSql: '$2',
                 ownerRunIdSql: '$4',
               })}
	        )
	        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM failed
	        UNION ALL
	        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM replay
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_FAILED,
          input.error,
          input.runId,
          leaseId,
          workReceiptFailureKindCodeForWrite(input.failureKind),
          runAttempt,
          input.errorPayload ? JSON.stringify(input.errorPayload) : null,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

export async function releaseRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    runAttempt?: number | null;
    leaseId?: string | null;
  },
): Promise<WorkReceipt | null> {
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const leaseId = input.leaseId?.trim() || null;
      const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH released AS (
          UPDATE ${workReceiptTable(session)}
             SET status = $2::smallint,
                 output = NULL,
                 error = NULL,
                 error_payload = NULL,
                 failure_kind = 0,
                 run_id = $3,
                 lease_id = NULL,
                 lease_owner_run_id = NULL,
                 lease_owner_attempt = NULL,
                 lease_expires_at = NULL,
                 updated_at = now()
           WHERE k = decode($1, 'hex')
             AND ${workReceiptActiveOwnerPredicateSql({
               receiptTable: workReceiptTable(session),
               ownerRunIdSql: '$3',
               ownerRunAttemptSql: '$5',
               leaseIdSql: '$4::text',
             })}
          RETURNING convert_from(k, 'UTF8') AS k,
                    status,
                    output,
                    error,
                    error_payload,
                    failure_kind,
                    run_id,
                    lease_id,
                    lease_owner_run_id,
                    lease_owner_attempt,
                    lease_expires_at,
                    updated_at
        )
        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM released
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_PENDING,
          input.runId,
          leaseId,
          runAttempt,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

export async function failRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    receipts: Array<{
      runId: string;
      key: string;
      error: string;
      errorPayload?: ToolExecutionFailureV1 | null;
      failureKind?: WorkReceiptFailureKind | null;
      leaseId?: string | null;
      runAttempt?: number | null;
    }>;
  },
): Promise<Array<WorkReceipt | null>> {
  const receipts = input.receipts.filter((receipt) => receipt.key.trim());
  if (receipts.length === 0) return [];
  const session = await getRuntimeWorkReceiptSessionForKeys(context, {
    playName: input.playName,
    keys: receipts.map((receipt) => receipt.key),
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
      WITH inputs AS (
        SELECT key_values.key_hex,
               run_values.run_id,
	               lease_values.lease_id,
	               attempt_values.run_attempt,
	               error_values.error,
	               payload_values.error_payload::jsonb,
	               failure_kind_values.failure_kind,
	               key_values.ord
          FROM unnest($1::text[]) WITH ORDINALITY AS key_values(key_hex, ord)
          JOIN unnest($2::text[]) WITH ORDINALITY AS run_values(run_id, ord)
            ON run_values.ord = key_values.ord
          JOIN unnest($3::text[]) WITH ORDINALITY AS lease_values(lease_id, ord)
            ON lease_values.ord = key_values.ord
	          JOIN unnest($4::integer[]) WITH ORDINALITY AS attempt_values(run_attempt, ord)
	            ON attempt_values.ord = key_values.ord
	          JOIN unnest($5::text[]) WITH ORDINALITY AS error_values(error, ord)
	            ON error_values.ord = key_values.ord
	          JOIN unnest($6::smallint[]) WITH ORDINALITY AS failure_kind_values(failure_kind, ord)
	            ON failure_kind_values.ord = key_values.ord
	          JOIN unnest($7::text[]) WITH ORDINALITY AS payload_values(error_payload, ord)
	            ON payload_values.ord = key_values.ord
      ),
      failed AS (
        UPDATE ${workReceiptTable(session)} AS target
	           SET status = $8::smallint,
	               output = NULL,
	               error = inputs.error,
	               error_payload = inputs.error_payload,
	               failure_kind = inputs.failure_kind,
	               run_id = inputs.run_id,
               lease_id = NULL,
               lease_owner_run_id = NULL,
               lease_owner_attempt = NULL,
               lease_expires_at = NULL,
               updated_at = now()
          FROM inputs
         WHERE target.k = decode(inputs.key_hex, 'hex')
	           AND ${workReceiptActiveOwnerPredicateSql({
               receiptTable: 'target',
               ownerRunIdSql: 'inputs.run_id',
               ownerRunAttemptSql: 'inputs.run_attempt',
               leaseIdSql: 'inputs.lease_id',
             })}
        RETURNING target.k,
                  target.status,
	                  target.output,
	                  target.error,
	                  target.error_payload,
	                  target.failure_kind,
	                  target.run_id,
                  target.lease_id,
                  target.lease_owner_run_id,
                  target.lease_owner_attempt,
                  target.lease_expires_at,
                  target.updated_at,
                  inputs.ord
      ),
      replay AS (
        SELECT target.k,
               target.status,
               target.output,
               target.error,
               target.error_payload,
               target.failure_kind,
               target.run_id,
               target.lease_id,
               target.lease_owner_run_id,
               target.lease_owner_attempt,
               target.lease_expires_at,
               target.updated_at,
               inputs.ord
          FROM inputs
          JOIN ${workReceiptTable(session)} AS target
            ON target.k = decode(inputs.key_hex, 'hex')
         WHERE NOT EXISTS (SELECT 1 FROM failed WHERE failed.ord = inputs.ord)
           AND ${workReceiptTerminalReplayPredicateSql({
             receiptTable: 'target',
             terminalStatusSql: '$8',
             ownerRunIdSql: 'inputs.run_id',
           })}
      ),
      resolved AS (
        SELECT ord, k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM failed
        UNION ALL
        SELECT ord, k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM replay
      ),
      returned AS (
        SELECT resolved.k,
               resolved.status,
	               resolved.output,
	               resolved.error,
	               resolved.error_payload,
	               resolved.failure_kind,
	               resolved.run_id,
               resolved.lease_id,
               resolved.lease_owner_run_id,
               resolved.lease_owner_attempt,
               resolved.lease_expires_at,
               resolved.updated_at,
               inputs.ord
          FROM inputs
          LEFT JOIN resolved ON resolved.ord = inputs.ord
      )
      SELECT convert_from(returned.k, 'UTF8') AS k,
             returned.status,
	             returned.output,
	             returned.error,
	             returned.error_payload,
	             returned.failure_kind,
		             returned.run_id,
             returned.lease_id,
             returned.lease_owner_run_id,
             returned.lease_owner_attempt,
             returned.lease_expires_at,
             returned.updated_at
        FROM returned
       ORDER BY returned.ord
    `,
        [
          receipts.map((receipt) => workReceiptKeyHex(receipt.key)),
          receipts.map((receipt) => receipt.runId),
          receipts.map((receipt) => receipt.leaseId?.trim() || null),
          receipts.map((receipt) =>
            normalizeRuntimeRunAttempt(receipt.runAttempt),
          ),
          receipts.map((receipt) => receipt.error),
          receipts.map((receipt) =>
            workReceiptFailureKindCodeForWrite(receipt.failureKind),
          ),
          receipts.map((receipt) =>
            receipt.errorPayload ? JSON.stringify(receipt.errorPayload) : null,
          ),
          RECEIPT_STATUS_FAILED,
        ],
      );
      return rows.map((row) =>
        row.k == null ? null : mapRuntimeWorkReceiptRow(row),
      );
    },
  );
}

export async function heartbeatRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    runAttempt?: number | null;
    leaseId?: string;
    leaseIds?: string[];
    keys: string[];
    leaseTtlMs?: number | null;
  },
): Promise<Array<WorkReceipt | null>> {
  if (
    input.leaseIds !== undefined &&
    input.leaseIds.length !== input.keys.length
  ) {
    throw new Error(
      `Runtime receipt bulk heartbeat requires one lease ID per key. Received ${input.leaseIds.length} lease IDs for ${input.keys.length} keys.`,
    );
  }
  const entries = input.keys
    .map((key, index) => ({
      key: key.trim(),
      leaseId: input.leaseIds?.[index]?.trim() || input.leaseId?.trim() || '',
    }))
    .filter((entry) => Boolean(entry.key));
  const keys = entries.map((entry) => entry.key);
  if (keys.length === 0) return [];
  const leaseIds = entries.map((entry) => entry.leaseId);
  if (leaseIds.some((leaseId) => !leaseId)) return keys.map(() => null);
  const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
  const leaseTtlMs = normalizeRuntimeLeaseTtlMs(
    input.leaseTtlMs,
    PLAY_RUNTIME_WORK_RECEIPT_LEASE_TTL_MS,
  );
  const session = await getRuntimeWorkReceiptSessionForKeys(context, {
    playName: input.playName,
    keys,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<Record<string, unknown>>(
        `
      WITH input_keys AS (
        SELECT key_values.key_hex, lease_values.lease_id, key_values.ord
        FROM unnest($1::text[]) WITH ORDINALITY AS key_values(key_hex, ord)
        JOIN unnest($3::text[]) WITH ORDINALITY AS lease_values(lease_id, ord)
          ON lease_values.ord = key_values.ord
      ),
      unique_inputs AS (
        SELECT DISTINCT key_hex, lease_id
          FROM input_keys
      ),
      renewed AS (
        UPDATE ${workReceiptTable(session)} AS target
           SET lease_expires_at = now() + ($4::double precision * interval '1 millisecond'),
               updated_at = now()
          FROM unique_inputs
         WHERE target.k = decode(unique_inputs.key_hex, 'hex')
           AND ${workReceiptHeartbeatPredicateSql({
             receiptTable: 'target',
             ownerRunIdSql: '$2::text',
             ownerRunAttemptSql: '$5',
             leaseIdSql: 'unique_inputs.lease_id',
           })}
        RETURNING target.k,
                  target.status,
                  target.output,
                  target.error,
                  target.error_payload,
                  target.failure_kind,
                  target.run_id,
                  target.lease_id,
                  target.lease_owner_run_id,
                  target.lease_owner_attempt,
                  target.lease_expires_at,
                  target.updated_at
      ),
      returned AS (
        SELECT renewed.k,
               renewed.status,
               renewed.output,
               renewed.error,
               renewed.error_payload,
               renewed.failure_kind,
               renewed.run_id,
               renewed.lease_id,
               renewed.lease_owner_run_id,
               renewed.lease_owner_attempt,
               renewed.lease_expires_at,
               renewed.updated_at,
               input_keys.ord
          FROM input_keys
          LEFT JOIN renewed
            ON renewed.k = decode(input_keys.key_hex, 'hex')
           AND renewed.lease_id = input_keys.lease_id
      )
      SELECT convert_from(returned.k, 'UTF8') AS k,
             returned.status,
             returned.output,
             returned.error,
             returned.error_payload,
             returned.failure_kind,
             returned.run_id,
             returned.lease_id,
             returned.lease_owner_run_id,
             returned.lease_owner_attempt,
             returned.lease_expires_at,
             returned.updated_at
        FROM returned
       ORDER BY returned.ord
    `,
        [
          keys.map(workReceiptKeyHex),
          input.runId,
          leaseIds,
          leaseTtlMs,
          runAttempt,
        ],
      );
      return rows.map((row) =>
        row.k == null ? null : mapRuntimeWorkReceiptRow(row),
      );
    },
  );
}

export async function skipRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    runAttempt?: number | null;
    leaseId?: string | null;
    output: unknown;
  },
): Promise<WorkReceipt | null> {
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const leaseId = input.leaseId?.trim() || null;
      const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH skipped AS (
          UPDATE ${workReceiptTable(session)}
	          SET status = $2::smallint,
	              output = $3::jsonb,
	              error = NULL,
	              error_payload = NULL,
	              failure_kind = 0,
              run_id = $4,
              lease_id = NULL,
              lease_owner_run_id = NULL,
              lease_owner_attempt = NULL,
              lease_expires_at = NULL,
              updated_at = now()
          WHERE k = decode($1, 'hex')
            AND ${workReceiptActiveOwnerPredicateSql({
              receiptTable: workReceiptTable(session),
              ownerRunIdSql: '$4',
              ownerRunAttemptSql: '$6',
              leaseIdSql: '$5::text',
            })}
	          RETURNING convert_from(k, 'UTF8') AS k,
	                    status,
	                    output,
	                    error,
	                    error_payload,
	                    failure_kind,
	                    run_id,
	                    lease_id,
	                    lease_owner_run_id,
	                    lease_owner_attempt,
	                    lease_expires_at,
	                    updated_at
	        )
	        SELECT k, status, output, error, error_payload, failure_kind, run_id, lease_id, lease_owner_run_id, lease_owner_attempt, lease_expires_at, updated_at FROM skipped
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_SKIPPED,
          input.output === null ? null : stringifyPostgresJson(input.output),
          input.runId,
          leaseId,
          runAttempt,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

export async function startRuntimeSheetDataset(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    playInput?: Record<string, unknown> | null;
    sheetContract: PlaySheetContract;
    rows: Record<string, unknown>[];
    runId: string;
    attemptId?: string | null;
    attemptOwnerRunId?: string | null;
    attemptExpiresAt?: string | null;
    attemptSeq?: number | null;
    attemptLeaseTtlMs?: number | null;
    writeVersion?: number | null;
    inputOffset?: number;
    force?: boolean;
    mode?: RuntimeSheetDatasetMode;
  },
): Promise<PrepareRuntimeSheetResult> {
  if (context.dbSessionStrategy === 'gateway_only') {
    return await postRuntimeApi<PrepareRuntimeSheetResult>(context, {
      action: 'runtime_sheet_start',
      input,
    });
  }
  const totalStartedAt = Date.now();
  const timings: RuntimeSheetTiming[] = [];
  const playName = context.playName?.trim() || input.playName;
  if (!playName) {
    throw new Error('Runtime DB sessions require a playName.');
  }
  const sessionStartedAt = Date.now();
  const session = requireRuntimePostgresSession(
    await getRuntimeDbSession(
      {
        ...context,
        playName,
        runId: context.runId ?? input.runId,
      },
      {
        tableNamespace: input.tableNamespace,
        logicalTable: 'sheet_rows',
        operations: ['rows.read', 'rows.upsert'],
        limits: {
          maxRows: runtimeDbSessionRowLimit(input.rows.length),
        },
        sheetContract: input.sheetContract,
        timings,
      },
    ),
  );
  timings.push({
    phase: 'db_session',
    ms: Date.now() - sessionStartedAt,
    rows: input.rows.length,
  });
  // Register the exact run-scoped Dataset Handle before any row can become
  // durable. A crash after a later sheet write can therefore never leave
  // customer data discoverable only through static pipeline inference.
  await registerRuntimeDataset(
    session,
    {
      playName,
      tableNamespace: input.tableNamespace,
      runId: input.runId,
    },
    context.abortSignal,
  );
  const normalizeStartedAt = Date.now();
  const uniqueRows = new Map<
    string,
    { row: Record<string, unknown>; inputIndex: number | null }
  >();
  for (const row of input.rows) {
    // Materializes projected CSV aliases as visible cells and drops internal
    // __deepline* keys — a plain spread would silently lose the
    // non-enumerable alias fields on the JSON payload boundary.
    const cleanedRow = toSerializableCsvAliasedRow(row);
    const key =
      resolveMapRowOutcomeKey(row) ??
      derivePlayRowIdentity(cleanedRow, input.tableNamespace);
    if (key && !uniqueRows.has(key)) {
      uniqueRows.set(key, {
        row: cleanedRow,
        inputIndex: normalizeRuntimeMapInputIndex(
          row[MAP_ROW_OUTCOME_RUNTIME_FIELDS.inputIndex],
        ),
      });
    }
  }
  const inputOffset = Math.max(0, Math.floor(input.inputOffset ?? 0));
  const rowEntries = [...uniqueRows.entries()].map(
    ([key, entry], inputIndex) => ({
      key,
      row: entry.row,
      inputIndex: entry.inputIndex ?? inputOffset + inputIndex,
    }),
  );
  timings.push({
    phase: 'normalize_rows',
    ms: Date.now() - normalizeStartedAt,
    rows: input.rows.length,
  });
  if (rowEntries.length === 0) {
    return {
      inserted: 0,
      skipped: input.rows.length,
      pendingRows: [],
      completedRows: [],
      blockedRows: [],
      tableNamespace: input.tableNamespace,
      timings: [
        ...timings,
        {
          phase: 'total',
          ms: Date.now() - totalStartedAt,
          rows: input.rows.length,
        },
      ],
    };
  }

  const physicalColumns = physicalSheetColumnNames(input.sheetContract);
  const physicalInsertColumnsSql =
    physicalColumns.length > 0
      ? `, ${physicalColumns.map(quoteIdentifier).join(', ')}`
      : '';
  const physicalInsertValuesSql =
    physicalColumns.length > 0
      ? `, ${physicalColumns
          .map((column) => `payload -> ${quoteLiteral(column)}`)
          .join(', ')}`
      : '';
  const outputPhysicalColumns = outputPhysicalSheetColumnProjections(
    input.sheetContract,
  );
  const outputPhysicalColumnNames = new Set(
    outputPhysicalColumns.map((column) => column.sqlName),
  );
  const physicalRefreshColumns = physicalColumns.filter(
    (column) => !outputPhysicalColumnNames.has(column),
  );
  const physicalRefreshSetSql =
    physicalRefreshColumns.length > 0
      ? `, ${physicalRefreshColumns
          .map(
            (column) =>
              `${quoteIdentifier(column)} = input_rows.payload -> ${quoteLiteral(column)}`,
          )
          .join(', ')}`
      : '';
  const physicalUpsertSetSql =
    physicalColumns.length > 0
      ? `, ${physicalColumns
          .map(
            (column) =>
              `${quoteIdentifier(column)} = EXCLUDED.${quoteIdentifier(column)}`,
          )
          .join(', ')}`
      : '';
  const normalizedPlayName = normalizePlayNameForSheet(playName);
  const normalizedTableNamespace = normalizeTableNamespace(
    input.tableNamespace,
  );
  const attemptId = normalizeRuntimeSheetAttemptId(input.attemptId);
  const attemptOwnerRunId = normalizeRuntimeSheetAttemptOwnerRunId({
    attemptOwnerRunId: input.attemptOwnerRunId,
    runId: input.runId,
  });
  const attemptSeq = normalizeRuntimeRunAttempt(input.attemptSeq);
  const chunks = chunkValues(rowEntries, DIRECT_POSTGRES_BATCH_SIZE);
  // Scheduling is one admission decision. Even when batching SQL for the
  // parameter limit, either every requested row receives this intent version
  // or none do; a run can never leave 42 pending rows after 8 conflicts.
  const needsTransaction = true;

  const result = await withRuntimeSheetQueryClient(
    context,
    session,
    {
      playName,
      tableNamespace: input.tableNamespace,
      sheetContract: input.sheetContract,
      transactional: needsTransaction,
      timings,
    },
    async (client) => {
      const attemptExpiryStartedAt = Date.now();
      const attemptExpiresAt = await mintRuntimeSheetAttemptExpiresAt(
        client,
        input.attemptLeaseTtlMs,
      );
      const writeVersion = await mintRuntimeSheetWriteVersion(
        client,
        session,
        input.writeVersion,
      );
      timings.push({
        phase: 'attempt_expiry_mint',
        ms: Date.now() - attemptExpiryStartedAt,
        rows: rowEntries.length,
      });
      const prepareStartedAt = Date.now();
      const prepared =
        input.mode === 'net_new'
          ? await prepareNetNewRuntimeSheetDatasetRows(client, session, {
              chunks,
              runId: input.runId,
              attemptId,
              attemptOwnerRunId,
              attemptExpiresAt,
              attemptSeq,
              writeVersion,
              physicalInsertColumnsSql,
              physicalInsertValuesSql,
            })
          : await prepareRuntimeSheetDatasetRows(client, session, {
              chunks,
              runId: input.runId,
              normalizedPlayName,
              normalizedTableNamespace,
              attemptId,
              attemptOwnerRunId,
              attemptExpiresAt,
              attemptSeq,
              writeVersion,
              physicalInsertColumnsSql,
              physicalInsertValuesSql,
              physicalRefreshSetSql:
                input.force === true ? physicalRefreshSetSql : '',
              physicalUpsertSetSql,
              outputPhysicalColumns,
              force: input.force === true,
            });
      timings.push({
        phase: 'prepare_rows_sql',
        ms: Date.now() - prepareStartedAt,
        rows: rowEntries.length,
        chunks: chunks.length,
        inserted: prepared.inserted,
        pending: prepared.rowDispositions.filter(
          (row) => row.disposition === 'pending',
        ).length,
        ...(input.force ? { force: true } : {}),
      });
      const buildStartedAt = Date.now();
      const built = await buildRuntimeSheetDatasetStartResult(client, session, {
        tableNamespace: input.tableNamespace,
        sourceRowsLength: input.rows.length,
        rowEntries,
        sheetContract: input.sheetContract,
        normalizedPlayName,
        normalizedTableNamespace,
        runId: input.runId,
        attemptId,
        attemptOwnerRunId,
        attemptExpiresAt,
        attemptSeq,
        writeVersion,
        timings,
        force: input.force === true,
        ...prepared,
      });
      timings.push({
        phase: 'build_result',
        ms: Date.now() - buildStartedAt,
        rows: rowEntries.length,
        inserted: built.inserted,
        skipped: built.skipped,
        pending: built.pendingRows.length,
        completed: built.completedRows.length,
      });
      return {
        ...built,
        attemptExpiresAt,
        attemptSeq,
        writeVersion,
      };
    },
  );
  timings.push({
    phase: 'total',
    ms: Date.now() - totalStartedAt,
    rows: input.rows.length,
    chunks: chunks.length,
    inserted: result.inserted,
    skipped: result.skipped,
    pending: result.pendingRows.length,
    completed: result.completedRows.length,
  });
  return {
    ...result,
    attemptId,
    attemptOwnerRunId,
    attemptExpiresAt: result.attemptExpiresAt,
    attemptSeq,
    writeVersion: result.writeVersion,
    timings,
  };
}

export async function heartbeatRuntimeSheetAttempt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    sheetContract: PlaySheetContract;
    runId: string;
    attemptId: string;
    attemptOwnerRunId?: string | null;
    attemptSeq?: number | null;
    attemptLeaseTtlMs?: number | null;
    keys: string[];
  },
): Promise<RuntimeSheetAttemptHeartbeatResult> {
  const playName = context.playName?.trim() || input.playName;
  if (!playName) {
    throw new Error('Runtime sheet attempt heartbeat requires a playName.');
  }
  const attemptId = input.attemptId.trim();
  const attemptOwnerRunId = normalizeRuntimeSheetAttemptOwnerRunId({
    attemptOwnerRunId: input.attemptOwnerRunId,
    runId: input.runId,
  });
  const attemptSeq = normalizeRuntimeRunAttempt(input.attemptSeq);
  const attemptLeaseTtlMs = normalizeRuntimeLeaseTtlMs(
    input.attemptLeaseTtlMs,
    PLAY_RUNTIME_SHEET_ATTEMPT_LEASE_TTL_MS,
  );
  const keys = [
    ...new Set(input.keys.map((key) => key.trim()).filter(Boolean)),
  ];
  if (!attemptId || keys.length === 0) {
    return { renewed: 0, renewedKeys: [], attemptExpiresAt: null };
  }
  const timings: RuntimeSheetTiming[] = [];
  const session = requireRuntimePostgresSession(
    await getRuntimeDbSession(
      {
        ...context,
        playName,
        runId: context.runId ?? input.runId,
      },
      {
        tableNamespace: input.tableNamespace,
        logicalTable: 'sheet_rows',
        operations: ['rows.upsert'],
        limits: {
          maxRows: runtimeDbSessionRowLimit(keys.length),
        },
        sheetContract: input.sheetContract,
        timings,
      },
    ),
  );
  return await withRuntimeSheetQueryClient(
    context,
    session,
    {
      playName,
      tableNamespace: input.tableNamespace,
      sheetContract: input.sheetContract,
      transactional: false,
      timings,
    },
    async (client) => {
      const { rows } = await client.query<{
        renewed_keys: string[];
        renewed: number;
        attempt_expires_at: Date | string | null;
      }>(
        `
        WITH input_keys AS (
          SELECT key_values._key
          FROM unnest($1::text[]) AS key_values(_key)
        ),
        owned_terminal AS (
          SELECT target._key
            FROM ${sheetTable(session)} AS target
            JOIN input_keys ON target._key = input_keys._key
           WHERE target._attempt_id = $2::text
             AND coalesce(target._attempt_owner_run_id, target._run_id) = $3::text
             AND target._run_id = $4::text
             AND COALESCE(target._attempt_seq, 0) = $6::integer
             AND target._status = 'enriched'
        ),
        renewed AS (
          UPDATE ${sheetTable(session)} AS target
             SET _attempt_expires_at = now() + ($5::double precision * interval '1 millisecond'),
                 _updated_at = now(),
                 _version = ${nextRuntimeSheetVersionExpression(session)}
            FROM input_keys
           WHERE target._key = input_keys._key
             AND target._attempt_id = $2::text
             AND coalesce(target._attempt_owner_run_id, target._run_id) = $3::text
             AND target._run_id = $4::text
             AND COALESCE(target._attempt_seq, 0) = $6::integer
             AND target._status IN ('pending', 'running', 'failed')
           RETURNING target._key, target._attempt_expires_at
        ),
        owned_keys AS (
          SELECT _key FROM renewed
          UNION
          SELECT _key FROM owned_terminal
        )
        SELECT coalesce((SELECT array_agg(_key ORDER BY _key) FROM owned_keys), '{}'::text[]) AS renewed_keys,
               count(*)::int AS renewed,
               max(_attempt_expires_at) AS attempt_expires_at
          FROM renewed
        `,
        [
          keys,
          attemptId,
          attemptOwnerRunId,
          input.runId,
          attemptLeaseTtlMs,
          attemptSeq,
        ],
      );
      const row = rows[0];
      const attemptExpiresAt = row?.attempt_expires_at;
      return {
        renewed: Number(row?.renewed ?? 0),
        renewedKeys: row?.renewed_keys ?? [],
        attemptExpiresAt:
          attemptExpiresAt instanceof Date
            ? attemptExpiresAt.toISOString()
            : typeof attemptExpiresAt === 'string'
              ? new Date(attemptExpiresAt).toISOString()
              : null,
      };
    },
  );
}

/**
 * Controlled run-fatal teardown for runtime sheet rows: clear the attempt lease
 * (`_attempt_*`) on every row the aborting attempt still owns, so an immediate
 * rerun can claim pending rows and overwrite the aborted attempt's rows without
 * waiting out the 10-minute lease TTL. The predicate is attempt-fenced (see
 * `releasableRuntimeSheetAttemptFenceSql` / `decideRuntimeSheetRelease`): it only
 * touches rows owned by this run's attempt generation, so it can never weaken a
 * different run's live lease. TTL expiry remains the recovery path for true
 * crashes where no teardown code runs. When `attemptId` is null every attempt id
 * the run holds at `attemptSeq` is released (the workers runner mints a distinct
 * attempt id per map chunk under one owner run id + seq).
 */
export async function releaseRuntimeSheetAttempt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    sheetContract: PlaySheetContract;
    runId: string;
    attemptId?: string | null;
    attemptOwnerRunId?: string | null;
    attemptSeq?: number | null;
  },
): Promise<RuntimeSheetAttemptReleaseResult> {
  if (context.dbSessionStrategy === 'gateway_only') {
    return await postRuntimeApi<RuntimeSheetAttemptReleaseResult>(context, {
      action: 'runtime_sheet_release_attempt',
      input,
    });
  }
  const playName = context.playName?.trim() || input.playName;
  if (!playName) {
    throw new Error('Runtime sheet attempt release requires a playName.');
  }
  const attemptId = input.attemptId?.trim() || null;
  const attemptOwnerRunId = normalizeRuntimeSheetAttemptOwnerRunId({
    attemptOwnerRunId: input.attemptOwnerRunId,
    runId: input.runId,
  });
  const attemptSeq = normalizeRuntimeRunAttempt(input.attemptSeq);
  if (!attemptOwnerRunId) {
    return { released: 0, releasedKeys: [] };
  }
  const timings: RuntimeSheetTiming[] = [];
  const session = requireRuntimePostgresSession(
    await getRuntimeDbSession(
      {
        ...context,
        playName,
        runId: context.runId ?? input.runId,
      },
      {
        tableNamespace: input.tableNamespace,
        logicalTable: 'sheet_rows',
        operations: ['rows.upsert'],
        limits: {},
        sheetContract: input.sheetContract,
        timings,
      },
    ),
  );
  return await withRuntimeSheetQueryClient(
    context,
    session,
    {
      playName,
      tableNamespace: input.tableNamespace,
      sheetContract: input.sheetContract,
      transactional: false,
      timings,
    },
    async (client) => {
      const { rows } = await client.query<{ _key: string }>(
        `
        UPDATE ${sheetTable(session)} AS target
           SET _attempt_id = NULL,
               _attempt_owner_run_id = NULL,
               _attempt_expires_at = NULL,
               _attempt_seq = NULL,
               _updated_at = now(),
               _version = ${nextRuntimeSheetVersionExpression(session)}
         WHERE ${releasableRuntimeSheetAttemptFenceSql(
           'target',
           '$1::text',
           '$2::text',
           '$3',
         )}
        RETURNING target._key
        `,
        [attemptId, attemptOwnerRunId, attemptSeq],
      );
      const releasedKeys = rows
        .map((row) => row._key)
        .filter((key): key is string => typeof key === 'string');
      return { released: releasedKeys.length, releasedKeys };
    },
  );
}

/**
 * Controlled run-fatal teardown for work receipts: return every receipt still
 * leased `running` by the aborting run+attempt back to `pending` (clearing the
 * lease) so an immediate rerun can reclaim and recompute the interrupted work
 * without waiting out the lease TTL. Fenced by owner run id + attempt (see
 * `workReceiptReleaseOwnerPredicateSql`), so it never touches a receipt a live
 * competitor legitimately holds. Completed/skipped/failed receipts are terminal
 * and untouched, preserving cross-run idempotency (already-persisted work is
 * still reused on rerun). Work receipts for a play share one table, so a single
 * run-scoped statement covers every stranded receipt.
 */
export async function releaseRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    runAttempt?: number | null;
  },
): Promise<RuntimeWorkReceiptReleaseResult> {
  if (context.dbSessionStrategy === 'gateway_only') {
    return await postRuntimeApi<RuntimeWorkReceiptReleaseResult>(context, {
      action: 'runtime_receipts_release_attempt',
      input,
    });
  }
  const runId = input.runId.trim();
  if (!runId) {
    return { released: 0, releasedKeys: [] };
  }
  const runAttempt = normalizeRuntimeRunAttempt(input.runAttempt);
  const session = await getRuntimeWorkReceiptSessionForRun(context, {
    playName: input.playName,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const { rows } = await client.query<{ k: string }>(
        `
        UPDATE ${workReceiptTable(session)} AS target
           SET status = $3::smallint,
               lease_id = NULL,
               lease_owner_run_id = NULL,
               lease_owner_attempt = NULL,
               lease_expires_at = NULL,
               updated_at = now()
         WHERE ${workReceiptReleaseOwnerPredicateSql({
           receiptTable: 'target',
           ownerRunIdSql: '$1',
           ownerRunAttemptSql: '$2',
         })}
        RETURNING convert_from(target.k, 'UTF8') AS k
        `,
        [runId, runAttempt, RECEIPT_STATUS_PENDING],
      );
      const releasedKeys = rows
        .map((row) => row.k)
        .filter((key): key is string => typeof key === 'string');
      return { released: releasedKeys.length, releasedKeys };
    },
  );
}

type CompleteRuntimeMapRowChunksInput = {
  chunks: RuntimePreparedCompletedRow[][];
  physicalUpdateSetSql: string;
  physicalColumnProjections: PhysicalSheetColumnProjection[];
  runId: string;
  attemptId: string | null;
  attemptOwnerRunId: string | null;
  attemptExpiresAt: string | null;
  attemptSeq: number;
  writeVersion: number | null;
  normalizedPlayName: string;
  normalizedTableNamespace: string;
  outputFields: string[];
};

export type RuntimeMapRowsWriteResult = {
  /** Submitted row identities durably accepted, including idempotent replays. */
  committedKeys: string[];
  /** Submitted row identities rejected because a newer write owns the row. */
  staleKeys: string[];
  /** Physical row mutations. Idempotent commits do not increment this count. */
  updated: number;
  /** Compatibility alias retained for callers that still call stale rows fenced. */
  fencedKeys: string[];
  staleDropped?: number;
  staleDroppedKeys?: string[];
  /** Synthetic Preview-only proof that the gateway applied the page-tail hold. */
  runtimeTestPageTailHoldMs?: number;
};

type RuntimeMapRowsMutationResult = {
  updated: number;
  fencedKeys: string[];
  conflictKeys: string[];
};

function classifyRuntimeMapRowsWrite(input: {
  submittedKeys: Iterable<string>;
  updated: number;
  fencedKeys: Iterable<string>;
}): RuntimeMapRowsWriteResult {
  const staleKeys = [...new Set(input.fencedKeys)];
  const staleSet = new Set(staleKeys);
  const committedKeys = [...new Set(input.submittedKeys)].filter(
    (key) => !staleSet.has(key),
  );
  return {
    committedKeys,
    staleKeys,
    updated: input.updated,
    fencedKeys: staleKeys,
    staleDropped: staleKeys.length,
    staleDroppedKeys: staleKeys,
  };
}

function writableEnrichedRuntimeSheetAttemptSql(
  tableAlias: string,
  attemptOwnerRunIdExpression: string,
  attemptExpiresAtExpression: string,
  attemptSeqExpression: string,
): string {
  return `(
    ${tableAlias}._status = 'enriched'
    AND (
      COALESCE(${tableAlias}._attempt_seq, 0) < ${attemptSeqExpression}::integer
    )
  )`;
}

async function completeRuntimeMapRowChunks(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: CompleteRuntimeMapRowChunksInput,
): Promise<RuntimeMapRowsMutationResult> {
  let updated = 0;
  const fencedKeys: string[] = [];
  const conflictKeys: string[] = [];
  for (const chunk of input.chunks) {
    const chunkKeys = chunk.map((row) => row.key);
    const chunkInputIndexes = chunk.map((row) => row.input_index);
    const chunkDataPatches = chunk.map((row) => row.data_patch_json);
    const chunkCellMetaPatches = chunk.map((row) => row.cell_meta_patch_json);
    const targetChangedPatchedCellSql = changedPatchedCellSql(
      'target',
      'updates.data_patch',
      input.physicalColumnProjections,
    );
    const writableEnrichedAttemptSql = writableEnrichedRuntimeSheetAttemptSql(
      'target',
      '$10::text',
      '$11::timestamptz',
      '$12',
    );
    const writableNonTerminalAttemptSql = `(
      target._status <> 'enriched'
      AND (
        ${activeRuntimeSheetAttemptFenceSql('target', '$9::text', '$10::text', '$11::timestamptz', '$12::integer')}
        OR coalesce(target._attempt_owner_run_id, target._run_id) IS DISTINCT FROM $10::text
      )
    )`;
    const sameOwnerTerminalEpochSql = sameOwnerTerminalAttemptEpochSql(
      'target',
      '$10::text',
      '$12',
    );
    const { rows } = await client.query<{
      updated: number;
      matched_keys: string[];
      fenced_keys: string[];
      conflict_keys: string[];
    }>(
      `WITH updates AS (
         SELECT key_values._key,
                input_index_values.input_index,
                data_values.data_patch,
                cell_meta_values.cell_meta_patch
         FROM unnest($1::text[]) WITH ORDINALITY AS key_values(_key, ord)
         JOIN unnest($2::bigint[]) WITH ORDINALITY AS input_index_values(input_index, ord)
           ON input_index_values.ord = key_values.ord
         JOIN unnest($3::jsonb[]) WITH ORDINALITY AS data_values(data_patch, ord)
           ON data_values.ord = key_values.ord
         JOIN unnest($4::jsonb[]) WITH ORDINALITY AS cell_meta_values(cell_meta_patch, ord)
           ON cell_meta_values.ord = key_values.ord
       ),
       matched_updates AS (
         SELECT target._key AS matched_key,
                target._status AS prev_status,
                target._cell_meta AS prev_cell_meta,
                updates.input_index,
                updates.data_patch,
                updates.cell_meta_patch
           FROM updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates._key
       ),
       same_version_conflicts AS (
         SELECT updates.matched_key AS _key
           FROM matched_updates AS updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates.matched_key
          WHERE $13::bigint IS NOT NULL
            AND target._write_version = $13::bigint
            AND target._run_id = $5::text
            AND target._status IN ('enriched', 'failed')
            AND (
              EXISTS (
                SELECT 1
                  FROM unnest($8::text[]) AS field_values(field)
                 WHERE target._cell_meta -> field_values.field ->> 'runId' = $5::text
              )
              OR (
                coalesce(target._attempt_owner_run_id, target._run_id) = $10::text
                AND COALESCE(target._attempt_seq, 0) = $12::integer
              )
            )
            AND (
              target._status <> 'enriched'
              OR (${targetChangedPatchedCellSql})
            )
       ),
       applied_rows AS (
         UPDATE ${sheetTable(session)} AS target
         SET _status = 'enriched',
             _run_id = $5::text,
             _writer_run_id = $5::text,
             _write_version = COALESCE($13::bigint, target._write_version),
             _error = NULL,
             _attempt_id = $9::text,
             _attempt_owner_run_id = $10::text,
             _attempt_expires_at = $11::timestamptz,
             _attempt_seq = $12::integer,
             _updated_at = now(),
             _version = ${nextRuntimeSheetVersionExpression(session)},
             _cell_meta = ${mergeRuntimeCellMetaPatchSql('target._cell_meta', 'updates.cell_meta_patch')}${input.physicalUpdateSetSql}
         FROM matched_updates AS updates
         WHERE target._key = updates.matched_key
                   AND NOT EXISTS (SELECT 1 FROM same_version_conflicts)
		           AND ($13::bigint IS NULL OR target._write_version = $13::bigint)
                   AND ($13::bigint IS NOT NULL OR target._status <> 'enriched')
		           AND NOT (${newerTerminalRuntimeSheetRowSql('target', '$11::timestamptz', '$12::integer', '$9::text', '$10::text')})
           AND NOT (${sameOwnerTerminalEpochSql})
           AND (
             ($9::text IS NULL AND target._run_id = $5::text)
             OR ($9::text IS NOT NULL AND ${writableNonTerminalAttemptSql})
             OR ($9::text IS NOT NULL AND ${activeRuntimeSheetAttemptFenceSql('target', '$9::text', '$10::text', '$11::timestamptz', '$12::integer')})
		             OR ($9::text IS NOT NULL AND ${writableEnrichedAttemptSql})
           )
           AND (
             target._status <> 'enriched'
             OR target._run_id IS DISTINCT FROM $5::text
             OR (${targetChangedPatchedCellSql})
             OR EXISTS (
               SELECT 1
                 FROM unnest($8::text[]) AS field_values(field)
                WHERE coalesce(target._cell_meta -> field_values.field ->> 'status', '') <> 'completed'
             )
           )
         RETURNING target._key, updates.prev_status, updates.prev_cell_meta
       ),
       applied_count AS (
         SELECT count(*)::bigint AS c,
                count(*) FILTER (WHERE prev_status = 'failed')::bigint AS from_failed,
                count(*) FILTER (WHERE prev_status = 'running')::bigint AS from_running,
                count(*) FILTER (WHERE prev_status <> 'enriched')::bigint AS newly_completed
           FROM applied_rows
       ),
       fenced_rows AS (
         SELECT updates.matched_key AS _key
           FROM matched_updates AS updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates.matched_key
          WHERE NOT EXISTS (
                  SELECT 1 FROM applied_rows WHERE applied_rows._key = updates.matched_key
                )
            AND (
	              ($13::bigint IS NOT NULL AND target._write_version <> $13::bigint)
	              OR ($13::bigint IS NULL AND target._status = 'enriched')
	              OR
	              (${newerTerminalRuntimeSheetRowSql('target', '$11::timestamptz', '$12::integer', '$9::text', '$10::text')})
              OR NOT (
                ($9::text IS NULL AND target._run_id = $5::text)
                OR ($9::text IS NOT NULL AND ${writableNonTerminalAttemptSql})
                OR ($9::text IS NOT NULL AND ${activeRuntimeSheetAttemptFenceSql('target', '$9::text', '$10::text', '$11::timestamptz', '$12::integer')})
                OR ($9::text IS NOT NULL AND ${writableEnrichedAttemptSql})
              )
            )
       ),
       summary_counts AS (
         SELECT newly_completed,
                from_failed,
                from_running,
                GREATEST(newly_completed - from_failed - from_running, 0)::bigint AS from_queued
           FROM applied_count
       ),
       summary_delta AS (
         INSERT INTO ${summaryTable(session)} AS target (
           play_name,
           table_namespace,
           total,
           queued,
           running,
           completed,
           failed
         )
         SELECT $6::text,
                $7::text,
                0,
                (-from_queued)::int,
                (-from_running)::int,
                newly_completed::int,
                (-from_failed)::int
           FROM summary_counts
          WHERE newly_completed > 0 OR from_failed > 0
         ON CONFLICT (play_name, table_namespace) DO UPDATE SET
           queued = GREATEST(target.queued + EXCLUDED.queued, 0),
           running = GREATEST(target.running + EXCLUDED.running, 0),
           completed = GREATEST(target.completed + EXCLUDED.completed, 0),
           failed = GREATEST(target.failed + EXCLUDED.failed, 0),
           total = ${runtimeSummaryTotalSql({
             currentTotal: 'target.total',
             totalDelta: 'EXCLUDED.total',
             queued: 'GREATEST(target.queued + EXCLUDED.queued, 0)',
             running: 'GREATEST(target.running + EXCLUDED.running, 0)',
             completed: 'GREATEST(target.completed + EXCLUDED.completed, 0)',
             failed: 'GREATEST(target.failed + EXCLUDED.failed, 0)',
           })},
           _updated_at = now()
         RETURNING 1
       ),
       completed_cell_delta AS (
         SELECT field_values.field, count(*)::bigint AS c
           FROM applied_rows
           JOIN unnest($8::text[]) AS field_values(field)
             ON applied_rows.prev_status <> 'enriched'
             OR coalesce(applied_rows.prev_cell_meta -> field_values.field ->> 'status', '') <> 'completed'
          GROUP BY field_values.field
       ),
       prev_failed_cells AS (
         SELECT field_values.field, count(*)::bigint AS c
           FROM applied_rows
           JOIN unnest($8::text[]) AS field_values(field)
             ON coalesce(applied_rows.prev_cell_meta -> field_values.field ->> 'status', '') = 'failed'
          GROUP BY field_values.field
       ),
       column_delta AS (
         INSERT INTO ${columnSummaryTable(session)} AS target (
           play_name,
           table_namespace,
           field,
           completed,
           failed
         )
         SELECT $6::text,
                $7::text,
                coalesce(completed_cell_delta.field, prev_failed_cells.field),
                coalesce(completed_cell_delta.c, 0)::int,
                (-coalesce(prev_failed_cells.c, 0))::int
           FROM completed_cell_delta
           FULL JOIN prev_failed_cells
             ON prev_failed_cells.field = completed_cell_delta.field
          WHERE coalesce(completed_cell_delta.c, 0) > 0
             OR coalesce(prev_failed_cells.c, 0) > 0
         ON CONFLICT (play_name, table_namespace, field) DO UPDATE SET
           completed = GREATEST(target.completed + EXCLUDED.completed, 0),
           failed = GREATEST(target.failed + EXCLUDED.failed, 0),
           _updated_at = now()
         RETURNING 1
       )
       SELECT
         (SELECT count(*)::int FROM applied_rows) AS updated,
         coalesce((SELECT array_agg(matched_key) FROM matched_updates), '{}'::text[]) AS matched_keys,
         coalesce((SELECT array_agg(_key) FROM fenced_rows), '{}'::text[]) AS fenced_keys,
         coalesce((SELECT array_agg(_key) FROM same_version_conflicts), '{}'::text[]) AS conflict_keys,
         (SELECT count(*)::int FROM summary_delta) AS summary_delta_count,
         (SELECT count(*)::int FROM column_delta) AS column_delta_count`,
      [
        chunkKeys,
        chunkInputIndexes,
        chunkDataPatches,
        chunkCellMetaPatches,
        input.runId,
        input.normalizedPlayName,
        input.normalizedTableNamespace,
        [...new Set(input.outputFields)],
        input.attemptId,
        input.attemptOwnerRunId,
        input.attemptExpiresAt,
        input.attemptSeq,
        input.writeVersion,
      ],
    );
    const matchedKeys = new Set(rows[0]?.matched_keys ?? []);
    const chunkFencedKeys = rows[0]?.fenced_keys ?? [];
    const chunkConflictKeys = rows[0]?.conflict_keys ?? [];
    // An exact-version replay that already produced the same terminal value is
    // accepted idempotently even when SQL has no physical mutation to apply.
    // Stale versions remain excluded through fenced_keys.
    const appliedCount = Number(rows[0]?.updated ?? 0);
    updated += appliedCount;
    fencedKeys.push(...chunkFencedKeys);
    conflictKeys.push(...chunkConflictKeys);

    if (chunkConflictKeys.length > 0) {
      continue;
    }

    if (matchedKeys.size === chunk.length) {
      continue;
    }

    const repairChunk = chunk.filter(
      (row) => !matchedKeys.has(row.key) && row.input_index !== null,
    );
    if (repairChunk.length === 0) {
      continue;
    }
    const repaired = await completeRuntimeMapRowChunksWithInputIndexRepair(
      client,
      session,
      { ...input, chunks: [repairChunk] },
    );
    updated += repaired.updated;
    fencedKeys.push(...repaired.fencedKeys);
    conflictKeys.push(...repaired.conflictKeys);
  }
  return { updated, fencedKeys, conflictKeys };
}

async function completeRuntimeMapRowChunksWithInputIndexRepair(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: CompleteRuntimeMapRowChunksInput,
): Promise<RuntimeMapRowsMutationResult> {
  let updated = 0;
  const fencedKeys: string[] = [];
  const conflictKeys: string[] = [];
  for (const chunk of input.chunks) {
    const chunkKeys = chunk.map((row) => row.key);
    const chunkInputIndexes = chunk.map((row) => row.input_index);
    const chunkDataPatches = chunk.map((row) => row.data_patch_json);
    const chunkCellMetaPatches = chunk.map((row) => row.cell_meta_patch_json);
    const targetChangedPatchedCellSql = changedPatchedCellSql(
      'target',
      'updates.data_patch',
      input.physicalColumnProjections,
    );
    const writableEnrichedAttemptSql = writableEnrichedRuntimeSheetAttemptSql(
      'target',
      '$10::text',
      '$11::timestamptz',
      '$12',
    );
    const sameOwnerTerminalEpochSql = sameOwnerTerminalAttemptEpochSql(
      'target',
      '$10::text',
      '$12',
    );
    const { rows } = await client.query<{
      _key?: string;
      fenced_keys?: string[];
      conflict_keys?: string[];
      updated?: number;
    }>(
      `WITH updates AS (
         SELECT key_values._key,
                input_index_values.input_index,
                data_values.data_patch,
                cell_meta_values.cell_meta_patch
         FROM unnest($1::text[]) WITH ORDINALITY AS key_values(_key, ord)
         JOIN unnest($2::bigint[]) WITH ORDINALITY AS input_index_values(input_index, ord)
           ON input_index_values.ord = key_values.ord
         JOIN unnest($3::jsonb[]) WITH ORDINALITY AS data_values(data_patch, ord)
           ON data_values.ord = key_values.ord
         JOIN unnest($4::jsonb[]) WITH ORDINALITY AS cell_meta_values(cell_meta_patch, ord)
           ON cell_meta_values.ord = key_values.ord
       ),
       matched_updates AS (
         SELECT DISTINCT ON (target._key)
                target._key AS matched_key,
                updates._key AS submitted_key,
                updates.data_patch,
                updates.cell_meta_patch
           FROM updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates._key
             OR (
               updates.input_index IS NOT NULL
               AND target._run_id = $5::text
               AND target._input_index = updates.input_index
             )
          ORDER BY target._key, (target._key = updates._key) DESC
       ),
       same_version_conflicts AS (
         SELECT updates.submitted_key AS _key
           FROM matched_updates AS updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates.matched_key
          WHERE $13::bigint IS NOT NULL
            AND target._write_version = $13::bigint
            AND target._run_id = $5::text
            AND target._status IN ('enriched', 'failed')
            AND (
              EXISTS (
                SELECT 1
                  FROM unnest($8::text[]) AS field_values(field)
                 WHERE target._cell_meta -> field_values.field ->> 'runId' = $5::text
              )
              OR (
                coalesce(target._attempt_owner_run_id, target._run_id) = $10::text
                AND COALESCE(target._attempt_seq, 0) = $12::integer
              )
            )
            AND (
              target._status <> 'enriched'
              OR (${targetChangedPatchedCellSql})
            )
       ),
       applied_rows AS (
         UPDATE ${sheetTable(session)} AS target
         SET _status = 'enriched',
             _run_id = $5::text,
             _writer_run_id = $5::text,
             _write_version = COALESCE($13::bigint, target._write_version),
             _error = NULL,
             _attempt_id = $9::text,
             _attempt_owner_run_id = $10::text,
             _attempt_expires_at = $11::timestamptz,
             _attempt_seq = $12::integer,
             _updated_at = now(),
             _version = ${nextRuntimeSheetVersionExpression(session)},
             _cell_meta = ${mergeRuntimeCellMetaPatchSql('target._cell_meta', 'updates.cell_meta_patch')}${input.physicalUpdateSetSql}
         FROM matched_updates AS updates, ${sheetTable(session)} AS prev
         WHERE target._key = updates.matched_key
           AND prev._key = target._key
           AND NOT EXISTS (SELECT 1 FROM same_version_conflicts)
           AND ($13::bigint IS NULL OR target._write_version = $13::bigint)
           AND ($13::bigint IS NOT NULL OR target._status <> 'enriched')
		           AND NOT (${newerTerminalRuntimeSheetRowSql('target', '$11::timestamptz', '$12::integer', '$9::text', '$10::text')})
           AND NOT (${sameOwnerTerminalEpochSql})
           AND (
             ($9::text IS NULL AND target._run_id = $5::text)
             OR ($9::text IS NOT NULL AND ${activeRuntimeSheetAttemptFenceSql('target', '$9::text', '$10::text', '$11::timestamptz', '$12::integer')})
		             OR ($9::text IS NOT NULL AND ${writableEnrichedAttemptSql})
           )
           AND (
             target._status <> 'enriched'
             OR target._run_id IS DISTINCT FROM $5::text
             OR (${targetChangedPatchedCellSql})
             OR EXISTS (
               SELECT 1
                 FROM unnest($8::text[]) AS field_values(field)
                WHERE coalesce(target._cell_meta -> field_values.field ->> 'status', '') <> 'completed'
             )
           )
         RETURNING target._key, prev._status AS prev_status, prev._cell_meta AS prev_cell_meta
       ),
       applied_count AS (
         SELECT count(*)::bigint AS c,
                count(*) FILTER (WHERE prev_status = 'failed')::bigint AS from_failed,
                count(*) FILTER (WHERE prev_status = 'running')::bigint AS from_running,
                count(*) FILTER (WHERE prev_status <> 'enriched')::bigint AS newly_completed
           FROM applied_rows
       ),
       fenced_rows AS (
         SELECT updates.submitted_key AS _key
           FROM matched_updates AS updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates.matched_key
          WHERE NOT EXISTS (
                  SELECT 1 FROM applied_rows WHERE applied_rows._key = updates.matched_key
            )
            AND (
              ($13::bigint IS NOT NULL AND target._write_version <> $13::bigint)
              OR ($13::bigint IS NULL AND target._status = 'enriched')
              OR (${newerTerminalRuntimeSheetRowSql('target', '$11::timestamptz', '$12::integer', '$9::text', '$10::text')})
              OR NOT (
                ($9::text IS NULL AND target._run_id = $5::text)
                OR ($9::text IS NOT NULL AND ${activeRuntimeSheetAttemptFenceSql('target', '$9::text', '$10::text', '$11::timestamptz', '$12::integer')})
                OR ($9::text IS NOT NULL AND ${writableEnrichedAttemptSql})
              )
            )
       ),
       summary_counts AS (
         SELECT newly_completed,
                from_failed,
                from_running,
                GREATEST(newly_completed - from_failed - from_running, 0)::bigint AS from_queued
           FROM applied_count
       ),
       summary_delta AS (
         INSERT INTO ${summaryTable(session)} AS target (
           play_name,
           table_namespace,
           total,
           queued,
           running,
           completed,
           failed
         )
         SELECT $6::text,
                $7::text,
                0,
                (-from_queued)::int,
                (-from_running)::int,
                newly_completed::int,
                (-from_failed)::int
           FROM summary_counts
          WHERE newly_completed > 0 OR from_failed > 0
         ON CONFLICT (play_name, table_namespace) DO UPDATE SET
           queued = GREATEST(target.queued + EXCLUDED.queued, 0),
           running = GREATEST(target.running + EXCLUDED.running, 0),
           completed = GREATEST(target.completed + EXCLUDED.completed, 0),
           failed = GREATEST(target.failed + EXCLUDED.failed, 0),
           total = ${runtimeSummaryTotalSql({
             currentTotal: 'target.total',
             totalDelta: 'EXCLUDED.total',
             queued: 'GREATEST(target.queued + EXCLUDED.queued, 0)',
             running: 'GREATEST(target.running + EXCLUDED.running, 0)',
             completed: 'GREATEST(target.completed + EXCLUDED.completed, 0)',
             failed: 'GREATEST(target.failed + EXCLUDED.failed, 0)',
           })},
           _updated_at = now()
         RETURNING 1
       ),
       completed_cell_delta AS (
         SELECT field_values.field, count(*)::bigint AS c
           FROM applied_rows
           JOIN unnest($8::text[]) AS field_values(field)
             ON applied_rows.prev_status <> 'enriched'
             OR coalesce(applied_rows.prev_cell_meta -> field_values.field ->> 'status', '') <> 'completed'
          GROUP BY field_values.field
       ),
       prev_failed_cells AS (
         SELECT field_values.field, count(*)::bigint AS c
           FROM applied_rows
           JOIN unnest($8::text[]) AS field_values(field)
             ON coalesce(applied_rows.prev_cell_meta -> field_values.field ->> 'status', '') = 'failed'
          GROUP BY field_values.field
       ),
       column_delta AS (
         INSERT INTO ${columnSummaryTable(session)} AS target (
           play_name,
           table_namespace,
           field,
           completed,
           failed
         )
         SELECT $6::text,
                $7::text,
                coalesce(completed_cell_delta.field, prev_failed_cells.field),
                coalesce(completed_cell_delta.c, 0)::int,
                (-coalesce(prev_failed_cells.c, 0))::int
           FROM completed_cell_delta
           FULL JOIN prev_failed_cells
             ON prev_failed_cells.field = completed_cell_delta.field
          WHERE coalesce(completed_cell_delta.c, 0) > 0
             OR coalesce(prev_failed_cells.c, 0) > 0
         ON CONFLICT (play_name, table_namespace, field) DO UPDATE SET
           completed = GREATEST(target.completed + EXCLUDED.completed, 0),
           failed = GREATEST(target.failed + EXCLUDED.failed, 0),
           _updated_at = now()
         RETURNING 1
       )
       SELECT
         (SELECT count(*)::int FROM applied_rows) AS updated,
         coalesce((SELECT array_agg(_key) FROM fenced_rows), '{}'::text[]) AS fenced_keys,
         coalesce((SELECT array_agg(_key) FROM same_version_conflicts), '{}'::text[]) AS conflict_keys`,
      [
        chunkKeys,
        chunkInputIndexes,
        chunkDataPatches,
        chunkCellMetaPatches,
        input.runId,
        input.normalizedPlayName,
        input.normalizedTableNamespace,
        [...new Set(input.outputFields)],
        input.attemptId,
        input.attemptOwnerRunId,
        input.attemptExpiresAt,
        input.attemptSeq,
        input.writeVersion,
      ],
    );
    updated += Number(rows[0]?.updated ?? 0);
    fencedKeys.push(...(rows[0]?.fenced_keys ?? []));
    conflictKeys.push(...(rows[0]?.conflict_keys ?? []));
  }
  return { updated, fencedKeys, conflictKeys };
}

async function insertMissingCompletedMapRowChunks(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: CompleteRuntimeMapRowChunksInput,
): Promise<{ inserted: number }> {
  let inserted = 0;
  const physicalInsertColumnsSql =
    input.physicalColumnProjections.length > 0
      ? `, ${input.physicalColumnProjections
          .map((column) => quoteIdentifier(column.sqlName))
          .join(', ')}`
      : '';
  const physicalInsertValuesSql =
    input.physicalColumnProjections.length > 0
      ? `, ${input.physicalColumnProjections
          .map(
            (column) =>
              `missing_rows.data_patch -> ${quoteLiteral(column.fieldName)}`,
          )
          .join(', ')}`
      : '';

  for (const chunk of input.chunks) {
    const chunkKeys = chunk.map((row) => row.key);
    const chunkInputIndexes = chunk.map((row) => row.input_index);
    const chunkDataPatches = chunk.map((row) => row.data_patch_json);
    const chunkCellMetaPatches = chunk.map((row) => row.cell_meta_patch_json);
    const { rows } = await client.query<{ inserted: number }>(
      `WITH input_rows AS (
         SELECT key_values._key,
                input_index_values.input_index,
                data_values.data_patch,
                cell_meta_values.cell_meta_patch
         FROM unnest($1::text[]) WITH ORDINALITY AS key_values(_key, ord)
         JOIN unnest($2::bigint[]) WITH ORDINALITY AS input_index_values(input_index, ord)
           ON input_index_values.ord = key_values.ord
         JOIN unnest($3::jsonb[]) WITH ORDINALITY AS data_values(data_patch, ord)
           ON data_values.ord = key_values.ord
         JOIN unnest($4::jsonb[]) WITH ORDINALITY AS cell_meta_values(cell_meta_patch, ord)
           ON cell_meta_values.ord = key_values.ord
       ),
       missing_rows AS (
         SELECT input_rows.*
           FROM input_rows
          WHERE NOT EXISTS (
            SELECT 1
              FROM ${sheetTable(session)} AS target
             WHERE target._key = input_rows._key
                OR (
                  input_rows.input_index IS NOT NULL
                  AND target._run_id = $5::text
                  AND target._input_index = input_rows.input_index
                )
          )
       ),
       inserted_rows AS (
         INSERT INTO ${sheetTable(session)} (
           _key,
           _status,
           _run_id,
           _input_index,
           _attempt_id,
           _attempt_owner_run_id,
           _attempt_expires_at,
           _attempt_seq,
           _cell_meta${physicalInsertColumnsSql}
         )
         SELECT _key,
                'enriched',
                $5::text,
                input_index,
                $8::text,
                $9::text,
                $10::timestamptz,
                $11::integer,
                cell_meta_patch${physicalInsertValuesSql}
           FROM missing_rows
         ON CONFLICT (_key) DO NOTHING
         RETURNING _key
       ),
       inserted_count AS (
         SELECT count(*)::bigint AS c FROM inserted_rows
       ),
       summary_delta AS (
         INSERT INTO ${summaryTable(session)} AS target (
           play_name,
           table_namespace,
           total,
           queued,
           running,
           completed,
           failed
         )
         SELECT $6::text,
                $7::text,
                c::int,
                0,
                0,
                c::int,
                0
           FROM inserted_count
          WHERE c > 0
         ON CONFLICT (play_name, table_namespace) DO UPDATE SET
           total = ${runtimeSummaryTotalSql({
             currentTotal: 'target.total',
             totalDelta: 'EXCLUDED.total',
             queued: 'target.queued',
             running: 'target.running',
             completed: 'target.completed + EXCLUDED.completed',
             failed: 'target.failed',
           })},
           completed = target.completed + EXCLUDED.completed,
           _updated_at = now()
         RETURNING 1
       ),
       completed_cell_delta AS (
         SELECT field_values.field, (SELECT c FROM inserted_count) AS c
           FROM unnest($12::text[]) AS field_values(field)
          WHERE (SELECT c FROM inserted_count) > 0
       ),
       column_delta AS (
         INSERT INTO ${columnSummaryTable(session)} AS target (
           play_name,
           table_namespace,
           field,
           completed,
           failed
         )
         SELECT $6::text,
                $7::text,
                field,
                c::int,
                0
           FROM completed_cell_delta
         ON CONFLICT (play_name, table_namespace, field) DO UPDATE SET
           completed = target.completed + EXCLUDED.completed,
           _updated_at = now()
         RETURNING 1
       )
       SELECT c::int AS inserted FROM inserted_count`,
      [
        chunkKeys,
        chunkInputIndexes,
        chunkDataPatches,
        chunkCellMetaPatches,
        input.runId,
        input.normalizedPlayName,
        input.normalizedTableNamespace,
        input.attemptId,
        input.attemptOwnerRunId,
        input.attemptExpiresAt,
        input.attemptSeq,
        [...new Set(input.outputFields)],
      ],
    );
    inserted += Number(rows[0]?.inserted ?? 0);
  }
  return { inserted };
}

/**
 * Mark map rows FAILED in the per-run scoped Postgres sheet table by key.
 *
 * Row failure isolation contract: one row's tool/provider error must not abort
 * sibling rows or the run. The failed row keeps every field value that
 * completed before the error (written through the same physical-column patch
 * as completed rows), flips `_status` to 'failed', records the row error in
 * `_error`, and merges the per-cell failure into `_cell_meta`. The next run's
 * `startRuntimeSheetDataset` returns failed rows as pending, so they re-execute
 * while their already-completed cells replay free via cell receipts.
 */
async function failRuntimeMapRowChunks(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    chunks: RuntimePreparedFailedRow[][];
    physicalUpdateSetSql: string;
    physicalColumnProjections: PhysicalSheetColumnProjection[];
    forceTerminal?: boolean;
    runId: string;
    attemptId: string | null;
    attemptOwnerRunId: string | null;
    attemptExpiresAt: string | null;
    attemptSeq: number;
    writeVersion: number | null;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
  },
): Promise<RuntimeMapRowsMutationResult> {
  let updated = 0;
  const fencedKeys: string[] = [];
  const conflictKeys: string[] = [];
  for (const chunk of input.chunks) {
    const chunkKeys = chunk.map((row) => row.key);
    const chunkInputIndexes = chunk.map((row) => row.input_index);
    const chunkDataPatches = chunk.map((row) => row.data_patch_json);
    const chunkCellMetaPatches = chunk.map((row) => row.cell_meta_patch_json);
    const chunkErrors = chunk.map((row) => row.error);
    const targetChangedPatchedCellSql = changedPatchedCellSql(
      'target',
      'updates.data_patch',
      input.physicalColumnProjections,
    );
    // Per-field failed-cell counts for the column summary, computed from the
    // cell meta patches (a failed row records exactly which cell failed).
    const failedCellCounts = new Map<string, number>();
    for (const row of chunk) {
      for (const [field, meta] of Object.entries(row.cell_meta_patch)) {
        if (
          meta &&
          typeof meta === 'object' &&
          (meta as { status?: unknown }).status === 'failed'
        ) {
          failedCellCounts.set(field, (failedCellCounts.get(field) ?? 0) + 1);
        }
      }
    }
    const failedCellFields = [...failedCellCounts.keys()];
    const failedCellTotals = failedCellFields.map(
      (field) => failedCellCounts.get(field) ?? 0,
    );
    const { rows } = await client.query<{
      updated?: number;
      fenced_keys?: string[];
      conflict_keys?: string[];
    }>(
      `WITH updates AS (
         SELECT key_values._key,
                input_index_values.input_index,
                data_values.data_patch,
                cell_meta_values.cell_meta_patch,
                error_values.error
         FROM unnest($1::text[]) WITH ORDINALITY AS key_values(_key, ord)
         JOIN unnest($2::bigint[]) WITH ORDINALITY AS input_index_values(input_index, ord)
           ON input_index_values.ord = key_values.ord
         JOIN unnest($3::jsonb[]) WITH ORDINALITY AS data_values(data_patch, ord)
           ON data_values.ord = key_values.ord
         JOIN unnest($4::jsonb[]) WITH ORDINALITY AS cell_meta_values(cell_meta_patch, ord)
           ON cell_meta_values.ord = key_values.ord
         JOIN unnest($5::text[]) WITH ORDINALITY AS error_values(error, ord)
           ON error_values.ord = key_values.ord
       ),
       matched_updates AS (
         SELECT DISTINCT ON (target._key)
                target._key AS matched_key,
                updates.data_patch,
                updates.cell_meta_patch,
                updates.error
           FROM updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates._key
             OR (
               updates.input_index IS NOT NULL
               AND target._run_id = $6::text
               AND target._input_index = updates.input_index
             )
          ORDER BY target._key, (target._key = updates._key) DESC
       ),
       same_version_conflicts AS (
         SELECT updates.matched_key AS _key
           FROM matched_updates AS updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates.matched_key
          WHERE $16::bigint IS NOT NULL
            AND target._write_version = $16::bigint
            AND target._run_id = $6::text
            AND target._status IN ('enriched', 'failed')
            AND (
              EXISTS (
                SELECT 1
                  FROM unnest($9::text[]) AS field_values(field)
                 WHERE target._cell_meta -> field_values.field ->> 'runId' = $6::text
              )
              OR (
                coalesce(target._attempt_owner_run_id, target._run_id) = $12::text
                AND COALESCE(target._attempt_seq, 0) = $14::integer
              )
            )
            AND (
              target._status <> 'failed'
              OR target._error IS DISTINCT FROM updates.error
              OR (${targetChangedPatchedCellSql})
            )
       ),
       applied_rows AS (
         UPDATE ${sheetTable(session)} AS target
         SET _status = 'failed',
             _run_id = $6::text,
             _writer_run_id = $6::text,
             _write_version = COALESCE($16::bigint, target._write_version),
             _error = updates.error,
             _attempt_id = $11::text,
             _attempt_owner_run_id = $12::text,
             _attempt_expires_at = $13::timestamptz,
             _attempt_seq = $14::integer,
             _updated_at = now(),
             _version = ${nextRuntimeSheetVersionExpression(session)},
             _cell_meta = ${mergeRuntimeCellMetaPatchSql('target._cell_meta', 'updates.cell_meta_patch')}${input.physicalUpdateSetSql}
         FROM matched_updates AS updates, ${sheetTable(session)} AS prev
         WHERE target._key = updates.matched_key
           AND NOT EXISTS (SELECT 1 FROM same_version_conflicts)
           AND ($16::bigint IS NULL OR target._write_version = $16::bigint)
           AND prev._key = target._key
           AND ($15::boolean OR target._status <> 'enriched')
		           AND NOT (${newerTerminalRuntimeSheetRowSql('target', '$13::timestamptz', '$14::integer', '$11::text', '$12::text')})
           AND (
             ($11::text IS NULL AND target._run_id = $6::text)
             OR ($11::text IS NOT NULL AND ${activeRuntimeSheetAttemptFenceSql('target', '$11::text', '$12::text', '$13::timestamptz', '$14::integer')})
	           )
           AND NOT (
             $16::bigint IS NOT NULL
             AND target._write_version = $16::bigint
             AND target._run_id = $6::text
             AND target._status = 'failed'
             AND target._error IS NOT DISTINCT FROM updates.error
             AND NOT (${targetChangedPatchedCellSql})
           )
         RETURNING target._key, prev._status AS prev_status
       ),
       applied_count AS (
         SELECT count(*)::bigint AS c,
                count(*) FILTER (WHERE prev_status = 'failed')::bigint AS already_failed,
                count(*) FILTER (WHERE prev_status = 'enriched')::bigint AS from_completed,
                count(*) FILTER (WHERE prev_status = 'running')::bigint AS from_running
           FROM applied_rows
       ),
       fenced_rows AS (
         SELECT updates.matched_key AS _key
           FROM matched_updates AS updates
           JOIN ${sheetTable(session)} AS target
             ON target._key = updates.matched_key
          WHERE NOT EXISTS (
                  SELECT 1 FROM applied_rows WHERE applied_rows._key = updates.matched_key
                )
            AND (
              ($16::bigint IS NOT NULL AND target._write_version <> $16::bigint)
              OR
              (NOT $15::boolean AND target._status = 'enriched')
              OR (${newerTerminalRuntimeSheetRowSql('target', '$13::timestamptz', '$14::integer', '$11::text', '$12::text')})
              OR NOT (
                ($11::text IS NULL AND target._run_id = $6::text)
                OR ($11::text IS NOT NULL AND ${activeRuntimeSheetAttemptFenceSql('target', '$11::text', '$12::text', '$13::timestamptz', '$14::integer')})
              )
            )
       ),
       summary_counts AS (
         SELECT (c - already_failed)::bigint AS newly_failed,
                from_completed,
                from_running,
                GREATEST(c - already_failed - from_completed - from_running, 0)::bigint AS from_queued
           FROM applied_count
       ),
       summary_delta AS (
         INSERT INTO ${summaryTable(session)} AS target (
           play_name,
           table_namespace,
           total,
           queued,
           running,
           completed,
           failed
         )
         SELECT $7::text,
                $8::text,
                0,
                (-from_queued)::int,
                (-from_running)::int,
                (-from_completed)::int,
                newly_failed::int
           FROM summary_counts
          WHERE newly_failed > 0
         ON CONFLICT (play_name, table_namespace) DO UPDATE SET
           queued = GREATEST(target.queued + EXCLUDED.queued, 0),
           running = GREATEST(target.running + EXCLUDED.running, 0),
           completed = GREATEST(target.completed + EXCLUDED.completed, 0),
           failed = GREATEST(target.failed + EXCLUDED.failed, 0),
           total = ${runtimeSummaryTotalSql({
             currentTotal: 'target.total',
             totalDelta: 'EXCLUDED.total',
             queued: 'GREATEST(target.queued + EXCLUDED.queued, 0)',
             running: 'GREATEST(target.running + EXCLUDED.running, 0)',
             completed: 'GREATEST(target.completed + EXCLUDED.completed, 0)',
             failed: 'GREATEST(target.failed + EXCLUDED.failed, 0)',
           })},
           _updated_at = now()
         RETURNING 1
       ),
       column_delta AS (
         INSERT INTO ${columnSummaryTable(session)} AS target (
           play_name,
           table_namespace,
           field,
           failed
         )
         SELECT $7::text, $8::text, field_values.field, count_values.c
           FROM unnest($9::text[]) WITH ORDINALITY AS field_values(field, ord)
           JOIN unnest($10::int[]) WITH ORDINALITY AS count_values(c, ord)
             ON count_values.ord = field_values.ord
          WHERE EXISTS (SELECT 1 FROM applied_rows)
         ON CONFLICT (play_name, table_namespace, field) DO UPDATE SET
           failed = GREATEST(target.failed + EXCLUDED.failed, 0),
           _updated_at = now()
         RETURNING 1
       )
       SELECT
         (SELECT count(*)::int FROM applied_rows) AS updated,
         coalesce((SELECT array_agg(_key) FROM fenced_rows), '{}'::text[]) AS fenced_keys,
         coalesce((SELECT array_agg(_key) FROM same_version_conflicts), '{}'::text[]) AS conflict_keys`,
      [
        chunkKeys,
        chunkInputIndexes,
        chunkDataPatches,
        chunkCellMetaPatches,
        chunkErrors,
        input.runId,
        input.normalizedPlayName,
        input.normalizedTableNamespace,
        failedCellFields,
        failedCellTotals,
        input.attemptId,
        input.attemptOwnerRunId,
        input.attemptExpiresAt,
        input.attemptSeq,
        input.forceTerminal === true,
        input.writeVersion,
      ],
    );
    updated += Number(rows[0]?.updated ?? 0);
    fencedKeys.push(...(rows[0]?.fenced_keys ?? []));
    conflictKeys.push(...(rows[0]?.conflict_keys ?? []));
  }
  return { updated, fencedKeys, conflictKeys };
}

/**
 * Mark map rows terminal in the per-run scoped Postgres sheet table by key.
 * Writes enriched output values into materialized physical columns, flips
 * _status to 'enriched', and advances the sheet cursor.
 *
 * Rows with `status: 'failed'` instead persist a row-isolated failure
 * (`_status='failed'`, `_error`, per-cell failure meta) while keeping the
 * field values that completed before the error — see
 * `failRuntimeMapRowChunks` for the recovery contract.
 *
 * Used by `persistCompletedMapRows` so map
 * completion writes go through the shared runtime storage plane and skip the
 * per-chunk Vercel hop.
 */
export async function completeRuntimeMapRows(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    sheetContract: PlaySheetContract;
    rows: RuntimeApiRowRecord[];
    outputFields?: string[];
    runId: string;
    attemptId?: string | null;
    attemptOwnerRunId?: string | null;
    attemptExpiresAt?: string | null;
    attemptSeq?: number | null;
    writeVersion?: number | null;
    forceFailedRows?: boolean;
  },
): Promise<RuntimeMapRowsWriteResult> {
  if (input.rows.length === 0) {
    return classifyRuntimeMapRowsWrite({
      submittedKeys: [],
      updated: 0,
      fencedKeys: [],
    });
  }
  if (context.dbSessionStrategy === 'gateway_only') {
    const keyedRows = input.rows.map((row) => {
      if (row.key) return row;
      const key = resolveMapRowOutcomeKey(
        row as unknown as Record<string, unknown>,
      );
      return key ? { ...row, key } : row;
    });
    const rows = prepareRuntimeSheetRowsForJsonTransport({
      rows: keyedRows,
      runId: input.runId,
      outputFields: input.outputFields ?? [],
    });
    const result = await postRuntimeApi<
      Partial<RuntimeMapRowsWriteResult> &
        Pick<RuntimeMapRowsWriteResult, 'updated' | 'fencedKeys'>
    >(context, {
      action: 'runtime_sheet_complete_map_rows',
      input: { ...input, rows },
    });
    const submittedKeys = keyedRows
      .map((row) => row.key)
      .filter((key): key is string => Boolean(key));
    const staleKeys =
      result.staleKeys ?? result.staleDroppedKeys ?? result.fencedKeys;
    return {
      ...classifyRuntimeMapRowsWrite({
        submittedKeys,
        updated: result.updated,
        fencedKeys: staleKeys,
      }),
      // Preserve the gateway's compatibility classification during rollout.
      fencedKeys: result.fencedKeys,
      ...(typeof result.runtimeTestPageTailHoldMs === 'number'
        ? { runtimeTestPageTailHoldMs: result.runtimeTestPageTailHoldMs }
        : {}),
    };
  }
  const sheetContract = augmentSheetContractWithDatasetFields({
    contract: input.sheetContract,
    rows: input.rows.map((row) => row.data),
    outputFields: input.outputFields,
  });
  const session = requireRuntimePostgresSession(
    await getRuntimeDbSession(
      {
        ...context,
        playName: context.playName,
        runId: context.runId ?? input.runId,
      },
      {
        tableNamespace: input.tableNamespace,
        logicalTable: 'sheet_rows',
        operations: ['rows.read', 'rows.upsert'],
        limits: {
          maxRows: runtimeDbSessionRowLimit(input.rows.length),
        },
        sheetContract,
      },
    ),
  );

  // Dedupe by key. Last write wins on payload, status, and error.
  const uniqueRows = new Map<string, RuntimeApiRowRecord>();
  for (const row of input.rows) {
    if (row.key) {
      uniqueRows.set(row.key, row);
      continue;
    }
    const resolvedKey = resolveMapRowOutcomeKey(
      row as unknown as Record<string, unknown>,
    );
    if (!resolvedKey) continue;
    uniqueRows.set(resolvedKey, { ...row, key: resolvedKey });
  }
  if (uniqueRows.size === 0) {
    return classifyRuntimeMapRowsWrite({
      submittedKeys: [],
      updated: 0,
      fencedKeys: [],
    });
  }

  const projections = physicalSheetColumnProjections(sheetContract);
  const physicalUpdateSetSql =
    projections.length > 0
      ? `,\n        ${projections
          .map((column) => {
            const quoted = quoteIdentifier(column.sqlName);
            const literal = quoteLiteral(column.fieldName);
            return `${quoted} = CASE
          WHEN coalesce(updates.data_patch, '{}'::jsonb) ? ${literal}
            THEN updates.data_patch -> ${literal}
          ELSE target.${quoted}
        END`;
          })
          .join(',\n        ')}`
      : '';

  const { completedRows, failedRows } = prepareRuntimeSheetRowTransitions({
    rows: uniqueRows.values(),
    runId: input.runId,
    outputFields: input.outputFields ?? [],
  });
  const chunks = chunkValues(completedRows, DIRECT_POSTGRES_BATCH_SIZE);
  const failedChunks = chunkValues(failedRows, DIRECT_POSTGRES_BATCH_SIZE);
  const needsTransaction = chunks.length + failedChunks.length > 1;
  const outputFields = [...new Set(input.outputFields ?? [])];
  const attemptId = input.attemptId?.trim() || null;
  const attemptOwnerRunId =
    input.attemptOwnerRunId?.trim() || (attemptId ? input.runId : null);
  const attemptExpiresAt = input.attemptExpiresAt?.trim() || null;
  const attemptSeq = normalizeRuntimeRunAttempt(input.attemptSeq);
  const writeVersion = input.writeVersion == null ? null : input.writeVersion;
  if (
    writeVersion !== null &&
    (!Number.isSafeInteger(writeVersion) || writeVersion <= 0)
  ) {
    throw new Error(
      'Runtime sheet writeVersion must be a positive safe integer.',
    );
  }

  return await withRuntimeSheetQueryClient(
    context,
    session,
    {
      playName: context.playName,
      tableNamespace: input.tableNamespace,
      sheetContract,
      transactional: needsTransaction,
      statementTimeoutMs: RUNTIME_SHEET_COMPLETION_STATEMENT_TIMEOUT_MS,
    },
    async (client) => {
      const completed =
        completedRows.length > 0
          ? await (async () => {
              const chunkInput = {
                chunks,
                physicalUpdateSetSql,
                physicalColumnProjections: projections,
                runId: input.runId,
                attemptId,
                attemptOwnerRunId,
                attemptExpiresAt,
                attemptSeq,
                writeVersion,
                normalizedPlayName: normalizePlayNameForSheet(session.playName),
                normalizedTableNamespace: normalizeTableNamespace(
                  input.tableNamespace,
                ),
                outputFields,
              };
              const updated = await completeRuntimeMapRowChunks(
                client,
                session,
                chunkInput,
              );
              const inserted = await insertMissingCompletedMapRowChunks(
                client,
                session,
                chunkInput,
              );
              return {
                updated: updated.updated + inserted.inserted,
                fencedKeys: updated.fencedKeys,
                conflictKeys: updated.conflictKeys,
              };
            })()
          : { updated: 0, fencedKeys: [], conflictKeys: [] };
      const failed =
        failedRows.length > 0
          ? await failRuntimeMapRowChunks(client, session, {
              chunks: failedChunks,
              physicalUpdateSetSql,
              physicalColumnProjections: projections,
              forceTerminal: input.forceFailedRows === true,
              runId: input.runId,
              attemptId,
              attemptOwnerRunId,
              attemptExpiresAt,
              attemptSeq,
              writeVersion,
              normalizedPlayName: normalizePlayNameForSheet(session.playName),
              normalizedTableNamespace: normalizeTableNamespace(
                input.tableNamespace,
              ),
            })
          : { updated: 0, fencedKeys: [], conflictKeys: [] };
      const conflictKeys = [...completed.conflictKeys, ...failed.conflictKeys];
      if (conflictKeys.length > 0) {
        throw new Error(
          `Runtime Sheet received the same write version with a different terminal payload for row(s): ${[
            ...new Set(conflictKeys),
          ]
            .slice(0, 10)
            .join(', ')}.`,
        );
      }
      return classifyRuntimeMapRowsWrite({
        submittedKeys: uniqueRows.keys(),
        updated: completed.updated + failed.updated,
        fencedKeys: [...completed.fencedKeys, ...failed.fencedKeys],
      });
    },
  );
}

/**
 * Freeze the transport-ready completion payload before a higher-level
 * capacity retry loop starts. In particular, completedAt must remain stable:
 * a retry is the same idempotent write, not a newly authored row transition.
 */
export function createCompleteRuntimeMapRowsOperation(
  context: RuntimeApiContext & { playName: string },
  input: Parameters<typeof completeRuntimeMapRows>[1],
): () => Promise<RuntimeMapRowsWriteResult> {
  if (context.dbSessionStrategy !== 'gateway_only') {
    return () => completeRuntimeMapRows(context, input);
  }
  const frozenInput = {
    ...input,
    rows: prepareRuntimeSheetRowsForJsonTransport({
      rows: input.rows,
      runId: input.runId,
      outputFields: input.outputFields ?? [],
    }),
  };
  return () => completeRuntimeMapRows(context, frozenInput);
}

export async function readRuntimeSheetDatasetRows(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    runId?: string | null;
    rowMode?: 'output' | 'all';
    limit: number;
    offset: number;
  },
): Promise<{ rows: Record<string, unknown>[]; limit: number; offset: number }> {
  if (context.dbSessionStrategy === 'gateway_only') {
    return await postRuntimeApi(context, {
      action: 'runtime_sheet_read_rows',
      input,
    });
  }
  const limit = Math.max(
    1,
    Math.min(DIRECT_POSTGRES_BATCH_SIZE, Math.floor(input.limit)),
  );
  const offset = Math.max(0, Math.floor(input.offset));
  const session = requireRuntimePostgresSession(
    await getRuntimeDbSession(
      {
        ...context,
        playName: context.playName,
        runId: context.runId ?? input.runId,
      },
      {
        tableNamespace: input.tableNamespace,
        logicalTable: 'sheet_rows',
        operations: ['rows.read'],
        limits: { maxRows: runtimeDbSessionRowLimit(limit) },
      },
    ),
  );
  const rows = await readRuntimeRows(session, {
    limit,
    offset,
    runId: input.runId ?? null,
    rowMode: input.rowMode,
  });
  return {
    rows: rows.map((row) => row.data),
    limit,
    offset,
  };
}

export async function readRuntimeSheetDatasetRowKeys(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    runId: string;
    keys: string[];
    rowMode?: 'output' | 'all' | 'terminalAnyRun';
  },
): Promise<{ keys: string[] }> {
  if (context.dbSessionStrategy === 'gateway_only') {
    return await postRuntimeApi(context, {
      action: 'runtime_sheet_read_row_keys',
      input,
    });
  }
  const keys = [
    ...new Set(
      input.keys.map((key) => key.trim()).filter((key) => key.length > 0),
    ),
  ];
  if (keys.length === 0) {
    return { keys: [] };
  }
  const session = requireRuntimePostgresSession(
    await getRuntimeDbSession(
      {
        ...context,
        playName: context.playName,
        runId: context.runId ?? input.runId,
      },
      {
        tableNamespace: input.tableNamespace,
        logicalTable: 'sheet_rows',
        operations: ['rows.read'],
        limits: { maxRows: runtimeDbSessionRowLimit(keys.length) },
      },
    ),
  );
  const rows = await withRuntimePostgres(session, async (client) => {
    if (input.rowMode === 'terminalAnyRun') {
      const { rows: matchedRows } = await client.query<{ _key: string }>(
        `SELECT _key
           FROM ${sheetTable(session)}
          WHERE _key = ANY($1::text[])
            AND _status IN ('enriched', 'failed')`,
        [keys],
      );
      return matchedRows;
    }
    if (input.rowMode === 'all') {
      const { rows: matchedRows } = await client.query<{ _key: string }>(
        `SELECT _key
           FROM ${sheetTable(session)}
          WHERE _run_id = $1::text
            AND _key = ANY($2::text[])
            AND _status IN ('enriched', 'failed')`,
        [input.runId, keys],
      );
      return matchedRows;
    }
    const { rows: matchedRows } = await client.query<{ _key: string }>(
      `SELECT _key
         FROM ${sheetTable(session)}
        WHERE _run_id = $1::text
          AND _key = ANY($2::text[])
          AND _status = 'enriched'`,
      [input.runId, keys],
    );
    return matchedRows;
  });
  return { keys: rows.map((row) => row._key) };
}

export type RuntimeSheetRowOutcomeCounts = {
  completed: number;
  failed: number;
  total: number;
};

/**
 * Count settled row outcomes for one physical runtime sheet (ADR 0016 rule 3).
 *
 * The sheet rows ARE the data plane: `_status = 'enriched'` is a durably
 * persisted success, `'failed'` a settled row failure, and everything except
 * `'stale'` counts toward the denominator. Split from the session wrapper so
 * the counting SQL runs identically against a test PGlite client and a
 * production tenant runtime Postgres.
 */
export async function countRuntimeSheetRowOutcomes(
  client: {
    query<Row extends Record<string, unknown> = Record<string, unknown>>(
      sql: string,
      params?: unknown[],
    ): Promise<{ rows: Row[] }>;
  },
  input: { schema: string; table: string; runId: string },
): Promise<RuntimeSheetRowOutcomeCounts> {
  const { rows } = await client.query<{
    completed: number | string;
    failed: number | string;
    total: number | string;
  }>(
    `SELECT
       count(*) FILTER (WHERE _status = 'enriched')::int AS completed,
       count(*) FILTER (WHERE _status = 'failed')::int AS failed,
       count(*) FILTER (WHERE _status <> 'stale')::int AS total
     FROM ${quoteIdentifier(input.schema)}.${quoteIdentifier(input.table)}
     WHERE _run_id = $1::text`,
    [input.runId],
  );
  const row = rows[0];
  return {
    completed: Math.max(0, Number(row?.completed ?? 0) || 0),
    failed: Math.max(0, Number(row?.failed ?? 0) || 0),
    total: Math.max(0, Number(row?.total ?? 0) || 0),
  };
}

/**
 * Read the settled row-outcome counts for a dataset's runtime sheet through the
 * run's preloaded DB session. Used by the scheduler worker at terminal finalize
 * as the reconciliation of record for per-map progress: sheet counts exist
 * regardless of runner mode (in-worker, detached Daytona push), unlike
 * checkpoint map frames or streamed events. Requires a direct session strategy
 * (`preloaded`/`trusted_dynamic`); there is deliberately no gateway action —
 * the worker holds the launch's preloaded sessions itself.
 */
export async function readRuntimeSheetRowOutcomeCounts(
  context: RuntimeApiContext & { playName: string; runId: string },
  input: { tableNamespace: string },
): Promise<RuntimeSheetRowOutcomeCounts> {
  if (context.dbSessionStrategy === 'gateway_only') {
    throw new Error(
      'readRuntimeSheetRowOutcomeCounts requires a direct DB session strategy.',
    );
  }
  const session = requireRuntimePostgresSession(
    await getRuntimeDbSession(
      { ...context, playName: context.playName },
      {
        tableNamespace: input.tableNamespace,
        logicalTable: 'sheet_rows',
        operations: ['rows.read'],
        limits: { maxRows: runtimeDbSessionRowLimit(1) },
      },
    ),
  );
  return await withRuntimePostgres(session, (client) =>
    countRuntimeSheetRowOutcomes(client, {
      schema: session.postgres.schema,
      table: session.postgres.sheetTable,
      runId: context.runId,
    }),
  );
}

export async function persistRuntimeCsvDataset(
  context: RuntimeApiContext & { playName: string },
  input: {
    tableNamespace: string;
    playInput?: Record<string, unknown> | null;
    sheetContract: PlaySheetContract;
    rows: Record<string, unknown>[];
    runId: string;
    sourceLabel?: string | null;
  },
): Promise<PlayDataset<Record<string, unknown>>> {
  await startRuntimeSheetDataset(context, {
    playName: context.playName,
    tableNamespace: input.tableNamespace,
    playInput: input.playInput,
    sheetContract: input.sheetContract,
    rows: input.rows,
    runId: input.runId,
  });
  return createRuntimeBackedPlayDataset({
    context,
    tableNamespace: input.tableNamespace,
    datasetKind: 'csv',
    sourceLabel: input.sourceLabel,
    initialCount: input.rows.length,
    initialPreviewRows: input.rows.slice(0, 10),
    runId: input.runId,
  });
}

export async function upsertRuntimeRows(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    rows: Record<string, unknown>[];
    runId: string;
    sheetContract: PlaySheetContract;
    skipEnsureSheet?: boolean;
  },
): Promise<RowsWriteResponse> {
  if (!context.playName) {
    throw new Error('Runtime DB sessions require a playName.');
  }
  const session = await getRuntimeDbSession(
    {
      ...context,
      playName: context.playName,
      runId: context.runId ?? input.runId,
    },
    {
      tableNamespace: input.tableNamespace,
      logicalTable: 'sheet_rows',
      operations: ['rows.upsert'],
      limits: {
        maxRows: runtimeDbSessionRowLimit(input.rows.length),
      },
      sheetContract: input.sheetContract,
    },
  );
  return rowsWriteResponseSchema.parse(
    await writeRuntimeRows(requireRuntimePostgresSession(session), {
      tableNamespace: input.tableNamespace,
      rows: input.rows,
      runId: input.runId,
      sheetContract: input.sheetContract,
      idempotencyKey: createHash('sha1')
        .update(
          `${input.playName}:${input.tableNamespace}:${input.runId}:${JSON.stringify(input.rows)}`,
        )
        .digest('hex'),
      mode: 'upsert',
    }),
  );
}

export async function appendRuntimeRows(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    rows: Record<string, unknown>[];
    runId: string;
    idempotencyKey: string;
    sheetContract: PlaySheetContract;
  },
): Promise<RowsWriteResponse> {
  if (!context.playName) {
    throw new Error('Runtime DB sessions require a playName.');
  }
  const session = await getRuntimeDbSession(
    {
      ...context,
      playName: context.playName,
      runId: context.runId ?? input.runId,
    },
    {
      tableNamespace: input.tableNamespace,
      logicalTable: 'sheet_rows',
      operations: ['rows.append'],
      limits: {
        maxRows: runtimeDbSessionRowLimit(input.rows.length),
      },
      sheetContract: input.sheetContract,
    },
  );
  return rowsWriteResponseSchema.parse(
    await writeRuntimeRows(requireRuntimePostgresSession(session), {
      tableNamespace: input.tableNamespace,
      rows: input.rows,
      runId: input.runId,
      sheetContract: input.sheetContract,
      idempotencyKey: input.idempotencyKey,
      mode: 'append',
    }),
  );
}

export async function replaceRuntimeRows(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    rows: Record<string, unknown>[];
    runId: string;
    idempotencyKey: string;
    sheetContract: PlaySheetContract;
  },
): Promise<RowsWriteResponse> {
  if (!context.playName) {
    throw new Error('Runtime DB sessions require a playName.');
  }
  const session = await getRuntimeDbSession(
    {
      ...context,
      playName: context.playName,
      runId: context.runId ?? input.runId,
    },
    {
      tableNamespace: input.tableNamespace,
      logicalTable: 'sheet_rows',
      operations: ['rows.replace'],
      limits: {
        maxRows: runtimeDbSessionRowLimit(input.rows.length),
      },
      sheetContract: input.sheetContract,
    },
  );
  return rowsWriteResponseSchema.parse(
    await writeRuntimeRows(requireRuntimePostgresSession(session), {
      tableNamespace: input.tableNamespace,
      rows: input.rows,
      runId: input.runId,
      sheetContract: input.sheetContract,
      idempotencyKey: input.idempotencyKey,
      mode: 'replace',
    }),
  );
}

export function createRuntimeBackedPlayDataset(input: {
  context: RuntimeApiContext & {
    playName: string;
  };
  tableNamespace: string;
  datasetKind: 'csv' | 'map';
  sourceLabel?: string | null;
  initialCount?: number;
  initialPreviewRows?: Record<string, unknown>[];
  runId?: string | null;
}): PlayDataset<Record<string, unknown>> {
  const csvRows =
    input.datasetKind === 'csv' &&
    typeof input.initialCount === 'number' &&
    (input.initialPreviewRows?.length ?? 0) >= input.initialCount
      ? [...(input.initialPreviewRows ?? [])]
      : null;
  if (csvRows) {
    return createDeferredPlayDataset({
      datasetKind: input.datasetKind,
      datasetId: createRuntimeDatasetId(
        input.context.playName,
        input.tableNamespace,
      ),
      count: input.initialCount ?? csvRows.length,
      previewRows: csvRows.slice(0, 10),
      residentRows: csvRows,
      sourceLabel: input.sourceLabel ?? null,
      resolvers: {
        count: async () => csvRows.length,
        peek: async (limit) => csvRows.slice(0, Math.max(0, limit)),
        materialize: async (limit) =>
          limit === undefined
            ? [...csvRows]
            : csvRows.slice(0, Math.max(0, limit)),
        iterate: () =>
          ({
            async *[Symbol.asyncIterator]() {
              for (const row of csvRows) {
                yield row;
              }
            },
          }) as AsyncIterable<Record<string, unknown>>,
      },
    });
  }
  const runtimeContext = {
    ...input.context,
    runId: input.context.runId ?? input.runId,
  };
  return createDeferredPlayDataset({
    datasetKind: input.datasetKind,
    datasetId: createRuntimeDatasetId(
      input.context.playName,
      input.tableNamespace,
    ),
    count: input.initialCount ?? 0,
    knownCount: input.initialCount ?? null,
    backing: {
      storage: 'neon_sheet',
      sheet: {
        playName: input.context.playName,
        tableNamespace: input.tableNamespace,
      },
    },
    previewRows: input.initialPreviewRows ?? [],
    sourceLabel: input.sourceLabel ?? null,
    tableNamespace: input.tableNamespace,
    resolvers: {
      count: async () => {
        if (typeof input.initialCount === 'number') {
          return input.initialCount;
        }
        const summarySession = await getRuntimeDbSession(runtimeContext, {
          tableNamespace: input.tableNamespace,
          logicalTable: 'sheet_rows',
          operations: ['rows.read'],
        });
        const summary = await readRuntimeSummary(
          requireRuntimePostgresSession(summarySession),
        );
        return Number(summary.stats.total ?? 0);
      },
      peek: async (limit) => {
        const session = await getRuntimeDbSession(runtimeContext, {
          tableNamespace: input.tableNamespace,
          logicalTable: 'sheet_rows',
          operations: ['rows.read'],
        });
        const rows = await readRuntimeRows(
          requireRuntimePostgresSession(session),
          {
            limit,
            offset: 0,
            runId: input.runId ?? null,
          },
        );
        return rows.map((row) => row.data);
      },
      at: async (index) => {
        const session = await getRuntimeDbSession(runtimeContext, {
          tableNamespace: input.tableNamespace,
          logicalTable: 'sheet_rows',
          operations: ['rows.read'],
        });
        const rows = await readRuntimeRows(
          requireRuntimePostgresSession(session),
          {
            limit: 1,
            offset: index,
            runId: input.runId ?? null,
          },
        );
        return rows[0]?.data;
      },
      materialize: async (limit) => {
        const pageSize = 1000;
        const materialized: Record<string, unknown>[] = [];
        let offset = 0;
        while (true) {
          const session = await getRuntimeDbSession(runtimeContext, {
            tableNamespace: input.tableNamespace,
            logicalTable: 'sheet_rows',
            operations: ['rows.read'],
          });
          const rows = await readRuntimeRows(
            requireRuntimePostgresSession(session),
            {
              limit:
                limit !== undefined
                  ? Math.min(pageSize, Math.max(0, limit - materialized.length))
                  : pageSize,
              offset,
              runId: input.runId ?? null,
            },
          );
          if (rows.length === 0) {
            break;
          }
          materialized.push(...rows.map((row) => row.data));
          if (limit !== undefined && materialized.length >= limit) {
            return materialized.slice(0, limit);
          }
          offset += rows.length;
        }
        return materialized;
      },
      iterate: () =>
        ({
          async *[Symbol.asyncIterator]() {
            const pageSize = 1000;
            let offset = 0;
            while (true) {
              const session = await getRuntimeDbSession(runtimeContext, {
                tableNamespace: input.tableNamespace,
                logicalTable: 'sheet_rows',
                operations: ['rows.read'],
              });
              const page = await readRuntimeRows(
                requireRuntimePostgresSession(session),
                {
                  limit: pageSize,
                  offset,
                  runId: input.runId ?? null,
                },
              );
              if (page.length === 0) {
                return;
              }
              for (const row of page) {
                yield row.data;
              }
              offset += page.length;
            }
          },
        }) as AsyncIterable<Record<string, unknown>>,
    },
  });
}

export function resolveRuntimeSheetContract(
  pipeline: PlayStaticPipeline | null | undefined,
  tableNamespace: string | null | undefined,
): PlaySheetContract | null {
  const requestedNamespace = tableNamespace?.trim();
  if (!pipeline || !requestedNamespace) {
    return null;
  }
  const normalizedNamespace = normalizeTableNamespace(requestedNamespace);

  const rootNamespace = pipeline.tableNamespace?.trim();
  if (
    rootNamespace &&
    normalizeTableNamespace(rootNamespace) === normalizedNamespace
  ) {
    return pipeline.sheetContract ?? null;
  }

  for (const substep of [...(pipeline.stages ?? []), ...pipeline.substeps]) {
    if (substep.type !== 'dataset') {
      continue;
    }
    const substepNamespace = substep.tableNamespace?.trim();
    if (
      substepNamespace &&
      normalizeTableNamespace(substepNamespace) === normalizedNamespace
    ) {
      return substep.sheetContract ?? null;
    }
  }

  return null;
}