agent-reviewer/agent.py

"""agent-reviewer — pre-deploy audit of an A2A agent's source."""
from __future__ import annotations

import asyncio
import re
from contextlib import suppress
from typing import Any

from pydantic import BaseModel

from a2a_pack import (
    A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill,
)

from agent_reviewer import ReviewerContext, build_reviewer_graph
from agent_reviewer.tools import ReviewReport


class ReviewerConfig(BaseModel):
    pass


DEEPAGENTS_RECURSION_LIMIT = 500


class AgentReviewer(A2AAgent[ReviewerConfig, NoAuth]):
    name = "agent-reviewer"
    description = (
        "Audit an A2A agent's source before deploy. Reads the target "
        "repo from Gitea, applies security / best-practice / grant-scope "
        "skill bundles, and returns a structured ReviewReport."
    )
    version = "0.1.0"

    config_model = ReviewerConfig
    auth_model = NoAuth
    tools_used = ("deepagents", "a2a-pack", "litellm", "microsandbox", "gitea")
    llm_provisioning = LLMProvisioning.PLATFORM
    wants_cp_jwt = True
    pricing = Pricing(
        price_per_call_usd=0.05,
        caller_pays_llm=True,
        notes=(
            "Reads target source via a short-lived, repo-scoped Gitea "
            "token minted on the caller's behalf. Never deploys or writes."
        ),
    )

    @skill(
        description=(
            "Audit an agent's source. Pass the target ``agent_name`` "
            "(kebab-case) and optional ``ref`` (defaults to ``main``). "
            "Returns a ReviewReport with severity-categorized findings."
        ),
        tags=["reviewer", "audit", "meta"],
        stream=True,
        timeout_seconds=900,
        cost_class="medium",
    )
    async def review(
        self,
        ctx: RunContext[NoAuth],
        agent_name: str,
        ref: str = "main",
        owner: str | None = None,
    ) -> dict[str, Any]:
        if not _is_valid_slug(agent_name):
            return {"error": f"invalid agent_name {agent_name!r} — use kebab-case"}
        if not ctx.cp_jwt or not ctx.cp_url:
            return {"error": "no cp_jwt; this agent must run through the orchestrator"}

        creds = ctx.llm
        if not creds.api_key:
            return {
                "error": "llm_credentials_missing",
                "final": (
                    "LLM key required. Add an LLM credential in Settings > "
                    "LLM credentials before running this review."
                ),
                "agent_name": agent_name,
                "ref": ref,
            }

        await ctx.emit_progress(f"requesting read-only Gitea token for {agent_name}@{ref}")
        try:
            mint = await ctx.mint_gitea_token(
                agent_name,
                scope="read",
                owner=owner,
                ttl_seconds=900,
                purpose=f"reviewer:{agent_name}@{ref}",
            )
        except RuntimeError as exc:
            return {"error": f"gitea token mint failed: {exc}"}

        token_name = mint["token_name"]
        gitea_token = mint["token"]
        gitea_owner = mint["owner"]
        await ctx.emit_progress(
            f"minted token {token_name} (read on {gitea_owner}/{agent_name})"
        )

        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")

        completion_box: dict[str, ReviewReport] = {}
        graph = build_reviewer_graph(
            ReviewerContext(
                agent_name=agent_name,
                ref=ref,
                gitea_token=gitea_token,
                gitea_owner=gitea_owner,
                llm_base_url=creds.base_url,
                llm_api_key=creds.api_key,
                llm_model=creds.model,
                llm_temperature_mode=creds.temperature_mode,
                llm_temperature=creds.temperature,
                llm_extra_body=creds.extra_body,
            ),
            completion_box=completion_box,
        )

        user_msg = (
            f"Review the agent at ``{gitea_owner}/{agent_name}`` ref ``{ref}``. "
            "Read the source, consult the bundled skills, run the sandbox checks, "
            "then call ``submit_review_report`` exactly once with the structured "
            "report."
        )

        async def _heartbeat() -> None:
            while True:
                await asyncio.sleep(20)
                await ctx.emit_progress(f"reviewer still working on {agent_name!r}")

        heartbeat_task = asyncio.create_task(_heartbeat())
        last_reply = ""
        try:
            async for event in graph.astream_events(
                {"messages": [{"role": "user", "content": user_msg}]},
                version="v2",
                config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
            ):
                kind = event.get("event")
                tname = event.get("name") or ""
                data = event.get("data") or {}
                if kind == "on_tool_start":
                    await ctx.emit_progress(f"→ {tname}")
                elif kind == "on_tool_end":
                    await ctx.emit_progress(f"  {tname} ✓")
                elif kind == "on_chain_end" and not event.get("parent_ids"):
                    out = data.get("output")
                    if isinstance(out, dict):
                        for msg in out.get("messages") or []:
                            content = (
                                msg.get("content")
                                if isinstance(msg, dict)
                                else getattr(msg, "content", None)
                            )
                            if isinstance(content, str) and content.strip():
                                last_reply = content
        except Exception as exc:  # noqa: BLE001
            return {"error": f"reviewer graph failed: {type(exc).__name__}: {exc}"}
        finally:
            heartbeat_task.cancel()
            with suppress(asyncio.CancelledError):
                await heartbeat_task
            with suppress(Exception):
                await ctx.release_gitea_token(token_name)

        report = completion_box.get("report")
        if report is None:
            return {
                "ok": False,
                "agent_name": agent_name,
                "ref": ref,
                "warning": "reviewer finished without submitting a structured report",
                "reply": last_reply[:2000],
            }
        return report.model_dump()


def _is_valid_slug(name: str) -> bool:
    return bool(re.match(r"^[a-z][a-z0-9-]{1,62}$", name))