diff --git a/create_issue_bootstrap.py b/create_issue_bootstrap.py new file mode 100644 index 0000000..74c8e15 --- /dev/null +++ b/create_issue_bootstrap.py @@ -0,0 +1,227 @@ +"""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 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, + 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))})" + ) + + remote_tip = (remote_master_sha or "").strip() or None + local_tip = (head_sha or "").strip() or None + 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, + ) + + 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", + ) + + +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, +) -> dict[str, Any]: + 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", + } diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index 213b661..ffe85aa 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -48,6 +48,16 @@ It extracts the issue-first, isolated-worktree, no-self-review, profile-safety, merge-cleanup, fail-closed, and recovery rules into a reusable package that can be adapted to other repositories. +### Sanctioned first mutation: `create_issue` from clean control (#749) + +Creating a tracking issue has no issue number yet, so no `branches/issue--*` +worktree can exist. The sanctioned path is: clean canonical control checkout +(accepted base branch, base-equivalent to live master, no tracked dirt) → +resolve exact `create_issue` → `gitea_create_issue`. After the issue exists, +all further author mutations require a registered issue-backed worktree and +lock. Do not improvise with dummy directories, borrowed worktrees, or pre-issue +worktrees. See `skills/llm-project-workflow/workflows/create-issue.md` §18a. + ## Principle: the profile is the role, not the LLM ```text diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index c1e8412..5e45fff 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -817,7 +817,11 @@ def _enforce_canonical_repository_root( ) -def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None: +def _enforce_branches_only_author_mutation( + worktree_path: str | None = None, + *, + task: str | None = None, +) -> None: """#274: author file/branch mutations must run from a branches/ worktree. Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr`` @@ -831,6 +835,12 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> genuine reconciler as an author and defeat this exemption. Keying off the real profile role as well preserves the exemption without weakening author blocking — an actual author profile classifies as ``author`` in both. + + #749: ``create_issue`` alone may proceed from a **clean** canonical control + checkout (pure remote mutation; no issue number exists yet for an + issue-backed worktree). Dirty roots, non-base branches, base races, and + every other author mutation keep the ordinary branches-only fail-closed + behaviour. """ if ( _effective_workspace_role() in nwb.NON_AUTHOR_ROLES @@ -845,10 +855,35 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> project_root=ctx["canonical_repo_root"], current_branch=git_state.get("current_branch"), ) - if assessment["block"]: - raise RuntimeError( - author_mutation_worktree.format_author_mutation_worktree_error(assessment) + if not assessment["block"]: + return + + # #749 create_issue bootstrap: narrow phase exemption only. + import create_issue_bootstrap as _cib + + remote_master_sha = None + try: + remote_master_sha = root_checkout_guard.resolve_remote_master_sha( + ctx["canonical_repo_root"] ) + except Exception: + remote_master_sha = None + bootstrap = _cib.assess_create_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, + task=task, + ) + if bootstrap.get("allowed"): + return + if bootstrap.get("block") and not bootstrap.get("not_applicable"): + raise RuntimeError(_cib.format_create_issue_bootstrap_error(bootstrap)) + raise RuntimeError( + author_mutation_worktree.format_author_mutation_worktree_error(assessment) + ) def _anti_stomp_in_test_mode() -> bool: @@ -1213,7 +1248,7 @@ def verify_preflight_purity( # Historical path: root + branches after purity-order when dirty paths live. _enforce_canonical_repository_root(worktree_path, remote=remote) _enforce_root_checkout_guard(worktree_path) - _enforce_branches_only_author_mutation(worktree_path) + _enforce_branches_only_author_mutation(worktree_path, task=task) _enforce_issue_scope_guard( worktree_path, task=task, @@ -1247,7 +1282,7 @@ def verify_preflight_purity( if production_active: _enforce_canonical_repository_root(worktree_path, remote=remote) _enforce_root_checkout_guard(worktree_path) - _enforce_branches_only_author_mutation(worktree_path) + _enforce_branches_only_author_mutation(worktree_path, task=task) _enforce_issue_scope_guard( worktree_path, task=task, @@ -1366,10 +1401,15 @@ def _enforce_issue_scope_guard( "mark_issue", } ) + # #749: create_issue is pre-ownership (no issue number / lock can exist yet). + import create_issue_bootstrap as _cib + + is_create_issue = _cib.is_create_issue_task(task) require_lock = bool(require_author_lock) or ( authorish and workflow_scope_guard.production_guards_forced() and role == "author" + and not is_create_issue ) assessment = workflow_scope_guard.assess_production_mutation_guards( workspace_path=workspace, @@ -1381,6 +1421,7 @@ def _enforce_issue_scope_guard( role_kind=role, require_author_lock=require_lock, in_test_mode=_preflight_in_test_mode(), + mutation_task=task, ) workflow_scope_guard.raise_if_blocked(assessment) @@ -1394,7 +1435,11 @@ def _production_guard_block_from_exc(exc: BaseException, **extra) -> dict | None kind = workflow_scope_guard.BLOCKER_PRODUCTION_GUARD if "root_diagnostic_edit" in text or "tracked source or test edits" in text: kind = workflow_scope_guard.BLOCKER_ROOT_DIAGNOSTIC_EDIT - elif "Branches-only mutation guard" in text or "stable control checkout" in text: + elif ( + "Branches-only mutation guard" in text + or "stable control checkout" in text + or "Create-issue bootstrap guard (#749)" in text + ): kind = workflow_scope_guard.BLOCKER_MISSING_WORKTREE elif "out-of-scope" in text or "locked to issue" in text: kind = workflow_scope_guard.BLOCKER_OUT_OF_SCOPE_ISSUE @@ -1405,10 +1450,15 @@ def _production_guard_block_from_exc(exc: BaseException, **extra) -> dict | None reasons=[text], **extra, ) - if "Branches-only mutation guard" in text: + if "Branches-only mutation guard" in text or "Create-issue bootstrap guard (#749)" in text: + # Prefer bootstrap next-action text when present (#749 AC4). + next_action = None + if "exact_next_action:" in text: + next_action = text.split("exact_next_action:", 1)[1].strip() return workflow_scope_guard.block_response( blocker_kind=workflow_scope_guard.BLOCKER_MISSING_WORKTREE, reasons=[text], + exact_next_action=next_action, **extra, ) if "Root checkout guard" in text: @@ -1650,6 +1700,7 @@ import issue_lock_worktree # noqa: E402 import issue_lock_provenance # noqa: E402 import issue_lock_store # noqa: E402 import issue_lock_adoption # noqa: E402 +import issue_lock_recovery # noqa: E402 import stacked_pr_support # noqa: E402 import merge_approval_gate # noqa: E402 import review_quarantine # noqa: E402 # #695 contaminated formal-review quarantine @@ -2081,6 +2132,7 @@ def _assess_issue_duplicate_gate( auth: str, locked_branch: str | None = None, phase: str, + recovered_owning_pr: dict | None = None, ) -> dict: open_prs, branch_names, claim_entry = _collect_issue_duplicate_context( h, o, r, auth, issue_number @@ -2092,6 +2144,7 @@ def _assess_issue_duplicate_gate( claim_entry=claim_entry, locked_branch=locked_branch, phase=phase, + recovered_owning_pr=recovered_owning_pr, ) @@ -3156,8 +3209,11 @@ def gitea_lock_issue( worktree_path, _canonical_local_git_root() ) h, o, r = _resolve(remote, host, org, repo) + existing_issue_lock = _load_existing_issue_lock( + remote=remote, org=o, repo=r, issue_number=issue_number + ) active_lease_block = issue_lock_store.assess_same_issue_lease_conflict( - _load_existing_issue_lock(remote=remote, org=o, repo=r, issue_number=issue_number), + existing_issue_lock, issue_number=issue_number, branch_name=branch_name, worktree_path=resolved_worktree, @@ -3199,6 +3255,83 @@ def gitea_lock_issue( org=org, repo=repo, ) + # ── Dead-session lock recovery assessment (#753) ── + # When the MCP session that took a lock exits, the lock goes non-live + # (stale by dead PID) even inside its lease TTL, and the branch it owns is + # ahead of its base by construction — so the base-equivalence gate below + # makes normal re-lock unreachable for every PR that already exists. + # + # This grants a waiver ONLY for that case, proven against the durable lock + # record plus live git/Gitea observation. A refused assessment never raises: + # it simply withholds the waiver, leaving the pre-existing guard to fail + # closed exactly as before. Recovery can only ever add permission. + recovery_assessment: dict | None = None + if ( + existing_issue_lock + and existing_issue_lock.get("issue_number") == issue_number + and not issue_lock_store.is_lease_live(existing_issue_lock) + ): + recovery_auth = _auth(h) + try: + recovery_branches = api_get_all( + f"{repo_api_url(h, o, r)}/branches", recovery_auth + ) + except Exception as exc: + raise RuntimeError( + f"Could not list branches to verify issue-lock recovery: {exc}" + ) + recovery_remote_head: str | None = None + recovery_candidates: list[str] = [] + for entry in recovery_branches: + entry_name = _branch_entry_name(entry) + if entry_name == branch_name: + recovery_remote_head = _branch_entry_commit_sha(entry) + if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number): + recovery_candidates.append(entry_name) + recovery_pr_head: str | None = None + recovery_pr_number: int | None = None + for pull in _list_open_pulls(h, o, r, recovery_auth): + pull_head = pull.get("head") or {} + if str(pull_head.get("ref") or "") == branch_name: + recovery_pr_head = pull_head.get("sha") + recovery_pr_number = pull.get("number") + break + recovery_claimant = _work_lease_claimant(h) + recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery( + existing_issue_lock, + issue_number=issue_number, + branch_name=branch_name, + worktree_path=resolved_worktree, + remote=remote, + org=o, + repo=r, + identity=recovery_claimant.get("username"), + profile=recovery_claimant.get("profile"), + current_branch=git_state.get("current_branch"), + porcelain_status=git_state.get("porcelain_status") or "", + head_sha=git_state.get("head_sha"), + remote_head_sha=recovery_remote_head, + pr_head_sha=recovery_pr_head, + pr_number=recovery_pr_number, + competing_live_locks=issue_lock_store.list_live_locks(), + candidate_branches=recovery_candidates, + current_pid=os.getpid(), + ) + + recovery_sanctioned = bool( + recovery_assessment and recovery_assessment.get("recovery_sanctioned") + ) + # #755: a sanctioned dead-session recovery always has an owning open PR — + # that is what makes it a recovery rather than a fresh claim. Carry the + # server-derived owning-PR evidence into the duplicate-work gate below so + # the PR this lock already owns is not mistaken for competing duplicate + # work. Withheld (None) unless recovery was granted, so the ordinary + # duplicate blocker is untouched on every other path. + recovered_owning_pr = ( + issue_lock_recovery.owning_pr_recovery_evidence(recovery_assessment) + if recovery_sanctioned + else None + ) lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( worktree_path=resolved_worktree, current_branch=git_state.get("current_branch"), @@ -3206,10 +3339,20 @@ def gitea_lock_issue( base_equivalent=git_state.get("base_equivalent"), inspected_git_root=git_state.get("inspected_git_root"), base_branch=git_state.get("base_branch"), + recovery_sanctioned=recovery_sanctioned, ) if lock_assessment["block"]: + reasons = list(lock_assessment.get("reasons") or []) + # Surface why recovery was unavailable, so a blocked caller sees the + # exact missing evidence instead of only the base-equivalence text. + if recovery_assessment and recovery_assessment.get("is_candidate"): + reasons.append( + issue_lock_recovery.format_recovery_refusal(recovery_assessment) + ) raise RuntimeError( - issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment) + issue_lock_worktree.format_issue_lock_worktree_error( + {**lock_assessment, "reasons": reasons} + ) ) auth = _auth(h) @@ -3221,6 +3364,7 @@ def gitea_lock_issue( auth=auth, locked_branch=branch_name, phase=issue_work_duplicate_gate.PHASE_LOCK, + recovered_owning_pr=recovered_owning_pr, ) if duplicate_gate.get("block"): raise ValueError("; ".join(duplicate_gate.get("reasons") or [ @@ -3272,6 +3416,14 @@ def gitea_lock_issue( } if stacked_approved: data["approved_stacked_base"] = stacked_approved + if recovery_sanctioned and recovery_assessment: + # #753 AC2/AC6: record that this claim was recovered after session + # death, with the prior and replacement session identity, so the + # takeover is auditable and never looks like an original claim. + data["dead_session_recovery"] = issue_lock_recovery.build_recovery_record( + recovery_assessment, + recovered_at=_work_lease_timestamp(_work_lease_now()), + ) lock_file_path = _save_issue_lock(data) lock_record = issue_lock_store.read_lock_file(lock_file_path) or data @@ -3305,6 +3457,14 @@ def gitea_lock_issue( "lock_freshness": freshness, "lock_proof": lock_proof, } + if recovery_sanctioned and recovery_assessment: + result["dead_session_recovery"] = data["dead_session_recovery"] + result["message"] = ( + f"Recovered the durable lock for issue #{issue_number} on branch " + f"'{branch_name}' after the owning MCP session (pid " + f"{recovery_assessment['evidence'].get('prior_session_pid')}) exited; " + "ownership evidence matched exactly (fail-closed check complete)." + ) if stacked_approved: result["approved_stacked_base"] = stacked_approved result["message"] = ( @@ -15499,46 +15659,151 @@ def _count_commits_behind( return None +def _matching_protection_rule(protections: Any, branch: str) -> dict | None: + """Select the branch-protection rule governing ``branch`` (#751).""" + if not isinstance(protections, list): + return None + for rule in protections: + if not isinstance(rule, dict): + continue + name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() + # Exact match or glob-ish contains for common patterns. + if name == branch or name in ("*", f"{branch}"): + return rule + # Fallback: any protection that mentions the base branch. + for rule in protections: + if not isinstance(rule, dict): + continue + name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() + if name and (branch in name or name.endswith(branch)): + return rule + return None + + +def _branch_protection_policy( + base_url: str, + auth: dict, + *, + base_branch: str, +) -> dict: + """Read the live branch-protection policy for ``base_branch`` (#751). + + Exposes both the current-base rule (``block_on_outdated_branch``) and the + status-check requirement (``enable_status_check`` / ``status_check_contexts``) + from the same payload, so the checks assessor can tell "no checks required" + apart from "required checks pending". + + ``determinable`` is False only when the policy genuinely could not be read + (missing branch or API failure) — never merely because no rule exists. A + successful read that finds no protection for the branch is authoritative + evidence that status checks are not required. + """ + policy: dict[str, Any] = { + "determinable": False, + "protection_found": False, + "requires_current_base": None, + "checks_enabled": None, + "required_contexts": [], + "base_branch": (base_branch or "").strip() or None, + } + branch = (base_branch or "").strip() + if not branch: + return policy + + def _apply(rule: dict) -> None: + policy["protection_found"] = True + if "block_on_outdated_branch" in rule: + policy["requires_current_base"] = bool( + rule.get("block_on_outdated_branch") + ) + if "enable_status_check" in rule: + policy["checks_enabled"] = bool(rule.get("enable_status_check")) + contexts = rule.get("status_check_contexts") + if isinstance(contexts, list): + policy["required_contexts"] = [ + str(ctx).strip() for ctx in contexts if str(ctx or "").strip() + ] + + try: + protections = api_request( + "GET", f"{base_url}/branch_protections", auth + ) + rule = _matching_protection_rule(protections, branch) + if rule is not None: + _apply(rule) + if not policy["protection_found"]: + # Branch payload may embed effective protection. + br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {} + prot = br.get("protection") or br.get("effective_branch_protection") or {} + if isinstance(prot, dict) and prot: + _apply(prot) + policy["determinable"] = True + except Exception: + # Genuine read failure — leave determinable False so callers fail closed. + return policy + + if not policy["protection_found"]: + # Authoritative absence: no protection governs the branch, so no status + # check is required by policy. + policy["checks_enabled"] = False + elif policy["checks_enabled"] is None: + # Protection exists but omits the status-check field entirely: Gitea + # only enforces contexts when the flag is set, so absence means off. + policy["checks_enabled"] = False + + return policy + + def _branch_protection_requires_current_base( base_url: str, auth: dict, *, base_branch: str, ) -> bool | None: - """Read Gitea branch protection ``block_on_outdated_branch`` when present.""" - branch = (base_branch or "").strip() - if not branch: - return None + """Read Gitea branch protection ``block_on_outdated_branch`` when present. + + Thin accessor over :func:`_branch_protection_policy`; return semantics are + unchanged (``None`` when the rule is absent or unreadable). + """ + policy = _branch_protection_policy(base_url, auth, base_branch=base_branch) + return policy.get("requires_current_base") + + +def _commit_checks_snapshot( + base_url: str, + auth: dict, + *, + sha: str, +) -> dict: + """Read the combined commit status *and its context collection* (#751). + + The combined ``state`` alone is not decisive: Gitea reports ``pending`` for + a commit with an empty status-context collection, which is indistinguishable + from executing CI unless the collection itself is inspected. + """ + snapshot: dict[str, Any] = { + "determinable": False, + "combined_state": None, + "statuses": [], + } + head = (sha or "").strip() + if not head: + return snapshot try: - # Prefer the named protection rule matching the base branch. - protections = api_request( - "GET", f"{base_url}/branch_protections", auth - ) or [] - if isinstance(protections, list): - for rule in protections: - if not isinstance(rule, dict): - continue - name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() - # Exact match or glob-ish contains for common patterns. - if name == branch or name in ("*", f"{branch}"): - if "block_on_outdated_branch" in rule: - return bool(rule.get("block_on_outdated_branch")) - # Fallback: any protection that mentions the base branch. - for rule in protections: - if not isinstance(rule, dict): - continue - name = (rule.get("branch_name") or rule.get("rule_name") or "").strip() - if branch in name or name.endswith(branch): - if "block_on_outdated_branch" in rule: - return bool(rule.get("block_on_outdated_branch")) - # Branch payload may embed effective protection. - br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {} - prot = br.get("protection") or br.get("effective_branch_protection") or {} - if isinstance(prot, dict) and "block_on_outdated_branch" in prot: - return bool(prot.get("block_on_outdated_branch")) + payload = api_request( + "GET", f"{base_url}/commits/{head}/status", auth + ) except Exception: - return None - return None + return snapshot + if payload is None: + return snapshot + snapshot["determinable"] = True + if not isinstance(payload, dict): + return snapshot + snapshot["combined_state"] = (payload.get("state") or "").strip().lower() or None + statuses = payload.get("statuses") + snapshot["statuses"] = statuses if isinstance(statuses, list) else [] + return snapshot def _prove_author_ownership_for_pr( @@ -15783,11 +16048,14 @@ def gitea_assess_pr_sync_status( # Fail closed on unknown behind-count only when SHAs differ. commits_behind = 0 if pr_head_sha == base_head_sha else None + # Single live read of the base-branch protection policy; it carries both + # the current-base rule and the status-check requirement (#751). + protection_policy = _branch_protection_policy( + base, auth, base_branch=base_branch + ) if branch_protection_requires_current_base is None: - branch_protection_requires_current_base = ( - _branch_protection_requires_current_base( - base, auth, base_branch=base_branch - ) + branch_protection_requires_current_base = protection_policy.get( + "requires_current_base" ) # When protection cannot be read, fail closed by requiring current base # whenever the PR is behind (safer for protected repos). Callers may pass @@ -15854,23 +16122,30 @@ def gitea_assess_pr_sync_status( except Exception: pass + # ── Live checks derivation (#751) ──────────────────────────────────── + # Classify from the actual status-context collection plus the live + # protection policy. The combined ``state`` is never treated as proof that + # CI is executing, because Gitea reports ``pending`` for an empty + # collection. ``checks_required`` is always derived from live evidence and + # is deliberately not a caller-supplied input, so no session can declare + # checks optional without proof. + checks_snapshot = _commit_checks_snapshot(base, auth, sha=pr_head_sha or "") + checks_classification = pr_sync_status.classify_commit_checks( + combined_state=checks_snapshot.get("combined_state"), + statuses=checks_snapshot.get("statuses"), + checks_enabled=protection_policy.get("checks_enabled"), + required_contexts=protection_policy.get("required_contexts"), + policy_determinable=bool(protection_policy.get("determinable")), + status_determinable=bool(checks_snapshot.get("determinable")), + ) + checks_required = bool(checks_classification.get("checks_required", True)) + caller_supplied_checks_status = checks_status if checks_status is None: - checks_status = "unknown" - # Combined status on PR head when Actions/status API is available. - if pr_head_sha: - try: - st = api_request( - "GET", - f"{base}/commits/{pr_head_sha}/status", - auth, - ) or {} - state = (st.get("state") or "").strip().lower() - if state: - checks_status = state - elif not st: - checks_status = "none" - except Exception: - checks_status = "unknown" + checks_status = checks_classification.get("checks_status") + elif checks_classification.get("checks_status") != pr_sync_status.CHECKS_UNKNOWN: + # Live evidence outranks a caller-supplied value; the override only + # applies when live status could not be classified at all. + checks_status = checks_classification.get("checks_status") assessment = pr_sync_status.assess_pr_sync_status( host=h, @@ -15891,9 +16166,26 @@ def gitea_assess_pr_sync_status( active_reviewer_lease=active_reviewer_lease, active_merger_lease=active_merger_lease, prepared_verdict_head_sha=prepared_verdict_head_sha, + checks_required=checks_required, ) assessment["remote"] = remote if remote in REMOTES else None assessment["base_branch"] = base_branch + # Evidence for the checks decision (#751) — no secrets, read-only. + assessment["checks_evidence"] = { + "combined_state": checks_classification.get("combined_state"), + "context_count": checks_classification.get("context_count"), + "observed_contexts": checks_classification.get("observed_contexts"), + "required_contexts": checks_classification.get("required_contexts"), + "missing_required_contexts": checks_classification.get( + "missing_required_contexts" + ), + "policy_determinable": checks_classification.get("policy_determinable"), + "status_determinable": checks_classification.get("status_determinable"), + "protection_found": protection_policy.get("protection_found"), + "checks_enabled": protection_policy.get("checks_enabled"), + "caller_supplied_checks_status": caller_supplied_checks_status, + "reasons": checks_classification.get("reasons"), + } assessment["success"] = True assessment["performed"] = False return assessment diff --git a/issue_lock_adoption.py b/issue_lock_adoption.py index 81a5a92..b6d52a4 100644 --- a/issue_lock_adoption.py +++ b/issue_lock_adoption.py @@ -85,6 +85,15 @@ def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool: return re.search(pattern, name) is not None +def branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool: + """Public accessor for the exact issue-marker match (#753). + + Dead-session lock recovery needs the same word-boundary matcher to detect + ambiguous branch claims, so it is exposed rather than reached into. + """ + return _branch_carries_issue_marker(branch_name, issue_number) + + def assess_own_branch_adoption( *, issue_number: int, diff --git a/issue_lock_recovery.py b/issue_lock_recovery.py new file mode 100644 index 0000000..3c2e5e5 --- /dev/null +++ b/issue_lock_recovery.py @@ -0,0 +1,451 @@ +"""Dead-session author issue-lock recovery (#753). + +A durable author issue lock records the PID of the MCP session that took it. +When that process exits, ``issue_lock_store.assess_lock_freshness`` classifies +the lock as ``stale`` (``live=False``) even while its lease is still within TTL, +so every ownership check that requires a *live* lock fails closed. + +Re-taking the lock through ``gitea_lock_issue`` is unreachable for real work: +``issue_lock_worktree.assess_issue_lock_worktree`` demands the worktree be +base-equivalent to ``master``/``main``/``dev``, and a branch that already +carries commits is ahead of its base by construction. The existing +``assess_expired_lock_reclaim`` affordance does not apply either, because +``assess_same_issue_lease_conflict`` only consults it once the lease has +*expired* — a dead PID under an unexpired lease never reaches it. + +This module is the pure evidence assessor for that one narrow case. It grants +recovery only when every element of durable ownership still matches exactly and +the recorded process is demonstrably dead. It never trusts caller assertions: +every field is compared against durable lock state or live observation supplied +by the caller. It performs no mutation and no network I/O. + +Recovery deliberately does **not** relax base-equivalence for brand-new issue +claims — only for a lock whose own prior record already proves the branch, +worktree, head, and author. +""" + +from __future__ import annotations + +import os +from typing import Any, Iterable, Mapping, Sequence + +from issue_lock_store import is_process_alive +from reviewer_worktree import parse_dirty_tracked_files + +# Outcome values +RECOVERY_SANCTIONED = "RECOVERY_SANCTIONED" +NO_CANDIDATE = "NO_CANDIDATE" +REFUSED = "REFUSED" + +# Durable fields a lock must carry before it can be considered at all. +REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path") + + +def _same_realpath(left: str | None, right: str | None) -> bool: + if not left or not right: + return False + try: + return os.path.realpath(left) == os.path.realpath(right) + except OSError: + return left == right + + +def _text(value: Any) -> str: + return str(value or "").strip() + + +def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]: + claimant = lock.get("claimant") + if not isinstance(claimant, Mapping): + lease = lock.get("work_lease") + claimant = lease.get("claimant") if isinstance(lease, Mapping) else None + return dict(claimant) if isinstance(claimant, Mapping) else {} + + +def _recorded_pid(lock: Mapping[str, Any]) -> Any: + pid = lock.get("session_pid") + if pid is None: + pid = lock.get("pid") + return pid + + +def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]: + """Names of durable fields that are missing or unusable.""" + missing: list[str] = [] + for field in REQUIRED_LOCK_FIELDS: + if not _text(lock.get(field)): + missing.append(field) + pid = _recorded_pid(lock) + if pid is None or _text(pid) == "": + missing.append("session_pid/pid") + else: + try: + if int(pid) <= 0: + missing.append("session_pid/pid") + except (TypeError, ValueError): + missing.append("session_pid/pid") + return missing + + +def assess_dead_session_lock_recovery( + existing_lock: Mapping[str, Any] | None, + *, + issue_number: int, + branch_name: str, + worktree_path: str, + remote: str, + org: str, + repo: str, + identity: str | None, + profile: str | None, + current_branch: str | None, + porcelain_status: str, + head_sha: str | None, + remote_head_sha: str | None, + pr_head_sha: str | None = None, + pr_number: int | None = None, + competing_live_locks: Sequence[Mapping[str, Any]] | None = None, + candidate_branches: Iterable[str] | None = None, + current_pid: int | None = None, +) -> dict[str, Any]: + """Decide whether a dead-session author lock may be natively recovered. + + Returns a dict with ``recovery_sanctioned`` (bool), ``outcome``, ``reasons`` + (why it was refused, or the positive proof when sanctioned), and + ``evidence`` (a redaction-safe record for auditing). + + ``NO_CANDIDATE`` means no recovery was attempted at all — there is no + existing lock, or the lock does not describe this issue. The caller must + treat that exactly as it treated the pre-#753 world. ``REFUSED`` means a + candidate existed but the evidence did not agree; the caller fails closed. + """ + reasons: list[str] = [] + evidence: dict[str, Any] = { + "issue_number": issue_number, + "branch_name": branch_name, + "worktree_path": worktree_path, + "remote": remote, + "org": org, + "repo": repo, + } + + if not existing_lock: + return _result( + NO_CANDIDATE, False, ["no existing durable lock for this issue"], evidence + ) + + lock = dict(existing_lock) + + # ── Candidate identification ──────────────────────────────────────────── + # Recovery only ever applies to a lock that already claims THIS issue. + # Anything else is not a recovery candidate and must not be reinterpreted. + if lock.get("issue_number") != issue_number: + return _result( + NO_CANDIDATE, + False, + [ + f"existing lock targets issue #{lock.get('issue_number')}, " + f"not #{issue_number}; not a recovery candidate" + ], + evidence, + ) + + # A malformed/incomplete durable record can never prove ownership. + missing = _malformed_reasons(lock) + if missing: + return _result( + REFUSED, + False, + [ + "durable lock record is incomplete and cannot prove ownership " + f"(missing/unusable: {', '.join(missing)})" + ], + evidence, + ) + + recorded_pid = _recorded_pid(lock) + evidence["prior_session_pid"] = recorded_pid + evidence["replacement_session_pid"] = ( + current_pid if current_pid is not None else os.getpid() + ) + + # ── Repository scope ──────────────────────────────────────────────────── + for field, expected in (("remote", remote), ("org", org), ("repo", repo)): + actual = _text(lock.get(field)) + if actual != _text(expected): + reasons.append( + f"lock {field} '{actual}' does not match requested '{_text(expected)}'" + ) + + # ── Branch identity ───────────────────────────────────────────────────── + locked_branch = _text(lock.get("branch_name")) + if locked_branch != _text(branch_name): + reasons.append( + f"lock branch '{locked_branch}' does not match requested " + f"'{_text(branch_name)}'" + ) + evidence["locked_branch"] = locked_branch + + # The worktree must actually be sitting on the locked branch. Without this + # a clean worktree parked elsewhere could stand in for the real work. + checked_out = _text(current_branch) + if not checked_out: + reasons.append( + "worktree is not on a named branch (detached HEAD); locked-branch " + "occupancy could not be proven" + ) + elif checked_out != locked_branch: + reasons.append( + f"worktree is on branch '{checked_out}', not the locked branch " + f"'{locked_branch}'" + ) + + # ── Worktree identity ─────────────────────────────────────────────────── + locked_worktree = _text(lock.get("worktree_path")) + if not _same_realpath(locked_worktree, worktree_path): + reasons.append( + f"lock worktree '{locked_worktree}' does not match declared " + f"'{_text(worktree_path)}'" + ) + evidence["locked_worktree_path"] = locked_worktree + + # ── Cleanliness (never waived) ────────────────────────────────────────── + dirty_files = parse_dirty_tracked_files(porcelain_status) + if dirty_files: + reasons.append( + "worktree has tracked local edits; recovery requires a clean " + f"worktree (dirty files: {', '.join(dirty_files)})" + ) + evidence["dirty_files"] = dirty_files + + # ── Head agreement: local == remote == PR ─────────────────────────────── + local_head = _text(head_sha) + remote_head = _text(remote_head_sha) + if not local_head: + reasons.append("local head SHA could not be determined") + if not remote_head: + reasons.append( + f"remote head for branch '{locked_branch}' could not be determined" + ) + if local_head and remote_head and local_head != remote_head: + reasons.append( + f"local head {local_head} does not match remote branch head {remote_head}" + ) + evidence["local_head"] = local_head or None + evidence["remote_head"] = remote_head or None + + pr_head = _text(pr_head_sha) + if pr_head: + evidence["pr_head"] = pr_head + evidence["pr_number"] = pr_number + if local_head and pr_head != local_head: + reasons.append( + f"open PR #{pr_number} head {pr_head} does not match local head " + f"{local_head}" + ) + + # ── Author identity ───────────────────────────────────────────────────── + claimant = _lock_claimant(lock) + locked_identity = _text(claimant.get("username")) + locked_profile = _text(claimant.get("profile")) + evidence["locked_identity"] = locked_identity or None + evidence["locked_profile"] = locked_profile or None + if not locked_identity or not locked_profile: + reasons.append( + "durable lock does not record a claimant identity/profile; " + "author ownership could not be proven" + ) + if not _text(identity) or not _text(profile): + reasons.append( + "active session identity/profile is unknown; author ownership " + "could not be proven" + ) + if locked_identity and _text(identity) and locked_identity != _text(identity): + reasons.append( + f"lock claimant '{locked_identity}' does not match active identity " + f"'{_text(identity)}'" + ) + if locked_profile and _text(profile) and locked_profile != _text(profile): + reasons.append( + f"lock profile '{locked_profile}' does not match active profile " + f"'{_text(profile)}'" + ) + + # ── The defining condition: the recorded owner must be dead ───────────── + prior_alive = is_process_alive(recorded_pid) + evidence["prior_pid_alive"] = prior_alive + if prior_alive: + reasons.append( + f"prior owner pid {recorded_pid} is still alive; this is not a " + "dead-session recovery" + ) + if current_pid is not None and recorded_pid is not None: + try: + if int(recorded_pid) == int(current_pid): + reasons.append( + "recorded pid is the current session; nothing to recover" + ) + except (TypeError, ValueError): + pass + + # ── Competing ownership ───────────────────────────────────────────────── + competing: list[dict[str, Any]] = [] + for entry in competing_live_locks or (): + if not isinstance(entry, Mapping): + continue + same_issue = entry.get("issue_number") == issue_number + same_branch = _text(entry.get("branch_name")) == locked_branch + if not (same_issue or same_branch): + continue + # The lock we are recovering is not competition with itself. + if ( + same_issue + and same_branch + and _same_realpath(_text(entry.get("worktree_path")), worktree_path) + ): + continue + competing.append( + { + "issue_number": entry.get("issue_number"), + "branch_name": entry.get("branch_name"), + "worktree_path": entry.get("worktree_path"), + "pid": entry.get("pid"), + } + ) + if competing: + described = ", ".join( + f"issue #{c['issue_number']} branch '{c['branch_name']}'" for c in competing + ) + reasons.append(f"competing live lock or lease exists ({described})") + evidence["competing_live_locks"] = competing + + # ── Ambiguous branch claims ───────────────────────────────────────────── + others = [ + name + for name in (candidate_branches or ()) + if _text(name) and _text(name) != locked_branch + ] + if others: + reasons.append( + "multiple branches claim this issue " + f"({', '.join(sorted(set(others)))}); ownership is ambiguous" + ) + evidence["other_candidate_branches"] = sorted(set(others)) + + if reasons: + return _result(REFUSED, False, reasons, evidence) + + return _result( + RECOVERY_SANCTIONED, + True, + [ + f"durable lock for issue #{issue_number} matches branch " + f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, " + f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead" + ], + evidence, + ) + + +def _result( + outcome: str, + sanctioned: bool, + reasons: list[str], + evidence: dict[str, Any], +) -> dict[str, Any]: + return { + "outcome": outcome, + "recovery_sanctioned": sanctioned, + "is_candidate": outcome != NO_CANDIDATE, + "reasons": reasons, + "evidence": evidence, + } + + +def owning_pr_recovery_evidence( + assessment: Mapping[str, Any] | None, +) -> dict[str, Any] | None: + """Server-derived proof of the open PR a sanctioned recovery already owns (#755). + + A dead-session recovery is, by construction, recovery of work that already + has an open PR — so the duplicate-work gate's linked-open-PR blocker would + otherwise discard every sanctioned recovery. This distils the completed + assessment into the minimum evidence that gate needs to tell "the PR this + lock already owns" apart from "a competing duplicate PR". + + Returns ``None`` unless recovery was actually granted and the assessment's + own evidence names exactly one owning PR whose head agrees with the local + and remote heads. Nothing here is caller-supplied: every field is copied + from evidence the assessor built out of durable lock state plus live + git/Gitea observation, so a caller cannot manufacture an exemption. + """ + if not isinstance(assessment, Mapping): + return None + if assessment.get("outcome") != RECOVERY_SANCTIONED: + return None + if not assessment.get("recovery_sanctioned"): + return None + + evidence = assessment.get("evidence") or {} + branch_name = _text(evidence.get("locked_branch")) + pr_head = _text(evidence.get("pr_head")) + local_head = _text(evidence.get("local_head")) + remote_head = _text(evidence.get("remote_head")) + raw_pr_number = evidence.get("pr_number") + + if raw_pr_number is None or not branch_name or not pr_head: + return None + # The assessor already required these to agree. Re-check, so a truncated or + # hand-built evidence map can never authorize an exemption. + if pr_head != local_head or pr_head != remote_head: + return None + try: + pr_number = int(raw_pr_number) + issue_number = int(evidence.get("issue_number")) + except (TypeError, ValueError): + return None + + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch_name, + "head_sha": pr_head, + } + + +def build_recovery_record( + assessment: Mapping[str, Any], + *, + recovered_at: str, +) -> dict[str, Any]: + """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6).""" + evidence = dict(assessment.get("evidence") or {}) + return { + "recovered": True, + "reason": "owning MCP session exited; durable ownership evidence matched", + "recovered_at": recovered_at, + "prior_session_pid": evidence.get("prior_session_pid"), + "replacement_session_pid": evidence.get("replacement_session_pid"), + "prior_pid_alive": evidence.get("prior_pid_alive"), + "branch_name": evidence.get("locked_branch"), + "worktree_path": evidence.get("locked_worktree_path"), + "local_head": evidence.get("local_head"), + "remote_head": evidence.get("remote_head"), + "pr_head": evidence.get("pr_head"), + "pr_number": evidence.get("pr_number"), + "identity": evidence.get("locked_identity"), + "profile": evidence.get("locked_profile"), + "proof": list(assessment.get("reasons") or []), + } + + +def format_recovery_refusal(assessment: Mapping[str, Any]) -> str: + """Single fail-closed message for a refused recovery attempt.""" + reasons = list(assessment.get("reasons") or []) or [ + "dead-session lock recovery evidence did not agree" + ] + return ( + "Dead-session issue-lock recovery refused: " + + "; ".join(reasons) + + " (fail closed)" + ) diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index 3c23134..d14f5a2 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -92,8 +92,19 @@ def assess_issue_lock_worktree( inspected_git_root: str | None = None, base_branch: str | None = None, base_branches: frozenset[str] | None = None, + recovery_sanctioned: bool = False, ) -> dict: - """Fail closed when lock preconditions are not met on the declared worktree.""" + """Fail closed when lock preconditions are not met on the declared worktree. + + ``recovery_sanctioned`` is set only when ``issue_lock_recovery`` has already + proven, from the durable lock itself, that this is a dead-session recovery of + an existing claim (#753): same issue, branch, worktree, author, and head, with + the recording process dead. In that one case the base-equivalence requirement + is waived, because a branch that already carries the work is ahead of its base + by construction and could never satisfy it. Every other precondition — + notably worktree cleanliness — still applies unchanged, and brand-new issue + claims keep the full base-equivalence requirement. + """ bases = base_branches or BASE_BRANCHES reasons: list[str] = [] path = (worktree_path or "").strip() @@ -111,7 +122,11 @@ def assess_issue_lock_worktree( f"(dirty files: {', '.join(dirty_files)})" ) - if base_equivalent is False: + if recovery_sanctioned: + # Base-equivalence intentionally not evaluated: ownership was proven + # against the durable lock record instead (#753). + pass + elif base_equivalent is False: reasons.append( "issue lock worktree must be base-equivalent to one of " f"{_base_list(bases)} before implementation work; inspected " @@ -139,6 +154,7 @@ def assess_issue_lock_worktree( inspected_git_root=inspected_git_root, base_branch=base_branch, base_equivalent=base_equivalent, + recovery_sanctioned=recovery_sanctioned, ) @@ -197,6 +213,7 @@ def _assessment( inspected_git_root: str | None = None, base_branch: str | None = None, base_equivalent: bool | None = None, + recovery_sanctioned: bool = False, ) -> dict: return { "proven": proven, @@ -208,6 +225,8 @@ def _assessment( "dirty_files": dirty_files, "base_branch": base_branch, "base_equivalent": base_equivalent, + "recovery_sanctioned": recovery_sanctioned, + "base_equivalence_waived": bool(recovery_sanctioned), } diff --git a/issue_work_duplicate_gate.py b/issue_work_duplicate_gate.py index fc70437..81350c7 100644 --- a/issue_work_duplicate_gate.py +++ b/issue_work_duplicate_gate.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, Mapping import issue_claim_heartbeat as claim_hb @@ -27,6 +27,116 @@ def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None: return claim_hb._linked_open_pr(issue_number, open_prs) +def _pr_links_issue(issue_number: int, pr: Mapping[str, Any]) -> bool: + """Same linkage rule ``claim_hb._linked_open_pr`` applies, per PR. + + ``_linked_open_pr`` only yields the *first* match, which cannot answer + "is there exactly one linked PR?" — a question the owning-PR exemption + below must answer before it can trust any of them. + """ + pattern = _issue_pattern(issue_number) + head = (pr.get("head") or {}).get("ref") or "" + text = f"{pr.get('title', '')} {pr.get('body', '')}".lower() + if pattern in head.lower(): + return True + return ( + f"closes #{int(issue_number)}" in text + or f"fixes #{int(issue_number)}" in text + ) + + +def _all_linked_open_prs( + issue_number: int, open_prs: list[dict] +) -> list[Mapping[str, Any]]: + return [pr for pr in (open_prs or []) if _pr_links_issue(issue_number, pr)] + + +def _assess_owning_pr_exemption( + issue_number: int, + *, + linked_open_prs: list[Mapping[str, Any]], + locked_branch: str | None, + recovered_owning_pr: Mapping[str, Any] | None, +) -> tuple[bool, list[str]]: + """Is the linked open PR provably the one a sanctioned recovery owns (#755)? + + ``recovered_owning_pr`` is produced by + ``issue_lock_recovery.owning_pr_recovery_evidence`` from a completed + server-side recovery assessment — it is never a caller-supplied field. + Every element is re-checked here against the live PR list this gate was + given, so a stale or partial token cannot widen the exemption. + + Returns ``(exempt, diagnostic_reasons)``. Diagnostics are only emitted when + a token was offered and rejected, so a blocked caller can see which element + of ownership disagreed. + """ + if not recovered_owning_pr: + return False, [] + + notes: list[str] = [] + token_issue = recovered_owning_pr.get("issue_number") + token_pr = recovered_owning_pr.get("pr_number") + token_branch = str(recovered_owning_pr.get("branch_name") or "").strip() + token_head = str(recovered_owning_pr.get("head_sha") or "").strip() + locked = (locked_branch or "").strip() + + if token_issue is not None and int(token_issue) != int(issue_number): + notes.append( + f"recovery evidence is for issue #{token_issue}, not " + f"#{issue_number} (no owning-PR exemption)" + ) + return False, notes + if not locked or not token_branch or locked != token_branch: + notes.append( + f"recovery evidence branch '{token_branch or 'unknown'}' does not " + f"match the branch being locked '{locked or 'unknown'}' " + "(no owning-PR exemption)" + ) + return False, notes + if len(linked_open_prs) != 1: + numbers = ", ".join( + f"#{pr.get('number')}" for pr in linked_open_prs + ) or "none" + notes.append( + f"{len(linked_open_prs)} open PRs link issue #{issue_number} " + f"({numbers}); recovery may only own exactly one " + "(no owning-PR exemption)" + ) + return False, notes + + only = linked_open_prs[0] + head_obj = only.get("head") or {} + only_number = only.get("number") + only_ref = str(head_obj.get("ref") or "").strip() + only_sha = str(head_obj.get("sha") or "").strip() + + if token_pr is None or only_number is None or int(only_number) != int(token_pr): + notes.append( + f"linked open PR #{only_number} is not the recovered owning PR " + f"#{token_pr} (no owning-PR exemption)" + ) + return False, notes + if only_ref != token_branch: + notes.append( + f"open PR #{only_number} head branch '{only_ref}' does not match " + f"the recovered branch '{token_branch}' (no owning-PR exemption)" + ) + return False, notes + if not token_head or not only_sha or only_sha != token_head: + notes.append( + f"open PR #{only_number} head {only_sha or 'unknown'} does not " + f"match the recovered head {token_head or 'unknown'} " + "(no owning-PR exemption)" + ) + return False, notes + + return True, [ + f"open PR #{only_number} is the exact PR already owned by the " + f"recovering lock for issue #{issue_number} (branch '{token_branch}', " + f"head {token_head}); not duplicate work" + ] + + def _matching_branches( issue_number: int, branch_names: list[str], @@ -52,8 +162,15 @@ def assess_work_issue_duplicate_gate( claim_entry: dict | None = None, locked_branch: str | None = None, phase: str = PHASE_LOCK, + recovered_owning_pr: Mapping[str, Any] | None = None, ) -> dict[str, Any]: - """Fail closed when duplicate work is already in flight for an issue.""" + """Fail closed when duplicate work is already in flight for an issue. + + ``recovered_owning_pr`` (#755) is server-derived evidence that a sanctioned + dead-session lock recovery already owns one specific open PR. It exempts + *only* that exact PR from the linked-open-PR blocker; every other duplicate + signal, and every mismatch, keeps failing closed. + """ reasons: list[str] = [] outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED prs = list(open_prs or []) @@ -61,12 +178,23 @@ def assess_work_issue_duplicate_gate( pattern = _issue_pattern(issue_number) linked = _linked_open_pr(issue_number, prs) + linked_open_prs = _all_linked_open_prs(issue_number, prs) + owning_pr_exempted = False + exemption_notes: list[str] = [] if linked: - reasons.append( - f"open PR #{linked.get('number')} already covers issue " - f"#{issue_number} (fail closed)" + owning_pr_exempted, exemption_notes = _assess_owning_pr_exemption( + issue_number, + linked_open_prs=linked_open_prs, + locked_branch=locked_branch, + recovered_owning_pr=recovered_owning_pr, ) - outcome = OUTCOME_DUPLICATE_PR_PREVENTED + if not owning_pr_exempted: + reasons.append( + f"open PR #{linked.get('number')} already covers issue " + f"#{issue_number} (fail closed)" + ) + reasons.extend(exemption_notes) + outcome = OUTCOME_DUPLICATE_PR_PREVENTED conflicting_branches = _matching_branches( issue_number, branches, locked_branch=locked_branch @@ -122,6 +250,9 @@ def assess_work_issue_duplicate_gate( "phase": phase, "outcome": outcome, "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"), + "linked_open_pr_count": len(linked_open_prs), + "owning_pr_recovery_exempted": owning_pr_exempted, + "owning_pr_recovery_notes": list(exemption_notes), "conflicting_branches": conflicting_branches, "claim_status": status or None, "reasons": reasons, diff --git a/pr_sync_status.py b/pr_sync_status.py index 546c42c..4d0e691 100644 --- a/pr_sync_status.py +++ b/pr_sync_status.py @@ -41,6 +41,186 @@ _DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", " UPDATE_STYLE_MERGE = "merge" _FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"}) +# ── Commit check classifications (#751) ────────────────────────────────── +# Gitea's *combined* commit status reports ``state: pending`` both when a real +# check is executing and when the status-context collection is empty. Reading +# ``state`` alone therefore cannot distinguish "CI is running" from "no CI +# exists", which permanently blocks a merge-ready PR that no check will ever +# report on. These classifications are derived from the actual context +# collection plus the live branch-protection policy. +CHECKS_SUCCESS = "success" +CHECKS_FAILURE = "failure" +CHECKS_PENDING = "pending" +CHECKS_NONE = "none" # configured/produced nothing +CHECKS_NOT_REQUIRED = "not_required" # protection does not require checks +CHECKS_MISSING_REQUIRED = "missing_required" # required contexts have no result +CHECKS_UNKNOWN = "unknown" # indeterminable — fail closed + +# Values that permit merge_now when checks are required. +_CHECKS_OK = frozenset({"success", "passed", "ok", "skipped", "not_required"}) + +# Raw per-context state vocabularies reported by Gitea. +_CTX_SUCCESS = frozenset({"success", "passed", "ok"}) +_CTX_FAILURE = frozenset({"failure", "failed", "error", "cancelled", "canceled"}) +_CTX_PENDING = frozenset({"pending", "running", "queued", "expected"}) +_CTX_SKIPPED = frozenset({"skipped", "neutral"}) + + +def _normalize_context_rows(statuses: Any) -> list[dict[str, str]]: + """Reduce a raw status collection to newest-wins ``{context, state}`` rows. + + Gitea returns the status collection newest-first, so the first row seen for + a context wins. Rows without a usable state are discarded rather than being + silently treated as passing. + """ + rows: list[dict[str, str]] = [] + seen: set[str] = set() + if not isinstance(statuses, list): + return rows + for raw in statuses: + if not isinstance(raw, dict): + continue + context = (raw.get("context") or raw.get("name") or "").strip() + state = (raw.get("status") or raw.get("state") or "").strip().lower() + if not state: + continue + key = context or f"__unnamed__{len(rows)}" + if key in seen: + continue + seen.add(key) + rows.append({"context": context, "state": state}) + return rows + + +def _aggregate_context_states(rows: list[dict[str, str]]) -> str: + """Fail-closed aggregate: failure > pending > unknown-state > success.""" + states = {row["state"] for row in rows} + if states & _CTX_FAILURE: + return CHECKS_FAILURE + if states & _CTX_PENDING: + return CHECKS_PENDING + unresolved = states - _CTX_SUCCESS - _CTX_SKIPPED + if unresolved: + # An unrecognized context state must never read as success. + return CHECKS_UNKNOWN + return CHECKS_SUCCESS + + +def classify_commit_checks( + *, + combined_state: str | None = None, + statuses: Any = None, + checks_enabled: bool | None = None, + required_contexts: Any = None, + policy_determinable: bool = True, + status_determinable: bool = True, +) -> dict[str, Any]: + """Classify head checks from live evidence (#751). + + ``combined_state`` is deliberately **not** authoritative: it is recorded for + observability but never used to infer that CI is executing. The context + collection and the live protection policy decide. + + Returns ``checks_status`` (one of the ``CHECKS_*`` values), the derived + ``checks_required`` flag, and structured ``reasons``. + """ + reasons: list[str] = [] + rows = _normalize_context_rows(statuses) + required = [ + str(ctx).strip() + for ctx in (required_contexts or []) + if str(ctx or "").strip() + ] + observed_combined = (combined_state or "").strip().lower() or None + + result: dict[str, Any] = { + "checks_status": CHECKS_UNKNOWN, + "checks_required": True, + "combined_state": observed_combined, + "context_count": len(rows), + "observed_contexts": [row["context"] for row in rows], + "required_contexts": required, + "missing_required_contexts": [], + "policy_determinable": bool(policy_determinable), + "status_determinable": bool(status_determinable), + "reasons": reasons, + } + + # Policy unreadable → never assume checks are optional. + if not policy_determinable: + reasons.append( + "branch-protection check policy could not be read; cannot prove " + "whether status checks are required (fail closed)" + ) + return result + + if checks_enabled is False: + result["checks_required"] = False + result["checks_status"] = CHECKS_NOT_REQUIRED + reasons.append( + "live branch protection does not require status checks for the base " + "branch; head status contexts do not gate merge" + ) + return result + + if checks_enabled is None: + reasons.append( + "branch-protection status-check requirement is indeterminate " + "(fail closed)" + ) + return result + + # Checks are required from here on. + if not status_determinable: + reasons.append( + "head commit status collection could not be read while branch " + "protection requires status checks (fail closed)" + ) + return result + + if required: + by_context = {row["context"]: row["state"] for row in rows if row["context"]} + missing = [ctx for ctx in required if ctx not in by_context] + if missing: + result["missing_required_contexts"] = missing + result["checks_status"] = CHECKS_MISSING_REQUIRED + reasons.append( + "branch protection requires status context(s) " + f"{', '.join(missing)} but no matching status result exists at " + "the head commit (fail closed)" + ) + return result + matched = [ + {"context": ctx, "state": by_context[ctx]} for ctx in required + ] + result["checks_status"] = _aggregate_context_states(matched) + reasons.append( + f"evaluated {len(matched)} required status context(s) from live " + "branch protection; unrelated contexts were ignored" + ) + return result + + # Status checks enabled with no specific required contexts configured. + if not rows: + result["checks_status"] = CHECKS_NONE + reasons.append( + "branch protection enables status checks but no status context was " + "produced for the head commit" + ) + if observed_combined in _CTX_PENDING: + reasons.append( + f"combined commit state '{observed_combined}' does not indicate " + "executing CI because the status-context collection is empty" + ) + return result + + result["checks_status"] = _aggregate_context_states(rows) + reasons.append( + f"aggregated {len(rows)} reported status context(s); branch protection " + "configures no explicit required-context list" + ) + return result + def _normalize_sha(value: str | None) -> str | None: text = (value or "").strip().lower() @@ -120,6 +300,7 @@ def assess_pr_sync_status( "branch_protection_requires_current_base": requires_current, "approval_at_current_head": approval_ok if approval_at_current_head is not None else None, "checks_status": checks, + "checks_required": bool(checks_required), "active_locks_and_leases": { "author_lock": bool(active_author_lock) if active_author_lock is not None else None, "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None, @@ -258,21 +439,41 @@ def assess_pr_sync_status( result["recommended_next_action"] = ACTION_BLOCKED return result - # ── Checks gate for merge_now ──────────────────────────────────────── - if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"): - if checks in ("pending", "running", "queued"): + # ── Checks gate for merge_now (#751) ───────────────────────────────── + # ``checks_required`` is derived from the live branch-protection policy by + # the production caller. When protection does not require status checks, + # head contexts cannot gate the merge and this whole gate is skipped. + if not checks_required: + reasons.append( + "live branch protection does not require status checks; head check " + f"state ({checks}) does not gate merge" + ) + elif checks not in _CHECKS_OK: + if checks in _CTX_PENDING: reasons.append(f"required checks are not finished (status={checks})") - result["recommended_next_action"] = ACTION_BLOCKED - return result - if checks in ("failure", "failed", "error", "cancelled"): + elif checks in _CTX_FAILURE: reasons.append(f"required checks failed (status={checks})") - result["recommended_next_action"] = ACTION_BLOCKED - return result - # unknown — fail closed when checks_required - if checks == "unknown": + elif checks == CHECKS_MISSING_REQUIRED: + reasons.append( + "branch protection configures required status context(s) but no " + "matching status result exists at the current head (fail closed)" + ) + elif checks == CHECKS_NONE: + reasons.append( + "branch protection requires status checks but no status context " + "was produced for the current head (fail closed); an empty " + "status collection is not executing CI" + ) + elif checks == CHECKS_UNKNOWN: reasons.append("checks status unknown (fail closed)") - result["recommended_next_action"] = ACTION_BLOCKED - return result + else: + # Unrecognized vocabulary must never fall through to merge_now. + reasons.append( + f"unrecognized checks status '{checks}' cannot prove required " + "checks passed (fail closed)" + ) + result["recommended_next_action"] = ACTION_BLOCKED + return result # ── Ready to merge without update ──────────────────────────────────── # Includes: current with approval; outdated when update is NOT required. diff --git a/skills/llm-project-workflow/workflows/create-issue.md b/skills/llm-project-workflow/workflows/create-issue.md index a0d287d..fd35377 100644 --- a/skills/llm-project-workflow/workflows/create-issue.md +++ b/skills/llm-project-workflow/workflows/create-issue.md @@ -450,6 +450,35 @@ If any gate fails, do not create the issue. Produce a recovery handoff or duplicate report. +### 18a. Sanctioned first-mutation path (#749) + +`gitea_create_issue` is a **pure remote mutation** (no local tree write). The +issue-first gate forbids creating `branches/issue--*` before the issue +number exists. Therefore the **only sanctioned first mutation** is: + +1. Read-only identity + capability + duplicate search from the control checkout. +2. Ensure the **canonical control checkout** is: + * the configured repository root for the requested remote/org/repo; + * on an accepted base branch (`master` / `main` / `dev`); + * base-equivalent to live master; + * clean (no tracked local edits); + * in runtime/master parity. +3. Resolve exact task `create_issue`, then call `gitea_create_issue` **from that + clean control checkout** (no `worktree_path` required for this step alone). +4. After the issue number exists: create a **registered** worktree under + `branches/issue--*`, claim/lock, and perform every subsequent author + mutation from that worktree only. + +**Forbidden improvisations (fail closed):** + +* `mkdir` dummy directories under `branches/` (#713) +* borrowing an unrelated pre-existing worktree +* creating a pre-issue worktree in violation of issue-first +* running create_issue from a dirty, drifted, detached, or non-canonical root + +Post-creation mutations (`lock_issue`, commit, push, `create_pr`, etc.) **never** +receive this bootstrap exemption. + ## 19. Issue commenting gate Before commenting on an existing issue, verify: diff --git a/tests/test_create_issue_bootstrap.py b/tests/test_create_issue_bootstrap.py new file mode 100644 index 0000000..4ec128b --- /dev/null +++ b/tests/test_create_issue_bootstrap.py @@ -0,0 +1,367 @@ +"""Regression tests for create_issue bootstrap (#749). + +TDD: these tests define the sanctioned first-mutation path for +``gitea_create_issue`` from a clean canonical control checkout, and prove +the exemption cannot widen to dirty roots, foreign clones, arbitrary +``branches/`` directories, or post-creation author mutations. +""" + +from __future__ import annotations + +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import create_issue_bootstrap as cib # noqa: E402 +import gitea_mcp_server as srv # noqa: E402 +import workflow_scope_guard as wsg # noqa: E402 + +FAKE_AUTH = {"Authorization": "token test-token"} +MASTER_SHA = "a" * 40 +STALE_SHA = "b" * 40 + +current_file_path = Path(__file__).resolve() +if "branches" in current_file_path.parts: + CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3]) +else: + CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1]) + + +class TestCreateIssueBootstrapAssessor(unittest.TestCase): + ROOT = "/repo/Gitea-Tools" + + def test_non_create_issue_task_not_applicable(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=self.ROOT, + canonical_repo_root=self.ROOT, + current_branch="master", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="lock_issue", + ) + self.assertTrue(res["not_applicable"]) + self.assertFalse(res["allowed"]) + self.assertFalse(res["block"]) + + def test_branches_worktree_not_applicable(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=f"{self.ROOT}/branches/issue-1-x", + canonical_repo_root=self.ROOT, + current_branch="fix/issue-1-x", + task="create_issue", + ) + self.assertTrue(res["not_applicable"]) + self.assertFalse(res["allowed"]) + + def test_clean_control_checkout_allowed(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=self.ROOT, + canonical_repo_root=self.ROOT, + current_branch="master", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="create_issue", + ) + self.assertFalse(res["not_applicable"]) + self.assertTrue(res["allowed"]) + self.assertFalse(res["block"]) + self.assertEqual(res["bootstrap_path"], "clean_canonical_control_checkout") + # Post-create next action must name issue-backed worktree after N exists. + self.assertIn("branches/issue--*", res["exact_next_action"]) + + def test_tool_alias_gitea_create_issue_allowed(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=self.ROOT, + canonical_repo_root=self.ROOT, + current_branch="main", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="gitea_create_issue", + ) + self.assertTrue(res["allowed"]) + + def test_dirty_control_checkout_blocked(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=self.ROOT, + canonical_repo_root=self.ROOT, + current_branch="master", + head_sha=MASTER_SHA, + porcelain_status=" M gitea_mcp_server.py\n", + remote_master_sha=MASTER_SHA, + task="create_issue", + ) + self.assertTrue(res["block"]) + self.assertFalse(res["allowed"]) + self.assertTrue(any("tracked local edits" in r for r in res["reasons"])) + # Pre-issue phase: next action must be satisfiable without inventing . + next_a = res["exact_next_action"] or "" + self.assertIn("clean accepted base branch", next_a) + self.assertIn("before the issue exists", next_a) + # Must not prescribe "bind branches/issue-" as the recovery step. + self.assertNotIn("Bind an issue-backed worktree", next_a) + + def test_stale_base_blocked(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=self.ROOT, + canonical_repo_root=self.ROOT, + current_branch="master", + head_sha=STALE_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="create_issue", + ) + self.assertTrue(res["block"]) + self.assertTrue(any("live master" in r for r in res["reasons"])) + + def test_non_base_branch_blocked(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=self.ROOT, + canonical_repo_root=self.ROOT, + current_branch="feat/something", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="create_issue", + ) + self.assertTrue(res["block"]) + + def test_detached_head_blocked(self): + res = cib.assess_create_issue_bootstrap( + workspace_path=self.ROOT, + canonical_repo_root=self.ROOT, + current_branch="", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="create_issue", + ) + self.assertTrue(res["block"]) + self.assertTrue(any("detached" in r for r in res["reasons"])) + + def test_foreign_workspace_blocked(self): + res = cib.assess_create_issue_bootstrap( + workspace_path="/other/clone", + canonical_repo_root=self.ROOT, + current_branch="master", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="create_issue", + ) + self.assertTrue(res["block"]) + self.assertTrue(any("canonical control checkout" in r for r in res["reasons"])) + + +class TestCreateIssueBootstrapIntegration(unittest.TestCase): + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "author" + srv._preflight_resolved_task = "create_issue" + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + self._orig_in_test = srv._preflight_in_test_mode + srv._preflight_in_test_mode = lambda: False + + def tearDown(self): + srv._preflight_in_test_mode = self._orig_in_test + srv._preflight_resolved_task = None + + def _git_state(self, branch="master", head=MASTER_SHA, porcelain=""): + return { + "current_branch": branch, + "head_sha": head, + "porcelain_status": porcelain, + } + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch( + "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, []), + ) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + @patch( + "gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", + return_value=MASTER_SHA, + ) + def test_clean_control_checkout_create_issue_succeeds( + self, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth + ): + mock_api.return_value = { + "number": 99, + "html_url": "https://gitea.example.com/issues/99", + } + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=self._git_state(), + ): + with patch( + "gitea_mcp_server._get_workspace_porcelain", return_value="" + ): + with patch( + "gitea_mcp_server._enforce_root_checkout_guard" + ): + # Anti-stomp / master parity: keep gates green. + with patch.object( + srv, + "_run_anti_stomp_preflight", + return_value=None, + ): + res = srv.gitea_create_issue( + title="Bootstrap issue from clean control", + body="Body text for content gate.", + ) + self.assertEqual(res.get("number"), 99) + mock_api.assert_called_once() + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch( + "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, []), + ) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + def test_dirty_control_checkout_create_issue_fails_closed( + self, _get_all, mock_api, _role, _ns, _prof, _auth + ): + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=self._git_state( + porcelain=" M author_mutation_worktree.py\n" + ), + ): + with patch( + "gitea_mcp_server._get_workspace_porcelain", + return_value=" M author_mutation_worktree.py\n", + ): + res = srv.gitea_create_issue( + title="Should fail on dirty root", + body="Body text for content gate.", + ) + self.assertFalse(res.get("success", True) and res.get("number")) + if isinstance(res, dict) and res.get("success") is False: + blob = " ".join(res.get("reasons") or []) + self.assertTrue( + "tracked local edits" in blob + or "dirty" in blob.lower() + or "control checkout" in blob.lower() + or res.get("blocker_kind") + ) + # Pre-issue phase must not demand issue- worktree. + next_a = res.get("exact_next_action") or "" + if next_a: + self.assertNotIn("issue--*", next_a) + mock_api.assert_not_called() + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch( + "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, []), + ) + def test_lock_issue_still_requires_branches_worktree(self, _role, _ns, _prof, _auth): + """Existing issue-backed mutations receive no exemption (#749 AC3/AC7).""" + srv._preflight_resolved_task = "lock_issue" + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=self._git_state(), + ): + with patch( + "gitea_mcp_server._get_workspace_porcelain", return_value="" + ): + with patch( + "gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", + return_value=MASTER_SHA, + ): + with patch.object( + srv, "_enforce_root_checkout_guard" + ): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity( + remote="prgs", + task="lock_issue", + ) + msg = str(ctx.exception) + self.assertTrue( + "Branches-only mutation guard" in msg + or "stable control checkout" in msg, + msg, + ) + self.assertIn("control checkout", msg) + + def test_workflow_scope_skips_missing_worktree_for_create_issue_clean_root(self): + """#683 root assessor must not block clean-root create_issue bootstrap.""" + res = wsg.assess_root_source_mutation( + workspace_path=CONTROL_CHECKOUT_ROOT, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + porcelain_status="", + role_kind="author", + mutation_task="create_issue", + ) + self.assertFalse(res["block"], res) + self.assertTrue(res.get("create_issue_bootstrap") or res["proven"]) + + def test_workflow_scope_still_blocks_clean_root_for_lock_issue(self): + res = wsg.assess_root_source_mutation( + workspace_path=CONTROL_CHECKOUT_ROOT, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + porcelain_status="", + role_kind="author", + mutation_task="lock_issue", + ) + self.assertTrue(res["block"]) + self.assertEqual(res["blocker_kind"], wsg.BLOCKER_MISSING_WORKTREE) + + def test_arbitrary_branches_directory_not_bootstrap(self): + """#713: mkdir fake under branches/ is not the bootstrap path.""" + fake = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "fake-mkdir-only") + res = cib.assess_create_issue_bootstrap( + workspace_path=fake, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + current_branch="master", + head_sha=MASTER_SHA, + porcelain_status="", + remote_master_sha=MASTER_SHA, + task="create_issue", + ) + # Under branches/ → not bootstrap; ordinary membership/registration applies. + self.assertTrue(res["not_applicable"]) + self.assertFalse(res["allowed"]) + + +class TestCreateIssueCapabilityAgreement(unittest.TestCase): + def test_map_and_alias_agree_on_create_issue(self): + import task_capability_map as tcm + + self.assertEqual( + tcm.required_permission("create_issue"), + "gitea.issue.create", + ) + # Tool alias must resolve to the same task contract. + alias = getattr(tcm, "TOOL_TASK_ALIASES", None) or getattr( + tcm, "TASK_ALIASES", None + ) + if alias is not None: + mapped = alias.get("gitea_create_issue") + if mapped is not None: + self.assertEqual(mapped, "create_issue") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_create_issue_workspace_guard.py b/tests/test_create_issue_workspace_guard.py index 43f516e..a61fb3a 100644 --- a/tests/test_create_issue_workspace_guard.py +++ b/tests/test_create_issue_workspace_guard.py @@ -51,29 +51,62 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase): "porcelain_status": "", }, ) - def test_create_issue_stable_checkout_rejected( + def test_create_issue_stable_checkout_bootstrap_allowed_when_clean( self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth, ): - # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that - # path is the stable control checkout (not under branches/), mutation must fail. + # #749: clean canonical control checkout is the sanctioned create_issue path. + mock_api.return_value = { + "number": 77, + "html_url": "https://gitea.example.com/issues/77", + } with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): - try: - res = srv.gitea_create_issue(title="Test issue", body="body text") - except RuntimeError as exc: - self.assertIn("stable control checkout", str(exc)) - else: - # #683: production guards return typed blockers at entrypoints - self.assertFalse(res.get("success")) - self.assertFalse(res.get("performed")) - blob = " ".join(res.get("reasons") or []) + " " + str( - res.get("blocker_kind") or "" - ) - self.assertTrue( - "stable control checkout" in blob - or "missing_issue_worktree" in blob - or "control checkout" in blob.lower() - ) - self.assertTrue(res.get("exact_next_action")) + with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""): + with patch.object(srv, "_run_anti_stomp_preflight", return_value=None): + with patch.object(srv, "_enforce_root_checkout_guard"): + res = srv.gitea_create_issue( + title="Test issue", body="body text for gate" + ) + self.assertEqual(res.get("number"), 77) + mock_api.assert_called_once() + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40) + @patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": "master", + "head_sha": "a" * 40, + "porcelain_status": " M dirty.py\n", + }, + ) + def test_create_issue_dirty_control_checkout_rejected( + self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth, + ): + # #749: dirty control checkout still fails closed (no bootstrap). + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch( + "gitea_mcp_server._get_workspace_porcelain", + return_value=" M dirty.py\n", + ): + try: + res = srv.gitea_create_issue(title="Test issue", body="body text") + except RuntimeError as exc: + self.assertTrue( + "tracked local edits" in str(exc) + or "dirty" in str(exc).lower() + or "bootstrap" in str(exc).lower() + or "control checkout" in str(exc).lower() + ) + else: + self.assertFalse(res.get("success", True) and res.get("number")) + blob = " ".join(res.get("reasons") or []) + self.assertTrue(blob or res.get("blocker_kind")) + mock_api.assert_not_called() @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._profile_permission_block", return_value=None) diff --git a/tests/test_issue_751_checks_assessor.py b/tests/test_issue_751_checks_assessor.py new file mode 100644 index 0000000..36ce374 --- /dev/null +++ b/tests/test_issue_751_checks_assessor.py @@ -0,0 +1,602 @@ +"""Regression coverage for the PR checks assessor defect (#751). + +Gitea's *combined* commit status reports ``state: pending`` both when a check is +executing and when the status-context collection is empty. The assessor read +``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head +had no status contexts — and never would — was routed to ``blocked`` forever. + +These tests pin the corrected semantics end to end: live branch protection +decides whether checks are required, and the actual context collection decides +what the checks say. +""" + +from __future__ import annotations + +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import pr_sync_status # noqa: E402 +from pr_sync_status import ( # noqa: E402 + ACTION_BLOCKED, + ACTION_MERGE_NOW, + ACTION_UPDATE_BRANCH_BY_MERGE, + CHECKS_FAILURE, + CHECKS_MISSING_REQUIRED, + CHECKS_NONE, + CHECKS_NOT_REQUIRED, + CHECKS_PENDING, + CHECKS_SUCCESS, + CHECKS_UNKNOWN, + assess_pr_sync_status, + classify_commit_checks, +) + + +def _sha(prefix: str) -> str: + return (prefix + "0" * 40)[:40] + + +PR_HEAD = _sha("aaaaaaaa") +BASE_HEAD = _sha("bbbbbbbb") + + +def _base_kwargs(**overrides): + data = { + "host": "gitea.prgs.cc", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + "pr_number": 751, + "pr_state": "open", + "source_branch": "fix/issue-751-checks-assessor", + "pr_head_sha": PR_HEAD, + "base_head_sha": BASE_HEAD, + "commits_behind": 0, + "mergeable": True, + "has_conflicts": False, + "branch_protection_requires_current_base": False, + "approval_at_current_head": True, + "checks_status": "success", + "active_author_lock": True, + "active_reviewer_lease": False, + "active_merger_lease": False, + } + data.update(overrides) + return data + + +def _reasons(result) -> str: + return " | ".join(result["reasons"]).lower() + + +class TestClassifyCommitChecks(unittest.TestCase): + """Pure classification from live evidence.""" + + def test_empty_collection_with_combined_pending_is_not_executing_ci(self): + # The exact PR #750 failure mode at the classification layer. + result = classify_commit_checks( + combined_state="pending", + statuses=[], + checks_enabled=True, + required_contexts=[], + ) + self.assertNotEqual(result["checks_status"], CHECKS_PENDING) + self.assertEqual(result["checks_status"], CHECKS_NONE) + self.assertEqual(result["context_count"], 0) + joined = " ".join(result["reasons"]).lower() + self.assertIn("empty", joined) + self.assertIn("does not indicate", joined) + + def test_protection_disables_status_checks(self): + result = classify_commit_checks( + combined_state="pending", + statuses=[], + checks_enabled=False, + required_contexts=[], + ) + self.assertFalse(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED) + + def test_no_required_contexts_aggregates_reported_contexts(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "success"}], + checks_enabled=True, + required_contexts=[], + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + + def test_required_checks_pending(self): + result = classify_commit_checks( + combined_state="pending", + statuses=[{"context": "build", "status": "pending"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_PENDING) + + def test_required_checks_failed(self): + result = classify_commit_checks( + combined_state="failure", + statuses=[{"context": "build", "status": "failure"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_FAILURE) + + def test_required_checks_successful(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "success"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + self.assertTrue(result["checks_required"]) + + def test_required_context_configured_with_no_matching_result(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "lint", "status": "success"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED) + self.assertEqual(result["missing_required_contexts"], ["build"]) + + def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self): + # An unrelated failing context must not fail a satisfied required set. + result = classify_commit_checks( + combined_state="failure", + statuses=[ + {"context": "build", "status": "success"}, + {"context": "optional-scan", "status": "failure"}, + ], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + # ...and a failing *required* context still fails despite passing extras. + failing = classify_commit_checks( + combined_state="success", + statuses=[ + {"context": "build", "status": "failure"}, + {"context": "optional-scan", "status": "success"}, + ], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(failing["checks_status"], CHECKS_FAILURE) + + def test_policy_unreadable_fails_closed(self): + result = classify_commit_checks( + combined_state=None, + statuses=[], + checks_enabled=None, + required_contexts=[], + policy_determinable=False, + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + self.assertIn("fail closed", " ".join(result["reasons"]).lower()) + + def test_malformed_policy_indeterminate_flag_fails_closed(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "success"}], + checks_enabled=None, + required_contexts=[], + policy_determinable=True, + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + + def test_status_collection_unreadable_fails_closed_when_required(self): + result = classify_commit_checks( + checks_enabled=True, + required_contexts=["build"], + status_determinable=False, + ) + self.assertTrue(result["checks_required"]) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + + def test_unrecognized_context_state_never_reads_as_success(self): + result = classify_commit_checks( + combined_state="success", + statuses=[{"context": "build", "status": "banana"}], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_UNKNOWN) + + def test_newest_wins_per_context(self): + # Gitea returns newest-first; the stale failure must not win. + result = classify_commit_checks( + combined_state="success", + statuses=[ + {"context": "build", "status": "success"}, + {"context": "build", "status": "failure"}, + ], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_SUCCESS) + + def test_malformed_status_rows_are_discarded_not_treated_as_passing(self): + result = classify_commit_checks( + combined_state="pending", + statuses=[{"context": "build"}, "not-a-dict", None], + checks_enabled=True, + required_contexts=["build"], + ) + self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED) + + +class TestChecksGateSemantics(unittest.TestCase): + """``assess_pr_sync_status`` routing and blocker reasons.""" + + def test_not_required_allows_merge_now_with_empty_checks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED), + checks_required=False, + ) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + self.assertTrue(result["approval_valid_for_merge"]) + self.assertFalse(result["checks_required"]) + self.assertIn("does not require status checks", _reasons(result)) + + def test_none_blocks_when_checks_required(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_NONE), + checks_required=True, + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("no status context", _reasons(result)) + self.assertIn("not executing ci", _reasons(result)) + + def test_missing_required_blocks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED), + checks_required=True, + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("no matching status result", _reasons(result)) + + def test_required_pending_blocks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("not finished", _reasons(result)) + + def test_required_failure_blocks(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("failed", _reasons(result)) + + def test_required_success_merges(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + + def test_unknown_blocks_when_required(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("unknown", _reasons(result)) + + def test_unrecognized_status_does_not_fall_through_to_merge(self): + # Regression: the previous gate only handled a fixed vocabulary and let + # anything else reach merge_now. + result = assess_pr_sync_status( + **_base_kwargs(checks_status="totally-unexpected"), + checks_required=True, + ) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertIn("unrecognized checks status", _reasons(result)) + + def test_checks_gate_does_not_bypass_approval_requirement(self): + result = assess_pr_sync_status( + **_base_kwargs( + checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False + ), + checks_required=False, + ) + self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + self.assertFalse(result["approval_valid_for_merge"]) + + def test_checks_gate_does_not_bypass_current_base_protection(self): + # Existing current-base behavior stays intact (#727). + result = assess_pr_sync_status( + **_base_kwargs( + checks_status=CHECKS_NOT_REQUIRED, + commits_behind=3, + branch_protection_requires_current_base=True, + ), + checks_required=False, + ) + self.assertEqual( + result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE + ) + + def test_checks_gate_does_not_bypass_conflicts(self): + result = assess_pr_sync_status( + **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False), + checks_required=False, + ) + self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + + +class TestBranchProtectionPolicy(unittest.TestCase): + """Live protection reader derives the status-check requirement.""" + + def setUp(self): + import gitea_mcp_server as gms + + self.gms = gms + + def _policy(self, responses): + def fake_api_request(method, url, auth, *args, **kwargs): + for fragment, payload in responses.items(): + if fragment in url: + if isinstance(payload, Exception): + raise payload + return payload + return None + + with patch.object(self.gms, "api_request", side_effect=fake_api_request): + return self.gms._branch_protection_policy( + "https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master" + ) + + def test_status_checks_enabled_with_contexts(self): + policy = self._policy({ + "branch_protections": [{ + "branch_name": "master", + "block_on_outdated_branch": True, + "enable_status_check": True, + "status_check_contexts": ["ci/build", " "], + }], + }) + self.assertTrue(policy["determinable"]) + self.assertTrue(policy["checks_enabled"]) + self.assertTrue(policy["requires_current_base"]) + self.assertEqual(policy["required_contexts"], ["ci/build"]) + + def test_status_checks_disabled(self): + policy = self._policy({ + "branch_protections": [{ + "branch_name": "master", + "enable_status_check": False, + }], + }) + self.assertTrue(policy["determinable"]) + self.assertFalse(policy["checks_enabled"]) + + def test_no_protection_rule_means_checks_not_required(self): + # PR #750's repository shape: protection list readable but empty. + policy = self._policy({"branch_protections": [], "branches/master": {}}) + self.assertTrue(policy["determinable"]) + self.assertFalse(policy["protection_found"]) + self.assertFalse(policy["checks_enabled"]) + self.assertIsNone(policy["requires_current_base"]) + + def test_protection_without_status_field_means_not_enabled(self): + policy = self._policy({ + "branch_protections": [{ + "branch_name": "master", + "block_on_outdated_branch": True, + }], + }) + self.assertTrue(policy["determinable"]) + self.assertFalse(policy["checks_enabled"]) + self.assertTrue(policy["requires_current_base"]) + + def test_api_failure_is_not_determinable(self): + policy = self._policy({ + "branch_protections": RuntimeError("boom"), + }) + self.assertFalse(policy["determinable"]) + self.assertIsNone(policy["checks_enabled"]) + + def test_missing_branch_is_not_determinable(self): + with patch.object(self.gms, "api_request", return_value=[]): + policy = self.gms._branch_protection_policy( + "https://example/api/v1/repos/o/r", {}, base_branch=" " + ) + self.assertFalse(policy["determinable"]) + + def test_requires_current_base_accessor_preserved(self): + def fake(method, url, auth, *a, **k): + if "branch_protections" in url: + return [{"branch_name": "master", + "block_on_outdated_branch": True}] + return None + + with patch.object(self.gms, "api_request", side_effect=fake): + value = self.gms._branch_protection_requires_current_base( + "https://example/api/v1/repos/o/r", {}, base_branch="master" + ) + self.assertTrue(value) + + +class TestCommitChecksSnapshot(unittest.TestCase): + def setUp(self): + import gitea_mcp_server as gms + + self.gms = gms + + def test_reads_state_and_context_collection(self): + payload = {"state": "pending", "statuses": [{"context": "b", + "status": "pending"}]} + with patch.object(self.gms, "api_request", return_value=payload): + snap = self.gms._commit_checks_snapshot( + "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD + ) + self.assertTrue(snap["determinable"]) + self.assertEqual(snap["combined_state"], "pending") + self.assertEqual(len(snap["statuses"]), 1) + + def test_empty_collection_recorded_as_determinable(self): + with patch.object( + self.gms, "api_request", return_value={"state": "pending", "statuses": []} + ): + snap = self.gms._commit_checks_snapshot( + "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD + ) + self.assertTrue(snap["determinable"]) + self.assertEqual(snap["statuses"], []) + + def test_api_failure_is_not_determinable(self): + with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")): + snap = self.gms._commit_checks_snapshot( + "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD + ) + self.assertFalse(snap["determinable"]) + + +class TestMcpWrapperForwarding(unittest.TestCase): + """The production MCP path must reach the derived ``checks_required``.""" + + def setUp(self): + import gitea_mcp_server as gms + + self.gms = gms + self.base = ( + "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools" + ) + + def _patches(self, fake_api_request, spy): + return [ + patch.object(self.gms, "_profile_operation_gate", return_value=None), + patch.object(self.gms, "_resolve", return_value=( + "gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")), + patch.object(self.gms, "_auth", return_value={"Authorization": "x"}), + patch.object(self.gms, "repo_api_url", return_value=self.base), + patch.object(self.gms, "api_request", side_effect=fake_api_request), + patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={ + "success": True, "approval_at_current_head": True}), + patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={ + "has_author_lock": True}), + patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy), + ] + + def _run(self, *, protections, status_payload, pr_number=750, + caller_checks_status=None): + captured = {} + real_assess = pr_sync_status.assess_pr_sync_status + + def spy(**kwargs): + captured.update(kwargs) + return real_assess(**kwargs) + + def fake_api_request(method, url, auth, *args, **kwargs): + if f"/pulls/{pr_number}" in url: + return { + "number": pr_number, + "state": "open", + "title": "t", + "body": "b", + "mergeable": True, + "head": {"sha": PR_HEAD, "ref": "fix/x"}, + "base": {"sha": BASE_HEAD, "ref": "master"}, + } + if "/branch_protections" in url: + if isinstance(protections, Exception): + raise protections + return protections + if "/branches/master" in url: + return {"commit": {"id": BASE_HEAD}} + if "/status" in url: + if isinstance(status_payload, Exception): + raise status_payload + return status_payload + if "/compare/" in url: + return {"total_commits": 0} + if "/comments" in url: + return [] + return None + + stack = self._patches(fake_api_request, spy) + for p in stack: + p.start() + try: + result = self.gms.gitea_assess_pr_sync_status( + pr_number=pr_number, remote="prgs", + checks_status=caller_checks_status, + ) + finally: + for p in reversed(stack): + p.stop() + return result, captured + + def test_derived_checks_required_is_forwarded(self): + result, captured = self._run( + protections=[{"branch_name": "master", "enable_status_check": True, + "status_check_contexts": ["ci/build"]}], + status_payload={"state": "pending", "statuses": [ + {"context": "ci/build", "status": "pending"}]}, + ) + self.assertIn("checks_required", captured) + self.assertTrue(captured["checks_required"]) + self.assertEqual(captured["checks_status"], CHECKS_PENDING) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + + def test_pr750_empty_status_with_no_protection_is_merge_ready(self): + # The exact reported failure: combined pending, zero contexts, and no + # branch protection requiring checks. + result, captured = self._run( + protections=[], + status_payload={"state": "pending", "statuses": []}, + ) + self.assertFalse(captured["checks_required"]) + self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED) + self.assertNotEqual(result["checks_status"], CHECKS_PENDING) + self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW) + + def test_protection_read_failure_blocks(self): + result, captured = self._run( + protections=RuntimeError("protection unavailable"), + status_payload={"state": "success", "statuses": []}, + ) + self.assertTrue(captured["checks_required"]) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + + def test_caller_supplied_status_cannot_mask_live_required_failure(self): + # A caller-declared "success" must not override live evidence. + result, captured = self._run( + protections=[{"branch_name": "master", "enable_status_check": True, + "status_check_contexts": ["ci/build"]}], + status_payload={"state": "failure", "statuses": [ + {"context": "ci/build", "status": "failure"}]}, + caller_checks_status="success", + ) + self.assertEqual(captured["checks_status"], CHECKS_FAILURE) + self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED) + self.assertEqual( + result["checks_evidence"]["caller_supplied_checks_status"], "success" + ) + + def test_checks_evidence_is_reported_without_secrets(self): + result, _ = self._run( + protections=[], + status_payload={"state": "pending", "statuses": []}, + ) + evidence = result["checks_evidence"] + self.assertEqual(evidence["context_count"], 0) + self.assertEqual(evidence["combined_state"], "pending") + self.assertFalse(evidence["protection_found"]) + blob = repr(result).lower() + self.assertNotIn("authorization", blob) + self.assertNotIn("token", blob) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_753_dead_pid_lock_recovery.py b/tests/test_issue_753_dead_pid_lock_recovery.py new file mode 100644 index 0000000..93a8a7c --- /dev/null +++ b/tests/test_issue_753_dead_pid_lock_recovery.py @@ -0,0 +1,364 @@ +"""Dead-session author issue-lock recovery (#753). + +Covers the narrow recovery path that lets an author re-acquire a durable lock +after the MCP session that recorded it exits, plus every rejection condition +that must keep failing closed. +""" + +import os +import subprocess +import sys +import unittest +from datetime import datetime, timedelta, timezone + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import issue_lock_recovery # noqa: E402 +import issue_lock_store # noqa: E402 +import issue_lock_worktree # noqa: E402 + +ISSUE = 4242 +BRANCH = f"fix/issue-{ISSUE}-demo" +WORKTREE = "/scratch/wt" +HEAD = "a" * 40 +OTHER_SHA = "b" * 40 +IDENTITY = "example-user" +PROFILE = "example-author" + + +def dead_pid() -> int: + """A PID that has certainly exited (spawned, then reaped).""" + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + return proc.pid + + +def future_ts(hours: int = 4) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +def make_lock(**overrides): + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": WORKTREE, + "remote": "prgs", + "org": "ExampleOrg", + "repo": "ExampleRepo", + "session_pid": dead_pid(), + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": BRANCH, + "worktree_path": WORKTREE, + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "expires_at": future_ts(), + }, + } + lock.update(overrides) + return lock + + +def assess(lock=None, **overrides): + kwargs = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": WORKTREE, + "remote": "prgs", + "org": "ExampleOrg", + "repo": "ExampleRepo", + "identity": IDENTITY, + "profile": PROFILE, + "current_branch": BRANCH, + "porcelain_status": "", + "head_sha": HEAD, + "remote_head_sha": HEAD, + "pr_head_sha": HEAD, + "pr_number": 99, + "competing_live_locks": [], + "candidate_branches": [BRANCH], + "current_pid": os.getpid(), + } + kwargs.update(overrides) + return issue_lock_recovery.assess_dead_session_lock_recovery( + make_lock() if lock is None else lock, **kwargs + ) + + +class TestDeadSessionRecoveryGranted(unittest.TestCase): + def test_dead_pid_with_exact_evidence_recovers(self): + result = assess() + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED) + + def test_recovery_still_granted_when_no_open_pr_exists(self): + # A locked branch need not have a PR yet; absence must not block. + result = assess(pr_head_sha=None, pr_number=None) + self.assertTrue(result["recovery_sanctioned"], result["reasons"]) + + def test_lease_expiry_is_not_required_for_recovery(self): + # The defining condition is PID death, not TTL expiry (the #601 gap). + lock = make_lock() + self.assertFalse(issue_lock_store.is_lease_expired(lock)) + self.assertFalse(issue_lock_store.assess_lock_freshness(lock)["live"]) + self.assertTrue(assess(lock)["recovery_sanctioned"]) + + +class TestDeadSessionRecoveryRefused(unittest.TestCase): + def assert_refused(self, result, needle): + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED) + self.assertTrue( + any(needle in reason for reason in result["reasons"]), + f"expected {needle!r} in {result['reasons']}", + ) + + def test_live_prior_pid_refused(self): + lock = make_lock(session_pid=os.getpid(), pid=os.getpid()) + # Distinct current pid so the refusal is attributable to liveness. + self.assert_refused(assess(lock, current_pid=os.getpid() + 1), "still alive") + + def test_different_author_identity_refused(self): + self.assert_refused( + assess(identity="someone-else"), "does not match active identity" + ) + + def test_different_profile_refused(self): + self.assert_refused( + assess(profile="other-profile"), "does not match active profile" + ) + + def test_different_branch_refused(self): + lock = make_lock(branch_name=f"fix/issue-{ISSUE}-other") + self.assert_refused(assess(lock), "does not match requested") + + def test_worktree_parked_on_another_branch_refused(self): + self.assert_refused(assess(current_branch="master"), "not the locked branch") + + def test_detached_head_worktree_refused(self): + self.assert_refused(assess(current_branch=None), "detached HEAD") + + def test_different_worktree_refused(self): + self.assert_refused( + assess(worktree_path="/scratch/elsewhere"), "does not match declared" + ) + + def test_dirty_worktree_refused(self): + self.assert_refused( + assess(porcelain_status=" M gitea_mcp_server.py\n"), "requires a clean" + ) + + def test_local_head_differing_from_remote_refused(self): + self.assert_refused( + assess(remote_head_sha=OTHER_SHA), "does not match remote branch head" + ) + + def test_pr_head_differing_refused(self): + self.assert_refused(assess(pr_head_sha=OTHER_SHA), "does not match local head") + + def test_missing_remote_head_refused(self): + self.assert_refused(assess(remote_head_sha=None), "remote head") + + def test_competing_live_lock_refused(self): + competing = [ + { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": "/scratch/other-wt", + "pid": os.getpid(), + } + ] + self.assert_refused( + assess(competing_live_locks=competing), "competing live lock" + ) + + def test_unrelated_live_lock_does_not_block(self): + unrelated = [ + { + "issue_number": 999, + "branch_name": "fix/issue-999-unrelated", + "worktree_path": "/scratch/unrelated", + "pid": os.getpid(), + } + ] + self.assertTrue(assess(competing_live_locks=unrelated)["recovery_sanctioned"]) + + def test_multiple_candidate_branches_refused(self): + self.assert_refused( + assess(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"]), + "multiple branches claim this issue", + ) + + def test_repository_scope_mismatch_refused(self): + self.assert_refused(assess(repo="OtherRepo"), "does not match requested") + + def test_malformed_lock_missing_worktree_refused(self): + lock = make_lock() + lock.pop("worktree_path") + self.assert_refused(assess(lock), "incomplete") + + def test_malformed_lock_missing_pid_refused(self): + lock = make_lock() + lock.pop("session_pid", None) + lock.pop("pid", None) + self.assert_refused(assess(lock), "incomplete") + + def test_lock_without_claimant_refused(self): + lock = make_lock() + lock["work_lease"] = dict(lock["work_lease"]) + lock["work_lease"].pop("claimant") + self.assert_refused(assess(lock), "claimant identity/profile") + + +class TestNotACandidate(unittest.TestCase): + def test_absent_lock_is_not_a_candidate(self): + result = issue_lock_recovery.assess_dead_session_lock_recovery( + None, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=WORKTREE, + remote="prgs", + org="ExampleOrg", + repo="ExampleRepo", + identity=IDENTITY, + profile=PROFILE, + current_branch=BRANCH, + porcelain_status="", + head_sha=HEAD, + remote_head_sha=HEAD, + ) + self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE) + self.assertFalse(result["recovery_sanctioned"]) + self.assertFalse(result["is_candidate"]) + + def test_lock_for_a_different_issue_is_not_a_candidate(self): + result = assess(make_lock(issue_number=7777)) + self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE) + self.assertFalse(result["recovery_sanctioned"]) + + +class TestWorktreeGateWaiver(unittest.TestCase): + def test_new_issue_claim_still_requires_base_equivalence(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status="", + base_equivalent=False, + ) + self.assertTrue(result["block"]) + self.assertFalse(result["base_equivalence_waived"]) + + def test_sanctioned_recovery_waives_base_equivalence(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status="", + base_equivalent=False, + recovery_sanctioned=True, + ) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["base_equivalence_waived"]) + + def test_recovery_never_waives_cleanliness(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status=" M gitea_mcp_server.py\n", + base_equivalent=False, + recovery_sanctioned=True, + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("tracked file edits" in reason for reason in result["reasons"]) + ) + + def test_unproven_base_equivalence_still_blocks_without_recovery(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=WORKTREE, + current_branch=BRANCH, + porcelain_status="", + base_equivalent=None, + ) + self.assertTrue(result["block"]) + + +class TestRecoveryRecordAndDownstream(unittest.TestCase): + def test_recovery_record_preserves_truthful_provenance(self): + assessment = assess() + prior = assessment["evidence"]["prior_session_pid"] + record = issue_lock_recovery.build_recovery_record( + assessment, recovered_at="2026-07-18T23:21:40Z" + ) + self.assertTrue(record["recovered"]) + self.assertEqual(record["prior_session_pid"], prior) + self.assertEqual(record["replacement_session_pid"], os.getpid()) + self.assertNotEqual( + record["prior_session_pid"], record["replacement_session_pid"] + ) + self.assertFalse(record["prior_pid_alive"]) + self.assertEqual(record["recovered_at"], "2026-07-18T23:21:40Z") + self.assertEqual(record["branch_name"], BRANCH) + self.assertEqual(record["local_head"], HEAD) + self.assertEqual(record["identity"], IDENTITY) + self.assertTrue(record["proof"]) + + def test_recovery_record_carries_no_secret_material(self): + record = issue_lock_recovery.build_recovery_record( + assess(), recovered_at="2026-07-18T23:21:40Z" + ) + blob = repr(record).lower() + for banned in ("token", "password", "authorization", "secret", "api_key"): + self.assertNotIn(banned, blob) + + def test_recovered_lock_satisfies_update_by_merge_ownership(self): + # After recovery the lock is rebound to the live session, so the + # ownership re-check used by gitea_update_pr_branch_by_merge passes. + assessment = assess() + recovered_lock = make_lock(session_pid=os.getpid(), pid=os.getpid()) + recovered_lock["dead_session_recovery"] = ( + issue_lock_recovery.build_recovery_record( + assessment, recovered_at="2026-07-18T23:21:40Z" + ) + ) + freshness = issue_lock_store.assess_lock_freshness(recovered_lock) + self.assertTrue(freshness["live"], freshness) + + verdict = issue_lock_store.verify_lock_for_mutation( + recovered_lock, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=WORKTREE, + ) + self.assertTrue(verdict["proven"], verdict["reasons"]) + self.assertFalse(verdict["block"]) + + def test_pre_recovery_lock_fails_ownership_check(self): + # Guards against a false positive above: the dead-PID lock must fail. + verdict = issue_lock_store.verify_lock_for_mutation( + make_lock(), + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=WORKTREE, + ) + self.assertTrue(verdict["block"]) + self.assertTrue(any("not live" in reason for reason in verdict["reasons"])) + + def test_no_manual_file_seeding_required(self): + # The whole decision is reachable from the durable record plus live + # observation; nothing is written to disk to reach a verdict. + self.assertTrue(assess()["recovery_sanctioned"]) + + def test_refusal_message_is_fail_closed(self): + message = issue_lock_recovery.format_recovery_refusal( + assess(porcelain_status=" M x.py\n") + ) + self.assertIn("fail closed", message) + self.assertIn("recovery refused", message.lower()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_755_owning_pr_recovery.py b/tests/test_issue_755_owning_pr_recovery.py new file mode 100644 index 0000000..d9fc617 --- /dev/null +++ b/tests/test_issue_755_owning_pr_recovery.py @@ -0,0 +1,609 @@ +import sys as _sys +from pathlib import Path as _Path +_sys.path.insert(0, str(_Path(__file__).resolve().parent)) +from mutation_profile_fixture import shared_mutation_env # noqa: E402 +"""Dead-session lock recovery when the issue already owns an open PR (#755). + +#753 added the recovery *assessor*, but the production ``gitea_lock_issue`` +path still rejected every sanctioned recovery: a dead-session lock is by +construction a lock for work that already has an open PR, and the #400 +duplicate-work gate blocked unconditionally on any linked open PR. These tests +drive the real MCP handler, not just the pure assessor, so that gap cannot +reopen. +""" +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import issue_lock_provenance # noqa: E402 +import issue_lock_recovery # noqa: E402 +import issue_lock_store # noqa: E402 +import mcp_server # noqa: E402 +from issue_work_duplicate_gate import ( # noqa: E402 + OUTCOME_DUPLICATE_PR_PREVENTED, + OUTCOME_DUPLICATE_WORK_NOT_PREVENTED, + PHASE_LOCK, + assess_work_issue_duplicate_gate, +) + +ISSUE = 4755 +BRANCH = f"fix/issue-{ISSUE}-owning-pr" +OTHER_BRANCH = f"fix/issue-{ISSUE}-competing" +HEAD = "c" * 40 +OTHER_HEAD = "d" * 40 +OWNING_PR = 4756 +OTHER_PR = 4757 +IDENTITY = "example-user" +PROFILE = "test-author-prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" + + +def dead_pid() -> int: + """A PID that has certainly exited (spawned, then reaped).""" + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + return proc.pid + + +def shifted_ts(hours: int = 4) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(hours=hours)) + .isoformat() + .replace("+00:00", "Z") + ) + + +def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE): + return { + "number": number, + "title": f"fix: something (Closes #{issue})", + "body": f"Closes #{issue}.", + "head": {"ref": ref, "sha": sha}, + } + + +def sanctioned_token( + issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD +): + """The evidence shape the server derives from a granted recovery.""" + return { + "issue_number": issue_number, + "pr_number": pr_number, + "branch_name": branch, + "head_sha": head, + } + + +# ───────────────────────── duplicate gate: exemption ───────────────────────── + + +class TestOwningPrExemptionGranted(unittest.TestCase): + def test_exact_owning_pr_is_not_duplicate_work(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED) + self.assertEqual(result["linked_open_pr"], OWNING_PR) + self.assertEqual(result["linked_open_pr_count"], 1) + + def test_unrelated_open_pr_alongside_owning_pr_is_ignored(self): + unrelated = { + "number": 999, + "title": "chore: unrelated", + "body": "no linkage", + "head": {"ref": "chore/unrelated", "sha": OTHER_HEAD}, + } + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[unrelated, owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assertFalse(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["linked_open_pr_count"], 1) + + +class TestOwningPrExemptionRefused(unittest.TestCase): + def assert_blocked(self, result): + self.assertTrue(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED) + + def test_no_recovery_evidence_keeps_ordinary_blocker(self): + self.assert_blocked( + assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + ) + ) + + def test_competing_pr_number_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(number=OTHER_PR)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_multiple_linked_open_prs_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + self.assertEqual(result["linked_open_pr_count"], 2) + + def test_different_branch_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(ref=OTHER_BRANCH)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_locked_branch_differing_from_evidence_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=OTHER_BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_different_head_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr(sha=OTHER_HEAD)], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + def test_evidence_for_another_issue_refused(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(issue_number=ISSUE + 1), + ) + self.assert_blocked(result) + + def test_missing_head_in_live_pr_refused(self): + pr = owning_pr() + pr["head"] = {"ref": BRANCH} + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[pr], + branch_names=[BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + self.assert_blocked(result) + + +class TestOrdinaryDuplicateBehaviorUnchanged(unittest.TestCase): + def test_clean_issue_still_passes(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[], + branch_names=["feat/other-issue-99"], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + ) + self.assertFalse(result["block"]) + self.assertFalse(result["owning_pr_recovery_exempted"]) + + def test_competing_branch_still_blocks_even_with_owning_pr_evidence(self): + result = assess_work_issue_duplicate_gate( + ISSUE, + open_prs=[owning_pr()], + branch_names=[BRANCH, OTHER_BRANCH], + claim_entry={"status": "not_claimed"}, + locked_branch=BRANCH, + phase=PHASE_LOCK, + recovered_owning_pr=sanctioned_token(), + ) + # The owning PR is exempt, but the competing branch is not. + self.assertTrue(result["block"]) + self.assertTrue(result["owning_pr_recovery_exempted"]) + self.assertIn(OTHER_BRANCH, result["conflicting_branches"]) + + +# ─────────────────── server-derived evidence cannot be forged ─────────────────── + + +class TestOwningPrEvidenceDerivation(unittest.TestCase): + def granted(self, **evidence_overrides): + evidence = { + "issue_number": ISSUE, + "locked_branch": BRANCH, + "local_head": HEAD, + "remote_head": HEAD, + "pr_head": HEAD, + "pr_number": OWNING_PR, + } + evidence.update(evidence_overrides) + return { + "outcome": issue_lock_recovery.RECOVERY_SANCTIONED, + "recovery_sanctioned": True, + "is_candidate": True, + "reasons": [], + "evidence": evidence, + } + + def test_granted_recovery_yields_evidence(self): + token = issue_lock_recovery.owning_pr_recovery_evidence(self.granted()) + self.assertEqual(token, sanctioned_token()) + + def test_none_assessment_yields_nothing(self): + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(None)) + + def test_refused_assessment_yields_nothing(self): + refused = self.granted() + refused["outcome"] = issue_lock_recovery.REFUSED + refused["recovery_sanctioned"] = False + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(refused)) + + def test_sanctioned_flag_without_outcome_yields_nothing(self): + forged = self.granted() + forged["outcome"] = "SOMETHING_ELSE" + self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(forged)) + + def test_missing_pr_number_yields_nothing(self): + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(pr_number=None) + ) + ) + + def test_head_disagreement_in_evidence_yields_nothing(self): + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(remote_head=OTHER_HEAD) + ) + ) + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence( + self.granted(local_head=OTHER_HEAD) + ) + ) + + def test_real_refused_assessment_yields_nothing(self): + """End-to-end against the real assessor, not a hand-built dict.""" + lock = { + "issue_number": ISSUE, + "branch_name": BRANCH, + "worktree_path": "/scratch/wt", + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + "session_pid": os.getpid(), # alive → must refuse + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": BRANCH, + "worktree_path": "/scratch/wt", + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "expires_at": shifted_ts(), + }, + } + assessment = issue_lock_recovery.assess_dead_session_lock_recovery( + lock, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote="prgs", + org="Example-Org", + repo="Example-Repo", + identity=IDENTITY, + profile=PROFILE, + current_branch=BRANCH, + porcelain_status="", + head_sha=HEAD, + remote_head_sha=HEAD, + pr_head_sha=HEAD, + pr_number=OWNING_PR, + competing_live_locks=[], + candidate_branches=[BRANCH], + current_pid=os.getpid(), + ) + self.assertFalse(assessment["recovery_sanctioned"]) + self.assertIsNone( + issue_lock_recovery.owning_pr_recovery_evidence(assessment) + ) + + +# ──────────────────── end-to-end: the real gitea_lock_issue ──────────────────── + + +class LockIssueEndToEndBase(unittest.TestCase): + """Drives ``mcp_server.gitea_lock_issue`` with live git/Gitea observation + stubbed at the module boundary — the production gate chain itself runs.""" + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.worktree = os.path.realpath(os.getcwd()) + # Bind host/org/repo to what the ``test-author-prgs`` fixture profile is + # pinned to, so the session-context gate under test is the real one and + # not a cross-host denial. The issue number and lock dir stay synthetic. + self.remotes = patch.dict(mcp_server.REMOTES, { + "prgs": { + "host": "gitea.prgs.cc", + "org": ORG, + "repo": REPO, + }, + }) + self.remotes.start() + self.addCleanup(patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + + def write_durable_lock(self, *, pid, branch=BRANCH, worktree=None): + path = issue_lock_store.lock_file_path( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + claimant = {"username": IDENTITY, "profile": PROFILE} + data = { + "issue_number": ISSUE, + "branch_name": branch, + "remote": "prgs", + "org": ORG, + "repo": REPO, + "worktree_path": worktree or self.worktree, + "session_pid": pid, + "pid": pid, + "work_lease": { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": ISSUE, + "branch": branch, + "worktree_path": worktree or self.worktree, + "claimant": claimant, + "created_at": shifted_ts(-1), + "last_heartbeat_at": shifted_ts(-1), + "expires_at": shifted_ts(), + }, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=claimant, + ), + } + issue_lock_store.save_lock_file(path, data) + return path + + def run_lock( + self, + *, + open_prs, + porcelain="", + current_branch=BRANCH, + base_equivalent=False, + branch_names=None, + head_sha=HEAD, + remote_head=HEAD, + ): + branch_names = branch_names if branch_names is not None else [BRANCH] + branch_entries = [ + {"name": name, "commit": {"id": remote_head}} for name in branch_names + ] + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + with patch( + "mcp_server.api_get_all", return_value=branch_entries + ), patch( + "mcp_server._list_open_pulls", return_value=list(open_prs) + ), patch( + "mcp_server.get_auth_header", return_value="token x" + ), patch( + "mcp_server._work_lease_claimant", + return_value={"username": IDENTITY, "profile": PROFILE}, + ), patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": current_branch, + "porcelain_status": porcelain, + "base_equivalent": base_equivalent, + "head_sha": head_sha, + "inspected_git_root": self.worktree, + "base_branch": "master", + }, + ), patch( + "mcp_server.issue_duplicate_context_fetcher", + side_effect=lambda h, o, r, auth, issue_number: ( + list(open_prs), list(branch_names), {"status": "not_claimed"} + ), + ): + with patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_lock_issue( + issue_number=ISSUE, + branch_name=BRANCH, + remote="prgs", + worktree_path=self.worktree, + ) + + +class TestRecoveryWithOwningPrSucceeds(LockIssueEndToEndBase): + def test_dead_session_recovery_with_owning_pr_relocks(self): + self.write_durable_lock(pid=dead_pid()) + result = self.run_lock(open_prs=[owning_pr()]) + + self.assertTrue(result["success"]) + self.assertEqual(result["issue_number"], ISSUE) + self.assertEqual(result["branch_name"], BRANCH) + self.assertTrue(result["lock_freshness"]["live"]) + self.assertTrue(result["lock_freshness"]["pid_alive"]) + + def test_recovered_lock_records_truthful_provenance(self): + prior = dead_pid() + self.write_durable_lock(pid=prior) + self.run_lock(open_prs=[owning_pr()]) + + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + record = lock.get("dead_session_recovery") or {} + self.assertTrue(record.get("recovered")) + self.assertEqual(record.get("prior_session_pid"), prior) + self.assertEqual(record.get("replacement_session_pid"), os.getpid()) + self.assertFalse(record.get("prior_pid_alive")) + self.assertEqual(record.get("pr_number"), OWNING_PR) + self.assertEqual(record.get("branch_name"), BRANCH) + self.assertEqual(record.get("identity"), IDENTITY) + + def test_recovered_lock_is_live_and_proves_pr_ownership(self): + """AC6: the persisted lock satisfies update-by-merge's ownership prover.""" + self.write_durable_lock(pid=dead_pid()) + self.run_lock(open_prs=[owning_pr()]) + + lock = issue_lock_store.load_issue_lock( + remote="prgs", + org=ORG, + repo=REPO, + issue_number=ISSUE, + lock_dir=self.lock_dir.name, + ) + self.assertTrue(issue_lock_store.is_lease_live(lock)) + + env = shared_mutation_env( + "test-author-prgs", + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + with patch.dict(os.environ, env, clear=True): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + ownership = mcp_server._prove_author_ownership_for_pr( + pr_number=OWNING_PR, + pr_title=f"fix: something (Closes #{ISSUE})", + pr_body=f"Closes #{ISSUE}.", + source_branch=BRANCH, + remote="prgs", + host="gitea.prgs.cc", + org=ORG, + repo=REPO, + worktree_path=self.worktree, + ) + self.assertTrue(ownership["proven"], ownership["reasons"]) + self.assertTrue(ownership["has_author_lock"]) + self.assertEqual(ownership["matched_issue"], ISSUE) + + +class TestRecoveryRejectionsEndToEnd(LockIssueEndToEndBase): + def assert_lock_refused(self, **kwargs): + with self.assertRaises((ValueError, RuntimeError)) as ctx: + self.run_lock(**kwargs) + return str(ctx.exception) + + def test_competing_pr_still_blocked(self): + self.write_durable_lock(pid=dead_pid()) + message = self.assert_lock_refused( + open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + branch_names=[BRANCH], + ) + self.assertIn("already covers issue", message) + + def test_multiple_linked_open_prs_blocked(self): + self.write_durable_lock(pid=dead_pid()) + message = self.assert_lock_refused( + open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)], + ) + self.assertIn("already covers issue", message) + + def test_owning_pr_on_a_different_head_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused(open_prs=[owning_pr(sha=OTHER_HEAD)]) + + def test_lock_registered_to_a_different_worktree_blocked(self): + self.write_durable_lock( + pid=dead_pid(), worktree=os.path.join(self.worktree, "elsewhere") + ) + self.assert_lock_refused(open_prs=[owning_pr()]) + + def test_dirty_worktree_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], porcelain=" M gitea_mcp_server.py" + ) + + def test_worktree_parked_on_another_branch_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], current_branch="master" + ) + + def test_local_head_differing_from_remote_blocked(self): + self.write_durable_lock(pid=dead_pid()) + self.assert_lock_refused( + open_prs=[owning_pr()], head_sha=OTHER_HEAD + ) + + def test_live_prior_pid_blocked(self): + self.write_durable_lock(pid=os.getpid()) + self.assert_lock_refused(open_prs=[owning_pr()]) + + def test_new_claim_without_prior_lock_still_requires_base_equivalence(self): + """AC10: no durable lock → no recovery → base-equivalence still rules.""" + self.assert_lock_refused(open_prs=[], branch_names=[]) + + +if __name__ == "__main__": + unittest.main() diff --git a/workflow_scope_guard.py b/workflow_scope_guard.py index 11b1790..3996102 100644 --- a/workflow_scope_guard.py +++ b/workflow_scope_guard.py @@ -278,12 +278,19 @@ def assess_root_source_mutation( current_branch: str | None = None, locked_issue_number: int | None = None, role_kind: str | None = None, + mutation_task: str | None = None, ) -> dict[str, Any]: """Fail closed for diagnostic/source edits on the control/root checkout. Allowed only when the active workspace is under ``branches/``. Dirty tracked source/test files on the control checkout always block, including temporary/diagnostic/test-only intent. + + #749: ``create_issue`` is a pure remote mutation with no local tree write. + When *mutation_task* is create_issue and the control checkout has no dirty + source/test files, the missing-worktree signal is suppressed so the + sanctioned bootstrap path can proceed. Dirty roots and every other task + still fail closed. """ role = (role_kind or "").strip().lower() if role == "reconciler": @@ -302,6 +309,7 @@ def assess_root_source_mutation( dirty_src = dirty_source_files(porcelain_status) reasons: list[str] = [] blocker_kind: str | None = None + create_issue_bootstrap = False if not under_branches and workspace == root and dirty_src: # Root workspace with source dirtiness is unattributed root WIP. @@ -313,32 +321,50 @@ def assess_root_source_mutation( ) blocker_kind = BLOCKER_ROOT_DIAGNOSTIC_EDIT + # Lazy import keeps workflow_scope_guard free of circular import at module load. + try: + import create_issue_bootstrap as _cib + except Exception: # pragma: no cover - import always available in-tree + _cib = None + if ( not under_branches and workspace == root and not dirty_src and role == "author" ): - # Explicit missing-worktree signal for force-on author entrypoints. - reasons.append( - "author source/test mutation from the stable control checkout is " - "forbidden; bind an issue-backed worktree under branches/ first" - ) - blocker_kind = BLOCKER_MISSING_WORKTREE + if _cib is not None and _cib.is_create_issue_task(mutation_task): + # #749: clean-root create_issue is the sanctioned bootstrap path. + create_issue_bootstrap = True + else: + # Explicit missing-worktree signal for force-on author entrypoints. + reasons.append( + "author source/test mutation from the stable control checkout is " + "forbidden; bind an issue-backed worktree under branches/ first" + ) + blocker_kind = BLOCKER_MISSING_WORKTREE if reasons: kind = blocker_kind or BLOCKER_ROOT_DIAGNOSTIC_EDIT + next_action = _NEXT_ACTIONS[kind] + if ( + kind == BLOCKER_ROOT_DIAGNOSTIC_EDIT + and _cib is not None + and _cib.is_create_issue_task(mutation_task) + ): + next_action = _cib.EXACT_NEXT_ACTION_BOOTSTRAP return { "proven": False, "block": True, "blocker_kind": kind, - "exact_next_action": _NEXT_ACTIONS[kind], + "exact_next_action": next_action, "reasons": reasons, "dirty_source_files": dirty_src, "workspace_path": workspace, "canonical_repo_root": root, "under_branches": under_branches, "locked_issue_number": locked_issue_number, + "create_issue_bootstrap": False, } return { "proven": True, @@ -351,6 +377,7 @@ def assess_root_source_mutation( "canonical_repo_root": root, "under_branches": under_branches, "locked_issue_number": locked_issue_number, + "create_issue_bootstrap": create_issue_bootstrap, } @@ -365,6 +392,7 @@ def assess_production_mutation_guards( role_kind: str | None = None, require_author_lock: bool = False, in_test_mode: bool = False, + mutation_task: str | None = None, ) -> dict[str, Any]: """Compose root + scope production guards when they must be active (#683).""" if not production_guards_active(in_test_mode=in_test_mode): @@ -385,6 +413,7 @@ def assess_production_mutation_guards( current_branch=current_branch, locked_issue_number=locked_issue_number, role_kind=role_kind, + mutation_task=mutation_task, ) if root_assess["block"]: return {**root_assess, "skipped": False} @@ -408,6 +437,7 @@ def assess_production_mutation_guards( "skipped": False, "root": root_assess, "scope": scope_assess, + "create_issue_bootstrap": bool(root_assess.get("create_issue_bootstrap")), }