test(recall): make the arm-candidate-k ceiling test non-vacuous

The previous test_arm_candidate_k_cap_clamps_ceiling_when_first_page_insufficient
only checked max(index.requested) <= 8. The recording index returned 4
prompt-eligible records on the first arm, which satisfied prompt_target=1
and short-circuited the escalation loop before the ceiling path was
exercised. With the cap removed, the test still passed -- the cap was
unverified.

Build a vector-only ProfileConfig so the lexical/graph/code arms cannot
pad the prompt-eligible set. Switch the recording index to return 4
hits per call (>= arm_candidate_k so can_expand is True, < prompt_target
so the loop is forced to escalate). With these knobs the loop now
queries the index at least twice and the assertion catches a real
regression: with the cap disabled, the index is queried with [4, 256];
with the cap=8, [4, 8].

Author: Coding-Dev-Tools

tests/test_recall_arm_candidate_k_cap.py

@@ -40,15 +40,29 @@ def _add(store, emb, wid, rid, text, **kw):
 class _RecordingIndex:
     """Vector-index double that records every arm size it was queried with."""
 
-    def __init__(self):
+    def __init__(self, hit_count: int | None = None, real_ids: list[str] | None = None):
+        """If ``hit_count`` is set, the index always returns exactly that many
+        hits per call (up to ``k``). ``None`` (default) returns the synthetic
+        up-to-4 series; ``0`` returns nothing; any positive int returns that
+        many. ``real_ids`` (if given) replaces the synthetic id prefix so the
+        returned ids resolve to real ``MemoryRecord`` rows in the store --
+        otherwise the prompt-eligibility filter discards them and the
+        escalation loop is short-circuited on empty recs.
+        """
         self.requested: list[int] = []
         self.records: list[tuple[str, float]] = []
+        self._hit_count = hit_count
+        self._real_ids = list(real_ids) if real_ids else None
 
     def search(self, query, k, *, filter=None):
         self.requested.append(int(k))
-        # Return synthetic (id, score) pairs so the prompt-eligible path has
-        # something to score.  Use distinct ids so the loop tests candidate count.
-        return [(f"mem_{i}", float(k - i)) for i in range(min(k, 4))]
+        if self._hit_count == 0:
+            return []
+        cap = min(k, 4) if self._hit_count is None else min(k, self._hit_count)
+        if self._real_ids is not None:
+            return [(self._real_ids[i % len(self._real_ids)], float(k - i))
+                    for i in range(cap)]
+        return [(f"mem_{i}", float(k - i)) for i in range(cap)]
 
 
 def test_arm_candidate_k_cap_default_is_none(monkeypatch):
@@ -106,22 +120,58 @@ def test_arm_candidate_k_cap_clamps_ceiling_when_first_page_insufficient(monkeyp
     """The second page must also be clamped so the escalation loop does not
     silently undo the savings by jumping to PROMPT_ONLY_MIN_CANDIDATES."""
     monkeypatch.setenv("ENGRAPHIS_RECALL_ARM_CANDIDATE_K", "8")
-    index = _RecordingIndex()
+    # Build an engine whose only enabled arm is the vector arm, so the
+    # lexical/graph/code arms cannot pad the prompt-eligible record set
+    # and short-circuit the ceiling path. The recording index returns
+    # exactly 1 hit per call (less than the prompt_target of 2 below and
+    # less than arm_candidate_k=4 so can_expand is True), so the
+    # escalation loop is forced into the arm_candidate_k =
+    # candidate_ceiling branch.
+    from engraphis.core.retrieval_policy import ProfileConfig
+    vector_only = ProfileConfig(
+        name="vector-only-test", vector=True, lexical=False, graph=False, code=False
+    )
     eng = RecallEngine(Store(":memory:"), _SemanticTestEmbedder(256),
-                       index, IdentityReranker())
+                       _RecordingIndex(), IdentityReranker())
     store = eng.store
     wid = store.get_or_create_workspace("w")
+    real_ids = []
     for i in range(20):
-        _add(store, eng.embedder, wid, None, f"fact {i}")
-
-    eng.recall("fact 0", SearchFilter(workspace_id=wid), k=1,
-               candidate_k=1, prompt_only=True)
-
-    # First arm is 1 + min(250, 1*3) = 4, ceiling would normally escalate to
-    # PROMPT_ONLY_MIN_CANDIDATES=256; with cap=8 the ceiling must also be 8.
+        real_ids.append(_add(store, eng.embedder, wid, None, f"fact {i}"))
+
+    index = _RecordingIndex(hit_count=4, real_ids=real_ids)
+    eng.index = index
+
+    eng.recall("zzz-unmatched-query-zzz", SearchFilter(workspace_id=wid), k=8,
+               candidate_k=1, prompt_only=True, arm_config=vector_only)
+
+    # First arm: 1 + min(250, 1*3) = 4 (clamped at min(8, 4) = 4, then
+    # floored at candidate_k=1, so 4). With the cap, the second-page
+    # ceiling is min(256, 8) = 8. Without the cap the index would have
+    # been queried with [4, 256]. The recording index returns 4 hits per
+    # call (>= arm_candidate_k=4 so can_expand=True; < prompt_target=8
+    # so the loop is forced to escalate to the ceiling).
+    assert index.requested, "index was never queried -- test setup is broken"
+    assert index.requested[0] == 4, (
+        f"first arm must be 4 (cap=8, candidate_k=1, prompt_only), "
+        f"got {index.requested[0]}"
+    )
+    assert all(requested <= 8 for requested in index.requested), (
+        f"every index query must respect the cap=8 ceiling, "
+        f"got {index.requested}"
+    )
     assert max(index.requested) <= 8
-    assert all(requested <= 8 for requested in index.requested)
-    # Without the cap the index would have been queried with [4, 256].
+    # Sanity: the loop must have actually escalated, i.e. it must have
+    # queried the index at least twice. If it queried only once, the
+    # first arm satisfied prompt_target and the ceiling-clamp code path
+    # was not exercised -- which is the vacuous case this test guards
+    # against.
+    assert len(index.requested) >= 2, (
+        f"ceiling clamp must be exercised via the escalation loop; "
+        f"only {len(index.requested)} index queries were recorded -- "
+        f"the first arm already satisfied prompt_target and the cap "
+        f"code path is not being run"
+    )
 
 
 def test_arm_candidate_k_cap_floor_protects_one_fact_corpus(monkeypatch):