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

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()},