Discover runtime topology facts

railiance_fabric/reconciliation.py

@@ -16,7 +16,32 @@ ORIGIN_PRECEDENCE = {
     "manual": 5,
 }
 
-PATH_SCOPED_NODE_KINDS = {"lockfile"}
+PATH_SCOPED_NODE_KINDS = {
+    "container-build",
+    "deployment-service",
+    "domain-name",
+    "kubernetes-config-map",
+    "kubernetes-cron-job",
+    "kubernetes-daemon-set",
+    "kubernetes-deployment",
+    "kubernetes-horizontal-pod-autoscaler",
+    "kubernetes-ingress",
+    "kubernetes-job",
+    "kubernetes-namespace",
+    "kubernetes-secret",
+    "kubernetes-service",
+    "kubernetes-stateful-set",
+    "lockfile",
+    "network-port",
+    "score-workload",
+    "service-config",
+}
+EVIDENCE_AGGREGATE_EDGE_TYPES = {
+    "exposes_port",
+    "opens_port",
+    "resolves_to",
+    "routes_to_port",
+}
 
 
 def reconcile_discovery_snapshots(
@@ -186,9 +211,12 @@ def _edge_conflicts(by_key: dict[str, dict[str, Any]]) -> list[dict[str, object]
     conflicts: list[dict[str, object]] = []
     seen: dict[tuple[str, str, str], str] = {}
     for key, edge in sorted(by_key.items()):
+        edge_type = str(edge.get("edge_type") or "")
+        if edge_type in EVIDENCE_AGGREGATE_EDGE_TYPES:
+            continue
         match_key = (
             str(edge.get("source_key") or ""),
-            str(edge.get("edge_type") or ""),
+            edge_type,
             str(edge.get("target_key") or ""),
         )
         other = seen.get(match_key)
@@ -292,11 +320,24 @@ def _path_scoped_nodes_are_distinct(left: dict[str, Any] | None, right: dict[str
     right_kind = normalize_identity_part(str(right.get("kind") or ""))
     if left_kind != right_kind or left_kind not in PATH_SCOPED_NODE_KINDS:
         return False
+    left_identities = _source_anchor_identities(left)
+    right_identities = _source_anchor_identities(right)
+    if left_identities and right_identities:
+        return left_identities.isdisjoint(right_identities)
     left_paths = _source_anchor_paths(left)
     right_paths = _source_anchor_paths(right)
     return bool(left_paths and right_paths and left_paths.isdisjoint(right_paths))
 
 
+def _source_anchor_identities(candidate: dict[str, Any]) -> set[str]:
+    anchors = candidate.get("source_anchors") if isinstance(candidate.get("source_anchors"), list) else []
+    return {
+        str(anchor.get("fingerprint") or "")
+        for anchor in anchors
+        if isinstance(anchor, dict) and anchor.get("fingerprint")
+    }
+
+
 def _source_anchor_paths(candidate: dict[str, Any]) -> set[str]:
     anchors = candidate.get("source_anchors") if isinstance(candidate.get("source_anchors"), list) else []
     return {