From 058732f0392aafc38fa82ce4eec5bf3ddb8b6a1f Mon Sep 17 00:00:00 2001 From: joeysbase Date: Wed, 12 Aug 2026 22:07:36 +0000 Subject: [PATCH 1/3] fix(antigravity): poll for backgrounded work instead of grading it incomplete AntigravityAgent.communicate() used to drain the SDK's step stream once and finalize the instant it went idle, so a task Gemini backgrounds and pauses on (intending to check back later) got graded before the work finished. Now the turn polls (sleep + re-drain) while an orphaned tool call is still ACTIVE, bounded by _MAX_BACKGROUND_POLLS and the existing turn watchdog. Hardened across three review rounds: the orphan signal allowlists ACTIVE (not "not yet closed") so a tool stuck on WAITING_FOR_USER/CANCELED/UNKNOWN is never polled forever; the fallback synthetic tool-call id (when the SDK's call.id is falsy) is stable across a step's own ACTIVE->DONE re-emissions and unique across trajectories; the poll loop exits promptly once the watchdog has decided to fire; and _drain() retries past the SDK's real two-layer generator re-entrancy window after a cooperative stop instead of crashing the next turn. Co-Authored-By: Claude Sonnet 5 --- .claude/harness-candidates.md | 46 ++ src/coder_eval/agents/antigravity_agent.py | 197 +++++- tests/test_antigravity_agent.py | 668 ++++++++++++++++++++- 3 files changed, 888 insertions(+), 23 deletions(-) diff --git a/.claude/harness-candidates.md b/.claude/harness-candidates.md index 01d812cc..edd6e003 100644 --- a/.claude/harness-candidates.md +++ b/.claude/harness-candidates.md @@ -314,3 +314,49 @@ with the two `action.yml` items above — one considered change to the action's advertises otherwise. Either mark it `required: true` (a published-input contract change, see the `working-directory` item) or fail with a clear message instead of an obscure discovery error. + +## From the coder-eval-code-review of fix/antigravity-wait-for-wakeup (2026-08-12) + +- [ ] **A retry/poll loop's continuation state must derive from a stable per-entity + key, never a mutable monotonic counter used as an id fallback.** `_AntigravityTurnState._handle_tool_call` + minted a synthetic tool-call id from `f"{raw_name}_{self._next_seq}"` when the SDK's + `call.id` was falsy; since `_next_seq` advances between a tool call's ACTIVE and DONE + emissions, the DONE step computed a *different* fallback id than the ACTIVE step, + stranding the ACTIVE entry as a permanent orphan and stalling `communicate()`'s new + poll loop for its full `_MAX_BACKGROUND_POLLS` budget on every id-less turn. Fixed by + deriving the fallback from `(step.step_index, call_index)` instead (stable across a + step's own re-emissions, per this class's own docstring). Not promoted to a CExxx rule: + this is the only id-fallback-driving-control-flow site in the codebase today (a + single call site, not a recurring class per the existing "single call-site fix, no + recurring pattern to guard" convention) — a mechanical AST rule for "no mutable + counter in a dict-key fallback" would need real design work to avoid false-positiving + on ordinary sequence-numbering counters elsewhere in the file. Caught by two + independent reviewers (Opus fallback pair) in this run's final code review. + +- [ ] **A `while` loop built around a cooperative-cancellation watchdog should read the + watchdog's own "already decided to fire" flag in its condition, not rely solely on a + later exception handler to notice.** The antigravity poll loop's condition checked + `not state.stopped_early_hit and state.has_orphaned_tool_call() and poll_count < cap` + but not `state.timeout_hit`, so if `ThreadedWatchdog`'s background thread set the flag + before its `task.cancel()` actually landed on this coroutine, the loop kept + sleeping/re-draining for up to the full poll budget before the pre-existing + post-loop `if state.timeout_hit:` check ever got a chance to run. Fixed by adding + `and not state.timeout_hit` to the condition. Not promoted: `ThreadedWatchdog` + + a bespoke poll loop reading its own state flag is a one-off shape unique to this + agent; no second instance exists to generalize a rule from. Caught in the same + final review as above. + +- [ ] **A regression test's fake dependency must model every layer the fix under test + actually touches, not just the outermost one.** `_drain()`'s cooperative-stop path + wraps a real SDK call (`Conversation.receive_steps()`) that is itself a delegating + async generator over an inner, connection-layer generator holding the real + re-entrancy guard. The first regression test written for this fix used a + single-layer fake (the guard lived on the SAME generator `_drain()` iterated), which + passed against an incomplete fix (`contextlib.aclosing` on the outer generator only) + that does not work against the real two-layer SDK shape — confirmed live that the + inner generator's cleanup is deferred to a LATER event-loop turn, not synchronous + with the outer's `aclose()`. Caught by a reviewer re-deriving the real dependency's + shape from its installed source, not by the test itself. Not promoted: detecting "a + test double is missing a delegation layer the source has" is a semantic match + against third-party source, not an AST pattern in our own code — no cheap mechanical + check exists. Caught in the round-3 coder-eval-code-review of this same branch. diff --git a/src/coder_eval/agents/antigravity_agent.py b/src/coder_eval/agents/antigravity_agent.py index 4e1475be..38715425 100644 --- a/src/coder_eval/agents/antigravity_agent.py +++ b/src/coder_eval/agents/antigravity_agent.py @@ -98,6 +98,47 @@ def _harness_spawn_lock() -> asyncio.Lock: # agentic benchmarks while running faster; ``medium`` thinking is its daily-driver default. _DEFAULT_MODEL = "gemini-3.5-flash" +# How often to re-check for progress once an orphaned (backgrounded) tool call is +# detected. receive_steps() returns instantly empty ONLY when the connection is +# already idle with nothing queued -- which is exactly the state right after a +# background job leaves the model idle, so the common re-check is cheap. (It CAN +# still await indefinitely if called while genuinely non-idle work is in flight; +# see the poll loop's own comment in communicate() for that case.) +# Conversation.wait_for_wakeup() is an unimplemented stub on the Local harness +# connection this agent uses (always returns False, regardless of pending state, +# confirmed against the installed SDK's source) — so this file drives its own +# sleep-and-retry poll instead. Not user-configurable: a tuning constant, not a +# feature. +_BACKGROUND_POLL_INTERVAL_SECONDS = 5.0 + +# Bound on retrying a receive_steps() call that hits the SDK's re-entrancy guard +# (see _drain()'s docstring) -- each retry yields one event-loop turn via +# asyncio.sleep(0) for the prior drain's already-scheduled generator cleanup to +# land. Confirmed live against the real SDK's generator-delegation shape that +# this clears within 2 turns; this constant carries a 2.5x margin, not a +# separately-tuned budget. +_RECEIVE_STEPS_REENTRY_RETRIES = 5 + +# Cap on poll *cycles* per turn -- bounds how many times communicate() will +# sleep-and-redrain. Wall-clock enforcement of the whole turn is a separate +# concern, left to the pre-existing ThreadedWatchdog (see communicate()); this +# cap only matters when a task sets no run_limits.turn_timeout/task_timeout at +# all, or when a single re-drain returns promptly each cycle. +# 120 * 5s = 10 minutes: real backgrounded jobs observed in confirmed-broken +# tasks ran 60-300s, so this carries a ~2x safety margin over the worst one +# seen while still bounding a truly pathological (never-resolving) orphan to a +# reasonable cap. +# +# Deliberately NOT "break after N consecutive empty polls" instead: the real +# SDK's receive_steps() returns identically empty whether a backgrounded job is +# still genuinely running OR will never resolve at all (confirmed live against +# the installed SDK) -- there is no signal that tells these two cases apart +# except waiting. A consecutive-empty-count small enough to matter would also +# abort real slow jobs (the confirmed cases needed up to ~60 consecutive 5s- +# empty polls before succeeding); one large enough to be safe barely improves +# over this flat cap. A flat, data-grounded cap is the honest option. +_MAX_BACKGROUND_POLLS = 120 + # Antigravity builtin tool name -> canonical Claude-ish tool name, so cross-agent # success criteria (command_executed / commands_efficiency / skill_triggered) and # reports key on the SAME tool names the Claude / Codex backends emit. Unmapped @@ -143,6 +184,7 @@ def _harness_spawn_lock() -> asyncio.Lock: # optional extra; only ``start()`` touches it). Named constants — not bare string # literals — so an antigravity StepStatus.ERROR comparison is not mistaken for a # coder_eval FinalStatus member-name denylist (lint rule CE018). +_STATUS_ACTIVE = "ACTIVE" _STATUS_DONE = "DONE" _STATUS_ERROR = "ERROR" _TYPE_THINKING = "THINKING" @@ -406,6 +448,59 @@ async def _harness_spawn_guard(self) -> AsyncIterator[None]: else: os.environ[path_key] = original + async def _drain( + self, + conversation: Any, + state: "_AntigravityTurnState", + should_stop: Callable[[], bool] | None, + ) -> None: + """Consume one ``receive_steps()`` cycle onto ``state``, honoring a + cooperative stop mid-stream. Shared by the initial drain and each poll + cycle's re-drain in ``communicate`` so this shape lives in one place. + + ``receive_steps()`` is actually TWO nested async generators: the public + ``Conversation.receive_steps()`` we call here delegates internally + (``async for step in self._connection.receive_steps(): yield step``) to + the connection layer, which guards re-entrancy with an ``_is_receiving`` + flag cleared only in its OWN ``finally``. ``aclosing`` on the outer + generator closes IT deterministically, but a ``GeneratorExit`` thrown + into a delegating generator does not synchronously propagate into the + inner one it was mid-iterating -- confirmed live: the inner ``finally`` + only ran after the outer's frame was unwound AND the event loop had + processed the abandoned inner generator's async-gen finalizer, i.e. on a + LATER event-loop turn, not within the ``aclosing`` block itself. So a + cooperative-stop ``break`` here can still leave the connection + "receiving" for a short, bounded window afterward, and the NEXT + ``receive_steps()`` call (the next poll cycle, or the next turn in a + multi-turn dialog) can raise ``RuntimeError`` during that window. The + retry below -- yielding via ``asyncio.sleep(0)`` and trying again -- + gives that already-scheduled finalizer a turn to run, mirroring the + SDK's OWN handling of this exact ``RuntimeError`` in + ``Conversation.send()`` (falls back to ``wait_for_idle()``); retrying + the drain itself is preferred here over that fallback since + ``wait_for_idle()`` discards any steps already queued, which would + silently drop real content instead of just retrying past a transient + window. + """ + for attempt in range(_RECEIVE_STEPS_REENTRY_RETRIES): + try: + async with contextlib.aclosing(conversation.receive_steps()) as steps: + async for step in 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 + return + except RuntimeError: + if attempt == _RECEIVE_STEPS_REENTRY_RETRIES - 1: + raise + self._log.debug( + "receive_steps() re-entrancy guard still set from a prior drain; retrying (attempt %d)", + attempt + 1, + ) + await asyncio.sleep(0) + async def communicate( self, user_input: str, @@ -470,21 +565,67 @@ def _on_turn_timeout() -> None: ): emit.on_event(TurnStartEvent(task_id=task_id, turn_id=turn_id, model=model)) conversation = self._sdk_agent.conversation + poll_count = 0 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) + await self._drain(conversation, state, should_stop) + + # The model may have kicked off a run_command as a background + # task and gone idle without waiting for it — receive_steps() + # then exhausts with that tool call still open (never reached + # DONE/ERROR). Conversation.wait_for_wakeup() is an unimplemented + # stub on this SDK's Local harness (always returns False, + # regardless of pending state — confirmed against the installed + # source and live-tested), so poll for progress ourselves + # instead, gated on that orphaned-tool signal so a normal turn + # (which always closes its tool calls before the stream + # exhausts) takes this branch zero times and finalizes exactly + # as fast as today. No separate in-loop deadline check against + # `timeout`: the ThreadedWatchdog above already enforces it by + # cancelling this whole coroutine, and a second check against + # the SAME value would just race it non-deterministically for + # who fires first. `_MAX_BACKGROUND_POLLS` bounds the number of + # poll *cycles*; the `not state.timeout_hit` check below is a + # fast exit once the watchdog has already decided to fire but + # its cancellation hasn't landed on this coroutine yet (a single + # re-drain can itself await indefinitely while genuinely + # non-idle work is in flight, so true wall-clock safety for an + # unbounded background job still depends on a configured + # `run_limits.turn_timeout`/`task_timeout` reaching the watchdog + # — not on this cap alone). + while ( + not state.stopped_early_hit + and not state.timeout_hit + and state.has_orphaned_tool_call() + and poll_count < _MAX_BACKGROUND_POLLS + ): + poll_count += 1 + self._log.debug("Polling for backgrounded work (orphaned tool call); attempt %d", poll_count) + await asyncio.sleep(_BACKGROUND_POLL_INTERVAL_SECONDS) + if state.timeout_hit: + # The watchdog decided to fire during the sleep above; skip + # the re-drain (which could itself await indefinitely on + # genuinely non-idle work) rather than waiting for the + # loop's own head check to catch it next cycle. + break 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 + await self._drain(conversation, state, should_stop) + + if poll_count >= _MAX_BACKGROUND_POLLS and state.has_orphaned_tool_call(): + msg = "Poll budget (_MAX_BACKGROUND_POLLS=%d) exhausted with a tool call still ACTIVE." + self._log.warning(msg, _MAX_BACKGROUND_POLLS) + if state.stopped_early_hit: # Best-effort server-side cancel, mirrors kill(); a raising - # cancel() lands in the guarded handler below. + # cancel() lands in the guarded handler below. Single check + # point covers a stop from either the initial drain or any + # poll cycle, so cancel() fires exactly once either way. with contextlib.suppress(Exception): await conversation.cancel() except asyncio.CancelledError: @@ -638,6 +779,10 @@ def __init__( # inputs). Used at DONE to distinguish inputs from harness-appended # result fields, whatever they're named for that tool. self._tool_input_keys: dict[str, set[str]] = {} + # Most recently seen StepStatus per tool id, for has_orphaned_tool_call + # below — deliberately separate from _closed_tools, which only tracks + # the DONE/ERROR terminal states relevant to result reporting. + self._tool_last_status: dict[str, Any] = {} # Content blocks accumulated since the last per-generation flush. self._blocks: list[ContentBlock] = [] @@ -654,8 +799,8 @@ def process_step(self, step: Any) -> None: self.emit.on_event(TextChunkEvent(task_id=self.task_id, turn_id=self.turn_id, text=step.content_delta)) # Tool calls: ToolStart on first sight, ToolEnd when the owning step is DONE. - for call in step.tool_calls: - self._handle_tool_call(call, step, done, sstatus) + for call_index, call in enumerate(step.tool_calls): + self._handle_tool_call(call, step, done, sstatus, call_index) # Capture content blocks on the terminal transition of a step. if done: @@ -671,9 +816,26 @@ def process_step(self, step: Any) -> None: self.total_usage = self.total_usage + gen self._flush_generation(gen, getattr(step.usage_metadata, "thoughts_token_count", 0) or 0) - def _handle_tool_call(self, call: Any, step: Any, done: bool, sstatus: Any) -> None: + def _handle_tool_call(self, call: Any, step: Any, done: bool, sstatus: Any, call_index: int) -> None: raw_name = _enum_value(call.name) - cid = call.id or f"{raw_name}_{self._next_seq}" + # call.id is usually present ("a tool call carries a stable id" per this + # class's docstring), but the SDK types it as optional. The fallback must + # be BOTH stable across a step's own ACTIVE -> DONE re-emissions (same + # step_index) -- so an id-less call's DONE step closes the SAME cid its + # ACTIVE step opened, rather than minting a fresh id from a counter that + # already advanced, which would strand the ACTIVE entry as a permanent, + # never-closing "orphan" and stall the poll loop for its full budget -- + # AND unique across trajectories: the SDK keys its own step tracking on + # (trajectory_id, step_index), since a sub-agent trajectory can reuse the + # same low step_index values as the main one. Mirrors the SDK's own + # `trajectory_id:step_index` id scheme (falling back to bare step_index + # when trajectory_id is empty, e.g. no sub-agent involved) rather than + # inventing a separate one; call_index further disambiguates multiple + # id-less tool calls within the same step, which the SDK's scheme does not. + trajectory_id = getattr(step, "trajectory_id", "") or "" + step_key = f"{trajectory_id}:{step.step_index}" if trajectory_id else str(step.step_index) + cid = call.id or f"{raw_name}_{step_key}_{call_index}" + self._tool_last_status[cid] = sstatus if cid not in self._seen_tools: self._seen_tools.add(cid) seq = self._next_seq @@ -786,6 +948,25 @@ def _agent_output(self) -> str: return self._agent._sdk_agent.conversation.last_response # type: ignore[union-attr] return "" + def has_orphaned_tool_call(self) -> bool: + """True if any NOT-YET-CLOSED tool call's most recently seen status is + ACTIVE — the structural signature of a backgrounded task the model went + idle on without waiting for. See ``communicate``'s poll loop. + + Deliberately an ALLOWLIST on ACTIVE, not a denylist on "not yet closed + via _closed_tools" alone: the SDK's StepStatus also has WAITING_FOR_USER + (the harness is blocked on a question that will never be answered in + this headless eval), CANCELED, and UNKNOWN — none of which _closed_tools + ever marks done (that set only tracks DONE/ERROR, the states relevant to + result reporting), but none of which the poll loop should ever wait out + either, since they will never become DONE on their own. Checking the + allowlisted ACTIVE status is what tells these apart from a genuine + in-flight background job. The `not in _closed_tools` guard is layered on + top (not a substitute) purely as a monotonicity backstop, in case a + closed id's last-seen entry were ever left at ACTIVE by a re-emission. + """ + return any(cid not in self._closed_tools and s == _STATUS_ACTIVE for cid, s in self._tool_last_status.items()) + def finalize(self, status: AgentEndStatus, *, crashed: bool = False, crash_reason: str | None = None) -> None: """Close orphaned tools, flush leftover blocks, emit TurnEnd + AgentEnd. diff --git a/tests/test_antigravity_agent.py b/tests/test_antigravity_agent.py index 99747950..5a76455a 100644 --- a/tests/test_antigravity_agent.py +++ b/tests/test_antigravity_agent.py @@ -6,6 +6,7 @@ import asyncio import os +from collections.abc import Callable from types import SimpleNamespace import pytest @@ -247,6 +248,8 @@ def _step( usage=None, complete=None, error="", + step_index=0, + trajectory_id="", ): # Plain strings stand in for the SDK's str-enums (_enum_value passes them through). return SimpleNamespace( @@ -262,24 +265,44 @@ def _step( usage_metadata=usage, is_complete_response=complete, error=error, - step_index=0, + step_index=step_index, + trajectory_id=trajectory_id, ) class _FakeConversation: + """Scriptable fake SDK conversation. + + ``steps`` is either a flat list (one batch, yielded on the first + ``receive_steps()`` call) or a list of batches (one per successive + ``receive_steps()`` call — the shape a poll loop drains repeatedly). Once + the authored batches are exhausted, further calls yield an EMPTY batch — + this mirrors the real SDK's local connection, which drains a queue and + returns immediately with nothing once idle; it never replays already- + yielded steps. A test standing in for a background job that never + resolves should author one batch that opens the orphan and let + exhaustion naturally fall through to empty polls, not repeat itself. + """ + def __init__(self, steps): - self._steps = steps + self._batches = list(steps) if steps and isinstance(steps[0], list) else [steps] + self._batch_index = 0 self.last_response = "" + self.receive_steps_call_count = 0 + self.cancel_call_count = 0 async def send(self, prompt, **kwargs): return None async def receive_steps(self): - for s in self._steps: + self.receive_steps_call_count += 1 + batch = self._batches[self._batch_index] if self._batch_index < len(self._batches) else [] + self._batch_index += 1 + for s in batch: yield s async def cancel(self): - return None + self.cancel_call_count += 1 def _agent_with_steps(steps): @@ -291,6 +314,31 @@ def _agent_with_steps(steps): return agent +async def _no_sleep(_seconds: float) -> None: + """Stand-in for asyncio.sleep in poll-loop tests — no real wait.""" + return None + + +class _FiringWatchdog: + """Fake ThreadedWatchdog that fires ``on_timeout`` synchronously at entry. + + Sets ``state.timeout_hit = True`` before any draining happens (exactly + like the real watchdog thread firing early), so a CancelledError raised + later — whether from the first drain or from a poll loop's re-drain — is + classified via the SAME existing ``if state.timeout_hit`` branch. + """ + + def __init__(self, *, on_timeout, **_kwargs): + self._on_timeout = on_timeout + + def __enter__(self): + self._on_timeout() + return self + + def __exit__(self, *_exc): + return False + + async def test_communicate_maps_steps_to_turn_record(): """A realistic think→edit→run→respond stream yields mapped commands, summed tokens, and the final assistant text — exercising the full mapping path.""" @@ -459,17 +507,6 @@ async def test_communicate_timeout_sets_pending_partial_turn(monkeypatch): from coder_eval.errors import TurnTimeoutError - class _FiringWatchdog: - def __init__(self, *, on_timeout, **_kwargs): - self._on_timeout = on_timeout - - def __enter__(self): - self._on_timeout() # watchdog fired: state.timeout_hit = True - return self - - def __exit__(self, *_exc): - return False - monkeypatch.setattr("coder_eval.agents.antigravity_agent.ThreadedWatchdog", _FiringWatchdog) class _Cancelled: @@ -503,6 +540,607 @@ async def test_communicate_requires_started_agent(): await agent.communicate("x") +# --- background-task poll loop (wait_for_wakeup is a dead stub on the Local ------ +# harness; see antigravity_agent.py's communicate() comment + the plan for the +# full evidence trail. The model leaves a tool call open (never DONE) when it +# backgrounds work and goes idle -- these tests drive that signal directly. ------ + + +def test_has_orphaned_tool_call_detects_active_vs_other_statuses(): + """Allowlist on ACTIVE, not a denylist on "not closed": a tool stuck in + WAITING_FOR_USER/CANCELED/UNKNOWN is also never added to _closed_tools + (that set only tracks DONE/ERROR), but must NOT be treated as pollable — + it will never become DONE on its own (Phase-2-review finding). Layered on + top: a cid already in _closed_tools is never orphaned even if its last-seen + status were ever left at ACTIVE by a re-emission (final-review finding).""" + from coder_eval.agents.antigravity_agent import _AntigravityTurnState + + state = _AntigravityTurnState.__new__(_AntigravityTurnState) + state._closed_tools = set() + state._tool_last_status = {} + assert state.has_orphaned_tool_call() is False # no tool calls at all + + state._tool_last_status = {"t1": "ACTIVE"} + assert state.has_orphaned_tool_call() is True # genuinely still running + + state._tool_last_status = {"t1": "DONE"} + assert state.has_orphaned_tool_call() is False # closed normally + + for stuck_status in ["WAITING_FOR_USER", "CANCELED", "UNKNOWN"]: + state._tool_last_status = {"t1": stuck_status} + assert state.has_orphaned_tool_call() is False, ( + f"a tool stuck in {stuck_status} must not trigger polling -- it will never become DONE" + ) + + # A second tool call still ACTIVE is enough, even if the first is DONE. + state._tool_last_status = {"t1": "DONE", "t2": "ACTIVE"} + assert state.has_orphaned_tool_call() is True + + # A closed cid stuck at ACTIVE (e.g. a stale re-emission) must not re-arm the + # poll loop -- closure is authoritative over the last-seen status. + state._closed_tools = {"t1"} + state._tool_last_status = {"t1": "ACTIVE"} + assert state.has_orphaned_tool_call() is False + + +async def test_communicate_fast_path_when_no_orphaned_tools(monkeypatch): + """A normal turn closes its tool call before the stream exhausts -- the poll + loop's condition is False on first check, so it's never entered: exactly one + receive_steps() call, no sleep, byte-identical to today's behavior.""" + from coder_eval.agents import antigravity_agent + + async def _sleep_should_not_be_called(_seconds: float) -> None: + raise AssertionError("asyncio.sleep must not be called on the no-orphan fast path") + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _sleep_should_not_be_called) + + steps = [ + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "t1", {"command_line": "echo hi", "exit_code": 0, "combined_output": "hi"})], + ), + _step("TEXT_RESPONSE", "DONE", content="done", complete=True, usage=_usage(10, 0, 1, 0)), + ] + agent = _agent_with_steps(steps) + tr = await agent.communicate("run it") + + conv = agent._sdk_agent.conversation + assert conv.receive_steps_call_count == 1 + assert tr.agent_output == "done" + + +async def test_communicate_does_not_poll_a_tool_stuck_waiting_for_user(monkeypatch): + """A tool call whose LAST status is WAITING_FOR_USER (not ACTIVE) is never + added to _closed_tools (that set only tracks DONE/ERROR) -- but it must + NOT be mistaken for a genuine backgrounded job either, since a headless + eval run will never actually answer the question. This is the exact gap a + final cross-cutting review found: has_orphaned_tool_call must allowlist + ACTIVE specifically, not just check "not yet closed".""" + from coder_eval.agents import antigravity_agent + + async def _sleep_should_not_be_called(_seconds: float) -> None: + raise AssertionError("asyncio.sleep must not be called for a tool stuck WAITING_FOR_USER") + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _sleep_should_not_be_called) + + steps = [ + _step( + "TOOL_CALL", + "WAITING_FOR_USER", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("ask_question", "t1", {"question": "which region?"})], + ), + _step("TEXT_RESPONSE", "DONE", content="waiting on you", complete=True, usage=_usage(10, 0, 1, 0)), + ] + agent = _agent_with_steps(steps) + tr = await agent.communicate("do it") + + conv = agent._sdk_agent.conversation + assert conv.receive_steps_call_count == 1 # poll loop never entered + ask = next(c for c in tr.commands if c.tool_name == "AskUser") + assert ask.result_status == "unknown" # force-closed as UNRESOLVED by finalize(), not polled forever + + +async def test_communicate_polls_and_resumes_after_orphaned_tool_closes(monkeypatch): + """The model backgrounds a run_command and goes idle -- the tool call stays + ACTIVE (never DONE) even past the final TEXT_RESPONSE. The orphaned-tool + signal triggers a poll; the second receive_steps() call closes the tool and + delivers the real result.""" + from coder_eval.agents import antigravity_agent + + sleep_calls: list[float] = [] + + async def _record_sleep(seconds: float) -> None: + sleep_calls.append(seconds) + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _record_sleep) + + batch1 = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "bg1", {"command_line": "sleep 12 && echo done"})], + ), + _step( + "TEXT_RESPONSE", + "DONE", + content="I've started this in the background.", + content_delta="I've started this in the background.", + complete=True, + usage=_usage(100, 0, 10, 0), + ), + ] + batch2 = [ + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[ + _tc( + "run_command", + "bg1", + {"command_line": "sleep 12 && echo done", "exit_code": 0, "combined_output": "done"}, + ) + ], + usage=_usage(50, 0, 5, 0), + ), + _step( + "TEXT_RESPONSE", + "DONE", + content="All finished.", + content_delta="All finished.", + complete=True, + usage=_usage(60, 0, 8, 0), + ), + ] + agent = _agent_with_steps([batch1, batch2]) + tr = await agent.communicate("do it") + + assert sleep_calls == [antigravity_agent._BACKGROUND_POLL_INTERVAL_SECONDS] + bash = next(c for c in tr.commands if c.tool_name == "Bash") + assert bash.result_status == "success" + assert "All finished." in tr.agent_output + assert agent._sdk_agent.conversation.receive_steps_call_count == 2 + + +async def test_communicate_resolves_backgrounded_tool_call_with_no_id(monkeypatch): + """The SDK types ToolCall.id as optional; the fallback synthetic id must be + stable across a step's own ACTIVE -> DONE re-emission (same step_index), not + derived from a mutable counter -- otherwise the DONE step mints a fresh id + and the ACTIVE entry is orphaned forever, stalling the poll loop for its + full budget on every id-less turn (final-review finding).""" + from coder_eval.agents import antigravity_agent + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _no_sleep) + + batch1 = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", None, {"command_line": "sleep 12 && echo done"})], + ), + _step("TEXT_RESPONSE", "DONE", content="started", complete=True, usage=_usage(10, 0, 1, 0)), + ] + batch2 = [ + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[ + _tc( + "run_command", + None, + {"command_line": "sleep 12 && echo done", "exit_code": 0, "combined_output": "done"}, + ) + ], + ), + _step("TEXT_RESPONSE", "DONE", content="All finished.", complete=True, usage=_usage(5, 0, 1, 0)), + ] + agent = _agent_with_steps([batch1, batch2]) + tr = await agent.communicate("do it") + + assert agent._sdk_agent.conversation.receive_steps_call_count == 2 # closed on the first poll, not the cap + bash = next(c for c in tr.commands if c.tool_name == "Bash") + assert bash.result_status == "success" + assert len(tr.commands) == 1 # the id-less ACTIVE and DONE steps collapsed to ONE tool call, not two + + +async def test_id_less_tool_calls_in_different_trajectories_do_not_collide(): + """step_index is only unique WITHIN a trajectory -- the SDK itself keys step + tracking on (trajectory_id, step_index), since a sub-agent trajectory can + reuse the same low step_index values as the main one. Two id-less tool + calls sharing a step_index but in DIFFERENT trajectories must still mint + distinct fallback cids and produce two separate commands, not collapse + into one (round-3 review finding); same trajectory + same step_index + still collapses to one, as test_communicate_resolves_backgrounded_tool_call_with_no_id covers.""" + steps = [ + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", None, {"command_line": "main job", "exit_code": 0, "output": "main"})], + step_index=1, + trajectory_id="", + ), + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", None, {"command_line": "subagent job", "exit_code": 0, "output": "sub"})], + step_index=1, # same index as the step above, different trajectory + trajectory_id="subagent-42", + ), + _step("TEXT_RESPONSE", "DONE", content="done", complete=True, usage=_usage(10, 0, 1, 0)), + ] + agent = _agent_with_steps(steps) + tr = await agent.communicate("do two things") + + bash_calls = [c for c in tr.commands if c.tool_name == "Bash"] + assert len(bash_calls) == 2 # distinct cids, not collapsed into one + assert {c.tool_id for c in bash_calls} == {"run_command_1_0", "run_command_subagent-42:1_0"} + + +async def test_communicate_handles_two_sequential_background_jobs(monkeypatch): + """paratransit-routing's real observed shape: the model backgrounds a job, + it resolves, and the model immediately backgrounds a SECOND job before + finally finishing -- the loop must not stop after just one poll cycle.""" + from coder_eval.agents import antigravity_agent + + sleep_calls: list[float] = [] + + async def _record_sleep(seconds: float) -> None: + sleep_calls.append(seconds) + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _record_sleep) + + batch1 = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "bgA", {"command_line": "job_a"})], + ), + _step("TEXT_RESPONSE", "DONE", content="started A", complete=True, usage=_usage(10, 0, 1, 0)), + ] + batch2 = [ + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[ + _tc("run_command", "bgA", {"command_line": "job_a", "exit_code": 0, "combined_output": "a done"}) + ], + ), + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "bgB", {"command_line": "job_b"})], + ), + _step("TEXT_RESPONSE", "DONE", content="started B", complete=True, usage=_usage(10, 0, 1, 0)), + ] + batch3 = [ + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[ + _tc("run_command", "bgB", {"command_line": "job_b", "exit_code": 0, "combined_output": "b done"}) + ], + ), + _step("TEXT_RESPONSE", "DONE", content="all done", complete=True, usage=_usage(10, 0, 1, 0)), + ] + agent = _agent_with_steps([batch1, batch2, batch3]) + tr = await agent.communicate("do two things") + + assert len(sleep_calls) == 2 # exactly two poll cycles, one per backgrounded job + bash_calls = [c for c in tr.commands if c.tool_name == "Bash"] + assert len(bash_calls) == 2 + assert all(c.result_status == "success" for c in bash_calls) + assert agent._sdk_agent.conversation.receive_steps_call_count == 3 + + +async def test_communicate_stops_polling_at_max_poll_cap(monkeypatch): + """A pathological, never-closing background job must not poll forever -- + the hard _MAX_BACKGROUND_POLLS cap bounds it independent of the turn budget.""" + from coder_eval.agents import antigravity_agent + + monkeypatch.setattr(antigravity_agent, "_MAX_BACKGROUND_POLLS", 3) + sleep_calls: list[float] = [] + + async def _record_sleep(seconds: float) -> None: + sleep_calls.append(seconds) + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _record_sleep) + + never_closing = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "stuck", {"command_line": "sleep 999999"})], + ), + _step("TEXT_RESPONSE", "DONE", content="waiting...", complete=True, usage=_usage(10, 0, 1, 0)), + ] + # A single batch that opens the orphan; every later call exhausts to an + # empty batch (see _FakeConversation's docstring, matching the real SDK) -- + # the orphan is never closed, simulating a job whose state never changes. + agent = _agent_with_steps([never_closing]) + tr = await agent.communicate("do it forever") + + assert len(sleep_calls) == 3 # exactly _MAX_BACKGROUND_POLLS, not infinite + bash = next(c for c in tr.commands if c.tool_name == "Bash") + assert bash.result_status == "unknown" # force-closed as UNRESOLVED by finalize() + + +class _WatchdogFiresLater: + """Fake ThreadedWatchdog that does NOT fire on entry (unlike _FiringWatchdog + above) -- it hands its ``on_timeout`` callback to the caller so the test can + invoke it mid-poll-loop, simulating a real watchdog thread firing between + poll cycles rather than before the turn even starts.""" + + captured_on_timeout: Callable[[], None] | None = None + + def __init__(self, *, on_timeout, **_kwargs): + _WatchdogFiresLater.captured_on_timeout = on_timeout + + def __enter__(self): + return self + + def __exit__(self, *_exc): + return False + + +async def test_communicate_poll_loop_exits_promptly_once_watchdog_flag_lands(monkeypatch): + """A watchdog timeout landing BETWEEN poll cycles (state.timeout_hit flips + to True while the loop is sleeping) must stop the loop on its next condition + check, not burn through the rest of _MAX_BACKGROUND_POLLS waiting for a + cancellation that may not land on this coroutine right away (final-review + finding: the loop condition must read the flag the watchdog already set).""" + from coder_eval.agents import antigravity_agent + + monkeypatch.setattr(antigravity_agent, "_MAX_BACKGROUND_POLLS", 50) + monkeypatch.setattr("coder_eval.agents.antigravity_agent.ThreadedWatchdog", _WatchdogFiresLater) + + sleep_calls: list[float] = [] + + async def _fire_watchdog_on_second_sleep(seconds: float) -> None: + sleep_calls.append(seconds) + if len(sleep_calls) == 2: + assert _WatchdogFiresLater.captured_on_timeout is not None + _WatchdogFiresLater.captured_on_timeout() + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _fire_watchdog_on_second_sleep) + + never_closing = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "stuck", {"command_line": "sleep 999999"})], + ), + _step("TEXT_RESPONSE", "DONE", content="waiting...", complete=True, usage=_usage(10, 0, 1, 0)), + ] + from coder_eval.errors import TurnTimeoutError + + agent = _agent_with_steps([never_closing]) + with pytest.raises(TurnTimeoutError): + await agent.communicate("do it forever", timeout=30.0) + + # Stopped right after the sleep that flipped timeout_hit -- NOT the (patched) cap of 50. + assert len(sleep_calls) == 2 + # 1 initial drain + 1 poll re-drain (after sleep #1) -- the mid-loop + # `if state.timeout_hit: break` skips the re-drain that would otherwise + # follow sleep #2, so no 3rd receive_steps() call happens. + assert agent._sdk_agent.conversation.receive_steps_call_count == 2 + assert agent.pending_turn is not None + bash = next(c for c in agent.pending_turn.commands if c.tool_name == "Bash") + assert bash.result_status == "unknown" + + +async def test_communicate_respects_should_stop_during_poll(monkeypatch): + """A cooperative-stop request arriving during the poll phase must be + honored before the next re-drain, not ignored until the job finishes.""" + from coder_eval.agents import antigravity_agent + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _no_sleep) + + batch1 = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "bg1", {"command_line": "sleep 999"})], + ), + _step("TEXT_RESPONSE", "DONE", content="started", complete=True, usage=_usage(10, 0, 1, 0)), + ] + batch2 = [ # must never be drained -- should_stop fires right after the sleep + _step( + "TOOL_CALL", + "DONE", + target="TARGET_ENVIRONMENT", + tool_calls=[ + _tc("run_command", "bg1", {"command_line": "sleep 999", "exit_code": 0, "combined_output": "x"}) + ], + ), + ] + agent = _agent_with_steps([batch1, batch2]) + conv = agent._sdk_agent.conversation + + call_count = 0 + + def should_stop() -> bool: + nonlocal call_count + call_count += 1 + return call_count > 2 # False for batch1's 2 steps; True on the post-sleep check + + await agent.communicate("do it", should_stop=should_stop) + + assert conv.receive_steps_call_count == 1 # the poll's re-drain never happened + assert conv.cancel_call_count == 1 + + +class _TwoLayerReentrancyGuardedConversation: + """Faithfully mirrors the REAL SDK's two-generator-layer shape: + ``Conversation.receive_steps()`` (the public method ``_drain()`` calls) is + ITSELF an async generator that delegates to + ``LocalConnection.receive_steps()`` (``async for step in + self._connection.receive_steps(): yield step``, verified against the + installed SDK) -- and the ``_is_receiving`` re-entrancy flag lives on that + INNER, connection-layer generator, not the outer one. A single-layer fake + (putting the flag directly on the generator ``_drain()`` iterates) cannot + catch a bug in how the outer/inner boundary is handled, since aclose()-ing + a generator always closes ITSELF -- the question this fake exists to probe + is whether that also reaches the inner one, and (confirmed live against + real asyncio semantics) it does NOT do so synchronously: a `GeneratorExit` + thrown into a delegating generator's frame does not immediately run the + generator it was mid-iterating -- that's deferred to the event loop's + async-gen finalizer, exactly like the original single-layer bug, just one + level down. ``_drain()``'s fix is therefore a bounded retry (yielding via + ``asyncio.sleep(0)`` for that already-scheduled finalizer to land), not a + claim that the inner generator closes synchronously.""" + + last_response = "" + + def __init__(self, batches): + self._batches = list(batches) + self._batch_index = 0 + self._is_receiving = False # lives on the "connection" layer, like the real SDK + self.receive_steps_call_count = 0 + + async def send(self, prompt, **kwargs): + return None + + async def _connection_receive_steps(self): + if self._is_receiving: + raise RuntimeError("Concurrent receive_steps() calls are not supported on this connection.") + self._is_receiving = True + try: + batch = self._batches[self._batch_index] if self._batch_index < len(self._batches) else [] + self._batch_index += 1 + for s in batch: + yield s + finally: + self._is_receiving = False + + async def receive_steps(self): + # The "Conversation" layer: delegates to the connection layer exactly + # like the real SDK's Conversation.receive_steps() does. + self.receive_steps_call_count += 1 + async for step in self._connection_receive_steps(): + yield step + + async def cancel(self): + return None + + +async def test_communicate_recovers_from_transient_reentrancy_after_cooperative_stop(): + """A cooperative-stop break on a PRIOR communicate() call can leave the + real SDK's inner (connection-layer) generator not-yet-closed for a short + window, since asyncio's async-gen finalizer runs it on a LATER event-loop + turn, not synchronously when the outer generator is aclose()'d (confirmed + live against the real two-layer delegation shape -- round-3 review finding; + see _drain()'s docstring). The NEXT communicate() call must recover by + retrying past that window (mirroring the SDK's own Conversation.send() + handling of this exact RuntimeError) instead of crashing with + AgentCrashError.""" + from pathlib import Path + + batch1 = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "bg1", {"command_line": "sleep 999"})], + ), + _step("TEXT_RESPONSE", "DONE", content="started", complete=True, usage=_usage(10, 0, 1, 0)), + ] + batch2 = [ + _step("TEXT_RESPONSE", "DONE", content="second turn", complete=True, usage=_usage(5, 0, 1, 0)), + ] + conversation = _TwoLayerReentrancyGuardedConversation([batch1, batch2]) + agent = AntigravityAgent(parse_agent_config(type="antigravity")) + agent.working_directory = Path("/tmp") + agent._sdk_agent = SimpleNamespace(conversation=conversation, is_started=True) + + await agent.communicate("do it", should_stop=lambda: True) # breaks after the first step + + # Without the retry, this second call raises AgentCrashError wrapping the + # fake's RuntimeError (verified live before the fix landed). With it, the + # transient window clears within a couple of asyncio.sleep(0) yields and + # the second turn's real content is delivered, not silently dropped. + tr = await agent.communicate("do it again") + assert tr.agent_output == "second turn" + + +async def test_communicate_poll_budget_exhausted_finalizes_via_existing_timeout_path(monkeypatch): + """A watchdog timeout landing during the poll loop's re-drain (not the first + drain) must surface as TurnTimeoutError via the SAME existing exception + branch -- the poll loop must not create a second, inconsistent timeout path. + + Uses ``_WatchdogFiresLater`` (not ``_FiringWatchdog``, which fires at entry + and would make the loop's head condition skip the poll cycle entirely, per + round-3 review) so ``state.timeout_hit`` only flips once a re-drain is + genuinely in flight -- mirroring the real watchdog, whose ``on_timeout`` + callback and the ``CancelledError`` it triggers are the same causal event, + not two independently-timed ones.""" + import asyncio + + from coder_eval.agents import antigravity_agent + from coder_eval.errors import TurnTimeoutError + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _no_sleep) + monkeypatch.setattr("coder_eval.agents.antigravity_agent.ThreadedWatchdog", _WatchdogFiresLater) + + class _FiresWatchdogThenCancelsOnSecondDrain: + last_response = "" + + def __init__(self) -> None: + self.call_count = 0 + + async def send(self, prompt, **kwargs): + return None + + async def receive_steps(self): + self.call_count += 1 + if self.call_count == 1: + yield _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "bg1", {"command_line": "sleep 999"})], + ) + yield _step("TEXT_RESPONSE", "DONE", content="started", complete=True, usage=_usage(10, 0, 1, 0)) + else: + assert _WatchdogFiresLater.captured_on_timeout is not None + _WatchdogFiresLater.captured_on_timeout() + raise asyncio.CancelledError + yield # pragma: no cover - makes this an async generator + + async def cancel(self): + return None + + from pathlib import Path + + agent = AntigravityAgent(parse_agent_config(type="antigravity")) + agent.working_directory = Path("/tmp") + conversation = _FiresWatchdogThenCancelsOnSecondDrain() + agent._sdk_agent = SimpleNamespace(conversation=conversation, is_started=True) + + with pytest.raises(TurnTimeoutError): + await agent.communicate("x", timeout=30.0) + assert conversation.call_count == 2 # the re-drain genuinely ran, not skipped + assert agent.pending_turn is not None + assert agent.pending_turn.crashed is True + + await agent.discard_pending_turn() + assert agent.pending_turn is None + + # --- env_path_prepend / harness-spawn PATH shadowing ------------------------------ # # The localharness subprocess inherits os.environ at Popen time (no SDK env seam), From 3abe902897e6f5972d08d5fd1a62e45399bf8f3b Mon Sep 17 00:00:00 2001 From: joeysbase Date: Wed, 12 Aug 2026 23:28:18 +0000 Subject: [PATCH 2/3] fix: remove redundant asyncio re-import flagged by CodeQL The module already imports asyncio at the top level; the local re-import inside test_communicate_poll_budget_exhausted_finalizes_via_existing_timeout_path was dead weight. The other CodeQL finding on this PR (an unreachable trailing yield in a fake receive_steps() that raises CancelledError first) is a deliberate, necessary idiom -- Python only classifies a function as an async generator if its body contains a yield anywhere, reachable or not, and the same pattern already exists twice, unflagged, on main. Co-Authored-By: Claude Sonnet 5 --- tests/test_antigravity_agent.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/test_antigravity_agent.py b/tests/test_antigravity_agent.py index 5a76455a..a6873eda 100644 --- a/tests/test_antigravity_agent.py +++ b/tests/test_antigravity_agent.py @@ -1088,8 +1088,6 @@ async def test_communicate_poll_budget_exhausted_finalizes_via_existing_timeout_ genuinely in flight -- mirroring the real watchdog, whose ``on_timeout`` callback and the ``CancelledError`` it triggers are the same causal event, not two independently-timed ones.""" - import asyncio - from coder_eval.agents import antigravity_agent from coder_eval.errors import TurnTimeoutError From 59552e8fe8d84d727c76c8f97caf6f6a5b59aff3 Mon Sep 17 00:00:00 2001 From: joeysbase Date: Thu, 13 Aug 2026 15:51:02 +0000 Subject: [PATCH 3/3] fix: derive the poll loop's exit bound from the turn's actual timeout The poll loop's graceful-exit path (force-close a never-resolving orphan as unresolved, finalize and grade normally) was bounded by a fixed cycle count (120 x 5s = 600s) that was double experiments/default.yaml's own default turn_timeout (300s). Since the pre-existing ThreadedWatchdog enforces timeout by cancelling the whole turn, it always won that race under default settings, making the graceful path dead code: a tool call spuriously left ACTIVE with no real background job behind it (a real, observed case from the PR's own validation run) went from "finalizes immediately, graded on whatever the agent wrote" pre-fix to "burns the full 300s, then crashes as TurnTimeoutError with zero criteria graded" post-fix -- a strict regression for that input class. Fixes by deriving a poll_deadline from 0.8x the actual timeout passed to communicate(), falling back to the cycle-based cap only when timeout is None. 0.8 is a fraction of the watchdog's own deadline, not an identical value, so it doesn't reintroduce the race an earlier review round removed -- it's a deliberately earlier internal deadline engineered to reliably win. Caught independently by two PR reviewers (bai-uipath, uipreliga) on the same line of arithmetic. Added a regression test proving a never-resolving orphan under a realistic 300s timeout now finalizes gracefully instead of crashing. Co-Authored-By: Claude Sonnet 5 --- .claude/harness-candidates.md | 39 +++++++++- src/coder_eval/agents/antigravity_agent.py | 91 +++++++++++++++------- tests/test_antigravity_agent.py | 50 ++++++++++++ 3 files changed, 147 insertions(+), 33 deletions(-) diff --git a/.claude/harness-candidates.md b/.claude/harness-candidates.md index edd6e003..3957153f 100644 --- a/.claude/harness-candidates.md +++ b/.claude/harness-candidates.md @@ -325,7 +325,11 @@ with the two `action.yml` items above — one considered change to the action's stranding the ACTIVE entry as a permanent orphan and stalling `communicate()`'s new poll loop for its full `_MAX_BACKGROUND_POLLS` budget on every id-less turn. Fixed by deriving the fallback from `(step.step_index, call_index)` instead (stable across a - step's own re-emissions, per this class's own docstring). Not promoted to a CExxx rule: + step's own re-emissions, per this class's own docstring) -- then, in the same PR, + further folded in `step.trajectory_id` (falling back to bare `step_index` when it's + empty, mirroring the SDK's own `trajectory_id:step_index` id scheme), since a + sub-agent trajectory can reuse the same low `step_index` values as the main one and + two id-less calls across trajectories would otherwise collide. Not promoted to a CExxx rule: this is the only id-fallback-driving-control-flow site in the codebase today (a single call site, not a recurring class per the existing "single call-site fix, no recurring pattern to guard" convention) — a mechanical AST rule for "no mutable @@ -341,9 +345,12 @@ with the two `action.yml` items above — one considered change to the action's before its `task.cancel()` actually landed on this coroutine, the loop kept sleeping/re-draining for up to the full poll budget before the pre-existing post-loop `if state.timeout_hit:` check ever got a chance to run. Fixed by adding - `and not state.timeout_hit` to the condition. Not promoted: `ThreadedWatchdog` + - a bespoke poll loop reading its own state flag is a one-off shape unique to this - agent; no second instance exists to generalize a rule from. Caught in the same + `and not state.timeout_hit` to the condition, plus a mid-body early exit right after + the sleep (`if state.timeout_hit: break`) so a flag landing DURING the sleep skips + the following re-drain too, instead of waiting for the loop's next head check. Not + promoted: `ThreadedWatchdog` + a bespoke poll loop reading its own state flag is a + one-off shape unique to this agent; no second instance exists to generalize a rule + from. Caught in the same final review as above. - [ ] **A regression test's fake dependency must model every layer the fix under test @@ -360,3 +367,27 @@ with the two `action.yml` items above — one considered change to the action's test double is missing a delegation layer the source has" is a semantic match against third-party source, not an AST pattern in our own code — no cheap mechanical check exists. Caught in the round-3 coder-eval-code-review of this same branch. + +- [ ] **An agent's internal sleep-and-retry loop must derive its own exit bound from + the turn's actual `timeout`, never a fixed cycle count picked independently.** The + poll loop's own graceful exit path (force-close a never-resolving orphan as + unresolved, finalize and grade normally) was bounded by `_MAX_BACKGROUND_POLLS * _BACKGROUND_POLL_INTERVAL_SECONDS` + (120 × 5s = 600s) — DOUBLE `experiments/default.yaml`'s own default `turn_timeout: 300`. + Since the pre-existing `ThreadedWatchdog` enforces `timeout` by cancelling the whole + turn, it always won that race under default settings, making the graceful path dead + code: a tool call spuriously left ACTIVE with no real background job behind it (a + real, observed case — see the final validation run) went from "finalizes immediately, + graded on whatever the agent wrote" pre-fix to "burns the full 300s, then crashes as + `TurnTimeoutError` with zero criteria graded" post-fix — a strict regression for that + input class. Fixed by deriving a `poll_deadline` from a fraction (0.8x) of the actual + `timeout` passed to `communicate()`, falling back to the cycle cap only when + `timeout is None`. Caught independently by two reviewers (`bai-uipath`, `uipreliga`) + on the PR, both citing the exact same arithmetic mismatch. **Not promoted in this + pass**, but a stronger candidate than most entries here: `uipreliga` proposed a + generic whole-tree rule (their CE035) — for every sleep-loop under + `src/coder_eval/agents/**`, assert its own cycle-count × interval either references a + timeout-derived name or is provably below `experiments/default.yaml`'s baseline — that + would catch this class of bug in ANY agent, not just this one (confirmed zero + violations on `main` before this bug, one on this PR). Worth a real look next time + `agents/` is touched, since a second agent adding its own disconnected sleep-loop + constant would reintroduce the exact same shape. diff --git a/src/coder_eval/agents/antigravity_agent.py b/src/coder_eval/agents/antigravity_agent.py index 38715425..79718ebe 100644 --- a/src/coder_eval/agents/antigravity_agent.py +++ b/src/coder_eval/agents/antigravity_agent.py @@ -119,15 +119,37 @@ def _harness_spawn_lock() -> asyncio.Lock: # separately-tuned budget. _RECEIVE_STEPS_REENTRY_RETRIES = 5 -# Cap on poll *cycles* per turn -- bounds how many times communicate() will -# sleep-and-redrain. Wall-clock enforcement of the whole turn is a separate -# concern, left to the pre-existing ThreadedWatchdog (see communicate()); this -# cap only matters when a task sets no run_limits.turn_timeout/task_timeout at -# all, or when a single re-drain returns promptly each cycle. -# 120 * 5s = 10 minutes: real backgrounded jobs observed in confirmed-broken -# tasks ran 60-300s, so this carries a ~2x safety margin over the worst one -# seen while still bounding a truly pathological (never-resolving) orphan to a -# reasonable cap. +# Fraction of the turn's configured `timeout` the poll loop is allowed to spend +# waiting on a backgrounded tool call, before giving up and finalizing through +# its OWN graceful path (force-close the orphan as unresolved, grade normally) +# instead of running into the ThreadedWatchdog's harder cutoff at `timeout` +# itself. Deliberately a FRACTION of `timeout`, not `timeout` itself: a check +# against the identical value the watchdog uses races it non-deterministically +# for who fires first (the bug an earlier review round removed); a check +# against a smaller fraction is a strictly earlier, non-racing internal +# deadline whose whole purpose is to reliably win that race. 0.8 leaves the +# watchdog a fifth of the turn's budget as margin for this loop's own exit +# bookkeeping (the warning log, finalize()'s force-close/grade pass) to +# complete before the harder cancellation would land anyway. +# +# This bound is what actually matters: without it, a tool call spuriously left +# ACTIVE with no real background job behind it (observed live -- see the final +# validation run) used to finalize immediately pre-fix and grade whatever the +# agent had already produced. Bounding this loop only by a fixed cycle count +# disconnected from `timeout` (as an earlier revision did: 120 * 5s = 600s, +# double the framework's own default `turn_timeout: 300` in +# experiments/default.yaml) makes the graceful path unreachable in practice -- +# the watchdog always wins first, and the SAME spurious-orphan turn now burns +# the full turn timeout before crashing as TurnTimeoutError with zero criteria +# evaluated, a strict regression for that input class. +_POLL_DEADLINE_TIMEOUT_FRACTION = 0.8 + +# Cap on poll *cycles* per turn -- the SOLE bound when a task sets no +# run_limits.turn_timeout/task_timeout at all (timeout=None), since +# _POLL_DEADLINE_TIMEOUT_FRACTION has nothing to multiply in that case. Also a +# backstop against a very large configured timeout turning this loop into an +# effectively unbounded wait: 120 * 5s = 10 minutes, ~2x the worst real +# backgrounded-job duration observed in confirmed-broken tasks (60-300s). # # Deliberately NOT "break after N consecutive empty polls" instead: the real # SDK's receive_steps() returns identically empty whether a backgrounded job is @@ -566,6 +588,15 @@ def _on_turn_timeout() -> None: emit.on_event(TurnStartEvent(task_id=task_id, turn_id=turn_id, model=model)) conversation = self._sdk_agent.conversation poll_count = 0 + # Bound the poll loop's OWN exit by a fraction of `timeout` so its + # graceful path (force-close the orphan, grade normally) reliably + # wins the race against the ThreadedWatchdog's harder cutoff at + # `timeout` itself, instead of the watchdog always firing first — + # see _POLL_DEADLINE_TIMEOUT_FRACTION's comment for why a FRACTION + # of `timeout` doesn't race it the way an identical value would. + # `timeout=None` has nothing to derive a fraction from, so the + # cycle-based _MAX_BACKGROUND_POLLS is the sole bound in that case. + poll_deadline = turn_start_time + timeout * _POLL_DEADLINE_TIMEOUT_FRACTION if timeout else None try: await conversation.send(user_input) # The cooperative should_stop poll runs AFTER process_step (the @@ -584,31 +615,24 @@ def _on_turn_timeout() -> None: # instead, gated on that orphaned-tool signal so a normal turn # (which always closes its tool calls before the stream # exhausts) takes this branch zero times and finalizes exactly - # as fast as today. No separate in-loop deadline check against - # `timeout`: the ThreadedWatchdog above already enforces it by - # cancelling this whole coroutine, and a second check against - # the SAME value would just race it non-deterministically for - # who fires first. `_MAX_BACKGROUND_POLLS` bounds the number of - # poll *cycles*; the `not state.timeout_hit` check below is a - # fast exit once the watchdog has already decided to fire but - # its cancellation hasn't landed on this coroutine yet (a single - # re-drain can itself await indefinitely while genuinely - # non-idle work is in flight, so true wall-clock safety for an - # unbounded background job still depends on a configured - # `run_limits.turn_timeout`/`task_timeout` reaching the watchdog - # — not on this cap alone). + # as fast as today. while ( not state.stopped_early_hit and not state.timeout_hit and state.has_orphaned_tool_call() - and poll_count < _MAX_BACKGROUND_POLLS + and ( + poll_count < _MAX_BACKGROUND_POLLS + if poll_deadline is None + else time.monotonic() < poll_deadline + ) ): poll_count += 1 self._log.debug("Polling for backgrounded work (orphaned tool call); attempt %d", poll_count) await asyncio.sleep(_BACKGROUND_POLL_INTERVAL_SECONDS) - if state.timeout_hit: - # The watchdog decided to fire during the sleep above; skip - # the re-drain (which could itself await indefinitely on + if state.timeout_hit or (poll_deadline is not None and time.monotonic() >= poll_deadline): + # The watchdog decided to fire during the sleep above, or + # this loop's own (earlier) deadline just passed: skip the + # re-drain (which could itself await indefinitely on # genuinely non-idle work) rather than waiting for the # loop's own head check to catch it next cycle. break @@ -617,9 +641,18 @@ def _on_turn_timeout() -> None: break await self._drain(conversation, state, should_stop) - if poll_count >= _MAX_BACKGROUND_POLLS and state.has_orphaned_tool_call(): - msg = "Poll budget (_MAX_BACKGROUND_POLLS=%d) exhausted with a tool call still ACTIVE." - self._log.warning(msg, _MAX_BACKGROUND_POLLS) + if state.has_orphaned_tool_call() and not state.stopped_early_hit and not state.timeout_hit: + # Exited via this loop's own bound (poll_deadline or the + # cycle cap), not an external stop/timeout -- the tool call + # is force-closed as unresolved in finalize() below and the + # turn is still graded normally on everything else. + bound = ( + f"poll_deadline ({_POLL_DEADLINE_TIMEOUT_FRACTION:.0%} of {timeout:g}s turn timeout)" + if poll_deadline is not None + else f"_MAX_BACKGROUND_POLLS ({_MAX_BACKGROUND_POLLS})" + ) + msg = "Poll budget exhausted (%s, poll_count=%d) with a tool call still ACTIVE." + self._log.warning(msg, bound, poll_count) if state.stopped_early_hit: # Best-effort server-side cancel, mirrors kill(); a raising diff --git a/tests/test_antigravity_agent.py b/tests/test_antigravity_agent.py index a6873eda..e7a15df7 100644 --- a/tests/test_antigravity_agent.py +++ b/tests/test_antigravity_agent.py @@ -877,6 +877,56 @@ async def _record_sleep(seconds: float) -> None: assert bash.result_status == "unknown" # force-closed as UNRESOLVED by finalize() +async def test_communicate_finalizes_gracefully_under_a_realistic_turn_timeout(monkeypatch): + """A never-resolving orphan under a REALISTIC configured timeout (300s, the + framework's own experiments/default.yaml turn_timeout) must finalize through + the poll loop's own graceful path -- force-close the orphan, grade normally + -- instead of the ThreadedWatchdog cutting the whole turn at `timeout` first. + + Pre-fix, `_MAX_BACKGROUND_POLLS * _BACKGROUND_POLL_INTERVAL_SECONDS` (120 * + 5s = 600s) was DOUBLE the 300s default, so the watchdog always won that race + and this exact scenario -- a tool call spuriously left ACTIVE with no real + background job behind it, confirmed live in the final validation run -- burned + the full turn timeout and crashed as TurnTimeoutError with zero criteria + graded, a strict regression versus the pre-fix immediate finalize. Deriving + the poll deadline from a FRACTION of the real `timeout` (not a disconnected + cycle count) fixes it: the loop now exits through its own graceful path with + room to spare before the watchdog's harder cutoff would ever fire.""" + from coder_eval.agents import antigravity_agent + + monkeypatch.setattr(antigravity_agent.asyncio, "sleep", _no_sleep) + + # Fake clock: turn_start_time=0.0, then +130s per subsequent call. The poll + # loop reads time.monotonic() at least twice per iteration (the while-head + # check, then the post-sleep deadline check), so this crosses the 240s + # deadline (0.8 * 300s) after exactly one poll cycle -- proving the exit is + # driven by the deadline, not by exhausting all 120 cycles. + clock = iter([0.0, 130.0, 260.0]) + monkeypatch.setattr(antigravity_agent.time, "monotonic", lambda: next(clock, 1_000_000.0)) + + never_closing = [ + _step( + "TOOL_CALL", + "ACTIVE", + target="TARGET_ENVIRONMENT", + tool_calls=[_tc("run_command", "stuck", {"command_line": "sleep 999999"})], + ), + _step("TEXT_RESPONSE", "DONE", content="waiting...", complete=True, usage=_usage(10, 0, 1, 0)), + ] + agent = _agent_with_steps([never_closing]) + + tr = await agent.communicate("do it forever", timeout=300.0) # the real default turn_timeout + + # Finalized and graded -- no TurnTimeoutError, no crash. + assert tr is not None + assert not tr.crashed + bash = next(c for c in tr.commands if c.tool_name == "Bash") + assert bash.result_status == "unknown" # force-closed as UNRESOLVED by finalize() + # Exited via the poll_deadline (well under the 120-cycle cap), matching a + # real turn where the watchdog's 300s cutoff never gets the chance to fire. + assert agent._sdk_agent.conversation.receive_steps_call_count < 5 + + class _WatchdogFiresLater: """Fake ThreadedWatchdog that does NOT fire on entry (unlike _FiringWatchdog above) -- it hands its ``on_timeout`` callback to the caller so the test can