From c763161702695ba7c4dff49882105299097e46cc Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Wed, 29 Jul 2026 06:31:52 -0500 Subject: [PATCH] fix(reconcile): server-enforced, revalidated missing-worktree cleanup (#970 review 644 B1-B5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the five blocking findings of review 644 on PR #972. B1 — live ownership and status revalidation. resolve_missing_worktree_binding now re-reads the authoritative lease, session, checkpoint, and issue-lock rows immediately before mutating and diffs them against the snapshot the audit recorded (binding path and identity, lease id/status/session/owner pid, checkpoint path/status, live-session evidence, trustworthy ownership evidence, issue-lock state). Any drift fails closed without mutation, and the binding is reclassified from the live values rather than the audit snapshot. A candidate carrying no audited snapshot is refused rather than trusted. B2 — server-enforced cleanup authorization. Apply mode no longer accepts a client-supplied operator_authorized boolean; it is rejected outright at the MCP tool and in the module (#709 F1 / review 434). Authorization is now the project's own reconciliation cleanup gate, required at both the task-capability boundary (new reconciler-only reconcile_missing_worktree_bindings capability, gitea.branch.delete, role-exclusive) and the production mutation boundary (an authorized audit_reconciliation_mode cleanup phase, re-checked at the point of mutation so a forged authorization mapping cannot stand in for the gate). Dry-run remains available to any gitea.read profile and stays non-mutating. Existing role, repository, parity, and provenance gates are unchanged. B3 — expected-path compare-and-swap. retire_session_checkpoint_worktree_path now requires expected_path and performs a guarded update keyed on the stored path, refusing without mutation when the stored path was moved, replaced, or concurrently changed, when the row is unknown, or when a selector matches more than one checkpoint. retire_lease_worktree_path gains the same treatment plus optional status/session/owner-pid compare-and-swap, and its UPDATE is keyed on the audited path. Both report an idempotent already_retired outcome instead of falsely reporting a retirement. B4 — live-session and issue-lock evidence. session_active is now derived from the control-plane sessions table instead of never being set, along two axes: genuine liveness (recorded active, PID not dead, heartbeat fresh — the rule reused from restart_coordinator) and weaker but still trustworthy recorded ownership. A non-terminal lease now protects its binding regardless of whether the recorded PID is alive, so dead-PID evidence alone can no longer retire a lease the control plane still holds. The previously unused issue_lock_store is now read: a live durable issue lock binding the path or branch blocks cleanup, and locks whose own paths are missing are reported for release through their own lifecycle rather than retired here. B5 — adversarial regression coverage. The suite now drives the registered MCP tools through mcp_server, the real ControlPlaneDB, and the real cleanup gate, covering lease status/ownership/session/path drift, expected-path mismatch, concurrent recreation, an unauthorized caller submitting operator_authorized, wrong profile and missing capability, live-session and trustworthy-owner evidence, conflicting issue locks, non-mutating dry-run, exact-binding-only retirement, preservation of unrelated worktrees and git metadata, idempotent re-execution, and worktrees-dimension resolution. All original #970 acceptance criteria are preserved, including the distinctions between deleted paths, moved paths, unavailable hosts or mounts, transient filesystem failures, live ownership, and concurrent recreation. Tests: focused #970 suite 47 passed. Adjacent suites (capability role invariants, audit reconciliation mode, control plane DB, lease lifecycle, reconciler cleanup integration, delete-branch capability, restart coordinator, bootstrap lock contract) 252 passed / 93 subtests. Full suite 28 failed / 6000 passed, an exact match of the pre-change baseline's 28 failing test ids at 3f584352 (28 failed / 5960 passed). Closes #970 Co-Authored-By: Claude Opus 4.8 (1M context) --- control_plane_db.py | 231 ++++- gitea_mcp_server.py | 68 +- missing_worktree_reconcile.py | 826 ++++++++++++++-- task_capability_map.py | 21 + ...st_issue_970_missing_worktree_reconcile.py | 882 +++++++++++++++++- tests/test_task_capability_role_invariants.py | 6 + 6 files changed, 1930 insertions(+), 104 deletions(-) diff --git a/control_plane_db.py b/control_plane_db.py index f495a55..b8bfcba 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -280,6 +280,22 @@ def _ts(dt: datetime | None = None) -> str: return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") +def _realpath_or_raw(value: str | None) -> str: + """Normalize a filesystem path for compare-and-swap equality (#970). + + Symlinks and ``..`` segments must not make two spellings of the same path + look different, but an unresolvable path must still compare as itself + rather than collapsing to empty — an empty result means "no path given". + """ + text = (value or "").strip() + if not text: + return "" + try: + return os.path.realpath(os.path.abspath(text)) + except Exception: + return text + + def _parse_ts(value: str | None) -> datetime | None: if not value: return None @@ -1918,16 +1934,36 @@ class ControlPlaneDB: lease_id: str, *, expected_path: str | None = None, + expected_status: str | None = None, + expected_session_id: str | None = None, + expected_owner_pid: int | None = None, reason: str = "missing_worktree_path_retired", ) -> dict[str, Any]: """Retire a missing worktree_path binding from a control-plane lease (#970). Clears worktree_path on the lease row, updates provenance_json with - durable retirement audit proof, and writes a worktree_binding_retired event. - Fail closed: if expected_path is provided and does not match current - worktree_path, the update is refused to prevent racing mutations. + durable retirement audit proof, and writes a worktree_binding_retired + event. + + The update is a compare-and-swap (#970 review 644 B1/B3): the caller + states the exact path it audited and, when known, the lease status, + owning session, and owner pid it classified against. Every stated value + must still match the stored row, and the ``UPDATE`` itself is keyed on + the stored ``worktree_path``, so a concurrent writer that moved or + replaced the binding between audit and apply loses the race instead of + having its value silently overwritten. A mismatch raises and mutates + nothing. + + ``expected_path`` is mandatory: a retirement that does not name the path + it intends to clear cannot be safe against concurrent recreation. """ now_s = _ts() + expected_norm = _realpath_or_raw(expected_path) + if not expected_norm: + raise ControlPlaneError( + f"cannot retire lease {lease_id} worktree_path: expected_path is " + "required for compare-and-swap retirement (fail closed)" + ) with self._tx() as conn: cols = self._lease_columns(conn) row = conn.execute( @@ -1937,15 +1973,56 @@ class ControlPlaneDB: if not row: raise ControlPlaneError(f"unknown lease_id {lease_id}") - current_wt = (dict(row).get("worktree_path") or "").strip() - if expected_path and current_wt and os.path.realpath(current_wt) != os.path.realpath(expected_path): + record = dict(row) + current_wt = (record.get("worktree_path") or "").strip() + + if not current_wt: + # Idempotent: the binding this caller audited is already gone. + return { + "lease_id": lease_id, + "retired": False, + "already_retired": True, + "prior_worktree_path": "", + "expected_worktree_path": expected_path, + "reason": reason, + "compare_and_swap": { + "matched": True, + "outcome": "already_retired", + }, + } + + if _realpath_or_raw(current_wt) != expected_norm: raise ControlPlaneError( - f"cannot retire lease {lease_id} worktree_path: expected '{expected_path}' " - f"does not match current '{current_wt}' (fail closed)" + f"cannot retire lease {lease_id} worktree_path: expected " + f"'{expected_path}' does not match current '{current_wt}' " + "(fail closed)" ) + for field, expected_value in ( + ("status", expected_status), + ("session_id", expected_session_id), + ): + if expected_value is None: + continue + current_value = record.get(field) + if str(current_value or "").strip() != str(expected_value).strip(): + raise ControlPlaneError( + f"cannot retire lease {lease_id} worktree_path: lease " + f"{field} changed since audit (expected " + f"'{expected_value}', found '{current_value}'); fail closed" + ) + + if expected_owner_pid is not None: + current_pid = record.get("owner_pid") + if current_pid is not None and int(current_pid) != int(expected_owner_pid): + raise ControlPlaneError( + f"cannot retire lease {lease_id} worktree_path: lease " + f"owner_pid changed since audit (expected " + f"{expected_owner_pid}, found {current_pid}); fail closed" + ) + # Parse and update provenance_json - raw_prov = dict(row).get("provenance_json") or "{}" + raw_prov = record.get("provenance_json") or "{}" try: prov = json.loads(raw_prov) if isinstance(raw_prov, str) else dict(raw_prov) except Exception: @@ -1953,21 +2030,31 @@ class ControlPlaneDB: if not isinstance(prov, dict): prov = {} - prior_path = current_wt or prov.get("worktree_path") + prior_path = current_wt prov.update({ "worktree_path_retired": True, "retired_worktree_path": prior_path, "retired_at": now_s, "retirement_reason": reason, + "retired_from_status": record.get("status"), + "retired_from_session_id": record.get("session_id"), "worktree_path": "", }) prov_json = json.dumps(prov) if "worktree_path" in cols: - conn.execute( - "UPDATE leases SET worktree_path = '', provenance_json = ? WHERE lease_id = ?", - (prov_json, lease_id), + # CAS: keyed on the exact stored path this caller audited. + cur = conn.execute( + "UPDATE leases SET worktree_path = '', provenance_json = ? " + "WHERE lease_id = ? AND worktree_path = ?", + (prov_json, lease_id, record.get("worktree_path")), ) + if cur.rowcount != 1: + raise ControlPlaneError( + f"cannot retire lease {lease_id} worktree_path: " + "compare-and-swap matched no row (concurrent change); " + "fail closed" + ) else: conn.execute( "UPDATE leases SET provenance_json = ? WHERE lease_id = ?", @@ -1980,7 +2067,7 @@ class ControlPlaneDB: VALUES (?, 'worktree_binding_retired', ?, ?) """, ( - row["work_item_id"], + record["work_item_id"], f"lease {lease_id} worktree_path '{prior_path}' retired: {reason}", now_s, ), @@ -1989,9 +2076,18 @@ class ControlPlaneDB: return { "lease_id": lease_id, "retired": True, + "already_retired": False, "prior_worktree_path": prior_path, + "expected_worktree_path": expected_path, "retired_at": now_s, "reason": reason, + "compare_and_swap": { + "matched": True, + "outcome": "retired", + "expected_status": expected_status, + "expected_session_id": expected_session_id, + "expected_owner_pid": expected_owner_pid, + }, } @@ -3140,25 +3236,116 @@ class ControlPlaneDB: *, checkpoint_id: str | None = None, expected_path: str | None = None, + expected_status: str | None = None, reason: str = "missing_worktree_path_retired", ) -> dict[str, Any]: - """Retire a missing worktree_path from session_checkpoints (#970).""" + """Retire a missing worktree_path from session_checkpoints (#970). + + Compare-and-swap, mirroring :meth:`retire_lease_worktree_path` (#970 + review 644 B3). ``expected_path`` names the exact stored path the caller + audited; the guarded ``UPDATE`` is keyed on that stored value, so a + checkpoint whose path was moved, replaced, or concurrently rewritten + after the audit is refused without mutation rather than blindly cleared. + + Exactly one checkpoint row is targeted: by ``checkpoint_id`` when given, + otherwise by ``session_id``, which must identify a single row. + """ now_s = _ts() + expected_norm = _realpath_or_raw(expected_path) + if not expected_norm: + raise ControlPlaneError( + "cannot retire session checkpoint worktree_path: expected_path " + "is required for compare-and-swap retirement (fail closed)" + ) + if not checkpoint_id and not (session_id or "").strip(): + raise ControlPlaneError( + "cannot retire session checkpoint worktree_path: checkpoint_id " + "or session_id is required (fail closed)" + ) + with self._tx() as conn: if checkpoint_id: - conn.execute( - "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? WHERE checkpoint_id = ?", - (now_s, checkpoint_id), + selector_sql = "SELECT * FROM session_checkpoints WHERE checkpoint_id = ?" + selector_params: tuple[Any, ...] = (checkpoint_id,) + selector_desc = f"checkpoint_id '{checkpoint_id}'" + else: + selector_sql = "SELECT * FROM session_checkpoints WHERE session_id = ?" + selector_params = (session_id,) + selector_desc = f"session_id '{session_id}'" + + rows = [dict(r) for r in conn.execute(selector_sql, selector_params).fetchall()] + if not rows: + raise ControlPlaneError( + f"cannot retire session checkpoint worktree_path: no " + f"checkpoint matches {selector_desc} (fail closed)" ) - elif session_id: - conn.execute( - "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? WHERE session_id = ?", - (now_s, session_id), + if len(rows) > 1: + raise ControlPlaneError( + f"cannot retire session checkpoint worktree_path: " + f"{selector_desc} matches {len(rows)} checkpoints; supply an " + "exact checkpoint_id (fail closed)" ) + + record = rows[0] + target_checkpoint_id = record.get("checkpoint_id") + current_wt = (record.get("worktree_path") or "").strip() + + if not current_wt: + # Idempotent: the binding this caller audited is already gone. + return { + "session_id": session_id, + "checkpoint_id": target_checkpoint_id, + "retired": False, + "already_retired": True, + "prior_worktree_path": "", + "expected_worktree_path": expected_path, + "reason": reason, + "compare_and_swap": { + "matched": True, + "outcome": "already_retired", + }, + } + + if _realpath_or_raw(current_wt) != expected_norm: + raise ControlPlaneError( + f"cannot retire session checkpoint worktree_path for " + f"{selector_desc}: expected '{expected_path}' does not match " + f"current '{current_wt}' (fail closed)" + ) + + if expected_status is not None: + current_status = record.get("status") + if str(current_status or "").strip() != str(expected_status).strip(): + raise ControlPlaneError( + f"cannot retire session checkpoint worktree_path for " + f"{selector_desc}: status changed since audit (expected " + f"'{expected_status}', found '{current_status}'); fail closed" + ) + + cur = conn.execute( + "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? " + "WHERE checkpoint_id = ? AND worktree_path = ?", + (now_s, target_checkpoint_id, record.get("worktree_path")), + ) + if cur.rowcount != 1: + raise ControlPlaneError( + f"cannot retire session checkpoint worktree_path for " + f"{selector_desc}: compare-and-swap matched no row " + "(concurrent change); fail closed" + ) + return { "session_id": session_id, - "checkpoint_id": checkpoint_id, + "checkpoint_id": target_checkpoint_id, "retired": True, + "already_retired": False, + "prior_worktree_path": current_wt, + "expected_worktree_path": expected_path, "retired_at": now_s, "reason": reason, + "compare_and_swap": { + "matched": True, + "outcome": "retired", + "expected_status": expected_status, + }, } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index df6d1a6..4ffaff7 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -13597,6 +13597,7 @@ def gitea_audit_missing_worktree_bindings( @mcp.tool() def gitea_reconcile_missing_worktree_bindings( dry_run: bool = True, + # Deprecated: retained so callers that still pass it get an explicit deny. operator_authorized: bool = False, remote: str = "dadeschools", host: str | None = None, @@ -13606,12 +13607,22 @@ def gitea_reconcile_missing_worktree_bindings( """Audit and safely resolve (retire) missing worktree path bindings (#970). Identifies confirmed stale deleted worktree bindings, re-validates them - immediately before mutation to prevent recreation races, and retires only the - exact stale bindings while preserving unrelated worktrees and Git metadata. + against a fresh authoritative read immediately before mutation to prevent + recreation and ownership races, and retires only the exact stale bindings + while preserving unrelated worktrees and Git metadata. + + Apply mode (``dry_run=False``) is authorized **server-side** (#970 review + 644 B2): it requires the reconciler-only ``gitea.branch.delete`` + capability, the resolved ``reconcile_missing_worktree_bindings`` task + capability, and an active cleanup phase minted through + ``gitea_authorize_reconciliation_cleanup_phase``. A caller-supplied + ``operator_authorized`` is never authorization evidence (#709 F1 / + review 434) and is rejected outright. Dry-run stays available to any + ``gitea.read`` profile and never mutates. Args: dry_run: If True (default), reports planned mutations without modifying state. - operator_authorized: Explicit operator authorization required for apply mode. + operator_authorized: Rejected. Authorization is a server-side artifact. remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. org: Override the owner/organization. @@ -13620,6 +13631,22 @@ def gitea_reconcile_missing_worktree_bindings( Returns: dict with before/after audit state, resolutions, and dimension status. """ + import missing_worktree_reconcile + + # Explicitly reject the self-assertable Boolean before anything else, so it + # can never combine with a legitimate gate to authorize a mutation. + if operator_authorized: + return { + "success": False, + "performed": False, + "reason": "operator_authorized_rejected", + "reasons": [missing_worktree_reconcile.OPERATOR_AUTHORIZED_REJECTION], + "exact_next_action": ( + "authorize cleanup via gitea_authorize_reconciliation_cleanup_phase " + "from a reconciler profile, then re-run with dry_run=False" + ), + } + read_block = _profile_operation_gate("gitea.read") if read_block: return { @@ -13629,14 +13656,39 @@ def gitea_reconcile_missing_worktree_bindings( "permission_report": _permission_block_report("gitea.read"), } - import missing_worktree_reconcile db, db_errs = _control_plane_db_or_error() root = _canonical_local_git_root() h, o, r = _resolve(remote, host, org, repo) profile = get_profile() - role = profile.get("role") or "" - workflow_authorized = role in ("reconciler", "author", "controller") + role = (profile.get("role") or profile.get("role_kind") or "").strip().lower() + + cleanup_authorization = None + if not dry_run: + cleanup_task = missing_worktree_reconcile.CLEANUP_TASK + required_permission = task_capability_map.required_permission(cleanup_task) + capability_blockers = _profile_operation_gate(required_permission) + role_matches = role == task_capability_map.required_role(cleanup_task) + cleanup_authorization = missing_worktree_reconcile.assess_cleanup_authorization( + role=role, + capability_blockers=capability_blockers, + operator_authorized=False, + task_capability_resolved=(not capability_blockers) and role_matches, + ) + if not cleanup_authorization.get("authorized"): + return { + "success": False, + "performed": False, + "reason": "cleanup_authorization_required", + "reasons": cleanup_authorization.get("reasons") or [], + "cleanup_authorization": cleanup_authorization, + "permission_report": _permission_block_report(required_permission), + "exact_next_action": ( + "resolve the reconcile_missing_worktree_bindings capability " + "from a reconciler profile and authorize cleanup via " + "gitea_authorize_reconciliation_cleanup_phase" + ), + } return missing_worktree_reconcile.reconcile_missing_worktree_bindings( db, @@ -13646,8 +13698,8 @@ def gitea_reconcile_missing_worktree_bindings( repo=r, host=h, dry_run=dry_run, - operator_authorized=operator_authorized, - workflow_authorized=workflow_authorized, + operator_authorized=False, + cleanup_authorization=cleanup_authorization, ) diff --git a/missing_worktree_reconcile.py b/missing_worktree_reconcile.py index 782df56..7590632 100644 --- a/missing_worktree_reconcile.py +++ b/missing_worktree_reconcile.py @@ -38,19 +38,29 @@ from __future__ import annotations import json import os import subprocess +from datetime import datetime, timezone from typing import Any, Mapping, Sequence +import audit_reconciliation_mode import control_plane_db as cpd import issue_lock_store import lease_lifecycle +import restart_coordinator from merged_cleanup_reconcile import list_local_worktrees +# Single authority for "this session's heartbeat has gone stale". Imported +# rather than redefined so session liveness cannot drift between the restart +# coordinator and this workflow. +SESSION_HEARTBEAT_STALE_SECONDS = restart_coordinator.DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS + # Classification vocabulary (#970 AC1-AC4). CLASS_CONFIRMED_STALE_DELETED = "confirmed_stale_deleted_worktree" CLASS_MOVED_WORKTREE = "moved_worktree_path" CLASS_UNAVAILABLE_HOST_MOUNT = "unavailable_host_or_mount" CLASS_LIVE_LEASE_PROTECTED = "live_lease_protected" CLASS_LIVE_SESSION_PROTECTED = "live_session_protected" +CLASS_TRUSTED_OWNER_PROTECTED = "trusted_owner_evidence_protected" +CLASS_ISSUE_LOCK_PROTECTED = "conflicting_issue_lock_protected" CLASS_PRESENT_VALID = "present_valid_worktree" CLASS_ALREADY_RETIRED = "already_retired_binding" @@ -60,6 +70,8 @@ ALL_CLASSIFICATIONS = frozenset({ CLASS_UNAVAILABLE_HOST_MOUNT, CLASS_LIVE_LEASE_PROTECTED, CLASS_LIVE_SESSION_PROTECTED, + CLASS_TRUSTED_OWNER_PROTECTED, + CLASS_ISSUE_LOCK_PROTECTED, CLASS_PRESENT_VALID, CLASS_ALREADY_RETIRED, }) @@ -68,6 +80,40 @@ RETIRE_ELIGIBLE_CLASSES = frozenset({ CLASS_CONFIRMED_STALE_DELETED, }) +# A lease stops protecting its binding only once it reaches a terminal status. +# Anything else — including an unrecognised status — keeps the binding +# protected, so a dead recorded PID can never on its own retire a lease the +# control plane still considers live (#970 review 644 B4). +TERMINAL_LEASE_STATUSES = frozenset({ + lease_lifecycle.LEASE_STATUS_RELEASED, + lease_lifecycle.LEASE_STATUS_EXPIRED, + lease_lifecycle.LEASE_STATUS_ABANDONED, +}) + +TERMINAL_SESSION_STATUSES = frozenset({ + "exited", "dead", "released", "terminated", "closed", "completed", +}) + +# Server-enforced cleanup authorization (#970 review 644 B2). +# +# Apply mode is authorized by the project's existing reconciliation cleanup +# gate — ``audit_reconciliation_mode.authorize_cleanup_phase``, minted through +# ``gitea_authorize_reconciliation_cleanup_phase``, which itself requires the +# reconciler-only ``gitea.branch.delete`` capability, safety proof, and a +# before/after snapshot. A caller-supplied ``operator_authorized`` boolean is +# self-assertable and is therefore never authorization evidence +# (#709 F1 / review 434). +CLEANUP_AUTHORIZATION_SOURCE = "audit_reconciliation_mode.authorize_cleanup_phase" +CLEANUP_REQUIRED_ROLE = "reconciler" +CLEANUP_REQUIRED_PERMISSION = "gitea.branch.delete" +CLEANUP_TASK = "reconcile_missing_worktree_bindings" + +OPERATOR_AUTHORIZED_REJECTION = ( + "operator_authorized is not accepted as authorization evidence " + "(#709 F1 / review 434); authorize cleanup server-side via " + "gitea_authorize_reconciliation_cleanup_phase" +) + def _norm_path(p: str | None) -> str: if not p or not str(p).strip(): @@ -167,6 +213,156 @@ def find_moved_worktree_path( return None +def _heartbeat_age_seconds(value: str | None, *, now: datetime | None = None) -> float | None: + """Seconds since *value*, or None when it cannot be parsed.""" + text = (value or "").strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + moment = now or datetime.now(timezone.utc) + return (moment - parsed).total_seconds() + + +def collect_session_evidence(db: cpd.ControlPlaneDB) -> dict[str, dict[str, Any]]: + """Return live-session evidence per session id (#970 review 644 B4). + + ``session_active`` was previously never derived, so every binding was + classified as if no session owned it. The control-plane ``sessions`` table + is the authoritative record of which process claimed a worktree, so it is + read here and reported along two distinct axes: + + * ``session_live`` — the project's existing liveness rule, reused verbatim + from ``restart_coordinator._classify_session``: recorded ``active``, the + PID not known-dead, and the heartbeat not stale. A live session is real + ownership and blocks cleanup outright. + * ``session_recorded_active`` — the row still says ``active`` but its + heartbeat has gone stale. That is weaker, still-trustworthy ownership + evidence: it blocks cleanup unless a terminal lease contradicts it. + + A merely-alive PID never establishes liveness on its own (#790 AC-N2): the + recorded PID is the long-lived MCP daemon, so it proves the daemon is up and + nothing about the task. It is carried as corroboration only. + """ + evidence: dict[str, dict[str, Any]] = {} + try: + rows = db.list_sessions(limit=500) + except Exception: # noqa: BLE001 + return evidence + + now = datetime.now(timezone.utc) + for row in rows: + sid = str(row.get("session_id") or "").strip() + if not sid: + continue + status = str(row.get("status") or "").strip().lower() + pid = row.get("pid") + pid_alive = lease_lifecycle.is_process_alive(pid) if pid else None + recorded_active = bool(status) and status not in TERMINAL_SESSION_STATUSES + age = _heartbeat_age_seconds(row.get("last_heartbeat_at"), now=now) + heartbeat_stale = bool( + age is not None and age > SESSION_HEARTBEAT_STALE_SECONDS + ) + session_live = bool( + recorded_active and pid_alive is not False and not heartbeat_stale + ) + evidence[sid] = { + "session_id": sid, + "session_status": status, + "session_pid": pid, + "session_pid_alive": bool(pid_alive), + "session_live": session_live, + "session_recorded_active": recorded_active, + "session_active": session_live or recorded_active, + "heartbeat_stale": heartbeat_stale, + "heartbeat_age_seconds": age, + "last_heartbeat_at": row.get("last_heartbeat_at"), + "role": row.get("role"), + "profile": row.get("profile"), + } + return evidence + + +def collect_issue_lock_bindings(lock_dir: str | None = None) -> list[dict[str, Any]]: + """Return durable author issue-lock bindings (#970 review 644 B4). + + ``issue_lock_store`` was imported but never used, so a worktree still owned + by a live author issue lock could be classified stale and retired. Every + durable lock file is read here and reported with its recorded path, branch, + owner pid, and lease liveness, so a conflicting lock can block cleanup. + """ + bindings: list[dict[str, Any]] = [] + try: + paths = issue_lock_store.iter_lock_files(lock_dir) + except Exception: # noqa: BLE001 + return bindings + + for lock_path in paths: + try: + lock = issue_lock_store.read_lock_file(lock_path) + except Exception: # noqa: BLE001 + continue + if not isinstance(lock, dict): + continue + wt = str(lock.get("worktree_path") or "").strip() + if not wt: + continue + pid = lock.get("session_pid") or lock.get("pid") + try: + pid_alive = issue_lock_store.is_process_alive(pid) + except Exception: # noqa: BLE001 + pid_alive = False + try: + lease_live = issue_lock_store.is_lease_live(lock) + except Exception: # noqa: BLE001 + lease_live = False + bindings.append({ + "source": "issue_lock", + "lock_file_path": lock_path, + "issue_number": lock.get("issue_number"), + "branch": lock.get("branch_name") or lock.get("branch"), + "worktree_path": wt, + "normalized_path": _norm_path(wt), + "owner_pid": pid, + "owner_pid_alive": pid_alive, + "lease_live": lease_live, + "claimant": (lock.get("claimant") or {}).get("username") + if isinstance(lock.get("claimant"), Mapping) + else lock.get("claimant"), + # A lock is trustworthy ownership evidence while its lease is live + # or its recorded process is still running. + "conflicting": bool(lease_live or pid_alive), + }) + return bindings + + +def find_conflicting_issue_lock( + recorded_path: str, + issue_locks: Sequence[Mapping[str, Any]] | None, + *, + branch: str | None = None, +) -> dict[str, Any] | None: + """Return a live issue lock that still binds *recorded_path* (or *branch*).""" + if not issue_locks: + return None + norm_p = _norm_path(recorded_path) + for lock in issue_locks: + if not lock.get("conflicting"): + continue + if norm_p and lock.get("normalized_path") == norm_p: + return dict(lock) + lock_branch = str(lock.get("branch") or "").strip() + if branch and lock_branch and lock_branch == str(branch).strip(): + return dict(lock) + return None + + def classify_worktree_binding( *, recorded_path: str, @@ -175,6 +371,10 @@ def classify_worktree_binding( owner_pid: int | None = None, session_id: str | None = None, session_active: bool = False, + session_live: bool = False, + session_pid_alive: bool = False, + session_status: str | None = None, + issue_lock: Mapping[str, Any] | None = None, branch: str | None = None, project_root: str, host_health: dict[str, Any] | None = None, @@ -242,43 +442,111 @@ def classify_worktree_binding( "reasons": reasons, } - # Check 4: Live lease protection + # Check 4: Conflicting durable issue lock. + # + # Checked before the lease, because an author issue lock can still own a + # worktree after its control-plane lease row has gone terminal. + if issue_lock: + reasons.append( + f"durable issue lock for issue #{issue_lock.get('issue_number')} " + f"(branch '{issue_lock.get('branch')}', owner pid " + f"{issue_lock.get('owner_pid')}, lease_live=" + f"{issue_lock.get('lease_live')}, pid_alive=" + f"{issue_lock.get('owner_pid_alive')}) still binds this worktree; " + "a conflicting issue lock blocks cleanup" + ) + return { + "classification": CLASS_ISSUE_LOCK_PROTECTED, + "retire_eligible": False, + "recorded_path": norm_p, + "issue_lock": dict(issue_lock), + "reasons": reasons, + } + + # Check 5: Lease protection. + # + # A non-terminal lease protects its binding regardless of whether the + # recorded PID is still alive: the recorded PID is the long-lived MCP + # daemon in normal operation, so its absence is weak evidence, while the + # lease status is the control plane's own record of ownership. Dead-PID + # evidence alone must never retire a lease the control plane still holds + # (#970 review 644 B4). st = (lease_status or "").strip().lower() pid_alive = lease_lifecycle.is_process_alive(owner_pid) if owner_pid else False - if st == "active" and (pid_alive or owner_pid is None): + if st and st not in TERMINAL_LEASE_STATUSES: reasons.append( - f"lease is status='active' (owner_pid={owner_pid}, alive={pid_alive}); " - "active lease protects worktree binding" + f"lease status '{st}' is not terminal (owner_pid={owner_pid}, " + f"alive={pid_alive}); a non-terminal lease protects its worktree " + "binding" ) return { "classification": CLASS_LIVE_LEASE_PROTECTED, "retire_eligible": False, "recorded_path": norm_p, + "lease_status": st, + "owner_pid_alive": pid_alive, "reasons": reasons, } - # Check 5: Live session / process protection - if session_active and pid_alive: + # Check 6: Live session protection. + # + # "Live" is the project's existing rule (recorded active, PID not dead, + # heartbeat fresh) — not bare PID liveness, which is only the daemon. + if session_live: reasons.append( - f"session '{session_id}' is active with live process PID {owner_pid}; " - "live session protects worktree binding" + f"session '{session_id}' is live (session_status='{session_status}', " + f"heartbeat fresh, session_pid_alive={session_pid_alive}); a live " + "session protects its worktree binding" ) return { "classification": CLASS_LIVE_SESSION_PROTECTED, "retire_eligible": False, "recorded_path": norm_p, + "session_status": session_status, + "reasons": reasons, + } + + # Check 7: Trustworthy ownership evidence without a live process. + # + # The owning process may be gone while the control plane still records the + # session as active. That is a valid ownership signal and must block + # cleanup, even though nothing is running. + # + # One signal outranks it: an explicitly terminal lease. Session rows are + # opened when work is assigned and are not closed on release, so a + # recorded-active session routinely outlives the work it owned. A lease + # that was explicitly released, expired, or abandoned is the later and more + # specific statement about *this* binding, so it wins. Without such a lease + # — a checkpoint binding, or a lease row that is simply gone — the session + # record is the best ownership evidence available and blocks cleanup. + lease_is_terminal = bool(st) and st in TERMINAL_LEASE_STATUSES + if session_active and not lease_is_terminal: + reasons.append( + f"session '{session_id}' is still recorded active " + f"(session_status='{session_status}') even though its process is " + "not running; trustworthy ownership evidence blocks cleanup" + ) + return { + "classification": CLASS_TRUSTED_OWNER_PROTECTED, + "retire_eligible": False, + "recorded_path": norm_p, + "session_status": session_status, "reasons": reasons, } # Confirmed stale deleted reasons.append( f"worktree path '{norm_p}' does not exist on disk, host is healthy, " - "no moved path was found, and no active lease or live process references it" + "no moved path was found, the lease status is terminal " + f"('{st or 'none recorded'}'), no live or recorded-active session owns " + "it, and no durable issue lock binds it" ) return { "classification": CLASS_CONFIRMED_STALE_DELETED, "retire_eligible": True, "recorded_path": norm_p, + "lease_status": st, + "owner_pid_alive": pid_alive, "reasons": reasons, } @@ -300,6 +568,11 @@ def audit_missing_worktree_bindings( hh = check_host_mount_health(root) cp_db = db or cpd.ControlPlaneDB() + # Ownership evidence the classifier needs. Populated once per audit and + # re-read live before any mutation (#970 review 644 B1/B4). + session_evidence = collect_session_evidence(cp_db) + issue_locks = collect_issue_lock_bindings() + bindings: list[dict[str, Any]] = [] # 1. Audit control-plane leases @@ -322,12 +595,20 @@ def audit_missing_worktree_bindings( if not br and isinstance(L.get("provenance"), dict): br = L["provenance"].get("branch") + sess = session_evidence.get(str(sess_id or "").strip(), {}) + conflicting_lock = find_conflicting_issue_lock(wt, issue_locks, branch=br) + cls_info = classify_worktree_binding( recorded_path=wt, lease_status=L.get("status"), lease_phase=L.get("phase"), owner_pid=owner_pid, session_id=sess_id, + session_active=bool(sess.get("session_active")), + session_live=bool(sess.get("session_live")), + session_pid_alive=bool(sess.get("session_pid_alive")), + session_status=sess.get("session_status"), + issue_lock=conflicting_lock, branch=br, project_root=root, host_health=hh, @@ -344,11 +625,25 @@ def audit_missing_worktree_bindings( "lease_status": L.get("status"), "lease_phase": L.get("phase"), "owner_pid": owner_pid, + "session_status": sess.get("session_status"), + "session_active": bool(sess.get("session_active")), + "session_pid_alive": bool(sess.get("session_pid_alive")), + "issue_lock_conflict": conflicting_lock, "remote": L.get("remote") or remote, "org": L.get("org") or org, "repo": L.get("repo") or repo, "host": L.get("host") or host, **cls_info, + "audited_state": _safety_snapshot( + source="lease", + worktree_path=wt, + lease_id=L.get("lease_id"), + lease_status=L.get("status"), + lease_session_id=sess_id, + lease_owner_pid=owner_pid, + session_evidence=sess, + issue_lock=conflicting_lock, + ), }) except Exception as exc: # noqa: BLE001 bindings.append({ @@ -364,9 +659,18 @@ def audit_missing_worktree_bindings( if not wt: continue sess_id = cp.get("session_id") + sess = session_evidence.get(str(sess_id or "").strip(), {}) + conflicting_lock = find_conflicting_issue_lock( + wt, issue_locks, branch=cp.get("branch") + ) cls_info = classify_worktree_binding( recorded_path=wt, session_id=sess_id, + session_active=bool(sess.get("session_active")), + session_live=bool(sess.get("session_live")), + session_pid_alive=bool(sess.get("session_pid_alive")), + session_status=sess.get("session_status"), + issue_lock=conflicting_lock, branch=cp.get("branch"), project_root=root, host_health=hh, @@ -380,14 +684,64 @@ def audit_missing_worktree_bindings( "branch": cp.get("branch"), "worktree_path": wt, "status": cp.get("status"), + "checkpoint_status": cp.get("status"), + "session_status": sess.get("session_status"), + "session_active": bool(sess.get("session_active")), + "session_pid_alive": bool(sess.get("session_pid_alive")), + "issue_lock_conflict": conflicting_lock, "remote": cp.get("remote") or remote, "org": cp.get("org") or org, "repo": cp.get("repo") or repo, **cls_info, + "audited_state": _safety_snapshot( + source="session_checkpoint", + worktree_path=wt, + checkpoint_id=cp.get("checkpoint_id"), + checkpoint_status=cp.get("status"), + lease_session_id=sess_id, + session_evidence=sess, + issue_lock=conflicting_lock, + ), }) except Exception: # noqa: BLE001 pass + # 3. Report durable issue-lock bindings whose paths are missing. + # + # These are reported, never retired: a durable author lock is released + # through its own sanctioned lifecycle. They are kept out of + # ``missing_bindings`` for exactly that reason — counting a binding this + # workflow must not touch would leave the worktrees dimension permanently + # unresolvable — but they are still surfaced so a missing path a lock + # claims is visible rather than silently ignored (#970 AC1). + issue_lock_missing: list[dict[str, Any]] = [] + for lock in issue_locks: + wt = lock.get("worktree_path") or "" + if not wt or os.path.exists(_norm_path(wt)): + continue + issue_lock_missing.append({ + "source": "issue_lock", + "lock_file_path": lock.get("lock_file_path"), + "issue_number": lock.get("issue_number"), + "branch": lock.get("branch"), + "worktree_path": wt, + "recorded_path": _norm_path(wt), + "owner_pid": lock.get("owner_pid"), + "owner_pid_alive": lock.get("owner_pid_alive"), + "lease_live": lock.get("lease_live"), + "claimant": lock.get("claimant"), + "conflicting": lock.get("conflicting"), + "classification": CLASS_ISSUE_LOCK_PROTECTED + if lock.get("conflicting") + else CLASS_ALREADY_RETIRED, + "retire_eligible": False, + "reasons": [ + "durable issue lock records a worktree path that is missing on " + "disk; release it through the sanctioned issue-lock lifecycle, " + "never by retiring the binding here" + ], + }) + missing_bindings = [ b for b in bindings if b.get("classification") != CLASS_PRESENT_VALID ] @@ -398,7 +752,15 @@ def audit_missing_worktree_bindings( b for b in missing_bindings if b.get("classification") == CLASS_MOVED_WORKTREE ] live_protected = [ - b for b in missing_bindings if b.get("classification") in (CLASS_LIVE_LEASE_PROTECTED, CLASS_LIVE_SESSION_PROTECTED) + b for b in missing_bindings + if b.get("classification") in ( + CLASS_LIVE_LEASE_PROTECTED, + CLASS_LIVE_SESSION_PROTECTED, + CLASS_TRUSTED_OWNER_PROTECTED, + ) + ] + issue_lock_protected = [ + b for b in missing_bindings if b.get("classification") == CLASS_ISSUE_LOCK_PROTECTED ] unavailable_host = [ b for b in missing_bindings if b.get("classification") == CLASS_UNAVAILABLE_HOST_MOUNT @@ -415,24 +777,314 @@ def audit_missing_worktree_bindings( "moved_count": len(moved), "live_protected_count": len(live_protected), "unavailable_host_count": len(unavailable_host), + "issue_lock_protected_count": len(issue_lock_protected), "missing_bindings": missing_bindings, "confirmed_stale_candidates": confirmed_stale, + "issue_lock_missing_bindings": issue_lock_missing, + "issue_lock_missing_count": len(issue_lock_missing), + "live_session_evidence_count": sum( + 1 for e in session_evidence.values() if e.get("session_active") + ), "worktrees_dimension_resolved": len(missing_bindings) == 0, } +def _safety_snapshot( + *, + source: str, + worktree_path: str | None, + lease_id: str | None = None, + lease_status: str | None = None, + lease_session_id: str | None = None, + lease_owner_pid: Any = None, + checkpoint_id: str | None = None, + checkpoint_status: str | None = None, + session_evidence: Mapping[str, Any] | None = None, + issue_lock: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Normalize the safety-relevant fields cleanup must re-verify (#970 B1). + + Every value cleanup relies on is captured here in one comparable shape, so + the apply path can diff a fresh authoritative read against what the audit + saw instead of trusting the audit's own snapshot. + """ + sess = dict(session_evidence or {}) + lock = dict(issue_lock or {}) + return { + "source": source, + "binding_path": _norm_path(worktree_path), + "binding_identity": f"{source}:{lease_id or checkpoint_id or lease_session_id or ''}", + "lease_id": lease_id, + "lease_status": (str(lease_status).strip().lower() if lease_status else None), + "lease_session_id": lease_session_id, + "lease_owner_pid": (int(lease_owner_pid) if isinstance(lease_owner_pid, int) else lease_owner_pid), + "checkpoint_id": checkpoint_id, + "checkpoint_status": (str(checkpoint_status).strip().lower() if checkpoint_status else None), + "checkpoint_path": _norm_path(worktree_path) if source == "session_checkpoint" else None, + "session_status": sess.get("session_status"), + "session_active": bool(sess.get("session_active")), + "session_live": bool(sess.get("session_live")), + "session_pid_alive": bool(sess.get("session_pid_alive")), + "issue_lock_conflicting": bool(lock.get("conflicting")), + "issue_lock_file": lock.get("lock_file_path"), + "issue_lock_issue_number": lock.get("issue_number"), + } + + +# Fields whose drift between audit and apply must fail closed without mutation. +SAFETY_SNAPSHOT_FIELDS = ( + "binding_path", + "binding_identity", + "lease_id", + "lease_status", + "lease_session_id", + "lease_owner_pid", + "checkpoint_id", + "checkpoint_status", + "checkpoint_path", + "session_status", + "session_active", + "session_live", + "session_pid_alive", + "issue_lock_conflicting", + "issue_lock_file", + "issue_lock_issue_number", +) + + +def read_live_safety_state( + db: cpd.ControlPlaneDB, + binding: Mapping[str, Any], +) -> dict[str, Any]: + """Re-read the authoritative safety state for one binding (#970 B1). + + Reads the lease row, session row, checkpoint row, and durable issue locks + from their authoritative stores — never from the audit snapshot — so + cleanup decides against current state. + """ + source = binding.get("source") or "lease" + lease_id = binding.get("lease_id") + checkpoint_id = binding.get("checkpoint_id") + session_id = binding.get("session_id") + + lease_row: dict[str, Any] = {} + if lease_id: + try: + state = db.get_lease_workflow_state(str(lease_id)) + except Exception: # noqa: BLE001 + state = None + if isinstance(state, Mapping): + lease_row = dict(state.get("lease") or {}) + + checkpoint_row: dict[str, Any] = {} + if source == "session_checkpoint": + try: + rows = db.list_session_checkpoints( + session_id=str(session_id) if session_id else None, + limit=500, + ) + except Exception: # noqa: BLE001 + rows = [] + for row in rows or []: + if checkpoint_id and row.get("checkpoint_id") != checkpoint_id: + continue + checkpoint_row = dict(row) + break + + session_evidence = collect_session_evidence(db) + live_session = session_evidence.get(str(session_id or "").strip(), {}) + + issue_locks = collect_issue_lock_bindings() + live_path = ( + lease_row.get("worktree_path") + if source == "lease" + else checkpoint_row.get("worktree_path") + ) + if live_path is None: + live_path = binding.get("worktree_path") + conflicting_lock = find_conflicting_issue_lock( + str(live_path or ""), issue_locks, branch=binding.get("branch") + ) + + snapshot = _safety_snapshot( + source=source, + worktree_path=str(live_path or ""), + lease_id=lease_id, + lease_status=lease_row.get("status") if lease_row else binding.get("lease_status"), + lease_session_id=lease_row.get("session_id") if lease_row else session_id, + lease_owner_pid=lease_row.get("owner_pid") if lease_row else binding.get("owner_pid"), + checkpoint_id=checkpoint_row.get("checkpoint_id") if checkpoint_row else checkpoint_id, + checkpoint_status=checkpoint_row.get("status") if checkpoint_row else binding.get("checkpoint_status"), + session_evidence=live_session, + issue_lock=conflicting_lock, + ) + return { + "snapshot": snapshot, + "lease_row": lease_row, + "checkpoint_row": checkpoint_row, + "session_evidence": live_session, + "issue_lock": conflicting_lock, + "live_worktree_path": str(live_path or ""), + } + + +def compare_safety_state( + audited: Mapping[str, Any] | None, + live: Mapping[str, Any], +) -> dict[str, Any]: + """Diff an audited safety snapshot against a fresh one (#970 B1).""" + if not audited: + return { + "comparable": False, + "changed": True, + "changed_fields": ["audited_state"], + "reasons": [ + "binding carries no audited safety snapshot; cleanup cannot " + "prove nothing changed since the audit (fail closed)" + ], + "differences": {}, + } + + differences: dict[str, Any] = {} + for field in SAFETY_SNAPSHOT_FIELDS: + before = audited.get(field) + after = live.get(field) + if before != after: + differences[field] = {"audited": before, "live": after} + + reasons = [ + f"safety-relevant value '{field}' changed between audit and apply " + f"(audited={diff['audited']!r}, live={diff['live']!r})" + for field, diff in differences.items() + ] + return { + "comparable": True, + "changed": bool(differences), + "changed_fields": sorted(differences), + "differences": differences, + "reasons": reasons, + } + + +def assess_cleanup_authorization( + *, + role: str | None = None, + capability_blockers: Sequence[str] | None = None, + operator_authorized: bool = False, + task_capability_resolved: bool = False, +) -> dict[str, Any]: + """Server-enforced authorization for apply-mode cleanup (#970 B2). + + Authorization is never derived from a caller-supplied boolean. It requires, + together: + + * the reconciler role — the only role the project grants cleanup authority; + * proven ``gitea.branch.delete`` capability, evaluated server-side by the + profile gate and passed in as its blocker list; + * a resolved cleanup task capability (task-capability resolution boundary); + * an active, authorized cleanup phase minted through + ``gitea_authorize_reconciliation_cleanup_phase`` (mutation boundary). + + An unauthorized or wrongly-profiled caller is rejected even when it submits + ``operator_authorized=True``. + """ + reasons: list[str] = [] + if operator_authorized: + reasons.append(OPERATOR_AUTHORIZED_REJECTION) + + normalized_role = (role or "").strip().lower() + if normalized_role != CLEANUP_REQUIRED_ROLE: + reasons.append( + f"cleanup requires role '{CLEANUP_REQUIRED_ROLE}'; active role is " + f"'{normalized_role or 'unknown'}' (fail closed)" + ) + + blockers = list(capability_blockers or []) + if blockers: + reasons.append( + f"active profile lacks {CLEANUP_REQUIRED_PERMISSION}: " + + "; ".join(str(b) for b in blockers) + ) + + if not task_capability_resolved: + reasons.append( + f"task capability '{CLEANUP_TASK}' is not resolved for this session; " + "resolve it before apply-mode cleanup (fail closed)" + ) + + phase_allowed, phase_reasons = audit_reconciliation_mode.check_cleanup_execution_allowed() + if not phase_allowed: + reasons.extend(phase_reasons) + + return { + "authorized": not reasons, + "source": CLEANUP_AUTHORIZATION_SOURCE, + "required_role": CLEANUP_REQUIRED_ROLE, + "required_permission": CLEANUP_REQUIRED_PERMISSION, + "required_task": CLEANUP_TASK, + "role": normalized_role or None, + "phase": audit_reconciliation_mode.current_phase(), + "operator_authorized_rejected": bool(operator_authorized), + "reasons": reasons, + } + + +def _authorization_valid_for_apply( + cleanup_authorization: Mapping[str, Any] | None, +) -> tuple[bool, list[str]]: + """Validate a minted authorization at the mutation boundary (#970 B2). + + The authorization artifact alone is not trusted: the process-local cleanup + phase is re-checked here, so a forged mapping cannot authorize a mutation + that the server never approved. + """ + reasons: list[str] = [] + if not isinstance(cleanup_authorization, Mapping): + reasons.append( + "apply mode requires a server-minted cleanup authorization from " + f"{CLEANUP_AUTHORIZATION_SOURCE} (fail closed)" + ) + return False, reasons + if not cleanup_authorization.get("authorized"): + reasons.extend( + list(cleanup_authorization.get("reasons") or []) + or ["cleanup authorization is not authorized (fail closed)"] + ) + if cleanup_authorization.get("source") != CLEANUP_AUTHORIZATION_SOURCE: + reasons.append( + "cleanup authorization did not come from " + f"{CLEANUP_AUTHORIZATION_SOURCE} (fail closed)" + ) + if (cleanup_authorization.get("role") or "").strip().lower() != CLEANUP_REQUIRED_ROLE: + reasons.append( + f"cleanup authorization role must be '{CLEANUP_REQUIRED_ROLE}' " + "(fail closed)" + ) + # Re-check the server-side phase at the mutation boundary itself. + phase_allowed, phase_reasons = audit_reconciliation_mode.check_cleanup_execution_allowed() + if not phase_allowed: + reasons.extend(phase_reasons) + return (not reasons), reasons + + def resolve_missing_worktree_binding( db: cpd.ControlPlaneDB | None = None, *, binding: dict[str, Any], dry_run: bool = True, operator_authorized: bool = False, - workflow_authorized: bool = False, + cleanup_authorization: Mapping[str, Any] | None = None, project_root: str | None = None, ) -> dict[str, Any]: """Safely resolve (retire) one confirmed stale missing worktree binding (#970). - Pre-mutation re-validation runs immediately before mutation to prevent races. + Apply mode requires a server-minted cleanup authorization; a client-supplied + ``operator_authorized`` is rejected outright (B2). Immediately before + mutating, the authoritative lease, session, checkpoint, and issue-lock state + is re-read and diffed against what the audit saw, and the binding is + reclassified from those live values — so a lease that was re-activated, an + owner that changed, or a path that moved after the audit fails closed + without mutation (B1/B4). """ root = _norm_path(project_root or os.getcwd()) cp_db = db or cpd.ControlPlaneDB() @@ -446,23 +1098,63 @@ def resolve_missing_worktree_binding( "message": "worktree_path is already empty (no action needed)", } - # 1. Authorization check - authorized = operator_authorized or workflow_authorized - if not authorized and not dry_run: + # 1. Authorization (apply mode only; dry-run is always non-mutating). + if operator_authorized: return { "success": False, "performed": False, - "reason": "authorization_required", - "message": "Mutation refused: operator_authorized or workflow_authorized is required to retire a stale binding", + "reason": "operator_authorized_rejected", + "reasons": [OPERATOR_AUTHORIZED_REJECTION], + "message": OPERATOR_AUTHORIZED_REJECTION, } - # 2. Immediate pre-mutation re-validation + if not dry_run: + auth_ok, auth_reasons = _authorization_valid_for_apply(cleanup_authorization) + if not auth_ok: + return { + "success": False, + "performed": False, + "reason": "authorization_required", + "reasons": auth_reasons, + "cleanup_authorization": dict(cleanup_authorization or {}), + "message": ( + "Mutation refused: server-enforced cleanup authorization is " + "required to retire a stale binding" + ), + } + + # 2. Fresh authoritative read of every safety-relevant record (B1). + live_state = read_live_safety_state(cp_db, binding) + live_snapshot = live_state["snapshot"] + drift = compare_safety_state(binding.get("audited_state"), live_snapshot) + if drift["changed"]: + return { + "success": False, + "performed": False, + "reason": "safety_state_changed", + "reasons": drift["reasons"], + "changed_fields": drift["changed_fields"], + "audited_state": dict(binding.get("audited_state") or {}), + "live_state": live_snapshot, + "differences": drift["differences"], + "message": ( + "Pre-mutation revalidation refused: safety-relevant state " + f"changed since the audit ({', '.join(drift['changed_fields'])})" + ), + } + + # 3. Reclassify from the live values, not the audit snapshot (B1/B4). reval = classify_worktree_binding( - recorded_path=rec_path, - lease_status=binding.get("lease_status"), + recorded_path=live_state["live_worktree_path"] or rec_path, + lease_status=live_snapshot.get("lease_status"), lease_phase=binding.get("lease_phase"), - owner_pid=binding.get("owner_pid"), - session_id=binding.get("session_id"), + owner_pid=live_snapshot.get("lease_owner_pid"), + session_id=live_snapshot.get("lease_session_id"), + session_active=bool(live_snapshot.get("session_active")), + session_live=bool(live_snapshot.get("session_live")), + session_pid_alive=bool(live_snapshot.get("session_pid_alive")), + session_status=live_snapshot.get("session_status"), + issue_lock=live_state.get("issue_lock"), branch=binding.get("branch"), project_root=root, ) @@ -473,13 +1165,14 @@ def resolve_missing_worktree_binding( "performed": False, "reason": "revalidation_failed", "revalidation": reval, + "live_state": live_snapshot, "message": f"Pre-mutation re-validation failed: binding classified as '{reval.get('classification')}' instead of '{CLASS_CONFIRMED_STALE_DELETED}'", } source = binding.get("source", "lease") lease_id = binding.get("lease_id") session_id = binding.get("session_id") - checkpoint_id = binding.get("checkpoint_id") + checkpoint_id = live_snapshot.get("checkpoint_id") or binding.get("checkpoint_id") planned_action = { "source": source, @@ -497,10 +1190,11 @@ def resolve_missing_worktree_binding( "dry_run": True, "action": "retire_stale_binding", "planned_action": planned_action, + "live_state": live_snapshot, "message": f"Dry-run: would retire stale missing worktree_path '{rec_path}' for {source} (lease={lease_id})", } - # 3. Apply mode: execute targeted retirement + # 4. Apply mode: targeted, compare-and-swapped retirement. audit_proof: dict[str, Any] = { "retired_worktree_path": rec_path, "source": source, @@ -508,31 +1202,53 @@ def resolve_missing_worktree_binding( "session_id": session_id, "checkpoint_id": checkpoint_id, "classification": CLASS_CONFIRMED_STALE_DELETED, + "authorization_source": CLEANUP_AUTHORIZATION_SOURCE, + "revalidated_against": live_snapshot, } - if source == "lease" and lease_id: - ret_proof = cp_db.retire_lease_worktree_path( - lease_id, - expected_path=rec_path, - reason="missing_worktree_path_retired_by_reconciler", - ) - audit_proof["db_lease_update"] = ret_proof - - if source == "session_checkpoint" and (checkpoint_id or session_id): - ret_cp = cp_db.retire_session_checkpoint_worktree_path( - session_id=session_id or "", - checkpoint_id=checkpoint_id, - expected_path=rec_path, - reason="missing_worktree_path_retired_by_reconciler", - ) - audit_proof["db_checkpoint_update"] = ret_cp - - # Also clean up any matching lease row if checkpoint was supplied or vice versa - if lease_id and source != "lease": - try: - cp_db.retire_lease_worktree_path(lease_id, expected_path=rec_path) - except Exception: # noqa: BLE001 - pass + try: + if source == "lease" and lease_id: + audit_proof["db_lease_update"] = cp_db.retire_lease_worktree_path( + lease_id, + expected_path=rec_path, + expected_status=live_snapshot.get("lease_status"), + expected_session_id=live_snapshot.get("lease_session_id"), + expected_owner_pid=live_snapshot.get("lease_owner_pid"), + reason="missing_worktree_path_retired_by_reconciler", + ) + elif source == "session_checkpoint" and (checkpoint_id or session_id): + audit_proof["db_checkpoint_update"] = ( + cp_db.retire_session_checkpoint_worktree_path( + session_id=session_id or "", + checkpoint_id=checkpoint_id, + expected_path=rec_path, + expected_status=live_snapshot.get("checkpoint_status"), + reason="missing_worktree_path_retired_by_reconciler", + ) + ) + else: + return { + "success": False, + "performed": False, + "reason": "unsupported_binding_source", + "message": ( + f"binding source '{source}' has no sanctioned retirement " + "path; refusing to mutate" + ), + } + except cpd.ControlPlaneError as exc: + # The compare-and-swap lost a race. Nothing was mutated. + return { + "success": False, + "performed": False, + "reason": "compare_and_swap_failed", + "reasons": [str(exc)], + "live_state": live_snapshot, + "message": ( + "Retirement refused by compare-and-swap: the stored binding " + "changed concurrently" + ), + } return { "success": True, @@ -554,13 +1270,15 @@ def reconcile_missing_worktree_bindings( host: str | None = None, dry_run: bool = True, operator_authorized: bool = False, - workflow_authorized: bool = False, + cleanup_authorization: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Audit and safely reconcile missing worktree path bindings (#970). Audits all recorded bindings, identifies confirmed stale missing paths, - retires eligible bindings (when dry_run=False and authorized), and returns - durable before/after audit evidence. + retires eligible bindings (when ``dry_run=False`` and a server-minted + ``cleanup_authorization`` proves authority), and returns durable + before/after audit evidence. A caller-supplied ``operator_authorized`` is + rejected, never honoured (#970 review 644 B2). """ cp_db = db or cpd.ControlPlaneDB() root = _norm_path(project_root or os.getcwd()) @@ -583,7 +1301,7 @@ def reconcile_missing_worktree_bindings( binding=candidate, dry_run=dry_run, operator_authorized=operator_authorized, - workflow_authorized=workflow_authorized, + cleanup_authorization=cleanup_authorization, project_root=root, ) resolutions.append(res) @@ -597,13 +1315,15 @@ def reconcile_missing_worktree_bindings( host=host, ) + refusals = [r for r in resolutions if not r.get("success")] return { "success": True, "dry_run": dry_run, - "operator_authorized": operator_authorized, - "workflow_authorized": workflow_authorized, + "operator_authorized_rejected": bool(operator_authorized), + "cleanup_authorization": dict(cleanup_authorization or {}), "before_audit": initial_audit, "resolutions": resolutions, + "refused_count": len(refusals), "after_audit": post_audit, "worktrees_dimension_resolved": post_audit.get("worktrees_dimension_resolved", False), "stale_candidates_count": len(candidates), diff --git a/task_capability_map.py b/task_capability_map.py index 393ce98..3766fb9 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -386,6 +386,25 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.delete", "role": "reconciler", }, + # #970: auditing missing worktree bindings is read-only; retiring one is a + # control-plane cleanup mutation and carries the same reconciler-only + # authority as any other reconciliation cleanup (review 644 B2). + "audit_missing_worktree_bindings": { + "permission": "gitea.read", + "role": "reconciler", + }, + "gitea_audit_missing_worktree_bindings": { + "permission": "gitea.read", + "role": "reconciler", + }, + "reconcile_missing_worktree_bindings": { + "permission": "gitea.branch.delete", + "role": "reconciler", + }, + "gitea_reconcile_missing_worktree_bindings": { + "permission": "gitea.branch.delete", + "role": "reconciler", + }, "work_issue": { "permission": "gitea.pr.create", "role": "author", @@ -653,6 +672,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset( "delete_branch", "cleanup_merged_pr_branch", "reconciliation_cleanup", + "reconcile_missing_worktree_bindings", + "gitea_reconcile_missing_worktree_bindings", "work_issue", "work-issue", } diff --git a/tests/test_issue_970_missing_worktree_reconcile.py b/tests/test_issue_970_missing_worktree_reconcile.py index 0f41034..a97b800 100644 --- a/tests/test_issue_970_missing_worktree_reconcile.py +++ b/tests/test_issue_970_missing_worktree_reconcile.py @@ -10,6 +10,18 @@ Covers all acceptance criteria for #970: - safe and idempotent repeated execution - worktrees dimension resolution post-cleanup - resolution of the two historical post-PR #968 records (#795 / #635) + +Plus the adversarial coverage required by review 644 (B5). Every test below +drives the *production* paths — the registered MCP tools loaded through +``mcp_server``, the real ``ControlPlaneDB`` compare-and-swap, and the real +``audit_reconciliation_mode`` cleanup gate — so each one fails if the +corresponding protection is removed: + +- B1: lease status / ownership / session / path drift between audit and apply +- B2: server-enforced cleanup authorization at both the task-capability and + the production mutation boundary, including a rejected ``operator_authorized`` +- B3: expected-path compare-and-swap on lease *and* checkpoint retirement +- B4: live-session, trustworthy-owner, and conflicting-issue-lock evidence """ from __future__ import annotations @@ -17,27 +29,211 @@ from __future__ import annotations import json import os import shutil +import subprocess +import sys import tempfile import unittest +from datetime import datetime, timedelta, timezone +from unittest.mock import patch +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import audit_reconciliation_mode as arm import control_plane_db as cpd -import lease_lifecycle +import issue_lock_store +import mcp_server import missing_worktree_reconcile as mwr +import task_capability_map + +# Profiles mirroring the configured role split. The reconciler is the only role +# the project grants cleanup authority to (#729 / #970 review 644 B2). +RECONCILER_PROFILE = { + "profile_name": "prgs-reconciler", + "role": "reconciler", + "role_kind": "reconciler", + "allowed_operations": [ + "gitea.read", + "gitea.pr.close", + "gitea.issue.comment", + "gitea.branch.delete", + ], + "forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"], + "audit_label": "prgs-reconciler", +} + +REVIEWER_PROFILE = { + "profile_name": "prgs-reviewer", + "role": "reviewer", + "role_kind": "reviewer", + "allowed_operations": [ + "gitea.read", + "gitea.pr.review", + "gitea.pr.approve", + "gitea.issue.comment", + ], + "forbidden_operations": ["gitea.branch.delete", "gitea.repo.commit"], + "audit_label": "prgs-reviewer", +} + +AUTHOR_WITH_DELETE_PROFILE = { + # Holds the cleanup *permission* but not the cleanup *role*: permission + # alone must never authorize the mutation. + "profile_name": "prgs-author-delete", + "role": "author", + "role_kind": "author", + "allowed_operations": [ + "gitea.read", + "gitea.branch.delete", + "gitea.repo.commit", + ], + "forbidden_operations": [], + "audit_label": "prgs-author-delete", +} -class TestMissingWorktreeReconcile(unittest.TestCase): +class _MissingWorktreeTestBase(unittest.TestCase): + """Shared fixture: a real control-plane DB, git root, and lock dir.""" + def setUp(self): self.temp_dir = tempfile.mkdtemp() self.db_path = os.path.join(self.temp_dir, "control_plane.sqlite3") self.db = cpd.ControlPlaneDB(self.db_path) self.project_root = os.path.join(self.temp_dir, "repo") os.makedirs(os.path.join(self.project_root, "branches"), exist_ok=True) - # Initialize git repo in project_root for git rev-parse checks - os.system(f"git -C '{self.project_root}' init -q") + # Initialize git repo in project_root for git rev-parse checks. + # Argument list form: no shell, so a temp path can never be interpreted + # as shell syntax. + subprocess.run( + ["git", "-C", self.project_root, "init", "-q"], + check=True, + capture_output=True, + ) + + # Point the durable issue-lock store at an isolated directory so the + # real lock reader runs against test data, not the operator's locks. + self.lock_dir = os.path.join(self.temp_dir, "issue-locks") + os.makedirs(self.lock_dir, exist_ok=True) + self._prior_lock_dir = os.environ.get(issue_lock_store.LOCK_DIR_ENV) + os.environ[issue_lock_store.LOCK_DIR_ENV] = self.lock_dir + + arm.clear_phase() def tearDown(self): + arm.clear_phase() + if self._prior_lock_dir is None: + os.environ.pop(issue_lock_store.LOCK_DIR_ENV, None) + else: + os.environ[issue_lock_store.LOCK_DIR_ENV] = self._prior_lock_dir shutil.rmtree(self.temp_dir, ignore_errors=True) + # ── helpers ────────────────────────────────────────────────────────── + + def _age_session_heartbeat(self, session_id, *, seconds=None): + """Age a session's heartbeat past the staleness grace. + + ``assign_and_lease`` opens a session row whose heartbeat is fresh by + construction, and nothing closes it on release. Real abandoned bindings + — the ones #970 exists to retire — have long-stale heartbeats, so the + fixture models that rather than testing against a session that is live + only because the test just created it. + """ + age = seconds or (mwr.SESSION_HEARTBEAT_STALE_SECONDS + 3600) + stale_ts = ( + datetime.now(timezone.utc) - timedelta(seconds=age) + ).replace(microsecond=0).isoformat().replace("+00:00", "Z") + with self.db._tx() as conn: # noqa: SLF001 - fixture ages durable state + conn.execute( + "UPDATE sessions SET last_heartbeat_at = ? WHERE session_id = ?", + (stale_ts, session_id), + ) + + def _stale_lease( + self, + *, + number=402, + path=None, + session_id="sess-402", + org="org", + repo="repo", + ): + """Create a released lease whose recorded worktree path is missing.""" + stale_path = path or os.path.join(self.temp_dir, f"stale_missing_{number}") + res = self.db.assign_and_lease( + remote="prgs", org=org, repo=repo, kind="issue", number=number, + role="author", session_id=session_id, worktree_path=stale_path, + ) + self.db.release_lease_recorded(res.lease_id, session_id=session_id) + self._age_session_heartbeat(session_id) + return res.lease_id, stale_path + + def _audit(self): + return mwr.audit_missing_worktree_bindings( + self.db, project_root=self.project_root, remote="prgs" + ) + + def _candidate(self): + audit = self._audit() + candidates = audit["confirmed_stale_candidates"] + self.assertTrue(candidates, "expected one confirmed stale candidate") + return candidates[0] + + def _mint_cleanup_authorization(self, role="reconciler"): + """Mint a genuine server-side cleanup authorization (#419 gate).""" + arm.clear_phase() + arm.enter_audit_phase("reconcile_merged_cleanups") + phase = arm.authorize_cleanup_phase( + operator_approved=True, + delete_capability_proven=True, + safety_proof={"safe_to_remove_worktree": True}, + before_after_snapshot={ + "before": "binding recorded with missing path", + "after": "binding retired", + }, + ) + self.assertTrue(phase["authorized"], phase) + return mwr.assess_cleanup_authorization( + role=role, + capability_blockers=[], + operator_authorized=False, + task_capability_resolved=True, + ) + + def _lease_row(self, lease_id): + state = self.db.get_lease_workflow_state(lease_id) + return dict(state["lease"]) + + def _set_lease_columns(self, lease_id, **columns): + """Simulate a concurrent writer changing the lease between audit/apply.""" + assignments = ", ".join(f"{k} = ?" for k in columns) + params = list(columns.values()) + [lease_id] + with self.db._tx() as conn: # noqa: SLF001 - deliberate race simulation + conn.execute( + f"UPDATE leases SET {assignments} WHERE lease_id = ?", params + ) + + def _write_issue_lock(self, *, issue_number, worktree_path, branch, pid=None): + """Write a durable issue lock through the production lock store.""" + path = issue_lock_store.lock_file_path( + remote="prgs", + org="org", + repo="repo", + issue_number=issue_number, + lock_dir=self.lock_dir, + ) + issue_lock_store.save_lock_file(path, { + "issue_number": issue_number, + "branch_name": branch, + "worktree_path": worktree_path, + "session_pid": pid if pid is not None else os.getpid(), + "pid": pid if pid is not None else os.getpid(), + "claimant": {"username": "jcwalker3", "profile": "prgs-author"}, + }) + return path + + +class TestMissingWorktreeReconcile(_MissingWorktreeTestBase): + """Original #970 acceptance criteria.""" + def test_audit_missing_worktree_bindings_reports_associations(self): res = self.db.assign_and_lease( remote="prgs", @@ -54,6 +250,7 @@ class TestMissingWorktreeReconcile(unittest.TestCase): self.assertIsNotNone(lease_id) # Release the lease (dead/released) self.db.release_lease_recorded(lease_id, session_id="reviewer-pr795-test-session") + self._age_session_heartbeat("reviewer-pr795-test-session") audit = mwr.audit_missing_worktree_bindings( self.db, @@ -89,6 +286,7 @@ class TestMissingWorktreeReconcile(unittest.TestCase): role="author", session_id="sess-100", worktree_path="/tmp/missing_100", ) self.db.release_lease_recorded(res.lease_id, session_id="sess-100") + self._age_session_heartbeat("sess-100") audit2 = mwr.audit_missing_worktree_bindings(self.db, project_root=bad_root, remote="prgs") self.assertEqual(audit2["unavailable_host_count"], 1) @@ -96,7 +294,7 @@ class TestMissingWorktreeReconcile(unittest.TestCase): self.assertFalse(audit2["missing_bindings"][0]["retire_eligible"]) def test_live_lease_and_live_session_block_retirement(self): - res = self.db.assign_and_lease( + self.db.assign_and_lease( remote="prgs", org="org", repo="repo", kind="issue", number=200, role="author", session_id="sess-live-200", worktree_path="/tmp/missing_live_200", owner_pid=os.getpid(), @@ -106,9 +304,15 @@ class TestMissingWorktreeReconcile(unittest.TestCase): self.assertEqual(audit["live_protected_count"], 1) self.assertEqual(audit["confirmed_stale_count"], 0) - # Attempting resolution on live protected binding is refused + # Attempting resolution on a live protected binding is refused. c = audit["missing_bindings"][0] - res_mut = mwr.resolve_missing_worktree_binding(self.db, binding=c, dry_run=False, operator_authorized=True, project_root=self.project_root) + res_mut = mwr.resolve_missing_worktree_binding( + self.db, + binding=c, + dry_run=False, + cleanup_authorization=self._mint_cleanup_authorization(), + project_root=self.project_root, + ) self.assertFalse(res_mut["success"]) self.assertEqual(res_mut["reason"], "revalidation_failed") @@ -122,6 +326,7 @@ class TestMissingWorktreeReconcile(unittest.TestCase): role="author", session_id="sess-635", worktree_path="/tmp/old_missing_path_635", ) self.db.release_lease_recorded(res.lease_id, session_id="sess-635") + self._age_session_heartbeat("sess-635") cls_info = mwr.classify_worktree_binding( recorded_path="/tmp/old_missing_path_635", @@ -134,15 +339,10 @@ class TestMissingWorktreeReconcile(unittest.TestCase): def test_recreation_race_revalidation_blocks_mutation(self): path = os.path.join(self.temp_dir, "recreated_wt") - res = self.db.assign_and_lease( - remote="prgs", org="org", repo="repo", kind="issue", number=300, - role="author", session_id="sess-300", worktree_path=path, - ) - self.db.release_lease_recorded(res.lease_id, session_id="sess-300") + lease_id, _ = self._stale_lease(number=300, path=path, session_id="sess-300") # Initial audit when path is missing - audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs") - candidate = audit["confirmed_stale_candidates"][0] + candidate = self._candidate() # Simulate concurrent recreation of worktree path os.makedirs(path, exist_ok=True) @@ -152,18 +352,22 @@ class TestMissingWorktreeReconcile(unittest.TestCase): self.db, binding=candidate, dry_run=False, - operator_authorized=True, + cleanup_authorization=self._mint_cleanup_authorization(), project_root=self.project_root, ) self.assertFalse(res_mut["success"]) self.assertEqual(res_mut["reason"], "revalidation_failed") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], path) def test_authorized_cleanup_retires_stale_binding_and_is_idempotent(self): # Create 2 leases: 1 stale missing, 1 valid existing worktree valid_path = os.path.join(self.project_root, "branches", "valid_wt") os.makedirs(valid_path, exist_ok=True) + unrelated_marker = os.path.join(valid_path, "keep-me.txt") + with open(unrelated_marker, "w", encoding="utf-8") as handle: + handle.write("unrelated worktree content") - stale_path = "/tmp/stale_missing_400" + stale_path = os.path.join(self.temp_dir, "stale_missing_400") res1 = self.db.assign_and_lease( remote="prgs", org="org", repo="repo", kind="issue", number=401, @@ -175,18 +379,28 @@ class TestMissingWorktreeReconcile(unittest.TestCase): role="author", session_id="sess-402", worktree_path=stale_path, ) self.db.release_lease_recorded(res2.lease_id, session_id="sess-402") + self._age_session_heartbeat("sess-402") # Dry-run reconciliation dry = mwr.reconcile_missing_worktree_bindings( - self.db, project_root=self.project_root, remote="prgs", dry_run=True, operator_authorized=True + self.db, project_root=self.project_root, remote="prgs", dry_run=True ) self.assertTrue(dry["success"]) self.assertTrue(dry["dry_run"]) self.assertEqual(dry["stale_candidates_count"], 1) - # Apply reconciliation + # Dry-run mutated nothing. + self.assertEqual( + self._lease_row(res2.lease_id)["worktree_path"], stale_path + ) + + # Apply reconciliation with a genuine server-minted authorization applied = mwr.reconcile_missing_worktree_bindings( - self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True + self.db, + project_root=self.project_root, + remote="prgs", + dry_run=False, + cleanup_authorization=self._mint_cleanup_authorization(), ) self.assertTrue(applied["success"]) self.assertFalse(applied["dry_run"]) @@ -203,9 +417,17 @@ class TestMissingWorktreeReconcile(unittest.TestCase): self.assertTrue(prov.get("worktree_path_retired")) self.assertEqual(prov.get("retired_worktree_path"), stale_path) + # Unrelated worktree contents and git metadata are untouched. + self.assertTrue(os.path.isfile(unrelated_marker)) + self.assertTrue(os.path.isdir(os.path.join(self.project_root, ".git"))) + # Re-run (idempotency check) rerun = mwr.reconcile_missing_worktree_bindings( - self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True + self.db, + project_root=self.project_root, + remote="prgs", + dry_run=False, + cleanup_authorization=self._mint_cleanup_authorization(), ) self.assertTrue(rerun["success"]) self.assertTrue(rerun["worktrees_dimension_resolved"]) @@ -219,6 +441,7 @@ class TestMissingWorktreeReconcile(unittest.TestCase): worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/review-feat-issue-628-autonomous-handoffs-orchestration", ) self.db.release_lease_recorded(res_795.lease_id, session_id="reviewer-pr795-hist") + self._age_session_heartbeat("reviewer-pr795-hist") res_635 = self.db.assign_and_lease( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", kind="issue", number=635, @@ -226,6 +449,7 @@ class TestMissingWorktreeReconcile(unittest.TestCase): worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/issue-635-project-registry-api", ) self.db.release_lease_recorded(res_635.lease_id, session_id="author-issue635-hist") + self._age_session_heartbeat("author-issue635-hist") # Audit finds both historical records audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs") @@ -233,11 +457,627 @@ class TestMissingWorktreeReconcile(unittest.TestCase): # Reconcile safely resolves both rec = mwr.reconcile_missing_worktree_bindings( - self.db, project_root=self.project_root, remote="prgs", dry_run=False, operator_authorized=True + self.db, + project_root=self.project_root, + remote="prgs", + dry_run=False, + cleanup_authorization=self._mint_cleanup_authorization(), ) self.assertTrue(rec["worktrees_dimension_resolved"]) self.assertEqual(rec["after_audit"]["missing_count"], 0) +class TestB1StaleSnapshotRevalidation(_MissingWorktreeTestBase): + """B1: safety-relevant drift between audit and apply must fail closed.""" + + def _apply(self, candidate): + return mwr.resolve_missing_worktree_binding( + self.db, + binding=candidate, + dry_run=False, + cleanup_authorization=self._mint_cleanup_authorization(), + project_root=self.project_root, + ) + + def test_lease_reactivated_between_audit_and_apply_fails_closed(self): + """The reviewer's exact probe: released at audit, active at apply.""" + lease_id, stale_path = self._stale_lease(number=501, session_id="sess-501") + candidate = self._candidate() + + # Concurrent writer re-activates the same lease with a live process. + self._set_lease_columns(lease_id, status="active", owner_pid=os.getpid()) + + result = self._apply(candidate) + self.assertFalse(result["success"]) + self.assertFalse(result["performed"]) + self.assertEqual(result["reason"], "safety_state_changed") + self.assertIn("lease_status", result["changed_fields"]) + # Nothing was mutated. + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_lease_owner_session_change_between_audit_and_apply_fails_closed(self): + lease_id, stale_path = self._stale_lease(number=502, session_id="sess-502") + candidate = self._candidate() + + self.db.upsert_session( + session_id="someone-else-503", role="author", profile="prgs-author", + pid=os.getpid(), + ) + self._set_lease_columns(lease_id, session_id="someone-else-503") + + result = self._apply(candidate) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "safety_state_changed") + self.assertIn("lease_session_id", result["changed_fields"]) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_lease_owner_pid_change_between_audit_and_apply_fails_closed(self): + lease_id, stale_path = self._stale_lease(number=504, session_id="sess-504") + candidate = self._candidate() + + self._set_lease_columns(lease_id, owner_pid=999_999_999) + + result = self._apply(candidate) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "safety_state_changed") + self.assertIn("lease_owner_pid", result["changed_fields"]) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_binding_path_change_between_audit_and_apply_fails_closed(self): + lease_id, _ = self._stale_lease(number=505, session_id="sess-505") + candidate = self._candidate() + + moved_target = os.path.join(self.temp_dir, "relocated_missing_505") + self._set_lease_columns(lease_id, worktree_path=moved_target) + + result = self._apply(candidate) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "safety_state_changed") + self.assertIn("binding_path", result["changed_fields"]) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], moved_target) + + def test_live_session_appearing_after_audit_fails_closed(self): + lease_id, stale_path = self._stale_lease(number=506, session_id="sess-506") + candidate = self._candidate() + + # A session row for the owning session appears (or revives) after audit. + self.db.upsert_session( + session_id="sess-506", role="author", profile="prgs-author", + pid=os.getpid(), + ) + + result = self._apply(candidate) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "safety_state_changed") + self.assertTrue( + {"session_live", "session_active", "session_status", "session_pid_alive"} + & set(result["changed_fields"]) + ) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_binding_without_audited_snapshot_is_refused(self): + """A caller cannot skip the drift check by omitting the snapshot.""" + lease_id, stale_path = self._stale_lease(number=507, session_id="sess-507") + candidate = dict(self._candidate()) + candidate.pop("audited_state", None) + + result = self._apply(candidate) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "safety_state_changed") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_revalidation_reads_live_rows_not_audit_snapshot(self): + """read_live_safety_state must come from the DB, not the candidate.""" + lease_id, _ = self._stale_lease(number=508, session_id="sess-508") + candidate = dict(self._candidate()) + # Lie in the snapshot: claim a status the DB never held. + candidate["lease_status"] = "active" + + live = mwr.read_live_safety_state(self.db, candidate) + self.assertEqual(live["snapshot"]["lease_status"], "released") + self.assertEqual(live["lease_row"]["lease_id"], lease_id) + + +class TestB2ServerEnforcedCleanupAuthorization(_MissingWorktreeTestBase): + """B2: authorization is a server artifact, never a client boolean.""" + + def test_operator_authorized_boolean_is_rejected_by_module(self): + self._stale_lease(number=601, session_id="sess-601") + candidate = self._candidate() + + result = mwr.resolve_missing_worktree_binding( + self.db, + binding=candidate, + dry_run=False, + operator_authorized=True, + project_root=self.project_root, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "operator_authorized_rejected") + + def test_apply_without_authorization_is_refused(self): + lease_id, stale_path = self._stale_lease(number=602, session_id="sess-602") + candidate = self._candidate() + + result = mwr.resolve_missing_worktree_binding( + self.db, + binding=candidate, + dry_run=False, + cleanup_authorization=None, + project_root=self.project_root, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "authorization_required") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_forged_authorization_without_cleanup_phase_is_refused(self): + """A hand-built authorization mapping cannot stand in for the gate.""" + lease_id, stale_path = self._stale_lease(number=603, session_id="sess-603") + candidate = self._candidate() + + forged = { + "authorized": True, + "source": mwr.CLEANUP_AUTHORIZATION_SOURCE, + "role": "reconciler", + "reasons": [], + } + arm.clear_phase() # the server never authorized a cleanup phase + result = mwr.resolve_missing_worktree_binding( + self.db, + binding=candidate, + dry_run=False, + cleanup_authorization=forged, + project_root=self.project_root, + ) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "authorization_required") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_authorization_requires_reconciler_role(self): + arm.clear_phase() + arm.enter_audit_phase("reconcile_merged_cleanups") + arm.authorize_cleanup_phase( + operator_approved=True, + delete_capability_proven=True, + safety_proof={"safe_to_remove_worktree": True}, + before_after_snapshot={"before": "a", "after": "b"}, + ) + auth = mwr.assess_cleanup_authorization( + role="author", capability_blockers=[], task_capability_resolved=True + ) + self.assertFalse(auth["authorized"]) + self.assertTrue(any("reconciler" in r for r in auth["reasons"])) + + def test_authorization_requires_delete_capability(self): + arm.clear_phase() + arm.enter_audit_phase("reconcile_merged_cleanups") + arm.authorize_cleanup_phase( + operator_approved=True, + delete_capability_proven=True, + safety_proof={"safe_to_remove_worktree": True}, + before_after_snapshot={"before": "a", "after": "b"}, + ) + auth = mwr.assess_cleanup_authorization( + role="reconciler", + capability_blockers=["profile forbids gitea.branch.delete"], + task_capability_resolved=True, + ) + self.assertFalse(auth["authorized"]) + + def test_authorization_requires_resolved_task_capability(self): + arm.clear_phase() + arm.enter_audit_phase("reconcile_merged_cleanups") + arm.authorize_cleanup_phase( + operator_approved=True, + delete_capability_proven=True, + safety_proof={"safe_to_remove_worktree": True}, + before_after_snapshot={"before": "a", "after": "b"}, + ) + auth = mwr.assess_cleanup_authorization( + role="reconciler", capability_blockers=[], task_capability_resolved=False + ) + self.assertFalse(auth["authorized"]) + self.assertTrue(any(mwr.CLEANUP_TASK in r for r in auth["reasons"])) + + def test_operator_authorized_rejected_even_for_authorized_reconciler(self): + auth = mwr.assess_cleanup_authorization( + role="reconciler", + capability_blockers=[], + operator_authorized=True, + task_capability_resolved=True, + ) + self.assertFalse(auth["authorized"]) + self.assertTrue(auth["operator_authorized_rejected"]) + + def test_cleanup_task_capability_is_reconciler_only(self): + self.assertEqual( + task_capability_map.required_role(mwr.CLEANUP_TASK), "reconciler" + ) + self.assertEqual( + task_capability_map.required_permission(mwr.CLEANUP_TASK), + mwr.CLEANUP_REQUIRED_PERMISSION, + ) + self.assertIn(mwr.CLEANUP_TASK, task_capability_map.ROLE_EXCLUSIVE_TASKS) + + +class TestB2ProductionMcpBoundary(_MissingWorktreeTestBase): + """B2 at the registered MCP tool — the real production entrypoint.""" + + CANONICAL_ORG = "Scaled-Tech-Consulting" + CANONICAL_REPO = "Gitea-Tools" + + def _mcp_stale_lease(self, *, number, session_id): + """A stale lease under the org/repo the tool resolves for 'prgs'.""" + return self._stale_lease( + number=number, + session_id=session_id, + org=self.CANONICAL_ORG, + repo=self.CANONICAL_REPO, + ) + + def _call_reconcile(self, **kwargs): + # org/repo are passed explicitly: the 'prgs' remote defaults to the + # Timesheet repo, so relying on resolution would audit a different + # repository than the fixture wrote to. + kwargs.setdefault("org", self.CANONICAL_ORG) + kwargs.setdefault("repo", self.CANONICAL_REPO) + with patch.object(mcp_server, "_control_plane_db_or_error", + return_value=(self.db, [])), \ + patch.object(mcp_server, "_canonical_local_git_root", + return_value=self.project_root): + return mcp_server.gitea_reconcile_missing_worktree_bindings(**kwargs) + + def test_production_tools_are_registered(self): + for name in ( + "gitea_audit_missing_worktree_bindings", + "gitea_reconcile_missing_worktree_bindings", + ): + self.assertTrue( + callable(getattr(mcp_server, name, None)), + f"{name} is not registered in the production MCP module", + ) + + @patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE) + def test_reviewer_with_operator_authorized_true_is_rejected(self, _profile): + lease_id, stale_path = self._mcp_stale_lease(number=701, session_id="sess-701") + + result = self._call_reconcile( + dry_run=False, operator_authorized=True, remote="prgs" + ) + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "operator_authorized_rejected") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + @patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE) + def test_reviewer_apply_without_cleanup_capability_is_rejected(self, _profile): + lease_id, stale_path = self._mcp_stale_lease(number=702, session_id="sess-702") + + result = self._call_reconcile(dry_run=False, remote="prgs") + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "cleanup_authorization_required") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + @patch.object(mcp_server, "get_profile", return_value=AUTHOR_WITH_DELETE_PROFILE) + def test_wrong_role_with_delete_permission_is_rejected(self, _profile): + """Permission alone must not authorize: the role gate must hold.""" + lease_id, stale_path = self._mcp_stale_lease(number=703, session_id="sess-703") + + result = self._call_reconcile(dry_run=False, remote="prgs") + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "cleanup_authorization_required") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + @patch.object(mcp_server, "get_profile", return_value=RECONCILER_PROFILE) + def test_reconciler_apply_without_authorized_cleanup_phase_is_rejected(self, _profile): + """Correct role and capability, but no minted cleanup phase.""" + lease_id, stale_path = self._mcp_stale_lease(number=704, session_id="sess-704") + arm.clear_phase() + + result = self._call_reconcile(dry_run=False, remote="prgs") + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "cleanup_authorization_required") + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + @patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE) + def test_dry_run_is_allowed_and_non_mutating_for_read_profile(self, _profile): + lease_id, stale_path = self._mcp_stale_lease(number=705, session_id="sess-705") + + result = self._call_reconcile(dry_run=True, remote="prgs") + self.assertTrue(result["success"]) + self.assertTrue(result["dry_run"]) + self.assertEqual(result["stale_candidates_count"], 1) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + @patch.object(mcp_server, "get_profile", return_value=RECONCILER_PROFILE) + def test_authorized_reconciler_apply_retires_binding(self, _profile): + lease_id, _ = self._mcp_stale_lease(number=706, session_id="sess-706") + arm.clear_phase() + arm.enter_audit_phase("reconcile_merged_cleanups") + arm.authorize_cleanup_phase( + operator_approved=True, + delete_capability_proven=True, + safety_proof={"safe_to_remove_worktree": True}, + before_after_snapshot={"before": "bound", "after": "retired"}, + ) + + result = self._call_reconcile(dry_run=False, remote="prgs") + self.assertTrue(result["success"], result) + self.assertTrue(result["worktrees_dimension_resolved"]) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], "") + + @patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE) + def test_audit_tool_is_read_only(self, _profile): + lease_id, stale_path = self._mcp_stale_lease(number=707, session_id="sess-707") + with patch.object(mcp_server, "_control_plane_db_or_error", + return_value=(self.db, [])), \ + patch.object(mcp_server, "_canonical_local_git_root", + return_value=self.project_root): + audit = mcp_server.gitea_audit_missing_worktree_bindings( + remote="prgs", org=self.CANONICAL_ORG, repo=self.CANONICAL_REPO + ) + self.assertEqual(audit["confirmed_stale_count"], 1) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + +class TestB3ExpectedPathCompareAndSwap(_MissingWorktreeTestBase): + """B3: both retirement paths enforce the expected stored path.""" + + def _checkpoint(self, *, session_id="cp-sess", worktree_path, number=800): + self.db.write_session_checkpoint( + remote="prgs", org="org", repo="repo", + session_id=session_id, role="author", + work_kind="issue", work_number=number, + worktree_path=worktree_path, branch=f"fix/issue-{number}-x", + ) + rows = self.db.list_session_checkpoints(session_id=session_id) + self.assertEqual(len(rows), 1) + return rows[0] + + def test_checkpoint_retirement_refuses_wrong_expected_path(self): + """The reviewer's probe: wrong expected_path returned retired=True.""" + missing = os.path.join(self.temp_dir, "cp_missing_801") + row = self._checkpoint(worktree_path=missing, number=801) + + with self.assertRaises(cpd.ControlPlaneError): + self.db.retire_session_checkpoint_worktree_path( + session_id="cp-sess", + checkpoint_id=row["checkpoint_id"], + expected_path="/some/other/path", + ) + + after = self.db.list_session_checkpoints(session_id="cp-sess")[0] + self.assertEqual(after["worktree_path"], missing) + + def test_checkpoint_retirement_requires_expected_path(self): + missing = os.path.join(self.temp_dir, "cp_missing_802") + row = self._checkpoint(worktree_path=missing, number=802) + + with self.assertRaises(cpd.ControlPlaneError): + self.db.retire_session_checkpoint_worktree_path( + session_id="cp-sess", checkpoint_id=row["checkpoint_id"] + ) + after = self.db.list_session_checkpoints(session_id="cp-sess")[0] + self.assertEqual(after["worktree_path"], missing) + + def test_checkpoint_retirement_refuses_unknown_row(self): + with self.assertRaises(cpd.ControlPlaneError): + self.db.retire_session_checkpoint_worktree_path( + session_id="no-such-session", expected_path="/tmp/whatever" + ) + + def test_checkpoint_retirement_succeeds_on_exact_path(self): + missing = os.path.join(self.temp_dir, "cp_missing_803") + row = self._checkpoint(worktree_path=missing, number=803) + + proof = self.db.retire_session_checkpoint_worktree_path( + session_id="cp-sess", + checkpoint_id=row["checkpoint_id"], + expected_path=missing, + ) + self.assertTrue(proof["retired"]) + after = self.db.list_session_checkpoints(session_id="cp-sess")[0] + self.assertEqual(after["worktree_path"], "") + + # Idempotent: a second retirement is a no-op, not a failure. + again = self.db.retire_session_checkpoint_worktree_path( + session_id="cp-sess", + checkpoint_id=row["checkpoint_id"], + expected_path=missing, + ) + self.assertFalse(again["retired"]) + self.assertTrue(again["already_retired"]) + + def test_lease_retirement_refuses_wrong_expected_path(self): + lease_id, stale_path = self._stale_lease(number=804, session_id="sess-804") + with self.assertRaises(cpd.ControlPlaneError): + self.db.retire_lease_worktree_path( + lease_id, expected_path="/not/the/stored/path" + ) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_lease_retirement_refuses_changed_status(self): + lease_id, stale_path = self._stale_lease(number=805, session_id="sess-805") + with self.assertRaises(cpd.ControlPlaneError): + self.db.retire_lease_worktree_path( + lease_id, expected_path=stale_path, expected_status="active" + ) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_lease_retirement_refuses_changed_session(self): + lease_id, stale_path = self._stale_lease(number=806, session_id="sess-806") + with self.assertRaises(cpd.ControlPlaneError): + self.db.retire_lease_worktree_path( + lease_id, + expected_path=stale_path, + expected_session_id="a-different-session", + ) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_lease_retirement_requires_expected_path(self): + lease_id, stale_path = self._stale_lease(number=807, session_id="sess-807") + with self.assertRaises(cpd.ControlPlaneError): + self.db.retire_lease_worktree_path(lease_id) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + +class TestB4OwnershipEvidence(_MissingWorktreeTestBase): + """B4: live-session, trustworthy-owner, and issue-lock evidence.""" + + def test_active_lease_with_dead_pid_is_not_confirmed_stale(self): + """Dead-PID evidence alone must not retire a lease still held.""" + dead_pid = 999_999_999 + self.db.assign_and_lease( + remote="prgs", org="org", repo="repo", kind="issue", number=901, + role="author", session_id="sess-901", + worktree_path=os.path.join(self.temp_dir, "missing_901"), + owner_pid=dead_pid, + ) + audit = self._audit() + self.assertEqual(audit["confirmed_stale_count"], 0) + self.assertEqual( + audit["missing_bindings"][0]["classification"], + mwr.CLASS_LIVE_LEASE_PROTECTED, + ) + + def test_session_active_is_populated_from_the_sessions_table(self): + self.db.assign_and_lease( + remote="prgs", org="org", repo="repo", kind="issue", number=902, + role="author", session_id="sess-902", + worktree_path=os.path.join(self.temp_dir, "missing_902"), + owner_pid=os.getpid(), + ) + evidence = mwr.collect_session_evidence(self.db) + self.assertIn("sess-902", evidence) + self.assertTrue(evidence["sess-902"]["session_active"]) + self.assertTrue(evidence["sess-902"]["session_pid_alive"]) + + binding = self._audit()["missing_bindings"][0] + self.assertTrue(binding["session_active"]) + + def test_live_session_blocks_cleanup_of_released_lease(self): + lease_id, stale_path = self._stale_lease(number=903, session_id="sess-903") + # The lease is released, but its session is live. + self.db.upsert_session( + session_id="sess-903", role="author", profile="prgs-author", + pid=os.getpid(), + ) + audit = self._audit() + self.assertEqual(audit["confirmed_stale_count"], 0) + self.assertEqual( + audit["missing_bindings"][0]["classification"], + mwr.CLASS_LIVE_SESSION_PROTECTED, + ) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_trusted_owner_evidence_without_live_process_blocks_cleanup(self): + """Session recorded active, process gone: still a valid owner signal. + + No terminal lease contradicts the session record here (a checkpoint-style + binding), so the recorded owner is the best evidence available and must + block cleanup even though nothing is running. + """ + result = mwr.classify_worktree_binding( + recorded_path=os.path.join(self.temp_dir, "missing_904"), + owner_pid=999_999_999, + session_id="sess-904", + session_active=True, + session_pid_alive=False, + session_status="active", + project_root=self.project_root, + ) + self.assertEqual(result["classification"], mwr.CLASS_TRUSTED_OWNER_PROTECTED) + self.assertFalse(result["retire_eligible"]) + + def test_released_lease_outranks_a_lingering_active_session_row(self): + """An explicit release is later and more specific than a session row. + + Session rows are opened on assignment and never closed on release, so + treating one as protective would make every released lease permanently + unretireable — the exact degradation #970 exists to clear. + """ + result = mwr.classify_worktree_binding( + recorded_path=os.path.join(self.temp_dir, "missing_908"), + lease_status="released", + owner_pid=999_999_999, + session_id="sess-908", + session_active=True, + session_pid_alive=False, + session_status="active", + project_root=self.project_root, + ) + self.assertEqual(result["classification"], mwr.CLASS_CONFIRMED_STALE_DELETED) + self.assertTrue(result["retire_eligible"]) + + def test_checkpoint_binding_with_active_session_is_protected(self): + """A checkpoint has no lease to contradict its owning session.""" + missing = os.path.join(self.temp_dir, "cp_missing_909") + self.db.write_session_checkpoint( + remote="prgs", org="org", repo="repo", + session_id="sess-909", role="author", + work_kind="issue", work_number=909, + worktree_path=missing, branch="fix/issue-909-x", + ) + self.db.upsert_session( + session_id="sess-909", role="author", profile="prgs-author", + pid=999_999_999, + ) + audit = self._audit() + checkpoint_bindings = [ + b for b in audit["missing_bindings"] + if b.get("source") == "session_checkpoint" + ] + self.assertEqual(len(checkpoint_bindings), 1) + self.assertEqual( + checkpoint_bindings[0]["classification"], + mwr.CLASS_TRUSTED_OWNER_PROTECTED, + ) + + def test_conflicting_issue_lock_blocks_cleanup(self): + stale_path = os.path.join(self.temp_dir, "missing_905") + lease_id, _ = self._stale_lease( + number=905, path=stale_path, session_id="sess-905" + ) + self._write_issue_lock( + issue_number=905, + worktree_path=stale_path, + branch="fix/issue-905-x", + ) + + audit = self._audit() + self.assertEqual(audit["confirmed_stale_count"], 0) + self.assertEqual( + audit["missing_bindings"][0]["classification"], + mwr.CLASS_ISSUE_LOCK_PROTECTED, + ) + self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path) + + def test_issue_lock_missing_path_is_reported_but_never_retired(self): + lock_path = os.path.join(self.temp_dir, "missing_906") + self._write_issue_lock( + issue_number=906, worktree_path=lock_path, branch="fix/issue-906-x" + ) + audit = self._audit() + self.assertEqual(audit["issue_lock_missing_count"], 1) + reported = audit["issue_lock_missing_bindings"][0] + self.assertFalse(reported["retire_eligible"]) + self.assertEqual(reported["issue_number"], 906) + + def test_issue_lock_with_dead_owner_does_not_block_unrelated_binding(self): + """A lock is only protective while it is live: no over-blocking.""" + stale_path = os.path.join(self.temp_dir, "missing_907") + self._stale_lease(number=907, path=stale_path, session_id="sess-907") + self._write_issue_lock( + issue_number=907, + worktree_path=os.path.join(self.temp_dir, "other_missing_907"), + branch="fix/issue-907-other", + pid=999_999_999, + ) + audit = self._audit() + self.assertEqual(audit["confirmed_stale_count"], 1) + self.assertEqual( + audit["confirmed_stale_candidates"][0]["worktree_path"], stale_path + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_task_capability_role_invariants.py b/tests/test_task_capability_role_invariants.py index 25e89a8..55fd318 100644 --- a/tests/test_task_capability_role_invariants.py +++ b/tests/test_task_capability_role_invariants.py @@ -153,6 +153,12 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset( "delete_branch", "cleanup_merged_pr_branch", "reconciliation_cleanup", + # #970 review 644 B2: retiring a missing worktree binding is a + # control-plane cleanup mutation, so it carries the same reconciler-only + # authority as every other reconciliation cleanup. Permission alone must + # not authorize it. + "reconcile_missing_worktree_bindings", + "gitea_reconcile_missing_worktree_bindings", "work_issue", "work-issue", }