feat: canonicalize duplicate repo identities

railiance_fabric/accountability_roots.py

@@ -611,17 +611,28 @@ def _identity_from_evidence(root: dict[str, Any], item: dict[str, Any]) -> dict[
     owner_actor_id = str(root.get("owner_actor_id") or "")
 
     if evidence_type in {"registered_repository", "repository_checkout"}:
-        label = str(source.get("repo_slug") or attributes.get("repo_slug") or Path(str(source.get("path") or "")).name)
+        declared_slug = str(source.get("repo_slug") or attributes.get("repo_slug") or "")
+        identity_slug = str(
+            source.get("identity_slug")
+            or attributes.get("canonical_slug")
+            or attributes.get("canonical_repo_slug")
+            or declared_slug
+            or Path(str(source.get("path") or "")).name
+        )
         return {
             "identity_type": "Repository",
-            "label": label,
-            "graph_id": label,
+            "label": identity_slug,
+            "graph_id": identity_slug,
             "fabric_id": fabric_id,
             "subfabric_id": subfabric_id,
             "owner_actor_id": owner_actor_id,
             "evidence_ids": evidence_ids,
-            "aliases": [label, str(source.get("path") or ""), str(source.get("remote_url") or "")],
-            "attributes": {**attributes, "source_evidence_type": evidence_type},
+            "aliases": [identity_slug, declared_slug, str(source.get("path") or ""), str(source.get("remote_url") or "")],
+            "attributes": {
+                **attributes,
+                "declared_repo_slug": declared_slug,
+                "source_evidence_type": evidence_type,
+            },
             "confidence": 0.9 if evidence_type == "repository_checkout" else 0.85,
         }
     if evidence_type in {"deployment_automation", "infrastructure_manifest"}:
@@ -1029,13 +1040,17 @@ def _registry_manifest_evidence(root: dict[str, Any], *, max_items: int) -> list
     evidence: list[dict[str, Any]] = [
         _file_evidence(root, manifest_path, "registry_manifest", summary=f"Registry manifest with {len(repositories)} repositories.")
     ]
+    canonical_slug_by_index = _registry_canonical_slugs_by_index(repositories)
     for index, repo in enumerate(repositories[:max_items]):
         if not isinstance(repo, dict):
             continue
+        repo_slug = str(repo.get("slug") or "")
+        identity_slug = canonical_slug_by_index.get(index) or _repository_identity_slug(repo) or repo_slug
         repo_source = {
             "manifest_path": _display_path(manifest_path),
             "json_pointer": f"/repositories/{index}",
-            "repo_slug": repo.get("slug", ""),
+            "repo_slug": repo_slug,
+            "identity_slug": identity_slug,
             "path": repo.get("path", ""),
             "remote_url": repo.get("remote_url", ""),
         }
@@ -1046,6 +1061,8 @@ def _registry_manifest_evidence(root: dict[str, Any], *, max_items: int) -> list
             "state_hub_repo_id": repo.get("state_hub_repo_id", ""),
             "has_local_path": bool(repo.get("path")),
             "has_remote_url": bool(repo.get("remote_url")),
+            "canonical_slug": identity_slug if identity_slug != repo_slug else "",
+            "identity_resolution": "duplicate_path_alias" if identity_slug != repo_slug else "",
         }
         evidence.append(
             _evidence_item(
@@ -1062,6 +1079,69 @@ def _registry_manifest_evidence(root: dict[str, Any], *, max_items: int) -> list
     return evidence
 
 
+def _registry_canonical_slugs_by_index(repositories: list[object]) -> dict[int, str]:
+    explicit: dict[int, str] = {}
+    path_groups: dict[str, list[tuple[int, dict[str, Any]]]] = {}
+    for index, repo in enumerate(repositories):
+        if not isinstance(repo, dict):
+            continue
+        explicit_slug = _repository_identity_slug(repo)
+        if explicit_slug:
+            explicit[index] = explicit_slug
+        if _boolish(repo.get("split_identity")):
+            continue
+        path = str(repo.get("path") or "").strip()
+        slug = str(repo.get("slug") or "").strip()
+        if path and slug:
+            path_groups.setdefault(path, []).append((index, repo))
+
+    canonical_by_index = dict(explicit)
+    for path, group in path_groups.items():
+        if len(group) < 2:
+            continue
+        explicit_group_slugs = [
+            explicit_slug
+            for index, _repo in group
+            if (explicit_slug := explicit.get(index))
+        ]
+        canonical_slug = explicit_group_slugs[0] if explicit_group_slugs else _select_canonical_repo_slug(path, group)
+        for index, _repo in group:
+            canonical_by_index.setdefault(index, canonical_slug)
+    return canonical_by_index
+
+
+def _repository_identity_slug(repo: dict[str, Any]) -> str:
+    return str(
+        repo.get("canonical_slug")
+        or repo.get("canonical_repo_slug")
+        or repo.get("identity_slug")
+        or repo.get("identity_repo_slug")
+        or ""
+    ).strip()
+
+
+def _select_canonical_repo_slug(path: str, group: list[tuple[int, dict[str, Any]]]) -> str:
+    path_name = normalize_identity_part(Path(path).name)
+
+    def sort_key(item: tuple[int, dict[str, Any]]) -> tuple[int, int, str]:
+        _index, repo = item
+        slug = str(repo.get("slug") or "")
+        return (
+            0 if normalize_identity_part(slug) == path_name else 1,
+            0 if repo.get("remote_url") else 1,
+            normalize_identity_part(slug),
+        )
+
+    _index, canonical = sorted(group, key=sort_key)[0]
+    return str(canonical.get("slug") or "").strip()
+
+
+def _boolish(value: object) -> bool:
+    if isinstance(value, bool):
+        return value
+    return str(value or "").strip().lower() in {"1", "true", "yes", "on"}
+
+
 def _repository_checkout_evidence(root: dict[str, Any]) -> list[dict[str, Any]]:
     source = _source(root)
     checkout = _resolve_path(source.get("path"))