FabledScribe/src/scribe/services/note_usage.py

"""Note usage telemetry — was a surfaced note ever actually pulled?

Two event streams, deliberately independent:

  - SURFACED: we put this note's title in front of the agent (auto-inject, or
    either arm of the write-path prior-art trigger).
  - PULLED: someone then opened it in full (get_snippet / get_note / the REST
    detail route).

The ratio between them is the signal. A snippet surfaced forty times and never
pulled is not neutral — it occupies the injection budget on every future turn
and dilutes the menu — so this is what makes dead weight visible and prunable.

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.
  - 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.
"""
from __future__ import annotations

import asyncio
import logging
import traceback

from sqlalchemy import case, func, select

from scribe.models import async_session
from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
from scribe.models.base import iso

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: failures degrade, visibly."""
    try:
        async with async_session() as session:
            session.add_all([NoteUsageEvent(**row) for row in rows])
            await session.commit()
    except Exception:
        await _report_failure("write")


def _schedule(rows: list[dict]) -> None:
    if not rows:
        return
    try:
        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(
    *, user_id: int | None, note_ids: list[int] | set[int], source: str
) -> None:
    """Fire-and-forget: record that these notes were shown to the agent.

    Takes the whole menu at once — one insert per surfacing event, not per note
    — because a menu is a single decision and its rows should land together.
    """
    try:
        rows = [
            {
                "user_id": user_id,
                "note_id": int(nid),
                "event": SURFACED,
                "source": source,
            }
            for nid in note_ids
        ]
    except Exception:
        logger.debug("note usage payload build failed", exc_info=True)
        return
    _schedule(rows)


def record_pulled(*, user_id: int | None, note_id: int, source: str) -> None:
    """Fire-and-forget: record that a note was opened in full."""
    try:
        rows = [
            {
                "user_id": user_id,
                "note_id": int(note_id),
                "event": PULLED,
                "source": source,
            }
        ]
    except Exception:
        logger.debug("note usage payload build failed", exc_info=True)
        return
    _schedule(rows)


# Surfacings that are NOT a ranked choice. enter_project returns whatever the
# top-N-by-recency happen to be; the skill sync installs every Process the
# operator can reach. Counting those alongside auto-inject would make a note's
# surfaced_count dominated by "it was recently updated in a project you
# opened", and dead-weight detection would read that as popularity (#2477).
# They still matter — a pull that follows one must not float unattributed — so
# they land in their own bucket rather than not landing at all.
AMBIENT_SOURCES = ("enter_project", "process_skill_sync")


def empty_usage() -> dict:
    """The zero readout — what a note with no recorded events looks like.

    Callers render this shape unconditionally, so a note predating the table
    reads as "never surfaced, never pulled" rather than as a missing key.

    `surfaced_count` is RANKED surfacings only — a scored surface chose this
    record. `ambient_count` is the rest (see AMBIENT_SOURCES). The split is the
    readout half of #2477: the "high surfaced, zero pulls → dead weight"
    reading is only valid over surfacings that were choices.
    """
    return {
        "surfaced_count": 0,
        "ambient_count": 0,
        "pull_count": 0,
        "last_surfaced_at": None,
        "last_pulled_at": None,
    }


async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
    """Aggregate usage for a set of notes: {note_id: {counts + timestamps}}.

    One GROUP BY for the whole page rather than a query per row — this feeds a
    list view, so the per-row shape would be N+1 by construction. Notes with no
    events are returned with `empty_usage()` so the caller never has to
    distinguish "no events" from "not in the result".
    """
    ids = [int(n) for n in note_ids]
    out: dict[int, dict] = {nid: empty_usage() for nid in ids}
    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 = (
                await session.execute(
                    select(
                        NoteUsageEvent.note_id,
                        NoteUsageEvent.event,
                        func.count().label("n"),
                        func.max(NoteUsageEvent.created_at).label("last_at"),
                        ambient,
                    )
                    .where(NoteUsageEvent.note_id.in_(ids))
                    .group_by(
                        NoteUsageEvent.note_id,
                        NoteUsageEvent.event,
                        ambient,
                    )
                )
            ).all()
    except Exception:
        # 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:
        slot = out.get(int(note_id))
        if slot is None:
            continue
        if event == SURFACED and ambient:
            slot["ambient_count"] = int(n)
        elif event == SURFACED:
            slot["surfaced_count"] = int(n)
            slot["last_surfaced_at"] = iso(last_at)
        elif event == PULLED:
            # Pulls are pulls regardless of what surfaced the record — the
            # question a pull answers ("did anyone ever open this?") doesn't
            # depend on how it was found.
            slot["pull_count"] = slot["pull_count"] + int(n)
            latest = iso(last_at)
            if latest and (slot["last_pulled_at"] or "") < latest:
                slot["last_pulled_at"] = latest
    return out