fix: do not recreate tasks that already have the derived UUID

Registrar-minted tasks keep the derived UUID with record_id unset.
Inserting them again collides. Match by UUID, then write record_id.

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

api/services/forge_projection.py

@@ -671,18 +671,25 @@ def _sync_existing_workplan_tasks(
         if t.record_id and t.record_id.strip()
     }
     matched: dict[str, Any] = {}
+    claimed: set[str] = set()
     stale: list[Any] = []
+    by_uuid = {str(ht.id): ht for ht in hub_tasks}
     for ht in hub_tasks:
         rid = (ht.record_id or "").strip().lower()
-        if not rid:
-            continue
-        if rid in want:
+        if rid and rid in want:
             matched[rid] = ht
-        else:
+            claimed.add(str(ht.id))
+        elif rid:
             stale.append(ht)
 
     for key, dt in want.items():
         ht = matched.get(key)
+        if ht is None and dt.uuid in by_uuid:
+            # Already on the hub under the derived UUID, but record_id was never
+            # written (the registrar-minted case). Overwriting would collide.
+            ht = by_uuid[dt.uuid]
+            matched[key] = ht
+            claimed.add(str(ht.id))
         if ht is None:
             kwargs: dict[str, Any] = {
                 "id": uuid.UUID(dt.uuid),
@@ -702,6 +709,9 @@ def _sync_existing_workplan_tasks(
             outcome.created_tasks.append(dt.record_id)
             continue
         changed = False
+        if not (ht.record_id or "").strip():
+            ht.record_id = dt.record_id
+            changed = True
         if dt.title and dt.title.strip() and ht.title != dt.title.strip():
             ht.title = dt.title.strip()
             changed = True

tests/test_forge_projection.py

@@ -1107,6 +1107,18 @@ class TestExistingWorkplanTasks:
         assert "DEMO-WP-0001-T09" in out.cancelled_tasks
         assert session.deleted == []
 
+    @pytest.mark.asyncio
+    async def test_sets_record_id_when_the_derived_uuid_already_exists(self):
+        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
+        existing = _TaskRow(None, status="done", title="Clear the stale hub rows", workplan_id=row.id)
+        derived = self._derived(("DEMO-WP-0001-T01", "Clear the stale hub rows", "done"))
+        existing.id = __import__("uuid").UUID(derived.workplans[0].tasks[0].uuid)
+        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[existing])
+        out = await fp.reset_repository_projection(session, "demo", derived=derived)
+        assert out.created_tasks == []
+        assert existing.record_id == "DEMO-WP-0001-T01"
+        assert session.added == []
+
     @pytest.mark.asyncio
     async def test_leaves_unidentified_tasks_alone(self):
         row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")