fix(consume): a withheld concept carries the rule that actually decided it

The quota filters the WHOLE candidate list, so every over-quota candidate
came back `source_quota_exceeded` -- including the ones the RANK had
already put outside k, which the quota only reached because it ran first.

`_fates_without_quota` asks the same cut what would have become of each
candidate with no quota, and the drop keeps THAT rule; only a candidate the
quota-off cut would have delivered is named as the quota's. The packer is
lifted into `_pack` and used by both, so the quota-off fate is decided by
the code the run itself uses and never by a second implementation.

The retrieval gate's row 3 goes 2 of 5 RED to 5 of 5 GREEN. Row 7 is
unchanged at 12 of 14; M01 and M02 lose their row-3 credit, which was the
lying label moving and not the ranking. Suite: 2288 passed, 0 failed; no
committed payload moves a byte.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/llm_ingestion_okf/consume.py

@@ -1873,6 +1873,77 @@ def knapsack(items: Sequence[tuple[float, int]], *, capacity: int) -> tuple[int,
     return max(best, key=lambda entry: entry[0])[1]
 
 
+def _pack(
+    shortlist: Sequence[tuple["Concept", float, dict[str, object], int]],
+    *,
+    limit: int,
+    reserve_top_rank: bool,
+) -> tuple[list[dict[str, object]], list[str], tuple[str, int] | None]:
+    """The budget step, alone: a shortlist in fused-rank order to delivered
+    excerpts, the ids the budget dropped, and the reservation that was made.
+
+    Lifted out of :func:`cut` so the quota-off fate of a dropped candidate is
+    decided by the SAME packer the run itself uses.
+    """
+    # The DP POOL is sorted by `concept_id`, so which of two equal-value subsets
+    # wins is a property of the input set rather than of the order the ranker
+    # happened to emit. The OUTPUT is not: excerpts come back in fused-rank
+    # order, because the rank is what a hit@k measurement reads, and an id-sorted
+    # list would silently turn "position in the payload" into a different number
+    # from "position in the ranking".
+    pool = sorted(shortlist, key=lambda entry: entry[0].concept_id)
+    reserved: tuple[str, int] | None = None
+    room = limit
+    if reserve_top_rank and shortlist:
+        # `shortlist` is in fused-rank order, so its first entry IS the
+        # top-ranked candidate -- not the heaviest, and not the first by id.
+        top = shortlist[0]
+        reserved = (top[0].concept_id, top[3])
+        room = limit - top[3]
+        pool = [entry for entry in pool if entry[0] is not top[0]]
+    capacity = room // WEIGHT_BUCKET
+    packed = {
+        id(pool[index][0])
+        for index in knapsack(
+            tuple((score, -(-weight // WEIGHT_BUCKET)) for _, score, _, weight in pool),
+            capacity=capacity,
+        )
+    }
+    if reserved is not None:
+        packed.add(id(shortlist[0][0]))
+    delivered: list[dict[str, object]] = []
+    over_budget: list[str] = []
+    for concept, _, excerpt, _ in shortlist:
+        if id(concept) in packed:
+            delivered.append({**excerpt, "rank": len(delivered) + 1})
+        else:
+            over_budget.append(concept.concept_id)
+    return delivered, over_budget, reserved
+
+
+def _fates_without_quota(
+    candidates: Sequence[tuple["Concept", float, dict[str, object], int]],
+    *,
+    k: int,
+    limit: int,
+    reserve_top_rank: bool,
+) -> dict[str, str]:
+    """What would have become of each candidate had the quota not run.
+
+    A candidate the quota-off cut would have DELIVERED is the only one the
+    quota can honestly claim: it is the place the quota took. Everything else
+    keeps the rule the quota-off cut gives it, because that is what actually
+    decided it.
+    """
+    fates = {concept.concept_id: "below_k" for concept, _, _, _ in candidates[k:]}
+    delivered, over_budget, _ = _pack(
+        candidates[:k], limit=limit, reserve_top_rank=reserve_top_rank
+    )
+    fates.update({concept_id: "over_budget_after_knapsack" for concept_id in over_budget})
+    fates.update({str(entry["concept_id"]): "source_quota_exceeded" for entry in delivered})
+    return fates
+
+
 def cut(
     ranked: Sequence[tuple[Concept, float, int]],
     *,
@@ -1957,8 +2028,21 @@ def cut(
         # to the quota being off.
         for index in over[: max(k - sum(keep), 0)]:
             keep[index] = True
+        # THE RULE A DROP CARRIES IS TRUE OF THAT DROP. The quota filters the
+        # WHOLE candidate list rather than the top k, so an over-quota
+        # candidate the RANK had already put outside k used to come back
+        # `source_quota_exceeded` -- the quota only reached it because it ran
+        # first. Measured on 25 real misses 2026-09-17: 13 were named by the
+        # quota and decided by the rank.
+        #
+        # The truth is the SAME cut without the quota, computed from the SAME
+        # candidate list by the SAME packer -- never a second implementation
+        # of the rule, which would be two rules with one name.
+        fates = _fates_without_quota(
+            candidates, k=k, limit=limit, reserve_top_rank=reserve_top_rank
+        )
         withheld.extend(
-            (candidates[index][0].concept_id, "source_quota_exceeded")
+            (candidates[index][0].concept_id, fates[candidates[index][0].concept_id])
             for index in range(len(candidates))
             if not keep[index]
         )
@@ -1966,38 +2050,10 @@ def cut(
     for concept, _, _, _ in candidates[k:]:
         withheld.append((concept.concept_id, "below_k"))
     shortlist = candidates[:k]
-    # The DP POOL is sorted by `concept_id`, so which of two equal-value subsets
+    delivered, over_budget, reserved = _pack(
-    # wins is a property of the input set rather than of the order the ranker
+        shortlist, limit=limit, reserve_top_rank=reserve_top_rank
-    # happened to emit. The OUTPUT is not: excerpts come back in fused-rank
-    # order, because the rank is what a hit@k measurement reads, and an id-sorted
-    # list would silently turn "position in the payload" into a different number
-    # from "position in the ranking".
-    pool = sorted(shortlist, key=lambda entry: entry[0].concept_id)
-    reserved: tuple[str, int] | None = None
-    room = limit
-    if reserve_top_rank and shortlist:
-        # `shortlist` is in fused-rank order, so its first entry IS the
-        # top-ranked candidate -- not the heaviest, and not the first by id.
-        top = shortlist[0]
-        reserved = (top[0].concept_id, top[3])
-        room = limit - top[3]
-        pool = [entry for entry in pool if entry[0] is not top[0]]
-    capacity = room // WEIGHT_BUCKET
-    packed = {
-        id(pool[index][0])
-        for index in knapsack(
-            tuple((score, -(-weight // WEIGHT_BUCKET)) for _, score, _, weight in pool),
-            capacity=capacity,
     )
-    }
+    withheld.extend((concept_id, "over_budget_after_knapsack") for concept_id in over_budget)
-    if reserved is not None:
-        packed.add(id(shortlist[0][0]))
-    delivered: list[dict[str, object]] = []
-    for concept, _, excerpt, _ in shortlist:
-        if id(concept) in packed:
-            delivered.append({**excerpt, "rank": len(delivered) + 1})
-        else:
-            withheld.append((concept.concept_id, "over_budget_after_knapsack"))
     withheld.sort()
     return tuple(delivered), tuple(withheld), reserved
 

tests/test_retrieval_gate.py

@@ -230,9 +230,22 @@ def test_a_miss_with_no_class_at_all_takes_the_whole_row_to_zero(tmp_path: Path)
 # --- row 3 --------------------------------------------------------------------
 
 
-def test_row_three_is_red_because_the_payload_names_the_quota_for_a_rank(
+def test_row_three_goes_red_again_the_moment_the_label_stops_being_true(
-    tmp_path: Path,
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    """The RED direction, driven by an input: the library's own behaviour.
+
+    This row was red on the shipped code until 2026-09-20 -- every over-quota
+    candidate came back `source_quota_exceeded`, including the ones the RANK
+    had already put outside k. Restoring exactly that naming puts the row back
+    where it was, with the same three fixtures and the same detail line, so
+    the green reading is not a green nobody can lose.
+    """
+
+    def every_drop_is_the_quotas(candidates: object, **_: object) -> dict[str, str]:
+        return _AlwaysTheQuota()
+
+    monkeypatch.setattr(consume, "_fates_without_quota", every_drop_is_the_quotas)
     row = gate.row_three([_case(tmp_path, "set-classes.json")])
     assert row.status == gate.RED
     lying = [detail for detail in row.details if "source_quota_exceeded" in detail]
@@ -240,6 +253,23 @@ def test_row_three_is_red_because_the_payload_names_the_quota_for_a_rank(
     assert all("the quota-off run says `below_k`" in detail for detail in lying)
 
 
+class _AlwaysTheQuota(dict[str, str]):
+    """The pre-2026-09-20 naming: every drop the quota reached is the quota's."""
+
+    def __missing__(self, key: str) -> str:
+        return "source_quota_exceeded"
+
+
+def test_row_three_is_green_on_the_shipped_code(tmp_path: Path) -> None:
+    """And the same three fixtures, unmutated, are the green direction.
+
+    Four judged units in this set alone; the fifth in the gate's own run comes
+    from `set-miss.json`."""
+    row = gate.row_three([_case(tmp_path, "set-classes.json")])
+    assert (row.k, row.m, row.status) == (4, 4, gate.GREEN)
+    assert not [detail for detail in row.details if "source_quota_exceeded" in detail]
+
+
 def test_row_three_is_green_when_the_printed_reason_is_the_true_one(tmp_path: Path) -> None:
     # The budget case: withheld by the pack in BOTH runs, so the label the
     # payload prints is the label the quota-off run confirms.
@@ -455,19 +485,28 @@ def test_row_seven_is_red_when_a_mutant_survives(tmp_path: Path) -> None:
     )
 
 
-def test_a_mutant_that_only_makes_a_red_row_greener_is_not_felled(tmp_path: Path) -> None:
+def test_a_mutant_is_felled_by_the_row_that_got_worse_and_never_by_one_that_did_not(
-    # The definition, held by a test: removing the quota makes row 3's label
+    tmp_path: Path,
-    # honest, because there is then no quota to name. Calling that a kill
+) -> None:
-    # would credit this gate with a check it does not have.
+    """PM's definition, held by a test: `worse`, never `different`.
+
+    Removing the quota costs row 1 deliveries -- that is the kill. It costs
+    row 3 nothing, because a cut with no quota has no quota to name falsely,
+    and crediting row 3 with the kill would give this gate a check it does not
+    have. Before 2026-09-20 the same mutation made row 3 GREENER, which was
+    the sharper form of the same point.
+    """
     cases, _ = gate.synthetic_cases(tmp_path / "bundles", FIXTURES)
-    baseline = gate.deterministic_rows(cases)
+    before = {row.number: row.k for row in gate.deterministic_rows(cases)}
-    before = {row.number: (row.k, row.m) for row in baseline}
     with gate._wrap_cut(source_quota=None):
         rows = gate.deterministic_rows(
             [gate.measure_case(case.question_set, case.bundles) for case in cases]
         )
-    after = {row.number: (row.k, row.m) for row in rows}
+    after = {row.number: row.k for row in rows}
-    assert after[3][0] - after[3][1] > before[3][0] - before[3][1]
+    worse = [number for number in sorted(before) if after[number] < before[number]]
+    assert 1 in worse, "row 1 is what fells this mutant"
+    assert 3 not in worse, after
+    assert after[3] >= before[3]
 
 
 # --- rows 8 and 9 -------------------------------------------------------------
@@ -600,12 +639,12 @@ def test_the_gate_is_red_today_and_says_which_rows(tmp_path: Path) -> None:
     rows = gate.evaluate(tmp_path / "bundles")
     by_number = {row.number: row for row in rows}
     assert sorted(by_number) == [1, 2, 3, 4, 5, 6, 7, 8, 9]
-    assert [row.number for row in rows if row.fails] == [3, 4, 5, 7, 8, 9]
+    assert [row.number for row in rows if row.fails] == [4, 5, 7, 8, 9]
     # 10, not 9: `set-quota.json` adds row 3's known-positive, one question the
     # source quota genuinely decides, and it is a hit.
     assert (by_number[1].k, by_number[1].m) == (10, 10)
     assert (by_number[2].k, by_number[2].m) == (7, 7)
-    assert (by_number[3].k, by_number[3].m) == (2, 5)
+    assert (by_number[3].k, by_number[3].m) == (5, 5)
     assert (by_number[6].k, by_number[6].m) == (10, 10)
 
 
@@ -622,7 +661,7 @@ def test_the_command_exits_one_and_prints_every_row(
     printed = capsys.readouterr().out
     for number in range(1, 10):
         assert f"\n{number} " in f"\n{printed}"
-    assert "GATE RED: rows 3, 4, 5, 7, 8, 9" in printed
+    assert "GATE RED: rows 4, 5, 7, 8, 9" in printed
 
 
 def test_the_json_form_carries_the_same_rows(capsys: pytest.CaptureFixture[str]) -> None: