Skip to content

fix(runner): keep stderr chatter from suppressing idle watchdogs #108

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Haloukaidi wants to merge 1 commit into
base: dev
Choose a base branch
from
+205 −82
Open

fix(runner): keep stderr chatter from suppressing idle watchdogs #108

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
176 changes: 94 additions & 82 deletions argus_skill/agent_cli/_run_exec.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Comment on lines +424 to +426
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:
Expand Down Expand Up @@ -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:
Expand All @@ -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)

Expand Down
111 changes: 111 additions & 0 deletions tests/test_run_exec_stream_callback.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@
from __future__ import annotations

import json
import os
import subprocess
import sys
import time

import pytest
Expand Down Expand Up @@ -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)