fix(cron): reap task subprocess group and bound post-kill pipe drain

controlmesh/cron/execution.py

@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import logging
 import os
+import signal
+import sys
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -21,6 +24,14 @@
 
 logger = logging.getLogger(__name__)
 
+_IS_WINDOWS = sys.platform == "win32"
+
+# Hard ceiling for draining a killed subprocess's pipes. Provider CLIs such as
+# ``claude`` spawn worker subprocesses that inherit stdout/stderr; if any of
+# those survivors keep a pipe write-end open, ``communicate()`` would otherwise
+# block the event loop forever. This bound guarantees the loop stays live.
+_POST_KILL_DRAIN_SECONDS = 10.0
+
 
 @dataclass(slots=True)
 class OneShotCommand:
@@ -267,11 +278,45 @@ class OneShotExecutionResult:
     timed_out: bool
 
 
-def _force_kill(proc: asyncio.subprocess.Process) -> None:
-    """Force-kill a subprocess and any descendants."""
+def _kill_subprocess_group(proc: asyncio.subprocess.Process) -> None:
+    """Force-kill a task subprocess together with every descendant it spawned.
+
+    Cron/webhook provider CLIs (e.g. ``claude``) routinely launch worker
+    subprocesses that inherit the parent's stdout/stderr pipes.  Killing only
+    the lead PID leaves those pipe write-ends held open by orphaned
+    grandchildren, so a following ``communicate()`` never observes EOF and the
+    asyncio loop deadlocks.
+
+    Each task is started in its own session (``start_new_session=True``), so
+    signalling the whole process group reliably reaps every pipe-holder even
+    when grandchildren were reparented to init between kill and reap.  The
+    PID-tree kill remains as a best-effort fallback for children that escaped
+    their session.
+    """
+    if _IS_WINDOWS:
+        force_kill_process_tree(proc.pid)
+        return
+    with contextlib.suppress(ProcessLookupError, PermissionError, OSError):
+        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
     force_kill_process_tree(proc.pid)
 
 
+async def _drain_or_abandon(proc: asyncio.subprocess.Process) -> tuple[bytes, bytes]:
+    """Best-effort pipe drain after killing the subprocess group.
+
+    Returns whatever output was flushed within ``_POST_KILL_DRAIN_SECONDS``.
+    If grandchildren still hold the pipes despite the group kill, the drain is
+    abandoned (returning empty buffers) instead of blocking the event loop.
+    """
+    try:
+        async with asyncio.timeout(_POST_KILL_DRAIN_SECONDS):
+            return await proc.communicate()
+    except TimeoutError:
+        return (b"", b"")
+    except asyncio.CancelledError:
+        return (b"", b"")
+
+
 async def execute_one_shot(
     one_shot: OneShotCommand,
     *,
@@ -292,6 +337,7 @@ async def execute_one_shot(
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
         env=env,
+        start_new_session=not _IS_WINDOWS,
         creationflags=_CREATION_FLAGS,
     )
 
@@ -301,11 +347,11 @@ async def execute_one_shot(
             stdout, stderr = await proc.communicate(input=stdin_input)
     except TimeoutError:
         timed_out = True
-        _force_kill(proc)
-        stdout, stderr = await proc.communicate()
+        _kill_subprocess_group(proc)
+        stdout, stderr = await _drain_or_abandon(proc)
     except asyncio.CancelledError:
-        _force_kill(proc)
-        await proc.wait()
+        _kill_subprocess_group(proc)
+        await _drain_or_abandon(proc)
         raise
 
     if timed_out:

tests/cron/test_execution.py

@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 from pathlib import Path
 from unittest.mock import AsyncMock, patch
 
+import pytest
+
 from controlmesh.cli.param_resolver import TaskExecutionConfig
 from controlmesh.cron.execution import (
     OneShotCommand,
@@ -352,3 +355,112 @@ def test_indents_lines(self) -> None:
 
     def test_single_line(self) -> None:
         assert indent("hello", ">> ") == ">> hello"
+
+
+class TestExecuteOneShotTimeoutKill:
+    """The timeout path must reap the whole process group and never block the loop.
+
+    Regression: provider CLIs (e.g. ``claude``) spawn worker subprocesses that
+    inherit stdout/stderr.  Killing only the lead PID left those pipe holders
+    alive, so the post-timeout ``communicate()`` blocked forever and froze the
+    main asyncio loop (no Telegram poller, no cron, no model-cache refresh).
+    """
+
+    def test_starts_subprocess_in_own_session_on_posix(self) -> None:
+        if sys.platform == "win32":
+            pytest.skip("POSIX-only: start_new_session enables process-group kill")
+
+        async def run() -> None:
+            with patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec:
+                proc = AsyncMock()
+                proc.communicate.return_value = (b'{"result":"ok"}', b"")
+                proc.returncode = 0
+                mock_exec.return_value = proc
+                await execute_one_shot(
+                    OneShotCommand(cmd=["/usr/bin/claude", "-p", "--", "hi"]),
+                    cwd=Path("/tmp"),
+                    provider="claude",
+                    timeout_seconds=60,
+                    timeout_label="Test",
+                )
+            assert mock_exec.call_args[1].get("start_new_session") is True
+
+        asyncio.run(run())
+
+    async def test_post_kill_drain_is_bounded(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """A pipe held open by an orphaned grandchild must not block the loop."""
+        monkeypatch.setattr("controlmesh.cron.execution._POST_KILL_DRAIN_SECONDS", 0.05)
+
+        async def hang_forever(_input: bytes | None = None) -> tuple[bytes, bytes]:
+            await asyncio.sleep(3600)
+            return (b"", b"")
+
+        proc = AsyncMock()
+        proc.pid = 424242
+        proc.returncode = None
+        proc.communicate = hang_forever
+
+        async def fake_create(*_args: object, **_kwargs: object) -> AsyncMock:
+            return proc
+
+        monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)
+
+        kill_calls: list[int] = []
+        monkeypatch.setattr(
+            "controlmesh.cron.execution.force_kill_process_tree",
+            lambda pid: kill_calls.append(pid),
+        )
+        monkeypatch.setattr("os.getpgid", lambda _pid: 424242)
+        monkeypatch.setattr("os.killpg", lambda *_a, **_k: None)
+
+        result = await execute_one_shot(
+            OneShotCommand(cmd=["/usr/bin/claude", "-p", "--", "hi"]),
+            cwd=Path("/tmp"),
+            provider="claude",
+            timeout_seconds=0.01,
+            timeout_label="Test",
+        )
+
+        # The bounded drain returned instead of hanging on the dead pipe.
+        assert result.timed_out is True
+        assert result.status == "error:timeout"
+        assert result.stdout == b""
+        assert 424242 in kill_calls
+
+    async def test_timeout_kills_process_group(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """On timeout the whole session/group is signalled, not just the lead PID."""
+        monkeypatch.setattr("controlmesh.cron.execution._POST_KILL_DRAIN_SECONDS", 0.05)
+
+        async def hang_forever(_input: bytes | None = None) -> tuple[bytes, bytes]:
+            await asyncio.sleep(3600)
+            return (b"", b"")
+
+        proc = AsyncMock()
+        proc.pid = 777
+        proc.returncode = None
+        proc.communicate = hang_forever
+
+        async def fake_create(*_args: object, **_kwargs: object) -> AsyncMock:
+            return proc
+
+        monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)
+
+        group_signalled: list[int] = []
+        monkeypatch.setattr(
+            "controlmesh.cron.execution.force_kill_process_tree", lambda _pid: None
+        )
+        monkeypatch.setattr("os.getpgid", lambda pid: pid)
+        monkeypatch.setattr(
+            "os.killpg",
+            lambda pgid, _sig: group_signalled.append(pgid),
+        )
+
+        await execute_one_shot(
+            OneShotCommand(cmd=["/usr/bin/claude", "-p", "--", "hi"]),
+            cwd=Path("/tmp"),
+            provider="claude",
+            timeout_seconds=0.01,
+            timeout_label="Test",
+        )
+
+        assert 777 in group_signalled