mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-19 15:18:03 +00:00
feat(kanban): typed block reasons + unblock-loop breaker (#52848)
* feat(kanban): typed block reasons + unblock-loop breaker Stops the kanban blocked-task loop: a worker blocks a task, a cron unblocks it, the worker re-blocks for the same reason, repeat forever. block_task now takes a typed kind and a persistent block_recurrences counter on the tasks table: - kind=dependency routes to todo (parent-gated, auto-resumed), never the human 'blocked' bucket a cron would keep unblocking. - needs_input/capability/transient/untyped land in blocked; each same-cause re-block after an unblock increments block_recurrences, and at BLOCK_RECURRENCE_LIMIT (default 2) the task routes to triage for a human instead of blocked. - unblock_task no longer resets block_recurrences (the amnesia that let the loop run unbounded); complete_task clears it on success. Wired through the worker kanban_block tool (new kind arg) and the hermes kanban block --kind CLI flag, both reporting where the task actually landed. Docs + 11 new tests; 536 existing kanban tests green. * test(kanban): make second-block notify test use a distinct block cause test_notifier_second_blocked_delivers blocked the same task twice with the same (untyped) reason, which now trips the new unblock-loop breaker and routes the second block to triage instead of blocked — so only one 'blocked' notification fired. The test's actual intent is that TWO distinct block cycles each notify; give the two cycles different kinds (needs_input then capability) so they're genuinely separate blocks. The same-cause loop→triage path is covered by test_kanban_block_kinds.py.
This commit is contained in:
Teknium
GitHub
parent
43b8ba4181
commit
5b5c79a8ef
6 changed files with 563 additions and 61 deletions
|
|
@ -559,6 +559,16 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu
|
|||
p_block.add_argument("reason", nargs="*", help="Reason (also appended as a comment)")
|
||||
p_block.add_argument("--ids", nargs="+", default=None,
|
||||
help="Additional task ids to block with the same reason (bulk mode)")
|
||||
p_block.add_argument(
|
||||
"--kind", default=None, choices=sorted(kb.VALID_BLOCK_KINDS),
|
||||
help=(
|
||||
"Typed block reason. 'dependency' waits in todo (auto-promoted "
|
||||
"when parents finish, no human); 'needs_input'/'capability' go to "
|
||||
"blocked for a human; 'transient' marks a maybe-flaky failure. "
|
||||
"Repeated same-kind re-blocks after unblock route the task to "
|
||||
"triage to break unblock loops. Omit for a generic block."
|
||||
),
|
||||
)
|
||||
|
||||
p_schedule = sub.add_parser("schedule", help="Park one or more tasks in Scheduled (waiting on time, not human input)")
|
||||
p_schedule.add_argument("task_id")
|
||||
|
|
@ -1928,6 +1938,7 @@ def _cmd_edit(args: argparse.Namespace) -> int:
|
|||
|
||||
def _cmd_block(args: argparse.Namespace) -> int:
|
||||
reason = " ".join(args.reason).strip() if args.reason else None
|
||||
kind = getattr(args, "kind", None)
|
||||
author = _profile_author()
|
||||
ids = [args.task_id] + list(getattr(args, "ids", None) or [])
|
||||
failed: list[str] = []
|
||||
|
|
@ -1939,12 +1950,26 @@ def _cmd_block(args: argparse.Namespace) -> int:
|
|||
conn,
|
||||
tid,
|
||||
reason=reason,
|
||||
kind=kind,
|
||||
expected_run_id=_worker_run_id_for(tid),
|
||||
):
|
||||
failed.append(tid)
|
||||
print(f"cannot block {tid}", file=sys.stderr)
|
||||
else:
|
||||
print(f"Blocked {tid}" + (f": {reason}" if reason else ""))
|
||||
# Report where the task actually landed — dependency blocks go
|
||||
# to todo, and a tripped unblock-loop breaker routes to triage.
|
||||
landed = kb.get_task(conn, tid)
|
||||
where = landed.status if landed else "blocked"
|
||||
suffix = f": {reason}" if reason else ""
|
||||
if where == "todo":
|
||||
print(f"{tid} → todo (dependency wait){suffix}")
|
||||
elif where == "triage":
|
||||
print(
|
||||
f"{tid} → triage (unblock loop detected — needs a "
|
||||
f"human decision){suffix}"
|
||||
)
|
||||
else:
|
||||
print(f"Blocked {tid}{suffix}")
|
||||
return 0 if not failed else 1
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -100,6 +100,37 @@ _log = logging.getLogger(__name__)
|
|||
|
||||
VALID_STATUSES = {"triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done", "archived"}
|
||||
VALID_INITIAL_STATUSES = {"running", "blocked"}
|
||||
|
||||
# Typed block reasons. Distinguishes the two fundamentally different things a
|
||||
# worker (or human) means by "blocked", so each can be routed differently
|
||||
# instead of all landing in one undifferentiated ``blocked`` bucket that a cron
|
||||
# unblocks → worker re-blocks → cron unblocks … forever.
|
||||
#
|
||||
# * ``dependency`` — can't proceed until another task finishes. Routed to
|
||||
# ``todo`` (NOT ``blocked``) so the existing
|
||||
# parent-gating / ``recompute_ready`` machinery promotes
|
||||
# it automatically once parents are done. No human, no
|
||||
# cron, no retry storm.
|
||||
# * ``needs_input`` — needs a human decision/answer it cannot derive.
|
||||
# * ``capability`` — hit a hard wall (no access, missing creds, an action no
|
||||
# AI agent can perform). Genuinely human-only.
|
||||
# * ``transient`` — a flaky/temporary failure that may clear on retry.
|
||||
#
|
||||
# ``needs_input`` and ``capability`` are "truly blocked": they go to ``blocked``
|
||||
# for a human, and the unblock-loop breaker (see ``block_task`` /
|
||||
# ``BLOCK_RECURRENCE_LIMIT``) escalates them to ``triage`` if a cron keeps
|
||||
# unblocking them only to have the worker re-block for the same reason.
|
||||
# ``None`` = legacy/un-typed block (treated as a generic human blocker).
|
||||
VALID_BLOCK_KINDS = {"dependency", "needs_input", "capability", "transient"}
|
||||
|
||||
# After a task has been blocked, unblocked, and re-blocked this many times for
|
||||
# the same (truly-blocked) reason, the unblock-loop breaker stops trusting the
|
||||
# unblocker (usually a cron) and routes the task to ``triage`` instead of back
|
||||
# to ``blocked`` — breaking the infinite unblock↔re-block loop and forcing a
|
||||
# human-in-the-loop decision. Mirrors the dispatcher's ``DEFAULT_FAILURE_LIMIT``
|
||||
# spirit (default 2) but counts a different signal: manual unblock recurrences,
|
||||
# not dispatcher spawn/crash/timeout failures.
|
||||
BLOCK_RECURRENCE_LIMIT = 2
|
||||
VALID_WORKSPACE_KINDS = {"scratch", "worktree", "dir"}
|
||||
KNOWN_TOOLSET_NAMES = frozenset(name.casefold() for name in get_toolset_names())
|
||||
_IS_WINDOWS = sys.platform == "win32"
|
||||
|
|
@ -838,6 +869,13 @@ class Task:
|
|||
# set the env var. Lets clients render a per-session board without
|
||||
# relying on tenant + time-window heuristics.
|
||||
session_id: Optional[str] = None
|
||||
# Typed block reason (one of VALID_BLOCK_KINDS) or None for legacy/un-typed
|
||||
# blocks. Set by ``block_task``; preserved across unblock so a re-block for
|
||||
# the same kind is recognisable as an unblock↔re-block loop.
|
||||
block_kind: Optional[str] = None
|
||||
# Unblock-loop counter. See the column comment in SCHEMA_SQL and
|
||||
# ``BLOCK_RECURRENCE_LIMIT``. Reset only on successful completion.
|
||||
block_recurrences: int = 0
|
||||
|
||||
@classmethod
|
||||
def from_row(cls, row: sqlite3.Row) -> "Task":
|
||||
|
|
@ -914,6 +952,14 @@ class Task:
|
|||
session_id=(
|
||||
row["session_id"] if "session_id" in keys else None
|
||||
),
|
||||
block_kind=(
|
||||
row["block_kind"] if "block_kind" in keys and row["block_kind"] else None
|
||||
),
|
||||
block_recurrences=(
|
||||
int(row["block_recurrences"])
|
||||
if "block_recurrences" in keys and row["block_recurrences"] is not None
|
||||
else 0
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -1078,7 +1124,20 @@ CREATE TABLE IF NOT EXISTS tasks (
|
|||
-- for tasks created from the CLI, dashboard, or any path that doesn't
|
||||
-- set the env var. Indexed so per-session list queries stay cheap on
|
||||
-- larger boards.
|
||||
session_id TEXT
|
||||
session_id TEXT,
|
||||
-- Typed block reason set by ``block_task`` (one of VALID_BLOCK_KINDS, or
|
||||
-- NULL for legacy/un-typed blocks). Drives routing: ``dependency`` never
|
||||
-- sits in ``blocked`` (goes to ``todo`` for parent-gating); the others go
|
||||
-- to ``blocked`` for a human. Preserved across unblock so a re-block for
|
||||
-- the SAME kind can be recognised as a loop.
|
||||
block_kind TEXT,
|
||||
-- Unblock-loop counter. Incremented each time a task is re-blocked for the
|
||||
-- same truly-blocked reason after having been unblocked. When it reaches
|
||||
-- BLOCK_RECURRENCE_LIMIT the task is routed to ``triage`` instead of
|
||||
-- ``blocked`` so a cron can't spin it forever. Reset to 0 only on a
|
||||
-- successful completion — NOT on unblock (resetting on unblock is exactly
|
||||
-- the amnesia that let the loop run unbounded).
|
||||
block_recurrences INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS task_links (
|
||||
|
|
@ -1873,6 +1932,22 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
|
|||
conn, "tasks", "session_id", "session_id TEXT"
|
||||
)
|
||||
|
||||
if "block_kind" not in cols:
|
||||
# Typed block reason (VALID_BLOCK_KINDS) or NULL for legacy/un-typed
|
||||
# blocks. Existing blocked rows get NULL, which is treated as a
|
||||
# generic human blocker — same behaviour they had before the column.
|
||||
_add_column_if_missing(conn, "tasks", "block_kind", "block_kind TEXT")
|
||||
|
||||
if "block_recurrences" not in cols:
|
||||
# Unblock-loop counter. Existing rows start at 0, so the loop breaker
|
||||
# only begins counting from the first re-block after this migration.
|
||||
_add_column_if_missing(
|
||||
conn,
|
||||
"tasks",
|
||||
"block_recurrences",
|
||||
"block_recurrences INTEGER NOT NULL DEFAULT 0",
|
||||
)
|
||||
|
||||
# Indexes over additive ``tasks`` columns must be created after the
|
||||
# columns exist. Keeping them in SCHEMA_SQL breaks legacy boards: SQLite
|
||||
# parses each statement in ``executescript`` against the live schema, so a
|
||||
|
|
@ -3898,7 +3973,9 @@ def complete_task(
|
|||
completed_at = ?,
|
||||
claim_lock = NULL,
|
||||
claim_expires= NULL,
|
||||
worker_pid = NULL
|
||||
worker_pid = NULL,
|
||||
block_kind = NULL,
|
||||
block_recurrences = 0
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready', 'blocked')
|
||||
""",
|
||||
|
|
@ -3913,7 +3990,9 @@ def complete_task(
|
|||
completed_at = ?,
|
||||
claim_lock = NULL,
|
||||
claim_expires= NULL,
|
||||
worker_pid = NULL
|
||||
worker_pid = NULL,
|
||||
block_kind = NULL,
|
||||
block_recurrences = 0
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready', 'blocked')
|
||||
AND current_run_id = ?
|
||||
|
|
@ -4385,53 +4464,204 @@ def block_task(
|
|||
task_id: str,
|
||||
*,
|
||||
reason: Optional[str] = None,
|
||||
kind: Optional[str] = None,
|
||||
expected_run_id: Optional[int] = None,
|
||||
) -> bool:
|
||||
"""Transition ``running -> blocked``."""
|
||||
with write_txn(conn):
|
||||
if expected_run_id is None:
|
||||
cur = conn.execute(
|
||||
"""
|
||||
UPDATE tasks
|
||||
SET status = 'blocked',
|
||||
claim_lock = NULL,
|
||||
claim_expires= NULL,
|
||||
worker_pid = NULL
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready')
|
||||
""",
|
||||
(task_id,),
|
||||
)
|
||||
else:
|
||||
cur = conn.execute(
|
||||
"""
|
||||
UPDATE tasks
|
||||
SET status = 'blocked',
|
||||
claim_lock = NULL,
|
||||
claim_expires= NULL,
|
||||
worker_pid = NULL
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready')
|
||||
AND current_run_id = ?
|
||||
""",
|
||||
(task_id, int(expected_run_id)),
|
||||
)
|
||||
if cur.rowcount != 1:
|
||||
return False
|
||||
run_id = _end_run(
|
||||
conn, task_id,
|
||||
outcome="blocked", status="blocked",
|
||||
summary=reason,
|
||||
"""Transition ``running``/``ready`` → ``blocked`` (or route elsewhere).
|
||||
|
||||
``kind`` (one of :data:`VALID_BLOCK_KINDS`, or ``None`` for a legacy
|
||||
un-typed block) drives routing instead of every block landing in one
|
||||
undifferentiated ``blocked`` bucket:
|
||||
|
||||
* ``dependency`` — the task is only waiting on another task. It does NOT
|
||||
sit in ``blocked`` (where a cron would keep "unblocking" it); it goes to
|
||||
``todo`` so the existing parent-gating / ``recompute_ready`` machinery
|
||||
promotes it automatically once its parents finish. No human, no cron, no
|
||||
retry storm. This is Dale's "Type 2 — dependency blocked".
|
||||
|
||||
* ``needs_input`` / ``capability`` / ``None`` — "truly blocked" (Dale's
|
||||
"Type 1"). Lands in ``blocked`` for a human. BUT: each time such a task
|
||||
is re-blocked for the SAME kind after having been unblocked, the
|
||||
unblock-loop counter (``block_recurrences``) increments. When it reaches
|
||||
:data:`BLOCK_RECURRENCE_LIMIT`, the task is routed to ``triage`` instead
|
||||
of ``blocked`` — breaking the cron-unblock ↔ worker-re-block loop and
|
||||
forcing a human-in-the-loop triage decision.
|
||||
|
||||
* ``transient`` — treated like a generic block for routing, but a worker
|
||||
can use it to signal "this might clear on its own"; it still participates
|
||||
in the loop breaker so a forever-flaky task eventually escalates.
|
||||
|
||||
Returns True on any successful transition (to ``blocked``, ``todo``, or
|
||||
``triage``), False when the task wasn't in a blockable state.
|
||||
"""
|
||||
if kind is not None and kind not in VALID_BLOCK_KINDS:
|
||||
raise ValueError(
|
||||
f"block kind must be one of {sorted(VALID_BLOCK_KINDS)} or None"
|
||||
)
|
||||
# Synthesize a run when blocking a never-claimed task so the
|
||||
# reason is preserved in attempt history.
|
||||
if run_id is None and reason:
|
||||
run_id = _synthesize_ended_run(
|
||||
routed_to = "blocked"
|
||||
recurrences = 0
|
||||
with write_txn(conn):
|
||||
cur_row = conn.execute(
|
||||
"SELECT status, block_kind, block_recurrences FROM tasks WHERE id = ?",
|
||||
(task_id,),
|
||||
).fetchone()
|
||||
if cur_row is None:
|
||||
return False
|
||||
prev_kind = cur_row["block_kind"] if "block_kind" in cur_row.keys() else None
|
||||
prev_recurrences = (
|
||||
int(cur_row["block_recurrences"])
|
||||
if "block_recurrences" in cur_row.keys()
|
||||
and cur_row["block_recurrences"] is not None
|
||||
else 0
|
||||
)
|
||||
|
||||
# Dependency blocks never enter the human ``blocked`` bucket — they
|
||||
# wait in ``todo`` and let ``recompute_ready`` gate on parents. Routing
|
||||
# here (rather than ``blocked``) is what keeps a cron from ever seeing
|
||||
# a dependency-wait as something to "unblock".
|
||||
if kind == "dependency":
|
||||
cur = conn.execute(
|
||||
"""
|
||||
UPDATE tasks
|
||||
SET status = 'todo',
|
||||
claim_lock = NULL,
|
||||
claim_expires = NULL,
|
||||
worker_pid = NULL,
|
||||
block_kind = ?
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready')
|
||||
""" + ("" if expected_run_id is None else " AND current_run_id = ?"),
|
||||
(kind, task_id) if expected_run_id is None
|
||||
else (kind, task_id, int(expected_run_id)),
|
||||
)
|
||||
if cur.rowcount != 1:
|
||||
return False
|
||||
run_id = _end_run(
|
||||
conn, task_id,
|
||||
outcome="blocked",
|
||||
outcome="blocked", status="blocked",
|
||||
summary=reason,
|
||||
)
|
||||
_append_event(conn, task_id, "blocked", {"reason": reason}, run_id=run_id)
|
||||
if run_id is None and reason:
|
||||
run_id = _synthesize_ended_run(
|
||||
conn, task_id, outcome="blocked", summary=reason,
|
||||
)
|
||||
_append_event(
|
||||
conn, task_id, "dependency_wait",
|
||||
{"reason": reason, "kind": kind}, run_id=run_id,
|
||||
)
|
||||
routed_to = "todo"
|
||||
_blocked_task = get_task(conn, task_id)
|
||||
_fire_kanban_lifecycle_hook(
|
||||
"kanban_task_blocked",
|
||||
task_id,
|
||||
board=get_current_board(),
|
||||
assignee=_blocked_task.assignee if _blocked_task else None,
|
||||
run_id=run_id,
|
||||
reason=reason,
|
||||
)
|
||||
return True
|
||||
|
||||
# Truly-blocked kinds. Increment the unblock-loop counter when this is a
|
||||
# re-block for the SAME reason after a prior unblock. block_task only
|
||||
# fires from running/ready (i.e. AFTER an unblock returned the task to
|
||||
# the work pool), so a stored block_kind that matches the incoming kind
|
||||
# means: blocked → unblocked → about-to-re-block for the same cause.
|
||||
# An un-typed (None) block compares as "same" to a prior un-typed block.
|
||||
same_cause = prev_kind == kind
|
||||
recurrences = prev_recurrences + 1 if same_cause else 1
|
||||
|
||||
if recurrences >= BLOCK_RECURRENCE_LIMIT:
|
||||
# Loop detected — stop letting the unblocker spin this task. Route
|
||||
# to triage for a human-in-the-loop decision instead of blocked.
|
||||
cur = conn.execute(
|
||||
"""
|
||||
UPDATE tasks
|
||||
SET status = 'triage',
|
||||
claim_lock = NULL,
|
||||
claim_expires = NULL,
|
||||
worker_pid = NULL,
|
||||
block_kind = ?,
|
||||
block_recurrences = ?
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready')
|
||||
""" + ("" if expected_run_id is None else " AND current_run_id = ?"),
|
||||
(kind, recurrences, task_id) if expected_run_id is None
|
||||
else (kind, recurrences, task_id, int(expected_run_id)),
|
||||
)
|
||||
if cur.rowcount != 1:
|
||||
return False
|
||||
run_id = _end_run(
|
||||
conn, task_id,
|
||||
outcome="blocked", status="blocked",
|
||||
summary=reason,
|
||||
)
|
||||
if run_id is None and reason:
|
||||
run_id = _synthesize_ended_run(
|
||||
conn, task_id, outcome="blocked", summary=reason,
|
||||
)
|
||||
_append_event(
|
||||
conn, task_id, "block_loop_detected",
|
||||
{
|
||||
"reason": reason,
|
||||
"kind": kind,
|
||||
"recurrences": recurrences,
|
||||
"limit": BLOCK_RECURRENCE_LIMIT,
|
||||
},
|
||||
run_id=run_id,
|
||||
)
|
||||
routed_to = "triage"
|
||||
else:
|
||||
if expected_run_id is None:
|
||||
cur = conn.execute(
|
||||
"""
|
||||
UPDATE tasks
|
||||
SET status = 'blocked',
|
||||
claim_lock = NULL,
|
||||
claim_expires = NULL,
|
||||
worker_pid = NULL,
|
||||
block_kind = ?,
|
||||
block_recurrences = ?
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready')
|
||||
""",
|
||||
(kind, recurrences, task_id),
|
||||
)
|
||||
else:
|
||||
cur = conn.execute(
|
||||
"""
|
||||
UPDATE tasks
|
||||
SET status = 'blocked',
|
||||
claim_lock = NULL,
|
||||
claim_expires = NULL,
|
||||
worker_pid = NULL,
|
||||
block_kind = ?,
|
||||
block_recurrences = ?
|
||||
WHERE id = ?
|
||||
AND status IN ('running', 'ready')
|
||||
AND current_run_id = ?
|
||||
""",
|
||||
(kind, recurrences, task_id, int(expected_run_id)),
|
||||
)
|
||||
if cur.rowcount != 1:
|
||||
return False
|
||||
run_id = _end_run(
|
||||
conn, task_id,
|
||||
outcome="blocked", status="blocked",
|
||||
summary=reason,
|
||||
)
|
||||
# Synthesize a run when blocking a never-claimed task so the
|
||||
# reason is preserved in attempt history.
|
||||
if run_id is None and reason:
|
||||
run_id = _synthesize_ended_run(
|
||||
conn, task_id,
|
||||
outcome="blocked",
|
||||
summary=reason,
|
||||
)
|
||||
_append_event(
|
||||
conn, task_id, "blocked",
|
||||
{"reason": reason, "kind": kind, "recurrences": recurrences},
|
||||
run_id=run_id,
|
||||
)
|
||||
_blocked_task = get_task(conn, task_id)
|
||||
_fire_kanban_lifecycle_hook(
|
||||
"kanban_task_blocked",
|
||||
|
|
@ -4556,6 +4786,16 @@ def unblock_task(conn: sqlite3.Connection, task_id: str) -> bool:
|
|||
(task_id,),
|
||||
).fetchone()
|
||||
new_status = "todo" if undone_parents else "ready"
|
||||
# NOTE: deliberately does NOT touch ``block_recurrences`` or
|
||||
# ``block_kind``. Resetting the recurrence counter on unblock is exactly
|
||||
# the amnesia that let a cron unblock → worker re-block loop run
|
||||
# unbounded (Dale's report). The counter survives the unblock so that a
|
||||
# subsequent same-cause ``block_task`` can detect the loop and route to
|
||||
# triage at ``BLOCK_RECURRENCE_LIMIT``. It is reset to 0 only on a
|
||||
# successful completion (see ``complete_task``). ``consecutive_failures``
|
||||
# (the *dispatcher* spawn/crash/timeout counter — a different signal) is
|
||||
# still reset here, which is correct: a deliberate unblock is a fresh
|
||||
# start for the dispatcher's retry budget.
|
||||
cur = conn.execute(
|
||||
"UPDATE tasks SET status = ?, current_run_id = NULL, "
|
||||
"consecutive_failures = 0, last_failure_error = NULL "
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue