feat(persistence): repository port + in-memory & SQLite adapters

Slice 1, PR 2 of 4 — the persistence boundary.

- panopticon.core.repository: the Repository ABC (the port, ADR 0001/0006) plus
  shared integrity checks (non-empty consistent history; append-only history).
- panopticon.taskservice.repository_memory.InMemoryRepository: dict-backed adapter
  (stores deep copies) for tests and the walking-skeleton runner.
- panopticon.taskservice.repository_sqlite.SqliteRepository: SQLite adapter with
  user_version-based forward migrations; SQLite specifics stay encapsulated.
- Contract tests parametrized over BOTH backends — proving the port is genuinely
  backend-agnostic and that both enforce identical integrity rules.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tildesrc

src/panopticon/core/repository.py

@@ -0,0 +1,98 @@
+"""The repository port: the abstraction over persisted task state.
+
+A backend-agnostic interface (ADR 0001/0006). The task service is its sole owner and the
+single writer; adapters implement it (in-memory and SQLite in this slice; Postgres later).
+
+Integrity rules every adapter must enforce — the "transition enforcement at the boundary":
+
+* ``create_task`` — the task id is unique, its ``repo_id`` exists, its history is
+  non-empty, and ``state`` equals the last history entry's ``to_state``.
+* ``save_task`` — the task exists, its history is **append-only** (the stored history is a
+  prefix of the supplied one), and ``state`` equals the last history entry's ``to_state``.
+
+The *legality* of a transition (which state may follow which) is decided by the engine
+before save; the repository guarantees the persisted record stays internally consistent.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Sequence
+
+from panopticon.core.models import HistoryEntry, Repo, Task
+
+
+class RepositoryError(Exception):
+    """Base class for repository failures."""
+
+
+class NotFound(RepositoryError):
+    """Raised when an entity referenced by id does not exist."""
+
+
+class AlreadyExists(RepositoryError):
+    """Raised when creating an entity whose id is already taken."""
+
+
+class IntegrityError(RepositoryError):
+    """Raised when a write would violate an integrity rule (e.g. non-append-only history)."""
+
+
+class Repository(ABC):
+    """Persistence boundary for repos and tasks."""
+
+    # -- repos --------------------------------------------------------------------
+
+    @abstractmethod
+    def create_repo(self, repo: Repo) -> None:
+        """Persist a new repo. Raises :class:`AlreadyExists` if its id is taken."""
+
+    @abstractmethod
+    def get_repo(self, repo_id: str) -> Repo | None:
+        """Return the repo, or ``None`` if it does not exist."""
+
+    @abstractmethod
+    def list_repos(self) -> list[Repo]:
+        """Return all repos."""
+
+    # -- tasks --------------------------------------------------------------------
+
+    @abstractmethod
+    def create_task(self, task: Task) -> None:
+        """Persist a new task and its initial history. Enforces the create-time rules."""
+
+    @abstractmethod
+    def get_task(self, task_id: str) -> Task | None:
+        """Return the task (with full history), or ``None`` if it does not exist."""
+
+    @abstractmethod
+    def list_tasks(self) -> list[Task]:
+        """Return all tasks (with full history)."""
+
+    @abstractmethod
+    def save_task(self, task: Task) -> None:
+        """Persist an updated task. Enforces the append-only and consistency rules."""
+
+
+# -- Shared integrity checks (adapters call these so the rules live in one place) --------
+
+
+def validate_task_invariants(task: Task) -> None:
+    """Check a task is internally consistent: non-empty history, state matches its tail."""
+    if not task.history:
+        raise IntegrityError(f"task {task.id!r} has empty history")
+    if task.state != task.history[-1].to_state:
+        raise IntegrityError(
+            f"task {task.id!r}: state {task.state!r} != last history to_state "
+            f"{task.history[-1].to_state!r}"
+        )
+
+
+def validate_append_only(
+    stored: Sequence[HistoryEntry], incoming: Sequence[HistoryEntry]
+) -> None:
+    """Check ``incoming`` history only extends ``stored`` (no edits, no shrink)."""
+    if len(incoming) < len(stored):
+        raise IntegrityError("history shrank (not append-only)")
+    if list(stored) != list(incoming[: len(stored)]):
+        raise IntegrityError("existing history was modified (not append-only)")

src/panopticon/taskservice/__init__.py

@@ -0,0 +1,5 @@
+"""The task service: the deterministic control plane.
+
+Owns the repository (the sole DB authority, ADR 0006), hosts the workflow registry, and
+drives task lifecycle. This package must remain LLM-free (enforced by a determinism test).
+"""

src/panopticon/taskservice/repository_memory.py

@@ -0,0 +1,66 @@
+"""In-memory repository adapter.
+
+Useful for tests and the walking-skeleton runner, and — together with the SQLite
+adapter — proof that the repository port is genuinely backend-agnostic (ADR 0006).
+Stores deep copies so callers can't mutate persisted state by holding a reference.
+"""
+
+from __future__ import annotations
+
+import copy
+
+from panopticon.core.models import Repo, Task
+from panopticon.core.repository import (
+    AlreadyExists,
+    NotFound,
+    Repository,
+    validate_append_only,
+    validate_task_invariants,
+)
+
+
+class InMemoryRepository(Repository):
+    """A dict-backed :class:`~panopticon.core.repository.Repository`."""
+
+    def __init__(self) -> None:
+        self._repos: dict[str, Repo] = {}
+        self._tasks: dict[str, Task] = {}
+
+    # -- repos --------------------------------------------------------------------
+
+    def create_repo(self, repo: Repo) -> None:
+        if repo.id in self._repos:
+            raise AlreadyExists(f"repo {repo.id!r} already exists")
+        self._repos[repo.id] = copy.deepcopy(repo)
+
+    def get_repo(self, repo_id: str) -> Repo | None:
+        repo = self._repos.get(repo_id)
+        return copy.deepcopy(repo) if repo is not None else None
+
+    def list_repos(self) -> list[Repo]:
+        return [copy.deepcopy(r) for r in self._repos.values()]
+
+    # -- tasks --------------------------------------------------------------------
+
+    def create_task(self, task: Task) -> None:
+        if task.id in self._tasks:
+            raise AlreadyExists(f"task {task.id!r} already exists")
+        if task.repo_id not in self._repos:
+            raise NotFound(f"repo {task.repo_id!r} does not exist")
+        validate_task_invariants(task)
+        self._tasks[task.id] = copy.deepcopy(task)
+
+    def get_task(self, task_id: str) -> Task | None:
+        task = self._tasks.get(task_id)
+        return copy.deepcopy(task) if task is not None else None
+
+    def list_tasks(self) -> list[Task]:
+        return [copy.deepcopy(t) for t in self._tasks.values()]
+
+    def save_task(self, task: Task) -> None:
+        stored = self._tasks.get(task.id)
+        if stored is None:
+            raise NotFound(f"task {task.id!r} does not exist")
+        validate_task_invariants(task)
+        validate_append_only(stored.history, task.history)
+        self._tasks[task.id] = copy.deepcopy(task)

src/panopticon/taskservice/repository_sqlite.py

@@ -0,0 +1,194 @@
+"""SQLite repository adapter (ADR 0001: SQLite first, behind the backend-agnostic port).
+
+Schema is versioned via ``PRAGMA user_version`` and migrated forward on open. SQLite-
+specific concerns stay encapsulated here; callers see only the :class:`Repository` port.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from collections.abc import Iterable
+from pathlib import Path
+from typing import cast
+
+from panopticon.core.models import Ball, HistoryEntry, Repo, Task
+from panopticon.core.repository import (
+    AlreadyExists,
+    NotFound,
+    Repository,
+    validate_append_only,
+    validate_task_invariants,
+)
+
+# Forward-only schema migrations. Index i in this list corresponds to user_version i+1.
+MIGRATIONS: tuple[str, ...] = (
+    # v1 — initial schema
+    """
+    CREATE TABLE repo (
+        id           TEXT PRIMARY KEY,
+        name         TEXT NOT NULL,
+        default_base TEXT NOT NULL
+    );
+    CREATE TABLE task (
+        id       TEXT PRIMARY KEY,
+        repo_id  TEXT NOT NULL REFERENCES repo(id),
+        workflow TEXT NOT NULL,
+        state    TEXT NOT NULL,
+        ball     TEXT NOT NULL,
+        slug     TEXT
+    );
+    CREATE TABLE history (
+        task_id    TEXT NOT NULL REFERENCES task(id),
+        seq        INTEGER NOT NULL,
+        at         TEXT NOT NULL,
+        from_state TEXT,
+        to_state   TEXT NOT NULL,
+        via        TEXT,
+        note       TEXT,
+        PRIMARY KEY (task_id, seq)
+    );
+    """,
+)
+
+
+class SqliteRepository(Repository):
+    """A SQLite-backed :class:`~panopticon.core.repository.Repository`."""
+
+    def __init__(self, path: str | Path = ":memory:") -> None:
+        self._conn = sqlite3.connect(str(path))
+        self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA foreign_keys = ON")
+        self._migrate()
+
+    def close(self) -> None:
+        self._conn.close()
+
+    def _migrate(self) -> None:
+        version = cast(int, self._conn.execute("PRAGMA user_version").fetchone()[0])
+        for i in range(version, len(MIGRATIONS)):
+            self._conn.executescript(MIGRATIONS[i])
+            self._conn.execute(f"PRAGMA user_version = {i + 1}")
+        self._conn.commit()
+
+    # -- repos --------------------------------------------------------------------
+
+    def create_repo(self, repo: Repo) -> None:
+        try:
+            self._conn.execute(
+                "INSERT INTO repo (id, name, default_base) VALUES (?, ?, ?)",
+                (repo.id, repo.name, repo.default_base),
+            )
+        except sqlite3.IntegrityError as exc:
+            raise AlreadyExists(f"repo {repo.id!r} already exists") from exc
+        self._conn.commit()
+
+    def get_repo(self, repo_id: str) -> Repo | None:
+        row = self._conn.execute(
+            "SELECT id, name, default_base FROM repo WHERE id = ?", (repo_id,)
+        ).fetchone()
+        return _row_to_repo(row) if row is not None else None
+
+    def list_repos(self) -> list[Repo]:
+        rows = self._conn.execute("SELECT id, name, default_base FROM repo ORDER BY id")
+        return [_row_to_repo(r) for r in rows]
+
+    # -- tasks --------------------------------------------------------------------
+
+    def create_task(self, task: Task) -> None:
+        validate_task_invariants(task)
+        if self._task_exists(task.id):
+            raise AlreadyExists(f"task {task.id!r} already exists")
+        if self.get_repo(task.repo_id) is None:
+            raise NotFound(f"repo {task.repo_id!r} does not exist")
+        with self._conn:  # transaction
+            self._conn.execute(
+                "INSERT INTO task (id, repo_id, workflow, state, ball, slug) "
+                "VALUES (?, ?, ?, ?, ?, ?)",
+                (task.id, task.repo_id, task.workflow, task.state, task.ball.value, task.slug),
+            )
+            self._insert_history(task.id, 0, task.history)
+
+    def get_task(self, task_id: str) -> Task | None:
+        row = self._conn.execute(
+            "SELECT id, repo_id, workflow, state, ball, slug FROM task WHERE id = ?",
+            (task_id,),
+        ).fetchone()
+        if row is None:
+            return None
+        return _row_to_task(row, self._load_history(task_id))
+
+    def list_tasks(self) -> list[Task]:
+        rows = self._conn.execute(
+            "SELECT id, repo_id, workflow, state, ball, slug FROM task ORDER BY id"
+        ).fetchall()
+        return [_row_to_task(r, self._load_history(cast(str, r["id"]))) for r in rows]
+
+    def save_task(self, task: Task) -> None:
+        validate_task_invariants(task)
+        stored = self._load_history(task.id)
+        if not self._task_exists(task.id):
+            raise NotFound(f"task {task.id!r} does not exist")
+        validate_append_only(stored, task.history)
+        with self._conn:  # transaction
+            self._conn.execute(
+                "UPDATE task SET state = ?, ball = ?, slug = ? WHERE id = ?",
+                (task.state, task.ball.value, task.slug, task.id),
+            )
+            self._insert_history(task.id, len(stored), task.history[len(stored) :])
+
+    # -- helpers ------------------------------------------------------------------
+
+    def _task_exists(self, task_id: str) -> bool:
+        return (
+            self._conn.execute("SELECT 1 FROM task WHERE id = ?", (task_id,)).fetchone()
+            is not None
+        )
+
+    def _insert_history(
+        self, task_id: str, start_seq: int, entries: Iterable[HistoryEntry]
+    ) -> None:
+        self._conn.executemany(
+            "INSERT INTO history (task_id, seq, at, from_state, to_state, via, note) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?)",
+            [
+                (task_id, start_seq + i, e.at, e.from_state, e.to_state, e.via, e.note)
+                for i, e in enumerate(entries)
+            ],
+        )
+
+    def _load_history(self, task_id: str) -> list[HistoryEntry]:
+        rows = self._conn.execute(
+            "SELECT at, from_state, to_state, via, note FROM history "
+            "WHERE task_id = ? ORDER BY seq",
+            (task_id,),
+        ).fetchall()
+        return [
+            HistoryEntry(
+                at=cast(str, r["at"]),
+                from_state=cast("str | None", r["from_state"]),
+                to_state=cast(str, r["to_state"]),
+                via=cast("str | None", r["via"]),
+                note=cast("str | None", r["note"]),
+            )
+            for r in rows
+        ]
+
+
+def _row_to_repo(row: sqlite3.Row) -> Repo:
+    return Repo(
+        id=cast(str, row["id"]),
+        name=cast(str, row["name"]),
+        default_base=cast(str, row["default_base"]),
+    )
+
+
+def _row_to_task(row: sqlite3.Row, history: list[HistoryEntry]) -> Task:
+    return Task(
+        id=cast(str, row["id"]),
+        repo_id=cast(str, row["repo_id"]),
+        workflow=cast(str, row["workflow"]),
+        state=cast(str, row["state"]),
+        ball=Ball(cast(str, row["ball"])),
+        slug=cast("str | None", row["slug"]),
+        history=history,
+    )