sec fixes

application/alembic/versions/0018_tool_attempts_attribution.py

@@ -3,8 +3,13 @@
 Adds ``tool_call_attempts.user_id`` / ``agent_id`` so tool analytics can
 attribute attempts that never get a ``message_id`` — headless runs
 (scheduled / webhook) execute tools before any conversation message
-exists, and parse-failure rows never reach one at all. Existing rows are
-backfilled through their parent message where one exists.
+exists, and parse-failure rows never reach one at all.
+
+DDL only — a whole-table UPDATE here would hold the ALTERs' ACCESS
+EXCLUSIVE lock across the rewrite and stall live tool journaling. The
+backfill lives in ``scripts/db/backfill_tool_attempts_attribution.py``;
+until it runs, the analytics reader falls back to the parent message's
+user for unstamped rows.
 
 Revision ID: 0018_tool_attempts_attribution
 Revises: 0017_oidc_scim
@@ -24,17 +29,6 @@ depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
     op.execute("ALTER TABLE tool_call_attempts ADD COLUMN user_id TEXT;")
     op.execute("ALTER TABLE tool_call_attempts ADD COLUMN agent_id UUID;")
-    op.execute(
-        """
-        UPDATE tool_call_attempts t
-           SET user_id = m.user_id,
-               agent_id = c.agent_id
-          FROM conversation_messages m
-          LEFT JOIN conversations c ON c.id = m.conversation_id
-         WHERE t.message_id = m.id
-           AND t.user_id IS NULL;
-        """
-    )
     op.execute(
         "CREATE INDEX tool_call_attempts_user_ts_idx "
         "ON tool_call_attempts (user_id, attempted_at DESC) "

application/api/user/analytics/routes.py

@@ -70,18 +70,21 @@ def _intervals_for_filter(filter_option, start_date, end_date):
 def _resolve_agent(conn, api_key_id, user_id):
     """Owner-scoped agent lookup for analytics filters.
 
-    Returns ``(agent, api_key, agent_pg_id)``. ``api_key`` falls back to
-    ``""`` and ``agent_pg_id`` to ``None``, neither of which matches any
-    row, so filtering by an unknown (or another user's) agent returns
-    nothing rather than everything. Accepts UUID or legacy Mongo
-    ObjectId ids.
+    Returns ``(agent, api_key, agent_pg_id)``. ``agent`` is ``None`` when
+    the id doesn't resolve to one of the caller's agents — callers must
+    short-circuit with an empty result, not fall back to sentinel filter
+    values. ``api_key`` is ``None`` (never ``""``) for key-less agents:
+    draft agents store ``key = ''``, and an ``''`` filter would match the
+    ``''`` that writers like ``stack_logs`` stamp on every key-less
+    request — leaking rows across users. NULL matches nothing. Accepts
+    UUID or legacy Mongo ObjectId ids.
     """
     agent = (
         AgentsRepository(conn).get_any(api_key_id, user_id)
         if api_key_id
         else None
     )
-    api_key = (agent or {}).get("key") or ""
+    api_key = (agent or {}).get("key") or None
     agent_pg_id = str(agent["id"]) if agent else None
     return agent, api_key, agent_pg_id
 
@@ -121,9 +124,24 @@ class GetMessageAnalytics(Resource):
 
         try:
             with db_readonly() as conn:
-                _agent, api_key, agent_pg_id = _resolve_agent(
+                agent, api_key, agent_pg_id = _resolve_agent(
                     conn, api_key_id, user
                 )
+                if api_key_id and agent is None:
+                    # Unknown / not-owned agent: empty result, not a
+                    # sentinel filter (see _resolve_agent).
+                    intervals = _intervals_for_filter(
+                        filter_option, start_date, end_date
+                    )
+                    return make_response(
+                        jsonify(
+                            {
+                                "success": True,
+                                "messages": {i: 0 for i in intervals},
+                            }
+                        ),
+                        200,
+                    )
 
                 # Count messages per bucket, filtered by the conversation's
                 # owner (user_id) and optionally the agent. The ``user_id``
@@ -214,7 +232,17 @@ class GetTokenAnalytics(Resource):
         api_key_id = data.get("api_key_id")
         filter_option = data.get("filter_option", "last_30_days")
         group_by = data.get("group_by") or "none"
-        include_side_channel = bool(data.get("include_side_channel", True))
+        # ``@api.expect`` documents but never validates/coerces — a JSON
+        # string like "false" must not truthy-coerce to True.
+        raw_side = data.get("include_side_channel", True)
+        if isinstance(raw_side, str):
+            include_side_channel = raw_side.strip().lower() not in (
+                "false",
+                "0",
+                "no",
+            )
+        else:
+            include_side_channel = bool(raw_side)
 
         window = _range_for_filter(filter_option)
         if window is None or group_by not in ("none", "model", "agent", "source"):
@@ -225,22 +253,29 @@ class GetTokenAnalytics(Resource):
 
         try:
             with db_readonly() as conn:
-                _agent, api_key, agent_pg_id = _resolve_agent(
+                agent, api_key, agent_pg_id = _resolve_agent(
                     conn, api_key_id, user
                 )
-                # The owner-scoped lookup above gates access, so the
-                # user_id filter is dropped when agent-filtering —
-                # external API-key rows have no user_id.
-                rows = TokenUsageRepository(conn).bucketed_totals(
-                    bucket_unit=bucket_unit,
-                    user_id=None if api_key_id else user,
-                    api_key=api_key if api_key_id else None,
-                    agent_id=agent_pg_id,
-                    timestamp_gte=start_date,
-                    timestamp_lt=end_date,
-                    group_by=None if group_by == "none" else group_by,
-                    include_side_channel=include_side_channel,
-                )
+                if api_key_id and agent is None:
+                    # Unknown / not-owned agent: empty result, not a
+                    # sentinel filter (see _resolve_agent).
+                    rows = []
+                else:
+                    # The owner-scoped lookup gates access, so the
+                    # user_id filter is dropped when agent-filtering
+                    # (shared-agent rows carry the caller's user_id).
+                    # The agent match is key-OR-id: chat stamps the
+                    # key, headless runs stamp agent_id.
+                    rows = TokenUsageRepository(conn).bucketed_totals(
+                        bucket_unit=bucket_unit,
+                        user_id=None if api_key_id else user,
+                        api_key=api_key,
+                        agent_id=agent_pg_id,
+                        timestamp_gte=start_date,
+                        timestamp_lt=end_date,
+                        group_by=None if group_by == "none" else group_by,
+                        include_side_channel=include_side_channel,
+                    )
 
             intervals = _intervals_for_filter(filter_option, start_date, end_date)
             daily_token_usage = {interval: 0 for interval in intervals}
@@ -324,9 +359,25 @@ class GetFeedbackAnalytics(Resource):
 
         try:
             with db_readonly() as conn:
-                _agent, api_key, agent_pg_id = _resolve_agent(
+                agent, api_key, agent_pg_id = _resolve_agent(
                     conn, api_key_id, user
                 )
+                if api_key_id and agent is None:
+                    intervals = _intervals_for_filter(
+                        filter_option, start_date, end_date
+                    )
+                    return make_response(
+                        jsonify(
+                            {
+                                "success": True,
+                                "feedback": {
+                                    i: {"positive": 0, "negative": 0}
+                                    for i in intervals
+                                },
+                            }
+                        ),
+                        200,
+                    )
 
                 # Feedback lives inside the ``conversation_messages.feedback``
                 # JSONB as ``{"text": "like"|"dislike", "timestamp": "..."}``.
@@ -421,21 +472,23 @@ class GetToolAnalytics(Resource):
 
         try:
             with db_readonly() as conn:
-                _agent, api_key, agent_pg_id = _resolve_agent(
+                agent, api_key, agent_pg_id = _resolve_agent(
                     conn, api_key_id, user
                 )
+                if api_key_id and agent is None:
+                    return make_response(
+                        jsonify({"success": True, "tools": []}), 200
+                    )
 
-                # Attribution: ``t.user_id`` is stamped at propose time
-                # (0018); rows from before the migration fall back to the
-                # parent message's user. Headless runs (scheduled / webhook)
-                # have no message, so the message join is LEFT.
+                # Exclude non-terminal rows: a stuck ``proposed`` attempt
+                # (stream died, approval never granted) would otherwise
+                # render as a phantom success (successful = calls - failures).
                 clauses = [
-                    "COALESCE(t.user_id, m.user_id) = :user_id",
+                    "t.status <> 'proposed'",
                     "t.attempted_at >= :start",
                     "t.attempted_at <= :end",
                 ]
                 params: dict = {
-                    "user_id": user,
                     "start": start_date,
                     "end": end_date,
                 }
@@ -444,10 +497,12 @@ class GetToolAnalytics(Resource):
                     "LEFT JOIN conversations c ON c.id = m.conversation_id "
                 )
                 if api_key_id:
-                    # Match by direct agent stamp (headless), by the
-                    # conversation's api_key (external chat), or by the
-                    # conversation's agent_id (owner chats / rows from
-                    # before 0018 stamped attempts directly).
+                    # Match by direct agent stamp (headless), the
+                    # conversation's api_key (external chat), or the
+                    # conversation's agent_id (owner chats / pre-0018
+                    # rows). The owner-scoped lookup gates access, so
+                    # no user clause — the owner also sees shared-agent
+                    # traffic logged under callers' user_ids.
                     clauses.append(
                         "(t.agent_id = CAST(:agent_pg_id AS uuid)"
                         " OR c.api_key = :api_key"
@@ -455,6 +510,16 @@ class GetToolAnalytics(Resource):
                     )
                     params["agent_pg_id"] = agent_pg_id
                     params["api_key"] = api_key
+                else:
+                    # ``t.user_id`` is stamped at propose time (0018);
+                    # pre-migration rows fall back to the parent
+                    # message's user (LEFT join — headless runs have no
+                    # message). OR rather than COALESCE keeps the first
+                    # arm index-sargable.
+                    clauses.append(
+                        "(t.user_id = :user_id OR m.user_id = :user_id)"
+                    )
+                    params["user_id"] = user
                 where = " AND ".join(clauses)
                 sql = (
                     "SELECT t.tool_name, "
@@ -519,9 +584,29 @@ class GetScheduleAnalytics(Resource):
 
         try:
             with db_readonly() as conn:
-                _agent, _api_key, agent_pg_id = _resolve_agent(
+                agent, _api_key, agent_pg_id = _resolve_agent(
                     conn, api_key_id, user
                 )
+                if api_key_id and agent is None:
+                    intervals = _intervals_for_filter(
+                        filter_option, start_date, end_date
+                    )
+                    return make_response(
+                        jsonify(
+                            {
+                                "success": True,
+                                "runs": {
+                                    i: {
+                                        "completed": 0,
+                                        "failed": 0,
+                                        "skipped": 0,
+                                    }
+                                    for i in intervals
+                                },
+                            }
+                        ),
+                        200,
+                    )
 
                 # A run's effective time is when it finished (fell back to
                 # started/scheduled for runs that never got that far).
@@ -538,8 +623,6 @@ class GetScheduleAnalytics(Resource):
                     "fmt": pg_fmt,
                 }
                 if api_key_id:
-                    # Filtering by an agent that doesn't exist (or isn't
-                    # the caller's) must return nothing, not everything.
                     clauses.append("r.agent_id = CAST(:agent_id AS uuid)")
                     params["agent_id"] = agent_pg_id
                 where = " AND ".join(clauses)
@@ -622,9 +705,14 @@ class GetUserLogs(Resource):
             return make_response(jsonify({"success": False}), 401)
         user = decoded_token.get("sub")
         data = request.get_json() or {}
-        page = int(data.get("page", 1))
+        try:
+            page = max(1, int(data.get("page") or 1))
+            page_size = max(1, min(100, int(data.get("page_size") or 10)))
+        except (TypeError, ValueError):
+            return make_response(
+                jsonify({"success": False, "message": "Invalid option"}), 400
+            )
         api_key_id = data.get("api_key_id")
-        page_size = int(data.get("page_size", 10))
         level = data.get("level")
         event_type = data.get("event_type")
         search = data.get("search")
@@ -645,6 +733,21 @@ class GetUserLogs(Resource):
                 agent, api_key, agent_pg_id = _resolve_agent(
                     conn, api_key_id, user
                 )
+                if api_key_id and agent is None:
+                    # Unknown / not-owned agent: empty page, not a
+                    # sentinel filter (see _resolve_agent).
+                    return make_response(
+                        jsonify(
+                            {
+                                "success": True,
+                                "logs": [],
+                                "page": page,
+                                "page_size": page_size,
+                                "has_more": False,
+                            }
+                        ),
+                        200,
+                    )
                 params: dict = {
                     "user_id": user,
                     "limit": page_size + 1,
@@ -653,34 +756,22 @@ class GetUserLogs(Resource):
 
                 # ``schedule`` / ``webhook`` errors are first-class events in
                 # their own branches; keep them out of ``system`` so a failed
-                # run doesn't appear twice.
-                chat_where = ["l.user_id = :user_id"]
-                webhook_where = [
-                    "s.user_id = :user_id",
-                    "COALESCE(s.endpoint, '') = 'webhook'",
-                ]
-                system_where = [
-                    "s.user_id = :user_id",
-                    "s.level = 'error'",
-                    "COALESCE(s.endpoint, '') NOT IN ('webhook', 'schedule')",
-                ]
-                # Terminal statuses only (worker writes ``success`` /
-                # ``failed`` / ``timeout`` / ``skipped``; ``completed`` kept
-                # defensively). Pending/running runs aren't log entries yet.
-                schedule_where = [
-                    "r.user_id = :user_id",
-                    "r.status IN ('success', 'completed', 'failed', 'timeout', 'skipped')",
-                ]
-                workflow_where = ["wr.user_id = :user_id"]
+                # run doesn't appear twice. A failed webhook activity writes
+                # both an error row and an info row for the same activity_id
+                # (logging.py:_consume_and_log); NOT-EXISTS drops the info twin.
+                webhook_dedupe = (
+                    "NOT (s.level = 'info' AND EXISTS ("
+                    "SELECT 1 FROM stack_logs e "
+                    "WHERE e.activity_id = s.activity_id "
+                    "AND e.level = 'error'))"
+                )
                 if api_key_id:
-                    # Filter each source by the selected agent. An unknown
-                    # agent (or one without a key) must match nothing. The
-                    # agent lookup above is already owner-scoped, so the
-                    # chat/webhook/system branches match on the agent key
-                    # alone — shared agents log external callers under the
-                    # caller's user_id, and the owner should still see that
-                    # traffic on the agent's own logs page (legacy
-                    # ``find_by_api_key`` behavior).
+                    # The owner-scoped lookup gates access, so the
+                    # chat/webhook/system branches match on the agent
+                    # key/id alone — the owner also sees shared-agent
+                    # traffic logged under callers' user_ids. The
+                    # agent_id arm covers key-less (draft) agents,
+                    # whose owner chats log a null api_key.
                     params["api_key"] = api_key
                     params["agent_pg_id"] = agent_pg_id
                     params["agent_workflow_id"] = (
@@ -688,77 +779,135 @@ class GetUserLogs(Resource):
                         if agent and agent.get("workflow_id")
                         else None
                     )
-                    chat_where = ["l.data->>'api_key' = :api_key"]
+                    chat_where = [
+                        "(l.data->>'api_key' = :api_key"
+                        " OR l.data->>'agent_id' = :agent_pg_id)"
+                    ]
                     webhook_where = [
                         "COALESCE(s.endpoint, '') = 'webhook'",
                         "s.api_key = :api_key",
+                        webhook_dedupe,
                     ]
                     system_where = [
                         "s.level = 'error'",
                         "COALESCE(s.endpoint, '') NOT IN ('webhook', 'schedule')",
                         "s.api_key = :api_key",
                     ]
-                    schedule_where.append(
-                        "r.agent_id = CAST(:agent_pg_id AS uuid)"
-                    )
-                    workflow_where.append(
-                        "wr.workflow_id = CAST(:agent_workflow_id AS uuid)"
-                    )
+                    schedule_where = [
+                        "r.user_id = :user_id",
+                        "r.status IN ('success', 'completed', 'failed', 'timeout', 'skipped')",
+                        "r.agent_id = CAST(:agent_pg_id AS uuid)",
+                    ]
+                    workflow_where = [
+                        "wr.user_id = :user_id",
+                        "wr.workflow_id = CAST(:agent_workflow_id AS uuid)",
+                    ]
+                else:
+                    chat_where = ["l.user_id = :user_id"]
+                    webhook_where = [
+                        "s.user_id = :user_id",
+                        "COALESCE(s.endpoint, '') = 'webhook'",
+                        webhook_dedupe,
+                    ]
+                    system_where = [
+                        "s.user_id = :user_id",
+                        "s.level = 'error'",
+                        "COALESCE(s.endpoint, '') NOT IN ('webhook', 'schedule')",
+                    ]
+                    # Terminal statuses only (worker writes ``success`` /
+                    # ``failed`` / ``timeout`` / ``skipped``; ``completed``
+                    # kept defensively). Pending/running runs aren't log
+                    # entries yet.
+                    schedule_where = [
+                        "r.user_id = :user_id",
+                        "r.status IN ('success', 'completed', 'failed', 'timeout', 'skipped')",
+                    ]
+                    workflow_where = ["wr.user_id = :user_id"]
 
-                # One normalized timeline over the three event sources.
+                # One normalized timeline over five event sources.
                 # ``payload`` carries the per-type detail; the outer query
-                # paginates the merged, time-ordered result.
-                sql = f"""
-                    SELECT * FROM (
+                # paginates the merged, time-ordered result. level /
+                # event_type / search are pushed into each branch so a
+                # filtered request only scans the branches it can match.
+                branches = [
+                    {
+                        "name": "chat",
+                        "level": "COALESCE(l.data->>'level', 'info')",
+                        "summary": "l.data->>'question'",
+                        "where": chat_where,
+                        "sql": """
                         SELECT 'chat' AS event_type,
                                CAST(l.id AS text) AS id,
                                l.user_id AS user_id,
                                l.timestamp AS timestamp,
-                               COALESCE(l.data->>'level', 'info') AS level,
+                               {level} AS level,
                                COALESCE(l.data->>'action', 'stream_answer') AS action,
-                               l.data->>'question' AS summary,
+                               {summary} AS summary,
                                l.data AS payload
                         FROM user_logs l
-                        WHERE {' AND '.join(chat_where)}
-                        UNION ALL
-                        SELECT 'system',
-                               CAST(s.id AS text),
-                               s.user_id,
-                               s.timestamp,
-                               COALESCE(s.level, 'error'),
-                               COALESCE(s.endpoint, 'request'),
-                               s.query,
+                        WHERE {where}
+                        """,
+                    },
+                    {
+                        "name": "system",
+                        "level": "'error'",
+                        "summary": "s.query",
+                        "where": system_where,
+                        "sql": """
+                        SELECT 'system' AS event_type,
+                               CAST(s.id AS text) AS id,
+                               s.user_id AS user_id,
+                               s.timestamp AS timestamp,
+                               {level} AS level,
+                               COALESCE(s.endpoint, 'request') AS action,
+                               {summary} AS summary,
                                jsonb_build_object(
                                    'endpoint', s.endpoint,
                                    'stacks', s.stacks
-                               )
+                               ) AS payload
                         FROM stack_logs s
-                        WHERE {' AND '.join(system_where)}
-                        UNION ALL
-                        SELECT 'webhook',
-                               CAST(s.id AS text),
-                               s.user_id,
-                               s.timestamp,
-                               COALESCE(s.level, 'info'),
-                               'webhook_run',
-                               s.query,
+                        WHERE {where}
+                        """,
+                    },
+                    {
+                        "name": "webhook",
+                        "level": "COALESCE(s.level, 'info')",
+                        "summary": "s.query",
+                        "where": webhook_where,
+                        "sql": """
+                        SELECT 'webhook' AS event_type,
+                               CAST(s.id AS text) AS id,
+                               s.user_id AS user_id,
+                               s.timestamp AS timestamp,
+                               {level} AS level,
+                               'webhook_run' AS action,
+                               {summary} AS summary,
                                jsonb_build_object(
                                    'endpoint', s.endpoint,
                                    'stacks', s.stacks
-                               )
+                               ) AS payload
                         FROM stack_logs s
-                        WHERE {' AND '.join(webhook_where)}
-                        UNION ALL
-                        SELECT 'workflow',
-                               CAST(wr.id AS text),
-                               wr.user_id,
-                               COALESCE(wr.ended_at, wr.started_at),
-                               CASE
-                                   WHEN wr.status = 'failed' THEN 'error'
-                                   ELSE 'info'
-                               END,
-                               'workflow_run',
-                               COALESCE(wr.inputs->>'query', w.name, 'Workflow run'),
+                        WHERE {where}
+                        """,
+                    },
+                    {
+                        "name": "workflow",
+                        "level": (
+                            "CASE WHEN wr.status = 'failed' "
+                            "THEN 'error' ELSE 'info' END"
+                        ),
+                        "summary": (
+                            "COALESCE(wr.inputs->>'query', w.name, 'Workflow run')"
+                        ),
+                        "where": workflow_where,
+                        "sql": """
+                        SELECT 'workflow' AS event_type,
+                               CAST(wr.id AS text) AS id,
+                               wr.user_id AS user_id,
+                               COALESCE(wr.ended_at, wr.started_at) AS timestamp,
+                               {level} AS level,
+                               'workflow_run' AS action,
+                               {summary} AS summary,
                                jsonb_build_object(
                                    'status', wr.status,
                                    'workflow_name', w.name,
@@ -766,22 +915,31 @@ class GetUserLogs(Resource):
                                    'steps', wr.steps,
                                    'started_at', wr.started_at,
                                    'finished_at', wr.ended_at
-                               )
+                               ) AS payload
                         FROM workflow_runs wr
                         LEFT JOIN workflows w ON w.id = wr.workflow_id
-                        WHERE {' AND '.join(workflow_where)}
-                        UNION ALL
-                        SELECT 'schedule',
-                               CAST(r.id AS text),
-                               r.user_id,
-                               COALESCE(r.finished_at, r.started_at, r.scheduled_for),
-                               CASE
-                                   WHEN r.status IN ('failed', 'timeout') THEN 'error'
-                                   WHEN r.status = 'skipped' THEN 'warning'
-                                   ELSE 'info'
-                               END,
-                               'scheduled_run',
-                               COALESCE(sc.name, sc.instruction, 'Scheduled run'),
+                        WHERE {where}
+                        """,
+                    },
+                    {
+                        "name": "schedule",
+                        "level": (
+                            "CASE WHEN r.status IN ('failed', 'timeout') "
+                            "THEN 'error' WHEN r.status = 'skipped' "
+                            "THEN 'warning' ELSE 'info' END"
+                        ),
+                        "summary": (
+                            "COALESCE(sc.name, sc.instruction, 'Scheduled run')"
+                        ),
+                        "where": schedule_where,
+                        "sql": """
+                        SELECT 'schedule' AS event_type,
+                               CAST(r.id AS text) AS id,
+                               r.user_id AS user_id,
+                               COALESCE(r.finished_at, r.started_at, r.scheduled_for) AS timestamp,
+                               {level} AS level,
+                               'scheduled_run' AS action,
+                               {summary} AS summary,
                                jsonb_build_object(
                                    'status', r.status,
                                    'trigger_source', r.trigger_source,
@@ -796,30 +954,49 @@ class GetUserLogs(Resource):
                                    'scheduled_for', r.scheduled_for,
                                    'started_at', r.started_at,
                                    'finished_at', r.finished_at
-                               )
+                               ) AS payload
                         FROM schedule_runs r
                         LEFT JOIN schedules sc ON sc.id = r.schedule_id
-                        WHERE {' AND '.join(schedule_where)}
-                    ) ev
-                """
-                outer = []
+                        WHERE {where}
+                        """,
+                    },
+                ]
+
                 if level:
-                    outer.append("ev.level = :level")
                     params["level"] = level
-                if event_type:
-                    outer.append("ev.event_type = :event_type")
-                    params["event_type"] = event_type
                 if search:
-                    outer.append("ev.summary ILIKE :search ESCAPE '\\'")
                     escaped = (
                         search.replace("\\", "\\\\")
                         .replace("%", "\\%")
                         .replace("_", "\\_")
                     )
                     params["search"] = f"%{escaped}%"
-                if outer:
-                    sql += " WHERE " + " AND ".join(outer)
-                sql += " ORDER BY ev.timestamp DESC LIMIT :limit OFFSET :offset"
+
+                branch_sqls = []
+                for branch in branches:
+                    if event_type and branch["name"] != event_type:
+                        continue
+                    where = list(branch["where"])
+                    if level:
+                        where.append(f"{branch['level']} = :level")
+                    if search:
+                        where.append(
+                            f"{branch['summary']} ILIKE :search ESCAPE '\\'"
+                        )
+                    branch_sqls.append(
+                        branch["sql"].format(
+                            level=branch["level"],
+                            summary=branch["summary"],
+                            where=" AND ".join(where),
+                        )
+                    )
+
+                sql = (
+                    "SELECT * FROM ("
+                    + " UNION ALL ".join(branch_sqls)
+                    + ") ev ORDER BY ev.timestamp DESC"
+                    " LIMIT :limit OFFSET :offset"
+                )
 
                 rows = conn.execute(_sql_text(sql), params).fetchall()
 

application/storage/db/repositories/tool_call_attempts.py

@@ -153,11 +153,18 @@ class ToolCallAttemptsRepository:
         return result_proxy.rowcount > 0
 
     def mark_failed(self, call_id: str, error: str) -> bool:
-        """Flip ``proposed`` → ``failed`` with the exception text."""
+        """Flip ``proposed`` → ``failed`` with the exception text.
+
+        The status guard matters: ``call_id`` is the table-wide PK and
+        LLMs have been observed reusing ids ("call_0"-style). Without it,
+        a duplicate id hitting an error path would flip an already
+        ``executed``/``confirmed`` row — possibly another request's —
+        to ``failed``, and nothing ever repairs that.
+        """
         result = self._conn.execute(
             text(
                 "UPDATE tool_call_attempts SET status = 'failed', error = :error "
-                "WHERE call_id = :call_id"
+                "WHERE call_id = :call_id AND status = 'proposed'"
             ),
             {"call_id": call_id, "error": error},
         )

application/storage/db/repositories/user_logs.py

@@ -1,14 +1,8 @@
-"""Repository for the ``user_logs`` table.
+"""Repository for the ``user_logs`` table (write-only).
 
-Covers every operation the legacy Mongo code performs on
-``user_logs_collection``:
-
-1. ``insert_one`` in logging.py (per-request activity log via
-   ``_log_to_mongodb`` — note: the *Mongo* variable is confusingly named
-   ``user_logs_collection`` but points at the ``user_logs`` Mongo
-   collection, not ``stack_logs``)
-2. ``insert_one`` in answer/routes/base.py (per-stream log entry)
-3. ``find`` with sort/skip/limit in analytics/routes.py (paginated log list)
+The single production write site is the per-stream log entry in
+answer/routes/base.py. Reads go through the unified timeline query in
+api/user/analytics/routes.py (GetUserLogs).
 """
 
 from __future__ import annotations
@@ -19,7 +13,6 @@ from typing import Optional
 
 from sqlalchemy import Connection, text
 
-from application.storage.db.base_repository import row_to_dict
 from application.storage.db.serialization import PGNativeJSONEncoder
 
 
@@ -52,65 +45,6 @@ class UserLogsRepository:
             },
         )
 
-    def list_paginated(
-        self,
-        *,
-        user_id: Optional[str] = None,
-        api_key: Optional[str] = None,
-        page: int = 1,
-        page_size: int = 10,
-    ) -> tuple[list[dict], bool]:
-        """Return ``(rows, has_more)`` for the requested page.
-
-        Mirrors the Mongo ``find(query).sort().skip().limit(page_size+1)``
-        pattern used in analytics/routes.py.
-        """
-        clauses: list[str] = []
-        params: dict = {"limit": page_size + 1, "offset": (page - 1) * page_size}
-        if user_id is not None:
-            clauses.append("user_id = :user_id")
-            params["user_id"] = user_id
-        if api_key is not None:
-            clauses.append("data->>'api_key' = :api_key")
-            params["api_key"] = api_key
-        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
-        result = self._conn.execute(
-            text(
-                f"SELECT * FROM user_logs {where} ORDER BY timestamp DESC LIMIT :limit OFFSET :offset"
-            ),
-            params,
-        )
-        rows = [row_to_dict(r) for r in result.fetchall()]
-        has_more = len(rows) > page_size
-        return rows[:page_size], has_more
-
-    def find_by_api_key(
-        self,
-        api_key: str,
-        *,
-        timestamp_gte: Optional[datetime] = None,
-        timestamp_lt: Optional[datetime] = None,
-        limit: Optional[int] = None,
-    ) -> list[dict]:
-        """Return user_logs rows whose ``data->>'api_key'`` matches ``api_key``.
-
-        Replacement for the legacy Mongo filter by top-level ``api_key``;
-        on the PG side the per-request payload lives in ``data`` JSONB,
-        so the filter reaches in via ``data->>'api_key'``. Rows are
-        ordered by ``timestamp DESC`` to match the Mongo sort.
-        """
-        clauses = ["data->>'api_key' = :api_key"]
-        params: dict = {"api_key": api_key}
-        if timestamp_gte is not None:
-            clauses.append("timestamp >= :timestamp_gte")
-            params["timestamp_gte"] = timestamp_gte
-        if timestamp_lt is not None:
-            clauses.append("timestamp < :timestamp_lt")
-            params["timestamp_lt"] = timestamp_lt
-        where = " AND ".join(clauses)
-        sql = f"SELECT * FROM user_logs WHERE {where} ORDER BY timestamp DESC"
-        if limit is not None:
-            sql += " LIMIT :limit"
-            params["limit"] = limit
-        result = self._conn.execute(text(sql), params)
-        return [row_to_dict(r) for r in result.fetchall()]
+    # NOTE: reads live in the unified timeline query in
+    # api/user/analytics/routes.py (GetUserLogs) — extend that rather
+    # than re-adding per-table readers here.

frontend/src/settings/Analytics.tsx

@@ -7,7 +7,7 @@ import {
   Title,
   Tooltip,
 } from 'chart.js';
-import { useEffect, useMemo, useState } from 'react';
+import { useEffect, useMemo, useRef, useState } from 'react';
 import { Bar } from 'react-chartjs-2';
 import { useTranslation } from 'react-i18next';
 import { useSelector } from 'react-redux';
@@ -162,7 +162,20 @@ export default function Analytics({ agentId }: AnalyticsProps) {
       ? primaryColor
       : SERIES_COLORS[(index - 1) % SERIES_COLORS.length];
 
+  // Monotonic request ids, one per chart: a response only lands if no
+  // newer request for that chart was issued meanwhile, so an
+  // out-of-order response can't leave a chart showing data for a
+  // different filter combination than the controls.
+  const requestIds = useRef({
+    messages: 0,
+    tokens: 0,
+    feedback: 0,
+    tools: 0,
+    schedules: 0,
+  });
+
   const fetchMessagesData = async (agent_id?: string, filter?: string) => {
+    const reqId = ++requestIds.current.messages;
     setLoadingMessages(true);
     try {
       const response = await userService.getMessageAnalytics(
@@ -174,11 +187,12 @@ export default function Analytics({ agentId }: AnalyticsProps) {
       );
       if (!response.ok) throw new Error('Failed to fetch analytics data');
       const data = await response.json();
+      if (reqId !== requestIds.current.messages) return;
       setMessagesData(data.messages);
     } catch (error) {
       console.error(error);
     } finally {
-      setLoadingMessages(false);
+      if (reqId === requestIds.current.messages) setLoadingMessages(false);
     }
   };
 
@@ -188,6 +202,7 @@ export default function Analytics({ agentId }: AnalyticsProps) {
     groupBy?: TokenGroupBy,
     sideChannel?: boolean,
   ) => {
+    const reqId = ++requestIds.current.tokens;
     setLoadingTokens(true);
     try {
       const response = await userService.getTokenAnalytics(
@@ -201,16 +216,18 @@ export default function Analytics({ agentId }: AnalyticsProps) {
       );
       if (!response.ok) throw new Error('Failed to fetch analytics data');
       const data = await response.json();
+      if (reqId !== requestIds.current.tokens) return;
       setTokenUsageData(data.token_usage);
       setTokenSeries(data.series);
     } catch (error) {
       console.error(error);
     } finally {
-      setLoadingTokens(false);
+      if (reqId === requestIds.current.tokens) setLoadingTokens(false);
     }
   };
 
   const fetchFeedbackData = async (agent_id?: string, filter?: string) => {
+    const reqId = ++requestIds.current.feedback;
     setLoadingFeedback(true);
     try {
       const response = await userService.getFeedbackAnalytics(
@@ -222,15 +239,17 @@ export default function Analytics({ agentId }: AnalyticsProps) {
       );
       if (!response.ok) throw new Error('Failed to fetch analytics data');
       const data = await response.json();
+      if (reqId !== requestIds.current.feedback) return;
       setFeedbackData(data.feedback);
     } catch (error) {
       console.error(error);
     } finally {
-      setLoadingFeedback(false);
+      if (reqId === requestIds.current.feedback) setLoadingFeedback(false);
     }
   };
 
   const fetchToolsData = async (agent_id?: string, filter?: string) => {
+    const reqId = ++requestIds.current.tools;
     setLoadingTools(true);
     try {
       const response = await userService.getToolAnalytics(
@@ -242,15 +261,17 @@ export default function Analytics({ agentId }: AnalyticsProps) {
       );
       if (!response.ok) throw new Error('Failed to fetch analytics data');
       const data = await response.json();
+      if (reqId !== requestIds.current.tools) return;
       setToolsData(data.tools);
     } catch (error) {
       console.error(error);
     } finally {
-      setLoadingTools(false);
+      if (reqId === requestIds.current.tools) setLoadingTools(false);
     }
   };
 
   const fetchScheduleData = async (agent_id?: string, filter?: string) => {
+    const reqId = ++requestIds.current.schedules;
     setLoadingSchedules(true);
     try {
       const response = await userService.getScheduleAnalytics(
@@ -262,11 +283,12 @@ export default function Analytics({ agentId }: AnalyticsProps) {
       );
       if (!response.ok) throw new Error('Failed to fetch analytics data');
       const data = await response.json();
+      if (reqId !== requestIds.current.schedules) return;
       setScheduleData(data.runs);
     } catch (error) {
       console.error(error);
     } finally {
-      setLoadingSchedules(false);
+      if (reqId === requestIds.current.schedules) setLoadingSchedules(false);
     }
   };
 

frontend/src/settings/Logs.tsx

@@ -45,6 +45,11 @@ export default function Logs({ agentId, tableHeader }: LogsProps) {
   const filterKeyRef = useRef(filterKey);
   filterKeyRef.current = filterKey;
   const isFirstRender = useRef(true);
+  // Set synchronously by the reset effect so the fetch effect — which
+  // runs later in the same commit, still seeing the pre-reset `page` —
+  // skips that cycle instead of fetching a stale page under the new
+  // filters (which could latch hasMore=false and freeze the list).
+  const resetPendingRef = useRef(false);
 
   useEffect(() => {
     const handle = setTimeout(() => setSearch(searchInput.trim()), 400);
@@ -56,6 +61,7 @@ export default function Logs({ agentId, tableHeader }: LogsProps) {
       isFirstRender.current = false;
       return;
     }
+    resetPendingRef.current = true;
     setLogsByPage({});
     setPage(1);
     setHasMore(true);
@@ -65,6 +71,7 @@ export default function Logs({ agentId, tableHeader }: LogsProps) {
     if (logsByPage[page] && logsByPage[page].length > 0) return;
 
     const issuedKey = filterKey;
+    const issuedPage = page;
     setLoadingLogs(true);
     try {
       const response = await userService.getLogs(
@@ -80,11 +87,11 @@ export default function Logs({ agentId, tableHeader }: LogsProps) {
       );
       if (!response.ok) throw new Error('Failed to fetch logs');
       const data = await response.json();
-      if (issuedKey !== filterKeyRef.current) return;
+      if (issuedKey !== filterKeyRef.current || resetPendingRef.current) return;
 
       setLogsByPage((prev) => ({
         ...prev,
-        [page]: data.logs,
+        [issuedPage]: data.logs,
       }));
       setHasMore(data.has_more);
     } catch (error) {
@@ -97,6 +104,13 @@ export default function Logs({ agentId, tableHeader }: LogsProps) {
   // `logsByPage` is a dependency so the fetch re-fires after a filter
   // change clears the cache; the early-return guard keeps it from looping.
   useEffect(() => {
+    if (resetPendingRef.current) {
+      // The reset effect ran in this commit; this closure still sees
+      // pre-reset state. Its queued updates re-run this effect with
+      // the clean values.
+      resetPendingRef.current = false;
+      return;
+    }
     if (hasMore) fetchLogs();
   }, [page, agentId, levelFilter, typeFilter, search, logsByPage]);
 

scripts/db/backfill_tool_attempts_attribution.py

@@ -0,0 +1,134 @@
+"""Backfill ``tool_call_attempts.user_id`` / ``agent_id`` (migration 0018).
+
+New rows are stamped at propose time by the tool executor. This script
+fills historical rows from data we already trust; tiers run in one
+transaction so each later tier sees only the rows still NULL.
+
+Tiers
+-----
+1. Parent message (high confidence). Rows with a ``message_id`` copy the
+   message's ``user_id`` and the conversation's ``agent_id``.
+2. Schedule-run window (medium confidence). Headless rows never had a
+   message; a run's tool calls always fall inside that run's
+   started/finished window, so copy attribution from ``schedule_runs`` —
+   but only when exactly one run window contains the attempt, to avoid
+   cross-user misattribution on overlapping runs.
+
+Rows matching neither tier (e.g. pre-0018 webhook runs, or attempts
+whose parent message was deleted) are left NULL on purpose: the
+analytics reader treats unattributable rows as invisible rather than
+guessing an owner.
+
+Usage::
+
+    # Dry-run (default): runs the fills in a rolled-back transaction and
+    # reports exactly how many rows each tier would touch.
+    python scripts/db/backfill_tool_attempts_attribution.py
+
+    # Commit the backfill.
+    python scripts/db/backfill_tool_attempts_attribution.py --apply
+
+Exit codes:
+    0 — success (dry-run or apply)
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from sqlalchemy import text  # noqa: E402
+
+from application.storage.db.engine import get_engine  # noqa: E402
+
+
+# Tier 1: parent message → user, conversation → agent.
+_TIER1 = text(
+    """
+    UPDATE tool_call_attempts t
+       SET user_id = m.user_id,
+           agent_id = c.agent_id
+      FROM conversation_messages m
+      LEFT JOIN conversations c ON c.id = m.conversation_id
+     WHERE t.message_id = m.id
+       AND t.user_id IS NULL
+    """
+)
+
+# Tier 2: headless rows via an unambiguous schedule-run time window.
+_TIER2 = text(
+    """
+    UPDATE tool_call_attempts t
+       SET user_id = r.user_id,
+           agent_id = r.agent_id
+      FROM schedule_runs r
+     WHERE t.user_id IS NULL
+       AND t.message_id IS NULL
+       AND r.started_at IS NOT NULL
+       AND r.finished_at IS NOT NULL
+       AND t.attempted_at BETWEEN r.started_at AND r.finished_at
+       AND (
+           SELECT COUNT(*) FROM schedule_runs r2
+            WHERE r2.started_at IS NOT NULL
+              AND r2.finished_at IS NOT NULL
+              AND t.attempted_at BETWEEN r2.started_at AND r2.finished_at
+       ) = 1
+    """
+)
+
+_COUNT_NULL = text(
+    "SELECT count(*) FROM tool_call_attempts WHERE user_id IS NULL"
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Backfill tool_call_attempts.user_id/agent_id from existing data."
+        ),
+    )
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Commit the backfill. Default is a rolled-back dry-run.",
+    )
+    args = parser.parse_args()
+
+    engine = get_engine()
+    with engine.connect() as conn:
+        trans = conn.begin()
+        try:
+            # A one-shot maintenance UPDATE can run well past the engine's
+            # 30s per-statement guardrail; lift it for this transaction.
+            conn.execute(text("SET LOCAL statement_timeout = 0"))
+
+            before = conn.execute(_COUNT_NULL).scalar_one()
+
+            t1 = conn.execute(_TIER1).rowcount or 0
+            t2 = conn.execute(_TIER2).rowcount or 0
+
+            after = conn.execute(_COUNT_NULL).scalar_one()
+
+            print(f"NULL user_id rows before:           {before}")
+            print(f"  tier 1 (parent message):          {t1}")
+            print(f"  tier 2 (schedule-run window):     {t2}")
+            print(f"NULL user_id rows remaining:        {after}")
+
+            if args.apply:
+                trans.commit()
+                print("\nCommitted.")
+            else:
+                trans.rollback()
+                print("\nDry run — rolled back. Re-run with --apply to commit.")
+        except Exception:
+            trans.rollback()
+            raise
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/agents/test_tool_executor_three_phase.py

@@ -251,6 +251,23 @@ class TestRepository:
         assert row["status"] == "failed"
         assert row["error"] == "kaboom"
 
+    def test_mark_failed_leaves_executed_row_untouched(self, pg_conn):
+        """A late error for a reused ``call_id`` ("call_0"-style) must
+        not flip an already-executed row (see ``mark_failed``)."""
+        from application.storage.db.repositories.tool_call_attempts import (
+            ToolCallAttemptsRepository,
+        )
+
+        repo = ToolCallAttemptsRepository(pg_conn)
+        repo.record_proposed("c-dup", "tool", "act", {})
+        # No message_id ⇒ lands straight in 'confirmed' (still non-proposed).
+        assert repo.mark_executed("c-dup", {"out": "ok"}) is True
+        # Late error path hits the same id — the guard rejects the update.
+        assert repo.mark_failed("c-dup", "boom") is False
+        row = _select_attempt(pg_conn, "c-dup")
+        assert row["status"] == "confirmed"
+        assert row["error"] is None
+
 
 @pytest.mark.unit
 class TestDefaultToolJournaling:

tests/api/user/test_analytics.py

@@ -78,8 +78,8 @@ class TestResolveAgent:
     def test_no_agent_when_no_id(self, pg_conn):
         from application.api.user.analytics.routes import _resolve_agent
 
-        assert _resolve_agent(pg_conn, None, "u") == (None, "", None)
-        assert _resolve_agent(pg_conn, "", "u") == (None, "", None)
+        assert _resolve_agent(pg_conn, None, "u") == (None, None, None)
+        assert _resolve_agent(pg_conn, "", "u") == (None, None, None)
 
     def test_returns_key_and_id_for_owned_agent(self, pg_conn):
         from application.api.user.analytics.routes import _resolve_agent
@@ -95,7 +95,7 @@ class TestResolveAgent:
         assert api_key == "secret-api-key"
         assert agent_pg_id == str(agent["id"])
 
-    def test_keyless_agent_gets_no_match_sentinel(self, pg_conn):
+    def test_keyless_agent_yields_none_key(self, pg_conn):
         from application.api.user.analytics.routes import _resolve_agent
         from application.storage.db.repositories.agents import AgentsRepository
 
@@ -105,9 +105,11 @@ class TestResolveAgent:
         resolved, api_key, agent_pg_id = _resolve_agent(
             pg_conn, str(agent["id"]), "owner"
         )
-        # The empty-string key matches no rows; agent_id still matches.
+        # Draft agents store key=''; the resolved api_key must be None
+        # so the filter can't match key-less rows across users (see
+        # _resolve_agent). agent_id still matches the agent's rows.
         assert resolved is not None
-        assert api_key == ""
+        assert api_key is None
         assert agent_pg_id == str(agent["id"])
 
     def test_no_match_for_other_users_agent(self, pg_conn):
@@ -119,7 +121,7 @@ class TestResolveAgent:
         )
         assert _resolve_agent(pg_conn, str(agent["id"]), "other-user") == (
             None,
-            "",
+            None,
             None,
         )
 
@@ -427,3 +429,678 @@ class TestGetUserLogs:
             request.decoded_token = {"sub": "u"}
             response = GetUserLogs().post()
         assert response.status_code == 400
+
+
+def _seed_stack_log(
+    pg_conn, *, user_id, api_key="", level="error",
+    endpoint="stream", query="secret prompt", activity_id=None,
+):
+    import uuid as _uuid
+
+    from application.storage.db.repositories.stack_logs import (
+        StackLogsRepository,
+    )
+
+    StackLogsRepository(pg_conn).insert(
+        activity_id=activity_id or str(_uuid.uuid4()),
+        endpoint=endpoint,
+        level=level,
+        user_id=user_id,
+        api_key=api_key,
+        query=query,
+        stacks=[{"component": "error", "data": {"message": "boom"}}],
+    )
+
+
+def _post_logs(app, pg_conn, user, body):
+    from application.api.user.analytics.routes import GetUserLogs
+
+    with _patch_analytics_db(pg_conn), app.test_request_context(
+        "/api/get_user_logs", method="POST", json=body
+    ):
+        from flask import request
+        request.decoded_token = {"sub": user}
+        return GetUserLogs().post()
+
+
+class TestCrossTenantIsolation:
+    """Pinned regression for the '' api_key sentinel leak.
+
+    Every normal (non-API-key) request's stack_logs rows store
+    api_key='' (logging.py defaults the missing attribute to ""). An
+    agent filter that resolved to an '' sentinel and dropped the
+    user_id clause therefore matched EVERY tenant's error logs.
+    """
+
+    def test_bogus_agent_id_leaks_nothing(self, app, pg_conn):
+        import uuid as _uuid
+
+        _seed_stack_log(pg_conn, user_id="victim", api_key="")
+
+        response = _post_logs(
+            app,
+            pg_conn,
+            "attacker",
+            {"api_key_id": str(_uuid.uuid4()), "event_type": "system"},
+        )
+        assert response.status_code == 200
+        assert response.json["logs"] == []
+        assert response.json["has_more"] is False
+
+    def test_own_draft_agent_leaks_nothing(self, app, pg_conn):
+        from application.storage.db.repositories.agents import AgentsRepository
+
+        _seed_stack_log(pg_conn, user_id="victim", api_key="")
+        # Draft agents legitimately store key='' — filtering by one must
+        # not match other tenants' key-less ('') stack_logs rows.
+        draft = AgentsRepository(pg_conn).create(
+            "attacker", "draft-agent", "draft", key="",
+        )
+
+        response = _post_logs(
+            app,
+            pg_conn,
+            "attacker",
+            {"api_key_id": str(draft["id"]), "event_type": "system"},
+        )
+        assert response.status_code == 200
+        assert response.json["logs"] == []
+
+    def test_other_users_agent_id_returns_empty_tokens(self, app, pg_conn):
+        from application.api.user.analytics.routes import GetTokenAnalytics
+        from application.storage.db.repositories.agents import AgentsRepository
+        from application.storage.db.repositories.token_usage import (
+            TokenUsageRepository,
+        )
+
+        b_agent = AgentsRepository(pg_conn).create(
+            "user-b", "b-agent", "published", key="b-key",
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="user-b", api_key="b-key", prompt_tokens=777,
+            generated_tokens=333,
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="user-a", prompt_tokens=10, generated_tokens=5,
+        )
+
+        with _patch_analytics_db(pg_conn), app.test_request_context(
+            "/api/get_token_analytics",
+            method="POST",
+            json={"api_key_id": str(b_agent["id"])},
+        ):
+            from flask import request
+            request.decoded_token = {"sub": "user-a"}
+            response = GetTokenAnalytics().post()
+        assert response.status_code == 200
+        # Unresolved filter == explicit empty, not "fall back to my data".
+        assert sum(response.json["token_usage"].values()) == 0
+
+
+class TestGetToolAnalytics:
+    def _post(self, app, pg_conn, user, body):
+        from application.api.user.analytics.routes import GetToolAnalytics
+
+        with _patch_analytics_db(pg_conn), app.test_request_context(
+            "/api/get_tool_analytics", method="POST", json=body
+        ):
+            from flask import request
+            request.decoded_token = {"sub": user}
+            return GetToolAnalytics().post()
+
+    def test_counts_terminal_attempts_only(self, app, pg_conn):
+        from application.storage.db.repositories.tool_call_attempts import (
+            ToolCallAttemptsRepository,
+        )
+
+        repo = ToolCallAttemptsRepository(pg_conn)
+        repo.record_proposed("c1", "brave", "search", {}, user_id="u-tools")
+        repo.mark_executed("c1", "ok")
+        repo.record_proposed("c2", "brave", "search", {}, user_id="u-tools")
+        repo.mark_failed("c2", "boom")
+        # Stuck proposed row: neither success nor failure — must not
+        # count as a phantom success.
+        repo.record_proposed("c3", "brave", "search", {}, user_id="u-tools")
+        # Another user's attempt is invisible.
+        repo.record_proposed("c4", "brave", "search", {}, user_id="someone")
+        repo.mark_executed("c4", "ok")
+
+        response = self._post(app, pg_conn, "u-tools", {})
+        assert response.status_code == 200
+        tools = {t["tool_name"]: t for t in response.json["tools"]}
+        assert tools["brave"]["calls"] == 2
+        assert tools["brave"]["failures"] == 1
+
+    def test_unknown_agent_returns_empty(self, app, pg_conn):
+        import uuid as _uuid
+
+        response = self._post(
+            app, pg_conn, "u-tools", {"api_key_id": str(_uuid.uuid4())}
+        )
+        assert response.status_code == 200
+        assert response.json["tools"] == []
+
+    def test_filters_by_agent_stamp(self, app, pg_conn):
+        from application.storage.db.repositories.agents import AgentsRepository
+        from application.storage.db.repositories.tool_call_attempts import (
+            ToolCallAttemptsRepository,
+        )
+
+        agent = AgentsRepository(pg_conn).create(
+            "u-tools", "a", "published", key="tk",
+        )
+        repo = ToolCallAttemptsRepository(pg_conn)
+        repo.record_proposed(
+            "c10", "ntfy", "send", {},
+            user_id="u-tools", agent_id=str(agent["id"]),
+        )
+        repo.mark_executed("c10", "ok")
+        repo.record_proposed("c11", "brave", "search", {}, user_id="u-tools")
+        repo.mark_executed("c11", "ok")
+
+        response = self._post(
+            app, pg_conn, "u-tools", {"api_key_id": str(agent["id"])}
+        )
+        assert response.status_code == 200
+        names = [t["tool_name"] for t in response.json["tools"]]
+        assert names == ["ntfy"]
+
+
+class TestGetScheduleAnalytics:
+    def _seed_run(self, pg_conn, user, status, *, agent_id=None):
+        import datetime as _dt
+
+        from application.storage.db.repositories.schedule_runs import (
+            ScheduleRunsRepository,
+        )
+        from application.storage.db.repositories.schedules import (
+            SchedulesRepository,
+        )
+
+        now = _dt.datetime.now(_dt.timezone.utc)
+        # The schedules_once_run_at_chk constraint requires run_at on
+        # once-type schedules.
+        schedule = SchedulesRepository(pg_conn).create(
+            user, agent_id, "once", "do the thing", run_at=now,
+        )
+        run = ScheduleRunsRepository(pg_conn).record_pending(
+            str(schedule["id"]), user, agent_id, now,
+        )
+        ScheduleRunsRepository(pg_conn).update(
+            str(run["id"]),
+            {"status": status, "started_at": now, "finished_at": now},
+        )
+        return schedule
+
+    def _post(self, app, pg_conn, user, body):
+        from application.api.user.analytics.routes import GetScheduleAnalytics
+
+        with _patch_analytics_db(pg_conn), app.test_request_context(
+            "/api/get_schedule_analytics", method="POST", json=body
+        ):
+            from flask import request
+            request.decoded_token = {"sub": user}
+            return GetScheduleAnalytics().post()
+
+    def test_maps_worker_statuses(self, app, pg_conn):
+        user = "u-sched"
+        self._seed_run(pg_conn, user, "success")
+        self._seed_run(pg_conn, user, "failed")
+        self._seed_run(pg_conn, user, "timeout")
+        self._seed_run(pg_conn, user, "skipped")
+
+        response = self._post(app, pg_conn, user, {})
+        assert response.status_code == 200
+        totals = {"completed": 0, "failed": 0, "skipped": 0}
+        for bucket in response.json["runs"].values():
+            for key in totals:
+                totals[key] += bucket[key]
+        # success → completed; failed + timeout → failed; skipped → skipped.
+        assert totals == {"completed": 1, "failed": 2, "skipped": 1}
+
+    def test_unknown_agent_returns_zeroes(self, app, pg_conn):
+        import uuid as _uuid
+
+        self._seed_run(pg_conn, "u-sched", "success")
+        response = self._post(
+            app, pg_conn, "u-sched", {"api_key_id": str(_uuid.uuid4())}
+        )
+        assert response.status_code == 200
+        assert all(
+            bucket == {"completed": 0, "failed": 0, "skipped": 0}
+            for bucket in response.json["runs"].values()
+        )
+
+
+class TestUnifiedLogsBranches:
+    def test_schedule_runs_appear_with_level_mapping(self, app, pg_conn):
+        helper = TestGetScheduleAnalytics()
+        helper._seed_run(pg_conn, "u-tl", "failed")
+        helper._seed_run(pg_conn, "u-tl", "success")
+
+        response = _post_logs(app, pg_conn, "u-tl", {"event_type": "schedule"})
+        assert response.status_code == 200
+        logs = response.json["logs"]
+        assert {log["event_type"] for log in logs} == {"schedule"}
+        levels = sorted(log["level"] for log in logs)
+        assert levels == ["error", "info"]
+
+    def test_failed_webhook_run_renders_once(self, app, pg_conn):
+        # A failed activity writes BOTH an error row (except) and an info
+        # row (finally) for the same activity_id — the timeline must show
+        # the run once, as the error.
+        _seed_stack_log(
+            pg_conn, user_id="u-wh", endpoint="webhook",
+            level="error", activity_id="act-1",
+        )
+        _seed_stack_log(
+            pg_conn, user_id="u-wh", endpoint="webhook",
+            level="info", activity_id="act-1",
+        )
+
+        response = _post_logs(app, pg_conn, "u-wh", {"event_type": "webhook"})
+        assert response.status_code == 200
+        logs = response.json["logs"]
+        assert len(logs) == 1
+        assert logs[0]["level"] == "error"
+
+    def test_successful_webhook_run_still_shows(self, app, pg_conn):
+        _seed_stack_log(
+            pg_conn, user_id="u-wh2", endpoint="webhook",
+            level="info", activity_id="act-ok",
+        )
+        response = _post_logs(app, pg_conn, "u-wh2", {"event_type": "webhook"})
+        assert response.status_code == 200
+        assert len(response.json["logs"]) == 1
+        assert response.json["logs"][0]["level"] == "info"
+
+    def test_level_filter_applies_per_branch(self, app, pg_conn):
+        from application.storage.db.repositories.user_logs import (
+            UserLogsRepository,
+        )
+
+        UserLogsRepository(pg_conn).insert(
+            user_id="u-lvl",
+            endpoint="stream_answer",
+            data={"action": "stream_answer", "level": "info", "question": "q"},
+        )
+        _seed_stack_log(pg_conn, user_id="u-lvl", level="error")
+
+        response = _post_logs(app, pg_conn, "u-lvl", {"level": "error"})
+        assert response.status_code == 200
+        assert [log["level"] for log in response.json["logs"]] == ["error"]
+
+    def test_non_numeric_page_returns_400(self, app, pg_conn):
+        response = _post_logs(app, pg_conn, "u", {"page": "abc"})
+        assert response.status_code == 400
+
+
+class TestTokenAnalyticsParamCoercion:
+    def test_string_false_disables_side_channel(self, app, pg_conn):
+        from application.api.user.analytics.routes import GetTokenAnalytics
+        from application.storage.db.repositories.token_usage import (
+            TokenUsageRepository,
+        )
+
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-coerce", prompt_tokens=100, generated_tokens=0,
+            source="title",
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-coerce", prompt_tokens=7, generated_tokens=0,
+        )
+
+        with _patch_analytics_db(pg_conn), app.test_request_context(
+            "/api/get_token_analytics",
+            method="POST",
+            json={"include_side_channel": "false"},
+        ):
+            from flask import request
+            request.decoded_token = {"sub": "u-coerce"}
+            response = GetTokenAnalytics().post()
+        assert response.status_code == 200
+        # The JSON string "false" must not truthy-coerce to True: the
+        # 100 side-channel (title) tokens stay excluded.
+        assert sum(response.json["token_usage"].values()) == 7
+
+def _post_resource(app, pg_conn, resource_cls, path, user, body):
+    with _patch_analytics_db(pg_conn), app.test_request_context(
+        path, method="POST", json=body
+    ):
+        from flask import request
+        request.decoded_token = {"sub": user}
+        return resource_cls().post()
+
+
+class TestUnknownAgentShortCircuits:
+    """Messages / feedback mirror the token / tool / schedule contract:
+    a filter that doesn't resolve to one of the caller's agents returns
+    an explicit empty result, never "all my data" (and never a sentinel
+    filter — see TestCrossTenantIsolation)."""
+
+    def test_message_analytics_returns_zeroes(self, app, pg_conn):
+        import uuid as _uuid
+
+        from application.api.user.analytics.routes import GetMessageAnalytics
+
+        _seed_conversation_with_messages(pg_conn, "u-msg", count=2)
+        response = _post_resource(
+            app, pg_conn, GetMessageAnalytics, "/api/get_message_analytics",
+            "u-msg", {"api_key_id": str(_uuid.uuid4())},
+        )
+        assert response.status_code == 200
+        assert sum(response.json["messages"].values()) == 0
+
+    def test_feedback_analytics_returns_zeroes(self, app, pg_conn):
+        import uuid as _uuid
+
+        from application.api.user.analytics.routes import GetFeedbackAnalytics
+
+        _seed_conversation_with_messages(
+            pg_conn, "u-fb0", count=2, feedback_text="like"
+        )
+        response = _post_resource(
+            app, pg_conn, GetFeedbackAnalytics, "/api/get_feedback_analytics",
+            "u-fb0", {"api_key_id": str(_uuid.uuid4())},
+        )
+        assert response.status_code == 200
+        assert all(
+            bucket == {"positive": 0, "negative": 0}
+            for bucket in response.json["feedback"].values()
+        )
+
+
+class TestMessageAnalyticsBuckets:
+    @pytest.mark.parametrize("option", ["last_hour", "last_24_hour"])
+    def test_minute_and_hour_buckets(self, app, pg_conn, option):
+        from application.api.user.analytics.routes import GetMessageAnalytics
+
+        _seed_conversation_with_messages(pg_conn, "u-bkt", count=2)
+        response = _post_resource(
+            app, pg_conn, GetMessageAnalytics, "/api/get_message_analytics",
+            "u-bkt", {"filter_option": option},
+        )
+        assert response.status_code == 200
+        assert sum(response.json["messages"].values()) == 2
+
+
+class TestFeedbackAnalyticsAgentFilter:
+    def test_filters_by_agent_key_or_id(self, app, pg_conn):
+        from application.api.user.analytics.routes import GetFeedbackAnalytics
+        from application.storage.db.repositories.agents import AgentsRepository
+
+        agent = AgentsRepository(pg_conn).create(
+            "u-fb", "fb-agent", "published", key="fb-key",
+        )
+        # count=2 → message 0 'like', message 1 'dislike' per conversation.
+        _seed_conversation_with_messages(
+            pg_conn, "u-fb", count=2, api_key="fb-key", feedback_text="like"
+        )
+        _seed_conversation_with_messages(
+            pg_conn, "u-fb", count=2, feedback_text="like"
+        )
+
+        response = _post_resource(
+            app, pg_conn, GetFeedbackAnalytics, "/api/get_feedback_analytics",
+            "u-fb", {"api_key_id": str(agent["id"])},
+        )
+        assert response.status_code == 200
+        totals = {"positive": 0, "negative": 0}
+        for bucket in response.json["feedback"].values():
+            totals["positive"] += bucket["positive"]
+            totals["negative"] += bucket["negative"]
+        # Only the agent conversation's feedback — not the keyless one.
+        assert totals == {"positive": 1, "negative": 1}
+
+
+class TestTokenAnalyticsGrouping:
+    def test_group_by_model_returns_series(self, app, pg_conn):
+        from application.api.user.analytics.routes import GetTokenAnalytics
+        from application.storage.db.repositories.token_usage import (
+            TokenUsageRepository,
+        )
+
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-grp", prompt_tokens=10, generated_tokens=5,
+            model_id="gpt-x",
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-grp", prompt_tokens=1, generated_tokens=2,
+        )
+
+        response = _post_resource(
+            app, pg_conn, GetTokenAnalytics, "/api/get_token_analytics",
+            "u-grp", {"group_by": "model"},
+        )
+        assert response.status_code == 200
+        assert sum(response.json["token_usage"].values()) == 18
+        series = response.json["series"]
+        # Rows without a model_id group under the 'unknown' key.
+        assert set(series) == {"gpt-x", "unknown"}
+        assert sum(series["gpt-x"].values()) == 15
+        assert sum(series["unknown"].values()) == 3
+
+    def test_filters_by_owned_agent_key_or_id(self, app, pg_conn):
+        from application.api.user.analytics.routes import GetTokenAnalytics
+        from application.storage.db.repositories.agents import AgentsRepository
+        from application.storage.db.repositories.token_usage import (
+            TokenUsageRepository,
+        )
+
+        agent = AgentsRepository(pg_conn).create(
+            "u-tok", "tok-agent", "published", key="tok-key",
+        )
+        # External chat traffic stamps the key; headless runs stamp the
+        # agent_id — the filter must match either shape.
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-tok", api_key="tok-key", prompt_tokens=4,
+            generated_tokens=0,
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-tok", agent_id=str(agent["id"]), prompt_tokens=2,
+            generated_tokens=0,
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-tok", prompt_tokens=100, generated_tokens=0,
+        )
+
+        response = _post_resource(
+            app, pg_conn, GetTokenAnalytics, "/api/get_token_analytics",
+            "u-tok", {"api_key_id": str(agent["id"])},
+        )
+        assert response.status_code == 200
+        assert sum(response.json["token_usage"].values()) == 6
+
+
+class TestScheduleAnalyticsAgentFilter:
+    def test_filters_by_owned_agent(self, app, pg_conn):
+        from application.storage.db.repositories.agents import AgentsRepository
+
+        helper = TestGetScheduleAnalytics()
+        agent = AgentsRepository(pg_conn).create(
+            "u-sched2", "sched-agent", "published", key="sk",
+        )
+        helper._seed_run(pg_conn, "u-sched2", "success", agent_id=str(agent["id"]))
+        helper._seed_run(pg_conn, "u-sched2", "failed")  # no agent
+
+        response = helper._post(
+            app, pg_conn, "u-sched2", {"api_key_id": str(agent["id"])}
+        )
+        assert response.status_code == 200
+        totals = {"completed": 0, "failed": 0, "skipped": 0}
+        for bucket in response.json["runs"].values():
+            for key in totals:
+                totals[key] += bucket[key]
+        assert totals == {"completed": 1, "failed": 0, "skipped": 0}
+
+
+class TestUnifiedLogsFilters:
+    def test_search_matches_summary(self, app, pg_conn):
+        from application.storage.db.repositories.user_logs import (
+            UserLogsRepository,
+        )
+
+        repo = UserLogsRepository(pg_conn)
+        repo.insert(
+            user_id="u-srch", endpoint="stream",
+            data={"question": "how do whales sleep"},
+        )
+        repo.insert(
+            user_id="u-srch", endpoint="stream",
+            data={"question": "unrelated"},
+        )
+
+        response = _post_logs(app, pg_conn, "u-srch", {"search": "whales"})
+        assert response.status_code == 200
+        logs = response.json["logs"]
+        assert len(logs) == 1
+        assert "whales" in logs[0]["question"]
+
+    def test_search_escapes_like_wildcards(self, app, pg_conn):
+        from application.storage.db.repositories.user_logs import (
+            UserLogsRepository,
+        )
+
+        UserLogsRepository(pg_conn).insert(
+            user_id="u-srch2", endpoint="stream",
+            data={"question": "plain text"},
+        )
+        # '%' must be matched literally, not as a wildcard.
+        response = _post_logs(app, pg_conn, "u-srch2", {"search": "%"})
+        assert response.status_code == 200
+        assert response.json["logs"] == []
+
+    def test_invalid_event_type_returns_400(self, app, pg_conn):
+        response = _post_logs(app, pg_conn, "u", {"event_type": "bogus"})
+        assert response.status_code == 400
+
+
+class TestNewEndpointGuards:
+    """401 / invalid-filter guards on the two endpoints new in this PR,
+    mirroring the existing per-endpoint guard tests above."""
+
+    @pytest.mark.parametrize(
+        "resource_name, path",
+        [
+            ("GetToolAnalytics", "/api/get_tool_analytics"),
+            ("GetScheduleAnalytics", "/api/get_schedule_analytics"),
+        ],
+    )
+    def test_returns_401_unauthenticated(self, app, resource_name, path):
+        import application.api.user.analytics.routes as routes
+
+        resource_cls = getattr(routes, resource_name)
+        with app.test_request_context(path, method="POST", json={}):
+            from flask import request
+            request.decoded_token = None
+            response = resource_cls().post()
+        assert response.status_code == 401
+
+    @pytest.mark.parametrize(
+        "resource_name, path",
+        [
+            ("GetToolAnalytics", "/api/get_tool_analytics"),
+            ("GetScheduleAnalytics", "/api/get_schedule_analytics"),
+        ],
+    )
+    def test_invalid_filter_returns_400(self, app, resource_name, path):
+        import application.api.user.analytics.routes as routes
+
+        resource_cls = getattr(routes, resource_name)
+        with app.test_request_context(
+            path, method="POST", json={"filter_option": "nope"}
+        ):
+            from flask import request
+            request.decoded_token = {"sub": "u"}
+            response = resource_cls().post()
+        assert response.status_code == 400
+
+
+class TestTokenAnalyticsGroupByAgent:
+    def test_group_by_agent_resolves_names(self, app, pg_conn):
+        from application.api.user.analytics.routes import GetTokenAnalytics
+        from application.storage.db.repositories.agents import AgentsRepository
+        from application.storage.db.repositories.token_usage import (
+            TokenUsageRepository,
+        )
+
+        agent = AgentsRepository(pg_conn).create(
+            "u-grpa", "Billing Bot", "published", key="ga-key",
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-grpa", agent_id=str(agent["id"]), prompt_tokens=3,
+            generated_tokens=0,
+        )
+        TokenUsageRepository(pg_conn).insert(
+            user_id="u-grpa", prompt_tokens=1, generated_tokens=0,
+        )
+
+        response = _post_resource(
+            app, pg_conn, GetTokenAnalytics, "/api/get_token_analytics",
+            "u-grpa", {"group_by": "agent"},
+        )
+        assert response.status_code == 200
+        # The series is keyed by the agent's display name, not its UUID.
+        assert set(response.json["series"]) == {"Billing Bot", "No agent"}
+
+
+class TestNewEndpointDbErrors:
+    @pytest.mark.parametrize(
+        "resource_name, path",
+        [
+            ("GetToolAnalytics", "/api/get_tool_analytics"),
+            ("GetScheduleAnalytics", "/api/get_schedule_analytics"),
+        ],
+    )
+    def test_db_error_returns_400(self, app, resource_name, path):
+        import application.api.user.analytics.routes as routes
+
+        resource_cls = getattr(routes, resource_name)
+
+        @contextmanager
+        def _broken():
+            raise RuntimeError("boom")
+            yield
+
+        with patch(
+            "application.api.user.analytics.routes.db_readonly", _broken
+        ), app.test_request_context(path, method="POST", json={}):
+            from flask import request
+            request.decoded_token = {"sub": "u"}
+            response = resource_cls().post()
+        assert response.status_code == 400
+
+
+class TestUnifiedLogsWorkflowBranch:
+    def test_workflow_runs_appear_with_payload(self, app, pg_conn):
+        import datetime as _dt
+
+        from application.storage.db.repositories.workflow_runs import (
+            WorkflowRunsRepository,
+        )
+        from application.storage.db.repositories.workflows import (
+            WorkflowsRepository,
+        )
+
+        now = _dt.datetime.now(_dt.timezone.utc)
+        wf = WorkflowsRepository(pg_conn).create("u-wf", "My Flow")
+        WorkflowRunsRepository(pg_conn).create(
+            str(wf["id"]), "u-wf", "failed",
+            inputs={"query": "run it"}, started_at=now, ended_at=now,
+        )
+        WorkflowRunsRepository(pg_conn).create(
+            str(wf["id"]), "u-wf", "completed",
+            started_at=now, ended_at=now,
+        )
+
+        response = _post_logs(app, pg_conn, "u-wf", {"event_type": "workflow"})
+        assert response.status_code == 200
+        logs = response.json["logs"]
+        assert len(logs) == 2
+        assert {log["action"] for log in logs} == {"workflow_run"}
+        assert sorted(log["level"] for log in logs) == ["error", "info"]
+        by_level = {log["level"]: log for log in logs}
+        assert by_level["error"]["status"] == "failed"
+        assert by_level["error"]["workflow_name"] == "My Flow"
+        # Summary falls back to the workflow name when inputs lack a query.
+        assert by_level["info"]["question"] == "My Flow"

tests/e2e/specs/tier-a/analytics.spec.ts

@@ -372,9 +372,10 @@ test.describe('tier-a · token usage analytics', () => {
     browser,
   }) => {
     // User A and user B each have data. A queries with B's api_key_id.
-    // `_resolve_api_key(conn, b_agent_id, a_sub)` returns None (agent
-    // lookup is user-scoped), so the filter is dropped and A sees only
-    // A's own data — never any of B's rows.
+    // `_resolve_agent(conn, b_agent_id, a_sub)` finds nothing (agent
+    // lookup is user-scoped), and the endpoint short-circuits to an
+    // explicit EMPTY result — a filter that doesn't resolve must never
+    // silently widen to "all your data" (and certainly not to B's).
     const aSub = 'e2e-analytics-cross-a';
     const bSub = 'e2e-analytics-cross-b';
     const aToken = signJwt(aSub);
@@ -429,14 +430,16 @@ test.describe('tier-a · token usage analytics', () => {
       expect(body.success).toBe(true);
       const buckets = body.token_usage ?? {};
 
-      // A's total is 10 + 5 = 15. B's 777+333=1110 MUST NOT appear.
+      // An unresolved agent filter returns all-zero buckets. B's
+      // 777+333=1110 MUST NOT appear, and neither must A's own 15 —
+      // the filter explicitly matched nothing.
       const total = Object.values(buckets).reduce((a, b) => a + b, 0);
-      expect(total).toBe(15);
+      expect(total).toBe(0);
 
       // Double-check: B's bucket value cannot have bled in under any
-      // key. No single bucket should exceed A's total.
+      // key.
       for (const value of Object.values(buckets)) {
-        expect(value).toBeLessThanOrEqual(15);
+        expect(value).toBe(0);
       }
     } finally {
       await apiA.dispose();

tests/storage/db/repositories/test_user_logs.py

@@ -1,103 +1,50 @@
-"""Tests for UserLogsRepository against a real Postgres instance."""
+"""Tests for UserLogsRepository against a real Postgres instance.
+
+The repository is now write-only: the read path is the unified timeline
+in ``api/user/analytics/routes.py`` (GetUserLogs), covered by the route
+tests. Inserts are verified with raw SQL here.
+"""
 
 from __future__ import annotations
 
+from sqlalchemy import text
 
 from application.storage.db.repositories.user_logs import UserLogsRepository
 
 
-def _repo(conn) -> UserLogsRepository:
-    return UserLogsRepository(conn)
+def _rows(conn, user_id):
+    result = conn.execute(
+        text(
+            "SELECT * FROM user_logs WHERE user_id = :u ORDER BY timestamp DESC"
+        ),
+        {"u": user_id},
+    )
+    return [dict(r._mapping) for r in result.fetchall()]
 
 
 class TestInsert:
     def test_inserts_log(self, pg_conn):
-        repo = _repo(pg_conn)
+        repo = UserLogsRepository(pg_conn)
         repo.insert(user_id="u1", endpoint="/api/answer", data={"question": "hi"})
-        rows, _ = repo.list_paginated(user_id="u1")
+        rows = _rows(pg_conn, "u1")
         assert len(rows) == 1
         assert rows[0]["data"]["question"] == "hi"
+        assert rows[0]["endpoint"] == "/api/answer"
 
     def test_allows_null_data(self, pg_conn):
-        repo = _repo(pg_conn)
+        repo = UserLogsRepository(pg_conn)
         repo.insert(user_id="u1")
-        rows, _ = repo.list_paginated(user_id="u1")
+        rows = _rows(pg_conn, "u1")
         assert len(rows) == 1
         assert rows[0]["data"] is None
 
-
-class TestListPaginated:
-    def test_paginates(self, pg_conn):
-        repo = _repo(pg_conn)
-        for i in range(5):
-            repo.insert(user_id="u1", data={"i": i})
-        page1, has_more1 = repo.list_paginated(user_id="u1", page=1, page_size=3)
-        assert len(page1) == 3
-        assert has_more1 is True
-        page2, has_more2 = repo.list_paginated(user_id="u1", page=2, page_size=3)
-        assert len(page2) == 2
-        assert has_more2 is False
-
-    def test_filters_by_user(self, pg_conn):
-        repo = _repo(pg_conn)
-        repo.insert(user_id="alice", data={"x": 1})
-        repo.insert(user_id="bob", data={"x": 2})
-        rows, _ = repo.list_paginated(user_id="alice")
-        assert len(rows) == 1
-        assert rows[0]["user_id"] == "alice"
-
-    def test_ordered_by_timestamp_desc(self, pg_conn):
+    def test_explicit_timestamp_is_stored(self, pg_conn):
         from datetime import datetime, timedelta, timezone
 
-        repo = _repo(pg_conn)
+        repo = UserLogsRepository(pg_conn)
         earlier = datetime.now(timezone.utc) - timedelta(minutes=5)
-        later = datetime.now(timezone.utc)
         repo.insert(user_id="u1", data={"order": "first"}, timestamp=earlier)
-        repo.insert(user_id="u1", data={"order": "second"}, timestamp=later)
-        rows, _ = repo.list_paginated(user_id="u1")
+        repo.insert(user_id="u1", data={"order": "second"})
+        rows = _rows(pg_conn, "u1")
         assert rows[0]["data"]["order"] == "second"
-
-
-class TestFindByApiKey:
-    def test_filters_by_api_key(self, pg_conn):
-        repo = _repo(pg_conn)
-        repo.insert(user_id="u1", data={"api_key": "k1", "note": "a"})
-        repo.insert(user_id="u1", data={"api_key": "k2", "note": "b"})
-        repo.insert(user_id="u1", data={"note": "c"})
-        rows = repo.find_by_api_key("k1")
-        assert len(rows) == 1
-        assert rows[0]["data"]["note"] == "a"
-
-    def test_respects_limit(self, pg_conn):
-        repo = _repo(pg_conn)
-        for i in range(5):
-            repo.insert(user_id="u1", data={"api_key": "k", "i": i})
-        rows = repo.find_by_api_key("k", limit=3)
-        assert len(rows) == 3
-
-    def test_ordered_by_timestamp_desc(self, pg_conn):
-        from datetime import datetime, timedelta, timezone
-
-        repo = _repo(pg_conn)
-        earlier = datetime.now(timezone.utc) - timedelta(minutes=5)
-        later = datetime.now(timezone.utc)
-        repo.insert(user_id="u1", data={"api_key": "k", "order": 1}, timestamp=earlier)
-        repo.insert(user_id="u1", data={"api_key": "k", "order": 2}, timestamp=later)
-        rows = repo.find_by_api_key("k")
-        assert rows[0]["data"]["order"] == 2
-
-    def test_respects_timestamp_range(self, pg_conn):
-        from datetime import datetime, timezone
-
-        repo = _repo(pg_conn)
-        inside = datetime(2026, 4, 10, 12, 0, tzinfo=timezone.utc)
-        outside = datetime(2026, 4, 9, 12, 0, tzinfo=timezone.utc)
-        repo.insert(user_id="u1", data={"api_key": "kt", "when": "in"}, timestamp=inside)
-        repo.insert(user_id="u1", data={"api_key": "kt", "when": "out"}, timestamp=outside)
-        rows = repo.find_by_api_key(
-            "kt",
-            timestamp_gte=datetime(2026, 4, 10, tzinfo=timezone.utc),
-            timestamp_lt=datetime(2026, 4, 11, tzinfo=timezone.utc),
-        )
-        assert len(rows) == 1
-        assert rows[0]["data"]["when"] == "in"
+        assert rows[1]["data"]["order"] == "first"