From 87fa4b4439869a9a06b494e97e0f1f31c3c8cd93 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Tue, 6 Oct 2026 14:09:30 -0700 Subject: [PATCH 1/4] fix(a2a): give up on an agent whose sandbox died, and cancel a prompt that timed out A prompt to an agent whose sandbox died waited out its whole timeout (20 minutes in practice) before failing, and a prompt that timed out left the agent working on it, spending tokens, until its sandbox went away. - poll_a2a_task takes the agent's sandbox_id. With it, once the agent has gone 60 s without answering, over at least 3 polls in a row, it raises AgentUnreachableError naming the sandbox. Transport errors and 502/503/504 count as no answer; any other reply, errors included, is the agent answering. A tunnel blip drops a poll or two and is ridden out. AgentUnreachableError is a TimeoutError, so callers handle it as they handled the same dead agent before, only sooner. - Out of time, poll_a2a_task sends tasks/cancel before raising. Best effort: a cancel that fails is logged and the timeout is still what's raised. - prompt_agent passes the solver's sandbox and a deployed user-sim's; a human keeps the full timeout. rubrics_verifier passes its judge's. - The agent framework implements tasks/cancel: it marks the task canceled and the SDK cancels run(), which sees CancelledError. It answered UnsupportedOperation before, so no agent stopped on a cancel. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/agentenv-protocol/README.md | 5 + .../agentenv_protocol/a2a_agent/framework.py | 10 +- .../agentenv-protocol/tests/test_a2a_agent.py | 74 ++++++++ src/agent_env/a2a_agent/protocol.py | 73 ++++++-- .../task_step/task_steps/prompt_agent.py | 6 +- .../task_steps/verifiers/rubrics_verifier.py | 1 + tst/unit/a2a_agent/protocol_test.py | 163 ++++++++++++++++++ .../test_prompt_agent_human_default.py | 2 +- .../task_step/test_prompt_agent_snapshots.py | 2 +- .../test_prompt_agent_unreachable.py | 91 ++++++++++ 10 files changed, 406 insertions(+), 21 deletions(-) create mode 100644 tst/unit/task_step/test_prompt_agent_unreachable.py diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index 0cc5bcad..fb719636 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -301,6 +301,11 @@ A successful `TaskResult` must contain at least one text, file, or data part; the framework rejects empty successes rather than emitting an ungradeable task. The SDK does not retry tasks. +A client that gives up on a task sends `tasks/cancel`. The framework marks the +task canceled and cancels `run()`, which gets `asyncio.CancelledError` at its +next `await`. A process `run()` started keeps running unless `run()` stops it, +so kill it before re-raising. + `enable(..., description="...")` is reserved for declarations carrying configuration or metadata. It preserves the agent-specific extension prose published in the Agent Card. The versioned SDK definition provides a generic 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 0ca4b06b..500aaad5 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/framework.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/a2a_agent/framework.py @@ -1226,12 +1226,12 @@ def _validate_status_event(cls, item: Any, request: TaskRequest[Any]) -> None: "streaming TaskStatusUpdateEvent must be non-terminal and final=False" ) - async def cancel(self, _context: Any, _event_queue: Any) -> None: - """Satisfy the upstream executor interface without offering cancellation.""" - from a2a.types import UnsupportedOperationError - from a2a.utils.errors import ServerError + async def cancel(self, context: Any, event_queue: Any) -> None: + """Mark the task canceled. The SDK then cancels the coroutine running it, so ``agent.run`` gets + ``CancelledError`` at its next ``await`` and should stop whatever it started, such as a CLI process.""" + from a2a.server.tasks import TaskUpdater - raise ServerError(error=UnsupportedOperationError()) + await TaskUpdater(event_queue, context.task_id, context.context_id).cancel() def _from_a2a_part(part: Any) -> Any: diff --git a/packages/agentenv-protocol/tests/test_a2a_agent.py b/packages/agentenv-protocol/tests/test_a2a_agent.py index 812ee352..2483651f 100644 --- a/packages/agentenv-protocol/tests/test_a2a_agent.py +++ b/packages/agentenv-protocol/tests/test_a2a_agent.py @@ -2908,3 +2908,77 @@ async def shadow(self): with pytest.raises(ValueError, match=rf"conflicts with {owner} route"): Agent().create_app() + + +def _send_in_context(client: TestClient, text: str, message_id: str) -> str: + response = client.post( + "/a2a", + json={ + "jsonrpc": "2.0", + "id": message_id, + "method": "message/send", + "params": { + "message": { + "kind": "message", + "messageId": message_id, + "role": "user", + "contextId": "context-1", + "parts": [{"kind": "text", "text": text}], + }, + "configuration": {"blocking": False}, + }, + }, + ) + return response.json()["result"]["id"] + + +def _task_rpc(client: TestClient, method: str, task_id: str) -> dict[str, Any]: + return client.post( + "/a2a", + json={"jsonrpc": "2.0", "id": method, "method": method, "params": {"id": task_id}}, + ).json() + + +def _within(seconds: float, condition) -> bool: + deadline = time.monotonic() + seconds + while not condition(): + if time.monotonic() > deadline: + return False + time.sleep(0.01) + return True + + +def test_cancel_stops_a_running_task_and_frees_its_context() -> None: + ran: list[str] = [] + cancelled: list[str] = [] + + @a2a_agent(identity=AgentIdentity(name="slow", description="test", version="1")) + class Agent(AgentEnvAgent): + async def run(self, request: TaskRequest) -> TaskResult: + text = request.parts[0].text + ran.append(text) + if text == "hang": + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.append(text) + raise + return TaskResult.text(f"did {text}") + + def state(client: TestClient, task_id: str) -> str: + return _task_rpc(client, "tasks/get", task_id)["result"]["status"]["state"] + + with TestClient(Agent().create_app()) as client: + hung = _send_in_context(client, "hang", "message-1") + assert _within(5, lambda: ran == ["hang"]) + + canceled = _task_rpc(client, "tasks/cancel", hung) + assert canceled["result"]["status"]["state"] == "canceled" + + # The next task in the same context runs only once the canceled one let go of it. + following = _send_in_context(client, "next", "message-2") + assert _within(5, lambda: state(client, following) == "completed") + assert state(client, hung) == "canceled" + + assert cancelled == ["hang"] + assert ran == ["hang", "next"] diff --git a/src/agent_env/a2a_agent/protocol.py b/src/agent_env/a2a_agent/protocol.py index e15393c2..9955d05a 100644 --- a/src/agent_env/a2a_agent/protocol.py +++ b/src/agent_env/a2a_agent/protocol.py @@ -18,6 +18,19 @@ TaskState.completed, TaskState.failed, TaskState.canceled, TaskState.rejected, }) +# How long an agent on a sandbox may go without answering a poll before ``poll_a2a_task`` gives up on it, over at +# least _UNREACHABLE_MIN_POLLS polls in a row: a sandbox that died answers none, a tunnel blip drops one or two. +UNREACHABLE_AFTER_SECONDS = 60 +_UNREACHABLE_MIN_POLLS = 3 +# What a proxy in front of the agent answers when it can't reach it; any other status is the agent answering. +_GATEWAY_STATUSES = frozenset({502, 503, 504}) +_CANCEL_TIMEOUT_SECONDS = 10 + + +class AgentUnreachableError(TimeoutError): + """The agent stopped answering, as one whose sandbox died does. A ``TimeoutError``, so whatever handles an + agent running out of time handles this the same way, only sooner.""" + _ERROR_CODE = re.compile(r"[a-z][a-z0-9_]{0,63}") @@ -134,10 +147,17 @@ async def poll_a2a_task( task_id: str, timeout_seconds: int, poll_interval_seconds: int = 10, + *, + sandbox_id: Optional[str] = None, ) -> dict: - """POST /a2a tasks/get until status.state is 'completed' or 'failed'. Returns the result dict.""" + """POST /a2a tasks/get until the task reaches a terminal state, and return it. + + Out of time, it asks the agent to cancel the task, so it stops working on it, and raises ``TimeoutError``. + ``sandbox_id`` names the sandbox the agent runs on: then it raises ``AgentUnreachableError``, naming it, once + the agent has gone UNREACHABLE_AFTER_SECONDS without answering, rather than waiting out the timeout.""" deadline = time.monotonic() + timeout_seconds consecutive_failures = 0 + answered_at, unanswered, last_failure = time.monotonic(), 0, "" while time.monotonic() < deadline: backoff = min(poll_interval_seconds + consecutive_failures * 5, 60) await asyncio.sleep(backoff) @@ -149,30 +169,59 @@ async def poll_a2a_task( }, timeout=30) resp.raise_for_status() data = resp.json() - if "error" in data: - consecutive_failures += 1 - logger.warning(f"A2A poll returned error (consec={consecutive_failures}, will retry): {data['error']}") - continue - result = data["result"] except httpx.HTTPStatusError as e: if 400 <= e.response.status_code < 500: raise consecutive_failures += 1 logger.warning(f"A2A poll got {e.response.status_code} (consec={consecutive_failures}, will retry)") - continue + if e.response.status_code in _GATEWAY_STATUSES: + unanswered, last_failure = unanswered + 1, f"HTTP {e.response.status_code}" + else: + answered_at, unanswered = time.monotonic(), 0 except httpx.HTTPError as e: consecutive_failures += 1 logger.warning(f"A2A poll failed (consec={consecutive_failures}, will retry): {type(e).__name__}: {e}") + unanswered, last_failure = unanswered + 1, f"{type(e).__name__}: {e}" + else: + answered_at, unanswered = time.monotonic(), 0 + if "error" in data: + consecutive_failures += 1 + logger.warning(f"A2A poll returned error (consec={consecutive_failures}, will retry): {data['error']}") + continue + consecutive_failures = 0 + if data["result"]["status"]["state"] in _TERMINAL_TASK_STATES: + return data["result"] continue + silent = time.monotonic() - answered_at + if sandbox_id is not None and unanswered >= _UNREACHABLE_MIN_POLLS and silent >= UNREACHABLE_AFTER_SECONDS: + raise AgentUnreachableError( + f"The agent on sandbox {sandbox_id} stopped answering: {unanswered} polls in a row over " + f"{silent:.0f}s got no reply (last: {last_failure})") - consecutive_failures = 0 - state = result["status"]["state"] - if state in _TERMINAL_TASK_STATES: - return result - + await cancel_a2a_task(a2a_url, task_id) raise TimeoutError(f"A2A task {task_id} did not complete within {timeout_seconds}s") +async def cancel_a2a_task(a2a_url: str, task_id: str) -> None: + """POST /a2a tasks/cancel, so the agent stops working on ``task_id``. Best effort: an agent that can't be + reached, or won't cancel, is only logged.""" + try: + async with httpx.AsyncClient() as client: + resp = await client.post(f"{a2a_url}/a2a", json={ + "jsonrpc": "2.0", "id": "cancel", "method": "tasks/cancel", + "params": {"id": task_id}, + }, timeout=_CANCEL_TIMEOUT_SECONDS) + resp.raise_for_status() + error = resp.json().get("error") + except Exception as e: # nothing here may replace the timeout being reported + logger.warning(f"Couldn't cancel A2A task {task_id}: {type(e).__name__}: {e}") + return + if error: + logger.warning(f"The agent didn't cancel A2A task {task_id}: {error}") + else: + logger.info(f"Canceled A2A task {task_id}") + + @dataclass(frozen=True) class TerminalResponse: """What's pulled from an A2A terminal status message: the text reply, the typed diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index 321454a2..af14884b 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -508,6 +508,7 @@ async def _execute_conversation( ) result = await protocol.poll_a2a_task( target_url, sent_task_id, self.timeout_seconds, self.poll_interval_seconds, + sandbox_id=agent.sandbox_id, ) target_state = result["status"]["state"] status_msg = (result.get("status") or {}).get("message") or {} @@ -579,9 +580,10 @@ async def _execute_conversation( user_result = await protocol.poll_a2a_task( user_url, sent_user_task_id, self.user_agent_timeout_seconds, self.poll_interval_seconds, + sandbox_id=user_sim.sandbox_id if is_user_sim else None, ) - except TimeoutError: - logger.warning(f"user_a2a_url timeout for conversation {conversation_id}; marking abandoned") + except TimeoutError as e: + logger.warning(f"user_a2a_url timeout for conversation {conversation_id} ({e}); marking abandoned") conversation_store.mark_closed(conversation_id) break diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py index 56377509..48e8a2a1 100644 --- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py @@ -1022,6 +1022,7 @@ async def _invoke_judge_a2a( task_id, timeout_seconds=self.judge_timeout_seconds, poll_interval_seconds=self.DEFAULT_POLL_INTERVAL_SECONDS, + sandbox_id=getattr(judge_agent, "sandbox_id", None), ) state = result["status"]["state"] status_msg = (result.get("status") or {}).get("message") or {} diff --git a/tst/unit/a2a_agent/protocol_test.py b/tst/unit/a2a_agent/protocol_test.py index 9973da6e..2595d20e 100644 --- a/tst/unit/a2a_agent/protocol_test.py +++ b/tst/unit/a2a_agent/protocol_test.py @@ -4,12 +4,20 @@ telemetry and — when the agent ran with output_format — the typed structured_output. """ +import json +import logging +from types import SimpleNamespace + import httpx import pytest +from agent_env.a2a_agent import protocol from agent_env.a2a_agent.protocol import ( + UNREACHABLE_AFTER_SECONDS, + AgentUnreachableError, TerminalResponse, extract_terminal_response, + poll_a2a_task, raise_for_extension_status, ) from agentenv_protocol.a2a_agent import TaskResult, Usage @@ -182,3 +190,158 @@ def test_extract_terminal_response_shim_returns_legacy_5_tuple(): {"kind": "data", "data": {"tool_call_count": 2, "error_type": "E"}}, ]} assert extract_terminal_response(msg) == ("hi", 2, "E", None, None) + + +class _Agent: + """An agent behind a mock transport, on a fake clock: ``answer(method, request, agent)`` replies to each + JSON-RPC call, and sleeping only moves the clock.""" + + def __init__(self, monkeypatch, answer): + self.now = 0.0 + self.calls: list[str] = [] + self._answer = answer + transport = httpx.MockTransport(self._handle) + client = httpx.AsyncClient + monkeypatch.setattr(protocol.httpx, "AsyncClient", lambda **kw: client(transport=transport, **kw)) + monkeypatch.setattr(protocol, "time", SimpleNamespace(monotonic=lambda: self.now)) + monkeypatch.setattr(protocol, "asyncio", SimpleNamespace(sleep=self._sleep)) + + async def _sleep(self, seconds): + self.now += seconds + + def _handle(self, request): + method = json.loads(request.content)["method"] + self.calls.append(method) + return self._answer(method, request, self) + + +def _task(state): + return httpx.Response(200, json={"jsonrpc": "2.0", "id": "poll", "result": {"id": "t-1", "status": {"state": state}}}) + + +def _disconnected(request, agent): + raise httpx.RemoteProtocolError("Server disconnected without sending a response.", request=request) + + +def _refused(request, agent): + raise httpx.ConnectError("Connection refused", request=request) + + +def _hung(request, agent): + agent.now += 30 + raise httpx.ReadTimeout("timed out", request=request) + + +def _bad_gateway(request, agent): + return httpx.Response(502) + + +def _working_then(failure): + def answer(method, request, agent): + return _task("working") if agent.calls.count("tasks/get") == 1 else failure(request, agent) + return answer + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure, last", [ + (_disconnected, "RemoteProtocolError"), (_refused, "ConnectError"), (_hung, "ReadTimeout"), (_bad_gateway, "HTTP 502"), +]) +async def test_an_agent_whose_sandbox_died_is_given_up_on_naming_it(monkeypatch, failure, last): + agent = _Agent(monkeypatch, _working_then(failure)) + + with pytest.raises(AgentUnreachableError, match=rf"The agent on sandbox sb-1 stopped answering: .*\(last: {last}") as raised: + await poll_a2a_task("http://agent", "t-1", 1200, 2, sandbox_id="sb-1") + + assert isinstance(raised.value, TimeoutError) # handled wherever running out of time is, only sooner + assert agent.now <= 2 * UNREACHABLE_AFTER_SECONDS + assert "tasks/cancel" not in agent.calls + + +@pytest.mark.asyncio +async def test_a_blip_shorter_than_the_window_is_ridden_out(monkeypatch): + def answer(method, request, agent): + if 10 < agent.now < 50: + _disconnected(request, agent) + return _task("completed" if agent.now > 50 else "working") + + agent = _Agent(monkeypatch, answer) + + result = await poll_a2a_task("http://agent", "t-1", 1200, 2, sandbox_id="sb-1") + + assert result["status"]["state"] == "completed" + + +@pytest.mark.asyncio +async def test_it_takes_three_unanswered_polls_however_far_apart(monkeypatch): + agent = _Agent(monkeypatch, _working_then(_refused)) + + with pytest.raises(AgentUnreachableError): + await poll_a2a_task("http://agent", "t-1", 1200, 60, sandbox_id="sb-1") + + assert agent.calls.count("tasks/get") == 4 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("reply", [ + httpx.Response(500), + httpx.Response(200, json={"jsonrpc": "2.0", "id": "poll", "error": {"code": -32603, "message": "boom"}}), +], ids=["http-500", "json-rpc-error"]) +async def test_an_agent_answering_with_errors_is_waited_on_until_the_timeout(monkeypatch, reply): + agent = _Agent(monkeypatch, lambda method, request, agent: reply if method == "tasks/get" else _task("canceled")) + + with pytest.raises(TimeoutError, match="did not complete within 600s") as raised: + await poll_a2a_task("http://agent", "t-1", 600, 2, sandbox_id="sb-1") + + assert not isinstance(raised.value, AgentUnreachableError) + assert agent.now >= 600 + + +@pytest.mark.asyncio +async def test_without_a_sandbox_a_silent_agent_is_waited_on_until_the_timeout(monkeypatch): + agent = _Agent(monkeypatch, lambda method, request, agent: _refused(request, agent)) + + with pytest.raises(TimeoutError, match="did not complete within 600s") as raised: + await poll_a2a_task("http://agent", "t-1", 600, 2) + + assert type(raised.value) is TimeoutError + assert agent.now >= 600 + + +@pytest.mark.asyncio +async def test_running_out_of_time_cancels_the_task(monkeypatch): + cancelled = [] + + def answer(method, request, agent): + if method == "tasks/cancel": + cancelled.append(json.loads(request.content)["params"]) + return _task("canceled") + return _task("working") + + agent = _Agent(monkeypatch, answer) + + with pytest.raises(TimeoutError, match="did not complete within 60s"): + await poll_a2a_task("http://agent", "t-1", 60, 2) + + assert cancelled == [{"id": "t-1"}] + assert agent.calls[-1] == "tasks/cancel" + + +def _unsupported(request, agent): + return httpx.Response(200, json={ + "jsonrpc": "2.0", "id": "cancel", "error": {"code": -32004, "message": "This operation is not supported"}}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("refusal, logged", [ + (_unsupported, "The agent didn't cancel A2A task t-1"), + (lambda request, agent: httpx.Response(500), "Couldn't cancel A2A task t-1: HTTPStatusError"), + (_refused, "Couldn't cancel A2A task t-1: ConnectError"), +], ids=["unsupported", "http-500", "unreachable"]) +async def test_a_cancel_that_fails_still_reports_the_timeout(monkeypatch, caplog, refusal, logged): + _Agent(monkeypatch, lambda method, request, agent: refusal(request, agent) if method == "tasks/cancel" else _task("working")) + + with caplog.at_level(logging.WARNING, logger=protocol.__name__): + with pytest.raises(TimeoutError, match="did not complete within 60s"): + await poll_a2a_task("http://agent", "t-1", 60, 2) + + assert logged in caplog.text diff --git a/tst/unit/task_step/test_prompt_agent_human_default.py b/tst/unit/task_step/test_prompt_agent_human_default.py index 9692fb73..a31b1072 100644 --- a/tst/unit/task_step/test_prompt_agent_human_default.py +++ b/tst/unit/task_step/test_prompt_agent_human_default.py @@ -20,7 +20,7 @@ def _wire(monkeypatch, cfg): async def send(url, parts, message_id, context_id, timeout): return "task-1", None - async def poll(url, task_id, timeout, interval): + async def poll(url, task_id, timeout, interval, **_): return { "status": { "state": TaskState.completed, diff --git a/tst/unit/task_step/test_prompt_agent_snapshots.py b/tst/unit/task_step/test_prompt_agent_snapshots.py index 4beef2c6..a66287a3 100644 --- a/tst/unit/task_step/test_prompt_agent_snapshots.py +++ b/tst/unit/task_step/test_prompt_agent_snapshots.py @@ -1270,7 +1270,7 @@ def __init__(self, monkeypatch, *, user_done_at: int | None = None): async def send(url, parts, message_id, wire_context_id, timeout_seconds): return f"{url}#{message_id}", None - async def poll(url, task_id, timeout_seconds, poll_interval_seconds): + async def poll(url, task_id, timeout_seconds, poll_interval_seconds, **_): if url == self.USER_URL: self.user_turns += 1 done = self.user_done_at == self.user_turns diff --git a/tst/unit/task_step/test_prompt_agent_unreachable.py b/tst/unit/task_step/test_prompt_agent_unreachable.py new file mode 100644 index 00000000..52306fcd --- /dev/null +++ b/tst/unit/task_step/test_prompt_agent_unreachable.py @@ -0,0 +1,91 @@ +"""prompt_agent gives up on an agent once its sandbox stops answering, naming the sandbox, but waits on a human.""" + +from unittest.mock import MagicMock + +import pytest +from a2a.types import TaskState + +from agent_env.a2a_agent.protocol import AgentUnreachableError +from agent_env.task_step.context import DeployedAgent, TaskStepContext +from agent_env.task_step.task_steps import prompt_agent as pa_mod +from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep + +_USER_SIM = DeployedAgent( + agent_name="human_agent", api_url="http://user-sim", a2a_url="http://user-sim", a2a_card={}, sandbox_id="sb-user", +) + + +def _completed(text): + return {"status": {"state": TaskState.completed, "message": {"parts": [{"kind": "text", "text": text}]}}} + + +def _wire(monkeypatch, poll): + cfg = MagicMock() + cfg.get_model_params.return_value = {} + cfg.get_default_human_a2a_url.return_value = "http://hub/a2a/human" + monkeypatch.setattr(pa_mod, "get_config", lambda: cfg) + for fn in ("create_conversation", "add_a2a_task", "complete_a2a_task", "mark_closed", "get_conversation"): + monkeypatch.setattr(pa_mod.conversation_store, fn, lambda *a, **kw: None) + + async def send(url, parts, message_id, context_id, timeout): + return f"{url}#task", None + + monkeypatch.setattr(pa_mod.protocol, "send_a2a_message", send) + monkeypatch.setattr(pa_mod.protocol, "poll_a2a_task", poll) + + +def _context(*agents): + ctx = TaskStepContext(instance_id="ti-1") + ctx.deployed_agents.append(DeployedAgent( + agent_name="solver", api_url="http://solver", a2a_url="http://solver", a2a_card={}, + sandbox_id="sb-solver", sandbox_type="modal", + )) + ctx.deployed_agents.extend(agents) + return ctx + + +@pytest.mark.asyncio +async def test_a_solver_that_stops_answering_fails_the_step_naming_its_sandbox(monkeypatch): + async def poll(url, task_id, timeout, interval, *, sandbox_id=None): + raise AgentUnreachableError(f"The agent on sandbox {sandbox_id} stopped answering") + + _wire(monkeypatch, poll) + step = PromptAgentTaskStep(id="solve", version=None, prompt="hi", agent_name="solver") + + with pytest.raises(AgentUnreachableError, match="sandbox sb-solver"): + await step.execute(_context()) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("user_sim, user_url, watched", [ + (_USER_SIM, "http://user-sim", "sb-user"), + (None, "http://hub/a2a/human", None), +], ids=["deployed-user-sim", "human"]) +async def test_a_user_sim_on_a_sandbox_is_watched_but_a_human_is_not(monkeypatch, user_sim, user_url, watched): + polls = [] + + async def poll(url, task_id, timeout, interval, *, sandbox_id=None): + polls.append((url, sandbox_id)) + return _completed("and then?") + + _wire(monkeypatch, poll) + step = PromptAgentTaskStep(id="solve", version=None, prompt="hi", agent_name="solver", max_conversation_turns=2) + + await step.execute(_context(*([user_sim] if user_sim else []))) + + assert polls == [("http://solver", "sb-solver"), (user_url, watched), ("http://solver", "sb-solver")] + + +@pytest.mark.asyncio +async def test_a_user_sim_that_stops_answering_ends_the_conversation_not_the_run(monkeypatch): + async def poll(url, task_id, timeout, interval, *, sandbox_id=None): + if url == "http://user-sim": + raise AgentUnreachableError(f"The agent on sandbox {sandbox_id} stopped answering") + return _completed("solver answer") + + _wire(monkeypatch, poll) + step = PromptAgentTaskStep(id="solve", version=None, prompt="hi", agent_name="solver", max_conversation_turns=3) + + result = await step.execute(_context(_USER_SIM)) + + assert result.prompt_responses[-1].response == "solver answer" From 3d4e4f0f8ceb7002bc261bd958f2fff18269e2bb Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Tue, 6 Oct 2026 14:16:08 -0700 Subject: [PATCH 2/4] fix(a2a): leave an empty error message out of the unreachable error Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/a2a_agent/protocol.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent_env/a2a_agent/protocol.py b/src/agent_env/a2a_agent/protocol.py index 9955d05a..afe7db47 100644 --- a/src/agent_env/a2a_agent/protocol.py +++ b/src/agent_env/a2a_agent/protocol.py @@ -181,7 +181,7 @@ async def poll_a2a_task( except httpx.HTTPError as e: consecutive_failures += 1 logger.warning(f"A2A poll failed (consec={consecutive_failures}, will retry): {type(e).__name__}: {e}") - unanswered, last_failure = unanswered + 1, f"{type(e).__name__}: {e}" + unanswered, last_failure = unanswered + 1, f"{type(e).__name__}: {e}" if str(e) else type(e).__name__ else: answered_at, unanswered = time.monotonic(), 0 if "error" in data: From 5ec660caeb88e89f4ab21bc789858c6f1bb88989 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Tue, 6 Oct 2026 14:16:59 -0700 Subject: [PATCH 3/4] fix(a2a): catch only the failures a cancel can meet, and drop test comments Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/agentenv-protocol/tests/test_a2a_agent.py | 1 - src/agent_env/a2a_agent/protocol.py | 2 +- tst/unit/a2a_agent/protocol_test.py | 2 +- 3 files changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/agentenv-protocol/tests/test_a2a_agent.py b/packages/agentenv-protocol/tests/test_a2a_agent.py index 2483651f..51e6c594 100644 --- a/packages/agentenv-protocol/tests/test_a2a_agent.py +++ b/packages/agentenv-protocol/tests/test_a2a_agent.py @@ -2975,7 +2975,6 @@ def state(client: TestClient, task_id: str) -> str: canceled = _task_rpc(client, "tasks/cancel", hung) assert canceled["result"]["status"]["state"] == "canceled" - # The next task in the same context runs only once the canceled one let go of it. following = _send_in_context(client, "next", "message-2") assert _within(5, lambda: state(client, following) == "completed") assert state(client, hung) == "canceled" diff --git a/src/agent_env/a2a_agent/protocol.py b/src/agent_env/a2a_agent/protocol.py index afe7db47..5cab6307 100644 --- a/src/agent_env/a2a_agent/protocol.py +++ b/src/agent_env/a2a_agent/protocol.py @@ -213,7 +213,7 @@ async def cancel_a2a_task(a2a_url: str, task_id: str) -> None: }, timeout=_CANCEL_TIMEOUT_SECONDS) resp.raise_for_status() error = resp.json().get("error") - except Exception as e: # nothing here may replace the timeout being reported + except (httpx.HTTPError, ValueError) as e: logger.warning(f"Couldn't cancel A2A task {task_id}: {type(e).__name__}: {e}") return if error: diff --git a/tst/unit/a2a_agent/protocol_test.py b/tst/unit/a2a_agent/protocol_test.py index 2595d20e..dcbb897a 100644 --- a/tst/unit/a2a_agent/protocol_test.py +++ b/tst/unit/a2a_agent/protocol_test.py @@ -252,7 +252,7 @@ async def test_an_agent_whose_sandbox_died_is_given_up_on_naming_it(monkeypatch, with pytest.raises(AgentUnreachableError, match=rf"The agent on sandbox sb-1 stopped answering: .*\(last: {last}") as raised: await poll_a2a_task("http://agent", "t-1", 1200, 2, sandbox_id="sb-1") - assert isinstance(raised.value, TimeoutError) # handled wherever running out of time is, only sooner + assert isinstance(raised.value, TimeoutError) assert agent.now <= 2 * UNREACHABLE_AFTER_SECONDS assert "tasks/cancel" not in agent.calls From c2d9063168d42003ef4488b4e59ae677f612f1d6 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Tue, 6 Oct 2026 14:30:20 -0700 Subject: [PATCH 4/4] fix(a2a): ask an agent given up on as unreachable to cancel too An agent cut off for over a minute, rather than dead, would otherwise keep working once it can be reached again. poll_a2a_task now sends tasks/cancel whenever it stops waiting on a task. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/a2a_agent/protocol.py | 8 +++++--- tst/unit/a2a_agent/protocol_test.py | 24 ++++++++++++++++++++++-- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/src/agent_env/a2a_agent/protocol.py b/src/agent_env/a2a_agent/protocol.py index 5cab6307..8c939eb6 100644 --- a/src/agent_env/a2a_agent/protocol.py +++ b/src/agent_env/a2a_agent/protocol.py @@ -152,9 +152,10 @@ async def poll_a2a_task( ) -> dict: """POST /a2a tasks/get until the task reaches a terminal state, and return it. - Out of time, it asks the agent to cancel the task, so it stops working on it, and raises ``TimeoutError``. - ``sandbox_id`` names the sandbox the agent runs on: then it raises ``AgentUnreachableError``, naming it, once - the agent has gone UNREACHABLE_AFTER_SECONDS without answering, rather than waiting out the timeout.""" + Out of time, it raises ``TimeoutError``. ``sandbox_id`` names the sandbox the agent runs on: then it raises + ``AgentUnreachableError``, naming it, once the agent has gone UNREACHABLE_AFTER_SECONDS without answering, + rather than waiting out the timeout. Either way it first asks the agent to cancel the task, so an agent still + working on it stops.""" deadline = time.monotonic() + timeout_seconds consecutive_failures = 0 answered_at, unanswered, last_failure = time.monotonic(), 0, "" @@ -194,6 +195,7 @@ async def poll_a2a_task( continue silent = time.monotonic() - answered_at if sandbox_id is not None and unanswered >= _UNREACHABLE_MIN_POLLS and silent >= UNREACHABLE_AFTER_SECONDS: + await cancel_a2a_task(a2a_url, task_id) raise AgentUnreachableError( f"The agent on sandbox {sandbox_id} stopped answering: {unanswered} polls in a row over " f"{silent:.0f}s got no reply (last: {last_failure})") diff --git a/tst/unit/a2a_agent/protocol_test.py b/tst/unit/a2a_agent/protocol_test.py index dcbb897a..70bd4261 100644 --- a/tst/unit/a2a_agent/protocol_test.py +++ b/tst/unit/a2a_agent/protocol_test.py @@ -199,6 +199,7 @@ class _Agent: def __init__(self, monkeypatch, answer): self.now = 0.0 self.calls: list[str] = [] + self.first_called_at: dict[str, float] = {} self._answer = answer transport = httpx.MockTransport(self._handle) client = httpx.AsyncClient @@ -212,6 +213,7 @@ async def _sleep(self, seconds): def _handle(self, request): method = json.loads(request.content)["method"] self.calls.append(method) + self.first_called_at.setdefault(method, self.now) return self._answer(method, request, self) @@ -253,8 +255,26 @@ async def test_an_agent_whose_sandbox_died_is_given_up_on_naming_it(monkeypatch, await poll_a2a_task("http://agent", "t-1", 1200, 2, sandbox_id="sb-1") assert isinstance(raised.value, TimeoutError) - assert agent.now <= 2 * UNREACHABLE_AFTER_SECONDS - assert "tasks/cancel" not in agent.calls + assert agent.first_called_at["tasks/cancel"] <= 2 * UNREACHABLE_AFTER_SECONDS + assert agent.calls[-1] == "tasks/cancel" + + +@pytest.mark.asyncio +async def test_an_agent_given_up_on_is_asked_to_cancel_in_case_it_is_still_working(monkeypatch): + cancelled = [] + + def answer(method, request, agent): + if method == "tasks/cancel": + cancelled.append(json.loads(request.content)["params"]) + return _task("canceled") + return _task("working") if agent.calls.count("tasks/get") == 1 else _refused(request, agent) + + _Agent(monkeypatch, answer) + + with pytest.raises(AgentUnreachableError): + await poll_a2a_task("http://agent", "t-1", 1200, 2, sandbox_id="sb-1") + + assert cancelled == [{"id": "t-1"}] @pytest.mark.asyncio