devplacepy/devplacepy/services/containers/files/d.py.bak

#!/usr/bin/env python
from __future__ import annotations
# retoor <retoor@molodetz.nl>
"""
Stealth HTTP client built on top of httpx that reproduces the network
behaviour of a modern Google Chrome desktop browser as faithfully as a
pure-Python stack permits.

WHY THIS IS THE BEST PRACTICAL STEALTH CLIENT
---------------------------------------------
Anti-bot platforms (Cloudflare, Akamai, DataDome, PerimeterX, Imperva and
the like) classify clients on three independent layers. A request only
looks "human" when all three agree with one another. Most scraping code
fails because it spoofs a single layer (usually just the User-Agent string)
while leaving the others screaming "automated tool". This client aligns
every layer that the standard library exposes:

1. TLS layer (JA3 / cipher fingerprint)
   The SSL context is configured with a Chrome-aligned TLS 1.2 cipher list,
   TLS 1.2 as the floor, TLS 1.3 negotiation, and an ALPN advertisement of
   ``h2`` before ``http/1.1`` exactly like Chrome. This moves the JA3 hash
   away from the default Python/OpenSSL fingerprint that detection vendors
   blocklist on sight. Note that OpenSSL does not let Python order the TLS 1.3
   ciphersuites, nor rewrite the supported-groups / signature-algorithm lists,
   so this is Chrome-*aligned* rather than byte-identical (see HONEST LIMITATIONS).

2. HTTP/2 layer (frame and header behaviour)
   Real Chrome speaks HTTP/2 to virtually every modern host. This client
   enables HTTP/2 by default, sends lowercase header names, and preserves a
   Chrome-accurate header *order* — the order itself is a fingerprint that
   naive clients get wrong even when the header values are correct.

3. Application layer (headers and client hints)
   A complete, correctly ordered set of Chrome headers is emitted, including
   the modern User-Agent Client Hints (``sec-ch-ua``, ``sec-ch-ua-mobile``,
   ``sec-ch-ua-platform``) and the request-context ``Sec-Fetch-*`` family.
   The ``Sec-Fetch-*`` values are recomputed per request type so a document
   navigation, a sub-resource fetch and a file download each carry the
   metadata Chrome would actually attach in that situation.

WHAT YOU CAN EXPECT
-------------------
- Requests that pass the overwhelming majority of header- and TLS-based
  bot heuristics, including Cloudflare's "I'm Under Attack" passive checks
  for endpoints that do not mandate a JavaScript challenge.
- Transparent HTTP/2, gzip/deflate/br/zstd decompression, cookie
  persistence across a session, and connection reuse.
- Single and bulk asynchronous file downloads with bounded concurrency,
  streaming to disk (constant memory regardless of file size), filename
  slugification and path-traversal protection.

HONEST LIMITATIONS
------------------
The Python standard ``ssl`` module cannot rewrite the TLS extension order,
the supported-groups list or the signature-algorithm list, so the JA3 hash
produced here is *Chrome-like* rather than *byte-identical* to Chrome. For
targets that fingerprint those low-level extension fields (a small minority,
but a growing one) a native TLS stack such as ``curl_cffi`` or BoringSSL is
required. This client is engineered to be the strongest stealth achievable
without leaving the httpx ecosystem, and it degrades gracefully: every layer
it can control is made indistinguishable from Chrome.

USE CASES
---------
- Resilient web scraping and data collection against header/TLS gating.
- Monitoring, price tracking and availability checks on protected sites.
- Mirroring and bulk asset downloading (images, documents, archives).
- Integration testing of CDN and WAF rules from a realistic client.

EXAMPLE
-------
    import asyncio
    from stealth import ChromeStealthClient

    async def main() -> None:
        async with ChromeStealthClient() as client:
            response = await client.get("https://example.com")
            print(response.status_code, len(response.text))

            results = await client.download_files(
                [
                    "https://example.com/a.jpg",
                    "https://example.com/b.pdf",
                ],
                destination="downloads",
                concurrency=4,
            )
            for result in results:
                print(result.url, result.ok, result.path)

    asyncio.run(main())
"""


import asyncio
import logging
import re
import ssl
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable, Mapping, Sequence
from urllib.parse import unquote, urlsplit

import httpx

logger = logging.getLogger("stealth")


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 as_dict(self) -> dict[str, object]:
        return {
            "url": self.url,
            "final_url": self.final_url,
            "status_code": self.status_code,
            "content_type": self.content_type,
            "data": self.data,
            "truncated": self.truncated,
            "error": self.error,
        }


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
    # Do NOT pin a single ECDH curve: Chrome advertises several groups
    # (X25519, P-256, P-384, plus an MLKEM hybrid), and pinning to one both
    # narrows the supported-groups fingerprint and breaks the handshake with
    # servers that lack that curve. Let OpenSSL advertise its default group set.
    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: httpx.Response) -> 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) -> httpx.Response:
        try:
            return await self._client.request(method, url, headers=headers, **kwargs)
        except httpx.DecodingError:
            # Some CDNs emit a br/zstd stream the decoder rejects. Recover the
            # content by retrying once with only the universally-safe encodings,
            # rather than failing a site that actually serves a valid body.
            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,
    ) -> httpx.Response:
        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) -> httpx.Response:
        return await self.request("GET", url, referer=referer, **kwargs)

    async def post(self, url: str, *, referer: str | None = None, **kwargs: object) -> httpx.Response:
        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)


__all__ = [
    "ChromeStealthClient",
    "ChromeProfile",
    "DownloadResult",
    "BytesResult",
    "build_chrome_ssl_context",
    "resource_kind_for_url",
    "slugify_filename",
]

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 sys
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Optional
from urllib.parse import urlencode, urlparse

def _resolve_llm_endpoint() -> str:
    base = os.environ.get("PRAVDA_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"


LLM_ENDPOINT = _resolve_llm_endpoint()
LLM_BASE_URL = LLM_ENDPOINT.rsplit("/chat/completions", 1)[0]
MODEL = "molodetz"
API_KEY = (
    os.environ.get("PRAVDA_API_KEY")
    or os.environ.get("LLM_API_KEY")
    or 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
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"

logging.basicConfig(level=logging.WARN, format="%(asctime)s %(levelname)s %(name)s %(message)s")
logger = logging.getLogger("x")


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))} {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))


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()


async def stream_subprocess(
    argv: list[str],
    timeout: Optional[int] = None,
    prefix: str = "",
    stdout_sink: Any = None,
    stderr_sink: Any = None,
) -> tuple[str, str, Optional[int], 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,
    )


_http_client: Optional[ChromeStealthClient] = None
_http_lock: Optional[asyncio.Lock] = 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: Optional[str]
    vision: bool


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


def get_backends() -> list[Backend]:
    global _backends
    if _backends is None:
        _backends = [
            Backend("molodetz", LLM_ENDPOINT, MODEL, 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: Optional[list[dict[str, Any]]] = 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: Optional[list[dict[str, Any]]] = 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}")


_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) -> Optional[Callable[..., Any]]:
    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]) -> Optional[str]:
    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


@dataclass
class AgentState:
    plan: Optional[dict[str, Any]] = 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[Optional[AgentState]] = contextvars.ContextVar("agent_state", default=None)


def _state() -> Optional[AgentState]:
    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) -> Optional[str]:
    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


@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: Optional[int] = 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[Optional[str], Optional[str]]:
    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: Optional[str] = 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: Optional[int] = 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?{_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?{_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: Optional[str] = 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 = "hawk .", timeout: int = 600):
    """Run a verification command (linter, tests, validator). Marks the task verified on success.
    command: Shell command, default 'hawk .'.
    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. The startup time is in the system message; use this only when you need the up-to-the-moment time."""
    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: Optional[list] = 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 x.py on a task. Use for truly parallel, independent workstreams.
    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())})


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: Optional[CorpusIndex] = None
_corpus_lock: Optional[asyncio.Lock] = 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


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: Optional[MarkdownRenderer], 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, Optional[str]]:
    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: Optional[MarkdownRenderer] = None,
    prefix: str = "",
) -> Optional[str]:
    token = _agent_state.set(state)
    md = renderer
    final_content: Optional[str] = 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() now (default 'hawk .') 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_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() (default 'hawk .') 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.
"""


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)}


async def run_once(prompt: str, renderer: Optional[MarkdownRenderer], messages: Optional[list[dict[str, Any]]] = None, max_iterations: int = MAX_ITERATIONS) -> tuple[Optional[str], 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)


async def amain() -> int:
    global MODEL
    parser = argparse.ArgumentParser(description="X — single-file autonomous software engineering agent")
    parser.add_argument("prompt_pos", nargs="?", help="Task prompt (positional). With a prompt the agent runs it and exits; without one it starts interactive chat mode.")
    parser.add_argument("-p", "--prompt", dest="prompt", help="Task prompt; same as the positional argument.")
    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
    renderer = MarkdownRenderer(use_color=not args.no_color)
    task = args.prompt or args.prompt_pos

    try:
        if task:
            final, _ = await run_once(task, renderer, max_iterations=args.max_iter)
            return 0 if final is not None else 1
        await interactive(renderer, args.max_iter)
        return 0
    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:
        sys.exit(130)


if __name__ == "__main__":
    main()