"""Sanctioned pre-issue bootstrap for ``create_issue`` (#749). ``gitea_create_issue`` is a pure remote mutation: it creates a tracking issue and writes nothing to the local working tree. The issue-first gate forbids creating ``branches/issue--*`` before the issue number exists, while the #274 branches-only guard previously demanded that worktree first — a deadlock. This module defines a **narrow, phase-scoped** exemption: * Only tasks in :data:`CREATE_ISSUE_TASKS` may use it. * Only the **canonical control checkout** may be used (never an arbitrary directory, unrelated worktree, or foreign clone). * The control checkout must be clean, on an accepted base branch, and base-equivalent to live master when a remote tip is known. * Every post-creation author mutation keeps the ordinary ``branches/`` rule. The exemption cannot widen: unknown tasks, dirty roots, drifted HEADs, non-base branches, and non-control workspaces fall through to the existing fail-closed guards. """ from __future__ import annotations import os from typing import Any from author_mutation_worktree import BASE_BRANCHES, is_path_under_branches from reviewer_worktree import parse_dirty_tracked_files CREATE_ISSUE_TASKS = frozenset({"create_issue", "gitea_create_issue"}) # Satisfiable before an issue number exists — never names issue-. EXACT_NEXT_ACTION_BOOTSTRAP = ( "Restore the canonical control checkout to a clean accepted base branch " "(master/main/dev) that matches live master, with no tracked local edits " "and no detached HEAD. Re-resolve the exact create_issue task, then re-run " "gitea_create_issue from that clean control checkout. Do not create " "branches/issue--* worktrees, dummy directories, or borrow unrelated " "worktrees before the issue exists." ) EXACT_NEXT_ACTION_POST_CREATE = ( "After the issue exists: create a registered worktree under " "branches/issue--* from clean master, claim/lock the issue, set " "GITEA_AUTHOR_WORKTREE / worktree_path to that path, then continue author " "mutations from the issue-backed worktree only." ) def is_create_issue_task(task: str | None) -> bool: """True when *task* is the create_issue mutation (or tool alias).""" return (task or "").strip() in CREATE_ISSUE_TASKS def normalize_sha(value: str | None) -> str | None: """Normalize a Git object id for comparison, or ``None`` when unknown. Whitespace and case are the only permitted variation between two spellings of the same commit; anything else is a different commit. Empty and whitespace-only values normalize to ``None`` so an unknown tip can never compare equal to another unknown tip. """ normalized = (value or "").strip().lower() return normalized or None def assess_create_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 create_issue may proceed from the control checkout. Returns a structured assessment: * ``not_applicable`` — not a create_issue task, or workspace is already a ``branches/`` worktree (use ordinary guards). * ``allowed`` — create_issue bootstrap may proceed from this control root. * ``block`` — create_issue was attempted from control checkout but gates failed (dirty, wrong branch, base race, etc.). """ reasons: list[str] = [] 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 = is_path_under_branches(workspace, root) if root else False if not is_create_issue_task(task): return _result( not_applicable=True, allowed=False, block=False, reasons=["task is not create_issue"], workspace=workspace, root=root, branch=branch, dirty=dirty, under_branches=under_branches, ) # Registered branches/ worktrees keep the normal path (no bootstrap). if under_branches: return _result( not_applicable=True, allowed=False, block=False, reasons=["workspace is under branches/; ordinary #274 path applies"], workspace=workspace, root=root, branch=branch, dirty=dirty, under_branches=True, ) # Only the exact canonical control checkout is eligible. if not root or workspace != root: reasons.append( "create_issue bootstrap requires the canonical control checkout; " f"workspace '{workspace}' is not the repository root '{root or '(unknown)'}'" ) return _result( not_applicable=False, allowed=False, block=True, reasons=reasons, workspace=workspace, root=root, branch=branch, dirty=dirty, under_branches=False, exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP, ) if dirty: reasons.append( "create_issue bootstrap blocked: control checkout has tracked local " f"edits (dirty files: {', '.join(dirty)})" ) if not branch: reasons.append( "create_issue bootstrap blocked: control checkout is detached HEAD; " "expected an accepted base branch (master/main/dev)" ) elif branch not in BASE_BRANCHES: reasons.append( f"create_issue bootstrap blocked: control checkout branch '{branch}' " f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})" ) # #757 AC3/AC4: base-equivalence must be *proven*, never assumed. An # unknown tip on either side is not evidence of agreement, so a missing # local HEAD, an unresolvable live master, or a resolver failure all block. remote_tip = normalize_sha(remote_master_sha) local_tip = normalize_sha(head_sha) resolver_error = (remote_master_sha_error or "").strip() or None if not local_tip: reasons.append( "create_issue bootstrap blocked: control checkout HEAD SHA is " "unknown; base equivalence to live master cannot be proven " "(fail closed)" ) if resolver_error: reasons.append( "create_issue bootstrap blocked: live master tip could not be " f"resolved ({resolver_error}); base equivalence cannot be proven " "(fail closed)" ) elif not remote_tip: reasons.append( "create_issue bootstrap blocked: live master tip is unknown; base " "equivalence to live master cannot be proven (fail closed)" ) if remote_tip and local_tip and remote_tip != local_tip: reasons.append( "create_issue bootstrap blocked: control checkout HEAD does not match " f"live master (HEAD {local_tip[:12]}, live master {remote_tip[:12]})" ) if reasons: return _result( not_applicable=False, allowed=False, block=True, reasons=reasons, workspace=workspace, root=root, branch=branch or None, dirty=dirty, under_branches=False, exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP, local_head_sha=local_tip, remote_master_sha=remote_tip, ) return _result( not_applicable=False, allowed=True, block=False, reasons=[], workspace=workspace, root=root, branch=branch or None, dirty=dirty, under_branches=False, exact_next_action=EXACT_NEXT_ACTION_POST_CREATE, bootstrap_path="clean_canonical_control_checkout", local_head_sha=local_tip, remote_master_sha=remote_tip, ) def bootstrap_permits_control_checkout( assessment: Any, *, task: str | None, workspace_path: str | None, canonical_repo_root: str | None, ) -> bool: """Single interpretation of a bootstrap assessment (#757). Both author-mutation guards — the #274 branches-only enforcer and the #604 anti-stomp preflight — route their "may this workspace mutate" decision through this predicate, so the two can never reach opposite conclusions about identical evidence. Fail-closed by construction. Every proof obligation must be present and affirmative in *assessment*, and the assessment must describe the very workspace and canonical root being guarded. A missing, malformed, refused, incomplete, or contradictory assessment returns ``False``, which leaves the caller's ordinary block in force. ``assessment`` is server-derived only: it is produced by :func:`assess_create_issue_bootstrap` from inspected repository state. It is never accepted from an MCP tool argument, so no caller can assert eligibility it has not proven. """ if not isinstance(assessment, dict): return False if not is_create_issue_task(task): return False # Positive proof: the assessment must affirmatively allow, with no # competing refusal or not-applicable disposition recorded alongside it. if assessment.get("allowed") is not True: return False if assessment.get("proven") is not True: return False if assessment.get("block") is not False: return False if assessment.get("not_applicable") is not False: return False if assessment.get("reasons"): return False # Scope proof: only the create_issue bootstrap, only via the clean # canonical control checkout path. if assessment.get("task_scope") != "create_issue_only": return False if assessment.get("bootstrap_path") != "clean_canonical_control_checkout": return False # State proof: clean, and not a branches/ worktree (those keep #274). if assessment.get("dirty_files"): return False if assessment.get("under_branches") is not False: return False # Base-equivalence proof (#757 AC3/AC4): both tips must be recorded, # nonempty, and equal. Re-derived here rather than trusted from the # assessment's own flag, so a hand-built or truncated assessment cannot # assert agreement it never proved. if assessment.get("base_tips_verified") is not True: return False local_tip = normalize_sha(assessment.get("local_head_sha")) remote_tip = normalize_sha(assessment.get("remote_master_sha")) if not local_tip or not remote_tip or local_tip != remote_tip: return False # Binding proof: the assessment must describe *this* workspace and root, # and that workspace must be exactly the canonical control checkout. root = os.path.realpath(canonical_repo_root or "") workspace = os.path.realpath(workspace_path or root or ".") if not root or workspace != root: return False if assessment.get("canonical_repo_root") != root: return False if assessment.get("workspace_path") != workspace: return False return True def format_create_issue_bootstrap_error(assessment: dict[str, Any]) -> str: """RuntimeError / typed-block message for a failed bootstrap assessment.""" reasons = "; ".join( assessment.get("reasons") or ["create_issue bootstrap failed"] ) next_action = ( assessment.get("exact_next_action") or EXACT_NEXT_ACTION_BOOTSTRAP ) root = assessment.get("canonical_repo_root") or "(unknown)" workspace = assessment.get("workspace_path") or "(unknown)" return ( f"Create-issue bootstrap guard (#749): {reasons}. " f"canonical repository root: {root}; workspace: {workspace}. " f"exact_next_action: {next_action}" ) def _result( *, not_applicable: bool, allowed: bool, block: bool, reasons: list[str], workspace: str, root: str, branch: str | None, dirty: list[str], under_branches: bool, exact_next_action: str | None = None, bootstrap_path: str | None = None, local_head_sha: str | None = None, remote_master_sha: str | None = None, ) -> dict[str, Any]: # #757 AC3/AC4: equality is recorded only when BOTH tips are known, so a # consumer can never read agreement out of two missing values. base_tips_verified = bool( local_head_sha and remote_master_sha and local_head_sha == remote_master_sha ) return { "not_applicable": not_applicable, "allowed": allowed, "block": block, "proven": allowed and not block, "reasons": list(reasons), "workspace_path": workspace, "canonical_repo_root": root, "current_branch": branch, "dirty_files": list(dirty), "under_branches": under_branches, "exact_next_action": exact_next_action, "bootstrap_path": bootstrap_path, "task_scope": "create_issue_only", "local_head_sha": local_head_sha, "remote_master_sha": remote_master_sha, "base_tips_verified": base_tips_verified, }