Merge pull request 'fix: control-plane guide org/repo params after #530 guard (Closes #588)' (#589) from fix/issue-588-guide-repo-guard into master

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',
+        with the reveal opt-in), 'resolved_repository' (org/repo actually
-        'workflows', and 'skills_tool'. Never secrets.
+        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,

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, "
+    "Pass explicit org= and repo= matching the local git remote on tools that "
-    "e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools."
+    "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<org>[^/]+)/(?P<repo>[^/]+?)(?:\.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(
     *,

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)

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(
+            "[email protected]: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()