Merge pull request 'Usage telemetry alive — readout fix, canaries, and the snippet-recording seam' (#110) from dev into main

.forgejo/workflows/ci.yml

@@ -225,6 +225,16 @@ jobs:
           UV_PROJECT_ENVIRONMENT: /opt/venv
         run: uv sync --locked --extra dev
 
+      # The hook-EXECUTION tests (test_write_path_trigger's nudge pair) run the
+      # real bash hook, which exits silently without jq — and those tests skip
+      # rather than fail when it's absent, so without this step they would
+      # quietly never be verified anywhere (ci-python ships without jq; same
+      # install the Plugin hooks job does).
+      - name: Install jq for hook execution tests
+        run: |
+          apt-get update -qq
+          apt-get install -y -qq --no-install-recommends jq
+
       - name: Run tests
         # Integration tests (real Postgres) run in the `integration` job below.
         run: /opt/venv/bin/python -m pytest tests/ -q -m "not integration"

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.28",
+  "version": "0.1.29",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/hooks/scribe_prior_art.sh

@@ -196,6 +196,21 @@ if [ -n "$body" ]; then
   fi
 fi
 
+# ARM 1½ — the RECORD nudge (#2664). The local arm just proved the thing being
+# written already exists elsewhere in this repo, and Scribe returned no record
+# of anything for it. That is the one moment "record it" is earned rather than
+# noise: the duplication is demonstrated, not guessed. Gated on BOTH sides so
+# an ordinary new helper (no other copies) and an already-recorded one (the
+# server spoke) stay nudge-free — a reflex that fires on everything is one
+# that gets skipped. An unreachable server counts as "nothing recorded": the
+# local finding needed no server, and the nudge fails open with it.
+if [ -n "$local_lines" ]; then
+  n_recorded=$(printf '%s' "$body" | jq -r '.note_ids | length' 2>/dev/null) || n_recorded=0
+  if [ "${n_recorded:-0}" = "0" ] || [ "$n_recorded" = "" ]; then
+    local_context="${local_context}"$'\n'"> None of those existing copies is recorded in Scribe. If the version being written is the canonical one — or this edit is consolidating the copies — record it now with create_snippet (name, code, when-to-reach-for-it, location) so the next session is offered it instead of writing another copy."
+  fi
+fi
+
 # Local first. It answers "this already EXISTS", which is a stronger claim than
 # "this resembles something recorded" — and it is the one the recorded arms are
 # structurally unable to make.

plugin/hooks/scribe_static_context.md

@@ -52,11 +52,16 @@ for the operator's work, and as your own working memory across sessions.
   it. An untagged project record carries the `systems_hint` question instead,
   on creates, updates, and work-logs alike — treat it as the tagging question
   asked at the moment of work, not as noise to skip past.
-- **Reuse before rebuilding** — before writing a new helper/utility/component,
-  search recorded **snippets** (reusable code recorded once for recall) and
-  reuse the prior art instead of re-solving it; when you build something
-  reusable, record it with `create_snippet` (name, code, when-to-reach-for-it,
-  location) so a later session is offered it, not left to write it again.
+- **Reuse before rebuilding — and record what you build** — before writing a
+  new helper/utility/component, search recorded **snippets** (reusable code
+  recorded once for recall) and reuse the prior art instead of re-solving it.
+  The recording half has NAMED TRIGGERS, not a vibe: the moment you extract a
+  shared component, hoist a helper into a common module, or notice you are
+  writing the second copy of anything, record it with `create_snippet` (name,
+  code, when-to-reach-for-it, location) in the same breath as the commit.
+  Work that "refactors X into a shared Y" is not finished until Y is recorded
+  — an unrecorded shared component is invisible to every later session, which
+  is how a codebase grows four `.btn-primary` definitions.
 - Do **not** keep the operator's rules, plans, or project notes in local
   memory / CLAUDE.md in parallel with Scribe — Scribe holds the single copy.
 - **Compact at clean seams** — because you record as you go, a context

src/scribe/services/note_usage.py

@@ -15,8 +15,13 @@ Design notes (mirrors retrieval_telemetry, for the same reasons):
   - Writes are fire-and-forget. `record_surfaced` / `record_pulled` extract
     plain ints synchronously and schedule the insert as a background task, so
     telemetry never adds latency to — or can break — the surface it observes.
-  - Every failure path is swallowed. Losing a usage row costs a data point;
-    raising would cost the operator their retrieval.
+  - Failures degrade, but they must not degrade SILENTLY. The original version
+    swallowed everything into logger.debug, and the deployed instance ran with
+    every counter at zero for weeks while surfacing demonstrably fired — a
+    total outage indistinguishable from "nobody uses this" (#2663). A
+    subsystem whose every failure mode is invisible cannot report its own
+    death, so failures now log at WARNING and drop one AppLog error row per
+    process per site, where the admin UI shows it.
   - Reads (`usage_for_notes`) are NOT fire-and-forget — a readout the caller
     awaits, aggregated in one round-trip for a whole page of snippets rather
     than per row.
@@ -25,6 +30,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import traceback
 
 from sqlalchemy import case, func, select
 
@@ -33,26 +39,67 @@ from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
 
 logger = logging.getLogger(__name__)
 
+# Strong references to in-flight inserts. The event loop keeps only a WEAK
+# reference to a task, so a fire-and-forget create_task with no other holder
+# can be garbage-collected before it completes — a write that never errors and
+# never lands. The done-callback discard keeps the set from growing.
+_pending: set[asyncio.Task] = set()
+
+# Sites that already dropped their once-per-process AppLog row. The readout
+# runs on every snippet list render — without this, a broken table would turn
+# the error log into a firehose that buries the finding it exists to surface.
+_reported: set[str] = set()
+
+
+async def _report_failure(site: str) -> None:
+    """Make a swallowed telemetry failure visible. Called from an except block.
+
+    WARNING to the process log every time; one AppLog error row per process per
+    site so the admin UI shows the outage without host access. The AppLog write
+    is itself guarded — when the whole database is down it fails too, and that
+    is fine: the WARNING already said so, and a canary must never take down the
+    surface it watches.
+    """
+    logger.warning("note usage telemetry %s failed", site, exc_info=True)
+    if site in _reported:
+        return
+    _reported.add(site)
+    try:
+        from scribe.services.logging import log_error
+
+        await log_error(
+            endpoint="note_usage",
+            error_type=f"note_usage_{site}_failed",
+            error_message=f"note usage telemetry {site} is failing; "
+            "usage counters will read zero until this is fixed",
+            traceback=traceback.format_exc(),
+        )
+    except Exception:
+        logger.debug("note usage canary write failed", exc_info=True)
+
 
 async def _insert_events(rows: list[dict]) -> None:
-    """Persist usage rows. Best-effort: all errors are swallowed."""
+    """Persist usage rows. Best-effort: failures degrade, visibly."""
     try:
         async with async_session() as session:
             session.add_all([NoteUsageEvent(**row) for row in rows])
             await session.commit()
     except Exception:
-        logger.debug("note usage telemetry write skipped", exc_info=True)
+        await _report_failure("write")
 
 
 def _schedule(rows: list[dict]) -> None:
     if not rows:
         return
     try:
-        asyncio.get_running_loop().create_task(_insert_events(rows))
+        task = asyncio.get_running_loop().create_task(_insert_events(rows))
     except RuntimeError:
         # No running loop (sync context outside the app) — skip rather than
         # block. Every app path runs on the loop.
         logger.debug("note usage telemetry skipped — no running event loop")
+        return
+    _pending.add(task)
+    task.add_done_callback(_pending.discard)
 
 
 def record_surfaced(
@@ -139,6 +186,18 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
     if not ids:
         return out
 
+    # Classified in SQL so the group count stays small: per note we get at most
+    # (surfaced-ranked, surfaced-ambient, pulled) rather than one row per
+    # distinct source. ONE labelled expression, grouped by its label — a second
+    # case() instance in GROUP BY renders with its own expanding-IN bind names
+    # under asyncpg, so the database sees two DIFFERENT expressions and rejects
+    # the query with a GroupingError. That rejection was swallowed, which is
+    # how every counter read zero in production while the writes were landing
+    # fine (#2663).
+    ambient = case(
+        (NoteUsageEvent.source.in_(AMBIENT_SOURCES), True),
+        else_=False,
+    ).label("ambient")
     try:
         async with async_session() as session:
             rows = (
@@ -148,28 +207,21 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
                         NoteUsageEvent.event,
                         func.count().label("n"),
                         func.max(NoteUsageEvent.created_at).label("last_at"),
-                        # Classified in SQL so the group count stays small: per
-                        # note we get at most (surfaced-ranked, surfaced-ambient,
-                        # pulled) rather than one row per distinct source.
-                        case(
-                            (NoteUsageEvent.source.in_(AMBIENT_SOURCES), True),
-                            else_=False,
-                        ).label("ambient"),
+                        ambient,
                     )
                     .where(NoteUsageEvent.note_id.in_(ids))
                     .group_by(
                         NoteUsageEvent.note_id,
                         NoteUsageEvent.event,
-                        case(
-                            (NoteUsageEvent.source.in_(AMBIENT_SOURCES), True),
-                            else_=False,
-                        ),
+                        ambient,
                     )
                 )
             ).all()
     except Exception:
-        # A telemetry readout must not be able to break the list it decorates.
-        logger.debug("note usage readout failed", exc_info=True)
+        # A telemetry readout must not be able to break the list it decorates —
+        # but it must say it failed, or a broken readout is indistinguishable
+        # from a corpus nobody uses (#2663).
+        await _report_failure("readout")
         return out
 
     for note_id, event, n, last_at, ambient in rows:

src/scribe/services/retrieval_telemetry.py

@@ -24,6 +24,14 @@ from scribe.models.retrieval_log import RetrievalLog
 
 logger = logging.getLogger(__name__)
 
+# Strong references to in-flight inserts — the loop holds tasks only weakly,
+# and an unreferenced fire-and-forget task can be collected before it runs
+# (same guard as note_usage, found via #2663).
+_pending: set[asyncio.Task] = set()
+
+# Whether this process already dropped its one warning about failing writes.
+_reported = False
+
 
 def _build_payload(
     *,
@@ -65,13 +73,24 @@ def _build_payload(
 
 
 async def _insert_retrieval_log(payload: dict) -> None:
-    """Persist one RetrievalLog row. Best-effort: all errors are swallowed."""
+    """Persist one RetrievalLog row. Best-effort: failures degrade, visibly.
+
+    WARNING rather than debug — this table is the empirical basis for threshold
+    tuning, and a silent write outage yields a dataset that looks complete while
+    covering only part of the traffic (#2663's shape). Once per process is
+    enough to be found; per-call would flood the log with what it already said.
+    """
+    global _reported
     try:
         async with async_session() as session:
             session.add(RetrievalLog(**payload))
             await session.commit()
     except Exception:
-        logger.debug("retrieval telemetry write skipped", exc_info=True)
+        if not _reported:
+            _reported = True
+            logger.warning("retrieval telemetry write failed", exc_info=True)
+        else:
+            logger.debug("retrieval telemetry write skipped", exc_info=True)
 
 
 def record_retrieval(
@@ -108,8 +127,11 @@ def record_retrieval(
         return
 
     try:
-        asyncio.get_running_loop().create_task(_insert_retrieval_log(payload))
+        task = asyncio.get_running_loop().create_task(_insert_retrieval_log(payload))
     except RuntimeError:
         # No running loop (e.g. called from sync context outside the app) —
         # skip rather than block. The app paths always run on the loop.
         logger.debug("retrieval telemetry skipped — no running event loop")
+        return
+    _pending.add(task)
+    task.add_done_callback(_pending.discard)

tests/test_instruction_surfaces_agree.py

@@ -128,6 +128,25 @@ def test_floor_states_the_systems_reflex():
         )
 
 
+def test_floor_names_the_snippet_recording_triggers():
+    """The recording half of reuse needs NAMED trigger moments on the floor.
+
+    #2664's behavioral finding: with recording guidance as a trailing clause of
+    the reuse bullet, zero snippets were ever recorded outside sessions already
+    thinking about snippets — extracting a shared component (Roundtable's
+    BaseModal) produced task prose and no record. The floor must name the
+    moments, not just the tool.
+    """
+    floor = (ROOT / "plugin" / "hooks" / "scribe_static_context.md").read_text()
+    for needle in ("create_snippet", "second copy"):
+        assert needle in floor, (
+            f"plugin/hooks/scribe_static_context.md no longer states the "
+            f"snippet-recording trigger ({needle!r}) — the record-as-you-build "
+            f"reflex must be stated on the floor with its trigger moments "
+            f"(#2664)."
+        )
+
+
 # Topics displaced from _INSTRUCTIONS when it was cut to fit the fold. Each
 # must remain stated on at least one DELIVERED surface: a tool docstring
 # (arrives with the tool schema), the plugin static context (always arrives),

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)

tests/test_write_path_trigger.py

@@ -789,3 +789,75 @@ def test_plugin_version_bumped_with_the_hook():
     manifest = json.loads((PLUGIN / ".claude-plugin" / "plugin.json").read_text())
     version = tuple(int(p) for p in manifest["version"].split("."))
     assert version >= (0, 1, 18)
+
+
+def _hook_runtime_env():
+    """Env for tests that need the hook's tools to actually RUN.
+
+    The silence-contract tests above deliberately restrict PATH — the hook must
+    exit quietly when its tools are missing, and asserting on empty output
+    doesn't care why it was empty. These tests assert on CONTENT, so the tools
+    must resolve wherever the image installed them; skip (don't fail) on an
+    image that lacks them, because that image cannot exercise this behaviour
+    at all.
+    """
+    import os
+    import shutil
+
+    for tool in ("git", "jq", "curl", "bash"):
+        if shutil.which(tool) is None:
+            pytest.skip(f"hook runtime tool {tool!r} not installed")
+    return {"PATH": os.environ["PATH"],
+            "SCRIBE_URL": "http://127.0.0.1:9", "SCRIBE_TOKEN": "t"}
+
+
+def test_hook_nudges_recording_when_copies_exist_but_nothing_is_recorded(tmp_path):
+    """#2664: the local arm proves duplication; when Scribe has no record of it,
+    the same context block must ask for create_snippet — the one moment the
+    recording nudge is earned rather than noise. An unreachable server counts
+    as "nothing recorded": the local finding needed no server, and the nudge
+    fails open with it (here: a refused connection stands in for the instance)."""
+    env = _hook_runtime_env()
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
+    (repo / "a.py").write_text("def debounce(fn):\n    return fn\n")
+    # git grep searches the index, so the existing copy must be staged.
+    subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
+    out = subprocess.run(
+        ["bash", str(HOOK)],
+        input=json.dumps({
+            "session_id": "s-nudge", "cwd": str(repo), "tool_name": "Write",
+            "tool_input": {"file_path": str(repo / "b.py"),
+                           "content": "def debounce(fn):\n    return fn\n"},
+        }),
+        capture_output=True, text=True, env=env,
+    )
+    assert out.returncode == 0
+    assert out.stdout.strip(), (
+        "hook produced no output — the local arm should have found the "
+        "staged duplicate and nudged"
+    )
+    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
+    assert "already defined" in ctx          # the duplication finding
+    assert "create_snippet" in ctx           # the recording ask riding it
+
+
+def test_hook_stays_quiet_about_recording_when_nothing_is_duplicated(tmp_path):
+    """A brand-new helper with no other copies earns no nudge — a reflex that
+    fires on every Write is one sessions learn to skip."""
+    env = _hook_runtime_env()
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
+    out = subprocess.run(
+        ["bash", str(HOOK)],
+        input=json.dumps({
+            "session_id": "s-quiet", "cwd": str(repo), "tool_name": "Write",
+            "tool_input": {"file_path": str(repo / "b.py"),
+                           "content": "def debounce(fn):\n    return fn\n"},
+        }),
+        capture_output=True, text=True, env=env,
+    )
+    assert out.returncode == 0
+    assert "create_snippet" not in out.stdout