ai-agent/backend/agents.py

"""The **agents catalog** — every ``.claude/agents/**/*.md`` on disk, joined
with the runs mined from the transcripts and the cron store.

Deliberately parallel to :mod:`skills`, because an agent is the same kind of
object: a markdown file that declares a capability, plus a usage history. What
differs is *how it runs*, and that's what makes the numbers here better than
the skills catalog's estimates — an agent run is a **whole conversation**, so
its cost is measured, not guessed. Two origins feed the same rollup:

* **subagent** (``origin: "task"``) — the parent's ``Task``/``Agent`` tool call
  names it in ``subagent_type``, and Claude Code writes the child's transcript
  to ``<convId>/subagents/agent-<id>.jsonl`` beside an ``agent-<id>.meta.json``
  carrying ``agentType``. That sidecar is the join key: it turns a sidechain
  summary (which already knows its own real cost) into "this Explore run cost
  $0.42" with no re-walk of the parent.
* **session** (``origin: "cron"`` / ``"manual"``) — ``cron.py`` spawns
  ``.claude/agents/<name>.md`` as a *top-level* ``claude -p`` run, so those are
  not subagents at all. They are found through the cron job's ``history`` and,
  for a one-off "Run now" from the agent page, through the ``agentRun`` stamp
  the backend writes into the conversation's metadata sidecar.

This module owns the **disk half** (:func:`catalog`) and the **arithmetic**
(:func:`rollup`, :func:`daily`). Assembling the run list needs the store, the
metadata sidecar and the cron store, so it lives in ``main.py`` — the same
split ``skills.py`` uses.

**Naming.** A subagent's ``agentType`` is what Claude Code resolved the type to
(``Explore``, ``general-purpose``, ``goal-keeper``), which for a file-defined
agent is its frontmatter ``name:``. So the catalog is keyed by frontmatter name
(falling back to the file stem), not by path — that is the identity the runs
are recorded under. Built-in types (``Explore``, ``Plan``, …) have no file here
and surface as ``sourceKind: "builtin"``, exactly like a built-in skill.
"""
from __future__ import annotations

import datetime
import os
import pathlib

from skills import IGNORE_DIRS, iso, parse_frontmatter

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

# Where Claude Code looks for agent definitions, relative to a repo root.
AGENT_DIR = ".claude/agents"

# Subdirectories of ``.claude/agents`` that hold prompt files rather than agent
# definitions. Empty today — kept as the one place to list such a directory if
# another prompt store ever moves in next to the agent definitions.
SKIP_SUBDIRS: set[str] = set()

# Frontmatter keys that describe *how* a run is launched rather than what the
# agent is. Echoed to the UI as chips so a scheduled agent's harness/model is
# visible without opening the file (`cron.run_config` reads the same keys).
RUN_KEYS = ("harness", "model", "effort", "thinking")


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

    Same shape (and same labelling rules) as ``skills._skill_roots``: ``repo``
    for this workspace's own agents, the project's dir-name for a project-local
    one.
    """
    roots: list[tuple[str, str, pathlib.Path]] = [
        ("repo", "repo", WORKSPACE / ".claude" / "agents"),
    ]
    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" / "agents"
        if d.is_dir():
            roots.append((entry.name, "project", d))
    return roots


def _run_config(fm: dict) -> dict:
    """The ``harness``/``model``/``effort``/``thinking`` keys, verbatim.

    Unlike ``cron.run_config`` this does **not** validate or normalize: the
    catalog is a mirror of what the file says, and an unrecognised value is
    worth *showing* (that's how a typo gets noticed) rather than silently
    dropping. The spawn path still runs everything through ``cron.run_config``.
    """
    out = {}
    for k in RUN_KEYS:
        v = fm.get(k)
        out[k] = None if v is None else str(v)
    return out


def _tools(fm: dict) -> list[str]:
    """The ``tools:`` frontmatter as a list. Claude Code accepts a
    comma-separated string (``Bash, Read, Grep``) or ``*``; an absent key means
    "inherit every tool", which the UI renders as ``*``."""
    raw = (fm.get("tools") or "").strip()
    if not raw:
        return []
    return [t.strip() for t in raw.split(",") if t.strip()]


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

    Agent files may be nested (``.claude/agents/hooks/post-task.md`` is a real
    one), so this walks the tree rather than listing one level — but only
    ``.md`` files are agents, and ``README.md`` is not.
    """
    out: list[dict] = []
    for source, source_kind, root in _agent_roots():
        if not root.is_dir():
            continue
        for dirpath, dirnames, filenames in os.walk(root):
            # Only the *top* level is checked — a nested dir of the same name
            # deeper down is somebody's agent folder.
            skip = SKIP_SUBDIRS if pathlib.Path(dirpath) == root else set()
            dirnames[:] = sorted(d for d in dirnames if d not in IGNORE_DIRS
                                 and d not in skip and not d.startswith("."))
            for fn in sorted(filenames):
                if not fn.endswith(".md") or fn.startswith(".") \
                        or fn.lower() == "readme.md":
                    continue
                f = pathlib.Path(dirpath) / fn
                try:
                    text = f.read_text(encoding="utf-8", errors="replace")
                    st = f.stat()
                except OSError:
                    continue
                fm = parse_frontmatter(text)
                try:
                    rel = f.relative_to(WORKSPACE).as_posix()
                except ValueError:
                    rel = f.as_posix()
                out.append({
                    # The invocation name: what `Task(subagent_type=…)` takes
                    # and what the run records are keyed by.
                    "name": fm.get("name") or f.stem,
                    "title": fm.get("name") or f.stem,
                    "description": fm.get("description", ""),
                    "path": rel,
                    "dir": pathlib.PurePosixPath(rel).parent.as_posix(),
                    "source": source,
                    "sourceKind": source_kind,
                    "tools": _tools(fm),
                    "bytes": st.st_size,
                    "updatedAt": iso(st.st_mtime),
                    **_run_config(fm),
                })
    out.sort(key=lambda a: a["updatedAt"] or "", reverse=True)
    return out


def rollup(runs: list[dict]) -> dict[str, dict]:
    """``{agent name -> aggregate}`` over a run list built by ``main.py``.

    Each run is ``{agent, origin, at, cost, tokens, …}``. ``lastRun`` is the
    newest ``at``; ``conversations`` counts the *distinct parent* conversations
    a subagent ran in (a session run is its own conversation, so it counts as
    one) — the same "how widely is this used" question the skills catalog asks.
    """
    out: dict[str, dict] = {}
    convs: dict[str, set[str]] = {}
    for r in runs:
        name = r.get("agent")
        if not name:
            continue
        e = out.setdefault(name, {
            "runs": 0, "taskRuns": 0, "sessionRuns": 0, "lastRun": None,
            "cost": 0.0, "tokens": 0, "conversations": 0,
        })
        e["runs"] += 1
        e["taskRuns" if r.get("origin") == "task" else "sessionRuns"] += 1
        e["cost"] += r.get("cost") or 0.0
        e["tokens"] += r.get("tokens") or 0
        at = r.get("at")
        if at and (e["lastRun"] is None or at > e["lastRun"]):
            e["lastRun"] = at
        cid = r.get("conversationId")
        if cid:
            convs.setdefault(name, set()).add(cid)
    for name, e in out.items():
        e["conversations"] = len(convs.get(name, ()))
        e["avgCost"] = (e["cost"] / e["runs"]) if e["runs"] else 0.0
    return out


def daily(runs: list[dict], days: int = 30) -> list[dict]:
    """``[{date, runs, cost}]`` for the last ``days`` calendar days, oldest
    first — the sparkline under an agent's stat cards.

    Days with no run are emitted as zeroes so the chart's x-axis is even; the
    window is anchored on the newest run rather than on "now", so an agent that
    last ran two months ago still shows its shape instead of a flat line.
    """
    by_day: dict[str, dict] = {}
    for r in runs:
        at = r.get("at") or ""
        if len(at) < 10:
            continue
        d = by_day.setdefault(at[:10], {"date": at[:10], "runs": 0, "cost": 0.0})
        d["runs"] += 1
        d["cost"] += r.get("cost") or 0.0
    if not by_day:
        return []
    end = max(by_day)
    try:
        end_d = datetime.date.fromisoformat(end)
    except ValueError:
        return sorted(by_day.values(), key=lambda d: d["date"])
    out = []
    for i in range(days - 1, -1, -1):
        key = (end_d - datetime.timedelta(days=i)).isoformat()
        out.append(by_day.get(key) or {"date": key, "runs": 0, "cost": 0.0})
    return out