From 2b567a39428deac735f314938d489496ddb44a84 Mon Sep 17 00:00:00 2001 From: William Loo Date: Fri, 13 Nov 2026 10:06:02 -0800 Subject: [PATCH 1/4] feat(agents): Claude Code follows the env's clock, in each prompt and after each tool call (FD-5420) The CLI states the date from its machine's clock ("Today's date is ..."), and no flag turns that off. While an env's clock is armed that date isn't the env's, and the model learns the env's only if it decides to call get_time. - Each prompt ends with the env's time in an note while an MCP server lists get_time, which the gateway does only while the clock is armed. - A PostToolUse hook (env_clock_hook.py) ends each tool call's result with a fresh reading, so a clock moved between prompts or running faster than real time never leaves the model on a stale date. It goes on the CLI's command line (--settings), never in the user settings a snapshot carries. - A failed or slow read leaves the note out and never fails the turn. Unarmed, nothing is added: the env runs on real time, the CLI's date. - The image and install.sh carry the two new modules; a test holds both to every module in the folder. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/agents/README.md | 6 + .../agents/agents/claude-code/Dockerfile | 2 +- .../agents/agents/claude-code/a2a_server.py | 8 + .../agents/claude-code/claude_bridge.py | 46 ++-- .../agents/agents/claude-code/env_clock.py | 131 +++++++++++ .../agents/claude-code/env_clock_hook.py | 41 ++++ .../agents/agents/claude-code/install.sh | 2 +- tst/unit/agents/claude_code/conftest.py | 9 + tst/unit/agents/claude_code/test_env_clock.py | 209 ++++++++++++++++++ .../agents/claude_code/test_image_contract.py | 10 + 10 files changed, 446 insertions(+), 18 deletions(-) create mode 100644 src/agent_env/agents/agents/claude-code/env_clock.py create mode 100644 src/agent_env/agents/agents/claude-code/env_clock_hook.py create mode 100644 tst/unit/agents/claude_code/test_env_clock.py diff --git a/src/agent_env/agents/README.md b/src/agent_env/agents/README.md index 62595d423..03fbb8cdf 100644 --- a/src/agent_env/agents/README.md +++ b/src/agent_env/agents/README.md @@ -29,6 +29,12 @@ turn off becomes `--disallowedTools`, which keeps the CLI's defaults for the res `claude_bridge.py`: every tool the pinned CLI has, though which it offers by default depends on the model. A test fails when the Dockerfile's `CLAUDE_CODE_VERSION` moves, until `ClaudeCodeTool` and `TOOLS_CLI_VERSION` move with it. +The CLI states the date from its machine's clock, which isn't the task env's while the env's clock is armed. So while +an MCP server lists the gateway's `get_time`, which it does only then, each prompt ends with the env's time in an +`` note, and a `PostToolUse` hook, given on the CLI's command line (`--settings`), ends each tool call's +result with a fresh reading (`env_clock.py`). A failed read leaves the note out and never fails the turn. Unarmed, +nothing is added: the env runs on real time, which is the CLI's date. + `install_agent` installs it too, into a container a task already runs or onto a `vm` sandbox's Linux host, with `install.sh` from its folder: Claude Code's native build for the target's CPU and libc, uv, and the agent's requirements in a Python of its own, so the target's Python is left alone. The CLI runs its commands in bash or zsh, so a target with diff --git a/src/agent_env/agents/agents/claude-code/Dockerfile b/src/agent_env/agents/agents/claude-code/Dockerfile index cbe95d6e6..7ec8b123f 100644 --- a/src/agent_env/agents/agents/claude-code/Dockerfile +++ b/src/agent_env/agents/agents/claude-code/Dockerfile @@ -23,7 +23,7 @@ WORKDIR /opt/agent COPY requirements.txt model-requirements.txt ./ RUN pip install --no-cache-dir -r requirements.txt -r model-requirements.txt # install.sh and run.sh are install/v1's: in the image, so they stay in its build context. -COPY a2a_server.py claude_bridge.py kit.py tar_changelog.py entrypoint.sh install.sh run.sh ./ +COPY a2a_server.py claude_bridge.py env_clock.py env_clock_hook.py kit.py tar_changelog.py entrypoint.sh install.sh run.sh ./ # The CLI's cwd, /app, and its state, fixed so neither follows the uid: Modal runs the image as root. RUN useradd --create-home --uid 10001 agent && mkdir -p /app /home/agent/.claude \ && chown agent:agent /app /home/agent/.claude && chmod 0755 entrypoint.sh diff --git a/src/agent_env/agents/agents/claude-code/a2a_server.py b/src/agent_env/agents/agents/claude-code/a2a_server.py index 7dc27625e..819d916e6 100644 --- a/src/agent_env/agents/agents/claude-code/a2a_server.py +++ b/src/agent_env/agents/agents/claude-code/a2a_server.py @@ -77,11 +77,13 @@ SESSION_JSONL_DIR, WORKSPACE_DIR, kill_running_clis, + mcp_server_entries, read_partial_session_trajectory, redact_session_records, run_claude, session_jsonl_path, ) +from env_clock import find_env_clock, prompt_note from kit import ( DurableSessionRefs, FilePartMaterializeError, @@ -472,6 +474,11 @@ async def _turn(self, request: TaskRequest[ClaudeCodeConfig], cfg: ClaudeCodeCon paths_block = "\n".join(attachment_paths) user_text = f"{user_text}\n\n{paths_block}" if user_text else paths_block + # The env's time, when its clock is armed: the date the CLI states is this machine's, not the env's. + env_clock = await find_env_clock(mcp_server_entries(request.mcp_servers, cfg.role)) + if env_clock is not None: + user_text = f"{user_text}\n\n{prompt_note(env_clock[1])}" if user_text else prompt_note(env_clock[1]) + # A turn that failed before the CLI wrote its transcript left no session to resume. resume = session_jsonl_path(session_id).exists() try: @@ -493,6 +500,7 @@ async def _turn(self, request: TaskRequest[ClaudeCodeConfig], cfg: ClaudeCodeCon max_thinking_tokens=cfg.max_thinking_tokens, builtin_tools=cfg.builtin_tools, event_callback=request.trajectory.append, + env_clock=env_clock[0] if env_clock is not None else None, ) except Exception as e: logger.error(f"claude CLI invocation raised: {e}", exc_info=True) diff --git a/src/agent_env/agents/agents/claude-code/claude_bridge.py b/src/agent_env/agents/agents/claude-code/claude_bridge.py index e8fcfc651..a8244bde0 100644 --- a/src/agent_env/agents/agents/claude-code/claude_bridge.py +++ b/src/agent_env/agents/agents/claude-code/claude_bridge.py @@ -20,6 +20,7 @@ from pydantic import BaseModel, ConfigDict, model_validator +from env_clock import ENV_CLOCK_SERVER_ENV, ClockServer, hook_settings from kit import in_thread_to_its_end logger = logging.getLogger(__name__) @@ -421,18 +422,20 @@ async def run_claude( max_thinking_tokens: int | None = None, builtin_tools: ClaudeCodeTools | None = None, event_callback: Callable[[dict], None] | None = None, + env_clock: ClockServer | None = None, ) -> ClaudeResult: """Run one turn, resuming the CLI session when a transient upstream failure cuts it short; a resume spends what is left of ``timeout_s``, so the turn never outlives its caller's deadline. ``CLAUDE_API_ERROR_RETRIES=0`` turns resumes off. ``event_callback`` gets each stdout event as the CLI prints it; an attempt's trailing `result` is - held until it's known not to be resumed.""" + held until it's known not to be resumed. ``env_clock``, the server of an armed env clock's get_time, has the CLI + add the env's time after each tool call (``env_clock``).""" deadline = time.monotonic() + timeout_s retries_left = _configured_api_error_retries() spawn = partial( _spawn_claude, model=model, allowed_tools=allowed_tools, mcp_servers=mcp_servers, system_prompt=system_prompt, output_format=output_format, project_id=project_id, task_id=task_id, role=role, effort=effort, max_turns=max_turns, max_thinking_tokens=max_thinking_tokens, - builtin_tools=builtin_tools, event_callback=event_callback, + builtin_tools=builtin_tools, event_callback=event_callback, env_clock=env_clock, ) attempt_prompt, attempt_resume, attempt_session = prompt, resume, session_id trajectory: list[dict] = [] @@ -610,10 +613,11 @@ def _with_1m_window(model: str) -> str: def _command(session_id: str, resume: bool, *, model: str | None, allowed_tools: list[str] | None, mcp_servers: dict[str, dict] | None, system_prompt: str | None, output_format: dict | None, role: str | None, effort: str | None, max_turns: int | None, - builtin_tools: ClaudeCodeTools | None = None) -> list[str]: + builtin_tools: ClaudeCodeTools | None = None, env_clock: bool = False) -> list[str]: """The CLI's argv. The prompt goes on stdin, since one argument over 128 KiB fails exec. ``builtin_tools`` selects the CLI's own tools: ``{"disable": "*"}`` offers none, ``{"only": [...]}`` just those, and ``{"disable": [...]}`` - the CLI's defaults but those; with any selection, its MCP servers are only ``mcp_servers``.""" + the CLI's defaults but those; with any selection, its MCP servers are only ``mcp_servers``. ``env_clock`` adds the + hook that ends each tool call's result with the env's time.""" cmd = ["claude", "-p", "--output-format", "stream-json", "--verbose", "--dangerously-skip-permissions", "--setting-sources", "user"] cmd += ["--resume", session_id] if resume else ["--session-id", session_id] @@ -640,18 +644,24 @@ def _command(session_id: str, resume: bool, *, model: str | None, allowed_tools: cmd += ["--disallowedTools", ",".join(builtin_tools.disable)] cmd.append("--strict-mcp-config") if mcp_servers: - def entry(info: dict) -> dict: - server: dict = {"type": "http", "url": info["url"]} - # The gateway filters an env's tools by the role header, which wins over a stored one. - headers = {**(info.get("headers") or {}), **({"AgentEnv-Role": role} if role else {})} - if headers: - server["headers"] = headers - return server - - cmd += ["--mcp-config", json.dumps({"mcpServers": {name: entry(info) for name, info in mcp_servers.items()}})] + def entry(server: ClockServer) -> dict: + return {"type": "http", "url": server.url, **({"headers": server.headers} if server.headers else {})} + + servers = mcp_server_entries(mcp_servers, role) + cmd += ["--mcp-config", json.dumps({"mcpServers": {name: entry(s) for name, s in servers.items()}})] + if env_clock: + cmd += ["--settings", hook_settings()] return cmd +def mcp_server_entries(mcp_servers: Mapping[str, dict], role: str | None) -> dict[str, ClockServer]: + """Each MCP server's url and the headers the CLI sends it.""" + # The gateway filters an env's tools by the role header, which wins over a stored one. + return {name: ClockServer(url=info["url"], + headers={**(info.get("headers") or {}), **({"AgentEnv-Role": role} if role else {})}) + for name, info in mcp_servers.items()} + + async def _spawn_claude( prompt: str, session_id: str, @@ -670,10 +680,15 @@ async def _spawn_claude( max_thinking_tokens: int | None = None, builtin_tools: ClaudeCodeTools | None = None, event_callback: Callable[[dict], None] | None = None, + env_clock: ClockServer | None = None, ) -> ClaudeResult: cmd = _command(session_id, resume, model=model, allowed_tools=allowed_tools, mcp_servers=mcp_servers, system_prompt=system_prompt, output_format=output_format, role=role, effort=effort, - max_turns=max_turns, builtin_tools=builtin_tools) + max_turns=max_turns, builtin_tools=builtin_tools, env_clock=env_clock is not None) + env = cli_env(os.environ, model=model, project_id=project_id, task_id=task_id, + max_thinking_tokens=max_thinking_tokens) + if env_clock is not None: + env[ENV_CLOCK_SERVER_ENV] = env_clock.to_env() logger.info(f"Running claude CLI (resume={resume}, session_id={session_id}, model={model})") WORKSPACE_DIR.mkdir(parents=True, exist_ok=True) proc = await _started(asyncio.create_subprocess_exec( @@ -681,8 +696,7 @@ async def _spawn_claude( stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - env=cli_env(os.environ, model=model, project_id=project_id, task_id=task_id, - max_thinking_tokens=max_thinking_tokens), + env=env, cwd=WORKSPACE_DIR, start_new_session=True, )) diff --git a/src/agent_env/agents/agents/claude-code/env_clock.py b/src/agent_env/agents/agents/claude-code/env_clock.py new file mode 100644 index 000000000..0eb1d1331 --- /dev/null +++ b/src/agent_env/agents/agents/claude-code/env_clock.py @@ -0,0 +1,131 @@ +"""The env's clock, read for the model. While agent-env's gateway runs an env's clock (urn:agentenv:clock/v1), it lists +``get_time`` on the env's MCP server, and the date the CLI states on its own ("Today's date is ...", from this +machine's clock) isn't the env's. So each prompt carries the env's time, and a PostToolUse hook (``env_clock_hook.py``) +adds a fresh reading after each tool call: a clock moved between prompts, or running faster than real time, never +leaves the model on a stale date. Unarmed, no server lists ``get_time`` and nothing is added: the env runs on real +time, which is the CLI's date.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import shlex +import sys +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path + +from mcp import ClientSession +from mcp.client.streamable_http import streamablehttp_client + +logger = logging.getLogger(__name__) + +ENV_CLOCK_TOOL_NAME = "get_time" +# The server whose get_time the hook reads, as {"url", "headers"} JSON in the CLI's environment while the clock is +# armed. The CLI's --mcp-config already carries the same url and headers on its command line. +ENV_CLOCK_SERVER_ENV = "AGENTENV_ENV_CLOCK_SERVER" +# How long a prompt waits for the clock before going without its note. +PROMPT_READ_TIMEOUT_S = 10.0 +# How long the hook waits for the clock, and how long the CLI lets the hook run, a little longer. +HOOK_READ_TIMEOUT_S = 5.0 +HOOK_TIMEOUT_S = 10 +HOOK_SCRIPT = Path(__file__).with_name("env_clock_hook.py") + + +@dataclass(frozen=True) +class ClockServer: + url: str + headers: dict[str, str] = field(default_factory=dict) + + def to_env(self) -> str: + return json.dumps({"url": self.url, "headers": self.headers}) + + @classmethod + def from_env(cls, value: str) -> ClockServer: + raw = json.loads(value) + return cls(url=raw["url"], headers=dict(raw.get("headers") or {})) + + +async def _read(server: ClockServer, *, find: bool) -> str | None: + """The env's current time from ``server``'s get_time; None when ``find`` and the server doesn't list it.""" + async with streamablehttp_client(server.url, headers=server.headers or None) as (read, write, _): + async with ClientSession(read, write) as session: + await session.initialize() + if find: + cursor, listed = None, False + while not listed: + page = await session.list_tools(cursor=cursor) + listed = any(t.name == ENV_CLOCK_TOOL_NAME for t in page.tools) + cursor = page.nextCursor + if not cursor: + break + if not listed: + return None + result = await session.call_tool(ENV_CLOCK_TOOL_NAME, {}) + if result.isError: + raise ValueError(f"{ENV_CLOCK_TOOL_NAME} failed: {_text(result)[:200]!r}") + if isinstance(result.structuredContent, dict) and result.structuredContent.get("current_time"): + return str(result.structuredContent["current_time"]) + text = _text(result) + current = json.loads(text).get("current_time") + if not current: + raise ValueError(f"no current_time in {text[:200]!r}") + return str(current) + + +def failure_reason(e: BaseException) -> str: + """What failed: the first error inside the MCP client's task groups, which say only how many there were.""" + while isinstance(e, BaseExceptionGroup) and e.exceptions: + e = e.exceptions[0] + return f"{type(e).__name__}: {e}" + + +def _text(result) -> str: + return "\n".join(c.text for c in (result.content or []) if getattr(c, "text", None)) + + +async def find_env_clock(servers: Mapping[str, ClockServer]) -> tuple[ClockServer, str] | None: + """The server listing get_time and the env's time it read, trying ``servers`` in order; None when none lists it or + the read fails, which is logged and never fails the prompt.""" + + async def search() -> tuple[ClockServer, str] | None: + for name, server in servers.items(): + try: + current = await _read(server, find=True) + except Exception as e: # one server's failure is no reason to skip the others + logger.warning(f"env clock: reading {ENV_CLOCK_TOOL_NAME} from MCP server {name!r} failed: " + f"{failure_reason(e)}") + continue + if current: + return server, current + return None + + try: + return await asyncio.wait_for(search(), timeout=PROMPT_READ_TIMEOUT_S) + except TimeoutError: + logger.warning(f"env clock: no reading within {PROMPT_READ_TIMEOUT_S:g} s; the prompt goes without it") + return None + + +async def read_env_clock(server: ClockServer, timeout_s: float = HOOK_READ_TIMEOUT_S) -> str: + return await asyncio.wait_for(_read(server, find=False), timeout=timeout_s) + + +def prompt_note(current: str) -> str: + # Same stance as get_time's own description: this machine's clock is not this environment's. + return (f"The current date and time in this environment is {current} (UTC). The clock on the machine " + f"you are running on, and the date you were given with it, may not match it, so use this value, or call " + f"{ENV_CLOCK_TOOL_NAME} for a fresh one before acting on a date.") + + +def turn_note(current: str) -> str: + return f"Environment time now: {current} (UTC)." + + +def hook_settings() -> str: + """The CLI's --settings for an armed clock: the hook after each tool call. On the command line rather than in the + user settings, so a snapshot of the CLI's state never carries it to an agent with no clock or no such script.""" + command = f"{shlex.quote(sys.executable)} {shlex.quote(str(HOOK_SCRIPT))}" + return json.dumps({"hooks": {"PostToolUse": [ + {"matcher": "", "hooks": [{"type": "command", "command": command, "timeout": HOOK_TIMEOUT_S}]}]}}) diff --git a/src/agent_env/agents/agents/claude-code/env_clock_hook.py b/src/agent_env/agents/agents/claude-code/env_clock_hook.py new file mode 100644 index 000000000..7af19974e --- /dev/null +++ b/src/agent_env/agents/agents/claude-code/env_clock_hook.py @@ -0,0 +1,41 @@ +"""The CLI's PostToolUse hook while the env's clock is armed: the env's time now, as context ending the tool's result +(``env_clock``). Any failure prints nothing, so a clock blip never fails a tool call; the model can still call +get_time itself.""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from env_clock import ( # noqa: E402 + ENV_CLOCK_SERVER_ENV, ENV_CLOCK_TOOL_NAME, ClockServer, failure_reason, read_env_clock, turn_note, +) + + +def main(stdin: str, environ: dict[str, str]) -> str: + """What the hook prints for the CLI's hook input ``stdin``: its JSON output, or nothing.""" + configured = environ.get(ENV_CLOCK_SERVER_ENV) + if not configured: + return "" + try: + tool = (json.loads(stdin or "{}") or {}).get("tool_name") or "" + except ValueError: + tool = "" + # A get_time call's own result already is the reading. + if tool == ENV_CLOCK_TOOL_NAME or tool.endswith(f"__{ENV_CLOCK_TOOL_NAME}"): + return "" + try: + current = asyncio.run(read_env_clock(ClockServer.from_env(configured))) + except Exception as e: + print(f"env clock: {failure_reason(e)}", file=sys.stderr) + return "" + return json.dumps({"hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": turn_note(current)}}) + + +if __name__ == "__main__": + out = main(sys.stdin.read(), dict(os.environ)) + if out: + print(out) diff --git a/src/agent_env/agents/agents/claude-code/install.sh b/src/agent_env/agents/agents/claude-code/install.sh index ec0ea30ce..c17b0ba2e 100755 --- a/src/agent_env/agents/agents/claude-code/install.sh +++ b/src/agent_env/agents/agents/claude-code/install.sh @@ -22,7 +22,7 @@ PYTHON_VERSION=3.12 START_TIMEOUT_S=120 STOP_TIMEOUT_S=10 CLAUDE_CODE_VERSION=$(sed -n 's/^ARG CLAUDE_CODE_VERSION=//p' "$here/Dockerfile") -AGENT_FILES="a2a_server.py claude_bridge.py kit.py requirements.txt run.sh tar_changelog.py" +AGENT_FILES="a2a_server.py claude_bridge.py env_clock.py env_clock_hook.py kit.py requirements.txt run.sh tar_changelog.py" # put SRC DEST replaces DEST by a rename, so a binary an earlier install's agent still runs can be replaced. case $mode in diff --git a/tst/unit/agents/claude_code/conftest.py b/tst/unit/agents/claude_code/conftest.py index ee1bd3676..b58034ea8 100644 --- a/tst/unit/agents/claude_code/conftest.py +++ b/tst/unit/agents/claude_code/conftest.py @@ -27,6 +27,15 @@ def agent_dirs(monkeypatch, tmp_path): return state +@pytest.fixture(autouse=True) +def no_env_clock(monkeypatch): + """No env clock unless a test gives one: a turn otherwise asks each MCP server for get_time.""" + async def _none(servers): + return None + + monkeypatch.setattr(a2a_server, "find_env_clock", _none) + + @pytest.fixture def app(): return a2a_server.build_app(a2a_server.ClaudeCodeAgent()) diff --git a/tst/unit/agents/claude_code/test_env_clock.py b/tst/unit/agents/claude_code/test_env_clock.py new file mode 100644 index 000000000..fcd896632 --- /dev/null +++ b/tst/unit/agents/claude_code/test_env_clock.py @@ -0,0 +1,209 @@ +"""The env's clock reaches the model: while an MCP server lists get_time (the gateway does only while the env's clock is +armed), each prompt carries the env's time, and the CLI runs a PostToolUse hook that ends each tool call's result with +a fresh reading. Unarmed, nothing changes. Read here from real MCP servers over streamable HTTP.""" + +from __future__ import annotations + +import asyncio +import importlib +import json +import os +import socket +import subprocess +import sys +import threading +import time + +import pytest +import uvicorn +from mcp.server.fastmcp import FastMCP + +from tst.unit.agents.claude_code.agent import AGENT_DIR, a2a_server, claude_bridge +from tst.unit.agents.claude_code.support import FakeCLI + +env_clock = importlib.import_module("env_clock") +env_clock_hook = importlib.import_module("env_clock_hook") + +NOW = "2026-05-01T09:00:00Z" + + +def _serve(server: FastMCP) -> tuple[str, uvicorn.Server]: + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + running = uvicorn.Server(uvicorn.Config(server.streamable_http_app(), log_level="warning")) + threading.Thread(target=lambda: asyncio.run(running.serve(sockets=[sock])), daemon=True).start() + deadline = time.monotonic() + 10 + while not running.started: + assert time.monotonic() < deadline, "the MCP server didn't start" + time.sleep(0.02) + return f"http://127.0.0.1:{port}/mcp", running + + +@pytest.fixture(scope="module") +def servers(): # every test using it asks for socket_enabled: they read it over this machine's loopback + """A gateway-like server listing get_time, and one without it.""" + gateway = FastMCP("env", stateless_http=True, json_response=True) + + @gateway.tool() + def get_time() -> dict: + return {"current_time": NOW} + + @gateway.tool() + def list_notes() -> list[str]: + return [] + + other = FastMCP("peers", stateless_http=True, json_response=True) + + @other.tool() + def peer_send_message(text: str) -> str: + return text + + started = [_serve(gateway), _serve(other)] + yield {"env": started[0][0], "peers": started[1][0]} + for _, running in started: + running.should_exit = True + + +def _entries(**urls: str) -> dict: + return {name: env_clock.ClockServer(url=url) for name, url in urls.items()} + + +def test_the_server_listing_get_time_is_found_and_read(socket_enabled, servers): + found = asyncio.run(env_clock.find_env_clock(_entries(peers=servers["peers"], env=servers["env"]))) + + assert found is not None + assert found[0].url == servers["env"] and found[1] == NOW + + +def test_unarmed_no_server_lists_get_time_and_nothing_is_found(socket_enabled, servers): + assert asyncio.run(env_clock.find_env_clock(_entries(peers=servers["peers"]))) is None + + +def test_a_server_that_fails_is_skipped_for_the_next(socket_enabled, servers): + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + closed = f"http://127.0.0.1:{sock.getsockname()[1]}/mcp" + sock.close() + + found = asyncio.run(env_clock.find_env_clock(_entries(gone=closed, env=servers["env"]))) + + assert found is not None and found[1] == NOW + + +def test_a_slow_clock_leaves_the_prompt_without_its_note(monkeypatch): + async def _hangs(server, *, find): + await asyncio.sleep(60) + + monkeypatch.setattr(env_clock, "_read", _hangs) + monkeypatch.setattr(env_clock, "PROMPT_READ_TIMEOUT_S", 0.05) + + assert asyncio.run(env_clock.find_env_clock(_entries(env="http://unused"))) is None + + +def test_the_mcp_servers_entries_carry_the_role_the_cli_sends(): + entries = claude_bridge.mcp_server_entries({"env": {"url": "http://gw/mcp", "headers": {"X": "1"}}}, "agent") + + assert entries == {"env": env_clock.ClockServer(url="http://gw/mcp", headers={"X": "1", "AgentEnv-Role": "agent"})} + + +def _turn(monkeypatch, agent, found) -> dict: + """One turn with ``find_env_clock`` answering ``found``; what run_claude was called with.""" + called: dict = {} + + async def _find(servers): + called["servers"] = servers + return found + + async def _run_claude(**kwargs): + called.update(kwargs) + return claude_bridge.ClaudeResult(response="ok", tool_call_count=0, session_id=kwargs["session_id"]) + + monkeypatch.setattr(a2a_server, "find_env_clock", _find) + monkeypatch.setattr(a2a_server, "run_claude", _run_claude) + request = a2a_server.TaskRequest(task_id="t", context_id="c", parts=(a2a_server.TextPart(text="book it"),), + config=a2a_server.ClaudeCodeConfig(role="agent"), + mcp_servers={"env": {"url": "http://gw/mcp"}}) + asyncio.run(agent.run(request)) + return called + + +def test_an_armed_clock_puts_the_envs_time_in_the_prompt_and_the_hook_on_the_cli(monkeypatch, agent): + server = env_clock.ClockServer(url="http://gw/mcp", headers={"AgentEnv-Role": "agent"}) + called = _turn(monkeypatch, agent, (server, NOW)) + + assert called["servers"] == {"env": server} + assert called["prompt"].startswith("book it\n\n") + assert NOW in called["prompt"] and called["env_clock"] == server + + +def test_an_unarmed_clock_leaves_the_prompt_and_the_cli_alone(monkeypatch, agent): + called = _turn(monkeypatch, agent, None) + + assert called["prompt"] == "book it" and called["env_clock"] is None + + +def _command(env_clock_on: bool) -> list[str]: + return claude_bridge._command("s", False, model=None, allowed_tools=None, mcp_servers=None, system_prompt=None, + output_format=None, role=None, effort=None, max_turns=None, env_clock=env_clock_on) + + +def test_the_hook_is_on_the_command_line_only_while_armed(): + assert "--settings" not in _command(False) + command = _command(True) + [hook] = json.loads(command[command.index("--settings") + 1])["hooks"]["PostToolUse"][0]["hooks"] + + assert hook["command"].endswith("env_clock_hook.py") and hook["timeout"] == env_clock.HOOK_TIMEOUT_S + + +def test_the_cli_gets_the_clocks_server_in_its_environment_only_while_armed(monkeypatch): + envs = [] + + async def _exec(*cmd, **kwargs): + envs.append(kwargs["env"]) + return FakeCLI((json.dumps({"type": "result", "subtype": "success", "result": "ok", + "session_id": "s"}) + "\n").encode()) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) + server = env_clock.ClockServer(url="http://gw/mcp", headers={"AgentEnv-Role": "agent"}) + asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False)) + asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False, env_clock=server)) + + assert env_clock.ENV_CLOCK_SERVER_ENV not in envs[0] + assert env_clock.ClockServer.from_env(envs[1][env_clock.ENV_CLOCK_SERVER_ENV]) == server + + +def _hook(stdin: dict, environ: dict) -> subprocess.CompletedProcess: + """The hook script as the CLI runs it: its own process, the hook input on stdin.""" + return subprocess.run([sys.executable, str(AGENT_DIR / "env_clock_hook.py")], input=json.dumps(stdin), + capture_output=True, text=True, timeout=30, + env={**os.environ, "PYTHONPATH": os.pathsep.join(sys.path), **environ}) + + +def test_the_hook_ends_a_tool_calls_result_with_the_envs_time(socket_enabled, servers): + ran = _hook({"hook_event_name": "PostToolUse", "tool_name": "Bash"}, + {env_clock.ENV_CLOCK_SERVER_ENV: env_clock.ClockServer(url=servers["env"]).to_env()}) + + assert ran.returncode == 0, ran.stderr + output = json.loads(ran.stdout)["hookSpecificOutput"] + assert output == {"hookEventName": "PostToolUse", "additionalContext": env_clock.turn_note(NOW)} + + +@pytest.mark.parametrize("stdin, environ", [ + ({"tool_name": "Bash"}, {}), # unarmed: no server configured + ({"tool_name": "mcp__env__get_time"}, {"configured": True}), # its own result is the reading + ({"tool_name": "Bash"}, {"unreachable": True}), # a clock blip prints nothing +]) +def test_the_hook_prints_nothing_unarmed_after_get_time_or_on_a_failed_read(socket_enabled, servers, stdin, environ): + if environ.get("configured"): + environ = {env_clock.ENV_CLOCK_SERVER_ENV: env_clock.ClockServer(url=servers["env"]).to_env()} + elif environ.get("unreachable"): + environ = {env_clock.ENV_CLOCK_SERVER_ENV: env_clock.ClockServer(url="http://127.0.0.1:9/mcp").to_env()} + + ran = _hook(stdin, environ) + + assert ran.returncode == 0 and ran.stdout == "", (ran.stdout, ran.stderr) + + +def test_the_hook_reads_its_input_as_the_cli_gives_it(): + assert env_clock_hook.main("not json", {}) == "" diff --git a/tst/unit/agents/claude_code/test_image_contract.py b/tst/unit/agents/claude_code/test_image_contract.py index 6445b09d6..9af21993a 100644 --- a/tst/unit/agents/claude_code/test_image_contract.py +++ b/tst/unit/agents/claude_code/test_image_contract.py @@ -19,6 +19,16 @@ def test_every_file_the_image_copies_is_in_the_folder(): assert {"a2a_server.py", "claude_bridge.py", "kit.py", "tar_changelog.py", "entrypoint.sh", "install.sh", "run.sh"} <= set(copied) +def test_the_image_and_an_install_carry_every_module_of_the_folder(): + modules = {path.name for path in AGENT_DIR.glob("*.py")} + copied = {name for line in re.findall(r"^COPY (?!--from=)(.+) \S+$", DOCKERFILE, re.MULTILINE) + for name in line.split()} + installed = set(re.search(r'^AGENT_FILES="([^"]+)"$', (AGENT_DIR / "install.sh").read_text(), + re.MULTILINE).group(1).split()) + + assert modules - copied == set() and modules - installed == set() + + def test_the_image_installs_the_protocol_alone_from_its_requirements(): requirements = [line for line in (AGENT_DIR / "requirements.txt").read_text().splitlines() if line and not line.startswith("#")] From aa1874771e46fea5d0498ac3d9e8b3f27455f795 Mon Sep 17 00:00:00 2001 From: William Loo Date: Fri, 13 Nov 2026 10:29:50 -0800 Subject: [PATCH 2/4] feat(agents): the clock hook follows failed tool calls too, and a resume carries the env's time now - The hook also runs on PostToolUseFailure, naming the event it ran for: a long command that fails takes as much of the clock as one that succeeds. - A resume after a transient API failure reads the clock again after its backoff and ends its prompt with the env's time then; a failed read leaves the resume prompt as it was. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/agents/README.md | 7 +-- .../agents/claude-code/claude_bridge.py | 14 +++++- .../agents/agents/claude-code/env_clock.py | 17 ++++--- .../agents/claude-code/env_clock_hook.py | 15 ++++--- tst/unit/agents/claude_code/test_env_clock.py | 44 ++++++++++++++++--- 5 files changed, 75 insertions(+), 22 deletions(-) diff --git a/src/agent_env/agents/README.md b/src/agent_env/agents/README.md index 03fbb8cdf..d80ff1ddd 100644 --- a/src/agent_env/agents/README.md +++ b/src/agent_env/agents/README.md @@ -31,9 +31,10 @@ when the Dockerfile's `CLAUDE_CODE_VERSION` moves, until `ClaudeCodeTool` and `T The CLI states the date from its machine's clock, which isn't the task env's while the env's clock is armed. So while an MCP server lists the gateway's `get_time`, which it does only then, each prompt ends with the env's time in an -`` note, and a `PostToolUse` hook, given on the CLI's command line (`--settings`), ends each tool call's -result with a fresh reading (`env_clock.py`). A failed read leaves the note out and never fails the turn. Unarmed, -nothing is added: the env runs on real time, which is the CLI's date. +`` note, a resume after an API failure too, and a `PostToolUse` and `PostToolUseFailure` hook, given on the +CLI's command line (`--settings`), ends each tool call's result, failed or not, with a fresh reading (`env_clock.py`). +A failed read leaves the note out and never fails the turn. Unarmed, nothing is added: the env runs on real time, which +is the CLI's date. `install_agent` installs it too, into a container a task already runs or onto a `vm` sandbox's Linux host, with `install.sh` from its folder: Claude Code's native build for the target's CPU and libc, uv, and the agent's requirements diff --git a/src/agent_env/agents/agents/claude-code/claude_bridge.py b/src/agent_env/agents/agents/claude-code/claude_bridge.py index a8244bde0..494c52b6d 100644 --- a/src/agent_env/agents/agents/claude-code/claude_bridge.py +++ b/src/agent_env/agents/agents/claude-code/claude_bridge.py @@ -20,7 +20,7 @@ from pydantic import BaseModel, ConfigDict, model_validator -from env_clock import ENV_CLOCK_SERVER_ENV, ClockServer, hook_settings +from env_clock import ENV_CLOCK_SERVER_ENV, ClockServer, failure_reason, hook_settings, prompt_note, read_env_clock from kit import in_thread_to_its_end logger = logging.getLogger(__name__) @@ -482,6 +482,18 @@ async def run_claude( logger.warning(f"claude CLI turn cut short ({_redact(failure)!r}); resuming session {attempt_session} " f"in {backoff}s with {int(budget_left)}s of the original {timeout_s}s budget left") await asyncio.sleep(backoff) + if env_clock is not None: + attempt_prompt = await _with_env_clock_note(_RESUME_PROMPT, env_clock) + + +async def _with_env_clock_note(prompt: str, server: ClockServer) -> str: + """``prompt`` with the env's time now: a resume after an API failure and its backoff mustn't carry on from the + time the turn's prompt gave. A failed read leaves it as it is.""" + try: + return f"{prompt}\n\n{prompt_note(await read_env_clock(server))}" + except Exception as e: + logger.warning(f"env clock: no reading for the resume: {failure_reason(e)}") + return prompt _PROCESS_CLEANUP_TIMEOUT_S = 5 diff --git a/src/agent_env/agents/agents/claude-code/env_clock.py b/src/agent_env/agents/agents/claude-code/env_clock.py index 0eb1d1331..bbef1c818 100644 --- a/src/agent_env/agents/agents/claude-code/env_clock.py +++ b/src/agent_env/agents/agents/claude-code/env_clock.py @@ -1,9 +1,9 @@ """The env's clock, read for the model. While agent-env's gateway runs an env's clock (urn:agentenv:clock/v1), it lists ``get_time`` on the env's MCP server, and the date the CLI states on its own ("Today's date is ...", from this -machine's clock) isn't the env's. So each prompt carries the env's time, and a PostToolUse hook (``env_clock_hook.py``) -adds a fresh reading after each tool call: a clock moved between prompts, or running faster than real time, never -leaves the model on a stale date. Unarmed, no server lists ``get_time`` and nothing is added: the env runs on real -time, which is the CLI's date.""" +machine's clock) isn't the env's. So each prompt, a resume's too, carries the env's time, and a hook +(``env_clock_hook.py``) adds a fresh reading after each tool call, failed ones too: a clock moved between prompts, or +running faster than real time, never leaves the model on a stale date. Unarmed, no server lists ``get_time`` and +nothing is added: the env runs on real time, which is the CLI's date.""" from __future__ import annotations @@ -31,6 +31,9 @@ HOOK_READ_TIMEOUT_S = 5.0 HOOK_TIMEOUT_S = 10 HOOK_SCRIPT = Path(__file__).with_name("env_clock_hook.py") +# After every tool call, whether it succeeded or failed: a long command that fails takes as much of the clock as one +# that succeeds. +HOOK_EVENTS = ("PostToolUse", "PostToolUseFailure") @dataclass(frozen=True) @@ -124,8 +127,8 @@ def turn_note(current: str) -> str: def hook_settings() -> str: - """The CLI's --settings for an armed clock: the hook after each tool call. On the command line rather than in the + """The CLI's --settings for an armed clock: the hook after each tool call, succeeded or failed. On the command line rather than in the user settings, so a snapshot of the CLI's state never carries it to an agent with no clock or no such script.""" command = f"{shlex.quote(sys.executable)} {shlex.quote(str(HOOK_SCRIPT))}" - return json.dumps({"hooks": {"PostToolUse": [ - {"matcher": "", "hooks": [{"type": "command", "command": command, "timeout": HOOK_TIMEOUT_S}]}]}}) + entry = {"matcher": "", "hooks": [{"type": "command", "command": command, "timeout": HOOK_TIMEOUT_S}]} + return json.dumps({"hooks": {event: [entry] for event in HOOK_EVENTS}}) diff --git a/src/agent_env/agents/agents/claude-code/env_clock_hook.py b/src/agent_env/agents/agents/claude-code/env_clock_hook.py index 7af19974e..c395474c2 100644 --- a/src/agent_env/agents/agents/claude-code/env_clock_hook.py +++ b/src/agent_env/agents/agents/claude-code/env_clock_hook.py @@ -1,5 +1,5 @@ -"""The CLI's PostToolUse hook while the env's clock is armed: the env's time now, as context ending the tool's result -(``env_clock``). Any failure prints nothing, so a clock blip never fails a tool call; the model can still call +"""The CLI's PostToolUse and PostToolUseFailure hook while the env's clock is armed: the env's time now, as context +ending the tool's result (``env_clock``). Any failure prints nothing, so a clock blip never fails a tool call; the model can still call get_time itself.""" from __future__ import annotations @@ -11,7 +11,7 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from env_clock import ( # noqa: E402 - ENV_CLOCK_SERVER_ENV, ENV_CLOCK_TOOL_NAME, ClockServer, failure_reason, read_env_clock, turn_note, + ENV_CLOCK_SERVER_ENV, ENV_CLOCK_TOOL_NAME, HOOK_EVENTS, ClockServer, failure_reason, read_env_clock, turn_note, ) @@ -21,9 +21,12 @@ def main(stdin: str, environ: dict[str, str]) -> str: if not configured: return "" try: - tool = (json.loads(stdin or "{}") or {}).get("tool_name") or "" + given = json.loads(stdin or "{}") or {} except ValueError: - tool = "" + given = {} + tool = given.get("tool_name") or "" + # The output names the event the CLI ran the hook for, which it checks. + event = given.get("hook_event_name") if given.get("hook_event_name") in HOOK_EVENTS else HOOK_EVENTS[0] # A get_time call's own result already is the reading. if tool == ENV_CLOCK_TOOL_NAME or tool.endswith(f"__{ENV_CLOCK_TOOL_NAME}"): return "" @@ -32,7 +35,7 @@ def main(stdin: str, environ: dict[str, str]) -> str: except Exception as e: print(f"env clock: {failure_reason(e)}", file=sys.stderr) return "" - return json.dumps({"hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": turn_note(current)}}) + return json.dumps({"hookSpecificOutput": {"hookEventName": event, "additionalContext": turn_note(current)}}) if __name__ == "__main__": diff --git a/tst/unit/agents/claude_code/test_env_clock.py b/tst/unit/agents/claude_code/test_env_clock.py index fcd896632..d27d83cf1 100644 --- a/tst/unit/agents/claude_code/test_env_clock.py +++ b/tst/unit/agents/claude_code/test_env_clock.py @@ -151,9 +151,42 @@ def _command(env_clock_on: bool) -> list[str]: def test_the_hook_is_on_the_command_line_only_while_armed(): assert "--settings" not in _command(False) command = _command(True) - [hook] = json.loads(command[command.index("--settings") + 1])["hooks"]["PostToolUse"][0]["hooks"] + hooks = json.loads(command[command.index("--settings") + 1])["hooks"] - assert hook["command"].endswith("env_clock_hook.py") and hook["timeout"] == env_clock.HOOK_TIMEOUT_S + # After a tool call that failed too: a long command that fails takes as much of the clock. + assert set(hooks) == {"PostToolUse", "PostToolUseFailure"} + for entries in hooks.values(): + [hook] = entries[0]["hooks"] + assert hook["command"].endswith("env_clock_hook.py") and hook["timeout"] == env_clock.HOOK_TIMEOUT_S + + +def test_a_resume_after_an_api_failure_carries_the_envs_time_now(monkeypatch, tmp_path): + procs: list[FakeCLI] = [] + answers = [ + {"type": "result", "subtype": "success", "is_error": True, "result": "API Error: The operation timed out.", + "session_id": "s"}, + {"type": "result", "subtype": "success", "result": "done", "session_id": "s"}, + ] + + async def _exec(*cmd, **kwargs): + procs.append(FakeCLI((json.dumps(answers.pop(0)) + "\n").encode(), 1 if answers else 0)) + return procs[-1] + + async def _read(server, timeout_s=env_clock.HOOK_READ_TIMEOUT_S): + return "2026-05-02T10:00:00Z" + + monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) + monkeypatch.setattr(claude_bridge, "SESSION_JSONL_DIR", tmp_path) + (tmp_path / "s.jsonl").write_text('{"type":"assistant"}\n') + monkeypatch.setattr(claude_bridge, "_RETRY_BACKOFF_S", (0, 0)) + monkeypatch.setattr(claude_bridge, "read_env_clock", _read) + result = asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False, + env_clock=env_clock.ClockServer(url="http://gw/mcp"))) + + prompts = [proc.stdin.data.decode() for proc in procs] + assert result.attempts == 2 and prompts[0] == "p" + assert prompts[1].startswith(claude_bridge._RESUME_PROMPT) + assert prompts[1].endswith(env_clock.prompt_note("2026-05-02T10:00:00Z")) def test_the_cli_gets_the_clocks_server_in_its_environment_only_while_armed(monkeypatch): @@ -180,13 +213,14 @@ def _hook(stdin: dict, environ: dict) -> subprocess.CompletedProcess: env={**os.environ, "PYTHONPATH": os.pathsep.join(sys.path), **environ}) -def test_the_hook_ends_a_tool_calls_result_with_the_envs_time(socket_enabled, servers): - ran = _hook({"hook_event_name": "PostToolUse", "tool_name": "Bash"}, +@pytest.mark.parametrize("event", ["PostToolUse", "PostToolUseFailure"]) +def test_the_hook_ends_a_tool_calls_result_with_the_envs_time(socket_enabled, servers, event): + ran = _hook({"hook_event_name": event, "tool_name": "Bash"}, {env_clock.ENV_CLOCK_SERVER_ENV: env_clock.ClockServer(url=servers["env"]).to_env()}) assert ran.returncode == 0, ran.stderr output = json.loads(ran.stdout)["hookSpecificOutput"] - assert output == {"hookEventName": "PostToolUse", "additionalContext": env_clock.turn_note(NOW)} + assert output == {"hookEventName": event, "additionalContext": env_clock.turn_note(NOW)} @pytest.mark.parametrize("stdin, environ", [ From 7fd181f0008a1f63a43688e2e75edde913f7c3da Mon Sep 17 00:00:00 2001 From: William Loo Date: Fri, 13 Nov 2026 10:46:18 -0800 Subject: [PATCH 3/4] refactor(agents): name the env clock's error text limit Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/agents/agents/claude-code/env_clock.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/agent_env/agents/agents/claude-code/env_clock.py b/src/agent_env/agents/agents/claude-code/env_clock.py index bbef1c818..2cd70f48a 100644 --- a/src/agent_env/agents/agents/claude-code/env_clock.py +++ b/src/agent_env/agents/agents/claude-code/env_clock.py @@ -31,6 +31,8 @@ HOOK_READ_TIMEOUT_S = 5.0 HOOK_TIMEOUT_S = 10 HOOK_SCRIPT = Path(__file__).with_name("env_clock_hook.py") +# How much of a bad get_time answer an error quotes. +_ERROR_TEXT_LIMIT = 200 # After every tool call, whether it succeeded or failed: a long command that fails takes as much of the clock as one # that succeeds. HOOK_EVENTS = ("PostToolUse", "PostToolUseFailure") @@ -67,13 +69,13 @@ async def _read(server: ClockServer, *, find: bool) -> str | None: return None result = await session.call_tool(ENV_CLOCK_TOOL_NAME, {}) if result.isError: - raise ValueError(f"{ENV_CLOCK_TOOL_NAME} failed: {_text(result)[:200]!r}") + raise ValueError(f"{ENV_CLOCK_TOOL_NAME} failed: {_text(result)[:_ERROR_TEXT_LIMIT]!r}") if isinstance(result.structuredContent, dict) and result.structuredContent.get("current_time"): return str(result.structuredContent["current_time"]) text = _text(result) current = json.loads(text).get("current_time") if not current: - raise ValueError(f"no current_time in {text[:200]!r}") + raise ValueError(f"no current_time in {text[:_ERROR_TEXT_LIMIT]!r}") return str(current) From 7dc1f53bd23d9323eea61a200c8d56eebefbaf28 Mon Sep 17 00:00:00 2001 From: William Loo Date: Fri, 13 Nov 2026 14:49:56 -0800 Subject: [PATCH 4/4] refactor(agents): every AgentEnvAgent gets the env's clock from the framework; Claude Code keeps its tool-call hook (FD-5420) The protocol's executor finds get_time on the registered env servers before each turn, ends the request with the env's time and carries TaskRequest.env_clock for a harness to read between steps. Claude Code keeps only its PostToolUse/PostToolUseFailure hook and the note on a resume, and degrades to no clock on an older protocol. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/agentenv-protocol/README.md | 18 ++ .../agentenv_protocol/a2a_agent/__init__.py | 12 ++ .../agentenv_protocol/a2a_agent/env_clock.py | 153 ++++++++++++++ .../agentenv_protocol/a2a_agent/framework.py | 22 +- .../agentenv_protocol/a2a_agent/tasks/v1.py | 8 +- packages/agentenv-protocol/tests/conftest.py | 14 ++ .../tests/test_a2a_env_clock.py | 192 +++++++++++++++++ src/agent_env/agents/README.md | 12 +- .../agents/agents/claude-code/Dockerfile | 2 +- .../agents/agents/claude-code/a2a_server.py | 11 +- .../agents/claude-code/claude_bridge.py | 47 +++-- .../agents/agents/claude-code/env_clock.py | 136 ------------ .../agents/claude-code/env_clock_hook.py | 36 +++- .../agents/agents/claude-code/install.sh | 2 +- tst/unit/agents/claude_code/conftest.py | 8 +- tst/unit/agents/claude_code/test_env_clock.py | 193 ++++++------------ 16 files changed, 544 insertions(+), 322 deletions(-) create mode 100644 packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/env_clock.py create mode 100644 packages/agentenv-protocol/tests/conftest.py create mode 100644 packages/agentenv-protocol/tests/test_a2a_env_clock.py delete mode 100644 src/agent_env/agents/agents/claude-code/env_clock.py diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index 401c5ea65..22f34e14d 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -1355,6 +1355,24 @@ request fields are additive and have framework defaults. Agent code returns a `T its factories or builder so additions to the result contract do not break existing handlers. +While the task env's clock is armed, the gateway lists `get_time` on the env's +MCP server, and the agent's machine clock, and any date its harness states from +it, isn't the env's. So before each turn the executor asks the MCP servers +agent-env registered (never the SDK's peer-agents server) for `get_time`, with +the task's role. When one lists it, the request's `parts` end with the env's +time in an `` note, and `TaskRequest.env_clock` (an `EnvClock`) +names that server and the time read. An agent that places the note in its +prompt itself sets `env_clock_prompt_note = False` and still gets +`env_clock`. A harness that runs many steps in one turn reads the clock again +between them, so a clock moved during the turn, or running faster than real +time, never leaves its model on a stale date: `await request.env_clock.read()` +and `turn_note(...)`, or, from a subprocess such as a CLI's hook, +`ClockServer.from_env(os.environ[ENV_CLOCK_SERVER_ENV])` and `read_env_clock` +(set the variable to `request.env_clock.server.to_env()` for it). Unarmed, no +server lists `get_time` and the request is unchanged: the env runs on real +time. A failed or slow read (`agentenv_protocol.a2a_agent.env_clock`) is +logged, leaves the note out, and never fails the turn. + Expected execution failures are returned as `TaskResult.failure(code, message)` and become failed A2A tasks with `error_type`, `error_code`, and `error_message` in the terminal message. `error_type` is the platform classification diff --git a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/__init__.py b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/__init__.py index 5f17fed1e..56e8a167f 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/__init__.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/__init__.py @@ -123,6 +123,13 @@ Usage, WriteOnly, ) +from .env_clock import ( + ENV_CLOCK_SERVER_ENV, + ClockServer, + EnvClock, + read_env_clock, + turn_note, +) from ..transfers import ( HttpGetGrant, HttpPostPolicyGrant, @@ -138,6 +145,11 @@ ) __all__ = [ + "ENV_CLOCK_SERVER_ENV", + "ClockServer", + "EnvClock", + "read_env_clock", + "turn_note", "AGENT_CONFIG_V1", "ATTRIBUTION_PROBE_V1", "CDP_CONFIG_V1", diff --git a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/env_clock.py b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/env_clock.py new file mode 100644 index 000000000..4b27c9b8d --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/env_clock.py @@ -0,0 +1,153 @@ +"""The env's clock, read for an agent's model. While agent-env's gateway runs a task env's clock +(``urn:agentenv:clock/v1``), it lists ``get_time`` on the env's MCP server, and the agent's own machine clock, and any +date its harness states from it, isn't the env's. + +The framework reads the clock before each turn (``find_env_clock``): when a registered MCP server lists ``get_time``, +the turn's request ends with the env's time in an ```` note (unless the agent places it itself, +``AgentEnvAgent.env_clock_prompt_note``) and carries ``TaskRequest.env_clock``, which an agent's harness reads again +between steps (``EnvClock.read``, ``turn_note``), say after each tool call, so a clock moved during a turn, or running +faster than real time, never leaves its model on a stale date. A subprocess such as a CLI hook gets the server through +its environment (``ENV_CLOCK_SERVER_ENV``, ``ClockServer.from_env``). Unarmed, no server lists ``get_time`` and nothing +changes: the env runs on real time. A failed or slow read is logged and never fails the turn.""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +from ..tool_context import ROLE_HEADER + +logger = logging.getLogger(__name__) + +ENV_CLOCK_TOOL_NAME = "get_time" +# The server whose get_time a subprocess reads, as ``ClockServer.to_env`` JSON. +ENV_CLOCK_SERVER_ENV = "AGENTENV_ENV_CLOCK_SERVER" +# How long a turn waits for the clock before going without it. +FIND_TIMEOUT_S = 10.0 +# How long a read between a turn's steps waits. +READ_TIMEOUT_S = 5.0 +# How much of a bad get_time answer an error quotes. +_ERROR_TEXT_LIMIT = 200 + + +@dataclass(frozen=True) +class ClockServer: + """An MCP server's url and the headers to send it.""" + + url: str + headers: dict[str, str] = field(default_factory=dict) + + def to_env(self) -> str: + return json.dumps({"url": self.url, "headers": self.headers}) + + @classmethod + def from_env(cls, value: str) -> ClockServer: + raw = json.loads(value) + return cls(url=raw["url"], headers=dict(raw.get("headers") or {})) + + +@dataclass(frozen=True) +class EnvClock: + """An armed env clock: the server whose get_time reads it, and the time it read at the turn's start.""" + + server: ClockServer + time: str + + async def read(self, timeout_s: float = READ_TIMEOUT_S) -> str: + """The env's time now.""" + return await read_env_clock(self.server, timeout_s) + + +def clock_servers(mcp_servers: Mapping[str, Any], role: str | None) -> dict[str, ClockServer]: + """Each MCP server's url and headers, with the task's role, which the gateway filters tools by and which wins over + one a registration names, as an agent sends them.""" + return {name: ClockServer(url=info["url"], + headers={**(info.get("headers") or {}), **({ROLE_HEADER: role} if role else {})}) + for name, info in mcp_servers.items()} + + +async def _read(server: ClockServer, *, find: bool) -> str | None: + """The env's time from ``server``'s get_time; None when ``find`` and the server doesn't list it.""" + from mcp import ClientSession + from mcp.client.streamable_http import streamablehttp_client + + async with streamablehttp_client(server.url, headers=server.headers or None) as (read, write, _): + async with ClientSession(read, write) as session: + await session.initialize() + if find: + cursor, listed = None, False + while not listed: + page = await session.list_tools(cursor=cursor) + listed = any(t.name == ENV_CLOCK_TOOL_NAME for t in page.tools) + cursor = page.nextCursor + if not cursor: + break + if not listed: + return None + result = await session.call_tool(ENV_CLOCK_TOOL_NAME, {}) + if result.isError: + raise ValueError(f"{ENV_CLOCK_TOOL_NAME} failed: {_text(result)[:_ERROR_TEXT_LIMIT]!r}") + if isinstance(result.structuredContent, dict) and result.structuredContent.get("current_time"): + return str(result.structuredContent["current_time"]) + text = _text(result) + current = json.loads(text).get("current_time") + if not current: + raise ValueError(f"no current_time in {text[:_ERROR_TEXT_LIMIT]!r}") + return str(current) + + +def _text(result: Any) -> str: + return "\n".join(c.text for c in (result.content or []) if getattr(c, "text", None)) + + +def failure_reason(e: BaseException) -> str: + """What failed: the first error inside the MCP client's task groups, which say only how many there were.""" + # By its shape, not BaseExceptionGroup, which Python 3.10 has only as a backport. + while isinstance(getattr(e, "exceptions", None), tuple) and e.exceptions: + e = e.exceptions[0] + return f"{type(e).__name__}: {e}" + + +async def find_env_clock(servers: Mapping[str, ClockServer], timeout_s: float = FIND_TIMEOUT_S) -> EnvClock | None: + """The armed clock of the first of ``servers`` that lists get_time; None when none does or the read fails or + takes longer than ``timeout_s``, which is logged.""" + + async def search() -> EnvClock | None: + for name, server in servers.items(): + try: + current = await _read(server, find=True) + except Exception as e: # one server's failure is no reason to skip the others + logger.warning(f"env clock: reading {ENV_CLOCK_TOOL_NAME} from MCP server {name!r} failed: " + f"{failure_reason(e)}") + continue + if current: + return EnvClock(server=server, time=current) + return None + + try: + return await asyncio.wait_for(search(), timeout=timeout_s) + except asyncio.TimeoutError: # not TimeoutError before Python 3.11 + logger.warning(f"env clock: no reading within {timeout_s:g} s; the turn goes without it") + return None + + +async def read_env_clock(server: ClockServer, timeout_s: float = READ_TIMEOUT_S) -> str: + """The env's time now, from ``server``'s get_time.""" + current = await asyncio.wait_for(_read(server, find=False), timeout=timeout_s) + assert current is not None + return current + + +def prompt_note(current: str) -> str: + # Same stance as get_time's own description: the machine's clock is not this environment's. + return (f"The current date and time in this environment is {current} (UTC). The clock on the machine " + f"you are running on, and any date you were given from it, may not match it, so use this value, or call " + f"{ENV_CLOCK_TOOL_NAME} for a fresh one before acting on a date.") + + +def turn_note(current: str) -> str: + return f"Environment time now: {current} (UTC)." diff --git a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/framework.py b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/framework.py index ce344d08a..3c18ac48b 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/framework.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/framework.py @@ -30,6 +30,7 @@ from ._trajectory_log import MAX_TRAJECTORY_READ_BYTES, TaskTrajectories from ._triggers import TriggerEngine, TriggerError +from .env_clock import clock_servers, find_env_clock, prompt_note from .extensions import ( AGENT_CONFIG_V1, CDP_CONFIG_V1, @@ -216,6 +217,10 @@ def _trim_idle(self) -> None: class AgentEnvAgent: """Typed base class for AgentEnv-compatible A2A agents.""" + #: While the task env's clock is armed, each turn's request ends with the env's time (``env_clock``). An agent + #: that places it in its prompt itself sets this False; ``TaskRequest.env_clock`` is set either way. + env_clock_prompt_note: bool = True + def run(self, request: TaskRequest[Any]) -> AgentRunResult: """Execute one normalized A2A task.""" raise NotImplementedError @@ -1047,6 +1052,7 @@ def __init__(self, agent: Any, definition: _AgentDefinition | None = None) -> No self.services, workspace=definition.workspace, streaming=self.streaming, + env_clock_prompt_note=bool(getattr(agent, "env_clock_prompt_note", True)), ) routes = [Route("/health", self._health, methods=["GET"])] @@ -1230,11 +1236,13 @@ def __init__( *, workspace: Path | None, streaming: bool, + env_clock_prompt_note: bool = True, ) -> None: self._handler = handler self._services = services self._workspace = workspace self._streaming = streaming + self._env_clock_prompt_note = env_clock_prompt_note self._context_locks = _BoundedContextLocks() async def execute(self, context: Any, event_queue: Any) -> None: @@ -1264,12 +1272,19 @@ async def execute(self, context: Any, event_queue: Any) -> None: } async with self._context_locks.acquire(task.context_id): trajectories.start(task.id) + request_parts = tuple( + _from_a2a_part(part) for part in (context.message.parts or []) + ) + # The env servers agent-env registered, never the SDK's peer-agents server, which has no clock. + env_clock = await find_env_clock( + clock_servers(self._services.mcp_servers, task_config.role) + ) if self._services.mcp_servers else None + if env_clock is not None and self._env_clock_prompt_note: + request_parts = (*request_parts, TaskTextPart(text=prompt_note(env_clock.time))) request = TaskRequest( task_id=task.id, context_id=task.context_id, - parts=tuple( - _from_a2a_part(part) for part in (context.message.parts or []) - ), + parts=request_parts, config=task_config, mcp_servers=self._services.task_mcp_servers(), cdp_endpoints=self._services.cdp_endpoints, @@ -1278,6 +1293,7 @@ async def execute(self, context: Any, event_queue: Any) -> None: session_ref=self._services.session_ref_for_context(task.context_id), workspace=self._workspace, trajectory=trajectory, + env_clock=env_clock, ) try: if self._streaming: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/tasks/v1.py b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/tasks/v1.py index 920ec7226..5f01db367 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/tasks/v1.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/tasks/v1.py @@ -14,11 +14,14 @@ from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Annotated, Any, Generic, Literal, TypeAlias, TypeVar, Union +from typing import TYPE_CHECKING, Annotated, Any, Generic, Literal, TypeAlias, TypeVar, Union from a2a.types import TaskStatusUpdateEvent from pydantic import BaseModel, ConfigDict, Field, model_validator +if TYPE_CHECKING: + from ..env_clock import EnvClock + logger = logging.getLogger(__name__) _LARGE_LIVE_TRAJECTORY_BYTES = 100 * 1024 * 1024 @@ -225,6 +228,9 @@ class TaskRequest(Generic[ConfigT]): trajectory: TrajectoryLog = field( default_factory=TrajectoryLog, repr=False, compare=False ) + # The task env's armed clock, read at the turn's start; None while unarmed. A harness reads it again between + # steps (``EnvClock.read``), and the framework has already ended ``parts`` with its time (``env_clock``). + env_clock: EnvClock | None = None def __post_init__(self) -> None: object.__setattr__(self, "parts", tuple(self.parts)) diff --git a/packages/agentenv-protocol/tests/conftest.py b/packages/agentenv-protocol/tests/conftest.py new file mode 100644 index 000000000..ec95e3305 --- /dev/null +++ b/packages/agentenv-protocol/tests/conftest.py @@ -0,0 +1,14 @@ +"""No env clock unless a test gives one: a turn with MCP servers registered otherwise asks each for get_time +(``agentenv_protocol.a2a_agent.env_clock``), and the servers these tests register don't exist.""" + +import pytest + + +@pytest.fixture(autouse=True) +def no_env_clock(monkeypatch): + from agentenv_protocol.a2a_agent import framework + + async def _none(servers, timeout_s=None): + return None + + monkeypatch.setattr(framework, "find_env_clock", _none) diff --git a/packages/agentenv-protocol/tests/test_a2a_env_clock.py b/packages/agentenv-protocol/tests/test_a2a_env_clock.py new file mode 100644 index 000000000..fe06676a4 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_a2a_env_clock.py @@ -0,0 +1,192 @@ +"""The task env's clock reaches every AgentEnvAgent: while a registered MCP server lists get_time (the gateway does only +while the env's clock is armed), each turn's request ends with the env's time and carries ``TaskRequest.env_clock`` +for the agent's harness to read again between steps. Unarmed, nothing changes. Read here from real MCP servers over +streamable HTTP on this machine's loopback.""" + +from __future__ import annotations + +import asyncio +import socket +import threading +import time +from typing import Any + +import pytest +import uvicorn +from mcp.server.fastmcp import FastMCP +from starlette.testclient import TestClient + +from agentenv_protocol.a2a_agent import ( + MCP_CONFIG_V1, + AgentEnvAgent, + AgentIdentity, + TaskRequest, + TaskResult, + TextPart, + a2a_agent, +) +from agentenv_protocol.a2a_agent import env_clock, framework + +NOW = "2026-05-01T09:00:00Z" + + +def _serve(server: FastMCP) -> tuple[str, uvicorn.Server]: + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + running = uvicorn.Server(uvicorn.Config(server.streamable_http_app(), log_level="warning")) + threading.Thread(target=lambda: asyncio.run(running.serve(sockets=[sock])), daemon=True).start() + deadline = time.monotonic() + 10 + while not running.started: + assert time.monotonic() < deadline, "the MCP server didn't start" + time.sleep(0.02) + return f"http://127.0.0.1:{port}/mcp", running + + +@pytest.fixture(scope="module") +def servers(): + """A gateway-like server listing get_time, and one without it.""" + gateway = FastMCP("env", stateless_http=True, json_response=True) + + @gateway.tool() + def get_time() -> dict: + return {"current_time": NOW} + + @gateway.tool() + def list_notes() -> list[str]: + return [] + + other = FastMCP("peers", stateless_http=True, json_response=True) + + @other.tool() + def peer_send_message(text: str) -> str: + return text + + started = [_serve(gateway), _serve(other)] + yield {"env": started[0][0], "peers": started[1][0]} + for _, running in started: + running.should_exit = True + + +def _servers(**urls: str) -> dict[str, env_clock.ClockServer]: + return {name: env_clock.ClockServer(url=url) for name, url in urls.items()} + + +def test_the_server_listing_get_time_is_found_and_read(servers): + found = asyncio.run(env_clock.find_env_clock(_servers(peers=servers["peers"], env=servers["env"]))) + + assert found == env_clock.EnvClock(server=env_clock.ClockServer(url=servers["env"]), time=NOW) + assert asyncio.run(found.read()) == NOW + + +def test_unarmed_no_server_lists_get_time_and_nothing_is_found(servers): + assert asyncio.run(env_clock.find_env_clock(_servers(peers=servers["peers"]))) is None + + +def test_a_server_that_fails_is_skipped_for_the_next(servers): + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + closed = f"http://127.0.0.1:{sock.getsockname()[1]}/mcp" + sock.close() + + found = asyncio.run(env_clock.find_env_clock(_servers(gone=closed, env=servers["env"]))) + + assert found is not None and found.time == NOW + + +def test_a_slow_clock_leaves_the_turn_without_it(monkeypatch): + async def _hangs(server, *, find): + await asyncio.sleep(60) + + monkeypatch.setattr(env_clock, "_read", _hangs) + + assert asyncio.run(env_clock.find_env_clock(_servers(env="http://unused"), timeout_s=0.05)) is None + + +def test_the_servers_carry_the_tasks_role_over_a_registrations(): + servers = env_clock.clock_servers({"env": {"url": "http://gw/mcp", "headers": {"AgentEnv-Role": "x", "X": "1"}}, + "bare": {"url": "http://b/mcp", "headers": None}}, "agent") + + assert servers == {"env": env_clock.ClockServer(url="http://gw/mcp", headers={"AgentEnv-Role": "agent", "X": "1"}), + "bare": env_clock.ClockServer(url="http://b/mcp", headers={"AgentEnv-Role": "agent"})} + assert env_clock.clock_servers({"bare": {"url": "http://b/mcp"}}, None)["bare"].headers == {} + + +def test_a_server_travels_to_a_subprocess_through_its_environment(): + server = env_clock.ClockServer(url="http://gw/mcp", headers={"AgentEnv-Role": "agent"}) + + assert env_clock.ClockServer.from_env(server.to_env()) == server + + +def test_a_failure_inside_the_mcp_clients_task_groups_is_named(): + class Group(Exception): + def __init__(self, *inner): + super().__init__("1 sub-exception") + self.exceptions = inner + + assert env_clock.failure_reason(Group(Group(ConnectionError("refused")))) == "ConnectionError: refused" + + +def _message() -> dict[str, Any]: + return {"jsonrpc": "2.0", "id": "r", "method": "message/send", "params": { + "message": {"kind": "message", "messageId": "m", "role": "user", "contextId": "c", + "parts": [{"kind": "text", "text": "book it"}]}, + "configuration": {"blocking": True}}} + + +def _turn(monkeypatch, found, *, note: bool = True, register: bool = True) -> tuple[TaskRequest, list]: + """One turn with ``find_env_clock`` answering ``found``; the request the agent got and what was searched.""" + seen: list[TaskRequest] = [] + searched: list = [] + + async def _find(servers, timeout_s=env_clock.FIND_TIMEOUT_S): + searched.append(servers) + return found + + monkeypatch.setattr(framework, "find_env_clock", _find) + + @a2a_agent(identity=AgentIdentity(name="clocked", description="test", version="1"), extensions=(MCP_CONFIG_V1,)) + class Agent(AgentEnvAgent): + env_clock_prompt_note = note + + async def run(self, request: TaskRequest) -> TaskResult: + seen.append(request) + return TaskResult.text("ok") + + with TestClient(Agent().create_app()) as client: + if register: + card = client.get("/.well-known/agent.json").json() + [extension] = [e for e in card["capabilities"]["extensions"] if e["uri"] == MCP_CONFIG_V1.uri] + add = extension["params"]["methods"]["add"] + path = add.get("endpoint", extension["params"].get("endpoint")) + assert client.request(add["method"], path, json={"name": "env", "url": "http://gw/mcp"}).is_success + assert client.post("/a2a", json=_message()).json()["result"]["status"]["state"] == "completed" + return seen[0], searched + + +def test_an_armed_clock_ends_every_agents_request_with_the_envs_time(monkeypatch): + clock = env_clock.EnvClock(server=env_clock.ClockServer(url="http://gw/mcp"), time=NOW) + request, searched = _turn(monkeypatch, clock) + + assert searched == [{"env": env_clock.ClockServer(url="http://gw/mcp")}] + assert request.parts == (TextPart(text="book it"), TextPart(text=env_clock.prompt_note(NOW))) + assert request.env_clock == clock + + +def test_an_agent_that_places_the_note_itself_gets_the_clock_alone(monkeypatch): + clock = env_clock.EnvClock(server=env_clock.ClockServer(url="http://gw/mcp"), time=NOW) + request, _ = _turn(monkeypatch, clock, note=False) + + assert request.parts == (TextPart(text="book it"),) and request.env_clock == clock + + +def test_an_unarmed_clock_leaves_the_request_alone(monkeypatch): + request, _ = _turn(monkeypatch, None) + + assert request.parts == (TextPart(text="book it"),) and request.env_clock is None + + +def test_with_no_mcp_server_registered_there_is_no_clock_to_ask(monkeypatch): + request, searched = _turn(monkeypatch, None, register=False) + + assert searched == [] and request.env_clock is None diff --git a/src/agent_env/agents/README.md b/src/agent_env/agents/README.md index d80ff1ddd..10fe6f2d2 100644 --- a/src/agent_env/agents/README.md +++ b/src/agent_env/agents/README.md @@ -29,12 +29,12 @@ turn off becomes `--disallowedTools`, which keeps the CLI's defaults for the res `claude_bridge.py`: every tool the pinned CLI has, though which it offers by default depends on the model. A test fails when the Dockerfile's `CLAUDE_CODE_VERSION` moves, until `ClaudeCodeTool` and `TOOLS_CLI_VERSION` move with it. -The CLI states the date from its machine's clock, which isn't the task env's while the env's clock is armed. So while -an MCP server lists the gateway's `get_time`, which it does only then, each prompt ends with the env's time in an -`` note, a resume after an API failure too, and a `PostToolUse` and `PostToolUseFailure` hook, given on the -CLI's command line (`--settings`), ends each tool call's result, failed or not, with a fresh reading (`env_clock.py`). -A failed read leaves the note out and never fails the turn. Unarmed, nothing is added: the env runs on real time, which -is the CLI's date. +The CLI states the date from its machine's clock, which isn't the task env's while the env's clock is armed. The +protocol's framework ends each prompt with the env's time then (`TaskRequest.env_clock`), and this agent has the CLI +run a `PostToolUse` and `PostToolUseFailure` hook, given on its command line (`--settings`), that ends each tool +call's result, failed or not, with a fresh reading (`env_clock_hook.py`); a resume after an API failure reads the clock +again too. A failed read leaves the note out and never fails the turn. Unarmed, nothing is added: the env runs on real +time, which is the CLI's date. `install_agent` installs it too, into a container a task already runs or onto a `vm` sandbox's Linux host, with `install.sh` from its folder: Claude Code's native build for the target's CPU and libc, uv, and the agent's requirements diff --git a/src/agent_env/agents/agents/claude-code/Dockerfile b/src/agent_env/agents/agents/claude-code/Dockerfile index 7ec8b123f..2f5e2ee3f 100644 --- a/src/agent_env/agents/agents/claude-code/Dockerfile +++ b/src/agent_env/agents/agents/claude-code/Dockerfile @@ -23,7 +23,7 @@ WORKDIR /opt/agent COPY requirements.txt model-requirements.txt ./ RUN pip install --no-cache-dir -r requirements.txt -r model-requirements.txt # install.sh and run.sh are install/v1's: in the image, so they stay in its build context. -COPY a2a_server.py claude_bridge.py env_clock.py env_clock_hook.py kit.py tar_changelog.py entrypoint.sh install.sh run.sh ./ +COPY a2a_server.py claude_bridge.py env_clock_hook.py kit.py tar_changelog.py entrypoint.sh install.sh run.sh ./ # The CLI's cwd, /app, and its state, fixed so neither follows the uid: Modal runs the image as root. RUN useradd --create-home --uid 10001 agent && mkdir -p /app /home/agent/.claude \ && chown agent:agent /app /home/agent/.claude && chmod 0755 entrypoint.sh diff --git a/src/agent_env/agents/agents/claude-code/a2a_server.py b/src/agent_env/agents/agents/claude-code/a2a_server.py index 819d916e6..ada06beab 100644 --- a/src/agent_env/agents/agents/claude-code/a2a_server.py +++ b/src/agent_env/agents/agents/claude-code/a2a_server.py @@ -77,13 +77,11 @@ SESSION_JSONL_DIR, WORKSPACE_DIR, kill_running_clis, - mcp_server_entries, read_partial_session_trajectory, redact_session_records, run_claude, session_jsonl_path, ) -from env_clock import find_env_clock, prompt_note from kit import ( DurableSessionRefs, FilePartMaterializeError, @@ -474,11 +472,6 @@ async def _turn(self, request: TaskRequest[ClaudeCodeConfig], cfg: ClaudeCodeCon paths_block = "\n".join(attachment_paths) user_text = f"{user_text}\n\n{paths_block}" if user_text else paths_block - # The env's time, when its clock is armed: the date the CLI states is this machine's, not the env's. - env_clock = await find_env_clock(mcp_server_entries(request.mcp_servers, cfg.role)) - if env_clock is not None: - user_text = f"{user_text}\n\n{prompt_note(env_clock[1])}" if user_text else prompt_note(env_clock[1]) - # A turn that failed before the CLI wrote its transcript left no session to resume. resume = session_jsonl_path(session_id).exists() try: @@ -500,7 +493,9 @@ async def _turn(self, request: TaskRequest[ClaudeCodeConfig], cfg: ClaudeCodeCon max_thinking_tokens=cfg.max_thinking_tokens, builtin_tools=cfg.builtin_tools, event_callback=request.trajectory.append, - env_clock=env_clock[0] if env_clock is not None else None, + # The framework ended the prompt with the env's time; the CLI's own date is this machine's. A protocol + # older than its env clock gives none. + env_clock=getattr(request, "env_clock", None), ) except Exception as e: logger.error(f"claude CLI invocation raised: {e}", exc_info=True) diff --git a/src/agent_env/agents/agents/claude-code/claude_bridge.py b/src/agent_env/agents/agents/claude-code/claude_bridge.py index 494c52b6d..efbeefe50 100644 --- a/src/agent_env/agents/agents/claude-code/claude_bridge.py +++ b/src/agent_env/agents/agents/claude-code/claude_bridge.py @@ -16,13 +16,16 @@ from dataclasses import dataclass, field, replace from functools import partial from pathlib import Path -from typing import Literal +from typing import TYPE_CHECKING, Literal from pydantic import BaseModel, ConfigDict, model_validator -from env_clock import ENV_CLOCK_SERVER_ENV, ClockServer, failure_reason, hook_settings, prompt_note, read_env_clock +from env_clock_hook import hook_settings from kit import in_thread_to_its_end +if TYPE_CHECKING: + from agentenv_protocol.a2a_agent.env_clock import EnvClock + logger = logging.getLogger(__name__) # The CLI's own state: settings, skills and session transcripts. The image fixes it, so it doesn't follow the uid. @@ -422,13 +425,13 @@ async def run_claude( max_thinking_tokens: int | None = None, builtin_tools: ClaudeCodeTools | None = None, event_callback: Callable[[dict], None] | None = None, - env_clock: ClockServer | None = None, + env_clock: EnvClock | None = None, ) -> ClaudeResult: """Run one turn, resuming the CLI session when a transient upstream failure cuts it short; a resume spends what is left of ``timeout_s``, so the turn never outlives its caller's deadline. ``CLAUDE_API_ERROR_RETRIES=0`` turns resumes off. ``event_callback`` gets each stdout event as the CLI prints it; an attempt's trailing `result` is - held until it's known not to be resumed. ``env_clock``, the server of an armed env clock's get_time, has the CLI - add the env's time after each tool call (``env_clock``).""" + held until it's known not to be resumed. ``env_clock``, the task env's armed clock (``TaskRequest.env_clock``), + has the CLI add the env's time after each tool call (``env_clock_hook``), and a resume carry it too.""" deadline = time.monotonic() + timeout_s retries_left = _configured_api_error_retries() spawn = partial( @@ -486,11 +489,13 @@ async def run_claude( attempt_prompt = await _with_env_clock_note(_RESUME_PROMPT, env_clock) -async def _with_env_clock_note(prompt: str, server: ClockServer) -> str: +async def _with_env_clock_note(prompt: str, env_clock: EnvClock) -> str: """``prompt`` with the env's time now: a resume after an API failure and its backoff mustn't carry on from the time the turn's prompt gave. A failed read leaves it as it is.""" + from agentenv_protocol.a2a_agent.env_clock import failure_reason, prompt_note + try: - return f"{prompt}\n\n{prompt_note(await read_env_clock(server))}" + return f"{prompt}\n\n{prompt_note(await env_clock.read())}" except Exception as e: logger.warning(f"env clock: no reading for the resume: {failure_reason(e)}") return prompt @@ -656,24 +661,20 @@ def _command(session_id: str, resume: bool, *, model: str | None, allowed_tools: cmd += ["--disallowedTools", ",".join(builtin_tools.disable)] cmd.append("--strict-mcp-config") if mcp_servers: - def entry(server: ClockServer) -> dict: - return {"type": "http", "url": server.url, **({"headers": server.headers} if server.headers else {})} - - servers = mcp_server_entries(mcp_servers, role) - cmd += ["--mcp-config", json.dumps({"mcpServers": {name: entry(s) for name, s in servers.items()}})] + def entry(info: dict) -> dict: + server: dict = {"type": "http", "url": info["url"]} + # The gateway filters an env's tools by the role header, which wins over a stored one. + headers = {**(info.get("headers") or {}), **({"AgentEnv-Role": role} if role else {})} + if headers: + server["headers"] = headers + return server + + cmd += ["--mcp-config", json.dumps({"mcpServers": {name: entry(info) for name, info in mcp_servers.items()}})] if env_clock: cmd += ["--settings", hook_settings()] return cmd -def mcp_server_entries(mcp_servers: Mapping[str, dict], role: str | None) -> dict[str, ClockServer]: - """Each MCP server's url and the headers the CLI sends it.""" - # The gateway filters an env's tools by the role header, which wins over a stored one. - return {name: ClockServer(url=info["url"], - headers={**(info.get("headers") or {}), **({"AgentEnv-Role": role} if role else {})}) - for name, info in mcp_servers.items()} - - async def _spawn_claude( prompt: str, session_id: str, @@ -692,7 +693,7 @@ async def _spawn_claude( max_thinking_tokens: int | None = None, builtin_tools: ClaudeCodeTools | None = None, event_callback: Callable[[dict], None] | None = None, - env_clock: ClockServer | None = None, + env_clock: EnvClock | None = None, ) -> ClaudeResult: cmd = _command(session_id, resume, model=model, allowed_tools=allowed_tools, mcp_servers=mcp_servers, system_prompt=system_prompt, output_format=output_format, role=role, effort=effort, @@ -700,7 +701,9 @@ async def _spawn_claude( env = cli_env(os.environ, model=model, project_id=project_id, task_id=task_id, max_thinking_tokens=max_thinking_tokens) if env_clock is not None: - env[ENV_CLOCK_SERVER_ENV] = env_clock.to_env() + from agentenv_protocol.a2a_agent.env_clock import ENV_CLOCK_SERVER_ENV + + env[ENV_CLOCK_SERVER_ENV] = env_clock.server.to_env() logger.info(f"Running claude CLI (resume={resume}, session_id={session_id}, model={model})") WORKSPACE_DIR.mkdir(parents=True, exist_ok=True) proc = await _started(asyncio.create_subprocess_exec( diff --git a/src/agent_env/agents/agents/claude-code/env_clock.py b/src/agent_env/agents/agents/claude-code/env_clock.py deleted file mode 100644 index 2cd70f48a..000000000 --- a/src/agent_env/agents/agents/claude-code/env_clock.py +++ /dev/null @@ -1,136 +0,0 @@ -"""The env's clock, read for the model. While agent-env's gateway runs an env's clock (urn:agentenv:clock/v1), it lists -``get_time`` on the env's MCP server, and the date the CLI states on its own ("Today's date is ...", from this -machine's clock) isn't the env's. So each prompt, a resume's too, carries the env's time, and a hook -(``env_clock_hook.py``) adds a fresh reading after each tool call, failed ones too: a clock moved between prompts, or -running faster than real time, never leaves the model on a stale date. Unarmed, no server lists ``get_time`` and -nothing is added: the env runs on real time, which is the CLI's date.""" - -from __future__ import annotations - -import asyncio -import json -import logging -import shlex -import sys -from collections.abc import Mapping -from dataclasses import dataclass, field -from pathlib import Path - -from mcp import ClientSession -from mcp.client.streamable_http import streamablehttp_client - -logger = logging.getLogger(__name__) - -ENV_CLOCK_TOOL_NAME = "get_time" -# The server whose get_time the hook reads, as {"url", "headers"} JSON in the CLI's environment while the clock is -# armed. The CLI's --mcp-config already carries the same url and headers on its command line. -ENV_CLOCK_SERVER_ENV = "AGENTENV_ENV_CLOCK_SERVER" -# How long a prompt waits for the clock before going without its note. -PROMPT_READ_TIMEOUT_S = 10.0 -# How long the hook waits for the clock, and how long the CLI lets the hook run, a little longer. -HOOK_READ_TIMEOUT_S = 5.0 -HOOK_TIMEOUT_S = 10 -HOOK_SCRIPT = Path(__file__).with_name("env_clock_hook.py") -# How much of a bad get_time answer an error quotes. -_ERROR_TEXT_LIMIT = 200 -# After every tool call, whether it succeeded or failed: a long command that fails takes as much of the clock as one -# that succeeds. -HOOK_EVENTS = ("PostToolUse", "PostToolUseFailure") - - -@dataclass(frozen=True) -class ClockServer: - url: str - headers: dict[str, str] = field(default_factory=dict) - - def to_env(self) -> str: - return json.dumps({"url": self.url, "headers": self.headers}) - - @classmethod - def from_env(cls, value: str) -> ClockServer: - raw = json.loads(value) - return cls(url=raw["url"], headers=dict(raw.get("headers") or {})) - - -async def _read(server: ClockServer, *, find: bool) -> str | None: - """The env's current time from ``server``'s get_time; None when ``find`` and the server doesn't list it.""" - async with streamablehttp_client(server.url, headers=server.headers or None) as (read, write, _): - async with ClientSession(read, write) as session: - await session.initialize() - if find: - cursor, listed = None, False - while not listed: - page = await session.list_tools(cursor=cursor) - listed = any(t.name == ENV_CLOCK_TOOL_NAME for t in page.tools) - cursor = page.nextCursor - if not cursor: - break - if not listed: - return None - result = await session.call_tool(ENV_CLOCK_TOOL_NAME, {}) - if result.isError: - raise ValueError(f"{ENV_CLOCK_TOOL_NAME} failed: {_text(result)[:_ERROR_TEXT_LIMIT]!r}") - if isinstance(result.structuredContent, dict) and result.structuredContent.get("current_time"): - return str(result.structuredContent["current_time"]) - text = _text(result) - current = json.loads(text).get("current_time") - if not current: - raise ValueError(f"no current_time in {text[:_ERROR_TEXT_LIMIT]!r}") - return str(current) - - -def failure_reason(e: BaseException) -> str: - """What failed: the first error inside the MCP client's task groups, which say only how many there were.""" - while isinstance(e, BaseExceptionGroup) and e.exceptions: - e = e.exceptions[0] - return f"{type(e).__name__}: {e}" - - -def _text(result) -> str: - return "\n".join(c.text for c in (result.content or []) if getattr(c, "text", None)) - - -async def find_env_clock(servers: Mapping[str, ClockServer]) -> tuple[ClockServer, str] | None: - """The server listing get_time and the env's time it read, trying ``servers`` in order; None when none lists it or - the read fails, which is logged and never fails the prompt.""" - - async def search() -> tuple[ClockServer, str] | None: - for name, server in servers.items(): - try: - current = await _read(server, find=True) - except Exception as e: # one server's failure is no reason to skip the others - logger.warning(f"env clock: reading {ENV_CLOCK_TOOL_NAME} from MCP server {name!r} failed: " - f"{failure_reason(e)}") - continue - if current: - return server, current - return None - - try: - return await asyncio.wait_for(search(), timeout=PROMPT_READ_TIMEOUT_S) - except TimeoutError: - logger.warning(f"env clock: no reading within {PROMPT_READ_TIMEOUT_S:g} s; the prompt goes without it") - return None - - -async def read_env_clock(server: ClockServer, timeout_s: float = HOOK_READ_TIMEOUT_S) -> str: - return await asyncio.wait_for(_read(server, find=False), timeout=timeout_s) - - -def prompt_note(current: str) -> str: - # Same stance as get_time's own description: this machine's clock is not this environment's. - return (f"The current date and time in this environment is {current} (UTC). The clock on the machine " - f"you are running on, and the date you were given with it, may not match it, so use this value, or call " - f"{ENV_CLOCK_TOOL_NAME} for a fresh one before acting on a date.") - - -def turn_note(current: str) -> str: - return f"Environment time now: {current} (UTC)." - - -def hook_settings() -> str: - """The CLI's --settings for an armed clock: the hook after each tool call, succeeded or failed. On the command line rather than in the - user settings, so a snapshot of the CLI's state never carries it to an agent with no clock or no such script.""" - command = f"{shlex.quote(sys.executable)} {shlex.quote(str(HOOK_SCRIPT))}" - entry = {"matcher": "", "hooks": [{"type": "command", "command": command, "timeout": HOOK_TIMEOUT_S}]} - return json.dumps({"hooks": {event: [entry] for event in HOOK_EVENTS}}) diff --git a/src/agent_env/agents/agents/claude-code/env_clock_hook.py b/src/agent_env/agents/agents/claude-code/env_clock_hook.py index c395474c2..9418908b4 100644 --- a/src/agent_env/agents/agents/claude-code/env_clock_hook.py +++ b/src/agent_env/agents/agents/claude-code/env_clock_hook.py @@ -1,22 +1,38 @@ -"""The CLI's PostToolUse and PostToolUseFailure hook while the env's clock is armed: the env's time now, as context -ending the tool's result (``env_clock``). Any failure prints nothing, so a clock blip never fails a tool call; the model can still call -get_time itself.""" +"""The CLI's PostToolUse and PostToolUseFailure hook while the task env's clock is armed: the env's time now, as context +ending the tool's result, read through the protocol's env clock (``agentenv_protocol.a2a_agent.env_clock``). Any +failure prints nothing, so a clock blip never fails a tool call; the model can still call get_time itself. + +``hook_settings`` is the CLI's --settings that runs it, on the command line rather than in the user settings, so a +snapshot of the CLI's state never carries it to an agent with no clock or no such script. The module loads only the +standard library until the hook runs.""" from __future__ import annotations import asyncio import json import os +import shlex import sys -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from env_clock import ( # noqa: E402 - ENV_CLOCK_SERVER_ENV, ENV_CLOCK_TOOL_NAME, HOOK_EVENTS, ClockServer, failure_reason, read_env_clock, turn_note, -) +# After every tool call, whether it succeeded or failed: a long command that fails takes as much of the clock. +HOOK_EVENTS = ("PostToolUse", "PostToolUseFailure") +# How long the CLI lets the hook run, a little longer than the protocol's read (READ_TIMEOUT_S). +HOOK_TIMEOUT_S = 10 +_GET_TIME = "get_time" + + +def hook_settings() -> str: + command = f"{shlex.quote(sys.executable)} {shlex.quote(os.path.abspath(__file__))}" + entry = {"matcher": "", "hooks": [{"type": "command", "command": command, "timeout": HOOK_TIMEOUT_S}]} + return json.dumps({"hooks": {event: [entry] for event in HOOK_EVENTS}}) def main(stdin: str, environ: dict[str, str]) -> str: """What the hook prints for the CLI's hook input ``stdin``: its JSON output, or nothing.""" + from agentenv_protocol.a2a_agent.env_clock import ( + ENV_CLOCK_SERVER_ENV, ClockServer, failure_reason, read_env_clock, turn_note, + ) + configured = environ.get(ENV_CLOCK_SERVER_ENV) if not configured: return "" @@ -25,11 +41,11 @@ def main(stdin: str, environ: dict[str, str]) -> str: except ValueError: given = {} tool = given.get("tool_name") or "" - # The output names the event the CLI ran the hook for, which it checks. - event = given.get("hook_event_name") if given.get("hook_event_name") in HOOK_EVENTS else HOOK_EVENTS[0] # A get_time call's own result already is the reading. - if tool == ENV_CLOCK_TOOL_NAME or tool.endswith(f"__{ENV_CLOCK_TOOL_NAME}"): + if tool == _GET_TIME or tool.endswith(f"__{_GET_TIME}"): return "" + # The output names the event the CLI ran the hook for, which it checks. + event = given.get("hook_event_name") if given.get("hook_event_name") in HOOK_EVENTS else HOOK_EVENTS[0] try: current = asyncio.run(read_env_clock(ClockServer.from_env(configured))) except Exception as e: diff --git a/src/agent_env/agents/agents/claude-code/install.sh b/src/agent_env/agents/agents/claude-code/install.sh index c17b0ba2e..ef53156bc 100755 --- a/src/agent_env/agents/agents/claude-code/install.sh +++ b/src/agent_env/agents/agents/claude-code/install.sh @@ -22,7 +22,7 @@ PYTHON_VERSION=3.12 START_TIMEOUT_S=120 STOP_TIMEOUT_S=10 CLAUDE_CODE_VERSION=$(sed -n 's/^ARG CLAUDE_CODE_VERSION=//p' "$here/Dockerfile") -AGENT_FILES="a2a_server.py claude_bridge.py env_clock.py env_clock_hook.py kit.py requirements.txt run.sh tar_changelog.py" +AGENT_FILES="a2a_server.py claude_bridge.py env_clock_hook.py kit.py requirements.txt run.sh tar_changelog.py" # put SRC DEST replaces DEST by a rename, so a binary an earlier install's agent still runs can be replaced. case $mode in diff --git a/tst/unit/agents/claude_code/conftest.py b/tst/unit/agents/claude_code/conftest.py index b58034ea8..e48d71450 100644 --- a/tst/unit/agents/claude_code/conftest.py +++ b/tst/unit/agents/claude_code/conftest.py @@ -29,11 +29,13 @@ def agent_dirs(monkeypatch, tmp_path): @pytest.fixture(autouse=True) def no_env_clock(monkeypatch): - """No env clock unless a test gives one: a turn otherwise asks each MCP server for get_time.""" - async def _none(servers): + """No env clock unless a test gives one: the framework otherwise asks each registered MCP server for get_time.""" + from agentenv_protocol.a2a_agent import framework + + async def _none(servers, timeout_s=None): return None - monkeypatch.setattr(a2a_server, "find_env_clock", _none) + monkeypatch.setattr(framework, "find_env_clock", _none) @pytest.fixture diff --git a/tst/unit/agents/claude_code/test_env_clock.py b/tst/unit/agents/claude_code/test_env_clock.py index d27d83cf1..b018b3992 100644 --- a/tst/unit/agents/claude_code/test_env_clock.py +++ b/tst/unit/agents/claude_code/test_env_clock.py @@ -1,6 +1,7 @@ -"""The env's clock reaches the model: while an MCP server lists get_time (the gateway does only while the env's clock is -armed), each prompt carries the env's time, and the CLI runs a PostToolUse hook that ends each tool call's result with -a fresh reading. Unarmed, nothing changes. Read here from real MCP servers over streamable HTTP.""" +"""Claude Code on the task env's clock. The framework ends each prompt with the env's time and gives the turn +``TaskRequest.env_clock`` (``agentenv_protocol.a2a_agent.env_clock``); this agent has the CLI run a PostToolUse and +PostToolUseFailure hook that ends each tool call's result with a fresh reading, and a resume after an API failure read +the clock again. Unarmed, nothing changes. The hook is run here as the CLI runs it, against a real MCP server.""" from __future__ import annotations @@ -16,131 +17,65 @@ import pytest import uvicorn +from agentenv_protocol.a2a_agent.env_clock import ENV_CLOCK_SERVER_ENV, ClockServer, EnvClock, prompt_note, turn_note from mcp.server.fastmcp import FastMCP from tst.unit.agents.claude_code.agent import AGENT_DIR, a2a_server, claude_bridge from tst.unit.agents.claude_code.support import FakeCLI -env_clock = importlib.import_module("env_clock") env_clock_hook = importlib.import_module("env_clock_hook") NOW = "2026-05-01T09:00:00Z" +CLOCK = EnvClock(server=ClockServer(url="http://gw/mcp", headers={"AgentEnv-Role": "agent"}), time=NOW) -def _serve(server: FastMCP) -> tuple[str, uvicorn.Server]: +@pytest.fixture(scope="module") +def gateway(): # every test using it asks for socket_enabled: it's read over this machine's loopback + """A gateway-like MCP server listing get_time.""" + server = FastMCP("env", stateless_http=True, json_response=True) + + @server.tool() + def get_time() -> dict: + return {"current_time": NOW} + sock = socket.socket() sock.bind(("127.0.0.1", 0)) - port = sock.getsockname()[1] running = uvicorn.Server(uvicorn.Config(server.streamable_http_app(), log_level="warning")) threading.Thread(target=lambda: asyncio.run(running.serve(sockets=[sock])), daemon=True).start() deadline = time.monotonic() + 10 while not running.started: assert time.monotonic() < deadline, "the MCP server didn't start" time.sleep(0.02) - return f"http://127.0.0.1:{port}/mcp", running - - -@pytest.fixture(scope="module") -def servers(): # every test using it asks for socket_enabled: they read it over this machine's loopback - """A gateway-like server listing get_time, and one without it.""" - gateway = FastMCP("env", stateless_http=True, json_response=True) - - @gateway.tool() - def get_time() -> dict: - return {"current_time": NOW} - - @gateway.tool() - def list_notes() -> list[str]: - return [] - - other = FastMCP("peers", stateless_http=True, json_response=True) - - @other.tool() - def peer_send_message(text: str) -> str: - return text - - started = [_serve(gateway), _serve(other)] - yield {"env": started[0][0], "peers": started[1][0]} - for _, running in started: - running.should_exit = True - - -def _entries(**urls: str) -> dict: - return {name: env_clock.ClockServer(url=url) for name, url in urls.items()} - - -def test_the_server_listing_get_time_is_found_and_read(socket_enabled, servers): - found = asyncio.run(env_clock.find_env_clock(_entries(peers=servers["peers"], env=servers["env"]))) - - assert found is not None - assert found[0].url == servers["env"] and found[1] == NOW - - -def test_unarmed_no_server_lists_get_time_and_nothing_is_found(socket_enabled, servers): - assert asyncio.run(env_clock.find_env_clock(_entries(peers=servers["peers"]))) is None - - -def test_a_server_that_fails_is_skipped_for_the_next(socket_enabled, servers): - sock = socket.socket() - sock.bind(("127.0.0.1", 0)) - closed = f"http://127.0.0.1:{sock.getsockname()[1]}/mcp" - sock.close() - - found = asyncio.run(env_clock.find_env_clock(_entries(gone=closed, env=servers["env"]))) - - assert found is not None and found[1] == NOW - - -def test_a_slow_clock_leaves_the_prompt_without_its_note(monkeypatch): - async def _hangs(server, *, find): - await asyncio.sleep(60) - - monkeypatch.setattr(env_clock, "_read", _hangs) - monkeypatch.setattr(env_clock, "PROMPT_READ_TIMEOUT_S", 0.05) - - assert asyncio.run(env_clock.find_env_clock(_entries(env="http://unused"))) is None - - -def test_the_mcp_servers_entries_carry_the_role_the_cli_sends(): - entries = claude_bridge.mcp_server_entries({"env": {"url": "http://gw/mcp", "headers": {"X": "1"}}}, "agent") + yield f"http://127.0.0.1:{sock.getsockname()[1]}/mcp" + running.should_exit = True - assert entries == {"env": env_clock.ClockServer(url="http://gw/mcp", headers={"X": "1", "AgentEnv-Role": "agent"})} - -def _turn(monkeypatch, agent, found) -> dict: - """One turn with ``find_env_clock`` answering ``found``; what run_claude was called with.""" +def _turn(monkeypatch, agent, clock: EnvClock | None) -> dict: + """One turn whose request carries ``clock``; what run_claude was called with.""" called: dict = {} - async def _find(servers): - called["servers"] = servers - return found - async def _run_claude(**kwargs): called.update(kwargs) return claude_bridge.ClaudeResult(response="ok", tool_call_count=0, session_id=kwargs["session_id"]) - monkeypatch.setattr(a2a_server, "find_env_clock", _find) monkeypatch.setattr(a2a_server, "run_claude", _run_claude) request = a2a_server.TaskRequest(task_id="t", context_id="c", parts=(a2a_server.TextPart(text="book it"),), - config=a2a_server.ClaudeCodeConfig(role="agent"), - mcp_servers={"env": {"url": "http://gw/mcp"}}) + config=a2a_server.ClaudeCodeConfig(), env_clock=clock) asyncio.run(agent.run(request)) return called -def test_an_armed_clock_puts_the_envs_time_in_the_prompt_and_the_hook_on_the_cli(monkeypatch, agent): - server = env_clock.ClockServer(url="http://gw/mcp", headers={"AgentEnv-Role": "agent"}) - called = _turn(monkeypatch, agent, (server, NOW)) +def test_an_armed_clock_reaches_the_cli(monkeypatch, agent): + assert _turn(monkeypatch, agent, CLOCK)["env_clock"] == CLOCK - assert called["servers"] == {"env": server} - assert called["prompt"].startswith("book it\n\n") - assert NOW in called["prompt"] and called["env_clock"] == server +def test_an_unarmed_clock_leaves_the_cli_alone(monkeypatch, agent): + assert _turn(monkeypatch, agent, None)["env_clock"] is None -def test_an_unarmed_clock_leaves_the_prompt_and_the_cli_alone(monkeypatch, agent): - called = _turn(monkeypatch, agent, None) - assert called["prompt"] == "book it" and called["env_clock"] is None +def test_the_agent_leaves_the_prompt_note_to_the_framework(): + # The framework ends the request's parts with the env's time; the agent adds none of its own. + assert a2a_server.ClaudeCodeAgent.env_clock_prompt_note is True def _command(env_clock_on: bool) -> list[str]: @@ -157,7 +92,28 @@ def test_the_hook_is_on_the_command_line_only_while_armed(): assert set(hooks) == {"PostToolUse", "PostToolUseFailure"} for entries in hooks.values(): [hook] = entries[0]["hooks"] - assert hook["command"].endswith("env_clock_hook.py") and hook["timeout"] == env_clock.HOOK_TIMEOUT_S + assert hook["command"].endswith("env_clock_hook.py") and hook["timeout"] == env_clock_hook.HOOK_TIMEOUT_S + + +def test_the_cli_gets_the_clocks_server_in_its_environment_only_while_armed(monkeypatch): + envs = [] + + async def _exec(*cmd, **kwargs): + envs.append(kwargs["env"]) + return FakeCLI((json.dumps({"type": "result", "subtype": "success", "result": "ok", + "session_id": "s"}) + "\n").encode()) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) + asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False)) + asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False, env_clock=CLOCK)) + + assert ENV_CLOCK_SERVER_ENV not in envs[0] + assert ClockServer.from_env(envs[1][ENV_CLOCK_SERVER_ENV]) == CLOCK.server + + +class _Clock(EnvClock): + async def read(self, timeout_s: float = 5.0) -> str: + return "2026-05-02T10:00:00Z" def test_a_resume_after_an_api_failure_carries_the_envs_time_now(monkeypatch, tmp_path): @@ -172,38 +128,16 @@ async def _exec(*cmd, **kwargs): procs.append(FakeCLI((json.dumps(answers.pop(0)) + "\n").encode(), 1 if answers else 0)) return procs[-1] - async def _read(server, timeout_s=env_clock.HOOK_READ_TIMEOUT_S): - return "2026-05-02T10:00:00Z" - monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) monkeypatch.setattr(claude_bridge, "SESSION_JSONL_DIR", tmp_path) (tmp_path / "s.jsonl").write_text('{"type":"assistant"}\n') monkeypatch.setattr(claude_bridge, "_RETRY_BACKOFF_S", (0, 0)) - monkeypatch.setattr(claude_bridge, "read_env_clock", _read) result = asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False, - env_clock=env_clock.ClockServer(url="http://gw/mcp"))) + env_clock=_Clock(server=CLOCK.server, time=NOW))) prompts = [proc.stdin.data.decode() for proc in procs] assert result.attempts == 2 and prompts[0] == "p" - assert prompts[1].startswith(claude_bridge._RESUME_PROMPT) - assert prompts[1].endswith(env_clock.prompt_note("2026-05-02T10:00:00Z")) - - -def test_the_cli_gets_the_clocks_server_in_its_environment_only_while_armed(monkeypatch): - envs = [] - - async def _exec(*cmd, **kwargs): - envs.append(kwargs["env"]) - return FakeCLI((json.dumps({"type": "result", "subtype": "success", "result": "ok", - "session_id": "s"}) + "\n").encode()) - - monkeypatch.setattr(asyncio, "create_subprocess_exec", _exec) - server = env_clock.ClockServer(url="http://gw/mcp", headers={"AgentEnv-Role": "agent"}) - asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False)) - asyncio.run(claude_bridge.run_claude(prompt="p", session_id="s", resume=False, env_clock=server)) - - assert env_clock.ENV_CLOCK_SERVER_ENV not in envs[0] - assert env_clock.ClockServer.from_env(envs[1][env_clock.ENV_CLOCK_SERVER_ENV]) == server + assert prompts[1] == f"{claude_bridge._RESUME_PROMPT}\n\n{prompt_note('2026-05-02T10:00:00Z')}" def _hook(stdin: dict, environ: dict) -> subprocess.CompletedProcess: @@ -214,27 +148,24 @@ def _hook(stdin: dict, environ: dict) -> subprocess.CompletedProcess: @pytest.mark.parametrize("event", ["PostToolUse", "PostToolUseFailure"]) -def test_the_hook_ends_a_tool_calls_result_with_the_envs_time(socket_enabled, servers, event): +def test_the_hook_ends_a_tool_calls_result_with_the_envs_time(socket_enabled, gateway, event): ran = _hook({"hook_event_name": event, "tool_name": "Bash"}, - {env_clock.ENV_CLOCK_SERVER_ENV: env_clock.ClockServer(url=servers["env"]).to_env()}) + {ENV_CLOCK_SERVER_ENV: ClockServer(url=gateway).to_env()}) assert ran.returncode == 0, ran.stderr output = json.loads(ran.stdout)["hookSpecificOutput"] - assert output == {"hookEventName": event, "additionalContext": env_clock.turn_note(NOW)} + assert output == {"hookEventName": event, "additionalContext": turn_note(NOW)} -@pytest.mark.parametrize("stdin, environ", [ - ({"tool_name": "Bash"}, {}), # unarmed: no server configured - ({"tool_name": "mcp__env__get_time"}, {"configured": True}), # its own result is the reading - ({"tool_name": "Bash"}, {"unreachable": True}), # a clock blip prints nothing +@pytest.mark.parametrize("stdin, configured", [ + ({"tool_name": "Bash"}, None), # unarmed: no server configured + ({"tool_name": "mcp__env__get_time"}, "gateway"), # its own result is the reading + ({"tool_name": "Bash"}, "http://127.0.0.1:9/mcp"), # a clock blip prints nothing ]) -def test_the_hook_prints_nothing_unarmed_after_get_time_or_on_a_failed_read(socket_enabled, servers, stdin, environ): - if environ.get("configured"): - environ = {env_clock.ENV_CLOCK_SERVER_ENV: env_clock.ClockServer(url=servers["env"]).to_env()} - elif environ.get("unreachable"): - environ = {env_clock.ENV_CLOCK_SERVER_ENV: env_clock.ClockServer(url="http://127.0.0.1:9/mcp").to_env()} - - ran = _hook(stdin, environ) +def test_the_hook_prints_nothing_unarmed_after_get_time_or_on_a_failed_read(socket_enabled, gateway, stdin, + configured): + url = gateway if configured == "gateway" else configured + ran = _hook(stdin, {ENV_CLOCK_SERVER_ENV: ClockServer(url=url).to_env()} if url else {}) assert ran.returncode == 0 and ran.stdout == "", (ran.stdout, ran.stderr)