Fix infra topology query-sensitive ranking

potpie/context-engine/application/readers/_common.py

@@ -121,6 +121,20 @@ def claim_corroboration(row: ClaimRow) -> int:
     return 1
 
 
+def claim_semantic_similarity(row: ClaimRow) -> float | None:
+    """Backend-stamped query similarity, when the read carried a ``fact_query``.
+
+    Returns ``None`` when the backend did not stamp a score so the ranker
+    falls back to its neutral default instead of a fabricated value.
+    """
+    sim = row.properties.get("semantic_similarity")
+    if isinstance(sim, bool):
+        return None
+    if isinstance(sim, (int, float)):
+        return float(sim)
+    return None
+
+
 def claim_environment(row: ClaimRow) -> str | None:
     env = row.environment
     if isinstance(env, str) and env.strip():
@@ -209,6 +223,7 @@ def row_in_anchor_set(row: ClaimRow, anchor_keys: Iterable[str]) -> bool:
     "claim_corroboration",
     "claim_environment",
     "claim_payload",
+    "claim_semantic_similarity",
     "coverage_status_from_count",
     "dedupe_claim_rows",
     "make_task_context",

potpie/context-engine/application/readers/coding_preferences.py

@@ -23,6 +23,7 @@
     claim_candidate_key,
     claim_corroboration,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -64,7 +65,6 @@ def read(self, req: ReadRequest) -> ReadResponse:
                 # Hard zero on overlap: skip — readers should not surface
                 # rules that demonstrably don't apply.
                 continue
-            sim = row.properties.get("semantic_similarity")
             candidates.append(
                 Candidate(
                     candidate_key=claim_candidate_key(row),
@@ -73,9 +73,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=row.valid_at,
                     corroboration_count=claim_corroboration(row),
                     scope_overlap=overlap if scope_keys else None,
-                    semantic_similarity=float(sim)
-                    if isinstance(sim, (int, float))
-                    else None,
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 

potpie/context-engine/application/readers/decisions.py

@@ -16,6 +16,7 @@
     claim_candidate_key,
     claim_corroboration,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -48,7 +49,6 @@ def read(self, req: ReadRequest) -> ReadResponse:
             overlap = _scope_overlap(row, anchor_keys=anchor_keys)
             if anchor_keys and overlap == 0.0:
                 continue
-            sim = row.properties.get("semantic_similarity")
             candidates.append(
                 Candidate(
                     candidate_key=claim_candidate_key(row),
@@ -57,9 +57,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=row.valid_at,
                     corroboration_count=claim_corroboration(row),
                     scope_overlap=overlap if anchor_keys else None,
-                    semantic_similarity=float(sim)
-                    if isinstance(sim, (int, float))
-                    else None,
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 

potpie/context-engine/application/readers/docs.py

@@ -16,6 +16,7 @@
     claim_candidate_key,
     claim_corroboration,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -45,7 +46,6 @@ def read(self, req: ReadRequest) -> ReadResponse:
             overlap = _scope_overlap(row, anchor_keys=anchor_keys)
             if anchor_keys and overlap == 0.0:
                 continue
-            sim = row.properties.get("semantic_similarity")
             candidates.append(
                 Candidate(
                     candidate_key=claim_candidate_key(row),
@@ -54,9 +54,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=row.valid_at,
                     corroboration_count=claim_corroboration(row),
                     scope_overlap=overlap if anchor_keys else None,
-                    semantic_similarity=float(sim)
-                    if isinstance(sim, (int, float))
-                    else None,
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 

potpie/context-engine/application/readers/features.py

@@ -16,6 +16,7 @@
     claim_candidate_key,
     claim_corroboration,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -48,7 +49,6 @@ def read(self, req: ReadRequest) -> ReadResponse:
             overlap = _scope_overlap(row, anchor_keys=anchor_keys)
             if anchor_keys and overlap == 0.0:
                 continue
-            sim = row.properties.get("semantic_similarity")
             candidates.append(
                 Candidate(
                     candidate_key=claim_candidate_key(row),
@@ -57,9 +57,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=row.valid_at,
                     corroboration_count=claim_corroboration(row),
                     scope_overlap=overlap if anchor_keys else None,
-                    semantic_similarity=float(sim)
-                    if isinstance(sim, (int, float))
-                    else None,
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 

potpie/context-engine/application/readers/infra_topology.py

@@ -11,10 +11,16 @@
 When a harness records ``Service DEPLOYED_TO Environment`` with the
 environment stamped on the edge, the question "what env runs auth-svc?"
 returns a real edge instead of 0% coverage.
+
+When the read carries a free-text ``query``, discovered rows are scored
+against it (``semantic_similarity``) so ranking is query-sensitive; the
+traversal itself stays query-free so the same topology is discovered
+either way.
 """
 
 from __future__ import annotations
 
+import dataclasses
 from dataclasses import dataclass
 from typing import Any, Iterable, Mapping
 
@@ -25,6 +31,7 @@
     claim_corroboration,
     claim_environment,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -92,6 +99,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=row.valid_at,
                     scope_overlap=overlap,
                     corroboration_count=claim_corroboration(row),
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 
@@ -135,6 +143,7 @@ def _traverse(
                     include_invalidated=req.include_invalidated,
                     as_of=req.as_of,
                     source_ref_in=req.source_refs,
+                    fact_query=req.query,
                     limit=max(req.max_items * 4, 16),
                 )
             )
@@ -219,7 +228,55 @@ def _traverse(
                     next_frontier.add(row.subject_key)
             frontier = next_frontier - visited_anchors
 
-        return list(seen_rows.values())
+        return self._stamp_query_similarity(req, list(seen_rows.values()))
+
+    def _stamp_query_similarity(
+        self, req: ReadRequest, rows: list[ClaimRow]
+    ) -> list[ClaimRow]:
+        """Stamp ``semantic_similarity`` onto already-traversed rows.
+
+        The traversal itself stays query-free: on vector backends a
+        ``fact_query`` turns ``find_claims`` into a top-k similarity search,
+        which would silently prune frontier edges and change *which* topology
+        is discovered. Scoring the discovered rows afterwards keeps discovery
+        query-insensitive while making ranking query-sensitive.
+        """
+        if not req.query or not rows:
+            return rows
+        claim_keys = tuple(sorted({row.claim_key for row in rows if row.claim_key}))
+        if not claim_keys:
+            return rows
+        scored = self.claim_query.find_claims(
+            ClaimQueryFilter(
+                pot_id=req.pot_id,
+                predicate_in=_INFRA_PREDICATES,
+                claim_key_in=claim_keys,
+                include_invalidated=req.include_invalidated,
+                as_of=req.as_of,
+                source_ref_in=req.source_refs,
+                fact_query=req.query,
+                limit=len(claim_keys),
+            )
+        )
+        similarity_by_key: dict[str, float] = {}
+        for row in scored:
+            sim = claim_semantic_similarity(row)
+            if row.claim_key and sim is not None:
+                similarity_by_key[row.claim_key] = sim
+
+        out: list[ClaimRow] = []
+        for row in rows:
+            sim = similarity_by_key.get(row.claim_key) if row.claim_key else None
+            if sim is None:
+                # No stamp for this row (e.g. missing claim_key or the vector
+                # top-k skipped it): leave it unset so the ranker uses its
+                # neutral default instead of a fabricated score.
+                out.append(row)
+                continue
+            props = dict(row.properties)
+            props["semantic_similarity"] = sim
+            out.append(dataclasses.replace(row, properties=props))
+        return out
 
 
 # ---------------------------------------------------------------------------

potpie/context-engine/application/readers/owners.py

@@ -15,6 +15,7 @@
     claim_candidate_key,
     claim_corroboration,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -47,7 +48,6 @@ def read(self, req: ReadRequest) -> ReadResponse:
             overlap = _scope_overlap(row, anchor_keys=anchor_keys)
             if anchor_keys and overlap == 0.0:
                 continue
-            sim = row.properties.get("semantic_similarity")
             candidates.append(
                 Candidate(
                     candidate_key=claim_candidate_key(row),
@@ -56,9 +56,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=row.valid_at,
                     corroboration_count=claim_corroboration(row),
                     scope_overlap=overlap if anchor_keys else None,
-                    semantic_similarity=float(sim)
-                    if isinstance(sim, (int, float))
-                    else None,
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 

potpie/context-engine/application/readers/prior_bugs.py

@@ -23,6 +23,7 @@
     ReadResponse,
     claim_candidate_key,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -88,7 +89,6 @@ def read(self, req: ReadRequest) -> ReadResponse:
             )
             if anchor_keys and overlap == 0.0:
                 continue
-            sim = row.properties.get("semantic_similarity")
             fix_key = _fix_key_for_row(row)
             verification_boost = verification_counts.get(fix_key, 0) if fix_key else 0
             candidates.append(
@@ -99,9 +99,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=row.valid_at,
                     corroboration_count=1 + verification_boost,
                     scope_overlap=overlap if anchor_keys else None,
-                    semantic_similarity=float(sim)
-                    if isinstance(sim, (int, float))
-                    else None,
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 

potpie/context-engine/application/readers/timeline_reader.py

@@ -24,6 +24,7 @@
     ReadResponse,
     claim_corroboration,
     claim_payload,
+    claim_semantic_similarity,
     coverage_status_from_count,
     dedupe_claim_rows,
     rank_candidates,
@@ -67,7 +68,6 @@ def read(self, req: ReadRequest) -> ReadResponse:
             overlap = (
                 1.0 if (not anchor_keys) or row_in_anchor_set(row, anchor_keys) else 0.4
             )
-            sim = row.properties.get("semantic_similarity")
             candidates.append(
                 Candidate(
                     candidate_key=_activity_key(row),
@@ -76,9 +76,7 @@ def read(self, req: ReadRequest) -> ReadResponse:
                     valid_at=event_time,
                     scope_overlap=overlap,
                     corroboration_count=claim_corroboration(row),
-                    semantic_similarity=float(sim)
-                    if isinstance(sim, (int, float))
-                    else None,
+                    semantic_similarity=claim_semantic_similarity(row),
                 )
             )
 
@@ -268,8 +266,7 @@ def sort_key(row: ClaimRow) -> tuple[int, int, float, float]:
             "PERFORMED": 1,
             "AUTHORED": 1,
         }.get(predicate, 0)
-        sim = row.properties.get("semantic_similarity")
-        similarity = float(sim) if isinstance(sim, (int, float)) else 0.0
+        similarity = claim_semantic_similarity(row) or 0.0
         event_time = _event_datetime(row)
         event_ts = event_time.timestamp() if event_time else 0.0
         return (direct_anchor, predicate_priority, similarity, event_ts)

potpie/context-engine/domain/graph_workbench_ontology.py

@@ -639,6 +639,7 @@ def to_dict(self, *, include_examples: bool = False) -> dict[str, Any]:
             "direction",
             "environment",
             "include_unqualified_environment",
+            "query",
         ),
         "supported_filters": (
             "service",
@@ -647,6 +648,7 @@ def to_dict(self, *, include_examples: bool = False) -> dict[str, Any]:
             "direction",
             "environment",
             "include_unqualified_environment",
+            "query",
         ),
         "keywords": (
             "service",