diff --git a/src/agent_env/agents/README.md b/src/agent_env/agents/README.md index 4b88aa0c..f4bc9f8a 100644 --- a/src/agent_env/agents/README.md +++ b/src/agent_env/agents/README.md @@ -30,5 +30,12 @@ the `[model]` endpoint as an OpenAI-compatible one and takes its model from the `[model] default` (or the `agent` role) set. It declares `install/v1` too, for Debian or Ubuntu targets. Its folder's `README.md` covers the extensions it implements. +`agents/openhands/` runs the [OpenHands](https://github.com/OpenHands/software-agent-sdk) agent behind the +protocol's A2A agent framework, driving the OpenHands SDK in process rather than a CLI: each turn is one run of a +persisted SDK conversation, the task env's MCP servers go to its `mcp_config`, and its SDK events are the turn's +trajectory. Name it as `@local/agentenv-framework/agents/openhands`. It calls the `[model]` endpoint through LiteLLM +and takes its model from the prompt, so its smoke task needs `[model] default` (or the `agent` role) set. It declares +`install/v1` too, for Debian or Ubuntu targets. Its folder's `README.md` covers the extensions it implements. + `agent-env run agents` runs every agent's smoke task, building each image; `--task claude-code-smoke` runs Claude -Code's alone, `--task pi-smoke` pi's. +Code's alone, `--task pi-smoke` pi's, `--task openhands-smoke` OpenHands'. diff --git a/src/agent_env/agents/agents/openhands/Dockerfile b/src/agent_env/agents/agents/openhands/Dockerfile new file mode 100644 index 00000000..d63d77ab --- /dev/null +++ b/src/agent_env/agents/agents/openhands/Dockerfile @@ -0,0 +1,13 @@ +# docker build -t openhands-a2a src/agent_env/agents/agents/openhands +FROM mirror.gcr.io/library/ubuntu:24.04@sha256:534baea6a22c03a63003dbc8dbe78fe34bc0d7e595d9a9dc9834884ff530eb55 +WORKDIR /app +# Named sources: agent-env packs the build context install/v1 copies from the COPY sources. +COPY a2a_server.py conversation_run.py install.sh openhands_agent.py openhands_runtime.py requirements.txt start.sh ./ +RUN sh install.sh \ + && useradd --create-home agent \ + && mkdir /workspace \ + && chown agent /app /workspace +ENV PATH="/opt/openhands-a2a-venv/bin:${PATH}" OPENHANDS_SUPPRESS_BANNER=1 +USER agent +EXPOSE 8000 +CMD ["python", "a2a_server.py"] diff --git a/src/agent_env/agents/agents/openhands/README.md b/src/agent_env/agents/agents/openhands/README.md new file mode 100644 index 00000000..13c32d6e --- /dev/null +++ b/src/agent_env/agents/agents/openhands/README.md @@ -0,0 +1,67 @@ +# OpenHands A2A agent + +The [OpenHands](https://github.com/OpenHands/software-agent-sdk) agent wrapped with the +`agentenv_protocol.a2a_agent` framework, shipped in the `agents` bundle as +`@local/agentenv-framework/agents/openhands`. This directory is the whole build context: the +protocol package and the OpenHands SDK come from PyPI (`requirements.txt`), and `install.sh` installs +them in a Python of uv's own, with tmux and git for the agent's terminal, both for the image and for `install/v1`. + +Instead of wrapping a CLI, the agent drives the OpenHands SDK in process. `openhands_agent.py` is the +A2A side and imports no SDK; `openhands_runtime.py` runs the conversation. Each task is one `arun()` +of an SDK conversation in `/workspace`, with the SDK's default tools (terminal, file editor, task +tracker; no browser) and its default condenser, which summarizes a conversation that outgrows the model's window. +The model is called through the `[model]` LiteLLM endpoint as `litellm_proxy/`, with `LITELLM_EXTRA_HEADERS` +sent on every call but kept out of the persisted conversation, which a snapshot uploads. Each turn runs on a thread +and event loop of its own. + +A context maps to one conversation, whose id is derived from the context id, persisted in +`~/.openhands-a2a/conversations/`, so follow-up tasks in the context resume it, across restarts too. +Images are sent to the model as images; every file part is also saved in the conversation's own directory under +`~/.openhands-a2a/inputs` and its path added to the prompt, for the agent's tools, and the conversation's snapshots +carry it. + +| Extension | Implementation | +|---|---| +| `agent-config/v1` | `OpenHandsConfig`: `model`, `effort` (the LLM's reasoning effort), `system_prompt` (replaces OpenHands' own), `append_system_prompt`, `max_iterations`, `timeout_seconds` | +| `mcp-config/v1` | SDK default; each server becomes an HTTP `MCPServer` in the agent's `mcp_config`, with its headers, sent as written: a `${VAR}` in a URL or header is not filled from the agent's environment | +| `skill-config/v1` | inline and bundle skills installed as `SKILL.md` directories under `~/.openhands-a2a/skills`, loaded into the agent's context | +| `trajectory/v1` | live: the task's OpenHands SDK events, appended as the conversation emits them (format `openhands-events/v1`) | +| `snapshot/v1` | `save` uploads the context's persisted conversation, with the files attached in the context, as a gzipped tar and, when asked, the workspace as another; `load` extracts both aside, checks the conversation and points it at the target context's paths before it replaces anything, then replaces the workspace's contents, puts the attached files, as they were saved, in the target's own directory and installs the conversation under the target context's id, so an archive that fails to extract, or holds no conversation, changes nothing | +| `peer-agents/v1` | SDK default: the `peer_agents` MCP server reaches the agent like any other | +| `install/v1` | copies this directory into the task container as `/opt/openhands-a2a`, runs `install.sh` (Debian or Ubuntu, as root, with network access), writes the LiteLLM credentials to a 0600 `agent.env` through a heredoc and starts `start.sh`, which sources it | +| `triggers/v1` | SDK default | + +`OPENHANDS_A2A_MODEL` sets the model when agent-config sets none, as for a peer that other agents +message directly, and `OPENHANDS_A2A_WORKSPACE` the workspace (default `/workspace`). `a2a_server.py` reads +these and the `LITELLM_*` variables once, at startup: a `LITELLM_EXTRA_HEADERS` that is not a JSON object stops +the agent there, while a missing `LITELLM_BASE_URL` fails each task as `openhands.no_endpoint`. + +Snapshot `load` answers 409 while a task is running, and 400 for an archive that holds no conversation. A +snapshot `save` waits for its context's running task to end, and a conversation runs one task at a time; one that +saves the workspace also waits for the tasks running in other contexts, which work in the same workspace. + +Usage is the conversation's spend over the task: tokens, cost as LiteLLM prices it, and tool calls. A +failed run's error code is `openhands.` of the SDK's last error event (`openhands.LLMAuthenticationError`, +`openhands.MaxIterationsReached`, ...). It is typed by the SDK's classification when the event carries one; +otherwise a run that raised is an `infra_error`, and one a run limit ended, or that ends stuck or paused, an +`agent_error`. A turn the SDK can't start, such as one whose persisted conversation it can't resume, fails as +`openhands.start` and one the agent's shutdown stops as `openhands.shutdown`, both `infra_error`s. A timeout +(`openhands.timeout`) counts reading the task's file parts too, reports what the run had spent, and interrupts the +conversation, which the context's next task resumes. + +A timeout, a `tasks/cancel` and the server shutting down each interrupt the run and close its tools, which kills +the shell commands it started. The SDK's terminal tool does not notice its closed session, though: it keeps waiting +on a command that was running until the command's own timeout, or its 30 s no-change timeout when the command has +none, and holds the run, and the agent's exit, until then. +Errors are answered as `{"detail": ...}`. + +`tst/integration/cli/run_bundle_openhands_test.py` runs the built image against a fake OpenAI endpoint. + +```bash +docker build -t openhands-a2a src/agent_env/agents/agents/openhands +docker run -p 8000:8000 -e LITELLM_BASE_URL=... -e LITELLM_API_KEY=... openhands-a2a +pytest tst/unit/agents/openhands +``` + +The OpenHands version is pinned in `requirements.txt` and `OPENHANDS_VERSION`; a new release is a pull +request that bumps both. diff --git a/src/agent_env/agents/agents/openhands/a2a_server.py b/src/agent_env/agents/agents/openhands/a2a_server.py new file mode 100644 index 00000000..bafff742 --- /dev/null +++ b/src/agent_env/agents/agents/openhands/a2a_server.py @@ -0,0 +1,24 @@ +"""Serves the OpenHands agent, configured from its environment. ``AGENT_CARD`` is what agent-env's install/v1 reads +from ``/app/a2a_server.py``.""" + +import os +from pathlib import Path +from typing import Final + +from a2a.types import AgentCard +from starlette.applications import Starlette + +from openhands_agent import DEFAULT_MODEL_ENV, Endpoint, OpenHandsAgent +from openhands_runtime import SdkRuntime + +agent: Final[OpenHandsAgent] = OpenHandsAgent( + SdkRuntime(), + endpoint=Endpoint.from_environ(os.environ), + default_model=os.environ.get(DEFAULT_MODEL_ENV), + workspace=Path(os.environ.get("OPENHANDS_A2A_WORKSPACE", "/workspace")), +) +app: Final[Starlette] = agent.create_app() +AGENT_CARD: Final[AgentCard] = app.state.agentenv_a2a.card + +if __name__ == "__main__": + agent.serve() diff --git a/src/agent_env/agents/agents/openhands/conversation_run.py b/src/agent_env/agents/agents/openhands/conversation_run.py new file mode 100644 index 00000000..8a0cf2d2 --- /dev/null +++ b/src/agent_env/agents/agents/openhands/conversation_run.py @@ -0,0 +1,87 @@ +"""One ``arun()`` of a conversation, on a thread and an event loop of its own. The SDK does blocking work between its +awaits (tools, persistence), which on the server's loop would stall every other task and the agent's own HTTP +endpoints, the peer-agents MCP server among them; and a run on the loop's default executor would hold one of the few +threads the server's other blocking calls share for as long as it runs.""" + +import asyncio +import threading +from contextlib import suppress +from typing import Protocol + + +class Conversation(Protocol): + async def arun(self) -> None: ... + + def interrupt(self) -> None: ... + + def close(self) -> None: ... + + +class ConversationRun: + def __init__(self, conversation: Conversation) -> None: + self.conversation = conversation + self.stopped = threading.Event() + self._loop: asyncio.AbstractEventLoop | None = None + self._lock = threading.Lock() + + def stop(self) -> None: + """Interrupt the run, or keep it from starting, and close its tool executors: the terminal tool does not heed + the interrupt, and a command it is running would otherwise hold the run until the command ends.""" + with self._lock: + self.stopped.set() + loop = self._loop + if loop is not None: + # A run that ends meanwhile closes its loop, and then has nothing left to interrupt. + with suppress(RuntimeError): + loop.call_soon_threadsafe(self.conversation.interrupt) + self.conversation.close() + + async def to_its_end(self) -> None: + """Run to the end; cancelled, stop the run, wait until it has stopped and re-raise.""" + loop = asyncio.get_running_loop() + done: asyncio.Future[None] = loop.create_future() + + def settle(error: BaseException | None) -> None: + if done.done(): + return + if error is None: + done.set_result(None) + else: + done.set_exception(error) + + def drive() -> None: + error: BaseException | None = None + try: + self._drive() + except BaseException as exc: + error = exc + # The server's loop is closed once it has stopped, and then nothing waits for the run. + with suppress(RuntimeError): + loop.call_soon_threadsafe(settle, error) + + threading.Thread(target=drive, name="openhands-run").start() + try: + await asyncio.shield(done) + except asyncio.CancelledError: + if not done.done(): + await asyncio.to_thread(self.stop) + await asyncio.wait([done]) + if not done.cancelled(): + done.exception() # the caller was cancelled, so how the run ended is no longer its to see + raise + + def _drive(self) -> None: + # Made on this thread, so the server's thread keeps its own event loop. Closing the runner cancels what the + # run left on its loop, LiteLLM's logging worker among them. + with asyncio.Runner() as runner: + with self._lock: + if self.stopped.is_set(): + return + self._loop = runner.get_loop() + runner.run(self._arun()) + + async def _arun(self) -> None: + # A stop sets the flag before it sends the interrupt here, so one that arrives before arun() has recorded the + # task it would cancel, and so only pauses, which arun() undoes, keeps the run from starting. + if not self.stopped.is_set(): + await self.conversation.arun() diff --git a/src/agent_env/agents/agents/openhands/install.sh b/src/agent_env/agents/agents/openhands/install.sh new file mode 100644 index 00000000..9a454cc8 --- /dev/null +++ b/src/agent_env/agents/agents/openhands/install.sh @@ -0,0 +1,32 @@ +#!/bin/sh +# Installs tmux and git for the agent's terminal and this agent's Python environment, in uv's own Python so the +# target's is never used, on Debian or Ubuntu, as root. The image build and install/v1 both run it, so an install +# into a task container matches the image. +set -eu +UV_VERSION=0.13.1 +UV_SHA256_X86_64=6c86040a8f045ab9adc9e3ec4cdafa03fd00e9a988ed2bf28971c7fc2d0debd0 +UV_SHA256_AARCH64=345a8308e4d1835cdfaadf692836fd3137a6d45864773f05ad7bf59442b1ba16 +PYTHON_VERSION=3.12 +here=$(cd "$(dirname "$0")" && pwd) + +export DEBIAN_FRONTEND=noninteractive +apt-get update +apt-get install -y --no-install-recommends ca-certificates curl git tmux +rm -rf /var/lib/apt/lists/* + +case "$(uname -m)" in + x86_64) arch=x86_64 sha256=$UV_SHA256_X86_64 ;; + aarch64 | arm64) arch=aarch64 sha256=$UV_SHA256_AARCH64 ;; + *) echo "unsupported architecture: $(uname -m)" >&2; exit 1 ;; +esac +stage=$(mktemp -d) +trap 'rm -rf "$stage"' EXIT +archive=uv-$arch-unknown-linux-musl.tar.gz +curl -fsSL --retry 3 "https://github.com/astral-sh/uv/releases/download/$UV_VERSION/$archive" -o "$stage/$archive" +echo "$sha256 $stage/$archive" | sha256sum -c - +tar -xzf "$stage/$archive" -C "$stage" +uv="$stage/uv-$arch-unknown-linux-musl/uv" + +export UV_PYTHON_INSTALL_DIR=/opt/openhands-a2a-python UV_PYTHON_PREFERENCE=only-managed UV_CACHE_DIR="$stage/cache" +"$uv" venv --quiet --clear --python "$PYTHON_VERSION" /opt/openhands-a2a-venv +"$uv" pip install --quiet --python /opt/openhands-a2a-venv/bin/python -r "$here/requirements.txt" diff --git a/src/agent_env/agents/agents/openhands/openhands_agent.py b/src/agent_env/agents/agents/openhands/openhands_agent.py new file mode 100644 index 00000000..02865253 --- /dev/null +++ b/src/agent_env/agents/agents/openhands/openhands_agent.py @@ -0,0 +1,563 @@ +"""OpenHands (https://github.com/OpenHands/software-agent-sdk) as an AgentEnv A2A agent. + +Each task is one run of an OpenHands SDK conversation in the workspace. A context maps to one persisted conversation, +so later tasks in the context resume it. This module is the A2A side and imports no SDK; ``openhands_runtime`` runs +the conversation. +""" + +import asyncio +import base64 +import io +import json +import mimetypes +import shutil +import tarfile +import tempfile +import threading +import uuid +from collections.abc import AsyncIterator, Callable, Mapping +from contextlib import asynccontextmanager, suppress +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Final, Literal, Protocol, Self +from urllib.parse import urlsplit + +import httpx +from agentenv_protocol import ROLE_HEADER +from agentenv_protocol.a2a_agent import ( + INSTALL_V1, + MCP_CONFIG_V1, + PEER_AGENTS_V1, + SNAPSHOT_V1, + TRAJECTORY_V1, + TRIGGERS_V1, + AgentConfig, + AgentEnvAgent, + AgentIdentity, + BundleSkillRequest, + DataPart, + FilePart, + InlineSkillRequest, + ObjectSnapshotLoadRequest, + ObjectSnapshotSaveRequest, + SKILL_CONFIG_V1, + TaskRequest, + TaskResult, + TextPart, + Usage, + a2a_agent, + download, + enable, + extension, + upload, +) +from starlette.applications import Starlette +from starlette.exceptions import HTTPException +from starlette.requests import Request +from starlette.responses import JSONResponse, Response + +OPENHANDS_VERSION: Final[str] = "1.57.0" +TRAJECTORY_FORMAT: Final[str] = "openhands-events/v1" +# The model when agent-config sets none, as for a peer that other agents message directly. +DEFAULT_MODEL_ENV: Final[str] = "OPENHANDS_A2A_MODEL" +A2A_PORT: Final[int] = 8000 +INSTALL_DIR: Final[str] = "/opt/openhands-a2a" +ENV_FILE: Final[str] = "agent.env" +BASE_STATE: Final[str] = "base_state.json" +# Where a conversation's archive keeps the files attached to it, and the directory they were saved in. +INPUTS: Final[str] = ".openhands-a2a-inputs" +INPUTS_SOURCE: Final[str] = ".openhands-a2a-inputs.json" +FETCH_TIMEOUT_SECONDS: Final[int] = 120 +MAX_ATTACHMENT_BYTES: Final[int] = 256 * 1024 * 1024 +# A conversation id is derived from its context id, so a context finds its conversation across restarts. +_CONVERSATIONS: Final[uuid.UUID] = uuid.uuid5(uuid.NAMESPACE_URL, "urn:agentenv:agents:openhands") +# Images go to the model as images as well as to disk; any other file is left on disk for the agent's tools. +_IMAGE_TYPES: Final[frozenset[str]] = frozenset({"image/png", "image/jpeg", "image/gif", "image/webp"}) +_NO_BODY_STATUSES: Final[frozenset[int]] = frozenset({204, 205, 304}) + +# sudo only where it needs no password: VM sandboxes have it, a local sandbox runs docker as the user. +_DOCKER: Final[str] = "$(sudo -n true 2>/dev/null && echo sudo) docker" +INSTALL_COMMANDS: Final[list[str]] = [ + f"{_DOCKER} cp {{agent_ctx_tar}} {{container}}:/tmp/openhands-a2a-ctx.tar.gz", + f"{_DOCKER} exec {{container}} sh -c 'mkdir -p {INSTALL_DIR} && tar -xf /tmp/openhands-a2a-ctx.tar.gz -C {INSTALL_DIR}" + " && rm /tmp/openhands-a2a-ctx.tar.gz'", + f"{_DOCKER} exec {{container}} sh {INSTALL_DIR}/install.sh", + # The credentials travel on a heredoc into a 0600 file: the installer logs only a command's first line, + # and neither a command line nor the container's config ever holds them. + f"{_DOCKER} exec -i {{container}} sh -c 'umask 077 && cat > {INSTALL_DIR}/{ENV_FILE}' <<'OPENHANDS_A2A_ENV'\n" + "LITELLM_API_KEY={litellm_api_key}\n" + "LITELLM_BASE_URL={litellm_base_url}\n" + "LITELLM_EXTRA_HEADERS={litellm_extra_headers}\n" + "OPENHANDS_A2A_ENV", + f"{_DOCKER} exec -d -e A2A_PORT={{a2a_port}} {{container}} sh {INSTALL_DIR}/start.sh", +] + + +class OpenHandsConfig(AgentConfig): + model: str | None = None + system_prompt: str | None = None + append_system_prompt: str | None = None + effort: Literal["none", "low", "medium", "high", "xhigh"] | None = None + max_iterations: int = 500 + timeout_seconds: int = 1800 + + +@dataclass(frozen=True, slots=True) +class Endpoint: + """The LiteLLM gateway the model is called through, with the headers every call carries.""" + + base_url: str + api_key: str | None = None + extra_headers: Mapping[str, str] = field(default_factory=dict) + + @classmethod + def from_environ(cls, environ: Mapping[str, str]) -> Self | None: + """The endpoint the ``LITELLM_*`` variables name; None without a base URL, which fails each task instead.""" + base_url = environ.get("LITELLM_BASE_URL") + if not base_url: + return None + try: + headers = json.loads(environ.get("LITELLM_EXTRA_HEADERS") or "{}") + except json.JSONDecodeError as exc: + raise ValueError(f"LITELLM_EXTRA_HEADERS is not valid JSON: {exc}") from exc + if not isinstance(headers, dict): + raise ValueError("LITELLM_EXTRA_HEADERS must be a JSON object of header names to values") + return cls(base_url, environ.get("LITELLM_API_KEY"), headers) + + +@dataclass(frozen=True, slots=True) +class Turn: + """One task, as the runtime runs it.""" + + conversation_id: str + conversations_dir: Path + workspace: Path + text: str + images: tuple[str, ...] + mcp_servers: Mapping[str, Any] + skills_dir: Path + skill_names: tuple[str, ...] + config: OpenHandsConfig + endpoint: Endpoint + on_event: Callable[[dict[str, Any]], None] + # What a run that was cancelled had spent by then. + on_usage: Callable[[Usage], None] + + +@dataclass(frozen=True, slots=True) +class Outcome: + text: str + usage: Usage + error_code: str | None = None + error_type: Literal["agent_error", "infra_error"] = "agent_error" + + +@dataclass(slots=True) +class _ConversationLock: + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + holders: int = 0 + + +class Runtime(Protocol): + async def run(self, turn: Turn) -> Outcome: + """Run ``turn`` to its end. Cancelled, it interrupts the conversation and re-raises once it has stopped.""" + ... + + def stop(self) -> None: + """Interrupt every turn still running: the server is shutting down.""" + ... + + +async def _detail_body(_request: Request, exc: HTTPException) -> Response: + """An ``HTTPException`` as the ``{"detail": ...}`` body agent-env reads errors from.""" + if exc.status_code < 200 or exc.status_code in _NO_BODY_STATUSES: + return Response(status_code=exc.status_code, headers=exc.headers) + return JSONResponse({"detail": exc.detail}, status_code=exc.status_code, headers=exc.headers) + + +@asynccontextmanager +async def _turns_stop_with_the_server(app: Starlette) -> AsyncIterator[None]: + """The server's lifespan: as it shuts down, turns still running are interrupted, so their runs, in threads of + their own, do not keep the process alive after the server.""" + try: + yield + finally: + app.state.agentenv_a2a.agent.runtime.stop() + + +def conversation_id(context_id: str) -> str: + return uuid.uuid5(_CONVERSATIONS, context_id).hex + + +def _with_role(mcp_servers: Mapping[str, Any], role: str | None) -> dict[str, Any]: + """The servers with the role header, which the env gateway filters tools by and which wins over a stored role.""" + if not role: + return dict(mcp_servers) + return { + name: {**registration, "headers": {**(registration.get("headers") or {}), ROLE_HEADER: role}} + for name, registration in mcp_servers.items() + } + + +def _media_type(part: FilePart) -> str: + return part.mime_type or mimetypes.guess_type(part.name or "")[0] or "application/octet-stream" + + +async def _materialize(part: FilePart, index: int, directory: Path) -> Path: + """Write a file part to disk; inline bytes and HTTP(S) URLs only.""" + name = Path(part.name).name if part.name else "" + destination = directory / f"{index}-{name or 'file'}" + if part.bytes is not None: + destination.write_bytes(base64.b64decode(part.bytes)) + return destination + if urlsplit(part.uri).scheme not in {"http", "https"}: + raise ValueError(f"unsupported file URI scheme for {name or 'file'}") + received = 0 + async with ( + httpx.AsyncClient(timeout=FETCH_TIMEOUT_SECONDS, follow_redirects=True) as client, + client.stream("GET", part.uri) as response, + ): + response.raise_for_status() + with destination.open("wb") as handle: + async for chunk in response.aiter_bytes(): + received += len(chunk) + if received > MAX_ATTACHMENT_BYTES: + raise ValueError(f"{name or 'file'} is larger than {MAX_ATTACHMENT_BYTES} bytes") + handle.write(chunk) + return destination + + +def _pack(directory: Path, archive: Path, inputs: Path | None = None) -> None: + """``directory``'s contents and, for a conversation, the files attached to it, from ``inputs``, under ``INPUTS``, + with the directory they were saved in.""" + with tarfile.open(archive, "w:gz") as tar: + for child in sorted(directory.iterdir()): + tar.add(child, arcname=child.name) + if inputs is not None and inputs.is_dir(): + tar.add(inputs, arcname=INPUTS) + source = json.dumps({"directory": str(inputs)}).encode() + info = tarfile.TarInfo(INPUTS_SOURCE) + info.size = len(source) + tar.addfile(info, io.BytesIO(source)) + + +def _extract(archive: Path, directory: Path) -> Path: + """The archive extracted into a new staging directory inside ``directory``.""" + staging = directory / f".openhands-a2a-restore-{uuid.uuid4().hex}" + # An empty workspace's archive has no members, and extracting it creates nothing. + staging.mkdir() + try: + with tarfile.open(archive) as tar: + tar.extractall(staging, filter="data") + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + return staging + + +def _replace_contents(directory: Path, staging: Path) -> None: + """Replace ``directory``'s contents with those of ``staging``, a directory inside it.""" + for child in directory.iterdir(): + if child == staging: + continue + if child.is_dir() and not child.is_symlink(): + shutil.rmtree(child) + else: + child.unlink() + for child in staging.iterdir(): + child.rename(directory / child.name) + staging.rmdir() + + +def _repoint(directory: Path, old: str, new: str, skip: Path) -> None: + """Make the files in ``directory``, but for those under ``skip``, name paths under ``new`` where they name paths + under ``old``, whether written as is or JSON-escaped.""" + replacements = {f"{old}/": f"{new}/", f"{json.dumps(old)[1:-1]}/": f"{json.dumps(new)[1:-1]}/"} + for path in directory.rglob("*"): + if not path.is_file() or path.is_symlink() or path.is_relative_to(skip): + continue + try: + text = path.read_text(encoding="utf-8") + except UnicodeDecodeError: + continue + repointed = text + for before, after in replacements.items(): + repointed = repointed.replace(before, after) + if repointed != text: + path.write_text(repointed, encoding="utf-8") + + +def _stage_conversation(archive: Path, conversations_dir: Path, target_id: str, inputs: Path) -> Path: + """The archived conversation extracted beside the others, rewritten to be ``target_id``'s with its attached files + in ``inputs``: the SDK resumes a conversation only under the id and directory its state records, and its messages + name the files by path. Raises ValueError for an archive that holds no conversation.""" + staging = _extract(archive, conversations_dir) + try: + state = json.loads((staging / BASE_STATE).read_text()) + uuid.UUID(state["id"]) + if not isinstance(state.get("agent"), dict): + raise ValueError("its state names no agent") + state.update(id=str(uuid.UUID(target_id)), persistence_dir=str(conversations_dir / target_id)) + source = staging / INPUTS_SOURCE + saved_in = json.loads(source.read_text())["directory"] if source.is_file() else None + except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc: + shutil.rmtree(staging, ignore_errors=True) + raise ValueError(f"the trajectory object holds no OpenHands conversation: {type(exc).__name__}: {exc}") from exc + (staging / BASE_STATE).write_text(json.dumps(state)) + if saved_in is not None: + source.unlink() + _repoint(staging, saved_in, str(inputs), skip=staging / INPUTS) + return staging + + +@a2a_agent( + identity=AgentIdentity( + name="OpenHands", + description="The OpenHands agent: reads, edits and runs code in its workspace with MCP tools and skills.", + version=OPENHANDS_VERSION, + input_modes=("text", "image/png", "image/jpeg", "image/gif", "image/webp"), + ), + config=OpenHandsConfig, + config_description="Configure the OpenHands runtime: model, effort, system prompt and iteration limit.", + lifespan=_turns_stop_with_the_server, + extensions=( + MCP_CONFIG_V1, + enable(TRAJECTORY_V1, live=True, description="The task's OpenHands SDK events, also while it runs."), + enable( + SNAPSHOT_V1, + description=( + "The trajectory object is the context's persisted OpenHands conversation, the workspace object a " + "gzipped tar." + ), + ), + PEER_AGENTS_V1, + enable( + INSTALL_V1, + install_commands=INSTALL_COMMANDS, + required_params=[ + "container", "agent_ctx_tar", "a2a_port", "litellm_api_key", "litellm_base_url", + "litellm_extra_headers", + ], + a2a_port=A2A_PORT, + ), + TRIGGERS_V1, + ), +) +class OpenHandsAgent(AgentEnvAgent): + def __init__( + self, + runtime: Runtime, + *, + endpoint: Endpoint | None, + default_model: str | None = None, + home: Path | None = None, + workspace: Path | None = None, + ) -> None: + self.runtime: Runtime = runtime + self.endpoint: Endpoint | None = endpoint + self.default_model: str | None = default_model + self.home: Path = home or Path.home() / ".openhands-a2a" + self.workspace: Path = workspace or Path("/workspace") + self.conversations_dir: Path = self.home / "conversations" + self.skills_dir: Path = self.home / "skills" + # The files attached in each conversation, in a directory of its own, which its snapshots carry. + self.inputs_dir: Path = self.home / "inputs" + # The framework already runs a context's tasks one at a time; this lets a snapshot save wait for the task + # running in its context. An entry lives only while it is held, so contexts never seen again leave nothing. + self._conversation_locks: dict[str, _ConversationLock] = {} + # A task starts once no restore or workspace save holds this. A restore is refused while tasks run, and a + # workspace save waits for the tasks running in other contexts to end. + self._workspace_lock: asyncio.Lock = asyncio.Lock() + self._running: int = 0 + self._idle: asyncio.Event = asyncio.Event() + self._idle.set() + + def create_app(self) -> Starlette: + app = super().create_app() + app.add_exception_handler(HTTPException, _detail_body) + return app + + async def run(self, request: TaskRequest[OpenHandsConfig]) -> TaskResult: + config = request.config + if not config.model and self.default_model: + config = config.model_copy(update={"model": self.default_model}) + if not config.model: + return TaskResult.failure("openhands.no_model", "no model is configured; set 'model' through agent-config") + if self.endpoint is None: + return TaskResult.failure("openhands.no_endpoint", "LITELLM_BASE_URL is not set", error_type="infra_error") + workspace = request.workspace or self.workspace + workspace.mkdir(parents=True, exist_ok=True) + request.trajectory.set_format(TRAJECTORY_FORMAT) + # The SDK calls back from its own threads too; the log takes one appender at a time. + appending = threading.Lock() + + def on_event(event: dict[str, Any]) -> None: + # A cancelled task's log ends before its run has stopped, and refuses the interrupt's last events. + with appending, suppress(RuntimeError): + request.trajectory.append(event) + + conversation = conversation_id(request.context_id) + # Bound before the run starts, so a live trajectory read or a mid-turn snapshot finds the conversation. + self.set_session_ref_for_context(request.context_id, conversation) + spent: list[Usage] = [] + async with self._holding(conversation), self._in_workspace(): + try: + # Reading the file parts is part of the task, and so counts toward its time. + async with asyncio.timeout(config.timeout_seconds): + try: + text, images = await self._prompt(request, conversation) + except (ValueError, OSError, httpx.HTTPError, httpx.InvalidURL) as exc: + return TaskResult.failure("openhands.attachment", f"could not read a file part: {exc}") + outcome = await self.runtime.run(Turn( + conversation_id=conversation, + conversations_dir=self.conversations_dir, + workspace=workspace, + text=text, + images=images, + mcp_servers=_with_role(request.mcp_servers, config.role), + skills_dir=self.skills_dir, + skill_names=tuple(str(skill["name"]) for skill in request.skills), + config=config, + endpoint=self.endpoint, + on_event=on_event, + on_usage=spent.append, + )) + except TimeoutError: + message = f"OpenHands did not finish within {config.timeout_seconds}s" + timed_out = TaskResult.builder().failed("openhands.timeout", message).add_text(message) + return (timed_out.usage(spent[-1]) if spent else timed_out).session_ref(conversation).build() + builder = TaskResult.builder().session_ref(conversation).usage(outcome.usage) + if outcome.error_code is None: + return builder.succeeded().add_text(outcome.text).build() + return builder.failed(outcome.error_code, outcome.text, error_type=outcome.error_type).add_text(outcome.text).build() + + async def _prompt(self, request: TaskRequest[OpenHandsConfig], conversation: str) -> tuple[str, tuple[str, ...]]: + """The message text and its images as data URIs. Files are kept on disk, in the conversation's own directory, + for its later turns and its snapshots.""" + files_dir = self.inputs_dir / conversation / request.task_id + chunks: list[str] = [] + images: list[str] = [] + for index, part in enumerate(request.parts): + if isinstance(part, TextPart): + chunks.append(part.text) + elif isinstance(part, DataPart): + chunks.append(json.dumps(part.data, indent=2, default=str)) + elif isinstance(part, FilePart): + files_dir.mkdir(parents=True, exist_ok=True) + path = await _materialize(part, index, files_dir) + media_type = _media_type(part) + if media_type in _IMAGE_TYPES: + images.append(f"data:{media_type};base64,{base64.b64encode(path.read_bytes()).decode()}") + chunks.append(f"[Attached file ({media_type}) saved at {path}]") + return "\n\n".join(chunks), tuple(images) + + @asynccontextmanager + async def _holding(self, conversation: str) -> AsyncIterator[None]: + entry = self._conversation_locks.setdefault(conversation, _ConversationLock()) + entry.holders += 1 + try: + async with entry.lock: + yield + finally: + entry.holders -= 1 + if not entry.holders: + del self._conversation_locks[conversation] + + @asynccontextmanager + async def _in_workspace(self) -> AsyncIterator[None]: + async with self._workspace_lock: + self._running += 1 + self._idle.clear() + try: + yield + finally: + self._running -= 1 + if not self._running: + self._idle.set() + + def _require_idle(self) -> None: + if self._running: + raise HTTPException(status_code=409, detail="a task is running; restore the workspace once it finishes") + + @extension(SNAPSHOT_V1.save) + async def save_snapshot(self, request: ObjectSnapshotSaveRequest) -> dict[str, Any]: + conversation = conversation_id(request.context_id) + directory = self.conversations_dir / conversation + with tempfile.TemporaryDirectory() as scratch: + # The conversation and the workspace are copied together, so a restore cannot fall between them. + conversation_archive = Path(scratch) / "conversation.tar.gz" + workspace_archive = Path(scratch) / "workspace.tar.gz" + async with self._holding(conversation), self._workspace_lock: + if not (directory / BASE_STATE).is_file(): + raise HTTPException(status_code=404, detail=f"No conversation for context '{request.context_id}'") + await asyncio.to_thread(_pack, directory, conversation_archive, self.inputs_dir / conversation) + if request.objects.workspace is not None: + # Tasks in other contexts work in the same workspace: those running end first, and new ones wait + # for the lock this holds. + await self._idle.wait() + self.workspace.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(_pack, self.workspace, workspace_archive) + objects = {"trajectory": (await upload(request.objects.trajectory, conversation_archive)).model_dump(mode="json")} + if request.objects.workspace is not None: + objects["workspace"] = (await upload(request.objects.workspace, workspace_archive)).model_dump(mode="json") + return {"context_id": request.context_id, "objects": objects} + + @extension(SNAPSHOT_V1.load) + async def load_snapshot(self, request: ObjectSnapshotLoadRequest) -> dict[str, Any]: + context_id = request.target_context_id or uuid.uuid4().hex + with tempfile.TemporaryDirectory() as scratch: + conversation_archive = Path(scratch) / "conversation.tar.gz" + await download(request.objects.trajectory, conversation_archive) + if request.objects.workspace is not None: + workspace_archive = Path(scratch) / "workspace.tar.gz" + await download(request.objects.workspace, workspace_archive) + async with self._workspace_lock: + self._require_idle() + workspace = workspace_archive if request.objects.workspace is not None else None + try: + await asyncio.to_thread(self._restore, conversation_archive, workspace, conversation_id(context_id)) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return {"context_id": context_id} + + def _restore(self, conversation_archive: Path, workspace_archive: Path | None, target: str) -> None: + """Extract both archives, check the conversation and point it at the target's paths, and only then put each in + place: one that fails to extract, or holds no conversation, changes nothing. The target gets the attached + files as they were saved, in its own directory, whatever other contexts have done to theirs.""" + self.conversations_dir.mkdir(parents=True, exist_ok=True) + self.inputs_dir.mkdir(parents=True, exist_ok=True) + conversation = _stage_conversation(conversation_archive, self.conversations_dir, target, self.inputs_dir / target) + staged = [conversation] + try: + if workspace_archive is not None: + self.workspace.mkdir(parents=True, exist_ok=True) + staged.append(_extract(workspace_archive, self.workspace)) + _replace_contents(self.workspace, staged[-1]) + shutil.rmtree(self.inputs_dir / target, ignore_errors=True) + if (conversation / INPUTS).is_dir(): + shutil.move(conversation / INPUTS, self.inputs_dir / target) + shutil.rmtree(self.conversations_dir / target, ignore_errors=True) + conversation.rename(self.conversations_dir / target) + finally: + for path in staged: + shutil.rmtree(path, ignore_errors=True) + + @extension(SKILL_CONFIG_V1.add.inline) + async def add_inline_skill(self, request: InlineSkillRequest) -> dict[str, Any]: + skill_dir = self.skills_dir / request.name + skill_dir.mkdir(parents=True, exist_ok=False) + (skill_dir / "SKILL.md").write_text(request.skill_md) + return {"name": request.name} + + @extension(SKILL_CONFIG_V1.add.bundle) + async def add_bundle_skill(self, request: BundleSkillRequest) -> dict[str, Any]: + self.skills_dir.mkdir(parents=True, exist_ok=True) + staging = self.skills_dir / f".{request.name}.{uuid.uuid4().hex}" + try: + # The protocol admits only normalized relative paths, so none leaves the staging directory. + for item in request.skill_bundle.files: + await download(item.object, staging / item.path) + staging.rename(self.skills_dir / request.name) + finally: + shutil.rmtree(staging, ignore_errors=True) + return {"name": request.name} diff --git a/src/agent_env/agents/agents/openhands/openhands_runtime.py b/src/agent_env/agents/agents/openhands/openhands_runtime.py new file mode 100644 index 00000000..9add55f5 --- /dev/null +++ b/src/agent_env/agents/agents/openhands/openhands_runtime.py @@ -0,0 +1,207 @@ +"""The OpenHands SDK side of the agent: a task is one ``arun()`` of the context's persisted conversation, with the +default tools, the task's MCP servers and its skills.""" + +import asyncio +import uuid +from dataclasses import dataclass, field +from typing import Any, Final, Literal + +from agentenv_protocol.a2a_agent import Usage +from openhands.sdk import LLM, Agent, AgentContext, Conversation, ImageContent, Message, TextContent +from openhands.sdk.conversation.base import BaseConversation +from openhands.sdk.conversation.exceptions import ConversationRunError +from openhands.sdk.conversation.response_utils import get_agent_final_response +from openhands.sdk.conversation.state import ConversationExecutionStatus +from openhands.sdk.event import ActionEvent, Event +from openhands.sdk.event.conversation_error import ConversationErrorEvent +from openhands.sdk.event.error_classification import FailureKind +from openhands.sdk.llm.utils.metrics import Metrics +from openhands.sdk.mcp import MCPServer +from openhands.sdk.skills.skill import load_skills_from_dir +from openhands.tools.preset.default import get_default_condenser, get_default_tools +from pydantic import PrivateAttr + +from conversation_run import ConversationRun +from openhands_agent import Outcome, Turn + +# The AgentEnv endpoint is a LiteLLM gateway; this provider prefix sends LiteLLM's requests to it as is. +_PROVIDER: Final[str] = "litellm_proxy" + + +class _GatewayLLM(LLM): + """The SDK's LLM with the gateway's headers sent on every call but kept out of the conversation's persisted state, + which a snapshot uploads: the SDK saves the agent's LLM there and masks only its key fields. The conversation's + stats don't count it, as they count only LLMs of exactly the SDK's class, so a turn's spend is read from its own.""" + + _gateway_headers: dict[str, str] = PrivateAttr(default_factory=dict) + + def _get_litellm_auth_values(self) -> tuple[str | None, dict[str, str]]: + api_key, headers = super()._get_litellm_auth_values() + return api_key, {**self._gateway_headers, **headers} + + +def _llm(turn: Turn) -> LLM: + options: dict[str, Any] = {} + if turn.config.effort is not None: + options["reasoning_effort"] = turn.config.effort + llm = _GatewayLLM( + usage_id="agent", + model=f"{_PROVIDER}/{turn.config.model}", + base_url=turn.endpoint.base_url, + api_key=turn.endpoint.api_key, + **options, + ) + llm._gateway_headers = dict(turn.endpoint.extra_headers) + return llm + + +def _agent(turn: Turn) -> Agent: + skills = [] + if turn.skill_names and turn.skills_dir.is_dir(): + loaded = load_skills_from_dir(turn.skills_dir, strict=False) + skills = [skill for group in loaded for skill in group.values() if skill.name in turn.skill_names] + mcp_config = { + # Literal, so a ${VAR} in a URL or header is sent as written, not filled from the agent's environment. + name: MCPServer(url=registration["url"], transport="http", headers=registration.get("headers") or None) + .as_literal() + for name, registration in turn.mcp_servers.items() + } + options: dict[str, Any] = {} + if turn.config.system_prompt: + options["system_prompt"] = turn.config.system_prompt + llm = _llm(turn) + return Agent( + llm=llm, + tools=get_default_tools(enable_browser=False), + mcp_config=mcp_config, + # The SDK's default agent condenses a conversation that outgrows its window; without one, a context's + # conversation, which every task adds to, fails every turn once it does. + condenser=get_default_condenser(llm=llm.model_copy(update={"usage_id": "condenser"})), + agent_context=AgentContext( + skills=skills, + system_message_suffix=turn.config.append_system_prompt, + load_user_skills=False, + load_public_skills=False, + ), + **options, + ) + + +def _usage(agent: Agent, tool_calls: int) -> Usage: + """What the turn's LLMs have spent: they are made for the turn, so they start at nothing.""" + llms = (agent.llm, getattr(agent.condenser, "llm", None)) + metrics = Metrics() + for each in {id(llm.metrics): llm.metrics for llm in llms if llm is not None}.values(): + metrics.merge(each) + + def spent(name: str) -> int: + return getattr(metrics.accumulated_token_usage, name, 0) or 0 + + input_tokens, output_tokens = spent("prompt_tokens"), spent("completion_tokens") + cost = metrics.accumulated_cost + return Usage( + tool_call_count=tool_calls, + input_tokens=input_tokens, + output_tokens=output_tokens, + total_tokens=input_tokens + output_tokens, + cost_usd=cost or None, + provider_details={ + "cache_read_tokens": spent("cache_read_tokens"), + "cache_write_tokens": spent("cache_write_tokens"), + "reasoning_tokens": spent("reasoning_tokens"), + }, + ) + + +def _unstarted(exc: Exception) -> Outcome: + return Outcome(f"OpenHands could not start the turn: {type(exc).__name__}: {exc}", Usage(), "openhands.start", + "infra_error") + + +def _failure( + error: ConversationErrorEvent | None, + fallback: str, + usage: Usage, + error_type: Literal["agent_error", "infra_error"], +) -> Outcome: + """The run's last error event as a failure, typed by its classification when it has one, else ``error_type``.""" + if error is None: + return Outcome(f"OpenHands failed: {fallback}", usage, "openhands.error", error_type) + if error.classification is not None: + error_type = "agent_error" if error.classification.kind == FailureKind.AGENT_ACTION else "infra_error" + return Outcome(f"OpenHands failed: {error.detail}", usage, f"openhands.{error.code}", error_type) + + +@dataclass(frozen=True, slots=True) +class SdkRuntime: + _runs: set[ConversationRun] = field(default_factory=set, init=False, repr=False) + + def stop(self) -> None: + for run in list(self._runs): + run.stop() + + async def run(self, turn: Turn) -> Outcome: + actions = 0 + # The SDK reports why a run failed only as events, whether or not it raises. + errors: list[ConversationErrorEvent] = [] + + def on_event(event: Event) -> None: + nonlocal actions + if isinstance(event, ActionEvent): + actions += 1 + elif isinstance(event, ConversationErrorEvent): + errors.append(event) + turn.on_event(event.model_dump(mode="json")) + + agent = _agent(turn) + + def open_conversation() -> BaseConversation: + return Conversation( + agent=agent, + workspace=str(turn.workspace), + persistence_dir=str(turn.conversations_dir), + conversation_id=uuid.UUID(turn.conversation_id), + callbacks=[on_event], + max_iteration_per_run=turn.config.max_iterations, + visualizer=None, + delete_on_close=False, + ) + + try: + conversation = await asyncio.to_thread(open_conversation) + except Exception as exc: # a persisted conversation the SDK can't resume, for one + return _unstarted(exc) + # Listed before it starts, so the server's shutdown stops it however far it has got. + run = ConversationRun(conversation) + self._runs.add(run) + try: + content = [TextContent(text=turn.text), *(ImageContent(image_urls=[image]) for image in turn.images)] + try: + await asyncio.to_thread(conversation.send_message, Message(role="user", content=content)) + except Exception as exc: # the agent's tools or MCP servers failing to start, for one + return _unstarted(exc) + try: + await run.to_its_end() + except ConversationRunError as exc: + usage = _usage(agent, actions) + error = exc.conversation_error or (errors[-1] if errors else None) + return _failure(error, str(exc.original_exception), usage, "infra_error") + except asyncio.CancelledError: + turn.on_usage(_usage(agent, actions)) + raise + usage = _usage(agent, actions) + status = conversation.state.execution_status + # Only the server's shutdown stops a run whose caller wasn't cancelled. + if run.stopped.is_set() and status != ConversationExecutionStatus.FINISHED: + return Outcome("OpenHands was stopped: the agent is shutting down", usage, "openhands.shutdown", + "infra_error") + if status == ConversationExecutionStatus.ERROR: + # A run limit, such as the iteration cap, ends the run without raising: the agent's own failure. + return _failure(errors[-1] if errors else None, "the run ended in error", usage, "agent_error") + if status != ConversationExecutionStatus.FINISHED: + return Outcome(f"OpenHands stopped with status '{status.value}'", usage, f"openhands.{status.value}") + reply = get_agent_final_response(conversation.state.events) + return Outcome(reply or "The agent finished without a reply.", usage) + finally: + self._runs.discard(run) + await asyncio.to_thread(conversation.close) diff --git a/src/agent_env/agents/agents/openhands/requirements.txt b/src/agent_env/agents/agents/openhands/requirements.txt new file mode 100644 index 00000000..63fd059e --- /dev/null +++ b/src/agent_env/agents/agents/openhands/requirements.txt @@ -0,0 +1,3 @@ +agentenv-framework-protocol[agent]>=0.1.351 +openhands-sdk==1.57.0 +openhands-tools==1.57.0 diff --git a/src/agent_env/agents/agents/openhands/start.sh b/src/agent_env/agents/agents/openhands/start.sh new file mode 100644 index 00000000..bd33facb --- /dev/null +++ b/src/agent_env/agents/agents/openhands/start.sh @@ -0,0 +1,8 @@ +#!/bin/sh +# Starts the installed agent (install/v1) with the credentials install wrote to agent.env; the caller sets A2A_PORT. +here=$(cd "$(dirname "$0")" && pwd) +set -a +. "$here/agent.env" +set +a +export PATH="/opt/openhands-a2a-venv/bin:$PATH" OPENHANDS_SUPPRESS_BANNER=1 +cd "$here" && exec python a2a_server.py >/tmp/openhands-a2a.log 2>&1 diff --git a/src/agent_env/agents/tasks/openhands-smoke.json b/src/agent_env/agents/tasks/openhands-smoke.json new file mode 100644 index 00000000..010ee0ee --- /dev/null +++ b/src/agent_env/agents/tasks/openhands-smoke.json @@ -0,0 +1,7 @@ +[ + {"id": "deploy", "type": "deploy_agent", "a2a_agent_id": "openhands", "env_ids": [], "sandbox_type": "local"}, + {"id": "ask", "type": "prompt_agent", "prompt_id": "ask", "depends_on": ["deploy"], + "prompt": "Run the shell command `echo $((6 * 7))` and reply with only the number it prints."}, + {"id": "check", "type": "agent_prompt_response_verifier", "prompt_id": "ask", "verifier_id": "answer", + "criteria": [{"type": "response_contains", "needles": ["42"]}], "depends_on": ["ask"]} +] diff --git a/tst/integration/cli/run_bundle_openhands_test.py b/tst/integration/cli/run_bundle_openhands_test.py new file mode 100644 index 00000000..2b4e27f7 --- /dev/null +++ b/tst/integration/cli/run_bundle_openhands_test.py @@ -0,0 +1,311 @@ +"""``agent-env run`` and the task steps on the agents bundle's OpenHands: its image built from the shipped folder with +real docker, the real OpenHands SDK, and a fake OpenAI Chat Completions endpoint as the configured model endpoint, so +no model key is needed. Needs a Docker daemon and network access for the image's downloads.""" + +import asyncio +import json +import logging +import subprocess +import time +import uuid +from typing import Final + +import httpx +import pytest +from click.testing import CliRunner + +from agent_env.a2a_agent import protocol +from agent_env.artifact.store import reset_artifact_store +from agent_env.cli import cli +from agent_env.config import configure, reset_config +from agent_env.providers.sandbox_providers.local_sandbox import _host_ips +from agent_env.store.object_store.local.grant_server import grant_server +from agent_env.store.routing import namespace_routing +from agent_env.task.teardown import teardown_run +from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep +from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep +from tst.util.fake_openai import USAGE, Fault, FakeOpenAI, Text, ToolCall, answer, tool_result_text, tool_then_answer + +pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] + +AGENT: Final[str] = "@local/agentenv-framework/agents/openhands" +MODEL: Final[str] = "fake-model-1" +ECHO_ENV: Final[str] = "@local/agentenv-framework/echo-mcp/echo" + + +@pytest.fixture +def endpoint(): + """The fake endpoint, where this machine's containers reach it: on Linux, the Docker bridge address.""" + with FakeOpenAI(host=_host_ips()[-1]) as fake: + yield fake + + +@pytest.fixture +def state(monkeypatch, tmp_path, endpoint): + """Local stores under this test's folder, and [model] pointed at the fake endpoint with a header of its own. + Whatever ran, no sandbox work folder may be left behind.""" + sandboxes = tmp_path / "sandboxes" + config = tmp_path / "config.toml" + config.write_text(f'[model]\nbase_url = "{endpoint.url}"\napi_key = "fake-key"\ndefault = "{MODEL}"\n\n' + '[model.headers]\nX-Agentenv-Test = "header-check"\n') + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("AGENT_ENV_CONFIG", str(config)) + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "state")) + monkeypatch.setenv("AGENT_ENV_DOCUMENT_STORE", "local") + monkeypatch.setenv("AGENT_ENV_OBJECT_STORE", "local") + monkeypatch.setenv("AGENT_ENV_LOCAL_SANDBOX_DIR", str(sandboxes)) + for name in ("LITELLM_API_KEY", "LITELLM_BASE_URL"): + monkeypatch.delenv(name, raising=False) + configure() + reset_artifact_store() + logging.disable(logging.CRITICAL) # pytest's live logging would take CliRunner's stdout + try: + yield tmp_path + assert not sandboxes.exists() or not any(sandboxes.iterdir()), "a run left a sandbox work folder" + finally: + logging.disable(logging.NOTSET) + grant_server(None, None).close() + reset_artifact_store() + reset_config() + + +def _bundle(root, name, steps): + (root / "tasks").mkdir(parents=True) + (root / f"tasks/{name}.json").write_text(json.dumps(steps)) + return root + + +def _ask(prompt, needle, env_ids=(), depends_on=()): + return [ + {"id": "agent", "type": "deploy_agent", "a2a_agent_id": AGENT, "env_ids": list(env_ids), + "sandbox_type": "local", **({"depends_on": list(depends_on)} if depends_on else {})}, + {"id": "ask", "type": "prompt_agent", "prompt_id": "ask", "prompt": prompt, "depends_on": ["agent"]}, + {"id": "check", "type": "agent_prompt_response_verifier", "prompt_id": "ask", "verifier_id": "answer", + "criteria": [{"type": "response_contains", "needles": [needle]}], "depends_on": ["ask"]}, + ] + + +def _run(root): + return CliRunner().invoke(cli, ["run", str(root)]) + + +def _with_agent(state, steps): + """Run ``steps`` against the shipped agent, after a bundle run has written it and its image to this test's + stores, and tear down what they deployed.""" + assert _run(_bundle(state / "setup", "ask", _ask("Say pong.", "pong"))).exit_code == 0 + context = TaskStepContext() + + async def run(): + try: + await steps(context) + finally: + await teardown_run(context) + + with namespace_routing(): + asyncio.run(run()) + + +async def _send(base_url: str, context_id: str, text: str) -> dict: + async with httpx.AsyncClient(timeout=300) as client: + response = await client.post(f"{base_url}/a2a", json={ + "jsonrpc": "2.0", "id": "1", "method": "message/send", "params": { + "message": {"kind": "message", "messageId": uuid.uuid4().hex, "role": "user", + "contextId": context_id, "parts": [{"kind": "text", "text": text}]}, + "configuration": {"blocking": True}}}) + task = response.json()["result"] + parts = task["status"]["message"]["parts"] + return {"state": task["status"]["state"], + "text": "".join(part.get("text", "") for part in parts if part["kind"] == "text"), + "data": next((part["data"] for part in parts if part["kind"] == "data"), {})} + + +def test_the_shipped_openhands_answers_through_the_configured_endpoint(state, endpoint): + endpoint.respond = answer("pong") + + result = _run(_bundle(state / "one", "ask", _ask("Say pong.", "pong"))) + + assert result.exit_code == 0, result.output + assert "tasks/ask.json v1: passed" in result.output, result.output + calls = endpoint.completions() + assert calls, "the agent never called the endpoint" + assert {call.headers["authorization"] for call in calls} == {"Bearer fake-key"} + assert {call.headers.get("x-agentenv-test") for call in calls} == {"header-check"} + assert {call.body.get("model") for call in calls} == {MODEL} + + +def test_the_shipped_openhands_runs_a_shell_command_with_its_terminal_tool(state, endpoint): + endpoint.respond = tool_then_answer("terminal", {"command": "echo $((6 * 7))", "security_risk": "LOW"}, + prefix="the shell said: ") + + result = _run(_bundle(state / "shell", "ask", _ask("What is six times seven? Use the shell.", "42"))) + + seen = (result.output, endpoint.tool_results()) + assert result.exit_code == 0 and "tasks/ask.json v1: passed" in result.output, seen + # The fake never computes, so 42 reached the answer through the container's shell. + assert [result.startswith("42") for result in endpoint.tool_results()] == [True], seen + + +def test_the_shipped_openhands_calls_the_tools_of_the_env_it_is_deployed_with(state, endpoint): + endpoint.respond = tool_then_answer("echo", {"text": "ping-from-tool"}, prefix="the tool said: ") + root = _bundle(state / "tools", "echo", [ + {"id": "env", "type": "deploy_env", "env_id": ECHO_ENV, "sandbox_type": "local"}, + *_ask("Call the echo tool with the text ping-from-tool.", "ping-from-tool", env_ids=[ECHO_ENV], + depends_on=["env"]), + ]) + + result = _run(root) + + offered = sorted({tool["function"]["name"] for call in endpoint.completions() for tool in call.body.get("tools") or []}) + seen = (result.output, offered, endpoint.tool_results()) + assert result.exit_code == 0 and "tasks/echo.json v1: passed" in result.output, seen + assert any(name.endswith("echo") for name in offered), seen + assert ["ping-from-tool" in result for result in endpoint.tool_results()] == [True], seen + + +def test_the_conversation_the_agent_keeps_condenses_and_holds_neither_the_key_nor_the_headers(state, endpoint): + # Spelled so the command, which the conversation records, holds neither value itself. + command = ("cd ~/.openhands-a2a/conversations && " + "(grep -rlF -e \"$(printf 'header-%s' check)\" -e \"$(printf 'fake-%s' key)\" . || echo no-secrets) && " + "grep -rl LLMSummarizingCondenser .") + endpoint.respond = tool_then_answer("terminal", {"command": command, "security_risk": "LOW"}, prefix="seen: ") + + result = _run(_bundle(state / "kept", "ask", _ask("Look at your saved conversation.", "seen"))) + + (seen,) = endpoint.tool_results() + assert result.exit_code == 0 and "tasks/ask.json v1: passed" in result.output, (result.output, seen) + assert "no-secrets" in seen and "base_state.json" in seen, seen + # Sent all the same. + assert {call.headers.get("x-agentenv-test") for call in endpoint.completions()} == {"header-check"} + + +def test_an_mcp_header_naming_a_variable_reaches_the_server_as_written(state, endpoint): + endpoint.respond = answer("pong") + + async def steps(context): + await DeployAgentTaskStep(id="agent", version=None, sandbox_type="local", a2a_agent_id=AGENT).execute(context) + (agent,) = context.deployed_agents + base_url = agent.a2a_url or agent.api_url + await protocol.post_agent_config(f"{base_url}/ext/agent-config", {"model": MODEL}) + async with httpx.AsyncClient() as client: + registered = await client.post(f"{base_url}/ext/mcp-config", json={ + "name": "probe", "url": f"{endpoint.url}/mcp-probe", "headers": {"X-Probe": "${LITELLM_API_KEY}"}}) + registered.raise_for_status() + await _send(base_url, "ctx", "Say pong.") + + _with_agent(state, steps) + + probes = [seen for seen in endpoint.seen if seen.path.startswith("/mcp-probe")] + assert probes, "the agent never called the MCP server" + assert {seen.headers.get("x-probe") for seen in probes} == {"${LITELLM_API_KEY}"} + + +def test_a_conversation_the_sdk_cannot_resume_fails_the_next_task_as_an_infra_error(state, endpoint): + outcomes = {} + endpoint.respond = answer("pong") + + async def steps(context): + await DeployAgentTaskStep(id="agent", version=None, sandbox_type="local", a2a_agent_id=AGENT).execute(context) + (agent,) = context.deployed_agents + base_url = agent.a2a_url or agent.api_url + await protocol.post_agent_config(f"{base_url}/ext/agent-config", {"model": MODEL}) + outcomes["first"] = await _send(base_url, "ctx", "Say pong.") + subprocess.run(["docker", "exec", f"agent-{agent.sandbox_id}", "sh", "-c", + "for f in ~/.openhands-a2a/conversations/*/base_state.json; do printf '{' > \"$f\"; done"], + check=True) + outcomes["second"] = await _send(base_url, "ctx", "Say pong again.") + + _with_agent(state, steps) + + assert outcomes["first"]["state"] == "completed", outcomes + second = outcomes["second"] + assert second["state"] == "failed", outcomes + assert (second["data"]["error_code"], second["data"]["error_type"]) == ("openhands.start", "infra_error"), outcomes + assert second["text"].startswith("OpenHands could not start the turn: "), outcomes + + +def test_a_model_endpoint_that_refuses_the_key_fails_the_task_as_an_infra_error(state, endpoint): + endpoint.respond = answer("pong") + + async def steps(context): + await DeployAgentTaskStep(id="agent", version=None, sandbox_type="local", a2a_agent_id=AGENT).execute(context) + endpoint.fail_next(Fault("status", 401), times=100) + with pytest.raises(Exception, match="error_type=infra_error, error_code=openhands.LLMAuthenticationError"): + await PromptAgentTaskStep(id="ask", version=None, prompt_id="ask", prompt="Say pong.").execute(context) + + _with_agent(state, steps) + + +def test_a_run_past_its_timeout_is_interrupted_and_its_context_keeps_working(state, endpoint): + outcomes = {} + + def respond(body): + # The first task's model call is the one that stalls past the timeout, and answers with a tool call that a + # run left going would carry out and report back on, calling the endpoint again. + if "Say pong again." in json.dumps(body["messages"][-1]): + return Text("pong") + return ToolCall("terminal", {"command": "echo late", "security_risk": "LOW"}) + + endpoint.respond = answer("pong") + + async def steps(context): + await DeployAgentTaskStep(id="agent", version=None, sandbox_type="local", a2a_agent_id=AGENT).execute(context) + (agent,) = context.deployed_agents + base_url = agent.a2a_url or agent.api_url + endpoint.respond = respond + await protocol.post_agent_config(f"{base_url}/ext/agent-config", {"model": MODEL, "timeout_seconds": 5}) + endpoint.fail_next(Fault("stall", seconds=15)) + started = time.monotonic() + outcomes["timed_out"] = await _send(base_url, "ctx", "Say pong.") + outcomes["seconds"] = time.monotonic() - started + calls = len(endpoint.completions()) + await asyncio.sleep(25) # past the stall, so a run left going would have called the endpoint again + outcomes["calls_after_the_timeout"] = len(endpoint.completions()) - calls + await protocol.post_agent_config(f"{base_url}/ext/agent-config", {"timeout_seconds": 120}) + outcomes["after"] = await _send(base_url, "ctx", "Say pong again.") + + _with_agent(state, steps) + + assert outcomes["timed_out"]["state"] == "failed", outcomes + assert outcomes["timed_out"]["data"]["error_code"] == "openhands.timeout", outcomes + assert outcomes["seconds"] < 15, outcomes + assert outcomes["calls_after_the_timeout"] == 0, outcomes + assert (outcomes["after"]["state"], outcomes["after"]["text"]) == ("completed", "pong"), outcomes + assert outcomes["after"]["data"]["usage"]["input_tokens"] == USAGE["prompt_tokens"], outcomes + # The interrupted turn stays in the conversation the next task resumes. + last = json.dumps(endpoint.completions()[-1].body["messages"]) + assert "Say pong." in last and "Say pong again." in last, last + + +def test_a_run_past_its_timeout_mid_command_leaves_no_process_behind(state, endpoint): + outcomes = {} + + def respond(body): + result = tool_result_text(body) + if result is not None: + return Text(f"processes: {result}") + if "List the processes." in json.dumps(body["messages"][-1]): + return ToolCall("terminal", {"command": "ps -eo args", "security_risk": "LOW"}) + return ToolCall("terminal", {"command": "sleep 120", "security_risk": "LOW"}) + + endpoint.respond = answer("pong") + + async def steps(context): + await DeployAgentTaskStep(id="agent", version=None, sandbox_type="local", a2a_agent_id=AGENT).execute(context) + (agent,) = context.deployed_agents + base_url = agent.a2a_url or agent.api_url + endpoint.respond = respond + await protocol.post_agent_config(f"{base_url}/ext/agent-config", {"model": MODEL, "timeout_seconds": 5}) + started = time.monotonic() + outcomes["timed_out"] = await _send(base_url, "ctx", "Sleep.") + outcomes["seconds"] = time.monotonic() - started + await protocol.post_agent_config(f"{base_url}/ext/agent-config", {"timeout_seconds": 120}) + outcomes["after"] = await _send(base_url, "ctx", "List the processes.") + + _with_agent(state, steps) + + assert outcomes["timed_out"]["data"]["error_code"] == "openhands.timeout", outcomes + # The terminal tool notices its killed session only at its no-change timeout, 30 s. + assert outcomes["seconds"] < 60, outcomes + assert outcomes["after"]["state"] == "completed", outcomes + assert "ps -eo args" in outcomes["after"]["text"] and "sleep 120" not in outcomes["after"]["text"], outcomes diff --git a/tst/unit/agents/openhands/__init__.py b/tst/unit/agents/openhands/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tst/unit/agents/openhands/agent.py b/tst/unit/agents/openhands/agent.py new file mode 100644 index 00000000..4066067a --- /dev/null +++ b/tst/unit/agents/openhands/agent.py @@ -0,0 +1,15 @@ +"""The OpenHands agent's A2A module and its run driver, imported from its folder as its image runs them: by their own +names, from the folder on the path. Its runtime module needs the OpenHands SDK, which only the agent's image installs, +so tests give the agent a fake runtime instead.""" + +import importlib +import sys +from pathlib import Path +from typing import Final + +import agent_env + +AGENT_DIR: Final[Path] = Path(agent_env.__file__).parent / "agents" / "agents" / "openhands" +sys.path.insert(0, str(AGENT_DIR)) +openhands_agent = importlib.import_module("openhands_agent") +conversation_run = importlib.import_module("conversation_run") diff --git a/tst/unit/agents/openhands/test_conversation_run.py b/tst/unit/agents/openhands/test_conversation_run.py new file mode 100644 index 00000000..909a115f --- /dev/null +++ b/tst/unit/agents/openhands/test_conversation_run.py @@ -0,0 +1,110 @@ +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from tst.unit.agents.openhands.agent import conversation_run + +ConversationRun = conversation_run.ConversationRun + + +class FakeConversation: + """Interrupts as the SDK's conversation does: it cancels the task ``arun()`` runs in once ``arun()`` has recorded + it, and before that only pauses, which ``arun()`` undoes as it starts. A run lasts until ``finish`` is set.""" + + def __init__(self) -> None: + self.finish = threading.Event() + self.started = threading.Event() + self.task: asyncio.Task | None = None + self.thread: str | None = None + self.paused = self.cancelled = False + self.closed = 0 + + async def arun(self) -> None: + self.task = asyncio.current_task() + self.paused = False + self.thread = threading.current_thread().name + self.started.set() + try: + while not self.finish.is_set(): + await asyncio.sleep(0.01) + except asyncio.CancelledError: + self.cancelled = True + + def interrupt(self) -> None: + if self.task is not None and not self.task.done(): + self.task.get_loop().call_soon_threadsafe(self.task.cancel) + else: + self.paused = True + + def close(self) -> None: + self.closed += 1 + + +def test_a_run_goes_to_its_end_on_a_thread_of_its_own() -> None: + conversation = FakeConversation() + conversation.finish.set() + + asyncio.run(ConversationRun(conversation).to_its_end()) + + assert conversation.started.is_set() and conversation.thread == "openhands-run" + + +def test_a_stop_before_the_run_starts_keeps_it_from_starting() -> None: + conversation = FakeConversation() + run = ConversationRun(conversation) + + run.stop() + asyncio.run(run.to_its_end()) + + assert not conversation.started.is_set() and conversation.closed == 1 + + +def test_a_stop_while_it_runs_interrupts_the_run() -> None: + conversation = FakeConversation() + run = ConversationRun(conversation) + + async def main() -> None: + running = asyncio.ensure_future(run.to_its_end()) + await asyncio.to_thread(conversation.started.wait) + run.stop() + await asyncio.wait_for(running, 5) + + asyncio.run(main()) + + assert conversation.cancelled and conversation.closed == 1 + + +def test_a_cancelled_caller_stops_the_run_and_waits_for_it_to_stop() -> None: + conversation = FakeConversation() + run = ConversationRun(conversation) + + async def main() -> None: + running = asyncio.ensure_future(run.to_its_end()) + await asyncio.to_thread(conversation.started.wait) + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + + asyncio.run(main()) + + assert conversation.cancelled and conversation.closed == 1 + + +def test_running_turns_leave_the_default_executor_free() -> None: + conversations = [FakeConversation() for _ in range(3)] + + async def main() -> str: + asyncio.get_running_loop().set_default_executor(ThreadPoolExecutor(max_workers=1)) + runs = [asyncio.ensure_future(ConversationRun(c).to_its_end()) for c in conversations] + for conversation in conversations: + await asyncio.to_thread(conversation.started.wait) + free = await asyncio.wait_for(asyncio.to_thread(lambda: "free"), 5) + for conversation in conversations: + conversation.finish.set() + await asyncio.gather(*runs) + return free + + assert asyncio.run(main()) == "free" + diff --git a/tst/unit/agents/openhands/test_openhands_agent.py b/tst/unit/agents/openhands/test_openhands_agent.py new file mode 100644 index 00000000..ac0505bc --- /dev/null +++ b/tst/unit/agents/openhands/test_openhands_agent.py @@ -0,0 +1,675 @@ +import asyncio +import base64 +import io +import json +import tarfile +import threading +import time +import uuid +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Final + +import httpx +import pytest +from agentenv_protocol.a2a_agent import Usage +from starlette.testclient import TestClient + +from tst.unit.agents.openhands.agent import openhands_agent + +OpenHandsAgent = openhands_agent.OpenHandsAgent +Endpoint = openhands_agent.Endpoint +Outcome = openhands_agent.Outcome + +# What a run cancelled midway had spent by then. +SPENT: Final[Usage] = Usage(tool_call_count=1, input_tokens=90, output_tokens=10, total_tokens=100, cost_usd=0.1) +GRANT: Final[dict[str, str]] = {"kind": "http-put", "url": "https://objects.test/put", "expires_at": "2099-01-01T00:00:00Z"} +READ: Final[dict[str, str]] = {"kind": "http-get", "url": "https://objects.test/get", "expires_at": "2099-01-01T00:00:00Z"} + + +@dataclass(slots=True) +class FakeRuntime: + """Persists each conversation as the SDK does, a directory named by its id holding ``base_state.json``, which it + resumes only under the id and directory the state records, and records the turns it ran. With ``gate`` set, a run + waits for it to open.""" + + turns: list[Any] = field(default_factory=list) + outcome: Outcome | None = None + hang: bool = False + cancelled: bool = False + gate: threading.Event | None = None + active: Counter = field(default_factory=Counter) + stopped: bool = False + + def stop(self) -> None: + self.stopped = True + + async def run(self, turn: Any) -> Outcome: + self.turns.append(turn) + directory = turn.conversations_dir / turn.conversation_id + state_path = directory / "base_state.json" + expected = {"id": str(uuid.UUID(turn.conversation_id)), "persistence_dir": str(directory)} + if state_path.exists(): + state = json.loads(state_path.read_text()) + if {key: state[key] for key in expected} != expected: + raise ValueError(f"Conversation ID mismatch: provided {expected}, persisted {state}") + else: + directory.mkdir(parents=True) + state = {**expected, "agent": {"kind": "Agent"}, "history": []} + state["history"].append(turn.text) + state_path.write_text(json.dumps(state)) + turn.on_event({"kind": "MessageEvent", "source": "user"}) + turn.on_event({"kind": "ActionEvent", "tool_name": "terminal"}) + self.active[turn.conversation_id] += 1 + try: + while self.gate is not None and not self.gate.is_set(): + await asyncio.sleep(0.01) + if self.hang: + # Bounded, so an agent that applies no timeout fails the test instead of hanging it. + await asyncio.sleep(10) + except asyncio.CancelledError: + self.cancelled = True + turn.on_usage(SPENT) + raise + finally: + self.active[turn.conversation_id] -= 1 + return self.outcome or Outcome("Done.", Usage(tool_call_count=1, input_tokens=200, output_tokens=40, + total_tokens=240, cost_usd=0.5)) + + def history(self, conversations_dir: Path, conversation_id: str) -> dict[str, Any]: + return json.loads((conversations_dir / conversation_id / "base_state.json").read_text()) + + +ENDPOINT: Final[Endpoint] = Endpoint("http://litellm.test", "sk-test", {"x-project": "p1"}) + + +@pytest.fixture +def runtime() -> FakeRuntime: + return FakeRuntime() + + +@pytest.fixture +def agent(tmp_path: Path, runtime: FakeRuntime) -> OpenHandsAgent: + return OpenHandsAgent(runtime, endpoint=ENDPOINT, home=tmp_path / "home", workspace=tmp_path / "ws") + + +@pytest.fixture +def client(agent: OpenHandsAgent): + with TestClient(agent.create_app(), client=("127.0.0.1", 50000)) as client: + client.post("/ext/agent-config", json={"model": "claude-sonnet", "effort": "high"}) + yield client + + +def _send(client: TestClient, parts: list[dict[str, Any]], context_id: str = "ctx-1") -> dict[str, Any]: + response = client.post("/a2a", json={ + "jsonrpc": "2.0", + "id": "1", + "method": "message/send", + "params": { + "message": {"kind": "message", "messageId": uuid.uuid4().hex, "role": "user", + "contextId": context_id, "parts": parts}, + "configuration": {"blocking": True}, + }, + }) + return response.json()["result"] + + +def _reply(task: dict[str, Any]) -> tuple[str, dict[str, Any]]: + parts = task["status"]["message"]["parts"] + text = "".join(part.get("text", "") for part in parts if part["kind"] == "text") + data = next((part["data"] for part in parts if part["kind"] == "data"), {}) + return text, data + + +def test_run_returns_the_reply_usage_and_live_trajectory(client: TestClient, runtime: FakeRuntime) -> None: + task = _send(client, [{"kind": "text", "text": "fix the bug"}, {"kind": "data", "data": {"k": 1}}]) + + text, data = _reply(task) + assert task["status"]["state"] == "completed" + assert text == "Done." + assert data["usage"] == {"tool_call_count": 1, "input_tokens": 200, "output_tokens": 40, "total_tokens": 240, + "cost_usd": 0.5} + turn = runtime.turns[0] + assert turn.text == 'fix the bug\n\n{\n "k": 1\n}' + assert turn.config.model == "claude-sonnet" + assert turn.config.effort == "high" + assert turn.endpoint == ENDPOINT + trajectory = client.post("/ext/trajectory", json={"task_id": task["id"]}).json()["trajectory"] + assert [event["kind"] for event in trajectory] == ["MessageEvent", "ActionEvent"] + + +def test_a_context_keeps_one_conversation(client: TestClient, runtime: FakeRuntime, tmp_path: Path) -> None: + _send(client, [{"kind": "text", "text": "one"}], context_id="ctx/a b") + _send(client, [{"kind": "text", "text": "two"}], context_id="ctx/a b") + _send(client, [{"kind": "text", "text": "three"}], context_id="ctx-2") + + first, second, other = (turn.conversation_id for turn in runtime.turns) + assert first == second == openhands_agent.conversation_id("ctx/a b") + assert other != first + assert runtime.history(tmp_path / "home" / "conversations", first)["history"] == ["one", "two"] + + +def test_mcp_servers_and_the_peer_tools_reach_the_runtime(client: TestClient, runtime: FakeRuntime) -> None: + client.post("/ext/mcp-config", json={"name": "files", "url": "http://mcp.test/mcp", + "headers": {"Authorization": "Bearer t"}}) + + _send(client, [{"kind": "text", "text": "go"}]) + + servers = runtime.turns[0].mcp_servers + assert servers["files"] == {"url": "http://mcp.test/mcp", "headers": {"Authorization": "Bearer t"}} + assert servers["peer_agents"]["url"].endswith("/ext/peer-agents/mcp") + + +def test_the_configured_role_is_sent_to_every_mcp_server(client: TestClient, runtime: FakeRuntime) -> None: + client.post("/ext/mcp-config", json={"name": "files", "url": "http://mcp.test/mcp", + "headers": {"Authorization": "Bearer t", "AgentEnv-Role": "stale"}}) + client.post("/ext/agent-config", json={"role": "buyer"}) + + _send(client, [{"kind": "text", "text": "go"}]) + + servers = runtime.turns[0].mcp_servers + assert servers["files"]["headers"] == {"Authorization": "Bearer t", "AgentEnv-Role": "buyer"} + assert servers["peer_agents"]["headers"] == {"AgentEnv-Role": "buyer"} + + +def test_images_are_attached_and_every_file_is_left_on_disk(client: TestClient, runtime: FakeRuntime) -> None: + png = base64.b64encode(b"\x89PNG").decode() + mp3 = base64.b64encode(b"ID3").decode() + _send(client, [{"kind": "text", "text": "look"}, + {"kind": "file", "file": {"name": "../red.png", "mimeType": "image/png", "bytes": png}}, + {"kind": "file", "file": {"name": "clip.mp3", "mimeType": "audio/mpeg", "bytes": mp3}}]) + + turn = runtime.turns[0] + assert turn.images == (f"data:image/png;base64,{png}",) + saved = [line.split(" saved at ")[1].rstrip("]") for line in turn.text.split("\n\n")[1:]] + assert [Path(path).name for path in saved] == ["1-red.png", "2-clip.mp3"] + assert Path(saved[1]).read_bytes() == b"ID3" + + +def test_inline_and_bundle_skills_are_installed_for_the_runtime( + client: TestClient, runtime: FakeRuntime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + async def fake_download(source, destination: Path) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(f"from {source.read.url}") + + monkeypatch.setattr(openhands_agent, "download", fake_download) + grant = {"media_type": "text/markdown", "max_bytes": 64, "size_bytes": 8, "read": READ} + client.post("/ext/skill-config", json={"name": "greet", "description": "Greet", "skill_md": "# Greet\nSay hi."}) + client.post("/ext/skill-config", json={ + "name": "lint", "description": "Lint", + "skill_bundle": {"max_total_bytes": 128, "files": [ + {"path": "SKILL.md", "object": grant}, {"path": "scripts/run.sh", "object": grant}]}, + }) + + _send(client, [{"kind": "text", "text": "go"}]) + + skills = tmp_path / "home" / "skills" + assert (skills / "greet" / "SKILL.md").read_text() == "# Greet\nSay hi." + assert (skills / "lint" / "scripts" / "run.sh").read_text() == f"from {READ['url']}" + assert sorted(path.name for path in skills.iterdir()) == ["greet", "lint"] + assert runtime.turns[0].skills_dir == skills + assert sorted(runtime.turns[0].skill_names) == ["greet", "lint"] + + +def test_a_runtime_failure_keeps_its_code_and_type(client: TestClient, runtime: FakeRuntime) -> None: + runtime.outcome = Outcome("OpenHands failed: rate limited", Usage(), "openhands.RateLimitError", "infra_error") + + task = _send(client, [{"kind": "text", "text": "go"}]) + + text, data = _reply(task) + assert task["status"]["state"] == "failed" + assert text == "OpenHands failed: rate limited" + assert (data["error_code"], data["error_type"]) == ("openhands.RateLimitError", "infra_error") + + +def test_timeout_cancels_the_runtime(client: TestClient, runtime: FakeRuntime) -> None: + runtime.hang = True + client.post("/ext/agent-config", json={"timeout_seconds": 1}) + + task = _send(client, [{"kind": "text", "text": "go"}]) + + assert task["status"]["state"] == "failed" + assert _reply(task)[1]["error_code"] == "openhands.timeout" + assert runtime.cancelled + + +def test_a_timed_out_task_reports_what_its_run_had_spent(client: TestClient, runtime: FakeRuntime) -> None: + runtime.hang = True + client.post("/ext/agent-config", json={"timeout_seconds": 1}) + + task = _send(client, [{"kind": "text", "text": "go"}]) + + _, data = _reply(task) + assert data["error_code"] == "openhands.timeout" + assert data["usage"] == {"tool_call_count": 1, "input_tokens": 90, "output_tokens": 10, "total_tokens": 100, + "cost_usd": 0.1} + + +def test_reading_a_file_part_counts_toward_the_task_timeout( + client: TestClient, runtime: FakeRuntime, monkeypatch: pytest.MonkeyPatch +) -> None: + real_client = httpx.AsyncClient + + async def stall(request: httpx.Request) -> httpx.Response: + await asyncio.sleep(30) + return httpx.Response(200, content=b"late") + + monkeypatch.setattr(openhands_agent.httpx, "AsyncClient", lambda **kwargs: real_client( + transport=httpx.MockTransport(stall), **kwargs)) + client.post("/ext/agent-config", json={"timeout_seconds": 1}) + started = time.monotonic() + + task = _send(client, [{"kind": "text", "text": "read it"}, + {"kind": "file", "file": {"name": "slow.txt", "uri": "https://files.test/slow.txt"}}]) + + assert _reply(task)[1]["error_code"] == "openhands.timeout" + assert time.monotonic() - started < 10 + assert runtime.turns == [] + + +def test_a_cancelled_task_interrupts_the_runtime(client: TestClient, runtime: FakeRuntime) -> None: + runtime.hang = True + sent = client.post("/a2a", json={"jsonrpc": "2.0", "id": "1", "method": "message/send", "params": { + "message": {"kind": "message", "messageId": uuid.uuid4().hex, "role": "user", "contextId": "ctx", + "parts": [{"kind": "text", "text": "go"}]}, + "configuration": {"blocking": False}}}).json()["result"] + _wait_until(lambda: sum(runtime.active.values()) == 1) + + client.post("/a2a", json={"jsonrpc": "2.0", "id": "2", "method": "tasks/cancel", "params": {"id": sent["id"]}}) + _wait_until(lambda: runtime.cancelled) + + task = client.post("/a2a", json={"jsonrpc": "2.0", "id": "3", "method": "tasks/get", + "params": {"id": sent["id"]}}).json()["result"] + assert task["status"]["state"] == "canceled" + + +def test_the_server_shutting_down_stops_the_runtime(agent: OpenHandsAgent, runtime: FakeRuntime) -> None: + with TestClient(agent.create_app()): + assert not runtime.stopped + + assert runtime.stopped + + +def test_errors_are_answered_as_json_details(client: TestClient, objects: dict[str, bytes]) -> None: + response = client.post("/ext/snapshot", json={"context_id": "nope", "objects": {"trajectory": _write()}}) + + assert response.status_code == 404 + assert response.json() == {"detail": "No conversation for context 'nope'"} + + +def test_missing_model_fails_without_running(agent: OpenHandsAgent, runtime: FakeRuntime) -> None: + with TestClient(agent.create_app()) as client: + task = _send(client, [{"kind": "text", "text": "go"}]) + + assert _reply(task)[1]["error_code"] == "openhands.no_model" + assert runtime.turns == [] + + +def test_the_default_model_is_the_fallback(tmp_path: Path, runtime: FakeRuntime) -> None: + agent = OpenHandsAgent(runtime, endpoint=ENDPOINT, default_model="fallback-model", home=tmp_path / "home", + workspace=tmp_path / "ws") + with TestClient(agent.create_app()) as client: + task = _send(client, [{"kind": "text", "text": "go"}]) + + assert task["status"]["state"] == "completed" + assert runtime.turns[0].config.model == "fallback-model" + + +def test_missing_endpoint_is_an_infra_failure(tmp_path: Path, runtime: FakeRuntime) -> None: + agent = OpenHandsAgent(runtime, endpoint=None, default_model="m", home=tmp_path / "home", workspace=tmp_path / "ws") + with TestClient(agent.create_app()) as client: + task = _send(client, [{"kind": "text", "text": "go"}]) + + _, data = _reply(task) + assert (data["error_code"], data["error_type"]) == ("openhands.no_endpoint", "infra_error") + assert runtime.turns == [] + + +def test_the_endpoint_is_read_from_the_litellm_variables() -> None: + environ = {"LITELLM_BASE_URL": "http://litellm.test", "LITELLM_API_KEY": "sk-test", + "LITELLM_EXTRA_HEADERS": '{"x-project": "p1"}'} + without_headers = {"LITELLM_BASE_URL": "http://litellm.test", "LITELLM_EXTRA_HEADERS": ""} + + assert Endpoint.from_environ(environ) == ENDPOINT + assert Endpoint.from_environ(without_headers) == Endpoint("http://litellm.test") + assert Endpoint.from_environ({"LITELLM_API_KEY": "sk-test"}) is None + + +@pytest.mark.parametrize("headers", ["{not json", '["x-project"]']) +def test_malformed_extra_headers_are_refused_at_startup(headers: str) -> None: + with pytest.raises(ValueError, match="LITELLM_EXTRA_HEADERS"): + Endpoint.from_environ({"LITELLM_BASE_URL": "http://litellm.test", "LITELLM_EXTRA_HEADERS": headers}) + + +def _write() -> dict[str, Any]: + return {"media_type": "application/octet-stream", "max_bytes": 1 << 20, "write": GRANT} + + +def _read(name: str) -> dict[str, Any]: + return {"media_type": "application/octet-stream", "max_bytes": 1 << 20, + "read": {**READ, "url": f"https://objects.test/{name}"}} + + +@pytest.fixture +def objects(monkeypatch: pytest.MonkeyPatch) -> dict[str, bytes]: + """An in-memory object store behind the transfer helpers the agent calls.""" + stored: dict[str, bytes] = {} + names = iter(["trajectory", "workspace"]) + + class FakeUploaded(dict): + def model_dump(self, mode: str = "json") -> dict: + return dict(self) + + async def fake_upload(target, source: Path): + body = source.read_bytes() + stored[next(names)] = body + return FakeUploaded(size_bytes=len(body)) + + async def fake_download(source, destination: Path) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(stored[source.read.url.rsplit("/", 1)[-1]]) + + monkeypatch.setattr(openhands_agent, "upload", fake_upload) + monkeypatch.setattr(openhands_agent, "download", fake_download) + return stored + + +def test_snapshot_round_trip_rebinds_the_conversation_to_the_target_context( + client: TestClient, runtime: FakeRuntime, tmp_path: Path, objects: dict[str, bytes] +) -> None: + _send(client, [{"kind": "text", "text": "remember TOKEN-1"}], context_id="source") + (tmp_path / "ws" / "notes.txt").write_text("remember TOKEN-1") + saved = client.post("/ext/snapshot", json={"context_id": "source", "objects": { + "trajectory": _write(), "workspace": _write()}}) + assert saved.status_code == 200, saved.text + (tmp_path / "ws" / "notes.txt").unlink() + (tmp_path / "ws" / "after.txt").write_text("written after the snapshot") + + loaded = client.put("/ext/snapshot", json={"objects": { + "trajectory": _read("trajectory"), "workspace": _read("workspace")}, "target_context_id": "target"}) + target = openhands_agent.conversation_id("target") + conversations = tmp_path / "home" / "conversations" + restored = runtime.history(conversations, target) + task = _send(client, [{"kind": "text", "text": "what was the token?"}], context_id="target") + + assert loaded.json() == {"context_id": "target"} + assert (tmp_path / "ws" / "notes.txt").read_text() == "remember TOKEN-1" + assert not (tmp_path / "ws" / "after.txt").exists() + assert (restored["id"], restored["persistence_dir"]) == (str(uuid.UUID(target)), str(conversations / target)) + assert task["status"]["state"] == "completed" + assert runtime.history(conversations, target)["history"] == ["remember TOKEN-1", "what was the token?"] + + +def test_an_empty_workspace_restores_as_empty( + client: TestClient, tmp_path: Path, objects: dict[str, bytes] +) -> None: + _send(client, [{"kind": "text", "text": "remember TOKEN-1"}], context_id="source") + client.post("/ext/snapshot", json={"context_id": "source", "objects": { + "trajectory": _write(), "workspace": _write()}}) + (tmp_path / "ws" / "after.txt").write_text("written after the snapshot") + + loaded = client.put("/ext/snapshot", json={"objects": { + "trajectory": _read("trajectory"), "workspace": _read("workspace")}, "target_context_id": "target"}) + + assert loaded.status_code == 200, loaded.text + assert list((tmp_path / "ws").iterdir()) == [] + + +def _attached(text: str) -> Path: + return Path(text.split(" saved at ")[1].rstrip("]")) + + +def test_a_snapshot_restores_the_files_attached_in_its_context_as_they_were_saved( + client: TestClient, runtime: FakeRuntime, tmp_path: Path, objects: dict[str, bytes] +) -> None: + notes = base64.b64encode(b"TOKEN-7").decode() + _send(client, [{"kind": "text", "text": "keep this"}, + {"kind": "file", "file": {"name": "notes.txt", "mimeType": "text/plain", "bytes": notes}}], + context_id="source") + attached = _attached(runtime.turns[0].text) + saved = client.post("/ext/snapshot", json={"context_id": "source", "objects": {"trajectory": _write()}}) + attached.write_bytes(b"changed after the snapshot") + + loaded = client.put("/ext/snapshot", json={"objects": {"trajectory": _read("trajectory")}, + "target_context_id": "target"}) + + inputs, conversations = tmp_path / "home" / "inputs", tmp_path / "home" / "conversations" + source, target = (openhands_agent.conversation_id(context) for context in ("source", "target")) + (message,) = runtime.history(conversations, target)["history"] + assert (saved.status_code, loaded.status_code) == (200, 200) + assert _attached(message) == inputs / target / attached.relative_to(inputs / source) + assert _attached(message).read_bytes() == b"TOKEN-7" + assert attached.read_bytes() == b"changed after the snapshot" + assert sorted(path.name for path in (conversations / target).iterdir()) == ["base_state.json"] + + +def test_a_snapshot_loaded_on_an_agent_with_another_home_names_its_files_there( + client: TestClient, runtime: FakeRuntime, tmp_path: Path, objects: dict[str, bytes] +) -> None: + notes = base64.b64encode(b"TOKEN-8").decode() + _send(client, [{"kind": "text", "text": "keep this"}, + {"kind": "file", "file": {"name": "notes.txt", "mimeType": "text/plain", "bytes": notes}}], + context_id="source") + client.post("/ext/snapshot", json={"context_id": "source", "objects": {"trajectory": _write()}}) + other = OpenHandsAgent(runtime, endpoint=ENDPOINT, home=tmp_path / "other", workspace=tmp_path / "other-ws") + + with TestClient(other.create_app()) as other_client: + loaded = other_client.put("/ext/snapshot", json={"objects": {"trajectory": _read("trajectory")}, + "target_context_id": "target"}) + + target = openhands_agent.conversation_id("target") + (message,) = runtime.history(tmp_path / "other" / "conversations", target)["history"] + assert loaded.status_code == 200 + assert _attached(message).is_relative_to(tmp_path / "other" / "inputs" / target) + assert _attached(message).read_bytes() == b"TOKEN-8" + + +@pytest.mark.parametrize("member, content", [ + ("README", b""), + ("base_state.json", b"{}"), + ("base_state.json", b'{"id": "not-a-uuid", "agent": {}}'), + ("base_state.json", json.dumps({"id": str(uuid.uuid4())}).encode()), +], ids=["no-state", "empty-state", "bad-id", "no-agent"]) +def test_a_snapshot_that_holds_no_conversation_changes_nothing( + client: TestClient, runtime: FakeRuntime, tmp_path: Path, objects: dict[str, bytes], member: str, content: bytes +) -> None: + _send(client, [{"kind": "text", "text": "remember TOKEN-1"}], context_id="source") + client.post("/ext/snapshot", json={"context_id": "source", "objects": {"trajectory": _write(), "workspace": _write()}}) + (tmp_path / "ws" / "notes.txt").write_text("current") + not_a_conversation = io.BytesIO() + with tarfile.open(fileobj=not_a_conversation, mode="w:gz") as tar: + info = tarfile.TarInfo(member) + info.size = len(content) + tar.addfile(info, io.BytesIO(content)) + objects["trajectory"] = not_a_conversation.getvalue() + conversations = tmp_path / "home" / "conversations" + source = openhands_agent.conversation_id("source") + + loaded = client.put("/ext/snapshot", json={"objects": { + "trajectory": _read("trajectory"), "workspace": _read("workspace")}, "target_context_id": "source"}) + + assert loaded.status_code == 400 + assert "holds no OpenHands conversation" in loaded.json()["detail"] + assert [path.name for path in (tmp_path / "ws").iterdir()] == ["notes.txt"] + assert (tmp_path / "ws" / "notes.txt").read_text() == "current" + assert [path.name for path in conversations.iterdir()] == [source] + assert runtime.history(conversations, source)["history"] == ["remember TOKEN-1"] + + +def test_snapshot_of_an_unknown_context_is_404(client: TestClient, objects: dict[str, bytes]) -> None: + response = client.post("/ext/snapshot", json={"context_id": "nope", "objects": {"trajectory": _write()}}) + + assert response.status_code == 404 + + +def _send_in_background(client: TestClient, text: str, context_id: str) -> threading.Thread: + thread = threading.Thread(target=_send, args=(client, [{"kind": "text", "text": text}], context_id)) + thread.start() + return thread + + +def _wait_until(condition, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while not condition(): + assert time.monotonic() < deadline, "condition not reached" + time.sleep(0.01) + + +def test_a_snapshot_load_is_refused_while_a_task_runs( + client: TestClient, runtime: FakeRuntime, objects: dict[str, bytes] +) -> None: + _send(client, [{"kind": "text", "text": "remember TOKEN-1"}], context_id="source") + client.post("/ext/snapshot", json={"context_id": "source", "objects": {"trajectory": _write()}}) + runtime.gate = threading.Event() + running = _send_in_background(client, "keep working", "busy") + _wait_until(lambda: sum(runtime.active.values()) == 1) + + busy = client.put("/ext/snapshot", json={"objects": {"trajectory": _read("trajectory")}, "target_context_id": "t"}) + runtime.gate.set() + running.join() + idle = client.put("/ext/snapshot", json={"objects": {"trajectory": _read("trajectory")}, "target_context_id": "t"}) + + assert (busy.status_code, idle.status_code) == (409, 200) + + +def test_a_snapshot_save_waits_for_its_context_running_task( + client: TestClient, runtime: FakeRuntime, objects: dict[str, bytes] +) -> None: + _send(client, [{"kind": "text", "text": "plant"}], context_id="ctx-a") + runtime.gate = threading.Event() + running = _send_in_background(client, "more", "ctx-a") + _wait_until(lambda: sum(runtime.active.values()) == 1) + saved: dict[str, httpx.Response] = {} + saver = threading.Thread(target=lambda: saved.update(response=client.post( + "/ext/snapshot", json={"context_id": "ctx-a", "objects": {"trajectory": _write()}}))) + saver.start() + time.sleep(0.3) + finished_while_running = "response" in saved + runtime.gate.set() + running.join() + saver.join() + + with tarfile.open(fileobj=io.BytesIO(objects["trajectory"])) as tar: + state = json.load(tar.extractfile("base_state.json")) + assert not finished_while_running + assert saved["response"].status_code == 200 + assert state["history"] == ["plant", "more"] + + +def test_a_workspace_snapshot_waits_for_the_tasks_running_in_other_contexts( + client: TestClient, runtime: FakeRuntime, objects: dict[str, bytes] +) -> None: + _send(client, [{"kind": "text", "text": "plant"}], context_id="ctx-a") + runtime.gate = threading.Event() + running = _send_in_background(client, "write files", "ctx-b") + _wait_until(lambda: sum(runtime.active.values()) == 1) + saved: dict[str, httpx.Response] = {} + saver = threading.Thread(target=lambda: saved.update(response=client.post( + "/ext/snapshot", json={"context_id": "ctx-a", "objects": {"trajectory": _write(), "workspace": _write()}}))) + saver.start() + time.sleep(0.3) + finished_while_running = "response" in saved + runtime.gate.set() + running.join() + saver.join() + + assert not finished_while_running + assert saved["response"].status_code == 200 + + +def test_conversation_locks_are_dropped_once_released( + agent: OpenHandsAgent, client: TestClient, objects: dict[str, bytes] +) -> None: + for context in ("one", "two", "three"): + _send(client, [{"kind": "text", "text": "go"}], context_id=context) + client.post("/ext/snapshot", json={"context_id": "one", "objects": {"trajectory": _write()}}) + + assert agent._conversation_locks == {} + + +def test_a_bundle_skill_cannot_write_outside_its_directory( + client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + async def fake_download(source, destination: Path) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text("payload") + + monkeypatch.setattr(openhands_agent, "download", fake_download) + grant = {"media_type": "text/plain", "max_bytes": 64, "size_bytes": 8, "read": READ} + responses = [client.post("/ext/skill-config", json={ + "name": "evil", "description": "Evil", + "skill_bundle": {"max_total_bytes": 128, "files": [ + {"path": "SKILL.md", "object": grant}, {"path": path, "object": grant}]}, + }) for path in ("../../escaped.sh", "/tmp/escaped.sh", "scripts/../../escaped.sh")] + + assert [response.status_code for response in responses] == [400, 400, 400] + assert not (tmp_path / "home" / "escaped.sh").exists() + assert not (tmp_path / "home" / "skills").exists() + + +def test_an_oversized_attachment_is_refused( + client: TestClient, runtime: FakeRuntime, monkeypatch: pytest.MonkeyPatch +) -> None: + real_client = httpx.AsyncClient + monkeypatch.setattr(openhands_agent, "MAX_ATTACHMENT_BYTES", 8) + monkeypatch.setattr(openhands_agent.httpx, "AsyncClient", lambda **kwargs: real_client( + transport=httpx.MockTransport(lambda request: httpx.Response(200, content=b"x" * 64)), **kwargs)) + + task = _send(client, [{"kind": "text", "text": "read it"}, + {"kind": "file", "file": {"name": "big.txt", "uri": "https://files.test/big.txt"}}]) + + assert task["status"]["state"] == "failed" + assert _reply(task)[1]["error_code"] == "openhands.attachment" + assert runtime.turns == [] + + +def test_only_http_file_uris_are_fetched(client: TestClient, runtime: FakeRuntime) -> None: + task = _send(client, [{"kind": "text", "text": "read it"}, + {"kind": "file", "file": {"name": "passwd", "uri": "file:///etc/passwd"}}]) + + text, data = _reply(task) + assert task["status"]["state"] == "failed" + assert data["error_code"] == "openhands.attachment" + assert "unsupported file URI scheme" in text + assert runtime.turns == [] + + +def test_a_malformed_file_uri_fails_as_an_attachment(client: TestClient, runtime: FakeRuntime) -> None: + task = _send(client, [{"kind": "text", "text": "read it"}, + {"kind": "file", "file": {"name": "x.txt", "uri": "https://[::1/x.txt"}}]) + + assert _reply(task)[1]["error_code"] == "openhands.attachment" + assert runtime.turns == [] + + +def test_install_extension_is_declared_on_the_card(client: TestClient) -> None: + card = client.get("/.well-known/agent.json").json() + install = next(e for e in card["capabilities"]["extensions"] if e["uri"] == "urn:agentenv:install/v1")["params"] + + assert install["a2a_port"] == 8000 + assert set(install["required_params"]) == {"container", "agent_ctx_tar", "a2a_port", "litellm_api_key", + "litellm_base_url", "litellm_extra_headers"} + formatted = [command.format(container="c", agent_ctx_tar="/t.tar", a2a_port="8000", litellm_api_key="'sk-secret'", + litellm_base_url="u", litellm_extra_headers="'{}'") + for command in install["install_commands"]] + assert formatted[-1].endswith("c sh /opt/openhands-a2a/start.sh") + assert all("sk-secret" not in command.splitlines()[0] for command in formatted) + assert "LITELLM_API_KEY='sk-secret'" in formatted[-2].splitlines() + + +def test_the_card_advertises_the_extensions_agent_env_validates(client: TestClient) -> None: + card = client.get("/.well-known/agent.json").json() + uris = {extension["uri"] for extension in card["capabilities"]["extensions"]} + + assert {f"urn:agentenv:{name}/v1" for name in ( + "agent-config", "mcp-config", "skill-config", "trajectory", "snapshot", "peer-agents", "install", "triggers", + )} <= uris + + +def test_the_card_version_is_the_pinned_sdk() -> None: + requirements = (openhands_agent.Path(openhands_agent.__file__).parent / "requirements.txt").read_text().splitlines() + + assert f"openhands-sdk=={openhands_agent.OPENHANDS_VERSION}" in requirements + assert f"openhands-tools=={openhands_agent.OPENHANDS_VERSION}" in requirements diff --git a/tst/util/fake_openai.py b/tst/util/fake_openai.py new file mode 100644 index 00000000..142d41a1 --- /dev/null +++ b/tst/util/fake_openai.py @@ -0,0 +1,153 @@ +"""A fake OpenAI Chat Completions endpoint, so an agent that calls the model through LiteLLM runs with no model key: +``POST .../chat/completions`` is answered by a responder, as one JSON completion, and every request is recorded. A +fault queued with ``fail_next`` replaces the next answer: an HTTP error, or an answer that stalls.""" + +import json +import threading +import time +import uuid +from collections import deque +from collections.abc import Callable +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Final + + +@dataclass(frozen=True, slots=True) +class Text: + text: str + + +@dataclass(frozen=True, slots=True) +class ToolCall: + name: str + arguments: dict + + +@dataclass(frozen=True, slots=True) +class Fault: + """``status``: answer with that HTTP error; ``stall``: send nothing for ``seconds``, then the answer.""" + + kind: str + status: int = 500 + seconds: float = 0.0 + + +@dataclass(slots=True) +class Seen: + path: str + headers: dict[str, str] + body: dict + + +Responder = Callable[[dict], Text | ToolCall] + +USAGE: Final[dict[str, int]] = {"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110} + + +def answer(text: str) -> Responder: + """Every request answered with ``text``.""" + return lambda body: Text(text) + + +def tool_then_answer(tool_suffix: str, arguments: dict, *, prefix: str = "") -> Responder: + """Call the request's tool whose name ends in ``tool_suffix``, then answer with ``prefix`` and the tool result's + text; answer ``prefix`` alone to a request that offers no such tool.""" + + def respond(body: dict) -> Text | ToolCall: + result = tool_result_text(body) + if result is not None: + return Text(prefix + result) + names = [tool["function"]["name"] for tool in body.get("tools") or []] + name = next((name for name in names if name.endswith(tool_suffix)), None) + return ToolCall(name, arguments) if name else Text(prefix) + + return respond + + +def tool_result_text(body: dict) -> str | None: + """The text of the last message when it is a tool result, else None.""" + messages = body.get("messages") or [] + if not messages or messages[-1].get("role") != "tool": + return None + content = messages[-1].get("content") + if isinstance(content, list): + return "".join(part.get("text", "") for part in content if isinstance(part, dict)) + return content or "" + + +@dataclass(slots=True) +class FakeOpenAI: + host: str = "127.0.0.1" + respond: Responder = field(default_factory=lambda: answer("ok")) + seen: list[Seen] = field(default_factory=list) + _faults: deque = field(default_factory=deque) + _server: ThreadingHTTPServer | None = None + + @property + def url(self) -> str: + return f"http://{self.host}:{self._server.server_address[1]}" + + def fail_next(self, fault: Fault, times: int = 1) -> None: + self._faults.extend([fault] * times) + + def completions(self) -> list[Seen]: + return [seen for seen in self.seen if seen.path.endswith("/chat/completions")] + + def tool_results(self) -> list[str]: + return [text for seen in self.completions() if (text := tool_result_text(seen.body)) is not None] + + def __enter__(self) -> "FakeOpenAI": + fake = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args) -> None: + pass + + def do_POST(self) -> None: + length = int(self.headers.get("content-length") or 0) + body = json.loads(self.rfile.read(length) or b"{}") + fake.seen.append(Seen(self.path, {k.lower(): v for k, v in self.headers.items()}, body)) + if not self.path.endswith("/chat/completions"): + self._send(404, {"error": {"message": f"no route {self.path}"}}) + return + fault = fake._faults.popleft() if fake._faults else None + if fault is not None and fault.kind == "status": + self._send(fault.status, {"error": {"message": f"fake {fault.status}", "type": "fake_error"}}) + return + if fault is not None and fault.kind == "stall": + time.sleep(fault.seconds) + self._send(200, _completion(body, fake.respond(body))) + + def _send(self, status: int, payload: dict) -> None: + data = json.dumps(payload).encode() + self.send_response(status) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(data))) + self.end_headers() + try: + self.wfile.write(data) + except (BrokenPipeError, ConnectionResetError): + pass + + self._server = ThreadingHTTPServer((self.host, 0), Handler) + self._server.daemon_threads = True + threading.Thread(target=self._server.serve_forever, daemon=True).start() + return self + + def __exit__(self, *exc) -> None: + self._server.shutdown() + self._server.server_close() + + +def _completion(body: dict, reply: Text | ToolCall) -> dict: + if isinstance(reply, Text): + message, finish = {"role": "assistant", "content": reply.text}, "stop" + else: + message = {"role": "assistant", "content": None, "tool_calls": [{ + "id": f"call_{uuid.uuid4().hex[:12]}", "type": "function", + "function": {"name": reply.name, "arguments": json.dumps(reply.arguments)}}]} + finish = "tool_calls" + return {"id": f"chatcmpl-{uuid.uuid4().hex[:12]}", "object": "chat.completion", "created": int(time.time()), + "model": body.get("model", ""), "choices": [{"index": 0, "message": message, "finish_reason": finish}], + "usage": USAGE}