gitea_mcp_server.py

@@ -834,6 +834,15 @@ def _anti_stomp_in_test_mode() -> bool:
     return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
 
 
+# Mutation tasks that carry an explicit cross-repository *target* contract and
+# must NOT auto-resolve the workspace repository for omitted coordinates. Only
+# ``delete_branch`` (#733) qualifies: a destructive raw-branch deletion names
+# its org/repo explicitly and fails closed otherwise. Every other mutation is
+# session-bound to the single workspace-aligned repository and resolves it from
+# the trusted git remote (mirroring :func:`_resolve`).
+_EXPLICIT_TARGET_MUTATION_TASKS = frozenset({"delete_branch"})
+
+
 def _run_anti_stomp_preflight(
     task: str | None,
     *,
@@ -899,21 +908,56 @@ def _run_anti_stomp_preflight(
     local_remote_url = None
     org_explicit = org is not None
     repo_explicit = repo is not None
+    filled_org = False
+    filled_repo = False
     if remote:
         try:
             local_remote_url = _local_git_remote_url(remote)
         except Exception:
             local_remote_url = None
         if resolved_org is None or resolved_repo is None:
-            try:
-                rem = _effective_remote(remote)
-                if rem in REMOTES:
-                    if resolved_org is None:
-                        resolved_org = REMOTES[rem]["org"]
-                    if resolved_repo is None:
-                        resolved_repo = REMOTES[rem]["repo"]
-            except Exception:
-                pass
+            # Session-bound mutations act on the single workspace-aligned
+            # repository, so prefer the trusted git-remote-derived identity for
+            # omitted coordinates — exactly as :func:`_resolve` does — instead
+            # of the remote-wide ``REMOTES`` default *target*. A bare ``prgs``
+            # default resolves to ``Timesheet`` and false-positived the #530
+            # repo guard against a ``Gitea-Tools`` checkout, blocking native
+            # issue/PR/review/merge/lock/cleanup mutations at preflight while
+            # the actual resolve stage (``_resolve``) would have targeted the
+            # workspace repo correctly. Workspace-filled sides are intentional
+            # alignment and are marked explicit so the guard accepts them.
+            # ``delete_branch`` keeps its explicit cross-repo target contract
+            # (#733) and is excluded, so it still fails closed on omitted
+            # coordinates. ``REMOTES`` stays a best-effort last resort only when
+            # no workspace identity can be derived (preserves the dadeschools
+            # default-target behavior and the existing unit suite).
+            workspace_parts = None
+            if task not in _EXPLICIT_TARGET_MUTATION_TASKS:
+                try:
+                    workspace_parts = (
+                        remote_repo_guard.parse_org_repo_from_remote_url(
+                            local_remote_url
+                        )
+                    )
+                except Exception:
+                    workspace_parts = None
+            if workspace_parts:
+                if resolved_org is None:
+                    resolved_org = workspace_parts[0]
+                    filled_org = True
+                if resolved_repo is None:
+                    resolved_repo = workspace_parts[1]
+                    filled_repo = True
+            if resolved_org is None or resolved_repo is None:
+                try:
+                    rem = _effective_remote(remote)
+                    if rem in REMOTES:
+                        if resolved_org is None:
+                            resolved_org = REMOTES[rem]["org"]
+                        if resolved_repo is None:
+                            resolved_repo = REMOTES[rem]["repo"]
+                except Exception:
+                    pass
 
     parity = _current_master_parity()
     startup_head = parity.get("startup_head")
@@ -967,8 +1011,10 @@ def _run_anti_stomp_preflight(
         resolved_org=resolved_org,
         resolved_repo=resolved_repo,
         local_remote_url=local_remote_url,
-        org_explicit=org_explicit,
-        repo_explicit=repo_explicit,
+        # Workspace-filled sides are intentional alignment for #530 (same
+        # contract as :func:`_resolve`), so treat them as explicit.
+        org_explicit=org_explicit or filled_org,
+        repo_explicit=repo_explicit or filled_repo,
         check_repo=check_repo,
         profile_name=profile_name,
         profile_role=profile_role,
@@ -2778,8 +2824,13 @@ def gitea_create_issue(
         return blocked
     try:
         with _mutation_stage("preflight_purity"):
+            # #735: forward explicit org/repo into shared anti-stomp preflight.
             verify_preflight_purity(
-                remote, worktree_path=worktree_path, task="create_issue"
+                remote,
+                worktree_path=worktree_path,
+                task="create_issue",
+                org=org,
+                repo=repo,
             )
     except Exception as exc:
         typed = _production_guard_block_from_exc(exc, number=None)
@@ -3001,7 +3052,14 @@ def gitea_lock_issue(
         )
     else:
         git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
+    # #735: forward explicit org/repo into shared anti-stomp preflight.
+    verify_preflight_purity(
+        remote,
+        worktree_path=resolved_worktree,
+        task="lock_issue",
+        org=org,
+        repo=repo,
+    )
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -3232,7 +3290,14 @@ def gitea_create_pr(
     if blocked:
         return blocked
     with _mutation_stage("preflight_purity"):
-        verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
+        # #735: forward explicit org/repo into shared anti-stomp preflight.
+        verify_preflight_purity(
+            remote,
+            worktree_path=worktree_path,
+            task="create_pr",
+            org=org,
+            repo=repo,
+        )
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -7528,9 +7593,9 @@ def gitea_edit_pr(
     # Closing uses close_pr; non-closing title/body/base edits use edit_pr so
     # the shared #604 anti-stomp inventory stays consistent with wiring.
     if closing:
-        verify_preflight_purity(remote, task="close_pr")
+        verify_preflight_purity(remote, task="close_pr", org=org, repo=repo)
     else:
-        verify_preflight_purity(remote, task="edit_pr")
+        verify_preflight_purity(remote, task="edit_pr", org=org, repo=repo)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
@@ -7781,7 +7846,8 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote, task="commit_files")
+    # #735: forward explicit org/repo into shared anti-stomp preflight.
+    verify_preflight_purity(remote, task="commit_files", org=org, repo=repo)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -8896,6 +8962,8 @@ def gitea_cleanup_merged_pr_branch(
         remote,
         worktree_path=worktree_path,
         task="cleanup_merged_pr_branch",
+        org=org,
+        repo=repo,
     )
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -9623,7 +9691,9 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
+    verify_preflight_purity(
+        remote, task="reconcile_merged_cleanups", org=org, repo=repo
+    )
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -10152,7 +10222,9 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
+    verify_preflight_purity(
+        remote, task="reconcile_already_landed_pr", org=org, repo=repo
+    )
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -10423,7 +10495,9 @@ def gitea_reconcile_superseded_by_merged_pr(
         result["safe_next_action"] = "rerun with explicit mutation flags"
         return result
 
-    verify_preflight_purity(remote, task="reconcile_close_superseded_pr")
+    verify_preflight_purity(
+        remote, task="reconcile_close_superseded_pr", org=org, repo=repo
+    )
 
     if post_comment:
         comment_url = f"{base}/issues/{target_pr_number}/comments"
@@ -10526,7 +10600,7 @@ def gitea_close_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="close_issue")
+    verify_preflight_purity(remote, task="close_issue", org=org, repo=repo)
 
     # #529: block closure that buries an unproven non-zero suite exit as an
     # "expected pre-existing failure". Skipped when no closure report is given.
@@ -12493,10 +12567,13 @@ def gitea_create_issue_comment(
         # allowed while an author holds a different implementation lock.
         # Scope ownership for source edits is enforced via branch/worktree
         # binding + root diagnostic checks (#683).
+        # #735: forward explicit org/repo into shared anti-stomp preflight.
         verify_preflight_purity(
             remote,
             worktree_path=worktree_path,
             task="comment_issue",
+            org=org,
+            repo=repo,
         )
     except Exception as exc:
         typed = _production_guard_block_from_exc(
@@ -14632,7 +14709,14 @@ def gitea_mark_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
+    # #735: forward explicit org/repo into shared anti-stomp preflight.
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task="mark_issue",
+        org=org,
+        repo=repo,
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -14722,7 +14806,7 @@ def gitea_post_heartbeat(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="post_heartbeat")
+    verify_preflight_purity(remote, task="post_heartbeat", org=org, repo=repo)
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -15801,7 +15885,9 @@ def gitea_cleanup_stale_claims(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="cleanup_stale_claims")
+    verify_preflight_purity(
+        remote, task="cleanup_stale_claims", org=org, repo=repo
+    )
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -15921,7 +16007,13 @@ def gitea_create_label(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_label")
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task="create_label",
+        org=org,
+        repo=repo,
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -15975,7 +16067,13 @@ def gitea_set_issue_labels(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="set_issue_labels")
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task="set_issue_labels",
+        org=org,
+        repo=repo,
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)

tests/test_preflight_workspace_repo_forwarding.py

@@ -0,0 +1,280 @@
+"""Regression matrix: session-bound mutation preflight resolves the workspace repo.
+
+Root-cause follow-up to #733. ``_resolve`` (the actual API-targeting resolver)
+already prefers the workspace-aligned git remote over the ``REMOTES`` default
+(bare ``prgs`` → ``Scaled-Tech-Consulting/Timesheet``). The shared #604
+anti-stomp preflight (``_run_anti_stomp_preflight``) did **not** mirror that: for
+every mutation task whose call site omitted ``org``/``repo`` it filled the
+remote-wide ``REMOTES`` default, so the #530 repo guard false-positived
+``wrong_repo`` against a ``Gitea-Tools`` checkout and blocked native
+issue/PR/review/merge/lock/cleanup mutations.
+
+The fix makes ``_run_anti_stomp_preflight`` prefer the trusted, git-remote-
+derived workspace identity for omitted coordinates on **session-bound** mutation
+tasks (marking those filled sides explicit, exactly like ``_resolve``), while
+``delete_branch`` keeps its explicit cross-repository *target* contract (#733)
+and continues to fail closed on omitted coordinates.
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+import remote_repo_guard
+
+# Workspace is bound to Gitea-Tools even though the prgs REMOTES default repo is
+# Timesheet (the exact cross-repo scenario).
+LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+
+RECONCILER = {
+    "profile_name": "prgs-reconciler",
+    "role": "reconciler",
+    "allowed_operations": [
+        "gitea.read", "gitea.issue.comment", "gitea.pr.comment",
+        "gitea.pr.close", "gitea.branch.delete",
+    ],
+    "forbidden_operations": ["gitea.pr.create", "gitea.pr.merge"],
+    "audit_label": "prgs-reconciler",
+}
+
+# Session-bound mutation tasks (every MUTATION_TASK except the explicit-target
+# delete_branch) must auto-resolve the workspace repository for omitted coords.
+SESSION_BOUND_TASKS = [
+    "create_issue",
+    "comment_issue",
+    "close_issue",
+    "mark_issue",
+    "lock_issue",
+    "set_issue_labels",
+    "create_label",
+    "create_pr",
+    "close_pr",
+    "edit_pr",
+    "commit_files",
+    "cleanup_merged_pr_branch",
+    "cleanup_stale_claims",
+    "reconcile_merged_cleanups",
+    "reconcile_already_landed_pr",
+    "reconcile_close_superseded_pr",
+    "post_heartbeat",
+    "review_pr",
+    "submit_pr_review",
+    "merge_pr",
+]
+
+
+class _AntiStompResolutionBase(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.prgs.cc",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Timesheet",
+            },
+        })
+        self._remotes.start()
+
+    def tearDown(self):
+        patch.stopall()
+
+    def _capture_resolution(self, task, org, repo, *, local_url=LOCAL_GITEA_TOOLS_URL):
+        """Drive the live anti-stomp runner and capture the org/repo it resolved
+        and passed to the pure assessor."""
+        passthrough = {
+            "allowed": True, "block": False, "blockers": [], "reasons": [],
+            "exact_next_action": "proceed", "blocker_kind": None, "checks": {},
+        }
+        with patch.dict(
+            os.environ, {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, clear=False
+        ), patch(
+            "mcp_server._local_git_remote_url", return_value=local_url
+        ), patch(
+            "mcp_server.get_profile", return_value=RECONCILER
+        ), patch.object(
+            mcp_server.anti_stomp_preflight,
+            "assess_anti_stomp_preflight",
+            return_value=passthrough,
+        ) as m:
+            mcp_server._run_anti_stomp_preflight(
+                task, remote="prgs", org=org, repo=repo
+            )
+        self.assertTrue(m.called)
+        return m.call_args.kwargs
+
+
+class TestSessionBoundResolution(_AntiStompResolutionBase):
+    def test_omitted_coords_resolve_workspace_repo_and_mark_explicit(self):
+        for task in SESSION_BOUND_TASKS:
+            with self.subTest(task=task):
+                kw = self._capture_resolution(task, None, None)
+                self.assertEqual(
+                    kw["resolved_org"], "Scaled-Tech-Consulting", task
+                )
+                self.assertEqual(kw["resolved_repo"], "Gitea-Tools", task)
+                # Workspace-filled sides are intentional alignment → explicit.
+                self.assertTrue(kw["org_explicit"], task)
+                self.assertTrue(kw["repo_explicit"], task)
+
+    def test_resolved_workspace_repo_passes_the_530_guard(self):
+        kw = self._capture_resolution("create_pr", None, None)
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org=kw["resolved_org"],
+            resolved_repo=kw["resolved_repo"],
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=kw["org_explicit"],
+            repo_explicit=kw["repo_explicit"],
+        )
+        self.assertFalse(assessment["block"], assessment)
+
+    def test_explicit_caller_coords_are_preserved(self):
+        kw = self._capture_resolution(
+            "create_pr", "Scaled-Tech-Consulting", "Gitea-Tools"
+        )
+        self.assertEqual(kw["resolved_org"], "Scaled-Tech-Consulting")
+        self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
+        self.assertTrue(kw["org_explicit"])
+        self.assertTrue(kw["repo_explicit"])
+
+    def test_no_workspace_identity_falls_back_to_remotes_default(self):
+        # No derivable local remote → best-effort REMOTES default preserved,
+        # not marked explicit (so the guard still corroborates when able).
+        kw = self._capture_resolution("create_pr", None, None, local_url=None)
+        self.assertEqual(kw["resolved_repo"], "Timesheet")
+        self.assertFalse(kw["org_explicit"])
+        self.assertFalse(kw["repo_explicit"])
+
+
+class TestDeleteBranchContractPreserved(_AntiStompResolutionBase):
+    """delete_branch keeps its explicit-target contract (#733): omitted coords
+    still resolve the remote-wide default and are NOT marked explicit, so the
+    guard fails closed against a mismatched workspace."""
+
+    def test_omitted_coords_still_remote_default_not_explicit(self):
+        kw = self._capture_resolution("delete_branch", None, None)
+        self.assertEqual(kw["resolved_repo"], "Timesheet")
+        self.assertFalse(kw["org_explicit"])
+        self.assertFalse(kw["repo_explicit"])
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org=kw["resolved_org"],
+            resolved_repo=kw["resolved_repo"],
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=kw["org_explicit"],
+            repo_explicit=kw["repo_explicit"],
+        )
+        self.assertTrue(assessment["block"])
+
+    def test_explicit_coords_still_forwarded(self):
+        kw = self._capture_resolution(
+            "delete_branch", "Scaled-Tech-Consulting", "Gitea-Tools"
+        )
+        self.assertEqual(kw["resolved_repo"], "Gitea-Tools")
+        self.assertTrue(kw["org_explicit"])
+        self.assertTrue(kw["repo_explicit"])
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestAuthorCallSiteForwardsOrgRepo(_AntiStompResolutionBase):
+    """#735: author mutation entrypoints must pass org/repo into preflight."""
+
+    def _assert_call_forwards(self, fn_name: str, *args, **kwargs):
+        import inspect
+        fn = getattr(mcp_server, fn_name)
+        # Force preflight path to raise with the kwargs it received so we can
+        # prove org/repo were forwarded without performing the full mutation.
+        captured = {}
+
+        def _capture(*a, **kw):
+            captured.update(kw)
+            captured["_args"] = a
+            raise RuntimeError("capture-only")
+
+        with patch.object(mcp_server, "verify_preflight_purity", side_effect=_capture), \
+             patch.object(mcp_server, "get_profile", return_value={
+                 "profile_name": "prgs-author",
+                 "role": "author",
+                 "allowed_operations": [
+                     "gitea.read", "gitea.issue.create", "gitea.issue.comment",
+                     "gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
+                 ],
+                 "forbidden_operations": [],
+                 "audit_label": "prgs-author",
+             }), patch.object(
+                 mcp_server, "_resolve",
+                 return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
+             ), patch.object(
+                 mcp_server, "_auth", return_value="token fake",
+             ), patch.object(
+                 mcp_server.role_session_router,
+                 "check_author_mutation_after_reviewer_stop",
+                 return_value=(True, []),
+             ), patch.object(
+                 mcp_server, "_namespace_mutation_block", return_value=None
+             ), patch.object(
+                 mcp_server, "_profile_permission_block", return_value=None
+             ), patch.object(
+                 mcp_server, "_production_guard_block_from_exc", return_value=None
+             ):
+            try:
+                fn(*args, **kwargs)
+            except RuntimeError as exc:
+                if "capture-only" not in str(exc):
+                    # Some tools re-raise; still require capture when preflight ran.
+                    if not captured:
+                        raise
+            self.assertTrue(captured, f"{fn_name} never called verify_preflight_purity")
+            self.assertEqual(captured.get("org"), "Scaled-Tech-Consulting", fn_name)
+            self.assertEqual(captured.get("repo"), "Gitea-Tools", fn_name)
+
+    def test_create_issue_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_create_issue",
+            "title",
+            body="body with acceptance criteria for gate",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/fake-wt",
+        )
+
+    def test_create_pr_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_create_pr",
+            title="t",
+            head="fix/issue-735-x",
+            base="master",
+            body="Closes #735",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/fake-wt",
+        )
+
+    def test_mark_issue_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_mark_issue",
+            issue_number=735,
+            action="start",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path="/tmp/fake-wt",
+        )
+
+    def test_commit_files_forwards_explicit_org_repo(self):
+        self._assert_call_forwards(
+            "gitea_commit_files",
+            files=[{"operation": "update", "path": "x", "content_plain": "y"}],
+            message="m",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )