From 8c1649b232df54fb6dbde7f2b5bae373a2e14f6d Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 15 Oct 2026 05:41:37 +0000 Subject: [PATCH] Stop stalled execution agents on idle timeout Why: prevent silent provider stalls from consuming the full execution timeout Evidence: 57 focused tests, configuration validation, 401-test governance regression, and diff check Impact: adds backward-compatible idle_timeout_seconds and fail-stop process supervision Refs: U5 runtime acceptance; docs/superpowers/plans/2026-10-15-project-agent-execution.md --- AGENT-EXECUTION.md | 5 +- scripts/agent_exec.py | 58 +++++++++- scripts/agent_exec_config.py | 17 ++- tests/test_agent_exec.py | 181 +++++++++++++++++++++++++++++++- tests/test_agent_exec_config.py | 71 +++++++++++++ 5 files changed, 324 insertions(+), 8 deletions(-) diff --git a/AGENT-EXECUTION.md b/AGENT-EXECUTION.md index c936190c..eab0483a 100644 --- a/AGENT-EXECUTION.md +++ b/AGENT-EXECUTION.md @@ -8,7 +8,9 @@ parser accepts exactly one `agent-execution` fenced block. U3's execution controller and U4's project wiring are integrated. This file is the execution-routing authority for implementation, repair, test, and review. Installed CLI identity stays unproven until a host runs the U5 probes one -profile at a time. +profile at a time. A provider that makes no successful stdin write or +stdout/stderr read for `idle_timeout_seconds` is stopped before +`timeout_seconds`. ```agent-execution schema: agent-execution/v1 @@ -19,6 +21,7 @@ execution: fallback_to_host: all-unavailable max_parallel: 3 timeout_seconds: 7200 + idle_timeout_seconds: 600 summary_bytes: 16384 raw_output_bytes: 10485760 diff --git a/scripts/agent_exec.py b/scripts/agent_exec.py index 83b2d24a..de6b0043 100644 --- a/scripts/agent_exec.py +++ b/scripts/agent_exec.py @@ -26,6 +26,7 @@ if str(_REPO_ROOT) not in sys.path: from scripts.agent_exec_adapters import build_invocation, probe_adapter from scripts.agent_exec_config import ( + DEFAULT_IDLE_TIMEOUT_SECONDS, SECRET_VALUE_RE, ConfigError, ExecutionConfig, @@ -133,6 +134,7 @@ class AttemptRecord: pid: int | None = None exit_code: int | None = None timed_out: bool = False + idle_timed_out: bool = False signal: int | None = None changed_paths: list[str] = field(default_factory=list) inventory_paths: list[str] = field(default_factory=list) @@ -608,11 +610,13 @@ def _run_attempt( pointer_before = _worktree_pointer_snapshot(worktree, require_git=True) canonical_before = _canonical_checkout_snapshot(repo) timeout = min(unit.timeout_seconds, cfg.execution.timeout_seconds) + idle_timeout = min(cfg.execution.idle_timeout_seconds, timeout) proc_meta = _supervise( invocation.argv, cwd=invocation.cwd, stdin=invocation.stdin, timeout=timeout, + idle_timeout=idle_timeout, raw_log=raw_log, raw_cap=cfg.execution.raw_output_bytes, secret_values=secret_values, @@ -637,6 +641,7 @@ def _run_attempt( pid=proc_meta["pid"], exit_code=proc_meta["exit_code"], timed_out=proc_meta["timed_out"], + idle_timed_out=bool(proc_meta.get("idle_timed_out")), signal=proc_meta["signal"], changed_paths=[], inventory_paths=[], @@ -694,6 +699,7 @@ def _run_attempt( pid=proc_meta["pid"], exit_code=proc_meta["exit_code"], timed_out=proc_meta["timed_out"], + idle_timed_out=bool(proc_meta.get("idle_timed_out")), signal=proc_meta["signal"], changed_paths=list(changed), inventory_paths=list(changed), @@ -756,6 +762,18 @@ def _run_attempt( cleanup_extra={"preserved": True}, ) + if proc_meta.get("idle_timed_out") and not proc_meta["timed_out"]: + # Idle expiry is not a proven-clean failure and must not continue routing. + return _preserved( + "failed-unknown", + "idle timeout; process group terminated", + cleanup_extra={ + "reason": "idle timeout", + "idle_timed_out": True, + "timed_out": False, + }, + ) + if proc_meta["timed_out"] or proc_meta["signal"] is not None: # Timeout/disconnect cannot prove absence of external side effects. return _preserved( @@ -920,6 +938,7 @@ def _supervise( raw_log: Path, raw_cap: int, secret_values: set[str], + idle_timeout: int | None = None, ) -> dict[str, Any]: env = _filtered_env() # Capture for redaction only — never serialize into receipts. @@ -929,6 +948,7 @@ def _supervise( secret_values.add(value) timed_out = False + idle_timed_out = False exit_code: int | None = None sig: int | None = None pid: int | None = None @@ -963,6 +983,7 @@ def _supervise( "pid": None, "exit_code": 127, "timed_out": False, + "idle_timed_out": False, "signal": None, "stdout_tail": "", "stderr_tail": "", @@ -993,11 +1014,25 @@ def _supervise( stdin_fh = None streams = {proc.stdout: stdout_tail, proc.stderr: stderr_tail} - deadline = time.monotonic() + max(timeout, 1) + bounded_timeout = max(int(timeout), 1) + if idle_timeout is None: + idle_limit = min(DEFAULT_IDLE_TIMEOUT_SECONDS, bounded_timeout) + else: + idle_limit = min(max(int(idle_timeout), 1), bounded_timeout) + started = time.monotonic() + deadline = started + bounded_timeout + idle_deadline = started + idle_limit grace_deadline: float | None = None with raw_log.open("wb") as raw_fh: + def _note_progress() -> None: + """Reset idle only after a successful stdin write or stdout/stderr read.""" + nonlocal idle_deadline + if grace_deadline is not None: + return + idle_deadline = time.monotonic() + idle_limit + def _seal_raw_prefix() -> None: nonlocal raw_written, prefix_sealed, raw_prefix if prefix_sealed: @@ -1045,6 +1080,7 @@ def _supervise( return False tail.extend(chunk) _emit_raw(chunk) + _note_progress() return True def _drain_fd_available(fh: Any, tail: _ByteTail) -> None: @@ -1095,6 +1131,8 @@ def _supervise( stdin_fh = None return False stdin_offset += written + if written > 0: + _note_progress() if stdin_offset >= len(stdin_view): try: stdin_fh.close() @@ -1106,8 +1144,12 @@ def _supervise( while streams or proc.poll() is None or stdin_fh is not None: now = time.monotonic() - if grace_deadline is None and now >= deadline: - timed_out = True + if grace_deadline is None and (now >= deadline or now >= idle_deadline): + # Total timeout stays the hard bound when both deadlines are due. + if now >= deadline: + timed_out = True + else: + idle_timed_out = True _kill_process_group(proc.pid) grace_deadline = now + 5 if stdin_fh is not None: @@ -1120,7 +1162,9 @@ def _supervise( break wait = 0.2 if grace_deadline is None: - wait = max(0.0, min(wait, deadline - now)) + wait = max( + 0.0, min(wait, deadline - now, idle_deadline - now) + ) else: wait = max(0.0, min(wait, grace_deadline - now)) readers = list(streams.keys()) @@ -1157,7 +1201,8 @@ def _supervise( _seal_raw_prefix() if proc.poll() is None: - timed_out = True + if not idle_timed_out: + timed_out = True _kill_process_group(proc.pid) try: proc.wait(timeout=5) @@ -1176,6 +1221,7 @@ def _supervise( "pid": pid, "exit_code": exit_code, "timed_out": timed_out, + "idle_timed_out": idle_timed_out, "signal": sig, "stdout_tail": stdout_text, "stderr_tail": stderr_text, @@ -1597,6 +1643,7 @@ def _base_receipt( "pid": a.pid, "exit_code": a.exit_code, "timed_out": a.timed_out, + "idle_timed_out": a.idle_timed_out, "signal": a.signal, "changed_paths": a.changed_paths, "inventory_paths": a.inventory_paths or a.changed_paths, @@ -1630,6 +1677,7 @@ def _config_summary(config: ExecutionConfig | None) -> dict[str, Any] | None: "permission": config.execution.permission, "fallback_to_host": config.execution.fallback_to_host, "timeout_seconds": config.execution.timeout_seconds, + "idle_timeout_seconds": config.execution.idle_timeout_seconds, "summary_bytes": config.execution.summary_bytes, "raw_output_bytes": config.execution.raw_output_bytes, "roles": {k: list(v) for k, v in config.roles.items()}, diff --git a/scripts/agent_exec_config.py b/scripts/agent_exec_config.py index 79b10c25..84aa59b0 100644 --- a/scripts/agent_exec_config.py +++ b/scripts/agent_exec_config.py @@ -31,6 +31,8 @@ MAX_PARALLEL = 32 MAX_TIMEOUT_SECONDS = 86400 MAX_SUMMARY_BYTES = 1_048_576 MAX_RAW_OUTPUT_BYTES = 104_857_600 +# Omitted idle_timeout_seconds uses this cap, then clamps to timeout_seconds. +DEFAULT_IDLE_TIMEOUT_SECONDS = 600 SHELL_META_RE = re.compile(r"[;&|`$<>(){}]|&&|\|\||\$\(") SECRET_KEY_RE = re.compile( @@ -49,10 +51,12 @@ EXECUTION_KEYS = frozenset( "fallback_to_host", "max_parallel", "timeout_seconds", + "idle_timeout_seconds", "summary_bytes", "raw_output_bytes", } ) +EXECUTION_REQUIRED = EXECUTION_KEYS - frozenset({"idle_timeout_seconds"}) PROFILE_KEYS = frozenset({"adapter", "executable", "model", "enabled"}) CUSTOM_ADAPTER_KEYS = frozenset( { @@ -79,6 +83,7 @@ class ExecutionPolicy: timeout_seconds: int summary_bytes: int raw_output_bytes: int + idle_timeout_seconds: int = DEFAULT_IDLE_TIMEOUT_SECONDS @dataclass(frozen=True) @@ -347,7 +352,7 @@ def _build_execution(raw: Any) -> ExecutionPolicy: unknown = set(raw) - EXECUTION_KEYS if unknown: raise ConfigError(f"unknown execution keys: {sorted(unknown)}") - missing = EXECUTION_KEYS - set(raw) + missing = EXECUTION_REQUIRED - set(raw) if missing: raise ConfigError(f"missing execution keys: {sorted(missing)}") @@ -368,6 +373,15 @@ def _build_execution(raw: Any) -> ExecutionPolicy: timeout_seconds = _positive_bounded( raw["timeout_seconds"], "timeout_seconds", 1, MAX_TIMEOUT_SECONDS ) + if "idle_timeout_seconds" in raw: + idle_timeout_seconds = _positive_bounded( + raw["idle_timeout_seconds"], + "idle_timeout_seconds", + 1, + timeout_seconds, + ) + else: + idle_timeout_seconds = min(DEFAULT_IDLE_TIMEOUT_SECONDS, timeout_seconds) summary_bytes = _positive_bounded( raw["summary_bytes"], "summary_bytes", 1, MAX_SUMMARY_BYTES ) @@ -385,6 +399,7 @@ def _build_execution(raw: Any) -> ExecutionPolicy: timeout_seconds=timeout_seconds, summary_bytes=summary_bytes, raw_output_bytes=raw_output_bytes, + idle_timeout_seconds=idle_timeout_seconds, ) diff --git a/tests/test_agent_exec.py b/tests/test_agent_exec.py index 7453376e..457ac1da 100644 --- a/tests/test_agent_exec.py +++ b/tests/test_agent_exec.py @@ -46,6 +46,7 @@ def _write_config( profiles: dict[str, dict[str, object]], role_order: list[str], timeout_seconds: int = 30, + idle_timeout_seconds: int | None = None, summary_bytes: int = 256, raw_output_bytes: int = 1024, ) -> None: @@ -67,6 +68,11 @@ def _write_config( custom_lines.append(" output_format: text") order = ", ".join(role_order) first = role_order[0] + idle_line = ( + f" idle_timeout_seconds: {idle_timeout_seconds}\n" + if idle_timeout_seconds is not None + else "" + ) block = textwrap.dedent( f"""\ schema: agent-execution/v1 @@ -77,7 +83,7 @@ def _write_config( fallback_to_host: all-unavailable max_parallel: 1 timeout_seconds: {timeout_seconds} - summary_bytes: {summary_bytes} +{idle_line} summary_bytes: {summary_bytes} raw_output_bytes: {raw_output_bytes} roles: @@ -721,6 +727,9 @@ class AgentExecControllerTests(unittest.TestCase): data = json.loads(receipt.read_text(encoding="utf-8")) self.assertEqual(data["terminal_status"], "failed-unknown") self.assertTrue(data["attempts"][0].get("timed_out")) + self.assertFalse(data["attempts"][0].get("idle_timed_out")) + self.assertIn("timed out", data["reason"]) + self.assertNotIn("idle timeout", data["reason"]) if child_pid_file.exists(): child = int(child_pid_file.read_text(encoding="utf-8").strip()) # Child must not still be alive. @@ -732,6 +741,176 @@ class AgentExecControllerTests(unittest.TestCase): self.assertNotEqual(still.returncode, 0) self._assert_canonical_untouched() + def test_idle_timeout_kills_silent_worker_before_total(self) -> None: + child_pid_file = self.root / "idle-child.pid" + launched = self.root / "idlenext.log" + launched.write_text("", encoding="utf-8") + secret = "sk-idletimeoutdeadbeef123456" + _write_fake_worker( + self.bindir, + "silent", + body=textwrap.dedent( + f"""\ + # No stdout/stderr. Child stays in this process group. + (sleep 120 & echo $! > "{child_pid_file}"; wait) + """ + ), + ) + _write_fake_worker( + self.bindir, + "idlenext", + body=textwrap.dedent( + f"""\ + echo next >> "{launched}" + exit 0 + """ + ), + ) + self._configure( + profiles={ + "silent": {"adapter": "custom", "executable": "silent"}, + "idlenext": {"adapter": "custom", "executable": "idlenext"}, + }, + role_order=["silent", "idlenext"], + timeout_seconds=40, + idle_timeout_seconds=2, + ) + unit = _write_unit(self.root / "unit.json", timeout_seconds=40) + receipt = self.receipts / "idle.json" + t0 = time.time() + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + env={"AGENT_EXEC_TEST_SECRET": secret}, + ) + elapsed = time.time() - t0 + self.assertLess(elapsed, 20) + self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout) + text = receipt.read_text(encoding="utf-8") + self.assertNotIn(secret, text) + data = json.loads(text) + self.assertEqual(data["terminal_status"], "failed-unknown") + self.assertEqual(data["reason"], "idle timeout; process group terminated") + self.assertEqual(data.get("host_fallback_authorizations", 0), 0) + self.assertNotEqual(data["terminal_status"], "host-fallback-authorized") + self.assertEqual(len(data["attempts"]), 1) + self.assertEqual(launched.read_text(encoding="utf-8"), "") + attempt = data["attempts"][0] + self.assertEqual(attempt["terminal_status"], "failed-unknown") + self.assertTrue(attempt["idle_timed_out"]) + self.assertFalse(attempt["timed_out"]) + self.assertTrue(attempt["workspace_preserved"]) + self.assertTrue(Path(attempt["worktree"]).is_dir()) + cleanup = attempt["cleanup"] + self.assertFalse(cleanup["exact_restore"]) + self.assertEqual(cleanup["reason"], "idle timeout") + self.assertTrue(child_pid_file.is_file()) + child = int(child_pid_file.read_text(encoding="utf-8").strip()) + still = subprocess.run( + ["kill", "-0", str(child)], + check=False, + capture_output=True, + ) + self.assertNotEqual(still.returncode, 0) + self._assert_canonical_untouched() + + def test_progress_keeps_worker_alive_past_one_idle_interval(self) -> None: + _write_fake_worker( + self.bindir, + "paced", + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + python3 - <<'PY' + import sys, time + end = time.monotonic() + 5.0 + while time.monotonic() < end: + sys.stdout.buffer.write(b"tick\\n") + sys.stdout.buffer.flush() + time.sleep(0.5) + PY + mkdir -p "$workspace/src" + printf 'ok\\n' > "$workspace/src/marker.txt" + printf '%s\\n' '{"terminal_status":"completed","summary":"progress","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + self._configure( + profiles={"paced": {"adapter": "custom", "executable": "paced"}}, + role_order=["paced"], + timeout_seconds=40, + idle_timeout_seconds=2, + ) + unit = _write_unit(self.root / "unit.json", timeout_seconds=40) + receipt = self.receipts / "progress.json" + t0 = time.time() + proc = self._run_cli( + "run", + "--repo", + str(self.repo), + "--role", + "implementation", + "--unit", + str(unit), + "--receipt", + str(receipt), + ) + elapsed = time.time() - t0 + self.assertGreaterEqual(elapsed, 5) + self.assertLess(elapsed, 25) + self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout) + data = json.loads(receipt.read_text(encoding="utf-8")) + self.assertEqual(data["terminal_status"], "success") + self.assertIn("src/marker.txt", data["changed_paths"]) + attempt = data["attempts"][0] + self.assertFalse(attempt["idle_timed_out"]) + self.assertFalse(attempt["timed_out"]) + raw = Path(attempt["raw_log"]).read_text(encoding="utf-8") + self.assertGreaterEqual(raw.count("tick"), 8) + self._assert_canonical_untouched() + + def test_stdin_and_stderr_progress_resets_idle_deadline(self) -> None: + code = "\n".join( + [ + "import sys, time", + "end = time.monotonic() + 3.0", + "while time.monotonic() < end:", + " data = sys.stdin.buffer.read(4096)", + " sys.stderr.buffer.write(b'e\\n')", + " sys.stderr.buffer.flush()", + " time.sleep(0.25)", + " if not data:", + " continue", + ] + ) + raw_log = self.root / "paced-raw.log" + t0 = time.monotonic() + meta = agent_exec._supervise( + ("python3", "-c", code), + cwd=self.root, + stdin=b"x" * 200_000, + timeout=8, + idle_timeout=1, + raw_log=raw_log, + raw_cap=1024, + secret_values=set(), + ) + elapsed = time.monotonic() - t0 + self.assertEqual(meta["exit_code"], 0, meta) + self.assertFalse(meta["idle_timed_out"]) + self.assertFalse(meta["timed_out"]) + self.assertGreaterEqual(elapsed, 2.5) + self.assertLess(elapsed, 8) + def test_summary_and_raw_log_caps_are_enforced(self) -> None: _write_fake_worker( self.bindir, diff --git a/tests/test_agent_exec_config.py b/tests/test_agent_exec_config.py index 6ee63078..1599c130 100644 --- a/tests/test_agent_exec_config.py +++ b/tests/test_agent_exec_config.py @@ -7,6 +7,7 @@ import unittest from pathlib import Path from scripts.agent_exec_config import ( + DEFAULT_IDLE_TIMEOUT_SECONDS, ConfigError, ExecutionConfig, load_config, @@ -334,6 +335,76 @@ class AgentExecConfigTests(unittest.TestCase): cfg = load_config(self.repo) self.assertEqual(cfg.profiles["claude"].executable, 'say "hi" 模型') + def test_idle_timeout_defaults_within_total_and_rejects_bounds(self) -> None: + self.assertEqual(DEFAULT_IDLE_TIMEOUT_SECONDS, 600) + write_repo(self.repo, MINIMAL_BLOCK) + cfg = load_config(self.repo) + self.assertEqual(cfg.execution.idle_timeout_seconds, 600) + self.assertLessEqual(cfg.execution.idle_timeout_seconds, cfg.execution.timeout_seconds) + + short = MINIMAL_BLOCK.replace("timeout_seconds: 7200", "timeout_seconds: 60") + write_repo(self.repo, short) + cfg = load_config(self.repo) + self.assertEqual(cfg.execution.timeout_seconds, 60) + self.assertEqual(cfg.execution.idle_timeout_seconds, 60) + + explicit = MINIMAL_BLOCK.replace( + "timeout_seconds: 7200\n", + "timeout_seconds: 7200\n idle_timeout_seconds: 600\n", + ) + write_repo(self.repo, explicit) + self.assertEqual(load_config(self.repo).execution.idle_timeout_seconds, 600) + + equal = MINIMAL_BLOCK.replace( + "timeout_seconds: 7200\n", + "timeout_seconds: 100\n idle_timeout_seconds: 100\n", + ) + write_repo(self.repo, equal) + cfg = load_config(self.repo) + self.assertEqual(cfg.execution.idle_timeout_seconds, 100) + self.assertEqual(cfg.execution.timeout_seconds, 100) + + above = MINIMAL_BLOCK.replace( + "timeout_seconds: 7200\n", + "timeout_seconds: 100\n idle_timeout_seconds: 101\n", + ) + write_repo(self.repo, above) + with self.assertRaises(ConfigError) as ctx: + load_config(self.repo) + self.assertIn("idle_timeout_seconds", str(ctx.exception)) + + zero = MINIMAL_BLOCK.replace( + "timeout_seconds: 7200\n", + "timeout_seconds: 7200\n idle_timeout_seconds: 0\n", + ) + write_repo(self.repo, zero) + with self.assertRaises(ConfigError): + load_config(self.repo) + + negative = MINIMAL_BLOCK.replace( + "timeout_seconds: 7200\n", + "timeout_seconds: 7200\n idle_timeout_seconds: -5\n", + ) + write_repo(self.repo, negative) + with self.assertRaises(ConfigError): + load_config(self.repo) + + flagged = MINIMAL_BLOCK.replace( + "timeout_seconds: 7200\n", + "timeout_seconds: 7200\n idle_timeout_seconds: true\n", + ) + write_repo(self.repo, flagged) + with self.assertRaises(ConfigError): + load_config(self.repo) + + fractional = MINIMAL_BLOCK.replace( + "timeout_seconds: 7200\n", + "timeout_seconds: 7200\n idle_timeout_seconds: 1.5\n", + ) + write_repo(self.repo, fractional) + with self.assertRaises(ConfigError): + load_config(self.repo) + if __name__ == "__main__": unittest.main()