wehub-resource-sync
fed8b2eed7
Backend release / release (push) Waiting to run
Bandit Security Scan / bandit_scan (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / manifest (push) Blocked by required conditions
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / manifest (push) Blocked by required conditions
Python linting / ruff (push) Waiting to run
Run python tests with pytest / Run tests and count coverage (3.12) (push) Waiting to run
React Widget Build / build (push) Waiting to run
1117 lines
48 KiB
Python
1117 lines
48 KiB
Python
"""Analytics and reporting routes."""
|
|
|
|
import datetime
|
|
|
|
from flask import current_app, jsonify, make_response, request
|
|
from flask_restx import fields, Namespace, Resource
|
|
from sqlalchemy import text as _sql_text
|
|
|
|
from application.api import api
|
|
from application.api.user.base import (
|
|
generate_date_range,
|
|
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.session import db_readonly
|
|
|
|
|
|
analytics_ns = Namespace(
|
|
"analytics", description="Analytics and reporting operations", path="/api"
|
|
)
|
|
|
|
|
|
_FILTER_BUCKETS = {
|
|
"last_hour": ("minute", "%Y-%m-%d %H:%M:00", "YYYY-MM-DD HH24:MI:00"),
|
|
"last_24_hour": ("hour", "%Y-%m-%d %H:00", "YYYY-MM-DD HH24:00"),
|
|
"last_7_days": ("day", "%Y-%m-%d", "YYYY-MM-DD"),
|
|
"last_15_days": ("day", "%Y-%m-%d", "YYYY-MM-DD"),
|
|
"last_30_days": ("day", "%Y-%m-%d", "YYYY-MM-DD"),
|
|
}
|
|
|
|
|
|
def _range_for_filter(filter_option: str):
|
|
"""Return ``(start_date, end_date, bucket_unit, pg_fmt)`` for the filter.
|
|
|
|
Returns ``None`` on invalid filter.
|
|
"""
|
|
if filter_option not in _FILTER_BUCKETS:
|
|
return None
|
|
end_date = datetime.datetime.now(datetime.timezone.utc)
|
|
bucket_unit, _py_fmt, pg_fmt = _FILTER_BUCKETS[filter_option]
|
|
|
|
if filter_option == "last_hour":
|
|
start_date = end_date - datetime.timedelta(hours=1)
|
|
elif filter_option == "last_24_hour":
|
|
start_date = end_date - datetime.timedelta(hours=24)
|
|
else:
|
|
days = {
|
|
"last_7_days": 6,
|
|
"last_15_days": 14,
|
|
"last_30_days": 29,
|
|
}[filter_option]
|
|
start_date = end_date - datetime.timedelta(days=days)
|
|
start_date = start_date.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
end_date = end_date.replace(
|
|
hour=23, minute=59, second=59, microsecond=999999
|
|
)
|
|
return start_date, end_date, bucket_unit, pg_fmt
|
|
|
|
|
|
def _intervals_for_filter(filter_option, start_date, end_date):
|
|
if filter_option == "last_hour":
|
|
return generate_minute_range(start_date, end_date)
|
|
if filter_option == "last_24_hour":
|
|
return generate_hourly_range(start_date, end_date)
|
|
return generate_date_range(start_date, end_date)
|
|
|
|
|
|
def _resolve_agent(conn, api_key_id, user_id):
|
|
"""Owner-scoped agent lookup for analytics filters.
|
|
|
|
Returns ``(agent, api_key, agent_pg_id)``. ``agent`` is ``None`` when
|
|
the id doesn't resolve to one of the caller's agents — callers must
|
|
short-circuit with an empty result, not fall back to sentinel filter
|
|
values. ``api_key`` is ``None`` (never ``""``) for key-less agents:
|
|
draft agents store ``key = ''``, and an ``''`` filter would match the
|
|
``''`` that writers like ``stack_logs`` stamp on every key-less
|
|
request — leaking rows across users. NULL matches nothing. Accepts
|
|
UUID or legacy Mongo ObjectId ids.
|
|
"""
|
|
agent = (
|
|
AgentsRepository(conn).get_any(api_key_id, user_id)
|
|
if api_key_id
|
|
else None
|
|
)
|
|
api_key = (agent or {}).get("key") or None
|
|
agent_pg_id = str(agent["id"]) if agent else None
|
|
return agent, api_key, agent_pg_id
|
|
|
|
|
|
@analytics_ns.route("/get_message_analytics")
|
|
class GetMessageAnalytics(Resource):
|
|
get_message_analytics_model = api.model(
|
|
"GetMessageAnalyticsModel",
|
|
{
|
|
"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_message_analytics_model)
|
|
@api.doc(description="Get message analytics based on filter option")
|
|
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:
|
|
# 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. 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 = [
|
|
"m.timestamp >= :start",
|
|
"m.timestamp <= :end",
|
|
]
|
|
params: dict = {
|
|
"start": start_date,
|
|
"end": end_date,
|
|
"fmt": pg_fmt,
|
|
}
|
|
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, "
|
|
"COUNT(*) AS count "
|
|
"FROM conversation_messages m "
|
|
"JOIN conversations c ON c.id = m.conversation_id "
|
|
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)
|
|
daily_messages = {interval: 0 for interval in intervals}
|
|
for row in rows:
|
|
daily_messages[row._mapping["bucket"]] = int(row._mapping["count"])
|
|
except Exception as err:
|
|
current_app.logger.error(
|
|
f"Error getting message analytics: {err}", exc_info=True
|
|
)
|
|
return make_response(jsonify({"success": False}), 400)
|
|
return make_response(
|
|
jsonify({"success": True, "messages": daily_messages}), 200
|
|
)
|
|
|
|
|
|
@analytics_ns.route("/get_token_analytics")
|
|
class GetTokenAnalytics(Resource):
|
|
get_token_analytics_model = api.model(
|
|
"GetTokenAnalyticsModel",
|
|
{
|
|
"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()),
|
|
),
|
|
"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,
|
|
),
|
|
},
|
|
)
|
|
|
|
@api.expect(get_token_analytics_model)
|
|
@api.doc(description="Get token analytics data")
|
|
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")
|
|
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 or group_by not in ("none", "model", "agent", "source"):
|
|
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:
|
|
# 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}
|
|
# ``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,
|
|
"group_by": group_by,
|
|
"series": series,
|
|
}
|
|
),
|
|
200,
|
|
)
|
|
|
|
|
|
@analytics_ns.route("/get_feedback_analytics")
|
|
class GetFeedbackAnalytics(Resource):
|
|
get_feedback_analytics_model = api.model(
|
|
"GetFeedbackAnalyticsModel",
|
|
{
|
|
"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_feedback_analytics_model)
|
|
@api.doc(description="Get feedback analytics data")
|
|
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,
|
|
"feedback": {
|
|
i: {"positive": 0, "negative": 0}
|
|
for i in intervals
|
|
},
|
|
}
|
|
),
|
|
200,
|
|
)
|
|
|
|
# Feedback lives inside the ``conversation_messages.feedback``
|
|
# JSONB as ``{"text": "like"|"dislike", "timestamp": "..."}``.
|
|
# There is no scalar ``feedback_timestamp`` column — extract
|
|
# the timestamp from the JSONB and cast it to timestamptz for
|
|
# the range filter + bucket grouping.
|
|
clauses = [
|
|
"m.feedback IS NOT NULL",
|
|
"(m.feedback->>'timestamp')::timestamptz >= :start",
|
|
"(m.feedback->>'timestamp')::timestamptz <= :end",
|
|
]
|
|
params: dict = {
|
|
"start": start_date,
|
|
"end": end_date,
|
|
"fmt": pg_fmt,
|
|
}
|
|
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("
|
|
"(m.feedback->>'timestamp')::timestamptz AT TIME ZONE 'UTC', :fmt"
|
|
") AS bucket, "
|
|
"SUM(CASE WHEN m.feedback->>'text' = 'like' THEN 1 ELSE 0 END) AS positive, "
|
|
"SUM(CASE WHEN m.feedback->>'text' = 'dislike' THEN 1 ELSE 0 END) AS negative "
|
|
"FROM conversation_messages m "
|
|
"JOIN conversations c ON c.id = m.conversation_id "
|
|
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)
|
|
daily_feedback = {
|
|
interval: {"positive": 0, "negative": 0} for interval in intervals
|
|
}
|
|
for row in rows:
|
|
bucket = row._mapping["bucket"]
|
|
daily_feedback[bucket] = {
|
|
"positive": int(row._mapping["positive"] or 0),
|
|
"negative": int(row._mapping["negative"] or 0),
|
|
}
|
|
except Exception as err:
|
|
current_app.logger.error(
|
|
f"Error getting feedback analytics: {err}", exc_info=True
|
|
)
|
|
return make_response(jsonify({"success": False}), 400)
|
|
return make_response(
|
|
jsonify({"success": True, "feedback": daily_feedback}), 200
|
|
)
|
|
|
|
|
|
@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(
|
|
"GetUserLogsModel",
|
|
{
|
|
"page": fields.Integer(
|
|
required=False,
|
|
description="Page number for pagination",
|
|
default=1,
|
|
),
|
|
"api_key_id": fields.String(required=False, description="API Key ID"),
|
|
"page_size": fields.Integer(
|
|
required=False,
|
|
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 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 {}
|
|
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")
|
|
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:
|
|
agent, api_key, agent_pg_id = _resolve_agent(
|
|
conn, api_key_id, user
|
|
)
|
|
if api_key_id and agent is None:
|
|
# Unknown / not-owned agent: empty page, not a
|
|
# sentinel filter (see _resolve_agent).
|
|
return make_response(
|
|
jsonify(
|
|
{
|
|
"success": True,
|
|
"logs": [],
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"has_more": False,
|
|
}
|
|
),
|
|
200,
|
|
)
|
|
params: dict = {
|
|
"user_id": user,
|
|
"limit": page_size + 1,
|
|
"offset": (page - 1) * page_size,
|
|
}
|
|
|
|
# ``schedule`` / ``webhook`` errors are first-class events in
|
|
# their own branches; keep them out of ``system`` so a failed
|
|
# run doesn't appear twice. 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": (
|
|
m["timestamp"].isoformat()
|
|
if hasattr(m["timestamp"], "isoformat")
|
|
else m["timestamp"]
|
|
),
|
|
}
|
|
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
|
|
)
|
|
return make_response(jsonify({"success": False}), 400)
|
|
|
|
return make_response(
|
|
jsonify(
|
|
{
|
|
"success": True,
|
|
"logs": results,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"has_more": has_more,
|
|
}
|
|
),
|
|
200,
|
|
)
|