diff --git a/.claude/harness-candidates.md b/.claude/harness-candidates.md index 01d812cc..3957153f 100644 --- a/.claude/harness-candidates.md +++ b/.claude/harness-candidates.md @@ -314,3 +314,80 @@ 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) -- 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 + 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, 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 + 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. + +- [ ] **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 4e1475be..79718ebe 100644 --- a/src/coder_eval/agents/antigravity_agent.py +++ b/src/coder_eval/agents/antigravity_agent.py @@ -98,6 +98,69 @@ 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 + +# 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 +# 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 +206,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 +470,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 +587,78 @@ 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 # 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. + while ( + not state.stopped_early_hit + and not state.timeout_hit + and state.has_orphaned_tool_call() + 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 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 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 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 - # 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 +812,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 +832,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 +849,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 +981,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..e7a15df7 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,655 @@ 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() + + +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 + 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.""" + 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),