Merge pull request #2822 from arc53-machine/feat/admin-activity-and-usage

feat(admin): activity feed, data-plane audit, and spend/latency in usage

tests/api/test_admin_activity.py

@@ -0,0 +1,285 @@
+"""Endpoint tests for the merged admin activity feed, its filters and export.
+
+The repository's SQL is covered by
+tests/storage/db/repositories/test_activity.py; these pin the route wiring,
+the @admin_required boundary, query-argument parsing and the export framing.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import json
+from contextlib import ExitStack, contextmanager
+from datetime import datetime, timezone
+from unittest.mock import Mock, patch
+
+import pytest
+
+
+@pytest.fixture
+def client():
+    from docsgpt.app import app as flask_app
+
+    flask_app.config["TESTING"] = True
+    return flask_app.test_client()
+
+
+@contextmanager
+def _fake_conn():
+    yield Mock()
+
+
+@contextmanager
+def _admin(repo: Mock):
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch("docsgpt.app.handle_auth", return_value={"sub": "admin1"})
+        )
+        stack.enter_context(
+            patch("docsgpt.app.resolve_roles", return_value=["admin", "user"])
+        )
+        stack.enter_context(patch("docsgpt.api.admin.activity.db_readonly", _fake_conn))
+        stack.enter_context(
+            patch(
+                "docsgpt.api.admin.activity.ActivityRepository", return_value=repo
+            )
+        )
+        yield
+
+
+def _row(**overrides):
+    base = {
+        "feed": "auth",
+        "id": "1",
+        "event": "oidc_login",
+        "category": "identity",
+        "actor_id": "u1",
+        "target_id": "u1",
+        "ip": "203.0.113.1",
+        "user_agent": "curl/8",
+        "outcome": None,
+        "detail": {"via": "oidc"},
+        "created_at": datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc),
+    }
+    base.update(overrides)
+    return base
+
+
+def _repo(rows=None, total=None):
+    repo = Mock()
+    rows = [_row()] if rows is None else rows
+    repo.list.return_value = rows
+    repo.count.return_value = len(rows) if total is None else total
+    repo.iter_all.return_value = iter(rows)
+    repo.event_names.return_value = [
+        {"event": "oidc_login", "category": "identity"},
+    ]
+    return repo
+
+
+@pytest.mark.unit
+class TestGuard:
+    def test_non_admin_forbidden(self, client):
+        with patch("docsgpt.app.handle_auth", return_value={"sub": "u"}), patch(
+            "docsgpt.app.resolve_roles", return_value=["user"]
+        ):
+            assert client.get("/api/admin/activity").status_code == 403
+            assert client.get("/api/admin/activity/export").status_code == 403
+            assert client.get("/api/admin/activity/events").status_code == 403
+
+    def test_unauthenticated(self, client):
+        with patch("docsgpt.app.handle_auth", return_value=None):
+            assert client.get("/api/admin/activity").status_code == 401
+
+
+@pytest.mark.unit
+class TestFeed:
+    def test_returns_rows_with_iso_timestamps(self, client):
+        repo = _repo()
+        with _admin(repo):
+            body = json.loads(client.get("/api/admin/activity").data)
+        assert body["success"] is True
+        assert body["activity"][0]["created_at"] == "2026-01-02T03:04:05+00:00"
+        assert body["total"] == 1
+        assert body["has_more"] is False
+
+    def test_has_more_when_the_page_is_short_of_the_total(self, client):
+        repo = _repo(total=500)
+        with _admin(repo):
+            body = json.loads(client.get("/api/admin/activity").data)
+        assert body["has_more"] is True
+
+    def test_filters_reach_the_repository(self, client):
+        repo = _repo()
+        with _admin(repo):
+            client.get(
+                "/api/admin/activity?feed=auth,device&category=data"
+                "&event=source.deleted&actor_id=admin1&user_id=u2"
+                "&since=2026-01-01T00:00:00Z&until=2026-02-01&search=handbook"
+            )
+        kwargs = repo.list.call_args.kwargs
+        assert kwargs["feeds"] == ["auth", "device"]
+        assert kwargs["categories"] == ["data"]
+        assert kwargs["events"] == ["source.deleted"]
+        assert kwargs["actor_id"] == "admin1"
+        assert kwargs["user_id"] == "u2"
+        assert kwargs["since"] == datetime(2026, 1, 1, tzinfo=timezone.utc)
+        # A naive bound is read as UTC.
+        assert kwargs["until"] == datetime(2026, 2, 1, tzinfo=timezone.utc)
+        assert kwargs["search"] == "handbook"
+
+    def test_unknown_facet_is_refused_not_dropped(self, client):
+        """Dropping it would mean no filter, and no filter means everything.
+
+        A typo or a stale chip must never widen an audit view.
+        """
+        repo = _repo()
+        with _admin(repo):
+            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()
+        with _admin(repo):
+            client.get("/api/admin/activity?category=data&category=access")
+        assert repo.list.call_args.kwargs["categories"] == ["data", "access"]
+
+    def test_search_is_length_capped(self, client):
+        repo = _repo()
+        with _admin(repo):
+            client.get("/api/admin/activity?search=" + "x" * 5000)
+        assert len(repo.list.call_args.kwargs["search"]) == 200
+
+
+@pytest.mark.unit
+class TestCatalogue:
+    def test_lists_events_categories_and_feeds(self, client):
+        repo = _repo()
+        with _admin(repo):
+            body = json.loads(client.get("/api/admin/activity/events").data)
+        assert body["events"] == [{"event": "oidc_login", "category": "identity"}]
+        assert "identity" in body["categories"]
+        assert body["feeds"] == ["auth", "device", "guardrail"]
+
+
+@pytest.mark.unit
+class TestExport:
+    """The export streams lazily, so every test consumes it inside the patch."""
+
+    def test_csv_has_a_header_and_one_row_per_event(self, client):
+        repo = _repo(rows=[_row(), _row(id="2", event="pat_created")])
+        with _admin(repo):
+            resp = client.get("/api/admin/activity/export?format=csv")
+            body = resp.get_data(as_text=True)
+        assert resp.status_code == 200
+        assert resp.mimetype == "text/csv"
+        assert "attachment; filename=" in resp.headers["Content-Disposition"]
+        rows = list(csv.DictReader(io.StringIO(body)))
+        assert [row["event"] for row in rows] == ["oidc_login", "pat_created"]
+        # ``detail`` is a JSON object, encoded into the cell.
+        assert json.loads(rows[0]["detail"]) == {"via": "oidc"}
+
+    def test_ndjson_emits_one_object_per_line(self, client):
+        repo = _repo(rows=[_row(), _row(id="2")])
+        with _admin(repo):
+            resp = client.get("/api/admin/activity/export?format=ndjson")
+            body = resp.get_data(as_text=True)
+        assert resp.mimetype == "application/x-ndjson"
+        lines = body.strip().split("\n")
+        assert [json.loads(line)["id"] for line in lines] == ["1", "2"]
+
+    def test_csv_cells_cannot_smuggle_a_spreadsheet_formula(self, client):
+        """``user_agent`` is an unauthenticated attacker's raw header."""
+        repo = _repo(
+            rows=[_row(user_agent="=cmd|'/c calc'!A0", actor_id="+1", ip="-2")]
+        )
+        with _admin(repo):
+            body = client.get(
+                "/api/admin/activity/export?format=csv"
+            ).get_data(as_text=True)
+        row = next(csv.DictReader(io.StringIO(body)))
+        assert row["user_agent"].startswith("'=")
+        assert row["actor_id"].startswith("'+")
+        assert row["ip"].startswith("'-")
+
+    def test_csv_leaves_ordinary_cells_alone(self, client):
+        with _admin(_repo()):
+            body = client.get(
+                "/api/admin/activity/export?format=csv"
+            ).get_data(as_text=True)
+        row = next(csv.DictReader(io.StringIO(body)))
+        assert row["user_agent"] == "curl/8"
+        assert row["event"] == "oidc_login"
+
+    def test_empty_csv_export_still_has_its_header(self, client):
+        with _admin(_repo(rows=[])):
+            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 (
+                client.get("/api/admin/activity/export?format=xlsx").status_code == 400
+            )
+
+    def test_export_applies_the_same_filters(self, client):
+        repo = _repo()
+        with _admin(repo):
+            client.get(
+                "/api/admin/activity/export?category=data&search=q"
+            ).get_data()
+        kwargs = repo.iter_all.call_args.kwargs
+        assert kwargs["categories"] == ["data"]
+        assert kwargs["search"] == "q"
+
+    def test_row_cap_is_bounded_and_reported(self, client):
+        repo = _repo()
+        with _admin(repo):
+            resp = client.get("/api/admin/activity/export?max_rows=10")
+            resp.get_data()
+        assert resp.headers["X-Export-Max-Rows"] == "10"
+        assert repo.iter_all.call_args.kwargs["max_rows"] == 10
+
+    def test_an_absurd_cap_clamps_to_the_ceiling(self, client):
+        repo = _repo()
+        with _admin(repo):
+            resp = client.get("/api/admin/activity/export?max_rows=999999999")
+            resp.get_data()
+        assert resp.headers["X-Export-Max-Rows"] == "100000"
+
+    def test_export_is_not_cached(self, client):
+        with _admin(_repo()):
+            resp = client.get("/api/admin/activity/export")
+            resp.get_data()
+        assert resp.headers["Cache-Control"] == "no-store"

tests/api/test_admin_dashboard.py

@@ -228,14 +228,35 @@ class TestUserLifecycle:
 
 @pytest.mark.unit
 class TestUsageAndAudit:
-    def test_usage(self, client):
+    @staticmethod
+    def _usage_repos():
         usage = Mock()
         usage.bucketed_totals.return_value = [
-            {"bucket": "2026-06-14", "prompt_tokens": 5, "generated_tokens": 3}
+            {
+                "bucket": "2026-06-14",
+                "prompt_tokens": 5,
+                "generated_tokens": 3,
+                "cost": 0.25,
+                "cached_tokens": None,
+                "cache_eligible_prompt_tokens": 0,
+            }
         ]
         usage.sum_tokens_in_range.return_value = 8
         stats = Mock()
-        stats.top_token_users.return_value = [{"user_id": "a", "tokens": 8}]
+        stats.top_token_users.return_value = [
+            {"user_id": "a", "tokens": 8, "cost": 0.25}
+        ]
+        stats.latency_summary.return_value = {
+            "samples": 2,
+            "p50_ms": 120,
+            "p95_ms": 400,
+            "ttft_samples": 2,
+            "ttft_p50_ms": 30,
+        }
+        return usage, stats
+
+    def test_usage(self, client):
+        usage, stats = self._usage_repos()
         with _admin(
             TokenUsageRepository=Mock(return_value=usage),
             AdminStatsRepository=Mock(return_value=stats),
@@ -247,6 +268,66 @@ class TestUsageAndAudit:
         assert len(data["series"]) == 1
         assert data["top_users"][0]["user_id"] == "a"
 
+    def test_usage_reports_spend_and_latency(self, client):
+        """Cost quotas were settable long before spend was visible."""
+        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 data["total_cost"] == 0.25
+        assert data["series"][0]["cost"] == 0.25
+        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(
+            TokenUsageRepository=Mock(return_value=usage),
+            AdminStatsRepository=Mock(return_value=stats),
+        ):
+            assert client.get("/api/admin/usage?group_by=model").status_code == 200
+        assert usage.bucketed_totals.call_args.kwargs["group_by"] == "model"
+
+    def test_usage_rejects_an_unknown_grouping(self, client):
+        with _admin():
+            assert client.get("/api/admin/usage?group_by=nonsense").status_code == 400
+
+    def test_per_user_usage_drilldown(self, client):
+        usage, stats = self._usage_repos()
+        stats.user_usage_breakdown.return_value = {
+            "totals": {"tokens": 8, "cost": 0.25, "calls": 1},
+            "by_model": [{"key": "gpt-x", "tokens": 8, "cost": 0.25}],
+            "by_source": [{"key": "agent_stream", "tokens": 8, "cost": 0.25}],
+        }
+        with _admin(
+            TokenUsageRepository=Mock(return_value=usage),
+            AdminStatsRepository=Mock(return_value=stats),
+        ):
+            data = _body(client.get("/api/admin/users/u1/usage?days=7"))
+        assert data["totals"]["cost"] == 0.25
+        assert data["by_source"][0]["key"] == "agent_stream"
+        assert len(data["series"]) == 1
+        assert usage.bucketed_totals.call_args.kwargs["user_id"] == "u1"
+
+    def test_per_user_usage_requires_admin(self, client):
+        with patch("docsgpt.app.handle_auth", return_value={"sub": "u1"}), patch(
+            "docsgpt.app.resolve_roles", return_value=["user"]
+        ):
+            assert client.get("/api/admin/users/u1/usage").status_code == 403
+
     def test_usage_invalid_bucket(self, client):
         with _admin():
             resp = client.get("/api/admin/usage?bucket=year")

tests/api/test_audit_helper.py

@@ -0,0 +1,124 @@
+"""Unit tests for the shared request-scoped audit helper."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+
+from docsgpt.api.audit import record_event
+from docsgpt.audit_events import ACTIVITY_CATEGORIES, category_for
+
+
+@pytest.fixture
+def app():
+    from docsgpt.app import app as flask_app
+
+    flask_app.config["TESTING"] = True
+    return flask_app
+
+
+@pytest.mark.unit
+class TestRecordEvent:
+    def test_writes_actor_target_and_request_context(self, app):
+        repo = Mock()
+        conn = MagicMock()
+        with app.test_request_context(
+            "/api/anything",
+            environ_base={"REMOTE_ADDR": "203.0.113.7"},
+            headers={"User-Agent": "curl/8"},
+        ):
+            with patch("docsgpt.api.audit.AuthEventsRepository", return_value=repo):
+                record_event(
+                    conn, "source.deleted", actor="u1", target=None, source_id="s1"
+                )
+        kwargs = repo.insert.call_args.kwargs
+        assert kwargs["event"] == "source.deleted"
+        assert kwargs["actor_id"] == "u1"
+        assert kwargs["target_id"] is None
+        assert kwargs["ip"] == "203.0.113.7"
+        assert kwargs["user_agent"] == "curl/8"
+        assert kwargs["metadata"] == {"source_id": "s1"}
+
+    def test_drops_none_metadata_values(self, app):
+        repo = Mock()
+        with app.test_request_context("/api/anything"):
+            with patch("docsgpt.api.audit.AuthEventsRepository", return_value=repo):
+                record_event(MagicMock(), "agent.created", actor="u1", name=None, id="a1")
+        assert repo.insert.call_args.kwargs["metadata"] == {"id": "a1"}
+
+    def test_insert_failure_never_raises(self, app):
+        """An audit write must not be able to fail the action it records."""
+        conn = Mock()
+        conn.begin_nested.side_effect = RuntimeError("boom")
+        with app.test_request_context("/api/anything"):
+            record_event(conn, "source.deleted", actor="u1")
+
+    def test_runs_in_a_savepoint(self, app):
+        """A poisoned audit insert must roll back alone, not the action."""
+        repo = Mock()
+        conn = MagicMock()
+        with app.test_request_context("/api/anything"):
+            with patch("docsgpt.api.audit.AuthEventsRepository", return_value=repo):
+                record_event(conn, "agent.deleted", actor="u1")
+        conn.begin_nested.assert_called_once()
+
+    def test_works_without_a_request_context(self, app):
+        """Celery tasks have no request; the row still records."""
+        repo = Mock()
+        with patch("docsgpt.api.audit.AuthEventsRepository", return_value=repo):
+            record_event(MagicMock(), "source.created", actor="u1")
+        kwargs = repo.insert.call_args.kwargs
+        assert kwargs["ip"] is None and kwargs["user_agent"] is None
+
+    def test_anonymous_actor_falls_back(self, app):
+        repo = Mock()
+        with app.test_request_context("/api/anything"):
+            with patch("docsgpt.api.audit.AuthEventsRepository", return_value=repo):
+                record_event(MagicMock(), "agent.created", actor=None)
+        assert repo.insert.call_args.kwargs["actor_id"] == "unknown"
+
+
+@pytest.mark.unit
+class TestCategories:
+    @pytest.mark.parametrize(
+        "event,expected",
+        [
+            ("oidc_login", "identity"),
+            ("oidc_login_denied", "identity"),
+            ("pat_created", "identity"),
+            ("scim_created", "identity"),
+            ("role_granted", "access"),
+            ("admin_user_deactivated", "access"),
+            ("team.member_add", "access"),
+            ("quota_policy_set", "config"),
+            ("source.deleted", "data"),
+            ("agent.created", "data"),
+            ("conversation.deleted", "data"),
+        ],
+    )
+    def test_known_events(self, event, expected):
+        assert category_for(event) == expected
+
+    def test_unknown_event_is_other(self):
+        assert category_for("something_new") == "other"
+
+    @pytest.mark.parametrize(
+        "event,expected",
+        [("device.run_command", "device"), ("guardrail.input", "safety")],
+    )
+    def test_merged_feed_namespaces(self, event, expected):
+        assert category_for(event) == expected
+
+    def test_every_produced_category_is_declared(self):
+        assert "other" in ACTIVITY_CATEGORIES
+        for event in (
+            "oidc_login",
+            "role_granted",
+            "quota_policy_set",
+            "source.deleted",
+            "device.run_command",
+            "guardrail.input",
+            "mystery",
+        ):
+            assert category_for(event) in ACTIVITY_CATEGORIES

tests/api/test_data_plane_audit.py

@@ -0,0 +1,218 @@
+"""The data-plane routes must leave an audit trail.
+
+Identity events were audited from the start; source/agent/conversation
+mutations were not, so "who deleted that source" had no answer. These pin the
+hooks at the route layer — the repositories are mocked, what is asserted is
+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
+
+import pytest
+
+
+@pytest.fixture
+def client():
+    from docsgpt.app import app as flask_app
+
+    flask_app.config["TESTING"] = True
+    return flask_app.test_client()
+
+
+@contextmanager
+def _authed(module: str, **patches):
+    """Authenticate as ``u1`` and patch names inside ``module``."""
+    recorded: list[tuple] = []
+
+    def _record(_conn, event, *, actor, target=None, **metadata):
+        recorded.append((event, actor, target, metadata))
+
+    @contextmanager
+    def _conn():
+        yield MagicMock()
+
+    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"{module}.db_session", _conn))
+        stack.enter_context(patch(f"{module}.db_readonly", _conn))
+        stack.enter_context(patch(f"{module}.record_event", _record))
+        for name, value in patches.items():
+            stack.enter_context(patch(f"{module}.{name}", value))
+        yield recorded
+
+
+_SOURCES = "docsgpt.api.user.sources.routes"
+_AGENTS = "docsgpt.api.user.agents.routes"
+_CONVERSATIONS = "docsgpt.api.user.conversations.routes"
+
+
+@pytest.mark.unit
+class TestSourceAudit:
+    def test_delete_records_source_deleted(self, client):
+        repo = Mock()
+        repo.get_any.return_value = {"id": "src-1", "name": "Handbook"}
+        storage = Mock()
+        storage.file_exists.return_value = False
+        with _authed(
+            _SOURCES,
+            SourcesRepository=Mock(return_value=repo),
+            StorageCreator=Mock(get_storage=Mock(return_value=storage)),
+        ) as recorded:
+            resp = client.get("/api/delete_old?source_id=src-1")
+        assert resp.status_code == 200
+        assert recorded[0][:3] == ("source.deleted", "u1", None)
+        assert recorded[0][3]["source_id"] == "src-1"
+        assert recorded[0][3]["name"] == "Handbook"
+
+    def test_nothing_recorded_when_the_source_is_missing(self, client):
+        repo = Mock()
+        repo.get_any.return_value = None
+        with _authed(_SOURCES, SourcesRepository=Mock(return_value=repo)) as recorded:
+            resp = client.get("/api/delete_old?source_id=nope")
+        assert resp.status_code == 404
+        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):
+        repo = Mock()
+        repo.get_any.return_value = {
+            "id": "agent-1",
+            "name": "Support bot",
+            "agent_type": "classic",
+        }
+        with _authed(
+            _AGENTS,
+            AgentsRepository=Mock(return_value=repo),
+            WorkflowsRepository=Mock(),
+            UsersRepository=Mock(),
+        ) as recorded:
+            resp = client.delete("/api/delete_agent?id=agent-1")
+        assert resp.status_code == 200
+        assert recorded[0][:3] == ("agent.deleted", "u1", None)
+        assert recorded[0][3]["agent_id"] == "agent-1"
+        assert recorded[0][3]["name"] == "Support bot"
+
+    def test_missing_agent_records_nothing(self, client):
+        repo = Mock()
+        repo.get_any.return_value = None
+        with _authed(_AGENTS, AgentsRepository=Mock(return_value=repo)) as recorded:
+            assert client.delete("/api/delete_agent?id=x").status_code == 404
+        assert recorded == []
+
+
+@pytest.mark.unit
+class TestConversationAudit:
+    def test_delete_records_conversation_deleted(self, client):
+        repo = Mock()
+        repo.get_any.return_value = {"id": "conv-1"}
+        with _authed(
+            _CONVERSATIONS, ConversationsRepository=Mock(return_value=repo)
+        ) as recorded:
+            resp = client.post("/api/delete_conversation?id=conv-1")
+        assert resp.status_code == 200
+        assert recorded[0][:3] == ("conversation.deleted", "u1", None)
+        assert recorded[0][3]["conversation_id"] == "conv-1"
+
+    def test_delete_all_records_the_count(self, client):
+        repo = Mock()
+        repo.delete_all_for_user.return_value = 7
+        with _authed(
+            _CONVERSATIONS, ConversationsRepository=Mock(return_value=repo)
+        ) as recorded:
+            resp = client.get("/api/delete_all_conversations")
+        assert resp.status_code == 200
+        assert recorded[0][0] == "conversation.deleted_all"
+        assert recorded[0][3]["deleted"] == 7
+
+    def test_unknown_conversation_records_nothing(self, client):
+        repo = Mock()
+        repo.get_any.return_value = None
+        with _authed(
+            _CONVERSATIONS, ConversationsRepository=Mock(return_value=repo)
+        ) as recorded:
+            assert client.post("/api/delete_conversation?id=x").status_code == 200
+        assert recorded == []

tests/llm/handlers/test_round_usage_persistence.py

@@ -167,11 +167,11 @@ class TestPerRoundUsageRows:
             return messages, None
 
         rows = []
-        with patch.object(
-            usage_mod,
-            "_persist_call_usage",
-            side_effect=lambda llm_, cu: rows.append(dict(cu)),
-        ):
+
+        def _record(llm_, cu, *, duration_ms=None, ttft_ms=None):
+            rows.append({**cu, "duration_ms": duration_ms, "ttft_ms": ttft_ms})
+
+        with patch.object(usage_mod, "_persist_call_usage", side_effect=_record):
             with patch.object(handler, "handle_tool_calls", fake_tool_calls):
                 first = llm.gen_stream(
                     model="m", messages=[{"role": "user", "content": "hi"}]
@@ -185,6 +185,9 @@ class TestPerRoundUsageRows:
         counts = lambda row: {k: row[k] for k in ("prompt_tokens", "generated_tokens")}  # noqa: E731
         assert counts(rows[0]) == {"prompt_tokens": 100, "generated_tokens": 10}
         assert counts(rows[1]) == {"prompt_tokens": 200, "generated_tokens": 20}
+        # Each round is timed independently, and each streamed a chunk.
+        assert all(row["duration_ms"] is not None for row in rows)
+        assert all(row["ttft_ms"] is not None for row in rows)
 
 
 class TestPreferProviderUsageClaim:

tests/storage/db/repositories/test_activity.py

@@ -0,0 +1,173 @@
+"""Repository tests for the merged admin activity feed."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import text
+
+from docsgpt.storage.db.repositories.activity import (
+    ACTIVITY_COLUMNS,
+    ActivityRepository,
+)
+from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository
+
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture()
+def seeded(pg_conn):
+    """One row in each journal, oldest to newest: guardrail, device, auth."""
+    now = datetime.now(timezone.utc)
+    pg_conn.execute(
+        text(
+            "INSERT INTO guardrail_events "
+            "(user_id, api_key, stage, check_name, detector_type, action, "
+            " outcome, matched_value, created_at) VALUES "
+            "('u-guard', 'sk-secret', 'input', 'pii', 'regex', 'block', "
+            " 'triggered', 'alex@example.com', :ts)"
+        ),
+        {"ts": now - timedelta(minutes=10)},
+    )
+    pg_conn.execute(
+        text(
+            "INSERT INTO devices "
+            "(id, user_id, name, machine_pubkey_fingerprint, token_hash, status) "
+            "VALUES ('dev-1', 'u-dev', 'laptop', 'fp-1', 'hash-1', 'active')"
+        )
+    )
+    pg_conn.execute(
+        text(
+            "INSERT INTO device_audit_log "
+            "(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', 'dispatched', :ts, :ts)"
+        ),
+        {"ts": now - timedelta(minutes=5)},
+    )
+    AuthEventsRepository(pg_conn).insert(
+        "victim", "admin_user_deactivated", ip="203.0.113.9",
+        actor_id="admin-1", target_id="victim",
+    )
+    return pg_conn
+
+
+@pytest.fixture()
+def repo(seeded):
+    return ActivityRepository(seeded)
+
+
+class TestMergedFeed:
+    def test_spans_all_three_journals_newest_first(self, repo):
+        rows = repo.list()
+        assert [row["feed"] for row in rows] == ["auth", "device", "guardrail"]
+        assert repo.count() == 3
+
+    def test_every_row_has_the_common_shape(self, repo):
+        for row in repo.list():
+            assert set(row) == set(ACTIVITY_COLUMNS)
+
+    def test_side_journals_get_namespaced_event_names(self, repo):
+        events = {row["feed"]: row["event"] for row in repo.list()}
+        assert events["device"] == "device.run_command"
+        assert events["guardrail"] == "guardrail.input"
+
+    def test_categories_are_derived(self, repo):
+        rows = {row["feed"]: row["category"] for row in repo.list()}
+        assert rows == {
+            "auth": "access",
+            "device": "device",
+            "guardrail": "safety",
+        }
+
+    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"] == "dispatched"
+        assert rows["guardrail"] == "triggered"
+        assert rows["auth"] is None
+
+
+class TestSecretExclusion:
+    def test_guardrail_secrets_never_reach_the_feed(self, repo):
+        """``api_key`` is a raw agent key and ``matched_value`` is source text."""
+        row = next(r for r in repo.list() if r["feed"] == "guardrail")
+        serialized = str(row)
+        assert "sk-secret" not in serialized
+        assert "alex@example.com" not in serialized
+
+
+class TestFilters:
+    def test_by_feed(self, repo):
+        assert [r["feed"] for r in repo.list(feeds=["device"])] == ["device"]
+
+    def test_by_category(self, repo):
+        rows = repo.list(categories=["safety"])
+        assert [r["feed"] for r in rows] == ["guardrail"]
+        assert repo.count(categories=["safety"]) == 1
+
+    def test_by_category_spanning_auth_and_a_side_journal(self, repo):
+        rows = repo.list(categories=["access", "device"])
+        assert {r["feed"] for r in rows} == {"auth", "device"}
+
+    def test_by_actor(self, repo):
+        assert [r["actor_id"] for r in repo.list(actor_id="admin-1")] == ["admin-1"]
+
+    def test_by_user_matches_actor_or_target(self, repo):
+        assert repo.count(user_id="victim") == 1
+        assert repo.count(user_id="admin-1") == 1
+
+    def test_by_event(self, repo):
+        assert repo.count(events=["device.run_command"]) == 1
+
+    def test_by_time_window(self, repo):
+        now = datetime.now(timezone.utc)
+        assert repo.count(since=now - timedelta(minutes=7)) == 2
+        assert repo.count(until=now - timedelta(minutes=7)) == 1
+
+    def test_search_spans_detail(self, repo):
+        assert repo.count(search="ls -la") == 1
+        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
+
+    def test_category_no_journal_can_produce(self, repo):
+        assert repo.count(feeds=["device"], categories=["data"]) == 0
+
+
+class TestPagination:
+    def test_limit_and_offset(self, repo):
+        assert len(repo.list(limit=2)) == 2
+        assert [r["feed"] for r in repo.list(limit=2, offset=2)] == ["guardrail"]
+
+
+class TestExportStream:
+    def test_iter_all_walks_every_row(self, repo):
+        assert len(list(repo.iter_all(chunk_size=1))) == 3
+
+    def test_iter_all_honours_max_rows(self, repo):
+        assert len(list(repo.iter_all(chunk_size=1, max_rows=2))) == 2
+
+    def test_iter_all_applies_filters(self, repo):
+        rows = list(repo.iter_all(chunk_size=1, feeds=["auth"]))
+        assert [r["feed"] for r in rows] == ["auth"]
+
+
+class TestCatalogue:
+    def test_event_names_are_paired_with_categories(self, repo):
+        catalogue = repo.event_names()
+        assert {"event": "device.run_command", "category": "device"} in catalogue
+        assert {"event": "admin_user_deactivated", "category": "access"} in catalogue

tests/storage/db/repositories/test_admin_usage_stats.py

@@ -0,0 +1,190 @@
+"""Repository tests for admin spend, latency and the per-user drill-down."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import text
+
+from docsgpt.storage.db.repositories.admin_stats import AdminStatsRepository
+from docsgpt.storage.db.repositories.token_usage import TokenUsageRepository
+
+
+pytestmark = pytest.mark.integration
+
+
+def _insert(conn, **fields):
+    defaults = {
+        "user_id": "u1",
+        "prompt_tokens": 100,
+        "generated_tokens": 20,
+        "cost": 0.5,
+        "source": "agent_stream",
+        "model_id": "gpt-x",
+        "cached_tokens": None,
+        "duration_ms": None,
+        "ttft_ms": None,
+        "timestamp": datetime.now(timezone.utc),
+    }
+    defaults.update(fields)
+    conn.execute(
+        text(
+            "INSERT INTO token_usage (user_id, prompt_tokens, generated_tokens, "
+            "cost, source, model_id, cached_tokens, duration_ms, ttft_ms, timestamp) "
+            "VALUES (:user_id, :prompt_tokens, :generated_tokens, :cost, :source, "
+            ":model_id, :cached_tokens, :duration_ms, :ttft_ms, :timestamp)"
+        ),
+        defaults,
+    )
+
+
+@pytest.fixture()
+def since():
+    return datetime.now(timezone.utc) - timedelta(days=7)
+
+
+class TestBucketedCost:
+    def test_buckets_carry_cost_alongside_tokens(self, pg_conn, since):
+        _insert(pg_conn, cost=1.25)
+        _insert(pg_conn, cost=0.75)
+        series = TokenUsageRepository(pg_conn).bucketed_totals(
+            bucket_unit="day", timestamp_gte=since
+        )
+        assert len(series) == 1
+        assert series[0]["cost"] == pytest.approx(2.0)
+        assert series[0]["prompt_tokens"] == 200
+
+    def test_unreported_cache_bins_stay_unknown(self, pg_conn, since):
+        """NULL means the provider said nothing, which is not the same as 0."""
+        _insert(pg_conn, cached_tokens=None)
+        series = TokenUsageRepository(pg_conn).bucketed_totals(
+            bucket_unit="day", timestamp_gte=since
+        )
+        assert series[0]["cached_tokens"] is None
+
+    def test_reported_cache_bins_sum(self, pg_conn, since):
+        _insert(pg_conn, cached_tokens=30)
+        _insert(pg_conn, cached_tokens=None)
+        series = TokenUsageRepository(pg_conn).bucketed_totals(
+            bucket_unit="day", timestamp_gte=since
+        )
+        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)
+        series = TokenUsageRepository(pg_conn).bucketed_totals(
+            bucket_unit="day", timestamp_gte=since, group_by="model"
+        )
+        costs = {row["group_key"]: row["cost"] for row in series}
+        assert costs == {"cheap": pytest.approx(0.1), "pricey": pytest.approx(9.9)}
+
+
+class TestTopUsers:
+    def test_reports_spend_next_to_tokens(self, pg_conn, since):
+        _insert(pg_conn, user_id="heavy", prompt_tokens=1000, cost=2.0)
+        _insert(pg_conn, user_id="light", prompt_tokens=10, cost=0.1)
+        rows = AdminStatsRepository(pg_conn).top_token_users(since=since)
+        assert [row["user_id"] for row in rows] == ["heavy", "light"]
+        assert rows[0]["cost"] == pytest.approx(2.0)
+
+    def test_rollup_rows_are_not_billed_twice(self, pg_conn, since):
+        _insert(pg_conn, user_id="u1", source="agent_stream", cost=1.0)
+        _insert(pg_conn, user_id="u1", source="schedule", cost=1.0)
+        rows = AdminStatsRepository(pg_conn).top_token_users(since=since)
+        assert rows[0]["cost"] == pytest.approx(1.0)
+
+
+class TestLatencySummary:
+    def test_percentiles_over_measured_calls(self, pg_conn, since):
+        for duration in (100, 200, 300, 400):
+            _insert(pg_conn, duration_ms=duration, ttft_ms=duration // 10)
+        summary = AdminStatsRepository(pg_conn).latency_summary(since=since)
+        assert summary["samples"] == 4
+        assert summary["p50_ms"] == 250
+        assert summary["p95_ms"] == 385
+        assert summary["ttft_p50_ms"] == 25
+
+    def test_unmeasured_rows_are_skipped_not_counted_as_zero(self, pg_conn, since):
+        _insert(pg_conn, duration_ms=1000, ttft_ms=None)
+        _insert(pg_conn, duration_ms=None, ttft_ms=None)
+        summary = AdminStatsRepository(pg_conn).latency_summary(since=since)
+        assert summary["samples"] == 1
+        assert summary["p50_ms"] == 1000
+        # A stream that never yielded contributes no first-token sample.
+        assert summary["ttft_samples"] == 0
+        assert summary["ttft_p50_ms"] is None
+
+    def test_no_data_reports_no_samples(self, pg_conn, since):
+        summary = AdminStatsRepository(pg_conn).latency_summary(since=since)
+        assert summary == {
+            "samples": 0,
+            "p50_ms": None,
+            "p95_ms": None,
+            "ttft_samples": 0,
+            "ttft_p50_ms": None,
+        }
+
+
+class TestUserUsageBreakdown:
+    def test_totals_and_splits(self, pg_conn, since):
+        _insert(pg_conn, user_id="u1", model_id="a", source="agent_stream", cost=1.0)
+        _insert(pg_conn, user_id="u1", model_id="b", source="webhook", cost=2.0)
+        _insert(pg_conn, user_id="other", model_id="a", cost=99.0)
+        detail = AdminStatsRepository(pg_conn).user_usage_breakdown("u1", since=since)
+        assert detail["totals"]["calls"] == 2
+        assert detail["totals"]["cost"] == pytest.approx(3.0)
+        assert {row["key"] for row in detail["by_model"]} == {"a", "b"}
+        assert {row["key"] for row in detail["by_source"]} == {
+            "agent_stream",
+            "webhook",
+        }
+
+    def test_rollup_rows_are_not_billed_twice(self, pg_conn, since):
+        """A scheduled run's rollup duplicates its own per-call rows."""
+        _insert(pg_conn, user_id="u1", source="agent_stream", cost=1.0)
+        _insert(pg_conn, user_id="u1", source="schedule", cost=1.0)
+        detail = AdminStatsRepository(pg_conn).user_usage_breakdown("u1", since=since)
+        assert detail["totals"]["calls"] == 1
+        assert detail["totals"]["cost"] == pytest.approx(1.0)
+        assert [row["key"] for row in detail["by_source"]] == ["agent_stream"]
+
+    def test_a_user_with_no_usage_reports_zeroes(self, pg_conn, since):
+        detail = AdminStatsRepository(pg_conn).user_usage_breakdown("ghost", since=since)
+        assert detail["totals"] == {"tokens": 0, "cost": 0.0, "calls": 0}
+        assert detail["by_model"] == [] and detail["by_source"] == []
+
+    def test_window_is_respected(self, pg_conn, since):
+        _insert(
+            pg_conn,
+            user_id="u1",
+            timestamp=datetime.now(timezone.utc) - timedelta(days=30),
+        )
+        detail = AdminStatsRepository(pg_conn).user_usage_breakdown("u1", since=since)
+        assert detail["totals"]["calls"] == 0

tests/storage/db/repositories/test_auth_events.py

@@ -0,0 +1,96 @@
+"""Repository tests for ``auth_events`` (actor/target attribution + feed filters)."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository
+
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture()
+def repo(pg_conn):
+    return AuthEventsRepository(pg_conn)
+
+
+class TestInsertAttribution:
+    def test_self_service_event_is_its_own_actor(self, repo):
+        row = repo.insert("u1", "oidc_login")
+        assert row["actor_id"] == "u1"
+        assert row["target_id"] == "u1"
+        assert row["user_id"] == "u1"
+
+    def test_admin_action_separates_actor_from_target(self, repo):
+        row = repo.insert(
+            "victim", "admin_user_deactivated", actor_id="admin-1", target_id="victim"
+        )
+        assert row["actor_id"] == "admin-1"
+        assert row["target_id"] == "victim"
+        # ``user_id`` stays the subject, so the per-user feed is unchanged.
+        assert row["user_id"] == "victim"
+
+    def test_actor_only_event_has_no_target(self, repo):
+        row = repo.insert("admin-1", "team.create", actor_id="admin-1", target_id=None)
+        assert row["actor_id"] == "admin-1"
+        assert row["target_id"] is None
+
+    def test_positional_user_id_still_supported(self, repo):
+        """Existing call sites pass only ``user_id``; it must keep working."""
+        row = repo.insert("u2", "pat_created", metadata={"token_id": "t"})
+        assert (row["actor_id"], row["target_id"]) == ("u2", "u2")
+
+
+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_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_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(since=past) == 3
+
+    def test_pagination(self, repo):
+        self._seed(repo)
+        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

tests/storage/db/repositories/test_token_usage.py

@@ -279,8 +279,12 @@ class TestBucketedTotals:
         t1 = datetime(2026, 4, 10, 10, 0, tzinfo=timezone.utc)
         t2 = datetime(2026, 4, 10, 23, 30, tzinfo=timezone.utc)
         t3 = datetime(2026, 4, 11, 0, 15, tzinfo=timezone.utc)
-        repo.insert(user_id="u-day", prompt_tokens=10, generated_tokens=5, timestamp=t1)
-        repo.insert(user_id="u-day", prompt_tokens=20, generated_tokens=7, timestamp=t2)
+        repo.insert(
+            user_id="u-day", prompt_tokens=10, generated_tokens=5, cost=0.5, timestamp=t1
+        )
+        repo.insert(
+            user_id="u-day", prompt_tokens=20, generated_tokens=7, cost=0.25, timestamp=t2
+        )
         repo.insert(user_id="u-day", prompt_tokens=1, generated_tokens=1, timestamp=t3)
         rows = repo.bucketed_totals(
             bucket_unit="day",
@@ -288,9 +292,25 @@ class TestBucketedTotals:
             timestamp_gte=datetime(2026, 4, 10, tzinfo=timezone.utc),
             timestamp_lt=datetime(2026, 4, 12, tzinfo=timezone.utc),
         )
+        # ``cached_tokens`` is None when no row in the bucket reported a cache
+        # breakdown: "the provider said nothing", not "no cache hits".
         assert rows == [
-            {"bucket": "2026-04-10", "prompt_tokens": 30, "generated_tokens": 12},
-            {"bucket": "2026-04-11", "prompt_tokens": 1, "generated_tokens": 1},
+            {
+                "bucket": "2026-04-10",
+                "prompt_tokens": 30,
+                "generated_tokens": 12,
+                "cost": 0.75,
+                "cached_tokens": None,
+                "cache_eligible_prompt_tokens": 0,
+            },
+            {
+                "bucket": "2026-04-11",
+                "prompt_tokens": 1,
+                "generated_tokens": 1,
+                "cost": 0.0,
+                "cached_tokens": None,
+                "cache_eligible_prompt_tokens": 0,
+            },
         ]
 
     def test_hour_bucket(self, pg_conn):

tests/storage/db/test_migration_0034.py

@@ -0,0 +1,218 @@
+"""Migration round-trip test for 0034_auth_events_actor_target."""
+
+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_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"

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)

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

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