feat(retrieval): every semantic search hands on the passage that matched

#4243 fixed one door. Scribe has three semantic searches over three chunk
tables, and all three collapsed chunk rows to the best one per record — each
of them KNEW which passage earned the hit, and each dropped it. Every surface
downstream then previewed the head of the document instead: a span the search
had already scored lower, with nothing saying so.

Mechanism, one place:
  - embeddings.record_best_chunk publishes {id: {index, text}} into `report`.
    Carried in `report`, NOT the return value: all three return
    list[tuple[float, Record]] and ~30 sites unpack that pair (lesson #4207).
  - semantic_search_rules and semantic_search_milestones now select
    chunk_index/chunk_text and publish the winner, as notes already did.
    semantic_search_milestones gains `report`, which it had no way to take.
  - services/text.matched_excerpt is the one choice of span, and
    excerpt_fields the one result block. Doors keep their own field names —
    the web renders `snippet`, MCP returns `excerpt` — because renaming a
    field a frontend reads is a different change from fixing what goes in it.

Surfaces:
  - knowledge.query_knowledge, whose own comment calls it "the human's MAIN
    search surface", was `(note.body or "")[:200]` on every row alike. Now the
    matched passage on a search, the opening on a browse, and `snippet_is`
    saying which. KnowledgeView renders that snippet, so this was live.
  - search(content_type='milestone') gains `matched` — the plan body stays
    out, but the passage that matched comes along, because recognising a plan
    means recognising the part you asked about and a description written at
    the start need not mention it.
  - The auto-inject menu and the write-path prior-art menu put the passage
    under their line. Both were title-only, which answers "does this apply?"
    for a lesson or snippet (the trigger is IN the title) and not at all for
    an issue or dev-log. No fallback to the body's opening: on a menu that is
    preamble dressed as a reason, and once indented it cannot be told apart.

Left alone deliberately: the rule arms. A rule hint already renders the rule's
TRIGGER, which is written to answer exactly "does this apply to me" and beats
a matched chunk at it; and that line's budget was measured at #3851. Adding a
passage there would duplicate the trigger and spend the budget twice.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy

tests/test_matched_passage_surfaces.py

@@ -0,0 +1,166 @@
+"""Every semantic search hands on the passage that matched (#4243, #4250).
+
+Three searches collapse chunk rows to the best one per record, so each of them
+KNOWS which passage earned the hit. Each used to drop it, leaving every door to
+preview the head of the document instead — a span the search had already scored
+lower, with nothing saying so.
+
+These pin the mechanism (`report["best_chunk"]` from all three searches) and
+each surface that reads it, because the failure mode is silent: a door that
+quietly reverts to the body's opening still returns a plausible-looking string
+and no test that only checks "a preview exists" would notice.
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services import embeddings as emb
+from scribe.services.text import (
+    BODY_OPENING,
+    MATCHED_PASSAGE,
+    excerpt_fields,
+    matched_excerpt,
+)
+from tests.helpers import make_mock_session
+
+
+def _searching(rows):
+    """A patched session whose one query returns `rows` — the shared
+    make_mock_session (#2834) rather than a third local copy of the
+    __aenter__/__aexit__ dance."""
+    session = make_mock_session()
+    result = MagicMock()
+    result.all.return_value = rows
+    session.execute = AsyncMock(return_value=result)
+    return session
+
+
+# ---------------------------------------------------------------------------
+# The shared choice of span
+# ---------------------------------------------------------------------------
+
+def test_the_matched_passage_wins_over_the_opening():
+    body = "An opening about nothing much. " * 20 + " THE ANSWER."
+    text, kind, _cut = matched_excerpt(body, {"index": 4, "text": "THE ANSWER."}, 1000)
+    assert text == "THE ANSWER."
+    assert kind == MATCHED_PASSAGE
+
+
+def test_without_a_chunk_the_opening_is_named_as_the_opening():
+    """The fallback is legitimate — a plain listing matched nothing — but it
+    must not pass for the passage that matched."""
+    text, kind, _cut = matched_excerpt("just a body", None, 1000)
+    assert kind == BODY_OPENING
+
+
+def test_an_empty_chunk_is_not_mistaken_for_a_passage():
+    """A record embedded from its title alone stores an empty body chunk;
+    rendering that as "the passage that matched" would be a blank line
+    presented as evidence."""
+    _t, kind, _c = matched_excerpt("real body", {"index": 0, "text": "   "}, 1000)
+    assert kind == BODY_OPENING
+
+
+def test_the_field_names_travel_together():
+    """A door renames the text field to keep its consumers working; the label
+    has to follow it, or a row carries an excerpt under one name and its
+    meaning under another."""
+    out = excerpt_fields("b" * 500, {"index": 1, "text": "hit"}, 100, key="snippet")
+    assert out["snippet"] == "hit"
+    assert out["snippet_is"] == MATCHED_PASSAGE
+    assert out["body_length"] == 500
+    assert "read_full" in out
+
+
+def test_a_record_shown_whole_advertises_nothing_further():
+    out = excerpt_fields("short", None, 1000)
+    assert out["excerpt"] == "short"
+    assert "read_full" not in out
+
+
+# ---------------------------------------------------------------------------
+# All three searches publish the winning chunk
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_rule_search_reports_the_clause_that_matched():
+    """A rule's `why` and `how_to_apply` run long. A caller shown only the head
+    cannot see the clause the query actually hit."""
+    r1, r2 = MagicMock(id=1), MagicMock(id=2)
+    session = _searching([
+        (r1, 0.10, 2, "the clause that matched"),
+        (r2, 0.20, 0, "r2 best"),
+        (r1, 0.40, 9, "a worse clause of r1"),
+    ])
+    report: dict = {}
+    with (
+        patch.object(emb, "async_session", return_value=session),
+        patch.object(emb, "get_embedding", AsyncMock(return_value=[0.0] * 384)),
+        patch.object(emb, "can_read_project", AsyncMock(return_value=True)),
+    ):
+        out = await emb.semantic_search_rules(1, "q", limit=5, threshold=0.0,
+                                              report=report)
+    assert [r.id for _s, r in out] == [1, 2]
+    assert report["best_chunk"][1] == {"index": 2, "text": "the clause that matched"}
+
+
+@pytest.mark.asyncio
+async def test_milestone_search_reports_the_passage_of_the_plan_that_matched():
+    """A milestone's body IS the plan and search shows its short description,
+    which need not mention the part the query was about."""
+    m1 = MagicMock(id=7)
+    session = _searching([(m1, 0.15, 5, "step 6 — the acceptance case")])
+    report: dict = {}
+    with (
+        patch.object(emb, "async_session", return_value=session),
+        patch.object(emb, "get_embedding", AsyncMock(return_value=[0.0] * 384)),
+        patch.object(emb, "can_read_project", AsyncMock(return_value=True)),
+    ):
+        out = await emb.semantic_search_milestones(
+            1, "acceptance", limit=5, threshold=0.0, report=report,
+        )
+    assert [m.id for _s, m in out] == [7]
+    assert report["best_chunk"][7]["text"] == "step 6 — the acceptance case"
+
+
+@pytest.mark.asyncio
+async def test_a_caller_that_passes_no_report_still_works():
+    """Every one of these searches fails open by design — a recall aid must
+    never break the call it serves — and that includes the chunk channel."""
+    session = _searching([(MagicMock(id=1), 0.1, 0, "text")])
+    with (
+        patch.object(emb, "async_session", return_value=session),
+        patch.object(emb, "get_embedding", AsyncMock(return_value=[0.0] * 384)),
+        patch.object(emb, "can_read_project", AsyncMock(return_value=True)),
+    ):
+        out = await emb.semantic_search_milestones(1, "q", limit=5, threshold=0.0)
+    assert len(out) == 1
+
+
+def test_record_best_chunk_on_no_report_is_a_no_op():
+    emb.record_best_chunk(None, {1: {"index": 0, "text": "x"}})  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# Only what survived the bar is published
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_chunks_are_published_only_for_results_that_came_back():
+    """Keyed off the returned rows, so a caller can look up every result it has
+    and never holds passages for records it was not shown."""
+    keep, drop = MagicMock(id=1), MagicMock(id=2)
+    session = _searching([
+        (keep, 0.10, 0, "kept"),
+        (drop, 0.95, 0, "below the bar"),
+    ])
+    report: dict = {}
+    with (
+        patch.object(emb, "async_session", return_value=session),
+        patch.object(emb, "get_embedding", AsyncMock(return_value=[0.0] * 384)),
+        patch.object(emb, "can_read_project", AsyncMock(return_value=True)),
+    ):
+        await emb.semantic_search_milestones(
+            1, "q", limit=5, threshold=0.5, report=report,
+        )
+    assert set(report["best_chunk"]) == {1}

tests/test_services_plugin_context.py

@@ -78,8 +78,12 @@ async def test_build_autoinject_hint_titles_only_with_margin_gate():
     assert out["note_ids"] == [11, 22]
     assert '#11 [note] "Pool sizing decision" (0.80)' in out["context"]
     assert "#33" not in out["context"]
-    # Title-first: no body text, ever.
+    # Title-first when the search reports no matched passage — which is this
+    # test, whose mock returns bare (score, note) pairs and fills no report.
+    # A record whose passage IS known gets it on a second line; that is
+    # test_the_menu_shows_the_passage_that_matched below.
     assert "get_note(id)" in out["context"]
+    assert "↳" not in out["context"]
     # Telemetry fired for BOTH retrievals this path runs: the scored menu and
     # the reuse-slot query competing against it. The slot's query used to be
     # the one unlogged retrieval on this path — the hit it displaced was in
@@ -552,3 +556,83 @@ async def test_the_config_stand_in_carries_every_key_the_real_one_does():
         "a key the real config has and the stand-in does not turns an arm "
         "into a silent no-op under test"
     )
+
+
+# ─── the passage that matched travels onto the menu (#4243, #4250) ───────────
+
+
+@pytest.mark.asyncio
+async def test_the_menu_shows_the_passage_that_matched():
+    """A title is a headline. For a lesson or a snippet it carries the trigger
+    and answers "does this apply to me"; for an issue or a dev-log the reason
+    this record matched is a sentence somewhere inside it, and the reader was
+    being handed the one part guaranteed not to say so."""
+    from scribe.services import plugin_context as pc
+
+    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1))]
+
+    # Only the FIRST call is the menu's own search; the reuse and lesson slots
+    # run their own queries afterwards and must not contribute chunks, which is
+    # also what makes the count assertion below deterministic. `report` is
+    # optional on this interface, so it is written only when one was passed.
+    calls: list[int] = []
+
+    async def _menu_search(*_a, **kw):
+        calls.append(1)
+        if len(calls) > 1:
+            return []
+        if kw.get("report") is not None:
+            kw["report"]["best_chunk"] = {
+                11: {"index": 3, "text": "we set max_overflow to 5 after the leak"}
+            }
+        return hits
+
+    with patch.object(pc, "get_autoinject_config",
+                      AsyncMock(return_value={"enabled": True, "threshold": 0.55,
+                                              "top_k": 3})), \
+         patch.object(pc, "semantic_search_notes", _menu_search), \
+         patch.object(pc, "record_retrieval", MagicMock()):
+        out = await pc.build_autoinject_hint(1, "pool", project_id=2)
+
+    assert "we set max_overflow to 5 after the leak" in out["context"]
+    # Indented under its line, so the menu still reads as a list of records
+    # rather than a wall of prose.
+    assert ">   ↳ we set max_overflow" in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_a_record_with_no_stored_chunk_gets_no_invented_passage():
+    """The reserved lesson and reuse slots are fetched by their own queries, so
+    they are absent from this search's report. Falling back to the body's
+    opening would put a line of preamble under them dressed as the reason they
+    matched — and once indented identically, a reader cannot tell the two
+    apart."""
+    from scribe.services import plugin_context as pc
+
+    hits = [(0.80, fake_note(id=11, title="Has a chunk", user_id=1)),
+            (0.78, fake_note(id=22, title="Has none", user_id=1,
+                             body="A long body whose opening says nothing."))]
+
+    calls: list[int] = []
+
+    async def _menu_search(*_a, **kw):
+        calls.append(1)
+        if len(calls) > 1:
+            return []
+        if kw.get("report") is not None:
+            kw["report"]["best_chunk"] = {
+                11: {"index": 0, "text": "the real reason"}
+            }
+        return hits
+
+    with patch.object(pc, "get_autoinject_config",
+                      AsyncMock(return_value={"enabled": True, "threshold": 0.55,
+                                              "top_k": 3})), \
+         patch.object(pc, "semantic_search_notes", _menu_search), \
+         patch.object(pc, "record_retrieval", MagicMock()):
+        out = await pc.build_autoinject_hint(1, "q", project_id=2)
+
+    assert "the real reason" in out["context"]
+    assert "A long body whose opening" not in out["context"]
+    # Exactly one passage line, for the one record that had a passage.
+    assert out["context"].count("↳") == 1