fix: run kanban dispatcher git work off the event loop (#6059)

* fix: run kanban dispatcher git work off the event loop (fixes #6039)

- dispatch_once / _spawn_worker / _shutdown hand _cleanup_completed_tasks
  and _prepare_worktree to a worker thread, serialised by a class-level
  lock so integrations never run git concurrently in the base checkout.
- Workers start in their own session; shutdown SIGTERMs every worker
  group first, waits for all of them off the loop, then SIGKILLs groups
  that are still alive.
- Sync helpers are unchanged, so the CLI one-shot dispatch path keeps
  working.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

* fix: clean up cancelled worktree prep and always SIGKILL worker groups on shutdown

- Cancelling dispatch during worktree preparation now waits for the git
  thread, discards the unused worktree/branch, closes the run and releases
  the claim before re-raising.
- Shutdown SIGKILLs every worker's process group even when the worker
  itself already exited, so SIGTERM-ignoring descendants cannot survive.
- Tests: new cancellation test; descendant test uses a marker file and a
  SIGTERM-ignoring child instead of os.kill(pid, 0), which zombies pass.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

---------

Co-authored-by: MervinPraison <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>

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):

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"