diff --git a/docs/operations.md b/docs/operations.md index 2960c57..f1f265d 100644 --- a/docs/operations.md +++ b/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 diff --git a/docs/persistence.md b/docs/persistence.md index 2263586..c5a927a 100644 --- a/docs/persistence.md +++ b/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 diff --git a/src/github_agent_bridge/actors.py b/src/github_agent_bridge/actors.py index f4c1588..421d72b 100644 --- a/src/github_agent_bridge/actors.py +++ b/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, + } diff --git a/src/github_agent_bridge/models.py b/src/github_agent_bridge/models.py index bf0d3ce..6d94e35 100644 --- a/src/github_agent_bridge/models.py +++ b/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] diff --git a/src/github_agent_bridge/persistence/__init__.py b/src/github_agent_bridge/persistence/__init__.py index 3db45d1..401e444 100644 --- a/src/github_agent_bridge/persistence/__init__.py +++ b/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", diff --git a/src/github_agent_bridge/persistence/actors.py b/src/github_agent_bridge/persistence/actors.py new file mode 100644 index 0000000..0911230 --- /dev/null +++ b/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 + ], + ) diff --git a/src/github_agent_bridge/persistence/ingestion.py b/src/github_agent_bridge/persistence/ingestion.py index a9bc7a1..cb23baf 100644 --- a/src/github_agent_bridge/persistence/ingestion.py +++ b/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 diff --git a/src/github_agent_bridge/queue.py b/src/github_agent_bridge/queue.py index d272367..6b8d464 100644 --- a/src/github_agent_bridge/queue.py +++ b/src/github_agent_bridge/queue.py @@ -2,7 +2,6 @@ import hashlib import sqlite3 -from importlib import resources from pathlib import Path from .models import GitHubContext, Job, Notification @@ -24,17 +23,7 @@ from . import feedback from .actors import trigger_actor_details_for_enqueue, trigger_actor_details_from_notification from .intent_classifier import ParserResult, classify_notification_with_llm, should_classify_with_llm -from .sql.migrations import apply_migrations, validate_migrations - -SCHEMA_PACKAGE = "github_agent_bridge.sql" - - -def load_schema() -> str: - """Read the packaged SQLite schema resource.""" - return resources.files(SCHEMA_PACKAGE).joinpath("schema.sql").read_text(encoding="utf-8") - - -SCHEMA = load_schema() +from .sql.migrations import SCHEMA, initialize_database, validate_migrations def canonical_event_key( @@ -103,38 +92,17 @@ def init(self) -> None: validate_migrations(con) return with self.database.read_write() as con: - self._initialize_database(con) + initialize_database(con) def migrate(self) -> None: """Initialize or migrate a database through an explicit operator path.""" with self.database.read_write() as con: - self._initialize_database(con) + initialize_database(con) def _ensure_initialized(self) -> None: if not self.path.exists(): self.init() - def _initialize_database(self, con: sqlite3.Connection) -> None: - history_exists = con.execute( - "SELECT 1 FROM sqlite_master WHERE type='table' AND name='schema_migrations'" - ).fetchone() - baseline_applied = bool( - history_exists - and con.execute( - "SELECT 1 FROM schema_migrations WHERE version=1" - ).fetchone() - ) - if baseline_applied: - # Validate/apply immutable steps before the rolling schema snapshot can - # touch a database created by this or a newer package version. - apply_migrations(con) - con.executescript(SCHEMA) - else: - # Legacy or incomplete histories need the snapshot to create missing - # tables before the baseline migration can perform its backfills. - con.executescript(SCHEMA) - apply_migrations(con) - def enqueue(self, n: Notification, policy: Policy) -> tuple[Job | None, str]: """Backward-compatible email enqueue entrypoint.""" return self.ingest(n, policy, source="email", source_key=n.message_id) diff --git a/src/github_agent_bridge/sql/migrations/__init__.py b/src/github_agent_bridge/sql/migrations/__init__.py index b4f04fe..af9083d 100644 --- a/src/github_agent_bridge/sql/migrations/__init__.py +++ b/src/github_agent_bridge/sql/migrations/__init__.py @@ -11,6 +11,19 @@ MIGRATION_FILENAME = re.compile(r"^v(?P[0-9]{4})_[a-z0-9_]+\.py$") +SCHEMA_PACKAGE = "github_agent_bridge.sql" + + +def load_schema() -> str: + """Read the packaged rolling schema snapshot.""" + return ( + resources.files(SCHEMA_PACKAGE) + .joinpath("schema.sql") + .read_text(encoding="utf-8") + ) + + +SCHEMA = load_schema() class MigrationError(RuntimeError): @@ -200,3 +213,26 @@ def apply_migrations( if pending is None: return tuple(completed) completed.append(pending.version) + + +def initialize_database(con: sqlite3.Connection) -> None: + """Initialize a fresh database or explicitly migrate an existing one.""" + history_exists = con.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='schema_migrations'" + ).fetchone() + baseline_applied = bool( + history_exists + and con.execute( + "SELECT 1 FROM schema_migrations WHERE version=1" + ).fetchone() + ) + if baseline_applied: + # Validate/apply immutable steps before the rolling schema snapshot can + # touch a database created by this or a newer package version. + apply_migrations(con) + con.executescript(SCHEMA) + else: + # Legacy or incomplete histories need the snapshot to create missing + # tables before the baseline migration can perform its backfills. + con.executescript(SCHEMA) + apply_migrations(con) diff --git a/tests/test_actors.py b/tests/test_actors.py index ab15d4b..b8f4047 100644 --- a/tests/test_actors.py +++ b/tests/test_actors.py @@ -4,6 +4,8 @@ import sqlite3 import subprocess +import pytest + from github_agent_bridge.actors import ( actor_details_from_github_payload, actor_endpoint, @@ -16,6 +18,7 @@ from github_agent_bridge.models import GitHubContext, Notification from github_agent_bridge.policy import Policy from github_agent_bridge.queue import JobQueue +from github_agent_bridge.sql.migrations import MigrationRequiredError def test_trigger_actor_from_notification_uses_github_sender_login(): @@ -168,6 +171,23 @@ def fake_run(args, check=False, stdout=None, stderr=None, text=False, timeout=No assert "trigger_actor_avatar_url" not in columns +def test_backfill_write_requires_migrated_schema(tmp_path): + db = tmp_path / "legacy.sqlite3" + con = sqlite3.connect(db) + con.execute("CREATE TABLE jobs (id INTEGER PRIMARY KEY, context_json TEXT NOT NULL)") + con.execute("INSERT INTO jobs(id, context_json) VALUES(1, '{}')") + con.commit() + con.close() + + with pytest.raises(MigrationRequiredError, match="run `gab migrate-db`"): + backfill_trigger_actors(db) + + con = sqlite3.connect(db) + columns = {row[1] for row in con.execute("PRAGMA table_info(jobs)")} + con.close() + assert columns == {"id", "context_json"} + + def test_backfill_trigger_actors_fills_missing_avatar_without_api(tmp_path, monkeypatch): db = tmp_path / "q.sqlite3" monkeypatch.setattr("github_agent_bridge.actors.github_actor_details_for_context", lambda ctx, *, gh_bin="gh": None) diff --git a/tests/test_persistence_architecture.py b/tests/test_persistence_architecture.py new file mode 100644 index 0000000..ca5e36c --- /dev/null +++ b/tests/test_persistence_architecture.py @@ -0,0 +1,80 @@ +import ast +from pathlib import Path + + +PACKAGE_ROOT = Path(__file__).parents[1] / "src" / "github_agent_bridge" +SQL_METHODS = {"execute", "executemany", "executescript"} + + +def _allows_sql(path: Path) -> bool: + relative = path.relative_to(PACKAGE_ROOT) + return ( + relative == Path("dashboard_data.py") + or relative.parts[0] == "persistence" + or relative.parts[:2] == ("sql", "migrations") + ) + + +def _sqlite_connect_calls(tree: ast.AST) -> list[int]: + module_aliases = {"sqlite3"} + connect_aliases: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + module_aliases.update( + alias.asname or alias.name + for alias in node.names + if alias.name == "sqlite3" + ) + elif isinstance(node, ast.ImportFrom) and node.module == "sqlite3": + connect_aliases.update( + alias.asname or alias.name + for alias in node.names + if alias.name == "connect" + ) + return [ + node.lineno + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and ( + ( + isinstance(node.func, ast.Name) + and node.func.id in connect_aliases + ) + or ( + isinstance(node.func, ast.Attribute) + and node.func.attr == "connect" + and isinstance(node.func.value, ast.Name) + and node.func.value.id in module_aliases + ) + ) + ] + + +def _sql_method_calls(tree: ast.AST) -> list[tuple[int, str]]: + return [ + (node.lineno, node.func.attr) + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr in SQL_METHODS + ] + + +def test_sqlite_connections_and_sql_stay_inside_persistence_boundaries(): + violations: list[str] = [] + for path in sorted(PACKAGE_ROOT.rglob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + relative = path.relative_to(PACKAGE_ROOT) + connect_lines = _sqlite_connect_calls(tree) + if relative != Path("persistence/database.py"): + violations.extend( + f"{relative}:{line}: direct sqlite3.connect" + for line in connect_lines + ) + if not _allows_sql(path): + violations.extend( + f"{relative}:{line}: direct .{method}()" + for line, method in _sql_method_calls(tree) + ) + + assert violations == []