Merge pull request #2534 from arc53/logs-and-anal-revamp

revamp analytics & logs — per-agent attribution, unified log timeline
This commit is contained in:
Alex authored and GitHub committed 2026-06-13 14:25:32 +01:00
commit f8e2fb715e
25 files changed
+3028 -431

No files matched your search

+72 -9
View File
@@ -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 = {
@@ -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;")
+777 -98
View File
@@ -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
+10 -6
View File
@@ -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": {
+5
View File
@@ -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),
+64
View File
@@ -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
@@ -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,
},
)
@@ -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()
]
@@ -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
@@ -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.
+2 -4
View File
@@ -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",
+2
View File
@@ -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',
+4
View File
@@ -73,6 +73,10 @@ const userService = {
apiClient.post(endpoints.USER.TOKEN_ANALYTICS, data, token),
getFeedbackAnalytics: (data: any, token: string | null): Promise<any> =>
apiClient.post(endpoints.USER.FEEDBACK_ANALYTICS, data, token),
getToolAnalytics: (data: any, token: string | null): Promise<any> =>
apiClient.post(endpoints.USER.TOOL_ANALYTICS, data, token),
getScheduleAnalytics: (data: any, token: string | null): Promise<any> =>
apiClient.post(endpoints.USER.SCHEDULE_ANALYTICS, data, token),
getLogs: (data: any, token: string | null): Promise<any> =>
apiClient.post(endpoints.USER.LOGS, data, token),
manageSync: (data: any, token: string | null): Promise<any> =>
+62 -3
View File
@@ -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",
+357 -99
View File
@@ -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<Record<
string,
number
@@ -87,35 +117,39 @@ export default function Analytics({ agentId }: AnalyticsProps) {
string,
number
> | null>(null);
const [tokenSeries, setTokenSeries] = useState<Record<
string,
Record<string, number>
> | null>(null);
const [feedbackData, setFeedbackData] = useState<Record<
string,
{ positive: number; negative: number }
> | 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<Record<
string,
{ completed: number; failed: number; skipped: number }
> | null>(null);
const [timeFilter, setTimeFilter] = useState<{
label: string;
value: string;
}>({
label: t('settings.analytics.filterOptions.last30Days'),
value: 'last_30_days',
});
const [tokenGroupBy, setTokenGroupBy] = useState<TokenGroupBy>('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 (
<div className="mt-8">
<p className="text-muted-foreground mb-5 text-sm leading-6">
{t('settings.analytics.subtitle')}
</p>
<div className="mb-5 flex flex-row flex-wrap items-center justify-between gap-3">
<p className="text-muted-foreground text-sm leading-6">
{t('settings.analytics.subtitle')}
</p>
<Select
value={timeFilter.value}
onValueChange={(value) => {
const opt = filterOptions.find((o) => o.value === value);
if (opt) setTimeFilter(opt);
}}
>
<SelectTrigger
className="w-[125px] rounded-3xl px-5 py-3 text-sm"
size="lg"
>
<SelectValue
placeholder={t('settings.analytics.filterPlaceholder')}
/>
</SelectTrigger>
<SelectContent>
{filterOptions.map((o) => (
<SelectItem key={o.value} value={o.value}>
{o.label}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
{/* Summary stat cards */}
<div className="grid grid-cols-2 gap-3 md:grid-cols-3 lg:grid-cols-5">
{statCards.map((card) => (
<div
key={card.label}
className="border-border dark:border-border rounded-2xl border px-6 py-5"
>
<p className="text-muted-foreground text-sm">{card.label}</p>
<p className="text-foreground dark:text-foreground mt-1 text-2xl font-bold">
{card.value}
</p>
</div>
))}
</div>
{/* Messages Analytics */}
<div className="mt-4 flex w-full flex-col gap-3 [@media(min-width:1080px)]:flex-row">
<div className="border-border dark:border-border h-[345px] w-full overflow-hidden rounded-2xl border px-6 py-5 [@media(min-width:1080px)]:w-1/2">
@@ -213,29 +427,6 @@ export default function Analytics({ agentId }: AnalyticsProps) {
<p className="text-foreground dark:text-foreground font-bold">
{t('settings.analytics.messages')}
</p>
<Select
value={messagesFilter?.value}
onValueChange={(value) => {
const opt = filterOptions.find((o) => o.value === value);
if (opt) setMessagesFilter(opt);
}}
>
<SelectTrigger
className="w-[125px] rounded-3xl px-5 py-3 text-sm"
size="lg"
>
<SelectValue
placeholder={t('settings.analytics.filterPlaceholder')}
/>
</SelectTrigger>
<SelectContent>
{filterOptions.map((o) => (
<SelectItem key={o.value} value={o.value}>
{o.label}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
<div className="relative mt-px h-[245px] w-full">
<div
@@ -268,33 +459,37 @@ export default function Analytics({ agentId }: AnalyticsProps) {
{/* Token Usage Analytics */}
<div className="border-border dark:border-border h-[345px] w-full overflow-hidden rounded-2xl border px-6 py-5 [@media(min-width:1080px)]:w-1/2">
<div className="flex flex-row items-center justify-start gap-3">
<div className="flex flex-row flex-wrap items-center justify-start gap-3">
<p className="text-foreground dark:text-foreground font-bold">
{t('settings.analytics.tokenUsage')}
</p>
<Select
value={tokenUsageFilter?.value}
onValueChange={(value) => {
const opt = filterOptions.find((o) => o.value === value);
if (opt) setTokenUsageFilter(opt);
}}
value={tokenGroupBy}
onValueChange={(value) => setTokenGroupBy(value as TokenGroupBy)}
>
<SelectTrigger
className="w-[125px] rounded-3xl px-5 py-3 text-sm"
className="w-[110px] rounded-3xl px-5 py-3 text-sm"
size="lg"
>
<SelectValue
placeholder={t('settings.analytics.filterPlaceholder')}
/>
<SelectValue placeholder={t('settings.analytics.groupBy')} />
</SelectTrigger>
<SelectContent>
{filterOptions.map((o) => (
{groupByOptions.map((o) => (
<SelectItem key={o.value} value={o.value}>
{o.label}
</SelectItem>
))}
</SelectContent>
</Select>
<div className="ml-auto flex flex-row items-center gap-2">
<p className="text-muted-foreground text-xs">
{t('settings.analytics.includeSideChannel')}
</p>
<Switch
checked={includeSideChannel}
onCheckedChange={setIncludeSideChannel}
/>
</div>
</div>
<div className="relative mt-px h-[245px] w-full">
<div
@@ -306,20 +501,106 @@ export default function Analytics({ agentId }: AnalyticsProps) {
) : (
<AnalyticsChart
data={{
labels: Object.keys(tokenUsageData || {}).map((item) =>
labels: tokenLabels,
datasets: tokenDatasets,
}}
legendID="legend-container-2"
maxTicksLimitInX={8}
isStacked={true}
/>
)}
</div>
</div>
</div>
{/* Scheduled runs + tool usage */}
<div className="mt-4 flex w-full flex-col gap-3 [@media(min-width:1080px)]:flex-row">
<div className="border-border dark:border-border h-[345px] w-full overflow-hidden rounded-2xl border px-6 py-5 [@media(min-width:1080px)]:w-1/2">
<div className="flex flex-row items-center justify-start gap-3">
<p className="text-foreground dark:text-foreground font-bold">
{t('settings.analytics.scheduledRuns')}
</p>
</div>
<div className="relative mt-px h-[245px] w-full">
<div
id="legend-container-4"
className="flex flex-row items-center justify-end"
></div>
{loadingSchedules ? (
<SkeletonLoader count={1} component={'analysis'} />
) : (
<AnalyticsChart
data={{
labels: Object.keys(scheduleData || {}).map((item) =>
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}
/>
)}
</div>
</div>
<div className="border-border dark:border-border h-[345px] w-full overflow-hidden rounded-2xl border px-6 py-5 [@media(min-width:1080px)]:w-1/2">
<div className="flex flex-row items-center justify-start gap-3">
<p className="text-foreground dark:text-foreground font-bold">
{t('settings.analytics.toolUsage')}
</p>
</div>
<div className="relative mt-px h-[245px] w-full">
<div
id="legend-container-5"
className="flex flex-row items-center justify-end"
></div>
{loadingTools ? (
<SkeletonLoader count={1} component={'analysis'} />
) : (
<AnalyticsChart
data={{
labels: (toolsData || []).map((tool) => 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}
/>
)}
</div>
@@ -327,35 +608,12 @@ export default function Analytics({ agentId }: AnalyticsProps) {
</div>
{/* Feedback Analytics */}
<div className="mt-8 flex w-full flex-col gap-3">
<div className="mt-4 flex w-full flex-col gap-3">
<div className="border-border dark:border-border h-[345px] w-full overflow-hidden rounded-2xl border px-6 py-5">
<div className="flex flex-row items-center justify-start gap-3">
<p className="text-foreground dark:text-foreground font-bold">
{t('settings.analytics.userFeedback')}
</p>
<Select
value={feedbackFilter?.value}
onValueChange={(value) => {
const opt = filterOptions.find((o) => o.value === value);
if (opt) setFeedbackFilter(opt);
}}
>
<SelectTrigger
className="w-[125px] rounded-3xl px-5 py-3 text-sm"
size="lg"
>
<SelectValue
placeholder={t('settings.analytics.filterPlaceholder')}
/>
</SelectTrigger>
<SelectContent>
{filterOptions.map((o) => (
<SelectItem key={o.value} value={o.value}>
{o.label}
</SelectItem>
))}
</SelectContent>
</Select>
</div>
<div className="relative mt-px h-[245px] w-full">
<div
+272 -10
View File
@@ -6,6 +6,14 @@ import userService from '../api/services/userService';
import ChevronRight from '../assets/chevron-right.svg';
import CopyButton from '../components/CopyButton';
import SkeletonLoader from '../components/SkeletonLoader';
import { Input } from '../components/ui/input';
import {
Select,
SelectContent,
SelectItem,
SelectTrigger,
SelectValue,
} from '../components/ui/select';
import { useLoaderState } from '../hooks';
import { selectToken } from '../preferences/preferenceSlice';
import { LogData } from './types';
@@ -23,11 +31,47 @@ export default function Logs({ agentId, tableHeader }: LogsProps) {
const [hasMore, setHasMore] = useState(true);
const [loadingLogs, setLoadingLogs] = useLoaderState(true);
const [levelFilter, setLevelFilter] = useState('all');
const [typeFilter, setTypeFilter] = useState('all');
const [searchInput, setSearchInput] = useState('');
const [search, setSearch] = useState('');
const logs = Object.values(logsByPage).flat();
// Identifies the filter combination a request was issued under, so a
// slow response from a previous combination is discarded instead of
// landing in the freshly reset cache.
const filterKey = [agentId ?? '', levelFilter, typeFilter, search].join('|');
const filterKeyRef = useRef(filterKey);
filterKeyRef.current = filterKey;
const isFirstRender = useRef(true);
// Set synchronously by the reset effect so the fetch effect — which
// runs later in the same commit, still seeing the pre-reset `page` —
// skips that cycle instead of fetching a stale page under the new
// filters (which could latch hasMore=false and freeze the list).
const resetPendingRef = useRef(false);
useEffect(() => {
const handle = setTimeout(() => setSearch(searchInput.trim()), 400);
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 (
<div className="mt-8">
<p className="text-muted-foreground mb-5 text-sm leading-6">
{t('settings.logs.subtitle')}
</p>
<div className="mb-3 flex flex-row flex-wrap items-center gap-3">
<Select value={levelFilter} onValueChange={setLevelFilter}>
<SelectTrigger
className="w-[125px] rounded-3xl px-5 py-3 text-sm"
size="lg"
>
<SelectValue placeholder={t('settings.logs.levels.all')} />
</SelectTrigger>
<SelectContent>
{levelOptions.map((o) => (
<SelectItem key={o.value} value={o.value}>
{o.label}
</SelectItem>
))}
</SelectContent>
</Select>
<Select value={typeFilter} onValueChange={setTypeFilter}>
<SelectTrigger
className="w-[140px] rounded-3xl px-5 py-3 text-sm"
size="lg"
>
<SelectValue placeholder={t('settings.logs.types.all')} />
</SelectTrigger>
<SelectContent>
{typeOptions.map((o) => (
<SelectItem key={o.value} value={o.value}>
{o.label}
</SelectItem>
))}
</SelectContent>
</Select>
<Input
value={searchInput}
onChange={(e) => setSearchInput(e.target.value)}
placeholder={t('settings.logs.searchPlaceholder')}
className="w-56 rounded-3xl"
/>
</div>
<div>
<LogsTable
logs={logs}
@@ -124,6 +235,11 @@ function LogsTable({ logs, setPage, loading, tableHeader }: LogsTableProps) {
</p>
</div>
<div className="relative flex h-[51vh] grow flex-col items-start gap-2 overflow-y-auto overscroll-contain bg-transparent p-4">
{!loading && logs.length === 0 && (
<p className="text-muted-foreground w-full py-4 text-center text-xs">
{t('settings.logs.noLogs')}
</p>
)}
{logs?.map((log, index) => {
if (index === logs.length - 1) {
return (
@@ -150,6 +266,14 @@ function LogsTable({ logs, setPage, loading, tableHeader }: LogsTableProps) {
</div>
);
}
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 (
<div className="group dark:hover:bg-accent hover:bg-muted w-full rounded-xl bg-transparent">
@@ -180,8 +404,13 @@ function Log({
alt="Expand log entry"
className={`mt-[3px] h-3 w-3 transition duration-300 ${isOpen ? 'rotate-90' : ''}`}
/>
<span className="flex flex-row gap-2">
<span className="flex flex-row flex-wrap gap-2">
<h2 className="dark:text-foreground text-xs text-black/60">{`${log.timestamp}`}</h2>
{log.event_type && (
<h2 className="text-muted-foreground text-xs">
{t(`settings.logs.types.${log.event_type}`)}
</h2>
)}
<h2 className="text-xs text-[#913400] dark:text-orange-500">{`[${log.action}]`}</h2>
<h2
className={`max-w-72 text-xs ${logLevelColor[log.level]} wrap-break-word`}
@@ -194,11 +423,44 @@ function Log({
</div>
{isOpen && (
<div className="dark:bg-background rounded-b-xl bg-[#F1F1F1] px-4 py-3">
<div className="scrollbar-overlay overflow-y-auto">
<pre className="px-2 font-mono text-xs leading-relaxed wrap-break-word whitespace-pre-wrap text-gray-700 dark:text-gray-400">
{JSON.stringify(filteredLog, null, 2)}
</pre>
</div>
{detailRows.length > 0 && (
<div className="flex flex-col gap-1 px-2 pb-2">
{detailRows.map(([label, value]) => (
<div key={label} className="flex flex-row gap-2 text-xs">
<span className="text-muted-foreground w-28 shrink-0">
{label}
</span>
<span className="text-foreground wrap-break-word">
{value}
</span>
</div>
))}
</div>
)}
{textBlocks.map((block) => (
<div key={block.label} className="px-2 pb-2">
<p className="text-muted-foreground text-xs">{block.label}</p>
<pre
className={`font-mono text-xs leading-relaxed wrap-break-word whitespace-pre-wrap ${
block.isError
? 'text-red-500'
: 'text-gray-700 dark:text-gray-400'
}`}
>
{block.text}
</pre>
</div>
))}
{jsonBlocks.map((block) => (
<div key={block.label} className="px-2 pb-2">
<p className="text-muted-foreground text-xs">{block.label}</p>
<div className="scrollbar-overlay max-h-60 overflow-y-auto">
<pre className="font-mono text-xs leading-relaxed wrap-break-word whitespace-pre-wrap text-gray-700 dark:text-gray-400">
{JSON.stringify(block.value, null, 2)}
</pre>
</div>
</div>
))}
<div className="my-px w-fit">
<CopyButton
textToCopy={JSON.stringify(filteredLog)}
+35 -3
View File
@@ -15,16 +15,48 @@ export type APIKeyData = {
chunks: string;
};
export type LogEventType =
| 'chat'
| 'schedule'
| 'webhook'
| 'workflow'
| 'system';
export type LogData = {
id: string;
event_type?: LogEventType;
action: string;
level: 'info' | 'error' | 'warning';
user: string;
question: string;
response: string;
sources: Record<string, any>[];
retriever_params: Record<string, any>;
timestamp: string;
// chat events (user_logs)
response?: string;
sources?: Record<string, any>[];
tool_calls?: Record<string, any>[];
agent_id?: string;
attachments?: string[];
// system + webhook events (stack_logs)
endpoint?: string;
stacks?: Record<string, any>[];
// workflow events (workflow_runs)
workflow_name?: string;
result?: Record<string, any>;
steps?: Record<string, any>[];
// 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 = {
+6 -1
View File
@@ -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',
@@ -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())
@@ -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:
+836 -14
View File
@@ -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"
+10 -7
View File
@@ -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();
-1
View File
@@ -47,7 +47,6 @@ interface GetLogsResponse {
user: string | null;
question: string | null;
sources: unknown;
retriever_params: unknown;
timestamp: string | null;
}>;
page: number;
@@ -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
+25 -78
View File
@@ -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"