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", }