import { createHash } from "node:crypto";
import { Pool } from "pg";
import type { PoolClient, QueryResultRow } from "pg";
import { AgentRunError } from "../common/errors.js";
import { redactJson } from "../common/redaction.js";
import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptListResult, QueueAttemptRecord, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueRetryActivation, QueueRetryReservation, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js";
import { newId, nowIso, stableHash } from "../common/validation.js";
import type { ActivateQueueRetryAttemptInput, AgentRunStore, CreateRunIdentity, DurableQueueClaim, EventOutboxConfig, ListQueueAttemptsInput, ListQueueTasksInput, ListRunsInput, ListSessionsInput, RecordQueueRetryAttemptFailureInput, ReserveQueueRetryAttemptInput, RunCollectionResult, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, SessionSteerAdmission, SessionSteerAdmissionInput, SessionTurnAdmission, SessionTurnAdmissionInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
import { activeSessionAdmissionConflict, assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, assertSessionCommandReplay, assertSessionSteerAdmissionContract, assertSessionTurnAdmissionContract, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, invalidSessionAdmissionProjection, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, receivableActiveSessionTurn, runCollectionItem, runCollectionResult, runCreatedEventPayload, runnerDispatchFailureKind, runnerDispatchFailureReason, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, staleActiveSessionTurn, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js";
import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js";
import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js";
import { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js";
import { assertRunnerDispatchReplay, attachRunnerDispatchIntent, newRunnerDispatchIntent } from "./runner-dispatch-intent.js";
import { durableDispatchAndKafkaOutboxMigrationSql, durableQueueStatusFromRow, kafkaEventOutboxFromRow, kafkaOutboxPendingKeyOrderMigrationSql, runnerDispatchIntentFromRow, runnerDispatchOutcomeMigrationSql } from "./postgres-durable-queues.js";
import { fenceActiveRunnerDispatchIntents, fenceTerminalRunnerDispatchIntents, lockRunnerDispatchClaim, staleClaimError } from "./postgres-runner-dispatch.js";

interface PostgresStoreOptions {
  connectionString: string;
  poolMax?: number;
  eventOutbox?: EventOutboxConfig;
  onWaitForRunChangeListening?: (runId: string) => Promise<void>;
}

interface MigrationDefinition {
  id: string;
  checksum: string;
  sql: string;
}

const initialMigrationSql = `
CREATE TABLE IF NOT EXISTS agentrun_runs (
  id text PRIMARY KEY,
  tenant_id text NOT NULL,
  project_id text NOT NULL,
  workspace_ref jsonb NOT NULL,
  provider_id text NOT NULL,
  backend_profile text NOT NULL,
  execution_policy jsonb NOT NULL,
  trace_sink jsonb,
  status text NOT NULL,
  terminal_status text,
  failure_kind text,
  failure_message text,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL,
  claimed_by text,
  lease_expires_at timestamptz
);

CREATE TABLE IF NOT EXISTS agentrun_commands (
  id text PRIMARY KEY,
  run_id text NOT NULL REFERENCES agentrun_runs(id) ON DELETE CASCADE,
  seq integer NOT NULL,
  type text NOT NULL,
  payload jsonb NOT NULL,
  payload_hash text NOT NULL,
  idempotency_key text,
  state text NOT NULL,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL,
  acknowledged_at timestamptz,
  UNIQUE (run_id, seq)
);

CREATE UNIQUE INDEX IF NOT EXISTS agentrun_commands_run_idempotency_key_idx
  ON agentrun_commands (run_id, idempotency_key)
  WHERE idempotency_key IS NOT NULL;

CREATE TABLE IF NOT EXISTS agentrun_events (
  id text PRIMARY KEY,
  run_id text NOT NULL REFERENCES agentrun_runs(id) ON DELETE CASCADE,
  seq integer NOT NULL,
  type text NOT NULL,
  payload jsonb NOT NULL,
  created_at timestamptz NOT NULL,
  UNIQUE (run_id, seq)
);

CREATE TABLE IF NOT EXISTS agentrun_runners (
  id text PRIMARY KEY,
  run_id text,
  attempt_id text,
  backend_profile text,
  placement text,
  source_commit text,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
  registered_at timestamptz NOT NULL,
  heartbeat_at timestamptz NOT NULL
);

CREATE TABLE IF NOT EXISTS agentrun_backends (
  profile text PRIMARY KEY,
  capabilities jsonb NOT NULL,
  capacity jsonb NOT NULL,
  health jsonb NOT NULL,
  updated_at timestamptz NOT NULL
);

CREATE TABLE IF NOT EXISTS agentrun_leases (
  run_id text PRIMARY KEY REFERENCES agentrun_runs(id) ON DELETE CASCADE,
  runner_id text NOT NULL,
  lease_expires_at timestamptz NOT NULL,
  stale_recovery_marker jsonb,
  updated_at timestamptz NOT NULL
);

CREATE INDEX IF NOT EXISTS agentrun_runs_status_idx ON agentrun_runs (status, updated_at);
CREATE INDEX IF NOT EXISTS agentrun_events_run_seq_idx ON agentrun_events (run_id, seq);
CREATE INDEX IF NOT EXISTS agentrun_commands_run_seq_idx ON agentrun_commands (run_id, seq);
CREATE INDEX IF NOT EXISTS agentrun_leases_expiry_idx ON agentrun_leases (lease_expires_at);

INSERT INTO agentrun_backends (profile, capabilities, capacity, health, updated_at)
VALUES (
  'codex',
  '{"protocol":"codex-app-server-jsonrpc-stdio","transport":"stdio","command":"codex app-server --listen stdio://"}'::jsonb,
  '{"mode":"manual-runner-v0.1"}'::jsonb,
  '{"status":"registered"}'::jsonb,
  now()
)
ON CONFLICT (profile) DO UPDATE SET
  capabilities = EXCLUDED.capabilities,
  capacity = EXCLUDED.capacity,
  health = EXCLUDED.health,
  updated_at = EXCLUDED.updated_at;
`;

const runChangeNotificationChannel = "agentrun_run_change";

const backendProfilesMigrationSql = `
INSERT INTO agentrun_backends (profile, capabilities, capacity, health, updated_at)
VALUES ${backendCapabilitiesSqlValues(["codex", "deepseek"])}
ON CONFLICT (profile) DO UPDATE SET
  capabilities = EXCLUDED.capabilities,
  capacity = EXCLUDED.capacity,
  health = EXCLUDED.health,
  updated_at = EXCLUDED.updated_at;
`;

const minimaxM3BackendProfileMigrationSql = `
INSERT INTO agentrun_backends (profile, capabilities, capacity, health, updated_at)
VALUES ${backendCapabilitiesSqlValues(["minimax-m3"])}
ON CONFLICT (profile) DO UPDATE SET
  capabilities = EXCLUDED.capabilities,
  capacity = EXCLUDED.capacity,
  health = EXCLUDED.health,
  updated_at = EXCLUDED.updated_at;
`;

const dsflashGoBackendProfileMigrationSql = `
UPDATE agentrun_runs SET backend_profile = 'dsflash-go' WHERE backend_profile = 'ofcx-go';
UPDATE agentrun_sessions SET backend_profile = 'dsflash-go' WHERE backend_profile = 'ofcx-go';
UPDATE agentrun_runners SET backend_profile = 'dsflash-go' WHERE backend_profile = 'ofcx-go';
UPDATE agentrun_queue_tasks SET backend_profile = 'dsflash-go' WHERE backend_profile = 'ofcx-go';
UPDATE agentrun_runs
SET execution_policy = replace(replace(replace(execution_policy::text,
  '"profile": "ofcx-go"',
  '"profile": "dsflash-go"'),
  'agentrun-v01-provider-ofcx-go',
  'agentrun-v01-provider-dsflash-go'),
  '/home/agentrun/.codex-ofcx-go',
  '/home/agentrun/.codex-dsflash-go')::jsonb
WHERE execution_policy::text LIKE '%ofcx-go%';
UPDATE agentrun_queue_tasks
SET execution_policy = replace(replace(replace(execution_policy::text,
  '"profile": "ofcx-go"',
  '"profile": "dsflash-go"'),
  'agentrun-v01-provider-ofcx-go',
  'agentrun-v01-provider-dsflash-go'),
  '/home/agentrun/.codex-ofcx-go',
  '/home/agentrun/.codex-dsflash-go')::jsonb
WHERE execution_policy IS NOT NULL AND execution_policy::text LIKE '%ofcx-go%';
DELETE FROM agentrun_backends WHERE profile = 'ofcx-go';
INSERT INTO agentrun_backends (profile, capabilities, capacity, health, updated_at)
VALUES ${backendCapabilitiesSqlValues(["dsflash-go"], { requiredSecretKeysByProfile: { "dsflash-go": ["auth.json", "config.toml"] } })}
ON CONFLICT (profile) DO UPDATE SET
  capabilities = EXCLUDED.capabilities,
  capacity = EXCLUDED.capacity,
  health = EXCLUDED.health,
  updated_at = EXCLUDED.updated_at;
`;

const dsflashGoModelCatalogBackendProfileMigrationSql = `
INSERT INTO agentrun_backends (profile, capabilities, capacity, health, updated_at)
VALUES ${backendCapabilitiesSqlValues(["dsflash-go"])}
ON CONFLICT (profile) DO UPDATE SET
  capabilities = EXCLUDED.capabilities,
  capacity = EXCLUDED.capacity,
  health = EXCLUDED.health,
  updated_at = EXCLUDED.updated_at;
`;

const sessionControlMigrationSql = `
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS version bigint NOT NULL DEFAULT 1;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS execution_state text NOT NULL DEFAULT 'idle';
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_run_id text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_command_id text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS active_run_id text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS active_command_id text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_event_seq integer NOT NULL DEFAULT 0;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS terminal_status text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS failure_kind text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS title text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS summary jsonb NOT NULL DEFAULT '{}'::jsonb;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_activity_at timestamptz;

CREATE SEQUENCE IF NOT EXISTS agentrun_session_version_seq;

SELECT setval(
  'agentrun_session_version_seq',
  GREATEST(
    COALESCE((SELECT MAX(version) FROM agentrun_sessions), 0),
    1
  ),
  true
);

CREATE TABLE IF NOT EXISTS agentrun_session_read_cursors (
  session_id text NOT NULL REFERENCES agentrun_sessions(session_id) ON DELETE CASCADE,
  reader_id text NOT NULL,
  session_version bigint NOT NULL,
  read_at timestamptz NOT NULL,
  PRIMARY KEY (session_id, reader_id)
);

CREATE INDEX IF NOT EXISTS agentrun_sessions_control_idx ON agentrun_sessions (execution_state, backend_profile, updated_at DESC);
CREATE INDEX IF NOT EXISTS agentrun_sessions_version_idx ON agentrun_sessions (version, updated_at DESC);
CREATE INDEX IF NOT EXISTS agentrun_runs_session_idx ON agentrun_runs ((session_ref->>'sessionId'), updated_at DESC);
`;

const sessionStateStorageMigrationSql = `
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_kind text NOT NULL DEFAULT 'none';
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_pvc_name text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_namespace text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_size_bytes bigint;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_files_count integer;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_sha256 text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_updated_at timestamptz;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS codex_rollout_subdir text NOT NULL DEFAULT 'sessions';
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_pvc_phase text;
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS storage_evicted_at timestamptz;

CREATE INDEX IF NOT EXISTS agentrun_sessions_storage_pvc_idx
  ON agentrun_sessions (storage_pvc_name)
  WHERE storage_pvc_name IS NOT NULL;

CREATE INDEX IF NOT EXISTS agentrun_sessions_storage_kind_idx
  ON agentrun_sessions (storage_kind, expires_at);
`;

const hwlabManualDispatchMigrationSql = `
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS session_ref jsonb;
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS resource_bundle_ref jsonb;

CREATE TABLE IF NOT EXISTS agentrun_sessions (
  session_id text PRIMARY KEY,
  tenant_id text NOT NULL,
  project_id text NOT NULL,
  backend_profile text NOT NULL,
  conversation_id text,
  thread_id text,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL,
  expires_at timestamptz
);

CREATE TABLE IF NOT EXISTS agentrun_runner_jobs (
  id text PRIMARY KEY,
  run_id text NOT NULL REFERENCES agentrun_runs(id) ON DELETE CASCADE,
  command_id text NOT NULL REFERENCES agentrun_commands(id) ON DELETE CASCADE,
  idempotency_key text,
  payload_hash text NOT NULL,
  attempt_id text NOT NULL,
  runner_id text NOT NULL,
  namespace text NOT NULL,
  job_name text NOT NULL,
  manager_url text NOT NULL,
  image text NOT NULL,
  source_commit text NOT NULL,
  service_account_name text,
  result jsonb NOT NULL,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL
);

CREATE UNIQUE INDEX IF NOT EXISTS agentrun_runner_jobs_run_idempotency_key_idx
  ON agentrun_runner_jobs (run_id, idempotency_key)
  WHERE idempotency_key IS NOT NULL;

CREATE INDEX IF NOT EXISTS agentrun_runner_jobs_run_command_idx ON agentrun_runner_jobs (run_id, command_id, created_at);
CREATE INDEX IF NOT EXISTS agentrun_sessions_tenant_project_idx ON agentrun_sessions (tenant_id, project_id, backend_profile, updated_at);
`;

const queueQ1MigrationSql = `
CREATE TABLE IF NOT EXISTS agentrun_queue_tasks (
  id text PRIMARY KEY,
  tenant_id text NOT NULL,
  project_id text NOT NULL,
  queue text NOT NULL,
  lane text NOT NULL,
  title text NOT NULL,
  priority integer NOT NULL,
  state text NOT NULL,
  version bigint NOT NULL,
  backend_profile text NOT NULL,
  provider_id text,
  workspace_ref jsonb,
  execution_policy jsonb,
  resource_bundle_ref jsonb,
  payload jsonb NOT NULL,
  payload_hash text NOT NULL,
  references_json jsonb NOT NULL DEFAULT '[]'::jsonb,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
  idempotency_key text,
  latest_attempt jsonb,
  session_path text,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL,
  cancelled_at timestamptz,
  cancel_reason text
);

CREATE UNIQUE INDEX IF NOT EXISTS agentrun_queue_tasks_idempotency_idx
  ON agentrun_queue_tasks (tenant_id, project_id, idempotency_key)
  WHERE idempotency_key IS NOT NULL;

CREATE INDEX IF NOT EXISTS agentrun_queue_tasks_list_idx ON agentrun_queue_tasks (queue, state, version, priority, created_at);
CREATE INDEX IF NOT EXISTS agentrun_queue_tasks_updated_idx ON agentrun_queue_tasks (version, updated_at);

CREATE TABLE IF NOT EXISTS agentrun_queue_read_cursors (
  task_id text NOT NULL REFERENCES agentrun_queue_tasks(id) ON DELETE CASCADE,
  reader_id text NOT NULL,
  task_version bigint NOT NULL,
  read_at timestamptz NOT NULL,
  PRIMARY KEY (task_id, reader_id)
);

CREATE SEQUENCE IF NOT EXISTS agentrun_queue_version_seq;
`;

const queueSessionRefMigrationSql = `
ALTER TABLE agentrun_queue_tasks
  ADD COLUMN IF NOT EXISTS session_ref jsonb;
`;

const cancelLifecycleMigrationSql = `
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS cancel_epoch integer NOT NULL DEFAULT 0;
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS cancel_request_id text;
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS cancel_requested_at timestamptz;
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS cancel_reason text;

ALTER TABLE agentrun_commands ADD COLUMN IF NOT EXISTS cancel_epoch integer NOT NULL DEFAULT 0;
ALTER TABLE agentrun_commands ADD COLUMN IF NOT EXISTS cancel_request_id text;
ALTER TABLE agentrun_commands ADD COLUMN IF NOT EXISTS cancel_requested_at timestamptz;
ALTER TABLE agentrun_commands ADD COLUMN IF NOT EXISTS cancel_reason text;

CREATE TABLE IF NOT EXISTS agentrun_cancel_requests (
  id text PRIMARY KEY,
  target_kind text NOT NULL,
  target_id text NOT NULL,
  run_id text REFERENCES agentrun_runs(id) ON DELETE CASCADE,
  command_id text REFERENCES agentrun_commands(id) ON DELETE SET NULL,
  session_id text,
  task_id text,
  reason text NOT NULL,
  requested_by text,
  epoch integer NOT NULL,
  stage text NOT NULL,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL
);

CREATE INDEX IF NOT EXISTS agentrun_cancel_requests_target_idx ON agentrun_cancel_requests (target_kind, target_id, created_at DESC);
CREATE INDEX IF NOT EXISTS agentrun_cancel_requests_run_idx ON agentrun_cancel_requests (run_id, created_at DESC);
CREATE INDEX IF NOT EXISTS agentrun_cancel_requests_command_idx ON agentrun_cancel_requests (command_id, created_at DESC);
`;

const queueAttemptsMigrationSql = `
CREATE TABLE IF NOT EXISTS agentrun_queue_attempts (
  attempt_id text NOT NULL,
  task_id text NOT NULL REFERENCES agentrun_queue_tasks(id) ON DELETE CASCADE,
  retry_index integer NOT NULL CHECK (retry_index >= 0),
  state text NOT NULL CHECK (state IN ('reserved', 'pending', 'running', 'completed', 'failed', 'blocked', 'cancelled')),
  idempotency_key text,
  reason text,
  payload_hash text NOT NULL,
  previous_attempt_id text,
  run_id text,
  command_id text,
  runner_job_id text,
  session_id text,
  session_path text,
  failure_kind text,
  failure_message text,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL,
  activated_at timestamptz,
  terminal_at timestamptz,
  PRIMARY KEY (task_id, attempt_id),
  UNIQUE (task_id, retry_index)
);

CREATE UNIQUE INDEX IF NOT EXISTS agentrun_queue_attempts_retry_key_idx
  ON agentrun_queue_attempts (task_id, idempotency_key)
  WHERE idempotency_key IS NOT NULL;

CREATE UNIQUE INDEX IF NOT EXISTS agentrun_queue_attempts_active_idx
  ON agentrun_queue_attempts (task_id)
  WHERE state IN ('reserved', 'running');

CREATE INDEX IF NOT EXISTS agentrun_queue_attempts_history_idx
  ON agentrun_queue_attempts (task_id, retry_index DESC, created_at DESC);

INSERT INTO agentrun_queue_attempts (
  attempt_id, task_id, retry_index, state, idempotency_key, reason, payload_hash,
  previous_attempt_id, run_id, command_id, runner_job_id, session_id, session_path,
  failure_kind, failure_message, created_at, updated_at, activated_at, terminal_at
)
SELECT
  latest_attempt->>'attemptId', id, 0, latest_attempt->>'state', NULL, NULL, payload_hash,
  NULL, latest_attempt->>'runId', latest_attempt->>'commandId', latest_attempt->>'runnerJobId',
  latest_attempt->>'sessionId', latest_attempt->>'sessionPath', latest_attempt->>'failureKind', latest_attempt->>'failureMessage',
  created_at, updated_at, updated_at,
  CASE WHEN latest_attempt->>'state' IN ('completed', 'failed', 'blocked', 'cancelled') THEN updated_at ELSE NULL END
FROM agentrun_queue_tasks
WHERE latest_attempt IS NOT NULL
  AND NULLIF(latest_attempt->>'attemptId', '') IS NOT NULL
ON CONFLICT (task_id, attempt_id) DO NOTHING;
`;

const postgresMigrations: MigrationDefinition[] = [
  {
    id: "001_v01_initial_durable_store",
    checksum: checksumSql(initialMigrationSql),
    sql: initialMigrationSql,
  },
  {
    id: "002_v01_backend_profiles",
    checksum: checksumSql(backendProfilesMigrationSql),
    sql: backendProfilesMigrationSql,
  },
  {
    id: "003_v01_hwlab_manual_dispatch",
    checksum: checksumSql(hwlabManualDispatchMigrationSql),
    sql: hwlabManualDispatchMigrationSql,
  },
  {
    id: "004_v01_queue_q1",
    checksum: checksumSql(queueQ1MigrationSql),
    sql: queueQ1MigrationSql,
  },
  {
    id: "005_v01_minimax_m3_backend_profile",
    checksum: checksumSql(minimaxM3BackendProfileMigrationSql),
    sql: minimaxM3BackendProfileMigrationSql,
  },
  {
    id: "006_v01_session_control",
    checksum: checksumSql(sessionControlMigrationSql),
    sql: sessionControlMigrationSql,
  },
  {
    id: "007_v01_session_state_storage",
    checksum: checksumSql(sessionStateStorageMigrationSql),
    sql: sessionStateStorageMigrationSql,
  },
  {
    id: "008_v01_dsflash_go_backend_profile",
    checksum: checksumSql(dsflashGoBackendProfileMigrationSql),
    sql: dsflashGoBackendProfileMigrationSql,
  },
  {
    id: "009_v01_dsflash_go_model_catalog",
    checksum: checksumSql(dsflashGoModelCatalogBackendProfileMigrationSql),
    sql: dsflashGoModelCatalogBackendProfileMigrationSql,
  },
  {
    id: "010_v01_queue_session_ref",
    checksum: checksumSql(queueSessionRefMigrationSql),
    sql: queueSessionRefMigrationSql,
  },
  {
    id: "011_v01_cancel_lifecycle",
    checksum: checksumSql(cancelLifecycleMigrationSql),
    sql: cancelLifecycleMigrationSql,
  },
  {
    id: "012_v02_durable_dispatch_kafka_outbox",
    checksum: checksumSql(durableDispatchAndKafkaOutboxMigrationSql),
    sql: durableDispatchAndKafkaOutboxMigrationSql,
  },
  {
    id: "013_v02_runner_dispatch_outcome",
    checksum: checksumSql(runnerDispatchOutcomeMigrationSql),
    sql: runnerDispatchOutcomeMigrationSql,
  },
  {
    id: "014_v02_queue_attempt_history_retry",
    checksum: checksumSql(queueAttemptsMigrationSql),
    sql: queueAttemptsMigrationSql,
  },
  {
    id: "015_v02_kafka_outbox_pending_key_order",
    checksum: checksumSql(kafkaOutboxPendingKeyOrderMigrationSql),
    sql: kafkaOutboxPendingKeyOrderMigrationSql,
  },
];

export function postgresMigrationContract(): JsonRecord {
  return {
    migrationIds: postgresMigrations.map((migration) => migration.id),
    latestMigrationId: latestMigrationId(),
    requiredTables: ["agentrun_schema_migrations", "agentrun_runs", "agentrun_commands", "agentrun_events", "agentrun_runners", "agentrun_backends", "agentrun_leases", "agentrun_sessions", "agentrun_session_read_cursors", "agentrun_runner_jobs", "agentrun_runner_dispatch_intents", "agentrun_kafka_event_outbox", "agentrun_queue_tasks", "agentrun_queue_attempts", "agentrun_queue_read_cursors", "agentrun_cancel_requests"],
    queueRetryConcurrency: {
      attemptIdentity: ["task_id", "attempt_id"],
      retryIdentity: ["task_id", "idempotency_key"],
      activeAttemptFence: ["task_id", "state IN (reserved,running)"],
      retryIndexIdentity: ["task_id", "retry_index"],
      taskReservationLock: "SELECT ... FOR UPDATE",
      legacyBackfill: "agentrun_queue_tasks.latest_attempt",
      failureVisibility: ["failure_kind", "failure_message"],
      valuesPrinted: false,
    },
    kafkaOutbox: {
      pendingKeyOrderIndexMigrationId: "015_v02_kafka_outbox_pending_key_order",
      pendingKeyOrderIndex: "agentrun_kafka_event_outbox_pending_key_order_idx",
      relayParallelismBound: "AGENTRUN_KAFKA_OUTBOX_BATCH_SIZE",
      valuesPrinted: false,
    },
    checksums: Object.fromEntries(postgresMigrations.map((migration) => [migration.id, migration.checksum])),
  };
}

export async function createPostgresAgentRunStore(options: PostgresStoreOptions): Promise<PostgresAgentRunStore> {
  const store = new PostgresAgentRunStore(options);
  await store.migrate();
  return store;
}

export class PostgresAgentRunStore implements AgentRunStore {
  private readonly pool: Pool;
  private readonly runnerCreateFencePool: Pool;
  private readonly eventOutboxConfig: EventOutboxConfig;
  private readonly onWaitForRunChangeListening: ((runId: string) => Promise<void>) | null;
  private migrationReady = false;
  private appliedMigrationId: string | null = null;

  constructor(options: PostgresStoreOptions) {
    this.pool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-v01", ...(options.poolMax === undefined ? {} : { max: options.poolMax }) });
    this.runnerCreateFencePool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-runner-capacity-fence", max: Math.max(2, options.poolMax ?? 10) });
    this.eventOutboxConfig = options.eventOutbox ?? { enabled: false, topic: null, source: null };
    this.onWaitForRunChangeListening = options.onWaitForRunChangeListening ?? null;
  }

  async migrate(): Promise<void> {
    await this.withTransaction(async (client) => {
      await client.query(`
CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
  id text PRIMARY KEY,
  checksum text NOT NULL,
  applied_at timestamptz NOT NULL DEFAULT now()
)
`);
      for (const migration of postgresMigrations) {
        const existing = await client.query<{ checksum: string }>("SELECT checksum FROM agentrun_schema_migrations WHERE id = $1 FOR UPDATE", [migration.id]);
        if (existing.rowCount && existing.rows[0]?.checksum !== migration.checksum) {
          throw new AgentRunError("infra-failed", `schema migration checksum mismatch for ${migration.id}`, { httpStatus: 503, details: { migrationId: migration.id } });
        }
        if (!existing.rowCount) {
          await client.query(migration.sql);
          await client.query("INSERT INTO agentrun_schema_migrations (id, checksum) VALUES ($1, $2)", [migration.id, migration.checksum]);
        }
      }
    });
    this.migrationReady = true;
    this.appliedMigrationId = latestMigrationId();
  }

  async health(): Promise<StoreHealth> {
    try {
      await this.pool.query("SELECT 1");
      const result = await this.pool.query<{ id: string }>("SELECT id FROM agentrun_schema_migrations ORDER BY applied_at DESC, id DESC LIMIT 1");
      const migrationId = result.rows[0]?.id ?? null;
      const migrationReady = this.migrationReady && migrationId === latestMigrationId();
      return { adapter: "postgres", ready: migrationReady, reachable: true, migrationReady, migrationId, failureKind: migrationReady ? null : "infra-failed", message: migrationReady ? null : "schema migration is not ready", credentialValuesPrinted: false };
    } catch (error) {
      return { adapter: "postgres", ready: false, reachable: false, migrationReady: false, migrationId: this.appliedMigrationId, failureKind: "infra-failed", message: error instanceof Error ? error.message : String(error), credentialValuesPrinted: false };
    }
  }

  async createRun(input: CreateRunInput, identity?: CreateRunIdentity): Promise<RunRecord> {
    return this.withTransaction(async (client) => await this.createRunWithClient(client, input, identity));
  }

  async getRun(runId: string): Promise<RunRecord> {
    const result = await this.pool.query("SELECT * FROM agentrun_runs WHERE id = $1", [runId]);
    const row = result.rows[0];
    if (!row) throw new AgentRunError("schema-invalid", `run ${runId} was not found`, { httpStatus: 404 });
    return runFromRow(row);
  }

  async listRuns(input: ListRunsInput): Promise<RunCollectionResult> {
    const parameters: unknown[] = [];
    const conditions: string[] = [];
    const equalFilter = (column: string, value: string | undefined) => {
      if (!value) return;
      parameters.push(value);
      conditions.push(`${column} = $${parameters.length}`);
    };
    if (input.statuses?.length) {
      parameters.push(input.statuses);
      conditions.push(`r.status = ANY($${parameters.length}::text[])`);
    }
    equalFilter("r.project_id", input.projectId);
    equalFilter("r.workspace_ref->>'path'", input.workspace);
    equalFilter("r.backend_profile", input.backendProfile);
    equalFilter("r.provider_id", input.providerId);
    if (input.updatedAfter) {
      parameters.push(input.updatedAfter);
      conditions.push(`r.updated_at >= $${parameters.length}::timestamptz`);
    }
    if (input.search) {
      parameters.push(`%${input.search.toLowerCase()}%`);
      const index = parameters.length;
      conditions.push(`LOWER(CONCAT_WS(E'\\n', r.id, r.project_id, r.workspace_ref->>'path', r.backend_profile, r.provider_id, r.session_ref->>'sessionId', r.session_ref->'metadata'->>'title', r.trace_sink->>'taskId', r.trace_sink->>'lane', r.failure_kind, r.failure_message)) LIKE $${index}`);
    }
    parameters.push(input.pageSize, (input.page - 1) * input.pageSize);
    const limitIndex = parameters.length - 1;
    const offsetIndex = parameters.length;
    const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
    const orderBy = input.sort === "updated-asc" ? "updated_at ASC, id ASC"
      : input.sort === "status-asc" ? "status ASC, updated_at DESC, id DESC"
      : input.sort === "id-asc" ? "id ASC"
      : "updated_at DESC, id DESC";
    const qualifiedOrderBy = orderBy.replaceAll("updated_at", "p.updated_at").replaceAll("status", "p.status").replaceAll("id", "p.id");
    const result = await this.pool.query(`
      WITH filtered AS (
        SELECT r.*
        FROM agentrun_runs r
        ${where}
      ), total AS (
        SELECT COUNT(*)::text AS total_count FROM filtered
      ), facets AS (
        SELECT jsonb_build_object(
          'projects', COALESCE((SELECT jsonb_agg(project_id ORDER BY project_id) FROM (SELECT DISTINCT project_id FROM agentrun_runs) values_by_project), '[]'::jsonb),
          'workspaces', COALESCE((SELECT jsonb_agg(workspace ORDER BY workspace) FROM (SELECT DISTINCT workspace_ref->>'path' AS workspace FROM agentrun_runs WHERE workspace_ref->>'path' IS NOT NULL) values_by_workspace), '[]'::jsonb),
          'backends', COALESCE((SELECT jsonb_agg(backend_profile ORDER BY backend_profile) FROM (SELECT DISTINCT backend_profile FROM agentrun_runs) values_by_backend), '[]'::jsonb),
          'providers', COALESCE((SELECT jsonb_agg(provider_id ORDER BY provider_id) FROM (SELECT DISTINCT provider_id FROM agentrun_runs) values_by_provider), '[]'::jsonb),
          'statusCounts', COALESCE((SELECT jsonb_object_agg(status, count) FROM (SELECT status, COUNT(*)::int AS count FROM agentrun_runs GROUP BY status) values_by_status), '{}'::jsonb)
        ) AS value
      ), paged AS (
        SELECT * FROM filtered ORDER BY ${orderBy} LIMIT $${limitIndex} OFFSET $${offsetIndex}
      )
      SELECT p.*, total.total_count, facets.value AS facets, row_to_json(c.*) AS latest_command, row_to_json(j.*) AS latest_runner_job
      FROM total CROSS JOIN facets
      LEFT JOIN paged p ON TRUE
      LEFT JOIN LATERAL (
        SELECT * FROM agentrun_commands command_row WHERE command_row.run_id = p.id ORDER BY command_row.seq DESC LIMIT 1
      ) c ON TRUE
      LEFT JOIN LATERAL (
        SELECT * FROM agentrun_runner_jobs job_row WHERE job_row.run_id = p.id ORDER BY job_row.created_at DESC, job_row.id DESC LIMIT 1
      ) j ON TRUE
      ORDER BY ${qualifiedOrderBy} NULLS LAST
    `, parameters);
    const total = Number(result.rows[0]?.total_count ?? 0);
    const items = result.rows
      .filter((row) => row.id !== null && row.id !== undefined)
      .map((row) => runCollectionItem(
        runFromRow(row),
        row.latest_command ? commandFromRow(jsonRecord(row.latest_command)) : null,
        row.latest_runner_job ? runnerJobFromRow(jsonRecord(row.latest_runner_job)) : null,
      ));
    return runCollectionResult(input, items, total, jsonRecord(result.rows[0]?.facets));
  }

  async listEvents(runId: string, afterSeq: number, limit: number): Promise<RunEvent[]> {
    await this.getRun(runId);
    const result = await this.pool.query("SELECT * FROM agentrun_events WHERE run_id = $1 AND seq > $2 ORDER BY seq ASC LIMIT $3", [runId, afterSeq, clamp(limit, 1, 500)]);
    return result.rows.map(eventFromRow);
  }

  async countEvents(runId: string, afterSeq: number): Promise<number> {
    await this.getRun(runId);
    const result = await this.pool.query<{ count: string }>("SELECT COUNT(*)::text AS count FROM agentrun_events WHERE run_id = $1 AND seq > $2", [runId, afterSeq]);
    return Number(result.rows[0]?.count ?? 0);
  }

  async waitForRunChange(runId: string, afterSeq: number, timeoutMs: number, signal?: AbortSignal): Promise<"changed" | "timeout" | "aborted"> {
    await this.getRun(runId);
    if (signal?.aborted) return "aborted";
    const client = await this.pool.connect();
    let notificationHandler: ((message: import("pg").Notification) => void) | null = null;
    let abortHandler: (() => void) | null = null;
    let timer: ReturnType<typeof setTimeout> | null = null;
    try {
      await client.query(`LISTEN ${runChangeNotificationChannel}`);
      let finish: (result: "changed" | "timeout" | "aborted") => void = () => {};
      const wait = new Promise<"changed" | "timeout" | "aborted">((resolve) => {
        let settled = false;
        finish = (result) => {
          if (settled) return;
          settled = true;
          resolve(result);
        };
        notificationHandler = (message) => {
          if (message.channel === runChangeNotificationChannel && message.payload === runId) finish("changed");
        };
        abortHandler = () => finish("aborted");
        timer = setTimeout(() => finish("timeout"), timeoutMs);
        client.on("notification", notificationHandler);
        signal?.addEventListener("abort", abortHandler, { once: true });
        if (signal?.aborted) finish("aborted");
      });
      await this.onWaitForRunChangeListening?.(runId);
      const changed = await client.query("SELECT EXISTS(SELECT 1 FROM agentrun_events WHERE run_id = $1 AND seq > $2) AS changed", [runId, afterSeq]);
      if (changed.rows[0]?.changed === true) finish("changed");
      return await wait;
    } finally {
      if (timer !== null) clearTimeout(timer);
      if (abortHandler !== null) signal?.removeEventListener("abort", abortHandler);
      if (notificationHandler !== null) client.removeListener("notification", notificationHandler);
      try {
        await client.query(`UNLISTEN ${runChangeNotificationChannel}`);
      } finally {
        client.release();
      }
    }
  }

  async listEventsForCommand(runId: string, commandId: string, limit: number): Promise<RunEvent[]> {
    await this.getRun(runId);
    const result = await this.pool.query(
      `SELECT * FROM agentrun_events
       WHERE run_id = $1
         AND (
           payload->>'commandId' = $2
           OR payload->>'targetCommandId' = $2
           OR (type = 'terminal_status' AND NOT (payload ? 'commandId'))
         )
       ORDER BY seq ASC
       LIMIT $3`,
      [runId, commandId, clamp(limit, 1, 2_000)],
    );
    return result.rows.map(eventFromRow);
  }

  async createCommand(runId: string, input: CreateCommandInput): Promise<CommandRecord> {
    return this.withTransaction(async (client) => await this.createCommandWithClient(client, runId, input));
  }

  async admitSessionSteer(input: SessionSteerAdmissionInput): Promise<SessionSteerAdmission | null> {
    assertSessionSteerAdmissionContract(input);
    return this.withTransaction(async (client) => {
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-session-turn-admission", input.sessionId]);
      const observedSessionResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1", [input.sessionId]);
      const observedSession = observedSessionResult.rows[0] ? sessionFromRow(observedSessionResult.rows[0]) : null;
      if (!observedSession?.activeRunId || !observedSession.activeCommandId) return null;
      const runResult = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [observedSession.activeRunId]);
      const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [observedSession.activeCommandId]);
      if (!runResult.rows[0] || !commandResult.rows[0]) return null;
      const run = runFromRow(runResult.rows[0]);
      const targetCommand = commandFromRow(commandResult.rows[0]);
      const sessionResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]);
      const session = sessionResult.rows[0] ? sessionFromRow(sessionResult.rows[0]) : null;
      if (!session) return null;
      const receivable = receivableActiveSessionTurn(session, run, targetCommand);
      if (!receivable) return null;
      const command = await this.createCommandWithClient(client, run.id, input.command);
      return { run, command, targetCommand, ...receivable };
    });
  }

  async admitSessionTurn(input: SessionTurnAdmissionInput): Promise<SessionTurnAdmission> {
    assertSessionTurnAdmissionContract(input);
    return this.withTransaction(async (client) => {
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-session-turn-admission", input.sessionId]);
      const sessionResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]);
      const session = sessionResult.rows[0] ? sessionFromRow(sessionResult.rows[0]) : null;
      if (session) assertSessionBoundary(session, input.run);

      if (input.command.idempotencyKey) {
        const replayResult = await client.query(
          `SELECT c.* FROM agentrun_commands c
           JOIN agentrun_runs r ON r.id = c.run_id
           WHERE r.session_ref->>'sessionId' = $1 AND c.idempotency_key = $2
           ORDER BY c.created_at DESC LIMIT 1 FOR UPDATE OF r, c`,
          [input.sessionId, input.command.idempotencyKey],
        );
        if (replayResult.rows[0]) {
          const command = commandFromRow(replayResult.rows[0]);
          const run = await this.requireRunForUpdate(client, command.runId);
          return await this.replayOrRecoverSessionTurnWithClient(client, run, command, input, "replayed");
        }
      }

      if (session?.activeRunId || session?.activeCommandId) {
        if (!session.activeRunId || !session.activeCommandId) throw invalidSessionAdmissionProjection(input.sessionId);
        const runResult = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [session.activeRunId]);
        const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [session.activeCommandId]);
        if (!runResult.rows[0] || !commandResult.rows[0]) throw invalidSessionAdmissionProjection(input.sessionId);
        const run = runFromRow(runResult.rows[0]);
        const command = commandFromRow(commandResult.rows[0]);
        if (command.runId !== run.id || run.sessionRef?.sessionId !== input.sessionId) throw invalidSessionAdmissionProjection(input.sessionId);
        if (!isTerminalRunStatus(run.status) && !isTerminalCommandState(command.state)) {
          try {
            return await this.replayOrRecoverSessionTurnWithClient(client, run, command, input, "replayed");
          } catch (error) {
            if (error instanceof AgentRunError && error.failureKind === "schema-invalid") {
              if (!staleActiveSessionTurn(session, run, command)) throw activeSessionAdmissionConflict(input.sessionId, run, command);
              await this.terminalizeStaleSessionAdmissionWithClient(client, run, command);
            } else {
              throw error;
            }
          }
        }
      }

      const run = await this.createRunWithClient(client, input.run);
      const command = await this.createCommandWithClient(client, run.id, input.command);
      return { run, command, disposition: "created", recoveredPriorPartialWrite: false, partialWrite: false };
    });
  }

  async getCommand(commandId: string): Promise<CommandRecord> {
    const result = await this.pool.query("SELECT * FROM agentrun_commands WHERE id = $1", [commandId]);
    const row = result.rows[0];
    if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
    return commandFromRow(row);
  }

  async listCommands(runId: string, afterSeq: number, limit: number): Promise<CommandRecord[]> {
    await this.getRun(runId);
    const result = await this.pool.query("SELECT * FROM agentrun_commands WHERE run_id = $1 AND seq > $2 ORDER BY seq ASC LIMIT $3", [runId, afterSeq, clamp(limit, 1, 100)]);
    return result.rows.map(commandFromRow);
  }

  async getRunnerDispatchIntent(commandId: string): Promise<RunnerDispatchIntentRecord | null> {
    const result = await this.pool.query("SELECT * FROM agentrun_runner_dispatch_intents WHERE command_id = $1", [commandId]);
    return result.rows[0] ? runnerDispatchIntentFromRow(result.rows[0]) : null;
  }

  async claimRunnerDispatchIntents(input: { owner: string; leaseMs: number; limit: number }): Promise<RunnerDispatchIntentRecord[]> {
    const leaseExpiresAt = new Date(Date.now() + input.leaseMs).toISOString();
    return this.withTransaction(async (client) => {
      await fenceTerminalRunnerDispatchIntents(client);
      const result = await client.query(
        `WITH due AS (
           SELECT i.id FROM agentrun_runner_dispatch_intents i
           JOIN agentrun_runs r ON r.id = i.run_id JOIN agentrun_commands c ON c.id = i.command_id
           WHERE i.next_attempt_at <= now()
             AND r.status NOT IN ('completed', 'failed', 'blocked', 'cancelled') AND c.state NOT IN ('completed', 'failed', 'cancelled')
             AND (i.state IN ('pending', 'retry') OR (i.state = 'dispatching' AND (i.lease_expires_at IS NULL OR i.lease_expires_at <= now())))
           ORDER BY i.next_attempt_at ASC, i.created_at ASC FOR UPDATE OF r, c, i SKIP LOCKED LIMIT $1
         )
         UPDATE agentrun_runner_dispatch_intents i
         SET state = 'dispatching', attempt_count = i.attempt_count + 1, lease_owner = $2, lease_expires_at = $3, updated_at = now()
         FROM due WHERE i.id = due.id RETURNING i.*`,
        [clamp(input.limit, 1, 500), input.owner, leaseExpiresAt],
      );
      return result.rows.map(runnerDispatchIntentFromRow);
    });
  }

  async completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): Promise<RunnerDispatchIntentRecord> {
    const settled = await this.withTransaction(async (client) => {
      const locked = await lockRunnerDispatchClaim(client, claim);
      if (locked.fenced) return runnerDispatchIntentFromRow(locked.intentRow);
      const result = await client.query(
        `UPDATE agentrun_runner_dispatch_intents
         SET state = 'dispatched', dispatch_outcome = $2, actual_runner_job_id = $3, active_runner_id = $4,
             lease_owner = NULL, lease_expires_at = NULL, last_error = NULL, dispatched_at = now(), updated_at = now()
         WHERE id = $1 RETURNING *`,
        [claim.id, completion.dispatchOutcome, completion.actualRunnerJobId, completion.activeRunnerId],
      );
      const intent = runnerDispatchIntentFromRow(result.rows[0]);
      await this.appendEventWithLockedRun(client, intent.runId, "backend_status", eventPayload);
      return intent;
    });
    if (settled.state === "cancelled") throw staleClaimError("runner dispatch", claim);
    return settled;
  }

  async retryRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string, eventPayload: JsonRecord): Promise<RunnerDispatchIntentRecord> {
    const settled = await this.withTransaction(async (client) => {
      const locked = await lockRunnerDispatchClaim(client, claim);
      if (locked.fenced) return runnerDispatchIntentFromRow(locked.intentRow);
      const result = await client.query(
        `UPDATE agentrun_runner_dispatch_intents
         SET state = 'retry', lease_owner = NULL, lease_expires_at = NULL, next_attempt_at = $2, last_error = $3::jsonb, updated_at = now()
         WHERE id = $1 RETURNING *`,
        [claim.id, nextAttemptAt, JSON.stringify(redactJson(error))],
      );
      const intent = runnerDispatchIntentFromRow(result.rows[0]);
      await this.appendEventWithLockedRun(client, intent.runId, "backend_status", eventPayload);
      return intent;
    });
    if (settled.state === "cancelled") throw staleClaimError("runner dispatch", claim);
    return settled;
  }

  async terminalizeRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord): Promise<JsonRecord> {
    const settled = await this.withTransaction(async (client) => {
      const locked = await lockRunnerDispatchClaim(client, claim);
      const intent = runnerDispatchIntentFromRow(locked.intentRow);
      const command = commandFromRow(locked.commandRow);
      const run = runFromRow(locked.runRow);
      if (locked.fenced) return { intent, command, run, dispatchFenced: true, valuesPrinted: false };
      const failure = redactJson(error);
      const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed";
      const failureKind = runnerDispatchFailureKind(failure.failureKind);
      const reason = runnerDispatchFailureReason(failure.reason);
      const at = nowIso();
      const updatedIntentResult = await client.query(
        `UPDATE agentrun_runner_dispatch_intents
         SET state = 'failed', lease_owner = NULL, lease_expires_at = NULL, next_attempt_at = $2,
             last_error = $3::jsonb, updated_at = $2
         WHERE id = $1 RETURNING *`,
        [intent.id, at, JSON.stringify(failure)],
      );
      let updatedCommand = command;
      if (!isTerminalCommandState(command.state)) {
        const updated = await client.query("UPDATE agentrun_commands SET state = 'failed', updated_at = $2 WHERE id = $1 RETURNING *", [command.id, at]);
        updatedCommand = commandFromRow(updated.rows[0]);
      }
      let updatedRun = run;
      if (!isTerminalRunStatus(run.status)) {
        const updated = await client.query(
          `UPDATE agentrun_runs SET status = 'failed', terminal_status = 'failed', failure_kind = $2, failure_message = $3, updated_at = $4
           WHERE id = $1 RETURNING *`,
          [run.id, failureKind, message, at],
        );
        updatedRun = runFromRow(updated.rows[0]);
        await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, "run terminalized by runner dispatch failure", at);
        await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "runner-dispatch-failed", reason, commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind, message, valuesPrinted: false });
        await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", reason, commandId: command.id, state: "failed", terminalStatus: "failed", failureKind, message, threadId: null, turnId: null });
        await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", reason, failureKind, message });
        await this.touchSessionForRun(client, updatedRun, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, lastCommandId: command.id, terminalStatus: "failed", failureKind, lastActivityAt: at }, { bumpVersion: true, at });
      }
      return { intent: runnerDispatchIntentFromRow(updatedIntentResult.rows[0]), command: updatedCommand, run: updatedRun, valuesPrinted: false };
    });
    if (settled.dispatchFenced === true) throw staleClaimError("runner dispatch", claim);
    return settled;
  }

  async runnerDispatchStatus(): Promise<JsonRecord> {
    const result = await this.pool.query(
      `SELECT count(*)::int AS total_count,
              count(*) FILTER (WHERE state IN ('pending', 'dispatching', 'retry'))::int AS backlog_count,
              count(*) FILTER (WHERE state = 'retry')::int AS retrying_count,
              min(created_at) FILTER (WHERE state IN ('pending', 'dispatching', 'retry')) AS oldest_pending_at,
              COALESCE(max(attempt_count), 0)::int AS max_attempt_count
       FROM agentrun_runner_dispatch_intents`,
    );
    return durableQueueStatusFromRow(result.rows[0], "runner-dispatch");
  }

  async claimKafkaEventOutbox(input: { owner: string; leaseMs: number; limit: number }): Promise<KafkaEventOutboxRecord[]> {
    const result = await this.pool.query(
       `WITH head_ids AS MATERIALIZED (
         SELECT DISTINCT ON (topic, partition_key) id
         FROM agentrun_kafka_event_outbox
         WHERE delivered_at IS NULL
         ORDER BY topic, partition_key, outbox_seq
       ), heads AS MATERIALIZED (
         SELECT candidate.topic, candidate.partition_key, candidate.outbox_seq
         FROM agentrun_kafka_event_outbox candidate
         JOIN head_ids ON head_ids.id = candidate.id
         WHERE candidate.next_attempt_at <= now()
           AND (candidate.lease_expires_at IS NULL OR candidate.lease_expires_at <= now())
         ORDER BY candidate.outbox_seq
         FOR UPDATE OF candidate SKIP LOCKED
         LIMIT $1
       ), prefix_candidates AS MATERIALIZED (
         SELECT candidate.id,
                candidate.topic,
                candidate.partition_key,
                candidate.outbox_seq,
                candidate.next_attempt_at,
                candidate.lease_expires_at
         FROM heads
         CROSS JOIN LATERAL (
           SELECT item.*
           FROM agentrun_kafka_event_outbox item
           WHERE item.topic = heads.topic
             AND item.partition_key = heads.partition_key
             AND item.delivered_at IS NULL
           ORDER BY item.outbox_seq
           LIMIT $1
         ) candidate
       ), ranked AS MATERIALIZED (
         SELECT prefix_candidates.id,
                prefix_candidates.outbox_seq,
                row_number() OVER (PARTITION BY prefix_candidates.topic, prefix_candidates.partition_key ORDER BY prefix_candidates.outbox_seq) AS key_offset,
                bool_and(prefix_candidates.next_attempt_at <= now() AND (prefix_candidates.lease_expires_at IS NULL OR prefix_candidates.lease_expires_at <= now())) OVER (PARTITION BY prefix_candidates.topic, prefix_candidates.partition_key ORDER BY prefix_candidates.outbox_seq) AS prefix_due
         FROM prefix_candidates
       ), due AS MATERIALIZED (
         SELECT candidate.id
         FROM agentrun_kafka_event_outbox candidate
         JOIN ranked ON ranked.id = candidate.id
         WHERE ranked.prefix_due
         ORDER BY ranked.key_offset ASC, candidate.outbox_seq ASC
         FOR UPDATE OF candidate SKIP LOCKED
         LIMIT $1
       )
       UPDATE agentrun_kafka_event_outbox o
       SET attempt_count = o.attempt_count + 1, lease_owner = $2, lease_expires_at = clock_timestamp() + ($3 * interval '1 millisecond'), updated_at = clock_timestamp()
       FROM due WHERE o.id = due.id
       RETURNING o.*`,
      [clamp(input.limit, 1, 500), input.owner, input.leaseMs],
    );
    return result.rows.map(kafkaEventOutboxFromRow);
  }

  async completeKafkaEventOutbox(claim: DurableQueueClaim): Promise<KafkaEventOutboxRecord> {
    const result = await this.pool.query(
      `UPDATE agentrun_kafka_event_outbox
       SET delivered_at = now(), lease_owner = NULL, lease_expires_at = NULL, last_error = NULL, updated_at = now()
       WHERE id = $1 AND delivered_at IS NULL AND lease_owner = $2 AND attempt_count = $3 AND lease_expires_at > now()
       RETURNING *`,
      [claim.id, claim.leaseOwner, claim.attemptCount],
    );
    if (!result.rows[0]) throw staleClaimError("Kafka event outbox", claim);
    return kafkaEventOutboxFromRow(result.rows[0]);
  }

  async retryKafkaEventOutbox(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string): Promise<KafkaEventOutboxRecord> {
    const result = await this.pool.query(
      `UPDATE agentrun_kafka_event_outbox
       SET lease_owner = NULL, lease_expires_at = NULL, next_attempt_at = $2, last_error = $3::jsonb, updated_at = now()
       WHERE id = $1 AND delivered_at IS NULL AND lease_owner = $4 AND attempt_count = $5 AND lease_expires_at > now()
       RETURNING *`,
      [claim.id, nextAttemptAt, JSON.stringify(redactJson(error)), claim.leaseOwner, claim.attemptCount],
    );
    if (!result.rows[0]) throw staleClaimError("Kafka event outbox", claim);
    return kafkaEventOutboxFromRow(result.rows[0]);
  }

  async kafkaEventOutboxStatus(): Promise<JsonRecord> {
    const result = await this.pool.query(
      `SELECT count(*)::int AS total_count,
              count(*) FILTER (WHERE delivered_at IS NULL)::int AS backlog_count,
              count(*) FILTER (WHERE delivered_at IS NULL AND attempt_count > 0)::int AS retrying_count,
              min(created_at) FILTER (WHERE delivered_at IS NULL) AS oldest_pending_at,
              COALESCE(max(attempt_count), 0)::int AS max_attempt_count,
              (array_agg(last_error ORDER BY outbox_seq ASC) FILTER (WHERE delivered_at IS NULL AND last_error IS NOT NULL))[1] AS first_error
       FROM agentrun_kafka_event_outbox`,
    );
    return { ...durableQueueStatusFromRow(result.rows[0], "kafka-event-outbox"), firstError: result.rows[0]?.first_error ?? null, enabled: this.eventOutboxConfig.enabled };
  }

  async registerRunner(input: Partial<RunnerRecord>): Promise<RunnerRecord> {
    const at = nowIso();
    const runner: RunnerRecord = { id: input.id ?? newId("runner"), registeredAt: at, heartbeatAt: at, ...input };
    const metadata = metadataForRunner(runner);
    const result = await this.pool.query(
      `INSERT INTO agentrun_runners (id, run_id, attempt_id, backend_profile, placement, source_commit, metadata, registered_at, heartbeat_at)
       VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9)
       ON CONFLICT (id) DO UPDATE SET
         run_id = EXCLUDED.run_id,
         attempt_id = EXCLUDED.attempt_id,
         backend_profile = EXCLUDED.backend_profile,
         placement = EXCLUDED.placement,
         source_commit = EXCLUDED.source_commit,
         metadata = EXCLUDED.metadata,
         heartbeat_at = EXCLUDED.heartbeat_at
       RETURNING *`,
      [runner.id, runner.runId ?? null, runner.attemptId ?? null, runner.backendProfile ?? null, runner.placement ?? null, runner.sourceCommit ?? null, JSON.stringify(metadata), runner.registeredAt, runner.heartbeatAt],
    );
    return runnerFromRow(result.rows[0]);
  }

  async listRunnerJobs(runId: string, commandId?: string): Promise<RunnerJobRecord[]> {
    await this.getRun(runId);
    const params: unknown[] = [runId];
    let where = "run_id = $1";
    if (commandId) {
      params.push(commandId);
      where += ` AND command_id = $${params.length}`;
    }
    const result = await this.pool.query(`SELECT * FROM agentrun_runner_jobs WHERE ${where} ORDER BY created_at ASC`, params);
    return result.rows.map(runnerJobFromRow);
  }

  async listRunnerJobsForReconciliation(limit: number): Promise<RunnerJobRecord[]> {
    const clamped = Math.max(1, Math.min(limit, 500));
    const result = await this.pool.query(
      `SELECT rj.*
       FROM agentrun_runner_jobs rj
       LEFT JOIN agentrun_runs r ON r.id = rj.run_id
       LEFT JOIN agentrun_commands c ON c.id = rj.command_id
       WHERE r.id IS NULL
          OR c.id IS NULL
          OR r.status NOT IN ('completed', 'failed', 'blocked', 'cancelled')
          OR c.state NOT IN ('completed', 'failed', 'cancelled')
       ORDER BY rj.updated_at ASC, rj.created_at ASC
       LIMIT $1`,
      [clamped],
    );
    return result.rows.map(runnerJobFromRow);
  }

  async getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): Promise<RunnerJobRecord | null> {
    const result = await this.pool.query("SELECT * FROM agentrun_runner_jobs WHERE run_id = $1 AND idempotency_key = $2", [runId, idempotencyKey]);
    const row = result.rows[0];
    if (!row) return null;
    const record = runnerJobFromRow(row);
    if (record.payloadHash !== payloadHash) throw new AgentRunError("schema-invalid", "runner job idempotency key reused with different payload", { httpStatus: 409 });
    return record;
  }

  async saveRunnerJob(input: SaveRunnerJobInput): Promise<RunnerJobRecord> {
    return this.withTransaction(async (client) => {
      await this.requireRunForUpdate(client, input.runId);
      if (input.idempotencyKey) {
        const existing = await client.query("SELECT * FROM agentrun_runner_jobs WHERE run_id = $1 AND idempotency_key = $2 FOR UPDATE", [input.runId, input.idempotencyKey]);
        if (existing.rows[0]) {
          const record = runnerJobFromRow(existing.rows[0]);
          if (record.payloadHash !== input.payloadHash) throw new AgentRunError("schema-invalid", "runner job idempotency key reused with different payload", { httpStatus: 409 });
          return record;
        }
      }
      const at = nowIso();
      const record: RunnerJobRecord = { ...input, id: input.id ?? newId("rjob"), idempotencyKey: input.idempotencyKey ?? null, serviceAccountName: input.serviceAccountName ?? null, createdAt: at, updatedAt: at };
      const inserted = await client.query(
        `INSERT INTO agentrun_runner_jobs (id, run_id, command_id, idempotency_key, payload_hash, attempt_id, runner_id, namespace, job_name, manager_url, image, source_commit, service_account_name, result, created_at, updated_at)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::jsonb, $15, $16)
         RETURNING *`,
        [record.id, record.runId, record.commandId, record.idempotencyKey, record.payloadHash, record.attemptId, record.runnerId, record.namespace, record.jobName, record.managerUrl, record.image, record.sourceCommit, record.serviceAccountName, JSON.stringify(record.result), record.createdAt, record.updatedAt],
      );
      return runnerJobFromRow(inserted.rows[0]);
    });
  }

  async updateRunnerJobResult(runnerJobId: string, patch: JsonRecord): Promise<RunnerJobRecord> {
    return this.withTransaction(async (client) => {
      const existing = await client.query("SELECT * FROM agentrun_runner_jobs WHERE id = $1 FOR UPDATE", [runnerJobId]);
      const row = existing.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `runner job ${runnerJobId} was not found`, { httpStatus: 404 });
      const record = runnerJobFromRow(row);
      const nextResult = { ...record.result, ...patch };
      const updated = await client.query("UPDATE agentrun_runner_jobs SET result = $2::jsonb, updated_at = $3 WHERE id = $1 RETURNING *", [runnerJobId, JSON.stringify(nextResult), nowIso()]);
      return runnerJobFromRow(updated.rows[0]);
    });
  }

  async withRunnerCreateFence<T>(fenceKey: string, operation: () => Promise<T>): Promise<T> {
    const client = await this.runnerCreateFencePool.connect();
    try {
      await client.query("BEGIN");
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-runner-create", fenceKey]);
      const result = await operation();
      await client.query("COMMIT");
      return result;
    } catch (error) {
      await client.query("ROLLBACK");
      throw error;
    } finally {
      client.release();
    }
  }

  async withRunnerRetentionFence<T>(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise<T>): Promise<T> {
    return this.withTransaction(async (client) => {
      const run = await this.requireRunForUpdate(client, input.runId);
      const commandRows = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [run.id]);
      const commands = commandRows.rows.slice(0, 100).map(commandFromRow);
      const commandRow = commandRows.rows.find((item) => String(item.id) === input.commandId);
      if (!commandRow) throw new AgentRunError("schema-invalid", `command ${input.commandId} does not belong to run ${input.runId}`, { httpStatus: 409 });
      const command = commandFromRow(commandRow);
      const eventRows = await client.query("SELECT * FROM agentrun_events WHERE run_id = $1 ORDER BY seq ASC LIMIT 500", [run.id]);
      const facts = { run, command, commands, events: eventRows.rows.map(eventFromRow) };
      if (!input.terminalizeStalePending || isTerminalRunStatus(run.status)) {
        if (input.releaseRunnerClaim && (isTerminalRunStatus(run.status) || isTerminalCommandState(command.state))) await this.releaseRunnerClaim(client, run);
        return await operation(facts);
      }

      const at = nowIso();
      const updatedCommands = await client.query("UPDATE agentrun_commands SET state = 'failed', updated_at = $2 WHERE run_id = $1 AND state NOT IN ('completed', 'failed', 'cancelled') RETURNING *", [run.id, at]);
      await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, input.reason, at);
      const updatedRun = await client.query("UPDATE agentrun_runs SET status = 'failed', terminal_status = 'failed', failure_kind = 'infra-failed', failure_message = $2, claimed_by = NULL, lease_expires_at = NULL, updated_at = $3 WHERE id = $1 RETURNING *", [run.id, input.reason, at]);
      await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [run.id]);
      for (const row of updatedCommands.rows) {
        const item = commandFromRow(row);
        await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", commandId: item.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message: input.reason, retentionFence: true });
      }
      await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message: input.reason, retentionFence: true });
      const next = runFromRow(updatedRun.rows[0]);
      await this.touchSessionForRun(client, next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at });
      return await operation(facts);
    });
  }

  async claimRun(runId: string, runnerId: string, leaseMs: number): Promise<RunRecord> {
    return this.withTransaction(async (client) => {
      const run = await this.requireRunForUpdate(client, runId);
      if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 });
      if (run.claimedBy && run.claimedBy !== runnerId && !isLeaseExpired(run.leaseExpiresAt)) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
      const leaseExpiresAt = new Date(Date.now() + leaseMs).toISOString();
      const updated = await client.query(
        `UPDATE agentrun_runs SET status = $2, claimed_by = $3, lease_expires_at = $4, updated_at = $5 WHERE id = $1 RETURNING *`,
        [runId, "claimed", runnerId, leaseExpiresAt, nowIso()],
      );
      await client.query(
        `INSERT INTO agentrun_leases (run_id, runner_id, lease_expires_at, stale_recovery_marker, updated_at)
         VALUES ($1, $2, $3, $4::jsonb, $5)
         ON CONFLICT (run_id) DO UPDATE SET runner_id = EXCLUDED.runner_id, lease_expires_at = EXCLUDED.lease_expires_at, updated_at = EXCLUDED.updated_at`,
        [runId, runnerId, leaseExpiresAt, null, nowIso()],
      );
      const next = runFromRow(updated.rows[0]);
      await this.touchSessionForRun(client, next, { executionState: "running", activeRunId: runId, lastRunId: runId, lastActivityAt: next.updatedAt }, { bumpVersion: false, at: next.updatedAt });
      await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "run-claimed", runnerId });
      return next;
    });
  }

  async heartbeat(runId: string, runnerId: string, leaseMs: number): Promise<RunRecord> {
    return this.withTransaction(async (client) => {
      const run = await this.requireRunForUpdate(client, runId);
      if (isTerminalRunStatus(run.status)) return run;
      if (run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is not claimed by ${runnerId}`, { httpStatus: 409 });
      const leaseExpiresAt = new Date(Date.now() + leaseMs).toISOString();
      const updated = await client.query("UPDATE agentrun_runs SET lease_expires_at = $2, updated_at = $3 WHERE id = $1 RETURNING *", [runId, leaseExpiresAt, nowIso()]);
      await client.query("UPDATE agentrun_runners SET heartbeat_at = $2 WHERE id = $1", [runnerId, nowIso()]);
      await client.query("UPDATE agentrun_leases SET lease_expires_at = $2, updated_at = $3 WHERE run_id = $1", [runId, leaseExpiresAt, nowIso()]);
      return runFromRow(updated.rows[0]);
    });
  }

  async ackCommand(commandId: string): Promise<CommandRecord> {
    return this.withTransaction(async (client) => {
      const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
      const row = existing.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
      const command = commandFromRow(row);
      if (isTerminalCommandState(command.state) || command.state === "acknowledged") return command;
      const result = await client.query("UPDATE agentrun_commands SET state = $2, acknowledged_at = $3, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, "acknowledged", nowIso()]);
      return commandFromRow(result.rows[0]);
    });
  }

  async finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): Promise<CommandRecord> {
    return this.withTransaction(async (client) => {
      const identity = await client.query("SELECT run_id FROM agentrun_commands WHERE id = $1", [commandId]);
      if (!identity.rows[0]) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
      let run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id));
      if (isTerminalRunStatus(run.status)) run = await this.releaseTerminalRunClaim(client, run);
      const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
      const row = existing.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
      const command = commandFromRow(row);
      if (isTerminalCommandState(command.state)) {
        await fenceActiveRunnerDispatchIntents(client, { commandId }, "command is terminal");
        if (command.state === "cancelled" && result.terminalStatus !== "cancelled") {
          await this.appendEventWithLockedRun(client, command.runId, "backend_status", lateWriteRejectedPayload(run, command, { source: "command-status", terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null }));
        }
        return command;
      }
      const state = commandStateFromTerminal(result.terminalStatus);
      const at = nowIso();
      const updated = await client.query("UPDATE agentrun_commands SET state = $2, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, state, at]);
      await fenceActiveRunnerDispatchIntents(client, { commandId }, `command terminalized as ${state}`, at);
      if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null);
      if (command.type === "turn") await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: command.runId, lastCommandId: command.id, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: nowIso() }, { bumpVersion: true });
      await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state, terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null });
      return commandFromRow(updated.rows[0]);
    });
  }

  async appendEvent(runId: string, type: EventType, payload: JsonRecord): Promise<RunEvent> {
    type = requireExternallyAppendableEventType(type);
    return this.withTransaction(async (client) => {
      await this.requireRunForUpdate(client, runId);
      return this.appendEventWithLockedRun(client, runId, type, payload);
    });
  }

  async finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): Promise<RunRecord> {
    return this.withTransaction(async (client) => {
      const existing = await this.requireRunForUpdate(client, runId);
      await client.query("SELECT id FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [runId]);
      if (isTerminalRunStatus(existing.status)) {
        await fenceActiveRunnerDispatchIntents(client, { runId }, "run is terminal");
        const released = await this.releaseTerminalRunClaim(client, existing);
        if (released.status === "cancelled" && result.terminalStatus !== "cancelled") await this.appendEventWithLockedRun(client, runId, "backend_status", lateWriteRejectedPayload(released, null, { source: "run-status", terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null }));
        return released;
      }
      const status = statusFromTerminal(result.terminalStatus);
      const at = nowIso();
      const updated = await client.query(
        `UPDATE agentrun_runs SET status = $2, terminal_status = $3, failure_kind = $4, failure_message = $5, claimed_by = NULL, lease_expires_at = NULL, updated_at = $6 WHERE id = $1 RETURNING *`,
        [runId, status, result.terminalStatus, result.failureKind, result.failureMessage, at],
      );
      await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [runId]);
      const run = runFromRow(updated.rows[0]);
      await fenceActiveRunnerDispatchIntents(client, { runId }, `run terminalized as ${status}`, at);
      if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null);
      await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage });
      const sessionId = run.sessionRef?.sessionId;
      const sessionResult = sessionId ? await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [sessionId]) : null;
      const session = sessionResult?.rows[0] ? sessionFromRow(sessionResult.rows[0]) : null;
      const stillCurrent = !session || ((session.activeRunId === null || session.activeRunId === runId) && (session.lastRunId === null || session.lastRunId === runId));
      if (stillCurrent) await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: runId, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: run.updatedAt }, { bumpVersion: true, at: run.updatedAt });
      return run;
    });
  }

  async cancelRun(runId: string, reason = "cancel requested"): Promise<RunRecord> {
    return this.withTransaction(async (client) => {
      const run = await this.requireRunForUpdate(client, runId);
      const commands = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [runId]);
      if (isTerminalRunStatus(run.status)) {
        await fenceActiveRunnerDispatchIntents(client, { runId }, reason);
        return await this.releaseTerminalRunClaim(client, run);
      }
      const at = nowIso();
      const cancel = await this.createCancelRequest(client, { targetKind: "run", targetId: runId, run, command: null, reason, at, stage: "accepted" });
      await this.appendCancelStage(client, runId, "accepted", cancel);
      const persistedRunResult = await client.query(
        `UPDATE agentrun_runs SET cancel_epoch = $2, cancel_request_id = $3, cancel_requested_at = $4, cancel_reason = $5, updated_at = $4 WHERE id = $1 RETURNING *`,
        [runId, cancel.epoch, cancel.id, at, reason],
      );
      const persistedRun = runFromRow(persistedRunResult.rows[0]);
      await this.appendCancelStage(client, runId, "persisted", cancel);
      const leaseExpired = Boolean(persistedRun.claimedBy && isLeaseExpired(persistedRun.leaseExpiresAt));
      if (leaseExpired) await this.appendCancelStage(client, runId, "fenced", cancel, { claimedBy: persistedRun.claimedBy, leaseExpiresAt: persistedRun.leaseExpiresAt });
      else if (persistedRun.claimedBy) {
        await this.appendCancelStage(client, runId, "delivered", cancel, { claimedBy: persistedRun.claimedBy });
        await this.appendCancelStage(client, runId, "aborting", cancel, { claimedBy: persistedRun.claimedBy });
      }
      for (const row of commands.rows) {
        const command = commandFromRow(row);
        if (isTerminalCommandState(command.state)) continue;
        await client.query("UPDATE agentrun_commands SET state = 'cancelled', cancel_epoch = $2, cancel_request_id = $3, cancel_requested_at = $4, cancel_reason = $5, updated_at = $4 WHERE id = $1", [command.id, cancel.epoch, cancel.id, at, reason]);
        await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-terminal", commandId: command.id, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", cancelRequestId: cancel.id, cancelEpoch: cancel.epoch });
      }
      await fenceActiveRunnerDispatchIntents(client, { runId }, reason, at);
      const updated = await client.query(
        `UPDATE agentrun_runs SET status = 'cancelled', terminal_status = 'cancelled', failure_kind = 'cancelled', failure_message = $2, claimed_by = NULL, lease_expires_at = NULL, updated_at = $3 WHERE id = $1 RETURNING *`,
        [runId, reason, at],
      );
      await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [runId]);
      await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason, cancelRequestId: cancel.id, cancelEpoch: cancel.epoch });
      await this.appendCancelStage(client, runId, "terminalized", cancel);
      const next = runFromRow(updated.rows[0]);
      await this.touchSessionForRun(client, next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: runId, terminalStatus: "cancelled", failureKind: "cancelled", lastActivityAt: at }, { bumpVersion: true, at });
      return next;
    });
  }

  async cancelCommand(commandId: string, reason = "cancel requested"): Promise<CommandRecord> {
    return this.withTransaction(async (client) => {
      const identity = await client.query("SELECT run_id FROM agentrun_commands WHERE id = $1", [commandId]);
      if (!identity.rows[0]) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
      let run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id));
      if (isTerminalRunStatus(run.status)) run = await this.releaseTerminalRunClaim(client, run);
      const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
      const row = existing.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
      const command = commandFromRow(row);
      if (isTerminalCommandState(command.state)) {
        await fenceActiveRunnerDispatchIntents(client, { commandId }, reason);
        return command;
      }
      const at = nowIso();
      const cancel = await this.createCancelRequest(client, { targetKind: "command", targetId: commandId, run, command, reason, at, stage: "accepted" });
      await this.appendCancelStage(client, command.runId, "accepted", cancel);
      const updated = await client.query("UPDATE agentrun_commands SET state = 'cancelled', cancel_epoch = $2, cancel_request_id = $3, cancel_requested_at = $4, cancel_reason = $5, updated_at = $4 WHERE id = $1 RETURNING *", [commandId, cancel.epoch, cancel.id, at, reason]);
      await fenceActiveRunnerDispatchIntents(client, { commandId }, reason, at);
      await this.appendCancelStage(client, command.runId, "persisted", cancel);
      const leaseExpired = Boolean(run.claimedBy && isLeaseExpired(run.leaseExpiresAt));
      if (leaseExpired) await this.appendCancelStage(client, command.runId, "fenced", cancel, { claimedBy: run.claimedBy, leaseExpiresAt: run.leaseExpiresAt });
      else if (run.claimedBy || command.state === "acknowledged") {
        await this.appendCancelStage(client, command.runId, "delivered", cancel, { claimedBy: run.claimedBy, commandState: command.state });
        await this.appendCancelStage(client, command.runId, "aborting", cancel, { claimedBy: run.claimedBy, commandState: command.state });
      }
      await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", message: reason, cancelRequestId: cancel.id, cancelEpoch: cancel.epoch });
      await this.appendCancelStage(client, command.runId, "terminalized", cancel);
      if (command.type === "turn") {
        await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: command.runId, lastCommandId: command.id, terminalStatus: "cancelled", failureKind: "cancelled", lastActivityAt: at }, { bumpVersion: true, at });
      }
      return commandFromRow(updated.rows[0]);
    });
  }

  async getSession(sessionId: string): Promise<SessionRecord | null> {
    const result = await this.pool.query("SELECT * FROM agentrun_sessions WHERE session_id = $1", [sessionId]);
    return result.rows[0] ? sessionFromRow(result.rows[0]) : null;
  }

  async getLatestSessionResourceRun(sessionId: string): Promise<RunRecord | null> {
    const result = await this.pool.query(
      `SELECT * FROM agentrun_runs
       WHERE session_ref->>'sessionId' = $1 AND jsonb_typeof(resource_bundle_ref) = 'object'
       ORDER BY created_at DESC, id DESC
       LIMIT 1`,
      [sessionId],
    );
    return result.rows[0] ? runFromRow(result.rows[0]) : null;
  }

  async getSessionSummary(sessionId: string, readerId: string | null = null): Promise<SessionSummary> {
    const session = await this.getSession(sessionId);
    if (!session) throw new AgentRunError("schema-invalid", `session ${sessionId} was not found`, { httpStatus: 404 });
    const cursor = readerId ? await this.getSessionReadCursor(sessionId, readerId) : null;
    return buildSessionSummary(session, readerId, cursor);
  }

  async listSessions(input: ListSessionsInput): Promise<SessionListResult> {
    const startVersion = parseSessionCursor(input.cursor) ?? 0;
    const state = input.state ?? "default";
    const params: unknown[] = [startVersion];
    const where = ["version > $1"];
    if (input.backendProfile) {
      params.push(input.backendProfile);
      where.push(`backend_profile = $${params.length}`);
    }
    const result = await this.pool.query(`SELECT * FROM agentrun_sessions WHERE ${where.join(" AND ")} ORDER BY updated_at DESC, session_id ASC LIMIT 500`, params);
    const cursors = input.readerId ? await this.loadSessionReadCursors(input.readerId, result.rows.map((row) => stringValue(row.session_id))) : new Map<string, SessionReadCursorRecord>();
    const items = result.rows
      .map(sessionFromRow)
      .map((session) => buildSessionSummary(session, input.readerId ?? null, input.readerId ? cursors.get(session.sessionId) ?? null : null))
      .filter((session) => sessionMatchesListState(session, state))
      .sort(sessionSort)
      .slice(0, clampSessionLimit(input.limit));
    return { items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.version ?? startVersion) : null, filters: sessionListFilters(input) };
  }

  async listSessionTrace(sessionId: string, input: SessionEventPageInput): Promise<SessionEventPage> {
    const runId = await this.resolveSessionRunId(sessionId, input.runId ?? null);
    if (!runId) return { sessionId, runId: null, items: [], count: 0, cursor: null };
    const items = await this.listEvents(runId, input.afterSeq, input.limit);
    return { sessionId, runId, items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.seq ?? input.afterSeq) : null };
  }

  async listSessionOutput(sessionId: string, input: SessionEventPageInput): Promise<SessionEventPage> {
    const page = await this.listSessionTrace(sessionId, input);
    const items = page.items.filter(isSessionOutputEvent);
    return { ...page, items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.seq ?? input.afterSeq) : null };
  }

  async markSessionRead(sessionId: string, readerId: string): Promise<SessionReadCursorRecord> {
    return this.withTransaction(async (client) => {
      const result = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [sessionId]);
      const row = result.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `session ${sessionId} was not found`, { httpStatus: 404 });
      const session = sessionFromRow(row);
      const record: SessionReadCursorRecord = { sessionId, readerId, sessionVersion: session.version, readAt: nowIso() };
      await client.query(
        `INSERT INTO agentrun_session_read_cursors (session_id, reader_id, session_version, read_at)
         VALUES ($1, $2, $3, $4)
         ON CONFLICT (session_id, reader_id) DO UPDATE SET session_version = EXCLUDED.session_version, read_at = EXCLUDED.read_at`,
        [record.sessionId, record.readerId, record.sessionVersion, record.readAt],
      );
      return record;
    });
  }

  async upsertSession(input: UpsertSessionInput): Promise<SessionRecord> {
    return this.withTransaction(async (client) => {
      const existing = await client.query<QueryResultRow>("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]);
      const at = nowIso();
      if (existing.rows[0]) {
        const session = sessionFromRow(existing.rows[0]);
        const next: SessionRecord = {
          ...session,
          tenantId: input.tenantId,
          projectId: input.projectId,
          backendProfile: input.backendProfile,
          conversationId: input.conversationId,
          threadId: input.threadId,
          metadata: input.metadata,
          expiresAt: input.expiresAt,
          codexRolloutSubdir: input.codexRolloutSubdir,
          version: session.version + 1,
          updatedAt: at,
        };
        await client.query(
          `UPDATE agentrun_sessions
           SET tenant_id = $2, project_id = $3, backend_profile = $4, conversation_id = $5, thread_id = $6,
               metadata = $7, expires_at = $8, codex_rollout_subdir = $9, version = $10, updated_at = $11
           WHERE session_id = $1`,
          [next.sessionId, next.tenantId, next.projectId, next.backendProfile, next.conversationId, next.threadId,
           JSON.stringify(next.metadata), next.expiresAt, next.codexRolloutSubdir, next.version, next.updatedAt],
        );
        return next;
      }
      const next: SessionRecord = {
        sessionId: input.sessionId,
        tenantId: input.tenantId,
        projectId: input.projectId,
        backendProfile: input.backendProfile,
        conversationId: input.conversationId,
        threadId: input.threadId,
        metadata: input.metadata,
        version: 1,
        executionState: "idle",
        lastRunId: null,
        lastCommandId: null,
        activeRunId: null,
        activeCommandId: null,
        lastEventSeq: 0,
        terminalStatus: null,
        failureKind: null,
        title: null,
        summary: {},
        lastActivityAt: null,
        createdAt: at,
        updatedAt: at,
        expiresAt: input.expiresAt,
        storageKind: "none",
        codexRolloutSubdir: input.codexRolloutSubdir,
      };
      await client.query(
        `INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id,
                                        metadata, version, execution_state, last_run_id, last_command_id, active_run_id,
                                        active_command_id, last_event_seq, terminal_status, failure_kind, title, summary,
                                        last_activity_at, created_at, updated_at, expires_at, storage_kind, codex_rollout_subdir)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'idle', null, null, null, null, 0, null, null, null, '{}'::jsonb, null, $9, $10, $11, 'none', $12)`,
        [next.sessionId, next.tenantId, next.projectId, next.backendProfile, next.conversationId, next.threadId,
         JSON.stringify(next.metadata), next.version, next.createdAt, next.updatedAt, next.expiresAt, next.codexRolloutSubdir],
      );
      return next;
    });
  }

  async refreshSessionStorage(input: SessionStoragePatch): Promise<SessionRecord> {
    return this.withTransaction(async (client) => {
      const existing = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]);
      if (!existing.rows[0]) throw new AgentRunError("schema-invalid", `session ${input.sessionId} was not found`, { httpStatus: 404 });
      const session = sessionFromRow(existing.rows[0]);
      const at = nowIso();
      const next: SessionRecord = {
        ...session,
        storageKind: input.storageKind,
        storagePvcName: input.pvcName ?? null,
        storageNamespace: input.storageNamespace ?? null,
        storagePvcPhase: input.pvcPhase ?? null,
        storageSizeBytes: input.storageSizeBytes ?? null,
        storageFilesCount: input.storageFilesCount ?? null,
        storageSha256: input.storageSha256 ?? null,
        codexRolloutSubdir: input.codexRolloutSubdir ?? session.codexRolloutSubdir ?? "sessions",
        storageUpdatedAt: at,
        version: session.version + 1,
        updatedAt: at,
      };
      await client.query(
        `UPDATE agentrun_sessions
         SET storage_kind = $2, storage_pvc_name = $3, storage_namespace = $4, storage_pvc_phase = $5,
             storage_size_bytes = $6, storage_files_count = $7, storage_sha256 = $8,
             storage_updated_at = $9, codex_rollout_subdir = $10, version = $11, updated_at = $12
         WHERE session_id = $1`,
        [next.sessionId, next.storageKind, next.storagePvcName, next.storageNamespace, next.storagePvcPhase,
         next.storageSizeBytes, next.storageFilesCount, next.storageSha256,
         next.storageUpdatedAt, next.codexRolloutSubdir, next.version, next.updatedAt],
      );
      return next;
    });
  }

  async markSessionStorageEvicted(input: { sessionId: string; pvcName: string }): Promise<SessionRecord> {
    return this.withTransaction(async (client) => {
      const existing = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]);
      if (!existing.rows[0]) throw new AgentRunError("schema-invalid", `session ${input.sessionId} was not found`, { httpStatus: 404 });
      const session = sessionFromRow(existing.rows[0]);
      const at = nowIso();
      const next: SessionRecord = {
        ...session,
        storageKind: "evicted",
        storagePvcName: input.pvcName,
        storageEvictedAt: at,
        storageUpdatedAt: at,
        version: session.version + 1,
        updatedAt: at,
      };
      await client.query(
        `UPDATE agentrun_sessions
         SET storage_kind = $2, storage_pvc_name = $3, storage_evicted_at = $4, storage_updated_at = $5,
             version = $6, updated_at = $7
         WHERE session_id = $1`,
        [next.sessionId, next.storageKind, next.storagePvcName, next.storageEvictedAt, next.storageUpdatedAt, next.version, next.updatedAt],
      );
      return next;
    });
  }

  async listGcExpiredSessions(input: ListGcExpiredSessionsInput): Promise<SessionRecord[]> {
    const limit = Math.max(1, input.limit);
    const result = await this.pool.query<QueryResultRow>(
      `SELECT * FROM agentrun_sessions
       WHERE storage_kind = 'pvc' AND storage_pvc_name IS NOT NULL
         AND expires_at IS NOT NULL AND expires_at <= to_timestamp($1)
       ORDER BY updated_at ASC
       LIMIT $2`,
      [input.now / 1000, limit],
    );
    return result.rows.map(sessionFromRow);
  }

  async createQueueTask(input: CreateQueueTaskInput): Promise<QueueTaskRecord> {
    const payloadHash = queueTaskPayloadHash(input);
    return this.withTransaction(async (client) => {
      if (input.idempotencyKey) {
        const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE tenant_id = $1 AND project_id = $2 AND idempotency_key = $3 FOR UPDATE", [input.tenantId, input.projectId, input.idempotencyKey]);
        if (existing.rows[0]) {
          const record = queueTaskFromRow(existing.rows[0]);
          if (record.payloadHash !== payloadHash) throw new AgentRunError("schema-invalid", "queue task idempotency key reused with different payload", { httpStatus: 409 });
          return record;
        }
      }
      const at = nowIso();
      const version = await this.nextQueueVersion(client);
      const sessionId = input.sessionRef?.sessionId ?? null;
      const sessionPath = sessionId ? `/api/v1/sessions/${encodeURIComponent(sessionId)}` : null;
      const task: QueueTaskRecord = { ...input, id: newId("qt"), state: "pending", version, payloadHash, latestAttempt: null, sessionPath, createdAt: at, updatedAt: at, cancelledAt: null, cancelReason: null };
      const inserted = await client.query(
        `INSERT INTO agentrun_queue_tasks (id, tenant_id, project_id, queue, lane, title, priority, state, version, backend_profile, provider_id, workspace_ref, session_ref, execution_policy, resource_bundle_ref, payload, payload_hash, references_json, metadata, idempotency_key, latest_attempt, session_path, created_at, updated_at, cancelled_at, cancel_reason)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb, $13::jsonb, $14::jsonb, $15::jsonb, $16::jsonb, $17, $18::jsonb, $19::jsonb, $20, $21::jsonb, $22, $23, $24, $25, $26)
         RETURNING *`,
        [task.id, task.tenantId, task.projectId, task.queue, task.lane, task.title, task.priority, task.state, task.version, task.backendProfile, task.providerId, JSON.stringify(task.workspaceRef), JSON.stringify(task.sessionRef), JSON.stringify(task.executionPolicy), JSON.stringify(task.resourceBundleRef), JSON.stringify(task.payload), task.payloadHash, JSON.stringify(task.references), JSON.stringify(task.metadata), task.idempotencyKey ?? null, JSON.stringify(task.latestAttempt), task.sessionPath, task.createdAt, task.updatedAt, task.cancelledAt, task.cancelReason],
      );
      return queueTaskFromRow(inserted.rows[0]);
    });
  }

  async listQueueTasks(input: ListQueueTasksInput): Promise<QueueTaskListResult> {
    const startVersion = parseQueueCursor(input.cursor) ?? input.updatedAfter ?? 0;
    const params: unknown[] = [startVersion];
    const where = ["version > $1"];
    if (input.queue) {
      params.push(input.queue);
      where.push(`queue = $${params.length}`);
    }
    if (input.state) {
      params.push(input.state);
      where.push(`state = $${params.length}`);
    }
    params.push(clampQueueLimit(input.limit));
    const result = await this.pool.query(`SELECT * FROM agentrun_queue_tasks WHERE ${where.join(" AND ")} ORDER BY priority DESC, created_at ASC, id ASC LIMIT $${params.length}`, params);
    const items = result.rows.map(queueTaskFromRow);
    return { items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.version ?? startVersion) : null };
  }

  async getQueueTask(taskId: string): Promise<QueueTaskRecord> {
    const result = await this.pool.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1", [taskId]);
    const row = result.rows[0];
    if (!row) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
    return queueTaskFromRow(row);
  }

  async listQueueAttempts(taskId: string, input: ListQueueAttemptsInput): Promise<QueueAttemptListResult> {
    await this.getQueueTask(taskId);
    const limit = clampQueueLimit(input.limit);
    const params: unknown[] = [taskId];
    let cursorClause = "";
    if (input.cursor !== undefined) {
      params.push(input.cursor);
      cursorClause = ` AND retry_index < $${params.length}`;
    }
    params.push(limit);
    const result = await this.pool.query(
      `SELECT * FROM agentrun_queue_attempts
       WHERE task_id = $1${cursorClause}
       ORDER BY retry_index DESC, created_at DESC
       LIMIT $${params.length}`,
      params,
    );
    const items = result.rows.map(queueAttemptFromRow);
    return { taskId, items, count: items.length, cursor: items.length === limit ? String(items.at(-1)?.retryIndex ?? "") : null };
  }

  async reserveQueueRetryAttempt(taskId: string, input: ReserveQueueRetryAttemptInput): Promise<QueueRetryReservation> {
    return this.withTransaction(async (client) => {
      const taskResult = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]);
      if (!taskResult.rows[0]) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
      const task = queueTaskFromRow(taskResult.rows[0]);
      assertQueueTaskPayloadHash(task);
      const attemptsResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 ORDER BY retry_index ASC, created_at ASC FOR UPDATE", [taskId]);
      const attempts = attemptsResult.rows.map(queueAttemptFromRow);
      const existing = attempts.find((attempt) => attempt.idempotencyKey === input.idempotencyKey) ?? null;
      if (existing) {
        if (existing.reason !== input.reason) throw new AgentRunError("schema-invalid", "queue retry idempotency key reused with different reason", { httpStatus: 409 });
        return queueRetryReservation(task, existing, attempts, false, true);
      }
      assertQueueRetryable(task, attempts);
      const retryIndex = nextQueueRetryIndex(attempts);
      if (input.dryRun) return queueRetryReservation(task, null, attempts, false, false, retryIndex);
      const at = nowIso();
      const attempt: QueueAttemptRecord = {
        attemptId: newId("qa"),
        taskId,
        retryIndex,
        state: "reserved",
        idempotencyKey: input.idempotencyKey,
        reason: input.reason,
        payloadHash: task.payloadHash,
        previousAttemptId: latestQueueAttempt(attempts)?.attemptId ?? task.latestAttempt?.attemptId ?? null,
        runId: newId("run"),
        commandId: null,
        runnerJobId: newId("rjob"),
        sessionId: task.sessionRef?.sessionId ?? null,
        sessionPath: task.sessionPath,
        failureKind: null,
        failureMessage: null,
        createdAt: at,
        updatedAt: at,
        activatedAt: null,
        terminalAt: null,
      };
      await client.query(
        `INSERT INTO agentrun_queue_attempts (
           attempt_id, task_id, retry_index, state, idempotency_key, reason, payload_hash,
           previous_attempt_id, run_id, command_id, runner_job_id, session_id, session_path,
           failure_kind, failure_message, created_at, updated_at, activated_at, terminal_at
         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)`,
        [attempt.attemptId, attempt.taskId, attempt.retryIndex, attempt.state, attempt.idempotencyKey, attempt.reason, attempt.payloadHash, attempt.previousAttemptId, attempt.runId, attempt.commandId, attempt.runnerJobId, attempt.sessionId, attempt.sessionPath, attempt.failureKind, attempt.failureMessage, attempt.createdAt, attempt.updatedAt, attempt.activatedAt, attempt.terminalAt],
      );
      return queueRetryReservation(task, attempt, attempts, true, false);
    });
  }

  async recordQueueRetryAttemptFailure(taskId: string, attemptId: string, input: RecordQueueRetryAttemptFailureInput): Promise<QueueAttemptRecord> {
    return this.withTransaction(async (client) => {
      const attemptResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 AND attempt_id = $2 FOR UPDATE", [taskId, attemptId]);
      if (!attemptResult.rows[0]) throw new AgentRunError("schema-invalid", `queue attempt ${attemptId} was not found for task ${taskId}`, { httpStatus: 404 });
      const attempt = queueAttemptFromRow(attemptResult.rows[0]);
      if (attempt.state !== "reserved") return attempt;
      const updated = await client.query(
        `UPDATE agentrun_queue_attempts
         SET failure_kind = $3, failure_message = $4, updated_at = $5
         WHERE task_id = $1 AND attempt_id = $2
         RETURNING *`,
        [taskId, attemptId, input.failureKind, input.failureMessage, nowIso()],
      );
      return queueAttemptFromRow(updated.rows[0]);
    });
  }

  async activateQueueRetryAttempt(taskId: string, attemptId: string, input: ActivateQueueRetryAttemptInput): Promise<QueueRetryActivation> {
    return this.withTransaction(async (client) => {
      const taskResult = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]);
      if (!taskResult.rows[0]) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
      const task = queueTaskFromRow(taskResult.rows[0]);
      const attemptResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 AND attempt_id = $2 FOR UPDATE", [taskId, attemptId]);
      if (!attemptResult.rows[0]) throw new AgentRunError("schema-invalid", `queue attempt ${attemptId} was not found for task ${taskId}`, { httpStatus: 404 });
      const attempt = queueAttemptFromRow(attemptResult.rows[0]);
      if (attempt.state !== "reserved") {
        assertQueueAttemptActivationReplay(attempt, input);
        return { task, attempt, activated: false };
      }
      if (task.state !== "failed" && task.state !== "blocked") throw queueRetryStateError(task);
      const at = nowIso();
      const terminalAt = isTerminalQueueTaskState(input.state) ? at : null;
      const updatedAttempt = await client.query(
        `UPDATE agentrun_queue_attempts
         SET state = $3, command_id = $4, session_id = $5, session_path = $6,
             failure_kind = $7, failure_message = $8, updated_at = $9, activated_at = $9, terminal_at = $10
         WHERE task_id = $1 AND attempt_id = $2
         RETURNING *`,
        [taskId, attemptId, input.state, input.commandId, input.sessionId, input.sessionPath, input.failureKind, input.failureMessage, at, terminalAt],
      );
      const nextAttempt = queueAttemptFromRow(updatedAttempt.rows[0]);
      const version = await this.nextQueueVersion(client);
      const updatedTask = await client.query(
        `UPDATE agentrun_queue_tasks
         SET state = $2, latest_attempt = $3::jsonb, session_path = $4, version = $5, updated_at = $6
         WHERE id = $1 RETURNING *`,
        [taskId, input.state, JSON.stringify(queueAttemptRef(nextAttempt)), input.sessionPath, version, at],
      );
      return { task: queueTaskFromRow(updatedTask.rows[0]), attempt: nextAttempt, activated: true };
    });
  }

  async updateQueueTaskAttempt(taskId: string, input: UpdateQueueTaskAttemptInput): Promise<QueueTaskRecord> {
    return this.withTransaction(async (client) => {
      const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]);
      const row = existing.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
      const task = queueTaskFromRow(row);
      if (isTerminalQueueTaskState(task.state)) throw new AgentRunError(task.state === "cancelled" ? "cancelled" : "schema-invalid", `queue task ${taskId} is already terminal: ${task.state}`, { httpStatus: 409 });
      const version = await this.nextQueueVersion(client);
      const at = nowIso();
      await this.upsertQueueAttemptProjection(client, task, input.latestAttempt, at);
      const updated = await client.query(
        "UPDATE agentrun_queue_tasks SET state = $2, latest_attempt = $3::jsonb, session_path = $4, version = $5, updated_at = $6 WHERE id = $1 RETURNING *",
        [taskId, input.state, JSON.stringify(input.latestAttempt), input.sessionPath, version, at],
      );
      return queueTaskFromRow(updated.rows[0]);
    });
  }

  async cancelQueueTask(taskId: string, reason = "cancel requested"): Promise<QueueTaskRecord> {
    const task = await this.getQueueTask(taskId);
    if (isTerminalQueueTaskState(task.state)) return task;
    if (task.latestAttempt?.runId) await this.cancelRun(task.latestAttempt.runId, reason);
    else if (task.latestAttempt?.commandId) await this.cancelCommand(task.latestAttempt.commandId, reason);
    return this.withTransaction(async (client) => {
      const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]);
      const row = existing.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
      const lockedTask = queueTaskFromRow(row);
      if (isTerminalQueueTaskState(lockedTask.state)) return lockedTask;
      const at = nowIso();
      const version = await this.nextQueueVersion(client);
      const latestAttempt = lockedTask.latestAttempt ? { ...lockedTask.latestAttempt, state: "cancelled" as const, failureKind: "cancelled" as const, failureMessage: reason } : null;
      if (latestAttempt) await this.upsertQueueAttemptProjection(client, lockedTask, latestAttempt, at);
      const updated = await client.query("UPDATE agentrun_queue_tasks SET state = 'cancelled', latest_attempt = $2::jsonb, version = $3, updated_at = $4, cancelled_at = $4, cancel_reason = $5 WHERE id = $1 RETURNING *", [taskId, JSON.stringify(latestAttempt), version, at, reason]);
      return queueTaskFromRow(updated.rows[0]);
    });
  }

  async markQueueTaskRead(taskId: string, readerId: string): Promise<QueueReadCursorRecord> {
    return this.withTransaction(async (client) => {
      const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]);
      const row = existing.rows[0];
      if (!row) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
      const task = queueTaskFromRow(row);
      const record: QueueReadCursorRecord = { taskId, readerId, taskVersion: task.version, readAt: nowIso() };
      await client.query(
        `INSERT INTO agentrun_queue_read_cursors (task_id, reader_id, task_version, read_at)
         VALUES ($1, $2, $3, $4)
         ON CONFLICT (task_id, reader_id) DO UPDATE SET task_version = EXCLUDED.task_version, read_at = EXCLUDED.read_at`,
        [record.taskId, record.readerId, record.taskVersion, record.readAt],
      );
      return record;
    });
  }

  async queueStats(queue?: string): Promise<QueueStats> {
    return buildQueueStats(await this.loadQueueTasksForProjection(queue), queue ?? null);
  }

  async queueCommander(queue?: string, readerId: string | null = null): Promise<QueueCommanderSnapshot> {
    const tasks = await this.loadQueueTasksForProjection(queue);
    const generatedAt = nowIso();
    const cursors = readerId ? await this.loadQueueReadCursors(readerId, tasks.map((task) => task.id)) : new Map<string, QueueReadCursorRecord>();
    const items = tasks
      .map((task) => buildQueueTaskSummary(task, readerId, readerId ? cursors.get(task.id) ?? null : null))
      .filter((task) => queueTaskMatchesCommander(task, readerId))
      .sort(queueTaskSort)
      .slice(0, 20);
    return { queue: queue ?? null, readerId, stats: buildQueueStats(tasks, queue ?? null, generatedAt), items, generatedAt };
  }

  async backends(): Promise<JsonRecord[]> {
    const result = await this.pool.query("SELECT * FROM agentrun_backends ORDER BY profile ASC");
    return result.rows.map((row) => {
      const profile = stringValue(row.profile);
      return { ...mergeBackendCapability(profile, jsonRecord(row.capabilities)), capacity: jsonValue(row.capacity), health: jsonValue(row.health), updatedAt: nullableIso(row.updated_at) };
    });
  }

  async close(): Promise<void> {
    await Promise.all([this.pool.end(), this.runnerCreateFencePool.end()]);
  }

  private async upsertQueueAttemptProjection(client: PoolClient, task: QueueTaskRecord, ref: QueueAttemptRef, at: string): Promise<void> {
    const existingResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 AND attempt_id = $2 FOR UPDATE", [task.id, ref.attemptId]);
    if (existingResult.rows[0]) {
      const existing = queueAttemptFromRow(existingResult.rows[0]);
      if (isTerminalQueueAttemptState(existing.state)) {
        if (stableHash(queueAttemptRef(existing)) !== stableHash(ref)) throw new AgentRunError("schema-invalid", `queue attempt ${ref.attemptId} is terminal and immutable`, { httpStatus: 409 });
        return;
      }
      await client.query(
        `UPDATE agentrun_queue_attempts
         SET state = $3, run_id = $4, command_id = $5, runner_job_id = $6,
             session_id = $7, session_path = $8, failure_kind = $9, failure_message = $10, updated_at = $11,
             activated_at = COALESCE(activated_at, $11),
             terminal_at = CASE WHEN $3 IN ('completed', 'failed', 'blocked', 'cancelled') THEN $11 ELSE NULL END
         WHERE task_id = $1 AND attempt_id = $2`,
        [task.id, ref.attemptId, ref.state, ref.runId, ref.commandId, ref.runnerJobId, ref.sessionId, ref.sessionPath, ref.failureKind ?? null, ref.failureMessage ?? null, at],
      );
      return;
    }
    const history = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 ORDER BY retry_index ASC FOR UPDATE", [task.id]);
    const attempts = history.rows.map(queueAttemptFromRow);
    const retryIndex = nextQueueRetryIndex(attempts);
    await client.query(
      `INSERT INTO agentrun_queue_attempts (
         attempt_id, task_id, retry_index, state, idempotency_key, reason, payload_hash,
         previous_attempt_id, run_id, command_id, runner_job_id, session_id, session_path,
         failure_kind, failure_message, created_at, updated_at, activated_at, terminal_at
       ) VALUES ($1, $2, $3, $4, NULL, NULL, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $14, $14, $15)`,
      [ref.attemptId, task.id, retryIndex, ref.state, task.payloadHash, latestQueueAttempt(attempts)?.attemptId ?? null, ref.runId, ref.commandId, ref.runnerJobId, ref.sessionId, ref.sessionPath, ref.failureKind ?? null, ref.failureMessage ?? null, at, isTerminalQueueTaskState(ref.state) ? at : null],
    );
  }

  private async createCancelRequest(client: PoolClient, input: { targetKind: CancelTargetKind; targetId: string; run: RunRecord; command: CommandRecord | null; reason: string; at: string; stage: CancelStage }): Promise<CancelRequestRecord> {
    const epoch = input.command ? input.command.cancelEpoch + 1 : input.run.cancelEpoch + 1;
    const record: CancelRequestRecord = {
      id: newId("cancel"),
      targetKind: input.targetKind,
      targetId: input.targetId,
      runId: input.run.id,
      commandId: input.command?.id ?? null,
      sessionId: input.run.sessionRef?.sessionId ?? null,
      taskId: null,
      reason: input.reason,
      requestedBy: null,
      epoch,
      stage: input.stage,
      metadata: {},
      createdAt: input.at,
      updatedAt: input.at,
    };
    await client.query(
      `INSERT INTO agentrun_cancel_requests (id, target_kind, target_id, run_id, command_id, session_id, task_id, reason, requested_by, epoch, stage, metadata, created_at, updated_at)
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb, $13, $14)`,
      [record.id, record.targetKind, record.targetId, record.runId, record.commandId, record.sessionId, record.taskId, record.reason, record.requestedBy, record.epoch, record.stage, JSON.stringify(record.metadata), record.createdAt, record.updatedAt],
    );
    return record;
  }

  private async appendCancelStage(client: PoolClient, runId: string, stage: CancelStage, cancel: CancelRequestRecord, extra: JsonRecord = {}): Promise<CancelRequestRecord> {
    const at = nowIso();
    const next: CancelRequestRecord = { ...cancel, stage, updatedAt: at };
    await client.query("UPDATE agentrun_cancel_requests SET stage = $2, updated_at = $3 WHERE id = $1", [cancel.id, stage, at]);
    await this.appendEventWithLockedRun(client, runId, "backend_status", cancelStagePayload(next, stage, extra));
    return next;
  }

  private async appendEventWithLockedRun(client: PoolClient, runId: string, type: EventType, payload: JsonRecord): Promise<RunEvent> {
    const run = await this.requireRunForUpdate(client, runId);
    const eventType = requireEventType(type);
    const fenced = fenceLateEventForCancelledRun(run, eventType, payload);
    const eventPayload = normalizeRunEventPayload(fenced.type, fenced.payload);
    const seq = await this.nextSeq(client, "agentrun_events", runId);
    const event: RunEvent = { id: newId("evt"), runId, seq, type: fenced.type, payload: redactJson(eventPayload), createdAt: nowIso() };
    await client.query("INSERT INTO agentrun_events (id, run_id, seq, type, payload, created_at) VALUES ($1, $2, $3, $4, $5::jsonb, $6)", [event.id, event.runId, event.seq, event.type, JSON.stringify(event.payload), event.createdAt]);
    await client.query("SELECT pg_notify($1, $2)", [runChangeNotificationChannel, runId]);
    await client.query(
      `UPDATE agentrun_sessions
       SET last_event_seq = $2, last_activity_at = $3, updated_at = $3
       WHERE session_id = (SELECT session_ref->>'sessionId' FROM agentrun_runs WHERE id = $1)`,
      [runId, event.seq, event.createdAt],
    );
    if (this.eventOutboxConfig.enabled) {
      const commandId = firstString(event.payload.commandId, event.payload.targetCommandId);
      let command: CommandRecord | null = null;
      if (commandId) {
        const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 AND run_id = $2", [commandId, runId]);
        if (commandResult.rows[0]) command = commandFromRow(commandResult.rows[0]);
      }
      const outboxTopic = this.eventOutboxConfig.topic as string;
      const partitionKey = canonicalAgentRunEventPartitionKey(run);
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", [outboxTopic, partitionKey]);
      const outboxSeqResult = await client.query<{ value: string | number }>("SELECT nextval('agentrun_kafka_event_outbox_seq') AS value");
      const outbox = buildKafkaEventOutboxRecord({ source: this.eventOutboxConfig.source as string, topic: this.eventOutboxConfig.topic as string, outboxSeq: Number(outboxSeqResult.rows[0]?.value), run, command, event });
      await client.query(
        `INSERT INTO agentrun_kafka_event_outbox (id, outbox_seq, event_id, run_id, source_seq, topic, partition_key, value, headers, attempt_count, lease_owner, lease_expires_at, next_attempt_at, last_error, delivered_at, created_at, updated_at)
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9::jsonb, $10, $11, $12, $13, $14::jsonb, $15, $16, $17)`,
        [outbox.id, outbox.outboxSeq, outbox.eventId, outbox.runId, outbox.sourceSeq, outbox.topic, outbox.partitionKey, JSON.stringify(outbox.value), JSON.stringify(outbox.headers), outbox.attemptCount, outbox.leaseOwner, outbox.leaseExpiresAt, outbox.nextAttemptAt, JSON.stringify(outbox.lastError), outbox.deliveredAt, outbox.createdAt, outbox.updatedAt],
      );
    }
    return event;
  }

  private async nextSeq(client: PoolClient, table: "agentrun_commands" | "agentrun_events", runId: string): Promise<number> {
    const result = await client.query<{ seq: number }>(`SELECT COALESCE(MAX(seq), 0) + 1 AS seq FROM ${table} WHERE run_id = $1`, [runId]);
    return Number(result.rows[0]?.seq ?? 1);
  }

  private async nextQueueVersion(client: PoolClient): Promise<number> {
    const result = await client.query<{ version: string | number }>("SELECT nextval('agentrun_queue_version_seq') AS version");
    return Number(result.rows[0]?.version ?? 1);
  }

  private async nextSessionVersion(client: PoolClient): Promise<number> {
    const result = await client.query<{ version: string | number }>("SELECT nextval('agentrun_session_version_seq') AS version");
    return Number(result.rows[0]?.version ?? 1);
  }

  private async getSessionReadCursor(sessionId: string, readerId: string): Promise<SessionReadCursorRecord | null> {
    const result = await this.pool.query("SELECT * FROM agentrun_session_read_cursors WHERE session_id = $1 AND reader_id = $2", [sessionId, readerId]);
    return result.rows[0] ? sessionReadCursorFromRow(result.rows[0]) : null;
  }

  private async loadSessionReadCursors(readerId: string, sessionIds: string[]): Promise<Map<string, SessionReadCursorRecord>> {
    if (sessionIds.length === 0) return new Map();
    const result = await this.pool.query("SELECT * FROM agentrun_session_read_cursors WHERE reader_id = $1 AND session_id = ANY($2::text[])", [readerId, sessionIds]);
    return new Map(result.rows.map((row) => {
      const cursor = sessionReadCursorFromRow(row);
      return [cursor.sessionId, cursor];
    }));
  }

  private async loadQueueReadCursors(readerId: string, taskIds: string[]): Promise<Map<string, QueueReadCursorRecord>> {
    if (taskIds.length === 0) return new Map();
    const result = await this.pool.query("SELECT * FROM agentrun_queue_read_cursors WHERE reader_id = $1 AND task_id = ANY($2::text[])", [readerId, taskIds]);
    return new Map(result.rows.map((row) => {
      const cursor = queueReadCursorFromRow(row);
      return [cursor.taskId, cursor];
    }));
  }

  private async loadQueueTasksForProjection(queue?: string): Promise<QueueTaskRecord[]> {
    if (queue) {
      const result = await this.pool.query("SELECT * FROM agentrun_queue_tasks WHERE queue = $1", [queue]);
      return result.rows.map(queueTaskFromRow);
    }
    const result = await this.pool.query("SELECT * FROM agentrun_queue_tasks");
    return result.rows.map(queueTaskFromRow);
  }

  private async requireRunForUpdate(client: PoolClient, runId: string): Promise<RunRecord> {
    const result = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [runId]);
    const row = result.rows[0];
    if (!row) throw new AgentRunError("schema-invalid", `run ${runId} was not found`, { httpStatus: 404 });
    return runFromRow(row);
  }

  private async createRunWithClient(client: PoolClient, input: CreateRunInput, identity?: CreateRunIdentity): Promise<RunRecord> {
    const at = nowIso();
    if (identity) {
      if (!identity.id.trim()) throw new AgentRunError("schema-invalid", "reserved run identity must be non-empty", { httpStatus: 400 });
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-create-run", identity.id]);
      const existing = await client.query("SELECT * FROM agentrun_runs WHERE id = $1", [identity.id]);
      if (existing.rows[0]) {
        const run = runFromRow(existing.rows[0]);
        assertRunCreateReplay(run, input);
        return run;
      }
    }
    const sessionRef = await this.resolveSessionForRun(client, input, at);
    const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: identity?.id ?? newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null };
    await client.query(
      `INSERT INTO agentrun_runs (id, tenant_id, project_id, workspace_ref, session_ref, resource_bundle_ref, provider_id, backend_profile, execution_policy, trace_sink, status, terminal_status, failure_kind, failure_message, created_at, updated_at, claimed_by, lease_expires_at)
       VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14, $15, $16, $17, $18)`,
      [run.id, run.tenantId, run.projectId, JSON.stringify(run.workspaceRef), JSON.stringify(run.sessionRef), JSON.stringify(run.resourceBundleRef), run.providerId, run.backendProfile, JSON.stringify(run.executionPolicy), JSON.stringify(run.traceSink), run.status, run.terminalStatus, run.failureKind, run.failureMessage, run.createdAt, run.updatedAt, run.claimedBy, run.leaseExpiresAt],
    );
    await this.touchSessionForRun(client, run, { lastRunId: run.id, lastActivityAt: at }, { bumpVersion: false, at });
    return run;
  }

  private async createCommandWithClient(client: PoolClient, runId: string, input: CreateCommandInput): Promise<CommandRecord> {
    const payloadHash = stableHash(input.payload);
    const run = await this.requireRunForUpdate(client, runId);
    if (input.idempotencyKey) {
      const existing = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 AND idempotency_key = $2", [runId, input.idempotencyKey]);
      if (existing.rows[0]) {
        const command = commandFromRow(existing.rows[0]);
        assertSessionCommandReplay(command, input);
        const intent = await this.runnerDispatchIntentForCommandWithClient(client, command.id);
        assertRunnerDispatchReplay(intent, command, input.dispatch);
        return attachRunnerDispatchIntent(command, intent);
      }
    }
    if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 });
    const runCreatedResult = await client.query<{ exists: boolean }>(
      `SELECT EXISTS (
         SELECT 1 FROM agentrun_events
         WHERE run_id = $1
           AND type = 'backend_status'
           AND payload->>'phase' = 'run-created'
       ) AS exists`,
      [runId],
    );
    const needsRunCreatedEvent = runCreatedResult.rows[0]?.exists !== true;
    const seq = await this.nextSeq(client, "agentrun_commands", runId);
    const at = nowIso();
    const { dispatch: _dispatch, ...commandInput } = input;
    const command: CommandRecord = { ...commandInput, id: newId("cmd"), runId, seq, state: "pending", payloadHash, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, acknowledgedAt: null };
    await client.query(
      `INSERT INTO agentrun_commands (id, run_id, seq, type, payload, payload_hash, idempotency_key, state, created_at, updated_at, acknowledged_at)
       VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10, $11)`,
      [command.id, command.runId, command.seq, command.type, JSON.stringify(command.payload), command.payloadHash, command.idempotencyKey ?? null, command.state, command.createdAt, command.updatedAt, command.acknowledgedAt],
    );
    const intent = input.dispatch ? newRunnerDispatchIntent(command, input.dispatch, at) : null;
    if (intent) await this.insertRunnerDispatchIntentWithClient(client, intent);
    if (command.type === "turn") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, activeCommandId: command.id, terminalStatus: null, failureKind: null, title: sessionTitleFromCommand(command), lastActivityAt: at }, { bumpVersion: true, at });
    else if (command.type === "steer") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, lastActivityAt: at }, { bumpVersion: true, at });
    const userMessage = userMessagePayloadForCommand(run, command);
    if (userMessage) await this.appendEventWithLockedRun(client, runId, "user_message", userMessage);
    if (needsRunCreatedEvent) await this.appendEventWithLockedRun(client, runId, "backend_status", runCreatedEventPayload(run));
    await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-created", commandId: command.id, commandType: command.type });
    return attachRunnerDispatchIntent(command, intent);
  }

  private async replayOrRecoverSessionTurnWithClient(client: PoolClient, run: RunRecord, command: CommandRecord, input: SessionTurnAdmissionInput, disposition: "replayed" | "recovered"): Promise<SessionTurnAdmission> {
    assertRunCreateReplay(run, input.run);
    assertSessionCommandReplay(command, input.command);
    const intent = await this.runnerDispatchIntentForCommandWithClient(client, command.id);
    if (intent) {
      assertRunnerDispatchReplay(intent, command, input.command.dispatch);
      return { run, command: attachRunnerDispatchIntent(command, intent), disposition: "replayed", recoveredPriorPartialWrite: false, partialWrite: false };
    }
    if (!input.command.dispatch) return { run, command: attachRunnerDispatchIntent(command, null), disposition, recoveredPriorPartialWrite: false, partialWrite: false };
    if (run.status !== "pending" || run.claimedBy || command.state !== "pending") throw activeSessionAdmissionConflict(input.sessionId, run, command);
    const recovered = newRunnerDispatchIntent(command, input.command.dispatch, nowIso());
    await this.insertRunnerDispatchIntentWithClient(client, recovered);
    await this.appendEventWithLockedRun(client, run.id, "backend_status", {
      phase: "runner-dispatch-recovered",
      reason: "session-runner-admission-partial-write-recovered",
      commandId: command.id,
      dispatchIntentId: recovered.id,
      plannedRunnerJobId: recovered.runnerJobId,
      valuesPrinted: false,
    });
    return { run, command: attachRunnerDispatchIntent(command, recovered), disposition: "recovered", recoveredPriorPartialWrite: true, partialWrite: false };
  }

  private async terminalizeStaleSessionAdmissionWithClient(client: PoolClient, run: RunRecord, command: CommandRecord): Promise<void> {
    const at = nowIso();
    const reason = "session-turn-admission-stale-lease";
    const message = "session turn admission replaced a stale runner lease";
    const updatedCommands = await client.query(
      "UPDATE agentrun_commands SET state = 'failed', updated_at = $2 WHERE run_id = $1 AND state NOT IN ('completed', 'failed', 'cancelled') RETURNING *",
      [run.id, at],
    );
    await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, message, at);
    await client.query(
      `UPDATE agentrun_runs
       SET status = 'failed', terminal_status = 'failed', failure_kind = 'infra-failed', failure_message = $2,
           claimed_by = NULL, lease_expires_at = NULL, updated_at = $3
       WHERE id = $1`,
      [run.id, message, at],
    );
    await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [run.id]);
    await this.appendEventWithLockedRun(client, run.id, "backend_status", {
      phase: "session-turn-admission-stale-terminalized",
      reason,
      commandId: command.id,
      claimedBy: run.claimedBy,
      leaseExpiresAt: run.leaseExpiresAt,
      failureKind: "infra-failed",
      valuesPrinted: false,
    });
    for (const row of updatedCommands.rows) {
      const item = commandFromRow(row);
      await this.appendEventWithLockedRun(client, run.id, "backend_status", {
        phase: "command-terminal",
        reason,
        commandId: item.id,
        state: "failed",
        terminalStatus: "failed",
        failureKind: "infra-failed",
        message,
        threadId: null,
        turnId: null,
      });
    }
    await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", reason, failureKind: "infra-failed", message });
    await this.touchSessionForRun(client, run, {
      executionState: "terminal",
      activeRunId: null,
      activeCommandId: null,
      lastRunId: run.id,
      lastCommandId: command.id,
      terminalStatus: "failed",
      failureKind: "infra-failed",
      lastActivityAt: at,
    }, { bumpVersion: true, at });
  }

  private async runnerDispatchIntentForCommandWithClient(client: PoolClient, commandId: string): Promise<RunnerDispatchIntentRecord | null> {
    const result = await client.query("SELECT * FROM agentrun_runner_dispatch_intents WHERE command_id = $1", [commandId]);
    return result.rows[0] ? runnerDispatchIntentFromRow(result.rows[0]) : null;
  }

  private async insertRunnerDispatchIntentWithClient(client: PoolClient, intent: RunnerDispatchIntentRecord): Promise<void> {
    await client.query(
      `INSERT INTO agentrun_runner_dispatch_intents (id, run_id, command_id, kind, input, input_hash, state, attempt_count, runner_job_id, lease_owner, lease_expires_at, next_attempt_at, last_error, dispatched_at, created_at, updated_at)
       VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10, $11, $12, $13::jsonb, $14, $15, $16)`,
      [intent.id, intent.runId, intent.commandId, intent.kind, JSON.stringify(intent.input), intent.inputHash, intent.state, intent.attemptCount, intent.runnerJobId, intent.leaseOwner, intent.leaseExpiresAt, intent.nextAttemptAt, JSON.stringify(intent.lastError), intent.dispatchedAt, intent.createdAt, intent.updatedAt],
    );
  }

  private async resolveSessionForRun(client: PoolClient, input: CreateRunInput, at: string): Promise<SessionRef | null> {
    if (!input.sessionRef) return null;
    const existing = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionRef.sessionId]);
    if (existing.rows[0]) {
      const session = sessionFromRow(existing.rows[0]);
      assertSessionBoundary(session, input);
      return sessionRefFromRecord(session, input.sessionRef);
    }
    const record: SessionRecord = {
      sessionId: input.sessionRef.sessionId,
      tenantId: input.tenantId,
      projectId: input.projectId,
      backendProfile: input.backendProfile,
      conversationId: input.sessionRef.conversationId ?? null,
      threadId: input.sessionRef.threadId ?? null,
      metadata: input.sessionRef.metadata ?? {},
      version: await this.nextSessionVersion(client),
      executionState: "idle",
      lastRunId: null,
      lastCommandId: null,
      activeRunId: null,
      activeCommandId: null,
      lastEventSeq: 0,
      terminalStatus: null,
      failureKind: null,
      title: titleFromMetadata(input.sessionRef.metadata ?? {}),
      summary: {},
      lastActivityAt: at,
      createdAt: at,
      updatedAt: at,
      expiresAt: input.sessionRef.expiresAt ?? null,
    };
    await client.query(
      `INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, version, execution_state, last_run_id, last_command_id, active_run_id, active_command_id, last_event_seq, terminal_status, failure_kind, title, summary, last_activity_at, created_at, updated_at, expires_at)
       VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18::jsonb, $19, $20, $21, $22)`,
      [record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.version, record.executionState, record.lastRunId, record.lastCommandId, record.activeRunId, record.activeCommandId, record.lastEventSeq, record.terminalStatus, record.failureKind, record.title, JSON.stringify(record.summary), record.lastActivityAt, record.createdAt, record.updatedAt, record.expiresAt],
    );
    return sessionRefFromRecord(record, input.sessionRef);
  }

  private async upsertSessionThread(client: PoolClient, run: RunRecord, threadId: string, turnId: string | null): Promise<void> {
    if (!run.sessionRef?.sessionId) return;
    const at = nowIso();
    const existingResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [run.sessionRef.sessionId]);
    const existing = existingResult.rows[0] ? sessionFromRow(existingResult.rows[0]) : null;
    const metadata = { ...(existing?.metadata ?? {}), ...(run.sessionRef.metadata ?? {}), ...(turnId ? { lastTurnId: turnId } : {}) };
    const record: SessionRecord = {
      sessionId: run.sessionRef.sessionId,
      tenantId: run.tenantId,
      projectId: run.projectId,
      backendProfile: run.backendProfile,
      conversationId: run.sessionRef.conversationId ?? existing?.conversationId ?? null,
      threadId,
      metadata,
      version: await this.nextSessionVersion(client),
      executionState: existing?.executionState ?? "idle",
      lastRunId: existing?.lastRunId ?? run.id,
      lastCommandId: existing?.lastCommandId ?? null,
      activeRunId: existing?.activeRunId ?? null,
      activeCommandId: existing?.activeCommandId ?? null,
      lastEventSeq: existing?.lastEventSeq ?? 0,
      terminalStatus: existing?.terminalStatus ?? null,
      failureKind: existing?.failureKind ?? null,
      title: existing?.title ?? titleFromMetadata(run.sessionRef.metadata ?? {}),
      summary: existing?.summary ?? {},
      lastActivityAt: at,
      createdAt: existing?.createdAt ?? at,
      updatedAt: at,
      expiresAt: run.sessionRef.expiresAt ?? existing?.expiresAt ?? null,
    };
    await client.query(
      `INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, version, execution_state, last_run_id, last_command_id, active_run_id, active_command_id, last_event_seq, terminal_status, failure_kind, title, summary, last_activity_at, created_at, updated_at, expires_at)
       VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18::jsonb, $19, $20, $21, $22)
       ON CONFLICT (session_id) DO UPDATE SET
         tenant_id = EXCLUDED.tenant_id,
         project_id = EXCLUDED.project_id,
         backend_profile = EXCLUDED.backend_profile,
         conversation_id = EXCLUDED.conversation_id,
         thread_id = EXCLUDED.thread_id,
         metadata = EXCLUDED.metadata,
         version = EXCLUDED.version,
         execution_state = EXCLUDED.execution_state,
         last_run_id = EXCLUDED.last_run_id,
         last_command_id = EXCLUDED.last_command_id,
         active_run_id = EXCLUDED.active_run_id,
         active_command_id = EXCLUDED.active_command_id,
         last_event_seq = EXCLUDED.last_event_seq,
         terminal_status = EXCLUDED.terminal_status,
         failure_kind = EXCLUDED.failure_kind,
         title = EXCLUDED.title,
         summary = EXCLUDED.summary,
         last_activity_at = EXCLUDED.last_activity_at,
         updated_at = EXCLUDED.updated_at,
         expires_at = EXCLUDED.expires_at`,
      [record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.version, record.executionState, record.lastRunId, record.lastCommandId, record.activeRunId, record.activeCommandId, record.lastEventSeq, record.terminalStatus, record.failureKind, record.title, JSON.stringify(record.summary), record.lastActivityAt, record.createdAt, record.updatedAt, record.expiresAt],
    );
    const nextSessionRef = sessionRefFromRecord(record, run.sessionRef);
    await client.query("UPDATE agentrun_runs SET session_ref = $2::jsonb, updated_at = $3 WHERE id = $1", [run.id, JSON.stringify(nextSessionRef), at]);
    await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "session-updated", sessionRef: summarizeSessionRef(nextSessionRef), turnId });
  }

  private async touchSessionForRun(client: PoolClient, run: RunRecord, patch: Partial<SessionRecord>, options: { bumpVersion: boolean; at?: string }): Promise<void> {
    const sessionId = run.sessionRef?.sessionId;
    if (!sessionId) return;
    const existingResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [sessionId]);
    const existing = existingResult.rows[0] ? sessionFromRow(existingResult.rows[0]) : null;
    if (!existing) return;
    const at = options.at ?? nowIso();
    const version = options.bumpVersion ? await this.nextSessionVersion(client) : existing.version;
    await client.query(
      `UPDATE agentrun_sessions SET
         version = $2,
         execution_state = $3,
         last_run_id = $4,
         last_command_id = $5,
         active_run_id = $6,
         active_command_id = $7,
         last_event_seq = $8,
         terminal_status = $9,
         failure_kind = $10,
         title = $11,
         summary = $12::jsonb,
         last_activity_at = $13,
         updated_at = $14
       WHERE session_id = $1`,
      [
        sessionId,
        version,
        patch.executionState ?? existing.executionState,
        patch.lastRunId === undefined ? existing.lastRunId : patch.lastRunId,
        patch.lastCommandId === undefined ? existing.lastCommandId : patch.lastCommandId,
        patch.activeRunId === undefined ? existing.activeRunId : patch.activeRunId,
        patch.activeCommandId === undefined ? existing.activeCommandId : patch.activeCommandId,
        patch.lastEventSeq ?? existing.lastEventSeq,
        patch.terminalStatus === undefined ? existing.terminalStatus : patch.terminalStatus,
        patch.failureKind === undefined ? existing.failureKind : patch.failureKind,
        patch.title === undefined ? existing.title : patch.title,
        JSON.stringify(patch.summary ?? existing.summary),
        patch.lastActivityAt ?? at,
        at,
      ],
    );
  }

  private async resolveSessionRunId(sessionId: string, requestedRunId: string | null): Promise<string | null> {
    const session = await this.getSession(sessionId);
    if (!session) throw new AgentRunError("schema-invalid", `session ${sessionId} was not found`, { httpStatus: 404 });
    if (requestedRunId) {
      const run = await this.getRun(requestedRunId);
      if (run.sessionRef?.sessionId !== sessionId) throw new AgentRunError("schema-invalid", `run ${requestedRunId} does not belong to session ${sessionId}`, { httpStatus: 404 });
      return requestedRunId;
    }
    return session.activeRunId ?? session.lastRunId;
  }

  private async releaseTerminalRunClaim(client: PoolClient, run: RunRecord): Promise<RunRecord> {
    if (!isTerminalRunStatus(run.status)) return run;
    return await this.releaseRunnerClaim(client, run);
  }

  private async releaseRunnerClaim(client: PoolClient, run: RunRecord): Promise<RunRecord> {
    let released = run;
    if (run.claimedBy !== null || run.leaseExpiresAt !== null) {
      const result = await client.query("UPDATE agentrun_runs SET claimed_by = NULL, lease_expires_at = NULL, updated_at = $2 WHERE id = $1 RETURNING *", [run.id, nowIso()]);
      released = runFromRow(result.rows[0]);
    }
    await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [run.id]);
    return released;
  }

  private async withTransaction<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
    const client = await this.pool.connect();
    try {
      await client.query("BEGIN");
      const result = await fn(client);
      await client.query("COMMIT");
      return result;
    } catch (error) {
      await client.query("ROLLBACK");
      throw error;
    } finally {
      client.release();
    }
  }
}

function checksumSql(sql: string): string {
  return createHash("sha256").update(sql.trim()).digest("hex");
}

function latestMigrationId(): string {
  return postgresMigrations[postgresMigrations.length - 1]?.id ?? "none";
}

function clamp(value: number, min: number, max: number): number {
  return Math.max(min, Math.min(value, max));
}

function runFromRow(row: QueryResultRow): RunRecord {
  return {
    id: stringValue(row.id),
    tenantId: stringValue(row.tenant_id),
    projectId: stringValue(row.project_id),
    workspaceRef: jsonRecord(row.workspace_ref) as RunRecord["workspaceRef"],
    sessionRef: jsonValue(row.session_ref) as RunRecord["sessionRef"],
    resourceBundleRef: jsonValue(row.resource_bundle_ref) as RunRecord["resourceBundleRef"],
    providerId: stringValue(row.provider_id),
    backendProfile: stringValue(row.backend_profile) as BackendProfile,
    executionPolicy: jsonRecord(row.execution_policy) as RunRecord["executionPolicy"],
    traceSink: jsonValue(row.trace_sink),
    status: stringValue(row.status) as RunStatus,
    terminalStatus: nullableString(row.terminal_status) as TerminalStatus | null,
    failureKind: nullableString(row.failure_kind) as FailureKind | null,
    failureMessage: nullableString(row.failure_message),
    cancelEpoch: Number(row.cancel_epoch ?? 0),
    cancelRequestId: nullableString(row.cancel_request_id),
    cancelRequestedAt: nullableIso(row.cancel_requested_at),
    cancelReason: nullableString(row.cancel_reason),
    createdAt: iso(row.created_at),
    updatedAt: iso(row.updated_at),
    claimedBy: nullableString(row.claimed_by),
    leaseExpiresAt: nullableIso(row.lease_expires_at),
  };
}

function commandFromRow(row: QueryResultRow): CommandRecord {
  return {
    id: stringValue(row.id),
    runId: stringValue(row.run_id),
    seq: Number(row.seq),
    type: stringValue(row.type) as CommandRecord["type"],
    payload: jsonRecord(row.payload),
    payloadHash: stringValue(row.payload_hash),
    ...(nullableString(row.idempotency_key) ? { idempotencyKey: stringValue(row.idempotency_key) } : {}),
    state: stringValue(row.state) as CommandState,
    cancelEpoch: Number(row.cancel_epoch ?? 0),
    cancelRequestId: nullableString(row.cancel_request_id),
    cancelRequestedAt: nullableIso(row.cancel_requested_at),
    cancelReason: nullableString(row.cancel_reason),
    createdAt: iso(row.created_at),
    updatedAt: iso(row.updated_at),
    acknowledgedAt: nullableIso(row.acknowledged_at),
  };
}

function eventFromRow(row: QueryResultRow): RunEvent {
  return { id: stringValue(row.id), runId: stringValue(row.run_id), seq: Number(row.seq), type: stringValue(row.type) as EventType, payload: jsonRecord(row.payload), createdAt: iso(row.created_at) };
}

function runnerFromRow(row: QueryResultRow): RunnerRecord {
  return {
    ...jsonRecord(row.metadata),
    id: stringValue(row.id),
    ...(nullableString(row.run_id) ? { runId: stringValue(row.run_id) } : {}),
    ...(nullableString(row.attempt_id) ? { attemptId: stringValue(row.attempt_id) } : {}),
    ...(nullableString(row.backend_profile) ? { backendProfile: stringValue(row.backend_profile) as BackendProfile } : {}),
    ...(nullableString(row.placement) ? { placement: stringValue(row.placement) } : {}),
    ...(nullableString(row.source_commit) ? { sourceCommit: stringValue(row.source_commit) } : {}),
    registeredAt: iso(row.registered_at),
    heartbeatAt: iso(row.heartbeat_at),
  };
}

function sessionFromRow(row: QueryResultRow): SessionRecord {
  return {
    sessionId: stringValue(row.session_id),
    tenantId: stringValue(row.tenant_id),
    projectId: stringValue(row.project_id),
    backendProfile: stringValue(row.backend_profile) as BackendProfile,
    conversationId: nullableString(row.conversation_id),
    threadId: nullableString(row.thread_id),
    metadata: jsonRecord(row.metadata),
    version: Number(row.version ?? 1),
    executionState: sessionExecutionState(row.execution_state),
    lastRunId: nullableString(row.last_run_id),
    lastCommandId: nullableString(row.last_command_id),
    activeRunId: nullableString(row.active_run_id),
    activeCommandId: nullableString(row.active_command_id),
    lastEventSeq: Number(row.last_event_seq ?? 0),
    terminalStatus: nullableString(row.terminal_status) as TerminalStatus | null,
    failureKind: nullableString(row.failure_kind) as FailureKind | null,
    title: nullableString(row.title),
    summary: jsonRecord(row.summary),
    lastActivityAt: nullableIso(row.last_activity_at),
    createdAt: iso(row.created_at),
    updatedAt: iso(row.updated_at),
    expiresAt: nullableIso(row.expires_at),
    storageKind: (stringValue(row.storage_kind ?? "none") as SessionRecord["storageKind"]) ?? "none",
    storagePvcName: nullableString(row.storage_pvc_name),
    storageNamespace: nullableString(row.storage_namespace),
    storageSizeBytes: row.storage_size_bytes !== null && row.storage_size_bytes !== undefined ? Number(row.storage_size_bytes) : null,
    storageFilesCount: row.storage_files_count !== null && row.storage_files_count !== undefined ? Number(row.storage_files_count) : null,
    storageSha256: nullableString(row.storage_sha256),
    storagePvcPhase: nullableString(row.storage_pvc_phase),
    storageUpdatedAt: nullableIso(row.storage_updated_at),
    storageEvictedAt: nullableIso(row.storage_evicted_at),
    codexRolloutSubdir: stringValue(row.codex_rollout_subdir ?? "sessions"),
  };
}

function sessionReadCursorFromRow(row: QueryResultRow): SessionReadCursorRecord {
  return { sessionId: stringValue(row.session_id), readerId: stringValue(row.reader_id), sessionVersion: Number(row.session_version), readAt: iso(row.read_at) };
}

function queueReadCursorFromRow(row: QueryResultRow): QueueReadCursorRecord {
  return { taskId: stringValue(row.task_id), readerId: stringValue(row.reader_id), taskVersion: Number(row.task_version), readAt: iso(row.read_at) };
}

function sessionExecutionState(value: unknown): SessionRecord["executionState"] {
  if (value === "running" || value === "terminal") return value;
  return "idle";
}

function runnerJobFromRow(row: QueryResultRow): RunnerJobRecord {
  return {
    id: stringValue(row.id),
    runId: stringValue(row.run_id),
    commandId: stringValue(row.command_id),
    idempotencyKey: nullableString(row.idempotency_key),
    payloadHash: stringValue(row.payload_hash),
    attemptId: stringValue(row.attempt_id),
    runnerId: stringValue(row.runner_id),
    namespace: stringValue(row.namespace),
    jobName: stringValue(row.job_name),
    managerUrl: stringValue(row.manager_url),
    image: stringValue(row.image),
    sourceCommit: stringValue(row.source_commit),
    serviceAccountName: nullableString(row.service_account_name),
    result: jsonRecord(row.result),
    createdAt: iso(row.created_at),
    updatedAt: iso(row.updated_at),
  };
}

function queueTaskFromRow(row: QueryResultRow): QueueTaskRecord {
  return {
    id: stringValue(row.id),
    tenantId: stringValue(row.tenant_id),
    projectId: stringValue(row.project_id),
    queue: stringValue(row.queue),
    lane: stringValue(row.lane),
    title: stringValue(row.title),
    priority: Number(row.priority),
    state: stringValue(row.state) as QueueTaskState,
    version: Number(row.version),
    backendProfile: stringValue(row.backend_profile) as BackendProfile,
    providerId: nullableString(row.provider_id),
    workspaceRef: jsonValue(row.workspace_ref) as QueueTaskRecord["workspaceRef"],
    sessionRef: jsonValue(row.session_ref) as QueueTaskRecord["sessionRef"],
    executionPolicy: jsonValue(row.execution_policy) as QueueTaskRecord["executionPolicy"],
    resourceBundleRef: jsonValue(row.resource_bundle_ref) as QueueTaskRecord["resourceBundleRef"],
    payload: jsonRecord(row.payload),
    payloadHash: stringValue(row.payload_hash),
    references: jsonArray(row.references_json) as JsonRecord[],
    metadata: jsonRecord(row.metadata),
    ...(nullableString(row.idempotency_key) ? { idempotencyKey: stringValue(row.idempotency_key) } : {}),
    latestAttempt: jsonValue(row.latest_attempt) as QueueTaskRecord["latestAttempt"],
    sessionPath: nullableString(row.session_path),
    createdAt: iso(row.created_at),
    updatedAt: iso(row.updated_at),
    cancelledAt: nullableIso(row.cancelled_at),
    cancelReason: nullableString(row.cancel_reason),
  };
}

function queueAttemptFromRow(row: QueryResultRow): QueueAttemptRecord {
  return {
    attemptId: stringValue(row.attempt_id),
    taskId: stringValue(row.task_id),
    retryIndex: Number(row.retry_index),
    state: stringValue(row.state) as QueueAttemptRecord["state"],
    idempotencyKey: nullableString(row.idempotency_key),
    reason: nullableString(row.reason),
    payloadHash: stringValue(row.payload_hash),
    previousAttemptId: nullableString(row.previous_attempt_id),
    runId: nullableString(row.run_id),
    commandId: nullableString(row.command_id),
    runnerJobId: nullableString(row.runner_job_id),
    sessionId: nullableString(row.session_id),
    sessionPath: nullableString(row.session_path),
    failureKind: nullableString(row.failure_kind) as FailureKind | null,
    failureMessage: nullableString(row.failure_message),
    createdAt: iso(row.created_at),
    updatedAt: iso(row.updated_at),
    activatedAt: nullableIso(row.activated_at),
    terminalAt: nullableIso(row.terminal_at),
  };
}

function metadataForRunner(runner: RunnerRecord): JsonRecord {
  const { id: _id, runId: _runId, attemptId: _attemptId, backendProfile: _backendProfile, placement: _placement, sourceCommit: _sourceCommit, registeredAt: _registeredAt, heartbeatAt: _heartbeatAt, ...metadata } = runner;
  return redactJson(metadata);
}

function stringValue(value: unknown): string {
  return typeof value === "string" ? value : String(value ?? "");
}

function firstString(...values: unknown[]): string | null {
  for (const value of values) {
    if (typeof value === "string" && value.trim().length > 0) return value.trim();
  }
  return null;
}

function nullableString(value: unknown): string | null {
  return value === null || value === undefined ? null : stringValue(value);
}

function jsonValue(value: unknown): JsonValue {
  if (value === undefined) return null;
  return value as JsonValue;
}

function jsonRecord(value: unknown): JsonRecord {
  return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : {};
}

function jsonArray(value: unknown): JsonValue[] {
  return Array.isArray(value) ? value as JsonValue[] : [];
}

function iso(value: unknown): string {
  if (value instanceof Date) return value.toISOString();
  if (typeof value === "string") return new Date(value).toISOString();
  return new Date(String(value)).toISOString();
}

function nullableIso(value: unknown): string | null {
  return value === null || value === undefined ? null : iso(value);
}