import { Database } from "bun:sqlite";
import { createHash } from "node:crypto";
import { existsSync, mkdirSync } from "node:fs";
import { dirname } from "node:path";
import { isWrapperCommand } from "./command-identity";
import { EventSchema, buildTransitionEvent, type ControllerEvent, type StoredEvent } from "./events";
import type { HostState } from "./status";
import { definitionDigest, ExactGitShaSchema, FeatureDefinitionSchema, resolveRuntimeFeatureDefinition, Sha256DigestSchema, type LoadedFeatureDefinition, type RuntimeFeatureRegistry } from "./delivery/definitions";
import { GateReceiptIdentitySchema, gateReceiptIdentityMatches, type GateReceiptIdentity } from "./delivery/gate-receipt";
import { evaluateDeliveryReadiness, type ReadinessDecision } from "./delivery/readiness";
const deliveryStoreCapability: unique symbol = Symbol("delivery-store-capability");

export const IDEMPOTENCY_RETENTION_MS = 24 * 60 * 60 * 1000;

export const NORMAL_CONTROLLER_META = {
  desired: "available",
  observed: "available",
  dispatch_state: "healthy",
  dispatch_detail: "",
  reconciler_healthy: "1",
} as const;

export const SPINE_CONFIG_INVALID_PREFIX = "spine-config-invalid:";
export const SPINE_CONFIG_INCIDENT_PREFIX = "spine-config:";

const SCHEDULER_TERMINAL_STATES = new Set([
  "completed",
  "failed",
  "blocked",
  "discarded",
  "cancelled",
  "succeeded",
]);

export interface SpineConfigStateRecord {
  configPath: string;
  lastAppliedBodyHash: string;
}

export class StaleRevisionError extends Error {
  constructor(readonly currentRevision: number) {
    super(`stale revision: ${currentRevision}`);
    this.name = "StaleRevisionError";
  }
}

export class ReportWriteError extends Error {
  constructor(message = "report write failed") {
    super(message);
    this.name = "ReportWriteError";
  }
}

export type DeliveryFeatureState = "OFF" | "INTERNAL" | "CANARY" | "ON" | "RETIRED";

export interface DeliveryFeatureStateRecord {
  featureId: string;
  definitionDigest: string;
  state: DeliveryFeatureState;
  deployedSha: string | null;
  acceptanceReceiptId?: string | null;
  smokeReceiptId?: string | null;
  revision: number;
  transitionedAt: string;
}

export interface DeliveryDeploymentRecord {
  deploymentId: string; targetId: string; deployedSha: string; deployedTree: string; artifactDigest: string;
  status: "ACTIVE" | "ROLLED_BACK"; observedAt: string; evidenceReference: string;
}
export interface DeliveryInstalledProofRecord {
  proofId: string; deploymentId: string; targetId: string; deployedSha: string; deployedTree: string; artifactDigest: string;
  entrypoint: string; result: "PASSED" | "FAILED"; observedAt: string; evidenceReference: string;
}
export interface DeliveryFeatureReceiptRecord {
  receiptId: string; featureId: string; definitionDigest: string; kind: "ACCEPTANCE" | "SMOKE"; checkId: string;
  candidateSha: string; candidateTree: string; deploymentId: string; targetId: string; deployedSha: string; deployedTree: string; artifactDigest: string;
  result: "PASSED" | "FAILED"; observedAt: string; evidenceReference: string;
}
export interface GateReceiptRecord {
  receiptId: string;
  identity: GateReceiptIdentity;
  result: "PASSED" | "FAILED";
  recordedAt: string;
  origin: string;
}
export interface ReusableGateReceipt extends GateReceiptRecord {
  ageMs: number;
}

export type TransitionVerb =
  | "box-drain"
  | "box-restore"
  | "host-quarantine"
  | "host-unquarantine"
  | "admission-reconcile"
  | "job-retry"
  | "ci-reconcile"
  | "recall-spill"
  | "delivery-feature-reconcile";

export interface FallbackLease {
  active: boolean;
  expiresAt: string | null;
  host: string | null;
  reason: string | null;
}

export interface GlobalRemoteJobReservationRecord {
  jobId: string;
  createdAt: number;
}

export interface HostSlotReservationRecord {
  jobId: string;
  hostname: string;
  ticketPosition: number;
  reservedAt: number;
  releasedAt: number | null;
}

export interface HostRecord {
  hostname: string;
  state: HostState;
  role: "builder" | "workstation";
  slotsTotal: number;
  slotsUsed: number;
  runningJobs: number;
  ciJobsRunning: number;
  healthStorage: boolean;
  healthRunner: boolean;
  healthOffload: boolean;
  capabilityOk: boolean;
  // Why the last admission probe refused this host; null when it admitted the host.
  capabilityReason: string | null;
  // ISO timestamp of the last admission probe; null when never probed.
  capabilityCheckedAt: string | null;
  primary: boolean;
  enrolling: boolean;
  dispatchPaused: boolean;
  quarantinedCommands: string[];
}

export interface LandConductHealthRecord {
  root: string;
  lastPassAt: string | null;
  lastOk: boolean;
  lastDetail: string;
  consecutiveFailures: number;
}

export type DeployFailureClass = "none" | "transient" | "permanent";

/** Retry-stop state for the S3 deploy watcher, keyed on the single target sha in flight. */
export interface DeployWatcherStateRecord {
  targetSha: string;
  attempts: number;
  lastStatus: string;
  lastDetail: string;
  lastAt: string;
  lastOk: boolean;
  failureClass: DeployFailureClass;
  nextRetryAt: string | null;
}

export type IncidentState = "open" | "resolved";

export interface IncidentRecord {
  key: string;
  firstSeen: string;
  lastSeen: string;
  count: number;
  affectedJobs: string[];
  remediation: string;
  cooldownUntil: string | null;
  autoResolveCondition: string;
  state: IncidentState;
}

export type CapabilityBreakerState = "closed" | "open" | "half-open";

export interface CapabilityBreakerRecord {
  hostname: string;
  command: string;
  state: CapabilityBreakerState;
  failureCount: number;
  missingEventEmitted: boolean;
}

export interface CapabilityManifestRecord {
  repo: string;
  command: string;
  version: string;
  writablePaths: string[];
  minimumDiskBytes: number;
  requiresSystemd: boolean;
}

export interface JobRecord {
  id: string;
  key?: string;
  mirror?: string;
  repo: string;
  host: string;
  snapshot: string;
  stage: string;
  attempt: number;
  rc: number | null;
  infraFailure: boolean;
  startedAt?: string;
  finishedAt?: string | null;
  lastReportAt?: string;
  timeoutSec?: number;
  publication?: {
    state: "none" | "staged" | "promoted" | "blocked" | "discarded";
    reason?: string;
  };
}

export type WorkspacePublicationState =
  | "none"
  | "staged"
  | "promoting"
  | "promoted"
  | "blocked"
  | "discarded";

export interface WorkspaceRecord {
  jobId: string;
  repo: string;
  host: string;
  snapshot: string;
  checkoutGeneration: string;
  checkoutPath: string;
  publicationPath: string;
  workspacePath: string;
  cachePath: string;
  snapshotPath: string;
  overlayPath: string;
  outputPath: string;
  stagingPath: string;
  backupPath: string;
  manifest: unknown | null;
  publicationState: WorkspacePublicationState;
  publicationReason: string | null;
  transportReattachCount: number;
  completedAt: number | null;
  stage: string;
}

export interface QueueTicketRecord {
  position: number;
  key: string;
  repo: string;
  owner: { pid: number; starttime: number; label?: string };
  command?: string;
  enqueuedAt?: number;
  enqueueAgeSeconds: number;
  state: string;
  dispatchTarget?: string;
  placement?: {
    kind: "builder" | "spill";
    host: string;
    jobId: string;
    placedAt: number;
  };
}

export interface TransitionIntent {
  verb: TransitionVerb;
  idempotencyKey: string;
  args: Record<string, unknown>;
  expectedRevision: number;
}

export interface IdempotencyRecord {
  revision: number;
  result: unknown;
}

export interface LandOperationReceipt {
  requestId: string;
  taskId: string;
  root: string;
  worktree: string;
  ref: string;
  commit: string;
  artifact: string;
  verification: string;
  pendingOperation: "land";
  blocker: string;
  nextAction: string;
  rollback: string;
  successorAccount: string;
  successorModel: string;
  queueDir: string;
  ticketId: string;
}

export interface LandOperationRecord {
  operationId: string;
  receipt: LandOperationReceipt;
  fence: number;
  leaseExpiresAt: number;
  state: "waiting" | "succeeded" | "failed";
  verdict: unknown | null;
  successorIntentId: string | null;
}

export interface LandSuccessorDispatch {
  intentId: string;
  dispatchKey: string;
  claimGeneration: number;
  operationId: string;
  fence: number;
  nextAction: string;
  receipt: LandOperationReceipt;
}

export class AuditWriteError extends Error {
  constructor(message = "audit write failed") {
    super(message);
    this.name = "AuditWriteError";
  }
}

export interface ControllerStoreOptions {
  /** Test hook: refuse journal writes. */
  auditWriteFails?: boolean;
  /** Test hook: throw after journaling intent, before commit. */
  crashAfterJournal?: boolean;
  /** Test hook: refuse event writes inside transition commits. */
  eventWriteFails?: boolean;
  now?: () => number;
}

interface MetaRow {
  revision: number;
  desired: HostState;
  observed: HostState;
  leaseActive: boolean;
  leaseExpiresAt: string | null;
  leaseHost: string | null;
  leaseReason: string | null;
  configIncidentEmitted: boolean;
  dispatchState: "healthy" | "wedged" | "paused";
  dispatchDetail: string | null;
  dispatchHost: string | null;
  reconcilerHealthy: boolean;
  reconcilerLastAt: string | null;
}

interface IncidentRow {
  key: string;
  first_seen: string;
  last_seen: string;
  count: number;
  affected_jobs_json: string;
  remediation: string;
  cooldown_until: string | null;
  auto_resolve_condition: string;
  state: string;
}

export class ControllerStore {
  private readonly db: Database;
  private readonly now: () => number;
  private auditWriteFails: boolean;
  private crashAfterJournal: boolean;
  private eventWriteFails: boolean;
  private mutateDepth = 0;

  constructor(
    dbPath: string,
    options: ControllerStoreOptions = {},
  ) {
    this.now = options.now ?? (() => Date.now());
    this.auditWriteFails = options.auditWriteFails ?? false;
    this.crashAfterJournal = options.crashAfterJournal ?? false;
    this.eventWriteFails = options.eventWriteFails ?? false;

    mkdirSync(dirname(dbPath), { recursive: true });
    this.db = new Database(dbPath);
    this.db.exec("PRAGMA journal_mode = WAL");
    this.db.exec("PRAGMA busy_timeout = 5000");
    this.initSchema();
    this.recoverPendingTransitions();
  }

  close(): void {
    this.db.close();
  }

  reopen(dbPath: string, options: ControllerStoreOptions = {}): ControllerStore {
    this.close();
    return new ControllerStore(dbPath, {
      ...options,
      now: options.now ?? this.now,
    });
  }

  static open(dbPath: string, options?: ControllerStoreOptions): ControllerStore {
    return new ControllerStore(dbPath, options);
  }

  private initSchema(): void {
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS meta (
        key TEXT PRIMARY KEY,
        value TEXT NOT NULL
      );

      CREATE TABLE IF NOT EXISTS hosts (
        hostname TEXT PRIMARY KEY,
        state TEXT NOT NULL DEFAULT 'available',
        role TEXT NOT NULL DEFAULT 'builder',
        slots_total INTEGER NOT NULL DEFAULT 4,
        slots_used INTEGER NOT NULL DEFAULT 0,
        running_jobs INTEGER NOT NULL DEFAULT 0,
        ci_jobs_running INTEGER NOT NULL DEFAULT 0,
        health_storage INTEGER NOT NULL DEFAULT 1,
        health_runner INTEGER NOT NULL DEFAULT 1,
        health_offload INTEGER NOT NULL DEFAULT 1,
        capability_ok INTEGER NOT NULL DEFAULT 1,
        primary_host INTEGER NOT NULL DEFAULT 0,
        enrolling INTEGER NOT NULL DEFAULT 0,
        dispatch_paused INTEGER NOT NULL DEFAULT 0
      );

      CREATE TABLE IF NOT EXISTS land_conduct_health (
        root TEXT PRIMARY KEY,
        last_pass_at TEXT,
        last_ok INTEGER NOT NULL DEFAULT 1,
        last_detail TEXT NOT NULL DEFAULT '',
        consecutive_failures INTEGER NOT NULL DEFAULT 0
      );

      CREATE TABLE IF NOT EXISTS deploy_watcher_state (
        id INTEGER PRIMARY KEY CHECK (id = 1),
        target_sha TEXT NOT NULL,
        attempts INTEGER NOT NULL DEFAULT 0,
        last_status TEXT NOT NULL DEFAULT '',
        last_detail TEXT NOT NULL DEFAULT '',
        last_at TEXT NOT NULL,
        last_ok INTEGER NOT NULL DEFAULT 1,
        failure_class TEXT NOT NULL DEFAULT 'none',
        next_retry_at TEXT
      );

      CREATE TABLE IF NOT EXISTS host_quarantines (
        hostname TEXT NOT NULL,
        command TEXT NOT NULL,
        circuit_open INTEGER NOT NULL DEFAULT 1,
        PRIMARY KEY (hostname, command)
      );

      CREATE TABLE IF NOT EXISTS capability_breakers (
        hostname TEXT NOT NULL,
        command TEXT NOT NULL,
        state TEXT NOT NULL DEFAULT 'closed',
        failure_count INTEGER NOT NULL DEFAULT 0,
        missing_event_emitted INTEGER NOT NULL DEFAULT 0,
        PRIMARY KEY (hostname, command)
      );

      CREATE TABLE IF NOT EXISTS capability_manifests (
        repo TEXT NOT NULL,
        command TEXT NOT NULL,
        manifest_json TEXT NOT NULL,
        PRIMARY KEY (repo, command)
      );

      CREATE TABLE IF NOT EXISTS jobs (
        id TEXT PRIMARY KEY,
        repo TEXT NOT NULL,
        host TEXT NOT NULL,
        snapshot TEXT NOT NULL,
        stage TEXT NOT NULL,
        attempt INTEGER NOT NULL DEFAULT 1,
        rc INTEGER,
        infra_failure INTEGER NOT NULL DEFAULT 0,
        publication_state TEXT NOT NULL DEFAULT 'none',
        publication_reason TEXT
      );

      CREATE TABLE IF NOT EXISTS remote_job_reservations (
        job_id TEXT PRIMARY KEY,
        created_at INTEGER NOT NULL
      );

      CREATE TABLE IF NOT EXISTS workspaces (
        job_id TEXT PRIMARY KEY,
        checkout_generation TEXT NOT NULL,
        checkout_path TEXT NOT NULL,
        publication_path TEXT NOT NULL,
        workspace_path TEXT NOT NULL,
        cache_path TEXT NOT NULL,
        snapshot_path TEXT NOT NULL,
        overlay_path TEXT NOT NULL,
        output_path TEXT NOT NULL,
        staging_path TEXT NOT NULL,
        backup_path TEXT NOT NULL,
        manifest_json TEXT,
        publication_state TEXT NOT NULL DEFAULT 'none',
        publication_reason TEXT,
        transport_reattach_count INTEGER NOT NULL DEFAULT 0,
        completed_at INTEGER
      );

      CREATE TABLE IF NOT EXISTS queue_tickets (
        position INTEGER PRIMARY KEY,
        ticket_key TEXT NOT NULL,
        repo TEXT NOT NULL,
        owner_pid INTEGER NOT NULL,
        owner_starttime INTEGER NOT NULL,
        owner_label TEXT,
        command TEXT NOT NULL DEFAULT '',
        enqueued_at INTEGER NOT NULL DEFAULT 0,
        enqueue_age_seconds REAL NOT NULL DEFAULT 0,
        state TEXT NOT NULL DEFAULT 'queued',
        dispatch_target TEXT,
        placement_kind TEXT,
        placement_job_id TEXT,
        placed_at INTEGER
      );

      CREATE TABLE IF NOT EXISTS host_slot_reservations (
        job_id TEXT PRIMARY KEY,
        hostname TEXT NOT NULL,
        ticket_position INTEGER NOT NULL UNIQUE,
        reserved_at INTEGER NOT NULL,
        released_at INTEGER
      );

      CREATE TABLE IF NOT EXISTS idempotency (
        idempotency_key TEXT PRIMARY KEY,
        revision INTEGER NOT NULL,
        result_json TEXT NOT NULL,
        created_at INTEGER NOT NULL
      );

      CREATE TABLE IF NOT EXISTS transition_journal (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        verb TEXT NOT NULL,
        idempotency_key TEXT NOT NULL UNIQUE,
        args_json TEXT NOT NULL,
        expected_revision INTEGER NOT NULL,
        status TEXT NOT NULL DEFAULT 'pending',
        result_json TEXT,
        created_at INTEGER NOT NULL
      );

      CREATE TABLE IF NOT EXISTS incidents (
        key TEXT PRIMARY KEY,
        first_seen TEXT NOT NULL,
        last_seen TEXT NOT NULL,
        count INTEGER NOT NULL,
        affected_jobs_json TEXT NOT NULL,
        remediation TEXT NOT NULL,
        cooldown_until TEXT,
        auto_resolve_condition TEXT NOT NULL,
        state TEXT NOT NULL
      );

      CREATE TABLE IF NOT EXISTS events (
        revision INTEGER PRIMARY KEY,
        payload_json TEXT NOT NULL
      );

      CREATE TABLE IF NOT EXISTS spine_config_state (
        config_path TEXT PRIMARY KEY,
        last_applied_body_hash TEXT NOT NULL
      );

      -- Feature definitions live in Git; only their digest and mutable state live here.
      CREATE TABLE IF NOT EXISTS delivery_feature_state (
        feature_id TEXT PRIMARY KEY,
        definition_digest TEXT NOT NULL,
        state TEXT NOT NULL CHECK(state IN ('OFF', 'INTERNAL', 'CANARY', 'ON', 'RETIRED')),
        deployed_sha TEXT,
        revision INTEGER NOT NULL,
        transitioned_at TEXT NOT NULL
      );

      CREATE TABLE IF NOT EXISTS delivery_deployments (
        deployment_id TEXT PRIMARY KEY, target_id TEXT NOT NULL, deployed_sha TEXT NOT NULL,
        deployed_tree TEXT NOT NULL, artifact_digest TEXT NOT NULL,
        status TEXT NOT NULL CHECK(status IN ('ACTIVE', 'ROLLED_BACK')),
        observed_at TEXT NOT NULL, evidence_reference TEXT NOT NULL
      );
      CREATE UNIQUE INDEX IF NOT EXISTS one_active_delivery_per_target
        ON delivery_deployments(target_id) WHERE status = 'ACTIVE';
      CREATE TABLE IF NOT EXISTS delivery_installed_proofs (
        proof_id TEXT PRIMARY KEY, deployment_id TEXT NOT NULL, target_id TEXT NOT NULL,
        deployed_sha TEXT NOT NULL, deployed_tree TEXT NOT NULL, artifact_digest TEXT NOT NULL, entrypoint TEXT NOT NULL,
        result TEXT NOT NULL CHECK(result IN ('PASSED', 'FAILED')),
        observed_at TEXT NOT NULL, evidence_reference TEXT NOT NULL
      );
      CREATE TRIGGER IF NOT EXISTS delivery_installed_proofs_immutable_update BEFORE UPDATE ON delivery_installed_proofs BEGIN SELECT RAISE(ABORT, 'delivery installed proofs immutable'); END;
      CREATE TRIGGER IF NOT EXISTS delivery_installed_proofs_immutable_delete BEFORE DELETE ON delivery_installed_proofs BEGIN SELECT RAISE(ABORT, 'delivery installed proofs immutable'); END;
      CREATE TABLE IF NOT EXISTS delivery_feature_receipts (
        receipt_id TEXT PRIMARY KEY, feature_id TEXT NOT NULL, definition_digest TEXT NOT NULL,
        kind TEXT NOT NULL CHECK(kind IN ('ACCEPTANCE', 'SMOKE')), check_id TEXT NOT NULL,
        candidate_sha TEXT NOT NULL, candidate_tree TEXT NOT NULL, deployment_id TEXT NOT NULL,
        target_id TEXT NOT NULL, deployed_sha TEXT NOT NULL, deployed_tree TEXT NOT NULL,
        artifact_digest TEXT NOT NULL, result TEXT NOT NULL CHECK(result IN ('PASSED', 'FAILED')),
        observed_at TEXT NOT NULL, evidence_reference TEXT NOT NULL
      );
      CREATE TRIGGER IF NOT EXISTS delivery_receipts_immutable_update BEFORE UPDATE ON delivery_feature_receipts BEGIN SELECT RAISE(ABORT, 'delivery receipts immutable'); END;
      CREATE TRIGGER IF NOT EXISTS delivery_receipts_immutable_delete BEFORE DELETE ON delivery_feature_receipts BEGIN SELECT RAISE(ABORT, 'delivery receipts immutable'); END;
      CREATE TABLE IF NOT EXISTS delivery_evidence_attestations (attestation_id TEXT PRIMARY KEY);
      CREATE TABLE IF NOT EXISTS gate_receipts (
        receipt_id TEXT PRIMARY KEY,
        identity_digest TEXT NOT NULL,
        identity_json TEXT NOT NULL,
        result TEXT NOT NULL CHECK(result IN ('PASSED', 'FAILED')),
        recorded_at TEXT NOT NULL,
        recorded_at_ms INTEGER NOT NULL,
        origin TEXT NOT NULL
      );
      CREATE TRIGGER IF NOT EXISTS gate_receipts_immutable_update BEFORE UPDATE ON gate_receipts BEGIN SELECT RAISE(ABORT, 'gate receipts immutable'); END;
      CREATE TRIGGER IF NOT EXISTS gate_receipts_immutable_delete BEFORE DELETE ON gate_receipts BEGIN SELECT RAISE(ABORT, 'gate receipts immutable'); END;

      CREATE TABLE IF NOT EXISTS land_operation_receipts (
        operation_id TEXT PRIMARY KEY,
        receipt_json TEXT NOT NULL,
        created_at INTEGER NOT NULL
      );
      CREATE TRIGGER IF NOT EXISTS land_operation_receipts_immutable_update BEFORE UPDATE ON land_operation_receipts BEGIN SELECT RAISE(ABORT, 'land operation receipts immutable'); END;
      CREATE TRIGGER IF NOT EXISTS land_operation_receipts_immutable_delete BEFORE DELETE ON land_operation_receipts BEGIN SELECT RAISE(ABORT, 'land operation receipts immutable'); END;
      CREATE TABLE IF NOT EXISTS land_operations (
        operation_id TEXT PRIMARY KEY,
        ticket_id TEXT NOT NULL UNIQUE,
        fence INTEGER NOT NULL,
        lease_expires_at INTEGER NOT NULL,
        state TEXT NOT NULL CHECK(state IN ('waiting', 'succeeded', 'failed')),
        verdict_json TEXT,
        successor_intent_id TEXT
      );
      CREATE TABLE IF NOT EXISTS land_operation_verdict_events (
        ticket_id TEXT PRIMARY KEY,
        operation_id TEXT NOT NULL UNIQUE,
        fence INTEGER NOT NULL,
        verdict_json TEXT NOT NULL,
        created_at INTEGER NOT NULL
      );
      CREATE TABLE IF NOT EXISTS land_successor_dispatches (
        intent_id TEXT PRIMARY KEY,
        operation_id TEXT NOT NULL UNIQUE,
        fence INTEGER NOT NULL,
        state TEXT NOT NULL CHECK(state IN ('pending', 'claimed', 'accepted')),
        next_action TEXT NOT NULL,
        receipt_json TEXT NOT NULL,
        claim_generation INTEGER NOT NULL DEFAULT 0,
        claim_expires_at INTEGER,
        dispatch_key TEXT NOT NULL,
        accepted_at INTEGER
      );
    `);

    this.migrateLandSuccessorDispatches();

    this.db.exec(`
      INSERT OR IGNORE INTO capability_breakers
        (hostname, command, state, failure_count, missing_event_emitted)
      SELECT hostname, command, 'open', 2, 0
      FROM host_quarantines
      WHERE circuit_open = 1
    `);

    this.dropWrapperBreakers();

    this.ensureColumn("delivery_feature_state", "acceptance_receipt_id", "TEXT");
    this.ensureColumn("gate_receipts", "identity_digest", "TEXT");
    this.ensureColumn("gate_receipts", "recorded_at_ms", "INTEGER");
    this.db.exec("CREATE INDEX IF NOT EXISTS reusable_gate_receipts ON gate_receipts(identity_digest, recorded_at_ms DESC) WHERE result = 'PASSED'");
    this.ensureColumn("delivery_feature_state", "smoke_receipt_id", "TEXT");
    this.ensureColumn("jobs", "publication_state", "TEXT NOT NULL DEFAULT 'none'");
    this.ensureColumn("jobs", "publication_reason", "TEXT");
    this.ensureColumn("queue_tickets", "command", "TEXT NOT NULL DEFAULT ''");
    this.ensureColumn("queue_tickets", "enqueued_at", "INTEGER NOT NULL DEFAULT 0");
    this.ensureColumn("queue_tickets", "placement_kind", "TEXT");
    this.ensureColumn("queue_tickets", "placement_job_id", "TEXT");
    this.ensureColumn("queue_tickets", "placed_at", "INTEGER");
    this.ensureColumn("hosts", "dispatch_paused", "INTEGER NOT NULL DEFAULT 0");
    this.ensureColumn("hosts", "capability_reason", "TEXT");
    this.ensureColumn("hosts", "capability_checked_at", "TEXT");
    this.ensureColumn("jobs", "job_key", "TEXT");
    this.ensureColumn("jobs", "mirror", "TEXT");
    this.ensureColumn("jobs", "started_at", "TEXT");
    this.ensureColumn("jobs", "finished_at", "TEXT");
    this.ensureColumn("jobs", "last_report_at", "TEXT");
    this.ensureColumn("jobs", "timeout_sec", "INTEGER");
    this.ensureColumn("land_successor_dispatches", "claim_generation", "INTEGER NOT NULL DEFAULT 0");
    this.ensureColumn("land_successor_dispatches", "claim_expires_at", "INTEGER");
    this.ensureColumn("land_successor_dispatches", "dispatch_key", "TEXT");
    this.ensureColumn("land_successor_dispatches", "accepted_at", "INTEGER");
    this.ensureColumn("deploy_watcher_state", "failure_class", "TEXT NOT NULL DEFAULT 'none'");
    this.ensureColumn("deploy_watcher_state", "next_retry_at", "TEXT");
    this.mutate(() => {
      this.db.prepare("UPDATE land_successor_dispatches SET dispatch_key = intent_id WHERE dispatch_key IS NULL").run();
    });
    this.migrateLegacyIncidents();
    this.mutate(() => {
      this.db.exec(`
        DELETE FROM queue_tickets
        WHERE rowid NOT IN (
          SELECT MIN(rowid) FROM queue_tickets GROUP BY ticket_key
        )
      `);
      this.db.exec(
        "CREATE UNIQUE INDEX IF NOT EXISTS queue_tickets_key ON queue_tickets(ticket_key)",
      );
    });

    const count = this.db
      .query<{ c: number }, []>("SELECT COUNT(*) AS c FROM meta")
      .get();
    if ((count?.c ?? 0) === 0) {
      this.setMetaDefaults();
    }
  }

  private migrateLandSuccessorDispatches(): void {
    const row = this.db.query<{ sql: string }, [string]>("SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?").get("land_successor_dispatches");
    if (!row || row.sql.includes("'claimed'") && row.sql.includes("'accepted'")) return;
    this.mutate(() => {
      this.db.exec(`
        ALTER TABLE land_successor_dispatches RENAME TO legacy_land_successor_dispatches;
        CREATE TABLE land_successor_dispatches (
          intent_id TEXT PRIMARY KEY,
          operation_id TEXT NOT NULL UNIQUE,
          fence INTEGER NOT NULL,
          state TEXT NOT NULL CHECK(state IN ('pending', 'claimed', 'accepted')),
          next_action TEXT NOT NULL,
          receipt_json TEXT NOT NULL,
          claim_generation INTEGER NOT NULL DEFAULT 0,
          claim_expires_at INTEGER,
          dispatch_key TEXT NOT NULL,
          accepted_at INTEGER
        );
        INSERT INTO land_successor_dispatches (
          intent_id, operation_id, fence, state, next_action, receipt_json, dispatch_key, accepted_at
        )
        SELECT intent_id, operation_id, fence,
          CASE state WHEN 'dispatched' THEN 'accepted' ELSE 'pending' END,
          next_action, receipt_json, intent_id,
          CASE WHEN state = 'dispatched' THEN dispatched_at ELSE NULL END
        FROM legacy_land_successor_dispatches;
        DROP TABLE legacy_land_successor_dispatches;
      `);
    });
  }

  private dropWrapperBreakers(): void {
    const rows = this.db
      .query<{ hostname: string; command: string }, []>(
        "SELECT hostname, command FROM capability_breakers",
      )
      .all();
    const drop = rows.filter((row) => isWrapperCommand(row.command));
    if (drop.length === 0) return;
    const statement = this.db.prepare(
      "DELETE FROM capability_breakers WHERE hostname = ? AND command = ?",
    );
    for (const row of drop) statement.run(row.hostname, row.command);
  }

  private ensureColumn(table: string, column: string, definition: string): void {
    const columns = this.db
      .query<{ name: string }, []>(`PRAGMA table_info(${table})`)
      .all();
    if (columns.some((entry) => entry.name === column)) return;
    this.db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`);
  }

  private migrateLegacyIncidents(): void {
    const columns = this.db
      .query<{ name: string }, []>("PRAGMA table_info(incidents)")
      .all();
    if (columns.some(({ name }) => name === "key")) return;
    this.mutate(() => {
      this.db.exec(`
        ALTER TABLE incidents RENAME TO legacy_incidents;
        CREATE TABLE incidents (
          key TEXT PRIMARY KEY,
          first_seen TEXT NOT NULL,
          last_seen TEXT NOT NULL,
          count INTEGER NOT NULL,
          affected_jobs_json TEXT NOT NULL,
          remediation TEXT NOT NULL,
          cooldown_until TEXT,
          auto_resolve_condition TEXT NOT NULL,
          state TEXT NOT NULL
        );
        INSERT INTO incidents (
          key, first_seen, last_seen, count, affected_jobs_json, remediation,
          cooldown_until, auto_resolve_condition, state
        )
        SELECT
          kind || ':' || id,
          strftime('%Y-%m-%dT%H:%M:%fZ', created_at / 1000.0, 'unixepoch'),
          strftime('%Y-%m-%dT%H:%M:%fZ', created_at / 1000.0, 'unixepoch'),
          1, '[]', detail, NULL, 'operator resolves legacy incident', 'open'
        FROM legacy_incidents;
        DROP TABLE legacy_incidents;
      `);
    });
  }

  private setMetaDefaults(): void {
    const defaults: Record<string, string> = {
      revision: "0",
      desired: NORMAL_CONTROLLER_META.desired,
      observed: NORMAL_CONTROLLER_META.observed,
      lease_active: "0",
      lease_expires_at: "",
      lease_host: "",
      lease_reason: "",
      config_incident_emitted: "0",
      dispatch_state: NORMAL_CONTROLLER_META.dispatch_state,
      dispatch_detail: NORMAL_CONTROLLER_META.dispatch_detail,
      dispatch_host: "",
      reconciler_healthy: NORMAL_CONTROLLER_META.reconciler_healthy,
      reconciler_last_at: "",
      incident_cursor: "0",
    };
    const insert = this.db.prepare(
      "INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)",
    );
    for (const [key, value] of Object.entries(defaults)) {
      insert.run(key, value);
    }
  }

  private getMeta(key: string): string {
    const row = this.db
      .query<{ value: string }, [string]>("SELECT value FROM meta WHERE key = ?")
      .get(key);
    return row?.value ?? "";
  }

  private setMeta(key: string, value: string): void {
    this.db
      .prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)")
      .run(key, value);
  }

  getMetaSnapshot(): MetaRow {
    return {
      revision: Number(this.getMeta("revision")),
      desired: this.getMeta("desired") as HostState,
      observed: this.getMeta("observed") as HostState,
      leaseActive: this.getMeta("lease_active") === "1",
      leaseExpiresAt: this.getMeta("lease_expires_at") || null,
      leaseHost: this.getMeta("lease_host") || null,
      leaseReason: this.getMeta("lease_reason") || null,
      configIncidentEmitted: this.getMeta("config_incident_emitted") === "1",
      dispatchState: (this.getMeta("dispatch_state") || "healthy") as
        | "healthy"
        | "wedged"
        | "paused",
      dispatchDetail: this.getMeta("dispatch_detail") || null,
      dispatchHost: this.getMeta("dispatch_host") || null,
      reconcilerHealthy: this.getMeta("reconciler_healthy") !== "0",
      reconcilerLastAt: this.getMeta("reconciler_last_at") || null,
    };
  }

  getRevision(): number {
    return this.getMetaSnapshot().revision;
  }

  getDeliveryFeatureState(featureId: string): DeliveryFeatureStateRecord | null {
    const row = this.db.query<{
      feature_id: string;
      definition_digest: string;
      state: DeliveryFeatureState;
      deployed_sha: string | null;
      acceptance_receipt_id: string | null;
      smoke_receipt_id: string | null;
      revision: number;
      transitioned_at: string;
    }, [string]>(`
      SELECT feature_id, definition_digest, state, deployed_sha, acceptance_receipt_id, smoke_receipt_id, revision, transitioned_at
      FROM delivery_feature_state WHERE feature_id = ?
    `).get(featureId);
    if (!row) return null;
    return {
      featureId: row.feature_id,
      definitionDigest: row.definition_digest,
      state: row.state,
      deployedSha: row.deployed_sha,
      acceptanceReceiptId: row.acceptance_receipt_id,
      smokeReceiptId: row.smoke_receipt_id,
      revision: row.revision,
      transitionedAt: row.transitioned_at,
    };
  }

  private transitionDeliveryFeatureState(input: {
    featureId: string;
    definitionDigest: string;
    state: DeliveryFeatureState;
    deployedSha?: string | null;
    acceptanceReceiptId?: string | null;
    smokeReceiptId?: string | null;
    expectedRevision: number;
  }): DeliveryFeatureStateRecord {
    let result: DeliveryFeatureStateRecord | null = null;
    this.mutate(() => {
      const currentRevision = this.getRevision();
      if (currentRevision !== input.expectedRevision) throw new StaleRevisionError(currentRevision);
      const current = this.getDeliveryFeatureState(input.featureId);
      const legal = current === null
        ? (input.state === "OFF" || input.state === "INTERNAL")
        : (current.state === "OFF" && input.state === "INTERNAL")
          || (current.state === "INTERNAL" && (input.state === "CANARY" || input.state === "RETIRED" || input.state === "OFF"))
          || (current.state === "CANARY" && (input.state === "ON" || input.state === "RETIRED" || input.state === "OFF"))
          || (current.state === "ON" && (input.state === "RETIRED" || input.state === "OFF"));
      if (!legal) throw new Error(`illegal delivery feature transition ${current?.state ?? "absent"} -> ${input.state}`);
      if (input.state !== "OFF" && (!ExactGitShaSchema.safeParse(input.deployedSha).success || !input.acceptanceReceiptId || !input.smokeReceiptId)) {
        throw new Error("enabled delivery feature state requires exact deployed SHA and receipt authorization");
      }
      if (input.state === "OFF" && (input.acceptanceReceiptId || input.smokeReceiptId)) throw new Error("OFF state cannot retain receipt authorization");
      const revision = currentRevision + 1;
      const transitionedAt = new Date(this.now()).toISOString();
      this.db.prepare(`
        INSERT INTO delivery_feature_state
          (feature_id, definition_digest, state, deployed_sha, acceptance_receipt_id, smoke_receipt_id, revision, transitioned_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(feature_id) DO UPDATE SET
          definition_digest = excluded.definition_digest,
          state = excluded.state,
          deployed_sha = excluded.deployed_sha,
          acceptance_receipt_id = excluded.acceptance_receipt_id,
          smoke_receipt_id = excluded.smoke_receipt_id,
          revision = excluded.revision,
          transitioned_at = excluded.transitioned_at
      `).run(input.featureId, input.definitionDigest, input.state, input.deployedSha ?? null, input.state === "OFF" ? null : input.acceptanceReceiptId ?? null, input.state === "OFF" ? null : input.smokeReceiptId ?? null, revision, transitionedAt);
      this.setMeta("revision", String(revision));
      result = {
        featureId: input.featureId,
        definitionDigest: input.definitionDigest,
        state: input.state,
        deployedSha: input.deployedSha ?? null,
        acceptanceReceiptId: input.state === "OFF" ? null : input.acceptanceReceiptId ?? null,
        smokeReceiptId: input.state === "OFF" ? null : input.smokeReceiptId ?? null,
        revision,
        transitionedAt,
      };
    });
    if (result === null) throw new Error("delivery feature transition did not persist");
    return result;
  }

  transitionDeliveryFeatureStateAuthorized(capability: typeof deliveryStoreCapability, input: {
    definition: LoadedFeatureDefinition;
    state: DeliveryFeatureState;
    targetId?: string;
    acceptanceReceiptId?: string;
    smokeReceiptId?: string;
    expectedRevision: number;
  }): DeliveryFeatureStateRecord {
    if (capability !== deliveryStoreCapability) throw new Error("delivery store capability required");
    const parsed = FeatureDefinitionSchema.parse(input.definition.definition);
    if (definitionDigest(parsed) !== input.definition.digest) throw new Error("definition digest authorization mismatch");
    if (input.state === "OFF") {
      return this.transitionDeliveryFeatureState({ featureId: parsed.id, definitionDigest: input.definition.digest, state: "OFF", expectedRevision: input.expectedRevision });
    }
    const reviewDeadline = Date.parse(`${parsed.reviewDate}T23:59:59.999Z`);
    if (input.state !== "RETIRED" && this.now() > reviewDeadline) throw new Error("expired definition cannot authorize state transition");
    if (input.state !== "RETIRED" && parsed.activationPolicy !== "auto-after-receipts") throw new Error("activation policy does not authorize automatic transition");
    if (!input.targetId || !input.acceptanceReceiptId || !input.smokeReceiptId) throw new Error("deployment and receipt authorization required");
    const deployment = this.getActiveDeliveryDeployment(input.targetId);
    if (!deployment) throw new Error("active deployment authorization required");
    const receipts = this.listDeliveryFeatureReceipts(parsed.id);
    const matching = (receipt: DeliveryFeatureReceiptRecord) => receipt.result === "PASSED" && receipt.featureId === parsed.id
      && receipt.definitionDigest === input.definition.digest && receipt.deploymentId === deployment.deploymentId
      && receipt.targetId === deployment.targetId && receipt.candidateSha === deployment.deployedSha
      && receipt.candidateTree === deployment.deployedTree && receipt.deployedSha === deployment.deployedSha
      && receipt.deployedTree === deployment.deployedTree && receipt.artifactDigest === deployment.artifactDigest;
    const acceptance = receipts.find((receipt) => receipt.receiptId === input.acceptanceReceiptId && receipt.kind === "ACCEPTANCE" && matching(receipt));
    const smoke = receipts.find((receipt) => receipt.receiptId === input.smokeReceiptId && receipt.kind === "SMOKE" && matching(receipt));
    if (!acceptance) throw new Error("exact acceptance receipt authorization required");
    if (!smoke) throw new Error("exact smoke receipt authorization required");
    if (!parsed.readinessChecks.every((check) => receipts.some((receipt) => receipt.checkId === check && matching(receipt)))) throw new Error("readiness receipt authorization incomplete");
    return this.transitionDeliveryFeatureState({
      featureId: parsed.id,
      definitionDigest: input.definition.digest,
      state: input.state,
      deployedSha: deployment.deployedSha,
      acceptanceReceiptId: acceptance.receiptId,
      smokeReceiptId: smoke.receiptId,
      expectedRevision: input.expectedRevision,
    });
  }

  recordGateReceipt(input: GateReceiptRecord): GateReceiptRecord {
    const identity = GateReceiptIdentitySchema.parse(input.identity);
    const recordedAtMs = Date.parse(input.recordedAt);
    if (!validId(input.receiptId) || !validId(input.origin) || !["PASSED", "FAILED"].includes(input.result)
      || !Number.isFinite(recordedAtMs) || new Date(recordedAtMs).toISOString() !== input.recordedAt) {
      throw new Error("invalid gate receipt");
    }
    const identityJson = JSON.stringify(identity);
    const identityDigest = createHash("sha256").update(identityJson).digest("hex");
    this.mutate(() => {
      const existing = this.db.query<any, [string]>("SELECT * FROM gate_receipts WHERE receipt_id = ?").get(input.receiptId);
      const stored = { ...input, identity };
      if (existing) {
        const current: GateReceiptRecord = { receiptId: existing.receipt_id, identity: GateReceiptIdentitySchema.parse(JSON.parse(existing.identity_json)), result: existing.result, recordedAt: existing.recorded_at, origin: existing.origin };
        if (JSON.stringify(current) !== JSON.stringify(stored)) throw new Error("immutable gate receipt ID payload mismatch");
        return;
      }
      this.db.prepare("INSERT INTO gate_receipts (receipt_id,identity_digest,identity_json,result,recorded_at,recorded_at_ms,origin) VALUES (?,?,?,?,?,?,?)")
        .run(input.receiptId, identityDigest, identityJson, input.result, input.recordedAt, recordedAtMs, input.origin);
    });
    return { ...input, identity };
  }

  findReusableGateReceipt(identity: GateReceiptIdentity, now = this.now()): ReusableGateReceipt | null {
    const requested = GateReceiptIdentitySchema.parse(identity);
    if (!Number.isFinite(now)) throw new Error("invalid gate receipt lookup time");
    const identityDigest = createHash("sha256").update(JSON.stringify(requested)).digest("hex");
    const row = this.db.query<any, [string, number]>("SELECT * FROM gate_receipts WHERE result = 'PASSED' AND identity_digest = ? AND recorded_at_ms <= ? ORDER BY recorded_at_ms DESC LIMIT 1").get(identityDigest, now);
    if (!row) return null;
    try {
      const receipt: GateReceiptRecord = { receiptId: row.receipt_id, identity: GateReceiptIdentitySchema.parse(JSON.parse(row.identity_json)), result: row.result, recordedAt: row.recorded_at, origin: row.origin };
      const recordedAtMs = Date.parse(receipt.recordedAt);
      if (!validId(receipt.receiptId) || !validId(receipt.origin) || receipt.result !== "PASSED"
        || !Number.isFinite(recordedAtMs) || new Date(recordedAtMs).toISOString() !== receipt.recordedAt
        || row.recorded_at_ms !== recordedAtMs || !gateReceiptIdentityMatches(receipt.identity, requested)) return null;
      return { ...receipt, ageMs: now - recordedAtMs };
    } catch {
      return null;
    }
  }

  recordDeliveryDeployment(capability: typeof deliveryStoreCapability, input: DeliveryDeploymentRecord): DeliveryDeploymentRecord {
    if (capability !== deliveryStoreCapability) throw new Error("delivery store capability required");
    validateDeployment(input);
    this.mutate(() => {
      const existing = this.getDeliveryDeployment(input.deploymentId);
      if (existing) { if (JSON.stringify(existing) !== JSON.stringify(input)) throw new Error("immutable deployment ID payload mismatch"); return; }
      if (input.status === "ACTIVE") this.db.prepare("UPDATE delivery_deployments SET status = 'ROLLED_BACK' WHERE target_id = ? AND status = 'ACTIVE'").run(input.targetId);
      this.db.prepare(`INSERT INTO delivery_deployments (deployment_id,target_id,deployed_sha,deployed_tree,artifact_digest,status,observed_at,evidence_reference) VALUES (?,?,?,?,?,?,?,?)`).run(input.deploymentId,input.targetId,input.deployedSha,input.deployedTree,input.artifactDigest,input.status,input.observedAt,input.evidenceReference);
    });
    return input;
  }

  claimDeliveryEvidenceAttestation(attestationId: string): void {
    if (!validId(attestationId)) throw new Error("invalid delivery evidence attestation ID");
    if (this.db.prepare("INSERT OR IGNORE INTO delivery_evidence_attestations (attestation_id) VALUES (?)").run(attestationId).changes !== 1) throw new Error("delivery evidence attestation replay");
  }

  getDeliveryDeployment(deploymentId: string): DeliveryDeploymentRecord | null {
    const row = this.db.query<any, [string]>("SELECT * FROM delivery_deployments WHERE deployment_id = ?").get(deploymentId);
    return row ? deploymentRow(row) : null;
  }
  getActiveDeliveryDeployment(targetId: string): DeliveryDeploymentRecord | null {
    const row = this.db.query<any, [string]>("SELECT * FROM delivery_deployments WHERE target_id = ? AND status = 'ACTIVE'").get(targetId);
    return row ? deploymentRow(row) : null;
  }
  recordDeliveryInstalledProof(capability: typeof deliveryStoreCapability, input: DeliveryInstalledProofRecord): DeliveryInstalledProofRecord {
    if (capability !== deliveryStoreCapability) throw new Error("delivery store capability required");
    validateInstalledProof(input);
    const deployment = this.getDeliveryDeployment(input.deploymentId);
    if (!deployment || deployment.targetId !== input.targetId || deployment.deployedSha !== input.deployedSha || deployment.deployedTree !== input.deployedTree || deployment.artifactDigest !== input.artifactDigest) throw new Error("exact deployment required for installed proof");
    this.mutate(() => {
      const existing = this.getDeliveryInstalledProof(input.proofId);
      if (existing) {
        if (JSON.stringify(existing) !== JSON.stringify(input)) throw new Error("immutable installed proof ID payload mismatch");
        return;
      }
      this.db.prepare(`INSERT INTO delivery_installed_proofs (proof_id,deployment_id,target_id,deployed_sha,deployed_tree,artifact_digest,entrypoint,result,observed_at,evidence_reference) VALUES (?,?,?,?,?,?,?,?,?,?)`).run(input.proofId,input.deploymentId,input.targetId,input.deployedSha,input.deployedTree,input.artifactDigest,input.entrypoint,input.result,input.observedAt,input.evidenceReference);
    });
    return input;
  }
  getDeliveryInstalledProof(proofId: string): DeliveryInstalledProofRecord | null {
    const row = this.db.query<any, [string]>("SELECT * FROM delivery_installed_proofs WHERE proof_id = ?").get(proofId);
    return row ? installedProofRow(row) : null;
  }
  recordDeliveryFeatureReceipt(capability: typeof deliveryStoreCapability, input: DeliveryFeatureReceiptRecord): DeliveryFeatureReceiptRecord {
    if (capability !== deliveryStoreCapability) throw new Error("delivery store capability required");
    validateReceipt(input);
    this.mutate(() => {
      const existing = this.getDeliveryFeatureReceipt(input.receiptId);
      if (existing) { if (JSON.stringify(existing) !== JSON.stringify(input)) throw new Error("immutable receipt ID payload mismatch"); return; }
      this.db.prepare(`INSERT INTO delivery_feature_receipts (receipt_id,feature_id,definition_digest,kind,check_id,candidate_sha,candidate_tree,deployment_id,target_id,deployed_sha,deployed_tree,artifact_digest,result,observed_at,evidence_reference) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`).run(input.receiptId,input.featureId,input.definitionDigest,input.kind,input.checkId,input.candidateSha,input.candidateTree,input.deploymentId,input.targetId,input.deployedSha,input.deployedTree,input.artifactDigest,input.result,input.observedAt,input.evidenceReference);
    });
    return input;
  }
  getDeliveryFeatureReceipt(receiptId: string): DeliveryFeatureReceiptRecord | null {
    const row = this.db.query<any, [string]>("SELECT * FROM delivery_feature_receipts WHERE receipt_id = ?").get(receiptId);
    return row ? receiptRow(row) : null;
  }
  listDeliveryFeatureReceipts(featureId: string): DeliveryFeatureReceiptRecord[] {
    return this.db.query<any, [string]>("SELECT * FROM delivery_feature_receipts WHERE feature_id = ? ORDER BY receipt_id").all(featureId).map(receiptRow);
  }

  removeRetiredDeliveryFeature(capability: typeof deliveryStoreCapability, definition: LoadedFeatureDefinition, expectedRevision: number): void {
    if (capability !== deliveryStoreCapability) throw new Error("delivery store capability required");
    const state = this.getDeliveryFeatureState(definition.definition.id);
    if (!state || state.state !== "RETIRED") throw new Error("retired delivery feature required");
    if (this.getRevision() !== expectedRevision) throw new StaleRevisionError(this.getRevision());
    const removeAfter = Date.parse(state.transitionedAt) + definition.definition.rollbackWindowDays * 86_400_000;
    if (this.now() < removeAfter) throw new Error("rollback window remains open");
    this.db.exec("DROP TRIGGER delivery_receipts_immutable_delete");
    this.db.prepare("DELETE FROM delivery_feature_receipts WHERE feature_id = ?").run(definition.definition.id);
    this.db.exec("CREATE TRIGGER delivery_receipts_immutable_delete BEFORE DELETE ON delivery_feature_receipts BEGIN SELECT RAISE(ABORT, 'delivery receipts immutable'); END");
    this.db.prepare("DELETE FROM delivery_feature_state WHERE feature_id = ?").run(definition.definition.id);
  }

  resolveDeliveryTarget(featureId: string): string | null {
    const targets = new Set(this.listDeliveryFeatureReceipts(featureId).map((receipt) => receipt.targetId));
    const active = [...targets].filter((targetId) => this.getActiveDeliveryDeployment(targetId) !== null);
    return active.length === 1 ? active[0]! : null;
  }

  getIncidentCursor(): number {
    return Number(this.getMeta("incident_cursor") || "0");
  }

  setIncidentCursor(revision: number): void {
    this.mutate(() => this.advanceIncidentCursor(revision));
  }

  getLandOperation(operationId: string): LandOperationRecord | null {
    const row = this.db.query<{
      operation_id: string; receipt_json: string; fence: number; lease_expires_at: number;
      state: "waiting" | "succeeded" | "failed"; verdict_json: string | null; successor_intent_id: string | null;
    }, [string]>(`
      SELECT o.operation_id, r.receipt_json, o.fence, o.lease_expires_at, o.state, o.verdict_json, o.successor_intent_id
      FROM land_operations o JOIN land_operation_receipts r ON r.operation_id = o.operation_id
      WHERE o.operation_id = ?
    `).get(operationId);
    return row ? landOperationRow(row) : null;
  }

  acceptLandOperation(input: {
    operationId: string;
    receipt: LandOperationReceipt;
    leaseExpiresAt: number;
  }): LandOperationRecord {
    let record: LandOperationRecord | null = null;
    this.mutate(() => {
      const existing = this.getLandOperation(input.operationId);
      if (existing) {
        if (JSON.stringify(existing.receipt) !== JSON.stringify(input.receipt)) throw new Error("immutable land operation receipt mismatch");
        record = existing;
        return;
      }
      const ticketExisting = this.db.query<{ operation_id: string }, [string]>(
        "SELECT operation_id FROM land_operations WHERE ticket_id = ?",
      ).get(input.receipt.ticketId);
      if (ticketExisting) throw new Error("land ticket already belongs to another operation");
      const fence = (this.db.query<{ fence: number }, []>("SELECT COALESCE(MAX(fence), 0) + 1 AS fence FROM land_operations").get()?.fence ?? 1);
      this.db.prepare("INSERT INTO land_operation_receipts (operation_id, receipt_json, created_at) VALUES (?, ?, ?)")
        .run(input.operationId, JSON.stringify(input.receipt), this.now());
      this.db.prepare("INSERT INTO land_operations (operation_id, ticket_id, fence, lease_expires_at, state) VALUES (?, ?, ?, ?, 'waiting')")
        .run(input.operationId, input.receipt.ticketId, fence, input.leaseExpiresAt);
      record = { operationId: input.operationId, receipt: input.receipt, fence, leaseExpiresAt: input.leaseExpiresAt, state: "waiting", verdict: null, successorIntentId: null };
    });
    if (!record) throw new Error("land operation acceptance did not persist");
    return record;
  }

  listLandOperationsAwaitingVerdict(): LandOperationRecord[] {
    return this.db.query<{
      operation_id: string; receipt_json: string; fence: number; lease_expires_at: number;
      state: "waiting" | "succeeded" | "failed"; verdict_json: string | null; successor_intent_id: string | null;
    }, []>(`
      SELECT o.operation_id, r.receipt_json, o.fence, o.lease_expires_at, o.state, o.verdict_json, o.successor_intent_id
      FROM land_operations o JOIN land_operation_receipts r ON r.operation_id = o.operation_id
      WHERE o.state = 'waiting' ORDER BY o.operation_id
    `).all().map(landOperationRow);
  }

  recoverExpiredLandOperationLease(operationId: string, leaseExpiresAt: number): LandOperationRecord | null {
    let record: LandOperationRecord | null = null;
    this.mutate(() => {
      const operation = this.getLandOperation(operationId);
      if (!operation || operation.state !== "waiting" || operation.leaseExpiresAt > this.now()) return;
      const changes = this.db.prepare(
        "UPDATE land_operations SET fence = ?, lease_expires_at = ? WHERE operation_id = ? AND state = 'waiting' AND fence = ? AND lease_expires_at = ?",
      ).run(operation.fence + 1, leaseExpiresAt, operationId, operation.fence, operation.leaseExpiresAt);
      if (changes.changes !== 1) return;
      record = { ...operation, fence: operation.fence + 1, leaseExpiresAt };
    });
    return record;
  }

  recordLandOperationVerdict(input: { operationId: string; ticketId: string; fence: number; verdict: unknown }): LandOperationRecord {
    let record: LandOperationRecord | null = null;
    this.mutate(() => {
      const operation = this.getLandOperation(input.operationId);
      if (!operation || operation.receipt.ticketId !== input.ticketId) throw new Error("unknown land operation verdict");
      if (operation.fence !== input.fence) throw new Error("stale land operation fence");
      if (operation.state !== "waiting") {
        if (JSON.stringify(operation.verdict) !== JSON.stringify(input.verdict)) throw new Error("immutable land verdict mismatch");
        record = operation;
        return;
      }
      const existing = this.db.query<{ operation_id: string; fence: number; verdict_json: string }, [string]>(
        "SELECT operation_id, fence, verdict_json FROM land_operation_verdict_events WHERE ticket_id = ?",
      ).get(input.ticketId);
      if (existing) {
        if (existing.operation_id !== input.operationId || existing.fence !== input.fence || existing.verdict_json !== JSON.stringify(input.verdict)) throw new Error("immutable land verdict event mismatch");
      } else {
        this.db.prepare("INSERT INTO land_operation_verdict_events (ticket_id, operation_id, fence, verdict_json, created_at) VALUES (?, ?, ?, ?, ?)")
          .run(input.ticketId, input.operationId, input.fence, JSON.stringify(input.verdict), this.now());
      }
      const succeeded = isSuccessfulLandVerdict(input.verdict);
      const state = succeeded ? "succeeded" : "failed";
      const intentId = succeeded ? `land-successor:${input.operationId}:${input.fence}` : null;
      this.db.prepare("UPDATE land_operations SET state = ?, verdict_json = ?, successor_intent_id = ? WHERE operation_id = ? AND state = 'waiting' AND fence = ?")
        .run(state, JSON.stringify(input.verdict), intentId, input.operationId, input.fence);
      if (succeeded) {
        this.db.prepare("INSERT INTO land_successor_dispatches (intent_id, operation_id, fence, state, next_action, receipt_json, dispatch_key) VALUES (?, ?, ?, 'pending', ?, ?, ?)")
          .run(intentId, input.operationId, input.fence, operation.receipt.nextAction, JSON.stringify(operation.receipt), intentId);
      }
      record = { ...operation, state, verdict: input.verdict, successorIntentId: intentId };
    });
    if (!record) throw new Error("land operation verdict did not persist");
    return record;
  }

  claimPendingLandSuccessorDispatches(leaseTtlMs = 60_000): LandSuccessorDispatch[] {
    const claimed: LandSuccessorDispatch[] = [];
    this.mutate(() => {
      const now = this.now();
      const rows = this.db.query<{
        intent_id: string; operation_id: string; fence: number; next_action: string; receipt_json: string;
        claim_generation: number; dispatch_key: string;
      }, [number]>(`
        SELECT intent_id, operation_id, fence, next_action, receipt_json, claim_generation, dispatch_key
        FROM land_successor_dispatches
        WHERE state = 'pending' OR (state = 'claimed' AND claim_expires_at <= ?)
        ORDER BY intent_id
      `).all(now);
      const claim = this.db.prepare(`
        UPDATE land_successor_dispatches
        SET state = 'claimed', claim_generation = claim_generation + 1, claim_expires_at = ?
        WHERE intent_id = ? AND claim_generation = ?
          AND (state = 'pending' OR (state = 'claimed' AND claim_expires_at <= ?))
      `);
      for (const row of rows) {
        if (claim.run(now + leaseTtlMs, row.intent_id, row.claim_generation, now).changes !== 1) continue;
        claimed.push({
          intentId: row.intent_id,
          dispatchKey: row.dispatch_key,
          claimGeneration: row.claim_generation + 1,
          operationId: row.operation_id,
          fence: row.fence,
          nextAction: row.next_action,
          receipt: JSON.parse(row.receipt_json) as LandOperationReceipt,
        });
      }
    });
    return claimed;
  }

  markLandSuccessorDispatched(intent: LandSuccessorDispatch): boolean {
    let accepted = false;
    this.mutate(() => {
      accepted = this.db.prepare(`
        UPDATE land_successor_dispatches
        SET state = 'accepted', accepted_at = ?, claim_expires_at = NULL
        WHERE intent_id = ? AND state = 'claimed' AND claim_generation = ? AND dispatch_key = ?
      `).run(this.now(), intent.intentId, intent.claimGeneration, intent.dispatchKey).changes === 1;
    });
    return accepted;
  }

  getLease(): FallbackLease {
    const meta = this.getMetaSnapshot();
    const expired =
      meta.leaseActive &&
      meta.leaseExpiresAt !== null &&
      Date.parse(meta.leaseExpiresAt) <= this.now();
    return {
      active: meta.leaseActive && !expired,
      expiresAt: meta.leaseExpiresAt,
      host: meta.leaseHost,
      reason: meta.leaseReason,
    };
  }

  setLease(lease: {
    active: boolean;
    expiresAt: string;
    host?: string | null;
    reason?: string | null;
  }): void {
    if (lease.active && !lease.expiresAt) {
      throw new Error("fallback lease must always expire");
    }
    this.mutate(() => {
      this.setMeta("lease_active", lease.active ? "1" : "0");
      this.setMeta("lease_expires_at", lease.expiresAt);
      this.setMeta("lease_host", lease.host ?? "");
      this.setMeta("lease_reason", lease.reason ?? "");
    });
  }

  clearLease(): void {
    this.mutate(() => {
      this.clearLeaseFields();
    });
  }

  setClusterState(desired: HostState, observed: HostState): void {
    this.mutate(() => {
      this.setMeta("desired", desired);
      this.setMeta("observed", observed);
    });
  }

  setDegraded(detail: string): void {
    this.mutate(() => {
      this.setMeta("desired", "degraded");
      this.setMeta("observed", "degraded");
      this.setMeta("dispatch_state", "paused");
      this.setMeta("dispatch_detail", detail);
      this.setMeta("reconciler_healthy", "0");
    });
  }

  emitConfigIncident(kind: string, detail: string): boolean {
    const meta = this.getMetaSnapshot();
    if (meta.configIncidentEmitted) {
      return false;
    }
    this.mutate(() => {
      const seenAt = new Date(this.now()).toISOString();
      this.writeIncident({
        key: kind,
        firstSeen: seenAt,
        lastSeen: seenAt,
        count: 1,
        affectedJobs: [],
        remediation: detail,
        cooldownUntil: null,
        autoResolveCondition: "operator supplies valid controller configuration",
        state: "open",
      });
      this.setMeta("config_incident_emitted", "1");
      this.setMeta("desired", "degraded");
      this.setMeta("observed", "degraded");
      this.setMeta("dispatch_state", "paused");
      this.setMeta("dispatch_detail", detail);
      this.setMeta("reconciler_healthy", "0");
    });
    return true;
  }

  recordLocalFallbackOverride(): void {
    this.mutate(() => {
      const seenAt = new Date(this.now()).toISOString();
      this.writeIncident({
        key: `local-fallback-override:${this.now()}`,
        firstSeen: seenAt,
        lastSeen: seenAt,
        count: 1,
        affectedJobs: [],
        remediation: "BUILD_REMOTE_LOCAL_FALLBACK=1 honored",
        cooldownUntil: null,
        autoResolveCondition: "override expires with process environment",
        state: "resolved",
      });
    });
  }

  getIncident(key: string): IncidentRecord | null {
    const row = this.db
      .query<IncidentRow, [string]>(
        `SELECT key, first_seen, last_seen, count, affected_jobs_json, remediation,
                cooldown_until, auto_resolve_condition, state
         FROM incidents WHERE key = ?`,
      )
      .get(key);
    return row ? parseIncidentRow(row) : null;
  }

  listIncidents(): IncidentRecord[] {
    return this.db
      .query<IncidentRow, []>(
        `SELECT key, first_seen, last_seen, count, affected_jobs_json, remediation,
                cooldown_until, auto_resolve_condition, state
         FROM incidents ORDER BY key`,
      )
      .all()
      .map(parseIncidentRow);
  }

  upsertIncident(
    input: Omit<IncidentRecord, "firstSeen" | "count">,
    openedEvent?: ControllerEvent,
    sourceRevision?: number,
  ): {
    incident: IncidentRecord;
    created: boolean;
  } {
    let result: { incident: IncidentRecord; created: boolean } | undefined;
    this.mutate(() => {
      const existing = this.getIncident(input.key);
      const incident: IncidentRecord = existing
        ? {
            ...existing,
            lastSeen: input.lastSeen,
            count: existing.count + 1,
            affectedJobs: [...new Set([...existing.affectedJobs, ...input.affectedJobs])].sort(),
            remediation: input.remediation,
            cooldownUntil: input.cooldownUntil,
            autoResolveCondition: input.autoResolveCondition,
            state: input.state,
          }
        : { ...input, firstSeen: input.lastSeen, count: 1 };
      const created = existing === null || existing.state === "resolved";
      this.writeIncident(incident);
      if (created && openedEvent) this.appendEventInTransaction(openedEvent);
      if (sourceRevision !== undefined) this.advanceIncidentCursor(sourceRevision);
      result = { incident, created };
    });
    if (!result) throw new Error("incident upsert did not execute");
    return result;
  }

  resolveIncident(
    key: string,
    lastSeen: string,
    resolvedEvent?: ControllerEvent,
    sourceRevision?: number,
  ): IncidentRecord | null {
    let incident: IncidentRecord | null = null;
    this.mutate(() => {
      const existing = this.getIncident(key);
      if (!existing || existing.state === "resolved") {
        if (sourceRevision !== undefined) this.advanceIncidentCursor(sourceRevision);
        return;
      }
      incident = { ...existing, lastSeen, state: "resolved" };
      this.writeIncident(incident);
      if (resolvedEvent) this.appendEventInTransaction(resolvedEvent);
      if (sourceRevision !== undefined) this.advanceIncidentCursor(sourceRevision);
    });
    return incident;
  }

  private writeIncident(incident: IncidentRecord): void {
    this.db.prepare(
      `INSERT INTO incidents (
         key, first_seen, last_seen, count, affected_jobs_json, remediation,
         cooldown_until, auto_resolve_condition, state
       ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
       ON CONFLICT(key) DO UPDATE SET
         first_seen = excluded.first_seen,
         last_seen = excluded.last_seen,
         count = excluded.count,
         affected_jobs_json = excluded.affected_jobs_json,
         remediation = excluded.remediation,
         cooldown_until = excluded.cooldown_until,
         auto_resolve_condition = excluded.auto_resolve_condition,
         state = excluded.state`,
    ).run(
      incident.key,
      incident.firstSeen,
      incident.lastSeen,
      incident.count,
      JSON.stringify(incident.affectedJobs),
      incident.remediation,
      incident.cooldownUntil,
      incident.autoResolveCondition,
      incident.state,
    );
  }

  private advanceIncidentCursor(revision: number): void {
    if (revision > this.getIncidentCursor()) {
      this.setMeta("incident_cursor", String(revision));
    }
  }

  hasConfigIncident(): boolean {
    return this.getMetaSnapshot().configIncidentEmitted;
  }

  upsertHost(host: Partial<HostRecord> & { hostname: string }): void {
    this.mutate(() => {
      const existing = this.getHost(host.hostname);
      const merged: HostRecord = {
        hostname: host.hostname,
        state: host.state ?? existing?.state ?? "available",
        role: host.role ?? existing?.role ?? "builder",
        slotsTotal: host.slotsTotal ?? existing?.slotsTotal ?? 4,
        slotsUsed: host.slotsUsed ?? existing?.slotsUsed ?? 0,
        runningJobs: host.runningJobs ?? existing?.runningJobs ?? 0,
        ciJobsRunning: host.ciJobsRunning ?? existing?.ciJobsRunning ?? 0,
        healthStorage: host.healthStorage ?? existing?.healthStorage ?? true,
        healthRunner: host.healthRunner ?? existing?.healthRunner ?? true,
        healthOffload: host.healthOffload ?? existing?.healthOffload ?? true,
        capabilityOk: host.capabilityOk ?? existing?.capabilityOk ?? true,
        capabilityReason:
          host.capabilityReason !== undefined
            ? host.capabilityReason
            : existing?.capabilityReason ?? null,
        capabilityCheckedAt:
          host.capabilityCheckedAt !== undefined
            ? host.capabilityCheckedAt
            : existing?.capabilityCheckedAt ?? null,
        primary: host.primary ?? existing?.primary ?? false,
        enrolling: host.enrolling ?? existing?.enrolling ?? false,
        dispatchPaused: host.dispatchPaused ?? existing?.dispatchPaused ?? false,
        quarantinedCommands:
          host.quarantinedCommands ?? existing?.quarantinedCommands ?? [],
      };
      this.db
        .prepare(
          `INSERT INTO hosts (
            hostname, state, role, slots_total, slots_used, running_jobs, ci_jobs_running,
            health_storage, health_runner, health_offload, capability_ok, primary_host, enrolling,
            dispatch_paused, capability_reason, capability_checked_at
          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
          ON CONFLICT(hostname) DO UPDATE SET
            state = excluded.state,
            role = excluded.role,
            slots_total = excluded.slots_total,
            slots_used = excluded.slots_used,
            running_jobs = excluded.running_jobs,
            ci_jobs_running = excluded.ci_jobs_running,
            health_storage = excluded.health_storage,
            health_runner = excluded.health_runner,
            health_offload = excluded.health_offload,
            capability_ok = excluded.capability_ok,
            primary_host = excluded.primary_host,
            enrolling = excluded.enrolling,
            dispatch_paused = excluded.dispatch_paused,
            capability_reason = excluded.capability_reason,
            capability_checked_at = excluded.capability_checked_at`,
        )
        .run(
          merged.hostname,
          merged.state,
          merged.role,
          merged.slotsTotal,
          merged.slotsUsed,
          merged.runningJobs,
          merged.ciJobsRunning,
          merged.healthStorage ? 1 : 0,
          merged.healthRunner ? 1 : 0,
          merged.healthOffload ? 1 : 0,
          merged.capabilityOk ? 1 : 0,
          merged.primary ? 1 : 0,
          merged.enrolling ? 1 : 0,
          merged.dispatchPaused ? 1 : 0,
          merged.capabilityReason,
          merged.capabilityCheckedAt,
        );
    });
  }

  getHost(hostname: string): HostRecord | null {
    const row = this.db
      .query<
        {
          hostname: string;
          state: string;
          role: string;
          slots_total: number;
          slots_used: number;
          running_jobs: number;
          ci_jobs_running: number;
          health_storage: number;
          health_runner: number;
          health_offload: number;
          capability_ok: number;
          primary_host: number;
          enrolling: number;
          dispatch_paused: number;
          capability_reason: string | null;
          capability_checked_at: string | null;
        },
        [string]
      >(
        `SELECT hostname, state, role, slots_total, slots_used, running_jobs, ci_jobs_running,
                health_storage, health_runner, health_offload, capability_ok, primary_host, enrolling,
                dispatch_paused, capability_reason, capability_checked_at
         FROM hosts WHERE hostname = ?`,
      )
      .get(hostname);
    if (!row) return null;
    const quarantines = this.db
      .query<{ command: string }, [string]>(
        "SELECT command FROM capability_breakers WHERE hostname = ? AND state = 'open' ORDER BY command",
      )
      .all(hostname)
      .map((r) => r.command);
    return {
      hostname: row.hostname,
      state: row.state as HostState,
      role: row.role as "builder" | "workstation",
      slotsTotal: row.slots_total,
      slotsUsed: row.slots_used,
      runningJobs: row.running_jobs,
      ciJobsRunning: row.ci_jobs_running,
      healthStorage: row.health_storage === 1,
      healthRunner: row.health_runner === 1,
      healthOffload: row.health_offload === 1,
      capabilityOk: row.capability_ok === 1,
      capabilityReason: row.capability_reason,
      capabilityCheckedAt: row.capability_checked_at,
      primary: row.primary_host === 1,
      enrolling: row.enrolling === 1,
      dispatchPaused: row.dispatch_paused === 1,
      quarantinedCommands: quarantines,
    };
  }

  listHosts(): HostRecord[] {
    const rows = this.db
      .query<{ hostname: string }, []>("SELECT hostname FROM hosts ORDER BY hostname")
      .all();
    return rows
      .map((row) => this.getHost(row.hostname))
      .filter((host): host is HostRecord => host !== null);
  }

  recordLandConductResult(root: string, at: string, ok: boolean, detail: string): void {
    this.mutate(() => {
      this.db
        .prepare(
          `INSERT INTO land_conduct_health (root, last_pass_at, last_ok, last_detail, consecutive_failures)
           VALUES (?, ?, ?, ?, ?)
           ON CONFLICT(root) DO UPDATE SET
             last_pass_at = excluded.last_pass_at,
             last_ok = excluded.last_ok,
             last_detail = excluded.last_detail,
             consecutive_failures = CASE
               WHEN excluded.last_ok = 1 THEN 0
               ELSE land_conduct_health.consecutive_failures + 1
             END`,
        )
        .run(root, at, ok ? 1 : 0, detail, ok ? 0 : 1);
    });
  }

  listLandConductHealth(): LandConductHealthRecord[] {
    return this.db
      .query<{
        root: string;
        last_pass_at: string | null;
        last_ok: number;
        last_detail: string;
        consecutive_failures: number;
      }, []>(
        `SELECT root, last_pass_at, last_ok, last_detail, consecutive_failures
         FROM land_conduct_health
         ORDER BY root`,
      )
      .all()
      .map((row) => ({
        root: row.root,
        lastPassAt: row.last_pass_at,
        lastOk: row.last_ok === 1,
        lastDetail: row.last_detail,
        consecutiveFailures: row.consecutive_failures,
      }));
  }

  getDeployWatcherState(): DeployWatcherStateRecord | null {
    const row = this.db
      .query<{
        target_sha: string;
        attempts: number;
        last_status: string;
        last_detail: string;
        last_at: string;
        last_ok: number;
        failure_class: string;
        next_retry_at: string | null;
      }, []>(
        `SELECT target_sha, attempts, last_status, last_detail, last_at, last_ok, failure_class, next_retry_at
         FROM deploy_watcher_state WHERE id = 1`,
      )
      .get();
    if (!row) return null;
    const lastOk = row.last_ok === 1;
    const storedClass: DeployFailureClass = row.failure_class === "transient" || row.failure_class === "permanent"
      ? row.failure_class
      : "none";
    // Rows written before classification had no taxonomy. A historical failure must
    // default permanent; treating its migrated DEFAULT 'none' as retryable would flood
    // the queue after a controller restart.
    const failureClass: DeployFailureClass = !lastOk && storedClass === "none"
      ? "permanent"
      : storedClass;
    return {
      targetSha: row.target_sha,
      attempts: row.attempts,
      lastStatus: row.last_status,
      lastDetail: row.last_detail,
      lastAt: row.last_at,
      lastOk,
      failureClass,
      nextRetryAt: failureClass === "transient" ? row.next_retry_at : null,
    };
  }

  recordDeployWatcherResult(record: DeployWatcherStateRecord): void {
    this.mutate(() => {
      this.db
        .prepare(
          `INSERT INTO deploy_watcher_state
             (id, target_sha, attempts, last_status, last_detail, last_at, last_ok, failure_class, next_retry_at)
           VALUES (1, ?, ?, ?, ?, ?, ?, ?, ?)
           ON CONFLICT(id) DO UPDATE SET
             target_sha = excluded.target_sha,
             attempts = excluded.attempts,
             last_status = excluded.last_status,
             last_detail = excluded.last_detail,
             last_at = excluded.last_at,
             last_ok = excluded.last_ok,
             failure_class = excluded.failure_class,
             next_retry_at = excluded.next_retry_at`,
        )
        .run(
          record.targetSha,
          record.attempts,
          record.lastStatus,
          record.lastDetail,
          record.lastAt,
          record.lastOk ? 1 : 0,
          record.failureClass,
          record.nextRetryAt,
        );
    });
  }

  setHostDispatchPaused(hostname: string, paused: boolean): void {
    this.mutate(() => {
      const result = this.db
        .prepare("UPDATE hosts SET dispatch_paused = ? WHERE hostname = ?")
        .run(paused ? 1 : 0, hostname);
      if (result.changes !== 1) throw new Error(`unknown host: ${hostname}`);
    });
  }

  upsertCapabilityManifest(manifest: CapabilityManifestRecord): void {
    this.mutate(() => {
      this.db
        .prepare(
          `INSERT INTO capability_manifests (repo, command, manifest_json)
           VALUES (?, ?, ?)
           ON CONFLICT(repo, command) DO UPDATE SET manifest_json = excluded.manifest_json`,
        )
        .run(manifest.repo, manifest.command, JSON.stringify(manifest));
    });
  }

  getManifest(repo: string, command: string): CapabilityManifestRecord | null {
    const row = this.db
      .query<{ manifest_json: string }, [string, string]>(
        "SELECT manifest_json FROM capability_manifests WHERE repo = ? AND command = ?",
      )
      .get(repo, command);
    return row ? JSON.parse(row.manifest_json) as CapabilityManifestRecord : null;
  }

  getCapabilityBreaker(hostname: string, command: string): CapabilityBreakerRecord | null {
    const row = this.db
      .query<
        { state: string; failure_count: number; missing_event_emitted: number },
        [string, string]
      >(
        `SELECT state, failure_count, missing_event_emitted
         FROM capability_breakers WHERE hostname = ? AND command = ?`,
      )
      .get(hostname, command);
    if (!row) return null;
    return {
      hostname,
      command,
      state: row.state as CapabilityBreakerState,
      failureCount: row.failure_count,
      missingEventEmitted: row.missing_event_emitted === 1,
    };
  }

  listCapabilityBreakers(hostname: string): CapabilityBreakerRecord[] {
    return this.db
      .query<{ command: string }, [string]>(
        "SELECT command FROM capability_breakers WHERE hostname = ? ORDER BY command",
      )
      .all(hostname)
      .map(({ command }) => this.getCapabilityBreaker(hostname, command))
      .filter((record): record is CapabilityBreakerRecord => record !== null);
  }

  setCapabilityBreaker(record: CapabilityBreakerRecord): void {
    this.mutate(() => {
      this.db
        .prepare(
          `INSERT INTO capability_breakers
           (hostname, command, state, failure_count, missing_event_emitted)
           VALUES (?, ?, ?, ?, ?)
           ON CONFLICT(hostname, command) DO UPDATE SET
             state = excluded.state,
             failure_count = excluded.failure_count,
             missing_event_emitted = excluded.missing_event_emitted`,
        )
        .run(
          record.hostname,
          record.command,
          record.state,
          record.failureCount,
          record.missingEventEmitted ? 1 : 0,
        );
    });
  }

  setCapabilityBreakerAndAppendEvent(
    record: CapabilityBreakerRecord,
    eventInput: ControllerEvent,
  ): number {
    let revision = 0;
    this.mutate(() => {
      if (this.eventWriteFails) throw new Error("event write failed");
      const event = EventSchema.parse(eventInput);
      this.db
        .prepare(
          `INSERT INTO capability_breakers
           (hostname, command, state, failure_count, missing_event_emitted)
           VALUES (?, ?, ?, ?, ?)
           ON CONFLICT(hostname, command) DO UPDATE SET
             state = excluded.state,
             failure_count = excluded.failure_count,
             missing_event_emitted = excluded.missing_event_emitted`,
        )
        .run(
          record.hostname,
          record.command,
          record.state,
          record.failureCount,
          record.missingEventEmitted ? 1 : 0,
        );
      revision = Number(this.getMeta("revision")) + 1;
      this.db
        .prepare("INSERT INTO events (revision, payload_json) VALUES (?, ?)")
        .run(revision, JSON.stringify(event));
      this.setMeta("revision", String(revision));
      this.setMeta("reconciler_last_at", new Date(this.now()).toISOString());
    });
    return revision;
  }

  setHostState(hostname: string, state: HostState): void {
    const host = this.getHost(hostname);
    if (!host) throw new Error(`unknown host: ${hostname}`);
    this.upsertHost({ ...host, state });
  }

  upsertJob(job: JobRecord): void {
    this.mutate(() => {
      this.db
        .prepare(
          `INSERT INTO jobs (
             id, job_key, mirror, repo, host, snapshot, stage, attempt, rc, infra_failure,
             started_at, finished_at, last_report_at, timeout_sec
           ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
           ON CONFLICT(id) DO UPDATE SET
             job_key = excluded.job_key,
             mirror = excluded.mirror,
             repo = excluded.repo,
             host = excluded.host,
             snapshot = excluded.snapshot,
             stage = excluded.stage,
             attempt = excluded.attempt,
             rc = excluded.rc,
             infra_failure = excluded.infra_failure,
             started_at = excluded.started_at,
             finished_at = excluded.finished_at,
             last_report_at = excluded.last_report_at,
             timeout_sec = excluded.timeout_sec`,
        )
        .run(
          job.id,
          job.key ?? null,
          job.mirror ?? null,
          job.repo,
          job.host,
          job.snapshot,
          job.stage,
          job.attempt,
          job.rc,
          job.infraFailure ? 1 : 0,
          job.startedAt ?? null,
          job.finishedAt ?? null,
          job.lastReportAt ?? null,
          job.timeoutSec ?? null,
        );
      if (SCHEDULER_TERMINAL_STATES.has(job.stage)) {
        this.releaseHostSlotForTerminalJob(job.id, job.stage);
      }
    });
  }

  getJob(id: string): JobRecord | null {
    const row = this.db
      .query<
        {
          id: string;
          job_key: string | null;
          mirror: string | null;
          repo: string;
          host: string;
          snapshot: string;
          stage: string;
          attempt: number;
          rc: number | null;
          infra_failure: number;
          publication_state: string;
          publication_reason: string | null;
          started_at: string | null;
          finished_at: string | null;
          last_report_at: string | null;
          timeout_sec: number | null;
        },
        [string]
      >(
        `SELECT id, job_key, mirror, repo, host, snapshot, stage, attempt, rc, infra_failure,
                publication_state, publication_reason, started_at, finished_at, last_report_at,
                timeout_sec
         FROM jobs WHERE id = ?`,
      )
      .get(id);
    if (!row) return null;
    return parseJobRow(row);
  }

  listJobs(): JobRecord[] {
    return this.db
      .query<
        {
          id: string;
          job_key: string | null;
          mirror: string | null;
          repo: string;
          host: string;
          snapshot: string;
          stage: string;
          attempt: number;
          rc: number | null;
          infra_failure: number;
          publication_state: string;
          publication_reason: string | null;
          started_at: string | null;
          finished_at: string | null;
          last_report_at: string | null;
          timeout_sec: number | null;
        },
        []
      >(
        `SELECT id, job_key, mirror, repo, host, snapshot, stage, attempt, rc, infra_failure,
                publication_state, publication_reason, started_at, finished_at, last_report_at,
                timeout_sec
         FROM jobs ORDER BY id`,
      )
      .all()
      .map(parseJobRow);
  }

  createWorkspace(record: WorkspaceRecord, maxRemoteJobs: number): boolean {
    if (!Number.isInteger(maxRemoteJobs) || maxRemoteJobs < 1) {
      throw new Error("maxRemoteJobs must be a positive integer");
    }
    let reserved = false;
    this.mutate(() => {
      const result = this.db
        .prepare(
          `INSERT INTO remote_job_reservations (job_id, created_at)
           SELECT ?, ?
           WHERE (SELECT COUNT(*) FROM remote_job_reservations) < ?
             AND NOT EXISTS (
               SELECT 1 FROM remote_job_reservations WHERE job_id = ?
             )`,
        )
        .run(record.jobId, this.now(), maxRemoteJobs, record.jobId);
      reserved = result.changes === 1;
      if (!reserved) return;

      this.db
        .prepare(
          `INSERT INTO jobs (
             id, repo, host, snapshot, stage, attempt, rc, infra_failure,
             publication_state, publication_reason
           ) VALUES (?, ?, ?, ?, ?, 1, NULL, 0, 'none', NULL)`,
        )
        .run(record.jobId, record.repo, record.host, record.snapshot, record.stage);
      this.db
        .prepare(
          `INSERT INTO workspaces (
             job_id, checkout_generation, checkout_path, publication_path,
             workspace_path, cache_path, snapshot_path, overlay_path, output_path,
             staging_path, backup_path, manifest_json, publication_state,
             publication_reason, transport_reattach_count, completed_at
           ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
        )
        .run(
          record.jobId,
          record.checkoutGeneration,
          record.checkoutPath,
          record.publicationPath,
          record.workspacePath,
          record.cachePath,
          record.snapshotPath,
          record.overlayPath,
          record.outputPath,
          record.stagingPath,
          record.backupPath,
          record.manifest === null ? null : JSON.stringify(record.manifest),
          record.publicationState,
          record.publicationReason,
          record.transportReattachCount,
          record.completedAt,
        );
    });
    return reserved;
  }

  getWorkspace(jobId: string): WorkspaceRecord | null {
    const row = this.db
      .query<
        {
          job_id: string;
          repo: string;
          host: string;
          snapshot: string;
          stage: string;
          checkout_generation: string;
          checkout_path: string;
          publication_path: string;
          workspace_path: string;
          cache_path: string;
          snapshot_path: string;
          overlay_path: string;
          output_path: string;
          staging_path: string;
          backup_path: string;
          manifest_json: string | null;
          publication_state: WorkspacePublicationState;
          publication_reason: string | null;
          transport_reattach_count: number;
          completed_at: number | null;
        },
        [string]
      >(
        `SELECT w.job_id, j.repo, j.host, j.snapshot, j.stage,
                w.checkout_generation, w.checkout_path, w.publication_path,
                w.workspace_path, w.cache_path, w.snapshot_path, w.overlay_path,
                w.output_path, w.staging_path, w.backup_path, w.manifest_json,
                w.publication_state, w.publication_reason,
                w.transport_reattach_count, w.completed_at
         FROM workspaces w
         JOIN jobs j ON j.id = w.job_id
         WHERE w.job_id = ?`,
      )
      .get(jobId);
    if (!row) return null;
    return {
      jobId: row.job_id,
      repo: row.repo,
      host: row.host,
      snapshot: row.snapshot,
      stage: row.stage,
      checkoutGeneration: row.checkout_generation,
      checkoutPath: row.checkout_path,
      publicationPath: row.publication_path,
      workspacePath: row.workspace_path,
      cachePath: row.cache_path,
      snapshotPath: row.snapshot_path,
      overlayPath: row.overlay_path,
      outputPath: row.output_path,
      stagingPath: row.staging_path,
      backupPath: row.backup_path,
      manifest: row.manifest_json === null ? null : JSON.parse(row.manifest_json),
      publicationState: row.publication_state,
      publicationReason: row.publication_reason,
      transportReattachCount: row.transport_reattach_count,
      completedAt: row.completed_at,
    };
  }

  listPromotingWorkspaces(): WorkspaceRecord[] {
    return this.db
      .query<{ job_id: string }, []>(
        "SELECT job_id FROM workspaces WHERE publication_state = 'promoting' ORDER BY job_id",
      )
      .all()
      .map((row) => this.getWorkspace(row.job_id))
      .filter((record): record is WorkspaceRecord => record !== null);
  }

  listExpiredWorkspaces(cutoff: number): WorkspaceRecord[] {
    return this.db
      .query<{ job_id: string }, [number]>(
        `SELECT job_id FROM workspaces
         WHERE completed_at IS NOT NULL AND completed_at <= ?
         ORDER BY job_id`,
      )
      .all(cutoff)
      .map((row) => this.getWorkspace(row.job_id))
      .filter((record): record is WorkspaceRecord => record !== null);
  }

  setWorkspaceManifest(jobId: string, manifest: unknown): void {
    this.mutate(() => {
      this.db
        .prepare(
          `UPDATE workspaces
           SET manifest_json = ?, publication_state = 'staged',
               publication_reason = NULL, completed_at = NULL
           WHERE job_id = ?`,
        )
        .run(JSON.stringify(manifest), jobId);
      this.db
        .prepare(
          `UPDATE jobs
           SET stage = 'staged', publication_state = 'staged', publication_reason = NULL
           WHERE id = ?`,
        )
        .run(jobId);
    });
  }

  setWorkspacePublication(
    jobId: string,
    state: WorkspacePublicationState,
    reason: string | null = null,
    completedAt: number | null = null,
  ): void {
    const publicState = state === "promoting" ? "staged" : state;
    const stage = state === "promoted"
      ? "completed"
      : state === "blocked"
        ? "blocked"
        : state === "discarded"
          ? "discarded"
          : state === "promoting"
            ? "staged"
            : state;
    this.mutate(() => {
      this.db
        .prepare(
          `UPDATE workspaces
           SET publication_state = ?, publication_reason = ?, completed_at = ?
           WHERE job_id = ?`,
        )
        .run(state, reason, completedAt, jobId);
      this.db
        .prepare(
          `UPDATE jobs
           SET stage = ?, publication_state = ?, publication_reason = ?
           WHERE id = ?`,
        )
        .run(stage, publicState, reason, jobId);
      if (completedAt !== null) {
        this.db.prepare("DELETE FROM remote_job_reservations WHERE job_id = ?").run(jobId);
        if (SCHEDULER_TERMINAL_STATES.has(stage)) {
          this.releaseHostSlotForTerminalJob(jobId, stage);
        }
      }
    });
  }

  recordPublicationBlocked(jobId: string, detail: string): void {
    this.mutate(() => {
      const job = this.getJob(jobId);
      if (!job) throw new Error(`unknown job: ${jobId}`);
      const nextRevision = this.getRevision() + 1;
      const event = EventSchema.parse({
        ts: new Date(this.now()).toISOString(),
        job: job.id,
        repo: job.repo,
        host: job.host,
        snapshot: job.snapshot,
        attempt: job.attempt,
        stage: "blocked",
        reason: "artifact-publication-blocked",
        rc: job.rc,
        durationSeconds: 0,
      });
      if (this.eventWriteFails) throw new Error("event write failed");
      this.db
        .prepare(
          `UPDATE workspaces
           SET publication_state = 'blocked', publication_reason = ?, completed_at = ?
           WHERE job_id = ?`,
        )
        .run(detail, this.now(), jobId);
      this.db
        .prepare(
          `UPDATE jobs
           SET stage = 'blocked', publication_state = 'blocked', publication_reason = ?
           WHERE id = ?`,
        )
        .run(detail, jobId);
      this.db.prepare("DELETE FROM remote_job_reservations WHERE job_id = ?").run(jobId);
      this.releaseHostSlotForTerminalJob(jobId, "blocked");
      this.db
        .prepare("INSERT INTO events (revision, payload_json) VALUES (?, ?)")
        .run(nextRevision, JSON.stringify(event));
      this.setMeta("revision", String(nextRevision));
      this.setMeta("reconciler_last_at", new Date(this.now()).toISOString());
    });
  }

  claimTransportReattach(jobId: string): boolean {
    const result = this.db
      .prepare(
        `UPDATE workspaces
         SET transport_reattach_count = transport_reattach_count + 1
         WHERE job_id = ? AND transport_reattach_count = 0 AND completed_at IS NULL`,
      )
      .run(jobId);
    return result.changes === 1;
  }

  markTransportFailed(jobId: string): void {
    this.mutate(() => {
      this.db
        .prepare("UPDATE jobs SET stage = 'failed', infra_failure = 1 WHERE id = ?")
        .run(jobId);
      this.db
        .prepare("UPDATE workspaces SET completed_at = ? WHERE job_id = ?")
        .run(this.now(), jobId);
      this.db.prepare("DELETE FROM remote_job_reservations WHERE job_id = ?").run(jobId);
      this.releaseHostSlotForTerminalJob(jobId, "failed");
    });
  }

  countRemoteJobReservations(): number {
    return this.db
      .query<{ count: number }, []>(
        "SELECT COUNT(*) AS count FROM remote_job_reservations",
      )
      .get()?.count ?? 0;
  }

  abandonWorkspace(jobId: string): void {
    this.mutate(() => {
      this.releaseHostSlotForTerminalJob(jobId, "cancelled");
      this.db.prepare("DELETE FROM workspaces WHERE job_id = ?").run(jobId);
      this.db.prepare("DELETE FROM remote_job_reservations WHERE job_id = ?").run(jobId);
      this.db.prepare("DELETE FROM jobs WHERE id = ?").run(jobId);
    });
  }

  deleteWorkspace(jobId: string): void {
    this.mutate(() => {
      this.releaseHostSlotForTerminalJob(jobId, "cancelled");
      this.db.prepare("DELETE FROM workspaces WHERE job_id = ?").run(jobId);
      this.db.prepare("DELETE FROM remote_job_reservations WHERE job_id = ?").run(jobId);
    });
  }

  listQueueTickets(): QueueTicketRecord[] {
    return this.db
      .query<
        {
          position: number;
          ticket_key: string;
          repo: string;
          owner_pid: number;
          owner_starttime: number;
          owner_label: string | null;
          command: string;
          enqueued_at: number;
          enqueue_age_seconds: number;
          state: string;
          dispatch_target: string | null;
          placement_kind: string | null;
          placement_job_id: string | null;
          placed_at: number | null;
        },
        []
      >(
        `SELECT position, ticket_key, repo, owner_pid, owner_starttime, owner_label,
                command, enqueued_at, enqueue_age_seconds, state, dispatch_target,
                placement_kind, placement_job_id, placed_at
         FROM queue_tickets ORDER BY position`,
      )
      .all()
      .map((row) => {
        const enqueuedAt = row.enqueued_at || this.now() - row.enqueue_age_seconds * 1000;
        return {
          position: row.position,
          key: row.ticket_key,
          repo: row.repo,
          owner: {
            pid: row.owner_pid,
            starttime: row.owner_starttime,
            ...(row.owner_label ? { label: row.owner_label } : {}),
          },
          command: row.command,
          enqueuedAt,
          enqueueAgeSeconds: Math.max(0, (this.now() - enqueuedAt) / 1000),
          state: row.state,
          ...(row.dispatch_target ? { dispatchTarget: row.dispatch_target } : {}),
          ...(row.placement_kind && row.dispatch_target && row.placement_job_id && row.placed_at !== null
            ? {
                placement: {
                  kind: row.placement_kind as "builder" | "spill",
                  host: row.dispatch_target,
                  jobId: row.placement_job_id,
                  placedAt: row.placed_at,
                },
              }
            : {}),
        };
      });
  }

  upsertQueueTicket(ticket: QueueTicketRecord): void {
    this.mutate(() => {
      this.db
        .prepare(
          `INSERT INTO queue_tickets (
            position, ticket_key, repo, owner_pid, owner_starttime, owner_label,
            command, enqueued_at, enqueue_age_seconds, state, dispatch_target,
            placement_kind, placement_job_id, placed_at
          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
          ON CONFLICT(position) DO UPDATE SET
            ticket_key = excluded.ticket_key,
            repo = excluded.repo,
            owner_pid = excluded.owner_pid,
            owner_starttime = excluded.owner_starttime,
            owner_label = excluded.owner_label,
            command = excluded.command,
            enqueued_at = excluded.enqueued_at,
            enqueue_age_seconds = excluded.enqueue_age_seconds,
            state = excluded.state,
            dispatch_target = excluded.dispatch_target,
            placement_kind = excluded.placement_kind,
            placement_job_id = excluded.placement_job_id,
            placed_at = excluded.placed_at`,
        )
        .run(
          ticket.position,
          ticket.key,
          ticket.repo,
          ticket.owner.pid,
          ticket.owner.starttime,
          ticket.owner.label ?? null,
          ticket.command ?? "",
          ticket.enqueuedAt ?? this.now() - ticket.enqueueAgeSeconds * 1000,
          ticket.enqueueAgeSeconds,
          ticket.state,
          ticket.dispatchTarget ?? null,
          ticket.placement?.kind ?? null,
          ticket.placement?.jobId ?? null,
          ticket.placement?.placedAt ?? null,
        );
    });
  }

  appendQueueTicket(
    ticket: Omit<QueueTicketRecord, "position">,
  ): QueueTicketRecord {
    let position = 0;
    this.mutate(() => {
      position = this.db
        .query<{ position: number }, []>(
          "SELECT COALESCE(MAX(position), 0) + 1 AS position FROM queue_tickets",
        )
        .get()?.position ?? 1;
      this.db
        .prepare(
          `INSERT INTO queue_tickets (
            position, ticket_key, repo, owner_pid, owner_starttime, owner_label,
            command, enqueued_at, enqueue_age_seconds, state, dispatch_target,
            placement_kind, placement_job_id, placed_at
          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
        )
        .run(
          position,
          ticket.key,
          ticket.repo,
          ticket.owner.pid,
          ticket.owner.starttime,
          ticket.owner.label ?? null,
          ticket.command ?? "",
          ticket.enqueuedAt ?? this.now() - ticket.enqueueAgeSeconds * 1000,
          ticket.enqueueAgeSeconds,
          ticket.state,
          ticket.dispatchTarget ?? null,
          ticket.placement?.kind ?? null,
          ticket.placement?.jobId ?? null,
          ticket.placement?.placedAt ?? null,
        );
    });
    return { ...ticket, position };
  }

  deleteQueueTicket(position: number): void {
    this.mutate(() => {
      const spill = this.db
        .query<{ count: number }, [number]>(
          "SELECT COUNT(*) AS count FROM queue_tickets WHERE position = ? AND placement_kind = 'spill'",
        )
        .get(position)?.count ?? 0;
      this.db
        .prepare(
          `UPDATE host_slot_reservations
           SET released_at = ?
           WHERE ticket_position = ? AND released_at IS NULL`,
        )
        .run(this.now(), position);
      this.db.prepare("DELETE FROM queue_tickets WHERE position = ?").run(position);
      if (spill > 0) this.clearLeaseFields();
    });
  }

  listHostSlotReservations(activeOnly = false): HostSlotReservationRecord[] {
    const sql = `SELECT job_id, hostname, ticket_position, reserved_at, released_at
                 FROM host_slot_reservations
                 ${activeOnly ? "WHERE released_at IS NULL" : ""}
                 ORDER BY reserved_at, job_id`;
    return this.db
      .query<
        {
          job_id: string;
          hostname: string;
          ticket_position: number;
          reserved_at: number;
          released_at: number | null;
        },
        []
      >(sql)
      .all()
      .map((row) => ({
        jobId: row.job_id,
        hostname: row.hostname,
        ticketPosition: row.ticket_position,
        reservedAt: row.reserved_at,
        releasedAt: row.released_at,
      }));
  }

  countHostSlotReservations(hostname?: string): number {
    if (hostname) {
      return this.db
        .query<{ count: number }, [string]>(
          `SELECT COUNT(*) AS count FROM host_slot_reservations
           WHERE hostname = ? AND released_at IS NULL`,
        )
        .get(hostname)?.count ?? 0;
    }
    return this.db
      .query<{ count: number }, []>(
        "SELECT COUNT(*) AS count FROM host_slot_reservations WHERE released_at IS NULL",
      )
      .get()?.count ?? 0;
  }

  tryReserveHostSlot(position: number, hostname: string, placedAt: number): boolean {
    let reserved = false;
    this.mutate(() => {
      const ticket = this.db
        .query<{ ticket_key: string; command: string }, [number]>(
          "SELECT ticket_key, command FROM queue_tickets WHERE position = ? AND state = 'queued'",
        )
        .get(position);
      if (!ticket) return;
      const result = this.db
        .prepare(
          `INSERT INTO host_slot_reservations
             (job_id, hostname, ticket_position, reserved_at, released_at)
           SELECT ?, h.hostname, ?, ?, NULL
           FROM hosts h
           WHERE h.hostname = ?
             AND h.role = 'builder'
             AND h.state = 'available'
             AND h.capability_ok = 1
             AND h.dispatch_paused = 0
             AND h.slots_used + (
               SELECT COUNT(*) FROM host_slot_reservations r
               WHERE r.hostname = h.hostname AND r.released_at IS NULL
             ) < h.slots_total
             AND NOT EXISTS (
               SELECT 1 FROM capability_breakers b
               WHERE b.hostname = h.hostname AND b.command = ? AND b.state = 'open'
             )`,
        )
        .run(ticket.ticket_key, position, placedAt, hostname, ticket.command);
      if (result.changes !== 1) return;
      const updated = this.db
        .prepare(
          `UPDATE queue_tickets
           SET state = 'placed', dispatch_target = ?, placement_kind = 'builder',
               placement_job_id = ?, placed_at = ?
           WHERE position = ? AND state = 'queued'`,
        )
        .run(hostname, ticket.ticket_key, placedAt, position);
      if (updated.changes !== 1) {
        throw new Error(`queue ticket ${position} changed during placement`);
      }
      reserved = true;
    });
    return reserved;
  }

  tryPlaceSpill(position: number, placedAt: number, expiresAt: string): boolean {
    let placed = false;
    this.mutate(() => {
      const policy = this.db
        .query<
          { builders: number; queued: number; builders_with_capacity: number },
          []
        >(
          `SELECT
             (SELECT COUNT(*) FROM hosts WHERE role = 'builder') AS builders,
             (SELECT COUNT(*) FROM queue_tickets WHERE state = 'queued') AS queued,
             (SELECT COUNT(*) FROM hosts h
              WHERE h.role = 'builder'
                AND h.slots_used + (
                  SELECT COUNT(*) FROM host_slot_reservations r
                  WHERE r.hostname = h.hostname AND r.released_at IS NULL
                ) < h.slots_total) AS builders_with_capacity`,
        )
        .get();
      if (
        !policy ||
        policy.builders === 0 ||
        policy.builders_with_capacity > 0 ||
        policy.queued <= policy.builders
      ) {
        return;
      }
      const ticket = this.db
        .query<{ ticket_key: string }, [number]>(
          "SELECT ticket_key FROM queue_tickets WHERE position = ? AND state = 'queued'",
        )
        .get(position);
      if (!ticket) return;
      const activeSpill = this.db
        .query<{ count: number }, []>(
          `SELECT COUNT(*) AS count FROM queue_tickets
           WHERE placement_kind = 'spill' AND state IN ('placed', 'recall-requested')`,
        )
        .get()?.count ?? 0;
      if (activeSpill > 0) return;
      const updated = this.db
        .prepare(
          `UPDATE queue_tickets
           SET state = 'placed', dispatch_target = 'laptop', placement_kind = 'spill',
               placement_job_id = ?, placed_at = ?
           WHERE position = ? AND state = 'queued'`,
        )
        .run(ticket.ticket_key, placedAt, position);
      if (updated.changes !== 1) return;
      this.setMeta("lease_active", "1");
      this.setMeta("lease_expires_at", expiresAt);
      this.setMeta("lease_host", "laptop");
      this.setMeta("lease_reason", "cluster-overloaded");
      placed = true;
    });
    return placed;
  }

  completeSchedulerPlacement(jobId: string, terminalState: string): boolean {
    let released = false;
    this.mutate(() => {
      const changes = this.releaseHostSlotForTerminalJob(jobId, terminalState);
      released = changes.reservation || changes.ticket;
    });
    return released;
  }

  reconcileTerminalHostReservations(): number {
    let released = 0;
    this.mutate(() => {
      for (const row of this.db
        .query<{ job_id: string; stage: string }, []>(
          `SELECT r.job_id, j.stage
           FROM host_slot_reservations r JOIN jobs j ON j.id = r.job_id
           WHERE r.released_at IS NULL`,
        )
        .all()) {
        if (!SCHEDULER_TERMINAL_STATES.has(row.stage)) continue;
        if (this.releaseHostSlotForTerminalJob(row.job_id, row.stage).reservation) {
          released += 1;
        }
      }
    });
    return released;
  }

  recallSpill(host?: string): number {
    let recalled = 0;
    this.mutate(() => {
      const leaseHost = this.getMeta("lease_host") || null;
      if (host && leaseHost && leaseHost !== host) {
        throw new Error(`spill lease is on ${leaseHost}, not ${host}`);
      }
      const result = this.db
        .prepare(
          `UPDATE queue_tickets SET state = 'recall-requested'
           WHERE placement_kind = 'spill' AND state = 'placed'`,
        )
        .run();
      recalled = result.changes;
      this.clearLeaseFields();
    });
    return recalled;
  }

  private releaseHostSlotForTerminalJob(
    jobId: string,
    terminalState: string,
  ): { reservation: boolean; ticket: boolean } {
    const spill = this.db
      .query<{ count: number }, [string]>(
        "SELECT COUNT(*) AS count FROM queue_tickets WHERE placement_job_id = ? AND placement_kind = 'spill'",
      )
      .get(jobId)?.count ?? 0;
    const reservation = this.db
      .prepare(
        `UPDATE host_slot_reservations SET released_at = ?
         WHERE job_id = ? AND released_at IS NULL`,
      )
      .run(this.now(), jobId);
    const ticket = this.db
      .prepare(
        `UPDATE queue_tickets SET state = ?
         WHERE placement_job_id = ? AND state IN ('placed', 'recall-requested')`,
      )
      .run(terminalState, jobId);
    if (spill > 0 && ticket.changes === 1) this.clearLeaseFields();
    return { reservation: reservation.changes === 1, ticket: ticket.changes === 1 };
  }

  private clearLeaseFields(): void {
    this.setMeta("lease_active", "0");
    this.setMeta("lease_expires_at", "");
    this.setMeta("lease_host", "");
    this.setMeta("lease_reason", "");
  }

  readEvents(afterRevision = 0): StoredEvent[] {
    return this.db
      .query<{ revision: number; payload_json: string }, [number]>(
        "SELECT revision, payload_json FROM events WHERE revision > ? ORDER BY revision ASC",
      )
      .all(afterRevision)
      .map((row) => ({
        revision: row.revision,
        event: EventSchema.parse(JSON.parse(row.payload_json)),
      }));
  }

  getLastEventTimestamp(): string | null {
    const row = this.db
      .query<{ payload_json: string }, []>(
        "SELECT payload_json FROM events ORDER BY revision DESC LIMIT 1",
      )
      .get();
    return row ? EventSchema.parse(JSON.parse(row.payload_json)).ts : null;
  }

  appendEvent(eventInput: ControllerEvent): number {
    let revision = 0;
    this.mutate(() => {
      revision = this.appendEventInTransaction(eventInput);
    });
    return revision;
  }

  private appendEventInTransaction(eventInput: ControllerEvent): number {
    if (this.eventWriteFails) throw new Error("event write failed");
    const revision = Number(this.getMeta("revision")) + 1;
    const event = EventSchema.parse(eventInput);
    this.db
      .prepare("INSERT INTO events (revision, payload_json) VALUES (?, ?)")
      .run(revision, JSON.stringify(event));
    this.setMeta("revision", String(revision));
    this.setMeta("reconciler_last_at", new Date(this.now()).toISOString());
    return revision;
  }

  pruneExpiredIdempotency(): void {
    const cutoff = this.now() - IDEMPOTENCY_RETENTION_MS;
    this.db
      .prepare("DELETE FROM idempotency WHERE created_at < ?")
      .run(cutoff);
  }

  getIdempotencyResult(key: string): IdempotencyRecord | null {
    this.pruneExpiredIdempotency();
    const row = this.db
      .query<
        { revision: number; result_json: string },
        [string]
      >("SELECT revision, result_json FROM idempotency WHERE idempotency_key = ?")
      .get(key);
    if (!row) return null;
    return {
      revision: row.revision,
      result: JSON.parse(row.result_json) as unknown,
    };
  }

  journalIntent(intent: TransitionIntent): number {
    if (this.auditWriteFails) {
      throw new AuditWriteError();
    }
    const createdAt = this.now();
    const result = this.db
      .prepare(
        `INSERT INTO transition_journal
         (verb, idempotency_key, args_json, expected_revision, status, created_at)
         VALUES (?, ?, ?, ?, 'pending', ?)
         RETURNING id`,
      )
      .get(
        intent.verb,
        intent.idempotencyKey,
        JSON.stringify(intent.args),
        intent.expectedRevision,
        createdAt,
      ) as { id: number } | null;
    if (!result) {
      throw new AuditWriteError("journal insert returned no row");
    }
    if (this.crashAfterJournal) {
      throw new Error("simulated crash after journal");
    }
    return result.id;
  }

  cancelJournal(journalId: number): void {
    this.db
      .prepare(
        "UPDATE transition_journal SET status = 'cancelled' WHERE id = ? AND status = 'pending'",
      )
      .run(journalId);
  }

  completeCancelledJournal(journalId: number, idempotencyKey: string, result: unknown): void {
    this.mutate(() => {
      const journal = this.db.query<{ status: string }, [number, string]>(
        "SELECT status FROM transition_journal WHERE id = ? AND idempotency_key = ?",
      ).get(journalId, idempotencyKey);
      if (!journal || journal.status !== "pending") throw new Error("pending transition journal required");
      this.db.prepare(
        "INSERT INTO idempotency (idempotency_key, revision, result_json, created_at) VALUES (?, ?, ?, ?)",
      ).run(idempotencyKey, this.getRevision(), JSON.stringify(result), this.now());
      this.db.prepare(
        "UPDATE transition_journal SET status = 'cancelled', result_json = ? WHERE id = ?",
      ).run(JSON.stringify(result), journalId);
    });
  }

  private recoverPendingTransitions(): void {
    const pending = this.db
      .query<
        {
          id: number;
          verb: string;
          idempotency_key: string;
          args_json: string;
          expected_revision: number;
        },
        []
      >(
        "SELECT id, verb, idempotency_key, args_json, expected_revision FROM transition_journal WHERE status = 'pending' ORDER BY id",
      )
      .all();

    for (const row of pending) {
      const existing = this.getIdempotencyResult(row.idempotency_key);
      if (existing) {
        this.db
          .prepare(
            "UPDATE transition_journal SET status = 'committed', result_json = ? WHERE id = ?",
          )
          .run(JSON.stringify(existing.result), row.id);
        continue;
      }
      // Resume will be handled by transitions layer via resumePendingJournal
    }
  }

  recoverOrphanHostSlotReservations(): void {
    this.mutate(() => {
      const orphaned = this.db
        .query<{ job_id: string; ticket_position: number }, []>(
          `SELECT r.job_id, r.ticket_position
           FROM host_slot_reservations r
           LEFT JOIN jobs j ON j.id = r.job_id
           WHERE r.released_at IS NULL AND j.id IS NULL`,
        )
        .all();
      for (const reservation of orphaned) {
        this.db
          .prepare(
            `UPDATE queue_tickets
             SET state = 'queued', dispatch_target = NULL, placement_kind = NULL,
                 placement_job_id = NULL, placed_at = NULL
             WHERE position = ? AND state = 'placed'`,
          )
          .run(reservation.ticket_position);
        this.db
          .prepare("DELETE FROM host_slot_reservations WHERE job_id = ?")
          .run(reservation.job_id);
      }
    });
  }

  getPendingJournal(idempotencyKey: string): ReturnType<ControllerStore["listPendingJournals"]>[number] | null {
    return this.listPendingJournals().find((journal) => journal.idempotencyKey === idempotencyKey) ?? null;
  }

  listPendingJournals(): Array<{
    id: number;
    verb: TransitionVerb;
    idempotencyKey: string;
    args: Record<string, unknown>;
    expectedRevision: number;
  }> {
    return this.db
      .query<
        {
          id: number;
          verb: string;
          idempotency_key: string;
          args_json: string;
          expected_revision: number;
        },
        []
      >(
        "SELECT id, verb, idempotency_key, args_json, expected_revision FROM transition_journal WHERE status = 'pending' ORDER BY id",
      )
      .all()
      .map((row) => ({
        id: row.id,
        verb: row.verb as TransitionVerb,
        idempotencyKey: row.idempotency_key,
        args: JSON.parse(row.args_json) as Record<string, unknown>,
        expectedRevision: row.expected_revision,
      }));
  }

  commitTransition(params: {
    journalId: number;
    idempotencyKey: string;
    result: unknown;
    apply: () => void;
    event?: ControllerEvent;
    acceptedRevision?: number;
  }): number {
    let nextRevision = 0;
    let staleRevision: number | null = null;
    this.mutate(() => {
      const current = Number(this.getMeta("revision"));
      if (
        params.acceptedRevision !== undefined &&
        current !== params.acceptedRevision
      ) {
        this.db
          .prepare(
            "UPDATE transition_journal SET status = 'cancelled' WHERE id = ? AND status = 'pending'",
          )
          .run(params.journalId);
        staleRevision = current;
        return;
      }
      nextRevision = current + 1;
      params.apply();
      if (params.event) {
        if (this.eventWriteFails) {
          throw new Error("event write failed");
        }
        const event = EventSchema.parse(params.event);
        this.db
          .prepare("INSERT INTO events (revision, payload_json) VALUES (?, ?)")
          .run(nextRevision, JSON.stringify(event));
      }
      this.setMeta("revision", String(nextRevision));
      this.setMeta("reconciler_last_at", new Date(this.now()).toISOString());
      this.db
        .prepare(
          "INSERT INTO idempotency (idempotency_key, revision, result_json, created_at) VALUES (?, ?, ?, ?)",
        )
        .run(
          params.idempotencyKey,
          nextRevision,
          JSON.stringify(params.result),
          this.now(),
        );
      this.db
        .prepare(
          "UPDATE transition_journal SET status = 'committed', result_json = ? WHERE id = ?",
        )
        .run(JSON.stringify(params.result), params.journalId);
    });
    if (staleRevision !== null) throw new StaleRevisionError(staleRevision);
    return nextRevision;
  }

  mutate<T>(fn: () => T): T {
    if (this.mutateDepth > 0) {
      return fn();
    }
    this.mutateDepth += 1;
    try {
      const run = this.db.transaction(fn);
      return run.immediate() as T;
    } finally {
      this.mutateDepth -= 1;
    }
  }

  isHostIdle(hostname: string): boolean {
    const host = this.getHost(hostname);
    if (!host) return true;
    return host.runningJobs === 0 && host.ciJobsRunning === 0;
  }

  isHostHealthGreen(hostname: string): boolean {
    const host = this.getHost(hostname);
    if (!host) return false;
    return (
      host.capabilityOk &&
      host.healthStorage &&
      host.healthRunner &&
      host.healthOffload
    );
  }

  countBuilderCapacity(): { builders: number; idleSlots: number } {
    const hosts = this.listHosts().filter((h) => h.role === "builder");
    const builders = hosts.length;
    const idleSlots = hosts.reduce(
      (sum, host) =>
        sum +
        Math.max(
          0,
          host.slotsTotal -
            host.slotsUsed -
            this.countHostSlotReservations(host.hostname),
        ),
      0,
    );
    return { builders, idleSlots };
  }

  hostEligible(
    hostname: string,
    command: string,
  ): { eligible: boolean; reason: string } {
    const host = this.getHost(hostname);
    if (!host) return { eligible: false, reason: "unknown-host" };
    if (host.role !== "builder") return { eligible: false, reason: "not-builder" };
    if (host.state !== "available") return { eligible: false, reason: "host-not-available" };
    if (host.capabilityOk !== true) return { eligible: false, reason: "capability-failed" };
    if (host.dispatchPaused) return { eligible: false, reason: "dispatch-paused" };
    const breaker = this.getCapabilityBreaker(hostname, command);
    if (breaker?.state === "open") {
      return { eligible: false, reason: "command-quarantined" };
    }
    if (host.slotsTotal - host.slotsUsed <= 0) {
      return { eligible: false, reason: "no-capacity" };
    }
    return { eligible: true, reason: "eligible" };
  }

  getSpineConfigState(configPath: string): SpineConfigStateRecord | null {
    const row = this.db
      .query<{ config_path: string; last_applied_body_hash: string }, [string]>(
        "SELECT config_path, last_applied_body_hash FROM spine_config_state WHERE config_path = ?",
      )
      .get(configPath);
    if (!row) return null;
    return {
      configPath: row.config_path,
      lastAppliedBodyHash: row.last_applied_body_hash,
    };
  }

  setSpineConfigState(record: SpineConfigStateRecord): void {
    this.db
      .prepare(
        `INSERT INTO spine_config_state (config_path, last_applied_body_hash)
         VALUES (?, ?)
         ON CONFLICT(config_path) DO UPDATE SET
           last_applied_body_hash = excluded.last_applied_body_hash`,
      )
      .run(record.configPath, record.lastAppliedBodyHash);
  }

  setSpineConfigDegraded(configPath: string): void {
    this.setMeta("desired", "degraded");
    this.setMeta("observed", "degraded");
    this.setMeta("dispatch_state", "paused");
    this.setMeta("dispatch_detail", `${SPINE_CONFIG_INVALID_PREFIX}${configPath}`);
    this.setMeta("reconciler_healthy", "0");
  }

  restoreNormalControllerMeta(configPath: string): void {
    const sentinel = `${SPINE_CONFIG_INVALID_PREFIX}${configPath}`;
    if (this.getMeta("dispatch_detail") !== sentinel) return;
    this.setMeta("desired", NORMAL_CONTROLLER_META.desired);
    this.setMeta("observed", NORMAL_CONTROLLER_META.observed);
    this.setMeta("dispatch_state", NORMAL_CONTROLLER_META.dispatch_state);
    this.setMeta("dispatch_detail", NORMAL_CONTROLLER_META.dispatch_detail);
    this.setMeta("reconciler_healthy", NORMAL_CONTROLLER_META.reconciler_healthy);
  }

  clearVanishedSpineConfigDegradation(now: string): string[] {
    const cleared: string[] = [];
    const detail = this.getMeta("dispatch_detail") ?? "";
    if (detail.startsWith(SPINE_CONFIG_INVALID_PREFIX)) {
      const path = detail.slice(SPINE_CONFIG_INVALID_PREFIX.length);
      if (path && !existsSync(path)) {
        this.restoreNormalControllerMeta(path);
        cleared.push(path);
      }
    }
    for (const incident of this.listIncidents()) {
      if (incident.state === "resolved") continue;
      if (!incident.key.startsWith(SPINE_CONFIG_INCIDENT_PREFIX)) continue;
      const path = incident.key.slice(SPINE_CONFIG_INCIDENT_PREFIX.length);
      if (!path || existsSync(path)) continue;
      this.resolveIncident(incident.key, now);
      if (!cleared.includes(path)) cleared.push(path);
    }
    return cleared;
  }

  appendEventInStore(eventInput: ControllerEvent): number {
    return this.appendEventInTransaction(eventInput);
  }
}


export type DeliveryReconcileResult = { activated: true; state: "OFF" | "INTERNAL" | "CANARY" | "ON" | "RETIRED" | "REMOVED"; receiptIds?: { acceptance: string; smoke: string }; deploymentId?: string } | { activated: false; reason: string };
export class DeliveryFeatureActivator {
  constructor(private readonly store: ControllerStore, private readonly registry: RuntimeFeatureRegistry, private readonly targetForFeature: (featureId: string) => string | null, private readonly now: () => number = () => Date.now()) {}
  reconcile(featureId: string, idempotencyKey: string, expectedRevision: number): DeliveryReconcileResult {
    return this.transition(featureId, "INTERNAL", idempotencyKey, expectedRevision);
  }
  transition(featureId: string, state: "INTERNAL" | "CANARY" | "ON" | "RETIRED" | "OFF", idempotencyKey: string, expectedRevision: number): DeliveryReconcileResult {
    const replay = this.store.getIdempotencyResult(idempotencyKey);
    if (replay) return replay.result as DeliveryReconcileResult;
    const targetId = this.targetForFeature(featureId);
    const pending = this.store.getPendingJournal(idempotencyKey);
    if (pending) {
      if (pending.verb !== "delivery-feature-reconcile" || pending.args.featureId !== featureId || pending.args.state !== state || pending.args.targetId !== targetId) throw new Error("idempotency key payload mismatch");
      return this.execute(featureId, state, idempotencyKey, pending.expectedRevision, pending.id, targetId);
    }
    const journalId = this.store.journalIntent({ verb: "delivery-feature-reconcile", idempotencyKey, args: { featureId, state, targetId }, expectedRevision });
    return this.execute(featureId, state, idempotencyKey, expectedRevision, journalId, targetId);
  }
  reconcileExpired(): DeliveryReconcileResult[] {
    const results: DeliveryReconcileResult[] = [];
    for (const [featureId, definition] of this.registry.definitions) {
      if (this.now() <= Date.parse(`${definition.definition.reviewDate}T23:59:59.999Z`)) continue;
      const current = this.store.getDeliveryFeatureState(featureId);
      if (!current) continue;
      if (current.state === "ON" || current.state === "INTERNAL" || current.state === "CANARY") { results.push(this.transition(featureId, "RETIRED", `delivery-expiry-retire:${featureId}:${definition.definition.reviewDate}`, this.store.getRevision())); continue; }
      if (current.state !== "RETIRED") continue;
      const removeAfter = Date.parse(current.transitionedAt) + definition.definition.rollbackWindowDays * 86_400_000;
      if (this.now() < removeAfter) continue;
      const idempotencyKey = `delivery-expiry-remove:${featureId}:${definition.definition.reviewDate}`;
      const replay = this.store.getIdempotencyResult(idempotencyKey);
      if (replay) { results.push(replay.result as DeliveryReconcileResult); continue; }
      const pending = this.store.getPendingJournal(idempotencyKey);
      if (pending) { results.push(this.executeRemoval(definition, idempotencyKey, pending.expectedRevision, pending.id)); continue; }
      const expectedRevision = this.store.getRevision();
      const journalId = this.store.journalIntent({ verb: "delivery-feature-reconcile", idempotencyKey, args: { featureId, state: "REMOVED", targetId: null }, expectedRevision });
      results.push(this.executeRemoval(definition, idempotencyKey, expectedRevision, journalId));
    }
    return results;
  }
  /** Called during startup before bind; only delivery intents are claimed here. */
  resumePending(): DeliveryReconcileResult[] {
    const results: DeliveryReconcileResult[] = [];
    for (const pending of this.store.listPendingJournals()) {
      if (pending.verb !== "delivery-feature-reconcile" || typeof pending.args.featureId !== "string") continue;
      const replay = this.store.getIdempotencyResult(pending.idempotencyKey);
      if (replay) { results.push(replay.result as DeliveryReconcileResult); continue; }
      if (typeof pending.args.state !== "string" || !["OFF", "INTERNAL", "CANARY", "ON", "RETIRED", "REMOVED"].includes(pending.args.state)) {
        results.push(this.refuse(pending.id, pending.idempotencyKey, "invalid-persisted-state"));
        continue;
      }
      if (pending.args.state === "REMOVED") {
        const definition = this.registry.definitions.get(pending.args.featureId);
        if (!definition) { results.push(this.refuse(pending.id, pending.idempotencyKey, "missing-definition")); continue; }
        try { results.push(this.executeRemoval(definition, pending.idempotencyKey, pending.expectedRevision, pending.id)); } catch { continue; }
        continue;
      }
      const state = pending.args.state as "OFF" | "INTERNAL" | "CANARY" | "ON" | "RETIRED";
      const targetId = pending.args.targetId === null || typeof pending.args.targetId === "string" ? pending.args.targetId : null;
      try { results.push(this.execute(pending.args.featureId, state, pending.idempotencyKey, pending.expectedRevision, pending.id, targetId)); }
      catch { continue; }
    }
    return results;
  }
  private executeRemoval(definition: LoadedFeatureDefinition, idempotencyKey: string, expectedRevision: number, journalId: number): DeliveryReconcileResult {
    if (this.store.getRevision() !== expectedRevision) return this.refuse(journalId, idempotencyKey, "stale-revision");
    const result: DeliveryReconcileResult = { activated: true, state: "REMOVED" };
    this.store.commitTransition({ journalId, idempotencyKey, result, acceptedRevision: expectedRevision, apply: () => this.store.removeRetiredDeliveryFeature(deliveryStoreCapability, definition, expectedRevision) });
    return result;
  }
  private execute(featureId: string, state: "OFF" | "INTERNAL" | "CANARY" | "ON" | "RETIRED", idempotencyKey: string, expectedRevision: number, journalId: number, targetId: string | null): DeliveryReconcileResult {
    if (this.store.getRevision() !== expectedRevision) return this.refuse(journalId, idempotencyKey, "stale-revision");
    const resolution = state === "OFF" || state === "RETIRED" ? this.registry.resolve(featureId) : resolveRuntimeFeatureDefinition(this.registry, featureId, this.now);
    if (!resolution.available) return this.refuse(journalId, idempotencyKey, resolution.reason);
    const current = this.store.getDeliveryFeatureState(featureId);
    const legal = state === "OFF" ? current !== null && current.state !== "OFF"
      : current === null ? state === "INTERNAL"
        : (current.state === "OFF" && state === "INTERNAL")
          || (current.state === "INTERNAL" && (state === "CANARY" || state === "RETIRED"))
          || (current.state === "CANARY" && (state === "ON" || state === "RETIRED"))
          || (current.state === "ON" && state === "RETIRED");
    if (!legal) return this.refuse(journalId, idempotencyKey, "illegal-state-transition");
    if (state === "OFF") {
      const result: DeliveryReconcileResult = { activated: true, state: "OFF" };
      this.store.commitTransition({ journalId, idempotencyKey, result, acceptedRevision: expectedRevision, event: buildTransitionEvent(this.store, "delivery-feature-reconcile", { featureId, state }, result, this.now), apply: () => {
        this.store.transitionDeliveryFeatureStateAuthorized(deliveryStoreCapability, { definition: resolution.definition, state, expectedRevision });
      } });
      return result;
    }
    const decision = this.decision(featureId, targetId, state === "RETIRED");
    if (!decision.admitted) return this.refuse(journalId, idempotencyKey, decision.reason);
    const result: DeliveryReconcileResult = { activated: true, state, deploymentId: decision.deployment.deploymentId, receiptIds: { acceptance: decision.acceptanceReceipt.receiptId, smoke: decision.smokeReceipt.receiptId } };
    this.store.commitTransition({ journalId, idempotencyKey, result, acceptedRevision: expectedRevision, event: buildTransitionEvent(this.store, "delivery-feature-reconcile", { featureId, state }, result, this.now), apply: () => {
      const finalResolution = state === "RETIRED" ? this.registry.resolve(featureId) : resolveRuntimeFeatureDefinition(this.registry, featureId, this.now);
      const final = this.decision(featureId, targetId, state === "RETIRED");
      if (!finalResolution.available || !final.admitted || final.deployment.deploymentId !== decision.deployment.deploymentId || final.acceptanceReceipt.receiptId !== decision.acceptanceReceipt.receiptId || final.smokeReceipt.receiptId !== decision.smokeReceipt.receiptId) throw new Error("delivery evidence drift");
      this.store.transitionDeliveryFeatureStateAuthorized(deliveryStoreCapability, { definition: finalResolution.definition, state, targetId: final.deployment.targetId, acceptanceReceiptId: final.acceptanceReceipt.receiptId, smokeReceiptId: final.smokeReceipt.receiptId, expectedRevision });
    } });
    return result;
  }
  private refuse(journalId: number, idempotencyKey: string, reason: string): DeliveryReconcileResult {
    const result: DeliveryReconcileResult = { activated: false, reason };
    this.store.completeCancelledJournal(journalId, idempotencyKey, result);
    return result;
  }
  private decision(featureId: string, targetId: string | null, allowExpired = false): ReadinessDecision { const definition = allowExpired ? this.registry.resolve(featureId) : resolveRuntimeFeatureDefinition(this.registry, featureId, this.now); return evaluateDeliveryReadiness(definition, { state: this.store.getDeliveryFeatureState(featureId), deployment: targetId ? this.store.getActiveDeliveryDeployment(targetId) : null, receipts: this.store.listDeliveryFeatureReceipts(featureId) }); }
}

export class DeliveryEvidenceAdmissionService {
  constructor(private readonly store: ControllerStore) {}
  admit(deployment: DeliveryDeploymentRecord, receipts: DeliveryFeatureReceiptRecord[], attestationId?: string, installedProof?: DeliveryInstalledProofRecord): void {
    if (!attestationId || !validId(attestationId)) throw new Error("delivery evidence attestation ID required");
    validateDeployment(deployment);
    if (receipts.length === 0) throw new Error("delivery evidence receipts required");
    for (const receipt of receipts) {
      validateReceipt(receipt);
      if (receipt.deploymentId !== deployment.deploymentId || receipt.targetId !== deployment.targetId || receipt.candidateSha !== deployment.deployedSha || receipt.candidateTree !== deployment.deployedTree || receipt.deployedSha !== deployment.deployedSha || receipt.deployedTree !== deployment.deployedTree || receipt.artifactDigest !== deployment.artifactDigest) throw new Error("delivery evidence identity mismatch");
    }
    this.store.mutate(() => {
      this.store.claimDeliveryEvidenceAttestation(attestationId);
      this.store.recordDeliveryDeployment(deliveryStoreCapability, deployment);
      for (const receipt of receipts) this.store.recordDeliveryFeatureReceipt(deliveryStoreCapability, receipt);
      if (installedProof) this.store.recordDeliveryInstalledProof(deliveryStoreCapability, installedProof);
    });
  }
}

function validId(value: string): boolean {
 return /^[a-zA-Z0-9][a-zA-Z0-9._-]{0,127}$/.test(value); }
function validTime(value: string): boolean { return !Number.isNaN(Date.parse(value)); }
function validateDeployment(value: DeliveryDeploymentRecord): void {
  if (!validId(value.deploymentId) || !validId(value.targetId) || !ExactGitShaSchema.safeParse(value.deployedSha).success || !ExactGitShaSchema.safeParse(value.deployedTree).success || !Sha256DigestSchema.safeParse(value.artifactDigest).success || !validTime(value.observedAt) || !value.evidenceReference) throw new Error("invalid delivery deployment evidence");
}
function validateInstalledProof(value: DeliveryInstalledProofRecord): void {
  if (!validId(value.proofId) || !validId(value.deploymentId) || !validId(value.targetId) || !ExactGitShaSchema.safeParse(value.deployedSha).success || !ExactGitShaSchema.safeParse(value.deployedTree).success || !Sha256DigestSchema.safeParse(value.artifactDigest).success || !value.entrypoint || !validTime(value.observedAt) || !value.evidenceReference) throw new Error("invalid delivery installed proof evidence");
}
function validateReceipt(value: DeliveryFeatureReceiptRecord): void {
  if (!validId(value.receiptId) || !validId(value.featureId) || !validId(value.targetId) || !validId(value.deploymentId) || !Sha256DigestSchema.safeParse(value.definitionDigest).success || ![value.candidateSha,value.candidateTree,value.deployedSha,value.deployedTree].every((sha) => ExactGitShaSchema.safeParse(sha).success) || !Sha256DigestSchema.safeParse(value.artifactDigest).success || !value.checkId || !value.evidenceReference || !validTime(value.observedAt)) throw new Error("invalid delivery feature receipt evidence");
}
function deploymentRow(row: any): DeliveryDeploymentRecord { return { deploymentId: row.deployment_id, targetId: row.target_id, deployedSha: row.deployed_sha, deployedTree: row.deployed_tree, artifactDigest: row.artifact_digest, status: row.status, observedAt: row.observed_at, evidenceReference: row.evidence_reference }; }
function installedProofRow(row: any): DeliveryInstalledProofRecord { return { proofId: row.proof_id, deploymentId: row.deployment_id, targetId: row.target_id, deployedSha: row.deployed_sha, deployedTree: row.deployed_tree, artifactDigest: row.artifact_digest, entrypoint: row.entrypoint, result: row.result, observedAt: row.observed_at, evidenceReference: row.evidence_reference }; }
function landOperationRow(row: {
  operation_id: string; receipt_json: string; fence: number; lease_expires_at: number;
  state: "waiting" | "succeeded" | "failed"; verdict_json: string | null; successor_intent_id: string | null;
}): LandOperationRecord {
  return {
    operationId: row.operation_id,
    receipt: JSON.parse(row.receipt_json) as LandOperationReceipt,
    fence: row.fence,
    leaseExpiresAt: row.lease_expires_at,
    state: row.state,
    verdict: row.verdict_json === null ? null : JSON.parse(row.verdict_json),
    successorIntentId: row.successor_intent_id,
  };
}

function isSuccessfulLandVerdict(verdict: unknown): boolean {
  if (!verdict || typeof verdict !== "object") return false;
  const value = verdict as { rc?: unknown; result?: { status?: unknown } };
  return value.rc === 0 && value.result?.status === "pushed";
}

function receiptRow(row: any): DeliveryFeatureReceiptRecord { return { receiptId: row.receipt_id, featureId: row.feature_id, definitionDigest: row.definition_digest, kind: row.kind, checkId: row.check_id, candidateSha: row.candidate_sha, candidateTree: row.candidate_tree, deploymentId: row.deployment_id, targetId: row.target_id, deployedSha: row.deployed_sha, deployedTree: row.deployed_tree, artifactDigest: row.artifact_digest, result: row.result, observedAt: row.observed_at, evidenceReference: row.evidence_reference }; }

function parseJobRow(row: {
  id: string;
  job_key: string | null;
  mirror: string | null;
  repo: string;
  host: string;
  snapshot: string;
  stage: string;
  attempt: number;
  rc: number | null;
  infra_failure: number;
  publication_state: string;
  publication_reason: string | null;
  started_at: string | null;
  finished_at: string | null;
  last_report_at: string | null;
  timeout_sec: number | null;
}): JobRecord {
  return {
    id: row.id,
    ...(row.job_key ? { key: row.job_key } : {}),
    ...(row.mirror ? { mirror: row.mirror } : {}),
    repo: row.repo,
    host: row.host,
    snapshot: row.snapshot,
    stage: row.stage,
    attempt: row.attempt,
    rc: row.rc,
    infraFailure: row.infra_failure === 1,
    ...(row.started_at ? { startedAt: row.started_at } : {}),
    ...(row.finished_at !== null ? { finishedAt: row.finished_at } : {}),
    ...(row.last_report_at ? { lastReportAt: row.last_report_at } : {}),
    ...(row.timeout_sec !== null ? { timeoutSec: row.timeout_sec } : {}),
    publication: {
      state: row.publication_state as NonNullable<JobRecord["publication"]>["state"],
      ...(row.publication_reason ? { reason: row.publication_reason } : {}),
    },
  };
}

function parseIncidentRow(row: IncidentRow): IncidentRecord {
  return {
    key: row.key,
    firstSeen: row.first_seen,
    lastSeen: row.last_seen,
    count: row.count,
    affectedJobs: JSON.parse(row.affected_jobs_json) as string[],
    remediation: row.remediation,
    cooldownUntil: row.cooldown_until,
    autoResolveCondition: row.auto_resolve_condition,
    state: row.state as IncidentState,
  };
}