diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 6682df343..d3aac7c20 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -43,6 +43,7 @@ updates: - "/src/agent_env/agents/agents/claude-code" - "/src/agent_env/agents/agents/codex" - "/src/agent_env/agents/agents/gemini" + - "/src/agent_env/agents/agents/hermes" - "/src/agent_env/agents/agents/letta-code" - "/src/agent_env/agents/agents/opencode" - "/src/agent_env/agents/agents/openhands" diff --git a/.github/scripts/infra_build.py b/.github/scripts/infra_build.py index fd9ed5a42..92bc20aff 100644 --- a/.github/scripts/infra_build.py +++ b/.github/scripts/infra_build.py @@ -34,6 +34,7 @@ "agentenv-claude-code": "src/agent_env/agents/agents/claude-code/Dockerfile", "agentenv-codex": "src/agent_env/agents/agents/codex/Dockerfile", "agentenv-gemini": "src/agent_env/agents/agents/gemini/Dockerfile", + "agentenv-hermes": "src/agent_env/agents/agents/hermes/Dockerfile", "agentenv-letta-code": "src/agent_env/agents/agents/letta-code/Dockerfile", "agentenv-opencode": "src/agent_env/agents/agents/opencode/Dockerfile", "agentenv-openhands": "src/agent_env/agents/agents/openhands/Dockerfile", @@ -59,7 +60,7 @@ _RETRY_WAITS_SECONDS = (30, 60, 120) # Each image's check, beyond its build: the gateway imports its server, Postgres turns healthy, the sidecars and the -# browser answer their own commands, Claude Code's, Codex's, Gemini's, OpenCode's and pi's agents run their CLIs and +# browser answer their own commands, Claude Code's, Codex's, Gemini's, Hermes', OpenCode's and pi's agents run their CLIs and # import their servers, OpenHands' imports its server, which loads the OpenHands SDK, Pydantic AI's imports its # server, which loads Pydantic AI, the proxy imports its server, the mail example's env imports its server, and the # base computer's desktop comes up, every program running. @@ -76,6 +77,8 @@ 'codex --version && mcp-remote --help >/dev/null && python -c "import a2a_server"'], "agentenv-gemini": ["docker", "run", "--rm", "--entrypoint", "sh", "{image}", "-c", 'gemini --version && python -c "import a2a_server"'], + "agentenv-hermes": ["docker", "run", "--rm", "--entrypoint", "sh", "{image}", "-c", + 'hermes --version && python -c "import a2a_server"'], "agentenv-letta-code": ["docker", "run", "--rm", "--entrypoint", "sh", "{image}", "-c", 'letta --version && python -c "import a2a_server"'], "agentenv-opencode": ["docker", "run", "--rm", "--entrypoint", "sh", "{image}", "-c", diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b97278465..2c6e81d47 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -91,7 +91,7 @@ the previous release, so a clear pull request title is also the release note. A A release also publishes the infra images agent-env ships, once the `PUBLISH_INFRA_IMAGES` repository variable is `true`: one per Dockerfile under `src/agent_env/env`, and one per shipped agent (`agentenv-claude-code`, its CLI -included, `agentenv-codex`, `agentenv-gemini`, `agentenv-opencode`, `agentenv-pi`, `agentenv-openhands` and `agentenv-pydantic-ai`), at `ghcr.io/scaleapi/agentenv-:src-` for linux/amd64 +included, `agentenv-codex`, `agentenv-gemini`, `agentenv-hermes`, `agentenv-opencode`, `agentenv-pi`, `agentenv-openhands` and `agentenv-pydantic-ai`), at `ghcr.io/scaleapi/agentenv-:src-` for linux/amd64 and linux/arm64, where the tag is the build context's source digest, so an unchanged Dockerfile is not built again. The agents' images install the newest `agentenv-framework-protocol` at or above the floor in their `requirements.txt`, which a release doesn't pin, so each is built again only when its folder changes: once the protocol diff --git a/src/agent_env/agents/README.md b/src/agent_env/agents/README.md index cfadf2c13..3ec6562c2 100644 --- a/src/agent_env/agents/README.md +++ b/src/agent_env/agents/README.md @@ -89,9 +89,17 @@ its smoke task needs `[model] default` (or the `agent` role) set; a `vllm:` pref requests to the model's family as a vLLM server needs. It declares `install/v1` too, for Debian or Ubuntu targets. Its folder's `README.md` covers the extensions it implements. -A release publishes Codex's, Gemini's, OpenCode's, pi's, OpenHands' and Pydantic AI's images too, at +`agents/hermes/` runs Nous Research's [Hermes Agent](https://github.com/NousResearch/hermes-agent) behind the +protocol's A2A agent framework: each turn is one `hermes agentenv turn`, a command the agent's own Hermes plugin adds, +which runs the turn in its process, resumed in the context's Hermes session; each context is a Hermes profile of its +own, the task env's MCP servers go to its `config.yaml`, and Hermes' plugin hook events are the turn's trajectory. Name +it as `hermes`, its alias, or by its id, `@local/agentenv-framework/agents/hermes`. It calls the `[model]` endpoint as +an OpenAI-compatible one, `/v1/chat/completions`, and takes its model from the prompt, else +`anthropic/claude-sonnet-4-6`. Its folder's `README.md` covers the extensions it implements. + +A release publishes Codex's, Gemini's, Hermes', OpenCode's, pi's, OpenHands' and Pydantic AI's images too, at `ghcr.io/scaleapi/agentenv-codex:src-`, `ghcr.io/scaleapi/agentenv-gemini:src-`, -`ghcr.io/scaleapi/agentenv-opencode:src-`, `ghcr.io/scaleapi/agentenv-pi:src-`, +`ghcr.io/scaleapi/agentenv-hermes:src-`, `ghcr.io/scaleapi/agentenv-opencode:src-`, `ghcr.io/scaleapi/agentenv-pi:src-`, `ghcr.io/scaleapi/agentenv-openhands:src-` and `ghcr.io/scaleapi/agentenv-pydantic-ai:src-`, each built again only when its folder changes, and a run deploys them as it does Claude Code's: it pulls the image published for the sources installed, by its digest, and builds it from the agent's folder only when it can't. @@ -107,5 +115,5 @@ store credentials. A release publishes its image at `ghcr.io/scaleapi/agentenv-p others'. It has no alias and no smoke task: with nobody driving it, a prompt only waits out its deadline. `agent-env run agents` runs every agent's smoke task, pulling or building each image; `--task claude-code-smoke` runs -Claude Code's alone, `--task codex-smoke` Codex's, `--task gemini-smoke` Gemini's, `--task opencode-smoke` OpenCode's, +Claude Code's alone, `--task codex-smoke` Codex's, `--task gemini-smoke` Gemini's, `--task hermes-smoke` Hermes', `--task opencode-smoke` OpenCode's, `--task pi-smoke` pi's, `--task openhands-smoke` OpenHands', `--task pydantic-ai-smoke` Pydantic AI's. diff --git a/src/agent_env/agents/agents/hermes/.dockerignore b/src/agent_env/agents/agents/hermes/.dockerignore new file mode 100644 index 000000000..8d35cb327 --- /dev/null +++ b/src/agent_env/agents/agents/hermes/.dockerignore @@ -0,0 +1,2 @@ +__pycache__ +*.pyc diff --git a/src/agent_env/agents/agents/hermes/Dockerfile b/src/agent_env/agents/agents/hermes/Dockerfile new file mode 100644 index 000000000..5b143517f --- /dev/null +++ b/src/agent_env/agents/agents/hermes/Dockerfile @@ -0,0 +1,34 @@ +FROM public.ecr.aws/docker/library/python:3.11-slim@sha256:c20888b6acdd1e63e1c433a185bf3ad162c0288fe484616ce062e0d28add2900 +# Hermes Agent at a pinned release, installed by its own installer, which the image checks against its checksum. The +# installer brings Hermes' runtime (its Python, Node and ripgrep among them) under its home, /data/.hermes; git and xz +# are what it needs to fetch them. poppler-utils renders a PDF attachment to the page images Hermes looks at. +ARG HERMES_COMMIT=bf0203a794a07aac3ff764972ab9087baa4e4f4e +ARG HERMES_INSTALLER_SHA256=99cb3691d55019c6973a7acb775a915648666cc24e483ca827a1cdbb3531a674 +RUN set -eu; \ + apt-get update && apt-get install -y --no-install-recommends ca-certificates curl git poppler-utils xz-utils \ + && rm -rf /var/lib/apt/lists/*; \ + useradd --create-home --uid 10001 agent && mkdir -p /app /data; \ + curl -fsSL --retry 5 --retry-all-errors \ + "https://raw.githubusercontent.com/NousResearch/hermes-agent/${HERMES_COMMIT}/scripts/install.sh" -o /tmp/install.sh; \ + echo "$HERMES_INSTALLER_SHA256 /tmp/install.sh" | sha256sum -c -; \ + HOME=/data bash /tmp/install.sh --commit "$HERMES_COMMIT" --dir /opt/hermes --hermes-home /data/.hermes \ + --non-interactive --skip-browser --skip-computer-use; \ + # What a turn never reads: the checkout's history, its UI builds' sources and dependencies, tests and docs, and + # the installer's download cache. + rm -rf /tmp/install.sh /opt/hermes/.git /opt/hermes/node_modules /opt/hermes/website /opt/hermes/tests \ + /data/.hermes/cache /data/.local /data/.bashrc /data/.profile; \ + ln -s /opt/hermes/.hermes/bin/hermes /usr/local/bin/hermes; \ + chown -R agent:agent /app /data +WORKDIR /opt/agent +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt +COPY a2a_server.py hermes_bridge.py hermes_kit.py entrypoint.sh ./ +COPY agentenv_plugin ./agentenv_plugin +RUN chmod 0755 entrypoint.sh +# Each context's Hermes profile is under /data/.hermes, Hermes' home, whatever the uid: Modal runs the image as root. +ENV A2A_PORT=8000 PYTHONPATH=/opt/agent HOME=/data HERMES_ROOT=/data/.hermes +USER agent +WORKDIR /app +EXPOSE 8000 +# CMD rather than ENTRYPOINT: Modal passes a sandbox's env to the image's CMD only. +CMD ["/opt/agent/entrypoint.sh"] diff --git a/src/agent_env/agents/agents/hermes/README.md b/src/agent_env/agents/agents/hermes/README.md new file mode 100644 index 000000000..ebaf5e09c --- /dev/null +++ b/src/agent_env/agents/agents/hermes/README.md @@ -0,0 +1,49 @@ +# Hermes Agent A2A agent + +Nous Research's [Hermes Agent](https://github.com/NousResearch/hermes-agent) wrapped with the +`agentenv_protocol.a2a_agent` framework, shipped in the `agents` bundle as `@local/agentenv-framework/agents/hermes`. +This directory is the whole build context: the protocol package comes from PyPI (`requirements.txt`), and Hermes from +its own installer at a pinned release, checked against its checksum, in the `Dockerfile` (`HERMES_COMMIT`, the +`v0.22.5` release). + +Each context is a Hermes profile of its own, `$HERMES_ROOT/profiles/ctx-/`, with its own config, sessions and +memory. Each task runs `hermes agentenv turn` once in `/app`, a command the agent's Hermes plugin, `agentenv_plugin/`, +adds: it runs the turn in its own process, as `hermes -z` does, rather than handing it to a Hermes gateway, so the turn +ends with that process and a timeout or cancel that kills it, and everything it started, stops the turn. A context's +first task starts a Hermes session, and each later task resumes it with `--resume `. Text parts are the +prompt, and file parts are written under `/app/inputs//` and named in it, for Hermes' file and vision tools; a +PDF goes with its pages' PNGs, rendered with `pdftoppm`. Turns of different contexts run at once, as many as memory fits +at about 400 MB each past the first 1 GB (`HERMES_MAX_CONCURRENT_TURNS` sets the number), and a context's turns one at +a time; a task's files are downloaded and rendered only once it has a turn's slot, so they count against that number. + +The profile's `config.yaml` is written before each turn: + +- the model endpoint, `$LITELLM_BASE_URL/v1`, as a custom OpenAI-compatible provider with its key from + `LITELLM_API_KEY`, `[model.headers]` (`LITELLM_EXTRA_HEADERS`) and the run's `project_id` and `task_id` as + `x-litellm-customer-id` and `x-litellm-tags` on every model call, Hermes' auxiliary calls (titles, compression, + vision) included. The run's tags join an `x-litellm-tags` that `[model.headers]` names, whose own keys, such as + `taskId`, win; +- the model: the task's, else `HERMES_MODEL`, else `anthropic/claude-sonnet-4-6`; +- every registered MCP server, the task's `peer_agents` tools included, as a remote server. Its `AgentEnv-Role` header + is the role its registration names, unless the task has a role of its own; +- the skills agent-env installed, as an external skills folder, and the `agentenv` plugin enabled. + +The task's `system_prompt` is the profile's `SOUL.md`, the slot Hermes' own identity fills, so it reaches the model as +the system prompt; a session keeps the system prompt it started with, as Hermes does. A turn with no system prompt gets +Hermes' own back. Hermes runs non-interactively: dangerous-command approvals are bypassed, and its `clarify` tool tells +the model no one is there to answer. + +| Extension | Implementation | +|---|---| +| `agent-config/v1` | `HermesConfig`: `model`, `effort` (Hermes' `--reasoning`), `system_prompt`, `output_format` (a JSON schema the answer is checked against), `project_id`, `task_id`, `timeout_seconds` (1800 by default) | +| `mcp-config/v1` | SDK default; each server written to the turn's profile config | +| `skill-config/v1` | inline and bundle skills installed under `$HERMES_ROOT/agentenv-skills//`, which every profile reads as `skills.external_dirs` | +| `trajectory/v1` | the turn's Hermes plugin hook events (`on_session_start`, `pre_tool_call`, `post_tool_call`, `post_api_request` with each model response, its tool calls and usage, `api_request_error`, `post_llm_call`, `on_session_end`), live while it runs (format `hermes-hook-events/1`) | +| `snapshot/v1` | `save` archives the context's profile (its sessions database, memory and `SOUL.md`; not the config, caches or logs Hermes rebuilds) with its session id, and `/app`, once no task runs; `load` restores the profile into the target context and replaces `/app`'s contents, both or neither, so a failed load leaves the context's profile, session and workspace as they were. No changelog | +| `peer-agents/v1` | SDK default: Hermes reaches the task's other agents through the SDK's `peer_agents` MCP tools | +| `attribution-probe/v1` | the `project_id` and `task_id` the last turn sent | + +A turn's usage is Hermes' own report for its main model calls: tokens, `api_calls`, and its estimated cost where the +endpoint's model list prices the model. The image is `python:3.11-slim` with Hermes, `pdftoppm` and the agent, running +as the `agent` user. Hermes' runtime (its own Python, Node and ripgrep) and the profiles are under `/data/.hermes`, +whatever the uid. It has no `install/v1`. diff --git a/src/agent_env/agents/agents/hermes/a2a_server.py b/src/agent_env/agents/agents/hermes/a2a_server.py new file mode 100644 index 000000000..54092f277 --- /dev/null +++ b/src/agent_env/agents/agents/hermes/a2a_server.py @@ -0,0 +1,594 @@ +"""A2A agent wrapping Hermes Agent. + +Built on agentenv-protocol's `AgentEnvAgent`: the framework serves the Agent Card, the A2A task lifecycle and the +SDK-owned extensions (agent-config, mcp-config, skill listing, task trajectories, peer agents). Each turn is one +`hermes agentenv turn` in the context's own Hermes profile, whose hook events are appended to the framework's +trajectory log as Hermes writes them. A context's first turn starts a Hermes session, which later turns resume. +""" +from __future__ import annotations + +import asyncio +import base64 +import binascii +import contextlib +import functools +import json +import logging +import os +import re +import shutil +import subprocess +import tarfile +import tempfile +import uuid +from collections.abc import Callable, Mapping +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any +from urllib.parse import urlparse +from weakref import WeakValueDictionary + +import httpx +from a2a.types import AgentSkill +from agentenv_protocol.a2a_agent import ( + ATTRIBUTION_PROBE_V1, + MCP_CONFIG_V1, + PEER_AGENTS_V1, + SKILL_CONFIG_V1, + SNAPSHOT_V1, + TRAJECTORY_V1, + AgentConfig, + AgentEnvAgent, + AgentIdentity, + AttributionProbeResponse, + BundleSkillRequest, + DataPart, + FilePart, + InlineSkillRequest, + ObjectSnapshotLoadRequest, + ObjectSnapshotSaveRequest, + SkillAddResponse, + TaskRequest, + TaskResult, + TextPart, + Usage, + a2a_agent, + download, + enable, + extension, + upload, +) +from pydantic import Field, field_validator +from starlette.applications import Starlette +from starlette.exceptions import HTTPException + +from hermes_bridge import ( + ERROR_MESSAGE_LIMIT, + HERMES_ROOT, + get_last_attribution_snapshot, + kill_running_clis, + profile_dir, + run_hermes, + validate_output_format, +) +from hermes_kit import DurableSessionRefs, in_thread_to_its_end, install_skill, json_error_bodies, to_its_end + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Skills agent-env installs, which every context's profile reads as an external skills folder. +SKILLS_DIR = Path(os.environ.get("HERMES_AGENTENV_SKILLS_DIR") or HERMES_ROOT / "agentenv-skills") +WORKSPACE_DIR = Path(os.environ.get("HERMES_WORKSPACE_DIR", "/app")) +INPUTS_DIR = WORKSPACE_DIR / "inputs" +_FILE_DOWNLOAD_TIMEOUT_S = 60 +_PDF_RENDER_TIMEOUT_S = 60 +_PDF_RENDER_DPI = 150 +_SAFE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]") +# The versioned format of the Hermes hook events this agent appends to its trajectory. +TRAJECTORY_FORMAT = "hermes-hook-events/1" +# A snapshot's conversation object: the context's profile, all but what Hermes rebuilds or this agent rewrites. +_SNAPSHOT_EXCLUDED = frozenset({"cache", "logs", "plugins", "skills", "config.yaml", "audio_cache", "image_cache"}) +_SNAPSHOT_SESSION = "agentenv-session.json" + + +class HermesConfig(AgentConfig): + model: str | None = None + effort: str | None = None + system_prompt: str | None = None + output_format: dict[str, Any] | None = None + project_id: str | None = None + task_id: str | None = None + timeout_seconds: int = Field(default=1800, gt=0) + + @field_validator("output_format") + @classmethod + def check_output_format(cls, value: dict[str, Any] | None) -> dict[str, Any] | None: + if value is not None: + validate_output_format(value) + return value + + +class FilePartMaterializeError(Exception): + """A supplied file cannot be passed to Hermes.""" + + +def _safe_name(candidate: str | None, fallback_ext: str = "") -> str: + if candidate: + candidate = _SAFE_NAME_RE.sub("_", Path(candidate).name) + return candidate or f"{uuid.uuid4().hex}{fallback_ext}" + + +async def _download_uri_to(uri: str, target: Path) -> None: + """Write an http(s) file to target as it arrives, so a large one never sits whole in memory; a cancelled turn's + download stops with it.""" + parsed = urlparse(uri) + if parsed.scheme not in ("http", "https"): + raise FilePartMaterializeError(f"unsupported file URI scheme {parsed.scheme!r}") + try: + async with httpx.AsyncClient(timeout=_FILE_DOWNLOAD_TIMEOUT_S, follow_redirects=True) as client: + async with client.stream("GET", uri) as response: + response.raise_for_status() + with target.open("wb") as out: + async for chunk in response.aiter_bytes(): + out.write(chunk) + except httpx.HTTPError as exc: + raise FilePartMaterializeError("HTTP file download failed") from exc + + +def _write_base64(target: Path, data: str) -> None: + target.write_bytes(base64.b64decode(data, validate=True)) + + +async def _render_pdf(target: Path, prefix: str) -> bool: + """Whether pdftoppm rendered the PDF's pages beside it; one that outlives its timeout, or a cancelled turn, is + killed and reaped before this returns.""" + try: + proc = await asyncio.create_subprocess_exec( + "pdftoppm", "-png", "-r", str(_PDF_RENDER_DPI), str(target), prefix, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + except FileNotFoundError: + return False + try: + return await asyncio.wait_for(proc.wait(), _PDF_RENDER_TIMEOUT_S) == 0 + except TimeoutError: + return False + finally: + if proc.returncode is None: + with contextlib.suppress(ProcessLookupError): + proc.kill() + await to_its_end(proc.wait()) + + +async def _materialize_and_render(part: FilePart, task_id: str) -> list[str]: + """A file part's path under the task's inputs, and a PDF's page images; each write ends before this returns, even + for a cancelled turn.""" + directory = INPUTS_DIR / task_id + directory.mkdir(parents=True, exist_ok=True) + name = _safe_name(part.name) + target = directory / name + if target.exists(): + target = directory / f"{uuid.uuid4().hex[:8]}-{name}" + if part.bytes is not None: + try: + await in_thread_to_its_end(_write_base64, target, part.bytes) + except (binascii.Error, ValueError, OSError) as exc: + raise FilePartMaterializeError(f"invalid file attachment {name!r}") from exc + else: + await _download_uri_to(part.uri, target) + if (part.mime_type or "").lower() != "application/pdf" and target.suffix.lower() != ".pdf": + return [str(target)] + # Hermes looks at images with its vision tool, so a PDF goes as its pages' images too: in a folder of their own, so + # two PDFs whose names share a stem (report.pdf, report.PDF) never mix their pages. The folder's name is short, as a + # name derived from the file's could pass the filesystem's 255-byte limit. + pages_dir = Path(tempfile.mkdtemp(prefix="pdf-pages-", dir=directory)) + if not await _render_pdf(target, str(pages_dir / "page")): + logger.warning("PDF rendering failed for %s; passing the PDF alone", target) + return [str(target)] + pages = sorted(pages_dir.glob("page*.png")) + return [str(target), *(str(page) for page in pages)] + + +def _with_attachments(prompt: str, paths: list[str]) -> str: + """The prompt, naming the files the task attached, which Hermes reads with its file and vision tools.""" + if not paths: + return prompt + listing = "\n".join(f"- {path}" for path in paths) + return f"{prompt}\n\nAttached files (read them, or look at images with your vision tool):\n{listing}" + + +class _WorkspaceBarrier: + """Turns share the workspace and snapshots take it alone, so a snapshot's session and files come from a point no + turn is writing at, while turns of different contexts run at once. A waiting snapshot holds new turns back, so + it is never starved.""" + + def __init__(self) -> None: + self._condition = asyncio.Condition() + self._readers = 0 + self._writer = False + self._waiting_writers = 0 + + @asynccontextmanager + async def shared(self): + async with self._condition: + await self._condition.wait_for(lambda: not self._writer and self._waiting_writers == 0) + self._readers += 1 + try: + yield + finally: + async with self._condition: + self._readers -= 1 + if self._readers == 0: + self._condition.notify_all() + + @asynccontextmanager + async def exclusive(self): + async with self._condition: + self._waiting_writers += 1 + try: + await self._condition.wait_for(lambda: not self._writer and self._readers == 0) + self._writer = True + finally: + self._waiting_writers -= 1 + if not self._writer: + self._condition.notify_all() + try: + yield + finally: + async with self._condition: + self._writer = False + self._condition.notify_all() + + +@asynccontextmanager +async def lifespan(app: Starlette): + """As the server stops, the processes of turns still running are killed, which would otherwise go on in the + workspace after it.""" + try: + yield + finally: + if stopped := kill_running_clis(): + logger.info(f"Killed {stopped} hermes turn(s) still running as the agent shut down") + + +# A Hermes turn's process peaked near 210 MB running shell and Python tools, and its tools' own programs can take as +# much again; the server and the OS need about 1 GB more. Turns past the cap wait rather than run the machine out of +# memory. +TURN_MEMORY_MB, RESERVED_MEMORY_MB = 400, 1024 + + +def max_concurrent_turns(environ: Mapping[str, str] = os.environ) -> int: + """How many Hermes turns may run at once: HERMES_MAX_CONCURRENT_TURNS, else as many as memory fits.""" + if configured := environ.get("HERMES_MAX_CONCURRENT_TURNS"): + return max(1, int(configured)) + return max(1, (memory_mb() - RESERVED_MEMORY_MB) // TURN_MEMORY_MB) + + +def memory_mb(cgroup_limit: Path = Path("/sys/fs/cgroup/memory.max")) -> int: + """The memory this server's processes may use: the machine's, or its cgroup's limit when that's lower.""" + total = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") + try: + limit = cgroup_limit.read_text().strip() + except OSError: + limit = "max" + if limit.isdigit(): + total = min(total, int(limit)) + return total // 2**20 + + +def _archive_profile(home: Path, session_id: str, archive: Path) -> None: + """The context's conversation as a tar.gz: its profile, all but what Hermes rebuilds, with its session id.""" + (home / _SNAPSHOT_SESSION).write_text(json.dumps({"session_id": session_id})) + try: + with tarfile.open(archive, "w:gz") as tar: + for child in sorted(home.iterdir()): + if child.name not in _SNAPSHOT_EXCLUDED: + tar.add(child, arcname=child.name) + finally: + (home / _SNAPSHOT_SESSION).unlink(missing_ok=True) + + +def _stage_profile(archive: Path, home: Path) -> tuple[Path, str]: + """An archived profile extracted beside the context's, which stays untouched; the staged folder and the archived + session id.""" + staged = Path(tempfile.mkdtemp(prefix=".restore-", dir=home.parent)) + try: + with tarfile.open(archive, "r:gz") as tar: + tar.extractall(staged, filter="data") + session_id = json.loads((staged / _SNAPSHOT_SESSION).read_text()).get("session_id") + if not isinstance(session_id, str) or not session_id: + raise ValueError("the snapshot names no Hermes session") + (staged / _SNAPSHOT_SESSION).unlink() + return staged, session_id + except BaseException: + shutil.rmtree(staged, ignore_errors=True) + raise + + +def _commit_profile(staged: Path, home: Path) -> None: + """Swap a staged profile in for the context's; the previous profile comes back if the move fails.""" + previous = home.with_name(home.name + ".previous") + shutil.rmtree(previous, ignore_errors=True) + if home.exists(): + home.rename(previous) + try: + staged.rename(home) + except BaseException: + shutil.rmtree(home, ignore_errors=True) + if previous.exists(): + previous.rename(home) + raise + # The swap has succeeded; removing the old profile is cleanup and must not undo it. + shutil.rmtree(previous, ignore_errors=True) + + +def _tar(*args: str) -> None: + subprocess.run(["tar", *args], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + + +@a2a_agent( + identity=AgentIdentity( + name="Hermes Agent", + description="A2A-compliant wrapper around Nous Research's Hermes Agent.", + version="1.0.0", + input_modes=("text", "image/png", "image/jpeg", "image/gif", "image/webp", "application/pdf"), + output_modes=("text",), + skills=(AgentSkill( + id="hermes", name="hermes", + description="Runs Hermes Agent with its full toolset (terminal, files, code execution, web, skills, " + "memory, subagents).", + tags=["coding", "agent"], + ),), + ), + config=HermesConfig, + extensions=( + MCP_CONFIG_V1, + enable(TRAJECTORY_V1, live=True), + enable(ATTRIBUTION_PROBE_V1), + # SDK default: Hermes reaches the task's other agents through the SDK's peer_agents MCP tools. + PEER_AGENTS_V1, + ), + lifespan=lifespan, + workspace=str(WORKSPACE_DIR), +) +class HermesAgent(DurableSessionRefs, AgentEnvAgent): + """Bridges A2A tasks onto `hermes agentenv turn`.""" + + def __init__(self) -> None: + self._turn_slots = asyncio.Semaphore(max_concurrent_turns()) + self._turn_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary() + self._workspace = _WorkspaceBarrier() + + def _turn_lock(self, context_id: str) -> asyncio.Lock: + return self._turn_locks.setdefault(context_id, asyncio.Lock()) + + async def run(self, request: TaskRequest[HermesConfig]) -> TaskResult: + # The plugin's hook events, appended as Hermes writes them, are the turn's trajectory: readable while it runs, + # and its final record once it ends. + request.trajectory.set_format(TRAJECTORY_FORMAT) + prompt = "\n".join(part.text for part in request.parts if isinstance(part, TextPart) and part.text) + context_id = request.context_id + async with self._workspace.shared(), self._turn_lock(context_id): + # Read under the lock rather than from the request: a snapshot load that held it may have just bound this + # context to a restored session. + session_id = self.session_ref_for_context(context_id) + + def record(event: dict) -> None: + request.trajectory.append(event) + # A first turn's session is bound as soon as Hermes names it, so a cancel can't lose it. + started = event.get("session_id") + if self.session_ref_for_context(context_id) is None and isinstance(started, str) and started: + self.set_session_ref_for_context(context_id, started) + + # Attachments count against the turn limit too: their downloads and PDF renders take memory and processes. + async with self._turn_slots: + try: + attachments = [] + for part in request.parts: + if isinstance(part, FilePart): + attachments.extend(await _materialize_and_render(part, request.task_id)) + except FilePartMaterializeError as exc: + return TaskResult.failure("filepart_error", str(exc), error_type="infra_error") + try: + result = await run_hermes( + prompt=_with_attachments(prompt, attachments), + context_id=context_id, + session_id=session_id, + model=request.config.model, + effort=request.config.effort, + mcp_servers=dict(request.mcp_servers), + system_prompt=request.config.system_prompt, + output_format=request.config.output_format, + timeout_s=request.config.timeout_seconds, + project_id=request.config.project_id, + task_id=request.config.task_id, + role=request.config.role, + skills_dir=SKILLS_DIR, + event_callback=record, + workspace=str(WORKSPACE_DIR), + ) + except Exception: + logger.exception("Hermes invocation failed") + return TaskResult.failure("hermes_invocation_error", "Hermes invocation failed", + error_type="infra_error") + if result.session_id: + self.set_session_ref_for_context(context_id, result.session_id) + + parts = [TextPart(text=result.response or result.error or "")] + usage = _usage(result) + if result.error: + details = {"error_type": "cli_error", "error_message": result.error[:ERROR_MESSAGE_LIMIT]} + if result.error_code: + details["error_code"] = result.error_code + parts.append(DataPart(data=details)) + return ( + TaskResult.builder() + .failed(result.error_code or "hermes_cli_error", result.error[:ERROR_MESSAGE_LIMIT]) + .parts(parts) + .session_ref(result.session_id) + .usage(usage) + .build() + ) + response = ( + TaskResult.builder() + .succeeded() + .parts(parts) + .session_ref(result.session_id) + .usage(usage) + ) + if request.config.output_format is not None: + response.add_structured_output(result.structured_output) + return response.build() + + @extension(SKILL_CONFIG_V1.add.inline) + async def add_inline_skill(self, request: InlineSkillRequest) -> SkillAddResponse: + return await install_skill(request, SKILLS_DIR) + + @extension(SKILL_CONFIG_V1.add.bundle) + async def add_bundle_skill(self, request: BundleSkillRequest) -> SkillAddResponse: + return await install_skill(request, SKILLS_DIR) + + @extension(ATTRIBUTION_PROBE_V1.probe) + async def probe_attribution(self) -> AttributionProbeResponse: + """The LiteLLM attribution the last turn sent with its model calls.""" + snapshot = get_last_attribution_snapshot() + return AttributionProbeResponse( + last_seen_attribution={ + key: value for key, value in snapshot.items() if key != "last_seen_at_utc" and isinstance(value, str) + }, + last_seen_at_utc=snapshot.get("last_seen_at_utc"), + ) + + @extension(SNAPSHOT_V1.save) + async def save_snapshot(self, request: ObjectSnapshotSaveRequest) -> dict[str, Any]: + with tempfile.TemporaryDirectory(prefix="hermes-snapshot-") as directory: + conversation = Path(directory) / "conversation.tgz" + workspace = Path(directory) / "workspace.tgz" + async with self._workspace.exclusive(), self._turn_lock(request.context_id): + # Read once no turn runs: a context's first turn, or a load into it, may be about to bind it. + session_id = self.session_ref_for_context(request.context_id) + if not session_id: + raise HTTPException(status_code=404, detail="No active session for context_id") + # Each to its end even for a cancelled save, so neither writes past the exclusive hold. + await in_thread_to_its_end(_archive_profile, profile_dir(request.context_id), session_id, conversation) + if request.objects.workspace is not None: + await in_thread_to_its_end(_tar, "-czf", str(workspace), "-C", str(WORKSPACE_DIR), ".") + objects = {"trajectory": (await upload(request.objects.trajectory, conversation)).model_dump(mode="json")} + if request.objects.workspace is not None: + objects["workspace"] = (await upload(request.objects.workspace, workspace)).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]: + target_context_id = request.target_context_id or uuid.uuid4().hex + with tempfile.TemporaryDirectory(prefix="hermes-restore-") as directory: + conversation = Path(directory) / "conversation.tgz" + await download(request.objects.trajectory, conversation) + staged = None + if request.objects.workspace is not None: + archive = Path(directory) / "workspace.tgz" + staged = Path(directory) / "workspace" + staged.mkdir() + await download(request.objects.workspace, archive) + await in_thread_to_its_end(_tar, "-xzf", str(archive), "-C", str(staged)) + async with self._workspace.exclusive(), self._turn_lock(target_context_id): + home = profile_dir(target_context_id) + home.parent.mkdir(parents=True, exist_ok=True) + # Stage the conversation first, so a bad snapshot changes nothing. The workspace and conversation swaps + # then run in one thread, and the context is bound to the restored session, in one step a cancel can't + # abandon: a failed conversation swap puts the workspace back, and the context's previous profile stays + # in place until both succeed. + try: + staged_home, session_id = await in_thread_to_its_end(_stage_profile, conversation, home) + except (tarfile.TarError, OSError, ValueError) as exc: + raise HTTPException(status_code=400, detail=f"Unreadable Hermes snapshot: {exc}") from exc + commit = functools.partial(_commit_profile, staged_home, home) + + async def swap() -> None: + try: + if staged is not None: + await asyncio.to_thread(_replace_workspace, staged, commit) + else: + await asyncio.to_thread(commit) + finally: + await asyncio.to_thread(shutil.rmtree, staged_home, True) + self.set_session_ref_for_context(target_context_id, session_id) + + await to_its_end(swap()) + return {"context_id": target_context_id} + + +def _usage(result) -> Usage: + """The turn's tool calls and the token counts and cost Hermes reported for its main model calls.""" + report = result.usage or {} + + def count(name: str) -> int | None: + value = report.get(name) + return value if isinstance(value, int) and not isinstance(value, bool) and value >= 0 else None + + cost = report.get("estimated_cost_usd") + details = {key: report[key] for key in ("api_calls", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens", + "cost_status", "turn_exit_reason") if report.get(key) is not None} + return Usage( + tool_call_count=result.tool_call_count, + input_tokens=count("input_tokens"), + output_tokens=count("output_tokens"), + total_tokens=count("total_tokens"), + cost_usd=cost if isinstance(cost, (int, float)) and not isinstance(cost, bool) and cost >= 0 else None, + provider_details=details, + ) + + +def _replace_workspace(staged: Path, then: Callable[[], None] | None = None) -> None: + """Swap staged in for the workspace's contents. When then() raises after the swap, the previous contents come + back.""" + WORKSPACE_DIR.mkdir(parents=True, exist_ok=True) + backup = Path(tempfile.mkdtemp(prefix=".hermes-backup-", dir=WORKSPACE_DIR)) + saved = backup / "saved" + saved.mkdir() + committed = False + try: + try: + for child in WORKSPACE_DIR.iterdir(): + if child != backup: + child.rename(saved / child.name) + except Exception: + for child in saved.iterdir(): + child.rename(WORKSPACE_DIR / child.name) + raise + try: + for child in staged.iterdir(): + shutil.move(str(child), str(WORKSPACE_DIR / child.name)) + if then is not None: + then() + committed = True + except Exception: + for child in WORKSPACE_DIR.iterdir(): + if child == backup: + continue + if child.is_dir() and not child.is_symlink(): + shutil.rmtree(child) + else: + child.unlink() + for child in saved.iterdir(): + child.rename(WORKSPACE_DIR / child.name) + raise + finally: + if committed: + try: + shutil.rmtree(backup) + except OSError: + logger.warning("Could not remove snapshot backup %s", backup, exc_info=True) + elif not any(saved.iterdir()): + saved.rmdir() + backup.rmdir() + + +def build_app(agent: HermesAgent) -> Starlette: + """The agent's Starlette app, answering errors with FastAPI's `{"detail": ...}` bodies.""" + return json_error_bodies(agent.create_app()) + + +app = build_app(HermesAgent()) +# agent-env reads the card from the image by importing this name. +AGENT_CARD = app.state.agentenv_a2a.card diff --git a/src/agent_env/agents/agents/hermes/agentenv_plugin/__init__.py b/src/agent_env/agents/agents/hermes/agentenv_plugin/__init__.py new file mode 100644 index 000000000..f12aa0ef3 --- /dev/null +++ b/src/agent_env/agents/agents/hermes/agentenv_plugin/__init__.py @@ -0,0 +1,75 @@ +"""A Hermes plugin, loaded by Hermes itself, that the agent's bridge drives each turn through. + +`hermes agentenv turn` runs one turn in this process, the way `hermes -z` once did, rather than handing it to a Hermes +gateway: so the turn ends with this process, and a timeout or cancel that kills it stops the turn. Hermes' plugin +hooks write the turn's events, each as one JSON line, to the pipe AGENTENV_EVENTS_FD names, as they happen; the bridge +reads them as the turn's trajectory. Each event is a hook's name as ``type`` and the hook's fields the bridge keeps. +""" + +from __future__ import annotations + +import json +import os +import threading +import time + +_EVENTS_FD = "AGENTENV_EVENTS_FD" +_write_lock = threading.Lock() + +# Each hook this plugin records, and the fields of its payload an event keeps. +_HOOKS = { + "on_session_start": ("session_id", "model"), + "pre_tool_call": ("session_id", "turn_id", "tool_call_id", "tool_name", "args"), + "post_tool_call": ("session_id", "turn_id", "tool_call_id", "tool_name", "result", "status", "duration_ms", + "error_type", "error_message"), + # ``response`` is Hermes' sanitized record of the model's answer: its assistant message, tool calls and usage. + "post_api_request": ("session_id", "turn_id", "api_call_count", "model", "finish_reason", "response"), + "api_request_error": ("session_id", "turn_id", "api_call_count", "model", "status_code", "reason", "retryable", + "error"), + "post_llm_call": ("session_id", "turn_id", "assistant_response"), + "on_session_end": ("session_id", "turn_id", "completed", "failed", "interrupted", "turn_exit_reason"), +} + + +def _write(event: dict) -> None: + descriptor = os.environ.get(_EVENTS_FD) + if not descriptor: + return + line = (json.dumps(event, ensure_ascii=False, default=str) + "\n").encode() + with _write_lock: + os.write(int(descriptor), line) + + +def _recorder(hook: str, fields: tuple[str, ...]): + def record(**payload) -> None: + event = {"type": hook, "timestamp": int(time.time() * 1000)} + event.update({name: payload[name] for name in fields if payload.get(name) is not None}) + _write(event) + return record + + +def _setup(parser) -> None: + commands = parser.add_subparsers(dest="agentenv_command", required=True) + turn = commands.add_parser("turn", help="Run one turn and exit") + turn.add_argument("--prompt-file", required=True, help="The turn's message, read from this file") + turn.add_argument("--model") + turn.add_argument("--provider") + turn.add_argument("--resume", help="The session the turn continues") + turn.add_argument("--reasoning", help="The reasoning effort") + turn.add_argument("--usage-file", help="Where to write the turn's usage report") + + +def _handle(args) -> None: + # The one-shot runner: the turn in this process, then Hermes' own cleanup and a hard exit. + from hermes_cli.main import _run_and_exit_oneshot + + with open(args.prompt_file, encoding="utf-8") as prompt: + message = prompt.read() + _run_and_exit_oneshot(message, model=args.model, provider=args.provider, resume=args.resume, + reasoning=args.reasoning, usage_file=args.usage_file) + + +def register(ctx) -> None: + ctx.register_cli_command("agentenv", "Run one agent-env turn in this process", _setup, _handle) + for hook, fields in _HOOKS.items(): + ctx.register_hook(hook, _recorder(hook, fields)) diff --git a/src/agent_env/agents/agents/hermes/agentenv_plugin/plugin.yaml b/src/agent_env/agents/agents/hermes/agentenv_plugin/plugin.yaml new file mode 100644 index 000000000..eba9a5298 --- /dev/null +++ b/src/agent_env/agents/agents/hermes/agentenv_plugin/plugin.yaml @@ -0,0 +1,3 @@ +name: agentenv +version: 1.0.0 +description: Runs one agent-env turn in process (`hermes agentenv turn`) and writes the turn's hook events as JSON lines to the file descriptor AGENTENV_EVENTS_FD names. diff --git a/src/agent_env/agents/agents/hermes/entrypoint.sh b/src/agent_env/agents/agents/hermes/entrypoint.sh new file mode 100644 index 000000000..8de93566d --- /dev/null +++ b/src/agent_env/agents/agents/hermes/entrypoint.sh @@ -0,0 +1,4 @@ +#!/bin/sh +set -e +# A stop waits this long for open requests, then the server shuts down and kills the turns still running. +exec uvicorn a2a_server:app --host 0.0.0.0 --port "${A2A_PORT:-8000}" --timeout-graceful-shutdown 5 diff --git a/src/agent_env/agents/agents/hermes/hermes_bridge.py b/src/agent_env/agents/agents/hermes/hermes_bridge.py new file mode 100644 index 000000000..c9c0c4b4d --- /dev/null +++ b/src/agent_env/agents/agents/hermes/hermes_bridge.py @@ -0,0 +1,490 @@ +"""Subprocess wrapper around Hermes Agent: one `hermes agentenv turn` per turn. + + - Each A2A context is a Hermes profile of its own, `/profiles/ctx-`, with its own config, + sessions and memory, so a session id resumes only that conversation and concurrent turns never share a file. + - The profile's config.yaml is written before each turn: the model endpoint as a named custom provider with its + headers and the run's attribution, the turn's MCP servers, the shared skills folder, and the agentenv plugin + enabled. The system prompt is the profile's SOUL.md, Hermes' identity slot. + - The turn runs in the plugin's process, not in a Hermes gateway, so killing that process stops the turn. The + plugin writes each hook event as a JSON line to a pipe this module reads as it goes: the turn's trajectory. +""" +from __future__ import annotations + +import asyncio +import contextlib +import hashlib +import json +import logging +import os +import shutil +import signal +import tempfile +import threading +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path + +from jsonschema import Draft202012Validator, SchemaError, ValidationError + +from hermes_kit import in_thread_to_its_end + +logger = logging.getLogger(__name__) + +# The model a run uses when neither the task nor the agent-config names one, sent to the endpoint as named. +DEFAULT_MODEL = os.environ.get("HERMES_MODEL", "anthropic/claude-sonnet-4-6") +ERROR_MESSAGE_LIMIT = 4000 +ROLE_HEADER = "AgentEnv-Role" +# What agent-env gives an agent for [model.headers]: a JSON object of header names to values. +MODEL_HEADERS_ENV = "LITELLM_EXTRA_HEADERS" +# The custom provider each profile's config names for the endpoint. +PROVIDER = "litellm" +# The Hermes install's home; each context's profile is under its profiles/. +HERMES_ROOT = Path(os.environ.get("HERMES_ROOT") or Path(os.environ.get("HOME", "/data")) / ".hermes") +HERMES_BIN = os.environ.get("HERMES_BIN", "hermes") +PLUGIN_DIR = Path(__file__).resolve().parent / "agentenv_plugin" +_PLUGIN_NAME = "agentenv" +# Marks a SOUL.md this agent wrote, so a later turn with no system prompt gives Hermes its own back. +_SOUL_MARKER = ".agentenv-soul" +_EVENTS_FD_ENV = "AGENTENV_EVENTS_FD" +# How long the event pipe may stay open after the turn's process exits, held by something it left behind. +_EVENTS_DRAIN_S = 2 + + +def _base_url(environ: Mapping[str, str]) -> str: + """The endpoint's chat-completions root: agent-env's LITELLM_BASE_URL has no /v1 suffix.""" + base = (environ.get("LITELLM_BASE_URL") or "").rstrip("/") + return base if base.endswith("/v1") else f"{base}/v1" + + +def profile_dir(context_id: str) -> Path: + """The context's Hermes profile home. A stable short hash keeps the path clean whatever the context id holds.""" + return HERMES_ROOT / "profiles" / f"ctx-{hashlib.sha1(context_id.encode()).hexdigest()[:16]}" + + +@dataclass +class HermesResult: + response: str + tool_call_count: int + session_id: str | None + trajectory: list[dict] = field(default_factory=list) + error: str | None = None + error_code: str | None = None + structured_output: object | None = None + usage: dict = field(default_factory=dict) + + +def validate_output_format(output_format: dict) -> dict: + if output_format.get("type") != "json_schema" or not isinstance(output_format.get("schema"), dict): + raise ValueError("output_format must contain type=json_schema and a schema object") + try: + Draft202012Validator.check_schema(output_format["schema"]) + except SchemaError as exc: + raise ValueError("output_format contains an invalid JSON schema") from exc + return output_format["schema"] + + +# The attribution the last run sent, which /ext/attribution-probe reads back so the validator can confirm +# project_id/task_id were wired through. +_last_attribution: dict[str, str | None] = {"project_id": None, "task_id": None, "last_seen_at_utc": None} +_attribution_lock = threading.Lock() + + +def _record_attribution(project_id: str | None, task_id: str | None) -> None: + if not project_id and not task_id: + return + with _attribution_lock: + _last_attribution["project_id"] = project_id or None + _last_attribution["task_id"] = task_id or None + _last_attribution["last_seen_at_utc"] = datetime.now(timezone.utc).isoformat() + + +def get_last_attribution_snapshot() -> dict[str, str | None]: + with _attribution_lock: + return dict(_last_attribution) + + +def _attribution_headers(project_id: str | None, task_id: str | None) -> dict[str, str]: + """LiteLLM cost-attribution headers, x-litellm-customer-id + x-litellm-tags; {} when nothing to attribute.""" + headers: dict[str, str] = {} + if project_id: + headers["x-litellm-customer-id"] = project_id + tags = [f"projectId:{project_id}"] + if task_id: + tags.append(f"taskId:{task_id}") + headers["x-litellm-tags"] = ",".join(tags) + elif task_id: + headers["x-litellm-tags"] = f"taskId:{task_id}" + return headers + + +def model_headers(raw: str | None) -> dict[str, str]: + if not raw: + return {} + try: + headers = json.loads(raw) + except json.JSONDecodeError: + logger.warning(f"ignoring {MODEL_HEADERS_ENV}, which isn't JSON") + return {} + if not isinstance(headers, dict): + logger.warning(f"ignoring {MODEL_HEADERS_ENV}, which isn't a JSON object") + return {} + return {str(name): str(value) for name, value in headers.items()} + + +_TAGS_HEADER = "x-litellm-tags" + + +def _join_tags(configured: str, run: str) -> str: + """x-litellm-tags as configured, then each of the run's tags whose key (the part before ``:``) they don't name.""" + tags = [tag.strip() for tag in configured.split(",") if tag.strip()] + keys = {tag.partition(":")[0] for tag in tags} + return ",".join([*tags, *(tag for tag in run.split(",") if tag and tag.partition(":")[0] not in keys)]) + + +def provider_headers(environ: Mapping[str, str], attribution: Mapping[str, str]) -> dict[str, str]: + """The endpoint's [model.headers], then the run's attribution, each replacing a header of the same name in any + case, except that the run's x-litellm-tags join the ones [model.headers] name.""" + configured = {name.lower(): value for name, value in model_headers(environ.get(MODEL_HEADERS_ENV)).items()} + if configured.get(_TAGS_HEADER) and _TAGS_HEADER in attribution: + attribution = {**attribution, _TAGS_HEADER: _join_tags(configured[_TAGS_HEADER], attribution[_TAGS_HEADER])} + return {name.lower(): value for layer in (configured, attribution) for name, value in layer.items()} + + +def _mcp_headers(headers: Mapping[str, str] | None, role: str | None) -> dict[str, str]: + """A server's registered headers; the run's role, when it has one, replaces a role header they name.""" + merged = dict(headers or {}) + if role: + merged = {name: value for name, value in merged.items() if name.lower() != ROLE_HEADER.lower()} + merged[ROLE_HEADER] = role + return merged + + +def build_config( + model: str, + mcp_servers: dict[str, dict] | None, + headers: dict[str, str], + role: str | None, + skills_dir: Path, + environ: Mapping[str, str] = os.environ, +) -> dict: + """The profile's config.yaml for one turn, as JSON (which YAML reads). Everything dynamic is rebuilt each turn, and + a context's turns run one at a time, so no turn reads another's.""" + provider: dict = {"api": _base_url(environ), "key_env": "LITELLM_API_KEY", "transport": "chat_completions"} + if headers: + provider["extra_headers"] = dict(headers) + config: dict = { + "model": {"provider": PROVIDER, "default": model}, + "providers": {PROVIDER: provider}, + "skills": {"external_dirs": [str(skills_dir)]}, + "plugins": {"enabled": [_PLUGIN_NAME]}, + } + if mcp_servers: + servers: dict[str, dict] = {} + for name, info in mcp_servers.items(): + entry: dict = {"url": info["url"]} + if hdrs := _mcp_headers(info.get("headers"), role): + entry["headers"] = hdrs + servers[name] = entry + config["mcp_servers"] = servers + return config + + +def prepare_profile(context_id: str, config: dict, system_prompt: str | None) -> Path: + """Write the context's profile for a turn: its config, the plugin, and the system prompt as SOUL.md.""" + home = profile_dir(context_id) + plugins = home / "plugins" + plugins.mkdir(parents=True, exist_ok=True) + (home / "config.yaml").write_text(json.dumps(config, indent=2)) + plugin = plugins / _PLUGIN_NAME + if not plugin.is_symlink(): + shutil.rmtree(plugin, ignore_errors=True) + plugin.symlink_to(PLUGIN_DIR, target_is_directory=True) + soul, marker = home / "SOUL.md", home / _SOUL_MARKER + if system_prompt: + soul.write_text(system_prompt) + marker.touch() + elif marker.exists(): + # Hermes writes its default identity back when SOUL.md is missing. + soul.unlink(missing_ok=True) + marker.unlink() + return home + + +_PROCESS_CLEANUP_TIMEOUT_S = 5 +_RUNNING: set = set() + + +def kill_running_clis() -> int: + """Kill every turn's process still running, and what each started; how many. The agent's shutdown does: the + processes, in sessions of their own, would outlive its server.""" + running = [proc for proc in _RUNNING if proc.returncode is None] + for proc in running: + _kill_cli(proc) + return len(running) + + +def _kill_cli(proc) -> None: + """Kill the turn's process and everything under it: its process group, and, found by parent, the tools it starts + in a session of their own, which no group or session kill reaches.""" + pid = getattr(proc, "pid", None) + if pid is None: + with contextlib.suppress(ProcessLookupError): + proc.kill() + return + if proc.returncode is not None: + return # the process is gone, and its pid may be another process's by now + tree = [pid, *_descendants(pid)] + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(pid, signal.SIGKILL) + for member in tree: + with contextlib.suppress(ProcessLookupError, PermissionError): + os.kill(member, signal.SIGKILL) + + +async def _started(spawn): + """The process ``spawn`` starts. A turn cancelled while it starts lets it finish starting, then kills it and what + it started: asyncio would kill the process alone, and leave its tools running.""" + start = asyncio.ensure_future(spawn) + try: + return await asyncio.shield(start) + except asyncio.CancelledError: + with contextlib.suppress(Exception): + proc = await start + _kill_cli(proc) + await _reap(proc) + raise + + +async def _reap(proc) -> None: + """Wait for the killed process, but not for one that left its tree and still holds its pipes open.""" + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(proc.wait(), _PROCESS_CLEANUP_TIMEOUT_S) + + +def _descendants(pid: int) -> list[int]: + """The processes under ``pid``, by parent, read from /proc; none where there is no /proc.""" + children: dict[int, list[int]] = {} + for entry in os.listdir("/proc") if os.path.isdir("/proc") else (): + if not entry.isdigit(): + continue + try: + with open(f"/proc/{entry}/stat") as stat: + parent = int(stat.read().rsplit(")", 1)[1].split()[1]) + except (OSError, ValueError, IndexError): + continue + children.setdefault(parent, []).append(int(entry)) + found, pending = [], [pid] + while pending: + for child in children.get(pending.pop(), ()): + found.append(child) + pending.append(child) + return found + + +async def run_hermes( + prompt: str, + context_id: str, + session_id: str | None = None, + model: str | None = None, + mcp_servers: dict[str, dict] | None = None, + system_prompt: str | None = None, + output_format: dict | None = None, + timeout_s: int = 600, + project_id: str | None = None, + task_id: str | None = None, + role: str | None = None, + effort: str | None = None, + skills_dir: Path = HERMES_ROOT / "agentenv-skills", + event_callback: Callable[[dict], None] | None = None, + workspace: str = "/app", +) -> HermesResult: + """Run one Hermes turn in ``workspace``. ``event_callback`` gets each event as the plugin writes it: exactly the + events of the returned trajectory, in order.""" + model = model or DEFAULT_MODEL + attribution = _attribution_headers(project_id, task_id) + if attribution: + _record_attribution(project_id, task_id) + schema = None + if output_format is not None: + schema = validate_output_format(output_format) + prompt += ( + "\n\nReturn only a JSON object matching this schema, with no Markdown or surrounding text: " + + json.dumps(schema, separators=(",", ":")) + ) + config = build_config(model, mcp_servers, provider_headers(os.environ, attribution), role, skills_dir) + # To its end even for a cancelled turn: a snapshot may archive the profile as soon as the turn lets go. + home = await in_thread_to_its_end(prepare_profile, context_id, config, system_prompt) + + with tempfile.TemporaryDirectory(prefix="hermes-turn-") as scratch: + prompt_file, usage_file = Path(scratch) / "prompt.txt", Path(scratch) / "usage.json" + prompt_file.write_text(prompt) + cmd = [HERMES_BIN, "agentenv", "turn", "--prompt-file", str(prompt_file), "--provider", PROVIDER, + "--model", model, "--usage-file", str(usage_file)] + if session_id: + cmd += ["--resume", session_id] + if effort: + cmd += ["--reasoning", effort] + logger.info(f"Running hermes (session={session_id}, model={model}, effort={effort}, ctx={context_id[:12]}, " + f"mcp={list((mcp_servers or {}).keys())})") + if attribution: + logger.info(f"hermes attribution headers: {attribution}") + + read_end, write_end = os.pipe() + try: + proc = await _started(asyncio.create_subprocess_exec( + *cmd, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=workspace, + env={**os.environ, "HERMES_HOME": str(home), _EVENTS_FD_ENV: str(write_end)}, + pass_fds=(write_end,), + start_new_session=True, + )) + except BaseException: + os.close(read_end) + raise + finally: + os.close(write_end) + _RUNNING.add(proc) + events = os.fdopen(read_end, "rb", buffering=0) + parser = _EventParser(event_callback) + try: + stdout, stderr = await asyncio.wait_for(_run_turn(proc, events, parser.on_line), timeout=timeout_s) + except asyncio.CancelledError: + _kill_cli(proc) + await _reap(proc) + raise + except asyncio.TimeoutError: + _kill_cli(proc) + await _reap(proc) + return HermesResult(response="", tool_call_count=parser.tool_call_count, session_id=parser.session_id, + trajectory=parser.trajectory, error=f"hermes timed out after {timeout_s}s") + except Exception: + # A failure while reading, such as an event_callback that raises, must not leave the turn running. + _kill_cli(proc) + await _reap(proc) + raise + finally: + _RUNNING.discard(proc) + events.close() + usage = _read_usage(usage_file) + + result = parser.result(stdout, stderr, proc.returncode, usage) + if schema is not None and result.error is None: + try: + structured_output = json.loads(result.response) + Draft202012Validator(schema).validate(structured_output) + except (json.JSONDecodeError, ValidationError) as exc: + result.error = f"Hermes returned a response that does not match output_format: {exc}" + result.error_code = "invalid_output_format" + else: + result.structured_output = structured_output + return result + + +def _read_usage(path: Path) -> dict: + try: + usage = json.loads(path.read_text()) + except (OSError, ValueError): + return {} + return usage if isinstance(usage, dict) else {} + + +async def _run_turn(proc, events, on_line: Callable[[str], None]) -> tuple[bytes, bytes]: + """Read the event pipe line by line as the turn writes it, and the process's stdout and stderr, until it exits. + The pipe is read on past the exit only briefly: a process the turn left behind may still hold it open.""" + loop = asyncio.get_running_loop() + reader = asyncio.StreamReader(limit=2**30) + transport, _ = await loop.connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), events) + + async def events() -> None: + while line := await reader.readline(): + on_line(line.decode(errors="replace")) + + reading = asyncio.create_task(events()) + try: + stdout, stderr = await proc.communicate() + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(asyncio.shield(reading), _EVENTS_DRAIN_S) + if reading.done(): + reading.result() + return stdout, stderr + finally: + reading.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await reading + transport.close() + + +class _EventParser: + """Parse the plugin's JSON event lines one at a time. + + Every event carries its hook's name as ``type``. ``pre_tool_call`` events are the turn's tool calls, + ``post_llm_call`` carries the final answer, and any event's ``session_id`` is the session. The whole event list is + the trajectory, and each event is handed to ``event_callback`` as it is parsed. + """ + + def __init__(self, event_callback: Callable[[dict], None] | None = None) -> None: + self._event_callback = event_callback + self.trajectory: list[dict] = [] + self.tool_calls: set[str] = set() + self.session_id: str | None = None + self.final: str | None = None + self.errors: list[str] = [] + self.error_code: str | None = None + + @property + def tool_call_count(self) -> int: + return len(self.tool_calls) + + def on_line(self, line: str) -> None: + line = line.strip() + if not line: + return + try: + event = json.loads(line) + except json.JSONDecodeError: + logger.debug(f"hermes emitted a non-JSON event line: {line[:200]}") + return + if not isinstance(event, dict): + return + self.trajectory.append(event) + if isinstance(event.get("session_id"), str) and event["session_id"]: + self.session_id = event["session_id"] + kind = event.get("type") + if kind == "pre_tool_call": + self.tool_calls.add(str(event.get("tool_call_id") or f"tool-{len(self.tool_calls)}")) + elif kind == "post_llm_call" and isinstance(event.get("assistant_response"), str): + self.final = event["assistant_response"] + elif kind == "api_request_error": + message = str(event.get("error") or event.get("reason") or "model request failed") + if message not in self.errors: + self.errors.append(message) + if event.get("status_code") is not None: + self.error_code = str(event["status_code"]) + if self._event_callback is not None: + self._event_callback(event) + + def result(self, stdout: bytes, stderr: bytes, returncode: int | None, usage: dict) -> HermesResult: + session_id = usage.get("session_id") or self.session_id + response = (self.final if self.final is not None else stdout.decode(errors="replace")).strip() + failed = returncode != 0 or usage.get("failed") or (usage and usage.get("completed") is False) + err: str | None = None + if failed: + stderr_text = stderr.decode(errors="replace").strip() + err = ("\n".join(self.errors) or usage.get("failure") or stderr_text + or (f"hermes stopped without completing the turn ({usage['turn_exit_reason']})" + if usage.get("turn_exit_reason") else f"hermes exited with code {returncode}") + )[:ERROR_MESSAGE_LIMIT] + logger.error(f"hermes turn failed: {err}") + return HermesResult( + response=response, + tool_call_count=self.tool_call_count, + session_id=session_id, + trajectory=self.trajectory, + error=err, + error_code=self.error_code if err else None, + usage=usage, + ) diff --git a/src/agent_env/agents/agents/hermes/hermes_kit.py b/src/agent_env/agents/agents/hermes/hermes_kit.py new file mode 100644 index 000000000..13892f638 --- /dev/null +++ b/src/agent_env/agents/agents/hermes/hermes_kit.py @@ -0,0 +1,101 @@ +"""The pieces of agent plumbing this agent shares in shape with other CLI agents, copied from the Claude Code agent's +``kit.py``: durable session references, skill installs, FastAPI-style error bodies and work a cancel can't abandon.""" + +from __future__ import annotations + +import asyncio +import logging +import shutil +from collections.abc import Awaitable, Callable +from pathlib import Path +from typing import TypeVar + +from agentenv_protocol.a2a_agent import BundleSkillRequest, InlineSkillRequest, SkillAddResponse +from agentenv_protocol.transfers import download +from starlette.applications import Starlette +from starlette.exceptions import HTTPException +from starlette.requests import Request +from starlette.responses import JSONResponse, Response + +logger = logging.getLogger(__name__) + +_REFS_ATTR = "_durable_session_refs" +_T = TypeVar("_T") + + +async def to_its_end(work: Awaitable[_T]) -> _T: + """``work`` awaited to its end even when the caller is cancelled, so nothing it does outlives the locks the caller + holds; the cancel is raised once it has ended.""" + task = asyncio.ensure_future(work) + cancelled = False + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError: + cancelled = True + except Exception: + break + if cancelled: + raise asyncio.CancelledError from (None if task.cancelled() else task.exception()) + return task.result() + + +async def in_thread_to_its_end(func: Callable[..., _T], /, *args: object) -> _T: + """``func(*args)`` in a thread, off the event loop, awaited to its end as :func:`to_its_end` awaits work.""" + return await to_its_end(asyncio.to_thread(func, *args)) + + +class DurableSessionRefs: + """Mixin, listed before ``AgentEnvAgent``, that keeps every context's session reference for the life of the + process: the framework's own record is a bounded LRU, so an idle context would otherwise start a fresh CLI + session.""" + + def session_ref_for_context(self, context_id: str) -> str | None: + return self.__dict__.get(_REFS_ATTR, {}).get(context_id) or super().session_ref_for_context(context_id) + + def set_session_ref_for_context(self, context_id: str, session_ref: str) -> None: + super().set_session_ref_for_context(context_id, session_ref) + self.__dict__.setdefault(_REFS_ATTR, {})[context_id] = session_ref + + +_NO_BODY_STATUSES = frozenset({204, 205, 304}) + + +async def _detail_body(_request: Request, exc: HTTPException) -> Response: + 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) + + +def json_error_bodies(app: Starlette) -> Starlette: + """Answer every ``HTTPException`` with a ``{"detail": ...}`` body, as agent-env reads errors.""" + app.add_exception_handler(HTTPException, _detail_body) + return app + + +def _validate_skill_name(name: str) -> None: + if name != Path(name).name or ".." in name or "/" in name or "\\" in name: + raise HTTPException(status_code=400, detail=f"Invalid skill name '{name}': must not contain path separators or '..'") + + +async def install_skill(request: InlineSkillRequest | BundleSkillRequest, skills_dir: Path) -> SkillAddResponse: + """Write ``skills_dir//``: the inline ``SKILL.md``, or every bundle file through its read grant. A failed + install leaves no directory behind.""" + _validate_skill_name(request.name) + skill_dir = skills_dir / request.name + if skill_dir.exists(): + raise HTTPException(status_code=409, detail=f"Skill '{request.name}' is already registered") + skill_dir.mkdir(parents=True) + try: + if isinstance(request, BundleSkillRequest): + for item in request.skill_bundle.files: + await download(item.object, skill_dir / item.path) + else: + if not request.skill_md or not request.skill_md.strip(): + raise HTTPException(status_code=400, detail="Skill content must not be empty") + (skill_dir / "SKILL.md").write_text(request.skill_md) + except BaseException: + shutil.rmtree(skill_dir, ignore_errors=True) + raise + logger.info(f"Registered skill '{request.name}' at {skill_dir}") + return SkillAddResponse(name=request.name) diff --git a/src/agent_env/agents/agents/hermes/requirements.txt b/src/agent_env/agents/agents/hermes/requirements.txt new file mode 100644 index 000000000..30c7f3a44 --- /dev/null +++ b/src/agent_env/agents/agents/hermes/requirements.txt @@ -0,0 +1,5 @@ +# The image installs the newest agentenv-framework-protocol at or above this floor; a release doesn't pin it, so the +# image is built again only when this folder changes. The floor is the first that serves peer-agents/v1's tools, which +# the agent advertises without handlers of its own; raise it when the agent needs a newer protocol. +agentenv-framework-protocol[agent]>=0.1.351 +jsonschema>=4.18,<5 diff --git a/src/agent_env/agents/aliases.toml b/src/agent_env/agents/aliases.toml index dfa477d8b..8f47f156e 100644 --- a/src/agent_env/agents/aliases.toml +++ b/src/agent_env/agents/aliases.toml @@ -3,6 +3,7 @@ claude-code = ["claude-code"] codex = ["codex"] gemini = ["gemini"] +hermes = ["hermes"] opencode = ["opencode"] pi = ["pi"] openhands = ["openhands"] diff --git a/src/agent_env/agents/tasks/hermes-smoke.json b/src/agent_env/agents/tasks/hermes-smoke.json new file mode 100644 index 000000000..4ce9bedbb --- /dev/null +++ b/src/agent_env/agents/tasks/hermes-smoke.json @@ -0,0 +1,7 @@ +[ + {"id": "deploy", "type": "deploy_agent", "a2a_agent_id": "hermes", "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/unit/agents/hermes/__init__.py b/tst/unit/agents/hermes/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tst/unit/agents/hermes/agent.py b/tst/unit/agents/hermes/agent.py new file mode 100644 index 000000000..774e1061e --- /dev/null +++ b/tst/unit/agents/hermes/agent.py @@ -0,0 +1,20 @@ +"""The Hermes agent's modules, imported from its folder as its image runs them: by their own names, from the folder on +the path.""" + +import importlib +import importlib.util +import sys +from pathlib import Path + +import agent_env + +AGENT_DIR = Path(agent_env.__file__).parent / "agents" / "agents" / "hermes" +sys.path.insert(0, str(AGENT_DIR)) +hermes_bridge = importlib.import_module("hermes_bridge") +hermes_kit = importlib.import_module("hermes_kit") +# By its path: another agent's a2a_server, imported by the same name, may be loaded already. +_spec = importlib.util.spec_from_file_location("hermes_a2a_server", AGENT_DIR / "a2a_server.py") +a2a_server = importlib.util.module_from_spec(_spec) +# Registered first: pydantic resolves its config's annotations through the module's entry. +sys.modules[_spec.name] = a2a_server +_spec.loader.exec_module(a2a_server) diff --git a/tst/unit/agents/hermes/conftest.py b/tst/unit/agents/hermes/conftest.py new file mode 100644 index 000000000..a10588f60 --- /dev/null +++ b/tst/unit/agents/hermes/conftest.py @@ -0,0 +1,86 @@ +"""A fresh agent per test, driven over ASGI the way agent-env reaches it, its workspace, Hermes profiles and skills +under the test's own folder.""" + +from __future__ import annotations + +import os +import stat +import sys +import uuid +from types import SimpleNamespace + +import httpx +import pytest + +from tst.unit.agents.hermes.agent import a2a_server, hermes_bridge + +FAKE_HERMES = os.path.join(os.path.dirname(__file__), "fake_hermes.py") + + +class Harness: + """One fresh agent and its app, reached the way agent-env reaches it: over HTTP.""" + + def __init__(self, agent: a2a_server.HermesAgent) -> None: + self.agent = agent + self.app = a2a_server.build_app(agent) + + @property + def card(self) -> dict: + return self.app.state.agentenv_a2a.card.model_dump(mode="json", exclude_none=True, by_alias=True) + + def client(self) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.ASGITransport(app=self.app), base_url="http://hermes") + + @staticmethod + async def send(client: httpx.AsyncClient, parts: str | list[dict], context_id: str = "ctx-1", *, + blocking: bool = True) -> dict: + """message/send; the task, terminal unless ``blocking`` is False.""" + if isinstance(parts, str): + parts = [{"kind": "text", "text": parts}] + message = {"kind": "message", "role": "user", "messageId": uuid.uuid4().hex, "contextId": context_id, + "parts": parts} + response = await client.post("/a2a", json={ + "jsonrpc": "2.0", "id": 1, "method": "message/send", + "params": {"message": message, "configuration": {"blocking": blocking}}, + }) + response.raise_for_status() + return response.json()["result"] + + +def result(**fields) -> SimpleNamespace: + """What run_hermes returns, with a finished turn's defaults.""" + defaults = {"response": "done", "error": None, "error_code": None, "session_id": "s-1", "trajectory": [], + "tool_call_count": 0, "structured_output": None, "usage": {}} + return SimpleNamespace(**{**defaults, **fields}) + + +@pytest.fixture(autouse=True) +def folders(monkeypatch, tmp_path): + """The agent's workspace, Hermes profiles and skills in the test's folder.""" + workspace = tmp_path / "workspace" + workspace.mkdir() + monkeypatch.setattr(a2a_server, "WORKSPACE_DIR", workspace) + monkeypatch.setattr(a2a_server, "INPUTS_DIR", workspace / "inputs") + monkeypatch.setattr(a2a_server, "SKILLS_DIR", tmp_path / "skills") + monkeypatch.setattr(hermes_bridge, "HERMES_ROOT", tmp_path / "hermes") + return workspace + + +@pytest.fixture +def fake_hermes(monkeypatch, tmp_path): + """A `hermes` that runs fake_hermes.py; what it was run with is in the returned record file.""" + launcher = tmp_path / "bin" / "hermes" + launcher.parent.mkdir() + launcher.write_text(f"#!/bin/sh\nexec {sys.executable} {FAKE_HERMES} \"$@\"\n") + launcher.chmod(launcher.stat().st_mode | stat.S_IXUSR) + record = tmp_path / "record.json" + monkeypatch.setattr(hermes_bridge, "HERMES_BIN", str(launcher)) + monkeypatch.setenv("FAKE_HERMES_RECORD", str(record)) + monkeypatch.setenv("FAKE_HERMES_PIDS", str(tmp_path / "pids")) + return record + + +@pytest.fixture +def hermes() -> Harness: + """The agent as the image serves it.""" + return Harness(a2a_server.HermesAgent()) diff --git a/tst/unit/agents/hermes/fake_hermes.py b/tst/unit/agents/hermes/fake_hermes.py new file mode 100644 index 000000000..e289ae55e --- /dev/null +++ b/tst/unit/agents/hermes/fake_hermes.py @@ -0,0 +1,48 @@ +"""A scripted `hermes agentenv turn` for the bridge tests, run as the real one is: a process of its own. It writes the +events, usage report and stdout a turn would, as FAKE_HERMES_MODE says: ``ok`` answers, ``fail`` fails the model call, +``hang`` starts a tool and never finishes. Its arguments, environment and cwd go to FAKE_HERMES_RECORD.""" +import json +import os +import subprocess +import sys +import time + +args = sys.argv[1:] +option = {name: args[i + 1] for i, name in enumerate(args) if name.startswith("--") and i + 1 < len(args)} +mode = os.environ.get("FAKE_HERMES_MODE", "ok") +events = int(os.environ["AGENTENV_EVENTS_FD"]) +session = option.get("--resume") or "20260101_000000_abcdef" +with open(os.environ["FAKE_HERMES_RECORD"], "w") as record: + json.dump({"args": args, "cwd": os.getcwd(), "home": os.environ.get("HERMES_HOME"), + "prompt": open(option["--prompt-file"]).read()}, record) + + +def emit(kind, **fields): + os.write(events, (json.dumps({"type": kind, "timestamp": 0, "session_id": session, **fields}) + "\n").encode()) + + +def report(**fields): + with open(option["--usage-file"], "w") as usage: + json.dump({"session_id": session, "input_tokens": 120, "output_tokens": 8, "total_tokens": 128, + "estimated_cost_usd": 0.002, "api_calls": 2, **fields}, usage) + + +emit("on_session_start", model=option.get("--model")) +emit("pre_tool_call", tool_call_id="call_1", tool_name="terminal", args={"command": "echo $((6 * 7))"}) +if mode == "hang": + tool = subprocess.Popen(["sleep", "600"]) + with open(os.environ["FAKE_HERMES_PIDS"], "w") as pids: + pids.write(f"{os.getpid()} {tool.pid}") + time.sleep(600) +emit("post_tool_call", tool_call_id="call_1", tool_name="terminal", result='{"output": "42"}', status="ok") +if mode == "fail": + emit("api_request_error", status_code=401, error="HTTP 401: invalid key") + emit("on_session_end", completed=False, failed=True, interrupted=False) + report(completed=False, failed=True) + sys.stderr.write("hermes: the model rejected the key\n") + sys.exit(1) +answer = os.environ.get("FAKE_HERMES_ANSWER", "42") +emit("post_llm_call", assistant_response=answer) +emit("on_session_end", completed=True, failed=False, interrupted=False) +report(completed=True, failed=False) +print(answer) diff --git a/tst/unit/agents/hermes/test_card_contract.py b/tst/unit/agents/hermes/test_card_contract.py new file mode 100644 index 000000000..e89f706eb --- /dev/null +++ b/tst/unit/agents/hermes/test_card_contract.py @@ -0,0 +1,75 @@ +"""The Agent Card agent-env negotiates against: the wrapper's identity and extensions, and the card as agent-env reads +it, by importing the server.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys + +import pytest + +from tst.unit.agents.hermes.agent import AGENT_DIR, hermes_bridge + + +def _methods(card: dict) -> dict[str, list[str]]: + return {ext["uri"]: sorted((ext.get("params") or {}).get("methods") or {}) for ext in card["capabilities"]["extensions"]} + + +def test_identity(hermes): + card = hermes.card + assert {key: card[key] for key in ("name", "version", "defaultInputModes", "defaultOutputModes")} == { + "name": "Hermes Agent", + "version": "1.0.0", + "defaultInputModes": ["text", "image/png", "image/jpeg", "image/gif", "image/webp", "application/pdf"], + "defaultOutputModes": ["text"], + } + assert [skill["id"] for skill in card["skills"]] == ["hermes"] + + +def test_the_extensions_are_the_wrappers(hermes): + assert _methods(hermes.card) == { + "urn:agentenv:agent-config/v1": ["get", "set"], + "urn:agentenv:mcp-config/v1": ["add", "list"], + "urn:agentenv:trajectory/v1": ["get"], + "urn:agentenv:attribution-probe/v1": ["probe"], + "urn:agentenv:peer-agents/v1": ["list", "set"], + "urn:agentenv:skill-config/v1": ["add", "list"], + "urn:agentenv:snapshot/v1": ["load", "save"], + "urn:agentenv:staging/v1": [], + } + + +@pytest.mark.asyncio +async def test_the_attribution_probe_reads_back_what_the_last_turn_sent(monkeypatch, hermes): + monkeypatch.setattr(hermes_bridge, "_last_attribution", {"project_id": None, "task_id": None, + "last_seen_at_utc": None}) + hermes_bridge._record_attribution("p-1", "t-1") + + async with hermes.client() as client: + probed = (await client.post("/ext/attribution-probe")).json() + + assert probed["last_seen_attribution"] == {"project_id": "p-1", "task_id": "t-1"} + assert probed["last_seen_at_utc"] + + +def test_the_card_is_read_by_importing_agent_card(): + """agent-env imports AGENT_CARD from the image with no agent environment, and parses its stdout as the card.""" + script = "import json; from a2a_server import AGENT_CARD; print(json.dumps(AGENT_CARD.model_dump(mode='json')))" + env = {key: value for key, value in os.environ.items() if not key.startswith("LITELLM_")} + out = subprocess.run([sys.executable, "-c", script], cwd=AGENT_DIR, env=env, capture_output=True, text=True, + check=True) + + assert json.loads(out.stdout)["name"] == "Hermes Agent" + + +@pytest.mark.asyncio +async def test_errors_are_answered_with_a_detail_body(hermes): + async with hermes.client() as client: + response = await client.post("/ext/snapshot", json={"context_id": "never-ran", "objects": { + "trajectory": {"media_type": "application/octet-stream", "max_bytes": 1024, + "write": {"kind": "http-put", "url": "https://objects.example/o", + "expires_at": "2099-01-01T00:00:00Z"}}}}) + + assert (response.status_code, response.json()) == (404, {"detail": "No active session for context_id"}) diff --git a/tst/unit/agents/hermes/test_held_work.py b/tst/unit/agents/hermes/test_held_work.py new file mode 100644 index 000000000..d9fd565ea --- /dev/null +++ b/tst/unit/agents/hermes/test_held_work.py @@ -0,0 +1,151 @@ +"""Work a turn or snapshot starts ends before it lets go of what it holds: a cancelled save keeps the workspace until its +archive is written, a cancelled turn's PDF render is killed and reaped, attachments wait for a turn slot, and a load +whose workspace swap fails leaves the context's previous conversation and session in place.""" + +from __future__ import annotations + +import asyncio +import os +import shutil +import stat +import threading + +import pytest +from agentenv_protocol.a2a_agent import Uploaded + +from tst.unit.agents.hermes.agent import a2a_server, hermes_bridge +from tst.unit.agents.hermes.conftest import Harness +from tst.unit.agents.hermes.test_message_contract import _GRANT, _WRITE, Turns, _until + +_READ = {"media_type": "application/octet-stream", "max_bytes": 2**20, "size_bytes": 1, + "read": {"kind": "http-get", **_GRANT}} +_PDF = {"kind": "file", "file": {"name": "report.pdf", "mimeType": "application/pdf", "bytes": "JVBERi0="}} + + +@pytest.fixture +def turns(monkeypatch) -> Turns: + stand_in = Turns() + monkeypatch.setattr(a2a_server, "run_hermes", stand_in) + return stand_in + + +def _alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + +@pytest.mark.asyncio +async def test_a_cancelled_save_holds_the_workspace_until_its_archive_ends(monkeypatch, hermes, turns): + started, finish = threading.Event(), threading.Event() + + def archive(home, session_id, target): + started.set() + finish.wait(10) + target.write_bytes(b"archive") + + monkeypatch.setattr(a2a_server, "_archive_profile", archive) + async with hermes.client() as client: + await hermes.send(client, "go") + save = asyncio.create_task( + client.post("/ext/snapshot", json={"context_id": "ctx-1", "objects": {"trajectory": _WRITE}})) + await _until(started.is_set) + save.cancel() + turn = asyncio.create_task(hermes.send(client, "next")) + await asyncio.sleep(0.3) + turned_during_archive = len(turns.calls) > 1 + finish.set() + await asyncio.gather(save, return_exceptions=True) + await turn + + assert not turned_during_archive + assert len(turns.calls) == 2 + + +@pytest.mark.asyncio +async def test_a_cancelled_turns_pdf_render_is_killed_and_reaped(monkeypatch, tmp_path, hermes, turns): + pids = tmp_path / "pdftoppm.pid" + fake = tmp_path / "pdfbin" / "pdftoppm" + fake.parent.mkdir() + fake.write_text(f'#!/bin/sh\necho $$ > "{pids}"\nexec sleep 600\n') + fake.chmod(fake.stat().st_mode | stat.S_IXUSR) + monkeypatch.setenv("PATH", f"{fake.parent}{os.pathsep}{os.environ['PATH']}") + async with hermes.client() as client: + task = await hermes.send(client, [{"kind": "text", "text": "read it"}, _PDF], blocking=False) + await _until(lambda: pids.exists() and pids.read_text().strip()) + pid = int(pids.read_text()) + await client.post("/a2a", json={"jsonrpc": "2.0", "id": 2, "method": "tasks/cancel", + "params": {"id": task["id"]}}) + try: + await _until(lambda: not _alive(pid), seconds=5) + finally: + if _alive(pid): + os.kill(pid, 9) + + assert turns.calls == [] + + +@pytest.mark.asyncio +async def test_attachments_wait_for_a_turn_slot(monkeypatch, folders, turns): + monkeypatch.setenv("HERMES_MAX_CONCURRENT_TURNS", "1") + hermes = Harness(a2a_server.HermesAgent()) + turns.release = asyncio.Event() + part = {"kind": "file", "file": {"name": "notes.txt", "mimeType": "text/plain", "bytes": "bm90ZXM="}} + async with hermes.client() as client: + first = asyncio.create_task(hermes.send(client, [{"kind": "text", "text": "one"}], "ctx-a")) + await _until(lambda: len(turns.calls) == 1) + second = asyncio.create_task(hermes.send(client, [{"kind": "text", "text": "two"}, part], "ctx-b")) + await asyncio.sleep(0.3) + written_while_waiting = list((folders / "inputs").rglob("notes.txt")) if (folders / "inputs").exists() else [] + turns.release.set() + await asyncio.gather(first, second) + + assert written_while_waiting == [] + assert len(turns.calls) == 2 + + +@pytest.mark.asyncio +async def test_a_load_whose_workspace_swap_fails_keeps_the_previous_conversation( + monkeypatch, hermes, turns, folders, tmp_path): + objects = tmp_path / "objects" + objects.mkdir() + + async def upload(target, source): + name = "workspace" if "workspace" in str(source) else "trajectory" + shutil.copyfile(source, objects / name) + return Uploaded(size_bytes=(objects / name).stat().st_size) + + async def download(source, destination): + shutil.copyfile(objects / ("workspace" if "workspace" in str(destination) else "trajectory"), destination) + + monkeypatch.setattr(a2a_server, "upload", upload) + monkeypatch.setattr(a2a_server, "download", download) + async with hermes.client() as client: + await hermes.send(client, "go", "ctx-src") + source = hermes_bridge.profile_dir("ctx-src") + source.mkdir(parents=True) + (source / "state.db").write_bytes(b"the snapshot's conversation") + (folders / "result.txt").write_text("snapshot") + saved = await client.post("/ext/snapshot", json={"context_id": "ctx-src", "objects": { + "trajectory": _WRITE, "workspace": _WRITE}}) + assert saved.status_code == 200, saved.text + await hermes.send(client, "go", "ctx-dst") + target = hermes_bridge.profile_dir("ctx-dst") + target.mkdir(parents=True, exist_ok=True) + (target / "state.db").write_bytes(b"the previous conversation") + (folders / "result.txt").write_text("previous") + + def full_disk(source, destination): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(a2a_server.shutil, "move", full_disk) + failed = await client.put("/ext/snapshot", json={"target_context_id": "ctx-dst", "objects": { + "trajectory": _READ, "workspace": _READ}}) + monkeypatch.undo() + + assert failed.status_code == 500 + assert (target / "state.db").read_bytes() == b"the previous conversation" + assert (folders / "result.txt").read_text() == "previous" + assert hermes.agent.session_ref_for_context("ctx-dst") == "s-ctx-dst" diff --git a/tst/unit/agents/hermes/test_hermes_bridge.py b/tst/unit/agents/hermes/test_hermes_bridge.py new file mode 100644 index 000000000..d8c335865 --- /dev/null +++ b/tst/unit/agents/hermes/test_hermes_bridge.py @@ -0,0 +1,155 @@ +"""One turn as the bridge runs it, against a scripted `hermes` in a process of its own: the profile it writes for the +turn, the events it reads off the pipe as they come, the result it makes of them, and a turn that times out or is +cancelled killing Hermes and every tool it started.""" + +from __future__ import annotations + +import asyncio +import json +import os +import time + +import pytest + +from tst.unit.agents.hermes.agent import hermes_bridge + +_MCP = {"env": {"url": "https://mcp.example/mcp", "headers": {"X-Key": "k", "agentenv-role": "registered"}}, + "peer_agents": {"url": "http://127.0.0.1:8000/ext/peer-agents/mcp", "headers": None}} + + +def _gone(pid: int, within: float = 5.0) -> bool: + deadline = time.monotonic() + within + while True: + try: + with open(f"/proc/{pid}/stat") as stat: + state = stat.read().rsplit(")", 1)[1].split()[0] + except FileNotFoundError: + return True + if state == "Z": + return True + if time.monotonic() > deadline: + return False + time.sleep(0.05) + + +@pytest.mark.asyncio +async def test_a_turn_runs_in_its_contexts_profile_with_the_endpoint_and_servers_it_was_given( + monkeypatch, fake_hermes, folders, tmp_path): + monkeypatch.setenv("LITELLM_BASE_URL", "http://litellm.example") + monkeypatch.setenv("LITELLM_EXTRA_HEADERS", json.dumps({"X-Team": "evals", "x-litellm-tags": "taskId:mine"})) + events: list[dict] = [] + + result = await hermes_bridge.run_hermes( + prompt="what is 6*7?", context_id="ctx-a", model="provider/model", mcp_servers=_MCP, system_prompt="be brief", + project_id="p-1", task_id="t-1", role="solver", effort="high", skills_dir=tmp_path / "skills", + event_callback=events.append, workspace=str(folders)) + + home = hermes_bridge.profile_dir("ctx-a") + ran = json.loads(fake_hermes.read_text()) + assert (ran["home"], ran["cwd"], ran["prompt"]) == (str(home), str(folders), "what is 6*7?") + assert ran["args"][:2] == ["agentenv", "turn"] + assert "--resume" not in ran["args"] + assert ran["args"][ran["args"].index("--model") + 1] == "provider/model" + assert ran["args"][ran["args"].index("--provider") + 1] == "litellm" + assert ran["args"][ran["args"].index("--reasoning") + 1] == "high" + + config = json.loads((home / "config.yaml").read_text()) + assert config["model"] == {"provider": "litellm", "default": "provider/model"} + assert config["providers"]["litellm"] == { + "api": "http://litellm.example/v1", "key_env": "LITELLM_API_KEY", "transport": "chat_completions", + # [model.headers], then the run's attribution, its tags joining the configured ones they don't name. + "extra_headers": {"x-team": "evals", "x-litellm-tags": "taskId:mine,projectId:p-1", + "x-litellm-customer-id": "p-1"}} + assert config["mcp_servers"] == { + "env": {"url": "https://mcp.example/mcp", "headers": {"X-Key": "k", "AgentEnv-Role": "solver"}}, + "peer_agents": {"url": "http://127.0.0.1:8000/ext/peer-agents/mcp", "headers": {"AgentEnv-Role": "solver"}}} + assert config["skills"] == {"external_dirs": [str(tmp_path / "skills")]} + assert config["plugins"] == {"enabled": ["agentenv"]} + assert (home / "plugins" / "agentenv").resolve() == hermes_bridge.PLUGIN_DIR + assert (home / "SOUL.md").read_text() == "be brief" + + assert [event["type"] for event in events] == ["on_session_start", "pre_tool_call", "post_tool_call", + "post_llm_call", "on_session_end"] + assert result.trajectory == events + assert (result.response, result.error, result.session_id, result.tool_call_count) == ( + "42", None, "20260101_000000_abcdef", 1) + assert result.usage["total_tokens"] == 128 + + +@pytest.mark.asyncio +async def test_a_later_turn_resumes_its_session_and_a_turn_without_a_system_prompt_gets_hermes_own_back( + fake_hermes, folders): + await hermes_bridge.run_hermes(prompt="one", context_id="ctx-b", system_prompt="be brief", workspace=str(folders)) + home = hermes_bridge.profile_dir("ctx-b") + + result = await hermes_bridge.run_hermes(prompt="two", context_id="ctx-b", session_id="s-earlier", + workspace=str(folders)) + + ran = json.loads(fake_hermes.read_text()) + assert ran["args"][ran["args"].index("--resume") + 1] == "s-earlier" + assert result.session_id == "s-earlier" + assert not (home / "SOUL.md").exists() + (home / "SOUL.md").write_text("Hermes' own") + await hermes_bridge.run_hermes(prompt="three", context_id="ctx-b", workspace=str(folders)) + assert (home / "SOUL.md").read_text() == "Hermes' own" + + +@pytest.mark.asyncio +async def test_a_failed_model_call_fails_the_turn_with_what_the_provider_said(monkeypatch, fake_hermes, folders): + monkeypatch.setenv("FAKE_HERMES_MODE", "fail") + + result = await hermes_bridge.run_hermes(prompt="go", context_id="ctx-c", workspace=str(folders)) + + assert (result.error, result.error_code) == ("HTTP 401: invalid key", "401") + assert result.session_id == "20260101_000000_abcdef" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("ending", ["timeout", "cancel"]) +async def test_a_turn_that_times_out_or_is_cancelled_kills_hermes_and_the_tools_it_started( + monkeypatch, fake_hermes, folders, tmp_path, ending): + monkeypatch.setenv("FAKE_HERMES_MODE", "hang") + pids, events = tmp_path / "pids", [] + turn = asyncio.create_task(hermes_bridge.run_hermes( + prompt="go", context_id="ctx-d", timeout_s=3 if ending == "timeout" else 600, event_callback=events.append, + workspace=str(folders))) + async with asyncio.timeout(10): + while not (pids.exists() and len(pids.read_text().split()) == 2): + await asyncio.sleep(0.02) + hermes, tool = (int(pid) for pid in pids.read_text().split()) + + if ending == "cancel": + turn.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(turn, 10) + else: + result = await asyncio.wait_for(turn, 10) + assert result.error == "hermes timed out after 3s" + # The session and the events Hermes wrote before it stopped are kept. + assert result.session_id == "20260101_000000_abcdef" + assert [event["type"] for event in result.trajectory] == ["on_session_start", "pre_tool_call"] + + assert _gone(hermes) and _gone(tool) + assert [event["type"] for event in events][:2] == ["on_session_start", "pre_tool_call"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("answer", "valid"), [('{"answer": 42}', True), ("42, I think", False)]) +async def test_an_output_format_is_asked_for_and_checked(monkeypatch, fake_hermes, folders, answer, valid): + monkeypatch.setenv("FAKE_HERMES_ANSWER", answer) + schema = {"type": "object", "properties": {"answer": {"type": "integer"}}, "required": ["answer"]} + + result = await hermes_bridge.run_hermes(prompt="go", context_id="ctx-e", workspace=str(folders), + output_format={"type": "json_schema", "schema": schema}) + + assert "Return only a JSON object matching this schema" in json.loads(fake_hermes.read_text())["prompt"] + if valid: + assert (result.error, result.structured_output) == (None, {"answer": 42}) + else: + assert result.error_code == "invalid_output_format" + + +def test_the_event_pipe_is_not_left_open_in_the_server(fake_hermes, folders): + before = len(os.listdir("/proc/self/fd")) + asyncio.run(hermes_bridge.run_hermes(prompt="go", context_id="ctx-f", workspace=str(folders))) + assert len(os.listdir("/proc/self/fd")) == before diff --git a/tst/unit/agents/hermes/test_message_contract.py b/tst/unit/agents/hermes/test_message_contract.py new file mode 100644 index 000000000..beb946a3b --- /dev/null +++ b/tst/unit/agents/hermes/test_message_contract.py @@ -0,0 +1,242 @@ +"""What a turn hands Hermes and when it runs: turns of different contexts at once and a context's own one at a time, the +session a context is bound to when its turn starts, the task's config, files and MCP servers, a failed turn, and a +snapshot of a context's conversation and workspace restored into another context.""" + +from __future__ import annotations + +import asyncio +import os +import shutil +import stat +import sys +from pathlib import Path + +import pytest +from agentenv_protocol.a2a_agent import Uploaded + +from tst.unit.agents.hermes.agent import a2a_server, hermes_bridge, hermes_kit +from tst.unit.agents.hermes.conftest import result + + +class Turns: + """Stand in for run_hermes: each call's arguments, held until ``release`` when it is set.""" + + def __init__(self) -> None: + self.calls: list[dict] = [] + self.release: asyncio.Event | None = None + self.answer = lambda kwargs: result(session_id=kwargs["session_id"] or f"s-{kwargs['context_id']}") + + async def __call__(self, **kwargs): + self.calls.append(kwargs) + if self.release is not None: + await self.release.wait() + return self.answer(kwargs) + + +@pytest.fixture +def turns(monkeypatch) -> Turns: + stand_in = Turns() + monkeypatch.setattr(a2a_server, "run_hermes", stand_in) + return stand_in + + +async def _until(condition, seconds: float = 5) -> None: + async with asyncio.timeout(seconds): + while not condition(): + await asyncio.sleep(0.01) + + +@pytest.mark.asyncio +async def test_turns_of_different_contexts_run_at_once(hermes, turns): + turns.release = asyncio.Event() + async with hermes.client() as client: + sent = [asyncio.create_task(hermes.send(client, "go", context_id)) for context_id in ("ctx-a", "ctx-b")] + await _until(lambda: len(turns.calls) == 2) + turns.release.set() + tasks = await asyncio.gather(*sent) + + assert {call["context_id"] for call in turns.calls} == {"ctx-a", "ctx-b"} + assert [task["status"]["state"] for task in tasks] == ["completed", "completed"] + + +@pytest.mark.asyncio +async def test_a_contexts_turns_run_one_at_a_time_each_resuming_the_last(hermes, turns): + turns.release = asyncio.Event() + async with hermes.client() as client: + first = asyncio.create_task(hermes.send(client, "one")) + await _until(lambda: len(turns.calls) == 1) + second = asyncio.create_task(hermes.send(client, "two")) + await asyncio.sleep(0.1) + running_together = len(turns.calls) + turns.release.set() + await asyncio.gather(first, second) + + assert running_together == 1 + assert [call["session_id"] for call in turns.calls] == [None, "s-ctx-1"] + + +@pytest.mark.asyncio +async def test_the_turns_config_servers_and_workspace_reach_hermes(monkeypatch, hermes, turns, folders): + monkeypatch.setenv("A2A_PORT", "8000") + config = {"model": "provider/model", "effort": "high", "system_prompt": "be brief", "project_id": "p-1", + "task_id": "t-1", "role": "solver", "timeout_seconds": 42} + async with hermes.client() as client: + assert (await client.post("/ext/agent-config", json=config)).status_code == 200 + await client.post("/ext/mcp-config", json={"name": "env", "url": "https://mcp.example/mcp", + "headers": {"X-Key": "k"}}) + task = await hermes.send(client, "go") + + (call,) = turns.calls + assert {key: call[key] for key in ("prompt", "model", "effort", "system_prompt", "project_id", "task_id", "role", + "timeout_s", "workspace", "skills_dir")} == { + "prompt": "go", "model": "provider/model", "effort": "high", "system_prompt": "be brief", + "project_id": "p-1", "task_id": "t-1", "role": "solver", "timeout_s": 42, "workspace": str(folders), + "skills_dir": a2a_server.SKILLS_DIR} + assert call["mcp_servers"]["env"] == {"url": "https://mcp.example/mcp", "headers": {"X-Key": "k"}} + assert call["mcp_servers"]["peer_agents"]["url"] == "http://127.0.0.1:8000/ext/peer-agents/mcp" + assert task["status"]["state"] == "completed" + + +@pytest.mark.asyncio +async def test_a_file_part_is_written_under_the_task_and_named_in_the_prompt(hermes, turns, folders): + parts = [{"kind": "text", "text": "describe"}, + {"kind": "file", "file": {"name": "../a b.png", "mimeType": "image/png", "bytes": "cG5n"}}] + async with hermes.client() as client: + task = await hermes.send(client, parts) + + attached = folders / "inputs" / task["id"] / "a_b.png" + assert attached.read_bytes() == b"png" + (call,) = turns.calls + assert call["prompt"].startswith("describe\n\nAttached files") + assert f"- {attached}" in call["prompt"] + + +@pytest.mark.asyncio +async def test_two_pdfs_whose_names_share_a_stem_keep_their_own_pages(monkeypatch, tmp_path, hermes, turns): + # A pdftoppm that renders as many pages as the file's text says, each naming the file it came from. + fake = tmp_path / "pdfbin" / "pdftoppm" + fake.parent.mkdir() + fake.write_text(f"#!{sys.executable}\nimport pathlib, sys\nsource, prefix = sys.argv[-2:]\n" + "for page in range(1, int(pathlib.Path(source).read_text()) + 1):\n" + " pathlib.Path(f'{prefix}-{page}.png').write_text(pathlib.Path(source).name)\n") + fake.chmod(fake.stat().st_mode | stat.S_IXUSR) + monkeypatch.setenv("PATH", f"{fake.parent}{os.pathsep}{os.environ['PATH']}") + parts = [{"kind": "text", "text": "compare"}, + {"kind": "file", "file": {"name": "report.pdf", "mimeType": "application/pdf", "bytes": "Mw=="}}, + {"kind": "file", "file": {"name": "report", "mimeType": "application/pdf", "bytes": "MQ=="}}] + async with hermes.client() as client: + await hermes.send(client, parts) + + (call,) = turns.calls + listed = [Path(line[2:]) for line in call["prompt"].splitlines() if line.startswith("- ")] + pages = [path for path in listed if path.suffix == ".png"] + assert [path.read_text() for path in pages] == ["report.pdf"] * 3 + ["report"] + assert len(set(pages)) == 4 + + +@pytest.mark.asyncio +async def test_a_pdf_with_a_name_near_the_filesystems_limit_renders(monkeypatch, tmp_path, hermes, turns): + fake = tmp_path / "pdfbin" / "pdftoppm" + fake.parent.mkdir() + fake.write_text(f"#!{sys.executable}\nimport pathlib, sys\npathlib.Path(sys.argv[-1] + '-1.png').write_text('p')\n") + fake.chmod(fake.stat().st_mode | stat.S_IXUSR) + monkeypatch.setenv("PATH", f"{fake.parent}{os.pathsep}{os.environ['PATH']}") + name = "r" * 238 + ".pdf" + parts = [{"kind": "text", "text": "read"}, + {"kind": "file", "file": {"name": name, "mimeType": "application/pdf", "bytes": "MQ=="}}] + async with hermes.client() as client: + task = await hermes.send(client, parts) + + assert task["status"]["state"] == "completed" + (call,) = turns.calls + assert sum(line.endswith(".png") for line in call["prompt"].splitlines()) == 1 + + +@pytest.mark.asyncio +async def test_a_failed_turn_is_a_failed_task_that_keeps_its_session(hermes, turns): + turns.answer = lambda kwargs: ( + result(response="", error="HTTP 401: invalid key", error_code="401", session_id="s-started") + if not kwargs["session_id"] else result(session_id=kwargs["session_id"])) + async with hermes.client() as client: + failed = await hermes.send(client, "one") + await hermes.send(client, "two") + + assert failed["status"]["state"] == "failed" + assert "HTTP 401: invalid key" in str(failed["status"]["message"]) + assert [call["session_id"] for call in turns.calls] == [None, "s-started"] + + +@pytest.mark.asyncio +async def test_a_first_turn_cancelled_after_hermes_named_its_session_keeps_it_for_the_next(monkeypatch, hermes): + calls: list[dict] = [] + started = asyncio.Event() + + async def run(**kwargs): + calls.append(kwargs) + if len(calls) == 1: + kwargs["event_callback"]({"type": "on_session_start", "session_id": "s-started"}) + started.set() + await asyncio.Event().wait() + return result(session_id=kwargs["session_id"]) + + monkeypatch.setattr(a2a_server, "run_hermes", run) + async with hermes.client() as client: + task = await hermes.send(client, "one", blocking=False) + await asyncio.wait_for(started.wait(), 5) + await client.post("/a2a", json={"jsonrpc": "2.0", "id": 2, "method": "tasks/cancel", "params": {"id": task["id"]}}) + await _until(lambda: hermes.agent.session_ref_for_context("ctx-1") == "s-started") + await hermes.send(client, "two") + + assert [call["session_id"] for call in calls] == [None, "s-started"] + + +def test_session_references_outlive_the_frameworks_bounded_record(): + assert issubclass(a2a_server.HermesAgent, hermes_kit.DurableSessionRefs) + + +_GRANT = {"expires_at": "2099-01-01T00:00:00Z", "url": "https://objects.example/o"} +_WRITE = {"media_type": "application/octet-stream", "max_bytes": 2**20, "write": {"kind": "http-put", **_GRANT}} + + +@pytest.mark.asyncio +async def test_a_snapshot_restores_the_conversation_and_workspace_into_another_context( + monkeypatch, hermes, turns, folders, tmp_path): + objects = tmp_path / "objects" + objects.mkdir() + + async def upload(target, source): + name = "workspace" if "workspace" in str(source) else "trajectory" + shutil.copyfile(source, objects / name) + return Uploaded(size_bytes=(objects / name).stat().st_size) + + async def download(source, destination): + shutil.copyfile(objects / ("workspace" if "workspace" in str(destination) else "trajectory"), destination) + + monkeypatch.setattr(a2a_server, "upload", upload) + monkeypatch.setattr(a2a_server, "download", download) + async with hermes.client() as client: + await hermes.send(client, "go", "ctx-src") + source = hermes_bridge.profile_dir("ctx-src") + source.mkdir(parents=True) + (source / "state.db").write_bytes(b"the conversation") + (source / "memories").mkdir() + (source / "memories" / "MEMORY.md").write_text("a memory") + (source / "config.yaml").write_text("rewritten each turn") + (folders / "result.txt").write_text("42") + saved = await client.post("/ext/snapshot", json={"context_id": "ctx-src", "objects": { + "trajectory": _WRITE, "workspace": _WRITE}}) + (folders / "result.txt").write_text("changed since") + read = {"media_type": "application/octet-stream", "max_bytes": 2**20, "size_bytes": 1, + "read": {"kind": "http-get", **_GRANT}} + loaded = await client.put("/ext/snapshot", json={"target_context_id": "ctx-dst", "objects": { + "trajectory": read, "workspace": read}}) + await hermes.send(client, "go on", "ctx-dst") + + assert saved.status_code == 200, saved.text + assert loaded.status_code == 200, loaded.text + target = hermes_bridge.profile_dir("ctx-dst") + assert (target / "state.db").read_bytes() == b"the conversation" + assert (target / "memories" / "MEMORY.md").read_text() == "a memory" + assert not (target / "config.yaml").exists() + assert (folders / "result.txt").read_text() == "42" + assert turns.calls[-1]["session_id"] == "s-ctx-src" diff --git a/tst/unit/bundle/shipped_bundles_test.py b/tst/unit/bundle/shipped_bundles_test.py index 8e1d687af..315f76591 100644 --- a/tst/unit/bundle/shipped_bundles_test.py +++ b/tst/unit/bundle/shipped_bundles_test.py @@ -47,7 +47,7 @@ def test_the_shipped_aliases_are_agent_envs_alone_and_claude_code_names_the_defa "openhands": "@local/agentenv-framework/agents/openhands", "codex": "@local/agentenv-framework/agents/codex", "gemini": "@local/agentenv-framework/agents/gemini", "opencode": "@local/agentenv-framework/agents/opencode", "letta-code": "@local/agentenv-framework/agents/letta-code", - "pydantic-ai": "@local/agentenv-framework/agents/pydantic-ai"} + "pydantic-ai": "@local/agentenv-framework/agents/pydantic-ai", "hermes": "@local/agentenv-framework/agents/hermes"} assert table.target("claude-code").target_id == get_config().get_default_a2a_agent_id() diff --git a/tst/unit/test_infra_images_script.py b/tst/unit/test_infra_images_script.py index b43427653..57685695a 100644 --- a/tst/unit/test_infra_images_script.py +++ b/tst/unit/test_infra_images_script.py @@ -63,9 +63,9 @@ def test_plan_only_lists_the_infra_images_or_the_agents_images(script, monkeypat assert planned == {"infra": [image.name for image in bootstrap.PUBLISHED_IMAGES if image.name not in after], "agents": after} assert "agentenv-gateway" in planned["infra"] and bundles == [ - "agentenv-claude-code", "agentenv-codex", "agentenv-gemini", "agentenv-letta-code", "agentenv-opencode", - "agentenv-openhands", "agentenv-pi", "agentenv-proxy", "agentenv-pydantic-ai", "agentenv-mail-mcp", - "agentenv-ubuntu-24.04"] + "agentenv-claude-code", "agentenv-codex", "agentenv-gemini", "agentenv-hermes", "agentenv-letta-code", + "agentenv-opencode", "agentenv-openhands", "agentenv-pi", "agentenv-proxy", "agentenv-pydantic-ai", + "agentenv-mail-mcp", "agentenv-ubuntu-24.04"] def test_plan_missing_lists_the_images_the_registry_lacks_or_refuses_to_show(script, monkeypatch, capsys):