brain/devtools/tools.py

"""The developer-assistant tools. Master prompt 6: a capability inside BOTH modes.

"The tone follows the mode; the substance is identical." So each tool runs a specialist,
mode-free sub-call that returns JSON, and Prova's own turn presents that result in whichever
voice the mode calls for. Nothing here knows or cares which mode it is in.

Each result is also returned as an *artifact* — the web app renders a review as a diff view and
an API design as a table (master prompt 6's UI list), rather than as prose.
"""

import json
import logging
from typing import Any

import httpx

from .. import config
from ..llm.base import user_content
from ..tools.registry import Tool, ToolContext, ToolOutcome, failed, ok, schema, string
from . import github, prompts
from .attachments import resolve

log = logging.getLogger(__name__)

SUBCALL_TEMPERATURE = 0.4
MAX_CODE_CHARS = 120_000


async def _specialist(ctx: ToolContext, instruction: str, material: str) -> dict[str, Any] | None:
    turn = await ctx.llm.step(
        instruction.strip(),
        [user_content(material[:MAX_CODE_CHARS])],
        temperature=SUBCALL_TEMPERATURE,
        max_output_tokens=config.MAX_OUTPUT_TOKENS_TECHNICAL,
        response_mime_type="application/json",
    )
    return parse_json(turn.text)


def parse_json(text: str) -> dict[str, Any] | None:
    """JSON mode is requested, but a fenced or prefixed reply must not lose the whole result."""
    raw = text.strip()
    if raw.startswith("```"):
        raw = raw.strip("`")
        raw = raw.split("\n", 1)[1] if "\n" in raw else raw
    start, end = raw.find("{"), raw.rfind("}")
    if start == -1 or end == -1:
        return None
    try:
        value = json.loads(raw[start : end + 1])
    except json.JSONDecodeError:
        return None
    return value if isinstance(value, dict) else None


async def _review_code(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    label, code = resolve(args.get("target"), ctx.attachments, ctx.user_message)
    if not code.strip():
        return ToolOutcome(failed("there is no code to review — ask him to paste or attach it"))
    material = f"FILE: {label}\n\n{code}"
    if args.get("focus"):
        material = f"FOCUS: {args['focus']}\n\n{material}"

    result = await _specialist(ctx, prompts.REVIEW, material)
    if result is None:
        return ToolOutcome(failed("the review came back unreadable"))

    findings = [f for f in result.get("findings", []) if isinstance(f, dict)][:12]
    artifact = {
        "type": "review",
        "target": label,
        "summary": result.get("summary", ""),
        "findings": findings,
    }
    return ToolOutcome(ok("reviewed", **artifact), artifacts=[artifact])


async def _explain_code(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    label, code = resolve(args.get("target"), ctx.attachments, ctx.user_message)
    if not code.strip():
        return ToolOutcome(failed("there is no code to explain"))
    material = f"FILE: {label}\n\n{code}"
    if args.get("focus"):
        material = f"QUESTION: {args['focus']}\n\n{material}"
    result = await _specialist(ctx, prompts.EXPLAIN, material)
    if result is None:
        return ToolOutcome(failed("the explanation came back unreadable"))
    return ToolOutcome(ok("explained", target=label, **result))


async def _debug_error(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    error = str(args["error"]).strip()
    label, code = resolve(args.get("target"), ctx.attachments, ctx.user_message)
    material = f"ERROR:\n{error}"
    if code.strip() and code.strip() != error:
        material += f"\n\nCODE ({label}):\n{code}"
    result = await _specialist(ctx, prompts.DEBUG, material)
    if result is None:
        return ToolOutcome(failed("the diagnosis came back unreadable"))
    artifact = {"type": "debug", "target": label, **result}
    return ToolOutcome(ok("diagnosed", **result), artifacts=[artifact])


async def _analyze_project(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    if not ctx.tree and not ctx.attachments:
        return ToolOutcome(failed("no project is open — ask him to open a folder or attach files"))
    parts = []
    if ctx.tree:
        parts.append("TREE:\n" + "\n".join(ctx.tree[:1500]))
    for name, content in list(ctx.attachments.items())[:20]:
        parts.append(f"FILE: {name}\n{content[:20_000]}")
    if args.get("focus"):
        parts.insert(0, f"FOCUS: {args['focus']}")
    result = await _specialist(ctx, prompts.ANALYZE_PROJECT, "\n\n".join(parts))
    if result is None:
        return ToolOutcome(failed("the analysis came back unreadable"))
    artifact = {"type": "analysis", **result}
    return ToolOutcome(ok("analysed", **result), artifacts=[artifact])


async def _design_api(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    result = await _specialist(ctx, prompts.DESIGN_API, f"REQUIREMENTS:\n{args['requirements']}")
    if result is None:
        return ToolOutcome(failed("the design came back unreadable"))
    artifact = {"type": "api", **result}
    return ToolOutcome(ok("designed", **result), artifacts=[artifact])


async def _fetch_github(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome:
    client = ctx.http or httpx.AsyncClient(timeout=20.0, follow_redirects=False)
    try:
        fetched = await github.fetch(client, str(args["url"]))
    except (github.GithubError, httpx.HTTPError) as exc:
        return ToolOutcome(failed(str(exc) or "could not fetch that link"))
    finally:
        if ctx.http is None:
            await client.aclose()

    if fetched["kind"] == "file":
        # Available to the other dev tools for the rest of this turn.
        ctx.attachments[fetched["name"]] = fetched["content"]
        return ToolOutcome(
            ok("fetched", file=fetched["name"], content=fetched["content"][:MAX_CODE_CHARS])
        )
    ctx.tree = fetched["paths"]
    return ToolOutcome(
        ok(
            "fetched tree",
            repo=fetched["repo"],
            files=len(fetched["paths"]),
            paths=fetched["paths"][:300],
        )
    )


_TARGET = string("An attached file's name, or omit to use the code pasted in his message")

DEV_TOOLS: list[Tool] = [
    Tool(
        "review_code",
        "Review code for bugs, security, performance and WordPress/WooCommerce/Laravel/NestJS "
        "best practice. Returns findings with suggested edits he can see as a diff.",
        schema({"target": _TARGET, "focus": string("Anything specific he wants checked")}),
        _review_code,
    ),
    Tool(
        "explain_code",
        "Explain what a piece of code does, step by step.",
        schema({"target": _TARGET, "focus": string("His specific question about it")}),
        _explain_code,
    ),
    Tool(
        "debug_error",
        "Diagnose an error message or stack trace, with the code if he gave it.",
        schema(
            {"error": string("The error or stack trace, verbatim"), "target": _TARGET}, ["error"]
        ),
        _debug_error,
    ),
    Tool(
        "analyze_project",
        "Get oriented in a whole project from its file tree and attached files.",
        schema({"focus": string("What he wants to understand about it")}),
        _analyze_project,
    ),
    Tool(
        "design_api",
        "Design a REST API — resources, endpoints, data model — from requirements.",
        schema({"requirements": string("What the API has to do")}, ["requirements"]),
        _design_api,
    ),
    Tool(
        "fetch_github",
        "Fetch a public GitHub file or repository tree from a github.com link he shared.",
        schema({"url": string("The github.com or raw.githubusercontent.com link")}, ["url"]),
        _fetch_github,
    ),
]