fix(registry): reject owner index drift and duplicate discovery rows

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

reuse_surface/cli.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+from collections import Counter
 import json
 import sys
 from pathlib import Path
@@ -87,6 +88,7 @@ from reuse_surface.registry import (
     parse_vector,
     registry_paths,
     resolve_repo_root,
+    vectors_match,
 )
 
 
@@ -114,12 +116,48 @@ def _check_index_drift(
     repo_root: Path,
 ) -> list[str]:
     warnings: list[str] = []
-    indexed_paths = {item["path"] for item in index.get("capabilities", [])}
+    if not isinstance(index, dict) or not isinstance(index.get("capabilities"), list):
+        return ["index drift: index must contain a capabilities list"]
+    rows = []
+    for position, row in enumerate(index["capabilities"]):
+        if not isinstance(row, dict) or any(
+            not isinstance(row.get(field), str) or not row[field]
+            for field in ("id", "path")
+        ):
+            warnings.append(f"index drift: row {position} requires nonempty string id and path")
+            continue
+        rows.append(row)
+    indexed_paths = {item["path"] for item in rows}
     file_paths = {str(path.relative_to(repo_root)) for path in entry_paths}
     for path in sorted(file_paths - indexed_paths):
         warnings.append(f"index drift: entry file not indexed: {path}")
     for path in sorted(indexed_paths - file_paths):
         warnings.append(f"index drift: index references missing file: {path}")
+    for field in ("id", "path"):
+        for value, count in sorted(Counter(row[field] for row in rows).items()):
+            if count > 1:
+                warnings.append(f"index drift: duplicate {field}: {value} ({count} rows)")
+    for row in rows:
+        path = repo_root / row["path"]
+        if not path.is_file():
+            continue
+        try:
+            entry = parse_front_matter(path)
+        except (ValueError, yaml.YAMLError):
+            continue  # The entry validation pass reports malformed source.
+        # The index projects the owner entry; index-only changes cannot replace it.
+        for field in ("id", "name", "summary", "domain", "status", "owner", "tags"):
+            if row.get(field) != entry.get(field):
+                warnings.append(f"index drift: {row['path']}: {field} differs from entry")
+        availability = entry.get("availability")
+        modes = availability.get("consumption_modes") if isinstance(availability, dict) else None
+        if row.get("consumption_modes") != modes:
+            warnings.append(f"index drift: {row['path']}: consumption_modes differs from entry")
+        try:
+            if not vectors_match(row["vector"], entry):
+                warnings.append(f"index drift: {row['path']}: vector differs from entry")
+        except (KeyError, TypeError, AttributeError):
+            warnings.append(f"index drift: {row['path']}: vector cannot be compared to entry")
     return warnings
 
 
@@ -145,7 +183,7 @@ def _run_validate(
     for path in paths:
         try:
             data = parse_front_matter(path)
-        except ValueError as exc:
+        except (ValueError, yaml.YAMLError) as exc:
             errors.append(str(exc))
             continue
         for error in sorted(validator.iter_errors(data), key=lambda e: e.path):
@@ -155,8 +193,12 @@ def _run_validate(
     if not target:
         index_path = registry_paths(repo_root)["index"]
         if index_path.exists():
-            index = load_index_at(index_path)
-            warnings.extend(_check_index_drift(paths, index, repo_root))
+            try:
+                index = load_index_at(index_path)
+            except (ValueError, yaml.YAMLError) as exc:
+                errors.append(f"{index_path}: {exc}")
+            else:
+                warnings.extend(_check_index_drift(paths, index, repo_root))
         if relations and repo_root == ROOT:
             warnings.extend(check_relations())
     return errors, warnings, paths