"""#685: gitea_resolve_task_capability must be side-effect free.

Stale-runtime detection remains fail-closed, but the resolver must never:
  * touch mcp_config.json (or any MCP client config)
  * spawn recovery threads
  * call os._exit / terminate the serving process
  * claim that an auto-restart was triggered

Recovery is owned by the IDE/client reconnect path only.
"""

from __future__ import annotations

import os
import tempfile
import threading
import unittest
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, patch

import sys

ROOT = str(Path(__file__).resolve().parent.parent)
if ROOT not in sys.path:
    sys.path.insert(0, ROOT)

import gitea_mcp_server as mcp_server


ROLE_PROFILES = (
    ("create_issue", "prgs-author", "author"),
    ("review_pr", "prgs-reviewer", "reviewer"),
    ("merge_pr", "prgs-merger", "merger"),
    ("reconciliation_cleanup", "prgs-reconciler", "reconciler"),
)


def _stale_self_ps_mocks(profile: str = "prgs-author"):
    """Build subprocess mocks: self PID is stale vs code mtime."""
    mock_getpid = MagicMock(return_value=12345)
    mock_exists = MagicMock(return_value=True)
    code_time = datetime(2026, 7, 8, 14, 0, 0)
    mock_getmtime = MagicMock(return_value=code_time.timestamp())

    ps_output = (
        "  PID LSTART COMMAND\n"
        "12345 Wed Jul  8 13:00:00 2026 /path/to/python mcp_server.py\n"
    )
    mock_run_ps = MagicMock()
    mock_run_ps.stdout = ps_output

    mock_run_env = MagicMock()
    mock_run_env.stdout = f"GITEA_MCP_PROFILE={profile}"

    mock_run_git = MagicMock()
    mock_run_git.stdout = "SAME"

    def side_effect(args, **kwargs):
        if args[0] == "ps" and "eww" in args:
            return mock_run_env
        if args[0] == "ps":
            return mock_run_ps
        if args[0] == "git":
            return mock_run_git
        raise ValueError(f"Unexpected subprocess args: {args}")

    mock_run = MagicMock(side_effect=side_effect)
    return mock_getpid, mock_exists, mock_getmtime, mock_run


class TestIssue685DiagnosticsNoSideEffects(unittest.TestCase):
    def setUp(self):
        mcp_server._process_boot_head_sha = None

    def tearDown(self):
        mcp_server._process_boot_head_sha = None

    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
    @patch("subprocess.run")
    @patch("os.path.getmtime")
    @patch("os.path.exists")
    @patch("os.getpid")
    @patch("os.utime")
    @patch("threading.Thread")
    @patch("os._exit")
    def test_stale_self_does_not_touch_config_or_exit(
        self,
        mock_exit,
        mock_thread,
        mock_utime,
        mock_getpid,
        mock_exists,
        mock_getmtime,
        mock_run,
    ):
        mock_getpid.return_value = 12345
        mock_exists.return_value = True
        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect

        before_threads = threading.active_count()
        reasons = mcp_server._check_mcp_runtimes_diagnostics(
            "create_issue", ["prgs-author"]
        )
        after_threads = threading.active_count()

        self.assertTrue(
            any("stale-runtime" in r and "active Gitea MCP server process is stale" in r
                for r in reasons),
            reasons,
        )
        # Must not claim auto-restart / config touch
        blob = " ".join(reasons)
        self.assertNotIn("Auto-restart has been triggered", blob)
        self.assertNotIn("touched mcp_config", blob)
        self.assertNotIn("will cleanly exit", blob)

        mock_utime.assert_not_called()
        mock_thread.assert_not_called()
        mock_exit.assert_not_called()
        self.assertEqual(before_threads, after_threads)

    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
    @patch("subprocess.run")
    @patch("os.path.getmtime")
    @patch("os.path.exists")
    @patch("os.getpid")
    @patch("os.utime")
    def test_repeated_stale_calls_do_not_trigger_restart_loop(
        self, mock_utime, mock_getpid, mock_exists, mock_getmtime, mock_run
    ):
        mock_getpid.return_value = 12345
        mock_exists.return_value = True
        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect

        for _ in range(5):
            reasons = mcp_server._check_mcp_runtimes_diagnostics(
                "create_issue", ["prgs-author"]
            )
            self.assertTrue(any("stale-runtime" in r for r in reasons))

        mock_utime.assert_not_called()

    def test_trigger_mcp_auto_restart_removed(self):
        """#685 AC: auto-restart helper is removed (unreachable from read-only)."""
        self.assertFalse(hasattr(mcp_server, "_trigger_mcp_auto_restart"))
        self.assertFalse(hasattr(mcp_server, "_restart_triggered"))


class TestIssue685ResolverTypedBlocker(unittest.TestCase):
    def setUp(self):
        mcp_server._process_boot_head_sha = None

    def tearDown(self):
        mcp_server._process_boot_head_sha = None
        if hasattr(mcp_server, "capability_stop_terminal"):
            mcp_server.capability_stop_terminal.clear()

    def _resolve_with_stale_runtime(self, task: str, profile_name: str, role: str):
        allowed = [
            "gitea.read",
            "gitea.issue.create",
            "gitea.issue.comment",
            "gitea.issue.close",
            "gitea.branch.create",
            "gitea.branch.push",
            "gitea.branch.delete",
            "gitea.pr.create",
            "gitea.pr.comment",
            "gitea.pr.review",
            "gitea.pr.approve",
            "gitea.pr.request_changes",
            "gitea.pr.merge",
            "gitea.pr.close",
            "gitea.repo.commit",
        ]
        profile = {
            "profile_name": profile_name,
            "role": role,
            "allowed_operations": allowed,
            "forbidden_operations": [],
        }
        config = {
            "profiles": {
                profile_name: {
                    "role": role,
                    "allowed_operations": allowed,
                    "forbidden_operations": [],
                }
            }
        }

        mock_getpid, mock_exists, mock_getmtime, mock_run = _stale_self_ps_mocks(
            profile_name
        )

        with patch.dict(
            os.environ,
            {
                "GITEA_FORCE_MCP_RUNTIME_CHECK": "1",
                "GITEA_MCP_PROFILE": profile_name,
            },
            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="test-user"
        ), patch.object(
            mcp_server, "_ensure_matching_profile", return_value=None
        ), 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
        ), patch(
            "subprocess.run", mock_run
        ), patch(
            "os.path.getmtime", mock_getmtime
        ), patch(
            "os.path.exists", mock_exists
        ), patch(
            "os.getpid", mock_getpid
        ), patch(
            "os.utime"
        ) as mock_utime, patch(
            "threading.Thread"
        ) as mock_thread, patch(
            "os._exit"
        ) as mock_exit:
            result = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
            return result, mock_utime, mock_thread, mock_exit

    def test_stale_returns_typed_blocker_fields(self):
        result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
            "create_issue", "prgs-author", "author"
        )
        self.assertTrue(result.get("restart_required"), result)
        self.assertTrue(result.get("stop_required"), result)
        self.assertEqual(result.get("blocker_kind"), "runtime_reconnect_required")
        self.assertIs(result.get("mutation_performed"), False)
        action = result.get("exact_safe_next_action") or ""
        self.assertIn("reconnect", action.lower())
        self.assertNotIn("None; ready for operations", action)
        reason = result.get("reason") or ""
        self.assertIn("stale-runtime", reason)
        self.assertNotIn("Auto-restart has been triggered", reason)
        mock_utime.assert_not_called()
        mock_thread.assert_not_called()
        mock_exit.assert_not_called()

    def test_all_four_role_profiles_get_same_side_effect_free_contract(self):
        for task, profile, role in ROLE_PROFILES:
            with self.subTest(task=task, profile=profile):
                # Skip tasks that may be unknown on this branch
                try:
                    import task_capability_map as tcm

                    tcm.required_permission(task)
                except Exception:
                    self.skipTest(f"task {task} not in capability map")

                result, mock_utime, mock_thread, mock_exit = (
                    self._resolve_with_stale_runtime(task, profile, role)
                )
                self.assertTrue(
                    result.get("restart_required") or result.get("stop_required"),
                    result,
                )
                self.assertEqual(
                    result.get("blocker_kind"), "runtime_reconnect_required", result
                )
                self.assertIs(result.get("mutation_performed"), False, result)
                mock_utime.assert_not_called()
                mock_thread.assert_not_called()
                mock_exit.assert_not_called()

    def test_config_mtime_and_contents_unchanged(self):
        with tempfile.TemporaryDirectory() as tmp:
            cfg = os.path.join(tmp, "mcp_config.json")
            original = '{"servers": {"gitea-author": {}}}'
            with open(cfg, "w", encoding="utf-8") as fh:
                fh.write(original)
            mtime_before = os.path.getmtime(cfg)

            result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
                "create_issue", "prgs-author", "author"
            )
            # Force-path also must not use real utime when diagnostics runs
            with open(cfg, encoding="utf-8") as fh:
                after = fh.read()
            self.assertEqual(after, original)
            self.assertEqual(os.path.getmtime(cfg), mtime_before)
            mock_utime.assert_not_called()
            self.assertTrue(result.get("restart_required"), result)


class TestIssue685MutationGatesStillFailClosed(unittest.TestCase):
    def test_parity_stale_still_reports_restart_required(self):
        """Mutation-facing parity gate remains fail-closed when heads differ."""
        import master_parity_gate as mpg

        out = mpg.assess_master_parity(
            {"startup_head": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
        )
        self.assertFalse(out.get("in_parity"))
        self.assertTrue(out.get("restart_required") or out.get("stale"))


class TestIssue685DocstringReadOnlyContract(unittest.TestCase):
    def test_resolve_docstring_declares_side_effect_free(self):
        doc = mcp_server.gitea_resolve_task_capability.__doc__ or ""
        lower = doc.lower()
        self.assertTrue(
            "side-effect" in lower or "read-only" in lower or "does not mutate" in lower,
            doc,
        )
        self.assertNotIn("auto-restart", lower)
        self.assertNotIn("os._exit", lower)


class TestIssue685MergeCoexistenceWithMasterAnnotations(unittest.TestCase):
    """After merging master: #685 reconnect blocker + #702 report-only binding."""

    def setUp(self):
        mcp_server._process_boot_head_sha = None

    def tearDown(self):
        mcp_server._process_boot_head_sha = None
        if hasattr(mcp_server, "capability_stop_terminal"):
            mcp_server.capability_stop_terminal.clear()

    def test_resolve_uses_report_only_stale_binding_assessment(self):
        """Capability resolve must call stale-binding assess with auto_recover=False."""
        allowed = [
            "gitea.read",
            "gitea.issue.create",
            "gitea.issue.comment",
            "gitea.branch.push",
            "gitea.pr.create",
            "gitea.pr.comment",
            "gitea.pr.review",
            "gitea.pr.merge",
            "gitea.repo.commit",
        ]
        profile = {
            "profile_name": "prgs-author",
            "role": "author",
            "allowed_operations": allowed,
            "forbidden_operations": [],
        }
        config = {
            "profiles": {
                "prgs-author": {
                    "role": "author",
                    "allowed_operations": allowed,
                    "forbidden_operations": [],
                }
            }
        }
        mock_getpid, mock_exists, mock_getmtime, mock_run = _stale_self_ps_mocks(
            "prgs-author"
        )
        fake_binding = {
            "classification": "missing_path",
            "active_worktree": "/tmp/gone",
            "reasons": ["path missing"],
        }

        with patch.dict(
            os.environ,
            {
                "GITEA_FORCE_MCP_RUNTIME_CHECK": "1",
                "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="test-user"
        ), patch.object(
            mcp_server, "_ensure_matching_profile", return_value=None
        ), 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
        ), patch.object(
            mcp_server,
            "_assess_stale_active_binding",
            return_value=fake_binding,
        ) as mock_assess, patch(
            "subprocess.run", mock_run
        ), patch(
            "os.path.getmtime", mock_getmtime
        ), patch(
            "os.path.exists", mock_exists
        ), patch(
            "os.getpid", mock_getpid
        ), patch(
            "os.utime"
        ) as mock_utime, patch(
            "threading.Thread"
        ) as mock_thread, patch(
            "os._exit"
        ) as mock_exit:
            result = mcp_server.gitea_resolve_task_capability(
                task="create_issue", remote="prgs"
            )

        mock_assess.assert_called()
        # Every call must be report-only (no env clear / session-state write).
        for call in mock_assess.call_args_list:
            self.assertFalse(call.kwargs.get("auto_recover", True))
        self.assertEqual(
            result.get("blocker_kind"), "runtime_reconnect_required", result
        )
        self.assertEqual(result.get("stale_binding_recovery"), fake_binding, result)
        self.assertIs(result.get("mutation_performed"), False, result)
        mock_utime.assert_not_called()
        mock_thread.assert_not_called()
        mock_exit.assert_not_called()


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