diff --git a/src/agent_env/task/task.py b/src/agent_env/task/task.py index 69d002d74..111c36c09 100644 --- a/src/agent_env/task/task.py +++ b/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): diff --git a/src/agent_env/task_step/__init__.py b/src/agent_env/task_step/__init__.py index dd249d93b..b10a72e4f 100644 --- a/src/agent_env/task_step/__init__.py +++ b/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", diff --git a/src/agent_env/task_step/context.py b/src/agent_env/task_step/context.py index 082a3d736..9891ec23a 100644 --- a/src/agent_env/task_step/context.py +++ b/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, 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 915fca57e..990253ce8 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 @@ -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}" + ) diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_fail_task_when_not_passing.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_fail_task_when_not_passing.py new file mode 100644 index 000000000..88183a577 --- /dev/null +++ b/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