diff --git a/studio/backend/routes/profile_stats.py b/studio/backend/routes/profile_stats.py index 239ad98275..f474cf0426 100644 --- a/studio/backend/routes/profile_stats.py +++ b/studio/backend/routes/profile_stats.py @@ -15,7 +15,11 @@ from fastapi import APIRouter, Depends, Query from auth.authentication import get_current_subject from loggers import get_logger -from storage.profile_stats_db import MAX_DAILY_DAYS, compute_profile_stats +from storage.profile_stats_db import ( + MAX_DAILY_DAYS, + MAX_TZ_OFFSET_MINUTES, + compute_profile_stats, +) from utils.utils import log_and_http_error router = APIRouter() @@ -26,14 +30,22 @@ logger = get_logger(__name__) @router.get("/stats") async def get_profile_stats( days: int = Query(MAX_DAILY_DAYS, ge = 1, le = MAX_DAILY_DAYS), + tz_offset_minutes: int = Query(0, ge = -MAX_TZ_OFFSET_MINUTES, le = MAX_TZ_OFFSET_MINUTES), current_subject: str = Depends(get_current_subject), ) -> dict[str, Any]: - """Usage stats for the signed-in user's local history.""" + """Usage stats for the signed-in user's local history. + + ``tz_offset_minutes`` is the caller's ``Date.getTimezoneOffset()``. Days and + hours are bucketed with it so a remote browser does not read the server's + calendar. + """ try: # A cold pass parses every message's metadata JSON: ~90 ms at 10k # messages, ~1.2 s at 260k. Off the event loop so it cannot stall token # streaming when Settings is opened mid-generation. - return await asyncio.to_thread(compute_profile_stats, days = days) + return await asyncio.to_thread( + compute_profile_stats, days = days, tz_offset_minutes = tz_offset_minutes + ) except Exception as exc: raise log_and_http_error( exc, 500, "Failed to compute profile statistics", log = logger diff --git a/studio/backend/storage/profile_stats_db.py b/studio/backend/storage/profile_stats_db.py index 94e82af88a..28865bf2c2 100644 --- a/studio/backend/storage/profile_stats_db.py +++ b/studio/backend/storage/profile_stats_db.py @@ -19,7 +19,7 @@ the Profile tab is free until history changes. import json import threading import time -from datetime import date, datetime, timedelta +from datetime import date, datetime, timedelta, timezone from typing import Any, Optional from loggers import get_logger @@ -33,6 +33,8 @@ logger = get_logger(__name__) SESSION_GAP_SECONDS = 30 * 60 # Cap on the daily activity series handed to the UI (the heatmap draws a year). MAX_DAILY_DAYS = 366 +# Widest real UTC offset is 14h; anything beyond that is a bad client value. +MAX_TZ_OFFSET_MINUTES = 14 * 60 # Top-N lists returned to the client. TOP_MODELS = 8 RECENT_RUNS = 5 @@ -66,6 +68,18 @@ def _iso(day: date) -> str: return day.isoformat() +def _local_stamp(created_at_ms: int, tz_offset_minutes: int) -> Optional[datetime]: + """Wall-clock time in the caller's timezone, not the server's. + + ``tz_offset_minutes`` follows the browser's ``getTimezoneOffset()``: minutes + to add to local time to reach UTC, so UTC-5 sends 300. + """ + if created_at_ms <= 0: + return None + utc = datetime.fromtimestamp(created_at_ms / 1000, tz = timezone.utc) + return (utc - timedelta(minutes = tz_offset_minutes)).replace(tzinfo = None) + + def _streaks(days: set[date], today: date) -> dict[str, Any]: """Current and longest run of consecutive active days. @@ -146,12 +160,13 @@ class _MessageFold: bucket["threads"].add(thread_id) -def _fold_messages(conn) -> _MessageFold: +def _fold_messages(conn, tz_offset_minutes: int = 0) -> _MessageFold: fold = _MessageFold() rows = conn.execute( """ SELECT m.thread_id, m.role, m.metadata_json, m.attachments_json, m.created_at, - t.title, t.model_id, t.model_type + t.title, t.model_id, t.model_type, + t.created_at AS thread_created_at, t.forked_from_thread_id FROM chat_messages m LEFT JOIN chat_threads t ON t.id = m.thread_id ORDER BY m.thread_id, m.created_at @@ -187,6 +202,13 @@ def _fold_messages(conn) -> _MessageFold: thread_messages = 0 previous_created = None + # Forking clones the whole ancestry into the new thread, keeping each + # copy's original timestamp. Counting those again would double every + # metric for the branched-from conversation, so skip anything older + # than the fork itself. + if row["forked_from_thread_id"] and created_at < _as_int(row["thread_created_at"]): + continue + fold.threads.add(thread_id) fold.messages += 1 thread_messages += 1 @@ -197,7 +219,7 @@ def _fold_messages(conn) -> _MessageFold: thread_seconds += gap previous_created = created_at - stamp = datetime.fromtimestamp(created_at / 1000) if created_at > 0 else None + stamp = _local_stamp(created_at, tz_offset_minutes) role = row["role"] if role == "user": fold.user_messages += 1 @@ -245,11 +267,13 @@ def _fold_messages(conn) -> _MessageFold: fold.tool_calls += _as_int(timing.get("toolCallCount")) message_tokens = total_tokens - model_id = usage.get("modelId") - if not isinstance(model_id, str) or not model_id.strip(): - model_id = row["model_id"] if isinstance(row["model_id"], str) else "" - if model_id.strip(): - fold.note_model(model_id.strip(), message_tokens) + # Only the checkpoint recorded on the turn itself. The thread's + # model_id tracks whatever is selected now, so using it as a + # fallback credits older turns to the wrong model after a switch. + raw_model_id = usage.get("modelId") + model_id = raw_model_id.strip() if isinstance(raw_model_id, str) else "" + if model_id: + fold.note_model(model_id, message_tokens) speed = _as_float(timing.get("tokensPerSecond")) # llama.cpp reports absurd rates on no-op turns; ignore those. @@ -262,10 +286,10 @@ def _fold_messages(conn) -> _MessageFold: stream_ms = _as_float(timing.get("totalStreamTime")) if stream_ms is not None and stream_ms > 0: fold.response_ms.append(stream_ms) - start_ms = _as_float(timing.get("streamStartTime")) + # firstTokenTime is already an elapsed duration, not a timestamp. first_token = _as_float(timing.get("firstTokenTime")) - if start_ms and first_token and first_token > start_ms: - fold.first_token_ms.append(first_token - start_ms) + if first_token is not None and first_token > 0: + fold.first_token_ms.append(first_token) if stamp is not None: fold.by_hour[stamp.hour] += 1 @@ -299,7 +323,6 @@ def _training_stats(conn) -> dict[str, Any]: """ SELECT COUNT(*) AS runs, SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) AS completed, - SUM(COALESCE(final_step, 0)) AS steps, SUM(COALESCE(duration_seconds, 0)) AS seconds, COUNT(DISTINCT model_name) AS models, COUNT(DISTINCT dataset_name) AS datasets, @@ -308,7 +331,28 @@ def _training_stats(conn) -> dict[str, Any]: """ ).fetchone() - tokens = conn.execute("SELECT COALESCE(SUM(num_tokens), 0) FROM training_metrics").fetchone()[0] + # A resumed run continues its source's step and token counters from the + # checkpoint, so both absolute totals already include the source's work. + # resume_blocked marks a run that some later run resumed from; counting + # only the unresumed tails avoids adding the same progress twice. + steps = conn.execute( + "SELECT COALESCE(SUM(final_step), 0) FROM training_runs WHERE resume_blocked = 0" + ).fetchone()[0] + + # num_tokens is state.num_input_tokens_seen, a running total logged at each + # step, so summing the samples multiplies the real figure. Take each run's + # final counter, the same value get_run_metrics reports. + tokens = conn.execute( + """ + SELECT COALESCE(SUM(run_tokens), 0) FROM ( + SELECT MAX(m.num_tokens) AS run_tokens + FROM training_metrics m + JOIN training_runs r ON r.id = m.run_id + WHERE r.resume_blocked = 0 + GROUP BY m.run_id + ) + """ + ).fetchone()[0] recent = conn.execute( """ @@ -324,7 +368,7 @@ def _training_stats(conn) -> dict[str, Any]: return { "runs": _as_int(row["runs"]), "completed": _as_int(row["completed"]), - "steps": _as_int(row["steps"]), + "steps": _as_int(steps), "tokens": _as_int(tokens), "seconds": _as_float(row["seconds"]) or 0.0, "models": _as_int(row["models"]), @@ -357,12 +401,15 @@ def _fingerprint(conn) -> tuple: return (message_row[0], message_row[1], run_row[0], run_row[1]) -def compute_profile_stats(days: int = MAX_DAILY_DAYS) -> dict[str, Any]: +def compute_profile_stats(days: int = MAX_DAILY_DAYS, tz_offset_minutes: int = 0) -> dict[str, Any]: """Aggregate every profile statistic in one pass, memoised per history state.""" days = max(1, min(int(days), MAX_DAILY_DAYS)) + tz_offset_minutes = max( + -MAX_TZ_OFFSET_MINUTES, min(int(tz_offset_minutes), MAX_TZ_OFFSET_MINUTES) + ) conn = get_connection() try: - fingerprint = (_fingerprint(conn), days) + fingerprint = (_fingerprint(conn), days, tz_offset_minutes) now = time.monotonic() with _cache_lock: if ( @@ -373,10 +420,12 @@ def compute_profile_stats(days: int = MAX_DAILY_DAYS) -> dict[str, Any]: return _cache["payload"] started = time.perf_counter() - fold = _fold_messages(conn) + fold = _fold_messages(conn, tz_offset_minutes) training = _training_stats(conn) - today = date.today() + # "Today" has to match the buckets above, or the newest column and the + # current streak drift by a day whenever the caller is elsewhere. + today = (_local_stamp(int(time.time() * 1000), tz_offset_minutes) or datetime.now()).date() streak = _streaks(set(fold.by_day.keys()), today) daily = _daily_series(fold, today, days) diff --git a/studio/backend/tests/test_profile_stats.py b/studio/backend/tests/test_profile_stats.py index 35a8d72a20..aa4bc52bcf 100644 --- a/studio/backend/tests/test_profile_stats.py +++ b/studio/backend/tests/test_profile_stats.py @@ -5,7 +5,7 @@ import json import time -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone import pytest @@ -76,8 +76,10 @@ def _metadata( "modelId": "unsloth/gpt-oss-20b", }, "timing": { - "streamStartTime": 1000, - "firstTokenTime": 1200, + # The adapter writes streamStartTime as an epoch stamp and + # firstTokenTime as the elapsed ms before the first chunk. + "streamStartTime": 1_760_000_000_000, + "firstTokenTime": 200, "totalStreamTime": 2000, "tokenCount": completion, "tokensPerSecond": speed, @@ -166,8 +168,10 @@ def test_completion_tokens_fall_back_to_adapter_count(stats_db): assert stats["totals"]["completionTokens"] == 64 assert stats["totals"]["totalTokens"] == 64 - # No modelId in metadata: the thread's model is used instead. - assert stats["models"][0]["id"] == "local-gguf" + # No modelId on the turn, so it is not credited to any model. The thread's + # model_id follows the current selection and would misattribute after a + # mid-conversation switch. + assert stats["models"] == [] def test_session_time_ignores_long_idle_gaps(stats_db): @@ -236,9 +240,10 @@ def test_training_totals(stats_db): "VALUES ('r2', 'error', 'unsloth/qwen3-4b', 'my/dataset', '{}', " "'2026-01-02T10:00:00', 100, 20, 1.8, 600)", ) + # num_tokens is a running total, so the last row is the run's figure. conn.executemany( "INSERT INTO training_metrics (run_id, step, loss, num_tokens) VALUES (?, ?, ?, ?)", - [("r1", step, 1.0, 1000) for step in range(10)], + [("r1", step, 1.0, (step + 1) * 1000) for step in range(10)], ) conn.commit() finally: @@ -258,6 +263,149 @@ def test_training_totals(stats_db): assert training["recent"][0]["modelLabel"] == "qwen3-4b" +def test_first_token_time_is_read_as_a_duration(stats_db): + """firstTokenTime is `Date.now() - streamStartTime`, not a wall-clock stamp. + + Treating it as a stamp and subtracting streamStartTime made the comparison + fail for every real message, so the average was always empty. + """ + now = datetime.now() + conn = studio_db.get_connection() + try: + _seed_thread(conn, "tft", "m", [(now, _metadata(10, 10))]) + conn.commit() + finally: + conn.close() + + stats = compute_profile_stats(days = 7) + + assert stats["speed"]["averageFirstTokenMs"] == pytest.approx(200.0) + + +def test_forked_threads_do_not_double_count_copied_history(stats_db): + """Forking clones the ancestry, so the copies must not be counted again.""" + now = datetime.now().replace(hour = 12, minute = 0, second = 0, microsecond = 0) + conn = studio_db.get_connection() + try: + _seed_thread(conn, "src", "m", [(now - timedelta(hours = 2), _metadata(100, 50))]) + conn.commit() + finally: + conn.close() + + before = compute_profile_stats(days = 7) + assert before["totals"]["totalTokens"] == 150 + assert before["totals"]["messages"] == 2 + + fork_at = now + conn = studio_db.get_connection() + try: + conn.execute( + "INSERT INTO chat_threads (id, title, model_type, model_id, created_at, " + "updated_at, forked_from_thread_id, forked_from_message_id) " + "VALUES ('fork', 'fork of src', 'base', 'm', ?, ?, 'src', 'src-a0')", + (_ms(fork_at), _ms(fork_at)), + ) + # The clone keeps the original timestamp, exactly as fork_chat_thread does. + conn.execute( + "INSERT INTO chat_messages (id, thread_id, role, content_json, metadata_json, " + "created_at) VALUES ('fork-a0', 'fork', 'assistant', '[]', ?, ?)", + (json.dumps(_metadata(100, 50)), _ms(now - timedelta(hours = 2))), + ) + conn.commit() + finally: + conn.close() + + invalidate_profile_stats_cache() + after = compute_profile_stats(days = 7) + + assert after["totals"]["totalTokens"] == 150 + assert after["totals"]["messages"] == 2 + + # A genuinely new turn in the fork still counts. + conn = studio_db.get_connection() + try: + conn.execute( + "INSERT INTO chat_messages (id, thread_id, role, content_json, metadata_json, " + "created_at) VALUES ('fork-a1', 'fork', 'assistant', '[]', ?, ?)", + (json.dumps(_metadata(10, 5)), _ms(fork_at + timedelta(minutes = 1))), + ) + conn.commit() + finally: + conn.close() + + invalidate_profile_stats_cache() + grown = compute_profile_stats(days = 7) + assert grown["totals"]["totalTokens"] == 165 + assert grown["totals"]["messages"] == 3 + + +def test_resumed_runs_do_not_double_count_steps_or_tokens(stats_db): + """A resume continues the source's counters, so only the tail is counted.""" + conn = studio_db.get_connection() + try: + # 'stopped' at step 10, then claimed by the resume below. + conn.execute( + "INSERT INTO training_runs (id, status, model_name, dataset_name, config_json, " + "started_at, total_steps, final_step, duration_seconds, resume_blocked) " + "VALUES ('src', 'stopped', 'm', 'd', '{}', '2026-01-01T10:00:00', 20, 10, 600, 1)", + ) + conn.execute( + "INSERT INTO training_runs (id, status, model_name, dataset_name, config_json, " + "started_at, total_steps, final_step, duration_seconds, resume_blocked) " + "VALUES ('cont', 'completed', 'm', 'd', '{}', '2026-01-02T10:00:00', 20, 15, 300, 0)", + ) + conn.executemany( + "INSERT INTO training_metrics (run_id, step, num_tokens) VALUES (?, ?, ?)", + # The continuation's counter picks up where the source stopped. + [("src", step, step * 100) for step in range(1, 11)] + + [("cont", step, step * 100) for step in range(11, 16)], + ) + conn.commit() + finally: + conn.close() + + training = compute_profile_stats(days = 7)["training"] + + # Training reached step 15, not 10 + 15. + assert training["steps"] == 15 + assert training["tokens"] == 1500 + # Both attempts still show up as runs. + assert training["runs"] == 2 + + +def test_days_and_hours_use_the_callers_timezone(stats_db): + """A remote browser must not be bucketed against the server's calendar.""" + # 01:30 UTC. In UTC that is one day; at UTC-4 it is 21:30 the day before. + when = datetime(2026, 3, 10, 1, 30, tzinfo = timezone.utc) + conn = studio_db.get_connection() + try: + conn.execute( + "INSERT INTO chat_threads (id, title, model_type, model_id, created_at, updated_at) " + "VALUES ('tz', 'tz', 'base', 'm', ?, ?)", + (int(when.timestamp() * 1000), int(when.timestamp() * 1000)), + ) + conn.execute( + "INSERT INTO chat_messages (id, thread_id, role, content_json, metadata_json, " + "created_at) VALUES ('tz-a0', 'tz', 'assistant', '[]', ?, ?)", + (json.dumps(_metadata(10, 10)), int(when.timestamp() * 1000)), + ) + conn.commit() + finally: + conn.close() + + at_utc = compute_profile_stats(days = 366, tz_offset_minutes = 0) + invalidate_profile_stats_cache() + at_minus_four = compute_profile_stats(days = 366, tz_offset_minutes = 240) + + assert at_utc["hourly"][1] == 1 + assert at_minus_four["hourly"][21] == 1 + + utc_days = {day["date"] for day in at_utc["daily"] if day["messages"]} + local_days = {day["date"] for day in at_minus_four["daily"] if day["messages"]} + assert utc_days == {"2026-03-10"} + assert local_days == {"2026-03-09"} + + def test_repeat_calls_are_served_from_cache_until_history_changes(stats_db): now = datetime.now() conn = studio_db.get_connection() @@ -302,7 +450,7 @@ def test_route_does_not_block_the_event_loop(stats_db, monkeypatch): from routes import profile_stats as route_module - def slow_compute(days = 366): + def slow_compute(days = 366, tz_offset_minutes = 0): time.sleep(0.5) return {"totals": {"messages": 0}} @@ -319,7 +467,9 @@ def test_route_does_not_block_the_event_loop(stats_db, monkeypatch): beat = asyncio.create_task(heartbeat()) try: - await route_module.get_profile_stats(days = 366, current_subject = "unsloth") + await route_module.get_profile_stats( + days = 366, tz_offset_minutes = 0, current_subject = "unsloth" + ) finally: beat.cancel() return ticks diff --git a/studio/frontend/src/features/profile/api/profile-stats.ts b/studio/frontend/src/features/profile/api/profile-stats.ts index e79878e880..529b5e66f6 100644 --- a/studio/frontend/src/features/profile/api/profile-stats.ts +++ b/studio/frontend/src/features/profile/api/profile-stats.ts @@ -90,7 +90,12 @@ export type ProfileStats = { export async function loadProfileStats( signal?: AbortSignal, ): Promise { - const res = await authFetch("/api/profile/stats", { signal }); + // Bucket days and hours in this browser's timezone, which is not the + // server's when Studio is reached over the network. + const query = new URLSearchParams({ + tz_offset_minutes: String(new Date().getTimezoneOffset()), + }); + const res = await authFetch(`/api/profile/stats?${query}`, { signal }); if (!res.ok) { throw new Error(await readFastApiError(res, "Failed to load your stats")); }