From a98aee47cecddab9ab9f58fc3a3b94b25f78d394 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:37:52 -0700 Subject: [PATCH] fix(kanban): move descendant invalidation to domain layer, make it non-silent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- hermes_cli/kanban_db.py | 170 +++++++++++++ plugins/kanban/dashboard/plugin_api.py | 71 ++---- .../test_kanban_parent_reopen_invalidation.py | 230 ++++++++++++++++++ 3 files changed, 415 insertions(+), 56 deletions(-) create mode 100644 tests/hermes_cli/test_kanban_parent_reopen_invalidation.py diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 359c88805b52f..9a92b03d3033d 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -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, diff --git a/plugins/kanban/dashboard/plugin_api.py b/plugins/kanban/dashboard/plugin_api.py index 16c76726208e2..d4548261ebeef 100644 --- a/plugins/kanban/dashboard/plugin_api.py +++ b/plugins/kanban/dashboard/plugin_api.py @@ -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( diff --git a/tests/hermes_cli/test_kanban_parent_reopen_invalidation.py b/tests/hermes_cli/test_kanban_parent_reopen_invalidation.py new file mode 100644 index 0000000000000..3429a633b988f --- /dev/null +++ b/tests/hermes_cli/test_kanban_parent_reopen_invalidation.py @@ -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