From 67e34baece39c225097c18cff5bc79d1afa8ee58 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 14:47:56 -0400 Subject: [PATCH 1/2] feat: align claim/lock gates with branches-only worktrees (#275) Allow issue lock from base-equivalent branches/ worktrees instead of requiring the literal master/main branch name. Runtime preflight and mark_issue/lock_issue now inspect the declared active task workspace so dirty control-checkout state does not block clean task worktrees. Closes #275 Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 115 ++++++++++++++++++++++++-- issue_lock_worktree.py | 95 +++++++++++++++++---- task_capability_map.py | 6 +- tests/test_issue_lock_worktree.py | 29 ++++++- tests/test_mcp_server.py | 68 +++++++++++++-- tests/test_resolve_task_capability.py | 11 +++ 6 files changed, 292 insertions(+), 32 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd1480c..c8e9cb1 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -171,6 +171,9 @@ _preflight_whoami_violation_files: list[str] = [] _preflight_capability_violation_files: list[str] = [] _preflight_reviewer_violation_files: list[str] = [] +ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE" +AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" + def _preflight_in_test_mode() -> bool: return "pytest" in sys.modules or "unittest" in sys.modules @@ -184,19 +187,47 @@ def _ensure_process_start_porcelain() -> str: return _process_start_porcelain -def _get_workspace_porcelain() -> str: +def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: + """Resolve the workspace root inspected by pre-flight guards.""" + path = (worktree_path or "").strip() + if not path: + path = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip() + if not path: + path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip() + if not path: + path = PROJECT_ROOT + return os.path.realpath(os.path.abspath(path)) + + +def _get_git_root(path: str) -> str | None: + try: + res = subprocess.run( + ["git", "-C", path, "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return None + if res.returncode != 0: + return None + return (res.stdout or "").strip() or None + + +def _get_workspace_porcelain(worktree_path: str | None = None) -> str: """Return tracked-workspace porcelain for pre-flight attribution.""" if os.environ.get("GITEA_TEST_FORCE_DIRTY"): return " M __gitea_test_force_dirty__.py\n" override = os.environ.get("GITEA_TEST_PORCELAIN") if override is not None: return override + workspace = _resolve_preflight_workspace_path(worktree_path) try: res = subprocess.run( ["git", "status", "--porcelain"], capture_output=True, text=True, - cwd=PROJECT_ROOT, + cwd=workspace, ) return res.stdout or "" except Exception: @@ -234,7 +265,35 @@ def _format_preflight_files(files: list[str]) -> str: return ", ".join(files) -def assess_preflight_status() -> dict: +def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict: + workspace = _resolve_preflight_workspace_path(worktree_path) + inspected_root = _get_git_root(workspace) + control_root = os.path.realpath(PROJECT_ROOT) + active_root = os.path.realpath(inspected_root or workspace) + if active_root == control_root: + dirty_scope = "control checkout" + else: + dirty_scope = "active task workspace" + return { + "mcp_server_process_root": control_root, + "active_task_workspace_root": active_root, + "inspected_git_root": inspected_root, + "dirty_files": list(dirty_files), + "dirty_scope": dirty_scope, + } + + +def _format_preflight_workspace_details(details: dict) -> str: + return ( + f"MCP server process root: {details.get('mcp_server_process_root')}; " + f"active task workspace root: {details.get('active_task_workspace_root')}; " + f"inspected git root: {details.get('inspected_git_root')}; " + f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; " + f"dirty scope: {details.get('dirty_scope')}" + ) + + +def assess_preflight_status(worktree_path: str | None = None) -> dict: """Non-throwing pre-flight readiness for runtime-context alignment (#252).""" reasons: list[str] = [] if not _preflight_whoami_called: @@ -245,6 +304,25 @@ def assess_preflight_status() -> dict: reasons.append( "Task capability (gitea_resolve_task_capability) has not been resolved" ) + workspace_details = None + if worktree_path: + dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path))) + workspace_details = _preflight_workspace_details(worktree_path, dirty_files) + if dirty_files: + reasons.append( + "Active task workspace has tracked file edits before mutation " + f"({_format_preflight_workspace_details(workspace_details)})" + ) + return { + "preflight_ready": not reasons, + "preflight_block_reasons": reasons, + "preflight_whoami_verified": _preflight_whoami_called, + "preflight_capability_resolved": _preflight_capability_called, + "preflight_whoami_violation_files": [], + "preflight_capability_violation_files": [], + "preflight_reviewer_violation_files": [], + "preflight_workspace": workspace_details, + } if _preflight_whoami_violation: reasons.append( "Workspace file edits occurred before gitea_whoami verification " @@ -268,6 +346,7 @@ def assess_preflight_status() -> dict: "preflight_whoami_violation_files": list(_preflight_whoami_violation_files), "preflight_capability_violation_files": list(_preflight_capability_violation_files), "preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files), + "preflight_workspace": _preflight_workspace_details(None, []), } @@ -308,7 +387,7 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): _preflight_resolved_role = resolved_role -def verify_preflight_purity(remote: str | None = None): +def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None): """Verify that identity and capability were verified prior to session edits.""" global _preflight_reviewer_violation_files @@ -328,6 +407,17 @@ def verify_preflight_purity(remote: str | None = None): "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" ) + if worktree_path: + dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path))) + if dirty_files: + details = _preflight_workspace_details(worktree_path, dirty_files) + raise RuntimeError( + "Pre-flight order violation: Active task workspace has tracked " + "file edits before mutation (fail closed). " + f"{_format_preflight_workspace_details(details)}" + ) + return + if _preflight_whoami_violation: raise RuntimeError( "Pre-flight order violation: Workspace file edits occurred before " @@ -818,14 +908,23 @@ def gitea_lock_issue( f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)" ) + blocked = _profile_permission_block( + task_capability_map.required_permission("lock_issue")) + if blocked: + return blocked + resolved_worktree = issue_lock_worktree.resolve_author_worktree_path( worktree_path, PROJECT_ROOT ) git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree) + verify_preflight_purity(remote, worktree_path=resolved_worktree) lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( worktree_path=resolved_worktree, current_branch=git_state.get("current_branch"), porcelain_status=git_state.get("porcelain_status") or "", + base_equivalent=git_state.get("base_equivalent"), + inspected_git_root=git_state.get("inspected_git_root"), + base_branch=git_state.get("base_branch"), ) if lock_assessment["block"]: raise RuntimeError( @@ -4191,6 +4290,7 @@ def _build_runtime_task_capabilities( def gitea_get_runtime_context( remote: str = "dadeschools", host: str | None = None, + worktree_path: str | None = None, ) -> dict: """Read-only: explicit visibility into active profile, configuration model, and eligibility. @@ -4278,7 +4378,7 @@ def gitea_get_runtime_context( allowed, forbidden, config ) - preflight = assess_preflight_status() + preflight = assess_preflight_status(worktree_path) if not preflight["preflight_ready"]: safe_next_action = ( "Complete pre-flight verification before mutating: call gitea_whoami, then " @@ -4302,6 +4402,7 @@ def gitea_get_runtime_context( "safe_next_action": safe_next_action, "preflight_ready": preflight["preflight_ready"], "preflight_block_reasons": preflight["preflight_block_reasons"], + "preflight_workspace": preflight.get("preflight_workspace"), "session_capabilities": session_capabilities, } @@ -4553,6 +4654,7 @@ def gitea_mark_issue( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Claim or release an issue via the status:in-progress label. @@ -4565,6 +4667,7 @@ def gitea_mark_issue( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + worktree_path: Active task worktree to inspect for pre-flight purity. Returns: dict with 'success' boolean and 'message'. @@ -4576,7 +4679,7 @@ def gitea_mark_issue( task_capability_map.required_permission("mark_issue")) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, worktree_path=worktree_path) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index 219c6b9..5d9dc4a 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -14,7 +14,7 @@ import subprocess from reviewer_worktree import parse_dirty_tracked_files AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" -BASE_BRANCHES = frozenset({"master", "main"}) +BASE_BRANCHES = frozenset({"master", "main", "dev"}) def resolve_author_worktree_path( @@ -50,9 +50,28 @@ def read_worktree_git_state(worktree_path: str) -> dict: text=True, check=False, ) + root_res = subprocess.run( + ["git", "-C", path, "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=False, + ) + head_res = subprocess.run( + ["git", "-C", path, "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=False, + ) + head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None + base_branch, base_sha = _find_matching_base_ref(path, head_sha) return { "current_branch": current_branch, "porcelain_status": status_res.stdout or "", + "inspected_git_root": (root_res.stdout or "").strip() if root_res.returncode == 0 else None, + "head_sha": head_sha, + "base_branch": base_branch, + "base_sha": base_sha, + "base_equivalent": bool(head_sha and base_sha and head_sha == base_sha), } @@ -61,6 +80,9 @@ def assess_issue_lock_worktree( worktree_path: str, current_branch: str | None, porcelain_status: str, + base_equivalent: bool | None = None, + inspected_git_root: str | None = None, + base_branch: str | None = None, base_branches: frozenset[str] | None = None, ) -> dict: """Fail closed when lock preconditions are not met on the declared worktree.""" @@ -77,21 +99,39 @@ def assess_issue_lock_worktree( if dirty_files: reasons.append( "tracked file edits exist before issue lock; " - "lock must precede implementation work" - ) - if not branch: - reasons.append( - "current branch unknown (detached HEAD?); issue lock must be taken " - f"from base branch ({_base_list(bases)})" - ) - elif branch not in bases: - reasons.append( - f"issue lock must be taken from base branch ({_base_list(bases)}), " - f"not '{branch}'" + f"lock must precede implementation work in '{path}' " + f"(dirty files: {', '.join(dirty_files)})" ) + if base_equivalent is False: + reasons.append( + "issue lock worktree must be base-equivalent to one of " + f"{_base_list(bases)} before implementation work; inspected " + f"branch '{branch or '(detached)'}' at '{path}'" + ) + elif base_equivalent is None: + if not branch: + reasons.append( + "current branch unknown (detached HEAD?); issue lock base-equivalence " + f"to {_base_list(bases)} could not be proven" + ) + elif branch not in bases: + reasons.append( + "issue lock worktree base-equivalence could not be proven; " + f"branch '{branch}' is not {_base_list(bases)}" + ) + proven = not reasons - return _assessment(proven, reasons, path, branch or None, dirty_files) + return _assessment( + proven, + reasons, + path, + branch or None, + dirty_files, + inspected_git_root=inspected_git_root, + base_branch=base_branch, + base_equivalent=base_equivalent, + ) def format_issue_lock_worktree_error(assessment: dict) -> str: @@ -145,12 +185,39 @@ def _assessment( worktree_path: str, current_branch: str | None, dirty_files: list[str], + *, + inspected_git_root: str | None = None, + base_branch: str | None = None, + base_equivalent: bool | None = None, ) -> dict: return { "proven": proven, "block": not proven, "reasons": reasons, "worktree_path": worktree_path or None, + "inspected_git_root": inspected_git_root, "current_branch": current_branch, "dirty_files": dirty_files, - } \ No newline at end of file + "base_branch": base_branch, + "base_equivalent": base_equivalent, + } + + +def _find_matching_base_ref(path: str, head_sha: str | None) -> tuple[str | None, str | None]: + """Return the stable branch ref whose commit matches HEAD, if any.""" + if not head_sha: + return None, None + candidates: list[str] = [] + for branch in sorted(BASE_BRANCHES): + candidates.extend((f"origin/{branch}", branch)) + for ref in candidates: + res = subprocess.run( + ["git", "-C", path, "rev-parse", "--verify", ref], + capture_output=True, + text=True, + check=False, + ) + sha = (res.stdout or "").strip() + if res.returncode == 0 and sha == head_sha: + return ref, sha + return None, None diff --git a/task_capability_map.py b/task_capability_map.py index fd3c344..1d91474 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -28,6 +28,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.issue.comment", "role": "author", }, + "lock_issue": { + "permission": "gitea.issue.comment", + "role": "author", + }, "set_issue_labels": { "permission": "gitea.issue.comment", "role": "author", @@ -110,4 +114,4 @@ def required_role(task: str) -> str: def tool_required_permission(tool_name: str) -> str: """Return the operation an issue-mutating tool must gate on.""" - return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name]) \ No newline at end of file + return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name]) diff --git a/tests/test_issue_lock_worktree.py b/tests/test_issue_lock_worktree.py index ae7c027..02347f4 100644 --- a/tests/test_issue_lock_worktree.py +++ b/tests/test_issue_lock_worktree.py @@ -36,7 +36,32 @@ class TestIssueLockWorktreeAssessment(unittest.TestCase): porcelain_status="", ) self.assertFalse(result["proven"]) - self.assertIn("issue lock must be taken from base branch", result["reasons"][0]) + self.assertIn("base-equivalence could not be proven", result["reasons"][0]) + + def test_base_equivalent_feature_branch_passes(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/repo/branches/issue-275", + current_branch="feat/issue-275-claim-lock-branches-worktree", + porcelain_status="", + base_equivalent=True, + inspected_git_root="/repo/branches/issue-275", + base_branch="origin/master", + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + self.assertEqual(result["base_branch"], "origin/master") + + def test_non_base_equivalent_branch_fails(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/repo/branches/issue-275", + current_branch="feat/issue-275-claim-lock-branches-worktree", + porcelain_status="", + base_equivalent=False, + inspected_git_root="/repo/branches/issue-275", + base_branch=None, + ) + self.assertFalse(result["proven"]) + self.assertIn("must be base-equivalent", result["reasons"][0]) def test_untracked_files_do_not_block(self): result = issue_lock_worktree.assess_issue_lock_worktree( @@ -96,4 +121,4 @@ class TestPrWorktreeMatch(unittest.TestCase): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 133e515..d09f6da 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2931,20 +2931,31 @@ class TestVerifyMutationAuthority(unittest.TestCase): mcp_server.verify_mutation_authority("prgs") +def _clean_master_git_state_for_lock(): + return { + "current_branch": "master", + "porcelain_status": "", + "base_equivalent": True, + "inspected_git_root": "/scratch/wt", + "base_branch": "origin/master", + } + + class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" - @staticmethod - def _clean_master_git_state(): - return {"current_branch": "master", "porcelain_status": ""} + def setUp(self): + self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True) + self._env_patcher.start() def tearDown(self): + self._env_patcher.stop() if os.path.exists(ISSUE_LOCK_FILE): os.remove(ISSUE_LOCK_FILE) @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2964,7 +2975,7 @@ class TestIssueLocking(unittest.TestCase): @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2981,7 +2992,7 @@ class TestIssueLocking(unittest.TestCase): @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -3002,7 +3013,7 @@ class TestIssueLocking(unittest.TestCase): scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean" with patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) as mock_git: res = gitea_lock_issue( issue_number=249, @@ -3022,6 +3033,7 @@ class TestIssueLocking(unittest.TestCase): return_value={ "current_branch": "master", "porcelain_status": " M gitea_mcp_server.py\n", + "base_equivalent": True, }, ): with self.assertRaises(RuntimeError) as ctx: @@ -3041,6 +3053,7 @@ class TestIssueLocking(unittest.TestCase): return_value={ "current_branch": "feat/issue-243-forbidden-git-gaps", "porcelain_status": "", + "base_equivalent": False, }, ): with self.assertRaises(RuntimeError) as ctx: @@ -3050,7 +3063,7 @@ class TestIssueLocking(unittest.TestCase): remote="prgs", worktree_path="/tmp/scratch/wt", ) - self.assertIn("issue lock must be taken from base branch", str(ctx.exception)) + self.assertIn("base-equivalent", str(ctx.exception)) @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @@ -3193,7 +3206,12 @@ class TestPreflightVerification(unittest.TestCase): mcp_server._preflight_whoami_violation_files = self.orig_whoami_files mcp_server._preflight_capability_violation_files = self.orig_capability_files mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files - for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"): + for key in ( + "GITEA_TEST_FORCE_DIRTY", + "GITEA_TEST_PORCELAIN", + "GITEA_ACTIVE_WORKTREE", + "GITEA_AUTHOR_WORKTREE", + ): if key in os.environ: del os.environ[key] @@ -3283,3 +3301,35 @@ class TestPreflightVerification(unittest.TestCase): status = mcp_server.assess_preflight_status() self.assertFalse(status["preflight_ready"]) self.assertIn("gitea_whoami", status["preflight_block_reasons"][0]) + + def test_declared_clean_task_worktree_ignores_control_checkout_violation(self): + """#275: active branches/ worktree is the inspected mutation workspace.""" + import mcp_server + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + mcp_server._preflight_whoami_violation = True + mcp_server._preflight_whoami_violation_files = ["mcp_server.py"] + os.environ["GITEA_TEST_PORCELAIN"] = "" + + worktree = "/repo/branches/issue-275-clean" + status = mcp_server.assess_preflight_status(worktree_path=worktree) + self.assertTrue(status["preflight_ready"]) + self.assertEqual(status["preflight_block_reasons"], []) + self.assertIn("preflight_workspace", status) + mcp_server.verify_preflight_purity(worktree_path=worktree) + + def test_declared_dirty_task_worktree_reports_workspace_diagnostics(self): + """#275: dirty failures name the inspected task workspace, not just files.""" + import mcp_server + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + os.environ["GITEA_TEST_PORCELAIN"] = " M task_file.py\n" + + worktree = "/repo/branches/issue-275-dirty" + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity(worktree_path=worktree) + msg = str(ctx.exception) + self.assertIn("active task workspace root", msg) + self.assertIn("inspected git root", msg) + self.assertIn("dirty files: task_file.py", msg) + self.assertIn("dirty scope:", msg) diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 463a618..3545675 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -141,6 +141,17 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertTrue(res["allowed_in_current_session"]) self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api): + # #275: lock_issue must be an exact resolver task for claim/lock gates. + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs") + self.assertEqual(res["requested_task"], "lock_issue") + self.assertEqual(res["required_operation_permission"], "gitea.issue.comment") + self.assertTrue(res["allowed_in_current_session"]) + self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) + def test_resolve_unknown_task_fails_closed(self): with patch.dict(os.environ, self._env("author-profile")): with self.assertRaises(ValueError): From 5e64c968512a1278e09c948cf83faaa59f7979a1 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 14:47:56 -0400 Subject: [PATCH 2/2] feat: align claim/lock gates with branches-only worktrees (#275) Allow issue lock from base-equivalent branches/ worktrees instead of requiring the literal master/main branch name. Runtime preflight and mark_issue/lock_issue now inspect the declared active task workspace so dirty control-checkout state does not block clean task worktrees. Closes #275 Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 119 ++++++++++++++++++++++++-- issue_lock_worktree.py | 95 +++++++++++++++++--- task_capability_map.py | 6 +- tests/test_issue_lock_worktree.py | 29 ++++++- tests/test_mcp_server.py | 68 +++++++++++++-- tests/test_resolve_task_capability.py | 11 +++ 6 files changed, 294 insertions(+), 34 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd1480c..fec1088 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -171,6 +171,9 @@ _preflight_whoami_violation_files: list[str] = [] _preflight_capability_violation_files: list[str] = [] _preflight_reviewer_violation_files: list[str] = [] +ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE" +AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" + def _preflight_in_test_mode() -> bool: return "pytest" in sys.modules or "unittest" in sys.modules @@ -184,19 +187,47 @@ def _ensure_process_start_porcelain() -> str: return _process_start_porcelain -def _get_workspace_porcelain() -> str: +def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: + """Resolve the workspace root inspected by pre-flight guards.""" + path = (worktree_path or "").strip() + if not path: + path = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip() + if not path: + path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip() + if not path: + path = PROJECT_ROOT + return os.path.realpath(os.path.abspath(path)) + + +def _get_git_root(path: str) -> str | None: + try: + res = subprocess.run( + ["git", "-C", path, "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return None + if res.returncode != 0: + return None + return (res.stdout or "").strip() or None + + +def _get_workspace_porcelain(worktree_path: str | None = None) -> str: """Return tracked-workspace porcelain for pre-flight attribution.""" if os.environ.get("GITEA_TEST_FORCE_DIRTY"): return " M __gitea_test_force_dirty__.py\n" override = os.environ.get("GITEA_TEST_PORCELAIN") if override is not None: return override + workspace = _resolve_preflight_workspace_path(worktree_path) try: res = subprocess.run( ["git", "status", "--porcelain"], capture_output=True, text=True, - cwd=PROJECT_ROOT, + cwd=workspace, ) return res.stdout or "" except Exception: @@ -234,7 +265,35 @@ def _format_preflight_files(files: list[str]) -> str: return ", ".join(files) -def assess_preflight_status() -> dict: +def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict: + workspace = _resolve_preflight_workspace_path(worktree_path) + inspected_root = _get_git_root(workspace) + control_root = os.path.realpath(PROJECT_ROOT) + active_root = os.path.realpath(inspected_root or workspace) + if active_root == control_root: + dirty_scope = "control checkout" + else: + dirty_scope = "active task workspace" + return { + "mcp_server_process_root": control_root, + "active_task_workspace_root": active_root, + "inspected_git_root": inspected_root, + "dirty_files": list(dirty_files), + "dirty_scope": dirty_scope, + } + + +def _format_preflight_workspace_details(details: dict) -> str: + return ( + f"MCP server process root: {details.get('mcp_server_process_root')}; " + f"active task workspace root: {details.get('active_task_workspace_root')}; " + f"inspected git root: {details.get('inspected_git_root')}; " + f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; " + f"dirty scope: {details.get('dirty_scope')}" + ) + + +def assess_preflight_status(worktree_path: str | None = None) -> dict: """Non-throwing pre-flight readiness for runtime-context alignment (#252).""" reasons: list[str] = [] if not _preflight_whoami_called: @@ -245,6 +304,25 @@ def assess_preflight_status() -> dict: reasons.append( "Task capability (gitea_resolve_task_capability) has not been resolved" ) + workspace_details = None + if worktree_path: + dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path))) + workspace_details = _preflight_workspace_details(worktree_path, dirty_files) + if dirty_files: + reasons.append( + "Active task workspace has tracked file edits before mutation " + f"({_format_preflight_workspace_details(workspace_details)})" + ) + return { + "preflight_ready": not reasons, + "preflight_block_reasons": reasons, + "preflight_whoami_verified": _preflight_whoami_called, + "preflight_capability_resolved": _preflight_capability_called, + "preflight_whoami_violation_files": [], + "preflight_capability_violation_files": [], + "preflight_reviewer_violation_files": [], + "preflight_workspace": workspace_details, + } if _preflight_whoami_violation: reasons.append( "Workspace file edits occurred before gitea_whoami verification " @@ -268,6 +346,7 @@ def assess_preflight_status() -> dict: "preflight_whoami_violation_files": list(_preflight_whoami_violation_files), "preflight_capability_violation_files": list(_preflight_capability_violation_files), "preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files), + "preflight_workspace": _preflight_workspace_details(None, []), } @@ -308,7 +387,7 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): _preflight_resolved_role = resolved_role -def verify_preflight_purity(remote: str | None = None): +def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None): """Verify that identity and capability were verified prior to session edits.""" global _preflight_reviewer_violation_files @@ -328,6 +407,17 @@ def verify_preflight_purity(remote: str | None = None): "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" ) + if worktree_path: + dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path))) + if dirty_files: + details = _preflight_workspace_details(worktree_path, dirty_files) + raise RuntimeError( + "Pre-flight order violation: Active task workspace has tracked " + "file edits before mutation (fail closed). " + f"{_format_preflight_workspace_details(details)}" + ) + return + if _preflight_whoami_violation: raise RuntimeError( "Pre-flight order violation: Workspace file edits occurred before " @@ -728,6 +818,8 @@ def gitea_create_issue( Returns: dict with 'number' of the created issue ('url' only with the reveal opt-in). """ + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( "create_issue" ) @@ -749,8 +841,6 @@ def gitea_create_issue( if blocked: return blocked verify_preflight_purity(remote) - h, o, r = _resolve(remote, host, org, repo) - auth = _auth(h) base = repo_api_url(h, o, r) open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth) closed_issues = api_get_all( @@ -818,14 +908,23 @@ def gitea_lock_issue( f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)" ) + blocked = _profile_permission_block( + task_capability_map.required_permission("lock_issue")) + if blocked: + return blocked + resolved_worktree = issue_lock_worktree.resolve_author_worktree_path( worktree_path, PROJECT_ROOT ) git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree) + verify_preflight_purity(remote, worktree_path=resolved_worktree) lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( worktree_path=resolved_worktree, current_branch=git_state.get("current_branch"), porcelain_status=git_state.get("porcelain_status") or "", + base_equivalent=git_state.get("base_equivalent"), + inspected_git_root=git_state.get("inspected_git_root"), + base_branch=git_state.get("base_branch"), ) if lock_assessment["block"]: raise RuntimeError( @@ -4191,6 +4290,7 @@ def _build_runtime_task_capabilities( def gitea_get_runtime_context( remote: str = "dadeschools", host: str | None = None, + worktree_path: str | None = None, ) -> dict: """Read-only: explicit visibility into active profile, configuration model, and eligibility. @@ -4278,7 +4378,7 @@ def gitea_get_runtime_context( allowed, forbidden, config ) - preflight = assess_preflight_status() + preflight = assess_preflight_status(worktree_path) if not preflight["preflight_ready"]: safe_next_action = ( "Complete pre-flight verification before mutating: call gitea_whoami, then " @@ -4302,6 +4402,7 @@ def gitea_get_runtime_context( "safe_next_action": safe_next_action, "preflight_ready": preflight["preflight_ready"], "preflight_block_reasons": preflight["preflight_block_reasons"], + "preflight_workspace": preflight.get("preflight_workspace"), "session_capabilities": session_capabilities, } @@ -4553,6 +4654,7 @@ def gitea_mark_issue( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Claim or release an issue via the status:in-progress label. @@ -4565,6 +4667,7 @@ def gitea_mark_issue( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + worktree_path: Active task worktree to inspect for pre-flight purity. Returns: dict with 'success' boolean and 'message'. @@ -4576,7 +4679,7 @@ def gitea_mark_issue( task_capability_map.required_permission("mark_issue")) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, worktree_path=worktree_path) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index 219c6b9..5d9dc4a 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -14,7 +14,7 @@ import subprocess from reviewer_worktree import parse_dirty_tracked_files AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" -BASE_BRANCHES = frozenset({"master", "main"}) +BASE_BRANCHES = frozenset({"master", "main", "dev"}) def resolve_author_worktree_path( @@ -50,9 +50,28 @@ def read_worktree_git_state(worktree_path: str) -> dict: text=True, check=False, ) + root_res = subprocess.run( + ["git", "-C", path, "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=False, + ) + head_res = subprocess.run( + ["git", "-C", path, "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=False, + ) + head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None + base_branch, base_sha = _find_matching_base_ref(path, head_sha) return { "current_branch": current_branch, "porcelain_status": status_res.stdout or "", + "inspected_git_root": (root_res.stdout or "").strip() if root_res.returncode == 0 else None, + "head_sha": head_sha, + "base_branch": base_branch, + "base_sha": base_sha, + "base_equivalent": bool(head_sha and base_sha and head_sha == base_sha), } @@ -61,6 +80,9 @@ def assess_issue_lock_worktree( worktree_path: str, current_branch: str | None, porcelain_status: str, + base_equivalent: bool | None = None, + inspected_git_root: str | None = None, + base_branch: str | None = None, base_branches: frozenset[str] | None = None, ) -> dict: """Fail closed when lock preconditions are not met on the declared worktree.""" @@ -77,21 +99,39 @@ def assess_issue_lock_worktree( if dirty_files: reasons.append( "tracked file edits exist before issue lock; " - "lock must precede implementation work" - ) - if not branch: - reasons.append( - "current branch unknown (detached HEAD?); issue lock must be taken " - f"from base branch ({_base_list(bases)})" - ) - elif branch not in bases: - reasons.append( - f"issue lock must be taken from base branch ({_base_list(bases)}), " - f"not '{branch}'" + f"lock must precede implementation work in '{path}' " + f"(dirty files: {', '.join(dirty_files)})" ) + if base_equivalent is False: + reasons.append( + "issue lock worktree must be base-equivalent to one of " + f"{_base_list(bases)} before implementation work; inspected " + f"branch '{branch or '(detached)'}' at '{path}'" + ) + elif base_equivalent is None: + if not branch: + reasons.append( + "current branch unknown (detached HEAD?); issue lock base-equivalence " + f"to {_base_list(bases)} could not be proven" + ) + elif branch not in bases: + reasons.append( + "issue lock worktree base-equivalence could not be proven; " + f"branch '{branch}' is not {_base_list(bases)}" + ) + proven = not reasons - return _assessment(proven, reasons, path, branch or None, dirty_files) + return _assessment( + proven, + reasons, + path, + branch or None, + dirty_files, + inspected_git_root=inspected_git_root, + base_branch=base_branch, + base_equivalent=base_equivalent, + ) def format_issue_lock_worktree_error(assessment: dict) -> str: @@ -145,12 +185,39 @@ def _assessment( worktree_path: str, current_branch: str | None, dirty_files: list[str], + *, + inspected_git_root: str | None = None, + base_branch: str | None = None, + base_equivalent: bool | None = None, ) -> dict: return { "proven": proven, "block": not proven, "reasons": reasons, "worktree_path": worktree_path or None, + "inspected_git_root": inspected_git_root, "current_branch": current_branch, "dirty_files": dirty_files, - } \ No newline at end of file + "base_branch": base_branch, + "base_equivalent": base_equivalent, + } + + +def _find_matching_base_ref(path: str, head_sha: str | None) -> tuple[str | None, str | None]: + """Return the stable branch ref whose commit matches HEAD, if any.""" + if not head_sha: + return None, None + candidates: list[str] = [] + for branch in sorted(BASE_BRANCHES): + candidates.extend((f"origin/{branch}", branch)) + for ref in candidates: + res = subprocess.run( + ["git", "-C", path, "rev-parse", "--verify", ref], + capture_output=True, + text=True, + check=False, + ) + sha = (res.stdout or "").strip() + if res.returncode == 0 and sha == head_sha: + return ref, sha + return None, None diff --git a/task_capability_map.py b/task_capability_map.py index fd3c344..1d91474 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -28,6 +28,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.issue.comment", "role": "author", }, + "lock_issue": { + "permission": "gitea.issue.comment", + "role": "author", + }, "set_issue_labels": { "permission": "gitea.issue.comment", "role": "author", @@ -110,4 +114,4 @@ def required_role(task: str) -> str: def tool_required_permission(tool_name: str) -> str: """Return the operation an issue-mutating tool must gate on.""" - return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name]) \ No newline at end of file + return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name]) diff --git a/tests/test_issue_lock_worktree.py b/tests/test_issue_lock_worktree.py index ae7c027..02347f4 100644 --- a/tests/test_issue_lock_worktree.py +++ b/tests/test_issue_lock_worktree.py @@ -36,7 +36,32 @@ class TestIssueLockWorktreeAssessment(unittest.TestCase): porcelain_status="", ) self.assertFalse(result["proven"]) - self.assertIn("issue lock must be taken from base branch", result["reasons"][0]) + self.assertIn("base-equivalence could not be proven", result["reasons"][0]) + + def test_base_equivalent_feature_branch_passes(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/repo/branches/issue-275", + current_branch="feat/issue-275-claim-lock-branches-worktree", + porcelain_status="", + base_equivalent=True, + inspected_git_root="/repo/branches/issue-275", + base_branch="origin/master", + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + self.assertEqual(result["base_branch"], "origin/master") + + def test_non_base_equivalent_branch_fails(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/repo/branches/issue-275", + current_branch="feat/issue-275-claim-lock-branches-worktree", + porcelain_status="", + base_equivalent=False, + inspected_git_root="/repo/branches/issue-275", + base_branch=None, + ) + self.assertFalse(result["proven"]) + self.assertIn("must be base-equivalent", result["reasons"][0]) def test_untracked_files_do_not_block(self): result = issue_lock_worktree.assess_issue_lock_worktree( @@ -96,4 +121,4 @@ class TestPrWorktreeMatch(unittest.TestCase): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 133e515..d09f6da 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2931,20 +2931,31 @@ class TestVerifyMutationAuthority(unittest.TestCase): mcp_server.verify_mutation_authority("prgs") +def _clean_master_git_state_for_lock(): + return { + "current_branch": "master", + "porcelain_status": "", + "base_equivalent": True, + "inspected_git_root": "/scratch/wt", + "base_branch": "origin/master", + } + + class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" - @staticmethod - def _clean_master_git_state(): - return {"current_branch": "master", "porcelain_status": ""} + def setUp(self): + self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True) + self._env_patcher.start() def tearDown(self): + self._env_patcher.stop() if os.path.exists(ISSUE_LOCK_FILE): os.remove(ISSUE_LOCK_FILE) @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2964,7 +2975,7 @@ class TestIssueLocking(unittest.TestCase): @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2981,7 +2992,7 @@ class TestIssueLocking(unittest.TestCase): @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -3002,7 +3013,7 @@ class TestIssueLocking(unittest.TestCase): scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean" with patch( "mcp_server.issue_lock_worktree.read_worktree_git_state", - return_value={"current_branch": "master", "porcelain_status": ""}, + return_value=_clean_master_git_state_for_lock(), ) as mock_git: res = gitea_lock_issue( issue_number=249, @@ -3022,6 +3033,7 @@ class TestIssueLocking(unittest.TestCase): return_value={ "current_branch": "master", "porcelain_status": " M gitea_mcp_server.py\n", + "base_equivalent": True, }, ): with self.assertRaises(RuntimeError) as ctx: @@ -3041,6 +3053,7 @@ class TestIssueLocking(unittest.TestCase): return_value={ "current_branch": "feat/issue-243-forbidden-git-gaps", "porcelain_status": "", + "base_equivalent": False, }, ): with self.assertRaises(RuntimeError) as ctx: @@ -3050,7 +3063,7 @@ class TestIssueLocking(unittest.TestCase): remote="prgs", worktree_path="/tmp/scratch/wt", ) - self.assertIn("issue lock must be taken from base branch", str(ctx.exception)) + self.assertIn("base-equivalent", str(ctx.exception)) @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @@ -3193,7 +3206,12 @@ class TestPreflightVerification(unittest.TestCase): mcp_server._preflight_whoami_violation_files = self.orig_whoami_files mcp_server._preflight_capability_violation_files = self.orig_capability_files mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files - for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"): + for key in ( + "GITEA_TEST_FORCE_DIRTY", + "GITEA_TEST_PORCELAIN", + "GITEA_ACTIVE_WORKTREE", + "GITEA_AUTHOR_WORKTREE", + ): if key in os.environ: del os.environ[key] @@ -3283,3 +3301,35 @@ class TestPreflightVerification(unittest.TestCase): status = mcp_server.assess_preflight_status() self.assertFalse(status["preflight_ready"]) self.assertIn("gitea_whoami", status["preflight_block_reasons"][0]) + + def test_declared_clean_task_worktree_ignores_control_checkout_violation(self): + """#275: active branches/ worktree is the inspected mutation workspace.""" + import mcp_server + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + mcp_server._preflight_whoami_violation = True + mcp_server._preflight_whoami_violation_files = ["mcp_server.py"] + os.environ["GITEA_TEST_PORCELAIN"] = "" + + worktree = "/repo/branches/issue-275-clean" + status = mcp_server.assess_preflight_status(worktree_path=worktree) + self.assertTrue(status["preflight_ready"]) + self.assertEqual(status["preflight_block_reasons"], []) + self.assertIn("preflight_workspace", status) + mcp_server.verify_preflight_purity(worktree_path=worktree) + + def test_declared_dirty_task_worktree_reports_workspace_diagnostics(self): + """#275: dirty failures name the inspected task workspace, not just files.""" + import mcp_server + mcp_server._preflight_whoami_called = True + mcp_server._preflight_capability_called = True + os.environ["GITEA_TEST_PORCELAIN"] = " M task_file.py\n" + + worktree = "/repo/branches/issue-275-dirty" + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity(worktree_path=worktree) + msg = str(ctx.exception) + self.assertIn("active task workspace root", msg) + self.assertIn("inspected git root", msg) + self.assertIn("dirty files: task_file.py", msg) + self.assertIn("dirty scope:", msg) diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 463a618..3545675 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -141,6 +141,17 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertTrue(res["allowed_in_current_session"]) self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api): + # #275: lock_issue must be an exact resolver task for claim/lock gates. + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs") + self.assertEqual(res["requested_task"], "lock_issue") + self.assertEqual(res["required_operation_permission"], "gitea.issue.comment") + self.assertTrue(res["allowed_in_current_session"]) + self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"]) + def test_resolve_unknown_task_fails_closed(self): with patch.dict(os.environ, self._env("author-profile")): with self.assertRaises(ValueError):