From 5985eb812c55902a8acf2307b0d8f8ec5d2a3648 Mon Sep 17 00:00:00 2001 From: Royce Date: Wed, 5 Aug 2026 23:06:04 -0700 Subject: [PATCH 1/3] [verified] feat: add safe Kanban worktree lifecycle --- .../emails/royce.personalassistant@gmail.com | 1 + docs/examples/kanban-worktree-exceptions.yaml | 12 + hermes_cli/config_defaults.py | 14 + hermes_cli/kanban.py | 162 +- hermes_cli/kanban_db.py | 263 ++- hermes_cli/kanban_workspaces.py | 1419 +++++++++++++++++ .../test_kanban_worktree_isolation.py | 30 + .../test_kanban_worktree_lifecycle.py | 1387 ++++++++++++++++ website/docs/user-guide/features/kanban.md | 78 +- 9 files changed, 3349 insertions(+), 17 deletions(-) create mode 100644 contributors/emails/royce.personalassistant@gmail.com create mode 100644 docs/examples/kanban-worktree-exceptions.yaml create mode 100644 hermes_cli/kanban_workspaces.py create mode 100644 tests/hermes_cli/test_kanban_worktree_lifecycle.py diff --git a/contributors/emails/royce.personalassistant@gmail.com b/contributors/emails/royce.personalassistant@gmail.com new file mode 100644 index 0000000000000..f56d47fe0ea65 --- /dev/null +++ b/contributors/emails/royce.personalassistant@gmail.com @@ -0,0 +1 @@ +roycepersonalassistant diff --git a/docs/examples/kanban-worktree-exceptions.yaml b/docs/examples/kanban-worktree-exceptions.yaml new file mode 100644 index 0000000000000..070c3d2fd11d7 --- /dev/null +++ b/docs/examples/kanban-worktree-exceptions.yaml @@ -0,0 +1,12 @@ +# Copy this file outside the source checkout and replace the placeholder paths +# before running the protected-exception import command. The importer only +# registers a workspace when Git confirms that the exact path is an existing +# linked worktree of repo_path; it never creates, removes, or force-adopts one. +version: 1 +exceptions: + - policy_id: pkgpulse-deepsec-operational + repo_path: /path/to/PkgPulse + workspace_path: /path/to/hermes/.worktrees/operational/pkgpulse-deepsec/PkgPulse + owner_profile: operator + purpose: operational + expires_at: null diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 41c493383ba5f..7f4106b862df3 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2309,6 +2309,20 @@ # worker process (if still running host-locally) is terminated # before the reclaim. 0 disables stale detection entirely. "dispatch_stale_timeout_seconds": 14400, + # Canonical worktree-lifecycle ownership and pressure thresholds. + # The default is intentionally report-only: creation records ownership + # and emits warnings when thresholds are crossed, but never rejects a + # valid single-owner workspace solely because the host is busy. Hard + # enforcement requires an explicit operator policy change after dry-run + # inventory/janitor evidence has been reviewed. + "worktree_lifecycle": { + "enforcement": "report_only", + "warning_worktree_count_per_repo": 20, + "warning_estimated_bytes_per_repo": 100 * 1024 * 1024 * 1024, + "warning_free_space_floor_bytes": 50 * 1024 * 1024 * 1024, + "approved_roots": [], + "exception_config": "", + }, }, # execute_code settings — controls the tool used for programmatic tool calls. diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index a08cb8f9b4076..2824a30bc4ae1 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -22,7 +22,7 @@ import sys import time from pathlib import Path -from typing import Any, Optional +from typing import Any, Optional, cast from hermes_cli import kanban_db as kb from hermes_cli import kanban_swarm as ks @@ -68,6 +68,7 @@ def _task_to_dict(t: kb.Task) -> dict[str, Any]: "workspace_kind": t.workspace_kind, "workspace_path": t.workspace_path, "branch_name": t.branch_name, + "workspace_replacement_reason": t.workspace_replacement_reason, "project_id": t.project_id, "created_by": t.created_by, "created_at": t.created_at, @@ -338,6 +339,12 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu "(default: scratch)") p_create.add_argument("--branch", default=None, help="Branch name for worktree tasks, e.g. wt/t6-wire") + p_create.add_argument( + "--workspace-replacement-reason", + default=None, + help="Required audit reason when replacing another active worktree " + "for the same task and repository", + ) p_create.add_argument("--project", default=None, help="Link to a project (id or slug). Anchors the task's " "worktree under the project's primary repo with a " @@ -698,6 +705,14 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu default=None, help="Permanently delete already-archived task ids from the board", ) + p_archive.add_argument( + "--workspace-disposition", + default=None, + help=( + "Required when directly archiving a task-owned worktree that has " + "not already been classified" + ), + ) # --- tail --- p_tail = sub.add_parser("tail", help="Follow a task's event stream") @@ -761,6 +776,60 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu ) p_stats.add_argument("--json", action="store_true") + # --- worktrees inventory / janitor --- + p_worktrees = sub.add_parser( + "worktrees", + help="Inventory and safely reconcile Kanban-managed Git worktrees", + ) + worktree_sub = p_worktrees.add_subparsers(dest="worktree_action", required=True) + p_wt_inventory = worktree_sub.add_parser( + "inventory", + help="Report deterministic Git/registry/task lifecycle findings", + ) + p_wt_inventory.add_argument( + "--repo", action="append", required=True, + help="Git repository to inventory (repeatable)", + ) + p_wt_inventory.add_argument( + "--approved-root", action="append", default=[], + help="Approved parent for secondary worktrees (repeatable)", + ) + p_wt_inventory.add_argument( + "--exception-config", + default=None, + help="Versioned YAML policy file for protected operational worktrees", + ) + p_wt_inventory.add_argument("--json", action="store_true") + p_wt_janitor = worktree_sub.add_parser( + "janitor", + help="Plan safe removals (dry-run only; never deletes worktrees)", + ) + p_wt_janitor.add_argument( + "--repo", action="append", required=True, + help="Git repository to evaluate (repeatable)", + ) + p_wt_janitor.add_argument( + "--approved-root", action="append", required=True, + help="Allowlisted parent for removable worktrees (repeatable)", + ) + p_wt_janitor.add_argument("--json", action="store_true") + p_wt_exceptions = worktree_sub.add_parser( + "exceptions", + help="Manage explicit protected-worktree policies", + ) + exception_sub = p_wt_exceptions.add_subparsers( + dest="exception_action", required=True + ) + p_wt_exception_import = exception_sub.add_parser( + "import", + help="Register existing Git worktrees from a versioned policy file", + ) + p_wt_exception_import.add_argument( + "--config", required=True, + help="Versioned YAML policy file to validate and import", + ) + p_wt_exception_import.add_argument("--json", action="store_true") + # --- notify subscribe / list / remove --- p_nsub = sub.add_parser( "notify-subscribe", @@ -1073,6 +1142,7 @@ def kanban_command(args: argparse.Namespace) -> int: "daemon": _cmd_daemon, "watch": _cmd_watch, "stats": _cmd_stats, + "worktrees": _cmd_worktrees, "log": _cmd_log, "runs": _cmd_runs, "heartbeat": _cmd_heartbeat, @@ -1505,6 +1575,9 @@ def _cmd_create(args: argparse.Namespace) -> int: workspace_kind=ws_kind, workspace_path=ws_path, branch_name=branch_name, + workspace_replacement_reason=getattr( + args, "workspace_replacement_reason", None + ), project_id=getattr(args, "project", None), tenant=args.tenant, priority=args.priority, @@ -2407,7 +2480,11 @@ def _cmd_archive(args: argparse.Namespace) -> int: print(f"Deleted {tid}") return 0 if not failed else 1 for tid in ids: - if not kb.archive_task(conn, tid): + if not kb.archive_task( + conn, + tid, + workspace_disposition=getattr(args, "workspace_disposition", None), + ): failed.append(tid) print(f"cannot archive {tid}", file=sys.stderr) else: @@ -2730,6 +2807,87 @@ def _cmd_watch(args: argparse.Namespace) -> int: return 0 +def _cmd_worktrees(args: argparse.Namespace) -> int: + from hermes_cli import kanban_workspaces as workspaces + + action = getattr(args, "worktree_action", None) + if action == "inventory": + report = cast( + dict[str, Any], + workspaces.reconcile_inventory( + repo_paths=getattr(args, "repo", []) or [], + approved_roots=getattr(args, "approved_root", []) or [], + exception_config=getattr(args, "exception_config", None), + ), + ) + elif action == "janitor": + report = cast( + dict[str, Any], + workspaces.plan_janitor( + repo_paths=getattr(args, "repo", []) or [], + approved_roots=getattr(args, "approved_root", []) or [], + ), + ) + elif action == "exceptions" and getattr(args, "exception_action", None) == "import": + report = cast( + dict[str, Any], + workspaces.import_protected_exceptions(args.config), + ) + else: + print(f"kanban worktrees: unknown action {action!r}", file=sys.stderr) + return 2 + if getattr(args, "json", False): + print(json.dumps(report, indent=2, ensure_ascii=False, sort_keys=True)) + return 0 + if action == "exceptions": + print( + "Protected worktree import: " + f"{len(report['imported'])} imported, {len(report['skipped'])} skipped" + ) + return 0 + if action == "janitor": + print( + "Worktree janitor dry-run: " + f"{len(report['candidates'])} candidate(s), " + f"{len(report['excluded'])} excluded, 0 removed" + ) + return 0 + totals = report["totals"] + print( + "Worktree inventory: " + f"{totals['git_worktrees']} Git worktree(s), " + f"{totals['registry_records']} registry record(s), " + f"{totals['estimated_bytes']} estimated byte(s)" + ) + print("Repositories:") + for repo in report["per_repo"]: + print( + f" {repo['repo_path']}: {repo['worktree_count']} worktree(s), " + f"{repo['estimated_bytes']} estimated byte(s)" + ) + print("Findings:") + for finding in ( + "unowned_paths", + "duplicate_task_repo", + "duplicate_heads", + "terminal_without_disposition", + "dirty_terminal", + "outside_approved_root", + "missing_expired_protected_exceptions", + ): + values = report[finding] + print(f" {finding} ({len(values)}):") + for value in values: + if isinstance(value, str): + rendered = value + else: + rendered = json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + print(f" {rendered}") + return 0 + + def _cmd_stats(args: argparse.Namespace) -> int: with kb.connect_closing() as conn: stats = kb.board_stats(conn) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 113e34842ec54..437a0aa64f5d0 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -921,6 +921,7 @@ class Task: claim_expires: Optional[int] tenant: Optional[str] branch_name: Optional[str] = None + workspace_replacement_reason: Optional[str] = None project_id: Optional[str] = None result: Optional[str] = None idempotency_key: Optional[str] = None @@ -1020,6 +1021,12 @@ def from_row(cls, row: sqlite3.Row) -> "Task": workspace_kind=row["workspace_kind"], workspace_path=row["workspace_path"], branch_name=row["branch_name"] if "branch_name" in keys else None, + workspace_replacement_reason=( + row["workspace_replacement_reason"] + if "workspace_replacement_reason" in keys + and row["workspace_replacement_reason"] + else None + ), project_id=row["project_id"] if "project_id" in keys else None, claim_lock=row["claim_lock"], claim_expires=row["claim_expires"], @@ -1196,6 +1203,9 @@ class Event: workspace_kind TEXT NOT NULL DEFAULT 'scratch', workspace_path TEXT, branch_name TEXT, + -- Required audit reason when a task/repository already owns a different + -- active worktree and creation is intentionally replacing it. + workspace_replacement_reason TEXT, -- Optional link to a first-class Project (hermes_cli/projects_db). When set, -- the task's worktree is anchored under the project's primary repo with a -- deterministic branch name instead of a random wt/ fallback. @@ -2175,6 +2185,37 @@ def connect( path = kanban_db_path(board=board) path.parent.mkdir(parents=True, exist_ok=True) + requested_board = _normalize_board_slug(board) + if requested_board is None: + resolved_path = path.expanduser().resolve(strict=False) + default_path = (kanban_home() / "kanban.db").resolve(strict=False) + if resolved_path == default_path: + requested_board = DEFAULT_BOARD + else: + matches = [ + str(item["slug"]) + for item in list_boards() + if str(item["slug"]) != DEFAULT_BOARD + and (board_dir(str(item["slug"])) / "kanban.db").resolve( + strict=False + ) == resolved_path + ] + requested_board = matches[0] if len(matches) == 1 else get_current_board() + + def _stamp_board_identity(conn: sqlite3.Connection) -> sqlite3.Connection: + # HERMES_KANBAN_DB can pin multiple logical boards to the same path, + # so reverse-mapping a connection path is not a reliable identity. + conn.execute( + "CREATE TEMP TABLE IF NOT EXISTS hermes_connection_context " + "(board_id TEXT NOT NULL)" + ) + conn.execute("DELETE FROM temp.hermes_connection_context") + conn.execute( + "INSERT INTO temp.hermes_connection_context (board_id) VALUES (?)", + (requested_board,), + ) + return conn + # Fast path: once THIS process has initialized this path, the expensive # first-open work (header validation, integrity probe, schema + additive # migrations) is already done and cached in _INITIALIZED_PATHS. Acquiring @@ -2201,7 +2242,7 @@ def connect( except Exception: conn.close() raise - return conn + return _stamp_board_identity(conn) with _cross_process_init_lock(path): # Read-only file/sidecar preflight (port of kilocode#12508) — @@ -2255,7 +2296,7 @@ def connect( except Exception: conn.close() raise - return conn + return _stamp_board_identity(conn) @contextlib.contextmanager @@ -2335,6 +2376,13 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: _add_column_if_missing(conn, "tasks", "result", "result TEXT") if "branch_name" not in cols: _add_column_if_missing(conn, "tasks", "branch_name", "branch_name TEXT") + if "workspace_replacement_reason" not in cols: + _add_column_if_missing( + conn, + "tasks", + "workspace_replacement_reason", + "workspace_replacement_reason TEXT", + ) if "project_id" not in cols: _add_column_if_missing(conn, "tasks", "project_id", "project_id TEXT") if "idempotency_key" not in cols: @@ -2888,6 +2936,7 @@ def create_task( workspace_kind: str = "scratch", workspace_path: Optional[str] = None, branch_name: Optional[str] = None, + workspace_replacement_reason: Optional[str] = None, tenant: Optional[str] = None, priority: int = 0, parents: Iterable[str] = (), @@ -2965,6 +3014,14 @@ def create_task( ) if branch_name is not None: branch_name = str(branch_name).strip() or None + if workspace_replacement_reason is not None: + workspace_replacement_reason = ( + str(workspace_replacement_reason).strip() or None + ) + if workspace_replacement_reason and workspace_kind != "worktree": + raise ValueError( + "workspace_replacement_reason is only valid for worktree workspaces" + ) if branch_name and workspace_kind != "worktree": raise ValueError("branch_name is only valid for worktree workspaces") @@ -3213,12 +3270,13 @@ def create_task( INSERT INTO tasks ( id, title, body, assignee, status, priority, created_by, created_at, workspace_kind, workspace_path, - branch_name, project_id, tenant, idempotency_key, + branch_name, workspace_replacement_reason, + project_id, tenant, idempotency_key, max_runtime_seconds, skills, max_retries, model_override, provider_override, reasoning_effort, goal_mode, goal_max_turns, session_id - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( task_id, @@ -3232,6 +3290,7 @@ def create_task( workspace_kind, workspace_path, branch_name, + workspace_replacement_reason, project_id, tenant, idempotency_key, @@ -3263,6 +3322,7 @@ def create_task( "workspace_kind": workspace_kind, "workspace_path": workspace_path, "branch_name": branch_name, + "workspace_replacement_reason": workspace_replacement_reason, "project_id": project_id, "skills": list(skills_list) if skills_list else None, "goal_mode": bool(goal_mode) or None, @@ -4900,6 +4960,37 @@ def complete_task( else: verified_cards = [] + # Worktree lifecycle gate. A task that owns registry worktrees cannot + # become terminal until the caller classifies each workspace explicitly. + # Legacy/unregistered worktree tasks remain compatible (no owned rows = + # no gate), while every newly materialized worktree is registered by + # ``_ensure_registered_git_worktree`` above. + _terminal_task = get_task(conn, task_id) + _terminal_plans = [] + if ( + _terminal_task is not None + and _terminal_task.status in {"running", "ready", "blocked"} + and _terminal_task.workspace_kind == "worktree" + ): + from hermes_cli import kanban_workspaces as _workspaces + + _terminal_plans = _workspaces.prepare_terminal_disposition( + task_id=task_id, + board_id=_board_for_connection(conn), + disposition=( + metadata.get("workspace_disposition") + if isinstance(metadata, dict) + else None + ), + pr_numbers=( + metadata.get("workspace_pr_numbers") + if isinstance(metadata, dict) + else None + ), + ) + if _terminal_plans: + _workspaces.attach_registry(conn) + metadata = _merge_completion_prose_artifacts( conn, task_id, metadata, summary=summary, result=result, ) @@ -4941,6 +5032,12 @@ def complete_task( ) if cur.rowcount != 1: return False + if _terminal_plans: + from hermes_cli import kanban_workspaces as _workspaces + + _workspaces.apply_terminal_disposition_in_transaction( + conn, _terminal_plans + ) if isinstance(metadata, dict): _persist_scratch_completion_artifacts(conn, task_id, metadata) for stored_path in metadata.pop("_staged_artifacts", []): @@ -6288,7 +6385,51 @@ def decompose_triage_task( return child_ids -def archive_task(conn: sqlite3.Connection, task_id: str) -> bool: +def _board_for_connection(conn: sqlite3.Connection) -> str: + """Resolve the board identity from the exact SQLite connection in use.""" + try: + row = conn.execute( + "SELECT board_id FROM temp.hermes_connection_context" + ).fetchone() + except sqlite3.Error: + row = None + if row is not None: + board_id = _normalize_board_slug(str(row["board_id"])) + if board_id: + return board_id + main_path = "" + for row in conn.execute("PRAGMA database_list").fetchall(): + if str(row[1]) == "main": + main_path = str(row[2] or "") + break + if not main_path: + raise RuntimeError("cannot resolve Kanban board from an in-memory connection") + resolved = Path(main_path).expanduser().resolve(strict=False) + for board in list_boards(): + slug = str(board["slug"]) + if kanban_db_path(slug).resolve(strict=False) == resolved: + return slug + raise RuntimeError(f"cannot resolve Kanban board for database {resolved}") + + +def archive_task( + conn: sqlite3.Connection, + task_id: str, + *, + workspace_disposition: object = None, +) -> bool: + task = get_task(conn, task_id) + terminal_plans = [] + if task is not None and task.status != "archived" and task.workspace_kind == "worktree": + from hermes_cli import kanban_workspaces as _workspaces + + terminal_plans = _workspaces.prepare_terminal_disposition( + task_id=task_id, + board_id=_board_for_connection(conn), + disposition=workspace_disposition, + ) + if terminal_plans: + _workspaces.attach_registry(conn) with write_txn(conn): cur = conn.execute( "UPDATE tasks SET status = 'archived', " @@ -6298,6 +6439,12 @@ def archive_task(conn: sqlite3.Connection, task_id: str) -> bool: ) if cur.rowcount != 1: return False + if terminal_plans: + from hermes_cli import kanban_workspaces as _workspaces + + _workspaces.apply_terminal_disposition_in_transaction( + conn, terminal_plans + ) # 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. @@ -6515,6 +6662,49 @@ def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> Non ) +def _ensure_registered_git_worktree( + task: Task, + repo_root: Path, + target: Path, + branch_name: str, + *, + board: Optional[str] = None, +) -> Path: + """Reserve lifecycle ownership, materialize, then verify one worktree. + + The existing branch/anchor/materialization behavior stays in + ``_ensure_git_worktree``. This wrapper adds the lifecycle contract around + that seam: the registry reservation commits before ``git worktree add`` and + is finalized only after Git reports a usable checkout. + """ + from hermes_cli import kanban_workspaces as workspaces + + board_id = board or get_current_board() + reservation = workspaces.reserve_workspace( + repo_path=repo_root, + workspace_path=target, + task_id=task.id, + board_id=board_id, + owner_profile=task.assignee, + branch=branch_name, + replacement_reason=getattr(task, "workspace_replacement_reason", None), + ) + reserved_target = Path(reservation.workspace_path) + try: + _ensure_git_worktree(repo_root, reserved_target, branch_name) + actual_branch = _git_current_branch(reserved_target) + if actual_branch != branch_name: + raise RuntimeError( + f"worktree branch mismatch at {reserved_target}: expected " + f"{branch_name!r}, found {actual_branch or 'detached HEAD'!r}" + ) + workspaces.verify_workspace(reservation.workspace_id, reserved_target) + except Exception: + workspaces.mark_creation_failed(reservation.workspace_id) + raise + return reserved_target + + def _resolve_worktree_workspace( task: Task, *, board: Optional[str] = None ) -> tuple[Path, str]: @@ -6555,7 +6745,9 @@ def _resolve_worktree_workspace( 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) + target = _ensure_registered_git_worktree( + task, repo_root, target, branch_name, board=board_slug + ) return target, branch_name requested = Path(task.workspace_path).expanduser() @@ -6569,7 +6761,40 @@ def _resolve_worktree_workspace( 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 + matched_branch = actual_branch or branch_name + existing_root = _repo_root_for_worktree_target(requested.parent) + if existing_root is None: + raise ValueError( + f"task {task.id} worktree path {task.workspace_path!r} is linked " + "but its primary repository cannot be resolved for lifecycle ownership" + ) + registered = _ensure_registered_git_worktree( + task, + existing_root, + requested_resolved, + matched_branch, + board=board, + ) + return registered.resolve(strict=False), matched_branch + # A path already registered to this exact task is not a sibling's + # checkout to route around: a different current branch is ownership + # drift. Fail closed without creating a replacement or downgrading the + # existing registry record. + from hermes_cli import kanban_workspaces as _workspaces + + registered_paths = { + Path(record.workspace_path).resolve(strict=False) + for record in _workspaces.list_workspace_records( + task_id=task.id, + board_id=board or get_current_board(), + ) + if record.status != "creation_failed" + } + if requested_resolved in registered_paths: + raise RuntimeError( + f"worktree branch mismatch at {requested_resolved}: expected " + f"{branch_name!r}, found {actual_branch or 'detached HEAD'!r}" + ) # 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 @@ -6581,17 +6806,25 @@ def _resolve_worktree_workspace( 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) + fallback = _ensure_registered_git_worktree( + task, fallback_root, fallback, branch_name, board=board + ) 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 + # No distinct safe fallback exists (including when the occupied path + # is this task's canonical target). Never silently run on another + # branch and never auto-switch a checkout that may contain unique work. + raise ValueError( + f"task {task.id} linked worktree target {str(requested)!r} has branch " + f"mismatch: expected {branch_name!r}, found " + f"{actual_branch or 'detached HEAD'!r}" + ) 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) + target = _ensure_registered_git_worktree( + task, repo_root, target, branch_name, board=board + ) return target, branch_name repo_root = _repo_root_for_worktree_target(requested.parent) @@ -6600,7 +6833,9 @@ def _resolve_worktree_workspace( 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) + requested = _ensure_registered_git_worktree( + task, repo_root, requested, branch_name, board=board + ) return requested, branch_name diff --git a/hermes_cli/kanban_workspaces.py b/hermes_cli/kanban_workspaces.py new file mode 100644 index 0000000000000..cc4508efa5fea --- /dev/null +++ b/hermes_cli/kanban_workspaces.py @@ -0,0 +1,1419 @@ +"""Canonical, profile-aware registry for Kanban-managed Git worktrees. + +The registry is shared across profiles and boards at +``/kanban/workspace-registry.db``. Git remains the authority for +which linked worktrees physically exist; this database is the authority for +ownership, lifecycle policy, and cleanup disposition. Unknown Git paths are +reported by reconciliation and are never adopted or deleted implicitly. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import logging +import os +import shutil +import sqlite3 +import subprocess +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable, Optional + +import yaml + + +_log = logging.getLogger(__name__) + + +VALID_DISPOSITION_PREFIXES = { + "retired", + "retained_until", + "preserved_dirty", + "operational_exception", +} +ACTIVE_REGISTRY_STATUSES = { + "reserved", + "active", + "blocked", + "terminal_clean", + "terminal_dirty", + "superseded", + "retirement_queued", + "protected", +} + + +@dataclass(frozen=True) +class WorkspaceRecord: + workspace_id: str + repo_id: str + repo_path: str + workspace_path: str + task_id: Optional[str] + board_id: Optional[str] + owner_profile: Optional[str] + purpose: str + branch: Optional[str] + head_sha: Optional[str] + pr_numbers: tuple[int, ...] + status: str + cleanup_policy: str + retention_condition: Optional[str] + disposition: Optional[str] + dirty_manifest_hash: Optional[str] + estimated_bytes: int + exception_policy_id: Optional[str] + exception_expires_at: Optional[int] + replacement_reason: Optional[str] + created_at: int + last_used_at: int + last_verified_at: Optional[int] + + @classmethod + def from_row(cls, row: sqlite3.Row) -> "WorkspaceRecord": + raw_prs = str(row["pr_numbers"] or "") + prs = tuple(int(value) for value in raw_prs.split(",") if value.strip()) + return cls( + workspace_id=row["workspace_id"], + repo_id=row["repo_id"], + repo_path=row["repo_path"], + workspace_path=row["workspace_path"], + task_id=row["task_id"], + board_id=row["board_id"], + owner_profile=row["owner_profile"], + purpose=row["purpose"], + branch=row["branch"], + head_sha=row["head_sha"], + pr_numbers=prs, + status=row["status"], + cleanup_policy=row["cleanup_policy"], + retention_condition=row["retention_condition"], + disposition=row["disposition"], + dirty_manifest_hash=row["dirty_manifest_hash"], + estimated_bytes=int(row["estimated_bytes"] or 0), + exception_policy_id=row["exception_policy_id"], + exception_expires_at=row["exception_expires_at"], + replacement_reason=row["replacement_reason"], + created_at=int(row["created_at"]), + last_used_at=int(row["last_used_at"]), + last_verified_at=row["last_verified_at"], + ) + + +@dataclass(frozen=True) +class _TerminalDispositionPlan: + record: WorkspaceRecord + disposition: str + status: str + dirty_manifest_hash: Optional[str] + exception_policy_id: Optional[str] + retention_condition: Optional[str] + head_sha: Optional[str] + branch: Optional[str] + estimated_bytes: int + pr_numbers: tuple[int, ...] + exception_expires_at: Optional[int] + verified_at: int + + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS workspaces ( + workspace_id TEXT PRIMARY KEY, + repo_id TEXT NOT NULL, + repo_path TEXT NOT NULL, + workspace_path TEXT NOT NULL UNIQUE, + task_id TEXT, + board_id TEXT, + owner_profile TEXT, + purpose TEXT NOT NULL, + branch TEXT, + head_sha TEXT, + pr_numbers TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + cleanup_policy TEXT NOT NULL, + retention_condition TEXT, + disposition TEXT, + dirty_manifest_hash TEXT, + estimated_bytes INTEGER NOT NULL DEFAULT 0, + exception_policy_id TEXT, + exception_expires_at INTEGER, + replacement_reason TEXT, + created_at INTEGER NOT NULL, + last_used_at INTEGER NOT NULL, + last_verified_at INTEGER +); +CREATE INDEX IF NOT EXISTS idx_workspaces_task_repo + ON workspaces(board_id, task_id, repo_id); +CREATE INDEX IF NOT EXISTS idx_workspaces_repo_head + ON workspaces(repo_id, head_sha); +CREATE INDEX IF NOT EXISTS idx_workspaces_status + ON workspaces(status); +""" + + +def registry_path() -> Path: + from hermes_cli import kanban_db + + return kanban_db.kanban_home() / "kanban" / "workspace-registry.db" + + +def connect_registry() -> sqlite3.Connection: + path = registry_path() + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(path), isolation_level=None, timeout=30) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA busy_timeout=30000") + conn.executescript(_SCHEMA) + columns = { + str(row[1]) + for row in conn.execute("PRAGMA table_info(workspaces)").fetchall() + } + if "exception_expires_at" not in columns: + conn.execute( + "ALTER TABLE workspaces ADD COLUMN exception_expires_at INTEGER" + ) + return conn + + +def _canonical(path: Path | str) -> str: + return str(Path(path).expanduser().resolve(strict=False)) + + +def _canonical_repo(path: Path | str) -> str: + """Return one identity path for a repository from any of its checkouts.""" + candidate = _canonical(path) + raw = _git(candidate, "worktree", "list", "--porcelain") + if raw: + for line in raw.splitlines(): + if line.startswith("worktree "): + return _canonical(line.removeprefix("worktree ")) + return candidate + + +def _repo_id(repo_path: str) -> str: + canonical = _canonical_repo(repo_path) + return "r_" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16] + + +def _git(path: Path | str, *args: str) -> Optional[str]: + try: + result = subprocess.run( + ["git", "-C", str(path), *args], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=30, + check=False, + ) + except Exception: + return None + if result.returncode != 0: + return None + return (result.stdout or "").strip() + + +def dirty_manifest_hash(path: Path | str) -> Optional[str]: + raw = _git(path, "status", "--porcelain=v1", "-z", "--untracked-files=all") + if raw is None: + return None + return hashlib.sha256(raw.encode("utf-8", errors="surrogateescape")).hexdigest() + + +def estimated_bytes(path: Path | str) -> int: + root = Path(path) + if not root.exists(): + return 0 + total = 0 + try: + for current, dirs, files in os.walk(root, followlinks=False): + dirs[:] = sorted(d for d in dirs if d != ".git") + for name in sorted(files): + candidate = Path(current) / name + try: + if not candidate.is_symlink(): + total += candidate.stat().st_size + except OSError: + continue + except OSError: + return total + return total + + +def list_workspace_records( + *, + task_id: Optional[str] = None, + board_id: Optional[str] = None, + repo_path: Optional[Path | str] = None, +) -> list[WorkspaceRecord]: + clauses: list[str] = [] + params: list[object] = [] + if task_id is not None: + clauses.append("task_id = ?") + params.append(task_id) + if board_id is not None: + clauses.append("board_id = ?") + params.append(board_id) + if repo_path is not None: + clauses.append("repo_id = ?") + params.append(_repo_id(_canonical_repo(repo_path))) + where = " WHERE " + " AND ".join(clauses) if clauses else "" + with contextlib.closing(connect_registry()) as conn: + rows = conn.execute( + "SELECT * FROM workspaces" + where + + " ORDER BY repo_path, workspace_path, workspace_id", + tuple(params), + ).fetchall() + return [WorkspaceRecord.from_row(row) for row in rows] + + +def creation_pressure_warnings(repo_path: Path | str) -> list[str]: + """Evaluate config-backed host/repository pressure thresholds.""" + try: + from hermes_cli.config import load_config + + config = load_config() + lifecycle = config.get("kanban", {}).get("worktree_lifecycle", {}) + except Exception: + lifecycle = {} + + def _positive(name: str, fallback: int) -> int: + try: + value = int(lifecycle.get(name, fallback)) + except (TypeError, ValueError): + return fallback + return value if value > 0 else fallback + + count_limit = _positive("warning_worktree_count_per_repo", 20) + byte_limit = _positive( + "warning_estimated_bytes_per_repo", 100 * 1024 * 1024 * 1024 + ) + free_floor = _positive( + "warning_free_space_floor_bytes", 50 * 1024 * 1024 * 1024 + ) + repo_path = _canonical_repo(repo_path) + entries = _git_worktrees(repo_path) + # Creation must not recursively walk every checkout just to emit a + # report-only warning. Daily inventory refreshes these cached estimates; + # creation reads the latest verified values in constant registry time. + total_bytes = sum( + record.estimated_bytes + for record in list_workspace_records(repo_path=repo_path) + if record.status != "creation_failed" + ) + warnings: list[str] = [] + if len(entries) >= count_limit: + warnings.append( + f"repository {_canonical(repo_path)} has {len(entries)} worktrees " + f"(warning threshold {count_limit})" + ) + if total_bytes >= byte_limit: + warnings.append( + f"repository {_canonical(repo_path)} worktrees use approximately " + f"{total_bytes} bytes (warning threshold {byte_limit})" + ) + try: + free = shutil.disk_usage(_canonical(repo_path)).free + except OSError: + free = None + if free is not None and free <= free_floor: + warnings.append( + f"filesystem free space is {free} bytes (warning floor {free_floor})" + ) + # The shipped contract is report-only. A future owner-approved change can + # add execution enforcement without changing these deterministic findings. + return warnings + + +def reserve_workspace( + *, + repo_path: Path | str, + workspace_path: Path | str, + task_id: str, + board_id: str, + owner_profile: Optional[str], + purpose: str = "implementation", + branch: Optional[str] = None, + cleanup_policy: str = "on_task_terminal", + retention_condition: Optional[str] = "task_terminal", + replacement_reason: Optional[str] = None, +) -> WorkspaceRecord: + """Reserve ownership before ``git worktree add`` can mutate the repo. + + A single compatible task/repository reservation is reused. Multiple active + rows, or one incompatible row, are ambiguous and fail closed unless the + task carries an explicit replacement reason. + """ + repo = _canonical_repo(repo_path) + target = _canonical(workspace_path) + rid = _repo_id(repo) + now = int(time.time()) + for warning in creation_pressure_warnings(repo): + _log.warning("kanban worktree lifecycle (report-only): %s", warning) + + with contextlib.closing(connect_registry()) as conn: + conn.execute("BEGIN IMMEDIATE") + try: + rows = conn.execute( + "SELECT * FROM workspaces WHERE board_id = ? AND task_id = ? " + "AND repo_id = ? AND status IN ('reserved', 'active') " + "ORDER BY workspace_path, workspace_id", + (board_id, task_id, rid), + ).fetchall() + compatible = [ + row for row in rows + if row["workspace_path"] == target + and row["owner_profile"] == owner_profile + and row["purpose"] == purpose + and row["branch"] == branch + and row["cleanup_policy"] == cleanup_policy + and row["retention_condition"] == retention_condition + ] + if len(rows) == 1 and len(compatible) == 1: + conn.execute( + "UPDATE workspaces SET last_used_at = ? " + "WHERE workspace_id = ?", + (now, compatible[0]["workspace_id"]), + ) + conn.execute("COMMIT") + return WorkspaceRecord.from_row( + conn.execute( + "SELECT * FROM workspaces WHERE workspace_id = ?", + (compatible[0]["workspace_id"],), + ).fetchone() + ) + if rows and not replacement_reason: + raise RuntimeError( + f"ambiguous worktree ownership for task {task_id} in repo {repo}: " + f"{len(rows)} active registry row(s); record an explicit " + "workspace replacement reason before creating another" + ) + + existing = conn.execute( + "SELECT * FROM workspaces WHERE workspace_path = ?", (target,) + ).fetchone() + if existing is not None: + exact_compatible_owner = ( + existing["task_id"] == task_id + and existing["board_id"] == board_id + and existing["repo_id"] == rid + and existing["status"] in {"reserved", "active"} + and existing["owner_profile"] == owner_profile + and existing["purpose"] == purpose + and existing["branch"] == branch + and existing["cleanup_policy"] == cleanup_policy + and existing["retention_condition"] == retention_condition + ) + if not exact_compatible_owner and not replacement_reason: + raise RuntimeError( + f"worktree path {target} is already owned by " + f"{existing['board_id']}:{existing['task_id']}; " + "record an explicit workspace replacement reason" + ) + if exact_compatible_owner: + conn.execute("COMMIT") + return WorkspaceRecord.from_row(existing) + raise RuntimeError( + f"replacement reason does not authorize stealing an existing " + f"workspace path: {target}; choose a new path" + ) + + workspace_id = "w_" + uuid.uuid4().hex[:16] + conn.execute( + "INSERT INTO workspaces (" + "workspace_id, repo_id, repo_path, workspace_path, task_id, board_id, " + "owner_profile, purpose, branch, status, cleanup_policy, " + "retention_condition, replacement_reason, created_at, last_used_at" + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'reserved', ?, ?, ?, ?, ?)", + ( + workspace_id, rid, repo, target, task_id, board_id, + owner_profile, purpose, branch, cleanup_policy, + retention_condition, replacement_reason, now, now, + ), + ) + row = conn.execute( + "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) + ).fetchone() + conn.execute("COMMIT") + return WorkspaceRecord.from_row(row) + except Exception: + conn.execute("ROLLBACK") + raise + + +def verify_workspace(workspace_id: str, workspace_path: Path | str) -> WorkspaceRecord: + path = _canonical(workspace_path) + now = int(time.time()) + head = _git(path, "rev-parse", "HEAD") + branch = _git(path, "branch", "--show-current") or None + dirty_hash = dirty_manifest_hash(path) + size = estimated_bytes(path) + with contextlib.closing(connect_registry()) as conn: + conn.execute("BEGIN IMMEDIATE") + try: + cur = conn.execute( + "UPDATE workspaces SET workspace_path = ?, branch = COALESCE(?, branch), " + "head_sha = ?, status = 'active', dirty_manifest_hash = ?, " + "estimated_bytes = ?, last_used_at = ?, last_verified_at = ? " + "WHERE workspace_id = ? AND status IN ('reserved', 'active')", + (path, branch, head, dirty_hash, size, now, now, workspace_id), + ) + if cur.rowcount != 1: + raise RuntimeError( + f"workspace reservation {workspace_id} is not reservable or active" + ) + row = conn.execute( + "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) + ).fetchone() + if row is None: + raise RuntimeError(f"unknown workspace reservation: {workspace_id}") + conn.execute("COMMIT") + return WorkspaceRecord.from_row(row) + except Exception: + conn.execute("ROLLBACK") + raise + + +def mark_creation_failed(workspace_id: str) -> None: + with contextlib.closing(connect_registry()) as conn: + conn.execute( + "UPDATE workspaces SET status = 'creation_failed', last_verified_at = ? " + "WHERE workspace_id = ? AND status = 'reserved'", + (int(time.time()), workspace_id), + ) + + +def _parse_disposition(value: object) -> tuple[str, Optional[str]]: + text = str(value or "").strip() + if text == "retired": + return "retired", None + prefix, separator, detail = text.partition(":") + if prefix not in VALID_DISPOSITION_PREFIXES or prefix == "retired": + raise RuntimeError( + "workspace disposition must be one of: retired, " + "retained_until:, preserved_dirty:, or " + "operational_exception:" + ) + if not separator or not detail.strip(): + raise RuntimeError(f"workspace disposition {prefix} requires a non-empty detail") + return prefix, detail.strip() + + +def _configured_exception_path() -> Optional[Path]: + try: + from hermes_cli.config import load_config + + raw = ( + load_config() + .get("kanban", {}) + .get("worktree_lifecycle", {}) + .get("exception_config", "") + ) + except Exception: + return None + text = os.path.expandvars(str(raw or "").strip()) + return Path(text).expanduser() if text else None + + +def _validate_recovery_detail(record: WorkspaceRecord, detail: str) -> None: + fields: dict[str, str] = {} + for item in detail.split(";"): + key, separator, value = item.partition("=") + if separator and key.strip() and value.strip(): + fields[key.strip()] = value.strip() + artifact = fields.get("artifact", "") + owner = fields.get("owner", "") + if not artifact or not owner: + raise RuntimeError( + "preserved_dirty requires artifact=;owner=" + ) + artifact_path = Path(artifact).expanduser() + if not artifact_path.is_absolute(): + artifact_path = Path(record.workspace_path) / artifact_path + if not artifact_path.exists(): + raise RuntimeError( + f"preserved_dirty recovery artifact does not exist: {artifact_path}" + ) + + +def _validate_operational_exception( + record: WorkspaceRecord, + policy_id: str, + *, + now: int, +) -> Optional[int]: + config_path = _configured_exception_path() + if config_path is None: + raise RuntimeError( + f"operational_exception:{policy_id} requires a current operational " + "exception policy configured at kanban.worktree_lifecycle.exception_config" + ) + policies = { + str(policy["policy_id"]): policy for policy in load_exception_config(config_path) + } + policy = policies.get(policy_id) + if policy is None or _exception_expired(policy, now=now): + raise RuntimeError( + f"operational_exception:{policy_id} does not name a current operational " + "exception policy" + ) + if ( + str(policy["repo_path"]) != _canonical_repo(record.repo_path) + or str(policy["workspace_path"]) != record.workspace_path + ): + raise RuntimeError( + f"operational_exception:{policy_id} policy does not match workspace " + f"{record.workspace_path}" + ) + return _exception_expires_at(policy) + + +def _normalize_pr_numbers(value: object) -> Optional[tuple[int, ...]]: + if value is None: + return None + if not isinstance(value, (list, tuple, set)): + raise RuntimeError("workspace_pr_numbers must be a list of positive integers") + numbers: set[int] = set() + for raw in value: + try: + number = int(str(raw)) + except (TypeError, ValueError) as exc: + raise RuntimeError( + "workspace_pr_numbers must be a list of positive integers" + ) from exc + if number <= 0: + raise RuntimeError("workspace_pr_numbers must contain positive integers") + numbers.add(number) + return tuple(sorted(numbers)) + + +def prepare_terminal_disposition( + *, + task_id: str, + board_id: str, + disposition: object, + pr_numbers: object = None, +) -> list[_TerminalDispositionPlan]: + """Validate terminal facts without mutating either lifecycle database.""" + owned = [ + record for record in list_workspace_records(task_id=task_id) + if record.board_id == board_id and record.status != "creation_failed" + ] + if not owned: + return [] + normalized_prs = _normalize_pr_numbers(pr_numbers) + by_id: dict[str, object] + if disposition is None or disposition == "": + if any(not record.disposition for record in owned): + raise RuntimeError( + f"task {task_id} owns {len(owned)} worktree(s); metadata must include " + "workspace_disposition (retired, retained_until:, " + "preserved_dirty:artifact=;owner=, or " + "operational_exception:)" + ) + by_id = { + record.workspace_id: str(record.disposition) for record in owned + } + elif isinstance(disposition, dict): + by_id = {str(key): value for key, value in disposition.items()} + missing = sorted( + record.workspace_id for record in owned + if record.workspace_id not in by_id + ) + extra = sorted(set(by_id) - {record.workspace_id for record in owned}) + if missing or extra: + raise RuntimeError( + "workspace_disposition mapping must cover exactly the task's owned " + f"workspace ids; missing={missing}, unknown={extra}" + ) + else: + by_id = {record.workspace_id: disposition for record in owned} + + plans: list[_TerminalDispositionPlan] = [] + now = int(time.time()) + for record in owned: + raw_text = str(by_id[record.workspace_id]).strip() + kind, detail = _parse_disposition(raw_text) + status_raw = _git( + record.workspace_path, + "status", + "--porcelain=v1", + "-z", + "--untracked-files=all", + ) + exists = Path(record.workspace_path).exists() + is_dirty = bool(status_raw) + if kind == "retired" and exists: + if is_dirty: + raise RuntimeError( + f"dirty worktree {record.workspace_path} cannot be marked retired; " + "use preserved_dirty with a recovery artifact and owner, or remove " + "it safely first" + ) + raise RuntimeError( + f"worktree {record.workspace_path} still exists and cannot be marked " + "retired; use retained_until: or remove it safely first" + ) + if kind == "preserved_dirty" and not is_dirty: + raise RuntimeError( + f"workspace {record.workspace_path} is clean; preserved_dirty would be " + "an inaccurate disposition" + ) + if kind == "preserved_dirty": + _validate_recovery_detail(record, detail or "") + if kind == "retained_until" and is_dirty: + raise RuntimeError( + f"dirty worktree {record.workspace_path} requires " + "preserved_dirty:artifact=;owner=; retained_until would " + "hide recovery-owned work" + ) + + exception_expires_at = record.exception_expires_at + if kind == "operational_exception": + exception_expires_at = _validate_operational_exception( + record, detail or "", now=now + ) + if kind == "retired": + lifecycle_status = "retired" + retention_condition = None + elif kind == "retained_until": + lifecycle_status = ( + "retirement_queued" if detail == "janitor" + else ("terminal_dirty" if is_dirty else "terminal_clean") + ) + retention_condition = detail + elif kind == "preserved_dirty": + lifecycle_status = "terminal_dirty" + retention_condition = "recovery_owned" + else: + lifecycle_status = "protected" + retention_condition = "operational_exception" + manifest = ( + hashlib.sha256((status_raw or "").encode("utf-8")).hexdigest() + if status_raw is not None else None + ) + plans.append( + _TerminalDispositionPlan( + record=record, + disposition=raw_text, + status=lifecycle_status, + dirty_manifest_hash=manifest, + exception_policy_id=( + detail if kind == "operational_exception" + else record.exception_policy_id + ), + retention_condition=retention_condition, + head_sha=_git(record.workspace_path, "rev-parse", "HEAD"), + branch=( + _git(record.workspace_path, "branch", "--show-current") or None + ), + estimated_bytes=estimated_bytes(record.workspace_path), + pr_numbers=( + normalized_prs if normalized_prs is not None else record.pr_numbers + ), + exception_expires_at=exception_expires_at, + verified_at=now, + ) + ) + return plans + + +def attach_registry(conn: sqlite3.Connection) -> None: + """Attach the initialized registry so task + lifecycle writes share a txn.""" + path = registry_path().expanduser().resolve(strict=False) + with contextlib.closing(connect_registry()): + pass + for row in conn.execute("PRAGMA database_list").fetchall(): + if str(row[1]) == "workspace_registry": + if Path(str(row[2])).expanduser().resolve(strict=False) != path: + raise RuntimeError("a different workspace registry is already attached") + return + conn.execute("ATTACH DATABASE ? AS workspace_registry", (str(path),)) + + +def _apply_terminal_plans( + conn: sqlite3.Connection, + plans: list[_TerminalDispositionPlan], + *, + table: str, +) -> None: + for plan in plans: + cur = conn.execute( + f"UPDATE {table} SET disposition = ?, status = ?, " + "dirty_manifest_hash = ?, exception_policy_id = ?, " + "exception_expires_at = ?, retention_condition = ?, head_sha = ?, " + "branch = ?, pr_numbers = ?, estimated_bytes = ?, last_used_at = ?, " + "last_verified_at = ? WHERE workspace_id = ? AND board_id = ? " + "AND task_id = ? AND repo_id = ? AND workspace_path = ? AND status = ?", + ( + plan.disposition, + plan.status, + plan.dirty_manifest_hash, + plan.exception_policy_id, + plan.exception_expires_at, + plan.retention_condition, + plan.head_sha, + plan.branch, + ",".join(str(number) for number in plan.pr_numbers), + plan.estimated_bytes, + plan.verified_at, + plan.verified_at, + plan.record.workspace_id, + plan.record.board_id, + plan.record.task_id, + plan.record.repo_id, + plan.record.workspace_path, + plan.record.status, + ), + ) + if cur.rowcount != 1: + raise RuntimeError( + f"workspace {plan.record.workspace_id} changed during terminal transition" + ) + + +def apply_terminal_disposition_in_transaction( + conn: sqlite3.Connection, + plans: list[_TerminalDispositionPlan], +) -> None: + """Apply prepared plans on an already-open task transaction.""" + _apply_terminal_plans(conn, plans, table="workspace_registry.workspaces") + + +def apply_terminal_disposition( + *, + task_id: str, + board_id: str, + disposition: object, + pr_numbers: object = None, +) -> list[WorkspaceRecord]: + """Compatibility wrapper for registry-only callers.""" + plans = prepare_terminal_disposition( + task_id=task_id, + board_id=board_id, + disposition=disposition, + pr_numbers=pr_numbers, + ) + with contextlib.closing(connect_registry()) as conn: + conn.execute("BEGIN IMMEDIATE") + try: + _apply_terminal_plans(conn, plans, table="workspaces") + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise + return [ + record for record in list_workspace_records(task_id=task_id) + if record.board_id == board_id and record.status != "creation_failed" + ] + + +def _git_worktrees(repo_path: Path | str) -> list[dict[str, object]]: + raw = _git(repo_path, "worktree", "list", "--porcelain") + if raw is None: + return [] + entries: list[dict[str, object]] = [] + current: dict[str, object] = {} + for line in raw.splitlines() + [""]: + if not line: + if current: + current["is_primary"] = not entries + current["workspace_path"] = _canonical(str(current["workspace_path"])) + entries.append(current) + current = {} + continue + key, _, value = line.partition(" ") + if key == "worktree": + current["workspace_path"] = value + elif key == "HEAD": + current["head_sha"] = value + elif key == "branch": + current["branch"] = value.removeprefix("refs/heads/") + elif key in {"detached", "bare", "locked", "prunable"}: + current[key] = value or True + return entries + + +def _is_within(path: str, roots: Iterable[str]) -> bool: + candidate = Path(path) + for root in roots: + try: + candidate.relative_to(Path(root)) + return True + except ValueError: + continue + return False + + +def _task_inventory() -> tuple[ + dict[tuple[str, str], object], set[str], set[tuple[str, str]] +]: + """Read task lifecycle and dependency evidence from every board.""" + from hermes_cli import kanban_db + + tasks: dict[tuple[str, str], object] = {} + worktree_paths: set[str] = set() + retained_dependencies: set[tuple[str, str]] = set() + for board in kanban_db.list_boards(): + slug = str(board["slug"]) + try: + with kanban_db.connect_closing(board=slug) as conn: + for task in kanban_db.list_tasks( + conn, include_archived=True, limit=100000 + ): + tasks[(slug, task.id)] = task + if task.workspace_kind == "worktree" and task.workspace_path: + worktree_paths.add(_canonical(task.workspace_path)) + for row in conn.execute( + "SELECT DISTINCT l.parent_id FROM task_links l " + "JOIN tasks child ON child.id = l.child_id " + "WHERE child.status NOT IN ('done', 'cancelled', 'archived')" + ).fetchall(): + retained_dependencies.add((slug, str(row["parent_id"]))) + except Exception: + # A board that cannot be opened is itself an operational problem, + # but inventory must stay useful for all other repositories. The + # CLI surfaces board-open failures separately. + continue + return tasks, worktree_paths, retained_dependencies + + +def load_exception_config(path: Path | str) -> list[dict[str, object]]: + source = Path(path).expanduser() + data = yaml.safe_load(source.read_text(encoding="utf-8")) or {} + if not isinstance(data, dict) or data.get("version") != 1: + raise RuntimeError("worktree exception config must be a version: 1 mapping") + raw = data.get("exceptions", []) + if not isinstance(raw, list): + raise RuntimeError("worktree exception config 'exceptions' must be a list") + policies: list[dict[str, object]] = [] + seen_policy_ids: set[str] = set() + seen_workspace_paths: set[str] = set() + for index, item in enumerate(raw): + if not isinstance(item, dict): + raise RuntimeError(f"exception #{index + 1} must be a mapping") + policy_id = str(item.get("policy_id") or "").strip() + repo_path = os.path.expandvars(str(item.get("repo_path") or "").strip()) + workspace_path = os.path.expandvars( + str(item.get("workspace_path") or "").strip() + ) + if not policy_id or not repo_path or not workspace_path: + raise RuntimeError( + f"exception #{index + 1} requires policy_id, repo_path, and workspace_path" + ) + policy = dict(item) + policy["policy_id"] = policy_id + policy["repo_path"] = _canonical(repo_path) + policy["workspace_path"] = _canonical(workspace_path) + if policy_id in seen_policy_ids: + raise RuntimeError(f"duplicate worktree exception policy_id: {policy_id}") + canonical_workspace = str(policy["workspace_path"]) + if canonical_workspace in seen_workspace_paths: + raise RuntimeError( + f"duplicate worktree exception workspace_path: {canonical_workspace}" + ) + seen_policy_ids.add(policy_id) + seen_workspace_paths.add(canonical_workspace) + policies.append(policy) + return sorted(policies, key=lambda item: str(item["policy_id"])) + + +def _exception_expired(policy: dict[str, object], now: Optional[int] = None) -> bool: + expires = _exception_expires_at(policy) + return expires is not None and expires <= ( + int(time.time()) if now is None else now + ) + + +def _exception_expires_at(policy: dict[str, object]) -> Optional[int]: + raw = policy.get("expires_at") + if raw in (None, "", 0, "0"): + return None + try: + expires = int(str(raw)) + except (TypeError, ValueError): + try: + from datetime import datetime + + expires = int(datetime.fromisoformat(str(raw).replace("Z", "+00:00")).timestamp()) + except (TypeError, ValueError) as exc: + raise RuntimeError( + f"invalid expires_at for exception {policy.get('policy_id')}: {raw!r}" + ) from exc + return expires + + +def import_protected_exceptions(config_path: Path | str) -> dict[str, object]: + """Safely register existing physical worktrees from an explicit config.""" + imported: list[dict[str, str]] = [] + skipped: list[dict[str, str]] = [] + for policy in load_exception_config(config_path): + policy_id = str(policy["policy_id"]) + repo = _canonical_repo(str(policy["repo_path"])) + workspace = str(policy["workspace_path"]) + if _exception_expired(policy): + skipped.append({ + "policy_id": policy_id, + "workspace_path": workspace, + "reason": "expired_policy", + }) + continue + physical = { + str(entry["workspace_path"]) for entry in _git_worktrees(repo) + } + if workspace not in physical: + skipped.append({ + "policy_id": policy_id, + "workspace_path": workspace, + "reason": "not_a_git_worktree", + }) + continue + now = int(time.time()) + expires_at = _exception_expires_at(policy) + head = _git(workspace, "rev-parse", "HEAD") + branch = _git(workspace, "branch", "--show-current") or None + manifest = dirty_manifest_hash(workspace) + size = estimated_bytes(workspace) + with contextlib.closing(connect_registry()) as conn: + conn.execute("BEGIN IMMEDIATE") + try: + existing = conn.execute( + "SELECT * FROM workspaces WHERE workspace_path = ?", + (workspace,), + ).fetchone() + if existing and existing["task_id"]: + conn.execute("ROLLBACK") + skipped.append({ + "policy_id": policy_id, + "workspace_path": workspace, + "reason": "task_owned_registry_record", + }) + continue + if existing: + conn.execute( + "UPDATE workspaces SET repo_id=?, repo_path=?, purpose=?, branch=?, " + "head_sha=?, status='protected', cleanup_policy='never_automatic', " + "retention_condition='operational_exception', disposition=?, " + "exception_policy_id=?, exception_expires_at=?, " + "dirty_manifest_hash=?, estimated_bytes=?, last_used_at=?, " + "last_verified_at=? WHERE workspace_id=?", + ( + _repo_id(repo), repo, + str(policy.get("purpose") or "operational_exception"), + branch, head, f"operational_exception:{policy_id}", + policy_id, expires_at, manifest, size, now, now, + existing["workspace_id"], + ), + ) + else: + conn.execute( + "INSERT INTO workspaces (workspace_id, repo_id, repo_path, " + "workspace_path, owner_profile, purpose, branch, head_sha, status, " + "cleanup_policy, retention_condition, disposition, " + "exception_policy_id, exception_expires_at, dirty_manifest_hash, estimated_bytes, " + "created_at, last_used_at, last_verified_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'protected', 'never_automatic', " + "'operational_exception', ?, ?, ?, ?, ?, ?, ?, ?)\n", + ( + "w_" + uuid.uuid4().hex[:16], _repo_id(repo), repo, workspace, + str(policy.get("owner_profile") or "operator"), + str(policy.get("purpose") or "operational_exception"), + branch, head, f"operational_exception:{policy_id}", policy_id, + expires_at, manifest, size, now, now, now, + ), + ) + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise + imported.append({"policy_id": policy_id, "workspace_path": workspace}) + return {"imported": imported, "skipped": skipped} + + +def reconcile_inventory( + *, + repo_paths: Iterable[Path | str], + approved_roots: Iterable[Path | str] = (), + exception_config: Optional[Path | str] = None, +) -> dict[str, object]: + """Return deterministic Git↔registry↔task reconciliation findings. + + The pass is report-only with respect to Git: it updates verification facts + on known registry rows but never adopts, removes, prunes, or repairs a + worktree. Unknown paths remain findings until an operator explicitly + registers or retires them. + """ + repos = sorted({_canonical_repo(path) for path in repo_paths}) + approved = sorted({_canonical(path) for path in approved_roots}) + repo_ids = {_repo_id(repo) for repo in repos} + exception_policies = ( + { + str(item["policy_id"]): item + for item in load_exception_config(exception_config) + if str(item["repo_path"]) in repos + } + if exception_config else {} + ) + tasks, _task_paths, _retained_dependencies = _task_inventory() + records = [ + record for record in list_workspace_records() + if record.repo_id in repo_ids + ] + records_by_path = { + record.workspace_path: record + for record in records + if record.status in ACTIVE_REGISTRY_STATUSES + } + physical_paths: set[str] = set() + per_repo: list[dict[str, object]] = [] + git_entries: list[dict[str, object]] = [] + unowned: list[str] = [] + outside: list[str] = [] + + for repo in repos: + entries = _git_worktrees(repo) + rid = _repo_id(repo) + repo_bytes = 0 + for entry in entries: + path = str(entry["workspace_path"]) + physical_paths.add(path) + size = estimated_bytes(path) + repo_bytes += size + entry["repo_id"] = rid + entry["repo_path"] = repo + entry["estimated_bytes"] = size + git_entries.append(entry) + if not bool(entry.get("is_primary")): + if path not in records_by_path: + unowned.append(path) + if approved and not _is_within(path, approved): + outside.append(path) + per_repo.append( + { + "repo_id": rid, + "repo_path": repo, + "worktree_count": len(entries), + "estimated_bytes": repo_bytes, + } + ) + + terminal_without: list[str] = [] + dirty_terminal: list[str] = [] + now = int(time.time()) + terminal_statuses = {"done", "cancelled", "archived"} + with contextlib.closing(connect_registry()) as conn: + conn.execute("BEGIN IMMEDIATE") + try: + for record in records: + if record.workspace_path not in physical_paths: + continue + task = tasks.get((record.board_id or "default", record.task_id or "")) + task_status = getattr(task, "status", None) + status_raw = _git( + record.workspace_path, + "status", + "--porcelain=v1", + "-z", + "--untracked-files=all", + ) + is_dirty = bool(status_raw) + head = _git(record.workspace_path, "rev-parse", "HEAD") + branch = _git(record.workspace_path, "branch", "--show-current") or None + size = estimated_bytes(record.workspace_path) + lifecycle_status = record.status + is_protected = ( + record.status == "protected" + or (record.disposition or "").startswith( + "operational_exception:" + ) + or bool(record.exception_policy_id) + ) + if is_protected: + lifecycle_status = "protected" + elif task_status in terminal_statuses: + lifecycle_status = "terminal_dirty" if is_dirty else "terminal_clean" + if not record.disposition: + terminal_without.append(record.workspace_path) + if is_dirty: + dirty_terminal.append(record.workspace_path) + elif record.status not in {"protected", "blocked", "superseded"}: + lifecycle_status = "active" + manifest = ( + hashlib.sha256((status_raw or "").encode("utf-8")).hexdigest() + if status_raw is not None else None + ) + # A terminal disposition is the cleanup drift baseline. + # Inventory may observe later changes but must never bless them + # by overwriting the recorded branch/head/manifest snapshot. + preserve_terminal_snapshot = bool(record.disposition) and ( + task_status in terminal_statuses or is_protected + ) + conn.execute( + "UPDATE workspaces SET branch = COALESCE(?, branch), head_sha = ?, " + "status = ?, dirty_manifest_hash = ?, estimated_bytes = ?, " + "last_verified_at = ? WHERE workspace_id = ?", + ( + record.branch if preserve_terminal_snapshot else branch, + record.head_sha if preserve_terminal_snapshot else head, + lifecycle_status, + ( + record.dirty_manifest_hash + if preserve_terminal_snapshot + else manifest + ), + size, + now, + record.workspace_id, + ), + ) + conn.execute("COMMIT") + except Exception: + conn.execute("ROLLBACK") + raise + + task_groups: dict[tuple[str, str, str], list[WorkspaceRecord]] = {} + for record in records: + if record.workspace_path not in physical_paths: + continue + if not record.task_id or not record.board_id: + continue + task_groups.setdefault( + (record.board_id, record.task_id, record.repo_id), [] + ).append(record) + duplicate_task_repo = [] + for (board_id, task_id, repo_id), group in sorted(task_groups.items()): + if len(group) < 2: + continue + duplicate_task_repo.append( + { + "board_id": board_id, + "task_id": task_id, + "repo_id": repo_id, + "workspace_ids": sorted(item.workspace_id for item in group), + "paths": sorted(item.workspace_path for item in group), + } + ) + + head_groups: dict[tuple[str, str], list[str]] = {} + for entry in git_entries: + head = str(entry.get("head_sha") or "") + if head: + head_groups.setdefault((str(entry["repo_id"]), head), []).append( + str(entry["workspace_path"]) + ) + duplicate_heads = [ + {"repo_id": repo_id, "head_sha": head, "paths": sorted(paths)} + for (repo_id, head), paths in sorted(head_groups.items()) + if len(paths) > 1 + ] + + protected_findings: list[dict[str, str]] = [] + for policy_id, policy in sorted(exception_policies.items()): + workspace_path = str(policy["workspace_path"]) + if _exception_expired(policy, now=now): + reason = "expired_policy" + elif workspace_path not in physical_paths: + reason = "missing_physical_worktree" + else: + record = records_by_path.get(workspace_path) + if record is None: + reason = "missing_registry_record" + elif ( + record.status != "protected" + or record.exception_policy_id != policy_id + or record.disposition != f"operational_exception:{policy_id}" + ): + reason = "registry_policy_mismatch" + else: + continue + protected_findings.append( + { + "policy_id": policy_id, + "workspace_path": workspace_path, + "reason": reason, + } + ) + + for record in records: + policy_id = record.exception_policy_id + is_protected = ( + record.status == "protected" + or (record.disposition or "").startswith("operational_exception:") + or bool(policy_id) + ) + if not is_protected or not policy_id: + continue + if ( + record.exception_expires_at is not None + and record.exception_expires_at <= now + ): + reason = "expired_policy" + elif exception_config is not None and policy_id not in exception_policies: + reason = "missing_policy_config" + else: + continue + finding = { + "policy_id": policy_id, + "workspace_path": record.workspace_path, + "reason": reason, + } + if finding not in protected_findings: + protected_findings.append(finding) + protected_findings.sort( + key=lambda item: ( + item["policy_id"], item["workspace_path"], item["reason"] + ) + ) + + return { + "totals": { + "repositories": len(per_repo), + "git_worktrees": len(git_entries), + "registry_records": len(records), + "estimated_bytes": sum( + int(str(repo["estimated_bytes"])) for repo in per_repo + ), + }, + "per_repo": per_repo, + "unowned_paths": sorted(set(unowned)), + "duplicate_task_repo": duplicate_task_repo, + "duplicate_heads": duplicate_heads, + "terminal_without_disposition": sorted(set(terminal_without)), + "dirty_terminal": sorted(set(dirty_terminal)), + "outside_approved_root": sorted(set(outside)), + "missing_expired_protected_exceptions": protected_findings, + } + + +def plan_janitor( + *, + repo_paths: Iterable[Path | str], + approved_roots: Iterable[Path | str], +) -> dict[str, object]: + """Build an auditable deletion plan without removing anything. + + There is intentionally no execute mode. Enabling deletion requires a + separate owner-approved change after this dry-run has proven conservative + against real inventories. + """ + repos = sorted({_canonical_repo(path) for path in repo_paths}) + repo_ids = {_repo_id(path) for path in repos} + approved = sorted({_canonical(path) for path in approved_roots}) + records = [ + record for record in list_workspace_records() + if record.repo_id in repo_ids and record.status != "creation_failed" + ] + tasks, _task_paths, retained_dependencies = _task_inventory() + terminal_task_statuses = {"done", "cancelled", "archived"} + physical_entries = { + str(entry["workspace_path"]): entry + for repo in repos + for entry in _git_worktrees(repo) + } + physical = set(physical_entries) + owner_counts: dict[tuple[Optional[str], Optional[str], str], int] = {} + for record in records: + key = (record.board_id, record.task_id, record.repo_id) + owner_counts[key] = owner_counts.get(key, 0) + 1 + + candidates: list[dict[str, object]] = [] + excluded: list[dict[str, object]] = [] + receipts: list[dict[str, object]] = [] + for record in sorted(records, key=lambda item: (item.workspace_path, item.workspace_id)): + reasons: list[str] = [] + task = tasks.get((record.board_id or "default", record.task_id or "")) + if getattr(task, "status", None) not in terminal_task_statuses: + reasons.append("task_not_terminal") + if record.workspace_path not in physical: + reasons.append("missing_from_git_inventory") + else: + physical_entry = physical_entries[record.workspace_path] + if bool(physical_entry.get("is_primary")): + reasons.append("primary_worktree") + if physical_entry.get("locked"): + reasons.append("locked_worktree") + if approved and not _is_within(record.workspace_path, approved): + reasons.append("outside_approved_root") + if ( + record.status == "protected" + or (record.disposition or "").startswith("operational_exception:") + or record.exception_policy_id + ): + reasons.append("protected_exception") + if record.pr_numbers: + # No network/API lookup occurs in janitor. Any associated PR is + # therefore open-or-unverifiable and must be preserved. + reasons.append("open_or_unverifiable_pr") + if record.disposition != "retained_until:janitor": + reasons.append("retention_condition_not_met") + if ( + record.cleanup_policy != "on_task_terminal" + or (record.board_id or "default", record.task_id or "") + in retained_dependencies + ): + reasons.append("dependency_retained") + if owner_counts.get((record.board_id, record.task_id, record.repo_id), 0) > 1: + reasons.append("ambiguous_duplicate_owner") + + current_status = _git( + record.workspace_path, + "status", + "--porcelain=v1", + "-z", + "--untracked-files=all", + ) + current_hash = ( + hashlib.sha256((current_status or "").encode("utf-8")).hexdigest() + if current_status is not None else None + ) + current_head = _git(record.workspace_path, "rev-parse", "HEAD") + if ( + current_status is None + or bool(current_status) + or current_hash != record.dirty_manifest_hash + or current_head != record.head_sha + ): + reasons.append("dirty_or_changed") + + if reasons: + excluded.append( + { + "workspace_id": record.workspace_id, + "workspace_path": record.workspace_path, + "reasons": sorted(set(reasons)), + } + ) + continue + candidates.append( + { + "workspace_id": record.workspace_id, + "workspace_path": record.workspace_path, + "repo_path": record.repo_path, + "task_id": record.task_id, + "board_id": record.board_id, + "estimated_bytes": record.estimated_bytes, + } + ) + receipts.append( + { + "workspace_id": record.workspace_id, + "workspace_path": record.workspace_path, + "repo_path": record.repo_path, + "head_sha": current_head, + "dirty_manifest_hash": current_hash, + "planned_action": "retire_worktree", + "execution": "disabled", + "requires_live_reverification": True, + } + ) + + return { + "dry_run": True, + "candidates": candidates, + "excluded": excluded, + "receipts": receipts, + "removed": [], + } diff --git a/tests/hermes_cli/test_kanban_worktree_isolation.py b/tests/hermes_cli/test_kanban_worktree_isolation.py index 02538ae0f3c87..d84fc8d453037 100644 --- a/tests/hermes_cli/test_kanban_worktree_isolation.py +++ b/tests/hermes_cli/test_kanban_worktree_isolation.py @@ -126,5 +126,35 @@ def test_resolve_worktree_falls_back_when_path_occupied(kanban_home, tmp_path): assert head == "wt/sibling" +def test_resolve_canonical_path_on_foreign_branch_fails_closed( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + with kb.connect() as conn: + tid = kb.create_task( + conn, + title="resume exact task", + workspace_kind="worktree", + workspace_path=str(repo), + ) + own = _add_worktree(repo, repo / ".worktrees" / tid, "wt/foreign") + conn.execute( + "UPDATE tasks SET workspace_path = ? WHERE id = ?", + (str(own), tid), + ) + conn.commit() + task = kb.get_task(conn, tid) + + assert task is not None + with pytest.raises(ValueError, match="branch mismatch"): + kb._resolve_worktree_workspace(task) + assert subprocess.run( + ["git", "-C", str(own), "branch", "--show-current"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() == "wt/foreign" + + diff --git a/tests/hermes_cli/test_kanban_worktree_lifecycle.py b/tests/hermes_cli/test_kanban_worktree_lifecycle.py new file mode 100644 index 0000000000000..6ef01b9ad88bc --- /dev/null +++ b/tests/hermes_cli/test_kanban_worktree_lifecycle.py @@ -0,0 +1,1387 @@ +"""E2E coverage for the canonical Kanban worktree lifecycle registry.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb +from hermes_cli import kanban_workspaces as kw + + +@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 + + +def _git(cwd: Path, *args: str) -> str: + result = subprocess.run( + [ + "git", + "-C", + str(cwd), + "-c", + "user.name=Kanban Test", + "-c", + "user.email=kanban@example.com", + "-c", + "commit.gpgsign=false", + *args, + ], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def _make_repo(tmp_path: Path) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init", "-b", "main") + (repo / "README.md").write_text("base\n", encoding="utf-8") + _git(repo, "add", "README.md") + _git(repo, "commit", "-m", "init") + return repo + + +def test_resolved_worktree_is_registered_before_creation_and_reused( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "owned" + + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="implement", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/owned", + ) + task = kb.get_task(conn, task_id) + first = kb.resolve_workspace(task) + kb.set_workspace_path(conn, task_id, first) + second = kb.resolve_workspace(kb.get_task(conn, task_id)) + + assert first == second == target + records = kw.list_workspace_records(task_id=task_id) + assert len(records) == 1 + record = records[0] + assert record.workspace_id.startswith("w_") + assert record.repo_path == str(repo.resolve()) + assert record.workspace_path == str(target.resolve()) + assert record.task_id == task_id + assert record.board_id == "default" + assert record.purpose == "implementation" + assert record.branch == "feat/owned" + assert record.head_sha == _git(target, "rev-parse", "HEAD") + assert record.status == "active" + assert record.cleanup_policy == "on_task_terminal" + assert record.retention_condition == "task_terminal" + assert record.estimated_bytes >= 0 + assert record.last_verified_at is not None + + +def test_preexisting_matching_worktree_is_adopted_only_for_its_task( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "preexisting" + _git(repo, "worktree", "add", "-b", "feat/preexisting", str(target), "HEAD") + + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="claim preexisting", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/preexisting", + ) + task = kb.get_task(conn, task_id) + assert task is not None + assert kb.resolve_workspace(task) == target.resolve() + + records = kw.list_workspace_records(task_id=task_id) + assert len(records) == 1 + assert records[0].workspace_path == str(target.resolve()) + assert records[0].head_sha == _git(target, "rev-parse", "HEAD") + assert records[0].status == "active" + + +def test_existing_target_on_wrong_branch_fails_closed_without_mutating_git( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="wrong branch collision", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(repo), + branch_name="feat/wanted", + ) + target = repo / ".worktrees" / task_id + _git(repo, "worktree", "add", "-b", "feat/occupied", str(target), "HEAD") + before = _git(repo, "worktree", "list", "--porcelain") + task = kb.get_task(conn, task_id) + assert task is not None + with pytest.raises(RuntimeError, match="branch mismatch"): + kb.resolve_workspace(task) + after = _git(repo, "worktree", "list", "--porcelain") + + assert after == before + records = kw.list_workspace_records(task_id=task_id) + assert len(records) == 1 + assert records[0].status == "creation_failed" + assert _git(target, "branch", "--show-current") == "feat/occupied" + report = kw.reconcile_inventory( + repo_paths=[repo], approved_roots=[repo / ".worktrees"] + ) + assert report["unowned_paths"] == [str(target.resolve())] + + +def test_drifted_existing_registration_stays_owned_when_branch_check_fails( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "drifted" + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="detect branch drift", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/expected", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + _git(target, "switch", "-c", "feat/drifted") + + current_task = kb.get_task(conn, task_id) + assert current_task is not None + with pytest.raises(RuntimeError, match="branch mismatch"): + kb.resolve_workspace(current_task) + + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.status == "active" + assert record.workspace_path == str(target.resolve()) + + +def test_duplicate_task_repo_requires_recorded_replacement_reason( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + first_path = repo / ".worktrees" / "first" + replacement_path = repo / ".worktrees" / "replacement" + + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="replace worktree", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(first_path), + branch_name="feat/first", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + + conn.execute( + "UPDATE tasks SET workspace_path=?, branch_name=? WHERE id=?", + (str(replacement_path), "feat/replacement", task_id), + ) + conn.commit() + with pytest.raises(RuntimeError, match="replacement reason"): + kb.resolve_workspace(kb.get_task(conn, task_id)) + + conn.execute( + "UPDATE tasks SET workspace_replacement_reason=? WHERE id=?", + ("first checkout was created from the wrong anchor", task_id), + ) + conn.commit() + replaced = kb.resolve_workspace(kb.get_task(conn, task_id)) + + assert replaced == replacement_path + records = kw.list_workspace_records(task_id=task_id) + assert len(records) == 2 + assert records[1].replacement_reason == "first checkout was created from the wrong anchor" + + +def test_complete_and_archive_require_explicit_workspace_disposition( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + dirty_path = repo / ".worktrees" / "dirty" + clean_path = repo / ".worktrees" / "clean" + + with kb.connect() as conn: + dirty_task_id = kb.create_task( + conn, + title="dirty terminal", + workspace_kind="worktree", + workspace_path=str(dirty_path), + branch_name="feat/dirty", + ) + dirty_task = kb.get_task(conn, dirty_task_id) + assert dirty_task is not None + kb.resolve_workspace(dirty_task) + (dirty_path / "evidence.txt").write_text("keep\n", encoding="utf-8") + + with pytest.raises(RuntimeError, match="workspace_disposition"): + kb.complete_task(conn, dirty_task_id, metadata={"tests_run": 1}) + assert kb.get_task(conn, dirty_task_id).status == "ready" + + with pytest.raises(RuntimeError, match="dirty worktree.*retired"): + kb.complete_task( + conn, + dirty_task_id, + metadata={"workspace_disposition": "retired"}, + ) + + with pytest.raises(RuntimeError, match="requires preserved_dirty"): + kb.complete_task( + conn, + dirty_task_id, + metadata={ + "workspace_disposition": "retained_until:review-required" + }, + ) + + assert kb.complete_task( + conn, + dirty_task_id, + metadata={ + "workspace_disposition": ( + "preserved_dirty:artifact=evidence.txt;owner=reviewer" + ) + }, + ) + dirty_record = kw.list_workspace_records(task_id=dirty_task_id)[0] + assert dirty_record.disposition == ( + "preserved_dirty:artifact=evidence.txt;owner=reviewer" + ) + assert dirty_record.status == "terminal_dirty" + assert kb.archive_task(conn, dirty_task_id) + assert kb.get_task(conn, dirty_task_id).status == "archived" + + clean_task_id = kb.create_task( + conn, + title="archive clean terminal", + workspace_kind="worktree", + workspace_path=str(clean_path), + branch_name="feat/clean", + ) + clean_task = kb.get_task(conn, clean_task_id) + assert clean_task is not None + kb.resolve_workspace(clean_task) + with pytest.raises(RuntimeError, match="workspace_disposition"): + kb.archive_task(conn, clean_task_id) + assert kb.archive_task( + conn, + clean_task_id, + workspace_disposition="retained_until:janitor", + ) + clean_record = kw.list_workspace_records(task_id=clean_task_id)[0] + assert clean_record.retention_condition == "janitor" + assert clean_record.status == "retirement_queued" + + +def test_terminal_disposition_uses_the_connection_board_not_ambient_board( + kanban_home, tmp_path, monkeypatch +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "beta-task" + kb.create_board("beta") + monkeypatch.setenv("HERMES_KANBAN_BOARD", "default") + + with kb.connect(board="beta") as conn: + task_id = kb.create_task( + conn, + title="beta worktree", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/beta-task", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task, board="beta") + assert kb.complete_task( + conn, + task_id, + metadata={"workspace_disposition": "retained_until:beta-review"}, + ) + + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.board_id == "beta" + assert record.disposition == "retained_until:beta-review" + assert record.status == "terminal_clean" + + +def test_archive_cli_records_direct_disposition_and_reuses_completed_one( + kanban_home, tmp_path +): + from hermes_cli import kanban as kc + + repo = _make_repo(tmp_path) + direct_path = repo / ".worktrees" / "direct-archive" + completed_path = repo / ".worktrees" / "completed-archive" + with kb.connect() as conn: + direct_id = kb.create_task( + conn, + title="direct archive", + workspace_kind="worktree", + workspace_path=str(direct_path), + branch_name="feat/direct-archive", + ) + kb.resolve_workspace(kb.get_task(conn, direct_id)) + completed_id = kb.create_task( + conn, + title="completed archive", + workspace_kind="worktree", + workspace_path=str(completed_path), + branch_name="feat/completed-archive", + ) + kb.resolve_workspace(kb.get_task(conn, completed_id)) + assert kb.complete_task( + conn, + completed_id, + metadata={"workspace_disposition": "retained_until:review-closeout"}, + ) + + assert kc.run_slash( + f"archive --workspace-disposition retained_until:review-closeout {direct_id}" + ) == "Archived " + direct_id + assert kc.run_slash(f"archive {completed_id}") == "Archived " + completed_id + with kb.connect() as conn: + assert kb.get_task(conn, direct_id).status == "archived" + assert kb.get_task(conn, completed_id).status == "archived" + + +def test_inventory_reconciles_git_counts_and_reports_deterministic_findings( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + approved_root = repo / ".worktrees" + owned = approved_root / "owned" + duplicate = approved_root / "duplicate" + outside = tmp_path / "outside-worktree" + + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="terminal implementation", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(owned), + branch_name="feat/owned", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + + _git(repo, "worktree", "add", "-b", "feat/duplicate", str(duplicate), "HEAD") + replacement = kw.reserve_workspace( + repo_path=repo, + workspace_path=duplicate, + task_id=task_id, + board_id="default", + owner_profile="developer", + branch="feat/duplicate", + replacement_reason="clean_replacement_after_dirty_collision", + ) + kw.verify_workspace(replacement.workspace_id, duplicate) + + _git(repo, "worktree", "add", "--detach", str(outside), "HEAD") + (owned / "dirty.txt").write_text("preserve me\n", encoding="utf-8") + conn.execute( + "UPDATE tasks SET status='done', completed_at=1 WHERE id = ?", + (task_id,), + ) + conn.commit() + + first = kw.reconcile_inventory( + repo_paths=[repo], approved_roots=[approved_root] + ) + second = kw.reconcile_inventory( + repo_paths=[repo], approved_roots=[approved_root] + ) + + assert first == second + assert first["totals"]["git_worktrees"] == 4 + assert first["per_repo"] == [ + { + "repo_id": first["per_repo"][0]["repo_id"], + "repo_path": str(repo.resolve()), + "worktree_count": 4, + "estimated_bytes": first["per_repo"][0]["estimated_bytes"], + } + ] + assert first["unowned_paths"] == [str(outside.resolve())] + assert first["duplicate_task_repo"] == [ + { + "board_id": "default", + "task_id": task_id, + "repo_id": first["per_repo"][0]["repo_id"], + "workspace_ids": sorted( + record.workspace_id + for record in kw.list_workspace_records(task_id=task_id) + ), + "paths": sorted([str(owned.resolve()), str(duplicate.resolve())]), + } + ] + assert len(first["duplicate_heads"]) == 1 + assert first["duplicate_heads"][0]["paths"] == sorted( + [str(repo.resolve()), str(owned.resolve()), str(duplicate.resolve()), str(outside.resolve())] + ) + assert first["terminal_without_disposition"] == sorted( + [str(owned.resolve()), str(duplicate.resolve())] + ) + assert first["dirty_terminal"] == [str(owned.resolve())] + assert first["outside_approved_root"] == [str(outside.resolve())] + + from hermes_cli import kanban as kc + + cli_payload = json.loads( + kc.run_slash( + f"worktrees inventory --repo {repo} " + f"--approved-root {approved_root} --json" + ) + ) + assert cli_payload == first + + terminal_report = kc.run_slash( + f"worktrees inventory --repo {repo} --approved-root {approved_root}" + ) + assert f"{repo.resolve()}: 4 worktree(s)" in terminal_report + assert "unowned_paths (1)" in terminal_report + assert str(outside.resolve()) in terminal_report + assert "duplicate_task_repo (1)" in terminal_report + assert "terminal_without_disposition (2)" in terminal_report + assert "dirty_terminal (1)" in terminal_report + assert "outside_approved_root (1)" in terminal_report + + +def test_janitor_is_dry_run_only_and_excludes_unsafe_worktrees( + kanban_home, tmp_path, monkeypatch +): + repo = _make_repo(tmp_path) + approved = repo / ".worktrees" + + def make_terminal(name: str, disposition: str) -> tuple[str, Path]: + path = approved / name + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title=name, + workspace_kind="worktree", + workspace_path=str(path), + branch_name=f"feat/{name}", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + assert kb.archive_task( + conn, task_id, workspace_disposition=disposition + ) + return task_id, path + + candidate_id, candidate = make_terminal("candidate", "retained_until:janitor") + pr_id, pr_path = make_terminal("open-pr", "retained_until:janitor") + protected_path = approved / "deepsec" + _git(repo, "worktree", "add", "-b", "ops/deepsec", str(protected_path), "HEAD") + exception_config = tmp_path / "exceptions.yaml" + exception_config.write_text( + "version: 1\nexceptions:\n" + " - policy_id: deepsec-operational\n" + f" repo_path: {repo}\n" + f" workspace_path: {protected_path}\n", + encoding="utf-8", + ) + imported = kw.import_protected_exceptions(exception_config) + assert imported["imported"] == [ + { + "policy_id": "deepsec-operational", + "workspace_path": str(protected_path.resolve()), + } + ] + dependency_id, dependency_path = make_terminal( + "dependency", "retained_until:successor_terminal" + ) + dirty_id, dirty_path = make_terminal("dirty", "retained_until:janitor") + (dirty_path / "late-change.txt").write_text("changed after terminal\n", encoding="utf-8") + active_path = approved / "active" + with kb.connect() as conn: + active_id = kb.create_task( + conn, + title="active", + workspace_kind="worktree", + workspace_path=str(active_path), + branch_name="feat/active", + ) + kb.resolve_workspace(kb.get_task(conn, active_id)) + + with kw.connect_registry() as registry: + registry.execute( + "UPDATE workspaces SET pr_numbers='71978' WHERE task_id=?", (pr_id,) + ) + registry.execute( + "UPDATE workspaces SET disposition='retained_until:janitor', " + "status='retirement_queued' WHERE task_id=?", + (active_id,), + ) + + real_subprocess_run = kw.subprocess.run + removal_attempts: list[list[str]] = [] + + def reject_removal(command, *args, **kwargs): + rendered = [str(part) for part in command] + if "worktree" in rendered and "remove" in rendered: + removal_attempts.append(rendered) + raise AssertionError("dry-run janitor attempted worktree removal") + return real_subprocess_run(command, *args, **kwargs) + + monkeypatch.setattr(kw.subprocess, "run", reject_removal) + plan = kw.plan_janitor(repo_paths=[repo], approved_roots=[approved]) + assert plan["dry_run"] is True + assert plan["removed"] == [] + assert [item["workspace_path"] for item in plan["candidates"]] == [ + str(candidate.resolve()) + ] + assert plan["receipts"] == [ + { + "workspace_id": kw.list_workspace_records(task_id=candidate_id)[0].workspace_id, + "workspace_path": str(candidate.resolve()), + "repo_path": str(repo.resolve()), + "head_sha": _git(candidate, "rev-parse", "HEAD"), + "dirty_manifest_hash": kw.dirty_manifest_hash(candidate), + "planned_action": "retire_worktree", + "execution": "disabled", + "requires_live_reverification": True, + } + ] + assert removal_attempts == [] + reasons = { + item["workspace_path"]: item["reasons"] for item in plan["excluded"] + } + assert "open_or_unverifiable_pr" in reasons[str(pr_path.resolve())] + assert "protected_exception" in reasons[str(protected_path.resolve())] + assert "retention_condition_not_met" in reasons[str(dependency_path.resolve())] + assert "dirty_or_changed" in reasons[str(dirty_path.resolve())] + assert "task_not_terminal" in reasons[str(active_path.resolve())] + assert dependency_id and dirty_id + + inventory = kw.reconcile_inventory( + repo_paths=[repo], approved_roots=[approved] + ) + assert inventory["missing_expired_protected_exceptions"] == [] + protected_record = next( + record + for record in kw.list_workspace_records(repo_path=repo) + if record.workspace_path == str(protected_path.resolve()) + ) + assert protected_record.status == "protected" + + from hermes_cli import kanban as kc + + cli_plan = json.loads( + kc.run_slash( + f"worktrees janitor --repo {repo} --approved-root {approved} --json" + ) + ) + assert cli_plan == plan + + +def test_janitor_never_candidates_primary_or_locked_worktrees( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + locked = repo / ".worktrees" / "locked" + + with kb.connect() as conn: + primary_task = kb.create_task( + conn, + title="primary safety", + workspace_kind="worktree", + workspace_path=str(repo), + branch_name="main", + ) + primary_reservation = kw.reserve_workspace( + repo_path=repo, + workspace_path=repo, + task_id=primary_task, + board_id="default", + owner_profile="developer", + branch="main", + ) + kw.verify_workspace(primary_reservation.workspace_id, repo) + assert kb.archive_task( + conn, + primary_task, + workspace_disposition="retained_until:janitor", + ) + + locked_task = kb.create_task( + conn, + title="locked safety", + workspace_kind="worktree", + workspace_path=str(locked), + branch_name="feat/locked", + ) + locked_record = kb.get_task(conn, locked_task) + assert locked_record is not None + kb.resolve_workspace(locked_record) + assert kb.archive_task( + conn, + locked_task, + workspace_disposition="retained_until:janitor", + ) + _git(repo, "worktree", "lock", str(locked)) + + plan = kw.plan_janitor(repo_paths=[repo], approved_roots=[tmp_path]) + assert plan["candidates"] == [] + excluded = plan["excluded"] + assert isinstance(excluded, list) + reasons = { + item["workspace_path"]: item["reasons"] for item in excluded + } + assert "primary_worktree" in reasons[str(repo.resolve())] + assert "locked_worktree" in reasons[str(locked.resolve())] + + +def test_terminal_snapshot_advances_once_and_inventory_does_not_bless_drift( + tmp_path, kanban_home +): + repo = _make_repo(tmp_path) + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="snapshot", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(repo), + branch_name="feat/snapshot", + ) + task = kb.get_task(conn, task_id) + assert task is not None + workspace = kb.resolve_workspace(task) + + (workspace / "before-terminal.txt").write_text("before\n", encoding="utf-8") + _git(workspace, "add", "before-terminal.txt") + _git(workspace, "commit", "-m", "before terminal") + terminal_head = _git(workspace, "rev-parse", "HEAD") + + assert kb.archive_task( + conn, + task_id, + workspace_disposition="retained_until:janitor", + ) + + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.head_sha == terminal_head + + (workspace / "after-terminal.txt").write_text("after\n", encoding="utf-8") + _git(workspace, "add", "after-terminal.txt") + _git(workspace, "commit", "-m", "after terminal") + drifted_head = _git(workspace, "rev-parse", "HEAD") + assert drifted_head != terminal_head + + kw.reconcile_inventory( + repo_paths=[repo], + approved_roots=[repo / ".worktrees"], + ) + refreshed = kw.list_workspace_records(task_id=task_id)[0] + assert refreshed.head_sha == terminal_head + + plan = kw.plan_janitor( + repo_paths=[repo], + approved_roots=[repo / ".worktrees"], + ) + excluded = plan["excluded"] + assert isinstance(excluded, list) + reasons = { + str(item["workspace_path"]): item["reasons"] + for item in excluded + if isinstance(item, dict) + } + assert "dirty_or_changed" in reasons[str(workspace.resolve())] + assert plan["candidates"] == [] + + +def test_creation_pressure_thresholds_are_config_backed_and_report_only_by_default(): + from hermes_cli.config_defaults import DEFAULT_CONFIG + + lifecycle = DEFAULT_CONFIG["kanban"]["worktree_lifecycle"] + assert lifecycle["enforcement"] == "report_only" + assert lifecycle["warning_worktree_count_per_repo"] == 20 + assert lifecycle["warning_estimated_bytes_per_repo"] == 100 * 1024 * 1024 * 1024 + assert lifecycle["warning_free_space_floor_bytes"] == 50 * 1024 * 1024 * 1024 + + + +def test_protected_exception_import_is_config_driven_and_reported(tmp_path, kanban_home): + import yaml + + from hermes_cli import kanban as kc + + repo = _make_repo(tmp_path) + deepsec = repo / ".worktrees" / "deepsec-operational" + missing = repo / ".worktrees" / "missing-protection" + expired = repo / ".worktrees" / "expired-protection" + _git(repo, "worktree", "add", "-b", "ops/deepsec", str(deepsec), "HEAD") + _git(repo, "worktree", "add", "-b", "ops/missing", str(missing), "HEAD") + _git(repo, "worktree", "add", "-b", "ops/expired", str(expired), "HEAD") + policy = tmp_path / "exceptions.yaml" + policy.write_text( + yaml.safe_dump( + { + "version": 1, + "exceptions": [ + { + "policy_id": "deepsec-operational", + "repo_path": str(repo), + "workspace_path": str(deepsec), + "purpose": "operational", + "expires_at": None, + }, + { + "policy_id": "expired-operational", + "repo_path": str(repo), + "workspace_path": str(expired), + "purpose": "operational", + "expires_at": 1, + }, + ], + }, + sort_keys=True, + ), + encoding="utf-8", + ) + + result = json.loads( + kc.run_slash(f"worktrees exceptions import --config {policy} --json") + ) + assert result == { + "imported": [ + { + "policy_id": "deepsec-operational", + "workspace_path": str(deepsec.resolve()), + } + ], + "skipped": [ + { + "policy_id": "expired-operational", + "workspace_path": str(expired.resolve()), + "reason": "expired_policy", + } + ], + } + record = next( + item for item in kw.list_workspace_records() + if item.workspace_path == str(deepsec.resolve()) + ) + assert record.task_id is None + assert record.status == "protected" + assert record.cleanup_policy == "never_automatic" + assert record.disposition == "operational_exception:deepsec-operational" + assert record.exception_policy_id == "deepsec-operational" + assert record.exception_expires_at is None + + missing_policy = tmp_path / "missing-exceptions.yaml" + missing_policy.write_text( + yaml.safe_dump( + { + "version": 1, + "exceptions": [ + { + "policy_id": "deepsec-operational", + "repo_path": str(repo), + "workspace_path": str(deepsec), + "purpose": "operational", + "expires_at": None, + }, + { + "policy_id": "missing-operational", + "repo_path": str(repo), + "workspace_path": str(missing), + "purpose": "operational", + "expires_at": None, + }, + { + "policy_id": "expired-operational", + "repo_path": str(repo), + "workspace_path": str(expired), + "purpose": "operational", + "expires_at": 1, + }, + ], + }, + sort_keys=True, + ), + encoding="utf-8", + ) + report = json.loads( + kc.run_slash( + f"worktrees inventory --repo {repo} " + f"--approved-root {repo / '.worktrees'} " + f"--exception-config {missing_policy} --json" + ) + ) + assert report["missing_expired_protected_exceptions"] == [ + { + "policy_id": "expired-operational", + "workspace_path": str(expired.resolve()), + "reason": "expired_policy", + }, + { + "policy_id": "missing-operational", + "workspace_path": str(missing.resolve()), + "reason": "missing_registry_record", + }, + ] + + checked_in = ( + Path(__file__).parents[2] + / "docs" / "examples" / "kanban-worktree-exceptions.yaml" + ) + example = checked_in.read_text(encoding="utf-8") + assert "deepsec-operational" in example + assert "/Users/royce" not in example + + +def test_inventory_reports_missing_and_expired_protected_exceptions( + tmp_path, kanban_home +): + import yaml + + repo = _make_repo(tmp_path) + approved = repo / ".worktrees" + current = approved / "deepsec-current" + expired = approved / "deepsec-expired" + _git(repo, "worktree", "add", "-b", "ops/deepsec-current", str(current), "HEAD") + _git(repo, "worktree", "add", "-b", "ops/deepsec-expired", str(expired), "HEAD") + + policy = tmp_path / "exceptions.yaml" + policy.write_text( + yaml.safe_dump( + { + "version": 1, + "exceptions": [ + { + "policy_id": "deepsec-current", + "repo_path": str(repo), + "workspace_path": str(current), + "purpose": "operational", + }, + { + "policy_id": "deepsec-expired", + "repo_path": str(repo), + "workspace_path": str(expired), + "purpose": "operational", + "expires_at": 1, + }, + ], + }, + sort_keys=True, + ), + encoding="utf-8", + ) + + missing = kw.reconcile_inventory( + repo_paths=[repo], + approved_roots=[approved], + exception_config=policy, + ) + assert missing["missing_expired_protected_exceptions"] == [ + { + "policy_id": "deepsec-current", + "workspace_path": str(current.resolve()), + "reason": "missing_registry_record", + }, + { + "policy_id": "deepsec-expired", + "workspace_path": str(expired.resolve()), + "reason": "expired_policy", + }, + ] + + imported = kw.import_protected_exceptions(policy) + assert imported["imported"] == [ + { + "policy_id": "deepsec-current", + "workspace_path": str(current.resolve()), + } + ] + assert imported["skipped"] == [ + { + "policy_id": "deepsec-expired", + "workspace_path": str(expired.resolve()), + "reason": "expired_policy", + } + ] + reconciled = kw.reconcile_inventory( + repo_paths=[repo], + approved_roots=[approved], + exception_config=policy, + ) + assert reconciled["missing_expired_protected_exceptions"] == [ + { + "policy_id": "deepsec-expired", + "workspace_path": str(expired.resolve()), + "reason": "expired_policy", + } + ] + + empty_policy = tmp_path / "empty-exceptions.yaml" + empty_policy.write_text("version: 1\nexceptions: []\n", encoding="utf-8") + missing_config = kw.reconcile_inventory( + repo_paths=[repo], + approved_roots=[approved], + exception_config=empty_policy, + ) + assert missing_config["missing_expired_protected_exceptions"] == [ + { + "policy_id": "deepsec-current", + "workspace_path": str(current.resolve()), + "reason": "missing_policy_config", + } + ] + + +def test_exception_config_rejects_duplicate_policy_ids_and_workspace_paths( + tmp_path, kanban_home +): + import yaml + + repo = _make_repo(tmp_path) + first = repo / ".worktrees" / "first" + second = repo / ".worktrees" / "second" + policy = tmp_path / "duplicates.yaml" + + policy.write_text( + yaml.safe_dump( + { + "version": 1, + "exceptions": [ + { + "policy_id": "duplicate", + "repo_path": str(repo), + "workspace_path": str(first), + }, + { + "policy_id": "duplicate", + "repo_path": str(repo), + "workspace_path": str(second), + }, + ], + }, + sort_keys=True, + ), + encoding="utf-8", + ) + with pytest.raises(RuntimeError, match="duplicate.*policy_id"): + kw.load_exception_config(policy) + + policy.write_text( + yaml.safe_dump( + { + "version": 1, + "exceptions": [ + { + "policy_id": "first", + "repo_path": str(repo), + "workspace_path": str(first), + }, + { + "policy_id": "second", + "repo_path": str(repo), + "workspace_path": str(first), + }, + ], + }, + sort_keys=True, + ), + encoding="utf-8", + ) + with pytest.raises(RuntimeError, match="duplicate.*workspace_path"): + kw.load_exception_config(policy) + + +def test_board_identity_stays_exact_when_database_path_is_overridden( + tmp_path, kanban_home, monkeypatch +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "beta-overridden" + kb.create_board("beta") + beta_db = kb.kanban_db_path("beta").resolve() + monkeypatch.setenv("HERMES_KANBAN_DB", str(beta_db)) + monkeypatch.setenv("HERMES_KANBAN_BOARD", "default") + + with kb.connect(board="beta") as conn: + task_id = kb.create_task( + conn, + title="beta override", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/beta-override", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task, board="beta") + with pytest.raises(RuntimeError, match="workspace_disposition"): + kb.complete_task(conn, task_id) + assert kb.get_task(conn, task_id).status == "ready" + + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.board_id == "beta" + assert record.disposition is None + + +def test_stale_completion_cannot_commit_workspace_disposition( + tmp_path, kanban_home +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "stale-run" + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="stale completion", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/stale-run", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + claimed = kb.claim_task(conn, task_id, claimer="test-worker") + assert claimed is not None and claimed.current_run_id is not None + assert not kb.complete_task( + conn, + task_id, + expected_run_id=claimed.current_run_id + 1, + metadata={"workspace_disposition": "retained_until:review"}, + ) + assert kb.get_task(conn, task_id).status == "running" + + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.status == "active" + assert record.disposition is None + + +def test_registry_failure_rolls_back_task_terminal_transition( + tmp_path, kanban_home, monkeypatch +): + import importlib + + live_kw = importlib.import_module("hermes_cli.kanban_workspaces") + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "registry-failure" + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="registry failure", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/registry-failure", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + + def fail_registry_update(*_args, **_kwargs): + raise RuntimeError("injected registry failure") + + monkeypatch.setattr( + live_kw, + "apply_terminal_disposition_in_transaction", + fail_registry_update, + raising=False, + ) + with pytest.raises(RuntimeError, match="injected registry failure"): + kb.complete_task( + conn, + task_id, + metadata={"workspace_disposition": "retained_until:review"}, + ) + assert kb.get_task(conn, task_id).status == "ready" + + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.status == "active" + assert record.disposition is None + + +def test_protected_import_never_reclassifies_task_owned_workspace( + tmp_path, kanban_home +): + import yaml + + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "task-owned" + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="task owned", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/task-owned", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + + policy = tmp_path / "task-owned-exception.yaml" + policy.write_text( + yaml.safe_dump( + { + "version": 1, + "exceptions": [ + { + "policy_id": "cannot-adopt-task-owned", + "repo_path": str(repo), + "workspace_path": str(target), + } + ], + } + ), + encoding="utf-8", + ) + result = kw.import_protected_exceptions(policy) + assert result["imported"] == [] + assert result["skipped"] == [ + { + "policy_id": "cannot-adopt-task-owned", + "workspace_path": str(target.resolve()), + "reason": "task_owned_registry_record", + } + ] + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.status == "active" + assert record.disposition is None + assert record.exception_policy_id is None + + +def test_repository_identity_is_shared_by_primary_and_linked_checkouts( + tmp_path, kanban_home +): + repo = _make_repo(tmp_path) + linked = repo / ".worktrees" / "linked-anchor" + target = repo / ".worktrees" / "owned" + _git(repo, "worktree", "add", "-b", "feat/linked-anchor", str(linked), "HEAD") + + first = kw.reserve_workspace( + repo_path=repo, + workspace_path=target, + task_id="t_repo_identity", + board_id="default", + owner_profile="developer", + branch="feat/owned", + ) + second = kw.reserve_workspace( + repo_path=linked, + workspace_path=target, + task_id="t_repo_identity", + board_id="default", + owner_profile="developer", + branch="feat/owned", + ) + + assert second.workspace_id == first.workspace_id + assert second.repo_id == first.repo_id + assert second.repo_path == str(repo.resolve()) + assert len(kw.list_workspace_records(task_id="t_repo_identity")) == 1 + + +def test_dispositions_require_recovery_owner_and_current_exception_policy( + tmp_path, kanban_home, monkeypatch +): + import importlib + import yaml + + live_kw = importlib.import_module("hermes_cli.kanban_workspaces") + repo = _make_repo(tmp_path) + dirty = repo / ".worktrees" / "dirty-policy" + protected = repo / ".worktrees" / "protected-policy" + with kb.connect() as conn: + dirty_id = kb.create_task( + conn, + title="dirty policy", + workspace_kind="worktree", + workspace_path=str(dirty), + branch_name="feat/dirty-policy", + ) + kb.resolve_workspace(kb.get_task(conn, dirty_id)) + (dirty / "recovery.patch").write_text("recover\n", encoding="utf-8") + with pytest.raises(RuntimeError, match="artifact.*owner"): + kb.complete_task( + conn, + dirty_id, + metadata={"workspace_disposition": "preserved_dirty:review later"}, + ) + + protected_id = kb.create_task( + conn, + title="protected policy", + workspace_kind="worktree", + workspace_path=str(protected), + branch_name="feat/protected-policy", + ) + kb.resolve_workspace(kb.get_task(conn, protected_id)) + with pytest.raises(RuntimeError, match="current operational exception policy"): + kb.archive_task( + conn, + protected_id, + workspace_disposition="operational_exception:current-policy", + ) + + policy = tmp_path / "current-policy.yaml" + policy.write_text( + yaml.safe_dump( + { + "version": 1, + "exceptions": [ + { + "policy_id": "current-policy", + "repo_path": str(repo), + "workspace_path": str(protected), + "expires_at": None, + } + ], + } + ), + encoding="utf-8", + ) + monkeypatch.setattr( + live_kw, "_configured_exception_path", lambda: policy + ) + assert kb.archive_task( + conn, + protected_id, + workspace_disposition="operational_exception:current-policy", + ) + + +def test_inventory_reports_expired_registered_exception_without_config( + tmp_path, kanban_home +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "expired-registered" + _git(repo, "worktree", "add", "-b", "ops/expired-registered", str(target), "HEAD") + now = 1 + with kw.connect_registry() as registry: + registry.execute( + "INSERT INTO workspaces (workspace_id, repo_id, repo_path, workspace_path, " + "owner_profile, purpose, branch, status, cleanup_policy, disposition, " + "exception_policy_id, exception_expires_at, created_at, last_used_at) " + "VALUES ('w_expired', ?, ?, ?, 'operator', 'operational', " + "'ops/expired-registered', 'protected', 'never_automatic', " + "'operational_exception:expired-registered', 'expired-registered', 1, ?, ?)", + (kw._repo_id(str(repo.resolve())), str(repo.resolve()), str(target.resolve()), now, now), + ) + + report = kw.reconcile_inventory(repo_paths=[repo], approved_roots=[repo / ".worktrees"]) + assert report["missing_expired_protected_exceptions"] == [ + { + "policy_id": "expired-registered", + "workspace_path": str(target.resolve()), + "reason": "expired_policy", + } + ] + + +def test_reservation_reuse_is_exact_and_creation_failure_cannot_downgrade_active( + tmp_path, kanban_home +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "concurrent" + reservation = kw.reserve_workspace( + repo_path=repo, + workspace_path=target, + task_id="t_concurrent", + board_id="default", + owner_profile="developer", + purpose="implementation", + branch="feat/concurrent", + cleanup_policy="on_task_terminal", + retention_condition="task_terminal", + ) + _git(repo, "worktree", "add", "-b", "feat/concurrent", str(target), "HEAD") + kw.verify_workspace(reservation.workspace_id, target) + kw.mark_creation_failed(reservation.workspace_id) + assert kw.list_workspace_records(task_id="t_concurrent")[0].status == "active" + + with pytest.raises(RuntimeError, match="ambiguous worktree ownership"): + kw.reserve_workspace( + repo_path=repo, + workspace_path=target, + task_id="t_concurrent", + board_id="default", + owner_profile="reviewer", + purpose="review", + branch="feat/concurrent", + cleanup_policy="never_automatic", + retention_condition="manual", + ) + + +def test_janitor_uses_production_pr_and_task_dependency_evidence( + tmp_path, kanban_home +): + repo = _make_repo(tmp_path) + approved = repo / ".worktrees" + pr_path = approved / "explicit-pr" + dependency_path = approved / "explicit-dependency" + with kb.connect() as conn: + pr_id = kb.create_task( + conn, + title="explicit PR", + workspace_kind="worktree", + workspace_path=str(pr_path), + branch_name="feat/explicit-pr", + ) + kb.resolve_workspace(kb.get_task(conn, pr_id)) + assert kb.complete_task( + conn, + pr_id, + metadata={ + "workspace_disposition": "retained_until:janitor", + "workspace_pr_numbers": [71978], + }, + ) + + dependency_id = kb.create_task( + conn, + title="explicit dependency", + workspace_kind="worktree", + workspace_path=str(dependency_path), + branch_name="feat/explicit-dependency", + ) + kb.resolve_workspace(kb.get_task(conn, dependency_id)) + assert kb.complete_task( + conn, + dependency_id, + metadata={"workspace_disposition": "retained_until:janitor"}, + ) + successor_id = kb.create_task(conn, title="active successor", parents=[dependency_id]) + assert kb.get_task(conn, successor_id).status == "ready" + + pr_record = kw.list_workspace_records(task_id=pr_id)[0] + assert pr_record.pr_numbers == (71978,) + plan = kw.plan_janitor(repo_paths=[repo], approved_roots=[approved]) + reasons = { + item["workspace_path"]: item["reasons"] for item in plan["excluded"] + } + assert "open_or_unverifiable_pr" in reasons[str(pr_path.resolve())] + assert "dependency_retained" in reasons[str(dependency_path.resolve())] diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index b1f16cfc269ae..14274a542223e 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -65,7 +65,7 @@ They coexist: a kanban worker may call `delegate_task` internally during its run - **Workspace** — the directory a worker operates in. Three kinds: - `scratch` (default) — fresh tmp dir under `~/.hermes/kanban/workspaces//` (or `~/.hermes/kanban/boards//workspaces//` on non-default boards). **Deleted when the task completes** — scratch is ephemeral by design. Files explicitly declared through `kanban_complete(artifacts=[...])` are copied into durable per-task attachment storage before cleanup; existing deliverable paths in legacy completion summaries receive the same treatment. Other scratch files are removed. A missing declared scratch artifact keeps the task in-flight so the worker can correct the path and retry. Use `worktree:` or `dir:` when the whole workspace should remain available. The first time a scratch workspace is created on an install, the dispatcher logs a warning and emits a `tip_scratch_workspace` event on the task (visible via `hermes kanban show `). - `dir:` — an existing shared directory (Obsidian vault, mail ops dir, per-account folder). **Must be an absolute path.** Relative paths like `dir:../tenants/foo/` are rejected at dispatch because they'd resolve against whatever CWD the dispatcher happens to be in, which is ambiguous and a confused-deputy escape vector. The path is otherwise trusted — it's your box, your filesystem, the worker runs with your uid. This is the trusted-local-user threat model; kanban is single-host by design. **Preserved on completion.** - - `worktree` — a git worktree under `.worktrees//` for coding tasks. Use `worktree:` to pin the exact target path. Worker-side `git worktree add` creates it, using `--branch` when provided. **Preserved on completion.** + - `worktree` — a git worktree under `.worktrees//` for coding tasks. Use `worktree:` to pin the exact target path. Worker-side `git worktree add` creates it, using `--branch` when provided. Every new worktree is registered before Git mutates the repository. Completion preserves the path only after an explicit lifecycle disposition is recorded. - **Dispatcher** — a long-lived loop that, every N seconds (default 60): reclaims stale claims, reclaims crashed workers (PID gone but TTL not yet expired), promotes ready tasks, atomically claims, spawns assigned profiles. Runs **inside the gateway** by default (`kanban.dispatch_in_gateway: true`). One dispatcher sweeps all boards per tick; workers are spawned with `HERMES_KANBAN_BOARD` pinned so they can't see other boards. After `kanban.failure_limit` consecutive spawn failures on the same task (default: 2) the dispatcher auto-blocks it with the last error as the reason — prevents thrashing on tasks whose profile doesn't exist, workspace can't mount, etc. - **Tenant** — optional string namespace *within* a board. One specialist fleet can serve multiple businesses (`--tenant business-a`) with data isolation by workspace path and memory key prefix. Tenants are a soft filter; boards are the hard isolation boundary. @@ -265,6 +265,82 @@ hermes kanban unblock t_abc t_def hermes kanban block t_abc "need input" --ids t_def t_hij ``` +### Managed worktree lifecycle + +Kanban keeps a shared, profile-aware ownership registry at +`/kanban/workspace-registry.db`. Git's +`worktree list --porcelain` output remains the authority for physical +checkouts; the registry is the authority for task/board ownership, purpose, +cleanup policy, retention conditions, protected exceptions, and verification +receipts. Inventory never adopts or deletes an unknown path. + +New worktree claims reuse the same compatible task/repository/path record. +Creating a second persistent worktree for that task and repository fails +closed unless the task has an explicit `workspace_replacement_reason` (CLI: +`--workspace-replacement-reason`). Pressure limits under +`kanban.worktree_lifecycle` are report-only by default. + +Worktree-owning tasks must classify every registered checkout before becoming +`done` or `archived`: + +- `retired` +- `retained_until:` +- `preserved_dirty:artifact=;owner=` +- `operational_exception:` + +A dirty checkout must name both its recovery artifact and accountable owner; +`retained_until` cannot be used to hide recovery-owned changes. Exception +policy IDs and workspace paths must each be unique within a policy file, and an +`operational_exception` disposition is accepted only when that exact current, +unexpired policy is configured for the workspace. Completion metadata may also +record associated pull requests with `workspace_pr_numbers: [123, ...]`; the +dry-run janitor treats every recorded PR as open or unverifiable and excludes it. + +For completion, include `workspace_disposition` in structured metadata. For a +direct archive, use `--workspace-disposition` unless the completed task already +has a valid recorded disposition: + +```bash +hermes kanban complete t_abc --summary "ready for review" \ + --metadata '{"workspace_disposition":"retained_until:review-and-pr-closeout"}' +hermes kanban archive t_def \ + --workspace-disposition retained_until:review-and-pr-closeout +``` + +Inventory is deterministic in both terminal and JSON modes: + +```bash +hermes kanban worktrees inventory \ + --repo /path/to/repo \ + --approved-root /path/to/repo/.worktrees +hermes kanban worktrees inventory \ + --repo /path/to/repo \ + --approved-root /path/to/repo/.worktrees \ + --exception-config /path/to/worktree-exceptions.yaml \ + --json +``` + +Protected operational checkouts are imported only from an explicit versioned +YAML policy. Start from +`docs/examples/kanban-worktree-exceptions.yaml`; the importer validates that +each path is already a Git worktree and skips expired or missing entries: + +```bash +hermes kanban worktrees exceptions import \ + --config /path/to/worktree-exceptions.yaml --json +``` + +The janitor is intentionally dry-run only. It emits candidates and drift +receipts but performs zero removals, never uses `--force`, and never mutates +branches or stashes. Primary and locked worktrees are always excluded, even +when an overly broad approved root is supplied: + +```bash +hermes kanban worktrees janitor \ + --repo /path/to/repo \ + --approved-root /path/to/repo/.worktrees --json +``` + :::note Where an unblocked task lands `unblock` itself only ever moves a task to **`ready`** (all parents `done`) or **`todo`** (a parent is still open — the task is dependency-gated and the From 783b59e47f530a8dd33a1a371851c4d070af8e6a Mon Sep 17 00:00:00 2001 From: Royce Date: Wed, 5 Aug 2026 23:13:04 -0700 Subject: [PATCH 2/3] ci: retry flaky Codex persistence test From 25d8026b1b9731cdb45469d061a4e9f62dbf3053 Mon Sep 17 00:00:00 2001 From: Royce Date: Thu, 6 Aug 2026 00:33:05 -0700 Subject: [PATCH 3/3] [verified] fix: harden Kanban worktree resolution --- hermes_cli/kanban_db.py | 210 +++++++--- hermes_cli/kanban_workspaces.py | 393 +++++++++++++++--- .../test_kanban_worktree_isolation.py | 31 +- .../test_kanban_worktree_lifecycle.py | 178 +++++++- website/docs/user-guide/features/kanban.md | 14 + 5 files changed, 695 insertions(+), 131 deletions(-) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 437a0aa64f5d0..18e4ab0cc1d4e 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -6632,6 +6632,41 @@ def _repo_root_for_worktree_target(path: Path) -> Optional[Path]: current = current.parent +def _primary_worktree_for_checkout(path: Path) -> Optional[Path]: + """Resolve a checkout's primary worktree from Git's shared metadata. + + Filesystem ancestry cannot identify the owner of a linked checkout placed + outside the primary repository hierarchy. ``git worktree list`` is rooted + in the common Git directory and reports the primary checkout first, so it + provides the same identity from any linked checkout. + """ + common_dir = _git_common_dir(path) + if common_dir is None: + return None + try: + result = subprocess.run( + ["git", "-C", str(path), "worktree", "list", "--porcelain"], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=30, + check=False, + ) + except Exception: + return None + if result.returncode != 0: + return None + for line in (result.stdout or "").splitlines(): + if not line.startswith("worktree "): + continue + primary = Path(line.removeprefix("worktree ")).resolve(strict=False) + if _git_common_dir(primary) == common_dir: + return primary + return None + return None + + 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() @@ -6680,29 +6715,71 @@ def _ensure_registered_git_worktree( from hermes_cli import kanban_workspaces as workspaces board_id = board or get_current_board() - reservation = workspaces.reserve_workspace( - repo_path=repo_root, - workspace_path=target, - task_id=task.id, - board_id=board_id, - owner_profile=task.assignee, - branch=branch_name, - replacement_reason=getattr(task, "workspace_replacement_reason", None), - ) - reserved_target = Path(reservation.workspace_path) - try: - _ensure_git_worktree(repo_root, reserved_target, branch_name) - actual_branch = _git_current_branch(reserved_target) - if actual_branch != branch_name: - raise RuntimeError( - f"worktree branch mismatch at {reserved_target}: expected " - f"{branch_name!r}, found {actual_branch or 'detached HEAD'!r}" - ) - workspaces.verify_workspace(reservation.workspace_id, reserved_target) - except Exception: - workspaces.mark_creation_failed(reservation.workspace_id) - raise - return reserved_target + pending_error: Optional[Exception] = None + for _attempt in range(2): + reservation = workspaces.reserve_workspace( + repo_path=repo_root, + workspace_path=target, + task_id=task.id, + board_id=board_id, + owner_profile=task.assignee, + branch=branch_name, + replacement_reason=getattr(task, "workspace_replacement_reason", None), + ) + reserved_target = Path(reservation.workspace_path) + + # An active checkout still passes through the existing materialization + # policy seam. This intentionally leaves branch realignment policy to + # _ensure_git_worktree (and compatible fixes such as PR #71978) while + # lifecycle ownership remains wrapped around it. + if reservation.owns_materialization or reservation.status == "active": + try: + if reservation.owns_materialization: + workspaces.renew_reservation( + reservation.workspace_id, + reservation.reservation_token, + ) + _ensure_git_worktree(repo_root, reserved_target, branch_name) + workspaces.verify_workspace( + reservation.workspace_id, + reserved_target, + reservation_token=reservation.reservation_token, + ) + except Exception: + if reservation.owns_materialization: + workspaces.mark_creation_failed( + reservation.workspace_id, + reservation.reservation_token, + ) + raise + return reserved_target + + # Another process owns this still-reserved row. Never run a second + # materialization attempt and never infer completion from Git metadata: + # HEAD/branch can become visible before `git worktree add` exits. Wait + # for the owner to publish active or creation_failed explicitly. + deadline = time.monotonic() + workspaces.RESERVATION_LEASE_SECONDS + 1 + while True: + try: + workspaces.verify_workspace( + reservation.workspace_id, reserved_target + ) + return reserved_target + except workspaces.WorkspaceReservationPending as exc: + pending_error = exc + except RuntimeError: + current = workspaces.get_workspace_record(reservation.workspace_id) + if current.status == "creation_failed": + break + raise + current = workspaces.get_workspace_record(reservation.workspace_id) + if current.status == "creation_failed" or time.monotonic() >= deadline: + break + time.sleep(0.05) + + raise RuntimeError( + f"timed out waiting for compatible workspace reservation at {target}" + ) from pending_error def _resolve_worktree_workspace( @@ -6762,7 +6839,7 @@ def _resolve_worktree_workspace( actual_branch = _git_current_branch(requested) if actual_branch == branch_name: matched_branch = actual_branch or branch_name - existing_root = _repo_root_for_worktree_target(requested.parent) + existing_root = _primary_worktree_for_checkout(requested) if existing_root is None: raise ValueError( f"task {task.id} worktree path {task.workspace_path!r} is linked " @@ -6776,10 +6853,11 @@ def _resolve_worktree_workspace( board=board, ) return registered.resolve(strict=False), matched_branch - # A path already registered to this exact task is not a sibling's - # checkout to route around: a different current branch is ownership - # drift. Fail closed without creating a replacement or downgrading the - # existing registry record. + + # A path already owned by this exact task is a concrete lifecycle + # target, even when it is not named /.worktrees/. Route + # it through the materialization policy seam rather than rejecting its + # current branch here; branch realignment policy composes at that seam. from hermes_cli import kanban_workspaces as _workspaces registered_paths = { @@ -6791,33 +6869,46 @@ def _resolve_worktree_workspace( if record.status != "creation_failed" } if requested_resolved in registered_paths: - raise RuntimeError( - f"worktree branch mismatch at {requested_resolved}: expected " - f"{branch_name!r}, found {actual_branch or 'detached HEAD'!r}" + registered_root = _primary_worktree_for_checkout(requested) + if registered_root is None: + raise ValueError( + f"task {task.id} registered worktree {requested_resolved} has no " + "resolvable primary repository" + ) + registered = _ensure_registered_git_worktree( + task, + registered_root, + requested_resolved, + branch_name, + board=board, ) - # 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. + return registered.resolve(strict=False), branch_name + + # A checkout owned by the surrounding repository is a concrete target. + # Route siblings to their canonical target, but let an exact target pass + # through the materialization policy seam so branch realignment remains + # composable with PR #71978 rather than being rejected beforehand. fallback_root = _repo_root_for_worktree_target(requested.parent) - if fallback_root is not None: + requested_common = _git_common_dir(requested) + if ( + fallback_root is not None + and _git_common_dir(fallback_root) == requested_common + ): fallback = fallback_root / ".worktrees" / task.id - if fallback.resolve(strict=False) != requested_resolved: - fallback = _ensure_registered_git_worktree( - task, fallback_root, fallback, branch_name, board=board - ) - return fallback.resolve(strict=False), branch_name - # No distinct safe fallback exists (including when the occupied path - # is this task's canonical target). Never silently run on another - # branch and never auto-switch a checkout that may contain unique work. - raise ValueError( - f"task {task.id} linked worktree target {str(requested)!r} has branch " - f"mismatch: expected {branch_name!r}, found " - f"{actual_branch or 'detached HEAD'!r}" + fallback = _ensure_registered_git_worktree( + task, fallback_root, fallback, branch_name, board=board + ) + return fallback.resolve(strict=False), branch_name + + # A linked checkout not owned by a surrounding checkout is itself the + # explicit linked-root anchor (PR #70585 semantics). Materialize below + # that exact anchor; lifecycle canonicalizes ownership to the primary + # repository through Git common-dir/worktree metadata. + linked_target = requested_resolved / ".worktrees" / task.id + linked_target = _ensure_registered_git_worktree( + task, requested_resolved, linked_target, branch_name, board=board ) + return linked_target.resolve(strict=False), branch_name repo_root = _git_toplevel(requested) if repo_root is not None and requested_resolved == repo_root: @@ -6854,13 +6945,16 @@ def resolve_workspace(task: Task, *, board: Optional[str] = None) -> Path: 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/``. + worktree at ``/.worktrees/``. Existing linked checkouts are + identified through Git common-dir/worktree metadata: exact same-branch + targets are reused even outside the primary repo hierarchy, while a + standalone wrong-branch linked checkout acts as a linked-root anchor. + Exact registered targets pass through ``_ensure_git_worktree`` so branch + policy remains composable at that seam. 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. diff --git a/hermes_cli/kanban_workspaces.py b/hermes_cli/kanban_workspaces.py index cc4508efa5fea..b9cdf66ccf019 100644 --- a/hermes_cli/kanban_workspaces.py +++ b/hermes_cli/kanban_workspaces.py @@ -44,6 +44,7 @@ "retirement_queued", "protected", } +RESERVATION_LEASE_SECONDS = 180 @dataclass(frozen=True) @@ -103,6 +104,38 @@ def from_row(cls, row: sqlite3.Row) -> "WorkspaceRecord": ) +@dataclass(frozen=True) +class WorkspaceReservation(WorkspaceRecord): + """A registry record plus this caller's materialization claim. + + Compatible concurrent callers can observe the same registry row, but only + the caller holding ``reservation_token`` may execute ``git worktree add``, + publish the row as active, or mark creation failed. Non-owners wait for that + explicit active publication before using the checkout. + """ + + reservation_token: Optional[str] = None + owns_materialization: bool = False + + @classmethod + def from_record( + cls, + record: WorkspaceRecord, + *, + reservation_token: Optional[str], + owns_materialization: bool, + ) -> "WorkspaceReservation": + return cls( + **record.__dict__, + reservation_token=reservation_token, + owns_materialization=owns_materialization, + ) + + +class WorkspaceReservationPending(RuntimeError): + """The reservation owner has not exposed a verifiable Git checkout yet.""" + + @dataclass(frozen=True) class _TerminalDispositionPlan: record: WorkspaceRecord @@ -141,6 +174,8 @@ class _TerminalDispositionPlan: exception_policy_id TEXT, exception_expires_at INTEGER, replacement_reason TEXT, + reservation_token TEXT, + reservation_expires_at INTEGER, created_at INTEGER NOT NULL, last_used_at INTEGER NOT NULL, last_verified_at INTEGER @@ -160,6 +195,21 @@ def registry_path() -> Path: return kanban_db.kanban_home() / "kanban" / "workspace-registry.db" +def _add_registry_column( + conn: sqlite3.Connection, columns: set[str], name: str, ddl: str +) -> None: + if name in columns: + return + try: + conn.execute(ddl) + except sqlite3.OperationalError as exc: + # Two first-use processes can both observe the pre-migration schema. + # ALTER TABLE is not IF-NOT-EXISTS-capable on supported SQLite builds; + # the loser may safely accept the winner's identical column. + if "duplicate column name" not in str(exc).lower(): + raise + + def connect_registry() -> sqlite3.Connection: path = registry_path() path.parent.mkdir(parents=True, exist_ok=True) @@ -171,10 +221,24 @@ def connect_registry() -> sqlite3.Connection: str(row[1]) for row in conn.execute("PRAGMA table_info(workspaces)").fetchall() } - if "exception_expires_at" not in columns: - conn.execute( - "ALTER TABLE workspaces ADD COLUMN exception_expires_at INTEGER" - ) + _add_registry_column( + conn, + columns, + "exception_expires_at", + "ALTER TABLE workspaces ADD COLUMN exception_expires_at INTEGER", + ) + _add_registry_column( + conn, + columns, + "reservation_token", + "ALTER TABLE workspaces ADD COLUMN reservation_token TEXT", + ) + _add_registry_column( + conn, + columns, + "reservation_expires_at", + "ALTER TABLE workspaces ADD COLUMN reservation_expires_at INTEGER", + ) return conn @@ -340,19 +404,53 @@ def reserve_workspace( cleanup_policy: str = "on_task_terminal", retention_condition: Optional[str] = "task_terminal", replacement_reason: Optional[str] = None, -) -> WorkspaceRecord: +) -> WorkspaceReservation: """Reserve ownership before ``git worktree add`` can mutate the repo. - A single compatible task/repository reservation is reused. Multiple active - rows, or one incompatible row, are ambiguous and fail closed unless the - task carries an explicit replacement reason. + The registry row remains the durable ownership identity. A short-lived, + compare-and-swap reservation token separately identifies the one caller + allowed to materialize it. Compatible concurrent callers observe the same + row without receiving that token and can only verify a live checkout. """ repo = _canonical_repo(repo_path) target = _canonical(workspace_path) rid = _repo_id(repo) - now = int(time.time()) for warning in creation_pressure_warnings(repo): _log.warning("kanban worktree lifecycle (report-only): %s", warning) + now = int(time.time()) + claim_token = uuid.uuid4().hex + claim_expires = now + RESERVATION_LEASE_SECONDS + + def is_compatible(row: sqlite3.Row) -> bool: + return ( + row["workspace_path"] == target + and row["task_id"] == task_id + and row["board_id"] == board_id + and row["repo_id"] == rid + and row["owner_profile"] == owner_profile + and row["purpose"] == purpose + and row["branch"] == branch + and row["cleanup_policy"] == cleanup_policy + and row["retention_condition"] == retention_condition + ) + + def result_for( + conn: sqlite3.Connection, + workspace_id: str, + *, + token: Optional[str], + owns: bool, + ) -> WorkspaceReservation: + row = conn.execute( + "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) + ).fetchone() + if row is None: + raise RuntimeError(f"unknown workspace reservation: {workspace_id}") + return WorkspaceReservation.from_record( + WorkspaceRecord.from_row(row), + reservation_token=token, + owns_materialization=owns, + ) with contextlib.closing(connect_registry()) as conn: conn.execute("BEGIN IMMEDIATE") @@ -363,27 +461,32 @@ def reserve_workspace( "ORDER BY workspace_path, workspace_id", (board_id, task_id, rid), ).fetchall() - compatible = [ - row for row in rows - if row["workspace_path"] == target - and row["owner_profile"] == owner_profile - and row["purpose"] == purpose - and row["branch"] == branch - and row["cleanup_policy"] == cleanup_policy - and row["retention_condition"] == retention_condition - ] + compatible = [row for row in rows if is_compatible(row)] if len(rows) == 1 and len(compatible) == 1: - conn.execute( - "UPDATE workspaces SET last_used_at = ? " - "WHERE workspace_id = ?", - (now, compatible[0]["workspace_id"]), - ) - conn.execute("COMMIT") - return WorkspaceRecord.from_row( + row = compatible[0] + owns = False + token: Optional[str] = None + if row["status"] == "reserved" and ( + row["reservation_token"] is None + or int(row["reservation_expires_at"] or 0) <= now + ): + cur = conn.execute( + "UPDATE workspaces SET reservation_token = ?, " + "reservation_expires_at = ?, last_used_at = ? " + "WHERE workspace_id = ? AND status = 'reserved' AND (" + "reservation_token IS NULL OR reservation_expires_at <= ?)", + (claim_token, claim_expires, now, row["workspace_id"], now), + ) + owns = cur.rowcount == 1 + token = claim_token if owns else None + else: conn.execute( - "SELECT * FROM workspaces WHERE workspace_id = ?", - (compatible[0]["workspace_id"],), - ).fetchone() + "UPDATE workspaces SET last_used_at = ? WHERE workspace_id = ?", + (now, row["workspace_id"]), + ) + conn.execute("COMMIT") + return result_for( + conn, row["workspace_id"], token=token, owns=owns ) if rows and not replacement_reason: raise RuntimeError( @@ -396,16 +499,33 @@ def reserve_workspace( "SELECT * FROM workspaces WHERE workspace_path = ?", (target,) ).fetchone() if existing is not None: + if is_compatible(existing) and existing["status"] == "creation_failed": + cur = conn.execute( + "UPDATE workspaces SET status = 'reserved', reservation_token = ?, " + "reservation_expires_at = ?, last_used_at = ? " + "WHERE workspace_id = ? AND status = 'creation_failed'", + ( + claim_token, + claim_expires, + now, + existing["workspace_id"], + ), + ) + if cur.rowcount != 1: + raise RuntimeError( + f"workspace reservation changed while retrying: " + f"{existing['workspace_id']}" + ) + conn.execute("COMMIT") + return result_for( + conn, + existing["workspace_id"], + token=claim_token, + owns=True, + ) exact_compatible_owner = ( - existing["task_id"] == task_id - and existing["board_id"] == board_id - and existing["repo_id"] == rid + is_compatible(existing) and existing["status"] in {"reserved", "active"} - and existing["owner_profile"] == owner_profile - and existing["purpose"] == purpose - and existing["branch"] == branch - and existing["cleanup_policy"] == cleanup_policy - and existing["retention_condition"] == retention_condition ) if not exact_compatible_owner and not replacement_reason: raise RuntimeError( @@ -415,7 +535,9 @@ def reserve_workspace( ) if exact_compatible_owner: conn.execute("COMMIT") - return WorkspaceRecord.from_row(existing) + return result_for( + conn, existing["workspace_id"], token=None, owns=False + ) raise RuntimeError( f"replacement reason does not authorize stealing an existing " f"workspace path: {target}; choose a new path" @@ -426,64 +548,199 @@ def reserve_workspace( "INSERT INTO workspaces (" "workspace_id, repo_id, repo_path, workspace_path, task_id, board_id, " "owner_profile, purpose, branch, status, cleanup_policy, " - "retention_condition, replacement_reason, created_at, last_used_at" - ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'reserved', ?, ?, ?, ?, ?)", + "retention_condition, replacement_reason, reservation_token, " + "reservation_expires_at, created_at, last_used_at" + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'reserved', ?, ?, ?, ?, ?, ?, ?)", ( - workspace_id, rid, repo, target, task_id, board_id, - owner_profile, purpose, branch, cleanup_policy, - retention_condition, replacement_reason, now, now, + workspace_id, + rid, + repo, + target, + task_id, + board_id, + owner_profile, + purpose, + branch, + cleanup_policy, + retention_condition, + replacement_reason, + claim_token, + claim_expires, + now, + now, ), ) - row = conn.execute( - "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) - ).fetchone() conn.execute("COMMIT") - return WorkspaceRecord.from_row(row) + return result_for( + conn, workspace_id, token=claim_token, owns=True + ) except Exception: - conn.execute("ROLLBACK") + if conn.in_transaction: + conn.execute("ROLLBACK") raise -def verify_workspace(workspace_id: str, workspace_path: Path | str) -> WorkspaceRecord: +def get_workspace_record(workspace_id: str) -> WorkspaceRecord: + with contextlib.closing(connect_registry()) as conn: + row = conn.execute( + "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) + ).fetchone() + if row is None: + raise RuntimeError(f"unknown workspace reservation: {workspace_id}") + return WorkspaceRecord.from_row(row) + + +def renew_reservation( + workspace_id: str, reservation_token: Optional[str] +) -> None: + """Renew a materializer's lease immediately before Git mutation.""" + if reservation_token is None: + raise RuntimeError(f"workspace reservation {workspace_id} has no owner token") + now = int(time.time()) + with contextlib.closing(connect_registry()) as conn: + cur = conn.execute( + "UPDATE workspaces SET reservation_expires_at = ?, last_used_at = ? " + "WHERE workspace_id = ? AND status = 'reserved' " + "AND reservation_token = ?", + ( + now + RESERVATION_LEASE_SECONDS, + now, + workspace_id, + reservation_token, + ), + ) + if cur.rowcount != 1: + raise RuntimeError( + f"workspace reservation {workspace_id} is no longer owned by this resolver" + ) + + +def verify_workspace( + workspace_id: str, + workspace_path: Path | str, + *, + reservation_token: Optional[str] = None, +) -> WorkspaceRecord: + """Verify Git facts and activate a reservation with compare-and-swap safety. + + The reservation owner supplies its token and is the only caller allowed to + promote ``reserved`` to ``active``. Non-owners may reverify an already-active + checkout, but cannot infer materialization completion from Git metadata that + becomes visible before ``git worktree add`` exits. + """ path = _canonical(workspace_path) now = int(time.time()) - head = _git(path, "rev-parse", "HEAD") - branch = _git(path, "branch", "--show-current") or None - dirty_hash = dirty_manifest_hash(path) - size = estimated_bytes(path) with contextlib.closing(connect_registry()) as conn: + row = conn.execute( + "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) + ).fetchone() + if row is None: + raise RuntimeError(f"unknown workspace reservation: {workspace_id}") + if row["status"] not in {"reserved", "active"}: + raise RuntimeError( + f"workspace reservation {workspace_id} is not reservable or active" + ) + if row["workspace_path"] != path: + raise RuntimeError( + f"workspace reservation {workspace_id} path mismatch: expected " + f"{row['workspace_path']}, found {path}" + ) + if row["status"] == "reserved": + if reservation_token is None: + raise WorkspaceReservationPending( + f"workspace reservation {workspace_id} is still materializing" + ) + if row["reservation_token"] != reservation_token: + raise RuntimeError( + f"workspace reservation {workspace_id} is owned by another resolver" + ) + + head = _git(path, "rev-parse", "HEAD") + actual_branch = _git(path, "branch", "--show-current") or None + missing_expected_branch = row["branch"] is not None and actual_branch is None + if head is None or missing_expected_branch: + if row["status"] == "reserved": + raise WorkspaceReservationPending( + f"workspace reservation {workspace_id} is not materialized yet" + ) + raise RuntimeError( + f"active workspace reservation {workspace_id} is not a usable checkout" + ) + actual_repo = _canonical_repo(path) + if actual_repo != row["repo_path"]: + raise RuntimeError( + f"workspace reservation {workspace_id} repository mismatch: expected " + f"{row['repo_path']}, found {actual_repo}" + ) + if row["branch"] is not None and actual_branch != row["branch"]: + raise RuntimeError( + f"workspace reservation {workspace_id} branch mismatch: expected " + f"{row['branch']!r}, found {actual_branch!r}" + ) + dirty_hash = dirty_manifest_hash(path) + size = estimated_bytes(path) + conn.execute("BEGIN IMMEDIATE") try: + current = conn.execute( + "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) + ).fetchone() + if current is None or current["status"] not in {"reserved", "active"}: + raise RuntimeError( + f"workspace reservation {workspace_id} is not reservable or active" + ) + if ( + current["status"] == "reserved" + and reservation_token is not None + and current["reservation_token"] != reservation_token + ): + raise RuntimeError( + f"workspace reservation {workspace_id} is owned by another resolver" + ) cur = conn.execute( - "UPDATE workspaces SET workspace_path = ?, branch = COALESCE(?, branch), " - "head_sha = ?, status = 'active', dirty_manifest_hash = ?, " - "estimated_bytes = ?, last_used_at = ?, last_verified_at = ? " + "UPDATE workspaces SET head_sha = ?, status = 'active', " + "dirty_manifest_hash = ?, estimated_bytes = ?, last_used_at = ?, " + "last_verified_at = ?, reservation_token = NULL, " + "reservation_expires_at = NULL " "WHERE workspace_id = ? AND status IN ('reserved', 'active')", - (path, branch, head, dirty_hash, size, now, now, workspace_id), + (head, dirty_hash, size, now, now, workspace_id), ) if cur.rowcount != 1: raise RuntimeError( - f"workspace reservation {workspace_id} is not reservable or active" + f"workspace reservation {workspace_id} changed during verification" ) - row = conn.execute( + verified = conn.execute( "SELECT * FROM workspaces WHERE workspace_id = ?", (workspace_id,) ).fetchone() - if row is None: - raise RuntimeError(f"unknown workspace reservation: {workspace_id}") conn.execute("COMMIT") - return WorkspaceRecord.from_row(row) + return WorkspaceRecord.from_row(verified) except Exception: - conn.execute("ROLLBACK") + if conn.in_transaction: + conn.execute("ROLLBACK") raise -def mark_creation_failed(workspace_id: str) -> None: +def mark_creation_failed( + workspace_id: str, reservation_token: Optional[str] = None +) -> None: + """Fail only the still-reserved row owned by this materializer.""" with contextlib.closing(connect_registry()) as conn: - conn.execute( - "UPDATE workspaces SET status = 'creation_failed', last_verified_at = ? " - "WHERE workspace_id = ? AND status = 'reserved'", - (int(time.time()), workspace_id), - ) + if reservation_token is None: + conn.execute( + "UPDATE workspaces SET status = 'creation_failed', " + "last_verified_at = ?, reservation_expires_at = NULL " + "WHERE workspace_id = ? AND status = 'reserved' " + "AND reservation_token IS NULL", + (int(time.time()), workspace_id), + ) + else: + conn.execute( + "UPDATE workspaces SET status = 'creation_failed', " + "last_verified_at = ?, reservation_token = NULL, " + "reservation_expires_at = NULL WHERE workspace_id = ? " + "AND status = 'reserved' AND reservation_token = ?", + (int(time.time()), workspace_id, reservation_token), + ) def _parse_disposition(value: object) -> tuple[str, Optional[str]]: diff --git a/tests/hermes_cli/test_kanban_worktree_isolation.py b/tests/hermes_cli/test_kanban_worktree_isolation.py index d84fc8d453037..0512ded48c270 100644 --- a/tests/hermes_cli/test_kanban_worktree_isolation.py +++ b/tests/hermes_cli/test_kanban_worktree_isolation.py @@ -66,6 +66,33 @@ def _add_worktree(repo: Path, target: Path, branch: str) -> Path: return target +def test_public_resolver_treats_external_wrong_branch_linked_root_as_anchor( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + linked_root = _add_worktree(repo, tmp_path / "linked-root", "anchor/main") + + with kb.connect() as conn: + tid = kb.create_task( + conn, + title="linked root anchor", + workspace_kind="worktree", + workspace_path=str(linked_root), + branch_name="wt/linked-anchor-task", + ) + task = kb.get_task(conn, tid) + + assert task is not None + workspace = kb.resolve_workspace(task) + assert workspace == linked_root / ".worktrees" / tid + assert subprocess.run( + ["git", "-C", str(workspace), "branch", "--show-current"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() == "wt/linked-anchor-task" + + def test_decompose_worktree_children_get_own_workspace(kanban_home): with kb.connect() as conn: root = kb.create_task(conn, title="build the feature", triage=True) @@ -146,8 +173,8 @@ def test_resolve_canonical_path_on_foreign_branch_fails_closed( task = kb.get_task(conn, tid) assert task is not None - with pytest.raises(ValueError, match="branch mismatch"): - kb._resolve_worktree_workspace(task) + with pytest.raises(RuntimeError, match="branch mismatch"): + kb.resolve_workspace(task) assert subprocess.run( ["git", "-C", str(own), "branch", "--show-current"], capture_output=True, diff --git a/tests/hermes_cli/test_kanban_worktree_lifecycle.py b/tests/hermes_cli/test_kanban_worktree_lifecycle.py index 6ef01b9ad88bc..31bbf08d56fa1 100644 --- a/tests/hermes_cli/test_kanban_worktree_lifecycle.py +++ b/tests/hermes_cli/test_kanban_worktree_lifecycle.py @@ -2,8 +2,10 @@ from __future__ import annotations +import concurrent.futures import json import subprocess +import threading from pathlib import Path import pytest @@ -119,6 +121,32 @@ def test_preexisting_matching_worktree_is_adopted_only_for_its_task( assert records[0].status == "active" +def test_preexisting_matching_external_linked_worktree_is_reused_via_public_resolver( + kanban_home, tmp_path +): + repo = _make_repo(tmp_path) + target = tmp_path / "external-linked-checkout" + _git(repo, "worktree", "add", "-b", "feat/external", str(target), "HEAD") + + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="reuse external linked checkout", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/external", + ) + task = kb.get_task(conn, task_id) + assert task is not None + assert kb.resolve_workspace(task) == target.resolve() + + record = kw.list_workspace_records(task_id=task_id)[0] + assert record.repo_path == str(repo.resolve()) + assert record.workspace_path == str(target.resolve()) + assert record.status == "active" + + def test_existing_target_on_wrong_branch_fails_closed_without_mutating_git( kanban_home, tmp_path ): @@ -180,6 +208,42 @@ def test_drifted_existing_registration_stays_owned_when_branch_check_fails( assert record.workspace_path == str(target.resolve()) +def test_registered_branch_drift_reaches_materialization_policy_seam( + kanban_home, tmp_path, monkeypatch +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "drifted-policy-seam" + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="allow branch policy integration", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/expected", + ) + task = kb.get_task(conn, task_id) + assert task is not None + kb.resolve_workspace(task) + _git(target, "switch", "-c", "feat/drifted") + + calls: list[tuple[Path, Path, str]] = [] + + def realign_at_materialization_seam( + repo_root: Path, requested: Path, branch_name: str + ) -> None: + calls.append((repo_root, requested, branch_name)) + _git(requested, "switch", branch_name) + + monkeypatch.setattr(kb, "_ensure_git_worktree", realign_at_materialization_seam) + current_task = kb.get_task(conn, task_id) + assert current_task is not None + assert kb.resolve_workspace(current_task) == target.resolve() + + assert calls == [(repo.resolve(), target.resolve(), "feat/expected")] + assert _git(target, "branch", "--show-current") == "feat/expected" + assert kw.list_workspace_records(task_id=task_id)[0].status == "active" + + def test_duplicate_task_repo_requires_recorded_replacement_reason( kanban_home, tmp_path ): @@ -403,7 +467,11 @@ def test_inventory_reconciles_git_counts_and_reports_deterministic_findings( branch="feat/duplicate", replacement_reason="clean_replacement_after_dirty_collision", ) - kw.verify_workspace(replacement.workspace_id, duplicate) + kw.verify_workspace( + replacement.workspace_id, + duplicate, + reservation_token=replacement.reservation_token, + ) _git(repo, "worktree", "add", "--detach", str(outside), "HEAD") (owned / "dirty.txt").write_text("preserve me\n", encoding="utf-8") @@ -627,7 +695,11 @@ def test_janitor_never_candidates_primary_or_locked_worktrees( owner_profile="developer", branch="main", ) - kw.verify_workspace(primary_reservation.workspace_id, repo) + kw.verify_workspace( + primary_reservation.workspace_id, + repo, + reservation_token=primary_reservation.reservation_token, + ) assert kb.archive_task( conn, primary_task, @@ -1301,6 +1373,30 @@ def test_inventory_reports_expired_registered_exception_without_config( ] +def test_detached_branchless_reservation_verifies_as_active(tmp_path, kanban_home): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "detached" + _git(repo, "worktree", "add", "--detach", str(target), "HEAD") + + reservation = kw.reserve_workspace( + repo_path=repo, + workspace_path=target, + task_id="t_detached", + board_id="default", + owner_profile="developer", + branch=None, + ) + verified = kw.verify_workspace( + reservation.workspace_id, + target, + reservation_token=reservation.reservation_token, + ) + + assert verified.status == "active" + assert verified.branch is None + assert verified.head_sha == _git(target, "rev-parse", "HEAD") + + def test_reservation_reuse_is_exact_and_creation_failure_cannot_downgrade_active( tmp_path, kanban_home ): @@ -1318,7 +1414,11 @@ def test_reservation_reuse_is_exact_and_creation_failure_cannot_downgrade_active retention_condition="task_terminal", ) _git(repo, "worktree", "add", "-b", "feat/concurrent", str(target), "HEAD") - kw.verify_workspace(reservation.workspace_id, target) + kw.verify_workspace( + reservation.workspace_id, + target, + reservation_token=reservation.reservation_token, + ) kw.mark_creation_failed(reservation.workspace_id) assert kw.list_workspace_records(task_id="t_concurrent")[0].status == "active" @@ -1336,6 +1436,78 @@ def test_reservation_reuse_is_exact_and_creation_failure_cannot_downgrade_active ) +def test_concurrent_public_resolvers_share_one_materialization_without_poisoning( + tmp_path, kanban_home, monkeypatch +): + repo = _make_repo(tmp_path) + target = repo / ".worktrees" / "shared-reservation" + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="concurrent shared reservation", + assignee="developer", + workspace_kind="worktree", + workspace_path=str(target), + branch_name="feat/shared-reservation", + ) + task = kb.get_task(conn, task_id) + assert task is not None + + real_ensure = kb._ensure_git_worktree + real_verify = kw.verify_workspace + materialized = threading.Event() + non_owner_waiting = threading.Event() + release_owner = threading.Event() + call_lock = threading.Lock() + materialization_calls = 0 + + def controlled_materialization( + repo_root: Path, requested: Path, branch_name: str + ) -> None: + nonlocal materialization_calls + with call_lock: + materialization_calls += 1 + call_number = materialization_calls + if call_number != 1: + raise RuntimeError("compatible resolver must not materialize shared reservation") + real_ensure(repo_root, requested, branch_name) + materialized.set() + assert release_owner.wait(timeout=10), "test did not release reservation owner" + + def observed_verification( + workspace_id: str, + workspace_path: Path | str, + *, + reservation_token: str | None = None, + ) -> kw.WorkspaceRecord: + if reservation_token is None: + non_owner_waiting.set() + return real_verify( + workspace_id, + workspace_path, + reservation_token=reservation_token, + ) + + monkeypatch.setattr(kb, "_ensure_git_worktree", controlled_materialization) + monkeypatch.setattr(kw, "verify_workspace", observed_verification) + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: + owner = pool.submit(kb.resolve_workspace, task) + assert materialized.wait(timeout=10), "owner did not materialize worktree" + compatible = pool.submit(kb.resolve_workspace, task) + assert non_owner_waiting.wait(timeout=10), "losing resolver did not observe claim" + with pytest.raises(concurrent.futures.TimeoutError): + compatible.result(timeout=0.1) + release_owner.set() + owner_result = owner.result(timeout=10) + compatible_result = compatible.result(timeout=10) + + assert owner_result == compatible_result == target.resolve() + assert materialization_calls == 1 + records = kw.list_workspace_records(task_id=task_id) + assert len(records) == 1 + assert records[0].status == "active" + + def test_janitor_uses_production_pr_and_task_dependency_evidence( tmp_path, kanban_home ): diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 14274a542223e..d3f19ebcfd42d 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -275,11 +275,25 @@ cleanup policy, retention conditions, protected exceptions, and verification receipts. Inventory never adopts or deletes an unknown path. New worktree claims reuse the same compatible task/repository/path record. +A short-lived reservation token gives only one concurrent resolver permission to +materialize and publish that record. Compatible losing resolvers wait for the +owner's explicit `active` publication instead of inferring completion from Git +metadata that can become visible before `git worktree add` exits. They never run +a second materialization or change the winner's still-reserved status. Creating a second persistent worktree for that task and repository fails closed unless the task has an explicit `workspace_replacement_reason` (CLI: `--workspace-replacement-reason`). Pressure limits under `kanban.worktree_lifecycle` are report-only by default. +Existing linked checkouts are identified through Git common-directory and +`worktree list --porcelain` metadata, not filesystem parent ancestry. An exact +checkout already on the requested branch is reused even outside the primary +repository hierarchy. A wrong-branch checkout owned by a surrounding repository +uses the canonical per-task target; a standalone linked checkout acts as the +explicit linked-root anchor. Exact registered targets continue through the +materialization policy seam so branch-realignment fixes can compose there while +lifecycle ownership and verification remain outside that policy. + Worktree-owning tasks must classify every registered checkout before becoming `done` or `archived`: