diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index 634e22c..d67769d 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -286,6 +286,28 @@ heartbeat timestamp. An active same-issue/same-operation lease blocks duplicate work. An expired lease still blocks takeover until a recovery review records why the prior work is abandoned, completed, or unsafe to continue. +**Stacked PRs (#484).** By default the lock worktree must be base-equivalent to +`master`/`main`/`dev` — ordinary work is unchanged. A *stacked* PR (deliberately +based on another unmerged PR's branch) is an explicit, opt-in path: pass +`stacked_base_branch` **and** `stacked_base_pr` to `gitea_lock_issue`. The lock +fails closed unless that branch is owned by a live **open** PR whose number +matches `stacked_base_pr`, so arbitrary or stale branches cannot be used as +bases. When approved, the lock payload records +`approved_stacked_base = {branch, pr_number, verified_open}` and the worktree may +be base-equivalent to that branch instead of master. `gitea_create_pr` then +allows `base = ` only when it matches the recorded approval, the +dependency PR is **still open**, and the PR body documents the stack: + +- `Stacked on PR # / issue #` +- `Base branch: ` +- `Head branch: ` +- `Do not merge before PR #` (merge ordering) +- retarget/rebase to `master` after the dependency lands, if required + +Stacked support never bypasses the issue lock — the base is recorded *on* the +lock and re-verified at PR time. A merged/closed dependency base fails closed; +retarget onto `master` or re-lock against a live base. + **Do not manually seed `/tmp/gitea_issue_lock.json` or any lock file as a normal recovery path.** That global slot is deprecated and can clobber unrelated live leases (#438). After an MCP restart, call `gitea_lock_issue` again — own-branch @@ -303,6 +325,17 @@ shared state and manual writes can clobber another session's live lease. Use 3. Operator override only when explicitly authorized — record `External-state mutations` and `operator override proof` in the final report. +**Adoption proof in the live lock response (#477):** when `gitea_lock_issue` +adopts an existing own branch, the response carries an `adoption` block with +citable fields — `adoption_decision` (`ADOPT`), `adopted` (`true`), +`adopted_branch`, `adopted_branch_head`, `matcher_summary` (boundary-safe reason +the branch qualified), `competing_branch_check`, and `safe_next_action`. A normal +lock instead returns an `adoption_check` block with `adoption_decision` +(`NO_MATCH`) and `adopted: false`, so a non-adoption response can never be misread +as claiming adoption. Recovery reports should quote the live lock response +`adoption`/`adoption_check` block directly instead of inferring adoption from +separate offline checks. + `gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance` metadata. Final-report validation blocks handoffs that hide lock read/write/delete under `External-state mutations: none` or mix author PR creation with reviewer diff --git a/final_report_validator.py b/final_report_validator.py index 6b6edac..9cfb4ab 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -508,6 +508,42 @@ def _rule_reviewer_validation_failure_history( ] +def _rule_reviewer_validation_cwd_proof( + report_text: str, + *, + validation_session: dict | None = None, +) -> list[dict[str, str]]: + from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report + + session = validation_session or {} + claims = ( + session.get("validation_ran") + or session.get("command") + or session.get("baseline_validation_ran") + ) + if not claims and "validation command:" not in (report_text or "").lower(): + return [] + + result = assess_validation_cwd_proof_report( + report_text, + validation_session=session, + ) + if result.get("proven") or not result.get("claims_validation"): + return [] + severity = "block" if result.get("violations") else "downgrade" + return [ + validator_finding( + "reviewer.validation_cwd_proof", + severity, + "Validation cwd/HEAD proof", + reason, + result.get("safe_next_action") + or "document pwd, HEAD SHA, and explicit cwd before validation", + ) + for reason in (result.get("violations") or result.get("reasons") or ["incomplete"]) + ] + + def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]: from pr_work_lease import assess_reviewer_stale_head_final_report @@ -1087,6 +1123,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_git_fetch_readonly, _rule_reviewer_validation_command, _rule_reviewer_validation_failure_history, + _rule_reviewer_validation_cwd_proof, _rule_reviewer_validation_structured, _rule_reviewer_linked_issue, _rule_reviewer_baseline_on_failure, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 12e456f..1b12950 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -165,6 +165,7 @@ _preflight_capability_called = False _preflight_whoami_violation = False _preflight_capability_violation = False _preflight_resolved_role = None +_preflight_resolved_task: str | None = None _process_start_porcelain: str | None = None _preflight_whoami_baseline_porcelain: str | None = None _preflight_capability_baseline_porcelain: str | None = None @@ -379,11 +380,30 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict: } -def record_preflight_check(type_name: str, resolved_role: str | None = None): +def _clear_preflight_capability_state() -> None: + """Drop resolved capability proof (consumed by a mutation or fresh resolve).""" + global _preflight_capability_called, _preflight_capability_violation + global _preflight_resolved_task + global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files + global _preflight_reviewer_violation_files + + _preflight_capability_called = False + _preflight_capability_violation = False + _preflight_capability_violation_files = [] + _preflight_capability_baseline_porcelain = None + _preflight_resolved_task = None + _preflight_reviewer_violation_files = [] + + +def record_preflight_check( + type_name: str, + resolved_role: str | None = None, + resolved_task: str | None = None, +): """Record a pre-flight check (whoami or capability) with session-scoped deltas.""" global _preflight_whoami_called, _preflight_capability_called global _preflight_whoami_violation, _preflight_capability_violation - global _preflight_resolved_role + global _preflight_resolved_role, _preflight_resolved_task global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain global _preflight_whoami_violation_files, _preflight_capability_violation_files global _preflight_reviewer_violation_files @@ -391,13 +411,24 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): current = _get_workspace_porcelain() if type_name == "whoami": - # Fresh whoami restarts the capability step and re-evaluates violations - # instead of replaying a sticky record (#252). - _preflight_capability_called = False - _preflight_capability_violation = False - _preflight_capability_violation_files = [] - _preflight_capability_baseline_porcelain = None - _preflight_reviewer_violation_files = [] + # Re-evaluate whoami violations instead of replaying sticky state (#252). + # Interleaved read-only whoami must not clear a valid capability proof (#469). + saved_capability = None + preserve_capability = ( + _preflight_capability_called + and _preflight_whoami_called + and not _preflight_whoami_violation + ) + if preserve_capability: + saved_capability = ( + _preflight_capability_violation, + list(_preflight_capability_violation_files), + _preflight_capability_baseline_porcelain, + _preflight_resolved_role, + _preflight_resolved_task, + ) + else: + _clear_preflight_capability_state() process_start = _ensure_process_start_porcelain() whoami_delta = _new_tracked_changes_since(process_start, current) @@ -405,6 +436,20 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): _preflight_whoami_violation_files = whoami_delta _preflight_whoami_baseline_porcelain = current _preflight_whoami_called = True + + if ( + preserve_capability + and saved_capability is not None + and not whoami_delta + ): + ( + _preflight_capability_violation, + _preflight_capability_violation_files, + _preflight_capability_baseline_porcelain, + _preflight_resolved_role, + _preflight_resolved_task, + ) = saved_capability + _preflight_capability_called = True elif type_name == "capability": baseline = _preflight_whoami_baseline_porcelain or "" capability_delta = _new_tracked_changes_since(baseline, current) @@ -414,11 +459,18 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): _preflight_capability_called = True if resolved_role: _preflight_resolved_role = resolved_role + if resolved_task: + _preflight_resolved_task = resolved_task def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None: - """#274: author mutations must run from a branches/ session worktree.""" - if _preflight_resolved_role == "reviewer": + """#274: author file/branch mutations must run from a branches/ worktree. + + Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a + Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE`` + (#468). + """ + if _preflight_resolved_role in ("reviewer", "reconciler"): return ctx = _resolve_author_mutation_context(worktree_path) workspace = ctx["workspace_path"] @@ -434,7 +486,11 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> ) -def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None): +def verify_preflight_purity( + remote: str | None = None, + worktree_path: str | None = None, + task: str | None = None, +): """Verify that identity and capability were verified prior to session edits.""" global _preflight_reviewer_violation_files @@ -453,6 +509,16 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None raise RuntimeError( "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" ) + if ( + task is not None + and _preflight_resolved_task is not None + and task != _preflight_resolved_task + ): + raise RuntimeError( + "Pre-flight task mismatch: " + f"resolved '{_preflight_resolved_task}' but mutation requires " + f"'{task}' (fail closed)" + ) ctx = _resolve_author_mutation_context(worktree_path) workspace = ctx["workspace_path"] @@ -508,6 +574,7 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None ) _enforce_branches_only_author_mutation(worktree_path) + _clear_preflight_capability_state() from mcp.server.fastmcp import FastMCP # noqa: E402 @@ -537,6 +604,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 stacked_pr_support # noqa: E402 import merge_approval_gate # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 @@ -1224,7 +1292,7 @@ def gitea_create_issue( ) if blocked: return blocked - verify_preflight_purity(remote, worktree_path=worktree_path) + verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue") base = repo_api_url(h, o, r) open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth) closed_issues = api_get_all( @@ -1264,6 +1332,16 @@ def gitea_create_issue( @mcp.tool() +def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]: + """Fetch all OPEN pull requests for a repo (used for stacked-base proof, #484).""" + try: + return api_get_all(f"{repo_api_url(h, o, r)}/pulls?state=open", auth) or [] + except Exception as exc: # fail closed: no proof of an open dependency PR + raise RuntimeError( + f"Could not list open pull requests to verify stacked base: {exc}" + ) + + def gitea_lock_issue( issue_number: int, branch_name: str, @@ -1272,6 +1350,8 @@ def gitea_lock_issue( org: str | None = None, repo: str | None = None, worktree_path: str | None = None, + stacked_base_branch: str | None = None, + stacked_base_pr: int | None = None, ) -> dict: """Lock exactly one Gitea issue and its branch name to ensure durable tracking. @@ -1284,6 +1364,15 @@ def gitea_lock_issue( repo: Override Repo. worktree_path: Author scratch-clone path to validate (defaults to GITEA_AUTHOR_WORKTREE or the MCP server project root). + stacked_base_branch: Opt-in. Declare a non-master base branch for a + *stacked* PR (a PR based on another unmerged PR's branch). Normal work + leaves this ``None`` and stays master-equivalent. When set, the + worktree may be base-equivalent to this branch instead of + master/main/dev, and the approved base is recorded on the lock (#484). + stacked_base_pr: Required when ``stacked_base_branch`` is set. The number + of the OPEN pull request that owns the stacked base branch. The lock + fails closed unless this open PR exists and owns that branch, so + arbitrary or stale branches cannot be used as stacked bases. """ # 1. Enforce branch name includes issue number expected_pattern = f"issue-{issue_number}" @@ -1311,8 +1400,32 @@ def gitea_lock_issue( if active_lease_block: raise RuntimeError(active_lease_block) - git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree) - verify_preflight_purity(remote, worktree_path=resolved_worktree) + # ── Stacked-PR base declaration (opt-in, #484) ── + # Normal work leaves stacked_base_branch None → master-equivalent path. + # A declared stacked base must be proven to own an OPEN PR before it can + # anchor base-equivalence; this never bypasses the lock. + stacked_extra_bases: tuple[str, ...] = () + stacked_approved: dict | None = None + if stacked_base_branch: + stacked_assessment = stacked_pr_support.assess_stacked_base_declaration( + stacked_base_branch=stacked_base_branch, + stacked_base_pr=stacked_base_pr, + open_prs=_list_open_pulls(h, o, r, _auth(h)), + ) + if stacked_assessment["block"]: + raise RuntimeError( + "; ".join(stacked_assessment["reasons"]) + " (fail closed)" + ) + stacked_approved = stacked_assessment["approved"] + stacked_extra_bases = (stacked_approved["branch"],) + + if stacked_extra_bases: + git_state = issue_lock_worktree.read_worktree_git_state( + resolved_worktree, extra_bases=stacked_extra_bases + ) + else: + git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree) + verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue") lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( worktree_path=resolved_worktree, current_branch=git_state.get("current_branch"), @@ -1384,6 +1497,8 @@ def gitea_lock_issue( claimant=work_lease.get("claimant"), ), } + if stacked_approved: + data["approved_stacked_base"] = stacked_approved lock_file_path = _save_issue_lock(data) lock_record = issue_lock_store.read_lock_file(lock_file_path) or data @@ -1417,6 +1532,13 @@ def gitea_lock_issue( "lock_freshness": freshness, "lock_proof": lock_proof, } + if stacked_approved: + result["approved_stacked_base"] = stacked_approved + result["message"] = ( + f"Successfully locked issue #{issue_number} to branch '{branch_name}' " + f"as a STACKED PR on base '{stacked_approved['branch']}' " + f"(open PR #{stacked_approved['pr_number']}); fail-closed check complete." + ) if adoption["adopt"]: result["adoption"] = issue_lock_adoption.build_adoption_proof( issue_number=issue_number, @@ -1431,6 +1553,14 @@ def gitea_lock_issue( f"Adopted existing branch '{branch_name}' and locked issue " f"#{issue_number} for recovery (fail-closed check complete)." ) + else: + # #477 AC2: normal (no-adoption) lock responses carry explicit, + # adoption-free proof metadata so they stay clear and cannot be + # misread as claiming a branch was adopted. + result["adoption_check"] = issue_lock_adoption.build_non_adoption_lock_proof( + issue_number=issue_number, + branch_name=branch_name, + ) if agent_artifacts: result["warnings"] = [ "Agent temp artifacts at repo root (delete before implementation): " @@ -1519,7 +1649,7 @@ def gitea_create_pr( ) if blocked: return blocked - verify_preflight_purity(remote, worktree_path=worktree_path) + verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr") h, o, r = _resolve(remote, host, org, repo) # ── Issue Lock Validation (Issue #194 / #196 / #443) ── @@ -1573,6 +1703,24 @@ def gitea_create_pr( f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)" ) + # ── Stacked-PR base validation (#484) ── + # Normal base branches (master/main/dev) pass unchanged. A non-base target is + # allowed only when it matches the lock's approved stacked base, that base still + # has an open PR, and the body documents the stack. This never bypasses the lock. + base_open_prs = ( + [] + if stacked_pr_support.is_base_branch(base) + else _list_open_pulls(h, o, r, _auth(h)) + ) + base_check = stacked_pr_support.assess_create_pr_base( + base=base, + approved_stacked_base=lock_data.get("approved_stacked_base"), + body=body, + open_prs=base_open_prs, + ) + if base_check["block"]: + raise ValueError("; ".join(base_check["reasons"]) + " (fail closed)") + duplicate_block = _enforce_locked_issue_duplicate_recheck( remote, issue_work_duplicate_gate.PHASE_CREATE_PR, @@ -2526,7 +2674,12 @@ def _list_pr_lease_comments( h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments" - comments = api_request("GET", api, auth) or [] + comments = api_request("GET", api, auth) + # Fail safe to no lease comments when the API returns a non-list payload + # (e.g. an error object such as an HTTP 401 body): lease state can only be + # proven from real comment entries, never inferred from an error shape (#485). + if not isinstance(comments, list): + return [] return list(comments[:limit]) @@ -2571,7 +2724,7 @@ def _evaluate_pr_review_submission( final_review_decision_ready: bool = False, ) -> dict: """Shared gate chain for live submit and dry-run review tools.""" - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="review_pr") action = (action or "").strip().lower() workflow_blockers = _review_workflow_load_gate_reasons() if live else [] result = { @@ -3090,12 +3243,12 @@ def gitea_edit_pr( if not payload: raise ValueError("At least one field to edit (title, body, state, base) must be provided.") - verify_preflight_purity(remote) + closing = payload.get("state") == "closed" + verify_preflight_purity(remote, task="close_pr" if closing else None) # PR closure is a first-class capability, distinct from retitling or # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close # never touches the network. - closing = payload.get("state") == "closed" if closing: gate_reasons = _profile_operation_gate("gitea.pr.close") if gate_reasons: @@ -3340,7 +3493,7 @@ def gitea_commit_files( branch="", ) - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="commit_files") processed_files, source_proofs = _prepare_commit_payload_files(files) h, o, r = _resolve(remote, host, org, repo) @@ -3441,7 +3594,7 @@ def gitea_merge_pr( reasons/gates passed or blocked, and merge result / merge commit if available. Never secrets. """ - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="merge_pr") workflow_blockers = _review_workflow_load_gate_reasons() do = (do or "").strip().lower() result = { @@ -3975,7 +4128,7 @@ def gitea_delete_branch( "permission_report": _permission_block_report("gitea.branch.delete"), } - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="delete_branch") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) import urllib.parse @@ -4084,7 +4237,7 @@ def gitea_reconcile_merged_cleanups( report["executed"] = False return {"success": True, "performed": False, **report} - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="reconcile_merged_cleanups") actions: list[dict] = [] for entry in report.get("entries") or []: head_branch = entry.get("head_branch") or "" @@ -4398,7 +4551,7 @@ def gitea_reconcile_already_landed_pr( ) return result - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="reconcile_already_landed_pr") if post_comment and comment_body.strip(): comment_block = _profile_operation_gate("gitea.pr.comment") @@ -4501,7 +4654,7 @@ def gitea_close_issue( task_capability_map.required_permission("close_issue")) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="close_issue") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/issues/{issue_number}" @@ -4929,7 +5082,7 @@ def gitea_acquire_reviewer_pr_lease( "permission_report": _permission_block_report("gitea.pr.comment"), } - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="review_pr") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) profile = get_profile() @@ -5030,7 +5183,7 @@ def gitea_heartbeat_reviewer_pr_lease( ], } - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="review_pr") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) body = reviewer_pr_lease.format_lease_body( @@ -5147,7 +5300,12 @@ def gitea_list_issue_comments( h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments" - comments = api_request("GET", api, auth) or [] + comments = api_request("GET", api, auth) + # Fail safe to no comments when the API returns a non-list payload (e.g. an + # error object such as an HTTP 401 body) so listing never crashes on a + # malformed response (#485). + if not isinstance(comments, list): + comments = [] reveal = _reveal_endpoints() out = [] for c in comments[:limit]: @@ -5201,7 +5359,7 @@ def gitea_create_issue_comment( (permission blocks also carry a structured 'permission_report', #142). """ - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="comment_issue") gate_reasons = _profile_operation_gate("gitea.issue.comment") reasons = list(gate_reasons) if not (body or "").strip(): @@ -6813,7 +6971,7 @@ def gitea_mark_issue( task_capability_map.required_permission("mark_issue")) if blocked: return blocked - verify_preflight_purity(remote, worktree_path=worktree_path) + verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) @@ -6891,7 +7049,7 @@ def gitea_post_heartbeat( task_capability_map.required_permission("post_heartbeat")) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="post_heartbeat") active_profile = profile or get_profile().get("profile_name") body = issue_claim_heartbeat.format_heartbeat_body( kind="progress", @@ -7134,7 +7292,7 @@ def gitea_cleanup_stale_claims( task_capability_map.required_permission("cleanup_stale_claims")) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="cleanup_stale_claims") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -7288,7 +7446,7 @@ def gitea_set_issue_labels( task_capability_map.required_permission("set_issue_labels")) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, task="set_issue_labels") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) @@ -7525,7 +7683,7 @@ def gitea_resolve_task_capability( "exact_safe_next_action": next_safe_action, } - record_preflight_check("capability", required_role) + record_preflight_check("capability", required_role, resolved_task=task) # Try automatic dispatch switching _ensure_matching_profile(required_permission, required_role, remote, host) diff --git a/issue_lock_adoption.py b/issue_lock_adoption.py index 8f8a8d6..81a5a92 100644 --- a/issue_lock_adoption.py +++ b/issue_lock_adoption.py @@ -20,6 +20,43 @@ ADOPT = "adopt_existing_branch" BLOCK_COMPETING = "block_competing_branch" NO_MATCH = "no_matching_branch" +# Citable decision labels aligned with the ``assess_own_branch_adoption`` +# outcomes, surfaced verbatim in the live ``gitea_lock_issue`` response so +# recovery reports (#473-style) can quote the lock tool output directly +# instead of inferring adoption from separate offline checks (#477). +DECISION_LABELS = { + ADOPT: "ADOPT", + BLOCK_COMPETING: "BLOCK_COMPETING", + NO_MATCH: "NO_MATCH", +} + +_SAFE_NEXT_ACTIONS = { + ADOPT: ( + "Own existing branch adopted for lock recovery; proceed to " + "gitea_create_pr for this issue and cite this adoption proof." + ), + BLOCK_COMPETING: ( + "Competing same-issue branch(es) exist; resolve branch ownership " + "before locking. No adoption performed (fail closed)." + ), + NO_MATCH: ( + "No existing branch carries this issue marker; normal lock path " + "applied. No adoption performed." + ), +} + + +def decision_label(outcome: str) -> str: + """Map an ``assess_own_branch_adoption`` outcome to its citable label.""" + return DECISION_LABELS.get(outcome, "UNKNOWN") + + +def safe_next_action(outcome: str) -> str: + """Return the safe next action string for an adoption *outcome*.""" + return _SAFE_NEXT_ACTIONS.get( + outcome, "Unknown adoption outcome; treat as fail closed." + ) + def _branch_name(entry) -> str: if isinstance(entry, dict): @@ -128,6 +165,42 @@ def assess_own_branch_adoption( } +def _matcher_summary(issue_number: int, assessment: dict) -> str: + """Explain, citably, why the assessed branch did or did not qualify. + + Names the numeric word-boundary rule so reports can show that + ``issue-42`` was not matched inside ``issue-420`` (#440 / #477 AC3). + """ + outcome = assessment.get("outcome") + matched = assessment.get("matched_branch") + competing = assessment.get("competing_branches") or [] + if outcome == ADOPT and matched: + return ( + f"branch '{matched}' exactly matches the issue-{int(issue_number)} " + f"marker (numeric word-boundary; 'issue-{int(issue_number)}' is not " + f"matched inside 'issue-{int(issue_number)}0')" + ) + if outcome == BLOCK_COMPETING: + return ( + f"competing same-issue branch(es) {competing} carry the " + f"issue-{int(issue_number)} marker but are not the requested " + f"branch; ownership is ambiguous (fail closed)" + ) + return ( + f"no existing branch carries the issue-{int(issue_number)} marker " + f"under the numeric word-boundary rule" + ) + + +def _competing_branch_check(assessment: dict) -> dict: + """Structured competing-branch verdict for the proof block.""" + competing = list(assessment.get("competing_branches") or []) + return { + "result": "blocked" if competing else "clear", + "competing_branches": competing, + } + + def build_adoption_proof( *, issue_number: int, @@ -143,7 +216,18 @@ def build_adoption_proof( Requirement #4: adoption results must carry issue number, branch name, branch head commit, adoption reason, no-existing-PR proof, no-competing- live-lock proof, and lock file path/status. + + #477: additionally surface explicit, citable adoption-proof fields tied to + the ``assess_own_branch_adoption`` outcome (``adoption_decision``, + ``adopted``, ``adopted_branch``, ``adopted_branch_head``, + ``matcher_summary``, ``competing_branch_check``, ``safe_next_action``) so a + recovery session can quote the live lock response directly. The explicit + fields are populated for any outcome; ``adopted_branch`` / + ``adopted_branch_head`` are set only when the outcome is ADOPT so a + non-adoption proof can never be misread as claiming adoption. """ + outcome = assessment.get("outcome") + adopted = outcome == ADOPT return { "issue_number": issue_number, "branch_name": branch_name, @@ -153,4 +237,36 @@ def build_adoption_proof( "no_competing_live_lock_proof": bool(competing_lock_checked), "lock_file_path": lock_file_path, "lock_file_status": lock_file_status, + # Explicit citable fields (#477). + "adoption_decision": decision_label(outcome), + "adopted": adopted, + "adopted_branch": branch_name if adopted else None, + "adopted_branch_head": assessment.get("matched_head_sha") if adopted else None, + "matcher_summary": _matcher_summary(issue_number, assessment), + "competing_branch_check": _competing_branch_check(assessment), + "safe_next_action": safe_next_action(outcome), + } + + +def build_non_adoption_lock_proof(*, issue_number: int, branch_name: str) -> dict: + """Safe, adoption-free proof metadata for a normal (NO_MATCH) lock. + + Requirement #477 AC2: non-adoption lock responses must stay clear and must + not imply adoption. This returns explicit ``adopted: False`` metadata with + the ``NO_MATCH`` decision so a normal lock response can carry citable proof + without ever asserting a branch was adopted. + """ + return { + "issue_number": issue_number, + "branch_name": branch_name, + "adoption_decision": DECISION_LABELS[NO_MATCH], + "adopted": False, + "adopted_branch": None, + "adopted_branch_head": None, + "matcher_summary": ( + f"no existing branch carries the issue-{int(issue_number)} marker; " + f"normal lock path (no adoption)" + ), + "competing_branch_check": {"result": "clear", "competing_branches": []}, + "safe_next_action": safe_next_action(NO_MATCH), } \ No newline at end of file diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index 5d9dc4a..3c23134 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -30,8 +30,16 @@ def resolve_author_worktree_path( return os.path.realpath(os.path.abspath(path)) -def read_worktree_git_state(worktree_path: str) -> dict: - """Read branch name and porcelain status from a git worktree.""" +def read_worktree_git_state( + worktree_path: str, + extra_bases: tuple[str, ...] | list[str] = (), +) -> dict: + """Read branch name and porcelain status from a git worktree. + + ``extra_bases`` names additional branches (e.g. an approved stacked base) + that may anchor base-equivalence in addition to master/main/dev. When empty + (the default), only the normal base branches are considered. + """ path = (worktree_path or "").strip() if not path: return {"current_branch": None, "porcelain_status": ""} @@ -63,7 +71,7 @@ def read_worktree_git_state(worktree_path: str) -> dict: check=False, ) head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None - base_branch, base_sha = _find_matching_base_ref(path, head_sha) + base_branch, base_sha = _find_matching_base_ref(path, head_sha, extra_bases) return { "current_branch": current_branch, "porcelain_status": status_res.stdout or "", @@ -203,13 +211,26 @@ def _assessment( } -def _find_matching_base_ref(path: str, head_sha: str | None) -> tuple[str | None, str | None]: - """Return the stable branch ref whose commit matches HEAD, if any.""" +def _find_matching_base_ref( + path: str, + head_sha: str | None, + extra_bases: tuple[str, ...] | list[str] = (), +) -> tuple[str | None, str | None]: + """Return the stable branch ref whose commit matches HEAD, if any. + + Normal base branches (master/main/dev) are always considered. ``extra_bases`` + adds explicitly-approved stacked bases; each is checked as a local ref and via + the ``prgs``/``origin`` remotes. + """ if not head_sha: return None, None candidates: list[str] = [] for branch in sorted(BASE_BRANCHES): candidates.extend((f"origin/{branch}", branch)) + for branch in extra_bases: + name = (branch or "").strip() + if name: + candidates.extend((f"prgs/{name}", f"origin/{name}", name)) for ref in candidates: res = subprocess.run( ["git", "-C", path, "rev-parse", "--verify", ref], diff --git a/review_proofs.py b/review_proofs.py index b74ebb8..febbcd0 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -5515,6 +5515,13 @@ def assess_validation_failure_history_report(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_validation_cwd_proof_report(report_text, **kwargs): + """#398: validation commands require explicit worktree cwd and HEAD proof.""" + from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess + + return _assess(report_text, **kwargs) + + def assess_already_landed_classification_report(report_text, **kwargs): """#295: already-landed PRs are reconciliation-only, not review eligible.""" from reviewer_already_landed_classification import ( diff --git a/reviewer_validation_cwd_proof.py b/reviewer_validation_cwd_proof.py new file mode 100644 index 0000000..15cfb02 --- /dev/null +++ b/reviewer_validation_cwd_proof.py @@ -0,0 +1,233 @@ +"""Explicit worktree and cwd proof for PR review validation (#398).""" + +from __future__ import annotations + +import re +from typing import Any + +_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE) + +_PWD_RE = re.compile( + r"(?:^|\n)\s*(?:pwd|working\s+directory|cwd)\s*:\s*(\S+)", + re.IGNORECASE, +) +_HEAD_RE = re.compile( + r"(?:git\s+rev-parse\s+head|observed\s+head\s+sha)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE | re.MULTILINE, +) +_EXPECTED_HEAD_RE = re.compile( + r"(?:expected\s+(?:pr\s+)?head\s+sha|candidate\s+head\s+sha|pinned\s+head)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_STATUS_RE = re.compile( + r"git\s+status\s+(?:--short\s+--branch|--short|-sb)\s*:\s*(.+)", + re.IGNORECASE, +) +_VALIDATION_CMD_RE = re.compile( + r"validation\s+command\s*:\s*(.+)", + re.IGNORECASE, +) +_GIT_C_CMD_RE = re.compile(r"git\s+-C\s+\S+", re.IGNORECASE) +_CD_CMD_RE = re.compile(r"(?:^|&&\s*)cd\s+\S+", re.IGNORECASE) +_BASELINE_CWD_RE = re.compile( + r"baseline\s+(?:worktree|working\s+directory|cwd)\s*:\s*(\S+)", + re.IGNORECASE, +) +_BASELINE_SHA_RE = re.compile( + r"baseline\s+(?:target\s+)?sha\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_BASELINE_CMD_RE = re.compile( + r"baseline\s+validation\s+command\s*:\s*(.+)", + re.IGNORECASE, +) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").rstrip("/") + + +def _path_under_branches(path: str, project_root: str | None = None) -> bool: + normalized = _normalize_path(path) + if not normalized: + return False + if "/branches/" in f"{normalized}/": + return True + if normalized.endswith("/branches"): + return True + if project_root: + root = _normalize_path(project_root) + if normalized.startswith(f"{root}/"): + rel = normalized[len(root) + 1 :] + return rel == "branches" or rel.startswith("branches/") + return False + + +def _expand_sha(sha: str) -> str: + return (sha or "").strip().lower() + + +def _sha_matches(expected: str, observed: str) -> bool: + exp = _expand_sha(expected) + obs = _expand_sha(observed) + if not exp or not obs: + return False + if len(exp) == 40 and len(obs) == 40: + return exp == obs + return obs.startswith(exp) or exp.startswith(obs) + + +def _command_has_explicit_cwd(command: str, cwd: str) -> bool: + text = (command or "").strip() + if not text: + return False + if _GIT_C_CMD_RE.search(text): + return True + if _CD_CMD_RE.search(text): + return True + if cwd and cwd in text: + return True + return False + + +def assess_validation_cwd_proof_report( + report_text: str, + *, + validation_session: dict | None = None, + project_root: str | None = None, +) -> dict[str, Any]: + """Require cwd/HEAD proof before reviewer validation claims (#398).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + violations: list[str] = [] + + claims_validation = bool( + session.get("validation_ran") + or _VALIDATION_CMD_RE.search(text) + or session.get("command") + ) + if not claims_validation: + return { + "proven": True, + "block": False, + "claims_validation": False, + "reasons": [], + "violations": [], + "safe_next_action": "proceed", + } + + expected_head = ( + session.get("expected_head_sha") + or session.get("candidate_head_sha") + or "" + ).strip() + if not expected_head: + match = _EXPECTED_HEAD_RE.search(text) + expected_head = (match.group(1) if match else "").strip() + + observed_head = (session.get("observed_head_sha") or "").strip() + if not observed_head: + match = _HEAD_RE.search(text) + observed_head = (match.group(1) if match else "").strip() + + cwd = ( + session.get("working_directory") + or session.get("cwd") + or session.get("pwd") + or "" + ).strip() + if not cwd: + match = _PWD_RE.search(text) + cwd = (match.group(1) if match else "").strip().rstrip(",.;") + + command = (session.get("command") or "").strip() + if not command: + match = _VALIDATION_CMD_RE.search(text) + command = (match.group(1) if match else "").strip().rstrip(".;") + + if not cwd: + reasons.append( + "validation claimed without pwd/working-directory proof (#398)" + ) + elif not _path_under_branches(cwd, project_root): + violations.append( + f"validation cwd {cwd!r} is not under branches/ (#398)" + ) + reasons.append( + "reviewer validation must run from a branches/ worktree, " + "not the main checkout (#398)" + ) + + if not observed_head: + reasons.append( + "validation claimed without git rev-parse HEAD / observed HEAD SHA " + "proof (#398)" + ) + elif expected_head and not _sha_matches(expected_head, observed_head): + violations.append( + f"observed HEAD {observed_head} does not match expected " + f"PR head {expected_head} (#398)" + ) + reasons.append("validation HEAD SHA must match pinned PR head (#398)") + + if not _STATUS_RE.search(text) and session.get("git_status") is None: + reasons.append( + "validation claimed without git status --short --branch proof (#398)" + ) + + if command and cwd and not _command_has_explicit_cwd(command, cwd): + if session.get("tool_working_directory") is not True: + reasons.append( + "validation command must use git -C , " + "cd && ..., or tool-provided cwd metadata (#398)" + ) + + baseline_ran = bool( + session.get("baseline_validation_ran") + or _BASELINE_CMD_RE.search(text) + ) + if baseline_ran: + baseline_cwd = (session.get("baseline_worktree_path") or "").strip() + if not baseline_cwd: + match = _BASELINE_CWD_RE.search(text) + baseline_cwd = (match.group(1) if match else "").strip().rstrip(",.;") + if not baseline_cwd or not _path_under_branches(baseline_cwd, project_root): + reasons.append( + "baseline validation claimed without baseline worktree cwd " + "under branches/ (#398)" + ) + baseline_sha = (session.get("baseline_target_sha") or "").strip() + if not baseline_sha: + match = _BASELINE_SHA_RE.search(text) + baseline_sha = (match.group(1) if match else "").strip() + if not baseline_sha: + reasons.append( + "baseline validation claimed without baseline target SHA (#398)" + ) + baseline_cmd = (session.get("baseline_command") or "").strip() + if not baseline_cmd: + match = _BASELINE_CMD_RE.search(text) + baseline_cmd = (match.group(1) if match else "").strip() + if not baseline_cmd: + reasons.append( + "baseline validation claimed without exact baseline command (#398)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations) or not proven, + "claims_validation": True, + "expected_head_sha": expected_head or None, + "observed_head_sha": observed_head or None, + "working_directory": cwd or None, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "before validation record pwd, git rev-parse HEAD, git status, " + "expected PR head SHA; run commands with git -C or cd in the same line" + if not proven + else "proceed" + ), + } \ No newline at end of file diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 0336d55..c8667f1 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -664,6 +664,28 @@ Do not claim “full-suite failures are pre-existing” unless baseline proof is ## 23. Validation command proof rule +Before any diff, test, or compile validation, record in the same command +transcript or final report: + +* `pwd` or explicit working directory +* `git rev-parse HEAD` +* `git status --short --branch` +* expected PR head SHA (candidate head SHA) + +Validation commands must use one of: + +* `git -C ...` +* `cd && ...` in the same command +* tool-provided explicit working-directory metadata + +Do not rely on inferred shell cwd from a prior command in a different block. + +`gitea_validate_review_final_report` rejects validation claims without +cwd/HEAD proof when `validation_session` is supplied. + +Baseline validation must document baseline worktree path, baseline target SHA, +cwd proof, exact baseline command, and baseline result using the same rules. + Report the exact validation command as executed. Report the working directory where validation ran. @@ -1227,6 +1249,7 @@ Controller Handoff: * Files reviewed: * Validation: * Validation failure history: +* Validation cwd/HEAD proof: * Official validation integrity status: * Terminal review mutation: * Review decision: diff --git a/skills/llm-project-workflow/workflows/work-issue.md b/skills/llm-project-workflow/workflows/work-issue.md index ceef51c..6766523 100644 --- a/skills/llm-project-workflow/workflows/work-issue.md +++ b/skills/llm-project-workflow/workflows/work-issue.md @@ -144,6 +144,14 @@ If the main checkout is dirty before selection, stop and produce a recovery hand If the main checkout becomes dirty during the run, stop and produce a recovery handoff unless the change is explicitly allowed by the canonical workflow. +### Stacked PRs (explicit exception, #484) + +Normal author work stays base-equivalent to `master`/`main`/`dev`. A **stacked PR** — deliberately based on another unmerged PR's branch — is the only sanctioned non-master base, and only when the operator/controller explicitly chooses it: + +- Branch the `branches/` worktree from the dependency's branch, then lock with `gitea_lock_issue(..., stacked_base_branch=, stacked_base_pr=)`. The lock fails closed unless that open PR owns the branch; arbitrary or stale branches are rejected. +- Open the PR with `gitea_create_pr(base=)`. The body must state: `Stacked on PR # / issue #`, `Base branch: `, `Head branch: `, `Do not merge before PR #`, and note retarget/rebase to `master` after the dependency lands if required. +- This does not relax the main-checkout rule or bypass the issue lock — work still happens under `branches/`, and the approved base is recorded on the lock. + ## 5. No raw MCP repair during normal issue work Do not run `pkill`, kill MCP processes, edit MCP config, restart servers, or perform control-checkout repair during normal issue work. diff --git a/stacked_pr_support.py b/stacked_pr_support.py new file mode 100644 index 0000000..e3cab47 --- /dev/null +++ b/stacked_pr_support.py @@ -0,0 +1,211 @@ +"""Stacked-PR support for author issue locks and PR creation (#484). + +Normal author work locks a worktree that is base-equivalent to ``master``/ +``main``/``dev`` and opens a PR against one of those base branches. A *stacked* +PR is deliberately based on another unmerged PR's branch, so its worktree is not +master-equivalent and its PR base is not a normal base branch. + +This module holds the pure decision logic that lets: + +* ``gitea_lock_issue`` approve a non-master base **only** when it is explicitly + declared and proven to correspond to an open pull request, and +* ``gitea_create_pr`` accept that approved base while still rejecting arbitrary, + mismatched, or stale (merged/closed) branches. + +The normal master-based path is unchanged: when no stacked base is declared, and +when the PR base is a normal base branch, these helpers are inert. Nothing here +bypasses the issue lock — a stacked base is recorded *on* the lock and re-checked +at PR time. +""" + +from __future__ import annotations + +BASE_BRANCHES = frozenset({"master", "main", "dev"}) + +# Phrases that satisfy the required merge-ordering statement in a stacked PR body. +MERGE_ORDER_PHRASES = ("do not merge before", "do not merge until") + + +def is_base_branch(base: str | None, base_branches: frozenset[str] | None = None) -> bool: + """True when ``base`` is a normal base branch (master/main/dev).""" + bases = base_branches or BASE_BRANCHES + return (base or "").strip() in bases + + +def _pr_head_ref(pr: dict) -> str: + head = pr.get("head") or {} + if isinstance(head, dict): + return (head.get("ref") or "").strip() + return (str(head) if head else "").strip() + + +def find_open_pr_for_branch(open_prs: list[dict] | None, branch: str | None) -> dict | None: + """Return the first OPEN PR whose head ref equals ``branch`` (else ``None``).""" + branch = (branch or "").strip() + if not branch: + return None + for pr in open_prs or []: + if (pr.get("state") or "").strip().lower() != "open": + continue + if _pr_head_ref(pr) == branch: + return pr + return None + + +def assess_stacked_base_declaration( + *, + stacked_base_branch: str | None, + stacked_base_pr: int | None, + open_prs: list[dict] | None, +) -> dict: + """Validate an explicit stacked-base declaration at lock time. + + Returns a dict with ``block`` (fail closed), ``reasons``, ``declared`` + (whether a stacked base was requested), and ``approved`` (the metadata to + persist on the lock when valid, else ``None``). + """ + branch = (stacked_base_branch or "").strip() + if not branch: + # No stacked base requested — normal master-based lock path. + return {"block": False, "reasons": [], "approved": None, "declared": False} + + if branch in BASE_BRANCHES: + return { + "block": True, + "declared": True, + "approved": None, + "reasons": [ + f"stacked base '{branch}' is already a normal base branch; do not " + "declare a base branch as a stacked base" + ], + } + + if stacked_base_pr is None: + return { + "block": True, + "declared": True, + "approved": None, + "reasons": [ + "stacked base branch declared without stacked_base_pr; a stacked PR " + "must cite the open PR that owns the base branch" + ], + } + + pr = find_open_pr_for_branch(open_prs, branch) + if pr is None: + return { + "block": True, + "declared": True, + "approved": None, + "reasons": [ + f"stacked base branch '{branch}' does not correspond to any OPEN pull " + "request; arbitrary or stale branches are not allowed as stacked bases" + ], + } + + if int(pr.get("number")) != int(stacked_base_pr): + return { + "block": True, + "declared": True, + "approved": None, + "reasons": [ + f"declared stacked_base_pr #{stacked_base_pr} does not match the open " + f"PR #{pr.get('number')} that owns base branch '{branch}'" + ], + } + + return { + "block": False, + "declared": True, + "reasons": [], + "approved": { + "branch": branch, + "pr_number": int(pr.get("number")), + "verified_open": True, + }, + } + + +def assess_stacked_pr_body( + body: str | None, *, base_branch: str | None, pr_number: int | None +) -> list[str]: + """Return the list of missing stacked-PR documentation fields (empty = ok).""" + text = body or "" + low = text.lower() + missing: list[str] = [] + if base_branch and base_branch not in text: + missing.append(f"base branch '{base_branch}'") + if pr_number is not None and f"#{pr_number}" not in text: + missing.append(f"stacked-on PR reference '#{pr_number}'") + if not any(phrase in low for phrase in MERGE_ORDER_PHRASES): + missing.append("merge-ordering statement (e.g. 'Do not merge before PR #')") + return missing + + +def assess_create_pr_base( + *, + base: str | None, + approved_stacked_base: dict | None, + body: str | None, + open_prs: list[dict] | None, + base_branches: frozenset[str] | None = None, +) -> dict: + """Validate the PR base at create time. + + Normal base branches pass through unchanged (``stacked`` False). A non-base + branch is allowed only when it matches the lock's approved stacked base, that + base still has an open PR, and the body documents the stack. + """ + bases = base_branches or BASE_BRANCHES + base = (base or "").strip() + if base in bases: + return {"block": False, "reasons": [], "stacked": False} + + approved = approved_stacked_base or {} + approved_branch = (approved.get("branch") or "").strip() + if not approved_branch: + return { + "block": True, + "stacked": True, + "reasons": [ + f"PR base '{base}' is not one of {'/'.join(sorted(bases))} and the " + "issue lock has no approved stacked base; re-lock with an explicit, " + "proof-backed stacked base to open a stacked PR" + ], + } + + if base != approved_branch: + return { + "block": True, + "stacked": True, + "reasons": [ + f"PR base '{base}' does not match the issue lock's approved stacked " + f"base '{approved_branch}'" + ], + } + + pr = find_open_pr_for_branch(open_prs, base) + if pr is None: + return { + "block": True, + "stacked": True, + "reasons": [ + f"approved stacked base '{base}' no longer corresponds to an OPEN pull " + "request (dependency merged, closed, or stale); retarget/rebase onto " + "master or re-lock against a live base" + ], + } + + pr_number = approved.get("pr_number") or pr.get("number") + missing = assess_stacked_pr_body(body, base_branch=base, pr_number=pr_number) + if missing: + return { + "block": True, + "stacked": True, + "reasons": [ + "stacked PR body must document the stack; missing: " + + ", ".join(missing) + ], + } + + return {"block": False, "reasons": [], "stacked": True, "stacked_base_pr": pr_number} diff --git a/tests/test_issue_lock_adoption.py b/tests/test_issue_lock_adoption.py index a579cbc..270248e 100644 --- a/tests/test_issue_lock_adoption.py +++ b/tests/test_issue_lock_adoption.py @@ -11,6 +11,7 @@ from issue_lock_adoption import ( # noqa: E402 NO_MATCH, assess_own_branch_adoption, build_adoption_proof, + build_non_adoption_lock_proof, ) REQ = "feat/issue-420-server-code-parity" @@ -134,5 +135,93 @@ class TestBuildAdoptionProof(unittest.TestCase): self.assertTrue(proof["no_competing_live_lock_proof"]) +class TestExplicitAdoptionProofFields(unittest.TestCase): + """#477: explicit, citable adoption-proof fields for all outcomes.""" + + def _proof(self, assessment, branch): + return build_adoption_proof( + issue_number=420, + branch_name=branch, + assessment=assessment, + open_pr_checked=True, + competing_lock_checked=True, + lock_file_path="/tmp/example-lock.json", + lock_file_status="written", + ) + + def test_adopt_proof_exposes_explicit_fields(self): + assessment = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": REQ, "commit_sha": "934688a"}], + ) + proof = self._proof(assessment, REQ) + self.assertEqual(proof["adoption_decision"], "ADOPT") + self.assertTrue(proof["adopted"]) + self.assertEqual(proof["adopted_branch"], REQ) + self.assertEqual(proof["adopted_branch_head"], "934688a") + self.assertEqual(proof["competing_branch_check"]["result"], "clear") + self.assertEqual(proof["competing_branch_check"]["competing_branches"], []) + self.assertIn("gitea_create_pr", proof["safe_next_action"]) + self.assertIn("exactly matches", proof["matcher_summary"]) + + def test_block_proof_reports_competing_and_does_not_claim_adoption(self): + assessment = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": "feat/issue-420-rogue"}], + ) + proof = self._proof(assessment, REQ) + self.assertEqual(proof["adoption_decision"], "BLOCK_COMPETING") + self.assertFalse(proof["adopted"]) + self.assertIsNone(proof["adopted_branch"]) + self.assertIsNone(proof["adopted_branch_head"]) + self.assertEqual(proof["competing_branch_check"]["result"], "blocked") + self.assertIn( + "feat/issue-420-rogue", + proof["competing_branch_check"]["competing_branches"], + ) + self.assertIn("fail closed", proof["safe_next_action"]) + + def test_no_match_proof_does_not_claim_adoption(self): + assessment = assess_own_branch_adoption( + issue_number=420, + requested_branch=REQ, + existing_branches=[{"name": "feat/issue-999-unrelated"}], + ) + proof = self._proof(assessment, REQ) + self.assertEqual(proof["adoption_decision"], "NO_MATCH") + self.assertFalse(proof["adopted"]) + self.assertIsNone(proof["adopted_branch"]) + self.assertEqual(proof["competing_branch_check"]["result"], "clear") + + def test_substring_collision_stays_boundary_safe(self): + # issue-42 must not adopt/claim against an issue-420 branch (#440/#477). + own = "feat/issue-42-widget" + assessment = assess_own_branch_adoption( + issue_number=42, + requested_branch=own, + existing_branches=[ + {"name": own, "commit_sha": "abc1234"}, + {"name": "feat/issue-420-server-code-parity"}, + ], + ) + proof = self._proof(assessment, own) + self.assertEqual(proof["adoption_decision"], "ADOPT") + self.assertEqual(proof["adopted_branch"], own) + self.assertEqual(proof["competing_branch_check"]["competing_branches"], []) + + def test_non_adoption_lock_proof_is_adoption_free(self): + proof = build_non_adoption_lock_proof( + issue_number=196, branch_name="feat/issue-196-mutations" + ) + self.assertEqual(proof["adoption_decision"], "NO_MATCH") + self.assertFalse(proof["adopted"]) + self.assertIsNone(proof["adopted_branch"]) + self.assertIsNone(proof["adopted_branch_head"]) + self.assertEqual(proof["competing_branch_check"]["result"], "clear") + self.assertIn("no adoption", proof["safe_next_action"].lower()) + + if __name__ == "__main__": unittest.main() \ No newline at end of file diff --git a/tests/test_issue_lock_store.py b/tests/test_issue_lock_store.py index d87fc73..3928b19 100644 --- a/tests/test_issue_lock_store.py +++ b/tests/test_issue_lock_store.py @@ -157,6 +157,30 @@ class TestIssueLockStore(unittest.TestCase): ) self.assertFalse(ils.has_active_issue_lock("feat/issue-999-other")) + def test_approved_stacked_base_survives_round_trip(self): + # #484: the approved stacked base recorded on the lock must persist so + # gitea_create_pr can validate the non-master base at PR time. + record = _lock_record( + issue_number=482, + branch_name="feat/issue-482-skip-stale-request-changes-pr", + approved_stacked_base={ + "branch": "feat/issue-478-mcp-menu-shell", + "pr_number": 479, + "verified_open": True, + }, + ) + path = ils.lock_file_path( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + issue_number=482, + ) + ils.save_lock_file(path, record) + stored = ils.read_lock_file(path) + self.assertEqual(stored["approved_stacked_base"]["branch"], "feat/issue-478-mcp-menu-shell") + self.assertEqual(stored["approved_stacked_base"]["pr_number"], 479) + self.assertTrue(stored["approved_stacked_base"]["verified_open"]) + def test_atomic_write_preserves_unrelated_lock(self): path_a = ils.lock_file_path( remote="prgs", diff --git a/tests/test_issue_lock_worktree.py b/tests/test_issue_lock_worktree.py index 02347f4..67baa78 100644 --- a/tests/test_issue_lock_worktree.py +++ b/tests/test_issue_lock_worktree.py @@ -72,6 +72,59 @@ class TestIssueLockWorktreeAssessment(unittest.TestCase): self.assertTrue(result["proven"]) +class TestStackedBaseEquivalence(unittest.TestCase): + """extra_bases (an approved stacked base) can anchor base-equivalence (#484).""" + + def _git(self, *args): + import subprocess + + subprocess.run( + ["git", "-C", self.repo, *args], + check=True, + capture_output=True, + text=True, + ) + + def setUp(self): + import subprocess + import tempfile + + self.tmp = tempfile.TemporaryDirectory() + self.repo = self.tmp.name + subprocess.run(["git", "init", "-q", self.repo], check=True, capture_output=True) + self._git("config", "user.email", "t@t") + self._git("config", "user.name", "t") + self._git("commit", "--allow-empty", "-q", "-m", "base") + # Rename the default branch away from master/main/dev so no *base* branch + # exists at HEAD — otherwise HEAD would be base-equivalent for free. + self._git("branch", "-m", "trunk") + # Create a non-master "dependency" branch at the same commit, then a + # feature branch off it — mirrors a stacked worktree. + self._git("branch", "feat/issue-100-dep") + self._git("checkout", "-q", "-b", "feat/issue-101-stacked") + + def tearDown(self): + self.tmp.cleanup() + + def test_stacked_base_not_equivalent_without_extra_bases(self): + state = issue_lock_worktree.read_worktree_git_state(self.repo) + # HEAD does not match master/main/dev, so base-equivalence is False. + self.assertFalse(state["base_equivalent"]) + + def test_stacked_base_equivalent_with_extra_bases(self): + state = issue_lock_worktree.read_worktree_git_state( + self.repo, extra_bases=("feat/issue-100-dep",) + ) + self.assertTrue(state["base_equivalent"]) + self.assertEqual(state["base_branch"], "feat/issue-100-dep") + + def test_unrelated_extra_base_does_not_anchor(self): + state = issue_lock_worktree.read_worktree_git_state( + self.repo, extra_bases=("feat/does-not-exist",) + ) + self.assertFalse(state["base_equivalent"]) + + class TestIssueLockWorktreeResolution(unittest.TestCase): def test_explicit_path_wins(self): resolved = issue_lock_worktree.resolve_author_worktree_path( diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 44d80ce..a5e5e33 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -213,6 +213,12 @@ def test_validation_failure_history_verifier_exported(): assert callable(assess_validation_failure_history_report) +def test_validation_cwd_proof_verifier_exported(): + from review_proofs import assess_validation_cwd_proof_report + + assert callable(assess_validation_cwd_proof_report) + + def test_prior_blocker_skip_verifier_exported(): from review_proofs import assess_prior_blocker_skip_proof diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 3e2e06b..3e7bd17 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3470,6 +3470,45 @@ class TestIssueLocking(unittest.TestCase): self.assertIn("adoption", res) self.assertEqual(res["adoption"]["branch_head_commit"], "abc123") + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=_clean_master_git_state_for_lock(), + ) + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_issue_adoption_response_has_explicit_proof(self, _auth, mock_api, _git_state): + # #477 AC1: the live lock response must carry citable adoption proof. + branch = "feat/issue-196-mutations" + self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"}) + mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}] + res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs") + proof = res["adoption"] + self.assertEqual(proof["adoption_decision"], "ADOPT") + self.assertTrue(proof["adopted"]) + self.assertEqual(proof["adopted_branch"], branch) + self.assertEqual(proof["adopted_branch_head"], "abc123") + self.assertEqual(proof["competing_branch_check"]["result"], "clear") + self.assertIn("gitea_create_pr", proof["safe_next_action"]) + self.assertIn("196", proof["matcher_summary"]) + + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=_clean_master_git_state_for_lock(), + ) + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_issue_no_match_response_does_not_claim_adoption(self, _auth, _api, _git_state): + # #477 AC2/AC3: a normal (NO_MATCH) lock must carry adoption-free proof + # and must NOT expose an ``adoption`` block. + res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") + self.assertTrue(res["success"]) + self.assertNotIn("adoption", res) + check = res["adoption_check"] + self.assertEqual(check["adoption_decision"], "NO_MATCH") + self.assertFalse(check["adopted"]) + self.assertIsNone(check["adopted_branch"]) + self.assertEqual(check["competing_branch_check"]["result"], "clear") + @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", return_value=_clean_master_git_state_for_lock(), diff --git a/tests/test_pr_lease_comments_non_list_guard.py b/tests/test_pr_lease_comments_non_list_guard.py new file mode 100644 index 0000000..08c9239 --- /dev/null +++ b/tests/test_pr_lease_comments_non_list_guard.py @@ -0,0 +1,76 @@ +"""Regression tests for non-list API payloads on PR/issue comment listing (#485).""" +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +from mcp_server import ( # noqa: E402 + _list_pr_lease_comments, + gitea_list_issue_comments, +) + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" +AUTHOR_ENV = { + "GITEA_PROFILE_NAME": "gitea-author", + "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment", +} + + +class TestPrLeaseCommentsNonListGuard(unittest.TestCase): + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_pr_lease_comments_non_list_payload_returns_empty(self, _auth, mock_api): + mock_api.return_value = {"message": "Unauthorized"} + result = _list_pr_lease_comments( + 12, remote="prgs", host=None, org=None, repo=None) + self.assertEqual(result, []) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_pr_lease_comments_none_returns_empty(self, _auth, mock_api): + mock_api.return_value = None + result = _list_pr_lease_comments( + 12, remote="prgs", host=None, org=None, repo=None) + self.assertEqual(result, []) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_pr_lease_comments_list_payload_unchanged(self, _auth, mock_api): + comment = {"id": 7, "body": ""} + mock_api.return_value = [comment] + result = _list_pr_lease_comments( + 12, remote="prgs", host=None, org=None, repo=None, limit=5) + self.assertEqual(result, [comment]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_issue_comments_non_list_payload_returns_empty(self, _auth, mock_api): + mock_api.return_value = {"message": "Unauthorized"} + with patch.dict(os.environ, AUTHOR_ENV, clear=True): + result = gitea_list_issue_comments(issue_number=9, remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(result["comments"], []) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_issue_comments_list_payload_unchanged(self, _auth, mock_api): + mock_api.return_value = [ + { + "id": 101, + "user": {"login": "alice"}, + "body": "hello", + "created_at": "2026-07-03T00:00:00Z", + "updated_at": "2026-07-03T01:00:00Z", + } + ] + with patch.dict(os.environ, AUTHOR_ENV, clear=True): + result = gitea_list_issue_comments(issue_number=9, remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(len(result["comments"]), 1) + self.assertEqual(result["comments"][0]["author"], "alice") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_preflight_read_survival.py b/tests/test_preflight_read_survival.py new file mode 100644 index 0000000..6da8af6 --- /dev/null +++ b/tests/test_preflight_read_survival.py @@ -0,0 +1,106 @@ +"""#469: capability preflight survives interleaved read-only whoami calls.""" + +import os +import unittest + +import gitea_mcp_server as mcp_server + + +class TestPreflightReadSurvival(unittest.TestCase): + def setUp(self): + self.orig_whoami = mcp_server._preflight_whoami_called + self.orig_capability = mcp_server._preflight_capability_called + self.orig_whoami_violation = mcp_server._preflight_whoami_violation + self.orig_capability_violation = mcp_server._preflight_capability_violation + self.orig_resolved_role = mcp_server._preflight_resolved_role + self.orig_resolved_task = mcp_server._preflight_resolved_task + self.orig_process_start = mcp_server._process_start_porcelain + self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain + self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain + for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"): + if key in os.environ: + del os.environ[key] + os.environ["GITEA_TEST_PORCELAIN"] = "" + mcp_server._preflight_whoami_called = False + mcp_server._preflight_capability_called = False + mcp_server._preflight_whoami_violation = False + mcp_server._preflight_capability_violation = False + mcp_server._preflight_resolved_role = None + mcp_server._preflight_resolved_task = None + mcp_server._process_start_porcelain = "" + mcp_server._preflight_whoami_baseline_porcelain = None + mcp_server._preflight_capability_baseline_porcelain = None + + def tearDown(self): + mcp_server._preflight_whoami_called = self.orig_whoami + mcp_server._preflight_capability_called = self.orig_capability + mcp_server._preflight_whoami_violation = self.orig_whoami_violation + mcp_server._preflight_capability_violation = self.orig_capability_violation + mcp_server._preflight_resolved_role = self.orig_resolved_role + mcp_server._preflight_resolved_task = self.orig_resolved_task + mcp_server._process_start_porcelain = self.orig_process_start + mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline + mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline + for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"): + if key in os.environ: + del os.environ[key] + + def test_interleaved_whoami_preserves_capability(self): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check( + "capability", resolved_role="reconciler", resolved_task="close_pr" + ) + self.assertTrue(mcp_server._preflight_capability_called) + self.assertEqual(mcp_server._preflight_resolved_task, "close_pr") + + mcp_server.record_preflight_check("whoami") + self.assertTrue(mcp_server._preflight_capability_called) + self.assertEqual(mcp_server._preflight_resolved_task, "close_pr") + + mcp_server.verify_preflight_purity(task="close_pr") + self.assertFalse(mcp_server._preflight_capability_called) + + def test_missing_capability_still_fails_closed(self): + mcp_server.record_preflight_check("whoami") + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity(task="close_pr") + self.assertIn("has not been resolved", str(ctx.exception)) + + def test_task_mismatch_fails_closed(self): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check( + "capability", resolved_role="author", resolved_task="create_issue" + ) + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity(task="close_pr") + self.assertIn("task mismatch", str(ctx.exception)) + + def test_capability_consumed_after_mutation_gate(self): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check( + "capability", resolved_role="author", resolved_task="create_issue" + ) + mcp_server.verify_preflight_purity(task="create_issue") + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity(task="create_issue") + self.assertIn("has not been resolved", str(ctx.exception)) + + def test_whoami_recovery_after_violation_clears_capability(self): + os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + mcp_server.record_preflight_check("whoami") + self.assertTrue(mcp_server._preflight_whoami_violation) + + del os.environ["GITEA_TEST_FORCE_DIRTY"] + os.environ["GITEA_TEST_PORCELAIN"] = "" + mcp_server.record_preflight_check("whoami") + self.assertFalse(mcp_server._preflight_whoami_violation) + self.assertFalse(mcp_server._preflight_capability_called) + + mcp_server.record_preflight_check( + "capability", resolved_role="reviewer", resolved_task="review_pr" + ) + mcp_server.verify_preflight_purity(task="review_pr") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reconciler_close_workspace_guard.py b/tests/test_reconciler_close_workspace_guard.py new file mode 100644 index 0000000..54a24e4 --- /dev/null +++ b/tests/test_reconciler_close_workspace_guard.py @@ -0,0 +1,89 @@ +"""Reconciler close_pr must not require author branches/ worktree (#468).""" +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server as srv + +FAKE_AUTH = "token test" +CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3]) + +RECONCILER_PROFILE = { + "profile_name": "prgs-reconciler", + "allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.review", + "gitea.pr.create", + "gitea.branch.push", + "gitea.repo.commit", + ], + "audit_label": "prgs-reconciler", +} + + +class TestReconcilerCloseWorkspaceGuard(unittest.TestCase): + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + 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 + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE) + @patch("gitea_mcp_server.api_request") + def test_reconciler_close_pr_from_control_checkout_succeeds( + self, mock_api, _profile, _ns, _auth + ): + srv._preflight_resolved_role = "reconciler" + mock_api.return_value = { + "number": 414, + "title": "old", + "body": "", + "state": "closed", + "html_url": "https://gitea.example.com/pulls/414", + } + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("GITEA_AUTHOR_WORKTREE", None) + os.environ.pop("GITEA_ACTIVE_WORKTREE", None) + result = srv.gitea_edit_pr(414, state="closed", remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(result["state"], "closed") + 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_get_all", return_value=[]) + @patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master"}, + ) + def test_author_create_issue_still_blocked_on_control_checkout( + self, _git, _get_all, _role, _ns, _prof, _auth + ): + srv._preflight_resolved_role = "author" + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue(title="Test", body="body") + self.assertIn("stable control checkout", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_validation_cwd_proof.py b/tests/test_reviewer_validation_cwd_proof.py new file mode 100644 index 0000000..59b81fc --- /dev/null +++ b/tests/test_reviewer_validation_cwd_proof.py @@ -0,0 +1,135 @@ +"""Tests for validation cwd/HEAD proof verifier (#398).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from final_report_validator import assess_final_report_validator # noqa: E402 +from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report # noqa: E402 + +ROOT = "/Users/jasonwalker/Development/Gitea-Tools" +WORKTREE = f"{ROOT}/branches/review-feat-issue-398" +HEAD = "f5953549aad5e822f14f52d3ea3c6d7990109384" + + +def _proof_backed_report() -> str: + return "\n".join([ + f"Candidate head SHA: {HEAD}", + f"pwd: {WORKTREE}", + f"git rev-parse HEAD: {HEAD}", + "git status --short --branch: ## feat/issue-398...prgs/master", + f"Validation command: cd {WORKTREE} && venv/bin/python -m pytest tests/ -q", + "Result: 1497 passed, 6 skipped", + ]) + + +class TestValidationCwdProof(unittest.TestCase): + def test_no_validation_claim_passes(self): + result = assess_validation_cwd_proof_report("Review decision: approve") + self.assertTrue(result["proven"]) + + def test_missing_cwd_proof_fails(self): + result = assess_validation_cwd_proof_report( + f"Validation command: pytest tests/\nCandidate head SHA: {HEAD}", + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_main_checkout_cwd_blocks(self): + result = assess_validation_cwd_proof_report( + "\n".join([ + f"Candidate head SHA: {HEAD}", + f"pwd: {ROOT}", + f"git rev-parse HEAD: {HEAD}", + "git status --short --branch: ## master", + "Validation command: pytest tests/ -q", + ]), + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + project_root=ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["violations"]) + + def test_wrong_head_blocks(self): + wrong = "a" * 40 + result = assess_validation_cwd_proof_report( + "\n".join([ + f"Candidate head SHA: {HEAD}", + f"pwd: {WORKTREE}", + f"git rev-parse HEAD: {wrong}", + "git status --short --branch: clean", + f"Validation command: cd {WORKTREE} && pytest -q", + ]), + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + project_root=ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["violations"]) + + def test_fully_proof_backed_passes(self): + result = assess_validation_cwd_proof_report( + _proof_backed_report(), + validation_session={"validation_ran": True, "expected_head_sha": HEAD}, + project_root=ROOT, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_baseline_without_cwd_fails(self): + result = assess_validation_cwd_proof_report( + "\n".join([ + _proof_backed_report(), + "Baseline validation command: pytest tests/ -q", + ]), + validation_session={ + "validation_ran": True, + "expected_head_sha": HEAD, + "baseline_validation_ran": True, + }, + project_root=ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue( + any("baseline" in r.lower() for r in result["reasons"]) + ) + + def test_baseline_with_full_proof_passes(self): + result = assess_validation_cwd_proof_report( + "\n".join([ + _proof_backed_report(), + f"Baseline worktree: {ROOT}/branches/baseline-master-pr376", + f"Baseline target SHA: {HEAD}", + f"Baseline validation command: cd {ROOT}/branches/baseline-master-pr376 && pytest -q", + ]), + validation_session={ + "validation_ran": True, + "expected_head_sha": HEAD, + "baseline_validation_ran": True, + }, + project_root=ROOT, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_final_report_validator_integration(self): + result = assess_final_report_validator( + "Validation command: pytest tests/ -q", + "review_pr", + validation_session={"validation_ran": True}, + ) + self.assertTrue(result["blocked"] or result["downgraded"]) + self.assertTrue( + any( + f.get("rule_id") == "reviewer.validation_cwd_proof" + for f in result.get("findings") or [] + ) + ) + + def test_exported_from_review_proofs(self): + from review_proofs import assess_validation_cwd_proof_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_stacked_pr_support.py b/tests/test_stacked_pr_support.py new file mode 100644 index 0000000..e592fc1 --- /dev/null +++ b/tests/test_stacked_pr_support.py @@ -0,0 +1,185 @@ +"""Unit tests for stacked-PR base policy (#484).""" +import unittest + +import stacked_pr_support as sps + + +def _pr(number, branch, state="open"): + return {"number": number, "state": state, "head": {"ref": branch}} + + +OPEN_PRS = [ + _pr(479, "feat/issue-478-mcp-menu-shell"), + _pr(481, "feat/issue-477-lock-adoption-proof"), +] + +# Motivating case (#482 stacked on #479 / #478). +STACKED_BODY = ( + "Closes #482.\n\n" + "Stacked on PR #479 / issue #478.\n" + "Base branch: feat/issue-478-mcp-menu-shell\n" + "Head branch: feat/issue-482-skip-stale-request-changes-pr\n" + "Do not merge before PR #479 lands." +) + + +class TestIsBaseBranch(unittest.TestCase): + def test_master_main_dev_are_base(self): + for b in ("master", "main", "dev"): + self.assertTrue(sps.is_base_branch(b)) + + def test_feature_branch_is_not_base(self): + self.assertFalse(sps.is_base_branch("feat/issue-478-mcp-menu-shell")) + + +class TestStackedBaseDeclaration(unittest.TestCase): + def test_no_declaration_is_normal_path(self): + out = sps.assess_stacked_base_declaration( + stacked_base_branch=None, stacked_base_pr=None, open_prs=OPEN_PRS + ) + self.assertFalse(out["block"]) + self.assertFalse(out["declared"]) + self.assertIsNone(out["approved"]) + + def test_valid_open_pr_base_is_approved(self): + out = sps.assess_stacked_base_declaration( + stacked_base_branch="feat/issue-478-mcp-menu-shell", + stacked_base_pr=479, + open_prs=OPEN_PRS, + ) + self.assertFalse(out["block"]) + self.assertEqual(out["approved"]["branch"], "feat/issue-478-mcp-menu-shell") + self.assertEqual(out["approved"]["pr_number"], 479) + self.assertTrue(out["approved"]["verified_open"]) + + def test_missing_pr_number_blocks(self): + out = sps.assess_stacked_base_declaration( + stacked_base_branch="feat/issue-478-mcp-menu-shell", + stacked_base_pr=None, + open_prs=OPEN_PRS, + ) + self.assertTrue(out["block"]) + self.assertIn("without stacked_base_pr", out["reasons"][0]) + + def test_arbitrary_branch_with_no_open_pr_blocks(self): + out = sps.assess_stacked_base_declaration( + stacked_base_branch="feat/random-unrelated-branch", + stacked_base_pr=999, + open_prs=OPEN_PRS, + ) + self.assertTrue(out["block"]) + self.assertIn("does not correspond to any OPEN pull request", out["reasons"][0]) + + def test_stale_merged_base_blocks(self): + merged = [_pr(479, "feat/issue-478-mcp-menu-shell", state="closed")] + out = sps.assess_stacked_base_declaration( + stacked_base_branch="feat/issue-478-mcp-menu-shell", + stacked_base_pr=479, + open_prs=merged, + ) + self.assertTrue(out["block"]) + self.assertIsNone(out["approved"]) + + def test_pr_number_mismatch_blocks(self): + out = sps.assess_stacked_base_declaration( + stacked_base_branch="feat/issue-478-mcp-menu-shell", + stacked_base_pr=481, # wrong PR for this branch + open_prs=OPEN_PRS, + ) + self.assertTrue(out["block"]) + self.assertIn("does not match the open", out["reasons"][0]) + + def test_declaring_a_base_branch_blocks(self): + out = sps.assess_stacked_base_declaration( + stacked_base_branch="master", stacked_base_pr=1, open_prs=OPEN_PRS + ) + self.assertTrue(out["block"]) + self.assertIn("already a normal base branch", out["reasons"][0]) + + +class TestStackedPrBody(unittest.TestCase): + def test_complete_body_has_no_missing_fields(self): + missing = sps.assess_stacked_pr_body( + STACKED_BODY, base_branch="feat/issue-478-mcp-menu-shell", pr_number=479 + ) + self.assertEqual(missing, []) + + def test_missing_all_fields(self): + missing = sps.assess_stacked_pr_body( + "just some text", base_branch="feat/issue-478-mcp-menu-shell", pr_number=479 + ) + self.assertEqual(len(missing), 3) + + def test_missing_merge_ordering_only(self): + body = "Base branch feat/issue-478-mcp-menu-shell for PR #479" + missing = sps.assess_stacked_pr_body( + body, base_branch="feat/issue-478-mcp-menu-shell", pr_number=479 + ) + self.assertEqual(len(missing), 1) + self.assertIn("merge-ordering", missing[0]) + + +class TestCreatePrBase(unittest.TestCase): + APPROVED = {"branch": "feat/issue-478-mcp-menu-shell", "pr_number": 479, "verified_open": True} + + def test_master_base_passes_without_stacked_metadata(self): + out = sps.assess_create_pr_base( + base="master", approved_stacked_base=None, body="Closes #1", open_prs=[] + ) + self.assertFalse(out["block"]) + self.assertFalse(out["stacked"]) + + def test_non_base_without_approval_blocks(self): + out = sps.assess_create_pr_base( + base="feat/issue-478-mcp-menu-shell", + approved_stacked_base=None, + body=STACKED_BODY, + open_prs=OPEN_PRS, + ) + self.assertTrue(out["block"]) + self.assertIn("no approved stacked base", out["reasons"][0]) + + def test_non_base_mismatched_approval_blocks(self): + out = sps.assess_create_pr_base( + base="feat/some-other-branch", + approved_stacked_base=self.APPROVED, + body=STACKED_BODY, + open_prs=OPEN_PRS, + ) + self.assertTrue(out["block"]) + self.assertIn("does not match the issue lock's approved stacked base", out["reasons"][0]) + + def test_approved_base_with_good_body_passes(self): + out = sps.assess_create_pr_base( + base="feat/issue-478-mcp-menu-shell", + approved_stacked_base=self.APPROVED, + body=STACKED_BODY, + open_prs=OPEN_PRS, + ) + self.assertFalse(out["block"]) + self.assertTrue(out["stacked"]) + self.assertEqual(out["stacked_base_pr"], 479) + + def test_approved_base_now_stale_blocks(self): + out = sps.assess_create_pr_base( + base="feat/issue-478-mcp-menu-shell", + approved_stacked_base=self.APPROVED, + body=STACKED_BODY, + open_prs=[_pr(479, "feat/issue-478-mcp-menu-shell", state="merged")], + ) + self.assertTrue(out["block"]) + self.assertIn("no longer corresponds to an OPEN", out["reasons"][0]) + + def test_approved_base_with_incomplete_body_blocks(self): + out = sps.assess_create_pr_base( + base="feat/issue-478-mcp-menu-shell", + approved_stacked_base=self.APPROVED, + body="Closes #482 only", + open_prs=OPEN_PRS, + ) + self.assertTrue(out["block"]) + self.assertIn("must document the stack", out["reasons"][0]) + + +if __name__ == "__main__": + unittest.main()