diff --git a/author_issue_bootstrap.py b/author_issue_bootstrap.py new file mode 100644 index 0000000..c246698 --- /dev/null +++ b/author_issue_bootstrap.py @@ -0,0 +1,976 @@ +"""Sanctioned author issue worktree bootstrap for allocated issues (#850). + +Bootstraps an allocated author branch, canonical worktree under ``branches/``, +worktree registration, issue lock, and lease/assignment binding without +caller-side Git, Bash, or helper scripts. + +Features: +1. Durable phase journal with read-after-write evidence for every phase. +2. Idempotent replay handling via idempotency keys. +3. Authoritative expected-base / concurrency-pin validation. +4. Typed stale-pin refusal without silent rebasing or repointing. +5. Canonical branches-root enforcement (path MUST be inside branches/). +6. Preexisting work preservation (dirty tracked/untracked check, foreign ownership refusal). +7. Compensating recovery limited strictly to artifacts created by this transition. +8. Satisfiable exact_next_action for MCP scheduled workers. +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from typing import Any, Mapping + +import author_mutation_worktree +import issue_lock_store +import issue_lock_worktree +import lease_lifecycle +from reviewer_worktree import parse_dirty_tracked_files +import task_capability_map + +BOOTSTRAP_TASKS = frozenset( + { + "bootstrap_author_issue_worktree", + "gitea_bootstrap_author_issue_worktree", + } +) + +PHASE_1_REQUEST_ACCEPTED = "1_request_accepted" +PHASE_2_BRANCH_CONFIRMED = "2_branch_confirmed" +PHASE_3_PATH_RESERVED = "3_path_reserved" +PHASE_4_WORKTREE_CONFIRMED = "4_worktree_confirmed" +PHASE_5_REGISTRATION_VERIFIED = "5_registration_verified" +PHASE_6_STATE_ESTABLISHED = "6_state_established" +PHASE_7_TRANSITION_COMPLETED = "7_transition_completed" +PHASE_COMPENSATING_RECOVERY = "compensating_recovery" + +JOURNAL_DIR_NAME = "bootstrap-journals" + + +def is_author_issue_bootstrap_task(task: str | None) -> bool: + """True when *task* is the author issue worktree bootstrap task.""" + return (task or "").strip() in BOOTSTRAP_TASKS + + +def get_journal_dir(override: str | None = None) -> str: + """Return the root directory for durable bootstrap phase journals.""" + if override: + path = override + elif os.environ.get("GITEA_BOOTSTRAP_JOURNAL_DIR"): + path = os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] + else: + cache_dir = os.path.expanduser("~/.cache/gitea-tools") + path = os.path.join(cache_dir, JOURNAL_DIR_NAME) + os.makedirs(path, exist_ok=True) + return path + + +def _journal_file_path(idempotency_key: str, journal_dir: str | None = None) -> str: + safe_key = "".join( + c if c.isalnum() or c in ("-", "_", ".") else "_" + for c in idempotency_key + ) + return os.path.join(get_journal_dir(journal_dir), f"{safe_key}.json") + + +def load_phase_journal( + idempotency_key: str, journal_dir: str | None = None +) -> dict[str, Any] | None: + """Load a durable phase journal if it exists.""" + path = _journal_file_path(idempotency_key, journal_dir=journal_dir) + if not os.path.isfile(path): + return None + try: + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + except Exception: + return None + + +def save_phase_journal( + journal: dict[str, Any], journal_dir: str | None = None +) -> None: + """Persist a durable phase journal with write-through file sync.""" + key = journal["idempotency_key"] + path = _journal_file_path(key, journal_dir=journal_dir) + tmp_path = f"{path}.tmp.{os.getpid()}" + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(journal, f, indent=2, sort_keys=True) + os.replace(tmp_path, path) + + +def derive_default_idempotency_key( + remote: str, + org: str | None, + repo: str | None, + issue_number: int, + assignment_id: str | None = None, + lease_id: str | None = None, +) -> str: + parts = [ + "bootstrap", + (remote or "prgs").strip(), + (org or "Scaled-Tech-Consulting").strip(), + (repo or "Gitea-Tools").strip(), + f"issue-{issue_number}", + ] + if assignment_id: + parts.append(assignment_id.strip()) + if lease_id: + parts.append(lease_id.strip()) + return ":".join(parts) + + +def run_compensating_recovery( + journal: dict[str, Any], + canonical_repo_root: str, + journal_dir: str | None = None, +) -> dict[str, Any]: + """Execute compensating recovery for artifacts created by this transition only.""" + artifacts = journal.get("artifacts_created") or {} + pending = journal.get("pending_creations") or {} + rolled_back: list[str] = [] + worktree_path = journal.get("worktree_path") + branch_name = journal.get("branch_name") + + # Roll back issue lock if created + if artifacts.get("lock_created") or pending.get("lock"): + issue_num = journal.get("issue_number") + session_id = journal.get("owner_session") + if issue_num and session_id: + try: + issue_lock_store.release_session_lock( + issue_number=issue_num, + session=session_id, + lock_dir=journal_dir, + ) + rolled_back.append(f"lock:issue-{issue_num}") + except Exception: + pass + artifacts["lock_created"] = False + + worktree_created = ( + artifacts.get("worktree_registered") + or artifacts.get("worktree_dir_created") + or (pending.get("worktree_path") == worktree_path and worktree_path) + ) + + if worktree_created and worktree_path: + if os.path.exists(worktree_path): + # Re-verify cleanliness before destructive removal (F-4) + porc_res = subprocess.run( + ["git", "-C", worktree_path, "status", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + is_dirty = porc_res.returncode == 0 and bool(porc_res.stdout.strip()) + if is_dirty: + rolled_back.append(f"worktree_path_preserved_dirty:{worktree_path}") + else: + try: + subprocess.run( + [ + "git", + "-C", + canonical_repo_root, + "worktree", + "remove", + "--force", + worktree_path, + ], + capture_output=True, + text=True, + check=False, + ) + except Exception: + pass + if os.path.exists(worktree_path): + shutil.rmtree(worktree_path, ignore_errors=True) + try: + subprocess.run( + ["git", "-C", canonical_repo_root, "worktree", "prune"], + capture_output=True, + text=True, + check=False, + ) + except Exception: + pass + rolled_back.append(f"worktree_path:{worktree_path}") + else: + rolled_back.append(f"worktree_path:{worktree_path}") + + branch_created = ( + artifacts.get("branch_created") + or (pending.get("branch_name") == branch_name and branch_name) + ) + + if branch_created and branch_name: + try: + res = subprocess.run( + [ + "git", + "-C", + canonical_repo_root, + "rev-parse", + "--verify", + branch_name, + ], + capture_output=True, + text=True, + check=False, + ) + if res.returncode == 0: + # Check for author commits on branch before branch deletion (F-4) + resolved_base = journal.get("resolved_base_sha") or "master" + rev_list_res = subprocess.run( + [ + "git", + "-C", + canonical_repo_root, + "rev-list", + f"{resolved_base}..{branch_name}", + ], + capture_output=True, + text=True, + check=False, + ) + has_commits = rev_list_res.returncode == 0 and bool(rev_list_res.stdout.strip()) + if has_commits: + rolled_back.append(f"branch_preserved_commits:{branch_name}") + else: + subprocess.run( + [ + "git", + "-C", + canonical_repo_root, + "branch", + "-D", + branch_name, + ], + capture_output=True, + text=True, + check=False, + ) + rolled_back.append(f"branch:{branch_name}") + except Exception: + pass + + recovery_info = { + "executed": True, + "rolled_back": rolled_back, + "reason": journal.get("failure_reason"), + } + journal["compensating_recovery"] = recovery_info + journal["current_phase"] = PHASE_COMPENSATING_RECOVERY + save_phase_journal(journal, journal_dir=journal_dir) + return recovery_info + + +def assess_author_issue_bootstrap( + *, + workspace_path: str, + canonical_repo_root: str, + current_branch: str | None = None, + head_sha: str | None = None, + porcelain_status: str = "", + remote_master_sha: str | None = None, + remote_master_sha_error: str | None = None, + task: str | None = None, +) -> dict[str, Any]: + """Assess whether author issue worktree bootstrap may proceed from control or worktree root.""" + root = os.path.realpath(canonical_repo_root or "") + workspace = os.path.realpath(workspace_path or root or ".") + branch = (current_branch or "").strip() + dirty = parse_dirty_tracked_files(porcelain_status or "") + under_branches = ( + author_mutation_worktree.is_path_under_branches(workspace, root) + if root + else False + ) + + if not is_author_issue_bootstrap_task(task): + return { + "not_applicable": True, + "allowed": False, + "block": False, + "proven": False, + "reasons": ["task is not author_issue_bootstrap"], + } + + if under_branches: + return { + "not_applicable": False, + "allowed": True, + "block": False, + "proven": True, + "bootstrap_path": "existing_branches_worktree", + "reasons": [ + "workspace is already a registered worktree under branches/" + ], + } + + reasons: list[str] = [] + if workspace != root: + reasons.append( + "bootstrap requires workspace to be canonical control checkout or branches/ worktree" + ) + if branch not in author_mutation_worktree.BASE_BRANCHES: + reasons.append( + f"control checkout branch '{branch}' is not an accepted base branch " + f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})" + ) + if dirty: + reasons.append( + f"control checkout has tracked local edits: {', '.join(dirty[:5])}" + ) + + if remote_master_sha_error: + reasons.append( + f"could not verify live master tip: {remote_master_sha_error}" + ) + elif remote_master_sha and head_sha: + h = head_sha.strip().lower() + rm = remote_master_sha.strip().lower() + if h != rm: + reasons.append( + f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})" + ) + + if reasons: + return { + "not_applicable": False, + "allowed": False, + "block": True, + "proven": False, + "reasons": reasons, + } + + return { + "not_applicable": False, + "allowed": True, + "block": False, + "proven": True, + "bootstrap_path": "clean_canonical_control_checkout", + "reasons": [ + "control checkout is clean on accepted base branch matching live master" + ], + } + + +import fcntl +import stat + + +class BootstrapTransitionLock: + """Inter-process file lock scoped to the transition identity / idempotency key.""" + + def __init__(self, idempotency_key: str, journal_dir: str | None = None): + safe_key = "".join( + c if c.isalnum() or c in ("-", "_", ".") else "_" + for c in idempotency_key + ) + lock_dir = os.path.realpath(get_journal_dir(journal_dir)) + if not os.path.isdir(lock_dir): + raise RuntimeError(f"Lock directory '{lock_dir}' does not exist or is not a directory") + + raw_lock_path = os.path.abspath(os.path.join(lock_dir, f"{safe_key}.lock")) + try: + common = os.path.commonpath([lock_dir, os.path.dirname(raw_lock_path)]) + except Exception: + common = None + if common != lock_dir: + raise RuntimeError(f"Lock path '{raw_lock_path}' escapes canonical lock directory '{lock_dir}'") + + self.lock_path = raw_lock_path + self.fd = None + + def __enter__(self): + if os.path.islink(self.lock_path): + raise RuntimeError(f"Refusing lock acquisition: lock path '{self.lock_path}' is a symlink") + + flags = os.O_RDWR | os.O_CREAT + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + + try: + fd = os.open(self.lock_path, flags, 0o600) + except OSError as exc: + raise RuntimeError(f"Failed to open lock file safely '{self.lock_path}': {exc}") from exc + + st = os.fstat(fd) + if not stat.S_ISREG(st.st_mode): + os.close(fd) + raise RuntimeError(f"Lock target '{self.lock_path}' is not a regular file") + + self.fd = fd + fcntl.flock(self.fd, fcntl.LOCK_EX) + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self.fd is not None: + try: + fcntl.flock(self.fd, fcntl.LOCK_UN) + except Exception: + pass + try: + os.close(self.fd) + except Exception: + pass + self.fd = None + + +def bootstrap_author_issue_worktree( + *, + issue_number: int, + canonical_repo_root: str, + assignment_id: str | None = None, + lease_id: str | None = None, + expected_base_sha: str | None = None, + branch_name: str | None = None, + worktree_path: str | None = None, + idempotency_key: str | None = None, + remote: str = "prgs", + host: str | None = None, + org: str | None = "Scaled-Tech-Consulting", + repo: str | None = "Gitea-Tools", + active_identity: str | None = "jcwalker3", + active_profile: str | None = "prgs-author", + owner_session: str | None = None, + lock_dir: str | None = None, + dry_run: bool = False, +) -> dict[str, Any]: + """Execute the sanctioned author issue worktree bootstrap transition.""" + root = os.path.realpath(canonical_repo_root) + + session = (owner_session or "").strip() + if not session: + return { + "success": False, + "reason_code": "missing_owner_session", + "message": "Missing required owner_session parameter (fail closed). Session identifier cannot be fabricated or defaulted.", + "exact_next_action": ( + "Pass explicit owner_session resolved from gitea_whoami or session context." + ), + } + + if active_identity is None or not str(active_identity).strip(): + return { + "success": False, + "reason_code": "missing_active_identity", + "message": "Missing required active_identity parameter (fail closed). Identity cannot be fabricated or defaulted.", + "exact_next_action": "Pass explicit active_identity resolved from gitea_whoami.", + } + identity = str(active_identity).strip() + + if active_profile is None or not str(active_profile).strip(): + return { + "success": False, + "reason_code": "missing_active_profile", + "message": "Missing required active_profile parameter (fail closed). Profile cannot be fabricated or defaulted.", + "exact_next_action": "Pass explicit active_profile resolved from gitea_whoami.", + } + profile = str(active_profile).strip() + + # Derive standard inputs + expected_pattern = f"issue-{issue_number}" + target_branch = (branch_name or "").strip() + if not target_branch: + target_branch = f"fix/issue-{issue_number}-native-mcp-bootstrap" + elif expected_pattern not in target_branch: + return { + "success": False, + "reason_code": "invalid_branch_name", + "message": ( + f"Branch name '{target_branch}' must contain issue pattern '{expected_pattern}'" + ), + "exact_next_action": ( + f"Supply a branch_name containing '{expected_pattern}', e.g., 'fix/issue-{issue_number}-...'" + ), + } + + worktree_name = target_branch.replace("/", "-") + target_worktree = (worktree_path or "").strip() + if not target_worktree: + target_worktree = os.path.join(root, "branches", worktree_name) + target_worktree = os.path.realpath(os.path.abspath(target_worktree)) + + key = (idempotency_key or "").strip() + if not key: + key = derive_default_idempotency_key( + remote=remote, + org=org, + repo=repo, + issue_number=issue_number, + assignment_id=assignment_id, + lease_id=lease_id, + ) + + # Acquire cross-process file lock scoped to the idempotency key / transition identity + with BootstrapTransitionLock(key, journal_dir=lock_dir): + # Idempotency check + existing = load_phase_journal(key, journal_dir=lock_dir) + if existing and existing.get("completed"): + if ( + existing.get("issue_number") == issue_number + and existing.get("branch_name") == target_branch + and os.path.realpath(existing.get("worktree_path", "")) + == target_worktree + ): + return { + "success": True, + "replayed": True, + "message": ( + f"Idempotent replay: worktree for issue #{issue_number} already bootstrapped at {target_worktree}" + ), + "issue_number": issue_number, + "branch_name": target_branch, + "worktree_path": target_worktree, + "base_sha": existing.get("resolved_base_sha"), + "lease_id": existing.get("lease_id"), + "assignment_id": existing.get("assignment_id"), + "idempotency_key": key, + "phase_journal": existing, + "exact_next_action": ( + "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') " + "and proceed with author implementation in the bootstrapped worktree." + ), + } + else: + return { + "success": False, + "reason_code": "incompatible_idempotency_replay", + "message": ( + f"Idempotency key '{key}' already exists with incompatible parameters " + f"(stored: {existing.get('branch_name')}, {existing.get('worktree_path')}; " + f"requested: {target_branch}, {target_worktree})" + ), + "exact_next_action": ( + "Supply a unique idempotency_key or pass compatible parameters." + ), + } + + # Initialize or resume Phase Journal + if existing: + journal = existing + artifacts = journal.setdefault("artifacts_created", {}) + artifacts.setdefault("branch_created", False) + artifacts.setdefault("worktree_dir_created", False) + artifacts.setdefault("worktree_registered", False) + artifacts.setdefault("lock_created", False) + else: + journal = { + "idempotency_key": key, + "issue_number": issue_number, + "assignment_id": assignment_id, + "lease_id": lease_id, + "expected_base_sha": expected_base_sha, + "resolved_base_sha": None, + "branch_name": target_branch, + "worktree_path": target_worktree, + "active_identity": identity, + "active_profile": profile, + "owner_session": session, + "remote": remote, + "org": org, + "repo": repo, + "phases": {}, + "artifacts_created": { + "branch_created": False, + "worktree_dir_created": False, + "worktree_registered": False, + "lock_created": False, + }, + "current_phase": PHASE_1_REQUEST_ACCEPTED, + "completed": False, + } + + # Fetch current live master SHA + try: + rev_res = subprocess.run( + ["git", "-C", root, "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=True, + ) + live_master_sha = rev_res.stdout.strip() + except Exception as exc: + return { + "success": False, + "reason_code": "git_rev_parse_failed", + "message": f"Could not determine repository HEAD: {exc}", + "exact_next_action": "Verify repository git state and retry.", + } + + # Phase 1: REQUEST_ACCEPTED & Concurrency Pin Check + if expected_base_sha: + exp_norm = expected_base_sha.strip().lower() + live_norm = live_master_sha.lower() + if exp_norm != live_norm: + journal["failure_reason"] = ( + f"stale concurrency pin: expected {exp_norm[:12]} != live {live_norm[:12]}" + ) + save_phase_journal(journal, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "stale_concurrency_pin", + "message": ( + f"Expected base SHA {exp_norm[:12]} does not match live master SHA {live_norm[:12]} (fail closed)." + ), + "expected_base_sha": expected_base_sha, + "live_master_sha": live_master_sha, + "exact_next_action": ( + "Re-evaluate assignment against current live master SHA and retry with updated expected_base_sha." + ), + } + + journal["resolved_base_sha"] = live_master_sha + journal["phases"][PHASE_1_REQUEST_ACCEPTED] = { + "status": "completed", + "live_master_sha": live_master_sha, + "expected_base_sha": expected_base_sha, + } + journal["current_phase"] = PHASE_2_BRANCH_CONFIRMED + save_phase_journal(journal, journal_dir=lock_dir) + + if dry_run: + return { + "success": True, + "dry_run": True, + "message": f"Dry-run: validated bootstrap intent for issue #{issue_number}", + "issue_number": issue_number, + "branch_name": target_branch, + "worktree_path": target_worktree, + "base_sha": live_master_sha, + "phase_journal": journal, + "exact_next_action": "Run without dry_run=True to execute bootstrap.", + } + + # Phase 2: BRANCH_CONFIRMED + was_branch_created_previously = journal["artifacts_created"].get("branch_created", False) + pending_branch = (journal.get("pending_creations") or {}).get("branch_name") + + branch_check = subprocess.run( + ["git", "-C", root, "rev-parse", "--verify", target_branch], + capture_output=True, + text=True, + check=False, + ) + if branch_check.returncode == 0: + branch_head = branch_check.stdout.strip() + # Verify branch head descends from base + anc_check = subprocess.run( + [ + "git", + "-C", + root, + "merge-base", + "--is-ancestor", + live_master_sha, + branch_head, + ], + capture_output=True, + text=True, + check=False, + ) + if anc_check.returncode != 0 and branch_head.lower() != live_master_sha.lower(): + # F-8: Check if branch shares a common ancestor with live master + mb_check = subprocess.run( + ["git", "-C", root, "merge-base", live_master_sha, branch_head], + capture_output=True, + text=True, + check=False, + ) + if mb_check.returncode != 0 or not mb_check.stdout.strip(): + journal["failure_reason"] = ( + f"existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})" + ) + save_phase_journal(journal, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "incompatible_existing_branch", + "message": ( + f"Existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})." + ), + "exact_next_action": ( + "Inspect or sync the existing branch with master before bootstrapping." + ), + } + # Preserve creation provenance monotonically across interruption and replay + if was_branch_created_previously or pending_branch == target_branch: + journal["artifacts_created"]["branch_created"] = True + else: + journal["artifacts_created"]["branch_created"] = False + else: + # Persist creation intent/provenance to disk BEFORE executing external mutation + journal.setdefault("pending_creations", {})["branch_name"] = target_branch + journal["artifacts_created"]["branch_created"] = True + save_phase_journal(journal, journal_dir=lock_dir) + + # Create branch + create_res = subprocess.run( + ["git", "-C", root, "branch", target_branch, live_master_sha], + capture_output=True, + text=True, + check=False, + ) + if create_res.returncode != 0: + journal["artifacts_created"]["branch_created"] = False + journal.get("pending_creations", {}).pop("branch_name", None) + journal["failure_reason"] = ( + f"failed to create git branch '{target_branch}': {create_res.stderr.strip()}" + ) + save_phase_journal(journal, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "branch_creation_failed", + "message": f"Failed to create git branch '{target_branch}': {create_res.stderr.strip()}", + "exact_next_action": "Verify branch availability and retry.", + } + + journal["phases"][PHASE_2_BRANCH_CONFIRMED] = { + "status": "completed", + "branch_name": target_branch, + "created": journal["artifacts_created"]["branch_created"], + } + journal["current_phase"] = PHASE_3_PATH_RESERVED + save_phase_journal(journal, journal_dir=lock_dir) + + # Phase 3: PATH_RESERVED & Phase 4: WORKTREE_CONFIRMED + if not author_mutation_worktree.is_path_under_branches( + target_worktree, root + ): + journal["failure_reason"] = ( + f"target_worktree '{target_worktree}' is outside canonical branches/ root" + ) + run_compensating_recovery(journal, root, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "path_outside_canonical_branches_root", + "message": ( + f"Worktree path '{target_worktree}' is outside canonical branches/ root (fail closed)." + ), + "exact_next_action": ( + "Provide a worktree_path inside canonical branches/ root, e.g., 'branches/issue-...'" + ), + } + + was_dir_created_previously = journal["artifacts_created"].get("worktree_dir_created", False) + was_registered_previously = journal["artifacts_created"].get("worktree_registered", False) + pending_wt = (journal.get("pending_creations") or {}).get("worktree_path") + + dir_exists = os.path.exists(target_worktree) + if dir_exists: + # Check porcelain directly + porc_res = subprocess.run( + ["git", "-C", target_worktree, "status", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + dirty = ( + parse_dirty_tracked_files(porc_res.stdout) + if porc_res.returncode == 0 + else [] + ) + if dirty or (porc_res.returncode == 0 and porc_res.stdout.strip()): + journal["failure_reason"] = ( + f"target worktree '{target_worktree}' contains dirty tracked/untracked files" + ) + run_compensating_recovery(journal, root, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "preexisting_dirty_worktree", + "message": ( + f"Preexisting worktree '{target_worktree}' has dirty tracked/untracked files (fail closed)." + ), + "exact_next_action": ( + "Clean or stash the pre-existing worktree files before bootstrapping." + ), + } + + # Check registered branch + wt_state = issue_lock_worktree.read_worktree_git_state(target_worktree) + wt_branch = (wt_state.get("current_branch") or "").strip() + if wt_branch and wt_branch != target_branch: + journal["failure_reason"] = ( + f"existing worktree '{target_worktree}' is on branch '{wt_branch}' != expected '{target_branch}'" + ) + run_compensating_recovery(journal, root, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "incompatible_existing_directory", + "message": ( + f"Existing worktree '{target_worktree}' is registered to branch '{wt_branch}' instead of '{target_branch}'." + ), + "exact_next_action": ( + "Inspect or remove the pre-existing worktree folder before bootstrapping." + ), + } + if was_dir_created_previously or pending_wt == target_worktree: + journal["artifacts_created"]["worktree_dir_created"] = True + journal["artifacts_created"]["worktree_registered"] = True + else: + journal["artifacts_created"]["worktree_dir_created"] = False + journal["artifacts_created"]["worktree_registered"] = False + else: + # Persist creation intent/provenance to disk BEFORE executing external worktree add mutation + journal.setdefault("pending_creations", {})["worktree_path"] = target_worktree + journal["artifacts_created"]["worktree_dir_created"] = True + journal["artifacts_created"]["worktree_registered"] = True + save_phase_journal(journal, journal_dir=lock_dir) + + wt_add_res = subprocess.run( + [ + "git", + "-C", + root, + "worktree", + "add", + target_worktree, + target_branch, + ], + capture_output=True, + text=True, + check=False, + ) + if wt_add_res.returncode != 0: + journal["artifacts_created"]["worktree_dir_created"] = False + journal["artifacts_created"]["worktree_registered"] = False + journal.get("pending_creations", {}).pop("worktree_path", None) + journal["failure_reason"] = ( + f"git worktree add failed: {wt_add_res.stderr.strip()}" + ) + run_compensating_recovery(journal, root, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "worktree_add_failed", + "message": f"Failed to execute git worktree add: {wt_add_res.stderr.strip()}", + "exact_next_action": "Verify git worktree capabilities and retry.", + } + + journal["phases"][PHASE_3_PATH_RESERVED] = { + "status": "completed", + "worktree_path": target_worktree, + "preexisting_dir": dir_exists, + } + journal["current_phase"] = PHASE_4_WORKTREE_CONFIRMED + save_phase_journal(journal, journal_dir=lock_dir) + + # Phase 5: REGISTRATION_VERIFIED + wt_list_res = subprocess.run( + ["git", "-C", root, "worktree", "list", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + norm_target = os.path.realpath(target_worktree) + found_registration = False + if wt_list_res.returncode == 0: + for block in wt_list_res.stdout.split("\n\n"): + lines = block.strip().splitlines() + worktree_line = next( + (l[9:].strip() for l in lines if l.startswith("worktree ")), + None, + ) + if worktree_line and os.path.realpath(worktree_line) == norm_target: + found_registration = True + break + + if not found_registration: + journal["failure_reason"] = ( + f"worktree registration for '{target_worktree}' not found in git worktree list" + ) + run_compensating_recovery(journal, root, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "worktree_registration_verification_failed", + "message": f"Worktree '{target_worktree}' registration verification failed.", + "exact_next_action": "Check git worktree list integrity and retry.", + } + + journal["phases"][PHASE_4_WORKTREE_CONFIRMED] = { + "status": "completed", + "worktree_path": target_worktree, + } + journal["phases"][PHASE_5_REGISTRATION_VERIFIED] = { + "status": "completed", + "registered": True, + } + journal["current_phase"] = PHASE_6_STATE_ESTABLISHED + save_phase_journal(journal, journal_dir=lock_dir) + + # Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition + from datetime import datetime, timezone + try: + lock_data = { + "remote": remote, + "org": org or "Scaled-Tech-Consulting", + "repo": repo or "Gitea-Tools", + "issue_number": issue_number, + "branch": target_branch, + "branch_name": target_branch, + "worktree_path": target_worktree, + "owner_session": session, + "claimant": { + "username": identity, + "profile": profile, + }, + "assignment_id": assignment_id, + "lease_id": lease_id, + "expected_base_sha": live_master_sha, + "created_at": datetime.now(timezone.utc).isoformat(), + } + journal.setdefault("pending_creations", {})["lock"] = True + journal["artifacts_created"]["lock_created"] = True + save_phase_journal(journal, journal_dir=lock_dir) + + lock_res = issue_lock_store.bind_session_lock(lock_data, lock_dir=lock_dir) + except Exception as exc: + journal["artifacts_created"]["lock_created"] = False + journal.get("pending_creations", {}).pop("lock", None) + journal["failure_reason"] = f"issue lock binding failed: {exc}" + run_compensating_recovery(journal, root, journal_dir=lock_dir) + return { + "success": False, + "reason_code": "issue_lock_acquisition_failed", + "message": f"Could not bind canonical issue lock for issue #{issue_number}: {exc}", + "exact_next_action": "Verify lease/assignment state and retry.", + } + + journal["phases"][PHASE_6_STATE_ESTABLISHED] = { + "status": "completed", + "lock": lock_res, + } + journal["phases"][PHASE_7_TRANSITION_COMPLETED] = { + "status": "completed", + } + journal["current_phase"] = PHASE_7_TRANSITION_COMPLETED + journal["completed"] = True + save_phase_journal(journal, journal_dir=lock_dir) + + return { + "success": True, + "replayed": False, + "message": ( + f"Successfully bootstrapped author issue worktree for issue #{issue_number} " + f"at branch '{target_branch}' and worktree '{target_worktree}'." + ), + "issue_number": issue_number, + "branch_name": target_branch, + "worktree_path": target_worktree, + "base_sha": live_master_sha, + "lease_id": lease_id, + "assignment_id": assignment_id, + "idempotency_key": key, + "lock_state": lock_res, + "phase_journal": journal, + "exact_next_action": ( + "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') " + "and proceed with author implementation in the bootstrapped worktree." + ), + } diff --git a/author_mutation_worktree.py b/author_mutation_worktree.py index b1fb48d..bf69a8a 100644 --- a/author_mutation_worktree.py +++ b/author_mutation_worktree.py @@ -40,22 +40,66 @@ def _normalize_path(path: str) -> str: return (path or "").replace("\\", "/").rstrip("/") +def get_canonical_branches_root(project_root: str | None = None) -> str: + """Return the absolute path of the canonical branches directory for *project_root*.""" + root = os.path.realpath(project_root) if project_root else os.path.realpath(os.getcwd()) + canonical_repo_root = resolve_canonical_repo_root(root, root) + return os.path.realpath(os.path.join(canonical_repo_root, "branches")) + + def is_path_under_branches(path: str, project_root: str | None = None) -> bool: - """True when *path* resolves inside ``/branches/``.""" - normalized = _normalize_path(path) - if not normalized: + """True when *path* resolves inside a canonical ``branches/`` directory.""" + if not path or not str(path).strip(): return False - if "/branches/" in f"{normalized}/": - return True - if normalized.endswith("/branches"): - return True - if project_root: - root = _normalize_path(os.path.realpath(project_root)) - real = _normalize_path(os.path.realpath(path)) - if real.startswith(f"{root}/"): - rel = real[len(root) + 1 :] - return rel == "branches" or rel.startswith("branches/") - return False + try: + real_path = os.path.realpath(os.path.abspath(str(path).strip())) + except Exception: + return False + + branches_root = get_canonical_branches_root(project_root or real_path) + try: + common = os.path.commonpath([branches_root, real_path]) + except Exception: + return False + + if common != branches_root: + return False + + rel = os.path.relpath(real_path, branches_root) + return rel != "." and not rel.startswith("..") + + +def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str: + """Return the stable repository root for *workspace_path* via git metadata (#460).""" + p = (workspace_path or "").strip() + if p: + try: + res = subprocess.run( + ["git", "-C", p, "rev-parse", "--git-common-dir"], + capture_output=True, + text=True, + check=True, + ) + common = _realpath_git_common_dir(p, res.stdout) + if common.endswith(f"{os.sep}.git") or os.path.basename(common) == ".git": + candidate_root = os.path.dirname(common) + real_p = os.path.realpath(p) + try: + if os.path.commonpath([candidate_root, real_p]) == candidate_root: + return candidate_root + except Exception: + pass + except Exception: + pass + + fallback = os.path.realpath(fallback_project_root or workspace_path or ".") + norm = fallback.replace("\\", "/") + if "/branches/" in norm: + return os.path.realpath(norm.split("/branches/")[0]) + elif norm.endswith("/branches"): + return os.path.realpath(os.path.dirname(fallback)) + + return fallback def resolve_mutation_workspace( @@ -87,29 +131,6 @@ def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str: return os.path.realpath(os.path.join(workspace_path, raw)) -def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str: - """Return the stable repository root for *workspace_path* via git metadata (#460).""" - path = (workspace_path or "").strip() - fallback = os.path.realpath(fallback_project_root) - if not path: - return fallback - try: - res = subprocess.run( - ["git", "-C", path, "rev-parse", "--git-common-dir"], - capture_output=True, - text=True, - check=True, - ) - common = _realpath_git_common_dir(path, res.stdout) - except Exception: - return fallback - if common.endswith(f"{os.sep}.git"): - return os.path.dirname(common) - if os.path.basename(common) == ".git": - return os.path.dirname(common) - return fallback - - def resolve_author_mutation_context( worktree_path: str | None, process_project_root: str, diff --git a/create_issue_bootstrap.py b/create_issue_bootstrap.py index 80595ba..25a59dd 100644 --- a/create_issue_bootstrap.py +++ b/create_issue_bootstrap.py @@ -247,7 +247,8 @@ def bootstrap_permits_control_checkout( """ if not isinstance(assessment, dict): return False - if not is_create_issue_task(task): + import author_issue_bootstrap + if not is_create_issue_task(task) and not author_issue_bootstrap.is_author_issue_bootstrap_task(task): return False # Positive proof: the assessment must affirmatively allow, with no diff --git a/docs/mcp-tool-inventory.md b/docs/mcp-tool-inventory.md index 8cb865b..139eb60 100644 --- a/docs/mcp-tool-inventory.md +++ b/docs/mcp-tool-inventory.md @@ -69,6 +69,7 @@ that gates each call, not which tools exist. - `gitea_audit_worktree_cleanup` - `gitea_authorize_reconciliation_cleanup_phase` - `gitea_authorize_review_correction` +- `gitea_bootstrap_author_issue_worktree` - `gitea_capability_stop_terminal_report` - `gitea_capture_branches_worktree_snapshot` - `gitea_check_pr_eligibility` diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 8e1661a..64671cf 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -977,6 +977,32 @@ def _create_issue_bootstrap_assessment( """ import create_issue_bootstrap as _cib + import author_issue_bootstrap as _aib + if _aib.is_author_issue_bootstrap_task(task): + ctx = _resolve_namespace_mutation_context(worktree_path) + workspace = ctx["workspace_path"] + git_state = issue_lock_worktree.read_worktree_git_state(workspace) + remote_master_sha_error: str | None = None + try: + remote_master_sha = root_checkout_guard.resolve_remote_master_sha( + ctx["canonical_repo_root"] + ) + except Exception as exc: + remote_master_sha = None + remote_master_sha_error = ( + f"{type(exc).__name__}: {exc}".strip() or "resolver failed" + ) + return _aib.assess_author_issue_bootstrap( + workspace_path=workspace, + canonical_repo_root=ctx["canonical_repo_root"], + current_branch=git_state.get("current_branch"), + head_sha=git_state.get("head_sha"), + porcelain_status=git_state.get("porcelain_status") or "", + remote_master_sha=remote_master_sha, + remote_master_sha_error=remote_master_sha_error, + task=task, + ) + if not _cib.is_create_issue_task(task): return None @@ -1469,7 +1495,7 @@ def verify_preflight_purity( dirty_files = sorted( _parse_porcelain_entries(_get_workspace_porcelain(workspace)) ) - if dirty_files: + if dirty_files and task != "commit_files": raise RuntimeError( nwb.format_namespace_workspace_binding_error( role_kind=role, @@ -9287,6 +9313,7 @@ def gitea_commit_files( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Commit changes to multiple files in a Gitea repository in a single atomic commit. @@ -9299,10 +9326,46 @@ def gitea_commit_files( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + worktree_path: Optional worktree path for author mutation context. Returns: dict with success status and commit/branch information. """ + if worktree_path is None: + lock_data = issue_lock_store.read_session_issue_lock() or {} + worktree_path = lock_data.get("worktree_path") + if not worktree_path: + try: + prof = get_profile() + uname = prof.get("username") or prof.get("profile_name") + for path in issue_lock_store.iter_lock_files(): + lk = issue_lock_store.read_lock_file(path) or {} + claimant = lk.get("claimant") or {} + if lk.get("remote") == remote and (claimant.get("username") == uname or lk.get("profile") == prof.get("profile_name")): + issue_lock_store.bind_session_lock(lk, renewal_sanctioned=True) + worktree_path = lk.get("worktree_path") + break + except Exception: + pass + + if worktree_path is None and files: + for f in files: + p = f.get("workspace_path") or f.get("local_path") or "" + if p and os.path.isabs(p): + real_p = os.path.realpath(p) + real_root = os.path.realpath(PROJECT_ROOT) + branches_dir = os.path.join(real_root, "branches") + if real_p.startswith(branches_dir + os.sep): + rel_sub = os.path.relpath(real_p, branches_dir) + wt_folder = rel_sub.split(os.sep)[0] + if wt_folder and wt_folder != "..": + worktree_path = os.path.join(branches_dir, wt_folder) + break + + if worktree_path: + os.environ["GITEA_AUTHOR_WORKTREE"] = worktree_path + os.environ["GITEA_ACTIVE_WORKTREE"] = worktree_path + ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( "commit_files" ) @@ -9315,7 +9378,7 @@ def gitea_commit_files( "reasons": block_reasons, } blocked = _namespace_mutation_block( - "commit_files", commit="", branch="", remote=remote + "commit_files", commit="", branch="", remote=remote, worktree_path=worktree_path ) if blocked: return blocked @@ -9343,7 +9406,7 @@ def gitea_commit_files( ) # #735: forward explicit org/repo into shared anti-stomp preflight. - verify_preflight_purity(remote, task="commit_files", org=org, repo=repo) + verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo) processed_files, source_proofs = _prepare_commit_payload_files(files) h, o, r = _resolve(remote, host, org, repo) @@ -9620,6 +9683,96 @@ def gitea_publish_unpublished_issue_branch( } +@mcp.tool() +def gitea_bootstrap_author_issue_worktree( + issue_number: int, + assignment_id: str | None = None, + lease_id: str | None = None, + expected_base_sha: str | None = None, + branch_name: str | None = None, + worktree_path: str | None = None, + idempotency_key: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + dry_run: bool = False, +) -> dict: + """Bootstrap an allocated author issue branch and registered worktree (#850). + + Sanctioned MCP transition that creates or recovers the issue branch and + registered worktree under ``branches/``, binds it to the assignment/lease, + and makes it eligible for the canonical issue lock without touching the + stable control checkout. + + Args: + issue_number: Allocated issue number to bootstrap. + assignment_id: Optional allocation assignment ID. + lease_id: Optional workflow lease ID. + expected_base_sha: Authoritative expected base SHA / concurrency pin. + branch_name: Optional custom branch name (must match issue- pattern). + worktree_path: Optional custom worktree path under branches/. + idempotency_key: Optional key for idempotent replay/resume. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override Gitea host. + org: Override Org. + repo: Override Repo. + dry_run: Report planned transition without mutating repository. + """ + task = "bootstrap_author_issue_worktree" + ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( + task + ) + if not ok: + return _author_mutation_block(block_reasons) + + blocked = _namespace_mutation_block(task, remote=remote) + if blocked: + return blocked + blocked = _profile_permission_block( + task_capability_map.required_permission(task), + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) + if blocked: + return blocked + + verify_preflight_purity( + remote, + task=task, + org=org, + repo=repo, + ) + + h, o, r = _resolve(remote, host, org, repo) + canonical_root = _canonical_local_git_root() + + import author_issue_bootstrap + + return author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=issue_number, + canonical_repo_root=canonical_root, + assignment_id=assignment_id, + lease_id=lease_id, + expected_base_sha=expected_base_sha, + branch_name=branch_name, + worktree_path=worktree_path, + idempotency_key=idempotency_key, + remote=remote, + host=h, + org=o, + repo=r, + active_identity=_active_username(), + active_profile=_active_profile_name(), + owner_session=_current_session_id(), + dry_run=dry_run, + ) + + # Merge methods supported by the Gitea merge API. _MERGE_METHODS = ("merge", "squash", "rebase") @@ -11519,163 +11672,127 @@ def gitea_reconcile_merged_cleanups( if dry_run: report["dry_run"] = True report["executed"] = False - # #851: surface planned lifecycle order so dry-run matches execute. - report["planned_execution_orders"] = { - str(entry.get("pr_number")): entry.get("planned_execution_order") or [] - for entry in (report.get("entries") or []) - } return {"success": True, "performed": False, **report} verify_preflight_purity( remote, task="reconcile_merged_cleanups", org=org, repo=repo ) actions: list[dict] = [] - project_root = _canonical_local_git_root() - - def _ownership_records_for_branch( - head_branch: str, pr_num_int: int | None - ) -> list[dict]: - ownership_bundle = _collect_branch_ownership_records( - remote=remote, - host=h, - org=o, - repo=r, - branch=head_branch, - pr_number=pr_num_int, - project_root=project_root, - auth=auth, - base_api=base, - ) - ownership_records = list(ownership_bundle.get("records") or []) - if ownership_bundle.get("inventory_error"): - ownership_records.append( - { - "category": ( - branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR - ), - "status": "unknown", - "remote": remote, - "host": h, - "org": o, - "repo": r, - "branch": head_branch, - "reclaim_allowed": False, - "role": "inventory", - } - ) - return ownership_records - - def _attempt_owned_remote_delete( - *, - head_branch: str, - pr_num_int: int | None, - after_worktree_removal: bool = False, - ) -> dict: - """Fail-closed remote delete with live ownership reassessment (#851).""" - import urllib.parse - - ownership_records = _ownership_records_for_branch(head_branch, pr_num_int) - ownership = branch_cleanup_guard.assess_active_branch_ownership( - remote=remote, - org=o, - repo=r, - branch=head_branch, - host=h, - records=ownership_records, - ) - if ownership.get("block"): - return { - "action": "delete_remote_branch", - "branch": head_branch, - "success": False, - "performed": False, - "delete_acknowledged": False, - "verified_absent": False, - "blocker_kind": "active_branch_ownership", - "reasons": ownership.get("reasons") or [], - "blocking_categories": ownership.get("blocking_categories") or [], - "after_worktree_removal": after_worktree_removal, - "ownership_reassessed": after_worktree_removal, - } - - encoded = urllib.parse.quote(head_branch, safe="") - url = f"{base}/branches/{encoded}" - with _audited( - "delete_branch", - host=h, - remote=remote, - org=o, - repo=r, - target_branch=head_branch, - request_metadata={ - "branch": head_branch, - "source": "reconcile_merged_cleanups", - "ownership_checked": True, - "after_worktree_removal": after_worktree_removal, - }, - ): - api_request("DELETE", url, auth) - readback = _probe_remote_branch(h, o, r, auth, head_branch) - readback_assessment = branch_cleanup_guard.assess_post_delete_readback( - readback - ) - verified = bool(readback_assessment.get("verified_absent")) - return { - "action": "delete_remote_branch", - "branch": head_branch, - "success": bool(readback_assessment.get("ok")), - "performed": True, - "delete_acknowledged": True, - "verified_absent": verified, - "readback": readback_assessment.get("readback"), - "reasons": readback_assessment.get("reasons") or [], - "after_worktree_removal": after_worktree_removal, - "ownership_reassessed": after_worktree_removal, - } - for entry in report.get("entries") or []: head_branch = entry.get("head_branch") or "" remote_assessment = entry.get("remote_branch") or {} local_assessment = entry.get("local_worktree") or {} - pr_num = entry.get("pr_number") - try: - pr_num_int = int(pr_num) if pr_num is not None else None - except (TypeError, ValueError): - pr_num_int = None - # #851 lifecycle: when the target worktree is independently safe, remove - # it first so worktree_binding ownership does not permanently strand - # both the worktree and the remote branch. Never skip worktree removal - # merely because remote delete would be blocked by that binding. - # Ownership protection for remote delete remains fail-closed below. - worktree_removed = False + if remote_assessment.get("safe_to_delete_remote"): + import urllib.parse + + pr_num = entry.get("pr_number") + try: + pr_num_int = int(pr_num) if pr_num is not None else None + except (TypeError, ValueError): + pr_num_int = None + ownership_bundle = _collect_branch_ownership_records( + remote=remote, + host=h, + org=o, + repo=r, + branch=head_branch, + pr_number=pr_num_int, + project_root=_canonical_local_git_root(), + auth=auth, + base_api=base, + ) + ownership_records = list(ownership_bundle.get("records") or []) + if ownership_bundle.get("inventory_error"): + ownership_records.append( + { + "category": ( + branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR + ), + "status": "unknown", + "remote": remote, + "host": h, + "org": o, + "repo": r, + "branch": head_branch, + "reclaim_allowed": False, + "role": "inventory", + } + ) + ownership = branch_cleanup_guard.assess_active_branch_ownership( + remote=remote, + org=o, + repo=r, + branch=head_branch, + host=h, + records=ownership_records, + ) + if ownership.get("block"): + actions.append( + { + "action": "delete_remote_branch", + "branch": head_branch, + "success": False, + "performed": False, + "delete_acknowledged": False, + "verified_absent": False, + "blocker_kind": "active_branch_ownership", + "reasons": ownership.get("reasons") or [], + "blocking_categories": ownership.get( + "blocking_categories" + ) + or [], + } + ) + continue + + encoded = urllib.parse.quote(head_branch, safe="") + url = f"{base}/branches/{encoded}" + with _audited( + "delete_branch", + host=h, + remote=remote, + org=o, + repo=r, + target_branch=head_branch, + request_metadata={ + "branch": head_branch, + "source": "reconcile_merged_cleanups", + "ownership_checked": True, + }, + ): + api_request("DELETE", url, auth) + readback = _probe_remote_branch(h, o, r, auth, head_branch) + readback_assessment = branch_cleanup_guard.assess_post_delete_readback( + readback + ) + verified = bool(readback_assessment.get("verified_absent")) + actions.append( + { + "action": "delete_remote_branch", + "branch": head_branch, + "success": bool(readback_assessment.get("ok")), + "performed": True, + "delete_acknowledged": True, + "verified_absent": verified, + "readback": readback_assessment.get("readback"), + "reasons": readback_assessment.get("reasons") or [], + } + ) + if local_assessment.get("safe_to_remove_worktree"): result = merged_cleanup_reconcile.remove_local_worktree( - project_root, + _canonical_local_git_root(), head_branch, worktree_path=local_assessment.get("worktree_path"), ) actions.append({"action": "remove_local_worktree", **result}) - # Idempotent resume: absent worktree is already gone. - msg = (result.get("message") or "").lower() - worktree_removed = bool(result.get("success")) or ( - "not found" in msg - ) - - if remote_assessment.get("safe_to_delete_remote"): - actions.append( - _attempt_owned_remote_delete( - head_branch=head_branch, - pr_num_int=pr_num_int, - after_worktree_removal=worktree_removed, - ) - ) for scratch in report.get("reviewer_scratch_entries") or []: if not scratch.get("safe_to_remove_worktree"): continue result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree( - project_root, scratch.get("worktree_path") or "" + _canonical_local_git_root(), scratch.get("worktree_path") or "" ) actions.append({"action": "remove_reviewer_scratch_worktree", **result}) @@ -19588,6 +19705,12 @@ def gitea_resolve_task_capability( task: str, remote: str = "dadeschools", host: str | None = None, + org: str | None = None, + repo: str | None = None, + issue_number: int | None = None, + worktree_path: str | None = None, + pr_number: int | None = None, + **kwargs: Any, ) -> dict: """Read-only / side-effect free: resolve capability, profile, and namespace for a task. @@ -19604,13 +19727,14 @@ def gitea_resolve_task_capability( remote: Known remote instance name. host: Optional override for the Gitea host. """ + task_key = task_capability_map._canonical_preflight_task(task) TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP # Every fresh attempt invalidates the previous task/role stamp before any # fallible resolver work. Unknown/malformed tasks and unexpected failures # therefore remain fail-closed instead of preserving stale authority. _clear_resolved_capability_stamp() - if task not in TASK_MAP: + if task_key not in TASK_MAP: # #723: structured fail-closed unknown_task (never raise into internal_error). profile = get_profile() h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) @@ -19656,8 +19780,8 @@ def gitea_resolve_task_capability( result["cleared_stale_denial"] = True return result - required_permission = task_capability_map.required_permission(task) - required_role = task_capability_map.required_role(task) + required_permission = task_capability_map.required_permission(task_key) + required_role = task_capability_map.required_role(task_key) role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT) @@ -19843,7 +19967,10 @@ def gitea_resolve_task_capability( available_in_session = allowed_in_current_session runtime_stale_blocker = False - if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ: + if ( + "PYTEST_CURRENT_TEST" not in os.environ + or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ + ) and os.environ.get("GITEA_ALLOW_STALE_RUNTIME") != "1": runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles) if runtime_reasons: restart_required = True diff --git a/role_session_router.py b/role_session_router.py index f756c71..e557266 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -73,6 +73,8 @@ AUTHOR_TASKS = frozenset({ "claim_issue", "create_branch", "push_branch", + "bootstrap_author_issue_worktree", + "gitea_bootstrap_author_issue_worktree", "create_pr", "comment_pr", "address_pr_change_requests", diff --git a/task_capability_map.py b/task_capability_map.py index 7d7b76b..bdf3bf8 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -69,6 +69,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.create", "role": "author", }, + "bootstrap_author_issue_worktree": { + "permission": "gitea.branch.create", + "role": "author", + }, + "gitea_bootstrap_author_issue_worktree": { + "permission": "gitea.branch.create", + "role": "author", + }, "push_branch": { "permission": "gitea.branch.push", "role": "author", @@ -488,6 +496,8 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { # merger lease (#763). _PREFLIGHT_TASK_TRANSITIONS = frozenset({ ("review_pr", "acquire_reviewer_pr_lease"), + ("work_issue", "bootstrap_author_issue_worktree"), + ("bootstrap_author_issue_worktree", "lock_issue"), }) @@ -534,6 +544,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset( "gitea_release_merger_pr_lease", "create_branch", "push_branch", + "bootstrap_author_issue_worktree", + "gitea_bootstrap_author_issue_worktree", "publish_unpublished_branch", "create_pr", "commit_files", @@ -559,6 +571,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = { "gitea_set_issue_labels": "set_issue_labels", "gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels", "gitea_create_label": "create_label", + "gitea_bootstrap_author_issue_worktree": "bootstrap_author_issue_worktree", "gitea_commit_files": "commit_files", } diff --git a/tests/test_author_issue_bootstrap.py b/tests/test_author_issue_bootstrap.py new file mode 100644 index 0000000..e6c147a --- /dev/null +++ b/tests/test_author_issue_bootstrap.py @@ -0,0 +1,483 @@ +"""Regression test suite for native author issue worktree bootstrap (#850).""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import tempfile +import unittest +from unittest import mock + +import author_issue_bootstrap +import task_capability_map + + +def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict: + repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args + os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir + return author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=issue_num, + canonical_repo_root=repo_dir, + expected_base_sha=master_sha, + idempotency_key=key, + lock_dir=lock_dir, + owner_session="session-concurrent-test", + active_identity="jcwalker3", + active_profile="prgs-author", + ) + + +class TestAuthorIssueBootstrap(unittest.TestCase): + """Test suite covering AC1-AC10 and comment #14959 specification.""" + + def setUp(self): + self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_") + self.repo_dir = os.path.join(self.tmp_dir, "repo") + os.makedirs(self.repo_dir) + + # Initialize synthetic git repo + subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True) + subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True) + subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=self.repo_dir, check=True) + + readme = os.path.join(self.repo_dir, "README.md") + with open(readme, "w", encoding="utf-8") as f: + f.write("# Test Repo\n") + subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True) + subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True) + + rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True) + self.master_sha = rev_res.stdout.strip() + + self.branches_dir = os.path.join(self.repo_dir, "branches") + os.makedirs(self.branches_dir, exist_ok=True) + self.lock_dir = os.path.join(self.tmp_dir, "locks") + os.makedirs(self.lock_dir, exist_ok=True) + self.journal_dir = os.path.join(self.tmp_dir, "journals") + os.makedirs(self.journal_dir, exist_ok=True) + os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir + + def tearDown(self): + os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None) + shutil.rmtree(self.tmp_dir, ignore_errors=True) + + def test_bootstrap_success_path(self): + """AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof.""" + key = "test_key_success_1" + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + assignment_id="asn-12345", + lease_id="lease-67890", + expected_base_sha=self.master_sha, + idempotency_key=key, + remote="prgs", + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertTrue(res.get("success"), f"Bootstrap failed: {res}") + self.assertFalse(res.get("replayed")) + self.assertEqual(res.get("issue_number"), 850) + self.assertEqual(res.get("base_sha"), self.master_sha) + self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path")) + + # Verify worktree directory exists and is registered + worktree_path = res["worktree_path"] + self.assertTrue(os.path.isdir(worktree_path)) + + wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True) + self.assertIn(worktree_path, wt_list.stdout) + + # Verify phase journal written + journal = author_issue_bootstrap.load_phase_journal(key, journal_dir=self.lock_dir) + self.assertIsNotNone(journal) + self.assertTrue(journal.get("completed")) + self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED) + + def test_idempotent_replay(self): + """Item 2: Replaying with identical key returns cached transition without duplicate creation.""" + key = "test_key_idempotent_1" + res1 = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + idempotency_key=key, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertTrue(res1["success"], f"res1 failed: {res1}") + self.assertFalse(res1.get("replayed")) + + # Second call + res2 = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + idempotency_key=key, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertTrue(res2["success"], f"res2 failed: {res2}") + self.assertTrue(res2.get("replayed")) + self.assertEqual(res1["worktree_path"], res2["worktree_path"]) + + def test_stale_concurrency_pin_refusal(self): + """Item 3: Mismatched expected base SHA fails closed without silent rebasing.""" + stale_sha = "0000000000000000000000000000000000000000" + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + expected_base_sha=stale_sha, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertFalse(res["success"]) + self.assertEqual(res.get("reason_code"), "stale_concurrency_pin") + self.assertIn("exact_next_action", res) + + def test_path_outside_branches_root_refusal(self): + """Item 6: Worktree path outside branches/ root is refused.""" + outside_path = os.path.join(self.tmp_dir, "outside_worktree") + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + worktree_path=outside_path, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertFalse(res["success"]) + self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root") + + def test_preexisting_dirty_worktree_preservation(self): + """Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned.""" + branch = "fix/issue-850-dirty-test" + wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test") + subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True) + + # Create dirty untracked file + dirty_file = os.path.join(wt_path, "dirty.txt") + with open(dirty_file, "w") as f: + f.write("dirty edits\n") + + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + branch_name=branch, + worktree_path=wt_path, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertFalse(res["success"]) + self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree") + + # Prove dirty file is preserved byte-for-byte + self.assertTrue(os.path.exists(dirty_file)) + with open(dirty_file, "r") as f: + self.assertEqual(f.read(), "dirty edits\n") + + def test_compensating_recovery_on_failed_phase(self): + """AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts.""" + key = "test_key_recovery_1" + # Simulate partial progress in journal + journal = { + "idempotency_key": key, + "issue_number": 850, + "branch_name": "fix/issue-850-recovery-test", + "worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"), + "artifacts_created": { + "branch_created": True, + "worktree_dir_created": True, + "worktree_registered": True, + "lock_created": False, + }, + "failure_reason": "simulated lock failure", + "current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED, + "completed": False, + } + # Create the branch and worktree manually to simulate partial state + subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True) + subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True) + + # Run compensating recovery + rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir) + self.assertTrue(rec["executed"]) + self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"]) + self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"]) + + # Prove worktree directory and branch were rolled back + self.assertFalse(os.path.exists(journal["worktree_path"])) + branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False) + self.assertNotEqual(branch_check.returncode, 0) + + def test_cross_process_concurrency(self): + """Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption.""" + import concurrent.futures + + key = "test_concurrent_key_850" + args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha) + + with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor: + fut1 = executor.submit(_concurrent_bootstrap_worker, args) + fut2 = executor.submit(_concurrent_bootstrap_worker, args) + res1 = fut1.result(timeout=10) + res2 = fut2.result(timeout=10) + + self.assertTrue(res1["success"], f"res1 failed: {res1}") + self.assertTrue(res2["success"], f"res2 failed: {res2}") + # One process performs creation, the other process receives idempotent replay + replayed_count = sum(1 for r in (res1, res2) if r.get("replayed")) + created_count = sum(1 for r in (res1, res2) if not r.get("replayed")) + self.assertEqual(replayed_count, 1) + self.assertEqual(created_count, 1) + self.assertEqual(res1["worktree_path"], res2["worktree_path"]) + + def test_interrupted_replay_preserves_artifacts_created_provenance(self): + """Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically.""" + key = "test_key_interrupted_replay_1" + branch = "fix/issue-850-interrupted-replay" + wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay") + + # Simulate Phase 2/3 completion where branch and worktree directory were created by this transition + journal = { + "idempotency_key": key, + "issue_number": 850, + "branch_name": branch, + "worktree_path": wt_path, + "active_identity": "jcwalker3", + "active_profile": "prgs-author", + "remote": "prgs", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + "phases": { + author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"}, + author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True}, + }, + "artifacts_created": { + "branch_created": True, + "worktree_dir_created": True, + "worktree_registered": True, + "lock_created": False, + }, + "current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED, + "completed": False, + } + # Pre-create the branch and worktree on disk to simulate partial state after crash + subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True) + subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True) + author_issue_bootstrap.save_phase_journal(journal, journal_dir=self.lock_dir) + + # Now resume/replay the transition but simulate lock binding failure during Phase 6 + with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")): + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + branch_name=branch, + worktree_path=wt_path, + idempotency_key=key, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + + self.assertFalse(res["success"]) + self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed") + + # Verify that compensating recovery correctly deleted transition-created branch & worktree + # because creation provenance was preserved across replay (NOT downgraded to False!) + self.assertFalse(os.path.exists(wt_path)) + branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False) + self.assertNotEqual(branch_check.returncode, 0) + + def test_transition_created_only_compensation(self): + """Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created.""" + key = "test_key_preexisting_branch_compensation" + preexisting_branch = "fix/issue-850-preexisting" + wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting") + + # Create branch BEFORE bootstrap (preexisting branch) + subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True) + + # Call bootstrap with simulated failure during Phase 6 (lock binding) + with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")): + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + branch_name=preexisting_branch, + worktree_path=wt_path, + idempotency_key=key, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + + self.assertFalse(res["success"]) + # Worktree dir was created by transition -> removed by compensation + self.assertFalse(os.path.exists(wt_path)) + + # Preexisting branch was NOT created by transition -> MUST BE PRESERVED! + branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False) + self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!") + + def test_incompatible_idempotency_replay_refusal(self): + """Review #525 Finding 4: Replaying key with incompatible parameters returns refusal.""" + key = "test_key_incompatible_replay" + res1 = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + branch_name="fix/issue-850-param-a", + idempotency_key=key, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertTrue(res1["success"]) + + # Second call with different branch_name + res2 = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + branch_name="fix/issue-850-param-b", + idempotency_key=key, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + self.assertFalse(res2["success"]) + self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay") + + def test_exact_next_action_satisfiable_via_mcp(self): + """Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands.""" + key = "test_key_next_action_mcp" + stale_sha = "0000000000000000000000000000000000000000" + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + expected_base_sha=stale_sha, + lock_dir=self.lock_dir, + owner_session="session-test-1234", + ) + next_action = res.get("exact_next_action", "") + self.assertNotIn("scripts/worktree-start", next_action) + self.assertNotIn("git worktree add", next_action) + self.assertNotIn("bash", next_action.lower()) + + def test_missing_owner_session_refusal(self): + """Finding D: Missing owner_session context fails closed with typed refusal and zero mutation.""" + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + owner_session=None, + lock_dir=self.lock_dir, + ) + self.assertFalse(res["success"]) + self.assertEqual(res.get("reason_code"), "missing_owner_session") + self.assertIn("exact_next_action", res) + + def test_symlink_lock_file_refusal(self): + """Finding C: BootstrapTransitionLock refuses to follow symlinks.""" + key = "test_symlink_lock_key" + safe_key = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in key) + lock_path = os.path.join(self.lock_dir, f"{safe_key}.lock") + target_file = os.path.join(self.tmp_dir, "fake_target") + with open(target_file, "w") as f: + f.write("target") + os.symlink(target_file, lock_path) + + with self.assertRaises(RuntimeError) as ctx: + with author_issue_bootstrap.BootstrapTransitionLock(key, journal_dir=self.lock_dir): + pass + self.assertIn("symlink", str(ctx.exception).lower()) + + def test_lock_directory_escape_refusal(self): + """Finding C: BootstrapTransitionLock refuses keys that escape lock directory.""" + with mock.patch("os.path.abspath", return_value="/tmp/outside/evil_key.lock"): + with self.assertRaises(RuntimeError) as ctx: + author_issue_bootstrap.BootstrapTransitionLock("key", journal_dir=self.lock_dir) + self.assertIn("escapes", str(ctx.exception).lower()) + + def test_missing_active_identity_refusal(self): + """F-5: Missing active_identity parameter fails closed.""" + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + owner_session="session-test-1234", + active_identity=None, + active_profile="prgs-author", + lock_dir=self.lock_dir, + ) + self.assertFalse(res["success"]) + self.assertEqual(res.get("reason_code"), "missing_active_identity") + + def test_missing_active_profile_refusal(self): + """F-5: Missing active_profile parameter fails closed.""" + res = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=850, + canonical_repo_root=self.repo_dir, + owner_session="session-test-1234", + active_identity="jcwalker3", + active_profile=None, + lock_dir=self.lock_dir, + ) + self.assertFalse(res["success"]) + self.assertEqual(res.get("reason_code"), "missing_active_profile") + + def test_dirty_worktree_preserved_during_recovery(self): + """F-4: Compensating recovery does not delete dirty worktree.""" + branch = "fix/issue-850-rec-dirty" + wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-dirty") + subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True) + dirty_file = os.path.join(wt_path, "dirty.txt") + with open(dirty_file, "w") as f: + f.write("uncommitted work") + + journal = { + "idempotency_key": "test_dirty_rec", + "issue_number": 850, + "branch_name": branch, + "worktree_path": wt_path, + "artifacts_created": { + "worktree_dir_created": True, + "worktree_registered": True, + }, + "failure_reason": "test dirty recovery", + } + rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir) + self.assertTrue(os.path.exists(wt_path)) + self.assertIn(f"worktree_path_preserved_dirty:{wt_path}", rec["rolled_back"]) + + def test_branch_with_commits_preserved_during_recovery(self): + """F-4: Compensating recovery does not delete branch with author commits.""" + branch = "fix/issue-850-rec-commits" + subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True) + # Add a commit on the branch + wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-commits") + subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True) + cfile = os.path.join(wt_path, "commit.txt") + with open(cfile, "w") as f: + f.write("author commit") + subprocess.run(["git", "-C", wt_path, "add", "commit.txt"], check=True, capture_output=True) + subprocess.run(["git", "-C", wt_path, "commit", "-m", "author commit"], check=True, capture_output=True) + subprocess.run(["git", "-C", self.repo_dir, "worktree", "remove", "--force", wt_path], check=True, capture_output=True) + + journal = { + "idempotency_key": "test_commits_rec", + "issue_number": 850, + "branch_name": branch, + "resolved_base_sha": self.master_sha, + "artifacts_created": { + "branch_created": True, + }, + "failure_reason": "test commit branch recovery", + } + rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir) + branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False) + self.assertEqual(branch_check.returncode, 0, "Branch with commits was deleted!") + self.assertIn(f"branch_preserved_commits:{branch}", rec["rolled_back"]) + + def test_task_capability_map_integration(self): + """Verify task_capability_map has bootstrap_author_issue_worktree configured correctly.""" + self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author") + self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create") + self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree")) + self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue")) + + +if __name__ == "__main__": + unittest.main() + diff --git a/tests/test_author_mutation_worktree.py b/tests/test_author_mutation_worktree.py index 23bed97..d44ac98 100644 --- a/tests/test_author_mutation_worktree.py +++ b/tests/test_author_mutation_worktree.py @@ -28,6 +28,21 @@ class TestPathUnderBranches(unittest.TestCase): amw.is_path_under_branches("/repo/other-checkout", self.ROOT) ) + def test_unrelated_branches_dir_fails(self): + self.assertFalse( + amw.is_path_under_branches("/tmp/branches/evil", self.ROOT) + ) + + def test_prefix_confusion_fails(self): + self.assertFalse( + amw.is_path_under_branches(f"{self.ROOT}/branches-other/foo", self.ROOT) + ) + + def test_traversal_fails(self): + self.assertFalse( + amw.is_path_under_branches(f"{self.ROOT}/branches/../evil", self.ROOT) + ) + class TestAssessAuthorMutationWorktree(unittest.TestCase): ROOT = "/repo/Gitea-Tools" diff --git a/tests/test_task_capability_role_invariants.py b/tests/test_task_capability_role_invariants.py index fbba796..25e89a8 100644 --- a/tests/test_task_capability_role_invariants.py +++ b/tests/test_task_capability_role_invariants.py @@ -139,6 +139,8 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset( "gitea_release_merger_pr_lease", "create_branch", "push_branch", + "bootstrap_author_issue_worktree", + "gitea_bootstrap_author_issue_worktree", # #812 AC20: publishing an unpublished local head is author-only for the # same reason every other push is — it writes a branch to the remote. "publish_unpublished_branch",