diff --git a/application/agents/tool_executor.py b/application/agents/tool_executor.py index ae23f6fd..0d059a0d 100644 --- a/application/agents/tool_executor.py +++ b/application/agents/tool_executor.py @@ -30,6 +30,9 @@ def _record_proposed( arguments: Any, *, tool_id: Optional[str] = None, + message_id: Optional[str] = None, + user_id: Optional[str] = None, + agent_id: Optional[str] = None, ) -> bool: """Insert a ``proposed`` row; swallow infra failures so tool calls still run when the journal is unreachable. Returns True iff the row @@ -43,6 +46,13 @@ def _record_proposed( action_name, arguments, tool_id=tool_id if tool_id and looks_like_uuid(tool_id) else None, + message_id=message_id, + user_id=user_id, + agent_id=( + str(agent_id) + if agent_id and looks_like_uuid(str(agent_id)) + else None + ), ) if not inserted: logger.warning( @@ -518,6 +528,17 @@ class ToolExecutor: "arguments": call_args or {}, "result": f"Failed to parse tool call. Invalid tool name format: {llm_name}", } + # Journal the malformed call so it still shows up in tool analytics. + if _record_proposed( + call_id, + "unknown", + llm_name or "unknown", + call_args if isinstance(call_args, dict) else {}, + message_id=self.message_id, + user_id=self.user, + agent_id=self.agent_id, + ): + _mark_failed(call_id, tool_call_data["result"]) yield {"type": "tool_call", "data": {**tool_call_data, "status": "error"}} self.tool_calls.append(tool_call_data) return "Failed to parse tool call.", call_id @@ -541,6 +562,17 @@ class ToolExecutor: "arguments": call_args, "result": f"Tool with ID {tool_id} not found. Available tools: {list(tools_dict.keys())}", } + # Journal the unresolvable call so it still shows up in tool analytics. + if _record_proposed( + call_id, + "unknown", + llm_name or "unknown", + call_args if isinstance(call_args, dict) else {}, + message_id=self.message_id, + user_id=self.user, + agent_id=self.agent_id, + ): + _mark_failed(call_id, f"Tool with ID {tool_id} not found.") yield {"type": "tool_call", "data": {**tool_call_data, "status": "error"}} self.tool_calls.append(tool_call_data) return f"Tool with ID {tool_id} not found.", call_id @@ -566,6 +598,9 @@ class ToolExecutor: action_name, call_args if isinstance(call_args, dict) else {}, tool_id=tool_data.get("id"), + message_id=self.message_id, + user_id=self.user, + agent_id=self.agent_id, ) # Defensive guard: a non-dict ``call_args`` (e.g. malformed # JSON on the resume path) would crash the param walk below diff --git a/application/alembic/versions/0018_tool_attempts_attribution.py b/application/alembic/versions/0018_tool_attempts_attribution.py new file mode 100644 index 00000000..27dc2a4b --- /dev/null +++ b/application/alembic/versions/0018_tool_attempts_attribution.py @@ -0,0 +1,48 @@ +"""0018 tool_call_attempts attribution — record user/agent on the journal. + +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. + +Revision ID: 0018_tool_attempts_attribution +Revises: 0017_oidc_scim +""" + +from typing import Sequence, Union + +from alembic import op + + +revision: str = "0018_tool_attempts_attribution" +down_revision: Union[str, None] = "0017_oidc_scim" +branch_labels: Union[str, Sequence[str], None] = None +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) " + "WHERE user_id IS NOT NULL;" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS tool_call_attempts_user_ts_idx;") + op.execute("ALTER TABLE tool_call_attempts DROP COLUMN IF EXISTS agent_id;") + op.execute("ALTER TABLE tool_call_attempts DROP COLUMN IF EXISTS user_id;") diff --git a/application/api/user/analytics/routes.py b/application/api/user/analytics/routes.py index a297491a..5ff8eb86 100644 --- a/application/api/user/analytics/routes.py +++ b/application/api/user/analytics/routes.py @@ -14,7 +14,6 @@ from application.api.user.base import ( ) from application.storage.db.repositories.agents import AgentsRepository from application.storage.db.repositories.token_usage import TokenUsageRepository -from application.storage.db.repositories.user_logs import UserLogsRepository from application.storage.db.session import db_readonly @@ -68,16 +67,23 @@ def _intervals_for_filter(filter_option, start_date, end_date): return generate_date_range(start_date, end_date) -def _resolve_api_key(conn, api_key_id, user_id): - """Look up the ``agents.key`` value for a given agent id. +def _resolve_agent(conn, api_key_id, user_id): + """Owner-scoped agent lookup for analytics filters. - Scoped by ``user_id`` so an authenticated caller can't probe another - user's agents. Accepts either UUID or legacy Mongo ObjectId shape. + 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. """ - if not api_key_id: - return None - agent = AgentsRepository(conn).get_any(api_key_id, user_id) - return (agent or {}).get("key") if agent else None + agent = ( + AgentsRepository(conn).get_any(api_key_id, user_id) + if api_key_id + else None + ) + api_key = (agent or {}).get("key") or "" + agent_pg_id = str(agent["id"]) if agent else None + return agent, api_key, agent_pg_id @analytics_ns.route("/get_message_analytics") @@ -115,12 +121,16 @@ class GetMessageAnalytics(Resource): try: with db_readonly() as conn: - api_key = _resolve_api_key(conn, api_key_id, user) + _agent, api_key, agent_pg_id = _resolve_agent( + conn, api_key_id, user + ) # Count messages per bucket, filtered by the conversation's - # owner (user_id) and optionally the agent api_key. The - # ``user_id`` filter is always applied post-cutover to - # prevent cross-tenant leakage on admin dashboards. + # owner (user_id) and optionally the agent. The ``user_id`` + # filter is always applied post-cutover to prevent + # cross-tenant leakage on admin dashboards. Agent matching + # covers both shapes: external traffic stamps ``api_key``, + # owner chats stamp ``agent_id``. clauses = [ "c.user_id = :user_id", "m.timestamp >= :start", @@ -132,9 +142,13 @@ class GetMessageAnalytics(Resource): "end": end_date, "fmt": pg_fmt, } - if api_key: - clauses.append("c.api_key = :api_key") + if api_key_id: + clauses.append( + "(c.api_key = :api_key" + " OR c.agent_id = CAST(:agent_pg_id AS uuid))" + ) params["api_key"] = api_key + params["agent_pg_id"] = agent_pg_id where = " AND ".join(clauses) sql = ( "SELECT to_char(m.timestamp AT TIME ZONE 'UTC', :fmt) AS bucket, " @@ -172,6 +186,20 @@ class GetTokenAnalytics(Resource): default="last_30_days", enum=list(_FILTER_BUCKETS.keys()), ), + "group_by": fields.String( + required=False, + description="Second grouping dimension for the series", + default="none", + enum=["none", "model", "agent", "source"], + ), + "include_side_channel": fields.Boolean( + required=False, + description=( + "Include non-user-initiated token usage (title " + "generation, compression, RAG condensing, fallback)" + ), + default=True, + ), }, ) @@ -185,9 +213,11 @@ class GetTokenAnalytics(Resource): data = request.get_json() or {} 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)) window = _range_for_filter(filter_option) - if window is None: + if window is None or group_by not in ("none", "model", "agent", "source"): return make_response( jsonify({"success": False, "message": "Invalid option"}), 400 ) @@ -195,30 +225,67 @@ class GetTokenAnalytics(Resource): try: with db_readonly() as conn: - api_key = _resolve_api_key(conn, api_key_id, user) - # ``bucketed_totals`` applies user_id / api_key filters - # directly — no need to reshape a Mongo pipeline. + _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=user, - api_key=api_key, + 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, ) intervals = _intervals_for_filter(filter_option, start_date, end_date) daily_token_usage = {interval: 0 for interval in intervals} - for entry in rows: - daily_token_usage[entry["bucket"]] = int( - entry["prompt_tokens"] + entry["generated_tokens"] - ) + # ``series`` is the multi-dataset shape the dashboard renders + # as stacked bars: {series_key: {bucket: tokens}}. Without + # grouping the two series are the prompt/generated split. + series: dict = {} + if group_by == "none": + series = { + "prompt": {interval: 0 for interval in intervals}, + "generated": {interval: 0 for interval in intervals}, + } + for entry in rows: + bucket = entry["bucket"] + daily_token_usage[bucket] = int( + entry["prompt_tokens"] + entry["generated_tokens"] + ) + series["prompt"][bucket] = int(entry["prompt_tokens"]) + series["generated"][bucket] = int(entry["generated_tokens"]) + else: + for entry in rows: + bucket = entry["bucket"] + total = int(entry["prompt_tokens"] + entry["generated_tokens"]) + daily_token_usage[bucket] = ( + daily_token_usage.get(bucket, 0) + total + ) + key = entry["group_key"] + if key not in series: + series[key] = {interval: 0 for interval in intervals} + series[key][bucket] = series[key].get(bucket, 0) + total except Exception as err: current_app.logger.error( f"Error getting token analytics: {err}", exc_info=True ) return make_response(jsonify({"success": False}), 400) return make_response( - jsonify({"success": True, "token_usage": daily_token_usage}), 200 + jsonify( + { + "success": True, + "token_usage": daily_token_usage, + "group_by": group_by, + "series": series, + } + ), + 200, ) @@ -257,7 +324,9 @@ class GetFeedbackAnalytics(Resource): try: with db_readonly() as conn: - api_key = _resolve_api_key(conn, api_key_id, user) + _agent, api_key, agent_pg_id = _resolve_agent( + conn, api_key_id, user + ) # Feedback lives inside the ``conversation_messages.feedback`` # JSONB as ``{"text": "like"|"dislike", "timestamp": "..."}``. @@ -276,9 +345,13 @@ class GetFeedbackAnalytics(Resource): "end": end_date, "fmt": pg_fmt, } - if api_key: - clauses.append("c.api_key = :api_key") + if api_key_id: + clauses.append( + "(c.api_key = :api_key" + " OR c.agent_id = CAST(:agent_pg_id AS uuid))" + ) params["api_key"] = api_key + params["agent_pg_id"] = agent_pg_id where = " AND ".join(clauses) sql = ( "SELECT to_char(" @@ -313,6 +386,196 @@ class GetFeedbackAnalytics(Resource): ) +@analytics_ns.route("/get_tool_analytics") +class GetToolAnalytics(Resource): + get_tool_analytics_model = api.model( + "GetToolAnalyticsModel", + { + "api_key_id": fields.String(required=False, description="API Key ID"), + "filter_option": fields.String( + required=False, + description="Filter option for analytics", + default="last_30_days", + enum=list(_FILTER_BUCKETS.keys()), + ), + }, + ) + + @api.expect(get_tool_analytics_model) + @api.doc(description="Get tool call analytics from the tool execution journal") + def post(self): + decoded_token = request.decoded_token + if not decoded_token: + return make_response(jsonify({"success": False}), 401) + user = decoded_token.get("sub") + data = request.get_json() or {} + api_key_id = data.get("api_key_id") + filter_option = data.get("filter_option", "last_30_days") + + window = _range_for_filter(filter_option) + if window is None: + return make_response( + jsonify({"success": False, "message": "Invalid option"}), 400 + ) + start_date, end_date, _bucket_unit, _pg_fmt = window + + try: + with db_readonly() as conn: + _agent, api_key, agent_pg_id = _resolve_agent( + conn, api_key_id, user + ) + + # 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. + clauses = [ + "COALESCE(t.user_id, m.user_id) = :user_id", + "t.attempted_at >= :start", + "t.attempted_at <= :end", + ] + params: dict = { + "user_id": user, + "start": start_date, + "end": end_date, + } + join = ( + "LEFT JOIN conversation_messages m ON m.id = t.message_id " + "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). + clauses.append( + "(t.agent_id = CAST(:agent_pg_id AS uuid)" + " OR c.api_key = :api_key" + " OR c.agent_id = CAST(:agent_pg_id AS uuid))" + ) + params["agent_pg_id"] = agent_pg_id + params["api_key"] = api_key + where = " AND ".join(clauses) + sql = ( + "SELECT t.tool_name, " + "COUNT(*) AS calls, " + "COUNT(*) FILTER (WHERE t.status = 'failed') AS failures " + "FROM tool_call_attempts t " + f"{join}" + f"WHERE {where} " + "GROUP BY t.tool_name " + "ORDER BY calls DESC" + ) + rows = conn.execute(_sql_text(sql), params).fetchall() + + tools = [ + { + "tool_name": row._mapping["tool_name"], + "calls": int(row._mapping["calls"]), + "failures": int(row._mapping["failures"]), + } + for row in rows + ] + except Exception as err: + current_app.logger.error( + f"Error getting tool analytics: {err}", exc_info=True + ) + return make_response(jsonify({"success": False}), 400) + return make_response(jsonify({"success": True, "tools": tools}), 200) + + +@analytics_ns.route("/get_schedule_analytics") +class GetScheduleAnalytics(Resource): + get_schedule_analytics_model = api.model( + "GetScheduleAnalyticsModel", + { + "api_key_id": fields.String(required=False, description="API Key ID"), + "filter_option": fields.String( + required=False, + description="Filter option for analytics", + default="last_30_days", + enum=list(_FILTER_BUCKETS.keys()), + ), + }, + ) + + @api.expect(get_schedule_analytics_model) + @api.doc(description="Get scheduled agent run outcomes over time") + def post(self): + decoded_token = request.decoded_token + if not decoded_token: + return make_response(jsonify({"success": False}), 401) + user = decoded_token.get("sub") + data = request.get_json() or {} + api_key_id = data.get("api_key_id") + filter_option = data.get("filter_option", "last_30_days") + + window = _range_for_filter(filter_option) + if window is None: + return make_response( + jsonify({"success": False, "message": "Invalid option"}), 400 + ) + start_date, end_date, _bucket_unit, pg_fmt = window + + try: + with db_readonly() as conn: + _agent, _api_key, agent_pg_id = _resolve_agent( + conn, api_key_id, user + ) + + # A run's effective time is when it finished (fell back to + # started/scheduled for runs that never got that far). + ts = "COALESCE(r.finished_at, r.started_at, r.scheduled_for)" + clauses = [ + "r.user_id = :user_id", + f"{ts} >= :start", + f"{ts} <= :end", + ] + params: dict = { + "user_id": user, + "start": start_date, + "end": end_date, + "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) + # The worker writes ``success`` / ``failed`` / ``timeout`` / + # ``skipped`` (scheduler_worker.py); ``completed`` is kept in + # the success bucket defensively. Timeouts count as failures. + sql = ( + f"SELECT to_char({ts} AT TIME ZONE 'UTC', :fmt) AS bucket, " + "COUNT(*) FILTER (WHERE r.status IN ('success', 'completed')) AS completed, " + "COUNT(*) FILTER (WHERE r.status IN ('failed', 'timeout')) AS failed, " + "COUNT(*) FILTER (WHERE r.status = 'skipped') AS skipped " + "FROM schedule_runs r " + f"WHERE {where} " + "GROUP BY bucket ORDER BY bucket ASC" + ) + rows = conn.execute(_sql_text(sql), params).fetchall() + + intervals = _intervals_for_filter(filter_option, start_date, end_date) + runs = { + interval: {"completed": 0, "failed": 0, "skipped": 0} + for interval in intervals + } + for row in rows: + runs[row._mapping["bucket"]] = { + "completed": int(row._mapping["completed"] or 0), + "failed": int(row._mapping["failed"] or 0), + "skipped": int(row._mapping["skipped"] or 0), + } + except Exception as err: + current_app.logger.error( + f"Error getting schedule analytics: {err}", exc_info=True + ) + return make_response(jsonify({"success": False}), 400) + return make_response(jsonify({"success": True, "runs": runs}), 200) + + @analytics_ns.route("/get_user_logs") class GetUserLogs(Resource): get_user_logs_model = api.model( @@ -329,11 +592,30 @@ class GetUserLogs(Resource): description="Number of logs per page", default=10, ), + "level": fields.String( + required=False, + description="Filter by log level", + enum=["info", "error", "warning"], + ), + "event_type": fields.String( + required=False, + description="Filter by event source", + enum=["chat", "schedule", "webhook", "workflow", "system"], + ), + "search": fields.String( + required=False, description="Substring filter on the summary" + ), }, ) @api.expect(get_user_logs_model) - @api.doc(description="Get user logs with pagination") + @api.doc( + description=( + "Get user activity logs with pagination. Merges chat answers " + "(user_logs), request errors (stack_logs) and scheduled agent " + "runs (schedule_runs) into one timeline." + ) + ) def post(self): decoded_token = request.decoded_token if not decoded_token: @@ -343,80 +625,271 @@ class GetUserLogs(Resource): page = int(data.get("page", 1)) 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") + if level not in (None, "info", "error", "warning") or event_type not in ( + None, + "chat", + "schedule", + "webhook", + "workflow", + "system", + ): + return make_response( + jsonify({"success": False, "message": "Invalid option"}), 400 + ) try: with db_readonly() as conn: - api_key = _resolve_api_key(conn, api_key_id, user) - logs_repo = UserLogsRepository(conn) - if api_key: - # ``find_by_api_key`` filters on ``data->>'api_key'`` - # — the PG shape of the legacy top-level ``api_key`` - # filter. Paginate client-side using offset/limit. - all_rows = logs_repo.find_by_api_key(api_key) - offset = (page - 1) * page_size - window = all_rows[offset: offset + page_size + 1] - items = window - else: - items, has_more_flag = logs_repo.list_paginated( - user_id=user, - page=page, - page_size=page_size, + agent, api_key, agent_pg_id = _resolve_agent( + conn, api_key_id, user + ) + params: dict = { + "user_id": user, + "limit": page_size + 1, + "offset": (page - 1) * page_size, + } + + # ``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"] + 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). + params["api_key"] = api_key + params["agent_pg_id"] = agent_pg_id + params["agent_workflow_id"] = ( + str(agent["workflow_id"]) + if agent and agent.get("workflow_id") + else None ) - # list_paginated already trims to page_size and - # returns has_more separately. - results = [ - { - "id": str(item.get("id") or item.get("_id")), - "action": (item.get("data") or {}).get("action"), - "level": (item.get("data") or {}).get("level"), - "user": item.get("user_id"), - "question": (item.get("data") or {}).get("question"), - "sources": (item.get("data") or {}).get("sources"), - "retriever_params": (item.get("data") or {}).get( - "retriever_params" - ), - "timestamp": ( - item["timestamp"].isoformat() - if hasattr(item.get("timestamp"), "isoformat") - else item.get("timestamp") - ), - } - for item in items + chat_where = ["l.data->>'api_key' = :api_key"] + webhook_where = [ + "COALESCE(s.endpoint, '') = 'webhook'", + "s.api_key = :api_key", ] - return make_response( - jsonify( - { - "success": True, - "logs": results, - "page": page, - "page_size": page_size, - "has_more": has_more_flag, - } - ), - 200, + 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)" ) - has_more = len(items) > page_size - items = items[:page_size] - results = [ - { - "id": str(item.get("id") or item.get("_id")), - "action": (item.get("data") or {}).get("action"), - "level": (item.get("data") or {}).get("level"), - "user": item.get("user_id"), - "question": (item.get("data") or {}).get("question"), - "sources": (item.get("data") or {}).get("sources"), - "retriever_params": (item.get("data") or {}).get( - "retriever_params" - ), + # One normalized timeline over the three event sources. + # ``payload`` carries the per-type detail; the outer query + # paginates the merged, time-ordered result. + sql = f""" + SELECT * FROM ( + 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, + COALESCE(l.data->>'action', 'stream_answer') AS action, + l.data->>'question' 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, + jsonb_build_object( + 'endpoint', s.endpoint, + 'stacks', s.stacks + ) + 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, + jsonb_build_object( + 'endpoint', s.endpoint, + 'stacks', s.stacks + ) + 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'), + jsonb_build_object( + 'status', wr.status, + 'workflow_name', w.name, + 'result', wr.result, + 'steps', wr.steps, + 'started_at', wr.started_at, + 'finished_at', wr.ended_at + ) + 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'), + jsonb_build_object( + 'status', r.status, + 'trigger_source', r.trigger_source, + 'schedule_name', sc.name, + 'instruction', sc.instruction, + 'output', r.output, + 'error', r.error, + 'error_type', r.error_type, + 'prompt_tokens', r.prompt_tokens, + 'generated_tokens', r.generated_tokens, + 'conversation_id', r.conversation_id, + 'scheduled_for', r.scheduled_for, + 'started_at', r.started_at, + 'finished_at', r.finished_at + ) + FROM schedule_runs r + LEFT JOIN schedules sc ON sc.id = r.schedule_id + WHERE {' AND '.join(schedule_where)} + ) ev + """ + outer = [] + 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" + + rows = conn.execute(_sql_text(sql), params).fetchall() + + has_more = len(rows) > page_size + results = [] + for row in rows[:page_size]: + m = row._mapping + payload = m["payload"] or {} + item = { + # Prefix with the source so ids stay unique across the + # merged tables (each has its own id sequence). + "id": f"{m['event_type']}-{m['id']}", + "event_type": m["event_type"], + "action": m["action"], + "level": m["level"], + "user": m["user_id"], + "question": m["summary"], "timestamp": ( - item["timestamp"].isoformat() - if hasattr(item.get("timestamp"), "isoformat") - else item.get("timestamp") + m["timestamp"].isoformat() + if hasattr(m["timestamp"], "isoformat") + else m["timestamp"] ), } - for item in items - ] + if m["event_type"] == "chat": + item.update( + { + "response": payload.get("response"), + "sources": payload.get("sources"), + "tool_calls": payload.get("tool_calls"), + "agent_id": payload.get("agent_id"), + "attachments": payload.get("attachments"), + } + ) + elif m["event_type"] in ("system", "webhook"): + item.update( + { + "endpoint": payload.get("endpoint"), + "stacks": payload.get("stacks"), + } + ) + elif m["event_type"] == "workflow": + item.update( + { + "status": payload.get("status"), + "workflow_name": payload.get("workflow_name"), + "result": payload.get("result"), + "steps": payload.get("steps"), + "started_at": payload.get("started_at"), + "finished_at": payload.get("finished_at"), + } + ) + else: # schedule + item.update( + { + "status": payload.get("status"), + "trigger_source": payload.get("trigger_source"), + "schedule_name": payload.get("schedule_name"), + "instruction": payload.get("instruction"), + "output": payload.get("output"), + "error": payload.get("error"), + "error_type": payload.get("error_type"), + "prompt_tokens": payload.get("prompt_tokens"), + "generated_tokens": payload.get("generated_tokens"), + "conversation_id": payload.get("conversation_id"), + "scheduled_for": payload.get("scheduled_for"), + "started_at": payload.get("started_at"), + "finished_at": payload.get("finished_at"), + } + ) + results.append(item) except Exception as err: current_app.logger.error( f"Error getting user logs: {err}", exc_info=True diff --git a/application/llm/handlers/base.py b/application/llm/handlers/base.py index 79ba8aab..c5963c54 100644 --- a/application/llm/handlers/base.py +++ b/application/llm/handlers/base.py @@ -956,6 +956,9 @@ class LLMHandler(ABC): pause_info["action_name"], pause_info.get("arguments") or {}, tool_id=pause_info.get("tool_id"), + message_id=agent.tool_executor.message_id, + user_id=agent.tool_executor.user, + agent_id=agent.tool_executor.agent_id, ) _mark_failed( pause_info["call_id"], diff --git a/application/storage/db/models.py b/application/storage/db/models.py index a8189d83..2099c2ec 100644 --- a/application/storage/db/models.py +++ b/application/storage/db/models.py @@ -508,6 +508,11 @@ tool_call_attempts_table = Table( ForeignKey("conversation_messages.id", ondelete="SET NULL"), ), Column("tool_id", UUID(as_uuid=True)), + # Direct attribution (0018): headless runs (scheduled / webhook) and + # parse-failure rows never get a message_id, so user/agent are stamped + # at propose time instead of derived through the parent message. + Column("user_id", Text), + Column("agent_id", UUID(as_uuid=True)), Column("tool_name", Text, nullable=False), Column("action_name", Text, nullable=False), Column("arguments", JSONB, nullable=False), diff --git a/application/storage/db/repositories/token_usage.py b/application/storage/db/repositories/token_usage.py index 033e51a3..127d213e 100644 --- a/application/storage/db/repositories/token_usage.py +++ b/application/storage/db/repositories/token_usage.py @@ -107,6 +107,29 @@ class TokenUsageRepository: ) return result.scalar() + # Token usage written outside a user-initiated request (conversation + # title generation, history compression, RAG question condensing, + # provider fallback). Mirrors the exclusion list in ``count_in_range``. + SIDE_CHANNEL_SOURCES = ("title", "compression", "rag_condense", "fallback") + + # Run-level roll-ups that duplicate per-call rows. The scheduler worker + # inserts one ``source='schedule'`` row summing a run's tokens, but the + # run's individual LLM calls were already persisted as ``agent_stream`` + # rows by the usage decorators — counting both doubles scheduled spend. + # The rollup is never used as a fallback: if a per-call insert failed + # (logged in usage.py), that call's tokens go uncounted, the same loss + # mode as any other traffic whose insert fails. + ROLLUP_SOURCES = ("schedule",) + + # Allowed ``group_by`` values → the SQL expression producing the + # group key. ``agent`` resolves to the agent's display name so the + # dashboard never has to map UUIDs client-side. + _GROUP_KEY_EXPRS = { + "model": "COALESCE(tu.model_id, 'unknown')", + "agent": "COALESCE(a.name, 'No agent')", + "source": "COALESCE(tu.source, 'agent_stream')", + } + def bucketed_totals( self, *, @@ -116,6 +139,8 @@ class TokenUsageRepository: agent_id: Optional[str] = None, timestamp_gte: Optional[datetime] = None, timestamp_lt: Optional[datetime] = None, + group_by: Optional[str] = None, + include_side_channel: bool = True, ) -> list[dict]: """Sum ``prompt_tokens`` / ``generated_tokens`` bucketed by time. @@ -124,6 +149,11 @@ class TokenUsageRepository: mirrors Mongo's output so the route layer doesn't reshape: ``"YYYY-MM-DD HH:MM:00"`` (minute), ``"YYYY-MM-DD HH:00"`` (hour), ``"YYYY-MM-DD"`` (day). Rows are ordered by bucket ASC. + + ``group_by`` (``"model"`` / ``"agent"`` / ``"source"``) adds a + second grouping dimension; each returned row then carries a + ``group_key``. ``include_side_channel=False`` drops rows whose + ``source`` is a side-channel call (title generation etc.). """ formats = { "minute": "YYYY-MM-DD HH24:MI:00", @@ -132,35 +162,65 @@ class TokenUsageRepository: } if bucket_unit not in formats: raise ValueError(f"unsupported bucket_unit: {bucket_unit!r}") + if group_by is not None and group_by not in self._GROUP_KEY_EXPRS: + raise ValueError(f"unsupported group_by: {group_by!r}") fmt = formats[bucket_unit] clauses: list[str] = [] params: dict = {"fmt": fmt} if user_id is not None: - clauses.append("user_id = :user_id") + clauses.append("tu.user_id = :user_id") params["user_id"] = user_id + # Rows stamp ``api_key`` (external traffic) or ``agent_id`` + # (owner chats / headless runs), so a per-agent filter must + # match either shape. + agent_clauses: list[str] = [] if api_key is not None: - clauses.append("api_key = :api_key") + agent_clauses.append("tu.api_key = :api_key") params["api_key"] = api_key if agent_id is not None: - clauses.append("agent_id = CAST(:agent_id AS uuid)") + agent_clauses.append("tu.agent_id = CAST(:agent_id AS uuid)") params["agent_id"] = agent_id + if agent_clauses: + clauses.append(f"({' OR '.join(agent_clauses)})") if timestamp_gte is not None: - clauses.append("timestamp >= :timestamp_gte") + clauses.append("tu.timestamp >= :timestamp_gte") params["timestamp_gte"] = timestamp_gte if timestamp_lt is not None: - clauses.append("timestamp < :timestamp_lt") + clauses.append("tu.timestamp < :timestamp_lt") params["timestamp_lt"] = timestamp_lt + excluded_sources = list(self.ROLLUP_SOURCES) + if not include_side_channel: + excluded_sources.extend(self.SIDE_CHANNEL_SOURCES) + placeholders = [] + for i, src in enumerate(excluded_sources): + key = f"excl_src_{i}" + placeholders.append(f":{key}") + params[key] = src + clauses.append( + f"COALESCE(tu.source, 'agent_stream') NOT IN ({', '.join(placeholders)})" + ) where = ("WHERE " + " AND ".join(clauses)) if clauses else "" + + group_select = "" + group_clause = "" + join = "" + if group_by is not None: + group_select = f", {self._GROUP_KEY_EXPRS[group_by]} AS group_key" + group_clause = ", group_key" + if group_by == "agent": + join = "LEFT JOIN agents a ON a.id = tu.agent_id" result = self._conn.execute( text( f""" - SELECT to_char(timestamp AT TIME ZONE 'UTC', :fmt) AS bucket, - COALESCE(SUM(prompt_tokens), 0) AS prompt_tokens, - COALESCE(SUM(generated_tokens), 0) AS generated_tokens - FROM token_usage + SELECT to_char(tu.timestamp AT TIME ZONE 'UTC', :fmt) AS bucket, + COALESCE(SUM(tu.prompt_tokens), 0) AS prompt_tokens, + COALESCE(SUM(tu.generated_tokens), 0) AS generated_tokens + {group_select} + FROM token_usage tu + {join} {where} - GROUP BY bucket + GROUP BY bucket{group_clause} ORDER BY bucket ASC """ ), @@ -171,6 +231,11 @@ class TokenUsageRepository: "bucket": row._mapping["bucket"], "prompt_tokens": int(row._mapping["prompt_tokens"]), "generated_tokens": int(row._mapping["generated_tokens"]), + **( + {"group_key": row._mapping["group_key"]} + if group_by is not None + else {} + ), } for row in result.fetchall() ] diff --git a/application/storage/db/repositories/tool_call_attempts.py b/application/storage/db/repositories/tool_call_attempts.py index 9b72c3ab..a6fa6c74 100644 --- a/application/storage/db/repositories/tool_call_attempts.py +++ b/application/storage/db/repositories/tool_call_attempts.py @@ -22,6 +22,9 @@ class ToolCallAttemptsRepository: arguments: Any, *, tool_id: Optional[str] = None, + message_id: Optional[str] = None, + user_id: Optional[str] = None, + agent_id: Optional[str] = None, ) -> bool: """Insert a ``proposed`` row before the tool executes. @@ -29,15 +32,22 @@ class ToolCallAttemptsRepository: guards against the LLM emitting a duplicate ``call_id``: the existing row stays put rather than a re-insert raising ``IntegrityError``. + + ``message_id`` / ``user_id`` / ``agent_id`` attribute the attempt + from the start — headless runs and parse-failure rows never reach + ``mark_executed``, and headless runs have no message at all. """ result = self._conn.execute( text( """ INSERT INTO tool_call_attempts - (call_id, tool_id, tool_name, action_name, arguments, status) + (call_id, tool_id, tool_name, action_name, arguments, + message_id, user_id, agent_id, status) VALUES (:call_id, CAST(:tool_id AS uuid), :tool_name, - :action_name, CAST(:arguments AS jsonb), 'proposed') + :action_name, CAST(:arguments AS jsonb), + CAST(:message_id AS uuid), :user_id, + CAST(:agent_id AS uuid), 'proposed') ON CONFLICT (call_id) DO NOTHING """ ), @@ -47,6 +57,9 @@ class ToolCallAttemptsRepository: "tool_name": tool_name, "action_name": action_name, "arguments": json.dumps(arguments if arguments is not None else {}, cls=PGNativeJSONEncoder), + "message_id": message_id, + "user_id": user_id, + "agent_id": agent_id, }, ) return result.rowcount > 0 diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 3ab57f4f..f0b32844 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -4640,7 +4640,6 @@ "version": "19.2.14", "resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.14.tgz", "integrity": "sha512-ilcTH/UniCkMdtexkoCN0bI7pMcJDvmQFPvuPvmEaYA/NSfFTAgdUSLAoVjaRJm7+6PvcM+q1zYOwS4wTYMF9w==", - "dev": true, "license": "MIT", "dependencies": { "csstype": "^3.2.2" @@ -4650,7 +4649,7 @@ "version": "19.2.3", "resolved": "https://registry.npmjs.org/@types/react-dom/-/react-dom-19.2.3.tgz", "integrity": "sha512-jp2L/eY6fn+KgVVQAOqYItbF0VY/YApe5Mz2F0aykSO8gx31bYCZyvSeYxCHKvzHG5eZjc+zyaS5BrBWya2+kQ==", - "dev": true, + "devOptional": true, "license": "MIT", "peerDependencies": { "@types/react": "^19.2.0" @@ -5961,7 +5960,6 @@ "version": "3.2.3", "resolved": "https://registry.npmjs.org/csstype/-/csstype-3.2.3.tgz", "integrity": "sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ==", - "dev": true, "license": "MIT" }, "node_modules/cytoscape": { @@ -12958,7 +12956,7 @@ "version": "6.0.3", "resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz", "integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==", - "dev": true, + "devOptional": true, "license": "Apache-2.0", "bin": { "tsc": "bin/tsc", diff --git a/frontend/src/api/endpoints.ts b/frontend/src/api/endpoints.ts index a897a70a..f7755bda 100644 --- a/frontend/src/api/endpoints.ts +++ b/frontend/src/api/endpoints.ts @@ -35,6 +35,8 @@ const endpoints = { MESSAGE_ANALYTICS: '/api/get_message_analytics', TOKEN_ANALYTICS: '/api/get_token_analytics', FEEDBACK_ANALYTICS: '/api/get_feedback_analytics', + TOOL_ANALYTICS: '/api/get_tool_analytics', + SCHEDULE_ANALYTICS: '/api/get_schedule_analytics', LOGS: `/api/get_user_logs`, MANAGE_SYNC: '/api/manage_sync', SYNC_SOURCE: '/api/sync_source', diff --git a/frontend/src/api/services/userService.ts b/frontend/src/api/services/userService.ts index 0ffac33e..03fb8575 100644 --- a/frontend/src/api/services/userService.ts +++ b/frontend/src/api/services/userService.ts @@ -73,6 +73,10 @@ const userService = { apiClient.post(endpoints.USER.TOKEN_ANALYTICS, data, token), getFeedbackAnalytics: (data: any, token: string | null): Promise => apiClient.post(endpoints.USER.FEEDBACK_ANALYTICS, data, token), + getToolAnalytics: (data: any, token: string | null): Promise => + apiClient.post(endpoints.USER.TOOL_ANALYTICS, data, token), + getScheduleAnalytics: (data: any, token: string | null): Promise => + apiClient.post(endpoints.USER.SCHEDULE_ANALYTICS, data, token), getLogs: (data: any, token: string | null): Promise => apiClient.post(endpoints.USER.LOGS, data, token), manageSync: (data: any, token: string | null): Promise => diff --git a/frontend/src/locale/en.json b/frontend/src/locale/en.json index 8d91f741..9fb8ff51 100644 --- a/frontend/src/locale/en.json +++ b/frontend/src/locale/en.json @@ -147,15 +147,74 @@ "filterPlaceholder": "Filter", "none": "None", "positiveFeedback": "Positive Feedback", - "negativeFeedback": "Negative Feedback" + "negativeFeedback": "Negative Feedback", + "groupBy": "Group by", + "groupOptions": { + "total": "Total", + "model": "By model", + "agent": "By agent", + "source": "By source" + }, + "includeSideChannel": "Background tokens", + "promptTokens": "Prompt Tokens", + "generatedTokens": "Generated Tokens", + "scheduledRuns": "Scheduled Runs", + "toolUsage": "Tool Usage", + "completed": "Completed", + "failed": "Failed", + "skipped": "Skipped", + "successful": "Successful", + "stats": { + "messages": "Messages", + "tokens": "Tokens", + "toolCalls": "Tool Calls", + "runSuccess": "Run Success", + "feedback": "Feedback" + } }, "logs": { "label": "Logs", - "subtitle": "View and monitor API calls and conversation logs", + "subtitle": "View and monitor API calls, scheduled runs and errors across your account", "filterByChatbot": "Filter by chatbot", "selectChatbot": "Select chatbot", "none": "None", - "tableHeader": "API generated / chatbot conversations" + "tableHeader": "API generated / chatbot conversations", + "searchPlaceholder": "Search logs...", + "noLogs": "No logs found", + "levels": { + "all": "All levels", + "info": "Info", + "error": "Error", + "warning": "Warning" + }, + "types": { + "all": "All events", + "chat": "Chat", + "schedule": "Scheduled", + "webhook": "Webhook", + "workflow": "Workflow", + "system": "System" + }, + "detail": { + "agent": "Agent", + "status": "Status", + "trigger": "Trigger", + "errorType": "Error type", + "tokens": "Tokens", + "duration": "Duration", + "conversation": "Conversation", + "endpoint": "Endpoint", + "instruction": "Instruction", + "response": "Response", + "output": "Output", + "error": "Error", + "toolCalls": "Tool calls", + "sources": "Sources", + "workflow": "Workflow", + "activity": "Activity", + "steps": "Steps", + "result": "Result" + } }, "tools": { "label": "Tools", diff --git a/frontend/src/settings/Analytics.tsx b/frontend/src/settings/Analytics.tsx index db1d4ac4..0cc65f80 100644 --- a/frontend/src/settings/Analytics.tsx +++ b/frontend/src/settings/Analytics.tsx @@ -21,6 +21,7 @@ import { SelectTrigger, SelectValue, } from '../components/ui/select'; +import { Switch } from '../components/ui/switch'; import { useDarkTheme, useLoaderState } from '../hooks'; import { selectToken } from '../preferences/preferenceSlice'; import { htmlLegendPlugin } from '../utils/chartUtils'; @@ -51,6 +52,20 @@ ChartJS.register( Legend, ); +// Extra series colors for grouped/stacked charts. The first dataset always +// uses the resolved `--primary` color; these follow for datasets 2..n. +const SERIES_COLORS = [ + '#FF6384', + '#36A2EB', + '#FFCE56', + '#4BC0C0', + '#9966FF', + '#FF9F40', + '#2BC596', +]; + +type TokenGroupBy = 'none' | 'model' | 'agent' | 'source'; + type AnalyticsProps = { agentId?: string; }; @@ -79,6 +94,21 @@ export default function Analytics({ agentId }: AnalyticsProps) { }, ]; + const groupByOptions: { label: string; value: TokenGroupBy }[] = [ + { label: t('settings.analytics.groupOptions.total'), value: 'none' }, + { label: t('settings.analytics.groupOptions.model'), value: 'model' }, + // Grouping by agent is meaningless on a single agent's page. + ...(agentId + ? [] + : [ + { + label: t('settings.analytics.groupOptions.agent'), + value: 'agent' as TokenGroupBy, + }, + ]), + { label: t('settings.analytics.groupOptions.source'), value: 'source' }, + ]; + const [messagesData, setMessagesData] = useState | null>(null); + const [tokenSeries, setTokenSeries] = useState + > | null>(null); const [feedbackData, setFeedbackData] = useState | null>(null); - const [messagesFilter, setMessagesFilter] = useState<{ - label: string; - value: string; - }>({ - label: t('settings.analytics.filterOptions.last30Days'), - value: 'last_30_days', - }); - const [tokenUsageFilter, setTokenUsageFilter] = useState<{ - label: string; - value: string; - }>({ - label: t('settings.analytics.filterOptions.last30Days'), - value: 'last_30_days', - }); - const [feedbackFilter, setFeedbackFilter] = useState<{ + const [toolsData, setToolsData] = useState< + { tool_name: string; calls: number; failures: number }[] | null + >(null); + const [scheduleData, setScheduleData] = useState | null>(null); + + const [timeFilter, setTimeFilter] = useState<{ label: string; value: string; }>({ label: t('settings.analytics.filterOptions.last30Days'), value: 'last_30_days', }); + const [tokenGroupBy, setTokenGroupBy] = useState('none'); + // Side-channel = tokens spent outside a user request (title generation, + // history compression, RAG condensing). + const [includeSideChannel, setIncludeSideChannel] = useState(false); const [loadingMessages, setLoadingMessages] = useLoaderState(true); const [loadingTokens, setLoadingTokens] = useLoaderState(true); const [loadingFeedback, setLoadingFeedback] = useLoaderState(true); + const [loadingTools, setLoadingTools] = useLoaderState(true); + const [loadingSchedules, setLoadingSchedules] = useLoaderState(true); const [isDarkTheme] = useDarkTheme(); const primaryColor = useMemo( () => readCssVar('--primary', '#7d54d1'), @@ -123,6 +157,10 @@ export default function Analytics({ agentId }: AnalyticsProps) { // resolved value of `--primary`; re-read whenever it flips. [isDarkTheme], ); + const seriesColor = (index: number) => + index === 0 + ? primaryColor + : SERIES_COLORS[(index - 1) % SERIES_COLORS.length]; const fetchMessagesData = async (agent_id?: string, filter?: string) => { setLoadingMessages(true); @@ -144,19 +182,27 @@ export default function Analytics({ agentId }: AnalyticsProps) { } }; - const fetchTokenData = async (agent_id?: string, filter?: string) => { + const fetchTokenData = async ( + agent_id?: string, + filter?: string, + groupBy?: TokenGroupBy, + sideChannel?: boolean, + ) => { setLoadingTokens(true); try { const response = await userService.getTokenAnalytics( { api_key_id: agent_id, filter_option: filter, + group_by: groupBy, + include_side_channel: sideChannel, }, token, ); if (!response.ok) throw new Error('Failed to fetch analytics data'); const data = await response.json(); setTokenUsageData(data.token_usage); + setTokenSeries(data.series); } catch (error) { console.error(error); } finally { @@ -184,28 +230,174 @@ export default function Analytics({ agentId }: AnalyticsProps) { } }; - useEffect(() => { - const id = agentId; - const filter = messagesFilter; - fetchMessagesData(id, filter?.value); - }, [agentId, messagesFilter]); + const fetchToolsData = async (agent_id?: string, filter?: string) => { + setLoadingTools(true); + try { + const response = await userService.getToolAnalytics( + { + api_key_id: agent_id, + filter_option: filter, + }, + token, + ); + if (!response.ok) throw new Error('Failed to fetch analytics data'); + const data = await response.json(); + setToolsData(data.tools); + } catch (error) { + console.error(error); + } finally { + setLoadingTools(false); + } + }; + + const fetchScheduleData = async (agent_id?: string, filter?: string) => { + setLoadingSchedules(true); + try { + const response = await userService.getScheduleAnalytics( + { + api_key_id: agent_id, + filter_option: filter, + }, + token, + ); + if (!response.ok) throw new Error('Failed to fetch analytics data'); + const data = await response.json(); + setScheduleData(data.runs); + } catch (error) { + console.error(error); + } finally { + setLoadingSchedules(false); + } + }; useEffect(() => { - const id = agentId; - const filter = tokenUsageFilter; - fetchTokenData(id, filter?.value); - }, [agentId, tokenUsageFilter]); + fetchMessagesData(agentId, timeFilter.value); + fetchFeedbackData(agentId, timeFilter.value); + fetchToolsData(agentId, timeFilter.value); + fetchScheduleData(agentId, timeFilter.value); + }, [agentId, timeFilter]); useEffect(() => { - const id = agentId; - const filter = feedbackFilter; - fetchFeedbackData(id, filter?.value); - }, [agentId, feedbackFilter]); + fetchTokenData(agentId, timeFilter.value, tokenGroupBy, includeSideChannel); + }, [agentId, timeFilter, tokenGroupBy, includeSideChannel]); + + const totalMessages = Object.values(messagesData || {}).reduce( + (sum, value) => sum + value, + 0, + ); + const totalTokens = Object.values(tokenUsageData || {}).reduce( + (sum, value) => sum + value, + 0, + ); + const totalToolCalls = (toolsData || []).reduce( + (sum, tool) => sum + tool.calls, + 0, + ); + const feedbackTotals = Object.values(feedbackData || {}).reduce( + (sum, value) => ({ + positive: sum.positive + value.positive, + negative: sum.negative + value.negative, + }), + { positive: 0, negative: 0 }, + ); + const scheduleTotals = Object.values(scheduleData || {}).reduce( + (sum, value) => ({ + completed: sum.completed + value.completed, + failed: sum.failed + value.failed, + skipped: sum.skipped + value.skipped, + }), + { completed: 0, failed: 0, skipped: 0 }, + ); + const scheduleRunsTotal = scheduleTotals.completed + scheduleTotals.failed; + const scheduleSuccessRate = + scheduleRunsTotal > 0 + ? `${Math.round((scheduleTotals.completed / scheduleRunsTotal) * 100)}%` + : '—'; + + const statCards = [ + { + label: t('settings.analytics.stats.messages'), + value: totalMessages.toLocaleString(), + }, + { + label: t('settings.analytics.stats.tokens'), + value: totalTokens.toLocaleString(), + }, + { + label: t('settings.analytics.stats.toolCalls'), + value: totalToolCalls.toLocaleString(), + }, + { + label: t('settings.analytics.stats.runSuccess'), + value: scheduleSuccessRate, + }, + { + label: t('settings.analytics.stats.feedback'), + value: `+${feedbackTotals.positive} / -${feedbackTotals.negative}`, + }, + ]; + + const tokenSeriesEntries = Object.entries(tokenSeries || {}); + const tokenLabels = Object.keys( + tokenSeriesEntries[0]?.[1] || tokenUsageData || {}, + ).map((item) => formatDate(item)); + const tokenDatasets = tokenSeriesEntries.map(([key, series], index) => ({ + label: + tokenGroupBy === 'none' + ? key === 'prompt' + ? t('settings.analytics.promptTokens') + : t('settings.analytics.generatedTokens') + : key, + data: Object.values(series), + backgroundColor: seriesColor(index), + })); + return (
-

- {t('settings.analytics.subtitle')} -

+
+

+ {t('settings.analytics.subtitle')} +

+ +
+ + {/* Summary stat cards */} +
+ {statCards.map((card) => ( +
+

{card.label}

+

+ {card.value} +

+
+ ))} +
+ {/* Messages Analytics */}
@@ -213,29 +405,6 @@ export default function Analytics({ agentId }: AnalyticsProps) {

{t('settings.analytics.messages')}

-
-
+

{t('settings.analytics.tokenUsage')}

+
+

+ {t('settings.analytics.includeSideChannel')} +

+ +
+ labels: tokenLabels, + datasets: tokenDatasets, + }} + legendID="legend-container-2" + maxTicksLimitInX={8} + isStacked={true} + /> + )} +
+
+
+ + {/* Scheduled runs + tool usage */} +
+
+
+

+ {t('settings.analytics.scheduledRuns')} +

+
+
+
+ {loadingSchedules ? ( + + ) : ( + formatDate(item), ), datasets: [ { - label: t('settings.analytics.tokenUsage'), - data: Object.values(tokenUsageData || {}), + label: t('settings.analytics.completed'), + data: Object.values(scheduleData || {}).map( + (item) => item.completed, + ), backgroundColor: primaryColor, }, + { + label: t('settings.analytics.failed'), + data: Object.values(scheduleData || {}).map( + (item) => item.failed, + ), + backgroundColor: '#FF6384', + }, + { + label: t('settings.analytics.skipped'), + data: Object.values(scheduleData || {}).map( + (item) => item.skipped, + ), + backgroundColor: '#FFCE56', + }, ], }} - legendID="legend-container-2" + legendID="legend-container-4" maxTicksLimitInX={8} - isStacked={false} + isStacked={true} + /> + )} +
+
+ +
+
+

+ {t('settings.analytics.toolUsage')} +

+
+
+
+ {loadingTools ? ( + + ) : ( + tool.tool_name), + datasets: [ + { + label: t('settings.analytics.successful'), + data: (toolsData || []).map( + (tool) => tool.calls - tool.failures, + ), + backgroundColor: primaryColor, + }, + { + label: t('settings.analytics.failed'), + data: (toolsData || []).map((tool) => tool.failures), + backgroundColor: '#FF6384', + }, + ], + }} + legendID="legend-container-5" + maxTicksLimitInX={8} + isStacked={true} /> )}
@@ -327,35 +586,12 @@ export default function Analytics({ agentId }: AnalyticsProps) {
{/* Feedback Analytics */} -
+

{t('settings.analytics.userFeedback')}

-
{ + const handle = setTimeout(() => setSearch(searchInput.trim()), 400); + return () => clearTimeout(handle); + }, [searchInput]); + + useEffect(() => { + if (isFirstRender.current) { + isFirstRender.current = false; + return; + } + setLogsByPage({}); + setPage(1); + setHasMore(true); + }, [levelFilter, typeFilter, search, agentId]); + const fetchLogs = async () => { if (logsByPage[page] && logsByPage[page].length > 0) return; + const issuedKey = filterKey; setLoadingLogs(true); try { const response = await userService.getLogs( @@ -35,11 +72,15 @@ export default function Logs({ agentId, tableHeader }: LogsProps) { page: page, api_key_id: agentId, page_size: 10, + level: levelFilter === 'all' ? undefined : levelFilter, + event_type: typeFilter === 'all' ? undefined : typeFilter, + search: search || undefined, }, token, ); if (!response.ok) throw new Error('Failed to fetch logs'); const data = await response.json(); + if (issuedKey !== filterKeyRef.current) return; setLogsByPage((prev) => ({ ...prev, @@ -49,18 +90,74 @@ export default function Logs({ agentId, tableHeader }: LogsProps) { } catch (error) { console.error(error); } finally { - setLoadingLogs(false); + if (issuedKey === filterKeyRef.current) setLoadingLogs(false); } }; + // `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 (hasMore) fetchLogs(); - }, [page, agentId]); + }, [page, agentId, levelFilter, typeFilter, search, logsByPage]); + + const levelOptions = [ + { label: t('settings.logs.levels.all'), value: 'all' }, + { label: t('settings.logs.levels.info'), value: 'info' }, + { label: t('settings.logs.levels.error'), value: 'error' }, + { label: t('settings.logs.levels.warning'), value: 'warning' }, + ]; + const typeOptions = [ + { label: t('settings.logs.types.all'), value: 'all' }, + { label: t('settings.logs.types.chat'), value: 'chat' }, + { label: t('settings.logs.types.schedule'), value: 'schedule' }, + { label: t('settings.logs.types.webhook'), value: 'webhook' }, + { label: t('settings.logs.types.workflow'), value: 'workflow' }, + { label: t('settings.logs.types.system'), value: 'system' }, + ]; + return (

{t('settings.logs.subtitle')}

+
+ + + setSearchInput(e.target.value)} + placeholder={t('settings.logs.searchPlaceholder')} + className="w-56 rounded-3xl" + /> +
+ {!loading && logs.length === 0 && ( +

+ {t('settings.logs.noLogs')} +

+ )} {logs?.map((log, index) => { if (index === logs.length - 1) { return ( @@ -150,6 +252,14 @@ function LogsTable({ logs, setPage, loading, tableHeader }: LogsTableProps) {
); } + +function formatDuration(start?: string, end?: string): string | null { + if (!start || !end) return null; + const ms = new Date(end).getTime() - new Date(start).getTime(); + if (isNaN(ms) || ms < 0) return null; + return ms < 1000 ? `${ms}ms` : `${(ms / 1000).toFixed(1)}s`; +} + function Log({ log, isOpen, @@ -165,7 +275,107 @@ function Log({ error: 'text-red-500', warning: 'text-yellow-500', }; - const { id, action, timestamp, ...filteredLog } = log; + const { id, action, timestamp, event_type, ...filteredLog } = log; + + const detailRows: [string, string][] = []; + if (log.event_type === 'chat') { + if (log.agent_id) + detailRows.push([t('settings.logs.detail.agent'), log.agent_id]); + if (log.tool_calls?.length) + detailRows.push([ + t('settings.logs.detail.toolCalls'), + String(log.tool_calls.length), + ]); + if (log.sources?.length) + detailRows.push([ + t('settings.logs.detail.sources'), + String(log.sources.length), + ]); + } else if (log.event_type === 'schedule') { + if (log.status) + detailRows.push([t('settings.logs.detail.status'), log.status]); + if (log.trigger_source) + detailRows.push([t('settings.logs.detail.trigger'), log.trigger_source]); + if (log.error_type) + detailRows.push([t('settings.logs.detail.errorType'), log.error_type]); + if (log.prompt_tokens || log.generated_tokens) + detailRows.push([ + t('settings.logs.detail.tokens'), + `${(log.prompt_tokens || 0) + (log.generated_tokens || 0)}`, + ]); + const duration = formatDuration(log.started_at, log.finished_at); + if (duration) + detailRows.push([t('settings.logs.detail.duration'), duration]); + if (log.conversation_id) + detailRows.push([ + t('settings.logs.detail.conversation'), + log.conversation_id, + ]); + } else if (log.event_type === 'workflow') { + if (log.workflow_name) + detailRows.push([t('settings.logs.detail.workflow'), log.workflow_name]); + if (log.status) + detailRows.push([t('settings.logs.detail.status'), log.status]); + const duration = formatDuration(log.started_at, log.finished_at); + if (duration) + detailRows.push([t('settings.logs.detail.duration'), duration]); + } else if (log.event_type === 'system' || log.event_type === 'webhook') { + if (log.endpoint) + detailRows.push([t('settings.logs.detail.endpoint'), log.endpoint]); + } + + const textBlocks: { label: string; text: string; isError?: boolean }[] = []; + if (log.event_type === 'schedule' && log.instruction) + textBlocks.push({ + label: t('settings.logs.detail.instruction'), + text: log.instruction, + }); + if (log.response) + textBlocks.push({ + label: t('settings.logs.detail.response'), + text: log.response, + }); + if (log.output) + textBlocks.push({ + label: t('settings.logs.detail.output'), + text: log.output, + }); + if (log.error) + textBlocks.push({ + label: t('settings.logs.detail.error'), + text: log.error, + isError: true, + }); + + const jsonBlocks: { label: string; value: unknown }[] = []; + if (log.tool_calls?.length) + jsonBlocks.push({ + label: t('settings.logs.detail.toolCalls'), + value: log.tool_calls, + }); + if (log.sources?.length) + jsonBlocks.push({ + label: t('settings.logs.detail.sources'), + value: log.sources, + }); + if (log.stacks?.length) + jsonBlocks.push({ + label: + log.event_type === 'webhook' + ? t('settings.logs.detail.activity') + : t('settings.logs.detail.error'), + value: log.stacks, + }); + if (log.steps?.length) + jsonBlocks.push({ + label: t('settings.logs.detail.steps'), + value: log.steps, + }); + if (log.event_type === 'workflow' && log.result) + jsonBlocks.push({ + label: t('settings.logs.detail.result'), + value: log.result, + }); return (
@@ -180,8 +390,13 @@ function Log({ alt="Expand log entry" className={`mt-[3px] h-3 w-3 transition duration-300 ${isOpen ? 'rotate-90' : ''}`} /> - +

{`${log.timestamp}`}

+ {log.event_type && ( +

+ {t(`settings.logs.types.${log.event_type}`)} +

+ )}

{`[${log.action}]`}

{isOpen && (
-
- 
-              {JSON.stringify(filteredLog, null, 2)}
-
-
+ {detailRows.length > 0 && ( +
+ {detailRows.map(([label, value]) => ( +
+ + {label} + + + {value} + +
+ ))} +
+ )} + {textBlocks.map((block) => ( +
+

{block.label}

+ 
+                {block.text}
+
+
+ ))} + {jsonBlocks.map((block) => ( +
+

{block.label}

+
+ 
+                  {JSON.stringify(block.value, null, 2)}
+
+
+
+ ))}
[]; - retriever_params: Record; timestamp: string; + // chat events (user_logs) + response?: string; + sources?: Record[]; + tool_calls?: Record[]; + agent_id?: string; + attachments?: string[]; + // system + webhook events (stack_logs) + endpoint?: string; + stacks?: Record[]; + // workflow events (workflow_runs) + workflow_name?: string; + result?: Record; + steps?: Record[]; + // schedule events (schedule_runs) + status?: string; + trigger_source?: string; + schedule_name?: string; + instruction?: string; + output?: string; + error?: string; + error_type?: string; + prompt_tokens?: number; + generated_tokens?: number; + conversation_id?: string; + scheduled_for?: string; + started_at?: string; + finished_at?: string; }; export type ParameterGroupType = { diff --git a/tests/api/user/test_analytics.py b/tests/api/user/test_analytics.py index f69b462b..d1869371 100644 --- a/tests/api/user/test_analytics.py +++ b/tests/api/user/test_analytics.py @@ -74,33 +74,54 @@ class TestRangeForFilter: assert bucket_unit in {"minute", "hour", "day"} -class TestResolveApiKey: - def test_none_when_no_id(self, pg_conn): - from application.api.user.analytics.routes import _resolve_api_key +class TestResolveAgent: + def test_no_agent_when_no_id(self, pg_conn): + from application.api.user.analytics.routes import _resolve_agent - assert _resolve_api_key(pg_conn, None, "u") is None - assert _resolve_api_key(pg_conn, "", "u") is None + assert _resolve_agent(pg_conn, None, "u") == (None, "", None) + assert _resolve_agent(pg_conn, "", "u") == (None, "", None) - def test_returns_key_for_owned_agent(self, pg_conn): - from application.api.user.analytics.routes import _resolve_api_key + def test_returns_key_and_id_for_owned_agent(self, pg_conn): + from application.api.user.analytics.routes import _resolve_agent from application.storage.db.repositories.agents import AgentsRepository agent = AgentsRepository(pg_conn).create( "owner", "test-agent", "published", key="secret-api-key", ) - assert ( - _resolve_api_key(pg_conn, str(agent["id"]), "owner") - == "secret-api-key" + resolved, api_key, agent_pg_id = _resolve_agent( + pg_conn, str(agent["id"]), "owner" ) + assert resolved is not None + assert api_key == "secret-api-key" + assert agent_pg_id == str(agent["id"]) - def test_none_for_other_users_agent(self, pg_conn): - from application.api.user.analytics.routes import _resolve_api_key + def test_keyless_agent_gets_no_match_sentinel(self, pg_conn): + from application.api.user.analytics.routes import _resolve_agent + from application.storage.db.repositories.agents import AgentsRepository + + agent = AgentsRepository(pg_conn).create( + "owner", "test-agent", "draft", key="", + ) + 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. + assert resolved is not None + assert api_key == "" + assert agent_pg_id == str(agent["id"]) + + def test_no_match_for_other_users_agent(self, pg_conn): + from application.api.user.analytics.routes import _resolve_agent from application.storage.db.repositories.agents import AgentsRepository agent = AgentsRepository(pg_conn).create( "owner", "test-agent", "published", key="secret" ) - assert _resolve_api_key(pg_conn, str(agent["id"]), "other-user") is None + assert _resolve_agent(pg_conn, str(agent["id"]), "other-user") == ( + None, + "", + None, + ) class TestGetMessageAnalytics: diff --git a/tests/e2e/specs/tier-a/user-logs.spec.ts b/tests/e2e/specs/tier-a/user-logs.spec.ts index 4b36d170..69a559e5 100644 --- a/tests/e2e/specs/tier-a/user-logs.spec.ts +++ b/tests/e2e/specs/tier-a/user-logs.spec.ts @@ -47,7 +47,6 @@ interface GetLogsResponse { user: string | null; question: string | null; sources: unknown; - retriever_params: unknown; timestamp: string | null; }>; page: number;