diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index cdc3e464fd655..3e7250dc3f063 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -210,6 +210,154 @@ def _fire_kanban_lifecycle_hook(event: str, task_id: str, **fields: Any) -> None _log.debug("kanban lifecycle hook %s failed: %s", event, exc) +def _kanban_observer_consumed(event: str) -> bool: + """Return whether any first-party observer or plugin consumes *event*. + + Hot-path short-circuit for the worker-lifecycle / task-mutation / + dispatch-tick observers (RFC #58548): those fire on every dispatcher + tick and every task write, so call sites skip payload assembly entirely + when nothing subscribes. Best-effort — if inspection fails the event is + treated as unconsumed (the invoke path would fail the same way, and + these are observers, so dropping is always safe). + """ + try: + from hermes_cli.lifecycle import has_hook + + return has_hook(event) + except Exception: # pragma: no cover - defensive + return False + + +def _fire_worker_spawned_hook( + conn: sqlite3.Connection, + task: "Task", + workspace_path: str, + pid: Optional[int], + *, + board: Optional[str] = None, +) -> None: + """Fire ``on_kanban_worker_spawned`` for one dispatched spawn. + + Called by the dispatch loop AFTER ``spawn_fn`` returned and the worker + PID (when one was reported) has been durably persisted — the RFC #58548 + timing contract. Fully best-effort: any failure is swallowed so a + misbehaving observer can never break the dispatch loop. + """ + if not _kanban_observer_consumed("on_kanban_worker_spawned"): + return + try: + _fire_kanban_lifecycle_hook( + "on_kanban_worker_spawned", + task.id, + board=board or get_current_board(), + assignee=task.assignee, + run_id=_current_run_id(conn, task.id), + worker_pid=int(pid) if pid else None, + workspace_path=str(workspace_path), + ) + except Exception as exc: # pragma: no cover - defensive + _log.debug("kanban worker spawned hook failed: %s", exc) + + +def notify_task_updated( + conn: sqlite3.Connection, + task_id: str, + changed_fields: Iterable[str], + *, + board: Optional[str] = None, +) -> None: + """Fire ``on_kanban_task_updated`` for a committed task-row mutation. + + Task-mutation boundary primitive from RFC #58548: a surface that mutates + a task row outside the claim/complete/block lifecycle calls this AFTER + its write txn has committed — including surfaces that write with direct + SQL and bypass every ``kanban_db`` mutator (the dashboard plugin API's + priority/title/body editors). ``changed_fields`` carries field NAMES + only, never values. Observer-only and fully best-effort: it can never + fail a task mutation, and it costs one ``has_hook`` probe when nothing + subscribes. + """ + if not _kanban_observer_consumed("on_kanban_task_updated"): + return + try: + row = conn.execute( + "SELECT assignee, current_run_id FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + _fire_kanban_lifecycle_hook( + "on_kanban_task_updated", + task_id, + board=board or get_current_board(), + assignee=row["assignee"] if row else None, + run_id=row["current_run_id"] if row else None, + changed_fields=list(changed_fields), + ) + except Exception as exc: # pragma: no cover - defensive + _log.debug("kanban task updated hook failed: %s", exc) + + +def _fire_dispatch_tick_hook( + result: "DispatchResult", + *, + board: Optional[str] = None, + dry_run: bool = False, +) -> None: + """Fire ``on_kanban_dispatch_tick`` after one dispatcher tick. + + Re-port of PR #56066 per the #64231 batch disposition: renamed to the + taxonomy form and called by ``dispatch_once`` strictly AFTER + ``_dispatch_tick_lock`` has been released — the original fired inside + the lock, so a slow subscriber could extend the single-writer critical + section and stall a sibling dispatcher's tick. Observer-only and fully + best-effort: any subscriber failure is swallowed. + """ + if not _kanban_observer_consumed("on_kanban_dispatch_tick"): + return + try: + from hermes_cli.lifecycle import invoke_hook + from hermes_cli.profiles import get_active_profile_name + + try: + profile_name = get_active_profile_name() + except Exception: + profile_name = "default" + if board is None: + try: + board = get_current_board() + except Exception: + board = None + outcome = "ok" + if result.skipped_locked: + outcome = "skipped_locked" + elif not any(( + result.spawned, + result.reclaimed, + result.promoted, + result.reconciled_orphans, + result.crashed, + result.stale, + result.timed_out, + result.auto_blocked, + result.rate_limited, + result.auto_assigned_default, + result.respawn_guarded, + result.skipped_per_profile_capped, + result.skipped_unassigned, + result.skipped_nonspawnable, + )): + outcome = "idle" + invoke_hook( + "on_kanban_dispatch_tick", + board=board, + profile_name=profile_name, + dry_run=bool(dry_run), + outcome=outcome, + result=result, + ) + except Exception as exc: # pragma: no cover - defensive + _log.debug("kanban dispatch tick hook failed: %s", 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 @@ -3490,7 +3638,10 @@ def assign_task(conn: sqlite3.Connection, task_id: str, profile: Optional[str]) else: conn.execute("UPDATE tasks SET assignee = ? WHERE id = ?", (profile, task_id)) _append_event(conn, task_id, "assigned", {"assignee": profile}) - return True + # Task-mutation observer (RFC #58548), fired AFTER the assignment txn + # has committed so subscribers always observe durable board state. + notify_task_updated(conn, task_id, ("assignee",)) + return True def set_model_override( @@ -3535,7 +3686,9 @@ def set_model_override( conn, task_id, "model_override_set", {"model": model, "provider": provider}, ) - return True + # Task-mutation observer (RFC #58548), fired AFTER the txn commits. + notify_task_updated(conn, task_id, ("model_override", "provider_override")) + return True def set_reasoning_effort( @@ -3573,7 +3726,9 @@ def set_reasoning_effort( _append_event( conn, task_id, "reasoning_effort_set", {"reasoning_effort": effort} ) - return True + # Task-mutation observer (RFC #58548), fired AFTER the txn commits. + notify_task_updated(conn, task_id, ("reasoning_effort",)) + return True # --------------------------------------------------------------------------- @@ -4732,7 +4887,8 @@ def release_stale_claims( reclaimed = 0 host_prefix = f"{_claimer_id().split(':', 1)[0]}:" stale = conn.execute( - "SELECT id, claim_lock, worker_pid, claim_expires, last_heartbeat_at " + "SELECT id, claim_lock, worker_pid, claim_expires, last_heartbeat_at, " + " assignee " "FROM tasks " "WHERE status = 'running' AND claim_expires IS NOT NULL " " AND claim_expires < ?", @@ -4843,6 +4999,24 @@ def release_stale_claims( run_id=run_id, ) reclaimed += 1 + # Worker-lifecycle observer (RFC #58548): the reclaim txn above has + # committed. The ``continue`` branches (rowcount mismatch, claim + # extension, deferred reclaim) never reach this point, so only a + # genuinely reclaimed stale claim fires. + if _kanban_observer_consumed("on_kanban_worker_stale_claim"): + _fire_kanban_lifecycle_hook( + "on_kanban_worker_stale_claim", + row["id"], + board=get_current_board(), + assignee=row["assignee"], + run_id=run_id, + worker_pid=( + int(row["worker_pid"]) + if row["worker_pid"] is not None else None + ), + heartbeat_stale=bool(heartbeat_stale), + retry_status=retry_status, + ) return reclaimed @@ -8531,9 +8705,13 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: # counter (see the post-txn loop below). crash_details: list[tuple[str, int, str, bool, str]] = [] # (task_id, pid, claimer, protocol_violation, error_text) + # Worker-exit observer payloads (RFC #58548), collected inside the main + # txn and fired only after every reclaim/accounting txn has committed. + exited_hook_payloads: list[dict] = [] with write_txn(conn): rows = conn.execute( - "SELECT id, worker_pid, claim_lock, started_at FROM tasks " + "SELECT id, worker_pid, claim_lock, started_at, assignee " + "FROM tasks " "WHERE status = 'running' AND worker_pid IS NOT NULL" ).fetchall() host_prefix = f"{_claimer_id().split(':', 1)[0]}:" @@ -8642,6 +8820,16 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: event_payload, run_id=run_id, ) + exited_hook_payloads.append({ + "task_id": row["id"], + "assignee": row["assignee"], + "run_id": run_id, + "worker_pid": pid, + "exit_kind": kind, + "exit_code": code, + "outcome": _run_outcome, + "retry_status": retry_status, + }) if rate_limited_exit: # Stamp the failure-error column so ``check_respawn_guard`` # recognizes this as a quota blocker and defers the @@ -8762,6 +8950,20 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: # Same side-channel for rate-limited requeues — these did NOT count a # failure and are NOT crashes, so they stay out of the ``crashed`` return. detect_crashed_workers._last_rate_limited = rate_limited # type: ignore[attr-defined] + # Worker-lifecycle observer (RFC #58548): exit events are tick-derived + # from this reclaim pass — fired only now, after the main reclaim txn + # AND the breaker accounting above have committed, so subscribers always + # observe fully durable board state. + if exited_hook_payloads and _kanban_observer_consumed("on_kanban_worker_exited"): + _board = get_current_board() + for hook_fields in exited_hook_payloads: + hook_fields = dict(hook_fields) + _fire_kanban_lifecycle_hook( + "on_kanban_worker_exited", + hook_fields.pop("task_id"), + board=_board, + **hook_fields, + ) return crashed @@ -9262,23 +9464,6 @@ def dispatch_once( # Path resolution should never fail, but if it somehow does we # must not lose the tick — fall through to an unguarded dispatch # rather than dropping work. - return _dispatch_once_locked( - conn, - spawn_fn=spawn_fn, - ttl_seconds=ttl_seconds, - dry_run=dry_run, - max_spawn=max_spawn, - max_in_progress=max_in_progress, - failure_limit=failure_limit, - stale_timeout_seconds=stale_timeout_seconds, - board=board, - default_assignee=default_assignee, - max_in_progress_per_profile=max_in_progress_per_profile, - reconcile_orphans=reconcile_orphans, - ) - with _dispatch_tick_lock(db_path) as held: - if not held: - return DispatchResult(skipped_locked=True) result = _dispatch_once_locked( conn, spawn_fn=spawn_fn, @@ -9293,10 +9478,36 @@ def dispatch_once( max_in_progress_per_profile=max_in_progress_per_profile, reconcile_orphans=reconcile_orphans, ) - # Still under the dispatch lock: opportunistically truncate the WAL - # at a coarse interval so it cannot grow unbounded between restarts. - _maybe_checkpoint_wal(conn, db_path) + _fire_dispatch_tick_hook(result, board=board, dry_run=dry_run) return result + with _dispatch_tick_lock(db_path) as held: + if not held: + result = DispatchResult(skipped_locked=True) + else: + result = _dispatch_once_locked( + conn, + spawn_fn=spawn_fn, + ttl_seconds=ttl_seconds, + dry_run=dry_run, + max_spawn=max_spawn, + max_in_progress=max_in_progress, + failure_limit=failure_limit, + stale_timeout_seconds=stale_timeout_seconds, + board=board, + default_assignee=default_assignee, + max_in_progress_per_profile=max_in_progress_per_profile, + reconcile_orphans=reconcile_orphans, + ) + # Still under the dispatch lock: run the periodic PASSIVE WAL + # checkpoint (see _maybe_checkpoint_wal; the -wal file size is + # bounded by journal_size_limit on the writer's natural reset). + _maybe_checkpoint_wal(conn, db_path) + # The dispatch lock has been released here. Fire the tick observer + # strictly OUTSIDE the single-writer critical section (#56066 sweeper + # finding / #64231 disposition): a slow subscriber must never extend + # the lock hold and stall a sibling dispatcher's tick. + _fire_dispatch_tick_hook(result, board=board, dry_run=dry_run) + return result def _dispatch_once_locked( @@ -9595,6 +9806,13 @@ def _dispatch_once_locked( pid = _spawn(claimed, str(workspace)) if pid: _set_worker_pid(conn, claimed.id, int(pid)) + # Worker-lifecycle observer (RFC #58548): fires AFTER spawn_fn + # returned and the PID (when reported) is durably persisted, + # per the RFC timing contract. Best-effort — can never break + # the dispatch loop. + _fire_worker_spawned_hook( + conn, claimed, str(workspace), pid, board=board, + ) # NOTE: we intentionally do NOT reset consecutive_failures # here. A successful spawn proves the worker can start but # doesn't prove the run will succeed. Under unified @@ -9720,6 +9938,11 @@ def _dispatch_once_locked( pid = _spawn(claimed, str(workspace)) if pid: _set_worker_pid(conn, claimed.id, int(pid)) + # Worker-lifecycle observer (RFC #58548): same contract as the + # ready-lane fire above — after spawn + PID persistence. + _fire_worker_spawned_hook( + conn, claimed, str(workspace), pid, board=board, + ) result.spawned.append((claimed.id, claimed.assignee or "", str(workspace))) spawned += 1 if _per_profile_cap is not None and claimed.assignee: diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 947a6e990e71c..7f5070ee2a2f3 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -253,6 +253,73 @@ VALID_HOOKS: Set[str] = { "kanban_task_claimed", "kanban_task_completed", "kanban_task_blocked", + # Kanban worker-lifecycle, task-mutation, and dispatcher-tick observers + # (RFC #58548, accepted as the design basis in the #64231 batch + # disposition; on_kanban_dispatch_tick is the re-port of PR #56066). + # All five are observers only: return values are ignored, and every fire + # site is fully best-effort, so a broken callback can never break + # dispatch or a task mutation. Cost rule: every call site short-circuits + # on has_hook(), so when nothing subscribes no payload is built and the + # hot paths (each dispatcher tick, each task write) pay one dict probe. + # + # WHICH PROCESS: worker spawn/exit/stale-claim and the dispatch tick + # fire in the DISPATCHER process (gateway-embedded dispatcher or + # ``hermes kanban dispatch``); on_kanban_task_updated fires in whichever + # process committed the mutation (CLI, worker, or the gateway-embedded + # dashboard API). + # + # Common kwargs (task-scoped hooks): task_id: str, profile_name: str, + # board: str | None, assignee: str | None, run_id: int | None. + # + # on_kanban_worker_spawned fires after ``spawn_fn`` returns AND the + # worker PID (when one was reported) is durably persisted, per the RFC + # timing contract; like kanban_task_claimed it runs inside the board's + # dispatch lock, so callbacks must stay fast. Adds: + # worker_pid: int | None, workspace_path: str. + # Privacy: workspace_path is a filesystem path and may reveal project + # layout or usernames. + "on_kanban_worker_spawned", + # on_kanban_worker_exited is tick-derived from detect_crashed_workers — + # it fires when a dead-PID running task is reclaimed, AFTER every + # reclaim/accounting txn has committed. Exit visibility latency is + # bounded by the dispatcher tick interval. Adds: + # worker_pid: int, + # exit_kind: "clean_exit" | "rate_limited" | "nonzero_exit" + # | "signaled" | "unknown", + # exit_code: int | None, + # outcome: "crashed" | "rate_limited", + # retry_status: str (the phase the task was released back to). + "on_kanban_worker_exited", + # on_kanban_worker_stale_claim fires when release_stale_claims reclaims + # a TTL-expired claim, after the reclaim txn commits. Live-PID claim + # extensions and deferred reclaims do NOT fire. Adds: + # worker_pid: int | None, heartbeat_stale: bool, retry_status: str. + "on_kanban_worker_stale_claim", + # on_kanban_task_updated is the task-mutation boundary observer: it + # fires after a committed task-row field write outside the + # claim/complete/block lifecycle — kanban_db.assign_task, + # set_model_override, and set_reasoning_effort, plus the dashboard + # plugin API's direct-SQL priority/title/body editors (single and + # bulk) via kanban_db.notify_task_updated. Adds: + # changed_fields: list[str] — field NAMES only; new values are never + # carried (fetch the task if you need them). + # Privacy: names only here, but title/body values in the board DB may + # contain user/project content. + "on_kanban_task_updated", + # on_kanban_dispatch_tick fires once per dispatcher tick in + # dispatch_once, strictly AFTER the board's single-writer dispatch lock + # has been released (the #56066 original fired inside the lock — the + # #64231 disposition mandates the post-lock re-port), so a slow + # subscriber can never extend the writer critical section. + # Kwargs: board: str | None, profile_name: str, dry_run: bool, + # outcome: "ok" | "skipped_locked" | "idle", + # result: hermes_cli.kanban_db.DispatchResult (spawned, reclaimed, + # promoted, reconciled_orphans, crashed, stale, timed_out, + # auto_blocked, rate_limited, auto_assigned_default, + # respawn_guarded, skipped_per_profile_capped, skipped_unassigned, + # skipped_nonspawnable, skipped_locked). + # Privacy: result carries task ids, assignees, and workspace paths. + "on_kanban_dispatch_tick", # Gateway platform-boundary observer hooks (#64176). Observer-only; each # callback isolated by invoke_hook. Payloads are normalized envelopes only, # never raw platform SDK objects (per #64176 / #64182 ground rule). This diff --git a/plugins/kanban/dashboard/plugin_api.py b/plugins/kanban/dashboard/plugin_api.py index d4548261ebeef..47077d939024a 100644 --- a/plugins/kanban/dashboard/plugin_api.py +++ b/plugins/kanban/dashboard/plugin_api.py @@ -1013,6 +1013,12 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu (task_id, json.dumps({"priority": int(payload.priority)}), int(time.time())), ) + # Mutation-boundary observer (RFC #58548): this direct-SQL write + # bypasses every kanban_db mutator, so report it here — after + # the txn commits. + kanban_db.notify_task_updated( + conn, task_id, ("priority",), board=board, + ) # --- title / body ------------------------------------------------- if payload.title is not None or payload.body is not None: @@ -1035,6 +1041,13 @@ def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Qu "VALUES (?, 'edited', NULL, ?)", (task_id, int(time.time())), ) + # Mutation-boundary observer (RFC #58548), post-commit. Field + # names only — values never leave the DB via this payload. + kanban_db.notify_task_updated( + conn, task_id, + [f for f in ("title", "body") if getattr(payload, f) is not None], + board=board, + ) updated = kanban_db.get_task(conn, task_id) return {"task": _task_dict(updated) if updated else None} @@ -1402,6 +1415,12 @@ def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)): (tid, json.dumps({"priority": int(payload.priority)}), int(time.time())), ) + # Mutation-boundary observer (RFC #58548): the bulk + # editor writes with direct SQL too — report each task's + # committed write. + kanban_db.notify_task_updated( + conn, tid, ("priority",), board=board, + ) if payload.clear_model_override or payload.model_override is not None: new_model = ( None if payload.clear_model_override diff --git a/tests/hermes_cli/test_kanban_dispatch_tick_hook.py b/tests/hermes_cli/test_kanban_dispatch_tick_hook.py new file mode 100644 index 0000000000000..d1c8b6cf7e129 --- /dev/null +++ b/tests/hermes_cli/test_kanban_dispatch_tick_hook.py @@ -0,0 +1,178 @@ +"""Tests for the ``on_kanban_dispatch_tick`` observer hook. + +Re-port of PR #56066 per the #64231 batch disposition: renamed to the +taxonomy form and fired by ``kanban_db.dispatch_once`` strictly AFTER the +board's single-writer dispatch lock has been released — the original fired +inside ``_dispatch_tick_lock``, so a slow subscriber could extend the +critical section and stall a sibling dispatcher. Verifies the post-lock +contract directly, the ``outcome`` classification (including idle and +contended ticks), ``dry_run`` propagation, and that a misbehaving +subscriber never breaks the dispatcher. +""" + +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_ticks(monkeypatch): + """Register a capturing callback for the dispatch tick hook.""" + mgr = get_plugin_manager() + events: list[dict] = [] + saved = {k: list(v) for k, v in mgr._hooks.items()} + mgr._hooks.setdefault("on_kanban_dispatch_tick", []).append( + lambda **kw: events.append(kw) + ) + try: + yield events + finally: + mgr._hooks = saved + + +def test_dispatch_tick_hook_is_registered_as_valid(): + assert "on_kanban_dispatch_tick" in VALID_HOOKS + + +def test_idle_tick_fires_hook_with_outcome_idle(kanban_home, captured_ticks): + """An empty board produces one hook fire with outcome='idle'.""" + conn = kb.connect() + try: + kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 12345) + finally: + conn.close() + assert len(captured_ticks) == 1 + kw = captured_ticks[0] + assert kw["outcome"] == "idle" + assert kw["dry_run"] is False + assert isinstance(kw["result"], kb.DispatchResult) + assert "board" in kw + assert "profile_name" in kw + + +def test_active_tick_fires_hook_with_outcome_ok( + kanban_home, all_assignees_spawnable, captured_ticks, +): + """A tick that spawns a worker fires the hook with outcome='ok'.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 4242) + finally: + conn.close() + ok_events = [kw for kw in captured_ticks if kw["outcome"] == "ok"] + assert ok_events, ( + f"expected an ok tick, got {[kw['outcome'] for kw in captured_ticks]}" + ) + result = ok_events[-1]["result"] + assert any(row[0] == tid for row in result.spawned) + + +def test_dry_run_tick_carries_dry_run_flag(kanban_home, captured_ticks): + """dry_run=True is propagated to the hook payload.""" + conn = kb.connect() + try: + kb.dispatch_once(conn, dry_run=True, spawn_fn=lambda *a, **k: None) + finally: + conn.close() + assert captured_ticks + assert all(kw["dry_run"] is True for kw in captured_ticks) + + +def test_contended_tick_fires_with_outcome_skipped_locked( + kanban_home, captured_ticks, +): + """A losing dispatcher still reports its (empty) tick to observers.""" + db_path = kb.kanban_db_path() + conn = kb.connect() + try: + with kb._dispatch_tick_lock(db_path) as held: + assert held is True + result = kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 1) + assert result.skipped_locked is True + finally: + conn.close() + assert len(captured_ticks) == 1 + assert captured_ticks[0]["outcome"] == "skipped_locked" + + +def test_tick_hook_fires_after_dispatch_lock_released(kanban_home): + """The #56066 sweeper finding, as a contract: subscribers run OUTSIDE + the single-writer critical section. From the callback, acquiring the + board's dispatch lock must succeed — flock conflicts across file + descriptors within one process, so this fails if the hook still fires + while ``dispatch_once`` holds the lock.""" + db_path = kb.kanban_db_path() + mgr = get_plugin_manager() + saved = {k: list(v) for k, v in mgr._hooks.items()} + acquired: list[bool] = [] + + def _probe_lock(**kw): + with kb._dispatch_tick_lock(db_path) as held: + acquired.append(held) + + mgr._hooks.setdefault("on_kanban_dispatch_tick", []).append(_probe_lock) + try: + conn = kb.connect() + try: + kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 1) + finally: + conn.close() + finally: + mgr._hooks = saved + assert acquired == [True] + + +def test_misbehaving_subscriber_does_not_break_dispatcher(kanban_home): + """A hook callback that raises must not break the dispatch tick.""" + mgr = get_plugin_manager() + saved = {k: list(v) for k, v in mgr._hooks.items()} + + def _boom(**kw): + raise RuntimeError("subscriber exploded") + + mgr._hooks.setdefault("on_kanban_dispatch_tick", []).append(_boom) + try: + conn = kb.connect() + try: + result = kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 1) + assert isinstance(result, kb.DispatchResult) + finally: + conn.close() + finally: + mgr._hooks = saved + + +def test_no_subscriber_short_circuits_tick_hook(kanban_home, monkeypatch): + """With nothing registered, the tick observer is never invoked.""" + from hermes_cli import lifecycle + + invoked: list[str] = [] + real_invoke = lifecycle.invoke_hook + + def _spy(hook_name, **kw): + invoked.append(hook_name) + return real_invoke(hook_name, **kw) + + monkeypatch.setattr(lifecycle, "invoke_hook", _spy) + conn = kb.connect() + try: + kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 1) + finally: + conn.close() + assert "on_kanban_dispatch_tick" not in invoked diff --git a/tests/hermes_cli/test_kanban_task_updated_hook.py b/tests/hermes_cli/test_kanban_task_updated_hook.py new file mode 100644 index 0000000000000..a20d4be05de4a --- /dev/null +++ b/tests/hermes_cli/test_kanban_task_updated_hook.py @@ -0,0 +1,181 @@ +"""Tests for the ``on_kanban_task_updated`` mutation observer (RFC #58548). + +Verifies the task-mutation boundary observer: ``assign_task`` fires it with +``changed_fields`` AFTER the assignment txn commits, refused/failed +mutations never fire, a raising callback never breaks the mutation, and +call sites short-circuit when nothing subscribes. +""" + +from __future__ import annotations + +import sqlite3 +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_updates(monkeypatch): + mgr = get_plugin_manager() + events: list[dict] = [] + saved = {k: list(v) for k, v in mgr._hooks.items()} + mgr._hooks.setdefault("on_kanban_task_updated", []).append( + lambda **kw: events.append(kw) + ) + try: + yield events + finally: + mgr._hooks = saved + + +def test_task_updated_hook_is_registered_as_valid(): + assert "on_kanban_task_updated" in VALID_HOOKS + + +def test_assign_fires_updated_with_changed_fields(kanban_home, captured_updates): + """assign_task fires the observer post-commit with the changed field.""" + assignee_at_fire_time: list = [] + + def _read_assignee(**kw): + # Fresh connection: proves the assignment was committed before + # the hook fired. + c2 = sqlite3.connect(kb.kanban_db_path()) + try: + row = c2.execute( + "SELECT assignee FROM tasks WHERE id = ?", (kw["task_id"],) + ).fetchone() + assignee_at_fire_time.append(row[0] if row else None) + finally: + c2.close() + + mgr = get_plugin_manager() + mgr._hooks.setdefault("on_kanban_task_updated", []).append(_read_assignee) + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + captured_updates.clear() # create-time bookkeeping is not under test + assert kb.assign_task(conn, tid, "bob") is True + finally: + conn.close() + + assert len(captured_updates) == 1 + kw = captured_updates[0] + assert kw["task_id"] == tid + assert kw["changed_fields"] == ["assignee"] + assert kw["assignee"] == "bob" + assert "profile_name" in kw + assert "board" in kw + assert "run_id" in kw + assert assignee_at_fire_time == ["bob"] + + +def test_set_model_override_fires_with_changed_fields( + kanban_home, captured_updates, +): + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + captured_updates.clear() + assert kb.set_model_override( + conn, tid, "some-model", provider="some-provider", + ) is True + finally: + conn.close() + assert len(captured_updates) == 1 + kw = captured_updates[0] + assert kw["task_id"] == tid + assert kw["changed_fields"] == ["model_override", "provider_override"] + + +def test_set_reasoning_effort_fires_with_changed_fields( + kanban_home, captured_updates, +): + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + captured_updates.clear() + assert kb.set_reasoning_effort(conn, tid, "high") is True + finally: + conn.close() + assert len(captured_updates) == 1 + kw = captured_updates[0] + assert kw["task_id"] == tid + assert kw["changed_fields"] == ["reasoning_effort"] + + +def test_assign_missing_task_does_not_fire(kanban_home, captured_updates): + conn = kb.connect() + try: + assert kb.assign_task(conn, "kb-nope", "bob") is False + finally: + conn.close() + assert captured_updates == [] + + +def test_assign_running_task_raises_and_does_not_fire( + kanban_home, captured_updates, +): + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + kb.claim_task(conn, tid) + captured_updates.clear() + with pytest.raises(RuntimeError): + kb.assign_task(conn, tid, "bob") + finally: + conn.close() + assert captured_updates == [] + + +def test_raising_callback_does_not_break_assign(kanban_home): + 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("on_kanban_task_updated", []).append(_boom) + try: + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + assert kb.assign_task(conn, tid, "bob") is True + assert kb.get_task(conn, tid).assignee == "bob" + finally: + conn.close() + finally: + mgr._hooks = saved + + +def test_no_subscriber_short_circuits_task_updated(kanban_home, monkeypatch): + from hermes_cli import lifecycle + + invoked: list[str] = [] + real_invoke = lifecycle.invoke_hook + + def _spy(hook_name, **kw): + invoked.append(hook_name) + return real_invoke(hook_name, **kw) + + monkeypatch.setattr(lifecycle, "invoke_hook", _spy) + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + assert kb.assign_task(conn, tid, "bob") is True + finally: + conn.close() + assert "on_kanban_task_updated" not in invoked diff --git a/tests/hermes_cli/test_kanban_worker_lifecycle_hooks.py b/tests/hermes_cli/test_kanban_worker_lifecycle_hooks.py new file mode 100644 index 0000000000000..5c277554dfd5b --- /dev/null +++ b/tests/hermes_cli/test_kanban_worker_lifecycle_hooks.py @@ -0,0 +1,283 @@ +"""Tests for the ``on_kanban_worker_*`` observer hooks (RFC #58548). + +Verifies the worker-lifecycle observers accepted in the #64231 batch +disposition: ``on_kanban_worker_spawned`` fires after ``spawn_fn`` returns +and the worker PID is durably persisted, ``on_kanban_worker_exited`` is +tick-derived from ``detect_crashed_workers`` and fires after every reclaim +transaction has committed, and ``on_kanban_worker_stale_claim`` fires when +``release_stale_claims`` reclaims a TTL-expired claim. All three are +observer-only, short-circuit on ``has_hook``, and can never break the +dispatcher. +""" + +from __future__ import annotations + +import sqlite3 +import time +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb +from hermes_cli.plugins import VALID_HOOKS, get_plugin_manager + +WORKER_HOOKS = ( + "on_kanban_worker_spawned", + "on_kanban_worker_exited", + "on_kanban_worker_stale_claim", +) + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + # Crash detection acts immediately in these tests (no launch grace). + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0") + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +@pytest.fixture +def captured_hooks(monkeypatch): + """Register capturing callbacks for the worker-lifecycle hooks.""" + mgr = get_plugin_manager() + events: list[tuple[str, dict]] = [] + saved = {k: list(v) for k, v in mgr._hooks.items()} + for hook in WORKER_HOOKS: + mgr._hooks.setdefault(hook, []).append( + lambda _h=hook, **kw: events.append((_h, kw)) + ) + try: + yield events + finally: + mgr._hooks = saved + + +def test_worker_lifecycle_hooks_are_registered_as_valid(): + for hook in WORKER_HOOKS: + assert hook in VALID_HOOKS, hook + + +def test_dispatch_spawn_fires_worker_spawned( + kanban_home, all_assignees_spawnable, captured_hooks, +): + """A dispatched spawn fires the hook AFTER the PID is durably persisted.""" + pid_at_fire_time: list = [] + + def _read_pid(**kw): + # Read through a FRESH connection: proves the PID write was + # committed before the hook fired (the RFC timing contract). + c2 = sqlite3.connect(kb.kanban_db_path()) + try: + row = c2.execute( + "SELECT worker_pid FROM tasks WHERE id = ?", (kw["task_id"],) + ).fetchone() + pid_at_fire_time.append(row[0] if row else None) + finally: + c2.close() + + mgr = get_plugin_manager() + mgr._hooks.setdefault("on_kanban_worker_spawned", []).append(_read_pid) + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + result = kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 4242) + assert any(row[0] == tid for row in result.spawned) + finally: + conn.close() + + fired = [e for e in captured_hooks if e[0] == "on_kanban_worker_spawned"] + assert len(fired) == 1 + kw = fired[0][1] + assert kw["task_id"] == tid + assert kw["assignee"] == "alice" + assert kw["worker_pid"] == 4242 + assert kw["workspace_path"] + assert kw["run_id"] is not None + assert "profile_name" in kw + assert "board" in kw + assert pid_at_fire_time == [4242] + + +def test_spawn_without_pid_fires_with_worker_pid_none( + kanban_home, all_assignees_spawnable, captured_hooks, +): + """A spawn_fn that reports no PID still fires, with worker_pid=None.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + kb.dispatch_once(conn, spawn_fn=lambda *a, **k: None) + finally: + conn.close() + fired = [e for e in captured_hooks if e[0] == "on_kanban_worker_spawned"] + assert len(fired) == 1 + kw = fired[0][1] + assert kw["task_id"] == tid + assert kw["worker_pid"] is None + + +def test_crash_reclaim_fires_worker_exited(kanban_home, captured_hooks, monkeypatch): + """A dead-PID reclaim fires the exit observer with the exit facts.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="worker") + kb.claim_task(conn, tid) + kb._set_worker_pid(conn, tid, 98765) + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + assert kb.detect_crashed_workers(conn) == [tid] + finally: + conn.close() + + fired = [e for e in captured_hooks if e[0] == "on_kanban_worker_exited"] + assert len(fired) == 1 + kw = fired[0][1] + assert kw["task_id"] == tid + assert kw["assignee"] == "worker" + assert kw["worker_pid"] == 98765 + assert kw["exit_kind"] == "unknown" + assert kw["exit_code"] is None + assert kw["outcome"] == "crashed" + assert kw["retry_status"] == "ready" + assert kw["run_id"] is not None + assert "profile_name" in kw + assert "board" in kw + + +def test_rate_limited_exit_fires_worker_exited_with_outcome( + kanban_home, captured_hooks, monkeypatch, +): + """The EX_TEMPFAIL sentinel classifies as a rate_limited exit.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="worker") + kb.claim_task(conn, tid) + kb._set_worker_pid(conn, tid, 24680) + # Simulate the reap loop having recorded the sentinel exit status. + kb._record_worker_exit(24680, kb.KANBAN_RATE_LIMIT_EXIT_CODE << 8) + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + assert kb.detect_crashed_workers(conn) == [] + finally: + conn.close() + + fired = [e for e in captured_hooks if e[0] == "on_kanban_worker_exited"] + assert len(fired) == 1 + kw = fired[0][1] + assert kw["task_id"] == tid + assert kw["exit_kind"] == "rate_limited" + assert kw["exit_code"] == kb.KANBAN_RATE_LIMIT_EXIT_CODE + assert kw["outcome"] == "rate_limited" + + +def test_stale_claim_reclaim_fires_hook(kanban_home, captured_hooks): + """A TTL-expired reclaim fires the stale-claim observer post-commit.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="worker") + kb.claim_task(conn, tid) + conn.execute( + "UPDATE tasks SET claim_expires = ? WHERE id = ?", + (int(time.time()) - 100, tid), + ) + conn.commit() + assert kb.release_stale_claims(conn) == 1 + finally: + conn.close() + + fired = [e for e in captured_hooks if e[0] == "on_kanban_worker_stale_claim"] + assert len(fired) == 1 + kw = fired[0][1] + assert kw["task_id"] == tid + assert kw["assignee"] == "worker" + assert kw["worker_pid"] is None + assert kw["heartbeat_stale"] is False + assert kw["retry_status"] == "ready" + assert kw["run_id"] is not None + assert "profile_name" in kw + assert "board" in kw + + +def test_claim_extension_does_not_fire_stale_claim( + kanban_home, captured_hooks, monkeypatch, +): + """A live-PID claim extension is not a reclaim and must not fire.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="worker") + kb.claim_task(conn, tid) + kb._set_worker_pid(conn, tid, 13579) + monkeypatch.setattr(kb, "_pid_alive", lambda pid: True) + conn.execute( + "UPDATE tasks SET claim_expires = ? WHERE id = ?", + (int(time.time()) - 100, tid), + ) + conn.commit() + assert kb.release_stale_claims(conn) == 0 + finally: + conn.close() + fired = [e for e in captured_hooks if e[0] == "on_kanban_worker_stale_claim"] + assert fired == [] + + +def test_raising_callbacks_never_break_worker_lifecycle( + kanban_home, all_assignees_spawnable, monkeypatch, +): + """Raising subscribers must not break spawn, crash reclaim, or stale reclaim.""" + mgr = get_plugin_manager() + saved = {k: list(v) for k, v in mgr._hooks.items()} + + def _boom(**kw): + raise RuntimeError("plugin exploded") + + for hook in WORKER_HOOKS: + mgr._hooks.setdefault(hook, []).append(_boom) + try: + conn = kb.connect() + try: + tid = kb.create_task(conn, title="t", assignee="alice") + result = kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 111) + assert any(row[0] == tid for row in result.spawned) + + monkeypatch.setattr(kb, "_pid_alive", lambda pid: False) + assert kb.detect_crashed_workers(conn) == [tid] + + kb.claim_task(conn, tid) + conn.execute( + "UPDATE tasks SET claim_expires = ?, worker_pid = NULL " + "WHERE id = ?", + (int(time.time()) - 100, tid), + ) + conn.commit() + assert kb.release_stale_claims(conn) == 1 + finally: + conn.close() + finally: + mgr._hooks = saved + + +def test_no_subscriber_short_circuits_worker_hooks( + kanban_home, all_assignees_spawnable, monkeypatch, +): + """With nothing registered, the new observers are never invoked at all.""" + from hermes_cli import lifecycle + + invoked: list[str] = [] + real_invoke = lifecycle.invoke_hook + + def _spy(hook_name, **kw): + invoked.append(hook_name) + return real_invoke(hook_name, **kw) + + monkeypatch.setattr(lifecycle, "invoke_hook", _spy) + conn = kb.connect() + try: + kb.create_task(conn, title="t", assignee="alice") + kb.dispatch_once(conn, spawn_fn=lambda *a, **k: 222) + finally: + conn.close() + assert "on_kanban_worker_spawned" not in invoked + # The shipped claimed hook has no short-circuit and still fires. + assert "kanban_task_claimed" in invoked diff --git a/tests/plugins/test_kanban_dashboard_task_updated_hook.py b/tests/plugins/test_kanban_dashboard_task_updated_hook.py new file mode 100644 index 0000000000000..4f52f48d72891 --- /dev/null +++ b/tests/plugins/test_kanban_dashboard_task_updated_hook.py @@ -0,0 +1,139 @@ +"""Dashboard mutation-boundary coverage for ``on_kanban_task_updated``. + +The dashboard plugin API's priority/title/body editors write task rows with +direct SQL, bypassing every ``kanban_db`` mutator — the exact gap the RFC +#58548 mutation-boundary review called out. These tests verify each +direct-SQL write path (single-task PATCH and bulk POST) reports through +``kanban_db.notify_task_updated`` with the right ``changed_fields``. +""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from hermes_cli import kanban_db as kb +from hermes_cli.plugins import get_plugin_manager + + +def _load_plugin_router(): + repo_root = Path(__file__).resolve().parents[2] + plugin_file = repo_root / "plugins" / "kanban" / "dashboard" / "plugin_api.py" + assert plugin_file.exists(), f"plugin file missing: {plugin_file}" + spec = importlib.util.spec_from_file_location( + "hermes_dashboard_plugin_kanban_task_updated_test", plugin_file, + ) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod.router + + +@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 client(kanban_home): + app = FastAPI() + app.include_router(_load_plugin_router(), prefix="/api/plugins/kanban") + return TestClient(app) + + +@pytest.fixture +def captured_updates(): + mgr = get_plugin_manager() + events: list[dict] = [] + saved = {k: list(v) for k, v in mgr._hooks.items()} + mgr._hooks.setdefault("on_kanban_task_updated", []).append( + lambda **kw: events.append(kw) + ) + try: + yield events + finally: + mgr._hooks = saved + + +def _make_task(title="t"): + conn = kb.connect() + try: + return kb.create_task(conn, title=title, assignee="alice") + finally: + conn.close() + + +def test_patch_priority_fires_task_updated(client, captured_updates): + tid = _make_task() + captured_updates.clear() + r = client.patch(f"/api/plugins/kanban/tasks/{tid}", json={"priority": 5}) + assert r.status_code == 200 + assert len(captured_updates) == 1 + kw = captured_updates[0] + assert kw["task_id"] == tid + assert kw["changed_fields"] == ["priority"] + assert kw["board"] + + +def test_patch_title_and_body_fires_task_updated(client, captured_updates): + tid = _make_task() + captured_updates.clear() + r = client.patch( + f"/api/plugins/kanban/tasks/{tid}", + json={"title": "new title", "body": "new body"}, + ) + assert r.status_code == 200 + assert len(captured_updates) == 1 + kw = captured_updates[0] + assert kw["task_id"] == tid + assert kw["changed_fields"] == ["title", "body"] + + +def test_patch_empty_title_rejected_does_not_fire(client, captured_updates): + tid = _make_task() + captured_updates.clear() + r = client.patch( + f"/api/plugins/kanban/tasks/{tid}", json={"title": " "}, + ) + assert r.status_code == 400 + assert captured_updates == [] + + +def test_bulk_priority_fires_task_updated_per_task(client, captured_updates): + tid1 = _make_task("a") + tid2 = _make_task("b") + captured_updates.clear() + r = client.post( + "/api/plugins/kanban/tasks/bulk", + json={"ids": [tid1, tid2], "priority": 3}, + ) + assert r.status_code == 200 + assert all(entry["ok"] for entry in r.json()["results"]) + fired = {kw["task_id"]: kw for kw in captured_updates} + assert set(fired) == {tid1, tid2} + assert all(kw["changed_fields"] == ["priority"] for kw in fired.values()) + + +def test_patch_assignee_fires_once_via_assign_task(client, captured_updates): + """The dashboard assignee path goes through kanban_db.assign_task; the + dashboard layer must not double-fire it.""" + tid = _make_task() + captured_updates.clear() + r = client.patch( + f"/api/plugins/kanban/tasks/{tid}", json={"assignee": "bob"}, + ) + assert r.status_code == 200 + fired = [kw for kw in captured_updates if kw["task_id"] == tid] + assert len(fired) == 1 + assert fired[0]["changed_fields"] == ["assignee"] diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md index bcad31ce22e17..b0f92deee8534 100644 --- a/website/docs/user-guide/features/hooks.md +++ b/website/docs/user-guide/features/hooks.md @@ -468,6 +468,11 @@ Payload fields below are the exact event-specific fields supplied by each call s | `kanban_task_claimed` | Observer | After claim commit, in dispatcher process before worker spawn; return ignored. | `task_id`, `profile_name`, `board`, `assignee`, `run_id` | Board/task/profile/assignee identifiers. | | `kanban_task_completed` | Observer | After completion and cleanup, usually in worker process; return ignored. | `task_id`, `profile_name`, `board`, `assignee`, `run_id`, `summary` | Summary may contain project/user content. | | `kanban_task_blocked` | Observer | After a blocked transition; the dependency-wait path fires before its transaction exits. Return ignored. | `task_id`, `profile_name`, `board`, `assignee`, `run_id`, `reason` | Reason may contain project/user content. | +| `on_kanban_worker_spawned` | Observer | In the dispatcher process, after `spawn_fn` returns and the worker PID (when one was reported) is durably persisted; runs inside the board dispatch lock like `kanban_task_claimed`, so callbacks must stay fast. Return ignored. | `task_id`, `profile_name`, `board`, `assignee`, `run_id`, `worker_pid`, `workspace_path` | `workspace_path` is a filesystem path and may reveal project layout or usernames. | +| `on_kanban_worker_exited` | Observer | Tick-derived in the dispatcher process: fires when `detect_crashed_workers` reclaims a dead-PID running task, after every reclaim and breaker-accounting transaction has committed; exit latency is bounded by the dispatcher tick interval. Return ignored. | `task_id`, `profile_name`, `board`, `assignee`, `run_id`, `worker_pid`, `exit_kind`, `exit_code`, `outcome`, `retry_status` | Identifiers and exit metadata only. | +| `on_kanban_worker_stale_claim` | Observer | In the dispatcher process when `release_stale_claims` reclaims a TTL-expired claim, after the reclaim transaction commits; live-PID claim extensions and deferred reclaims do not fire. Return ignored. | `task_id`, `profile_name`, `board`, `assignee`, `run_id`, `worker_pid`, `heartbeat_stale`, `retry_status` | Identifiers and claim metadata only. | +| `on_kanban_task_updated` | Observer | After a committed task-row field mutation outside the claim/complete/block lifecycle: `assign_task`, `set_model_override`, `set_reasoning_effort`, plus the dashboard plugin API's direct-SQL priority/title/body editors (single and bulk) via `kanban_db.notify_task_updated`. Return ignored. | `task_id`, `profile_name`, `board`, `assignee`, `run_id`, `changed_fields` | `changed_fields` carries field names only, never values; the named title/body values in the board DB may contain user/project content. | +| `on_kanban_dispatch_tick` | Observer | Once per dispatcher tick in `dispatch_once`, strictly after the board's single-writer dispatch lock is released (never inside the writer critical section); also fires for idle and lock-contended ticks. Return ignored. | `board`, `profile_name`, `dry_run`, `outcome`, `result` | `result` is the tick's `DispatchResult` and carries task ids, assignees, and workspace paths. | --- @@ -1521,6 +1526,30 @@ Fires after a normal blocked transition. The dependency-wait path invokes it bef All three kanban hooks are observer-only and carry `task_id`, `profile_name`, `board`, `assignee`, and `run_id`; completed adds `summary`, and blocked adds `reason`. +### Kanban worker-lifecycle, task-mutation, and dispatch observers + +These observers extend the kanban lifecycle family with the worker and mutation events accepted from RFC #58548 (batch disposition in #64231). All of them are observer-only, fire only after the relevant write transaction has committed, and short-circuit on `has_hook` — with no subscriber registered, no payload is built and dispatch behavior is unchanged. The task-scoped hooks carry the same common fields as the shipped kanban hooks: `task_id`, `profile_name`, `board`, `assignee`, and `run_id`. + +#### `on_kanban_worker_spawned` + +Fires in the dispatcher process after `spawn_fn` returns and the worker PID (when one was reported) has been durably persisted. Adds `worker_pid` (may be `None` when the spawn function reports no PID) and `workspace_path`. Like `kanban_task_claimed`, it runs inside the board's dispatch lock, so callbacks must stay fast. + +#### `on_kanban_worker_exited` + +Worker exits are tick-derived: this fires when `detect_crashed_workers` reclaims a running task whose worker PID is no longer alive, after every reclaim and breaker-accounting transaction has committed. Exit visibility latency is therefore bounded by the dispatcher tick interval. Adds `worker_pid`, `exit_kind` (`clean_exit` | `rate_limited` | `nonzero_exit` | `signaled` | `unknown`), `exit_code`, `outcome` (`crashed` | `rate_limited`), and `retry_status` (the phase the task was released back to). + +#### `on_kanban_worker_stale_claim` + +Fires in the dispatcher process when `release_stale_claims` reclaims a TTL-expired claim, after the reclaim transaction commits. Live-PID claim extensions and deferred reclaims do not fire. Adds `worker_pid`, `heartbeat_stale`, and `retry_status`. + +#### `on_kanban_task_updated` + +The task-mutation boundary observer. Fires after a committed task-row field write outside the claim/complete/block lifecycle: `kanban_db.assign_task`, `set_model_override`, and `set_reasoning_effort`, plus the dashboard plugin API's direct-SQL priority/title/body editors (single-task and bulk), which report through `kanban_db.notify_task_updated`. Adds `changed_fields`, a list of field names only — new values are never carried in the payload; fetch the task if you need them. Status transitions stay in the lifecycle hook family, and dispatcher bookkeeping fields (worker PID, workspace path, claim columns) are deliberately excluded as noise. + +#### `on_kanban_dispatch_tick` + +Fires once per dispatcher tick in `dispatch_once`, strictly after the board's single-writer dispatch lock has been released — never inside the writer critical section — so a slow subscriber cannot stall a sibling dispatcher. Idle and lock-contended ticks fire too. It is not task-scoped; its payload is `board`, `profile_name`, `dry_run`, `outcome` (`ok` | `skipped_locked` | `idle`), and `result` (the tick's `DispatchResult`). + --- ## Shell Hooks