diff --git a/api/services/task_record_id_backfill.py b/api/services/task_record_id_backfill.py
index c7d43ec..c2ff3c4 100644
--- a/api/services/task_record_id_backfill.py
+++ b/api/services/task_record_id_backfill.py
@@ -24,6 +24,23 @@ 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})"?')
+_FRONTMATTER_ID_RE = re.compile(r"^id:\s*(\S+)", re.MULTILINE)
+# A task id written as a bare "T01" is unique only inside its own workplan.
+# Storing it as a canonical identifier makes every workplan's first task share
+# one identity — 51 such ids were assigned to 148 rows before this was caught.
+_SHORT_TASK_ID_RE = re.compile(r"^T\d+$", re.IGNORECASE)
+
+
+def qualify_task_id(task_id: str, workplan_id: str | None) -> str | None:
+    """Return a canonical task id, or None when identity cannot be established."""
+    task_id = task_id.strip()
+    if not _SHORT_TASK_ID_RE.fullmatch(task_id):
+        return task_id
+    if not workplan_id:
+        # Unqualifiable: leaving it unidentified is correct, since an identity
+        # that is not unique is worse than none.
+        return None
+    return f"{workplan_id.strip()}-{task_id.upper()}"
 
 
 @dataclass
@@ -63,13 +80,19 @@ def collect_pairs(roots: list[Path]) -> tuple[dict[str, str], BackfillReport]:
             except (OSError, UnicodeDecodeError):
                 continue
             report.scanned_files += 1
+            head = text.split("---", 2)[1] if text.startswith("---") and text.count("---") >= 2 else ""
+            wp_match = _FRONTMATTER_ID_RE.search(head)
+            workplan_id = wp_match.group(1).strip() if wp_match else None
             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)
+                record_id = qualify_task_id(rid.group(1), workplan_id)
+                if record_id is None:
+                    continue
+                task_uuid = uid.group(1)
                 prior = pairs.get(task_uuid)
                 if prior and prior != record_id:
                     # One UUID claimed by two canonical ids: a duplicate
diff --git a/tests/test_task_record_id_backfill.py b/tests/test_task_record_id_backfill.py
index 9eff772..0b2f905 100644
--- a/tests/test_task_record_id_backfill.py
+++ b/tests/test_task_record_id_backfill.py
@@ -121,3 +121,41 @@ async def test_existing_identity_is_never_overwritten(tmp_path):
     rep = await bf.backfill_task_record_ids(_Session([row]), [root], dry_run=False)
     assert row.record_id == "OTHER-WP-0001-T09"
     assert rep.conflicts and rep.updated == 0
+
+
+def test_a_bare_task_id_is_qualified_by_its_workplan(tmp_path):
+    """`T01` is unique inside one workplan and meaningless outside it.
+
+    Stored raw, every workplan's first task shares one identity — 51 such ids
+    were assigned to 148 rows on central before this was caught.
+    """
+    root = _repo(tmp_path, "a", (
+        "---\nid: LLM-WP-0001\ntype: workplan\n---\n\n"
+        '```task\nid: T01\nstate_hub_task_id: "11111111-1111-5111-8111-111111111111"\n```\n'
+    ))
+    pairs, _ = bf.collect_pairs([root])
+    assert pairs == {"11111111-1111-5111-8111-111111111111": "LLM-WP-0001-T01"}
+
+
+def test_two_workplans_first_tasks_do_not_collide(tmp_path):
+    a = _repo(tmp_path, "a", (
+        "---\nid: A-WP-0001\ntype: workplan\n---\n\n"
+        '```task\nid: T01\nstate_hub_task_id: "11111111-1111-5111-8111-111111111111"\n```\n'
+    ))
+    b = _repo(tmp_path, "b", (
+        "---\nid: B-WP-0001\ntype: workplan\n---\n\n"
+        '```task\nid: T01\nstate_hub_task_id: "22222222-2222-5222-8222-222222222222"\n```\n'
+    ))
+    pairs, rep = bf.collect_pairs([a, b])
+    assert set(pairs.values()) == {"A-WP-0001-T01", "B-WP-0001-T01"}
+    assert rep.conflicts == []
+
+
+def test_unqualifiable_short_id_gets_no_identity(tmp_path):
+    """No identity beats a non-unique one."""
+    root = _repo(tmp_path, "a", (
+        "id: not-frontmatter\n\n"
+        '```task\nid: T01\nstate_hub_task_id: "11111111-1111-5111-8111-111111111111"\n```\n'
+    ))
+    pairs, _ = bf.collect_pairs([root])
+    assert pairs == {}