import { createHash } 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,
} from './work-receipts';
import {
  sanitizePostgresJsonValue,
  stringifyPostgresJson,
} from './postgres-json';
import { RECEIPT_STATUS_CODE, receiptStatusFromCode } from './receipt-status';
import type { MapRowOutcome } from './durability-store';
import {
  normalizeRuntimeMapInputIndex,
  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 type { PlayArtifactKind } from './backend';

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' | 'sandbox_public_key' | null;
  preloadedDbSessions?: PreloadedRuntimeDbSession[] | null;
  vercelProtectionBypassToken?: string | null;
  disablePostgresPoolCache?: boolean | null;
  postgresSessionUnwrapKey?: string | null;
};

type DbSessionCacheEntry = {
  session: CreateDbSessionResponse;
};

type RuntimeQueryClient = Pick<RuntimeOneShotQueryClient, 'query'>;
type RuntimeDatasetRowEntry = {
  key: string;
  row: Record<string, unknown>;
  inputIndex: number;
};
const dbSessionCache = new Map<string, DbSessionCacheEntry>();
const dbSessionInFlight = new Map<string, Promise<CreateDbSessionResponse>>();
const postgresPools = new Map<string, RuntimePool>();
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;
const RUNTIME_DB_SESSION_TTL_SECONDS = 10 * 60;
const RUNTIME_SHEET_ENSURE_CACHE_TTL_MS = 10 * 60_000;
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_MAX_ATTEMPTS = RUNTIME_API_RETRY_DELAYS_MS.length + 1;
const RUNTIME_API_DEFAULT_RETRY_AFTER_MS = 2_000;
const RUNTIME_API_REQUEST_TIMEOUT_MS = 30_000;
const vercelProtectionCookieCache = new Map<string, Promise<string | null>>();
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 = 5_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.
const RUNTIME_POSTGRES_POOL_MAX_CONNECTIONS = 8;
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';
  contractSnapshot?: Record<string, unknown> | null;
};

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

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

type RuntimeApiActionRequest =
  | {
      action: 'resolve_play';
      playRef: string;
      /**
       * Artifact kind the caller can execute. Node runners (postgres
       * scheduler inline executor, Daytona/local-process) request
       * `cjs_node20` so a child published with an esm_workers artifact is
       * re-bundled server-side into a `Module._compile`-loadable module.
       * Omitted = the stored artifact is returned verbatim.
       */
      artifactKind?: PlayArtifactKind;
    }
  | {
      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;
    };

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.');
  }
  const url = new URL(
    `${baseUrl.replace(/\/$/, '')}/api/v2/plays/internal/runtime`,
  );
  const bypassToken = context.vercelProtectionBypassToken?.trim();
  if (bypassToken) {
    url.searchParams.set('x-vercel-set-bypass-cookie', 'true');
    url.searchParams.set('x-vercel-protection-bypass', bypassToken);
  }
  return url.toString();
}

function resolveRuntimeApiHeaders(
  context: RuntimeApiContext,
  vercelProtectionCookie?: string | null,
): 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}`,
    ...(context.vercelProtectionBypassToken
      ? { 'x-vercel-protection-bypass': context.vercelProtectionBypassToken }
      : {}),
    ...(vercelProtectionCookie ? { cookie: vercelProtectionCookie } : {}),
  };
}

function setCookieHeaders(headers: Headers): string[] {
  const getter = (headers as Headers & { getSetCookie?: () => string[] })
    .getSetCookie;
  if (typeof getter === 'function') {
    return getter.call(headers);
  }
  const single = headers.get('set-cookie');
  return single ? [single] : [];
}

function cookieHeaderFromSetCookie(headers: Headers): string | null {
  const cookies = setCookieHeaders(headers)
    .flatMap((header) => header.split(/,(?=\s*[^;,=]+=[^;,]+)/g))
    .map((header) => header.split(';', 1)[0]?.trim() ?? '')
    .filter(Boolean);
  return cookies.length > 0 ? cookies.join('; ') : null;
}

function resolveVercelProtectionCookie(
  context: RuntimeApiContext,
): Promise<string | null> {
  const bypassToken = context.vercelProtectionBypassToken?.trim();
  const baseUrl = context.baseUrl?.trim().replace(/\/$/, '');
  if (!bypassToken || !baseUrl) return Promise.resolve(null);

  const cacheKey = `${baseUrl}\n${bypassToken}`;
  const cached = vercelProtectionCookieCache.get(cacheKey);
  if (cached) return cached;

  const promise = (async () => {
    const url = new URL(`${baseUrl}/api/v2/health`);
    url.searchParams.set('x-vercel-set-bypass-cookie', 'true');
    url.searchParams.set('x-vercel-protection-bypass', bypassToken);
    const response = await fetch(url.toString(), {
      headers: { 'x-vercel-protection-bypass': bypassToken },
    }).catch(() => null);
    return response ? cookieHeaderFromSetCookie(response.headers) : null;
  })();
  vercelProtectionCookieCache.set(cacheKey, promise);
  return promise;
}

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 vercelProtectionCookie = await resolveVercelProtectionCookie(context);
  for (let attempt = 1; attempt <= RUNTIME_API_MAX_ATTEMPTS; attempt += 1) {
    let response: Response;
    const abortController = new AbortController();
    const timeout = setTimeout(
      () => abortController.abort(),
      RUNTIME_API_REQUEST_TIMEOUT_MS,
    );
    try {
      response = await fetch(url, {
        method: 'POST',
        headers: resolveRuntimeApiHeaders(context, vercelProtectionCookie),
        body: JSON.stringify(body),
        signal: abortController.signal,
      });
    } catch (error) {
      if (attempt < RUNTIME_API_MAX_ATTEMPTS) {
        await sleepRuntimeApiRetry(attempt);
        continue;
      }
      const message = error instanceof Error ? error.message : String(error);
      throw new Error(
        `Runtime API request to ${url} failed before receiving a response: ${message}`,
      );
    } finally {
      clearTimeout(timeout);
    }

    const parsed = (await response.json().catch(() => null)) as Record<
      string,
      unknown
    > | null;
    if (response.ok) {
      return parsed as TResponse;
    }

    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?.details === 'string'
        ? parsed.details
        : typeof parsed?.debug_error === 'string'
          ? parsed.debug_error
          : null;
    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 < RUNTIME_API_MAX_ATTEMPTS) {
      await sleepRuntimeApiRetry(attempt, retryAfterMs);
      continue;
    }

    const errorMessage =
      typeof parsed?.error === 'string'
        ? parsed.error
        : `Runtime API request failed with status ${response.status}.`;
    throw new Error(
      details
        ? `${errorMessage} (status ${response.status}): ${details}`
        : `${errorMessage} (status ${response.status}).`,
    );
  }

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

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

async function sleepRuntimeApiRetry(
  attempt: number,
  retryAfterMs?: number,
): Promise<void> {
  await new Promise((resolve) =>
    setTimeout(resolve, runtimeApiRetryDelayMs(attempt, retryAfterMs)),
  );
}

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/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<void> => {
    await client.query(
      `SELECT ${selectList}
       FROM ${sheetTable(session)}
       LIMIT 0`,
    );
  };
  try {
    if (isRuntimeOneShotQueryFactoryRegistered()) {
      await withRuntimeOneShotPostgres(session, query);
    } else {
      await withRuntimePostgres(session, query);
    }
    return true;
  } catch (error) {
    if (isMissingRelationError(error)) {
      return false;
    }
    throw error;
  }
}

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 = ensureRuntimeSheet(context, {
    playName: context.playName,
    tableNamespace: input.tableNamespace,
    sheetContract: input.sheetContract,
  });
  runtimeSheetEnsureCache.set(cacheKey, {
    expiresAt: now + RUNTIME_SHEET_ENSURE_CACHE_TTL_MS,
    promise,
  });
  try {
    await promise;
    input.timings?.push({
      phase: 'ensure_sheet_for_preloaded_session',
      ms: Date.now() - ensureStartedAt,
    });
    const recheckStartedAt = Date.now();
    const readyAfterEnsure = await isRuntimeSheetSchemaReady(input.session, {
      sheetContract: input.sheetContract,
    });
    input.timings?.push({
      phase: 'schema_check_after_preloaded_session_ensure',
      ms: Date.now() - recheckStartedAt,
      ready: readyAfterEnsure,
    });
    if (!readyAfterEnsure) {
      if (runtimeSheetEnsureCache.get(cacheKey)?.promise === promise) {
        runtimeSheetEnsureCache.delete(cacheKey);
      }
      throw new Error(
        `Runtime sheet schema for ctx.dataset("${input.tableNamespace}") is still not ready after ensure_sheet.`,
      );
    }
  } 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 {
  if (context.dbSessionStrategy === 'sandbox_public_key') {
    return 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 (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) {
    const pool = postgresPools.get(session.postgresUrl);
    postgresPools.delete(session.postgresUrl);
    await pool?.end().catch(() => {});
  }
}

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> {
  if (context.dbSessionStrategy === 'sandbox_public_key') {
    return;
  }
  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');
      });
      return;
    } catch (error) {
      if (
        attempt >= RUNTIME_POSTGRES_PREWARM_MAX_ATTEMPTS ||
        !isTransientRuntimePostgresConnectionError(error)
      ) {
        const pool = postgresPools.get(session.postgresUrl);
        postgresPools.delete(session.postgresUrl);
        if (pool) {
          await Promise.resolve(pool.end()).catch(() => {});
        }
        throw error;
      }
      const pool = postgresPools.get(session.postgresUrl);
      postgresPools.delete(session.postgresUrl);
      if (pool) {
        await Promise.resolve(pool.end()).catch(() => {});
      }
      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 === '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/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,
): Promise<RuntimePoolClient> {
  let timedOut = false;
  let timeout: ReturnType<typeof setTimeout> | null = null;
  const connectPromise = pool.connect();
  connectPromise
    .then((client) => {
      if (timedOut) {
        client.release();
      }
    })
    .catch(() => {});
  try {
    return await Promise.race([
      connectPromise,
      new Promise<never>((_, reject) => {
        timeout = setTimeout(() => {
          timedOut = true;
          reject(
            new Error(
              `Runtime Postgres connection timed out after ${RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS}ms.`,
            ),
          );
        }, RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS);
      }),
    ]);
  } finally {
    if (timeout) {
      clearTimeout(timeout);
    }
  }
}

function getPostgresPool(postgresUrl: string, cachePool = true): RuntimePool {
  if (!cachePool) {
    return createRuntimePool({
      connectionString: postgresUrl,
      maxConnections: 1,
      idleTimeoutMs: 0,
      connectTimeoutMs: RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS,
    });
  }
  const existing = postgresPools.get(postgresUrl);
  if (existing) {
    return existing;
  }
  const pool = createRuntimePool({
    connectionString: postgresUrl,
    maxConnections: RUNTIME_POSTGRES_POOL_MAX_CONNECTIONS,
    idleTimeoutMs: 15_000,
    connectTimeoutMs: RUNTIME_POSTGRES_CONNECT_TIMEOUT_MS,
  });
  postgresPools.set(postgresUrl, pool);
  return pool;
}

async function resetRuntimePostgresPool(postgresUrl: string): Promise<void> {
  const pool = postgresPools.get(postgresUrl);
  postgresPools.delete(postgresUrl);
  if (pool) {
    await Promise.resolve(pool.end()).catch(() => {});
  }
}

async function withRuntimePostgres<T>(
  session: RuntimePostgresSession,
  fn: (client: RuntimePoolClient) => Promise<T>,
  options: { cachePool?: boolean } = {},
): Promise<T> {
  let client: RuntimePoolClient | null = null;
  let requestLocalPool: RuntimePool | null = null;
  const cachePool =
    (options.cachePool ?? true) && canReuseRuntimePostgresPoolsAcrossRequests();
  for (
    let attempt = 1;
    attempt <= RUNTIME_POSTGRES_CONNECT_MAX_ATTEMPTS;
    attempt += 1
  ) {
    try {
      const pool = getPostgresPool(session.postgresUrl, cachePool);
      if (!cachePool) {
        requestLocalPool = pool;
      }
      client = await connectRuntimePostgresPool(pool);
      break;
    } catch (error) {
      if (cachePool) {
        const pool = postgresPools.get(session.postgresUrl);
        postgresPools.delete(session.postgresUrl);
        if (pool) {
          await Promise.resolve(pool.end()).catch(() => {});
        }
      } else if (requestLocalPool) {
        await Promise.resolve(requestLocalPool.end()).catch(() => {});
        requestLocalPool = null;
      }
      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
          ],
      );
    }
  }
  if (!client) {
    throw new Error('Runtime Postgres connection was not acquired.');
  }
  try {
    return await fn(client);
  } finally {
    client.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 {
  return (
    error !== null &&
    typeof error === 'object' &&
    'code' in error &&
    String(error.code) === '42501'
  );
}

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;
    timings?: RuntimeSheetTiming[];
  },
  operation: (client: RuntimeQueryClient) => Promise<T>,
): Promise<T> {
  const totalStartedAt = Date.now();
  const result = await withRuntimeSheetProvisioningRetry(
    context,
    input,
    async () => {
      if (!input.transactional && 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) => {
          if (input.transactional) await client.query('BEGIN');
          try {
            const result = await operation(client);
            if (input.transactional) await client.query('COMMIT');
            return result;
          } catch (error) {
            if (input.transactional) {
              await client.query('ROLLBACK').catch(() => {});
            }
            throw error;
          }
        },
        { cachePool: !context.disablePostgresPoolCache },
      );
    },
  );
  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 workReceiptTable(session: RuntimePostgresSession): string {
  return fqRuntimeTable(
    session,
    session.postgres.receiptTable ?? RUNTIME_WORK_RECEIPT_POSTGRES_TABLE,
  );
}

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 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 {
  return {
    key: String(raw.k ?? ''),
    status: receiptStatusFromCode(raw.status),
    output: raw.output == null ? null : raw.output,
    error: raw.error == null ? null : String(raw.error),
    runId: raw.run_id == null ? null : String(raw.run_id),
    updatedAt: raw.updated_at == null ? null : String(raw.updated_at),
  };
}

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

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,
      run_id text,
      updated_at timestamptz NOT NULL DEFAULT now()
    )
  `,
    )
    .then(() => undefined);
  runtimeWorkReceiptEnsureCache.set(cacheKey, promise);
  try {
    await promise;
  } catch (error) {
    if (runtimeWorkReceiptEnsureCache.get(cacheKey) === promise) {
      runtimeWorkReceiptEnsureCache.delete(cacheKey);
    }
    throw error;
  }
}

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. A genuinely
  // missing schema, permission error, or second failure still rethrows loudly.
  const runWithSelfHeal = async (client: RuntimeQueryClient): Promise<T> => {
    try {
      return await operation(client);
    } catch (error) {
      if (isMissingRelationError(error)) {
        runtimeWorkReceiptEnsureCache.delete(
          runtimeWorkReceiptEnsureCacheKey(session),
        );
        await ensureRuntimeWorkReceiptTable(session, client);
      } else if (isPostgresPermissionDeniedError(error)) {
        await repairRuntimeStorageGrants(context, {
          playName: context.playName?.trim() || session.playName,
        });
      } else {
        throw error;
      }
      return await operation(client);
    }
  };

  for (
    let attempt = 1;
    attempt <= RUNTIME_WORK_RECEIPT_QUERY_MAX_ATTEMPTS;
    attempt += 1
  ) {
    try {
      return isRuntimeOneShotQueryFactoryRegistered()
        ? await withRuntimeOneShotPostgres(session, runWithSelfHeal)
        : await withRuntimePostgres(
            session,
            (client) => runWithSelfHeal(client),
            { cachePool: !context.disablePostgresPoolCache },
          );
    } catch (error) {
      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 '_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 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) => {
    if (needsTransaction) 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 (needsTransaction) await client.query('COMMIT');
      return { disposition: 'completed', writtenRows };
    } catch (error) {
      if (needsTransaction) {
        await client.query('ROLLBACK').catch(() => {});
      }
      throw error;
    }
  });
}

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

export async function ensureRuntimeSheet(
  context: RuntimeApiContext,
  input: {
    playName: string;
    tableNamespace: string;
    sheetContract: PlaySheetContract;
  },
): Promise<void> {
  await postRuntimeApi<{ ok: true }>(context, {
    action: 'ensure_sheet',
    ...input,
    runId: context.runId ?? null,
    userEmail: normalizeRuntimeUserEmail(context.userEmail),
  });
}

async function prepareRuntimeSheetDatasetRows(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    chunks: RuntimeDatasetRowEntry[][];
    runId: string;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
    physicalInsertColumnsSql: string;
    physicalInsertValuesSql: string;
    outputPhysicalColumns: PhysicalSheetColumnProjection[];
  },
): Promise<{ inserted: number; pendingKeys: string[] }> {
  let inserted = 0;
  const pendingKeys: string[] = [];
  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 { 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
       ),
       existing_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _input_index = input_rows._input_index,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND 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${input.physicalInsertColumnsSql})
          SELECT _key, 'pending', $4::text, _input_index${input.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
        ),
        missing_output_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _status = 'pending',
              _run_id = $4::text,
              _input_index = input_rows._input_index,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._status = 'enriched'
            AND (${targetMissingOutputSql})
          RETURNING target._key
        ),
        claimed_existing_rows AS (
          UPDATE ${sheetTable(session)} AS target
          SET _run_id = $4::text,
              _input_index = input_rows._input_index,
              _updated_at = now(),
              _version = ${nextRuntimeSheetVersionExpression(session)}
          FROM input_rows
          WHERE target._key = input_rows._key
            AND target._status IN ('pending', 'running', 'failed')
            AND (
              target._run_id IS DISTINCT FROM $4::text
              OR target._input_index IS DISTINCT FROM input_rows._input_index
            )
          RETURNING target._key
        ),
        pending_rows AS (
          SELECT _key
          FROM inserted_rows
          UNION
          SELECT _key
          FROM missing_output_rows
          UNION
          SELECT _key
          FROM claimed_existing_rows
          UNION
          SELECT existing._key
          FROM ${sheetTable(session)} AS existing
          JOIN input_rows ON input_rows._key = existing._key
          WHERE existing._status IN ('pending', 'running', 'failed')
             OR (
               existing._status = 'enriched'
               AND (${existingMissingOutputSql})
             )
        ),
        inserted_count_cte AS (
          SELECT count(*)::bigint AS c FROM inserted_rows
        ),
        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 array_agg(_key) FROM pending_rows), '{}'::text[]) AS pending_keys,
          (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,
      ],
    );
    inserted += Number(rows[0]?.inserted_count ?? 0);
    if (Array.isArray(rows[0]?.pending_keys)) {
      pendingKeys.push(...(rows[0]?.pending_keys as string[]));
    }
  }
  return { inserted, pendingKeys };
}

async function buildRuntimeSheetDatasetStartResult(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    tableNamespace: string;
    sourceRowsLength: number;
    rowEntries: RuntimeDatasetRowEntry[];
    sheetContract: PlaySheetContract;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
    runId: string;
    inserted: number;
    pendingKeys: string[];
    timings?: RuntimeSheetTiming[];
  },
): Promise<PrepareRuntimeSheetResult> {
  if (input.inserted === input.rowEntries.length) {
    const datasetFields = input.sheetContract.columns.flatMap((column) =>
      column.source === 'datasetColumn' && typeof column.field === 'string'
        ? [column.field]
        : [],
    );
    return {
      inserted: input.inserted,
      skipped: input.sourceRowsLength - input.rowEntries.length,
      pendingRows: input.rowEntries.map((entry) => {
        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;
      }),
      completedRows: [],
      tableNamespace: input.tableNamespace,
    };
  }

  const pendingKeys = new Set(input.pendingKeys);
  const startedAt = Date.now();
  await markRuntimeRowsPendingForRecompute(client, session, {
    keys: [...pendingKeys],
    runId: input.runId,
    normalizedPlayName: input.normalizedPlayName,
    normalizedTableNamespace: input.normalizedTableNamespace,
  });
  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]),
  );
  if (pendingKeys.size > 0) {
    input.timings?.push({
      phase: 'mark_rows_pending_for_recompute',
      ms: Date.now() - startedAt,
      rows: pendingKeys.size,
    });
  }
  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>[] = [];
  for (const entry of input.rowEntries) {
    pendingRows.push(buildMergedRow(entry));
  }
  return {
    inserted: input.inserted,
    skipped: input.sourceRowsLength - input.rowEntries.length,
    pendingRows,
    completedRows: [],
    tableNamespace: input.tableNamespace,
  };
}

async function markRuntimeRowsPendingForRecompute(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: {
    keys: string[];
    runId: string;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
  },
): Promise<void> {
  if (input.keys.length === 0) return;
  await client.query(
    `WITH target_rows AS (
       SELECT _key, _status
         FROM ${sheetTable(session)}
        WHERE _key = ANY($1::text[])
        FOR UPDATE
     ),
     updated AS (
       UPDATE ${sheetTable(session)} AS target
        SET _status = 'pending',
            _run_id = $2::text,
            _updated_at = now(),
            _version = ${nextRuntimeSheetVersionExpression(session)}
        FROM target_rows
      WHERE target._key = target_rows._key
         AND (
           target._status <> 'pending'
           OR target._run_id IS DISTINCT FROM $2::text
         )
       RETURNING target_rows._status AS previous_status
     ),
     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
     ),
     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
     )
     SELECT 1`,
    [
      input.keys,
      input.runId,
      input.normalizedPlayName,
      input.normalizedTableNamespace,
    ],
  );
}

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

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, run_id, 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<WorkReceipt[]> {
  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.run_id,
             receipts.updated_at
        FROM input_keys
        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(mapRuntimeWorkReceiptRow);
    },
  );
}

export async function claimRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    reclaimRunning?: boolean;
    forceRefresh?: boolean;
  },
): Promise<WorkReceiptClaim> {
  const session = await getRuntimeWorkReceiptSession(context, {
    playName: input.playName,
    key: input.key,
  });
  return await withRuntimeWorkReceiptClient(
    context,
    session,
    async (client) => {
      const claimableStatuses = input.forceRefresh
        ? [
            RECEIPT_STATUS_PENDING,
            RECEIPT_STATUS_FAILED,
            RECEIPT_STATUS_RUNNING,
            RECEIPT_STATUS_COMPLETED,
            RECEIPT_STATUS_SKIPPED,
          ]
        : input.reclaimRunning
          ? [
              RECEIPT_STATUS_PENDING,
              RECEIPT_STATUS_FAILED,
              RECEIPT_STATUS_RUNNING,
            ]
          : [RECEIPT_STATUS_PENDING, RECEIPT_STATUS_FAILED];
      const { rows } = await client.query<Record<string, unknown>>(
        `
        WITH claimed AS (
          INSERT INTO ${workReceiptTable(session)} (k, status, run_id, updated_at)
          VALUES (decode($1, 'hex'), $2::smallint, $3, now())
          ON CONFLICT (k) DO UPDATE
          SET status = $4::smallint,
              output = NULL,
              run_id = $3,
              error = NULL,
              updated_at = now()
          WHERE ${workReceiptTable(session)}.status = ANY($5::smallint[])
          RETURNING convert_from(k, 'UTF8') AS k, status, output, error, run_id, updated_at
        )
        SELECT k, status, output, error, run_id, updated_at
        FROM claimed
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_RUNNING,
          input.runId,
          RECEIPT_STATUS_RUNNING,
          claimableStatuses,
        ],
      );
      const claimed = rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
      if (claimed) {
        return { disposition: 'claimed', receipt: claimed };
      }

      const latest = await readRuntimeWorkReceipt(client, session, input.key);
      if (latest && isReusableWorkReceipt(latest)) {
        return { disposition: 'reused', receipt: latest };
      }
      if (latest?.status === 'running') {
        return { disposition: 'running', receipt: latest };
      }
      if (latest?.status === 'failed') {
        return { disposition: 'failed', receipt: latest };
      }
      throw new Error(
        `Runtime receipt ${input.key} claim did not return execution ownership.`,
      );
    },
  );
}

export async function claimRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    keys: string[];
    reclaimRunning?: boolean;
    forceRefresh?: boolean;
  },
): Promise<WorkReceiptClaim[]> {
  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 keyHexes = keys.map(workReceiptKeyHex);
      const reclaimStatuses = input.forceRefresh
        ? [
            RECEIPT_STATUS_PENDING,
            RECEIPT_STATUS_FAILED,
            RECEIPT_STATUS_RUNNING,
            RECEIPT_STATUS_COMPLETED,
            RECEIPT_STATUS_SKIPPED,
          ]
        : input.reclaimRunning
          ? [
              RECEIPT_STATUS_PENDING,
              RECEIPT_STATUS_FAILED,
              RECEIPT_STATUS_RUNNING,
            ]
          : [RECEIPT_STATUS_PENDING, RECEIPT_STATUS_FAILED];
      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)
      ),
      claimed AS (
        INSERT INTO ${workReceiptTable(session)} (k, status, run_id, updated_at)
        SELECT decode(input_keys.key_hex, 'hex'), $2::smallint, $3::text, now()
          FROM input_keys
        ON CONFLICT (k) DO UPDATE
        SET status = $4::smallint,
            output = NULL,
            run_id = $3::text,
            error = NULL,
            updated_at = now()
        WHERE ${workReceiptTable(session)}.status = ANY($5::smallint[])
        RETURNING k, status, output, error, run_id, updated_at
      ),
      latest AS (
        SELECT receipts.k,
               receipts.status,
               receipts.output,
               receipts.error,
               receipts.run_id,
               receipts.updated_at,
               input_keys.ord,
               false 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.run_id,
               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, run_id, updated_at, ord, claimed
          FROM latest
      )
      SELECT convert_from(k, 'UTF8') AS k,
             status,
             output,
             error,
             run_id,
             updated_at,
             claimed
        FROM returned
       ORDER BY ord
    `,
        [
          keyHexes,
          RECEIPT_STATUS_RUNNING,
          input.runId,
          RECEIPT_STATUS_RUNNING,
          reclaimStatuses,
        ],
      );
      return rows.map((row) => {
        const receipt = mapRuntimeWorkReceiptRow(row);
        if (row.claimed === true) {
          return { disposition: 'claimed', receipt } satisfies WorkReceiptClaim;
        }
        if (isReusableWorkReceipt(receipt)) {
          return { disposition: 'reused', receipt } satisfies WorkReceiptClaim;
        }
        if (receipt.status === 'running') {
          return { disposition: 'running', receipt } satisfies WorkReceiptClaim;
        }
        if (receipt.status === 'failed') {
          return { disposition: 'failed', receipt } satisfies WorkReceiptClaim;
        }
        return { disposition: 'running', receipt } satisfies WorkReceiptClaim;
      });
    },
  );
}

export async function completeRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    output: unknown;
  },
): Promise<WorkReceipt | null> {
  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 completed AS (
          UPDATE ${workReceiptTable(session)}
          SET status = $2::smallint,
              output = $3::jsonb,
              error = NULL,
              run_id = $4,
              updated_at = now()
          WHERE k = decode($1, 'hex')
            AND status = $5::smallint
            AND run_id = $4
          RETURNING convert_from(k, 'UTF8') AS k, status, output, error, run_id, updated_at
        ),
        latest AS (
          SELECT convert_from(k, 'UTF8') AS k, status, output, error, run_id, updated_at
            FROM ${workReceiptTable(session)}
           WHERE k = decode($1, 'hex')
             AND NOT EXISTS (SELECT 1 FROM completed)
        )
        SELECT k, status, output, error, run_id, updated_at FROM completed
        UNION ALL
        SELECT k, status, output, error, run_id, updated_at FROM latest
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_COMPLETED,
          input.output === null ? null : stringifyPostgresJson(input.output),
          input.runId,
          RECEIPT_STATUS_RUNNING,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

export async function completeRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    receipts: Array<{ runId: string; key: string; output: unknown }>;
  },
): Promise<WorkReceipt[]> {
  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,
               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 jsonb_array_elements($3::jsonb) WITH ORDINALITY AS output_values(output, ord)
            ON output_values.ord = key_values.ord
      ),
      completed AS (
        UPDATE ${workReceiptTable(session)} AS target
           SET status = $4::smallint,
               output = inputs.output,
               error = NULL,
               run_id = inputs.run_id,
               updated_at = now()
          FROM inputs
         WHERE target.k = decode(inputs.key_hex, 'hex')
           AND target.status = $5::smallint
           AND target.run_id = inputs.run_id
        RETURNING target.k, target.status, target.output, target.error, target.run_id, target.updated_at, inputs.ord
      ),
      latest AS (
        SELECT receipts.k,
               receipts.status,
               receipts.output,
               receipts.error,
               receipts.run_id,
               receipts.updated_at,
               inputs.ord
          FROM inputs
          JOIN ${workReceiptTable(session)} AS receipts
            ON receipts.k = decode(inputs.key_hex, 'hex')
         WHERE NOT EXISTS (
           SELECT 1 FROM completed WHERE completed.k = receipts.k
         )
      ),
      returned AS (
        SELECT k, status, output, error, run_id, updated_at, ord FROM completed
        UNION ALL
        SELECT k, status, output, error, run_id, updated_at, ord FROM latest
      )
      SELECT convert_from(returned.k, 'UTF8') AS k,
             returned.status,
             returned.output,
             returned.error,
             returned.run_id,
             returned.updated_at
        FROM returned
       ORDER BY returned.ord
    `,
        [
          receipts.map((receipt) => workReceiptKeyHex(receipt.key)),
          receipts.map((receipt) => receipt.runId),
          stringifyPostgresJson(
            receipts.map((receipt) =>
              receipt.output === null ? null : receipt.output,
            ),
          ),
          RECEIPT_STATUS_COMPLETED,
          RECEIPT_STATUS_RUNNING,
        ],
      );
      return rows.map(mapRuntimeWorkReceiptRow);
    },
  );
}

export async function failRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    error: string;
  },
): Promise<WorkReceipt | null> {
  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,
              run_id = $4,
              updated_at = now()
          WHERE k = decode($1, 'hex')
            AND status = $5::smallint
            AND run_id = $4
          RETURNING convert_from(k, 'UTF8') AS k, status, output, error, run_id, updated_at
        ),
        latest AS (
          SELECT convert_from(k, 'UTF8') AS k, status, output, error, run_id, updated_at
            FROM ${workReceiptTable(session)}
           WHERE k = decode($1, 'hex')
             AND NOT EXISTS (SELECT 1 FROM failed)
        )
        SELECT k, status, output, error, run_id, updated_at FROM failed
        UNION ALL
        SELECT k, status, output, error, run_id, updated_at FROM latest
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_FAILED,
          input.error,
          input.runId,
          RECEIPT_STATUS_RUNNING,
        ],
      );
      return rows[0] ? mapRuntimeWorkReceiptRow(rows[0]) : null;
    },
  );
}

export async function failRuntimeWorkReceipts(
  context: RuntimeApiContext,
  input: {
    playName: string;
    receipts: Array<{ runId: string; key: string; error: string }>;
  },
): Promise<WorkReceipt[]> {
  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,
               error_values.error,
               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 error_values(error, ord)
            ON error_values.ord = key_values.ord
      ),
      failed AS (
        UPDATE ${workReceiptTable(session)} AS target
           SET status = $4::smallint,
               output = NULL,
               error = inputs.error,
               run_id = inputs.run_id,
               updated_at = now()
          FROM inputs
         WHERE target.k = decode(inputs.key_hex, 'hex')
           AND target.status = $5::smallint
           AND target.run_id = inputs.run_id
        RETURNING target.k, target.status, target.output, target.error, target.run_id, target.updated_at, inputs.ord
      ),
      latest AS (
        SELECT receipts.k,
               receipts.status,
               receipts.output,
               receipts.error,
               receipts.run_id,
               receipts.updated_at,
               inputs.ord
          FROM inputs
          JOIN ${workReceiptTable(session)} AS receipts
            ON receipts.k = decode(inputs.key_hex, 'hex')
         WHERE NOT EXISTS (
           SELECT 1 FROM failed WHERE failed.k = receipts.k
         )
      ),
      returned AS (
        SELECT k, status, output, error, run_id, updated_at, ord FROM failed
        UNION ALL
        SELECT k, status, output, error, run_id, updated_at, ord FROM latest
      )
      SELECT convert_from(returned.k, 'UTF8') AS k,
             returned.status,
             returned.output,
             returned.error,
             returned.run_id,
             returned.updated_at
        FROM returned
       ORDER BY returned.ord
    `,
        [
          receipts.map((receipt) => workReceiptKeyHex(receipt.key)),
          receipts.map((receipt) => receipt.runId),
          receipts.map((receipt) => receipt.error),
          RECEIPT_STATUS_FAILED,
          RECEIPT_STATUS_RUNNING,
        ],
      );
      return rows.map(mapRuntimeWorkReceiptRow);
    },
  );
}

export async function skipRuntimeWorkReceipt(
  context: RuntimeApiContext,
  input: {
    playName: string;
    runId: string;
    key: string;
    output: unknown;
  },
): Promise<WorkReceipt | null> {
  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 skipped AS (
          UPDATE ${workReceiptTable(session)}
          SET status = $2::smallint,
              output = $3::jsonb,
              error = NULL,
              run_id = $4,
              updated_at = now()
          WHERE k = decode($1, 'hex')
            AND status = $5::smallint
            AND run_id = $4
          RETURNING convert_from(k, 'UTF8') AS k, status, output, error, run_id, updated_at
        ),
        latest AS (
          SELECT convert_from(k, 'UTF8') AS k, status, output, error, run_id, updated_at
            FROM ${workReceiptTable(session)}
           WHERE k = decode($1, 'hex')
             AND NOT EXISTS (SELECT 1 FROM skipped)
        )
        SELECT k, status, output, error, run_id, updated_at FROM skipped
        UNION ALL
        SELECT k, status, output, error, run_id, updated_at FROM latest
      `,
        [
          workReceiptKeyHex(input.key),
          RECEIPT_STATUS_SKIPPED,
          input.output === null ? null : stringifyPostgresJson(input.output),
          input.runId,
          RECEIPT_STATUS_RUNNING,
        ],
      );
      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;
    inputOffset?: number;
  },
): Promise<PrepareRuntimeSheetResult> {
  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,
  });
  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: [],
      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 normalizedPlayName = normalizePlayNameForSheet(playName);
  const normalizedTableNamespace = normalizeTableNamespace(
    input.tableNamespace,
  );

  const chunks = chunkValues(rowEntries, DIRECT_POSTGRES_BATCH_SIZE);
  const needsTransaction = chunks.length > 1;

  const result = await withRuntimeSheetQueryClient(
    context,
    session,
    {
      playName,
      tableNamespace: input.tableNamespace,
      sheetContract: input.sheetContract,
      transactional: needsTransaction,
      timings,
    },
    async (client) => {
      const prepareStartedAt = Date.now();
      const prepared = await prepareRuntimeSheetDatasetRows(client, session, {
        chunks,
        runId: input.runId,
        normalizedPlayName,
        normalizedTableNamespace,
        physicalInsertColumnsSql,
        physicalInsertValuesSql,
        outputPhysicalColumns,
      });
      timings.push({
        phase: 'prepare_rows_sql',
        ms: Date.now() - prepareStartedAt,
        rows: rowEntries.length,
        chunks: chunks.length,
        inserted: prepared.inserted,
        pending: prepared.pendingKeys.length,
      });
      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,
        timings,
        ...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;
    },
  );
  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, timings };
}

type CompleteRuntimeMapRowChunksInput = {
  chunks: RuntimePreparedCompletedRow[][];
  physicalUpdateSetSql: string;
  physicalColumnProjections: PhysicalSheetColumnProjection[];
  runId: string;
  normalizedPlayName: string;
  normalizedTableNamespace: string;
  outputFields: string[];
};

async function completeRuntimeMapRowChunks(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: CompleteRuntimeMapRowChunksInput,
): Promise<{ updated: number }> {
  let updated = 0;
  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) =>
      stringifyPostgresJson(row.data_patch),
    );
    const chunkCellMetaPatches = chunk.map((row) =>
      stringifyPostgresJson(row.cell_meta_patch),
    );
    const targetChangedPatchedCellSql = changedPatchedCellSql(
      'target',
      'updates.data_patch',
      input.physicalColumnProjections,
    );
    const { rows } = await client.query<{
      updated: number;
      matched_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
       ),
       applied_rows AS (
         UPDATE ${sheetTable(session)} AS target
         SET _status = 'enriched',
             _run_id = $5::text,
             _error = NULL,
             _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 (target._run_id IS NULL OR target._run_id <= $5::text)
           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
       ),
       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,
         (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)],
      ],
    );
    updated += Number(rows[0]?.updated ?? 0);

    const matchedKeys = new Set(rows[0]?.matched_keys ?? []);
    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;
  }
  return { updated };
}

async function completeRuntimeMapRowChunksWithInputIndexRepair(
  client: RuntimeQueryClient,
  session: RuntimePostgresSession,
  input: CompleteRuntimeMapRowChunksInput,
): Promise<{ updated: number }> {
  let updated = 0;
  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) =>
      stringifyPostgresJson(row.data_patch),
    );
    const chunkCellMetaPatches = chunk.map((row) =>
      stringifyPostgresJson(row.cell_meta_patch),
    );
    const targetChangedPatchedCellSql = changedPatchedCellSql(
      'target',
      'updates.data_patch',
      input.physicalColumnProjections,
    );
    const { rows: appliedRows } = await client.query<{ _key: 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 DISTINCT ON (target._key)
                target._key AS matched_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
       ),
       applied_rows AS (
         UPDATE ${sheetTable(session)} AS target
         SET _status = 'enriched',
             _run_id = $5::text,
             _error = NULL,
             _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 (target._run_id IS NULL OR target._run_id <= $5::text)
           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
       ),
       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 _key FROM applied_rows`,
      [
        chunkKeys,
        chunkInputIndexes,
        chunkDataPatches,
        chunkCellMetaPatches,
        input.runId,
        input.normalizedPlayName,
        input.normalizedTableNamespace,
        [...new Set(input.outputFields)],
      ],
    );
    updated += appliedRows.length;
  }
  return { updated };
}

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) =>
      stringifyPostgresJson(row.data_patch),
    );
    const chunkCellMetaPatches = chunk.map((row) =>
      stringifyPostgresJson(row.cell_meta_patch),
    );
    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, _cell_meta${physicalInsertColumnsSql})
         SELECT _key, 'enriched', $5::text, input_index, 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($8::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,
        [...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;
    runId: string;
    normalizedPlayName: string;
    normalizedTableNamespace: string;
  },
): Promise<{ updated: number }> {
  let updated = 0;
  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) =>
      stringifyPostgresJson(row.data_patch),
    );
    const chunkCellMetaPatches = chunk.map((row) =>
      stringifyPostgresJson(row.cell_meta_patch),
    );
    const chunkErrors = chunk.map((row) => row.error);
    // 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: appliedRows } = await client.query<{ _key: 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
       ),
       applied_rows AS (
         UPDATE ${sheetTable(session)} AS target
         SET _status = 'failed',
             _run_id = $6::text,
             _error = updates.error,
             _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 target._status <> 'enriched'
           AND (target._run_id IS NULL OR target._run_id <= $6::text)
         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 = 'running')::bigint AS from_running
           FROM applied_rows
       ),
       summary_counts AS (
         SELECT (c - already_failed)::bigint AS newly_failed,
                from_running,
                GREATEST(c - already_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 $7::text,
                $8::text,
                0,
                (-from_queued)::int,
                (-from_running)::int,
                0,
                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),
           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: 'target.completed',
             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 _key FROM applied_rows`,
      [
        chunkKeys,
        chunkInputIndexes,
        chunkDataPatches,
        chunkCellMetaPatches,
        chunkErrors,
        input.runId,
        input.normalizedPlayName,
        input.normalizedTableNamespace,
        failedCellFields,
        failedCellTotals,
      ],
    );
    updated += appliedRows.length;
  }
  return { updated };
}

/**
 * Mark map rows terminal in the per-run scoped Postgres sheet table by key.
 * Mirrors server-side `store.completeSheetRows` semantics: UPDATE-by-key with
 * the row's enriched output values written into materialized physical
 * columns, _status flipped to 'enriched', and the sheet cursor advanced.
 *
 * 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 the workers_edge harness's `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;
  },
): Promise<{ updated: number }> {
  if (input.rows.length === 0) {
    return { updated: 0 };
  }
  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) continue;
    uniqueRows.set(row.key, row);
  }
  if (uniqueRows.size === 0) {
    return { updated: 0 };
  }

  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 ?? [])];

  return await withRuntimeSheetQueryClient(
    context,
    session,
    {
      playName: context.playName,
      tableNamespace: input.tableNamespace,
      sheetContract,
      transactional: needsTransaction,
    },
    async (client) => {
      const completed =
        completedRows.length > 0
          ? await (async () => {
              const chunkInput = {
                chunks,
                physicalUpdateSetSql,
                physicalColumnProjections: projections,
                runId: input.runId,
                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 };
            })()
          : { updated: 0 };
      const failed =
        failedRows.length > 0
          ? await failRuntimeMapRowChunks(client, session, {
              chunks: failedChunks,
              physicalUpdateSetSql,
              runId: input.runId,
              normalizedPlayName: normalizePlayNameForSheet(session.playName),
              normalizedTableNamespace: normalizeTableNamespace(
                input.tableNamespace,
              ),
            })
          : { updated: 0 };
      return { updated: completed.updated + failed.updated };
    },
  );
}

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 }> {
  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';
  },
): Promise<{ keys: string[] }> {
  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 === '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 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' &&
    (input.initialPreviewRows?.length ?? 0) >= (input.initialCount ?? 0)
      ? [...(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),
      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,
    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);
      },
      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;
}