feat(audit): record data-plane events, not just identity ones

Logins, role grants and provisioning were audited from the first release;
creating and deleting sources, agents, agent keys and conversations were not.
An operator reviewing the trail could see who signed in but not who deleted
the source they were asking about.

Adds docsgpt/api/audit.py: one helper that records an event inside the
caller's transaction, in a savepoint, swallowing failures -- an audit write
must never be able to fail the action it describes -- and tolerating the
absence of a Flask request so a Celery task can record too.

Events added: source.created (upload and wiki), source.deleted,
source.reingested, agent.created, agent.updated, agent.deleted,
agent.key_regenerated, conversation.deleted, conversation.deleted_all.
agent.updated records field names only, never values, which can carry prompts
and credentials.

The same module carries the event -> category map (identity / access / config
/ data) that the admin activity feed filters on.

docsgpt/api/audit.py

@@ -0,0 +1,121 @@
+"""Shared audit-trail helper and the event → category map.
+
+Identity and access events (login, role grants, provisioning) have been
+audited since ``auth_events`` was introduced. Data-plane actions — creating and
+deleting sources, agents, agent keys and conversations — were not, which left
+the trail unable to answer "who deleted that source". :func:`record_event` is
+the one-line hook those routes use.
+
+Two properties matter and are enforced here rather than at every call site:
+
+1. **An audit write can never fail the action it records.** The insert runs in
+   a SAVEPOINT so a rejected row rolls back alone, and any exception is
+   swallowed with a log line.
+2. **Request context is optional.** Celery tasks (ingestion finishing, a
+   scheduled run) have no Flask request; the row simply records without an IP
+   or user agent instead of raising.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Optional
+
+from flask import has_request_context, request
+
+from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository
+
+logger = logging.getLogger(__name__)
+
+
+# Coarse grouping for the admin activity feed's facet filter. Keyed by exact
+# event name or by a ``prefix.`` namespace; see :func:`category_for`.
+EVENT_CATEGORIES: dict[str, str] = {
+    # Who you are: authentication, provisioning, personal credentials.
+    "oidc_login": "identity",
+    "oidc_login_denied": "identity",
+    "oidc_refresh": "identity",
+    "backchannel_logout": "identity",
+    "scim_created": "identity",
+    "scim_deactivated": "identity",
+    "scim_reactivated": "identity",
+    "pat_created": "identity",
+    "pat_revoked": "identity",
+    "pat_regenerated": "identity",
+    # What you may do: roles, account state, team membership and sharing.
+    "role_granted": "access",
+    "role_revoked": "access",
+    "admin_user_activated": "access",
+    "admin_user_deactivated": "access",
+    "admin_sessions_revoked": "access",
+    "team.": "access",
+    # How the instance is configured.
+    "quota_policy_set": "config",
+    "quota_policy_deleted": "config",
+    # What the data looks like.
+    "source.": "data",
+    "agent.": "data",
+    "conversation.": "data",
+    # Everything an older release wrote that this one does not know about.
+    "other": "other",
+}
+
+
+def category_for(event: str) -> str:
+    """Return the activity category for ``event``.
+
+    Args:
+        event: The audit event name.
+
+    Returns:
+        One of ``identity``, ``access``, ``config``, ``data``, or ``other``
+        for names this release does not recognise.
+    """
+    exact = EVENT_CATEGORIES.get(event)
+    if exact is not None and not event.endswith("."):
+        return exact
+    namespace, dot, _ = event.partition(".")
+    if dot:
+        return EVENT_CATEGORIES.get(f"{namespace}.", "other")
+    return "other"
+
+
+def record_event(
+    conn,
+    event: str,
+    *,
+    actor: Optional[str],
+    target: Optional[str] = None,
+    **metadata: Any,
+) -> None:
+    """Append one audit event, best-effort, inside the caller's transaction.
+
+    Args:
+        conn: Open database connection, inside the action's transaction so the
+            audit row commits atomically with the change it describes.
+        event: Dotted event name (``source.deleted``, ``agent.created``).
+        actor: Who performed the action; ``"unknown"`` when unauthenticated.
+        target: The user acted upon, when the event is about a user. Data-plane
+            events act on a resource, not an account, so they leave this None.
+        **metadata: Free-form detail for the audit drill-down. ``None`` values
+            are dropped so the stored object only carries what was known.
+    """
+    detail = {key: value for key, value in metadata.items() if value is not None}
+    ip = request.remote_addr if has_request_context() else None
+    user_agent = request.headers.get("User-Agent") if has_request_context() else None
+    resolved_actor = actor or "unknown"
+    try:
+        with conn.begin_nested():
+            AuthEventsRepository(conn).insert(
+                # ``user_id`` is the per-user feed key: the target when the
+                # event is about someone, else the actor's own trail.
+                user_id=target or resolved_actor,
+                event=event,
+                ip=ip,
+                user_agent=user_agent,
+                metadata=detail,
+                actor_id=resolved_actor,
+                target_id=target,
+            )
+    except Exception:
+        logger.warning("audit insert failed for event=%s", event, exc_info=True)

docsgpt/api/user/agents/routes.py

@@ -9,6 +9,7 @@ from flask_restx import fields, Namespace, Resource
 from pydantic import ValidationError as PydanticValidationError
 
 from docsgpt.api import api
+from docsgpt.api.audit import record_event
 from docsgpt.api.pat.rules import filter_listing, mask_agent_key, may_see_agent_keys
 from docsgpt.guardrails.config import AgentConfig
 from docsgpt.api.user.base import (
@@ -760,6 +761,15 @@ class CreateAgent(Resource):
                     **kwargs,
                 )
                 new_id = str(agent_row["id"])
+                record_event(
+                    conn,
+                    "agent.created",
+                    actor=user,
+                    agent_id=new_id,
+                    name=data["name"],
+                    status=data["status"],
+                    agent_type=agent_type,
+                )
         except Exception as err:
             current_app.logger.error(f"Error creating agent: {err}", exc_info=True)
             return make_response(jsonify({"success": False}), 400)
@@ -1284,6 +1294,15 @@ class UpdateAgent(Resource):
                     updated = bool(result)
                 else:
                     updated = agents_repo.update(pg_agent_id, user, update_fields)
+                if updated:
+                    record_event(
+                        conn,
+                        "agent.updated",
+                        actor=user,
+                        agent_id=pg_agent_id,
+                        # Names only: values can hold prompts and credentials.
+                        fields=sorted(update_fields),
+                    )
                 if not updated:
                     return make_response(
                         jsonify(
@@ -1403,6 +1422,13 @@ class RegenerateAgentKey(Resource):
                 SharedConversationsRepository(conn).reassign_api_key(
                     old_key=old_key, new_key=new_key
                 )
+                record_event(
+                    conn,
+                    "agent.key_regenerated",
+                    actor=user,
+                    agent_id=pg_agent_id,
+                    name=existing_agent.get("name"),
+                )
         except Exception as err:
             current_app.logger.error(
                 f"Error regenerating agent key: {err}", exc_info=True
@@ -1457,6 +1483,14 @@ class DeleteAgent(Resource):
                 agents_repo.delete(pg_agent_id, user)
                 # Strip pinned/shared entries for this agent from the owner's prefs.
                 UsersRepository(conn).remove_agent_from_all(user, pg_agent_id)
+                record_event(
+                    conn,
+                    "agent.deleted",
+                    actor=user,
+                    agent_id=pg_agent_id,
+                    name=agent.get("name"),
+                    agent_type=agent.get("agent_type"),
+                )
         except Exception as err:
             current_app.logger.error(f"Error deleting agent: {err}", exc_info=True)
             return make_response(jsonify({"success": False}), 400)

docsgpt/api/user/conversations/routes.py

@@ -7,6 +7,7 @@ from flask_restx import fields, Namespace, Resource
 from sqlalchemy import text as sql_text
 
 from docsgpt.api import api
+from docsgpt.api.audit import record_event
 from docsgpt.api.answer.services.conversation_service import (
     TERMINATED_RESPONSE_PLACEHOLDER,
 )
@@ -45,6 +46,12 @@ class DeleteConversation(Resource):
                 conv = repo.get_any(conversation_id, user_id)
                 if conv is not None:
                     repo.delete(str(conv["id"]), user_id)
+                    record_event(
+                        conn,
+                        "conversation.deleted",
+                        actor=user_id,
+                        conversation_id=str(conv["id"]),
+                    )
         except Exception as err:
             current_app.logger.error(
                 f"Error deleting conversation: {err}", exc_info=True
@@ -65,7 +72,13 @@ class DeleteAllConversations(Resource):
         user_id = decoded_token.get("sub")
         try:
             with db_session() as conn:
-                ConversationsRepository(conn).delete_all_for_user(user_id)
+                deleted = ConversationsRepository(conn).delete_all_for_user(user_id)
+                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

docsgpt/api/user/sources/routes.py

@@ -10,6 +10,7 @@ from pydantic import ValidationError
 
 from docsgpt.agents.tools.path_utils import validate_tool_path
 from docsgpt.api import api
+from docsgpt.api.audit import record_event
 from docsgpt.api.pat.rules import filter_listing
 from docsgpt.api.user.tasks import (
     convert_source_to_wiki,
@@ -280,6 +281,13 @@ class DeleteOldIndexes(Resource):
         try:
             with db_session() as conn:
                 SourcesRepository(conn).delete(resolved_id, user)
+                record_event(
+                    conn,
+                    "source.deleted",
+                    actor=user,
+                    source_id=resolved_id,
+                    name=doc.get("name"),
+                )
         except Exception as err:
             current_app.logger.error(
                 f"Error deleting source row: {err}", exc_info=True
@@ -500,6 +508,23 @@ class ReingestSource(Resource):
                 exc_info=True,
             )
             return make_response(jsonify({"success": False}), 400)
+        try:
+            with db_session() as conn:
+                record_event(
+                    conn,
+                    "source.reingested",
+                    actor=user,
+                    source_id=resolved_source_id,
+                    name=doc.get("name"),
+                    # A team editor can reingest a source they do not own.
+                    owner=owner if owner != user else None,
+                    task_id=task.id,
+                )
+        except Exception as err:
+            current_app.logger.warning(
+                f"Could not audit reingest for {resolved_source_id}: {err}",
+                exc_info=True,
+            )
         return make_response(jsonify({"success": True, "task_id": task.id}), 200)
 
 
@@ -811,6 +836,14 @@ class CreateWikiSource(Resource):
                         updated_via="human",
                     )
                     rebuild_wiki_directory_structure(conn, source_id, user)
+                record_event(
+                    conn,
+                    "source.created",
+                    actor=user,
+                    source_id=source_id,
+                    name=name,
+                    type="wiki",
+                )
         except Exception as err:
             current_app.logger.error(
                 f"Error creating wiki source: {err}", exc_info=True

docsgpt/api/user/sources/upload.py

@@ -11,6 +11,7 @@ from flask_restx import fields, Namespace, Resource
 from sqlalchemy import text as sql_text
 
 from docsgpt.api import api
+from docsgpt.api.audit import record_event
 from docsgpt.api.user.tasks import ingest, ingest_connector_task, ingest_remote
 from docsgpt.api.user.team_sharing import effective_write_owner
 from docsgpt.core.settings import settings
@@ -400,6 +401,24 @@ class UploadFile(Resource):
         # ``source_id``; the worker uses it verbatim for every SSE event,
         # so the frontend can correlate inbound ``source.ingest.*`` to
         # this upload regardless of whether an idempotency key was set.
+        # 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,
+            )
         response_payload: dict = {
             "success": True,
             "task_id": response_task_id,

tests/api/test_audit_helper.py

@@ -0,0 +1,106 @@
+"""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 EVENT_CATEGORIES, category_for, record_event
+
+
+@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"
+
+    def test_every_category_is_declared(self):
+        assert "other" in EVENT_CATEGORIES

tests/api/test_data_plane_audit.py

@@ -0,0 +1,142 @@
+"""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
+
+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 == []
+
+
+@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 == []