feat(audit): split actor from subject on auth events

auth_events.user_id was overloaded and meant something different per call
site: admin mutations filed the row under the target user and hid the acting
admin in metadata->>'by', team events filed it under the actor, and quota
events switched between the two depending on scope. The practical consequence
was that "show me everything admin X did" had no answer.

Adds actor_id (never NULL) and target_id (NULL when the event is not about a
user) alongside the existing column, which keeps its meaning as the per-user
feed key so nothing historical is rewritten. The backfill recovers the actor
from metadata for rows that predate the columns.

Also adds the indexes the cross-user admin feed needs. The only index was
(user_id, created_at DESC), so the global feed -- which orders by created_at
with no user predicate -- fell back to a sequential scan plus a sort on every
page.

The feed gains actor, multi-event, until and free-text search filters, plus a
distinct-event catalogue so the UI can offer what the instance recorded
instead of asking operators to type a name from memory.

tests/storage/db/repositories/test_auth_events.py

@@ -0,0 +1,85 @@
+"""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:
+    def _seed(self, repo):
+        repo.insert("a", "oidc_login")
+        repo.insert("b", "oidc_login_denied")
+        repo.insert("b", "admin_user_deactivated", actor_id="admin-9", target_id="b")
+
+    def test_filter_by_actor(self, repo):
+        self._seed(repo)
+        rows = repo.list_all(actor_id="admin-9")
+        assert [r["event"] for r in rows] == ["admin_user_deactivated"]
+        assert repo.count_all(actor_id="admin-9") == 1
+
+    def test_filter_by_multiple_events(self, repo):
+        self._seed(repo)
+        rows = repo.list_all(events=["oidc_login", "oidc_login_denied"])
+        assert {r["event"] for r in rows} == {"oidc_login", "oidc_login_denied"}
+        assert repo.count_all(events=["oidc_login", "oidc_login_denied"]) == 2
+
+    def test_single_event_filter_still_supported(self, repo):
+        self._seed(repo)
+        assert repo.count_all(event="oidc_login") == 1
+
+    def test_filter_by_until(self, repo):
+        self._seed(repo)
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        assert repo.count_all(until=past) == 0
+        assert repo.count_all(until=datetime.now(timezone.utc) + timedelta(days=1)) == 3
+
+    def test_search_matches_user_actor_and_metadata(self, repo):
+        repo.insert("someone", "pat_created", metadata={"name": "ci-deploy-key"})
+        assert repo.count_all(search="ci-deploy") == 1
+        assert repo.count_all(search="someone") == 1
+        assert repo.count_all(search="nothing-here") == 0
+
+    def test_event_names_catalogue(self, repo):
+        self._seed(repo)
+        names = repo.event_names()
+        assert names == sorted(names)
+        assert "oidc_login" in names and "admin_user_deactivated" in names

tests/storage/db/test_migration_0034.py

@@ -0,0 +1,127 @@
+"""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|-"