diff --git a/application/alembic/versions/0018_tool_attempts_attribution.py b/application/alembic/versions/0018_tool_attempts_attribution.py index 27dc2a4b..0758d5ea 100644 --- a/application/alembic/versions/0018_tool_attempts_attribution.py +++ b/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) " diff --git a/application/api/user/analytics/routes.py b/application/api/user/analytics/routes.py index 5ff8eb86..0c6839f2 100644 --- a/application/api/user/analytics/routes.py +++ b/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() diff --git a/application/storage/db/repositories/tool_call_attempts.py b/application/storage/db/repositories/tool_call_attempts.py index a6fa6c74..6326f0cf 100644 --- a/application/storage/db/repositories/tool_call_attempts.py +++ b/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}, ) diff --git a/application/storage/db/repositories/user_logs.py b/application/storage/db/repositories/user_logs.py index 7b8cafc1..cb8fc23c 100644 --- a/application/storage/db/repositories/user_logs.py +++ b/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. diff --git a/frontend/src/settings/Analytics.tsx b/frontend/src/settings/Analytics.tsx index 0cc65f80..47b4e135 100644 --- a/frontend/src/settings/Analytics.tsx +++ b/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); } }; diff --git a/frontend/src/settings/Logs.tsx b/frontend/src/settings/Logs.tsx index e50b68a1..d308acd9 100644 --- a/frontend/src/settings/Logs.tsx +++ b/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]); diff --git a/scripts/db/backfill_tool_attempts_attribution.py b/scripts/db/backfill_tool_attempts_attribution.py new file mode 100644 index 00000000..d588f93a --- /dev/null +++ b/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()) diff --git a/tests/agents/test_tool_executor_three_phase.py b/tests/agents/test_tool_executor_three_phase.py index b7f9e5a4..88b01226 100644 --- a/tests/agents/test_tool_executor_three_phase.py +++ b/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: diff --git a/tests/api/user/test_analytics.py b/tests/api/user/test_analytics.py index d1869371..0ee49c0b 100644 --- a/tests/api/user/test_analytics.py +++ b/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" diff --git a/tests/e2e/specs/tier-a/analytics.spec.ts b/tests/e2e/specs/tier-a/analytics.spec.ts index 33d59ea3..b23c3a8b 100644 --- a/tests/e2e/specs/tier-a/analytics.spec.ts +++ b/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(); diff --git a/tests/storage/db/repositories/test_user_logs.py b/tests/storage/db/repositories/test_user_logs.py index 3ae7fea8..aaea9341 100644 --- a/tests/storage/db/repositories/test_user_logs.py +++ b/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"