fix(cron): tree-kill script timeout descendants via agent.deadline.kill_process_tree

The script-timeout path used a site-local process-group kill, which
cannot reach a grandchild that created its OWN session (start_new_session
background jobs, watchdogs). Such descendants kept running after the job
reported failure (#71148, #59549). Migrate the timeout handler to the
unified deadline layer's kill_process_tree (#85147, d6a5cb9725): psutil
snapshots the descendant set before signalling, so own-session
grandchildren are reached too. Fallback to the site-local group kill if
the import ever fails, so the path cannot re-wedge.

The explicit script-timeout message stays the classification anchor
(#85536's contract), keeping cron timeouts distinct from provider
timeouts.

Salvage additions on review (#85125 Phase 4a):
- migrate the sibling kill site too — the cancel_event/"ownership was
  lost" path orphaned setsid grandchildren the same way (whole-bug-class
  rule); pinned by test_cancel_path_also_tree_kills
- proc.poll() early-return in _terminate_cron_script_tree so a script
  that exits right at the deadline doesn't log a spurious "no signal"
  warning (mirrors _terminate_cron_script_process); pinned by
  test_already_exited_proc_is_left_alone
- acceptance test's script timeout 1s -> 2s: interpreter startup under
  CI load could eat the whole 1s window before the spawner wrote its
  pid file
- note: kill_process_tree hard-kills (SIGKILL) immediately, whereas the
  old path gave a 1s SIGTERM grace window; intended for a deadline-
  expiry hard stop (both docstrings say "hard stop")

Based on #86791 by @ayushnangia; cherry-picked to preserve authorship.

Co-authored-by: dante32683 <dante32683@users.noreply.github.com>
Co-authored-by: supotato-ipj <supotato-ipj@users.noreply.github.com>

contributors/emails/dante32683@users.noreply.github.com

@@ -0,0 +1 @@
+dante32683

contributors/emails/supotato-ipj@users.noreply.github.com

@@ -0,0 +1 @@
+supotato-ipj

cron/scheduler.py

@@ -4092,6 +4092,53 @@ def _terminate_cron_script_process(proc: subprocess.Popen) -> None:
         proc.wait(timeout=1.0)
 
 
+def _terminate_cron_script_tree(proc: subprocess.Popen) -> None:
+    """Terminate a script tree, then fall back to the local process-group path."""
+    if proc.poll() is not None:
+        # Already exited (e.g. finished right at the deadline): nothing to
+        # signal, and calling kill_process_tree on a reaped pid would log a
+        # spurious "no signal" warning. Mirrors _terminate_cron_script_process.
+        return
+    pid = getattr(proc, "pid", None)
+    if not isinstance(pid, int) or pid <= 0:
+        logger.warning(
+            "Cron script tree-kill received invalid pid %r; "
+            "falling back to process-group termination",
+            pid,
+        )
+        _terminate_cron_script_process(proc)
+        return
+    try:
+        # Function-local so tests can monkeypatch agent.deadline.kill_process_tree;
+        # separate from the kill try below so a packaging/import problem
+        # surfaces as what it is instead of masquerading as a kill failure.
+        from agent.deadline import kill_process_tree
+    except Exception:
+        logger.warning(
+            "agent.deadline.kill_process_tree unavailable; "
+            "falling back to process-group termination",
+            exc_info=True,
+        )
+        _terminate_cron_script_process(proc)
+        return
+    try:
+        if kill_process_tree(pid):
+            return
+        logger.warning(
+            "Cron script tree-kill reported no signal for pid %s; "
+            "falling back to process-group termination",
+            pid,
+        )
+    except Exception:
+        logger.warning(
+            "Cron script tree-kill failed for pid %s; "
+            "falling back to process-group termination",
+            pid,
+            exc_info=True,
+        )
+    _terminate_cron_script_process(proc)
+
+
 def _drain_script_pipes(proc: subprocess.Popen) -> None:
     """Reap a terminated script process without ever blocking indefinitely.
 
@@ -4320,12 +4367,23 @@ def _run_job_script(
         deadline = time.monotonic() + script_timeout
         while True:
             if cancel_event is not None and cancel_event.is_set():
-                _terminate_cron_script_process(proc)
+                # Same bug class as the timeout site below: a cancelled fire
+                # must not orphan own-session grandchildren either.
+                _terminate_cron_script_tree(proc)
                 _drain_script_pipes(proc)
                 return False, "Script cancelled because cron fire ownership was lost"
             remaining = deadline - time.monotonic()
             if remaining <= 0:
-                _terminate_cron_script_process(proc)
+                # Phase 4a (#85125): a script timeout must leave ZERO living
+                # descendants. killpg only reaches the script's own process
+                # group — a grandchild that called setsid (backgrounded
+                # shell jobs, watchdogs) escapes it and keeps running after
+                # the job reports failure (#71148 / #59549).
+                # agent.deadline.kill_process_tree snapshots the descendant
+                # set via psutil BEFORE signalling, so own-session
+                # grandchildren are reached too — the unified deadline
+                # layer's tree-kill (#85147, d6a5cb9725).
+                _terminate_cron_script_tree(proc)
                 _drain_script_pipes(proc)
                 return False, f"Script timed out after {script_timeout}s: {path}"
             try:

tests/cron/test_cron_script.py

@@ -10,11 +10,13 @@ Tests cover:
 import json
 import os
 import re
+import subprocess
 import sys
 import textwrap
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from types import SimpleNamespace
+from typing import cast
 
 import pytest
 
@@ -629,3 +631,179 @@ class TestRunJobEnvVarCleanup:
         assert os.environ.get("HERMES_SESSION_PLATFORM") is None
         assert os.environ.get("HERMES_SESSION_CHAT_ID") is None
         assert os.environ.get("HERMES_SESSION_CHAT_NAME") is None
+
+
+class TestScriptTimeoutTreeKill:
+    """Phase 4a (#85125): a script timeout must leave zero living descendants."""
+
+    def test_unified_tree_kill_failure_falls_back(self, monkeypatch, caplog):
+        from agent import deadline
+        from cron import scheduler as sched
+
+        proc = SimpleNamespace(pid=12345, poll=lambda: None)
+        fallback_calls = []
+        monkeypatch.setattr(deadline, "kill_process_tree", lambda _pid: False)
+        monkeypatch.setattr(
+            sched,
+            "_terminate_cron_script_process",
+            lambda candidate: fallback_calls.append(candidate),
+        )
+
+        with caplog.at_level("WARNING", logger=sched.__name__):
+            sched._terminate_cron_script_tree(cast("subprocess.Popen", proc))
+
+        assert fallback_calls == [proc]
+        assert "falling back to process-group termination" in caplog.text
+
+    def test_invalid_pid_never_reaches_unified_tree_kill(self, monkeypatch, caplog):
+        from agent import deadline
+        from cron import scheduler as sched
+
+        proc = SimpleNamespace(pid=0, poll=lambda: None)
+        tree_kill_calls = []
+        fallback_calls = []
+        monkeypatch.setattr(
+            deadline,
+            "kill_process_tree",
+            lambda pid: tree_kill_calls.append(pid),
+        )
+        monkeypatch.setattr(
+            sched,
+            "_terminate_cron_script_process",
+            lambda candidate: fallback_calls.append(candidate),
+        )
+
+        with caplog.at_level("WARNING", logger=sched.__name__):
+            sched._terminate_cron_script_tree(cast("subprocess.Popen", proc))
+
+        assert tree_kill_calls == []
+        assert fallback_calls == [proc]
+        assert "invalid pid 0" in caplog.text
+
+    def test_already_exited_proc_is_left_alone(self, monkeypatch):
+        """A script that finished right at the deadline needs no signalling —
+        and must not produce a spurious "no signal" warning."""
+        from agent import deadline
+        from cron import scheduler as sched
+
+        proc = SimpleNamespace(pid=12345, poll=lambda: 0)
+        tree_kill_calls = []
+        fallback_calls = []
+        monkeypatch.setattr(
+            deadline,
+            "kill_process_tree",
+            lambda pid: tree_kill_calls.append(pid) or True,
+        )
+        monkeypatch.setattr(
+            sched,
+            "_terminate_cron_script_process",
+            lambda candidate: fallback_calls.append(candidate),
+        )
+
+        sched._terminate_cron_script_tree(cast("subprocess.Popen", proc))
+
+        assert tree_kill_calls == []
+        assert fallback_calls == []
+
+    def test_cancel_path_also_tree_kills(self, monkeypatch, cron_env):
+        """The ownership-lost/cancel kill site is the timeout site's sibling:
+        it must go through the same tree-kill (#71148 class)."""
+        from cron import scheduler as sched
+
+        tree_calls = []
+
+        def _record_and_kill(proc):
+            # Record the routing, then really kill so _drain_script_pipes
+            # reaps instantly instead of waiting out its 5s communicate().
+            tree_calls.append(proc.pid)
+            proc.kill()
+
+        monkeypatch.setattr(sched, "_terminate_cron_script_tree", _record_and_kill)
+
+        class _Cancelled:
+            def is_set(self):
+                return True
+
+            def set(self):
+                pass
+
+        scripts_dir = cron_env / "scripts"
+        (scripts_dir / "long.py").write_text(
+            "import time; time.sleep(30)\n", encoding="utf-8"
+        )
+        ok, out = sched._run_job_script(
+            str(scripts_dir / "long.py"),
+            workdir=str(cron_env),
+            cancel_event=_Cancelled(),
+        )
+        assert not ok
+        assert "ownership was lost" in out
+        assert len(tree_calls) == 1
+
+    @pytest.mark.live_system_guard_bypass
+    def test_timeout_leaves_no_setsid_grandchild(self, cron_env, monkeypatch):
+        """The script spawns a grandchild in its OWN session (start_new_session).
+        killpg alone cannot reach it; agent.deadline.kill_process_tree must —
+        after the timeout the grandchild must no longer be running."""
+        import time
+
+        psutil = pytest.importorskip(
+            "psutil",
+            reason="kill_process_tree needs psutil to reach own-session descendants",
+        )
+
+        from cron import scheduler as sched
+
+        def is_live(pid):
+            try:
+                process = psutil.Process(pid)
+                return process.is_running() and process.status() != psutil.STATUS_ZOMBIE
+            except (psutil.NoSuchProcess, psutil.ZombieProcess):
+                return False
+
+        scripts_dir = cron_env / "scripts"
+        pid_file = cron_env / "grandchild.pid"
+        (scripts_dir / "spawner.py").write_text(
+            "import subprocess, sys, time\n"
+            "p = subprocess.Popen(\n"
+            "    [sys.executable, '-c', 'import time; time.sleep(30)'],\n"
+            "    start_new_session=True,\n"
+            "    stdin=subprocess.DEVNULL,\n"
+            "    stdout=subprocess.DEVNULL,\n"
+            "    stderr=subprocess.DEVNULL,\n"
+            ")\n"
+            f"open({str(pid_file)!r}, 'w').write(str(p.pid))\n"
+            "time.sleep(30)\n",
+            encoding="utf-8",
+        )
+        monkeypatch.setenv("HERMES_CRON_SCRIPT_TIMEOUT", "2")
+        monkeypatch.setattr(sched, "_SCRIPT_TIMEOUT", sched._DEFAULT_SCRIPT_TIMEOUT)
+
+        ok, out = sched._run_job_script(
+            str(scripts_dir / "spawner.py"), workdir=str(cron_env)
+        )
+        assert not ok, f"script should have timed out, got {out!r}"
+
+        deadline = time.monotonic() + 5
+        gpid = None
+        while time.monotonic() < deadline and gpid is None:
+            try:
+                gpid = int(pid_file.read_text().strip())
+            except (FileNotFoundError, ValueError):
+                time.sleep(0.05)
+        assert gpid is not None, "spawner never wrote the grandchild pid"
+
+        try:
+            deadline = time.monotonic() + 5
+            while is_live(gpid) and time.monotonic() < deadline:
+                time.sleep(0.05)
+            assert not is_live(gpid), (
+                f"grandchild pid {gpid} survived the script timeout — the "
+                "timeout path orphaned an own-session descendant"
+            )
+        finally:
+            if is_live(gpid):
+                try:
+                    psutil.Process(gpid).kill()
+                except psutil.NoSuchProcess:
+                    pass