retoorandClaude Opus 5
91fac7fd67
A member whose account has not accepted the terms in force now gets one dialog on the action they attempted instead of a dead-end refusal. The client handler is the single TermsGate, wired into every Http POST helper so the four optimistic controllers cannot swallow the gate into an error flash, and the original request is replayed once the acceptance is recorded. Reading the site and deleting an account stay unblocked. apple.md is the source brief the compliance research documents reference. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2338 lines
89 KiB
Python
Executable File
2338 lines
89 KiB
Python
Executable File
#!/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("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"
|
||
|
||
|
||
LLM_ENDPOINT = _resolve_llm_endpoint()
|
||
LLM_BASE_URL = LLM_ENDPOINT.rsplit("/chat/completions", 1)[0]
|
||
MODEL = "molodetz"
|
||
API_KEY = (
|
||
os.environ.get("DEVPLACE_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, 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. 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() 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_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.
|
||
"""
|
||
|
||
|
||
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()
|