feat(plugins): kanban worker-lifecycle, task-mutation, and dispatch-tick observers
Implements the remaining observers from RFC #58548 (@thebizfixer), accepted as the design basis in the #64231 batch disposition: - on_kanban_worker_spawned: fires in the dispatch loop after spawn_fn returns and the worker PID is durably persisted (the RFC timing contract), in both the ready and review lanes. - on_kanban_worker_exited: tick-derived from detect_crashed_workers; fires after every reclaim/accounting txn has committed, carrying exit_kind / exit_code / outcome / retry_status. - on_kanban_worker_stale_claim: fires when release_stale_claims reclaims a TTL-expired claim; live-PID claim extensions and deferred reclaims stay silent. - on_kanban_task_updated: task-mutation boundary observer carrying changed_fields (field names only); fired by assign_task, set_model_override, and set_reasoning_effort, and by the dashboard plugin API's direct-SQL priority/title/body editors (single and bulk) through the new kanban_db.notify_task_updated seam. - on_kanban_dispatch_tick: re-port of PR #56066 (@laboratoiresonore), renamed per the taxonomy and fired strictly AFTER _dispatch_tick_lock is released; the sweeper found the original fired inside the lock, where a slow subscriber could extend the single-writer critical section and stall a sibling dispatcher. All five are observer-only (return values ignored), fire after the relevant write txn commits, and short-circuit on has_hook() so nothing is built when no consumer registers; every fire site is fully best-effort so a broken plugin can never break dispatch or a task mutation. No config surface added. Existing plugins and hook payloads are untouched. Mutation-boundary scope: every user-facing task-FIELD editor fires (assignee, priority, title, body, model/provider override, reasoning effort). Deliberately not wired: status transitions (they belong to the lifecycle hook family), dispatcher bookkeeping columns (worker_pid, workspace_path, claim columns — surfaced through the worker hooks instead), link/comment/attachment tables (not task-row writes), and the dispatcher's default-assignee auto-assign (already surfaced via DispatchResult.auto_assigned_default in the tick payload). notify_task_updated is the seam for wiring further paths. Docs: new rows plus a detail section in the shipped plugin-hook catalog. Tests: 30 new (9 worker lifecycle, 8 dispatch tick, 8 task updated, 5 dashboard mutation boundary), including a lock-probe contract test that fails if the tick hook ever fires inside the dispatch lock. Refs: RFC #58548, #64231 batch disposition, folds #56066.
This commit is contained in:
webdevtodayjason
Teknium
parent
2a26693e22
commit
5e10351683
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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"]
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue