# Agent Session Stores

Where each agent keeps local session history, and how far the claim is verified.
Read this before adding or fixing a provider in `scripts/collect.py`.

## Status of Each Provider

`verified` means the path and format were confirmed against a real session store
and the resulting counts were checked against an independent count. `inferred`
means the path and format come from documentation or published research and have
not been run against real data - those providers set `verified_format: false`, and
the dashboard labels their rows so a zero from them is not read as proof.

| agent | store | format | status |
|---|---|---|---|
| `claude-code` | `~/.claude/projects/<encoded-cwd>/<id>.jsonl` | JSONL | verified |
| `opencode` | `~/.local/share/opencode/opencode.db` | SQLite | verified |
| `codex` | `$CODEX_HOME/sessions/YYYY/MM/DD/rollout-*.jsonl` (default `~/.codex`) | JSONL | inferred |
| `cursor` | `~/.cursor/projects/*/agent-transcripts/**/*.jsonl` | JSONL | inferred |
| `github-copilot` | `~/.copilot/session-state/<id>/events.jsonl` | JSONL | inferred |
| `goose` | `~/.local/share/goose/sessions.db` | SQLite | inferred |

Every other agent in kasetto's roster is unmapped. That is "nobody has checked",
not "impossible" - most were simply not installed on the machine where this was
built. Adding one means finding its store, confirming tool calls are recoverable,
and checking the counts against a manual grep before marking it verified.

## Details Worth Knowing

**Claude Code.** Skill invocations appear as `"skill":"<name>"` inside the `Skill`
tool's input. MCP calls appear as tool names shaped `mcp__<server>__<tool>`. Both
are reliable. History can run to hundreds of megabytes, which is why the
per-file cache exists.

**OpenCode.** The `part` table holds one row per event with the payload in a
`data` text column of JSON; `json_extract(data,'$.type')='tool'` selects tool
calls and `$.tool` is the name. Migrated from per-file JSON to SQLite in v1.2.
Builtin tools (`edit`, `read`, `bash`) dominate, so a machine with OpenCode
installed but no MCP or skill use reports a working provider with zero findings -
which is correct, and deliberately renders differently from a missing provider.

**Codex.** `CODEX_HOME` relocates the whole tree, so never hardcode `~/.codex`.
The project is explicitly tolerant of schema drift and field names vary between
client versions, which is another argument for pattern matching over parsing.

**Cursor.** Splits storage. The agent transcripts are readable JSONL. The IDE
chat lives in protobuf blobs inside `state.vscdb` under `workspaceStorage` and is
not decoded by anything public - so Cursor coverage is real but partial, and a
user who works mainly in the IDE chat will look less active than they are.

**Goose.** Moved from per-session JSONL to a `sessions.db` SQLite database in
1.10.0. Legacy JSONL files may still sit in `~/.local/share/goose/sessions/`
untouched. The SQLite schema is not pinned here; the reader discovers plausible
(timestamp, payload) column pairs instead of guessing table names.

**Copilot CLI.** Restructured from flat JSONL to per-session directories in
v1.0.11, so the glob has to cover both shapes.

## Why This Churns

Three of the six formats above changed shape recently - Goose, OpenCode and
Copilot CLI all migrated within a short window. These are internal formats with
no compatibility contract and no deprecation notice, so treat any provider here
as a moving target.

That churn is the reason this lives in a skill rather than in the kasetto binary.
Fixing a drifted format is editing a file that reaches users on their next `kst
sync`, instead of cutting a release and waiting for everyone to upgrade.

It is also why the readers match tool-name strings rather than parsing records.
Every one of those three migrations changed the record structure. None of them
changed what a tool call is named.

#!/usr/bin/env python3
"""Collect local usage telemetry for kasetto-installed skills and MCP servers.

Reads the installed inventory from `kst list --json`, scans each supported
agent's local session store for tool invocations, and emits a single JSON
document joining the two. Standard library only.

The scan is deliberately pattern-based rather than schema-based. Agent session
formats churn (Goose moved JSONL -> SQLite, OpenCode JSON -> SQLite, Copilot CLI
flat -> per-session dirs), but the tool-name strings inside them do not move.
Matching on `mcp__<server>__<tool>` and on known asset names survives a schema
change that would break a field-by-field parser.
"""

import argparse
import json
import os
import re
import sqlite3
import subprocess
import sys
import time
from datetime import date
from pathlib import Path

CACHE_VERSION = 2
STALE_DAYS = 60

# == Providers ==

# Each provider: where its session store lives and how to read it. `kind` picks
# the reader. Paths honour the agent's own env override where one exists.
def providers():
    home = Path.home()
    codex_home = Path(os.environ.get("CODEX_HOME", home / ".codex"))
    return [
        {
            "agent": "claude-code",
            "kind": "jsonl",
            "root": home / ".claude" / "projects",
            "glob": "**/*.jsonl",
            "verified": True,
        },
        {
            "agent": "codex",
            "kind": "jsonl",
            "root": codex_home / "sessions",
            "glob": "**/*.jsonl",
            "verified": False,
        },
        {
            "agent": "cursor",
            "kind": "jsonl",
            "root": home / ".cursor" / "projects",
            "glob": "**/agent-transcripts/**/*.jsonl",
            "verified": False,
        },
        {
            "agent": "github-copilot",
            "kind": "jsonl",
            "root": home / ".copilot" / "session-state",
            "glob": "**/*.jsonl",
            "verified": False,
        },
        {
            "agent": "opencode",
            "kind": "sqlite",
            "root": home / ".local" / "share" / "opencode" / "opencode.db",
            "sql": "SELECT time_created, data FROM part",
            "verified": True,
        },
        {
            "agent": "goose",
            "kind": "sqlite",
            "root": home / ".local" / "share" / "goose" / "sessions.db",
            "sql": None,  # table discovered at runtime; schema unverified
            "verified": False,
        },
    ]


# == Matching ==

# Anchored on a name-key position on purpose. Session logs also carry arrays of
# the tools *offered* each turn (`"mcp__logfire__alert_create","mcp__logfire__..."`),
# which repeat once per request. Matching the bare `mcp__x__y` string anywhere
# counted those as calls and inflated one server from 0 real invocations to
# 38,743. Any new provider must anchor the same way.
MCP_RE = re.compile(r'"name"\s*:\s*"mcp__([A-Za-z0-9_.-]+?)__([A-Za-z0-9_.-]+)"')
# ISO-8601 with optional fractional seconds and zone; the first one on a line is
# close enough to the event for a "last used" readout.
TS_RE = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?")


def skill_pattern(names):
    """Match an installed skill name only in a structured key position.

    A bare name like `init` appears constantly in prose, so anchoring on
    `"skill": "init"` / `"tool": "init"` is what separates an invocation from a
    mention. The generic `"name"` key is deliberately excluded: it is the most
    common key in these logs and matches file names, agent names and tool
    definitions. Over-matching marks a dead skill alive, and a skill wrongly
    shown as used is the one error that costs the user nothing to keep and
    everything to trust.
    """
    if not names:
        return None
    alt = "|".join(sorted((re.escape(n) for n in names), key=len, reverse=True))
    return re.compile(r'"(?:skill|skill_name|tool)"\s*:\s*"(' + alt + r')"')


def scan_text(text, skill_re, acc, ts_hint):
    """Accumulate matches from one chunk of raw session text.

    Per-day buckets are kept alongside the totals so the report can show when an
    asset was used, not just how often. A total alone cannot distinguish a tool
    used heavily last year from one used steadily this week.
    """
    ts = None
    m = TS_RE.search(text)
    if m:
        ts = m.group(0)
    ts = ts or ts_hint
    day = ts[:10] if ts else None

    for server, tool in MCP_RE.findall(text):
        e = acc["mcp_servers"].setdefault(
            server, {"count": 0, "tools": {}, "last": None, "days": {}})
        e["count"] += 1
        e["tools"][tool] = e["tools"].get(tool, 0) + 1
        if day:
            e["days"][day] = e["days"].get(day, 0) + 1
        if ts and (e["last"] is None or ts > e["last"]):
            e["last"] = ts

    if skill_re:
        for name in skill_re.findall(text):
            e = acc["skills"].setdefault(name, {"count": 0, "last": None, "days": {}})
            e["count"] += 1
            if day:
                e["days"][day] = e["days"].get(day, 0) + 1
            if ts and (e["last"] is None or ts > e["last"]):
                e["last"] = ts


def empty_acc():
    return {"skills": {}, "mcp_servers": {}}


def attribute(acc, project):
    """Stamp one file's results with its session and project.

    One JSONL file is one session, so presence in the file means one session -
    which separates a skill run 16 times in a single burst from one run once a
    week for 16 weeks. The project label comes from the directory name the agent
    encodes the working directory into.
    """
    for group in ("skills", "mcp_servers"):
        for e in acc[group].values():
            e["sessions"] = 1
            if project:
                e["projects"] = {project: e["count"]}


def merge_days(into, other):
    for d, n in other.items():
        into[d] = into.get(d, 0) + n


def merge_acc(into, other):
    for name, e in other["skills"].items():
        t = into["skills"].setdefault(
            name, {"count": 0, "last": None, "days": {}, "sessions": 0, "projects": {}})
        t["count"] += e["count"]
        t["sessions"] = t.get("sessions", 0) + e.get("sessions", 0)
        merge_days(t["projects"], e.get("projects", {}))
        merge_days(t["days"], e.get("days", {}))
        if e["last"] and (t["last"] is None or e["last"] > t["last"]):
            t["last"] = e["last"]
    for server, e in other["mcp_servers"].items():
        t = into["mcp_servers"].setdefault(
            server, {"count": 0, "tools": {}, "last": None, "days": {},
                     "sessions": 0, "projects": {}})
        t["count"] += e["count"]
        t["sessions"] = t.get("sessions", 0) + e.get("sessions", 0)
        merge_days(t["projects"], e.get("projects", {}))
        for tool, n in e["tools"].items():
            t["tools"][tool] = t["tools"].get(tool, 0) + n
        merge_days(t["days"], e.get("days", {}))
        if e["last"] and (t["last"] is None or e["last"] > t["last"]):
            t["last"] = e["last"]


# == Readers ==

def project_label(path):
    """Readable project name from the directory an agent stores sessions under.

    Claude Code encodes the working directory into the folder name by replacing
    every non-alphanumeric run with `-`, which is lossy: a real hyphen and a path
    separator become the same character, so `pivoshenko-wallpapers` cannot be
    told from a `pivoshenko/wallpapers` directory. Rather than reconstruct a path
    and get it confidently wrong, strip the encoded home prefix and show what is
    left verbatim. It groups correctly, which is what the rollup needs.
    """
    name = path.parent.name
    if not name.startswith("-"):
        return name
    home = re.sub(r"[^A-Za-z0-9]+", "-", str(Path.home()))
    if name.startswith(home):
        name = name[len(home):]
    return name.strip("-") or "-"


def read_jsonl_tree(root, pattern, skill_re, cache):
    """Scan a tree of JSONL session files, reusing cached per-file results.

    Closed sessions never change, so caching on (size, mtime) means a rerun only
    touches the handful of files the agent actually appended to.
    """
    acc, files, events = empty_acc(), 0, 0
    for path in sorted(root.glob(pattern)):
        if not path.is_file():
            continue
        files += 1
        try:
            st = path.stat()
        except OSError:
            continue
        key = str(path)
        hit = cache.get(key)
        if hit and hit.get("size") == st.st_size and hit.get("mtime") == int(st.st_mtime):
            merge_acc(acc, hit["result"])
            events += hit.get("events", 0)
            continue

        one = empty_acc()
        ts_hint = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(st.st_mtime))
        n = 0
        try:
            with path.open("r", encoding="utf-8", errors="replace") as fh:
                for line in fh:
                    if "mcp__" not in line and '"skill' not in line and '"tool"' not in line and '"name"' not in line:
                        continue
                    scan_text(line, skill_re, one, ts_hint)
                    n += 1
        except OSError:
            continue
        attribute(one, project_label(path))
        cache[key] = {
            "size": st.st_size,
            "mtime": int(st.st_mtime),
            "events": n,
            "result": one,
        }
        merge_acc(acc, one)
        events += n
    return acc, files, events


def read_sqlite(path, sql, skill_re):
    """Scan a SQLite session store. Rows are opaque JSON blobs; same patterns apply."""
    acc, rows = empty_acc(), 0
    try:
        con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    except sqlite3.Error:
        return acc, 0, 0

    try:
        queries = [sql] if sql else discover_sqlite_queries(con)
        for q in queries:
            try:
                for ts, data in con.execute(q):
                    if not data:
                        continue
                    text = data if isinstance(data, str) else str(data)
                    hint = None
                    if isinstance(ts, int) and ts > 0:
                        secs = ts / 1000 if ts > 10_000_000_000 else ts
                        hint = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(secs))
                    scan_text(text, skill_re, acc, hint)
                    rows += 1
            except sqlite3.Error:
                continue
    finally:
        con.close()
    return acc, 1, rows


def discover_sqlite_queries(con):
    """Find (timestamp, blob) column pairs in an unknown schema.

    Goose's post-1.10 schema is not pinned here on purpose. Rather than guess
    table names that may be wrong, look for any text column wide enough to hold
    a serialized message and pair it with a plausible time column.
    """
    out = []
    try:
        tables = [r[0] for r in con.execute(
            "SELECT name FROM sqlite_master WHERE type='table'")]
    except sqlite3.Error:
        return out
    for t in tables:
        try:
            cols = [(r[1], (r[2] or "").upper()) for r in con.execute(f'PRAGMA table_info("{t}")')]
        except sqlite3.Error:
            continue
        names = [c for c, _ in cols]
        blob = next((c for c in names if c.lower() in
                     ("data", "content", "body", "message", "payload", "json")), None)
        if not blob:
            continue
        tcol = next((c for c in names if "time" in c.lower() or c.lower() in ("created", "ts")), None)
        out.append(f'SELECT {tcol or "NULL"}, "{blob}" FROM "{t}"')
    return out


# == Inventory ==

def installed_inventory():
    """Ask kasetto what it installed. `kst list --json` is the contract."""
    for exe in ("kst", "kasetto"):
        try:
            proc = subprocess.run([exe, "list", "--json"], capture_output=True, text=True, timeout=60)
        except (FileNotFoundError, subprocess.SubprocessError):
            continue
        if proc.returncode == 0 and proc.stdout.strip():
            try:
                return json.loads(proc.stdout), None
            except json.JSONDecodeError as exc:
                return None, f"{exe} list --json returned unparseable output: {exc}"
    return None, "kasetto CLI not found on PATH (tried `kst`, `kasetto`)"


FRESH_DAYS = 7
AGE_RE = re.compile(r"(\d+)\s*([smhdwy])")


def days_since(last):
    """Whole days since an ISO timestamp, or None when never used."""
    if not last:
        return None
    try:
        return max(0, (date.today() - date.fromisoformat(last[:10])).days)
    except ValueError:
        return None


def is_stale(last, total):
    """A skill used heavily long ago still reads as alive without this.

    Live/idle is a binary that hides the middle: something invoked twenty times
    a year ago and never since is not in use, but it is also not unused, and it
    escapes the idle list forever. Anything past STALE_DAYS gets its own tier.
    """
    if not total:
        return False
    n = days_since(last)
    return n is not None and n > STALE_DAYS


def is_fresh(age):
    """True when an asset was installed too recently to judge as unused.

    `kst list` renders age as "12h ago" / "3d ago" / "35d ago". Something
    installed yesterday has had no chance to be invoked, and listing it as a
    removal candidate is how a report loses the user's trust.
    """
    if not age:
        return False
    m = AGE_RE.search(age)
    if not m:
        return False
    n, unit = int(m.group(1)), m.group(2)
    days = {"s": 0, "m": 0, "h": n / 24, "d": n, "w": n * 7, "y": n * 365}[unit]
    return days <= FRESH_DAYS


def skill_inventory(entries):
    """Normalize skill rows from `kst list --json`.

    A row carries both a display `name` ("/analyze - Answer Data Questions") and
    the invocable slug `skill` ("analyze"). Matching must use the slug; the
    display name never appears in a session log.
    """
    out = []
    for e in entries or []:
        if isinstance(e, str):
            out.append({"slug": e, "label": e, "source": "", "scope": "", "age": ""})
            continue
        slug = e.get("skill") or e.get("name")
        if not slug:
            continue
        out.append({
            "slug": slug,
            "label": e.get("name") or slug,
            "source": e.get("source", ""),
            "scope": e.get("scope", ""),
            "age": e.get("updated_ago", ""),
        })
    return out


MCP_LOCK_RE = re.compile(
    r"^\s{2}mcp::(?P<src>[^:\n]*(?:::[^:\n]*)*?)::(?P<pack>[^:\n]+):\s*$"
    r"(?P<body>(?:\n\s{4}\S.*)*)", re.MULTILINE)


def lock_mcp_servers():
    """Map MCP pack -> server names by reading `kasetto.lock`.

    `kst list --json` reports the pack but not the servers it merged, and the
    telemetry only ever sees server names (`mcp__<server>__<tool>`). The lock is
    the only place that mapping exists. It is YAML and the standard library has
    no YAML parser, but these entries are machine-generated with a fixed shape,
    so a narrow regex over just the `mcp::` blocks is safer than taking on a
    dependency. If the shape ever changes this returns nothing and callers fall
    back to assuming pack name == server name.
    """
    candidates = [
        Path.cwd() / "kasetto.lock",
        Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))
        / "kasetto" / "kasetto.lock",
    ]
    mapping = {}
    for lock in candidates:
        if not lock.exists():
            continue
        try:
            text = lock.read_text(encoding="utf-8", errors="replace")
        except OSError:
            continue
        for m in MCP_LOCK_RE.finditer(text):
            body = m.group("body")
            dest = re.search(r"^\s{4}destination:\s*(.+)$", body, re.MULTILINE)
            name = re.search(r"^\s{4}name:\s*(.+)$", body, re.MULTILINE)
            pack = (name.group(1).strip() if name else m.group("pack")).strip()
            pack = pack[:-5] if pack.endswith(".json") else pack
            servers = [s.strip() for s in dest.group(1).split(",")] if dest else []
            if servers:
                mapping.setdefault(pack, []).extend(s for s in servers if s)
    return mapping


def mcp_inventory(entries, lock_map):
    """Normalize MCP rows, resolving each pack to the servers it installed."""
    out = []
    for e in entries or []:
        if isinstance(e, str):
            pack = e[:-5] if e.endswith(".json") else e
            out.append({"pack": pack, "servers": lock_map.get(pack, [pack]),
                        "source": "", "scope": ""})
            continue
        pack = e.get("name") or e.get("id") or ""
        pack = pack[:-5] if pack.endswith(".json") else pack
        if not pack:
            continue
        out.append({
            "pack": pack,
            "servers": lock_map.get(pack, [pack]),
            "source": e.get("source", ""),
            "scope": e.get("scope", ""),
        })
    return out


# == Insights ==

# Severity drives colour in the dashboard, so the vocabulary is deliberately
# narrow. "bad" is reserved for concrete waste the user can delete; an unused
# skill is dormant, not a defect, and colouring it red would spend the strongest
# signal on the least urgent finding.
SEV_BAD, SEV_WARN, SEV_INFO, SEV_GOOD = "bad", "warn", "info", "good"


def insights(skills, mcps, by_source, unmanaged, coverage, activity, idle, fresh,
             by_project=None):
    """Derive the handful of statements worth acting on.

    Computed here rather than left to the reader because these are arithmetic
    over the whole report, and a number stated once beats a chart the reader has
    to integrate by eye.
    """
    out = []

    # Whole source repos with nothing in use: the cheapest large cleanup, since
    # each is a single `source:` block rather than N individual removals.
    dead_src = {s: v for s, v in by_source.items()
                if v["used"] == 0 and v["fresh"] == 0 and (v["idle"] or 0) > 0}
    if dead_src:
        n = sum(v["idle"] for v in dead_src.values())
        out.append({
            "severity": SEV_BAD,
            "title": f"{len(dead_src)} source repos have no asset in use",
            "detail": f"They account for {n} idle assets. Each is one `source:` block, "
                      f"so removing them is {len(dead_src)} edits rather than {n}.",
            "items": [short_src(s) for s in sorted(
                dead_src, key=lambda x: -dead_src[x]["idle"])][:6],
        })

    # Idle MCP packs carry a standing cost that idle skills do not.
    mcp_idle = [k for k, v in mcps.items() if not v["total"]]
    if mcp_idle:
        out.append({
            "severity": SEV_WARN,
            "title": f"{len(mcp_idle)} MCP packs load into context but are never called",
            "detail": "MCP tool definitions are sent on every request whether or not the "
                      "server is used, so an idle pack is a recurring cost, not just clutter.",
            "items": sorted(mcp_idle),
        })

    # Coverage gaps invalidate every "idle" claim for the agents involved.
    absent = [c["agent"] for c in coverage if c["status"] != "ok"]
    unver = [c["agent"] for c in coverage if c["status"] == "ok"
             and not c.get("verified_format")]
    if absent:
        out.append({
            "severity": SEV_WARN,
            "title": f"{len(absent)} of {len(coverage)} known agents left no readable history",
            "detail": "Assets driven from these agents appear idle here and are not. "
                      "Treat the idle list as a shortlist to review, not a cut list.",
            "items": sorted(absent),
        })
    if unver:
        out.append({
            "severity": SEV_WARN,
            "title": f"{len(unver)} providers parsed an unverified format",
            "detail": "Their layout has not been confirmed against a real sample, so a zero "
                      "from them is weak evidence rather than proof.",
            "items": sorted(unver),
        })

    # Used, but long ago - invisible in a live/idle split.
    stale = [k for k, v in skills.items() if v.get("stale")]
    if stale:
        out.append({
            "severity": SEV_WARN,
            "title": f"{len(stale)} skills were used once but not in the last {STALE_DAYS} days",
            "detail": "They count as live on every other view, which is how a tool you "
                      "stopped reaching for stays in the config indefinitely.",
            "items": sorted(stale, key=lambda k: -(skills[k].get("days_since") or 0))[:6],
        })

    # One-burst assets: a high total from a single session is not a habit.
    burst = [k for k, v in skills.items()
             if v["total"] >= 5 and v.get("sessions", 0) == 1]
    if burst:
        out.append({
            "severity": SEV_INFO,
            "title": f"{len(burst)} skills look busy but ran in a single session",
            "detail": "A high call count from one sitting is a trial, not a habit. "
                      "Session count separates the two.",
            "items": sorted(burst),
        })

    # Where the toolkit actually gets exercised.
    if by_project:
        top = sorted(by_project.items(), key=lambda x: -x[1]["calls"])
        share = round(100 * top[0][1]["calls"] / max(1, sum(v["calls"] for v in by_project.values())))
        out.append({
            "severity": SEV_INFO,
            "title": f"{len(by_project)} projects used these assets, and the top one is "
                     f"{share}% of all calls",
            "detail": "Assets that only ever fire in one project may belong in that "
                      "project's kasetto.yaml rather than the global config.",
            "items": [t[0] for t in top[:4]],
        })

    # Concentration: how much of the toolkit is actually carrying the work.
    used = {k: v["total"] for k, v in skills.items() if v["total"]}
    total_calls = sum(used.values())
    if used and total_calls:
        top3 = sorted(used.values(), reverse=True)[:3]
        pct = round(100 * sum(top3) / total_calls)
        out.append({
            "severity": SEV_INFO,
            "title": f"Your top 3 skills are {pct}% of all skill invocations",
            "detail": f"{len(used)} of {len(skills)} installed skills have ever run. "
                      f"Usage concentrates hard, which is normal - it just means the long "
                      f"tail is cheaper to review than it looks.",
            "items": [k for k, _ in sorted(used.items(), key=lambda x: -x[1])[:3]],
        })

    # Unmanaged assets in heavy use are the inverse finding: not what to remove,
    # but what is missing from the config.
    um = sorted(unmanaged["mcps"].items(), key=lambda x: -x[1])[:1]
    if um and used and um[0][1] > max(used.values() or [0]):
        name, n = um[0]
        out.append({
            "severity": SEV_INFO,
            "title": f"`{name}` is your most-used asset and kasetto does not manage it",
            "detail": f"{n} calls, more than any managed skill. Adding it to kasetto.yaml "
                      f"would sync it across machines and agents like the rest.",
            "items": [],
        })

    if fresh:
        out.append({
            "severity": SEV_INFO,
            "title": f"{len(fresh)} skills were installed in the last 7 days and not yet run",
            "detail": "Held out of the cut list deliberately - a zero here means no chance "
                      "to be used, not a verdict.",
            "items": sorted(fresh)[:6],
        })

    # A long silence is worth naming; a sparse chart is easy to misread as dense.
    days = sorted(set(activity["skills"]) | set(activity["mcps"]))
    if days:
        try:
            last = date.fromisoformat(days[-1])
            quiet = (date.today() - last).days
            if quiet >= 14:
                out.append({
                    "severity": SEV_INFO,
                    "title": f"No recorded invocation in {quiet} days",
                    "detail": "Either the toolkit is idle or sessions are being written "
                              "somewhere this run did not read.",
                    "items": [],
                })
        except ValueError:
            pass

    if not idle and not mcp_idle:
        out.append({
            "severity": SEV_GOOD,
            "title": "Everything installed has been used",
            "detail": "No idle assets in the agents this run could read.",
            "items": [],
        })

    order = {SEV_BAD: 0, SEV_WARN: 1, SEV_INFO: 2, SEV_GOOD: 3}
    out.sort(key=lambda x: order.get(x["severity"], 9))
    return out


def short_src(src):
    if not src or src == "-":
        return "-"
    s = str(src).rstrip("/")
    return "/".join(s.split("/")[-2:]) if "/" in s else s


# == Main ==

def main():
    ap = argparse.ArgumentParser(description="Collect kasetto asset usage telemetry.")
    ap.add_argument("--out", default="usage.json", help="where to write the report JSON")
    ap.add_argument("--cache", default=None, help="scan cache path (default: XDG cache)")
    ap.add_argument("--quiet", action="store_true")
    args = ap.parse_args()

    cache_path = Path(args.cache) if args.cache else (
        Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache"))
        / "kasetto" / "usage-scan.json")
    cache = {}
    if cache_path.exists():
        try:
            blob = json.loads(cache_path.read_text())
            if blob.get("version") == CACHE_VERSION:
                cache = blob.get("files", {})
        except (OSError, json.JSONDecodeError):
            cache = {}

    inventory, inv_err = installed_inventory()
    if inventory is None:
        print(f"error: {inv_err}", file=sys.stderr)
        return 2

    skills = skill_inventory(inventory.get("skills"))
    mcps = mcp_inventory(inventory.get("mcps"), lock_mcp_servers())
    skill_re = skill_pattern([s["slug"] for s in skills])

    # server name -> owning pack, so per-server counts roll up to the pack
    pack_of = {}
    for entry in mcps:
        for server in entry["servers"]:
            pack_of[server] = entry["pack"]

    coverage, totals = [], {}
    for prov in providers():
        root = prov["root"]
        if not root.exists():
            coverage.append({
                "agent": prov["agent"], "status": "absent",
                "reason": f"no session store at {root}",
                "verified_format": prov["verified"],
            })
            continue
        if prov["kind"] == "jsonl":
            acc, files, events = read_jsonl_tree(root, prov["glob"], skill_re, cache)
        else:
            acc, files, events = read_sqlite(root, prov.get("sql"), skill_re)
        totals[prov["agent"]] = acc
        coverage.append({
            "agent": prov["agent"], "status": "ok", "root": str(root),
            "sources": files, "events": events,
            "verified_format": prov["verified"],
        })

    # == Join ==
    covered = {c["agent"] for c in coverage if c["status"] == "ok"}

    skill_usage = {}
    for entry in skills:
        slug = entry["slug"]
        by_agent, days, projects, last, total, sessions = {}, {}, {}, None, 0, 0
        for agent, acc in totals.items():
            e = acc["skills"].get(slug)
            if not e:
                continue
            by_agent[agent] = e["count"]
            total += e["count"]
            sessions += e.get("sessions", 0)
            merge_days(days, e.get("days", {}))
            merge_days(projects, e.get("projects", {}))
            if e["last"] and (last is None or e["last"] > last):
                last = e["last"]
        skill_usage[slug] = {
            "total": total, "last_used": last, "by_agent": by_agent, "days": days,
            "sessions": sessions, "projects": projects,
            "days_since": days_since(last), "stale": is_stale(last, total),
            "label": entry["label"], "source": entry["source"],
            "scope": entry["scope"], "age": entry["age"],
            "fresh": is_fresh(entry["age"]),
        }

    mcp_usage = {}
    for entry in mcps:
        by_agent, per_server, per_tool, days, projects = {}, {}, {}, {}, {}
        last, total, sessions = None, 0, 0
        for server in entry["servers"]:
            for agent, acc in totals.items():
                e = acc["mcp_servers"].get(server)
                if not e:
                    continue
                by_agent[agent] = by_agent.get(agent, 0) + e["count"]
                per_server[server] = per_server.get(server, 0) + e["count"]
                total += e["count"]
                merge_days(days, e.get("days", {}))
                merge_days(projects, e.get("projects", {}))
                sessions += e.get("sessions", 0)
                for tool, n in e.get("tools", {}).items():
                    per_tool[tool] = per_tool.get(tool, 0) + n
                if e["last"] and (last is None or e["last"] > last):
                    last = e["last"]
            per_server.setdefault(server, 0)
        mcp_usage[entry["pack"]] = {
            "total": total, "last_used": last, "days": days,
            "sessions": sessions, "projects": projects,
            "days_since": days_since(last), "stale": is_stale(last, total),
            "by_agent": by_agent, "by_server": per_server, "by_tool": per_tool,
            "source": entry["source"], "scope": entry["scope"],
        }

    # Assets seen in the logs that kasetto did not install.
    managed_servers = set(pack_of)
    unmanaged = {"skills": {}, "mcps": {}}
    for acc in totals.values():
        for name, e in acc["skills"].items():
            if name not in skill_usage:
                unmanaged["skills"][name] = unmanaged["skills"].get(name, 0) + e["count"]
        for server, e in acc["mcp_servers"].items():
            if server not in managed_servers:
                unmanaged["mcps"][server] = unmanaged["mcps"].get(server, 0) + e["count"]

    # Daily activity across everything managed, for the timeline.
    activity = {"skills": {}, "mcps": {}}
    for v in skill_usage.values():
        merge_days(activity["skills"], v["days"])
    for v in mcp_usage.values():
        merge_days(activity["mcps"], v["days"])

    # Whole source repos often go idle together, and a dead source is one block
    # to delete from kasetto.yaml rather than N rows to prune individually.
    by_source = {}
    for slug, v in skill_usage.items():
        s = by_source.setdefault(v["source"] or "-", {
            "used": 0, "idle": 0, "fresh": 0, "calls": 0, "assets": []})
        s["calls"] += v["total"]
        s["assets"].append(slug)
        if v["total"]:
            s["used"] += 1
        elif v["fresh"]:
            s["fresh"] += 1
        else:
            s["idle"] += 1
    for pack, v in mcp_usage.items():
        s = by_source.setdefault(v["source"] or "-", {
            "used": 0, "idle": 0, "fresh": 0, "calls": 0, "assets": []})
        s["calls"] += v["total"]
        s["assets"].append(pack)
        s["used" if v["total"] else "idle"] += 1

    # Which projects the toolkit is actually exercised in.
    by_project = {}
    for group, kind in ((skill_usage, "skills"), (mcp_usage, "mcps")):
        for name, v in group.items():
            for proj, cnt in v.get("projects", {}).items():
                e = by_project.setdefault(proj, {"calls": 0, "skills": 0, "mcps": 0,
                                                 "assets": []})
                e["calls"] += cnt
                e[kind] += 1
                e["assets"].append(name)

    idle_skills = [k for k, v in skill_usage.items() if not v["total"] and not v["fresh"]]
    fresh_skills = [k for k, v in skill_usage.items() if not v["total"] and v["fresh"]]

    findings = insights(skill_usage, mcp_usage, by_source, unmanaged, coverage,
                        activity, idle_skills, fresh_skills, by_project)

    report = {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "insights": findings,
        "coverage": coverage,
        "agents_covered": sorted(covered),
        "counts": {
            "skills_installed": len(skill_usage),
            "skills_used": sum(1 for v in skill_usage.values() if v["total"] > 0),
            "skills_idle": len(idle_skills),
            "skills_fresh": len(fresh_skills),
            "skills_stale": sum(1 for v in skill_usage.values() if v.get("stale")),
            "projects": len(by_project),
            "sessions": sum(v.get("sessions", 0) for v in skill_usage.values())
                        + sum(v.get("sessions", 0) for v in mcp_usage.values()),
            "mcps_installed": len(mcp_usage),
            "mcps_used": sum(1 for v in mcp_usage.values() if v["total"] > 0),
            "mcps_idle": sum(1 for v in mcp_usage.values() if not v["total"]),
            "total_calls": sum(v["total"] for v in skill_usage.values())
                           + sum(v["total"] for v in mcp_usage.values()),
        },
        "activity": activity,
        "by_source": by_source,
        "by_project": by_project,
        "skills": skill_usage,
        "mcps": mcp_usage,
        "unmanaged": unmanaged,
    }

    Path(args.out).write_text(json.dumps(report, indent=2) + "\n")
    try:
        cache_path.parent.mkdir(parents=True, exist_ok=True)
        cache_path.write_text(json.dumps({"version": CACHE_VERSION, "files": cache}))
    except OSError:
        pass

    if not args.quiet:
        c = report["counts"]
        print(f"wrote {args.out}")
        print(f"  agents covered : {', '.join(sorted(covered)) or 'none'}")
        print(f"  skills         : {c['skills_used']}/{c['skills_installed']} used")
        print(f"  mcps           : {c['mcps_used']}/{c['mcps_installed']} used")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Render usage.json as a self-contained, kasetto-branded HTML dashboard.

Palette and roles come from `src/colors.rs`, which the site also derives from,
so the dashboard reads as the same tool as the CLI. The role semantics carry
over directly: SUCCESS marks what is alive, ERROR marks removal candidates,
ATTENTION marks coverage gaps and freshly installed assets, INFO labels sources
and unmanaged finds, BRAND violet is reserved for the wordmark, and INFRA draws
structure only - never content.

Layout is a 12-column panel grid with a sticky table-of-contents rail, rather
than a stack of sections. Panels render at full height - nothing scrolls inside
a panel - so the page is complete when printed or saved, and the rail carries
navigation instead.

Charts are hand-rolled inline SVG. No chart library, no CDN, no webfont fetch:
the output is one file that opens offline and can be mailed as-is.
"""

import argparse
import html
import json
import re
import sys
from datetime import date, timedelta
from pathlib import Path

# == Palette (src/colors.rs is the source of truth) ==
C = {
    "crust": "#151514", "base": "#1f1f1e", "mantle": "#1a1a19",
    "s0": "#262625", "s1": "#2e2e2c", "s2": "#373634",
    "text": "#e4e2de", "sub": "#b8b3a8", "secondary": "#a8a195", "infra": "#6e6759",
    "attention": "#e8a94d", "success": "#84c578", "error": "#e87e6c",
    "info": "#6cbfd3", "brand": "#b6a6ef",
}

WORDMARK = r"""██╗  ██╗ █████╗ ███████╗███████╗████████╗████████╗ ██████╗
██║ ██╔╝██╔══██╗██╔════╝██╔════╝╚══██╔══╝╚══██╔══╝██╔═══██╗
█████╔╝ ███████║███████╗█████╗     ██║      ██║   ██║   ██║
██╔═██╗ ██╔══██║╚════██║██╔══╝     ██║      ██║   ██║   ██║
██║  ██╗██║  ██║███████║███████╗   ██║      ██║   ╚██████╔╝
╚═╝  ╚═╝╚═╝  ╚═╝╚══════╝╚══════╝   ╚═╝      ╚═╝    ╚═════╝"""

CSS = """
*{box-sizing:border-box}
html,body{margin:0;padding:0}
body{background:%(crust)s;color:%(text)s;
  font-family:'JetBrains Mono',Menlo,Consolas,'Noto Sans Mono',monospace;
  font-size:12.5px;line-height:1.55;-webkit-font-smoothing:antialiased}
.shell{max-width:1440px;margin:0 auto;padding-left:20px;padding-right:20px}
code{background:%(s0)s;border:1px solid %(s1)s;border-radius:3px;padding:0 4px;font-size:11.5px}

/* == Header == */
header.site{background:%(mantle)s;border-bottom:1px solid %(s2)s;padding:16px 0}
.hrow{display:flex;align-items:center;gap:22px;flex-wrap:wrap}
.mark{border:1px solid %(brand)s;border-radius:3px;padding:8px 11px;flex:0 0 auto}
.mark pre{margin:0;color:%(brand)s;font-size:6px;line-height:1.1;font-weight:700}
.htitle{flex:1 1 220px;min-width:180px}
.htitle .t{font-size:16px;font-weight:700;letter-spacing:-.01em;line-height:1.3}
.hmeta{flex:0 0 auto;text-align:right;color:%(infra)s;font-size:11px;line-height:1.8}
.hmeta b{color:%(sub)s;font-weight:400}

/* == Grid == */
main{padding-top:28px;padding-bottom:44px}
.layout{display:flex;gap:22px;align-items:flex-start}
.grid{display:grid;grid-template-columns:repeat(12,1fr);gap:12px;flex:1 1 auto;min-width:0}

/* == TOC == */
.toc{position:sticky;top:18px;flex:0 0 176px;width:176px}
.toc-h{color:%(infra)s;font-size:9.5px;letter-spacing:.12em;text-transform:uppercase;
  padding:0 0 7px;border-bottom:1px solid %(s1)s;margin-bottom:6px}
.toc a{display:block;color:%(secondary)s;text-decoration:none;font-size:11.5px;
  padding:3px 9px;border-left:2px solid transparent;line-height:1.45}
.toc a:hover{color:%(text)s;background:%(s0)s}
.toc a.on{color:%(text)s;border-left-color:%(attention)s}
@media(max-width:1080px){.toc{display:none}}
.c3{grid-column:span 3}.c4{grid-column:span 4}.c5{grid-column:span 5}
.c6{grid-column:span 6}.c7{grid-column:span 7}.c8{grid-column:span 8}
.c9{grid-column:span 9}.c12{grid-column:span 12}
@media(max-width:1080px){.c3,.c4,.c5,.c6,.c7{grid-column:span 6}
  .c8,.c9{grid-column:span 12}}
@media(max-width:680px){.grid>*{grid-column:span 12 !important}.mark{display:none}}

/* == Panel == */
.panel{background:%(base)s;border:1px solid %(s1)s;border-radius:4px;display:flex;
  flex-direction:column;overflow:hidden;min-width:0}
.phead{display:flex;align-items:baseline;justify-content:space-between;gap:10px;
  padding:9px 13px;border-bottom:1px solid %(s1)s;background:%(mantle)s;flex:0 0 auto}
.phead h2{margin:0;font-size:12px;font-weight:700;letter-spacing:0;color:%(text)s}
.phead .hint{color:%(infra)s;font-size:10.5px;text-align:right;white-space:nowrap}
.pbody{padding:13px;overflow:auto;flex:1 1 auto;min-height:0}
.pbody.flush{padding:0}
.mix-i{text-align:center;padding:16px 8px 14px}
.mix-i+.mix-i{border-top:1px solid %(s1)s}
.mix-l{color:%(sub)s;font-size:11.5px;margin-top:4px}

/* == KPI strip == */
.kstrip{display:grid;grid-template-columns:repeat(5,1fr);gap:12px}
@media(max-width:1080px){.kstrip{grid-template-columns:repeat(2,1fr)}}
@media(max-width:520px){.kstrip{grid-template-columns:1fr}}
.kpi{background:%(base)s;border:1px solid %(s1)s;border-radius:4px;padding:13px 15px}
.kpi .n{font-size:26px;font-weight:700;line-height:1.1;font-variant-numeric:tabular-nums}
.kpi .l{color:%(sub)s;font-size:11px;margin-top:3px}
.kpi .s{color:%(infra)s;font-size:10.5px}

/* == Tables == */
table{width:100%%;border-collapse:collapse}
th{text-align:left;font-weight:400;font-size:10.5px;letter-spacing:.02em;
  color:%(infra)s;padding:8px 12px;border-bottom:1px solid %(s1)s;white-space:nowrap}
td{padding:6px 12px;border-bottom:1px solid %(s0)s;vertical-align:middle}
tbody tr:last-child td{border-bottom:none}
tbody tr:hover td{background:%(s0)s}
td.num,th.num{text-align:right;font-variant-numeric:tabular-nums;white-space:nowrap}
.dim{color:%(secondary)s}.faint{color:%(infra)s}
.src{color:%(info)s;font-size:10.5px}
.nm{white-space:nowrap;overflow:hidden;text-overflow:ellipsis;max-width:210px;display:block}

/* == Bars == */
.track{background:%(s1)s;border-radius:2px;height:7px;width:100%%;min-width:60px;
  overflow:hidden;display:flex}
.track i{height:7px;display:block}
.legend{display:flex;gap:14px;flex-wrap:wrap;color:%(secondary)s;font-size:10.5px;
  margin-top:10px;justify-content:center}
.legend span{display:flex;align-items:center;gap:5px}
.sw{width:8px;height:8px;border-radius:2px;display:inline-block}

.tag{display:inline-block;border:1px solid;border-radius:3px;padding:0 5px;
  font-size:10px;line-height:16px;white-space:nowrap}
.ok{color:%(success)s;border-color:%(success)s}
.dead{color:%(error)s;border-color:%(error)s}
.warn{color:%(attention)s;border-color:%(attention)s}
.nfo{color:%(info)s;border-color:%(info)s}
.idle{color:%(secondary)s;border-color:%(s2)s}

.more{padding:8px 12px;color:%(infra)s;font-size:10.5px;border-top:1px solid %(s1)s}

/* == Insights == */
.ins{border-left:2px solid;border-bottom:1px solid %(s0)s;padding:11px 14px}
.ins:last-child{border-bottom:none}
.ins-h{display:flex;align-items:baseline;gap:9px;flex-wrap:wrap}
.ins-b{border:1px solid;border-radius:3px;padding:0 6px;font-size:9.5px;line-height:15px;
  letter-spacing:.05em;flex:0 0 auto}
.ins-t{font-weight:700;font-size:12.5px}
.ins-d{margin:4px 0 0;color:%(sub)s;font-size:11.5px}
.ins-i{margin-top:7px;line-height:1.9}

footer.site{border-top:1px solid %(s1)s;padding:20px 0 40px;color:%(infra)s;font-size:10.5px}
"""


def esc(s):
    return html.escape(str(s if s is not None else ""))


def short(src):
    if not src or src == "-":
        return "-"
    s = str(src).rstrip("/")
    return "/".join(s.split("/")[-2:]) if "/" in s else s


AGE_RE = re.compile(r"(\d+)\s*([smhdwy])")
AGE_UNIT = {"s": 0.0, "m": 0.0, "h": 1 / 24, "d": 1.0, "w": 7.0, "y": 365.0}


def age_rank(age):
    """Install age in days from `kst list`'s "12h ago" / "35d ago" rendering."""
    m = AGE_RE.search(age or "")
    return int(m.group(1)) * AGE_UNIT[m.group(2)] if m else 0.0


def day_only(ts):
    return ts.split("T")[0] if ts else "Never"


def slug(title):
    return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")


def panel(title, body, hint="", cls="c6", body_cls=""):
    h = f'<div class="hint">{hint}</div>' if hint else ""
    return (f'<section id="{slug(title)}" class="panel {cls}">'
            f'<div class="phead"><h2>{title}</h2>{h}</div>'
            f'<div class="pbody {body_cls}">{body}</div></section>')


# == Charts ==

def svg_timeline(skills, mcps, height=132):
    """Daily invocation volume, skills stacked over MCP calls.

    Empty days are drawn as gaps rather than skipped, because "nothing happened
    for two weeks" is the signal an evenly spaced series would hide.
    """
    width = 900
    days = sorted(set(skills) | set(mcps))
    if not days:
        return '<p class="faint" style="margin:0">No dated activity recorded.</p>'

    start, end = date.fromisoformat(days[0]), date.fromisoformat(days[-1])
    span = [(start + timedelta(days=i)).isoformat() for i in range((end - start).days + 1)]
    span = span[-180:]

    pad_l, pad_b, pad_t = 30, 18, 6
    pw, ph = width - pad_l - 6, height - pad_b - pad_t
    top = max([skills.get(d, 0) + mcps.get(d, 0) for d in span] + [1])
    step = pw / len(span)
    bw = max(1.5, step - 1.4)

    o = [f'<svg viewBox="0 0 {width} {height}" width="100%" height="{height}" '
         f'preserveAspectRatio="none" role="img" aria-label="daily invocations">']
    for frac in (0, 0.5, 1):
        y = pad_t + ph * (1 - frac)
        o.append(f'<line x1="{pad_l}" y1="{y:.1f}" x2="{width-6}" y2="{y:.1f}" '
                 f'stroke="{C["s1"]}" stroke-width="1"/>')
        o.append(f'<text x="{pad_l-6}" y="{y+3.5:.1f}" text-anchor="end" font-size="9" '
                 f'fill="{C["infra"]}" font-family="monospace">{round(top*frac)}</text>')
    for i, dd in enumerate(span):
        s, m = skills.get(dd, 0), mcps.get(dd, 0)
        if not (s or m):
            continue
        x = pad_l + i * step
        hm, hs = ph * m / top, ph * s / top
        if m:
            o.append(f'<rect x="{x:.1f}" y="{pad_t+ph-hm:.1f}" width="{bw:.1f}" height="{hm:.1f}" '
                     f'fill="{C["info"]}"><title>{dd}: {m} mcp</title></rect>')
        if s:
            o.append(f'<rect x="{x:.1f}" y="{pad_t+ph-hm-hs:.1f}" width="{bw:.1f}" height="{hs:.1f}" '
                     f'fill="{C["success"]}"><title>{dd}: {s} skill</title></rect>')
    o.append(f'<text x="{pad_l}" y="{height-4}" font-size="9" fill="{C["infra"]}" '
             f'font-family="monospace">{span[0]}</text>')
    o.append(f'<text x="{width-6}" y="{height-4}" text-anchor="end" font-size="9" '
             f'fill="{C["infra"]}" font-family="monospace">{span[-1]}</text>')
    o.append('</svg>')
    return "".join(o)


def svg_donut(segments, size=118, thickness=14):
    """Flat composition ring. segments = [(label, value, color)]."""
    total = sum(v for _, v, _ in segments) or 1
    r = (size - thickness) / 2
    circ = 2 * 3.141592653589793 * r
    cx = cy = size / 2
    o = [f'<svg viewBox="0 0 {size} {size}" width="{size}" height="{size}" role="img">',
         f'<circle cx="{cx}" cy="{cy}" r="{r:.2f}" fill="none" stroke="{C["s1"]}" '
         f'stroke-width="{thickness}"/>']
    off = 0.0
    for label, val, col in segments:
        if not val:
            continue
        frac = val / total
        o.append(f'<circle cx="{cx}" cy="{cy}" r="{r:.2f}" fill="none" stroke="{col}" '
                 f'stroke-width="{thickness}" stroke-dasharray="{circ*frac:.2f} {circ:.2f}" '
                 f'stroke-dashoffset="{-circ*off:.2f}" transform="rotate(-90 {cx} {cy})">'
                 f'<title>{esc(label)}: {val}</title></circle>')
        off += frac
    live = segments[0][1] if segments else 0
    o.append(f'<text x="{cx}" y="{cy}" text-anchor="middle" font-size="20" font-weight="700" '
             f'fill="{C["text"]}" font-family="monospace">{live}</text>')
    o.append(f'<text x="{cx}" y="{cy+13}" text-anchor="middle" font-size="8" '
             f'fill="{C["infra"]}" font-family="monospace">Of {total}</text>')
    o.append('</svg>')
    return "".join(o)


def stacked(parts):
    """Inline composition bar. parts = [(value, color, title)]."""
    total = sum(p[0] for p in parts) or 1
    cells = "".join(f'<i style="width:{100*v/total:.2f}%;background:{col}" title="{esc(t)}"></i>'
                    for v, col, t in parts if v)
    return f'<span class="track">{cells}</span>'


def legend(items):
    return ('<div class="legend">' + "".join(
        f'<span><i class="sw" style="background:{col}"></i>{esc(l)}</span>'
        for l, col in items) + '</div>')


def table(head, rows, sticky=True):
    th = "".join(f'<th class="{c}">{t}</th>' for t, c in head)
    thead = f'<thead><tr>{th}</tr></thead>' if sticky else f'<tr>{th}</tr>'
    return f'<table>{thead}<tbody>{"".join(rows)}</tbody></table>'


# == Page ==

def render(d):
    c = d["counts"]
    skills, mcps = d["skills"], d["mcps"]
    used = {k: v for k, v in skills.items() if v["total"]}
    fresh = {k: v for k, v in skills.items() if not v["total"] and v.get("fresh")}
    idle = {k: v for k, v in skills.items() if not v["total"] and not v.get("fresh")}
    mcp_idle = {k: v for k, v in mcps.items() if not v["total"]}

    o = []
    A = o.append
    A(f'<!doctype html><html lang="en"><head><meta charset="utf-8">'
      f'<meta name="viewport" content="width=device-width,initial-scale=1">'
      f'<meta name="color-scheme" content="dark"><meta name="theme-color" content="{C["crust"]}">'
      f'<title>Kasetto Usage</title><style>{CSS % C}</style></head><body>')

    A(f'<header class="site"><div class="shell hrow">'
      f'<div class="mark"><pre>{esc(WORDMARK)}</pre></div>'
      f'<div class="htitle"><div class="t">Kasetto Usage</div></div>'
      f'<div class="hmeta">Generated <b>{esc(d["generated_at"][:16].replace("T", " "))}</b></div>'
      f'</div></header><main class="shell"><div class="layout"><div class="grid">')

    # Coverage is not banner-ed here: `insights` already carries it as a Review
    # finding naming the unread agents, and repeating it above the fold was the
    # third telling on the same page.

    # == KPI strip ==
    A('<div id="overview" class="c12 kstrip">')
    for n, lab, sub, col in [
        (c.get("total_calls", 0), "Invocations", "Across all agents", C["text"]),
        (f'{c["skills_used"]}/{c["skills_installed"]}', "Skills Live", "Invoked at least once", C["success"]),
        (c.get("skills_idle", 0), "Skills Idle", "Older than 7d, never called",
         C["secondary"]),
        (c.get("skills_fresh", 0), "Skills New", "Newer than 7d, too soon to judge", C["info"]),
        (f'{c.get("mcps_idle", 0)}/{c["mcps_installed"]}', "MCP Packs Idle", "Standing context cost",
         C["attention"] if c.get("mcps_idle") else C["success"]),
    ]:
        A(f'<div class="kpi"><div class="n" style="color:{col}">{n}</div>'
          f'<div class="l">{lab}</div><div class="s">{sub}</div></div>')
    A('</div>')

    # == Insights: the read, stated before the charts that support it ==
    sev = {"bad": C["error"], "warn": C["attention"], "info": C["info"], "good": C["success"]}
    words = {"bad": "Waste", "warn": "Review", "info": "Note", "good": "Clear"}
    cards = []
    for ins in d.get("insights", []):
        col = sev.get(ins["severity"], C["secondary"])
        items = ("".join(f'<span class="tag" style="color:{col};border-color:{col}">'
                         f'{esc(i)}</span> ' for i in ins.get("items", []))
                 if ins.get("items") else "")
        cards.append(
            f'<div class="ins" style="border-left-color:{col}">'
            f'<div class="ins-h"><span class="ins-b" style="color:{col};border-color:{col}">'
            f'{words.get(ins["severity"], "Note")}</span>'
            f'<span class="ins-t">{esc(ins["title"])}</span></div>'
            f'<p class="ins-d">{esc(ins["detail"])}</p>'
            f'{f"<div class=ins-i>{items}</div>" if items else ""}</div>')
    if cards:
        A(panel("Insights", "".join(cards), hint="Computed, not guessed",
                cls="c12", body_cls="flush"))

    # == Activity ==
    # Full width: coverage lives in the banner and the insights, so the timeline
    # gets the whole row rather than sharing it with a table that repeats them.
    A(panel("Activity", svg_timeline(d["activity"]["skills"], d["activity"]["mcps"])
            + legend([("Skill invocations", C["success"]), ("MCP tool calls", C["info"])]),
            hint="Daily invocations", cls="c12"))

    # == Composition ==
    # Both rings in one panel, stacked, so skills and MCP packs are read as two
    # views of the same portfolio rather than two unrelated widgets.
    mix = (
        '<div class="mix-i">'
        + svg_donut([("live", len(used), C["success"]), ("new", len(fresh), C["info"]),
                     ("idle", len(idle), C["secondary"])])
        + '<div class="mix-l">Skills</div>'
        + legend([("Live", C["success"]), ("New", C["info"]), ("Idle", C["secondary"])])
        + '</div><div class="mix-i">'
        + svg_donut([("live", c["mcps_used"], C["success"]),
                     ("idle", c.get("mcps_idle", 0), C["attention"])])
        + '<div class="mix-l">MCP Packs</div>'
        + legend([("Live", C["success"]), ("Idle", C["attention"])])
        + '</div>')
    A(panel("Asset Mix", mix, hint="Skills and MCP packs", cls="c3", body_cls="flush"))

    # == Top skills ==
    if used:
        top = max(v["total"] for v in used.values())
        rows = []
        for k, v in sorted(used.items(), key=lambda x: -x[1]["total"]):
            agents = ", ".join(f"{a}&nbsp;{n}" for a, n in sorted(v["by_agent"].items()))
            # A stale asset reads as live on every other view; flag it where it shows.
            lastc = C["attention"] if v.get("stale") else C["secondary"]
            since = v.get("days_since")
            ago = f"{since}d ago" if since is not None else "Never"
            rows.append(
                f'<tr><td><span class="nm">{esc(k)}</span></td>'
                f'<td style="width:26%">{stacked([(v["total"], C["success"], "{} calls".format(v["total"])), (top - v["total"], C["s1"], "")])}</td>'
                f'<td class="num" style="color:{C["success"]}">{v["total"]}</td>'
                f'<td class="num dim">{v.get("sessions", 0)}</td>'
                f'<td style="color:{lastc}">{esc(ago)}</td>'
                f'<td class="faint">{agents}</td></tr>')
        body = table([("Skill", ""), ("Calls", ""), ("N", "num"), ("Sessions", "num"),
                      ("Last Used", ""), ("Agents", "")], rows)
    else:
        body = '<p class="faint" style="margin:0">No installed skill was invoked.</p>'
    A(panel("Skills in Use", body, hint=f"Live: {len(used)}", cls="c9", body_cls="flush"))

    # == By source ==
    rows = []
    for src, s in sorted(d["by_source"].items(), key=lambda x: (-x[1]["idle"], -x[1]["calls"])):
        allde = s["used"] == 0 and s["fresh"] == 0
        nm = (f'<span class="nm" style="color:{C["error"]}">{esc(short(src))}</span>'
              if allde else f'<span class="nm">{esc(short(src))}</span>')
        rows.append(
            f'<tr><td>{nm}</td><td style="width:30%">'
            f'{stacked([(s["used"], C["success"], "live"), (s["fresh"], C["info"], "new"), (s["idle"], C["secondary"], "idle")])}</td>'
            f'<td class="num" style="color:{C["success"]}">{s["used"] or ""}</td>'
            f'<td class="num" style="color:{C["info"]}">{s["fresh"] or ""}</td>'
            f'<td class="num dim">{s["idle"] or ""}</td>'
            f'<td class="num dim">{s["calls"]}</td></tr>')
    A(panel("By Source", table([("Source", ""), ("Composition", ""), ("Live", "num"),
                                ("New", "num"), ("Idle", "num"), ("Calls", "num")], rows),
            hint="A fully idle repo is one config block", cls="c6", body_cls="flush"))

    # == By project ==
    proj = d.get("by_project") or {}
    if proj:
        ptop = max(v["calls"] for v in proj.values())
        rows = []
        for name, v in sorted(proj.items(), key=lambda x: -x[1]["calls"]):
            rows.append(
                f'<tr><td><span class="nm">{esc(name)}</span></td>'
                f'<td style="width:30%">{stacked([(v["calls"], C["success"], str(v["calls"])), (ptop - v["calls"], C["s1"], "")])}</td>'
                f'<td class="num" style="color:{C["success"]}">{v["calls"]}</td>'
                f'<td class="num dim">{v["skills"]}</td>'
                f'<td class="num dim">{v["mcps"]}</td></tr>')
        A(panel("By Project", table([("Project", ""), ("Calls", ""), ("N", "num"),
                                     ("Skills", "num"), ("MCPs", "num")], rows),
                hint="Where the toolkit gets used", cls="c6", body_cls="flush"))

    # == MCP detail ==
    rows = []
    for k, v in sorted(mcps.items(), key=lambda x: -x[1]["total"]):
        col = C["success"] if v["total"] else C["attention"]
        tools = v.get("by_tool") or {}
        tl = (" ".join(f'<span class="tag ok">{esc(t)}&nbsp;{n}</span>'
                       for t, n in sorted(tools.items(), key=lambda x: -x[1])[:6])
              or '<span class="faint">None called</span>')
        rows.append(f'<tr><td><span class="nm">{esc(k)}</span></td>'
                    f'<td class="num" style="color:{col}">{v["total"]}</td>'
                    f'<td class="dim">{esc(day_only(v["last_used"]))}</td>'
                    f'<td>{tl}</td></tr>')
    A(panel("MCP Packs",
            table([("Pack", ""), ("Calls", "num"), ("Last", ""), ("Tools Used", "")], rows),
            hint="Idle packs cost context every request", cls="c6", body_cls="flush"))

    # == Cut candidates ==
    # Every candidate has zero calls, so usage cannot rank them. MCP packs go
    # first because they cost context on every request, then skills by longest
    # time installed - the ones that have had the most chance to be used.
    if idle or mcp_idle:
        ranked = ([(k, v, "MCP") for k, v in mcp_idle.items()]
                  + sorted(((k, v, "Skill") for k, v in idle.items()),
                           key=lambda x: -age_rank(x[1].get("age"))))
        total_cut = len(ranked)
        rows = []
        for k, v, kind in ranked[:5]:
            tag = "warn" if kind == "MCP" else "idle"
            rows.append(f'<tr><td><span class="nm">{esc(k)}</span></td>'
                        f'<td><span class="tag {tag}">{kind}</span></td>'
                        f'<td class="dim">{esc(v.get("age") or "-")}</td>'
                        f'<td class="src">{esc(short(v.get("source")))}</td></tr>')
        body = table([("Asset", ""), ("Kind", ""), ("Installed", ""), ("Source", "")], rows)
        # Never let a truncated list read as the whole list.
        if total_cut > 5:
            body += (f'<div class="more">{total_cut - 5} more not shown &middot; '
                     f'full list in usage.json</div>')
        A(panel("Cut Candidates", body,
                hint=f"Top 5 of {total_cut}", cls="c6", body_cls="flush"))
    else:
        A(panel("Cut Candidates", '<p class="faint" style="margin:0">Nothing installed is idle.</p>',
                cls="c6"))

    # == Unmanaged ==
    um_s, um_m = d["unmanaged"]["skills"], d["unmanaged"]["mcps"]
    if um_s or um_m:
        allrows = ([(k, "MCP", n) for k, n in um_m.items()]
                   + [(k, "Skill", n) for k, n in um_s.items()])
        top = max([n for _, _, n in allrows] + [1])
        rows = [f'<tr><td><span class="nm">{esc(k)}</span></td>'
                f'<td><span class="tag nfo">{kind}</span></td>'
                f'<td style="width:30%">{stacked([(n, C["info"], str(n)), (top - n, C["s1"], "")])}</td>'
                f'<td class="num dim">{n}</td></tr>'
                for k, kind, n in sorted(allrows, key=lambda x: -x[2])[:24]]
        A(panel("Not Managed by Kasetto",
                table([("Name", ""), ("Kind", ""), ("Calls", ""), ("N", "num")], rows),
                hint="Candidates to bring under kasetto.yaml", cls="c6",
                body_cls="flush"))

    # == TOC ==
    # Built by reading back what was actually emitted, so a panel that was
    # skipped for lack of data can never leave a dead link in the rail.
    emitted = re.findall(r'id="([^"]+)" class="panel[^"]*"><div class="phead"><h2>([^<]+)</h2>',
                         "".join(o))
    links = [('overview', 'Overview')] + emitted
    rail = "".join(f'<a href="#{i}">{esc(t)}</a>' for i, t in links)
    A(f'</div><aside class="toc"><div class="toc-h">Contents</div>{rail}</aside>'
      f'</div></main>')

    # Year comes from the report rather than a literal so the footer cannot go stale.
    year = (d.get("generated_at") or "")[:4] or ""
    A('<footer class="site"><div class="shell">'
      f'<a href="https://kasetto.dev">kasetto.dev</a> &middot; {esc(year)}'
      '</div></footer>')

    # Highlight the rail entry for whatever is on screen. Plain observer, no
    # scroll handler, so it costs nothing while reading.
    A('<script>'
      'var L=[].slice.call(document.querySelectorAll(".toc a")),'
      'M={};L.forEach(function(a){var e=document.getElementById(a.hash.slice(1));'
      'if(e)M[a.hash.slice(1)]=a;});'
      'var seen={};'
      'var io=new IntersectionObserver(function(es){'
      'es.forEach(function(e){seen[e.target.id]=e.isIntersecting;});'
      'var first=Object.keys(M).filter(function(k){return seen[k];})[0];'
      'L.forEach(function(a){a.classList.toggle("on",a.hash.slice(1)===first);});'
      '},{rootMargin:"-10% 0px -70% 0px"});'
      'Object.keys(M).forEach(function(k){io.observe(document.getElementById(k));});'
      '</script></body></html>')
    return "".join(o)


def main():
    ap = argparse.ArgumentParser(description="Render usage.json as an HTML dashboard.")
    ap.add_argument("usage_json", nargs="?", default="usage.json")
    ap.add_argument("--out", default="usage.html")
    args = ap.parse_args()
    try:
        data = json.loads(Path(args.usage_json).read_text())
    except (OSError, json.JSONDecodeError) as exc:
        print(f"error: cannot read {args.usage_json}: {exc}", file=sys.stderr)
        return 2
    Path(args.out).write_text(render(data))
    print(f"wrote {args.out}")
    return 0


if __name__ == "__main__":
    sys.exit(main())