From 2f4672218c7edb0641a098b8ad4e4b3ed50fc06c Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 14:44:41 -0400 Subject: [PATCH 1/4] docs: forbid improvised commit fallbacks when MCP commit exists (#260) Document MCP-native gitea_commit_files as the only approved commit path when gitea.repo.commit is allowed; ban WebFetch, Playwright, and manual base64 workarounds in runbooks and safety model. Add doc tests. Closes #260. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/llm-workflow-runbooks.md | 30 ++++++++++++++++++++++ docs/safety-model.md | 18 ++++++++++++++ tests/test_workflow_commit_path_docs.py | 33 +++++++++++++++++++++++++ 3 files changed, 81 insertions(+) create mode 100644 tests/test_workflow_commit_path_docs.py diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index 1b04cb3..425ce86 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -370,6 +370,36 @@ git branch -d fix/issue-123-example All three helpers accept `--dry-run` to print the exact commands/paths without touching anything. +### MCP-native commit path (#260) + +When the active author profile allows **`gitea.repo.commit`** and +**`gitea_commit_files`** is visible in the client, that is the **only** approved +path for committing files to the tracked repository. Do not improvise alternate +encoding or transport when MCP commit is available. + +**Required before commit:** + +1. Call `gitea_resolve_task_capability` for `commit_files` or + `gitea_commit_files` and confirm `allowed_in_current_session` is true. +2. Use `gitea_commit_files` with file payloads prepared in the author worktree. +3. Stage only issue-scoped paths; never commit throwaway `_encode_*` / + `_emit_*` / `_inline_*` helpers. + +**Explicitly forbidden workarounds** when MCP commit is reachable: + +- `WebFetch` / HTTP calls to external decode sites (for example httpbin base64 + endpoints) +- Playwright or other browser automation to bypass MCP +- Manual LLM-generated base64 pasted into ad-hoc scripts +- Delegating commit authority to a subagent while the main session has + `gitea.repo.commit` on an author profile + +**If shell encoding is unavailable** (spawn failure, hung terminal) **and** MCP +commit cannot run: **stop** with a recovery report. Mention restarting the +session, clearing hung background terminals, switching to MCP-native commit, and +the agent temp artifact cleanup checklist. Do **not** retry shell encoding in a +loop and do **not** substitute WebFetch/Playwright/manual base64. + ### Create an issue / child issues - **Profile:** issue-manager or author (any profile allowed to create issues). diff --git a/docs/safety-model.md b/docs/safety-model.md index 1f80b51..31c740a 100644 --- a/docs/safety-model.md +++ b/docs/safety-model.md @@ -28,3 +28,21 @@ Any mutating action (e.g., Gitea issue creation from GlitchTip, or Jenkins build No default profile carries trigger capability (#152 / mcp-control-plane #56). - **GlitchTip to Gitea issue filing** is a library-only orchestrator in mcp-control-plane (not on `glitchtip-mcp`). See #153 / mcp-control-plane #57. + +## 6. Agent Commit Path (no improvised fallbacks) + +When an author execution profile allows **`gitea.repo.commit`** and the +**`gitea_commit_files`** tool is visible, agents must use that MCP path for +repository commits. Fail closed instead of improvising alternate transports. + +Forbidden when MCP commit is available: + +- WebFetch or other HTTP calls to external base64/decode services +- Playwright or browser automation used to work around MCP commit +- Manual LLM-generated base64 embedded in throwaway scripts as the primary + commit transport + +If shell helpers are unavailable and MCP commit cannot run, stop with a recovery +report (restart session, clear hung terminals, use MCP-native commit). See +[`llm-workflow-runbooks.md`](llm-workflow-runbooks.md) § MCP-native commit path +(#260) and agent temp artifact cleanup (#261). diff --git a/tests/test_workflow_commit_path_docs.py b/tests/test_workflow_commit_path_docs.py new file mode 100644 index 0000000..7322127 --- /dev/null +++ b/tests/test_workflow_commit_path_docs.py @@ -0,0 +1,33 @@ +"""Documentation checks for MCP-native commit path rules (#260).""" +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md" +SAFETY = REPO_ROOT / "docs" / "safety-model.md" + + +def test_runbook_requires_mcp_native_commit_path(): + text = RUNBOOK.read_text(encoding="utf-8") + assert "MCP-native commit path" in text + assert "gitea_commit_files" in text + assert "gitea.repo.commit" in text + assert "only" in text.lower() and "approved" in text.lower() + lower = text.lower() + for forbidden in ("webfetch", "playwright", "manual llm-generated base64"): + assert forbidden in lower + + +def test_runbook_forbids_fallback_loops(): + lower = RUNBOOK.read_text(encoding="utf-8").lower() + assert "retry shell encoding" in lower + assert "loop" in lower + assert "recovery report" in lower + + +def test_safety_model_documents_commit_fallback_ban(): + text = SAFETY.read_text(encoding="utf-8") + assert "Agent Commit Path" in text + assert "gitea_commit_files" in text + assert "WebFetch" in text + assert "Playwright" in text + assert "llm-workflow-runbooks.md" in text \ No newline at end of file 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 2/4] 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 3/4] 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): From cf292e41663c48289b44325da69ad64b46baaf47 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 14:52:12 -0400 Subject: [PATCH 4/4] docs: add Shell Spawn Hard-Stop Rule for agent workflows (#258) Documents fail-closed behavior when the shell executor fails to spawn (exit_code: -1, empty stdout/stderr): probe once, mark shell unavailable, hard-stop after two consecutive spawn failures, and emit a recovery report (restart session, kill hung background terminals, prefer MCP-native paths) instead of retry-spiralling. - skills/llm-project-workflow/SKILL.md: portable rule section - docs/llm-workflow-runbooks.md: Gitea runbook section - gitea_mcp_server.py: shell_spawn_hard_stop operator guide rule - tests/test_shell_spawn_hard_stop_docs.py: doc-contract tests (5) - tests/test_operator_guide.py: require the new guide key Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/llm-workflow-runbooks.md | 27 ++++++++ gitea_mcp_server.py | 10 +++ skills/llm-project-workflow/SKILL.md | 20 ++++++ tests/test_operator_guide.py | 3 +- tests/test_shell_spawn_hard_stop_docs.py | 82 ++++++++++++++++++++++++ 5 files changed, 141 insertions(+), 1 deletion(-) create mode 100644 tests/test_shell_spawn_hard_stop_docs.py diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index 1b04cb3..13e3b4d 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -285,6 +285,33 @@ explicit control-checkout repair. Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md). +## Shell Spawn Hard-Stop Rule + +Symptom: a shell tool call returns `exit_code: -1` with empty stdout/stderr. +That is an executor spawn failure, not a command failure — the command never +ran, and retrying the identical call cannot succeed. + +Required behavior (fail closed, issue #258): + +1. **Probe once.** On the first spawn failure, run one trivial probe + (`echo ok` or `pwd`). If the probe also returns `exit_code: -1`, mark + shell unavailable for the session. +2. **Hard-stop at two.** After two consecutive spawn failures, stop all + further shell tool use for the session; never retry the same failing + spawn. A hundred retries produce a hundred identical failures (session + `019f382e`: 100+ tool calls stalled on a trivial encode-and-commit task). +3. **Emit a recovery report.** The report must direct the operator to: + - restart the session, + - kill hung background terminals (a hung test runner holding the + executor is a known contributor), + - prefer MCP-native paths for remaining mutations (for example + `gitea_commit_files` under `gitea.repo.commit`) instead of shell. +4. **No improvised fallbacks.** Shell unavailability never authorizes + WebFetch/browser/manual-encoding workarounds (see #260). No shell means + stop-and-report. + +Doc-contract tests: `tests/test_shell_spawn_hard_stop_docs.py`. + ## Branch worktree isolation All LLM implementation and review work happens in an isolated branch worktree diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd1480c..5d605b7 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3483,6 +3483,16 @@ _GUIDE_RULES = { "small fixes, review fixes, conflicts, emergencies). Main checkout: " "read-only inspect, fetch, create worktrees, post-merge stable " "update, explicit repair only."), + "shell_spawn_hard_stop": ( + "A shell result of exit_code: -1 with empty stdout/stderr is an " + "executor spawn failure, not a command failure. Probe once (echo/pwd); " + "if the probe fails, mark shell unavailable for the session. After " + "two consecutive spawn failures, hard-stop all shell use — never " + "retry the same failing spawn — and emit a recovery report: restart " + "the session, kill hung background terminals, prefer MCP-native " + "paths (e.g. gitea_commit_files) for remaining mutations. Shell " + "unavailability never authorizes WebFetch/browser/manual-encoding " + "fallbacks (#258)."), } _COMMON_WORKFLOWS = [ diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index cfb723a..b95f2a9 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -115,6 +115,26 @@ The main checkout may only be used for read-only inspection, fetching, stable-branch update after merged PRs, creating `branches/` worktrees, or explicit control-checkout repair. +## Shell Spawn Hard-Stop Rule + +A shell tool result of `exit_code: -1` with empty stdout/stderr means the +executor failed to spawn — it is not a command failure, and retrying the same +call cannot succeed. + +1. On the first spawn failure, run one trivial probe (`echo ok` or `pwd`). + If the probe also fails, mark shell unavailable for the session. +2. After two consecutive spawn failures, hard-stop all further shell tool + use for the session. Never retry the same failing spawn. +3. Stop and emit a recovery report instead of improvising fallbacks. The + report must tell the operator to: restart the session, kill hung + background terminals (a hung test runner is a known contributor), and + prefer MCP-native paths (for example `gitea_commit_files`) for any + remaining mutations. + +Retry spirals are a real failure mode (issue #258: 100+ tool calls on a +trivial encode-and-commit task). The hard-stop is fail-closed: no shell means +stop-and-report, not workarounds. + ## B. Isolated worktree rule **Never implement or review in the main checkout** (Global LLM Worktree Rule). diff --git a/tests/test_operator_guide.py b/tests/test_operator_guide.py index 602160a..fd62809 100644 --- a/tests/test_operator_guide.py +++ b/tests/test_operator_guide.py @@ -133,7 +133,8 @@ class TestControlPlaneGuide(GuideTestBase): for key in ("hard_stops", "fail_closed", "head_sha_pinning", "merge_confirmation", "redaction", "separation", "profile_switching", "identity_verification", - "work_selection", "global_worktree"): + "work_selection", "global_worktree", + "shell_spawn_hard_stop"): self.assertIn(key, rules) self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"])) self.assertTrue(rules["hard_stops"]) diff --git a/tests/test_shell_spawn_hard_stop_docs.py b/tests/test_shell_spawn_hard_stop_docs.py new file mode 100644 index 0000000..f4596e3 --- /dev/null +++ b/tests/test_shell_spawn_hard_stop_docs.py @@ -0,0 +1,82 @@ +"""Doc-contract checks for the shell-spawn hard-stop rule (#258). + +When the shell executor fails to spawn (exit_code: -1 with empty +stdout/stderr), agents must probe once, mark shell unavailable, hard-stop +after two consecutive spawn failures, and emit a recovery report instead of +retry-spiralling. These tests pin that guidance in the runbooks, the portable +skill doc, and the operator guide so it cannot silently regress. +""" +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent + +RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md" +SKILL = REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md" + + +def _normalize(text: str) -> str: + """Collapse whitespace so phrase checks survive markdown line wrapping.""" + return " ".join(text.split()) + + +def _runbook_text() -> str: + return _normalize(RUNBOOK.read_text(encoding="utf-8")) + + +def _skill_text() -> str: + return _normalize(SKILL.read_text(encoding="utf-8")) + + +def test_runbook_has_shell_spawn_hard_stop_section(): + text = _runbook_text() + assert "## Shell Spawn Hard-Stop Rule" in text + for phrase in ( + "exit_code: -1", + "empty stdout/stderr", + "trivial probe", + "two consecutive spawn failures", + "mark shell unavailable", + "recovery report", + ): + assert phrase in text, f"runbook missing phrase: {phrase!r}" + + +def test_runbook_recovery_report_names_required_steps(): + text = _runbook_text() + for phrase in ( + "restart the session", + "kill hung background terminals", + "MCP-native", + ): + assert phrase in text, f"runbook recovery guidance missing: {phrase!r}" + + +def test_runbook_forbids_retry_spirals(): + text = _runbook_text() + assert "never retry the same failing spawn" in text + + +def test_skill_doc_declares_shell_spawn_hard_stop_rule(): + text = _skill_text() + assert "## Shell Spawn Hard-Stop Rule" in text + for phrase in ( + "exit_code: -1", + "two consecutive spawn failures", + "recovery report", + ): + assert phrase in text, f"SKILL.md missing phrase: {phrase!r}" + + +def test_operator_guide_rules_include_shell_spawn_hard_stop(): + import sys + + sys.path.insert(0, str(REPO_ROOT)) + import gitea_mcp_server + + rule = gitea_mcp_server._GUIDE_RULES["shell_spawn_hard_stop"] + for phrase in ( + "exit_code: -1", + "two consecutive", + "recovery report", + ): + assert phrase in rule, f"operator guide rule missing: {phrase!r}"