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>
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""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 share_ref(item: dict) -> str | None:
|
||
"""How the guest-sharing list (users/store ``sharedSkills``) names a
|
||
catalog skill: bare for the repo's, ``<project>/<name>`` for a project's —
|
||
the shape the host sidecar resolves (sidecar ``shared_skills.resolve``)."""
|
||
if item.get("sourceKind") == "repo":
|
||
return item["name"]
|
||
if item.get("sourceKind") == "project":
|
||
return f"{item['source']}/{item['name']}"
|
||
return None
|
||
|
||
|
||
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 merge_same_name(items: list[dict]) -> list[dict]:
|
||
"""One card per skill name — the catalog lists a skill once per repo that
|
||
ships it (``complete`` lives in both the homelab and the ai-agent repo).
|
||
|
||
``Skill(name)`` resolves by bare name and the usage is keyed by it, so two
|
||
cards showed the same counts twice (and the totals summed them). The repo's
|
||
own copy leads (else the newest-edited); the others ride along in
|
||
``copies`` — their paths and share refs stay reachable, trigger words are
|
||
pooled. Order is the input's, at the first occurrence of each name.
|
||
"""
|
||
rank = lambda s: (s.get("sourceKind") == "repo", s.get("updatedAt") or "") # noqa: E731
|
||
copy_keys = ("source", "sourceKind", "dir", "path", "updatedAt",
|
||
"shareRef", "sharedWithGuests")
|
||
by_name: dict[str, dict] = {}
|
||
order: list[str] = []
|
||
for s in items:
|
||
cur = by_name.get(s["name"])
|
||
if cur is None:
|
||
by_name[s["name"]] = s
|
||
order.append(s["name"])
|
||
continue
|
||
lead, other = (cur, s) if rank(cur) >= rank(s) else (s, cur)
|
||
merged = {**lead,
|
||
"copies": [*(lead.get("copies") or []),
|
||
{k: other.get(k) for k in copy_keys},
|
||
*(other.get("copies") or [])],
|
||
"description": lead.get("description") or other.get("description", ""),
|
||
"triggerWords": list(dict.fromkeys(
|
||
[*(lead.get("triggerWords") or []),
|
||
*(other.get("triggerWords") or [])]))}
|
||
by_name[s["name"]] = merged
|
||
return [by_name[n] for n in order]
|
||
|
||
|
||
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
|