diff --git a/src/panopticon/taskservice/store_sqlalchemy.py b/src/panopticon/taskservice/store_sqlalchemy.py index 79d7b9d0..e9fa1aa2 100644 --- a/src/panopticon/taskservice/store_sqlalchemy.py +++ b/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( diff --git a/tests/core/test_store.py b/tests/core/test_store.py index c6b24140..ddc0afb4 100644 --- a/tests/core/test_store.py +++ b/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() diff --git a/tests/taskservice/test_client.py b/tests/taskservice/test_client.py index aa81e97c..506f692c 100644 --- a/tests/taskservice/test_client.py +++ b/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: