feat(telemetry): pull-through per surface, not just per corpus (#3311)

The readout already grouped usage by source — `group_by(event, source)` —
and the loop directly below it threw the source away, collapsing every
surface into one corpus-wide ratio. So the question a threshold is
actually tuned against, "is THIS surface worth its noise", could not be
asked of any surface, while the data to answer it sat in the table.

`usage.by_source` reports notes_surfaced / notes_pulled / pull_through
per surface. The grain is the note, not the call: a pull records the
door it came through, not the surface that led there, so grouping the
pulled rows by source would answer a different question. Joining
surfaced rows to pulled rows on note_id answers this one without the
session identity #2085 declined to invent — at the cost of being an
upper bound per surface, which the docstring says where it is read.

Ambient surfaces report counts and a null ratio: nothing chose those
records, so "surfaced often, opened never" is not a judgment about them.
A surface that genuinely produced nothing reports 0.0, which must not
look like the null.

The join is guarded separately from the two reads above it. #2663 was a
novel SQL shape the database rejected inside a broad except; this is the
novel shape here, and it must not take down two readouts that work.

Tests are integration for that same reason — a mock passes on a query
Postgres refuses. They pin the distinct-first property (three surfacings
of one note are one note), the ambient null, and the LIKE escape, since
an unescaped `mcp_%` also matches `mcpXget_note` and nothing else in the
payload would show the difference.

tests/test_services_retrieval_telemetry.py

@@ -195,6 +195,10 @@ async def test_retrieval_summary_is_empty_not_broken_for_a_fresh_install(_dispos
     assert out["sources"] == {}
     assert out["usage"]["pull_through"] is None   # no division by zero
     assert out["usage"]["surfaced"] == 0
+    # An empty dict, not a missing key and not a failure flag — the same
+    # "no rows" / "read broke" distinction the rest of this readout keeps.
+    assert out["usage"]["by_source"] == {}
+    assert "by_source_failed" not in out["usage"]
 
 
 @pytest.mark.integration
@@ -222,3 +226,136 @@ async def test_retrieval_summary_sees_only_its_own_users_telemetry(_dispose_engi
         async with async_session() as s:
             await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == 990004))
             await s.commit()
+
+
+# ─── per-source pull-through (#3311) ─────────────────────────────────────────
+# Integration for the same reason the block above is: this is a self-join with
+# two DISTINCT subqueries and a LIKE escape, which is a new SQL shape in a
+# module whose one production outage (#2663) was a new SQL shape the database
+# rejected inside a broad except. A mock would pass on a query Postgres refuses.
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_by_source_separates_a_surface_that_earns_its_noise_from_one_that_does_not(
+    _dispose_engine,
+):
+    """The whole point: the corpus average cannot say WHICH surface is working.
+
+    Two ranked surfaces, identical volume, opposite outcomes — and a top-level
+    ratio that describes neither of them.
+    """
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.note_usage import NoteUsageEvent
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    UID = 990010
+    async with async_session() as s:
+        s.add_all([
+            # auto_inject chose note 1 three times and note 2 once. Three
+            # surfacings of one note is ONE note surfaced — the DISTINCT that
+            # keeps the join from multiplying rows is what this pins.
+            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
+            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
+            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
+            NoteUsageEvent(user_id=UID, note_id=2, event="surfaced", source="auto_inject"),
+            # write_path_semantic chose two notes and got nothing opened.
+            NoteUsageEvent(user_id=UID, note_id=3, event="surfaced", source="write_path_semantic"),
+            NoteUsageEvent(user_id=UID, note_id=4, event="surfaced", source="write_path_semantic"),
+            # One agent pull, of a note only auto_inject surfaced.
+            NoteUsageEvent(user_id=UID, note_id=1, event="pulled", source="mcp_get_note"),
+        ])
+        await s.commit()
+
+    try:
+        out = await retrieval_summary(UID, days=30)
+        assert out["read_failed"] is False
+        by_source = out["usage"]["by_source"]
+        assert "by_source_failed" not in out["usage"], "the join did not execute"
+
+        ai = by_source["auto_inject"]
+        assert ai["notes_surfaced"] == 2, "three surfacings of note 1 are one note"
+        assert ai["notes_pulled"] == 1
+        assert ai["pull_through"] == pytest.approx(0.5)
+
+        wp = by_source["write_path_semantic"]
+        assert wp["notes_surfaced"] == 2
+        assert wp["notes_pulled"] == 0
+        # 0.0, NOT None. "This surface produced nothing" is a finding; None is
+        # what a surface with no data reads as, and they must not look alike.
+        assert wp["pull_through"] == 0.0
+
+        # And the number that exists today, which is true of neither surface:
+        # one agent pull over six ranked surfacings.
+        assert out["usage"]["pull_through"] == pytest.approx(1 / 6, abs=1e-4)
+    finally:
+        async with async_session() as s:
+            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
+            await s.commit()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_an_ambient_surface_reports_its_counts_but_no_ratio(_dispose_engine):
+    """`enter_project` bulk-loads records; nothing CHOSE them. "Surfaced often,
+    opened never" is not a judgment about a record that was never picked, so the
+    counts stay visible and the ratio that would be misread is null."""
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.note_usage import NoteUsageEvent
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    UID = 990011
+    async with async_session() as s:
+        s.add_all([
+            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="enter_project"),
+            NoteUsageEvent(user_id=UID, note_id=2, event="surfaced", source="enter_project"),
+        ])
+        await s.commit()
+
+    try:
+        row = (await retrieval_summary(UID, days=30))["usage"]["by_source"]["enter_project"]
+        assert row["ambient"] is True
+        assert row["notes_surfaced"] == 2
+        assert row["pull_through"] is None
+    finally:
+        async with async_session() as s:
+            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
+            await s.commit()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_the_agent_pull_filter_does_not_treat_its_underscore_as_a_wildcard(
+    _dispose_engine,
+):
+    """`_` is a LIKE wildcard, so an unescaped `LIKE 'mcp_%'` also matches
+    `mcpXsomething`. The Python half of this readout uses str.startswith and
+    cannot have the bug; the SQL half needs autoescape to match it, and nothing
+    else in the payload would reveal the difference."""
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.note_usage import NoteUsageEvent
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    UID = 990012
+    async with async_session() as s:
+        s.add_all([
+            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
+            # Not an agent pull: the door is `mcpXget_note`, not `mcp_get_note`.
+            NoteUsageEvent(user_id=UID, note_id=1, event="pulled", source="mcpXget_note"),
+        ])
+        await s.commit()
+
+    try:
+        row = (await retrieval_summary(UID, days=30))["usage"]["by_source"]["auto_inject"]
+        assert row["notes_pulled"] == 0, "a wildcard match counted a non-agent pull"
+        assert row["pull_through"] == 0.0
+    finally:
+        async with async_session() as s:
+            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
+            await s.commit()