fix: control-plane guide org/repo params and local repo alignment (#588)
#530's remote/repo guard blocked mcp_get_control_plane_guide(remote=prgs) because the bare default resolves to Timesheet while local git is Gitea-Tools, and remediation told callers to pass org/repo on a tool that did not accept those parameters. - Accept optional org/repo on mcp_get_control_plane_guide - Prefer local git remote org/repo when bare defaults mismatch (read-only guide) - Keep mutation tools fail-closed on bare mismatch - Align remediation text with tools that accept org/repo Closes #588
This commit is contained in:
+68
-4
@@ -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,
|
||||
|
||||
+26
-2
@@ -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<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(
|
||||
*,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user