From 17a79f0450e757eb7a5369eae26a97b4f22e0b96 Mon Sep 17 00:00:00 2001 From: anonymous Date: Thu, 9 Jul 2026 15:01:48 -0700 Subject: [PATCH 1/7] feat(early-stop): config surface, live-verdict contract, resolution guardrails (inert) Add the opt-in surface for early-stop-on-criterion and its resolution-time guardrails, all inert at runtime (nothing invokes an interrupt yet): - run_limits.stop_early (master opt-in) and per-criterion stop_when (pass/fail/decided) config fields. - BaseCriterion.live_verdict + live_stop_polarities observability contract; overrides on the two mid-run-observable criteria (skill_triggered's first-engagement policy, command_executed's monotone min/max rules, with a _matching_commands helper shared with _check_impl). - Agent.supports_cooperative_stop capability flag (True on ClaudeCodeAgent). - validate_early_stop(): rejects every unsupported arming (non-Claude agent, no armed criterion, unobservable criterion, unsupported polarity, simulation mode) as a hard error, wired into resolve_all_tasks, plan, and _setup. --- src/coder_eval/agent.py | 9 +- src/coder_eval/agents/claude_code_agent.py | 6 +- src/coder_eval/cli/plan_command.py | 8 + src/coder_eval/criteria/base.py | 36 ++- src/coder_eval/criteria/command_executed.py | 135 +++++--- src/coder_eval/criteria/skill_triggered.py | 70 +++- src/coder_eval/models/criteria.py | 12 + src/coder_eval/models/limits.py | 11 + src/coder_eval/orchestration/early_stop.py | 110 +++++++ src/coder_eval/orchestration/experiment.py | 8 + src/coder_eval/orchestrator.py | 6 + tests/test_early_stop.py | 339 ++++++++++++++++++++ 12 files changed, 708 insertions(+), 42 deletions(-) create mode 100644 src/coder_eval/orchestration/early_stop.py create mode 100644 tests/test_early_stop.py diff --git a/src/coder_eval/agent.py b/src/coder_eval/agent.py index c5a9e001..8c3dc509 100644 --- a/src/coder_eval/agent.py +++ b/src/coder_eval/agent.py @@ -5,7 +5,7 @@ import logging from abc import ABC, abstractmethod -from typing import Any, NoReturn, Protocol +from typing import Any, ClassVar, NoReturn, Protocol from .errors import AgentCrashError, TurnTimeoutError from .errors.agent import format_timeout_reason, truncate_crash_message @@ -71,6 +71,13 @@ def __init__(self, config: ClaudeCodeAgentConfig, ...): _iteration: int = 0 _iteration_was_incremented: bool = False + # Capability flag: whether this agent honors the cooperative ``should_stop`` + # interrupt threaded through ``communicate()`` (early-stop-on-criterion). + # Default False — arming early-stop on an agent that does not set this True + # is rejected at resolution time. Concrete agents that check ``should_stop`` + # between messages override it to True. + supports_cooperative_stop: ClassVar[bool] = False + def _begin_turn(self) -> None: """Mark the start of a ``communicate()`` turn: reset the pending slot and bump the iteration counter so a mid-turn failure can be rolled back. diff --git a/src/coder_eval/agents/claude_code_agent.py b/src/coder_eval/agents/claude_code_agent.py index eb5b506e..c6c62855 100644 --- a/src/coder_eval/agents/claude_code_agent.py +++ b/src/coder_eval/agents/claude_code_agent.py @@ -10,7 +10,7 @@ from contextlib import suppress from datetime import datetime, timedelta from pathlib import Path -from typing import Any +from typing import Any, ClassVar from claude_agent_sdk import ( ClaudeAgentOptions, @@ -633,6 +633,10 @@ def finalize(self, status: AgentEndStatus, *, crashed: bool = False, crash_reaso class ClaudeCodeAgent(Agent[ClaudeCodeAgentConfig]): """Implementation of the Agent interface for Claude Code using the SDK.""" + # The Claude message loop has a between-messages guard where the cooperative + # ``should_stop`` check runs, so this agent supports early-stop-on-criterion. + supports_cooperative_stop: ClassVar[bool] = True + def __init__( self, config: ClaudeCodeAgentConfig, diff --git a/src/coder_eval/cli/plan_command.py b/src/coder_eval/cli/plan_command.py index fa612d09..55778c92 100644 --- a/src/coder_eval/cli/plan_command.py +++ b/src/coder_eval/cli/plan_command.py @@ -59,6 +59,7 @@ def plan_command( check_api_keys() # Lazy import to avoid circular dependency at module level + from ..orchestration.early_stop import EarlyStopConfigError, validate_early_stop from ..orchestration.experiment import DEFAULT_EXPERIMENT_PATH, load_experiment, resolve_task_for_variant # Always load experiment (defaults to experiments/default.yaml) @@ -133,10 +134,17 @@ def plan_command( for variant in exp_def.variants: try: resolved, _lineage, _ = resolve_task_for_variant(default_exp, task, exp_def, variant) + # Early-stop guardrails (no-op unless run_limits.stop_early is armed). + validate_early_stop(resolved) agent_type = str(resolved.agent.type) if resolved.agent else "unknown" agent_model = resolved.agent.model if resolved.agent else None model_str = f" ({agent_model})" if agent_model else "" console.print(f" [dim]Variant '{variant.variant_id}': {agent_type}{model_str}[/dim]") + except EarlyStopConfigError as e: + # A hard config error (unlike generic per-variant resolution + # failures, which stay soft): flip the plan exit code. + console.print(f" [red]Variant '{variant.variant_id}': early-stop config error - {e}[/red]") + all_valid = False except Exception as e: console.print(f" [red]Variant '{variant.variant_id}': resolution failed - {e}[/red]") diff --git a/src/coder_eval/criteria/base.py b/src/coder_eval/criteria/base.py index ac07f3c5..e2c4e2dd 100644 --- a/src/coder_eval/criteria/base.py +++ b/src/coder_eval/criteria/base.py @@ -8,7 +8,7 @@ from collections.abc import Callable from dataclasses import dataclass from functools import wraps -from typing import TYPE_CHECKING, Any, ClassVar +from typing import TYPE_CHECKING, Any, ClassVar, Literal from coder_eval.errors import JudgeInfrastructureError from coder_eval.models import BaseSuccessCriterion, CriterionAggregate, CriterionResult @@ -23,6 +23,10 @@ logger = logging.getLogger(__name__) +# A criterion's verdict from a PARTIAL, mid-run trajectory (early-stop observability). +# "undecided" means the outcome is not yet knowable from the events seen so far. +LiveVerdict = Literal["pass", "fail", "undecided"] + @dataclass(frozen=True) class CheckContext: @@ -124,6 +128,13 @@ def _check_impl( # Subclasses MUST define this as a class variable criterion_type: ClassVar[str] + # Which polarities this criterion can decide from a PARTIAL, mid-run trajectory. + # Empty (base default) = not observable mid-run, so it can never arm early-stop. + # A subclass that reads only turn_records and can decide mid-run declares the + # polarities it supports (e.g. frozenset({"pass", "fail"})) AND overrides + # live_verdict; CE025 enforces that the two stay consistent. + live_stop_polarities: ClassVar[frozenset[str]] = frozenset() + @handle_criterion_errors def check( self, @@ -191,6 +202,29 @@ def _check_impl( """ pass + def live_verdict( + self, + criterion: C, + turn_records: list["TurnRecord"], + ) -> LiveVerdict: + """Decide this criterion from a PARTIAL, mid-run trajectory (early-stop). + + Reads ONLY ``turn_records`` — a live verdict, by definition, may not peek + at the finished sandbox (that would invite end-state peeking), so there is + no ``sandbox`` parameter. Returns ``"pass"``/``"fail"`` only when the + outcome is already knowable from the events seen so far, else + ``"undecided"``. + + This only *triggers* an early stop; the authoritative scores always come + from ``check()``/``_check_impl`` run on the frozen trajectory after the + stop, so a live/final divergence can never corrupt scoring. + + Base default: ``"undecided"`` (not observable mid-run). Subclasses that + override this MUST also declare a non-empty ``live_stop_polarities`` (and + vice versa) — enforced by lint rule CE025. + """ + return "undecided" + def aggregate( self, criterion: C, diff --git a/src/coder_eval/criteria/command_executed.py b/src/coder_eval/criteria/command_executed.py index df40114d..733b94c4 100644 --- a/src/coder_eval/criteria/command_executed.py +++ b/src/coder_eval/criteria/command_executed.py @@ -3,14 +3,15 @@ import json import logging import re -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, ClassVar -from coder_eval.criteria.base import BaseCriterion, CheckContext, register_criterion +from coder_eval.criteria.base import BaseCriterion, CheckContext, LiveVerdict, register_criterion from coder_eval.models import CommandExecutedCriterion, CriterionResult if TYPE_CHECKING: from coder_eval.models.results import TurnRecord + from coder_eval.models.telemetry import CommandTelemetry from coder_eval.sandbox import Sandbox logger = logging.getLogger(__name__) @@ -29,6 +30,98 @@ class CommandExecutedChecker(BaseCriterion[CommandExecutedCriterion]): criterion_type = "command_executed" + # Observable mid-run: command matches accumulate monotonically in the live + # stream, so a min_count pass (no upper bound) and a max_count exceedance + # (incl. the must-NOT-run 0/0 form) are both decidable before end-of-run. + live_stop_polarities: ClassVar[frozenset[str]] = frozenset({"pass", "fail"}) + + @staticmethod + def _matching_commands( + criterion: CommandExecutedCriterion, + all_commands: list["CommandTelemetry"], + pattern: re.Pattern[str] | None, + exclude_re: re.Pattern[str] | None, + ) -> list[str]: + """Filter + label the commands matching this criterion. + + Shared by ``_check_impl`` and ``live_verdict`` so a command can never + count for one and not the other (the live trigger and the authoritative + score always agree on WHICH commands match). Callers compile the + patterns and own error handling. + """ + matching: list[str] = [] + for cmd in all_commands: + # Filter by tool name + if criterion.tool_name is not None and cmd.tool_name != criterion.tool_name: + continue + + # Filter by success status + if criterion.require_success and cmd.result_status != "success": + continue + + # Extract text for pattern matching (Bash: command param; others: JSON-serialized params) + if cmd.tool_name == "Bash" and cmd.parameters.get("command"): + cmd_text = cmd.parameters["command"] + else: + cmd_text = json.dumps(cmd.parameters) + + # Truncate to mitigate ReDoS on large command strings + if len(cmd_text) > _MAX_PATTERN_SEARCH_LEN: + cmd_text = cmd_text[:_MAX_PATTERN_SEARCH_LEN] + + # Filter by command pattern + if pattern is not None and not pattern.search(cmd_text): + continue + + # Apply exclusion pattern (skip commands matching the exclusion) + if exclude_re is not None and exclude_re.search(cmd_text): + continue + + # Build a display label for the matched command + if cmd.tool_name == "Bash" and cmd.parameters.get("command"): + label = cmd.parameters["command"] + else: + label = f"{cmd.tool_name}({json.dumps(cmd.parameters)[:80]})" + matching.append(label) + return matching + + def live_verdict( + self, + criterion: CommandExecutedCriterion, + turn_records: list["TurnRecord"], + ) -> LiveVerdict: + """Decide from the partial trajectory (early-stop trigger). + + - ``fail`` the moment ``match_count`` exceeds ``max_count`` — monotone + (once over, always over). Covers the ``min_count: 0, max_count: 0`` + "must NOT run" form: the first forbidden match is a definitive fail. + - ``pass`` the moment ``match_count`` reaches ``min_count`` when there is + no upper bound (``max_count is None`` and ``min_count > 0``). With an + upper bound the pass is only final at end-of-run, so it stays + ``undecided`` here. + - otherwise ``undecided``. + + A malformed regex yields ``undecided`` (the final ``_check_impl`` surfaces + the config error as a scored failure). + """ + all_commands = [cmd for turn in turn_records for cmd in turn.commands] + try: + pattern = ( + re.compile(criterion.command_pattern, re.DOTALL) if criterion.command_pattern is not None else None + ) + exclude_re = ( + re.compile(criterion.exclude_pattern, re.DOTALL) if criterion.exclude_pattern is not None else None + ) + except re.error: + return "undecided" + + match_count = len(self._matching_commands(criterion, all_commands, pattern, exclude_re)) + if criterion.max_count is not None and match_count > criterion.max_count: + return "fail" + if criterion.max_count is None and criterion.min_count > 0 and match_count >= criterion.min_count: + return "pass" + return "undecided" + def _check_impl( self, criterion: CommandExecutedCriterion, @@ -98,41 +191,9 @@ def _check_impl( error=f"Invalid regex: {e}", ) - # Filter and count matching commands - matching_commands: list[str] = [] - for cmd in all_commands: - # Filter by tool name - if criterion.tool_name is not None and cmd.tool_name != criterion.tool_name: - continue - - # Filter by success status - if criterion.require_success and cmd.result_status != "success": - continue - - # Extract text for pattern matching (Bash: command param; others: JSON-serialized params) - if cmd.tool_name == "Bash" and cmd.parameters.get("command"): - cmd_text = cmd.parameters["command"] - else: - cmd_text = json.dumps(cmd.parameters) - - # Truncate to mitigate ReDoS on large command strings - if len(cmd_text) > _MAX_PATTERN_SEARCH_LEN: - cmd_text = cmd_text[:_MAX_PATTERN_SEARCH_LEN] - - # Filter by command pattern - if pattern is not None and not pattern.search(cmd_text): - continue - - # Apply exclusion pattern (skip commands matching the exclusion) - if exclude_re is not None and exclude_re.search(cmd_text): - continue - - # Build a display label for the matched command - if cmd.tool_name == "Bash" and cmd.parameters.get("command"): - label = cmd.parameters["command"] - else: - label = f"{cmd.tool_name}({json.dumps(cmd.parameters)[:80]})" - matching_commands.append(label) + # Filter and count matching commands (shared with live_verdict so the + # live trigger and the authoritative score never disagree on matches). + matching_commands = self._matching_commands(criterion, all_commands, pattern, exclude_re) match_count = len(matching_commands) diff --git a/src/coder_eval/criteria/skill_triggered.py b/src/coder_eval/criteria/skill_triggered.py index 284631ab..06b90d0e 100644 --- a/src/coder_eval/criteria/skill_triggered.py +++ b/src/coder_eval/criteria/skill_triggered.py @@ -9,10 +9,11 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING +import re +from typing import TYPE_CHECKING, ClassVar from coder_eval.criteria._classification_aggregate import overlay_classification_metrics -from coder_eval.criteria.base import BaseCriterion, register_criterion +from coder_eval.criteria.base import BaseCriterion, LiveVerdict, register_criterion from coder_eval.models import ( ClassificationCriterionResult, CriterionAggregate, @@ -32,6 +33,35 @@ _YES = "yes" _NO = "no" +# Extracts the skill name between ``skills//`` path segments (the Codex +# file-read signal). The leading class keeps a bare ``skills//`` from matching. +_SKILL_PATH_RE = re.compile(r"skills/([A-Za-z0-9][A-Za-z0-9_-]*)/") + + +def _engaged_skill_names(cmd: CommandTelemetry) -> set[str]: + """All skill names engaged by ONE command, agent-agnostically (any-skill). + + Generalizes ``_engaged_skill`` (which asks about one named skill) so a + live verdict can detect a *competing* skill engagement. Collects: + + - the Claude ``Skill`` tool call's ``skill`` parameter, namespace-stripped + via ``.split(":")[-1]`` (the SDK reports ``plugin:skill-name``); and + - every ``skills//`` path segment appearing in any string parameter + (the Codex / file-read signal — repo layout or the ``.agents/skills`` + sandbox symlink). + + Returns the (possibly empty) set of engaged skill names for this command. + """ + names: set[str] = set() + if cmd.tool_name == "Skill": + skill = cmd.parameters.get("skill", "") + if isinstance(skill, str) and skill: + names.add(skill.split(":")[-1]) + for value in cmd.parameters.values(): + if isinstance(value, str): + names.update(_SKILL_PATH_RE.findall(value)) + return names + def _engaged_skill(cmd: CommandTelemetry, skill_name: str) -> bool: """True when one command engaged ``skill_name`` — agent-agnostically. @@ -63,6 +93,11 @@ class SkillTriggeredChecker(BaseCriterion[SkillTriggeredCriterion]): criterion_type = "skill_triggered" + # Observable mid-run: a Skill tool call (or a skill file read) is a positive + # event in the live stream, so both polarities are decidable the moment the + # agent first engages ANY skill (the first-engagement policy in live_verdict). + live_stop_polarities: ClassVar[frozenset[str]] = frozenset({"pass", "fail"}) + def _check_impl( self, criterion: SkillTriggeredCriterion, @@ -99,6 +134,37 @@ def _check_impl( expected_label=expected, ) + def live_verdict( + self, + criterion: SkillTriggeredCriterion, + turn_records: list[TurnRecord], + ) -> LiveVerdict: + """First-engagement policy: the FIRST observed skill engagement decides. + + Activation measures which skill the agent selects *first*, so every + stacked ``skill_triggered`` criterion is decided simultaneously by the + first command that engages any skill: + + - before any engagement -> ``"undecided"``; + - on the first command engaging some skill: ``observed = (skill_name in + engaged)``, ``expected = (expected_skill == skill_name)`` -> + ``"pass"`` iff they match, else ``"fail"``. + + This covers the "wrong skill loads" case (a positive wrong signal) and + negative rows (``expected_skill == ""`` -> any engagement of the target + fails that criterion). It is a *policy* made consistent by stopping: the + trajectory is frozen at the deciding event, so the standard checker on the + truncated trajectory agrees with this verdict by construction. + """ + expected_yes = criterion.expected_skill == criterion.skill_name + for turn in turn_records: + for cmd in turn.commands: + engaged = _engaged_skill_names(cmd) + if engaged: + observed_yes = criterion.skill_name in engaged + return "pass" if observed_yes == expected_yes else "fail" + return "undecided" + def aggregate( self, criterion: SkillTriggeredCriterion, diff --git a/src/coder_eval/models/criteria.py b/src/coder_eval/models/criteria.py index b3d67d19..1bb59c8f 100644 --- a/src/coder_eval/models/criteria.py +++ b/src/coder_eval/models/criteria.py @@ -122,6 +122,18 @@ class BaseSuccessCriterion(BaseModel, ABC): ), ) + stop_when: Literal["pass", "fail", "decided"] | None = Field( + default=None, + description=( + "Opt-in early-stop polarity for this criterion (membership in the run's 'armed set'). " + "None (default) = not a stop criterion. 'pass' = a live PASS may contribute to a stop; " + "'fail' = a live definitive FAIL may trigger a stop; 'decided' = either. Inert unless " + "run_limits.stop_early is True. Only valid on criteria observable mid-run (e.g. " + "skill_triggered, command_executed); an unobservable armed criterion is rejected at " + "resolution time." + ), + ) + requires_agent: ClassVar[bool] = False """True if this criterion requires agent turn records to evaluate correctly.""" diff --git a/src/coder_eval/models/limits.py b/src/coder_eval/models/limits.py index 3ddf34bf..5a99dfae 100644 --- a/src/coder_eval/models/limits.py +++ b/src/coder_eval/models/limits.py @@ -87,3 +87,14 @@ class RunLimits(BaseModel): "existing behavior (Claude reports real cache-creation writes here)." ), ) + stop_early: bool = Field( + default=False, + description=( + "Opt-in master switch for early-stop-on-criterion. When True, the run ends as " + "soon as the armed criteria (those with stop_when set) are decided mid-run - on " + "pass or on a definitive fail - so a raised max_turns is not wasted once the " + "measured signal has happened. Default False keeps behavior identical. Requires a " + "Claude single-shot task with at least one observable armed criterion; every " + "unsupported combination is rejected at resolution time." + ), + ) diff --git a/src/coder_eval/orchestration/early_stop.py b/src/coder_eval/orchestration/early_stop.py new file mode 100644 index 00000000..8c43c79a --- /dev/null +++ b/src/coder_eval/orchestration/early_stop.py @@ -0,0 +1,110 @@ +"""Early-stop-on-criterion: resolution-time validation. + +Opt-in mechanism (``run_limits.stop_early``) that ends a single-shot run as soon +as its *armed* criteria (those with ``stop_when`` set) are decided mid-run — on +pass or on a definitive fail. This module owns the resolution-time guardrails: +``validate_early_stop`` rejects every configuration v1 cannot honor as a hard +error, so an unsupported arming is never a silent no-op. The runtime watcher that +actually observes the event stream and trips the interrupt lands in a later +commit; until then arming is inert at runtime. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from coder_eval.models import TaskDefinition + + +class EarlyStopConfigError(ValueError): + """Raised when a task arms ``run_limits.stop_early`` in a way v1 cannot honor. + + Subclasses ``ValueError`` so the run path's resolve -> ``typer.BadParameter`` + conversion covers it transparently; the ``plan`` command catches this + subclass specifically to flip its exit code (generic per-variant resolution + failures intentionally stay soft). + """ + + +def validate_early_stop(task: TaskDefinition) -> None: + """Validate an armed early-stop task at resolution time; no-op when unarmed. + + Called after the config layers have merged (``resolve_all_tasks`` post-CLI + overrides, the ``plan`` per-variant loop, and defensively in + ``Orchestrator._setup``). All checks below are skipped unless + ``run_limits.stop_early`` is True, so default runs are entirely unaffected. + + Raise order (matters for which error a multiply-invalid task reports first): + 5. armed together with ``simulation.enabled`` -> error + 1. agent does not declare ``supports_cooperative_stop`` -> error + 2. armed but no criterion sets ``stop_when`` -> error + then per armed criterion: + 3. criterion is not observable mid-run -> error + 4. requested ``stop_when`` polarity the criterion cannot decide -> error + + Raises: + EarlyStopConfigError: on any unsupported armed configuration. + """ + limits = task.run_limits + if limits is None or not limits.stop_early: + return + + # (5) Simulation/dialog mode has its own criteria-driven stop. + if task.simulation is not None and task.simulation.enabled: + raise EarlyStopConfigError( + "run_limits.stop_early is not supported together with simulation.enabled " + + "(early-stop v1 is single-shot only); use simulation.stop_on_criteria_pass " + + "for dialog-mode criteria stopping." + ) + + # (1) The agent must honor the cooperative interrupt. Lazily import the + # registry + plugin loader so this module stays free of runtime coder_eval + # imports at load time. + from coder_eval.agents.registry import AgentRegistry + from coder_eval.plugins import ensure_plugins_loaded + + ensure_plugins_loaded() + agent_type = str(task.agent.type) if task.agent is not None and task.agent.type is not None else None + registration = AgentRegistry.get(agent_type) if agent_type is not None else None + supports = bool(getattr(registration.agent_class, "supports_cooperative_stop", False)) if registration else False + if not supports: + raise EarlyStopConfigError( + "run_limits.stop_early requires an agent that supports cooperative stopping " + + f"(currently Claude Code); agent type {agent_type!r} does not." + ) + + # (2) Arming requires at least one stop criterion. + armed = [c for c in task.success_criteria if c.stop_when is not None] + if not armed: + raise EarlyStopConfigError( + "run_limits.stop_early is armed but no success criterion sets stop_when; " + + "arming requires at least one stop criterion (e.g. stop_when: decided)." + ) + + # (3)+(4) Per armed criterion: observable, then the requested polarity is + # decidable. The criteria registry is not initialized at resolution time. + from coder_eval.criteria import CriterionRegistry, init_criteria + + init_criteria(validate=False) + for c in armed: + # `armed` filtered on `stop_when is not None`; re-bind + assert so pyright + # narrows the Literal away from None for the set arithmetic below. + polarity = c.stop_when + assert polarity is not None + checker_cls = CriterionRegistry.get_checker(c.type) + polarities = checker_cls.live_stop_polarities + if not polarities: + raise EarlyStopConfigError( + f"criterion type {c.type!r} is armed (stop_when={polarity!r}) but is not " + + "observable mid-run; early-stop supports only live-observable criteria " + + "(e.g. skill_triggered, command_executed)." + ) + needed = {"pass", "fail"} if polarity == "decided" else {polarity} + missing = sorted(needed - polarities) + if missing: + raise EarlyStopConfigError( + f"criterion type {c.type!r} cannot decide polarity {missing} mid-run " + + f"(stop_when={polarity!r}); it supports {sorted(polarities)}." + ) diff --git a/src/coder_eval/orchestration/experiment.py b/src/coder_eval/orchestration/experiment.py index a03505a7..0962c362 100644 --- a/src/coder_eval/orchestration/experiment.py +++ b/src/coder_eval/orchestration/experiment.py @@ -572,6 +572,8 @@ def resolve_all_tasks( Raises: ValueError: If duplicate task IDs are found after resolution. """ + from .early_stop import validate_early_stop + resolved: list[ResolvedTask] = [] skipped: list[SkippedTask] = [] @@ -637,6 +639,12 @@ def resolve_all_tasks( # Apply layer 5 (CLI overrides) _apply_cli_overrides(resolved_task, config, lineage) + # Early-stop guardrails: run once the task is fully resolved (all 5 + # layers merged, incl. -D run_limits.stop_early). No-op unless armed; + # a bad arming raises EarlyStopConfigError (a ValueError) which the + # run path converts to a clean CLI error. + validate_early_stop(resolved_task) + # Fan-out: simulation n_trials takes precedence over experiment repeats # when simulation is active; otherwise use experiment-level repeats. sim = resolved_task.simulation diff --git a/src/coder_eval/orchestrator.py b/src/coder_eval/orchestrator.py index 050b16fa..d6820976 100644 --- a/src/coder_eval/orchestrator.py +++ b/src/coder_eval/orchestrator.py @@ -50,6 +50,7 @@ UserMessage, resolve_route, ) +from .orchestration.early_stop import validate_early_stop from .orchestration.evaluation import load_reference from .path_utils import format_task_log_id, task_log_path from .sandbox import Sandbox @@ -845,6 +846,11 @@ async def _setup(self) -> None: Raises: RuntimeError: If setup fails """ + # Defensive early-stop guardrails for the library-use and in-container + # paths (the CLI already validated during resolution). No-op unless + # run_limits.stop_early is armed. + validate_early_stop(self.task) + if self.sandbox is not None: # evaluate-only mode: sandbox already set up, skip agent assert self.result is not None diff --git a/tests/test_early_stop.py b/tests/test_early_stop.py new file mode 100644 index 00000000..4baa5d0a --- /dev/null +++ b/tests/test_early_stop.py @@ -0,0 +1,339 @@ +"""Tests for early-stop-on-criterion: config surface, live-verdict +contract, and resolution-time guardrails. + +This suite is inert at runtime (nothing invokes the interrupt yet); these tests +cover the pieces that DO exist: the two new opt-in config fields, the +``live_verdict`` / ``live_stop_polarities`` observability contract on the two +observable criteria, and ``validate_early_stop``'s guardrails on both the +``plan`` and ``run`` surfaces. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +import pytest + +from coder_eval.criteria import CriterionRegistry, init_criteria +from coder_eval.criteria.base import BaseCriterion +from coder_eval.criteria.command_executed import CommandExecutedChecker +from coder_eval.criteria.skill_triggered import SkillTriggeredChecker, _engaged_skill_names +from coder_eval.models import ( + AgentKind, + CommandExecutedCriterion, + CommandTelemetry, + FileExistsCriterion, + RunLimits, + SandboxConfig, + SimulationConfig, + SkillTriggeredCriterion, + TaskDefinition, + TurnRecord, + parse_agent_config, +) +from coder_eval.orchestration.early_stop import EarlyStopConfigError, validate_early_stop + + +# --------------------------------------------------------------------------- # +# Helpers +# --------------------------------------------------------------------------- # + +_TS = datetime(2026, 1, 1, 0, 0, 0) + + +def _cmd(tool_name: str, parameters: dict[str, Any], *, result_status: str = "success") -> CommandTelemetry: + return CommandTelemetry( + tool_name=tool_name, + tool_id=f"tool-{tool_name}", + timestamp=_TS, + parameters=parameters, + result_status=result_status, + ) + + +def _turn(*commands: CommandTelemetry) -> TurnRecord: + return TurnRecord(iteration=1, user_input="", agent_output="", commands=list(commands)) + + +def _task( + *, + criteria: list[Any], + stop_early: bool = False, + agent_type: AgentKind = AgentKind.CLAUDE_CODE, + simulation: SimulationConfig | None = None, +) -> TaskDefinition: + """Build a minimal resolved-style TaskDefinition for guardrail tests.""" + return TaskDefinition( + task_id="early-stop-test", + description="early-stop test task", + initial_prompt="do the thing", + agent=parse_agent_config(type=agent_type), + sandbox=SandboxConfig(driver="tempdir"), + success_criteria=criteria, + run_limits=RunLimits(stop_early=stop_early, max_turns=20), + simulation=simulation, + ) + + +def _skill_crit(skill_name: str, expected_skill: str, *, stop_when: str | None = None) -> SkillTriggeredCriterion: + return SkillTriggeredCriterion( + type="skill_triggered", + description=f"{skill_name} activation", + skill_name=skill_name, + expected_skill=expected_skill, + stop_when=stop_when, # type: ignore[arg-type] + ) + + +def _cmd_crit( + *, min_count: int = 1, max_count: int | None = None, pattern: str | None = "curl", stop_when: str | None = None +) -> CommandExecutedCriterion: + return CommandExecutedCriterion( + type="command_executed", + description="command check", + tool_name="Bash", + command_pattern=pattern, + min_count=min_count, + max_count=max_count, + stop_when=stop_when, # type: ignore[arg-type] + ) + + +# --------------------------------------------------------------------------- # +# Config surface +# --------------------------------------------------------------------------- # + + +class TestConfigSurface: + def test_stop_early_defaults_false(self) -> None: + assert RunLimits().stop_early is False + + def test_stop_early_settable(self) -> None: + assert RunLimits(stop_early=True).stop_early is True + + def test_stop_when_defaults_none(self) -> None: + assert _skill_crit("s", "s").stop_when is None + + @pytest.mark.parametrize("value", ["pass", "fail", "decided"]) + def test_stop_when_accepts_valid_polarities(self, value: str) -> None: + assert _skill_crit("s", "s", stop_when=value).stop_when == value + + def test_stop_when_rejects_invalid_polarity(self) -> None: + with pytest.raises(ValueError): + _skill_crit("s", "s", stop_when="maybe") + + +# --------------------------------------------------------------------------- # +# skill_triggered live verdict +# --------------------------------------------------------------------------- # + + +class TestEngagedSkillNames: + def test_claude_skill_tool_namespaced(self) -> None: + assert _engaged_skill_names(_cmd("Skill", {"skill": "plugin:date-teller"})) == {"date-teller"} + + def test_bare_skill_param(self) -> None: + assert _engaged_skill_names(_cmd("Skill", {"skill": "date-teller"})) == {"date-teller"} + + def test_file_read_path_segment(self) -> None: + got = _engaged_skill_names(_cmd("Read", {"file_path": "/repo/skills/date-teller/SKILL.md"})) + assert got == {"date-teller"} + + def test_bash_command_path_segment(self) -> None: + assert _engaged_skill_names(_cmd("Bash", {"command": "cat skills/foo/refs.md"})) == {"foo"} + + def test_no_engagement(self) -> None: + assert _engaged_skill_names(_cmd("Bash", {"command": "echo hi"})) == set() + + +class TestSkillTriggeredLiveVerdict: + checker = SkillTriggeredChecker() + + def test_undecided_before_any_engagement(self) -> None: + crit = _skill_crit("date-teller", "date-teller") + assert self.checker.live_verdict(crit, [_turn(_cmd("Bash", {"command": "ls"}))]) == "undecided" + + def test_pass_when_expected_skill_engaged(self) -> None: + crit = _skill_crit("date-teller", "date-teller") + rec = [_turn(_cmd("Skill", {"skill": "plugin:date-teller"}))] + assert self.checker.live_verdict(crit, rec) == "pass" + + def test_fail_when_wrong_skill_engaged(self) -> None: + # Positive row expecting date-teller, but a different skill loads first. + crit = _skill_crit("date-teller", "date-teller") + rec = [_turn(_cmd("Skill", {"skill": "other-skill"}))] + assert self.checker.live_verdict(crit, rec) == "fail" + + def test_negative_row_target_engaged_is_fail(self) -> None: + # expected_skill == "" (negative): engaging the target skill fails. + crit = _skill_crit("date-teller", "") + rec = [_turn(_cmd("Skill", {"skill": "date-teller"}))] + assert self.checker.live_verdict(crit, rec) == "fail" + + def test_negative_row_other_engaged_is_pass(self) -> None: + crit = _skill_crit("date-teller", "") + rec = [_turn(_cmd("Skill", {"skill": "unrelated"}))] + assert self.checker.live_verdict(crit, rec) == "pass" + + def test_first_engagement_decides(self) -> None: + # The wrong skill engages first, the expected one later — first wins. + crit = _skill_crit("date-teller", "date-teller") + rec = [_turn(_cmd("Skill", {"skill": "wrong"}), _cmd("Skill", {"skill": "date-teller"}))] + assert self.checker.live_verdict(crit, rec) == "fail" + + def test_polarities_declared(self) -> None: + assert SkillTriggeredChecker.live_stop_polarities == frozenset({"pass", "fail"}) + + +# --------------------------------------------------------------------------- # +# command_executed live verdict +# --------------------------------------------------------------------------- # + + +class TestCommandExecutedLiveVerdict: + checker = CommandExecutedChecker() + + def _bash(self, cmd: str) -> CommandTelemetry: + return _cmd("Bash", {"command": cmd}) + + def test_undecided_below_min(self) -> None: + crit = _cmd_crit(min_count=2, pattern="curl") + assert self.checker.live_verdict(crit, [_turn(self._bash("curl x"))]) == "undecided" + + def test_pass_at_min_when_no_upper_bound(self) -> None: + crit = _cmd_crit(min_count=1, pattern="curl") + assert self.checker.live_verdict(crit, [_turn(self._bash("curl x"))]) == "pass" + + def test_must_not_run_first_match_is_fail(self) -> None: + crit = _cmd_crit(min_count=0, max_count=0, pattern="rm ") + assert self.checker.live_verdict(crit, [_turn(self._bash("rm -rf /"))]) == "fail" + + def test_must_not_run_no_match_is_undecided(self) -> None: + # Absence of the forbidden event isn't decidable mid-run. + crit = _cmd_crit(min_count=0, max_count=0, pattern="rm ") + assert self.checker.live_verdict(crit, [_turn(self._bash("ls"))]) == "undecided" + + def test_range_never_live_passes(self) -> None: + crit = _cmd_crit(min_count=2, max_count=3, pattern="x") + rec = [_turn(self._bash("x"), self._bash("x"))] # within range + assert self.checker.live_verdict(crit, rec) == "undecided" + + def test_range_over_max_is_fail(self) -> None: + crit = _cmd_crit(min_count=1, max_count=2, pattern="x") + rec = [_turn(self._bash("x"), self._bash("x"), self._bash("x"))] + assert self.checker.live_verdict(crit, rec) == "fail" + + def test_invalid_regex_is_undecided(self) -> None: + crit = _cmd_crit(min_count=1, pattern="[") + assert self.checker.live_verdict(crit, [_turn(self._bash("anything"))]) == "undecided" + + def test_zero_min_no_max_is_undecided(self) -> None: + # min_count=0 + no upper bound has nothing to wait for and no fail edge. + crit = _cmd_crit(min_count=0, max_count=None, pattern="curl") + assert self.checker.live_verdict(crit, [_turn(self._bash("ls"))]) == "undecided" + + def test_matching_shared_with_check_impl(self) -> None: + # The live count and the authoritative check agree on matches. + crit = _cmd_crit(min_count=1, pattern="curl") + rec = [_turn(self._bash("curl a"), self._bash("wget b"), self._bash("curl c"))] + assert self.checker.live_verdict(crit, rec) == "pass" + # _check_impl scores 1.0 on the same trajectory (>= min_count). + result = self.checker.check(crit, sandbox=None, turn_records=rec) # type: ignore[arg-type] + assert result.score == 1.0 + + def test_polarities_declared(self) -> None: + assert CommandExecutedChecker.live_stop_polarities == frozenset({"pass", "fail"}) + + +# --------------------------------------------------------------------------- # +# Base default: unobservable criteria +# --------------------------------------------------------------------------- # + + +class TestBaseLiveVerdictDefault: + def test_base_polarities_empty(self) -> None: + assert BaseCriterion.live_stop_polarities == frozenset() + + def test_unobservable_checker_is_undecided(self) -> None: + init_criteria(validate=False) + checker = CriterionRegistry.get_checker("file_exists")() + assert checker.live_stop_polarities == frozenset() + crit = FileExistsCriterion(type="file_exists", path="x.txt", description="x") + assert checker.live_verdict(crit, [_turn()]) == "undecided" + + +# --------------------------------------------------------------------------- # +# Resolution-time guardrails +# --------------------------------------------------------------------------- # + + +class TestValidateEarlyStop: + def test_unarmed_is_noop_even_with_bad_shape(self) -> None: + # stop_early=False → validator never inspects anything. + task = _task(criteria=[_skill_crit("s", "s")], stop_early=False, agent_type=AgentKind.CODEX) + validate_early_stop(task) # no raise + + def test_unarmed_with_stop_when_is_inert(self) -> None: + # A criterion may declare stop_when; without stop_early it stays inert. + task = _task(criteria=[_skill_crit("s", "s", stop_when="decided")], stop_early=False) + validate_early_stop(task) # no raise + + def test_armed_happy_path_accepts(self) -> None: + task = _task(criteria=[_skill_crit("s", "s", stop_when="decided")], stop_early=True) + validate_early_stop(task) # no raise + + def test_armed_command_executed_accepts(self) -> None: + task = _task(criteria=[_cmd_crit(stop_when="fail")], stop_early=True) + validate_early_stop(task) # no raise + + def test_guardrail5_simulation_rejected(self) -> None: + sim = SimulationConfig(enabled=True, persona="user", goal="get it done") + task = _task(criteria=[_skill_crit("s", "s", stop_when="decided")], stop_early=True, simulation=sim) + with pytest.raises(EarlyStopConfigError, match="simulation"): + validate_early_stop(task) + + def test_guardrail1_non_claude_agent_rejected(self) -> None: + task = _task(criteria=[_skill_crit("s", "s", stop_when="decided")], stop_early=True, agent_type=AgentKind.CODEX) + with pytest.raises(EarlyStopConfigError, match="cooperative stopping"): + validate_early_stop(task) + + def test_guardrail2_no_stop_criterion_rejected(self) -> None: + task = _task(criteria=[_skill_crit("s", "s")], stop_early=True) + with pytest.raises(EarlyStopConfigError, match="at least one stop criterion"): + validate_early_stop(task) + + def test_guardrail3_unobservable_criterion_rejected(self) -> None: + crit = FileExistsCriterion(type="file_exists", path="x.txt", description="x", stop_when="pass") + task = _task(criteria=[crit], stop_early=True) + with pytest.raises(EarlyStopConfigError, match="observable"): + validate_early_stop(task) + + def test_guardrail4_unsupported_polarity_rejected(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Force command_executed to be pass-only, then arm it with stop_when="fail". + init_criteria(validate=False) + monkeypatch.setattr(CommandExecutedChecker, "live_stop_polarities", frozenset({"pass"})) + task = _task(criteria=[_cmd_crit(stop_when="fail")], stop_early=True) + with pytest.raises(EarlyStopConfigError, match="polarity"): + validate_early_stop(task) + + def test_raise_order_simulation_before_agent(self) -> None: + # Both simulation AND a non-Claude agent are invalid; simulation reports first. + sim = SimulationConfig(enabled=True, persona="user", goal="g") + task = _task( + criteria=[_skill_crit("s", "s", stop_when="decided")], + stop_early=True, + agent_type=AgentKind.CODEX, + simulation=sim, + ) + with pytest.raises(EarlyStopConfigError, match="simulation"): + validate_early_stop(task) + + def test_stacked_activation_criteria_accept(self) -> None: + # Multiple armed skill_triggered criteria (the activation pattern). + crits = [ + _skill_crit("skill-a", "skill-a", stop_when="decided"), + _skill_crit("skill-b", "skill-a", stop_when="decided"), + ] + task = _task(criteria=crits, stop_early=True) + validate_early_stop(task) # no raise From 6a3206682b0d7c945e205c5bed4ef76945b6b74b Mon Sep 17 00:00:00 2001 From: anonymous Date: Thu, 9 Jul 2026 15:37:52 -0700 Subject: [PATCH 2/7] feat(agents): cooperative should_stop seam + STOPPED_EARLY status (unwired) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the mechanism to interrupt a Claude turn cleanly between SDK messages, still unwired (the orchestrator does not pass should_stop yet — zero behavioral change): - should_stop: Callable[[], bool] | None kwarg on the communicate() ABC; the three non-Claude agents (codex/antigravity/noop) accept-and-ignore it for override compatibility. - ClaudeCodeAgent honors it: the message pump is extracted to _pump_messages (keeps communicate under the statement cap; query still resolved as a module global so test mocks keep working); the stop check runs AFTER dispatch so a watcher observing events on the deciding message stops before the next is pulled. A stopped_early_hit flag drives the finalize ladder (TIMEOUT -> STOPPED_EARLY -> COMPLETED) — max_turns promotion only fires for COMPLETED, and the post-finally timeout raise is gated on timeout_hit, so an early stop returns a clean crashed=False TurnRecord and never raises. - New STOPPED_EARLY member on TurnEndStatus + AgentEndStatus (consumers: renderers format status.value generically; TurnEndStatus(status.value) conversion round-trips; wire is by-value). - Seam tests: stop after the first dispatched message, one AgentEndEvent (STOPPED_EARLY, crashed=False), no raise; should_stop=None/False consume the full stream and finalize COMPLETED. --- src/coder_eval/agent.py | 11 ++ src/coder_eval/agents/antigravity_agent.py | 8 +- src/coder_eval/agents/claude_code_agent.py | 63 +++++++++-- src/coder_eval/agents/codex_agent.py | 6 + src/coder_eval/agents/noop_agent.py | 6 + src/coder_eval/streaming/events.py | 2 + tests/test_early_stop.py | 121 +++++++++++++++++++++ 7 files changed, 206 insertions(+), 11 deletions(-) diff --git a/src/coder_eval/agent.py b/src/coder_eval/agent.py index 8c3dc509..2b244ce2 100644 --- a/src/coder_eval/agent.py +++ b/src/coder_eval/agent.py @@ -5,6 +5,7 @@ import logging from abc import ABC, abstractmethod +from collections.abc import Callable from typing import Any, ClassVar, NoReturn, Protocol from .errors import AgentCrashError, TurnTimeoutError @@ -190,6 +191,7 @@ async def communicate( stream_callback: StreamCallback | None = None, timeout: float | None = None, max_turns: int | None = None, + should_stop: Callable[[], bool] | None = None, ) -> TurnRecord: """Send a message to the agent and receive its response. @@ -205,6 +207,15 @@ async def communicate( ``communicate()`` call. When the agent would exceed it, the returned ``TurnRecord`` has ``max_turns_exhausted=True``. None defers to the underlying SDK default. + should_stop: Cooperative early-stop poll for early-stop-on-criterion. + When provided, an implementation that supports cooperative + stopping (``supports_cooperative_stop=True``) should call it at + each safe message boundary and, when it returns True, stop + pulling further work and finalize the turn cleanly + (``crashed=False``, no raise). ``None`` (default) preserves the + pre-existing behavior exactly. Agents that do not support it + accept the argument and ignore it (the orchestrator only passes + it to a capable agent). Returns: TurnRecord containing the complete interaction diff --git a/src/coder_eval/agents/antigravity_agent.py b/src/coder_eval/agents/antigravity_agent.py index cc19cda9..953cf4f4 100644 --- a/src/coder_eval/agents/antigravity_agent.py +++ b/src/coder_eval/agents/antigravity_agent.py @@ -23,7 +23,7 @@ import logging import os import time -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Callable from contextlib import AsyncExitStack from datetime import datetime from pathlib import Path @@ -409,9 +409,15 @@ async def communicate( stream_callback: StreamCallback | None = None, timeout: float | None = None, max_turns: int | None = None, + should_stop: Callable[[], bool] | None = None, ) -> TurnRecord: """Send a message to the Antigravity agent and receive its response. + ``should_stop`` is accepted for ``Agent.communicate`` override + compatibility and ignored — Antigravity does not support cooperative + early-stop (``supports_cooperative_stop`` is False), so the orchestrator + never passes it. + Drives one logical turn: ``conversation.send(prompt)`` then iterate ``receive_steps()`` until the turn goes idle, mapping the Gemini step stream onto the standardized event protocol. diff --git a/src/coder_eval/agents/claude_code_agent.py b/src/coder_eval/agents/claude_code_agent.py index c6c62855..cd24451a 100644 --- a/src/coder_eval/agents/claude_code_agent.py +++ b/src/coder_eval/agents/claude_code_agent.py @@ -218,6 +218,9 @@ def __init__( self.deadline = deadline # Set True by the in-loop deadline break OR the watchdog callback. self.timeout_hit = False + # Set True by the in-loop cooperative-stop break (early-stop-on-criterion). + # Distinct from timeout_hit: a clean, non-crash stop that must NOT raise. + self.stopped_early_hit = False # Resolved by _build_claude_query, set on the state before any finalize # path. Stays None if we crash before setup (finalize reads it for cost # backfill). @@ -790,6 +793,7 @@ async def communicate( stream_callback: StreamCallback | None = None, timeout: float | None = None, max_turns: int | None = None, + should_stop: Callable[[], bool] | None = None, ) -> TurnRecord: """Send a message to Claude and receive its response. @@ -802,6 +806,12 @@ async def communicate( asyncio.wait_for is not sufficient). max_turns: Hard cap on inner-loop turns for this call. None defers to the SDK default. + should_stop: Cooperative early-stop poll (early-stop-on-criterion). + When provided, it is checked after each dispatched SDK message; + the first True finalizes the turn cleanly as STOPPED_EARLY + (``crashed=False``, no raise) at the next message boundary. + ``None`` (default) leaves the message loop behaviorally + identical to before. Returns: TurnRecord containing the complete interaction @@ -902,16 +912,7 @@ def _on_turn_timeout() -> None: asyncio_task_to_cancel=asyncio.current_task(), label=f"Turn timeout ({timeout:g}s)" if timeout else "turn_timeout", ): - async for message in query(**query_kwargs): - # Wall-clock guard at the TOP of the loop: it breaks BEFORE - # the message is dispatched (and appended), so the - # over-deadline message is DISCARDED — no append, no events. - # Do NOT relocate this to a post-loop check. - if deadline is not None and time.monotonic() > deadline: - state.timeout_hit = True - self._log.warning("Turn timeout reached mid-stream; breaking out of message loop") - break - state.dispatch(message) + await self._pump_messages(state, query_kwargs, deadline, should_stop) self._log.debug("Agent query stream ended") @@ -968,6 +969,12 @@ def _on_turn_timeout() -> None: if state.timeout_hit: assert timeout is not None state.finalize(AgentEndStatus.TIMEOUT, crashed=True, crash_reason=format_timeout_reason(timeout)) + elif state.stopped_early_hit: + # Clean cooperative stop: NOT a crash, NOT a timeout. The + # max_turns_exhausted promotion in finalize() only fires for + # COMPLETED, so STOPPED_EARLY survives; the post-finally + # timeout raise is gated on timeout_hit, which is False here. + state.finalize(AgentEndStatus.STOPPED_EARLY, crashed=False, crash_reason=None) else: state.finalize(AgentEndStatus.COMPLETED, crashed=False, crash_reason=None) self._active_transport = None @@ -990,6 +997,42 @@ def _on_turn_timeout() -> None: # above — single, agent-agnostic capture path (no parallel record build). return collector.build_turn_record() + async def _pump_messages( + self, + state: _ClaudeTurnState, + query_kwargs: dict[str, Any], + deadline: float | None, + should_stop: Callable[[], bool] | None, + ) -> None: + """Drive the SDK message stream for one turn (extracted from ``communicate``). + + Kept separate so the added cooperative-stop check keeps ``communicate`` + under ruff's statement cap. ``query`` is still resolved as a module global + at call time, so ``patch("...claude_code_agent.query", ...)`` test mocks + keep working. + + Two break conditions, and the order matters: + + - The wall-clock guard runs at the TOP of the loop — it breaks BEFORE the + message is dispatched, so the over-deadline message is DISCARDED (no + append, no events). Do NOT relocate this to a post-loop check. + - The cooperative stop runs AFTER ``state.dispatch(message)`` so a watcher + observing events during dispatch can flip its flag on THIS message and + the next message is never pulled — the deciding message is kept, the + next is not. No-op when ``should_stop is None`` (behaviorally identical + to before). + """ + async for message in query(**query_kwargs): + if deadline is not None and time.monotonic() > deadline: + state.timeout_hit = True + self._log.warning("Turn timeout reached mid-stream; breaking out of message loop") + break + state.dispatch(message) + if should_stop is not None and should_stop(): + state.stopped_early_hit = True + self._log.debug("Cooperative stop requested; ending message loop at this boundary") + break + def _build_claude_query( self, user_input: str, diff --git a/src/coder_eval/agents/codex_agent.py b/src/coder_eval/agents/codex_agent.py index 7431aad7..9e05637d 100644 --- a/src/coder_eval/agents/codex_agent.py +++ b/src/coder_eval/agents/codex_agent.py @@ -7,6 +7,7 @@ import os import shutil import time +from collections.abc import Callable from datetime import datetime from pathlib import Path from typing import Any @@ -723,6 +724,7 @@ async def communicate( stream_callback: StreamCallback | None = None, timeout: float | None = None, max_turns: int | None = None, + should_stop: Callable[[], bool] | None = None, ) -> TurnRecord: """Send a message to Codex and receive its response. @@ -731,6 +733,10 @@ async def communicate( stream_callback: Optional callback for real-time event streaming timeout: Hard wall-clock deadline in seconds max_turns: Hard cap on inner-loop turns (unused for Codex single-turn) + should_stop: Accepted for ``Agent.communicate`` override compatibility + and ignored — Codex does not support cooperative early-stop + (``supports_cooperative_stop`` is False), so the orchestrator + never passes it. Returns: TurnRecord containing the complete interaction diff --git a/src/coder_eval/agents/noop_agent.py b/src/coder_eval/agents/noop_agent.py index f5e57f2a..e39c45df 100644 --- a/src/coder_eval/agents/noop_agent.py +++ b/src/coder_eval/agents/noop_agent.py @@ -15,6 +15,8 @@ from __future__ import annotations +from collections.abc import Callable + from coder_eval.agent import Agent, AgentState from coder_eval.agents.registry import AgentRegistry from coder_eval.models import AgentKind, ApiRoute, NoneAgentConfig, TurnRecord @@ -63,9 +65,13 @@ async def communicate( stream_callback: StreamCallback | None = None, timeout: float | None = None, max_turns: int | None = None, + should_stop: Callable[[], bool] | None = None, ) -> TurnRecord: """Return an empty turn without contacting any model. + ``should_stop`` is accepted for ``Agent.communicate`` override + compatibility and ignored — a no-op turn has nothing to interrupt. + Honors the streaming contract — sole emitter of a balanced event tree (``AgentStart`` -> ``TurnStart`` -> ``TurnEnd`` -> ``AgentEnd``) — so the task-log handler and renderers see a clean turn boundary. The returned diff --git a/src/coder_eval/streaming/events.py b/src/coder_eval/streaming/events.py index e1e06033..8e8bf43f 100644 --- a/src/coder_eval/streaming/events.py +++ b/src/coder_eval/streaming/events.py @@ -61,6 +61,7 @@ class TurnEndStatus(StrEnum): CRASHED = "crashed" TIMEOUT = "timeout" MAX_TURNS_EXHAUSTED = "max_turns_exhausted" + STOPPED_EARLY = "stopped_early" # cooperative early-stop-on-criterion (clean, non-crash) class AgentEndStatus(StrEnum): @@ -70,6 +71,7 @@ class AgentEndStatus(StrEnum): CRASHED = "crashed" TIMEOUT = "timeout" MAX_TURNS_EXHAUSTED = "max_turns_exhausted" + STOPPED_EARLY = "stopped_early" # cooperative early-stop-on-criterion (clean, non-crash) # Reuse the canonical TranscriptMessage union (defined once in telemetry.py) so diff --git a/tests/test_early_stop.py b/tests/test_early_stop.py index 4baa5d0a..c2636c00 100644 --- a/tests/test_early_stop.py +++ b/tests/test_early_stop.py @@ -10,11 +10,15 @@ from __future__ import annotations +import tempfile +from collections.abc import Callable from datetime import datetime from typing import Any +from unittest.mock import patch import pytest +from coder_eval.agents.claude_code_agent import ClaudeCodeAgent from coder_eval.criteria import CriterionRegistry, init_criteria from coder_eval.criteria.base import BaseCriterion from coder_eval.criteria.command_executed import CommandExecutedChecker @@ -33,6 +37,7 @@ parse_agent_config, ) from coder_eval.orchestration.early_stop import EarlyStopConfigError, validate_early_stop +from coder_eval.streaming.events import AgentEndEvent, AgentEndStatus, TurnEndStatus # --------------------------------------------------------------------------- # @@ -337,3 +342,119 @@ def test_stacked_activation_criteria_accept(self) -> None: ] task = _task(criteria=crits, stop_early=True) validate_early_stop(task) # no raise + + +# --------------------------------------------------------------------------- # +# Cooperative should_stop seam on ClaudeCodeAgent — still UNWIRED: the +# orchestrator does not pass should_stop yet, so these drive the agent directly. +# --------------------------------------------------------------------------- # + + +class _DummyMsg: + """Minimal SDK-message stand-in. + + ``dispatch`` records it and matches no message predicate (so it is ignored), + and it exposes no ``.error`` attribute so ``_update_state_from_messages`` + leaves the agent in WORKING — unlike a MagicMock, whose ``.error`` would be a + truthy mock and wrongly flip the state to ERROR. + """ + + def __init__(self, index: int) -> None: + self.index = index + + +class _EventSink: + """StreamCallback that records every emitted event.""" + + def __init__(self) -> None: + self.events: list[Any] = [] + + def on_event(self, event: Any) -> None: + self.events.append(event) + + +async def _run_claude_communicate( + *, stop_after: int | None = None, never: bool = False, n_messages: int = 3 +) -> tuple[ClaudeCodeAgent, TurnRecord, _EventSink, int]: + """Drive ``ClaudeCodeAgent.communicate`` over a mocked ``query`` yielding + ``n_messages`` dummy messages. + + ``stop_after``: build a should_stop that returns True once that many messages + have been pulled (checked after each dispatch). ``never``: pass an explicit + always-False should_stop. Neither: pass ``should_stop=None``. Returns + ``(agent, record, sink, pulled_count)``. + """ + config = parse_agent_config(type=AgentKind.CLAUDE_CODE, permission_mode="acceptEdits") + agent = ClaudeCodeAgent(config) + pulled = {"n": 0} + + should_stop: Callable[[], bool] | None + if stop_after is not None: + + def should_stop() -> bool: + return pulled["n"] >= stop_after + + elif never: + + def should_stop() -> bool: + return False + + else: + should_stop = None + + with tempfile.TemporaryDirectory() as tmpdir: + await agent.start(tmpdir) + + async def mock_query(prompt: Any, options: Any, transport: Any = None) -> Any: + for i in range(n_messages): + pulled["n"] += 1 + yield _DummyMsg(i) + + sink = _EventSink() + with patch("coder_eval.agents.claude_code_agent.query", mock_query): + record = await agent.communicate("prompt", stream_callback=sink, should_stop=should_stop) + return agent, record, sink, pulled["n"] + + +def _agent_end_events(sink: _EventSink) -> list[AgentEndEvent]: + return [e for e in sink.events if isinstance(e, AgentEndEvent)] + + +class TestCooperativeStopSeam: + def test_stopped_early_member_on_both_enums(self) -> None: + assert AgentEndStatus.STOPPED_EARLY.value == "stopped_early" + assert TurnEndStatus.STOPPED_EARLY.value == "stopped_early" + + def test_turnendstatus_conversion_from_agentendstatus(self) -> None: + # finalize() (and antigravity_agent) convert via TurnEndStatus(status.value); + # the new member must round-trip so an early-stopped open turn isn't mislabeled. + assert TurnEndStatus(AgentEndStatus.STOPPED_EARLY.value) == TurnEndStatus.STOPPED_EARLY + + async def test_stop_after_first_dispatched_message(self) -> None: + _agent, record, sink, pulled = await _run_claude_communicate(stop_after=1, n_messages=3) + # The deciding message is kept; the next is never pulled. + assert pulled == 1 + assert record.crashed is False + ends = _agent_end_events(sink) + assert len(ends) == 1 + assert ends[0].status == AgentEndStatus.STOPPED_EARLY + assert ends[0].crashed is False + + async def test_early_stop_is_clean_not_crashed(self) -> None: + agent, record, _sink, _pulled = await _run_claude_communicate(stop_after=1) + # A clean stop: no partial pending_turn, no ERROR state, no raise (we got here). + assert agent.pending_turn is None + assert agent.get_state().value != "error" + assert record.crashed is False + + async def test_should_stop_none_consumes_full_stream(self) -> None: + _agent, _record, sink, pulled = await _run_claude_communicate(stop_after=None, n_messages=3) + assert pulled == 3 + ends = _agent_end_events(sink) + assert len(ends) == 1 + assert ends[0].status == AgentEndStatus.COMPLETED + + async def test_should_stop_false_consumes_full_stream(self) -> None: + _agent, _record, sink, pulled = await _run_claude_communicate(never=True, n_messages=3) + assert pulled == 3 + assert _agent_end_events(sink)[0].status == AgentEndStatus.COMPLETED From d278736808b1ff464b179de26f3eb844ce8b5527 Mon Sep 17 00:00:00 2001 From: anonymous Date: Thu, 9 Jul 2026 16:09:20 -0700 Subject: [PATCH 3/7] feat(orchestrator): early-stop watcher, armed-set gate, EarlyStopInfo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Activate early-stop-on-criterion: when run_limits.stop_early is armed, an EarlyStopWatcher composed into the agent's event stream evaluates each armed criterion's live_verdict on every tool completion and trips the cooperative should_stop the moment the armed set is decided (pass or definitive fail). - models/results.py: EarlyStopReason (StrEnum), EarlyStopInfo (why/when the run stopped), EvaluationResult.early_stop, and armed_criteria_passed — the armed-subset gate sibling of all_criteria_passed. Exports added. - orchestration/early_stop.py: EarlyStopWatcher (own EventCollector + the stop rule; fail-open on a raising verdict; latched decision; turn/tool/elapsed attribution). - orchestrator.py: build the watcher once in _setup when armed; compose it into the stream regardless of --stream; pass should_stop; populate result.early_stop before check_all; gate an early-stopped run on the armed subset (others advisory) vs the full gate on a completed run. - tests: 28 new (models, watcher stop-rule/fail-open/latching, orchestrator wiring, timeout-beats-stop precedence). --- src/coder_eval/models/__init__.py | 4 + src/coder_eval/models/results.py | 88 ++++ src/coder_eval/orchestration/early_stop.py | 207 ++++++++- src/coder_eval/orchestrator.py | 45 +- tests/test_early_stop.py | 513 ++++++++++++++++++++- 5 files changed, 839 insertions(+), 18 deletions(-) diff --git a/src/coder_eval/models/__init__.py b/src/coder_eval/models/__init__.py index 38a6a765..82354063 100644 --- a/src/coder_eval/models/__init__.py +++ b/src/coder_eval/models/__init__.py @@ -111,6 +111,8 @@ CriterionResult, CriterionResultUnion, CriterionStats, + EarlyStopInfo, + EarlyStopReason, EvaluationResult, FailedRowSummary, JudgeCriterionResult, @@ -277,6 +279,8 @@ "PostRunResult", "ResultSummary", "TurnRecord", + "EarlyStopInfo", + "EarlyStopReason", "EvaluationResult", "SimulationTelemetry", "SuiteRollup", diff --git a/src/coder_eval/models/results.py b/src/coder_eval/models/results.py index a3e5ed05..05dbbb1d 100644 --- a/src/coder_eval/models/results.py +++ b/src/coder_eval/models/results.py @@ -3,6 +3,7 @@ from __future__ import annotations from datetime import datetime +from enum import StrEnum from typing import Annotated, Any, Literal from pydantic import AliasChoices, BaseModel, ConfigDict, Discriminator, Field, Tag, model_validator @@ -378,6 +379,56 @@ class SimulationTelemetry(BaseModel): total_turns: int = Field(description="Number of user↔agent exchanges completed in this dialog", ge=0) +class EarlyStopReason(StrEnum): + """Why an early-stop-on-criterion run was cut short. + + Lives in ``coder_eval.models`` (not ``simulation/``) because + ``EvaluationResult`` carries it and ``models/`` is a leaf package that + cannot import from ``simulation``. ``DialogStopReason`` is a stylistic + reference only. Early-stop is orthogonal telemetry, NOT a ``FinalStatus`` — + the terminal-status set stays closed. + """ + + CRITERION_PASSED = "criterion_passed" + CRITERION_FAILED = "criterion_failed" + + +class EarlyStopInfo(BaseModel): + """Records why and when a run stopped early (``None`` when it ran to completion). + + Populated by the orchestrator's ``EarlyStopWatcher`` at the moment the armed + criteria are decided mid-run. ``early_stop is not None`` is itself the + "stopped early" flag — no separate bool. Serialized as part of + ``EvaluationResult`` to ``task.json``; defaults to ``None`` on old files, so + the round-trip is safe. + """ + + reason: EarlyStopReason = Field(description="Why the run stopped: armed criteria passed, or definitively failed.") + deciding_criterion_type: str = Field( + description="Type of the criterion whose live verdict fired the stop (the failing one on " + + "fail-stop; the last-to-pass on pass-stop)." + ) + deciding_criterion_description: str = Field( + description="Description of the deciding criterion (human-readable label from the task YAML)." + ) + armed_criteria: list[str] = Field( + default_factory=list, + description="'type: description' strings for the armed set, so reports can mark the rest " + + "advisory without re-deriving from task_config.", + ) + sdk_turn_index: int = Field( + description="SDK inner-turn count at the stop (watcher counts TurnStartEvents). NOT the " + + "orchestrator iteration, which is always 1 in single-shot." + ) + tool_call_index: int = Field(description="1-based index of the tool call that decided the stop.") + elapsed_seconds: float = Field(description="Wall-clock seconds from the first agent-start event to the stop.") + turns_remaining_at_stop: int | None = Field( + default=None, + description="max_turns - sdk_turn_index (an upper bound on turns avoided, not a measured " + + "saving); None when max_turns is unset.", + ) + + class EvaluationResult(BaseModel): """Complete result of a task evaluation.""" @@ -503,6 +554,16 @@ class EvaluationResult(BaseModel): ), ) + # Early-stop telemetry (only populated when the run was cut short by the + # armed-criteria watcher; None on a full run). See EarlyStopInfo. + early_stop: EarlyStopInfo | None = Field( + default=None, + description=( + "Why/when the run stopped early (reason, deciding criterion, SDK turn/tool index, elapsed). " + "None on a full run — 'early_stop is not None' is itself the stopped-early flag." + ), + ) + def calculate_weighted_score(self, criteria: list[SuccessCriterion]) -> None: """Calculate weighted average score from criterion results. @@ -547,6 +608,33 @@ def all_criteria_passed(self, criteria: list[SuccessCriterion]) -> bool: ) return all(r.score >= c.pass_threshold for r, c in zip(self.success_criteria_results, criteria, strict=True)) + def armed_criteria_passed(self, criteria: list[SuccessCriterion]) -> bool: + """True iff every ARMED criterion (``stop_when`` set) meets its pass_threshold. + + The early-stop gate: on an early-stopped run only the armed subset gates + ``final_status`` (non-armed criteria are advisory — recorded but never + decisive), so a smoke flavor is not dragged to FAILURE by criteria whose + work it deliberately skipped. Shares the same results/criteria length + pre-check as ``all_criteria_passed`` so the gate logic stays + single-sourced. Raises ``ValueError`` on an empty armed set — unreachable + when a stop actually fired (a stop requires an armed criterion), so this + is a defensive guard against misuse. + """ + if len(self.success_criteria_results) != len(criteria): + raise ValueError( + f"Results/criteria length mismatch for task {self.task_id}: " + + f"{len(self.success_criteria_results)} results vs {len(criteria)} criteria." + ) + armed = [ + (r, c) for r, c in zip(self.success_criteria_results, criteria, strict=True) if c.stop_when is not None + ] + if not armed: + raise ValueError( + f"armed_criteria_passed called with no armed criteria for task {self.task_id}; " + + "the early-stop gate is only valid when at least one criterion sets stop_when." + ) + return all(r.score >= c.pass_threshold for r, c in armed) + class CriterionStats(BaseModel): """Per-criterion-type aggregate stats across a suite's rows.""" diff --git a/src/coder_eval/orchestration/early_stop.py b/src/coder_eval/orchestration/early_stop.py index 8c43c79a..71e59b83 100644 --- a/src/coder_eval/orchestration/early_stop.py +++ b/src/coder_eval/orchestration/early_stop.py @@ -1,21 +1,42 @@ -"""Early-stop-on-criterion: resolution-time validation. +"""Early-stop-on-criterion: resolution-time validation + runtime watcher. Opt-in mechanism (``run_limits.stop_early``) that ends a single-shot run as soon as its *armed* criteria (those with ``stop_when`` set) are decided mid-run — on -pass or on a definitive fail. This module owns the resolution-time guardrails: -``validate_early_stop`` rejects every configuration v1 cannot honor as a hard -error, so an unsupported arming is never a silent no-op. The runtime watcher that -actually observes the event stream and trips the interrupt lands in a later -commit; until then arming is inert at runtime. +pass or on a definitive fail. This module owns the whole feature: + +* ``validate_early_stop`` — resolution-time guardrails. Rejects every + configuration v1 cannot honor as a hard error, so an unsupported arming is + never a silent no-op. +* ``EarlyStopWatcher`` — the runtime observer. A ``StreamCallback`` composed + into the agent's event stream that maintains its own ``EventCollector``, + evaluates every armed criterion's ``live_verdict`` on each tool completion, + applies the stop rule, and exposes ``should_stop()`` (the cooperative + interrupt the agent polls) plus ``info`` (the ``EarlyStopInfo`` the + orchestrator records). Fail-open: a raising ``live_verdict`` disarms the + watcher and degrades to a full run — a verdict bug can never cause a *false* + early stop. + +Live verdicts only *trigger* the stop; the authoritative scores always come +from the standard ``check_all`` on the frozen trajectory after the cut. """ from __future__ import annotations -from typing import TYPE_CHECKING +import logging +import time +from typing import TYPE_CHECKING, Any + +from coder_eval.models import EarlyStopInfo, EarlyStopReason +from coder_eval.streaming.collector import EventCollector +from coder_eval.streaming.events import AgentStartEvent, StreamEvent, ToolEndEvent, TurnStartEvent if TYPE_CHECKING: - from coder_eval.models import TaskDefinition + from coder_eval.criteria.base import BaseCriterion, LiveVerdict + from coder_eval.models import BaseSuccessCriterion, TaskDefinition + + +logger = logging.getLogger(__name__) class EarlyStopConfigError(ValueError): @@ -108,3 +129,173 @@ def validate_early_stop(task: TaskDefinition) -> None: f"criterion type {c.type!r} cannot decide polarity {missing} mid-run " + f"(stop_when={polarity!r}); it supports {sorted(polarities)}." ) + + +# Type alias for the armed pairs the watcher holds: (criterion model, its checker). +_ArmedPair = tuple["BaseSuccessCriterion", "BaseCriterion[Any]"] + + +class EarlyStopWatcher: + """Observes the agent event stream and trips the cooperative interrupt. + + A ``StreamCallback`` composed into the agent's callback chain (alone when + ``--stream`` is off, else beside the ``TaskScopedCallback``). It maintains + its OWN ``EventCollector`` — independent of the one the agent builds its + returned ``TurnRecord`` from — so each ``live_verdict`` sees a fresh, + single-element partial-trajectory list. On every tool completion it computes + all armed verdicts and applies the stop rule; once a stop fires (or the + watcher disarms on a raising verdict) the decision is latched and further + events are ignored. + + The orchestrator polls ``should_stop`` (passed to ``agent.communicate``) and, + after the turn, reads ``info`` to populate ``EvaluationResult.early_stop``. + + Fail-open: a ``live_verdict`` that raises disarms the watcher, logs + loudly, and degrades the run to a full run (``info`` stays ``None``). Because + live verdicts are triggers — not truth — this can never produce a *false* + early stop; it only ever errs toward running more. + """ + + def __init__( + self, + task_id: str, + armed: list[_ArmedPair], + *, + max_turns: int | None, + ) -> None: + self._task_id = task_id + self._armed = armed + self._max_turns = max_turns + self._collector = EventCollector() + self._sdk_turn_index = 0 + self._tool_call_index = 0 + self._started_monotonic: float | None = None + # Previous round's verdicts, for the "which criterion flipped to pass this + # round" attribution on pass-stop. Reassigned ONLY at the end of a + # non-firing evaluation, so it always holds the PREVIOUS round when a stop + # fires. Starts all-"undecided". + self._prev_verdicts: list[LiveVerdict] = ["undecided"] * len(armed) + self._info: EarlyStopInfo | None = None + self._disarmed = False + + @classmethod + def for_task(cls, task: TaskDefinition) -> EarlyStopWatcher: + """Build a watcher for an armed task (instantiates the armed criteria's checkers). + + The criteria registry is imported lazily here — it is not initialized at + module import time. Checker classes take no ctor args. + """ + from coder_eval.criteria import CriterionRegistry, init_criteria + + init_criteria(validate=False) + armed: list[_ArmedPair] = [ + (c, CriterionRegistry.get_checker(c.type)()) for c in task.success_criteria if c.stop_when is not None + ] + max_turns = task.run_limits.max_turns if task.run_limits is not None else None + return cls(task.task_id, armed, max_turns=max_turns) + + # --- StreamCallback -------------------------------------------------- # + + def on_event(self, event: StreamEvent) -> None: + """Forward the event to the internal collector; evaluate on tool completions. + + Short-circuits once the decision is latched (fired or disarmed). Counts + ``TurnStartEvent`` for ``sdk_turn_index`` and ``ToolEndEvent`` for the + 1-based ``tool_call_index``, and stamps the wall-clock origin at the + FIRST ``AgentStartEvent`` only (a retry's second AgentStart does not + reset it). + """ + if self._info is not None or self._disarmed: + return + if isinstance(event, AgentStartEvent): + if self._started_monotonic is None: + self._started_monotonic = time.monotonic() + elif isinstance(event, TurnStartEvent): + self._sdk_turn_index += 1 + elif isinstance(event, ToolEndEvent): + self._tool_call_index += 1 + self._collector.on_event(event) + if isinstance(event, ToolEndEvent): + self._evaluate() + + def should_stop(self) -> bool: + """The cooperative interrupt the agent polls after each dispatched message.""" + return self._info is not None + + @property + def info(self) -> EarlyStopInfo | None: + """The recorded stop info, or ``None`` if no stop fired (incl. after disarm).""" + return self._info + + @property + def disarmed(self) -> bool: + """True once a ``live_verdict`` raised and the watcher degraded to a full run.""" + return self._disarmed + + # --- Stop rule -------------------------------------------------- # + + def _evaluate(self) -> None: + records = [self._collector.build_turn_record()] + verdicts: list[LiveVerdict] = [] + for criterion, checker in self._armed: + try: + verdicts.append(checker.live_verdict(criterion, records)) + except Exception: + # Fail-open: a raising verdict disarms; the run degrades to full. + self._disarmed = True + logger.error( + "[%s] early-stop live_verdict raised for criterion %r; disarming watcher, " + + "run degrades to a full run", + self._task_id, + criterion.type, + exc_info=True, + ) + return + + # Fail-stop: first armed criterion (criteria order) that live-fails AND + # whose stop_when permits fail decides the run. + for (criterion, _checker), verdict in zip(self._armed, verdicts, strict=True): + if verdict == "fail" and criterion.stop_when in ("fail", "decided"): + self._fire(EarlyStopReason.CRITERION_FAILED, criterion) + return + + # Pass-stop: EVERY armed criterion live-passes AND each permits pass. + all_pass = all(v == "pass" for v in verdicts) + all_permit_pass = all(c.stop_when in ("pass", "decided") for c, _ in self._armed) + if all_pass and all_permit_pass: + # Deciding criterion = the last armed (criteria order) whose verdict + # flipped vs the previous round; fall back to the last armed. + deciding = self._armed[-1][0] + for (criterion, _checker), verdict, prev in zip(self._armed, verdicts, self._prev_verdicts, strict=True): + if verdict != prev: + deciding = criterion + self._fire(EarlyStopReason.CRITERION_PASSED, deciding) + return + + # No stop this round — record the verdicts so the next round can detect flips. + self._prev_verdicts = verdicts + + def _fire(self, reason: EarlyStopReason, criterion: BaseSuccessCriterion) -> None: + elapsed = 0.0 + if self._started_monotonic is not None: + elapsed = max(time.monotonic() - self._started_monotonic, 0.0) + turns_remaining = None if self._max_turns is None else max(self._max_turns - self._sdk_turn_index, 0) + self._info = EarlyStopInfo( + reason=reason, + deciding_criterion_type=criterion.type, + deciding_criterion_description=criterion.description, + armed_criteria=[f"{c.type}: {c.description}" for c, _ in self._armed], + sdk_turn_index=self._sdk_turn_index, + tool_call_index=self._tool_call_index, + elapsed_seconds=elapsed, + turns_remaining_at_stop=turns_remaining, + ) + logger.info( + "[%s] early-stop fired: reason=%s deciding=%s sdk_turn=%d tool_call=%d elapsed=%.2fs", + self._task_id, + reason.value, + criterion.type, + self._sdk_turn_index, + self._tool_call_index, + elapsed, + ) diff --git a/src/coder_eval/orchestrator.py b/src/coder_eval/orchestrator.py index d6820976..963e5ff2 100644 --- a/src/coder_eval/orchestrator.py +++ b/src/coder_eval/orchestrator.py @@ -50,12 +50,12 @@ UserMessage, resolve_route, ) -from .orchestration.early_stop import validate_early_stop +from .orchestration.early_stop import EarlyStopWatcher, validate_early_stop from .orchestration.evaluation import load_reference from .path_utils import format_task_log_id, task_log_path from .sandbox import Sandbox from .simulation import DialogStopReason, SimulatorResult, UserSimulator, evaluate_stop -from .streaming.callbacks import StreamCallback, TaskScopedCallback, safe_emit +from .streaming.callbacks import CompositeStreamCallback, StreamCallback, TaskScopedCallback, safe_emit from .streaming.events import CriteriaCheckEvent, CriterionSummary from .telemetry import Scalar, hash_identifier from .utils import get_version_info, looks_like_version, runtime_uip_versions @@ -366,6 +366,10 @@ def __init__( # Reference solution cache (loaded on-demand) self._reference_code: str | None = None + # Early-stop watcher (created in _setup only when run_limits.stop_early is + # armed; None otherwise, so the default path is entirely unaffected). + self._early_stop_watcher: EarlyStopWatcher | None = None + # One-shot flag: emit the "cost budget configured but no cost data" warning # exactly once per task even if _check_run_limits fires every turn. self._cost_budget_skipped_logged: bool = False @@ -851,6 +855,13 @@ async def _setup(self) -> None: # run_limits.stop_early is armed. validate_early_stop(self.task) + # Build the early-stop watcher once, up front, when armed. This sits BEFORE + # the evaluate-only early return below, so an armed evaluate-only re-grade + # builds an inert (never-fed) watcher — harmless, and keeps a single + # creation point. + if self.task.run_limits is not None and self.task.run_limits.stop_early: + self._early_stop_watcher = EarlyStopWatcher.for_task(self.task) + if self.sandbox is not None: # evaluate-only mode: sandbox already set up, skip agent assert self.result is not None @@ -1147,6 +1158,17 @@ async def _communicate_with_retry( if self.stream_callback is not None: agent_callback = TaskScopedCallback(self.stream_callback, self._log_task_id) + # Compose the early-stop watcher into the stream when armed. It is the + # sole callback when --stream is off, else it runs alongside the + # TaskScopedCallback. The same watcher instance persists across retry + # attempts (created once in _setup), so its turn/tool counters and + # wall-clock origin accumulate correctly. The closures below read `watcher`. + watcher = self._early_stop_watcher + if watcher is not None: + agent_callback = ( + CompositeStreamCallback([watcher, agent_callback]) if agent_callback is not None else watcher + ) + def _drain_pending_turn(*, attempt: int) -> None: """Read agent.pending_turn and, if set, append it to result.iterations.""" partial = agent.pending_turn @@ -1190,6 +1212,7 @@ async def _communicate_attempt() -> TurnRecord: stream_callback=agent_callback, timeout=turn_timeout, max_turns=max_turns, + should_stop=watcher.should_stop if watcher is not None else None, ) if turn_timeout is None: return await coro @@ -1347,6 +1370,10 @@ async def _evaluation_loop(self) -> bool: self.result.iterations.append(turn_record) self._sync_sandbox_command_path_with_agent() + # Record early-stop info (if the watcher tripped) BEFORE check_all, so it + # survives even if a checker raises. None on a full run or when unarmed. + self.result.early_stop = self._early_stop_watcher.info if self._early_stop_watcher is not None else None + logger.debug(f"Agent response received ({len(turn_record.agent_output)} chars)") # Check success criteria (pass reference code for reference_comparison criterion) @@ -1371,7 +1398,19 @@ async def _evaluation_loop(self) -> bool: pairs = list(zip(criteria_results, self.task.success_criteria, strict=True)) passed_count = sum(1 for r, c in pairs if r.score >= c.pass_threshold) total_count = len(pairs) - all_passed = self.result.all_criteria_passed(self.task.success_criteria) + if self.result.early_stop is not None: + # Early-stopped run: only the armed subset gates final_status; the rest + # are advisory (recorded, never decisive) so a smoke flavor is not + # dragged to FAILURE by criteria whose work it deliberately skipped. + all_passed = self.result.armed_criteria_passed(self.task.success_criteria) + armed_count = sum(1 for c in self.task.success_criteria if c.stop_when is not None) + logger.info( + "Early-stopped run: gating on %d armed criteria (%d advisory, not gated).", + armed_count, + total_count - armed_count, + ) + else: + all_passed = self.result.all_criteria_passed(self.task.success_criteria) # Reuse the model method for weighted score (single source of truth) self.result.calculate_weighted_score(self.task.success_criteria) diff --git a/tests/test_early_stop.py b/tests/test_early_stop.py index c2636c00..04f69419 100644 --- a/tests/test_early_stop.py +++ b/tests/test_early_stop.py @@ -1,20 +1,29 @@ -"""Tests for early-stop-on-criterion: config surface, live-verdict -contract, and resolution-time guardrails. +"""Tests for early-stop-on-criterion (phases 1-3). -This suite is inert at runtime (nothing invokes the interrupt yet); these tests -cover the pieces that DO exist: the two new opt-in config fields, the +Phase 1 (config + contract): the two new opt-in config fields, the ``live_verdict`` / ``live_stop_polarities`` observability contract on the two observable criteria, and ``validate_early_stop``'s guardrails on both the ``plan`` and ``run`` surfaces. + +Phase 2 (agent seam): the cooperative ``should_stop`` seam on +``ClaudeCodeAgent.communicate`` and the ``STOPPED_EARLY`` status, plus the +timeout-beats-stop precedence (a deadline breach wins over a pending stop). + +Phase 3 (feature live): the ``EarlyStopReason`` / ``EarlyStopInfo`` models and +the ``armed_criteria_passed`` gate; the ``EarlyStopWatcher`` runtime observer +(stop rule, fail-open, latching, attribution); and the orchestrator wiring +(watcher composed into the stream, ``result.early_stop`` populated, armed-subset +gate on an early-stopped run vs the full gate on a completed run). """ from __future__ import annotations +import asyncio import tempfile from collections.abc import Callable from datetime import datetime from typing import Any -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest @@ -23,11 +32,17 @@ from coder_eval.criteria.base import BaseCriterion from coder_eval.criteria.command_executed import CommandExecutedChecker from coder_eval.criteria.skill_triggered import SkillTriggeredChecker, _engaged_skill_names +from coder_eval.errors import TurnTimeoutError from coder_eval.models import ( AgentKind, CommandExecutedCriterion, CommandTelemetry, + CriterionResult, + EarlyStopInfo, + EarlyStopReason, + EvaluationResult, FileExistsCriterion, + FinalStatus, RunLimits, SandboxConfig, SimulationConfig, @@ -36,8 +51,16 @@ TurnRecord, parse_agent_config, ) -from coder_eval.orchestration.early_stop import EarlyStopConfigError, validate_early_stop -from coder_eval.streaming.events import AgentEndEvent, AgentEndStatus, TurnEndStatus +from coder_eval.orchestration.early_stop import EarlyStopConfigError, EarlyStopWatcher, validate_early_stop +from coder_eval.orchestrator import Orchestrator +from coder_eval.streaming.events import ( + AgentEndEvent, + AgentEndStatus, + AgentStartEvent, + ToolEndEvent, + TurnEndStatus, + TurnStartEvent, +) # --------------------------------------------------------------------------- # @@ -105,6 +128,63 @@ def _cmd_crit( ) +# --- Phase-3 helpers: results / info / events ------------------------------ # + + +def _crit_result(ctype: str, score: float) -> CriterionResult: + return CriterionResult(criterion_type=ctype, description=f"{ctype} result", score=score) + + +def _result(*, criteria_results: list[CriterionResult] | None = None) -> EvaluationResult: + return EvaluationResult( + task_id="r", + task_description="d", + agent_type="claude-code", + started_at=_TS, + final_status=FinalStatus.FAILURE, + iteration_count=1, + success_criteria_results=criteria_results or [], + ) + + +def _info(**overrides: Any) -> EarlyStopInfo: + base: dict[str, Any] = dict( + reason=EarlyStopReason.CRITERION_PASSED, + deciding_criterion_type="skill_triggered", + deciding_criterion_description="skill activation", + armed_criteria=["skill_triggered: skill activation"], + sdk_turn_index=1, + tool_call_index=1, + elapsed_seconds=1.0, + turns_remaining_at_stop=14, + ) + base.update(overrides) + return EarlyStopInfo(**base) + + +def _agent_start() -> AgentStartEvent: + return AgentStartEvent(task_id="t") + + +def _turn_start() -> TurnStartEvent: + return TurnStartEvent(task_id="t") + + +def _skill_cmd(skill: str, *, tool_id: str) -> CommandTelemetry: + return CommandTelemetry( + tool_name="Skill", tool_id=tool_id, timestamp=_TS, parameters={"skill": skill}, result_status="success" + ) + + +def _tool_end(cmd: CommandTelemetry) -> ToolEndEvent: + return ToolEndEvent(task_id="t", tool=cmd) + + +def _skill_events(skill: str, *, tool_id: str = "sk-1") -> list[Any]: + """AgentStart + TurnStart + a Skill ToolEnd engaging ``skill``.""" + return [_agent_start(), _turn_start(), _tool_end(_skill_cmd(skill, tool_id=tool_id))] + + # --------------------------------------------------------------------------- # # Config surface # --------------------------------------------------------------------------- # @@ -420,6 +500,45 @@ def _agent_end_events(sink: _EventSink) -> list[AgentEndEvent]: return [e for e in sink.events if isinstance(e, AgentEndEvent)] +class _NoopWatchdog: + """No-op stand-in for ThreadedWatchdog so only the in-loop deadline guard fires.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + def __enter__(self) -> _NoopWatchdog: + return self + + def __exit__(self, *args: Any) -> bool: + return False + + +async def _run_claude_communicate_timeout() -> tuple[ClaudeCodeAgent, _EventSink, BaseException | None]: + """Drive ``communicate`` with a slow query (50ms) against a 10ms deadline AND + ``should_stop=True`` — the deadline guard must win. Returns + ``(agent, sink, raised_exception)``.""" + config = parse_agent_config(type=AgentKind.CLAUDE_CODE, permission_mode="acceptEdits") + agent = ClaudeCodeAgent(config) + sink = _EventSink() + raised: BaseException | None = None + with tempfile.TemporaryDirectory() as tmpdir: + await agent.start(tmpdir) + + async def slow_query(prompt: Any, options: Any, transport: Any = None) -> Any: + await asyncio.sleep(0.05) + yield _DummyMsg(0) + + with ( + patch("coder_eval.agents.claude_code_agent.query", slow_query), + patch("coder_eval.agents.claude_code_agent.ThreadedWatchdog", _NoopWatchdog), + ): + try: + await agent.communicate("p", stream_callback=sink, timeout=0.01, should_stop=lambda: True) + except TurnTimeoutError as exc: + raised = exc + return agent, sink, raised + + class TestCooperativeStopSeam: def test_stopped_early_member_on_both_enums(self) -> None: assert AgentEndStatus.STOPPED_EARLY.value == "stopped_early" @@ -458,3 +577,383 @@ async def test_should_stop_false_consumes_full_stream(self) -> None: _agent, _record, sink, pulled = await _run_claude_communicate(never=True, n_messages=3) assert pulled == 3 assert _agent_end_events(sink)[0].status == AgentEndStatus.COMPLETED + + async def test_timeout_beats_stop_precedence(self) -> None: + # Both signals live in one turn: a deadline breach AND should_stop=True. + # The top-of-loop deadline guard returns BEFORE dispatch, so the stop + # check is never reached — TIMEOUT wins over the pending stop. + agent, sink, raised = await _run_claude_communicate_timeout() + assert isinstance(raised, TurnTimeoutError) + ends = _agent_end_events(sink) + assert len(ends) == 1 + assert ends[0].status == AgentEndStatus.TIMEOUT + assert ends[0].crashed is True + # The crashed partial is preserved for the orchestrator to drain. + assert agent.pending_turn is not None and agent.pending_turn.crashed is True + # STOPPED_EARLY must NOT appear — the stop lost the race. + assert AgentEndStatus.STOPPED_EARLY not in {e.status for e in ends} + + +# --------------------------------------------------------------------------- # +# Phase 3: EarlyStopReason / EarlyStopInfo / armed_criteria_passed +# --------------------------------------------------------------------------- # + + +class TestEarlyStopModels: + def test_reason_values(self) -> None: + assert EarlyStopReason.CRITERION_PASSED.value == "criterion_passed" + assert EarlyStopReason.CRITERION_FAILED.value == "criterion_failed" + + def test_info_defaults(self) -> None: + info = EarlyStopInfo( + reason=EarlyStopReason.CRITERION_FAILED, + deciding_criterion_type="command_executed", + deciding_criterion_description="d", + sdk_turn_index=2, + tool_call_index=3, + elapsed_seconds=1.5, + ) + assert info.armed_criteria == [] + assert info.turns_remaining_at_stop is None + + def test_info_roundtrip(self) -> None: + info = _info() + assert EarlyStopInfo.model_validate_json(info.model_dump_json()) == info + + def test_evaluation_result_early_stop_defaults_none(self) -> None: + assert _result().early_stop is None + + def test_evaluation_result_roundtrip_with_early_stop(self) -> None: + result = _result(criteria_results=[_crit_result("skill_triggered", 1.0)]) + result.early_stop = _info() + reloaded = EvaluationResult.model_validate_json(result.model_dump_json()) + assert reloaded.early_stop == _info() + + def test_armed_criteria_passed_gates_armed_only(self) -> None: + # Armed skill passes; advisory file_exists fails. armed gate -> True. + criteria = [ + _skill_crit("date-teller", "date-teller", stop_when="decided"), + FileExistsCriterion(path="x", description="x must exist"), + ] + result = _result(criteria_results=[_crit_result("skill_triggered", 1.0), _crit_result("file_exists", 0.0)]) + assert result.armed_criteria_passed(criteria) is True + # The full gate would (correctly) fail on the advisory 0.0. + assert result.all_criteria_passed(criteria) is False + + def test_armed_criteria_passed_fails_when_armed_fails(self) -> None: + criteria = [ + _skill_crit("date-teller", "date-teller", stop_when="decided"), + FileExistsCriterion(path="x", description="x must exist"), + ] + result = _result(criteria_results=[_crit_result("skill_triggered", 0.0), _crit_result("file_exists", 1.0)]) + assert result.armed_criteria_passed(criteria) is False + + def test_armed_criteria_passed_raises_on_empty_armed(self) -> None: + criteria = [FileExistsCriterion(path="x", description="x must exist")] + result = _result(criteria_results=[_crit_result("file_exists", 1.0)]) + with pytest.raises(ValueError, match="no armed criteria"): + result.armed_criteria_passed(criteria) + + +# --------------------------------------------------------------------------- # +# Phase 3: EarlyStopWatcher +# --------------------------------------------------------------------------- # + + +def _watcher(criteria: list[Any], *, max_turns: int | None = 20) -> EarlyStopWatcher: + task = _task(criteria=criteria, stop_early=True) + assert task.run_limits is not None + task.run_limits.max_turns = max_turns + return EarlyStopWatcher.for_task(task) + + +def _feed(watcher: EarlyStopWatcher, events: list[Any]) -> None: + for event in events: + watcher.on_event(event) + + +class TestEarlyStopWatcher: + def test_for_task_arms_only_stop_criteria(self) -> None: + watcher = _watcher( + [ + _skill_crit("date-teller", "date-teller", stop_when="decided"), + FileExistsCriterion(path="x", description="x must exist"), + ] + ) + # Only the armed criterion is tracked; the unarmed file_exists is ignored. + assert len(watcher._armed) == 1 + + def test_undecided_before_engagement_no_stop(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")]) + _feed(watcher, [_agent_start(), _turn_start()]) + assert watcher.should_stop() is False + assert watcher.info is None + + def test_pass_stop_fires_on_expected_skill(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")]) + _feed(watcher, _skill_events("date-teller")) + assert watcher.should_stop() is True + assert watcher.info is not None + assert watcher.info.reason == EarlyStopReason.CRITERION_PASSED + + def test_fail_stop_fires_on_wrong_skill(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")]) + _feed(watcher, _skill_events("weather-teller")) + assert watcher.should_stop() is True + assert watcher.info is not None + assert watcher.info.reason == EarlyStopReason.CRITERION_FAILED + + def test_pass_polarity_does_not_fire_on_fail(self) -> None: + # stop_when="pass": a wrong-skill (live-fail) engagement must NOT stop. + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="pass")]) + _feed(watcher, _skill_events("weather-teller")) + assert watcher.should_stop() is False + + def test_stacked_pass_stop(self) -> None: + # Two armed skill criteria, both expecting date-teller; engaging it passes both. + watcher = _watcher( + [ + _skill_crit("date-teller", "date-teller", stop_when="decided"), + _skill_crit("weather-teller", "", stop_when="decided"), + ] + ) + _feed(watcher, _skill_events("date-teller")) + assert watcher.should_stop() is True + assert watcher.info is not None + assert watcher.info.reason == EarlyStopReason.CRITERION_PASSED + + def test_stacked_wrong_skill_fail_stop(self) -> None: + # Engaging weather-teller: date-teller row -> fail; the fail-stop fires first. + watcher = _watcher( + [ + _skill_crit("date-teller", "date-teller", stop_when="decided"), + _skill_crit("weather-teller", "date-teller", stop_when="decided"), + ] + ) + _feed(watcher, _skill_events("weather-teller")) + assert watcher.info is not None + assert watcher.info.reason == EarlyStopReason.CRITERION_FAILED + + def test_records_turn_and_tool_index(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")]) + _feed(watcher, _skill_events("date-teller")) + assert watcher.info is not None + assert watcher.info.sdk_turn_index == 1 + assert watcher.info.tool_call_index == 1 + + def test_turns_remaining_from_max_turns(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")], max_turns=15) + _feed(watcher, _skill_events("date-teller")) + assert watcher.info is not None + assert watcher.info.turns_remaining_at_stop == 14 # 15 - sdk_turn_index(1) + + def test_turns_remaining_none_when_max_turns_unset(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")], max_turns=None) + _feed(watcher, _skill_events("date-teller")) + assert watcher.info is not None + assert watcher.info.turns_remaining_at_stop is None + + def test_fail_open_on_raising_verdict(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")]) + with patch.object(SkillTriggeredChecker, "live_verdict", side_effect=RuntimeError("boom")): + _feed(watcher, _skill_events("date-teller")) + # Fail-open: disarmed, no false stop, degrades to a full run. + assert watcher.disarmed is True + assert watcher.should_stop() is False + assert watcher.info is None + + def test_decision_latched_after_fire(self) -> None: + watcher = _watcher([_skill_crit("date-teller", "date-teller", stop_when="decided")]) + _feed(watcher, _skill_events("date-teller")) + fired = watcher.info + # A subsequent (wrong-skill) engagement must not overwrite the latched decision. + _feed(watcher, [_tool_end(_skill_cmd("weather-teller", tool_id="sk-2"))]) + assert watcher.info is fired + assert watcher.info is not None + assert watcher.info.reason == EarlyStopReason.CRITERION_PASSED + + +# --------------------------------------------------------------------------- # +# Phase 3: Orchestrator wiring +# --------------------------------------------------------------------------- # + + +class _ScriptedAgent: + """Duck-typed agent: replays scripted events through the callback, polling + ``should_stop`` after each and breaking when it flips (mirrors the real + message-boundary cut). Returns a fixed ``TurnRecord``.""" + + def __init__(self, events: list[Any], turn: TurnRecord) -> None: + self._events = events + self._turn = turn + self.pending_turn: TurnRecord | None = None + self.delivered = 0 + + def get_sdk_options(self) -> dict[str, Any] | None: + return None + + async def communicate( + self, + prompt: str, + *, + stream_callback: Any = None, + timeout: float | None = None, + max_turns: int | None = None, + should_stop: Callable[[], bool] | None = None, + ) -> TurnRecord: + for event in self._events: + if stream_callback is not None: + stream_callback.on_event(event) + self.delivered += 1 + if should_stop is not None and should_stop(): + break + return self._turn + + +async def _run_wiring( + *, + criteria: list[Any], + events: list[Any], + scores: list[float], + stop_early: bool, + tmp_path, +) -> tuple[EvaluationResult, _ScriptedAgent]: + """Drive ``Orchestrator._evaluation_loop`` with a scripted agent + mock checker. + + ``scores`` are positional CriterionResult scores matching ``criteria``. + The early-stop watcher is built directly (_setup is not invoked here). + """ + task = _task(criteria=criteria, stop_early=stop_early) + run_dir = tmp_path / "run" + run_dir.mkdir(parents=True) + orch = Orchestrator(task=task, run_dir=run_dir, variant_id="default") + orch.result = EvaluationResult( + task_id=task.task_id, + task_description=task.description, + variant_id="default", + agent_type=AgentKind.CLAUDE_CODE, + started_at=datetime.now(), + final_status=FinalStatus.FAILURE, + iteration_count=0, + environment_info={}, + ) + sandbox = MagicMock() + sandbox.sandbox_dir = tmp_path / "sandbox" + sandbox.sandbox_dir.mkdir() + orch.sandbox = sandbox + + checker = MagicMock() + checker.check_all = MagicMock(return_value=[_crit_result(c.type, s) for c, s in zip(criteria, scores, strict=True)]) + orch.success_checker = checker + + if stop_early: + orch._early_stop_watcher = EarlyStopWatcher.for_task(task) + + turn = TurnRecord(iteration=1, user_input="p", agent_output="done") + agent = _ScriptedAgent(events, turn) + orch.agent = agent # type: ignore[assignment] + + with patch("coder_eval.orchestrator.load_reference", return_value=(None, None, None)): + await orch._evaluation_loop() + assert orch.result is not None + return orch.result, agent + + +class TestOrchestratorEarlyStopWiring: + _SKILL = "date-teller" + + def _criteria(self, *, expected: str = "date-teller", stop_when: str | None = "decided") -> list[Any]: + # Armed skill_triggered + advisory file_exists (deliberately failing). + return [ + _skill_crit(self._SKILL, expected, stop_when=stop_when), + FileExistsCriterion(path="artifact.txt", description="artifact must exist"), + ] + + async def test_default_off_full_gate_no_early_stop(self, tmp_path) -> None: + # Unarmed: no watcher, all criteria gate, advisory 0.0 drags to FAILURE. + result, agent = await _run_wiring( + criteria=self._criteria(stop_when=None), + events=_skill_events(self._SKILL), + scores=[1.0, 0.0], + stop_early=False, + tmp_path=tmp_path, + ) + assert result.early_stop is None + assert agent.delivered == 3 # full stream consumed (should_stop=None) + + async def test_pass_stop_cuts_the_stream(self, tmp_path) -> None: + # A trailing event AFTER the deciding ToolEnd proves the cut: delivered == 3. + events = [*_skill_events(self._SKILL), _turn_start()] + result, agent = await _run_wiring( + criteria=self._criteria(), + events=events, + scores=[1.0, 0.0], + stop_early=True, + tmp_path=tmp_path, + ) + assert agent.delivered == 3 + assert result.early_stop is not None + assert result.early_stop.reason == EarlyStopReason.CRITERION_PASSED + + async def test_fail_stop_wiring(self, tmp_path) -> None: + result, _agent = await _run_wiring( + criteria=self._criteria(), + events=_skill_events("weather-teller"), + scores=[0.0, 0.0], + stop_early=True, + tmp_path=tmp_path, + ) + assert result.early_stop is not None + assert result.early_stop.reason == EarlyStopReason.CRITERION_FAILED + + async def test_early_stop_info_fields_populated(self, tmp_path) -> None: + result, _agent = await _run_wiring( + criteria=self._criteria(), + events=_skill_events(self._SKILL), + scores=[1.0, 0.0], + stop_early=True, + tmp_path=tmp_path, + ) + assert result.early_stop is not None + assert result.early_stop.sdk_turn_index == 1 + assert result.early_stop.tool_call_index == 1 + assert result.early_stop.deciding_criterion_type == "skill_triggered" + + async def test_advisory_not_gated_on_early_stop(self, tmp_path) -> None: + # Armed skill passes (1.0), advisory file_exists fails (0.0): armed gate -> SUCCESS. + result, _agent = await _run_wiring( + criteria=self._criteria(), + events=_skill_events(self._SKILL), + scores=[1.0, 0.0], + stop_early=True, + tmp_path=tmp_path, + ) + assert result.early_stop is not None + assert result.all_criteria_passed(self._criteria()) is False # full gate would fail + assert result.armed_criteria_passed(self._criteria()) is True # armed gate passes + + async def test_completed_naturally_uses_full_gate(self, tmp_path) -> None: + # Armed, but the skill is never engaged -> watcher never fires -> full gate, + # so the advisory 0.0 legitimately drags the completed run to FAILURE. + result, agent = await _run_wiring( + criteria=self._criteria(), + events=[_agent_start(), _turn_start()], # no skill engagement + scores=[0.0, 0.0], + stop_early=True, + tmp_path=tmp_path, + ) + assert result.early_stop is None + assert agent.delivered == 2 # full (short) stream consumed + assert result.all_criteria_passed(self._criteria()) is False + + async def test_fail_open_wiring_degrades_to_full_run(self, tmp_path) -> None: + with patch.object(SkillTriggeredChecker, "live_verdict", side_effect=RuntimeError("boom")): + result, _agent = await _run_wiring( + criteria=self._criteria(), + events=_skill_events(self._SKILL), + scores=[1.0, 0.0], + stop_early=True, + tmp_path=tmp_path, + ) + # Fail-open: no early_stop recorded, full gate applies. + assert result.early_stop is None From 1448cfafdbc1ce43f630a7a4d0e48281e75589e2 Mon Sep 17 00:00:00 2001 From: anonymous Date: Thu, 9 Jul 2026 16:39:15 -0700 Subject: [PATCH 4/7] feat(reports): early-stop surfaces, CE025 lint, docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Surface the opt-in early-stop feature end-to-end and lock the live-verdict contract with a lint rule. All render/telemetry-only for unarmed runs, so the default path is behavior-unchanged. - reports_experiment: eval_result_to_task_dict emits stopped_early / early_stop_reason / turns_remaining_at_stop so downstream analysis never confuses a truncated run with a full one. - reports: _runtime_notes_lines adds a per-task early-stop NOTE (reads the task-dict keys), with the "<= N turn(s) avoided" clause when known. - reports_html: header badge + per-criterion "advisory — not gated" markers on the non-armed rows of an early-stopped run. - orchestrator: build_task_event gains EarlyStopped / EarlyStopReason dims on CoderEval.Task.End. - CE025 (live-verdict consistency): a criterion's live_stop_polarities and live_verdict must agree — a non-empty polarity set without an override arms a dead criterion; an override without polarities is unreachable. Scoped to criteria/, base.py exempt. 10 rule tests + 7 report-surface tests. - experiments/early-stop-ab.yaml: smoke vs. e2e flavors from one file via the stop_early boolean per variant. - docs: TASK_DEFINITION_GUIDE (stop_early subsection + stop_when field row), AB_EXPERIMENTS (smoke-vs-e2e recipe + ToC), CLAUDE.md (architecture bullet, criteria-fields mention, tree updates, CE range CE001-CE025). --- CLAUDE.md | 10 +- docs/AB_EXPERIMENTS.md | 34 ++++++ docs/TASK_DEFINITION_GUIDE.md | 51 ++++++++ experiments/early-stop-ab.yaml | 23 ++++ src/coder_eval/orchestrator.py | 2 + src/coder_eval/reports.py | 8 ++ src/coder_eval/reports_experiment.py | 8 ++ src/coder_eval/reports_html.py | 18 ++- .../rules/ce025_live_verdict_consistency.py | 111 ++++++++++++++++++ tests/lint/runner.py | 2 + tests/test_custom_lint.py | 65 ++++++++++ tests/test_early_stop.py | 90 +++++++++++++- 12 files changed, 414 insertions(+), 8 deletions(-) create mode 100644 experiments/early-stop-ab.yaml create mode 100644 tests/lint/rules/ce025_live_verdict_consistency.py diff --git a/CLAUDE.md b/CLAUDE.md index 667acac3..5c5cf392 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -36,7 +36,7 @@ coder_eval/ │ ├── experiment.py # ExperimentDefinition, ExperimentVariant, ResolvedTask, result models │ ├── judge_defaults.py # DEFAULT_JUDGE_MODEL constant (cycle-free leaf) │ ├── mutations.py # PromptMutation variants (prefix/suffix/replace/template/rephrase) -│ ├── results.py # CriterionResult (+ ClassificationCriterionResult), TurnRecord, EvaluationResult, CriterionAggregate, ThresholdCheck, SuiteRollup +│ ├── results.py # CriterionResult (+ ClassificationCriterionResult), TurnRecord, EvaluationResult, EarlyStopInfo/EarlyStopReason, CriterionAggregate, ThresholdCheck, SuiteRollup │ ├── routing.py # ApiRoute (DirectRoute/BedrockRoute) │ ├── sandbox.py # SandboxConfig, ResourceLimits │ ├── tasks.py # TaskDefinition, AgentConfig, Dataset (dataset fan-out + sample) @@ -79,6 +79,7 @@ coder_eval/ ├── orchestration/ # Batch execution utilities │ ├── batch.py # Parallel task execution (run_batch + run_batch_resolved) │ ├── config.py # Batch run configuration +│ ├── early_stop.py # validate_early_stop guardrails + EarlyStopWatcher (armed live-verdict observer) │ ├── evaluation.py # Evaluation helpers │ ├── experiment.py # ExperimentRunner, resolve_task_for_variant, load_experiment │ └── task_loader.py # YAML task loading @@ -138,6 +139,7 @@ templates/ # Sandbox template directories - **Reconciliation message (stream self-reconciles to the turn total)**: The per-message stream consistently under-reports the authoritative turn total — a fixed prompt slice (~512 input tokens on Claude) is billed on no SDK-emitted message, and sub-agent input/cache only partially bubbles up. So `EventCollector.build_turn_record` appends one synthetic `ReconciliationMessage` (`role="reconciliation"`, in the `TranscriptMessage` union) per turn, carrying the per-bucket residual = `token_usage` − Σ(assistant message buckets). The invariant: **summing the four token buckets across `TurnRecord.messages` (assistant + reconciliation) equals `token_usage` exactly**, for both Claude and Codex (Codex's stream is already complete after `_recover_subagent_tool_calls`, so its residual is usually 0 and no entry is emitted). This is what lets the evalboard SUM the message stream as the source of truth instead of reading a separate aggregate ("agent tokens"): `selectTokenTotals` returns the stream sum whenever a reconciliation entry is present, and the timeline renders it as its own row. It is agent-agnostic (booked at the single `EventCollector` seam), carries no cost (cost stays on `token_usage`), and is excluded from generation/turn counts and the cost simulator. The Python `token_usage`/`total_token_usage` aggregate is unchanged and still authoritative for budget/judges/reports. - **sandbox isolation**: Tasks that don't need MCP servers should set `setting_sources: []` in their `agent:` block to isolate the sandbox from the host project's CLAUDE.md and settings. Without this, the host project's CLAUDE.md (often 20 KB+) is injected into every API call, inflating cache-creation tokens and cost significantly. - **Run-time caps (non-criterion enforcement)**: `TaskDefinition.run_limits` (`RunLimits` model) is the single namespace for all run-time caps — `max_turns` / `task_timeout` / `turn_timeout` (structural) and `max_input_tokens` / `max_output_tokens` / `max_total_tokens` / `max_usd` (cumulative budget). Token/USD breaches abort with `FinalStatus.TOKEN_BUDGET_EXCEEDED` or `COST_BUDGET_EXCEEDED` (both `category == "failed"`). Structural caps are set from the CLI via `-D run_limits.max_turns=…` / `-D run_limits.task_timeout=…` / `-D run_limits.turn_timeout=…` (field-merged into `run_limits`); budget caps via `-D run_limits.max_usd=…` etc. or YAML. Layered config uses field-merge — a variant block overrides individual keys without replacing the task's block. +- **Early stop on criterion (opt-in)**: `run_limits.stop_early` (default off) ends a single-shot Claude run early once the run's **armed** criteria are decided, so a raised `max_turns` isn't wasted on the smoke flavor. A criterion is armed by `stop_when: pass|fail|decided`; only criteria that can decide from a partial trajectory may arm (they declare a non-empty `live_stop_polarities` ClassVar and override `live_verdict` on `BaseCriterion` — currently `skill_triggered`, `command_executed`; CE025 enforces the two stay consistent). It uses a cooperative `should_stop` seam on the Claude agent's between-messages guard (tool-call granularity, no SIGKILL) driven by `orchestration/early_stop.py::EarlyStopWatcher` (own `EventCollector` + stop rule). Live verdicts only *trigger* the stop; the standard `check_all` on the frozen trajectory is authoritative. An early-stopped run gates on the **armed subset** (`EvaluationResult.armed_criteria_passed`); a completed run gates on the full set. Every unsupported use (non-observable criterion, non-Claude agent, wrong polarity, no armed criterion, simulation mode) is an error at resolution (plan *and* run), and a runtime verdict bug **fails open** to a full run. Surfaces: `EarlyStopInfo` (reason + deciding criterion + when), report notes/badges, `stopped_early` run.json rows, and `EarlyStopped`/`EarlyStopReason` telemetry dims. Defaults off ⇒ behavior byte-for-behavior unchanged. ## Success Criteria (14 types) @@ -158,7 +160,7 @@ templates/ # Sandbox template directories | `llm_judge` | Continuous | LLM grades artifacts + optional trajectory + optional reference; routes through the run's backend (Bedrock / Anthropic) | | `agent_judge` | Continuous | Spawns a Claude Code SDK agent in an isolated sandbox copy; judge uses tools (Bash/Read/Grep/…) to investigate and returns a JSON verdict. Expensive; runs with evaluator credentials — see SECURITY note in the criterion docstring. | -All criteria support `weight` (default 1.0) and `pass_threshold` (default 0.9). On dataset-backed tasks, criteria may also set `suite_thresholds: {metric: min_value}` — the suite gate passes iff every listed metric (from the criterion's `aggregate()` output) meets its minimum. +All criteria support `weight` (default 1.0) and `pass_threshold` (default 0.9), plus `stop_when` (`pass`/`fail`/`decided`, default `null`) which arms the criterion for early stop when `run_limits.stop_early` is set (observable criteria only). On dataset-backed tasks, criteria may also set `suite_thresholds: {metric: min_value}` — the suite gate passes iff every listed metric (from the criterion's `aggregate()` output) meets its minimum. ## Evaluation Flow @@ -191,11 +193,11 @@ make format # ruff format make check # ruff check (lint) make typecheck # pyright make test # pytest -make lint # custom architectural lint rules (CE001–CE013) +make lint # custom architectural lint rules (CE001–CE025) make verify # All of the above + coverage check (CI equivalent) ``` -When fixing a bug, ask: *could a custom lint rule have prevented this?* If the root cause is a mechanically detectable pattern (e.g., "always import from `coder_eval.models`", "never call blocking IO in async"), add a rule to `tests/lint/rules/` following the CE001–CE013 pattern and wire it up in `tests/lint/runner.py`. This turns a one-time fix into permanent enforcement. See `tests/test_custom_lint.py` for how rules are tested. +When fixing a bug, ask: *could a custom lint rule have prevented this?* If the root cause is a mechanically detectable pattern (e.g., "always import from `coder_eval.models`", "never call blocking IO in async"), add a rule to `tests/lint/rules/` following the CE001–CE025 pattern and wire it up in `tests/lint/runner.py`. This turns a one-time fix into permanent enforcement. See `tests/test_custom_lint.py` for how rules are tested. ## Configuration diff --git a/docs/AB_EXPERIMENTS.md b/docs/AB_EXPERIMENTS.md index 4ba7c2b0..9bc0da09 100644 --- a/docs/AB_EXPERIMENTS.md +++ b/docs/AB_EXPERIMENTS.md @@ -17,6 +17,7 @@ to the orchestrator are involved. - [Recipe: A/B a Skill](#recipe-ab-a-skill) - [Recipe: A/B a Model](#recipe-ab-a-model) - [Recipe: A/B a Prompt](#recipe-ab-a-prompt) +- [Recipe: Smoke vs. e2e Flavors (Early Stop)](#recipe-smoke-vs-e2e-flavors-early-stop) - [Replicates (Statistical Power)](#replicates-statistical-power) - [Measuring the Difference](#measuring-the-difference) - [CLI Reference](#cli-reference) @@ -222,6 +223,39 @@ variants: The full mutation catalog (prefix / suffix / replace / template / rephrase) is defined in `coder_eval/models/mutations.py`. +## Recipe: Smoke vs. e2e Flavors (Early Stop) + +Run the **same** task file as both a fast `smoke` flavor and a full `e2e` flavor +by flipping one boolean per variant — `run_limits.stop_early`. Arm the criteria +that define "the interesting thing happened" with `stop_when` in the task file; +the `smoke` variant cuts off as soon as they're decided, while `e2e` runs to +completion. Because the field merge is per-key, the variant sets only +`stop_early` without disturbing the task's `max_turns`. + +```yaml +experiment_id: early-stop-ab +description: "Smoke vs. e2e from one file via opt-in early stop" + +variants: + - variant_id: e2e + run_limits: + stop_early: false # full run to completion (the reference flavor) + - variant_id: smoke + run_limits: + stop_early: true # cut off once the armed criteria are decided +``` + +The task file supplies the arming (`stop_when` on the criteria that gate the +flavor) and a `max_turns` generous enough for `e2e`; see +[`stop_early`](TASK_DEFINITION_GUIDE.md#stop_early-opt-in-early-stop). This recipe +ships as `experiments/early-stop-ab.yaml`. + +Expect **identical pass/fail verdicts** between the two variants — an +early-stopped run is gated on the armed subset only, and the non-armed criteria +become advisory (clearly marked in the report), so the `smoke` flavor can't +"pass for free" — with the `smoke` variant significantly lower on turns, +duration, and tokens. + ## Replicates (Statistical Power) Agents are stochastic — a single run per arm is noise, not signal. Set `repeats` diff --git a/docs/TASK_DEFINITION_GUIDE.md b/docs/TASK_DEFINITION_GUIDE.md index 9cb58983..8d26121f 100644 --- a/docs/TASK_DEFINITION_GUIDE.md +++ b/docs/TASK_DEFINITION_GUIDE.md @@ -192,6 +192,56 @@ budgets consistently across a suite — the headline % is only comparable when tasks are measured against realistic, like-for-like targets. Omit it (the default) to exclude a task from the metric entirely. +### `stop_early` (opt-in early stop) + +`run_limits.stop_early` (default `false`) ends a single-shot run **early** once +the run's **armed** criteria are decided — so you can raise `max_turns` for the +full-run flavor without paying for turns the smoke flavor doesn't need. A +criterion is *armed* by giving it a `stop_when` (see the criterion-fields table); +`stop_early` is the master switch that turns arming on for the run. + +```yaml +run_limits: + max_turns: 30 + stop_early: true # opt in; default false leaves behavior unchanged +success_criteria: + - type: skill_triggered + skill_name: date-teller + expected_skill: date-teller + stop_when: decided # arm on pass OR definitive fail + - type: file_exists # not armed → advisory on an early-stopped run + path: report.md +``` + +Semantics: + +- **Opt-in, per run.** With `stop_early: false` (the default) the run behaves + exactly as before — `stop_when` is inert and every criterion gates normally. +- **Polarity.** `stop_when: pass` stops the moment all armed criteria are decided + in the pass direction; `stop_when: fail` stops on a definitive wrong-signal + fail; `stop_when: decided` stops on either. Only criteria that can decide from a + partial trajectory (currently `skill_triggered`, `command_executed`) may be + armed — arming any other criterion is a hard error at resolution (plan *and* + run), never a silent no-op. +- **Verdict.** An early-stopped run is gated on the **armed subset only**; the + non-armed criteria become **advisory** and are clearly marked (report badge + + per-criterion note + `stopped_early` row). A run that completes naturally is + gated on the **full** set, as always. This is what lets one file serve both a + `smoke` flavor (`stop_early: true`) and an `e2e` flavor (`stop_early: false`) + with identical verdicts — see [AB_EXPERIMENTS.md](AB_EXPERIMENTS.md). +- **Fail-safe.** A live-verdict bug **fails open** to a full run (logged loudly) — + it can never silently disable a criterion or cause a false early stop. + +Observability (every early-stopped run is flagged everywhere so analysis never +compares a truncated run against a full one): + +| Surface | Field / marker | +|---------|----------------| +| `run.json` row | `stopped_early`, `early_stop_reason`, `turns_remaining_at_stop` | +| `run.md` | `> **NOTE:** […] stopped early (); <= N turn(s) avoided …` | +| `task.html` | header badge `stopped early ()` + `advisory — not gated` markers | +| Telemetry | `EarlyStopped` / `EarlyStopReason` dimensions on `CoderEval.Task.End` | + ## Sandbox Configuration The `sandbox` block is optional. When omitted, it defaults to `driver: "tempdir"` with standard Python environment. @@ -319,6 +369,7 @@ All criteria share these fields: | `description` | — | Human-readable description (required) | | `weight` | 1.0 | Relative importance for weighted score | | `pass_threshold` | 0.9 | Minimum score (0.0–1.0) to pass | +| `stop_when` | `null` | Arms this criterion for early stop (`pass`/`fail`/`decided`); requires `run_limits.stop_early: true` and an observable criterion type (`skill_triggered`, `command_executed`). See [`stop_early`](#stop_early-opt-in-early-stop). | **Scoring types:** - **Binary** (1.0 or 0.0): `file_exists`, `run_command`, `file_matches_regex` diff --git a/experiments/early-stop-ab.yaml b/experiments/early-stop-ab.yaml new file mode 100644 index 00000000..f5116836 --- /dev/null +++ b/experiments/early-stop-ab.yaml @@ -0,0 +1,23 @@ +experiment_id: early-stop-ab +description: | + Smoke vs. e2e flavors from ONE task file via opt-in early-stop. Both variants + share the same tasks, criteria, and max_turns; they differ only in the + run_limits.stop_early boolean (field-merged, so neither replaces the task's + run_limits block). The task's armed criteria (those carrying `stop_when`) + decide when the `smoke` variant cuts off — as soon as the designated criteria + are decided — while `e2e` runs every task to completion. + + Expect identical pass/fail verdicts between the two variants (the armed subset + gates an early-stopped run; other criteria are advisory and clearly marked), + with the `smoke` variant significantly lower on turns, duration, and tokens. + +variants: + - variant_id: e2e + description: "Full run to completion — no early stop (the reference flavor)." + run_limits: + stop_early: false + + - variant_id: smoke + description: "Opt-in early stop — cut off once the armed criteria are decided." + run_limits: + stop_early: true diff --git a/src/coder_eval/orchestrator.py b/src/coder_eval/orchestrator.py index 963e5ff2..0c629c95 100644 --- a/src/coder_eval/orchestrator.py +++ b/src/coder_eval/orchestrator.py @@ -249,6 +249,8 @@ def build_task_event(result: EvaluationResult, *, driver: str, variant_id: str) "AgentType": result.agent_type or "", "Model": result.model_used or "", "Driver": driver, + "EarlyStopped": result.early_stop is not None, + "EarlyStopReason": (result.early_stop.reason.value if result.early_stop is not None else ""), } return "CoderEval.Task.End", props diff --git a/src/coder_eval/reports.py b/src/coder_eval/reports.py index e62fd6a8..6e5b9deb 100644 --- a/src/coder_eval/reports.py +++ b/src/coder_eval/reports.py @@ -418,6 +418,14 @@ def _runtime_notes_lines(summary: RunSummary) -> list[str]: notes.append( f"> **WARNING:** [{task_id}] expected_turns exceeded: {actual}/{expected} (cumulative SDK turns)" ) + if t.get("stopped_early"): + reason = t.get("early_stop_reason") or "unknown" + turns_remaining = t.get("turns_remaining_at_stop") + avoided = f" <= {turns_remaining} turn(s) avoided —" if isinstance(turns_remaining, int) else "" + notes.append( + f"> **NOTE:** [{task_id}] stopped early ({reason});{avoided}" + + " gated on armed criteria only; other criteria are advisory" + ) if not notes: return [] return ["## Run-time Notes", "", *notes] diff --git a/src/coder_eval/reports_experiment.py b/src/coder_eval/reports_experiment.py index bacb1a29..9f9935b7 100644 --- a/src/coder_eval/reports_experiment.py +++ b/src/coder_eval/reports_experiment.py @@ -138,6 +138,14 @@ def eval_result_to_task_dict( "visible_turns": visible_turn_count(result), "expected_turns": expected_turns_value, "has_final_reply": has_reply, + # Early-stop surfaces (opt-in run_limits.stop_early). None/False on the + # default path so downstream analysis never confuses a truncated run + # with a full one. + "stopped_early": result.early_stop is not None, + "early_stop_reason": (result.early_stop.reason.value if result.early_stop is not None else None), + "turns_remaining_at_stop": ( + result.early_stop.turns_remaining_at_stop if result.early_stop is not None else None + ), } d["variant_id"] = variant_id return d diff --git a/src/coder_eval/reports_html.py b/src/coder_eval/reports_html.py index f29687ff..57184eff 100644 --- a/src/coder_eval/reports_html.py +++ b/src/coder_eval/reports_html.py @@ -24,6 +24,7 @@ from coder_eval.models import ( CommandTelemetry, CriterionResult, + EarlyStopInfo, EvaluationResult, ExperimentDefinition, ExperimentResult, @@ -341,6 +342,12 @@ def _render_header(result: EvaluationResult) -> str: if overage is not None: actual, expected = overage expected_turns_badge = f'expected_turns exceeded ({actual}/{expected})' + early_stop_badge = "" + if result.early_stop is not None: + early_stop_badge = ( + 'stopped early ({_esc(result.early_stop.reason.value)})' + ) return f"""
@@ -354,6 +361,7 @@ def _render_header(result: EvaluationResult) -> str: {_esc(duration)} {cost_badge} {expected_turns_badge} + {early_stop_badge} Toggle theme
@@ -554,7 +562,7 @@ def _render_judge_transcript(transcript: Any) -> str: ) -def _render_criteria(results: list[CriterionResult]) -> str: +def _render_criteria(results: list[CriterionResult], early_stop: EarlyStopInfo | None = None) -> str: if not results: return """
@@ -563,13 +571,17 @@ def _render_criteria(results: list[CriterionResult]) -> str: the checker ran).

""" + armed = set(early_stop.armed_criteria) if early_stop is not None else set() rows: list[str] = [] for cr in results: + advisory = "" + if early_stop is not None and f"{cr.criterion_type}: {cr.description}" not in armed: + advisory = '
advisory — not gated (run stopped early)
' rows.append( f""" {_esc(cr.criterion_type)} - {_esc(cr.description)} + {_esc(cr.description)}{advisory} {_score_pill(cr.score)} {_render_criteria_details(cr)} @@ -1443,7 +1455,7 @@ def generate_task_html(result: EvaluationResult) -> str: body = ( _render_header(result) + _render_simulation(result) - + _render_criteria(result.success_criteria_results or []) + + _render_criteria(result.success_criteria_results or [], result.early_stop) + _render_judge_section(result.success_criteria_results or []) + _render_error_details(result) + f"

Conversation Trace ({trace_count})

" diff --git a/tests/lint/rules/ce025_live_verdict_consistency.py b/tests/lint/rules/ce025_live_verdict_consistency.py new file mode 100644 index 00000000..162c1c8d --- /dev/null +++ b/tests/lint/rules/ce025_live_verdict_consistency.py @@ -0,0 +1,111 @@ +"""CE025: a criterion's ``live_stop_polarities`` and ``live_verdict`` must agree. + +The early-stop live-verdict contract (``criteria/base.py``) pairs two members on +every ``BaseCriterion`` subclass: + +- ``live_stop_polarities: ClassVar[frozenset[str]]`` — the polarities the + criterion can decide from a PARTIAL, mid-run trajectory. Empty (the base + default) means "not observable mid-run", so the criterion can never arm + early-stop. +- ``live_verdict(...)`` — the method that actually reads the partial trajectory + and returns ``"pass"``/``"fail"``/``"undecided"``. + +The two must move together: a non-empty ``live_stop_polarities`` without a +``live_verdict`` override arms a criterion whose base ``live_verdict`` always +returns ``"undecided"`` (it can never stop — a silent dead arm); a +``live_verdict`` override without a non-empty ``live_stop_polarities`` writes +decision logic the arming path can never reach (dead code that reads as +supported). Either drift is a mechanically detectable bug, so flag it. + +Scoped to ``src/coder_eval/criteria/`` and exempts ``criteria/base.py`` (which +legitimately declares the empty default alongside the default ``live_verdict``). + +Non-empty detection: ``frozenset({...})`` / ``frozenset([...])`` with a +non-empty literal is non-empty; bare ``frozenset()`` is empty; a ``frozenset`` +call over a non-literal argument (or any other RHS) is conservatively treated as +non-empty. + +Add ``# noqa: CE025`` on the class line for a deliberate exception. +""" + +import ast +import re + +from tests.lint.rules.base import BaseRule + + +_CRITERIA_DIR = re.compile(r"[/\\]coder_eval[/\\]criteria[/\\]") +_BASE_FILE = re.compile(r"[/\\]coder_eval[/\\]criteria[/\\]base\.py$") + + +def _is_frozenset(func: ast.expr) -> bool: + return isinstance(func, ast.Name) and func.id == "frozenset" + + +def _polarities_nonempty(value: ast.expr) -> bool: + """Whether the RHS of a ``live_stop_polarities`` assignment is non-empty. + + ``frozenset()`` reads the literal's element count; bare + ``frozenset()`` is empty; anything else is conservatively non-empty so a + computed value is never mistaken for a dead arm. + """ + if isinstance(value, ast.Call) and _is_frozenset(value.func): + if not value.args: + return False + arg = value.args[0] + if isinstance(arg, ast.Set | ast.List | ast.Tuple): + return len(arg.elts) > 0 + if isinstance(arg, ast.Dict): + return len(arg.keys) > 0 + # Non-literal argument (a name, comprehension, …) — conservatively non-empty. + return True + # Not a frozenset() call at all — conservatively treat the declaration as non-empty. + return True + + +class LiveVerdictConsistency(BaseRule): + id = "CE025" + + def __init__(self, filepath: str) -> None: + super().__init__(filepath) + self._active = bool(_CRITERIA_DIR.search(filepath)) and not bool(_BASE_FILE.search(filepath)) + + def check(self, tree: ast.AST) -> list: # type: ignore[override] + if not self._active or not isinstance(tree, ast.Module): + return [] + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef): + self._check_class(node) + return self.violations + + def _check_class(self, node: ast.ClassDef) -> None: + has_polarities = False + polarities_nonempty = False + has_live_verdict = False + for stmt in node.body: + if isinstance(stmt, ast.FunctionDef) and stmt.name == "live_verdict": + has_live_verdict = True + elif isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name): + if stmt.target.id == "live_stop_polarities" and stmt.value is not None: + has_polarities = True + polarities_nonempty = _polarities_nonempty(stmt.value) + elif isinstance(stmt, ast.Assign): + for target in stmt.targets: + if isinstance(target, ast.Name) and target.id == "live_stop_polarities": + has_polarities = True + polarities_nonempty = _polarities_nonempty(stmt.value) + + if has_polarities and polarities_nonempty and not has_live_verdict: + self.violation( + node, + "declares a non-empty `live_stop_polarities` but no `live_verdict` override; " + "the base `live_verdict` always returns 'undecided', so this arms a criterion " + "that can never stop (add a `live_verdict` override or clear the polarities).", + ) + elif has_live_verdict and not (has_polarities and polarities_nonempty): + self.violation( + node, + "overrides `live_verdict` but declares no non-empty `live_stop_polarities`; " + "the arming path can never reach this decision logic (declare the polarities " + "it supports or drop the override).", + ) diff --git a/tests/lint/runner.py b/tests/lint/runner.py index e360b8ec..966c08ad 100644 --- a/tests/lint/runner.py +++ b/tests/lint/runner.py @@ -20,6 +20,7 @@ from tests.lint.rules.ce022_dialog_loop_statement_cap import SimulationDialogLoopStatementCap from tests.lint.rules.ce023_no_proxy_shim_import import NoProxyShimImports from tests.lint.rules.ce024_discriminated_unions import DiscriminatedUnions +from tests.lint.rules.ce025_live_verdict_consistency import LiveVerdictConsistency from tests.lint.rules.no_agent_timing_access import NoAgentTimingAccess from tests.lint.rules.no_blocking_io_in_async import NoBlockingIoInAsync from tests.lint.rules.no_cli_imports_in_core import NoCliImportsInCore @@ -63,6 +64,7 @@ SimulationDialogLoopStatementCap, NoProxyShimImports, DiscriminatedUnions, + LiveVerdictConsistency, ] # Anti-shadow invariant (mirrors AgentRegistry / register_pricing): every CE rule diff --git a/tests/test_custom_lint.py b/tests/test_custom_lint.py index 76a0ae4d..125c649f 100644 --- a/tests/test_custom_lint.py +++ b/tests/test_custom_lint.py @@ -530,6 +530,71 @@ def test_flags_annassign_union(self): assert self._run(self._TAGGED_CLASSES + "X: object = A | B") +@pytest.mark.lint +class TestCE025LiveVerdictConsistency: + """CE025 flags criteria whose `live_stop_polarities` and `live_verdict` disagree.""" + + _POLARITIES_NONEMPTY = ' live_stop_polarities: ClassVar[frozenset[str]] = frozenset({"pass", "fail"})\n' + _POLARITIES_EMPTY = " live_stop_polarities: ClassVar[frozenset[str]] = frozenset()\n" + _POLARITIES_PLAIN = ' live_stop_polarities = frozenset({"pass"})\n' + _POLARITIES_NONLITERAL = " live_stop_polarities: ClassVar[frozenset[str]] = frozenset(_SOME_SET)\n" + _LIVE_VERDICT = ' def live_verdict(self, criterion, turn_records):\n return "undecided"\n' + + @staticmethod + def _cls(*members: str) -> str: + body = "".join(members) if members else " pass\n" + return "from typing import ClassVar\nclass FakeChecker:\n" + body + + @staticmethod + def _run(src: str, *, path: str = "src/coder_eval/criteria/fake_checker.py"): + import ast + + from tests.lint.rules.ce025_live_verdict_consistency import LiveVerdictConsistency + + return LiveVerdictConsistency(path).check(ast.parse(src)) + + def test_flags_polarities_without_live_verdict(self): + assert self._run(self._cls(self._POLARITIES_NONEMPTY)) + + def test_flags_live_verdict_without_polarities(self): + assert self._run(self._cls(self._LIVE_VERDICT)) + + def test_flags_live_verdict_with_empty_polarities(self): + assert self._run(self._cls(self._POLARITIES_EMPTY, self._LIVE_VERDICT)) + + def test_allows_polarities_with_live_verdict(self): + assert not self._run(self._cls(self._POLARITIES_NONEMPTY, self._LIVE_VERDICT)) + + def test_allows_neither_declared(self): + # A plain checker (e.g. file_exists) that arms nothing declares neither member. + assert not self._run(self._cls(" path: str\n")) + + def test_allows_empty_polarities_without_live_verdict(self): + assert not self._run(self._cls(self._POLARITIES_EMPTY)) + + def test_detects_plain_assign_polarities(self): + # Non-empty `live_stop_polarities` via a plain (non-annotated) assignment still arms. + assert self._run(self._cls(self._POLARITIES_PLAIN)) + + def test_non_literal_arg_treated_nonempty(self): + # A computed frozenset() argument is conservatively non-empty → live_verdict required. + assert self._run(self._cls(self._POLARITIES_NONLITERAL)) + assert not self._run(self._cls(self._POLARITIES_NONLITERAL, self._LIVE_VERDICT)) + + def test_scope_exemptions(self): + # base.py legitimately pairs the empty default with the default live_verdict; and + # files outside criteria/ are out of scope entirely. + offending = self._cls(self._LIVE_VERDICT) + assert not self._run(offending, path="src/coder_eval/criteria/base.py") + assert not self._run(offending, path="src/coder_eval/orchestration/x.py") + + def test_real_criteria_tree_is_clean(self): + from tests.lint.rules.ce025_live_verdict_consistency import LiveVerdictConsistency + + criteria_dir = SRC / "coder_eval" / "criteria" + assert not check_paths([criteria_dir], rules=[LiveVerdictConsistency]) + + @pytest.mark.lint class TestCE009YamlModelsForbidExtras: """CE009 fires on non-compliant BaseModel classes in every scoped module.""" diff --git a/tests/test_early_stop.py b/tests/test_early_stop.py index 04f69419..cc098050 100644 --- a/tests/test_early_stop.py +++ b/tests/test_early_stop.py @@ -44,6 +44,7 @@ FileExistsCriterion, FinalStatus, RunLimits, + RunSummary, SandboxConfig, SimulationConfig, SkillTriggeredCriterion, @@ -52,7 +53,10 @@ parse_agent_config, ) from coder_eval.orchestration.early_stop import EarlyStopConfigError, EarlyStopWatcher, validate_early_stop -from coder_eval.orchestrator import Orchestrator +from coder_eval.orchestrator import Orchestrator, build_task_event +from coder_eval.reports import ReportGenerator +from coder_eval.reports_experiment import eval_result_to_task_dict +from coder_eval.reports_html import _render_criteria, _render_header from coder_eval.streaming.events import ( AgentEndEvent, AgentEndStatus, @@ -957,3 +961,87 @@ async def test_fail_open_wiring_degrades_to_full_run(self, tmp_path) -> None: ) # Fail-open: no early_stop recorded, full gate applies. assert result.early_stop is None + + +# --------------------------------------------------------------------------- # +# Report / telemetry surfaces +# --------------------------------------------------------------------------- # + + +def _stopped_result( + *, + reason: EarlyStopReason = EarlyStopReason.CRITERION_PASSED, + turns_remaining: int | None = 14, + criteria_results: list[CriterionResult] | None = None, +) -> EvaluationResult: + result = _result(criteria_results=criteria_results) + result.early_stop = _info(reason=reason, turns_remaining_at_stop=turns_remaining) + return result + + +def _run_summary(task_dicts: list[dict[str, Any]]) -> RunSummary: + # framework_version is a required RunSummary field; the count invariant needs + # succeeded + failed + error == tasks_run. + return RunSummary( + run_id="r", + start_time=_TS, + end_time=_TS, + total_duration_seconds=0.0, + tasks_run=len(task_dicts), + tasks_succeeded=len(task_dicts), + tasks_failed=0, + tasks_error=0, + task_results=task_dicts, + framework_version="test", + ) + + +class TestEarlyStopReportSurfaces: + def test_task_dict_keys_present_when_early_stopped(self) -> None: + d = eval_result_to_task_dict(_stopped_result()) + assert d["stopped_early"] is True + assert d["early_stop_reason"] == "criterion_passed" + assert d["turns_remaining_at_stop"] == 14 + + def test_task_dict_keys_defaulted_when_not_early_stopped(self) -> None: + d = eval_result_to_task_dict(_result()) + assert d["stopped_early"] is False + assert d["early_stop_reason"] is None + assert d["turns_remaining_at_stop"] is None + + def test_runtime_note_rendered_with_turns_avoided(self) -> None: + lines = ReportGenerator._runtime_notes_lines(_run_summary([eval_result_to_task_dict(_stopped_result())])) + blob = "\n".join(lines) + assert "stopped early (criterion_passed)" in blob + assert "<= 14 turn(s) avoided" in blob + assert "gated on armed criteria only; other criteria are advisory" in blob + + def test_runtime_note_absent_for_unarmed_run(self) -> None: + lines = ReportGenerator._runtime_notes_lines(_run_summary([eval_result_to_task_dict(_result())])) + assert not any("stopped early" in line for line in lines) + + def test_html_header_shows_early_stop_badge(self) -> None: + html = _render_header(_stopped_result()) + assert "stopped early (criterion_passed)" in html + # No badge on a normal run. + assert "stopped early" not in _render_header(_result()) + + def test_html_criteria_marks_only_advisory_rows(self) -> None: + # Armed skill_triggered (matches _info.armed_criteria) + advisory file_exists. + armed = _crit_result("skill_triggered", 1.0) + armed.description = "skill activation" # match the armed_criteria key format + advisory = _crit_result("file_exists", 0.0) + result = _stopped_result(criteria_results=[armed, advisory]) + html = _render_criteria(result.success_criteria_results, result.early_stop) + assert html.count("advisory — not gated (run stopped early)") == 1 + # A completed (non-early-stopped) run gets no advisory markers at all. + assert "advisory — not gated" not in _render_criteria([armed, advisory], None) + + def test_telemetry_dims_reflect_early_stop(self) -> None: + _name, props = build_task_event(_stopped_result(), driver="tempdir", variant_id="v") + assert props["EarlyStopped"] is True + assert props["EarlyStopReason"] == "criterion_passed" + # Defaulted on a normal run. + _n2, props2 = build_task_event(_result(), driver="tempdir", variant_id="v") + assert props2["EarlyStopped"] is False + assert props2["EarlyStopReason"] == "" From 20320a40a29c620c954772ab37aeb73d842a6a61 Mon Sep 17 00:00:00 2001 From: anonymous Date: Fri, 10 Jul 2026 10:09:01 -0700 Subject: [PATCH 5/7] docs(early-stop): tutorial pointer + renumber harness-candidate CE ids - docs/tutorials/04-writing-a-task.md: add a "Where to go deeper" pointer to the stop_early / stop_when early-stop docs so the feature is discoverable from the intro tutorial (field detail stays in the Task Definition Guide). - .claude/harness-candidates.md: CE024 (discriminated-unions) and CE025 (live-verdict consistency) are now implemented rules, so the proposed candidates were renumbered off the taken ids to CE026 / CE027 / CE028, with a note that the tests/lint/runner.py id-uniqueness assert is the source of truth. --- .claude/harness-candidates.md | 14 ++++++++++---- docs/tutorials/04-writing-a-task.md | 1 + 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/.claude/harness-candidates.md b/.claude/harness-candidates.md index a4eee6eb..cbfb4970 100644 --- a/.claude/harness-candidates.md +++ b/.claude/harness-candidates.md @@ -5,21 +5,27 @@ Deferred lint/test guardrails surfaced during reviews. Promote to a `CExxx` rule ## From code review 260701-1954 (fix-review-top5 run) — deferred to a dedicated guardrail plan -- **CE024** — workflow-YAML rule: forbid any `uses:` step pinned to a floating ref +> Numbering note: these are *proposed* ids. `CE024` (discriminated-unions) and +> `CE025` (live-verdict consistency) have since been **implemented** for other +> rules, so the candidates below were renumbered to the next free ids. Always +> claim the next unused number in `tests/lint/rules/` — the id-uniqueness assert +> in `tests/lint/runner.py` is the source of truth. + +- **CE026** — workflow-YAML rule: forbid any `uses:` step pinned to a floating ref (`@v3`, `@main`) rather than a 40-hex commit SHA. Would have caught `mxschmitt/action-tmate@v3` (fixed manually in this run). >30 min: needs a non-Python file-walk branch in `tests/lint/runner.py`. -- **CE025** — retired-token grep gate: fail when a removed-subsystem token +- **CE027** — retired-token grep gate: fail when a removed-subsystem token (`LLMGW_`, `API_BACKEND=proxy`, `uipath_llmgw_client`) reappears outside an allowlist across docs/config/src. Would have caught the LLM-Gateway residue swept in this run. >30 min: needs an allowlist + repo-wide text scan. -- **CE026** — assert the Makefile `lint:` help does not hardcode a stale `CE0NN` +- **CE028** — assert the Makefile `lint:` help does not hardcode a stale `CE0NN` upper bound (use `CE001+`). Would have caught the `CE001–CE005` drift fixed here. - **docs-vs-harness smoke test** — execute the CI tutorial's `coder-eval run` command against a NoOp task and assert the produced tree matches the documented globs. Would have caught the `--run-dir runs` layout bug. Not statically reachable (needs a live run). -- [ ] CE-rule: `type: Literal[...]` fields on models in `coder_eval/models/` must declare their tag default (`type: Literal["x"] = "x"`) — a member without the default degrades `validate_registry` diagnostics (PydanticUndefined in expected_types) and breaks direct construction. Nothing guards it today; needs a rule-design call (second violation class inside CE024 vs. a new CE025), and the failure is already double-caught by the MINIMAL_PAYLOADS parity test + direct-construction tests — caught in the 2026-07-03 top5-review-fixes run (Phase 1 quality review). +- [ ] CE-rule: `type: Literal[...]` fields on models in `coder_eval/models/` must declare their tag default (`type: Literal["x"] = "x"`) — a member without the default degrades `validate_registry` diagnostics (PydanticUndefined in expected_types) and breaks direct construction. Nothing guards it today; needs a rule-design call (second violation class inside CE024 vs. a new CExxx at the next free id), and the failure is already double-caught by the MINIMAL_PAYLOADS parity test + direct-construction tests — caught in the 2026-07-03 top5-review-fixes run (Phase 1 quality review). ## From 2026-07-03 open-source docs cleanup diff --git a/docs/tutorials/04-writing-a-task.md b/docs/tutorials/04-writing-a-task.md index 0683d523..bba05de5 100644 --- a/docs/tutorials/04-writing-a-task.md +++ b/docs/tutorials/04-writing-a-task.md @@ -95,6 +95,7 @@ created are preserved under `runs/latest////` (`task.json`, ## Where to go deeper - **All 14 criterion types, weights, thresholds** → [Task Definition Guide](../TASK_DEFINITION_GUIDE.md) +- **Stop a run early once the key criteria are decided** (opt-in `run_limits.stop_early` + `stop_when` on a criterion) → [Task Definition Guide → `stop_early`](../TASK_DEFINITION_GUIDE.md#stop_early-opt-in-early-stop) - **Fan one task out over a dataset of rows** → [Bring Your Own Data](../BYOD.md) - **Full CLI & config reference** → [User Guide](../USER_GUIDE.md) - **Compare two configurations on this task** → [Tutorial 05](05-comparing-models.md) From a6a51779f8535ecfe2da87f521c366fd3920a2bb Mon Sep 17 00:00:00 2001 From: anonymous Date: Fri, 10 Jul 2026 14:00:57 -0700 Subject: [PATCH 6/7] test(early-stop): cover plan/run resolution-surface guardrail wiring The five early-stop guardrails were unit-tested against validate_early_stop directly, but nothing asserted the resolution surfaces actually invoke it. Add integration tests driving the real wiring with real task YAMLs: - run surface: resolve_all_tasks raises EarlyStopConfigError for a bad arming (not demoted to a skipped task), accepts a valid armed task, and validates an arming introduced only by the layer-5 -D override (proving the guardrails run after _apply_cli_overrides). - plan surface: plan_command flips its exit code to 1 and names the early-stop config error for a bad arming; a valid armed task still plans clean. --- tests/test_early_stop.py | 127 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 127 insertions(+) diff --git a/tests/test_early_stop.py b/tests/test_early_stop.py index cc098050..705eed6a 100644 --- a/tests/test_early_stop.py +++ b/tests/test_early_stop.py @@ -22,12 +22,15 @@ import tempfile from collections.abc import Callable from datetime import datetime +from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch import pytest +import typer from coder_eval.agents.claude_code_agent import ClaudeCodeAgent +from coder_eval.cli.plan_command import plan_command from coder_eval.criteria import CriterionRegistry, init_criteria from coder_eval.criteria.base import BaseCriterion from coder_eval.criteria.command_executed import CommandExecutedChecker @@ -41,6 +44,8 @@ EarlyStopInfo, EarlyStopReason, EvaluationResult, + ExperimentDefinition, + ExperimentVariant, FileExistsCriterion, FinalStatus, RunLimits, @@ -52,7 +57,9 @@ TurnRecord, parse_agent_config, ) +from coder_eval.orchestration.config import BatchRunConfig from coder_eval.orchestration.early_stop import EarlyStopConfigError, EarlyStopWatcher, validate_early_stop +from coder_eval.orchestration.experiment import resolve_all_tasks from coder_eval.orchestrator import Orchestrator, build_task_event from coder_eval.reports import ReportGenerator from coder_eval.reports_experiment import eval_result_to_task_dict @@ -428,6 +435,126 @@ def test_stacked_activation_criteria_accept(self) -> None: validate_early_stop(task) # no raise +# --------------------------------------------------------------------------- # +# Guardrail integration: the plan and run resolution surfaces actually invoke +# validate_early_stop (not just the helper in isolation). Real task YAMLs go +# through the real load + 5-layer merge; a bad arming must surface as a clean +# CLI-level error on BOTH surfaces, never a silent no-op. +# --------------------------------------------------------------------------- # + +_ARMED_UNOBSERVABLE_CRITERION = """\ + - type: file_exists + description: out exists + path: out.txt + stop_when: pass +""" + +_ARMED_OBSERVABLE_CRITERION = """\ + - type: skill_triggered + description: date-teller activation + skill_name: date-teller + expected_skill: date-teller + stop_when: decided +""" + + +def _write_task_yaml(tmp_path: Path, *, criterion_yaml: str, stop_early: bool) -> Path: + task_file = tmp_path / "es_task.yaml" + task_file.write_text( + "task_id: es-guardrail-task\n" + + "description: early-stop guardrail surface test\n" + + "initial_prompt: do the thing\n" + + "agent:\n" + + " type: claude-code\n" + + "sandbox:\n" + + " driver: tempdir\n" + + "run_limits:\n" + + " max_turns: 20\n" + + f" stop_early: {str(stop_early).lower()}\n" + + "success_criteria:\n" + + criterion_yaml + ) + return task_file + + +def _resolve_surface(task_file: Path, tmp_path: Path, *, overrides: dict[str, Any] | None = None): + """Drive the real run-surface resolution (load + 5-layer merge + guardrails).""" + single_variant = [ExperimentVariant(variant_id="default")] + return resolve_all_tasks( + task_files=[task_file], + experiment=ExperimentDefinition(experiment_id="exp", variants=single_variant), + default_experiment=ExperimentDefinition(experiment_id="default", variants=single_variant), + config=BatchRunConfig(run_dir=tmp_path / "runs", overrides=overrides or {}), + ) + + +class TestGuardrailResolutionSurfaces: + """A bad arming is rejected by the real plan/run wiring, not only the helper.""" + + def test_run_surface_rejects_bad_armed_task(self, tmp_path: Path) -> None: + # The armed unobservable criterion propagates out of resolve_all_tasks as + # EarlyStopConfigError (a ValueError, so the run CLI converts it to a + # clean BadParameter) instead of being demoted to a skipped task. + task_file = _write_task_yaml(tmp_path, criterion_yaml=_ARMED_UNOBSERVABLE_CRITERION, stop_early=True) + with pytest.raises(EarlyStopConfigError, match="observable"): + _resolve_surface(task_file, tmp_path) + + def test_run_surface_accepts_valid_armed_task(self, tmp_path: Path) -> None: + task_file = _write_task_yaml(tmp_path, criterion_yaml=_ARMED_OBSERVABLE_CRITERION, stop_early=True) + resolved, skipped = _resolve_surface(task_file, tmp_path) + assert not skipped + assert len(resolved) == 1 + limits = resolved[0].task.run_limits + assert limits is not None and limits.stop_early is True + + def test_run_surface_validates_cli_override_arming(self, tmp_path: Path) -> None: + # The YAML alone is inert (stop_when set, stop_early false) and must be + # accepted; arming via the layer-5 -D override must then be validated, + # proving the guardrails run AFTER _apply_cli_overrides. + task_file = _write_task_yaml(tmp_path, criterion_yaml=_ARMED_UNOBSERVABLE_CRITERION, stop_early=False) + resolved, _ = _resolve_surface(task_file, tmp_path) + assert len(resolved) == 1 # inert without the override + with pytest.raises(EarlyStopConfigError, match="observable"): + _resolve_surface(task_file, tmp_path, overrides={"run_limits.stop_early": True}) + + def _run_plan(self, task_file: Path, exp_dir: Path) -> tuple[str, int]: + """Invoke the real plan_command against a minimal single-variant experiment. + + Returns the concatenated console output and the exit code (0 when plan + returned normally). + """ + exp_file = exp_dir / "es_experiment.yaml" + exp_file.write_text("experiment_id: es-guardrail\nvariants:\n - variant_id: default\n") + exit_code = 0 + with ( + patch("coder_eval.cli.plan_command.check_tools"), + patch("coder_eval.cli.plan_command.check_api_keys"), + # Point the default-experiment lookup at a nonexistent file so plan + # falls back to the explicit experiment (hermetic vs. the repo tree). + patch("coder_eval.orchestration.experiment.DEFAULT_EXPERIMENT_PATH", exp_dir / "missing.yaml"), + patch("coder_eval.cli.plan_command.console") as mock_console, + ): + try: + plan_command(task_files=[task_file], experiment=exp_file) + except typer.Exit as exc: + exit_code = exc.exit_code + printed = " ".join(str(call) for call in mock_console.print.call_args_list) + return printed, exit_code + + def test_plan_surface_flips_exit_code_on_bad_armed_task(self, tmp_path: Path) -> None: + task_file = _write_task_yaml(tmp_path, criterion_yaml=_ARMED_UNOBSERVABLE_CRITERION, stop_early=True) + printed, exit_code = self._run_plan(task_file, tmp_path) + assert exit_code == 1 + assert "early-stop config error" in printed + assert "observable" in printed + + def test_plan_surface_accepts_valid_armed_task(self, tmp_path: Path) -> None: + task_file = _write_task_yaml(tmp_path, criterion_yaml=_ARMED_OBSERVABLE_CRITERION, stop_early=True) + printed, exit_code = self._run_plan(task_file, tmp_path) + assert exit_code == 0 + assert "All tasks are valid!" in printed + + # --------------------------------------------------------------------------- # # Cooperative should_stop seam on ClaudeCodeAgent — still UNWIRED: the # orchestrator does not pass should_stop yet, so these drive the agent directly. From fcb9cccd286639a80e8d127d1957505c0f61ed4c Mon Sep 17 00:00:00 2001 From: anonymous Date: Fri, 10 Jul 2026 14:51:57 -0700 Subject: [PATCH 7/7] refactor(early-stop): make the _ArmedPair alias TYPE_CHECKING-only CodeQL flagged the Any and BaseCriterion imports as unused: their only references sat inside the quoted type-alias string tuple["BaseSuccessCriterion", "BaseCriterion[Any]"], which static analyzers do not resolve. Move the alias into the TYPE_CHECKING block unquoted so the usages are real name references. Behavior-neutral: the alias is referenced only from annotations, which are lazy under 'from __future__ import annotations'. --- src/coder_eval/orchestration/early_stop.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/coder_eval/orchestration/early_stop.py b/src/coder_eval/orchestration/early_stop.py index 71e59b83..6e5b3bff 100644 --- a/src/coder_eval/orchestration/early_stop.py +++ b/src/coder_eval/orchestration/early_stop.py @@ -35,6 +35,12 @@ from coder_eval.criteria.base import BaseCriterion, LiveVerdict from coder_eval.models import BaseSuccessCriterion, TaskDefinition + # Armed pair the watcher holds: (criterion model, its checker). Lives in the + # TYPE_CHECKING block (only annotations reference it, and those are lazy under + # `from __future__ import annotations`), so the names are real references + # rather than quoted strings static analyzers cannot resolve. + _ArmedPair = tuple[BaseSuccessCriterion, BaseCriterion[Any]] + logger = logging.getLogger(__name__) @@ -131,10 +137,6 @@ def validate_early_stop(task: TaskDefinition) -> None: ) -# Type alias for the armed pairs the watcher holds: (criterion model, its checker). -_ArmedPair = tuple["BaseSuccessCriterion", "BaseCriterion[Any]"] - - class EarlyStopWatcher: """Observes the agent event stream and trips the cooperative interrupt.