diff --git a/docs/gitea-execution-profiles.md b/docs/gitea-execution-profiles.md index 3b15843..226ddd8 100644 --- a/docs/gitea-execution-profiles.md +++ b/docs/gitea-execution-profiles.md @@ -317,6 +317,44 @@ Least-privilege constraints: canonical names such as `gitea.pr.close` (never bare `pr.close` / `issue.close`, which the production normalizer rejects or drops). +### Post-merge moot-lease cleanup ownership (`gitea.pr.comment`) + +Neutralising a reviewer lease left behind on an already-merged/closed PR is +reconciliation work too. `task_capability_map` maps +`cleanup_post_merge_moot_lease` — and its tool-name alias +`gitea_cleanup_post_merge_moot_lease` — to role `reconciler` with permission +`gitea.pr.comment` (#745). Both names carry the **same** contract. + +The permission alone is deliberately not sufficient: author, reviewer and +merger profiles all hold `gitea.pr.comment` for ordinary PR discussion, so the +role gate — not the permission gate — is what keeps the terminal lease marker +reconciler-owned. + +`gitea_cleanup_post_merge_moot_lease` splits its two modes on purpose: + +- **`apply=false` (assessment) requires only `gitea.read`, with no role gate.** + This matches `gitea_cleanup_stale_review_decision_lock` and + `gitea_cleanup_obsolete_reviewer_comment_lease`, whose assessment paths are + likewise read-gated, so an operator can diagnose a stuck lease from whichever + namespace happens to be attached without switching roles. The dry run + performs no mutation and records append-only evidence in-session. +- **`apply=true` (mutation) requires all of the following**, in order: the + session must have resolved exactly `cleanup_post_merge_moot_lease` (resolving + any other task — including a sibling reconciler task — does not authorize + it); the active role must be `reconciler`; the profile must hold + `gitea.pr.comment`; the explicit `org`/`repo` must agree with the canonical + repository identity, which is derived from the session binding and can never + be overridden by request parameters; and matching dry-run evidence must show + `lease_moot`, `cleanup_allowed`, and the same PR, lease session, candidate + head and lease marker id that are live at apply time. + +Everything else fails closed: a live lease on an open PR, an already-terminal +(idempotent) lease, a lease superseded between the dry run and the apply, a +malformed lease missing session/head/marker, and any foreign-repository target. +The cleanup only ever appends a terminal `phase: released` marker +(`blocker: post-merge-moot`) — it never edits or deletes another session's +comment, and it never merges or adopts a lease. + Launch a static `gitea-reconciler` MCP namespace with `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by `reconciler_profile.assess_reconciler_profile` (#304). Use the diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 456f94b..9551534 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1829,6 +1829,7 @@ def _seed_session_context( import issue_work_duplicate_gate # noqa: E402 import issue_workflow_labels # noqa: E402 import reviewer_pr_lease # noqa: E402 +import post_merge_moot_lease_gate # noqa: E402 # #745 reconciler cleanup gate import merger_lease_adoption # noqa: E402 import merged_cleanup_reconcile # noqa: E402 import worktree_cleanup_audit # noqa: E402 @@ -8859,13 +8860,14 @@ def gitea_review_pr( return out -def _delete_branch_repository_binding_block( +def _repository_binding_block( remote: str | None, *, org: str | None, repo: str | None, + required_permission: str = "gitea.branch.delete", ) -> dict | None: - """#733: validate an explicit delete target against the workspace binding. + """#733: validate an explicit mutation target against the workspace binding. The trusted repository identity comes only from the verified, workspace-aligned git remote (never ``REMOTES`` defaults, never @@ -8898,7 +8900,7 @@ def _delete_branch_repository_binding_block( return { "success": False, "performed": False, - "required_permission": "gitea.branch.delete", + "required_permission": required_permission, "reasons": canonical_reasons, "blocker_kind": "repository_binding", } @@ -8917,9 +8919,9 @@ def _delete_branch_repository_binding_block( return { "success": False, "performed": False, - "required_permission": "gitea.branch.delete", + "required_permission": required_permission, "reasons": list(override.get("reasons") or [ - "delete target repository does not match the workspace " + "target repository does not match the workspace " "binding (fail closed)" ]), "blocker_kind": "repository_binding", @@ -8929,17 +8931,51 @@ def _delete_branch_repository_binding_block( return { "success": False, "performed": False, - "required_permission": "gitea.branch.delete", + "required_permission": required_permission, "reasons": [ "repository binding unverified: no workspace repository " "identity could be established to corroborate the explicit " - "delete target (fail closed)" + "target (fail closed)" ], "blocker_kind": "repository_binding", } return None +def _delete_branch_repository_binding_block( + remote: str | None, + *, + org: str | None, + repo: str | None, +) -> dict | None: + """Delete-branch view of :func:`_repository_binding_block` (#733). + + Retained as the named entry point for the delete path so #733/#739 + regression coverage keeps exercising the exact permission label that + ``gitea_delete_branch`` reports. + """ + return _repository_binding_block( + remote, org=org, repo=repo, + required_permission="gitea.branch.delete", + ) + + +def _bound_repository_slug(remote: str | None) -> str | None: + """Canonical repository slug for the active session, or None (#745). + + Same trust order as ``_repository_binding_block``: the configured canonical + root when the namespace declares one, otherwise the workspace-derived + identity. Request parameters are never consulted, so they cannot redirect a + mutation at a foreign repository. + """ + canonical_slug, canonical_reasons = _canonical_repository_slug( + get_profile(), remote + ) + if canonical_reasons: + return None + return canonical_slug or _workspace_repository_slug(remote) + + @mcp.tool() def gitea_delete_branch( branch: str, @@ -12352,6 +12388,10 @@ def gitea_diagnose_reviewer_pr_lease_handoff( def gitea_cleanup_post_merge_moot_lease( pr_number: int, apply: bool = False, + expected_session_id: str | None = None, + expected_candidate_head: str | None = None, + expected_lease_comment_id: int | None = None, + worktree_path: str | None = None, remote: str = "dadeschools", host: str | None = None, org: str | None = None, @@ -12367,14 +12407,37 @@ def gitea_cleanup_post_merge_moot_lease( an append-only comment that neutralises the moot lease without deleting any other session's comment. + Role binding (#745). The two modes are gated differently, on purpose: + + * ``apply=false`` (assessment) stays reachable under ``gitea.read`` for + **any** role, matching ``gitea_cleanup_stale_review_decision_lock`` and + ``gitea_cleanup_obsolete_reviewer_comment_lease``, so an operator can + diagnose a stuck lease from whichever namespace is attached. It performs + no mutation and records append-only dry-run evidence in-session. + * ``apply=true`` (mutation) additionally requires, in order: the session to + have resolved exactly ``cleanup_post_merge_moot_lease``; the + ``reconciler`` role; ``gitea.pr.comment``; a validated repository / + canonical-root / workspace binding; and matching dry-run evidence proving + ``lease_moot`` and ``cleanup_allowed`` for the same PR, lease session, + candidate head and lease marker. Author, reviewer and merger fail closed + here even though their profiles carry ``gitea.pr.comment``. + Args: pr_number: The PR whose lingering lease to assess/clean. apply: When false (default) report only (read-only). When true, post the - terminal released marker if — and only if — cleanup is allowed. + terminal released marker if — and only if — cleanup is authorized. + expected_session_id: Optional lease session the caller expects; a + mismatch against the live lease fails closed. + expected_candidate_head: Optional leased head the caller expects; a + mismatch against the live lease fails closed. + expected_lease_comment_id: Optional lease marker id the caller expects; + a mismatch against the live lease fails closed. + worktree_path: Reconciler worktree to bind the apply-path preflight to. remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. - org: Override the owner/organization. - repo: Override the repository name. + org: Override the owner/organization. Validated against the canonical + repository identity; it can never redirect the mutation. + repo: Override the repository name. Validated as above. Returns: dict reporting PR merged/closed state, merge_commit_sha, linked-issue @@ -12434,14 +12497,62 @@ def gitea_cleanup_post_merge_moot_lease( "reasons": assessment.get("reasons") or [], } + # The canonical repository identity comes from the session binding only — + # never from the org/repo request parameters (#745 requirement 10). + repository_slug = _bound_repository_slug(remote) + report["repository_slug"] = repository_slug + report["required_task"] = post_merge_moot_lease_gate.CLEANUP_TASK + report["required_role_kind"] = post_merge_moot_lease_gate.REQUIRED_ROLE + if not apply: + # Append-only dry-run evidence. Recorded for every assessment, allowed + # or not, so a later apply can prove the lease it saw is the lease that + # is still there. Prior entries are never rewritten. + evidence = post_merge_moot_lease_gate.record_dry_run( + pr_number=pr_number, + repository_slug=repository_slug, + lease_moot=bool(assessment.get("is_moot")), + cleanup_allowed=bool(assessment.get("cleanup_allowed")), + session_id=active.get("session_id"), + candidate_head=active.get("candidate_head"), + lease_comment_id=active.get("comment_id"), + ) + report["dry_run_evidence"] = evidence return report + + # ---------------------------------------------------------------- apply -- + # Preserve the existing dry-run safety assessment: a lease that is not moot + # (open PR, already-terminal, absent) is refused here exactly as before, and + # no mutation is attempted. if not assessment.get("cleanup_allowed"): report["cleanup_skipped_reason"] = ( assessment.get("reasons") or ["cleanup not allowed"] ) return report + # 1 + 2 + 6: exact resolved cleanup task, reconciler role, and matching + # append-only dry-run evidence. Permission alone is deliberately not enough. + authorization = post_merge_moot_lease_gate.assess_apply_authorization( + pr_number=pr_number, + repository_slug=repository_slug, + resolved_task=_preflight_resolved_task, + active_role_kind=_profile_role_kind(get_profile()), + assessment=assessment, + evidence=post_merge_moot_lease_gate.latest_dry_run( + pr_number=pr_number, repository_slug=repository_slug + ), + expected_session_id=expected_session_id, + expected_candidate_head=expected_candidate_head, + expected_lease_comment_id=expected_lease_comment_id, + ) + report["authorization"] = authorization + if not authorization["allowed"]: + report["success"] = False + report["reasons"] = authorization["reasons"] + report["blocker_kind"] = authorization["blocker_kind"] + return report + + # 3: the dedicated mutation permission. comment_block = _profile_operation_gate("gitea.pr.comment") if comment_block: report["success"] = False @@ -12449,7 +12560,28 @@ def gitea_cleanup_post_merge_moot_lease( report["permission_report"] = _permission_block_report("gitea.pr.comment") return report - verify_preflight_purity(remote) + # 4: explicit target must agree with the canonical repository identity + # before any preflight or mutation (#733/#739). + repo_binding_block = _repository_binding_block( + remote, org=org, repo=repo, + required_permission="gitea.pr.comment", + ) + if repo_binding_block: + report["success"] = False + report["reasons"] = repo_binding_block["reasons"] + report["blocker_kind"] = repo_binding_block["blocker_kind"] + return report + + # 4 (cont.): canonical root, workspace binding and the resolved-task match + # are enforced by the shared preflight, with the explicit target forwarded + # so the #604 anti-stomp resolution validates the targeted repository. + verify_preflight_purity( + remote, + worktree_path=worktree_path, + task=post_merge_moot_lease_gate.CLEANUP_TASK, + org=org, + repo=repo, + ) body = assessment["release_body"] comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments" with _audited( diff --git a/post_merge_moot_lease_gate.py b/post_merge_moot_lease_gate.py new file mode 100644 index 0000000..846f4dc --- /dev/null +++ b/post_merge_moot_lease_gate.py @@ -0,0 +1,279 @@ +"""Reconciler authorization gate for post-merge moot-lease cleanup (#745). + +``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase: +released`` lease marker — a real, durable mutation of the PR lease ledger. +Before #745 it was gated on permissions alone (``gitea.read`` to enter, +``gitea.pr.comment`` to apply) with no canonical task and no role binding, so +any profile carrying ``gitea.pr.comment`` reached the mutation path while the +reconciler could not satisfy the operator-required resolve-exact-task -> +mutation sequence. + +This module holds the pure half of that gate: + +* the canonical task name and its tool-name alias; +* an **append-only** in-process ledger of read-only dry-run assessments; +* ``assess_apply_authorization``, which decides whether an apply may proceed. + +Apply is authorized only when all of the following hold: + +* the session resolved exactly the cleanup task (no other task substitutes); +* the active profile role is ``reconciler``; +* a prior dry run in this session recorded ``lease_moot`` and + ``cleanup_allowed`` for the *same* repository, PR, lease session, candidate + head and lease marker id; +* the live assessment still agrees with that evidence, so a lease superseded + between the dry run and the apply fails closed; +* any caller-supplied expectations match the live lease exactly. + +Everything else fails closed. The ledger is only ever appended to — a +superseded dry run stays visible as history instead of being rewritten — which +keeps the cleanup audit trail append-only end to end. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +CLEANUP_TASK = "cleanup_post_merge_moot_lease" +CLEANUP_TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease" +REQUIRED_ROLE = "reconciler" +REQUIRED_PERMISSION = "gitea.pr.comment" + +# The read-only assessment stays reachable under gitea.read for every role — +# the convention shared with cleanup_stale_review_decision_lock and +# cleanup_obsolete_reviewer_comment_lease — so any namespace can diagnose a +# stuck lease. Only the apply path demands CLEANUP_TASK + REQUIRED_ROLE. +ASSESSMENT_PERMISSION = "gitea.read" + +_DRY_RUN_LEDGER: list[dict[str, Any]] = [] + + +def _norm(value: Any) -> str: + return str(value or "").strip() + + +def _norm_comment_id(value: Any) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def record_dry_run( + *, + pr_number: int, + repository_slug: str | None, + lease_moot: bool, + cleanup_allowed: bool, + session_id: str | None, + candidate_head: str | None, + lease_comment_id: Any, + recorded_at: datetime | None = None, +) -> dict[str, Any]: + """Append one read-only assessment to the dry-run ledger. + + Never rewrites or removes a prior entry: repeated dry runs accumulate and + ``latest_dry_run`` returns the newest matching one. + """ + entry = { + "task": CLEANUP_TASK, + "pr_number": int(pr_number), + "repository_slug": _norm(repository_slug) or None, + "lease_moot": bool(lease_moot), + "cleanup_allowed": bool(cleanup_allowed), + "session_id": _norm(session_id) or None, + "candidate_head": _norm(candidate_head) or None, + "lease_comment_id": _norm_comment_id(lease_comment_id), + "recorded_at": (recorded_at or datetime.now(timezone.utc)).isoformat(), + } + _DRY_RUN_LEDGER.append(entry) + return dict(entry) + + +def dry_run_history() -> tuple[dict[str, Any], ...]: + """Immutable view of every recorded dry run, oldest first.""" + return tuple(dict(entry) for entry in _DRY_RUN_LEDGER) + + +def latest_dry_run( + *, pr_number: int, repository_slug: str | None +) -> dict[str, Any] | None: + """Newest dry-run evidence for this repository + PR, or None.""" + wanted_repo = _norm(repository_slug) + for entry in reversed(_DRY_RUN_LEDGER): + if entry["pr_number"] != int(pr_number): + continue + if _norm(entry.get("repository_slug")) != wanted_repo: + continue + return dict(entry) + return None + + +def _reset_for_testing() -> None: + """Drop ledger state between tests. Never called by production paths.""" + _DRY_RUN_LEDGER.clear() + + +def assess_apply_authorization( + *, + pr_number: int, + repository_slug: str | None, + resolved_task: str | None, + active_role_kind: str | None, + assessment: dict[str, Any], + evidence: dict[str, Any] | None, + expected_session_id: str | None = None, + expected_candidate_head: str | None = None, + expected_lease_comment_id: Any = None, +) -> dict[str, Any]: + """Decide whether a moot-lease cleanup apply is authorized (fail closed). + + Returns ``{"allowed", "reasons", "blocker_kind", "evidence_matched", ...}``. + ``allowed`` is True only when every check passes; each failure contributes a + reason so the caller can report all of them together. + """ + reasons: list[str] = [] + blocker_kind: str | None = None + + def _block(kind: str, reason: str) -> None: + nonlocal blocker_kind + reasons.append(reason) + if blocker_kind is None: + blocker_kind = kind + + # 1. Exact resolved cleanup task. Resolving any other task — including a + # sibling reconciler task — does not authorize this mutation. + if _norm(resolved_task) != CLEANUP_TASK: + _block( + "unresolved_cleanup_task", + "post-merge moot-lease cleanup requires the session to resolve " + f"task '{CLEANUP_TASK}' immediately before apply; resolved task is " + f"{resolved_task!r} (fail closed)", + ) + + # 2. Dedicated reconciler role, enforced independently of the permission. + if _norm(active_role_kind) != REQUIRED_ROLE: + _block( + "wrong_role", + f"profile role {active_role_kind!r} cannot apply post-merge " + f"moot-lease cleanup; required role is {REQUIRED_ROLE} even when " + f"{REQUIRED_PERMISSION} is present (fail closed)", + ) + + # 3. Canonical repository identity must be established, never inferred from + # request parameters. + if not _norm(repository_slug): + _block( + "repository_binding", + "no canonical repository identity could be established for the " + "cleanup target (fail closed)", + ) + + # 4. The live safety assessment must still say the lease is moot/cleanable. + if not assessment.get("is_moot") or not assessment.get("cleanup_allowed"): + _block( + "lease_not_moot", + "live assessment does not report a moot, cleanable lease on PR " + f"#{pr_number} (lease_moot={bool(assessment.get('is_moot'))}, " + f"cleanup_allowed={bool(assessment.get('cleanup_allowed'))}) " + "(fail closed)", + ) + + live = assessment.get("active_lease") or {} + live_session = _norm(live.get("session_id")) + live_head = _norm(live.get("candidate_head")) + live_comment_id = _norm_comment_id(live.get("comment_id")) + + # 5. A lease missing identifying fields is malformed and unsafe to act on. + if not live_session or not live_head or live_comment_id is None: + _block( + "malformed_lease", + "active lease is malformed: session_id / candidate_head / " + "comment_id must all be present to authorize cleanup " + f"(session_id={live.get('session_id')!r}, " + f"candidate_head={live.get('candidate_head')!r}, " + f"comment_id={live.get('comment_id')!r}) (fail closed)", + ) + + # 6. Caller expectations, when supplied, must match the live lease exactly. + if expected_session_id is not None and _norm(expected_session_id) != live_session: + _block( + "lease_mismatch", + f"expected lease session {expected_session_id!r} does not match the " + f"live lease session {live.get('session_id')!r} (fail closed)", + ) + if ( + expected_candidate_head is not None + and _norm(expected_candidate_head) != live_head + ): + _block( + "lease_mismatch", + f"expected candidate head {expected_candidate_head!r} does not " + f"match the live lease head {live.get('candidate_head')!r} " + "(fail closed)", + ) + if expected_lease_comment_id is not None and ( + _norm_comment_id(expected_lease_comment_id) != live_comment_id + ): + _block( + "lease_mismatch", + f"expected lease marker {expected_lease_comment_id!r} does not " + f"match the live lease marker {live.get('comment_id')!r} " + "(fail closed)", + ) + + # 7. Matching dry-run evidence recorded earlier in this session. + evidence_matched = False + if evidence is None: + _block( + "missing_dry_run_evidence", + "no read-only dry run recorded for this repository and PR; run the " + "tool with apply=false and confirm lease_moot / cleanup_allowed " + "before applying (fail closed)", + ) + elif not evidence.get("lease_moot") or not evidence.get("cleanup_allowed"): + _block( + "dry_run_not_allowed", + "recorded dry run did not report an allowed cleanup " + f"(lease_moot={bool(evidence.get('lease_moot'))}, " + f"cleanup_allowed={bool(evidence.get('cleanup_allowed'))}) " + "(fail closed)", + ) + elif int(evidence.get("pr_number") or -1) != int(pr_number) or _norm( + evidence.get("repository_slug") + ) != _norm(repository_slug): + _block( + "dry_run_mismatch", + "recorded dry run targets a different repository or PR " + f"({evidence.get('repository_slug')}#{evidence.get('pr_number')} vs " + f"{repository_slug}#{pr_number}) (fail closed)", + ) + elif ( + _norm(evidence.get("session_id")) != live_session + or _norm(evidence.get("candidate_head")) != live_head + or _norm_comment_id(evidence.get("lease_comment_id")) != live_comment_id + ): + _block( + "superseded_lease", + "the lease changed after the recorded dry run (dry run: " + f"session={evidence.get('session_id')!r}, " + f"head={evidence.get('candidate_head')!r}, " + f"marker={evidence.get('lease_comment_id')!r}; live: " + f"session={live.get('session_id')!r}, " + f"head={live.get('candidate_head')!r}, " + f"marker={live.get('comment_id')!r}); re-run the dry run " + "(fail closed)", + ) + else: + evidence_matched = True + + return { + "allowed": not reasons, + "reasons": reasons, + "blocker_kind": blocker_kind, + "evidence_matched": evidence_matched, + "required_task": CLEANUP_TASK, + "required_role_kind": REQUIRED_ROLE, + "required_permission": REQUIRED_PERMISSION, + } diff --git a/role_session_router.py b/role_session_router.py index 8ae69a4..4f331b8 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -87,6 +87,11 @@ RECONCILER_TASKS = frozenset({ # only to the reconciler profile). Raw gitea_delete_branch redirects here to # the guarded gitea_cleanup_merged_pr_branch path (#514/#687). "delete_branch", + # #745: post-merge moot reviewer-lease cleanup is reconciler-owned; the + # apply path posts a terminal lease marker. Kept in step with + # task_capability_map so map and router cannot disagree (#723 defect A). + "cleanup_post_merge_moot_lease", + "gitea_cleanup_post_merge_moot_lease", "reconcile_already_landed_pr", "reconcile_already_landed", "reconcile-landed-pr", @@ -118,6 +123,9 @@ TASK_REQUIRED_ROLE = { "reconcile_already_landed": "reconciler", "reconcile-landed-pr": "reconciler", "cleanup_merged_pr_branch": "reconciler", + # #745: post-merge moot reviewer-lease cleanup (canonical task + tool alias). + "cleanup_post_merge_moot_lease": "reconciler", + "gitea_cleanup_post_merge_moot_lease": "reconciler", # #309: reconciler tasks close already-landed PRs/issues only. "reconcile_close_landed_pr": "reconciler", "reconcile_close_landed_issue": "reconciler", diff --git a/task_capability_map.py b/task_capability_map.py index 83e9dcb..adf3524 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -158,6 +158,23 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.comment", "role": "reviewer", }, + # #745: post-merge moot reviewer-lease cleanup is reconciler-owned. The + # apply path posts an append-only terminal `phase: released` lease marker + # (gitea.pr.comment), so holding the comment permission alone must not + # authorize it — author, reviewer and merger fail closed on the role gate + # even though their profiles carry gitea.pr.comment. The read-only + # `apply=false` assessment deliberately stays reachable under gitea.read + # inside the tool (the same convention as cleanup_stale_review_decision_lock + # below), so any namespace can diagnose a stuck lease; only apply requires + # this task plus the reconciler role. + "cleanup_post_merge_moot_lease": { + "permission": "gitea.pr.comment", + "role": "reconciler", + }, + "gitea_cleanup_post_merge_moot_lease": { + "permission": "gitea.pr.comment", + "role": "reconciler", + }, "blind_pr_queue_review": { "permission": "gitea.pr.review", "role": "reviewer", diff --git a/tests/test_issue_745_moot_lease_reconciler_gate.py b/tests/test_issue_745_moot_lease_reconciler_gate.py new file mode 100644 index 0000000..20d3ecb --- /dev/null +++ b/tests/test_issue_745_moot_lease_reconciler_gate.py @@ -0,0 +1,658 @@ +"""Reconciler role binding for post-merge moot-lease cleanup (#745). + +``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase: +released`` lease marker but had no capability-map entry and no role gate: entry +required only ``gitea.read``, apply required only ``gitea.pr.comment``, so any +profile holding the comment permission reached the mutation while the +reconciler could not satisfy resolve-exact-task -> mutation. + +These tests pin the fixed contract: + +- the canonical task and its tool-name alias resolve identically + (``gitea.pr.comment`` + ``reconciler``); +- only a reconciler profile satisfies permission AND role; +- ``apply=false`` assessment stays reachable under ``gitea.read`` for any role + and mutates nothing (documented, deliberate divergence from the apply path); +- ``apply=true`` requires the exact resolved task, the reconciler role, the + comment permission, a validated repository binding and matching dry-run + evidence; +- live/non-moot/superseded/mismatched/malformed leases fail closed; +- the dry-run ledger is append-only and cleanup stays idempotent. + +Every fixture is synthetic. No production PR, lease session or marker is used +anywhere in this module (see ``TestNoProductionLeaseTouched``). +""" + +import sys as _sys +from pathlib import Path as _Path + +_sys.path.insert(0, str(_Path(__file__).resolve().parent.parent)) + +import os # noqa: E402 +import unittest # noqa: E402 +from datetime import datetime, timezone # noqa: E402 +from unittest.mock import patch # noqa: E402 + +import mcp_server # noqa: E402 +import post_merge_moot_lease_gate as gate # noqa: E402 +import reviewer_pr_lease as leases # noqa: E402 +from mcp_server import gitea_cleanup_post_merge_moot_lease # noqa: E402 +from role_session_router import RECONCILER_TASKS, TASK_REQUIRED_ROLE # noqa: E402 +from task_capability_map import required_permission, required_role # noqa: E402 + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" +SLUG = "Scaled-Tech-Consulting/Gitea-Tools" +PR = 487 +ISSUE = 485 +SESSION = "97274-676d20a825c4" +HEAD_A = "a" * 40 +HEAD_B = "d" * 40 +LEASE_COMMENT_ID = 6603 + +CANONICAL_TASK = "cleanup_post_merge_moot_lease" +TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease" + +_BASE_OPS = "gitea.read,gitea.pr.comment" +RECONCILER_ENV = { + "GITEA_PROFILE_NAME": "prgs-reconciler", + "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.close,gitea.branch.delete", +} +AUTHOR_ENV = { + "GITEA_PROFILE_NAME": "prgs-author", + "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.create,gitea.issue.create", +} +REVIEWER_ENV = { + "GITEA_PROFILE_NAME": "prgs-reviewer", + "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.review,gitea.pr.approve", +} +MERGER_ENV = { + "GITEA_PROFILE_NAME": "prgs-merger", + "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.merge", +} + +# Permission shape of the configured role profiles (mirrors +# tests/test_task_capability_role_invariants.py CANONICAL_ROLE_PROFILES). +ROLE_PROFILE_PERMISSIONS = { + "author": {"gitea.read", "gitea.pr.comment", "gitea.pr.create", + "gitea.issue.create", "gitea.issue.comment", "gitea.issue.close", + "gitea.branch.create", "gitea.branch.push", "gitea.repo.commit"}, + "reviewer": {"gitea.read", "gitea.pr.comment", "gitea.pr.review", + "gitea.pr.approve", "gitea.pr.request_changes", + "gitea.issue.comment"}, + "merger": {"gitea.read", "gitea.pr.comment", "gitea.pr.merge", + "gitea.issue.comment"}, + "reconciler": {"gitea.read", "gitea.pr.comment", "gitea.pr.close", + "gitea.issue.close", "gitea.branch.delete"}, +} + + +def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed", + candidate_head=HEAD_A, comment_id=LEASE_COMMENT_ID): + body = leases.format_lease_body( + repo=SLUG, + pr_number=pr_number, + issue_number=ISSUE, + reviewer_identity="sysadmin", + profile="prgs-reviewer", + session_id=session_id, + worktree="branches/review-pr487", + phase=phase, + candidate_head=candidate_head, + target_branch="master", + target_branch_sha="b" * 40, + last_activity=datetime.now(timezone.utc), + ) + return {"id": comment_id, "body": body, "user": {"login": "sysadmin"}} + + +def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999): + """api_request side effect keyed on method + url; records POSTs.""" + calls = {"post": []} + + def _side(method, url, auth=None, payload=None, *a, **k): + if (method or "").upper() == "POST": + calls["post"].append({"url": url, "payload": payload}) + return {"id": posted_id} + if "/comments" in url: + return list(comments) + if "/pulls/" in url: + pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40} + if pr_merged: + pr["merged"] = True + pr["merged_at"] = "2026-07-08T07:46:04Z" + return pr + if "/issues/" in url: + return {"state": "closed" if pr_merged else "open", "number": ISSUE} + return {} + + return _side, calls + + +def _assessment(comments, *, pr_merged=True, pr_state="closed"): + return leases.assess_post_merge_moot_lease( + comments, pr_number=PR, pr_merged=pr_merged, pr_state=pr_state, + merge_commit_sha="c" * 40, + ) + + +# --------------------------------------------------------------------------- # +# 1-5. Capability map / router contract +# --------------------------------------------------------------------------- # +class TestCleanupTaskContract(unittest.TestCase): + def test_canonical_task_is_reconciler_owned(self): + """1. The reconciler is the role that can resolve the cleanup task.""" + self.assertEqual(required_permission(CANONICAL_TASK), "gitea.pr.comment") + self.assertEqual(required_role(CANONICAL_TASK), "reconciler") + + def test_tool_alias_resolves_to_identical_contract(self): + """5. Alias and canonical task must not diverge.""" + self.assertEqual( + (required_permission(TOOL_ALIAS), required_role(TOOL_ALIAS)), + (required_permission(CANONICAL_TASK), required_role(CANONICAL_TASK)), + ) + + def test_author_reviewer_merger_cannot_resolve_the_task(self): + """2-4. No non-reconciler role satisfies permission AND role.""" + for role in ("author", "reviewer", "merger"): + with self.subTest(role=role): + self.assertNotEqual(required_role(CANONICAL_TASK), role) + # They hold the permission — which is exactly why the role gate + # is required rather than optional. + self.assertIn( + "gitea.pr.comment", ROLE_PROFILE_PERMISSIONS[role], + "test premise: non-reconciler roles do hold pr.comment", + ) + + def test_reconciler_profile_satisfies_permission_and_role(self): + self.assertIn( + required_permission(CANONICAL_TASK), + ROLE_PROFILE_PERMISSIONS[required_role(CANONICAL_TASK)], + ) + + def test_router_agrees_with_capability_map(self): + for task in (CANONICAL_TASK, TOOL_ALIAS): + with self.subTest(task=task): + self.assertIn(task, RECONCILER_TASKS) + self.assertEqual(TASK_REQUIRED_ROLE[task], required_role(task)) + + def test_unknown_alias_still_rejected(self): + for bogus in ("cleanup_post_merge_moot_leases", "cleanup_moot_lease", ""): + with self.subTest(task=bogus): + with self.assertRaises(KeyError): + required_role(bogus) + with self.assertRaises(KeyError): + required_permission(bogus) + + +# --------------------------------------------------------------------------- # +# Authorization gate unit tests +# --------------------------------------------------------------------------- # +class TestApplyAuthorizationGate(unittest.TestCase): + def setUp(self): + gate._reset_for_testing() + self.addCleanup(gate._reset_for_testing) + self.assessment = _assessment([_lease_comment()]) + + def _evidence(self, **over): + base = dict( + pr_number=PR, repository_slug=SLUG, lease_moot=True, + cleanup_allowed=True, session_id=SESSION, candidate_head=HEAD_A, + lease_comment_id=LEASE_COMMENT_ID, + ) + base.update(over) + return gate.record_dry_run(**base) + + def _assess(self, **over): + kwargs = dict( + pr_number=PR, repository_slug=SLUG, resolved_task=CANONICAL_TASK, + active_role_kind="reconciler", assessment=self.assessment, + evidence=self._evidence(), + ) + kwargs.update(over) + return gate.assess_apply_authorization(**kwargs) + + def test_reconciler_with_matching_evidence_is_authorized(self): + result = self._assess() + self.assertTrue(result["allowed"], result["reasons"]) + self.assertTrue(result["evidence_matched"]) + + def test_apply_without_exact_task_resolution_fails(self): + """8. Apply without exact task resolution fails preflight.""" + result = self._assess(resolved_task=None) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task") + + def test_resolving_another_task_does_not_authorize_cleanup(self): + """9. A sibling reconciler task is not a substitute.""" + for other in ("delete_branch", "reconcile_already_landed_pr", + "cleanup_merged_pr_branch", "comment_pr"): + with self.subTest(task=other): + result = self._assess(resolved_task=other) + self.assertFalse(result["allowed"]) + self.assertEqual( + result["blocker_kind"], "unresolved_cleanup_task") + + def test_non_reconciler_roles_fail_closed(self): + """10. Author/reviewer/merger cannot apply despite pr.comment.""" + for role in ("author", "reviewer", "merger", None, ""): + with self.subTest(role=role): + result = self._assess(active_role_kind=role) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "wrong_role") + + def test_missing_repository_identity_fails_closed(self): + result = self._assess(repository_slug=None) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "repository_binding") + + def test_missing_dry_run_evidence_fails_closed(self): + result = self._assess(evidence=None) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence") + + def test_dry_run_that_disallowed_cleanup_fails_closed(self): + result = self._assess(evidence=self._evidence(cleanup_allowed=False)) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "dry_run_not_allowed") + + def test_dry_run_for_another_pr_or_repo_fails_closed(self): + """11. Wrong PR / repository fails closed.""" + for over in ({"pr_number": PR + 1}, {"repository_slug": "Other/Repo"}): + with self.subTest(**over): + result = self._assess(evidence=self._evidence(**over)) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "dry_run_mismatch") + + def test_superseded_lease_fails_closed(self): + """12. Head/session/marker drift since the dry run fails closed.""" + for over in ({"session_id": "other-session"}, + {"candidate_head": HEAD_B}, + {"lease_comment_id": 7777}): + with self.subTest(**over): + result = self._assess(evidence=self._evidence(**over)) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "superseded_lease") + + def test_expectation_mismatch_fails_closed(self): + """11. Wrong session / head / marker expectations fail closed.""" + for over in ({"expected_session_id": "nope"}, + {"expected_candidate_head": HEAD_B}, + {"expected_lease_comment_id": 7777}): + with self.subTest(**over): + result = self._assess(**over) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "lease_mismatch") + + def test_matching_expectations_are_authorized(self): + result = self._assess( + expected_session_id=SESSION, + expected_candidate_head=HEAD_A, + expected_lease_comment_id=LEASE_COMMENT_ID, + ) + self.assertTrue(result["allowed"], result["reasons"]) + + def test_non_moot_lease_fails_closed(self): + """12. A live lease on an open PR is never cleanable.""" + open_pr = _assessment( + [_lease_comment()], pr_merged=False, pr_state="open") + result = self._assess(assessment=open_pr) + self.assertFalse(result["allowed"]) + self.assertIn( + result["blocker_kind"], ("lease_not_moot", "malformed_lease")) + + def test_malformed_lease_fails_closed(self): + malformed = dict(self.assessment) + malformed["active_lease"] = { + "session_id": "", "candidate_head": None, "comment_id": None} + result = self._assess(assessment=malformed) + self.assertFalse(result["allowed"]) + self.assertEqual(result["blocker_kind"], "malformed_lease") + + def test_ledger_is_append_only(self): + """14. Recording never rewrites or drops prior entries.""" + first = self._evidence() + second = self._evidence(candidate_head=HEAD_B, lease_comment_id=7777) + history = gate.dry_run_history() + self.assertEqual(len(history), 2) + self.assertEqual(history[0]["candidate_head"], first["candidate_head"]) + self.assertEqual(history[1]["candidate_head"], HEAD_B) + # Newest-wins for lookup, but the older entry survives in history. + latest = gate.latest_dry_run(pr_number=PR, repository_slug=SLUG) + self.assertEqual(latest["lease_comment_id"], second["lease_comment_id"]) + self.assertEqual(gate.dry_run_history()[0]["lease_comment_id"], + LEASE_COMMENT_ID) + + def test_history_view_cannot_mutate_the_ledger(self): + self._evidence() + snapshot = gate.dry_run_history() + snapshot[0]["pr_number"] = 999999 + self.assertEqual( + gate.dry_run_history()[0]["pr_number"], PR, + "dry_run_history must hand out copies, not live rows") + + +# --------------------------------------------------------------------------- # +# Tool-level behavior +# --------------------------------------------------------------------------- # +class _ToolCase(unittest.TestCase): + def setUp(self): + leases.clear_session_lease() + gate._reset_for_testing() + self.addCleanup(gate._reset_for_testing) + + def _run(self, env, *, apply, comments, pr_state="closed", pr_merged=True, + resolved_task=CANONICAL_TASK, slug=SLUG, **kwargs): + side, calls = _api_side_effect( + pr_state=pr_state, pr_merged=pr_merged, comments=comments) + with patch("mcp_server.api_request", side_effect=side), \ + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ + patch("mcp_server.verify_preflight_purity", return_value=None), \ + patch("mcp_server._bound_repository_slug", return_value=slug), \ + patch("mcp_server._repository_binding_block", return_value=None), \ + patch.object(mcp_server, "_preflight_resolved_task", + resolved_task), \ + patch.dict(os.environ, env, clear=True): + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=apply, remote="prgs", **kwargs) + return result, calls + + +class TestDryRunOpenToEveryRole(_ToolCase): + """7. Dry run performs no mutation and stays under the read capability.""" + + def test_dry_run_reports_moot_and_mutates_nothing_for_every_role(self): + for name, env in (("reconciler", RECONCILER_ENV), ("author", AUTHOR_ENV), + ("reviewer", REVIEWER_ENV), ("merger", MERGER_ENV)): + with self.subTest(role=name): + gate._reset_for_testing() + result, calls = self._run( + env, apply=False, comments=[_lease_comment()], + resolved_task=None) + self.assertTrue(result["success"]) + self.assertTrue(result["lease_moot"]) + self.assertTrue(result["cleanup_allowed"]) + self.assertFalse(result["cleanup_performed"]) + self.assertEqual(result["mode"], "read_only") + self.assertEqual(calls["post"], [], "dry run must not mutate") + + def test_dry_run_records_evidence(self): + result, _ = self._run( + RECONCILER_ENV, apply=False, comments=[_lease_comment()]) + evidence = result["dry_run_evidence"] + self.assertEqual(evidence["pr_number"], PR) + self.assertEqual(evidence["repository_slug"], SLUG) + self.assertEqual(evidence["session_id"], SESSION) + self.assertEqual(evidence["candidate_head"], HEAD_A) + self.assertEqual(evidence["lease_comment_id"], LEASE_COMMENT_ID) + self.assertTrue(evidence["lease_moot"]) + self.assertTrue(evidence["cleanup_allowed"]) + + +class TestApplyRequiresReconciler(_ToolCase): + def test_reconciler_apply_succeeds_after_matching_dry_run(self): + """7 (apply). Allowed dry run then apply posts exactly one marker.""" + comments = [_lease_comment()] + dry, dry_calls = self._run( + RECONCILER_ENV, apply=False, comments=comments) + self.assertTrue(dry["cleanup_allowed"]) + self.assertEqual(dry_calls["post"], []) + + result, calls = self._run( + RECONCILER_ENV, apply=True, comments=comments) + self.assertTrue(result["success"], result.get("reasons")) + self.assertTrue(result["cleanup_performed"]) + self.assertEqual(result["released_comment_id"], 9999) + self.assertEqual(len(calls["post"]), 1) + body = calls["post"][0]["payload"]["body"] + self.assertIn("phase: released", body) + self.assertIn("post-merge-moot", body) + + def test_apply_without_dry_run_fails_closed(self): + """6. Apply must be preceded by a matching dry run.""" + result, calls = self._run( + RECONCILER_ENV, apply=True, comments=[_lease_comment()]) + self.assertFalse(result["success"]) + self.assertFalse(result["cleanup_performed"]) + self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence") + self.assertEqual(calls["post"], []) + + def test_apply_without_exact_task_resolution_fails_closed(self): + """8. No resolved cleanup task -> no mutation.""" + comments = [_lease_comment()] + self._run(RECONCILER_ENV, apply=False, comments=comments) + result, calls = self._run( + RECONCILER_ENV, apply=True, comments=comments, resolved_task=None) + self.assertFalse(result["success"]) + self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task") + self.assertEqual(calls["post"], []) + + def test_resolving_a_different_task_does_not_authorize_apply(self): + """9. Another resolved task is not a substitute.""" + comments = [_lease_comment()] + self._run(RECONCILER_ENV, apply=False, comments=comments) + result, calls = self._run( + RECONCILER_ENV, apply=True, comments=comments, + resolved_task="delete_branch") + self.assertFalse(result["success"]) + self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task") + self.assertEqual(calls["post"], []) + + def test_author_reviewer_merger_cannot_apply(self): + """10. Permission-only roles are refused at the role gate.""" + for name, env in (("author", AUTHOR_ENV), ("reviewer", REVIEWER_ENV), + ("merger", MERGER_ENV)): + with self.subTest(role=name): + gate._reset_for_testing() + comments = [_lease_comment()] + self._run(env, apply=False, comments=comments) + result, calls = self._run(env, apply=True, comments=comments) + self.assertFalse(result["success"]) + self.assertFalse(result["cleanup_performed"]) + self.assertEqual(result["blocker_kind"], "wrong_role") + self.assertEqual( + calls["post"], [], + f"{name} must not post a terminal lease marker") + + +class TestApplyFailsClosedOnLeaseState(_ToolCase): + def test_open_pr_lease_is_never_force_cleaned(self): + """12. Non-moot: an active lease on an open PR stays untouched.""" + comments = [_lease_comment()] + result, calls = self._run( + RECONCILER_ENV, apply=True, comments=comments, + pr_state="open", pr_merged=False) + self.assertFalse(result["cleanup_performed"]) + self.assertFalse(result["pr_merged_or_closed"]) + self.assertEqual(calls["post"], [], "never force-clean an open PR lease") + self.assertTrue(any( + "still open" in r for r in result.get("cleanup_skipped_reason", []))) + + def test_superseded_lease_between_dry_run_and_apply_fails_closed(self): + """12. The lease moved on after the dry run -> refuse.""" + self._run(RECONCILER_ENV, apply=False, comments=[_lease_comment()]) + moved = [_lease_comment(session_id="fresh-session", + candidate_head=HEAD_B, comment_id=7777)] + result, calls = self._run(RECONCILER_ENV, apply=True, comments=moved) + self.assertFalse(result["success"]) + self.assertEqual(result["blocker_kind"], "superseded_lease") + self.assertEqual(calls["post"], []) + + def test_expectation_mismatch_fails_closed(self): + """11. Wrong session / head / marker expectations refuse the apply.""" + comments = [_lease_comment()] + for kwargs in ({"expected_session_id": "wrong-session"}, + {"expected_candidate_head": HEAD_B}, + {"expected_lease_comment_id": 7777}): + with self.subTest(**kwargs): + gate._reset_for_testing() + self._run(RECONCILER_ENV, apply=False, comments=comments) + result, calls = self._run( + RECONCILER_ENV, apply=True, comments=comments, **kwargs) + self.assertFalse(result["success"]) + self.assertEqual(result["blocker_kind"], "lease_mismatch") + self.assertEqual(calls["post"], []) + + def test_already_terminal_cleanup_is_idempotent(self): + """13. A released lease reports nothing to clean and posts nothing.""" + first = _assessment([_lease_comment()]) + released = {"id": 7000, "body": first["release_body"], + "user": {"login": "sysadmin"}} + comments = [_lease_comment(), released] + self._run(RECONCILER_ENV, apply=False, comments=comments) + result, calls = self._run(RECONCILER_ENV, apply=True, comments=comments) + self.assertFalse(result["cleanup_performed"]) + self.assertFalse(result["lease_moot"]) + self.assertEqual(calls["post"], [], "no second terminal marker") + self.assertTrue(any( + "already released/terminal" in r + for r in result.get("cleanup_skipped_reason", []))) + + def test_apply_is_append_only_never_deletes(self): + """14. The only write is a POST; nothing is edited or deleted.""" + comments = [_lease_comment()] + self._run(RECONCILER_ENV, apply=False, comments=comments) + side, _calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=comments) + seen = [] + + def _recording(method, url, auth=None, payload=None, *a, **k): + seen.append((method or "").upper()) + return side(method, url, auth, payload, *a, **k) + + with patch("mcp_server.api_request", side_effect=_recording), \ + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ + patch("mcp_server.verify_preflight_purity", return_value=None), \ + patch("mcp_server._bound_repository_slug", return_value=SLUG), \ + patch("mcp_server._repository_binding_block", return_value=None), \ + patch.object(mcp_server, "_preflight_resolved_task", + CANONICAL_TASK), \ + patch.dict(os.environ, RECONCILER_ENV, clear=True): + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=True, remote="prgs") + self.assertTrue(result["cleanup_performed"]) + self.assertNotIn("DELETE", seen) + self.assertNotIn("PATCH", seen) + self.assertNotIn("PUT", seen) + self.assertEqual(seen.count("POST"), 1) + + +class TestRepositoryBinding(_ToolCase): + """11. Foreign-repository targets fail closed before any mutation.""" + + def test_explicit_foreign_repository_is_rejected(self): + comments = [_lease_comment()] + side, calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=comments) + with patch("mcp_server.api_request", side_effect=side), \ + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ + patch("mcp_server.verify_preflight_purity", return_value=None), \ + patch("mcp_server._canonical_repository_slug", + return_value=(None, [])), \ + patch("mcp_server._workspace_repository_slug", + return_value=SLUG), \ + patch.object(mcp_server, "_preflight_resolved_task", + CANONICAL_TASK), \ + patch.dict(os.environ, RECONCILER_ENV, clear=True): + gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=False, remote="prgs") + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=True, remote="prgs", + org="Some-Other-Org", repo="Some-Other-Repo") + self.assertFalse(result["success"]) + self.assertEqual(result["blocker_kind"], "repository_binding") + self.assertEqual(calls["post"], []) + + def test_unresolvable_canonical_root_fails_closed(self): + comments = [_lease_comment()] + side, calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=comments) + with patch("mcp_server.api_request", side_effect=side), \ + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ + patch("mcp_server.verify_preflight_purity", return_value=None), \ + patch("mcp_server._canonical_repository_slug", + return_value=(None, ["canonical root unresolvable"])), \ + patch.object(mcp_server, "_preflight_resolved_task", + CANONICAL_TASK), \ + patch.dict(os.environ, RECONCILER_ENV, clear=True): + gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=False, remote="prgs") + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=True, remote="prgs") + self.assertFalse(result["success"]) + self.assertEqual(result["blocker_kind"], "repository_binding") + self.assertEqual(calls["post"], []) + + +class TestPreflightBinding(_ToolCase): + """4. The apply path binds the shared preflight to the exact task.""" + + def test_apply_forwards_task_and_target_to_preflight(self): + comments = [_lease_comment()] + self._run(RECONCILER_ENV, apply=False, comments=comments) + side, _calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=comments) + seen = {} + + def _purity(remote=None, worktree_path=None, task=None, **kw): + seen.update({"remote": remote, "worktree_path": worktree_path, + "task": task, **kw}) + return None + + with patch("mcp_server.api_request", side_effect=side), \ + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ + patch("mcp_server.verify_preflight_purity", _purity), \ + patch("mcp_server._bound_repository_slug", return_value=SLUG), \ + patch("mcp_server._repository_binding_block", return_value=None), \ + patch.object(mcp_server, "_preflight_resolved_task", + CANONICAL_TASK), \ + patch.dict(os.environ, RECONCILER_ENV, clear=True): + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=True, remote="prgs", + worktree_path="/tmp/branches/reconciler-745", + org="Scaled-Tech-Consulting", repo="Gitea-Tools") + self.assertTrue(result["cleanup_performed"]) + self.assertEqual(seen["task"], CANONICAL_TASK) + self.assertEqual(seen["worktree_path"], "/tmp/branches/reconciler-745") + self.assertEqual(seen["org"], "Scaled-Tech-Consulting") + self.assertEqual(seen["repo"], "Gitea-Tools") + + def test_dry_run_does_not_require_preflight(self): + def _boom(*a, **k): + raise AssertionError("dry run must not run mutation preflight") + + side, calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=[_lease_comment()]) + with patch("mcp_server.api_request", side_effect=side), \ + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ + patch("mcp_server.verify_preflight_purity", _boom), \ + patch("mcp_server._bound_repository_slug", return_value=SLUG), \ + patch.dict(os.environ, AUTHOR_ENV, clear=True): + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=False, remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(calls["post"], []) + + +class TestNoProductionLeaseTouched(unittest.TestCase): + """16. No real production lease or PR is referenced by these tests.""" + + PRODUCTION_PR = 744 + PRODUCTION_SESSION = "33673-1d54887a0415" + PRODUCTION_MARKER = 12452 + + def test_fixtures_are_synthetic(self): + self.assertNotEqual(PR, self.PRODUCTION_PR) + self.assertNotEqual(SESSION, self.PRODUCTION_SESSION) + self.assertNotEqual(LEASE_COMMENT_ID, self.PRODUCTION_MARKER) + + def test_module_source_never_names_the_production_lease(self): + source = _Path(__file__).read_text() + for token in (self.PRODUCTION_SESSION, str(self.PRODUCTION_MARKER)): + self.assertEqual( + source.count(token), 1, + f"{token!r} must appear only in this guard's own constants", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_post_merge_moot_lease.py b/tests/test_post_merge_moot_lease.py index 9f9095b..29ea74e 100644 --- a/tests/test_post_merge_moot_lease.py +++ b/tests/test_post_merge_moot_lease.py @@ -19,6 +19,8 @@ from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) +import mcp_server # noqa: E402 +import post_merge_moot_lease_gate as moot_gate # noqa: E402 import reviewer_pr_lease as leases # noqa: E402 from mcp_server import ( # noqa: E402 gitea_acquire_reviewer_pr_lease, @@ -30,6 +32,13 @@ MERGER_ENV = { "GITEA_PROFILE_NAME": "prgs-merger", "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment", } +# #745: applying the terminal marker is reconciler-owned. +RECONCILER_ENV = { + "GITEA_PROFILE_NAME": "prgs-reconciler", + "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment,gitea.pr.close", +} +CLEANUP_TASK = moot_gate.CLEANUP_TASK +REPO_SLUG = "Scaled-Tech-Consulting/Gitea-Tools" PR = 487 ISSUE = 485 SESSION = "97274-676d20a825c4" @@ -204,6 +213,8 @@ class TestAcquireToolRefusesMergedPR(unittest.TestCase): class TestCleanupTool(unittest.TestCase): def setUp(self): leases.clear_session_lease() + moot_gate._reset_for_testing() + self.addCleanup(moot_gate._reset_for_testing) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.api_request") @@ -223,23 +234,53 @@ class TestCleanupTool(unittest.TestCase): self.assertEqual(calls["post"], []) @patch("mcp_server.verify_preflight_purity", return_value=None) + @patch("mcp_server._repository_binding_block", return_value=None) + @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.api_request") def test_apply_posts_released_marker_on_merged_pr( - self, mock_api, _auth, _purity): + self, mock_api, _auth, _slug, _binding, _purity): + """#745: apply is reconciler-only and needs a matching dry run first.""" side, calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=[_lease_comment()]) mock_api.side_effect = side - with patch.dict(os.environ, MERGER_ENV, clear=True): + with patch.object(mcp_server, "_preflight_resolved_task", + CLEANUP_TASK), \ + patch.dict(os.environ, RECONCILER_ENV, clear=True): + gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=False, remote="prgs") result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=True, remote="prgs") - self.assertTrue(result["success"]) + self.assertTrue(result["success"], result.get("reasons")) self.assertTrue(result["cleanup_performed"]) self.assertEqual(result["released_comment_id"], 9999) self.assertEqual(len(calls["post"]), 1) self.assertIn("phase: released", calls["post"][0]["payload"]["body"]) self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"]) + @patch("mcp_server.verify_preflight_purity", return_value=None) + @patch("mcp_server._repository_binding_block", return_value=None) + @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.api_request") + def test_merger_can_no_longer_apply( + self, mock_api, _auth, _slug, _binding, _purity): + """#745: holding gitea.pr.comment is no longer sufficient to apply.""" + side, calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=[_lease_comment()]) + mock_api.side_effect = side + with patch.object(mcp_server, "_preflight_resolved_task", + CLEANUP_TASK), \ + patch.dict(os.environ, MERGER_ENV, clear=True): + gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=False, remote="prgs") + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=True, remote="prgs") + self.assertFalse(result["success"]) + self.assertFalse(result["cleanup_performed"]) + self.assertEqual(result["blocker_kind"], "wrong_role") + self.assertEqual(calls["post"], []) + @patch("mcp_server.verify_preflight_purity", return_value=None) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.api_request")