diff --git a/application/agents/tool_executor.py b/application/agents/tool_executor.py index c86c298b..f197f762 100644 --- a/application/agents/tool_executor.py +++ b/application/agents/tool_executor.py @@ -40,10 +40,18 @@ 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 - is now journaled (newly created or already present). + still run when the journal is unreachable. Returns True iff THIS call + created the row. + + A duplicate ``call_id`` (LLMs reuse "call_0"-style ids) hits + ``ON CONFLICT DO NOTHING`` and returns False: the existing row may + belong to another in-flight request, so callers must not then flip it + via ``_mark_failed`` / ``_mark_executed``. """ try: with db_session() as conn: @@ -53,6 +61,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( @@ -60,7 +75,7 @@ def _record_proposed( call_id, extra={"alert": "tool_call_id_collision", "call_id": call_id}, ) - return True + return inserted except Exception: logger.exception("tool_call_attempts proposed write failed for %s", call_id) return False @@ -77,11 +92,14 @@ def _mark_executed( action_name: Optional[str] = None, arguments: Any = None, tool_id: Optional[str] = None, + user_id: Optional[str] = None, + agent_id: Optional[str] = None, ) -> None: """Flip the row to ``executed``. If ``proposed_ok`` is False (the proposed write failed earlier), upsert a fresh row in ``executed`` so the reconciler can still see the attempt — without this, the side - effect would be invisible to the journal. + effect would be invisible to the journal. Both paths are scoped to + the owning ``user_id`` so a reused ``call_id`` can't cross tenants. """ try: with db_session() as conn: @@ -92,6 +110,7 @@ def _mark_executed( result, message_id=message_id, artifact_id=artifact_id, + user_id=user_id, ) if updated: return @@ -105,15 +124,23 @@ def _mark_executed( tool_id=tool_id if tool_id and looks_like_uuid(tool_id) else None, message_id=message_id, artifact_id=artifact_id, + user_id=user_id, + agent_id=( + str(agent_id) + if agent_id and looks_like_uuid(str(agent_id)) + else None + ), ) except Exception: logger.exception("tool_call_attempts executed write failed for %s", call_id) -def _mark_failed(call_id: str, error: str) -> None: +def _mark_failed(call_id: str, error: str, *, user_id: Optional[str] = None) -> None: try: with db_session() as conn: - ToolCallAttemptsRepository(conn).mark_failed(call_id, error) + ToolCallAttemptsRepository(conn).mark_failed( + call_id, error, user_id=user_id + ) except Exception: logger.exception("tool_call_attempts failed-write failed for %s", call_id) @@ -551,6 +578,19 @@ 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"], user_id=self.user + ) 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 @@ -574,6 +614,21 @@ 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.", + user_id=self.user, + ) 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 @@ -599,6 +654,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 @@ -612,7 +670,8 @@ class ToolExecutor: ) tool_call_data["result"] = error_message tool_call_data["arguments"] = {} - _mark_failed(call_id, error_message) + if proposed_ok: + _mark_failed(call_id, error_message, user_id=self.user) yield { "type": "tool_call", "data": {**tool_call_data, "status": "error"}, @@ -675,7 +734,8 @@ class ToolExecutor: }, ) tool_call_data["result"] = error_message - _mark_failed(call_id, error_message) + if proposed_ok: + _mark_failed(call_id, error_message, user_id=self.user) yield {"type": "tool_call", "data": {**tool_call_data, "status": "error"}} self.tool_calls.append(tool_call_data) return error_message, call_id @@ -695,7 +755,8 @@ class ToolExecutor: logger.debug(f"Executing tool: {action_name} with args: {call_args}") result = tool.execute_action(action_name, **parameters) except Exception as exc: - _mark_failed(call_id, str(exc)) + if proposed_ok: + _mark_failed(call_id, str(exc), user_id=self.user) raise get_artifact_id = ( @@ -739,6 +800,8 @@ class ToolExecutor: action_name=action_name, arguments=call_args, tool_id=tool_data.get("id"), + user_id=self.user, + agent_id=self.agent_id, ) stream_tool_call_data = { 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..851d62c2 --- /dev/null +++ b/application/alembic/versions/0018_tool_attempts_attribution.py @@ -0,0 +1,56 @@ +"""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. + +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. + +The index is built ``CONCURRENTLY`` in an autocommit block: the +``ADD COLUMN`` ALTERs commit first (a fast metadata-only change), so the +O(table-size) index build never runs while their ACCESS EXCLUSIVE lock +is held and never blocks live journaling. + +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;") + # CONCURRENTLY can't run inside a transaction; the autocommit block + # commits the ALTERs above (releasing their lock) before building. + # DROP IF EXISTS first so a failed concurrent build (which leaves an + # INVALID index) is cleaned up on re-run. + with op.get_context().autocommit_block(): + op.execute("DROP INDEX IF EXISTS tool_call_attempts_user_ts_idx;") + op.execute( + "CREATE INDEX CONCURRENTLY tool_call_attempts_user_ts_idx " + "ON tool_call_attempts (user_id, attempted_at DESC) " + "WHERE user_id IS NOT NULL;" + ) + + +def downgrade() -> None: + with op.get_context().autocommit_block(): + op.execute( + "DROP INDEX CONCURRENTLY 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..9bf3ba1a 100644 --- a/application/api/user/analytics/routes.py +++ b/application/api/user/analytics/routes.py @@ -12,9 +12,9 @@ from application.api.user.base import ( generate_hourly_range, generate_minute_range, ) +from application.storage.db.redaction import redact_secrets 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 +68,26 @@ 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)``. ``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. """ - 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 None + agent_pg_id = str(agent["id"]) if agent else None + return agent, api_key, agent_pg_id @analytics_ns.route("/get_message_analytics") @@ -115,26 +125,52 @@ 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 + ) + 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 api_key. The - # ``user_id`` filter is always applied post-cutover to - # prevent cross-tenant leakage on admin dashboards. + # Count messages per bucket. When filtering by agent the + # owner-scoped lookup above already gates access, so the + # user clause is dropped (matching tokens / tools / logs): + # a shared agent's conversations carry the caller's + # user_id, and the owner should see that traffic on their + # own agent's dashboard. Agent matching covers both + # shapes: external traffic stamps ``api_key``, owner / + # shared chats stamp ``agent_id``. clauses = [ - "c.user_id = :user_id", "m.timestamp >= :start", "m.timestamp <= :end", ] params: dict = { - "user_id": user, "start": start_date, "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 + else: + clauses.append("c.user_id = :user_id") + params["user_id"] = user where = " AND ".join(clauses) sql = ( "SELECT to_char(m.timestamp AT TIME ZONE 'UTC', :fmt) AS bucket, " @@ -172,6 +208,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 +235,21 @@ 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" + # ``@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: + 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 +257,74 @@ 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. - rows = TokenUsageRepository(conn).bucketed_totals( - bucket_unit=bucket_unit, - user_id=user, - api_key=api_key, - timestamp_gte=start_date, - timestamp_lt=end_date, + 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). + 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} - 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 +363,25 @@ 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 + ) + 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": "..."}``. @@ -265,20 +389,29 @@ class GetFeedbackAnalytics(Resource): # the timestamp from the JSONB and cast it to timestamptz for # the range filter + bucket grouping. clauses = [ - "c.user_id = :user_id", "m.feedback IS NOT NULL", "(m.feedback->>'timestamp')::timestamptz >= :start", "(m.feedback->>'timestamp')::timestamptz <= :end", ] params: dict = { - "user_id": user, "start": start_date, "end": end_date, "fmt": pg_fmt, } - if api_key: - clauses.append("c.api_key = :api_key") + if api_key_id: + # Owner-gated agent match (see GetMessageAnalytics): + # drop the user clause so shared-agent feedback is + # visible to the owner, consistent with the other + # per-agent charts. + 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 + else: + clauses.append("c.user_id = :user_id") + params["user_id"] = user where = " AND ".join(clauses) sql = ( "SELECT to_char(" @@ -313,6 +446,237 @@ 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 + ) + if api_key_id and agent is None: + return make_response( + jsonify({"success": True, "tools": []}), 200 + ) + + # Terminal rows only. ``proposed`` (pending) and + # ``executed`` (ran, not yet finalized) are non-terminal: + # counting them inflates ``calls`` and — since the client + # computes successful = calls - failures — renders them as + # phantom successes that later flip to failures when the + # reconciler escalates a stuck row. ``confirmed`` is the + # only success state; ``failed`` the only failure. + clauses = [ + "t.status IN ('confirmed', 'failed')", + "t.attempted_at >= :start", + "t.attempted_at <= :end", + ] + params: dict = { + "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), 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" + " OR c.agent_id = CAST(:agent_pg_id AS uuid))" + ) + 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, " + "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 + ) + 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). + ts = "COALESCE(r.finished_at, r.started_at, r.scheduled_for)" + clauses = [ + f"{ts} >= :start", + f"{ts} <= :end", + ] + params: dict = { + "start": start_date, + "end": end_date, + "fmt": pg_fmt, + } + if api_key_id: + # Owner-gated agent match: drop the user clause so a + # shared agent's runs (created by callers under their + # own user_id via the scheduler tool) are visible to + # the owner, consistent with the per-agent timeline. + clauses.append("r.agent_id = CAST(:agent_id AS uuid)") + params["agent_id"] = agent_pg_id + else: + clauses.append("r.user_id = :user_id") + params["user_id"] = user + 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,94 +693,409 @@ 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: 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") + 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, - ) - # 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 - ] + 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": results, + "logs": [], "page": page, "page_size": page_size, - "has_more": has_more_flag, + "has_more": False, } ), 200, ) + params: dict = { + "user_id": user, + "limit": page_size + 1, + "offset": (page - 1) * page_size, + } - 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" - ), + # ``schedule`` / ``webhook`` errors are first-class events in + # their own branches; keep them out of ``system`` so a failed + # 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: + # 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"] = ( + str(agent["workflow_id"]) + if agent and agent.get("workflow_id") + else None + ) + 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", + ] + # Owner-gated agent match: drop the user clause so a + # shared agent's runs (stamped with the caller's + # user_id by the scheduler tool) appear on the owner's + # per-agent timeline, consistent with the chat branch. + schedule_where = [ + "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 five event sources. + # ``payload`` carries the per-type detail; the outer query + # 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, + {level} AS level, + COALESCE(l.data->>'action', 'stream_answer') AS action, + {summary} AS summary, + l.data AS payload + FROM user_logs l + 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 {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 {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, + 'result', wr.result, + '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 {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, + '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 + ) AS payload + FROM schedule_runs r + LEFT JOIN schedules sc ON sc.id = r.schedule_id + WHERE {where} + """, + }, + ] + + if level: + params["level"] = level + if search: + escaped = ( + search.replace("\\", "\\\\") + .replace("%", "\\%") + .replace("_", "\\_") + ) + params["search"] = f"%{escaped}%" + + 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), + ) + ) + + # ``ev.id`` is a unique per-branch tiebreaker: equal + # timestamps (transaction-stable ``now()`` makes ties + # routine) would otherwise sort non-deterministically + # across page queries, duplicating a row on one page and + # dropping its sibling from the next under OFFSET paging. + sql = ( + "SELECT * FROM (" + + " UNION ALL ".join(branch_sqls) + + ") ev ORDER BY ev.timestamp DESC, ev.id 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"), + # Redact on the way out too: rows written + # before write-time redaction still carry the + # reflected provider/user secrets in ``stacks``. + "stacks": redact_secrets(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..6d325815 100644 --- a/application/llm/handlers/base.py +++ b/application/llm/handlers/base.py @@ -950,17 +950,21 @@ class LLMHandler(ABC): _record_proposed, ) - _record_proposed( + if _record_proposed( pause_info["call_id"], pause_info["tool_name"], pause_info["action_name"], pause_info.get("arguments") or {}, tool_id=pause_info.get("tool_id"), - ) - _mark_failed( - pause_info["call_id"], - f"headless: {deny_reason}", - ) + 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"], + f"headless: {deny_reason}", + user_id=agent.tool_executor.user, + ) yield { "type": "tool_call", "data": { 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/redaction.py b/application/storage/db/redaction.py new file mode 100644 index 00000000..cb1e4754 --- /dev/null +++ b/application/storage/db/redaction.py @@ -0,0 +1,64 @@ +"""Secret redaction for reflected ``stack_logs`` data. + +``stacks`` is built by reflecting every public attribute of runtime +objects (the ``llm`` component carries the deployment provider +``api_key`` and the caller's ``user_api_key``). The unified-logs endpoint +returns ``stacks`` to the client, so credentials must be scrubbed — at +write time for new rows, and at read time for rows written before +redaction existed. +""" + +from __future__ import annotations + +REDACTED = "[REDACTED]" + +# Substrings marking a key as a credential. Compound token-count fields +# (``prompt_tokens`` / ``generated_tokens`` / ``token_budget`` / ...) are +# intentionally not matched: secret token forms are spelled out +# (``access_token`` etc.) and a bare ``token`` is handled separately. +_SECRET_SUBSTRINGS = ( + "api_key", + "apikey", + "api_token", + "access_token", + "refresh_token", + "auth_token", + "id_token", + "session_token", + "secret", + "password", + "passwd", + "passphrase", + "private_key", + "credential", + "authorization", + "bearer", +) + + +def is_secret_key(key: str) -> bool: + """True when ``key`` names a credential that must not be persisted/returned.""" + k = key.lower() + if k == "token": + return True + return any(s in k for s in _SECRET_SUBSTRINGS) + + +def redact_secrets(obj): + """Recursively replace secret-keyed values with ``[REDACTED]``. + + Walks dicts and lists; leaf scalars pass through unchanged. ``None`` + returns ``None`` so callers can redact an optional payload directly. + """ + if isinstance(obj, dict): + return { + k: ( + REDACTED + if isinstance(k, str) and is_secret_key(k) + else redact_secrets(v) + ) + for k, v in obj.items() + } + if isinstance(obj, list): + return [redact_secrets(v) for v in obj] + return obj diff --git a/application/storage/db/repositories/stack_logs.py b/application/storage/db/repositories/stack_logs.py index a3cc3413..fc1a6387 100644 --- a/application/storage/db/repositories/stack_logs.py +++ b/application/storage/db/repositories/stack_logs.py @@ -13,6 +13,7 @@ import json from datetime import datetime from typing import Optional +from application.storage.db.redaction import redact_secrets from application.storage.db.serialization import PGNativeJSONEncoder from sqlalchemy import Connection, text @@ -54,7 +55,9 @@ class StackLogsRepository: "user_id": user_id, "api_key": api_key, "query": query, - "stacks": json.dumps(stacks or [], cls=PGNativeJSONEncoder), + "stacks": json.dumps( + redact_secrets(stacks or []), cls=PGNativeJSONEncoder + ), "timestamp": timestamp, }, ) 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..0954d733 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 @@ -62,13 +75,22 @@ class ToolCallAttemptsRepository: tool_id: Optional[str] = None, message_id: Optional[str] = None, artifact_id: Optional[str] = None, + user_id: Optional[str] = None, + agent_id: Optional[str] = None, ) -> None: """Insert OR upgrade a row to ``executed`` — or ``confirmed`` when there is no ``message_id``, as in ``mark_executed``. Used as a fallback when ``record_proposed`` failed (DB outage) and the tool ran anyway — preserves the journal so the - reconciler can still see the attempt. + reconciler can still see the attempt. ``user_id`` / ``agent_id`` + attribute the synthesized row so it stays visible to per-user / + per-agent analytics. + + The ``ON CONFLICT`` upgrade is guarded to a still-``proposed`` row + owned by the same ``user_id``: ``call_id`` is the table-wide PK + and LLMs reuse ids, so without it a colliding write could upgrade + another request's (possibly another tenant's) row. """ result_payload: dict = {"result": result} if artifact_id: @@ -79,16 +101,18 @@ class ToolCallAttemptsRepository: """ INSERT INTO tool_call_attempts (call_id, tool_id, tool_name, action_name, arguments, - result, message_id, status) + result, message_id, user_id, agent_id, status) VALUES (:call_id, CAST(:tool_id AS uuid), :tool_name, :action_name, CAST(:arguments AS jsonb), CAST(:result AS jsonb), CAST(:message_id AS uuid), - :status) + :user_id, CAST(:agent_id AS uuid), :status) ON CONFLICT (call_id) DO UPDATE SET status = :status, result = EXCLUDED.result, message_id = COALESCE(EXCLUDED.message_id, tool_call_attempts.message_id) + WHERE tool_call_attempts.status = 'proposed' + AND tool_call_attempts.user_id IS NOT DISTINCT FROM EXCLUDED.user_id """ ), { @@ -99,6 +123,8 @@ class ToolCallAttemptsRepository: "arguments": json.dumps(arguments if arguments is not None else {}, cls=PGNativeJSONEncoder), "result": json.dumps(result_payload, cls=PGNativeJSONEncoder), "message_id": message_id, + "user_id": user_id, + "agent_id": agent_id, "status": status, }, ) @@ -110,6 +136,7 @@ class ToolCallAttemptsRepository: *, message_id: Optional[str] = None, artifact_id: Optional[str] = None, + user_id: Optional[str] = None, ) -> bool: """Flip ``proposed`` → ``executed``, or straight to ``confirmed`` when there is no ``message_id`` (a ``save_conversation=False`` @@ -118,6 +145,11 @@ class ToolCallAttemptsRepository: ``artifact_id`` (when present) is stored alongside ``result`` in the JSONB as audit data — the reconciler reads it for diagnostic alerts when escalating stuck rows to ``failed``. + + Guarded to a still-``proposed`` row (and, when ``user_id`` is + given, to one owned by that user): ``call_id`` is the table-wide + PK and LLMs reuse ids, so an unguarded update could flip another + request's already-terminal — or another tenant's — row. """ result_payload: dict = {"result": result} if artifact_id: @@ -135,17 +167,33 @@ class ToolCallAttemptsRepository: if message_id is not None: sql += ", message_id = CAST(:message_id AS uuid)" params["message_id"] = message_id - sql += " WHERE call_id = :call_id" + sql += " WHERE call_id = :call_id AND status = 'proposed'" + if user_id is not None: + sql += " AND user_id IS NOT DISTINCT FROM :user_id" + params["user_id"] = user_id result_proxy = self._conn.execute(text(sql), params) return result_proxy.rowcount > 0 - def mark_failed(self, call_id: str, error: str) -> bool: - """Flip ``proposed`` → ``failed`` with the exception text.""" - result = self._conn.execute( - text( - "UPDATE tool_call_attempts SET status = 'failed', error = :error " - "WHERE call_id = :call_id" - ), - {"call_id": call_id, "error": error}, + def mark_failed( + self, call_id: str, error: str, *, user_id: Optional[str] = None + ) -> bool: + """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. When ``user_id`` is + given the update is additionally scoped to that owner, so a + colliding id from another tenant can't fail this row. + """ + sql = ( + "UPDATE tool_call_attempts SET status = 'failed', error = :error " + "WHERE call_id = :call_id AND status = 'proposed'" ) + params: dict[str, Any] = {"call_id": call_id, "error": error} + if user_id is not None: + sql += " AND user_id IS NOT DISTINCT FROM :user_id" + params["user_id"] = user_id + result = self._conn.execute(text(sql), params) return result.rowcount > 0 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/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 2007d474..54540f95 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..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'; @@ -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,8 +157,25 @@ 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]; + + // 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( @@ -136,35 +187,47 @@ 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); } }; - const fetchTokenData = async (agent_id?: string, filter?: string) => { + const fetchTokenData = async ( + agent_id?: string, + filter?: string, + groupBy?: TokenGroupBy, + sideChannel?: boolean, + ) => { + const reqId = ++requestIds.current.tokens; 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(); + 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( @@ -176,36 +239,187 @@ 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( + { + 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(); + if (reqId !== requestIds.current.tools) return; + setToolsData(data.tools); + } catch (error) { + console.error(error); + } finally { + 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( + { + 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(); + if (reqId !== requestIds.current.schedules) return; + setScheduleData(data.runs); + } catch (error) { + console.error(error); + } finally { + if (reqId === requestIds.current.schedules) setLoadingSchedules(false); } }; useEffect(() => { - const id = agentId; - const filter = messagesFilter; - fetchMessagesData(id, filter?.value); - }, [agentId, messagesFilter]); + fetchMessagesData(agentId, timeFilter.value); + fetchFeedbackData(agentId, timeFilter.value); + fetchToolsData(agentId, timeFilter.value); + fetchScheduleData(agentId, timeFilter.value); + }, [agentId, timeFilter]); useEffect(() => { - const id = agentId; - const filter = tokenUsageFilter; - fetchTokenData(id, filter?.value); - }, [agentId, tokenUsageFilter]); + 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), + })); - useEffect(() => { - const id = agentId; - const filter = feedbackFilter; - fetchFeedbackData(id, filter?.value); - }, [agentId, feedbackFilter]); return (
-

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

+
+

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

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

{card.label}

+

+ {card.value} +

+
+ ))} +
+ {/* Messages Analytics */}
@@ -213,29 +427,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 +608,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; + } + resetPendingRef.current = true; + setLogsByPage({}); + setPage(1); + setHasMore(true); + }, [levelFilter, typeFilter, search, agentId]); + const fetchLogs = async () => { if (logsByPage[page] && logsByPage[page].length > 0) return; + const issuedKey = filterKey; + const issuedPage = page; setLoadingLogs(true); try { const response = await userService.getLogs( @@ -35,32 +79,99 @@ 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 || resetPendingRef.current) return; setLogsByPage((prev) => ({ ...prev, - [page]: data.logs, + [issuedPage]: data.logs, })); setHasMore(data.has_more); } 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 (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]); + }, [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 +266,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 +289,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 +404,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/frontend/src/utils/dateTimeUtils.ts b/frontend/src/utils/dateTimeUtils.ts index b2fb65e8..ce7a2c39 100644 --- a/frontend/src/utils/dateTimeUtils.ts +++ b/frontend/src/utils/dateTimeUtils.ts @@ -28,7 +28,12 @@ export function formatDate(dateString: string): string { minute: '2-digit', }); } else if (/^\d{4}-\d{2}-\d{2}$/.test(dateString)) { - const date = new Date(dateString); + // `new Date('YYYY-MM-DD')` parses as UTC midnight, so rendering it + // in a timezone west of UTC shows the previous day. Construct from + // parts so the date is interpreted in local time (matching the + // space-separated hourly buckets, which already parse as local). + const [year, month, day] = dateString.split('-').map(Number); + const date = new Date(year, month - 1, day); return date.toLocaleDateString('en-US', { month: 'short', day: 'numeric', diff --git a/scripts/db/backfill_tool_attempts_attribution.py b/scripts/db/backfill_tool_attempts_attribution.py new file mode 100644 index 00000000..f60f2dbe --- /dev/null +++ b/scripts/db/backfill_tool_attempts_attribution.py @@ -0,0 +1,110 @@ +"""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``. + +Message-less rows (headless: scheduled / webhook runs, plus pre-0018 +parse-failure rows) are left NULL on purpose: there is no FK linking an +attempt to its run, so any inference from a schedule-run *time window* +would also catch webhook attempts and misattribute them to an unrelated +tenant whose run happened to span the same instant. The analytics reader +treats unattributable rows as invisible rather than guessing an owner, +and new headless rows are stamped at propose time by the executor. + +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 + """ +) + +_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 + + after = conn.execute(_COUNT_NULL).scalar_one() + + print(f"NULL user_id rows before: {before}") + print(f" tier 1 (parent message): {t1}") + print(f"NULL user_id rows remaining: {after}") + print(" (message-less headless rows left NULL by design)") + + 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 b13b994f..f899dc9f 100644 --- a/tests/agents/test_tool_executor_three_phase.py +++ b/tests/agents/test_tool_executor_three_phase.py @@ -251,6 +251,105 @@ 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 + + def test_mark_executed_guarded_to_proposed(self, pg_conn): + """A reused ``call_id`` must not flip an already-terminal row back + to executed (the status guard mirrors ``mark_failed``).""" + from application.storage.db.repositories.tool_call_attempts import ( + ToolCallAttemptsRepository, + ) + + repo = ToolCallAttemptsRepository(pg_conn) + repo.record_proposed("c-me", "tool", "act", {}) + assert repo.mark_failed("c-me", "boom") is True + # Late executed emission for the same id: no proposed row remains. + assert repo.mark_executed("c-me", {"out": "ok"}) is False + row = _select_attempt(pg_conn, "c-me") + assert row["status"] == "failed" + assert row["result"] is None + + def test_mark_executed_scoped_to_owner(self, pg_conn): + """A colliding ``call_id`` from another tenant can't flip — or + read its result into — this user's proposed row.""" + from application.storage.db.repositories.tool_call_attempts import ( + ToolCallAttemptsRepository, + ) + + repo = ToolCallAttemptsRepository(pg_conn) + repo.record_proposed("call_0", "tool", "act", {}, user_id="victim") + # Attacker's request reuses the id and tries to land its result. + assert ( + repo.mark_executed("call_0", {"out": "leak"}, user_id="attacker") + is False + ) + row = _select_attempt(pg_conn, "call_0") + assert row["status"] == "proposed" + assert row["result"] is None + assert row["user_id"] == "victim" + + def test_mark_failed_scoped_to_owner(self, pg_conn): + from application.storage.db.repositories.tool_call_attempts import ( + ToolCallAttemptsRepository, + ) + + repo = ToolCallAttemptsRepository(pg_conn) + repo.record_proposed("call_0", "tool", "act", {}, user_id="victim") + assert repo.mark_failed("call_0", "boom", user_id="attacker") is False + row = _select_attempt(pg_conn, "call_0") + assert row["status"] == "proposed" + + def test_upsert_executed_stamps_attribution(self, pg_conn): + """The DB-outage fallback row must carry user/agent so it stays + visible to per-user / per-agent analytics (was born unattributed).""" + import uuid as _uuid + + from application.storage.db.repositories.tool_call_attempts import ( + ToolCallAttemptsRepository, + ) + + agent_id = str(_uuid.uuid4()) + repo = ToolCallAttemptsRepository(pg_conn) + repo.upsert_executed( + "c-up2", "tool", "act", {}, {"out": "ok"}, + user_id="u-up", agent_id=agent_id, + ) + row = _select_attempt(pg_conn, "c-up2") + assert row["user_id"] == "u-up" + assert str(row["agent_id"]) == agent_id + + def test_upsert_executed_wont_clobber_other_tenant(self, pg_conn): + """A colliding fallback upsert must not upgrade another tenant's + proposed row or overwrite its result.""" + from application.storage.db.repositories.tool_call_attempts import ( + ToolCallAttemptsRepository, + ) + + repo = ToolCallAttemptsRepository(pg_conn) + repo.record_proposed("call_0", "tool", "act", {}, user_id="victim") + repo.upsert_executed( + "call_0", "tool", "act", {}, {"out": "leak"}, user_id="attacker" + ) + row = _select_attempt(pg_conn, "call_0") + assert row["status"] == "proposed" + assert row["result"] is None + assert row["user_id"] == "victim" + @pytest.mark.unit class TestDefaultToolJournaling: diff --git a/tests/api/user/test_analytics.py b/tests/api/user/test_analytics.py index f69b462b..4afd3aec 100644 --- a/tests/api/user/test_analytics.py +++ b/tests/api/user/test_analytics.py @@ -31,12 +31,13 @@ def _patch_analytics_db(conn): def _seed_conversation_with_messages( pg_conn, user_id, *, count=3, api_key=None, feedback_text=None, + agent_id=None, ): from application.storage.db.repositories.conversations import ( ConversationsRepository, ) repo = ConversationsRepository(pg_conn) - conv = repo.create(user_id, name="t", api_key=api_key) + conv = repo.create(user_id, name="t", api_key=api_key, agent_id=agent_id) conv_id = str(conv["id"]) for i in range(count): repo.append_message(conv_id, {"prompt": f"p{i}", "response": f"r{i}"}) @@ -74,33 +75,56 @@ 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, None) + assert _resolve_agent(pg_conn, "", "u") == (None, 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_yields_none_key(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" + ) + # 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 is None + 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, + None, + ) class TestGetMessageAnalytics: @@ -406,3 +430,801 @@ 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_system_stacks_redacted_on_read(self, app, pg_conn): + # A row written before write-time redaction existed still holds + # the reflected provider secret in ``stacks``. The endpoint must + # scrub it on the way out, not hand it back to the client. + import json + + from sqlalchemy import text + + pg_conn.execute( + text( + """ + INSERT INTO stack_logs + (activity_id, endpoint, level, user_id, stacks) + VALUES + ('act-leak', 'stream', 'error', 'u-redact', + CAST(:stacks AS jsonb)) + """ + ), + { + "stacks": json.dumps( + [ + { + "component": "llm", + "data": { + "api_key": "sk-deployment-secret", + "user_api_key": "agent-key", + "model": "gpt-x", + }, + } + ] + ) + }, + ) + + response = _post_logs(app, pg_conn, "u-redact", {"event_type": "system"}) + assert response.status_code == 200 + logs = response.json["logs"] + assert len(logs) == 1 + data = logs[0]["stacks"][0]["data"] + assert data["api_key"] == "[REDACTED]" + assert data["user_api_key"] == "[REDACTED]" + assert data["model"] == "gpt-x" + + 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" + + +class TestSharedAgentVisibility: + """Per-agent dashboards show the agent's full traffic, including + conversations a *shared* agent's callers create under their own + user_id. Messages / feedback / schedule must agree with the chat + timeline, which already exposes that traffic — they used to keep a + ``c.user_id``/``r.user_id`` clause that hid it (one screen, two + populations).""" + + def _owned_agent(self, pg_conn, owner): + from application.storage.db.repositories.agents import AgentsRepository + + return AgentsRepository(pg_conn).create( + owner, "shared-agent", "published", key="shared-key", + ) + + def test_message_analytics_includes_shared_caller(self, app, pg_conn): + from application.api.user.analytics.routes import GetMessageAnalytics + + agent = self._owned_agent(pg_conn, "owner-a") + # Caller B chats with A's shared agent: conversation.user_id = B, + # agent_id = A's agent. + _seed_conversation_with_messages( + pg_conn, "caller-b", count=3, agent_id=str(agent["id"]) + ) + response = _post_resource( + app, pg_conn, GetMessageAnalytics, "/api/get_message_analytics", + "owner-a", {"api_key_id": str(agent["id"])}, + ) + assert response.status_code == 200 + assert sum(response.json["messages"].values()) == 3 + + def test_feedback_analytics_includes_shared_caller(self, app, pg_conn): + from application.api.user.analytics.routes import GetFeedbackAnalytics + + agent = self._owned_agent(pg_conn, "owner-a") + _seed_conversation_with_messages( + pg_conn, "caller-b", count=2, agent_id=str(agent["id"]), + feedback_text="like", + ) + response = _post_resource( + app, pg_conn, GetFeedbackAnalytics, "/api/get_feedback_analytics", + "owner-a", {"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"] + assert totals == {"positive": 1, "negative": 1} + + def test_schedule_analytics_includes_shared_caller(self, app, pg_conn): + helper = TestGetScheduleAnalytics() + agent = self._owned_agent(pg_conn, "owner-a") + # Run created by caller B against A's agent (scheduler tool stamps + # the caller's user_id). + helper._seed_run( + pg_conn, "caller-b", "success", agent_id=str(agent["id"]) + ) + response = helper._post( + app, pg_conn, "owner-a", {"api_key_id": str(agent["id"])} + ) + assert response.status_code == 200 + completed = sum(b["completed"] for b in response.json["runs"].values()) + assert completed == 1 + + def test_schedule_timeline_includes_shared_caller(self, app, pg_conn): + helper = TestGetScheduleAnalytics() + agent = self._owned_agent(pg_conn, "owner-a") + helper._seed_run( + pg_conn, "caller-b", "failed", agent_id=str(agent["id"]) + ) + response = _post_logs( + app, pg_conn, "owner-a", + {"api_key_id": str(agent["id"]), "event_type": "schedule"}, + ) + assert response.status_code == 200 + assert len(response.json["logs"]) == 1 + assert response.json["logs"][0]["level"] == "error" 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/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; diff --git a/tests/storage/db/repositories/test_stack_logs.py b/tests/storage/db/repositories/test_stack_logs.py index 0a9218dc..44325e7d 100644 --- a/tests/storage/db/repositories/test_stack_logs.py +++ b/tests/storage/db/repositories/test_stack_logs.py @@ -50,3 +50,79 @@ class TestInsert: text("SELECT query FROM stack_logs WHERE activity_id = 'act-3'") ).fetchone() assert len(dict(row._mapping)["query"]) == 20000 + + def test_secrets_redacted_from_stacks(self, pg_conn): + # The ``llm`` stack component is built from every public attr of + # the LLM and includes the provider secret + the caller's key. + # These must never persist (the unified-logs endpoint returns + # stacks verbatim). + repo = _repo(pg_conn) + repo.insert( + activity_id="act-secret", + level="error", + stacks=[ + { + "component": "llm", + "data": { + "api_key": "sk-deployment-secret", + "user_api_key": "agent-key", + "OPENAI_API_KEY": "sk-env", + "model": "gpt-x", + "prompt_tokens": 42, + }, + } + ], + ) + row = pg_conn.execute( + text("SELECT stacks FROM stack_logs WHERE activity_id = 'act-secret'") + ).fetchone() + data = dict(row._mapping)["stacks"][0]["data"] + assert data["api_key"] == "[REDACTED]" + assert data["user_api_key"] == "[REDACTED]" + assert data["OPENAI_API_KEY"] == "[REDACTED]" + # Non-secret fields (incl. token *counts*) are untouched. + assert data["model"] == "gpt-x" + assert data["prompt_tokens"] == 42 + + def test_redacts_broadened_secret_keys(self, pg_conn): + # Credentials beyond ``*api_key`` — OAuth tokens, bearer/auth + # headers, private keys — must also be scrubbed, while token + # *count* fields are preserved. + repo = _repo(pg_conn) + repo.insert( + activity_id="act-broad", + level="error", + stacks=[ + { + "component": "tool", + "data": { + "access_token": "ya29.secret", + "api_token": "r8_secret", + "token": "hf_secret", + "Authorization": "Bearer xyz", + "private_key": "-----BEGIN-----", + "client_secret": "cs_secret", + "prompt_tokens": 10, + "generated_tokens": 5, + "token_budget": 1000, + }, + } + ], + ) + row = pg_conn.execute( + text("SELECT stacks FROM stack_logs WHERE activity_id = 'act-broad'") + ).fetchone() + data = dict(row._mapping)["stacks"][0]["data"] + for key in ( + "access_token", + "api_token", + "token", + "Authorization", + "private_key", + "client_secret", + ): + assert data[key] == "[REDACTED]", key + # Token *counts* / budgets are not credentials and must survive. + assert data["prompt_tokens"] == 10 + assert data["generated_tokens"] == 5 + assert data["token_budget"] == 1000 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"