feat(rubrics_verifier): fail_task_when_not_passing fails the task on a verdict that isn't passing

src/agent_env/task/task.py

@@ -25,7 +25,7 @@
 from agent_env.store.routing import run_scope
 from agent_env.task_step.preflight import pinned
 from agent_env.task.teardown import env_record_ids
-from agent_env.task_step.context import TaskStepContext, regraft_redacted_keys
+from agent_env.task_step.context import TaskStepContext, VerifierNotPassingError, regraft_redacted_keys
 from agent_env.task_step.context_ops import ContextUpdateOps, build_context_update_ops
 from agent_env.task_step.task_step import RetryConfig, TaskStep
 
@@ -731,7 +731,8 @@ async def _drain_in_flight() -> set[str]:
                             first_error, first_error_step_id = exc, step_id
                         continue
                     retry_cfg = getattr(step_obj, "retry_config", None)
-                    if retry_cfg is not None and retry_counts.get(step_id, 0) < retry_cfg.max_retries:
+                    if (retry_cfg is not None and retry_counts.get(step_id, 0) < retry_cfg.max_retries
+                            and not isinstance(exc, VerifierNotPassingError)):
                         retryable.append((step_obj, retry_cfg, exc))
                         continue
                     if not getattr(step_obj, "fail_task_on_error", True):

src/agent_env/task_step/__init__.py

@@ -1,7 +1,7 @@
 """TaskStep module for AgentEnv."""
 
 from agent_env.env.env import DeployedEnv
-from .context import DeployedAgent, PromptResponse, StepContextError, TaskStepContext
+from .context import DeployedAgent, PromptResponse, StepContextError, TaskStepContext, VerifierNotPassingError
 from .task_step import TaskStep
 from .task_steps import AddSkillsTaskStep, AgentPromptResponseVerifierTaskStep, BuildMcpCliTaskStep, CombineUniverseVerdictsStep, DeployAgentTaskStep, DeployEnvTaskStep, EnvOutcomeVerifierTaskStep, LoadArtifactTaskStep, ModifyEnvToolAccessStep, PromptAgentTaskStep, RubricsVerifierTaskStep, ScoreAggregator, ValidationGateAggregatorStep, VerifyCoreA2AProtocolStep, VerifyCoreEnvironmentProtocolStep, VerifyEnvironmentCardStep, VerifyMCPEnvAssessmentStep, VerifyMCPToolSchemaTaskStep, VerifySpecConformanceTaskStep, VerifyUniverseLoadExportRoundtripStep, VerifyA2AAgentCardStep, VerifyA2AAgentMCPStep, VerifyA2ASkillConfigStep, VerifyA2ATrajectoryStep
 from .task_steps.verifiers.judge_utils.judge_output_format import JudgeOutputFormat, get_judge_output_format_spec
@@ -16,6 +16,7 @@
     "ScoreAggregator",
     "StepContextError",
     "TaskStepContext",
+    "VerifierNotPassingError",
     "TaskStep",
     "AddSkillsTaskStep",
     "AgentPromptResponseVerifierTaskStep",

src/agent_env/task_step/context.py

@@ -20,6 +20,11 @@ class StepContextError(RuntimeError):
     something it would create is already there. Running the step again on the same context fails the same
     way, so a runner should not retry it."""
 
+
+class VerifierNotPassingError(RuntimeError):
+    """A verifier set to fail its task when its verdict isn't passing recorded one that isn't. Grading the same run
+    again could pass it, so a runner should not retry it."""
+
 _REDACTED_KEYS = {
     "litellm_api_key", "judge_litellm_api_key", "usersim_api_key", "remote_tokens", "cf_access_client_secret",
     HEARTBEAT_KEY,

src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py

@@ -20,7 +20,13 @@
 from agent_env.config import get_config
 from agent_env.config.model import ModelParam
 from agent_env.providers.sandbox_providers.sandbox import into_container
-from agent_env.task_step.context import PromptResponse, StepContextError, TaskStepContext, dual_keyed
+from agent_env.task_step.context import (
+    PromptResponse,
+    StepContextError,
+    TaskStepContext,
+    VerifierNotPassingError,
+    dual_keyed,
+)
 from agent_env.task_step.task_steps.verifiers.judge_utils.judge_output_format import (
     SCREENSHOT_GROUNDING,
     JudgeOutputFormat,
@@ -456,8 +462,16 @@ def __init__(
         judge_batch_concurrency: Optional[int] = None,
         depends_on: Optional[list[TaskStepDependency]] = None,
         fail_task_on_error: bool = True,
+        fail_task_when_not_passing: bool = False,
     ):
         super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
+        if fail_task_when_not_passing and not fail_task_on_error:
+            raise ValueError(
+                "fail_task_when_not_passing needs fail_task_on_error: a tolerated failure lets the steps after it "
+                "run and score the run."
+            )
+        # A verdict below 1.0 fails the step with VerifierNotPassingError once the verdict is recorded.
+        self.fail_task_when_not_passing = fail_task_when_not_passing
         # RUN_CONTEXT_PARTS the judge also sees, cut at grade_up_to_turn like the trajectory.
         self.run_context_for_judge = _validated_run_context_parts(run_context_for_judge)
         # Split the criteria into batches of this many, each graded by its own judge call (up to
@@ -591,6 +605,8 @@ def to_dict(self) -> dict:
             base["judge_batch_size"] = self.judge_batch_size
         if self.judge_batch_concurrency is not None:
             base["judge_batch_concurrency"] = self.judge_batch_concurrency
+        if self.fail_task_when_not_passing:
+            base["fail_task_when_not_passing"] = True
         return base
 
     @classmethod
@@ -621,6 +637,7 @@ def from_dict(cls, data: dict) -> RubricsVerifierTaskStep:
             run_context_for_judge=data.get("run_context_for_judge"),
             judge_batch_size=data.get("judge_batch_size"),
             judge_batch_concurrency=data.get("judge_batch_concurrency"),
+            fail_task_when_not_passing=data.get("fail_task_when_not_passing", False),
         )
 
     def _resolved_criteria(self, context: TaskStepContext) -> list[dict]:
@@ -634,6 +651,12 @@ def _resolved_criteria(self, context: TaskStepContext) -> list[dict]:
         return list(step_overrides["criteria"]) if "criteria" in step_overrides else self.criteria
 
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
+        context = await self._grade(context)
+        if self.fail_task_when_not_passing:
+            _raise_unless_passing(self.verifier_id, context.metadata["verifications"][self.verifier_id])
+        return context
+
+    async def _grade(self, context: TaskStepContext) -> TaskStepContext:
         prompt_response = next((pr for pr in context.prompt_responses if pr.prompt_id == self.prompt_id), None)
         if prompt_response is None:
             raise StepContextError(f"PromptResponse with prompt_id='{self.prompt_id}' not found in context")
@@ -1588,3 +1611,13 @@ def _build_loaded_artifacts_block(entries: list[dict]) -> str:
             "Do not rely solely on the agent's response text — verify against the actual files when possible.",
         ]
         return "\n".join(lines)
+
+
+def _raise_unless_passing(verifier_id: str, entry: dict) -> None:
+    """Raise VerifierNotPassingError when the recorded verdict scores below 1.0, naming the rows that weren't met."""
+    if entry["score"] >= 1:
+        return
+    unmet = [str(row.get("id")) for row in entry["results"] if row.get("score", 1.0 if row.get("result") else 0.0) < 1]
+    raise VerifierNotPassingError(
+        f"Verifier '{verifier_id}' isn't passing (score {entry['score']}); rows not met: {unmet}"
+    )

tst/unit/task_step/verifiers/test_rubrics_verifier_fail_task_when_not_passing.py

@@ -0,0 +1,154 @@
+"""fail_task_when_not_passing: a verifier whose verdict isn't passing (score below 1.0) fails its step, and so its
+task, once the verdict is recorded; the score itself is computed as before."""
+from __future__ import annotations
+
+import pytest
+
+from agent_env.runner.runner import RunStatus, run_outcome
+from agent_env.task import Task
+from agent_env.task_step.context import PromptResponse, TaskStepContext, VerifierNotPassingError
+from agent_env.task_step.task_step import RetryConfig
+from agent_env.task_step.task_steps.verifiers.aggregate_verifiers import AggregateVerifiersTaskStep
+from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep
+
+CRITERIA = [{"id": "world", "description": "the events fired"}, {"id": "replies", "description": "everyone replied"}]
+
+
+def _verifier(**kw) -> RubricsVerifierTaskStep:
+    params = dict(
+        id="validity", version=1, criteria=CRITERIA, prompt_id="p1", use_agent_judge=False, use_trajectory=True,
+        verifier_id="validity", fail_task_when_not_passing=True,
+    )
+    params.update(kw)
+    return RubricsVerifierTaskStep(**params)
+
+
+def _context(**prompt) -> TaskStepContext:
+    return TaskStepContext(prompt_responses=[PromptResponse(
+        prompt_id="p1", response="done", prompt_text="do it", agent_trajectory_object_url="s3://t/1.json", **prompt)])
+
+
+def _judge(monkeypatch, verifier, met: dict[str, bool]) -> None:
+    async def fake_run(**kwargs):
+        rows = [{**c, "score": 1.0 if met[c["id"]] else 0.0, "result": met[c["id"]]} for c in verifier.criteria]
+        return rows, 0, [], None
+
+    monkeypatch.setattr(verifier, "_read_trajectory_text", lambda uri: "EVENTS")
+    monkeypatch.setattr(verifier, "_run_judge_with_output_retries", fake_run)
+
+
+@pytest.mark.asyncio
+async def test_a_passing_verdict_leaves_the_step_successful(monkeypatch):
+    v = _verifier()
+    _judge(monkeypatch, v, {"world": True, "replies": True})
+    ctx = _context()
+
+    await v.execute(ctx)
+
+    assert ctx.metadata["verifications"]["validity"]["score"] == 1.0
+
+
+@pytest.mark.asyncio
+async def test_an_unmet_row_fails_the_step_naming_it_and_keeps_the_verdict(monkeypatch):
+    v = _verifier()
+    _judge(monkeypatch, v, {"world": False, "replies": True})
+    ctx = _context()
+
+    not_passing = r"Verifier 'validity' isn't passing \(score 0.0\); rows not met: \['world'\]"
+    with pytest.raises(VerifierNotPassingError, match=not_passing):
+        await v.execute(ctx)
+
+    assert [r["result"] for r in ctx.metadata["verifications"]["validity"]["results"]] == [False, True]
+
+
+@pytest.mark.asyncio
+async def test_a_partial_weighted_score_is_not_passing(monkeypatch):
+    v = _verifier(score_aggregator="weighted_average")
+    _judge(monkeypatch, v, {"world": True, "replies": False})
+
+    with pytest.raises(VerifierNotPassingError, match=r"\(score 0.5\); rows not met: \['replies'\]"):
+        await v.execute(_context())
+
+
+@pytest.mark.asyncio
+async def test_a_penalized_negative_row_is_not_passing(monkeypatch):
+    v = _verifier(score_aggregator="weighted_average",
+                  criteria=[CRITERIA[0], {"id": "penalty", "description": "a negative-weight row", "weight": -1}])
+    _judge(monkeypatch, v, {"world": True, "penalty": False})
+
+    with pytest.raises(VerifierNotPassingError, match=r"rows not met: \['penalty'\]"):
+        await v.execute(_context())
+
+
+@pytest.mark.asyncio
+async def test_a_run_the_judge_could_not_grade_is_not_passing(monkeypatch):
+    v = _verifier()
+    _judge(monkeypatch, v, {"world": True, "replies": True})
+    ctx = _context(error_type="timeout")
+
+    with pytest.raises(VerifierNotPassingError, match=r"rows not met: \['prompt_error'\]"):
+        await v.execute(ctx)
+
+    assert ctx.metadata["verifications"]["validity"]["score"] == 0
+
+
+@pytest.mark.asyncio
+async def test_without_the_option_a_failing_verdict_is_only_recorded(monkeypatch):
+    v = _verifier(fail_task_when_not_passing=False)
+    _judge(monkeypatch, v, {"world": False, "replies": False})
+    ctx = _context()
+
+    await v.execute(ctx)
+
+    assert ctx.metadata["verifications"]["validity"]["score"] == 0.0
+
+
+def test_the_option_needs_fail_task_on_error():
+    problem = "fail_task_when_not_passing needs fail_task_on_error"
+    with pytest.raises(ValueError, match=problem):
+        _verifier(fail_task_on_error=False)
+    with pytest.raises(ValueError, match=problem):
+        RubricsVerifierTaskStep.from_dict(
+            {**_verifier(fail_task_when_not_passing=False).to_dict(), "fail_task_on_error": False,
+             "fail_task_when_not_passing": True})
+
+
+def test_the_option_is_stored_only_when_set():
+    assert RubricsVerifierTaskStep.from_dict(_verifier().to_dict()).fail_task_when_not_passing is True
+    assert "fail_task_when_not_passing" not in _verifier(fail_task_when_not_passing=False).to_dict()
+
+
+@pytest.mark.asyncio
+async def test_a_task_fails_at_the_verifier_and_never_scores(monkeypatch, local_stores):
+    v = _verifier()
+    _judge(monkeypatch, v, {"world": False, "replies": True})
+    final = AggregateVerifiersTaskStep(id="final-score", version=None, verifier_ids=["validity"],
+                                       verifier_id="final-score", depends_on=["validity"])
+    ctx = _context()
+
+    with pytest.raises(VerifierNotPassingError) as caught:
+        await Task(id="t", version=None, steps=[v, final]).run(context=ctx)
+
+    assert run_outcome(ctx, caught.value)[0] is RunStatus.FAILED
+    assert [f["step_id"] for f in ctx.metadata["failed_steps"]] == ["validity"]
+    assert "final-score" not in ctx.metadata["verifications"]
+
+
+@pytest.mark.asyncio
+async def test_a_retry_config_does_not_grade_the_run_again(monkeypatch, local_stores):
+    v = _verifier()
+    v.retry_config = RetryConfig(retry_from_step_id="validity", max_retries=2)
+    _judge(monkeypatch, v, {"world": False, "replies": True})
+    graded = []
+    grade = v._grade
+
+    async def counting_grade(context):
+        graded.append(context)
+        return await grade(context)
+
+    monkeypatch.setattr(v, "_grade", counting_grade)
+
+    with pytest.raises(VerifierNotPassingError):
+        await Task(id="t", version=None, steps=[v]).run(context=_context())
+
+    assert len(graded) == 1