Fix repo update silently dropping agent_cli; single-source the domain→row column mapping

src/panopticon/taskservice/store_sqlalchemy.py

@@ -3,7 +3,10 @@
 One adapter serves every SQL backend SQLAlchemy speaks; **"in-memory" is just an in-memory
 SQLite engine**. The pure, frozen domain models (:mod:`panopticon.core.models`) never touch
 the ORM — this adapter owns mutable *row* classes and each knows how to translate itself
-``to_domain`` / ``from_domain``. Parent→child links are ORM ``relationship``\\ s (loaded
+``to_domain`` / ``from_domain``. Each row class states its domain→column mapping **once**, in
+``column_values``, which both the insert (``from_domain``) and the update (``_apply_columns``)
+go through — so a newly added column can't be carried on create and silently dropped on update.
+Parent→child links are ORM ``relationship``\\ s (loaded
 eagerly via ``selectin``), so reading a task pulls in its history and responsibilities and
 writing one cascades — no hand-written load/insert code.
 
@@ -72,6 +75,18 @@ class _Base(DeclarativeBase):
 metadata = _Base.metadata
 
 
+def _apply_columns(row: _Base, values: dict[str, Any]) -> None:
+    """Overwrite ``row``'s columns from a ``column_values`` mapping, leaving its ``id`` alone.
+
+    The counterpart to each row class's ``from_domain``: an update writes exactly the columns an
+    insert does, so a newly added column can't be persisted on create and then silently dropped
+    on update (which is how ``Repo.agent_cli`` was lost — see ``column_values``).
+    """
+    for key, value in values.items():
+        if key != "id":
+            setattr(row, key, value)
+
+
 class _RepoRow(_Base):
     __tablename__ = "repo"
 
@@ -104,22 +119,32 @@ def to_domain(self) -> Repo:
             disabled_workflows=list(self.disabled_workflows or []),
         )
 
+    @classmethod
+    def column_values(cls, repo: Repo) -> dict[str, Any]:
+        """The domain→row column mapping: the one place a new repo column gets wired up.
+
+        Both the insert (:meth:`from_domain`) and the update (:func:`_apply_columns`) read it, so
+        the two can't drift — the failure mode this replaces was ``agent_cli`` being carried on
+        create and dropped by a hand-copied update, making a repo un-switchable to another CLI.
+        """
+        return {
+            "id": repo.id,
+            "name": repo.name,
+            "git_url": repo.git_url,
+            "default_base": repo.default_base,
+            "env_file": repo.env_file,
+            "credential_dir": repo.credential_dir,
+            "image_layer_file": repo.image_layer_file,
+            "capabilities": dict(repo.capabilities),
+            "hook_file": repo.hook_file,
+            "agent_cli": repo.agent_cli,
+            "enabled_workflows": list(repo.enabled_workflows),
+            "disabled_workflows": list(repo.disabled_workflows),
+        }
+
     @classmethod
     def from_domain(cls, repo: Repo) -> _RepoRow:
-        return cls(
-            id=repo.id,
-            name=repo.name,
-            git_url=repo.git_url,
-            default_base=repo.default_base,
-            env_file=repo.env_file,
-            credential_dir=repo.credential_dir,
-            image_layer_file=repo.image_layer_file,
-            capabilities=dict(repo.capabilities),
-            hook_file=repo.hook_file,
-            agent_cli=repo.agent_cli,
-            enabled_workflows=list(repo.enabled_workflows),
-            disabled_workflows=list(repo.disabled_workflows),
-        )
+        return cls(**cls.column_values(repo))
 
 
 class _TaskRow(_Base):
@@ -179,30 +204,41 @@ def to_domain(self) -> Task:
             history=[h.to_domain() for h in self.history],
         )
 
+    @classmethod
+    def column_values(cls, task: Task) -> dict[str, Any]:
+        """The domain→row column mapping: the one place a new task column gets wired up.
+
+        Columns only — ``history`` is a relationship, persisted append-only by ``_update_task``
+        (see the module docstring), never through this.
+        """
+        return {
+            "id": task.id,
+            "repo_id": task.repo_id,
+            "workflow": task.workflow,
+            "state": task.state,
+            "turn": task.turn.value,
+            "blocked": task.blocked,
+            "memo": task.memo,
+            "initial_prompt": task.initial_prompt,
+            "slug": task.slug,
+            "url": task.url,
+            "snoozed_until": task.snoozed_until,
+            "branch": task.branch,
+            "clone": task.clone,
+            "claimed_by": task.claimed_by,
+            "starting_model": task.starting_model,
+            "agent_cli": task.agent_cli,
+            "governor_task_id": task.governor_task_id,
+            "created_at": task.created_at,
+            "updated_at": task.updated_at,
+            "sort_weight": task.sort_weight,
+            "depends_on_task_ids": list(task.depends_on_task_ids),
+        }
+
     @classmethod
     def from_domain(cls, task: Task) -> _TaskRow:
         return cls(
-            id=task.id,
-            repo_id=task.repo_id,
-            workflow=task.workflow,
-            state=task.state,
-            turn=task.turn.value,
-            blocked=task.blocked,
-            memo=task.memo,
-            initial_prompt=task.initial_prompt,
-            slug=task.slug,
-            url=task.url,
-            snoozed_until=task.snoozed_until,
-            branch=task.branch,
-            clone=task.clone,
-            claimed_by=task.claimed_by,
-            starting_model=task.starting_model,
-            agent_cli=task.agent_cli,
-            governor_task_id=task.governor_task_id,
-            created_at=task.created_at,
-            updated_at=task.updated_at,
-            sort_weight=task.sort_weight,
-            depends_on_task_ids=list(task.depends_on_task_ids),
+            **cls.column_values(task),
             history=[_HistoryRow.from_domain(e, seq) for seq, e in enumerate(task.history)],
         )
 
@@ -348,16 +384,7 @@ async def _update_repo(self, repo: Repo) -> None:
             row = await s.get(_RepoRow, repo.id)
             if row is None:
                 raise NotFound(f"repo {repo.id!r} does not exist")
-            row.name = repo.name
-            row.git_url = repo.git_url
-            row.default_base = repo.default_base
-            row.env_file = repo.env_file
-            row.credential_dir = repo.credential_dir
-            row.image_layer_file = repo.image_layer_file
-            row.capabilities = dict(repo.capabilities)
-            row.hook_file = repo.hook_file
-            row.enabled_workflows = list(repo.enabled_workflows)
-            row.disabled_workflows = list(repo.disabled_workflows)
+            _apply_columns(row, _RepoRow.column_values(repo))
 
     # -- tasks: reads + persistence primitives (the base's template methods drive these) --
 
@@ -398,19 +425,7 @@ async def _update_task(self, task: Task, stored: Sequence[HistoryEntry]) -> None
             row = await s.get(_TaskRow, task.id)
             if row is None:  # defensive: single-writer, so it still exists after _stored_history
                 raise NotFound(f"task {task.id!r} does not exist")
-            row.state = task.state
-            row.turn = task.turn.value
-            row.blocked = task.blocked
-            row.slug = task.slug
-            row.url = task.url
-            row.snoozed_until = task.snoozed_until
-            row.branch = task.branch
-            row.clone = task.clone
-            row.claimed_by = task.claimed_by
-            row.governor_task_id = task.governor_task_id
-            row.updated_at = task.updated_at
-            row.sort_weight = task.sort_weight
-            row.depends_on_task_ids = list(task.depends_on_task_ids)
+            _apply_columns(row, _TaskRow.column_values(task))
             # The current (last stored) entry's promises may have been fulfilled in place.
             if stored:
                 _fulfil_current_promises(

tests/core/test_store.py

@@ -9,7 +9,7 @@
 from __future__ import annotations
 
 from collections.abc import AsyncIterator
-from dataclasses import MISSING, fields, is_dataclass
+from dataclasses import MISSING, fields, is_dataclass, replace
 from pathlib import Path
 from typing import Any, get_args, get_origin, get_type_hints
 
@@ -477,6 +477,42 @@ def test_rows_and_domain_models_stay_in_sync(domain: type) -> None:
     )
 
 
+def _fully_populated_repo() -> Repo:
+    """A repo touching every field of Repo with a non-default value."""
+    return Repo(
+        id="r-full",
+        name="acme/widgets",
+        git_url="https://github.com/acme/widgets.git",
+        default_base="trunk",
+        env_file="r-full.env",
+        credential_dir="r-full-creds",
+        image_layer_file="r-full.layer",
+        capabilities={"docker_in_docker": True},
+        hook_file="prep.sh",
+        enabled_workflows=["spike"],
+        disabled_workflows=["github-peer-reviewed"],
+        agent_cli="codex",
+    )
+
+
+def _mutated_repo(repo: Repo) -> Repo:
+    """``repo`` with every updatable field moved to a second distinct value."""
+    return replace(
+        repo,
+        name="acme/gadgets",
+        git_url="https://github.com/acme/gadgets.git",
+        default_base="main",
+        env_file="other.env",
+        credential_dir="other-creds",
+        image_layer_file="other.layer",
+        capabilities={"docker_in_docker": False},
+        hook_file="other.sh",
+        enabled_workflows=["github-self-reviewed"],
+        disabled_workflows=["spike"],
+        agent_cli="claude",
+    )
+
+
 def _fully_populated_task() -> Task:
     """A task touching every field of Task/HistoryEntry/Responsibility with a non-default value."""
     return Task(
@@ -530,6 +566,38 @@ def _fully_populated_task() -> Task:
     )
 
 
+def _mutated_task(task: Task) -> Task:
+    """``task`` with every updatable field moved to a second distinct value.
+
+    The appended history entry carries the new state (a task's state must match its history
+    tail) — updating in place is not allowed for recorded transitions.
+    """
+    return replace(
+        task,
+        state="REVIEW",
+        turn=Actor.USER,
+        blocked=False,
+        memo="make the widget blue",
+        initial_prompt="have another look",
+        slug="fix-the-gadget",
+        url="https://github.com/acme/widgets/pull/8",
+        snoozed_until="2026-09-06T03:00:00+00:00",
+        branch="panopticon/fix-the-gadget",
+        clone="/clones/t-full-again",
+        claimed_by="remote",
+        starting_model="secondary",
+        agent_cli="claude",
+        governor_task_id=None,
+        updated_at="t3",
+        sort_weight=9,
+        depends_on_task_ids=["t-dep-2"],
+        history=[
+            *task.history,
+            HistoryEntry(at="t3", from_state="WORKING", to_state="REVIEW", trigger="advance"),
+        ],
+    )
+
+
 def _assert_every_field_exercised(instances: list[Any], domain: type) -> None:
     """Fail if any field of ``domain`` equals its default across *all* ``instances``.
 
@@ -549,6 +617,52 @@ def _assert_every_field_exercised(instances: list[Any], domain: type) -> None:
             )
 
 
+def _assert_every_field_changed(before: Any, after: Any, domain: type, immutable: set[str]) -> None:
+    """Fail if any updatable field of ``domain`` holds the same value in ``before`` and ``after``.
+
+    The mirror of :func:`_assert_every_field_exercised` for the *update* path: it forces the
+    mutation fixture to move every new field, so the round-trip below genuinely proves the field
+    is written by an update — the gap that let ``Repo.agent_cli`` be carried on create and
+    silently dropped on update.
+    """
+    for f in fields(domain):
+        if f.name in immutable:
+            continue
+        if getattr(before, f.name) == getattr(after, f.name):
+            pytest.fail(
+                f"{domain.__name__}.{f.name} is never changed — extend the mutation fixture"
+            )
+
+
+async def test_full_repo_round_trips_on_create_and_update(store: Store) -> None:
+    repo = _fully_populated_repo()
+    updated = _mutated_repo(repo)
+    _assert_every_field_exercised([repo], Repo)
+    _assert_every_field_changed(repo, updated, Repo, immutable={"id"})
+
+    await store.create_repo(repo)
+    assert await store.get_repo(repo.id) == repo  # dataclass __eq__: every field survives
+
+    await store.update_repo(updated)
+    assert await store.get_repo(repo.id) == updated  # ...and an update writes every one of them
+
+
+# Identity and creation facts: set once by the insert and never rewritten, so the update
+# round-trip below doesn't assert on them. Any *new* field is guarded by default.
+_TASK_IMMUTABLE = {"id", "repo_id", "workflow", "created_at", "history"}
+
+
+async def test_full_task_update_round_trips_and_exercises_every_field(store: Store) -> None:
+    await _seed_repo(store)
+    task = _fully_populated_task()
+    updated = _mutated_task(task)
+    _assert_every_field_changed(task, updated, Task, immutable=_TASK_IMMUTABLE)
+
+    await store.create_task(task)
+    await store.save_task(updated)
+    assert await store.get_task(task.id) == updated
+
+
 async def test_full_task_round_trips_and_exercises_every_field(store: Store) -> None:
     await _seed_repo(store)
     task = _fully_populated_task()

tests/taskservice/test_client.py

@@ -130,7 +130,8 @@ def test_repo_agent_cli_defaults_and_round_trips_over_rest(client: TaskServiceCl
     assert codex["agent_cli"] == "codex"
     assert client.get_repo("r9")["agent_cli"] == "codex"  # persisted
     patched = client.update_repo("r9", agent_cli="claude")
-    assert patched["agent_cli"] == "claude"  # PATCH updates it
+    assert patched["agent_cli"] == "claude"  # PATCH updates it...
+    assert client.get_repo("r9")["agent_cli"] == "claude"  # ...and the update is persisted
 
 
 def test_create_task_carries_agent_cli_over_rest(client: TaskServiceClient) -> None: