sync identified tasks of existing workplans on projection reset

Hub tasks now carry record_id, so reset_repository_projection can create
missing derived tasks, update title/status, and cancel open tasks the
forge no longer derives. Rows with no record_id are left alone.

Assistant: grok
Assistant-Session: 01a04996-76e8-7f53-b971-1885cfbed436

tests/test_forge_projection.py

@@ -218,11 +218,12 @@ class _Row:
 class _FakeSession:
     """Stands in for AsyncSession: enough to prove intent without a database."""
 
-    def __init__(self, repo, rows, foreign=None, slug_clash=None):
+    def __init__(self, repo, rows, foreign=None, slug_clash=None, task_rows=None):
         self._repo = repo
         self.rows = list(rows)
         self._foreign = list(foreign or [])
         self._slug_clash = list(slug_clash or [])
+        self._task_rows = task_rows
         self.added = []
         self.deleted = []
         self.committed = False
@@ -231,16 +232,24 @@ class _FakeSession:
     async def execute(self, *_a, **_k):
         self._calls += 1
         repo, rows = self._repo, self.rows
-        # 1st call resolves the repo, 2nd loads its workplans, 3rd is the
-        # foreign-identifier lookup.
         # 1 resolves the repo, 2 loads its workplans, 3 is the identifier
-        # lookup, 4 the slug lookup.
+        # lookup, 4 the slug lookup. On the update-only path (no creates)
+        # the 3rd call is the existing-task load instead.
+        update_only = (
+            self._task_rows is not None
+            and not self._foreign
+            and not self._slug_clash
+        )
         if self._calls == 2:
             payload = rows
+        elif update_only and self._calls == 3:
+            payload = self._task_rows
         elif self._calls == 3:
             payload = self._foreign
         elif self._calls == 4:
             payload = self._slug_clash
+        elif self._calls == 5:
+            payload = self._task_rows or []
         else:
             payload = []
 
@@ -1030,3 +1039,81 @@ class TestTitleSyncBehaviour:
         session = _FakeSession(repo=_Repo(), rows=[row])
         await fp.reset_repository_projection(session, "demo", derived=d)
         assert row.title == "Real Title"
+
+
+class _TaskRow:
+    def __init__(self, record_id, status="todo", title="t", workplan_id=None):
+        import uuid as _u
+        self.id = _u.uuid4()
+        self.record_id = record_id
+        self.status = status
+        self.title = title
+        self.priority = "medium"
+        self.workplan_id = workplan_id
+
+
+class TestExistingWorkplanTasks:
+    """Tasks of an existing workplan are matched by record_id (CUST-WP-0068-T09)."""
+
+    def _derived(self, *tasks):
+        dts = [
+            fp.DerivedTask(
+                record_id=tid,
+                uuid=fp.derived_record_uuid(tid),
+                title=title,
+                status=st,
+                priority="medium",
+            )
+            for tid, title, st in tasks
+        ]
+        return fp.DerivedProjection(
+            repo_slug="demo",
+            commit="c0ffee",
+            workplans=[
+                fp.DerivedWorkplan(
+                    record_id="DEMO-WP-0001",
+                    uuid=fp.derived_record_uuid("DEMO-WP-0001"),
+                    title="DEMO-WP-0001",
+                    status="active",
+                    relative_path="workplans/a.md",
+                    archived=False,
+                    tasks=dts,
+                )
+            ],
+        )
+
+    @pytest.mark.asyncio
+    async def test_creates_a_missing_identified_task(self):
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[])
+        out = await fp.reset_repository_projection(
+            session, "demo",
+            derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")),
+        )
+        assert out.created_tasks == ["DEMO-WP-0001-T01"]
+        assert session.added
+        assert session.added[0].record_id == "DEMO-WP-0001-T01"
+
+    @pytest.mark.asyncio
+    async def test_cancels_an_open_task_the_file_no_longer_derives(self):
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        stale = _TaskRow("DEMO-WP-0001-T09", status="todo", workplan_id=row.id)
+        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[stale])
+        out = await fp.reset_repository_projection(
+            session, "demo",
+            derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")),
+        )
+        assert stale.status.value == "cancel" or stale.status == "cancel" or str(stale.status).endswith("cancel")
+        assert "DEMO-WP-0001-T09" in out.cancelled_tasks
+        assert session.deleted == []
+
+    @pytest.mark.asyncio
+    async def test_leaves_unidentified_tasks_alone(self):
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        orphan = _TaskRow(None, status="todo", title="legacy", workplan_id=row.id)
+        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[orphan])
+        await fp.reset_repository_projection(
+            session, "demo",
+            derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")),
+        )
+        assert orphan.status == "todo"