fix(graph): traverse SUPERSEDES in reverse so BFS surfaces the replacement (#1170)

docs/design/bfs_multihop.md

@@ -183,7 +183,7 @@ is the noisiest signal in the v1.2 ingest output).
 
 | Edge type      | Weight | Class          | Rationale |
 |----------------|--------|----------------|-----------|
-| `SUPERSEDES`   | 0.90   | decisional     | "B replaces A" — the most actionable adjacency. If the query hit A, the user almost certainly wants B. Highest weight. |
+| `SUPERSEDES`   | 0.90   | decisional     | "B replaces A" — the most actionable adjacency. If the query hit A, the user almost certainly wants B. Highest weight. **Traversed in reverse (#1170):** producers store this edge as `src=B` (the replacement) → `dst=A` (the replaced), which is what the type name means and what `contradiction.resolve_contradiction` and the triple extractor both write. Walking it outbound therefore did the opposite of the rationale in this row — a hit on the current belief surfaced its *stale* predecessor at 0.90, and a hit on the stale one surfaced nothing, so the case this weight was chosen for never fired. `expand_bfs` now reads this type from the inbound side (`REVERSE_TRAVERSED_EDGE_TYPES`) and does not follow it outbound. |
 | `CONTRADICTS`  | 0.85   | decisional     | "B disagrees with A" — surfacing it lets the agent flag the conflict instead of acting on a contradicted belief. Slightly below SUPERSEDES because contradictions are not always resolved (the v1.0.1 contradiction tie-breaker may not have fired yet). |
 | `DERIVED_FROM` | 0.70   | provenance     | "B's content depends on A" — strong contextual coupling, per the v1.2 ingest enrichment spec ("sibling becomes stale if A is superseded"). Following it surfaces parent decisions. Triple extractor produces `DERIVED_FROM` from "X is derived from Y" / "X is based on Y" / "X extends Y". **Retroactive ship-gate (#388):** shipped pre-bench-gate at v1.2; now must clear the same ≥+5pp BFS multi-hop hit@k uplift bar as the other Track A edges per #382 ratification; gate harness at `tests/bench_gate/test_bfs_multihop_derived_from.py`. Below-floor closes #388 as `wontfix`. |
 | `IMPLEMENTS`   | 0.65   | provenance     | "B implements A" — source is an implementation, target is the spec/claim being implemented. Slightly below DERIVED_FROM (0.70) because IMPLEMENTS is a more specific kind of derivation, but the dependency is almost as strong: an implementation becomes stale when its spec is superseded. Triple extractor produces `IMPLEMENTS` from "X implements Y" / "X is an implementation of Y" / "X realizes Y" / "X fulfills Y". **v2.0 ship-gate (#385):** the edge stays at weight 0.65 only while it clears a ≥+5pp BFS multi-hop hit@k uplift on the labeled `implements_edge/` corpus vs. the same fixture run with this entry zeroed; gate harness lives at `tests/bench_gate/test_bfs_multihop_implements.py`. Below-floor closes #385 as `wontfix`. |
@@ -240,6 +240,19 @@ valence carrier. `SUPERSEDES` is the **highest-relevance** adjacency,
 not a structural-zero. Reusing `EDGE_VALENCE` would therefore actively
 mis-score retrieval. Two tables, two purposes.
 
+The `SUPERSEDES` gap between the tables (0.0 vs 0.90) is the starkest
+and is **intentional, not drift** — re-confirmed under #1170. The two
+values answer different questions about the same edge, and neither
+should move:
+
+- `EDGE_VALENCE = 0.0` — "does a feedback signal cross this edge?" No.
+  Reinforcing a replacement says nothing about the confidence of what it
+  replaced. Because valence never traverses `SUPERSEDES` at all, the
+  edge-direction question that #1170 fixed does not arise there.
+- `BFS_EDGE_WEIGHTS = 0.90` — "how relevant is the belief on the other
+  end?" Maximally. This is the one that had to be traversed in reverse
+  to match its own rationale.
+
 ## Depth cap and budget
 
 Defaults, with the rationale that gives them concrete numbers
@@ -433,8 +446,8 @@ fix targeted at v2.0.**
 ### What "temporal coherence" means here
 
 When a belief A is superseded by A', and A' is superseded by A'',
-the BFS frontier walking outbound from a tier-0 hit on A naively
-surfaces A'' as the latest. That is correct *for that one hop*. But
+the BFS frontier walking *inbound along SUPERSEDES* (#1170) from a
+tier-0 hit on A naively surfaces A'' as the latest. That is correct *for that one hop*. But
 when the seed is itself a session-scoped belief from session S₁ and
 the chain has crossed `SUPERSEDES` boundaries that postdate S₁, the
 "latest serial" the agent receives may not be the one that was

src/aelfrice/bfs_multihop.py

@@ -75,6 +75,25 @@
     EDGE_POTENTIALLY_STALE: 0.0,
 }
 
+# Edge types the walk follows AGAINST their stored direction (#1170).
+#
+# Producers write SUPERSEDES as src=winner(new) -> dst=loser(old):
+# `contradiction.resolve_contradiction` (src=winner.id, dst=loser.id) and
+# `triple_extractor` parsing "X supersedes Y" as src=X. That direction is
+# the one the edge type's name means, and it is kept.
+#
+# But the spec's justification for the 0.90 weight is "'B replaces A' — the
+# most actionable adjacency. If the query hit A, the user almost certainly
+# wants B", which needs an old -> new hop. Walking outbound delivered the
+# exact opposite: a hit on the *current* belief surfaced its stale
+# predecessor at the highest available path score, and a hit on the stale
+# one surfaced nothing — so the case the weight was chosen for never fired.
+#
+# Following SUPERSEDES in reverse fixes the direction without a migration
+# that would leave the edge type reading backwards. Types listed here are
+# NOT also followed outbound; that is what produced the inversion.
+REVERSE_TRAVERSED_EDGE_TYPES: frozenset[str] = frozenset({EDGE_SUPERSEDES})
+
 
 @dataclass(frozen=True)
 class ScoredHop:
@@ -125,14 +144,31 @@ def expand_bfs(
     min_path_score: float = DEFAULT_MIN_PATH_SCORE,
     seed_scopes: dict[str, str | None] | None = None,
 ) -> list[ScoredHop]:
-    """Walk outbound edges from `seeds`, returning ranked expansions.
+    """Walk the edge graph from `seeds`, returning ranked expansions.
 
     Pseudocode + properties: see `docs/design/bfs_multihop.md § Algorithm`.
 
+    Traversal direction (#1170). Most edge types are followed
+    **outbound** — a hop from `current_id` reaches each edge's `dst`,
+    read via ``edges_from_in_scope``. The types in
+    ``REVERSE_TRAVERSED_EDGE_TYPES`` are followed **inbound** instead:
+    they are read via ``edges_to_in_scope`` and the hop reaches each
+    edge's `src`. Those types are never also followed outbound. Today
+    that set is ``{SUPERSEDES}``, whose producers write `src` = the new
+    belief and `dst` = the one it retires, so an inbound read is what
+    steps a hit on a retired belief forward to its replacement. The two
+    reads are merged into one candidate list before ranking, so
+    "neighbour" below means either kind.
+
     Determinism contract:
-      - Edges at each frontier expansion are ranked by
-        (-edge_type_weight, -edge.weight, dst_id_ascending). Any
+      - Candidates at each frontier expansion are ranked by
+        (-edge_type_weight, -edge.weight, neighbour_id_ascending). Any
         ranking tie thus breaks on belief id ascending.
+      - Candidates are then deduplicated by neighbour id, keeping the
+        strongest edge to each, BEFORE the `nodes_per_hop` slice — two
+        edges in one hop can name the same neighbour, and letting a
+        duplicate consume a slot would underfill the hop and drop an
+        otherwise-eligible belief.
       - Final results are sorted by (-score, belief.id) so two
         identical inputs always produce byte-identical output.
 
@@ -142,7 +178,8 @@ def expand_bfs(
 
     Budget bookkeeping:
       - `nodes_per_hop` caps fanout per frontier entry (top-k after
-        edge-type ranking).
+        edge-type ranking and neighbour dedup, so the cap counts
+        distinct beliefs rather than distinct edges).
       - `total_budget` caps the cumulative number of expanded
         beliefs across all hops.
       - `min_path_score` prunes paths whose multiplicative score has
@@ -156,8 +193,8 @@ def expand_bfs(
     Federation (#690): ``seed_scopes`` is an optional ``{belief_id:
     owning_scope}`` mapping. When a seed id appears in the dict with
     a non-None scope, the walk follows that peer's edges (via
-    ``store.edges_from_in_scope`` / ``get_belief_in_scope``) instead
-    of local. The scope propagates from each frontier entry to its
+    ``store.edges_from_in_scope`` / ``edges_to_in_scope`` /
+    ``get_belief_in_scope``) instead of local. The scope propagates from each frontier entry to its
     children — once the walk enters a peer, subsequent hops stay
     inside that peer's edge graph. Seeds not in the dict (and the
     default ``seed_scopes=None`` case) walk local edges only, so
@@ -193,23 +230,65 @@ def expand_bfs(
                 continue
             if nodes_used >= total_budget:
                 break
-            edges: list[Edge] = store.edges_from_in_scope(current_id, scope)
+            # Neighbours reachable from `current_id`, normalised to
+            # (neighbour_id, edge_type, edge.weight). Outbound edges give
+            # their `dst`; the reverse-traversed types (#1170) are read
+            # from the inbound side and give their `src`, so a hit on a
+            # superseded belief steps forward to its replacement rather
+            # than the other way round.
+            neighbours: list[tuple[str, str, float]] = [
+                (e.dst, e.type, e.weight)
+                for e in store.edges_from_in_scope(current_id, scope)
+                if e.type not in REVERSE_TRAVERSED_EDGE_TYPES
+            ]
+            neighbours += [
+                (e.src, e.type, e.weight)
+                for e in store.edges_to_in_scope(current_id, scope)
+                if e.type in REVERSE_TRAVERSED_EDGE_TYPES
+            ]
             # Determinism: rank by (-edge-type-weight, -edge.weight,
-            # dst id). Filter already-visited dsts BEFORE ranking so
+            # neighbour id). Filter already-visited ids BEFORE ranking so
             # the top-k slice is over genuinely-novel candidates.
-            candidates = [e for e in edges if e.dst not in visited]
-            ranked = sorted(
+            candidates = [n for n in neighbours if n[0] not in visited]
+            ordered = sorted(
                 candidates,
-                key=lambda e: (
-                    -BFS_EDGE_WEIGHTS.get(e.type, 0.0),
-                    -e.weight,
-                    e.dst,
+                key=lambda n: (
+                    -BFS_EDGE_WEIGHTS.get(n[1], 0.0),
+                    -n[2],
+                    n[0],
                 ),
-            )[:nodes_per_hop]
-            for edge in ranked:
+            )
+            # Deduplicate by neighbour id BEFORE the top-k slice, keeping
+            # the strongest edge to each. Two edges in one hop can name
+            # the same neighbour — different types between one pair are
+            # permitted by the `(src, dst, type)` PK, and since #1170 an
+            # outbound edge and a reverse-traversed inbound one can also
+            # collide. Slicing first would let the duplicate consume a
+            # slot and drop an otherwise-eligible neighbour, underfilling
+            # the hop. That is not exotic: `resolve_contradiction` writes
+            # SUPERSEDES between a pair that already carries CONTRADICTS,
+            # and those are the two highest weights in the table, so the
+            # duplicate reliably lands at the top of the ranking.
+            ranked: list[tuple[str, str, float]] = []
+            seen_this_hop: set[str] = set()
+            for cand in ordered:
+                if cand[0] in seen_this_hop:
+                    continue
+                seen_this_hop.add(cand[0])
+                ranked.append(cand)
+                if len(ranked) >= nodes_per_hop:
+                    break
+            for neighbour_id, edge_type, _edge_weight in ranked:
                 if nodes_used >= total_budget:
                     break
-                edge_w = BFS_EDGE_WEIGHTS.get(edge.type, 0.0)
+                if neighbour_id in visited:
+                    # Defence in depth. `candidates` is filtered against
+                    # `visited` before ranking and `ranked` is deduped
+                    # within the hop, so nothing should reach here —
+                    # emitting a duplicate would return the same belief
+                    # twice and charge the node budget twice.
+                    continue
+                edge_w = BFS_EDGE_WEIGHTS.get(edge_type, 0.0)
                 if edge_w == 0.0:
                     # Unknown / zero-weighted edge type — skip,
                     # don't mark visited (a future hop might still
@@ -221,16 +300,16 @@ def expand_bfs(
                 # Mark visited BEFORE the materialisation guard so a
                 # missing-belief race doesn't re-queue the same id
                 # later in this same call.
-                visited.add(edge.dst)
-                belief = store.get_belief_in_scope(edge.dst, scope)
+                visited.add(neighbour_id)
+                belief = store.get_belief_in_scope(neighbour_id, scope)
                 if belief is None:
-                    # Race: belief was deleted between edges_from
+                    # Race: belief was deleted between the edge read
                     # and get_belief. Skip; the next mutation cycle
                     # will fire the cache invalidation that re-runs
                     # this query.
                     continue
-                new_path = path + [edge.type]
-                new_trail = trail + (edge.dst,)
+                new_path = path + [edge_type]
+                new_trail = trail + (neighbour_id,)
                 expanded.append(
                     ScoredHop(
                         belief=belief,
@@ -242,7 +321,10 @@ def expand_bfs(
                     )
                 )
                 next_frontier.append(
-                    (edge.dst, new_score, depth + 1, new_path, new_trail, scope)
+                    (
+                        neighbour_id, new_score, depth + 1,
+                        new_path, new_trail, scope,
+                    )
                 )
                 nodes_used += 1
         frontier = next_frontier

src/aelfrice/store.py

@@ -5440,16 +5440,47 @@ def entity_persistence_scores(
         return out
 
     def edges_to(self, dst: str) -> list[Edge]:
-        """Return every edge whose `dst` is `dst`. Symmetric companion
-        to `edges_from`. Used by the edge-type-keyed rerank pass
-        (#421) to detect marker edges (e.g., POTENTIALLY_STALE)
-        targeting a surfaced belief.
+        """Return every edge whose `dst` is `dst`, in a total order.
+
+        Symmetric companion to `edges_from`. Used by the
+        edge-type-keyed rerank pass (#421) to detect marker edges
+        (e.g., POTENTIALLY_STALE) targeting a surfaced belief, and by
+        the BFS walk to traverse SUPERSEDES in reverse (#1170).
+
+        `ORDER BY src, type` for the same reason `edges_from` orders
+        (#1169): the BFS frontier is determinism-load-bearing, and the
+        raw row order here is physical layout. Unlike `edges_from` this
+        one does cost a sort — the PK is `(src, dst, type)`, so a
+        `dst`-keyed lookup cannot be index-ordered — but the row count
+        per `dst` is the belief's in-degree, which is small.
         """
         cur = self._conn.execute(
-            "SELECT * FROM edges WHERE dst = ?", (dst,)
+            "SELECT * FROM edges WHERE dst = ? ORDER BY src, type", (dst,)
         )
         return [_row_to_edge(r) for r in cur.fetchall()]
 
+    def edges_to_in_scope(
+        self, dst: str, owning_scope: str | None
+    ) -> list[Edge]:
+        """Read-only ``edges_to`` against a named peer (or local).
+
+        Mirrors :meth:`edges_from_in_scope`. ``owning_scope=None`` is the
+        local DB. Unreachable peers and schema-drift peer DBs return
+        ``[]`` rather than raising — federation is opportunistic per #661.
+        """
+        if owning_scope is None:
+            return self.edges_to(dst)
+        conn = self._peer_conn(owning_scope)
+        if conn is None:
+            return []
+        try:
+            cur = conn.execute(
+                "SELECT * FROM edges WHERE dst = ? ORDER BY src, type", (dst,)
+            )
+            return [_row_to_edge(r) for r in cur.fetchall()]
+        except sqlite3.OperationalError:
+            return []
+
     def iter_all_edges(self) -> Iterator[Edge]:
         """Stream every edge in the store. Ordering is insertion order
         (sqlite ROWID). Used by graph-wide builders such as the signed