diff --git a/docs/agents/ANTIGRAVITY.md b/docs/agents/ANTIGRAVITY.md index a7cee48f..522dc8bf 100644 --- a/docs/agents/ANTIGRAVITY.md +++ b/docs/agents/ANTIGRAVITY.md @@ -168,18 +168,15 @@ as every other agent. ## Known limitations -1. **No cooperative early stop.** Antigravity does not support the `should_stop` - seam, so `run_limits.stop_early` is unsupported for this agent (it errors at - resolution, as it does for any non-Claude agent). -2. **No endpoint routing.** Only `GEMINI_API_KEY` + `ANTIGRAVITY_MODEL` are read — +1. **No endpoint routing.** Only `GEMINI_API_KEY` + `ANTIGRAVITY_MODEL` are read — there is no base-URL, project, region, or gateway override. -3. **Default-model drift.** The runtime fallback (`gemini-3.5-flash`) may differ from +2. **Default-model drift.** The runtime fallback (`gemini-3.5-flash`) may differ from what a given release's docs or example tasks pin; always set `agent.model` explicitly for reproducible runs. -4. **`kill_sync()` is best-effort.** The SDK's cancel/disconnect are async-only, so +3. **`kill_sync()` is best-effort.** The SDK's cancel/disconnect are async-only, so the watchdog's synchronous kill only flips agent state to `ERROR`; real teardown happens on the subsequent async `stop()`. -5. **Process-global spawn lock.** The SDK spawns `localharness` via a subprocess with +4. **Process-global spawn lock.** The SDK spawns `localharness` via a subprocess with no env-injection seam, so the agent transiently mutates `PATH` across the spawn under a process-wide lock. This serializes harness startup across concurrent tasks (it does not serialize the turns themselves). diff --git a/docs/agents/CLAUDE_CODE.md b/docs/agents/CLAUDE_CODE.md index d8e6075c..7bc830a0 100644 --- a/docs/agents/CLAUDE_CODE.md +++ b/docs/agents/CLAUDE_CODE.md @@ -145,13 +145,14 @@ simulator force `[]` for the same reason.) UiPath CLI plugin discovery; when unset the sandbox derives it from the resolved `uip` binary. See [User Guide → Environment Variables](../USER_GUIDE.md#environment-variables). -## Early stop (Claude-only) +## Early stop -Claude Code is the only agent that supports the cooperative early-stop seam. With +Claude Code supports the cooperative early-stop seam, as do the +[Codex](CODEX.md) and [Antigravity](ANTIGRAVITY.md) agents. With `run_limits.stop_early: true`, a single-shot run ends cleanly at the next tool-call boundary once its **armed** criteria (`stop_when: pass|fail|decided`) are decided — so a raised `max_turns` isn't wasted on a smoke run. Early stop errors at -resolution for any other agent. See the +resolution for any agent that does not declare `supports_cooperative_stop`. See the [Task Definition Guide](../TASK_DEFINITION_GUIDE.md) for the full contract. ## Telemetry diff --git a/docs/agents/CODEX.md b/docs/agents/CODEX.md index ce84b727..b020e76b 100644 --- a/docs/agents/CODEX.md +++ b/docs/agents/CODEX.md @@ -216,6 +216,7 @@ The Codex SDK is synchronous. The agent uses `_run_async()` helper to detect and | **Session Resume** | `--resume {session_id}` | Via thread ID | | **Permissions** | `permission_mode` + `allowed_tools` | `permission_mode` → sandbox/approval + `allowed_tools`/`disallowed_tools` → thread config | | **Tool Enforcement** | Not enforced by Coder Eval wrapper | `enabled_tools` honored; `disabled_tools` NOT enforced by the SDK | +| **Early stop** | Supported (cooperative `should_stop`, polled between messages) | Supported — polled after each streamed notification; the in-flight turn is interrupted best-effort | ## Known Limitations diff --git a/src/coder_eval/agents/antigravity_agent.py b/src/coder_eval/agents/antigravity_agent.py index 8506154b..b27527f9 100644 --- a/src/coder_eval/agents/antigravity_agent.py +++ b/src/coder_eval/agents/antigravity_agent.py @@ -27,7 +27,7 @@ from contextlib import AsyncExitStack from datetime import datetime from pathlib import Path -from typing import Any +from typing import Any, ClassVar from coder_eval.agent import Agent, AgentState from coder_eval.agents._logging import PrefixedAdapter @@ -185,6 +185,10 @@ def _to_token_usage(usage: Any, model: str | None) -> TokenUsage: class AntigravityAgent(Agent[AntigravityAgentConfig]): """Implementation of the Agent interface for Google Antigravity (Gemini).""" + # The step loop has a between-steps 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: AntigravityAgentConfig, @@ -413,10 +417,10 @@ async def communicate( ) -> 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. + ``should_stop`` is the cooperative early-stop callback, polled after each + processed step. When it returns True the step loop breaks, the + conversation is cancelled (best-effort) and the turn finalizes cleanly as + ``STOPPED_EARLY`` (``crashed=False``). Drives one logical turn: ``conversation.send(prompt)`` then iterate ``receive_steps()`` until the turn goes idle, mapping the Gemini step @@ -468,8 +472,21 @@ def _on_turn_timeout() -> None: conversation = self._sdk_agent.conversation try: await conversation.send(user_input) + # The cooperative should_stop poll runs AFTER process_step (the + # emission that lets the watcher latch on the deciding tool + # call) and BEFORE the next step is pulled — the deciding step + # is kept, the next is not. No-op when should_stop is None. async for step in conversation.receive_steps(): state.process_step(step) + if should_stop is not None and should_stop(): + state.stopped_early_hit = True + self._log.debug("Cooperative stop requested; ending step loop at this boundary") + break + if state.stopped_early_hit: + # Best-effort server-side cancel, mirrors kill(); a raising + # cancel() lands in the guarded handler below. + with contextlib.suppress(Exception): + await conversation.cancel() except asyncio.CancelledError: if state.timeout_hit: self._finalize_and_raise_timeout(state.finalize, timeout or 0) @@ -477,9 +494,17 @@ def _on_turn_timeout() -> None: except Exception as e: if state.timeout_hit: self._finalize_and_raise_timeout(state.finalize, timeout or 0, cause=e) - self._finalize_and_raise_crash( - state.finalize, truncate_crash_message(f"Antigravity turn failed: {e!s}"), cause=e - ) + if state.stopped_early_hit: + # The turn already stopped cleanly (e.g. the generator's + # aclose() raised on the break); escalating to a crash + # would trigger the orchestrator's retry with the watcher's + # decision still latched → immediate stop-at-turn-0 on the + # retry (wasted spend). Fall through to the clean tail. + self._log.warning("Ignoring post-stop exception; finalizing as STOPPED_EARLY: %s", e) + else: + self._finalize_and_raise_crash( + state.finalize, truncate_crash_message(f"Antigravity turn failed: {e!s}"), cause=e + ) if state.timeout_hit: # Watchdog fired but the pump finished before the cancel landed. @@ -493,13 +518,20 @@ def _on_turn_timeout() -> None: state.finalize(AgentEndStatus.CRASHED, crashed=True, crash_reason="turn cancelled") raise except Exception as e: - self._finalize_and_raise_crash( - state.finalize, truncate_crash_message(f"Antigravity turn failed: {e!s}"), cause=e - ) + if state.stopped_early_hit and not state.timeout_hit: + # Same retry-poisoning guard as the inner handler: a cooperative + # stop already happened, so finalize cleanly instead of crashing. + self._log.warning("Ignoring post-stop exception; finalizing as STOPPED_EARLY: %s", e) + else: + self._finalize_and_raise_crash( + state.finalize, truncate_crash_message(f"Antigravity turn failed: {e!s}"), cause=e + ) self._state = AgentState.WORKING self._end_turn_ok() - state.finalize(AgentEndStatus.COMPLETED, crashed=False, crash_reason=None) + # Precedence matches Claude: timeout (raised above) > stopped_early > completed. + status = AgentEndStatus.STOPPED_EARLY if state.stopped_early_hit else AgentEndStatus.COMPLETED + state.finalize(status, crashed=False, crash_reason=None) return collector.build_turn_record() async def stop(self) -> None: @@ -589,6 +621,7 @@ def __init__( self.turn_start_time = turn_start_time self.timeout_hit = False + self.stopped_early_hit = False self.finalized = False self.total_usage = TokenUsage() diff --git a/src/coder_eval/agents/codex_agent.py b/src/coder_eval/agents/codex_agent.py index 9be3274c..b22407bc 100644 --- a/src/coder_eval/agents/codex_agent.py +++ b/src/coder_eval/agents/codex_agent.py @@ -12,7 +12,7 @@ from collections.abc import Callable from datetime import datetime from pathlib import Path -from typing import Any +from typing import Any, ClassVar from urllib.parse import urlparse from coder_eval.agent import Agent, AgentState @@ -280,6 +280,7 @@ def __init__( self.iteration = iteration self.turn_start_time = turn_start_time self.timeout_hit = False + self.stopped_early_hit = False self.finalized = False # Live pump scratch (set during streaming). @@ -634,6 +635,10 @@ def finalize(self, status: AgentEndStatus, *, crashed: bool = False, crash_reaso class CodexAgent(Agent[CodexAgentConfig]): """Implementation of the Agent interface for OpenAI Codex using the Codex SDK.""" + # The notification pump has a between-items 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: CodexAgentConfig, @@ -740,10 +745,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. + should_stop: Cooperative early-stop callback, polled after each + dispatched notification. When it returns True the pump breaks, + the in-flight turn is interrupted (best-effort) and the turn + finalizes cleanly as ``STOPPED_EARLY`` (``crashed=False``). Returns: TurnRecord containing the complete interaction @@ -822,7 +827,7 @@ def _on_turn_timeout() -> None: # return; a crash skips this, so finalize reads the crash # defaults (None/"") and falls back to the captured messages. state.result_turn, state.sdk_token_usage, state.result_text = await self._run_turn_with_streaming( - state + state, should_stop ) except asyncio.CancelledError: if state.timeout_hit: @@ -831,9 +836,16 @@ def _on_turn_timeout() -> None: except Exception as e: if state.timeout_hit: self._finalize_and_raise_timeout(state.finalize, timeout or 0, cause=e) - self._finalize_and_raise_crash( - state.finalize, truncate_crash_message(f"Codex turn failed: {e!s}"), cause=e - ) + if state.stopped_early_hit: + # The turn already stopped cleanly; escalating to a crash + # would trigger the orchestrator's retry with the watcher's + # decision still latched → immediate stop-at-turn-0 on the + # retry (wasted spend). Fall through to the clean tail. + self._log.warning("Ignoring post-stop exception; finalizing as STOPPED_EARLY: %s", e) + else: + self._finalize_and_raise_crash( + state.finalize, truncate_crash_message(f"Codex turn failed: {e!s}"), cause=e + ) if state.timeout_hit: # Watchdog fired but the pump finished before the cancel landed. @@ -860,13 +872,22 @@ def _on_turn_timeout() -> None: # and _format_turn_result. Without this, such errors escape as a bare # exception: the orchestrator never drains pending_turn and _iteration # stays incremented, violating the pending-turn contract. - self._finalize_and_raise_crash(state.finalize, truncate_crash_message(f"Codex turn failed: {e!s}"), cause=e) + if state.stopped_early_hit and not state.timeout_hit: + # Same retry-poisoning guard as the inner handler: a cooperative + # stop already happened, so finalize cleanly instead of crashing. + self._log.warning("Ignoring post-stop exception; finalizing as STOPPED_EARLY: %s", e) + else: + self._finalize_and_raise_crash( + state.finalize, truncate_crash_message(f"Codex turn failed: {e!s}"), cause=e + ) self._state = AgentState.WORKING self._end_turn_ok() # The TurnRecord is the EventCollector's reduction of the emitted events. - state.finalize(AgentEndStatus.COMPLETED, crashed=False, crash_reason=None) + # Precedence matches Claude: timeout (raised above) > stopped_early > completed. + status = AgentEndStatus.STOPPED_EARLY if state.stopped_early_hit else AgentEndStatus.COMPLETED + state.finalize(status, crashed=False, crash_reason=None) return collector.build_turn_record() async def stop(self) -> None: @@ -1380,7 +1401,9 @@ def _format_turn_result(self, turn_result: Any) -> str: self._log.warning(f"Failed to format turn result: {e}") return str(turn_result) - async def _run_turn_with_streaming(self, state: _CodexTurnState) -> tuple[Any, Any, str]: + async def _run_turn_with_streaming( + self, state: _CodexTurnState, should_stop: Callable[[], bool] | None = None + ) -> tuple[Any, Any, str]: """Drive ``turn.stream()`` through the per-turn state, emitting the standard event protocol; returns ``(turn_result, latest_token_usage, agent_text)``. @@ -1388,6 +1411,11 @@ async def _run_turn_with_streaming(self, state: _CodexTurnState) -> tuple[Any, A boundaries; this drives the inner notification pump. ``state`` accumulates commands, the assistant transcript, sub-agent spawns and per-generation tokens — all mutated in place so a mid-turn crash keeps the partial. + + The cooperative ``should_stop`` poll runs AFTER ``state.dispatch`` (the + emission that lets the watcher latch on the deciding tool call) and BEFORE + the next notification is pulled — the deciding item is kept, the next is + not. No-op when ``should_stop is None`` (behaviorally identical to before). """ # Create the turn handle (starts the turn but doesn't block) + event stream. turn_handle = await self._run_async(self.thread.turn, state.user_input) @@ -1405,6 +1433,11 @@ async def _run_turn_with_streaming(self, state: _CodexTurnState) -> tuple[Any, A break if state.dispatch(notification): # True on a valid turn/completed break + if should_stop is not None and should_stop(): + state.stopped_early_hit = True + self._log.debug("Cooperative stop requested; ending notification pump at this boundary") + self._interrupt_active_turn() # best-effort; stops server-side spend + break finally: self._active_turn_handle = None # Close any orphan tool (item/started without item/completed), flush any @@ -1415,7 +1448,7 @@ async def _run_turn_with_streaming(self, state: _CodexTurnState) -> tuple[Any, A with contextlib.suppress(Exception): await self._run_async(stream.close) - if state.turn_result is None: + if state.turn_result is None and not state.stopped_early_hit: raise RuntimeError("Turn did not complete (no turn/completed notification received)") # Belt-and-suspenders: if streaming surfaced no assistant transcript, @@ -1427,7 +1460,9 @@ async def _run_turn_with_streaming(self, state: _CodexTurnState) -> tuple[Any, A # and nest them under the spawning Agent call. The parent stream never # carries the child's commands (Limited persistence drops them), but its # rollout always persists the raw function_call/local_shell_call items. - if state.spawned_children: + # Skipped on a cooperative stop: children may have no rollout yet and the + # run is already decided — recovery adds nothing the armed gate uses. + if state.spawned_children and not state.stopped_early_hit: await self._recover_subagent_tool_calls( state.spawned_children, state.collab_results, diff --git a/src/coder_eval/orchestration/early_stop.py b/src/coder_eval/orchestration/early_stop.py index fc9f0e5d..09d7acc7 100644 --- a/src/coder_eval/orchestration/early_stop.py +++ b/src/coder_eval/orchestration/early_stop.py @@ -159,7 +159,7 @@ def validate_early_stop(task: TaskDefinition) -> None: 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." + + f"(claude-code, codex, antigravity); agent type {agent_type!r} does not." ) # (2) Arming requires at least one stop criterion. diff --git a/tests/test_codex_agent_live.py b/tests/test_codex_agent_live.py index df19f534..bf684cb8 100644 --- a/tests/test_codex_agent_live.py +++ b/tests/test_codex_agent_live.py @@ -105,6 +105,45 @@ async def test_codex_live_edits_file_and_records_telemetry(tmp_path): assert record.commands, "expected command telemetry for the file creation" +@_live +async def test_codex_live_cooperative_stop_ends_turn_promptly(tmp_path): + """A ``should_stop`` that flips True after the first ToolStart ends the turn + cleanly as STOPPED_EARLY instead of running the multi-step task out — the + only automated check of real ``handle.interrupt()`` behavior.""" + from coder_eval.streaming.events import AgentEndEvent, AgentEndStatus, ToolStartEvent + + class _Sink: + def __init__(self): + self.tool_started = False + self.ends = [] + + def on_event(self, event): + if isinstance(event, ToolStartEvent): + self.tool_started = True + if isinstance(event, AgentEndEvent): + self.ends.append(event) + + sink = _Sink() + agent = _make_agent() + await agent.start(str(tmp_path)) + try: + record = await agent.communicate( + "Run `echo one`, then `echo two`, then `echo three`, each as a separate shell command, " + "then create three files a.txt, b.txt and c.txt.", + timeout=180, + stream_callback=sink, + should_stop=lambda: sink.tool_started, + ) + finally: + await agent.stop() + + # Clean cooperative stop: no crash, no pending partial, STOPPED_EARLY status. + assert record.crashed is False + assert agent.pending_turn is None + assert sink.ends, "expected an AgentEndEvent" + assert sink.ends[-1].status == AgentEndStatus.STOPPED_EARLY + + @_live async def test_codex_live_token_usage_populated(tmp_path): """Token usage is captured from the SDK and attached to the TurnRecord.""" diff --git a/tests/test_early_stop.py b/tests/test_early_stop.py index 09fe6024..36daa3e0 100644 --- a/tests/test_early_stop.py +++ b/tests/test_early_stop.py @@ -8,6 +8,10 @@ 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). +The same seam contract on ``CodexAgent`` and ``AntigravityAgent`` (stop cuts +the stream, clean STOPPED_EARLY finalize, timeout precedence, and the +post-stop-exception guard that prevents retry poisoning) is covered at the +end of this file. Phase 3 (feature live): the ``EarlyStopReason`` / ``EarlyStopInfo`` models and the ``armed_criteria_passed`` gate; the ``EarlyStopWatcher`` runtime observer @@ -20,24 +24,29 @@ import asyncio import tempfile -from collections.abc import Callable +from collections.abc import Callable, Iterator from datetime import datetime from pathlib import Path +from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import pytest import typer +from coder_eval.agents.antigravity_agent import AntigravityAgent from coder_eval.agents.claude_code_agent import ClaudeCodeAgent +from coder_eval.agents.codex_agent import CodexAgent, _CodexTurnState +from coder_eval.agents.registry import AgentRegistry 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 from coder_eval.criteria.skill_triggered import SkillTriggeredChecker, _engaged_skill_names -from coder_eval.errors import TurnTimeoutError +from coder_eval.errors import AgentCrashError, TurnTimeoutError from coder_eval.models import ( AgentKind, + BaseAgentConfig, CommandExecutedCriterion, CommandTelemetry, CriterionResult, @@ -101,7 +110,7 @@ def _task( *, criteria: list[Any], stop_early: bool = False, - agent_type: AgentKind = AgentKind.CLAUDE_CODE, + agent_type: AgentKind | str = AgentKind.CLAUDE_CODE, simulation: SimulationConfig | None = None, ) -> TaskDefinition: """Build a minimal resolved-style TaskDefinition for guardrail tests.""" @@ -117,6 +126,32 @@ def _task( ) +class _DummyNoStopConfig(BaseAgentConfig): + """Config for the dummy non-supporting agent registered by the fixture below.""" + + +class _DummyNoStopAgent: + """Agent stand-in that leaves ``supports_cooperative_stop`` at the default False. + + ``validate_early_stop`` only reads the flag off the registered class, so no + ``Agent`` machinery is needed. Guardrail 1 must keep rejecting agents that + have not opted into the cooperative interrupt (all built-ins now support it). + """ + + supports_cooperative_stop = False + + +@pytest.fixture +def dummy_no_stop_kind() -> Iterator[str]: + """Register a non-supporting agent kind for guardrail-1 tests, then clean up.""" + kind = "dummy-no-stop" + AgentRegistry.register(kind, _DummyNoStopConfig)(_DummyNoStopAgent) + try: + yield kind + finally: + AgentRegistry._registry.pop(kind, None) + + def _skill_crit(skill_name: str, expected_skill: str, *, stop_when: str | None = None) -> SkillTriggeredCriterion: return SkillTriggeredCriterion( type="skill_triggered", @@ -532,11 +567,23 @@ def test_guardrail5_simulation_rejected(self) -> None: 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) + def test_guardrail1_non_supporting_agent_rejected(self, dummy_no_stop_kind: str) -> None: + # Codex/antigravity now support the cooperative interrupt, so guardrail 1 + # is exercised with a dummy agent that leaves the flag at False. + task = _task(criteria=[_skill_crit("s", "s", stop_when="pass")], stop_early=True, agent_type=dummy_no_stop_kind) with pytest.raises(EarlyStopConfigError, match="cooperative stopping"): validate_early_stop(task) + def test_guardrail1_armed_codex_accepts(self) -> None: + task = _task(criteria=[_skill_crit("s", "s", stop_when="pass")], stop_early=True, agent_type=AgentKind.CODEX) + validate_early_stop(task) # no raise + + def test_guardrail1_armed_antigravity_accepts(self) -> None: + task = _task( + criteria=[_skill_crit("s", "s", stop_when="pass")], stop_early=True, agent_type=AgentKind.ANTIGRAVITY + ) + validate_early_stop(task) # no raise + 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"): @@ -556,13 +603,13 @@ def test_guardrail4_unsupported_polarity_rejected(self, monkeypatch: pytest.Monk 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. + def test_raise_order_simulation_before_agent(self, dummy_no_stop_kind: str) -> None: + # Both simulation AND a non-supporting 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, + agent_type=dummy_no_stop_kind, simulation=sim, ) with pytest.raises(EarlyStopConfigError, match="simulation"): @@ -1353,13 +1400,14 @@ async def _run_wiring( scores: list[float], stop_early: bool, tmp_path, + agent_type: AgentKind = AgentKind.CLAUDE_CODE, ) -> 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) + task = _task(criteria=criteria, stop_early=stop_early, agent_type=agent_type) run_dir = tmp_path / "run" run_dir.mkdir(parents=True) orch = Orchestrator(task=task, run_dir=run_dir, variant_id="default") @@ -1367,7 +1415,7 @@ async def _run_wiring( task_id=task.task_id, task_description=task.description, variant_id="default", - agent_type=AgentKind.CLAUDE_CODE, + agent_type=agent_type, started_at=datetime.now(), final_status=FinalStatus.FAILURE, iteration_count=0, @@ -1621,3 +1669,475 @@ def test_telemetry_dims_reflect_early_stop(self) -> None: _n2, props2 = build_task_event(_result(), driver="tempdir", variant_id="v") assert props2["EarlyStopped"] is False assert props2["EarlyStopReason"] == "" + + +# --------------------------------------------------------------------------- # +# Cooperative should_stop seam on CodexAgent — mirrors TestCooperativeStopSeam. +# SDK-independent: the pump is driven over fake notifications (agentMessage +# deltas need no openai_codex types) and turn/completed handling is stubbed. +# --------------------------------------------------------------------------- # + + +class _CodexNotifIter: + """Counting iterator over fake notifications (the pump pulls via ``next``).""" + + def __init__(self, notifications: list[Any]) -> None: + self._it = iter(notifications) + self.pulled = 0 + + def __iter__(self) -> _CodexNotifIter: + return self + + def __next__(self) -> Any: + item = next(self._it) + self.pulled += 1 + return item + + +class _FakeCodexStream: + def __init__(self, notifications: list[Any]) -> None: + self.iter = _CodexNotifIter(notifications) + self.closed = False + + def __iter__(self) -> _CodexNotifIter: + return self.iter + + def close(self) -> None: + self.closed = True + + +class _FakeCodexTurnHandle: + def __init__(self, stream: _FakeCodexStream) -> None: + self._stream = stream + self.interrupts = 0 + + def stream(self) -> _FakeCodexStream: + return self._stream + + def interrupt(self) -> None: + self.interrupts += 1 + + +def _codex_delta(i: int) -> SimpleNamespace: + """A fake ``item/agentMessage/delta`` notification (no SDK types involved).""" + return SimpleNamespace(method="item/agentMessage/delta", payload=SimpleNamespace(delta=f"chunk{i} ", item=None)) + + +def _codex_completed() -> SimpleNamespace: + """A fake ``turn/completed`` notification, consumed by the stubbed handler.""" + return SimpleNamespace(method="turn/completed", payload=SimpleNamespace(items=None)) + + +def _stub_on_turn_completed(self: Any, notification: Any) -> bool: + """Stand-in for ``_CodexTurnState.on_turn_completed`` (the real one isinstance- + checks an openai_codex type). Sets the terminal turn and breaks the pump.""" + self.turn_result = notification.payload + return True + + +def _codex_agent() -> CodexAgent: + agent = CodexAgent(parse_agent_config(type=AgentKind.CODEX, model="gpt-5-codex")) # type: ignore[arg-type] + agent.working_directory = Path("/tmp") + agent.codex_client = SimpleNamespace() # truthy: communicate()'s started check + return agent + + +async def _run_codex_communicate( + *, + notifications: list[Any], + stop_after: int | None = None, + never: bool = False, + timeout: float | None = None, +) -> tuple[CodexAgent, TurnRecord, _EventSink, _FakeCodexStream, _FakeCodexTurnHandle]: + """Drive ``CodexAgent.communicate`` over a fake notification stream. + + ``stop_after``: should_stop returns True once that many notifications have + been pulled (checked after each dispatch). ``never``: an always-False + should_stop. Neither: ``should_stop=None``. + """ + agent = _codex_agent() + stream = _FakeCodexStream(notifications) + handle = _FakeCodexTurnHandle(stream) + agent.thread = SimpleNamespace(turn=lambda _prompt: handle) + + should_stop: Callable[[], bool] | None + if stop_after is not None: + should_stop = lambda: stream.iter.pulled >= stop_after # noqa: E731 + elif never: + should_stop = lambda: False # noqa: E731 + else: + should_stop = None + + sink = _EventSink() + with patch.object(_CodexTurnState, "on_turn_completed", _stub_on_turn_completed): + record = await agent.communicate("prompt", stream_callback=sink, timeout=timeout, should_stop=should_stop) + return agent, record, sink, stream, handle + + +class TestCodexCooperativeStopSeam: + async def test_stop_after_first_dispatched_notification(self) -> None: + notifications = [_codex_delta(0), _codex_delta(1), _codex_delta(2), _codex_completed()] + agent, record, sink, stream, handle = await _run_codex_communicate(notifications=notifications, stop_after=1) + # The deciding notification is kept; the next is never pulled. + assert stream.iter.pulled == 1 + # The in-flight turn was interrupted exactly once (server-side spend cut). + assert handle.interrupts == 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 + # A clean stop: no partial pending_turn, no ERROR state, no raise. + assert agent.pending_turn is None + assert agent.get_state().value != "error" + + async def test_should_stop_none_consumes_full_stream(self) -> None: + notifications = [_codex_delta(0), _codex_delta(1), _codex_completed()] + _agent, record, sink, stream, handle = await _run_codex_communicate(notifications=notifications) + assert stream.iter.pulled == 3 + assert handle.interrupts == 0 + assert record.crashed is False + assert _agent_end_events(sink)[0].status == AgentEndStatus.COMPLETED + + async def test_should_stop_false_consumes_full_stream(self) -> None: + notifications = [_codex_delta(0), _codex_delta(1), _codex_completed()] + _agent, _record, sink, stream, _handle = await _run_codex_communicate(notifications=notifications, never=True) + assert stream.iter.pulled == 3 + assert _agent_end_events(sink)[0].status == AgentEndStatus.COMPLETED + + async def test_stop_before_turn_completed_does_not_raise(self) -> None: + # The stream is cut before any turn/completed: turn_result is None, but the + # stop makes the "turn never completed" raise conditional — no crash. + notifications = [_codex_delta(0), _codex_delta(1), _codex_delta(2)] + _agent, record, sink, _stream, _handle = await _run_codex_communicate(notifications=notifications, stop_after=1) + assert record.crashed is False + assert _agent_end_events(sink)[0].status == AgentEndStatus.STOPPED_EARLY + + async def test_stream_dying_without_stop_still_raises(self) -> None: + # Regression guard: a stream that ends with NO turn/completed and NO stop + # is still a crash (the RuntimeError survives for genuine stream deaths). + agent = _codex_agent() + stream = _FakeCodexStream([_codex_delta(0)]) + agent.thread = SimpleNamespace(turn=lambda _prompt: _FakeCodexTurnHandle(stream)) + with pytest.raises(AgentCrashError, match="did not complete"): + await agent.communicate("prompt", stream_callback=_EventSink(), should_stop=None) + assert agent.pending_turn is not None + assert agent.pending_turn.crashed is True + await agent.discard_pending_turn() + + async def test_timeout_beats_stop_precedence(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Both signals in one turn: the watchdog fires (timeout_hit) AND should_stop + # is True. The post-pump timeout check must win — TIMEOUT, crashed=True. + class _FiringWatchdog: + def __init__(self, *, on_timeout: Callable[[], None], **_kwargs: Any) -> None: + self._on_timeout = on_timeout + + def __enter__(self) -> _FiringWatchdog: + self._on_timeout() # watchdog fired: state.timeout_hit = True + return self + + def __exit__(self, *_exc: Any) -> bool: + return False + + monkeypatch.setattr("coder_eval.agents.codex_agent.ThreadedWatchdog", _FiringWatchdog) + agent = _codex_agent() + stream = _FakeCodexStream([_codex_delta(0), _codex_delta(1)]) + agent.thread = SimpleNamespace(turn=lambda _prompt: _FakeCodexTurnHandle(stream)) + sink = _EventSink() + with pytest.raises(TurnTimeoutError): + await agent.communicate("prompt", stream_callback=sink, timeout=30.0, should_stop=lambda: True) + ends = _agent_end_events(sink) + assert len(ends) == 1 + assert ends[0].status == AgentEndStatus.TIMEOUT + assert ends[0].crashed is True + 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} + await agent.discard_pending_turn() + + async def test_post_stop_exception_stays_clean(self, monkeypatch: pytest.MonkeyPatch) -> None: + # The retry-poisoning gap: an exception AFTER the cooperative break (here: + # the pump's finally-side cleanup) must NOT crash-finalize the turn — a + # crash would trigger the orchestrator retry with the watcher's decision + # still latched, stopping the retry at turn 0. + def _boom(self: Any) -> None: + raise RuntimeError("post-stop cleanup boom") + + monkeypatch.setattr(_CodexTurnState, "close_open_tools", _boom) + notifications = [_codex_delta(0), _codex_delta(1)] + agent, record, sink, _stream, _handle = await _run_codex_communicate(notifications=notifications, stop_after=1) + # No AgentCrashError raised (we got a record back), clean STOPPED_EARLY. + 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 + assert agent.pending_turn is None + + async def test_post_stop_cleanup_exception_without_stop_still_crashes( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # The guard is scoped to stopped turns only: the same cleanup exception on + # a NON-stopped turn keeps crashing (no behavior change for real failures). + def _boom(self: Any) -> None: + raise RuntimeError("cleanup boom") + + monkeypatch.setattr(_CodexTurnState, "close_open_tools", _boom) + agent = _codex_agent() + stream = _FakeCodexStream([_codex_delta(0)]) + agent.thread = SimpleNamespace(turn=lambda _prompt: _FakeCodexTurnHandle(stream)) + with pytest.raises(AgentCrashError): + await agent.communicate("prompt", stream_callback=_EventSink(), should_stop=None) + await agent.discard_pending_turn() + + async def test_stopped_turn_skips_subagent_recovery(self) -> None: + # A stopped turn must not attempt rollout recovery: children may have no + # rollout yet and the run is already decided. + agent = _codex_agent() + stream = _FakeCodexStream([_codex_delta(0), _codex_delta(1)]) + agent.thread = SimpleNamespace(turn=lambda _prompt: _FakeCodexTurnHandle(stream)) + recover = AsyncMock() + captured: dict[str, Any] = {} + + original_init = _CodexTurnState.__init__ + + def _capturing_init(self: Any, *args: Any, **kwargs: Any) -> None: + original_init(self, *args, **kwargs) + self.spawned_children = [("child-thread", "tool-1", None)] + captured["state"] = self + + with ( + patch.object(_CodexTurnState, "__init__", _capturing_init), + patch.object(CodexAgent, "_recover_subagent_tool_calls", recover), + ): + await agent.communicate("prompt", stream_callback=_EventSink(), should_stop=lambda: stream.iter.pulled >= 1) + assert captured["state"].stopped_early_hit is True + recover.assert_not_awaited() + + +# --------------------------------------------------------------------------- # +# Cooperative should_stop seam on AntigravityAgent — same contract, driven over +# a fake step stream (mirrors tests/test_antigravity_agent.py's conventions). +# --------------------------------------------------------------------------- # + + +def _ag_step(i: int) -> SimpleNamespace: + """A minimal streamed text step (plain strings stand in for the SDK enums).""" + return SimpleNamespace( + type="TEXT_RESPONSE", + status="ACTIVE", + source="MODEL", + target="TARGET_USER", + tool_calls=[], + content="", + content_delta=f"c{i}", + thinking="", + thinking_delta="", + usage_metadata=None, + is_complete_response=None, + error="", + step_index=i, + ) + + +class _CountingConversation: + def __init__(self, steps: list[Any], *, cancel_raises: bool = False) -> None: + self._steps = steps + self._cancel_raises = cancel_raises + self.yielded = 0 + self.cancels = 0 + self.last_response = "" + + async def send(self, prompt: Any, **_kwargs: Any) -> None: + return None + + async def receive_steps(self) -> Any: + for step in self._steps: + self.yielded += 1 + yield step + + async def cancel(self) -> None: + self.cancels += 1 + if self._cancel_raises: + raise RuntimeError("cancel boom") + + +def _antigravity_agent(conversation: _CountingConversation) -> AntigravityAgent: + agent = AntigravityAgent(parse_agent_config(type=AgentKind.ANTIGRAVITY, model="gemini-3-flash")) # type: ignore[arg-type] + agent.working_directory = Path("/tmp") + agent._sdk_agent = SimpleNamespace(conversation=conversation, is_started=True) + return agent + + +async def _run_antigravity_communicate( + *, + n_steps: int = 3, + stop_after: int | None = None, + never: bool = False, + cancel_raises: bool = False, +) -> tuple[AntigravityAgent, TurnRecord, _EventSink, _CountingConversation]: + """Drive ``AntigravityAgent.communicate`` over a fake step stream (same + stop_after / never / None semantics as the Claude and Codex drivers).""" + conversation = _CountingConversation([_ag_step(i) for i in range(n_steps)], cancel_raises=cancel_raises) + agent = _antigravity_agent(conversation) + + should_stop: Callable[[], bool] | None + if stop_after is not None: + should_stop = lambda: conversation.yielded >= stop_after # noqa: E731 + elif never: + should_stop = lambda: False # noqa: E731 + else: + should_stop = None + + sink = _EventSink() + record = await agent.communicate("prompt", stream_callback=sink, should_stop=should_stop) + return agent, record, sink, conversation + + +class TestAntigravityCooperativeStopSeam: + async def test_stop_after_first_processed_step(self) -> None: + agent, record, sink, conversation = await _run_antigravity_communicate(stop_after=1, n_steps=3) + # The deciding step is kept; the next is never pulled. + assert conversation.yielded == 1 + # The conversation was cancelled once (best-effort server-side cut). + assert conversation.cancels == 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 + assert agent.pending_turn is None + assert agent.get_state().value != "error" + + async def test_should_stop_none_consumes_full_stream(self) -> None: + _agent, record, sink, conversation = await _run_antigravity_communicate(n_steps=3) + assert conversation.yielded == 3 + assert conversation.cancels == 0 + assert record.crashed is False + assert _agent_end_events(sink)[0].status == AgentEndStatus.COMPLETED + + async def test_should_stop_false_consumes_full_stream(self) -> None: + _agent, _record, sink, conversation = await _run_antigravity_communicate(never=True, n_steps=3) + assert conversation.yielded == 3 + assert _agent_end_events(sink)[0].status == AgentEndStatus.COMPLETED + + async def test_raising_cancel_still_stops_clean(self) -> None: + # conversation.cancel() is best-effort: a raising cancel must not escalate + # a stopped turn to a crash. + agent, record, sink, conversation = await _run_antigravity_communicate(stop_after=1, cancel_raises=True) + assert conversation.cancels == 1 + assert record.crashed is False + assert _agent_end_events(sink)[0].status == AgentEndStatus.STOPPED_EARLY + assert agent.pending_turn is None + + async def test_timeout_beats_stop_precedence(self, monkeypatch: pytest.MonkeyPatch) -> None: + class _FiringWatchdog: + def __init__(self, *, on_timeout: Callable[[], None], **_kwargs: Any) -> None: + self._on_timeout = on_timeout + + def __enter__(self) -> _FiringWatchdog: + self._on_timeout() + return self + + def __exit__(self, *_exc: Any) -> bool: + return False + + monkeypatch.setattr("coder_eval.agents.antigravity_agent.ThreadedWatchdog", _FiringWatchdog) + conversation = _CountingConversation([_ag_step(0), _ag_step(1)]) + agent = _antigravity_agent(conversation) + sink = _EventSink() + with pytest.raises(TurnTimeoutError): + await agent.communicate("prompt", stream_callback=sink, timeout=30.0, should_stop=lambda: True) + ends = _agent_end_events(sink) + assert len(ends) == 1 + assert ends[0].status == AgentEndStatus.TIMEOUT + assert ends[0].crashed is True + assert agent.pending_turn is not None and agent.pending_turn.crashed is True + assert AgentEndStatus.STOPPED_EARLY not in {e.status for e in ends} + await agent.discard_pending_turn() + + async def test_post_stop_exception_stays_clean(self, monkeypatch: pytest.MonkeyPatch) -> None: + # The retry-poisoning gap, antigravity flavor: an exception raised by + # post-stop cleanup (stand-in for the step generator's aclose() raising + # after the break) lands in the generic handler, which must fall through + # to the clean STOPPED_EARLY finalize instead of crash-finalizing. + class _ExplodingExitWatchdog: + def __init__(self, *_args: Any, **_kwargs: Any) -> None: + pass + + def __enter__(self) -> _ExplodingExitWatchdog: + return self + + def __exit__(self, exc_type: Any, *_exc: Any) -> bool: + if exc_type is None: + raise RuntimeError("post-stop cleanup boom") + return False + + monkeypatch.setattr("coder_eval.agents.antigravity_agent.ThreadedWatchdog", _ExplodingExitWatchdog) + agent, record, sink, _conversation = await _run_antigravity_communicate(stop_after=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 + assert agent.pending_turn is None + + async def test_post_stop_cleanup_exception_without_stop_still_crashes( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Guard scoping: the same cleanup exception on a NON-stopped turn keeps + # crashing (no behavior change for real failures). + class _ExplodingExitWatchdog: + def __init__(self, *_args: Any, **_kwargs: Any) -> None: + pass + + def __enter__(self) -> _ExplodingExitWatchdog: + return self + + def __exit__(self, exc_type: Any, *_exc: Any) -> bool: + if exc_type is None: + raise RuntimeError("cleanup boom") + return False + + monkeypatch.setattr("coder_eval.agents.antigravity_agent.ThreadedWatchdog", _ExplodingExitWatchdog) + conversation = _CountingConversation([_ag_step(0)]) + agent = _antigravity_agent(conversation) + with pytest.raises(AgentCrashError): + await agent.communicate("prompt", stream_callback=_EventSink(), should_stop=None) + await agent.discard_pending_turn() + + +# --------------------------------------------------------------------------- # +# Orchestrator-level wiring on a non-Claude agent type: the watcher, gating and +# report row are agent-agnostic — an armed codex task flows end to end. +# --------------------------------------------------------------------------- # + + +class TestOrchestratorEarlyStopWiringCodex: + _SKILL = "date-teller" + + def _criteria(self) -> list[Any]: + return [ + _skill_crit(self._SKILL, self._SKILL, stop_when="pass"), + FileExistsCriterion(path="artifact.txt", description="artifact must exist"), + ] + + async def test_pass_stop_populates_early_stop_and_armed_gate(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, + agent_type=AgentKind.CODEX, + ) + assert agent.delivered == 3 + assert result.early_stop is not None + assert result.early_stop.reason == EarlyStopReason.CRITERION_PASSED + # Armed-subset gate: advisory 0.0 is not gated on an early-stopped run. + assert result.armed_criteria_passed(self._criteria()) is True + # The report row carries the early-stop marker. + row = eval_result_to_task_dict(result) + assert row["stopped_early"] is True