Gabriel Vidal
091fafd881
`complete` ships in both the homelab and the ai-agent repo, so /api/skills listed it twice with the same (name-keyed) usage, counted twice in the totals. merge_same_name keeps the repo's copy (else the newest) and lists the others under `copies` — shown as a +source badge on the card; their share refs stay pickable in Settings → Users. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
221 lines
8.9 KiB
Python
221 lines
8.9 KiB
Python
"""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),
|
||
}
|
||
|
||
shared = set(deps.users.defaults()["sharedSkills"])
|
||
items = skills_mod.merge_same_name(
|
||
[{**s, **stats(s["name"]), "shareRef": skills_mod.share_ref(s),
|
||
"sharedWithGuests": skills_mod.share_ref(s) in shared}
|
||
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),
|
||
},
|
||
}
|