diff --git a/src/agent_env/a2a_agent/protocol.py b/src/agent_env/a2a_agent/protocol.py index 8c939eb6..4199c5dd 100644 --- a/src/agent_env/a2a_agent/protocol.py +++ b/src/agent_env/a2a_agent/protocol.py @@ -34,6 +34,12 @@ class AgentUnreachableError(TimeoutError): _ERROR_CODE = re.compile(r"[a-z][a-z0-9_]{0,63}") +# Terminal ``error_code`` a harness sets when it stops the agent for exhausting +# its step/turn budget (as opposed to a genuine crash). Consumers treat it as an +# expected budget outcome rather than a retryable failure. The string is the +# cross-process contract — harnesses in other repos emit this exact value. +MAX_TURNS_ERROR_CODE = "max_turns_reached" + def raise_for_extension_status( response: httpx.Response, *, operation: str, include_body: bool = False diff --git a/src/agent_env/task_step/context.py b/src/agent_env/task_step/context.py index 8d635d3b..c60229a6 100644 --- a/src/agent_env/task_step/context.py +++ b/src/agent_env/task_step/context.py @@ -6,6 +6,7 @@ from dataclasses import dataclass, field from typing import Any +from agent_env.a2a_agent.protocol import MAX_TURNS_ERROR_CODE from agent_env.env.env import DeployedEnv from agent_env.providers.sandbox_providers.local_sandbox import LocalSandbox, host_url_for from agent_env.providers.sandbox_providers.sandbox_provider import reachable_url @@ -202,6 +203,16 @@ def from_dict(cls, data: dict) -> PromptResponse: structured_output=data.get("structured_output"), ) + @property + def is_gradable_despite_error(self) -> bool: + """True when this response carries an error that is nonetheless gradable. + + A max-turns/step budget exhaustion is an expected outcome, not a crash: + the partial trajectory recorded before the failure is still worth grading. + Verifiers skip responses with an ``error_type`` *unless* this is set. + """ + return self.error_code == MAX_TURNS_ERROR_CODE + @dataclass class TaskStepContext: 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 3765cf4a..392a72fe 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -14,6 +14,7 @@ from agent_env.a2a_agent import A2AAgent, conversation_store from agent_env.a2a_agent import protocol +from agent_env.a2a_agent.protocol import MAX_TURNS_ERROR_CODE from agent_env.a2a_agent.object_transfer import ( TrajectoryUpload, fetch_trajectory, @@ -694,6 +695,20 @@ def record_turn() -> None: structured_output=structured_output, )) + if final_state == TaskState.failed and final_terminal.error_code == MAX_TURNS_ERROR_CODE: + # Exhausting the step/turn budget is an expected outcome, not a + # retryable crash: re-running the prompt just burns another full + # budget and never grades. Return the context (the prompt_response + # with its trajectory was already recorded above) so the task flows + # on to grading instead of raising. + logger.warning( + "Agent '%s' hit its step budget (task_id=%s, error_code=%s); grading the " + "partial trajectory without retrying. %s", + self.agent_name, final_task_id, final_terminal.error_code, + agent_error_text(final_terminal.error_message, final_terminal.response_text), + ) + return context + if final_state == TaskState.failed: detail_parts = [] if final_terminal.error_type: diff --git a/src/agent_env/task_step/task_steps/verifiers/agent_prompt_response_verifier.py b/src/agent_env/task_step/task_steps/verifiers/agent_prompt_response_verifier.py index a4fb779f..e375b919 100644 --- a/src/agent_env/task_step/task_steps/verifiers/agent_prompt_response_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/agent_prompt_response_verifier.py @@ -83,7 +83,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: if prompt_response is None: raise RuntimeError(f"PromptResponse with prompt_id='{self.prompt_id}' not found in context") - if prompt_response.error_type: + if prompt_response.error_type and not prompt_response.is_gradable_despite_error: logger.warning( f"Skipping verification '{self.verifier_id}': prompt had error_type={prompt_response.error_type}" ) 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 7f8d5a3c..6180f1d9 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 @@ -457,7 +457,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: if prompt_response is None: raise RuntimeError(f"PromptResponse with prompt_id='{self.prompt_id}' not found in context") - if prompt_response.error_type: + if prompt_response.error_type and not prompt_response.is_gradable_despite_error: logger.warning(f"Skipping verification '{self.verifier_id}': prompt had error_type={prompt_response.error_type}") self._write_skipped_entries( context, "prompt_error", f"Skipped: prompt had error_type={prompt_response.error_type}") diff --git a/tst/unit/task_step/test_prompt_agent.py b/tst/unit/task_step/test_prompt_agent.py index 6f821ae7..6f5b8b21 100644 --- a/tst/unit/task_step/test_prompt_agent.py +++ b/tst/unit/task_step/test_prompt_agent.py @@ -27,6 +27,7 @@ from agent_env.task_step.context import DeployedAgent, TaskStepContext from agent_env.task_step.snapshot_utils import agent_state_capture from agent_env.task_step.task_steps import prompt_agent as pa +from agent_env.a2a_agent.protocol import MAX_TURNS_ERROR_CODE from agent_env.task_step.task_steps.prompt_agent import ( PromptAgentTaskStep, _duplicates_prompt_text, @@ -269,3 +270,141 @@ def trajectory(request: httpx.Request) -> httpx.Response: assert json.loads(store.get(uri)) == [{"type": "echo", "output": "Echo: hi"}] +def _serve_failed_terminal(monkeypatch, *, error_code): + """Serve an A2A agent whose task terminates in state=failed carrying error_code.""" + for fn in ("create_conversation", "add_a2a_task", "complete_a2a_task", + "mark_closed", "get_conversation"): + monkeypatch.setattr(pa.conversation_store, fn, lambda *a, **kw: None) + monkeypatch.setattr(pa, "fetch_container_logs", AsyncMock(return_value=None)) + + data = {"error_type": "agent_error", "error_message": "Max turns (500) reached without completion."} + if error_code is not None: + data["error_code"] = error_code + status = {"state": "failed", "message": {"parts": [ + {"kind": "text", "text": "Max turns (500) reached without completion."}, + {"kind": "data", "data": data}, + ]}} + + def answer(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + result = ( + {"id": "task-1", "contextId": "ctx"} + if body["method"] == "message/send" + else {"status": status} + ) + return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": result}) + + transport = httpx.MockTransport(answer) + real_client = httpx.AsyncClient + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **kw: real_client(*a, transport=transport, **kw)) + + +def _solver_context_and_step(): + context = TaskStepContext(instance_id="ti-1") + context.deployed_agents.append(DeployedAgent( + agent_name="solver", api_url="http://agent.test", a2a_url="http://agent.test", + a2a_card={"capabilities": {"extensions": []}}, + )) + step = PromptAgentTaskStep( + id="solve", version=None, prompt="hi", agent_name="solver", poll_interval_seconds=0, + ) + return context, step + + +@pytest.mark.asyncio +async def test_max_turns_terminal_grades_without_raising(monkeypatch): + # A max-steps terminal is an expected budget outcome: execute() must return + # the context (so the task flows on to grading) rather than raising, which + # would be retried into another full budget. + _serve_failed_terminal(monkeypatch, error_code=MAX_TURNS_ERROR_CODE) + context, step = _solver_context_and_step() + + result = await step.execute(context) + + assert result is context + assert result.prompt_responses[-1].error_code == MAX_TURNS_ERROR_CODE + + +@pytest.mark.asyncio +async def test_other_failed_terminal_still_raises(monkeypatch): + # A failure without the max-steps error_code is a real failure: still raise + # so Temporal retries it as before. + _serve_failed_terminal(monkeypatch, error_code=None) + context, step = _solver_context_and_step() + + with pytest.raises(RuntimeError, match="A2A task failed"): + await step.execute(context) + + +def _serve_failed_terminal_with_trajectory(monkeypatch, *, error_code, trajectory): + """Like _serve_failed_terminal, but also answers the trajectory extension so the + partial trajectory recorded before the failure can be fetched and stored.""" + for fn in ("create_conversation", "add_a2a_task", "complete_a2a_task", + "mark_closed", "get_conversation"): + monkeypatch.setattr(pa.conversation_store, fn, lambda *a, **kw: None) + monkeypatch.setattr(pa, "fetch_container_logs", AsyncMock(return_value=None)) + + data = {"error_type": "agent_error", "error_message": "Max turns (500) reached without completion."} + if error_code is not None: + data["error_code"] = error_code + status = {"state": "failed", "message": {"parts": [ + {"kind": "text", "text": "Max turns (500) reached without completion."}, + {"kind": "data", "data": data}, + ]}} + + def answer(request: httpx.Request) -> httpx.Response: + if request.url.path == "/ext/trajectory": + assert json.loads(request.content) == {"task_id": "task-1"} + return httpx.Response(200, json={"trajectory": trajectory}) + body = json.loads(request.content) + result = ( + {"id": "task-1", "contextId": "ctx"} + if body["method"] == "message/send" + else {"status": status} + ) + return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": result}) + + transport = httpx.MockTransport(answer) + real_client = httpx.AsyncClient + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **kw: real_client(*a, transport=transport, **kw)) + + +@pytest.mark.asyncio +async def test_max_turns_terminal_records_partial_trajectory(monkeypatch, tmp_path): + # Grading a budget-exhausted run is only worthwhile because of its partial + # trajectory: a failed max-turns terminal must still fetch and record the + # partial actions, and the returned response must point at the stored copy. + store = LocalFilesystemObjectStore(str(tmp_path)) + configure(object_store=store) + + async def sdk_get(request: TaskTrajectoryRequest | TaskObjectTrajectoryRequest): + return {"trajectory": []} + + extensions = build_registry( + object(), [enable(TRAJECTORY_V1)], sdk_handlers={(TRAJECTORY_V1.uri, "get"): sdk_get} + ).card_extensions() + + partial = [{"type": "action", "output": "click(1)"}, {"type": "action", "output": "type('x')"}] + _serve_failed_terminal_with_trajectory( + monkeypatch, error_code=MAX_TURNS_ERROR_CODE, trajectory=partial) + + context = TaskStepContext(instance_id="ti-1") + context.deployed_agents.append(DeployedAgent( + agent_name="solver", api_url="http://agent.test", a2a_url="http://agent.test", + a2a_card={"capabilities": {"extensions": extensions}}, + )) + step = PromptAgentTaskStep( + id="solve", version=None, prompt="hi", agent_name="solver", poll_interval_seconds=0, + trajectory_output_prefix=store.object_url("prompt_agent_trajectories"), + ) + + result = await step.execute(context) + + assert result is context + pr = result.prompt_responses[-1] + assert pr.error_code == MAX_TURNS_ERROR_CODE + uri = pr.agent_trajectory_object_url + assert uri is not None + assert json.loads(store.get(uri)) == partial + + diff --git a/tst/unit/task_step/verifiers/test_max_turns_grading.py b/tst/unit/task_step/verifiers/test_max_turns_grading.py new file mode 100644 index 00000000..99f47b7a --- /dev/null +++ b/tst/unit/task_step/verifiers/test_max_turns_grading.py @@ -0,0 +1,61 @@ +"""A max-turns budget exhaustion records an ``error_type``, but its partial work is +still gradable. Verifiers skip responses with an ``error_type`` *unless* the error is +a ``MAX_TURNS_ERROR_CODE`` budget outcome — these tests pin that carve-out so a +budget-exhausted run reaches grading instead of being silently skipped.""" + +import pytest + +from agent_env.a2a_agent.protocol import MAX_TURNS_ERROR_CODE +from agent_env.task_step.context import PromptResponse, TaskStepContext +from agent_env.task_step.task_steps.verifiers.agent_prompt_response_verifier import ( + AgentPromptResponseVerifierTaskStep, +) + + +def test_is_gradable_despite_error_only_for_max_turns(): + assert PromptResponse( + prompt_id="p1", response="x", error_type="agent_error", error_code=MAX_TURNS_ERROR_CODE, + ).is_gradable_despite_error + assert not PromptResponse( + prompt_id="p1", response="x", error_type="agent_error", error_code="some_crash", + ).is_gradable_despite_error + assert not PromptResponse(prompt_id="p1", response="x").is_gradable_despite_error + + +def _context_with_response(**kw) -> TaskStepContext: + ctx = TaskStepContext(instance_id="ti-1") + ctx.prompt_responses.append(PromptResponse(prompt_id="p1", response="the answer is 42", **kw)) + return ctx + + +def _verifier() -> AgentPromptResponseVerifierTaskStep: + return AgentPromptResponseVerifierTaskStep( + id="v", version=None, prompt_id="p1", verifier_id="vid", + criteria=[{"type": "response_contains", "needles": ["42"]}], + ) + + +@pytest.mark.asyncio +async def test_max_turns_error_reaches_grading(): + # A budget-exhausted response carries error_type but is gradable: the verifier + # must evaluate its criteria against the partial response, not skip it. + ctx = _context_with_response(error_type="agent_error", error_code=MAX_TURNS_ERROR_CODE) + + result = await _verifier().execute(ctx) + + entry = result.metadata["verifications"]["vid"] + assert entry["results"][0].get("id") != "prompt_error" # not the skipped sentinel + assert entry["results"][0]["result"] is True + assert entry["score"] == 1.0 + + +@pytest.mark.asyncio +async def test_other_error_type_still_skips_grading(): + # A real crash (any other error_code) must still short-circuit to a skip. + ctx = _context_with_response(error_type="agent_error", error_code="some_crash") + + result = await _verifier().execute(ctx) + + entry = result.metadata["verifications"]["vid"] + assert entry["results"][0]["id"] == "prompt_error" + assert entry["score"] == 0