C-43: flag resolved decisions whose superseded predecessor is unlinked (STATE-WP-0097-T01)

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 356991@bnt-lap001
Assistant-Session: 5d99ea69-ed55-4a1e-8d97-f4f312db247f

scripts/consistency_check.py

@@ -46,6 +46,7 @@ Checks:
   C-40  task-wait-blocker-satisfied WARN No  wait task whose depends_on targets are all terminal, whose decision_id is resolved, or whose wait_until has passed (CUST-WP-0074 rules 3/4)
   C-41  task-wait-qualifier-drift WARN Yes  needs_human/blocking_reason/decision_id/blocked_on/wait_until differ between file and DB (file wins)
   C-42  workplan-blocked-no-wait WARN No   blocked workplan has no task in status wait (CUST-WP-0074-T06)
+  C-43  decision-supersession-unlinked WARN No  resolved decision of this repo says "supersedes <id>" but the cited decision is not superseded_by it (STATE-WP-0097)
   (C-20 also indexes task-block depends_on as workplan_dependencies rows with from_task_id set)
   (finished¬DoD-Ok is listed by `statehub quality-debt`, not per-file C-warn — avoids historical flood)
 
@@ -2478,6 +2479,7 @@ def check_repo(
         if scan_source_revision(repo_dir) != (report.source_revision, None):
             report.source_revision, report.source_revision_reason = None, "changed_during_scan"
     _check_inbox_hygiene(api_base, repo_slug, report)
+    _check_unlinked_supersession(api_base, set(file_ws_ids), report)
     _check_workplan_prefixes(repo_dir, repo_slug, workplan_infos, report)
     _check_workplan_id_collisions(api_base, repo_slug, repo_dir, workplan_infos, report)
     _check_scope_freshness(repo_dir, workplan_infos, report)
@@ -3047,6 +3049,62 @@ def _check_blocked_unblock_sweep(
             )
 
 
+_SUPERSEDES_RE = re.compile(
+    r"\bsupersedes\s+(?:decision\s+)?([0-9a-f]{8}(?:-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})?)\b",
+    re.IGNORECASE,
+)
+
+
+def find_unlinked_supersessions(
+    decisions: list[dict], workplan_ids: set[str]
+) -> list[tuple[dict, dict]]:
+    """(citing, cited) pairs where a resolved decision of these workplans says
+    "supersedes <id>" but the cited decision is not superseded_by it. "Superseded
+    by" is the other direction and is not matched; ambiguous prefixes are skipped."""
+    pairs: list[tuple[dict, dict]] = []
+    for citing in decisions:
+        if str(citing.get("status") or "") != "resolved":
+            continue
+        if str(citing.get("workplan_id") or "") not in workplan_ids:
+            continue
+        text = " ".join(str(citing.get(k) or "") for k in ("title", "rationale", "description"))
+        for ref in dict.fromkeys(m.lower() for m in _SUPERSEDES_RE.findall(text)):
+            matches = [d for d in decisions if str(d.get("id") or "").lower().startswith(ref)]
+            if len(matches) != 1 or matches[0] is citing:
+                continue
+            cited = matches[0]
+            if (
+                str(cited.get("status") or "") != "superseded"
+                or str(cited.get("superseded_by") or "") != str(citing["id"])
+            ):
+                pairs.append((citing, cited))
+    return pairs
+
+
+def _check_unlinked_supersession(
+    api_base: str, workplan_ids: set[str], report: ConsistencyReport
+) -> None:
+    workplan_ids = {str(w) for w in workplan_ids if w}
+    if not workplan_ids:
+        return
+    decisions = _api_get(api_base, "/decisions/")
+    if not isinstance(decisions, list):
+        return
+    for citing, cited in find_unlinked_supersessions(decisions, workplan_ids):
+        report.add(
+            severity="WARN",
+            check_id="C-43",
+            message=(
+                f"Decision {citing['id']} supersedes {cited['id']}, but the cited "
+                f"decision is status={cited.get('status')!r} "
+                f"superseded_by={cited.get('superseded_by')!r}; the decider should "
+                f"set status=superseded and superseded_by"
+            ),
+            db_id=str(cited["id"]),
+            fixable=False,
+        )
+
+
 def _check_inbox_hygiene(api_base: str, repo_slug: str, report: ConsistencyReport) -> None:
     hygiene = collect_inbox_hygiene(api_base, repo_slug)
     if hygiene["stale_unread_count"]: