Merge prgs/master into feat/issue-266-subagent-gate-inheritance

Resolve _GUIDE_RULES conflict by keeping both operator guide keys:
shell_spawn_hard_stop (#258, landed on master via PR #281) and
subagent_delegation (#266, this branch).

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

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()
+    unittest.main()

tests/test_mcp_server.py

@@ -2937,20 +2937,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)
@@ -2970,7 +2981,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)
@@ -2987,7 +2998,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)
@@ -3008,7 +3019,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,
@@ -3028,6 +3039,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:
@@ -3047,6 +3059,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:
@@ -3056,7 +3069,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, []))
@@ -3199,7 +3212,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]
 
@@ -3289,3 +3307,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)

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"])

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):

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}"

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