sqlite.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/backends/sqlite.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_storage_foundation.py"
"""
SQLite backend for the orchestration registry.
 
Purpose
-------
Own all SQL used by the M11 registry storage foundation, including
canonical schema loading, connection PRAGMAs, init-time seeding, and
backend maintenance operations.
 
Responsibilities
----------------
- Load the canonical registry schema from the design-lock SQL file.
- Apply required SQLite PRAGMAs on every connection.
- Seed singleton rows for ``python -m oc_chat.registry init``.
- Execute the single mutating seam and insert one operation-log row.
- Own lock acquisition, release, renewal, and expiry sweep SQL.
- Provide integrity check, backup/restore, and vacuum hooks.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- SQL remains in this module; higher registry layers call backend methods.
- Init is all-or-nothing: schema load and seed inserts share one transaction.
- No SQLite ``datetime('now')`` defaults are used; timestamps come from Clock.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- sqlite3 (registry storage engine)
- oc_chat.registry.clock (injected time)
- oc_chat.registry.backends.base (StorageBackend contract, capability error and return types)
- oc_chat.registry.canonical (canonical_json for operation payloads and lock metadata)
- oc_chat.registry.errors (registry error types raised by the mutating seam and lock SQL)
- oc_chat.registry.migrations.v0_3_1_auth_events (auth events schema migration and version row)
- oc_chat.registry.migrations.v0_3_2_roadmap_source_fields (roadmap source fields schema migration and version row)
- oc_chat.registry.migrations.v0_3_roadmap (roadmap schema SQL, version row and upgrade checks)
- oc_chat.registry.store_scope (assert_live_store guard before each domain command)
"""
from __future__ import annotations
 
from contextlib import contextmanager
from dataclasses import dataclass
import json
from pathlib import Path
import shutil
import sqlite3
from typing import Iterator, Mapping
import uuid
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.backends.base import (
    BackendCapabilityError,
    LockHandle,
    OperationLogRow,
    StorageBackend,
)
from oc_chat.registry.canonical import canonical_json
from oc_chat.registry.clock import Clock, RealClock, wall_plus_seconds
from oc_chat.registry.errors import (
    ActorTypeMismatchError,
    HandoverStateError,
    LockHeldError,
    LockOwnerError,
    ParentCycleError,
    PayloadSchemaViolationError,
    PayloadVersionMismatchError,
    RoundStateError,
    UnknownActorError,
)
from oc_chat.registry.migrations import (
    v0_3_1_auth_events,
    v0_3_2_roadmap_source_fields,
    v0_3_roadmap,
)
from oc_chat.registry import store_scope as _store_scope
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
SCHEMA_VERSION = 1
SCHEMA_DESCRIPTION = "baseline REGISTRY_SCHEMA_V0_2_3"
_PRAGMA_PREFIX = "PRAGMA "
_ACTOR_TYPES_BY_SLOT_KIND = {
    "operator": "human_operator",
    "orchestrator": "orchestrator_agent",
    "worker": "worker_agent",
}
_MAX_PARENT_DEPTH = 1000
 
 
@dataclass(frozen=True)
class RegistrySeed:
    """Values substituted into the canonical seed template at init time."""
 
    workspace_id: str
    registry_id: str
    workspace_root: str
    git_common_dir: str
    registry_path: str
    checkpoint_path: str
    operator_slot: str
    hmac_key_id: str
    secret_fingerprint: str
    app_version: str
 
 
class SQLiteBackend(StorageBackend):
    """SQLite implementation of the registry storage backend."""
 
    def __init__(
        self,
        path: str | Path,
        clock: Clock | None = None,
        *,
        check_capabilities: bool = True,
    ) -> None:
        self.path = Path(path)
        self.clock = RealClock() if clock is None else clock
        if not self._is_memory:
            self.path.parent.mkdir(parents=True, exist_ok=True)
        if check_capabilities:
            super().__init__()
            self._auto_migrate_existing_database()
 
    @property
    def _is_memory(self) -> bool:
        return str(self.path) == ":memory:"
 
    def connect(self) -> sqlite3.Connection:
        """Open a connection with all required PRAGMAs applied."""
 
        conn = sqlite3.connect(str(self.path))
        conn.row_factory = sqlite3.Row
        self._apply_connection_pragmas(conn)
        return conn
 
    def apply_schema(self) -> None:
        """Apply the canonical schema baseline atomically."""
 
        with self.connect() as conn:
            self._run_schema_script(conn, commit=True)
 
    def initialize(self, seed: RegistrySeed) -> None:
        """Apply schema and seed singleton rows in one transaction."""
 
        # NOT `with self.connect()`: sqlite3's connection context manager is a
        # transaction manager and never closes — the leaked handle held a WAL
        # lock that made regenerate's os.replace swap fail under coverage.
        conn = self.connect()
        try:
            if self._table_count(conn) > 0:
                raise ValueError(f"registry database is not empty: {self.path}")
            self._run_schema_script(conn, commit=False)
            try:
                self._seed_initial_rows(conn, seed)
                conn.execute("COMMIT")
            except Exception:
                conn.execute("ROLLBACK")
                raise
        finally:
            conn.close()
 
    @contextmanager
    def transaction(self, mode: str = "immediate") -> Iterator[sqlite3.Connection]:
        conn = self.connect()
        try:
            begin = "BEGIN IMMEDIATE" if mode in {"immediate", "write"} else "BEGIN"
            conn.execute(begin)
            try:
                yield conn
                conn.execute("COMMIT")
            except Exception:
                conn.execute("ROLLBACK")
                raise
        finally:
            conn.close()
 
    def read_model(
        self,
        name: str,
        filters: Mapping[str, object] | None = None,
    ) -> object:
        raise NotImplementedError("read models are implemented in a later M11 plan")
 
    def execute_domain_command(
        self,
        operation_type: str,
        actor_slot: str,
        payload: Mapping[str, object],
        *,
        actor_type: str,
        payload_version: int = 1,
        command: str | None = None,
        parent_operation_id: int | None = None,
        related_plan_id: str | None = None,
        related_round: int | None = None,
        operation_uuid: str | None = None,
        hmac_key_id: str | None = None,
        system_actor: bool = False,
    ) -> OperationLogRow:
        with self.transaction("immediate") as conn:
            _store_scope.assert_live_store(conn)
            return self._execute_domain_command_in_conn(
                conn,
                operation_type,
                actor_slot,
                payload,
                actor_type=actor_type,
                payload_version=payload_version,
                command=command,
                parent_operation_id=parent_operation_id,
                related_plan_id=related_plan_id,
                related_round=related_round,
                operation_uuid=operation_uuid,
                hmac_key_id=hmac_key_id,
                system_actor=system_actor,
            )
 
    def apply_migration_projection(
        self,
        projection: Mapping[str, object],
        *,
        actor_slot: str,
        actor_type: str,
    ) -> list[OperationLogRow]:
        """Apply a legacy migration projection and emit seam audit rows."""
 
        emitted: list[OperationLogRow] = []
        with self.transaction("immediate") as conn:
            for agent in _mapping_list(projection.get("agents")):
                self._upsert_migration_agent(conn, agent)
                emitted.append(
                    self._execute_domain_command_in_conn(
                        conn,
                        "migrate.import.agent",
                        actor_slot,
                        _required_mapping(agent, "payload"),
                        actor_type=actor_type,
                        related_plan_id=None,
                    )
                )
 
            for round_row in _mapping_list(projection.get("rounds")):
                self._upsert_migration_round(conn, round_row)
 
            import_rows_by_plan: dict[str, OperationLogRow] = {}
            for plan in _mapping_list(projection.get("plans")):
                self._upsert_migration_plan(conn, plan)
                self._insert_migration_status_event(conn, plan, actor_slot)
                plan_id = str(plan["plan_id"])
                import_row = self._execute_domain_command_in_conn(
                    conn,
                    "migrate.import.plan",
                    actor_slot,
                    _required_mapping(plan, "payload"),
                    actor_type=actor_type,
                    related_plan_id=plan_id,
                    related_round=_optional_int(plan.get("closed_round")),
                )
                import_rows_by_plan[plan_id] = import_row
                emitted.append(import_row)
 
                backfill_payload = plan.get("backfill_payload")
                if isinstance(backfill_payload, Mapping):
                    emitted.append(
                        self._execute_domain_command_in_conn(
                            conn,
                            "migrate.backfill_completed",
                            actor_slot,
                            backfill_payload,
                            actor_type=actor_type,
                            parent_operation_id=import_row.id,
                            related_plan_id=plan_id,
                        )
                    )
 
                blocked_payload = plan.get("blocked_payload")
                if isinstance(blocked_payload, Mapping):
                    emitted.append(
                        self._execute_domain_command_in_conn(
                            conn,
                            "migrate.note_blocked",
                            actor_slot,
                            blocked_payload,
                            actor_type=actor_type,
                            parent_operation_id=import_row.id,
                            related_plan_id=plan_id,
                        )
                    )
 
            for round_plan in _mapping_list(projection.get("round_plans")):
                self._upsert_migration_round_plan(conn, round_plan)
 
            for queue_entry in _mapping_list(projection.get("queue_entries")):
                plan_id = str(queue_entry["plan_id"])
                emitted.append(
                    self._execute_domain_command_in_conn(
                        conn,
                        "migrate.import.queue_entry",
                        actor_slot,
                        _required_mapping(queue_entry, "payload"),
                        actor_type=actor_type,
                        parent_operation_id=(
                            import_rows_by_plan[plan_id].id
                            if plan_id in import_rows_by_plan
                            else None
                        ),
                        related_plan_id=plan_id,
                    )
                )
 
            max_plan_number = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT COALESCE(MAX(plan_number), 0) FROM plans"
            ).fetchone()[0]
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE plan_seq SET next_number = ? WHERE id = 1",
                (int(max_plan_number) + 1,),
            )
        return emitted
 
    def migration_snapshot(self) -> dict[str, object]:
        """Return the current-state rows that ``migrate verify`` compares."""
 
        with self.connect() as conn:
            agents = [
                _row_dict(row)
                for row in conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "SELECT slot, slot_kind, identity_kind, model_class, "
                    "orchestrator_eligible, provider, profile, active "
                    "FROM agents WHERE identity_kind = 'agent' "
                    "ORDER BY slot"
                ).fetchall()
            ]
            plans = [
                _row_dict(row)
                for row in conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "SELECT plan_id, plan_number, title, plan_type, status, "
                    "assigned_slot, queue_position, branch, worktree, body_path, "
                    "body_sha256, body_observed_at, model_override, "
                    "thinking_override, session_override, ready_for_merge_at, "
                    "completed_at, closed_round FROM plans ORDER BY plan_id"
                ).fetchall()
            ]
            queue_entries = [
                _row_dict(row)
                for row in conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "SELECT plan_id, assigned_slot AS slot, queue_position AS position "
                    "FROM plans WHERE queue_position IS NOT NULL "
                    "ORDER BY slot, position"
                ).fetchall()
            ]
            rounds = [
                _row_dict(row)
                for row in conn.execute(
                    "SELECT round_no, status, cleanup_status, orchestrator_slot, "
                    "opened_at, closed_at, final_commit_sha FROM rounds "
                    "ORDER BY round_no"
                ).fetchall()
            ]
            round_plans = [
                _row_dict(row)
                for row in conn.execute(
                    "SELECT round_no, plan_id, merge_order, merge_commit_sha, "
                    "test_result FROM round_plans ORDER BY round_no, merge_order"
                ).fetchall()
            ]
        return {
            "agents": agents,
            "plans": plans,
            "queue_entries": queue_entries,
            "rounds": rounds,
            "round_plans": round_plans,
        }
 
    def migration_null_round_backfill_violations(self) -> list[str]:
        """Return completed null-round plans missing a backfill audit marker."""
 
        with self.connect() as conn:
            plan_ids = [
                str(row["plan_id"])
                for row in conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "SELECT plan_id FROM plans "
                    "WHERE status = 'completed' AND closed_round IS NULL "
                    "ORDER BY plan_id"
                ).fetchall()
            ]
            backfilled: set[str] = set()
            for row in conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT args_json FROM operation_log "
                "WHERE operation_type = 'migrate.backfill_completed'"
            ).fetchall():
                try:
                    payload = json.loads(str(row["args_json"]))
                except (TypeError, ValueError):
                    continue
                plan_id = payload.get("plan_id") if isinstance(payload, dict) else None
                if isinstance(plan_id, str):
                    backfilled.add(plan_id)
        return [plan_id for plan_id in plan_ids if plan_id not in backfilled]
 
    def _execute_domain_command_in_conn(
        self,
        conn: sqlite3.Connection,
        operation_type: str,
        actor_slot: str,
        payload: Mapping[str, object],
        *,
        actor_type: str,
        payload_version: int = 1,
        command: str | None = None,
        parent_operation_id: int | None = None,
        related_plan_id: str | None = None,
        related_round: int | None = None,
        operation_uuid: str | None = None,
        hmac_key_id: str | None = None,
        system_actor: bool = False,
    ) -> OperationLogRow:
        command_text = command or operation_type
        operation_uuid = operation_uuid or str(uuid.uuid4())
        args_json = self._validate_payload(
            operation_type,
            payload,
            payload_version,
        )
 
        slot_kind = self._actor_slot_kind(conn, actor_slot)
        expected_actor_type = _ACTOR_TYPES_BY_SLOT_KIND[slot_kind]
        if actor_type != expected_actor_type:
            raise ActorTypeMismatchError(
                actor_slot=actor_slot,
                expected_actor_type=expected_actor_type,
                supplied_actor_type=actor_type,
            )
        self._assert_handover_allows(
            conn,
            operation_type,
            actor_slot,
            slot_kind,
        )
        if parent_operation_id is not None:
            self._assert_no_parent_cycle(
                conn,
                parent_operation_id,
                operation_uuid,
            )
        row_hmac_key_id = (
            None
            if system_actor
            else hmac_key_id or self._active_hmac_key_id(conn, actor_slot)
        )
        current_version = self._current_state_version(conn)
        next_version = current_version + 1
        now = self.clock.now_wall()
        cursor = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO operation_log("
            "operation_uuid, ts, state_version, actor_slot, actor_type, "
            "hmac_key_id, command, operation_type, payload_version, "
            "args_json, before_hash, after_hash, auth_ok, parent_operation_id, "
            "related_plan_id, related_round"
            ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, 1, ?, ?, ?)",
            (
                operation_uuid,
                now,
                next_version,
                actor_slot,
                actor_type,
                row_hmac_key_id,
                command_text,
                operation_type,
                payload_version,
                args_json,
                parent_operation_id,
                related_plan_id,
                related_round,
            ),
        )
        conn.execute(
            "UPDATE registry_meta SET state_version = ?, updated_at = ? "
            "WHERE id = 1",
            (next_version, now),
        )
        self._apply_domain_projection(
            conn,
            operation_type,
            payload,
            actor_slot=actor_slot,
            ts=now,
            related_round=related_round,
            source_operation_id=int(cursor.lastrowid),
        )
        return OperationLogRow(
            id=int(cursor.lastrowid),
            operation_uuid=operation_uuid,
            ts=now,
            state_version=next_version,
            actor_slot=actor_slot,
            actor_type=actor_type,
            command=command_text,
            operation_type=operation_type,
            payload_version=payload_version,
            args_json=args_json,
            parent_operation_id=parent_operation_id,
            related_plan_id=related_plan_id,
            related_round=related_round,
            hmac_key_id=row_hmac_key_id,
        )
 
    def _apply_domain_projection(
        self,
        conn: sqlite3.Connection,
        operation_type: str,
        payload: Mapping[str, object],
        *,
        actor_slot: str,
        ts: str,
        related_round: int | None,
        source_operation_id: int,
    ) -> None:
        if operation_type == "plan.create":
            self._project_plan_create(conn, payload, ts)
        elif operation_type == "plan.set_status":
            self._project_plan_set_status(conn, payload, actor_slot, ts)
        elif operation_type == "plan.assign":
            self._project_plan_assign(conn, payload, ts)
        elif operation_type == "plan.supersede":
            self._project_plan_supersede(conn, payload, ts)
        elif operation_type == "plan.reimport":
            self._project_plan_reimport(conn, payload, ts)
        elif operation_type == "queue.enqueue":
            self._project_queue_enqueue(conn, payload, ts)
        elif operation_type == "queue.move":
            self._project_queue_move(conn, payload, ts)
        elif operation_type == "queue.dequeue":
            self._project_queue_dequeue(conn, payload, ts)
        elif operation_type == "queue.complete":
            self._project_queue_complete(conn, payload, ts, related_round)
        elif operation_type in {"role.handover.start", "handover.initiate"}:
            self._project_handover_start(conn, payload, actor_slot, ts)
        elif operation_type in {"role.handover.complete", "handover.accept"}:
            self._project_handover_complete(conn, payload, actor_slot, ts)
        elif operation_type == "roadmap_item.create":
            self._project_roadmap_item_create(
                conn,
                payload,
                actor_slot,
                ts,
                source_operation_id,
            )
        elif operation_type == "roadmap_item.set_status":
            self._project_roadmap_item_set_status(
                conn,
                payload,
                ts,
                source_operation_id,
            )
        elif operation_type == "roadmap_item.add_dependency":
            self._project_roadmap_item_add_dependency(
                conn,
                payload,
                actor_slot,
                ts,
                source_operation_id,
            )
        elif operation_type == "roadmap_version.bump":
            self._project_roadmap_version_bump(conn, payload, ts)
        elif operation_type == "roadmap_phase.complete":
            self._project_roadmap_phase_complete(
                conn,
                payload,
                ts,
                source_operation_id,
            )
        elif operation_type == "roadmap_section.upsert":
            self._project_roadmap_section_upsert(conn, payload)
        elif operation_type == "roadmap_plan_link.create":
            self._project_roadmap_plan_link_create(conn, payload, actor_slot, ts)
        elif operation_type == "round.open":
            self._project_round_open(conn, payload, actor_slot, ts)
        elif operation_type == "round.merge":
            self._project_round_merge(conn, payload, ts)
        elif operation_type == "round.test":
            self._project_round_test(conn, payload)
        elif operation_type == "round.lint":
            self._project_round_lint(conn, payload)
        elif operation_type == "round.consolidate":
            self._project_round_consolidate(conn, payload)
        elif operation_type == "round.ff_main":
            self._project_round_ff_main(conn, payload)
        elif operation_type == "round.cleanup":
            self._project_round_cleanup(conn, payload, ts)
        elif operation_type == "round.abort":
            self._project_round_abort(conn, payload, ts)
 
    def _project_handover_start(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        actor_slot: str,
        ts: str,
    ) -> None:
        state = self._orchestration_state(conn)
        if state["handover_status"] != "stable":
            raise HandoverStateError(
                state="handover_in_flight",
                actor_slot=actor_slot,
                operation_type="role.handover.start",
            )
        from_slot = str(payload["from_slot"])
        to_slot = str(payload["to_slot"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE orchestration_state SET "
            "orchestrator_slot = COALESCE(orchestrator_slot, ?), "
            "handover_status = 'requested', handover_outgoing = ?, "
            "handover_incoming = ?, handover_requested_at = ?, "
            "ready_to_handover = 0, updated_at = ?, updated_by = ? "
            "WHERE id = 1",
            (from_slot, from_slot, to_slot, ts, ts, actor_slot),
        )
 
    def _project_handover_complete(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        actor_slot: str,
        ts: str,
    ) -> None:
        state = self._orchestration_state(conn)
        to_slot = str(payload["to_slot"])
        if state["handover_status"] not in {"requested", "draining"}:
            raise HandoverStateError(
                state="no_handover_pending",
                actor_slot=actor_slot,
                operation_type="role.handover.complete",
                handover_id=_optional_int(payload.get("parent_operation_id")),
            )
        if str(state["handover_incoming"]) != to_slot:
            raise HandoverStateError(
                state="handover_target_mismatch",
                actor_slot=actor_slot,
                operation_type="role.handover.complete",
                handover_id=_optional_int(payload.get("parent_operation_id")),
            )
        if actor_slot != to_slot and self._actor_slot_kind(conn, actor_slot) != "operator":
            raise HandoverStateError(
                state="wrong_accepting_slot",
                actor_slot=actor_slot,
                operation_type="role.handover.complete",
                handover_id=_optional_int(payload.get("parent_operation_id")),
            )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE orchestration_state SET "
            "orchestrator_slot = ?, handover_status = 'stable', "
            "handover_outgoing = NULL, handover_incoming = NULL, "
            "handover_requested_at = NULL, ready_to_handover = 0, "
            "updated_at = ?, updated_by = ? WHERE id = 1",
            (to_slot, ts, actor_slot),
        )
 
    def _project_plan_create(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        plan_number = int(payload["plan_number"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO plans("
            "plan_id, plan_number, title, plan_type, status, assigned_slot, "
            "queue_position, branch, worktree, body_path, body_sha256, "
            "body_observed_at, model_override, thinking_override, "
            "session_override, created_at, updated_at"
            ") VALUES (?, ?, ?, ?, 'draft', NULL, NULL, NULL, NULL, ?, ?, "
            "?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(plan_id) DO UPDATE SET "
            "plan_number = excluded.plan_number, "
            "title = excluded.title, "
            "plan_type = excluded.plan_type, "
            "body_path = excluded.body_path, "
            "body_sha256 = excluded.body_sha256, "
            "body_observed_at = excluded.body_observed_at, "
            "model_override = excluded.model_override, "
            "thinking_override = excluded.thinking_override, "
            "session_override = excluded.session_override, "
            "updated_at = excluded.updated_at",
            (
                str(payload["plan_id"]),
                plan_number,
                str(payload["title"]),
                _optional_str(payload.get("plan_type")),
                str(payload["body_path"]),
                str(payload["body_sha256"]),
                _optional_str(payload.get("body_observed_at")) or ts,
                _optional_str(payload.get("model_override")),
                _optional_str(payload.get("thinking_override")),
                _optional_str(payload.get("session_override")),
                ts,
                ts,
            ),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plan_seq SET next_number = CASE "
            "WHEN next_number <= ? THEN ? ELSE next_number END WHERE id = 1",
            (plan_number, plan_number + 1),
        )
 
    def _project_plan_set_status(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        actor_slot: str,
        ts: str,
    ) -> None:
        plan_id = str(payload["plan_id"])
        to_status = str(payload["to_status"])
        conn.execute(
            "INSERT INTO plan_status_events("
            "plan_id, from_status, to_status, actor_slot, source, created_at, note"
            ") VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                plan_id,
                _optional_str(payload.get("from_status")),
                to_status,
                actor_slot,
                _optional_str(payload.get("source")),
                ts,
                _optional_str(payload.get("note")),
            ),
        )
        queue_position_expr = (
            "queue_position"
            if to_status in {"pending", "implementing", "ready_for_merge"}
            else "NULL"
        )
        conn.execute(
            "UPDATE plans SET "
            "status = ?, "
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            f"queue_position = {queue_position_expr}, "
            "ready_for_merge_at = CASE "
            "WHEN ? = 'ready_for_merge' AND ready_for_merge_at IS NULL THEN ? "
            "ELSE ready_for_merge_at END, "
            "completed_at = CASE "
            "WHEN ? = 'completed' AND completed_at IS NULL THEN ? "
            "ELSE completed_at END, "
            "updated_at = ? "
            "WHERE plan_id = ?",
            (to_status, to_status, ts, to_status, ts, ts, plan_id),
        )
 
    def _project_plan_assign(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET assigned_slot = ?, branch = ?, worktree = ?, "
            "queue_position = ?, status = CASE "
            "WHEN status = 'draft' THEN 'pending' ELSE status END, "
            "updated_at = ? WHERE plan_id = ?",
            (
                str(payload["to_slot"]),
                str(payload["branch"]),
                str(payload["worktree"]),
                int(payload["queue_position"]),
                ts,
                str(payload["plan_id"]),
            ),
        )
 
    def _project_plan_supersede(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET status = 'superseded', queue_position = NULL, "
            "superseded_by = ?, updated_at = ? WHERE plan_id = ?",
            (
                str(payload["superseded_by_plan_id"]),
                ts,
                str(payload["plan_id"]),
            ),
        )
 
    def _project_plan_reimport(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET body_sha256 = ?, body_observed_at = ?, "
            "updated_at = ? WHERE plan_id = ?",
            (
                str(payload["new_body_sha256"]),
                str(payload["new_body_observed_at"]),
                ts,
                str(payload["plan_id"]),
            ),
        )
 
    def _project_queue_enqueue(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        plan_id = str(payload["plan_id"])
        slot = str(payload["slot"])
        position = int(payload["position"])
        if not self._plan_exists(conn, plan_id):
            return
        current = self._plan_queue_location(conn, plan_id)
        if current is not None:
            self._remove_queue_position(conn, current[0], current[1], plan_id)
        self._shift_queue_positions_up(conn, slot, position, exclude_plan_id=plan_id)
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET assigned_slot = ?, queue_position = ?, "
            "status = CASE WHEN status IN "
            "('draft','partial','failed','superseded','completed') "
            "THEN 'pending' ELSE status END, updated_at = ? "
            "WHERE plan_id = ?",
            (slot, position, ts, plan_id),
        )
 
    def _project_queue_move(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        plan_id = str(payload["plan_id"])
        if not self._plan_exists(conn, plan_id):
            return
        current = self._plan_queue_location(conn, plan_id)
        from_slot = current[0] if current is not None else str(payload["from_slot"])
        to_slot = str(payload["to_slot"])
        from_position = current[1] if current is not None else int(payload["from_position"])
        to_position = int(payload["to_position"])
        self._remove_queue_position(conn, from_slot, from_position, plan_id)
        self._shift_queue_positions_up(conn, to_slot, to_position, exclude_plan_id=plan_id)
        conn.execute(
            "UPDATE plans SET assigned_slot = ?, queue_position = ?, "
            "updated_at = ? WHERE plan_id = ?",
            (to_slot, to_position, ts, plan_id),
        )
 
    def _project_queue_dequeue(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        plan_id = str(payload["plan_id"])
        current = self._plan_queue_location(conn, plan_id)
        slot = current[0] if current is not None else str(payload["slot"])
        position = current[1] if current is not None else int(payload["position"])
        reason = str(payload["reason"])
        self._remove_queue_position(conn, slot, position, plan_id)
        if reason == "transition_to_failed":
            conn.execute(
                "UPDATE plans SET status = 'failed', updated_at = ? WHERE plan_id = ?",
                (ts, plan_id),
            )
        elif reason == "transition_to_integrating":
            conn.execute(
                "UPDATE plans SET status = 'integrating', updated_at = ? "
                "WHERE plan_id = ?",
                (ts, plan_id),
            )
        else:
            conn.execute(
                "UPDATE plans SET updated_at = ? WHERE plan_id = ?",
                (ts, plan_id),
            )
 
    def _project_queue_complete(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
        related_round: int | None,
    ) -> None:
        round_no = int(payload["round_no"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET status = 'completed', queue_position = NULL, "
            "completed_at = COALESCE(completed_at, ?), closed_round = ?, "
            "updated_at = ? WHERE plan_id = ?",
            (
                ts,
                related_round if related_round is not None else round_no,
                ts,
                str(payload["plan_id"]),
            ),
        )
 
    def _project_roadmap_item_create(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        actor_slot: str,
        ts: str,
        source_operation_id: int,
    ) -> None:
        item_id = str(payload["item_id"])
        track_kind = str(payload.get("track_kind") or "m_phase")
        phase_code = _optional_str(payload.get("phase_code"))
        phase_id = None
        if track_kind == "m_phase":
            phase_code = phase_code or _phase_code_for_item_id(item_id)
            if phase_code is None:
                raise ValueError("roadmap m_phase item requires phase_code")
            phase_id = self._roadmap_phase_id(
                conn,
                phase_code,
                ts=ts,
                label=_optional_str(payload.get("phase_label")) or phase_code,
            )
        parent_item_id = _optional_str(payload.get("parent_item_id"))
        if parent_item_id is not None:
            parent_row = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT 1 FROM roadmap_items_v3 WHERE item_id = ?",
                (parent_item_id,),
            ).fetchone()
            if parent_row is None:
                parent_item_id = None
        introduced = _optional_str(payload.get("introduced_in_version_id"))
        status = str(payload.get("status") or "todo")
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO roadmap_items_v3("
            "item_id, track_kind, phase_id, parent_item_id, title, prose, prose_path, "
            "status, priority, sort_order, target_version, introduced_in_version_id, "
            "closed_in_version_id, closed_round_no, superseded_by_item_id, "
            "created_at, updated_at"
            ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(item_id) DO UPDATE SET "
            "track_kind = excluded.track_kind, "
            "phase_id = excluded.phase_id, "
            "parent_item_id = excluded.parent_item_id, "
            "title = excluded.title, "
            "prose = excluded.prose, "
            "prose_path = excluded.prose_path, "
            "status = excluded.status, "
            "priority = excluded.priority, "
            "sort_order = excluded.sort_order, "
            "target_version = excluded.target_version, "
            "introduced_in_version_id = COALESCE("
            "roadmap_items_v3.introduced_in_version_id, "
            "excluded.introduced_in_version_id"
            "), "
            "closed_in_version_id = excluded.closed_in_version_id, "
            "closed_round_no = excluded.closed_round_no, "
            "superseded_by_item_id = excluded.superseded_by_item_id, "
            "updated_at = excluded.updated_at",
            (
                item_id,
                track_kind,
                phase_id,
                parent_item_id,
                str(payload["title"]),
                _optional_str(payload.get("prose")),
                _optional_str(payload.get("prose_path")),
                status,
                _optional_str(payload.get("priority")),
                int(payload.get("sort_order") or 0),
                _optional_str(payload.get("target_version")),
                introduced,
                _optional_str(payload.get("closed_in_version_id")),
                _optional_int(payload.get("closed_round_no")),
                _optional_str(payload.get("superseded_by_item_id")),
                ts,
                ts,
            ),
        )
        self._insert_roadmap_event(
            conn,
            item_id,
            "created",
            from_value_json=None,
            to_value_json=canonical_json(dict(payload)),
            in_version_id=introduced,
            source_operation_id=source_operation_id,
            ts=ts,
            note=f"created by {actor_slot}",
        )
 
    def _project_roadmap_item_set_status(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
        source_operation_id: int,
    ) -> None:
        item_id = str(payload["item_id"])
        from_status = _optional_str(payload.get("from_status"))
        to_status = str(payload["to_status"])
        in_version_id = _optional_str(payload.get("in_version_id"))
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE roadmap_items_v3 SET status = ?, "
            "closed_in_version_id = CASE "
            "WHEN ? IN ('done','deferred','superseded') THEN COALESCE(?, closed_in_version_id) "
            "ELSE closed_in_version_id END, "
            "closed_round_no = COALESCE(?, closed_round_no), "
            "superseded_by_item_id = COALESCE(?, superseded_by_item_id), "
            "updated_at = ? WHERE item_id = ?",
            (
                to_status,
                to_status,
                in_version_id,
                _optional_int(payload.get("closed_round_no")),
                _optional_str(payload.get("superseded_by_item_id")),
                ts,
                item_id,
            ),
        )
        self._insert_roadmap_event(
            conn,
            item_id,
            "status_change",
            from_value_json=canonical_json({"status": from_status}),
            to_value_json=canonical_json({"status": to_status}),
            in_version_id=in_version_id,
            source_operation_id=source_operation_id,
            ts=ts,
            note=_optional_str(payload.get("note")),
        )
 
    def _project_roadmap_item_add_dependency(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        actor_slot: str,
        ts: str,
        source_operation_id: int,
    ) -> None:
        from_item_id = str(payload["from_item_id"])
        to_item_id = str(payload["to_item_id"])
        kind = str(payload.get("kind") or "gates")
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT OR IGNORE INTO roadmap_item_dependencies("
            "from_item_id, to_item_id, kind, created_at, created_by_slot"
            ") VALUES (?, ?, ?, ?, ?)",
            (from_item_id, to_item_id, kind, ts, actor_slot),
        )
        self._insert_roadmap_event(
            conn,
            from_item_id,
            "dependency_added",
            from_value_json=None,
            to_value_json=canonical_json({"to_item_id": to_item_id, "kind": kind}),
            in_version_id=_optional_str(payload.get("in_version_id")),
            source_operation_id=source_operation_id,
            ts=ts,
            note=_optional_str(payload.get("note")),
        )
 
    def _project_roadmap_version_bump(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        from_version = _optional_str(payload.get("from_version_id"))
        to_version = str(payload["to_version_id"])
        era_id = _optional_str(payload.get("era_id")) or _era_id_for_version(to_version)
        self._roadmap_era(conn, era_id, ts)
        status = str(payload.get("status") or "draft")
        if status == "current":
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE roadmap_versions SET status = 'superseded', "
                "superseded_by_version_id = ?, last_updated_at = ? "
                "WHERE era_id = ? AND status = 'current'",
                (to_version, ts, era_id),
            )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO roadmap_versions("
            "version_id, era_id, released_at, status, parent_version_id, "
            "supersedes_version_id, superseded_by_version_id, last_updated_at, "
            "source_round_no, bump_kind, notes"
            ") VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?) "
            "ON CONFLICT(version_id) DO UPDATE SET "
            "status = excluded.status, "
            "last_updated_at = excluded.last_updated_at, "
            "source_round_no = excluded.source_round_no, "
            "bump_kind = excluded.bump_kind, "
            "notes = excluded.notes",
            (
                to_version,
                era_id,
                _optional_str(payload.get("released_at")) or ts,
                status,
                from_version,
                from_version,
                _optional_str(payload.get("last_updated_at")) or ts,
                _optional_int(payload.get("source_round_no")),
                str(payload.get("bump_kind") or "patch"),
                _optional_str(payload.get("notes")),
            ),
        )
 
    def _project_roadmap_phase_complete(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
        source_operation_id: int,
    ) -> None:
        phase_code = str(payload["phase_code"])
        in_version_id = _optional_str(payload.get("in_version_id"))
        row = conn.execute(
            "SELECT item_id FROM roadmap_items_v3 WHERE item_id = ?",
            (phase_code,),
        ).fetchone()
        if row is not None:
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE roadmap_items_v3 SET status = 'done', "
                "closed_in_version_id = COALESCE(?, closed_in_version_id), "
                "updated_at = ? WHERE item_id = ?",
                (in_version_id, ts, phase_code),
            )
            self._insert_roadmap_event(
                conn,
                phase_code,
                "status_change",
                from_value_json=None,
                to_value_json=canonical_json({"status": "done"}),
                in_version_id=in_version_id,
                source_operation_id=source_operation_id,
                ts=ts,
                note=_optional_str(payload.get("note")),
            )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE roadmap_phases SET retired_in_version_id = COALESCE(?, retired_in_version_id), "
            "updated_at = ? WHERE phase_code = ?",
            (in_version_id, ts, phase_code),
        )
 
    def _project_roadmap_section_upsert(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "DELETE FROM roadmap_render_sections "
            "WHERE file_kind = ? AND section_path = ?",
            (str(payload["file_kind"]), str(payload["section_path"])),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO roadmap_render_sections("
            "file_kind, section_path, sort_order, query_kind, "
            "query_params_json, required_min_version, active"
            ") VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                str(payload["file_kind"]),
                str(payload["section_path"]),
                int(payload["sort_order"]),
                str(payload["query_kind"]),
                _optional_str(payload.get("query_params_json")),
                _optional_str(payload.get("required_min_version")),
                _bool_int(payload.get("active", True)),
            ),
        )
 
    def _project_roadmap_plan_link_create(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        actor_slot: str,
        ts: str,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT OR REPLACE INTO roadmap_plan_links("
            "plan_id, item_id, contribution, created_at, created_by_slot"
            ") VALUES (?, ?, ?, ?, ?)",
            (
                str(payload["plan_id"]),
                str(payload["item_id"]),
                str(payload.get("contribution") or "partial"),
                ts,
                actor_slot,
            ),
        )
 
    def _roadmap_era(
        self,
        conn: sqlite3.Connection,
        era_id: str,
        ts: str,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT OR IGNORE INTO roadmap_eras("
            "era_id, era_label, major_bump_reason, started_at, ended_at, "
            "archived_under, sort_order"
            ") VALUES (?, ?, NULL, ?, NULL, NULL, ?)",
            (era_id, f"{era_id} roadmap era", ts, _era_sort_order(era_id)),
        )
 
    def _roadmap_phase_id(
        self,
        conn: sqlite3.Connection,
        phase_code: str,
        *,
        ts: str,
        label: str,
    ) -> int:
        parent_code = _parent_phase_code(phase_code)
        parent_phase_id = None
        if parent_code is not None and parent_code != phase_code:
            parent_phase_id = self._roadmap_phase_id(
                conn,
                parent_code,
                ts=ts,
                label=parent_code,
            )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO roadmap_phases("
            "phase_code, label, parent_phase_id, branch_suggestion, "
            "commit_prefix, target_version, priority_label, parallel_track, "
            "full_spec_url, sort_order, introduced_in_version_id, "
            "retired_in_version_id, created_at, updated_at"
            ") VALUES (?, ?, ?, NULL, NULL, NULL, NULL, ?, NULL, ?, NULL, NULL, ?, ?) "
            "ON CONFLICT(phase_code) DO UPDATE SET "
            "label = excluded.label, "
            "parent_phase_id = COALESCE(roadmap_phases.parent_phase_id, excluded.parent_phase_id), "
            "updated_at = excluded.updated_at",
            (
                phase_code,
                label,
                parent_phase_id,
                1 if phase_code == "M11" else 0,
                _phase_sort_order(phase_code),
                ts,
                ts,
            ),
        )
        row = conn.execute(
            "SELECT phase_id FROM roadmap_phases WHERE phase_code = ?",
            (phase_code,),
        ).fetchone()
        if row is None:
            raise ValueError(f"failed to create roadmap phase: {phase_code}")
        return int(row["phase_id"])
 
    def _insert_roadmap_event(
        self,
        conn: sqlite3.Connection,
        item_id: str,
        event_kind: str,
        *,
        from_value_json: str | None,
        to_value_json: str | None,
        in_version_id: str | None,
        source_operation_id: int,
        ts: str,
        note: str | None,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO roadmap_item_events("
            "item_id, event_kind, from_value_json, to_value_json, "
            "in_version_id, source_operation_id, source_version_seen, ts, note"
            ") VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?)",
            (
                item_id,
                event_kind,
                from_value_json,
                to_value_json,
                in_version_id,
                source_operation_id,
                ts,
                note,
            ),
        )
 
    def _plan_exists(self, conn: sqlite3.Connection, plan_id: str) -> bool:
        row = conn.execute("SELECT 1 FROM plans WHERE plan_id = ?", (plan_id,)).fetchone()
        return row is not None
 
    def _plan_queue_location(
        self,
        conn: sqlite3.Connection,
        plan_id: str,
    ) -> tuple[str, int] | None:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "SELECT assigned_slot, queue_position FROM plans WHERE plan_id = ?",
            (plan_id,),
        ).fetchone()
        if row is None or row["assigned_slot"] is None or row["queue_position"] is None:
            return None
        return str(row["assigned_slot"]), int(row["queue_position"])
 
    def _shift_queue_positions_up(
        self,
        conn: sqlite3.Connection,
        slot: str,
        start_position: int,
        *,
        exclude_plan_id: str | None = None,
    ) -> None:
        if exclude_plan_id is None:
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE plans SET queue_position = queue_position + 100000 "
                "WHERE assigned_slot = ? AND queue_position >= ?",
                (slot, start_position),
            )
        else:
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE plans SET queue_position = queue_position + 100000 "
                "WHERE assigned_slot = ? AND queue_position >= ? AND plan_id != ?",
                (slot, start_position, exclude_plan_id),
            )
        conn.execute(
            "UPDATE plans SET queue_position = queue_position - 99999 "
            "WHERE assigned_slot = ? AND queue_position >= ?",
            (slot, start_position + 100000),
        )
 
    def _remove_queue_position(
        self,
        conn: sqlite3.Connection,
        slot: str,
        position: int,
        plan_id: str,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET queue_position = NULL WHERE plan_id = ?",
            (plan_id,),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET queue_position = queue_position + 100000 "
            "WHERE assigned_slot = ? AND queue_position > ?",
            (slot, position),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE plans SET queue_position = queue_position - 100001 "
            "WHERE assigned_slot = ? AND queue_position >= ?",
            (slot, position + 100001),
        )
 
    def _upsert_migration_agent(
        self,
        conn: sqlite3.Connection,
        agent: Mapping[str, object],
    ) -> None:
        now = self.clock.now_wall()
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO agents("
            "slot, slot_kind, identity_kind, model_class, orchestrator_eligible, "
            "provider, profile, active, created_at"
            ") VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?) "
            "ON CONFLICT(slot) DO UPDATE SET "
            "slot_kind = excluded.slot_kind, "
            "identity_kind = excluded.identity_kind, "
            "model_class = excluded.model_class, "
            "orchestrator_eligible = excluded.orchestrator_eligible, "
            "provider = excluded.provider, "
            "profile = excluded.profile, "
            "active = 1",
            (
                str(agent["slot"]),
                str(agent["slot_kind"]),
                str(agent["identity_kind"]),
                str(agent["model_class"]),
                _bool_int(agent["orchestrator_eligible"]),
                _optional_str(agent.get("provider")),
                _optional_str(agent.get("profile")),
                now,
            ),
        )
 
    # -- M11.7 round-close projections (operation_log seam) ------------------
 
    def _project_round_open(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        actor_slot: str,
        ts: str,
    ) -> None:
        round_no = int(payload["round_no"])
        existing = conn.execute(
            "SELECT status FROM rounds WHERE round_no = ?",
            (round_no,),
        ).fetchone()
        if existing is not None and str(existing["status"]) in {"closed", "aborted"}:
            raise RoundStateError(
                round_no=round_no,
                status=str(existing["status"]),
                operation_type="round.open",
            )
        orchestrator_slot = _optional_str(payload.get("orchestrator_slot")) or actor_slot
        # SF-001: idempotent-additive — a re-open never clobbers an open round.
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO rounds(round_no, status, cleanup_status, "
            "orchestrator_slot, opened_at) VALUES (?, 'open', 'pending', ?, ?) "
            "ON CONFLICT(round_no) DO NOTHING",
            (round_no, orchestrator_slot, ts),
        )
        plan_id = _optional_str(payload.get("plan_id"))
        if plan_id is not None:
            merge_order = _optional_int(payload.get("merge_order"))
            if merge_order is None:
                merge_order = self._next_round_merge_order(conn, round_no)
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "INSERT INTO round_plans(round_no, plan_id, merge_order) "
                "VALUES (?, ?, ?) ON CONFLICT(round_no, plan_id) DO NOTHING",
                (round_no, plan_id, merge_order),
            )
 
    def _next_round_merge_order(
        self,
        conn: sqlite3.Connection,
        round_no: int,
    ) -> int:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "SELECT COALESCE(MAX(merge_order), 0) AS hi FROM round_plans "
            "WHERE round_no = ?",
            (round_no,),
        ).fetchone()
        return int(row["hi"]) + 1
 
    def _project_round_merge(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        round_no = int(payload["round_no"])
        plan_id = str(payload["plan_id"])
        staging_branch = _optional_str(payload.get("staging_branch"))
        merge_base_sha = _optional_str(payload.get("merge_base_sha"))
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE rounds SET status = 'closing', "
            "staging_branch = COALESCE(?, staging_branch), "
            "merge_base_sha = COALESCE(?, merge_base_sha) "
            "WHERE round_no = ? AND status IN ('open', 'closing')",
            (staging_branch, merge_base_sha, round_no),
        )
        merge_order = _optional_int(payload.get("merge_order"))
        if merge_order is None:
            merge_order = self._next_round_merge_order(conn, round_no)
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO round_plans(round_no, plan_id, merge_order, base_sha, "
            "merge_commit_sha) VALUES (?, ?, ?, ?, ?) "
            "ON CONFLICT(round_no, plan_id) DO UPDATE SET "
            "base_sha = COALESCE(excluded.base_sha, round_plans.base_sha), "
            "merge_commit_sha = COALESCE(excluded.merge_commit_sha, "
            "round_plans.merge_commit_sha)",
            (
                round_no,
                plan_id,
                merge_order,
                _optional_str(payload.get("base_sha")),
                _optional_str(payload.get("merge_commit_sha")),
            ),
        )
 
    def _project_round_test(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
    ) -> None:
        round_no = int(payload["round_no"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE rounds SET "
            "test_count_before = COALESCE(?, test_count_before), "
            "test_count_after = COALESCE(?, test_count_after) "
            "WHERE round_no = ?",
            (
                _optional_int(payload.get("test_count_before")),
                _optional_int(payload.get("test_count_after")),
                round_no,
            ),
        )
        for entry in _mapping_list(payload.get("plan_results")):
            conn.execute(
                "UPDATE round_plans SET test_result = ? "
                "WHERE round_no = ? AND plan_id = ?",
                (
                    _optional_str(entry.get("test_result")),
                    round_no,
                    str(entry["plan_id"]),
                ),
            )
 
    def _project_round_lint(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
    ) -> None:
        # Lint is audit-only at the rounds projection level; the operation_log
        # row carries errors/warnings counts. No rounds-table column to update.
        _ = (conn, payload)
 
    def _project_round_consolidate(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
    ) -> None:
        round_no = int(payload["round_no"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE rounds SET status = 'committed', final_commit_sha = ? "
            "WHERE round_no = ? AND status IN ('open', 'closing', 'committed')",
            (_optional_str(payload.get("final_commit_sha")), round_no),
        )
 
    def _project_round_ff_main(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
    ) -> None:
        # Fast-forwarding main is a git-only effect; nothing to project.
        _ = (conn, payload)
 
    def _project_round_cleanup(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        round_no = int(payload["round_no"])
        outcome = _optional_str(payload.get("outcome")) or "clean"
        if outcome == "clean":
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE rounds SET cleanup_status = 'clean', status = 'closed', "
                "closed_at = ?, "
                "roadmap_version = COALESCE(?, roadmap_version), "
                "app_version = COALESCE(?, app_version) "
                "WHERE round_no = ? AND status IN ('committed', 'closed')",
                (
                    ts,
                    _optional_str(payload.get("roadmap_version")),
                    _optional_str(payload.get("app_version")),
                    round_no,
                ),
            )
        else:
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE rounds SET cleanup_status = 'failed' WHERE round_no = ?",
                (round_no,),
            )
 
    def _project_round_abort(
        self,
        conn: sqlite3.Connection,
        payload: Mapping[str, object],
        ts: str,
    ) -> None:
        round_no = int(payload["round_no"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "UPDATE rounds SET status = 'aborted', closed_at = ? "
            "WHERE round_no = ? AND status NOT IN ('closed')",
            (ts, round_no),
        )
 
    # -- M11.7 round-job tracking seam (not operation_log audit rows) ---------
 
    def create_round_job(
        self,
        *,
        job_id: str,
        round_no: int,
        created_by: str,
        staging_branch_head_sha: str | None = None,
        deadline_at: str | None = None,
    ) -> None:
        """Insert a new resumable round-close job row."""
 
        now = self.clock.now_wall()
        with self.transaction("immediate") as conn:
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "INSERT INTO round_jobs(job_id, round_no, status, current_step, "
                "created_by, deadline_at, staging_branch_head_sha, created_at, "
                "updated_at) VALUES (?, ?, 'running', NULL, ?, ?, ?, ?, ?)",
                (
                    job_id,
                    round_no,
                    created_by,
                    deadline_at,
                    staging_branch_head_sha,
                    now,
                    now,
                ),
            )
 
    def get_round_job(self, job_id: str) -> dict[str, object] | None:
        """Return a round-job row by id, or None."""
 
        with self.connect() as conn:
            row = conn.execute(
                "SELECT * FROM round_jobs WHERE job_id = ?",
                (job_id,),
            ).fetchone()
        return None if row is None else _row_dict(row)
 
    def latest_round_job(
        self,
        round_no: int,
        *,
        statuses: tuple[str, ...] | None = None,
    ) -> dict[str, object] | None:
        """Return the most-recent round-job for a round, optional status filter."""
 
        query = "SELECT * FROM round_jobs WHERE round_no = ?"
        params: list[object] = [round_no]
        if statuses:
            placeholders = ", ".join("?" for _ in statuses)
            query += f" AND status IN ({placeholders})"
            params.extend(statuses)
        query += " ORDER BY created_at DESC, rowid DESC LIMIT 1"
        with self.connect() as conn:
            row = conn.execute(query, tuple(params)).fetchone()
        return None if row is None else _row_dict(row)
 
    def set_round_job_status(
        self,
        job_id: str,
        *,
        status: str,
        current_step: str | None = None,
        update_step: bool = True,
    ) -> None:
        """Update a round-job's status (and optionally current_step)."""
 
        now = self.clock.now_wall()
        with self.transaction("immediate") as conn:
            if update_step:
                conn.execute(
                    "UPDATE round_jobs SET status = ?, current_step = ?, "
                    "updated_at = ? WHERE job_id = ?",
                    (status, current_step, now, job_id),
                )
            else:
                conn.execute(
                    "UPDATE round_jobs SET status = ?, updated_at = ? "
                    "WHERE job_id = ?",
                    (status, now, job_id),
                )
 
    def set_round_job_head_sha(self, job_id: str, sha: str | None) -> None:
        """Record the staging-branch HEAD snapshot for a round-job."""
 
        now = self.clock.now_wall()
        with self.transaction("immediate") as conn:
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "UPDATE round_jobs SET staging_branch_head_sha = ?, "
                "updated_at = ? WHERE job_id = ?",
                (sha, now, job_id),
            )
 
    def upsert_round_job_step(
        self,
        *,
        job_id: str,
        step_no: int,
        step_name: str,
        status: str,
        input_hash: str | None = None,
        post_step_head_sha: str | None = None,
        result_summary: str | None = None,
        error_summary: str | None = None,
        artifact_paths_json: str | None = None,
        started_at: str | None = None,
        finished_at: str | None = None,
    ) -> None:
        """Insert or update a single round-job step row."""
 
        with self.transaction("immediate") as conn:
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "INSERT INTO round_job_steps(job_id, step_no, step_name, status, "
                "input_hash, post_step_head_sha, started_at, finished_at, "
                "result_summary, error_summary, artifact_paths_json) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "
                "ON CONFLICT(job_id, step_no) DO UPDATE SET "
                "step_name = excluded.step_name, status = excluded.status, "
                "input_hash = COALESCE(excluded.input_hash, "
                "round_job_steps.input_hash), "
                "post_step_head_sha = COALESCE(excluded.post_step_head_sha, "
                "round_job_steps.post_step_head_sha), "
                "started_at = COALESCE(excluded.started_at, "
                "round_job_steps.started_at), "
                "finished_at = COALESCE(excluded.finished_at, "
                "round_job_steps.finished_at), "
                "result_summary = COALESCE(excluded.result_summary, "
                "round_job_steps.result_summary), "
                "error_summary = excluded.error_summary, "
                "artifact_paths_json = COALESCE(excluded.artifact_paths_json, "
                "round_job_steps.artifact_paths_json)",
                (
                    job_id,
                    step_no,
                    step_name,
                    status,
                    input_hash,
                    post_step_head_sha,
                    started_at,
                    finished_at,
                    result_summary,
                    error_summary,
                    artifact_paths_json,
                ),
            )
 
    def list_round_job_steps(self, job_id: str) -> list[dict[str, object]]:
        """Return all step rows for a round-job ordered by step number."""
 
        with self.connect() as conn:
            rows = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT * FROM round_job_steps WHERE job_id = ? ORDER BY step_no",
                (job_id,),
            ).fetchall()
        return [_row_dict(row) for row in rows]
 
    def get_round(self, round_no: int) -> dict[str, object] | None:
        """Return a round row by number, or None."""
 
        with self.connect() as conn:
            row = conn.execute(
                "SELECT * FROM rounds WHERE round_no = ?",
                (round_no,),
            ).fetchone()
        return None if row is None else _row_dict(row)
 
    def list_round_plans(self, round_no: int) -> list[dict[str, object]]:
        """Return round_plans rows for a round ordered by merge_order."""
 
        with self.connect() as conn:
            rows = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT * FROM round_plans WHERE round_no = ? ORDER BY merge_order",
                (round_no,),
            ).fetchall()
        return [_row_dict(row) for row in rows]
 
    def get_round_plan(
        self,
        round_no: int,
        plan_id: str,
    ) -> dict[str, object] | None:
        """Return a single round_plans row, or None."""
 
        with self.connect() as conn:
            row = conn.execute(
                "SELECT * FROM round_plans WHERE round_no = ? AND plan_id = ?",
                (round_no, plan_id),
            ).fetchone()
        return None if row is None else _row_dict(row)
 
    def _upsert_migration_round(
        self,
        conn: sqlite3.Connection,
        round_row: Mapping[str, object],
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO rounds("
            "round_no, status, cleanup_status, orchestrator_slot, opened_at, "
            "closed_at, final_commit_sha"
            ") VALUES (?, ?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(round_no) DO UPDATE SET "
            "status = excluded.status, "
            "cleanup_status = excluded.cleanup_status, "
            "orchestrator_slot = excluded.orchestrator_slot, "
            "opened_at = excluded.opened_at, "
            "closed_at = excluded.closed_at, "
            "final_commit_sha = excluded.final_commit_sha",
            (
                int(round_row["round_no"]),
                str(round_row["status"]),
                str(round_row["cleanup_status"]),
                str(round_row["orchestrator_slot"]),
                str(round_row["opened_at"]),
                _optional_str(round_row.get("closed_at")),
                _optional_str(round_row.get("final_commit_sha")),
            ),
        )
 
    def _upsert_migration_plan(
        self,
        conn: sqlite3.Connection,
        plan: Mapping[str, object],
    ) -> None:
        now = self.clock.now_wall()
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO plans("
            "plan_id, plan_number, title, plan_type, status, assigned_slot, "
            "queue_position, branch, worktree, body_path, body_sha256, "
            "body_observed_at, model_override, thinking_override, "
            "session_override, created_at, updated_at, ready_for_merge_at, "
            "completed_at, closed_round"
            ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(plan_id) DO UPDATE SET "
            "plan_number = excluded.plan_number, "
            "title = excluded.title, "
            "plan_type = excluded.plan_type, "
            "status = excluded.status, "
            "assigned_slot = excluded.assigned_slot, "
            "queue_position = excluded.queue_position, "
            "branch = excluded.branch, "
            "worktree = excluded.worktree, "
            "body_path = excluded.body_path, "
            "body_sha256 = excluded.body_sha256, "
            "body_observed_at = excluded.body_observed_at, "
            "model_override = excluded.model_override, "
            "thinking_override = excluded.thinking_override, "
            "session_override = excluded.session_override, "
            "updated_at = excluded.updated_at, "
            "ready_for_merge_at = excluded.ready_for_merge_at, "
            "completed_at = excluded.completed_at, "
            "closed_round = excluded.closed_round",
            (
                str(plan["plan_id"]),
                int(plan["plan_number"]),
                str(plan["title"]),
                _optional_str(plan.get("plan_type")),
                str(plan["status"]),
                _optional_str(plan.get("assigned_slot")),
                _optional_int(plan.get("queue_position")),
                _optional_str(plan.get("branch")),
                _optional_str(plan.get("worktree")),
                str(plan["body_path"]),
                _optional_str(plan.get("body_sha256")),
                _optional_str(plan.get("body_observed_at")),
                _optional_str(plan.get("model_override")),
                _optional_str(plan.get("thinking_override")),
                _optional_str(plan.get("session_override")),
                now,
                now,
                _optional_str(plan.get("ready_for_merge_at")),
                _optional_str(plan.get("completed_at")),
                _optional_int(plan.get("closed_round")),
            ),
        )
 
    def _insert_migration_status_event(
        self,
        conn: sqlite3.Connection,
        plan: Mapping[str, object],
        actor_slot: str,
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO plan_status_events("
            "plan_id, from_status, to_status, actor_slot, source, created_at, note"
            ") VALUES (?, NULL, ?, ?, 'migrate', ?, ?)",
            (
                str(plan["plan_id"]),
                str(plan["status"]),
                actor_slot,
                self.clock.now_wall(),
                _optional_str(plan.get("legacy_status")),
            ),
        )
 
    def _upsert_migration_round_plan(
        self,
        conn: sqlite3.Connection,
        round_plan: Mapping[str, object],
    ) -> None:
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO round_plans("
            "round_no, plan_id, merge_order, merge_commit_sha, test_result"
            ") VALUES (?, ?, ?, ?, ?) "
            "ON CONFLICT(round_no, plan_id) DO UPDATE SET "
            "merge_order = excluded.merge_order, "
            "merge_commit_sha = excluded.merge_commit_sha, "
            "test_result = excluded.test_result",
            (
                int(round_plan["round_no"]),
                str(round_plan["plan_id"]),
                int(round_plan["merge_order"]),
                _optional_str(round_plan.get("merge_commit_sha")),
                _optional_str(round_plan.get("test_result")),
            ),
        )
 
    def append_operation_log(self, entry: Mapping[str, object]) -> object:
        raise NotImplementedError("operation-log writing is implemented in plan 082")
 
    def acquire_lock(
        self,
        name: str,
        owner: str,
        ttl_seconds: int,
        metadata: Mapping[str, object] | None = None,
    ) -> LockHandle:
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be positive")
        self.startup_sweep()
        acquired_at = self.clock.now_wall()
        expires_at = wall_plus_seconds(acquired_at, ttl_seconds)
        metadata_json = None if metadata is None else canonical_json(metadata)
        try:
            with self.transaction("immediate") as conn:
                conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "INSERT INTO locks("
                    "name, owner_slot, acquired_at, expires_at, metadata_json"
                    ") VALUES (?, ?, ?, ?, ?)",
                    (name, owner, acquired_at, expires_at, metadata_json),
                )
        except sqlite3.IntegrityError as exc:
            holder = self._lock_holder(name)
            if holder is None:
                raise
            raise LockHeldError(
                name=name,
                owner_slot=holder["owner_slot"],
                expires_at=holder["expires_at"],
            ) from exc
        return LockHandle(
            name=name,
            owner_slot=owner,
            expires_at=expires_at,
            _release=self.release_lock,
            _renew=self.renew_lock,
        )
 
    def release_lock(self, name: str, owner: str) -> None:
        with self.transaction("immediate") as conn:
            row = conn.execute(
                "SELECT owner_slot FROM locks WHERE name = ?",
                (name,),
            ).fetchone()
            if row is None:
                return
            actual_owner = str(row["owner_slot"])
            if actual_owner != owner:
                raise LockOwnerError(
                    name=name,
                    owner_slot=owner,
                    actual_owner_slot=actual_owner,
                )
            conn.execute("DELETE FROM locks WHERE name = ?", (name,))
 
    def renew_lock(self, name: str, owner: str, ttl_seconds: int) -> str | None:
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be positive")
        expires_at = wall_plus_seconds(self.clock.now_wall(), ttl_seconds)
        with self.transaction("immediate") as conn:
            row = conn.execute(
                "SELECT owner_slot FROM locks WHERE name = ?",
                (name,),
            ).fetchone()
            if row is None:
                raise LockOwnerError(
                    name=name,
                    owner_slot=owner,
                    actual_owner_slot=None,
                )
            actual_owner = str(row["owner_slot"])
            if actual_owner != owner:
                raise LockOwnerError(
                    name=name,
                    owner_slot=owner,
                    actual_owner_slot=actual_owner,
                )
            conn.execute(
                "UPDATE locks SET expires_at = ? WHERE name = ?",
                (expires_at, name),
            )
        return expires_at
 
    def startup_sweep(self) -> int:
        return len(self.startup_sweep_with_names())
 
    def startup_sweep_with_names(self) -> list[str]:
        """Remove expired lock rows and return their names."""
 
        with self.transaction("immediate") as conn:
            now = self.clock.now_wall()
            names = [
                str(row["name"])
                for row in conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "SELECT name FROM locks "
                    "WHERE expires_at IS NOT NULL AND expires_at < ? "
                    "ORDER BY name",
                    (now,),
                ).fetchall()
            ]
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "DELETE FROM locks WHERE expires_at IS NOT NULL AND expires_at < ?",
                (now,),
            )
            return names
 
    def active_operator_slot(self) -> str | None:
        """Return the first active human operator slot, if one exists."""
 
        with self.connect() as conn:
            row = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT slot FROM agents "
                "WHERE slot_kind = 'operator' AND active = 1 "
                "ORDER BY slot LIMIT 1"
            ).fetchone()
        return None if row is None else str(row["slot"])
 
    def actor_slot_kind(self, actor_slot: str) -> str:
        """Return an actor's slot kind or raise ``UnknownActorError``."""
 
        with self.connect() as conn:
            return self._actor_slot_kind(conn, actor_slot)
 
    def credential_fingerprints(self) -> list[dict[str, object]]:
        """Return credential rows needed for doctor fingerprint checks."""
 
        with self.connect() as conn:
            rows = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT hmac_key_id, slot, secret_fingerprint, rotation_phase, "
                "revoked_at FROM agent_credentials ORDER BY slot, rotation_phase"
            ).fetchall()
        return [
            {
                "hmac_key_id": str(row["hmac_key_id"]),
                "slot": str(row["slot"]),
                "secret_fingerprint": str(row["secret_fingerprint"]),
                "rotation_phase": str(row["rotation_phase"]),
                "revoked_at": row["revoked_at"],
            }
            for row in rows
        ]
 
    def backup(self, target: str | Path) -> Path:
        target_path = Path(target)
        target_path.parent.mkdir(parents=True, exist_ok=True)
        with self.connect() as source, sqlite3.connect(str(target_path)) as dest:
            source.backup(dest)
        return target_path
 
    def restore(self, source: str | Path) -> None:
        if self._is_memory:
            raise BackendCapabilityError("cannot restore over an in-memory database")
        source_path = Path(source)
        if not source_path.is_file():
            raise FileNotFoundError(source_path)
        self._checkpoint_for_restore()
        self._delete_sqlite_sidecars()
        shutil.copy2(source_path, self.path)
        self._delete_sqlite_sidecars()
 
    def integrity_check(self) -> dict[str, object]:
        errors: list[dict[str, object]] = []
        try:
            # conn closed explicitly: `with conn` alone never closes and the
            # leaked handle holds a WAL lock (breaks regenerate's swap).
            conn = self.connect()
            try:
                with conn:
                    integrity_rows = [
                        row[0] for row in conn.execute("PRAGMA integrity_check").fetchall()
                    ]
                    if integrity_rows != ["ok"]:
                        errors.extend(
                            {
                                "kind": "physical_corruption",
                                "details": str(row),
                            }
                            for row in integrity_rows
                        )
                    fk_rows = conn.execute("PRAGMA foreign_key_check").fetchall()
                    errors.extend(_foreign_key_error(row) for row in fk_rows)
                    self._check_append_only_triggers(conn, errors)
            finally:
                conn.close()
        except sqlite3.Error as exc:
            errors.append({"kind": "physical_corruption", "details": str(exc)})
        return {"ok": not errors, "errors": errors}
 
    def vacuum(self) -> None:
        with self.connect() as conn:
            conn.execute("VACUUM")
 
    def connection_pragmas(self) -> dict[str, object]:
        """Return the effective PRAGMA values for a fresh backend connection."""
 
        # Closed explicitly: `with conn` never closes, and this runs in every
        # backend construction (capability assert) — the leaked handle held a
        # WAL lock that broke regenerate's os.replace swap under coverage.
        conn = self.connect()
        try:
            return self._read_pragmas(conn)
        finally:
            conn.close()
 
    def _sqlite_sidecars(self) -> tuple[Path, Path]:
        return (
            self.path.with_name(f"{self.path.name}-wal"),
            self.path.with_name(f"{self.path.name}-shm"),
        )
 
    def _checkpoint_for_restore(self) -> None:
        try:
            conn = self.connect()
        except sqlite3.Error:
            return
        try:
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchall()
        except sqlite3.Error:
            pass
        finally:
            conn.close()
 
    def _delete_sqlite_sidecars(self) -> None:
        for sidecar in self._sqlite_sidecars():
            try:
                sidecar.unlink(missing_ok=True)
            except PermissionError:
                # A just-closed SQLite handle can retain the sidecar briefly on
                # Windows. The checkpoint above makes the sidecar non-authoritative.
                pass
 
    def _assert_capabilities(self) -> None:
        if not self._is_memory:
            pragmas = self.connection_pragmas()
            if pragmas["foreign_keys"] != 1:
                raise BackendCapabilityError("SQLite foreign_keys pragma is not ON")
            if str(pragmas["journal_mode"]).lower() != "wal":
                raise BackendCapabilityError("SQLite journal_mode pragma is not WAL")
            if int(pragmas["busy_timeout"]) < 5000:
                raise BackendCapabilityError("SQLite busy_timeout is below 5000 ms")
        super()._assert_capabilities()
 
    def _apply_connection_pragmas(self, conn: sqlite3.Connection) -> None:
        conn.execute("PRAGMA foreign_keys = ON")
        conn.execute("PRAGMA journal_mode = WAL")
        conn.execute("PRAGMA busy_timeout = 5000")
 
    def _read_pragmas(self, conn: sqlite3.Connection) -> dict[str, object]:
        return {
            "foreign_keys": conn.execute("PRAGMA foreign_keys").fetchone()[0],
            "journal_mode": conn.execute("PRAGMA journal_mode").fetchone()[0],
            "busy_timeout": conn.execute("PRAGMA busy_timeout").fetchone()[0],
        }
 
    def _run_schema_script(self, conn: sqlite3.Connection, *, commit: bool) -> None:
        schema = canonical_schema_sql_without_pragmas()
        roadmap_schema = v0_3_roadmap.load_schema_sql()
        auth_events_schema = v0_3_1_auth_events.migration_sql()
        source_fields_schema = v0_3_2_roadmap_source_fields.migration_sql()
        suffix = "\nCOMMIT;" if commit else ""
        try:
            conn.executescript(
                "BEGIN IMMEDIATE;\n"
                f"{schema}\n{roadmap_schema}\n{auth_events_schema}\n"
                f"{source_fields_schema}{suffix}"
            )
        except Exception:
            try:
                conn.execute("ROLLBACK")
            except sqlite3.Error:
                pass
            raise
 
    def _seed_initial_rows(self, conn: sqlite3.Connection, seed: RegistrySeed) -> None:
        now = self.clock.now_wall()
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO schema_migrations(version, applied_at, description) "
            "VALUES (?, ?, ?)",
            (SCHEMA_VERSION, now, SCHEMA_DESCRIPTION),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT OR IGNORE INTO schema_migrations(version, applied_at, description) "
            "VALUES (?, ?, ?)",
            (
                v0_3_roadmap.ROADMAP_SCHEMA_VERSION,
                now,
                v0_3_roadmap.ROADMAP_SCHEMA_DESCRIPTION,
            ),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT OR IGNORE INTO schema_migrations(version, applied_at, description) "
            "VALUES (?, ?, ?)",
            (
                v0_3_1_auth_events.AUTH_EVENTS_SCHEMA_VERSION,
                now,
                v0_3_1_auth_events.AUTH_EVENTS_SCHEMA_DESCRIPTION,
            ),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT OR IGNORE INTO schema_migrations(version, applied_at, description) "
            "VALUES (?, ?, ?)",
            (
                v0_3_2_roadmap_source_fields.ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION,
                now,
                v0_3_2_roadmap_source_fields.ROADMAP_SOURCE_FIELDS_SCHEMA_DESCRIPTION,
            ),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO workspaces("
            "workspace_id, workspace_root, git_common_dir, registry_path, "
            "checkpoint_path, created_at, last_seen_at, active"
            ") VALUES (?, ?, ?, ?, ?, ?, ?, 1)",
            (
                seed.workspace_id,
                seed.workspace_root,
                seed.git_common_dir,
                seed.registry_path,
                seed.checkpoint_path,
                now,
                now,
            ),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO registry_meta("
            "id, registry_id, workspace_id, schema_version, state_version, "
            "last_checkpoint_id, created_at, updated_at"
            ") VALUES (1, ?, ?, ?, 0, NULL, ?, ?)",
            (seed.registry_id, seed.workspace_id, SCHEMA_VERSION, now, now),
        )
        # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
        conn.execute("INSERT INTO plan_seq(id, next_number) VALUES (1, 1)")
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO agents("
            "slot, slot_kind, identity_kind, model_class, "
            "orchestrator_eligible, active, created_at"
            ") VALUES (?, 'operator', 'human', 'human', 0, 1, ?)",
            (seed.operator_slot, now),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO orchestration_state("
            "id, orchestrator_slot, handover_status, ready_to_handover, "
            "updated_at, updated_by"
            ") VALUES (1, NULL, 'stable', 0, ?, ?)",
            (now, seed.operator_slot),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO leases("
            "id, holder_slot, lease_expires_at, heartbeat_interval_sec, "
            "last_heartbeat_at"
            ") VALUES (1, NULL, NULL, 30, NULL)"
        )
        conn.executemany(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO settings(key, value, updated_at, updated_by) "
            "VALUES (?, ?, ?, ?)",
            [
                (key, value, now, seed.operator_slot)
                for key, value in _default_settings(seed.app_version)
            ],
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "INSERT INTO agent_credentials("
            "hmac_key_id, slot, secret_fingerprint, created_at, rotation_phase"
            ") VALUES (?, ?, ?, ?, 'active')",
            (
                seed.hmac_key_id,
                seed.operator_slot,
                seed.secret_fingerprint,
                now,
            ),
        )
 
    def _table_count(self, conn: sqlite3.Connection) -> int:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table'"
        ).fetchone()
        return int(row[0])
 
    def _check_append_only_triggers(
        self,
        conn: sqlite3.Connection,
        errors: list[dict[str, object]],
    ) -> None:
        table_row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
            ("operation_log",),
        ).fetchone()
        if table_row is None:
            return
        trigger_rows = conn.execute(
            "SELECT name FROM sqlite_master WHERE type = 'trigger'"
        ).fetchall()
        triggers = {str(row[0]) for row in trigger_rows}
        required = {"operation_log_no_update", "operation_log_no_delete"}
        missing = sorted(required - triggers)
        if missing:
            errors.append(
                {
                    "kind": "append_only_trigger",
                    "details": f"missing append-only triggers: {', '.join(missing)}",
                    "missing": missing,
                }
            )
        roadmap_table = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
            ("roadmap_item_events",),
        ).fetchone()
        if roadmap_table is None:
            return
        roadmap_required = {
            "roadmap_item_events_no_update",
            "roadmap_item_events_no_delete",
        }
        roadmap_missing = sorted(roadmap_required - triggers)
        if roadmap_missing:
            errors.append(
                {
                    "kind": "append_only_trigger",
                    "details": "missing roadmap append-only triggers: "
                    + ", ".join(roadmap_missing),
                    "missing": roadmap_missing,
                }
            )
 
    def _auto_migrate_existing_database(self) -> None:
        """Apply the v0.3 roadmap extension to initialized registries."""
 
        if self._is_memory:
            return
        # conn closed explicitly (leak held a WAL lock); the inner `with conn`
        # keeps the commit-on-success semantics the migration stamps rely on.
        conn = self.connect()
        try:
            self._auto_migrate_in_conn(conn)
        finally:
            conn.close()
 
    def _auto_migrate_in_conn(self, conn: sqlite3.Connection) -> None:
        with conn:
            if self._table_count(conn) == 0:
                return
            row = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
                ("registry_meta",),
            ).fetchone()
            if row is None:
                return
            now = self.clock.now_wall()
            if not (
                v0_3_roadmap.current_user_version(conn)
                >= v0_3_roadmap.ROADMAP_SCHEMA_VERSION
                and v0_3_roadmap.tables_exist(conn)
            ):
                conn.executescript(
                    "BEGIN IMMEDIATE;\n"
                    + v0_3_roadmap.load_schema_sql()
                    + "\nCOMMIT;"
                )
                conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "INSERT OR IGNORE INTO schema_migrations("
                    "version, applied_at, description"
                    ") VALUES (?, ?, ?)",
                    (
                        v0_3_roadmap.ROADMAP_SCHEMA_VERSION,
                        now,
                        v0_3_roadmap.ROADMAP_SCHEMA_DESCRIPTION,
                    ),
                )
            if not v0_3_1_auth_events.is_applied(conn):
                v0_3_1_auth_events.apply(conn)
                conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "INSERT OR IGNORE INTO schema_migrations("
                    "version, applied_at, description"
                    ") VALUES (?, ?, ?)",
                    (
                        v0_3_1_auth_events.AUTH_EVENTS_SCHEMA_VERSION,
                        now,
                        v0_3_1_auth_events.AUTH_EVENTS_SCHEMA_DESCRIPTION,
                    ),
                )
            elif (
                v0_3_1_auth_events.current_user_version(conn)
                < v0_3_1_auth_events.AUTH_EVENTS_SCHEMA_VERSION
            ):
                conn.execute(
                    f"PRAGMA user_version = "
                    f"{v0_3_1_auth_events.AUTH_EVENTS_SCHEMA_VERSION}"
                )
            if not v0_3_2_roadmap_source_fields.is_applied(conn):
                v0_3_2_roadmap_source_fields.apply(conn)
                conn.execute(
                    # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                    "INSERT OR IGNORE INTO schema_migrations("
                    "version, applied_at, description"
                    ") VALUES (?, ?, ?)",
                    (
                        v0_3_2_roadmap_source_fields.ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION,
                        now,
                        v0_3_2_roadmap_source_fields.ROADMAP_SOURCE_FIELDS_SCHEMA_DESCRIPTION,
                    ),
                )
            elif (
                v0_3_1_auth_events.current_user_version(conn)
                < v0_3_2_roadmap_source_fields.ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION
            ):
                conn.execute(
                    f"PRAGMA user_version = "
                    f"{v0_3_2_roadmap_source_fields.ROADMAP_SOURCE_FIELDS_SCHEMA_VERSION}"
                )
            conn.commit()
 
    def _validate_payload(
        self,
        operation_type: str,
        payload: Mapping[str, object],
        payload_version: int,
    ) -> str:
        if not isinstance(operation_type, str) or not operation_type.strip():
            raise PayloadSchemaViolationError(
                operation_type=str(operation_type),
                message="operation_type must be a non-empty string",
            )
        if not isinstance(payload, Mapping):
            raise PayloadSchemaViolationError(
                operation_type=operation_type,
                message="payload must be a mapping",
            )
        if "version" not in payload:
            raise PayloadSchemaViolationError(
                operation_type=operation_type,
                message="payload must include version",
            )
        supplied_version = payload["version"]
        if not isinstance(supplied_version, int) or supplied_version != payload_version:
            raise PayloadVersionMismatchError(
                operation_type=operation_type,
                payload_version=payload_version,
                supplied_version=supplied_version,
            )
        try:
            return canonical_json(payload)
        except (TypeError, ValueError) as exc:
            raise PayloadSchemaViolationError(
                operation_type=operation_type,
                message=str(exc),
            ) from exc
 
    def _actor_slot_kind(self, conn: sqlite3.Connection, actor_slot: str) -> str:
        row = conn.execute(
            "SELECT slot_kind FROM agents WHERE slot = ?",
            (actor_slot,),
        ).fetchone()
        if row is None:
            raise UnknownActorError(actor_slot)
        return str(row["slot_kind"])
 
    def _active_hmac_key_id(
        self,
        conn: sqlite3.Connection,
        actor_slot: str,
    ) -> str | None:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "SELECT hmac_key_id FROM agent_credentials "
            "WHERE slot = ? AND revoked_at IS NULL "
            "ORDER BY CASE rotation_phase "
            "WHEN 'rotating_new' THEN 0 "
            "WHEN 'active' THEN 1 "
            "WHEN 'rotating_old' THEN 2 "
            "ELSE 3 END LIMIT 1",
            (actor_slot,),
        ).fetchone()
        return None if row is None else str(row["hmac_key_id"])
 
    def _assert_handover_allows(
        self,
        conn: sqlite3.Connection,
        operation_type: str,
        actor_slot: str,
        slot_kind: str,
    ) -> None:
        if not operation_type.startswith("round.") or slot_kind != "orchestrator":
            return
        state = self._orchestration_state(conn)
        holder = state["orchestrator_slot"]
        status = str(state["handover_status"])
        outgoing = state["handover_outgoing"]
        incoming = state["handover_incoming"]
        if status == "stable":
            if holder is None:
                conn.execute(
                    "UPDATE orchestration_state SET orchestrator_slot = ?, "
                    "updated_at = ?, updated_by = ? WHERE id = 1",
                    (actor_slot, self.clock.now_wall(), actor_slot),
                )
                return
            if str(holder) != actor_slot:
                raise HandoverStateError(
                    state="handover_in_flight",
                    actor_slot=actor_slot,
                    operation_type=operation_type,
                )
            return
        if outgoing is not None and str(outgoing) == actor_slot:
            return
        state_name = (
            "pending_not_accepted"
            if incoming is not None and str(incoming) == actor_slot
            else "handover_in_flight"
        )
        raise HandoverStateError(
            state=state_name,
            actor_slot=actor_slot,
            operation_type=operation_type,
        )
 
    def _orchestration_state(self, conn: sqlite3.Connection) -> sqlite3.Row:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
            "SELECT orchestrator_slot, handover_status, handover_outgoing, "
            "handover_incoming, handover_requested_at FROM orchestration_state "
            "WHERE id = 1"
        ).fetchone()
        if row is None:
            raise BackendCapabilityError("orchestration_state singleton row is missing")
        return row
 
    def _assert_no_parent_cycle(
        self,
        conn: sqlite3.Connection,
        parent_operation_id: int,
        operation_uuid: str,
    ) -> None:
        seen: set[int] = set()
        current_id: int | None = parent_operation_id
        depth = 0
        while current_id is not None:
            if current_id in seen or depth >= _MAX_PARENT_DEPTH:
                raise ParentCycleError(parent_operation_id, operation_uuid)
            seen.add(current_id)
            row = conn.execute(
                "SELECT operation_uuid, parent_operation_id "
                "FROM operation_log WHERE id = ?",
                (current_id,),
            ).fetchone()
            if row is None:
                return
            if row["operation_uuid"] == operation_uuid:
                raise ParentCycleError(parent_operation_id, operation_uuid)
            current_id = row["parent_operation_id"]
            depth += 1
 
    def _current_state_version(self, conn: sqlite3.Connection) -> int:
        row = conn.execute(
            "SELECT state_version FROM registry_meta WHERE id = 1"
        ).fetchone()
        if row is None:
            raise BackendCapabilityError("registry_meta singleton row is missing")
        return int(row["state_version"])
 
    def _lock_holder(self, name: str) -> sqlite3.Row | None:
        with self.connect() as conn:
            return conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential-related column names are schema fields, not embedded secrets"
                "SELECT owner_slot, expires_at FROM locks WHERE name = ?",
                (name,),
            ).fetchone()
 
 
def _mapping_list(value: object) -> list[Mapping[str, object]]:
    if value is None:
        return []
    if not isinstance(value, list):
        raise TypeError("migration projection sections must be lists")
    rows: list[Mapping[str, object]] = []
    for item in value:
        if not isinstance(item, Mapping):
            raise TypeError("migration projection rows must be mappings")
        rows.append(item)
    return rows
 
 
def _required_mapping(row: Mapping[str, object], key: str) -> Mapping[str, object]:
    value = row.get(key)
    if not isinstance(value, Mapping):
        raise TypeError(f"migration projection row missing mapping: {key}")
    return value
 
 
def _row_dict(row: sqlite3.Row) -> dict[str, object]:
    return {key: row[key] for key in row.keys()}
 
 
def _optional_str(value: object) -> str | None:
    if value is None:
        return None
    text = str(value)
    return text if text else None
 
 
def _optional_int(value: object) -> int | None:
    return None if value is None else int(value)
 
 
def _bool_int(value: object) -> int:
    return 1 if bool(value) else 0
 
 
def _phase_code_for_item_id(item_id: str) -> str | None:
    if not item_id.startswith("M"):
        return None
    if "." not in item_id:
        return item_id
    return item_id.split(".", 1)[0]
 
 
def _parent_phase_code(phase_code: str) -> str | None:
    if "." not in phase_code:
        return None
    return phase_code.split(".", 1)[0]
 
 
def _phase_sort_order(phase_code: str) -> int:
    digits = "".join(ch if ch.isdigit() else " " for ch in phase_code).split()
    if not digits:
        return 999000
    value = int(digits[0]) * 1000
    if len(digits) > 1:
        value += int(digits[1])
    return value
 
 
def _era_id_for_version(version_id: str) -> str:
    major = version_id.removeprefix("v").split(".", 1)[0]
    return f"v{major}"
 
 
def _era_sort_order(era_id: str) -> int:
    text = era_id.removeprefix("v")
    return int(text) if text.isdigit() else 999
 
 
def canonical_schema_path() -> Path:
    return (
        Path(__file__).resolve().parents[2]
        / "docs"
        / "architecture"
        / "orchestration"
        / "REGISTRY_SCHEMA_V0_2_3.sql"
    )
 
 
def load_canonical_schema_sql() -> str:
    return canonical_schema_path().read_text(encoding="utf-8")
 
 
def canonical_schema_sql_without_pragmas() -> str:
    lines = []
    for line in load_canonical_schema_sql().splitlines():
        if line.lstrip().upper().startswith(_PRAGMA_PREFIX):
            continue
        lines.append(line)
    return "\n".join(lines) + "\n"
 
 
def _foreign_key_error(row: sqlite3.Row) -> dict[str, object]:
    values = tuple(row)
    table = "" if len(values) < 1 else str(values[0])
    row_id = None if len(values) < 2 else values[1]
    parent = "" if len(values) < 3 else str(values[2])
    fkid = None if len(values) < 4 else values[3]
    return {
        "kind": "fk_violation",
        "table": table,
        "row": row_id,
        "parent": parent,
        "fkid": fkid,
        "details": repr(values),
    }
 
 
def _default_settings(app_version: str) -> list[tuple[str, str]]:
    return [
        ("workflow_version", "1"),
        ("renderer_version", "1"),
        ("roadmap_version", "1"),
        ("app_version", app_version),
        ("hmac_max_skew_seconds", "300"),
        ("hmac_rotation_grace_seconds", "86400"),
        ("round_close_timeout_seconds", "3600"),
        ("round_auto_batch_limit", "5"),
        ("handover_drain_timeout_seconds", "300"),
        ("grant_default_ttl_seconds", "3600"),
        ("max_replay_rows_for_merge", "10000"),
        ("oplog_retention_days", "90"),
        ("auth_events_retention_days", "30"),
        ("auth_failure_alert_threshold_per_hour", "10"),
    ]