From 36980f2ab44f410fbf17205256d1389614a61501 Mon Sep 17 00:00:00 2001 From: muqiao215 <268317993+muqiao215@users.noreply.github.com> Date: Mon, 22 Jun 2026 22:29:01 +0800 Subject: [PATCH] fix(cron): reap task subprocess group and bound post-kill pipe drain Provider CLIs spawned by cron/webhook one-shot tasks (e.g. `claude`) launch worker subprocesses that inherit the parent's stdout/stderr pipes. The previous cleanup only force-killed the lead PID and then called an unbounded `proc.communicate()`. When an orphaned grandchild kept a pipe write-end open, `communicate()` never saw EOF and blocked the main asyncio loop indefinitely. Symptom across the fleet: controlmesh.service stayed active (running) with Errors: 0, but every periodic task (Telegram/Lark polling, model-cache refresh, further cron jobs) froze in the morning cron window and bots stopped responding. Fix: - Start each task subprocess in its own session/process-group (`start_new_session=True` on POSIX) so the whole tree can be signalled. - Replace the lead-PID kill with a group-aware kill (`os.killpg`) plus the existing PID-tree kill as a best-effort fallback. - Bound the post-kill pipe drain (`_POST_KILL_DRAIN_SECONDS`); abandon the pipes instead of blocking the loop if descendants survive the group kill. Tests: - subprocess is started with `start_new_session=True` on POSIX - a pipe held open by a simulated orphan no longer blocks the loop - the process group (not just the lead PID) is signalled on timeout --- controlmesh/cron/execution.py | 58 ++++++++++++++++-- tests/cron/test_execution.py | 112 ++++++++++++++++++++++++++++++++++ 2 files changed, 164 insertions(+), 6 deletions(-) diff --git a/controlmesh/cron/execution.py b/controlmesh/cron/execution.py index 2a17f6b..3374f10 100644 --- a/controlmesh/cron/execution.py +++ b/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: diff --git a/tests/cron/test_execution.py b/tests/cron/test_execution.py index 376db40..5470a44 100644 --- a/tests/cron/test_execution.py +++ b/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