"""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()
