diff --git a/.claude/agents/senior-engineer.md b/.claude/agents/senior-engineer.md index 82eea3767..88bb3909b 100644 --- a/.claude/agents/senior-engineer.md +++ b/.claude/agents/senior-engineer.md @@ -46,6 +46,9 @@ to what the task needs. **Cross-cutting:** when a service or crud function wraps an external SDK or raw HTTP call, also Read `.claude/conventions/error-handling.md` and apply its source-tagged, fault-based pattern. + **Coding Style:** unroll loops for better readability and maintainability. Prefer clarity over clever + language tricks. Do not write one-liner loops that involves 1) nested loops 2) complex pydantic interfaces. + 4. **Before writing a helper, check it doesn't already exist.** Anything generic — wrapping an external SDK or its error handling, building/parsing a domain payload, hitting cloud storage, loading config — usually has a canonical version already, and it rarely lives in a neighbor file. diff --git a/backend/app/api/routes/cron.py b/backend/app/api/routes/cron.py index c31288ce8..9e11d033a 100644 --- a/backend/app/api/routes/cron.py +++ b/backend/app/api/routes/cron.py @@ -9,6 +9,8 @@ from app.core.config import settings from app.crud.evaluations import process_all_pending_evaluations from app.services.job_monitoring import monitor_pending_jobs +from app.crud.stats import StatRow, get_daily_stats +from app.services.stats import format_sections, post_to_discord logger = logging.getLogger(__name__) @@ -33,6 +35,16 @@ "recovery_threshold": 1, } +DAILY_STATS_CRON_MONITOR_CONFIG: MonitorConfig = { + "schedule": {"type": "crontab", "value": "0 9 * * *"}, + "timezone": "UTC", + "checkin_margin": 5, + "max_runtime": 10, + "failure_issue_threshold": 1, + "recovery_threshold": 1, +} + + PENDING_JOBS_CRON_MONITOR_CONFIG: MonitorConfig = { "schedule": { "type": "interval", @@ -123,6 +135,29 @@ async def evaluation_cron_job( raise +@router.get( + "/cron/daily-stats", + include_in_schema=False, + dependencies=[Depends(require_permission(Permission.SUPERUSER))], +) +@sentry_sdk.monitor( + monitor_slug="daily-stats-cron-job", + monitor_config=DAILY_STATS_CRON_MONITOR_CONFIG, +) +def daily_stats_cron_job(session: SessionDep) -> dict[str, list[StatRow]]: + try: + stats = get_daily_stats(session=session) + post_to_discord(format_sections(stats)) + return stats + except Exception as e: + logger.error( + f"[daily_stats_cron_job] Error executing cron job: {e}", + exc_info=True, + ) + sentry_sdk.capture_exception(e) + raise + + @router.get( "/cron/pending-jobs", include_in_schema=False, diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 4d1544da3..3fa659cfa 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -47,6 +47,7 @@ class Settings(BaseSettings): PROJECT_NAME: str API_VERSION: str = "0.5.0" SENTRY_DSN: HttpUrl | None = None + DISCORD_STATS_WEBHOOK_URL: HttpUrl | None = None POSTGRES_SERVER: str POSTGRES_PORT: int = 5432 POSTGRES_USER: str diff --git a/backend/app/crud/stats.py b/backend/app/crud/stats.py new file mode 100644 index 000000000..a13d6b38f --- /dev/null +++ b/backend/app/crud/stats.py @@ -0,0 +1,122 @@ +import re + +from sqlalchemy import text +from sqlmodel import Session + +StatValue = str | int | float +StatRow = dict[str, StatValue] + + +LLM_CALLS = """ + SELECT + o.name AS organization, + p.name AS project, + COUNT(*) FILTER (WHERE l.inserted_at >= now() - INTERVAL '24 hours') AS calls_24h, + COUNT(*) AS calls_7d + FROM llm_call l + INNER JOIN organization o ON l.organization_id = o.id + INNER JOIN project p ON l.project_id = p.id + WHERE l.inserted_at >= now() - INTERVAL '168 hours' + AND l.deleted_at IS NULL + GROUP BY o.name, p.name + ORDER BY calls_7d DESC +""" + +LLM_TOKENS = """ + SELECT + o.name AS organization, + p.name AS project, + l.model AS model, + COALESCE(SUM((l.usage->>'total_tokens')::INTEGER) + FILTER (WHERE l.inserted_at >= now() - INTERVAL '24 hours'), 0) AS tokens_24h, + COALESCE(SUM((l.usage->>'total_tokens')::INTEGER), 0) AS tokens_7d + FROM llm_call l + INNER JOIN organization o ON l.organization_id = o.id + INNER JOIN project p ON l.project_id = p.id + WHERE l.inserted_at >= now() - INTERVAL '168 hours' + AND l.deleted_at IS NULL + GROUP BY o.name, p.name, l.model + ORDER BY tokens_7d DESC +""" + +LLM_MODALITY = """ + SELECT + o.name AS organization, + p.name AS project, + CASE + WHEN l.input_type = 'text' AND l.output_type = 'text' THEN 'TEXT' + WHEN l.input_type = 'audio' AND l.output_type = 'text' THEN 'STT' + WHEN l.input_type = 'text' AND l.output_type = 'audio' THEN 'TTS' + ELSE 'OTHER' + END AS modality, + COUNT(*) FILTER (WHERE l.inserted_at >= now() - INTERVAL '24 hours') AS calls_24h, + COUNT(*) AS calls_7d + FROM llm_call l + INNER JOIN organization o ON l.organization_id = o.id + INNER JOIN project p ON l.project_id = p.id + WHERE l.inserted_at >= now() - INTERVAL '168 hours' + AND l.deleted_at IS NULL + GROUP BY o.name, p.name, modality + ORDER BY o.name, p.name, modality +""" + +JOBS = """ + SELECT + o.name AS organization, + p.name AS project, + j.job_type AS job_type, + COUNT(*) FILTER (WHERE j.inserted_at >= now() - INTERVAL '24 hours') AS jobs_24h, + COUNT(*) AS jobs_7d + FROM job j + INNER JOIN project p ON j.project_id = p.id + INNER JOIN organization o ON p.organization_id = o.id + WHERE j.inserted_at >= now() - INTERVAL '168 hours' + GROUP BY o.name, p.name, j.job_type + ORDER BY o.name, p.name, j.job_type +""" + +_SIMPLE_COUNT = """ + SELECT + o.name AS organization, + p.name AS project, + COUNT(*) FILTER (WHERE t.inserted_at >= now() - INTERVAL '24 hours') AS count_24h, + COUNT(*) AS count_7d + FROM {table} t + INNER JOIN organization o ON t.organization_id = o.id + INNER JOIN project p ON t.project_id = p.id + WHERE t.inserted_at >= now() - INTERVAL '168 hours' + GROUP BY o.name, p.name + ORDER BY count_7d DESC +""" + +SIMPLE_COUNT_TABLES = { + "Evaluation Runs": "evaluation_run", + "STT Results": "stt_result", + "TTS Results": "tts_result", + "Assessments": "assessment", +} + +_IDENTIFIER = re.compile(r"^[a-z_][a-z0-9_]*$") + + +def _simple_count_sql(table: str) -> str: + if not _IDENTIFIER.match(table): + raise ValueError(f"unsafe table identifier: {table!r}") + return _SIMPLE_COUNT.format(table=table) + + +def _rows(session: Session, sql: str) -> list[StatRow]: + result = session.connection().execute(text(sql)) + return [dict(row) for row in result.mappings().all()] + + +def get_daily_stats(*, session: Session) -> dict[str, list[StatRow]]: + stats: dict[str, list[StatRow]] = { + "LLM Calls": _rows(session, LLM_CALLS), + "LLM Tokens": _rows(session, LLM_TOKENS), + "LLM Modality": _rows(session, LLM_MODALITY), + "Jobs by Type": _rows(session, JOBS), + } + for label, table in SIMPLE_COUNT_TABLES.items(): + stats[label] = _rows(session, _simple_count_sql(table)) + return stats diff --git a/backend/app/services/stats.py b/backend/app/services/stats.py new file mode 100644 index 000000000..da92b1b0b --- /dev/null +++ b/backend/app/services/stats.py @@ -0,0 +1,165 @@ +import logging +from datetime import date +from typing import TypedDict + +import requests + +from app.core.config import settings +from app.crud.stats import StatRow, StatValue + +logger = logging.getLogger(__name__) + +DISCORD_EMBED_TOTAL_TEXT_LIMIT = ( + 5900 # Discord caps all embed text in a message at 6000 chars. +) +DISCORD_EMBED_FIELD_VALUE_LIMIT = ( + 1000 # Discord caps a single field value at 1024 chars. +) +DISCORD_EMBED_FIELD_COUNT_LIMIT = 25 # Discord embed field cap. +DISCORD_EMBED_BORDER_COLOR = 0x3B82F6 # Blue left-border accent on the Discord embed. +COLUMN_LABELS = {"24h": "Last 24hrs", "7d": "Last 7 days"} + + +class EmbedField(TypedDict): + name: str + value: str + + +class DiscordEmbed(TypedDict): + title: str + description: str + color: int + fields: list[EmbedField] + + +def _column_label(column: str) -> str: + suffix = column.rsplit("_", 1)[-1] + return COLUMN_LABELS.get(suffix, column) + + +def format_sections(stats: dict[str, list[StatRow]]) -> list[EmbedField]: + """Build one Discord embed field per stat section (name = title, value = table).""" + fields: list[EmbedField] = [] + inactive_titles: list[str] = [] + for title, rows in stats.items(): + if not rows: + inactive_titles.append(title) + continue + + columns = list(rows[0].keys()) + numeric_cols = [ + c for c in columns if all(isinstance(row[c], (int, float)) for row in rows) + ] + text_cols = [c for c in columns if c not in numeric_cols] + + # Repeating org/project on every row is what pushed rows past Discord's + # embed width and wrapped the trailing columns. Instead, group rows by + # every text column but the last one and print that group once as a + # bold heading (bold only works outside the code block), leaving just + # the varying column + metrics — short enough to fit — in the table. + group_cols = text_cols[:-1] + row_col = text_cols[-1] if text_cols else None + table_cols = ([row_col] if row_col else []) + numeric_cols + headers = {c: _column_label(c) for c in table_cols} + + def cell(column: str, row: StatRow) -> str: + value = row[column] + return f"{value:,}" if column in numeric_cols else str(value) + + groups: dict[tuple[StatValue, ...], list[StatRow]] = {} + for row in rows: + groups.setdefault(tuple(row[c] for c in group_cols), []).append(row) + + blocks = [] + for key, group_rows in groups.items(): + # Column width is driven by the widest value, uncapped, so long + # names (e.g. model ids) are never truncated. + widths = { + c: max(len(headers[c]), max(len(cell(c, r)) for r in group_rows)) + for c in table_cols + } + + def align(text: str, column: str) -> str: + width = widths[column] + return ( + text.rjust(width) if column in numeric_cols else text.ljust(width) + ) + + lines = [ + " ".join(align(headers[c], c) for c in table_cols).rstrip(), + ] + for row in group_rows: + lines.append( + " ".join(align(cell(c, row), c) for c in table_cols).rstrip() + ) + + table = "\n".join(lines) + block = f"```\n{table}\n```" + if group_cols: + block = f"**{' / '.join(str(k) for k in key)}**\n{block}" + blocks.append(block) + + value = "\n".join(blocks) + if len(value) > DISCORD_EMBED_FIELD_VALUE_LIMIT: + cutoff = DISCORD_EMBED_FIELD_VALUE_LIMIT - len("\n…\n```") + truncated = value[:cutoff] + value = ( + f"{truncated}\n…\n```" + if truncated.count("```") % 2 == 1 + else f"{truncated}\n…" + ) + fields.append({"name": title, "value": value}) + + if inactive_titles: + bullets = "\n".join(f"• {t}" for t in inactive_titles) + fields.append({"name": "No activity this week", "value": bullets}) + return fields + + +def post_to_discord(fields: list[EmbedField], *, today: date | None = None) -> None: + url = settings.DISCORD_STATS_WEBHOOK_URL + if not url: + return + + stat_date = today or date.today() + title = f"Date: {stat_date.day}/{stat_date.month}/{stat_date.year}" + description = "Daily platform feature stats" + + embed = _new_embed(title, description) + for field in fields: + field_cost = len(field["name"]) + len(field["value"]) + if embed["fields"] and ( + len(embed["fields"]) >= DISCORD_EMBED_FIELD_COUNT_LIMIT + or _embed_size(embed) + field_cost > DISCORD_EMBED_TOTAL_TEXT_LIMIT + ): + _post(str(url), embed) + embed = _new_embed(title, description) + embed["fields"].append(field) + if embed["fields"]: + _post(str(url), embed) + + +def _new_embed(title: str, description: str) -> DiscordEmbed: + return { + "title": title, + "description": description, + "color": DISCORD_EMBED_BORDER_COLOR, + "fields": [], + } + + +def _embed_size(embed: DiscordEmbed) -> int: + return ( + len(embed["title"]) + + len(embed["description"]) + + sum(len(f["name"]) + len(f["value"]) for f in embed["fields"]) + ) + + +def _post(url: str, embed: DiscordEmbed) -> None: + try: + response = requests.post(url, json={"embeds": [embed]}, timeout=5) + response.raise_for_status() + except requests.RequestException as e: + # Log only the exception type — the message can contain the webhook URL. + logger.warning(f"[_post] Webhook post failed: {type(e).__name__}") diff --git a/backend/app/tests/api/routes/test_cron.py b/backend/app/tests/api/routes/test_cron.py index 7b52ee4f9..399ed6eee 100644 --- a/backend/app/tests/api/routes/test_cron.py +++ b/backend/app/tests/api/routes/test_cron.py @@ -1,7 +1,9 @@ -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch +import pytest from fastapi.testclient import TestClient +from app.api.routes import cron from app.core.config import settings from app.tests.utils.auth import TestAuthContext @@ -272,6 +274,58 @@ def test_pending_jobs_cron_job_requires_superuser( assert "superuser" in response_data["error"].lower() +def test_daily_stats_cron_job_success( + client: TestClient, + superuser_api_key: TestAuthContext, +) -> None: + """Returns the collected stats and posts them to Discord.""" + stats = { + "LLM Calls": [ + {"organization": "Acme", "project": "Alpha", "calls_24h": 1, "calls_7d": 2} + ], + } + with ( + patch("app.api.routes.cron.get_daily_stats", return_value=stats), + patch("app.api.routes.cron.post_to_discord") as post, + ): + response = client.get( + f"{settings.API_V1_STR}/cron/daily-stats", + headers={"X-API-KEY": superuser_api_key.key}, + ) + + assert response.status_code == 200 + assert response.json() == stats + post.assert_called_once() + + +def test_daily_stats_cron_job_requires_superuser( + client: TestClient, + user_api_key: TestAuthContext, +) -> None: + """Non-superuser cannot access the daily stats cron endpoint.""" + response = client.get( + f"{settings.API_V1_STR}/cron/daily-stats", + headers={"X-API-KEY": user_api_key.key}, + ) + + assert response.status_code == 403 + + +def test_daily_stats_cron_job_captures_and_reraises_on_error() -> None: + """On failure the job reports to Sentry and re-raises.""" + with ( + patch( + "app.api.routes.cron.get_daily_stats", + side_effect=RuntimeError("boom"), + ), + patch("app.api.routes.cron.sentry_sdk") as sentry, + ): + with pytest.raises(RuntimeError): + cron.daily_stats_cron_job(session=MagicMock()) + + sentry.capture_exception.assert_called_once() + + def test_evaluation_cron_job_not_in_schema( client: TestClient, ) -> None: @@ -285,6 +339,7 @@ def test_evaluation_cron_job_not_in_schema( # Endpoint should not be in the schema due to include_in_schema=False assert f"{settings.API_V1_STR}/cron/evaluations" not in paths assert f"{settings.API_V1_STR}/cron/pending-jobs" not in paths + assert f"{settings.API_V1_STR}/cron/daily-stats" not in paths def test_cron_intervals_match_to_prevent_sentry_monitor_drift() -> None: diff --git a/backend/app/tests/crud/test_stats.py b/backend/app/tests/crud/test_stats.py new file mode 100644 index 000000000..b7453ed24 --- /dev/null +++ b/backend/app/tests/crud/test_stats.py @@ -0,0 +1,27 @@ +from unittest.mock import MagicMock + +from app.crud.stats import get_daily_stats + +EXPECTED_SECTIONS = { + "LLM Calls", + "LLM Tokens", + "LLM Modality", + "Jobs by Type", + "Evaluation Runs", + "STT Results", + "TTS Results", + "Assessments", +} + + +def test_get_daily_stats_runs_every_section_and_maps_rows(): + row = {"organization": "Acme", "project": "Alpha", "calls_7d": 5} + session = MagicMock() + execute = session.connection.return_value.execute + execute.return_value.mappings.return_value.all.return_value = [row] + + stats = get_daily_stats(session=session) + + assert set(stats) == EXPECTED_SECTIONS + assert execute.call_count == len(EXPECTED_SECTIONS) # one query per section + assert stats["LLM Calls"] == [row] # _rows unpacks each mapping into a dict diff --git a/backend/app/tests/services/test_stats.py b/backend/app/tests/services/test_stats.py new file mode 100644 index 000000000..a6704677e --- /dev/null +++ b/backend/app/tests/services/test_stats.py @@ -0,0 +1,138 @@ +from datetime import date +from unittest.mock import MagicMock, patch + +import requests + +from app.services import stats as stats_mod +from app.services.stats import format_sections, post_to_discord + + +def _sample_stats() -> dict: + return { + "LLM Calls": [ + { + "organization": "Acme", + "project": "Alpha", + "calls_24h": 3, + "calls_7d": 15, + }, + ], + "STT Results": [], + } + + +def test_format_sections_groups_by_leading_columns_with_bold_heading(): + fields = format_sections(_sample_stats()) + llm_field = next(f for f in fields if f["name"] == "LLM Calls") + value = llm_field["value"] + assert value.startswith("**Acme**\n```\n") # bold heading outside the code block + assert "project Last 24hrs Last 7 days" in value + assert "Alpha 3 15" in value + assert value.count("```") == 2 # rows wrapped in one code block + + +def test_format_sections_groups_empty_sections_into_one_field(): + stats = { + "LLM Calls": [ + {"organization": "Acme", "project": "Alpha", "calls_24h": 3, "calls_7d": 15} + ], + "Evaluation Runs": [], + "STT Results": [], + "TTS Results": [], + "Assessments": [], + } + fields = format_sections(stats) + assert [f["name"] for f in fields] == ["LLM Calls", "No activity this week"] + inactive_field = fields[-1] + assert inactive_field["value"] == ( + "• Evaluation Runs\n• STT Results\n• TTS Results\n• Assessments" + ) + + +def test_format_sections_does_not_truncate_long_model_names(): + stats = { + "LLM Tokens": [ + { + "organization": "Org", + "model": "gemini-3.1-flash-tts-preview", # long, must stay whole + "tokens_7d": 89271, + }, + ], + } + value = format_sections(stats)[0]["value"] + assert "gemini-3.1-flash-tts-preview" in value # not clipped + assert "89,271" in value # thousands separator applied + + +def test_post_to_discord_noop_when_webhook_unset(): + with patch.object(stats_mod.settings, "DISCORD_STATS_WEBHOOK_URL", None), patch( + "app.services.stats.requests.post" + ) as mock_post: + post_to_discord([{"name": "X", "value": "y"}]) + mock_post.assert_not_called() + + +def test_post_to_discord_sets_title_description_and_border_color(): + posted: list[dict] = [] + + def fake_post(url, json, timeout): + posted.append(json["embeds"][0]) + return MagicMock() + + with patch.object( + stats_mod.settings, "DISCORD_STATS_WEBHOOK_URL", "https://x/hook" + ), patch("app.services.stats.requests.post", side_effect=fake_post): + post_to_discord( + [{"name": "LLM Calls", "value": "```\nx\n```"}], + today=date(2026, 8, 10), + ) + + assert len(posted) == 1 + embed = posted[0] + assert embed["title"] == "Date: 10/8/2026" + assert embed["description"] == "Daily platform feature stats" + assert embed["color"] == stats_mod.DISCORD_EMBED_BORDER_COLOR + assert embed["fields"] == [{"name": "LLM Calls", "value": "```\nx\n```"}] + + +def test_post_to_discord_splits_fields_across_embeds_under_size_limit(): + posted: list[dict] = [] + + def fake_post(url, json, timeout): + posted.append(json["embeds"][0]) + return MagicMock() + + big_fields = [{"name": f"Section {i}", "value": "x" * 2000} for i in range(4)] + with patch.object( + stats_mod.settings, "DISCORD_STATS_WEBHOOK_URL", "https://x/hook" + ), patch("app.services.stats.requests.post", side_effect=fake_post): + post_to_discord(big_fields) + + assert len(posted) >= 2 # split into multiple embeds/messages + for embed in posted: + size = ( + len(embed["title"]) + + len(embed["description"]) + + sum(len(f["name"]) + len(f["value"]) for f in embed["fields"]) + ) + assert size <= stats_mod.DISCORD_EMBED_TOTAL_TEXT_LIMIT + + +def test_post_to_discord_swallows_request_exception(): + with patch.object( + stats_mod.settings, "DISCORD_STATS_WEBHOOK_URL", "https://x/hook" + ), patch( + "app.services.stats.requests.post", + side_effect=requests.ConnectionError("boom"), + ): + post_to_discord([{"name": "X", "value": "y"}]) # must not raise + + +def test_post_to_discord_swallows_non_success_status(): + response = MagicMock() + response.raise_for_status.side_effect = requests.HTTPError("429 Too Many Requests") + with patch.object( + stats_mod.settings, "DISCORD_STATS_WEBHOOK_URL", "https://x/hook" + ), patch("app.services.stats.requests.post", return_value=response): + post_to_discord([{"name": "X", "value": "y"}]) # must not raise + response.raise_for_status.assert_called_once() diff --git a/scripts/python/invoke-cron.py b/scripts/python/invoke-cron.py index 64df37b25..e589c65aa 100644 --- a/scripts/python/invoke-cron.py +++ b/scripts/python/invoke-cron.py @@ -18,6 +18,7 @@ ENDPOINTS = [ "/api/v1/cron/evaluations", "/api/v1/cron/pending-jobs", + "/api/v1/cron/daily-stats", ] REQUEST_TIMEOUT = 30 # Timeout for requests in seconds