Add discovery registry review flow

railiance_fabric/registry.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import json
+import hashlib
+import re
 import sqlite3
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -50,6 +52,20 @@ class RegistryStore:
                 create index if not exists idx_snapshots_repo_latest
                   on snapshots(repo_slug, id desc);
 
+                create table if not exists discovery_snapshots (
+                  id integer primary key autoincrement,
+                  repo_slug text not null references repositories(slug),
+                  commit_sha text not null,
+                  profile text not null,
+                  generated_at text not null,
+                  snapshot_json text not null,
+                  accepted_graph_snapshot_id integer references snapshots(id),
+                  created_at text not null
+                );
+
+                create index if not exists idx_discovery_snapshots_repo_latest
+                  on discovery_snapshots(repo_slug, profile, id desc);
+
                 create table if not exists artifacts (
                   id integer primary key autoincrement,
                   repo_slug text not null references repositories(slug),
@@ -230,6 +246,166 @@ class RegistryStore:
             ).fetchall()
         return [_snapshot_dict(row) for row in rows]
 
+    def add_discovery_snapshot(self, repo_slug: str, payload: dict[str, Any]) -> dict[str, Any]:
+        self.get_repository(repo_slug)
+        if not isinstance(payload, dict):
+            raise RegistryError("discovery snapshot payload must be an object")
+        validate_discovery_snapshot(payload)
+        source = payload.get("source") if isinstance(payload.get("source"), dict) else {}
+        if source.get("repo_slug") and source.get("repo_slug") != repo_slug:
+            raise RegistryError("discovery snapshot repo_slug does not match request path")
+        commit = str(source.get("commit") or "working-tree")
+        scan = payload.get("scan") if isinstance(payload.get("scan"), dict) else {}
+        profile = str(scan.get("profile") or "deterministic")
+        generated_at = str(payload.get("generated_at") or _utc_now())
+        now = _utc_now()
+        with self._connect() as db:
+            cursor = db.execute(
+                """
+                insert into discovery_snapshots (
+                  repo_slug, commit_sha, profile, generated_at, snapshot_json,
+                  accepted_graph_snapshot_id, created_at
+                )
+                values (?, ?, ?, ?, ?, null, ?)
+                """,
+                (repo_slug, commit, profile, generated_at, json.dumps(payload, sort_keys=True), now),
+            )
+            discovery_id = int(cursor.lastrowid)
+        return self.get_discovery_snapshot(discovery_id)
+
+    def get_discovery_snapshot(self, discovery_snapshot_id: int) -> dict[str, Any]:
+        with self._connect() as db:
+            row = db.execute(
+                """
+                select id, repo_slug, commit_sha, profile, generated_at, snapshot_json,
+                       accepted_graph_snapshot_id, created_at
+                from discovery_snapshots
+                where id = ?
+                """,
+                (discovery_snapshot_id,),
+            ).fetchone()
+        if row is None:
+            raise RegistryError(f"discovery snapshot not found: {discovery_snapshot_id}", 404)
+        return _discovery_snapshot_dict(row)
+
+    def list_discovery_snapshots(self, repo_slug: str, profile: str | None = None) -> list[dict[str, Any]]:
+        self.get_repository(repo_slug)
+        params: list[Any] = [repo_slug]
+        where = "where repo_slug = ?"
+        if profile:
+            where += " and profile = ?"
+            params.append(profile)
+        with self._connect() as db:
+            rows = db.execute(
+                f"""
+                select id, repo_slug, commit_sha, profile, generated_at, snapshot_json,
+                       accepted_graph_snapshot_id, created_at
+                from discovery_snapshots
+                {where}
+                order by id desc
+                """,
+                params,
+            ).fetchall()
+        return [_discovery_snapshot_summary(row) for row in rows]
+
+    def latest_discovery_snapshot(self, repo_slug: str, profile: str | None = None) -> dict[str, Any]:
+        self.get_repository(repo_slug)
+        params: list[Any] = [repo_slug]
+        where = "where repo_slug = ?"
+        if profile:
+            where += " and profile = ?"
+            params.append(profile)
+        with self._connect() as db:
+            row = db.execute(
+                f"""
+                select id, repo_slug, commit_sha, profile, generated_at, snapshot_json,
+                       accepted_graph_snapshot_id, created_at
+                from discovery_snapshots
+                {where}
+                order by id desc
+                limit 1
+                """,
+                params,
+            ).fetchone()
+        if row is None:
+            raise RegistryError(f"no discovery snapshots for repository: {repo_slug}", 404)
+        return _discovery_snapshot_dict(row)
+
+    def discovery_snapshot_diff(
+        self,
+        repo_slug: str,
+        from_id: int | None = None,
+        to_id: int | None = None,
+        profile: str | None = None,
+    ) -> dict[str, Any]:
+        self.get_repository(repo_slug)
+        if from_id is None or to_id is None:
+            snapshots = self.list_discovery_snapshots(repo_slug, profile=profile)
+            if len(snapshots) < 2:
+                raise RegistryError(f"at least two discovery snapshots are required for diff: {repo_slug}", 404)
+            to_id = snapshots[0]["id"] if to_id is None else to_id
+            from_id = snapshots[1]["id"] if from_id is None else from_id
+
+        before = self.get_discovery_snapshot(from_id)
+        after = self.get_discovery_snapshot(to_id)
+        if before["repo_slug"] != repo_slug or after["repo_slug"] != repo_slug:
+            raise RegistryError("discovery snapshot ids must belong to the requested repository")
+
+        return {
+            "repo_slug": repo_slug,
+            "from": _discovery_snapshot_public_summary(before),
+            "to": _discovery_snapshot_public_summary(after),
+            "discovery": _discovery_diff(before["snapshot"], after["snapshot"]),
+            "reconciliation": after["snapshot"].get("reconciliation", {}),
+        }
+
+    def accept_discovery_snapshot(
+        self,
+        repo_slug: str,
+        discovery_snapshot_id: int,
+        payload: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        payload = payload or {}
+        discovery = self.get_discovery_snapshot(discovery_snapshot_id)
+        if discovery["repo_slug"] != repo_slug:
+            raise RegistryError("discovery snapshot id must belong to the requested repository")
+        base_graph = self._latest_graph_or_empty(repo_slug)
+        accepted_keys = payload.get("accepted_keys")
+        if accepted_keys is not None and not isinstance(accepted_keys, list):
+            raise RegistryError("field 'accepted_keys' must be an array when provided")
+        accepted_key_set = {str(key) for key in accepted_keys or []}
+        accept_review_states = payload.get("accept_review_states", ["accepted"])
+        if not isinstance(accept_review_states, list):
+            raise RegistryError("field 'accept_review_states' must be an array")
+        graph = _project_discovery_snapshot(
+            base_graph,
+            discovery["snapshot"],
+            accepted_keys=accepted_key_set,
+            accept_review_states={str(state) for state in accept_review_states},
+        )
+        snapshot = self.add_snapshot(
+            repo_slug,
+            {
+                "commit": str(payload.get("commit") or f"discovery:{discovery['commit']}"),
+                "generated_at": _utc_now(),
+                "graph": graph,
+            },
+        )
+        with self._connect() as db:
+            db.execute(
+                "update discovery_snapshots set accepted_graph_snapshot_id = ? where id = ?",
+                (snapshot["id"], discovery_snapshot_id),
+            )
+        return {
+            "repo_slug": repo_slug,
+            "discovery_snapshot": self.get_discovery_snapshot(discovery_snapshot_id),
+            "graph_snapshot": snapshot,
+            "projected": {
+                "node_count": len(graph.get("nodes", [])),
+                "edge_count": len(graph.get("edges", [])),
+            },
+        }
+
     def combined_graph(self) -> dict[str, Any]:
         nodes: dict[str, dict[str, Any]] = {}
         edges: list[dict[str, str]] = []
@@ -358,17 +534,30 @@ class RegistryStore:
             if exc.status_code != 404:
                 raise
             latest_snapshot = None
+        try:
+            latest_discovery_snapshot = self.latest_discovery_snapshot(repo_slug)
+        except RegistryError as exc:
+            if exc.status_code != 404:
+                raise
+            latest_discovery_snapshot = None
 
         graph = latest_snapshot["graph"] if latest_snapshot else _empty_graph()
         nodes = [node for node in graph.get("nodes", []) if isinstance(node, dict)]
         edges = [edge for edge in graph.get("edges", []) if isinstance(edge, dict)]
         artifacts = self.list_artifacts(repo_slug=repo_slug)
         libraries = self.list_libraries(repo_slug=repo_slug)
+        discovery_snapshots = self.list_discovery_snapshots(repo_slug)
         return {
             "repository": repository,
             "latest_snapshot": _snapshot_public_summary(latest_snapshot) if latest_snapshot else None,
+            "latest_discovery_snapshot": (
+                _discovery_snapshot_public_summary(latest_discovery_snapshot)
+                if latest_discovery_snapshot
+                else None
+            ),
             "counts": {
                 "snapshots": len(self.list_snapshots(repo_slug)),
+                "discovery_snapshots": len(discovery_snapshots),
                 "nodes": len(nodes),
                 "edges": len(edges),
                 "artifacts": len(artifacts),
@@ -534,6 +723,7 @@ class RegistryStore:
             counts = {
                 "repositories": db.execute("select count(*) from repositories").fetchone()[0],
                 "snapshots": db.execute("select count(*) from snapshots").fetchone()[0],
+                "discovery_snapshots": db.execute("select count(*) from discovery_snapshots").fetchone()[0],
                 "artifacts": db.execute("select count(*) from artifacts").fetchone()[0],
                 "libraries": db.execute("select count(*) from libraries").fetchone()[0],
             }
@@ -553,6 +743,14 @@ class RegistryStore:
             "latest_snapshots": latest,
         }
 
+    def _latest_graph_or_empty(self, repo_slug: str) -> dict[str, Any]:
+        try:
+            return self.latest_snapshot(repo_slug)["graph"]
+        except RegistryError as exc:
+            if exc.status_code != 404:
+                raise
+            return _empty_graph()
+
     def _connect(self) -> sqlite3.Connection:
         db = sqlite3.connect(self.path)
         db.row_factory = sqlite3.Row
@@ -569,6 +767,16 @@ def validate_graph_export(graph: dict[str, Any]) -> None:
         raise RegistryError(f"invalid FabricGraphExport at {location}: {error.message}")
 
 
+def validate_discovery_snapshot(snapshot: dict[str, Any]) -> None:
+    schema_path = repo_root() / "schemas" / "discovery-snapshot.schema.yaml"
+    validator = draft202012_validator(schema_path)
+    errors = sorted(validator.iter_errors(snapshot), key=lambda error: list(error.path))
+    if errors:
+        error = errors[0]
+        location = ".".join(str(part) for part in error.path) or "<root>"
+        raise RegistryError(f"invalid FabricDiscoverySnapshot at {location}: {error.message}")
+
+
 def providers(graph: dict[str, Any], capability: str) -> list[dict[str, Any]]:
     result = []
     for node in _nodes(graph):
@@ -911,6 +1119,79 @@ def _snapshot_public_summary(snapshot: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _discovery_snapshot_dict(row: sqlite3.Row) -> dict[str, Any]:
+    snapshot = json.loads(row["snapshot_json"])
+    return {
+        "id": row["id"],
+        "repo_slug": row["repo_slug"],
+        "commit": row["commit_sha"],
+        "profile": row["profile"],
+        "generated_at": row["generated_at"],
+        "snapshot": snapshot,
+        "accepted_graph_snapshot_id": row["accepted_graph_snapshot_id"],
+        "created_at": row["created_at"],
+    }
+
+
+def _discovery_snapshot_summary(row: sqlite3.Row) -> dict[str, Any]:
+    snapshot = json.loads(row["snapshot_json"])
+    candidates = snapshot.get("candidates") if isinstance(snapshot.get("candidates"), dict) else {}
+    reconciliation = snapshot.get("reconciliation") if isinstance(snapshot.get("reconciliation"), dict) else {}
+    diff = reconciliation.get("diff") if isinstance(reconciliation.get("diff"), dict) else {}
+    return {
+        "id": row["id"],
+        "repo_slug": row["repo_slug"],
+        "commit": row["commit_sha"],
+        "profile": row["profile"],
+        "generated_at": row["generated_at"],
+        "accepted_graph_snapshot_id": row["accepted_graph_snapshot_id"],
+        "created_at": row["created_at"],
+        "candidate_counts": {
+            "nodes": len(candidates.get("nodes", [])) if isinstance(candidates.get("nodes"), list) else 0,
+            "edges": len(candidates.get("edges", [])) if isinstance(candidates.get("edges"), list) else 0,
+            "attributes": len(candidates.get("attributes", [])) if isinstance(candidates.get("attributes"), list) else 0,
+        },
+        "diff_counts": {
+            "added": len(diff.get("added", [])) if isinstance(diff.get("added"), list) else 0,
+            "changed": len(diff.get("changed", [])) if isinstance(diff.get("changed"), list) else 0,
+            "retired": len(diff.get("retired", [])) if isinstance(diff.get("retired"), list) else 0,
+            "conflicted": len(diff.get("conflicted", [])) if isinstance(diff.get("conflicted"), list) else 0,
+        },
+        "review_artifact_count": len(snapshot.get("review_artifacts", []))
+        if isinstance(snapshot.get("review_artifacts"), list)
+        else 0,
+        "connector_run_count": len(snapshot.get("connector_runs", []))
+        if isinstance(snapshot.get("connector_runs"), list)
+        else 0,
+    }
+
+
+def _discovery_snapshot_public_summary(snapshot: dict[str, Any]) -> dict[str, Any]:
+    candidates = snapshot["snapshot"].get("candidates", {})
+    reconciliation = snapshot["snapshot"].get("reconciliation", {})
+    diff = reconciliation.get("diff", {}) if isinstance(reconciliation, dict) else {}
+    return {
+        "id": snapshot["id"],
+        "repo_slug": snapshot["repo_slug"],
+        "commit": snapshot["commit"],
+        "profile": snapshot["profile"],
+        "generated_at": snapshot["generated_at"],
+        "accepted_graph_snapshot_id": snapshot["accepted_graph_snapshot_id"],
+        "created_at": snapshot["created_at"],
+        "candidate_counts": {
+            "nodes": len(candidates.get("nodes", [])) if isinstance(candidates.get("nodes"), list) else 0,
+            "edges": len(candidates.get("edges", [])) if isinstance(candidates.get("edges"), list) else 0,
+            "attributes": len(candidates.get("attributes", [])) if isinstance(candidates.get("attributes"), list) else 0,
+        },
+        "diff_counts": {
+            "added": len(diff.get("added", [])) if isinstance(diff.get("added"), list) else 0,
+            "changed": len(diff.get("changed", [])) if isinstance(diff.get("changed"), list) else 0,
+            "retired": len(diff.get("retired", [])) if isinstance(diff.get("retired"), list) else 0,
+            "conflicted": len(diff.get("conflicted", [])) if isinstance(diff.get("conflicted"), list) else 0,
+        },
+    }
+
+
 def _row_dict(row: sqlite3.Row) -> dict[str, Any]:
     return {key: row[key] for key in row.keys()}
 
@@ -1027,6 +1308,186 @@ def _empty_graph() -> dict[str, Any]:
     }
 
 
+def _discovery_diff(before: dict[str, Any], after: dict[str, Any]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for collection in ("nodes", "edges", "attributes"):
+        before_items = _discovery_candidates_by_key(before, collection)
+        after_items = _discovery_candidates_by_key(after, collection)
+        added = sorted(set(after_items) - set(before_items))
+        removed = sorted(set(before_items) - set(after_items))
+        common = sorted(set(before_items) & set(after_items))
+        changed = [
+            key for key in common
+            if _stable_json(_without_review_noise(before_items[key])) != _stable_json(_without_review_noise(after_items[key]))
+        ]
+        confidence_changed = [
+            {
+                "stable_key": key,
+                "before": before_items[key].get("confidence"),
+                "after": after_items[key].get("confidence"),
+            }
+            for key in common
+            if before_items[key].get("confidence") != after_items[key].get("confidence")
+        ]
+        result[collection] = {
+            "added": [after_items[key] for key in added],
+            "removed": [before_items[key] for key in removed],
+            "changed": [
+                {
+                    "stable_key": key,
+                    "before": before_items[key],
+                    "after": after_items[key],
+                }
+                for key in changed
+            ],
+            "confidence_changed": confidence_changed,
+        }
+    after_reconciliation = after.get("reconciliation") if isinstance(after.get("reconciliation"), dict) else {}
+    result["review"] = {
+        "retired": after_reconciliation.get("diff", {}).get("retired", [])
+        if isinstance(after_reconciliation.get("diff"), dict)
+        else [],
+        "conflicted": after_reconciliation.get("diff", {}).get("conflicted", [])
+        if isinstance(after_reconciliation.get("diff"), dict)
+        else [],
+        "conflicts": after_reconciliation.get("conflicts", []),
+    }
+    return result
+
+
+def _project_discovery_snapshot(
+    base_graph: dict[str, Any],
+    discovery_snapshot: dict[str, Any],
+    *,
+    accepted_keys: set[str],
+    accept_review_states: set[str],
+) -> dict[str, Any]:
+    graph = json.loads(json.dumps(base_graph))
+    graph.setdefault("apiVersion", "railiance.fabric/v1alpha1")
+    graph.setdefault("kind", "FabricGraphExport")
+    graph.setdefault("nodes", [])
+    graph.setdefault("edges", [])
+    candidates = discovery_snapshot.get("candidates") if isinstance(discovery_snapshot.get("candidates"), dict) else {}
+    candidate_nodes = [
+        node for node in candidates.get("nodes", [])
+        if isinstance(node, dict) and _candidate_is_accepted(node, accepted_keys, accept_review_states)
+    ]
+    candidate_edges = [
+        edge for edge in candidates.get("edges", [])
+        if isinstance(edge, dict) and _candidate_is_accepted(edge, accepted_keys, accept_review_states)
+    ]
+    existing_nodes = {
+        str(node.get("id")): node
+        for node in graph.get("nodes", [])
+        if isinstance(node, dict) and node.get("id")
+    }
+    key_to_graph_id = {
+        str(node.get("id")): str(node.get("id"))
+        for node in graph.get("nodes", [])
+        if isinstance(node, dict) and node.get("id")
+    }
+    for candidate in candidate_nodes:
+        graph_id = _candidate_graph_id(candidate)
+        key_to_graph_id[str(candidate.get("stable_key"))] = graph_id
+        if candidate.get("graph_id"):
+            key_to_graph_id[str(candidate.get("graph_id"))] = graph_id
+        if graph_id in existing_nodes:
+            continue
+        projected = _project_candidate_node(candidate, graph_id)
+        existing_nodes[graph_id] = projected
+        graph["nodes"].append(projected)
+
+    existing_edges = {
+        _edge_key(edge)
+        for edge in graph.get("edges", [])
+        if isinstance(edge, dict)
+    }
+    for candidate in candidate_edges:
+        source = key_to_graph_id.get(str(candidate.get("source_key") or ""))
+        target = key_to_graph_id.get(str(candidate.get("target_key") or ""))
+        if not source or not target:
+            continue
+        edge = {"from": source, "to": target, "type": str(candidate.get("edge_type") or "")}
+        edge_key = _edge_key(edge)
+        if edge["type"] and edge_key not in existing_edges:
+            existing_edges.add(edge_key)
+            graph["edges"].append(edge)
+    validate_graph_export(graph)
+    return graph
+
+
+def _project_candidate_node(candidate: dict[str, Any], graph_id: str) -> dict[str, Any]:
+    attributes = candidate.get("attributes") if isinstance(candidate.get("attributes"), dict) else {}
+    return {
+        "id": graph_id,
+        "kind": str(candidate.get("kind") or "DiscoveredEntity"),
+        "name": str(candidate.get("label") or graph_id),
+        "repo": str(candidate.get("repo") or ""),
+        "domain": str(candidate.get("domain") or ""),
+        "lifecycle": str(candidate.get("lifecycle") or "active"),
+        "attributes": {
+            **attributes,
+            "discovery_stable_key": candidate.get("stable_key"),
+            "discovery_origin": candidate.get("origin"),
+            "discovery_review_state": candidate.get("review_state"),
+            "discovery_confidence": candidate.get("confidence"),
+            "discovery_source_anchors": candidate.get("source_anchors", []),
+            "discovery_provenance": candidate.get("provenance", []),
+        },
+    }
+
+
+def _candidate_graph_id(candidate: dict[str, Any]) -> str:
+    graph_id = str(candidate.get("graph_id") or "").strip()
+    if graph_id:
+        return graph_id
+    repo = _graph_id_part(str(candidate.get("repo") or "repo"))
+    kind = _graph_id_part(str(candidate.get("kind") or "entity"))
+    label = _graph_id_part(str(candidate.get("label") or candidate.get("stable_key") or "candidate"))
+    candidate_id = f"{repo}.{kind}.{label}".strip(".")
+    if len(candidate_id) > 150:
+        digest = hashlib.sha256(str(candidate.get("stable_key") or candidate_id).encode("utf-8")).hexdigest()[:10]
+        candidate_id = f"{candidate_id[:139].rstrip('.')}.{digest}"
+    if len(candidate_id) < 3:
+        candidate_id = f"{candidate_id}.id"
+    return candidate_id
+
+
+def _graph_id_part(value: str) -> str:
+    text = re.sub(r"[^a-z0-9.-]+", "-", value.lower()).strip(".-")
+    text = re.sub(r"-+", "-", text)
+    text = re.sub(r"\.+", ".", text)
+    return text or "unknown"
+
+
+def _candidate_is_accepted(
+    candidate: dict[str, Any],
+    accepted_keys: set[str],
+    accept_review_states: set[str],
+) -> bool:
+    stable_key = str(candidate.get("stable_key") or "")
+    if stable_key in accepted_keys:
+        return True
+    return str(candidate.get("review_state") or "") in accept_review_states
+
+
+def _discovery_candidates_by_key(snapshot: dict[str, Any], collection: str) -> dict[str, dict[str, Any]]:
+    candidates = snapshot.get("candidates") if isinstance(snapshot.get("candidates"), dict) else {}
+    return {
+        str(item["stable_key"]): item
+        for item in candidates.get(collection, [])
+        if isinstance(item, dict) and item.get("stable_key")
+    }
+
+
+def _without_review_noise(candidate: dict[str, Any]) -> dict[str, Any]:
+    return {
+        key: value
+        for key, value in candidate.items()
+        if key not in {"provenance"}
+    }
+
+
 def _graph_diff(before: dict[str, Any], after: dict[str, Any]) -> dict[str, Any]:
     before_nodes = _nodes_by_id(before)
     after_nodes = _nodes_by_id(after)