Merge branch 'master' into feat/issue-271-validation-ledger-verifier

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
@@ -394,6 +421,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).

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

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 "
@@ -278,6 +356,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, []),
     }
 
 
@@ -318,7 +397,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
 
@@ -338,6 +417,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 "
@@ -739,6 +829,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"
     )
@@ -760,8 +852,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(
@@ -829,14 +919,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(
@@ -3629,6 +3728,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 = [
@@ -4337,6 +4446,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.
 
@@ -4424,7 +4534,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 "
@@ -4448,6 +4558,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,
     }
 
@@ -4699,6 +4810,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.
 
@@ -4711,6 +4823,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'.
@@ -4722,7 +4835,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)

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

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

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",
@@ -119,4 +123,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])
+    return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])

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