feat(admin): merged activity feed with filters and export

DocsGPT keeps three append-only journals -- auth_events, device_audit_log and
guardrail_events -- each readable only on its own terms, and the second of
those had an admin endpoint that no UI ever called. An operator reviewing an
incident wants one timeline, not three.

Adds GET /api/admin/activity: all three journals projected onto a common row
shape and merged, ordered newest first, with facets for journal, category,
event name, actor, affected user, a time window and free-text search over the
detail payload. A category filter that only one journal can satisfy drops the
others from the union rather than scanning and discarding them.

GET /api/admin/activity/events returns the distinct (event, category) pairs
the instance has actually recorded, so the filter UI can offer real choices
instead of asking an operator to type "oidc_login_denied" from memory.

GET /api/admin/activity/export streams the same filtered feed as CSV or
NDJSON. Streamed rather than buffered, and capped, so a compliance export of a
busy instance is bounded work.

guardrail_events.api_key (a raw agent key) and matched_value (unredacted
source text) are excluded from the projection, mirroring the exclusion the
per-agent guardrail view already makes. Admin-gating is not a reason to widen
what a list response carries.

docsgpt/api/admin/__init__.py

@@ -1,4 +1,5 @@
 from .routes import admin_ns
 from . import quotas  # noqa: F401  (registers the quota resources on admin_ns)
+from . import activity  # noqa: F401  (registers the activity resources on admin_ns)
 
 __all__ = ["admin_ns"]

docsgpt/api/admin/activity.py

@@ -0,0 +1,234 @@
+"""Admin activity feed: one timeline over all three audit journals.
+
+``GET /api/admin/audit`` remains the ``auth_events``-only feed. These endpoints
+add the merged view an incident review actually wants — identity, access,
+configuration, data-plane, remote-device and guardrail activity in one ordered
+list — plus the filters and the export that make a review possible without
+shelling into the database.
+
+Every resource here is ``@admin_required``, like the rest of the namespace.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Iterator, Optional
+
+from flask import Response, jsonify, make_response, request, stream_with_context
+from flask_restx import Resource
+
+from docsgpt.api.admin.routes import _int_arg, _page, admin_ns
+from docsgpt.api.user.authz import admin_required
+from docsgpt.audit_events import ACTIVITY_CATEGORIES
+from docsgpt.storage.db.repositories.activity import (
+    ACTIVITY_COLUMNS,
+    ActivityRepository,
+)
+from docsgpt.storage.db.session import db_readonly
+
+logger = logging.getLogger(__name__)
+
+_FEEDS = ("auth", "device", "guardrail")
+_EXPORT_FORMATS = ("csv", "ndjson")
+
+# An export runs against a live instance, so it is bounded. Operators needing
+# the full history take a database dump, not an HTTP request.
+_EXPORT_MAX_ROWS = 100_000
+_EXPORT_CHUNK = 1_000
+
+_MAX_SEARCH_LENGTH = 200
+
+
+def _csv_list(name: str, allowed: tuple[str, ...]) -> Optional[list[str]]:
+    """Parse a repeated/comma-separated query arg, dropping unknown values.
+
+    Args:
+        name: Query parameter name.
+        allowed: The permitted values; anything else is ignored rather than
+            rejected, so an old bookmark degrades instead of erroring.
+
+    Returns:
+        The distinct valid values in the order given, or None when absent.
+    """
+    raw: list[str] = []
+    for value in request.args.getlist(name):
+        raw.extend(part.strip() for part in value.split(","))
+    seen: list[str] = []
+    for value in raw:
+        if value in allowed and value not in seen:
+            seen.append(value)
+    return seen or None
+
+
+def _event_list() -> Optional[list[str]]:
+    """Event names to filter on. Unconstrained — the catalogue is data, not an enum."""
+    names: list[str] = []
+    for value in request.args.getlist("event"):
+        for part in value.split(","):
+            part = part.strip()
+            if part and part not in names:
+                names.append(part)
+    return names or None
+
+
+def _time_arg(name: str) -> Optional[datetime]:
+    """Parse an ISO-8601 timestamp query arg, or None when absent/unparseable."""
+    raw = (request.args.get(name) or "").strip()
+    if not raw:
+        return None
+    try:
+        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    # A naive bound is read as UTC, matching how the feed stores timestamps.
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def _search_arg() -> Optional[str]:
+    value = (request.args.get("search") or "").strip()
+    return value[:_MAX_SEARCH_LENGTH] or None
+
+
+def _filters() -> dict:
+    """The filter set shared by the feed and the export."""
+    return {
+        "feeds": _csv_list("feed", _FEEDS),
+        "categories": _csv_list("category", ACTIVITY_CATEGORIES),
+        "events": _event_list(),
+        "actor_id": request.args.get("actor_id") or None,
+        "user_id": request.args.get("user_id") or None,
+        "since": _time_arg("since"),
+        "until": _time_arg("until"),
+        "search": _search_arg(),
+    }
+
+
+def _serialize(row: dict) -> dict:
+    """Make one feed row JSON-safe (timestamps to ISO-8601)."""
+    created = row.get("created_at")
+    return {
+        **row,
+        "created_at": created.isoformat() if hasattr(created, "isoformat") else created,
+    }
+
+
+@admin_ns.route("/admin/activity")
+class AdminActivityResource(Resource):
+    @admin_required
+    def get(self):
+        """Merged audit feed, newest first, with filters and a total."""
+        page, page_size, offset = _page()
+        filters = _filters()
+        with db_readonly() as conn:
+            repo = ActivityRepository(conn)
+            total = repo.count(**filters)
+            rows = repo.list(limit=page_size, offset=offset, **filters)
+        return make_response(
+            jsonify(
+                {
+                    "success": True,
+                    "activity": [_serialize(row) for row in rows],
+                    "page": page,
+                    "page_size": page_size,
+                    "total": total,
+                    "has_more": offset + len(rows) < total,
+                }
+            ),
+            200,
+        )
+
+
+@admin_ns.route("/admin/activity/events")
+class AdminActivityCatalogueResource(Resource):
+    @admin_required
+    def get(self):
+        """Distinct ``(event, category)`` pairs this instance has recorded.
+
+        The filter UI offers these instead of a free-text box, so an operator
+        never has to guess that denied logins are ``oidc_login_denied``.
+        """
+        with db_readonly() as conn:
+            catalogue = ActivityRepository(conn).event_names()
+        return make_response(
+            jsonify(
+                {
+                    "success": True,
+                    "events": catalogue,
+                    "categories": list(ACTIVITY_CATEGORIES),
+                    "feeds": list(_FEEDS),
+                }
+            ),
+            200,
+        )
+
+
+@admin_ns.route("/admin/activity/export")
+class AdminActivityExportResource(Resource):
+    @admin_required
+    def get(self):
+        """Stream the filtered feed as CSV or NDJSON.
+
+        Streamed, not buffered: a compliance export of a busy instance should
+        not be built in memory first. Capped at ``_EXPORT_MAX_ROWS``; the
+        response header says whether the cap was reached.
+        """
+        fmt = (request.args.get("format") or "csv").lower()
+        if fmt not in _EXPORT_FORMATS:
+            return make_response(
+                jsonify({"success": False, "message": "Invalid format"}), 400
+            )
+        limit = max(1, min(_EXPORT_MAX_ROWS, _int_arg("max_rows", _EXPORT_MAX_ROWS)))
+        filters = _filters()
+        stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
+        generator = _csv_rows if fmt == "csv" else _ndjson_rows
+        return Response(
+            stream_with_context(generator(filters, limit)),
+            mimetype="text/csv" if fmt == "csv" else "application/x-ndjson",
+            headers={
+                "Content-Disposition": (
+                    f'attachment; filename="docsgpt-activity-{stamp}.{fmt}"'
+                ),
+                # Partial content would otherwise be indistinguishable from a
+                # complete export.
+                "X-Export-Max-Rows": str(limit),
+                "Cache-Control": "no-store",
+            },
+        )
+
+
+def _export_rows(filters: dict, limit: int) -> Iterator[dict]:
+    """Walk the filtered feed, holding the connection open for the whole stream."""
+    with db_readonly() as conn:
+        yield from ActivityRepository(conn).iter_all(
+            chunk_size=_EXPORT_CHUNK, max_rows=limit, **filters
+        )
+
+
+def _csv_rows(filters: dict, limit: int) -> Iterator[str]:
+    buffer = io.StringIO()
+    writer = csv.DictWriter(buffer, fieldnames=list(ACTIVITY_COLUMNS))
+    writer.writeheader()
+    yield _drain(buffer)
+    for row in _export_rows(filters, limit):
+        serialized = _serialize(row)
+        # ``detail`` is a JSON object; a CSV cell holds its compact encoding.
+        serialized["detail"] = json.dumps(serialized.get("detail") or {})
+        writer.writerow(serialized)
+        yield _drain(buffer)
+
+
+def _ndjson_rows(filters: dict, limit: int) -> Iterator[str]:
+    for row in _export_rows(filters, limit):
+        yield json.dumps(_serialize(row), default=str) + "\n"
+
+
+def _drain(buffer: io.StringIO) -> str:
+    """Return and clear the writer's buffer, so each row streams as written."""
+    value = buffer.getvalue()
+    buffer.seek(0)
+    buffer.truncate(0)
+    return value

docsgpt/api/audit.py

@@ -1,4 +1,4 @@
-"""Shared audit-trail helper and the event → category map.
+"""Shared audit-trail helper for recording data-plane and identity events.
 
 Identity and access events (login, role grants, provisioning) have been
 audited since ``auth_events`` was introduced. Data-plane actions — creating and
@@ -23,61 +23,13 @@ from typing import Any, Optional
 
 from flask import has_request_context, request
 
+from docsgpt.audit_events import ACTIVITY_CATEGORIES, category_for
 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"
+# Re-exported so route modules have one import for recording and classifying.
+__all__ = ["ACTIVITY_CATEGORIES", "category_for", "record_event"]
 
 
 def record_event(

docsgpt/audit_events.py

@@ -0,0 +1,104 @@
+"""Audit event taxonomy: the categories the admin activity feed filters on.
+
+Lives at the top level (like ``docsgpt/pricing.py``) because both the API
+layer — which records events — and the storage layer — which has to filter a
+merged feed in SQL — need the same map, and neither should import the other.
+
+Two representations of one table:
+
+* :func:`category_for` classifies a single event name in Python.
+* :func:`category_case_sql` emits the equivalent ``CASE`` expression so a
+  category filter can be pushed into the query instead of enumerating every
+  event name a release might have written.
+"""
+
+from __future__ import annotations
+
+# Categories in the order the UI shows them. ``other`` catches event names an
+# older release wrote that this one does not know about — the feed must never
+# hide a row just because its name is unfamiliar.
+ACTIVITY_CATEGORIES = ("identity", "access", "config", "data", "device", "safety", "other")
+
+# Exact event names.
+_EXACT: dict[str, str] = {
+    # Who you are: authentication, provisioning, personal credentials.
+    "oidc_login": "identity",
+    "oidc_login_denied": "identity",
+    "oidc_refresh": "identity",
+    "backchannel_logout": "identity",
+    "scim_created": "identity",
+    "scim_deactivated": "identity",
+    "scim_reactivated": "identity",
+    "pat_created": "identity",
+    "pat_revoked": "identity",
+    "pat_regenerated": "identity",
+    # What you may do: roles, account state.
+    "role_granted": "access",
+    "role_revoked": "access",
+    "admin_user_activated": "access",
+    "admin_user_deactivated": "access",
+    "admin_sessions_revoked": "access",
+    # How the instance is configured.
+    "quota_policy_set": "config",
+    "quota_policy_deleted": "config",
+}
+
+# Dotted namespaces, matched on the part before the first ``.``.
+_NAMESPACES: dict[str, str] = {
+    "team": "access",
+    "source": "data",
+    "agent": "data",
+    "conversation": "data",
+    "device": "device",
+    "guardrail": "safety",
+}
+
+
+def category_for(event: str) -> str:
+    """Return the activity category for ``event``.
+
+    Args:
+        event: The audit event name (``oidc_login``, ``source.deleted``).
+
+    Returns:
+        One of :data:`ACTIVITY_CATEGORIES`; ``"other"`` for names this release
+        does not recognise.
+    """
+    namespace, dot, _ = event.partition(".")
+    if dot:
+        return _NAMESPACES.get(namespace, "other")
+    return _EXACT.get(event, "other")
+
+
+def category_case_sql(column: str) -> str:
+    """Build the SQL ``CASE`` that classifies ``column`` the way :func:`category_for` does.
+
+    Uses ``split_part`` rather than ``LIKE`` so the expression carries no ``%``
+    — which a pyformat DBAPI would otherwise read as a parameter marker.
+
+    Args:
+        column: The SQL expression holding an event name.
+
+    Returns:
+        A parenthesised ``CASE`` expression. Every literal in it comes from the
+        maps above, never from user input, so it is safe to interpolate.
+    """
+    namespace_branches = " ".join(
+        f"WHEN '{namespace}' THEN '{category}'"
+        for namespace, category in _NAMESPACES.items()
+    )
+    by_category: dict[str, list[str]] = {}
+    for event, category in _EXACT.items():
+        by_category.setdefault(category, []).append(event)
+    exact_branches = " ".join(
+        f"WHEN {column} IN ({', '.join(repr(event) for event in sorted(events))}) "
+        f"THEN '{category}'"
+        for category, events in sorted(by_category.items())
+    )
+    return (
+        "(CASE "
+        f"WHEN position('.' in {column}) > 0 "
+        f"THEN (CASE split_part({column}, '.', 1) {namespace_branches} ELSE 'other' END) "
+        f"{exact_branches} "
+        "ELSE 'other' END)"
+    )

docsgpt/storage/db/repositories/activity.py

@@ -0,0 +1,332 @@
+"""One merged, normalized feed over the instance's three audit journals.
+
+DocsGPT keeps three append-only journals, each written by a different part of
+the system and each previously readable only on its own terms:
+
+* ``auth_events`` — identity, access, configuration and (since the data-plane
+  hooks) resource mutations.
+* ``device_audit_log`` — every remote-device command dispatch and its outcome.
+* ``guardrail_events`` — every guardrail decision on a request.
+
+An operator reviewing an incident wants one timeline, not three. This
+repository projects all three onto a common row shape and merges them with
+``UNION ALL``, so a single ordered, filtered, paginated feed spans the lot.
+
+Two exclusions are deliberate and mirror the per-agent guardrail view: the
+``guardrail_events.api_key`` column is a raw agent key, and ``matched_value``
+is unredacted source text. Neither is projected here, admin-gated or not.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Iterator, Optional, Sequence
+
+from sqlalchemy import Connection, text
+
+from docsgpt.audit_events import category_case_sql
+
+
+# Columns every branch of the union produces, in order.
+ACTIVITY_COLUMNS = (
+    "feed",
+    "id",
+    "event",
+    "category",
+    "actor_id",
+    "target_id",
+    "ip",
+    "user_agent",
+    "outcome",
+    "detail",
+    "created_at",
+)
+
+# Which journal each feed name reads from. Used to skip a branch entirely when
+# the category filter cannot match it.
+_FEED_CATEGORIES = {"auth": None, "device": "device", "guardrail": "safety"}
+
+_AUTH_BRANCH = f"""
+    SELECT 'auth' AS feed,
+           id::text AS id,
+           event,
+           {category_case_sql("event")} AS category,
+           actor_id,
+           target_id,
+           ip,
+           user_agent,
+           NULL::text AS outcome,
+           metadata AS detail,
+           created_at
+    FROM auth_events
+"""
+
+_DEVICE_BRANCH = """
+    SELECT 'device' AS feed,
+           id::text AS id,
+           'device.' || action AS event,
+           'device' AS category,
+           user_id AS actor_id,
+           NULL::text AS target_id,
+           NULL::text AS ip,
+           NULL::text AS user_agent,
+           decision AS outcome,
+           jsonb_strip_nulls(jsonb_build_object(
+               'device_id', device_id,
+               'invocation_id', invocation_id,
+               'command', command,
+               'working_dir', working_dir,
+               'approval_mode', approval_mode,
+               'decision_reason', decision_reason,
+               'exit_code', exit_code,
+               'duration_ms', duration_ms,
+               'agent_id', agent_id,
+               'conversation_id', conversation_id
+           )) AS detail,
+           created_at
+    FROM device_audit_log
+"""
+
+_GUARDRAIL_BRANCH = """
+    SELECT 'guardrail' AS feed,
+           g.id::text AS id,
+           'guardrail.' || g.stage AS event,
+           'safety' AS category,
+           g.user_id AS actor_id,
+           NULL::text AS target_id,
+           NULL::text AS ip,
+           NULL::text AS user_agent,
+           g.outcome,
+           jsonb_strip_nulls(jsonb_build_object(
+               'check_name', g.check_name,
+               'detector_type', g.detector_type,
+               'action', g.action,
+               'category', g.category,
+               'score', g.score,
+               'match_count', g.match_count,
+               'agent_id', g.agent_id,
+               'message_id', g.message_id,
+               'request_id', g.request_id,
+               'detail', g.detail
+           )) AS detail,
+           g.created_at
+    FROM guardrail_events g
+"""
+
+_BRANCHES = {
+    "auth": _AUTH_BRANCH,
+    "device": _DEVICE_BRANCH,
+    "guardrail": _GUARDRAIL_BRANCH,
+}
+
+
+class ActivityRepository:
+    """Read-only merged feed over ``auth_events`` + the two side journals."""
+
+    def __init__(self, conn: Connection) -> None:
+        self._conn = conn
+
+    @staticmethod
+    def _feeds_for(
+        feeds: Optional[Sequence[str]], categories: Optional[Sequence[str]]
+    ) -> list[str]:
+        """Which union branches can still produce a matching row.
+
+        A ``category=device`` filter cannot be satisfied by ``auth_events``, so
+        that branch is dropped from the union rather than scanned and
+        discarded. ``auth`` spans several categories and is only dropped when
+        the filter names none of them.
+        """
+        selected = list(feeds) if feeds else list(_BRANCHES)
+        selected = [feed for feed in selected if feed in _BRANCHES]
+        if not categories:
+            return selected
+        wanted = set(categories)
+        fixed_only = {category for category in _FEED_CATEGORIES.values() if category}
+        keep = []
+        for feed in selected:
+            fixed = _FEED_CATEGORIES[feed]
+            if fixed is None:
+                # ``auth`` spans several categories, so keep it unless the
+                # filter names only categories another journal owns outright.
+                if wanted - fixed_only:
+                    keep.append(feed)
+            elif fixed in wanted:
+                keep.append(feed)
+        return keep
+
+    def _query(
+        self,
+        *,
+        feeds: Optional[Sequence[str]],
+        categories: Optional[Sequence[str]],
+        events: Optional[Sequence[str]],
+        actor_id: Optional[str],
+        user_id: Optional[str],
+        since: Optional[datetime],
+        until: Optional[datetime],
+        search: Optional[str],
+    ) -> tuple[str, dict]:
+        selected = self._feeds_for(feeds, categories)
+        if not selected:
+            return "", {}
+        union = " UNION ALL ".join(_BRANCHES[feed] for feed in selected)
+
+        clauses: list[str] = []
+        params: dict[str, Any] = {}
+        if categories:
+            clauses.append("category = ANY(:categories)")
+            params["categories"] = list(categories)
+        if events:
+            clauses.append("event = ANY(:events)")
+            params["events"] = list(events)
+        if actor_id:
+            clauses.append("actor_id = :actor_id")
+            params["actor_id"] = actor_id
+        if user_id:
+            clauses.append("(actor_id = :user_id OR target_id = :user_id)")
+            params["user_id"] = user_id
+        if since is not None:
+            clauses.append("created_at >= :since")
+            params["since"] = since
+        if until is not None:
+            clauses.append("created_at <= :until")
+            params["until"] = until
+        if search:
+            clauses.append(
+                "(actor_id ILIKE :search OR target_id ILIKE :search "
+                "OR event ILIKE :search OR ip ILIKE :search "
+                "OR COALESCE(outcome, '') ILIKE :search OR detail::text ILIKE :search)"
+            )
+            params["search"] = f"%{search}%"
+        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
+        return f"SELECT * FROM ({union}) AS activity {where}", params
+
+    def list(
+        self,
+        *,
+        feeds: Optional[Sequence[str]] = None,
+        categories: Optional[Sequence[str]] = None,
+        events: Optional[Sequence[str]] = None,
+        actor_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+        since: Optional[datetime] = None,
+        until: Optional[datetime] = None,
+        search: Optional[str] = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> list[dict]:
+        """Return one page of the merged feed, newest first.
+
+        Args:
+            feeds: Restrict to named journals (``auth``, ``device``, ``guardrail``).
+            categories: Restrict to activity categories.
+            events: Restrict to exact event names.
+            actor_id: Only rows this user performed.
+            user_id: Rows where this user is the actor or the target.
+            since: Inclusive lower bound on ``created_at``.
+            until: Inclusive upper bound on ``created_at``.
+            search: Case-insensitive substring over ids, event, IP and detail.
+            limit: Page size.
+            offset: Rows to skip.
+
+        Returns:
+            Row dicts carrying :data:`ACTIVITY_COLUMNS`.
+        """
+        sql, params = self._query(
+            feeds=feeds,
+            categories=categories,
+            events=events,
+            actor_id=actor_id,
+            user_id=user_id,
+            since=since,
+            until=until,
+            search=search,
+        )
+        if not sql:
+            return []
+        params.update({"limit": int(limit), "offset": int(offset)})
+        result = self._conn.execute(
+            text(f"{sql} ORDER BY created_at DESC, feed, id DESC LIMIT :limit OFFSET :offset"),
+            params,
+        )
+        return [dict(row._mapping) for row in result.fetchall()]
+
+    def count(
+        self,
+        *,
+        feeds: Optional[Sequence[str]] = None,
+        categories: Optional[Sequence[str]] = None,
+        events: Optional[Sequence[str]] = None,
+        actor_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+        since: Optional[datetime] = None,
+        until: Optional[datetime] = None,
+        search: Optional[str] = None,
+    ) -> int:
+        """Total rows matching the same filters as :meth:`list`."""
+        sql, params = self._query(
+            feeds=feeds,
+            categories=categories,
+            events=events,
+            actor_id=actor_id,
+            user_id=user_id,
+            since=since,
+            until=until,
+            search=search,
+        )
+        if not sql:
+            return 0
+        scalar = self._conn.execute(
+            text(f"SELECT count(*) FROM ({sql}) AS matched"), params
+        ).scalar()
+        return int(scalar or 0)
+
+    def iter_all(
+        self,
+        *,
+        chunk_size: int = 1000,
+        max_rows: int = 100_000,
+        **filters: Any,
+    ) -> Iterator[dict]:
+        """Yield every matching row, newest first, for the export endpoint.
+
+        Pages through the feed rather than materializing it, so an export of a
+        large instance streams instead of building the whole result in memory.
+
+        Args:
+            chunk_size: Rows fetched per round trip.
+            max_rows: Hard ceiling; an export stops here rather than running
+                unbounded against a busy instance.
+            **filters: As :meth:`list`, minus ``limit`` / ``offset``.
+
+        Yields:
+            Row dicts carrying :data:`ACTIVITY_COLUMNS`.
+        """
+        emitted = 0
+        offset = 0
+        while emitted < max_rows:
+            page = self.list(
+                limit=min(chunk_size, max_rows - emitted), offset=offset, **filters
+            )
+            if not page:
+                return
+            for row in page:
+                yield row
+            emitted += len(page)
+            offset += len(page)
+
+    def event_names(self) -> list[dict]:
+        """Distinct ``(event, category)`` pairs across all three journals.
+
+        Feeds the filter's event picker, so operators choose from what the
+        instance actually recorded rather than typing a name from memory.
+        """
+        union = " UNION ALL ".join(_BRANCHES.values())
+        result = self._conn.execute(
+            text(
+                f"SELECT DISTINCT event, category FROM ({union}) AS activity "
+                "ORDER BY category, event"
+            )
+        )
+        return [{"event": row[0], "category": row[1]} for row in result.fetchall()]