diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 4e975bb3e8d7..b4ce32f3fe79 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -91,6 +91,11 @@ def _run_state_kwargs(args: argparse.Namespace) -> Optional[dict[str, str]]: return {"state_type": st, "state_name": sn} +def _infer_subscribe_from_env() -> Optional[dict[str, str]]: + """Delegate to the canonical env-var resolution in ``kanban_db``.""" + return kb._resolve_subscribe_from_env() + + def _parse_workspace_flag(value: str) -> tuple[str, Optional[str]]: """Parse ``--workspace`` into ``(kind, path|None)``. @@ -341,6 +346,14 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu "two retries. Omit to use the dispatcher's " "kanban.failure_limit config " f"(default {kb.DEFAULT_FAILURE_LIMIT}).") + p_create.add_argument("--subscribe-platform", default=None, + help="Auto-subscribe task notifications for this platform") + p_create.add_argument("--subscribe-chat-id", default=None, + help="Chat ID for the notification subscription") + p_create.add_argument("--subscribe-thread-id", default=None, + help="Optional thread/topic ID for the notification subscription") + p_create.add_argument("--subscribe-user-id", default=None, + help="Optional user ID for the notification subscription") p_create.add_argument("--initial-status", choices=sorted(kb.VALID_INITIAL_STATUSES), default="running", @@ -550,6 +563,13 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu p_unblock = sub.add_parser("unblock", help="Return one or more blocked/scheduled tasks to ready") p_unblock.add_argument("task_ids", nargs="+") + p_rerun = sub.add_parser("rerun", help="Reset a completed/blocked task for another attempt") + p_rerun.add_argument("task_id") + p_rerun.add_argument("--reason", default=None, + help="Optional reason recorded on the rerun event") + p_rerun.add_argument("--reassign-to", dest="new_assignee", default=None, + help="Optional assignee profile to use for the rerun") + p_archive = sub.add_parser("archive", help="Archive one or more tasks") p_archive.add_argument("task_ids", nargs="*", help="Task ids to archive (default mode)") @@ -899,6 +919,7 @@ def _restore_board_env() -> None: "block": _cmd_block, "schedule": _cmd_schedule, "unblock": _cmd_unblock, + "rerun": _cmd_rerun, "archive": _cmd_archive, "tail": _cmd_tail, "dispatch": _cmd_dispatch, @@ -1286,6 +1307,28 @@ def _cmd_create(args: argparse.Namespace) -> int: file=sys.stderr, ) return 2 + subscribe = None + sub_platform = (getattr(args, "subscribe_platform", None) or "").strip() + sub_chat_id = (getattr(args, "subscribe_chat_id", None) or "").strip() + sub_thread_id = (getattr(args, "subscribe_thread_id", None) or "").strip() + sub_user_id = (getattr(args, "subscribe_user_id", None) or "").strip() + if any((sub_platform, sub_chat_id, sub_thread_id, sub_user_id)): + if not sub_platform or not sub_chat_id: + print( + "kanban: --subscribe-platform and --subscribe-chat-id must be passed together", + file=sys.stderr, + ) + return 2 + subscribe = { + "platform": sub_platform, + "chat_id": sub_chat_id, + } + if sub_thread_id: + subscribe["thread_id"] = sub_thread_id + if sub_user_id: + subscribe["user_id"] = sub_user_id + if subscribe is None: + subscribe = _infer_subscribe_from_env() with kb.connect() as conn: task_id = kb.create_task( conn, @@ -1304,6 +1347,7 @@ def _cmd_create(args: argparse.Namespace) -> int: max_runtime_seconds=max_runtime, skills=getattr(args, "skills", None) or None, max_retries=max_retries, + subscribe=subscribe, initial_status=getattr(args, "initial_status", "running"), ) task = kb.get_task(conn, task_id) @@ -1955,6 +1999,26 @@ def _cmd_unblock(args: argparse.Namespace) -> int: return 0 if not failed else 1 +def _cmd_rerun(args: argparse.Namespace) -> int: + with kb.connect() as conn: + ok = kb.rerun_task( + conn, + args.task_id, + reason=getattr(args, "reason", None), + new_assignee=getattr(args, "new_assignee", None), + ) + if not ok: + print( + f"cannot rerun {args.task_id} (not completed/blocked/archived?)", + file=sys.stderr, + ) + return 1 + task = kb.get_task(conn, args.task_id) + status = task.status if task else "?" + print(f"Rerun {args.task_id} ({status})") + return 0 + + def _cmd_archive(args: argparse.Namespace) -> int: ids = list(args.task_ids or []) purge_ids = list(getattr(args, "purge_ids", None) or []) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 7a30b70987f6..28ed0c796250 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1391,6 +1391,34 @@ def _canonical_assignee(assignee: Optional[str]) -> Optional[str]: return normalize_profile_name(assignee) +def _resolve_subscribe_from_env() -> Optional[dict[str, str]]: + """Build a ``subscribe`` dict from ``HERMES_NOTIFY_*`` env vars. + + Returns ``None`` when ``HERMES_NOTIFY_PLATFORM`` or + ``HERMES_NOTIFY_CHAT_ID`` is missing / empty. Optional fields + (``thread_id``, ``user_id``) are included only when non-empty. + + This is the single source of truth for env-var → subscribe + resolution, shared by :func:`create_task` and the CLI + ``kanban create`` handler. + """ + platform = (os.environ.get("HERMES_NOTIFY_PLATFORM") or "").strip() + chat_id = (os.environ.get("HERMES_NOTIFY_CHAT_ID") or "").strip() + if not platform or not chat_id: + return None + subscribe: dict[str, str] = { + "platform": platform, + "chat_id": chat_id, + } + thread_id = (os.environ.get("HERMES_NOTIFY_THREAD_ID") or "").strip() + user_id = (os.environ.get("HERMES_NOTIFY_USER_ID") or "").strip() + if thread_id: + subscribe["thread_id"] = thread_id + if user_id: + subscribe["user_id"] = user_id + return subscribe + + def create_task( conn: sqlite3.Connection, *, @@ -1409,6 +1437,7 @@ def create_task( max_runtime_seconds: Optional[int] = None, skills: Optional[Iterable[str]] = None, max_retries: Optional[int] = None, + subscribe: Optional[dict] = None, initial_status: str = "running", session_id: Optional[str] = None, board: Optional[str] = None, @@ -1436,6 +1465,13 @@ def create_task( ``kanban-worker``. Use this to pin a task to a specialist skill (e.g. ``skills=["translation"]`` so the worker loads the translation skill regardless of the profile's default config). + + ``subscribe`` optionally creates a notification subscription in the + same write transaction as the task row. Expected keys: ``platform`` + and ``chat_id`` (required), plus optional ``thread_id`` and + ``user_id``. When ``subscribe`` is omitted, the gateway-provided + ``HERMES_NOTIFY_PLATFORM`` / ``HERMES_NOTIFY_CHAT_ID`` env vars are + used as a best-effort fallback. """ assignee = _canonical_assignee(assignee) if not title or not title.strip(): @@ -1526,6 +1562,25 @@ def create_task( if board_default: workspace_path = str(board_default) + # Resolve notification subscription from explicit arg or env vars. + resolved_subscribe: Optional[dict[str, str]] = None + if not subscribe: + subscribe = _resolve_subscribe_from_env() + if isinstance(subscribe, dict): + platform = str(subscribe.get("platform") or "").strip() + chat_id = str(subscribe.get("chat_id") or "").strip() + if platform and chat_id: + resolved_subscribe = { + "platform": platform, + "chat_id": chat_id, + } + thread_id = str(subscribe.get("thread_id") or "").strip() + if thread_id: + resolved_subscribe["thread_id"] = thread_id + user_id = str(subscribe.get("user_id") or "").strip() + if user_id: + resolved_subscribe["user_id"] = user_id + # Retry once on the extremely unlikely id collision. for attempt in range(2): task_id = _new_task_id() @@ -1610,6 +1665,15 @@ def create_task( "skills": list(skills_list) if skills_list else None, }, ) + if resolved_subscribe: + _add_notify_sub_inner( + conn, + task_id=task_id, + platform=resolved_subscribe["platform"], + chat_id=resolved_subscribe["chat_id"], + thread_id=resolved_subscribe.get("thread_id"), + user_id=resolved_subscribe.get("user_id"), + ) return task_id except sqlite3.IntegrityError: if attempt == 1: @@ -3139,6 +3203,126 @@ def unblock_task(conn: sqlite3.Connection, task_id: str) -> bool: return True +def _parent_counts_as_done_for_rerun( + conn: sqlite3.Connection, + parent_id: str, +) -> bool: + """Return True when ``parent_id`` should satisfy rerun parent gates.""" + row = conn.execute( + "SELECT status FROM tasks WHERE id = ?", + (parent_id,), + ).fetchone() + if row is None: + return False + status = row["status"] + if status in ("done", "archived"): + return True + if status != "blocked": + return False + event = conn.execute( + "SELECT kind, payload FROM task_events " + "WHERE task_id = ? AND kind IN ('blocked', 'unblocked') " + "ORDER BY id DESC LIMIT 1", + (parent_id,), + ).fetchone() + if not event or event["kind"] != "blocked" or not event["payload"]: + return False + try: + payload = json.loads(event["payload"]) + except Exception: + payload = None + reason = payload.get("reason") if isinstance(payload, dict) else None + return isinstance(reason, str) and reason.startswith("review-required") + + +def rerun_task( + conn: sqlite3.Connection, + task_id: str, + *, + reason: Optional[str] = None, + new_assignee: Optional[str] = None, +) -> bool: + """Reset a completed/blocked task to ready for another attempt. + + Clears claim state, the active run pointer, and the failure counter. + Parent gates are re-evaluated: undone parents send the task to + ``todo``; ``review-required`` blocked parents count as satisfied for + this rerun decision. + + Returns True when the task was reset, False when the task is not in + a rerunnable state. + """ + with write_txn(conn): + task = get_task(conn, task_id) + if not task: + return False + if task.status not in ("done", "blocked", "archived", "gave_up"): + return False + + parent_rows = conn.execute( + "SELECT parent_id FROM task_links WHERE child_id = ? ORDER BY parent_id", + (task_id,), + ).fetchall() + new_status = "ready" + for row in parent_rows: + if not _parent_counts_as_done_for_rerun(conn, row["parent_id"]): + new_status = "todo" + break + + # Defensive invariant recovery: terminal tasks should not carry + # an open run pointer, but clear it safely if some external path + # leaked one. + if task.current_run_id is not None: + _end_run( + conn, + task_id, + outcome="reclaimed", + status="reclaimed", + summary="invariant recovery on rerun", + ) + + assignee = _canonical_assignee(new_assignee) if new_assignee is not None else task.assignee + cur = conn.execute( + """ + UPDATE tasks + SET assignee = ?, + status = ?, + started_at = NULL, + completed_at = NULL, + claim_lock = NULL, + claim_expires = NULL, + worker_pid = NULL, + last_heartbeat_at = NULL, + consecutive_failures = 0, + last_failure_error = NULL, + current_run_id = NULL + WHERE id = ? + """, + (assignee, new_status, task_id), + ) + if cur.rowcount != 1: + return False + _append_event( + conn, + task_id, + "rerun", + { + "reason": reason, + "status": new_status, + "assignee": assignee, + }, + ) + max_event_id = conn.execute( + "SELECT COALESCE(MAX(id), 0) AS max_id FROM task_events WHERE task_id = ?", + (task_id,), + ).fetchone()["max_id"] + conn.execute( + "UPDATE kanban_notify_subs SET last_event_id = ? WHERE task_id = ?", + (int(max_event_id), task_id), + ) + return True + + def specify_triage_task( conn: sqlite3.Connection, task_id: str, @@ -5786,6 +5970,34 @@ def task_age(task: Task) -> dict: # Notification subscriptions (used by the gateway kanban-notifier) # --------------------------------------------------------------------------- +def _add_notify_sub_inner( + conn: sqlite3.Connection, + *, + task_id: str, + platform: str, + chat_id: str, + thread_id: Optional[str] = None, + user_id: Optional[str] = None, + notifier_profile: Optional[str] = None, +) -> None: + """Insert/update a notify sub — assumes caller already holds a write_txn.""" + now = int(time.time()) + conn.execute( + """INSERT OR IGNORE INTO kanban_notify_subs + (task_id, platform, chat_id, thread_id, user_id, notifier_profile, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (task_id, platform, chat_id, thread_id or "", user_id, notifier_profile, now), + ) + if notifier_profile: + conn.execute( + """UPDATE kanban_notify_subs + SET notifier_profile = ? + WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? + AND (notifier_profile IS NULL OR notifier_profile = '')""", + (notifier_profile, task_id, platform, chat_id, thread_id or ""), + ) + + def add_notify_sub( conn: sqlite3.Connection, *, @@ -5798,28 +6010,16 @@ def add_notify_sub( ) -> None: """Register a gateway source that wants terminal-state notifications for ``task_id``. Idempotent on (task, platform, chat, thread).""" - now = int(time.time()) with write_txn(conn): - conn.execute( - """ - INSERT OR IGNORE INTO kanban_notify_subs - (task_id, platform, chat_id, thread_id, user_id, notifier_profile, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?) - """, - (task_id, platform, chat_id, thread_id or "", user_id, notifier_profile, now), + _add_notify_sub_inner( + conn, + task_id=task_id, + platform=platform, + chat_id=chat_id, + thread_id=thread_id, + user_id=user_id, + notifier_profile=notifier_profile, ) - if notifier_profile: - # Self-heal legacy rows that predate notifier ownership by - # backfilling only when the existing value is unset. - conn.execute( - """ - UPDATE kanban_notify_subs - SET notifier_profile = ? - WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? - AND (notifier_profile IS NULL OR notifier_profile = '') - """, - (notifier_profile, task_id, platform, chat_id, thread_id or ""), - ) def list_notify_subs( diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 435ef41001a9..9faedc7f77ef 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -1605,6 +1605,82 @@ def test_session_id_compose_with_tenant_filter(kanban_home): assert [t.title for t in rows] == ["match"] +def test_create_task_without_subscribe_creates_no_subs(kanban_home, monkeypatch): + """No notify sub is created when subscribe=None and no env vars set.""" + monkeypatch.delenv("HERMES_NOTIFY_PLATFORM", raising=False) + monkeypatch.delenv("HERMES_NOTIFY_CHAT_ID", raising=False) + monkeypatch.delenv("HERMES_NOTIFY_THREAD_ID", raising=False) + monkeypatch.delenv("HERMES_NOTIFY_USER_ID", raising=False) + with kb.connect() as conn: + tid = kb.create_task(conn, title="no-sub") + subs = kb.list_notify_subs(conn, task_id=tid) + assert subs == [] + + +def test_create_task_with_subscribe_dict_creates_sub(kanban_home): + """Passing subscribe=dict creates a row in kanban_notify_subs.""" + with kb.connect() as conn: + tid = kb.create_task( + conn, title="with-sub", + subscribe={"platform": "telegram", "chat_id": "123"}, + ) + subs = kb.list_notify_subs(conn, task_id=tid) + assert len(subs) == 1 + assert subs[0]["platform"] == "telegram" + assert subs[0]["chat_id"] == "123" + assert subs[0]["task_id"] == tid + + +def test_create_task_with_subscribe_includes_optional_fields(kanban_home): + """thread_id and user_id are persisted when passed in subscribe dict.""" + with kb.connect() as conn: + tid = kb.create_task( + conn, title="with-optional", + subscribe={ + "platform": "discord", + "chat_id": "456", + "thread_id": "thread-1", + "user_id": "user-1", + }, + ) + subs = kb.list_notify_subs(conn, task_id=tid) + assert len(subs) == 1 + assert subs[0]["platform"] == "discord" + assert subs[0]["chat_id"] == "456" + assert subs[0]["thread_id"] == "thread-1" + assert subs[0]["user_id"] == "user-1" + + +def test_create_task_falls_back_to_env_vars(kanban_home, monkeypatch): + """When subscribe=None but HERMES_NOTIFY_PLATFORM/CHAT_ID are set, + the env vars are used as fallback.""" + monkeypatch.setenv("HERMES_NOTIFY_PLATFORM", "telegram") + monkeypatch.setenv("HERMES_NOTIFY_CHAT_ID", "999") + monkeypatch.setenv("HERMES_NOTIFY_THREAD_ID", "topic-x") + with kb.connect() as conn: + tid = kb.create_task(conn, title="env-fallback") + subs = kb.list_notify_subs(conn, task_id=tid) + assert len(subs) == 1 + assert subs[0]["platform"] == "telegram" + assert subs[0]["chat_id"] == "999" + assert subs[0]["thread_id"] == "topic-x" + + +def test_create_task_env_vars_skipped_when_subscribe_passed(kanban_home, monkeypatch): + """Explicit subscribe=dict takes precedence over env vars.""" + monkeypatch.setenv("HERMES_NOTIFY_PLATFORM", "slack") + monkeypatch.setenv("HERMES_NOTIFY_CHAT_ID", "111") + with kb.connect() as conn: + tid = kb.create_task( + conn, title="explicit-override", + subscribe={"platform": "telegram", "chat_id": "222"}, + ) + subs = kb.list_notify_subs(conn, task_id=tid) + assert len(subs) == 1 + assert subs[0]["platform"] == "telegram" + assert subs[0]["chat_id"] == "222" + + # --------------------------------------------------------------------------- # Shared-board path resolution (issue #19348) # diff --git a/tests/hermes_cli/test_kanban_db_rerun.py b/tests/hermes_cli/test_kanban_db_rerun.py new file mode 100644 index 000000000000..60673e2e8509 --- /dev/null +++ b/tests/hermes_cli/test_kanban_db_rerun.py @@ -0,0 +1,420 @@ +"""Tests for Kanban DB rerun_task (commit 97c6f4e42 coverage).""" + +from __future__ import annotations + +import os +import sqlite3 +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + """Isolated HERMES_HOME with an empty kanban DB.""" + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +# --------------------------------------------------------------------------- +# Basic: terminal statuses → ready +# --------------------------------------------------------------------------- + +def test_rerun_completed_task_resets_to_ready(kanban_home): + """rerun_task on a completed task resets status to ready.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="done-task", assignee="worker") + kb.claim_task(conn, t) + with kb.connect() as conn: + assert kb.complete_task(conn, t, result="all good") + assert kb.get_task(conn, t).status == "done" + + with kb.connect() as conn: + assert kb.rerun_task(conn, t) + task = kb.get_task(conn, t) + + assert task.status == "ready" + assert task.completed_at is None + assert task.started_at is None + assert task.claim_lock is None + assert task.claim_expires is None + assert task.worker_pid is None + assert task.consecutive_failures == 0 + assert task.last_failure_error is None + assert task.current_run_id is None + + +def test_rerun_blocked_task_resets_to_ready(kanban_home): + """rerun_task on a blocked task resets status to ready.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="blocked-task", assignee="worker") + kb.claim_task(conn, t) + assert kb.block_task(conn, t, reason="need input") + assert kb.get_task(conn, t).status == "blocked" + + with kb.connect() as conn: + assert kb.rerun_task(conn, t) + task = kb.get_task(conn, t) + + assert task.status == "ready" + + +def test_rerun_archived_task_resets_to_ready(kanban_home): + """rerun_task on an archived task resets status to ready.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="archived-task", assignee="worker") + kb.claim_task(conn, t) + assert kb.complete_task(conn, t, result="ok") + assert kb.archive_task(conn, t) + assert kb.get_task(conn, t).status == "archived" + + with kb.connect() as conn: + assert kb.rerun_task(conn, t) + task = kb.get_task(conn, t) + + assert task.status == "ready" + + +# --------------------------------------------------------------------------- +# Edge cases: non-rerunnable inputs +# --------------------------------------------------------------------------- + +def test_rerun_non_terminal_status_returns_false(kanban_home): + """rerun_task returns False for running / ready / todo tasks.""" + with kb.connect() as conn: + # Create and test a fresh ready task first + t_ready = kb.create_task(conn, title="fresh-ready", assignee="w") + assert not kb.rerun_task(conn, t_ready), "Ready task should not be rerunnable" + + # Create a running task + t_running = kb.create_task(conn, title="running-task", assignee="w") + kb.claim_task(conn, t_running) + assert kb.get_task(conn, t_running).status == "running" + assert not kb.rerun_task(conn, t_running), "Running task should not be rerunnable" + + # Create a todo task (has undone parent) + p = kb.create_task(conn, title="p-todo", assignee="w") + t_todo = kb.create_task(conn, title="todo-task", parents=[p], assignee="w") + assert kb.get_task(conn, t_todo).status == "todo" + assert not kb.rerun_task(conn, t_todo), "Todo task should not be rerunnable" + + +def test_rerun_nonexistent_task_returns_false(kanban_home): + """rerun_task returns False for a bogus task_id.""" + with kb.connect() as conn: + assert not kb.rerun_task(conn, "t_nonexistent_999") + + +# --------------------------------------------------------------------------- +# Parent gate logic +# --------------------------------------------------------------------------- + +def test_rerun_with_done_parents_goes_to_ready(kanban_home): + """When all parents are done, rerun promotes to ready.""" + with kb.connect() as conn: + p = kb.create_task(conn, title="parent", assignee="w") + kb.claim_task(conn, p) + kb.complete_task(conn, p, result="ok") + c = kb.create_task(conn, title="child", parents=[p], assignee="w") + assert kb.get_task(conn, c).status == "ready" + kb.claim_task(conn, c) + kb.complete_task(conn, c, result="ok") + assert kb.get_task(conn, c).status == "done" + + with kb.connect() as conn: + assert kb.rerun_task(conn, c) + child = kb.get_task(conn, c) + + assert child.status == "ready" + + +def test_rerun_with_undone_parents_goes_to_todo(kanban_home): + """When a parent is NOT done, rerun demotes to todo.""" + with kb.connect() as conn: + p = kb.create_task(conn, title="parent-undone", assignee="w") + c = kb.create_task(conn, title="child-undone", parents=[p], assignee="w") + assert kb.get_task(conn, c).status == "todo" + # Force child to done via direct update (simulate external completion) + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (c,), + ) + conn.commit() + assert kb.get_task(conn, c).status == "done" + + with kb.connect() as conn: + assert kb.rerun_task(conn, c) + child = kb.get_task(conn, c) + + assert child.status == "todo" + + +def test_rerun_with_review_required_blocked_parent_counts_as_done(kanban_home): + """Parent blocked with 'review-required' reason satisfies the gate.""" + with kb.connect() as conn: + p = kb.create_task(conn, title="parent-review", assignee="w") + kb.claim_task(conn, p) + kb.block_task(conn, p, reason="review-required: needs eyes on PR") + c = kb.create_task(conn, title="child-review", parents=[p], assignee="w") + assert kb.get_task(conn, c).status == "todo" + # Force child to done via direct update + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (c,), + ) + conn.commit() + # Verify parent is blocked with review-required + parent = kb.get_task(conn, p) + assert parent.status == "blocked" + + with kb.connect() as conn: + assert kb.rerun_task(conn, c) + child = kb.get_task(conn, c) + + assert child.status == "ready", ( + "Child with review-required blocked parent should be 'ready'" + ) + + +def test_rerun_with_non_review_blocked_parent_goes_to_todo(kanban_home): + """Parent blocked with a non-review reason does NOT satisfy the gate.""" + with kb.connect() as conn: + p = kb.create_task(conn, title="parent-blocked-other", assignee="w") + kb.claim_task(conn, p) + kb.block_task(conn, p, reason="waiting for API key") + c = kb.create_task( + conn, title="child-blocked-other", parents=[p], assignee="w" + ) + # Force child to done + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (c,), + ) + conn.commit() + + with kb.connect() as conn: + assert kb.rerun_task(conn, c) + child = kb.get_task(conn, c) + + assert child.status == "todo", ( + "Child with non-review blocked parent should be 'todo'" + ) + + +# --------------------------------------------------------------------------- +# Notification subscription retention +# --------------------------------------------------------------------------- + +def test_rerun_preserves_notification_subscription(kanban_home): + """Notification subs persist across rerun — last_event_id is refreshed.""" + with kb.connect() as conn: + t = kb.create_task( + conn, + title="sub-task", + assignee="worker", + subscribe={"platform": "telegram", "chat_id": "123"}, + ) + kb.claim_task(conn, t) + kb.complete_task(conn, t, result="ok") + + # Capture the sub row before rerun + subs_before = kb.list_notify_subs(conn, task_id=t) + assert len(subs_before) == 1 + assert subs_before[0]["platform"] == "telegram" + assert subs_before[0]["chat_id"] == "123" + + with kb.connect() as conn: + assert kb.rerun_task(conn, t) + + # Subscription must still exist + subs_after = kb.list_notify_subs(conn, task_id=t) + assert len(subs_after) == 1, "Subscription should survive rerun" + assert subs_after[0]["platform"] == "telegram" + assert subs_after[0]["chat_id"] == "123" + assert subs_after[0]["task_id"] == t + + +# --------------------------------------------------------------------------- +# Claim + failure state clearing +# --------------------------------------------------------------------------- + +def test_rerun_clears_claim_and_failure_state(kanban_home): + """rerun_task zeros out consecutive_failures and last_failure_error.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="failure-task", assignee="worker") + kb.claim_task(conn, t) + kb.complete_task(conn, t, result="ok") + + # Inject artificial failure state + conn.execute( + "UPDATE tasks SET consecutive_failures = 3, " + "last_failure_error = 'something broke' WHERE id = ?", + (t,), + ) + conn.commit() + task_before = kb.get_task(conn, t) + assert task_before.consecutive_failures == 3 + assert task_before.last_failure_error == "something broke" + + with kb.connect() as conn: + assert kb.rerun_task(conn, t) + task = kb.get_task(conn, t) + + assert task.consecutive_failures == 0 + assert task.last_failure_error is None + + +# --------------------------------------------------------------------------- +# Assignee override +# --------------------------------------------------------------------------- + +def test_rerun_with_new_assignee(kanban_home): + """rerun_task accepts an optional new_assignee.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="reassign-task", assignee="alice") + kb.claim_task(conn, t) + kb.complete_task(conn, t, result="done") + + with kb.connect() as conn: + assert kb.rerun_task(conn, t, new_assignee="bob") + task = kb.get_task(conn, t) + + assert task.status == "ready" + assert task.assignee == "bob" + + +def test_rerun_keeps_original_assignee_when_new_not_given(kanban_home): + """rerun_task preserves the original assignee when new_assignee is None.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="keep-assignee", assignee="alice") + kb.claim_task(conn, t) + kb.complete_task(conn, t, result="done") + + with kb.connect() as conn: + assert kb.rerun_task(conn, t) + task = kb.get_task(conn, t) + + assert task.status == "ready" + assert task.assignee == "alice" + + +# --------------------------------------------------------------------------- +# Event recording +# --------------------------------------------------------------------------- + +def test_rerun_records_event(kanban_home): + """rerun_task emits a 'rerun' event with reason and status.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="event-task", assignee="worker") + kb.claim_task(conn, t) + kb.complete_task(conn, t, result="done") + + with kb.connect() as conn: + assert kb.rerun_task(conn, t, reason="retry after upstream fix") + + events = conn.execute( + "SELECT kind, payload FROM task_events WHERE task_id = ? ORDER BY id", + (t,), + ).fetchall() + + kinds = [e["kind"] for e in events] + assert "rerun" in kinds, f"Expected 'rerun' event in {kinds}" + + rerun_event = next(e for e in events if e["kind"] == "rerun") + import json + + payload = json.loads(rerun_event["payload"]) + assert payload["reason"] == "retry after upstream fix" + assert payload["status"] == "ready" + assert payload["assignee"] == "worker" + + +# --------------------------------------------------------------------------- +# Multiple parents (fan-in) +# --------------------------------------------------------------------------- + +def test_rerun_fan_in_all_done_goes_to_ready(kanban_home): + """Fan-in: rerun goes to ready only when ALL parents are done.""" + with kb.connect() as conn: + p1 = kb.create_task(conn, title="p1", assignee="w") + p2 = kb.create_task(conn, title="p2", assignee="w") + kb.claim_task(conn, p1) + kb.complete_task(conn, p1, result="ok") + kb.claim_task(conn, p2) + kb.complete_task(conn, p2, result="ok") + + c = kb.create_task(conn, title="fanin-child", parents=[p1, p2], assignee="w") + assert kb.get_task(conn, c).status == "ready" + kb.claim_task(conn, c) + kb.complete_task(conn, c, result="ok") + assert kb.get_task(conn, c).status == "done" + + with kb.connect() as conn: + assert kb.rerun_task(conn, c) + child = kb.get_task(conn, c) + + assert child.status == "ready" + + +def test_rerun_fan_in_one_undone_goes_to_todo(kanban_home): + """Fan-in: rerun goes to todo when ANY parent is undone.""" + with kb.connect() as conn: + p1 = kb.create_task(conn, title="p1-done", assignee="w") + kb.claim_task(conn, p1) + kb.complete_task(conn, p1, result="ok") + + p2 = kb.create_task(conn, title="p2-undone", assignee="w") + + c = kb.create_task( + conn, title="fanin-partial", parents=[p1, p2], assignee="w" + ) + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (c,), + ) + conn.commit() + + with kb.connect() as conn: + assert kb.rerun_task(conn, c) + child = kb.get_task(conn, c) + + assert child.status == "todo" + + +# --------------------------------------------------------------------------- +# Invariant: stale current_run_id cleaned up +# --------------------------------------------------------------------------- + +def test_rerun_cleans_stale_current_run_id(kanban_home): + """If a terminal task somehow has current_run_id set, rerun cleans it.""" + with kb.connect() as conn: + t = kb.create_task(conn, title="leaky-task", assignee="worker") + kb.claim_task(conn, t) + # Inject a synthetic run so there's a current_run_id + run_id = conn.execute( + "INSERT INTO task_runs (task_id, profile, status, started_at) " + "VALUES (?, 'worker', 'running', ?)", + (t, int(__import__("time").time())), + ).lastrowid + conn.execute("UPDATE tasks SET current_run_id = ? WHERE id = ?", (run_id, t)) + conn.commit() + # Now complete the task — but current_run_id persists (this is the "leak") + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (t,), + ) + conn.commit() + task_before = kb.get_task(conn, t) + assert task_before.current_run_id is not None, "Precondition: leaked run id" + + with kb.connect() as conn: + assert kb.rerun_task(conn, t) + task = kb.get_task(conn, t) + + assert task.current_run_id is None, "Rerun must clear stale current_run_id" diff --git a/tests/hermes_cli/test_kanban_scratch_gc.py b/tests/hermes_cli/test_kanban_scratch_gc.py new file mode 100644 index 000000000000..6ca16c051ac4 --- /dev/null +++ b/tests/hermes_cli/test_kanban_scratch_gc.py @@ -0,0 +1,313 @@ +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def kanban_conn(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> sqlite3.Connection: + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + + conn = sqlite3.connect(":memory:", isolation_level=None) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys=ON") + conn.executescript(kb.SCHEMA_SQL) + kb._migrate_add_optional_columns(conn) + try: + yield conn + finally: + conn.close() + + +def test_parent_scratch_survives_until_child_done( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + workspace = tmp_path / "parent-scratch" + workspace.mkdir() + + parent = kb.create_task( + kanban_conn, + title="parent", + assignee="lead", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + child = kb.create_task( + kanban_conn, + title="child", + assignee="worker", + parents=[parent], + ) + + assert kb.complete_task(kanban_conn, parent, result="parent complete") + assert kb.get_task(kanban_conn, child).status == "ready" + assert workspace.is_dir() + + +def test_parent_scratch_cleaned_after_last_child_done( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + workspace = tmp_path / "parent-scratch-last-child" + workspace.mkdir() + + parent = kb.create_task( + kanban_conn, + title="parent", + assignee="lead", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + child = kb.create_task( + kanban_conn, + title="child", + assignee="worker", + parents=[parent], + ) + + assert kb.complete_task(kanban_conn, parent, result="parent complete") + assert workspace.is_dir() + assert kb.complete_task(kanban_conn, child, result="child complete") + assert not workspace.exists() + + +# --------------------------------------------------------------------------- +# Multi-child deferred cleanup (commit 03e231971) +# --------------------------------------------------------------------------- + +def test_parent_survives_when_one_of_many_children_still_active( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + """Parent scratch workspace survives when at least one child is active.""" + workspace = tmp_path / "parent-multi-child-active" + workspace.mkdir() + + parent = kb.create_task( + kanban_conn, + title="orchestrator", + assignee="lead", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + child_a = kb.create_task( + kanban_conn, + title="worker-a", + assignee="alice", + parents=[parent], + ) + child_b = kb.create_task( + kanban_conn, + title="worker-b", + assignee="bob", + parents=[parent], + ) + child_c = kb.create_task( + kanban_conn, + title="worker-c", + assignee="carol", + parents=[parent], + ) + + assert kb.complete_task(kanban_conn, parent, result="orchestration done") + + # Complete child A and C, but leave B in todo (still active) + conn = kanban_conn + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (child_a,), + ) + conn.commit() + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (child_c,), + ) + conn.commit() + + # Simulate completing A (calls _cleanup_workspace on parent) + # A is done → parent cleanup deferred because B is still active + assert workspace.is_dir(), "Workspace must survive with active child B" + + # Now complete child B — last active child + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (child_b,), + ) + conn.commit() + # Simulate completing B → triggers parent cleanup + kb._cleanup_workspace(kanban_conn, parent) + + assert not workspace.exists(), "Workspace cleaned after last child done" + + +def test_parent_cleaned_when_all_children_done( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + """Parent scratch workspace is removed when ALL children are done.""" + workspace = tmp_path / "parent-all-children-done" + workspace.mkdir() + + parent = kb.create_task( + kanban_conn, + title="orchestrator", + assignee="lead", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + child_a = kb.create_task( + kanban_conn, + title="worker-a", + assignee="alice", + parents=[parent], + ) + child_b = kb.create_task( + kanban_conn, + title="worker-b", + assignee="bob", + parents=[parent], + ) + + assert kb.complete_task(kanban_conn, parent, result="orchestration done") + + # Mark both children as done + conn = kanban_conn + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (child_a,), + ) + conn.commit() + conn.execute( + "UPDATE tasks SET status='done', started_at=0, completed_at=0 WHERE id=?", + (child_b,), + ) + conn.commit() + + # Simulate completing last child → parent cleanup + kb._cleanup_workspace(kanban_conn, parent) + + assert not workspace.exists(), "Workspace cleaned when all children done" + + +def test_parent_survives_child_in_archived_state( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + """Archived children do NOT count as active — parent can be cleaned.""" + workspace = tmp_path / "parent-archived-child" + workspace.mkdir() + + parent = kb.create_task( + kanban_conn, + title="orchestrator", + assignee="lead", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + child = kb.create_task( + kanban_conn, + title="worker", + assignee="alice", + parents=[parent], + ) + + assert kb.complete_task(kanban_conn, parent, result="orchestration done") + + # Archive the child (archived = terminal, not active) + conn = kanban_conn + conn.execute( + "UPDATE tasks SET status='archived', started_at=0, completed_at=0 WHERE id=?", + (child,), + ) + conn.commit() + + kb._cleanup_workspace(kanban_conn, parent) + assert not workspace.exists(), "Workspace cleaned when child is archived" + + +def test_parent_survives_child_in_gave_up_state( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + """Children in 'gave_up' status do NOT count as active.""" + workspace = tmp_path / "parent-gave-up-child" + workspace.mkdir() + + parent = kb.create_task( + kanban_conn, + title="orchestrator", + assignee="lead", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + child = kb.create_task( + kanban_conn, + title="worker", + assignee="alice", + parents=[parent], + ) + + assert kb.complete_task(kanban_conn, parent, result="orchestration done") + + conn = kanban_conn + conn.execute( + "UPDATE tasks SET status='gave_up', started_at=0, completed_at=0 WHERE id=?", + (child,), + ) + conn.commit() + + kb._cleanup_workspace(kanban_conn, parent) + assert not workspace.exists(), "Workspace cleaned when child gave up" + + +def test_dir_workspace_not_cleaned( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + """``dir`` workspaces are never cleaned — only ``scratch``.""" + workspace = tmp_path / "persistent-workspace" + workspace.mkdir() + marker = workspace / "important.txt" + marker.write_text("preserve me") + + parent = kb.create_task( + kanban_conn, + title="persistent-parent", + assignee="lead", + workspace_kind="dir", + workspace_path=str(workspace), + ) + + assert kb.complete_task(kanban_conn, parent, result="done") + kb._cleanup_workspace(kanban_conn, parent) + + assert workspace.is_dir(), "dir workspace must survive" + assert marker.read_text() == "preserve me" + + +def test_scratch_cleanup_no_active_children( + kanban_conn: sqlite3.Connection, + tmp_path: Path, +) -> None: + """Scratch workspace cleaned immediately when there are no children.""" + workspace = tmp_path / "solo-scratch" + workspace.mkdir() + + task = kb.create_task( + kanban_conn, + title="solo-task", + assignee="worker", + workspace_kind="scratch", + workspace_path=str(workspace), + ) + + assert kb.complete_task(kanban_conn, task, result="done") + assert not workspace.exists(), "Solo scratch workspace cleaned on completion"