diff --git a/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py b/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py index 0b8d2a71b..81bb6e0d1 100644 --- a/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py +++ b/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py @@ -11,11 +11,13 @@ import logging import math import os import re +import signal import subprocess +import threading import time from typing import Dict, Any, List, Optional -from .._guarded_subprocess import run_guarded +from .._guarded_subprocess import _kill_process_group, run_guarded logger = logging.getLogger(__name__) @@ -86,6 +88,17 @@ class KanbanDispatcher: self._task_runs: Dict[str, int] = {} # task_id -> open run_id self.worker_id = f"gateway_{os.getpid()}" + # Serialises blocking git/SQLite work handed to worker threads, so two + # integrations never run git concurrently in the same base checkout. + _blocking_lock = threading.Lock() + + async def _run_blocking(self, fn, *args): + """Run a blocking (git / SQLite) helper off the event loop.""" + def _locked(): + with self._blocking_lock: + return fn(*args) + return await asyncio.to_thread(_locked) + def _close_run_safe(self, store, run_id, outcome, *, summary=None, metadata=None, error=None) -> bool: """Close a run, swallowing errors so dispatch stays resilient. @@ -196,8 +209,9 @@ class KanbanDispatcher: if not store: return 0 - # Clean up completed processes first - self._cleanup_completed_tasks(store) + # Clean up completed processes first. Integration runs several git + # commands, so keep it off the event loop. + await self._run_blocking(self._cleanup_completed_tasks, store) # Prove liveness for our own children before the reclaim sweep runs, # so a healthy long-running worker is never mistaken for a stale one. @@ -257,8 +271,24 @@ class KanbanDispatcher: ) continue - # Spawn worker process - success = await self._spawn_worker(task, store) + # Spawn worker process. If the dispatcher is stopped while the + # worktree is being prepared, no worker exists yet: close the + # run and hand the claim back before propagating. + try: + success = await self._spawn_worker(task, store) + except asyncio.CancelledError: + self._close_run_safe( + store, run_id, 'failed', + error='Dispatcher stopped before worker spawn', + ) + try: + store.release_claim(task.id, self.worker_id) + except Exception as release_err: + logger.error( + "Failed to release claim for %s on cancel: %s", + task.id, release_err, + ) + raise if not success: # Failed to spawn: close run as failed, count it, and only # release the claim for retry when the failure was recorded @@ -664,6 +694,11 @@ class KanbanDispatcher: return f"worktree removal error: {e}" return None + def _discard_worktree(self, path: str, branch: str) -> None: + """Remove a worktree and branch that no worker ever used.""" + if self._remove_worktree(path, force=True) is None: + self._run_git("branch", "-D", branch) + def _commit_worktree_changes(self, path: str, branch: str) -> None: """Commit any uncommitted edits the worker left in its worktree. @@ -804,7 +839,22 @@ class KanbanDispatcher: # cwd so nothing regresses. worktree_path = None if getattr(task, 'workspace_kind', 'default') == 'worktree': - worktree_path = self._prepare_worktree(task, store) + prep = asyncio.ensure_future( + self._run_blocking(self._prepare_worktree, task, store) + ) + try: + worktree_path = await asyncio.shield(prep) + except asyncio.CancelledError: + # The thread keeps running after cancellation. Let it + # finish, then drop the fresh (unused) worktree and branch + # so a retry can recreate them with `worktree add -b`. + path = await asyncio.shield(prep) + if path: + await self._run_blocking( + self._discard_worktree, path, + f"kanban/{self._safe_task_id(task.id)}", + ) + raise if worktree_path: if not hasattr(self, '_worktrees'): self._worktrees = {} @@ -833,6 +883,9 @@ class KanbanDispatcher: stderr=subprocess.STDOUT, text=True, cwd=worktree_path, + # Own process group so shutdown can stop the worker's + # descendants too (POSIX; ignored on Windows). + start_new_session=(os.name != 'nt'), ) # parent FD closed here; child still has its duped copy @@ -1085,30 +1138,27 @@ class KanbanDispatcher: while self.running_tasks and (time.time() - start_time) < timeout: store = self._get_kanban_store() if store: - self._cleanup_completed_tasks(store) + await self._run_blocking(self._cleanup_completed_tasks, store) await asyncio.sleep(1) - # Force terminate remaining tasks + # Force terminate remaining tasks: signal every worker group + # first, then wait for them together off the event loop. store = self._get_kanban_store() for task_id, process in self.running_tasks.items(): logger.warning(f"Force terminating task {task_id}") - try: - process.terminate() - process.wait(timeout=5) - except subprocess.TimeoutExpired: + _terminate_worker(process) + await asyncio.gather(*( + asyncio.to_thread(_wait_quiet, process, 5) + for process in self.running_tasks.values() + )) + for task_id, process in self.running_tasks.items(): + if process.poll() is None: logger.warning(f"Task {task_id} did not terminate in 5s; sending SIGKILL") - try: - process.kill() - except OSError as kill_err: - logger.error(f"Failed to kill task {task_id}: {kill_err}") - except OSError as os_err: - logger.error(f"OS error while terminating task {task_id}: {os_err}") - try: - process.kill() - except OSError as kill_err: - logger.error(f"Failed to kill task {task_id}: {kill_err}") - # KeyboardInterrupt, SystemExit, CancelledError now propagate as intended + # Even if the worker exited, a descendant may have ignored + # SIGTERM, so always finish off the whole group. + _kill_worker_group(process) + for task_id in list(self.running_tasks): # Close any open run for the interrupted attempt so we do not # leave dangling task_runs rows / stale current_run_id pointers. run_id = self._task_runs.pop(task_id, None) @@ -1124,6 +1174,49 @@ class KanbanDispatcher: logger.info("Kanban dispatcher shutdown complete") +def _terminate_worker(process: subprocess.Popen) -> None: + """Ask a worker and its descendants to exit (SIGTERM to its group).""" + try: + if os.name != 'nt': + os.killpg(process.pid, signal.SIGTERM) + else: + process.terminate() + except OSError: + try: + process.terminate() + except OSError as err: + logger.error(f"Failed to terminate worker {process.pid}: {err}") + + +def _kill_worker_group(process: subprocess.Popen) -> None: + """SIGKILL a worker's group, even after the worker itself was reaped. + + The worker leads its own session, so its group id equals its pid and the + group stays addressable while any descendant is alive. + """ + if os.name == 'nt': + if process.poll() is None: + _kill_process_group(process) + return + try: + os.killpg(process.pid, signal.SIGKILL) + except OSError: + pass # group already empty + if process.poll() is None: + try: + process.kill() + except OSError: + pass + + +def _wait_quiet(process: subprocess.Popen, timeout: float) -> None: + """Wait for a process, ignoring a timeout (the caller escalates).""" + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + pass + + # Global dispatcher management functions async def start_kanban_dispatcher(max_concurrent: int = 3, poll_interval: float = 5.0): diff --git a/src/praisonai-bot/tests/unit/gateway/test_kanban_dispatcher_nonblocking.py b/src/praisonai-bot/tests/unit/gateway/test_kanban_dispatcher_nonblocking.py new file mode 100644 index 000000000..bdc148f02 --- /dev/null +++ b/src/praisonai-bot/tests/unit/gateway/test_kanban_dispatcher_nonblocking.py @@ -0,0 +1,136 @@ +"""The kanban dispatcher must not run blocking git work on the event loop.""" + +import asyncio +import itertools +import os +import subprocess +import sys +import threading +import time +import types + +import pytest + +import praisonai_bot.gateway.kanban_dispatcher as kd +from praisonai_bot.kanban.sqlite_store import SQLiteKanbanStore + +GIT_DELAY = 0.5 + + +class _Stub(kd.KanbanDispatcher): + def _build_execution_command(self, task): + return ["true"] + + +async def _heartbeat(gaps, stop): + last = time.monotonic() + while not stop.is_set(): + await asyncio.sleep(0.05) + now = time.monotonic() + gaps.append(now - last) + last = now + + +@pytest.fixture +def slow_git(tmp_path, monkeypatch): + """Patch ``run_guarded`` with a synchronous sleep, recording overlap.""" + monkeypatch.setenv("PRAISONAI_KANBAN_DB", str(tmp_path / "kanban.db")) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.chdir(tmp_path) + state = {"active": 0, "max_active": 0, "calls": []} + lock = threading.Lock() + + def fake(cmd, *, timeout=None, cwd=None, **kw): + with lock: + state["active"] += 1 + state["max_active"] = max(state["max_active"], state["active"]) + state["calls"].append(" ".join(cmd[:3])) + try: + time.sleep(GIT_DELAY) + if list(cmd[1:3]) == ["worktree", "add"]: + os.makedirs(cmd[3], exist_ok=True) + out = {"rev-parse": "main\n", "rev-list": "0\n"}.get(cmd[1], "") + return subprocess.CompletedProcess(cmd, 0, stdout=out, stderr="") + finally: + with lock: + state["active"] -= 1 + + monkeypatch.setattr(kd, "run_guarded", fake) + return state + + +@pytest.mark.skipif(sys.platform == "win32", reason="uses the POSIX `true` worker") +async def test_dispatch_does_not_block_event_loop(slow_git): + store = SQLiteKanbanStore() + store.create_task({"id": "w1", "title": "isolated", "status": "ready", + "workspace_kind": "worktree"}) + d = _Stub(max_concurrent=1) + gaps, stop = [], asyncio.Event() + hb = asyncio.create_task(_heartbeat(gaps, stop)) + await asyncio.sleep(0.1) + + assert await d.dispatch_once() == 1 # spawn: rev-parse + worktree add + await asyncio.to_thread(d.running_tasks["w1"].wait) + await d.dispatch_once() # completion: commit/merge/cleanup + await asyncio.sleep(0.1) + stop.set() + await hb + + assert "git worktree add" in slow_git["calls"] + assert "git merge --no-ff" in slow_git["calls"] + assert max(gaps) < 0.3, f"event loop stalled for {max(gaps):.2f}s" + + +async def test_blocking_work_is_serialised(slow_git): + a, b = _Stub(), _Stub() + await asyncio.gather( + a._run_blocking(a._run_git, "status"), + b._run_blocking(b._run_git, "status"), + ) + assert slow_git["max_active"] == 1 + + +async def test_cancel_during_worktree_prep_cleans_up(slow_git): + store = SQLiteKanbanStore() + store.create_task({"id": "w1", "title": "isolated", "status": "ready", + "workspace_kind": "worktree"}) + d = _Stub(max_concurrent=1) + t = asyncio.create_task(d.dispatch_once()) + await asyncio.sleep(GIT_DELAY + 0.1) # inside `git worktree add` + t.cancel() + with pytest.raises(asyncio.CancelledError): + await t + + assert "git worktree remove" in slow_git["calls"] + assert "git branch -D" in slow_git["calls"] + assert not d.running_tasks + assert store.get_task("w1").status == "ready" + assert all(r.ended_at for r in store.get_runs("w1")) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups") +async def test_shutdown_terminates_worker_descendants(tmp_path, monkeypatch): + monkeypatch.setenv("PRAISONAI_KANBAN_DB", str(tmp_path / "kanban.db")) + marker = tmp_path / "survived" + + class _Spawner(kd.KanbanDispatcher): + def _build_execution_command(self, task): + # Worker exits on SIGTERM, but its child ignores SIGTERM and + # would leave a marker if it outlived shutdown. + return ["sh", "-c", + f"(trap '' TERM; sleep 1; touch {marker}) & wait"] + + class _Task: + id, board, title, body = "t1", "default", "t1", "" + + d = _Spawner() + assert await d._spawn_worker(_Task(), store=None) + await asyncio.sleep(0.2) + + # Skip the 30 s graceful window; go straight to force-terminate. + ticks = itertools.count(step=1000) + monkeypatch.setattr(kd, "time", types.SimpleNamespace(time=lambda: next(ticks))) + await d._shutdown() + + await asyncio.sleep(1.5) + assert not marker.exists(), "worker's descendant survived shutdown"