diff --git a/contributors/emails/andrexibiza@gmail.com b/contributors/emails/andrexibiza@gmail.com new file mode 100644 index 000000000000..efa930813a29 --- /dev/null +++ b/contributors/emails/andrexibiza@gmail.com @@ -0,0 +1 @@ +andrexibiza diff --git a/contributors/emails/andrexibiza@users.noreply.github.com b/contributors/emails/andrexibiza@users.noreply.github.com new file mode 100644 index 000000000000..efa930813a29 --- /dev/null +++ b/contributors/emails/andrexibiza@users.noreply.github.com @@ -0,0 +1 @@ +andrexibiza diff --git a/hermes_cli/failure_accounting.py b/hermes_cli/failure_accounting.py new file mode 100644 index 000000000000..d09a1dbdf845 --- /dev/null +++ b/hermes_cli/failure_accounting.py @@ -0,0 +1,246 @@ +"""Failure accounting for the kanban dispatcher circuit breaker. + +Extracted verbatim from ``hermes_cli.kanban_db`` (wave 1, shard s4, +cluster c10 / failure_accounting). ``_record_task_failure`` is the unified +non-success bookkeeper; the old spawn-only name and the +``_clear_spawn_failures`` alias are preserved for back-compat. +""" +from __future__ import annotations + +import sqlite3 +from typing import Optional + +# DEFAULT_FAILURE_LIMIT and the txn/run/event helpers stay in kanban_db.py; +# imported at the bottom to avoid a circular import. + +def _record_task_failure( + conn: sqlite3.Connection, + task_id: str, + error: str, + *, + outcome: str, + failure_limit: int = None, + force_trip: bool = False, + release_claim: bool = False, + end_run: bool = False, + event_payload_extra: Optional[dict] = None, +) -> bool: + """Record a non-success outcome (spawn_failed / crashed / timed_out) + and maybe trip the circuit breaker. + + Unified replacement for the old spawn-only ``_record_spawn_failure``. + Every path that ends a task with a non-success outcome funnels + through here so the ``consecutive_failures`` counter and the + auto-block threshold stay consistent. + + Returns True when the task was auto-blocked (counter reached + ``failure_limit``), False when it was just updated in place. + + Modes: + + * ``release_claim=True, end_run=True`` — spawn-failure path. + Caller has a running task with an open run; this transitions + it back to ``ready`` (or ``blocked`` when the breaker trips), + releases the claim, and closes the run with ``outcome=``. + + * ``release_claim=False, end_run=False`` — timeout/crash path. + Caller has ALREADY flipped the task to ``ready`` and closed the + run with the appropriate outcome. This just increments the + counter; if the breaker trips, the task is re-transitioned + ``ready → blocked`` and a ``gave_up`` event is emitted. + + ``event_payload_extra`` merges into the ``gave_up`` event payload + when the breaker trips, so callers can include outcome-specific + context (e.g. pid on crash, elapsed on timeout). + + Resolution order for the effective threshold: + 1. per-task ``max_retries`` if set (nothing else overrides) + 2. caller-supplied ``failure_limit`` (gateway passes the config + value from ``kanban.failure_limit``; tests pass fixed values) + 3. ``DEFAULT_FAILURE_LIMIT`` + + ``force_trip=True`` trips the breaker unconditionally, skipping the + counter-vs-threshold comparison (the resolution order above is then + only reported in the ``gave_up`` payload, not re-evaluated). Callers + use it when they have already applied their own bounded-retry policy + — e.g. the clean-exit protocol-violation streak in + ``detect_crashed_workers``, which resolves the per-task + ``max_retries`` override against the violation streak itself. The + failure is still counted into ``consecutive_failures``. + """ + if failure_limit is None: + failure_limit = DEFAULT_FAILURE_LIMIT + blocked = False + with write_txn(conn): + row = conn.execute( + "SELECT consecutive_failures, status, max_retries " + "FROM tasks WHERE id = ?", (task_id,), + ).fetchone() + if row is None: + return False + failures = int(row["consecutive_failures"]) + 1 + + # Per-task override wins over both caller-supplied and default + # thresholds. None (the common case) falls through. + task_override = ( + row["max_retries"] if "max_retries" in row.keys() else None + ) + if task_override is not None: + effective_limit = int(task_override) + limit_source = "task" + else: + effective_limit = int(failure_limit) + limit_source = "dispatcher" + + if force_trip or failures >= effective_limit: + # Trip the breaker. + if release_claim: + # Spawn path: still running, also clear claim state. + conn.execute( + "UPDATE tasks SET status = 'blocked', claim_lock = NULL, " + "claim_expires = NULL, worker_pid = NULL, " + "consecutive_failures = ?, last_failure_error = ? " + "WHERE id = ? AND status IN ('running', 'ready')", + (failures, error[:500], task_id), + ) + else: + # Timeout/crash path: task is already at ``ready`` + # with claim cleared; just flip to blocked + update + # counter fields. + conn.execute( + "UPDATE tasks SET status = 'blocked', " + "consecutive_failures = ?, last_failure_error = ? " + "WHERE id = ? AND status IN ('ready', 'running')", + (failures, error[:500], task_id), + ) + run_id = None + if end_run: + # Only the spawn path has an open run to close. + run_id = _end_run( + conn, task_id, + outcome="gave_up", status="gave_up", + error=error[:500], + metadata={ + "failures": failures, + "trigger_outcome": outcome, + "effective_limit": effective_limit, + "limit_source": limit_source, + }, + ) + payload = { + "failures": failures, + "effective_limit": effective_limit, + "limit_source": limit_source, + "error": error[:500], + "trigger_outcome": outcome, + } + if event_payload_extra: + payload.update(event_payload_extra) + _append_event( + conn, task_id, "gave_up", payload, run_id=run_id, + ) + blocked = True + else: + # Below threshold. + if release_claim: + # Spawn path: transition running → ready + clear claim. + conn.execute( + "UPDATE tasks SET status = 'ready', claim_lock = NULL, " + "claim_expires = NULL, worker_pid = NULL, " + "consecutive_failures = ?, last_failure_error = ? " + "WHERE id = ? AND status = 'running'", + (failures, error[:500], task_id), + ) + else: + # Timeout/crash path: task is already at ``ready`` via + # its own UPDATE. Just bookkeep the counter + last error. + conn.execute( + "UPDATE tasks SET consecutive_failures = ?, " + "last_failure_error = ? WHERE id = ?", + (failures, error[:500], task_id), + ) + if end_run: + # Spawn path: close the open run with outcome. + run_id = _end_run( + conn, task_id, + outcome=outcome, status=outcome, + error=error[:500], + metadata={"failures": failures}, + ) + _append_event( + conn, task_id, outcome, + {"error": error[:500], "failures": failures}, + run_id=run_id, + ) + # Timeout/crash path's caller already emitted its own event. + return blocked + + +# Backward-compat alias. Old name is referenced from tests and possibly +# third-party callers. New code should call ``_record_task_failure``. +def _record_spawn_failure( + conn: sqlite3.Connection, + task_id: str, + error: str, + *, + failure_limit: int = None, +) -> bool: + return _record_task_failure( + conn, task_id, error, + outcome="spawn_failed", + failure_limit=failure_limit, + release_claim=True, + end_run=True, + ) + + +def _set_worker_pid(conn: sqlite3.Connection, task_id: str, pid: int) -> None: + """Record the spawned child's pid + emit a ``spawned`` event. + + The event's payload carries the pid so a human reading ``hermes kanban + tail`` can correlate log lines with OS-level traces without opening + the drawer. + """ + with write_txn(conn): + conn.execute( + "UPDATE tasks SET worker_pid = ? WHERE id = ?", + (int(pid), task_id), + ) + run_id = _current_run_id(conn, task_id) + if run_id is not None: + conn.execute( + "UPDATE task_runs SET worker_pid = ? WHERE id = ?", + (int(pid), run_id), + ) + _append_event(conn, task_id, "spawned", {"pid": int(pid)}, run_id=run_id) + + +def _clear_failure_counter(conn: sqlite3.Connection, task_id: str) -> None: + """Reset the unified consecutive-failures counter. + + Called from ``complete_task`` on successful completion — a fresh + success means the task + profile combination is working and any + past failures are history. NOT called on spawn success anymore: + a successful spawn proves the worker could start but says nothing + about whether the run will succeed, so we need to let timeouts and + crashes accumulate across spawn boundaries. + """ + with write_txn(conn): + conn.execute( + "UPDATE tasks SET consecutive_failures = 0, " + "last_failure_error = NULL WHERE id = ?", + (task_id,), + ) + + +# Legacy alias for test-code and anything else that still imports it. +_clear_spawn_failures = _clear_failure_counter + + +from hermes_cli.kanban_db import ( # noqa: E402 + DEFAULT_FAILURE_LIMIT, + _append_event, + _current_run_id, + _end_run, + write_txn, +) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 113e34842ec5..76b0889646c2 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -6288,404 +6288,6 @@ def decompose_triage_task( return child_ids -def archive_task(conn: sqlite3.Connection, task_id: str) -> bool: - with write_txn(conn): - cur = conn.execute( - "UPDATE tasks SET status = 'archived', " - " claim_lock = NULL, claim_expires = NULL, worker_pid = NULL " - "WHERE id = ? AND status != 'archived'", - (task_id,), - ) - if cur.rowcount != 1: - return False - # If archive happened while a run was still in flight (e.g. user - # archived a running task from the dashboard), close that run with - # outcome='reclaimed' so attempt history isn't orphaned. - run_id = _end_run( - conn, task_id, - outcome="reclaimed", status="reclaimed", - summary="task archived with run still active", - ) - _append_event(conn, task_id, "archived", None, run_id=run_id) - # ``archived`` parents no longer block children, same as ``done``. - # Promote newly-unblocked dependents immediately instead of waiting - # for a later dispatcher tick. - recompute_ready(conn) - return True - - -def delete_archived_task(conn: sqlite3.Connection, task_id: str) -> bool: - """Permanently remove an already-archived task and its related rows. - - Safety guard: only archived tasks can be deleted. Active / blocked / done - tasks must be explicitly archived first so accidental data loss requires a - second deliberate action. - """ - with write_txn(conn): - row = conn.execute( - "SELECT status FROM tasks WHERE id = ?", - (task_id,), - ).fetchone() - if not row or row["status"] != "archived": - return False - conn.execute( - "DELETE FROM task_links WHERE parent_id = ? OR child_id = ?", - (task_id, task_id), - ) - conn.execute("DELETE FROM task_comments WHERE task_id = ?", (task_id,)) - conn.execute("DELETE FROM task_events WHERE task_id = ?", (task_id,)) - conn.execute("DELETE FROM task_runs WHERE task_id = ?", (task_id,)) - conn.execute("DELETE FROM kanban_notify_subs WHERE task_id = ?", (task_id,)) - cur = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) - return cur.rowcount == 1 - - -def delete_task(conn: sqlite3.Connection, task_id: str) -> bool: - """Hard-delete a task and cascade to all related rows. - - Because the schema does not use ``ON DELETE CASCADE`` foreign keys, - we explicitly delete from child tables first, then the task row. - This keeps the operation atomic (single ``write_txn``). - - Returns ``True`` if the task existed and was deleted, ``False`` - if the task was not found. - """ - with write_txn(conn): - cur = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) - if cur.rowcount != 1: - return False - conn.execute("DELETE FROM task_links WHERE parent_id = ? OR child_id = ?", (task_id, task_id)) - conn.execute("DELETE FROM task_comments WHERE task_id = ?", (task_id,)) - conn.execute("DELETE FROM task_events WHERE task_id = ?", (task_id,)) - conn.execute("DELETE FROM task_runs WHERE task_id = ?", (task_id,)) - conn.execute("DELETE FROM kanban_notify_subs WHERE task_id = ?", (task_id,)) - recompute_ready(conn) - return True - - -# --------------------------------------------------------------------------- -# Workspace resolution -# --------------------------------------------------------------------------- - -def _git_toplevel(path: Path) -> Optional[Path]: - """Return the git toplevel containing ``path``, or ``None`` if not in a repo.""" - try: - result = subprocess.run( - ["git", "-C", str(path), "rev-parse", "--show-toplevel"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=30, - check=False, - ) - except Exception: - return None - if result.returncode != 0: - return None - out = (result.stdout or "").strip() - if not out: - return None - try: - return Path(out).expanduser().resolve() - except Exception: - return Path(out).expanduser() - - -def _git_branch_exists(repo_root: Path, branch_name: str) -> bool: - try: - result = subprocess.run( - ["git", "-C", str(repo_root), "show-ref", "--verify", f"refs/heads/{branch_name}"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=30, - check=False, - ) - except Exception: - return False - return result.returncode == 0 - - -def _git_common_dir(path: Path) -> Optional[Path]: - try: - result = subprocess.run( - ["git", "-C", str(path), "rev-parse", "--path-format=absolute", "--git-common-dir"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=30, - check=False, - ) - except Exception: - return None - if result.returncode != 0: - return None - out = (result.stdout or "").strip() - if not out: - return None - return Path(out).expanduser().resolve(strict=False) - - -def _git_dir(path: Path) -> Optional[Path]: - try: - result = subprocess.run( - ["git", "-C", str(path), "rev-parse", "--path-format=absolute", "--git-dir"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=30, - check=False, - ) - except Exception: - return None - if result.returncode != 0: - return None - out = (result.stdout or "").strip() - if not out: - return None - return Path(out).expanduser().resolve(strict=False) - - -def _git_current_branch(path: Path) -> Optional[str]: - try: - result = subprocess.run( - ["git", "-C", str(path), "branch", "--show-current"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=30, - check=False, - ) - except Exception: - return None - if result.returncode != 0: - return None - branch = (result.stdout or "").strip() - return branch or None - - -def _is_linked_worktree_checkout(path: Path) -> bool: - git_dir = _git_dir(path) - common_dir = _git_common_dir(path) - if git_dir is None or common_dir is None: - return False - return git_dir != common_dir - - -def _nearest_existing_path(path: Path) -> Path: - current = path - while not current.exists() and current != current.parent: - current = current.parent - return current - - -def _repo_root_for_worktree_target(path: Path) -> Optional[Path]: - current = _nearest_existing_path(path).resolve(strict=False) - while True: - repo_root = _git_toplevel(current) - if repo_root is not None: - return repo_root - if current == current.parent: - return None - current = current.parent - - -def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> None: - """Materialize ``target`` as a linked git worktree under ``repo_root``.""" - target = target.expanduser() - repo_common = _git_common_dir(repo_root) - if target.exists() and repo_common is not None: - target_common = _git_common_dir(target) - if target_common == repo_common: - return - target.parent.mkdir(parents=True, exist_ok=True) - if _git_branch_exists(repo_root, branch_name): - cmd = ["git", "-C", str(repo_root), "worktree", "add", str(target), branch_name] - else: - cmd = [ - "git", "-C", str(repo_root), "worktree", "add", "-b", branch_name, - str(target), "HEAD", - ] - result = subprocess.run( - cmd, - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=60, - check=False, - ) - if result.returncode != 0: - stderr = (result.stderr or result.stdout or "").strip() - raise RuntimeError( - f"git worktree add failed for {target} on branch {branch_name}: {stderr}" - ) - - -def _resolve_worktree_workspace( - task: Task, *, board: Optional[str] = None -) -> tuple[Path, str]: - """Resolve + materialize a linked git worktree for ``task``. - - When ``task.workspace_path`` is unset, the anchor is the board's - ``default_workdir`` (a persistent project checkout). This keeps every - worktree task under a meaningful, board-owned repo — ``/.worktrees/ - `` — instead of silently landing under the dispatcher's current - working directory (which is whatever directory the gateway happened to be - launched from, e.g. the Hermes checkout). If no anchor is configured - anywhere, we fail loudly rather than guess. - """ - branch_name = (task.branch_name or "").strip() or f"wt/{task.id}" - if not task.workspace_path: - # Anchor on the board's configured default_workdir, not Path.cwd(). - # The dispatcher's CWD is incidental (gateway launch dir) and using it - # scatters worktrees under whatever repo the gateway started in. - board_slug = board if board else get_current_board() - board_default = (read_board_metadata(board_slug).get("default_workdir") or "").strip() - if not board_default: - raise ValueError( - f"task {task.id} has workspace_kind=worktree but no workspace_path, " - f"and board {board_slug!r} has no default_workdir set. Set a board " - "default workdir (a git repo) or create the task with " - "--workspace worktree:." - ) - anchor = Path(board_default).expanduser() - if not anchor.is_absolute(): - raise ValueError( - f"board {board_slug!r} default_workdir {board_default!r} is not " - "absolute; use an absolute path to a git repo" - ) - repo_root = _git_toplevel(anchor) - if repo_root is None: - raise ValueError( - f"task {task.id} has workspace_kind=worktree but board " - f"{board_slug!r} default_workdir {board_default!r} is not inside a git repo" - ) - target = repo_root / ".worktrees" / task.id - _ensure_git_worktree(repo_root, target, branch_name) - return target, branch_name - - requested = Path(task.workspace_path).expanduser() - if not requested.is_absolute(): - raise ValueError( - f"task {task.id} has non-absolute worktree path " - f"{task.workspace_path!r}; use an absolute path" - ) - requested_resolved = requested.resolve(strict=False) - - if requested.exists() and _is_linked_worktree_checkout(requested): - actual_branch = _git_current_branch(requested) - if actual_branch == branch_name: - return requested_resolved, actual_branch - # The requested path is an existing checkout of a DIFFERENT - # task's branch. Decompose children inherit the root's - # workspace_path verbatim, so siblings all point here; reusing - # the checkout as-is would run this task on the other task's - # branch — silent cross-task provenance corruption, and unsafe - # when siblings run concurrently. Fall back to a fresh worktree - # of our own under the same repo. - fallback_root = _repo_root_for_worktree_target(requested.parent) - if fallback_root is not None: - fallback = fallback_root / ".worktrees" / task.id - if fallback.resolve(strict=False) != requested_resolved: - _ensure_git_worktree(fallback_root, fallback, branch_name) - return fallback.resolve(strict=False), branch_name - # No repo to anchor a fallback on (or the occupied path IS this - # task's own canonical worktree): keep the legacy reuse rather - # than failing dispatch. - return requested_resolved, actual_branch or branch_name - - repo_root = _git_toplevel(requested) - if repo_root is not None and requested_resolved == repo_root: - target = repo_root / ".worktrees" / task.id - _ensure_git_worktree(repo_root, target, branch_name) - return target, branch_name - - repo_root = _repo_root_for_worktree_target(requested.parent) - if repo_root is None: - raise ValueError( - f"task {task.id} worktree path {task.workspace_path!r} is not inside a git repo " - "and does not point at a git repo root" - ) - _ensure_git_worktree(repo_root, requested, branch_name) - return requested, branch_name - - -def resolve_workspace(task: Task, *, board: Optional[str] = None) -> Path: - """Resolve (and create if needed) the workspace for a task. - - - ``scratch``: a fresh dir under ``/workspaces//``, - where ```` is the active board's root. The path is the - same for the dispatcher and every profile worker, so handoff is - path-stable. - - ``dir:``: the path stored in ``workspace_path``. Created - if missing. MUST be absolute — relative paths are rejected to - prevent confused-deputy traversal where ``../../../tmp/attacker`` - resolves against the dispatcher's CWD instead of a meaningful - root. Users who want a kanban-root-relative workspace should - compute the absolute path themselves. - - ``worktree``: a real linked git worktree. If ``workspace_path`` names - a repo root, Hermes treats it as an anchor and materializes a linked - worktree at ``/.worktrees/``. If ``workspace_path`` names - a concrete target path, Hermes creates/reuses that linked worktree. With - no ``workspace_path``, Hermes anchors on the board's ``default_workdir`` - and materializes ``/.worktrees/`` per task; if no - ``default_workdir`` is configured it raises rather than guessing from the - dispatcher's CWD. When ``branch_name`` is empty, Hermes uses - ``wt/``. - - Persist the resolved path back to the task row via ``set_workspace_path`` - so subsequent runs reuse the same directory. - """ - kind = task.workspace_kind or "scratch" - if kind == "scratch": - if task.workspace_path: - # Legacy scratch tasks that were set to an explicit path get the - # same absolute-path guard as dir: — consistent with the - # threat model. - p = Path(task.workspace_path).expanduser() - if not p.is_absolute(): - raise ValueError( - f"task {task.id} has non-absolute workspace_path " - f"{task.workspace_path!r}; workspace paths must be absolute" - ) - else: - p = workspaces_root(board=board) / task.id - p.mkdir(parents=True, exist_ok=True) - return p - if kind == "dir": - if not task.workspace_path: - raise ValueError( - f"task {task.id} has workspace_kind=dir but no workspace_path" - ) - p = Path(task.workspace_path).expanduser() - if not p.is_absolute(): - raise ValueError( - f"task {task.id} has non-absolute workspace_path " - f"{task.workspace_path!r}; use an absolute path " - f"(relative paths are ambiguous against the dispatcher's CWD)" - ) - p.mkdir(parents=True, exist_ok=True) - return p - if kind == "worktree": - p, _branch_name = _resolve_worktree_workspace(task, board=board) - return p - raise ValueError(f"unknown workspace_kind: {kind}") - - -def set_workspace_path( - conn: sqlite3.Connection, task_id: str, path: Path | str -) -> None: - with write_txn(conn): - conn.execute( - "UPDATE tasks SET workspace_path = ? WHERE id = ?", - (str(path), task_id), - ) - - -def set_branch_name( - conn: sqlite3.Connection, task_id: str, branch_name: str -) -> None: - with write_txn(conn): - conn.execute( - "UPDATE tasks SET branch_name = ? WHERE id = ?", - (str(branch_name), task_id), - ) - - # --------------------------------------------------------------------------- def schedule_task( conn: sqlite3.Connection, @@ -6848,172 +6450,6 @@ class DispatchResult: actively preventing two dispatchers from racing on ``kanban.db``.""" -# Bounded registry of recently-reaped worker child exits, populated by the -# reap loop at the top of ``dispatch_once`` and consulted by -# ``detect_crashed_workers`` to classify a dead-pid task. -# -# Entry: ``pid -> (raw_wait_status, reaped_at_epoch)``. We keep raw status -# so both ``os.WIFEXITED`` / ``os.WEXITSTATUS`` and ``os.WIFSIGNALED`` can -# be consulted. Entries are trimmed by age (and total size cap as a -# belt-and-braces against unbounded growth on exotic platforms). -_RECENT_WORKER_EXIT_TTL_SECONDS = 600 -_RECENT_WORKER_EXITS_MAX = 4096 -_recent_worker_exits: "dict[int, tuple[int, float]]" = {} - - -def _record_worker_exit(pid: int, raw_status: int) -> None: - """Record a reaped child's exit status for later classification. - - Called from the reap loop in ``dispatch_once``. Safe to call many - times; duplicate pids overwrite (pids can cycle, latest wins). - """ - if not pid or pid <= 0: - return - now = time.time() - _recent_worker_exits[int(pid)] = (int(raw_status), now) - # Age-based trim: drop entries older than the TTL. - if len(_recent_worker_exits) > _RECENT_WORKER_EXITS_MAX // 2: - cutoff = now - _RECENT_WORKER_EXIT_TTL_SECONDS - for _pid in [p for p, (_s, t) in _recent_worker_exits.items() if t < cutoff]: - _recent_worker_exits.pop(_pid, None) - # Size cap as a final guard. - if len(_recent_worker_exits) > _RECENT_WORKER_EXITS_MAX: - # Drop oldest half. - ordered = sorted(_recent_worker_exits.items(), key=lambda kv: kv[1][1]) - for _pid, _ in ordered[: len(ordered) // 2]: - _recent_worker_exits.pop(_pid, None) - - -def _classify_worker_exit(pid: int) -> "tuple[str, Optional[int]]": - """Classify a recently-reaped worker by pid. - - Returns ``(kind, code)`` where ``kind`` is one of: - - * ``"clean_exit"`` — ``WIFEXITED`` with ``WEXITSTATUS == 0``. When the - task is still ``running`` in the DB, this is a protocol violation - (worker exited without calling ``kanban_complete`` / ``kanban_block``) - and should be auto-blocked immediately — retrying will just loop. - * ``"rate_limited"`` — ``WIFEXITED`` with status - ``KANBAN_RATE_LIMIT_EXIT_CODE``. The worker bailed because the - provider rate-limited / exhausted quota, NOT because the task failed. - ``detect_crashed_workers`` releases the task back to ``ready`` without - counting a failure, so a long quota window can't trip the breaker. - * ``"nonzero_exit"`` — ``WIFEXITED`` with non-zero status. Real error. - * ``"signaled"`` — ``WIFSIGNALED`` (OOM killer, SIGKILL, etc). Real crash. - * ``"unknown"`` — pid was not in the reap registry (either reaped by - something else, or died between reap tick and liveness check). Fall - back to existing crashed-counter behavior. - - ``code`` is the exit status (for ``clean_exit`` / ``rate_limited`` / - ``nonzero_exit``) or the signal number (for ``signaled``), or ``None`` - for ``unknown``. - """ - entry = _recent_worker_exits.get(int(pid)) - if entry is None: - return ("unknown", None) - raw, _ = entry - try: - if os.WIFEXITED(raw): - code = os.WEXITSTATUS(raw) - if code == 0: - return ("clean_exit", 0) - if code == KANBAN_RATE_LIMIT_EXIT_CODE: - return ("rate_limited", code) - return ("nonzero_exit", code) - if os.WIFSIGNALED(raw): - return ("signaled", os.WTERMSIG(raw)) - except Exception: - pass - return ("unknown", None) - - -def reap_worker_zombies() -> "list[int]": - """Reap all zombie children of this process without blocking. - - Returns the list of reaped PIDs. Safe to call when there are no - children (returns []). No-op on Windows. - """ - reaped: "list[int]" = [] - if os.name != "nt": - try: - while True: - try: - pid, status = os.waitpid(-1, os.WNOHANG) - except ChildProcessError: - break - if pid == 0: - break - _record_worker_exit(pid, status) - reaped.append(pid) - except Exception: - pass - return reaped - - -def _pid_alive(pid: Optional[int]) -> bool: - """Return True if ``pid`` is still running on this host. - - Cross-platform: uses ``OpenProcess`` + ``WaitForSingleObject`` on - Windows (via ``gateway.status._pid_exists``) and ``os.kill(pid, 0)`` - on POSIX. Returns False for falsy PIDs or on any OS error. - - **DO NOT** use ``os.kill(pid, 0)`` directly on Windows — Python's - Windows ``os.kill`` treats ``sig=0`` as ``CTRL_C_EVENT`` (bpo-14484) - and will broadcast it to the target's console group, potentially - killing unrelated processes. - - **Zombie handling:** the existence check succeeds against zombie - processes (post-exit, pre-reap) because the process table entry - still exists. A worker that exits without being reaped by its - parent would stay "alive" to the dispatcher forever. Dispatcher - workers are started via ``start_new_session=True`` + intentional - Popen handle abandonment, so init reaps them quickly — but during - the window between exit and reap, we'd otherwise see stale "alive" - signals. On Linux we peek at ``/proc//status`` and treat - ``State: Z`` as dead. On macOS we ask ``ps`` for the BSD ``stat`` - field and treat values containing ``Z`` as dead. - """ - if not pid or pid <= 0: - return False - from gateway.status import _pid_exists - if not _pid_exists(int(pid)): - return False - # Still here → process exists. Check for zombie on platforms - # where we have a cheap, deterministic process-state probe. - if sys.platform == "linux": - try: - with open(f"/proc/{int(pid)}/status", "r", encoding="utf-8") as f: - for line in f: - if line.startswith("State:"): - # "State:\tZ (zombie)" → dead - if "Z" in line.split(":", 1)[1]: - return False - break - except (FileNotFoundError, PermissionError, OSError): - # proc entry gone → already reaped; treat as dead. - # PermissionError shouldn't happen for our own children but - # be defensive. - pass - elif sys.platform == "darwin": - try: - proc = subprocess.run( - ["ps", "-o", "stat=", "-p", str(int(pid))], - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, encoding='utf-8', errors='replace', - timeout=1, - check=False, - ) - if proc.returncode != 0: - return False - if "Z" in (proc.stdout or "").strip(): - return False - except (OSError, subprocess.SubprocessError, TimeoutError): - # If the secondary probe fails, keep the kill(0) answer. - pass - return True - - def _terminate_reclaimed_worker( pid: Optional[int], claim_lock: Optional[str], @@ -7785,230 +7221,6 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: return crashed -def _record_task_failure( - conn: sqlite3.Connection, - task_id: str, - error: str, - *, - outcome: str, - failure_limit: int = None, - force_trip: bool = False, - release_claim: bool = False, - end_run: bool = False, - event_payload_extra: Optional[dict] = None, -) -> bool: - """Record a non-success outcome (spawn_failed / crashed / timed_out) - and maybe trip the circuit breaker. - - Unified replacement for the old spawn-only ``_record_spawn_failure``. - Every path that ends a task with a non-success outcome funnels - through here so the ``consecutive_failures`` counter and the - auto-block threshold stay consistent. - - Returns True when the task was auto-blocked (counter reached - ``failure_limit``), False when it was just updated in place. - - Modes: - - * ``release_claim=True, end_run=True`` — spawn-failure path. - Caller has a running task with an open run; this transitions - it back to ``ready`` (or ``blocked`` when the breaker trips), - releases the claim, and closes the run with ``outcome=``. - - * ``release_claim=False, end_run=False`` — timeout/crash path. - Caller has ALREADY flipped the task to ``ready`` and closed the - run with the appropriate outcome. This just increments the - counter; if the breaker trips, the task is re-transitioned - ``ready → blocked`` and a ``gave_up`` event is emitted. - - ``event_payload_extra`` merges into the ``gave_up`` event payload - when the breaker trips, so callers can include outcome-specific - context (e.g. pid on crash, elapsed on timeout). - - Resolution order for the effective threshold: - 1. per-task ``max_retries`` if set (nothing else overrides) - 2. caller-supplied ``failure_limit`` (gateway passes the config - value from ``kanban.failure_limit``; tests pass fixed values) - 3. ``DEFAULT_FAILURE_LIMIT`` - - ``force_trip=True`` trips the breaker unconditionally, skipping the - counter-vs-threshold comparison (the resolution order above is then - only reported in the ``gave_up`` payload, not re-evaluated). Callers - use it when they have already applied their own bounded-retry policy - — e.g. the clean-exit protocol-violation streak in - ``detect_crashed_workers``, which resolves the per-task - ``max_retries`` override against the violation streak itself. The - failure is still counted into ``consecutive_failures``. - """ - if failure_limit is None: - failure_limit = DEFAULT_FAILURE_LIMIT - blocked = False - with write_txn(conn): - row = conn.execute( - "SELECT consecutive_failures, status, max_retries " - "FROM tasks WHERE id = ?", (task_id,), - ).fetchone() - if row is None: - return False - failures = int(row["consecutive_failures"]) + 1 - - # Per-task override wins over both caller-supplied and default - # thresholds. None (the common case) falls through. - task_override = ( - row["max_retries"] if "max_retries" in row.keys() else None - ) - if task_override is not None: - effective_limit = int(task_override) - limit_source = "task" - else: - effective_limit = int(failure_limit) - limit_source = "dispatcher" - - if force_trip or failures >= effective_limit: - # Trip the breaker. - if release_claim: - # Spawn path: still running, also clear claim state. - conn.execute( - "UPDATE tasks SET status = 'blocked', claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, " - "consecutive_failures = ?, last_failure_error = ? " - "WHERE id = ? AND status IN ('running', 'ready')", - (failures, error[:500], task_id), - ) - else: - # Timeout/crash path: task is already at ``ready`` - # with claim cleared; just flip to blocked + update - # counter fields. - conn.execute( - "UPDATE tasks SET status = 'blocked', " - "consecutive_failures = ?, last_failure_error = ? " - "WHERE id = ? AND status IN ('ready', 'running')", - (failures, error[:500], task_id), - ) - run_id = None - if end_run: - # Only the spawn path has an open run to close. - run_id = _end_run( - conn, task_id, - outcome="gave_up", status="gave_up", - error=error[:500], - metadata={ - "failures": failures, - "trigger_outcome": outcome, - "effective_limit": effective_limit, - "limit_source": limit_source, - }, - ) - payload = { - "failures": failures, - "effective_limit": effective_limit, - "limit_source": limit_source, - "error": error[:500], - "trigger_outcome": outcome, - } - if event_payload_extra: - payload.update(event_payload_extra) - _append_event( - conn, task_id, "gave_up", payload, run_id=run_id, - ) - blocked = True - else: - # Below threshold. - if release_claim: - # Spawn path: transition running → ready + clear claim. - conn.execute( - "UPDATE tasks SET status = 'ready', claim_lock = NULL, " - "claim_expires = NULL, worker_pid = NULL, " - "consecutive_failures = ?, last_failure_error = ? " - "WHERE id = ? AND status = 'running'", - (failures, error[:500], task_id), - ) - else: - # Timeout/crash path: task is already at ``ready`` via - # its own UPDATE. Just bookkeep the counter + last error. - conn.execute( - "UPDATE tasks SET consecutive_failures = ?, " - "last_failure_error = ? WHERE id = ?", - (failures, error[:500], task_id), - ) - if end_run: - # Spawn path: close the open run with outcome. - run_id = _end_run( - conn, task_id, - outcome=outcome, status=outcome, - error=error[:500], - metadata={"failures": failures}, - ) - _append_event( - conn, task_id, outcome, - {"error": error[:500], "failures": failures}, - run_id=run_id, - ) - # Timeout/crash path's caller already emitted its own event. - return blocked - - -# Backward-compat alias. Old name is referenced from tests and possibly -# third-party callers. New code should call ``_record_task_failure``. -def _record_spawn_failure( - conn: sqlite3.Connection, - task_id: str, - error: str, - *, - failure_limit: int = None, -) -> bool: - return _record_task_failure( - conn, task_id, error, - outcome="spawn_failed", - failure_limit=failure_limit, - release_claim=True, - end_run=True, - ) - - -def _set_worker_pid(conn: sqlite3.Connection, task_id: str, pid: int) -> None: - """Record the spawned child's pid + emit a ``spawned`` event. - - The event's payload carries the pid so a human reading ``hermes kanban - tail`` can correlate log lines with OS-level traces without opening - the drawer. - """ - with write_txn(conn): - conn.execute( - "UPDATE tasks SET worker_pid = ? WHERE id = ?", - (int(pid), task_id), - ) - run_id = _current_run_id(conn, task_id) - if run_id is not None: - conn.execute( - "UPDATE task_runs SET worker_pid = ? WHERE id = ?", - (int(pid), run_id), - ) - _append_event(conn, task_id, "spawned", {"pid": int(pid)}, run_id=run_id) - - -def _clear_failure_counter(conn: sqlite3.Connection, task_id: str) -> None: - """Reset the unified consecutive-failures counter. - - Called from ``complete_task`` on successful completion — a fresh - success means the task + profile combination is working and any - past failures are history. NOT called on spawn success anymore: - a successful spawn proves the worker could start but says nothing - about whether the run will succeed, so we need to let timeouts and - crashes accumulate across spawn boundaries. - """ - with write_txn(conn): - conn.execute( - "UPDATE tasks SET consecutive_failures = 0, " - "last_failure_error = NULL WHERE id = ?", - (task_id,), - ) - - -# Legacy alias for test-code and anything else that still imports it. -_clear_spawn_failures = _clear_failure_counter - - def check_respawn_guard(conn: sqlite3.Connection, task_id: str) -> Optional[str]: """Return a guard reason if ``task_id`` should NOT be re-spawned, else None. @@ -10273,3 +9485,45 @@ def latest_summaries( ids, ).fetchall() return {r["task_id"]: r["summary"] for r in rows} + + +# --------------------------------------------------------------------------- +# Re-export of extracted modules (wave 1, shard s4, implementer w1b). +# The moved functions live in their own modules; these bottom imports keep +# them importable from ``hermes_cli.kanban_db`` for back-compat (tests and +# the CLI reference them by name on this module). Deferred to the bottom so +# the new modules can import shared helpers from here without a cycle. +# --------------------------------------------------------------------------- +from hermes_cli.task_lifecycle import ( # noqa: E402,F401 + archive_task, + delete_archived_task, + delete_task, +) +from hermes_cli.workspace_resolution import ( # noqa: E402,F401 + _ensure_git_worktree, + _git_branch_exists, + _git_common_dir, + _git_current_branch, + _git_dir, + _git_toplevel, + _is_linked_worktree_checkout, + _nearest_existing_path, + _repo_root_for_worktree_target, + _resolve_worktree_workspace, + resolve_workspace, + set_branch_name, + set_workspace_path, +) +from hermes_cli.worker_process import ( # noqa: E402,F401 + _classify_worker_exit, + _pid_alive, + _record_worker_exit, + reap_worker_zombies, +) +from hermes_cli.failure_accounting import ( # noqa: E402,F401 + _clear_failure_counter, + _clear_spawn_failures, + _record_spawn_failure, + _record_task_failure, + _set_worker_pid, +) diff --git a/hermes_cli/task_lifecycle.py b/hermes_cli/task_lifecycle.py new file mode 100644 index 000000000000..10964cb980ae --- /dev/null +++ b/hermes_cli/task_lifecycle.py @@ -0,0 +1,91 @@ +"""Task lifecycle removal (archive / hard delete) for kanban boards. + +Extracted verbatim from ``hermes_cli.kanban_db`` (wave 1, shard s4, +cluster c1 / task_lifecycle). Shared txn/run/event helpers stay in +kanban_db.py and are imported at the bottom to avoid a circular import. +""" +from __future__ import annotations + +import sqlite3 + +def archive_task(conn: sqlite3.Connection, task_id: str) -> bool: + with write_txn(conn): + cur = conn.execute( + "UPDATE tasks SET status = 'archived', " + " claim_lock = NULL, claim_expires = NULL, worker_pid = NULL " + "WHERE id = ? AND status != 'archived'", + (task_id,), + ) + if cur.rowcount != 1: + return False + # If archive happened while a run was still in flight (e.g. user + # archived a running task from the dashboard), close that run with + # outcome='reclaimed' so attempt history isn't orphaned. + run_id = _end_run( + conn, task_id, + outcome="reclaimed", status="reclaimed", + summary="task archived with run still active", + ) + _append_event(conn, task_id, "archived", None, run_id=run_id) + # ``archived`` parents no longer block children, same as ``done``. + # Promote newly-unblocked dependents immediately instead of waiting + # for a later dispatcher tick. + recompute_ready(conn) + return True + + +def delete_archived_task(conn: sqlite3.Connection, task_id: str) -> bool: + """Permanently remove an already-archived task and its related rows. + + Safety guard: only archived tasks can be deleted. Active / blocked / done + tasks must be explicitly archived first so accidental data loss requires a + second deliberate action. + """ + with write_txn(conn): + row = conn.execute( + "SELECT status FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + if not row or row["status"] != "archived": + return False + conn.execute( + "DELETE FROM task_links WHERE parent_id = ? OR child_id = ?", + (task_id, task_id), + ) + conn.execute("DELETE FROM task_comments WHERE task_id = ?", (task_id,)) + conn.execute("DELETE FROM task_events WHERE task_id = ?", (task_id,)) + conn.execute("DELETE FROM task_runs WHERE task_id = ?", (task_id,)) + conn.execute("DELETE FROM kanban_notify_subs WHERE task_id = ?", (task_id,)) + cur = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) + return cur.rowcount == 1 + + +def delete_task(conn: sqlite3.Connection, task_id: str) -> bool: + """Hard-delete a task and cascade to all related rows. + + Because the schema does not use ``ON DELETE CASCADE`` foreign keys, + we explicitly delete from child tables first, then the task row. + This keeps the operation atomic (single ``write_txn``). + + Returns ``True`` if the task existed and was deleted, ``False`` + if the task was not found. + """ + with write_txn(conn): + cur = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) + if cur.rowcount != 1: + return False + conn.execute("DELETE FROM task_links WHERE parent_id = ? OR child_id = ?", (task_id, task_id)) + conn.execute("DELETE FROM task_comments WHERE task_id = ?", (task_id,)) + conn.execute("DELETE FROM task_events WHERE task_id = ?", (task_id,)) + conn.execute("DELETE FROM task_runs WHERE task_id = ?", (task_id,)) + conn.execute("DELETE FROM kanban_notify_subs WHERE task_id = ?", (task_id,)) + recompute_ready(conn) + return True + + +from hermes_cli.kanban_db import ( # noqa: E402 + _append_event, + _end_run, + recompute_ready, + write_txn, +) diff --git a/hermes_cli/worker_process.py b/hermes_cli/worker_process.py new file mode 100644 index 000000000000..6cdc88e1cb96 --- /dev/null +++ b/hermes_cli/worker_process.py @@ -0,0 +1,184 @@ +"""Worker process supervision for the kanban dispatcher. + +Extracted verbatim from ``hermes_cli.kanban_db`` (wave 1, shard s4, +cluster c5 / worker_process). The ``_recent_worker_exits`` registry and +its bounds are module state used only by this cluster and move with it. +""" +from __future__ import annotations + +import os +import subprocess +import sys +import time +from typing import Optional + +# KANBAN_RATE_LIMIT_EXIT_CODE stays in kanban_db.py (tests import it from +# there); imported at the bottom to avoid a circular import. + +# Bounded registry of recently-reaped worker child exits, populated by the +# reap loop at the top of ``dispatch_once`` and consulted by +# ``detect_crashed_workers`` to classify a dead-pid task. +# +# Entry: ``pid -> (raw_wait_status, reaped_at_epoch)``. We keep raw status +# so both ``os.WIFEXITED`` / ``os.WEXITSTATUS`` and ``os.WIFSIGNALED`` can +# be consulted. Entries are trimmed by age (and total size cap as a +# belt-and-braces against unbounded growth on exotic platforms). +_RECENT_WORKER_EXIT_TTL_SECONDS = 600 +_RECENT_WORKER_EXITS_MAX = 4096 +_recent_worker_exits: "dict[int, tuple[int, float]]" = {} + + +def _record_worker_exit(pid: int, raw_status: int) -> None: + """Record a reaped child's exit status for later classification. + + Called from the reap loop in ``dispatch_once``. Safe to call many + times; duplicate pids overwrite (pids can cycle, latest wins). + """ + if not pid or pid <= 0: + return + now = time.time() + _recent_worker_exits[int(pid)] = (int(raw_status), now) + # Age-based trim: drop entries older than the TTL. + if len(_recent_worker_exits) > _RECENT_WORKER_EXITS_MAX // 2: + cutoff = now - _RECENT_WORKER_EXIT_TTL_SECONDS + for _pid in [p for p, (_s, t) in _recent_worker_exits.items() if t < cutoff]: + _recent_worker_exits.pop(_pid, None) + # Size cap as a final guard. + if len(_recent_worker_exits) > _RECENT_WORKER_EXITS_MAX: + # Drop oldest half. + ordered = sorted(_recent_worker_exits.items(), key=lambda kv: kv[1][1]) + for _pid, _ in ordered[: len(ordered) // 2]: + _recent_worker_exits.pop(_pid, None) + + +def _classify_worker_exit(pid: int) -> "tuple[str, Optional[int]]": + """Classify a recently-reaped worker by pid. + + Returns ``(kind, code)`` where ``kind`` is one of: + + * ``"clean_exit"`` — ``WIFEXITED`` with ``WEXITSTATUS == 0``. When the + task is still ``running`` in the DB, this is a protocol violation + (worker exited without calling ``kanban_complete`` / ``kanban_block``) + and should be auto-blocked immediately — retrying will just loop. + * ``"rate_limited"`` — ``WIFEXITED`` with status + ``KANBAN_RATE_LIMIT_EXIT_CODE``. The worker bailed because the + provider rate-limited / exhausted quota, NOT because the task failed. + ``detect_crashed_workers`` releases the task back to ``ready`` without + counting a failure, so a long quota window can't trip the breaker. + * ``"nonzero_exit"`` — ``WIFEXITED`` with non-zero status. Real error. + * ``"signaled"`` — ``WIFSIGNALED`` (OOM killer, SIGKILL, etc). Real crash. + * ``"unknown"`` — pid was not in the reap registry (either reaped by + something else, or died between reap tick and liveness check). Fall + back to existing crashed-counter behavior. + + ``code`` is the exit status (for ``clean_exit`` / ``rate_limited`` / + ``nonzero_exit``) or the signal number (for ``signaled``), or ``None`` + for ``unknown``. + """ + entry = _recent_worker_exits.get(int(pid)) + if entry is None: + return ("unknown", None) + raw, _ = entry + try: + if os.WIFEXITED(raw): + code = os.WEXITSTATUS(raw) + if code == 0: + return ("clean_exit", 0) + if code == KANBAN_RATE_LIMIT_EXIT_CODE: + return ("rate_limited", code) + return ("nonzero_exit", code) + if os.WIFSIGNALED(raw): + return ("signaled", os.WTERMSIG(raw)) + except Exception: + pass + return ("unknown", None) + + +def reap_worker_zombies() -> "list[int]": + """Reap all zombie children of this process without blocking. + + Returns the list of reaped PIDs. Safe to call when there are no + children (returns []). No-op on Windows. + """ + reaped: "list[int]" = [] + if os.name != "nt": + try: + while True: + try: + pid, status = os.waitpid(-1, os.WNOHANG) + except ChildProcessError: + break + if pid == 0: + break + _record_worker_exit(pid, status) + reaped.append(pid) + except Exception: + pass + return reaped + + +def _pid_alive(pid: Optional[int]) -> bool: + """Return True if ``pid`` is still running on this host. + + Cross-platform: uses ``OpenProcess`` + ``WaitForSingleObject`` on + Windows (via ``gateway.status._pid_exists``) and ``os.kill(pid, 0)`` + on POSIX. Returns False for falsy PIDs or on any OS error. + + **DO NOT** use ``os.kill(pid, 0)`` directly on Windows — Python's + Windows ``os.kill`` treats ``sig=0`` as ``CTRL_C_EVENT`` (bpo-14484) + and will broadcast it to the target's console group, potentially + killing unrelated processes. + + **Zombie handling:** the existence check succeeds against zombie + processes (post-exit, pre-reap) because the process table entry + still exists. A worker that exits without being reaped by its + parent would stay "alive" to the dispatcher forever. Dispatcher + workers are started via ``start_new_session=True`` + intentional + Popen handle abandonment, so init reaps them quickly — but during + the window between exit and reap, we'd otherwise see stale "alive" + signals. On Linux we peek at ``/proc//status`` and treat + ``State: Z`` as dead. On macOS we ask ``ps`` for the BSD ``stat`` + field and treat values containing ``Z`` as dead. + """ + if not pid or pid <= 0: + return False + from gateway.status import _pid_exists + if not _pid_exists(int(pid)): + return False + # Still here → process exists. Check for zombie on platforms + # where we have a cheap, deterministic process-state probe. + if sys.platform == "linux": + try: + with open(f"/proc/{int(pid)}/status", "r", encoding="utf-8") as f: + for line in f: + if line.startswith("State:"): + # "State:\tZ (zombie)" → dead + if "Z" in line.split(":", 1)[1]: + return False + break + except (FileNotFoundError, PermissionError, OSError): + # proc entry gone → already reaped; treat as dead. + # PermissionError shouldn't happen for our own children but + # be defensive. + pass + elif sys.platform == "darwin": + try: + proc = subprocess.run( + ["ps", "-o", "stat=", "-p", str(int(pid))], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, encoding='utf-8', errors='replace', + timeout=1, + check=False, + ) + if proc.returncode != 0: + return False + if "Z" in (proc.stdout or "").strip(): + return False + except (OSError, subprocess.SubprocessError, TimeoutError): + # If the secondary probe fails, keep the kill(0) answer. + pass + return True + + +from hermes_cli.kanban_db import KANBAN_RATE_LIMIT_EXIT_CODE # noqa: E402 diff --git a/hermes_cli/workspace_resolution.py b/hermes_cli/workspace_resolution.py new file mode 100644 index 000000000000..43af3bedda40 --- /dev/null +++ b/hermes_cli/workspace_resolution.py @@ -0,0 +1,348 @@ +"""Workspace resolution for kanban tasks. + +Extracted verbatim from ``hermes_cli.kanban_db`` (wave 1, shard s4, +cluster c2 / workspace_resolution) so the git worktree machinery is +reviewable in isolation. The public entry points (``resolve_workspace``, +``set_workspace_path``, ``set_branch_name``) stay importable from +``hermes_cli.kanban_db`` via the bottom re-export in that module. +""" +from __future__ import annotations + +import sqlite3 +import subprocess +from pathlib import Path +from typing import Optional + +# Shared helpers stay in kanban_db.py; imported at the bottom to avoid a +# circular import (kanban_db re-imports the moved names from here). + +# --------------------------------------------------------------------------- +# Workspace resolution +# --------------------------------------------------------------------------- + +def _git_toplevel(path: Path) -> Optional[Path]: + """Return the git toplevel containing ``path``, or ``None`` if not in a repo.""" + try: + result = subprocess.run( + ["git", "-C", str(path), "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, encoding='utf-8', errors='replace', + timeout=30, + check=False, + ) + except Exception: + return None + if result.returncode != 0: + return None + out = (result.stdout or "").strip() + if not out: + return None + try: + return Path(out).expanduser().resolve() + except Exception: + return Path(out).expanduser() + + +def _git_branch_exists(repo_root: Path, branch_name: str) -> bool: + try: + result = subprocess.run( + ["git", "-C", str(repo_root), "show-ref", "--verify", f"refs/heads/{branch_name}"], + capture_output=True, + text=True, encoding='utf-8', errors='replace', + timeout=30, + check=False, + ) + except Exception: + return False + return result.returncode == 0 + + +def _git_common_dir(path: Path) -> Optional[Path]: + try: + result = subprocess.run( + ["git", "-C", str(path), "rev-parse", "--path-format=absolute", "--git-common-dir"], + capture_output=True, + text=True, encoding='utf-8', errors='replace', + timeout=30, + check=False, + ) + except Exception: + return None + if result.returncode != 0: + return None + out = (result.stdout or "").strip() + if not out: + return None + return Path(out).expanduser().resolve(strict=False) + + +def _git_dir(path: Path) -> Optional[Path]: + try: + result = subprocess.run( + ["git", "-C", str(path), "rev-parse", "--path-format=absolute", "--git-dir"], + capture_output=True, + text=True, encoding='utf-8', errors='replace', + timeout=30, + check=False, + ) + except Exception: + return None + if result.returncode != 0: + return None + out = (result.stdout or "").strip() + if not out: + return None + return Path(out).expanduser().resolve(strict=False) + + +def _git_current_branch(path: Path) -> Optional[str]: + try: + result = subprocess.run( + ["git", "-C", str(path), "branch", "--show-current"], + capture_output=True, + text=True, encoding='utf-8', errors='replace', + timeout=30, + check=False, + ) + except Exception: + return None + if result.returncode != 0: + return None + branch = (result.stdout or "").strip() + return branch or None + + +def _is_linked_worktree_checkout(path: Path) -> bool: + git_dir = _git_dir(path) + common_dir = _git_common_dir(path) + if git_dir is None or common_dir is None: + return False + return git_dir != common_dir + + +def _nearest_existing_path(path: Path) -> Path: + current = path + while not current.exists() and current != current.parent: + current = current.parent + return current + + +def _repo_root_for_worktree_target(path: Path) -> Optional[Path]: + current = _nearest_existing_path(path).resolve(strict=False) + while True: + repo_root = _git_toplevel(current) + if repo_root is not None: + return repo_root + if current == current.parent: + return None + current = current.parent + + +def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> None: + """Materialize ``target`` as a linked git worktree under ``repo_root``.""" + target = target.expanduser() + repo_common = _git_common_dir(repo_root) + if target.exists() and repo_common is not None: + target_common = _git_common_dir(target) + if target_common == repo_common: + return + target.parent.mkdir(parents=True, exist_ok=True) + if _git_branch_exists(repo_root, branch_name): + cmd = ["git", "-C", str(repo_root), "worktree", "add", str(target), branch_name] + else: + cmd = [ + "git", "-C", str(repo_root), "worktree", "add", "-b", branch_name, + str(target), "HEAD", + ] + result = subprocess.run( + cmd, + capture_output=True, + text=True, encoding='utf-8', errors='replace', + timeout=60, + check=False, + ) + if result.returncode != 0: + stderr = (result.stderr or result.stdout or "").strip() + raise RuntimeError( + f"git worktree add failed for {target} on branch {branch_name}: {stderr}" + ) + + +def _resolve_worktree_workspace( + task: Task, *, board: Optional[str] = None +) -> tuple[Path, str]: + """Resolve + materialize a linked git worktree for ``task``. + + When ``task.workspace_path`` is unset, the anchor is the board's + ``default_workdir`` (a persistent project checkout). This keeps every + worktree task under a meaningful, board-owned repo — ``/.worktrees/ + `` — instead of silently landing under the dispatcher's current + working directory (which is whatever directory the gateway happened to be + launched from, e.g. the Hermes checkout). If no anchor is configured + anywhere, we fail loudly rather than guess. + """ + branch_name = (task.branch_name or "").strip() or f"wt/{task.id}" + if not task.workspace_path: + # Anchor on the board's configured default_workdir, not Path.cwd(). + # The dispatcher's CWD is incidental (gateway launch dir) and using it + # scatters worktrees under whatever repo the gateway started in. + board_slug = board if board else get_current_board() + board_default = (read_board_metadata(board_slug).get("default_workdir") or "").strip() + if not board_default: + raise ValueError( + f"task {task.id} has workspace_kind=worktree but no workspace_path, " + f"and board {board_slug!r} has no default_workdir set. Set a board " + "default workdir (a git repo) or create the task with " + "--workspace worktree:." + ) + anchor = Path(board_default).expanduser() + if not anchor.is_absolute(): + raise ValueError( + f"board {board_slug!r} default_workdir {board_default!r} is not " + "absolute; use an absolute path to a git repo" + ) + repo_root = _git_toplevel(anchor) + if repo_root is None: + raise ValueError( + f"task {task.id} has workspace_kind=worktree but board " + f"{board_slug!r} default_workdir {board_default!r} is not inside a git repo" + ) + target = repo_root / ".worktrees" / task.id + _ensure_git_worktree(repo_root, target, branch_name) + return target, branch_name + + requested = Path(task.workspace_path).expanduser() + if not requested.is_absolute(): + raise ValueError( + f"task {task.id} has non-absolute worktree path " + f"{task.workspace_path!r}; use an absolute path" + ) + requested_resolved = requested.resolve(strict=False) + + if requested.exists() and _is_linked_worktree_checkout(requested): + actual_branch = _git_current_branch(requested) + if actual_branch == branch_name: + return requested_resolved, actual_branch + # The requested path is an existing checkout of a DIFFERENT + # task's branch. Decompose children inherit the root's + # workspace_path verbatim, so siblings all point here; reusing + # the checkout as-is would run this task on the other task's + # branch — silent cross-task provenance corruption, and unsafe + # when siblings run concurrently. Fall back to a fresh worktree + # of our own under the same repo. + fallback_root = _repo_root_for_worktree_target(requested.parent) + if fallback_root is not None: + fallback = fallback_root / ".worktrees" / task.id + if fallback.resolve(strict=False) != requested_resolved: + _ensure_git_worktree(fallback_root, fallback, branch_name) + return fallback.resolve(strict=False), branch_name + # No repo to anchor a fallback on (or the occupied path IS this + # task's own canonical worktree): keep the legacy reuse rather + # than failing dispatch. + return requested_resolved, actual_branch or branch_name + + repo_root = _git_toplevel(requested) + if repo_root is not None and requested_resolved == repo_root: + target = repo_root / ".worktrees" / task.id + _ensure_git_worktree(repo_root, target, branch_name) + return target, branch_name + + repo_root = _repo_root_for_worktree_target(requested.parent) + if repo_root is None: + raise ValueError( + f"task {task.id} worktree path {task.workspace_path!r} is not inside a git repo " + "and does not point at a git repo root" + ) + _ensure_git_worktree(repo_root, requested, branch_name) + return requested, branch_name + + +def resolve_workspace(task: Task, *, board: Optional[str] = None) -> Path: + """Resolve (and create if needed) the workspace for a task. + + - ``scratch``: a fresh dir under ``/workspaces//``, + where ```` is the active board's root. The path is the + same for the dispatcher and every profile worker, so handoff is + path-stable. + - ``dir:``: the path stored in ``workspace_path``. Created + if missing. MUST be absolute — relative paths are rejected to + prevent confused-deputy traversal where ``../../../tmp/attacker`` + resolves against the dispatcher's CWD instead of a meaningful + root. Users who want a kanban-root-relative workspace should + compute the absolute path themselves. + - ``worktree``: a real linked git worktree. If ``workspace_path`` names + a repo root, Hermes treats it as an anchor and materializes a linked + worktree at ``/.worktrees/``. If ``workspace_path`` names + a concrete target path, Hermes creates/reuses that linked worktree. With + no ``workspace_path``, Hermes anchors on the board's ``default_workdir`` + and materializes ``/.worktrees/`` per task; if no + ``default_workdir`` is configured it raises rather than guessing from the + dispatcher's CWD. When ``branch_name`` is empty, Hermes uses + ``wt/``. + + Persist the resolved path back to the task row via ``set_workspace_path`` + so subsequent runs reuse the same directory. + """ + kind = task.workspace_kind or "scratch" + if kind == "scratch": + if task.workspace_path: + # Legacy scratch tasks that were set to an explicit path get the + # same absolute-path guard as dir: — consistent with the + # threat model. + p = Path(task.workspace_path).expanduser() + if not p.is_absolute(): + raise ValueError( + f"task {task.id} has non-absolute workspace_path " + f"{task.workspace_path!r}; workspace paths must be absolute" + ) + else: + p = workspaces_root(board=board) / task.id + p.mkdir(parents=True, exist_ok=True) + return p + if kind == "dir": + if not task.workspace_path: + raise ValueError( + f"task {task.id} has workspace_kind=dir but no workspace_path" + ) + p = Path(task.workspace_path).expanduser() + if not p.is_absolute(): + raise ValueError( + f"task {task.id} has non-absolute workspace_path " + f"{task.workspace_path!r}; use an absolute path " + f"(relative paths are ambiguous against the dispatcher's CWD)" + ) + p.mkdir(parents=True, exist_ok=True) + return p + if kind == "worktree": + p, _branch_name = _resolve_worktree_workspace(task, board=board) + return p + raise ValueError(f"unknown workspace_kind: {kind}") + + +def set_workspace_path( + conn: sqlite3.Connection, task_id: str, path: Path | str +) -> None: + with write_txn(conn): + conn.execute( + "UPDATE tasks SET workspace_path = ? WHERE id = ?", + (str(path), task_id), + ) + + +def set_branch_name( + conn: sqlite3.Connection, task_id: str, branch_name: str +) -> None: + with write_txn(conn): + conn.execute( + "UPDATE tasks SET branch_name = ? WHERE id = ?", + (str(branch_name), task_id), + ) + + +from hermes_cli.kanban_db import ( # noqa: E402 + Task, + get_current_board, + read_board_metadata, + workspaces_root, + write_txn, +) diff --git a/tests/hermes_cli/test_kanban_failure_accounting.py b/tests/hermes_cli/test_kanban_failure_accounting.py new file mode 100644 index 000000000000..cc47eabf9a89 --- /dev/null +++ b/tests/hermes_cli/test_kanban_failure_accounting.py @@ -0,0 +1,190 @@ +"""Regression tests for the failure_accounting cluster (s4-w1b extraction). + +Covers the circuit-breaker bookkeeping moved verbatim from +``hermes_cli.kanban_db`` (cluster c10 / failure_accounting) into +``hermes_cli.failure_accounting``: ``_record_task_failure``, +``_record_spawn_failure`` (back-compat alias), ``_set_worker_pid``, +``_clear_failure_counter``, and the ``_clear_spawn_failures`` alias. +""" + +from __future__ import annotations + +import pytest + +import hermes_cli.kanban_db as kb +from hermes_cli.failure_accounting import ( + _clear_failure_counter, + _clear_spawn_failures, + _record_spawn_failure, + _record_task_failure, + _set_worker_pid, +) + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(__import__("pathlib").Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +# --------------------------------------------------------------------------- +# Re-export parity +# --------------------------------------------------------------------------- + + +def test_moved_names_reexported_on_kanban_db_module(): + for name in ("_record_task_failure", "_record_spawn_failure", + "_set_worker_pid", "_clear_failure_counter", + "_clear_spawn_failures"): + assert getattr(kb, name) is globals()[name], name + + +def test_legacy_alias_preserved(): + assert _clear_spawn_failures is _clear_failure_counter + assert kb._clear_spawn_failures is kb._clear_failure_counter + + +def test_direct_module_import_works(): + import hermes_cli.failure_accounting as fa + assert fa._record_task_failure is _record_task_failure + + +# --------------------------------------------------------------------------- +# _record_task_failure +# --------------------------------------------------------------------------- + + +def test_record_task_failure_below_limit_does_not_block(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="below") + conn.execute("UPDATE tasks SET status = 'ready' WHERE id = ?", (t,)) + conn.commit() + blocked = _record_task_failure( + conn, t, error="boom", outcome="crashed", + ) + assert blocked is False + row = conn.execute( + "SELECT consecutive_failures, status FROM tasks WHERE id = ?", (t,), + ).fetchone() + assert row["consecutive_failures"] == 1 + assert row["status"] == "ready" + + +def test_record_task_failure_trips_breaker_at_limit(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="trip") + conn.execute("UPDATE tasks SET status = 'ready' WHERE id = ?", (t,)) + conn.commit() + # First hit: 1 < DEFAULT_FAILURE_LIMIT(2) -> not blocked. + assert _record_task_failure(conn, t, error="e1", outcome="crashed") is False + # Second hit: 2 >= 2 -> blocked. + assert _record_task_failure(conn, t, error="e2", outcome="crashed") is True + row = conn.execute( + "SELECT consecutive_failures, status FROM tasks WHERE id = ?", (t,), + ).fetchone() + assert row["consecutive_failures"] == 2 + assert row["status"] == "blocked" + ev = conn.execute( + "SELECT 1 FROM task_events " + "WHERE task_id = ? AND kind = 'gave_up' ORDER BY id DESC LIMIT 1", + (t,), + ).fetchone() + assert ev is not None + + +def test_record_task_failure_respects_caller_failure_limit(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="limit1") + conn.execute("UPDATE tasks SET status = 'ready' WHERE id = ?", (t,)) + conn.commit() + blocked = _record_task_failure( + conn, t, error="x", outcome="spawn_failed", failure_limit=1, + release_claim=False, end_run=False, + ) + assert blocked is True + row = conn.execute( + "SELECT status FROM tasks WHERE id = ?", (t,), + ).fetchone() + assert row["status"] == "blocked" + + +def test_record_task_failure_missing_task_returns_false(kanban_home): + with kb.connect() as conn: + assert _record_task_failure(conn, "nope", error="x", outcome="crashed") is False + + +def test_record_task_failure_force_trip(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="force") + blocked = _record_task_failure( + conn, t, error="x", outcome="crashed", force_trip=True, + ) + assert blocked is True + + +# --------------------------------------------------------------------------- +# _record_spawn_failure (back-compat wrapper) +# --------------------------------------------------------------------------- + + +def test_record_spawn_failure_wraps_record_task_failure(kanban_home, monkeypatch): + import hermes_cli.failure_accounting as fa + calls = [] + + def _spy(conn, task_id, error, *, outcome, failure_limit=None, + force_trip=False, release_claim=False, end_run=False, + event_payload_extra=None): + calls.append((task_id, error, outcome, release_claim, end_run)) + return False + + monkeypatch.setattr(fa, "_record_task_failure", _spy) + with kb.connect() as conn: + t = kb.create_task(conn, title="spawnfail") + _record_spawn_failure(conn, t, "spawn blew up") + assert calls == [(t, "spawn blew up", "spawn_failed", True, True)] + + +# --------------------------------------------------------------------------- +# _set_worker_pid +# --------------------------------------------------------------------------- + + +def test_set_worker_pid_persists_and_emits_event(kanban_home): + import json + with kb.connect() as conn: + t = kb.create_task(conn, title="pid") + kb.claim_task(conn, t) + _set_worker_pid(conn, t, 12345) + row = conn.execute( + "SELECT worker_pid FROM tasks WHERE id = ?", (t,), + ).fetchone() + assert row["worker_pid"] == 12345 + ev = conn.execute( + "SELECT payload FROM task_events " + "WHERE task_id = ? AND kind = 'spawned' ORDER BY id DESC LIMIT 1", + (t,), + ).fetchone() + assert ev is not None + assert json.loads(ev["payload"])["pid"] == 12345 + + +# --------------------------------------------------------------------------- +# _clear_failure_counter +# --------------------------------------------------------------------------- + + +def test_clear_failure_counter_resets(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="clear") + _record_task_failure(conn, t, error="e1", outcome="crashed") + _clear_failure_counter(conn, t) + row = conn.execute( + "SELECT consecutive_failures, last_failure_error " + "FROM tasks WHERE id = ?", (t,), + ).fetchone() + assert row["consecutive_failures"] == 0 + assert row["last_failure_error"] is None diff --git a/tests/hermes_cli/test_kanban_task_lifecycle.py b/tests/hermes_cli/test_kanban_task_lifecycle.py new file mode 100644 index 000000000000..fa3836f54798 --- /dev/null +++ b/tests/hermes_cli/test_kanban_task_lifecycle.py @@ -0,0 +1,170 @@ +"""Regression tests for the task_lifecycle cluster (s4-w1b extraction). + +Covers archive / hard-delete moved verbatim from ``hermes_cli.kanban_db`` +(cluster c1 / task_lifecycle) into ``hermes_cli.task_lifecycle``. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +import hermes_cli.kanban_db as kb +from hermes_cli.task_lifecycle import ( + archive_task, + delete_archived_task, + delete_task, +) + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +# --------------------------------------------------------------------------- +# Re-export parity +# --------------------------------------------------------------------------- + + +def test_moved_names_reexported_on_kanban_db_module(): + for name in ("archive_task", "delete_archived_task", "delete_task"): + assert getattr(kb, name) is globals()[name], name + + +def test_direct_module_import_works(): + import hermes_cli.task_lifecycle as tl + assert tl.archive_task is archive_task + + +# --------------------------------------------------------------------------- +# archive_task +# --------------------------------------------------------------------------- + + +def test_archive_task_archives_and_emits_event(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="to archive") + assert archive_task(conn, t) is True + row = conn.execute( + "SELECT status FROM tasks WHERE id = ?", (t,), + ).fetchone() + assert row["status"] == "archived" + ev = conn.execute( + "SELECT 1 FROM task_events WHERE task_id = ? AND kind = 'archived'", + (t,), + ).fetchone() + assert ev is not None + + +def test_archive_task_missing_returns_false(kanban_home): + with kb.connect() as conn: + assert archive_task(conn, "missing-task") is False + + +def test_archive_task_already_archived_returns_false(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="twice") + assert archive_task(conn, t) is True + assert archive_task(conn, t) is False + + +def test_archive_task_promotes_dependents(kanban_home): + """Archiving a parent must unblock its children (recompute_ready).""" + with kb.connect() as conn: + parent = kb.create_task(conn, title="parent") + child = kb.create_task(conn, title="child") + conn.execute( + "INSERT INTO task_links (parent_id, child_id) VALUES (?, ?)", + (parent, child), + ) + conn.execute( + "UPDATE tasks SET status = 'blocked' WHERE id = ?", (child,), + ) + conn.commit() + assert archive_task(conn, parent) is True + row = conn.execute( + "SELECT status FROM tasks WHERE id = ?", (child,), + ).fetchone() + assert row["status"] == "ready" + + +# --------------------------------------------------------------------------- +# delete_archived_task +# --------------------------------------------------------------------------- + + +def test_delete_archived_task_removes_row(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="purge me") + archive_task(conn, t) + assert delete_archived_task(conn, t) is True + assert kb.get_task(conn, t) is None + + +def test_delete_archived_task_refuses_non_archived(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="active") + assert delete_archived_task(conn, t) is False + assert kb.get_task(conn, t) is not None + + +def test_delete_archived_task_cascades_related_rows(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="cascade") + conn.execute("INSERT INTO task_comments (task_id, author, body, created_at) " + "VALUES (?, 'a', 'b', 1)", (t,)) + conn.execute("INSERT INTO task_events (task_id, kind, created_at) " + "VALUES (?, 'created', 1)", (t,)) + conn.commit() + archive_task(conn, t) + assert delete_archived_task(conn, t) is True + assert conn.execute( + "SELECT 1 FROM task_comments WHERE task_id = ?", (t,), + ).fetchone() is None + assert conn.execute( + "SELECT 1 FROM task_events WHERE task_id = ?", (t,), + ).fetchone() is None + + +# --------------------------------------------------------------------------- +# delete_task +# --------------------------------------------------------------------------- + + +def test_delete_task_hard_deletes(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="hard delete") + assert delete_task(conn, t) is True + assert kb.get_task(conn, t) is None + + +def test_delete_task_missing_returns_false(kanban_home): + with kb.connect() as conn: + assert delete_task(conn, "missing-task") is False + + +def test_delete_task_cascades_links_and_comments(kanban_home): + with kb.connect() as conn: + a = kb.create_task(conn, title="a") + b = kb.create_task(conn, title="b") + conn.execute( + "INSERT INTO task_links (parent_id, child_id) VALUES (?, ?)", (a, b), + ) + conn.execute("INSERT INTO task_comments (task_id, author, body, created_at) " + "VALUES (?, 'x', 'y', 1)", (a,)) + conn.commit() + assert delete_task(conn, a) is True + assert conn.execute( + "SELECT 1 FROM task_links WHERE parent_id = ? OR child_id = ?", + (a, a), + ).fetchone() is None + assert conn.execute( + "SELECT 1 FROM task_comments WHERE task_id = ?", (a,), + ).fetchone() is None diff --git a/tests/hermes_cli/test_kanban_worker_process.py b/tests/hermes_cli/test_kanban_worker_process.py new file mode 100644 index 000000000000..7e39a9281b0e --- /dev/null +++ b/tests/hermes_cli/test_kanban_worker_process.py @@ -0,0 +1,208 @@ +"""Regression tests for the worker_process cluster (s4-w1b extraction). + +Covers the reap/classify/liveness helpers moved verbatim from +``hermes_cli.kanban_db`` (cluster c5 / worker_process) into +``hermes_cli.worker_process``, including the module-level +``_recent_worker_exits`` registry that moved with the cluster. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +import time + +import pytest + +import hermes_cli.kanban_db as kb +from hermes_cli.worker_process import ( + _classify_worker_exit, + _pid_alive, + _record_worker_exit, + reap_worker_zombies, +) + + +def _exited_status(code: int) -> int: + """Raw wait-status for a WIFEXITED child with the given exit code.""" + return code << 8 + + +# --------------------------------------------------------------------------- +# Re-export parity +# --------------------------------------------------------------------------- + + +def test_moved_names_reexported_on_kanban_db_module(): + for name in ("_record_worker_exit", "_classify_worker_exit", + "reap_worker_zombies", "_pid_alive"): + assert getattr(kb, name) is globals()[name], name + + +def test_direct_module_import_works(): + import hermes_cli.worker_process as wp + assert wp._pid_alive is _pid_alive + + +# --------------------------------------------------------------------------- +# Exit registry: record -> classify roundtrip +# +# NOTE: ``os.WIFEXITED``/``os.WIFSIGNALED`` do not exist on Windows, so the +# classification branch degrades to ``("unknown", None)`` there — identical +# to the behavior of the same code before extraction (verified against the +# live repo). The classification-semantics tests are POSIX-only; the registry +# sharing tests run everywhere. +# --------------------------------------------------------------------------- + +_is_posix = os.name != "nt" + + +def test_record_then_classify_clean_exit(): + if not _is_posix: + pytest.skip("os.WIFEXITED unavailable on Windows") + pid = 424242 + _record_worker_exit(pid, _exited_status(0)) + assert _classify_worker_exit(pid) == ("clean_exit", 0) + + +def test_record_then_classify_rate_limited(): + if not _is_posix: + pytest.skip("os.WIFEXITED unavailable on Windows") + pid = 424243 + _record_worker_exit(pid, _exited_status(kb.KANBAN_RATE_LIMIT_EXIT_CODE)) + assert _classify_worker_exit(pid) == ("rate_limited", kb.KANBAN_RATE_LIMIT_EXIT_CODE) + + +def test_record_then_classify_nonzero_exit(): + if not _is_posix: + pytest.skip("os.WIFEXITED unavailable on Windows") + pid = 424244 + _record_worker_exit(pid, _exited_status(7)) + assert _classify_worker_exit(pid) == ("nonzero_exit", 7) + + +def test_classify_unknown_pid(): + assert _classify_worker_exit(99999999) == ("unknown", None) + + +def test_record_worker_exit_ignores_falsy_pid(): + import hermes_cli.worker_process as wp + _record_worker_exit(0, _exited_status(1)) + _record_worker_exit(-5, _exited_status(1)) + # No crash, and nothing recorded for pid 0. + assert 0 not in wp._recent_worker_exits + assert -5 not in wp._recent_worker_exits + + +def test_duplicate_pid_overwrites_latest_wins(): + import hermes_cli.worker_process as wp + pid = 424245 + _record_worker_exit(pid, _exited_status(1)) + _record_worker_exit(pid, _exited_status(0)) + # The registry holds the raw status; the latest write wins. + assert wp._recent_worker_exits[pid][0] == _exited_status(0) + if _is_posix: + assert _classify_worker_exit(pid) == ("clean_exit", 0) + + +def test_registry_shared_with_kanban_db_namespace(): + """The registry must be one object shared by both import surfaces.""" + import hermes_cli.worker_process as wp + pid = 424246 + _record_worker_exit(pid, _exited_status(3)) + assert pid in wp._recent_worker_exits + # Both surfaces observe the same registry (same function objects). + assert kb._record_worker_exit is _record_worker_exit + assert kb._classify_worker_exit is _classify_worker_exit + assert _classify_worker_exit(pid) == kb._classify_worker_exit(pid) + + +# --------------------------------------------------------------------------- +# reap_worker_zombies +# --------------------------------------------------------------------------- + + +def test_reap_worker_zombies_noop_when_no_children(): + reaped = reap_worker_zombies() + assert isinstance(reaped, list) + + +@pytest.mark.skipif(os.name == "nt", reason="fork-based reap test is POSIX-only") +def test_reap_worker_zombies_reaps_and_records(): + """POSIX: spawn a child that exits; reap it and see it in the registry.""" + # fork() so the child becomes a real zombie: subprocess.wait() would reap + # the child itself (waitpid consumes it), leaving nothing for + # reap_worker_zombies() to find. The child may not have reached os._exit + # by the time the parent runs waitpid(WNOHANG), so poll with a bounded + # deadline instead of a single reap call. + pid = os.fork() + if pid == 0: + os._exit(0) # child exits immediately; parent does not wait -> zombie + reaped = [] + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + reaped = reap_worker_zombies() + if pid in reaped: + break + time.sleep(0.01) + assert pid in reaped + kind, code = _classify_worker_exit(pid) + assert kind == "clean_exit" + assert code == 0 + + +# --------------------------------------------------------------------------- +# _pid_alive +# --------------------------------------------------------------------------- + + +def test_pid_alive_falsy_and_negative(): + assert _pid_alive(None) is False + assert _pid_alive(0) is False + assert _pid_alive(-1) is False + + +def test_pid_alive_own_process(): + assert _pid_alive(os.getpid()) is True + + +def test_pid_alive_dead_pid_false(): + # A pid that can never be alive (Windows PIDs stay < 2^32; this is far + # above the range and no process table entry can exist for it). + assert _pid_alive(2**31 + 99999) is False + + +# --------------------------------------------------------------------------- +# TTL / size-cap trimming (module state behavior) +# --------------------------------------------------------------------------- + + +def test_record_worker_exit_trims_by_age(monkeypatch): + import hermes_cli.worker_process as wp + pid = 424247 + fake_now = [1_000_000.0] + + monkeypatch.setattr(wp.time, "time", lambda: fake_now[0]) + _record_worker_exit(pid, _exited_status(0)) + fake_now[0] += wp._RECENT_WORKER_EXIT_TTL_SECONDS + 1 + # Trigger a trim by pushing the registry past half its cap: since the + # entry is older than the TTL, it must be dropped. + for i in range(wp._RECENT_WORKER_EXITS_MAX // 2 + 1): + _record_worker_exit(90000 + i, _exited_status(0)) + assert _classify_worker_exit(pid) == ("unknown", None) + + +def test_record_worker_exit_size_cap_evicts_oldest(monkeypatch): + import hermes_cli.worker_process as wp + fake_now = [1_000_000.0] + monkeypatch.setattr(wp.time, "time", lambda: fake_now[0]) + + # Fill the registry past the hard cap; the oldest half must be dropped. + for i in range(wp._RECENT_WORKER_EXITS_MAX + 20): + _record_worker_exit(100000 + i, _exited_status(0)) + fake_now[0] += 0.001 # each entry strictly newer than the last + + assert len(wp._recent_worker_exits) <= wp._RECENT_WORKER_EXITS_MAX + # The very first entry (oldest) must be gone. + assert _classify_worker_exit(100000) == ("unknown", None) diff --git a/tests/hermes_cli/test_kanban_workspace_resolution.py b/tests/hermes_cli/test_kanban_workspace_resolution.py new file mode 100644 index 000000000000..5e027663ee43 --- /dev/null +++ b/tests/hermes_cli/test_kanban_workspace_resolution.py @@ -0,0 +1,243 @@ +"""Regression tests for the workspace_resolution cluster (s4-w1b extraction). + +Covers the pure git helpers moved verbatim from ``hermes_cli.kanban_db`` +(cluster c2 / workspace_resolution) into ``hermes_cli.workspace_resolution``. +Both import surfaces are exercised: the new module directly, and the +re-exported names on ``hermes_cli.kanban_db`` (the public API the CLI and +existing tests use). +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +import hermes_cli.kanban_db as kb +from hermes_cli.workspace_resolution import ( + _ensure_git_worktree, + _git_branch_exists, + _git_common_dir, + _git_current_branch, + _git_dir, + _git_toplevel, + _is_linked_worktree_checkout, + _nearest_existing_path, + _repo_root_for_worktree_target, + _resolve_worktree_workspace, + resolve_workspace, + set_branch_name, + set_workspace_path, +) + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + """Isolated HERMES_HOME with an empty kanban DB (mirrors test_kanban_db.py).""" + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +def _init_git_repo(repo: Path) -> None: + repo.mkdir(parents=True, exist_ok=True) + subprocess.run(["git", "init", "-b", "main", str(repo)], check=True, capture_output=True, text=True) + subprocess.run(["git", "-C", str(repo), "config", "user.email", "kanban@example.com"], check=True, capture_output=True, text=True) + subprocess.run(["git", "-C", str(repo), "config", "user.name", "Kanban Test"], check=True, capture_output=True, text=True) + (repo / "README.md").write_text("hello\n", encoding="utf-8") + subprocess.run(["git", "-C", str(repo), "add", "README.md"], check=True, capture_output=True, text=True) + subprocess.run(["git", "-C", str(repo), "commit", "-m", "init"], check=True, capture_output=True, text=True) + + +# --------------------------------------------------------------------------- +# Re-export parity: the moved functions must be THE SAME objects the public +# kanban_db module exposes (tests and CLI reference them by name there). +# --------------------------------------------------------------------------- + + +def test_moved_names_reexported_on_kanban_db_module(): + for name in ( + "_git_toplevel", "_git_branch_exists", "_git_common_dir", "_git_dir", + "_git_current_branch", "_is_linked_worktree_checkout", + "_nearest_existing_path", "_repo_root_for_worktree_target", + "_ensure_git_worktree", "_resolve_worktree_workspace", + "resolve_workspace", "set_workspace_path", "set_branch_name", + ): + assert getattr(kb, name) is globals()[name], name + + +def test_direct_module_import_works(): + """The new module must be importable on its own (no import cycle).""" + import hermes_cli.workspace_resolution as ws + assert ws.resolve_workspace is resolve_workspace + + +# --------------------------------------------------------------------------- +# Git helpers +# --------------------------------------------------------------------------- + + +def test_git_toplevel_finds_repo_root(tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + assert _git_toplevel(repo) == repo.resolve() + nested = repo / "sub" / "dir" + nested.mkdir(parents=True) + assert _git_toplevel(nested) == repo.resolve() + + +def test_git_toplevel_none_outside_repo(tmp_path): + assert _git_toplevel(tmp_path / "not-a-repo") is None + + +def test_git_branch_exists(tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + assert _git_branch_exists(repo, "main") is True + assert _git_branch_exists(repo, "no-such-branch") is False + + +def test_git_common_dir_and_git_dir_agree_on_plain_repo(tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + assert _git_common_dir(repo) == _git_dir(repo) + + +def test_git_current_branch(tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + assert _git_current_branch(repo) == "main" + + +def test_is_linked_worktree_checkout_false_for_plain_repo(tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + assert _is_linked_worktree_checkout(repo) is False + + +def test_nearest_existing_path_walks_up(tmp_path): + existing = tmp_path / "a" + existing.mkdir() + deep = existing / "b" / "c" / "d" + assert _nearest_existing_path(deep) == existing + + +def test_repo_root_for_worktree_target(tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + target = repo / ".worktrees" / "t1" + assert _repo_root_for_worktree_target(target) == repo.resolve() + + +# --------------------------------------------------------------------------- +# resolve_workspace +# --------------------------------------------------------------------------- + + +def test_resolve_workspace_scratch_creates_dir(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="scratch task") + task = kb.get_task(conn, t) + ws = resolve_workspace(task) + assert ws.exists() + assert ws.name == t + + +def test_resolve_workspace_rejects_relative_dir_path(kanban_home): + with kb.connect() as conn: + t = kb.create_task( + conn, title="rel", workspace_kind="dir", workspace_path="relative/path", + ) + task = kb.get_task(conn, t) + with pytest.raises(ValueError, match="absolute"): + resolve_workspace(task) + + +def test_resolve_workspace_dir_creates_absolute_path(kanban_home, tmp_path): + target = tmp_path / "workdir" + with kb.connect() as conn: + t = kb.create_task( + conn, title="dir task", workspace_kind="dir", workspace_path=str(target), + ) + task = kb.get_task(conn, t) + ws = resolve_workspace(task) + assert ws == target + assert ws.exists() + + +def test_resolve_workspace_unknown_kind_raises(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="bogus") # create_task validates the kind, + # so corrupt the row directly to simulate a kind no resolver knows. + conn.execute( + "UPDATE tasks SET workspace_kind = 'teleport', workspace_path = ? " + "WHERE id = ?", + (str(kanban_home / "x"), t), + ) + conn.commit() + task = kb.get_task(conn, t) + with pytest.raises(ValueError, match="unknown workspace_kind"): + resolve_workspace(task) + + +def test_resolve_workspace_worktree_materializes(kanban_home, tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + target = repo / ".worktrees" / "wt-task" + with kb.connect() as conn: + t = kb.create_task( + conn, + title="wt", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="wt/wt-task", + ) + task = kb.get_task(conn, t) + ws = resolve_workspace(task) + assert ws == target + assert ws.exists() + # The worktree must share the repo's common dir (a real linked worktree). + repo_common = subprocess.run( + ["git", "-C", str(repo), "rev-parse", "--path-format=absolute", "--git-common-dir"], + check=True, capture_output=True, text=True, + ).stdout.strip() + ws_common = subprocess.run( + ["git", "-C", str(ws), "rev-parse", "--path-format=absolute", "--git-common-dir"], + check=True, capture_output=True, text=True, + ).stdout.strip() + assert ws_common == repo_common + + +def test_set_workspace_path_and_branch_persist(kanban_home): + with kb.connect() as conn: + t = kb.create_task(conn, title="persist") + set_workspace_path(conn, t, "/some/absolute/ws") + set_branch_name(conn, t, "feature/x") + row = conn.execute( + "SELECT workspace_path, branch_name FROM tasks WHERE id = ?", (t,), + ).fetchone() + assert row["workspace_path"] == "/some/absolute/ws" + assert row["branch_name"] == "feature/x" + + +def test_ensure_git_worktree_creates_branch_and_checkout(kanban_home, tmp_path): + repo = tmp_path / "repo" + _init_git_repo(repo) + target = repo / ".worktrees" / "ensured" + _ensure_git_worktree(repo, target, "wt/ensured") + assert target.exists() + assert _git_current_branch(target) == "wt/ensured" + + +def test_resolve_worktree_workspace_requires_absolute(kanban_home): + with kb.connect() as conn: + t = kb.create_task( + conn, title="rel wt", workspace_kind="worktree", workspace_path="relative", + ) + task = kb.get_task(conn, t) + with pytest.raises(ValueError, match="absolute"): + _resolve_worktree_workspace(task) diff --git a/tests/tools/test_windows_native_support.py b/tests/tools/test_windows_native_support.py index 81be4319b557..d8f2d62267a4 100644 --- a/tests/tools/test_windows_native_support.py +++ b/tests/tools/test_windows_native_support.py @@ -543,7 +543,9 @@ class TestKanbanWaitpidWindowsGuard: def test_source_gates_waitpid_loop(self): root = Path(__file__).resolve().parents[2] - source = (root / "hermes_cli" / "kanban_db.py").read_text(encoding="utf-8") + # The waitpid loop lives in the worker-process module extracted from + # kanban_db.py (godfile-kill extraction PR); grep the moved module. + source = (root / "hermes_cli" / "worker_process.py").read_text(encoding="utf-8") # Find the waitpid call and confirm it's inside a POSIX gate. idx = source.find("os.waitpid(-1, os.WNOHANG)") assert idx > 0, "waitpid call must exist"