feat(edge_rerank): edge-type-keyed rerank consumer (#421)

docs/edge_rerank.md

@@ -0,0 +1,96 @@
+# Edge-type-keyed rerank consumer (#421)
+
+`aelfrice.edge_rerank.apply_edge_type_rerank` is a post-BFS rescore
+pass that demotes surfaced beliefs based on the **edge types of their
+incoming edges**. It is the demotion-half of the retrieval pipeline,
+orthogonal to `BFS_EDGE_WEIGHTS` (which biases reachability during
+expansion).
+
+## Where it lives in the pipeline
+
+```
+seeds (L0 + L2.5 + L1)
+   │
+   ▼
+expand_bfs                ← BFS_EDGE_WEIGHTS bias what is reached
+   │  list[ScoredHop]
+   ▼
+apply_edge_type_rerank    ← per-edge-type penalties demote what was reached  (#421)
+   │  list[ScoredHop] (rescored, re-sorted)
+   ▼
+caller (e.g., retrieve_with_tiers token-budget pack)
+```
+
+The pass is pure: same `(hops, store, penalties)` produces
+byte-identical output. It uses `MemoryStore.edges_to(dst)` to query
+incoming edges per surfaced belief.
+
+## Skip-during-BFS contract
+
+`POTENTIALLY_STALE` is a **marker** edge type: it tags a target belief
+as suspected stale, with no relational/propagation semantics. BFS must
+not walk through it, so it is pinned at weight 0.0:
+
+```python
+BFS_EDGE_WEIGHTS[POTENTIALLY_STALE] = 0.0
+```
+
+The `BFS_EDGE_WEIGHTS.get(t, 0.0)` default would do this implicitly,
+but the contract is pinned explicitly so the intent is reviewable.
+The actual *demotion* of beliefs reached via other edges that happen
+to have stale incoming markers occurs only in the rerank pass.
+
+## Config knobs
+
+```python
+DEFAULT_STALE_PENALTY: float = 0.5
+EDGE_TYPE_PENALTIES_DEFAULT: Mapping[str, float] = {
+    EDGE_POTENTIALLY_STALE: 0.5,
+}
+```
+
+Pass-level override:
+
+```python
+apply_edge_type_rerank(
+    hops, store,
+    penalties={EDGE_POTENTIALLY_STALE: 0.25, ...},
+)
+```
+
+- `penalties=None` → `EDGE_TYPE_PENALTIES_DEFAULT`.
+- `penalties={}` → identity (re-sort only; no per-hop edge query).
+- Penalty values are typically in `[0.0, 1.0]` for demotion semantics;
+  `> 1.0` would amplify and is allowed but contradicts "demotion".
+
+## Multi-edge interaction
+
+When a single surfaced belief has incoming edges of more than one
+penalty-keyed type, the penalties **compose multiplicatively**:
+
+```
+score_after = score_before × ∏ penalties[t]   for t in firing_types
+```
+
+`firing_types` is a *set*: the same edge type appearing on multiple
+incoming edges fires once. The trigger is "at least one matching
+incoming edge of this type," not edge count.
+
+## Determinism
+
+The returned list is sorted by `(-score, belief.id)` — the same
+tie-break used by `expand_bfs`. Two passes compose without
+ordering surprises.
+
+## Bench gate
+
+`tests/bench_gate/test_edge_rerank_potentially_stale.py` enforces
+#421 acceptance #3 / #387 acceptance #3: **≥1pp@k drop** in
+stale-tagged retrieval after the rerank pass vs. before, on the
+`bfs_potentially_stale` corpus module. Skips cleanly on public CI
+when the corpus is unmounted.
+
+## Producer side
+
+`POTENTIALLY_STALE` edges are produced by `aelf doctor` (#387) — the
+edge-writer is out of scope for #421. This module is the consumer.

src/aelfrice/bfs_multihop.py

@@ -34,6 +34,7 @@
     EDGE_CONTRADICTS,
     EDGE_DERIVED_FROM,
     EDGE_IMPLEMENTS,
+    EDGE_POTENTIALLY_STALE,
     EDGE_RELATES_TO,
     EDGE_SUPERSEDES,
     EDGE_SUPPORTS,
@@ -67,6 +68,11 @@
     EDGE_CITES: 0.40,
     EDGE_RELATES_TO: 0.30,
     EDGE_TEMPORAL_NEXT: 0.25,
+    # Marker edge — skipped during BFS expansion. Demotion happens in
+    # the rerank pass (`aelfrice.edge_rerank`), not here. Pinned at 0.0
+    # explicitly so the contract is reviewable rather than implicit
+    # via the `BFS_EDGE_WEIGHTS.get(..., 0.0)` default. See #421.
+    EDGE_POTENTIALLY_STALE: 0.0,
 }
 
 

src/aelfrice/edge_rerank.py

@@ -0,0 +1,111 @@
+"""Edge-type-keyed rerank consumer for BFS expansion results (#421).
+
+Problem: `BFS_EDGE_WEIGHTS` in `bfs_multihop` are non-negative path
+multipliers — they bias what is *reached*, not how reached results
+are *ranked*. Marker edges like ``POTENTIALLY_STALE`` need a
+separate demotion pass that downgrades reachable beliefs after BFS,
+not by path-multiplication during BFS expansion.
+
+This module is that pass. It runs downstream of BFS / lane fusion,
+takes the `ScoredHop` list, examines each hop's belief's incoming
+edges, and applies a configurable multiplicative penalty per
+matching edge type. The result is a new `list[ScoredHop]` with
+rescored scores, re-sorted by ``(-score, belief.id)`` — the same
+tie-breaking rule used by `expand_bfs` so two passes compose without
+order surprises.
+
+The producer for ``POTENTIALLY_STALE`` edges is `aelf doctor` (#387);
+this module is its consumer-side substrate.
+
+Multi-edge composition: when more than one penalty-keyed incoming
+edge type fires on the same belief, penalties compose
+**multiplicatively** (a belief reached via two penalty-keyed edge
+types has its score multiplied by *both* factors). Same edge type
+firing multiple times collapses to one factor — "at least one
+matching incoming edge" is the trigger, not edge count.
+
+Stdlib only.
+"""
+from __future__ import annotations
+
+from typing import Final, Mapping
+
+from aelfrice.bfs_multihop import ScoredHop
+from aelfrice.models import EDGE_POTENTIALLY_STALE
+from aelfrice.store import MemoryStore
+
+# Default penalty factor applied when at least one ``POTENTIALLY_STALE``
+# incoming edge targets a surfaced belief. Conservative starting point:
+# 0.5 halves the score, which preserves the belief's relative position
+# below non-stale equivalents but does not drop it below the BFS floor
+# (the rerank pass is a re-rank, not a hard filter). Operator-tunable
+# per call via the `penalties` kwarg on `apply_edge_type_rerank`.
+DEFAULT_STALE_PENALTY: Final[float] = 0.5
+
+# Default penalty config — only ``POTENTIALLY_STALE`` is keyed today.
+# Future marker edges that need rerank-time demotion add their entries
+# here; non-marker (positive-weight) edges should never be in this
+# table — those bias retrieval through `BFS_EDGE_WEIGHTS` at expansion
+# time, not at rerank time.
+EDGE_TYPE_PENALTIES_DEFAULT: Final[Mapping[str, float]] = {
+    EDGE_POTENTIALLY_STALE: DEFAULT_STALE_PENALTY,
+}
+
+
+def apply_edge_type_rerank(
+    hops: list[ScoredHop],
+    store: MemoryStore,
+    *,
+    penalties: Mapping[str, float] | None = None,
+) -> list[ScoredHop]:
+    """Rerank `hops` by applying per-edge-type penalties.
+
+    For each hop's belief, query incoming edges via `store.edges_to`.
+    If any incoming edge type appears in `penalties`, multiply the
+    hop's score by the corresponding penalty factor. Multiple distinct
+    matching edge types compose multiplicatively. The same matching
+    edge type appearing on multiple incoming edges fires once — the
+    presence test is "at least one matching edge of this type."
+
+    Determinism contract: the returned list is sorted by
+    ``(-score, belief.id)``, matching `expand_bfs`'s tie-breaking
+    rule. The same `(hops, store, penalties)` input produces
+    byte-identical output.
+
+    Args:
+      hops: BFS expansion results from `expand_bfs`. Empty list is
+        a no-op (returns empty list).
+      store: MemoryStore providing `edges_to(dst)`.
+      penalties: per-edge-type penalty factors, typically in
+        ``[0.0, 1.0]`` for demotion semantics. ``None`` selects
+        `EDGE_TYPE_PENALTIES_DEFAULT` (``POTENTIALLY_STALE`` @ 0.5).
+        An explicit ``{}`` is identity (re-sort only).
+
+    Returns:
+      A new `list[ScoredHop]` with rescored `score` fields, sorted
+      by ``(-score, belief.id)``.
+    """
+    if not hops:
+        return []
+    cfg: Mapping[str, float] = (
+        EDGE_TYPE_PENALTIES_DEFAULT if penalties is None else penalties
+    )
+    if not cfg:
+        return sorted(hops, key=lambda h: (-h.score, h.belief.id))
+    rescored: list[ScoredHop] = []
+    for hop in hops:
+        incoming = store.edges_to(hop.belief.id)
+        firing = {e.type for e in incoming if e.type in cfg}
+        new_score = hop.score
+        for edge_type in firing:
+            new_score *= cfg[edge_type]
+        rescored.append(
+            ScoredHop(
+                belief=hop.belief,
+                score=new_score,
+                depth=hop.depth,
+                path=hop.path,
+            )
+        )
+    rescored.sort(key=lambda h: (-h.score, h.belief.id))
+    return rescored

src/aelfrice/models.py

@@ -33,6 +33,16 @@
 EDGE_TEMPORAL_NEXT: Final[str] = "TEMPORAL_NEXT"
 EDGE_TESTS: Final[str] = "TESTS"
 
+# Marker edge — semantically distinct from the relational edge types
+# above. POTENTIALLY_STALE tags a target belief as suspected stale; it
+# carries no propagation valence and is skipped during BFS expansion
+# (`BFS_EDGE_WEIGHTS[POTENTIALLY_STALE] = 0.0`). The consumer is the
+# edge-type-keyed rerank pass in `aelfrice.edge_rerank` (#421); the
+# producer is `aelf doctor` (#387). Deliberately NOT in `EDGE_TYPES`
+# or `EDGE_VALENCE` — those enumerate structural relational edges,
+# this is a tag.
+EDGE_POTENTIALLY_STALE: Final[str] = "POTENTIALLY_STALE"
+
 # Edge-type valence multipliers for propagation.
 # Positive = propagate same sign; negative = invert; 0.0 = no propagation.
 # DERIVED_FROM mirrors CITES (0.5): both indicate B's content depends on A,

src/aelfrice/store.py

@@ -2318,6 +2318,17 @@ def edges_from(self, src: str) -> list[Edge]:
         )
         return [_row_to_edge(r) for r in cur.fetchall()]
 
+    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.
+        """
+        cur = self._conn.execute(
+            "SELECT * FROM edges WHERE dst = ?", (dst,)
+        )
+        return [_row_to_edge(r) for r in cur.fetchall()]
+
     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

tests/bench_gate/test_edge_rerank_potentially_stale.py

@@ -0,0 +1,133 @@
+"""Bench gate for #421 / #387 — POTENTIALLY_STALE rerank demotion.
+
+Per #421 acceptance #3 and #387 acceptance #3: the edge-type-keyed
+rerank consumer must demonstrate **≥1pp@k drop on stale-tagged
+retrieval** when the rerank pass runs over a labeled corpus where
+some retrievable beliefs have at least one ``POTENTIALLY_STALE``
+incoming edge.
+
+Metric: stale rate at k = ``count(stale_ids ∩ top_k) / total_stale``,
+summed across rows. Compute pre-rerank vs post-rerank; assert the
+drop is at least `STALE_DROP_FLOOR` (0.01 = 1pp).
+
+Skips cleanly when ``AELFRICE_CORPUS_ROOT`` is unset (public CI),
+when the ``bfs_potentially_stale/`` module dir is missing, or when
+the corpus has fewer than ``MIN_ROWS`` non-seed rows (the gate
+requires a row floor before rate-difference measurement is
+statistically meaningful).
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.conftest import load_corpus_module
+
+STALE_DROP_FLOOR = 0.01  # +1pp per #421 / #387 acceptance #3
+MIN_ROWS = 30  # public-tree floor; lab corpus is expected to exceed this
+
+
+def _build_store(tmp_path: Path, row: dict, arm: str):
+    from aelfrice.models import BELIEF_FACTUAL, Belief, Edge
+    from aelfrice.store import MemoryStore
+
+    db_path = tmp_path / f"{row['id']}-{arm}.db"
+    store = MemoryStore(str(db_path))
+    for b in row["beliefs"]:
+        belief = Belief(
+            id=b["id"],
+            content=b["text"],
+            content_hash=f"h_{b['id']}",
+            alpha=1.0,
+            beta=1.0,
+            type=BELIEF_FACTUAL,
+            lock_level="none",
+            locked_at=None,
+            demotion_pressure=0,
+            created_at="2026-05-04T00:00:00Z",
+            last_retrieved_at=None,
+        )
+        store.insert_belief(belief)
+    for e in row["edges"]:
+        store.insert_edge(
+            Edge(
+                src=e["src"],
+                dst=e["dst"],
+                type=e["type"],
+                weight=float(e["weight"]),
+            )
+        )
+    return store
+
+
+def _row_top_k_ids(row: dict, store, *, rerank: bool) -> list[str]:
+    """Run BFS expansion (POTENTIALLY_STALE is skip-during-BFS per
+    `BFS_EDGE_WEIGHTS[POTENTIALLY_STALE] = 0.0`) and either return
+    BFS top-k directly (`rerank=False`) or apply the rerank pass
+    first and then take top-k (`rerank=True`)."""
+    from aelfrice.bfs_multihop import expand_bfs
+    from aelfrice.edge_rerank import apply_edge_type_rerank
+
+    seeds = []
+    for sid in row["seed_ids"]:
+        b = store.get_belief(sid)
+        assert b is not None, f"row {row['id']}: seed {sid} not in row beliefs"
+        seeds.append(b)
+    expansions = expand_bfs(seeds, store)
+    if rerank:
+        expansions = apply_edge_type_rerank(expansions, store)
+    k = int(row["k"])
+    return [hop.belief.id for hop in expansions[:k]]
+
+
+@pytest.mark.bench_gated
+def test_potentially_stale_rerank_drops_stale_in_top_k(
+    aelfrice_corpus_root: Path,
+    tmp_path: Path,
+) -> None:
+    rows = [
+        r
+        for r in load_corpus_module(
+            aelfrice_corpus_root, "bfs_potentially_stale"
+        )
+        if not r.get("seed", False)
+    ]
+
+    if len(rows) < MIN_ROWS:
+        pytest.skip(
+            f"bfs_potentially_stale corpus has {len(rows)} non-seed rows; gate "
+            f"requires ≥{MIN_ROWS} for stable rate-difference measurement"
+        )
+
+    total_stale = sum(len(set(r["stale_ids"])) for r in rows)
+    assert total_stale > 0, (
+        "corpus has zero stale_ids across all rows; cannot grade "
+        "rerank-demotion impact"
+    )
+
+    pre_stale_in_top_k = 0
+    post_stale_in_top_k = 0
+    for row in rows:
+        stale = set(row["stale_ids"])
+        store = _build_store(tmp_path, row, arm="run")
+        try:
+            pre_top = set(_row_top_k_ids(row, store, rerank=False))
+            post_top = set(_row_top_k_ids(row, store, rerank=True))
+            pre_stale_in_top_k += len(pre_top & stale)
+            post_stale_in_top_k += len(post_top & stale)
+        finally:
+            store.close()
+
+    pre_rate = pre_stale_in_top_k / total_stale
+    post_rate = post_stale_in_top_k / total_stale
+    drop = pre_rate - post_rate
+
+    assert drop >= STALE_DROP_FLOOR, (
+        f"POTENTIALLY_STALE rerank drop {drop:+.3f} below "
+        f"+{STALE_DROP_FLOOR:.2f} floor (pre={pre_rate:.3f}, "
+        f"post={post_rate:.3f}, n_rows={len(rows)}, "
+        f"n_stale={total_stale}). Per #421 / #387 acceptance #3, the "
+        f"rerank consumer ships only on ≥+1pp drop in stale-tagged "
+        f"retrieval; below-floor blocks #387 closure."
+    )