diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f3f6edc..d974a68 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3580,6 +3580,64 @@ def _authenticated_username(host: str): return user +# #943: process-local session identifier. The pre-existing call sites that mint a +# session id (workflow dashboard, lease adopt, lease reclaim) all build the same +# "--" shape when the caller supplies none. Binding it once per +# process keeps lease-ownership comparisons stable for the life of the session +# instead of minting a fresh identifier — and therefore a fresh owner — on every +# call. Process-local only, never shared to a file (same rationale as the +# immutable session context in session_context_binding). +_ACTIVE_SESSION_ID: str | None = None + + +def _active_username() -> str | None: + """Authenticated identity bound to this session, or None when unbound. + + Reads the immutable #714 session context that ``gitea_whoami`` seeds; it is + the authoritative identity pin every other mutation gate already consults. + Never re-derives or fabricates an identity: an unbound context returns None + so callers fail closed instead of acting as an unverified actor. + """ + ctx = session_ctx.get_session_context() or {} + return ((ctx.get("identity") or "").strip()) or None + + +def _active_profile_name() -> str | None: + """Active runtime profile name, or None when it cannot be determined. + + The live profile is authoritative (``get_profile``); the bound session + context is consulted only when the profile cannot be read, so the reported + name always describes the profile actually serving this process. + """ + try: + profile = get_profile() or {} + except Exception: + profile = {} + name = (profile.get("profile_name") or "").strip() + if name: + return name + ctx = session_ctx.get_session_context() or {} + return ((ctx.get("profile_name") or "").strip()) or None + + +def _current_session_id() -> str | None: + """Session identifier for this process, or None when the profile is unknown. + + Uses the same "--" shape as the existing lease call + sites. Bound once per process so repeated calls describe one session; fails + soft to None when the profile is undeterminable, letting callers fail closed + rather than inventing an owner. + """ + global _ACTIVE_SESSION_ID + if _ACTIVE_SESSION_ID: + return _ACTIVE_SESSION_ID + profile_name = _active_profile_name() + if not profile_name: + return None + _ACTIVE_SESSION_ID = f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}" + return _ACTIVE_SESSION_ID + + def _authenticated_actor(host: str) -> dict: """Resolve the authenticated actor's stable identity (#709 F7 review 438). @@ -9902,6 +9960,24 @@ def gitea_commit_files( } +def _author_mutation_block(reasons: list[str], **extra) -> dict: + """Uniform fail-closed shape for an author mutation refused after a reviewer stop. + + #943: referenced by ``gitea_bootstrap_author_issue_worktree`` and never + defined, so the reviewer-stop refusal path raised ``NameError`` instead of + returning its refusal. Mirrors the inline shape the other author mutations + return for the same ``check_author_mutation_after_reviewer_stop`` block. + """ + payload = { + "success": False, + "performed": False, + "outcome": "REFUSED", + "reasons": reasons, + } + payload.update(extra) + return payload + + def _publication_block(reasons: list[str], **extra) -> dict: """Uniform fail-closed shape for publication refusals (#812 AC20).""" payload = { diff --git a/tests/test_issue_943_runtime_context_helpers.py b/tests/test_issue_943_runtime_context_helpers.py new file mode 100644 index 0000000..f954249 --- /dev/null +++ b/tests/test_issue_943_runtime_context_helpers.py @@ -0,0 +1,458 @@ +"""Regression: the author bootstrap wrapper's runtime-context helpers (#943). + +``gitea_bootstrap_author_issue_worktree`` passed three values down to +``author_issue_bootstrap.bootstrap_author_issue_worktree``:: + + active_identity=_active_username(), + active_profile=_active_profile_name(), + owner_session=_current_session_id(), + +None of those three names was ever defined. Commit ``a942afe`` (#850) introduced +the references and no definition, so every call — dry-run included — raised +``NameError: name '_active_username' is not defined`` while evaluating the +arguments, before the bootstrap service was entered. + +The defect was unreachable until PR #942 (#941) wired the bootstrap scope into +``workflow_scope_guard``: until then ``verify_preflight_purity`` refused first +with ``missing_issue_worktree``, so the guard fix is what exposed this. + +``test_every_global_referenced_by_the_wrapper_resolves`` is the test that would +have caught the original defect: it resolves every global name the wrapper's +body references. Asserting only that the three known helpers now exist would +not generalise to the next missing reference. +""" + +from __future__ import annotations + +import ast +import builtins +import os +import re +import subprocess +import tempfile +import unittest +from unittest import mock + +import author_issue_bootstrap as aib +import create_issue_bootstrap as cib +import gitea_mcp_server as gms +import workflow_scope_guard + +BOOTSTRAP_TASK = "bootstrap_author_issue_worktree" +WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree" +RUNTIME_HELPERS = ("_active_username", "_active_profile_name", "_current_session_id") + +# "--", the shape the pre-existing lease call sites mint. +SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+-\d+-[0-9a-f]{8}$") + + +def _make_control_repo(tmp: str) -> tuple[str, str]: + """Create a clean control checkout on master and return (path, head).""" + repo = os.path.join(tmp, "repo") + os.makedirs(os.path.join(repo, "branches")) + subprocess.check_call( + ["git", "init", "-b", "master", repo], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + subprocess.check_call( + [ + "git", "-C", repo, + "-c", "user.email=t@t", "-c", "user.name=t", + "commit", "--allow-empty", "-m", "init", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + head = subprocess.check_output( + ["git", "-C", repo, "rev-parse", "HEAD"], text=True + ).strip() + return repo, head + + +def _wrapper_ast() -> ast.FunctionDef: + """Return the AST of the bootstrap wrapper as it exists on disk. + + Read from source rather than ``inspect``: the tool decorator may replace the + callable, and the defect lived in the *source* argument expressions. + """ + path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "gitea_mcp_server.py") + with open(path, encoding="utf-8") as fh: + tree = ast.parse(fh.read()) + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name == WRAPPER_NAME: + return node + raise AssertionError(f"{WRAPPER_NAME} not found in gitea_mcp_server.py") + + +class RuntimeHelperResolutionTests(unittest.TestCase): + """AC: every runtime helper the wrapper references is defined and callable.""" + + def test_three_named_helpers_are_defined_and_callable(self): + for name in RUNTIME_HELPERS: + with self.subTest(helper=name): + self.assertTrue( + hasattr(gms, name), f"{name} is referenced but not defined" + ) + self.assertTrue(callable(getattr(gms, name)), f"{name} not callable") + + def test_helpers_accept_zero_arguments_as_called(self): + """The wrapper calls each with no arguments; the signature must allow it.""" + prior = gms._ACTIVE_SESSION_ID + self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior) + for name in RUNTIME_HELPERS: + with self.subTest(helper=name): + with mock.patch.object(gms, "get_profile", return_value={}), \ + mock.patch.object( + gms.session_ctx, "get_session_context", return_value=None): + gms._ACTIVE_SESSION_ID = None + getattr(gms, name)() # must not raise TypeError + + def test_every_global_referenced_by_the_wrapper_resolves(self): + """The generalised form of this defect: an unresolvable global name. + + Collects every ``Name`` load in the wrapper body, subtracts locals + (arguments, assignments, comprehension targets, imports), and asserts the + remainder resolves against module globals or builtins. + """ + fn = _wrapper_ast() + bound: set[str] = {a.arg for a in fn.args.args} + bound |= {a.arg for a in fn.args.kwonlyargs} + if fn.args.vararg: + bound.add(fn.args.vararg.arg) + if fn.args.kwarg: + bound.add(fn.args.kwarg.arg) + for node in ast.walk(fn): + if isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)): + bound.add(node.id) + elif isinstance(node, (ast.Import, ast.ImportFrom)): + for alias in node.names: + bound.add((alias.asname or alias.name).split(".")[0]) + elif isinstance(node, ast.ExceptHandler) and node.name: + bound.add(node.name) + + unresolved = sorted( + node.id + for node in ast.walk(fn) + if isinstance(node, ast.Name) + and isinstance(node.ctx, ast.Load) + and node.id not in bound + and not hasattr(gms, node.id) + and not hasattr(builtins, node.id) + ) + self.assertEqual( + unresolved, [], f"{WRAPPER_NAME} references undefined globals: {unresolved}" + ) + + def test_wrapper_still_passes_all_three_runtime_values(self): + """Guard the wiring itself: the fix must not be 'stop passing them'.""" + fn = _wrapper_ast() + called = { + node.func.id + for node in ast.walk(fn) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) + } + for name in RUNTIME_HELPERS: + with self.subTest(helper=name): + self.assertIn(name, called) + + +class ActiveUsernameTests(unittest.TestCase): + """AC: identity comes from the authoritative pin, and fails closed.""" + + def test_returns_identity_from_bound_session_context(self): + with mock.patch.object( + gms.session_ctx, "get_session_context", + return_value={"identity": "jcwalker3", "profile_name": "prgs-author"}, + ): + self.assertEqual(gms._active_username(), "jcwalker3") + + def test_unbound_context_returns_none_so_callers_fail_closed(self): + with mock.patch.object( + gms.session_ctx, "get_session_context", return_value=None + ): + self.assertIsNone(gms._active_username()) + + def test_blank_identity_is_not_treated_as_an_identity(self): + for blank in ("", " ", None): + with self.subTest(identity=blank): + with mock.patch.object( + gms.session_ctx, "get_session_context", + return_value={"identity": blank}, + ): + self.assertIsNone(gms._active_username()) + + def test_identity_is_not_fabricated_from_the_profile(self): + """A profile's expected username must never stand in for a real identity.""" + with mock.patch.object( + gms.session_ctx, "get_session_context", + return_value={"identity": None, "expected_username": "jcwalker3"}, + ): + self.assertIsNone(gms._active_username()) + + +class ActiveProfileNameTests(unittest.TestCase): + """AC: profile name comes from the live profile, context only as fallback.""" + + def test_prefers_the_live_profile(self): + with mock.patch.object( + gms, "get_profile", return_value={"profile_name": "prgs-author"} + ), mock.patch.object( + gms.session_ctx, "get_session_context", + return_value={"profile_name": "prgs-reviewer"}, + ): + self.assertEqual(gms._active_profile_name(), "prgs-author") + + def test_falls_back_to_session_context_when_profile_unreadable(self): + with mock.patch.object(gms, "get_profile", side_effect=RuntimeError("no cfg")), \ + mock.patch.object( + gms.session_ctx, "get_session_context", + return_value={"profile_name": "prgs-author"}): + self.assertEqual(gms._active_profile_name(), "prgs-author") + + def test_returns_none_when_neither_source_knows(self): + with mock.patch.object(gms, "get_profile", return_value={}), \ + mock.patch.object( + gms.session_ctx, "get_session_context", return_value=None): + self.assertIsNone(gms._active_profile_name()) + + +class CurrentSessionIdTests(unittest.TestCase): + """AC: a real, stable session identifier — never a fresh owner per call.""" + + def setUp(self): + self._prior = gms._ACTIVE_SESSION_ID + gms._ACTIVE_SESSION_ID = None + self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", self._prior) + + def test_shape_matches_the_existing_lease_call_sites(self): + with mock.patch.object( + gms, "get_profile", return_value={"profile_name": "prgs-author"} + ): + sid = gms._current_session_id() + self.assertRegex(sid, SESSION_ID_RE) + self.assertTrue(sid.startswith("prgs-author-")) + self.assertIn(str(os.getpid()), sid) + + def test_stable_across_calls_within_one_process(self): + """A new id per call would make lease-ownership checks unsatisfiable.""" + with mock.patch.object( + gms, "get_profile", return_value={"profile_name": "prgs-author"} + ): + first = gms._current_session_id() + second = gms._current_session_id() + third = gms._current_session_id() + self.assertEqual(first, second) + self.assertEqual(second, third) + + def test_returns_none_when_profile_undeterminable(self): + with mock.patch.object(gms, "get_profile", return_value={}), \ + mock.patch.object( + gms.session_ctx, "get_session_context", return_value=None): + self.assertIsNone(gms._current_session_id()) + + def test_none_result_is_not_memoised_as_a_session(self): + with mock.patch.object(gms, "get_profile", return_value={}), \ + mock.patch.object( + gms.session_ctx, "get_session_context", return_value=None): + self.assertIsNone(gms._current_session_id()) + with mock.patch.object( + gms, "get_profile", return_value={"profile_name": "prgs-author"} + ): + self.assertIsNotNone(gms._current_session_id()) + + +class BootstrapServiceReachedTests(unittest.TestCase): + """AC: the values the helpers produce carry a dry-run into the service.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.tmp = self._tmp.name + self.repo, self.head = _make_control_repo(self.tmp) + self.journals = os.path.join(self.tmp, "journals") + os.makedirs(self.journals) + + def _bootstrap(self, **over): + kwargs = dict( + issue_number=943, + canonical_repo_root=self.repo, + expected_base_sha=self.head, + branch_name="fix/issue-943-runtime-context-helpers", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + active_identity="jcwalker3", + active_profile="prgs-author", + owner_session="prgs-author-4242-abcdef12", + lock_dir=self.journals, + idempotency_key="test-943", + dry_run=True, + ) + kwargs.update(over) + return aib.bootstrap_author_issue_worktree(**kwargs) + + def test_dry_run_succeeds_with_helper_produced_bindings(self): + """Feed the service exactly what the live helpers return.""" + prior = gms._ACTIVE_SESSION_ID + self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior) + with mock.patch.object( + gms, "get_profile", return_value={"profile_name": "prgs-author"} + ), mock.patch.object( + gms.session_ctx, "get_session_context", + return_value={"identity": "jcwalker3", "profile_name": "prgs-author"}, + ): + gms._ACTIVE_SESSION_ID = None + identity = gms._active_username() + profile = gms._active_profile_name() + session = gms._current_session_id() + + res = self._bootstrap( + active_identity=identity, active_profile=profile, owner_session=session + ) + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("dry_run")) + self.assertEqual(res.get("issue_number"), 943) + self.assertEqual(res.get("base_sha"), self.head) + + def test_dry_run_creates_no_branch_worktree_or_lease(self): + res = self._bootstrap() + self.assertTrue(res.get("success"), res) + + branches = subprocess.check_output( + ["git", "-C", self.repo, "branch", "--list"], text=True + ) + self.assertNotIn("issue-943", branches) + worktrees = subprocess.check_output( + ["git", "-C", self.repo, "worktree", "list"], text=True + ) + self.assertNotIn("issue-943", worktrees) + self.assertFalse( + os.path.exists(os.path.join(self.repo, "branches", + "fix-issue-943-runtime-context-helpers")) + ) + journal = res.get("phase_journal") or {} + self.assertFalse(journal.get("completed")) + self.assertFalse(any((journal.get("artifacts_created") or {}).values())) + self.assertIsNone(journal.get("lease_id")) + self.assertIsNone(journal.get("assignment_id")) + + def test_missing_identity_fails_closed(self): + res = self._bootstrap(active_identity=None) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "missing_active_identity") + + def test_missing_profile_fails_closed(self): + res = self._bootstrap(active_profile=" ") + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "missing_active_profile") + + def test_missing_session_fails_closed(self): + res = self._bootstrap(owner_session=None) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "missing_owner_session") + + def test_expected_base_mismatch_fails_closed(self): + res = self._bootstrap(expected_base_sha="0" * 40) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("reason_code"), "stale_concurrency_pin") + + def test_unbound_runtime_context_cannot_reach_the_service(self): + """With nothing bound, the helpers yield None and the service refuses.""" + prior = gms._ACTIVE_SESSION_ID + self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior) + with mock.patch.object(gms, "get_profile", return_value={}), \ + mock.patch.object( + gms.session_ctx, "get_session_context", return_value=None): + gms._ACTIVE_SESSION_ID = None + res = self._bootstrap( + active_identity=gms._active_username(), + active_profile=gms._active_profile_name(), + owner_session=gms._current_session_id(), + ) + self.assertFalse(res.get("success")) + self.assertIn( + res.get("reason_code"), + {"missing_owner_session", "missing_active_identity", + "missing_active_profile"}, + ) + + def test_apply_reaches_the_intended_transition(self): + res = self._bootstrap(dry_run=False) + self.assertTrue(res.get("success"), res) + self.assertNotEqual(res.get("dry_run"), True) + branches = subprocess.check_output( + ["git", "-C", self.repo, "branch", "--list"], text=True + ) + self.assertIn("issue-943", branches) + self.assertTrue(os.path.isdir(res.get("worktree_path") or "")) + + +class Issue941ScopeGuardNotRegressedTests(unittest.TestCase): + """AC: PR #942's bootstrap-scope wiring still holds.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.repo, self.head = _make_control_repo(self._tmp.name) + + def _assessment(self, task: str = BOOTSTRAP_TASK) -> dict: + return aib.assess_author_issue_bootstrap( + workspace_path=self.repo, + canonical_repo_root=self.repo, + current_branch="master", + head_sha=self.head, + porcelain_status="", + remote_master_sha=self.head, + task=task, + ) + + def test_bootstrap_task_still_permitted_from_clean_control_checkout(self): + res = workflow_scope_guard.assess_root_source_mutation( + workspace_path=self.repo, + canonical_repo_root=self.repo, + role_kind="author", + mutation_task=BOOTSTRAP_TASK, + porcelain_status="", + bootstrap_assessment=self._assessment(), + ) + self.assertFalse(res.get("block"), res) + self.assertNotEqual( + res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE + ) + + def test_bootstrap_task_still_blocked_without_evidence(self): + res = workflow_scope_guard.assess_root_source_mutation( + workspace_path=self.repo, + canonical_repo_root=self.repo, + role_kind="author", + mutation_task=BOOTSTRAP_TASK, + porcelain_status="", + ) + self.assertTrue(res.get("block")) + self.assertEqual( + res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE + ) + + def test_ordinary_author_mutation_still_blocked_from_control_checkout(self): + res = workflow_scope_guard.assess_root_source_mutation( + workspace_path=self.repo, + canonical_repo_root=self.repo, + role_kind="author", + mutation_task="commit_files", + porcelain_status="", + bootstrap_assessment=self._assessment(), + ) + self.assertTrue(res.get("block")) + self.assertEqual( + res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE + ) + + def test_create_issue_bootstrap_unchanged(self): + self.assertTrue(cib.is_create_issue_task("create_issue")) + self.assertFalse(cib.is_create_issue_task(BOOTSTRAP_TASK)) + + +if __name__ == "__main__": + unittest.main()