bvandeusenandClaude Opus 5
7a2aff7bc1
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / TypeScript typecheck (push) Successful in 23s
CI & Build / integration (push) Successful in 34s
CI & Build / Python tests (push) Successful in 1m6s
CI & Build / Build & push image (push) Successful in 24s
`out["sources"]` was built only from the windowed aggregate, so a source with rows in `retrieval_logs` but none inside the window got no bucket at all. Absent is exactly how a source that never existed renders, so a surface that WAS recording and went silent became unreadable — #2663 one level up, the failure that looks like the correct answer. Two queries at different scopes, and only one shaped the output. `_complete_from` reads all-time and knows every source the table has ever held; the windowed loop dropped whatever it did not return. Every such source now gets a zero bucket. Zero is a real measurement here rather than a manufactured one: the all-time query proves the source was recording, and it made no calls across a window it fully covers. No `covers_window` special case is needed either — a source whose first row fell after `since` would have that row IN the window and already hold a bucket, so anything reaching this branch began before it. The counts are 0 and everything else is null. A sampled distribution is not the same claim as a call count, and rendering p50 as 0.0 for a source nobody sampled would assert a measurement — #3311's mistake, in the readout built to prevent it. Found while fixing #3712's fixture, which failed with KeyError for this exact reason. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011cPyzNnegXHr5iRMzzy5KJ
756 lines
36 KiB
Python
756 lines
36 KiB
Python
"""Retrieval telemetry — one RetrievalLog row per semantic-retrieval call.
|
|
|
|
This is the empirical basis for KB-injection tuning: it records what each query
|
|
asked for, the score distribution of what came back, and the effective params,
|
|
so the similarity threshold and top-k can be tuned from data rather than guessed.
|
|
|
|
Design notes:
|
|
- Fire-and-forget, mirroring upsert_note_embedding: `record_retrieval` extracts
|
|
the primitives it needs SYNCHRONOUSLY (while the caller's Note objects are
|
|
still valid) and schedules the DB insert as a background task, so logging
|
|
never adds latency to — or can break — the search response.
|
|
- Result objects are reduced to {id, score, rank} before scheduling; the
|
|
background writer touches only plain data, never a possibly-detached ORM row.
|
|
- Every failure path is swallowed: telemetry must never take down retrieval.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import case, func, select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.base import iso
|
|
from scribe.models.note import Note
|
|
from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
|
|
from scribe.models.rule_usage import PULLED as RULE_PULLED
|
|
from scribe.models.rule_usage import SURFACED as RULE_SURFACED
|
|
from scribe.models.rule_usage import RuleUsageEvent
|
|
from scribe.services.rule_usage import is_ambient
|
|
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(
|
|
*,
|
|
user_id: int | None,
|
|
source: str,
|
|
query: str | None,
|
|
threshold: float | None,
|
|
limit: int | None,
|
|
project_id: int | None,
|
|
is_task: bool | None,
|
|
results: list[tuple[float, Note]],
|
|
duration_ms: float | None,
|
|
suppressed: int | None = None,
|
|
) -> dict:
|
|
"""Reduce a retrieval call to a flat, JSON-safe RetrievalLog payload.
|
|
|
|
Pure and synchronous (no DB, no event loop) so it is unit-testable and safe
|
|
to run inline before scheduling the write. `results` is the
|
|
`(score, Note)` list from semantic_search_notes, already highest-first.
|
|
|
|
`suppressed` is how many scored hits the caller dropped because the session
|
|
had already been shown them, and it stays None for callers that cannot
|
|
know. See the column's comment: None means "not measured here", which is a
|
|
different fact from 0 and must never render as one.
|
|
"""
|
|
items = [
|
|
{"id": int(note.id), "score": round(float(score), 5), "rank": rank}
|
|
for rank, (score, note) in enumerate(results)
|
|
]
|
|
scores = [it["score"] for it in items]
|
|
return {
|
|
"user_id": user_id,
|
|
"source": source,
|
|
"query": query,
|
|
"threshold": threshold,
|
|
"limit_n": limit,
|
|
"project_id": project_id,
|
|
"is_task": is_task,
|
|
"result_count": len(items),
|
|
"suppressed_count": (None if suppressed is None else int(suppressed)),
|
|
"top_score": (scores[0] if scores else None),
|
|
"min_score": (scores[-1] if scores else None),
|
|
"result_ids": items,
|
|
"duration_ms": (round(duration_ms, 2) if duration_ms is not None else None),
|
|
}
|
|
|
|
|
|
async def _insert_retrieval_log(payload: dict) -> None:
|
|
"""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:
|
|
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(
|
|
*,
|
|
user_id: int | None,
|
|
source: str,
|
|
query: str | None,
|
|
threshold: float | None,
|
|
limit: int | None,
|
|
project_id: int | None,
|
|
is_task: bool | None,
|
|
results: list[tuple[float, Any]],
|
|
duration_ms: float | None = None,
|
|
suppressed: int | None = None,
|
|
) -> None:
|
|
"""Fire-and-forget: record one retrieval call.
|
|
|
|
`results` needs only `.id` on each record, which is why it is not typed to
|
|
Note: rules are retrieved too (milestone 307) and land here rather than in
|
|
note_usage_events. That table's ids are REMAPPED on a backup restore, so a
|
|
rule id written into it would come back attached to whatever note happened
|
|
to take that number — silent corruption of the very evidence this exists to
|
|
provide. retrieval_logs is not restored at all, so it has no such hazard,
|
|
and `source` already distinguishes the surfaces.
|
|
|
|
Builds the payload inline (synchronously) then schedules the insert so the
|
|
caller returns immediately. Never raises — telemetry must not affect search.
|
|
"""
|
|
try:
|
|
payload = _build_payload(
|
|
user_id=user_id,
|
|
source=source,
|
|
query=query,
|
|
threshold=threshold,
|
|
limit=limit,
|
|
project_id=project_id,
|
|
is_task=is_task,
|
|
results=results,
|
|
duration_ms=duration_ms,
|
|
suppressed=suppressed,
|
|
)
|
|
except Exception:
|
|
logger.debug("retrieval telemetry payload build failed", exc_info=True)
|
|
return
|
|
|
|
try:
|
|
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)
|
|
|
|
|
|
# --- The read half (#2975) ---------------------------------------------------
|
|
# Until this existed, `retrieval_logs` was WRITE-ONLY: rows accrued and the only
|
|
# `select()` over them in the whole tree lived in a test. That made #1038's gate
|
|
# — "build the reranker once telemetry shows precision is the bottleneck" —
|
|
# unsatisfiable by construction, and it is why the one real tuning decision on
|
|
# record (the 0.68 write-path threshold, #2223) was reached by hand-probing the
|
|
# live instance with eight payloads instead of by reading what was collected.
|
|
|
|
def _bucket(rows: list) -> dict:
|
|
"""A score readout a human can act on, from one aggregate row."""
|
|
(calls, zero, cleared, p10, p50, p90, lo, hi, avg_n, dur,
|
|
measured, supp_calls, supp_zero) = rows
|
|
return {
|
|
"calls": int(calls or 0),
|
|
# A call that returned nothing is not a low-scoring call — it is a
|
|
# different failure (nothing indexed, filter too narrow), and averaging
|
|
# it into the score distribution would hide both.
|
|
"zero_result_calls": int(zero or 0),
|
|
# How often the best hit actually cleared the threshold in force for
|
|
# that call. THE precision-adjacent number: a surface that clears its
|
|
# bar on almost every call is either well-tuned or too loose, and the
|
|
# score spread below says which.
|
|
"cleared_threshold": int(cleared or 0),
|
|
# Of the zeros above, which were the RANKER declining and which were
|
|
# the reader having seen it already? `zero_result_calls` cannot say,
|
|
# and only the first kind is evidence about the threshold.
|
|
#
|
|
# None — not a zeroed dict — when no row in the window reported it. A
|
|
# surface that filters inside the search genuinely does not know, and
|
|
# rendering that as `{"calls": 0}` would state a measurement nobody
|
|
# made. That substitution is the whole of #3311.
|
|
"suppression": (
|
|
None if not int(measured or 0) else {
|
|
"measured_calls": int(measured or 0),
|
|
"calls_with_suppression": int(supp_calls or 0),
|
|
# Subtract from zero_result_calls for the true ranker declines.
|
|
"zero_because_already_shown": int(supp_zero or 0),
|
|
}
|
|
),
|
|
"top_score": {
|
|
"p10": _round(p10), "p50": _round(p50), "p90": _round(p90),
|
|
"min": _round(lo), "max": _round(hi),
|
|
},
|
|
"avg_result_count": _round(avg_n),
|
|
"p90_duration_ms": _round(dur, 1),
|
|
}
|
|
|
|
|
|
# The aggregate row Postgres would have returned for a source with no rows in
|
|
# the window: nothing counted, nothing scored. Positional, matching the SELECT
|
|
# `_bucket` unpacks — calls, zero, cleared, p10, p50, p90, min, max, avg_n,
|
|
# dur, measured, supp_calls, supp_zero. The three counts are 0 because zero
|
|
# calls is a real observation; everything else is None because a distribution
|
|
# nobody sampled has no value, and rendering it as 0.0 would state one.
|
|
_NO_ROWS_IN_WINDOW = [0, 0, 0, None, None, None, None, None, None, None, 0, 0, 0]
|
|
|
|
|
|
def _round(v, places: int = 4):
|
|
return None if v is None else round(float(v), places)
|
|
|
|
|
|
async def _complete_from(session, model, user_id) -> dict[str, Any]:
|
|
"""When each source in `model` started being recorded, and the instant the
|
|
WHOLE table is complete from. Returns {source: earliest_row, "*": latest}.
|
|
|
|
THE GRAIN IS THE SOURCE, and that is the whole point. `retrieval_logs` has
|
|
rows going back months, so a table-level "earliest row" says months and
|
|
tells a reader their window is fully covered — while a source added last
|
|
week has a week of rows and a counter that silently means something else.
|
|
Per-source is the only grain at which partial coverage is visible.
|
|
|
|
THE AGGREGATE USES THE LATEST, NOT THE EARLIEST. A number that sums several
|
|
sources is complete only once EVERY contributor was recording, so "*" is a
|
|
max over the sources, not a min. Taking the min here would reproduce the
|
|
exact reading this exists to prevent: the oldest source vouching for the
|
|
youngest.
|
|
|
|
All-time, deliberately unfiltered by the window — a query bounded by
|
|
`since` can only ever report something at or after `since`, which answers
|
|
nothing.
|
|
"""
|
|
rows = (
|
|
await session.execute(
|
|
select(model.source, func.min(model.created_at))
|
|
.where(model.user_id == user_id)
|
|
.group_by(model.source)
|
|
)
|
|
).all()
|
|
out: dict[str, Any] = {src: ts for src, ts in rows if ts is not None}
|
|
stamps = list(out.values())
|
|
out["*"] = max(stamps) if stamps else None
|
|
return out
|
|
|
|
|
|
def _coverage(complete_from, since) -> dict:
|
|
"""The two keys every counter block carries, from one timestamp.
|
|
|
|
`covers_window` is None — never False — when nothing was ever recorded.
|
|
"No rows at all" is not "partial coverage", it is no measurement, and the
|
|
null convention #3497 established for `suppression` holds here for the
|
|
same reason: absent must not read as a verdict.
|
|
"""
|
|
return {
|
|
# iso() already returns None for an unset value (#2845) — the guard
|
|
# belongs on covers_window, which is a verdict, not a serialisation.
|
|
"complete_from": iso(complete_from),
|
|
"covers_window": (
|
|
None if complete_from is None else complete_from <= since
|
|
),
|
|
}
|
|
|
|
|
|
async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
|
|
"""What the retrieval telemetry says, per surface, over a window.
|
|
|
|
Three aggregates side by side, each read from the table built for it — NOT
|
|
a join. `usage` is notes, `rule_usage` is rules, and they stay apart
|
|
because a few dozen eligible rules blended into thousands of notes is the
|
|
note ratio with noise on it (milestone 333). `NoteUsageEvent`'s own docstring is explicit that the two are
|
|
complements ("RetrievalLog tunes the threshold, this tunes the corpus") and
|
|
that RetrievalLog's JSONB `result_ids` "can't be indexed at" the per-note
|
|
grain. So the score distribution comes from `retrieval_logs` on its indexed
|
|
columns, and surfaced-vs-pulled comes from `note_usage_events` at the grain
|
|
it was built for. Reading each from its own table is both cheaper and more
|
|
honest than correlating them through JSONB.
|
|
|
|
`usage["by_source"]` is the one join, and it stays INSIDE
|
|
`note_usage_events` — surfaced rows against pulled rows on note_id. That
|
|
answers "of the notes this surface chose, how many were opened", which the
|
|
top-level ratio averages away. It does not cross into `retrieval_logs`, so
|
|
the sentence above still holds.
|
|
|
|
Scoped to one user's own telemetry. There is no sharing model for a
|
|
retrieval log — it records what THIS user's agent asked for, including the
|
|
query text — so an owner filter is the whole access rule here rather than a
|
|
shortcut around `services/access.py` (P#78 governs shared record kinds).
|
|
|
|
Never raises: a telemetry readout that can break its caller is worse than
|
|
no readout. It does distinguish "no rows" from "the read failed", because
|
|
#2663 is exactly the bug where those two looked identical for weeks.
|
|
"""
|
|
since = datetime.now(timezone.utc) - timedelta(days=max(1, int(days)))
|
|
out: dict = {
|
|
"window_days": int(days),
|
|
"since": iso(since),
|
|
"sources": {},
|
|
"usage": {},
|
|
"rule_usage": {},
|
|
"read_failed": False,
|
|
}
|
|
|
|
cleared = case(
|
|
(
|
|
(RetrievalLog.threshold.isnot(None))
|
|
& (RetrievalLog.top_score.isnot(None))
|
|
& (RetrievalLog.top_score >= RetrievalLog.threshold),
|
|
1,
|
|
),
|
|
else_=0,
|
|
)
|
|
zero = case((RetrievalLog.result_count == 0, 1), else_=0)
|
|
# Three sums rather than one, because "not measured" and "measured as zero"
|
|
# are different answers and a single counter cannot hold both.
|
|
measured = case((RetrievalLog.suppressed_count.isnot(None), 1), else_=0)
|
|
supp_calls = case((RetrievalLog.suppressed_count > 0, 1), else_=0)
|
|
supp_zero = case(
|
|
((RetrievalLog.result_count == 0) & (RetrievalLog.suppressed_count > 0), 1),
|
|
else_=0,
|
|
)
|
|
|
|
def pct(p: float):
|
|
return func.percentile_cont(p).within_group(RetrievalLog.top_score.asc())
|
|
|
|
# Assigned inside the try below; named here so the readout can tell
|
|
# "this query failed" from "this window has no rows" (#2663).
|
|
by_source_rows = None
|
|
rule_rows = None
|
|
distinct_rules_surfaced = distinct_rules_pulled = 0
|
|
# None means the coverage read did not happen — distinct from a table with
|
|
# no rows, which is {"*": None}. Same reason `read_failed` exists.
|
|
note_complete = rule_complete = None
|
|
|
|
try:
|
|
async with async_session() as session:
|
|
rows = (
|
|
await session.execute(
|
|
select(
|
|
RetrievalLog.source,
|
|
func.count().label("calls"),
|
|
func.sum(zero).label("zero"),
|
|
func.sum(cleared).label("cleared"),
|
|
pct(0.1), pct(0.5), pct(0.9),
|
|
func.min(RetrievalLog.top_score),
|
|
func.max(RetrievalLog.top_score),
|
|
func.avg(RetrievalLog.result_count),
|
|
func.percentile_cont(0.9).within_group(
|
|
RetrievalLog.duration_ms.asc()
|
|
),
|
|
func.sum(measured).label("measured"),
|
|
func.sum(supp_calls).label("supp_calls"),
|
|
func.sum(supp_zero).label("supp_zero"),
|
|
)
|
|
.where(
|
|
RetrievalLog.created_at >= since,
|
|
RetrievalLog.user_id == user_id,
|
|
)
|
|
.group_by(RetrievalLog.source)
|
|
)
|
|
).all()
|
|
log_complete = await _complete_from(session, RetrievalLog, user_id)
|
|
for row in rows:
|
|
source = row[0]
|
|
bucket = _bucket(list(row[1:]))
|
|
# Per SOURCE, not per table: retrieval_logs goes back months
|
|
# while any individual arm may be days old, and the table's
|
|
# age would vouch for an arm that has barely started.
|
|
bucket.update(_coverage(log_complete.get(source), since))
|
|
out["sources"][source] = bucket
|
|
|
|
# A source with rows in the table but NONE in this window would
|
|
# otherwise be absent from the readout — and absent is exactly how
|
|
# a source that never existed renders, so a surface that WAS
|
|
# recording and went silent is unreadable (#3720). That is #2663
|
|
# one level up: the failure that looks like the correct answer.
|
|
#
|
|
# Zero here is a real measurement, not a manufactured one. The
|
|
# all-time query proves the source was recording, and it made no
|
|
# calls across a window it fully covers — which is why no
|
|
# `covers_window` special case is needed: a source whose first row
|
|
# fell after `since` would have that row IN the window and already
|
|
# hold a bucket, so anything reaching here began before it.
|
|
for src, first_row in log_complete.items():
|
|
if src == "*" or first_row is None or src in out["sources"]:
|
|
continue
|
|
quiet = _bucket(list(_NO_ROWS_IN_WINDOW))
|
|
quiet.update(_coverage(first_row, since))
|
|
out["sources"][src] = quiet
|
|
|
|
# The corpus side, at its own grain. `ambient` mirrors
|
|
# note_usage.usage_for_notes: an ambient surfacing was not a scored
|
|
# CHOICE, so folding it into pull-through would understate it.
|
|
# Grouped by RAW source, then classified in Python. The
|
|
# alternative — CASE expressions in the GROUP BY — is the shape
|
|
# that produced #2663: a second case() renders its own expanding
|
|
# bind names, the database sees two different expressions and
|
|
# rejects the query, and the broad except swallows it. One CASE is
|
|
# provably fine (usage_for_notes does it); two is where it broke.
|
|
# `source` has a handful of distinct values, so grouping on it
|
|
# directly is cheap and cannot fail that way at all.
|
|
urows = (
|
|
await session.execute(
|
|
select(
|
|
NoteUsageEvent.event,
|
|
NoteUsageEvent.source,
|
|
func.count().label("n"),
|
|
)
|
|
.where(
|
|
NoteUsageEvent.created_at >= since,
|
|
NoteUsageEvent.user_id == user_id,
|
|
)
|
|
.group_by(NoteUsageEvent.event, NoteUsageEvent.source)
|
|
)
|
|
).all()
|
|
note_complete = await _complete_from(session, NoteUsageEvent, user_id)
|
|
|
|
# Distinct-note counts need their OWN queries, and this is not
|
|
# fussiness: count(distinct note_id) per (event, source) group
|
|
# cannot be summed across groups — a note surfaced by two sources
|
|
# is one distinct note and would be counted twice. A wrong number
|
|
# labelled "distinct" is worse than no number.
|
|
from scribe.services.note_usage import AMBIENT_SOURCES as _AMB
|
|
|
|
distinct_surfaced = (
|
|
await session.execute(
|
|
select(func.count(func.distinct(NoteUsageEvent.note_id))).where(
|
|
NoteUsageEvent.created_at >= since,
|
|
NoteUsageEvent.user_id == user_id,
|
|
NoteUsageEvent.event == SURFACED,
|
|
NoteUsageEvent.source.notin_(_AMB),
|
|
)
|
|
)
|
|
).scalar_one()
|
|
distinct_pulled = (
|
|
await session.execute(
|
|
select(func.count(func.distinct(NoteUsageEvent.note_id))).where(
|
|
NoteUsageEvent.created_at >= since,
|
|
NoteUsageEvent.user_id == user_id,
|
|
NoteUsageEvent.event == PULLED,
|
|
)
|
|
)
|
|
).scalar_one()
|
|
|
|
# Per-source pull-through, at the NOTE grain (#3311).
|
|
#
|
|
# The `urows` query above already groups by source and the loop
|
|
# below then throws the source away, so until now this readout
|
|
# could say what the corpus's overall pull-through was and nothing
|
|
# about WHICH surface earned it. The data was always here; only
|
|
# the aggregation discarded it.
|
|
#
|
|
# It cannot be had by grouping the PULLED rows by source: a pull
|
|
# records the door it came through (`mcp_get_note`), not the
|
|
# surface that put the record in front of the agent. Correlating
|
|
# those within a session is what #2085 ruled out — there is no
|
|
# session identity server-side and inventing one would mean
|
|
# threading a client-supplied token through every read path. The
|
|
# note grain answers the question without one: of the distinct
|
|
# notes surface X chose, how many did an agent open in this window?
|
|
#
|
|
# Guarded separately from the reads above, on #2663's actual
|
|
# lesson. That outage was a NOVEL SQL SHAPE the database rejected
|
|
# inside a broad except. This join is the novel shape here, and a
|
|
# failure in it must not take down two readouts that already work.
|
|
try:
|
|
pulled_ids = (
|
|
select(NoteUsageEvent.note_id)
|
|
.where(
|
|
NoteUsageEvent.created_at >= since,
|
|
NoteUsageEvent.user_id == user_id,
|
|
NoteUsageEvent.event == PULLED,
|
|
# autoescape because `_` is a LIKE wildcard: a bare
|
|
# like("mcp_%") also matches "mcpX…". The Python half
|
|
# of this readout uses str.startswith and has no such
|
|
# hazard; this is the SQL half's version of it.
|
|
NoteUsageEvent.source.startswith("mcp_", autoescape=True),
|
|
)
|
|
.distinct()
|
|
.subquery()
|
|
)
|
|
surfaced_pairs = (
|
|
select(NoteUsageEvent.source, NoteUsageEvent.note_id)
|
|
.where(
|
|
NoteUsageEvent.created_at >= since,
|
|
NoteUsageEvent.user_id == user_id,
|
|
NoteUsageEvent.event == SURFACED,
|
|
)
|
|
.distinct()
|
|
.subquery()
|
|
)
|
|
# DISTINCT on (source, note_id) FIRST, which is what lets the
|
|
# outer aggregate be a plain count(): the pairs are already
|
|
# unique, so the left join cannot multiply them and no
|
|
# count(DISTINCT) is needed to undo damage that never happens.
|
|
by_source_rows = (
|
|
await session.execute(
|
|
select(
|
|
surfaced_pairs.c.source,
|
|
func.count().label("notes_surfaced"),
|
|
func.count(pulled_ids.c.note_id).label("notes_pulled"),
|
|
)
|
|
.select_from(
|
|
surfaced_pairs.outerjoin(
|
|
pulled_ids,
|
|
pulled_ids.c.note_id == surfaced_pairs.c.note_id,
|
|
)
|
|
)
|
|
.group_by(surfaced_pairs.c.source)
|
|
)
|
|
).all()
|
|
except Exception:
|
|
logger.warning("per-source pull-through read failed", exc_info=True)
|
|
by_source_rows = None
|
|
|
|
# Rules, at their own grain and in their own block (milestone 333).
|
|
#
|
|
# Guarded separately from the reads above for the reason `by_source`
|
|
# is: this table is NEW, and an instance running upgraded code
|
|
# against un-migrated schema would otherwise take down two readouts
|
|
# that work perfectly in order to report a third that cannot.
|
|
#
|
|
# The queries themselves are the note block's shapes, not novel
|
|
# ones — a group-by on two indexed columns and two count(distinct).
|
|
# The distinct counts need their own queries for the same reason
|
|
# the note ones do: count(distinct rule_id) per group cannot be
|
|
# summed across groups without double-counting a rule two sources
|
|
# both touched.
|
|
try:
|
|
rule_rows = (
|
|
await session.execute(
|
|
select(
|
|
RuleUsageEvent.event,
|
|
RuleUsageEvent.source,
|
|
func.count().label("n"),
|
|
)
|
|
.where(
|
|
RuleUsageEvent.created_at >= since,
|
|
RuleUsageEvent.user_id == user_id,
|
|
)
|
|
.group_by(RuleUsageEvent.event, RuleUsageEvent.source)
|
|
)
|
|
).all()
|
|
rule_complete = await _complete_from(
|
|
session, RuleUsageEvent, user_id,
|
|
)
|
|
# The rows carry `source`, so the ranked/ambient split is done
|
|
# below rather than in SQL — the bulk surfaces started emitting
|
|
# on 2026-09-03 (#3473), so there IS an ambient class now.
|
|
#
|
|
# `distinct_rules_surfaced` deliberately counts BOTH classes. It
|
|
# answers "how many distinct rules did this install put in front
|
|
# of an agent at all", which is the denominator for dead weight
|
|
# — and a rule delivered by the preload a hundred times and
|
|
# never opened is the most important case that question has.
|
|
distinct_rules_surfaced = (
|
|
await session.execute(
|
|
select(func.count(func.distinct(RuleUsageEvent.rule_id)))
|
|
.where(
|
|
RuleUsageEvent.created_at >= since,
|
|
RuleUsageEvent.user_id == user_id,
|
|
RuleUsageEvent.event == RULE_SURFACED,
|
|
)
|
|
)
|
|
).scalar_one()
|
|
distinct_rules_pulled = (
|
|
await session.execute(
|
|
select(func.count(func.distinct(RuleUsageEvent.rule_id)))
|
|
.where(
|
|
RuleUsageEvent.created_at >= since,
|
|
RuleUsageEvent.user_id == user_id,
|
|
RuleUsageEvent.event == RULE_PULLED,
|
|
)
|
|
)
|
|
).scalar_one()
|
|
except Exception:
|
|
logger.warning("rule usage read failed", exc_info=True)
|
|
rule_rows = None
|
|
distinct_rules_surfaced = distinct_rules_pulled = 0
|
|
except Exception:
|
|
logger.warning("retrieval summary read failed", exc_info=True)
|
|
out["read_failed"] = True
|
|
return out
|
|
|
|
from scribe.services.note_usage import AMBIENT_SOURCES
|
|
|
|
usage = {
|
|
"surfaced": 0, "ambient": 0,
|
|
"pulled": 0, "pulled_by_agent": 0, "pulled_by_human": 0,
|
|
"distinct_notes_surfaced": int(distinct_surfaced or 0),
|
|
"distinct_notes_pulled": int(distinct_pulled or 0),
|
|
}
|
|
for event, source, n in urows:
|
|
n = int(n)
|
|
if event == SURFACED:
|
|
if source in AMBIENT_SOURCES:
|
|
usage["ambient"] += n
|
|
else:
|
|
usage["surfaced"] += n
|
|
elif event == PULLED:
|
|
usage["pulled"] += n
|
|
# The mcp_/rest_ split is load-bearing (see NoteUsageEvent's own
|
|
# comment, which names #1038 — this readout's whole purpose). "Is
|
|
# this record dead weight?" is answered by ANY pull; "was that
|
|
# injected line useful to the agent?" only by an AGENT pull. So
|
|
# pull-through, which exists to answer the second, counts mcp_*
|
|
# only. Both halves are reported so the first question is still
|
|
# answerable from the same payload.
|
|
if source.startswith("mcp_"):
|
|
usage["pulled_by_agent"] += n
|
|
else:
|
|
usage["pulled_by_human"] += n
|
|
# Ranked surfacings in the denominator, agent pulls in the numerator: the
|
|
# "surfaced often, opened never" reading is only valid where a scored
|
|
# surface CHOSE the record and an agent was the one who declined it.
|
|
usage["pull_through"] = (
|
|
round(usage["pulled_by_agent"] / usage["surfaced"], 4)
|
|
if usage["surfaced"] else None
|
|
)
|
|
|
|
# The same question, per surface — which is the one the top-level ratio
|
|
# cannot answer. A corpus average of 0.05 is compatible with one surface
|
|
# earning its noise and another producing none, and tuning a threshold
|
|
# needs to know which.
|
|
#
|
|
# UPPER BOUND, and say so where it will be read: a pull records the door,
|
|
# not the surface that led to it, so a note surfaced by two surfaces and
|
|
# opened once counts as pulled for both. Attribution would need the session
|
|
# identity #2085 declined to invent. The bound is still decisive in the
|
|
# direction that matters — a surface reading near zero here is not being
|
|
# flattered by the double-count.
|
|
if by_source_rows is None:
|
|
usage["by_source"] = {}
|
|
# Distinct from an empty window, for the same reason `read_failed` is.
|
|
usage["by_source_failed"] = True
|
|
else:
|
|
by_source: dict[str, dict] = {}
|
|
for source, n_surfaced, n_pulled in by_source_rows:
|
|
n_surfaced, n_pulled = int(n_surfaced or 0), int(n_pulled or 0)
|
|
ambient = source in AMBIENT_SOURCES
|
|
by_source[source] = {
|
|
"notes_surfaced": n_surfaced,
|
|
"notes_pulled": n_pulled,
|
|
# None rather than a number on an ambient surface: nothing
|
|
# CHOSE those records, so "surfaced often, opened never" is not
|
|
# a judgment about them. The counts stay visible; the ratio
|
|
# that would be misread does not.
|
|
"pull_through": (
|
|
None if ambient or not n_surfaced
|
|
else round(n_pulled / n_surfaced, 4)
|
|
),
|
|
"ambient": ambient,
|
|
}
|
|
usage["by_source"] = by_source
|
|
|
|
# The SECTION's coverage, from the latest source to start recording — a
|
|
# figure that sums several sources is complete only once every one of them
|
|
# was being written. `_complete_from` computes that as "*".
|
|
usage.update(_coverage((note_complete or {}).get("*"), since))
|
|
out["usage"] = usage
|
|
|
|
# ── Rules, deliberately a SEPARATE block ────────────────────────────
|
|
#
|
|
# Not folded into `usage`, for two reasons and the second is the one that
|
|
# bites. The corpora differ by orders of magnitude — a few dozen eligible
|
|
# rules against thousands of notes — so one blended ratio would be the note
|
|
# ratio with a little noise on it, and the rule arm's own behaviour would
|
|
# be undetectable inside it. And `usage` is what existing callers already
|
|
# read: silently changing what it counts would move a number people have
|
|
# been comparing across windows, without telling them it now measures
|
|
# something else.
|
|
#
|
|
# `ambient` now carries the bulk deliveries — the SessionStart preload,
|
|
# `list_always_on_rules`, and every `rules_payload` surface (#3473). Before
|
|
# they emitted, this block had no ambient key and said the absence was a
|
|
# fact about the data. It was, and it was also the thing that made the
|
|
# always-on set impossible to judge: the largest rule surface in the
|
|
# product was the one surface its own scoreboard could not see.
|
|
#
|
|
# READ THE TWO SEPARATELY, ALWAYS. `surfaced` is a claim a ranker made and
|
|
# a pull can settle. `ambient` is a delivery nobody chose, so a high count
|
|
# says the set is large and resident, never that it is useful.
|
|
rule_usage = {
|
|
"surfaced": 0, "ambient": 0,
|
|
"pulled": 0, "pulled_by_agent": 0, "pulled_by_human": 0,
|
|
"distinct_rules_surfaced": int(distinct_rules_surfaced or 0),
|
|
"distinct_rules_pulled": int(distinct_rules_pulled or 0),
|
|
}
|
|
if rule_rows is None:
|
|
# The FLAG is added, the shape is kept — matching `by_source_failed`
|
|
# one block up. A caller that renders this must not have to choose
|
|
# between crashing on a missing key and quietly showing zeros it has no
|
|
# right to: the keys let it render, and the flag tells it the zeros are
|
|
# "we could not find out" rather than "nothing happened" (#2663).
|
|
rule_usage["rule_usage_failed"] = True
|
|
else:
|
|
for event, source, n in rule_rows:
|
|
n = int(n)
|
|
if event == RULE_SURFACED:
|
|
# One definition of ranked-vs-ambient, imported rather than
|
|
# restated — the per-rule badge readout reads the same
|
|
# predicate, and two spellings of "what counts as surfaced" is
|
|
# precisely the uneven wiring #3246 found across this system.
|
|
if is_ambient(source):
|
|
rule_usage["ambient"] += n
|
|
else:
|
|
rule_usage["surfaced"] += n
|
|
elif event == RULE_PULLED:
|
|
rule_usage["pulled"] += n
|
|
# Same split, and it carries MORE weight here than for notes.
|
|
# The arm's whole claim is "this rule may apply to what you are
|
|
# writing", and only an agent opening it says the claim landed.
|
|
# A person browsing the rule list says nothing about the hint.
|
|
if source.startswith("mcp_"):
|
|
rule_usage["pulled_by_agent"] += n
|
|
else:
|
|
rule_usage["pulled_by_human"] += n
|
|
|
|
# None, not 0.0, when nothing was surfaced — matching the note block. A
|
|
# ratio of zero asserts "we showed rules and none were opened"; with an
|
|
# empty numerator AND denominator that is a claim the data does not
|
|
# support, and it is the reading that would make a brand-new install look
|
|
# like a broken one.
|
|
#
|
|
# RANKED SURFACINGS ONLY in the denominator, and this is the load-bearing
|
|
# line of the whole change. Pull-through asks "was that hint any use", and
|
|
# only a surface that CHOSE what it showed can be judged by it. Folding the
|
|
# preload in would divide the same pulls by a number that grows with every
|
|
# session and every rule added to the resident set — so enlarging the
|
|
# always-on set would DEPRESS the arm's measured precision, and trimming it
|
|
# would flatter it, neither for any reason to do with the arm. The ambient
|
|
# count sits beside it, unaveraged, and is read as size rather than skill.
|
|
rule_usage["pull_through"] = (
|
|
round(rule_usage["pulled_by_agent"] / rule_usage["surfaced"], 4)
|
|
if rule_usage["surfaced"] else None
|
|
)
|
|
rule_usage.update(_coverage((rule_complete or {}).get("*"), since))
|
|
out["rule_usage"] = rule_usage
|
|
|
|
return out
|