executor.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/executor.py
 
"""
Purpose
-------
Tool Executor (Layer 4, inner logic) for the ``oct-mcp`` six-layer
pipeline. Maps validated tool arguments to ``oct`` CLI argv lists and
delegates execution to :class:`oct.mcp.sandbox.SandboxExecutor`.
 
Responsibilities
----------------
- Provide one ``_build_argv_<tool>()`` function per Phase 5A tool that
  converts a validated Pydantic model to an ``["oct", subcommand, ...]``
  argv list.
- Provide :class:`ToolExecutor` which wires together the Pydantic model
  dict and the sandbox.
- Never import oct internals directly — always call the ``oct`` CLI as
  a subprocess (blueprint §5.5 isolation principle, design decision D3).
 
Diagnostics
-----------
Domain: MCP
Levels:
    L2 — lifecycle: tool execution
    L3 — argv details
    L4 — deep trace
 
Contracts
---------
- This module does not import ``click`` or the MCP SDK.
- ``_build_argv_*()`` functions accept a ``dict`` (from
  ``model.model_dump()``) and return a ``list[str]``. They never raise
  for valid validated input.
- :meth:`ToolExecutor.execute` never raises — it returns
  :class:`~oct.mcp.sandbox.SandboxResult` directly from the sandbox.
"""
 
from __future__ import annotations
 
import sys
from pathlib import Path
from typing import Callable
 
try:
    from oc_diagnostics import _dbg as _real_dbg
 
    def _dbg(*args, **kwargs) -> None:
        _real_dbg(*args, **kwargs)
except ImportError:  # pragma: no cover
    def _dbg(*args, **kwargs) -> None:
        return None
 
from oct.mcp.sandbox import SandboxExecutor, SandboxResult
 
 
# ---------------------------------------------------------------------
# argv builders (one per tool)
# ---------------------------------------------------------------------
 
 
def _build_argv_lint(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct lint``."""
    cmd = [sys.executable, "-m", "oct", "lint"]
    if args.get("json_output", True):
        cmd.append("--json")
    profile = args.get("profile")
    if profile:
        cmd.extend(["--profile", profile])
    if args.get("changed", False):
        cmd.append("--changed")
    if args.get("fix_headers", False):
        cmd.append("--fix-headers")
    jobs = args.get("jobs", 1)
    if jobs > 1:
        cmd.extend(["--jobs", str(jobs)])
    paths = args.get("paths", [])
    cmd.extend(paths)
    return cmd
 
 
def _build_argv_health(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct health``."""
    cmd = [sys.executable, "-m", "oct", "health"]
    if args.get("json_output", True):
        cmd.append("--json")
    if args.get("verbose", False):
        cmd.append("--verbose")
    return cmd
 
 
def _build_argv_skeleton(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct export-skeleton``.
 
    OI-502: no longer emits ``--output-dir``; that flag is not accepted
    by the CLI. The output destination is derived by the CLI from the
    project root.
    """
    cmd = [sys.executable, "-m", "oct", "export-skeleton"]
    if args.get("json_output", True):
        cmd.append("--json")
    return cmd
 
 
def _build_argv_deps(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct deps``.
 
    OI-502: no longer emits ``--check``; that flag is not accepted by
    the CLI. Pin auditing is instead available via ``--audit-pins`` (not
    modelled here yet).
    """
    cmd = [sys.executable, "-m", "oct", "deps"]
    if args.get("json_output", True):
        cmd.append("--json")
    return cmd
 
 
def _build_argv_test(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct test``.
 
    OI-502: no longer emits ``--jobs``; the CLI does not accept it and
    the ``jobs`` field was removed from :class:`OctTestArgs`.
    """
    cmd = [sys.executable, "-m", "oct", "test"]
    if args.get("json_output", True):
        cmd.append("--json")
    if args.get("changed", False):
        cmd.append("--changed")
    paths = args.get("paths", [])
    cmd.extend(paths)
    return cmd
 
 
def _build_argv_typecheck(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct typecheck``."""
    cmd = [sys.executable, "-m", "oct", "typecheck"]
    if args.get("json_output", True):
        cmd.append("--json")
    paths = args.get("paths", [])
    cmd.extend(paths)
    return cmd
 
 
def _build_argv_diag(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct diag <subcommand>``.
 
    OI-502: no longer emits ``--json``; no diag subcommand currently
    accepts that flag (``validate-config`` may grow one via OI-534, at
    which point this builder will conditionally re-emit it).
    """
    subcommand = args.get("subcommand", "validate-config")
    cmd = [sys.executable, "-m", "oct", "diag", subcommand]
    return cmd
 
 
def _build_argv_git_status(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct git status``."""
    cmd = [sys.executable, "-m", "oct", "git", "status"]
    if args.get("json_output", True):
        cmd.append("--json")
    if args.get("verbose", False):
        cmd.append("--verbose")
    return cmd
 
 
def _build_argv_git_check(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct git check``."""
    cmd = [sys.executable, "-m", "oct", "git", "check"]
    if args.get("json_output", True):
        cmd.append("--json")
    if args.get("verbose", False):
        cmd.append("--verbose")
    return cmd
 
 
# ---------------------------------------------------------------------
# Phase 5B — argv builders for write tools (11)
# Note: dry_run is determined by SafetyGate, not just by the args dict.
# Each builder receives a ``_dry_run_override`` key injected by
# ToolExecutor.execute_write() when SafetyDecision.dry_run is True.
# ---------------------------------------------------------------------
 
 
def _build_argv_format(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct format``.
 
    Uses ``--dry-run`` unless ``apply=True`` AND ``confirm=True``.
    The safety gate enforces this rule upstream; the argv builder also
    checks ``_dry_run_override`` (set by execute_write()) as a redundant guard.
    """
    cmd = [sys.executable, "-m", "oct", "format"]
    # SafetyGate passes _dry_run_override=True when apply=False + confirm=True.
    dry_run = args.get("_dry_run_override", not (args.get("apply") and args.get("confirm")))
    if dry_run:
        cmd.append("--dry-run")
    else:
        cmd.append("--fix")
    if args.get("json_output", True):
        cmd.append("--json")
    if args.get("changed", False):
        cmd.append("--changed")
    if args.get("verbose", False):
        cmd.append("--verbose")
    # OI-502: ``jobs`` removed; the formatter argparser does not accept it.
    cmd.extend(args.get("paths", []))
    return cmd
 
 
def _build_argv_docs(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct docs``."""
    cmd = [sys.executable, "-m", "oct", "docs"]
    if args.get("all_docs", False):
        cmd.append("--all")
    elif args.get("sphinx", False):
        cmd.append("--sphinx")
    elif args.get("doxygen", False):
        cmd.append("--doxygen")
    if args.get("clean", False):
        cmd.append("--clean")
    if args.get("dry_run", True):
        cmd.append("--dry-run")
    return cmd
 
 
def _build_argv_export_source(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct export-source``."""
    cmd = [sys.executable, "-m", "oct", "export-source"]
    if args.get("verbose", False):
        cmd.append("--verbose")
    if args.get("single_dir", False):
        cmd.append("--single-dir")
    if args.get("warn_secrets", True):
        cmd.append("--warn-secrets")
    if args.get("block_secrets", False):
        cmd.append("--block-secrets")
    return cmd
 
 
def _build_argv_new(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct new``."""
    path = args.get("path", "")
    cmd = [sys.executable, "-m", "oct", "new", path]
    if args.get("dry_run", True):
        cmd.append("--dry-run")
    if args.get("force", False):
        cmd.append("--force")
    if args.get("create_test", False):
        cmd.append("--test")
    if args.get("verbose", False):
        cmd.append("--verbose")
    return cmd
 
 
def _build_argv_scaffold(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct scaffold``."""
    name = args.get("name", "")
    cmd = [sys.executable, "-m", "oct", "scaffold", name]
    if args.get("dry_run", True):
        cmd.append("--dry-run")
    return cmd
 
 
def _build_argv_clean(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct clean``."""
    cmd = [sys.executable, "-m", "oct", "clean"]
    path = args.get("path")
    if path:
        cmd.append(path)
    if args.get("dry_run", True):
        cmd.append("--dry-run")
    return cmd
 
 
def _build_argv_install_hooks(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct install-hooks`` (deprecated; calls git hooks install)."""
    cmd = [sys.executable, "-m", "oct", "git", "hooks", "install"]
    if args.get("force", False):
        cmd.append("--force")
    return cmd
 
 
def _build_argv_git_commit(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct git commit``.
 
    Note: ``--no-verify`` is intentionally omitted — bypassing hooks
    is never permitted via MCP (design decision D-5B-6).
    """
    message = args.get("message", "")
    cmd = [sys.executable, "-m", "oct", "git", "commit", "-m", message]
    if args.get("json_output", True):
        cmd.append("--json")
    if args.get("force", False):
        cmd.append("--force")
    return cmd
 
 
def _build_argv_git_hooks_install(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct git hooks install``."""
    cmd = [sys.executable, "-m", "oct", "git", "hooks", "install"]
    if args.get("force", False):
        cmd.append("--force")
    return cmd
 
 
def _build_argv_git_init(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct git init``."""
    cmd = [sys.executable, "-m", "oct", "git", "init"]
    if args.get("dry_run", True):
        cmd.append("--dry-run")
    if args.get("github", False):
        cmd.append("--github")
    if args.get("no_hooks", False):
        cmd.append("--no-hooks")
    return cmd
 
 
def _build_argv_git_changelog(args: dict, project_root: Path) -> list[str]:
    """Build argv for ``oct git changelog``."""
    cmd = [sys.executable, "-m", "oct", "git", "changelog"]
    if args.get("json_output", True):
        cmd.append("--json")
    if args.get("dry_run", True):
        cmd.append("--dry-run")
    since = args.get("since")
    if since:
        cmd.extend(["--since", since])
    version = args.get("version")
    if version:
        cmd.extend(["--release", version])
    return cmd
 
 
# ---------------------------------------------------------------------
# Argv builder registry
# ---------------------------------------------------------------------
 
_ARGV_BUILDERS: dict[str, Callable[[dict, Path], list[str]]] = {
    # Phase 5A — read-only
    "oct_lint":             _build_argv_lint,
    "oct_health":           _build_argv_health,
    "oct_skeleton":         _build_argv_skeleton,
    "oct_deps":             _build_argv_deps,
    "oct_test":             _build_argv_test,
    "oct_typecheck":        _build_argv_typecheck,
    "oct_diag":             _build_argv_diag,
    "oct_git_status":       _build_argv_git_status,
    "oct_git_check":        _build_argv_git_check,
    # Phase 5B — write tools
    "oct_format":           _build_argv_format,
    "oct_docs":             _build_argv_docs,
    "oct_export_source":    _build_argv_export_source,
    "oct_new":              _build_argv_new,
    "oct_scaffold":         _build_argv_scaffold,
    "oct_clean":            _build_argv_clean,
    "oct_install_hooks":    _build_argv_install_hooks,
    "oct_git_commit":       _build_argv_git_commit,
    "oct_git_hooks_install": _build_argv_git_hooks_install,
    "oct_git_init":         _build_argv_git_init,
    "oct_git_changelog":    _build_argv_git_changelog,
}
 
 
# ---------------------------------------------------------------------
# Tool executor
# ---------------------------------------------------------------------
 
 
class ToolExecutor:
    """Maps MCP tool names and validated args to sandboxed ``oct`` calls.
 
    Parameters
    ----------
    sandbox
        :class:`~oct.mcp.sandbox.SandboxExecutor` instance (injected
        so tests can substitute a mock).
    timeout_default
        Default subprocess timeout in seconds.
    timeout_test
        Override timeout for ``oct_test`` (pytest may need more time).
    max_output_bytes
        Hard cap on combined stdout+stderr passed to the sandbox.
    """
 
    def __init__(
        self,
        sandbox: SandboxExecutor | None = None,
        timeout_default: int = 60,
        timeout_test: int = 300,
        max_output_bytes: int = 1_048_576,
    ) -> None:
        func = "ToolExecutor.__init__"
        self._sandbox = sandbox or SandboxExecutor()
        self._timeout_default = timeout_default
        self._timeout_test = timeout_test
        self._max_output_bytes = max_output_bytes
        _dbg("MCP", func, "executor initialised", 2)
 
    def execute(
        self,
        tool_name: str,
        validated_args: dict,
        project_root: Path,
    ) -> SandboxResult:
        """Execute *tool_name* with *validated_args* inside the sandbox.
 
        Parameters
        ----------
        tool_name
            One of the 9 Phase 5A tool names.
        validated_args
            A ``model_dump()`` dict from the corresponding Pydantic model.
        project_root
            Absolute project root path (used as subprocess cwd and
            forwarded to the argv builder).
 
        Returns
        -------
        :class:`~oct.mcp.sandbox.SandboxResult` — never raises.
        """
        func = "ToolExecutor.execute"
        builder = _ARGV_BUILDERS.get(tool_name)
        if builder is None:
            msg = f"No argv builder for tool {tool_name!r}"
            _dbg("MCP", func, f"SYSTEM_ERROR: {msg}", 1)
            return SandboxResult(
                exit_code=1,
                stdout="",
                stderr=msg,
                error_message=msg,
            )
 
        cmd = builder(validated_args, project_root)
        _dbg("MCP", func, f"tool={tool_name} cmd={cmd!r}", 3)
 
        # oct test gets a longer timeout (pytest can be slow).
        timeout = (
            self._timeout_test
            if tool_name == "oct_test"
            else self._timeout_default
        )
 
        return self._sandbox.run(
            cmd=cmd,
            cwd=project_root,
            timeout=timeout,
            max_output_bytes=self._max_output_bytes,
        )
 
    def execute_write(
        self,
        tool_name: str,
        validated_args: dict,
        project_root: Path,
        dry_run_override: bool = False,
    ) -> SandboxResult:
        """Execute a write tool, injecting the safety-gate dry-run decision.
 
        Parameters
        ----------
        tool_name
            One of the 11 Phase 5B write tool names.
        validated_args
            A ``model_dump()`` dict from the corresponding Pydantic model.
        project_root
            Absolute project root path.
        dry_run_override
            When ``True`` (as determined by :class:`~oct.mcp.safety.SafetyGate`),
            injects ``_dry_run_override=True`` into the args dict so that
            ``_build_argv_format`` always passes ``--dry-run`` regardless of
            the ``apply`` field value.
        """
        func = "ToolExecutor.execute_write"
        args = dict(validated_args)
        if dry_run_override:
            args["_dry_run_override"] = True
 
        builder = _ARGV_BUILDERS.get(tool_name)
        if builder is None:
            msg = f"No argv builder for write tool {tool_name!r}"
            _dbg("MCP", func, f"SYSTEM_ERROR: {msg}", 1)
            return SandboxResult(
                exit_code=1,
                stdout="",
                stderr=msg,
                error_message=msg,
            )
 
        cmd = builder(args, project_root)
        _dbg("MCP", func, f"tool={tool_name} cmd={cmd!r}", 3)
 
        # Write tools use the default timeout (doc builds can be slow;
        # git commit runs the quality gate; both are bounded by timeout_default).
        return self._sandbox.run(
            cmd=cmd,
            cwd=project_root,
            timeout=self._timeout_default,
            max_output_bytes=self._max_output_bytes,
        )