diff --git a/docs/content/Deploying/Access-Control.mdx b/docs/content/Deploying/Access-Control.mdx index 61cbd432..208c378e 100644 --- a/docs/content/Deploying/Access-Control.mdx +++ b/docs/content/Deploying/Access-Control.mdx @@ -171,6 +171,10 @@ The **Admin → Activity** tab merges three append-only journals into one timeli 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 4b7c537f..6282b104 100644 --- a/docs/content/Deploying/OIDC-SSO.mdx +++ b/docs/content/Deploying/OIDC-SSO.mdx @@ -190,7 +190,7 @@ Authentication activity is recorded in `auth_events`, an append-only Postgres ta - `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). +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). `actor_id` defaults to `unknown`, which only a process from a release predating the column can produce. | Event | Recorded when | | --- | --- | diff --git a/docsgpt/alembic/versions/0034_auth_events_actor_target.py b/docsgpt/alembic/versions/0034_auth_events_actor_target.py index 0854f730..91887209 100644 --- a/docsgpt/alembic/versions/0034_auth_events_actor_target.py +++ b/docsgpt/alembic/versions/0034_auth_events_actor_target.py @@ -76,6 +76,15 @@ def upgrade() -> None: # Backfilled every row above, so the NOT NULL is safe to assert now. New # rows always carry an actor (the repository derives one from ``user_id``). + # + # The DEFAULT is what makes this safe during a rolling deploy: a + # still-running previous-release process inserts (user_id, event, ip, + # user_agent, metadata) with no actor, and those inserts share the + # request's transaction in admin/routes.py -- a NotNullViolation there + # would 500 the deactivate or role grant and roll its write back with it. + # 'unknown' is the same sentinel ``record_event`` writes for an + # unauthenticated actor, so the column stays readable either way. + op.execute("ALTER TABLE auth_events ALTER COLUMN actor_id SET DEFAULT 'unknown';") op.execute("ALTER TABLE auth_events ALTER COLUMN actor_id SET NOT NULL;") op.execute( 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/activity.py b/docsgpt/api/admin/activity.py index ec6add8f..736a0baf 100644 --- a/docsgpt/api/admin/activity.py +++ b/docsgpt/api/admin/activity.py @@ -43,23 +43,38 @@ _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, dropping unknown values. + """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; anything else is ignored rather than - rejected, so an old bookmark degrades instead of erroring. + allowed: The permitted values. Returns: - The distinct valid values in the order given, or None when absent. + 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(",")) + raw.extend(part.strip() for part in value.split(",") if part.strip()) seen: list[str] = [] for value in raw: - if value in allowed and value not in seen: + 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 @@ -76,14 +91,21 @@ def _event_list() -> Optional[list[str]]: def _time_arg(name: str) -> Optional[datetime]: - """Parse an ISO-8601 timestamp query arg, or None when absent/unparseable.""" + """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: - return None + 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) @@ -122,7 +144,10 @@ class AdminActivityResource(Resource): def get(self): """Merged audit feed, newest first, with filters and a total.""" page, page_size, offset = _page() - filters = _filters() + 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) @@ -183,7 +208,10 @@ class AdminActivityExportResource(Resource): jsonify({"success": False, "message": "Invalid format"}), 400 ) limit = max(1, min(_EXPORT_MAX_ROWS, _int_arg("max_rows", _EXPORT_MAX_ROWS))) - filters = _filters() + 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( diff --git a/docsgpt/api/admin/routes.py b/docsgpt/api/admin/routes.py index 318c6b76..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), } @@ -347,7 +354,6 @@ class AdminUsageResource(Resource): total = usage_repo.sum_tokens_in_range( start=start, end=datetime.now(timezone.utc) ) - by_model = usage_repo.tokens_by_model(start=start) latency = stats_repo.latency_summary(since=start) top_users = stats_repo.top_token_users(since=start, limit=10) return make_response( @@ -362,7 +368,6 @@ class AdminUsageResource(Resource): # 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), - "by_model": by_model, "latency": latency, "top_users": top_users, } diff --git a/docsgpt/api/audit.py b/docsgpt/api/audit.py index 26c6bd4a..a5d131b8 100644 --- a/docsgpt/api/audit.py +++ b/docsgpt/api/audit.py @@ -25,6 +25,7 @@ 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 @@ -32,7 +33,7 @@ logger = logging.getLogger(__name__) def record_event( - conn, + conn: Connection, event: str, *, actor: Optional[str], @@ -42,8 +43,9 @@ def record_event( """Append one audit event, best-effort, inside the caller's transaction. Args: - conn: Open database connection, inside the action's transaction so the - audit row commits atomically with the change it describes. + 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 diff --git a/docsgpt/api/user/agents/routes.py b/docsgpt/api/user/agents/routes.py index 308462ce..84f7fbbe 100644 --- a/docsgpt/api/user/agents/routes.py +++ b/docsgpt/api/user/agents/routes.py @@ -1294,7 +1294,8 @@ class UpdateAgent(Resource): updated = bool(result) else: updated = agents_repo.update(pg_agent_id, user, update_fields) - if updated: + # An update that changed no field is not an event. + if updated and update_fields: record_event( conn, "agent.updated", diff --git a/docsgpt/api/user/conversations/routes.py b/docsgpt/api/user/conversations/routes.py index 36578378..189763bf 100644 --- a/docsgpt/api/user/conversations/routes.py +++ b/docsgpt/api/user/conversations/routes.py @@ -73,12 +73,15 @@ class DeleteAllConversations(Resource): try: with db_session() as conn: deleted = ConversationsRepository(conn).delete_all_for_user(user_id) - record_event( - conn, - "conversation.deleted_all", - actor=user_id, - deleted=deleted, - ) + # 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/upload.py b/docsgpt/api/user/sources/upload.py index 31f2d700..f15bee65 100644 --- a/docsgpt/api/user/sources/upload.py +++ b/docsgpt/api/user/sources/upload.py @@ -5,6 +5,7 @@ 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 @@ -179,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( @@ -397,26 +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 - # Audited here, not in the worker: this is the user action. The - # ingest may still fail, which the source's own status records. - try: - with db_session() as conn: - record_event( - conn, - "source.created", - actor=user, - source_id=str(source_uuid), - name=job_name, - type="local", - task_id=response_task_id, - ) - except Exception as err: - current_app.logger.warning( - "Could not audit upload for source %s: %s", - source_uuid, - err, - exc_info=True, - ) + _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 @@ -568,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 @@ -601,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/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 edab66a5..79773b6e 100644 --- a/docsgpt/storage/db/models.py +++ b/docsgpt/storage/db/models.py @@ -250,9 +250,11 @@ 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``. Wall-clock for the call, 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. + # 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), ) diff --git a/docsgpt/storage/db/repositories/activity.py b/docsgpt/storage/db/repositories/activity.py index 2cfa8281..4c62bf65 100644 --- a/docsgpt/storage/db/repositories/activity.py +++ b/docsgpt/storage/db/repositories/activity.py @@ -25,6 +25,7 @@ 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. @@ -197,12 +198,18 @@ class ActivityRepository: 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 OR target_id ILIKE :search " - "OR event ILIKE :search OR ip ILIKE :search " - "OR COALESCE(outcome, '') ILIKE :search OR detail::text ILIKE :search)" + "(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"%{search}%" + 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"]) @@ -259,7 +266,10 @@ class ActivityRepository: return [dict(row._mapping) for row in result.fetchall()] def _page_after( - self, cursor: Optional[tuple], limit: int, filters: dict + self, + cursor: Optional[tuple[datetime, str, str]], + limit: int, + filters: dict, ) -> list[dict]: """One keyset page strictly older than ``cursor``. @@ -332,7 +342,7 @@ class ActivityRepository: Row dicts carrying :data:`ACTIVITY_COLUMNS`. """ emitted = 0 - cursor: Optional[tuple] = None + cursor: Optional[tuple[datetime, str, str]] = None while emitted < max_rows: page = self._page_after( cursor, min(chunk_size, max_rows - emitted), filters diff --git a/docsgpt/storage/db/repositories/admin_stats.py b/docsgpt/storage/db/repositories/admin_stats.py index bd4241f9..b4c54bee 100644 --- a/docsgpt/storage/db/repositories/admin_stats.py +++ b/docsgpt/storage/db/repositories/admin_stats.py @@ -17,7 +17,7 @@ from docsgpt.storage.db.base_repository import row_to_dict from docsgpt.storage.db.repositories.token_usage import TokenUsageRepository -def _round_ms(value) -> Optional[int]: +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 diff --git a/docsgpt/storage/db/repositories/auth_events.py b/docsgpt/storage/db/repositories/auth_events.py index b87e6ae7..04f6172a 100644 --- a/docsgpt/storage/db/repositories/auth_events.py +++ b/docsgpt/storage/db/repositories/auth_events.py @@ -88,18 +88,44 @@ 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()] @@ -108,41 +134,18 @@ class AuthEventsRepository: event: Optional[str], user_id: Optional[str], since: Any, - *, - events: Optional[Sequence[str]] = None, - actor_id: Optional[str] = None, - until: Any = None, - search: Optional[str] = None, ) -> tuple[str, dict]: clauses: list[str] = [] params: dict = {} if event: clauses.append("event = :event") params["event"] = event - if events: - clauses.append("event = ANY(:events)") - params["events"] = list(events) if user_id: clauses.append("(user_id = :user_id OR target_id = :user_id)") params["user_id"] = user_id - if actor_id: - clauses.append("actor_id = :actor_id") - params["actor_id"] = actor_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: - # Substring match across the human-meaningful columns plus the - # serialized metadata, so "ci-deploy-key" finds the PAT it named. - clauses.append( - "(user_id ILIKE :search OR actor_id ILIKE :search " - "OR target_id ILIKE :search OR event ILIKE :search " - "OR ip ILIKE :search OR metadata::text ILIKE :search)" - ) - params["search"] = f"%{search}%" where = ("WHERE " + " AND ".join(clauses)) if clauses else "" return where, params @@ -150,25 +153,18 @@ class AuthEventsRepository: self, *, event: Optional[str] = None, - events: Optional[Sequence[str]] = None, user_id: Optional[str] = None, - actor_id: Optional[str] = None, since=None, - until=None, - search: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> list[dict]: - """Global audit feed (admin), newest first; see :meth:`_filter_clauses`.""" - where, params = self._filter_clauses( - event, - user_id, - since, - events=events, - actor_id=actor_id, - until=until, - search=search, - ) + """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( text( @@ -183,37 +179,14 @@ class AuthEventsRepository: self, *, event: Optional[str] = None, - events: Optional[Sequence[str]] = None, user_id: Optional[str] = None, - actor_id: Optional[str] = None, since=None, - until=None, - search: Optional[str] = None, ) -> int: """Total matching the same filters as :meth:`list_all` (for pagination).""" - where, params = self._filter_clauses( - event, - user_id, - since, - events=events, - actor_id=actor_id, - until=until, - search=search, - ) + where, params = self._filter_clauses(event, user_id, since) return int( self._conn.execute( text(f"SELECT count(*) FROM auth_events {where}"), params ).scalar() or 0 ) - - def event_names(self) -> list[str]: - """Distinct event names present in the table, alphabetically. - - Feeds the admin filter's event picker, so operators pick from what the - instance actually recorded instead of typing a name from memory. - """ - result = self._conn.execute( - text("SELECT DISTINCT event FROM auth_events ORDER BY event") - ) - return [row[0] for row in result.fetchall()] diff --git a/docsgpt/storage/db/repositories/token_usage.py b/docsgpt/storage/db/repositories/token_usage.py index b0b29e0d..6247bd9e 100644 --- a/docsgpt/storage/db/repositories/token_usage.py +++ b/docsgpt/storage/db/repositories/token_usage.py @@ -306,7 +306,18 @@ class TokenUsageRepository: 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 + 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} @@ -328,6 +339,9 @@ class TokenUsageRepository: 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 17adc24a..8ac6c436 100644 --- a/docsgpt/usage.py +++ b/docsgpt/usage.py @@ -305,12 +305,28 @@ 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 = time.monotonic() + first_chunk_at = pull_started + provider_seconds batch.append(r) yield r except Exception as exc: @@ -320,7 +336,7 @@ def stream_token_usage(func): error = exc raise finally: - duration_ms = int((time.monotonic() - started_at) * 1000) + 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 = ( @@ -343,7 +359,10 @@ def stream_token_usage(func): model, prompt_tokens=call_usage["prompt_tokens"], completion_tokens=call_usage["generated_tokens"], - latency_ms=duration_ms, + # 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"), error=error, diff --git a/frontend/src/admin/Activity.tsx b/frontend/src/admin/Activity.tsx index fc2f8ee3..c5494635 100644 --- a/frontend/src/admin/Activity.tsx +++ b/frontend/src/admin/Activity.tsx @@ -29,6 +29,7 @@ import { fmtNumber, fmtRelative, isLoopback, + outcomeLabel, outcomeTone, } from './AdminUI'; @@ -373,7 +374,7 @@ export default function Activity() { {row.outcome ? ( - {row.outcome} + {outcomeLabel(row.outcome)} ) : null} diff --git a/frontend/src/admin/AdminUI.test.ts b/frontend/src/admin/AdminUI.test.ts index 03a14186..26d74212 100644 --- a/frontend/src/admin/AdminUI.test.ts +++ b/frontend/src/admin/AdminUI.test.ts @@ -1,6 +1,12 @@ import { describe, expect, it } from 'vitest'; -import { categoryTone, eventLabel, eventTone, outcomeTone } from './AdminUI'; +import { + categoryTone, + eventLabel, + eventTone, + outcomeLabel, + outcomeTone, +} from './AdminUI'; describe('eventLabel', () => { it('humanizes the events it knows', () => { @@ -57,16 +63,32 @@ describe('categoryTone', () => { }); describe('outcomeTone', () => { - it('reads a block or a denial as dangerous', () => { - expect(outcomeTone('blocked')).toBe('danger'); - expect(outcomeTone('denied')).toBe('danger'); + // 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('reads an allowed command as a success', () => { - expect(outcomeTone('allowed')).toBe('success'); + 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('pending')).toBe('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 6efc5010..250a7410 100644 --- a/frontend/src/admin/AdminUI.tsx +++ b/frontend/src/admin/AdminUI.tsx @@ -223,14 +223,35 @@ export function categoryTone(category: string): Tone { return CATEGORY_TONES[category] ?? 'muted'; } -// A guardrail "blocked" or a device "denied" is the row's headline, so it gets -// the same treatment as a dangerous event name. -const DANGER_OUTCOMES = new Set(['blocked', 'denied', 'error', 'failed']); +// 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 { - if (DANGER_OUTCOMES.has(outcome)) return 'danger'; - if (outcome === 'allowed' || outcome === 'passed') return 'success'; - return 'muted'; + 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 diff --git a/frontend/src/admin/Usage.tsx b/frontend/src/admin/Usage.tsx index 20de3855..4904515b 100644 --- a/frontend/src/admin/Usage.tsx +++ b/frontend/src/admin/Usage.tsx @@ -171,7 +171,7 @@ export default function Usage() { sub={ cacheHitRate === null ? 'Provider reported no cache data' - : 'Of prompt tokens the provider reported' + : 'Of prompt tokens on calls that reported' } /> diff --git a/frontend/src/admin/UserUsageModal.tsx b/frontend/src/admin/UserUsageModal.tsx index 276e42b1..2a9f853b 100644 --- a/frontend/src/admin/UserUsageModal.tsx +++ b/frontend/src/admin/UserUsageModal.tsx @@ -29,10 +29,15 @@ 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. */ +/** + * 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', - schedule: 'Scheduled run', webhook: 'Webhook', workflow: 'Workflow', title: 'Title generation', diff --git a/frontend/src/admin/usageChartData.test.ts b/frontend/src/admin/usageChartData.test.ts index b596910c..a92de3ef 100644 --- a/frontend/src/admin/usageChartData.test.ts +++ b/frontend/src/admin/usageChartData.test.ts @@ -18,6 +18,7 @@ const bucket = (overrides: Partial): UsageBucket => ({ generated_tokens: 20, cost: 1, cached_tokens: null, + cache_eligible_prompt_tokens: 0, ...overrides, }); @@ -83,19 +84,40 @@ describe('cacheHitRate', () => { expect(cacheHitRate([])).toBeNull(); }); - it('measures only the rows that reported', () => { + 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: 100, cached_tokens: 25 }), - // Unreported: excluded entirely rather than counted as a 0% row, which - // would halve the rate. - bucket({ prompt_tokens: 100, cached_tokens: null }), + bucket({ + prompt_tokens: 200, + cached_tokens: 50, + cache_eligible_prompt_tokens: 100, + }), ]); - expect(rate).toBeCloseTo(25); + expect(rate).toBeCloseTo(50); }); - it('avoids dividing by zero prompt tokens', () => { + 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: 0, cached_tokens: 0 })]), + 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 index 7bc5a2ff..ee862e98 100644 --- a/frontend/src/admin/usageChartData.ts +++ b/frontend/src/admin/usageChartData.ts @@ -7,6 +7,8 @@ export type UsageBucket = { 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; }; @@ -94,20 +96,21 @@ export function buildUsageChart( } /** - * Prompt-cache hit rate over the rows whose provider reported a breakdown. + * Prompt-cache hit rate over the calls whose provider reported a breakdown. * - * NULL cached_tokens means "the provider said nothing", which is not the same - * as "no cache hits" — counting those rows as 0 would understate the rate. - * Returns null when nothing reported. + * 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 reported = series.filter((row) => row.cached_tokens !== null); - if (reported.length === 0) return null; - const prompt = reported.reduce((sum, row) => sum + row.prompt_tokens, 0); - if (prompt === 0) return null; - const cached = reported.reduce( - (sum, row) => sum + (row.cached_tokens ?? 0), + const eligible = series.reduce( + (sum, row) => sum + (row.cache_eligible_prompt_tokens ?? 0), 0, ); - return (cached / prompt) * 100; + if (eligible === 0) return null; + const cached = series.reduce((sum, row) => sum + (row.cached_tokens ?? 0), 0); + return (cached / eligible) * 100; } diff --git a/tests/api/test_admin_activity.py b/tests/api/test_admin_activity.py index c0929190..4a7fd1c9 100644 --- a/tests/api/test_admin_activity.py +++ b/tests/api/test_admin_activity.py @@ -129,19 +129,36 @@ class TestFeed: assert kwargs["until"] == datetime(2026, 2, 1, tzinfo=timezone.utc) assert kwargs["search"] == "handbook" - def test_unknown_facet_values_are_ignored_not_rejected(self, client): - repo = _repo() - with _admin(repo): - resp = client.get("/api/admin/activity?feed=bogus&category=bogus") - assert resp.status_code == 200 - kwargs = repo.list.call_args.kwargs - assert kwargs["feeds"] is None and kwargs["categories"] is None + def test_unknown_facet_is_refused_not_dropped(self, client): + """Dropping it would mean no filter, and no filter means everything. - def test_unparseable_time_is_ignored(self, client): + A typo or a stale chip must never widen an audit view. + """ repo = _repo() with _admin(repo): - assert client.get("/api/admin/activity?since=nonsense").status_code == 200 - assert repo.list.call_args.kwargs["since"] is None + resp = client.get("/api/admin/activity?category=bogus") + assert resp.status_code == 400 + assert "bogus" in json.loads(resp.data)["message"] + repo.list.assert_not_called() + + def test_unknown_feed_is_refused(self, client): + repo = _repo() + with _admin(repo): + assert client.get("/api/admin/activity?feed=bogus").status_code == 400 + repo.list.assert_not_called() + + def test_unparseable_time_is_refused(self, client): + """Ignoring it would drop the bound and return the full history.""" + repo = _repo() + with _admin(repo): + assert client.get("/api/admin/activity?since=nonsense").status_code == 400 + repo.list.assert_not_called() + + def test_a_valid_facet_still_passes_through(self, client): + repo = _repo() + with _admin(repo): + assert client.get("/api/admin/activity?category=data").status_code == 200 + assert repo.list.call_args.kwargs["categories"] == ["data"] def test_repeated_params_accumulate(self, client): repo = _repo() @@ -221,6 +238,15 @@ class TestExport: body = client.get("/api/admin/activity/export").get_data(as_text=True) assert body.strip().startswith("feed,id,event,category") + def test_export_refuses_an_unknown_facet(self, client): + """Failing open here would stream the whole history to the 100k cap.""" + repo = _repo() + with _admin(repo): + resp = client.get("/api/admin/activity/export?category=bogus") + resp.get_data() + assert resp.status_code == 400 + repo.iter_all.assert_not_called() + def test_rejects_an_unknown_format(self, client): with _admin(_repo()): assert ( diff --git a/tests/api/test_admin_dashboard.py b/tests/api/test_admin_dashboard.py index ad76ad9f..0ed98975 100644 --- a/tests/api/test_admin_dashboard.py +++ b/tests/api/test_admin_dashboard.py @@ -238,12 +238,10 @@ class TestUsageAndAudit: "generated_tokens": 3, "cost": 0.25, "cached_tokens": None, + "cache_eligible_prompt_tokens": 0, } ] usage.sum_tokens_in_range.return_value = 8 - usage.tokens_by_model.return_value = [ - {"model_id": "gpt-x", "tokens": 8, "cost": 0.25} - ] stats = Mock() stats.top_token_users.return_value = [ {"user_id": "a", "tokens": 8, "cost": 0.25} @@ -280,10 +278,20 @@ class TestUsageAndAudit: data = _body(client.get("/api/admin/usage")) assert data["total_cost"] == 0.25 assert data["series"][0]["cost"] == 0.25 - assert data["by_model"][0]["model_id"] == "gpt-x" assert data["latency"]["p95_ms"] == 400 assert data["top_users"][0]["cost"] == 0.25 + def test_usage_does_not_run_an_aggregate_nothing_reads(self, client): + """Per-model spend is served by the per-user endpoint and by group_by.""" + usage, stats = self._usage_repos() + with _admin( + TokenUsageRepository=Mock(return_value=usage), + AdminStatsRepository=Mock(return_value=stats), + ): + data = _body(client.get("/api/admin/usage")) + assert "by_model" not in data + usage.tokens_by_model.assert_not_called() + def test_usage_group_by_reaches_the_repository(self, client): usage, stats = self._usage_repos() with _admin( diff --git a/tests/api/test_data_plane_audit.py b/tests/api/test_data_plane_audit.py index 7a574988..b298917b 100644 --- a/tests/api/test_data_plane_audit.py +++ b/tests/api/test_data_plane_audit.py @@ -8,6 +8,7 @@ that the audit row is written with the acting user and the resource id. from __future__ import annotations +import json from contextlib import ExitStack, contextmanager from unittest.mock import MagicMock, Mock, patch @@ -79,6 +80,81 @@ class TestSourceAudit: assert recorded == [] +_UPLOAD = "docsgpt.api.user.sources.upload" + + +@pytest.mark.unit +class TestRemoteSourceAudit: + """``source.deleted`` is recorded for remote sources, so creation must be. + + A trail showing a source deleted that was apparently never created is + worse than no trail, and the docs state ``source.created`` is recorded. + """ + + @contextmanager + def _remote_env(self): + recorded: list[tuple] = [] + + def _record(_conn, event, *, actor, target=None, **metadata): + recorded.append((event, actor, target, metadata)) + + @contextmanager + def _conn(): + yield MagicMock() + + task = Mock(id="task-9") + with ExitStack() as stack: + stack.enter_context( + patch("docsgpt.app.handle_auth", return_value={"sub": "u1"}) + ) + stack.enter_context( + patch("docsgpt.app.resolve_roles", return_value=["user"]) + ) + stack.enter_context(patch(f"{_UPLOAD}.db_session", _conn)) + stack.enter_context(patch(f"{_UPLOAD}.record_event", _record)) + stack.enter_context( + patch( + f"{_UPLOAD}.ingest_remote", + Mock(apply_async=Mock(return_value=task)), + ) + ) + yield recorded + + def test_url_source_records_source_created(self, client): + # The route reads ``request.form``; ``data`` is a JSON string. + with self._remote_env() as recorded: + resp = client.post( + "/api/remote", + data={ + "user": "u1", + "source": "url", + "name": "Handbook", + "data": json.dumps({"url": "https://example.com"}), + }, + ) + assert resp.status_code == 200 + assert [row[0] for row in recorded] == ["source.created"] + event, actor, target, metadata = recorded[0] + assert (actor, target) == ("u1", None) + assert metadata["name"] == "Handbook" + assert metadata["type"] == "url" + assert metadata["source_id"] == json.loads(resp.data)["source_id"] + + def test_github_source_records_source_created(self, client): + with self._remote_env() as recorded: + resp = client.post( + "/api/remote", + data={ + "user": "u1", + "source": "github", + "name": "Repo", + "data": json.dumps({"repo_url": "https://github.com/a/b"}), + }, + ) + assert resp.status_code == 200 + assert recorded[0][3]["type"] == "github" + + @pytest.mark.unit class TestAgentAudit: def test_delete_records_agent_deleted(self, client): diff --git a/tests/storage/db/repositories/test_activity.py b/tests/storage/db/repositories/test_activity.py index a22b01f3..36b897e3 100644 --- a/tests/storage/db/repositories/test_activity.py +++ b/tests/storage/db/repositories/test_activity.py @@ -27,7 +27,7 @@ def seeded(pg_conn): "(user_id, api_key, stage, check_name, detector_type, action, " " outcome, matched_value, created_at) VALUES " "('u-guard', 'sk-secret', 'input', 'pii', 'regex', 'block', " - " 'blocked', 'alex@example.com', :ts)" + " 'triggered', 'alex@example.com', :ts)" ), {"ts": now - timedelta(minutes=10)}, ) @@ -44,7 +44,7 @@ def seeded(pg_conn): "(device_id, user_id, invocation_id, action, command, " " approval_mode, decision, issued_at, created_at) VALUES " "('dev-1', 'u-dev', 'inv-1', 'run_command', 'ls -la', " - " 'ask', 'allowed', :ts, :ts)" + " 'ask', 'dispatched', :ts, :ts)" ), {"ts": now - timedelta(minutes=5)}, ) @@ -84,9 +84,11 @@ class TestMergedFeed: } def test_outcome_carries_the_side_journal_verdict(self, repo): + # The values the writers actually produce: guardrails/runtime.py emits + # triggered/not_evaluated, remote_device.py emits dispatched. rows = {row["feed"]: row["outcome"] for row in repo.list()} - assert rows["device"] == "allowed" - assert rows["guardrail"] == "blocked" + assert rows["device"] == "dispatched" + assert rows["guardrail"] == "triggered" assert rows["auth"] is None @@ -132,6 +134,12 @@ class TestFilters: assert repo.count(search="203.0.113") == 1 assert repo.count(search="nope") == 0 + def test_search_treats_wildcards_literally(self, repo): + """A search box takes a substring; '%' must not match everything.""" + assert repo.count(search="%") == 0 + # '_' would otherwise match any single character, e.g. "local". + assert repo.count(search="l_cal") == 0 + def test_unknown_feed_yields_nothing(self, repo): assert repo.list(feeds=["made-up"]) == [] assert repo.count(feeds=["made-up"]) == 0 diff --git a/tests/storage/db/repositories/test_admin_usage_stats.py b/tests/storage/db/repositories/test_admin_usage_stats.py index 75a8eed2..b5c84a06 100644 --- a/tests/storage/db/repositories/test_admin_usage_stats.py +++ b/tests/storage/db/repositories/test_admin_usage_stats.py @@ -71,6 +71,31 @@ class TestBucketedCost: ) assert series[0]["cached_tokens"] == 30 + def test_cache_denominator_counts_only_reporting_rows(self, pg_conn, since): + """A hit rate needs the prompt tokens of the rows that reported. + + Both rows land in the same bucket, so filtering buckets client-side + cannot separate them: ``prompt_tokens`` there is the whole day. A rate + built on it is understated by however much traffic ran on a provider + that reports no cache breakdown. + """ + _insert(pg_conn, prompt_tokens=100, cached_tokens=50) + _insert(pg_conn, prompt_tokens=100, cached_tokens=None) + series = TokenUsageRepository(pg_conn).bucketed_totals( + bucket_unit="day", timestamp_gte=since + ) + assert series[0]["prompt_tokens"] == 200 + assert series[0]["cached_tokens"] == 50 + # 50 / 100 = 50%, not 50 / 200 = 25%. + assert series[0]["cache_eligible_prompt_tokens"] == 100 + + def test_cache_denominator_is_zero_when_nothing_reported(self, pg_conn, since): + _insert(pg_conn, prompt_tokens=100, cached_tokens=None) + series = TokenUsageRepository(pg_conn).bucketed_totals( + bucket_unit="day", timestamp_gte=since + ) + assert series[0]["cache_eligible_prompt_tokens"] == 0 + def test_grouping_keeps_cost_per_group(self, pg_conn, since): _insert(pg_conn, model_id="cheap", cost=0.1) _insert(pg_conn, model_id="pricey", cost=9.9) diff --git a/tests/storage/db/repositories/test_auth_events.py b/tests/storage/db/repositories/test_auth_events.py index d76fea9b..47f3e136 100644 --- a/tests/storage/db/repositories/test_auth_events.py +++ b/tests/storage/db/repositories/test_auth_events.py @@ -45,41 +45,52 @@ class TestInsertAttribution: class TestFeedFilters: + """``/api/admin/audit`` is the single-journal feed; the merged one has its + own repository, so the filter set here stays deliberately small.""" + def _seed(self, repo): repo.insert("a", "oidc_login") repo.insert("b", "oidc_login_denied") repo.insert("b", "admin_user_deactivated", actor_id="admin-9", target_id="b") - def test_filter_by_actor(self, repo): - self._seed(repo) - rows = repo.list_all(actor_id="admin-9") - assert [r["event"] for r in rows] == ["admin_user_deactivated"] - assert repo.count_all(actor_id="admin-9") == 1 - - def test_filter_by_multiple_events(self, repo): - self._seed(repo) - rows = repo.list_all(events=["oidc_login", "oidc_login_denied"]) - assert {r["event"] for r in rows} == {"oidc_login", "oidc_login_denied"} - assert repo.count_all(events=["oidc_login", "oidc_login_denied"]) == 2 - - def test_single_event_filter_still_supported(self, repo): + def test_filter_by_event(self, repo): self._seed(repo) + rows = repo.list_all(event="oidc_login") + assert [r["event"] for r in rows] == ["oidc_login"] assert repo.count_all(event="oidc_login") == 1 - def test_filter_by_until(self, repo): + def test_filter_by_user_matches_subject_or_target(self, repo): self._seed(repo) + assert repo.count_all(user_id="b") == 2 + + def test_filter_by_since(self, repo): + self._seed(repo) + future = datetime.now(timezone.utc) + timedelta(days=1) + assert repo.count_all(since=future) == 0 past = datetime.now(timezone.utc) - timedelta(days=1) - assert repo.count_all(until=past) == 0 - assert repo.count_all(until=datetime.now(timezone.utc) + timedelta(days=1)) == 3 + assert repo.count_all(since=past) == 3 - def test_search_matches_user_actor_and_metadata(self, repo): - repo.insert("someone", "pat_created", metadata={"name": "ci-deploy-key"}) - assert repo.count_all(search="ci-deploy") == 1 - assert repo.count_all(search="someone") == 1 - assert repo.count_all(search="nothing-here") == 0 - - def test_event_names_catalogue(self, repo): + def test_pagination(self, repo): self._seed(repo) - names = repo.event_names() - assert names == sorted(names) - assert "oidc_login" in names and "admin_user_deactivated" in names + assert len(repo.list_all(limit=2)) == 2 + assert len(repo.list_all(limit=2, offset=2)) == 1 + + +class TestPerUserPanel: + """The admin user drill-down is a security panel, not an activity log.""" + + def test_data_plane_events_can_be_excluded(self, repo): + repo.insert("u1", "oidc_login_denied") + for _ in range(5): + repo.insert("u1", "conversation.deleted", target_id=None, actor_id="u1") + rows = repo.list_recent( + "u1", limit=3, exclude_events_like=("source.", "agent.", "conversation.") + ) + # Without the exclusion the five deletes would fill the window and + # push the denied login out of it. + assert [r["event"] for r in rows] == ["oidc_login_denied"] + + def test_without_exclusions_everything_is_returned(self, repo): + repo.insert("u1", "oidc_login") + repo.insert("u1", "conversation.deleted", target_id=None, actor_id="u1") + assert len(repo.list_recent("u1")) == 2 diff --git a/tests/storage/db/repositories/test_token_usage.py b/tests/storage/db/repositories/test_token_usage.py index 67a62e19..db7692b3 100644 --- a/tests/storage/db/repositories/test_token_usage.py +++ b/tests/storage/db/repositories/test_token_usage.py @@ -301,6 +301,7 @@ class TestBucketedTotals: "generated_tokens": 12, "cost": 0.75, "cached_tokens": None, + "cache_eligible_prompt_tokens": 0, }, { "bucket": "2026-04-11", @@ -308,6 +309,7 @@ class TestBucketedTotals: "generated_tokens": 1, "cost": 0.0, "cached_tokens": None, + "cache_eligible_prompt_tokens": 0, }, ] diff --git a/tests/storage/db/test_migration_0034.py b/tests/storage/db/test_migration_0034.py index 7abf4d11..27aeb007 100644 --- a/tests/storage/db/test_migration_0034.py +++ b/tests/storage/db/test_migration_0034.py @@ -126,6 +126,25 @@ class TestMigration0034RoundTrip: # Team events were always filed under the actor; they have no user target. assert rows["team.create"] == "actor-3|-" + def test_actor_id_has_a_default_for_rolling_deploys(self, pg_engine): + """A previous-release process must not 500 mid-rollout. + + Its inserts name only the old columns; without a DEFAULT the NOT NULL + raises, and in admin/routes.py that insert shares the request's + transaction, so the role grant beside it would roll back too. + """ + with pg_engine.begin() as conn: + conn.execute( + text( + "INSERT INTO auth_events (user_id, event, metadata) " + "VALUES ('legacy', 'oidc_login', '{}'::jsonb)" + ) + ) + actor = conn.execute( + text("SELECT actor_id FROM auth_events WHERE user_id = 'legacy'") + ).scalar() + assert actor == "unknown" + def test_backfill_scopes_quota_policy_targets(self, pg_engine): """Only a user-scoped quota policy has a user target. diff --git a/tests/test_usage_latency.py b/tests/test_usage_latency.py index bf54a4ff..0bc1cc1e 100644 --- a/tests/test_usage_latency.py +++ b/tests/test_usage_latency.py @@ -83,6 +83,25 @@ class TestStreaming: 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):