diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 209359e..48a94cc 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1359,6 +1359,55 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None): return (resolved_host, resolved_org, resolved_repo) +def _resolve_control_plane_guide_target( + remote: str, + host: str | None, + org: str | None, + repo: str | None, +) -> tuple[str, str, str]: + """Resolve guide target with schema-supported org/repo + local context (#530 follow-up). + + ``mcp_get_control_plane_guide`` is a session bootstrap tool. When called with a + bare ``remote`` whose REMOTES default repo mismatches the local git remote + (classic prgs→Timesheet vs Gitea-Tools), prefer the local remote's org/repo so + the guide remains usable. Explicit org/repo always win and are validated by + :func:`_resolve`. + """ + if org is not None and repo is not None: + return _resolve(remote, host, org, repo) + + if remote not in REMOTES: + raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}") + + profile = REMOTES[remote] + candidate_org = org if org is not None else profile["org"] + candidate_repo = repo if repo is not None else profile["repo"] + local_url = _local_git_remote_url(remote) + assessment = remote_repo_guard.assess_remote_repo_match( + remote=remote, + resolved_org=candidate_org, + resolved_repo=candidate_repo, + local_remote_url=local_url, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) + if assessment.get("block"): + parsed = remote_repo_guard.parse_org_repo_from_remote_url(local_url) + if parsed: + local_org, local_repo = parsed + # Fill only unspecified sides from local git context, then resolve + # as explicit so the guard accepts intentional local alignment. + use_org = org if org is not None else local_org + use_repo = repo if repo is not None else local_repo + return _resolve(remote, host, use_org, use_repo) + # No local context to recover — surface the original guard error. + raise RuntimeError( + remote_repo_guard.format_remote_repo_guard_error(assessment) + ) + + return _resolve(remote, host, org, repo) + + def _enforce_remote_repo_guard( remote: str, resolved_org: str, @@ -7706,6 +7755,8 @@ _PROJECT_SKILLS.update(workflow_skill.workflow_skill_registry_entries()) def mcp_get_control_plane_guide( remote: str = "dadeschools", host: str | None = None, + org: str | None = None, + repo: str | None = None, ) -> dict: """Structured operator guide for the current Gitea MCP session (#128). @@ -7725,14 +7776,19 @@ def mcp_get_control_plane_guide( Args: remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. + org: Optional owner/org override (explicit; recommended when the bare + remote default differs from the local git remote). + repo: Optional repository name override (use with org). Returns: dict with 'read_only', 'remote', 'profile' (name, operations, role_kind), 'identity' (username/status/instruction; 'server' only - with the reveal opt-in), 'guidance' (profile-specific), 'rules', - 'workflows', and 'skills_tool'. Never secrets. + with the reveal opt-in), 'resolved_repository' (org/repo actually + used after guard/local-context alignment), 'guidance' + (profile-specific), 'rules', 'workflows', and 'skills_tool'. + Never secrets. """ - h, _, _ = _resolve(remote, host, None, None) + h, o, r = _resolve_control_plane_guide_target(remote, host, org, repo) profile = get_profile() allowed = profile["allowed_operations"] forbidden = profile["forbidden_operations"] @@ -7761,7 +7817,11 @@ def mcp_get_control_plane_guide( "Do not hardcode identity or guess the launcher profile. " "Verify the active identity/profile matches the required role " "for the task. Use the correct namespace (author vs reviewer) " - "as reported." + "as reported.", + "When the bare remote default repo mismatches the local git remote " + "(e.g. prgs→Timesheet vs Gitea-Tools), pass explicit " + "org=Scaled-Tech-Consulting repo=Gitea-Tools to " + "mcp_get_control_plane_guide and other tools that accept org/repo.", ] if not username: guidance.append( @@ -7806,6 +7866,10 @@ def mcp_get_control_plane_guide( return { "read_only": True, "remote": remote, + "resolved_repository": { + "org": o, + "repo": r, + }, "profile": { "name": profile["profile_name"], "allowed_operations": allowed, diff --git a/remote_repo_guard.py b/remote_repo_guard.py index f4938bd..2757eec 100644 --- a/remote_repo_guard.py +++ b/remote_repo_guard.py @@ -16,11 +16,35 @@ or the local remote URL is unavailable (best-effort corroboration only). from __future__ import annotations +import re + REMEDIATION = ( - "Pass explicit org= and repo= matching the local git remote, " - "e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools." + "Pass explicit org= and repo= matching the local git remote on tools that " + "accept those parameters (including mcp_get_control_plane_guide), " + "e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools. " + "Do not pass org/repo to tools whose schema does not list them." ) +# https://host/org/repo.git or git@host:org/repo.git +_REMOTE_URL_SLUG_RE = re.compile( + r"(?:[:/])(?P[^/]+)/(?P[^/]+?)(?:\.git)?/*$" +) + + +def parse_org_repo_from_remote_url(remote_url: str | None) -> tuple[str, str] | None: + """Best-effort parse of org/repo from a git remote URL.""" + url = (remote_url or "").strip() + if not url: + return None + match = _REMOTE_URL_SLUG_RE.search(url) + if not match: + return None + org = (match.group("org") or "").strip() + repo = (match.group("repo") or "").strip() + if not org or not repo: + return None + return org, repo + def assess_remote_repo_match( *, diff --git a/tests/test_operator_guide.py b/tests/test_operator_guide.py index b73806d..5c751d1 100644 --- a/tests/test_operator_guide.py +++ b/tests/test_operator_guide.py @@ -88,6 +88,21 @@ class TestControlPlaneGuide(GuideTestBase): blob = " ".join(g["guidance"]).lower() self.assertIn("forbidden", blob) self.assertIn("review", blob) + self.assertIn("resolved_repository", g) + + @patch("mcp_server.api_request", return_value={"login": "author-bot"}) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_guide_explicit_org_repo_parameters(self, _auth, _api): + with patch.dict(os.environ, AUTHOR_ENV, clear=True): + g = mcp_get_control_plane_guide( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertEqual( + g["resolved_repository"], + {"org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"}, + ) @patch("mcp_server.api_request", return_value={"login": "reviewer-bot"}) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) diff --git a/tests/test_remote_repo_guard.py b/tests/test_remote_repo_guard.py index 9fbfa47..7c0ce53 100644 --- a/tests/test_remote_repo_guard.py +++ b/tests/test_remote_repo_guard.py @@ -111,6 +111,23 @@ class TestAssessRemoteRepoMatch(unittest.TestCase): self.assertIn("Gitea-Tools", message) self.assertIn("org=", message) self.assertIn("repo=", message) + # Recovery text must name a tool that actually accepts org/repo (#530 follow-up). + self.assertIn("mcp_get_control_plane_guide", message) + self.assertIn("schema does not list", message) + + def test_parse_org_repo_from_https_url(self): + parsed = remote_repo_guard.parse_org_repo_from_remote_url(LOCAL_GITEA_TOOLS_URL) + self.assertEqual(parsed, ("Scaled-Tech-Consulting", "Gitea-Tools")) + + def test_parse_org_repo_from_ssh_url(self): + parsed = remote_repo_guard.parse_org_repo_from_remote_url( + "git@gitea.prgs.cc:Scaled-Tech-Consulting/Gitea-Tools.git" + ) + self.assertEqual(parsed, ("Scaled-Tech-Consulting", "Gitea-Tools")) + + def test_parse_org_repo_empty(self): + self.assertIsNone(remote_repo_guard.parse_org_repo_from_remote_url(None)) + self.assertIsNone(remote_repo_guard.parse_org_repo_from_remote_url("")) class TestResolveServerWiring(unittest.TestCase): @@ -173,6 +190,82 @@ class TestResolveServerWiring(unittest.TestCase): with self.assertRaises(RuntimeError): self.server.gitea_view_issue(issue_number=1, remote="prgs") + def test_control_plane_guide_accepts_explicit_gitea_tools(self): + """Guide schema accepts org/repo and resolves Gitea-Tools, not Timesheet.""" + self._set_prgs_default_repo("Timesheet") + with mock.patch.object( + self.server, "api_request", return_value={"login": "guide-bot"} + ), mock.patch.object( + self.server, "get_auth_header", return_value="Basic dGVzdA==" + ), mock.patch.dict( + os.environ, + { + "GITEA_PROFILE_NAME": "author-test", + "GITEA_ALLOWED_OPERATIONS": "gitea.read", + "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.merge", + }, + clear=False, + ): + guide = self.server.mcp_get_control_plane_guide( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertTrue(guide["read_only"]) + self.assertEqual( + guide["resolved_repository"], + {"org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"}, + ) + self.assertNotEqual(guide["resolved_repository"]["repo"], "Timesheet") + + def test_control_plane_guide_bare_remote_aligns_to_local_gitea_tools(self): + """Bare remote=prgs in a Gitea-Tools worktree must not stay on Timesheet.""" + self._set_prgs_default_repo("Timesheet") + with mock.patch.object( + self.server, "api_request", return_value={"login": "guide-bot"} + ), mock.patch.object( + self.server, "get_auth_header", return_value="Basic dGVzdA==" + ), mock.patch.dict( + os.environ, + { + "GITEA_PROFILE_NAME": "author-test", + "GITEA_ALLOWED_OPERATIONS": "gitea.read", + "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.merge", + }, + clear=False, + ): + guide = self.server.mcp_get_control_plane_guide(remote="prgs") + self.assertEqual(guide["resolved_repository"]["repo"], "Gitea-Tools") + self.assertEqual( + guide["resolved_repository"]["org"], "Scaled-Tech-Consulting" + ) + + def test_control_plane_guide_wrong_default_fails_closed_without_local(self): + """Without local remote context, mismatched defaults still fail closed.""" + self._set_prgs_default_repo("Timesheet") + with mock.patch.object( + self.server, "_local_git_remote_url", return_value=None + ): + # No local URL → preference cannot recover; bare resolve uses Timesheet + # and guard is skipped only because local URL is missing (best-effort). + host, org, repo = self.server._resolve_control_plane_guide_target( + "prgs", None, None, None + ) + self.assertEqual(repo, "Timesheet") + + def test_control_plane_guide_schema_recovery_matches_parameters(self): + """Remediation must only recommend parameters the guide actually accepts.""" + import inspect + + sig = inspect.signature(self.server.mcp_get_control_plane_guide) + self.assertIn("org", sig.parameters) + self.assertIn("repo", sig.parameters) + self.assertIn("remote", sig.parameters) + remediation = remote_repo_guard.REMEDIATION + self.assertIn("mcp_get_control_plane_guide", remediation) + self.assertIn("org=", remediation) + self.assertIn("repo=", remediation) + if __name__ == "__main__": unittest.main()