refactor: enforce SQLite persistence boundaries

docs/operations.md

@@ -642,6 +642,10 @@ gab --db ~/.local/state/github-agent-bridge/bridge.sqlite3 \
   backfill-trigger-actors
 ```
 
+The write pass requires the current migrated schema and never mutates the
+schema itself. Run `gab migrate-db` first when the command reports a pending
+migration. `--dry-run` can inspect a legacy database without changing it.
+
 ### Detect install drift
 
 Set `GITHUB_AGENT_BRIDGE_RELEASE_REPO` in the systemd environment file to the

docs/persistence.md

@@ -30,6 +30,51 @@ Backup and restore are deliberately excluded from repositories. The specialized
 SQLite's backup API and are shared by autoupdate and the explicit migration
 command.
 
+## Adding or changing persisted data
+
+Keep schema, write-model and read-model changes separate and explicit:
+
+1. Add every schema change as the next immutable module under
+   `sql/migrations/`, and update `sql/schema.sql` so a fresh database and an
+   upgraded database converge on the same schema. Test both paths. Application
+   services and maintenance commands must never add columns or indexes as a
+   compatibility side effect.
+2. Add write and domain lookup methods under `persistence/`. Return typed DTOs
+   rather than leaking `sqlite3.Row`; use `Database.read_only()` for reads and
+   an explicit `Database.transaction()` mode for writes. A service may own a
+   cross-repository transaction and pass its connection into repository
+   methods when one invariant spans several tables.
+3. Add denormalized UI/operational queries to `DashboardQueries`, not to HTTP
+   handlers, CLI commands or write repositories. Accept typed, allowlisted
+   filters and bind all values. Keep hot endpoints to one query where practical.
+4. Exercise repository behavior against a real temporary SQLite database. For
+   hot queries, pin the intended indexes with `EXPLAIN QUERY PLAN`; do not mock
+   SQL strings as a substitute for schema-level coverage.
+
+`tests/test_persistence_architecture.py` enforces these ownership boundaries.
+Direct `sqlite3.connect()` is reserved for `persistence/database.py`, including
+the specialized backup/restore functions. SQL execution is restricted to
+repositories, migrations and the dashboard read model.
+
+## Retention and cleanup ownership
+
+Retention belongs to the repository that writes each dataset; HTTP handlers
+and read models do not delete data as a side effect of reads.
+
+| Data | Owner | Lifecycle |
+| --- | --- | --- |
+| Webhook receipt payloads | `WebhookRepository` | Pruned on receipt ingestion using the configured webhook retention window. |
+| Process samples | `ObservabilityRepository` | Pruned when the monitor records a sample, using the configured sample-retention window. |
+| Job runs, acknowledgements, commit statuses, coalesced notifications, session events and progress | `JobRepository` via the `jobs` parent | Deleted by foreign-key cascade when a future explicit job-retention operation deletes the parent job. Runs are never pruned independently because that would corrupt runtime totals. |
+| Worklog | Future explicit job-retention operation | The legacy table has no job foreign key. Job cleanup must delete matching `job_id` rows in the same transaction before deleting the job; ordinary reads must not prune it. |
+| Ingest receipts and canonical GitHub events | Ingestion/audit retention | They detach from a deleted job with `ON DELETE SET NULL` and remain as deduplication/audit records until a separately configured ingestion-retention policy exists. |
+| Feedback events, proposals and learned rules | Feedback operator workflow | Kept until an explicit operator action removes or supersedes them; job cleanup does not own learned policy data. |
+
+There is intentionally no automatic completed-job retention policy yet. When
+one is added, it must be an explicit repository/CLI operation with a documented
+window, one transaction for parent and non-FK children, and tests proving that
+active jobs and retained audit records are untouched.
+
 ## Dashboard read model
 
 `github_agent_bridge.dashboard_data.DashboardQueries` is the read-only boundary

src/github_agent_bridge/actors.py

@@ -3,14 +3,13 @@
 import json
 import os
 import re
-import sqlite3
 import subprocess
-from dataclasses import dataclass
 from email.utils import parseaddr
 from pathlib import Path
 from typing import Any
 
-from .models import GitHubContext, Notification
+from .models import GitHubContext, Notification, TriggerActor
+from .persistence import ActorBackfillRepository, ActorBackfillUpdate, Database
 
 LOGIN_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?(?:\[bot\])?$")
 ISSUE_EVENT_URL_RE = re.compile(
@@ -23,13 +22,6 @@ def default_gh_bin() -> str:
     return os.getenv("GITHUB_AGENT_BRIDGE_GH_BIN", "gh")
 
 
-@dataclass(frozen=True)
-class TriggerActor:
-    login: str
-    avatar_url: str | None = None
-    user_id: int | None = None
-
-
 def normalize_github_login(value: str | None) -> str | None:
     if not value:
         return None
@@ -150,87 +142,53 @@ def backfill_trigger_actors(db: str | Path, *, gh_bin: str | None = None, limit:
     path = Path(db).expanduser()
     if not path.exists():
         return {"db_exists": False, "checked": 0, "updated": 0, "missing": 0, "dry_run": dry_run}
-    con = sqlite3.connect(path)
-    con.row_factory = sqlite3.Row
-    try:
-        jobs_table = con.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='jobs'").fetchone()
-        if jobs_table is None:
-            return {"db_exists": True, "checked": 0, "updated": 0, "missing": 0, "dry_run": dry_run, "updates": []}
-        has_actor_column = _has_column(con, "jobs", "trigger_actor")
-        has_avatar_column = _has_column(con, "jobs", "trigger_actor_avatar_url")
-        if not dry_run:
-            _ensure_trigger_actor_columns(con)
-            has_actor_column = has_avatar_column = True
-        conditions = []
-        if has_actor_column:
-            conditions.append("(trigger_actor IS NULL OR trigger_actor='')")
-        if has_avatar_column:
-            conditions.append("(trigger_actor_avatar_url IS NULL OR trigger_actor_avatar_url='')")
-        where = f"WHERE {' OR '.join(conditions)}" if conditions else ""
-        select_columns = "id, context_json"
-        if has_actor_column:
-            select_columns += ", trigger_actor"
-        if has_avatar_column:
-            select_columns += ", trigger_actor_avatar_url"
-        rows = con.execute(
-            f"""
-            SELECT {select_columns}
-            FROM jobs
-            {where}
-            ORDER BY id
-            LIMIT ?
-            """,
-            (max(1, limit or 1000000),),
-        ).fetchall()
-        checked = updated = missing = 0
-        updates: list[dict[str, Any]] = []
-        for row in rows:
-            checked += 1
-            try:
-                ctx = GitHubContext.from_json(row["context_json"])
-            except (TypeError, json.JSONDecodeError):
-                missing += 1
-                continue
-            existing_actor = normalize_github_login(row["trigger_actor"]) if has_actor_column else None
-            actor = (
-                TriggerActor(login=existing_actor, avatar_url=github_avatar_url(existing_actor))
-                if existing_actor and (not has_avatar_column or not row["trigger_actor_avatar_url"])
-                else github_actor_details_for_context(ctx, gh_bin=gh_bin)
+    repository = ActorBackfillRepository(Database(path))
+    if not dry_run:
+        repository.validate_current_schema()
+    rows = repository.list_candidates(limit)
+    checked = updated = missing = 0
+    persisted_updates: list[ActorBackfillUpdate] = []
+    updates: list[dict[str, Any]] = []
+    for row in rows:
+        checked += 1
+        try:
+            ctx = GitHubContext.from_json(row.context_json)
+        except (TypeError, json.JSONDecodeError):
+            missing += 1
+            continue
+        existing_actor = normalize_github_login(row.trigger_actor)
+        actor = (
+            TriggerActor(
+                login=existing_actor,
+                avatar_url=github_avatar_url(existing_actor),
             )
-            if not actor:
-                missing += 1
-                continue
-            updates.append({"job_id": int(row["id"]), "trigger_actor": actor.login, "trigger_actor_avatar_url": actor.avatar_url})
-            updated += 1
-            if not dry_run:
-                con.execute(
-                    "UPDATE jobs SET trigger_actor=?, trigger_actor_avatar_url=? WHERE id=?",
-                    (actor.login, actor.avatar_url, row["id"]),
-                )
-        if not dry_run:
-            con.commit()
-        return {"db_exists": True, "checked": checked, "updated": updated, "missing": missing, "dry_run": dry_run, "updates": updates}
-    finally:
-        con.close()
-
-
-def _has_column(con: sqlite3.Connection, table: str, column: str) -> bool:
-    exists = con.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone()
-    if exists is None:
-        return False
-    return column in {row["name"] for row in con.execute(f"PRAGMA table_info({table})")}
-
-
-def _ensure_trigger_actor_columns(con: sqlite3.Connection) -> None:
-    tables = {
-        "jobs": {"trigger_actor": "TEXT", "trigger_actor_avatar_url": "TEXT"},
-        "coalesced_notifications": {"trigger_actor": "TEXT", "trigger_actor_avatar_url": "TEXT"},
-    }
-    for table, columns in tables.items():
-        exists = con.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone()
-        if exists is None:
+            if existing_actor and not row.trigger_actor_avatar_url
+            else github_actor_details_for_context(ctx, gh_bin=gh_bin)
+        )
+        if not actor:
+            missing += 1
             continue
-        existing = {row["name"] for row in con.execute(f"PRAGMA table_info({table})")}
-        for column, definition in columns.items():
-            if column not in existing:
-                con.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}")
+        update = ActorBackfillUpdate(
+            job_id=row.job_id,
+            trigger_actor=actor.login,
+            trigger_actor_avatar_url=actor.avatar_url,
+        )
+        persisted_updates.append(update)
+        updates.append(
+            {
+                "job_id": update.job_id,
+                "trigger_actor": update.trigger_actor,
+                "trigger_actor_avatar_url": update.trigger_actor_avatar_url,
+            }
+        )
+        updated += 1
+    if not dry_run:
+        repository.apply_updates(persisted_updates)
+    return {
+        "db_exists": True,
+        "checked": checked,
+        "updated": updated,
+        "missing": missing,
+        "dry_run": dry_run,
+        "updates": updates,
+    }

src/github_agent_bridge/models.py

@@ -10,6 +10,13 @@ def utc_now() -> str:
     return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
 
 
+@dataclass(frozen=True)
+class TriggerActor:
+    login: str
+    avatar_url: str | None = None
+    user_id: int | None = None
+
+
 @dataclass(frozen=True)
 class GitHubContext:
     urls: list[str]

src/github_agent_bridge/persistence/__init__.py

@@ -1,6 +1,11 @@
 """Explicit SQLite connection and transaction boundaries."""
 
 from .acknowledgements import AcknowledgementClaim, AcknowledgementRepository
+from .actors import (
+    ActorBackfillCandidate,
+    ActorBackfillRepository,
+    ActorBackfillUpdate,
+)
 from .commit_statuses import CommitStatusClaim, CommitStatusRepository
 from .database import (
     ClosingConnection,
@@ -28,6 +33,9 @@
     "AcknowledgementClaim",
     "AcknowledgementRepository",
     "ACTIVE_JOB_STATUSES",
+    "ActorBackfillCandidate",
+    "ActorBackfillRepository",
+    "ActorBackfillUpdate",
     "CommitStatusClaim",
     "CommitStatusRepository",
     "ClosingConnection",

src/github_agent_bridge/persistence/actors.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ..sql.migrations import validate_migrations
+from .database import Database
+
+
+@dataclass(frozen=True)
+class ActorBackfillCandidate:
+    job_id: int
+    context_json: str
+    trigger_actor: str | None
+    trigger_actor_avatar_url: str | None
+
+
+@dataclass(frozen=True)
+class ActorBackfillUpdate:
+    job_id: int
+    trigger_actor: str
+    trigger_actor_avatar_url: str | None
+
+
+class ActorBackfillRepository:
+    """Read and update persisted trigger-actor data for the operator backfill."""
+
+    def __init__(self, database: Database) -> None:
+        self.database = database
+
+    def validate_current_schema(self) -> None:
+        with self.database.read_only() as con:
+            validate_migrations(con)
+
+    def list_candidates(self, limit: int | None = None) -> list[ActorBackfillCandidate]:
+        with self.database.read_only() as con:
+            jobs_table = con.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='jobs'"
+            ).fetchone()
+            if jobs_table is None:
+                return []
+            columns = {
+                str(row["name"])
+                for row in con.execute("PRAGMA table_info(jobs)")
+            }
+            has_actor = "trigger_actor" in columns
+            has_avatar = "trigger_actor_avatar_url" in columns
+            conditions = []
+            if has_actor:
+                conditions.append("(trigger_actor IS NULL OR trigger_actor='')")
+            if has_avatar:
+                conditions.append(
+                    "(trigger_actor_avatar_url IS NULL OR trigger_actor_avatar_url='')"
+                )
+            where = f"WHERE {' OR '.join(conditions)}" if conditions else ""
+            actor_column = "trigger_actor" if has_actor else "NULL AS trigger_actor"
+            avatar_column = (
+                "trigger_actor_avatar_url"
+                if has_avatar
+                else "NULL AS trigger_actor_avatar_url"
+            )
+            rows = con.execute(
+                f"""SELECT id, context_json, {actor_column}, {avatar_column}
+                FROM jobs
+                {where}
+                ORDER BY id
+                LIMIT ?""",
+                (max(1, limit or 1000000),),
+            ).fetchall()
+        return [
+            ActorBackfillCandidate(
+                job_id=int(row["id"]),
+                context_json=str(row["context_json"]),
+                trigger_actor=(
+                    str(row["trigger_actor"]) if row["trigger_actor"] else None
+                ),
+                trigger_actor_avatar_url=(
+                    str(row["trigger_actor_avatar_url"])
+                    if row["trigger_actor_avatar_url"]
+                    else None
+                ),
+            )
+            for row in rows
+        ]
+
+    def apply_updates(self, updates: list[ActorBackfillUpdate]) -> None:
+        if not updates:
+            return
+        with self.database.transaction() as con:
+            con.executemany(
+                "UPDATE jobs SET trigger_actor=?, trigger_actor_avatar_url=? WHERE id=?",
+                [
+                    (
+                        update.trigger_actor,
+                        update.trigger_actor_avatar_url,
+                        update.job_id,
+                    )
+                    for update in updates
+                ],
+            )

src/github_agent_bridge/persistence/ingestion.py

@@ -5,8 +5,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 
-from ..actors import TriggerActor
-from ..models import GitHubContext, Job, Notification, utc_now
+from ..models import GitHubContext, Job, Notification, TriggerActor, utc_now
 from .acknowledgements import AcknowledgementRepository
 from .commit_statuses import CommitStatusRepository
 from .database import Database, TransactionMode