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

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
from scribe.services.background import report_telemetry_failure

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()


async def _report_failure(site: str) -> None:
    """This subsystem's canary, now the shared one.

    The per-site dedup, the WARNING and the single AppLog row all moved to
    `background.report_telemetry_failure` unchanged when `rule_usage` needed
    the identical behaviour — two hand-kept copies of a thing whose whole job
    is to be reliable is the wrong number. `retrieval_telemetry` deliberately
    still has its own: its canary is a different shape (one process-wide flag,
    no AppLog row), so repointing it would change behaviour rather than
    consolidate it.
    """
    await report_telemetry_failure("note_usage", site)


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