"""#714: fail closed on cross-host MCP profile drift and capability substitution."""

from __future__ import annotations

import json
import os
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest.mock import patch

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

import gitea_config  # noqa: E402
import mcp_server  # noqa: E402
import session_context_binding as session_ctx  # noqa: E402


CONFIG_714 = {
    "version": 2,
    "allow_runtime_switching": True,
    "contexts": {
        "prgs": {
            "enabled": True,
            "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
        },
        "mdcps": {
            "enabled": True,
            "gitea": {"enabled": True, "base_url": "https://gitea.dadeschools.net"},
        },
    },
    "profiles": {
        "prgs-author": {
            "enabled": True,
            "context": "prgs",
            "role": "author",
            "username": "jcwalker3",
            "base_url": "https://gitea.prgs.cc",
            "auth": {"type": "env", "name": "GITEA_TOKEN_PRGS_AUTHOR"},
            "allowed_operations": [
                "gitea.read",
                "gitea.issue.create",
                "gitea.issue.comment",
                "gitea.issue.close",
                "gitea.pr.create",
                "gitea.pr.comment",
                "gitea.branch.create",
                "gitea.branch.push",
                "gitea.repo.commit",
            ],
            "forbidden_operations": [
                "gitea.pr.approve",
                "gitea.pr.merge",
                "gitea.pr.request_changes",
            ],
            "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
            "execution_profile": "prgs-author",
        },
        "mdcps-reviewer": {
            "enabled": True,
            "context": "mdcps",
            "role": "reviewer",
            "username": "913443",
            "base_url": "https://gitea.dadeschools.net",
            "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_REVIEWER"},
            "allowed_operations": [
                "gitea.read",
                "gitea.pr.review",
                "gitea.pr.comment",
                "gitea.pr.approve",
                "gitea.pr.request_changes",
            ],
            "forbidden_operations": [
                "gitea.branch.create",
                "gitea.branch.push",
                "gitea.pr.create",
                "gitea.pr.merge",
                "gitea.repo.commit",
            ],
            "allowed_repositories": ["913443/eAgenda"],
            "execution_profile": "mdcps-reviewer",
        },
        "mdcps-author": {
            "enabled": True,
            "context": "mdcps",
            "role": "author",
            "username": "913443",
            "base_url": "https://gitea.dadeschools.net",
            "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_AUTHOR"},
            "allowed_operations": [
                "gitea.read",
                "gitea.issue.create",
                "gitea.issue.comment",
                "gitea.pr.create",
                "gitea.pr.comment",
                "gitea.branch.create",
                "gitea.branch.push",
                "gitea.repo.commit",
            ],
            "forbidden_operations": [
                "gitea.pr.approve",
                "gitea.pr.merge",
                "gitea.pr.request_changes",
            ],
            "allowed_repositories": ["913443/eAgenda"],
            "execution_profile": "mdcps-author",
        },
    },
}


class TestIssue714SessionContextImmutability(unittest.TestCase):
    def setUp(self):
        self._dir = tempfile.TemporaryDirectory()
        self.config_path = os.path.join(self._dir.name, "profiles.json")
        with open(self.config_path, "w", encoding="utf-8") as fh:
            fh.write(json.dumps(CONFIG_714))
        self._remotes = patch.dict(
            mcp_server.REMOTES,
            {
                "dadeschools": {
                    "host": "gitea.dadeschools.net",
                    "org": "913443",
                    "repo": "eAgenda",
                },
                "prgs": {
                    "host": "gitea.prgs.cc",
                    "org": "Scaled-Tech-Consulting",
                    "repo": "Gitea-Tools",
                },
            },
            clear=False,
        )
        self._remotes.start()
        def _url(name):
            if name == "dadeschools":
                return "https://gitea.dadeschools.net/913443/eAgenda.git"
            if name == "prgs":
                return "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
            return None
        self._url_patch = patch.object(mcp_server, "_local_git_remote_url", side_effect=_url)
        self._url_patch.start()
        mcp_server._IDENTITY_CACHE.clear()
        gitea_config._active_profile_override = None
        mcp_server._MUTATION_AUTHORITY = None
        self._env = {
            "GITEA_MCP_CONFIG": self.config_path,
            "GITEA_MCP_PROFILE": "mdcps-reviewer",
            "GITEA_TOKEN_MDCPS_REVIEWER": "mdcps-reviewer-token",
            "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
            "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
        }

    def tearDown(self):
        self._url_patch.stop()
        self._remotes.stop()
        mcp_server._IDENTITY_CACHE.clear()
        gitea_config._active_profile_override = None
        mcp_server._MUTATION_AUTHORITY = None
        self._dir.cleanup()

    def _api_side_effect(self, method, url, header):
        # Token identity mapping for whoami endpoint
        auth = str(header)
        if "mdcps-reviewer-token" in auth or "mdcps-author-token" in auth:
            login = "913443"
        else:
            login = "jcwalker3"
        return {"login": login, "full_name": "Test", "id": 1, "email": "t@example.com"}

    def test_pin_mdcps_reviewer_never_drifts_to_prgs_author(self):
        """Reproduce the incident: pin mdcps-reviewer then resolve comment_issue."""
        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                act = mcp_server.gitea_activate_profile(
                    profile_name="mdcps-reviewer", remote="dadeschools"
                )
                self.assertTrue(act["success"])
                self.assertEqual(act["after_profile"], "mdcps-reviewer")

                who1 = mcp_server.gitea_whoami(remote="dadeschools")
                self.assertEqual(who1["profile"]["profile_name"], "mdcps-reviewer")
                self.assertEqual(who1["username"], "913443")

                # Capability that mdcps-reviewer lacks — must NOT switch to prgs-author
                res = mcp_server.gitea_resolve_task_capability(
                    task="comment_issue", remote="dadeschools"
                )
                self.assertEqual(res["active_profile"], "mdcps-reviewer")
                self.assertFalse(res["allowed_in_current_session"])
                self.assertFalse(res.get("auto_profile_substitution", True))
                self.assertNotIn("prgs-author", res["matching_configured_profile"])
                self.assertIn("mdcps-author", res["matching_configured_profile"])

                who2 = mcp_server.gitea_whoami(remote="dadeschools")
                rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
                self.assertEqual(who2["profile"]["profile_name"], "mdcps-reviewer")
                self.assertEqual(rt["active_profile"], "mdcps-reviewer")
                # Still pinned — never prgs-author
                self.assertEqual(
                    gitea_config.selected_profile_name(), "mdcps-reviewer"
                )

    def test_repeated_calls_remain_on_mdcps_reviewer(self):
        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                mcp_server.gitea_activate_profile(
                    profile_name="mdcps-reviewer", remote="dadeschools"
                )
                for _ in range(5):
                    res = mcp_server.gitea_resolve_task_capability(
                        task="review_pr", remote="dadeschools"
                    )
                    self.assertEqual(res["active_profile"], "mdcps-reviewer")
                    who = mcp_server.gitea_whoami(remote="dadeschools")
                    self.assertEqual(who["profile"]["profile_name"], "mdcps-reviewer")
                    rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
                    self.assertEqual(rt["active_profile"], "mdcps-reviewer")

    def test_dadeschools_request_cannot_resolve_through_prgs_author(self):
        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                mcp_server.gitea_activate_profile(
                    profile_name="mdcps-reviewer", remote="dadeschools"
                )
                res = mcp_server.gitea_resolve_task_capability(
                    task="comment_issue", remote="dadeschools"
                )
                self.assertNotEqual(res["active_profile"], "prgs-author")
                self.assertNotIn("prgs-author", res["matching_configured_profile"])
                # Gate must not silently switch either
                blocked = mcp_server._profile_permission_block(
                    "gitea.issue.comment", remote="dadeschools"
                )
                self.assertIsNotNone(blocked)
                self.assertFalse(blocked.get("success", True))
                self.assertEqual(
                    gitea_config.selected_profile_name(), "mdcps-reviewer"
                )

    def test_prgs_request_cannot_resolve_through_mdcps_profile(self):
        """Reverse of the incident: a pinned MDCPS profile must never serve a
        prgs request, nor be silently swapped for the prgs-side profile."""
        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                mcp_server.gitea_activate_profile(
                    profile_name="mdcps-author", remote="dadeschools"
                )
                res = mcp_server.gitea_resolve_task_capability(
                    task="create_issue", remote="prgs"
                )
                self.assertNotEqual(res["active_profile"], "prgs-author")
                self.assertEqual(res["active_profile"], "mdcps-author")
                self.assertFalse(res["allowed_in_current_session"])
                self.assertTrue(res["stop_required"])
                blocked = mcp_server._session_context_mutation_block(remote="prgs")
                self.assertIsNotNone(blocked)
                self.assertTrue(any("cross-host" in r for r in blocked["reasons"]))
                self.assertEqual(gitea_config.selected_profile_name(), "mdcps-author")

    def test_unsupported_capability_fails_closed_structured(self):
        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                mcp_server.gitea_activate_profile(
                    profile_name="mdcps-reviewer", remote="dadeschools"
                )
                res = mcp_server.gitea_resolve_task_capability(
                    task="comment_issue", remote="dadeschools"
                )
                self.assertFalse(res["allowed_in_current_session"])
                self.assertTrue(res["stop_required"])
                self.assertIn("reason", res)
                self.assertIn("exact_safe_next_action", res)
                self.assertIn("activate_profile", res["exact_safe_next_action"])
                # Unknown task still fail closed (structured denial after #723;
                # never raises into internal_error and never substitutes profile).
                unknown = mcp_server.gitea_resolve_task_capability(
                    task="reopen_issue", remote="dadeschools"
                )
                self.assertFalse(unknown.get("allowed_in_current_session"))
                self.assertTrue(unknown.get("stop_required"))
                self.assertEqual(unknown.get("reason_code"), "unknown_task")
                self.assertEqual(unknown.get("mutation_performed"), False)
                self.assertEqual(unknown.get("active_profile"), "mdcps-reviewer")
                self.assertNotEqual(unknown.get("active_profile"), "prgs-author")

    def test_identity_mismatch_blocks_mutation(self):
        """Profile expects 913443 but authenticated as jcwalker3."""

        def wrong_identity(method, url, header):
            return {
                "login": "jcwalker3",
                "full_name": "Wrong",
                "id": 9,
                "email": "w@example.com",
            }

        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=wrong_identity):
                mcp_server.gitea_activate_profile(
                    profile_name="mdcps-reviewer", remote="dadeschools"
                )
                blocked = mcp_server._session_context_mutation_block(
                    remote="dadeschools"
                )
                self.assertIsNotNone(blocked)
                self.assertTrue(
                    any("identity mismatch" in r for r in blocked["reasons"])
                )

    def test_repository_host_mismatch_blocks_mutation(self):
        """mdcps-reviewer cannot serve prgs remote."""
        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                mcp_server.gitea_activate_profile(
                    profile_name="mdcps-reviewer", remote="dadeschools"
                )
                blocked = mcp_server._session_context_mutation_block(remote="prgs")
                self.assertIsNotNone(blocked)
                self.assertTrue(
                    any("cross-host" in r for r in blocked["reasons"])
                )

    def test_drift_cannot_reach_write(self):
        """Simulated mid-session profile override cannot pass permission gate."""
        with patch.dict(os.environ, self._env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                mcp_server.gitea_activate_profile(
                    profile_name="mdcps-reviewer", remote="dadeschools"
                )
                # Bind session as mdcps-reviewer
                self.assertEqual(
                    session_ctx.get_session_context()["profile_name"],
                    "mdcps-reviewer",
                )
                # Hostile override without activate_profile
                gitea_config._active_profile_override = "prgs-author"
                blocked = mcp_server._session_context_mutation_block(
                    remote="dadeschools"
                )
                self.assertIsNotNone(blocked)
                self.assertTrue(any("drift" in r or "cross-host" in r for r in blocked["reasons"]))

    def test_static_prgs_author_still_works(self):
        env = {
            "GITEA_MCP_CONFIG": self.config_path,
            "GITEA_MCP_PROFILE": "prgs-author",
            "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
        }
        with patch.dict(os.environ, env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                gitea_config._active_profile_override = None
                res = mcp_server.gitea_resolve_task_capability(
                    task="create_issue", remote="prgs"
                )
                self.assertEqual(res["active_profile"], "prgs-author")
                self.assertTrue(res["allowed_in_current_session"])
                self.assertEqual(res["active_identity"], "jcwalker3")

    def test_static_mdcps_author_still_works(self):
        env = {
            "GITEA_MCP_CONFIG": self.config_path,
            "GITEA_MCP_PROFILE": "mdcps-author",
            "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
        }
        with patch.dict(os.environ, env, clear=False):
            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
                gitea_config._active_profile_override = None
                res = mcp_server.gitea_resolve_task_capability(
                    task="comment_issue", remote="dadeschools"
                )
                self.assertEqual(res["active_profile"], "mdcps-author")
                self.assertTrue(res["allowed_in_current_session"])
                self.assertNotIn("prgs-author", res["matching_configured_profile"])


class TestSessionContextBindingUnit(unittest.TestCase):
    def test_profile_matches_remote_by_base_url(self):
        remotes = {
            "dadeschools": {"host": "gitea.dadeschools.net"},
            "prgs": {"host": "gitea.prgs.cc"},
        }
        mdcps = {"base_url": "https://gitea.dadeschools.net", "context": "mdcps"}
        prgs = {"base_url": "https://gitea.prgs.cc", "context": "prgs"}
        self.assertTrue(
            session_ctx.profile_matches_remote(mdcps, "dadeschools", remotes)
        )
        self.assertFalse(session_ctx.profile_matches_remote(mdcps, "prgs", remotes))
        self.assertTrue(session_ctx.profile_matches_remote(prgs, "prgs", remotes))

    def test_bind_and_detect_drift(self):
        session_ctx.bind_session_context(
            profile_name="mdcps-reviewer",
            remote="dadeschools",
            host="gitea.dadeschools.net",
            identity="913443",
            source="test",
        )
        ok = session_ctx.assess_session_context(
            profile_name="mdcps-reviewer",
            remote="dadeschools",
            host="gitea.dadeschools.net",
            identity="913443",
        )
        self.assertTrue(ok["proven"])
        bad = session_ctx.assess_session_context(
            profile_name="prgs-author",
            remote="dadeschools",
            host="gitea.dadeschools.net",
            identity="jcwalker3",
        )
        self.assertTrue(bad["block"])
        self.assertTrue(any("drift" in r for r in bad["reasons"]))

    def test_bound_context_survives_multiple_calls_and_exceptions(self):
        original = session_ctx.bind_session_context(
            profile_name="mdcps-reviewer",
            remote="dadeschools",
            host="gitea.dadeschools.net",
            identity="913443",
            source="test",
        )

        try:
            for _ in range(3):
                observed = session_ctx.seed_session_context_if_unbound(
                    profile_name="prgs-author",
                    remote="prgs",
                    host="gitea.prgs.cc",
                    identity="jcwalker3",
                    source="interleaved-test-call",
                )
                self.assertEqual(observed, original)
            raise RuntimeError("simulated caller failure")
        except RuntimeError:
            pass

        self.assertEqual(session_ctx.get_session_context(), original)
        drift = session_ctx.assess_session_context(
            profile_name="prgs-author",
            remote="prgs",
            host="gitea.prgs.cc",
            identity="jcwalker3",
            require_bound=True,
        )
        self.assertTrue(drift["block"])

    def test_parallel_calls_cannot_overwrite_established_binding(self):
        original = session_ctx.bind_session_context(
            profile_name="mdcps-reviewer",
            remote="dadeschools",
            host="gitea.dadeschools.net",
            identity="913443",
            source="test",
        )
        barrier = threading.Barrier(9)
        results = []
        results_lock = threading.Lock()

        def competing_seed(index: int) -> None:
            barrier.wait()
            result = session_ctx.seed_session_context_if_unbound(
                profile_name=f"prgs-author-{index}",
                remote="prgs",
                host="gitea.prgs.cc",
                identity=f"other-{index}",
                source="parallel-test-call",
            )
            with results_lock:
                results.append(result)

        threads = [threading.Thread(target=competing_seed, args=(i,)) for i in range(8)]
        for thread in threads:
            thread.start()
        barrier.wait()
        for thread in threads:
            thread.join(timeout=5)

        self.assertFalse(any(thread.is_alive() for thread in threads))
        self.assertEqual(results, [original] * 8)
        self.assertEqual(session_ctx.get_session_context(), original)

    def test_concurrent_initialization_has_single_winning_binding(self):
        """Racing first-binds from an unbound context: exactly one winner, and
        every racer observes that same complete binding (never a partial one)."""
        self.assertIsNone(session_ctx.get_session_context())
        thread_count = 8
        barrier = threading.Barrier(thread_count)
        results = []
        results_lock = threading.Lock()

        def racing_seed(index: int) -> None:
            barrier.wait()
            result = session_ctx.seed_session_context_if_unbound(
                profile_name=f"prgs-author-{index}",
                remote="prgs",
                host="gitea.prgs.cc",
                identity=f"user-{index}",
                source="concurrent-init-test",
            )
            with results_lock:
                results.append(result)

        threads = [
            threading.Thread(target=racing_seed, args=(i,))
            for i in range(thread_count)
        ]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(timeout=5)

        self.assertFalse(any(thread.is_alive() for thread in threads))
        winner = session_ctx.get_session_context()
        self.assertIsNotNone(winner)
        self.assertEqual(len(results), thread_count)
        self.assertEqual(results, [winner] * thread_count)
        # A losing racer's identity must never be spliced into the winner.
        self.assertEqual(
            winner["identity"], winner["profile_name"].replace("prgs-author-", "user-")
        )

    def test_repository_mismatch_blocks_before_mutation(self):
        """Same host and profile, different repository, must still fail closed."""
        session_ctx.bind_session_context(
            profile_name="prgs-author",
            remote="prgs",
            host="gitea.prgs.cc",
            identity="jcwalker3",
            repository="Gitea-Tools",
            org="Scaled-Tech-Consulting",
            source="test",
        )
        same = session_ctx.assess_session_context(
            profile_name="prgs-author",
            remote="prgs",
            host="gitea.prgs.cc",
            identity="jcwalker3",
            repository="Gitea-Tools",
            org="Scaled-Tech-Consulting",
            require_bound=True,
        )
        self.assertTrue(same["proven"])
        other_repo = session_ctx.assess_session_context(
            profile_name="prgs-author",
            remote="prgs",
            host="gitea.prgs.cc",
            identity="jcwalker3",
            repository="eAgenda",
            org="Scaled-Tech-Consulting",
            require_bound=True,
        )
        self.assertTrue(other_repo["block"])
        self.assertTrue(
            any("repository drift" in r for r in other_repo["reasons"])
        )
        other_org = session_ctx.assess_session_context(
            profile_name="prgs-author",
            remote="prgs",
            host="gitea.prgs.cc",
            identity="jcwalker3",
            repository="Gitea-Tools",
            org="913443",
            require_bound=True,
        )
        self.assertTrue(other_org["block"])
        self.assertTrue(any("org drift" in r for r in other_org["reasons"]))

    def test_returned_snapshot_cannot_mutate_binding(self):
        session_ctx.bind_session_context(
            profile_name="mdcps-reviewer",
            remote="dadeschools",
            host="gitea.dadeschools.net",
            identity="913443",
            source="test",
        )
        snapshot = session_ctx.get_session_context()
        snapshot["profile_name"] = "prgs-author"
        self.assertEqual(
            session_ctx.get_session_context()["profile_name"], "mdcps-reviewer"
        )


class TestSessionContextTestBoundaryIsolation(unittest.TestCase):
    def test_mdcps_binding_starts_clean_and_does_not_escape_test(self):
        self.assertIsNone(session_ctx.get_session_context())
        session_ctx.bind_session_context(
            profile_name="mdcps-reviewer",
            remote="dadeschools",
            host="gitea.dadeschools.net",
            identity="913443",
            source="test-boundary",
        )

    def test_prgs_binding_starts_clean_and_does_not_escape_test(self):
        self.assertIsNone(session_ctx.get_session_context())
        session_ctx.bind_session_context(
            profile_name="prgs-author",
            remote="prgs",
            host="gitea.prgs.cc",
            identity="jcwalker3",
            source="test-boundary",
        )

    def test_reset_helper_is_rejected_outside_pytest_boundary(self):
        with patch.dict(os.environ, {}, clear=True):
            with self.assertRaises(RuntimeError):
                session_ctx._reset_session_context_for_testing()


class TestIssue714MergeCoexistenceWithMaster(unittest.TestCase):
    """Conflict-resolution regressions after merging advanced master into #715.

    Preserves:
    - #714 immutable session / no auto profile substitution
    - #685 side-effect-free resolve (auto_recover=False)
    - #709 actor identity helper from master
    """

    def test_auto_switch_helpers_remain_fail_closed(self):
        self.assertFalse(mcp_server._try_auto_switch_for_operation("gitea.pr.review"))
        self.assertFalse(
            mcp_server._try_auto_switch_for_operation(
                "gitea.issue.comment", host="gitea.prgs.cc"
            )
        )
        # Active profile is prgs-author; cannot match mdcps review permission
        # and must not switch.
        with patch.object(
            mcp_server,
            "get_profile",
            return_value={
                "profile_name": "prgs-author",
                "allowed_operations": ["gitea.read", "gitea.issue.create"],
                "forbidden_operations": ["gitea.pr.approve"],
                "base_url": "https://gitea.prgs.cc",
                "context": "prgs",
            },
        ):
            matched = mcp_server._ensure_matching_profile(
                "gitea.pr.approve", "reviewer", remote="prgs"
            )
            self.assertIsNone(matched)
            self.assertNotEqual(
                gitea_config.selected_profile_name() or "prgs-author",
                "mdcps-reviewer",
            )

    def test_authenticated_actor_helper_survives_merge(self):
        """Master #709 F7 actor identity coexists with #714 immutability."""
        self.assertTrue(hasattr(mcp_server, "_authenticated_actor"))
        with patch(
            "mcp_server.get_auth_header", return_value={"Authorization": "token x"}
        ), patch(
            "mcp_server.api_request",
            return_value={"id": 42, "login": "sysadmin"},
        ):
            mcp_server._ACTOR_IDENTITY_CACHE.clear()
            actor = mcp_server._authenticated_actor("gitea.prgs.cc")
            self.assertEqual(actor.get("user_id"), 42)
            self.assertEqual(actor.get("login"), "sysadmin")
            # Cached; second call does not re-request
            with patch("mcp_server.api_request") as mock_api:
                actor2 = mcp_server._authenticated_actor("gitea.prgs.cc")
                mock_api.assert_not_called()
                self.assertEqual(actor2.get("user_id"), 42)

    def test_resolve_still_report_only_after_master_merge(self):
        """#685: resolve path must not pass auto_recover=True after conflict merge."""
        allowed = [
            "gitea.read",
            "gitea.issue.create",
            "gitea.issue.comment",
            "gitea.pr.create",
            "gitea.pr.comment",
            "gitea.branch.create",
            "gitea.branch.push",
            "gitea.repo.commit",
        ]
        profile = {
            "profile_name": "prgs-author",
            "role": "author",
            "allowed_operations": allowed,
            "forbidden_operations": [],
            "base_url": "https://gitea.prgs.cc",
            "context": "prgs",
        }
        config = {
            "contexts": {
                "prgs": {
                    "enabled": True,
                    "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
                }
            },
            "profiles": {
                "prgs-author": {
                    "role": "author",
                    "allowed_operations": allowed,
                    "forbidden_operations": [],
                    "base_url": "https://gitea.prgs.cc",
                    "context": "prgs",
                }
            },
        }
        fake_binding = {
            "classification": "unbound",
            "active_worktree": None,
            "reasons": [],
        }
        with patch.dict(
            os.environ, {"GITEA_MCP_PROFILE": "prgs-author"}, clear=False
        ), patch.object(
            mcp_server, "get_profile", return_value=profile
        ), patch.object(
            mcp_server.gitea_config, "load_config", return_value=config
        ), patch.object(
            mcp_server, "_authenticated_username", return_value="jcwalker3"
        ), patch.object(
            mcp_server,
            "_assess_stale_active_binding",
            return_value=fake_binding,
        ) as mock_assess, patch.object(
            mcp_server, "record_preflight_check", return_value=None
        ), patch.object(
            mcp_server, "record_mutation_authority", return_value=None
        ), patch.object(
            mcp_server, "init_review_decision_lock", return_value=None
        ):
            result = mcp_server.gitea_resolve_task_capability(
                task="create_issue", remote="prgs"
            )
        mock_assess.assert_called()
        for call in mock_assess.call_args_list:
            self.assertFalse(call.kwargs.get("auto_recover", True))
        self.assertEqual(result.get("mutation_performed"), False)
        # Session context must not have been rewritten by resolve
        # (report-only; any prior binding stays; unbound stays unbound unless
        # seed-on-read path is intentional — resolve must not activate peers).
        self.assertNotEqual(result.get("active_profile"), "mdcps-reviewer")


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