diff --git a/docs/content/Deploying/Access-Control.mdx b/docs/content/Deploying/Access-Control.mdx index dc238349..208c378e 100644 --- a/docs/content/Deploying/Access-Control.mdx +++ b/docs/content/Deploying/Access-Control.mdx @@ -72,7 +72,7 @@ If your IdP only exposes groups via the userinfo endpoint, DocsGPT backfills the ## Admin dashboard -Admins get a dashboard backed by a REST surface under `/api/admin` (every endpoint requires the admin role). All mutating actions are written to the `auth_events` audit log with the acting admin recorded in the metadata. +Admins get a dashboard backed by a REST surface under `/api/admin` (every endpoint requires the admin role). All mutating actions are written to the `auth_events` audit log with the acting admin recorded as the event's `actor_id`. | Method | Path | Description | | --- | --- | --- | @@ -84,8 +84,12 @@ Admins get a dashboard backed by a REST surface under `/api/admin` (every endpoi | `DELETE` | `/api/admin/users//role` | Revoke the admin role. | | `POST` | `/api/admin/users//revoke-sessions` | Force-logout a user. | | `GET` | `/api/admin/admins` | List current admins. | -| `GET` | `/api/admin/usage` | Token-usage series and top users. | -| `GET` | `/api/admin/audit` | Authentication/admin audit feed. | +| `GET` | `/api/admin/usage` | Usage series with tokens and spend, per-model split, latency percentiles, top users. Takes `days`, `bucket` and `group_by=model\|agent\|source`. | +| `GET` | `/api/admin/users//usage` | One user's spend: daily series plus a split by model and by flow. | +| `GET` | `/api/admin/activity` | [Merged activity feed](#activity-feed) across all three audit journals. | +| `GET` | `/api/admin/activity/events` | The `(event, category)` pairs this instance has recorded, for the filter UI. | +| `GET` | `/api/admin/activity/export` | The filtered feed as `format=csv` or `format=ndjson`. | +| `GET` | `/api/admin/audit` | Authentication/admin audit feed (`auth_events` only). | | `GET` | `/api/admin/devices/audit` | Remote-device audit feed. | | `GET` | `/api/admin/teams` | Instance-wide oversight of all teams. | | `GET` `PUT` `DELETE` | `/api/admin/quotas/...` | [Usage quotas](/Deploying/Usage-Quotas) for the instance, teams and users. | @@ -142,7 +146,36 @@ Sharing rules: ## Audit log -Access-control actions are appended to the `auth_events` table alongside the [authentication events](/Deploying/OIDC-SSO#login-auditing). This includes admin actions — `admin_user_activated` / `admin_user_deactivated`, `admin_sessions_revoked`, `role_granted` / `role_revoked` (with `metadata.source` = `manual` or `oidc_group`), `quota_policy_set` / `quota_policy_deleted` — and team events (`team.create`, `team.member_add`, `team.member_role`, `team.member_remove`, `team.share`, `team.unshare`, `team.transfer_owner`, `team.delete`). The acting admin is recorded in the event metadata. +Access-control actions are appended to the `auth_events` table alongside the [authentication events](/Deploying/OIDC-SSO#login-auditing). This includes admin actions — `admin_user_activated` / `admin_user_deactivated`, `admin_sessions_revoked`, `role_granted` / `role_revoked` (with `metadata.source` = `manual` or `oidc_group`), `quota_policy_set` / `quota_policy_deleted` — and team events (`team.create`, `team.member_add`, `team.member_role`, `team.member_remove`, `team.share`, `team.unshare`, `team.transfer_owner`, `team.delete`). + +Data-plane actions are recorded too: `source.created` / `source.deleted` / `source.reingested`, `agent.created` / `agent.updated` / `agent.deleted` / `agent.key_regenerated`, and `conversation.deleted` / `conversation.deleted_all`. + +Every row carries `actor_id` (who did it) and `target_id` (the user it was done to, or `NULL` when the event is not about a user), so "everything this admin did" is a single query: + +```sql +SELECT created_at, event, target_id, metadata +FROM auth_events +WHERE actor_id = 'the-admin' +ORDER BY created_at DESC; +``` + +### Activity feed + +The **Admin → Activity** tab merges three append-only journals into one timeline: + +| Journal | Category | What it records | +| --- | --- | --- | +| `auth_events` | `identity`, `access`, `config`, `data` | Sign-ins, provisioning, role and account changes, quota changes, resource mutations. | +| `device_audit_log` | `device` | Every remote-device command dispatch and its verdict. | +| `guardrail_events` | `safety` | Every guardrail decision on a request. | + +Filter by category, event name, actor, affected user, a time window, or a free-text search that reaches into each row's detail payload, then export the filtered feed as CSV or NDJSON. Exports stream and are capped at 100,000 rows — take a database dump for a full history. + +An unknown filter value is refused with a 400 rather than dropped. Dropping it would leave that facet unfiltered, and no filter means every row, so a typo or a stale bookmark would silently widen an audit view instead of narrowing it. + +The per-user panel in **Admin → Users** deliberately lists identity and access events only. Data-plane events are filed under the user who performed them, so on an active account routine activity would push a denied login or a role grant out of the window; use the Activity tab filtered by that user to see everything. + +Two guardrail columns are deliberately never projected into the feed or the export: `api_key` (a raw agent key) and `matched_value` (unredacted source text). Admin-gating is not a reason to widen what a list response carries. ## Related diff --git a/docs/content/Deploying/OIDC-SSO.mdx b/docs/content/Deploying/OIDC-SSO.mdx index 8851994e..0df8b73b 100644 --- a/docs/content/Deploying/OIDC-SSO.mdx +++ b/docs/content/Deploying/OIDC-SSO.mdx @@ -185,7 +185,14 @@ SCIM identifies users by `userName`, which DocsGPT matches against its user id ## Login auditing -Authentication activity is recorded in `auth_events`, an append-only Postgres table carrying the user id, event name, IP address, user agent, a JSONB `metadata` column, and a timestamp: +Authentication activity is recorded in `auth_events`, an append-only Postgres table carrying the event name, IP address, user agent, a JSONB `metadata` column, a timestamp, and two attribution columns: + +- `actor_id` — who performed the action. Never empty. +- `target_id` — the user the action was performed on, or `NULL` when the event is not about a user (a team change, an instance-wide policy). + +They differ whenever someone acts on someone else: an admin deactivating an account is `actor_id = `, `target_id = `. For self-service events such as a login they are the same user. Automated actors are named `system:oidc` (a group-driven role change at login) and `system:scim` (provisioning). + +A `BEFORE INSERT` trigger fills both columns for any row that arrives without an `actor_id`, applying the same rules as the migration's backfill. That keeps attribution intact for a writer from a release that predates the columns — during a rolling deployment, for instance — instead of rejecting the insert or flattening the actor to a sentinel. | Event | Recorded when | | --- | --- | @@ -197,14 +204,19 @@ Authentication activity is recorded in `auth_events`, an append-only Postgres ta | `role_granted` / `role_revoked` | The admin role is granted/revoked (`metadata.source` is `manual` or `oidc_group`). | | `admin_user_activated` / `admin_user_deactivated` | An admin activates/deactivates a user. | | `admin_sessions_revoked` | An admin force-logs-out a user. | +| `pat_created` / `pat_revoked` / `pat_regenerated` | Personal access token lifecycle. | +| `quota_policy_set` / `quota_policy_deleted` | A [usage quota](/Deploying/Usage-Quotas) is changed. | | `team.*` | Team management — `team.create`, `team.member_add`, `team.member_role`, `team.member_remove`, `team.share`, `team.unshare`, `team.transfer_owner`, `team.delete`. | +| `source.*` | `source.created`, `source.deleted`, `source.reingested`. | +| `agent.*` | `agent.created`, `agent.updated`, `agent.deleted`, `agent.key_regenerated`. | +| `conversation.*` | `conversation.deleted`, `conversation.deleted_all`. | -The acting admin is recorded in the event metadata. See [Access Control, Roles & Teams](/Deploying/Access-Control) for the admin and team features that emit these events. +`agent.updated` records the names of the fields that changed, never their values — an agent's fields carry prompts and credentials. -There is no UI for these events yet — query the table directly: +See [Access Control, Roles & Teams](/Deploying/Access-Control#activity-feed) for the admin Activity view that reads these events, and for the two other journals it merges them with. To query the table directly: ```sql -SELECT created_at, event, user_id, ip, metadata +SELECT created_at, event, actor_id, target_id, ip, metadata FROM auth_events ORDER BY created_at DESC LIMIT 50; diff --git a/docsgpt/alembic/versions/0034_auth_events_actor_target.py b/docsgpt/alembic/versions/0034_auth_events_actor_target.py new file mode 100644 index 00000000..f3eb0877 --- /dev/null +++ b/docsgpt/alembic/versions/0034_auth_events_actor_target.py @@ -0,0 +1,181 @@ +"""0034 auth events actor/target — who did it, and to whom. + +``auth_events.user_id`` was overloaded: admin mutations file the row under the +*target* user and hide the acting admin in ``metadata->>'by'``, team events file +it under the *actor*, and quota events switch between the two depending on +scope. The practical consequence is that "show me everything admin X did" is +unanswerable — the first question any audit review asks. + +This adds two explicit columns: + +* ``actor_id`` — who performed the action (never NULL going forward). +* ``target_id`` — the user the action was performed on; NULL when the event is + not about a user (a team or an instance-wide policy change). + +``user_id`` is deliberately left alone. It stays the per-user feed key +(``list_recent``) and every historical row keeps its meaning, so this migration +is additive and the backfill never rewrites it. + +Backfill rules, applied only to rows that predate the columns: + +* ``actor_id`` = the first present of ``metadata->>'by'``, + ``metadata->>'granted_by'``, ``metadata->>'revoked_by'``, else ``user_id``. +* ``target_id`` = ``user_id``, except for events that were always filed under + the actor and have no single user target: ``team.*``, and instance- or + team-scoped ``quota_policy_*`` changes. + +The same two rules are installed as SQL functions and applied by a BEFORE +INSERT trigger to any row that arrives without an ``actor_id``, so a writer +from a release that predates these columns keeps its attribution instead of +being rejected or flattened to a sentinel. + +Also adds the indexes the global admin feed needs. Before this the only index +was ``(user_id, created_at DESC)``, so the cross-user feed — which orders by +``created_at DESC`` with no user predicate — degraded to a sequential scan plus +a sort on every page. + +Idempotent both ways. + +Revision ID: 0034_auth_events_actor_target +Revises: 0033_quotas +""" + +from typing import Sequence, Union + +from alembic import op + + +revision: str = "0034_auth_events_actor_target" +down_revision: Union[str, None] = "0033_quotas" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute("ALTER TABLE auth_events ADD COLUMN IF NOT EXISTS actor_id TEXT;") + op.execute("ALTER TABLE auth_events ADD COLUMN IF NOT EXISTS target_id TEXT;") + + # The derivation rules, as functions, so the backfill below and the + # compatibility trigger share one definition instead of two copies of the + # same CASE drifting apart. + op.execute( + """ + CREATE OR REPLACE FUNCTION auth_events_derive_actor( + metadata jsonb, user_id text + ) RETURNS text + LANGUAGE sql IMMUTABLE AS $$ + SELECT COALESCE( + metadata->>'by', + metadata->>'granted_by', + metadata->>'revoked_by', + user_id + ); + $$; + """ + ) + op.execute( + """ + CREATE OR REPLACE FUNCTION auth_events_derive_target( + event text, metadata jsonb, user_id text + ) RETURNS text + LANGUAGE sql IMMUTABLE AS $$ + SELECT CASE + -- Team events were always filed under the actor. + WHEN event LIKE 'team.%' THEN NULL + -- An instance or team quota policy is filed under the acting + -- admin, not a target user (see api/admin/quotas.py::_audit); + -- only a user-scoped policy has one. + WHEN event IN ('quota_policy_set', 'quota_policy_deleted') + AND COALESCE(metadata->>'scope', '') <> 'user' THEN NULL + ELSE user_id + END; + $$; + """ + ) + + # Backfill. ``actor_id IS NULL`` scopes this to pre-migration rows, so a + # re-run (or a downgrade/upgrade cycle) never clobbers written values. + op.execute( + """ + UPDATE auth_events + SET actor_id = auth_events_derive_actor(metadata, user_id), + target_id = auth_events_derive_target(event, metadata, user_id) + WHERE actor_id IS NULL; + """ + ) + + # A writer from a release that predates these columns names only the old + # ones, so ``actor_id`` arrives NULL. Two things follow, and the trigger + # handles both: + # + # 1. The NOT NULL below would reject the insert. In admin/routes.py that + # insert shares the request's transaction, so a NotNullViolation would + # 500 the deactivate or role grant and roll its write back with it. + # 2. Defaulting it to a sentinel would keep the write but lose the + # attribution -- and the highest-volume events are the recoverable + # ones: for a login or a silent renewal the actor is simply the user. + # + # ``actor_id IS NULL`` identifies a legacy insert exactly, because the + # repository always supplies one. That matters for ``target_id``: a current + # writer sets it to NULL deliberately for events with no user target + # (``conversation.deleted``), and deriving there would wrongly name the + # actor as their own target. + # + # Kept rather than dropped in a follow-up: it costs nothing on the path the + # repository takes, and it keeps the column's contract true for any writer + # that bypasses it. + op.execute( + """ + CREATE OR REPLACE FUNCTION auth_events_fill_attribution() + RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.actor_id IS NULL THEN + NEW.actor_id := auth_events_derive_actor( + NEW.metadata, NEW.user_id + ); + NEW.target_id := auth_events_derive_target( + NEW.event, NEW.metadata, NEW.user_id + ); + END IF; + RETURN NEW; + END; + $$; + """ + ) + op.execute("DROP TRIGGER IF EXISTS auth_events_fill_attribution_trg ON auth_events;") + op.execute( + """ + CREATE TRIGGER auth_events_fill_attribution_trg + BEFORE INSERT ON auth_events + FOR EACH ROW EXECUTE FUNCTION auth_events_fill_attribution(); + """ + ) + + # Backfilled above and guaranteed by the trigger, so this is safe to assert. + op.execute("ALTER TABLE auth_events ALTER COLUMN actor_id SET NOT NULL;") + + op.execute( + "CREATE INDEX IF NOT EXISTS auth_events_created_idx " + "ON auth_events (created_at DESC);" + ) + op.execute( + "CREATE INDEX IF NOT EXISTS auth_events_event_created_idx " + "ON auth_events (event, created_at DESC);" + ) + op.execute( + "CREATE INDEX IF NOT EXISTS auth_events_actor_idx " + "ON auth_events (actor_id, created_at DESC);" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS auth_events_actor_idx;") + op.execute("DROP INDEX IF EXISTS auth_events_event_created_idx;") + op.execute("DROP INDEX IF EXISTS auth_events_created_idx;") + op.execute("DROP TRIGGER IF EXISTS auth_events_fill_attribution_trg ON auth_events;") + op.execute("DROP FUNCTION IF EXISTS auth_events_fill_attribution();") + op.execute("ALTER TABLE auth_events DROP COLUMN IF EXISTS target_id;") + op.execute("ALTER TABLE auth_events DROP COLUMN IF EXISTS actor_id;") + op.execute("DROP FUNCTION IF EXISTS auth_events_derive_target(text, jsonb, text);") + op.execute("DROP FUNCTION IF EXISTS auth_events_derive_actor(jsonb, text);") diff --git a/docsgpt/alembic/versions/0035_token_usage_latency.py b/docsgpt/alembic/versions/0035_token_usage_latency.py new file mode 100644 index 00000000..b06d3203 --- /dev/null +++ b/docsgpt/alembic/versions/0035_token_usage_latency.py @@ -0,0 +1,42 @@ +"""0035 token usage latency — how long the call took, and time to first token. + +Nothing in the schema recorded how long an LLM call took. The wrappers in +``docsgpt/usage.py`` already measured it for the ``llm_gen_finished`` / +``llm_stream_finished`` log lines and then threw the number away, so an +operator could see what an instance spent but never how slow it was. + +Two columns, both nullable because both are genuinely unknown for some rows: + +* ``duration_ms`` — wall-clock for the whole call. +* ``ttft_ms`` — time to the first streamed chunk. NULL on non-streaming calls + and on streams that failed before yielding anything, which is the honest + reading: "no first token", not "instant". + +NULL also means "written before this migration", so historical rows never +drag a p50 towards zero. + +Idempotent both ways. + +Revision ID: 0035_token_usage_latency +Revises: 0034_auth_events_actor_target +""" + +from typing import Sequence, Union + +from alembic import op + + +revision: str = "0035_token_usage_latency" +down_revision: Union[str, None] = "0034_auth_events_actor_target" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute("ALTER TABLE token_usage ADD COLUMN IF NOT EXISTS duration_ms INTEGER;") + op.execute("ALTER TABLE token_usage ADD COLUMN IF NOT EXISTS ttft_ms INTEGER;") + + +def downgrade() -> None: + op.execute("ALTER TABLE token_usage DROP COLUMN IF EXISTS ttft_ms;") + op.execute("ALTER TABLE token_usage DROP COLUMN IF EXISTS duration_ms;") diff --git a/docsgpt/alembic/versions/0036_device_audit_created_idx.py b/docsgpt/alembic/versions/0036_device_audit_created_idx.py new file mode 100644 index 00000000..6f1d0217 --- /dev/null +++ b/docsgpt/alembic/versions/0036_device_audit_created_idx.py @@ -0,0 +1,37 @@ +"""0036 device audit feed index — created_at for the merged activity feed. + +``device_audit_log`` carried only ``(device_id, created_at DESC)`` and +``(user_id, created_at DESC)``. Both serve a per-device or per-user lookup; +neither can serve the merged admin activity feed, which orders by +``created_at DESC`` across every user and device, so that branch of the union +was sequentially scanned on every page. + +``auth_events`` got the equivalent index in 0034 and ``guardrail_events`` has +had ``ix_guardrail_events_created`` since 0029; this closes the third journal. + +Idempotent both ways. + +Revision ID: 0036_device_audit_created_idx +Revises: 0035_token_usage_latency +""" + +from typing import Sequence, Union + +from alembic import op + + +revision: str = "0036_device_audit_created_idx" +down_revision: Union[str, None] = "0035_token_usage_latency" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute( + "CREATE INDEX IF NOT EXISTS device_audit_created_idx " + "ON device_audit_log (created_at DESC);" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS device_audit_created_idx;") diff --git a/docsgpt/api/admin/__init__.py b/docsgpt/api/admin/__init__.py index 7bfc3059..1dcdd686 100644 --- a/docsgpt/api/admin/__init__.py +++ b/docsgpt/api/admin/__init__.py @@ -1,4 +1,5 @@ from .routes import admin_ns from . import quotas # noqa: F401 (registers the quota resources on admin_ns) +from . import activity # noqa: F401 (registers the activity resources on admin_ns) __all__ = ["admin_ns"] diff --git a/docsgpt/api/admin/activity.py b/docsgpt/api/admin/activity.py new file mode 100644 index 00000000..736a0baf --- /dev/null +++ b/docsgpt/api/admin/activity.py @@ -0,0 +1,283 @@ +"""Admin activity feed: one timeline over all three audit journals. + +``GET /api/admin/audit`` remains the ``auth_events``-only feed. These endpoints +add the merged view an incident review actually wants — identity, access, +configuration, data-plane, remote-device and guardrail activity in one ordered +list — plus the filters and the export that make a review possible without +shelling into the database. + +Every resource here is ``@admin_required``, like the rest of the namespace. +""" + +from __future__ import annotations + +import csv +import io +import json +import logging +from datetime import datetime, timezone +from typing import Iterator, Optional + +from flask import Response, jsonify, make_response, request, stream_with_context +from flask_restx import Resource + +from docsgpt.api.admin.routes import _int_arg, _page, admin_ns +from docsgpt.api.user.authz import admin_required +from docsgpt.audit_events import ACTIVITY_CATEGORIES +from docsgpt.storage.db.repositories.activity import ( + ACTIVITY_COLUMNS, + ActivityRepository, +) +from docsgpt.storage.db.session import db_readonly + +logger = logging.getLogger(__name__) + +_FEEDS = ("auth", "device", "guardrail") +_EXPORT_FORMATS = ("csv", "ndjson") + +# An export runs against a live instance, so it is bounded. Operators needing +# the full history take a database dump, not an HTTP request. +_EXPORT_MAX_ROWS = 100_000 +_EXPORT_CHUNK = 1_000 + +_MAX_SEARCH_LENGTH = 200 + + +class _InvalidFilter(ValueError): + """A filter argument the caller got wrong. Surfaces as a 400.""" + + +def _facet_list(name: str, allowed: tuple[str, ...]) -> Optional[list[str]]: + """Parse a repeated/comma-separated query arg into known facet values. + + Rejects unknown values rather than dropping them. Dropping would leave the + request with no filter on that facet, and "no filter" means *everything* -- + so a typo or a stale chip would silently widen an audit view instead of + narrowing it, and on the export would stream the whole history. + + Args: + name: Query parameter name. + allowed: The permitted values. + + Returns: + The distinct values in the order given, or None when absent. + + Raises: + _InvalidFilter: If any supplied value is not in ``allowed``. + """ + raw: list[str] = [] + for value in request.args.getlist(name): + raw.extend(part.strip() for part in value.split(",") if part.strip()) + seen: list[str] = [] + for value in raw: + if value not in allowed: + raise _InvalidFilter( + f"Unknown {name}: {value!r}. Expected one of {', '.join(allowed)}." + ) + if value not in seen: + seen.append(value) + return seen or None + + +def _event_list() -> Optional[list[str]]: + """Event names to filter on. Unconstrained — the catalogue is data, not an enum.""" + names: list[str] = [] + for value in request.args.getlist("event"): + for part in value.split(","): + part = part.strip() + if part and part not in names: + names.append(part) + return names or None + + +def _time_arg(name: str) -> Optional[datetime]: + """Parse an ISO-8601 timestamp query arg, or None when absent. + + Raises: + _InvalidFilter: If present but unparseable. Silently ignoring it would + drop the time bound and return the full history. + """ + raw = (request.args.get(name) or "").strip() + if not raw: + return None + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError as exc: + raise _InvalidFilter( + f"{name} must be an ISO-8601 timestamp, got {raw!r}." + ) from exc + # A naive bound is read as UTC, matching how the feed stores timestamps. + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _search_arg() -> Optional[str]: + value = (request.args.get("search") or "").strip() + return value[:_MAX_SEARCH_LENGTH] or None + + +def _filters() -> dict: + """The filter set shared by the feed and the export.""" + return { + "feeds": _facet_list("feed", _FEEDS), + "categories": _facet_list("category", ACTIVITY_CATEGORIES), + "events": _event_list(), + "actor_id": request.args.get("actor_id") or None, + "user_id": request.args.get("user_id") or None, + "since": _time_arg("since"), + "until": _time_arg("until"), + "search": _search_arg(), + } + + +def _serialize(row: dict) -> dict: + """Make one feed row JSON-safe (timestamps to ISO-8601).""" + created = row.get("created_at") + return { + **row, + "created_at": created.isoformat() if hasattr(created, "isoformat") else created, + } + + +@admin_ns.route("/admin/activity") +class AdminActivityResource(Resource): + @admin_required + def get(self): + """Merged audit feed, newest first, with filters and a total.""" + page, page_size, offset = _page() + try: + filters = _filters() + except _InvalidFilter as exc: + return make_response(jsonify({"success": False, "message": str(exc)}), 400) + with db_readonly() as conn: + repo = ActivityRepository(conn) + total = repo.count(**filters) + rows = repo.list(limit=page_size, offset=offset, **filters) + return make_response( + jsonify( + { + "success": True, + "activity": [_serialize(row) for row in rows], + "page": page, + "page_size": page_size, + "total": total, + "has_more": offset + len(rows) < total, + } + ), + 200, + ) + + +@admin_ns.route("/admin/activity/events") +class AdminActivityCatalogueResource(Resource): + @admin_required + def get(self): + """Distinct ``(event, category)`` pairs this instance has recorded. + + The filter UI offers these instead of a free-text box, so an operator + never has to guess that denied logins are ``oidc_login_denied``. + """ + with db_readonly() as conn: + catalogue = ActivityRepository(conn).event_names() + return make_response( + jsonify( + { + "success": True, + "events": catalogue, + "categories": list(ACTIVITY_CATEGORIES), + "feeds": list(_FEEDS), + } + ), + 200, + ) + + +@admin_ns.route("/admin/activity/export") +class AdminActivityExportResource(Resource): + @admin_required + def get(self): + """Stream the filtered feed as CSV or NDJSON. + + Streamed, not buffered: a compliance export of a busy instance should + not be built in memory first. Capped at ``_EXPORT_MAX_ROWS``, and the + ``X-Export-Max-Rows`` header reports the cap that was in effect so a + truncated export is not mistaken for a complete one. + """ + fmt = (request.args.get("format") or "csv").lower() + if fmt not in _EXPORT_FORMATS: + return make_response( + jsonify({"success": False, "message": "Invalid format"}), 400 + ) + limit = max(1, min(_EXPORT_MAX_ROWS, _int_arg("max_rows", _EXPORT_MAX_ROWS))) + try: + filters = _filters() + except _InvalidFilter as exc: + return make_response(jsonify({"success": False, "message": str(exc)}), 400) + stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + generator = _csv_rows if fmt == "csv" else _ndjson_rows + return Response( + stream_with_context(generator(filters, limit)), + mimetype="text/csv" if fmt == "csv" else "application/x-ndjson", + headers={ + "Content-Disposition": ( + f'attachment; filename="docsgpt-activity-{stamp}.{fmt}"' + ), + # The cap in effect. Without it a truncated export reads as + # a complete one. + "X-Export-Max-Rows": str(limit), + "Cache-Control": "no-store", + }, + ) + + +def _export_rows(filters: dict, limit: int) -> Iterator[dict]: + """Walk the filtered feed, holding the connection open for the whole stream.""" + with db_readonly() as conn: + yield from ActivityRepository(conn).iter_all( + chunk_size=_EXPORT_CHUNK, max_rows=limit, **filters + ) + + +# Characters a spreadsheet reads as the start of a formula. +_FORMULA_TRIGGERS = ("=", "+", "-", "@", "\t", "\r") + + +def _csv_safe(value: object) -> object: + """Neutralize a leading formula trigger before the value becomes a CSV cell. + + The feed carries attacker-controlled text: ``user_agent`` is the raw header + of whoever made the request, and a denied login records one without ever + authenticating. An export is opened in a spreadsheet by an admin, so a cell + beginning ``=`` would be evaluated there. Prefixing an apostrophe keeps the + value readable and inert. + """ + if isinstance(value, str) and value.startswith(_FORMULA_TRIGGERS): + return f"'{value}" + return value + + +def _csv_rows(filters: dict, limit: int) -> Iterator[str]: + buffer = io.StringIO() + writer = csv.DictWriter(buffer, fieldnames=list(ACTIVITY_COLUMNS)) + writer.writeheader() + yield _drain(buffer) + for row in _export_rows(filters, limit): + serialized = _serialize(row) + # ``detail`` is a JSON object; a CSV cell holds its compact encoding. + serialized["detail"] = json.dumps( + serialized.get("detail") or {}, default=str + ) + writer.writerow({key: _csv_safe(value) for key, value in serialized.items()}) + yield _drain(buffer) + + +def _ndjson_rows(filters: dict, limit: int) -> Iterator[str]: + for row in _export_rows(filters, limit): + yield json.dumps(_serialize(row), default=str) + "\n" + + +def _drain(buffer: io.StringIO) -> str: + """Return and clear the writer's buffer, so each row streams as written.""" + value = buffer.getvalue() + buffer.seek(0) + buffer.truncate(0) + return value diff --git a/docsgpt/api/admin/quotas.py b/docsgpt/api/admin/quotas.py index 0235f36b..83d2ec52 100644 --- a/docsgpt/api/admin/quotas.py +++ b/docsgpt/api/admin/quotas.py @@ -119,6 +119,10 @@ def _audit(conn, event: str, scope: str, subject_id: Optional[str], detail: dict ip=request.remote_addr, user_agent=request.headers.get("User-Agent"), metadata={"by": actor, "via": "admin_api", "scope": scope, "subject_id": subject_id, **detail}, + actor_id=actor or "unknown", + # Only a user-scoped policy targets a user; instance and team policies + # change configuration, not an account. + target_id=subject_id if scope == "user" else None, ) diff --git a/docsgpt/api/admin/routes.py b/docsgpt/api/admin/routes.py index 3d178904..7f86e5fa 100644 --- a/docsgpt/api/admin/routes.py +++ b/docsgpt/api/admin/routes.py @@ -38,6 +38,9 @@ admin_ns = Namespace("admin", description="Admin-only management endpoints", pat _DEFAULT_PAGE_SIZE = 25 _MAX_PAGE_SIZE = 100 +# Event namespaces the per-user security panel leaves out; see the drill-down. +_DATA_PLANE_PREFIXES = ("source.", "agent.", "conversation.") + def _int_arg(name: str, default: int) -> int: try: @@ -135,8 +138,12 @@ class AdminUserResource(Resource): }, "roles": sorted({"user", *roles_repo.role_names_for(user_id)}), "grants": roles_repo.list_for(user_id), + # Identity and access only. Data-plane events file under the + # actor, so an active user's routine deletes would push a + # denied login or a role grant out of a 20-row window. + # The Activity tab, filtered by user, shows everything. "recent_events": AuthEventsRepository(conn).list_recent( - user_id, limit=20 + user_id, limit=20, exclude_events_like=_DATA_PLANE_PREFIXES ), "counts": AdminStatsRepository(conn).user_counts(user_id), } @@ -171,6 +178,8 @@ class AdminUserResource(Resource): ip=request.remote_addr, user_agent=request.headers.get("User-Agent"), metadata={"by": actor, "via": "admin_api"}, + actor_id=actor, + target_id=user_id, ) if not active: # Best-effort live-session revocation (mirrors SCIM deactivation). @@ -203,6 +212,8 @@ class AdminUserRoleResource(Resource): "granted_by": actor, "via": "admin_api", }, + actor_id=actor, + target_id=user_id, ) return make_response( jsonify({"success": True, "granted": inserted, "role": ROLE_ADMIN}), 200 @@ -238,6 +249,8 @@ class AdminUserRoleResource(Resource): "revoked_by": actor, "via": "admin_api", }, + actor_id=actor, + target_id=user_id, ) return make_response(jsonify({"success": True, "revoked": removed}), 200) @@ -261,6 +274,8 @@ class AdminUserSessionsResource(Resource): ip=request.remote_addr, user_agent=request.headers.get("User-Agent"), metadata={"token_id": token_id, "by": _actor(), "via": "admin_sessions_revoked"}, + actor_id=_actor(), + target_id=user_id, ) AuthEventsRepository(conn).insert( user_id, @@ -273,10 +288,32 @@ class AdminUserSessionsResource(Resource): "persisted": ok, "personal_access_tokens_revoked": len(revoked_token_ids), }, + actor_id=_actor(), + target_id=user_id, ) return make_response(jsonify({"success": True, "revoked": ok}), 200) +@admin_ns.route("/admin/users//usage") +class AdminUserUsageResource(Resource): + @admin_required + def get(self, user_id): + """One user's spend: daily series plus a split by model and by flow.""" + days = max(1, min(365, _int_arg("days", 30))) + start = datetime.now(timezone.utc) - timedelta(days=days) + with db_readonly() as conn: + series = TokenUsageRepository(conn).bucketed_totals( + bucket_unit="day", user_id=user_id, timestamp_gte=start + ) + breakdown = AdminStatsRepository(conn).user_usage_breakdown( + user_id, since=start + ) + return make_response( + jsonify({"success": True, "days": days, "series": series, **breakdown}), + 200, + ) + + @admin_ns.route("/admin/admins") class AdminAdminsResource(Resource): @admin_required @@ -294,7 +331,12 @@ class AdminUsageResource(Resource): @admin_required def get(self): - """Global token usage: time-bucketed series + total + top users.""" + """Global usage: time-bucketed series, totals, spend, latency, top users. + + Every bucket carries both tokens and USD cost. Quota policies have + always been settable in dollars; until now nothing showed the spend + they were capping. + """ days = max(1, min(365, _int_arg("days", 30))) bucket = request.args.get("bucket", "day") group_by = request.args.get("group_by", "none") @@ -303,6 +345,7 @@ class AdminUsageResource(Resource): start = datetime.now(timezone.utc) - timedelta(days=days) with db_readonly() as conn: usage_repo = TokenUsageRepository(conn) + stats_repo = AdminStatsRepository(conn) series = usage_repo.bucketed_totals( bucket_unit=bucket, timestamp_gte=start, @@ -311,7 +354,8 @@ class AdminUsageResource(Resource): total = usage_repo.sum_tokens_in_range( start=start, end=datetime.now(timezone.utc) ) - top_users = AdminStatsRepository(conn).top_token_users(since=start, limit=10) + latency = stats_repo.latency_summary(since=start) + top_users = stats_repo.top_token_users(since=start, limit=10) return make_response( jsonify( { @@ -321,6 +365,10 @@ class AdminUsageResource(Resource): "group_by": group_by, "series": series, "total_tokens": int(total), + # Summed from the series, not from ``by_model``: the latter + # drops rows with no model_id, so it would undercount. + "total_cost": round(sum(row["cost"] for row in series), 4), + "latency": latency, "top_users": top_users, } ), diff --git a/docsgpt/api/audit.py b/docsgpt/api/audit.py new file mode 100644 index 00000000..a5d131b8 --- /dev/null +++ b/docsgpt/api/audit.py @@ -0,0 +1,74 @@ +"""Shared audit-trail helper for recording data-plane and identity events. + +Identity and access events (login, role grants, provisioning) have been +audited since ``auth_events`` was introduced. Data-plane actions — creating and +deleting sources, agents, agent keys and conversations — were not, which left +the trail unable to answer "who deleted that source". :func:`record_event` is +the one-line hook those routes use. + +Two properties matter and are enforced here rather than at every call site: + +1. **An audit write can never fail the action it records.** The insert runs in + a SAVEPOINT so a rejected row rolls back alone, and any exception is + swallowed with a log line. +2. **Request context is optional.** Celery tasks (ingestion finishing, a + scheduled run) have no Flask request; the row simply records without an IP + or user agent instead of raising. + +The event → category taxonomy the admin activity feed filters on lives in +``docsgpt/audit_events.py``, which the storage layer needs too. +""" + +from __future__ import annotations + +import logging +from typing import Any, Optional + +from flask import has_request_context, request +from sqlalchemy import Connection + +from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository + +logger = logging.getLogger(__name__) + + +def record_event( + conn: Connection, + event: str, + *, + actor: Optional[str], + target: Optional[str] = None, + **metadata: Any, +) -> None: + """Append one audit event, best-effort, inside the caller's transaction. + + Args: + conn: Open SQLAlchemy ``Connection`` -- not a Session -- inside the + action's transaction, so the audit row commits atomically with the + change it describes and ``begin_nested`` gives it a real SAVEPOINT. + event: Dotted event name (``source.deleted``, ``agent.created``). + actor: Who performed the action; ``"unknown"`` when unauthenticated. + target: The user acted upon, when the event is about a user. Data-plane + events act on a resource, not an account, so they leave this None. + **metadata: Free-form detail for the audit drill-down. ``None`` values + are dropped so the stored object only carries what was known. + """ + detail = {key: value for key, value in metadata.items() if value is not None} + ip = request.remote_addr if has_request_context() else None + user_agent = request.headers.get("User-Agent") if has_request_context() else None + resolved_actor = actor or "unknown" + try: + with conn.begin_nested(): + AuthEventsRepository(conn).insert( + # ``user_id`` is the per-user feed key: the target when the + # event is about someone, else the actor's own trail. + user_id=target or resolved_actor, + event=event, + ip=ip, + user_agent=user_agent, + metadata=detail, + actor_id=resolved_actor, + target_id=target, + ) + except Exception: + logger.warning("audit insert failed for event=%s", event, exc_info=True) diff --git a/docsgpt/api/oidc/routes.py b/docsgpt/api/oidc/routes.py index 90ca72b6..5d86333a 100644 --- a/docsgpt/api/oidc/routes.py +++ b/docsgpt/api/oidc/routes.py @@ -26,7 +26,10 @@ from docsgpt.api.oidc import denylist, provider from docsgpt.auth import handle_auth from docsgpt.cache import get_redis_instance from docsgpt.core.settings import settings -from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository +from docsgpt.storage.db.repositories.auth_events import ( + SYSTEM_ACTOR_OIDC, + AuthEventsRepository, +) from docsgpt.storage.db.repositories.user_roles import UserRolesRepository from docsgpt.storage.db.repositories.users import UsersRepository from docsgpt.storage.db.session import db_readonly, db_session @@ -129,6 +132,9 @@ def _reconcile_oidc_admin(user_id: str, groups: list[str] | None) -> None: ip=request.remote_addr, user_agent=request.headers.get("User-Agent"), metadata={"role": "admin", "source": "oidc_group"}, + # The provider's group membership drove this, not the user. + actor_id=SYSTEM_ACTOR_OIDC, + target_id=user_id, ) except Exception: logger.error("OIDC admin reconcile failed for %s", user_id, exc_info=True) diff --git a/docsgpt/api/pat/routes.py b/docsgpt/api/pat/routes.py index b8a9e00d..a8c0f0d6 100644 --- a/docsgpt/api/pat/routes.py +++ b/docsgpt/api/pat/routes.py @@ -307,6 +307,8 @@ class AdminToken(Resource): ip=request.remote_addr, user_agent=request.headers.get("User-Agent"), metadata={"token_id": token_id, "by": actor, "via": "admin_api"}, + actor_id=actor, + target_id=row["user_id"], ) if not revoked: return _error("Token not found", 404) diff --git a/docsgpt/api/scim/routes.py b/docsgpt/api/scim/routes.py index cb9338c7..e2393ab0 100644 --- a/docsgpt/api/scim/routes.py +++ b/docsgpt/api/scim/routes.py @@ -20,7 +20,10 @@ from sqlalchemy import Connection from docsgpt.api.oidc.denylist import deny_user from docsgpt.core.settings import settings -from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository +from docsgpt.storage.db.repositories.auth_events import ( + SYSTEM_ACTOR_SCIM, + AuthEventsRepository, +) from docsgpt.storage.db.repositories.users import UsersRepository from docsgpt.storage.db.session import db_readonly, db_session @@ -201,7 +204,13 @@ def _audit(conn: Connection, user_id: str, event: str) -> None: """Best-effort audit insert in a savepoint; failure never fails the request.""" try: with conn.begin_nested(): - AuthEventsRepository(conn).insert(user_id, event, metadata={"via": "scim"}) + AuthEventsRepository(conn).insert( + user_id, + event, + metadata={"via": "scim"}, + actor_id=SYSTEM_ACTOR_SCIM, + target_id=user_id, + ) except Exception: logger.error("SCIM audit insert failed for user %s event %s", user_id, event, exc_info=True) diff --git a/docsgpt/api/user/agents/routes.py b/docsgpt/api/user/agents/routes.py index 6a0ec2b0..84f7fbbe 100644 --- a/docsgpt/api/user/agents/routes.py +++ b/docsgpt/api/user/agents/routes.py @@ -9,6 +9,7 @@ from flask_restx import fields, Namespace, Resource from pydantic import ValidationError as PydanticValidationError from docsgpt.api import api +from docsgpt.api.audit import record_event from docsgpt.api.pat.rules import filter_listing, mask_agent_key, may_see_agent_keys from docsgpt.guardrails.config import AgentConfig from docsgpt.api.user.base import ( @@ -760,6 +761,15 @@ class CreateAgent(Resource): **kwargs, ) new_id = str(agent_row["id"]) + record_event( + conn, + "agent.created", + actor=user, + agent_id=new_id, + name=data["name"], + status=data["status"], + agent_type=agent_type, + ) except Exception as err: current_app.logger.error(f"Error creating agent: {err}", exc_info=True) return make_response(jsonify({"success": False}), 400) @@ -1284,6 +1294,16 @@ class UpdateAgent(Resource): updated = bool(result) else: updated = agents_repo.update(pg_agent_id, user, update_fields) + # An update that changed no field is not an event. + if updated and update_fields: + record_event( + conn, + "agent.updated", + actor=user, + agent_id=pg_agent_id, + # Names only: values can hold prompts and credentials. + fields=sorted(update_fields), + ) if not updated: return make_response( jsonify( @@ -1403,6 +1423,13 @@ class RegenerateAgentKey(Resource): SharedConversationsRepository(conn).reassign_api_key( old_key=old_key, new_key=new_key ) + record_event( + conn, + "agent.key_regenerated", + actor=user, + agent_id=pg_agent_id, + name=existing_agent.get("name"), + ) except Exception as err: current_app.logger.error( f"Error regenerating agent key: {err}", exc_info=True @@ -1457,6 +1484,14 @@ class DeleteAgent(Resource): agents_repo.delete(pg_agent_id, user) # Strip pinned/shared entries for this agent from the owner's prefs. UsersRepository(conn).remove_agent_from_all(user, pg_agent_id) + record_event( + conn, + "agent.deleted", + actor=user, + agent_id=pg_agent_id, + name=agent.get("name"), + agent_type=agent.get("agent_type"), + ) except Exception as err: current_app.logger.error(f"Error deleting agent: {err}", exc_info=True) return make_response(jsonify({"success": False}), 400) diff --git a/docsgpt/api/user/conversations/routes.py b/docsgpt/api/user/conversations/routes.py index a6c26807..189763bf 100644 --- a/docsgpt/api/user/conversations/routes.py +++ b/docsgpt/api/user/conversations/routes.py @@ -7,6 +7,7 @@ from flask_restx import fields, Namespace, Resource from sqlalchemy import text as sql_text from docsgpt.api import api +from docsgpt.api.audit import record_event from docsgpt.api.answer.services.conversation_service import ( TERMINATED_RESPONSE_PLACEHOLDER, ) @@ -45,6 +46,12 @@ class DeleteConversation(Resource): conv = repo.get_any(conversation_id, user_id) if conv is not None: repo.delete(str(conv["id"]), user_id) + record_event( + conn, + "conversation.deleted", + actor=user_id, + conversation_id=str(conv["id"]), + ) except Exception as err: current_app.logger.error( f"Error deleting conversation: {err}", exc_info=True @@ -65,7 +72,16 @@ class DeleteAllConversations(Resource): user_id = decoded_token.get("sub") try: with db_session() as conn: - ConversationsRepository(conn).delete_all_for_user(user_id) + deleted = ConversationsRepository(conn).delete_all_for_user(user_id) + # Nothing deleted is not an event; the endpoint is idempotent + # and a row here would render as a destructive action. + if deleted: + record_event( + conn, + "conversation.deleted_all", + actor=user_id, + deleted=deleted, + ) except Exception as err: current_app.logger.error( f"Error deleting all conversations: {err}", exc_info=True diff --git a/docsgpt/api/user/sources/routes.py b/docsgpt/api/user/sources/routes.py index 2708b387..6826c9e9 100644 --- a/docsgpt/api/user/sources/routes.py +++ b/docsgpt/api/user/sources/routes.py @@ -10,6 +10,7 @@ from pydantic import ValidationError from docsgpt.agents.tools.path_utils import validate_tool_path from docsgpt.api import api +from docsgpt.api.audit import record_event from docsgpt.api.pat.rules import filter_listing from docsgpt.api.user.tasks import ( convert_source_to_wiki, @@ -280,6 +281,13 @@ class DeleteOldIndexes(Resource): try: with db_session() as conn: SourcesRepository(conn).delete(resolved_id, user) + record_event( + conn, + "source.deleted", + actor=user, + source_id=resolved_id, + name=doc.get("name"), + ) except Exception as err: current_app.logger.error( f"Error deleting source row: {err}", exc_info=True @@ -500,6 +508,23 @@ class ReingestSource(Resource): exc_info=True, ) return make_response(jsonify({"success": False}), 400) + try: + with db_session() as conn: + record_event( + conn, + "source.reingested", + actor=user, + source_id=resolved_source_id, + name=doc.get("name"), + # A team editor can reingest a source they do not own. + owner=owner if owner != user else None, + task_id=task.id, + ) + except Exception as err: + current_app.logger.warning( + f"Could not audit reingest for {resolved_source_id}: {err}", + exc_info=True, + ) return make_response(jsonify({"success": True, "task_id": task.id}), 200) @@ -811,6 +836,14 @@ class CreateWikiSource(Resource): updated_via="human", ) rebuild_wiki_directory_structure(conn, source_id, user) + record_event( + conn, + "source.created", + actor=user, + source_id=source_id, + name=name, + type="wiki", + ) except Exception as err: current_app.logger.error( f"Error creating wiki source: {err}", exc_info=True diff --git a/docsgpt/api/user/sources/upload.py b/docsgpt/api/user/sources/upload.py index 004d9a4f..f15bee65 100644 --- a/docsgpt/api/user/sources/upload.py +++ b/docsgpt/api/user/sources/upload.py @@ -5,12 +5,14 @@ import os import tempfile import uuid import zipfile +from typing import Optional from flask import current_app, jsonify, make_response, request from flask_restx import fields, Namespace, Resource from sqlalchemy import text as sql_text from docsgpt.api import api +from docsgpt.api.audit import record_event from docsgpt.api.user.tasks import ingest, ingest_connector_task, ingest_remote from docsgpt.api.user.team_sharing import effective_write_owner from docsgpt.core.settings import settings @@ -178,6 +180,40 @@ def _source_archive_limits() -> ZipExtractionLimits: ) +def _audit_source_created( + *, + source_id: str, + user: Optional[str], + name: Optional[str], + source_type: str, + task_id: Optional[str], +) -> None: + """Record ``source.created`` for an accepted ingestion job. + + Audited at the request, not in the worker: enqueuing is the user's action. + The ingest may still fail, which the source's own status records. + + Every path that mints a ``source_id`` and dispatches must call this -- + ``source.deleted`` is recorded for remote sources too, and a trail showing + a deletion with no matching creation is worse than no trail. + """ + try: + with db_session() as conn: + record_event( + conn, + "source.created", + actor=user, + source_id=source_id, + name=name, + type=source_type, + task_id=task_id, + ) + except Exception as err: + current_app.logger.warning( + "Could not audit source.created for %s: %s", source_id, err, exc_info=True + ) + + @sources_upload_ns.route("/upload") class UploadFile(Resource): @api.expect( @@ -396,6 +432,13 @@ class UploadFile(Resource): return make_response(jsonify({"success": False}), 400) # Predetermined id matches the dedup-claim row; loser GET sees same. response_task_id = predetermined_task_id or task.id + _audit_source_created( + source_id=str(source_uuid), + user=user, + name=job_name, + source_type="local", + task_id=response_task_id, + ) # ``source_uuid`` was minted above and passed to the worker as # ``source_id``; the worker uses it verbatim for every SSE event, # so the frontend can correlate inbound ``source.ingest.*`` to @@ -547,6 +590,13 @@ class UploadRemote(Resource): connector_kwargs["task_id"] = predetermined_task_id task = ingest_connector_task.apply_async(**connector_kwargs) response_task_id = predetermined_task_id or task.id + _audit_source_created( + source_id=str(source_uuid), + user=user, + name=data["name"], + source_type=data["source"], + task_id=response_task_id, + ) # ``source_uuid`` was minted above and passed to the # worker as ``source_id``; the worker uses it verbatim # for every SSE event, so the frontend can correlate @@ -580,6 +630,13 @@ class UploadRemote(Resource): _release_claim(scoped_key) return make_response(jsonify({"success": False}), 400) response_task_id = predetermined_task_id or task.id + _audit_source_created( + source_id=str(source_uuid), + user=user, + name=data["name"], + source_type=data["source"], + task_id=response_task_id, + ) response_payload = { "success": True, "task_id": response_task_id, diff --git a/docsgpt/api/user/teams/routes.py b/docsgpt/api/user/teams/routes.py index ca3de7de..df51ec0e 100644 --- a/docsgpt/api/user/teams/routes.py +++ b/docsgpt/api/user/teams/routes.py @@ -62,13 +62,26 @@ def _current_user() -> str | None: return token.get("sub") if isinstance(token, dict) else None +# Metadata keys naming the user a team event acted on, most specific first. +# Lets ``_audit`` fill ``target_id`` without every call site repeating it. +_TARGET_METADATA_KEYS = ("target_user", "target_user_id", "new_owner") + + def _audit(conn, actor: str | None, event: str, **metadata) -> None: """Append a team management event to the audit trail (best-effort). Runs inside the action's transaction so the audit row commits atomically with the change. Never raises into the request path — an audit failure must not fail the operation. + + Team events are filed under the acting user. When the action names another + member (add, role change, removal, ownership transfer) that member becomes + the row's ``target_id``; otherwise the event has no user target. """ + detail = {k: v for k, v in metadata.items() if v is not None} + target = next( + (detail[key] for key in _TARGET_METADATA_KEYS if detail.get(key)), None + ) try: # SAVEPOINT: a failed audit insert poisons the surrounding txn, so # nest it — on failure only the audit rolls back, not the action. @@ -78,7 +91,9 @@ def _audit(conn, actor: str | None, event: str, **metadata) -> None: event=event, ip=request.remote_addr, user_agent=request.headers.get("User-Agent"), - metadata={k: v for k, v in metadata.items() if v is not None}, + metadata=detail, + actor_id=actor or "unknown", + target_id=target, ) except Exception: logger.warning("team audit insert failed for event=%s", event, exc_info=True) diff --git a/docsgpt/audit_events.py b/docsgpt/audit_events.py new file mode 100644 index 00000000..db6bf2e3 --- /dev/null +++ b/docsgpt/audit_events.py @@ -0,0 +1,104 @@ +"""Audit event taxonomy: the categories the admin activity feed filters on. + +Lives at the top level (like ``docsgpt/pricing.py``) because both the API +layer — which records events — and the storage layer — which has to filter a +merged feed in SQL — need the same map, and neither should import the other. + +Two representations of one table: + +* :func:`category_for` classifies a single event name in Python. +* :func:`category_case_sql` emits the equivalent ``CASE`` expression so a + category filter can be pushed into the query instead of enumerating every + event name a release might have written. +""" + +from __future__ import annotations + +# Categories in the order the UI shows them. ``other`` catches event names an +# older release wrote that this one does not know about — the feed must never +# hide a row just because its name is unfamiliar. +ACTIVITY_CATEGORIES = ("identity", "access", "config", "data", "device", "safety", "other") + +# Exact event names. +_EXACT: dict[str, str] = { + # Who you are: authentication, provisioning, personal credentials. + "oidc_login": "identity", + "oidc_login_denied": "identity", + "oidc_refresh": "identity", + "backchannel_logout": "identity", + "scim_created": "identity", + "scim_deactivated": "identity", + "scim_reactivated": "identity", + "pat_created": "identity", + "pat_revoked": "identity", + "pat_regenerated": "identity", + # What you may do: roles, account state. + "role_granted": "access", + "role_revoked": "access", + "admin_user_activated": "access", + "admin_user_deactivated": "access", + "admin_sessions_revoked": "access", + # How the instance is configured. + "quota_policy_set": "config", + "quota_policy_deleted": "config", +} + +# Dotted namespaces, matched on the part before the first ``.``. +_NAMESPACES: dict[str, str] = { + "team": "access", + "source": "data", + "agent": "data", + "conversation": "data", + "device": "device", + "guardrail": "safety", +} + + +def category_for(event: str) -> str: + """Return the activity category for ``event``. + + Args: + event: The audit event name (``oidc_login``, ``source.deleted``). + + Returns: + One of :data:`ACTIVITY_CATEGORIES`; ``"other"`` for names this release + does not recognise. + """ + namespace, dot, _ = event.partition(".") + if dot: + return _NAMESPACES.get(namespace, "other") + return _EXACT.get(event, "other") + + +def category_case_sql(column: str) -> str: + """Build the SQL ``CASE`` that classifies ``column`` the way :func:`category_for` does. + + Uses ``split_part`` rather than ``LIKE`` so the expression carries no ``%`` + — which a pyformat DBAPI would otherwise read as a parameter marker. + + Args: + column: The SQL expression holding an event name. + + Returns: + A parenthesised ``CASE`` expression. Every literal in it comes from the + maps above, never from user input, so it is safe to interpolate. + """ + namespace_branches = " ".join( + f"WHEN '{namespace}' THEN '{category}'" + for namespace, category in _NAMESPACES.items() + ) + by_category: dict[str, list[str]] = {} + for event, category in _EXACT.items(): + by_category.setdefault(category, []).append(event) + exact_branches = " ".join( + f"WHEN {column} IN ({', '.join(repr(event) for event in sorted(events))}) " + f"THEN '{category}'" + for category, events in sorted(by_category.items()) + ) + return ( + "(CASE " + f"WHEN position('.' in {column}) > 0 " + f"THEN (CASE split_part({column}, '.', 1) {namespace_branches} ELSE 'other' END) " + f"{exact_branches} " + "ELSE 'other' END)" + ) diff --git a/docsgpt/storage/db/base_repository.py b/docsgpt/storage/db/base_repository.py index 802c263d..3a8bee0d 100644 --- a/docsgpt/storage/db/base_repository.py +++ b/docsgpt/storage/db/base_repository.py @@ -65,3 +65,21 @@ def row_to_dict(row: Any) -> dict: out["_id"] = out["id"] return out + + +def like_escape(term: str) -> str: + """Escape LIKE/ILIKE metacharacters so ``term`` matches literally. + + A search box takes a substring, not a pattern. Interpolated raw, ``%`` + matches everything and ``_`` matches any single character, so searching + for ``100%`` or ``q1_report`` silently returns the wrong rows. + + Callers must pair this with ``ESCAPE '\\'`` on the comparison. + + Args: + term: The user-supplied substring. + + Returns: + ``term`` with ``\\``, ``%`` and ``_`` backslash-escaped. + """ + return term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") diff --git a/docsgpt/storage/db/models.py b/docsgpt/storage/db/models.py index 1582e32d..79773b6e 100644 --- a/docsgpt/storage/db/models.py +++ b/docsgpt/storage/db/models.py @@ -68,6 +68,11 @@ auth_events_table = Table( metadata, Column("id", UUID(as_uuid=True), primary_key=True, server_default=func.gen_random_uuid()), Column("user_id", Text, nullable=False), + # Who performed the action, and (when the action is about a user) whom it + # was performed on. ``user_id`` predates both and is kept as the per-user + # feed key; see migration 0034. + Column("actor_id", Text, nullable=False), + Column("target_id", Text), Column("event", Text, nullable=False), Column("ip", Text), Column("user_agent", Text), @@ -245,6 +250,13 @@ token_usage_table = Table( # Added in ``0033_quotas``. USD cost of the call at write time; 0 for # unpriced and bring-your-own models. Column("cost", Numeric(12, 8), nullable=False, server_default="0"), + # Added in ``0035_token_usage_latency``. Time spent waiting on the + # provider (a stream excludes consumer backpressure; see ``usage.py``), + # and time to the first streamed chunk. ``ttft_ms`` is NULL for + # non-streaming calls and for streams that failed before yielding -- + # "no first token", not 0. + Column("duration_ms", Integer), + Column("ttft_ms", Integer), ) user_logs_table = Table( diff --git a/docsgpt/storage/db/repositories/activity.py b/docsgpt/storage/db/repositories/activity.py new file mode 100644 index 00000000..4c62bf65 --- /dev/null +++ b/docsgpt/storage/db/repositories/activity.py @@ -0,0 +1,371 @@ +"""One merged, normalized feed over the instance's three audit journals. + +DocsGPT keeps three append-only journals, each written by a different part of +the system and each previously readable only on its own terms: + +* ``auth_events`` — identity, access, configuration and (since the data-plane + hooks) resource mutations. +* ``device_audit_log`` — every remote-device command dispatch and its outcome. +* ``guardrail_events`` — every guardrail decision on a request. + +An operator reviewing an incident wants one timeline, not three. This +repository projects all three onto a common row shape and merges them with +``UNION ALL``, so a single ordered, filtered, paginated feed spans the lot. + +Two exclusions are deliberate and mirror the per-agent guardrail view: the +``guardrail_events.api_key`` column is a raw agent key, and ``matched_value`` +is unredacted source text. Neither is projected here, admin-gated or not. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Iterator, Optional, Sequence + +from sqlalchemy import Connection, text + +from docsgpt.audit_events import category_case_sql +from docsgpt.storage.db.base_repository import like_escape + + +# Columns every branch of the union produces, in order. +ACTIVITY_COLUMNS = ( + "feed", + "id", + "event", + "category", + "actor_id", + "target_id", + "ip", + "user_agent", + "outcome", + "detail", + "created_at", +) + +# Which journal each feed name reads from. Used to skip a branch entirely when +# the category filter cannot match it. +_FEED_CATEGORIES = {"auth": None, "device": "device", "guardrail": "safety"} + +# ``feed`` and ``id`` break ties so a page boundary is deterministic and the +# keyset cursor below has something total to compare against. +_ORDER_BY = "ORDER BY created_at DESC, feed DESC, id DESC" + +_AUTH_BRANCH = f""" + SELECT 'auth' AS feed, + id::text AS id, + event, + {category_case_sql("event")} AS category, + actor_id, + target_id, + ip, + user_agent, + NULL::text AS outcome, + metadata AS detail, + created_at + FROM auth_events +""" + +_DEVICE_BRANCH = """ + SELECT 'device' AS feed, + id::text AS id, + 'device.' || action AS event, + 'device' AS category, + user_id AS actor_id, + NULL::text AS target_id, + NULL::text AS ip, + NULL::text AS user_agent, + decision AS outcome, + jsonb_strip_nulls(jsonb_build_object( + 'device_id', device_id, + 'invocation_id', invocation_id, + 'command', command, + 'working_dir', working_dir, + 'approval_mode', approval_mode, + 'decision_reason', decision_reason, + 'exit_code', exit_code, + 'duration_ms', duration_ms, + 'agent_id', agent_id, + 'conversation_id', conversation_id + )) AS detail, + created_at + FROM device_audit_log +""" + +_GUARDRAIL_BRANCH = """ + SELECT 'guardrail' AS feed, + g.id::text AS id, + 'guardrail.' || g.stage AS event, + 'safety' AS category, + g.user_id AS actor_id, + NULL::text AS target_id, + NULL::text AS ip, + NULL::text AS user_agent, + g.outcome, + jsonb_strip_nulls(jsonb_build_object( + 'check_name', g.check_name, + 'detector_type', g.detector_type, + 'action', g.action, + 'category', g.category, + 'score', g.score, + 'match_count', g.match_count, + 'agent_id', g.agent_id, + 'message_id', g.message_id, + 'request_id', g.request_id, + 'detail', g.detail + )) AS detail, + g.created_at + FROM guardrail_events g +""" + +_BRANCHES = { + "auth": _AUTH_BRANCH, + "device": _DEVICE_BRANCH, + "guardrail": _GUARDRAIL_BRANCH, +} + + +class ActivityRepository: + """Read-only merged feed over ``auth_events`` + the two side journals.""" + + def __init__(self, conn: Connection) -> None: + self._conn = conn + + @staticmethod + def _feeds_for( + feeds: Optional[Sequence[str]], categories: Optional[Sequence[str]] + ) -> list[str]: + """Which union branches can still produce a matching row. + + A ``category=device`` filter cannot be satisfied by ``auth_events``, so + that branch is dropped from the union rather than scanned and + discarded. ``auth`` spans several categories and is only dropped when + the filter names none of them. + """ + selected = list(feeds) if feeds else list(_BRANCHES) + selected = [feed for feed in selected if feed in _BRANCHES] + if not categories: + return selected + wanted = set(categories) + fixed_only = {category for category in _FEED_CATEGORIES.values() if category} + keep = [] + for feed in selected: + fixed = _FEED_CATEGORIES[feed] + if fixed is None: + # ``auth`` spans several categories, so keep it unless the + # filter names only categories another journal owns outright. + if wanted - fixed_only: + keep.append(feed) + elif fixed in wanted: + keep.append(feed) + return keep + + def _query( + self, + *, + feeds: Optional[Sequence[str]] = None, + categories: Optional[Sequence[str]] = None, + events: Optional[Sequence[str]] = None, + actor_id: Optional[str] = None, + user_id: Optional[str] = None, + since: Optional[datetime] = None, + until: Optional[datetime] = None, + search: Optional[str] = None, + ) -> tuple[str, dict]: + selected = self._feeds_for(feeds, categories) + if not selected: + return "", {} + union = " UNION ALL ".join(_BRANCHES[feed] for feed in selected) + + clauses: list[str] = [] + params: dict[str, Any] = {} + if categories: + clauses.append("category = ANY(:categories)") + params["categories"] = list(categories) + if events: + clauses.append("event = ANY(:events)") + params["events"] = list(events) + if actor_id: + clauses.append("actor_id = :actor_id") + params["actor_id"] = actor_id + if user_id: + clauses.append("(actor_id = :user_id OR target_id = :user_id)") + params["user_id"] = user_id + if since is not None: + clauses.append("created_at >= :since") + params["since"] = since + if until is not None: + clauses.append("created_at <= :until") + params["until"] = until + if search: + # ``search`` is a literal substring, not a pattern: an operator + # looking for "100%" or "q1_report" must not have those characters + # read as wildcards. + clauses.append( + "(actor_id ILIKE :search ESCAPE '\\' " + "OR target_id ILIKE :search ESCAPE '\\' " + "OR event ILIKE :search ESCAPE '\\' " + "OR ip ILIKE :search ESCAPE '\\' " + "OR COALESCE(outcome, '') ILIKE :search ESCAPE '\\' " + "OR detail::text ILIKE :search ESCAPE '\\')" + ) + params["search"] = f"%{like_escape(search)}%" + # Always a WHERE, even with no filters, so the keyset cursor below can + # append " AND ..." without sniffing the string it was handed. + where = "WHERE " + " AND ".join(clauses or ["TRUE"]) + return f"SELECT * FROM ({union}) AS activity {where}", params + + def list( + self, + *, + feeds: Optional[Sequence[str]] = None, + categories: Optional[Sequence[str]] = None, + events: Optional[Sequence[str]] = None, + actor_id: Optional[str] = None, + user_id: Optional[str] = None, + since: Optional[datetime] = None, + until: Optional[datetime] = None, + search: Optional[str] = None, + limit: int = 50, + offset: int = 0, + ) -> list[dict]: + """Return one page of the merged feed, newest first. + + Args: + feeds: Restrict to named journals (``auth``, ``device``, ``guardrail``). + categories: Restrict to activity categories. + events: Restrict to exact event names. + actor_id: Only rows this user performed. + user_id: Rows where this user is the actor or the target. + since: Inclusive lower bound on ``created_at``. + until: Inclusive upper bound on ``created_at``. + search: Case-insensitive substring over ids, event, IP and detail. + limit: Page size. + offset: Rows to skip. + + Returns: + Row dicts carrying :data:`ACTIVITY_COLUMNS`. + """ + sql, params = self._query( + feeds=feeds, + categories=categories, + events=events, + actor_id=actor_id, + user_id=user_id, + since=since, + until=until, + search=search, + ) + if not sql: + return [] + params.update({"limit": int(limit), "offset": int(offset)}) + result = self._conn.execute( + text(f"{sql} {_ORDER_BY} LIMIT :limit OFFSET :offset"), + params, + ) + return [dict(row._mapping) for row in result.fetchall()] + + def _page_after( + self, + cursor: Optional[tuple[datetime, str, str]], + limit: int, + filters: dict, + ) -> list[dict]: + """One keyset page strictly older than ``cursor``. + + OFFSET paging would drift here: the feed is ordered newest-first and + the journals are append-only, so rows inserted mid-export push the + window down and repeat rows that were already emitted. + """ + sql, params = self._query(**filters) + if not sql: + return [] + if cursor is not None: + sql += " AND (created_at, feed, id) < (:cur_at, :cur_feed, :cur_id)" + params.update( + {"cur_at": cursor[0], "cur_feed": cursor[1], "cur_id": cursor[2]} + ) + params["limit"] = int(limit) + result = self._conn.execute(text(f"{sql} {_ORDER_BY} LIMIT :limit"), params) + return [dict(row._mapping) for row in result.fetchall()] + + def count( + self, + *, + feeds: Optional[Sequence[str]] = None, + categories: Optional[Sequence[str]] = None, + events: Optional[Sequence[str]] = None, + actor_id: Optional[str] = None, + user_id: Optional[str] = None, + since: Optional[datetime] = None, + until: Optional[datetime] = None, + search: Optional[str] = None, + ) -> int: + """Total rows matching the same filters as :meth:`list`.""" + sql, params = self._query( + feeds=feeds, + categories=categories, + events=events, + actor_id=actor_id, + user_id=user_id, + since=since, + until=until, + search=search, + ) + if not sql: + return 0 + scalar = self._conn.execute( + text(f"SELECT count(*) FROM ({sql}) AS matched"), params + ).scalar() + return int(scalar or 0) + + def iter_all( + self, + *, + chunk_size: int = 1000, + max_rows: int = 100_000, + **filters: Any, + ) -> Iterator[dict]: + """Yield every matching row, newest first, for the export endpoint. + + Pages by keyset, not offset: the journals are append-only and the feed + is newest-first, so rows written mid-export would otherwise shift the + window and repeat rows already emitted. + + Args: + chunk_size: Rows fetched per round trip. + max_rows: Hard ceiling; an export stops here rather than running + unbounded against a busy instance. + **filters: As :meth:`list`, minus ``limit`` / ``offset``. + + Yields: + Row dicts carrying :data:`ACTIVITY_COLUMNS`. + """ + emitted = 0 + cursor: Optional[tuple[datetime, str, str]] = None + while emitted < max_rows: + page = self._page_after( + cursor, min(chunk_size, max_rows - emitted), filters + ) + if not page: + return + for row in page: + yield row + last = page[-1] + cursor = (last["created_at"], last["feed"], last["id"]) + emitted += len(page) + + def event_names(self) -> list[dict]: + """Distinct ``(event, category)`` pairs across all three journals. + + Feeds the filter's event picker, so operators choose from what the + instance actually recorded rather than typing a name from memory. + """ + union = " UNION ALL ".join(_BRANCHES.values()) + result = self._conn.execute( + text( + f"SELECT DISTINCT event, category FROM ({union}) AS activity " + "ORDER BY category, event" + ) + ) + return [{"event": row[0], "category": row[1]} for row in result.fetchall()] diff --git a/docsgpt/storage/db/repositories/admin_stats.py b/docsgpt/storage/db/repositories/admin_stats.py index cec29e54..b4c54bee 100644 --- a/docsgpt/storage/db/repositories/admin_stats.py +++ b/docsgpt/storage/db/repositories/admin_stats.py @@ -14,6 +14,12 @@ from typing import Optional from sqlalchemy import Connection, text from docsgpt.storage.db.base_repository import row_to_dict +from docsgpt.storage.db.repositories.token_usage import TokenUsageRepository + + +def _round_ms(value: Optional[float]) -> Optional[int]: + """Round a percentile to whole milliseconds, preserving NULL as None.""" + return int(round(float(value))) if value is not None else None class AdminStatsRepository: @@ -66,21 +72,146 @@ class AdminStatsRepository: } def top_token_users(self, *, since: datetime, limit: int = 10) -> list[dict]: - """Highest token consumers since ``since`` (admin usage view).""" + """Highest consumers since ``since``, with spend (admin usage view). + + Ordered by tokens, not cost: a cheap model can dominate token volume + while a costly one dominates the bill, and the operator wants to see + both columns rather than have the ranking pick for them. + + Run-level rollup rows are excluded, matching every other spend query: + a scheduled run already has a row per LLM call, so counting the rollup + too would bill it twice. + """ result = self._conn.execute( text( """ - SELECT user_id, SUM(prompt_tokens + generated_tokens) AS tokens + SELECT user_id, + SUM(prompt_tokens + generated_tokens) AS tokens, + COALESCE(SUM(cost), 0) AS cost FROM token_usage WHERE timestamp >= :since AND user_id IS NOT NULL + AND source <> ALL(:rollup_sources) GROUP BY user_id ORDER BY tokens DESC LIMIT :limit """ ), - {"since": since, "limit": int(limit)}, + { + "since": since, + "limit": int(limit), + "rollup_sources": list(TokenUsageRepository.ROLLUP_SOURCES), + }, ) - return [{"user_id": r[0], "tokens": int(r[1])} for r in result.fetchall()] + return [ + {"user_id": r[0], "tokens": int(r[1]), "cost": float(r[2])} + for r in result.fetchall() + ] + + def latency_summary(self, *, since: datetime) -> dict: + """Call-latency percentiles since ``since``. + + Rows written before the latency columns existed, and streams that + never yielded, hold NULL; percentile_cont skips them, so the summary + describes the calls that were actually measured. ``samples`` says how + many those were, so a percentile from three calls is not read as a + service-level figure. + """ + row = self._conn.execute( + text( + """ + SELECT count(duration_ms) AS samples, + percentile_cont(0.5) WITHIN GROUP (ORDER BY duration_ms) + AS p50_ms, + percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms) + AS p95_ms, + count(ttft_ms) AS ttft_samples, + percentile_cont(0.5) WITHIN GROUP (ORDER BY ttft_ms) + AS ttft_p50_ms + FROM token_usage + WHERE timestamp >= :since + """ + ), + {"since": since}, + ).one() + return { + "samples": int(row.samples or 0), + "p50_ms": _round_ms(row.p50_ms), + "p95_ms": _round_ms(row.p95_ms), + "ttft_samples": int(row.ttft_samples or 0), + "ttft_p50_ms": _round_ms(row.ttft_p50_ms), + } + + def user_usage_breakdown( + self, user_id: str, *, since: datetime, limit: int = 10 + ) -> dict: + """Per-user spend detail: totals, and a split by model and by flow. + + The admin user drill-down showed a single ``tokens_30d`` number, which + cannot answer either question an operator actually has -- what is this + person costing, and what is driving it. + + Run-level rollup rows are excluded, so a scheduled run is counted once + (by its per-call rows), not twice. + + Args: + user_id: The billable user (auth ``sub``). + since: Inclusive window start. + limit: How many models / flows to return, busiest first. + + Returns: + ``{"totals": {...}, "by_model": [...], "by_source": [...]}``. + """ + rollups = list(TokenUsageRepository.ROLLUP_SOURCES) + totals = self._conn.execute( + text( + """ + SELECT COALESCE(SUM(prompt_tokens + generated_tokens), 0) AS tokens, + COALESCE(SUM(cost), 0) AS cost, + count(*) AS calls + FROM token_usage + WHERE user_id = :u AND timestamp >= :since + AND source <> ALL(:rollup_sources) + """ + ), + {"u": user_id, "since": since, "rollup_sources": rollups}, + ).one() + + def _breakdown(key_expr: str) -> list[dict]: + rows = self._conn.execute( + text( + f""" + SELECT {key_expr} AS key, + COALESCE(SUM(prompt_tokens + generated_tokens), 0) AS tokens, + COALESCE(SUM(cost), 0) AS cost + FROM token_usage + WHERE user_id = :u AND timestamp >= :since + AND source <> ALL(:rollup_sources) + GROUP BY key + ORDER BY tokens DESC, key + LIMIT :limit + """ + ), + { + "u": user_id, + "since": since, + "limit": int(limit), + "rollup_sources": rollups, + }, + ) + return [ + {"key": r.key, "tokens": int(r.tokens), "cost": float(r.cost)} + for r in rows.fetchall() + ] + + return { + "totals": { + "tokens": int(totals.tokens), + "cost": float(totals.cost), + "calls": int(totals.calls), + }, + "by_model": _breakdown("COALESCE(model_id, 'unknown')"), + "by_source": _breakdown("COALESCE(source, 'agent_stream')"), + } def list_users( self, user_id_filter: Optional[str], offset: int, limit: int diff --git a/docsgpt/storage/db/repositories/auth_events.py b/docsgpt/storage/db/repositories/auth_events.py index 834b31e7..04f6172a 100644 --- a/docsgpt/storage/db/repositories/auth_events.py +++ b/docsgpt/storage/db/repositories/auth_events.py @@ -3,15 +3,33 @@ from __future__ import annotations import json -from typing import Optional +from typing import Any, Optional, Sequence from sqlalchemy import Connection, text from docsgpt.storage.db.base_repository import row_to_dict +# Actors that are not a person. Used where an automated system performs the +# action on a user's behalf, so the feed never reads as if the user did it. +SYSTEM_ACTOR_SCIM = "system:scim" +SYSTEM_ACTOR_OIDC = "system:oidc" + + class AuthEventsRepository: - """Append-only audit trail of login / logout / provisioning events.""" + """Append-only audit trail of identity, access and data-plane events. + + Every row carries three attribution columns: + + * ``actor_id`` — who performed the action. Never NULL. + * ``target_id`` — the user the action was performed on, or NULL when the + event is not about a user (a team change, an instance-wide policy). + * ``user_id`` — the legacy subject column, kept as the key for the + per-user feed (:meth:`list_recent`). Defaults to the target, falling + back to the actor for actor-only events. + + See migration ``0034_auth_events_actor_target``. + """ def __init__(self, conn: Connection) -> None: self._conn = conn @@ -23,18 +41,45 @@ class AuthEventsRepository: ip: Optional[str] = None, user_agent: Optional[str] = None, metadata: Optional[dict] = None, + *, + actor_id: Optional[str] = None, + target_id: Optional[str] = "", ) -> dict: - """Record one auth event and return the inserted row.""" + """Record one audit event and return the inserted row. + + Args: + user_id: The row's subject, kept for the per-user feed. + event: Event name (``oidc_login``, ``source.deleted``, ...). + ip: Client IP, when the event came from a request. + user_agent: Client ``User-Agent``, when the event came from a request. + metadata: Free-form JSON detail shown in the audit drill-down. + actor_id: Who performed the action. Defaults to ``user_id``, which + is correct for self-service events (login, PAT create). + target_id: The user acted upon. Defaults to ``user_id``; pass + ``None`` explicitly for events with no user target. + + Returns: + The inserted row as a dict. + """ + # Sentinel default: ``None`` is a meaningful value here ("no target"), + # so it cannot double as "caller did not say". + resolved_target = user_id if target_id == "" else target_id + resolved_actor = actor_id or user_id result = self._conn.execute( text( """ - INSERT INTO auth_events (user_id, event, ip, user_agent, metadata) - VALUES (:user_id, :event, :ip, :user_agent, CAST(:metadata AS jsonb)) + INSERT INTO auth_events + (user_id, actor_id, target_id, event, ip, user_agent, metadata) + VALUES + (:user_id, :actor_id, :target_id, :event, :ip, :user_agent, + CAST(:metadata AS jsonb)) RETURNING * """ ), { "user_id": user_id, + "actor_id": resolved_actor, + "target_id": resolved_target, "event": event, "ip": ip, "user_agent": user_agent, @@ -43,30 +88,60 @@ class AuthEventsRepository: ) return row_to_dict(result.fetchone()) - def list_recent(self, user_id: str, limit: int = 50) -> list[dict]: - """Return the newest events for ``user_id``, newest first.""" + def list_recent( + self, + user_id: str, + limit: int = 50, + *, + exclude_events_like: Optional[Sequence[str]] = None, + ) -> list[dict]: + """Return the newest events for ``user_id``, newest first. + + Args: + user_id: The subject whose trail to read. + limit: How many rows to return. + exclude_events_like: Event-name prefixes to drop, e.g. + ``("source.", "agent.")``. Data-plane events are filed under + the actor, so on an active account they would otherwise push + a denied login or a role grant out of a short window within + minutes. + + Returns: + Matching rows, newest first. + """ + clauses = ["user_id = :user_id"] + params: dict = {"user_id": user_id, "limit": limit} + for index, prefix in enumerate(exclude_events_like or ()): + key = f"excl_{index}" + clauses.append(f"event NOT LIKE :{key}") + params[key] = f"{prefix}%" + where = " AND ".join(clauses) result = self._conn.execute( text( - """ + f""" SELECT * FROM auth_events - WHERE user_id = :user_id + WHERE {where} ORDER BY created_at DESC LIMIT :limit """ ), - {"user_id": user_id, "limit": limit}, + params, ) return [row_to_dict(row) for row in result.fetchall()] @staticmethod - def _filter_clauses(event, user_id, since) -> tuple[str, dict]: + def _filter_clauses( + event: Optional[str], + user_id: Optional[str], + since: Any, + ) -> tuple[str, dict]: clauses: list[str] = [] params: dict = {} if event: clauses.append("event = :event") params["event"] = event if user_id: - clauses.append("user_id = :user_id") + clauses.append("(user_id = :user_id OR target_id = :user_id)") params["user_id"] = user_id if since is not None: clauses.append("created_at >= :since") @@ -83,7 +158,12 @@ class AuthEventsRepository: limit: int = 50, offset: int = 0, ) -> list[dict]: - """Global audit feed (admin), newest first; optional event/user/since filters.""" + """Single-journal audit feed for ``GET /api/admin/audit``, newest first. + + The admin UI reads the merged feed instead; richer filtering lives on + :class:`~docsgpt.storage.db.repositories.activity.ActivityRepository` + rather than being implemented twice here. + """ where, params = self._filter_clauses(event, user_id, since) params.update({"limit": int(limit), "offset": int(offset)}) result = self._conn.execute( @@ -96,7 +176,11 @@ class AuthEventsRepository: return [row_to_dict(row) for row in result.fetchall()] def count_all( - self, *, event: Optional[str] = None, user_id: Optional[str] = None, since=None + self, + *, + event: Optional[str] = None, + user_id: Optional[str] = None, + since=None, ) -> int: """Total matching the same filters as :meth:`list_all` (for pagination).""" where, params = self._filter_clauses(event, user_id, since) diff --git a/docsgpt/storage/db/repositories/token_usage.py b/docsgpt/storage/db/repositories/token_usage.py index fb256b55..6247bd9e 100644 --- a/docsgpt/storage/db/repositories/token_usage.py +++ b/docsgpt/storage/db/repositories/token_usage.py @@ -38,6 +38,8 @@ class TokenUsageRepository: cached_tokens: Optional[int] = None, cache_write_tokens: Optional[int] = None, cost: float = 0.0, + duration_ms: Optional[int] = None, + ttft_ms: Optional[int] = None, ) -> None: # Attribution guard: the ``token_usage_attribution_chk`` CHECK # constraint requires at least one of ``user_id`` / ``api_key`` @@ -64,6 +66,7 @@ class TokenUsageRepository: user_id, api_key, agent_id, prompt_tokens, generated_tokens, cached_tokens, cache_write_tokens, cost, + duration_ms, ttft_ms, source, request_id, model_id, timestamp ) VALUES ( @@ -71,6 +74,7 @@ class TokenUsageRepository: CAST(:agent_id AS uuid), :prompt_tokens, :generated_tokens, :cached_tokens, :cache_write_tokens, :cost, + :duration_ms, :ttft_ms, :source, :request_id, :model_id, COALESCE(:timestamp, now()) ) """ @@ -84,6 +88,8 @@ class TokenUsageRepository: "cached_tokens": cached_tokens, "cache_write_tokens": cache_write_tokens, "cost": cost, + "duration_ms": duration_ms, + "ttft_ms": ttft_ms, "source": source, "request_id": request_id, "model_id": model_id, @@ -296,7 +302,22 @@ class TokenUsageRepository: f""" SELECT to_char(tu.timestamp AT TIME ZONE 'UTC', :fmt) AS bucket, COALESCE(SUM(tu.prompt_tokens), 0) AS prompt_tokens, - COALESCE(SUM(tu.generated_tokens), 0) AS generated_tokens + COALESCE(SUM(tu.generated_tokens), 0) AS generated_tokens, + COALESCE(SUM(tu.cost), 0) AS cost, + -- NULL cache bins mean "provider did not report", so + -- they stay out of the sum rather than reading as 0. + SUM(tu.cached_tokens) AS cached_tokens, + -- The matching denominator: prompt tokens from the + -- rows that reported. Dividing ``cached_tokens`` by + -- ``prompt_tokens`` instead would understate the hit + -- rate by however much traffic ran on a provider that + -- reports nothing, since those rows are in one sum but + -- not the other. + COALESCE( + SUM(tu.prompt_tokens) + FILTER (WHERE tu.cached_tokens IS NOT NULL), + 0 + ) AS cache_eligible_prompt_tokens {group_select} FROM token_usage tu {join} @@ -312,6 +333,15 @@ class TokenUsageRepository: "bucket": row._mapping["bucket"], "prompt_tokens": int(row._mapping["prompt_tokens"]), "generated_tokens": int(row._mapping["generated_tokens"]), + "cost": float(row._mapping["cost"]), + "cached_tokens": ( + int(row._mapping["cached_tokens"]) + if row._mapping["cached_tokens"] is not None + else None + ), + "cache_eligible_prompt_tokens": int( + row._mapping["cache_eligible_prompt_tokens"] + ), **( {"group_key": row._mapping["group_key"]} if group_by is not None diff --git a/docsgpt/usage.py b/docsgpt/usage.py index 82fc3eb8..8ac6c436 100644 --- a/docsgpt/usage.py +++ b/docsgpt/usage.py @@ -93,7 +93,7 @@ def _count_prompt_tokens(messages, tools=None, usage_attachments=None, **kwargs) return prompt_tokens -def _persist_call_usage(llm, call_usage): +def _persist_call_usage(llm, call_usage, *, duration_ms=None, ttft_ms=None): """Write one ``token_usage`` row per LLM call. Always-on; no flag. Source defaults to ``agent_stream`` and can be overridden per @@ -101,6 +101,13 @@ def _persist_call_usage(llm, call_usage): title / compression / rag_condense / fallback). A ``_request_id`` stamped on the LLM lets ``count_in_range`` deduplicate the multiple rows produced by a single multi-tool agent run. + + Args: + llm: The LLM instance the call ran on. + call_usage: The call's token counts. + duration_ms: Wall-clock for the call, measured by the wrapper. + ttft_ms: Time to the first streamed chunk; None for a non-streaming + call and for a stream that failed before yielding anything. """ if call_usage["prompt_tokens"] == 0 and call_usage["generated_tokens"] == 0: return @@ -149,6 +156,8 @@ def _persist_call_usage(llm, call_usage): ), request_id=getattr(llm, "_request_id", None), model_id=model_id, + duration_ms=duration_ms, + ttft_ms=ttft_ms, ) except Exception: logger.exception("token_usage persist failed") @@ -258,10 +267,12 @@ def gen_token_usage(func): error = exc raise finally: + duration_ms = int((time.monotonic() - started_at) * 1000) call_usage = _prefer_provider_usage(self, call_usage) self.token_usage["prompt_tokens"] += call_usage["prompt_tokens"] self.token_usage["generated_tokens"] += call_usage["generated_tokens"] - _persist_call_usage(self, call_usage) + # A non-streaming call has no first-token moment; ttft stays NULL. + _persist_call_usage(self, call_usage, duration_ms=duration_ms) emit = getattr(self, "_emit_gen_finished_log", None) if callable(emit): try: @@ -269,7 +280,7 @@ def gen_token_usage(func): model, prompt_tokens=call_usage["prompt_tokens"], completion_tokens=call_usage["generated_tokens"], - latency_ms=int((time.monotonic() - started_at) * 1000), + latency_ms=duration_ms, cached_tokens=call_usage.get("cached_tokens"), cache_write_tokens=call_usage.get("cache_write_tokens"), error=error, @@ -293,10 +304,29 @@ def stream_token_usage(func): ) batch = [] started_at = time.monotonic() + first_chunk_at: float | None = None + # Time spent waiting on the provider, accumulated across ``next()`` + # calls. The wall clock cannot be used here: this is a generator, so + # every ``yield`` suspends until the consumer comes back, and the span + # from start to exhaustion includes the agent loop's tool handling and + # the SSE client's backpressure. A slow browser would otherwise record + # 30s for a 900ms call, and latency_summary would mix that with true + # non-streaming durations under one p50. + provider_seconds = 0.0 error: BaseException | None = None try: result = func(self, model, messages, stream, tools, **kwargs) - for r in result: + stream_iter = iter(result) + while True: + pull_started = time.monotonic() + try: + r = next(stream_iter) + except StopIteration: + provider_seconds += time.monotonic() - pull_started + break + provider_seconds += time.monotonic() - pull_started + if first_chunk_at is None: + first_chunk_at = pull_started + provider_seconds batch.append(r) yield r except Exception as exc: @@ -306,12 +336,22 @@ def stream_token_usage(func): error = exc raise finally: + duration_ms = int(provider_seconds * 1000) + # NULL, not 0, when the stream failed before yielding: "no first + # token" must not read as an instant one in a p50. + ttft_ms = ( + int((first_chunk_at - started_at) * 1000) + if first_chunk_at is not None + else None + ) for line in batch: call_usage["generated_tokens"] += _count_tokens(line) call_usage = _prefer_provider_usage(self, call_usage) self.token_usage["prompt_tokens"] += call_usage["prompt_tokens"] self.token_usage["generated_tokens"] += call_usage["generated_tokens"] - _persist_call_usage(self, call_usage) + _persist_call_usage( + self, call_usage, duration_ms=duration_ms, ttft_ms=ttft_ms + ) emit = getattr(self, "_emit_stream_finished_log", None) if callable(emit): try: @@ -319,6 +359,9 @@ def stream_token_usage(func): model, prompt_tokens=call_usage["prompt_tokens"], completion_tokens=call_usage["generated_tokens"], + # The log line has always meant end-to-end wall clock + # for the streamed response; only the persisted column + # isolates provider time. latency_ms=int((time.monotonic() - started_at) * 1000), cached_tokens=call_usage.get("cached_tokens"), cache_write_tokens=call_usage.get("cache_write_tokens"), diff --git a/frontend/src/admin/Activity.tsx b/frontend/src/admin/Activity.tsx new file mode 100644 index 00000000..c5494635 --- /dev/null +++ b/frontend/src/admin/Activity.tsx @@ -0,0 +1,459 @@ +import { ChevronDown, ChevronRight, Download } from 'lucide-react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { useSelector } from 'react-redux'; +import { useSearchParams } from 'react-router-dom'; + +import adminService, { + type ActivityFilters, +} from '../api/services/adminService'; +import { Button } from '../components/ui/button'; +import { Input } from '../components/ui/input'; +import { MultiSelect } from '../components/ui/multi-select'; +import { + Table, + TableBody, + TableCell, + TableContainer, + TableHead, + TableHeader, + TableRow, +} from '../components/ui/table'; +import { selectToken } from '../preferences/preferenceSlice'; +import { + Loading, + Pill, + categoryTone, + eventLabel, + eventTone, + fmtDate, + fmtNumber, + fmtRelative, + isLoopback, + outcomeLabel, + outcomeTone, +} from './AdminUI'; + +type ActivityRow = { + feed: string; + id: string; + event: string; + category: string; + actor_id: string | null; + target_id: string | null; + ip: string | null; + user_agent: string | null; + outcome: string | null; + detail: Record | null; + created_at: string | null; +}; + +type Catalogue = { + events: { event: string; category: string }[]; + categories: string[]; +}; + +const PAGE_SIZE = 50; + +/** Quick ranges, in days. `0` means "no lower bound". */ +const RANGES: { label: string; days: number }[] = [ + { label: '24h', days: 1 }, + { label: '7d', days: 7 }, + { label: '30d', days: 30 }, + { label: '90d', days: 90 }, + { label: 'All', days: 0 }, +]; + +function sinceFor(days: number): string | undefined { + if (days <= 0) return undefined; + return new Date(Date.now() - days * 86400000).toISOString(); +} + +/** One detail value, rendered as a readable line rather than raw JSON. */ +function DetailRow({ label, value }: { label: string; value: unknown }) { + const text = + typeof value === 'object' && value !== null + ? JSON.stringify(value) + : String(value); + return ( +
+ {label} + {text} +
+ ); +} + +/** + * The expanded row. auth_events captures `metadata` and `user_agent` on every + * insert and the table showed neither, so "Admin revoked · user-42" never said + * who revoked it. Everything the feed carries lands here. + */ +function RowDetail({ row }: { row: ActivityRow }) { + const detail = row.detail ?? {}; + const entries = Object.entries(detail); + return ( +
+ + + {row.actor_id ? : null} + {row.target_id ? ( + + ) : null} + {row.outcome ? : null} + {row.ip ? : null} + {row.user_agent ? ( + + ) : null} + + {entries.length > 0 ? ( +
+ {entries.map(([key, value]) => ( + + ))} +
+ ) : null} +
+ ); +} + +export default function Activity() { + const token = useSelector(selectToken); + const [searchParams] = useSearchParams(); + + // Deep links from Overview land here pre-filtered (?event=oidc_login_denied). + const initialEvent = searchParams.get('event'); + + const [rows, setRows] = useState([]); + const [total, setTotal] = useState(0); + const [page, setPage] = useState(1); + const [loading, setLoading] = useState(true); + const [expanded, setExpanded] = useState(null); + const [exporting, setExporting] = useState(false); + const [catalogue, setCatalogue] = useState(null); + + const [categories, setCategories] = useState([]); + const [events, setEvents] = useState( + initialEvent ? [initialEvent] : [], + ); + const [rangeDays, setRangeDays] = useState(30); + const [searchDraft, setSearchDraft] = useState(''); + const [search, setSearch] = useState(''); + + /** + * Narrowing the feed invalidates whatever page you were on, so every filter + * setter resets it. Resetting in an effect instead would fetch the old page + * against the new filters first, then fetch again. + */ + const withPageReset = + (set: (value: T) => void) => + (value: T) => { + setPage(1); + set(value); + }; + + const filters: ActivityFilters = useMemo( + () => ({ + category: categories.length ? categories : undefined, + event: events.length ? events : undefined, + since: sinceFor(rangeDays), + search: search || undefined, + }), + [categories, events, rangeDays, search], + ); + + useEffect(() => { + let cancelled = false; + adminService + .getActivityEvents(token) + .then((res) => res.json()) + .then((json) => { + if (!cancelled && json?.success) setCatalogue(json); + }) + .catch(() => undefined); + return () => { + cancelled = true; + }; + }, [token]); + + // Monotonic request id: a response only lands if no newer request was + // issued meanwhile, so an out-of-order reply cannot leave the table showing + // a different filter combination than the controls. Mirrors the guard in + // settings/Analytics. + const requestId = useRef(0); + + const load = useCallback(async () => { + const id = ++requestId.current; + setLoading(true); + try { + const res = await adminService.getActivity( + { ...filters, page, page_size: PAGE_SIZE }, + token, + ); + const json = await res.json().catch(() => ({})); + if (id !== requestId.current) return; + setRows(json.activity ?? []); + setTotal(json.total ?? 0); + } finally { + if (id === requestId.current) setLoading(false); + } + }, [token, page, filters]); + + useEffect(() => { + load(); + }, [load]); + + const exportAs = async (format: 'csv' | 'ndjson') => { + setExporting(true); + try { + const res = await adminService.exportActivity(filters, format, token); + if (!res.ok) return; + const blob = await res.blob(); + const url = URL.createObjectURL(blob); + const link = document.createElement('a'); + link.href = url; + link.download = `docsgpt-activity.${format}`; + document.body.appendChild(link); + link.click(); + document.body.removeChild(link); + URL.revokeObjectURL(url); + } finally { + setExporting(false); + } + }; + + const eventOptions = useMemo( + () => + (catalogue?.events ?? []).map(({ event }) => ({ + value: event, + label: eventLabel(event), + })), + [catalogue], + ); + const categoryOptions = useMemo( + () => + (catalogue?.categories ?? []).map((category) => ({ + value: category, + label: category.charAt(0).toUpperCase() + category.slice(1), + })), + [catalogue], + ); + + const totalPages = Math.max(1, Math.ceil(total / PAGE_SIZE)); + const hasFilters = + categories.length > 0 || events.length > 0 || search !== ''; + + const applySearch = () => withPageReset(setSearch)(searchDraft.trim()); + + return ( +
+
+ setSearchDraft(e.target.value)} + onKeyDown={(e) => e.key === 'Enter' && applySearch()} + onBlur={applySearch} + className="max-w-xs" + /> + + +
+ {RANGES.map((range) => ( + + ))} +
+ {hasFilters ? ( + + ) : null} +
+ + +
+
+ + {loading ? ( + + ) : rows.length === 0 ? ( +

No activity.

+ ) : ( + <> + + + + + + Event + Category + Actor + Affected + IP + When + + + + {rows.map((row, idx) => { + const key = `${row.feed}:${row.id}`; + const isOpen = expanded === key; + // De-emphasize an actor repeated from the row above (ditto), + // so distinct actors stand out in a long single-user stream. + const repeat = + idx > 0 && rows[idx - 1].actor_id === row.actor_id; + return [ + + + + + +
+ + {eventLabel(row.event)} + + {row.outcome ? ( + + {outcomeLabel(row.outcome)} + + ) : null} +
+ + + + {row.category} + + + + {repeat ? '〃' : (row.actor_id ?? '—')} + + + {row.target_id && row.target_id !== row.actor_id + ? row.target_id + : '—'} + + + {row.ip === null ? ( + + — + + ) : isLoopback(row.ip) ? ( + + local + + ) : ( + + {row.ip} + + )} + + + + {fmtRelative(row.created_at)} + + + , + isOpen ? ( + + + + + + ) : null, + ]; + })} + +
+ +
+

+ {fmtNumber(total)} events · page {page} of {totalPages} +

+
+ + +
+
+ + )} +
+ ); +} diff --git a/frontend/src/admin/AdminUI.test.ts b/frontend/src/admin/AdminUI.test.ts new file mode 100644 index 00000000..26d74212 --- /dev/null +++ b/frontend/src/admin/AdminUI.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from 'vitest'; + +import { + categoryTone, + eventLabel, + eventTone, + outcomeLabel, + outcomeTone, +} from './AdminUI'; + +describe('eventLabel', () => { + it('humanizes the events it knows', () => { + expect(eventLabel('oidc_login_denied')).toBe('Login denied'); + expect(eventLabel('source.deleted')).toBe('Source deleted'); + expect(eventLabel('device.run_command')).toBe('Device command'); + }); + + it('stays legible for an event a newer release added', () => { + expect(eventLabel('source.archived')).toBe('Source archived'); + expect(eventLabel('brand_new_event')).toBe('Brand new event'); + }); +}); + +describe('eventTone', () => { + it('flags removals and refusals', () => { + expect(eventTone('oidc_login_denied')).toBe('danger'); + expect(eventTone('agent.deleted')).toBe('danger'); + }); + + it('flags revocations as a warning, not a failure', () => { + expect(eventTone('role_revoked')).toBe('warning'); + expect(eventTone('agent.key_regenerated')).toBe('warning'); + }); + + it('marks an admin grant with the brand tone', () => { + expect(eventTone('role_granted')).toBe('brand'); + }); + + it('leaves routine events muted', () => { + expect(eventTone('oidc_login')).toBe('muted'); + expect(eventTone('anything_else')).toBe('muted'); + }); +}); + +describe('categoryTone', () => { + it('has a tone for every category the API declares', () => { + for (const category of [ + 'identity', + 'access', + 'config', + 'data', + 'device', + 'safety', + 'other', + ]) { + expect(categoryTone(category)).toBeTruthy(); + } + }); + + it('falls back for an unknown category', () => { + expect(categoryTone('invented')).toBe('muted'); + }); +}); + +describe('outcomeTone', () => { + // The values below are the only ones the writers produce: guardrails emit + // triggered/not_evaluated, the device feed emits dispatched. + it('flags a guardrail that fired', () => { + expect(outcomeTone('triggered')).toBe('danger'); + }); + + it('leaves a check that never ran neutral', () => { + expect(outcomeTone('not_evaluated')).toBe('muted'); + }); + + it('reads a dispatched device command as a success', () => { + expect(outcomeTone('dispatched')).toBe('success'); + }); + + it('leaves an unfamiliar outcome muted', () => { + expect(outcomeTone('something-new')).toBe('muted'); + }); +}); + +describe('outcomeLabel', () => { + it('humanizes the values the journals write', () => { + expect(outcomeLabel('not_evaluated')).toBe('Not evaluated'); + expect(outcomeLabel('dispatched')).toBe('Dispatched'); + }); + + it('stays legible for an unknown value', () => { + expect(outcomeLabel('rate_limited')).toBe('Rate limited'); + }); +}); diff --git a/frontend/src/admin/AdminUI.tsx b/frontend/src/admin/AdminUI.tsx index 6bc5caeb..250a7410 100644 --- a/frontend/src/admin/AdminUI.tsx +++ b/frontend/src/admin/AdminUI.tsx @@ -97,6 +97,28 @@ export function fmtNumber(n?: number | null): string { return new Intl.NumberFormat().format(n ?? 0); } +/** + * USD, with enough precision that a fraction of a cent is not rendered as $0. + * An instance on a cheap model can run whole days under a dollar, and "$0" + * next to a cost quota reads as "nothing is being counted". + */ +export function fmtUsd(n?: number | null): string { + const value = n ?? 0; + const digits = value !== 0 && Math.abs(value) < 0.01 ? 4 : 2; + return new Intl.NumberFormat(undefined, { + style: 'currency', + currency: 'USD', + minimumFractionDigits: 2, + maximumFractionDigits: digits, + }).format(value); +} + +/** Milliseconds as ms or s, or an em dash when nothing was measured. */ +export function fmtMs(n?: number | null): string { + if (n === null || n === undefined) return '—'; + return n < 1000 ? `${Math.round(n)}ms` : `${(n / 1000).toFixed(1)}s`; +} + export function fmtCompact(n?: number | null): string { return new Intl.NumberFormat(undefined, { notation: 'compact', @@ -104,8 +126,10 @@ export function fmtCompact(n?: number | null): string { }).format(n ?? 0); } -// Humanized auth-event labels + semantic color tier, shared by the Audit feed +// Humanized event labels + semantic color tier, shared by the activity feed // and the user-detail modal so the same event reads identically everywhere. +// Unknown names fall back to a title-cased form of the name itself, so an +// event added by a newer release is still legible here. const EVENT_LABELS: Record = { oidc_login: 'Login', oidc_login_denied: 'Login denied', @@ -121,28 +145,115 @@ const EVENT_LABELS: Record = { scim_reactivated: 'Activated (SCIM)', quota_policy_set: 'Quota set', quota_policy_deleted: 'Quota removed', + pat_created: 'Token created', + pat_revoked: 'Token revoked', + pat_regenerated: 'Token regenerated', + 'team.create': 'Team created', + 'team.delete': 'Team deleted', + 'team.member_add': 'Member added', + 'team.member_role': 'Member role changed', + 'team.member_remove': 'Member removed', + 'team.share': 'Resource shared', + 'team.unshare': 'Resource unshared', + 'team.transfer_owner': 'Ownership transferred', + 'source.created': 'Source created', + 'source.deleted': 'Source deleted', + 'source.reingested': 'Source reingested', + 'agent.created': 'Agent created', + 'agent.updated': 'Agent updated', + 'agent.deleted': 'Agent deleted', + 'agent.key_regenerated': 'Agent key rotated', + 'conversation.deleted': 'Conversation deleted', + 'conversation.deleted_all': 'All conversations deleted', + 'device.run_command': 'Device command', + 'guardrail.input': 'Guardrail (input)', + 'guardrail.retrieval': 'Guardrail (retrieval)', + 'guardrail.tool_result': 'Guardrail (tool result)', + 'guardrail.output': 'Guardrail (output)', }; export function eventLabel(event: string): string { - return ( - EVENT_LABELS[event] ?? - event.replace(/_/g, ' ').replace(/^\w/, (c) => c.toUpperCase()) - ); + const known = EVENT_LABELS[event]; + if (known) return known; + // "source.reingested" -> "Source reingested" + return event.replace(/[._]/g, ' ').replace(/^\w/, (c) => c.toUpperCase()); } +// Events that describe something being taken away or refused. +const DANGER_EVENTS = new Set([ + 'oidc_login_denied', + 'admin_user_deactivated', + 'scim_deactivated', + 'source.deleted', + 'agent.deleted', + 'conversation.deleted', + 'conversation.deleted_all', + 'team.delete', +]); + +const WARNING_EVENTS = new Set([ + 'role_revoked', + 'admin_sessions_revoked', + 'pat_revoked', + 'agent.key_regenerated', + 'quota_policy_deleted', + 'team.member_remove', + 'team.unshare', +]); + export function eventTone(event: string): Tone { - if ( - event === 'oidc_login_denied' || - event === 'admin_user_deactivated' || - event === 'scim_deactivated' - ) - return 'danger'; + if (DANGER_EVENTS.has(event)) return 'danger'; if (event === 'role_granted') return 'brand'; - if (event === 'role_revoked' || event === 'admin_sessions_revoked') - return 'warning'; + if (WARNING_EVENTS.has(event)) return 'warning'; return 'muted'; } +// Category facet colors. Mirrors docsgpt/audit_events.py. +const CATEGORY_TONES: Record = { + identity: 'default', + access: 'brand', + config: 'warning', + data: 'success', + device: 'warning', + safety: 'danger', + other: 'muted', +}; + +export function categoryTone(category: string): Tone { + return CATEGORY_TONES[category] ?? 'muted'; +} + +// Outcome values the two side journals actually write. +// +// guardrail_events.outcome is "triggered" / "not_evaluated" +// (guardrails/runtime.py) and its .action carries flag/block; the device feed +// writes decision="dispatched" (agents/tools/remote_device.py). An earlier +// map here guessed at blocked/denied/allowed, none of which are ever +// produced, so every outcome rendered neutral grey -- including a guardrail +// that fired, which is the one signal the merged feed exists to surface. +const OUTCOME_TONES: Record = { + triggered: 'danger', + not_evaluated: 'muted', + dispatched: 'success', +}; + +export function outcomeTone(outcome: string): Tone { + return OUTCOME_TONES[outcome] ?? 'muted'; +} + +const OUTCOME_LABELS: Record = { + triggered: 'Triggered', + not_evaluated: 'Not evaluated', + dispatched: 'Dispatched', +}; + +export function outcomeLabel(outcome: string): string { + return ( + OUTCOME_LABELS[outcome] ?? + outcome.replace(/[._]/g, ' ').replace(/^\w/, (c) => c.toUpperCase()) + ); +} + // 127.0.0.1 / ::1 are noise in dev — collapse to a muted "local" chip so real // external IPs stand out in the audit feed. export function isLoopback(ip?: string | null): boolean { diff --git a/frontend/src/admin/Audit.tsx b/frontend/src/admin/Audit.tsx deleted file mode 100644 index 02c5e492..00000000 --- a/frontend/src/admin/Audit.tsx +++ /dev/null @@ -1,202 +0,0 @@ -import { useCallback, useEffect, useState } from 'react'; -import { useSelector } from 'react-redux'; -import { useSearchParams } from 'react-router-dom'; - -import adminService from '../api/services/adminService'; -import { Button } from '../components/ui/button'; -import { Input } from '../components/ui/input'; -import { - Table, - TableBody, - TableCell, - TableContainer, - TableHead, - TableHeader, - TableRow, -} from '../components/ui/table'; -import { selectToken } from '../preferences/preferenceSlice'; -import { - Loading, - Pill, - eventLabel, - eventTone, - fmtDate, - fmtNumber, - fmtRelative, - isLoopback, -} from './AdminUI'; - -type Event = { - id: string; - user_id: string; - event: string; - ip?: string; - created_at?: string; -}; - -const PAGE_SIZE = 50; - -export default function Audit() { - const token = useSelector(selectToken); - const [searchParams] = useSearchParams(); - const initialEvent = searchParams.get('event') ?? ''; - const [events, setEvents] = useState([]); - const [total, setTotal] = useState(0); - const [page, setPage] = useState(1); - const [eventFilter, setEventFilter] = useState(initialEvent); - const [userFilter, setUserFilter] = useState(''); - const [search, setSearch] = useState({ event: initialEvent, user_id: '' }); - const [loading, setLoading] = useState(true); - - const load = useCallback(async () => { - setLoading(true); - try { - const res = await adminService.getAudit( - { - page, - page_size: PAGE_SIZE, - event: search.event || undefined, - user_id: search.user_id || undefined, - }, - token, - ); - const json = await res.json().catch(() => ({})); - setEvents(json.events ?? []); - setTotal(json.total ?? 0); - } finally { - setLoading(false); - } - }, [token, page, search]); - - useEffect(() => { - load(); - }, [load]); - - const totalPages = Math.max(1, Math.ceil(total / PAGE_SIZE)); - const applyFilters = () => { - setPage(1); - setSearch({ event: eventFilter.trim(), user_id: userFilter.trim() }); - }; - - return ( -
-
- setEventFilter(e.target.value)} - onKeyDown={(e) => e.key === 'Enter' && applyFilters()} - className="max-w-xs" - /> - setUserFilter(e.target.value)} - onKeyDown={(e) => e.key === 'Enter' && applyFilters()} - className="max-w-xs" - /> - - {(search.event || search.user_id) && ( - - )} -
- - {loading ? ( - - ) : events.length === 0 ? ( -

No events.

- ) : ( - <> - - - - - Event - User - IP - When - - - - {events.map((ev, idx) => { - // De-emphasize a user id repeated from the row above (ditto), - // so distinct users stand out in a long single-user stream. - const repeat = - idx > 0 && events[idx - 1].user_id === ev.user_id; - return ( - - - - {eventLabel(ev.event)} - - - - {repeat ? '〃' : ev.user_id} - - - {isLoopback(ev.ip) ? ( - - local - - ) : ( - - {ev.ip ?? '—'} - - )} - - - - {fmtRelative(ev.created_at)} - - - - ); - })} - -
- -
-

- {fmtNumber(total)} events · page {page} of {totalPages} -

-
- - -
-
- - )} -
- ); -} diff --git a/frontend/src/admin/Usage.tsx b/frontend/src/admin/Usage.tsx index edd78743..4904515b 100644 --- a/frontend/src/admin/Usage.tsx +++ b/frontend/src/admin/Usage.tsx @@ -4,6 +4,13 @@ import { useSelector } from 'react-redux'; import adminService from '../api/services/adminService'; import SkeletonLoader from '../components/SkeletonLoader'; import { Button } from '../components/ui/button'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '../components/ui/select'; import { Table, TableBody, @@ -15,16 +22,33 @@ import { } from '../components/ui/table'; import { useDarkTheme } from '../hooks'; import { selectToken } from '../preferences/preferenceSlice'; -import { formatDate } from '../utils/dateTimeUtils'; -import { Loading, LoadError, StatCard, fmtNumber } from './AdminUI'; -import UsageChart, { usageColors } from './UsageChart'; +import { + Loading, + LoadError, + StatCard, + fmtMs, + fmtNumber, + fmtUsd, +} from './AdminUI'; +import UsageChart from './UsageChart'; +import { + GROUP_OPTIONS, + buildUsageChart, + cacheHitRate as computeCacheHitRate, + type GroupBy, + type Metric, + type UsageBucket, +} from './usageChartData'; +import UserUsageModal from './UserUsageModal'; -type Bucket = { - bucket: string; - prompt_tokens: number; - generated_tokens: number; +type TopUser = { user_id: string; tokens: number; cost: number }; +type Latency = { + samples: number; + p50_ms: number | null; + p95_ms: number | null; + ttft_samples: number; + ttft_p50_ms: number | null; }; -type TopUser = { user_id: string; tokens: number }; const RANGES = [7, 30, 90]; @@ -32,47 +56,40 @@ export default function Usage() { const token = useSelector(selectToken); const [isDarkTheme] = useDarkTheme(); const [days, setDays] = useState(30); + const [groupBy, setGroupBy] = useState('none'); + const [metric, setMetric] = useState('tokens'); const [data, setData] = useState(null); const [loading, setLoading] = useState(true); + const [drilldown, setDrilldown] = useState(null); const load = useCallback(async () => { setLoading(true); try { - const res = await adminService.getUsage({ days, bucket: 'day' }, token); + const res = await adminService.getUsage( + { days, bucket: 'day', group_by: groupBy }, + token, + ); setData(await res.json().catch(() => ({}))); } finally { setLoading(false); } - }, [token, days]); + }, [token, days, groupBy]); useEffect(() => { load(); }, [load]); - const series: Bucket[] = data?.series ?? []; + const series: UsageBucket[] = data?.series ?? []; const topUsers: TopUser[] = data?.top_users ?? []; - const promptTotal = series.reduce((sum, b) => sum + b.prompt_tokens, 0); - const generatedTotal = series.reduce((sum, b) => sum + b.generated_tokens, 0); + const latency: Latency | undefined = data?.latency; - const chartData = useMemo(() => { - const colors = usageColors(); - return { - labels: series.map((b) => formatDate(b.bucket)), - datasets: [ - { - label: 'Prompt', - data: series.map((b) => b.prompt_tokens), - backgroundColor: colors.prompt, - }, - { - label: 'Generated', - data: series.map((b) => b.generated_tokens), - backgroundColor: colors.generated, - }, - ], - }; + const chartData = useMemo( + () => buildUsageChart(series, groupBy, metric), // isDarkTheme re-resolves the canvas colors when the theme toggles. - }, [series, isDarkTheme]); + [series, groupBy, metric, isDarkTheme], + ); + + const cacheHitRate = useMemo(() => computeCacheHitRate(series), [series]); if (data === null && loading) return ; if (data && !data.success) @@ -80,34 +97,89 @@ export default function Usage() { return (
-
- {RANGES.map((r) => ( +
+
+ {RANGES.map((range) => ( + + ))} +
+ +
- ))} + +
+ - - - + +

- Token usage + {metric === 'cost' ? 'Spend' : 'Token usage'}

) : ( - + )}

- Top users by tokens + Top users

{topUsers.length === 0 ? (

No usage.

@@ -139,17 +215,33 @@ export default function Usage() { User - Tokens + Tokens + Cost - {topUsers.map((u) => ( - + {topUsers.map((user) => ( + - {u.user_id} + - - {fmtNumber(u.tokens)} + + {fmtNumber(user.tokens)} + + + {fmtUsd(user.cost)} ))} @@ -158,6 +250,8 @@ export default function Usage() { )}
+ + setDrilldown(null)} />
); } diff --git a/frontend/src/admin/UsageChart.tsx b/frontend/src/admin/UsageChart.tsx index d9c27f4d..f7d58496 100644 --- a/frontend/src/admin/UsageChart.tsx +++ b/frontend/src/admin/UsageChart.tsx @@ -39,6 +39,25 @@ export function usageColors(): { prompt: string; generated: string } { }; } +// Follow-on colors for a grouped chart (by model / agent / source), matching +// the palette settings/Analytics already uses for its stacked series. +const SERIES_COLORS = [ + '#FF6384', + '#36A2EB', + '#FFCE56', + '#4BC0C0', + '#9966FF', + '#FF9F40', + '#2BC596', +]; + +/** Color for dataset ``index``; the first uses the resolved brand color. */ +export function seriesColor(index: number): string { + return index === 0 + ? readCssVar('--primary', '#7D54D1') + : SERIES_COLORS[(index - 1) % SERIES_COLORS.length]; +} + function compactTick(value: number | string): string { const n = typeof value === 'number' ? value : Number(value); if (Number.isNaN(n)) return String(value); @@ -48,16 +67,28 @@ function compactTick(value: number | string): string { }).format(n); } +function currencyTick(value: number | string): string { + const n = typeof value === 'number' ? value : Number(value); + if (Number.isNaN(n)) return String(value); + return `$${new Intl.NumberFormat(undefined, { + notation: 'compact', + maximumFractionDigits: 2, + }).format(n)}`; +} + type UsageChartProps = { data: ChartData<'bar'>; legendID: string; maxTicksLimitInX?: number; + /** Format the y axis as USD rather than a token count. */ + currency?: boolean; }; export default function UsageChart({ data, legendID, maxTicksLimitInX = 8, + currency = false, }: UsageChartProps) { const options = { responsive: true, @@ -76,7 +107,7 @@ export default function UsageChart({ y: { grid: { lineWidth: 0.2, color: '#C4C4C4' }, border: { width: 0.2, color: '#C4C4C4' }, - ticks: { callback: compactTick }, + ticks: { callback: currency ? currencyTick : compactTick }, stacked: true, }, }, diff --git a/frontend/src/admin/UserUsageModal.tsx b/frontend/src/admin/UserUsageModal.tsx new file mode 100644 index 00000000..2a9f853b --- /dev/null +++ b/frontend/src/admin/UserUsageModal.tsx @@ -0,0 +1,222 @@ +import { useEffect, useMemo, useState } from 'react'; +import { useSelector } from 'react-redux'; + +import adminService from '../api/services/adminService'; +import { Button } from '../components/ui/button'; +import { Modal } from '../components/ui/modal'; +import { + Table, + TableBody, + TableCell, + TableContainer, + TableHead, + TableHeader, + TableRow, +} from '../components/ui/table'; +import { useDarkTheme } from '../hooks'; +import { selectToken } from '../preferences/preferenceSlice'; +import { formatDate } from '../utils/dateTimeUtils'; +import { Loading, LoadError, StatCard, fmtNumber, fmtUsd } from './AdminUI'; +import UsageChart, { usageColors } from './UsageChart'; + +type Bucket = { + bucket: string; + prompt_tokens: number; + generated_tokens: number; + cost: number; +}; +type Split = { key: string; tokens: number; cost: number }; + +const RANGES = [7, 30, 90]; + +/** + * Label for a token_usage.source value — what the spend was for. + * + * No `schedule` entry: that source marks the run-level rollup row, which the + * breakdown excludes to avoid double-counting. A scheduled run's individual + * calls are recorded as `agent_stream`, so its spend appears under Chat. + */ +const FLOW_LABELS: Record = { + agent_stream: 'Chat', + webhook: 'Webhook', + workflow: 'Workflow', + title: 'Title generation', + compression: 'History compression', + rag_condense: 'Question condensing', + fallback: 'Provider fallback', +}; + +function SplitTable({ + title, + rows, + labels, +}: { + title: string; + rows: Split[]; + /** Only the flow table maps keys; a model named `fallback` is a model. */ + labels?: Record; +}) { + return ( +
+

{title}

+ {rows.length === 0 ? ( +

No usage.

+ ) : ( + + + + + {title} + Tokens + Cost + + + + {rows.map((row) => ( + + + {labels?.[row.key] ?? row.key} + + + {fmtNumber(row.tokens)} + + + {fmtUsd(row.cost)} + + + ))} + +
+ + )} +
+ ); +} + +/** + * Per-user spend drill-down. The user detail view carried a single + * `tokens_30d` number, which answers neither question an operator has: what is + * this person costing, and what is driving it. + */ +export default function UserUsageModal({ + userId, + onClose, +}: { + userId: string | null; + onClose: () => void; +}) { + const token = useSelector(selectToken); + const [isDarkTheme] = useDarkTheme(); + const [days, setDays] = useState(30); + const [data, setData] = useState(null); + const [loading, setLoading] = useState(false); + + useEffect(() => { + if (!userId) { + setData(null); + return; + } + let cancelled = false; + setLoading(true); + // Drop the previous range's numbers rather than showing them under the + // newly selected one. + setData(null); + adminService + .getUserUsage(userId, { days }, token) + .then((res) => res.json()) + .then((json) => { + if (!cancelled) setData(json); + }) + .catch(() => !cancelled && setData({ success: false })) + .finally(() => !cancelled && setLoading(false)); + return () => { + cancelled = true; + }; + }, [userId, days, token]); + + const series: Bucket[] = data?.series ?? []; + const chartData = useMemo(() => { + const colors = usageColors(); + return { + labels: series.map((bucket) => formatDate(bucket.bucket)), + datasets: [ + { + label: 'Prompt', + data: series.map((bucket) => bucket.prompt_tokens), + backgroundColor: colors.prompt, + }, + { + label: 'Generated', + data: series.map((bucket) => bucket.generated_tokens), + backgroundColor: colors.generated, + }, + ], + }; + // isDarkTheme re-resolves the canvas colors when the theme toggles. + }, [series, isDarkTheme]); + + const totals = data?.totals ?? {}; + + return ( + !open && onClose()} + title={userId ? `Usage · ${userId}` : 'Usage'} + size="lg" + > +
+ {RANGES.map((range) => ( + + ))} +
+ + {loading ? ( + + ) : data && !data.success ? ( + + ) : ( +
+
+ + + +
+ +
+
+

Daily tokens

+
+
+
+ {series.length === 0 ? ( +

+ No usage in this period. +

+ ) : ( + + )} +
+
+ + + +
+ )} + + ); +} diff --git a/frontend/src/admin/Users.tsx b/frontend/src/admin/Users.tsx index d7ae7f8e..242e678e 100644 --- a/frontend/src/admin/Users.tsx +++ b/frontend/src/admin/Users.tsx @@ -42,6 +42,7 @@ import { fmtRelative, } from './AdminUI'; import UserQuotaModal from './UserQuotaModal'; +import UserUsageModal from './UserUsageModal'; type AdminUser = { user_id: string; @@ -72,6 +73,7 @@ export default function Users() { const [busy, setBusy] = useState(null); const [menuUserId, setMenuUserId] = useState(null); const [detail, setDetail] = useState(null); + const [usageFor, setUsageFor] = useState(null); const [quotaUserId, setQuotaUserId] = useState(null); const [feedback, setFeedback] = useState<{ ok: boolean; @@ -443,6 +445,7 @@ export default function Users() { /> ) : null} + setUsageFor(null)} /> setQuotaUserId(null)} @@ -528,6 +531,19 @@ export default function Users() {
+

Recent auth events diff --git a/frontend/src/admin/index.tsx b/frontend/src/admin/index.tsx index f8ad6157..65ec1c7c 100644 --- a/frontend/src/admin/index.tsx +++ b/frontend/src/admin/index.tsx @@ -5,7 +5,7 @@ import SectionIndexPage from '../navigation/SectionIndexPage'; import { CurrentSectionHeader } from '../navigation/SectionPageHeader'; import { ADMIN_SECTION } from '../navigation/sections'; import Admins from './Admins'; -import Audit from './Audit'; +import Activity from './Activity'; import Overview from './Overview'; import Quotas from './Quotas'; import Usage from './Usage'; @@ -38,7 +38,7 @@ export default function Admin() { } /> } /> } /> - } /> + } /> } /> diff --git a/frontend/src/admin/usageChartData.test.ts b/frontend/src/admin/usageChartData.test.ts new file mode 100644 index 00000000..a92de3ef --- /dev/null +++ b/frontend/src/admin/usageChartData.test.ts @@ -0,0 +1,123 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + buildUsageChart, + cacheHitRate, + type UsageBucket, +} from './usageChartData'; + +// Chart.js colors resolve against a live DOM; the shapes under test do not. +vi.mock('./UsageChart', () => ({ + seriesColor: (index: number) => `color-${index}`, + usageColors: () => ({ prompt: 'prompt-color', generated: 'generated-color' }), +})); + +const bucket = (overrides: Partial): UsageBucket => ({ + bucket: '2026-01-01', + prompt_tokens: 100, + generated_tokens: 20, + cost: 1, + cached_tokens: null, + cache_eligible_prompt_tokens: 0, + ...overrides, +}); + +describe('buildUsageChart', () => { + it('splits prompt and generated tokens when ungrouped', () => { + const chart = buildUsageChart([bucket({})], 'none', 'tokens'); + expect(chart.datasets.map((d) => d.label)).toEqual(['Prompt', 'Generated']); + expect(chart.datasets[0].data).toEqual([100]); + }); + + it('plots a single series for cost', () => { + const chart = buildUsageChart([bucket({ cost: 2.5 })], 'none', 'cost'); + expect(chart.datasets).toHaveLength(1); + expect(chart.datasets[0].data).toEqual([2.5]); + }); + + it('gives each group its own dataset', () => { + const chart = buildUsageChart( + [ + bucket({ group_key: 'gpt-a' }), + bucket({ group_key: 'gpt-b', prompt_tokens: 5, generated_tokens: 5 }), + ], + 'model', + 'tokens', + ); + expect(chart.datasets.map((d) => d.label)).toEqual(['gpt-a', 'gpt-b']); + expect(chart.datasets[1].data).toEqual([10]); + }); + + it('pads a group that is missing a bucket so the axes stay aligned', () => { + const chart = buildUsageChart( + [ + bucket({ bucket: '2026-01-01', group_key: 'a' }), + bucket({ bucket: '2026-01-02', group_key: 'b' }), + ], + 'model', + 'tokens', + ); + expect(chart.labels).toHaveLength(2); + // 'a' has no row on the second day: a zero, not a shifted bar. + expect(chart.datasets[0].data).toEqual([120, 0]); + expect(chart.datasets[1].data).toEqual([0, 120]); + }); + + it('labels a group with no key rather than dropping the row', () => { + const chart = buildUsageChart([bucket({})], 'model', 'tokens'); + expect(chart.datasets[0].label).toBe('unknown'); + }); + + it('handles an empty series', () => { + const chart = buildUsageChart([], 'model', 'cost'); + expect(chart.labels).toEqual([]); + expect(chart.datasets).toEqual([]); + }); +}); + +describe('cacheHitRate', () => { + it('is unknown when no provider reported cache bins', () => { + expect(cacheHitRate([bucket({ cached_tokens: null })])).toBeNull(); + }); + + it('is unknown for an empty series', () => { + expect(cacheHitRate([])).toBeNull(); + }); + + it('divides by the prompt tokens of the reporting calls only', () => { + // One bucket is one day, and a day mixes calls whose provider reported a + // cache breakdown with calls whose provider did not. 50/100, not 50/200. + const rate = cacheHitRate([ + bucket({ + prompt_tokens: 200, + cached_tokens: 50, + cache_eligible_prompt_tokens: 100, + }), + ]); + expect(rate).toBeCloseTo(50); + }); + + it('sums across buckets', () => { + const rate = cacheHitRate([ + bucket({ + prompt_tokens: 100, + cached_tokens: 10, + cache_eligible_prompt_tokens: 100, + }), + bucket({ + prompt_tokens: 100, + cached_tokens: 30, + cache_eligible_prompt_tokens: 100, + }), + ]); + expect(rate).toBeCloseTo(20); + }); + + it('avoids dividing by zero eligible tokens', () => { + expect( + cacheHitRate([ + bucket({ prompt_tokens: 500, cache_eligible_prompt_tokens: 0 }), + ]), + ).toBeNull(); + }); +}); diff --git a/frontend/src/admin/usageChartData.ts b/frontend/src/admin/usageChartData.ts new file mode 100644 index 00000000..ee862e98 --- /dev/null +++ b/frontend/src/admin/usageChartData.ts @@ -0,0 +1,116 @@ +import { formatDate } from '../utils/dateTimeUtils'; +import { seriesColor, usageColors } from './UsageChart'; + +export type UsageBucket = { + bucket: string; + prompt_tokens: number; + generated_tokens: number; + cost: number; + cached_tokens: number | null; + /** Prompt tokens from the rows that reported a cache breakdown. */ + cache_eligible_prompt_tokens: number; + group_key?: string; +}; + +export type GroupBy = 'none' | 'model' | 'agent' | 'source'; +export type Metric = 'tokens' | 'cost'; + +export const GROUP_OPTIONS: { value: GroupBy; label: string }[] = [ + { value: 'none', label: 'Total' }, + { value: 'model', label: 'By model' }, + { value: 'agent', label: 'By agent' }, + { value: 'source', label: 'By flow' }, +]; + +function valueOf(row: UsageBucket, metric: Metric): number { + return metric === 'cost' + ? row.cost + : row.prompt_tokens + row.generated_tokens; +} + +/** + * Shape the admin usage series for the chart. + * + * Ungrouped gives one dataset per token kind (or a single cost series); + * grouped gives one dataset per group key, each aligned to the shared bucket + * axis so a group missing a day plots 0 rather than shifting the whole series + * left by one bar. + */ +export function buildUsageChart( + series: UsageBucket[], + groupBy: GroupBy, + metric: Metric, +) { + const buckets = Array.from(new Set(series.map((row) => row.bucket))).sort(); + const labels = buckets.map((bucket) => formatDate(bucket)); + + if (groupBy === 'none') { + if (metric === 'cost') { + return { + labels, + datasets: [ + { + label: 'Cost', + data: series.map((row) => row.cost), + backgroundColor: seriesColor(0), + }, + ], + }; + } + const colors = usageColors(); + return { + labels, + datasets: [ + { + label: 'Prompt', + data: series.map((row) => row.prompt_tokens), + backgroundColor: colors.prompt, + }, + { + label: 'Generated', + data: series.map((row) => row.generated_tokens), + backgroundColor: colors.generated, + }, + ], + }; + } + + const keys = Array.from( + new Set(series.map((row) => row.group_key ?? 'unknown')), + ); + return { + labels, + datasets: keys.map((key, index) => { + const byBucket = new Map( + series + .filter((row) => (row.group_key ?? 'unknown') === key) + .map((row) => [row.bucket, valueOf(row, metric)]), + ); + return { + label: key, + data: buckets.map((bucket) => byBucket.get(bucket) ?? 0), + backgroundColor: seriesColor(index), + }; + }), + }; +} + +/** + * Prompt-cache hit rate over the calls whose provider reported a breakdown. + * + * Both numerator and denominator come from the server, because a bucket is a + * whole day and mixes reporting with non-reporting calls: filtering *buckets* + * here would still divide by the day's entire prompt volume and understate + * the rate by however much traffic ran on a provider that reports nothing. + * + * Returns null when no call in the window reported one. + */ +export function cacheHitRate(series: UsageBucket[]): number | null { + const eligible = series.reduce( + (sum, row) => sum + (row.cache_eligible_prompt_tokens ?? 0), + 0, + ); + if (eligible === 0) return null; + const cached = series.reduce((sum, row) => sum + (row.cached_tokens ?? 0), 0); + return (cached / eligible) * 100; +} diff --git a/frontend/src/api/endpoints.ts b/frontend/src/api/endpoints.ts index 89187528..28a95dbc 100644 --- a/frontend/src/api/endpoints.ts +++ b/frontend/src/api/endpoints.ts @@ -170,10 +170,13 @@ const endpoints = { `/api/admin/users/${encodeURIComponent(id)}/role`, USER_REVOKE_SESSIONS: (id: string) => `/api/admin/users/${encodeURIComponent(id)}/revoke-sessions`, + USER_USAGE: (id: string) => + `/api/admin/users/${encodeURIComponent(id)}/usage`, ADMINS: '/api/admin/admins', USAGE: '/api/admin/usage', - AUDIT: '/api/admin/audit', - DEVICE_AUDIT: '/api/admin/devices/audit', + ACTIVITY: '/api/admin/activity', + ACTIVITY_EVENTS: '/api/admin/activity/events', + ACTIVITY_EXPORT: '/api/admin/activity/export', QUOTAS: '/api/admin/quotas', QUOTA_INSTANCE: '/api/admin/quotas/instance', QUOTA_TEAM: (id: string) => diff --git a/frontend/src/api/services/adminService.ts b/frontend/src/api/services/adminService.ts index adf67e71..aef39370 100644 --- a/frontend/src/api/services/adminService.ts +++ b/frontend/src/api/services/adminService.ts @@ -1,15 +1,36 @@ import apiClient from '../client'; import endpoints from '../endpoints'; -const qs = (params: Record): string => { +type QueryValue = string | number | string[] | undefined; + +const qs = (params: Record): string => { const search = new URLSearchParams(); Object.entries(params).forEach(([key, value]) => { - if (value !== undefined && value !== '') search.append(key, String(value)); + if (value === undefined || value === '') return; + // Array facets repeat the key; the API accumulates repeats. + if (Array.isArray(value)) { + value.forEach((item) => item && search.append(key, item)); + return; + } + search.append(key, String(value)); }); const str = search.toString(); return str ? `?${str}` : ''; }; +export type ActivityFilters = { + page?: number; + page_size?: number; + category?: string[]; + event?: string[]; + feed?: string[]; + actor_id?: string; + user_id?: string; + since?: string; + until?: string; + search?: string; +}; + export type QuotaScope = 'instance' | 'team' | 'user'; const quotaUrl = (scope: QuotaScope, subjectId?: string | null): string => { @@ -47,21 +68,25 @@ const adminService = { token: string | null, ): Promise => apiClient.get(`${endpoints.ADMIN.USAGE}${qs(params)}`, token), - getAudit: ( - params: { - page?: number; - page_size?: number; - event?: string; - user_id?: string; - }, + getUserUsage: ( + userId: string, + params: { days?: number }, token: string | null, ): Promise => - apiClient.get(`${endpoints.ADMIN.AUDIT}${qs(params)}`, token), - getDeviceAudit: ( - params: { page?: number; page_size?: number; decision?: string }, + apiClient.get(`${endpoints.ADMIN.USER_USAGE(userId)}${qs(params)}`, token), + getActivity: (filters: ActivityFilters, token: string | null): Promise => + apiClient.get(`${endpoints.ADMIN.ACTIVITY}${qs(filters)}`, token), + getActivityEvents: (token: string | null): Promise => + apiClient.get(endpoints.ADMIN.ACTIVITY_EVENTS, token), + exportActivity: ( + filters: ActivityFilters, + format: 'csv' | 'ndjson', token: string | null, ): Promise => - apiClient.get(`${endpoints.ADMIN.DEVICE_AUDIT}${qs(params)}`, token), + apiClient.get( + `${endpoints.ADMIN.ACTIVITY_EXPORT}${qs({ ...filters, format })}`, + token, + ), getQuotas: (token: string | null): Promise => apiClient.get(endpoints.ADMIN.QUOTAS, token), getUserQuota: (userId: string, token: string | null): Promise => diff --git a/frontend/src/components/ui/table.tsx b/frontend/src/components/ui/table.tsx index 4f3b7c7f..b15cfb6e 100644 --- a/frontend/src/components/ui/table.tsx +++ b/frontend/src/components/ui/table.tsx @@ -32,6 +32,8 @@ interface TableCellProps { minWidth?: string; width?: string; align?: 'left' | 'right' | 'center'; + /** Span several columns, e.g. a detail row under the row it expands. */ + colSpan?: number; } const TableContainer = React.forwardRef( @@ -134,6 +136,7 @@ const TableHeader: React.FC = ({ minWidth, width, align = 'left', + colSpan, }) => { const alignmentClass = align === 'right' @@ -145,6 +148,7 @@ const TableHeader: React.FC = ({ return ( = ({ minWidth, width, align = 'left', + colSpan, }) => { const alignmentClass = align === 'right' @@ -175,6 +180,7 @@ const TableCell: React.FC = ({ return ( Path: + return Path(__file__).resolve().parents[3] / "docsgpt" / "alembic.ini" + + +def _run_alembic(url: str, *args: str) -> None: + subprocess.check_call( + [sys.executable, "-m", "alembic", "-c", str(_alembic_ini()), *args], + timeout=60, + env={**os.environ, "POSTGRES_URI": url}, + ) + + +def _alembic_heads(url: str) -> list[str]: + out = subprocess.check_output( + [sys.executable, "-m", "alembic", "-c", str(_alembic_ini()), "heads"], + timeout=60, + env={**os.environ, "POSTGRES_URI": url}, + text=True, + ) + return [line for line in out.splitlines() if line.strip()] + + +def _alembic_version(conn) -> str: + return conn.execute(text("SELECT version_num FROM alembic_version")).scalar() + + +def _column_exists(conn, table: str, column: str) -> bool: + row = conn.execute( + text( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = :t AND column_name = :c AND table_schema = 'public'" + ), + {"t": table, "c": column}, + ).fetchone() + return row is not None + + +def _index_exists(conn, name: str) -> bool: + row = conn.execute( + text("SELECT 1 FROM pg_indexes WHERE schemaname = 'public' AND indexname = :n"), + {"n": name}, + ).fetchone() + return row is not None + + +_0034 = "0034_auth_events_actor_target" +_0033 = "0033_quotas" + + +class TestMigration0034RoundTrip: + def test_single_head(self, pg_engine): + url = pg_engine.url.render_as_string(hide_password=False) + assert len(_alembic_heads(url)) == 1 + + def test_head_has_actor_target_columns(self, pg_engine): + with pg_engine.connect() as conn: + assert _alembic_version(conn) >= _0034 + assert _column_exists(conn, "auth_events", "actor_id") + assert _column_exists(conn, "auth_events", "target_id") + + def test_head_has_feed_indexes(self, pg_engine): + """The global feed orders by created_at with no user filter.""" + with pg_engine.connect() as conn: + assert _index_exists(conn, "auth_events_created_idx") + assert _index_exists(conn, "auth_events_event_created_idx") + assert _index_exists(conn, "auth_events_actor_idx") + + def test_downgrade_drops_then_upgrade_restores(self, pg_engine): + url = pg_engine.url.render_as_string(hide_password=False) + _run_alembic(url, "downgrade", _0033) + with pg_engine.connect() as conn: + assert _alembic_version(conn) == _0033 + assert not _column_exists(conn, "auth_events", "actor_id") + assert not _column_exists(conn, "auth_events", "target_id") + assert not _index_exists(conn, "auth_events_created_idx") + _run_alembic(url, "upgrade", "head") + with pg_engine.connect() as conn: + assert _alembic_version(conn) >= _0034 + assert _column_exists(conn, "auth_events", "actor_id") + + def test_backfill_reads_actor_out_of_metadata(self, pg_engine): + """An admin action filed under its target keeps the acting admin.""" + url = pg_engine.url.render_as_string(hide_password=False) + _run_alembic(url, "downgrade", _0033) + with pg_engine.begin() as conn: + conn.execute( + text( + "INSERT INTO auth_events (user_id, event, metadata) VALUES " + "('victim', 'admin_user_deactivated', '{\"by\": \"admin-1\"}'::jsonb), " + "('grantee', 'role_granted', '{\"granted_by\": \"admin-2\"}'::jsonb), " + "('someone', 'oidc_login', '{}'::jsonb), " + "('actor-3', 'team.create', '{\"team_id\": \"t1\"}'::jsonb)" + ) + ) + _run_alembic(url, "upgrade", "head") + with pg_engine.connect() as conn: + rows = dict( + conn.execute( + text( + "SELECT event, actor_id || '|' || COALESCE(target_id, '-') " + "FROM auth_events WHERE user_id IN " + "('victim', 'grantee', 'someone', 'actor-3')" + ) + ).fetchall() + ) + assert rows["admin_user_deactivated"] == "admin-1|victim" + assert rows["role_granted"] == "admin-2|grantee" + # Self-service events: the user is both actor and target. + assert rows["oidc_login"] == "someone|someone" + # Team events were always filed under the actor; they have no user target. + assert rows["team.create"] == "actor-3|-" + + def test_legacy_insert_keeps_its_attribution(self, pg_engine): + """A previous-release writer must neither 500 nor lose attribution. + + Its inserts name only the old columns, so ``actor_id`` arrives NULL. + The NOT NULL alone would reject them -- and in admin/routes.py that + insert shares the request's transaction, so the role grant beside it + would roll back too. A bare sentinel would keep the row but throw the + actor away, and the highest-volume events are the recoverable ones. + """ + with pg_engine.begin() as conn: + conn.execute( + text( + "INSERT INTO auth_events (user_id, event, metadata) VALUES " + # A login: the user is their own actor and target. + "('legacy-login', 'oidc_login', '{}'::jsonb), " + # An admin action: the actor is recoverable from metadata. + "('legacy-victim', 'admin_user_deactivated', " + " '{\"by\": \"admin-legacy\"}'::jsonb), " + # A team event: filed under the actor, no user target. + "('legacy-actor', 'team.create', '{}'::jsonb)" + ) + ) + rows = dict( + conn.execute( + text( + "SELECT user_id, actor_id || '|' || COALESCE(target_id, '-') " + "FROM auth_events WHERE user_id LIKE 'legacy-%'" + ) + ).fetchall() + ) + assert rows["legacy-login"] == "legacy-login|legacy-login" + assert rows["legacy-victim"] == "admin-legacy|legacy-victim" + assert rows["legacy-actor"] == "legacy-actor|-" + + def test_trigger_leaves_a_current_writer_alone(self, pg_engine): + """A deliberate NULL target must survive. + + ``conversation.deleted`` acts on a resource, not an account. Deriving + a target there would name the actor as their own target. + """ + with pg_engine.begin() as conn: + conn.execute( + text( + "INSERT INTO auth_events " + "(user_id, actor_id, target_id, event, metadata) " + "VALUES ('u9', 'u9', NULL, 'conversation.deleted', '{}'::jsonb)" + ) + ) + row = conn.execute( + text( + "SELECT actor_id, target_id FROM auth_events " + "WHERE event = 'conversation.deleted'" + ) + ).fetchone() + assert tuple(row) == ("u9", None) + + def test_backfill_scopes_quota_policy_targets(self, pg_engine): + """Only a user-scoped quota policy has a user target. + + Instance and team policies are filed under the acting admin, so + backfilling ``target_id = user_id`` would claim the admin was acted on. + """ + url = pg_engine.url.render_as_string(hide_password=False) + _run_alembic(url, "downgrade", _0033) + with pg_engine.begin() as conn: + conn.execute( + text( + "INSERT INTO auth_events (user_id, event, metadata) VALUES " + "('admin-q', 'quota_policy_set', " + " '{\"by\": \"admin-q\", \"scope\": \"instance\"}'::jsonb), " + "('admin-q2', 'quota_policy_deleted', " + " '{\"by\": \"admin-q2\", \"scope\": \"team\"}'::jsonb), " + "('capped-user', 'quota_policy_set', " + " '{\"by\": \"admin-q3\", \"scope\": \"user\"}'::jsonb)" + ) + ) + _run_alembic(url, "upgrade", "head") + with pg_engine.connect() as conn: + rows = dict( + conn.execute( + text( + "SELECT user_id, actor_id || '|' || COALESCE(target_id, '-') " + "FROM auth_events WHERE user_id IN " + "('admin-q', 'admin-q2', 'capped-user')" + ) + ).fetchall() + ) + assert rows["admin-q"] == "admin-q|-" + assert rows["admin-q2"] == "admin-q2|-" + assert rows["capped-user"] == "admin-q3|capped-user" diff --git a/tests/storage/db/test_migration_0035.py b/tests/storage/db/test_migration_0035.py new file mode 100644 index 00000000..4fe13f9a --- /dev/null +++ b/tests/storage/db/test_migration_0035.py @@ -0,0 +1,85 @@ +"""Migration round-trip test for 0035_token_usage_latency.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +import pytest +from sqlalchemy import text + + +pytestmark = pytest.mark.integration + + +def _alembic_ini() -> Path: + return Path(__file__).resolve().parents[3] / "docsgpt" / "alembic.ini" + + +def _run_alembic(url: str, *args: str) -> None: + subprocess.check_call( + [sys.executable, "-m", "alembic", "-c", str(_alembic_ini()), *args], + timeout=60, + env={**os.environ, "POSTGRES_URI": url}, + ) + + +def _alembic_version(conn) -> str: + return conn.execute(text("SELECT version_num FROM alembic_version")).scalar() + + +def _column_exists(conn, table: str, column: str) -> bool: + row = conn.execute( + text( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = :t AND column_name = :c AND table_schema = 'public'" + ), + {"t": table, "c": column}, + ).fetchone() + return row is not None + + +_0035 = "0035_token_usage_latency" +_0034 = "0034_auth_events_actor_target" + + +class TestMigration0035RoundTrip: + def test_head_has_latency_columns(self, pg_engine): + with pg_engine.connect() as conn: + assert _alembic_version(conn) >= _0035 + assert _column_exists(conn, "token_usage", "duration_ms") + assert _column_exists(conn, "token_usage", "ttft_ms") + + def test_downgrade_drops_then_upgrade_restores(self, pg_engine): + url = pg_engine.url.render_as_string(hide_password=False) + _run_alembic(url, "downgrade", _0034) + with pg_engine.connect() as conn: + assert _alembic_version(conn) == _0034 + assert not _column_exists(conn, "token_usage", "duration_ms") + assert not _column_exists(conn, "token_usage", "ttft_ms") + _run_alembic(url, "upgrade", "head") + with pg_engine.connect() as conn: + assert _column_exists(conn, "token_usage", "duration_ms") + + def test_existing_rows_read_null_latency(self, pg_engine): + """Rows written before 0035 must read NULL, not 0 — a 0ms call is a lie.""" + url = pg_engine.url.render_as_string(hide_password=False) + _run_alembic(url, "downgrade", _0034) + with pg_engine.begin() as conn: + conn.execute( + text( + "INSERT INTO token_usage (user_id, prompt_tokens, generated_tokens) " + "VALUES ('u-mig35', 10, 1)" + ) + ) + _run_alembic(url, "upgrade", "head") + with pg_engine.connect() as conn: + row = conn.execute( + text( + "SELECT duration_ms, ttft_ms FROM token_usage " + "WHERE user_id = 'u-mig35'" + ) + ).fetchone() + assert tuple(row) == (None, None) diff --git a/tests/test_scim.py b/tests/test_scim.py index 77725655..5b3e96fc 100644 --- a/tests/test_scim.py +++ b/tests/test_scim.py @@ -276,7 +276,11 @@ class TestCreateUser: assert response.headers["Location"].endswith(f"/scim/v2/Users/{USER_PK}") scim_mocks.users.create.assert_called_once_with("alice@example.com", active=True) scim_mocks.audit.insert.assert_called_once_with( - "alice@example.com", "scim_created", metadata={"via": "scim"} + "alice@example.com", + "scim_created", + metadata={"via": "scim"}, + actor_id="system:scim", + target_id="alice@example.com", ) body = response.get_json() assert body["id"] == USER_PK @@ -373,7 +377,11 @@ class TestReplaceUser: # already lets a fresh login through on a newer iat). scim_mocks.deny_user.assert_not_called() scim_mocks.audit.insert.assert_called_once_with( - "alice@example.com", "scim_reactivated", metadata={"via": "scim"} + "alice@example.com", + "scim_reactivated", + metadata={"via": "scim"}, + actor_id="system:scim", + target_id="alice@example.com", ) def test_put_active_false_triggers_deny(self, client, scim_settings, scim_mocks): @@ -390,7 +398,11 @@ class TestReplaceUser: scim_mocks.users.set_active.assert_called_once_with(USER_PK, False) scim_mocks.deny_user.assert_called_once_with("alice@example.com") scim_mocks.audit.insert.assert_called_once_with( - "alice@example.com", "scim_deactivated", metadata={"via": "scim"} + "alice@example.com", + "scim_deactivated", + metadata={"via": "scim"}, + actor_id="system:scim", + target_id="alice@example.com", ) def test_put_differently_cased_username_deprovisions(self, client, scim_settings, scim_mocks): @@ -445,7 +457,11 @@ class TestPatchUser: scim_mocks.users.set_active.assert_called_once_with(USER_PK, False) scim_mocks.deny_user.assert_called_once_with("alice@example.com") scim_mocks.audit.insert.assert_called_once_with( - "alice@example.com", "scim_deactivated", metadata={"via": "scim"} + "alice@example.com", + "scim_deactivated", + metadata={"via": "scim"}, + actor_id="system:scim", + target_id="alice@example.com", ) def test_replace_with_path_deactivates(self, client, scim_settings, scim_mocks): @@ -479,7 +495,11 @@ class TestPatchUser: scim_mocks.users.set_active.assert_called_once_with(USER_PK, True) scim_mocks.deny_user.assert_not_called() scim_mocks.audit.insert.assert_called_once_with( - "alice@example.com", "scim_reactivated", metadata={"via": "scim"} + "alice@example.com", + "scim_reactivated", + metadata={"via": "scim"}, + actor_id="system:scim", + target_id="alice@example.com", ) def test_bogus_path_rejected(self, client, scim_settings, scim_mocks): @@ -540,7 +560,11 @@ class TestDeleteUser: scim_mocks.users.set_active.assert_called_once_with(USER_PK, False) scim_mocks.deny_user.assert_called_once_with("alice@example.com") scim_mocks.audit.insert.assert_called_once_with( - "alice@example.com", "scim_deactivated", metadata={"via": "scim"} + "alice@example.com", + "scim_deactivated", + metadata={"via": "scim"}, + actor_id="system:scim", + target_id="alice@example.com", ) def test_delete_already_inactive_has_no_side_effects(self, client, scim_settings, scim_mocks): diff --git a/tests/test_usage_latency.py b/tests/test_usage_latency.py new file mode 100644 index 00000000..0bc1cc1e --- /dev/null +++ b/tests/test_usage_latency.py @@ -0,0 +1,113 @@ +"""Latency is measured by the usage wrappers and persisted with the call. + +``duration_ms`` and ``ttft_ms`` were already computed for the finish log lines +and then discarded; these pin that they now reach ``token_usage``. +""" + +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from docsgpt.usage import gen_token_usage, stream_token_usage + + +class _LLM: + """Minimal stand-in for an LLM instance the wrappers decorate.""" + + def __init__(self): + self.token_usage = {"prompt_tokens": 0, "generated_tokens": 0} + self.decoded_token = {"sub": "u1"} + self.user_api_key = None + self.agent_id = None + + +@pytest.fixture +def persisted(): + calls: list[dict] = [] + + def _capture(llm, call_usage, *, duration_ms=None, ttft_ms=None): + calls.append({"duration_ms": duration_ms, "ttft_ms": ttft_ms}) + + with patch("docsgpt.usage._persist_call_usage", _capture): + yield calls + + +@pytest.mark.unit +class TestNonStreaming: + def test_records_a_duration_and_no_first_token(self, persisted): + @gen_token_usage + def _gen(self, model, messages, stream, tools, **kwargs): + return "hello" + + assert _gen(_LLM(), "m", [], False, None) == "hello" + assert persisted[0]["duration_ms"] >= 0 + # A non-streaming call has no first-token moment. + assert persisted[0]["ttft_ms"] is None + + def test_a_failed_call_still_records_its_duration(self, persisted): + @gen_token_usage + def _gen(self, model, messages, stream, tools, **kwargs): + raise RuntimeError("upstream down") + + with pytest.raises(RuntimeError): + _gen(_LLM(), "m", [], False, None) + assert persisted[0]["duration_ms"] >= 0 + + +@pytest.mark.unit +class TestStreaming: + def test_records_time_to_first_chunk(self, persisted): + @stream_token_usage + def _stream(self, model, messages, stream, tools, **kwargs): + yield "a" + yield "b" + + assert list(_stream(_LLM(), "m", [], True, None)) == ["a", "b"] + row = persisted[0] + assert row["ttft_ms"] is not None + # First token cannot land after the call finished. + assert row["ttft_ms"] <= row["duration_ms"] + + def test_a_stream_that_never_yields_has_no_first_token(self, persisted): + """NULL, not 0 — an instant p50 would be a lie.""" + + @stream_token_usage + def _stream(self, model, messages, stream, tools, **kwargs): + raise RuntimeError("refused") + yield # pragma: no cover - unreachable, makes this a generator + + with pytest.raises(RuntimeError): + list(_stream(_LLM(), "m", [], True, None)) + assert persisted[0]["ttft_ms"] is None + assert persisted[0]["duration_ms"] >= 0 + + def test_duration_excludes_consumer_backpressure(self, persisted): + """The clock must measure the provider, not a slow reader. + + ``stream_token_usage`` is a generator, so every yield suspends until + the consumer comes back. Timing start-to-exhaustion would bill the + agent loop's tool handling and the SSE client's pace to the model. + """ + import time as _time + + @stream_token_usage + def _stream(self, model, messages, stream, tools, **kwargs): + yield "a" + yield "b" + + for _ in _stream(_LLM(), "m", [], True, None): + # A consumer that takes far longer than the provider did. + _time.sleep(0.05) + assert persisted[0]["duration_ms"] < 50 + + def test_a_stream_cut_short_keeps_the_first_token_it_saw(self, persisted): + @stream_token_usage + def _stream(self, model, messages, stream, tools, **kwargs): + yield "a" + raise RuntimeError("dropped") + + with pytest.raises(RuntimeError): + list(_stream(_LLM(), "m", [], True, None)) + assert persisted[0]["ttft_ms"] is not None