RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(kanban): move descendant invalidation to domain layer, make it non-silent 

Ancestor-reopen descendant invalidation previously lived only in the
dashboard plugin (_set_status_direct), so board semantics diverged by
surface and the retraction was silent: completed work snapped back to
todo and live workers were killed with no operator-visible signal.

Move it into kanban_db.invalidate_descendants_for_parent_reopen as THE
single domain implementation (recursive-CTE discovery and per-run
_retry_status_for_run handling preserved). It composes under a caller's
open transaction via write_txn(allow_nested=True) — the ancestor flip
and the descendant retractions must commit atomically — and opens its
own transaction standalone. The dashboard shim now delegates; the CLI
deliberately has no done-reopen verb (reopen-review is review-phase
only), so the DB-layer function being the single implementation is the
fix, documented in its docstring.

Non-silent: every invalidated descendant gets a descendant_invalidated
event ({ancestor, prior_status, new_status, resume_status}), the legacy
status event for existing live-feed consumers, and a task comment
naming the reopened ancestor. Running descendants keep the termination
behavior (a child building on a retracted premise is wasted spend), but
the events/comment are committed BEFORE the kill, which routes through
_terminate_reclaimed_worker — the same helper the reclaim paths use.

consecutive_failures resets to 0 on invalidated descendants: operator-
initiated invalidation is a deliberate fresh start, deliberately the
opposite of the review-loop rule (reopen_review_task preserves the
counter, #35072) so the autonomous review loop can't launder its own
failure streak.

Regression: DB-function reopen demotes done descendants with events +
comments; running descendant's audit trail is durable before its worker
dies; counter resets; dashboard and DB paths produce identical task
states, event kinds, and comment counts.
This commit is contained in:
Teknium 2026-08-10 12:37:52 -07:00
parent 917c27d4a5
commit a98aee47ce
3 changed files with 415 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

170
hermes_cli/kanban_db.py
View File

@ -6652,6 +6652,176 @@ def reopen_review_task(conn: sqlite3.Connection, task_id: str) -> bool:
return True
def invalidate_descendants_for_parent_reopen(
conn: sqlite3.Connection,
task_id: str,
*,
author: str,
) -> dict[str, Any]:
"""Retract every dispatchable/completed descendant of a reopened ancestor.
THE single domain implementation of done-reopen descendant invalidation.
When a ``done`` (or ``archived``) ancestor is reopened, every descendant
whose state assumed the ancestor's result — ``ready``, ``review``,
``running`` or ``done`` — is building on a retracted premise, so it is
demoted to ``todo`` and re-gated on the graph. The CLI deliberately has
NO done-reopen verb on this branch (``reopen-review`` only handles the
review-phase transition via :func:`reopen_review_task`), so every surface
that reopens a done task (dashboard drag-drop / PATCH — single and bulk —
via ``_set_status_direct``) must route through this function; keeping the
implementation here means a future CLI or tool reopen verb inherits
identical semantics for free.
Transactionality: composes under the caller's already-open transaction
via ``write_txn(conn, allow_nested=True)`` — the dashboard's status
writer must commit the ancestor's status flip and the descendant
retractions atomically (a crash between the two would leave stale done
descendants claiming a premise that no longer holds). Called standalone
it opens its own transaction. All SQL is inline per this file's txn
conventions (no calls into other txn-opening helpers).
Non-silent contract: every invalidated descendant gets
* a ``descendant_invalidated`` event with ``{ancestor, prior_status,
new_status}`` (plus ``resume_status``) for board/notifier surfaces,
* the legacy ``status`` event (``reason=ancestor_reopened``) the live
feed already renders, and
* a ``task_comments`` row naming the reopened ancestor, so operators see
WHY a card moved instead of watching it silently teleport.
Live ``running`` descendants keep the termination behavior (a running
child building on a retracted premise is wasted spend): their run is
closed ``reclaimed`` and their worker is killed via
:func:`_terminate_reclaimed_worker` — the same helper the reclaim paths
use. Events/comments are written inside the transaction and the kill
happens strictly post-commit, so the audit trail exists BEFORE the
worker dies. When this function opened its own transaction it performs
the terminations itself after commit; when composing under a caller's
transaction the caller MUST drain the returned ``terminations`` list
with ``_terminate_reclaimed_worker`` after its own commit.
``consecutive_failures`` is reset to 0 on every invalidated descendant:
ancestor reopen is a deliberate operator action, so demoted work gets a
fresh start with the breaker (a previously auto-blocked-then-completed
descendant should not re-enter the queue one failure from the breaker).
This is deliberately the OPPOSITE of the review-transition rule
(:func:`reopen_review_task` / #35072 preserves the counter) because the
autonomous review loop must not be able to launder its own failure
streak, while an operator invalidating a subtree is an explicit reset
signal.
Returns ``{"invalidated": [...], "terminations": [...]}`` where each
invalidated entry is ``{id, prior_status, new_status, resume_status}``
and each termination is a ``(worker_pid, claim_lock)`` tuple.
"""
caller_owns_txn = bool(getattr(conn, "in_transaction", False))
now = int(time.time())
invalidated: list[dict[str, Any]] = []
terminations: list[tuple[Optional[int], Optional[str]]] = []
with write_txn(conn, allow_nested=True):
rows = conn.execute(
"""
WITH RECURSIVE descendants(id) AS (
SELECT child_id FROM task_links WHERE parent_id = ?
UNION
SELECT l.child_id
FROM task_links l
JOIN descendants d ON d.id = l.parent_id
)
SELECT t.id, t.status, t.current_run_id, t.worker_pid, t.claim_lock
FROM descendants d
JOIN tasks t ON t.id = d.id
ORDER BY t.id
""",
(task_id,),
).fetchall()
for row in rows:
previous_status = row["status"]
if previous_status not in {"ready", "review", "running", "done"}:
continue
resume_status = "ready"
run_id = None
if previous_status == "review":
resume_status = "review"
elif previous_status == "running":
resume_status = _retry_status_for_run(
conn, row["id"], row["current_run_id"]
)
terminations.append((row["worker_pid"], row["claim_lock"]))
run_id = _end_run(
conn,
row["id"],
outcome="reclaimed",
status="todo",
summary=f"ancestor {task_id} reopened",
)
# consecutive_failures = 0: deliberate operator reset — see
# docstring for why this diverges from reopen_review_task.
conn.execute(
"UPDATE tasks SET status = 'todo', completed_at = NULL, "
"claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, "
"current_run_id = NULL, consecutive_failures = 0 WHERE id = ?",
(row["id"],),
)
_append_event(
conn,
row["id"],
"descendant_invalidated",
{
"ancestor": task_id,
"prior_status": previous_status,
"new_status": "todo",
"resume_status": resume_status,
},
run_id=run_id,
)
# Legacy 'status' event kept so existing live-feed consumers
# still see the move without learning the new event kind.
_append_event(
conn,
row["id"],
"status",
{
"status": "todo",
"reason": "ancestor_reopened",
"parent": task_id,
"previous_status": previous_status,
"resume_status": resume_status,
},
run_id=run_id,
)
# Inline comment insert (not add_comment: no txn-opening helper
# calls inside a txn per file convention).
conn.execute(
"INSERT INTO task_comments (task_id, author, body, created_at) "
"VALUES (?, ?, ?, ?)",
(
row["id"],
author,
(
f"Invalidated: ancestor {task_id} was reopened; "
f"retracted from '{previous_status}' to 'todo' "
f"(will resume via '{resume_status}')."
),
now,
),
)
invalidated.append(
{
"id": row["id"],
"prior_status": previous_status,
"new_status": "todo",
"resume_status": resume_status,
}
)
if not caller_owns_txn:
# Standalone call: we committed above, so the audit trail is durable
# — safe to kill workers now. Composed calls leave this to the
# caller (post-commit), preserving events-before-termination.
for pid, claim_lock in terminations:
_terminate_reclaimed_worker(pid, claim_lock)
return {"invalidated": invalidated, "terminations": terminations}
def specify_triage_task(
conn: sqlite3.Connection,
task_id: str,

71
plugins/kanban/dashboard/plugin_api.py
View File

@ -1087,62 +1087,21 @@ def _invalidate_descendants_for_parent_reopen(
parent_id: str,
terminations: list[tuple[Optional[int], Optional[str]]],
) -> None:
"""Retract every dispatchable/completed descendant of a reopened parent."""
rows = conn.execute(
"""
WITH RECURSIVE descendants(id) AS (
SELECT child_id FROM task_links WHERE parent_id = ?
UNION
SELECT l.child_id
FROM task_links l
JOIN descendants d ON d.id = l.parent_id
)
SELECT t.id, t.status, t.current_run_id, t.worker_pid, t.claim_lock
FROM descendants d
JOIN tasks t ON t.id = d.id
ORDER BY t.id
""",
(parent_id,),
).fetchall()
for row in rows:
previous_status = row["status"]
if previous_status not in {"ready", "review", "running", "done"}:
continue
resume_status = "ready"
run_id = None
if previous_status == "review":
resume_status = "review"
elif previous_status == "running":
resume_status = kanban_db._retry_status_for_run(
conn, row["id"], row["current_run_id"]
)
terminations.append((row["worker_pid"], row["claim_lock"]))
run_id = kanban_db._end_run(
conn,
row["id"],
outcome="reclaimed",
status="todo",
summary=f"ancestor {parent_id} reopened",
)
conn.execute(
"UPDATE tasks SET status = 'todo', completed_at = NULL, "
"claim_lock = NULL, claim_expires = NULL, worker_pid = NULL, "
"current_run_id = NULL WHERE id = ?",
(row["id"],),
)
kanban_db._append_event(
conn,
row["id"],
"status",
{
"status": "todo",
"reason": "ancestor_reopened",
"parent": parent_id,
"previous_status": previous_status,
"resume_status": resume_status,
},
run_id=run_id,
)
"""Delegate to the domain-layer implementation in :mod:`kanban_db`.
Kept as a thin shim so ``_set_status_direct`` stays readable; the actual
invalidation (recursive-CTE discovery, per-descendant events + comments,
run closing, failure-counter reset) lives in
:func:`kanban_db.invalidate_descendants_for_parent_reopen` so every
reopen surface shares one implementation. We run inside the caller's
open transaction, so the domain function composes via a savepoint and
returns the worker terminations for us to perform post-commit (events
must be durable BEFORE the kill).
"""
result = kanban_db.invalidate_descendants_for_parent_reopen(
conn, parent_id, author="dashboard",
)
terminations.extend(result["terminations"])
def _set_status_direct(

230
tests/hermes_cli/test_kanban_parent_reopen_invalidation.py Normal file
View File

@ -0,0 +1,230 @@
"""Regressions for domain-layer descendant invalidation on ancestor reopen.
``kanban_db.invalidate_descendants_for_parent_reopen`` is the single
implementation of "a done ancestor was reopened, retract everything that
assumed its result" (M3). These tests pin:
* done descendants are demoted to ``todo`` with a ``descendant_invalidated``
event AND a comment naming the ancestor (non-silent),
* running descendants have their audit trail committed BEFORE their worker
is terminated, and the kill routes through ``_terminate_reclaimed_worker``
(the same helper the reclaim paths use),
* ``consecutive_failures`` resets to 0 (deliberate operator action —
opposite of the review-loop rule pinned in M2), and
* the dashboard ``_set_status_direct`` reopen path and the DB function
produce identical descendant outcomes.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from hermes_cli import kanban_db as kb
@pytest.fixture
def conn(tmp_path: Path):
db = kb.connect(tmp_path / "kanban.db")
try:
yield db
finally:
db.close()
def _done_parent_with_done_child(conn):
parent_id = kb.create_task(conn, title="ancestor", assignee="planner")
assert kb.complete_task(conn, parent_id)
child_id = kb.create_task(
conn, title="child", assignee="builder", parents=[parent_id],
)
assert kb.complete_task(conn, child_id)
return parent_id, child_id
def _reopen_parent_directly(conn, parent_id: str) -> None:
"""Minimal stand-in for a reopen surface: flip done -> todo."""
with kb.write_txn(conn):
conn.execute(
"UPDATE tasks SET status = 'todo', completed_at = NULL WHERE id = ?",
(parent_id,),
)
def test_reopen_demotes_done_descendants_with_events_and_comments(conn):
parent_id, child_id = _done_parent_with_done_child(conn)
grandchild_id = kb.create_task(
conn, title="grandchild", assignee="writer", parents=[child_id],
)
assert kb.complete_task(conn, grandchild_id)
_reopen_parent_directly(conn, parent_id)
result = kb.invalidate_descendants_for_parent_reopen(
conn, parent_id, author="operator",
)
demoted = {entry["id"]: entry for entry in result["invalidated"]}
assert set(demoted) == {child_id, grandchild_id}
for tid in (child_id, grandchild_id):
assert demoted[tid]["prior_status"] == "done"
assert demoted[tid]["new_status"] == "todo"
task = kb.get_task(conn, tid)
assert task is not None and task.status == "todo"
assert task.completed_at is None
events = kb.list_events(conn, tid)
inval = [e for e in events if e.kind == "descendant_invalidated"]
assert len(inval) == 1
payload = inval[0].payload
assert payload["ancestor"] == parent_id
assert payload["prior_status"] == "done"
assert payload["new_status"] == "todo"
comments = kb.list_comments(conn, tid)
assert any(
parent_id in c.body and c.author == "operator" for c in comments
), f"no invalidation comment naming {parent_id} on {tid}"
assert result["terminations"] == []
def test_running_descendant_event_precedes_termination_via_reclaim_helper(
conn, tmp_path, monkeypatch,
):
parent_id = kb.create_task(conn, title="ancestor", assignee="planner")
assert kb.complete_task(conn, parent_id)
child_id = kb.create_task(
conn, title="running child", assignee="builder", parents=[parent_id],
)
claimed = kb.claim_task(conn, child_id)
assert claimed is not None and claimed.status == "running"
kb._set_worker_pid(conn, child_id, 424242)
kills: list[tuple] = []
def fake_terminate(pid, claim_lock, **kwargs):
# The audit trail must already be durable when the kill fires:
# standalone calls commit before terminating.
side = kb.connect(tmp_path / "kanban.db")
try:
kinds = [e.kind for e in kb.list_events(side, child_id)]
finally:
side.close()
assert "descendant_invalidated" in kinds
kills.append((pid, claim_lock))
return {"terminated": True}
monkeypatch.setattr(kb, "_terminate_reclaimed_worker", fake_terminate)
_reopen_parent_directly(conn, parent_id)
result = kb.invalidate_descendants_for_parent_reopen(
conn, parent_id, author="operator",
)
assert kills and kills[0][0] == 424242
assert result["terminations"] == kills
child = kb.get_task(conn, child_id)
assert child is not None
assert child.status == "todo"
assert child.current_run_id is None
run = kb.latest_run(conn, child_id)
assert run is not None and run.outcome == "reclaimed"
def test_counter_reset_on_invalidated_descendants(conn):
parent_id, child_id = _done_parent_with_done_child(conn)
with kb.write_txn(conn):
conn.execute(
"UPDATE tasks SET consecutive_failures = 4 WHERE id = ?",
(child_id,),
)
_reopen_parent_directly(conn, parent_id)
kb.invalidate_descendants_for_parent_reopen(conn, parent_id, author="op")
child = kb.get_task(conn, child_id)
assert child is not None
# Deliberate operator action = fresh start with the breaker; contrast
# with reopen_review_task, which PRESERVES the counter (M2 rule).
assert child.consecutive_failures == 0
def test_dashboard_and_db_paths_produce_identical_outcomes(tmp_path, monkeypatch):
fastapi = pytest.importorskip("fastapi")
from fastapi.testclient import TestClient
import importlib.util
import sys
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
kb.init_db()
repo_root = Path(__file__).resolve().parents[2]
plugin_file = repo_root / "plugins" / "kanban" / "dashboard" / "plugin_api.py"
spec = importlib.util.spec_from_file_location(
"hermes_dashboard_plugin_kanban_m3_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)
app = fastapi.FastAPI()
app.include_router(mod.router, prefix="/api/plugins/kanban")
client = TestClient(app)
def build_graph(tag: str):
with kb.connect() as c:
parent = kb.create_task(c, title=f"{tag}-parent", assignee="planner")
assert kb.complete_task(c, parent)
child = kb.create_task(
c, title=f"{tag}-child", assignee="builder", parents=[parent],
)
assert kb.complete_task(c, child)
return parent, child
dash_parent, dash_child = build_graph("dash")
db_parent, db_child = build_graph("db")
# Surface 1: dashboard drag (done -> todo) via _set_status_direct.
r = client.patch(
f"/api/plugins/kanban/tasks/{dash_parent}", json={"status": "todo"},
)
assert r.status_code == 200, r.text
# Surface 2: DB function directly (the single domain implementation).
with kb.connect() as c:
with kb.write_txn(c):
c.execute(
"UPDATE tasks SET status = 'todo', completed_at = NULL "
"WHERE id = ?",
(db_parent,),
)
kb.invalidate_descendants_for_parent_reopen(
c, db_parent, author="dashboard",
)
with kb.connect() as c:
def snapshot(tid: str):
t = kb.get_task(c, tid)
assert t is not None
kinds = sorted(e.kind for e in kb.list_events(c, tid))
n_comments = len(kb.list_comments(c, tid))
return (
t.status,
t.completed_at,
t.current_run_id,
t.consecutive_failures,
kinds,
n_comments,
)
assert snapshot(dash_child) == snapshot(db_child)
status, completed_at, _run, failures, kinds, n_comments = snapshot(db_child)
assert status == "todo"
assert completed_at is None
assert failures == 0
assert "descendant_invalidated" in kinds
assert n_comments >= 1