feat(tasks): give task rows a canonical record identifier

Every work-record type carried a stable identifier except tasks, whose rows held
only id, workplan_id, title, status and priority — nothing connecting a row to
CUST-WP-0067-T01 in the file it came from. Matching was therefore by title, so a
renamed heading looked like one task vanishing and another appearing, and the
forge-derived reset had to refuse to touch tasks at all.

Adds tasks.record_id (nullable: no migration can invent an identity for an
existing row) and a backfill that reads the pairing from the repository files,
where a task declares both its canonical id and its projection UUID. 5516 pairs
across 121 repositories with zero conflicts; 4456 of 6073 cache task rows
identified.

Diff and reset now key on record_id where present, falling back to a
title-prefixed key so an unidentified row stays visibly unidentified.

Unknown stays unknown: a row the files do not claim keeps no identity and the
reset keeps refusing to act on it, and an existing identity is never
overwritten — a mismatch is recorded as a conflict rather than resolved.

Refs STATE-WP-0083-T06

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 2583210@bnt-lap001
Assistant-Session: f2bff2d5-e9b2-4338-92ca-10282a927006

api/services/forge_projection.py

@@ -350,11 +350,19 @@ def diff_against_hub(
             cur = have_wp.get(w.record_id.strip().lower())
             if cur:
                 hub_rows = hub_tasks_by_workplan.get(str(cur["id"]), [])
-        # Tasks carry no canonical id on the hub, only a title, so compare on
-        # title. Imperfect, and the reason task removal needs the same explicit
-        # acknowledgement as everything else.
-        have = {(t.get("title") or "").strip().lower(): t for t in hub_rows}
-        want = {(t.title or t.record_id).strip().lower(): t for t in w.tasks}
+        # Match on the canonical record id where the hub has one. Rows created
+        # before STATE-WP-0083-T06 fall back to title, which is why those are
+        # reported rather than acted on: a renamed heading is indistinguishable
+        # from a replaced task under title matching.
+        def _task_key(record_id: str | None, title: str | None) -> str:
+            if record_id:
+                return record_id.strip().lower()
+            return "title:" + (title or "").strip().lower()
+
+        have = {
+            _task_key(t.get("record_id"), t.get("title")): t for t in hub_rows
+        }
+        want = {_task_key(t.record_id, t.title): t for t in w.tasks}
         for key, t in want.items():
             if key not in have:
                 d.missing.append(
@@ -530,6 +538,7 @@ async def reset_repository_projection(
                     Task(
                         id=uuid.UUID(t.uuid),
                         workplan_id=row.id,
+                        record_id=t.record_id,
                         title=t.title or t.record_id,
                         status=t.status or "todo",
                         priority=t.priority or "medium",

api/services/task_record_id_backfill.py

@@ -0,0 +1,116 @@
+"""Backfill canonical record ids onto existing task rows (STATE-WP-0083-T06).
+
+Only the repository files hold the mapping. A file task declares both its
+canonical id and the projection UUID it was registered under:
+
+    ```task
+    id: CUST-WP-0067-T01
+    state_hub_task_id: "f3608db4-..."
+    ```
+
+so the pairing can be read directly rather than guessed from titles. Anything a
+file does not claim is left alone: a task row whose canonical id cannot be
+established keeps `record_id` null, and the reset continues to refuse to act on
+it. An unknown identity must stay unknown rather than be inferred.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+_TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL)
+_ID_RE = re.compile(r"^id:\s*(\S+)", re.MULTILINE)
+_UUID_RE = re.compile(r'state_hub_task_id:\s*"?([0-9a-f-]{36})"?')
+
+
+@dataclass
+class BackfillReport:
+    scanned_files: int = 0
+    pairs_found: int = 0
+    updated: int = 0
+    already_set: int = 0
+    conflicts: list[dict[str, str]] = field(default_factory=list)
+    unmatched_uuids: int = 0
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "schema": "state-hub.task-record-id-backfill.v1",
+            "scanned_files": self.scanned_files,
+            "pairs_found": self.pairs_found,
+            "updated": self.updated,
+            "already_set": self.already_set,
+            "unmatched_uuids": self.unmatched_uuids,
+            "conflicts": self.conflicts,
+        }
+
+
+def collect_pairs(roots: list[Path]) -> tuple[dict[str, str], BackfillReport]:
+    """Map projection UUID -> canonical record id, from workplan files."""
+    report = BackfillReport()
+    pairs: dict[str, str] = {}
+    for root in roots:
+        wp_dir = root / "workplans"
+        if not wp_dir.is_dir():
+            continue
+        for path in sorted(wp_dir.rglob("*.md")):
+            if path.name.startswith("."):
+                continue
+            try:
+                text = path.read_text(encoding="utf-8")
+            except (OSError, UnicodeDecodeError):
+                continue
+            report.scanned_files += 1
+            for block in _TASK_BLOCK_RE.finditer(text):
+                body = block.group(1)
+                rid = _ID_RE.search(body)
+                uid = _UUID_RE.search(body)
+                if not rid or not uid:
+                    continue
+                record_id, task_uuid = rid.group(1).strip(), uid.group(1)
+                prior = pairs.get(task_uuid)
+                if prior and prior != record_id:
+                    # One UUID claimed by two canonical ids: a duplicate
+                    # registration. Recording it and skipping is the only safe
+                    # option — picking one would fabricate an identity.
+                    report.conflicts.append(
+                        {"uuid": task_uuid, "first": prior, "second": record_id}
+                    )
+                    continue
+                pairs[task_uuid] = record_id
+    report.pairs_found = len(pairs)
+    return pairs, report
+
+
+async def backfill_task_record_ids(
+    session: Any, roots: list[Path], *, dry_run: bool = True
+) -> BackfillReport:
+    from sqlalchemy import select
+
+    from api.models.task import Task
+
+    pairs, report = collect_pairs(roots)
+    if not pairs:
+        return report
+
+    rows = list((await session.execute(select(Task))).scalars())
+    by_id = {str(r.id): r for r in rows}
+    for task_uuid, record_id in pairs.items():
+        row = by_id.get(task_uuid)
+        if row is None:
+            report.unmatched_uuids += 1
+            continue
+        if row.record_id == record_id:
+            report.already_set += 1
+            continue
+        if row.record_id and row.record_id != record_id:
+            report.conflicts.append(
+                {"uuid": task_uuid, "first": row.record_id, "second": record_id}
+            )
+            continue
+        report.updated += 1
+        if not dry_run:
+            row.record_id = record_id
+    return report