"""LangChain tools the inner deepagents loop uses.

  - workspace file CRUD through the caller's scoped workspace grant
  - sandbox python (for `a2a card` / `a2a validate` round-trips)
  - cp_deploy_tarball (call /v1/agents/from-tarball on the user's behalf)
  - cp_compose_meta_agent (call /v1/agents/compose with a manifest)
"""
from __future__ import annotations

import asyncio
import ast
import base64
import io
import json
import re
import tarfile
import time
from dataclasses import dataclass
from importlib import resources
from pathlib import Path
from typing import TYPE_CHECKING, Any

import boto3
import httpx
from botocore.config import Config as _BotoConfig
from botocore.exceptions import ClientError
from langchain_core.tools import tool

if TYPE_CHECKING:
    from .config import Settings


A2A_PACK_MIN_VERSION = "0.1.51"


@dataclass(frozen=True)
class ToolContext:
    bucket: str
    settings: "Settings"
    cp_jwt: str | None = None
    workspace: Any | None = None
    grant_token: str | None = None


async def _wait_for_live_card(
    url: str | None,
    expected_version: str | None,
    timeout_s: float = 900.0,
) -> tuple[bool, dict[str, Any]]:
    """Poll ``{url}/.well-known/agent-card`` until ``version`` matches
    ``expected_version`` or ``timeout_s`` elapses. Returns
    ``(matched, last_seen_card)``. ``last_seen_card`` is ``{}`` if the
    endpoint never responded with a parseable card.

    Backoff: 2s → 1.5x per try, capped at 10s. Tolerates pod-not-ready
    (any HTTPError or 4xx/5xx) — just keeps polling.
    """
    if not url or not expected_version:
        return False, {}
    deadline = time.monotonic() + timeout_s
    last_card: dict[str, Any] = {}
    interval = 2.0
    while time.monotonic() < deadline:
        try:
            async with httpx.AsyncClient(timeout=5.0) as c:
                r = await c.get(f"{url.rstrip('/')}/.well-known/agent-card")
            if r.status_code == 200:
                card = r.json() or {}
                if isinstance(card, dict):
                    last_card = card
                    if card.get("version") == expected_version:
                        return True, card
        except (httpx.HTTPError, ValueError):
            pass
        await asyncio.sleep(interval)
        interval = min(interval * 1.5, 10.0)
    return False, last_card


def _s3(ctx: ToolContext) -> Any:
    return boto3.client(
        "s3",
        endpoint_url=ctx.settings.minio_endpoint,
        aws_access_key_id=ctx.settings.minio_access_key,
        aws_secret_access_key=ctx.settings.minio_secret_key,
        region_name="us-east-1",
        config=_BotoConfig(s3={"addressing_style": "path"}),
    )


def _ensure_bucket(s3: Any, bucket: str) -> None:
    try:
        s3.head_bucket(Bucket=bucket)
    except ClientError as exc:
        code = exc.response.get("Error", {}).get("Code")
        if code in {"404", "NoSuchBucket", "NotFound"}:
            s3.create_bucket(Bucket=bucket)
        elif code != "403":
            raise


class _ObjectStore:
    def iter_keys(self, prefix: str) -> list[str]:
        raise NotImplementedError

    def get(self, key: str) -> bytes:
        raise NotImplementedError

    def put(self, key: str, body: bytes, content_type: str) -> None:
        raise NotImplementedError

    def delete(self, key: str) -> None:
        raise NotImplementedError


class _S3ObjectStore(_ObjectStore):
    def __init__(self, s3: Any, bucket: str) -> None:
        self._s3 = s3
        self._bucket = bucket

    def iter_keys(self, prefix: str) -> list[str]:
        keys: list[str] = []
        paginator = self._s3.get_paginator("list_objects_v2")
        for page in paginator.paginate(Bucket=self._bucket, Prefix=prefix):
            for obj in page.get("Contents") or ():
                key = obj.get("Key")
                if isinstance(key, str):
                    keys.append(key)
        return sorted(keys)

    def get(self, key: str) -> bytes:
        obj = self._s3.get_object(Bucket=self._bucket, Key=key)
        return obj["Body"].read()

    def put(self, key: str, body: bytes, content_type: str) -> None:
        self._s3.put_object(
            Bucket=self._bucket,
            Key=key,
            Body=body,
            ContentType=content_type,
        )

    def delete(self, key: str) -> None:
        self._s3.delete_object(Bucket=self._bucket, Key=key)


class _WorkspaceObjectStore(_ObjectStore):
    def __init__(self, workspace: Any) -> None:
        self._workspace = workspace

    def iter_keys(self, prefix: str) -> list[str]:
        return sorted(
            key
            for key in self._workspace.iter_paths()
            if isinstance(key, str) and key.startswith(prefix)
        )

    def get(self, key: str) -> bytes:
        return self._workspace.read_bytes(key)

    def put(self, key: str, body: bytes, content_type: str) -> None:
        del content_type
        self._workspace.write_bytes(key, body)

    def delete(self, key: str) -> None:
        self._workspace.delete_path(key)


def _store_from_args(
    store_or_s3: Any,
    bucket_or_prefix: str,
    prefix: str | None,
) -> tuple[_ObjectStore, str]:
    """Support new helper calls with ObjectStore and old tests with fake S3."""
    if isinstance(store_or_s3, _ObjectStore):
        return store_or_s3, bucket_or_prefix
    if hasattr(store_or_s3, "iter_keys") and hasattr(store_or_s3, "get"):
        return store_or_s3, bucket_or_prefix
    if prefix is None:
        raise TypeError("prefix is required when passing an S3 client")
    return _S3ObjectStore(store_or_s3, bucket_or_prefix), prefix


def _sandbox_headers(
    bearer_token: str | None,
    grant_token: str | None,
) -> dict[str, str] | None:
    if bearer_token:
        return {"authorization": f"Bearer {bearer_token}"}
    if grant_token:
        return {"X-A2A-Grant": grant_token}
    return None


_AGENT_NAME_RE = re.compile(r"^[a-z][a-z0-9-]{1,62}$")
_SKILL_NAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$")
_BUILDER_STATE_FILE = ".a2a-builder-state.json"
_BUILDER_INTERNAL_PREFIX = ".agent-builder/"
_DERIVED_PRICING_FIELDS = frozenset({"compute", "total_usd"})


def _validate_name(name: str) -> None:
    if not _AGENT_NAME_RE.match(name):
        raise ValueError(
            f"invalid agent name {name!r}: must match {_AGENT_NAME_RE.pattern}"
        )


def _validate_skill_name(name: str) -> None:
    if not _SKILL_NAME_RE.match(name) or "--" in name:
        raise ValueError(
            "invalid skill name "
            f"{name!r}: use lowercase alphanumeric kebab-case, max 64 chars"
        )


def _agent_prefix(name: str) -> str:
    _validate_name(name)
    return f"agents/{name}/"


def _builder_state_key(name: str) -> str:
    return _agent_prefix(name) + _BUILDER_STATE_FILE


def build_tools(ctx: ToolContext) -> list[Any]:
    bucket = ctx.bucket
    settings = ctx.settings
    if ctx.workspace is not None:
        store: _ObjectStore = _WorkspaceObjectStore(ctx.workspace)
    else:
        s3 = _s3(ctx)
        _ensure_bucket(s3, bucket)
        store = _S3ObjectStore(s3, bucket)

    @tool
    def init_agent_template(
        name: str,
        description: str = "A new A2A agent",
        frontend: str = "none",
    ) -> str:
        """Initialize ``agents/<name>/`` from the installed a2a-pack template.

        Use this FIRST for a new project. It writes the same baseline files
        as ``a2a init`` for the SDK version installed in this runtime:
        ``agent.py``, ``a2a.yaml``, and ``requirements.txt``. Set
        ``frontend`` to ``"react"`` for a Vite packed app, ``"static"`` for a
        minimal bundled HTML app, or ``"none"`` for a headless agent. After
        this, edit those files with ``read_agent_file`` + ``write_agent_file``
        to implement the user's requested behavior.
        """
        try:
            frontend_kind = _normalize_frontend_kind(frontend)
            files = _render_a2a_init_template(
                name,
                description=description,
                frontend=frontend_kind,
            )
            prefix = _agent_prefix(name)
        except (OSError, RuntimeError, ValueError) as exc:
            return json.dumps({"error": str(exc)})
        written: list[dict[str, Any]] = []
        for path, content in files.items():
            key = prefix + path
            encoded = content.encode("utf-8")
            store.put(key, encoded, "text/plain; charset=utf-8")
            written.append({"path": key, "size": len(encoded)})
        return json.dumps({
            "ok": True,
            "agent": name,
            "frontend": frontend_kind,
            "files": written,
            "docs": (
                "https://docs.a2acloud.io/concepts/packed-frontends"
                if frontend_kind != "none"
                else "https://docs.a2acloud.io/quickstart"
            ),
        })

    @tool
    def list_agent_files(name: str) -> str:
        """List every file under ``agents/<name>/`` in the user's workspace.

        Use this before edits to see what you already wrote / what's left.
        """
        try:
            prefix = _agent_prefix(name)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})
        out: list[dict[str, Any]] = []
        for key in store.iter_keys(prefix):
            rel = key[len(prefix):]
            if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
                continue
            out.append({"path": rel, "size": len(store.get(key))})
        return json.dumps({"agent": name, "files": out})

    @tool
    def write_agent_file(name: str, path: str, content: str) -> str:
        """Write a file under ``agents/<name>/<path>`` in the user's workspace.

        Use this for ``agent.py``, ``a2a.yaml``, ``requirements.txt``, and
        any auxiliary modules or packed frontend files the agent needs.
        """
        try:
            prefix = _agent_prefix(name)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})
        rel = path.lstrip("/")
        if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
            return json.dumps({"error": f"{rel} is managed by agent-builder"})
        sanitized_fields: tuple[str, ...] = ()
        if rel == "agent.py":
            content, sanitized_fields = _strip_unsupported_pricing_fields(content)
        key = prefix + rel
        encoded = content.encode("utf-8")
        out: dict[str, Any] = {"ok": True, "path": key, "size": len(encoded)}
        if sanitized_fields:
            out["sanitized_agent_card_fields"] = list(sanitized_fields)
            out["warning"] = (
                "Removed derived Pricing fields that the Agent Card schema "
                "does not accept. Declare Resources separately; compute and "
                "total_usd are filled by the platform."
            )
        store.put(key, encoded, "text/plain; charset=utf-8")
        return json.dumps(out)

    @tool
    def read_agent_file(name: str, path: str) -> str:
        """Read a single file from the scaffolded agent project."""
        try:
            prefix = _agent_prefix(name)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})
        rel = path.lstrip("/")
        if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
            return json.dumps({"error": f"{rel} is managed by agent-builder"})
        key = prefix + rel
        try:
            data = store.get(key)
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": f"read failed: {exc}"})
        body = data.decode("utf-8", errors="replace")
        return json.dumps({"path": key, "content": body[:200_000]})

    @tool
    def write_agent_skill(
        name: str,
        skill_name: str,
        description: str,
        instructions: str,
        supporting_files_json: str = "{}",
    ) -> str:
        """Create a DeepAgents skill bundle in ``agents/<name>/skills/``.

        Use this for generated agents that should rely on DeepAgents'
        progressive-disclosure skills instead of fake one-off tools.
        ``supporting_files_json`` is a JSON object mapping relative paths
        inside the skill directory to text content, for example
        ``{"references/schema.md": "..."}``.

        After writing skills, update ``agent.py`` so it seeds packaged
        ``skills/`` files into ``ctx.workspace_backend()`` and passes
        ``skills=[...]`` to ``create_deep_agent``.
        """
        try:
            prefix = _agent_prefix(name)
            _validate_skill_name(skill_name)
            files = _parse_supporting_skill_files(supporting_files_json)
            skill_md = _render_skill_md(skill_name, description, instructions)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})

        base = prefix + f"skills/{skill_name}/"
        written: list[dict[str, Any]] = []
        payloads = {"SKILL.md": skill_md, **files}
        for rel, content in payloads.items():
            key = base + rel
            encoded = content.encode("utf-8")
            store.put(key, encoded, "text/plain; charset=utf-8")
            written.append({"path": key, "size": len(encoded)})
        return json.dumps({
            "ok": True,
            "agent": name,
            "skill": skill_name,
            "source_path": f"skills/{skill_name}/",
            "files": written,
            "wire_agent_py": (
                "Seed project skills into ctx.workspace_backend() and pass "
                "skills=[RUNTIME_SKILLS_ROOT] to create_deep_agent."
            ),
        })

    @tool
    async def test_agent_in_sandbox(name: str) -> str:
        """Spin a microVM, ``pip install a2a-pack`` + the agent's own
        requirements.txt, then run ``a2a card`` to verify the scaffold
        produces a valid Card. When a packed frontend is declared, it also
        prints ``a2a frontend info`` metadata. Returns stdout/stderr.
        """
        try:
            prefix = _agent_prefix(name)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})
        sanitized_fields = _sanitize_agent_workspace_pricing(store, prefix)

        # Bundle the workspace files into the sandbox via base64 tar so
        # the sandbox doesn't need MinIO read access for this skill.
        bundle_bytes = _tarball_workspace_dir(store, prefix)
        if not bundle_bytes:
            return json.dumps({
                "error": "no files in agents/" + name + "/ — write some first",
            })
        b64 = base64.b64encode(bundle_bytes).decode("ascii")
        script = (
            "set -e\n"
            f"pip install --quiet 'a2a-pack>={A2A_PACK_MIN_VERSION}' >/dev/null\n"
            "mkdir -p /tmp/agent\n"
            f"echo '{b64}' | base64 -d | tar -xzf - -C /tmp/agent\n"
            "cd /tmp/agent\n"
            "ls -la\n"
            "if [ -f requirements.txt ]; then\n"
            "  pip install --quiet -r requirements.txt >/dev/null || true\n"
            "fi\n"
            "echo '--- card ---'\n"
            "a2a card --project .\n"
            "if grep -q '^frontend:' a2a.yaml 2>/dev/null; then\n"
            "  echo '--- frontend ---'\n"
            "  a2a frontend info --project . || true\n"
            "fi\n"
        )
        headers = _sandbox_headers(settings.sandbox_token, ctx.grant_token)
        async with httpx.AsyncClient(timeout=settings.sandbox_timeout_s + 30) as c:
            r = await c.post(
                f"{settings.sandbox_url}/v1/run_shell",
                headers=headers,
                json={
                    "bucket": bucket, "script": script,
                    "image": settings.image, "memory_mib": 512,
                    "timeout_seconds": settings.sandbox_timeout_s,
                },
            )
        if r.status_code >= 400:
            return json.dumps({"error": f"sandbox {r.status_code}", "detail": r.text[:1000]})
        out = r.json()
        result: dict[str, Any] = {
            "exit_code": out.get("exit_code"),
            "stdout": (out.get("stdout") or "")[:4000],
            "stderr": (out.get("stderr") or "")[:4000],
        }
        if sanitized_fields:
            result["sanitized_agent_card_fields"] = list(sanitized_fields)
        return json.dumps(result)

    @tool
    async def cp_deploy_tarball(
        name: str,
        version: str = "0.1.0",
        public: bool = True,
        force: bool = False,
    ) -> str:
        """Tar the agent at ``agents/<name>/`` and POST it to the control
        plane's ``/v1/agents/from-tarball`` endpoint on the user's behalf,
        then BLOCK until the live ``.well-known/agent-card`` reports the
        new version (or 180s timeout).

        Returns JSON with ``ok``, ``live``, ``url``, ``version``,
        ``live_version``, and ``live_skills[{name, input_schema}]``. When
        ``live=true`` the agent is callable; when ``live=false`` the
        deploy POST succeeded but the pod never came up with the new
        version — caller should treat this as a failure, not success.

        Compare ``live_skills[].input_schema`` against the code you wrote
        to confirm the advertised schema matches the params the function
        actually accepts. Mismatches cause downstream ``agent 400`` from
        ``call_agent``.

        ``force`` defaults to false so a stale MinIO workspace cannot
        silently overwrite source that was deployed from a user's computer
        or another tool. Only set ``force=True`` after the user explicitly
        accepts replacing the current managed repo contents.

        Requires the orchestrator to have forwarded the user's CP JWT
        (the platform does this automatically when this agent's Card
        declared ``wants_cp_jwt=True``).
        """
        try:
            prefix = _agent_prefix(name)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})
        if not ctx.cp_jwt:
            return json.dumps({
                "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True",
            })

        try:
            latest_deploy = await _latest_cp_deployment(settings.cp_url, ctx.cp_jwt, name)
        except RuntimeError as exc:
            return json.dumps({"error": str(exc)})
        builder_state = _read_builder_state(store, name)
        drift = _deployment_drift_error(
            name,
            latest_deploy,
            builder_state,
            force=force,
        )
        if drift is not None:
            return json.dumps(drift)
        base_head = builder_state.get("repo_head_sha")
        if force:
            base_head = None

        sanitized_fields = _sanitize_agent_workspace_pricing(store, prefix)
        bundle_bytes = _tarball_workspace_dir(store, prefix)
        if not bundle_bytes:
            return json.dumps({"error": f"no files at agents/{name}/"})
        # Read the entrypoint from a2a.yaml so we can pass it verbatim.
        entrypoint = _read_entrypoint(bundle_bytes) or f"agent:{_class_name(name)}"

        # NOTE: the CP from-tarball endpoint names the file field
        # ``source`` (see apps/control-plane/control_plane/routes/agents.py).
        files = {"source": (f"{name}.tar.gz", bundle_bytes, "application/gzip")}
        data = {
            "name": name, "version": version,
            "entrypoint": entrypoint, "public": "true" if public else "false",
            "description": "Built by agent-builder.",
        }
        if isinstance(base_head, str) and base_head:
            data["base_head_sha"] = base_head
        try:
            async with httpx.AsyncClient(timeout=60.0) as c:
                r = await c.post(
                    f"{settings.cp_url}/v1/agents/from-tarball",
                    headers={"authorization": f"bearer {ctx.cp_jwt}"},
                    files=files, data=data,
                )
        except httpx.HTTPError as exc:
            return json.dumps({"error": f"cp unreachable: {exc}"})
        if r.status_code >= 400:
            return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]})
        body = r.json()
        name_ = body.get("name")
        version_ = body.get("version")
        url_ = body.get("url")
        head_sha = body.get("head_sha")
        if isinstance(head_sha, str) and head_sha:
            _write_builder_state(
                store,
                name,
                {
                    "agent": name,
                    "repo_head_sha": head_sha,
                    "deployment_id": body.get("deployment_id"),
                    "version": version_,
                    "updated_at": int(time.time()),
                    "source": "agent-builder",
                },
            )

        # CP returns immediately with status="building". The agent isn't
        # actually callable until the pod rolls + serves the new card.
        # Poll /.well-known/agent-card until the live card reports the
        # version we just deployed (or timeout). Surface the live skills
        # so the LLM can verify the input_schema actually matches the
        # code it wrote — catches the schema/code skew that causes
        # downstream 400s on call_agent.
        live, live_card = await _wait_for_live_card(
            url_,
            version_,
            timeout_s=settings.deploy_wait_timeout_s,
        )
        out: dict[str, Any] = {
            "ok": live,
            "name": name_,
            "version": version_,
            "status": body.get("status"),
            "url": url_,
            "head_sha": head_sha,
            "deployment_id": body.get("deployment_id"),
            "workspace_base_head_sha": base_head,
            "live": live,
        }
        if sanitized_fields:
            out["sanitized_agent_card_fields"] = list(sanitized_fields)
        if live:
            out["live_version"] = live_card.get("version")
            out["live_skills"] = [
                {
                    "name": s.get("name"),
                    "input_schema": s.get("input_schema") or {},
                }
                for s in (live_card.get("skills") or [])
            ]
        else:
            out["error"] = (
                "deploy did not become live within timeout — pod may still"
                " be rolling or build failed. Call cp_deploy_tarball again"
                " or inspect via discover_agent before declaring done."
            )
            if live_card:
                out["last_seen_version"] = live_card.get("version")
        return json.dumps(out)

    @tool
    async def cp_compose_meta_agent(
        name: str,
        manifest_json: str,
        description: str = "",
        version: str = "0.1.0",
        public: bool = True,
        refresh_existing: bool = False,
    ) -> str:
        """Deploy a declarative meta-agent composition through the control plane.

        Use this when the user asks to compose existing agents toward a goal.
        ``manifest_json`` must be a JSON object with ``composition`` and
        optional ``goal`` / ``memory`` blocks. The control plane validates
        referenced sub-agents and skills, generates editable MetaAgent source,
        commits it to the user's managed source repo, and stamps the runtime
        repo. This is the preferred path for meta-agents; do not hand-write
        orchestration source when a manifest is enough.
        """
        try:
            prefix = _agent_prefix(name)
            manifest = _parse_manifest_json(manifest_json)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})
        if not ctx.cp_jwt:
            return json.dumps({
                "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True",
            })
        body = {
            "name": name,
            "description": description,
            "version": version,
            "public": public,
            "manifest": manifest,
            "refresh_existing": refresh_existing,
        }
        try:
            async with httpx.AsyncClient(timeout=60.0) as c:
                r = await c.post(
                    f"{settings.cp_url}/v1/agents/compose",
                    headers={"authorization": f"bearer {ctx.cp_jwt}"},
                    json=body,
                )
        except httpx.HTTPError as exc:
            return json.dumps({"error": f"cp unreachable: {exc}"})
        if r.status_code >= 400:
            return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]})
        payload = r.json() or {}
        head_sha = payload.get("head_sha")
        store.put(
            prefix + "meta_agent_manifest.json",
            (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"),
            "application/json",
        )
        if isinstance(head_sha, str) and head_sha:
            _write_builder_state(
                store,
                name,
                {
                    "agent": name,
                    "repo_head_sha": head_sha,
                    "deployment_id": payload.get("deployment_id"),
                    "version": payload.get("version"),
                    "updated_at": int(time.time()),
                    "source": "agent-builder-compose",
                },
            )
        url = payload.get("expected_url") or payload.get("url")
        live, live_card = await _wait_for_live_card(
            url,
            payload.get("version"),
            timeout_s=settings.deploy_wait_timeout_s,
        )
        out: dict[str, Any] = {
            "ok": live,
            "name": payload.get("name"),
            "version": payload.get("version"),
            "status": payload.get("status"),
            "url": url,
            "head_sha": head_sha,
            "deployment_id": payload.get("deployment_id"),
            "preview": payload.get("preview") or {},
            "live": live,
        }
        if live:
            out["live_version"] = live_card.get("version")
            out["live_skills"] = [
                {
                    "name": s.get("name"),
                    "input_schema": s.get("input_schema") or {},
                }
                for s in (live_card.get("skills") or [])
            ]
        else:
            out["error"] = (
                "compose deploy did not become live within timeout — build may "
                "still be rolling. Inspect deployment_id or call cp_refresh_agent later."
            )
            if live_card:
                out["last_seen_version"] = live_card.get("version")
        return json.dumps(out)

    @tool
    async def cp_refresh_agent(name: str) -> str:
        """Force the control plane to re-fetch the agent's live
        ``/.well-known/agent-card`` and update its stored copy.

        Use this when the pod was redeployed out-of-band (CI bump, manual
        rollout) and the CP's cached card is stale — symptoms are stale
        ``input_schema`` in ``list_my_agents`` / ``discover_agent`` and
        4xx from ``call_agent`` on what should be a valid skill call.

        ``cp_deploy_tarball`` already does this automatically by polling
        the live card, so you only need ``cp_refresh_agent`` when you
        did NOT deploy through this tool.

        Returns JSON: the fresh ``AgentDetailOut`` (with ``card`` +
        ``skills`` + ``input_schema``), or ``{error, ...}``.
        """
        if not ctx.cp_jwt:
            return json.dumps({
                "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True",
            })
        try:
            async with httpx.AsyncClient(timeout=15.0) as c:
                r = await c.get(
                    f"{settings.cp_url}/v1/agents/{name}",
                    headers={"authorization": f"bearer {ctx.cp_jwt}"},
                )
        except httpx.HTTPError as exc:
            return json.dumps({"error": f"cp unreachable: {exc}"})
        if r.status_code >= 400:
            return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]})
        body = r.json() or {}
        return json.dumps({
            "ok": True,
            "name": body.get("name"),
            "version": body.get("version"),
            "status": body.get("status"),
            "url": body.get("url"),
            "card": body.get("card") or {},
        })

    @tool
    async def sync_agent_workspace_from_repo(name: str) -> str:
        """Replace ``agents/<name>/`` in MinIO with the current managed repo source.

        Use this before editing an existing deployed agent, or after
        ``cp_deploy_tarball`` returns ``workspace_untracked`` /
        ``workspace_drift``. It downloads the owner's current managed repo
        through the control plane, deletes the old MinIO files for that
        agent, writes the repo source files, and records the repo head as the
        deploy base marker. This intentionally overwrites the builder
        workspace copy so it matches the repo.
        """
        try:
            prefix = _agent_prefix(name)
        except ValueError as exc:
            return json.dumps({"error": str(exc)})
        if not ctx.cp_jwt:
            return json.dumps({
                "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True",
            })
        try:
            async with httpx.AsyncClient(timeout=60.0) as c:
                r = await c.get(
                    f"{settings.cp_url}/v1/agents/{name}/source",
                    headers={"authorization": f"bearer {ctx.cp_jwt}"},
                )
        except httpx.HTTPError as exc:
            return json.dumps({"error": f"cp unreachable: {exc}"})
        if r.status_code >= 400:
            return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]})
        head_sha = r.headers.get("x-a2a-repo-head-sha")
        try:
            files = _files_from_tarball(r.content)
        except ValueError as exc:
            return json.dumps({"error": f"invalid source tarball: {exc}"})
        if not files:
            return json.dumps({"error": "managed repo source export was empty"})
        written = _replace_workspace_files(store, prefix, files)
        _write_builder_state(
            store,
            name,
            {
                "agent": name,
                "repo_head_sha": head_sha,
                "file_count": len(written),
                "updated_at": int(time.time()),
                "source": "repo-sync",
            },
        )
        return json.dumps({
            "ok": True,
            "agent": name,
            "repo_head_sha": head_sha,
            "files": written,
        })

    @tool
    def list_a2a_pack(subdir: str = "") -> str:
        """List ``.py`` files under the installed ``a2a_pack`` package.

        Use this when you need to recall what the SDK actually exposes
        — what decorators exist, what ``RunContext`` methods are
        available, how ``WorkspaceClient`` is shaped, what the
        ``AgentRuntime`` fields are, etc. The agent code you scaffold
        runs on this exact package, so reading it is the authoritative
        reference (the system-prompt examples are intentionally
        abridged).

        ``subdir`` is an optional path segment relative to the package
        root (``""``, ``"serve"``, ``"mcp"`` …). Returns JSON:
        ``{root, files: [{path, size}]}``.
        """
        try:
            import a2a_pack
            import os as _os
            base = _os.path.dirname(_os.path.abspath(a2a_pack.__file__))
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": f"a2a_pack not importable: {exc}"})
        if subdir.startswith("/") or ".." in subdir.split("/"):
            return json.dumps({"error": "subdir must be relative; '..' disallowed"})
        root = _os.path.join(base, subdir) if subdir else base
        if not _os.path.commonpath([base, _os.path.abspath(root)]) == base:
            return json.dumps({"error": "subdir escapes package root"})
        if not _os.path.isdir(root):
            return json.dumps({"error": f"not a directory: {subdir}"})
        out: list[dict[str, Any]] = []
        for dirpath, _, filenames in _os.walk(root):
            for fname in filenames:
                if not fname.endswith(".py"):
                    continue
                full = _os.path.join(dirpath, fname)
                rel = _os.path.relpath(full, base)
                try:
                    size = _os.path.getsize(full)
                except OSError:
                    continue
                out.append({"path": rel, "size": size})
        out.sort(key=lambda r: r["path"])
        return json.dumps({"root": base, "files": out})

    @tool
    def read_a2a_pack(path: str) -> str:
        """Read a file from the installed ``a2a_pack`` package.

        Path is relative to the package root, e.g. ``agent.py``,
        ``context.py``, ``runtime.py``, ``workspace.py``, ``card.py``,
        ``mcp/server.py``. Returns JSON: ``{path, content}``. Capped
        at 64 KiB; reach for ``list_a2a_pack`` first if you need to
        chase a deeper file.
        """
        try:
            import a2a_pack
            import os as _os
            base = _os.path.dirname(_os.path.abspath(a2a_pack.__file__))
        except Exception as exc:  # noqa: BLE001
            return json.dumps({"error": f"a2a_pack not importable: {exc}"})
        if path.startswith("/") or ".." in path.split("/"):
            return json.dumps({"error": "path must be relative; '..' disallowed"})
        full = _os.path.abspath(_os.path.join(base, path))
        if _os.path.commonpath([base, full]) != base:
            return json.dumps({"error": "path escapes package root"})
        if not _os.path.isfile(full):
            return json.dumps({"error": f"not a file: {path}"})
        try:
            with open(full, "r", encoding="utf-8") as fh:
                content = fh.read(64 * 1024 + 1)
        except OSError as exc:
            return json.dumps({"error": str(exc)})
        truncated = len(content) > 64 * 1024
        if truncated:
            content = content[: 64 * 1024]
        return json.dumps({
            "path": path,
            "content": content,
            "truncated": truncated,
        })

    return [
        init_agent_template, list_agent_files, write_agent_file, read_agent_file,
        write_agent_skill, test_agent_in_sandbox, cp_deploy_tarball,
        cp_compose_meta_agent, cp_refresh_agent,
        sync_agent_workspace_from_repo,
        list_a2a_pack, read_a2a_pack,
    ]


# --- helpers (not tools) ---


def _tarball_workspace_dir(
    store_or_s3: Any,
    bucket_or_prefix: str,
    prefix: str | None = None,
) -> bytes:
    """Pull every object under ``prefix`` and produce a relative tarball."""
    store, prefix = _store_from_args(store_or_s3, bucket_or_prefix, prefix)
    buf = io.BytesIO()
    added = False
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        for key in store.iter_keys(prefix):
            rel = key[len(prefix):]
            if not rel:
                continue
            if rel == _BUILDER_STATE_FILE or rel.startswith(_BUILDER_INTERNAL_PREFIX):
                continue
            body = store.get(key)
            info = tarfile.TarInfo(name=rel)
            info.size = len(body)
            info.mode = 0o644
            tf.addfile(info, io.BytesIO(body))
            added = True
    return buf.getvalue() if added else b""


def _sanitize_agent_workspace_pricing(
    store: _ObjectStore,
    prefix: str,
) -> tuple[str, ...]:
    key = prefix + "agent.py"
    try:
        source = store.get(key).decode("utf-8", errors="replace")
    except Exception:  # noqa: BLE001
        return ()
    sanitized, fields = _strip_unsupported_pricing_fields(source)
    if fields and sanitized != source:
        store.put(key, sanitized.encode("utf-8"), "text/plain; charset=utf-8")
    return fields


def _strip_unsupported_pricing_fields(source: str) -> tuple[str, tuple[str, ...]]:
    """Remove platform-derived pricing kwargs from generated Agent source."""
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return source, ()

    offsets = _line_offsets(source)
    ranges: list[tuple[int, int, str]] = []
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call) or not _is_pricing_call(node.func):
            continue
        for keyword in node.keywords:
            if keyword.arg not in _DERIVED_PRICING_FIELDS:
                continue
            if (
                keyword.lineno is None
                or keyword.end_lineno is None
                or keyword.col_offset is None
                or keyword.end_col_offset is None
            ):
                continue
            start = offsets[keyword.lineno - 1] + keyword.col_offset
            end = offsets[keyword.end_lineno - 1] + keyword.end_col_offset
            ranges.append((start, end, keyword.arg))

    if not ranges:
        return source, ()

    sanitized = source
    removed: set[str] = set()
    for start, end, field in sorted(ranges, key=lambda item: item[0], reverse=True):
        sanitized = _remove_keyword_source_range(sanitized, start, end)
        removed.add(field)
    return sanitized, tuple(sorted(removed))


def _line_offsets(source: str) -> list[int]:
    offsets: list[int] = []
    offset = 0
    for line in source.splitlines(keepends=True):
        offsets.append(offset)
        offset += len(line)
    return offsets or [0]


def _is_pricing_call(func: ast.expr) -> bool:
    if isinstance(func, ast.Name):
        return func.id == "Pricing"
    if isinstance(func, ast.Attribute):
        return func.attr == "Pricing"
    return False


def _remove_keyword_source_range(source: str, start: int, end: int) -> str:
    line_start = source.rfind("\n", 0, start) + 1
    if source[line_start:start].strip() == "":
        trailing = _skip_horizontal_whitespace(source, end)
        if trailing < len(source) and source[trailing] == ",":
            trailing = _skip_horizontal_whitespace(source, trailing + 1)
        if trailing >= len(source) or source[trailing] in "\r\n":
            if source[trailing: trailing + 2] == "\r\n":
                trailing += 2
            elif trailing < len(source):
                trailing += 1
            return source[:line_start] + source[trailing:]

    trailing = _skip_horizontal_whitespace(source, end)
    if trailing < len(source) and source[trailing] == ",":
        trailing = _skip_horizontal_whitespace(source, trailing + 1)
        if trailing < len(source) and source[trailing] == " ":
            trailing += 1
        return source[:start] + source[trailing:]

    leading = start
    while leading > 0 and source[leading - 1] in " \t":
        leading -= 1
    if leading > 0 and source[leading - 1] == ",":
        return source[: leading - 1] + source[end:]
    return source[:start] + source[end:]


def _skip_horizontal_whitespace(source: str, offset: int) -> int:
    while offset < len(source) and source[offset] in " \t":
        offset += 1
    return offset


def _read_builder_state(store: _ObjectStore, name: str) -> dict[str, Any]:
    try:
        data = json.loads(store.get(_builder_state_key(name)).decode("utf-8"))
    except Exception:  # noqa: BLE001
        return {}
    return data if isinstance(data, dict) else {}


def _write_builder_state(
    store: _ObjectStore,
    name: str,
    state: dict[str, Any],
) -> None:
    body = json.dumps(state, sort_keys=True, indent=2).encode("utf-8")
    store.put(_builder_state_key(name), body, "application/json")


def _parse_manifest_json(raw: str) -> dict[str, Any]:
    try:
        parsed = json.loads(raw or "{}")
    except (TypeError, ValueError) as exc:
        raise ValueError("manifest_json must be a JSON object") from exc
    if not isinstance(parsed, dict):
        raise ValueError("manifest_json must be a JSON object")
    composition = parsed.get("composition")
    if not isinstance(composition, (dict, list)):
        raise ValueError("manifest_json requires a composition object or list")
    return parsed


def _render_skill_md(skill_name: str, description: str, instructions: str) -> str:
    desc = " ".join(description.strip().split())
    body = instructions.strip()
    if not desc:
        raise ValueError("skill description is required")
    if len(desc) > 1024:
        raise ValueError("skill description must be 1024 characters or less")
    if not body:
        raise ValueError("skill instructions are required")
    if "\n---" in body or body.startswith("---"):
        raise ValueError("skill instructions must not contain frontmatter delimiters")
    return (
        "---\n"
        f"name: {skill_name}\n"
        "description: "
        + json.dumps(desc)
        + "\n"
        "---\n"
        f"# {skill_name}\n\n"
        + body
        + "\n"
    )


def _parse_supporting_skill_files(raw: str) -> dict[str, str]:
    try:
        parsed = json.loads(raw or "{}")
    except (TypeError, ValueError) as exc:
        raise ValueError("supporting_files_json must be a JSON object") from exc
    if not isinstance(parsed, dict):
        raise ValueError("supporting_files_json must be a JSON object")
    out: dict[str, str] = {}
    for path, content in parsed.items():
        if not isinstance(path, str) or not isinstance(content, str):
            raise ValueError("supporting skill file paths and contents must be strings")
        rel = _safe_supporting_skill_path(path)
        if rel == "SKILL.md":
            raise ValueError("supporting_files_json cannot override SKILL.md")
        out[rel] = content
    return out


def _safe_supporting_skill_path(path: str) -> str:
    rel = path.replace("\\", "/").lstrip("/")
    while rel.startswith("./"):
        rel = rel[2:]
    if not rel or ".." in rel.split("/") or rel.startswith("."):
        raise ValueError(f"unsafe supporting skill file path: {path!r}")
    if rel.endswith("/") or "//" in rel:
        raise ValueError(f"unsafe supporting skill file path: {path!r}")
    return rel


def _delete_workspace_objects(store: _ObjectStore, prefix: str) -> None:
    for key in store.iter_keys(prefix):
        store.delete(key)


def _files_from_tarball(bundle: bytes) -> dict[str, bytes]:
    out: dict[str, bytes] = {}
    try:
        with tarfile.open(fileobj=io.BytesIO(bundle), mode="r:gz") as tf:
            for member in tf.getmembers():
                if not member.isfile():
                    continue
                rel = member.name
                while rel.startswith("./"):
                    rel = rel[2:]
                if not rel or rel.startswith("/") or ".." in rel.split("/"):
                    raise ValueError(f"unsafe member path: {member.name}")
                if rel == _BUILDER_STATE_FILE:
                    continue
                extracted = tf.extractfile(member)
                if extracted is None:
                    continue
                out[rel] = extracted.read()
    except tarfile.TarError as exc:
        raise ValueError(str(exc)) from exc
    return out


def _replace_workspace_files(
    store_or_s3: Any,
    bucket_or_prefix: str,
    prefix_or_files: str | dict[str, bytes],
    files: dict[str, bytes] | None = None,
) -> list[dict[str, Any]]:
    if files is None:
        if not isinstance(prefix_or_files, dict):
            raise TypeError("files are required")
        store, prefix = _store_from_args(store_or_s3, bucket_or_prefix, None)
        files = prefix_or_files
    else:
        if not isinstance(prefix_or_files, str):
            raise TypeError("prefix must be a string")
        store, prefix = _store_from_args(store_or_s3, bucket_or_prefix, prefix_or_files)
    written: list[dict[str, Any]] = []
    _delete_workspace_objects(store, prefix)
    for rel, body in sorted(files.items()):
        key = prefix + rel
        store.put(key, body, "application/octet-stream")
        written.append({"path": rel, "size": len(body)})
    return written


async def _latest_cp_deployment(
    cp_url: str,
    cp_jwt: str,
    name: str,
) -> dict[str, Any] | None:
    try:
        async with httpx.AsyncClient(timeout=15.0) as c:
            r = await c.get(
                f"{cp_url}/v1/agents/{name}/deployments",
                headers={"authorization": f"bearer {cp_jwt}"},
            )
    except httpx.HTTPError as exc:
        raise RuntimeError(f"cp unreachable: {exc}") from exc
    if r.status_code == 404:
        return None
    if r.status_code >= 400:
        raise RuntimeError(f"cp {r.status_code}: {r.text[:1000]}")
    body = r.json() or []
    if not isinstance(body, list) or not body:
        return None
    first = body[0]
    return first if isinstance(first, dict) else None


def _deployment_drift_error(
    name: str,
    latest_deployment: dict[str, Any] | None,
    builder_state: dict[str, Any],
    *,
    force: bool = False,
) -> dict[str, Any] | None:
    if force or latest_deployment is None:
        return None
    latest_head = latest_deployment.get("head_sha")
    base_head = builder_state.get("repo_head_sha")
    if not isinstance(base_head, str) or not base_head:
        return {
            "ok": False,
            "error": "workspace_untracked",
            "agent": name,
            "message": (
                "This agent already has managed source, but the MinIO "
                "workspace has no deploy base marker. Refresh it before "
                "deploying, or use force=True only if the user wants to "
                "replace the repo."
            ),
            "current_head_sha": latest_head,
            "current_deployment_id": latest_deployment.get("deploy_id"),
        }
    return None


def _read_entrypoint(bundle: bytes) -> str | None:
    try:
        with tarfile.open(fileobj=io.BytesIO(bundle), mode="r:gz") as tf:
            for member in tf.getmembers():
                if member.name.endswith("a2a.yaml") and tf.extractfile(member):
                    text = tf.extractfile(member).read().decode("utf-8", "replace")  # type: ignore[union-attr]
                    for line in text.splitlines():
                        if line.lstrip().startswith("entrypoint:"):
                            return line.split(":", 1)[1].strip().strip('"\'')
    except Exception:  # noqa: BLE001
        pass
    return None


def _read_bundle_text(bundle: bytes, path: str) -> str | None:
    try:
        with tarfile.open(fileobj=io.BytesIO(bundle), mode="r:gz") as tf:
            for member in tf.getmembers():
                if member.name == path and member.isfile():
                    extracted = tf.extractfile(member)
                    if extracted is None:
                        return None
                    return extracted.read().decode("utf-8", "replace")
    except Exception:  # noqa: BLE001
        return None
    return None


def _class_name(slug: str) -> str:
    return "".join(p.capitalize() for p in re.split(r"[-_]+", slug) if p) or "MyAgent"


_FRONTEND_TEMPLATE_FILES: dict[str, dict[str, str]] = {
    "static": {
        "frontend/dist/index.html": "frontend/static-index.html.tmpl",
    },
    "react": {
        "frontend/package.json": "frontend/react-package.json.tmpl",
        "frontend/index.html": "frontend/react-index.html.tmpl",
        "frontend/vite.config.js": "frontend/react-vite.config.js.tmpl",
        "frontend/src/main.jsx": "frontend/react-main.jsx.tmpl",
        "frontend/src/App.jsx": "frontend/react-app.jsx.tmpl",
        "frontend/src/a2a.js": "frontend/react-a2a.js.tmpl",
        "frontend/src/style.css": "frontend/react-style.css.tmpl",
    },
}


def _normalize_frontend_kind(frontend: str | None) -> str:
    kind = (frontend or "none").strip().lower()
    if kind in {"", "no", "off", "false"}:
        kind = "none"
    if kind not in {"none", "static", "react"}:
        raise ValueError("frontend must be one of: none, static, react")
    return kind


def _frontend_block(frontend: str) -> str:
    kind = _normalize_frontend_kind(frontend)
    if kind == "none":
        return ""
    build = "\n  build: npm run build" if kind == "react" else ""
    return (
        "frontend:\n"
        "  path: frontend"
        f"{build}\n"
        "  dist: dist\n"
        "  mount: /app\n"
        "  auth: inherit"
    )


def _render_frontend_files(frontend: str, *, name: str) -> dict[str, str]:
    kind = _normalize_frontend_kind(frontend)
    if kind == "none":
        return {}
    return {
        path: _render_sdk_template(template, name=name)
        for path, template in _FRONTEND_TEMPLATE_FILES[kind].items()
    }


def _render_a2a_init_template(
    name: str,
    *,
    description: str = "A new A2A agent",
    frontend: str = "none",
) -> dict[str, str]:
    _validate_name(name)
    frontend_kind = _normalize_frontend_kind(frontend)
    class_name = _class_name(name)
    files = {
        "agent.py": _render_sdk_template(
            "agent.py.tmpl",
            name=name,
            class_name=class_name,
            description=description,
        ),
        "a2a.yaml": _render_sdk_template(
            "a2a.yaml.tmpl",
            name=name,
            class_name=class_name,
            frontend_block=_frontend_block(frontend_kind),
        ),
        "requirements.txt": _render_sdk_template("requirements.txt.tmpl"),
    }
    files.update(_render_frontend_files(frontend_kind, name=name))
    return files


def _render_sdk_template(template: str, /, **vars: str) -> str:
    parts = template.split("/")
    local_templates = (
        Path(__file__).resolve().parents[2]
        / "a2a"
        / "a2a_pack"
        / "cli"
        / "templates"
    )
    if local_templates.exists():
        text = local_templates.joinpath(*parts).read_text(encoding="utf-8")
    else:
        resource = resources.files("a2a_pack.cli.templates")
        for part in parts:
            resource = resource.joinpath(part)
        text = resource.read_text(encoding="utf-8")
    for key, value in vars.items():
        text = text.replace("{{ " + key + " }}", value)
    return text