ai-agent/backend/dashboards/routes.py

"""Dashboards: the skills/cost rollup, the activity heatmap and the skills catalog."""

import threading

from fastapi import APIRouter

import schemas
from conversations.cards import (
    _by_account,
    _summaries_for,
    _summaries_with_account,
)
from core.http import _json_in_worker, _r
from core.state import AppState, State
from dashboards import activity as activity_mod
from dashboards import skills as skills_mod
from indexer import INPUT_PRICE_PER_MTOK, MODEL

router = APIRouter()


def _context_by_skill(files: dict) -> dict[str, dict]:
    """Group the indexed files' tokens/cost by the skill that owns them.

    A skill's *context cost* is what loading its whole dir
    (``.claude/skills/<name>/…``) into a session costs once. Shared by the
    dashboard rollup and the skills catalog."""
    by_skill: dict[str, dict] = {}
    for path, r in files.items():
        parts = path.split("/")
        if ".claude" in parts:
            i = parts.index(".claude")
            if len(parts) > i + 2 and parts[i + 1] == "skills":
                name = parts[i + 2]
                e = by_skill.setdefault(name, {"tokens": 0, "cost": 0.0})
                e["tokens"] += r["tokens"] or 0
                e["cost"] += r["cost"] or 0.0
    return by_skill


@router.get("/api/dashboard", responses=_r(schemas.Dashboard))
def dashboard(deps: State):
    """Skill-usage analytics + cost rollups for the homepage."""
    files = deps.store.all_files()

    total_tokens = sum((r["tokens"] or 0) for r in files.values())
    total_cost = sum((r["cost"] or 0.0) for r in files.values())
    estimated = sum(1 for r in files.values() if r["estimated"])

    by_skill = _context_by_skill(files)

    skills = []
    for row in deps.store.all_skills():
        ctx = by_skill.get(row["name"])
        ctx_tokens = ctx["tokens"] if ctx else None
        ctx_cost = ctx["cost"] if ctx else None
        skills.append({
            "name": row["name"],
            "count": row["count"],
            "lastUsed": row["last_used"],
            "contextTokens": ctx_tokens,
            "contextCost": ctx_cost,
            # estimated total spend = cost of loading the skill once × times used
            "estSpend": (ctx_cost * row["count"]) if ctx_cost is not None else None,
        })

    total_invocations = sum(s["count"] for s in skills)
    skill_spend = sum(s["estSpend"] for s in skills if s["estSpend"] is not None)

    top_files = sorted(
        ({"path": p, "tokens": r["tokens"] or 0, "cost": r["cost"] or 0.0,
          "estimated": bool(r["estimated"])}
         for p, r in files.items()),
        key=lambda f: f["cost"], reverse=True,
    )[:12]

    return {
        "pricing": {"model": MODEL, "inputPerMTok": INPUT_PRICE_PER_MTOK},
        "totals": {
            "files": len(files),
            "tokens": total_tokens,
            "cost": total_cost,
            "estimated": estimated,
            "skills": len(skills),
            "invocations": total_invocations,
            "skillSpend": skill_spend,
        },
        "skills": skills,
        "topFiles": top_files,
    }


# ── activity dashboard ──────────────────────────────────────────────────────
# The rollup walks every summary's pre-computed `activeBins`, so it is cheap —
# but the whole archive's worth of it is still a few thousand cells, and the
# page refetches on every SSE list ping. Memoized on (summaries version,
# timezone), same pattern as _skill_usage.
_activity_cache: tuple[tuple, dict] | None = None
_activity_lock = threading.Lock()


@router.get("/api/activity", responses=_r(schemas.ActivityResponse))
@_json_in_worker
def activity_rollup(deps: State, tzOffset: int = 0, account: str | None = None):
    """When the agents ran: per day, per 5-minute bin per model, per model.

    ``tzOffset`` is the caller's minutes east of UTC (the browser's
    ``-new Date().getTimezoneOffset()``); days and histogram slots are bucketed
    in that local time. Clamped to ±16 h — a nonsense offset would otherwise
    shift every day silently.

    ``account`` narrows it to one Claude account (``all`` = every one); unset
    is the aggregated view, without the accounts excluded from totals.
    ``byAccount`` always splits the whole archive, whatever the filter.
    """
    tz = max(-960, min(960, int(tzOffset)))
    global _activity_cache
    key = (deps.store.summaries_version, deps.meta_store.version, tz, account or "")
    with _activity_lock:
        if _activity_cache and _activity_cache[0] == key:
            return _activity_cache[1]
    out = activity_mod.rollup(_summaries_for(deps, account), tz)
    out["account"] = account or None
    out["byAccount"] = _by_account(_summaries_with_account(deps, ))
    with _activity_lock:
        _activity_cache = (key, out)
    return out


# ── skills catalog ──────────────────────────────────────────────────────────
# The per-skill usage rollup walks every transcript's skills blob and joins it to
# the conversation summaries; memoized on the summaries version so repeat
# requests between changes are a dict lookup (same pattern as _project_costs).
_skill_usage_cache: tuple[tuple, dict] | None = None
_skill_usage_lock = threading.Lock()


def _skill_usage(deps: AppState, account: str | None = None) -> dict:
    """Per-skill usage over one ``?account=`` view (default: the totals) —
    a skill's calls only count where their conversation's account does."""
    global _skill_usage_cache
    key = (deps.store.summaries_version, deps.meta_store.version, account or "")
    with _skill_usage_lock:
        if _skill_usage_cache and _skill_usage_cache[0] == key:
            return _skill_usage_cache[1]
    summaries = _summaries_for(deps, account)
    keep = {p for p, _ in summaries}
    usage = skills_mod.usage_by_skill(
        summaries, [r for r in deps.store.all_transcript_skill_rows()
                    if r[0] in keep])
    with _skill_usage_lock:
        _skill_usage_cache = (key, usage)
    return usage


@router.get("/api/skills", responses=_r(schemas.SkillsResponse))
@_json_in_worker
def skills_list(deps: State, account: str | None = None):
    """Card metadata + usage/cost stats for every skill, last-used first.

    The list is the **union** of the skills on disk and the ones the transcripts
    saw invoked: a built-in or plugin skill (``code-review``, ``artifact-design``)
    has no ``SKILL.md`` in this repo but still earned its calls, so it appears
    with ``sourceKind: "builtin"`` and no editor link.

    ``account`` scopes the usage/spend to one Claude account (``all`` = every
    one); unset = the aggregated view, without the excluded accounts.
    """
    usage = _skill_usage(deps, account)
    ctx = _context_by_skill(deps.store.all_files())

    def stats(name: str) -> dict:
        u = usage.get(name) or {}
        c = ctx.get(name)
        count = u.get("count", 0)
        ctx_cost = c["cost"] if c else None
        return {
            "count": count,
            "lastUsed": u.get("lastUsed"),
            "conversations": u.get("conversations", 0),
            "contextTokens": c["tokens"] if c else None,
            "contextCost": ctx_cost,
            # what invoking it has plausibly cost: loading its context, per call
            "estSpend": (ctx_cost * count) if ctx_cost is not None else None,
            # total spend of the conversations that used it (coarse — a session
            # is credited in full to every skill it invoked)
            "sessionCost": u.get("sessionCost", 0.0),
            "sessionTokens": u.get("sessionTokens", 0),
        }

    items = [{**s, **stats(s["name"])} for s in skills_mod.catalog()]
    known = {s["name"] for s in items}
    for name in sorted(usage.keys() - known):
        items.append({
            "name": name, "title": name, "description": "",
            "dir": None, "path": None,
            "source": "builtin", "sourceKind": "builtin",
            "triggerWords": [],
            "files": 0, "updatedAt": None,
            **stats(name),
        })

    # Last-used first (never-used skills sink to the bottom, alphabetical) — the
    # frontend re-sorts by the same field, this just makes the payload sane.
    items.sort(key=lambda s: (s["lastUsed"] or "", s["name"]), reverse=True)

    return {
        "skills": items,
        "totals": {
            "skills": len(items),
            "invocations": sum(s["count"] for s in items),
            "used": sum(1 for s in items if s["count"] > 0),
            "contextCost": sum(s["contextCost"] or 0.0 for s in items),
            "estSpend": sum(s["estSpend"] or 0.0 for s in items),
        },
    }