From e74959a08ad137b8928a0f14f1819ad51b139889 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:24:24 -0700 Subject: [PATCH 1/4] feat(agents): add a Hermes Agent agent Adds agents/hermes/, Nous Research's Hermes Agent behind the A2A agent framework: each turn is one in-process `hermes agentenv turn` (a subcommand of the agent's own Hermes plugin) in the context's own Hermes profile, its plugin hook events streamed as the trajectory. Registers the image, its alias, a smoke task, and the release/dependabot entries. --- .github/dependabot.yml | 1 + .github/scripts/infra_build.py | 5 +- CONTRIBUTING.md | 2 +- src/agent_env/agents/README.md | 14 +- .../agents/agents/hermes/.dockerignore | 2 + src/agent_env/agents/agents/hermes/Dockerfile | 34 ++ src/agent_env/agents/agents/hermes/README.md | 49 ++ .../agents/agents/hermes/a2a_server.py | 530 ++++++++++++++++++ .../agents/hermes/agentenv_plugin/__init__.py | 75 +++ .../agents/hermes/agentenv_plugin/plugin.yaml | 3 + .../agents/agents/hermes/entrypoint.sh | 4 + .../agents/agents/hermes/hermes_bridge.py | 487 ++++++++++++++++ .../agents/agents/hermes/hermes_kit.py | 75 +++ .../agents/agents/hermes/requirements.txt | 5 + src/agent_env/agents/aliases.toml | 1 + src/agent_env/agents/tasks/hermes-smoke.json | 7 + tst/unit/agents/hermes/__init__.py | 0 tst/unit/agents/hermes/agent.py | 20 + tst/unit/agents/hermes/conftest.py | 86 +++ tst/unit/agents/hermes/fake_hermes.py | 48 ++ tst/unit/agents/hermes/test_card_contract.py | 75 +++ tst/unit/agents/hermes/test_hermes_bridge.py | 155 +++++ .../agents/hermes/test_message_contract.py | 197 +++++++ tst/unit/bundle/shipped_bundles_test.py | 2 +- tst/unit/test_infra_images_script.py | 6 +- 25 files changed, 1874 insertions(+), 9 deletions(-) create mode 100644 src/agent_env/agents/agents/hermes/.dockerignore create mode 100644 src/agent_env/agents/agents/hermes/Dockerfile create mode 100644 src/agent_env/agents/agents/hermes/README.md create mode 100644 src/agent_env/agents/agents/hermes/a2a_server.py create mode 100644 src/agent_env/agents/agents/hermes/agentenv_plugin/__init__.py create mode 100644 src/agent_env/agents/agents/hermes/agentenv_plugin/plugin.yaml create mode 100644 src/agent_env/agents/agents/hermes/entrypoint.sh create mode 100644 src/agent_env/agents/agents/hermes/hermes_bridge.py create mode 100644 src/agent_env/agents/agents/hermes/hermes_kit.py create mode 100644 src/agent_env/agents/agents/hermes/requirements.txt create mode 100644 src/agent_env/agents/tasks/hermes-smoke.json create mode 100644 tst/unit/agents/hermes/__init__.py create mode 100644 tst/unit/agents/hermes/agent.py create mode 100644 tst/unit/agents/hermes/conftest.py create mode 100644 tst/unit/agents/hermes/fake_hermes.py create mode 100644 tst/unit/agents/hermes/test_card_contract.py create mode 100644 tst/unit/agents/hermes/test_hermes_bridge.py create mode 100644 tst/unit/agents/hermes/test_message_contract.py 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..e06e88d28 --- /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. + +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, so a failed load leaves the workspace as it was. 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..7d2c925bc --- /dev/null +++ b/src/agent_env/agents/agents/hermes/a2a_server.py @@ -0,0 +1,530 @@ +"""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 json +import logging +import os +import re +import shutil +import subprocess +import tarfile +import tempfile +import uuid +from collections.abc import 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, install_skill, json_error_bodies + +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}" + + +def _download_uri_to(uri: str, target: Path) -> None: + parsed = urlparse(uri) + if parsed.scheme in ("http", "https"): + try: + with httpx.Client(timeout=_FILE_DOWNLOAD_TIMEOUT_S, follow_redirects=True) as client: + response = client.get(uri) + response.raise_for_status() + target.write_bytes(response.content) + except httpx.HTTPError as exc: + raise FilePartMaterializeError("HTTP file download failed") from exc + else: + raise FilePartMaterializeError(f"unsupported file URI scheme {parsed.scheme!r}") + + +def _materialize_and_render(part: FilePart, task_id: str) -> list[str]: + 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: + target.write_bytes(base64.b64decode(part.bytes, validate=True)) + except (binascii.Error, ValueError, OSError) as exc: + raise FilePartMaterializeError(f"invalid file attachment {name!r}") from exc + else: + _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. + prefix = str(target.with_suffix("")) + "_page" + try: + subprocess.run( + ["pdftoppm", "-png", "-r", str(_PDF_RENDER_DPI), str(target), prefix], + check=True, capture_output=True, timeout=_PDF_RENDER_TIMEOUT_S, + ) + except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): + logger.warning("PDF rendering failed for %s; passing the PDF alone", target) + return [str(target)] + pages = sorted(directory.glob(f"{Path(prefix).name}*.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 _restore_profile(archive: Path, home: Path) -> str: + """Replace the context's profile with an archived one; 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() + shutil.rmtree(home, ignore_errors=True) + staged.rename(home) + return session_id + except BaseException: + shutil.rmtree(staged, ignore_errors=True) + raise + + +@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) + try: + attachments = [] + for part in request.parts: + if isinstance(part, FilePart): + attachments.extend(await asyncio.to_thread(_materialize_and_render, part, request.task_id)) + except FilePartMaterializeError as exc: + return TaskResult.failure("filepart_error", str(exc), error_type="infra_error") + try: + async with self._turn_slots: + 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") + await asyncio.to_thread(_archive_profile, profile_dir(request.context_id), session_id, conversation) + if request.objects.workspace is not None: + await asyncio.to_thread( + subprocess.run, ["tar", "-czf", str(workspace), "-C", str(WORKSPACE_DIR), "."], + check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + ) + 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 asyncio.to_thread( + subprocess.run, ["tar", "-xzf", str(archive), "-C", str(staged)], + check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + ) + 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) + try: + session_id = await asyncio.to_thread(_restore_profile, conversation, home) + except (tarfile.TarError, OSError, ValueError) as exc: + raise HTTPException(status_code=400, detail=f"Unreadable Hermes snapshot: {exc}") from exc + if staged is not None: + await asyncio.to_thread(_replace_workspace, staged) + self.set_session_ref_for_context(target_context_id, session_id) + 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) -> None: + 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)) + 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..e22797b1d --- /dev/null +++ b/src/agent_env/agents/agents/hermes/hermes_bridge.py @@ -0,0 +1,487 @@ +"""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 + +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) + home = await asyncio.to_thread(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..d38eb0634 --- /dev/null +++ b/src/agent_env/agents/agents/hermes/hermes_kit.py @@ -0,0 +1,75 @@ +"""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 and FastAPI-style error bodies.""" + +from __future__ import annotations + +import logging +import shutil +from pathlib import Path + +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" + + +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_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..5605cd4a6 --- /dev/null +++ b/tst/unit/agents/hermes/test_message_contract.py @@ -0,0 +1,197 @@ +"""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 shutil + +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_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): From fa5a90fbc5dc94ad08050f59ed4d763c9f783adf Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 29 Oct 2026 09:07:06 -0700 Subject: [PATCH 2/4] fix(agents): roll back Hermes snapshot loads, finish cancelled work before letting go, and hold attachments to the turn limit - A snapshot load stages the archived profile beside the context's, then swaps the workspace and the profile in one thread, with the profile swap inside the workspace's, and binds the session, in one step a cancel can't abandon. A failed swap leaves the previous profile, session and workspace. - Thread work a cancel would have left running (the snapshot archives, tar, the profile write before a turn) now runs to its end before the caller lets go of the workspace or the context's lock. A PDF render is an async pdftoppm, killed and reaped on timeout or cancel, and http(s) attachments stream to disk with an async client. - A task's attachments are downloaded and rendered inside the turn's slot, so HERMES_MAX_CONCURRENT_TURNS bounds them too. Co-Authored-By: Claude Opus 5.5 --- src/agent_env/agents/agents/hermes/README.md | 4 +- .../agents/agents/hermes/a2a_server.py | 168 ++++++++++++------ .../agents/agents/hermes/hermes_bridge.py | 5 +- .../agents/agents/hermes/hermes_kit.py | 28 ++- tst/unit/agents/hermes/test_held_work.py | 151 ++++++++++++++++ 5 files changed, 299 insertions(+), 57 deletions(-) create mode 100644 tst/unit/agents/hermes/test_held_work.py diff --git a/src/agent_env/agents/agents/hermes/README.md b/src/agent_env/agents/agents/hermes/README.md index e06e88d28..ebaf5e09c 100644 --- a/src/agent_env/agents/agents/hermes/README.md +++ b/src/agent_env/agents/agents/hermes/README.md @@ -14,7 +14,7 @@ first task starts a Hermes session, and each later task resumes it with `--resum 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 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: @@ -39,7 +39,7 @@ the model no one is there to answer. | `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, so a failed load leaves the workspace as it was. No changelog | +| `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 | diff --git a/src/agent_env/agents/agents/hermes/a2a_server.py b/src/agent_env/agents/agents/hermes/a2a_server.py index 7d2c925bc..50c0b6a60 100644 --- a/src/agent_env/agents/agents/hermes/a2a_server.py +++ b/src/agent_env/agents/agents/hermes/a2a_server.py @@ -10,6 +10,8 @@ import asyncio import base64 import binascii +import contextlib +import functools import json import logging import os @@ -19,7 +21,7 @@ import tarfile import tempfile import uuid -from collections.abc import Mapping +from collections.abc import Callable, Mapping from contextlib import asynccontextmanager from pathlib import Path from typing import Any @@ -69,7 +71,7 @@ run_hermes, validate_output_format, ) -from hermes_kit import DurableSessionRefs, install_skill, json_error_bodies +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__) @@ -116,21 +118,51 @@ def _safe_name(candidate: str | None, fallback_ext: str = "") -> str: return candidate or f"{uuid.uuid4().hex}{fallback_ext}" -def _download_uri_to(uri: str, target: Path) -> None: +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 in ("http", "https"): - try: - with httpx.Client(timeout=_FILE_DOWNLOAD_TIMEOUT_S, follow_redirects=True) as client: - response = client.get(uri) - response.raise_for_status() - target.write_bytes(response.content) - except httpx.HTTPError as exc: - raise FilePartMaterializeError("HTTP file download failed") from exc - else: + 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()) -def _materialize_and_render(part: FilePart, task_id: str) -> list[str]: + +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) @@ -139,21 +171,16 @@ def _materialize_and_render(part: FilePart, task_id: str) -> list[str]: target = directory / f"{uuid.uuid4().hex[:8]}-{name}" if part.bytes is not None: try: - target.write_bytes(base64.b64decode(part.bytes, validate=True)) + 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: - _download_uri_to(part.uri, target) + 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. prefix = str(target.with_suffix("")) + "_page" - try: - subprocess.run( - ["pdftoppm", "-png", "-r", str(_PDF_RENDER_DPI), str(target), prefix], - check=True, capture_output=True, timeout=_PDF_RENDER_TIMEOUT_S, - ) - except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): + if not await _render_pdf(target, prefix): logger.warning("PDF rendering failed for %s; passing the PDF alone", target) return [str(target)] pages = sorted(directory.glob(f"{Path(prefix).name}*.png")) @@ -259,8 +286,9 @@ def _archive_profile(home: Path, session_id: str, archive: Path) -> None: (home / _SNAPSHOT_SESSION).unlink(missing_ok=True) -def _restore_profile(archive: Path, home: Path) -> str: - """Replace the context's profile with an archived one; the archived session id.""" +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: @@ -269,14 +297,33 @@ def _restore_profile(archive: Path, home: Path) -> str: if not isinstance(session_id, str) or not session_id: raise ValueError("the snapshot names no Hermes session") (staged / _SNAPSHOT_SESSION).unlink() - shutil.rmtree(home, ignore_errors=True) - staged.rename(home) - return session_id + 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", @@ -330,15 +377,17 @@ def record(event: dict) -> None: 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) - try: - attachments = [] - for part in request.parts: - if isinstance(part, FilePart): - attachments.extend(await asyncio.to_thread(_materialize_and_render, part, request.task_id)) - except FilePartMaterializeError as exc: - return TaskResult.failure("filepart_error", str(exc), error_type="infra_error") - try: - async with self._turn_slots: + + # 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, @@ -356,10 +405,10 @@ def record(event: dict) -> None: 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") + 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) @@ -418,12 +467,10 @@ async def save_snapshot(self, request: ObjectSnapshotSaveRequest) -> dict[str, A 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") - await asyncio.to_thread(_archive_profile, profile_dir(request.context_id), session_id, conversation) + # 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 asyncio.to_thread( - subprocess.run, ["tar", "-czf", str(workspace), "-C", str(WORKSPACE_DIR), "."], - check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, - ) + 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") @@ -441,20 +488,31 @@ async def load_snapshot(self, request: ObjectSnapshotLoadRequest) -> dict[str, A staged = Path(directory) / "workspace" staged.mkdir() await download(request.objects.workspace, archive) - await asyncio.to_thread( - subprocess.run, ["tar", "-xzf", str(archive), "-C", str(staged)], - check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, - ) + 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: - session_id = await asyncio.to_thread(_restore_profile, conversation, home) + 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 - if staged is not None: - await asyncio.to_thread(_replace_workspace, staged) - self.set_session_ref_for_context(target_context_id, session_id) + 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} @@ -479,7 +537,9 @@ def count(name: str) -> int | None: ) -def _replace_workspace(staged: Path) -> None: +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" @@ -497,6 +557,8 @@ def _replace_workspace(staged: Path) -> None: 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(): diff --git a/src/agent_env/agents/agents/hermes/hermes_bridge.py b/src/agent_env/agents/agents/hermes/hermes_bridge.py index e22797b1d..c9c0c4b4d 100644 --- a/src/agent_env/agents/agents/hermes/hermes_bridge.py +++ b/src/agent_env/agents/agents/hermes/hermes_bridge.py @@ -27,6 +27,8 @@ 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. @@ -311,7 +313,8 @@ async def run_hermes( + json.dumps(schema, separators=(",", ":")) ) config = build_config(model, mcp_servers, provider_headers(os.environ, attribution), role, skills_dir) - home = await asyncio.to_thread(prepare_profile, context_id, config, system_prompt) + # 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" diff --git a/src/agent_env/agents/agents/hermes/hermes_kit.py b/src/agent_env/agents/agents/hermes/hermes_kit.py index d38eb0634..13892f638 100644 --- a/src/agent_env/agents/agents/hermes/hermes_kit.py +++ b/src/agent_env/agents/agents/hermes/hermes_kit.py @@ -1,11 +1,14 @@ """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 and FastAPI-style error bodies.""" +``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 @@ -17,6 +20,29 @@ 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: 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" From c6473cded5de0fcbf251fe30657a86634302f06e Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 29 Oct 2026 10:29:36 -0700 Subject: [PATCH 3/4] fix(agents): render each Hermes PDF attachment's pages in a folder of its own Two PDFs whose names share a stem (report.pdf and report, or report.PDF on a case-sensitive filesystem) rendered their pages under one prefix, so the second overwrote the first's and listed its leftovers. Each PDF's pages now go in a folder of their own. Co-Authored-By: Claude Opus 5.5 --- .../agents/agents/hermes/a2a_server.py | 9 ++++--- .../agents/hermes/test_message_contract.py | 27 +++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/src/agent_env/agents/agents/hermes/a2a_server.py b/src/agent_env/agents/agents/hermes/a2a_server.py index 50c0b6a60..5c93c8871 100644 --- a/src/agent_env/agents/agents/hermes/a2a_server.py +++ b/src/agent_env/agents/agents/hermes/a2a_server.py @@ -178,12 +178,13 @@ async def _materialize_and_render(part: FilePart, task_id: str) -> list[str]: 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. - prefix = str(target.with_suffix("")) + "_page" - if not await _render_pdf(target, prefix): + # 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. + pages_dir = Path(tempfile.mkdtemp(prefix=f"{target.name}-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(directory.glob(f"{Path(prefix).name}*.png")) + pages = sorted(pages_dir.glob("page*.png")) return [str(target), *(str(page) for page in pages)] diff --git a/tst/unit/agents/hermes/test_message_contract.py b/tst/unit/agents/hermes/test_message_contract.py index 5605cd4a6..288da17cf 100644 --- a/tst/unit/agents/hermes/test_message_contract.py +++ b/tst/unit/agents/hermes/test_message_contract.py @@ -5,7 +5,11 @@ 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 @@ -107,6 +111,29 @@ async def test_a_file_part_is_written_under_the_task_and_named_in_the_prompt(her 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_failed_turn_is_a_failed_task_that_keeps_its_session(hermes, turns): turns.answer = lambda kwargs: ( From 0d8ad84e01e45a3abe60bfb97f209737e7e6c066 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 29 Oct 2026 11:08:35 -0700 Subject: [PATCH 4/4] fix(agents): give Hermes PDF page folders a short fixed name A page folder named after the PDF passed the filesystem's 255-byte limit for a name near it, and the OSError failed the task. The folder now takes a fixed prefix; its random suffix keeps two PDFs apart. Co-Authored-By: Claude Opus 5.5 --- .../agents/agents/hermes/a2a_server.py | 5 +++-- .../agents/hermes/test_message_contract.py | 18 ++++++++++++++++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/src/agent_env/agents/agents/hermes/a2a_server.py b/src/agent_env/agents/agents/hermes/a2a_server.py index 5c93c8871..54092f277 100644 --- a/src/agent_env/agents/agents/hermes/a2a_server.py +++ b/src/agent_env/agents/agents/hermes/a2a_server.py @@ -179,8 +179,9 @@ async def _materialize_and_render(part: FilePart, task_id: str) -> list[str]: 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. - pages_dir = Path(tempfile.mkdtemp(prefix=f"{target.name}-pages-", dir=directory)) + # 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)] diff --git a/tst/unit/agents/hermes/test_message_contract.py b/tst/unit/agents/hermes/test_message_contract.py index 288da17cf..beb946a3b 100644 --- a/tst/unit/agents/hermes/test_message_contract.py +++ b/tst/unit/agents/hermes/test_message_contract.py @@ -134,6 +134,24 @@ async def test_two_pdfs_whose_names_share_a_stem_keep_their_own_pages(monkeypatc 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: (