From 271d630531dde76a31f2aabe1e1cd81798838344 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Fri, 8 May 2026 12:30:34 -0700 Subject: [PATCH 1/2] feat(clustering): RetrievalCluster module + multi-fact corpus mount (436) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the pure-library half of intentional clustering per docs/feature-intentional-clustering.md. New `src/aelfrice/clustering.py` exposes: - `RetrievalCluster` dataclass — dense cluster_id, score-ranked member_ids, representative_id, seed_score. - `cluster_candidates()` — path-compressed union-find pass over the candidate-induced edge subgraph. Edges below `edge_weight_floor` (default 0.4: includes CITES, excludes RELATES_TO) and edges with one endpoint outside the candidate pool are filtered out. - `pack_with_clusters()` — diversity-aware greedy fill. Stage 1 picks one representative per cluster (descending seed_score) until `cluster_diversity_target=3` distinct clusters are covered or budget is exhausted. Stage 2 fills remaining budget from the score-ranked tail. `fallback_to_score=True` (default) bails Stage 1 on the first budget miss. `MemoryStore.edges_for_beliefs(belief_ids)` — batched fetch of every edge whose src OR dst is in `belief_ids`. Single SQL with `IN (...)`, empty input → empty list (no SQL). Bench-gate harness scaffold at `tests/bench_gate/test_intentional_clustering.py` + corpus mount point at `tests/corpus/v2_0/multi_fact/`. Public CI skips via the autouse `bench_gated` marker; corpus content lives lab-side per the directory-of-origin rule. The retrieval-side wiring (use_intentional_clustering flag in retrieve_v2 + flag-resolution helpers) is the next gate; the module ships independently so the substrate lands without a hot-path edit. Spec status updated to "module shipped; retrieval wiring + bench-gate evidence are the next gates". Reuses the path-compressed union-find pattern from src/aelfrice/dedup.py rather than importing it — neither owns the primitive yet, and a future refactor can promote one of the two. --- docs/feature-intentional-clustering.md | 2 +- src/aelfrice/clustering.py | 247 +++++++++++++++ src/aelfrice/store.py | 21 ++ .../bench_gate/test_intentional_clustering.py | 65 ++++ tests/corpus/v2_0/README.md | 31 +- tests/corpus/v2_0/multi_fact/.gitkeep | 0 tests/test_clustering.py | 299 ++++++++++++++++++ tests/test_corpus_schema.py | 25 ++ 8 files changed, 688 insertions(+), 2 deletions(-) create mode 100644 src/aelfrice/clustering.py create mode 100644 tests/bench_gate/test_intentional_clustering.py create mode 100644 tests/corpus/v2_0/multi_fact/.gitkeep create mode 100644 tests/test_clustering.py diff --git a/docs/feature-intentional-clustering.md b/docs/feature-intentional-clustering.md index 9724ac781..1fbab8fa4 100644 --- a/docs/feature-intentional-clustering.md +++ b/docs/feature-intentional-clustering.md @@ -1,6 +1,6 @@ # Feature spec: Intentional clustering (#436) -**Status:** spec, no implementation +**Status:** module shipped (`src/aelfrice/clustering.py`); retrieval-side wiring + bench-gate evidence are the next gates **Issue:** #436 **Recovery-inventory line:** [`docs/ROADMAP.md`](ROADMAP.md) — *"Intentional clustering | v2.0.0"* **Substrate prereqs:** edge graph (foundation), `dedup.DuplicateCluster` union-find pattern (`src/aelfrice/dedup.py:155-185`, shipped #197), heat kernel authority (#150, shipped v1.7.0), BFS multi-hop (#143, shipped v1.3.0) diff --git a/src/aelfrice/clustering.py b/src/aelfrice/clustering.py new file mode 100644 index 000000000..adae5ad81 --- /dev/null +++ b/src/aelfrice/clustering.py @@ -0,0 +1,247 @@ +"""Intentional clustering (#436). + +Retrieval-time pass that biases the top-K output toward cluster-diverse +beliefs — when a multi-fact query needs more than one belief to answer, +the existing rank+pack returns K beliefs from the highest-scoring graph +neighbourhood and a complementary cluster never makes the cut. +Clustering replaces the pack loop with a diversity-aware greedy fill. + +Spec: ``docs/feature-intentional-clustering.md``. + +This module owns the pure-library half of the contract: + +- ``cluster_candidates`` — union-find pass over the candidate-induced + edge subgraph. Returns one ``RetrievalCluster`` per connected + component. +- ``pack_with_clusters`` — diversity-aware greedy fill. Stage 1 picks + one representative per cluster up to ``cluster_diversity_target`` + distinct clusters; Stage 2 fills the remaining budget by score. + +The retrieval-side wiring (flag resolution, integration with +``retrieve_v2``) lands separately so this module can ship + bench +without a hot-path edit. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Final, Iterable + +from aelfrice.models import Belief, Edge + +# Default edge-weight floor: 0.4. Picked to include `EDGE_CITES` (0.5 +# in `EDGE_VALENCE`) but exclude `EDGE_RELATES_TO` (0.3) — beliefs that +# only relate are too weak a signal to be considered the same cluster. +# Tunable via `[retrieval] cluster_edge_weight_floor`. +DEFAULT_CLUSTER_EDGE_FLOOR: Final[float] = 0.4 + +# Default diversity target: 3 distinct clusters in the top-K. Three +# covers most multi-fact queries without crowding out the score-ranked +# tail. Tunable via `[retrieval] cluster_diversity_target`. +DEFAULT_CLUSTER_DIVERSITY_TARGET: Final[int] = 3 + +_CHARS_PER_TOKEN: Final[float] = 4.0 + + +def _belief_tokens(b: Belief) -> int: + """Char-based token estimate, conservative (rounds up). + + Mirrors `retrieval._belief_tokens`. Duplicated here rather than + imported to keep this module free of a `retrieval`-side dependency + (the wiring direction is retrieval → clustering, not vice versa). + """ + if not b.content: + return 0 + n = len(b.content) + return int((n + _CHARS_PER_TOKEN - 1) // _CHARS_PER_TOKEN) + + +@dataclass(frozen=True) +class RetrievalCluster: + """One connected-component cluster within the post-rank candidate pool. + + ``cluster_id`` is dense (zero-indexed in deterministic insertion + order). ``member_ids`` is sorted by descending rank score so + ``member_ids[0]`` is the representative — the highest-scoring member + that Stage 1 of the pack picks first. + """ + + cluster_id: int + member_ids: tuple[str, ...] + representative_id: str + seed_score: float + + +class _UnionFind: + """Path-compressed, union-by-size DSU. Mirrors `dedup._UnionFind`. + + Duplicated rather than imported so a future refactor can promote + one of the two to a shared primitive; today neither owns it. + """ + + __slots__ = ("_parent", "_size") + + def __init__(self) -> None: + self._parent: dict[str, str] = {} + self._size: dict[str, int] = {} + + def make(self, x: str) -> None: + if x not in self._parent: + self._parent[x] = x + self._size[x] = 1 + + def find(self, x: str) -> str: + path: list[str] = [] + while self._parent[x] != x: + path.append(x) + x = self._parent[x] + for p in path: + self._parent[p] = x + return x + + def union(self, a: str, b: str) -> None: + ra, rb = self.find(a), self.find(b) + if ra == rb: + return + if self._size[ra] < self._size[rb]: + ra, rb = rb, ra + self._parent[rb] = ra + self._size[ra] += self._size[rb] + + +def cluster_candidates( + candidates: list[Belief], + candidate_scores: dict[str, float], + *, + edges: Iterable[Edge], + edge_weight_floor: float = DEFAULT_CLUSTER_EDGE_FLOOR, +) -> list[RetrievalCluster]: + """Group ``candidates`` into connected components on the + candidate-induced edge subgraph. + + The subgraph's vertex set is ``{c.id for c in candidates}``; edges + are the ones in ``edges`` whose ``weight >= edge_weight_floor`` AND + both endpoints are in the vertex set (candidate-induced — non- + candidate beliefs are out of consideration per spec § Open question 1). + + ``candidate_scores`` is the per-belief rank score; clusters' + ``seed_score`` is the max over the component, ``member_ids`` is + sorted by descending score with ties broken by id ASC for determinism. + + Singletons (candidates with no in-pool neighbours above the floor) + are returned as size-1 clusters. + + Cluster ordering in the returned list is by descending ``seed_score``; + ties broken by ``representative_id`` ASC. ``cluster_id`` reflects + that order. + """ + if not candidates: + return [] + + candidate_ids = {c.id for c in candidates} + uf = _UnionFind() + for cid in candidate_ids: + uf.make(cid) + for e in edges: + if e.weight < edge_weight_floor: + continue + if e.src not in candidate_ids or e.dst not in candidate_ids: + continue + uf.union(e.src, e.dst) + + groups: dict[str, list[str]] = {} + for cid in candidate_ids: + groups.setdefault(uf.find(cid), []).append(cid) + + raw_clusters: list[tuple[float, str, tuple[str, ...]]] = [] + for members in groups.values(): + ranked = sorted( + members, + key=lambda mid: (-candidate_scores.get(mid, 0.0), mid), + ) + seed = candidate_scores.get(ranked[0], 0.0) + raw_clusters.append((seed, ranked[0], tuple(ranked))) + + raw_clusters.sort(key=lambda t: (-t[0], t[1])) + return [ + RetrievalCluster( + cluster_id=i, + member_ids=members, + representative_id=members[0], + seed_score=seed, + ) + for i, (seed, _rep, members) in enumerate(raw_clusters) + ] + + +def pack_with_clusters( + clusters: list[RetrievalCluster], + belief_by_id: dict[str, Belief], + *, + token_budget: int, + cluster_diversity_target: int = DEFAULT_CLUSTER_DIVERSITY_TARGET, + fallback_to_score: bool = True, +) -> list[Belief]: + """Diversity-aware greedy fill at fixed ``token_budget``. + + Stage 1: walk clusters in descending ``seed_score``; pick each + cluster's representative until ``cluster_diversity_target`` distinct + clusters are covered or the budget is exhausted. ``fallback_to_score=True`` + (default) abandons Stage 1 the first time a representative does not + fit the remaining budget; ``False`` skip-but-continues for strict- + diversity benchmarks. + + Stage 2: fill the remaining budget from the score-ranked tail + (members across all clusters in descending seed_score), skipping + beliefs already in the output. + + ``belief_by_id`` must have an entry for every member id in every + cluster; missing ids are silently skipped (treated as "deleted + between rank and pack", same race-handling pattern as the existing + L2.5 pack loop). + """ + out: list[Belief] = [] + used_tokens = 0 + seen: set[str] = set() + covered_clusters: set[int] = set() + + sorted_clusters = sorted(clusters, key=lambda c: -c.seed_score) + + # Stage 1: representatives. + for cluster in sorted_clusters: + if len(covered_clusters) >= cluster_diversity_target: + break + rep_id = cluster.representative_id + if rep_id in seen: + continue + rep = belief_by_id.get(rep_id) + if rep is None: + continue + cost = _belief_tokens(rep) + if used_tokens + cost > token_budget: + if fallback_to_score: + break + continue + out.append(rep) + seen.add(rep_id) + used_tokens += cost + covered_clusters.add(cluster.cluster_id) + + # Stage 2: score-ranked tail. Cluster traversal in descending seed + # order; within a cluster, member_ids[0] is the representative + # (already considered) and member_ids[1:] is the rest in score + # order. Across clusters this is approximately score-order overall. + for cluster in sorted_clusters: + for mid in cluster.member_ids: + if mid in seen: + continue + b = belief_by_id.get(mid) + if b is None: + continue + cost = _belief_tokens(b) + if used_tokens + cost > token_budget: + continue + out.append(b) + seen.add(mid) + used_tokens += cost + + return out diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 2b0e19f6a..6196abb51 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -2432,6 +2432,27 @@ def edges_from(self, src: str) -> list[Edge]: ) return [_row_to_edge(r) for r in cur.fetchall()] + def edges_for_beliefs(self, belief_ids: list[str]) -> list[Edge]: + """Batched edge fetch for clustering (#436). + + Returns every edge whose `src` OR `dst` is in `belief_ids` — + the candidate-induced subgraph plus its boundary. The clusterer + filters down to the candidate-induced subgraph (both endpoints + in the candidate set); the boundary edges come along for free + because the SQL is one read. + + Empty input → empty list (no SQL). + """ + if not belief_ids: + return [] + ph = ",".join("?" * len(belief_ids)) + params = tuple(belief_ids) + tuple(belief_ids) + cur = self._conn.execute( + f"SELECT * FROM edges WHERE src IN ({ph}) OR dst IN ({ph})", + params, + ) + 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 diff --git a/tests/bench_gate/test_intentional_clustering.py b/tests/bench_gate/test_intentional_clustering.py new file mode 100644 index 000000000..111e6a0c4 --- /dev/null +++ b/tests/bench_gate/test_intentional_clustering.py @@ -0,0 +1,65 @@ +"""Bench gate for #436 intentional clustering. + +Spec § A2 (multi-fact recall uplift) + § A3 (single-fact non-regression) ++ § A4 (latency). The full gate evaluates all three; this scaffold runs +the multi-fact corpus through ``cluster_candidates`` + ``pack_with_clusters`` +directly, then checks ``cluster_coverage@k`` against a baseline. + +Public CI skips when ``AELFRICE_CORPUS_ROOT`` is unset (corpus content +lives lab-side per the directory-of-origin rule). The retrieval-side +wiring (``use_intentional_clustering`` flag in ``retrieve_v2``) is the +follow-up gate; this scaffold tests the module independently so the +substrate can land before the wiring. +""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from tests.conftest import load_corpus_module + + +@pytest.mark.bench_gated +def test_multi_fact_corpus_round_trip(aelfrice_corpus_root: Path) -> None: + """Smoke check: the multi_fact corpus parses + every row exposes the + spec § A1 fields. Skips when the directory is empty.""" + rows = load_corpus_module(aelfrice_corpus_root, "multi_fact") + assert rows, "multi_fact corpus produced zero rows" + + for row in rows: + assert "query" in row + assert "expected_belief_ids" in row + assert "expected_clusters" in row + assert "n_clusters_required" in row + assert isinstance(row["expected_clusters"], list) + + +@pytest.mark.bench_gated +def test_clustering_ship_gate_runner_present( + aelfrice_corpus_root: Path, +) -> None: + """The full A2 + A3 ship gate runs from + ``tests.retrieve_uplift_runner.run_clustering_uplift``. This test + skips when the runner is absent — the runner is the operator-side + gate for flipping ``use_intentional_clustering`` to default-on.""" + rows = load_corpus_module(aelfrice_corpus_root, "multi_fact") + assert rows, "multi_fact corpus produced zero rows" + + try: + from tests.retrieve_uplift_runner import ( # noqa: F401 + run_clustering_uplift, + ) + except ImportError: + pytest.skip( + "intentional-clustering uplift runner not yet wired " + "(operator gate; spec § A2 + A3 — pending lab-side corpus + scorer)", + ) + + results = run_clustering_uplift(rows) # type: ignore[name-defined] + assert results.cluster_coverage_uplift > 0, ( + "intentional clustering must show strictly positive cluster_coverage@k uplift\n" + f" ON={results.cluster_coverage_on:.4f} " + f"OFF={results.cluster_coverage_off:.4f} " + f"uplift={results.cluster_coverage_uplift:+.4f}" + ) diff --git a/tests/corpus/v2_0/README.md b/tests/corpus/v2_0/README.md index fd53a11c0..2d12e0ca5 100644 --- a/tests/corpus/v2_0/README.md +++ b/tests/corpus/v2_0/README.md @@ -49,7 +49,9 @@ tests/corpus/v2_0/ │ └── *.jsonl ├── reasoning/ #389 (Track B: aelf reason) │ └── *.jsonl -└── wonder_online/ #389 (Track B: aelf wonder) +├── wonder_online/ #389 (Track B: aelf wonder) +│ └── *.jsonl +└── multi_fact/ #436 (intentional clustering — multi-fact recall) └── *.jsonl ``` @@ -90,6 +92,7 @@ required for **all** modules: | `bfs_potentially_stale` | `beliefs` (list[obj]), `edges` (list[obj]), `seed_ids` (list[string]), `expected_hit_ids` (list[string]), `stale_ids` (list[string]), `k` (int) | `graded` | | `reasoning` | `query` (string), `beliefs` (list[obj]), `edges` (list[obj]), `expected_hit_ids` (list[string]), `baseline_search_only_top_k` (list[string]), `k` (int) | `graded` | | `wonder_online` | `beliefs` (list[obj]), `edges` (list[obj]), `seed_id` (string), `expected_candidate_ids` (list[string]) | `graded` | +| `multi_fact` | `query` (string), `expected_belief_ids` (list[string]), `expected_clusters` (list[list[string]]), `n_clusters_required` (int), `tag` (string) | `graded` | ### `directive_detection` re-entry gate (#374) @@ -239,6 +242,32 @@ stay lab-side per directory-of-origin rules. The bench-gate test at `tests/bench_gate/test_bfs_multihop_tests.py` skips cleanly when the module dir is empty or has fewer rows than the floor below. +### `multi_fact` ship gate (#436) + +Per `docs/feature-intentional-clustering.md` § A2 the intentional +clustering module ships when the multi_fact corpus shows a strictly +positive `cluster_coverage@k` uplift on `use_intentional_clustering=ON` +versus OFF, with no `recall@k` regression. Public CI cannot run this — +labelled rows live lab-side per the directory-of-origin rule. + +Per-row shape: + +- `query` — string. The retrieve_v2 input under test. +- `expected_belief_ids` — non-empty list of belief ids the top-K must + contain. +- `expected_clusters` — list of lists; each inner list is the labeller's + partition of `expected_belief_ids` into one cluster. Two beliefs in + the same inner list are "the same cluster" for the purposes of the + uplift metric. +- `n_clusters_required` — the minimum number of distinct clusters that + must appear in the top-K for the row to count as "covered." +- `tag` — one of `complementary`, `conjunctive`, `sequential`. Free + text describing the multi-fact relationship; used for per-tag + uplift slicing. + +The bench-gate test at `tests/bench_gate/test_intentional_clustering.py` +skips cleanly when the module dir is empty. + ## v0.1 acceptance (per #307) - ≥ 50 non-seed entries per module file (300 total). diff --git a/tests/corpus/v2_0/multi_fact/.gitkeep b/tests/corpus/v2_0/multi_fact/.gitkeep new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_clustering.py b/tests/test_clustering.py new file mode 100644 index 000000000..4860a130a --- /dev/null +++ b/tests/test_clustering.py @@ -0,0 +1,299 @@ +"""Tests for intentional clustering (#436) module + edges_for_beliefs.""" +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from aelfrice.clustering import ( + DEFAULT_CLUSTER_DIVERSITY_TARGET, + DEFAULT_CLUSTER_EDGE_FLOOR, + RetrievalCluster, + cluster_candidates, + pack_with_clusters, +) +from aelfrice.models import ( + BELIEF_FACTUAL, + EDGE_CITES, + EDGE_RELATES_TO, + EDGE_SUPPORTS, + LOCK_NONE, + ORIGIN_AGENT_INFERRED, + RETENTION_FACT, + Belief, + Edge, +) +from aelfrice.store import MemoryStore + + +def _b(bid: str, content: str) -> Belief: + ts = datetime.now(timezone.utc).isoformat() + return Belief( + id=bid, + content=content, + content_hash=f"h-{bid}", + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=LOCK_NONE, + locked_at=None, + demotion_pressure=0, + created_at=ts, + last_retrieved_at=None, + session_id=None, + origin=ORIGIN_AGENT_INFERRED, + retention_class=RETENTION_FACT, + ) + + +def _e(src: str, dst: str, t: str, w: float) -> Edge: + return Edge(src=src, dst=dst, type=t, weight=w) + + +# --- cluster_candidates ------------------------------------------------ + + +def test_empty_candidates_returns_empty_list() -> None: + assert cluster_candidates([], {}, edges=[]) == [] + + +def test_singleton_candidates_become_size_1_clusters() -> None: + """Three isolated candidates → three size-1 clusters, ordered by + descending score.""" + cands = [_b("a", "x"), _b("b", "y"), _b("c", "z")] + scores = {"a": 0.9, "b": 0.7, "c": 0.5} + + clusters = cluster_candidates(cands, scores, edges=[]) + assert [c.cluster_id for c in clusters] == [0, 1, 2] + assert [c.representative_id for c in clusters] == ["a", "b", "c"] + for c in clusters: + assert len(c.member_ids) == 1 + + +def test_two_clusters_via_strong_edge() -> None: + """A SUPPORTS edge (weight≥floor) merges its endpoints into one + cluster. A separate isolated belief stays its own cluster.""" + cands = [_b("a", "x"), _b("b", "y"), _b("c", "z")] + scores = {"a": 0.9, "b": 0.4, "c": 0.6} + edges = [_e("a", "b", EDGE_SUPPORTS, 0.8)] + + clusters = cluster_candidates(cands, scores, edges=edges) + # Cluster {a, b} has seed_score = 0.9 (a's score, the max member). + # Cluster {c} has seed_score = 0.6. + assert len(clusters) == 2 + assert clusters[0].member_ids == ("a", "b") # ranked by score + assert clusters[0].representative_id == "a" + assert clusters[0].seed_score == pytest.approx(0.9) + assert clusters[1].member_ids == ("c",) + assert clusters[1].representative_id == "c" + + +def test_edge_below_floor_does_not_merge() -> None: + """RELATES_TO at weight 0.3 is below the default 0.4 floor.""" + cands = [_b("a", "x"), _b("b", "y")] + scores = {"a": 0.9, "b": 0.7} + edges = [_e("a", "b", EDGE_RELATES_TO, 0.3)] + + clusters = cluster_candidates(cands, scores, edges=edges) + assert len(clusters) == 2 # not merged + + +def test_edge_outside_candidate_pool_is_ignored() -> None: + """Edges with one endpoint outside the candidate set must not + propagate connections (candidate-induced subgraph per spec).""" + cands = [_b("a", "x"), _b("b", "y")] + scores = {"a": 0.9, "b": 0.7} + # Both a and b cite "outside" — but "outside" is not a candidate, + # so a and b must not end up in the same cluster. + edges = [_e("a", "outside", EDGE_CITES, 0.7), + _e("b", "outside", EDGE_CITES, 0.7)] + + clusters = cluster_candidates(cands, scores, edges=edges) + assert len(clusters) == 2 + + +def test_cluster_member_order_descending_score() -> None: + cands = [_b(x, "x") for x in ("a", "b", "c", "d")] + scores = {"a": 0.5, "b": 0.9, "c": 0.7, "d": 0.3} + # Chain: all four merged into one cluster via strong SUPPORTS. + edges = [ + _e("a", "b", EDGE_SUPPORTS, 0.8), + _e("b", "c", EDGE_SUPPORTS, 0.8), + _e("c", "d", EDGE_SUPPORTS, 0.8), + ] + + clusters = cluster_candidates(cands, scores, edges=edges) + assert len(clusters) == 1 + # Sorted by descending score: b > c > a > d. + assert clusters[0].member_ids == ("b", "c", "a", "d") + assert clusters[0].representative_id == "b" + assert clusters[0].seed_score == pytest.approx(0.9) + + +def test_tie_breaking_is_deterministic_by_id_asc() -> None: + cands = [_b("zzz", "x"), _b("aaa", "y")] + scores = {"zzz": 0.5, "aaa": 0.5} + + clusters = cluster_candidates(cands, scores, edges=[]) + assert [c.representative_id for c in clusters] == ["aaa", "zzz"] + + +# --- pack_with_clusters ------------------------------------------------- + + +def test_pack_picks_one_representative_per_cluster_until_target() -> None: + """Stage 1: with diversity_target=2, pack picks the top 2 reps.""" + a = _b("a", "alpha alpha alpha alpha") # ~5 tokens + b = _b("b", "beta beta beta beta") + c = _b("c", "gamma gamma gamma") + cands = [a, b, c] + clusters = [ + RetrievalCluster(0, ("a",), "a", 0.9), + RetrievalCluster(1, ("b",), "b", 0.7), + RetrievalCluster(2, ("c",), "c", 0.5), + ] + + out = pack_with_clusters( + clusters, + {b.id: b for b in cands}, + token_budget=10_000, + cluster_diversity_target=2, + ) + # Stage 1 stops at 2 covered clusters; Stage 2 fills remaining + # budget with non-rep members (none here, so just c gets added too + # because it's a singleton-cluster member in the score-ranked tail). + assert [b.id for b in out] == ["a", "b", "c"] + + +def test_pack_stage1_yields_to_stage2_on_tight_budget() -> None: + """When a cluster representative does not fit the remaining budget, + fallback_to_score=True abandons Stage 1 and Stage 2 fills from + the score-ranked tail. Default behaviour.""" + big = _b("a", "x" * 200) # ~50 tokens + small = _b("b", "y" * 10) # ~3 tokens + cands = [big, small] + clusters = [ + RetrievalCluster(0, ("a",), "a", 0.9), + RetrievalCluster(1, ("b",), "b", 0.5), + ] + # Budget too tight for `big` to fit alongside even one more belief. + out = pack_with_clusters( + clusters, + {b.id: b for b in cands}, + token_budget=5, + ) + # `a` consumes ~50 tokens, doesn't fit at budget=5. + # fallback_to_score=True → Stage 1 abandons after the miss; Stage 2 + # picks `b` from the tail. + assert [b.id for b in out] == ["b"] + + +def test_pack_strict_diversity_skips_oversize_rep_and_continues() -> None: + """fallback_to_score=False keeps trying Stage 1 reps even after a + miss — strict-diversity mode.""" + big = _b("a", "x" * 200) # too big + small = _b("b", "y" * 10) + cands = [big, small] + clusters = [ + RetrievalCluster(0, ("a",), "a", 0.9), + RetrievalCluster(1, ("b",), "b", 0.5), + ] + out = pack_with_clusters( + clusters, + {b.id: b for b in cands}, + token_budget=5, + fallback_to_score=False, + ) + assert [b.id for b in out] == ["b"] + + +def test_pack_fills_stage2_from_remaining_cluster_members() -> None: + """A 3-member cluster: Stage 1 picks the rep; Stage 2 fills the + remaining budget with the other two members in score order.""" + a = _b("a", "x" * 20) + b = _b("b", "y" * 20) + c = _b("c", "z" * 20) + cluster = RetrievalCluster(0, ("a", "b", "c"), "a", 0.9) + + out = pack_with_clusters( + [cluster], + {x.id: x for x in (a, b, c)}, + token_budget=10_000, + cluster_diversity_target=1, + ) + # Stage 1: picks `a` (covers the only cluster, target=1, done). + # Stage 2: walks cluster members in member_ids order, picks b, c. + assert [x.id for x in out] == ["a", "b", "c"] + + +def test_pack_skips_missing_belief_id() -> None: + """Race: a belief id is in a cluster but missing from belief_by_id + (deleted between rank and pack). Pack quietly skips it.""" + cluster = RetrievalCluster(0, ("missing",), "missing", 0.9) + out = pack_with_clusters([cluster], {}, token_budget=10_000) + assert out == [] + + +def test_pack_no_clusters_returns_empty() -> None: + out = pack_with_clusters([], {}, token_budget=10_000) + assert out == [] + + +def test_pack_default_diversity_target_is_three() -> None: + """Smoke test on the default constant — five clusters, default + target=3 means stage 1 picks the top 3 reps.""" + cands = [_b(x, "content " * 5) for x in "abcde"] + clusters = [ + RetrievalCluster(i, (cands[i].id,), cands[i].id, 1.0 - i * 0.1) + for i in range(5) + ] + out = pack_with_clusters( + clusters, + {b.id: b for b in cands}, + token_budget=10_000, + # explicit default for clarity + cluster_diversity_target=DEFAULT_CLUSTER_DIVERSITY_TARGET, + ) + # Stage 1 picks first 3 reps in seed_score order; Stage 2 fills + # remaining budget with the last 2 members. + assert [b.id for b in out[:3]] == ["a", "b", "c"] + assert sorted(b.id for b in out) == list("abcde") + + +# --- edges_for_beliefs -------------------------------------------------- + + +def test_edges_for_beliefs_batched_lookup(tmp_path: Path) -> None: + store = MemoryStore(str(tmp_path / "ef.db")) + try: + for bid in ("a", "b", "c", "d"): + store.insert_belief(_b(bid, "x")) + store.insert_edge(_e("a", "b", EDGE_SUPPORTS, 0.8)) + store.insert_edge(_e("b", "c", EDGE_CITES, 0.5)) + store.insert_edge(_e("d", "a", EDGE_RELATES_TO, 0.3)) + + # Query for {a, b}: should return all three edges (a→b, b→c, d→a) + # because each touches at least one of {a, b}. + out = store.edges_for_beliefs(["a", "b"]) + ids = {(e.src, e.dst, e.type) for e in out} + assert ids == { + ("a", "b", EDGE_SUPPORTS), + ("b", "c", EDGE_CITES), + ("d", "a", EDGE_RELATES_TO), + } + + # Empty input → empty output, no SQL. + assert store.edges_for_beliefs([]) == [] + + # Query for an unrelated id → empty result. + store.insert_belief(_b("loner", "x")) + assert store.edges_for_beliefs(["loner"]) == [] + finally: + store.close() + + +def test_default_edge_floor_excludes_relates_to_and_includes_cites() -> None: + """Defaults are calibrated against `EDGE_VALENCE`: 0.4 floor includes + CITES (0.5) and excludes RELATES_TO (0.3).""" + assert DEFAULT_CLUSTER_EDGE_FLOOR == 0.4 diff --git a/tests/test_corpus_schema.py b/tests/test_corpus_schema.py index e58d6d3aa..b88e33661 100644 --- a/tests/test_corpus_schema.py +++ b/tests/test_corpus_schema.py @@ -171,6 +171,20 @@ "expected_candidate_ids": "list[str]", }, ), + # #436 intentional clustering. cluster_coverage@k uplift on the + # multi-fact corpus; expected_clusters partitions expected_belief_ids + # into the labeller's cluster groupings so the bench can distinguish + # "found two beliefs from one cluster" from "found two clusters." + "multi_fact": ( + {"graded"}, + { + "query": "str", + "expected_belief_ids": "list[str]", + "expected_clusters": "list[list_str]", + "n_clusters_required": "int", + "tag": "str", + }, + ), } COMMON_REQUIRED = ("id", "provenance", "labeller_note", "label") @@ -218,6 +232,17 @@ def _check_field(row: dict, field: str, spec: str, where: str) -> None: f"{where}: {field}[{i}] duplicate belief id {bid!r}" ) row_belief_ids.add(bid) + elif spec == "list[list_str]": + assert isinstance(val, list) and val, ( + f"{where}: field {field!r} must be non-empty list of lists" + ) + for i, inner in enumerate(val): + assert isinstance(inner, list) and inner, ( + f"{where}: {field}[{i}] must be a non-empty list, got {inner!r}" + ) + assert all(isinstance(x, str) and x for x in inner), ( + f"{where}: {field}[{i}] must contain only non-empty strings" + ) elif spec == "list[edge]": assert isinstance(val, list) and val, ( f"{where}: field {field!r} must be non-empty list of edges" From 5465b3cd7fb2d9d7778a85d0e87be0a6b780ddf5 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Fri, 8 May 2026 12:41:37 -0700 Subject: [PATCH 2/2] fix(test): use pytest.importorskip in clustering bench-gate (#436) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeQL flagged 'run_clustering_uplift may be uninitialized' on the try/import + pytest.skip pattern — CodeQL flow analysis can't see that pytest.skip raises. pytest.importorskip is the idiomatic equivalent that returns the module on success and skips otherwise, with no post-import name-binding ambiguity. --- tests/bench_gate/test_intentional_clustering.py | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/tests/bench_gate/test_intentional_clustering.py b/tests/bench_gate/test_intentional_clustering.py index 111e6a0c4..3988ecb14 100644 --- a/tests/bench_gate/test_intentional_clustering.py +++ b/tests/bench_gate/test_intentional_clustering.py @@ -46,17 +46,15 @@ def test_clustering_ship_gate_runner_present( rows = load_corpus_module(aelfrice_corpus_root, "multi_fact") assert rows, "multi_fact corpus produced zero rows" - try: - from tests.retrieve_uplift_runner import ( # noqa: F401 - run_clustering_uplift, - ) - except ImportError: - pytest.skip( + runner_mod = pytest.importorskip( + "tests.retrieve_uplift_runner", + reason=( "intentional-clustering uplift runner not yet wired " - "(operator gate; spec § A2 + A3 — pending lab-side corpus + scorer)", - ) + "(operator gate; spec § A2 + A3 — pending lab-side corpus + scorer)" + ), + ) - results = run_clustering_uplift(rows) # type: ignore[name-defined] + results = runner_mod.run_clustering_uplift(rows) assert results.cluster_coverage_uplift > 0, ( "intentional clustering must show strictly positive cluster_coverage@k uplift\n" f" ON={results.cluster_coverage_on:.4f} "