ai-agent/backend/agents/routes.py

"""Agents catalog routes (/api/agents)."""


from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

import accounts as accounts_mod
import agents as agents_mod
import cron as cron_mod
import schemas
from agents.runs import _agent_card, _agent_runs
from core.config import WORKSPACE
from core.http import _r
from core.state import AppState, State
from cron.service import _cron_job_out
from notifications import store as notify_mod
from runs.service import _send_message

router = APIRouter()


@router.get("/api/agents", responses=_r(schemas.AgentsResponse))
def agents_list(deps: State, account: str | None = None):
    """Card metadata + run/cost stats for every agent, last-run first.

    The **union** of the agent definitions on disk and the types the transcripts
    saw run — same shape as `/api/skills`, but the money here is real spend.
    ``account`` scopes the runs to one Claude account (unset = the totals).
    """
    sel = accounts_mod.selector(account)
    runs = [r for r in _agent_runs(deps, ) if sel(r.get("account"))]
    usage = agents_mod.rollup(runs)
    files = deps.store.all_files()

    items = [_agent_card(a, a["name"], usage, files)
             for a in agents_mod.catalog()]
    known = {a["name"] for a in items}
    for name in sorted(usage.keys() - known):
        items.append(_agent_card(None, name, usage, files))

    items.sort(key=lambda a: (a["lastRun"] or "", a["name"]), reverse=True)

    return {
        "agents": items,
        "totals": {
            "agents": len(items),
            "used": sum(1 for a in items if a["runs"] > 0),
            "runs": sum(a["runs"] for a in items),
            "cost": sum(a["cost"] for a in items),
            "tokens": sum(a["tokens"] for a in items),
        },
    }


def _agent_entry(name: str) -> dict | None:
    return next((a for a in agents_mod.catalog() if a["name"] == name), None)


def _agent_cron_job(deps: AppState, path: str | None) -> dict | None:
    """The cron job scheduled on this agent's file, if one exists — what turns
    the detail page's "Schedule" button into a "next run in 3h" line."""
    if not path:
        return None
    for job in deps.cron_store.list():
        if (job.get("promptFile") or "").lstrip("/") == path:
            return _cron_job_out(deps, job)
    return None


# How many runs the detail page's history list shows. The rollup counts them
# all; only the tail is worth shipping.
AGENT_RUN_LIMIT = 40


@router.get("/api/agents/{name}", responses=_r(schemas.AgentDetail))
def agent_detail(deps: State, name: str, account: str | None = None):
    """One agent: its definition, its stats, its recent runs and its schedule.

    404s only when the name is neither on disk nor in any transcript — an agent
    that exists but has never run is a perfectly good (empty) page.
    """
    entry = _agent_entry(name)
    runs = [r for r in _agent_runs(deps, ) if r["agent"] == name]
    if entry is None and not runs:
        raise HTTPException(404, "agent not found")
    sel = accounts_mod.selector(account)
    runs = [r for r in runs if sel(r.get("account"))]

    usage = agents_mod.rollup(runs)
    card = _agent_card(entry, name, usage, deps.store.all_files())

    content = ""
    if entry:
        try:
            content = (WORKSPACE / entry["path"]).read_text(
                encoding="utf-8", errors="replace")
        except OSError:
            content = ""

    # Schedulable = cron would accept this path as a prompt file. Nested
    # definitions (`.claude/agents/hooks/post-task.md`) and project-local ones
    # aren't, so the UI can say why instead of offering a button that 400s.
    can_schedule = False
    if entry:
        try:
            cron_mod.valid_prompt_file(entry["path"])
            can_schedule = True
        except ValueError:
            can_schedule = False

    return {
        "agent": card,
        "content": content,
        "runs": runs[:AGENT_RUN_LIMIT],
        "daily": agents_mod.daily(runs),
        "cron": _agent_cron_job(deps, entry["path"] if entry else None),
        "canSchedule": can_schedule,
    }


class AgentRunBody(BaseModel):
    # An extra instruction appended under the definition — the "what should it
    # do this time?" box on the agent page. Empty ⇒ run the definition as-is,
    # exactly like a cron firing does.
    prompt: str | None = None


@router.post("/api/agents/{name}/run", responses=_r(schemas.SpawnResult))
def agent_run(deps: State, name: str, body: AgentRunBody):
    """Run an agent now, as a top-level session — the agent page's "Run now".

    Same machinery as a cron firing (`_fire_cron_job`): the definition's body is
    the prompt and its frontmatter picks the harness/model/effort, so a "Run
    now" and a scheduled run are the same run. The session is stamped
    `agentRun` in the metadata sidecar, which is both what puts it in this
    agent's history and what makes the conversation card say where it came from.
    """
    entry = _agent_entry(name)
    if entry is None:
        raise HTTPException(404, "agent not found")
    try:
        raw = (WORKSPACE / entry["path"]).read_text(encoding="utf-8")
    except OSError:
        raise HTTPException(409, f"could not read {entry['path']}")
    cfg = cron_mod.run_config(raw)
    prompt = cron_mod.strip_frontmatter(raw).strip()
    if not prompt:
        raise HTTPException(409, f"{entry['path']} has no prompt body")
    extra = (body.prompt or "").strip()
    if extra:
        prompt = f"{prompt}\n\n---\n\n**This run's task:** {extra}"

    out = _send_message(deps, prompt, model=cfg["model"], harness=cfg["harness"],
                        thinking=cfg["thinking"], effort=cfg["effort"],
                        account=cfg.get("account"))
    sid = out["sessionId"]
    deps.meta_store.update(sid, {"agentRun": {"name": name,
                                         "at": notify_mod._now_iso()}})
    deps.hub.publish({"type": "meta", "id": sid})
    return out