From 091ac498fd947fda67035dd398a2cbe21081c102 Mon Sep 17 00:00:00 2001 From: Chengming Li <81849676+Haloukaidi@users.noreply.github.com> Date: Mon, 7 Sep 2026 22:05:01 +0800 Subject: [PATCH] fix(runner): keep stderr chatter from suppressing idle watchdogs --- argus_skill/agent_cli/_run_exec.py | 176 +++++++++++++------------ tests/test_run_exec_stream_callback.py | 111 ++++++++++++++++ 2 files changed, 205 insertions(+), 82 deletions(-) diff --git a/argus_skill/agent_cli/_run_exec.py b/argus_skill/agent_cli/_run_exec.py index 459feffce..2f53ff6fd 100644 --- a/argus_skill/agent_cli/_run_exec.py +++ b/argus_skill/agent_cli/_run_exec.py @@ -414,6 +414,92 @@ def check_wall_clock_limit() -> bool: state.watchdog_terminated = True return True + def check_idle_deadlines() -> None: + nonlocal last_soft_check_at + if state.watchdog_terminated or process.poll() is not None: + return + now = time.monotonic() + idle_seconds = now - last_activity_at + + check_external_interrupt() + check_wall_clock_limit() + + if ( + soft_idle > 0 + and options.inactivity_callback is not None + and process.poll() is None + and idle_seconds >= soft_idle + and (now - last_soft_check_at) >= soft_idle + ): + last_soft_check_at = now + snapshot = InactivitySnapshot( + idle_seconds=idle_seconds, + command=command, + thread_id=state.thread_id, + last_agent_message=( + state.agent_messages[-1] if state.agent_messages else "" + ), + stdout_tail=list(state.stdout_lines)[-50:], + stderr_tail=list(state.stderr_lines)[-50:], + run_label=run_label, + ) + decision = options.inactivity_callback(snapshot) + if decision == "restart": + state.watchdog_reason = ( + f"Restart requested by stall sub-agent after {int(idle_seconds)}s idle." + ) + self._emit( + self._stream_name("stderr", run_label), + f"[watchdog] {state.watchdog_reason}", + ) + self._terminate_process( + process, + include_detached_children=self.backend == BACKEND_OPENCODE, + ) + state.watchdog_terminated = True + + last_message_chars = len(state.agent_messages[-1]) if state.agent_messages else 0 + for stage in idle_escalation.newly_due(idle_seconds): + if process.poll() is not None: + break + if stage == WARNING_STAGE: + self._emit( + self._stream_name("stderr", run_label), + "[watchdog] No model stream event for " + f"{int(idle_seconds)}s (warning threshold " + f"{soft_idle}s, pid={process.pid}, " + f"thread={state.thread_id or '-'}, " + f"stdout_lines={state.stdout_line_count}, " + f"stderr_lines={state.stderr_line_count}, " + f"last_message_chars={last_message_chars}); " + "capturing diagnostics and continuing.", + ) + elif stage == STALLED_STAGE: + self._emit( + self._stream_name("stderr", run_label), + "[watchdog] Model call is likely stalled after " + f"{int(idle_seconds)}s without a stream event " + f"(threshold {stalled_idle}s, pid={process.pid}); " + f"stdout_lines={state.stdout_line_count}, " + f"stderr_lines={state.stderr_line_count}; continuing " + "until the hard deadline.", + ) + elif stage == TERMINATE_STAGE: + state.watchdog_reason = ( + "Forced restart after hard idle timeout " + f"({hard_idle}s without a model stream event)." + ) + self._emit( + self._stream_name("stderr", run_label), + f"[watchdog] {state.watchdog_reason}", + ) + self._terminate_process( + process, + include_detached_children=self.backend == BACKEND_OPENCODE, + ) + state.watchdog_terminated = True + + while True: if process.poll() is not None: if provider_exited_at is None: @@ -459,86 +545,7 @@ def check_wall_clock_limit() -> bool: ) raise except queue.Empty: - now = time.monotonic() - idle_seconds = now - last_activity_at - - check_external_interrupt() - check_wall_clock_limit() - - if ( - soft_idle > 0 - and options.inactivity_callback is not None - and process.poll() is None - and idle_seconds >= soft_idle - and (now - last_soft_check_at) >= soft_idle - ): - last_soft_check_at = now - snapshot = InactivitySnapshot( - idle_seconds=idle_seconds, - command=command, - thread_id=state.thread_id, - last_agent_message=( - state.agent_messages[-1] if state.agent_messages else "" - ), - stdout_tail=list(state.stdout_lines)[-50:], - stderr_tail=list(state.stderr_lines)[-50:], - run_label=run_label, - ) - decision = options.inactivity_callback(snapshot) - if decision == "restart": - state.watchdog_reason = ( - f"Restart requested by stall sub-agent after {int(idle_seconds)}s idle." - ) - self._emit( - self._stream_name("stderr", run_label), - f"[watchdog] {state.watchdog_reason}", - ) - self._terminate_process( - process, - include_detached_children=self.backend == BACKEND_OPENCODE, - ) - state.watchdog_terminated = True - - last_message_chars = len(state.agent_messages[-1]) if state.agent_messages else 0 - for stage in idle_escalation.newly_due(idle_seconds): - if process.poll() is not None: - break - if stage == WARNING_STAGE: - self._emit( - self._stream_name("stderr", run_label), - "[watchdog] No model stream event for " - f"{int(idle_seconds)}s (warning threshold " - f"{soft_idle}s, pid={process.pid}, " - f"thread={state.thread_id or '-'}, " - f"stdout_lines={state.stdout_line_count}, " - f"stderr_lines={state.stderr_line_count}, " - f"last_message_chars={last_message_chars}); " - "capturing diagnostics and continuing.", - ) - elif stage == STALLED_STAGE: - self._emit( - self._stream_name("stderr", run_label), - "[watchdog] Model call is likely stalled after " - f"{int(idle_seconds)}s without a stream event " - f"(threshold {stalled_idle}s, pid={process.pid}); " - f"stdout_lines={state.stdout_line_count}, " - f"stderr_lines={state.stderr_line_count}; continuing " - "until the hard deadline.", - ) - elif stage == TERMINATE_STAGE: - state.watchdog_reason = ( - "Forced restart after hard idle timeout " - f"({hard_idle}s without a model stream event)." - ) - self._emit( - self._stream_name("stderr", run_label), - f"[watchdog] {state.watchdog_reason}", - ) - self._terminate_process( - process, - include_detached_children=self.backend == BACKEND_OPENCODE, - ) - state.watchdog_terminated = True + check_idle_deadlines() continue if text is None: @@ -548,8 +555,13 @@ def check_wall_clock_limit() -> bool: stderr_closed = True continue - last_activity_at = time.monotonic() - idle_escalation.reset() + if stream_name == "stdout": + last_activity_at = time.monotonic() + idle_escalation.reset() + else: + # Diagnostics are retained but are not model stream progress. + # Check deadlines even when stderr keeps the queue nonempty. + check_idle_deadlines() output_stream = self._stream_name(stream_name, run_label) self._emit(output_stream, text) diff --git a/tests/test_run_exec_stream_callback.py b/tests/test_run_exec_stream_callback.py index 83f0050d1..5001c529a 100644 --- a/tests/test_run_exec_stream_callback.py +++ b/tests/test_run_exec_stream_callback.py @@ -10,6 +10,9 @@ from __future__ import annotations import json +import os +import subprocess +import sys import time import pytest @@ -547,3 +550,111 @@ def test_scientist_skill_distill_wall_clock_is_opt_in(monkeypatch) -> None: assert _turn_wall_clock_seconds("scientist.skill_distill") == 45 monkeypatch.setenv("ARGUS_SKILL_SCIENTIST_TURN_MAX_SECONDS", "0") assert _turn_wall_clock_seconds("scientist.skill_distill") == 0 + + +CHATTER = """ +def chatter(seconds): + until = time.monotonic() + seconds + while time.monotonic() < until: + print('ERROR live_writer expected ordinal 23, got 22', file=sys.stderr, flush=True) + time.sleep(0.02) +""" + + +def run_stream(body: str, options: RunnerOptions): + events: list[tuple[str, str]] = [] + runner = AgentCliRunner( + agent_bin=sys.executable, + backend="codex", + event_callback=lambda stream, line: events.append((stream, line)), + ) + command = [sys.executable, "-u", "-c", "import sys, time, json\n" + CHATTER + body] + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, + start_new_session=os.name != "nt", + ) + try: + state = runner._stream_turn_output( + process=process, + command=command, + options=options, + run_label="engineer-r1", + thread_id="stored-thread", + ) + return state, events + finally: + if process.poll() is None: + process.terminate() + process.wait(timeout=3) + + +def test_stderr_chatter_does_not_hide_warning_or_stalled_stage(): + state, events = run_stream( + "chatter(2.6)\n", + RunnerOptions( + watchdog_soft_idle_seconds=1, + watchdog_stalled_idle_seconds=2, + watchdog_hard_idle_seconds=0, + ), + ) + warnings = [line for _, line in events if line.startswith("[watchdog]")] + assert sum("No model stream event" in line for line in warnings) == 1 + assert sum("likely stalled" in line for line in warnings) == 1 + assert not state.watchdog_terminated # Default warning-only policy is preserved. + assert state.stderr_line_count > 20 + assert any("expected ordinal 23, got 22" in line for line in state.stderr_lines) + + +def test_explicit_hard_idle_is_not_bypassed_by_stderr_chatter(): + state, events = run_stream( + "chatter(2.6)\n", + RunnerOptions(watchdog_hard_idle_seconds=1), + ) + assert state.watchdog_terminated + assert "hard idle timeout" in str(state.watchdog_reason).lower() + assert sum("Forced restart" in line for _, line in events) == 1 + + +def test_stdout_tool_progress_keeps_the_stream_active(): + event = {"type": "item.updated", "item": {"id": "command-1", "type": "command_execution", "status": "in_progress", "aggregated_output": "tick"}} + body = "for _ in range(13):\n print(" + repr(json.dumps(event)) + ", flush=True)\n chatter(0.2)\n" + state, events = run_stream( + body, + RunnerOptions(watchdog_soft_idle_seconds=1, watchdog_hard_idle_seconds=1), + ) + assert not state.watchdog_terminated + assert state.stdout_line_count == 13 + assert not any(line.startswith("[watchdog]") for _, line in events) + + +def test_real_stdout_event_resets_warning_stage_after_stderr_chatter(): + state, events = run_stream( + "chatter(1.3)\nprint(json.dumps({'type': 'turn.started'}), flush=True)\nchatter(1.3)\n", + RunnerOptions(watchdog_soft_idle_seconds=1, watchdog_stalled_idle_seconds=2), + ) + assert not state.watchdog_terminated + assert sum("No model stream event" in line for _, line in events) == 2 + assert not any("likely stalled" in line for _, line in events) + + +def test_inactivity_callback_is_reached_during_stderr_chatter(): + snapshots = [] + + def on_idle(snapshot): + snapshots.append(snapshot) + return "restart" + + state, _ = run_stream( + "chatter(2.6)\n", + RunnerOptions(watchdog_soft_idle_seconds=1, inactivity_callback=on_idle), + ) + assert state.watchdog_terminated + assert len(snapshots) == 1 + assert snapshots[0].idle_seconds >= 1 + assert snapshots[0].stderr_tail + assert "stall sub-agent" in str(state.watchdog_reason)