[FD-3805] feat(prompt_agent): grade on max-steps terminal instead of retrying

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

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:

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:

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}"
             )

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}")

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
+
+

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