feat(rulings): system_usage_events is read back — per-System counts and a telemetry block (#4769)

#4769 "Rulings are counted where someone will read them": milestone 444
step 4 wrote system_usage_events and nothing read it.

- retrieval_telemetry gains a `system_usage` block: surfacings and opens by
  source, distinct counts, and `by_system` naming the areas most shown.
  There is deliberately no pull-through ratio, because rulings travel in full
  in the line and opens are the exception.
- usage_for_systems (one GROUP BY) adds `usage` to the REST Systems list and
  detail, and to MCP get_system. MCP list_systems is unchanged.
- The Systems UI shows a "rulings shown N×" chip.
- rulings_pre_tool, rulings_write_path and mcp_get_system are now declared
  registry points; the registry guard covers their recorders.
- The Systems store merges a PATCH reply instead of replacing the row.

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

tests/test_system_usage_readout.py

@@ -0,0 +1,161 @@
+"""Rulings are counted where someone will read them (#4769).
+
+`system_usage_events` was written from milestone 444 step 4 and read by
+nothing. These pin the two readouts:
+
+- PER SYSTEM: `usage_for_systems` — one GROUP BY, a zero shape for a System
+  nothing touched — carried by the Systems list, the REST detail and MCP
+  `get_system`.
+- IN AGGREGATE: `retrieval_summary`'s `system_usage` block — counts by source,
+  the areas named, and deliberately NO pull-through ratio.
+
+Integration for both aggregates, not mocked: a GROUP BY the database rejects
+is swallowed by the fail-open guard and reads as zero, which is #2663 exactly.
+"""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+
+from scribe.models import async_session
+from scribe.services.system_usage import empty_system_usage
+# Bound at import, before conftest's autouse stub replaces the module attribute.
+from scribe.services.system_usage import usage_for_systems as real_usage_for_systems
+from tests.helpers import ensure_user
+
+
+# ── the payloads carry it ────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_get_system_carries_its_usage():
+    from tests.helpers import fake_system
+
+    shown = {**empty_system_usage(), "surfaced_count": 4, "pull_count": 1}
+    with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
+         patch("scribe.mcp.tools.systems.systems_svc") as svc, \
+         patch("scribe.services.system_usage.usage_for_systems",
+               AsyncMock(return_value={3: shown})):
+        svc.get_system = AsyncMock(return_value=fake_system(id=3))
+        svc.list_records_for_system = AsyncMock(return_value=[])
+        from scribe.mcp.tools.systems import get_system
+        result = await get_system(system_id=3)
+    assert result["usage"] == shown
+
+
+def test_the_zero_shape_matches_the_client_type():
+    """The four keys `RecordUsage` in frontend/src/types/usage.ts declares —
+    a fifth here would be a field the chip never reads, a missing one a
+    crash on every System nothing has touched."""
+    assert set(empty_system_usage()) == {
+        "surfaced_count", "pull_count", "last_surfaced_at", "last_pulled_at",
+    }
+
+
+@pytest.mark.asyncio
+async def test_an_empty_id_list_reads_nothing():
+    with patch("scribe.services.system_usage.async_session") as session:
+        assert await real_usage_for_systems([]) == {}
+    session.assert_not_called()
+
+
+# ── the aggregates, against Postgres ─────────────────────────────────────
+
+
+@pytest_asyncio.fixture
+async def ruled_areas():
+    """An owner, a project and two Systems, with usage rows on both."""
+    from scribe.models.project import Project
+    from scribe.models.system import System
+    from scribe.models.system_usage import SystemUsageEvent
+
+    async with async_session() as s:
+        owner = await ensure_user(s, "system_usage_owner")
+        other = await ensure_user(s, "system_usage_other")
+        project = Project(user_id=owner.id, title="Ruled areas")
+        s.add(project)
+        await s.flush()
+        billing = System(user_id=owner.id, project_id=project.id, name="Billing")
+        storage = System(user_id=owner.id, project_id=project.id, name="Storage")
+        s.add_all([billing, storage])
+        await s.flush()
+
+        def ev(uid, sid, event, source):
+            return SystemUsageEvent(
+                user_id=uid, system_id=sid, event=event, source=source,
+                project_id=project.id,
+            )
+
+        s.add_all([
+            ev(owner.id, billing.id, "surfaced", "rulings_pre_tool"),
+            ev(owner.id, billing.id, "surfaced", "rulings_pre_tool"),
+            ev(owner.id, billing.id, "surfaced", "rulings_write_path"),
+            ev(owner.id, billing.id, "pulled", "mcp_get_system"),
+            ev(owner.id, storage.id, "surfaced", "rulings_write_path"),
+            # Another user's session: in the per-System count, which is about
+            # the System, and NOT in the owner's telemetry, which is theirs.
+            ev(other.id, storage.id, "surfaced", "rulings_pre_tool"),
+        ])
+        ids = {
+            "owner": owner.id, "other": other.id, "pid": project.id,
+            "billing": billing.id, "storage": storage.id,
+        }
+        await s.commit()
+    yield ids
+
+    from sqlalchemy import delete
+    async with async_session() as s:
+        await s.execute(delete(SystemUsageEvent).where(
+            SystemUsageEvent.system_id.in_([ids["billing"], ids["storage"]])))
+        await s.execute(delete(System).where(System.project_id == ids["pid"]))
+        await s.execute(delete(Project).where(Project.id == ids["pid"]))
+        await s.commit()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_dispose_engine")
+async def test_usage_for_systems_counts_each_system(ruled_areas):
+    ids = ruled_areas
+    out = await real_usage_for_systems([ids["billing"], ids["storage"], 987654321])
+    billing, storage = out[ids["billing"]], out[ids["storage"]]
+    assert billing["surfaced_count"] == 3 and billing["pull_count"] == 1
+    assert billing["last_surfaced_at"] and billing["last_pulled_at"]
+    assert storage["surfaced_count"] == 2 and storage["pull_count"] == 0
+    assert storage["last_pulled_at"] is None
+    assert out[987654321] == empty_system_usage()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_dispose_engine")
+async def test_the_telemetry_block_names_the_areas_and_offers_no_ratio(ruled_areas):
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    ids = ruled_areas
+    block = (await retrieval_summary(ids["owner"], days=30))["system_usage"]
+    assert "system_usage_failed" not in block, "the aggregate did not execute"
+    assert block["surfaced"] == 4 and block["pulled"] == 1
+    assert block["by_source"] == {
+        "rulings_pre_tool": 2, "rulings_write_path": 2, "mcp_get_system": 1,
+    }
+    assert block["distinct_systems_surfaced"] == 2
+    assert block["distinct_systems_pulled"] == 1
+    assert [(r["name"], r["surfaced"], r["pulled"]) for r in block["by_system"]] == [
+        ("Billing", 3, 1), ("Storage", 1, 0),
+    ]
+    # Rulings are delivered in full in the line; a ratio of opens would read
+    # near zero on an arm that is working.
+    assert "pull_through" not in block
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("_dispose_engine")
+async def test_a_fresh_install_reads_zero_not_failed():
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    block = (await retrieval_summary(990031, days=30))["system_usage"]
+    assert "system_usage_failed" not in block
+    assert block["surfaced"] == 0 and block["by_system"] == []
+    assert block["covers_window"] is None