Skip to content

fix(kanban): terminate workers when tasks are blocked or archived #57713

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nankingjing wants to merge 1 commit into from
Closed

fix(kanban): terminate workers when tasks are blocked or archived #57713

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
218 changes: 118 additions & 100 deletions hermes_cli/kanban_db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4545,6 +4545,7 @@ def block_task(
reason: Optional[str] = None,
kind: Optional[str] = None,
expected_run_id: Optional[int] = None,
signal_fn=None,
) -> bool:
"""Transition ``running``/``ready`` → ``blocked`` (or route elsewhere).

Expand Down Expand Up @@ -4577,8 +4578,11 @@ def block_task(
raise ValueError(
f"block kind must be one of {sorted(VALID_BLOCK_KINDS)} or None"
)
routed_to = "blocked"
worker_pid, claim_lock = _read_task_worker_handles(conn, task_id)
recurrences = 0
run_id = None
_blocked_task = None
blocked_ok = False
with write_txn(conn):
cur_row = conn.execute(
"SELECT status, block_kind, block_recurrences FROM tasks WHERE id = ?",
Expand Down Expand Up @@ -4628,120 +4632,117 @@ def block_task(
conn, task_id, "dependency_wait",
{"reason": reason, "kind": kind}, run_id=run_id,
)
routed_to = "todo"
_blocked_task = get_task(conn, task_id)
_fire_kanban_lifecycle_hook(
"kanban_task_blocked",
task_id,
board=get_current_board(),
assignee=_blocked_task.assignee if _blocked_task else None,
run_id=run_id,
reason=reason,
)
return True

# Truly-blocked kinds. Increment the unblock-loop counter when this is a
# re-block for the SAME reason after a prior unblock. block_task only
# fires from running/ready (i.e. AFTER an unblock returned the task to
# the work pool), so a stored block_kind that matches the incoming kind
# means: blocked → unblocked → about-to-re-block for the same cause.
# An un-typed (None) block compares as "same" to a prior un-typed block.
same_cause = prev_kind == kind
recurrences = prev_recurrences + 1 if same_cause else 1

if recurrences >= BLOCK_RECURRENCE_LIMIT:
# Loop detected — stop letting the unblocker spin this task. Route
# to triage for a human-in-the-loop decision instead of blocked.
cur = conn.execute(
"""
UPDATE tasks
SET status = 'triage',
claim_lock = NULL,
claim_expires = NULL,
worker_pid = NULL,
block_kind = ?,
block_recurrences = ?
WHERE id = ?
AND status IN ('running', 'ready')
""" + ("" if expected_run_id is None else " AND current_run_id = ?"),
(kind, recurrences, task_id) if expected_run_id is None
else (kind, recurrences, task_id, int(expected_run_id)),
)
if cur.rowcount != 1:
return False
run_id = _end_run(
conn, task_id,
outcome="blocked", status="blocked",
summary=reason,
)
if run_id is None and reason:
run_id = _synthesize_ended_run(
conn, task_id, outcome="blocked", summary=reason,
)
_append_event(
conn, task_id, "block_loop_detected",
{
"reason": reason,
"kind": kind,
"recurrences": recurrences,
"limit": BLOCK_RECURRENCE_LIMIT,
},
run_id=run_id,
)
routed_to = "triage"
blocked_ok = True
else:
if expected_run_id is None:
# Truly-blocked kinds. Increment the unblock-loop counter when this is a
# re-block for the SAME reason after a prior unblock. block_task only
# fires from running/ready (i.e. AFTER an unblock returned the task to
# the work pool), so a stored block_kind that matches the incoming kind
# means: blocked → unblocked → about-to-re-block for the same cause.
# An un-typed (None) block compares as "same" to a prior un-typed block.
same_cause = prev_kind == kind
recurrences = prev_recurrences + 1 if same_cause else 1

if recurrences >= BLOCK_RECURRENCE_LIMIT:
# Loop detected — stop letting the unblocker spin this task. Route
# to triage for a human-in-the-loop decision instead of blocked.
cur = conn.execute(
"""
UPDATE tasks
SET status = 'blocked',
SET status = 'triage',
claim_lock = NULL,
claim_expires = NULL,
worker_pid = NULL,
block_kind = ?,
block_recurrences = ?
WHERE id = ?
AND status IN ('running', 'ready')
""",
(kind, recurrences, task_id),
""" + ("" if expected_run_id is None else " AND current_run_id = ?"),
(kind, recurrences, task_id) if expected_run_id is None
else (kind, recurrences, task_id, int(expected_run_id)),
)
else:
cur = conn.execute(
"""
UPDATE tasks
SET status = 'blocked',
claim_lock = NULL,
claim_expires = NULL,
worker_pid = NULL,
block_kind = ?,
block_recurrences = ?
WHERE id = ?
AND status IN ('running', 'ready')
AND current_run_id = ?
""",
(kind, recurrences, task_id, int(expected_run_id)),
if cur.rowcount != 1:
return False
run_id = _end_run(
conn, task_id,
outcome="blocked", status="blocked",
summary=reason,
)
if cur.rowcount != 1:
return False
run_id = _end_run(
conn, task_id,
outcome="blocked", status="blocked",
summary=reason,
)
# Synthesize a run when blocking a never-claimed task so the
# reason is preserved in attempt history.
if run_id is None and reason:
run_id = _synthesize_ended_run(
if run_id is None and reason:
run_id = _synthesize_ended_run(
conn, task_id, outcome="blocked", summary=reason,
)
_append_event(
conn, task_id, "block_loop_detected",
{
"reason": reason,
"kind": kind,
"recurrences": recurrences,
"limit": BLOCK_RECURRENCE_LIMIT,
},
run_id=run_id,
)
blocked_ok = True
else:
if expected_run_id is None:
cur = conn.execute(
"""
UPDATE tasks
SET status = 'blocked',
claim_lock = NULL,
claim_expires = NULL,
worker_pid = NULL,
block_kind = ?,
block_recurrences = ?
WHERE id = ?
AND status IN ('running', 'ready')
""",
(kind, recurrences, task_id),
)
else:
cur = conn.execute(
"""
UPDATE tasks
SET status = 'blocked',
claim_lock = NULL,
claim_expires = NULL,
worker_pid = NULL,
block_kind = ?,
block_recurrences = ?
WHERE id = ?
AND status IN ('running', 'ready')
AND current_run_id = ?
""",
(kind, recurrences, task_id, int(expected_run_id)),
)
if cur.rowcount != 1:
return False
run_id = _end_run(
conn, task_id,
outcome="blocked",
outcome="blocked", status="blocked",
summary=reason,
)
_append_event(
conn, task_id, "blocked",
{"reason": reason, "kind": kind, "recurrences": recurrences},
run_id=run_id,
)
_blocked_task = get_task(conn, task_id)
# Synthesize a run when blocking a never-claimed task so the
# reason is preserved in attempt history.
if run_id is None and reason:
run_id = _synthesize_ended_run(
conn, task_id,
outcome="blocked",
summary=reason,
)
_append_event(
conn, task_id, "blocked",
{"reason": reason, "kind": kind, "recurrences": recurrences},
run_id=run_id,
)
blocked_ok = True
if blocked_ok:
_blocked_task = get_task(conn, task_id)
if not blocked_ok:
return False
_terminate_reclaimed_worker(
worker_pid, claim_lock, signal_fn=signal_fn,
)
_fire_kanban_lifecycle_hook(
"kanban_task_blocked",
task_id,
Expand Down Expand Up @@ -5204,7 +5205,8 @@ def decompose_triage_task(
return child_ids


def archive_task(conn: sqlite3.Connection, task_id: str) -> bool:
def archive_task(conn: sqlite3.Connection, task_id: str, *, signal_fn=None) -> bool:
worker_pid, claim_lock = _read_task_worker_handles(conn, task_id)
with write_txn(conn):
cur = conn.execute(
"UPDATE tasks SET status = 'archived', "
Expand All @@ -5223,6 +5225,9 @@ def archive_task(conn: sqlite3.Connection, task_id: str) -> bool:
summary="task archived with run still active",
)
_append_event(conn, task_id, "archived", None, run_id=run_id)
_terminate_reclaimed_worker(
worker_pid, claim_lock, signal_fn=signal_fn,
)
# ``archived`` parents no longer block children, same as ``done``.
# Promote newly-unblocked dependents immediately instead of waiting
# for a later dispatcher tick.
Expand Down Expand Up @@ -5912,6 +5917,19 @@ def _pid_alive(pid: Optional[int]) -> bool:
return True


def _read_task_worker_handles(
conn: sqlite3.Connection, task_id: str,
) -> tuple[Optional[int], Optional[str]]:
"""Return ``(worker_pid, claim_lock)`` for a task before claim release."""
row = conn.execute(
"SELECT worker_pid, claim_lock FROM tasks WHERE id = ?",
(task_id,),
).fetchone()
if not row:
return None, None
return row["worker_pid"], row["claim_lock"]


def _terminate_reclaimed_worker(
pid: Optional[int],
claim_lock: Optional[str],
Expand Down
97 changes: 97 additions & 0 deletions tests/hermes_cli/test_kanban_worker_terminate_on_block_archive.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
"""Regression tests for #57596 — terminate host-local workers on block/archive.

When a kanban task transitions to ``blocked`` or ``archived``, the dispatcher
must SIGTERM the associated host-local worker process instead of only clearing
``worker_pid`` in the database and leaving the process running.
"""

from __future__ import annotations

from pathlib import Path

import pytest

from hermes_cli import kanban_db as kb


@pytest.fixture
def kanban_home(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setenv("HERMES_KANBAN_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
db_path = kb.kanban_db_path(board="default")
kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
kb.init_db()
return home


@pytest.fixture
def conn(kanban_home):
with kb.connect() as c:
yield c


def _claim_host_worker(conn, task_id: str, *, pid: int = 4242) -> str:
host = kb._claimer_id().split(":", 1)[0]
claimer = f"{host}:worker"
claimed = kb.claim_task(conn, task_id, claimer=claimer)
assert claimed is not None
kb._set_worker_pid(conn, task_id, pid)
return claimer


def test_block_task_terminates_host_local_worker(conn, monkeypatch):
calls: list[tuple] = []

def _fake_terminate(pid, claim_lock, *, signal_fn=None):
calls.append((pid, claim_lock, signal_fn))
return {"termination_attempted": True, "host_local": True, "terminated": True}

monkeypatch.setattr(kb, "_terminate_reclaimed_worker", _fake_terminate)

tid = kb.create_task(conn, title="running", assignee="w")
claimer = _claim_host_worker(conn, tid)

assert kb.block_task(conn, tid, reason="needs human") is True
assert calls == [(4242, claimer, None)]
task = kb.get_task(conn, tid)
assert task is not None
assert task.status == "blocked"
assert task.worker_pid is None


def test_archive_task_terminates_host_local_worker(conn, monkeypatch):
calls: list[tuple] = []

def _fake_terminate(pid, claim_lock, *, signal_fn=None):
calls.append((pid, claim_lock, signal_fn))
return {"termination_attempted": True, "host_local": True, "terminated": True}

monkeypatch.setattr(kb, "_terminate_reclaimed_worker", _fake_terminate)

tid = kb.create_task(conn, title="running", assignee="w")
claimer = _claim_host_worker(conn, tid)

assert kb.archive_task(conn, tid) is True
assert calls == [(4242, claimer, None)]
task = kb.get_task(conn, tid)
assert task is not None
assert task.status == "archived"
assert task.worker_pid is None


def test_block_task_noops_termination_without_worker_handles(conn, monkeypatch):
calls: list[tuple] = []

def _track(pid, claim_lock, *, signal_fn=None):
calls.append((pid, claim_lock))
return {}

monkeypatch.setattr(kb, "_terminate_reclaimed_worker", _track)

tid = kb.create_task(conn, title="ready", assignee="w")
conn.execute("UPDATE tasks SET status = 'running' WHERE id = ?", (tid,))
assert kb.block_task(conn, tid, reason="stuck") is True
assert calls == [(None, None)]
Loading