fix(cron): distinguish and clean up script timeouts

cron/scheduler.py

@@ -17,6 +17,7 @@
 import os
 import re
 import shutil
+import signal
 import subprocess
 import sys
 import threading
@@ -59,6 +60,20 @@ def _summarize_cron_failure_for_delivery(job: dict, error: str | None) -> str:
     text = (error or "unknown error").strip()
     lower = text.lower()
 
+    # Script execution failures never involve provider fallback. Check these
+    # before generic provider wording because a no-agent script can emit text
+    # such as "timeout" or "rate limit" itself.
+    if "script timed out" in lower:
+        return (
+            f"⚠️ Cron '{job_name}' failed: script timeout. "
+            "Full details saved in cron output."
+        )
+    if job.get("no_agent"):
+        return (
+            f"⚠️ Cron '{job_name}' failed: script execution failed. "
+            "Full details saved in cron output."
+        )
+
     # Provider/API failures are the common noisy path. Keep these short.
     if "429" in text or "rate limit" in lower or "usage limit" in lower:
         reason = "rate limit"
@@ -2097,17 +2112,44 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
         from tools.environments.local import _sanitize_subprocess_env
 
         popen_kwargs = {"creationflags": windows_hide_flags()} if sys.platform == "win32" else {}
-        result = subprocess.run(
+        if os.name == "posix":
+            # Give the script and every descendant a dedicated process group so
+            # timeout cleanup cannot leave data collectors running in the
+            # background after their direct parent is killed.
+            popen_kwargs["start_new_session"] = True
+        process = subprocess.Popen(
             argv,
-            capture_output=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
             text=True,
-            timeout=script_timeout,
             cwd=str(path.parent),
             env=_sanitize_subprocess_env(os.environ.copy()),
             **popen_kwargs,
         )
-        stdout = (result.stdout or "").strip()
-        stderr = (result.stderr or "").strip()
+        try:
+            stdout, stderr = process.communicate(timeout=script_timeout)
+        except subprocess.TimeoutExpired:
+            if os.name == "posix":
+                try:
+                    os.killpg(process.pid, signal.SIGTERM)
+                except ProcessLookupError:
+                    pass
+                try:
+                    process.communicate(timeout=5)
+                except subprocess.TimeoutExpired:
+                    try:
+                        os.killpg(process.pid, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+                    process.communicate()
+            else:
+                # Keep Windows on the existing direct-child cleanup behavior.
+                process.kill()
+                process.communicate()
+            return False, f"Script timed out after {script_timeout}s: {path}"
+
+        stdout = (stdout or "").strip()
+        stderr = (stderr or "").strip()
 
         # Redact secrets from both stdout and stderr before any return path.
         try:
@@ -2119,8 +2161,8 @@ def _run_job_script(script_path: str) -> tuple[bool, str]:
             stdout = "[REDACTED - redaction failed]"
             stderr = "[REDACTED - redaction failed]"
 
-        if result.returncode != 0:
-            parts = [f"Script exited with code {result.returncode}"]
+        if process.returncode != 0:
+            parts = [f"Script exited with code {process.returncode}"]
             if stderr:
                 parts.append(f"stderr:\n{stderr}")
             if stdout:

tests/cron/test_cron_script.py

@@ -9,8 +9,10 @@
 
 import json
 import os
+import signal
 import sys
 import textwrap
+import time
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
@@ -195,6 +197,49 @@ def test_script_timeout(self, cron_env, monkeypatch):
         assert success is False
         assert "timed out" in output.lower()
 
+    @pytest.mark.skipif(sys.platform == "win32", reason="Process groups require POSIX")
+    @pytest.mark.live_system_guard_bypass
+    def test_script_timeout_terminates_descendants(self, cron_env, monkeypatch):
+        """A timed-out script must not leave its child process running."""
+        from cron import scheduler as sched_mod
+        from cron.scheduler import _run_job_script
+
+        monkeypatch.setattr(sched_mod, "_SCRIPT_TIMEOUT", 1)
+        pid_file = cron_env / "child.pid"
+        script = cron_env / "scripts" / "spawns_child.py"
+        script.write_text(textwrap.dedent(f"""\
+            import subprocess
+            import sys
+            import time
+
+            child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
+            {str(pid_file)!r} and open({str(pid_file)!r}, "w").write(str(child.pid))
+            time.sleep(60)
+        """))
+
+        child_pid = None
+        try:
+            success, output = _run_job_script(str(script))
+            assert success is False
+            assert "timed out" in output.lower()
+            child_pid = int(pid_file.read_text())
+
+            deadline = time.monotonic() + 5
+            while time.monotonic() < deadline:
+                try:
+                    os.kill(child_pid, 0)
+                except ProcessLookupError:
+                    break
+                time.sleep(0.05)
+            else:
+                pytest.fail("timed-out script left its child process running")
+        finally:
+            if child_pid is not None:
+                try:
+                    os.kill(child_pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+
     def test_script_json_output(self, cron_env):
         """Scripts can output structured JSON for the LLM to parse."""
         from cron.scheduler import _run_job_script

tests/cron/test_run_one_job.py

@@ -13,6 +13,29 @@
 import cron.scheduler as s
 
 
+def test_script_timeout_delivery_is_not_mislabeled_as_provider_timeout():
+    """no_agent script failures never imply provider fallback exhaustion."""
+    message = s._summarize_cron_failure_for_delivery(
+        {"id": "j-script", "name": "watchdog", "no_agent": True},
+        "Script timed out after 7200s: /tmp/watchdog.py",
+    )
+
+    assert "script timeout" in message.lower()
+    assert "provider timeout" not in message.lower()
+    assert "fallback chain" not in message.lower()
+
+
+def test_agent_timeout_delivery_keeps_provider_timeout_summary():
+    """Provider failures for agent jobs retain the concise fallback summary."""
+    message = s._summarize_cron_failure_for_delivery(
+        {"id": "j-agent", "name": "briefing"},
+        "ReadTimeout: provider request timed out",
+    )
+
+    assert "provider timeout" in message.lower()
+    assert "fallback chain was exhausted or unavailable" in message.lower()
+
+
 def _patch_pipeline(monkeypatch, *, success=True, output="out", final="final response",
                     error=None, silent_marker_in=None):
     """Patch the job pipeline primitives and record the call order."""