Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 75 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix(mcp): bind post-merge moot-lease cleanup to the reconciler capability (Closes #745)' (#746) from fix/issue-745-reconciler-moot-lease-gate into master

Browse Source
This commit was merged in pull request #746.
This commit is contained in:
sysadmin
2026-07-18 22:04:42 -05:00
parent 447595b72b a517655aad
commit bde5c5fb20
7 changed files with 1187 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/gitea-execution-profiles.md
+38
View File
@@ -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
gitea_mcp_server.py
+143 -11
View File
@@ -1880,6 +1880,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
@@ -9019,13 +9020,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
@@ -9058,7 +9060,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",
}
@@ -9077,9 +9079,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",
@@ -9089,17 +9091,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,
@@ -12528,6 +12564,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,
@@ -12543,14 +12583,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
@@ -12610,14 +12673,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
@@ -12625,7 +12736,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(
post_merge_moot_lease_gate.py
+279
View File
@@ -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,
}
role_session_router.py
+8
View File
@@ -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",
task_capability_map.py
+17
View File
@@ -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",
tests/test_issue_745_moot_lease_reconciler_gate.py
+658
View File
@@ -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()
tests/test_post_merge_moot_lease.py
+44 -3
View File
@@ -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")