test(consume, retrieval-gate): the payload must say what of the question it covered, and row 4 must see it

Red first, on behaviour: three asserts about what the payload carries and
what row 4 scores, none on an import or an attribute.

Measured 2026-09-19: N3, N4 and N5 come back with 8, 8 and 1 excerpts and
nothing that says they are weak, so `marked = nothing delivered` reads
three uncovered questions as answered ones. The second gate test is the
known-negative that a marking firing on everything would fail: the 10
questions the three synthetic sets DO answer must stay unmarked.

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

tests/test_okf_consume.py

@@ -2646,3 +2646,75 @@ def test_a_drop_the_rank_had_already_made_is_not_named_as_the_quotas() -> None:
     # And the two the rank had already lost say so.
     assert rules["c4"] == "below_k"
     assert rules["c5"] == "below_k"
+
+
+# --- what the payload says about the question it was cut for -----------------
+
+
+def test_the_payload_names_the_question_terms_the_bundle_answers_to_none_of() -> None:
+    """A consumer holding eight excerpts cannot tell a bundle that ANSWERED
+    its question from one that merely ranked something.
+
+    The payload carries no verdict about that -- see `build_payload` for the
+    two measured falsifications -- but it carries the facts a reader needs:
+    the question's own terms, the ones no concept in the bundle answers, and
+    the ones no delivered excerpt answers.
+    """
+    payload = okf_consume.build_payload(
+        FIXTURE, question="Hva koster et doegn paa hytta for gjester?"
+    )
+    assert "coverage" in payload, "the payload says nothing about what it covered"
+    coverage = payload["coverage"]
+    assert isinstance(coverage, dict)
+    # The terms are the ranker's own, counted here rather than read back.
+    expected = list(
+        dict.fromkeys(okf_consume.normalise("Hva koster et doegn paa hytta for gjester?"))
+    )
+    assert coverage["question_terms"] == expected
+    assert expected, "a question with no terms would make every assertion below vacuous"
+    # Measured independently: the bundle's own searchable text, term by term.
+    concepts = [
+        okf_consume.read_concept(
+            FIXTURE / f"{concept_id}.md",
+            bundle_root=FIXTURE,
+            root_bundle_id=okf_consume.root_bundle_id_of(FIXTURE),
+        )
+        for concept_id in okf_consume.enumerate_concepts(FIXTURE)
+    ]
+    texts = okf_consume.searchable_text(concepts)
+    stems = frozenset(token for text in texts for token in okf_consume.normalise(text))
+    absent = [
+        term
+        for term in expected
+        if not any(
+            okf_consume.tokens_match(term, other, stems=stems)
+            for text in texts
+            for other in okf_consume.normalise(text)
+        )
+    ]
+    assert coverage["unanswered_in_bundle"] == absent
+    # KNOWN-POSITIVE and KNOWN-NEGATIVE in the same measurement: this question
+    # is answered by no concept here, and some of its terms are.
+    assert absent, "every term is in the bundle, so the list under test is empty"
+    assert len(absent) < len(expected), "no term is in the bundle; the list cannot discriminate"
+
+
+def test_a_question_the_bundle_answers_leaves_the_unanswered_lists_short() -> None:
+    """The control: a question the fixture bundle does answer.
+
+    Without it, a payload reporting every term as unanswered would pass the
+    test above.
+    """
+    payload = okf_consume.build_payload(FIXTURE, question="Hvordan skal prisene fylles ut?")
+    assert "coverage" in payload, "the payload says nothing about what it covered"
+    coverage = payload["coverage"]
+    assert isinstance(coverage, dict)
+    terms = coverage["question_terms"]
+    assert isinstance(terms, list) and terms
+    in_payload = [
+        term
+        for term in terms
+        if term not in coverage["unanswered_in_payload"]  # type: ignore[operator]
+    ]
+    assert in_payload, "the delivered excerpts answer nothing of a question they were cut for"
+    assert len(coverage["unanswered_in_bundle"]) < len(terms)  # type: ignore[arg-type]

tests/test_retrieval_gate.py

@@ -1287,3 +1287,49 @@ def test_the_threshold_is_compared_with_the_measured_hold_out(tmp_path: Path) ->
     )
     assert any("clears the threshold: NO" in detail for detail in falls.details)
     assert (falls.k, falls.m, falls.status) == (10, 11, gate.RED)
+
+
+# --- step 1: row 4, the marking a consumer can act on -------------------------
+
+
+def test_row_four_marks_every_uncovered_control_and_leaves_the_covered_one(
+    tmp_path: Path,
+) -> None:
+    """Six controls, five uncovered and one covered, and the row must get all
+    six right.
+
+    Three of the five were measured red on 2026-09-19: N3, N4 and N5 came back
+    with 8, 8 and 1 excerpts and nothing in the payload that said they were
+    weak, so `marked = nothing delivered` could not see them.
+    """
+    case = _case(tmp_path, "set-controls.json")
+    # The denominator, counted from the pinned file rather than from the run.
+    declared = json.loads((FIXTURES / "set-controls.json").read_text(encoding="utf-8"))
+    assert len(declared["controls"]) == 6
+    assert sum(1 for control in declared["controls"] if control.get("covered")) == 1
+    row = gate.row_four([case])
+    assert (row.k, row.m, row.status) == (6, 6, gate.GREEN)
+    assert not row.details
+
+
+def test_the_marking_stays_off_every_question_the_synthetic_sets_do_answer(
+    tmp_path: Path,
+) -> None:
+    """The known-negative, and it is wider than row 4's own denominator.
+
+    A marking that fires on the questions the bundles DO answer would take row
+    4 to 6 of 6 and be worthless. Every question of the three sets whose fasit
+    row 1 finds is measured here, and none of them may come back marked.
+    """
+    bundles = _bundles(tmp_path)
+    answered = 0
+    for name in ("set-positive.json", "set-signals.json", "set-quota.json"):
+        question_set = gate.load_set(FIXTURES / name, gate.SYNTHETIC_SETS[name])
+        for question in question_set.questions:
+            bundle = bundles[question.bundle or question_set.bundle]
+            payload = consume.build_payload(
+                bundle, question=question.question, k=question.k, limit=question.limit
+            )
+            assert not gate.marked(payload), f"{question_set.set_id}/{question.id}"
+            answered += 1
+    assert answered == 10, answered