"""#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) if __name__ == "__main__": unittest.main()