fix: control-plane guide org/repo params after #530 guard (Closes #588) #589

Merged
sysadmin merged 1 commits from fix/issue-588-guide-repo-guard into master 2026-07-09 12:49:25 -05:00
4 changed files with 202 additions and 6 deletions
+68 -4
View File
@@ -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 prgsTimesheet 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
View File
@@ -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(
*,
+15
View File
@@ -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)
+93
View File
@@ -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()