#!/usr/bin/env python3
"""
botje.py — DevPlace bot with full X-agent capabilities.

Merges:
  • bot.py        — ChromeStealthClient, all @tool functions, LLM agent loop
  • docs.md       — DevPlace XML-RPC API client, mention/DM bot patterns

When invoked without arguments it runs as a DevPlace bot, polling for @mentions
and direct messages and answering them with the full X-agent tool set (web
search, deep research, file operations, etc.).  When invoked with --prompt or
a positional prompt it runs as a normal X-agent, executes the prompt and exits.

Usage:
  python botje.py                         # DevPlace bot (polling loop)
  python botje.py "Deep research cat pictures"   # one-shot X-agent task
  python botje.py -p "deploy my project"  # same with --prompt
"""

from __future__ import annotations
import httpx
import argparse
import ast
import asyncio
import base64
import collections
import contextvars
import functools
import hashlib
import inspect
import json
import logging
import math
import mimetypes
import os
import re
import ssl
import sys
import uuid
import xmlrpc.client
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Optional, Sequence
from urllib.parse import quote, unquote, urlencode, urlparse, urlsplit, urlunsplit

# ═══════════════════════════════════════════════════════════════════════════════
#  Configuration
# ═══════════════════════════════════════════════════════════════════════════════

def _resolve_devplace_url() -> str:
    base = os.environ.get("DEVPLACE_BASE_URL", "").strip().rstrip("/")
    if base:
        return base
    return os.environ.get("DEVPLACE_URL", "").strip().rstrip("/")


def _resolve_llm_endpoint() -> str:
    base = os.environ.get("DEVPLACE_OPENAI_URL", "").strip().rstrip("/")
    if not base:
        base = "https://openai.app.molodetz.nl/v1"
    return base if base.endswith("/chat/completions") else base + "/chat/completions"


DEVPLACE_URL = _resolve_devplace_url()
DEVPLACE_API_KEY = os.environ.get("DEVPLACE_API_KEY", "").strip()
MENTION_POLL_SECONDS = int(os.environ.get("MENTION_POLL_SECONDS", "30"))
DM_POLL_SECONDS = int(os.environ.get("DM_POLL_SECONDS", "10"))
BOT_USERNAME = os.environ.get("BOT_USERNAME", "")

COMMENT_CHAR_LIMIT = 125000
MESSAGE_CHAR_LIMIT = 2000
PART_SUFFIX_RESERVE = 12
PART_DELIVERY_DELAY = 0.5

LLM_ENDPOINT = _resolve_llm_endpoint()
LLM_BASE_URL = LLM_ENDPOINT.rsplit("/chat/completions", 1)[0]
MODEL = "molodetz"
LLM_API_KEY = str(
    os.environ.get("DEVPLACE_API_KEY")
    or os.environ.get("LLM_API_KEY")
    or ""
)
if not LLM_API_KEY:
    LLM_API_KEY = str(uuid.uuid4())

_BOOT_DT = datetime.now().astimezone()
BOOT_DATETIME = _BOOT_DT.isoformat()
BOOT_DAY_NAME = _BOOT_DT.strftime("%A")

RSEARCH_BASE_URL = "https://rsearch.app.molodetz.nl"
RSEARCH_TIMEOUT = 300
RSEARCH_MAX_BYTES = 8 * 1024 * 1024

LLM_HTTP_TIMEOUT = 600
LLM_MAX_RETRIES = 3
LLM_RETRY_BACKOFF = 2.0
DEFAULT_COMMAND_TIMEOUT = 300
DEFAULT_HTTP_TIMEOUT = 120

CONTEXT_COMPACT_THRESHOLD_CHARS = 500_000
CONTEXT_KEEP_TAIL_MESSAGES = 14
MAX_ITERATIONS = 1000
DELEGATE_MAX_ITERATIONS = 60
DEVPLACE_MAX_ITERATIONS = int(os.environ.get("DEVPLACE_MAX_ITERATIONS", str(MAX_ITERATIONS)))
OUTPUT_CAP_BYTES = 256 * 1024
TOOL_ARG_PREVIEW = 220
IMAGE_MAX_BYTES = 20 * 1024 * 1024

WORKDIR = Path.cwd()

INDEX_EXTS = (
    ".py", ".js", ".ts", ".tsx", ".jsx", ".md", ".html", ".css",
    ".json", ".yaml", ".yml", ".toml", ".rs", ".go", ".java",
    ".c", ".h", ".cpp", ".hpp", ".sh", ".rb", ".php", ".lua", ".sql",
)
INDEX_SKIP_DIRS = {
    ".git", "__pycache__", "node_modules", ".venv", "venv",
    "dist", "build", ".cache", ".mypy_cache", ".pytest_cache",
    ".tox", ".idea", ".vscode", "target", "out",
}
INDEX_MAX_FILE_BYTES = 512 * 1024

SWARM_DIR = Path("/tmp") / "x_swarm"
POST_SLUG_RE = re.compile(r"/posts/([^/#?]+)")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("botje")

# ═══════════════════════════════════════════════════════════════════════════════
#  DevPlace XML-RPC Client (from docs.md)
# ═══════════════════════════════════════════════════════════════════════════════


class DevPlace:
    """Thin XML-RPC client for DevPlace (from docs.md)."""

    def __init__(
        self,
        base_url: str,
        api_key: str | None = None,
        username: str | None = None,
        password: str | None = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        endpoint = f"{self.base_url}/xmlrpc"
        if username and password:
            parts = urlsplit(self.base_url)
            creds = f"{quote(username, safe='')}:{quote(password, safe='')}"
            endpoint = urlunsplit((parts.scheme, f"{creds}@{parts.netloc}", "/xmlrpc", "", ""))
        self.proxy = xmlrpc.client.ServerProxy(endpoint, allow_none=True)

    def call(self, method: str, **params: Any) -> Any:
        if self.api_key and "api_key" not in params:
            params["api_key"] = self.api_key
        return getattr(self.proxy, method)(params)


# ═══════════════════════════════════════════════════════════════════════════════
#  ChromeStealthClient (from bot.py)
# ═══════════════════════════════════════════════════════════════════════════════


def _supported_accept_encoding() -> str:
    encodings = ["gzip", "deflate"]
    try:
        import brotli  # noqa: F401
        encodings.append("br")
    except ImportError:
        logger.debug("brotli not installed; not advertising 'br'")
    try:
        import zstandard  # noqa: F401
        encodings.append("zstd")
    except ImportError:
        logger.debug("zstandard not installed; not advertising 'zstd'")
    return ", ".join(encodings)


ACCEPT_ENCODING: str = _supported_accept_encoding()

CHROME_CIPHER_SUITES: str = ":".join(
    [
        "TLS_AES_128_GCM_SHA256",
        "TLS_AES_256_GCM_SHA384",
        "TLS_CHACHA20_POLY1305_SHA256",
        "ECDHE-ECDSA-AES128-GCM-SHA256",
        "ECDHE-RSA-AES128-GCM-SHA256",
        "ECDHE-ECDSA-AES256-GCM-SHA384",
        "ECDHE-RSA-AES256-GCM-SHA384",
        "ECDHE-ECDSA-CHACHA20-POLY1305",
        "ECDHE-RSA-CHACHA20-POLY1305",
        "ECDHE-RSA-AES128-SHA",
        "ECDHE-RSA-AES256-SHA",
        "AES128-GCM-SHA256",
        "AES256-GCM-SHA384",
        "AES128-SHA",
        "AES256-SHA",
    ]
)

ALPN_PROTOCOLS: tuple[str, ...] = ("h2", "http/1.1")
DEFAULT_CHROME_VERSION: str = "131"
DEFAULT_CHUNK_SIZE: int = 65536
IN_MEMORY_DOWNLOAD_CAP: int = 100 * 1024 * 1024

IMAGE_EXTENSIONS: tuple[str, ...] = (
    ".jpg", ".jpeg", ".png", ".gif", ".webp", ".avif",
    ".bmp", ".svg", ".ico", ".tiff",
)

MAX_FILENAME_LENGTH: int = 200
MAX_CONCURRENT_DOWNLOADS: int = 64


@dataclass(frozen=True)
class ChromeProfile:
    version: str = DEFAULT_CHROME_VERSION
    platform: str = "Windows"
    platform_token: str = "Windows NT 10.0; Win64; x64"
    accept_language: str = "en-US,en;q=0.9"

    @property
    def user_agent(self) -> str:
        return (
            f"Mozilla/5.0 ({self.platform_token}) AppleWebKit/537.36 "
            f"(KHTML, like Gecko) Chrome/{self.version}.0.0.0 Safari/537.36"
        )

    @property
    def sec_ch_ua(self) -> str:
        return (
            f'"Google Chrome";v="{self.version}", '
            f'"Chromium";v="{self.version}", '
            f'"Not_A Brand";v="24"'
        )

    @classmethod
    def windows(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
        return cls(version=version, platform="Windows", platform_token="Windows NT 10.0; Win64; x64")

    @classmethod
    def macos(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
        return cls(
            version=version,
            platform="macOS",
            platform_token="Macintosh; Intel Mac OS X 10_15_7",
        )

    @classmethod
    def linux(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
        return cls(version=version, platform="Linux", platform_token="X11; Linux x86_64")


@dataclass
class DownloadResult:
    url: str
    path: Path | None
    status_code: int | None
    bytes_written: int
    ok: bool
    error: str | None = None


@dataclass
class BytesResult:
    url: str
    final_url: str
    status_code: int
    content_type: str
    data: bytes
    truncated: bool
    error: str | None = None


def resource_kind_for_url(url: str) -> str:
    path = urlsplit(url).path.lower()
    return "image" if path.endswith(IMAGE_EXTENSIONS) else "empty"


def build_chrome_ssl_context() -> ssl.SSLContext:
    context = ssl.create_default_context()
    context.check_hostname = True
    context.verify_mode = ssl.CERT_REQUIRED
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    context.maximum_version = ssl.TLSVersion.TLSv1_3
    context.set_ciphers(CHROME_CIPHER_SUITES)
    context.set_alpn_protocols(list(ALPN_PROTOCOLS))
    context.options |= ssl.OP_NO_COMPRESSION
    logger.debug(
        "Built Chrome-aligned SSL context with %d ciphers",
        CHROME_CIPHER_SUITES.count(":") + 1,
    )
    return context


def slugify_filename(name: str) -> str:
    name = unquote(name).strip()
    name = name.replace("\\", "/").split("/")[-1]
    name = name.split("?")[0].split("#")[0]
    name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-._")
    if not name:
        name = "download"
    if len(name) > MAX_FILENAME_LENGTH:
        stem, _, suffix = name.rpartition(".")
        if stem and len(suffix) <= 16:
            keep = MAX_FILENAME_LENGTH - len(suffix) - 1
            name = f"{stem[:keep]}.{suffix}"
        else:
            name = name[:MAX_FILENAME_LENGTH]
    return name


def resolve_destination(directory: Path, url: str, override_name: str | None) -> Path:
    directory = directory.resolve()
    raw_name = override_name if override_name else Path(urlsplit(url).path).name
    safe_name = slugify_filename(raw_name)
    candidate = (directory / safe_name).resolve()
    if directory != candidate.parent:
        raise ValueError(f"Refusing path traversal for {url!r} -> {candidate}")
    return candidate


class ChromeStealthClient:

    def __init__(
        self,
        profile: ChromeProfile | None = None,
        *,
        http2: bool = True,
        timeout: float = 30.0,
        follow_redirects: bool = True,
        proxy: str | None = None,
        trust_env: bool = True,
        max_connections: int = 100,
        max_keepalive_connections: int = 20,
    ) -> None:
        self.profile = profile or ChromeProfile.windows()
        self._ssl_context = build_chrome_ssl_context()
        limits = httpx.Limits(
            max_connections=max_connections,
            max_keepalive_connections=max_keepalive_connections,
        )
        self._client = httpx.AsyncClient(
            http2=http2,
            verify=self._ssl_context,
            timeout=httpx.Timeout(timeout),
            follow_redirects=follow_redirects,
            limits=limits,
            proxy=proxy,
            trust_env=trust_env,
            headers=self._base_headers(),
        )
        logger.info(
            "ChromeStealthClient ready: Chrome %s on %s, http2=%s, proxy=%s",
            self.profile.version,
            self.profile.platform,
            http2,
            bool(proxy),
        )

    def _base_headers(self) -> dict[str, str]:
        return {
            "sec-ch-ua": self.profile.sec_ch_ua,
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": f'"{self.profile.platform}"',
            "user-agent": self.profile.user_agent,
            "accept-encoding": ACCEPT_ENCODING,
            "accept-language": self.profile.accept_language,
        }

    def _navigation_headers(self, referer: str | None) -> dict[str, str]:
        headers = {
            "upgrade-insecure-requests": "1",
            "accept": (
                "text/html,application/xhtml+xml,application/xml;q=0.9,"
                "image/avif,image/webp,image/apng,*/*;q=0.8,"
                "application/signed-exchange;v=b3;q=0.7"
            ),
            "sec-fetch-site": "same-origin" if referer else "none",
            "sec-fetch-mode": "navigate",
            "sec-fetch-user": "?1",
            "sec-fetch-dest": "document",
        }
        if referer:
            headers["referer"] = referer
        return headers

    def _resource_headers(self, dest: str, referer: str | None) -> dict[str, str]:
        accept = {
            "image": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",
            "empty": "*/*",
        }.get(dest, "*/*")
        headers = {
            "accept": accept,
            "sec-fetch-site": "same-origin" if referer else "cross-site",
            "sec-fetch-mode": "no-cors" if dest == "image" else "cors",
            "sec-fetch-dest": dest,
        }
        if referer:
            headers["referer"] = referer
        return headers

    def _api_headers(self, referer: str | None) -> dict[str, str]:
        headers = {
            "accept": "application/json, text/plain, */*",
            "sec-fetch-site": "same-origin" if referer else "cross-site",
            "sec-fetch-mode": "cors",
            "sec-fetch-dest": "empty",
        }
        if referer:
            headers["referer"] = referer
        return headers

    @staticmethod
    def _json_or_error(response: Any) -> dict[str, object]:
        if response.status_code >= 400:
            return {
                "error": f"HTTP {response.status_code}",
                "status_code": response.status_code,
                "body": response.text[:5000],
            }
        try:
            return response.json()
        except ValueError:
            return {
                "error": "Invalid JSON response",
                "status_code": response.status_code,
                "body": response.text[:5000],
            }

    async def __aenter__(self) -> "ChromeStealthClient":
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.aclose()

    async def aclose(self) -> None:
        await self._client.aclose()
        logger.debug("ChromeStealthClient closed")

    async def _send(self, method: str, url: str, headers: dict[str, str], **kwargs: object) -> Any:
        try:
            return await self._client.request(method, url, headers=headers, **kwargs)
        except httpx.DecodingError:
            logger.warning("content decode failed for %s; retrying without br/zstd", url)
            retry_headers = dict(headers)
            retry_headers["accept-encoding"] = "gzip, deflate"
            return await self._client.request(method, url, headers=retry_headers, **kwargs)

    async def request(
        self,
        method: str,
        url: str,
        *,
        referer: str | None = None,
        headers: Mapping[str, str] | None = None,
        **kwargs: object,
    ) -> Any:
        merged = self._navigation_headers(referer)
        if headers:
            merged.update({key.lower(): value for key, value in headers.items()})
        logger.info("%s %s", method.upper(), url)
        response = await self._send(method, url, merged, **kwargs)
        logger.debug(
            "%s %s -> %s (%s)",
            method.upper(), url, response.status_code, response.http_version,
        )
        return response

    async def get(self, url: str, *, referer: str | None = None, **kwargs: object) -> Any:
        return await self.request("GET", url, referer=referer, **kwargs)

    async def post(self, url: str, *, referer: str | None = None, **kwargs: object) -> Any:
        return await self.request("POST", url, referer=referer, **kwargs)

    def stream(
        self,
        method: str,
        url: str,
        *,
        referer: str | None = None,
        dest: str = "empty",
        headers: Mapping[str, str] | None = None,
        **kwargs: object,
    ):
        merged = self._resource_headers(dest, referer)
        if headers:
            merged.update({key.lower(): value for key, value in headers.items()})
        return self._client.stream(method, url, headers=merged, **kwargs)

    async def download_file(
        self,
        url: str,
        destination: str | Path,
        *,
        filename: str | None = None,
        referer: str | None = None,
        chunk_size: int = DEFAULT_CHUNK_SIZE,
        dest_type: str = "empty",
    ) -> DownloadResult:
        directory = Path(destination)
        directory.mkdir(parents=True, exist_ok=True)
        try:
            target = resolve_destination(directory, url, filename)
        except ValueError as error:
            logger.error("Destination rejected for %s: %s", url, error)
            return DownloadResult(
                url=url, path=None, status_code=None, bytes_written=0, ok=False, error=str(error),
            )

        written = 0
        try:
            async with self.stream("GET", url, referer=referer, dest=dest_type) as response:
                response.raise_for_status()
                with target.open("wb") as handle:
                    async for chunk in response.aiter_bytes(chunk_size):
                        handle.write(chunk)
                        written += len(chunk)
            logger.info("Downloaded %s -> %s (%d bytes)", url, target, written)
            return DownloadResult(
                url=url, path=target, status_code=response.status_code,
                bytes_written=written, ok=True,
            )
        except httpx.HTTPStatusError as error:
            logger.error("HTTP %s downloading %s", error.response.status_code, url)
            return DownloadResult(
                url=url, path=None, status_code=error.response.status_code,
                bytes_written=written, ok=False, error=f"HTTP {error.response.status_code}",
            )
        except httpx.HTTPError as error:
            logger.error("Transport error downloading %s: %s", url, error)
            return DownloadResult(
                url=url, path=None, status_code=None,
                bytes_written=written, ok=False, error=str(error),
            )

    async def download_files(
        self,
        urls: Iterable[str] | Mapping[str, str],
        destination: str | Path,
        *,
        concurrency: int = 8,
        referer: str | None = None,
        chunk_size: int = DEFAULT_CHUNK_SIZE,
        dest_type: str = "empty",
    ) -> list[DownloadResult]:
        if isinstance(urls, Mapping):
            jobs: Sequence[tuple[str, str | None]] = [(url, name) for url, name in urls.items()]
        else:
            jobs = [(url, None) for url in urls]

        bounded = max(1, min(concurrency, MAX_CONCURRENT_DOWNLOADS))
        semaphore = asyncio.Semaphore(bounded)
        logger.info("Starting bulk download of %d files with concurrency %d", len(jobs), bounded)

        async def worker(url: str, name: str | None) -> DownloadResult:
            async with semaphore:
                return await self.download_file(
                    url, destination, filename=name, referer=referer,
                    chunk_size=chunk_size, dest_type=dest_type,
                )

        results = await asyncio.gather(*(worker(url, name) for url, name in jobs))
        succeeded = sum(1 for result in results if result.ok)
        logger.info("Bulk download finished: %d succeeded, %d failed", succeeded, len(results) - succeeded)
        return list(results)

    async def download_bytes(
        self,
        url: str,
        *,
        referer: str | None = None,
        max_bytes: int = IN_MEMORY_DOWNLOAD_CAP,
        chunk_size: int = DEFAULT_CHUNK_SIZE,
        dest_type: str | None = None,
    ) -> BytesResult:
        kind = dest_type if dest_type else resource_kind_for_url(url)
        buffer = bytearray()
        try:
            async with self.stream("GET", url, referer=referer, dest=kind) as response:
                content_type = response.headers.get("content-type", "")
                async for chunk in response.aiter_bytes(chunk_size):
                    buffer.extend(chunk)
                    if len(buffer) > max_bytes:
                        break
                final_url = str(response.url)
            truncated = len(buffer) > max_bytes
            data = bytes(buffer[:max_bytes])
            logger.debug(
                "download_bytes %s -> %d bytes (status %s)",
                url, len(data), response.status_code,
            )
            return BytesResult(
                url=url, final_url=final_url, status_code=response.status_code,
                content_type=content_type, data=data, truncated=truncated,
            )
        except httpx.HTTPError as error:
            logger.error("download_bytes failed for %s: %s", url, error)
            return BytesResult(
                url=url, final_url=url, status_code=0,
                content_type="", data=b"", truncated=False, error=str(error),
            )

    async def post_json(
        self,
        url: str,
        payload: object,
        *,
        headers: Mapping[str, str] | None = None,
        referer: str | None = None,
        timeout: float | None = None,
    ) -> dict[str, object]:
        request_headers = self._api_headers(referer)
        if headers:
            request_headers.update({key.lower(): value for key, value in headers.items()})
        kwargs: dict[str, object] = {"json": payload}
        if timeout is not None:
            kwargs["timeout"] = httpx.Timeout(timeout)
        logger.info("POST(json) %s", url)
        try:
            response = await self._send("POST", url, request_headers, **kwargs)
        except httpx.HTTPError as error:
            logger.error("post_json transport error for %s: %s", url, error)
            return {"error": f"{type(error).__name__}: {error}"}
        return self._json_or_error(response)

    async def get_json(
        self,
        url: str,
        *,
        referer: str | None = None,
        timeout: float | None = None,
    ) -> dict[str, object]:
        kwargs: dict[str, object] = {}
        if timeout is not None:
            kwargs["timeout"] = httpx.Timeout(timeout)
        logger.info("GET(json) %s", url)
        try:
            response = await self._send("GET", url, self._api_headers(referer), **kwargs)
        except httpx.HTTPError as error:
            logger.error("get_json transport error for %s: %s", url, error)
            return {"error": f"{type(error).__name__}: {error}"}
        return self._json_or_error(response)


# ═══════════════════════════════════════════════════════════════════════════════
#  MarkdownRenderer
# ═══════════════════════════════════════════════════════════════════════════════


class MarkdownRenderer:
    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"
    ITALIC = "\033[3m"
    UNDERLINE = "\033[4m"
    RED = "\033[31m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    BLUE = "\033[34m"
    MAGENTA = "\033[35m"
    CYAN = "\033[36m"
    WHITE = "\033[37m"
    GRAY = "\033[90m"
    BG_CODE = "\033[48;5;235m"

    STYLES = {
        "h1": f"{BOLD}{CYAN}",
        "h2": f"{BOLD}{BLUE}",
        "h3": f"{BOLD}{YELLOW}",
        "h4": f"{BOLD}{GREEN}",
        "code": f"{BG_CODE}{GREEN}",
        "blockquote": f"{DIM}{GRAY}",
        "list_marker": f"{CYAN}",
        "hr": f"{DIM}{WHITE}",
        "bold": f"{BOLD}",
        "italic": f"{ITALIC}",
        "link": f"{UNDERLINE}{BLUE}",
        "dim": f"{DIM}{GRAY}",
        "ok": f"{GREEN}",
        "err": f"{RED}",
        "warn": f"{YELLOW}",
    }

    def __init__(self, use_color: bool = True) -> None:
        self.use_color = bool(use_color) and sys.stdout.isatty()

    def _c(self, style_key: str, text: str = "") -> str:
        if not self.use_color:
            return text
        return f"{self.STYLES.get(style_key, '')}{text}{self.RESET}"

    def _render_inline(self, text: str) -> str:
        if not text:
            return text
        text = re.sub(r"`([^`]+)`", lambda m: self._c("code", m.group(1)), text)
        text = re.sub(
            r"\*\*(.+?)\*\*|__(.+?)__",
            lambda m: self._c("bold", m.group(1) or m.group(2)), text,
        )
        text = re.sub(r"~~(.+?)~~", lambda m: self._c("dim", m.group(1)), text)
        text = re.sub(
            r"\[([^\]]+)\]\(([^)]+)\)",
            lambda m: f"{self._c('link', m.group(1))} ({self._c('dim', m.group(2))})",
            text,
        )
        return text

    def _render_code_block(self, code: str, lang: str = "") -> str:
        lines = code.rstrip("\n").split("\n")
        prefix = self._c("code", " ▎")
        header = f"\n{self._c('dim', f' ── {lang} ──')}\n" if lang else "\n"
        body = "\n".join(f"{prefix}{line}" for line in lines)
        return f"{header}{body}\n"

    def render(self, text: str) -> str:
        if not text:
            return text
        lines = text.split("\n")
        out: list[str] = []
        in_code = False
        code_buf: list[str] = []
        code_lang = ""
        for line in lines:
            if line.startswith("```"):
                if in_code:
                    out.append(self._render_code_block("\n".join(code_buf), code_lang))
                    code_buf, code_lang, in_code = [], "", False
                else:
                    in_code, code_lang = True, line[3:].strip()
                continue
            if in_code:
                code_buf.append(line)
                continue
            if not line.strip():
                out.append("")
                continue
            heading = re.match(r"^(#{1,6})\s+(.+)$", line)
            if heading:
                level = min(len(heading.group(1)), 4)
                out.append(
                    f"\n{self._c(f'h{level}', '#' * level + ' ' + self._render_inline(heading.group(2)))}\n",
                )
                continue
            if re.match(r"^[-*_]{3,}$", line.strip()):
                out.append(self._c("hr", "─" * 60))
                continue
            if line.startswith(">"):
                out.append(
                    f"{self._c('blockquote', '│')} {self._render_inline(re.sub(r'^>\s?', '', line))}",
                )
                continue
            list_item = re.match(r"^(\s*)([-*+]|\d+\.)\s+(.+)$", line)
            if list_item:
                out.append(
                    f"{list_item.group(1)}{self._c('list_marker', list_item.group(2))} "
                    f"{self._render_inline(list_item.group(3))}",
                )
                continue
            out.append(self._render_inline(line))
        return "\n".join(out)

    def print(self, text: str) -> None:
        print(self.render(text))


# ═══════════════════════════════════════════════════════════════════════════════
#  Utility functions
# ═══════════════════════════════════════════════════════════════════════════════


def _truncate_output(text: str, cap: int = OUTPUT_CAP_BYTES) -> str:
    raw = text.encode("utf-8", errors="replace")
    if len(raw) <= cap:
        return text
    head = raw[: cap // 2].decode("utf-8", errors="replace")
    tail = raw[-cap // 2:].decode("utf-8", errors="replace")
    return f"{head}\n…[truncated {len(raw) - cap} bytes]…\n{tail}"


def _sha(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()


SENTENCE_BOUNDARY = re.compile(r"(?<=[.!?])\s+")


def _hard_chunks(text: str, limit: int) -> list[str]:
    return [text[index : index + limit] for index in range(0, len(text), limit)]


def _pack(units: list[str], limit: int, glue: str) -> list[str]:
    parts: list[str] = []
    current = ""
    for unit in units:
        if not current:
            current = unit
            continue
        candidate = current + glue + unit
        if len(candidate) <= limit:
            current = candidate
        else:
            parts.append(current)
            current = unit
    if current:
        parts.append(current)
    return parts


def _split_sentences(line: str, limit: int) -> list[str]:
    units: list[str] = []
    for sentence in SENTENCE_BOUNDARY.split(line):
        if len(sentence) <= limit:
            units.append(sentence)
            continue
        words: list[str] = []
        for word in sentence.split(" "):
            words.extend([word] if len(word) <= limit else _hard_chunks(word, limit))
        units.extend(_pack(words, limit, " "))
    return _pack(units, limit, " ")


def _split_text_block(block: str, limit: int) -> list[str]:
    units: list[str] = []
    for line in block.split("\n"):
        if len(line) <= limit:
            units.append(line)
        else:
            units.extend(_split_sentences(line, limit))
    return _pack(units, limit, "\n")


def _is_code_block(block: str) -> bool:
    return block.lstrip().startswith("```")


def _split_code_block(block: str, limit: int) -> list[str]:
    lines = block.split("\n")
    open_fence = lines[0]
    if len(lines) > 1 and lines[-1].strip().startswith("```"):
        close_fence = lines[-1]
        body = lines[1:-1]
    else:
        close_fence = "```"
        body = lines[1:]
    budget = max(1, limit - len(open_fence) - len(close_fence) - 2)
    safe: list[str] = []
    for line in body:
        safe.extend([line] if len(line) <= budget else _hard_chunks(line, budget))
    groups = _pack(safe, budget, "\n")
    return [f"{open_fence}\n{group}\n{close_fence}" for group in groups]


def _markdown_blocks(text: str) -> list[str]:
    blocks: list[str] = []
    buffer: list[str] = []
    in_code = False

    def flush() -> None:
        if buffer:
            blocks.append("\n".join(buffer))
            buffer.clear()

    for line in text.split("\n"):
        if line.lstrip().startswith("```"):
            if in_code:
                buffer.append(line)
                flush()
                in_code = False
            else:
                flush()
                buffer.append(line)
                in_code = True
            continue
        if in_code:
            buffer.append(line)
            continue
        if not line.strip():
            flush()
            continue
        buffer.append(line)
    flush()
    return blocks


def _pack_blocks(text: str, limit: int) -> list[str]:
    parts: list[str] = []
    current = ""
    for block in _markdown_blocks(text):
        if len(block) <= limit:
            pieces = [block]
        elif _is_code_block(block):
            pieces = _split_code_block(block, limit)
        else:
            pieces = _split_text_block(block, limit)
        for piece in pieces:
            if not current:
                current = piece
            elif len(current) + 2 + len(piece) <= limit:
                current = f"{current}\n\n{piece}"
            else:
                parts.append(current)
                current = piece
    if current:
        parts.append(current)
    return parts


def split_reply(text: str, limit: int) -> list[str]:
    """Split a long reply into several posts at markdown/sentence boundaries.

    Each returned part stays within limit. Fenced code blocks are never broken
    mid-block unless a single block exceeds the limit (then it is re-fenced per
    part), sentences and words are never cut while any softer boundary remains,
    and multi-part replies are tagged "(i/N)" the way a person would.
    """
    text = (text or "").strip()
    if len(text) <= limit:
        return [text]
    parts = _pack_blocks(text, limit - PART_SUFFIX_RESERVE)
    if len(parts) <= 1:
        return parts
    total = len(parts)
    return [f"{part}\n\n({index}/{total})" for index, part in enumerate(parts, 1)]


async def _deliver_in_parts(send: Callable[[str], Any], text: str, limit: int) -> int:
    parts = split_reply(text, limit)
    delivered = 0
    for index, part in enumerate(parts):
        body = part.strip()
        if not body:
            continue
        send(body)
        delivered += 1
        if index + 1 < len(parts):
            await asyncio.sleep(PART_DELIVERY_DELAY)
    return delivered


async def stream_subprocess(
    argv: list[str],
    timeout: int | None = None,
    prefix: str = "",
    stdout_sink: Any = None,
    stderr_sink: Any = None,
) -> tuple[str, str, int | None, bool]:
    proc = await asyncio.create_subprocess_exec(
        *argv,
        stdin=asyncio.subprocess.DEVNULL,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout_buf: list[str] = []
    stderr_buf: list[str] = []

    async def consume(stream: Any, buf: list[str], sink: Any, color: str) -> None:
        use_color = bool(color) and sink is not None and sink.isatty()
        reset = MarkdownRenderer.RESET if use_color else ""
        open_color = color if use_color else ""
        while True:
            line = await stream.readline()
            if not line:
                break
            text = line.decode("utf-8", errors="replace")
            buf.append(text)
            if sink is not None:
                sink.write(f"{open_color}{prefix}{text.rstrip(chr(10))}{reset}\n")
                sink.flush()

    out_task = asyncio.create_task(consume(proc.stdout, stdout_buf, stdout_sink, ""))
    err_task = asyncio.create_task(consume(proc.stderr, stderr_buf, stderr_sink, MarkdownRenderer.RED))

    timed_out = False
    try:
        returncode = await asyncio.wait_for(proc.wait(), timeout=timeout)
    except asyncio.TimeoutError:
        timed_out = True
        try:
            proc.kill()
        except ProcessLookupError:
            pass
        await proc.wait()
        returncode = proc.returncode

    await out_task
    await err_task
    return (
        _truncate_output("".join(stdout_buf)),
        _truncate_output("".join(stderr_buf)),
        returncode,
        timed_out,
    )


# ═══════════════════════════════════════════════════════════════════════════════
#  LLM / Backend layer (from bot.py)
# ═══════════════════════════════════════════════════════════════════════════════

_http_client: ChromeStealthClient | None = None
_http_lock: asyncio.Lock | None = None


async def get_http_client() -> ChromeStealthClient:
    global _http_client, _http_lock
    if _http_lock is None:
        _http_lock = asyncio.Lock()
    async with _http_lock:
        if _http_client is None:
            _http_client = ChromeStealthClient(timeout=float(LLM_HTTP_TIMEOUT))
        return _http_client


async def close_http_client() -> None:
    global _http_client
    if _http_client is not None:
        await _http_client.aclose()
        _http_client = None


@dataclass(frozen=True)
class Backend:
    name: str
    endpoint: str
    model: str
    api_key: str | None
    vision: bool


_backends: list[Backend] | None = None
_active_backend: int = 0


def get_backends() -> list[Backend]:
    global _backends
    if _backends is None:
        _backends = [
            Backend("molodetz", LLM_ENDPOINT, MODEL, LLM_API_KEY, True),
        ]
    return _backends


def vision_available() -> bool:
    backends = get_backends()
    return _active_backend == 0 and bool(backends[0].api_key)


async def _call_backend(
    backend: Backend,
    messages: list[dict[str, Any]],
    tools: list[dict[str, Any]] | None = None,
    tool_choice: str = "auto",
    temperature: float = 0.0,
    timeout: int = LLM_HTTP_TIMEOUT,
) -> dict[str, Any]:
    if not backend.api_key:
        raise RuntimeError(f"{backend.name}: no API key configured")
    client = await get_http_client()
    payload: dict[str, Any] = {
        "model": backend.model,
        "messages": messages,
        "temperature": temperature,
    }
    if tools:
        payload["tools"] = tools
        payload["tool_choice"] = tool_choice
    headers = {"authorization": f"Bearer {backend.api_key}"}
    last_error = "unknown error"
    for attempt in range(1, LLM_MAX_RETRIES + 1):
        result = await client.post_json(
            backend.endpoint, payload, headers=headers, timeout=float(timeout),
        )
        if isinstance(result, dict) and result.get("choices"):
            return result
        last_error = (
            json.dumps(result)[:1000] if isinstance(result, dict) else str(result)
        )
        logger.warning(
            "Backend %s attempt %d/%d failed: %s",
            backend.name, attempt, LLM_MAX_RETRIES, last_error,
        )
        if attempt < LLM_MAX_RETRIES:
            await asyncio.sleep(LLM_RETRY_BACKOFF * attempt)
    raise RuntimeError(
        f"{backend.name} failed after {LLM_MAX_RETRIES} attempts: {last_error}",
    )


async def llm_call(
    messages: list[dict[str, Any]],
    tools: list[dict[str, Any]] | None = None,
    tool_choice: str = "auto",
    temperature: float = 0.0,
    timeout: int = LLM_HTTP_TIMEOUT,
) -> dict[str, Any]:
    global _active_backend
    backends = get_backends()
    last_error = "no usable backend"
    for idx in range(_active_backend, len(backends)):
        backend = backends[idx]
        if not backend.api_key:
            last_error = f"{backend.name}: no API key"
            continue
        try:
            result = await _call_backend(
                backend, messages, tools, tool_choice, temperature, timeout,
            )
        except RuntimeError as e:
            last_error = str(e)
            logger.warning("Backend %s unavailable: %s", backend.name, e)
            continue
        if idx != _active_backend:
            logger.warning(
                "LLM backend switched %s -> %s (vision=%s)",
                backends[_active_backend].name, backend.name, backend.vision,
            )
            _active_backend = idx
        return result
    raise RuntimeError(f"All LLM backends failed: {last_error}")


# ═══════════════════════════════════════════════════════════════════════════════
#  Tool system (from bot.py)
# ═══════════════════════════════════════════════════════════════════════════════

_registry: dict[str, Callable[..., Any]] = {}


def _type_to_json_schema(tp: Any) -> Any:
    if tp is list:
        return {"type": "array", "items": {"type": "object"}}
    if tp is dict:
        return {"type": "object"}
    origin = getattr(tp, "__origin__", None)
    if origin is list:
        items_type = tp.__args__[0] if getattr(tp, "__args__", None) else str
        item_schema = _type_to_json_schema(items_type)
        return {"type": "array", "items": item_schema if isinstance(item_schema, dict) else {"type": item_schema}}
    if origin is dict:
        return {"type": "object"}
    args = getattr(tp, "__args__", None)
    if args and type(None) in args:
        non_none = [a for a in args if a is not type(None)]
        if non_none:
            return _type_to_json_schema(non_none[0])
    if tp is str:
        return "string"
    if tp is int or tp is float:
        return "number"
    if tp is bool:
        return "boolean"
    return "string"


def _build_function_payload(func: Callable[..., Any]) -> dict[str, Any]:
    try:
        sig = inspect.signature(func, eval_str=True)
    except (NameError, TypeError):
        sig = inspect.signature(func)
    doc = inspect.getdoc(func) or ""
    description = doc.split("\n")[0] if doc else func.__name__.replace("_", " ").title()
    properties: dict[str, Any] = {}
    required: list[str] = []
    for name, param in sig.parameters.items():
        if name == "self":
            continue
        param_doc = ""
        for line in doc.split("\n")[1:]:
            stripped = line.strip()
            if stripped.startswith(f"{name}:"):
                param_doc = stripped.split(":", 1)[1].strip()
                break
        prop: dict[str, Any] = {"type": "string"}
        if param.annotation is not inspect.Parameter.empty:
            js = _type_to_json_schema(param.annotation)
            prop = dict(js) if isinstance(js, dict) else {"type": js}
        if param.default is not inspect.Parameter.empty:
            if param.default is not None:
                prop["default"] = param.default
        else:
            required.append(name)
        if param_doc:
            prop["description"] = param_doc
        properties[name] = prop
    payload: dict[str, Any] = {
        "type": "function",
        "function": {
            "name": func.__name__,
            "description": description,
            "parameters": {"type": "object", "properties": properties},
        },
    }
    if required:
        payload["function"]["parameters"]["required"] = required
    return payload


def tool(func: Callable[..., Any]) -> Callable[..., Any]:
    is_async = asyncio.iscoroutinefunction(func)
    if is_async:
        @functools.wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            return await func(*args, **kwargs)
    else:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return func(*args, **kwargs)
    _registry[func.__name__] = wrapper
    wrapper._tool_payload = _build_function_payload(func)  # type: ignore[attr-defined]
    wrapper._is_async = is_async  # type: ignore[attr-defined]
    return wrapper


def get_tool_payloads(exclude: tuple[str, ...] = ()) -> list[dict[str, Any]]:
    return [f._tool_payload for n, f in _registry.items() if n not in exclude]


def get_tool(name: str) -> Callable[..., Any] | None:
    if name in _registry:
        return _registry[name]
    flat = name.replace("_", "")
    for known in _registry:
        if flat == known.replace("_", ""):
            return _registry[known]
    nl = name.lower()
    for known in _registry:
        if known.lower() == nl:
            return _registry[known]
    return None


def coerce_tool_args(payload: dict[str, Any], args: dict[str, Any]) -> dict[str, Any]:
    props = payload["function"]["parameters"].get("properties", {})
    out: dict[str, Any] = {}
    for key, value in args.items():
        expected = props.get(key, {}).get("type")
        if isinstance(value, str):
            if expected in ("array", "object"):
                try:
                    value = json.loads(value)
                except json.JSONDecodeError:
                    pass
            elif expected == "number":
                try:
                    value = float(value) if "." in value or "e" in value.lower() else int(value)
                except ValueError:
                    pass
            elif expected == "boolean" and value.lower() in ("true", "false"):
                value = value.lower() == "true"
        out[key] = value
    return out


def validate_tool_args(payload: dict[str, Any], args: dict[str, Any]) -> str | None:
    schema = payload["function"]["parameters"]
    required = schema.get("required", [])
    props = schema.get("properties", {})
    type_map = {
        "string": str,
        "number": (int, float),
        "boolean": bool,
        "array": list,
        "object": dict,
    }
    for key in required:
        if key not in args:
            return f"Missing required parameter '{key}'"
    for key, value in args.items():
        if key not in props:
            return f"Unknown parameter '{key}' (allowed: {sorted(props.keys())})"
        if value is None:
            continue
        expected = props[key].get("type")
        if expected not in type_map:
            continue
        if expected == "boolean" and not isinstance(value, bool):
            return f"Parameter '{key}' must be a boolean"
        if expected == "number" and (isinstance(value, bool) or not isinstance(value, (int, float))):
            return f"Parameter '{key}' must be a number"
        if expected not in ("boolean", "number") and not isinstance(value, type_map[expected]):
            return f"Parameter '{key}' must be of type {expected}"
    return None


# ═══════════════════════════════════════════════════════════════════════════════
#  Agent state
# ═══════════════════════════════════════════════════════════════════════════════


@dataclass
class AgentState:
    plan: dict[str, Any] | None = None
    reflections: list[dict[str, str]] = field(default_factory=list)
    iteration: int = 0
    verified: bool = False
    modified_files: set[str] = field(default_factory=set)
    read_files: dict[str, str] = field(default_factory=dict)
    gate_triggered: bool = False
    last_error: bool = False


@dataclass
class SwarmProcess:
    pid: int
    task: str
    proc: Any
    log_path: Path
    err_path: Path
    timeout: int


_swarm: dict[int, SwarmProcess] = {}
_agent_state: contextvars.ContextVar[AgentState | None] = contextvars.ContextVar(
    "agent_state", default=None,
)


def _state() -> AgentState | None:
    return _agent_state.get()


def _record_read(path: Path, content: str) -> None:
    state = _state()
    if state is not None:
        state.read_files[str(path.resolve())] = _sha(content)


def _record_modification(path: Path, content: str) -> None:
    state = _state()
    if state is not None:
        key = str(path.resolve())
        state.modified_files.add(key)
        state.read_files[key] = _sha(content)


def _mutation_guard(path: Path) -> str | None:
    if not path.exists():
        return None
    state = _state()
    if state is None:
        return None
    key = str(path.resolve())
    if key not in state.read_files:
        return (
            f"Read before write: you must read_file('{path}') "
            "before modifying an existing file."
        )
    current = _sha(path.read_text(encoding="utf-8", errors="replace"))
    if current != state.read_files[key]:
        return (
            f"File '{path}' changed on disk since you last read it. "
            "Re-read it before modifying."
        )
    return None


# ═══════════════════════════════════════════════════════════════════════════════
#  All @tool functions (from bot.py)
# ═══════════════════════════════════════════════════════════════════════════════


@tool
async def read_file(path: str):
    """Read the full contents of a UTF-8 text file. Required before editing an existing file.
    path: Path to the file.
    """
    def _do() -> str:
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "File not found"})
        try:
            content = p.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            return json.dumps({"status": "error", "error": "File is not UTF-8 text"})
        _record_read(p, content)
        return json.dumps({
            "status": "success",
            "path": str(p),
            "content": content,
            "lines": content.count("\n") + 1,
            "bytes": len(content.encode("utf-8")),
        })
    return await asyncio.to_thread(_do)


@tool
async def read_lines(path: str, start: int = 1, end: int | None = None):
    """Read a 1-indexed inclusive line range from a file. Use for large files. Records the file as read.
    path: File path.
    start: First line, 1-indexed.
    end: Last line inclusive; omit to read to end.
    """
    def _do() -> str:
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "File not found"})
        full = p.read_text(encoding="utf-8", errors="replace")
        _record_read(p, full)
        lines = full.split("\n")
        s = max(1, int(start)) - 1
        e = int(end) if end is not None else len(lines)
        excerpt = lines[s:e]
        return json.dumps({
            "status": "success",
            "path": str(p),
            "start": s + 1,
            "end": s + len(excerpt),
            "total_lines": len(lines),
            "content": "\n".join(excerpt),
        })
    return await asyncio.to_thread(_do)


@tool
async def create_file(path: str, content: str):
    """Create a new file. Fails if the file already exists.
    path: File path.
    content: Full file contents.
    """
    def _do() -> str:
        p = Path(path)
        if p.exists():
            return json.dumps({
                "status": "error",
                "error": "File already exists; use edit_file or write_file",
            })
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        _record_modification(p, content)
        return json.dumps({
            "status": "success",
            "path": str(p),
            "bytes": len(content.encode("utf-8")),
        })
    return await asyncio.to_thread(_do)


@tool
async def write_file(path: str, content: str):
    """Overwrite a file with new content. For an existing file you must read_file it first.
    path: File path.
    content: Full new contents.
    """
    def _do() -> str:
        p = Path(path)
        guard = _mutation_guard(p)
        if guard:
            return json.dumps({"status": "error", "error": guard})
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        _record_modification(p, content)
        return json.dumps({
            "status": "success",
            "path": str(p),
            "bytes": len(content.encode("utf-8")),
        })
    return await asyncio.to_thread(_do)


@tool
async def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False):
    """Replace exact text in an existing file. old_string must match uniquely unless replace_all is true. Read the file first.
    path: File path.
    old_string: Exact text to replace.
    new_string: Replacement text.
    replace_all: Replace every occurrence when true.
    """
    def _do() -> str:
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "File not found"})
        guard = _mutation_guard(p)
        if guard:
            return json.dumps({"status": "error", "error": guard})
        src = p.read_text(encoding="utf-8")
        count = src.count(old_string)
        if count == 0:
            return json.dumps({"status": "error", "error": "old_string not found in file"})
        if not replace_all and count > 1:
            return json.dumps({
                "status": "error",
                "error": f"old_string is not unique ({count} matches); set replace_all=true or add surrounding context",
            })
        updated = src.replace(old_string, new_string) if replace_all else src.replace(old_string, new_string, 1)
        p.write_text(updated, encoding="utf-8")
        _record_modification(p, updated)
        return json.dumps({"status": "success", "path": str(p), "replacements": count if replace_all else 1})
    return await asyncio.to_thread(_do)


def _apply_unified_diff(source: str, patch: str) -> tuple[str | None, str | None]:
    lines = patch.splitlines()
    hunks: list[tuple[list[str], list[str]]] = []
    before: list[str] = []
    after: list[str] = []
    in_hunk = False
    for line in lines:
        if line.startswith("@@"):
            if in_hunk:
                hunks.append((before, after))
            before, after, in_hunk = [], [], True
            continue
        if line.startswith("---") or line.startswith("+++"):
            continue
        if not in_hunk:
            continue
        if line.startswith("-"):
            before.append(line[1:])
        elif line.startswith("+"):
            after.append(line[1:])
        elif line.startswith(" "):
            before.append(line[1:])
            after.append(line[1:])
        elif line == "":
            before.append("")
            after.append("")
    if in_hunk:
        hunks.append((before, after))
    if not hunks:
        return None, "No hunks found in patch"

    result = source
    for index, (before_lines, after_lines) in enumerate(hunks):
        before_block = "\n".join(before_lines)
        after_block = "\n".join(after_lines)
        if before_block and before_block in result:
            result = result.replace(before_block, after_block, 1)
            continue
        stripped = before_block.strip("\n")
        if stripped and stripped in result:
            result = result.replace(stripped, after_block.strip("\n"), 1)
            continue
        if not before_block.strip():
            result = result + ("\n" if not result.endswith("\n") else "") + after_block
            continue
        return None, f"Hunk {index + 1} did not match the file content"
    return result, None


@tool
async def patch_file(path: str, patch: str):
    """Apply a unified diff to an existing file. Hunks are matched by context with whitespace-tolerant fallback. Read the file first.
    path: File to patch.
    patch: Unified diff text with @@ hunk headers and -/+/space line prefixes.
    """
    def _do() -> str:
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "File not found"})
        guard = _mutation_guard(p)
        if guard:
            return json.dumps({"status": "error", "error": guard})
        source = p.read_text(encoding="utf-8")
        updated, err = _apply_unified_diff(source, patch)
        if err is not None or updated is None:
            return json.dumps({"status": "error", "error": err or "Patch failed"})
        p.write_text(updated, encoding="utf-8")
        _record_modification(p, updated)
        return json.dumps({"status": "success", "path": str(p), "bytes": len(updated.encode("utf-8"))})
    return await asyncio.to_thread(_do)


@tool
async def list_dir(path: str = ".", show_hidden: bool = False):
    """List entries of a directory, single level, directories first.
    path: Directory to list.
    show_hidden: Include dotfiles when true.
    """
    def _do() -> str:
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "Path does not exist"})
        if not p.is_dir():
            return json.dumps({"status": "error", "error": "Not a directory"})
        entries = []
        for e in sorted(p.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
            if not show_hidden and e.name.startswith("."):
                continue
            try:
                size = e.stat().st_size if e.is_file() else None
            except OSError:
                size = None
            entries.append({"name": e.name, "type": "dir" if e.is_dir() else "file", "size": size})
        return json.dumps({"status": "success", "path": str(p), "entries": entries})
    return await asyncio.to_thread(_do)


@tool
async def glob_files(pattern: str, path: str = "."):
    """List files matching a glob pattern; supports ** for recursive matching.
    pattern: Glob pattern, e.g. '**/*.py'.
    path: Root directory.
    """
    def _do() -> str:
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "Path does not exist"})
        results = []
        for f in p.glob(pattern):
            if any(part in INDEX_SKIP_DIRS for part in f.parts):
                continue
            results.append(str(f))
            if len(results) >= 1000:
                break
        return json.dumps({
            "status": "success",
            "files": sorted(results),
            "count": len(results),
        })
    return await asyncio.to_thread(_do)


@tool
async def grep(
    pattern: str,
    path: str = ".",
    glob: str | None = None,
    ignore_case: bool = False,
    max_matches: int = 200,
):
    """Recursively search for a regex pattern in files, skipping build and cache directories.
    pattern: Python regex.
    path: Directory or file to search.
    glob: Optional filename glob filter, e.g. '*.py'.
    ignore_case: Case-insensitive when true.
    max_matches: Maximum match lines to return.
    """
    def _do() -> str:
        try:
            rx = re.compile(pattern, re.IGNORECASE if ignore_case else 0)
        except re.error as e:
            return json.dumps({"status": "error", "error": f"Invalid regex: {e}"})
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "Path does not exist"})
        files = [p] if p.is_file() else [
            sub for sub in p.rglob("*")
            if sub.is_file()
            and not any(part in INDEX_SKIP_DIRS for part in sub.parts)
            and (not glob or sub.match(glob))
        ]
        matches = []
        scanned = 0
        for f in files:
            try:
                if f.stat().st_size > INDEX_MAX_FILE_BYTES:
                    continue
                scanned += 1
                with f.open("r", encoding="utf-8", errors="replace") as fh:
                    for lineno, line in enumerate(fh, start=1):
                        if rx.search(line):
                            matches.append({
                                "file": str(f),
                                "line": lineno,
                                "text": line.rstrip("\n")[:240],
                            })
                            if len(matches) >= max_matches:
                                return json.dumps({
                                    "status": "success",
                                    "matches": matches,
                                    "truncated": True,
                                    "files_scanned": scanned,
                                })
            except (OSError, UnicodeDecodeError):
                continue
        return json.dumps({
            "status": "success",
            "matches": matches,
            "truncated": False,
            "files_scanned": scanned,
        })
    return await asyncio.to_thread(_do)


@tool
async def find_symbol(name: str, path: str = "."):
    """Find Python class or function definitions matching a name using the AST.
    name: Exact symbol name.
    path: Root directory or .py file.
    """
    def _do() -> str:
        p = Path(path)
        if not p.exists():
            return json.dumps({"status": "error", "error": "Path does not exist"})
        files = [p] if p.is_file() else [
            f for f in p.rglob("*.py")
            if not any(part in INDEX_SKIP_DIRS for part in f.parts)
        ]
        matches = []
        for f in files:
            try:
                tree = ast.parse(
                    f.read_text(encoding="utf-8", errors="replace"), filename=str(f),
                )
            except (OSError, SyntaxError):
                continue
            for node in ast.walk(tree):
                if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name == name:
                    matches.append({
                        "file": str(f),
                        "line": node.lineno,
                        "kind": type(node).__name__,
                        "name": node.name,
                    })
        return json.dumps({
            "status": "success",
            "matches": matches,
            "files_scanned": len(files),
        })
    return await asyncio.to_thread(_do)


@tool
async def run_command(command: str, timeout: int | None = None):
    """Execute a shell command via bash with live stdout/stderr streaming. Default timeout 300 seconds.
    command: Shell command line.
    timeout: Timeout in seconds.
    """
    effective = int(timeout) if timeout is not None else DEFAULT_COMMAND_TIMEOUT
    try:
        stdout, stderr, exit_code, timed_out = await stream_subprocess(
            ["bash", "-c", command],
            timeout=effective,
            stdout_sink=sys.stdout,
            stderr_sink=sys.stderr,
        )
    except Exception as e:  # noqa: BLE001
        return json.dumps({"status": "error", "error": f"{type(e).__name__}: {e}"})
    status = "error" if (timed_out or exit_code != 0) else "success"
    return json.dumps({
        "status": status,
        "command": command,
        "exit_code": exit_code,
        "timed_out": timed_out,
        "stdout": stdout,
        "stderr": stderr,
    })


def _rsearch_querystring(params: dict[str, Any]) -> str:
    cleaned: dict[str, str] = {}
    for key, value in params.items():
        if value is None:
            continue
        cleaned[key] = ("true" if value else "false") if isinstance(value, bool) else str(value)
    return urlencode(cleaned)


async def _rsearch_get(endpoint: str) -> dict[str, Any]:
    client = await get_http_client()
    try:
        response = await client.get(endpoint, timeout=float(RSEARCH_TIMEOUT))
    except Exception as e:  # noqa: BLE001
        return {"status": "error", "error": f"{type(e).__name__}: {e}", "endpoint": endpoint}
    body = response.text[:RSEARCH_MAX_BYTES]
    try:
        parsed: Any = json.loads(body)
    except json.JSONDecodeError:
        parsed = body
    return {"status": "success", "endpoint": endpoint, "status_code": response.status_code, "result": parsed}


@tool
async def web_search(query: str, content: bool = False, count: int = 10, images: bool = False):
    """Web search via rsearch across many providers. Returns titles, urls and descriptions.
    query: Search query.
    content: Fetch and include full page content for each result when true (large response).
    count: Number of results 1-100.
    images: Return image results when true.
    """
    q = (query or "").strip()
    if not q:
        return json.dumps({"status": "error", "error": "query is required"})
    params: dict[str, Any] = {
        "query": q[:1024],
        "count": max(1, min(int(count), 100)),
        "content": content,
    }
    if images:
        params["type"] = "images"
    endpoint = f"{RSEARCH_BASE_URL}/search?{_rsearch_querystring(params)}"
    return json.dumps(await _rsearch_get(endpoint))


@tool
async def deep_search(query: str, content: bool = True):
    """Deep multi-step research via rsearch (deep=true). Slower; aggregates and synthesizes many sources.
    query: Research question.
    content: Include full page content for sources when true.
    """
    q = (query or "").strip()
    if not q:
        return json.dumps({"status": "error", "error": "query is required"})
    endpoint = (
        f"{RSEARCH_BASE_URL}/search?"
        f"{_rsearch_querystring({'query': q[:1024], 'deep': True, 'content': content})}"
    )
    return json.dumps(await _rsearch_get(endpoint))


@tool
async def ai_search(query: str):
    """AI-answered web search via rsearch (ai=true). Returns a synthesized natural-language answer with citations.
    query: Question to answer from the web.
    """
    q = (query or "").strip()
    if not q:
        return json.dumps({"status": "error", "error": "query is required"})
    endpoint = (
        f"{RSEARCH_BASE_URL}/search?"
        f"{_rsearch_querystring({'query': q[:1024], 'ai': True})}"
    )
    return json.dumps(await _rsearch_get(endpoint))


@tool
async def fetch_url(url: str, max_bytes: int = 1048576):
    """Fetch the body of an http(s) URL using the stealth Chrome client.
    url: Absolute http or https URL.
    max_bytes: Cap on bytes read from the response body.
    """
    target = (url or "").strip()
    parsed = urlparse(target)
    if parsed.scheme not in ("http", "https") or not parsed.netloc:
        return json.dumps({"status": "error", "error": "Only absolute http(s) URLs are allowed"})
    cap = max(1, min(int(max_bytes), 10 * 1024 * 1024))
    client = await get_http_client()
    try:
        result = await client.download_bytes(target, max_bytes=cap, dest_type="empty")
    except Exception as e:  # noqa: BLE001
        return json.dumps({"status": "error", "error": f"{type(e).__name__}: {e}"})
    if result.error:
        return json.dumps({"status": "error", "error": result.error})
    body = result.data.decode("utf-8", errors="replace")
    return json.dumps({
        "status": "success",
        "url": result.final_url,
        "status_code": result.status_code,
        "content_type": result.content_type,
        "truncated": result.truncated,
        "body": body,
    })


@tool
async def download_file(url: str, destination: str = "downloads", filename: str | None = None):
    """Download a single binary file (image, archive, document, media) to disk via the stealth client.
    url: Absolute http or https URL of the file.
    destination: Target directory; created if missing.
    filename: Optional output filename; derived from the URL when omitted.
    """
    target = (url or "").strip()
    parsed = urlparse(target)
    if parsed.scheme not in ("http", "https") or not parsed.netloc:
        return json.dumps({"status": "error", "error": "Only absolute http(s) URLs are allowed"})
    client = await get_http_client()
    result = await client.download_file(
        target, destination, filename=filename,
        dest_type=resource_kind_for_url(target),
    )
    return json.dumps({
        "status": "success" if result.ok else "error",
        "url": result.url,
        "path": str(result.path) if result.path else None,
        "bytes": result.bytes_written,
        "status_code": result.status_code,
        "error": result.error,
    })


@tool
async def download_files(urls: list, destination: str = "downloads", concurrency: int = 8):
    """Download many binary files concurrently to disk via the stealth client with bounded concurrency.
    urls: List of absolute http(s) URLs to download.
    destination: Target directory; created if missing.
    concurrency: Maximum simultaneous downloads.
    """
    clean = [str(u).strip() for u in urls if str(u).strip()]
    if not clean:
        return json.dumps({"status": "error", "error": "urls is required and must be non-empty"})
    client = await get_http_client()
    results = await client.download_files(clean, destination, concurrency=int(concurrency))
    items = [
        {
            "url": r.url, "ok": r.ok, "path": str(r.path) if r.path else None,
            "bytes": r.bytes_written, "error": r.error,
        }
        for r in results
    ]
    succeeded = sum(1 for r in results if r.ok)
    return json.dumps({
        "status": "success" if succeeded == len(results) else "error",
        "succeeded": succeeded,
        "failed": len(results) - succeeded,
        "results": items,
    })


@tool
async def describe_image(prompt: str, image_path: str):
    """Describe or answer a question about a local image using the vision-capable model. Disabled on the DeepSeek fallback.
    prompt: Instruction for what to describe or ask about the image.
    image_path: Path to a local image file.
    """
    if not vision_available():
        return json.dumps({
            "status": "error",
            "error": (
                "Image recognition is disabled: the molodetz vision backend is unavailable; "
                "running on the DeepSeek text-only fallback."
            ),
        })
    p = Path(image_path)
    if not p.is_file():
        return json.dumps({"status": "error", "error": f"File not found: {image_path}"})
    mime_type = mimetypes.guess_type(p.name)[0]
    if not mime_type or not mime_type.startswith("image/"):
        return json.dumps({"status": "error", "error": f"Unsupported image type: {mime_type or 'unknown'}"})
    if p.stat().st_size > IMAGE_MAX_BYTES:
        return json.dumps({"status": "error", "error": f"Image too large (max {IMAGE_MAX_BYTES} bytes)"})
    try:
        raw = await asyncio.to_thread(p.read_bytes)
    except OSError as e:
        return json.dumps({"status": "error", "error": f"Failed to read image: {e}"})
    data_url = f"data:{mime_type};base64,{base64.b64encode(raw).decode('utf-8')}"
    messages = [{
        "role": "user",
        "content": [
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": data_url}},
        ],
    }]
    try:
        result = await _call_backend(get_backends()[0], messages, temperature=0.0)
        description = result["choices"][0]["message"]["content"]
    except (RuntimeError, KeyError, IndexError, TypeError) as e:
        return json.dumps({
            "status": "error",
            "error": f"Image recognition unavailable (molodetz backend error): {type(e).__name__}: {e}",
        })
    return json.dumps({"status": "success", "description": description})


@tool
async def plan(goal: str, steps: list, success_criteria: str, confidence: float = 0.8):
    """Record the structured execution plan. MUST be the very first tool call on a new task.
    goal: One-line restatement of the user goal.
    steps: Ordered list of step objects, each with keys id, action, depends_on.
    success_criteria: Concrete criteria for declaring the task complete.
    confidence: 0.0-1.0 self-estimate of plan correctness.
    """
    try:
        confidence_value = float(confidence)
    except (TypeError, ValueError):
        confidence_value = 0.8
    state = _state()
    if state is not None:
        state.plan = {
            "goal": goal, "steps": steps,
            "success_criteria": success_criteria, "confidence": confidence_value,
        }
    advice = ""
    if confidence_value < 0.6:
        advice = "Plan confidence below 0.6 — gather more context (read/grep/retrieve) before executing."
    return json.dumps({
        "status": "success", "plan_recorded": True,
        "step_count": len(steps), "advice": advice,
    })


@tool
async def reflect(observation: str, conclusion: str, next_action: str):
    """Record a reflection: what was observed, what it means, what to do next. Call after errors and at task end.
    observation: What was observed (failure mode, unexpected output).
    conclusion: Diagnosis or interpretation.
    next_action: The chosen next step.
    """
    state = _state()
    total = 1
    if state is not None:
        state.reflections.append({
            "observation": observation,
            "conclusion": conclusion,
            "next_action": next_action,
        })
        total = len(state.reflections)
    return json.dumps({
        "status": "success",
        "reflection_recorded": True,
        "total_reflections": total,
    })


@tool
async def verify(command: str, timeout: int = 600):
    """Run a verification command (linter, tests, validator). Marks the task verified on success.
    command: Shell command that verifies the change, for example the project's test or lint command.
    timeout: Timeout in seconds.
    """
    try:
        stdout, stderr, exit_code, timed_out = await stream_subprocess(
            ["bash", "-c", command],
            timeout=int(timeout),
            stdout_sink=sys.stdout,
            stderr_sink=sys.stderr,
        )
    except Exception as e:  # noqa: BLE001
        return json.dumps({"status": "error", "error": f"{type(e).__name__}: {e}"})
    passed = exit_code == 0 and not timed_out
    state = _state()
    if state is not None and passed:
        state.verified = True
    return json.dumps({
        "status": "success" if passed else "error",
        "passed": passed,
        "command": command,
        "exit_code": exit_code,
        "timed_out": timed_out,
        "stdout": stdout,
        "stderr": stderr,
    })


@tool
async def get_current_isodate():
    """Return the live current local date and time as a full ISO 8601 string with timezone, plus the day name."""
    now = datetime.now().astimezone()
    return json.dumps({
        "status": "success",
        "isodate": now.isoformat(),
        "day": now.strftime("%A"),
        "timestamp": now.timestamp(),
    })


@tool
async def retrieve(query: str, k: int = 5):
    """Retrieve the top-k files most relevant to a query using BM25 over the working directory.
    query: Natural language or keyword query.
    k: Number of results.
    """
    idx = await get_corpus_index()
    return json.dumps({
        "status": "success",
        "results": idx.search(query, k=int(k)),
        "indexed_files": idx.n,
    })


@tool
async def delegate(task: str, allowed_tools: list | None = None):
    """Spawn an in-process sub-agent with a fresh context to complete a self-contained sub-task.
    task: Clear, scoped task description.
    allowed_tools: Optional whitelist of tool names; defaults to all tools except delegate and spawn.
    """
    excluded = (
        "delegate", "spawn", "swarm_wait",
        "swarm_result", "swarm_tail", "swarm_kill", "swarm_cleanup",
    )
    if allowed_tools:
        sub_payloads = [
            t for t in get_tool_payloads()
            if t["function"]["name"] in allowed_tools and t["function"]["name"] not in excluded
        ]
    else:
        sub_payloads = get_tool_payloads(exclude=excluded)
    sub_messages = [
        {"role": "system", "content": _with_datetime(SUB_AGENT_SYSTEM_PROMPT)},
        {"role": "user", "content": task},
    ]
    sub_state = AgentState()
    final = await react_loop(
        messages=sub_messages,
        tools_payload=sub_payloads,
        state=sub_state,
        max_iterations=DELEGATE_MAX_ITERATIONS,
        renderer=None,
        prefix="[delegate] ",
    )
    return json.dumps({
        "status": "success",
        "iterations": sub_state.iteration,
        "verified": sub_state.verified,
        "reflections": len(sub_state.reflections),
        "result": final or "",
    })


@tool
async def spawn(task: str, timeout: int = 1800):
    """Launch an independent OS subprocess running botje.py on a task.
    task: Self-contained task for the subprocess agent.
    timeout: Wall-clock timeout in seconds for the subprocess.
    """
    SWARM_DIR.mkdir(parents=True, exist_ok=True)
    pid_seed = len(_swarm) + 1
    log_path = SWARM_DIR / f"job_{pid_seed}.out"
    err_path = SWARM_DIR / f"job_{pid_seed}.err"
    out_handle = log_path.open("wb")
    err_handle = err_path.open("wb")
    proc = await asyncio.create_subprocess_exec(
        sys.executable, str(Path(__file__).resolve()), "--prompt", task, "--no-color",
        stdin=asyncio.subprocess.DEVNULL,
        stdout=out_handle,
        stderr=err_handle,
    )
    record = SwarmProcess(
        pid=proc.pid, task=task, proc=proc,
        log_path=log_path, err_path=err_path, timeout=int(timeout),
    )
    _swarm[proc.pid] = record
    return json.dumps({"status": "success", "pid": proc.pid, "log": str(log_path)})


@tool
async def swarm_wait(pid: int):
    """Wait for a spawned subprocess to finish and return its exit code.
    pid: Process id returned by spawn().
    """
    record = _swarm.get(int(pid))
    if record is None:
        return json.dumps({"status": "error", "error": f"No spawned process with pid {pid}"})
    try:
        returncode = await asyncio.wait_for(record.proc.wait(), timeout=record.timeout)
        timed_out = False
    except asyncio.TimeoutError:
        record.proc.kill()
        await record.proc.wait()
        returncode = record.proc.returncode
        timed_out = True
    return json.dumps({
        "status": "success", "pid": int(pid),
        "exit_code": returncode, "timed_out": timed_out,
    })


@tool
async def swarm_result(pid: int, max_bytes: int = 65536):
    """Read the captured stdout and stderr of a spawned subprocess.
    pid: Process id returned by spawn().
    max_bytes: Cap on bytes returned from each stream.
    """
    record = _swarm.get(int(pid))
    if record is None:
        return json.dumps({"status": "error", "error": f"No spawned process with pid {pid}"})
    cap = max(1, int(max_bytes))
    out = (
        record.log_path.read_text(encoding="utf-8", errors="replace")[-cap:]
        if record.log_path.exists() else ""
    )
    err = (
        record.err_path.read_text(encoding="utf-8", errors="replace")[-cap:]
        if record.err_path.exists() else ""
    )
    return json.dumps({
        "status": "success", "pid": int(pid),
        "running": record.proc.returncode is None,
        "stdout": out, "stderr": err,
    })


@tool
async def swarm_tail(pid: int, lines: int = 40):
    """Return the last N lines of a spawned subprocess's stdout.
    pid: Process id returned by spawn().
    lines: Number of trailing lines.
    """
    record = _swarm.get(int(pid))
    if record is None:
        return json.dumps({"status": "error", "error": f"No spawned process with pid {pid}"})
    text = (
        record.log_path.read_text(encoding="utf-8", errors="replace")
        if record.log_path.exists() else ""
    )
    tail = "\n".join(text.splitlines()[-int(lines):])
    return json.dumps({
        "status": "success", "pid": int(pid),
        "running": record.proc.returncode is None,
        "tail": tail,
    })


@tool
async def swarm_kill(pid: int):
    """Terminate a spawned subprocess.
    pid: Process id returned by spawn().
    """
    record = _swarm.get(int(pid))
    if record is None:
        return json.dumps({"status": "error", "error": f"No spawned process with pid {pid}"})
    if record.proc.returncode is None:
        record.proc.kill()
        await record.proc.wait()
    return json.dumps({
        "status": "success", "pid": int(pid), "exit_code": record.proc.returncode,
    })


@tool
async def swarm_cleanup():
    """Remove finished spawned-process records and their captured output files."""
    removed = []
    for pid in list(_swarm.keys()):
        record = _swarm[pid]
        if record.proc.returncode is not None:
            for fp in (record.log_path, record.err_path):
                try:
                    fp.unlink(missing_ok=True)
                except OSError:
                    pass
            removed.append(pid)
            del _swarm[pid]
    return json.dumps({
        "status": "success", "removed": removed,
        "remaining": list(_swarm.keys()),
    })


# ═══════════════════════════════════════════════════════════════════════════════
#  CorpusIndex (for retrieve)
# ═══════════════════════════════════════════════════════════════════════════════


class CorpusIndex:
    def __init__(self, root: Path, exts: tuple[str, ...] = INDEX_EXTS) -> None:
        self.root = root
        self.exts = exts
        self.docs: list[dict[str, Any]] = []
        self.tf: list[collections.Counter] = []
        self.dl: list[int] = []
        self.idf: dict[str, float] = {}
        self.avgdl: float = 0.0
        self.n: int = 0

    @staticmethod
    def _tokenize(text: str) -> list[str]:
        tokens = re.findall(r"[A-Za-z_][A-Za-z0-9_]*|\d+", text.lower())
        extra: list[str] = []
        for t in tokens:
            extra.extend(re.findall(r"[A-Z]?[a-z]+", t))
        return list(dict.fromkeys(tokens + extra))

    def _scan_files(self) -> list[Path]:
        out: list[Path] = []
        for p in self.root.rglob("*"):
            if not p.is_file() or any(part in INDEX_SKIP_DIRS for part in p.parts):
                continue
            if p.suffix.lower() not in self.exts:
                continue
            try:
                if p.stat().st_size > INDEX_MAX_FILE_BYTES:
                    continue
            except OSError:
                continue
            out.append(p)
        return out

    async def build(self) -> None:
        files = await asyncio.to_thread(self._scan_files)
        if not files:
            return
        texts = await asyncio.gather(*[
            asyncio.to_thread(lambda fp=f: fp.read_text(encoding="utf-8", errors="replace"))
            for f in files
        ])
        df: collections.Counter = collections.Counter()
        for path, text in zip(files, texts):
            if not text:
                continue
            tokens = self._tokenize(text)
            if not tokens:
                continue
            tf = collections.Counter(tokens)
            for term in tf:
                df[term] += 1
            self.docs.append({"path": str(path), "size": len(text)})
            self.tf.append(tf)
            self.dl.append(len(tokens))
        self.n = len(self.docs)
        self.avgdl = sum(self.dl) / max(self.n, 1)
        self.idf = {
            t: math.log((self.n - dft + 0.5) / (dft + 0.5) + 1)
            for t, dft in df.items()
        }

    def search(self, query: str, k: int = 5) -> list[dict[str, Any]]:
        tokens = self._tokenize(query)
        if not tokens or not self.docs:
            return []
        k1, b = 1.5, 0.75
        scored: list[tuple[float, int]] = []
        for i, tf in enumerate(self.tf):
            score = 0.0
            for t in tokens:
                f = tf.get(t, 0)
                if f == 0:
                    continue
                norm = 1 - b + b * (self.dl[i] / max(self.avgdl, 1))
                score += self.idf.get(t, 0.0) * (f * (k1 + 1)) / (f + k1 * norm)
            if score > 0:
                scored.append((score, i))
        scored.sort(reverse=True)
        return [
            {"path": self.docs[i]["path"], "score": round(s, 3), "size": self.docs[i]["size"]}
            for s, i in scored[:k]
        ]


_corpus_index: CorpusIndex | None = None
_corpus_lock: asyncio.Lock | None = None


async def get_corpus_index() -> CorpusIndex:
    global _corpus_index, _corpus_lock
    if _corpus_lock is None:
        _corpus_lock = asyncio.Lock()
    async with _corpus_lock:
        if _corpus_index is None:
            idx = CorpusIndex(WORKDIR)
            await idx.build()
            _corpus_index = idx
        return _corpus_index


# ═══════════════════════════════════════════════════════════════════════════════
#  Context management / react loop
# ═══════════════════════════════════════════════════════════════════════════════


def context_size(messages: list[dict[str, Any]]) -> int:
    return len(json.dumps(messages, default=str))


def find_compaction_split(messages: list[dict[str, Any]], target_keep: int) -> int:
    if len(messages) <= target_keep:
        return 1
    candidate = len(messages) - target_keep
    while candidate > 1:
        if messages[candidate].get("role") == "user":
            return candidate
        candidate -= 1
    return 1


async def compact_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if len(messages) < CONTEXT_KEEP_TAIL_MESSAGES + 3:
        return messages
    split = find_compaction_split(messages, CONTEXT_KEEP_TAIL_MESSAGES)
    if split <= 1:
        return messages
    system_msg = messages[0]
    middle = messages[1:split]
    tail = messages[split:]
    if not middle:
        return messages
    summary_prompt = (
        "Summarize the following agent conversation segment as a concise factual log of actions taken, "
        "files inspected/modified, conclusions reached, and outstanding tasks. Keep file paths, exact "
        "identifiers, and decisions verbatim. Maximum 800 words.\n\n---\n\n"
        + json.dumps(middle, default=str)[:120000]
    )
    try:
        response = await llm_call(
            [
                {"role": "system", "content": "You are a precise technical summarizer."},
                {"role": "user", "content": summary_prompt},
            ],
            temperature=0.0,
        )
        summary = response["choices"][0]["message"]["content"]
    except (RuntimeError, KeyError, IndexError, TypeError):
        return messages
    return [system_msg, {"role": "assistant", "content": f"[Compacted earlier turns]\n\n{summary}"}, *tail]


async def execute_tool_call(
    tool_call: dict[str, Any],
    md: MarkdownRenderer | None,
    prefix: str = "",
) -> str:
    name = tool_call["function"]["name"]
    raw = tool_call["function"].get("arguments") or "{}"
    try:
        args = json.loads(raw)
    except json.JSONDecodeError as e:
        return json.dumps({"status": "error", "error": f"Invalid JSON arguments: {e}"})
    func = get_tool(name)
    if not func:
        return json.dumps({
            "status": "error",
            "error": f"Tool '{name}' not found. Available: {sorted(_registry.keys())}",
        })
    args = coerce_tool_args(func._tool_payload, args)
    err = validate_tool_args(func._tool_payload, args)
    if err:
        return json.dumps({"status": "error", "error": err})
    if md is not None:
        arg_repr = json.dumps(args, default=str)
        if len(arg_repr) > TOOL_ARG_PREVIEW:
            arg_repr = arg_repr[:TOOL_ARG_PREVIEW] + "…"
        sys.stderr.write(
            f"{prefix}{md._c('bold', '↳')} {md._c('h3', name)}{md._c('dim', ' ' + arg_repr)}\n",
        )
        sys.stderr.flush()
    try:
        if func._is_async:
            result = await func(**args)
        else:
            result = await asyncio.to_thread(func, **args)
        return result if isinstance(result, str) else json.dumps({"status": "success", "result": result})
    except Exception as e:  # noqa: BLE001
        return json.dumps({"status": "error", "error": f"{type(e).__name__}: {e}"})


def _summarize_tool_result(result_str: str) -> tuple[str, str | None]:
    try:
        parsed = json.loads(result_str)
    except (json.JSONDecodeError, TypeError):
        return "unknown", None
    status = parsed.get("status", "unknown")
    extra = []
    for key in ("exit_code", "bytes", "lines", "replacements", "pid"):
        if key in parsed:
            extra.append(f"{key}={parsed[key]}")
    for key in ("matches", "files", "results"):
        if isinstance(parsed.get(key), list):
            extra.append(f"{key}={len(parsed[key])}")
    return status, ", ".join(extra) if extra else None


async def react_loop(
    messages: list[dict[str, Any]],
    tools_payload: list[dict[str, Any]],
    state: AgentState,
    max_iterations: int = MAX_ITERATIONS,
    renderer: MarkdownRenderer | None = None,
    prefix: str = "",
) -> str | None:
    token = _agent_state.set(state)
    md = renderer
    final_content: str | None = None
    tool_names = {t["function"]["name"] for t in tools_payload}
    plan_required = "plan" in tool_names
    verify_required = "verify" in tool_names
    try:
        while state.iteration < max_iterations:
            state.iteration += 1
            if context_size(messages) > CONTEXT_COMPACT_THRESHOLD_CHARS:
                if md is not None:
                    sys.stderr.write(f"{prefix}{md._c('dim', '[context compaction]')}\n")
                messages[:] = await compact_messages(messages)
            try:
                response = await llm_call(messages, tools=tools_payload, tool_choice="auto")
            except RuntimeError as e:
                if md is not None:
                    md.print(f"**LLM error:** `{e}`")
                return None
            msg = response["choices"][0]["message"]
            messages.append(msg)
            tool_calls = msg.get("tool_calls") or []

            if tool_calls:
                if plan_required and state.plan is None and tool_calls[0]["function"]["name"] != "plan":
                    for tc in tool_calls:
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tc["id"],
                            "content": json.dumps({
                                "status": "error",
                                "error": (
                                    "Protocol violation: the first tool call MUST be plan(). "
                                    "Restart with a structured plan."
                                ),
                            }),
                        })
                    continue
                results = await asyncio.gather(*[
                    execute_tool_call(tc, md, prefix=prefix) for tc in tool_calls
                ])
                any_error = False
                for tc, res in zip(tool_calls, results):
                    if len(res) > OUTPUT_CAP_BYTES:
                        res = res[:OUTPUT_CAP_BYTES] + f"\n\n[truncated {len(res)} bytes to {OUTPUT_CAP_BYTES}]"
                    messages.append({"role": "tool", "tool_call_id": tc["id"], "content": res})
                    status, summary = _summarize_tool_result(res)
                    if status == "error":
                        any_error = True
                    if md is not None:
                        tag = "✓" if status == "success" else "✗"
                        style = "ok" if status == "success" else "err"
                        line = f"{prefix}{md._c(style, tag)} {tc['function']['name']}"
                        if summary:
                            line += f" {md._c('dim', '(' + summary + ')')}"
                        sys.stderr.write(line + "\n")
                        sys.stderr.flush()
                state.last_error = any_error
                if any_error:
                    messages.append({
                        "role": "user",
                        "content": (
                            "[reflection-trigger] One or more tool calls returned status=error. "
                            "Call reflect() with the observation, root-cause conclusion, and next action "
                            "before retrying. Do not repeat the same call without diagnosis."
                        ),
                    })
                continue

            content = msg.get("content")
            if content:
                if verify_required and not state.verified and not state.gate_triggered and state.modified_files:
                    state.gate_triggered = True
                    if md is not None:
                        sys.stderr.write(
                            f"{prefix}{md._c('warn', '⚠ verification gate: re-prompting')}\n",
                        )
                        sys.stderr.flush()
                    messages.append({
                        "role": "user",
                        "content": (
                            "[verification-gate] You produced a final answer after modifying files "
                            "without a successful verify(). Call verify() with the command that verifies this project now and "
                            "report the result. If verification truly does not apply, reply starting "
                            "with: 'No verification applicable: <reason>'."
                        ),
                    })
                    continue
                final_content = content
                if md is not None:
                    print()
                    md.print(content)
                    print()
                break
        if state.iteration >= max_iterations and md is not None:
            md.print("**Iteration limit reached:** agent did not converge.")
        return final_content
    finally:
        _agent_state.reset(token)


# ═══════════════════════════════════════════════════════════════════════════════
#  System prompts
# ═══════════════════════════════════════════════════════════════════════════════

SYSTEM_PROMPT = """You are X, an autonomous software engineer running an asynchronous, parallel agent loop with structured planning, automatic reflection on errors, and a verification gate. You build, modify, debug, research, and verify software end to end.

OPERATING PROTOCOL

1. PLAN FIRST. On every new task your VERY FIRST tool call MUST be plan() with goal, steps (each {id, action, depends_on}), success_criteria, and confidence. The harness rejects any other first call.

2. EXECUTE IN PARALLEL. Independent tool calls in a single turn are dispatched concurrently. Batch independent reads, greps, searches, and downloads together.

3. INVESTIGATE BEFORE EDITING. Use grep, glob_files, list_dir, find_symbol, and retrieve to navigate. You MUST read_file (or read_lines) an existing file before edit_file, patch_file, or write_file touches it — the harness enforces this. Prefer edit_file for surgical replacements, patch_file for multi-hunk diffs, create_file for new files, write_file for full rewrites of files you have read.

4. VERIFY BEFORE FINISHING. Whenever you modify files, call verify() with the command that verifies this project before your final answer. The harness rejects a final answer that changed files without a successful verify().

5. REFLECT ON FAILURE. After any tool returns status=error, the harness injects a reflection trigger. Respond with reflect() (observation, conclusion, next_action), then proceed. Never blindly retry the same call.

6. DELEGATE AND SPAWN. Use delegate(task) for a self-contained sub-problem in an isolated in-process context that returns a concise result. Use spawn(task) to launch an independent OS subprocess for truly parallel, long-running, or independent workstreams; track it with swarm_wait, swarm_result, swarm_tail, swarm_kill, swarm_cleanup.

RESEARCH AND VISION
- web_search(query, content, count, images) for normal multi-provider web search; set content=true to include full page text.
- ai_search(query) for a synthesized natural-language answer with citations.
- deep_search(query, content) for slow, thorough multi-step research.
- fetch_url(url) to retrieve a page body via the stealth Chrome client.
- download_file(url, destination) and download_files(urls, destination) to save binary files (images, archives, documents, media) to disk via the stealth client.
- describe_image(prompt, image_path) analyzes a local image with the vision model. It works only on the primary backend and is disabled automatically when the agent falls back to the DeepSeek text-only backend.

TOOL HYGIENE
- run_command has a 300-second default timeout; raise it explicitly for known-long commands.
- Prefer the dedicated navigation tools over shelling out.
- Final replies are a short summary of what changed and how it was verified."""

SUB_AGENT_SYSTEM_PROMPT = """You are a focused sub-agent completing a single scoped task. The harness enforces plan-first, parallel dispatch, error-reflection, and a verification gate.
- Begin with a brief plan() call.
- You must read an existing file before modifying it.
- Investigate, then act with the most specific tools available.
- If you modify files, call verify() before returning.
- Return a concise factual result string, under 2000 characters unless more is essential."""

DEVPLACE_SYSTEM_PROMPT = SYSTEM_PROMPT + """

DEVPLACE CHAT MODE
You are also "botje", reachable through DevPlace chat at https://devplace.net via @mentions and direct messages. You have the complete X-agent tool set and follow the exact same operating protocol described above: plan-first, parallel dispatch, investigate-before-editing, verify-before-finishing, reflect-on-failure, and delegate/spawn for heavy work. Nothing is withheld in chat mode — file operations, shell execution, planning, verification, research, vision, retrieval and the swarm are all available.

Only your FINAL message is delivered to the user as a DevPlace comment or direct message; intermediate tool calls and their output are never shown. Therefore:
- Your final message IS the chat reply. It is posted verbatim to the user. It is NOT a status report, summary, or description of work — it is the message itself, written as botje speaking directly to the user.
- This OVERRIDES the engineering "final reply is a summary of what changed and how it was verified" rule above. In chat mode you do NOT summarise your own process. Never begin with "All done", "Here's a summary", "Context:", "Reply drafted:", "I composed", "I fetched" or any meta-commentary about what you did or were going to do. Just say the thing to the user.
- Write the final reply in concise GitHub-flavoured Markdown suitable for a chat message.
- The user cannot see your tools running, so never tell them to "wait" — just do the work and deliver the answer.
- When the user asks for pictures, photos, images, or anything visual, keep researching with web_search(query, images=true), ai_search, deep_search and fetch_url until you have gathered direct URLs that end in an image extension (.jpg, .jpeg, .png, .gif, .webp, .avif, .svg, .bmp). Do not stop at page links — drill down to the actual image file URLs. Then embed each one DIRECTLY in the final reply using Markdown image syntax `![description](https://.../image.jpg)` so it renders inline. Do NOT merely describe the images in prose, do NOT list them as plain links, and do NOT say you "included" or "attached" them — the actual `![...](...)` markdown for every image must be present in the final message. Provide several relevant images when available."""


def _with_datetime(base: str) -> str:
    return (
        f"{base}\n\nCURRENT DATE AND TIME (set once at startup; it does NOT update during the session — "
        f"call get_current_isodate() whenever you need the live current time): {BOOT_DAY_NAME}, {BOOT_DATETIME}."
    )


def system_message() -> dict[str, Any]:
    return {"role": "system", "content": _with_datetime(SYSTEM_PROMPT)}


# ═══════════════════════════════════════════════════════════════════════════════
#  Agent run helpers
# ═══════════════════════════════════════════════════════════════════════════════


async def run_once(
    prompt: str,
    renderer: MarkdownRenderer | None,
    messages: list[dict[str, Any]] | None = None,
    max_iterations: int = MAX_ITERATIONS,
) -> tuple[str | None, list[dict[str, Any]]]:
    if messages is None:
        messages = [system_message()]
    messages.append({"role": "user", "content": prompt})
    state = AgentState()
    final = await react_loop(
        messages=messages,
        tools_payload=get_tool_payloads(),
        state=state,
        max_iterations=max_iterations,
        renderer=renderer,
    )
    return final, messages


async def interactive(renderer: MarkdownRenderer, max_iterations: int) -> None:
    renderer.print("# X agent\nInteractive mode. Type a task, or `exit` to quit.")
    messages: list[dict[str, Any]] = [system_message()]
    loop = asyncio.get_event_loop()
    while True:
        try:
            line = await loop.run_in_executor(None, lambda: input("\n› "))
        except (EOFError, KeyboardInterrupt):
            break
        line = line.strip()
        if not line:
            continue
        if line.lower() in ("exit", "quit"):
            break
        await run_once(line, renderer, messages=messages, max_iterations=max_iterations)


# ═══════════════════════════════════════════════════════════════════════════════
#  DevPlace bot: mention handler and DM handler (from docs.md)
# ═══════════════════════════════════════════════════════════════════════════════


async def handle_mentions(dp: DevPlace, answered: set[str]) -> None:
    """Poll for unread @mentions and reply to each with the X-agent."""
    try:
        result = dp.call("notifications.list")
    except (xmlrpc.client.Fault, OSError) as e:
        logger.warning("Mention poll error: %s", e)
        return

    for group in result.get("notification_groups", []):
        for entry in group.get("entries", []):
            note = entry["notification"]
            if note.get("type") != "mention":
                continue
            nid = note.get("uid", "")
            if note.get("read") or nid in answered:
                continue
            answered.add(nid)

            message = (note.get("message") or "").strip()
            target_url = note.get("target_url") or ""

            logger.info("New mention: %s", message[:120])

            try:
                match = POST_SLUG_RE.search(target_url)
                if not match:
                    logger.warning("Cannot parse post slug from: %s", target_url)
                    dp.call("notifications.mark.read", notification_uid=nid)
                    continue

                slug = match.group(1)
                detail = dp.call("posts.detail", post_slug=slug)
                post_uid = detail["post"]["uid"]

                # Use the full X-agent to generate an intelligent reply
                reply = await _agent_answer_for_devplace(
                    f"You were @-mentioned in a DevPlace post. The message to you is:\n\n{message}\n\n"
                    f"Do what it asks, then answer it directly. Your reply is posted verbatim as a "
                    f"comment, so write it as botje talking to the user — not as a report about what you did. "
                    f"Write a complete answer; do not truncate or abbreviate it to fit a length limit. "
                    f"A reply longer than the comment limit is automatically split into several "
                    f"comments at sentence and markdown boundaries, so never cut a thought short.",
                    context=f"Post URL: {DEVPLACE_URL}/posts/{slug}",
                )

                count = await _deliver_in_parts(
                    lambda body: dp.call(
                        "comments.create",
                        content=body,
                        target_uid=post_uid,
                        target_type="post",
                    ),
                    reply,
                    COMMENT_CHAR_LIMIT,
                )
                logger.info("Replied to mention on /posts/%s in %d comment(s)", slug, count)

            except (xmlrpc.client.Fault, Exception) as e:
                logger.error("Could not reply to mention: %s", e)
            finally:
                try:
                    dp.call("notifications.mark.read", notification_uid=nid)
                except Exception:
                    pass


async def handle_dms(dp: DevPlace, answered: set[str]) -> None:
    """Poll for unread DMs and reply to each with the X-agent."""
    try:
        inbox = dp.call("messages.inbox")
    except (xmlrpc.client.Fault, OSError) as e:
        logger.warning("DM poll error: %s", e)
        return

    for conversation in inbox.get("conversations", []):
        if not conversation.get("unread"):
            continue
        other = conversation.get("other_user") or {}
        other_uid = other.get("uid")
        other_name = other.get("username", "?")
        if not other_uid:
            continue

        try:
            thread = dp.call("messages.inbox", with_uid=other_uid)
        except xmlrpc.client.Fault as e:
            logger.warning("Cannot open thread with %s: %s", other_name, e)
            continue

        incoming = [m for m in thread.get("messages", []) if not m.get("is_mine")]
        if not incoming:
            continue

        last = incoming[-1]
        mid = last["message"]["uid"]
        if mid in answered:
            continue
        answered.add(mid)

        content = last["message"].get("content", "")
        logger.info("New DM from @%s: %s", other_name, content[:120])

        try:
            reply = await _agent_answer_for_devplace(
                content,
                context=(
                    f"The user @{other_name} (uid {other_uid}) sent you a direct message. "
                    f"Write a complete answer; do not truncate or abbreviate it to fit a length "
                    f"limit. A reply longer than the message limit is automatically split into "
                    f"several messages at sentence and markdown boundaries, so never cut a thought short."
                ),
            )

            count = await _deliver_in_parts(
                lambda body: dp.call("messages.send", content=body, receiver_uid=other_uid),
                reply,
                MESSAGE_CHAR_LIMIT,
            )
            logger.info("Replied to DM from @%s in %d message(s)", other_name, count)

        except (xmlrpc.client.Fault, Exception) as e:
            logger.error("Could not reply to DM from @%s: %s", other_name, e)


async def _agent_answer_for_devplace(
    user_message: str,
    context: str = "",
) -> str:
    """Answer a DevPlace message with the full X-agent, identical in capability to bot.py."""
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": _with_datetime(DEVPLACE_SYSTEM_PROMPT)},
    ]
    if context:
        messages.append({"role": "system", "content": f"Context: {context}"})
    messages.append({"role": "user", "content": user_message})

    state = AgentState()
    final = await react_loop(
        messages=messages,
        tools_payload=get_tool_payloads(),
        state=state,
        max_iterations=DEVPLACE_MAX_ITERATIONS,
        renderer=None,
        prefix="[devplace] ",
    )
    return final or "I couldn't produce an answer. Please try rephrasing your question."


# ═══════════════════════════════════════════════════════════════════════════════
#  DevPlace bot main loop
# ═══════════════════════════════════════════════════════════════════════════════


async def devplace_bot_loop() -> None:
    """Run the DevPlace bot: poll mentions and DMs forever."""
    logger.info("Botje starting — DevPlace bot with full X-agent capabilities")

    if not DEVPLACE_URL or not DEVPLACE_API_KEY:
        logger.error(
            "DEVPLACE_BASE_URL and DEVPLACE_API_KEY are not set. Both are injected "
            "automatically inside a DevPlace-managed container; set them manually "
            "only when running botje.py outside one.",
        )
        return

    logger.info("DevPlace URL: %s", DEVPLACE_URL)
    logger.info("API key: %s...", DEVPLACE_API_KEY[:12])

    dp = DevPlace(DEVPLACE_URL, DEVPLACE_API_KEY)

    # Verify connectivity
    try:
        me = dp.call("notifications.list")
        logger.info(
            "API connection OK — %d notification groups",
            len(me.get("notification_groups", [])),
        )
    except Exception as e:
        logger.error("API connection failed: %s", e)
        logger.error("Check DEVPLACE_URL and DEVPLACE_API_KEY")
        return

    answered_mentions: set[str] = set()
    answered_dms: set[str] = set()
    tick = 0

    logger.info(
        "Bot running. Polling mentions every %ss, DMs every %ss.",
        MENTION_POLL_SECONDS, DM_POLL_SECONDS,
    )

    try:
        while True:
            tick += 1
            try:
                await handle_mentions(dp, answered_mentions)
            except Exception as e:
                logger.error("Mention handler error: %s", e)

            try:
                await handle_dms(dp, answered_dms)
            except Exception as e:
                logger.error("DM handler error: %s", e)

            if tick % 6 == 0:
                logger.debug(
                    "Heartbeat — handled %d mentions, %d DMs so far",
                    len(answered_mentions), len(answered_dms),
                )

            await asyncio.sleep(min(MENTION_POLL_SECONDS, DM_POLL_SECONDS))
    except asyncio.CancelledError:
        logger.info("Bot loop cancelled.")
    finally:
        await close_http_client()


# ═══════════════════════════════════════════════════════════════════════════════
#  Main entry point
# ═══════════════════════════════════════════════════════════════════════════════

async def amain() -> int:
    global MODEL
    parser = argparse.ArgumentParser(
        description="botje — DevPlace bot with full X-agent capabilities",
    )
    parser.add_argument(
        "prompt_pos", nargs="?",
        help="Task prompt (positional). Runs the X-agent and exits. Omit to run as DevPlace bot.",
    )
    parser.add_argument(
        "-p", "--prompt", dest="prompt",
        help="Task prompt; same as the positional argument.",
    )
    parser.add_argument(
        "-b", "--bot", action="store_true",
        help="Force DevPlace bot mode even with a prompt argument present.",
    )
    parser.add_argument(
        "-m", "--model", default=MODEL,
        help="Primary model name (default: molodetz)",
    )
    parser.add_argument(
        "--max-iter", type=int, default=MAX_ITERATIONS,
        help="Maximum agent iterations",
    )
    parser.add_argument(
        "--no-color", action="store_true",
        help="Disable ANSI colour output",
    )
    parser.add_argument(
        "-v", "--verbose", action="store_true",
        help="Enable debug logging",
    )
    args = parser.parse_args()

    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    MODEL = args.model

    # Determine mode: DevPlace bot vs X-agent
    task = args.prompt or args.prompt_pos

    if args.bot or not task:
        # DevPlace bot mode
        await devplace_bot_loop()
        return 0

    # X-agent mode: run the prompt and exit
    renderer = MarkdownRenderer(use_color=not args.no_color)
    try:
        final, _ = await run_once(task, renderer, max_iterations=args.max_iter)
        return 0 if final is not None else 1
    except Exception as e:  # noqa: BLE001
        logger.error("Agent run failed: %s", e)
        return 1
    finally:
        await close_http_client()


def main() -> None:
    try:
        sys.exit(asyncio.run(amain()))
    except KeyboardInterrupt:
        print()  # clean exit
        sys.exit(130)


if __name__ == "__main__":
    main()
