feat(observability): capture LLM token usage + cost (#118)

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e:

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e:

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e:

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e:

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e:

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e:

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e:

backend/agents/document.py

@@ -52,6 +52,7 @@
 from agents.concept_extraction import concept_extraction_agent, ConceptList
 from agents.syllabus_extraction import syllabus_extraction_agent, SyllabusAssignments
 from agents.tools.graph import apply_concepts_to_graph
+from agents.usage import record_agent_usage
 from services.durable import workflow as durable_workflow, step as durable_step
 
 
@@ -86,25 +87,37 @@ class _WorkerResults:
 
 @durable_step
 async def _step_classify(text: str, deps: SaplingDeps) -> DocumentClassification:
- result = await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await classifier_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="classifier", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_summary(text: str, deps: SaplingDeps) -> Summary:
- result = await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await summary_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="summary", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_concepts(text: str, deps: SaplingDeps) -> ConceptList:
- result = await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await concept_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="concepts", user_id=deps.user_id,
+ )
  return result.output
 
 
 @durable_step
 async def _step_syllabus(text: str, deps: SaplingDeps) -> SyllabusAssignments:
- result = await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS)
+ result = record_agent_usage(
+ await syllabus_extraction_agent.run(text, deps=deps, usage_limits=WORKER_LIMITS),
+ feature="document", task="syllabus", user_id=deps.user_id,
+ )
  return result.output
 
 

backend/agents/usage.py

@@ -0,0 +1,82 @@
+"""One-line usage capture for Pydantic AI agent runs (issue #118).
+
+Every agent call site wraps its result in ``record_agent_usage(result,
+feature=..., task=...)``. The helper reads ``result.usage()`` and the model
+actually used, then hands them to ``events_service.log_llm_usage`` (which
+normalizes tokens, computes cost, and enqueues off the request thread).
+
+Two properties matter:
+
+* **One line per call site.** Because it returns ``result`` unchanged, a call
+ site can wrap inline (``result = record_agent_usage(await agent.run(...),
+ feature=..., task=...)``) or add it as a trailing statement.
+* **Never raises.** Instrumentation must not break an agent run, so every
+ failure — a result shape we don't recognize, a usage extraction slip — is
+ swallowed and logged at debug level.
+
+The ``model`` is read from the result's final ``ModelResponse`` (the model the
+provider actually served); if that's unavailable it falls back to the task's
+configured model via ``_providers.model_for``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents._providers import AgentTask, model_for
+from services import events_service
+
+logger = logging.getLogger("sapling.agents.usage")
+
+
+def _model_name(result: Any, task: AgentTask | None) -> str:
+ """Best-effort model id for the run, resilient to Pydantic AI churn."""
+ # Preferred: the final ModelResponse carries the served model name.
+ try:
+ name = getattr(result.response, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Fallback: scan the message history for the last response with a model.
+ try:
+ for msg in reversed(result.all_messages()):
+ name = getattr(msg, "model_name", None)
+ if name:
+ return name
+ except Exception:
+ pass
+ # Last resort: the task's configured default model.
+ if task is not None:
+ try:
+ return model_for(task).model_name
+ except Exception:
+ pass
+ return "unknown"
+
+
+def record_agent_usage(
+ result: Any,
+ *,
+ feature: str,
+ task: AgentTask | None = None,
+ user_id: str | None = None,
+) -> Any:
+ """Record token usage for an agent run and return ``result`` unchanged.
+
+ ``user_id`` is optional: pass it where the actor is in scope (routes with a
+ ``deps.user_id`` / request body) for per-user rollups; omit it and the
+ request_id from the contextvar still attributes the row.
+ """
+ try:
+ events_service.log_llm_usage(
+ feature=feature,
+ task=task,
+ model=_model_name(result, task),
+ usage=result.usage(),
+ user_id=user_id,
+ )
+ except Exception:
+ logger.debug("record_agent_usage: could not capture usage", exc_info=True)
+ return result

backend/db/migrations/0035_observability.sql

@@ -0,0 +1,57 @@
+-- 0035: observability tables (issue #115) — the two tables backing the site
+-- logging + usage-tracking system.
+--
+-- events — flexible, high-volume analytics / audit / error events.
+-- llm_usage — structured per-call LLM token + cost rows, queried with
+-- SUM/GROUP BY for billing rollups.
+--
+-- Notes / deviations from the original issue spec:
+-- * Ids and user_id are TEXT, not uuid, to match the rest of this schema
+-- (users.id is TEXT, e.g. 'user_andres'); a uuid user_id column could not
+-- hold the existing text ids. Same gen_random_uuid()::text PK style as
+-- 0026_ops.sql.
+-- * No FK on user_id. These are append-only, high-write analytics tables and
+-- user_id is intentionally nullable (system/anonymous actors); we don't
+-- want a per-row FK check on the write path or cascade coupling to the
+-- users lifecycle. Kept as a plain nullable column.
+-- * NO raw-content columns. Sensitive content is represented only by
+-- content_fp (a 16-hex sha256 fingerprint). Never store message text,
+-- document text, or names here.
+-- * admin_audit_log is deliberately untouched — events complements it.
+--
+-- Idempotent (IF NOT EXISTS throughout) so it is safe to re-run.
+
+-- ── events ──────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS events (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ event_type TEXT NOT NULL, -- dotted taxonomy: document.upload, auth.login, error.5xx
+ category TEXT NOT NULL, -- usage | audit | error
+ user_id TEXT, -- actor; NULL for anonymous/system
+ request_id TEXT, -- correlates to RequestIDMiddleware + Logfire
+ payload JSONB NOT NULL DEFAULT '{}', -- type-specific metadata (counts, ids, status_code, duration_ms…)
+ content_fp TEXT, -- sha256 fingerprint (16 hex) — never raw content
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_events_user_created ON events (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_type_created ON events (event_type, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_events_cat_created ON events (category, created_at DESC);
+
+-- ── llm_usage ────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS llm_usage (
+ id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ user_id TEXT, -- per-user rollups; NULL for system
+ request_id TEXT,
+ feature TEXT NOT NULL, -- quiz | chat_tutor | document | notes …
+ task TEXT, -- matches agents/_providers.py task slots
+ model TEXT NOT NULL, -- e.g. gemini-2.5-flash
+ provider TEXT NOT NULL DEFAULT 'gemini',
+ prompt_tokens INTEGER NOT NULL DEFAULT 0,
+ completion_tokens INTEGER NOT NULL DEFAULT 0,
+ total_tokens INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(12,6), -- NULL when the model isn't priced
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_llm_usage_user_created ON llm_usage (user_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_llm_usage_feature_created ON llm_usage (feature, created_at DESC);

backend/main.py

@@ -24,6 +24,7 @@
 from routes import graph, learn, quiz, calendar, social, extract, auth, documents, flashcards, study_guide, feedback, careers, onboarding, gradebook, gradescope, notes, academics
 from routes.profile import router as profile_router
 from routes.admin import router as admin_router
+from routes.admin_analytics import router as admin_analytics_router
 from routes.newsletter import router as newsletter_router
 from services.logfire_scrubber import EXTRA_PATTERNS, scrub_value
 from services.request_context import RequestIDMiddleware, current_request_id
@@ -80,8 +81,14 @@ async def _lifespan(_app: FastAPI):
  file_size_limit=MAX_AVATAR_SIZE,
  allowed_mime_types=sorted(ALLOWED_CONTENT_TYPES),
  )
+ # #116/#118: start the fire-and-forget observability drain thread so LLM
+ # usage + event rows flush off the request path.
+ from services import events_service
+ events_service.start_worker()
  yield
- # No shutdown hooks today.
+ # Stop the drain thread and flush anything still queued so the last batch
+ # of usage rows isn't lost on shutdown.
+ events_service.shutdown()
 
 
 def _drop_request_arguments(_request, _attributes):
@@ -199,6 +206,7 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
 app.include_router(onboarding.router, prefix="/api/onboarding")
 app.include_router(profile_router, prefix="/api/profile")
 app.include_router(admin_router, prefix="/api/admin")
+app.include_router(admin_analytics_router, prefix="/api/admin/analytics")
 app.include_router(newsletter_router, prefix="/api/newsletter")
 app.include_router(gradebook.router, prefix="/api/gradebook")
 app.include_router(gradescope.router, prefix="/api/gradescope")
@@ -267,9 +275,13 @@ def gemini_test(request: Request):
  require_admin(request) # 403 unless the session belongs to an admin; 401 if unauthenticated
  from agents._run import run_agent_sync
  from agents.health import health_probe_agent
+ from agents.usage import record_agent_usage
  try:
- result = run_agent_sync(
- health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ result = record_agent_usage(
+ run_agent_sync(
+ health_probe_agent.run('Reply with exactly the text: Gemini OK')
+ ),
+ feature="health",
  )
  return {"ok": True, "reply": result.output.strip()}
  except Exception as e: