qa/tests: cover capacity metrics + retrieval-keywords across all 3 layers

Three layers, 14 new tests, full suite 611 passed (was 597):

UNIT — tests/test_query.py
  test_query_returns_prompt_chars_breakdown
    Asserts the result dict's prompt_chars carries exactly the five
    expected keys and messages_total equals sum of message contents
    the StubClient saw.
  test_query_answer_chars_matches_answer_text
    answer_chars == len(answer_text) — drift check.
  test_query_cache_hit_also_returns_capacity_metrics
    Cache-hit path populates prompt_chars + answer_chars (operators
    inspecting cached records still want the breakdown).
  test_query_evidence_chars_grows_with_topk
    Sanity: more sources / larger budget → more evidence chars
    (the metric tracks actual context build, not a stale constant).

INTEGRATION — tests/test_query.py (retrieval_keywords)
  test_retrieval_keywords_does_not_alter_question_to_llm
    Keywords don't appear in the LLM-facing question segment;
    system prompt unchanged across runs. Pins the substrate
    contract: keywords are FTS5/title-filter-only.
  test_retrieval_keywords_changes_retrieved_sources
    Different keyword sets surface different docs (the actual
    user-visible behavior).

UNIT — tests/test_cli_render.py
  test_render_shows_capacity_line_when_prompt_chars_present
    Capacity line appears with messages_total + breakdown when
    prompt_chars is in the result dict.
  test_render_omits_capacity_line_on_legacy_results_without_prompt_chars
    Backwards-compat: legacy results render cleanly without the
    capacity line — no KeyError, no '0 chars' noise.
  test_render_capacity_thousand_separators
    61,550 not 61550 — operator legibility on daily renders.

UNIT/INTEGRATION — tests/test_bench_qa_sweep.py (NEW FILE)
  Imports bench/qa_sweep.py via importlib.util so the module's
  not in the Python path doesn't matter. Five tests:
    - _summarize counts verdicts by mode
    - deflections counted only on STRICT/HYBRID rows (not UNGROUNDED)
    - rendered markdown has the headline summary + size buckets
    - size buckets correctly stratify strict-rate by prompt_chars_total
    - empty buckets are skipped (no '0 runs' noise)

FUNCTIONAL — live verification (no automated test, manual)
  `make query Q="what is the capital of france?" BURN=1` confirmed
  in commit f927298 to render the capacity one-liner under the
  source list. Documented in that commit's body.

Also corrected the docstring on query()'s `retrieval_keywords` to
reflect that keywords don't enter cache_key DIRECTLY but do change
context_root + conversation_hash via source selection — so the same
question with different keywords lands under different cache_keys
(legitimately, since the LLM saw different contexts).

tests/test_bench_qa_sweep.py

@@ -0,0 +1,165 @@
+"""Bench-side aggregation tests for capacity metrics + deflection rate.
+
+The bench harness in ``bench/qa_sweep.py`` is the analytics surface for
+QA-quality work — it consumes per-row ``prompt_chars_*`` / ``answer_chars``
+fields from the query result and renders a markdown summary including
+strict-rate-by-size buckets. These tests pin the aggregation behavior on
+synthetic rows so the rendering doesn't drift silently.
+"""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture(scope="module")
+def qa_sweep():
+    """Import bench/qa_sweep.py as a module despite living outside the
+    aborist package. Avoids polluting the package or requiring a
+    pip-install of the bench harness."""
+    bench_path = Path(__file__).parent.parent / "bench" / "qa_sweep.py"
+    spec = importlib.util.spec_from_file_location("qa_sweep_under_test", bench_path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _row(**overrides) -> dict:
+    """Synthetic bench-row fixture with sensible defaults for every
+    field the summarizer/renderer reads."""
+    base = {
+        "question": "q",
+        "answer_mode": "claim_lattice",
+        "status": "cache_miss_then_written",
+        "audit_mode": "STRICT",
+        "n_quotes": 1,
+        "n_verified": 1,
+        "ratio": 1.0,
+        "verifier_method": "claim_lattice",
+        "lookup_path": "miss",
+        "failure_stage": None,
+        "lazy_anchor_ratio": None,
+        "pointer_id_distribution": None,
+        "cache_key": "abc123",
+        "n_sources": 1,
+        "elapsed_s": 1.0,
+        "deflection_kind": "on_topic",
+        "subject_anchor": "x",
+        "subject_in_answer": True,
+        "prompt_chars_total": 5000,
+        "prompt_chars_evidence": 4000,
+        "prompt_chars_system": 800,
+        "prompt_chars_question": 30,
+        "answer_chars": 100,
+        "error": None,
+    }
+    base.update(overrides)
+    return base
+
+
+def test_summarize_counts_verdicts_by_mode(qa_sweep):
+    rows = [
+        _row(answer_mode="quote", audit_mode="STRICT"),
+        _row(answer_mode="quote", audit_mode="HYBRID"),
+        _row(answer_mode="claim_lattice", audit_mode="STRICT"),
+        _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED"),
+    ]
+    summary = qa_sweep._summarize(rows)
+    assert summary["quote"]["STRICT"] == 1
+    assert summary["quote"]["HYBRID"] == 1
+    assert summary["claim_lattice"]["STRICT"] == 1
+    assert summary["claim_lattice"]["UNGROUNDED"] == 1
+
+
+def test_summarize_counts_deflections_only_on_grounded_rows(qa_sweep):
+    """A deflection_kind=='deflection' on UNGROUNDED isn't a real
+    topic shift (no answer to deflect with). Only STRICT/HYBRID rows
+    with deflection_kind='deflection' should count toward the
+    deflections column."""
+    rows = [
+        _row(answer_mode="claim_lattice", audit_mode="STRICT", deflection_kind="deflection"),
+        _row(answer_mode="claim_lattice", audit_mode="HYBRID", deflection_kind="deflection"),
+        _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED", deflection_kind="deflection"),
+        _row(answer_mode="claim_lattice", audit_mode="STRICT", deflection_kind="on_topic"),
+    ]
+    summary = qa_sweep._summarize(rows)
+    # 2 grounded deflections (STRICT + HYBRID); UNGROUNDED + on_topic don't count.
+    assert summary["claim_lattice"]["deflections"] == 2
+
+
+def test_render_markdown_includes_summary_table_with_capacity_metrics(qa_sweep):
+    rows = [
+        _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=5000),
+        _row(answer_mode="claim_lattice", audit_mode="HYBRID", prompt_chars_total=10000),
+    ]
+    summary = qa_sweep._summarize(rows)
+    md = qa_sweep._render_markdown(
+        rows, summary, "2026-05-01T00-00-00Z",
+        ["claim_lattice"], ["q"], n_samples=2,
+    )
+    # Headline table has the deflections column (added before capacity work).
+    assert "deflections" in md
+    assert "strict-rate" in md
+    # Strict-rate-by-prompt-size section exists.
+    assert "strict-rate by prompt size" in md
+    # Buckets are present in the markdown body.
+    assert "<8KB" in md
+    assert "8-16KB" in md
+
+
+def test_render_markdown_buckets_strict_rate_by_size(qa_sweep):
+    """Two STRICT rows in <8KB bucket + two UNGROUNDED rows in
+    8-16KB bucket → <8KB strict-rate=1.0, 8-16KB strict-rate=0.0."""
+    rows = [
+        _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=4000),
+        _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=5000),
+        _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED", prompt_chars_total=10000),
+        _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED", prompt_chars_total=12000),
+    ]
+    summary = qa_sweep._summarize(rows)
+    md = qa_sweep._render_markdown(
+        rows, summary, "2026-05-01T00-00-00Z",
+        ["claim_lattice"], ["q"], n_samples=4,
+    )
+    # Find the size-bucket section; verify both buckets show their
+    # expected strict-rate. Markdown table cell format is `| value |`.
+    bucket_section = md.split("strict-rate by prompt size")[1]
+    # <8KB row: 2 runs, 2 STRICT, rate 1.00
+    lines_8kb = [
+        L for L in bucket_section.splitlines()
+        if "<8KB" in L and "claim_lattice" in L
+    ]
+    assert lines_8kb, f"missing <8KB row in:\n{bucket_section[:600]}"
+    assert " 1.00 " in lines_8kb[0]
+    # 8-16KB row: 2 runs, 0 STRICT, rate 0.00
+    lines_16kb = [
+        L for L in bucket_section.splitlines()
+        if "8-16KB" in L and "claim_lattice" in L
+    ]
+    assert lines_16kb, f"missing 8-16KB row in:\n{bucket_section[:600]}"
+    assert " 0.00 " in lines_16kb[0]
+
+
+def test_render_markdown_skips_empty_buckets(qa_sweep):
+    """Bucket with zero runs in a mode shouldn't render an empty row
+    — the table should only include cells with data."""
+    rows = [
+        _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=5000),
+    ]
+    summary = qa_sweep._summarize(rows)
+    md = qa_sweep._render_markdown(
+        rows, summary, "2026-05-01T00-00-00Z",
+        ["claim_lattice"], ["q"], n_samples=1,
+    )
+    bucket_section = md.split("strict-rate by prompt size")[1]
+    # Only <8KB has data; >=64KB / 32-64KB / 16-32KB / 8-16KB rows
+    # should be absent.
+    for missing_label in (">=64KB", "32-64KB", "16-32KB", "8-16KB"):
+        assert missing_label not in bucket_section, (
+            f"empty bucket {missing_label} leaked into rendering:\n"
+            f"{bucket_section[:600]}"
+        )

tests/test_cli_render.py

@@ -146,3 +146,60 @@ def test_json_mode_also_uses_unicode():
     s = json.dumps(payload, ensure_ascii=False)
     assert "Pokémon" in s
     assert "\\u00e9" not in s
+
+
+# ---------------------------------------------------------------------------
+# capacity metrics — human render surfaces prompt_chars breakdown
+# ---------------------------------------------------------------------------
+
+
+def test_render_shows_capacity_line_when_prompt_chars_present():
+    """When the result carries `prompt_chars`, the human render
+    emits a one-liner summarizing prompt size + answer size so an
+    operator can tell at a glance whether STRICT came from a small
+    prompt or a context-stuffed one."""
+    r = _result(
+        prompt_chars={
+            "system_prompt": 1292,
+            "grounding_reminder": 342,
+            "user_question": 30,
+            "evidence_or_context": 20304,
+            "messages_total": 21996,
+        },
+        answer_chars=247,
+    )
+    out = _render_query_human(r, "what is the capital of france?")
+    assert "capacity:" in out
+    assert "21,996 chars" in out  # messages_total
+    assert "sys 1,292" in out
+    assert "evidence 20,304" in out
+    assert "answer 247 chars" in out
+
+
+def test_render_omits_capacity_line_on_legacy_results_without_prompt_chars():
+    """Backwards-compat: results that pre-date capacity metrics (no
+    prompt_chars key) render without the capacity line — no KeyError,
+    no awkward 'capacity: 0 chars' noise."""
+    r = _result()  # no prompt_chars
+    out = _render_query_human(r, "q")
+    assert "capacity:" not in out
+
+
+def test_render_capacity_thousand_separators():
+    """Large prompts must format with thousand-separator commas so
+    a 60K prompt reads as '60,000' not '60000' — operator legibility
+    on the daily render."""
+    r = _result(
+        prompt_chars={
+            "system_prompt": 1500,
+            "grounding_reminder": 0,
+            "user_question": 50,
+            "evidence_or_context": 60000,
+            "messages_total": 61550,
+        },
+        answer_chars=1234,
+    )
+    out = _render_query_human(r, "q")
+    assert "61,550 chars" in out
+    assert "60,000" in out
+    assert "1,234 chars" in out

tests/test_query.py

@@ -719,3 +719,220 @@ def test_query_persists_audit_event(tmp_path):
         qc.close()
     types = [e["event_type"] for e in events]
     assert "providence_query" in types
+
+
+# ---------------------------------------------------------------------------
+# capacity metrics — prompt_chars + answer_chars surface
+# ---------------------------------------------------------------------------
+
+
+def _setup_corpus(tmp_path):
+    main_db = tmp_path / "corpus.db"
+    qa_db = tmp_path / "qa.db"
+    conn = connect(main_db)
+    try:
+        ingest_source(conn, FakeSource(DOCS))
+    finally:
+        conn.close()
+    return main_db, qa_db
+
+
+def test_query_returns_prompt_chars_breakdown(tmp_path):
+    """prompt_chars must contain the five expected keys and
+    messages_total must equal the sum of system + reminder + user
+    message lengths actually sent to the chat client."""
+    main_db, qa_db = _setup_corpus(tmp_path)
+    client = StubClient(answer="A short answer.")
+    result = query(
+        question="What is anarcho-capitalism?",
+        qa_db=qa_db,
+        chat_client=client,
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+    )
+
+    pc = result["prompt_chars"]
+    assert set(pc.keys()) == {
+        "system_prompt",
+        "grounding_reminder",
+        "user_question",
+        "evidence_or_context",
+        "messages_total",
+    }
+    # System prompt is from DEFAULT_QUERY_POLICY — non-empty.
+    assert pc["system_prompt"] > 0
+    # Question is exactly the chars passed in.
+    assert pc["user_question"] == len("What is anarcho-capitalism?")
+    # Evidence/context is the chunk text the model saw — non-trivial.
+    assert pc["evidence_or_context"] > 0
+    # messages_total equals the sum of message contents the stub saw.
+    sent = client.calls[0]["messages"]
+    actual_total = sum(len(m["content"]) for m in sent)
+    assert pc["messages_total"] == actual_total
+
+
+def test_query_answer_chars_matches_answer_text(tmp_path):
+    main_db, qa_db = _setup_corpus(tmp_path)
+    client = StubClient(answer="Exactly twenty-six chars!!")
+    result = query(
+        question="What is anarchism?",
+        qa_db=qa_db,
+        chat_client=client,
+        model_id="m",
+        single_db=main_db,
+    )
+    assert result["answer_chars"] == len(result["answer_text"])
+
+
+def test_query_cache_hit_also_returns_capacity_metrics(tmp_path):
+    """Cache-hit path must populate prompt_chars too — operators
+    inspecting cached records still want the capacity breakdown."""
+    main_db, qa_db = _setup_corpus(tmp_path)
+    client = StubClient(answer="Same answer twice.")
+    first = query(
+        question="What is anarcho-capitalism?",
+        qa_db=qa_db,
+        chat_client=client,
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+    )
+    assert first["status"] == "cache_miss_then_written"
+    second = query(
+        question="What is anarcho-capitalism?",
+        qa_db=qa_db,
+        chat_client=client,
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+    )
+    assert second["status"] == "cache_hit"
+    assert "prompt_chars" in second
+    assert second["prompt_chars"]["messages_total"] > 0
+    assert second["answer_chars"] > 0
+
+
+def test_query_evidence_chars_grows_with_topk(tmp_path):
+    """Adding more sources to the context budget should increase
+    `evidence_or_context`. Sanity check that the metric tracks
+    actual context build, not a stale constant."""
+    main_db, qa_db = _setup_corpus(tmp_path)
+
+    qa_db_small = tmp_path / "qa_small.db"
+    qa_db_big = tmp_path / "qa_big.db"
+
+    small = query(
+        question="What is anarchism?",
+        qa_db=qa_db_small,
+        chat_client=StubClient(answer="X"),
+        model_id="m",
+        single_db=main_db,
+        top_k=1,
+        max_context_chars=2000,
+    )
+    big = query(
+        question="What is anarchism?",
+        qa_db=qa_db_big,
+        chat_client=StubClient(answer="X"),
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+        max_context_chars=20000,
+    )
+    assert big["prompt_chars"]["evidence_or_context"] >= small["prompt_chars"]["evidence_or_context"]
+
+
+# ---------------------------------------------------------------------------
+# --retrieval-keywords flag (operator-supplied retrieval augmentation)
+# ---------------------------------------------------------------------------
+
+
+def test_retrieval_keywords_does_not_alter_question_to_llm(tmp_path):
+    """The LLM must see the original question string, not the
+    augmented retrieval query. Cache_key must match the same call
+    without retrieval_keywords."""
+    main_db, qa_db = _setup_corpus(tmp_path)
+
+    captured = {}
+
+    def _record_answer(messages, **kwargs):
+        captured["messages"] = messages
+        return "Some answer."
+
+    client = StubClient(answer=_record_answer)
+    bare = query(
+        question="What is anarchism?",
+        qa_db=qa_db,
+        chat_client=client,
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+    )
+
+    # Same question, with retrieval keywords. New qa_db so cache lookup
+    # under fresh state — both runs are cache-miss.
+    qa_db_2 = tmp_path / "qa_with_keywords.db"
+    client2 = StubClient(answer=_record_answer)
+    augmented = query(
+        question="What is anarchism?",
+        qa_db=qa_db_2,
+        chat_client=client2,
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+        retrieval_keywords="mutual aid theory",
+    )
+
+    # User question text is byte-identical between the two runs.
+    bare_user_msg = [m["content"] for m in client.calls[0]["messages"]
+                     if m["role"] == "user"][-1]
+    aug_user_msg = [m["content"] for m in client2.calls[0]["messages"]
+                    if m["role"] == "user"][-1]
+    # Both end with "Question: What is anarchism?" — the keywords
+    # never appear inside the LLM-facing question segment (they
+    # only flow into FTS5 retrieval + title-filter token sets).
+    assert "What is anarchism?" in bare_user_msg
+    assert "What is anarchism?" in aug_user_msg
+    assert "mutual aid theory" not in aug_user_msg.split("Question:")[-1]
+    # System prompt unchanged across runs.
+    bare_sys = [m["content"] for m in client.calls[0]["messages"]
+                if m["role"] == "system"][0]
+    aug_sys = [m["content"] for m in client2.calls[0]["messages"]
+               if m["role"] == "system"][0]
+    assert bare_sys == aug_sys
+
+
+def test_retrieval_keywords_changes_retrieved_sources(tmp_path):
+    """Different keywords surface different docs, even with the
+    same question text. Same-keywords across runs is reproducible."""
+    main_db, qa_db = _setup_corpus(tmp_path)
+    qa_db_a = tmp_path / "a.db"
+    qa_db_b = tmp_path / "b.db"
+    a = query(
+        question="define the system",
+        qa_db=qa_db_a,
+        chat_client=StubClient(answer="X"),
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+        retrieval_keywords="anarcho-capitalism rothbard",
+    )
+    b = query(
+        question="define the system",
+        qa_db=qa_db_b,
+        chat_client=StubClient(answer="X"),
+        model_id="m",
+        single_db=main_db,
+        top_k=3,
+        retrieval_keywords="capitalism markets",
+    )
+    a_uris = sorted(s["document_uri"] for s in a["sources"])
+    b_uris = sorted(s["document_uri"] for s in b["sources"])
+    # The two retrieval-keyword sets pull different topical sources.
+    # (Sanity: same question without keywords is the no-keyword
+    # baseline; the assertion is that adding distinct keywords does
+    # something to the result.)
+    assert a_uris != b_uris or set(a_uris) != set(b_uris), (
+        f"keywords should affect retrieval; got identical sources {a_uris}"
+    )