feat(kanban): add task lifecycle plugin hooks (claimed/completed/blocked) (#50349)

Plugins could observe session/tool/approval lifecycle but had no way to
observe kanban task transitions. Adds three observer hooks fired by the
board's claim/complete/block transitions:

  - kanban_task_claimed   (dispatcher process, before worker spawn)
  - kanban_task_completed (worker process, carries summary)
  - kanban_task_blocked   (worker process, carries reason)

Each fires AFTER the DB write txn commits, so a plugin observes durable
state and a slow/hanging callback can never hold the SQLite write lock.
All firing is best-effort: a raising hook is logged and swallowed and
never breaks a board transition. profile_name is resolved from
HERMES_HOME so dispatcher- and worker-side hooks carry the right profile.

Requested by @Smithangshu on Discord.

hermes_cli/kanban_db.py

@@ -103,6 +103,32 @@ VALID_WORKSPACE_KINDS = {"scratch", "worktree", "dir"}
 KNOWN_TOOLSET_NAMES = frozenset(name.casefold() for name in get_toolset_names())
 _IS_WINDOWS = sys.platform == "win32"
 
+
+def _fire_kanban_lifecycle_hook(event: str, task_id: str, **fields: Any) -> None:
+    """Fire a kanban lifecycle plugin hook, fully best-effort.
+
+    Called by the claim/complete/block transitions AFTER their write txn has
+    committed, so plugin code never runs while a SQLite write lock is held and
+    always observes durable board state. Any failure (plugins unavailable,
+    a plugin raising, import error) is swallowed — a misbehaving observer must
+    never break a board state transition.
+
+    ``profile_name`` is resolved from the active HERMES_HOME so dispatcher- and
+    worker-side hooks both carry the right profile without the caller plumbing
+    it through.
+    """
+    try:
+        from hermes_cli.plugins import invoke_hook
+        from hermes_cli.profiles import get_active_profile_name
+        try:
+            profile_name = get_active_profile_name()
+        except Exception:
+            profile_name = "default"
+        invoke_hook(event, task_id=task_id, profile_name=profile_name, **fields)
+    except Exception as exc:  # pragma: no cover - defensive
+        _log.debug("kanban lifecycle hook %s failed: %s", event, exc)
+
+
 # A running task's claim is valid for 15 minutes by default; after that the
 # next dispatcher tick reclaims it. Workers that outlive this window should
 # call ``heartbeat_claim(task_id)`` periodically. In practice most kanban
@@ -3175,7 +3201,15 @@ def claim_task(
             {"lock": lock, "expires": expires, "run_id": run_id},
             run_id=run_id,
         )
-        return get_task(conn, task_id)
+        claimed = get_task(conn, task_id)
+    _fire_kanban_lifecycle_hook(
+        "kanban_task_claimed",
+        task_id,
+        board=get_current_board(),
+        assignee=claimed.assignee if claimed else None,
+        run_id=run_id,
+    )
+    return claimed
 
 
 def claim_review_task(
@@ -3841,6 +3875,15 @@ def complete_task(
     recompute_ready(conn)
     # Clean up the scratch workspace and any stale tmux session for the worker.
     _cleanup_workspace(conn, task_id)
+    _done_task = get_task(conn, task_id)
+    _fire_kanban_lifecycle_hook(
+        "kanban_task_completed",
+        task_id,
+        board=get_current_board(),
+        assignee=_done_task.assignee if _done_task else None,
+        run_id=run_id,
+        summary=(summary if summary is not None else result),
+    )
     return True
 
 
@@ -4264,7 +4307,16 @@ def block_task(
                 summary=reason,
             )
         _append_event(conn, task_id, "blocked", {"reason": reason}, run_id=run_id)
-        return True
+        _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
 
 
 

hermes_cli/plugins.py

@@ -167,6 +167,31 @@ VALID_HOOKS: Set[str] = {
     #   choice: "once" | "session" | "always" | "deny" | "timeout"
     "pre_approval_request",
     "post_approval_response",
+    # Kanban task lifecycle hooks. Fired by hermes_cli.kanban_db when a task
+    # transitions state, AFTER the change is committed to the board DB (so the
+    # hook always sees durable state and a slow plugin can never hold the
+    # SQLite write lock). Observers only: return values are ignored.
+    #
+    # WHICH PROCESS each fires in matters, because kanban workers run as
+    # separate `hermes -p <profile> chat -q` subprocesses:
+    #   - kanban_task_claimed   -> the DISPATCHER process (gateway-embedded
+    #                              dispatcher or `hermes kanban dispatch`),
+    #                              right before the worker subprocess spawns.
+    #   - kanban_task_completed -> the WORKER process, when it calls
+    #                              kanban_complete (or a CLI/manual complete).
+    #   - kanban_task_blocked   -> the WORKER process (worker-initiated block)
+    #                              or whichever process drove the block.
+    # A plugin that needs to observe every transition centrally should hook in
+    # the dispatcher; one that needs per-task in-session context should hook in
+    # the worker.
+    #
+    # Common kwargs: task_id: str, board: str | None, assignee: str | None,
+    #   run_id: int | None, profile_name: str.
+    # kanban_task_completed adds: summary: str | None.
+    # kanban_task_blocked adds:   reason: str | None.
+    "kanban_task_claimed",
+    "kanban_task_completed",
+    "kanban_task_blocked",
 }
 
 ENTRY_POINTS_GROUP = "hermes_agent.plugins"

tests/hermes_cli/test_kanban_lifecycle_hooks.py

@@ -0,0 +1,135 @@
+"""Tests for kanban lifecycle plugin hooks.
+
+Verifies that claim/complete/block transitions fire the
+kanban_task_claimed / kanban_task_completed / kanban_task_blocked plugin
+hooks AFTER the board DB change is committed, with the documented kwargs,
+and that a misbehaving hook callback never breaks the transition.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import kanban_db as kb
+from hermes_cli.plugins import VALID_HOOKS, get_plugin_manager
+
+
+@pytest.fixture
+def kanban_home(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    kb.init_db()
+    return home
+
+
+@pytest.fixture
+def captured_hooks(monkeypatch):
+    """Register capturing callbacks for the three kanban lifecycle hooks.
+
+    Patches the plugin manager's _hooks dict directly (the same registry
+    invoke_hook reads) and restores it afterward.
+    """
+    mgr = get_plugin_manager()
+    events: list[tuple[str, dict]] = []
+    saved = {k: list(v) for k, v in mgr._hooks.items()}
+    for hook in ("kanban_task_claimed", "kanban_task_completed", "kanban_task_blocked"):
+        mgr._hooks.setdefault(hook, []).append(
+            lambda _h=hook, **kw: events.append((_h, kw))
+        )
+    try:
+        yield events
+    finally:
+        mgr._hooks = saved
+
+
+def test_hooks_are_registered_as_valid():
+    """The three lifecycle hook names are part of VALID_HOOKS."""
+    assert "kanban_task_claimed" in VALID_HOOKS
+    assert "kanban_task_completed" in VALID_HOOKS
+    assert "kanban_task_blocked" in VALID_HOOKS
+
+
+def test_claim_fires_hook(kanban_home, captured_hooks):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="t", assignee="worker")
+        claimed = kb.claim_task(conn, tid)
+        assert claimed is not None
+    finally:
+        conn.close()
+    fired = [e for e in captured_hooks if e[0] == "kanban_task_claimed"]
+    assert len(fired) == 1
+    kw = fired[0][1]
+    assert kw["task_id"] == tid
+    assert kw["assignee"] == "worker"
+    assert "profile_name" in kw
+    assert kw["run_id"] is not None
+
+
+def test_complete_fires_hook_with_summary(kanban_home, captured_hooks):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="t", assignee="worker")
+        kb.claim_task(conn, tid)
+        assert kb.complete_task(conn, tid, summary="all done")
+    finally:
+        conn.close()
+    fired = [e for e in captured_hooks if e[0] == "kanban_task_completed"]
+    assert len(fired) == 1
+    kw = fired[0][1]
+    assert kw["task_id"] == tid
+    assert kw["summary"] == "all done"
+    assert kw["assignee"] == "worker"
+
+
+def test_block_fires_hook_with_reason(kanban_home, captured_hooks):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="t", assignee="worker")
+        kb.claim_task(conn, tid)
+        assert kb.block_task(conn, tid, reason="needs human")
+    finally:
+        conn.close()
+    fired = [e for e in captured_hooks if e[0] == "kanban_task_blocked"]
+    assert len(fired) == 1
+    kw = fired[0][1]
+    assert kw["task_id"] == tid
+    assert kw["reason"] == "needs human"
+
+
+def test_no_hook_on_failed_transition(kanban_home, captured_hooks):
+    """complete_task on an unclaimed/nonexistent task fires no hook."""
+    conn = kb.connect()
+    try:
+        # Completing a task that doesn't exist returns False without firing.
+        assert kb.complete_task(conn, "t_doesnotexist", summary="x") is False
+    finally:
+        conn.close()
+    assert [e for e in captured_hooks if e[0] == "kanban_task_completed"] == []
+
+
+def test_misbehaving_hook_does_not_break_transition(kanban_home, monkeypatch):
+    """A hook callback that raises must not break the board transition."""
+    mgr = get_plugin_manager()
+    saved = {k: list(v) for k, v in mgr._hooks.items()}
+
+    def _boom(**kw):
+        raise RuntimeError("plugin exploded")
+
+    mgr._hooks.setdefault("kanban_task_completed", []).append(_boom)
+    try:
+        conn = kb.connect()
+        try:
+            tid = kb.create_task(conn, title="t", assignee="worker")
+            kb.claim_task(conn, tid)
+            # Despite the raising hook, completion succeeds and persists.
+            assert kb.complete_task(conn, tid, summary="ok") is True
+            assert kb.get_task(conn, tid).status == "done"
+        finally:
+            conn.close()
+    finally:
+        mgr._hooks = saved