fix(telemetry): usage counters get canaries, task references, and the missing integration tests (#2663)

The deployed instance ran with every usage counter at zero while surfacing
demonstrably fired. Every unit test was green because every unit test mocked
either _schedule or the session — the two functions that touch the database
ran against real Postgres nowhere. Both telemetry writers also held no
reference to their fire-and-forget tasks (the loop keeps only weak ones), and
swallowed every failure into logger.debug, so a total outage was
indistinguishable from an unused corpus.

- note_usage + retrieval_telemetry keep strong task references until done
- failures log at WARNING; note_usage additionally drops one AppLog error row
  per process per site, so the admin UI shows the outage without host access
- integration tests cover _insert_events -> usage_for_notes and the full
  record_pulled chain on a running loop, splitting the write and read halves
  so a failure names its side

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

tests/test_note_usage.py

@@ -8,6 +8,7 @@ before this it surfaced snippets while leaving no trace anywhere.
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+import pytest_asyncio
 
 
 @pytest.fixture(autouse=True)
@@ -249,3 +250,78 @@ def test_every_getter_that_can_be_surfaced_also_records_a_pull():
             f"{name} can be surfaced in an auto-inject menu but records no pull — "
             "its pull-through rate will read as zero regardless of real usage"
         )
+
+
+# --- persistence (integration) --------------------------------------------
+# Everything above mocks _schedule or the session — deliberately, for the hot
+# path. But that left the two functions that actually touch the database
+# (_insert_events and usage_for_notes' real SQL) running against real Postgres
+# nowhere, which is how the deployed instance reported zero for every counter
+# while surfacing demonstrably fired (#2663): all-green mocked units over a
+# dead real path, the #2109 shape. These two run in the CI integration lane
+# and split the chain so a failure names its half.
+
+
+@pytest_asyncio.fixture
+async def _dispose_engine():
+    from scribe.models import engine
+    yield
+    await engine.dispose()
+
+
+async def _purge(note_id: int) -> None:
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.note_usage import NoteUsageEvent
+
+    async with async_session() as s:
+        await s.execute(
+            delete(NoteUsageEvent).where(NoteUsageEvent.note_id == note_id)
+        )
+        await s.commit()
+
+
+@pytest.mark.integration
+async def test_insert_and_readout_roundtrip_on_real_postgres(_dispose_engine):
+    """WRITE half + READ half against the real table, one assertion per counter."""
+    from scribe.services.note_usage import _insert_events
+
+    nid = 990101
+    try:
+        await _insert_events([
+            {"user_id": 7, "note_id": nid, "event": "surfaced",
+             "source": "write_path_place"},
+            {"user_id": 7, "note_id": nid, "event": "surfaced",
+             "source": "enter_project"},
+            {"user_id": 7, "note_id": nid, "event": "pulled",
+             "source": "mcp_get_snippet"},
+        ])
+        out = await usage_for_notes([nid])
+        # write_path_place is a ranked choice; enter_project is ambient (#2477).
+        assert out[nid]["surfaced_count"] == 1
+        assert out[nid]["ambient_count"] == 1
+        assert out[nid]["pull_count"] == 1
+        assert out[nid]["last_surfaced_at"] is not None
+        assert out[nid]["last_pulled_at"] is not None
+    finally:
+        await _purge(nid)
+
+
+@pytest.mark.integration
+async def test_record_pulled_lands_end_to_end_from_a_running_loop(_dispose_engine):
+    """The exact chain the deployed instance runs: record_pulled schedules a
+    fire-and-forget task on the running loop, and the row must land. The
+    _pending set (which exists to keep the loop's weak-ref'd tasks alive) is
+    also what lets this test await a write that is fire-and-forget by design."""
+    import asyncio
+
+    nid = 990102
+    try:
+        record_pulled(user_id=7, note_id=nid, source="mcp_get_snippet")
+        assert note_usage._pending, "record_pulled scheduled no task"
+        await asyncio.gather(*note_usage._pending)
+        out = await usage_for_notes([nid])
+        assert out[nid]["pull_count"] == 1
+    finally:
+        await _purge(nid)