import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from mutation_profile_fixture import (
    mutation_profile_env,
    shared_mutation_env,
    install_deterministic_remote_urls,
)  # noqa: E402
"""Tests for the MCP server tool functions.

Each tool is tested by calling the underlying function directly (not through
the MCP protocol) with mocked API responses.
"""
import json
import os
import shutil
import sys
import tempfile
import unittest
from unittest.mock import patch, MagicMock

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

from mcp_server import (  # noqa: E402
    gitea_create_issue,
    gitea_create_pr,
    gitea_close_issue,
    gitea_list_issues,
    gitea_view_issue,
    gitea_mark_issue,
    gitea_mirror_refs,
    gitea_list_prs,
    gitea_view_pr,
    gitea_merge_pr,
    gitea_review_pr,
    gitea_delete_branch,
    gitea_reconcile_merged_cleanups,
    gitea_edit_pr,
    gitea_get_file,
    gitea_commit_files,
    gitea_whoami,
    gitea_get_profile,
    gitea_check_pr_eligibility,
    gitea_submit_pr_review,
    gitea_dry_run_pr_review,
    gitea_mark_final_review_decision,
    gitea_authorize_review_correction,
    init_review_decision_lock,

    gitea_list_issue_comments,
    gitea_create_issue_comment,
    gitea_lock_issue,
)
from gitea_auth import get_profile  # noqa: E402
import gitea_config  # noqa: E402

import mcp_server
install_deterministic_remote_urls()
import issue_lock_store
import gitea_auth

_orig_api_request = gitea_auth.api_request
def mockable_api_request(*args, **kwargs):
    import mcp_server
    return mcp_server.api_request(*args, **kwargs)

def setUpModule():
    gitea_auth.api_request = mockable_api_request
    patch("mcp_server._enforce_root_checkout_guard").start()
    patch("mcp_server._enforce_branches_only_author_mutation").start()

def tearDownModule():
    gitea_auth.api_request = _orig_api_request
    patch.stopall()


FAKE_AUTH = "Basic dGVzdDp0ZXN0"
FULL_HEAD_SHA = "a" * 40

_NO_BLOCKER_FEEDBACK = {
    "success": True,
    "has_blocking_change_requests": False,
    "review_feedback_stale": False,
}


def _init_reviewer_session(remote="prgs"):
    """Seed review decision lock and required workflow-load proof (#389)."""
    init_review_decision_lock(remote, "review_pr")
    mcp_server.gitea_load_review_workflow()


def _mark_request_changes_ready(pr_number=8, **kwargs):
    """Mark a request_changes decision ready with the #332 duplicate-
    suppression feedback fetch stubbed to 'no existing blocker'."""
    with patch("mcp_server.gitea_get_pr_review_feedback",
               return_value=dict(_NO_BLOCKER_FEEDBACK)):
        kwargs.setdefault("expected_head_sha", "abc123")
        return gitea_mark_final_review_decision(
            pr_number, "request_changes", **kwargs)


def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
    return {
        "id": review_id,
        "user": {"login": reviewer},
        "state": verdict,
        "commit_id": sha,
        "submitted_at": "2026-07-06T10:00:00Z",
        "dismissed": False,
    }


def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
    return [_formal_review(reviewer, "APPROVED", sha=sha)]


_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
WORKFLOW_REPO_LABELS = [
    {"id": 1, "name": "type:feature"},
    {"id": 2, "name": "status:in-progress"},
    {"id": 3, "name": "status:ready"},
    {"id": 4, "name": "status:pr-open"},
]


def _issue_with_labels(*names):
    return {"labels": [{"name": name} for name in names]}


def _reviewer_lease_comment(
    pr_number,
    *,
    session_id=_DEFAULT_LEASE_SESSION,
    head_sha="abc123",
    reviewer="reviewer-bot",
):
    from datetime import datetime, timezone

    import reviewer_pr_lease

    body = reviewer_pr_lease.format_lease_body(
        repo="Scaled-Tech-Consulting/Gitea-Tools",
        pr_number=pr_number,
        issue_number=407,
        reviewer_identity=reviewer,
        profile="gitea-reviewer",
        session_id=session_id,
        worktree="branches/review-test",
        phase="claimed",
        candidate_head=head_sha,
        target_branch="master",
        target_branch_sha="b" * 40,
        last_activity=datetime.now(timezone.utc),
    )
    return {"id": 9001, "body": body, "user": {"login": reviewer}}


def _install_owned_reviewer_lease(
    pr_number,
    *,
    session_id=_DEFAULT_LEASE_SESSION,
    head_sha="abc123",
):
    import merger_lease_adoption as mla
    import reviewer_pr_lease

    reviewer_pr_lease.clear_session_lease()
    reviewer_pr_lease.record_session_lease({
        "pr_number": pr_number,
        "session_id": session_id,
        "candidate_head": head_sha,
        "target_branch": "master",
        "comment_id": 9001,
    }, lease_provenance=mla.build_lease_provenance(
        source=mla.SOURCE_ACQUIRE,
        comment_id=9001,
    ))
    return patch(
        "mcp_server._fetch_pr_comments",
        return_value=[
            _reviewer_lease_comment(
                pr_number,
                session_id=session_id,
                head_sha=head_sha,
            )
        ],
    )


def _seed_ready_review_decision(
    pr_number,
    action,
    *,
    sha=FULL_HEAD_SHA,
    remote="prgs",
    org=None,
    repo=None,
):
    """Mark the review-decision lock ready without mark_final API calls (#399)."""
    import mcp_server as _m

    resolved_org, resolved_repo = org, repo
    if remote in _m.REMOTES:
        _, resolved_org, resolved_repo = _m._resolve(remote, None, org, repo)
    profile_name = (_m.get_profile().get("profile_name") or "").strip()
    session_lock = (
        (os.environ.get(_m.SESSION_PROFILE_LOCK_ENV) or "").strip()
        or profile_name
    )
    _m._save_review_decision_lock({
        "task": "review_pr",
        "remote": remote,
        "session_pid": os.getpid(),
        "session_profile": profile_name,
        "session_profile_lock": session_lock,
        "final_review_decision_ready": True,
        "ready_pr_number": pr_number,
        "ready_action": action,
        "ready_expected_head_sha": sha,
        "ready_remote": remote,
        "ready_org": resolved_org,
        "ready_repo": resolved_repo,
        "live_mutations": [],
        "correction_authorized": False,
        "correction_reason": None,
    })
    _m.gitea_load_review_workflow()


# Issue-write tools are profile-gated (#69).
# Default remote is dadeschools — use the host-aligned config profile so
# cross-host session binding does not fail closed (#714).
ISSUE_WRITE_ENV = shared_mutation_env(
    "test-author-dadeschools",
    GITEA_ALLOWED_OPERATIONS=(
        "gitea.issue.create,gitea.issue.close,gitea.issue.comment,"
        "gitea.pr.create,gitea.branch.push,gitea.repo.commit,gitea.read"
    ),
)

# create_pr is gated on author profile + namespace (#69, #209).
# Default remote is dadeschools; use matching config-backed author profile.
CREATE_PR_ENV = shared_mutation_env(
    "test-author-dadeschools",
    GITEA_ALLOWED_OPERATIONS=(
        "gitea.read,gitea.pr.create,gitea.branch.push,"
        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
    ),
    GITEA_FORBIDDEN_OPERATIONS="gitea.pr.approve,gitea.pr.merge,gitea.pr.review",
)

ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"


def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
    import issue_lock_provenance

    work_lease = {
        "operation_type": "author_issue_work",
        "issue_number": issue_number,
        "branch": branch_name,
        "claimant": {"username": "test-user", "profile": "test-author"},
        "expires_at": "2999-01-01T00:00:00Z",
    }
    record = {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "remote": "dadeschools",
        "org": "Scaled-Tech-Consulting",
        "repo": "Gitea-Tools",
        "worktree_path": "/tmp/test-worktree",
        "work_lease": work_lease,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant=work_lease.get("claimant"),
        ),
    }
    record.update(overrides)
    return record


def _clear_duplicate_context_fetcher(*_args, **_kwargs):
    """Default injectable duplicate-work context for lock/create_pr tests."""
    return [], [], {"status": "not_claimed"}


def _bind_test_lock(**overrides) -> str:
    remote = overrides.get("remote", "dadeschools")
    record = _sample_issue_lock(**overrides)
    if remote in mcp_server.REMOTES:
        profile = mcp_server.REMOTES[remote]
        record.setdefault("org", profile["org"])
        record.setdefault("repo", profile["repo"])
    record["remote"] = remote
    return issue_lock_store.bind_session_lock(record)


# ---------------------------------------------------------------------------
# Create Issue
# ---------------------------------------------------------------------------

class TestCreateIssue(unittest.TestCase):

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_issue(self, _auth, _get_all, mock_api, _role):
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            result = gitea_create_issue(title="Test issue", body="body text")
        self.assertEqual(result["number"], 1)
        self.assertNotIn("url", result)
        mock_api.assert_called_once()
        # Verify payload
        call_args = mock_api.call_args
        self.assertEqual(call_args[0][0], "POST")
        self.assertEqual(call_args[0][3]["title"], "Test issue")

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
        env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
        with patch.dict(os.environ, env, clear=False):
            result = gitea_create_issue(title="Test issue", body="body text")
        self.assertIn("issues/1", result["url"])

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
        mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
        env = {
            **ISSUE_WRITE_ENV,
            "GITEA_MCP_PROFILE": "test-author-prgs",
        }
        with patch.dict(os.environ, env, clear=False):
            import gitea_config
            gitea_config._active_profile_override = None
            try:
                import session_context_binding as _sc
                _sc._reset_session_context_for_testing()
            except Exception:
                pass
            result = gitea_create_issue(title="Test", remote="prgs")
        self.assertEqual(result["number"], 5)
        url = mock_api.call_args[0][1]
        self.assertIn("gitea.prgs.cc", url)
        self.assertIn("Scaled-Tech-Consulting", url)

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_issue_required_workflow_labels_blocks(self, _auth, _get_all, mock_api, _role):
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            result = gitea_create_issue(
                title="Test issue",
                require_workflow_labels=True,
            )
        self.assertFalse(result["success"])
        self.assertFalse(result["performed"])
        self.assertIn("issue is missing a type:* label", result["reasons"])
        mock_api.assert_not_called()


# ---------------------------------------------------------------------------
# Create PR
# ---------------------------------------------------------------------------
class TestCreatePR(unittest.TestCase):

    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    @patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_pr(self, _auth, mock_api, _role, _labels, _dup_fetcher):
        worktree = os.path.realpath(os.getcwd())

        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/issues/123" in url:
                return _issue_with_labels("type:feature", "status:in-progress")
            if method == "POST" and url.endswith("/pulls"):
                return {"number": 3, "html_url": "https://example.com/pulls/3"}
            if method == "PUT" and url.endswith("/issues/123/labels"):
                return [{"name": "type:feature"}, {"name": "status:pr-open"}]
            return {}

        mock_api.side_effect = api_side_effect
        with tempfile.TemporaryDirectory() as lock_dir:
            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
            with patch.dict(os.environ, env, clear=False):
                _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
                result = gitea_create_pr(
                    title="feat: X Closes #123",
                    head="feat/x",
                    base="main",
                    worktree_path=worktree,
                )
        self.assertEqual(result["number"], 3)
        self.assertNotIn("url", result)
        post_call = next(c for c in mock_api.call_args_list if c[0][0] == "POST")
        payload = post_call[0][3]
        self.assertEqual(payload["head"], "feat/x")
        self.assertEqual(payload["base"], "main")
        self.assertIn("Closes #123", payload["title"])
        put_call = next(c for c in mock_api.call_args_list if c[0][0] == "PUT")
        self.assertEqual(put_call[0][3], {"labels": [1, 4]})
        self.assertTrue(result["issue_status_transition"]["performed"])

    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    @patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role, _labels, _dup_fetcher):
        worktree = os.path.realpath(os.getcwd())

        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/issues/123" in url:
                return _issue_with_labels("type:feature", "status:in-progress")
            if method == "POST" and url.endswith("/pulls"):
                return {"number": 3, "html_url": "https://example.com/pulls/3"}
            if method == "PUT" and url.endswith("/issues/123/labels"):
                return [{"name": "type:feature"}, {"name": "status:pr-open"}]
            return {}

        mock_api.side_effect = api_side_effect
        with tempfile.TemporaryDirectory() as lock_dir:
            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
            with patch.dict(os.environ, env, clear=False):
                _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
                result = gitea_create_pr(
                    title="feat: X Closes #123",
                    head="feat/x",
                    base="main",
                    worktree_path=worktree,
                )
        self.assertIn("pulls/3", result["url"])

    @patch(
        "mcp_server.issue_duplicate_context_fetcher",
        return_value=([], [], {"status": "not_claimed"}),
    )
    @patch("mcp_server.api_get_all", return_value=[{"id": 1, "name": "type:feature"}])
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_blocks_when_pr_open_status_label_missing(
        self, _auth, mock_api, _role, _labels, _dup_fetcher
    ):
        worktree = os.path.realpath(os.getcwd())
        mock_api.return_value = _issue_with_labels("type:feature", "status:in-progress")
        with tempfile.TemporaryDirectory() as lock_dir:
            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
            with patch.dict(os.environ, env, clear=False):
                _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
                result = gitea_create_pr(
                    title="feat: X Closes #123",
                    head="feat/x",
                    base="main",
                    worktree_path=worktree,
                )
        self.assertFalse(result["success"])
        self.assertFalse(result["performed"])
        self.assertIn("status:pr-open", result["reasons"][0])
        self.assertFalse(
            any(c[0][0] == "POST" and c[0][1].endswith("/pulls")
                for c in mock_api.call_args_list)
        )

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_locked_issue_mismatch_fails(self, _auth, _role):
        worktree = os.path.realpath(os.getcwd())
        with tempfile.TemporaryDirectory() as lock_dir:
            env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
            with patch.dict(os.environ, env, clear=False):
                _bind_test_lock(
                    issue_number=123,
                    branch_name="feat/x",
                    worktree_path=worktree,
                )
                with self.assertRaises(ValueError) as ctx:
                    gitea_create_pr(
                        title="feat: X Closes #999",
                        head="feat/x",
                        base="main",
                        worktree_path=worktree,
                    )
        self.assertIn("Closes #123", str(ctx.exception))


# ---------------------------------------------------------------------------
# Close Issue
# ---------------------------------------------------------------------------
class TestCloseIssue(unittest.TestCase):

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_closes_issue(self, _auth, mock_api):
        mock_api.return_value = {"state": "closed"}
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            result = gitea_close_issue(issue_number=42)
        self.assertTrue(result["success"])
        self.assertIn("42", result["message"])
        patch_call = next(call for call in mock_api.call_args_list if call[0][0] == "PATCH")
        payload = patch_call[0][3]
        self.assertEqual(payload["state"], "closed")


# ---------------------------------------------------------------------------
# List Issues
# ---------------------------------------------------------------------------
class TestListIssues(unittest.TestCase):

    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_returns_formatted_list(self, _auth, mock_api):
        mock_api.return_value = [
            {
                "number": 1, "title": "Bug", "state": "open",
                "labels": [{"name": "bug"}],
                "assignee": {"login": "alice"},
            },
            {
                "number": 2, "title": "Feature", "state": "open",
                "labels": [], "assignee": None,
            },
        ]
        result = gitea_list_issues()
        self.assertEqual(len(result), 2)
        self.assertEqual(result[0]["number"], 1)
        self.assertEqual(result[0]["labels"], ["bug"])
        self.assertEqual(result[0]["assignee"], "alice")
        self.assertEqual(result[1]["assignee"], "")

    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_passes_label_filter(self, _auth, mock_api):
        mock_api.return_value = []
        gitea_list_issues(label="important")
        url = mock_api.call_args[0][0]
        self.assertIn("labels=important", url)

    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_passes_state_filter(self, _auth, mock_api):
        mock_api.return_value = []
        gitea_list_issues(state="closed")
        url = mock_api.call_args[0][0]
        self.assertIn("state=closed", url)


# ---------------------------------------------------------------------------
# View Issue
# ---------------------------------------------------------------------------
class TestViewIssue(unittest.TestCase):

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_returns_full_details(self, _auth, mock_api):
        mock_api.return_value = {
            "number": 7, "title": "MCP server", "body": "Build it",
            "state": "open", "labels": [{"name": "important"}],
            "assignee": {"login": "jason"},
            "html_url": "https://gitea.prgs.cc/issues/7",
        }
        with patch.dict(os.environ, {}, clear=True):
            result = gitea_view_issue(issue_number=7, remote="prgs")
        self.assertEqual(result["number"], 7)
        self.assertEqual(result["body"], "Build it")
        self.assertEqual(result["labels"], ["important"])
        self.assertEqual(result["assignee"], "jason")
        self.assertNotIn("url", result)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_view_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
        mock_api.return_value = {
            "number": 7, "title": "MCP server", "body": "Build it",
            "state": "open", "labels": [], "assignee": None,
            "html_url": "https://gitea.prgs.cc/issues/7",
        }
        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
            result = gitea_view_issue(issue_number=7, remote="prgs")
        self.assertIn("issues/7", result["url"])


# ---------------------------------------------------------------------------
# Mark Issue
# ---------------------------------------------------------------------------
class TestMarkIssue(unittest.TestCase):

    @patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_start_adds_label(self, _auth, mock_api, _labels):
        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/issues/5" in url:
                return _issue_with_labels("type:feature", "status:ready")
            if method == "PUT" and url.endswith("/issues/5/labels"):
                return [{"name": "type:feature"}, {"name": "status:in-progress"}]
            if method == "POST" and url.endswith("/issues/5/comments"):
                return {"id": 99}
            return {}

        mock_api.side_effect = api_side_effect
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            result = gitea_mark_issue(issue_number=5, action="start")
        self.assertTrue(result["success"])
        self.assertIn("claimed", result["message"])
        self.assertTrue(result.get("heartbeat_posted"))
        put_call = next(c for c in mock_api.call_args_list if c[0][0] == "PUT")
        self.assertEqual(put_call[0][3], {"labels": [1, 2]})

    @patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_done_removes_label(self, _auth, mock_api, _labels):
        mock_api.side_effect = [
            None,
        ]
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            result = gitea_mark_issue(issue_number=5, action="done")
        self.assertTrue(result["success"])
        self.assertIn("released", result["message"])

    def test_invalid_action_raises(self):
        with self.assertRaises(ValueError):
            gitea_mark_issue(issue_number=5, action="pause")

    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_missing_label_raises(self, _auth, mock_api, _labels):
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            with self.assertRaises(RuntimeError):
                gitea_mark_issue(issue_number=5, action="start")


# ---------------------------------------------------------------------------
# Auth errors
# ---------------------------------------------------------------------------
class TestAuthErrors(unittest.TestCase):

    @patch("mcp_server.get_auth_header", return_value=None)
    def test_no_credentials_raises(self, _auth):
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            with self.assertRaises(RuntimeError):
                gitea_create_issue(title="test")

    def test_unknown_remote_raises(self):
        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
            with self.assertRaises(ValueError):
                gitea_create_issue(title="test", remote="nonexistent")


# ---------------------------------------------------------------------------
# Mirror Refs
# ---------------------------------------------------------------------------
class TestMirrorRefs(unittest.TestCase):

    @patch("mcp_server.subprocess.run")
    def test_dry_run_default(self, mock_run):
        mock_run.return_value = MagicMock(
            stdout="DRY RUN\n", stderr="", returncode=0
        )
        result = gitea_mirror_refs()
        self.assertEqual(result["return_code"], 0)
        # Should NOT have --apply
        args = mock_run.call_args[0][0]
        self.assertNotIn("--apply", args)
        self.assertNotIn("--force", args)

    @patch("mcp_server.subprocess.run")
    def test_apply_flag(self, mock_run):
        mock_run.return_value = MagicMock(
            stdout="done\n", stderr="", returncode=0
        )
        result = gitea_mirror_refs(apply=True)
        args = mock_run.call_args[0][0]
        self.assertIn("--apply", args)

    @patch("mcp_server.subprocess.run")
    def test_force_flag(self, mock_run):
        mock_run.return_value = MagicMock(
            stdout="", stderr="", returncode=0
        )
        gitea_mirror_refs(apply=True, force=True)
        args = mock_run.call_args[0][0]
        self.assertIn("--apply", args)
        self.assertIn("--force", args)


# ---------------------------------------------------------------------------
# List PRs
# ---------------------------------------------------------------------------
class TestListPRs(unittest.TestCase):

    @patch("mcp_server.api_fetch_page")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_list_prs(self, _auth, mock_fetch):
        mock_fetch.return_value = (
            [{
                "number": 1, "title": "PR 1", "state": "open",
                "head": {"ref": "branch1"}, "base": {"ref": "main"},
                "html_url": "http://url1", "mergeable": True,
                "updated_at": "2026-07-05T12:00:00Z"
            }],
            {
                "page": 1, "per_page": 50, "returned_count": 1,
                "has_more": False, "next_page": None, "is_final_page": True,
            },
        )
        with patch.dict(os.environ, {}, clear=True):
            result = gitea_list_prs()
        self.assertEqual(len(result["prs"]), 1)
        self.assertEqual(result["prs"][0]["number"], 1)
        self.assertEqual(result["prs"][0]["head"], "branch1")
        self.assertNotIn("url", result["prs"][0])
        self.assertTrue(result["pagination"]["inventory_complete"])
        # Staleness fields for queue reconciliation (#166)
        self.assertEqual(result["prs"][0]["updated_at"], "2026-07-05T12:00:00Z")

    @patch("mcp_server.api_fetch_page")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_list_prs_reveal_opt_in_includes_url(self, _auth, mock_fetch):
        mock_fetch.return_value = (
            [{
                "number": 1, "title": "PR 1", "state": "open",
                "head": {"ref": "branch1"}, "base": {"ref": "main"},
                "html_url": "http://url1", "mergeable": True,
                "updated_at": "2026-07-05T12:00:00Z"
            }],
            {
                "page": 1, "per_page": 50, "returned_count": 1,
                "has_more": False, "next_page": None, "is_final_page": True,
            },
        )
        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
            result = gitea_list_prs()
        self.assertEqual(result["prs"][0]["url"], "http://url1")


# ---------------------------------------------------------------------------
# View PR
# ---------------------------------------------------------------------------
class TestViewPR(unittest.TestCase):

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_view_pr(self, _auth, mock_api):
        mock_api.return_value = {
            "number": 1, "title": "PR 1", "state": "open",
            "head": {"ref": "branch1"}, "base": {"ref": "main"},
            "html_url": "http://url1", "mergeable": True, "body": "description",
            "user": {"login": "user1"},
            "updated_at": "2026-07-05T12:00:00Z",
            "merged_at": None,
            "merge_commit_sha": None,
            "closed_at": None
        }
        with patch.dict(os.environ, {}, clear=True):
            result = gitea_view_pr(pr_number=1)
        self.assertEqual(result["number"], 1)
        self.assertEqual(result["body"], "description")
        self.assertEqual(result["user"], "user1")
        self.assertNotIn("url", result)
        # Staleness / merge fields for live reconciliation (#166) - passthrough
        self.assertEqual(result["updated_at"], "2026-07-05T12:00:00Z")
        self.assertIsNone(result["merged_at"])
        self.assertIsNone(result["merge_commit_sha"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_view_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
        mock_api.return_value = {
            "number": 1, "title": "PR 1", "state": "open",
            "head": {"ref": "branch1"}, "base": {"ref": "main"},
            "html_url": "http://url1", "mergeable": True, "body": "description",
            "user": {"login": "user1"},
            "updated_at": "2026-07-05T12:00:00Z", "merged_at": None, "merge_commit_sha": None, "closed_at": None
        }
        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
            result = gitea_view_pr(pr_number=1)
        self.assertEqual(result["url"], "http://url1")


# ---------------------------------------------------------------------------
# Merge PR
# ---------------------------------------------------------------------------
class TestMergePR(unittest.TestCase):
    """Gated merge workflow (#16). gitea_merge_pr is the only merge path."""

    def setUp(self):
        import reviewer_pr_lease

        # Clean ledger each merge test so residual/host durable #332 state
        # cannot short-circuit eligibility gates (#590 / #594).
        mcp_server._save_review_decision_lock(None)
        mcp_server.gitea_load_review_workflow()
        self._lease_patch = _install_owned_reviewer_lease(8)
        self._lease_patch.start()
        self._auth_identity_patch = patch(
            "mcp_server._authenticated_username", return_value="reviewer-bot"
        )
        self._auth_identity_patch.start()
        self.addCleanup(self._auth_identity_patch.stop)
        self.addCleanup(self._lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)
        self._pr_lease_comments_patch = patch(
            "mcp_server._list_pr_lease_comments", return_value=[]
        )
        self._pr_lease_comments_patch.start()
        self.addCleanup(self._pr_lease_comments_patch.stop)
        self._pr_work_lease_patch = patch(
            "mcp_server._pr_work_lease_reviewer_block",
            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
        )
        self._pr_work_lease_patch.start()
        self.addCleanup(self._pr_work_lease_patch.stop)

    def _pr(self, author, state="open", sha="abc123", mergeable=True):
        return {
            "user": {"login": author},
            "state": state,
            "head": {"sha": sha},
            "mergeable": mergeable,
        }

    def _confirm(self, n):
        return f"MERGE PR {n}"

    def _assert_no_merge_call(self, mock_api):
        for c in mock_api.call_args_list:
            method, url = c.args[0], c.args[1]
            self.assertFalse(
                method == "POST" and url.endswith("/merge"),
                f"unexpected merge mutation: {method} {url}",
            )

    def _feedback_reads(self, author="author-bot", sha="abc123"):
        """PR + reviews GETs for one gitea_get_pr_review_feedback call."""
        return [self._pr(author, sha=sha), _visible_approval_reviews(sha=sha)]

    def _eligibility_merge_reads(self, author="author-bot", sha="abc123"):
        """Eligibility user/PR plus #695 merge-approval feedback PR+reviews."""
        return [
            {"login": "merger-bot"},
            self._pr(author, sha=sha),
            *self._feedback_reads(author=author, sha=sha),
        ]

    # -- success --------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api):
        mock_api.side_effect = [
            *self._eligibility_merge_reads(),
            *self._feedback_reads(),
            {},  # merge POST
            {"merged_commit_sha": "mergecommit99"},  # read-back
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="abc123", do="squash",
                title="T", message="M", remote="prgs")
        self.assertTrue(r["performed"])
        self.assertEqual(r["authenticated_user"], "merger-bot")
        self.assertEqual(r["pr_author"], "author-bot")
        self.assertEqual(r["head_sha"], "abc123")
        self.assertEqual(r["merge_method"], "squash")
        self.assertEqual(r["merge_commit"], "mergecommit99")
        # 5th call is the merge POST with the requested method/title/message.
        merge_call = mock_api.call_args_list[6]
        self.assertEqual(merge_call.args[0], "POST")
        self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge"))
        payload = merge_call.args[3]
        self.assertEqual(payload["Do"], "squash")
        self.assertEqual(payload["MergeTitleField"], "T")
        self.assertEqual(payload["MergeMessageField"], "M")
        self.assertNotIn("force_merge", payload)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_expected_changed_files_match_allows(self, _auth, mock_api):
        mock_api.side_effect = [
            *self._eligibility_merge_reads(),
            [{"filename": "a.py"}, {"filename": "b.py"}],  # files
            *self._feedback_reads(),
            {},  # merge POST
            {"merged_commit_sha": "c1"},  # read-back
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_changed_files=["b.py", "a.py"],
                expected_head_sha="abc123", remote="prgs")
        self.assertTrue(r["performed"])

    # -- read-back / cleanup surfacing (#98) -----------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_readback_failure_reports_skipped_cleanup(self, _auth, mock_api):
        """Merge OK + read-back GET failure => explicit cleanup skip, not silence."""
        mock_api.side_effect = [
            *self._eligibility_merge_reads(),
            *self._feedback_reads(),
            {},  # merge POST
            RuntimeError("HTTP 502: Gitea upstream unavailable"),  # read-back fails
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="abc123", remote="prgs",
            )
        # The merge itself is still reported performed/successful.
        self.assertTrue(r["performed"])
        self.assertEqual(r["merge_result"], "PR #8 merged via 'merge'.")
        self.assertIsNone(r["merge_commit"])
        # The skip is explicit, not silent.
        self.assertEqual(r["cleanup_status"], "skipped (merge read-back failed)")
        # No tracker-cleanup API traffic after the failed read-back:
        # user, PR (eligibility), feedback PR+reviews, merge POST, read-back.
        self.assertEqual(mock_api.call_count, 8)
        for c in mock_api.call_args_list:
            self.assertNotEqual(c.args[0], "DELETE")

    @patch("mcp_server.cleanup_in_progress_for_pr",
           side_effect=RuntimeError("boom token secret-xyz"))
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_cleanup_exception_surfaced_and_redacted(self, _auth, mock_api, _cleanup):
        """Unexpected cleanup exception => merge still succeeds; error surfaced redacted."""
        mock_api.side_effect = [
            *self._eligibility_merge_reads(),
            *self._feedback_reads(),
            {},  # merge POST
            {"merged_commit_sha": "c9"},  # read-back OK
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="abc123", remote="prgs",
            )
        self.assertTrue(r["performed"])
        self.assertEqual(r["merge_commit"], "c9")
        self.assertTrue(r["cleanup_status"].startswith("skipped (cleanup error:"))
        self.assertNotIn("secret-xyz", r["cleanup_status"])

    # -- confirmation ---------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_missing_confirmation_blocks_with_no_api_call(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(pr_number=8, confirmation="", expected_head_sha="abc123", remote="prgs")
        self.assertFalse(r["performed"])
        self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"]))
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_wrong_confirmation_blocks(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", expected_head_sha="abc123", remote="prgs")
        self.assertFalse(r["performed"])
        mock_api.assert_not_called()

    # -- identity / profile / eligibility fail-closed -------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_reviewer_profile_cannot_merge(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "prgs-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,approve"}
        with patch.dict(os.environ, env, clear=False):
            with self.assertRaises(RuntimeError) as ctx:
                gitea_merge_pr(
                    pr_number=8, confirmation=self._confirm(8), remote="prgs"
                )
        self.assertIn("Reviewer profile cannot merge. Use a merger profile/session after formal approval at current head.", str(ctx.exception))
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_self_author_cannot_merge(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8), remote="prgs")
        self.assertFalse(r["performed"])
        self.assertIn("authenticated user is PR author", r["reasons"])
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.get_auth_header", return_value=None)
    def test_unknown_identity_blocks(self, _auth):
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8), remote="prgs")
        self.assertFalse(r["performed"])
        self.assertIsNone(r["authenticated_user"])
        self.assertIn("authenticated identity could not be determined", r["reasons"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_unknown_profile_blocks(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
        env = {}  # no GITEA_ALLOWED_OPERATIONS → empty allowed ops
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8), remote="prgs")
        self.assertFalse(r["performed"])
        self.assertIn(
            "profile has no configured allowed operations (fail closed)",
            r["reasons"])
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_env_clear_does_not_adopt_host_review_mutation_ledger(
        self, _auth, mock_api
    ):
        """#590: patch.dict(clear=True) must not load host terminal mutations.

        Simulates a dirty host session-state file for profile gitea-default
        (the profile active when env is fully cleared). Isolation must keep
        the #332 hard-stop ledger clean so the empty-profile gate fires.
        """
        import tempfile

        import mcp_session_state

        poison_dir = tempfile.mkdtemp(prefix="gitea-host-poison-")
        mcp_session_state.save_state(
            kind=mcp_session_state.KIND_DECISION_LOCK,
            profile_identity="gitea-default",
            state_dir=poison_dir,
            payload={
                "task": "review_pr",
                "remote": "prgs",
                "final_review_decision_ready": True,
                "ready_pr_number": 8,
                "ready_action": "request_changes",
                "live_mutations": [
                    {
                        "pr_number": 8,
                        "action": "request_changes",
                        "review_id": 99,
                        "review_state": "request_changes",
                    }
                ],
            },
        )
        # If isolation only used the env var, clear=True would fall back to
        # DEFAULT_STATE_DIR and adopt this poison ledger.
        mcp_server._REVIEW_DECISION_LOCK = None
        mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
        with patch.object(mcp_session_state, "DEFAULT_STATE_DIR", poison_dir):
            with patch.dict(os.environ, {}, clear=True):
                r = gitea_merge_pr(
                    pr_number=8,
                    confirmation=self._confirm(8),
                    remote="prgs",
                )
        self.assertFalse(r["performed"])
        self.assertIn(
            "profile has no configured allowed operations (fail closed)",
            r["reasons"],
        )
        self.assertFalse(
            any("terminal review mutation already consumed" in x for x in r["reasons"]),
            msg=f"host ledger leaked into reasons: {r['reasons']}",
        )
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_profile_without_merge_permission_blocks(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            with self.assertRaises(RuntimeError) as ctx:
                gitea_merge_pr(
                    pr_number=8, confirmation=self._confirm(8), remote="prgs")
        self.assertIn("Reviewer profile cannot merge. Use a merger profile/session after formal approval at current head.", str(ctx.exception))
        self._assert_no_merge_call(mock_api)

    # -- PR state / mergeability ----------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_closed_pr_blocks(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "merger-bot"}, self._pr("author-bot", state="closed")]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8), remote="prgs")
        self.assertFalse(r["performed"])
        self.assertIn("PR is not open (state=closed)", r["reasons"])
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_non_mergeable_pr_blocks(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "merger-bot"}, self._pr("author-bot", mergeable=False)]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8), remote="prgs")
        self.assertFalse(r["performed"])
        self.assertIn("PR is not mergeable", r["reasons"])
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_unknown_mergeability_fails_closed(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "merger-bot"}, self._pr("author-bot", mergeable=None)]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8), remote="prgs")
        self.assertFalse(r["performed"])
        self.assertIn("PR mergeability unknown", r["reasons"])
        self._assert_no_merge_call(mock_api)

    # -- head SHA / changed files ---------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_head_sha_mismatch_blocks(self, _auth, mock_api):
        # Eligibility (#695) needs approval feedback before head-gate runs.
        mock_api.side_effect = [
            *self._eligibility_merge_reads(sha="abc123"),
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="deadbeef", remote="prgs")
        self.assertFalse(r["performed"])
        self.assertTrue(any(
            "expected head SHA does not match current PR head (fail closed)" in reason
            or "PR head changed during lease" in reason
            for reason in r["reasons"]
        ))
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_changed_files_mismatch_blocks(self, _auth, mock_api):
        mock_api.side_effect = [
            *self._eligibility_merge_reads(),
            [{"filename": "a.py"}, {"filename": "c.py"}],  # actual files
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_changed_files=["a.py", "b.py"],
                expected_head_sha="abc123", remote="prgs")
        self.assertFalse(r["performed"])
        self.assertIn(
            "PR changed files do not match expected_changed_files (fail closed)",
            r["reasons"])
        self._assert_no_merge_call(mock_api)

    # -- misc -----------------------------------------------------------------

    @patch("mcp_server.api_request")
    def test_invalid_merge_method_rejected(self, mock_api):
        with patch.dict(os.environ, {}, clear=True):
            r = gitea_merge_pr(
                pr_number=8, confirmation="MERGE PR 8", do="octopus", expected_head_sha="abc123", remote="prgs")
        self.assertFalse(r["performed"])
        self.assertTrue(any("unknown merge method" in x for x in r["reasons"]))
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_output_redacts_secrets(self, _auth, mock_api):
        mock_api.side_effect = [
            *self._eligibility_merge_reads(),
            *self._feedback_reads(),
            {}, {"merged_commit_sha": "c1"},
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge",
               "GITEA_TOKEN": "super-secret-token"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="abc123", remote="prgs",
            )
        blob = repr(r).lower()
        for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
            self.assertNotIn(secret, blob)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_merge_error_message_redacts_credential(self, _auth, mock_api):
        mock_api.side_effect = [
            *self._eligibility_merge_reads(),
            *self._feedback_reads(),
            RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="abc123", remote="prgs",
            )
        self.assertFalse(r["performed"])
        blob = repr(r)
        self.assertIn("[REDACTED]", blob)
        self.assertNotIn("abc-secret-xyz", blob)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_merge_blocked_on_stale_approval_head(self, _auth, mock_api):
        old_sha = "8b61c4b41f1b49b271ed3b99657431cf06eeda3e"
        new_sha = "3e4b721d60e97147ba0704773cf57cd0d42cbe31"
        # #695: eligibility itself now fails closed on stale approval head.
        mock_api.side_effect = [
            {"login": "merger-bot"}, self._pr("author-bot", sha=new_sha),
            self._pr("author-bot", sha=new_sha),
            [_formal_review("reviewer-bot", "APPROVED", sha=old_sha)],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8), remote="prgs",
                expected_head_sha=new_sha)
        self.assertFalse(r["performed"])
        self.assertTrue(r.get("approval_visible"))
        self.assertFalse(r.get("approval_at_current_head"))
        self.assertTrue(any("stale approval" in x for x in r["reasons"]))
        self.assertTrue(any(old_sha in x for x in r["reasons"]))
        self.assertTrue(any(new_sha in x for x in r["reasons"]))
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_merge_blocked_without_visible_approval(self, _auth, mock_api):
        # #695: eligibility denies merge when no non-quarantined APPROVED review.
        mock_api.side_effect = [
            {"login": "merger-bot"}, self._pr("author-bot"),
            self._pr("author-bot"),
            [_formal_review("sysadmin", "PENDING")],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="abc123", remote="prgs",
            )
        self.assertFalse(r["performed"])
        self.assertFalse(r.get("approval_visible"))
        self.assertTrue(
            any(
                "no non-quarantined APPROVED review" in x
                or "no visible APPROVED review" in x
                or "merge eligibility denied" in x
                for x in r["reasons"]
            ),
            msg=r["reasons"],
        )
        self._assert_no_merge_call(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_merge_blocked_on_request_changes(self, _auth, mock_api):
        # #695: eligibility denies merge on undismissed REQUEST_CHANGES.
        mock_api.side_effect = [
            {"login": "merger-bot"}, self._pr("author-bot"),
            self._pr("author-bot"),
            [
                _formal_review("reviewer-bot", "APPROVED"),
                _formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=2),
            ],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_merge_pr(
                pr_number=8, confirmation=self._confirm(8),
                expected_head_sha="abc123", remote="prgs",
            )
        self.assertFalse(r["performed"])
        self.assertTrue(r.get("has_blocking_change_requests"))
        self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
        self._assert_no_merge_call(mock_api)


class TestNoUngatedMergePath(unittest.TestCase):
    """Prove no other exposed tool can merge (#16 surface audit)."""

    def test_submit_pr_review_has_no_merge(self):
        import inspect
        import mcp_server
        # No merge parameter on the review-mutation tool.
        params = inspect.signature(mcp_server.gitea_submit_pr_review).parameters
        self.assertNotIn("merge", params)
        # And 'merge' is not a reviewable action.
        self.assertNotIn("merge", mcp_server._REVIEW_ACTIONS)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_review_pr_merge_true_makes_no_api_call(self, _auth, mock_api):
        r = gitea_review_pr(pr_number=1, event="APPROVE", body="x", merge=True)
        self.assertFalse(r["success"])
        self.assertEqual(mock_api.call_count, 0)

    def test_only_merge_endpoint_is_in_gated_tool(self):
        # Source-level audit: the merge endpoint appears only inside
        # gitea_merge_pr, which is the gated path.
        import inspect
        import mcp_server
        src = inspect.getsource(mcp_server)
        self.assertEqual(src.count("/merge\""), 1)
        merge_src = inspect.getsource(mcp_server.gitea_merge_pr)
        self.assertIn("/merge\"", merge_src)

    def test_readme_no_longer_advertises_ungated_cli_merge(self):
        import pathlib
        readme = (pathlib.Path(__file__).resolve().parent.parent
                  / "README.md").read_text(encoding="utf-8")
        # The old ungated example command must be gone.
        self.assertNotIn('--body "Approved" --merge', readme)
        # And the gated messaging / audit deferral must be present.
        self.assertIn("disabled", readme.lower())
        self.assertIn("gitea_merge_pr", readme)
        self.assertIn("#18", readme)


# ---------------------------------------------------------------------------
# Review PR
# ---------------------------------------------------------------------------
class TestReviewPR(unittest.TestCase):

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_legacy_review_pr_merge_fails_closed(self, _auth, mock_api):
        result = gitea_review_pr(
            pr_number=1,
            event="APPROVE",
            body="Looks good",
            merge=True
        )
        self.assertFalse(result["success"])
        self.assertIn("no longer supported", result["message"])
        self.assertEqual(mock_api.call_count, 0)

    @patch("mcp_server.api_get_all")
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.get_profile")
    def test_legacy_review_pr_uses_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
        # Mock profile to lack approve capability (fails gate)
        mock_get_profile.return_value = {
            "profile_name": "gitea-readonly",
            "allowed_operations": ["read"],
            "forbidden_operations": [],
            "base_url": None,
        }
        mock_get_all.return_value = [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "author1"}}]
        mock_api.return_value = {"login": "reviewer1"}
        result = gitea_review_pr(
            pr_number=1,
            event="APPROVE",
            body="Looks good",
            merge=False
        )
        self.assertFalse(result["success"])
        self.assertIn("Review/Merge Blocked", result["message"])
        self.assertIn("Author profile", result["message"])

    @patch("mcp_server.api_fetch_page")
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.get_profile")
    def test_legacy_review_pr_self_approval_blocked(self, mock_get_profile, _auth, mock_api, mock_fetch):
        mock_get_profile.return_value = {
            "profile_name": "gitea-reviewer",
            "allowed_operations": ["read", "approve"],
            "forbidden_operations": [],
            "base_url": None,
        }
        head_sha = FULL_HEAD_SHA
        mock_fetch.return_value = (
            [{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": head_sha}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}],
            {"page": 1, "per_page": 50, "returned_count": 1, "has_more": False, "next_page": None, "is_final_page": True}
        )
        # mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
        mock_api.side_effect = [
            {"login": "jcwalker3"}, # /api/v1/user (inventory)
            {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
            {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": head_sha}, "mergeable": True}, # /pulls/1
        ]
        from mcp_server import init_review_decision_lock
        init_review_decision_lock("prgs", "review_pr")
        with patch("mcp_server._list_pr_lease_comments", return_value=[]):
            _seed_ready_review_decision(1, "approve", sha=head_sha, remote="prgs")
        result = gitea_review_pr(
            pr_number=1,
            event="APPROVE",
            body="Self approve",
            merge=False,
            remote="prgs",
            final_review_decision_ready=True,
        )
        self.assertFalse(result["success"])
        self.assertIn("Review submission failed eligibility gates", result["message"])
        self.assertIn("authenticated user is PR author", result["message"])


# ---------------------------------------------------------------------------
# Delete Branch
# ---------------------------------------------------------------------------
class TestDeleteBranch(unittest.TestCase):

    DELETE_PROFILE = {
        # #729: delete_branch is reconciler-owned.
        "profile_name": "test-reconciler-deleter",
        "role": "reconciler",
        "allowed_operations": [
            "gitea.read",
            "gitea.pr.create",
            "gitea.branch.push",
            "gitea.branch.delete",
        ],
        "forbidden_operations": [],
        "audit_label": "test-author-deleter",
    }

    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_delete_branch(self, _auth, mock_api, _profile):
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", resolved_role="reconciler")
        mock_api.return_value = {}
        result = gitea_delete_branch(branch="feat/branch")
        self.assertTrue(result["success"])
        self.assertIn("deleted", result["message"])
        # Check url encoding of branch name
        url = mock_api.call_args[0][1]
        self.assertIn("feat%2Fbranch", url)


# ---------------------------------------------------------------------------
# Reconcile merged cleanups (#269)
# ---------------------------------------------------------------------------
READ_ENV = {
    "GITEA_ALLOWED_OPERATIONS": "gitea.read",
}


class TestReconcileMergedCleanups(unittest.TestCase):

    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_dry_run_builds_report_without_mutations(self, _auth, mock_api_get_all):
        mock_api_get_all.side_effect = [
            [
                {
                    "number": 50,
                    "title": "merged feature",
                    "body": "Closes #50",
                    "merged": True,
                    "merged_at": "2026-07-06T12:00:00Z",
                    "merge_commit_sha": "deadbeef",
                    "head": {"ref": "feat/issue-50-example", "sha": "cafebabe"},
                }
            ],
            [],
        ]
        with patch.dict(os.environ, READ_ENV, clear=True):
            with patch(
                "mcp_server._remote_branch_exists",
                return_value=True,
            ):
                with patch(
                    "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
                    return_value=True,
                ):
                    result = gitea_reconcile_merged_cleanups(
                        dry_run=True,
                        remote="prgs",
                        limit=10,
                    )
        self.assertTrue(result["success"])
        self.assertTrue(result["dry_run"])
        self.assertFalse(result["executed"])
        self.assertEqual(result["merged_pr_count"], 1)
        self.assertEqual(result["entries"][0]["issue_number"], 50)

    def test_execute_requires_confirmation(self):
        with patch.dict(os.environ, READ_ENV, clear=True):
            with self.assertRaises(ValueError) as ctx:
                gitea_reconcile_merged_cleanups(dry_run=False, execute_confirmed=False)
        self.assertIn("execute_confirmed", str(ctx.exception))


# ---------------------------------------------------------------------------
# Edit PR
# ---------------------------------------------------------------------------
class TestEditPR(unittest.TestCase):

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_edit_pr_success(self, _auth, mock_api):
        mock_api.return_value = {
            "number": 1,
            "title": "New Title",
            "body": "New description",
            "state": "open",
            "html_url": "https://gitea.example.com/pulls/1"
        }
        result = gitea_edit_pr(
            pr_number=1,
            title="New Title",
            body="New description",
            state="open"
        )
        self.assertTrue(result["success"])
        self.assertEqual(result["title"], "New Title")
        self.assertEqual(result["body"], "New description")

        # Verify PATCH and payload
        call_args = mock_api.call_args
        self.assertEqual(call_args[0][0], "PATCH")
        self.assertEqual(call_args[0][3]["title"], "New Title")
        self.assertEqual(call_args[0][3]["body"], "New description")
        self.assertEqual(call_args[0][3]["state"], "open")

    def test_edit_pr_no_fields_raises(self):
        with self.assertRaises(ValueError):
            gitea_edit_pr(pr_number=1)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header")
    def test_edit_pr_no_fields_validates_before_auth(self, mock_auth, mock_api):
        # No-fields validation must not depend on credentials/network: it raises
        # ValueError before touching auth or the API, even with no creds.
        mock_auth.return_value = None  # simulate an unauthenticated environment
        with self.assertRaises(ValueError):
            gitea_edit_pr(pr_number=1)
        mock_auth.assert_not_called()
        mock_api.assert_not_called()


# ---------------------------------------------------------------------------
# Get File
# ---------------------------------------------------------------------------
class TestGetFile(unittest.TestCase):

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_get_file_success(self, _auth, mock_api):
        mock_api.return_value = {
            "name": "README.md",
            "path": "README.md",
            "sha": "3a0b123",
            "size": 100,
            "encoding": "base64",
            "content": "SGVsbG8gV29ybGQ="
        }
        result = gitea_get_file(filepath="README.md", ref="main")
        self.assertEqual(result["name"], "README.md")
        self.assertEqual(result["sha"], "3a0b123")
        self.assertEqual(result["content"], "SGVsbG8gV29ybGQ=")

        # Verify endpoint and GET method
        call_args = mock_api.call_args
        self.assertEqual(call_args[0][0], "GET")
        self.assertIn("contents/README.md", call_args[0][1])
        self.assertIn("ref=main", call_args[0][1])


# ---------------------------------------------------------------------------
# Commit Files
# ---------------------------------------------------------------------------
class TestCommitFiles(unittest.TestCase):

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_commit_files_success(self, _auth, mock_api, _role):
        mock_api.return_value = {
            "commit": {"sha": "commit-sha-123"},
            "branch": {"name": "test-branch"}
        }
        files = [
            {"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
        ]
        env = shared_mutation_env("test-author-dadeschools")
        with patch.dict(os.environ, env, clear=False):
            result = gitea_commit_files(
                files=files,
                message="Initial commit",
                new_branch="test-branch"
            )
        self.assertTrue(result["success"])
        self.assertEqual(result["commit"], "commit-sha-123")
        self.assertEqual(result["branch"], "test-branch")

        # Verify POST method and payload
        call_args = mock_api.call_args
        self.assertEqual(call_args[0][0], "POST")
        self.assertIn("/contents", call_args[0][1])
        payload = call_args[0][3]
        self.assertEqual(payload["message"], "Initial commit")
        self.assertEqual(payload["new_branch"], "test-branch")
        self.assertEqual(payload["files"], files)


# ---------------------------------------------------------------------------
# Whoami (authenticated-user identity lookup)
# ---------------------------------------------------------------------------
class TestWhoami(unittest.TestCase):

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_returns_safe_identity(self, _auth, mock_api):
        mock_api.return_value = {
            "id": 42,
            "login": "reviewer-bot",
            "full_name": "Reviewer Bot",
            "email": "reviewer@example.com",
        }
        result = gitea_whoami(remote="prgs")
        self.assertTrue(result["authenticated"])
        self.assertEqual(result["username"], "reviewer-bot")
        self.assertEqual(result["display_name"], "Reviewer Bot")
        self.assertEqual(result["user_id"], 42)
        # Endpoint URLs are hidden from normal LLM-facing output (#120);
        # the logical remote name is the addressing surface.
        self.assertNotIn("server", result)
        self.assertEqual(result["remote"], "prgs")
        # Read-only: GET against the authenticated-user endpoint.
        call_args = mock_api.call_args
        self.assertEqual(call_args[0][0], "GET")
        self.assertTrue(call_args[0][1].endswith("/api/v1/user"))

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_never_exposes_secrets(self, _auth, mock_api):
        mock_api.return_value = {"id": 1, "login": "someone"}
        result = gitea_whoami(remote="prgs")
        blob = repr(result).lower()
        for secret in ("token", "authorization", "basic ", "password", FAKE_AUTH.lower()):
            self.assertNotIn(secret, blob)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_fails_closed_without_login(self, _auth, mock_api):
        mock_api.return_value = {"id": 1}  # no 'login'
        with self.assertRaises(RuntimeError):
            gitea_whoami(remote="prgs")

    def test_rejects_unknown_remote(self):
        with self.assertRaises(ValueError):
            gitea_whoami(remote="nope")


# ---------------------------------------------------------------------------
# Runtime profile (env-configured profile metadata) — issue #19
# ---------------------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server._auth")
    def test_gitea_get_authenticated_user_alias(self, _auth, mock_api):
        mock_api.return_value = {
            "login": "alias_user",
            "full_name": "Alias User",
            "id": 999,
            "email": "alias@example.com"
        }
        from mcp_server import gitea_get_authenticated_user
        result = gitea_get_authenticated_user(remote="prgs")
        self.assertEqual(result["username"], "alias_user")

    @patch("mcp_server.api_request")
    @patch("mcp_server._auth")
    def test_gitea_get_current_user_alias(self, _auth, mock_api):
        mock_api.return_value = {
            "login": "alias_user",
            "full_name": "Alias User",
            "id": 999,
            "email": "alias@example.com"
        }
        from mcp_server import gitea_get_current_user
        result = gitea_get_current_user(remote="prgs")
        self.assertEqual(result["username"], "alias_user")


class TestRuntimeProfile(unittest.TestCase):

    def test_defaults_when_unset(self):
        with patch.dict(os.environ, {}, clear=True):
            p = get_profile()
        self.assertEqual(p["profile_name"], "gitea-default")
        self.assertEqual(p["allowed_operations"], [])
        self.assertIsNone(p["base_url"])

    def test_reads_env_metadata(self):
        env = {
            "GITEA_PROFILE_NAME": "gitea-reviewer",
            "GITEA_ALLOWED_OPERATIONS": "read, review , approve",
            "GITEA_BASE_URL": "https://gitea.example.invalid",
        }
        with patch.dict(os.environ, env, clear=False):
            p = get_profile()
        self.assertEqual(p["profile_name"], "gitea-reviewer")
        self.assertEqual(p["allowed_operations"], ["read", "review", "approve"])
        self.assertEqual(p["base_url"], "https://gitea.example.invalid")

    def test_never_includes_token(self):
        env = {
            "GITEA_PROFILE_NAME": "gitea-author",
            "GITEA_TOKEN": "super-secret-token",
        }
        with patch.dict(os.environ, env, clear=False):
            p = get_profile()
        blob = repr(p).lower()
        # The token VALUE must never appear. (The field name
        # 'token_source_name' is non-secret metadata and may exist.)
        self.assertNotIn("super-secret-token", blob)
        self.assertIsNone(p.get("token_source_name"))  # GITEA_TOKEN_SOURCE unset

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_whoami_surfaces_profile_without_token(self, _auth, mock_api):
        mock_api.return_value = {"id": 7, "login": "rev"}
        env = {
            "GITEA_PROFILE_NAME": "gitea-reviewer",
            "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
            "GITEA_TOKEN": "super-secret-token",
        }
        with patch.dict(os.environ, env, clear=False):
            result = gitea_whoami(remote="prgs")
        self.assertEqual(result["profile"]["profile_name"], "gitea-reviewer")
        self.assertEqual(
            result["profile"]["allowed_operations"], ["read", "review", "approve"]
        )
        blob = repr(result).lower()
        for secret in ("super-secret-token", "token", "authorization", "basic "):
            self.assertNotIn(secret, blob)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_whoami_v2_metadata(self, _auth, mock_api):
        mock_api.return_value = {"id": 7, "login": "rev"}
        env = {
            "GITEA_PROFILE_NAME": "gitea-reviewer",
            "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
            "GITEA_FORBIDDEN_OPERATIONS": "merge",
            "GITEA_AUDIT_LABEL": "reviewer-runtime",
            "GITEA_TOKEN_SOURCE": "keychain:prgs-reviewer-token",
        }
        with patch.dict(os.environ, env, clear=False):
            result = gitea_whoami(remote="prgs")
        profile = result["profile"]
        self.assertEqual(profile["environment"], None)
        self.assertEqual(profile["service"], None)
        self.assertEqual(profile["identity"], None)
        self.assertEqual(profile["role"], None)
        self.assertEqual(profile["profile_address"], None)
        self.assertEqual(profile["execution_profile"], None)
        self.assertEqual(profile["audit_label"], "reviewer-runtime")
        self.assertEqual(profile["auth_source_type"], "keychain")
        self.assertEqual(profile["forbidden_operations"], ["merge"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.get_profile")
    def test_whoami_v2_resolved_metadata(self, mock_get_profile, _auth, mock_api):
        mock_api.return_value = {"id": 7, "login": "rev"}
        mock_get_profile.return_value = {
            "profile_name": "prgs.gitea.reviewer",
            "allowed_operations": ["read", "review"],
            "forbidden_operations": ["merge"],
            "audit_label": "rev-audit",
            "token_source_name": "keychain:prgs-reviewer-token",
            "auth_source_type": "keychain",
            "base_url": "https://gitea.prgs.cc",
            "username": "sysadmin",
            "default_owner": "Scaled-Tech-Consulting",
            "profile_path": "prgs.gitea.reviewer",
            "environment": "prgs",
            "service": "gitea",
            "identity": "reviewer",
            "role": "reviewer",
            "execution_profile": "reviewer-profile",
        }
        result = gitea_whoami(remote="prgs")
        profile = result["profile"]
        self.assertEqual(profile["environment"], "prgs")
        self.assertEqual(profile["service"], "gitea")
        self.assertEqual(profile["identity"], "reviewer")
        self.assertEqual(profile["role"], "reviewer")
        self.assertEqual(profile["profile_address"], "prgs.gitea.reviewer")
        self.assertEqual(profile["execution_profile"], "reviewer-profile")
        self.assertEqual(profile["audit_label"], "rev-audit")
        self.assertEqual(profile["auth_source_type"], "keychain")
        self.assertEqual(profile["forbidden_operations"], ["merge"])



# ---------------------------------------------------------------------------
# Profile discovery (read-only) — issue #13
# ---------------------------------------------------------------------------
class TestProfileDiscovery(unittest.TestCase):

    def test_get_profile_new_fields_default_empty(self):
        with patch.dict(os.environ, {}, clear=True):
            p = get_profile()
        self.assertEqual(p["forbidden_operations"], [])
        self.assertIsNone(p["audit_label"])
        self.assertIsNone(p["token_source_name"])

    def test_get_profile_reads_all_metadata(self):
        env = {
            "GITEA_PROFILE_NAME": "gitea-reviewer",
            "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
            "GITEA_FORBIDDEN_OPERATIONS": "merge, branch.push",
            "GITEA_AUDIT_LABEL": "reviewer-runtime",
            "GITEA_TOKEN_SOURCE": "GITEA_TOKEN",
        }
        with patch.dict(os.environ, env, clear=False):
            p = get_profile()
        self.assertEqual(p["forbidden_operations"], ["merge", "branch.push"])
        self.assertEqual(p["audit_label"], "reviewer-runtime")
        self.assertEqual(p["token_source_name"], "GITEA_TOKEN")

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_discovery_verified_identity(self, _auth, mock_api):
        mock_api.return_value = {"id": 3, "login": "reviewer-bot"}
        env = {
            "GITEA_PROFILE_NAME": "gitea-reviewer",
            "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
            "GITEA_TOKEN_SOURCE": "GITEA_TOKEN",
            "GITEA_TOKEN": "super-secret-token",
        }
        with patch.dict(os.environ, env, clear=False):
            result = gitea_get_profile(remote="prgs")
        self.assertEqual(result["profile_name"], "gitea-reviewer")
        self.assertEqual(result["allowed_operations"], ["read", "review", "approve"])
        self.assertEqual(result["authenticated_username"], "reviewer-bot")
        self.assertEqual(result["identity_status"], "verified")
        # Endpoint URLs and token source names are hidden from normal
        # LLM-facing output (#120); auth is reported as a status only.
        self.assertNotIn("server", result)
        self.assertNotIn("base_url", result)
        self.assertNotIn("token_source_name", result)
        self.assertEqual(result["auth_status"], "configured")
        # Read-only: only a GET to the user endpoint was issued.
        self.assertEqual(mock_api.call_args[0][0], "GET")
        self.assertTrue(mock_api.call_args[0][1].endswith("/api/v1/user"))

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_discovery_never_exposes_secrets(self, _auth, mock_api):
        mock_api.return_value = {"id": 3, "login": "reviewer-bot"}
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_TOKEN": "super-secret-token"}
        with patch.dict(os.environ, env, clear=False):
            result = gitea_get_profile(remote="prgs")
        blob = repr(result).lower()
        for secret in ("super-secret-token", "token dgvzd", "authorization", "basic ", FAKE_AUTH.lower()):
            self.assertNotIn(secret, blob)

    @patch("mcp_server.get_auth_header", return_value=None)
    def test_discovery_identity_unavailable_fails_soft(self, _auth):
        # No credentials -> _auth raises inside the tool; identity marked
        # unavailable but the profile config is still returned (not raised).
        with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
            result = gitea_get_profile(remote="prgs")
        self.assertEqual(result["profile_name"], "gitea-author")
        self.assertIsNone(result["authenticated_username"])
        self.assertEqual(result["identity_status"], "unavailable")

    def test_discovery_can_skip_identity(self):
        with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
            result = gitea_get_profile(remote="prgs", resolve_identity=False)
        self.assertEqual(result["identity_status"], "not_resolved")
        self.assertIsNone(result["authenticated_username"])

    def test_discovery_unknown_remote_marks_unknown(self):
        with patch.dict(os.environ, {}, clear=True):
            result = gitea_get_profile(remote="nope")
        self.assertEqual(result["identity_status"], "unknown")
        self.assertIsNone(result["remote"])
        self.assertIn("remote_error", result)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    @patch("mcp_server.get_profile")
    def test_get_profile_v2_resolved_metadata(self, mock_get_profile, _auth, mock_api):
        mock_api.return_value = {"id": 7, "login": "rev"}
        mock_get_profile.return_value = {
            "profile_name": "prgs.gitea.reviewer",
            "allowed_operations": ["read", "review"],
            "forbidden_operations": ["merge"],
            "audit_label": "rev-audit",
            "token_source_name": "keychain:prgs-reviewer-token",
            "auth_source_type": "keychain",
            "base_url": "https://gitea.prgs.cc",
            "username": "sysadmin",
            "default_owner": "Scaled-Tech-Consulting",
            "profile_path": "prgs.gitea.reviewer",
            "environment": "prgs",
            "service": "gitea",
            "identity": "reviewer",
            "role": "reviewer",
            "execution_profile": "reviewer-profile",
        }
        result = gitea_get_profile(remote="prgs")
        self.assertEqual(result["environment"], "prgs")
        self.assertEqual(result["service"], "gitea")
        self.assertEqual(result["identity"], "reviewer")
        self.assertEqual(result["role"], "reviewer")
        self.assertEqual(result["profile_address"], "prgs.gitea.reviewer")
        self.assertEqual(result["execution_profile"], "reviewer-profile")
        self.assertEqual(result["auth_source_type"], "keychain")
        self.assertEqual(result["forbidden_operations"], ["merge"])



# ---------------------------------------------------------------------------
# PR eligibility checks (read-only) — issue #14
# ---------------------------------------------------------------------------
class TestPrEligibility(unittest.TestCase):

    def _pr(self, author, state="open", sha="abc123", mergeable=True):
        return {
            "user": {"login": author},
            "state": state,
            "head": {"sha": sha},
            "mergeable": mergeable,
        }

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_reviewer_eligible_to_review(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=5, action="review", remote="prgs")
        self.assertTrue(r["eligible"])
        self.assertEqual(r["authenticated_user"], "reviewer-bot")
        self.assertEqual(r["pr_author"], "author-bot")
        self.assertEqual(r["head_sha"], "abc123")

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_self_author_cannot_merge(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertIn("authenticated user is PR author", r["reasons"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_self_author_cannot_approve(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=8, action="approve", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertIn("authenticated user is PR author", r["reasons"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_merge_fails_closed_when_mergeability_unknown(self, _auth, mock_api):
        # Gitea reports mergeable as None/null (not yet computed).
        mock_api.side_effect = [
            {"login": "merger-bot"},
            self._pr("author-bot", mergeable=None),
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertIsNone(r["mergeable"])
        self.assertIn("PR mergeability unknown", r["reasons"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_profile_not_allowed_to_merge(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "author-bot"}, self._pr("someone-else")]
        env = {"GITEA_PROFILE_NAME": "gitea-author",
               "GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertIn("profile is not allowed to merge", r["reasons"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_forbidden_operation_blocks(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review",
               "GITEA_FORBIDDEN_OPERATIONS": "merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertIn("profile forbids 'merge'", r["reasons"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_closed_pr_not_eligible(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot", state="closed")]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=8, action="approve", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertIn("PR is not open (state=closed)", r["reasons"])

    @patch("mcp_server.get_auth_header", return_value=None)
    def test_unknown_identity_fails_closed(self, _auth):
        env = {"GITEA_PROFILE_NAME": "gitea-merger",
               "GITEA_ALLOWED_OPERATIONS": "read,merge"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertIn("authenticated identity could not be determined", r["reasons"])
        self.assertIsNone(r["authenticated_user"])

    def test_unknown_action_rejected(self):
        with patch.dict(os.environ, {}, clear=True):
            r = gitea_check_pr_eligibility(pr_number=8, action="delete", remote="prgs")
        self.assertFalse(r["eligible"])
        self.assertTrue(any("unknown action" in x for x in r["reasons"]))

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_never_exposes_secrets(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review",
               "GITEA_TOKEN": "super-secret-token"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_check_pr_eligibility(pr_number=5, action="review", remote="prgs")
        blob = repr(r).lower()
        for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
            self.assertNotIn(secret, blob)


class TestReviewDecisionValidationGate(unittest.TestCase):
    """Block incidental live review mutations during validation."""

    PR = 203
    SHA = FULL_HEAD_SHA

    def _pr(self, author, sha=SHA):
        return {
            "user": {"login": author},
            "state": "open",
            "head": {"sha": sha},
            "mergeable": True,
        }

    def setUp(self):
        import reviewer_pr_lease

        _init_reviewer_session("prgs")
        self._lease_patch = _install_owned_reviewer_lease(
            self.PR, head_sha=self.SHA,
        )
        self._lease_patch.start()
        self._auth_identity_patch = patch(
            "mcp_server._authenticated_username", return_value="reviewer-bot"
        )
        self._auth_identity_patch.start()
        self.addCleanup(self._auth_identity_patch.stop)
        self.addCleanup(self._lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)
        self._pr_lease_comments_patch = patch(
            "mcp_server._list_pr_lease_comments", return_value=[]
        )
        self._pr_lease_comments_patch.start()
        self._pr_work_lease_patch = patch(
            "mcp_server._pr_work_lease_reviewer_block",
            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
        )
        self._pr_work_lease_patch.start()
        self.addCleanup(self._pr_lease_comments_patch.stop)
        self.addCleanup(self._pr_work_lease_patch.stop)


    def _env(self):
        return patch.dict(os.environ, {
            "GITEA_PROFILE_NAME": "gitea-reviewer",
            "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes",
        }, clear=True)

    def _assert_no_mutation(self, mock_api):
        for c in mock_api.call_args_list:
            method, url = c.args[0], c.args[1]
            self.assertFalse(
                method == "POST" and url.endswith("/reviews"),
                f"unexpected review mutation: {method} {url}",
            )

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_approve_during_validation_blocked(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
        ]
        with self._env():
            r = gitea_submit_pr_review(
                pr_number=self.PR, action="approve", remote="prgs",
                final_review_decision_ready=False,
            )
        self.assertFalse(r["performed"])
        self.assertTrue(any(
            "final_review_decision_ready must be true" in x for x in r["reasons"]
        ))
        self._assert_no_mutation(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_request_changes_during_validation_blocked(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
        ]
        with self._env():
            r = gitea_submit_pr_review(
                pr_number=self.PR, action="request_changes", remote="prgs",
                final_review_decision_ready=False,
            )
        self.assertFalse(r["performed"])
        self.assertTrue(any(
            "final_review_decision_ready must be true" in x for x in r["reasons"]
        ))
        self._assert_no_mutation(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_duplicate_terminal_decision_blocked(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 1, "state": "APPROVED"},
            [_formal_review("reviewer-bot", "APPROVED", review_id=1)],
            {"login": "reviewer-bot"}, self._pr("author-bot"),
        ]
        gitea_mark_final_review_decision(
            self.PR, "approve", expected_head_sha=self.SHA, remote="prgs")
        with self._env():
            first = gitea_submit_pr_review(
                pr_number=self.PR, action="approve", remote="prgs",
                final_review_decision_ready=True,
            )
            second = gitea_submit_pr_review(
                pr_number=self.PR, action="request_changes", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertTrue(first["performed"])
        self.assertFalse(second["performed"])
        self.assertTrue(any(
            "live review mutation already recorded" in x for x in second["reasons"]
        ))

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_dry_run_proves_submission_without_live_mutation(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
        ]
        with self._env():
            r = gitea_dry_run_pr_review(
                pr_number=self.PR, action="approve", remote="prgs")
        self.assertFalse(r["performed"])
        self.assertTrue(r["would_perform"])
        self.assertTrue(r["dry_run"])
        self._assert_no_mutation(mock_api)


class TestSubmitPrReview(unittest.TestCase):
    """Gated review-mutation tool (#15)."""

    def setUp(self):
        import reviewer_pr_lease

        _init_reviewer_session("prgs")
        self._lease_patch = _install_owned_reviewer_lease(8)
        self._lease_patch.start()
        self._auth_identity_patch = patch(
            "mcp_server._authenticated_username", return_value="reviewer-bot"
        )
        self._auth_identity_patch.start()
        self._pr_lease_comments_patch = patch(
            "mcp_server._list_pr_lease_comments", return_value=[]
        )
        self._pr_lease_comments_patch.start()
        self._pr_work_lease_patch = patch(
            "mcp_server._pr_work_lease_reviewer_block",
            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
        )
        self._pr_work_lease_patch.start()
        gitea_mark_final_review_decision(
            8, "approve", remote="prgs", expected_head_sha="abc123",
        )
        self.addCleanup(self._auth_identity_patch.stop)
        self.addCleanup(self._lease_patch.stop)
        self.addCleanup(self._pr_lease_comments_patch.stop)
        self.addCleanup(self._pr_work_lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)

    def _pr(self, author, state="open", sha="abc123", mergeable=True):
        return {
            "user": {"login": author},
            "state": state,
            "head": {"sha": sha},
            "mergeable": mergeable,
        }

    def _methods(self, mock_api):
        return [c.args[0] for c in mock_api.call_args_list]

    def _assert_no_mutation(self, mock_api):
        # A review mutation is POST .../reviews; eligibility only ever GETs.
        for c in mock_api.call_args_list:
            method, url = c.args[0], c.args[1]
            self.assertFalse(
                method == "POST" and url.endswith("/reviews"),
                f"unexpected review mutation: {method} {url}",
            )

    # -- approve --------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_approve_blocked_when_author(self, _auth, mock_api):
        mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertFalse(r["performed"])
        self.assertIn("authenticated user is PR author", r["reasons"])
        self._assert_no_mutation(mock_api)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_approve_succeeds_when_eligible(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 7, "state": "APPROVED"},
            [_formal_review("reviewer-bot", "APPROVED", review_id=7)],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", body="LGTM", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertTrue(r["performed"])
        self.assertTrue(r.get("review_verdict_visible"))
        self.assertEqual(r["authenticated_user"], "reviewer-bot")
        self.assertEqual(r["pr_author"], "author-bot")
        self.assertEqual(r["head_sha"], "abc123")
        post_calls = [
            c for c in mock_api.call_args_list
            if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
        ]
        self.assertEqual(len(post_calls), 1)
        payload = post_calls[0].args[3]
        self.assertEqual(payload["event"], "APPROVED")
        self.assertEqual(payload["commit_id"], "abc123")

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_approve_fails_when_verdict_stays_pending(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 7, "state": "PENDING"},
            {"id": 7, "state": "PENDING"},
            [_formal_review("reviewer-bot", "PENDING", review_id=7)],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", body="LGTM", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertFalse(r["performed"])
        self.assertFalse(r.get("review_verdict_visible"))
        self.assertTrue(any("expected visible 'APPROVED'" in x for x in r["reasons"]))

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_approve_submits_pending_draft_when_api_returns_pending(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 7, "state": "PENDING"},
            {"id": 7, "state": "APPROVED"},
            [_formal_review("reviewer-bot", "APPROVED", review_id=7)],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", body="LGTM", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertTrue(r["performed"])
        submit_calls = [
            c for c in mock_api.call_args_list
            if c.args[0] == "POST" and "/reviews/7" in c.args[1]
        ]
        self.assertEqual(len(submit_calls), 1)
        self.assertEqual(submit_calls[0].args[3]["event"], "APPROVED")

    # -- request_changes ------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
        _mark_request_changes_ready(remote="prgs")
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 9, "state": "REQUEST_CHANGES"},
            [_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=9)],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="request_changes",
                body="needs work", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertTrue(r["performed"])
        post_calls = [
            c for c in mock_api.call_args_list
            if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
        ]
        self.assertEqual(len(post_calls), 1)
        self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")

    def test_request_changes_blocked_without_eligibility(self):
        _mark_request_changes_ready(remote="prgs")
        with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
                patch("mcp_server.api_request") as mock_api:
            mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
            env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                   "GITEA_ALLOWED_OPERATIONS": "read,review"}  # no request_changes
            with patch.dict(os.environ, env, clear=False):
                r = gitea_submit_pr_review(
                    pr_number=8, action="request_changes", remote="prgs",
                    final_review_decision_ready=True,
                )
            self.assertFalse(r["performed"])
            self.assertIn("profile is not allowed to request_changes", r["reasons"])
            self._assert_no_mutation(mock_api)

    # -- comment --------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
        ]
        gitea_mark_final_review_decision(8, "comment", remote="prgs", expected_head_sha="abc123")
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="comment", body="finding", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertTrue(r["performed"])
        self.assertEqual(mock_api.call_args.args[3]["event"], "COMMENT")

    def test_comment_by_author_allowed(self):
        # Commenting on your own PR is fine — only approve is self-blocked.
        with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
                patch("mcp_server.api_request") as mock_api:
            mock_api.side_effect = [
                {"login": "jcwalker3"}, self._pr("jcwalker3"),
                {"login": "jcwalker3"}, self._pr("jcwalker3"), {"id": 4},
            ]
            env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                   "GITEA_ALLOWED_OPERATIONS": "read,review"}
            with patch.dict(os.environ, env, clear=False):
                gitea_mark_final_review_decision(
                    8, "comment", remote="prgs", expected_head_sha="abc123",
                )
                r = gitea_submit_pr_review(
                    pr_number=8, action="comment", body="note", remote="prgs",
                    final_review_decision_ready=True,
                )
            self.assertTrue(r["performed"])

    # -- identity / profile fail-closed ---------------------------------------

    @patch("mcp_server.get_auth_header", return_value=None)
    def test_unknown_identity_blocks(self, _auth):
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertFalse(r["performed"])
        self.assertIsNone(r["authenticated_user"])
        self.assertIn("authenticated identity could not be determined", r["reasons"])

    def test_disallowed_profile_operation_blocks(self):
        with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
                patch("mcp_server.api_request") as mock_api:
            mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
            env = {"GITEA_PROFILE_NAME": "gitea-author",
                   "GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
            with patch.dict(os.environ, env, clear=False):
                r = gitea_submit_pr_review(
                    pr_number=8, action="approve", remote="prgs",
                    final_review_decision_ready=True,
                )
            self.assertFalse(r["performed"])
            self.assertIn("profile is not allowed to approve", r["reasons"])
            self._assert_no_mutation(mock_api)

    # -- head SHA guard -------------------------------------------------------

    def test_head_sha_mismatch_blocks(self):
        with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
                patch("mcp_server.api_request") as mock_api:
            mock_api.side_effect = [
                {"login": "reviewer-bot"}, self._pr("author-bot", sha="deadbeef"),
            ]
            env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                   "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
            with patch.dict(os.environ, env, clear=False):
                r = gitea_submit_pr_review(
                    pr_number=8, action="approve",
                    expected_head_sha="abc123", remote="prgs",
                    final_review_decision_ready=True,
                )
            self.assertFalse(r["performed"])
            self.assertTrue(any(
                "expected head SHA does not match current PR head (fail closed)" in reason
                or "PR head changed during lease" in reason
                for reason in r["reasons"]
            ))
            self._assert_no_mutation(mock_api)

    def test_head_sha_match_allows(self):
        with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
                patch("mcp_server.api_request") as mock_api:
            mock_api.side_effect = [
                {"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
                {"id": 5, "state": "APPROVED"},
                [_formal_review("reviewer-bot", "APPROVED", review_id=5)],
            ]
            env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                   "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
            with patch.dict(os.environ, env, clear=False):
                r = gitea_submit_pr_review(
                    pr_number=8, action="approve",
                    expected_head_sha="abc123", remote="prgs",
                    final_review_decision_ready=True,
                )
            self.assertTrue(r["performed"])

    # -- invalid action -------------------------------------------------------

    @patch("mcp_server.api_request")
    def test_invalid_review_action_rejected(self, mock_api):
        with patch.dict(os.environ, {}, clear=True):
            r = gitea_submit_pr_review(pr_number=8, action="delete", remote="prgs")
        self.assertFalse(r["performed"])
        self.assertTrue(any("unknown review action" in x for x in r["reasons"]))
        mock_api.assert_not_called()  # never touches the API on a bad action

    # -- redaction ------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_output_redacts_secrets(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
               "GITEA_TOKEN": "super-secret-token"}
        with patch.dict(os.environ, env, clear=False):
            gitea_mark_final_review_decision(5, "approve", remote="prgs", expected_head_sha="abc123")
            r = gitea_submit_pr_review(
                pr_number=5, action="approve", remote="prgs",
                final_review_decision_ready=True,
            )
        blob = repr(r).lower()
        for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
            self.assertNotIn(secret, blob)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_error_message_redacts_credential(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"},
            self._pr("author-bot"),
            RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertFalse(r["performed"])
        blob = repr(r)
        self.assertIn("[REDACTED]", blob)
        self.assertNotIn("abc-secret-xyz", blob)

    def test_authorize_review_correction_success(self):
        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
        lock = _load_review_decision_lock() or {}
        lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
        _save_review_decision_lock(lock)
        with patch("mcp_server.api_request", return_value={"id": 9001}):
            r = gitea_authorize_review_correction(
                prior_review_id=42,
                prior_review_state="approve",
                reason="typo",
                operator_authorized=True,
                target_pr_number=8,
            )
        self.assertTrue(r["authorized"])
        lock = _load_review_decision_lock()
        self.assertTrue(lock["correction_authorized"])
        self.assertEqual(lock["correction_pr_number"], 8)

    def test_authorize_review_correction_mismatch(self):
        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
        lock = _load_review_decision_lock() or {}
        lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
        _save_review_decision_lock(lock)

        # Mismatched ID
        r = gitea_authorize_review_correction(
            prior_review_id=99,
            prior_review_state="approve",
            reason="typo",
            operator_authorized=True,
            target_pr_number=8,
        )
        self.assertFalse(r["authorized"])
        self.assertTrue(any("prior review ID" in x for x in r["reasons"]))

        # Mismatched State
        r = gitea_authorize_review_correction(
            prior_review_id=42,
            prior_review_state="request_changes",
            reason="typo",
            operator_authorized=True,
            target_pr_number=8,
        )
        self.assertFalse(r["authorized"])
        self.assertTrue(any("prior review state" in x for x in r["reasons"]))

        # Cross-PR unlock refused (#693)
        r = gitea_authorize_review_correction(
            prior_review_id=42,
            prior_review_state="approve",
            reason="unlock other pr",
            operator_authorized=True,
            target_pr_number=692,
        )
        self.assertFalse(r["authorized"])
        self.assertTrue(any("different PR" in x for x in r["reasons"]))

    def test_authorize_review_correction_requires_operator_auth(self):
        from mcp_server import _load_review_decision_lock, _save_review_decision_lock
        lock = _load_review_decision_lock() or {}
        lock["live_mutations"] = [
            {"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}
        ]
        _save_review_decision_lock(lock)
        r = gitea_authorize_review_correction(
            prior_review_id=42,
            prior_review_state="approve",
            reason="typo",
            operator_authorized=False,
            target_pr_number=8,
        )
        self.assertFalse(r["authorized"])
        self.assertTrue(any("operator authorization" in x for x in r["reasons"]))

    def test_spoofed_tmp_lock_file_does_not_bypass_gate(self):
        import json
        import mcp_server
        mcp_server._REVIEW_DECISION_LOCK = None
        spoof_path = "/tmp/gitea_review_decision.lock"
        with open(spoof_path, "w", encoding="utf-8") as fh:
            json.dump({"final_review_decision_ready": True}, fh)
        try:
            env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                   "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
            with patch.dict(os.environ, env, clear=False):
                r = gitea_submit_pr_review(
                    pr_number=8, action="approve", remote="prgs",
                    final_review_decision_ready=True,
                )
            self.assertFalse(r["performed"])
            self.assertTrue(any("review decision lock missing" in x for x in r["reasons"]))
        finally:
            if os.path.exists(spoof_path):
                os.remove(spoof_path)

    def test_mark_final_decision_rejects_remote_mismatch(self):
        _init_reviewer_session("prgs")
        r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools", expected_head_sha="abc123")
        self.assertFalse(r["marked_ready"])
        self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_correction_flow_allows_second_terminal_review(self, _auth, mock_api):
        mock_api.side_effect = [
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 42, "state": "APPROVED"},
            [_formal_review("reviewer-bot", "APPROVED", review_id=42)],
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"login": "reviewer-bot"}, self._pr("author-bot"),
            {"id": 43, "state": "REQUEST_CHANGES"},
            [_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
        ]
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
        with patch("mcp_server.gitea_get_pr_review_feedback",
                   return_value=dict(_NO_BLOCKER_FEEDBACK)):
            with patch.dict(os.environ, env, clear=False):
                first = gitea_submit_pr_review(
                    pr_number=8, action="approve", remote="prgs",
                    final_review_decision_ready=True,
                )
                auth = gitea_authorize_review_correction(
                    prior_review_id=42,
                    prior_review_state="approve",
                    reason="operator approved correcting mistaken approve",
                    operator_authorized=True,
                    target_pr_number=8,
                )
                _mark_request_changes_ready(remote="prgs")
                second = gitea_submit_pr_review(
                    pr_number=8, action="request_changes", remote="prgs",
                    final_review_decision_ready=True,
                )
        self.assertTrue(first["performed"])
        self.assertTrue(auth["authorized"])
        self.assertTrue(second["performed"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_remote_org_repo_validation_mismatch(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
        with patch.dict(os.environ, env, clear=False):
            # Mark decision for specific remote, org, repo
            gitea_mark_final_review_decision(8, "approve", remote="prgs", expected_head_sha="abc123", org="MyOrg", repo="MyRepo")

            # Mismatched remote
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", remote="dadeschools", org="MyOrg", repo="MyRepo",
                final_review_decision_ready=True,
            )
            self.assertFalse(r["performed"])
            self.assertTrue(any("ready remote" in x for x in r["reasons"]))

            # Mismatched org
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", remote="prgs", org="OtherOrg", repo="MyRepo",
                final_review_decision_ready=True,
            )
            self.assertFalse(r["performed"])
            self.assertTrue(any("ready org" in x for x in r["reasons"]))

            # Mismatched repo
            r = gitea_submit_pr_review(
                pr_number=8, action="approve", remote="prgs", org="MyOrg", repo="OtherRepo",
                final_review_decision_ready=True,
            )
            self.assertFalse(r["performed"])
            self.assertTrue(any("ready repo" in x for x in r["reasons"]))


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

# ---------------------------------------------------------------------------
# Tracker Hygiene Cleanup Tests
# ---------------------------------------------------------------------------
class TestTrackerHygieneCleanup(unittest.TestCase):

    def setUp(self):
        self._mut_env = patch.dict(
            os.environ,
            shared_mutation_env("test-author-dadeschools"),
            clear=False,
        )
        self._mut_env.start()
        mcp_server.gitea_load_review_workflow()
        self.mock_api = patch("mcp_server.api_request").start()
        self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        patch("gitea_audit.audit_enabled", return_value=True).start()
        self.mock_audit = patch("gitea_audit.write_event").start()
        # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
        patch("mcp_server.get_profile", return_value={
            "profile_name": "test-author-dadeschools",
            "allowed_operations": [
                "read", "merge", "edit", "close",
                "gitea.pr.close", "gitea.issue.close", "gitea.read",
            ],
            "audit_label": "test",
            "forbidden_operations": [],
            "allowed_repositories": ["913443/eAgenda"],
            "base_url": "https://gitea.dadeschools.net",
        }).start()
        patch("mcp_server._list_pr_lease_comments", return_value=[]).start()
        patch(
            "mcp_server._pr_work_lease_reviewer_block",
            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
        ).start()

    def tearDown(self):
        try:
            self._mut_env.stop()
        except Exception:
            pass

        patch.stopall()

    def test_close_issue_removes_in_progress(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "issues/1" in url:
                return {"state": "closed"}
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
            if method == "GET" and "issues/1" in url:
                return {"labels": [{"name": "status:in-progress"}, {"name": "bug"}]}
            if method == "DELETE" and url.endswith("/issues/1/labels/1"):
                return {}
            if method == "PUT" and "labels" in url:
                self.fail("Should not replace the issue label set")
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_close_issue(issue_number=1)
        self.assertTrue(res["success"])
        self.assertEqual(res["cleanup_status"].get(1), "released")
        self.mock_audit.assert_called()

    def test_close_issue_no_label_is_noop(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "issues/1" in url:
                return {"state": "closed"}
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
            if method == "GET" and "issues/1" in url:
                return {"labels": [{"name": "bug"}]}
            if method == "DELETE" and "labels" in url:
                self.fail("Should not DELETE labels")
            if method == "PUT" and "labels" in url:
                self.fail("Should not replace the issue label set")
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_close_issue(issue_number=1)
        self.assertTrue(res["success"])
        self.assertEqual(res["cleanup_status"].get(1), "not present")

    def test_close_issue_blocks_unproven_expected_preexisting_failure(self):
        # #529: a closure report calling a non-zero suite exit an "expected
        # pre-existing failure" without pre-merge proof must fail closed and
        # never PATCH the issue state.
        patched = []

        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH":
                patched.append(url)
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_close_issue(
            issue_number=1,
            closure_report=(
                "Full suite: 1 failed. Expected pre-existing failure, safe to close."
            ),
        )
        self.assertFalse(res["success"])
        self.assertTrue(res.get("blocked"))
        self.assertTrue(res.get("reasons"))
        self.assertFalse(
            patched, "must not PATCH issue state when the closure gate blocks"
        )

    def test_close_issue_allows_proven_baseline_closure(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "issues/1" in url:
                return {"state": "closed"}
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "bug", "id": 2}]
            if method == "GET" and "issues/1" in url:
                return {"labels": [{"name": "bug"}]}
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_close_issue(
            issue_number=1,
            closure_report=(
                "Full suite: 1 failed, expected pre-existing baseline failure "
                "proven on the PR pre-merge base commit.\n"
                "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
                "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
                "Command: venv/bin/python -m pytest -q\n"
                "Exit status: 1\n"
                "Failure signature: AssertionError: None is not true\n"
            ),
        )
        self.assertTrue(res["success"])

    def test_merge_pr_with_closes_removes_label(self):
        import reviewer_pr_lease

        head_sha = FULL_HEAD_SHA
        lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
        lease_patch.start()
        self.addCleanup(lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)

        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/user" in url:
                return {"login": "merger"}
            if method == "GET" and url.endswith("/reviews"):
                return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
            if method == "GET" and "pulls/1" in url and "/files" not in url:
                return {
                    "user": {"login": "author"},
                    "state": "open",
                    "head": {"sha": head_sha, "ref": "feat/my-branch"},
                    "base": {"ref": "main"},
                    "mergeable": True,
                    "merged_commit_sha": "merge123",
                    "title": "My PR",
                    "body": "Closes #123"
                }
            if method == "POST" and "merge" in url:
                return {}
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
            if method == "GET" and "issues/123" in url:
                return {"labels": [{"name": "status:in-progress"}, {"name": "bug"}]}
            if method == "DELETE" and url.endswith("/issues/123/labels/1"):
                return {}
            if method == "PUT" and "labels" in url:
                self.fail("Should not replace the issue label set")
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_merge_pr(
            pr_number=1, confirmation="MERGE PR 1", do="merge",
            expected_head_sha=head_sha,
        )
        self.assertTrue(res["performed"])
        self.assertEqual(res["cleanup_status"].get(123), "released")

    def test_merge_pr_with_branch_name_removes_label(self):
        import reviewer_pr_lease

        head_sha = FULL_HEAD_SHA
        lease_patch = _install_owned_reviewer_lease(1, head_sha=head_sha)
        lease_patch.start()
        self.addCleanup(lease_patch.stop)
        self.addCleanup(reviewer_pr_lease.clear_session_lease)

        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/user" in url:
                return {"login": "merger"}
            if method == "GET" and url.endswith("/reviews"):
                return [_formal_review("reviewer", "APPROVED", sha=head_sha)]
            if method == "GET" and "pulls/1" in url and "/files" not in url:
                return {
                    "user": {"login": "author"},
                    "state": "open",
                    "head": {"sha": head_sha, "ref": "fix/issue-123-slug"},
                    "base": {"ref": "main"},
                    "mergeable": True,
                    "merged_commit_sha": "merge123",
                    "title": "My PR",
                    "body": "Fixing things"
                }
            if method == "POST" and "merge" in url:
                return {}
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "status:in-progress", "id": 1}, {"name": "bug", "id": 2}]
            if method == "GET" and "issues/123" in url:
                return {"labels": [{"name": "status:in-progress"}, {"name": "bug"}]}
            if method == "DELETE" and url.endswith("/issues/123/labels/1"):
                return {}
            if method == "PUT" and "labels" in url:
                self.fail("Should not replace the issue label set")
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_merge_pr(
            pr_number=1, confirmation="MERGE PR 1", do="merge",
            expected_head_sha=head_sha,
        )
        self.assertTrue(res["performed"])
        self.assertEqual(res["cleanup_status"].get(123), "released")

    def test_close_pr_removes_label_but_does_not_close_issue(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "pulls/1" in url:
                return {
                    "number": 1,
                    "title": "My PR",
                    "state": "closed",
                    "html_url": "url",
                    "body": "Closes #123",
                    "head": {"ref": "feat/my-branch"}
                }
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "status:in-progress", "id": 1}]
            if method == "GET" and "issues/123" in url:
                return {"labels": [{"name": "status:in-progress"}]}
            if method == "DELETE" and url.endswith("/issues/123/labels/1"):
                return {}
            if method == "PUT" and "labels" in url:
                self.fail("Should not replace the issue label set")
            if method == "POST" and "comments" in url:
                return {}
            return {}
        self.mock_api.side_effect = api_side_effect

        with patch.dict(os.environ, {}, clear=True):
            res = gitea_edit_pr(pr_number=1, state="closed")
        self.assertTrue(res["success"])
        self.assertEqual(res["cleanup_status"].get(123), "released")
        self.assertNotIn("url", res)

    def test_edit_pr_reveal_opt_in_includes_url(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "pulls/1" in url:
                return {
                    "number": 1,
                    "title": "My PR",
                    "state": "open",
                    "html_url": "http://url1",
                    "body": "No issue link",
                    "head": {"ref": "feat/my-branch"}
                }
            return {}
        self.mock_api.side_effect = api_side_effect

        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
            res = gitea_edit_pr(pr_number=1, title="Updated")
        self.assertTrue(res["success"])
        self.assertEqual(res["url"], "http://url1")

    def test_multiple_linked_issues(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "pulls/1" in url:
                return {
                    "number": 1,
                    "title": "My PR",
                    "state": "closed",
                    "html_url": "url",
                    "body": "Closes #123\nFixes #124",
                    "head": {"ref": "issue-125"}
                }
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "status:in-progress", "id": 1}]
            if method == "GET" and "issues/123" in url:
                return {"labels": [{"name": "status:in-progress"}]}
            if method == "GET" and "issues/124" in url:
                return {"labels": [{"name": "status:in-progress"}]}
            if method == "GET" and "issues/125" in url:
                return {"labels": []}
            if method == "DELETE" and url.endswith("/issues/123/labels/1"):
                return {}
            if method == "DELETE" and url.endswith("/issues/124/labels/1"):
                return {}
            if method == "PUT" and "labels" in url:
                self.fail("Should not replace the issue label set")
            if method == "POST" and "comments" in url:
                return {}
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_edit_pr(pr_number=1, state="closed")
        self.assertTrue(res["success"])
        self.assertEqual(res["cleanup_status"].get(123), "released")
        self.assertEqual(res["cleanup_status"].get(124), "released")
        self.assertEqual(res["cleanup_status"].get(125), "not present")

    def test_no_linked_issue_found(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "pulls/1" in url:
                return {
                    "number": 1,
                    "title": "My PR",
                    "state": "closed",
                    "html_url": "url",
                    "body": "No issue link",
                    "head": {"ref": "main"}
                }
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_edit_pr(pr_number=1, state="closed")
        self.assertTrue(res["success"])
        self.assertEqual(res["cleanup_status"], "no linked issue found")

    def test_label_removal_failure_reported(self):
        def api_side_effect(method, url, auth, payload=None):
            if method == "PATCH" and "issues/1" in url:
                return {"state": "closed"}
            if method == "GET" and "labels" in url and "issues" not in url:
                return [{"name": "status:in-progress", "id": 1}]
            if method == "GET" and "issues/1" in url:
                return {"labels": [{"name": "status:in-progress"}]}
            if method == "DELETE" and url.endswith("/issues/1/labels/1"):
                raise RuntimeError("API failure")
            if method == "PUT" and "labels" in url:
                self.fail("Should not replace the issue label set")
            return {}
        self.mock_api.side_effect = api_side_effect

        res = gitea_close_issue(issue_number=1)
        self.assertTrue(res["success"])
        self.assertIn("error:", res["cleanup_status"].get(1))

    def test_extract_linked_issue_numbers_hygiene(self):
        from mcp_server import extract_linked_issue_numbers
        # Standard closing keywords
        self.assertEqual(extract_linked_issue_numbers("Closes #123"), [123])
        self.assertEqual(extract_linked_issue_numbers("Fixes #123"), [123])
        self.assertEqual(extract_linked_issue_numbers("Resolves #123"), [123])

        # New implements/implemented keywords
        self.assertEqual(extract_linked_issue_numbers("Implements #123"), [123])
        self.assertEqual(extract_linked_issue_numbers("implemented #123"), [123])
        self.assertEqual(extract_linked_issue_numbers("implement #123"), [123])

        # refs / ref should NOT match
        self.assertEqual(extract_linked_issue_numbers("Refs #123"), [])
        self.assertEqual(extract_linked_issue_numbers("ref #123"), [])

        # branch name fallback
        self.assertEqual(extract_linked_issue_numbers("", branch_name="issue-123"), [123])
        self.assertEqual(extract_linked_issue_numbers("", branch_name="feat/issue-123-foo"), [123])


# ---------------------------------------------------------------------------
# Endpoint/keychain redaction in LLM-facing output — issue #120
# ---------------------------------------------------------------------------
class TestEndpointRedaction(unittest.TestCase):
    """Normal MCP output hides endpoint URLs and keychain ids; the admin
    opt-in (GITEA_MCP_REVEAL_ENDPOINTS) restores them for local diagnostics
    without ever revealing token values."""

    def _contexts_config_file(self):
        import tempfile
        config = {
            "version": 2,
            "contexts": {
                "prgs": {
                    "enabled": True,
                    "gitea": {"enabled": True, "kind": "gitea",
                              "base_url": "https://gitea.prgs.cc"},
                    "services": {
                        "jenkins": {
                            "enabled": True, "kind": "jenkins",
                            "label": "PRGS Jenkins",
                            "base_url": "https://jenkins.prgs.cc",
                            "auth": {"type": "keychain",
                                     "id": "prgs-jenkins-token"},
                            "capabilities": ["read"],
                        },
                        "sentry": {
                            "enabled": False, "kind": "sentry",
                            "label": "PRGS Sentry", "base_url": "",
                            "auth": {"type": "keychain",
                                     "id": "prgs-sentry-token"},
                            "capabilities": ["read"],
                        },
                    },
                },
            },
            "profiles": {
                "prgs-author": {
                    "enabled": True, "context": "prgs", "role": "author",
                    "username": "jcwalker3",
                    "base_url": "https://gitea.prgs.cc",
                    "auth": {"type": "keychain",
                             "id": "prgs-gitea-author-token"},
                    "allowed_operations": ["read"],
                    "forbidden_operations": [],
                },
            },
            "projects": {},
            "rules": {"hide_service_urls_from_llm": True,
                      "hide_keychain_ids_from_llm": True,
                      "mcp_resolves_endpoints": True},
        }
        fd, path = tempfile.mkstemp(suffix=".json")
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            json.dump(config, fh)
        return path

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_whoami_hides_endpoint_url_by_default(self, _auth, mock_api):
        mock_api.return_value = {"id": 1, "login": "someone"}
        with patch.dict(os.environ, {}, clear=True):
            result = gitea_whoami(remote="prgs")
        self.assertNotIn("server", result)
        self.assertNotIn("https://", repr(result))

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_whoami_reveals_endpoint_with_admin_optin(self, _auth, mock_api):
        mock_api.return_value = {"id": 1, "login": "someone"}
        with patch.dict(os.environ,
                        {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
            result = gitea_whoami(remote="prgs")
        self.assertEqual(result["server"], "https://gitea.prgs.cc")

    def test_get_profile_hides_url_and_token_source_by_default(self):
        env = {
            "GITEA_PROFILE_NAME": "gitea-author",
            "GITEA_BASE_URL": "https://gitea.example.invalid",
            "GITEA_TOKEN_SOURCE": "keychain:some-item-id",
        }
        with patch.dict(os.environ, env, clear=False):
            result = gitea_get_profile(remote="prgs",
                                       resolve_identity=False)
        blob = repr(result)
        for leaked in ("https://", "keychain:", "some-item-id",
                       "base_url", "server", "token_source_name"):
            self.assertNotIn(leaked, blob)
        self.assertEqual(result["auth_status"], "configured")

    def test_get_profile_reports_unconfigured_auth(self):
        with patch.dict(os.environ,
                        {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
            result = gitea_get_profile(remote="prgs",
                                       resolve_identity=False)
        self.assertEqual(result["auth_status"], "unconfigured")

    def test_get_profile_reveals_with_admin_optin(self):
        env = {
            "GITEA_PROFILE_NAME": "gitea-author",
            "GITEA_TOKEN_SOURCE": "keychain:some-item-id",
            "GITEA_MCP_REVEAL_ENDPOINTS": "1",
        }
        with patch.dict(os.environ, env, clear=False):
            result = gitea_get_profile(remote="prgs",
                                       resolve_identity=False)
        self.assertEqual(result["server"], "https://gitea.prgs.cc")
        self.assertEqual(result["token_source_name"], "keychain:some-item-id")

    def test_audit_tool_reports_state_without_urls_or_ids(self):
        from mcp_server import gitea_audit_config
        path = self._contexts_config_file()
        try:
            env = {"GITEA_MCP_CONFIG": path,
                   "GITEA_MCP_PROFILE": "prgs-author"}
            with patch.dict(os.environ, env, clear=False), \
                 patch("gitea_config._keychain_token", return_value="x"):
                result = gitea_audit_config()
        finally:
            os.unlink(path)
        blob = json.dumps(result)
        self.assertIn("PRGS Jenkins: enabled, read-only, authenticated",
                      result["summaries"])
        self.assertIn("PRGS Sentry: disabled", result["summaries"])
        for leaked in ("https://", "http://", "prgs-jenkins-token",
                       "prgs-gitea-author-token", "base_url"):
            self.assertNotIn(leaked, blob)

    def test_audit_tool_without_config_reports_off(self):
        with patch.dict(os.environ, {}, clear=True):
            from mcp_server import gitea_audit_config
            result = gitea_audit_config()
        self.assertFalse(result["configured"])


# ---------------------------------------------------------------------------
# Issue comment tools (#126)
# ---------------------------------------------------------------------------
class TestIssueCommentTools(unittest.TestCase):
    """gitea_list_issue_comments / gitea_create_issue_comment (#126).

    Issue discussion comments are distinct from PR reviews: they hit the
    /issues/{n}/comments endpoint, are gated on gitea.read (list) and
    gitea.issue.comment (create) — never on the gitea.pr.* review/merge
    family — and their normal output is LLM-safe (no endpoint URLs; the
    GITEA_MCP_REVEAL_ENDPOINTS opt-in restores links for local diagnostics).
    """

    AUTHOR_ENV = {
        "GITEA_PROFILE_NAME": "gitea-author",
        "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
    }

    @staticmethod
    def _comment(cid=101, login="alice", body="hello world"):
        return {
            "id": cid,
            "user": {"login": login},
            "body": body,
            "created_at": "2026-07-03T00:00:00Z",
            "updated_at": "2026-07-03T01:00:00Z",
            "html_url": (
                "https://gitea.example.com/o/r/issues/9#issuecomment-%d" % cid
            ),
        }

    # -- list ----------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lists_comments(self, _auth, mock_api):
        mock_api.return_value = [self._comment(101), self._comment(102, "bob")]
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
        self.assertTrue(result["success"])
        self.assertEqual(result["issue_number"], 9)
        self.assertEqual(len(result["comments"]), 2)
        first = result["comments"][0]
        self.assertEqual(first["id"], 101)
        self.assertEqual(first["author"], "alice")
        self.assertEqual(first["body"], "hello world")
        self.assertEqual(first["created_at"], "2026-07-03T00:00:00Z")
        self.assertEqual(first["updated_at"], "2026-07-03T01:00:00Z")
        url = mock_api.call_args[0][1]
        self.assertIn("/issues/9/comments", url)
        self.assertEqual(mock_api.call_args[0][0], "GET")

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_list_output_has_no_urls_by_default(self, _auth, mock_api):
        mock_api.return_value = [self._comment()]
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
        blob = json.dumps(result)
        self.assertNotIn("https://", blob)
        self.assertNotIn("http://", blob)
        self.assertNotIn("url", blob)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_list_reveal_opt_in_includes_url(self, _auth, mock_api):
        mock_api.return_value = [self._comment()]
        env = dict(self.AUTHOR_ENV, GITEA_MCP_REVEAL_ENDPOINTS="1")
        with patch.dict(os.environ, env, clear=False):
            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
        self.assertIn("issuecomment-101", result["comments"][0]["url"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_list_blocked_without_read_permission(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "gitea-writer-only",
               "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment"}
        with patch.dict(os.environ, env, clear=False):
            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
        self.assertFalse(result["success"])
        self.assertTrue(result["reasons"])
        mock_api.assert_not_called()

    def test_list_unknown_remote_raises(self):
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            with self.assertRaises(ValueError):
                gitea_list_issue_comments(issue_number=9, remote="nope")

    # -- create ----------------------------------------------------------------

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_comment(self, _auth, mock_api):
        mock_api.return_value = self._comment(555, "gitea-author", "posted")
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_create_issue_comment(
                issue_number=9, body="posted", remote="prgs")
        self.assertTrue(result["success"])
        self.assertEqual(result["comment_id"], 555)
        self.assertEqual(result["issue_number"], 9)
        self.assertEqual(mock_api.call_args[0][0], "POST")
        url = mock_api.call_args[0][1]
        self.assertIn("/issues/9/comments", url)
        self.assertIn("gitea.prgs.cc", url)
        self.assertIn("Scaled-Tech-Consulting", url)
        self.assertEqual(mock_api.call_args[0][3], {"body": "posted"})

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_output_has_no_urls_by_default(self, _auth, mock_api):
        mock_api.return_value = self._comment(555)
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_create_issue_comment(
                issue_number=9, body="posted", remote="prgs")
        blob = json.dumps(result)
        self.assertNotIn("https://", blob)
        self.assertNotIn("http://", blob)
        self.assertNotIn("url", blob)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_reveal_opt_in_includes_url(self, _auth, mock_api):
        mock_api.return_value = self._comment(555)
        env = dict(self.AUTHOR_ENV, GITEA_MCP_REVEAL_ENDPOINTS="1")
        with patch.dict(os.environ, env, clear=False):
            result = gitea_create_issue_comment(
                issue_number=9, body="posted", remote="prgs")
        self.assertIn("issuecomment-555", result["url"])

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_review_permissions_do_not_grant_issue_comments(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
               "GITEA_ALLOWED_OPERATIONS":
                   "gitea.read,gitea.pr.review,gitea.pr.comment,"
                   "gitea.pr.approve,gitea.pr.merge"}
        with patch.dict(os.environ, env, clear=False):
            result = gitea_create_issue_comment(
                issue_number=9, body="posted", remote="prgs")
        self.assertFalse(result["success"])
        self.assertFalse(result["performed"])
        self.assertTrue(result["reasons"])
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_forbidden_overrides_allowed(self, _auth, mock_api):
        env = {"GITEA_PROFILE_NAME": "gitea-author",
               "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
               "GITEA_FORBIDDEN_OPERATIONS": "gitea.issue.comment"}
        with patch.dict(os.environ, env, clear=False):
            result = gitea_create_issue_comment(
                issue_number=9, body="posted", remote="prgs")
        self.assertFalse(result["success"])
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_tagged_handoff_missing_mutation_ledger_blocked(
        self, _auth, mock_api
    ):
        body = "[CONTROLLER HANDOFF] PR #1 — test\n\nBlockers:\n- none"
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_create_issue_comment(
                issue_number=9, body=body, remote="prgs")
        self.assertFalse(result["success"])
        self.assertFalse(result["performed"])
        self.assertTrue(result["reasons"])
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_empty_body_blocked(self, _auth, mock_api):
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_create_issue_comment(
                issue_number=9, body="   ", remote="prgs")
        self.assertFalse(result["success"])
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_casual_comment_allowed(self, _auth, mock_api):
        mock_api.return_value = self._comment(556, "gitea-author", "casual")
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_create_issue_comment(
                issue_number=9,
                body="Thanks, I will check this.",
                remote="prgs",
            )
        self.assertTrue(result["success"])
        self.assertTrue(result["performed"])
        mock_api.assert_called_once()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_workflow_comment_missing_fields_blocked(self, _auth, mock_api):
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            result = gitea_create_issue_comment(
                issue_number=9,
                body="Blocked, author should fix.",
                remote="prgs",
            )
        self.assertFalse(result["success"])
        self.assertFalse(result["performed"])
        self.assertIn("canonical_comment_validation", result)
        self.assertFalse(result["canonical_comment_validation"]["allowed"])
        self.assertTrue(result["canonical_comment_validation"]["is_workflow_comment"])
        mock_api.assert_not_called()

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_missing_issue_error_is_redacted(self, _auth, mock_api):
        mock_api.side_effect = RuntimeError(
            "Gitea API error 404 for issue (auth was token abc123secret)")
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            with self.assertRaises(RuntimeError) as cm:
                gitea_create_issue_comment(
                    issue_number=99999, body="posted", remote="prgs")
        msg = str(cm.exception)
        self.assertNotIn("abc123secret", msg)
        self.assertIn("404", msg)

    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_never_touches_review_endpoints(self, _auth, mock_api):
        mock_api.return_value = self._comment(555)
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            gitea_create_issue_comment(
                issue_number=9, body="posted", remote="prgs")
        for call in mock_api.call_args_list:
            self.assertNotIn("reviews", call[0][1])
            self.assertNotIn("/pulls/", call[0][1])

    def test_create_unknown_remote_raises(self):
        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
            with self.assertRaises(ValueError):
                gitea_create_issue_comment(
                    issue_number=9, body="x", remote="nope")


# ---------------------------------------------------------------------------
# Issue-comment permission separation (#137 follow-up; #139 groundwork)
# ---------------------------------------------------------------------------
class TestIssueCommentPermissionSeparation(unittest.TestCase):
    """gitea.issue.comment grants exactly issue-thread commenting.

    #138 added the ``issue.comment`` alias and granted it in the example
    configs. These are the forward-direction separation guards: holding the
    permission must never imply issue close, PR review/approve/merge, or
    branch push / repo commit — and the shipped examples must keep granting
    it so the live-config migration gap (#137) cannot silently regress.
    """

    ISSUE_COMMENT_ONLY = ["gitea.issue.comment"]

    def test_issue_comment_does_not_imply_issue_close(self):
        ok, reason = gitea_config.check_operation(
            "gitea.issue.close", self.ISSUE_COMMENT_ONLY)
        self.assertFalse(ok)
        self.assertEqual(reason, "not-allowed")

    def test_issue_comment_does_not_imply_review_or_merge(self):
        for op in ("gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge",
                   "gitea.pr.request_changes"):
            ok, _ = gitea_config.check_operation(op, self.ISSUE_COMMENT_ONLY)
            self.assertFalse(ok, msg=f"{op} must not be implied")

    def test_issue_comment_does_not_imply_authoring_ops(self):
        for op in ("gitea.branch.push", "gitea.repo.commit",
                   "gitea.branch.create", "gitea.pr.create"):
            ok, _ = gitea_config.check_operation(op, self.ISSUE_COMMENT_ONLY)
            self.assertFalse(ok, msg=f"{op} must not be implied")

    def test_pr_comment_does_not_imply_issue_comment(self):
        ok, reason = gitea_config.check_operation(
            "gitea.issue.comment", ["gitea.read", "gitea.pr.comment"])
        self.assertFalse(ok)
        self.assertEqual(reason, "not-allowed")

    def test_pre_137_author_op_set_fails_closed(self):
        # Exactly the live prgs-author op set that blocked the #51 audit
        # comment before the #137/#138 migration.
        env = {"GITEA_PROFILE_NAME": "gitea-author",
               "GITEA_ALLOWED_OPERATIONS":
                   "gitea.branch.create,gitea.branch.push,gitea.pr.comment,"
                   "gitea.pr.create,gitea.read,gitea.repo.commit"}
        with patch.dict(os.environ, env, clear=False):
            result = gitea_create_issue_comment(
                issue_number=51, body="audit findings", remote="prgs")
        self.assertFalse(result["success"])
        self.assertIn("profile is not allowed to gitea.issue.comment",
                      result["reasons"])

    # -- shipped examples must keep granting the op --------------------------

    @staticmethod
    def _example(name):
        path = os.path.join(os.path.dirname(__file__), "..", name)
        with open(path, encoding="utf-8") as fh:
            return json.load(fh)

    def test_v2_example_profiles_grant_issue_comment(self):
        cfg = self._example("gitea-mcp.v2-contexts.example.json")
        for name, profile in cfg["profiles"].items():
            allowed = profile.get("allowed_operations", [])
            ok, _ = gitea_config.check_operation(
                "gitea.issue.comment", allowed,
                profile.get("forbidden_operations", []))
            self.assertTrue(
                ok, msg=f"example profile {name} must grant issue comments")

    def test_v2_example_preserves_role_separation(self):
        cfg = self._example("gitea-mcp.v2-contexts.example.json")
        author = cfg["profiles"]["example-author"]
        reviewer = cfg["profiles"]["example-reviewer"]
        for op in ("gitea.pr.approve", "gitea.pr.merge"):
            ok, _ = gitea_config.check_operation(
                op, author["allowed_operations"],
                author.get("forbidden_operations", []))
            self.assertFalse(ok, msg=f"author example must not allow {op}")
        for op in ("gitea.branch.push", "gitea.repo.commit"):
            ok, _ = gitea_config.check_operation(
                op, reviewer["allowed_operations"],
                reviewer.get("forbidden_operations", []))
            self.assertFalse(ok, msg=f"reviewer example must not allow {op}")

    def test_v1_example_reviewer_grants_issue_comment(self):
        cfg = self._example("gitea-mcp.example.json")
        reviewer = cfg["profiles"]["mdcps-reviewer"]
        ok, _ = gitea_config.check_operation(
            "gitea.issue.comment", reviewer["allowed_operations"],
            reviewer.get("forbidden_operations", []))
        self.assertTrue(ok)


class TestVerifyMutationAuthority(unittest.TestCase):
    """In-process mutation authority (#199, refs #194).

    The authority record lives in mcp_server._MUTATION_AUTHORITY (per
    process, reset between tests by conftest); the CLI side-channel is
    covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
    lock file — nothing here touches /tmp.
    """

    def setUp(self):
        self.patch_profile = patch("mcp_server.get_profile")
        self.mock_profile = self.patch_profile.start()
        self.patch_username = patch("mcp_server._authenticated_username")
        self.mock_username = self.patch_username.start()
        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
        self.mock_username.return_value = "sysadmin"

    def tearDown(self):
        self.patch_profile.stop()
        self.patch_username.stop()

    def _authority(self, **overrides):
        data = {
            "initial_profile": "prgs-reviewer",
            "initial_identity": "sysadmin",
            "current_profile": "prgs-reviewer",
            "current_identity": "sysadmin",
            "remote": "prgs",
            "task": "review_pr",
            "role_pivot_authorized": False,
            "role_pivot_record": None,
            "pid": os.getpid(),
        }
        data.update(overrides)
        mcp_server._MUTATION_AUTHORITY = data

    def test_missing_authority_seeds_from_live_context(self):
        # Approved preflight path (whoami → eligibility → mutation): the
        # first mutation gate seeds the authority instead of failing closed,
        # so the standard reviewer workflow keeps working.
        mcp_server._MUTATION_AUTHORITY = None
        mcp_server.verify_mutation_authority("prgs")
        seeded = mcp_server._MUTATION_AUTHORITY
        self.assertIsNotNone(seeded)
        self.assertEqual(seeded["current_profile"], "prgs-reviewer")
        self.assertEqual(seeded["current_identity"], "sysadmin")
        self.assertEqual(seeded["remote"], "prgs")

    def test_unresolved_profile_fails_closed(self):
        self.mock_profile.return_value = {}
        mcp_server._MUTATION_AUTHORITY = None
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs")
        self.assertIn("profile unresolved", str(ctx.exception))

    def test_mismatched_remote_fails(self):
        self._authority(remote="dadeschools")
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs")
        self.assertIn("does not match locked remote", str(ctx.exception))

    def test_profile_flip_after_record_fails(self):
        # Authority was recorded as author; the active profile now resolves
        # as reviewer (e.g. an env-var flip mid-session) — refuse.
        self._authority(
            initial_profile="prgs-author",
            initial_identity="jcwalker3",
            current_profile="prgs-author",
            current_identity="jcwalker3",
        )
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs")
        self.assertIn("does not match locked authority", str(ctx.exception))

    def test_session_lock_env_mismatch_fails(self):
        # The launching session locked the environment to the author
        # profile; the active profile resolves as reviewer — side-channel
        # override rejected even with a matching in-process authority.
        self._authority()
        with patch("mcp_server.gitea_config.is_runtime_switching_enabled",
                   return_value=False):
            with patch.dict(os.environ,
                            {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
                with self.assertRaises(RuntimeError) as ctx:
                    mcp_server.verify_mutation_authority("prgs")
        self.assertIn("side-channel override rejected", str(ctx.exception))

    def test_foreign_pid_authority_is_not_trusted(self):
        # An authority record from another process (fork leftovers) is
        # discarded and reseeded from the live context, never reused.
        self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
        mcp_server.verify_mutation_authority("prgs")
        self.assertEqual(
            mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
        )
        self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())

    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
        self._authority(
            initial_profile="prgs-author",
            initial_identity="jcwalker3",
        )
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
        self.assertIn("without authorized role pivot", str(ctx.exception))

    def test_authorized_pivot_is_allowed(self):
        self._authority(
            initial_profile="prgs-author",
            initial_identity="jcwalker3",
            role_pivot_authorized=True,
        )
        mcp_server.verify_mutation_authority("prgs", required_role="reviewer")

    def test_allowed_when_match(self):
        self._authority()
        with patch.dict(os.environ,
                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
            mcp_server.verify_mutation_authority("prgs")


def _clean_master_git_state_for_lock():
    return {
        "current_branch": "master",
        "porcelain_status": "",
        "base_equivalent": True,
        "inspected_git_root": "/scratch/wt",
        "base_branch": "origin/master",
    }


class TestIssueLocking(unittest.TestCase):
    """Test issue locking and PR gating constraints."""

    def setUp(self):
        self._lock_dir = tempfile.TemporaryDirectory()
        # Most locking tests target remote=prgs; host-align profile (#714).
        env = {
            **shared_mutation_env("test-author-prgs"),
            "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
        }
        self._env_patcher = patch.dict(os.environ, env, clear=False)
        self._env_patcher.start()
        self._dup_fetcher_patcher = patch(
            "mcp_server.issue_duplicate_context_fetcher",
            return_value=([], [], {"status": "not_claimed"}),
        )
        self.mock_dup_fetcher = self._dup_fetcher_patcher.start()

    def tearDown(self):
        self._dup_fetcher_patcher.stop()
        self._env_patcher.stop()
        self._lock_dir.cleanup()

    def _create_pr_env(self) -> dict:
        # PR-locking tests call create_pr with remote=prgs.
        return {
            **shared_mutation_env("test-author-prgs"),
            "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
        }

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_success(self, _auth, _api, _git_state):
        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
        self.assertTrue(res["success"])
        self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
        self.assertEqual(res["work_lease"]["issue_number"], 196)
        self.assertEqual(res["work_lease"]["branch"], "feat/issue-196-mutations")
        self.assertIn("created_at", res["work_lease"])
        self.assertIn("expires_at", res["work_lease"])
        self.assertIn("last_heartbeat_at", res["work_lease"])
        self.assertIn(res["work_lease"]["claimant"]["profile"], ("gitea-default", "test-author-prgs", "prgs-author", "gitea-author"))
        self.assertIn("lock_file_path", res)
        lock = issue_lock_store.read_lock_file(res["lock_file_path"])
        self.assertIn("worktree_path", lock)
        self.assertIn("work_lease", lock)

    def test_lock_issue_mismatch_branch_fails(self):
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
        self.assertIn("must contain locked issue pattern", str(ctx.exception))

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_reused_by_open_pr_branch(self, _auth, _git_state):
        self.mock_dup_fetcher.return_value = ([{
            "number": 200,
            "head": {"ref": "feat/issue-196-boundary"},
            "title": "Some PR",
            "body": "No closes ref",
        }], [], {"status": "not_claimed"})
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
        self.assertIn("open PR #200 already covers issue", str(ctx.exception))

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, _git_state):
        self.mock_dup_fetcher.return_value = ([{
            "number": 200,
            "head": {"ref": "feat/other-branch"},
            "title": "Some PR",
            "body": "fixes #196",
        }], [], {"status": "not_claimed"})
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
        self.assertIn("open PR #200 already covers issue", str(ctx.exception))

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_reused_by_remote_branch(self, _auth, _git_state):
        self.mock_dup_fetcher.return_value = (
            [],
            ["feat/issue-196-existing-work"],
            {"status": "not_claimed"},
        )
        with self.assertRaises(ValueError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
        self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_adopts_exact_own_branch(self, _auth, mock_api, _git_state):
        branch = "feat/issue-196-mutations"
        self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
        mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
        res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
        self.assertTrue(res["success"])
        self.assertIn("adoption", res)
        self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_adoption_response_has_explicit_proof(self, _auth, mock_api, _git_state):
        # #477 AC1: the live lock response must carry citable adoption proof.
        branch = "feat/issue-196-mutations"
        self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
        mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
        res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
        proof = res["adoption"]
        self.assertEqual(proof["adoption_decision"], "ADOPT")
        self.assertTrue(proof["adopted"])
        self.assertEqual(proof["adopted_branch"], branch)
        self.assertEqual(proof["adopted_branch_head"], "abc123")
        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
        self.assertIn("gitea_create_pr", proof["safe_next_action"])
        self.assertIn("196", proof["matcher_summary"])

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_no_match_response_does_not_claim_adoption(self, _auth, _api, _git_state):
        # #477 AC2/AC3: a normal (NO_MATCH) lock must carry adoption-free proof
        # and must NOT expose an ``adoption`` block.
        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
        self.assertTrue(res["success"])
        self.assertNotIn("adoption", res)
        check = res["adoption_check"]
        self.assertEqual(check["adoption_decision"], "NO_MATCH")
        self.assertFalse(check["adopted"])
        self.assertIsNone(check["adopted_branch"])
        self.assertEqual(check["competing_branch_check"]["result"], "clear")

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_blocks_active_same_operation_lease(self, _auth, _api, _git_state):
        prgs_repo = "Gitea-Tools"  # #714 session-bound repo (not REMOTES Timesheet default)
        issue_lock_store.save_lock_file(
            issue_lock_store.lock_file_path(
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo=prgs_repo,
                issue_number=196,
            ),
            {
                "issue_number": 196,
                "branch_name": "feat/issue-196-other-work",
                "remote": "prgs",
                "org": "Scaled-Tech-Consulting",
                "repo": prgs_repo,
                "worktree_path": "/tmp/other-worktree",
                "work_lease": {
                    "operation_type": "author_issue_work",
                    "expires_at": "2999-01-01T00:00:00Z",
                },
            },
        )
        with self.assertRaises(RuntimeError) as ctx:
            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
        self.assertIn("already has an active author_issue_work lease", str(ctx.exception))

    @patch(
        "mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value=_clean_master_git_state_for_lock(),
    )
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self, _auth, _api, _git_state):
        """#601: expired lease still blocks takeover when owner pid is live AND worktree exists.

        Dead-pid / missing-worktree reclaim is covered by issue_lock_store unit tests.
        This MCP path must remain fail-closed for sticky expired foreign ownership.
        """
        prgs_repo = "Gitea-Tools"  # #714 session-bound repo (not REMOTES Timesheet default)
        # Present worktree + live pid => reclaim_allowed=False even though expires_at is past.
        sticky_worktree = tempfile.mkdtemp(prefix="gitea-sticky-lease-")
        self.addCleanup(lambda: shutil.rmtree(sticky_worktree, ignore_errors=True))
        issue_lock_store.save_lock_file(
            issue_lock_store.lock_file_path(
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo=prgs_repo,
                issue_number=196,
            ),
            {
                "issue_number": 196,
                "branch_name": "feat/issue-196-other-work",
                "remote": "prgs",
                "org": "Scaled-Tech-Consulting",
                "repo": prgs_repo,
                "worktree_path": sticky_worktree,
                "session_pid": os.getpid(),
                "pid": os.getpid(),
                "work_lease": {
                    "operation_type": "author_issue_work",
                    "expires_at": "2000-01-01T00:00:00Z",
                },
            },
        )
        with patch.object(issue_lock_store, "is_process_alive", return_value=True):
            with self.assertRaises(RuntimeError) as ctx:
                gitea_lock_issue(
                    issue_number=196,
                    branch_name="feat/issue-196-mutations",
                    remote="prgs",
                )
        self.assertIn("Recovery review is required before takeover", str(ctx.exception))

    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_from_clean_scratch_worktree(self, _auth, _api):
        scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
        with patch(
            "mcp_server.issue_lock_worktree.read_worktree_git_state",
            return_value=_clean_master_git_state_for_lock(),
        ) as mock_git:
            res = gitea_lock_issue(
                issue_number=249,
                branch_name="feat/issue-249-issue-lock-scratch-worktree",
                remote="prgs",
                worktree_path=scratch,
            )
        self.assertTrue(res["success"])
        self.assertEqual(res["worktree_path"], os.path.realpath(scratch))
        mock_git.assert_called_once_with(os.path.realpath(scratch))

    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_fails_when_declared_worktree_dirty(self, _auth, _api):
        with patch(
            "mcp_server.issue_lock_worktree.read_worktree_git_state",
            return_value={
                "current_branch": "master",
                "porcelain_status": " M gitea_mcp_server.py\n",
                "base_equivalent": True,
            },
        ):
            with self.assertRaises(RuntimeError) as ctx:
                gitea_lock_issue(
                    issue_number=249,
                    branch_name="feat/issue-249-issue-lock-scratch-worktree",
                    remote="prgs",
                    worktree_path="/tmp/scratch/wt",
                )
        self.assertIn("tracked file edits exist before issue lock", str(ctx.exception))

    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_lock_fails_when_declared_worktree_not_on_base(self, _auth, _api):
        with patch(
            "mcp_server.issue_lock_worktree.read_worktree_git_state",
            return_value={
                "current_branch": "feat/issue-243-forbidden-git-gaps",
                "porcelain_status": "",
                "base_equivalent": False,
            },
        ):
            with self.assertRaises(RuntimeError) as ctx:
                gitea_lock_issue(
                    issue_number=249,
                    branch_name="feat/issue-249-issue-lock-scratch-worktree",
                    remote="prgs",
                    worktree_path="/tmp/scratch/wt",
                )
        self.assertIn("base-equivalent", str(ctx.exception))

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_missing_lock_fails(self, _auth, _role):
        with patch.dict(os.environ, self._create_pr_env(), clear=True):
            with self.assertRaises(RuntimeError) as ctx:
                gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
        self.assertIn("Issue lock is missing", str(ctx.exception))

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_branch_mismatch_fails(self, _auth, _role):
        worktree = os.path.realpath(os.getcwd())
        _bind_test_lock(
            issue_number=196,
            branch_name="feat/issue-196-mutations",
            remote="prgs",
            worktree_path=worktree,
        )
        with patch.dict(os.environ, self._create_pr_env(), clear=True):
            with self.assertRaises(ValueError) as ctx:
                gitea_create_pr(
                    title="feat: X Closes #196",
                    head="feat/issue-196-different",
                    remote="prgs",
                    worktree_path=worktree,
                )
        self.assertIn("does not match locked branch", str(ctx.exception))

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_forbidden_terms_fails(self, _auth, _role):
        worktree = os.path.realpath(os.getcwd())
        _bind_test_lock(
            issue_number=196,
            branch_name="feat/issue-196-mutations",
            remote="prgs",
            worktree_path=worktree,
        )
        with patch.dict(os.environ, self._create_pr_env(), clear=True):
            for term in ("equivalent to #196", "related to #196", "same as #196"):
                with self.assertRaises(ValueError) as ctx:
                    gitea_create_pr(
                        title=f"feat: X {term}",
                        head="feat/issue-196-mutations",
                        remote="prgs",
                        worktree_path=worktree,
                    )
                self.assertIn("contains forbidden term", str(ctx.exception))

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
        worktree = os.path.realpath(os.getcwd())
        _bind_test_lock(
            issue_number=196,
            branch_name="feat/issue-196-mutations",
            remote="prgs",
            worktree_path=worktree,
        )
        with patch.dict(os.environ, self._create_pr_env(), clear=True):
            with self.assertRaises(ValueError) as ctx:
                gitea_create_pr(
                    title="feat: X refs #196",
                    head="feat/issue-196-mutations",
                    remote="prgs",
                    worktree_path=worktree,
                )
        self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_worktree_mismatch_fails(self, _auth, _role):
        scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr")
        _bind_test_lock(
            issue_number=249,
            branch_name="feat/issue-249-issue-lock-scratch-worktree",
            worktree_path=scratch,
            remote="prgs",
        )
        with patch.dict(os.environ, self._create_pr_env(), clear=True):
            with self.assertRaises(ValueError) as ctx:
                gitea_create_pr(
                    title="feat: lock scratch worktree Closes #249",
                    head="feat/issue-249-issue-lock-scratch-worktree",
                    remote="prgs",
                    worktree_path="/tmp/other-scratch",
                )
        self.assertIn("does not match locked worktree", str(ctx.exception))

    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
        worktree = os.path.realpath(os.getcwd())
        with tempfile.TemporaryDirectory() as lock_dir:
            env = {**self._create_pr_env(), "GITEA_ISSUE_LOCK_DIR": lock_dir}
            with patch.dict(os.environ, env, clear=False):
                issue_lock_store.save_lock_file(
                    issue_lock_store.lock_file_path(
                        remote="prgs",
                        org="Scaled-Tech-Consulting",
                        repo="Gitea-Tools",  # #714 session-bound
                        issue_number=447,
                        lock_dir=lock_dir,
                    ),
                _sample_issue_lock(
                    issue_number=447,
                    branch_name="feat/issue-447-lock-provenance",
                    remote="prgs",
                    org="Scaled-Tech-Consulting",
                    repo="Gitea-Tools",  # #714 session-bound
                    worktree_path=worktree,
                    lock_provenance=None,
                ),
                )
                with self.assertRaises(RuntimeError) as ctx:
                    gitea_create_pr(
                        title="feat: lock provenance Closes #447",
                        head="feat/issue-447-lock-provenance",
                        remote="prgs",
                        worktree_path=worktree,
                    )
        self.assertIn("lock provenance", str(ctx.exception).lower())

    @patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
    @patch("mcp_server.api_request")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api, _labels):
        scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")

        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/issues/249" in url:
                return _issue_with_labels("type:feature", "status:in-progress")
            if method == "POST" and url.endswith("/pulls"):
                return {"number": 250, "html_url": "https://example/pr/250"}
            if method == "PUT" and url.endswith("/issues/249/labels"):
                return [{"name": "type:feature"}, {"name": "status:pr-open"}]
            return {}

        mock_api.side_effect = api_side_effect
        _bind_test_lock(
            issue_number=249,
            branch_name="feat/issue-249-issue-lock-scratch-worktree",
            worktree_path=scratch,
            remote="prgs",
        )
        with patch.dict(os.environ, self._create_pr_env(), clear=True):
            res = gitea_create_pr(
                title="feat: issue-lock scratch worktree Closes #249",
                head="feat/issue-249-issue-lock-scratch-worktree",
                remote="prgs",
                worktree_path=scratch,
            )
        self.assertEqual(res["number"], 250)
        self.assertTrue(res["issue_status_transition"]["performed"])


# ---------------------------------------------------------------------------
# Pre-flight ordering and workspace edit block (#210)
# ---------------------------------------------------------------------------
class TestPreflightVerification(unittest.TestCase):
    """Assert that pre-flight verification gates order correctly."""

    def setUp(self):
        # Save real global variables
        import mcp_server
        self.orig_whoami_called = mcp_server._preflight_whoami_called
        self.orig_capability_called = mcp_server._preflight_capability_called
        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
        self.orig_capability_violation = mcp_server._preflight_capability_violation
        self.orig_resolved_role = mcp_server._preflight_resolved_role
        self.orig_process_start = mcp_server._process_start_porcelain
        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
        self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
        self.orig_capability_files = mcp_server._preflight_capability_violation_files
        self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files

        # Reset state for each test
        mcp_server._preflight_whoami_called = False
        mcp_server._preflight_capability_called = False
        mcp_server._preflight_whoami_violation = False
        mcp_server._preflight_capability_violation = False
        mcp_server._preflight_resolved_role = None
        mcp_server._process_start_porcelain = ""
        mcp_server._preflight_whoami_baseline_porcelain = None
        mcp_server._preflight_capability_baseline_porcelain = None
        mcp_server._preflight_whoami_violation_files = []
        mcp_server._preflight_capability_violation_files = []
        mcp_server._preflight_reviewer_violation_files = []
        os.environ["GITEA_TEST_PORCELAIN"] = ""

    def tearDown(self):
        # Restore real global variables
        import mcp_server
        mcp_server._preflight_whoami_called = self.orig_whoami_called
        mcp_server._preflight_capability_called = self.orig_capability_called
        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
        mcp_server._preflight_capability_violation = self.orig_capability_violation
        mcp_server._preflight_resolved_role = self.orig_resolved_role
        mcp_server._process_start_porcelain = self.orig_process_start
        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
        mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
        mcp_server._preflight_capability_violation_files = self.orig_capability_files
        mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
        for key in (
            "GITEA_TEST_FORCE_DIRTY",
            "GITEA_TEST_PORCELAIN",
            "GITEA_ACTIVE_WORKTREE",
            "GITEA_AUTHOR_WORKTREE",
        ):
            if key in os.environ:
                del os.environ[key]

    def test_preflight_whoami_violation(self):
        import mcp_server
        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
        mcp_server.record_preflight_check("whoami")
        self.assertTrue(mcp_server._preflight_whoami_violation)
        mcp_server.record_preflight_check("capability", resolved_role="author")

        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_preflight_purity()
        self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
        self.assertIn("Offending files:", str(ctx.exception))

    def test_preflight_capability_violation(self):
        import mcp_server
        mcp_server.record_preflight_check("whoami")
        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
        mcp_server.record_preflight_check("capability", resolved_role="author")
        self.assertTrue(mcp_server._preflight_capability_violation)

        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_preflight_purity()
        self.assertIn(
            "Workspace file edits occurred before gitea_resolve_task_capability verification",
            str(ctx.exception),
        )
        self.assertIn("Offending files:", str(ctx.exception))

    def test_preflight_not_called_fails_closed(self):
        import mcp_server
        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_preflight_purity()
        self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))

        mcp_server._preflight_whoami_called = True
        with self.assertRaises(RuntimeError) as ctx2:
            mcp_server.verify_preflight_purity()
        self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))

    def test_preflight_reviewer_mutation_violation(self):
        import mcp_server
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", resolved_role="reviewer")

        # Session-owned reviewer edits after capability are blocked.
        os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_preflight_purity()
        self.assertIn("forbidden from modifying tracked workspace files", str(ctx.exception))
        self.assertIn("reviewer_edit.py", str(ctx.exception))

        # Foreign pre-existing dirty state does not block when unchanged.
        os.environ["GITEA_TEST_PORCELAIN"] = ""
        mcp_server.verify_preflight_purity()

    def test_foreign_workspace_edits_do_not_block_clean_reviewer(self):
        """#252: concurrent-session dirt in the shared worktree is not attributed."""
        import mcp_server
        mcp_server._process_start_porcelain = " M foreign_author.py\n"
        os.environ["GITEA_TEST_PORCELAIN"] = " M foreign_author.py\n"
        mcp_server.record_preflight_check("whoami")
        self.assertFalse(mcp_server._preflight_whoami_violation)
        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
        self.assertFalse(mcp_server._preflight_capability_violation)
        mcp_server.verify_preflight_purity()

    def test_fresh_whoami_clears_sticky_violation(self):
        """#252: re-running whoami re-evaluates instead of replaying sticky state."""
        import mcp_server
        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
        mcp_server.record_preflight_check("whoami")
        self.assertTrue(mcp_server._preflight_whoami_violation)

        del os.environ["GITEA_TEST_FORCE_DIRTY"]
        os.environ["GITEA_TEST_PORCELAIN"] = ""
        mcp_server.record_preflight_check("whoami")
        self.assertFalse(mcp_server._preflight_whoami_violation)
        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
        mcp_server.verify_preflight_purity()

    def test_runtime_context_matches_preflight_block(self):
        """#252: safe_next_action must not claim ready while pre-flight blocks."""
        import mcp_server
        status = mcp_server.assess_preflight_status()
        self.assertFalse(status["preflight_ready"])
        self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])

    def test_declared_clean_task_worktree_ignores_control_checkout_violation(self):
        """#275: active branches/ worktree is the inspected mutation workspace."""
        import mcp_server
        mcp_server._preflight_whoami_called = True
        mcp_server._preflight_capability_called = True
        mcp_server._preflight_whoami_violation = True
        mcp_server._preflight_whoami_violation_files = ["mcp_server.py"]
        os.environ["GITEA_TEST_PORCELAIN"] = ""

        worktree = "/repo/branches/issue-275-clean"
        status = mcp_server.assess_preflight_status(worktree_path=worktree)
        self.assertTrue(status["preflight_ready"])
        self.assertEqual(status["preflight_block_reasons"], [])
        self.assertIn("preflight_workspace", status)
        mcp_server.verify_preflight_purity(worktree_path=worktree)

    def test_declared_dirty_task_worktree_reports_workspace_diagnostics(self):
        """#275: dirty failures name the inspected task workspace, not just files."""
        import mcp_server
        mcp_server._preflight_whoami_called = True
        mcp_server._preflight_capability_called = True
        os.environ["GITEA_TEST_PORCELAIN"] = " M task_file.py\n"

        worktree = "/repo/branches/issue-275-dirty"
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_preflight_purity(worktree_path=worktree)
        msg = str(ctx.exception)
        self.assertIn("resolved workspace", msg)
        self.assertIn(worktree, msg)
        self.assertIn("worktree_path argument", msg)
        self.assertIn("task_file.py", msg)
        self.assertIn("author namespace", msg)


class TestIssue546Deadlock(unittest.TestCase):
    def setUp(self):
        import reviewer_pr_lease
        import review_workflow_load
        import mcp_server
        reviewer_pr_lease.clear_session_lease()
        review_workflow_load.clear_review_workflow_load()
        mcp_server._preflight_capability_called = False
        mcp_server._preflight_resolved_role = None
        mcp_server._REVIEW_DECISION_LOCK = None

        self.auth_user_patch = patch("mcp_server._authenticated_username", return_value="reviewer-bot")
        self.auth_user_patch.start()
        self.addCleanup(self.auth_user_patch.stop)

        self.verify_purity_patch = patch("mcp_server.verify_preflight_purity", return_value=None)
        self.verify_purity_patch.start()
        self.addCleanup(self.verify_purity_patch.stop)

        self.verify_workspace_patch = patch("mcp_server._verify_role_mutation_workspace", return_value="/workspace")
        self.verify_workspace_patch.start()
        self.addCleanup(self.verify_workspace_patch.stop)

        self.get_profile_patch = patch("mcp_server.get_profile", return_value={
            "profile_name": "prgs-reviewer",
            "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.review", "gitea.pr.approve"],
            "forbidden_operations": [],
        })
        self.get_profile_patch.start()
        self.addCleanup(self.get_profile_patch.stop)

        self.pr_work_lease_patch = patch(
            "mcp_server._pr_work_lease_reviewer_block",
            return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
        )
        self.pr_work_lease_patch.start()
        self.addCleanup(self.pr_work_lease_patch.stop)

        self.auth_header_patch = patch("mcp_server.get_auth_header", return_value="token test")
        self.auth_header_patch.start()
        self.addCleanup(self.auth_header_patch.stop)

    def test_workflow_no_deadlock(self):
        import mcp_server
        import reviewer_pr_lease

        # 1. Resolve capability
        res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
        self.assertTrue(res["allowed_in_current_session"])

        # 2. Load review workflow
        res = mcp_server.gitea_load_review_workflow()
        self.assertTrue(res["success"])

        # Mock API requests for lease comments and PR retrieval
        comments = []
        def mock_api_side(method, url, auth, data=None):
            if "/api/v1/user" in url:
                return {"login": "reviewer-bot"}
            if "/pulls/" in url:
                if "/reviews" in url:
                    if method == "POST":
                        return {"id": 2001, "state": "APPROVED"}
                    return [{"id": 2001, "user": {"login": "reviewer-bot"}, "state": "APPROVED", "commit_id": "abc123"}]
                return {"user": {"login": "author-user"}, "state": "open", "head": {"sha": "abc123"}, "mergeable": True}
            if "/comments" in url:
                if method == "POST":
                    comments.append({"id": 1001, "body": data["body"], "user": {"login": "reviewer-bot"}})
                    return {"id": 1001}
                return comments
            return {}

        with patch("mcp_server.api_request", side_effect=mock_api_side):
            # 3. Acquire reviewer lease
            res = mcp_server.gitea_acquire_reviewer_pr_lease(
                pr_number=550,
                worktree="/workspace",
                candidate_head="abc123",
                remote="prgs",
            )
            self.assertTrue(res["success"])
            self.assertEqual(res["comment_id"], 1001)

            # verify lease is recorded in session
            session_lease = reviewer_pr_lease.get_session_lease()
            self.assertIsNotNone(session_lease)
            self.assertEqual(session_lease["pr_number"], 550)

            # 4. Resolve capability again (simulating subsequent tool call preflight check/token refresh)
            # This should NOT clear the session lease or decision lock!
            mcp_server._preflight_capability_called = False  # simulate clearance from prior mutation
            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertIsNotNone(reviewer_pr_lease.get_session_lease()) # Preserved!

            # 5. Mark final review decision APPROVED
            res = mcp_server.gitea_mark_final_review_decision(
                pr_number=550,
                action="approve",
                expected_head_sha="abc123",
                remote="prgs",
            )
            self.assertTrue(res["marked_ready"])

            # 6. Submit review (should succeed without deadlock)
            res = mcp_server.gitea_submit_pr_review(
                pr_number=550,
                action="approve",
                body="",
                expected_head_sha="abc123",
                final_review_decision_ready=True,
                remote="prgs",
                worktree_path="/workspace",
            )
            self.assertTrue(res["performed"])

    def test_release_reviewer_pr_lease(self):
        import mcp_server
        import reviewer_pr_lease

        # Resolve capability and load workflow
        mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
        mcp_server.gitea_load_review_workflow()

        comments = [_reviewer_lease_comment(550, session_id=_DEFAULT_LEASE_SESSION, head_sha="abc123")]
        posted_comments = []

        def mock_api_side(method, url, auth, data=None):
            if "/api/v1/user" in url:
                return {"login": "reviewer-bot"}
            if "/pulls/" in url:
                return {"user": {"login": "author-user"}, "state": "open", "head": {"sha": "abc123"}, "mergeable": True}
            if "/comments" in url:
                if method == "POST":
                    new_c = {"id": 1002, "body": data["body"], "user": {"login": "reviewer-bot"}}
                    comments.append(new_c)
                    posted_comments.append(new_c)
                    return {"id": 1002}
                return comments
            return {}

        # Set session lease in memory
        import merger_lease_adoption as mla
        reviewer_pr_lease.record_session_lease({
            "pr_number": 550,
            "session_id": _DEFAULT_LEASE_SESSION,
            "candidate_head": "abc123",
            "target_branch": "master",
            "comment_id": 9001,
        }, lease_provenance=mla.build_lease_provenance(source=mla.SOURCE_ACQUIRE, comment_id=9001))

        with patch("mcp_server.api_request", side_effect=mock_api_side):
            # Release lease
            res = mcp_server.gitea_release_reviewer_pr_lease(pr_number=550, remote="prgs")
            self.assertTrue(res["success"])
            self.assertTrue(res["released"])
            self.assertEqual(res["comment_id"], 1002)

            # verify lease is cleared in session
            self.assertIsNone(reviewer_pr_lease.get_session_lease())
            # verify comment phase is released
            self.assertIn("phase: released", posted_comments[0]["body"])

    def test_reviewer_pr_lease_gate_resolves_identity_with_host(self):
        """Lease mutation gate must resolve identity using host, not remote alias."""
        import mcp_server

        resolved_hosts = []

        def tracking_username(host):
            resolved_hosts.append(host)
            return None if host in {"prgs", "dadeschools"} else "reviewer-bot"

        with _install_owned_reviewer_lease(
            550,
            session_id=_DEFAULT_LEASE_SESSION,
            head_sha="abc123",
        ), patch(
            "mcp_server._resolve",
            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
        ), patch(
            "mcp_server._authenticated_username",
            side_effect=tracking_username,
        ):
            reasons = mcp_server._reviewer_pr_lease_gate(
                pr_number=550,
                remote="prgs",
                host=None,
                org=None,
                repo=None,
                mutation="approve",
                live_head_sha="abc123",
                pinned_head_sha="abc123",
            )

        self.assertEqual(reasons, [])
        self.assertIn("gitea.prgs.cc", resolved_hosts)
        self.assertNotIn("prgs", resolved_hosts)

    def test_acquire_reviewer_pr_lease_resolves_identity_with_host(self):
        """Acquire path must not pass a remote alias into identity lookup."""
        import mcp_server

        resolved_hosts = []
        posted = []

        def tracking_username(host):
            resolved_hosts.append(host)
            return None if host in {"prgs", "dadeschools"} else "reviewer-bot"

        def mock_api(method, url, auth, data=None):
            if "/pulls/" in url:
                return {
                    "user": {"login": "author-user"},
                    "state": "open",
                    "head": {"sha": "abc123"},
                    "mergeable": True,
                }
            if "/comments" in url:
                if method == "POST":
                    posted.append(data["body"])
                    return {"id": 1003}
                return []
            return {}

        with patch("mcp_server.api_request", side_effect=mock_api), patch(
            "mcp_server._authenticated_username",
            side_effect=tracking_username,
        ), patch(
            "mcp_server._resolve",
            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
        ):
            res = mcp_server.gitea_acquire_reviewer_pr_lease(
                pr_number=550,
                worktree="/workspace",
                candidate_head="abc123",
                remote="prgs",
            )

        self.assertTrue(res.get("success"), res)
        self.assertTrue(res.get("acquired"), res)
        self.assertEqual(res.get("comment_id"), 1003)
        self.assertTrue(posted)
        self.assertIn("gitea.prgs.cc", resolved_hosts)
        self.assertNotIn("prgs", resolved_hosts)

    def test_adopt_merger_pr_lease_resolves_identity_with_host(self):
        """Adopt path must not pass a remote alias into identity lookup."""
        import mcp_server
        import reviewer_pr_lease

        reviewer_pr_lease.clear_session_lease()
        head = "a" * 40
        resolved_hosts = []
        posted = []
        reviewer_comment = _reviewer_lease_comment(
            550,
            session_id="reviewer-session",
            head_sha=head,
            reviewer="reviewer-bot",
        )

        def tracking_username(host):
            resolved_hosts.append(host)
            return None if host in {"prgs", "dadeschools"} else "merger-bot"

        def mock_api(method, url, auth, data=None):
            if "/pulls/" in url and "/comments" not in url:
                return {
                    "number": 550,
                    "user": {"login": "author-user"},
                    "state": "open",
                    "head": {"sha": head},
                    "mergeable": True,
                    "merged": False,
                }
            if "/comments" in url:
                if method == "POST":
                    posted.append(data["body"])
                    return {"id": 1004}
                return [reviewer_comment]
            return {}

        merger_profile = {
            "profile_name": "prgs-merger",
            "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"],
            "forbidden_operations": [],
        }
        with patch("mcp_server.get_profile", return_value=merger_profile), patch(
            "mcp_server.api_request",
            side_effect=mock_api,
        ), patch(
            "mcp_server._authenticated_username",
            side_effect=tracking_username,
        ), patch(
            "mcp_server._resolve",
            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
        ), patch(
            "mcp_server.gitea_get_pr_review_feedback",
            return_value={
                "approval_at_current_head": True,
                "latest_approved_head_sha": head,
            },
        ):
            res = mcp_server.gitea_adopt_merger_pr_lease(
                pr_number=550,
                worktree="/workspace",
                expected_head_sha=head,
                remote="prgs",
            )

        self.assertTrue(res.get("success"), res)
        self.assertTrue(res.get("adopted"), res)
        self.assertEqual(res.get("adoption_comment_id"), 1004)
        self.assertTrue(posted)
        self.assertIn("gitea.prgs.cc", resolved_hosts)
        self.assertNotIn("prgs", resolved_hosts)

    def test_release_reviewer_pr_lease_same_identity_cross_session(self):
        """Same reviewer identity may release a lease claimed by another session.

        Regression: release used to call ``_authenticated_username(remote)``
        with the remote alias (e.g. ``prgs``) instead of the resolved host,
        so identity resolved empty and same-identity release fail-closed even
        when the authenticated user owned the lease.
        """
        import mcp_server
        import reviewer_pr_lease

        mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
        mcp_server.gitea_load_review_workflow()
        reviewer_pr_lease.clear_session_lease()

        foreign_session = "46820-dd78cdc4aacc"
        comments = [
            _reviewer_lease_comment(
                457,
                session_id=foreign_session,
                head_sha="ef287eaf5841d9e542a4e888ce0ae7d679dbd685",
                reviewer="sysadmin",
            )
        ]
        posted = []
        resolved_hosts = []

        def mock_api(method, url, auth, data=None):
            if "/api/v1/user" in url:
                return {"login": "sysadmin"}
            if "/comments" in url:
                if method == "POST":
                    new_c = {
                        "id": 8071,
                        "body": data["body"],
                        "user": {"login": "sysadmin"},
                    }
                    comments.append(new_c)
                    posted.append(new_c)
                    return {"id": 8071}
                return comments
            if "/pulls/" in url:
                return {
                    "user": {"login": "jcwalker3"},
                    "state": "open",
                    "head": {"sha": "ef287eaf5841d9e542a4e888ce0ae7d679dbd685"},
                    "mergeable": False,
                }
            return {}

        def tracking_username(host):
            resolved_hosts.append(host)
            # Empty identity when passed a remote alias (the pre-fix bug path).
            if host in {"prgs", "dadeschools"}:
                return None
            return "sysadmin"

        with patch("mcp_server.api_request", side_effect=mock_api), patch(
            "mcp_server._authenticated_username", side_effect=tracking_username
        ), patch(
            "mcp_server._resolve",
            return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
        ), patch("mcp_server._auth", return_value={"Authorization": "token x"}), patch(
            "mcp_server._verify_role_mutation_workspace", return_value=None
        ):
            res = mcp_server.gitea_release_reviewer_pr_lease(
                pr_number=457, remote="prgs"
            )

        self.assertTrue(res.get("success"), res)
        self.assertTrue(res.get("released"), res)
        self.assertEqual(res.get("comment_id"), 8071)
        self.assertTrue(posted)
        self.assertIn("phase: released", posted[0]["body"])
        # Must resolve identity with host, never the remote alias alone.
        self.assertIn("gitea.prgs.cc", resolved_hosts)
        self.assertNotIn("prgs", resolved_hosts)
