ai-agent/backend/skills.py

"""The **skills catalog** — every ``.claude/skills/<name>/SKILL.md`` on disk,
joined with the usage mined from the Claude Code transcripts.

Two halves, deliberately kept apart:

* :func:`catalog` is the *disk* half — a static scan of the skill dirs the
  container can see (the repo's own ``.claude/skills`` plus each project's),
  reading each ``SKILL.md``'s frontmatter for its name/description and the dir's
  file count + newest mtime. No DB, no transcripts.
* :func:`usage_by_skill` is the *transcript* half — it turns the per-transcript
  ``skills_json`` blobs (``{skill: {count, last}}``, written by the indexer) plus
  the matching conversation summaries into a per-skill rollup: how many calls,
  in how many conversations, when it last ran, and what those conversations cost
  in total. Attribution is deliberately coarse — a conversation's whole spend is
  credited to *each* skill it invoked, so ``sessionCost`` answers "how much money
  flowed through work that used this skill", not "what did this skill cost".
  The narrower per-invocation figure is ``estSpend`` (context load × calls),
  computed by the caller from the file-token table.

``main.py`` merges the two: the union of disk skills and used-skill names, so a
built-in/plugin skill (``code-review``, ``artifact-design``, …) that never had a
``SKILL.md`` in this repo still shows up with its usage, marked ``builtin``.

**Name collisions merge.** The ``skills`` table is keyed by bare skill name, so a
project-local skill sharing a name with a repo one shares its counts. That's the
same identity Claude Code itself uses when a session invokes ``Skill(name)``.
"""
from __future__ import annotations

import datetime
import json
import os
import pathlib

from conversations import usage_cost, usage_tokens, zero_usage

WORKSPACE = pathlib.Path(os.environ.get("WORKSPACE", "/workspace")).resolve()
PROJECTS_DIR = pathlib.Path(
    os.environ.get("PROJECTS_DIR", "/workspace/projects")).resolve()

# Dirs a skill scan never descends into (mirrors main.IGNORE_DIRS' intent, but
# a skill dir is small and flat — this only guards against a stray vendor tree).
IGNORE_DIRS = {"node_modules", ".git", "__pycache__", "dist", "build", ".venv"}

# A skill dir is only a skill if it holds this file.
SKILL_FILE = "SKILL.md"


def iso(epoch: float) -> str:
    return (datetime.datetime.fromtimestamp(epoch, datetime.timezone.utc)
            .isoformat().replace("+00:00", "Z"))


# Kept under the old private name for this module's own call sites; `agents.py`
# reuses both helpers (an agent definition's frontmatter has the same shape).
_iso = iso


def parse_frontmatter(text: str) -> dict:
    """Top-level entries from a ``---`` frontmatter block.

    Hand-rolled rather than PyYAML on purpose: a skill ``description`` is one
    long unquoted sentence that routinely contains ``: `` (``"via the gh CLI —
    PRs, issues: …"``), which is a YAML syntax error. Splitting on the first
    colon is both correct and total for that shape.

    Two value shapes, because ``trigger_words`` needs a list:

    * ``key: value``          -> ``str``
    * ``key:`` + ``- item``   -> ``list[str]`` (a block sequence, one per line)

    A key with an empty value and no ``-`` lines under it yields ``""``, as
    before — nothing that reads a scalar key has to learn about lists.
    """
    if not text.startswith("---"):
        return {}
    end = text.find("\n---", 3)
    if end == -1:
        return {}
    out: dict[str, str | list[str]] = {}
    last_key: str | None = None
    for line in text[3:end].split("\n"):
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if stripped.startswith("- "):
            # A sequence item belongs to the key that opened the block.
            if last_key is not None and out.get(last_key) in ("", None):
                out[last_key] = []
            if last_key is not None and isinstance(out.get(last_key), list):
                out[last_key].append(stripped[2:].strip().strip("'\""))  # type: ignore[union-attr]
            continue
        if line.startswith((" ", "\t")):
            continue  # nested map entries: not used here
        key, sep, val = line.partition(":")
        if not sep:
            continue
        last_key = key.strip()
        out[last_key] = val.strip().strip("'\"")
    return out


def _as_list(v) -> list[str]:
    """A frontmatter value as a list — tolerating the scalar spellings.

    ``trigger_words`` is written as a block sequence, but a hand-edited
    ``SKILL.md`` may well carry ``trigger_words: a, b`` instead; both should
    reach the composer rather than one of them silently doing nothing.
    """
    if isinstance(v, list):
        return [str(x).strip() for x in v]
    if isinstance(v, str) and v.strip():
        return [p.strip() for p in v.split(",")]
    return []


def _dir_stats(d: pathlib.Path) -> tuple[int, float]:
    """(file count, newest mtime) for a skill dir, walked once."""
    files = 0
    newest = 0.0
    for dirpath, dirnames, filenames in os.walk(d):
        dirnames[:] = [x for x in dirnames if x not in IGNORE_DIRS]
        for fn in filenames:
            files += 1
            try:
                mt = (pathlib.Path(dirpath) / fn).stat().st_mtime
            except OSError:
                continue
            if mt > newest:
                newest = mt
    return files, newest


def _skill_roots() -> list[tuple[str, str, pathlib.Path]]:
    """The ``.claude/skills`` dirs to scan, as (source, sourceKind, path).

    ``source`` labels where a skill comes from in the UI: ``"repo"`` for the
    homelab's own skills, the project's dir-name for a project-local one.
    """
    roots: list[tuple[str, str, pathlib.Path]] = [
        ("repo", "repo", WORKSPACE / ".claude" / "skills"),
    ]
    try:
        entries = sorted(PROJECTS_DIR.iterdir())
    except OSError:
        entries = []
    for entry in entries:
        if entry.name.startswith(".") or not entry.is_dir():
            continue
        d = entry / ".claude" / "skills"
        if d.is_dir():
            roots.append((entry.name, "project", d))
    return roots


def catalog() -> list[dict]:
    """Every skill found on disk, newest-edited first.

    Each record is the *static* half of a skill card — ``path`` is repo-relative
    so the frontend can deep-link it into the file editor (``/_/<path>``).
    """
    out: list[dict] = []
    for source, source_kind, root in _skill_roots():
        try:
            entries = sorted(root.iterdir())
        except OSError:
            continue
        for entry in entries:
            if entry.name.startswith(".") or not entry.is_dir():
                continue
            md = entry / SKILL_FILE
            if not md.is_file():
                continue
            try:
                fm = parse_frontmatter(md.read_text(encoding="utf-8", errors="replace"))
            except OSError:
                fm = {}
            files, newest = _dir_stats(entry)
            try:
                rel = entry.relative_to(WORKSPACE).as_posix()
            except ValueError:
                rel = entry.as_posix()
            out.append({
                # The invocation name is the dir name — that's what `Skill(...)`
                # takes, and what the transcript counts are keyed by. Frontmatter
                # `name:` is only a fallback for a dir/frontmatter mismatch.
                "name": entry.name,
                "title": fm.get("name") or entry.name,
                "description": fm.get("description", ""),
                # What makes the composer's auto-tag notice this skill. Derived
                # by `scripts/skill-trigger-words.py` in the homelab repo, and
                # extendable per-tag in Settings → Tags.
                "triggerWords": [t for t in _as_list(fm.get("trigger_words")) if t],
                "dir": rel,
                "path": f"{rel}/{SKILL_FILE}",
                "source": source,
                "sourceKind": source_kind,
                "files": files,
                "updatedAt": _iso(newest) if newest else None,
            })
    out.sort(key=lambda s: s["updatedAt"] or "", reverse=True)
    return out


def usage_by_skill(summaries: list[tuple[str, dict]],
                   skill_rows: list[tuple[str, str]]) -> dict[str, dict]:
    """``{skill name -> {count, lastUsed, conversations, sessionCost, sessionTokens}}``.

    ``summaries`` is ``store.all_summaries()`` (``[(path, summary), …]``) and
    ``skill_rows`` is ``store.all_transcript_skill_rows()`` — the same transcript
    path keys both, which is what lets a skill's calls be joined to the spend of
    the conversations they ran in.
    """
    by_path = dict(summaries)
    out: dict[str, dict] = {}
    for path, raw in skill_rows:
        try:
            contrib = json.loads(raw or "{}")
        except ValueError:
            continue
        if not contrib:
            continue
        s = by_path.get(path) or {}
        # Same fallback as project_costs: older summaries predate the stored
        # cost/tokens rollup and only carry the raw `usage` block.
        cost = s.get("cost")
        tokens = s.get("tokens")
        if cost is None or tokens is None:
            u = {**zero_usage(), **(s.get("usage") or {})}
            model = s.get("model") or ""
            if cost is None:
                cost = usage_cost(u, model)
            if tokens is None:
                tokens = usage_tokens(u)
        for name, d in contrib.items():
            e = out.setdefault(name, {
                "count": 0, "lastUsed": None, "conversations": 0,
                "sessionCost": 0.0, "sessionTokens": 0,
            })
            e["count"] += d.get("count", 0)
            last = d.get("last")
            if last and (e["lastUsed"] is None or last > e["lastUsed"]):
                e["lastUsed"] = last
            e["conversations"] += 1
            e["sessionCost"] += cost
            e["sessionTokens"] += tokens
    return out