import sys
import unittest
from unittest.mock import patch

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))

import branch_cleanup_guard as guard  # noqa: E402
import mcp_server  # noqa: E402
import task_capability_map  # noqa: E402
from final_report_validator import assess_final_report_validator  # noqa: E402
from mcp_server import gitea_cleanup_merged_pr_branch, gitea_delete_branch  # noqa: E402

FAKE_AUTH = "token fake"

# Reconciler-shaped profile that holds branch.delete (recommended) plus
# required pr.close/read so _role_kind classifies as reconciler.
RECONCILER_WITH_DELETE = {
    "profile_name": "prgs-reconciler",
    "role": "reconciler",
    "allowed_operations": [
        "gitea.read",
        "gitea.pr.close",
        "gitea.pr.comment",
        "gitea.issue.comment",
        "gitea.issue.close",
        "gitea.branch.delete",
    ],
    "forbidden_operations": [
        "gitea.pr.approve",
        "gitea.pr.merge",
        "gitea.pr.review",
        "gitea.pr.create",
        "gitea.branch.push",
        "gitea.repo.commit",
    ],
}


class TestRawBranchDeleteGuard(unittest.TestCase):
    def test_detects_local_and_remote_raw_git_delete_commands(self):
        text = """
        Cleanup mutations:
        - git branch -d feat/issue-485-lease-comments-non-list-guard
        - git push prgs --delete feat/issue-485-lease-comments-non-list-guard
        """
        commands = guard.raw_branch_delete_commands(text)
        self.assertEqual(len(commands), 2)
        self.assertIn("git branch -d", commands[0])
        self.assertIn("git push prgs --delete", commands[1])

    def test_final_report_blocks_raw_delete_as_cleanup_proof(self):
        report = """
        ## Controller Handoff
        - Task: review_pr
        - Cleanup mutations: git push prgs --delete feat/branch
        """
        result = assess_final_report_validator(report, "review_pr")
        self.assertTrue(result["blocked"])
        self.assertTrue(
            any("shared.raw_branch_delete_bypass" in r for r in result["reasons"])
        )

    def test_cleanup_task_requires_branch_delete_capability(self):
        self.assertEqual(
            task_capability_map.required_permission("cleanup_merged_pr_branch"),
            "gitea.branch.delete",
        )
        self.assertEqual(
            task_capability_map.required_role("cleanup_merged_pr_branch"),
            "reconciler",
        )


class TestMergedPrBranchCleanupTool(unittest.TestCase):
    def setUp(self):
        self._remotes = patch.dict(
            mcp_server.REMOTES,
            {
                "prgs": {
                    "host": "gitea.example.com",
                    "org": "Example-Org",
                    "repo": "Example-Repo",
                }
            },
        )
        self._remotes.start()
        patch("gitea_audit.audit_enabled", return_value=False).start()
        self.mock_api = patch("mcp_server.api_request").start()
        self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start()
        patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
            return_value=True,
        ).start()
        # Default: no active ownership records (tests that need ownership patch this).
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()

    def tearDown(self):
        patch.stopall()

    def test_reviewer_without_delete_authority_fails_closed(self):
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "reviewer",
                "allowed_operations": ["gitea.read", "gitea.pr.review"],
                "forbidden_operations": ["gitea.branch.delete"],
            },
        ).start()
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
            branch="feat/branch",
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res["performed"])
        self.assertEqual(res["required_permission"], "gitea.branch.delete")
        self.mock_api.assert_not_called()

    def test_author_and_merger_without_delete_authority_fail_closed(self):
        role_profiles = {
            "author": [
                "gitea.read",
                "gitea.branch.create",
                "gitea.branch.push",
                "gitea.repo.commit",
                "gitea.pr.create",
            ],
            "merger": ["gitea.read", "gitea.pr.merge"],
        }
        for name, allowed in role_profiles.items():
            with self.subTest(role=name):
                profile_patch = patch(
                    "mcp_server.get_profile",
                    return_value={
                        "profile_name": name,
                        "allowed_operations": allowed,
                        "forbidden_operations": [],
                    },
                )
                profile_patch.start()
                try:
                    res = gitea_cleanup_merged_pr_branch(
                        pr_number=487,
                        confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
                        branch="feat/branch",
                        remote="prgs",
                        worktree_path="/tmp/repo/branches/cleanup",
                    )
                finally:
                    profile_patch.stop()
                self.assertFalse(res["performed"])
                self.assertEqual(
                    res["required_permission"], "gitea.branch.delete"
                )
                self.mock_api.assert_not_called()

    def test_root_checkout_cleanup_fails_closed(self):
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
            branch="feat/branch",
            remote="prgs",
            worktree_path="/tmp/repo/Gitea-Tools",
        )
        self.assertFalse(res["performed"])
        self.assertIn("root checkout", " ".join(res["reasons"]))
        self.mock_api.assert_not_called()

    def test_authorized_cleanup_deletes_through_api_path(self):
        branch = "feat/issue-485-lease-comments-non-list-guard"
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()

        def _api(method, url, *args, **kwargs):
            if method == "GET" and "/pulls/" in url:
                return {
                    "number": 487,
                    "merged": True,
                    "merged_at": "2026-07-08T01:00:00Z",
                    "head": {"ref": branch, "sha": "a" * 40},
                    "base": {"ref": "master"},
                }
            if method == "GET" and "/branches/" in url:
                # First pre-delete probe: present. Post-delete: not found.
                get_branch_calls = [
                    c
                    for c in self.mock_api.call_args_list
                    if c.args and c.args[0] == "GET" and "/branches/" in c.args[1]
                ]
                if len(get_branch_calls) <= 1:
                    return {"name": branch}
                raise RuntimeError("HTTP 404: not found")
            if method == "GET" and url.rstrip("/").endswith("/Example-Repo"):
                # Repo reachability probe after branch 404 (R1 branch-scoped).
                return {"full_name": "Example-Org/Example-Repo"}
            if method == "DELETE":
                return {}
            raise AssertionError(f"unexpected {method} {url}")

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertTrue(res["success"])
        self.assertTrue(res["performed"])
        self.assertTrue(res["delete_acknowledged"])
        self.assertTrue(res["verified_absent"])
        self.assertTrue((res.get("readback") or {}).get("verified_absent"))
        delete_calls = [
            call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
        ]
        self.assertEqual(len(delete_calls), 1)
        self.assertIn("branches/feat%2Fissue-485", delete_calls[0].args[1])

    def test_confirmation_required_before_delete(self):
        branch = "feat/issue-485-lease-comments-non-list-guard"
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()
        self.mock_api.side_effect = [
            {
                "number": 487,
                "merged": True,
                "merged_at": "2026-07-08T01:00:00Z",
                "head": {"ref": branch, "sha": "a" * 40},
                "base": {"ref": "master"},
            },
            {},
        ]
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation="wrong",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res["performed"])
        self.assertIn("confirmation", " ".join(res["reasons"]))
        delete_calls = [
            call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
        ]
        self.assertFalse(delete_calls)

    def test_reconciler_with_branch_delete_can_raw_delete(self):
        """#729: delete_branch is re-homed from author to reconciler. The
        reconciler is the delete-capable role and performs raw deletion of an
        eligible (non-preservation, non-protected) branch — superseding the
        #687 redirect that previously blocked it."""
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", resolved_role="reconciler")
        self.mock_api.return_value = {}
        res = gitea_delete_branch(
            branch="fix/issue-683-workflow-guard-hardening",
            remote="prgs",
        )
        self.assertTrue(res.get("success"))
        delete_calls = [
            call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
        ]
        self.assertTrue(delete_calls)

    def test_reconciler_raw_delete_denies_preservation_branch(self):
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()
        res = gitea_delete_branch(
            branch="chore/issue-681-preserve-review-session-wip",
            remote="prgs",
        )
        self.assertFalse(res.get("performed", True))
        self.mock_api.assert_not_called()

    def test_reconciler_raw_delete_role_ok_but_preserve_blocked(self):
        """#729: reconciler role may use raw delete path when permitted;
        preservation branch still fails closed."""
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "prgs-reconciler",
                "role": "reconciler",
                "allowed_operations": [
                    "gitea.read",
                    "gitea.issue.comment",
                    "gitea.pr.comment",
                    "gitea.branch.delete",
                ],
                "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
            },
        ).start()
        res = gitea_delete_branch(
            branch="chore/issue-681-preserve-review-session-wip",
            remote="prgs",
        )
        self.assertFalse(res.get("performed", True))
        self.assertIn("preservation", " ".join(res.get("reasons") or []))
        self.mock_api.assert_not_called()

    def test_unmerged_branch_cleanup_rejected(self):
        branch = "feat/unmerged-work"
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()
        self.mock_api.side_effect = [
            {
                "number": 999,
                "merged": False,
                "merged_at": None,
                "head": {"ref": branch, "sha": "b" * 40},
                "base": {"ref": "master"},
            },
            {},
        ]
        res = gitea_cleanup_merged_pr_branch(
            pr_number=999,
            confirmation=f"CLEANUP MERGED PR 999 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res["performed"])
        self.assertTrue(
            any("not merged" in r for r in (res.get("reasons") or []))
        )
        delete_calls = [
            call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
        ]
        self.assertFalse(delete_calls)

    def test_preservation_branch_cleanup_rejected(self):
        branch = "chore/issue-681-preserve-review-session-wip"
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()
        self.mock_api.side_effect = [
            {
                "number": 681,
                "merged": True,
                "merged_at": "2026-07-08T01:00:00Z",
                "head": {"ref": branch, "sha": "c" * 40},
                "base": {"ref": "master"},
            },
            {},
        ]
        res = gitea_cleanup_merged_pr_branch(
            pr_number=681,
            confirmation=f"CLEANUP MERGED PR 681 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res["performed"])
        self.assertTrue(
            any("preservation" in r for r in (res.get("reasons") or []))
        )
        delete_calls = [
            call for call in self.mock_api.call_args_list if call.args[0] == "DELETE"
        ]
        self.assertFalse(delete_calls)

    def test_non_reconciler_with_delete_denied_cleanup(self):
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "prgs-author",
                "role": "author",
                "allowed_operations": [
                    "gitea.read",
                    "gitea.pr.create",
                    "gitea.branch.push",
                    "gitea.branch.delete",
                ],
                "forbidden_operations": [],
            },
        ).start()
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation="CLEANUP MERGED PR 487 BRANCH feat/branch",
            branch="feat/branch",
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res["performed"])
        self.assertEqual(res.get("required_role_kind"), "reconciler")
        self.mock_api.assert_not_called()

    def test_assess_guard_rejects_preservation_branch(self):
        assessment = guard.assess_merged_pr_branch_cleanup(
            pr_number=681,
            head_branch="chore/issue-681-preserve-review-session-wip",
            merged=True,
            remote_branch_exists=True,
            open_pr_heads=set(),
            head_on_target=True,
            delete_capability_allowed=True,
            confirmation=(
                "CLEANUP MERGED PR 681 BRANCH "
                "chore/issue-681-preserve-review-session-wip"
            ),
        )
        self.assertFalse(assessment["safe_to_delete"])
        self.assertTrue(
            any("preservation" in r for r in assessment["block_reasons"])
        )



class TestPostDeleteReadback(unittest.TestCase):
    def test_branch_scoped_not_found_is_verified_success(self):
        readback = guard.classify_branch_readback_http_status(
            404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
        )
        result = guard.assess_post_delete_readback(readback)
        self.assertTrue(result["ok"])
        self.assertTrue(result["verified_absent"])
        self.assertTrue(result["readback"]["verified_absent"])

    def test_generic_404_not_verified_absent(self):
        # R1: bare 404 must not verify absence
        for scope in (None, guard.NOT_FOUND_SCOPE_UNKNOWN,
                      guard.NOT_FOUND_SCOPE_REPOSITORY,
                      guard.NOT_FOUND_SCOPE_HOST):
            with self.subTest(scope=scope):
                readback = guard.classify_branch_readback_http_status(
                    404, not_found_scope=scope
                )
                self.assertFalse(readback["verified_absent"])
                result = guard.assess_post_delete_readback(readback)
                self.assertFalse(result["ok"])
                self.assertFalse(result["verified_absent"])

    def test_exception_substring_404_not_verified(self):
        # R1: substring/generic 404 without scope stays unverified
        result = guard.classify_branch_readback_exception(
            RuntimeError("HTTP 404: something not found")
        )
        self.assertFalse(result["verified_absent"])
        self.assertNotEqual(result.get("not_found_scope"), guard.NOT_FOUND_SCOPE_BRANCH)

    def test_exists_is_structured_failure(self):
        readback = guard.classify_branch_readback_http_status(200)
        result = guard.assess_post_delete_readback(readback)
        self.assertFalse(result["ok"])
        self.assertFalse(result["readback"]["verified_absent"])
        self.assertTrue(result["readback"]["branch_present"])
        self.assertIn("still present", " ".join(result["reasons"]))

    def test_auth_failure_preserved(self):
        readback = guard.classify_branch_readback_http_status(401)
        result = guard.assess_post_delete_readback(readback)
        self.assertFalse(result["ok"])
        self.assertEqual(result["readback"]["error_class"], "authentication")
        self.assertIn("authentication", " ".join(result["reasons"]))

    def test_authz_and_transport_failures(self):
        for code, err in ((403, "authorization"), (503, "transport")):
            with self.subTest(code=code):
                readback = guard.classify_branch_readback_http_status(code)
                result = guard.assess_post_delete_readback(readback)
                self.assertFalse(result["ok"])
                self.assertEqual(result["readback"]["error_class"], err)

    def test_exception_classifier_no_secret_leak(self):
        class FakeHTTPError(Exception):
            def __init__(self):
                super().__init__("HTTP 401: token=super-secret-value Authorization: Bearer xyz")
                self.code = 401

        result = guard.classify_branch_readback_exception(FakeHTTPError())
        blob = str(result)
        self.assertNotIn("super-secret", blob)
        self.assertNotIn("Bearer", blob)
        self.assertEqual(result["error_class"], "authentication")


class TestActiveBranchOwnership(unittest.TestCase):
    def _base(self, **overrides):
        rec = {
            "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
            "status": "active",
            "remote": "prgs",
            "host": "gitea.prgs.cc",
            "org": "Scaled-Tech-Consulting",
            "repo": "Gitea-Tools",
            "branch": "feat/target",
            "reclaim_allowed": False,
        }
        rec.update(overrides)
        return rec

    def test_active_author_blocks(self):
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            branch="feat/target",
            host="gitea.prgs.cc",
            records=[self._base()],
        )
        self.assertTrue(result["block"])
        self.assertIn(guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, result["blocking_categories"])

    def test_active_reviewer_lease_blocks(self):
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            branch="feat/target",
            host="gitea.prgs.cc",
            records=[
                self._base(
                    category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE,
                    status="active",
                )
            ],
        )
        self.assertTrue(result["block"])
        self.assertIn(guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, result["blocking_categories"])

    def test_merger_controller_reconciler_binding_blocks(self):
        for cat in (
            guard.OWNERSHIP_CATEGORY_MERGER_LEASE,
            guard.OWNERSHIP_CATEGORY_CONTROLLER_LEASE,
            guard.OWNERSHIP_CATEGORY_RECONCILER_LEASE,
            guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
        ):
            with self.subTest(category=cat):
                result = guard.assess_active_branch_ownership(
                    remote="prgs",
                    org="Scaled-Tech-Consulting",
                    repo="Gitea-Tools",
                    branch="feat/target",
                    host="gitea.prgs.cc",
                    records=[self._base(category=cat, status="active")],
                )
                self.assertTrue(result["block"])
                self.assertIn(cat, result["blocking_categories"])

    def test_released_and_expired_reclaimable_do_not_block(self):
        records = [
            self._base(status="released", reclaim_allowed=True),
            self._base(
                category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE,
                status="expired",
                reclaim_allowed=True,
            ),
        ]
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            branch="feat/target",
            host="gitea.prgs.cc",
            records=records,
        )
        self.assertFalse(result["block"])

    def test_sticky_stale_blocks_per_recovery_policy(self):
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            branch="feat/target",
            host="gitea.prgs.cc",
            records=[self._base(status="stale", reclaim_allowed=False)],
        )
        self.assertTrue(result["block"])
        self.assertTrue(
            any(
                token in " ".join(result["reasons"])
                for token in ("sticky", "reclaim not proven", "fail closed")
            )
        )

    def test_other_repo_or_branch_no_false_block(self):
        records = [
            self._base(repo="Other-Repo", status="active"),
            self._base(branch="feat/other", status="active"),
            self._base(remote="dadeschools", status="active"),
            self._base(host="gitea.other.host", status="active"),
        ]
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            branch="feat/target",
            host="gitea.prgs.cc",
            records=records,
        )
        self.assertFalse(result["block"])
        self.assertEqual(len(result["ignored_out_of_scope"]), 4)

    def test_normalized_host_matching(self):
        # Host identity is normalized (scheme/path stripped)
        records = [self._base(host="https://gitea.prgs.cc/")]
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            branch="feat/target",
            host="gitea.prgs.cc",
            records=records,
        )
        self.assertTrue(result["block"])

    def test_denial_has_no_secrets(self):
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            branch="feat/target",
            host="gitea.prgs.cc",
            records=[
                self._base(
                    status="active",
                    # Poison fields that must never be echoed as secrets
                    token="sekrit-token",
                    authorization="Bearer abc",
                )
            ],
        )
        blob = str(result)
        self.assertNotIn("sekrit", blob)
        self.assertNotIn("Bearer", blob)
        self.assertIn("author_lease", blob)


class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase):
    def setUp(self):
        self._remotes = patch.dict(
            mcp_server.REMOTES,
            {
                "prgs": {
                    "host": "gitea.example.com",
                    "org": "Example-Org",
                    "repo": "Example-Repo",
                }
            },
        )
        self._remotes.start()
        patch("gitea_audit.audit_enabled", return_value=False).start()
        self.mock_api = patch("mcp_server.api_request").start()
        patch("mcp_server.api_get_all", return_value=[]).start()
        patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
            return_value=True,
        ).start()
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()

    def tearDown(self):
        patch.stopall()

    def _pr_payload(self, branch, number=487):
        return {
            "number": number,
            "merged": True,
            "merged_at": "2026-07-08T01:00:00Z",
            "head": {"ref": branch, "sha": "a" * 40},
            "base": {"ref": "master"},
        }

    def test_delete_success_but_branch_remains(self):
        branch = "feat/still-there"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr_payload(branch)
            if method == "GET" and "/branches/" in url:
                return {"name": branch}  # present before and after
            if method == "DELETE":
                return {}
            raise AssertionError(method)

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertTrue(res.get("performed"))
        self.assertFalse(res.get("success"))
        self.assertTrue(res.get("delete_acknowledged"))
        self.assertFalse(res.get("verified_absent"))
        self.assertIn("still present", " ".join(res.get("reasons") or []))

    def test_readback_authentication_failure(self):
        branch = "feat/auth-fail-readback"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()
        state = {"branch_gets": 0}

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr_payload(branch)
            if method == "GET" and "/branches/" in url:
                state["branch_gets"] += 1
                if state["branch_gets"] == 1:
                    return {"name": branch}
                raise RuntimeError("HTTP 401: unauthorized")
            if method == "DELETE":
                return {}
            raise AssertionError(method)

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertTrue(res.get("performed"))
        self.assertFalse(res.get("success"))
        self.assertEqual((res.get("readback") or {}).get("error_class"), "authentication")

    def test_readback_transport_failure(self):
        branch = "feat/transport-fail-readback"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()
        state = {"branch_gets": 0}

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr_payload(branch)
            if method == "GET" and "/branches/" in url:
                state["branch_gets"] += 1
                if state["branch_gets"] == 1:
                    return {"name": branch}
                raise RuntimeError("HTTP 503: temporarily unavailable")
            if method == "DELETE":
                return {}
            raise AssertionError(method)

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res.get("success"))
        self.assertEqual((res.get("readback") or {}).get("error_class"), "transport")

    def test_active_author_ownership_blocks_before_delete(self):
        branch = "feat/owned"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={
                "records": [
                    {
                        "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
                        "status": "active",
                        "remote": "prgs",
                        "host": "gitea.example.com",
                        "org": "Example-Org",
                        "repo": "Example-Repo",
                        "branch": branch,
                        "reclaim_allowed": False,
                    }
                ],
                "inventory_error": False,
            },
        ).start()

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr_payload(branch)
            if method == "GET" and "/branches/" in url:
                return {"name": branch}
            raise AssertionError(f"unexpected mutation {method}")

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res.get("performed"))
        self.assertFalse(res.get("success"))
        self.assertEqual(res.get("blocker_kind"), "active_branch_ownership")
        self.assertIn(
            guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
            (res.get("ownership") or {}).get("blocking_categories") or [],
        )
        delete_calls = [
            c for c in self.mock_api.call_args_list if c.args and c.args[0] == "DELETE"
        ]
        self.assertFalse(delete_calls)

    def test_open_pr_guard_still_blocks(self):
        branch = "feat/open-pr-head"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()
        patch(
            "mcp_server.api_get_all",
            return_value=[{"head": {"ref": branch}, "number": 999}],
        ).start()

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr_payload(branch, number=487)
            if method == "GET" and "/branches/" in url:
                return {"name": branch}
            raise AssertionError(method)

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res.get("performed"))
        self.assertTrue(any("open PR" in r for r in (res.get("reasons") or [])))

    def test_non_ancestor_guard_still_blocks(self):
        branch = "feat/not-ancestor"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
            return_value=False,
        ).start()

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr_payload(branch)
            if method == "GET" and "/branches/" in url:
                return {"name": branch}
            raise AssertionError(method)

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res.get("performed"))
        self.assertTrue(any("ancestor" in r for r in (res.get("reasons") or [])))

    def test_protected_default_branch_guard(self):
        branch = "master"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr_payload(branch)
            if method == "GET" and "/branches/" in url:
                return {"name": branch}
            raise AssertionError(method)

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res.get("performed"))
        self.assertTrue(any("protected" in r for r in (res.get("reasons") or [])))




class TestSecondRemediationR1R2(unittest.TestCase):
    """R1/R2 second-remediation: branch-scoped 404 and top-level fields."""

    def test_cleanup_envelope_always_has_top_level_fields(self):
        env = guard.cleanup_result_envelope(
            success=False,
            performed=False,
            delete_acknowledged=False,
            verified_absent=False,
            reasons=["x"],
        )
        for key in ("success", "performed", "delete_acknowledged", "verified_absent"):
            self.assertIn(key, env)
            self.assertIsInstance(env[key], bool)

    def test_repo_scoped_404_never_verified(self):
        rb = guard.classify_branch_readback_http_status(
            404, not_found_scope=guard.NOT_FOUND_SCOPE_REPOSITORY
        )
        self.assertFalse(rb["verified_absent"])
        assessed = guard.assess_post_delete_readback(rb)
        self.assertFalse(assessed["ok"])
        self.assertFalse(assessed["verified_absent"])

    def test_wrong_host_404_never_verified(self):
        rb = guard.classify_branch_readback_http_status(
            404, not_found_scope=guard.NOT_FOUND_SCOPE_HOST
        )
        self.assertFalse(rb["verified_absent"])


class TestSecondRemediationOwnership(unittest.TestCase):
    """O1/O2/O3 ownership second-remediation."""

    def test_inventory_error_category_blocks(self):
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Org",
            repo="Repo",
            branch="feat/x",
            host="gitea.example.com",
            records=[
                {
                    "category": guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR,
                    "status": "unknown",
                    "remote": "prgs",
                    "host": "gitea.example.com",
                    "org": "Org",
                    "repo": "Repo",
                    "branch": "feat/x",
                    "reclaim_allowed": False,
                }
            ],
        )
        self.assertTrue(result["block"])
        self.assertIn(
            guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR,
            result["blocking_categories"],
        )

    def test_expired_without_explicit_reclaim_blocks(self):
        # O2: expired must not auto-allow
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Org",
            repo="Repo",
            branch="feat/x",
            host="gitea.example.com",
            records=[
                {
                    "category": guard.OWNERSHIP_CATEGORY_MERGER_LEASE,
                    "status": "expired",
                    "remote": "prgs",
                    "host": "gitea.example.com",
                    "org": "Org",
                    "repo": "Repo",
                    "branch": "feat/x",
                    # reclaim_allowed omitted / False
                    "reclaim_allowed": False,
                }
            ],
        )
        self.assertTrue(result["block"])

    def test_expired_with_explicit_reclaim_allowed_does_not_block(self):
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Org",
            repo="Repo",
            branch="feat/x",
            host="gitea.example.com",
            records=[
                {
                    "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
                    "status": "expired",
                    "remote": "prgs",
                    "host": "gitea.example.com",
                    "org": "Org",
                    "repo": "Repo",
                    "branch": "feat/x",
                    "reclaim_allowed": True,
                }
            ],
        )
        self.assertFalse(result["block"])

    def test_active_reviewer_comment_lease_blocks(self):
        result = guard.assess_active_branch_ownership(
            remote="prgs",
            org="Org",
            repo="Repo",
            branch="feat/x",
            host="gitea.example.com",
            records=[
                {
                    "category": guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE,
                    "status": "active",
                    "remote": "prgs",
                    "host": "gitea.example.com",
                    "org": "Org",
                    "repo": "Repo",
                    "branch": "feat/x",
                    "reclaim_allowed": False,
                    "role": "reviewer",
                }
            ],
        )
        self.assertTrue(result["block"])
        self.assertIn(
            guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE,
            result["blocking_categories"],
        )


class TestSecondRemediationIntegration(unittest.TestCase):
    def setUp(self):
        self._remotes = patch.dict(
            mcp_server.REMOTES,
            {
                "prgs": {
                    "host": "gitea.example.com",
                    "org": "Example-Org",
                    "repo": "Example-Repo",
                }
            },
        )
        self._remotes.start()
        patch("gitea_audit.audit_enabled", return_value=False).start()
        self.mock_api = patch("mcp_server.api_request").start()
        patch("mcp_server.api_get_all", return_value=[]).start()
        patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
            return_value=True,
        ).start()
        patch(
            "mcp_server.get_profile",
            return_value=dict(RECONCILER_WITH_DELETE),
        ).start()

    def tearDown(self):
        patch.stopall()

    def _pr(self, branch, number=487):
        return {
            "number": number,
            "merged": True,
            "merged_at": "2026-07-08T01:00:00Z",
            "head": {"ref": branch, "sha": "a" * 40},
            "base": {"ref": "master"},
        }

    def test_r1_repo_404_after_delete_not_verified(self):
        """After DELETE, branch 404 + repo 404 must not verify absence."""
        branch = "feat/r1-repo-404"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": False},
        ).start()
        state = {"branch_gets": 0}

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr(branch)
            if method == "GET" and "/branches/" in url:
                state["branch_gets"] += 1
                if state["branch_gets"] == 1:
                    return {"name": branch}
                raise RuntimeError("HTTP 404: not found")
            if method == "GET" and url.rstrip("/").endswith("/Example-Repo"):
                raise RuntimeError("HTTP 404: repository not found")
            if method == "DELETE":
                return {}
            raise AssertionError(method + " " + url)

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertTrue(res["performed"])
        self.assertTrue(res["delete_acknowledged"])
        self.assertFalse(res["success"])
        self.assertFalse(res["verified_absent"])

    def test_o1_inventory_error_blocks_before_delete(self):
        branch = "feat/o1-inv"
        patch(
            "mcp_server._collect_branch_ownership_records",
            return_value={"records": [], "inventory_error": True},
        ).start()

        def _api(method, url, *a, **k):
            if method == "GET" and "/pulls/" in url:
                return self._pr(branch)
            if method == "GET" and "/branches/" in url:
                return {"name": branch}
            raise AssertionError(f"no mutation expected {method}")

        self.mock_api.side_effect = _api
        res = gitea_cleanup_merged_pr_branch(
            pr_number=487,
            confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}",
            branch=branch,
            remote="prgs",
            worktree_path="/tmp/repo/branches/cleanup",
        )
        self.assertFalse(res["performed"])
        self.assertFalse(res["delete_acknowledged"])
        self.assertFalse(res["verified_absent"])
        self.assertEqual(res.get("blocker_kind"), "active_branch_ownership")

    def test_o3_comment_reviewer_lease_in_collector(self):
        """Collector includes active comment-backed reviewer leases."""
        active_lease = {
            "pr_number": 10,
            "phase": "claimed",
            "session_id": "s1",
            "expires_at": "2099-01-02T00:00:00Z",
        }
        with patch(
            "mcp_server.api_get_all", return_value=[{"id": 1, "body": "x"}]
        ), patch(
            "mcp_server.reviewer_pr_lease.find_active_reviewer_lease",
            return_value=active_lease,
        ), patch(
            "mcp_server.issue_lock_store.iter_lock_files", return_value=[]
        ), patch(
            "mcp_server.worktree_cleanup_audit.list_worktrees", return_value=[]
        ), patch.object(
            mcp_server.control_plane_db,
            "ControlPlaneDB",
            side_effect=RuntimeError("no cp"),
        ):
            bundle = mcp_server._collect_branch_ownership_records(
                remote="prgs",
                host="gitea.example.com",
                org="Example-Org",
                repo="Example-Repo",
                branch="feat/x",
                pr_number=10,
                project_root="/tmp/repo",
                auth=FAKE_AUTH,
                base_api=(
                    "https://gitea.example.com/api/v1/repos/"
                    "Example-Org/Example-Repo"
                ),
            )
        cats = {r.get("category") for r in bundle.get("records") or []}
        self.assertIn(guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, cats)

    def test_o4_reconcile_merged_cleanups_runs_ownership_and_readback(self):
        from mcp_server import gitea_reconcile_merged_cleanups

        branch = "feat/reconcile-o4"
        ownership_calls = []

        def fake_collect(**kwargs):
            ownership_calls.append(kwargs)
            return {"records": [], "inventory_error": False}

        probe_calls = []

        def fake_probe(h, o, r, auth, br):
            probe_calls.append(br)
            return guard.classify_branch_readback_http_status(
                404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
            )

        report = {
            "entries": [
                {
                    "pr_number": 1,
                    "head_branch": branch,
                    "remote_branch": {"safe_to_delete_remote": True},
                    "local_worktree": {"safe_to_remove_worktree": False},
                }
            ],
            "reviewer_scratch_entries": [],
        }
        patch(
            "mcp_server.get_profile",
            return_value={
                "profile_name": "prgs-reconciler",
                "role": "reconciler",
                "allowed_operations": [
                    "gitea.read",
                    "gitea.branch.delete",
                    "gitea.pr.close",
                ],
                "forbidden_operations": [],
            },
        ).start()
        patch("mcp_server.api_get_all", return_value=[]).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
            return_value=report,
        ).start()
        patch(
            "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
            return_value=[],
        ).start()
        patch(
            "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
            return_value=(True, []),
        ).start()
        patch("mcp_server.verify_preflight_purity", return_value=None).start()
        patch(
            "mcp_server._collect_branch_ownership_records",
            side_effect=fake_collect,
        ).start()
        patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
        self.mock_api.side_effect = lambda *a, **k: {}

        res = gitea_reconcile_merged_cleanups(
            dry_run=False,
            execute_confirmed=True,
            remote="prgs",
        )
        self.assertTrue(res.get("performed") or res.get("executed"))
        self.assertTrue(ownership_calls, "ownership must run before delete")
        self.assertTrue(probe_calls, "post-delete readback must run")
        actions = res.get("actions") or []
        delete_actions = [
            a for a in actions if a.get("action") == "delete_remote_branch"
        ]
        self.assertEqual(len(delete_actions), 1)
        self.assertIn("verified_absent", delete_actions[0])
        self.assertIn("delete_acknowledged", delete_actions[0])
        self.assertTrue(delete_actions[0].get("verified_absent"))



if __name__ == "__main__":
    unittest.main()
