feat: terminal launcher diagnostics and mutation preflight (Closes #556)

Add categorize-and-probe helpers for shell spawn failures, expose
gitea_diagnose_terminal, and fail closed on mutation capability resolve
when the terminal launcher is unhealthy (BLOCKED + DIAGNOSE). Document
the operator path in the workflow runbooks.

docs/llm-workflow-runbooks.md

@@ -661,6 +661,21 @@ 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.
 
+### Terminal launcher diagnostics (#556)
+
+When git/pytest finalization fails with opaque spawn errors (e.g. `os error 2`),
+do **not** improvise shell wrappers or fall back to direct API / temp scripts.
+
+1. Call `gitea_diagnose_terminal` (optional `cwd`, optional `command`) — returns
+   categorized failure: missing cwd, cwd not a directory, missing executable,
+   missing runtime wrapper, missing shell, probe timeout, or session launcher
+   failure.
+2. Call `gitea_get_shell_health` for the shell circuit-breaker state.
+3. Mutation-capable `gitea_resolve_task_capability` probes the terminal launcher
+   before allowing git/pytest-oriented work; on failure it returns
+   `BLOCKED + DIAGNOSE` with `terminal_launcher_unhealthy` and diagnostics.
+4. Emit the canonical `blocked-diagnose-report.md` template and stop.
+
 ### Create an issue / child issues
 
 - **Profile:** issue-manager or author (any profile allowed to create issues).

gitea_mcp_server.py

@@ -7906,6 +7906,36 @@ def gitea_get_shell_health() -> dict:
     return native_mcp_preference.shell_health_status()
 
 
+@mcp.tool()
+def gitea_diagnose_terminal(
+    cwd: str | None = None,
+    command: list[str] | None = None,
+) -> dict:
+    """Read-only: Run active diagnostics on terminal launcher spawn ability (#556).
+
+    Args:
+        cwd: Active worktree or current directory to test (defaults to GITEA_ACTIVE_WORKTREE).
+        command: Test command to attempt (defaults to ['git', '--version'] or ['echo', 'ok']).
+    """
+    resolved_cwd = cwd or os.environ.get("GITEA_ACTIVE_WORKTREE")
+    if resolved_cwd:
+        resolved_cwd = os.path.realpath(resolved_cwd)
+
+    probe_cmd = command or native_mcp_preference.default_terminal_probe_command()
+
+    diag = native_mcp_preference.diagnose_terminal_failure(resolved_cwd, probe_cmd)
+    probe_res = native_mcp_preference.probe_terminal_spawn(resolved_cwd, probe_cmd)
+
+    return {
+        "healthy": probe_res["healthy"],
+        "error_type": None if probe_res["healthy"] else probe_res["error_type"] or diag["error_type"],
+        "error_msg": "" if probe_res["healthy"] else probe_res["error_msg"] or diag["error_msg"],
+        "cwd": resolved_cwd,
+        "command": probe_cmd,
+        "diagnostics": diag,
+    }
+
+
 @mcp.tool()
 def gitea_validate_review_final_report(
     report_text: str,
@@ -9412,6 +9442,43 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
+    # ── Terminal Launcher Preflight Check (#556) ──
+    if task in native_mcp_preference.GITEA_MUTATION_TASKS:
+        in_test = _preflight_in_test_mode()
+        if not in_test or os.environ.get("GITEA_FORCE_TERMINAL_PROBE") == "1":
+            resolved_cwd = os.environ.get("GITEA_ACTIVE_WORKTREE") or PROJECT_ROOT
+            probe_res = native_mcp_preference.probe_terminal_spawn(resolved_cwd)
+            if not probe_res["healthy"]:
+                profile = get_profile()
+                h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+                username = _authenticated_username(h) if h else None
+                next_safe_action = (
+                    "BLOCKED + DIAGNOSE: terminal-launcher-failure "
+                    f"({probe_res['error_type']}): {probe_res['error_msg']}. "
+                    "Do not attempt git/pytest finalization or unsafe fallbacks. "
+                    "Run gitea_diagnose_terminal, emit blocked-diagnose-report, and stop."
+                )
+                return {
+                    "requested_task": task,
+                    "required_operation_permission": required_permission,
+                    "required_role_kind": required_role,
+                    "active_profile": profile["profile_name"],
+                    "active_identity": username,
+                    "active_profile_allowed_operations": profile.get("allowed_operations") or [],
+                    "allowed_in_current_session": False,
+                    "stop_required": True,
+                    "infra_stop": True,
+                    "blocked": True,
+                    "blocked_reason": "BLOCKED + DIAGNOSE",
+                    "terminal_launcher_unhealthy": True,
+                    "terminal_launcher_diagnostics": probe_res,
+                    "task_role_guidance": [next_safe_action],
+                    "matching_configured_profile": [],
+                    "runtime_switching_supported": gitea_config.is_runtime_switching_enabled(),
+                    "different_mcp_namespace_required": False,
+                    "exact_safe_next_action": next_safe_action,
+                }
+
     record_preflight_check("capability", required_role, resolved_task=task)
 
     # Try automatic dispatch switching

native_mcp_preference.py

@@ -8,6 +8,9 @@ available unless explicit recovery-mode proof is supplied.
 from __future__ import annotations
 
 import re
+import os
+import shutil
+import subprocess
 from typing import Any
 
 from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES
@@ -366,4 +369,154 @@ def assess_gitea_operation_path(
                 else "proceed with native MCP"
             )
         ),
-    }
+    }
+
+
+def default_terminal_probe_command() -> list[str]:
+    return ["git", "--version"] if shutil.which("git") else ["echo", "ok"]
+
+
+def _resolve_executable(cwd: str | None, executable: str | None) -> str | None:
+    if not executable:
+        return None
+    if os.path.isabs(executable):
+        return executable if os.path.exists(executable) and os.access(executable, os.X_OK) else None
+    if "/" in executable or "\\" in executable:
+        target_cwd = cwd or os.getcwd()
+        full_path = os.path.abspath(os.path.join(target_cwd, executable))
+        return full_path if os.path.exists(full_path) and os.access(full_path, os.X_OK) else None
+    return shutil.which(executable)
+
+
+def diagnose_terminal_failure(cwd: str | None, command: list[str]) -> dict[str, Any]:
+    """Inspect and categorize command spawn failures (#556)."""
+    diagnostics = {
+        "cwd": cwd,
+        "command": command,
+        "cwd_exists": True,
+        "cwd_is_dir": True,
+        "shell_exists": True,
+        "executable_exists": True,
+        "resolved_executable": None,
+        "error_type": "session launcher failure",
+        "error_msg": (
+            "Subprocess spawn failed despite valid CWD and executable. This may be due "
+            "to sandboxing, permission restrictions, or system resource limits."
+        ),
+    }
+
+    if cwd:
+        if not os.path.exists(cwd):
+            diagnostics["cwd_exists"] = False
+            diagnostics["error_type"] = "missing cwd"
+            diagnostics["error_msg"] = f"Current working directory does not exist: {cwd}"
+            return diagnostics
+        if not os.path.isdir(cwd):
+            diagnostics["cwd_is_dir"] = False
+            diagnostics["error_type"] = "cwd is not a directory"
+            diagnostics["error_msg"] = f"Current working directory is not a directory: {cwd}"
+            return diagnostics
+
+    exe = command[0] if command else None
+    if exe:
+        resolved_exe = _resolve_executable(cwd, exe)
+        diagnostics["resolved_executable"] = resolved_exe
+        if not resolved_exe:
+            diagnostics["executable_exists"] = False
+            if "venv" in exe or "pytest" in exe:
+                diagnostics["error_type"] = "missing runtime wrapper"
+                diagnostics["error_msg"] = (
+                    "Virtual environment runtime wrapper/executable not found or not "
+                    f"executable: {exe}"
+                )
+            else:
+                diagnostics["error_type"] = "missing executable"
+                diagnostics["error_msg"] = f"Executable not found in PATH or CWD: {exe}"
+            return diagnostics
+
+    # Check common shell availability
+    has_any_shell = False
+    for shell_path in ("/bin/sh", "/bin/zsh", "/bin/bash"):
+        if os.path.exists(shell_path) and os.access(shell_path, os.X_OK):
+            has_any_shell = True
+            break
+    if not has_any_shell:
+        diagnostics["shell_exists"] = False
+        diagnostics["error_type"] = "missing shell"
+        diagnostics["error_msg"] = "No standard system shell (/bin/sh, /bin/zsh, /bin/bash) is available or executable."
+        return diagnostics
+
+    return diagnostics
+
+
+def probe_terminal_spawn(
+    cwd: str | None = None,
+    command: list[str] | None = None,
+) -> dict[str, Any]:
+    """Proactively probe command spawning to detect launcher health (#556)."""
+    probe_cmd = command or default_terminal_probe_command()
+    diag = diagnose_terminal_failure(cwd, probe_cmd)
+    if diag["error_type"] != "session launcher failure":
+        return {
+            "healthy": False,
+            "error_type": diag["error_type"],
+            "error_msg": diag["error_msg"],
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }
+
+    try:
+        proc = subprocess.run(
+            probe_cmd,
+            capture_output=True,
+            text=True,
+            cwd=cwd,
+            timeout=5,
+        )
+        if proc.returncode == 0:
+            return {
+                "healthy": True,
+                "error_type": None,
+                "error_msg": "",
+                "cwd": cwd,
+                "command": probe_cmd,
+                "diagnostics": diag,
+            }
+        # Non-zero status still proves the launcher spawned the process.
+        return {
+            "healthy": True,
+            "error_type": None,
+            "error_msg": "",
+            "cwd": cwd,
+            "command": probe_cmd,
+            "exit_code": proc.returncode,
+            "diagnostics": diag,
+        }
+    except FileNotFoundError:
+        return {
+            "healthy": False,
+            "error_type": diag["error_type"],
+            "error_msg": diag["error_msg"],
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }
+    except subprocess.TimeoutExpired as exc:
+        return {
+            "healthy": False,
+            "error_type": "probe timeout",
+            "error_msg": f"Subprocess probe timed out after {exc.timeout} seconds.",
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }
+    except Exception as exc:
+        return {
+            "healthy": False,
+            "error_type": diag["error_type"],
+            "error_msg": f"Subprocess spawn failed with exception: {exc}",
+            "cwd": cwd,
+            "command": probe_cmd,
+            "diagnostics": diag,
+        }

tests/test_native_mcp_preference.py

@@ -26,6 +26,68 @@ class TestShellHealthCircuitBreaker(unittest.TestCase):
         self.assertEqual(status["consecutive_spawn_failures"], 0)
 
 
+class TestTerminalDiagnostics(unittest.TestCase):
+    def test_diagnose_missing_cwd(self):
+        res = nmp.diagnose_terminal_failure("/nonexistent_directory_for_test", ["git", "status"])
+        self.assertFalse(res["cwd_exists"])
+        self.assertEqual(res["error_type"], "missing cwd")
+        self.assertIn("does not exist", res["error_msg"])
+
+    def test_diagnose_cwd_is_not_dir(self):
+        import tempfile
+        import os
+        with tempfile.NamedTemporaryFile() as tmp:
+            res = nmp.diagnose_terminal_failure(tmp.name, ["git", "status"])
+            self.assertFalse(res["cwd_is_dir"])
+            self.assertEqual(res["error_type"], "cwd is not a directory")
+
+    def test_diagnose_missing_executable(self):
+        res = nmp.diagnose_terminal_failure(None, ["nonexistent_exec_for_test"])
+        self.assertFalse(res["executable_exists"])
+        self.assertEqual(res["error_type"], "missing executable")
+
+    def test_diagnose_missing_runtime_wrapper(self):
+        res = nmp.diagnose_terminal_failure("/tmp", ["venv/bin/pytest"])
+        self.assertFalse(res["executable_exists"])
+        self.assertEqual(res["error_type"], "missing runtime wrapper")
+
+    def test_diagnose_relative_executable_from_cwd(self):
+        import os
+        import tempfile
+
+        with tempfile.TemporaryDirectory() as tmp:
+            script = Path(tmp) / "tool"
+            script.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
+            os.chmod(script, 0o755)
+
+            res = nmp.diagnose_terminal_failure(tmp, ["./tool"])
+            self.assertTrue(res["executable_exists"])
+            self.assertEqual(res["resolved_executable"], str(script))
+
+    def test_probe_terminal_spawn_success(self):
+        res = nmp.probe_terminal_spawn()
+        self.assertTrue(res["healthy"])
+        self.assertIn("command", res)
+        self.assertIn("diagnostics", res)
+
+    def test_probe_terminal_spawn_missing_cwd(self):
+        res = nmp.probe_terminal_spawn("/nonexistent_directory_for_test")
+        self.assertFalse(res["healthy"])
+        self.assertEqual(res["error_type"], "missing cwd")
+
+    def test_probe_terminal_spawn_honors_custom_command(self):
+        res = nmp.probe_terminal_spawn(command=[sys.executable, "-c", "raise SystemExit(3)"])
+        self.assertTrue(res["healthy"])
+        self.assertEqual(res["command"][0], sys.executable)
+        self.assertEqual(res["exit_code"], 3)
+
+    def test_probe_terminal_spawn_blocks_missing_custom_command(self):
+        res = nmp.probe_terminal_spawn(command=["nonexistent_exec_for_test"])
+        self.assertFalse(res["healthy"])
+        self.assertEqual(res["error_type"], "missing executable")
+        self.assertEqual(res["command"], ["nonexistent_exec_for_test"])
+
+
 class TestNativeMcpPreferenceGate(unittest.TestCase):
     def setUp(self):
         nmp.clear_shell_health_for_tests()
@@ -118,4 +180,4 @@ class TestNativeMcpPreferenceGate(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -96,6 +96,14 @@ class TestResolveTaskCapability(unittest.TestCase):
             "GITEA_TOKEN_MERGER": "merger-pass",
         }
 
+    def test_diagnose_terminal_success_has_no_error(self):
+        res = mcp_server.gitea_diagnose_terminal(
+            command=[sys.executable, "-c", "raise SystemExit(0)"]
+        )
+        self.assertTrue(res["healthy"])
+        self.assertIsNone(res["error_type"])
+        self.assertEqual(res["error_msg"], "")
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_author_capable_task(self, _auth, _api):
@@ -110,6 +118,39 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertFalse(res["different_mcp_namespace_required"])
             self.assertIn("ready for operations", res["exact_safe_next_action"])
 
+    @patch("mcp_server.native_mcp_preference.probe_terminal_spawn")
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_mutation_task_blocks_when_terminal_launcher_unhealthy(
+        self,
+        _auth,
+        _api,
+        mock_probe,
+    ):
+        mock_probe.return_value = {
+            "healthy": False,
+            "error_type": "missing cwd",
+            "error_msg": "Current working directory does not exist: /missing",
+            "cwd": "/missing",
+            "command": ["git", "--version"],
+            "diagnostics": {"cwd_exists": False},
+        }
+        env = self._env("author-profile")
+        # Probe is skipped under pytest unless forced (mirrors production path).
+        env["GITEA_FORCE_TERMINAL_PROBE"] = "1"
+        with patch.dict(os.environ, env):
+            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
+
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertTrue(res["stop_required"])
+        self.assertTrue(res["infra_stop"])
+        self.assertTrue(res["terminal_launcher_unhealthy"])
+        self.assertTrue(res["blocked"])
+        self.assertEqual(res["blocked_reason"], "BLOCKED + DIAGNOSE")
+        self.assertEqual(res["terminal_launcher_diagnostics"]["error_type"], "missing cwd")
+        self.assertIn("terminal-launcher-failure", res["exact_safe_next_action"])
+        self.assertIn("BLOCKED + DIAGNOSE", res["exact_safe_next_action"])
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_reviewer_capable_task_blocked(self, _auth, _api):