tools.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/tools.py
 
"""
Purpose
-------
Phase 5A + 5B tool manifest - 20 MCP tools (9 read-only + 11 write)
registered with the :class:`mcp.server.MCPServer` instance. Each tool
function wires the validated input through the six-layer pipeline:
validator → policy → executor → redactor → audit logger.  Write tools
add a HITL safety gate step between policy and execution.
 
Responsibilities
----------------
- Define :func:`register_tools` which attaches all 9 Phase 5A tools.
- Define :func:`register_write_tools` which attaches all 11 Phase 5B
  write tools.
- Each read tool function follows the six-layer pipeline.
- Each write tool function follows the same pipeline but calls
  :func:`_run_tool_write` which inserts the safety gate (Layer 2.5)
  between policy and execution.
- Hold ``_TOOL_CALL_LOCK``, the process-wide choke point that keeps the
  pipeline single-threaded under MCP SDK v2 (plan 244, P3).
 
Write tool pipeline (HITL gate)
--------------------------------
  rate-limiter → policy → validator → **safety gate** → executor →
  redactor → audit
 
Diagnostics
-----------
Domain: MCP
Levels:
    L2 - lifecycle: tool called, tool returned
    L3 - request details
    L4 - deep trace
 
Contracts
---------
- Tool functions never raise for expected error conditions - they return
  a structured error string that the MCP client sees as tool output.
  Nothing here raises an SDK error, so a failure reaches the host as a
  result and never as a JSON-RPC error (plan 244, P4).
- The :func:`register_tools` function is idempotent for a given
  :class:`~mcp.server.MCPServer` instance.
- Tool functions always write one audit record per invocation, even on
  policy denial or validation failure.
- MCP SDK v2 dispatches synchronous tool handlers to worker threads and
  lets requests overlap. Every call therefore passes through
  ``_TOOL_CALL_LOCK`` before entering the pipeline, so the rate limiter,
  executor, audit logger and metrics behind the server-state singleton
  still see one caller at a time (plan 244, P3).
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- oct.mcp.audit (McpAuditRecord and the timestamp and project-root hash helpers for each audit record)
- oct.mcp.safety (SafetyDecision and the confirmation-request text of the write-tool safety gate)
- oct.mcp.server (get_server_state, the shared pipeline instances)
- oct.mcp.validator (validate_tool_args, the Layer 2 argument validation)
- oct.mcp.policy (lazy; TOOL_MANIFEST, the tool spec a write call hands the safety gate)
- mcp (MCPServer, the SDK server the tools register on)
"""
 
from __future__ import annotations
 
import getpass
import hashlib
import threading
import time
import uuid
from pathlib import Path
from typing import Any
 
from mcp.server import MCPServer
 
from oct.core.diagnostics import _dbg
 
from oct.mcp.audit import McpAuditRecord, _hash_project_root, _utc_now_iso
from oct.mcp.safety import SafetyDecision, _format_confirmation_request
from oct.mcp.server import get_server_state
from oct.mcp.validator import validate_tool_args
 
 
# ---------------------------------------------------------------------
# Concurrency choke point (plan 244, P3)
# ---------------------------------------------------------------------
 
#: Process-wide tool-call lock. MCP SDK v2 runs synchronous tool
#: handlers on worker threads via ``anyio.to_thread.run_sync()`` and
#: overlaps concurrent requests, but the pipeline layers reached through
#: :func:`~oct.mcp.server.get_server_state` - rate limiter, executor,
#: audit logger, metrics - were written for a single thread. Every tool
#: call is serialised here so that contract still holds. Serialising is
#: the conservative choice; making those layers individually thread-safe
#: is separate, operator-gated work.
#:
#: Re-entrant so that a nested call on the same thread cannot deadlock.
_TOOL_CALL_LOCK: threading.RLock = threading.RLock()
 
 
# ---------------------------------------------------------------------
# Shared pipeline runner
# ---------------------------------------------------------------------
 
 
def _run_tool(tool_name: str, raw_args: dict) -> str:
    """Execute one tool through the full six-layer pipeline.
 
    This is the single shared entry point called by every read tool
    function. It takes ``_TOOL_CALL_LOCK`` for the whole pipeline so no
    two calls are ever inside it at once, then delegates to
    :func:`_run_tool_pipeline`.
 
    Parameters
    ----------
    tool_name
        MCP tool name, e.g. ``"oct_lint"``.
    raw_args
        Unvalidated dict from the MCP client.
 
    Returns
    -------
    Redacted tool output string (or a structured error message).
    """
    with _TOOL_CALL_LOCK:
        return _run_tool_pipeline(tool_name, raw_args)
 
 
def _run_tool_pipeline(tool_name: str, raw_args: dict) -> str:
    """Run the six-layer read pipeline; caller holds ``_TOOL_CALL_LOCK``.
 
    It handles all error conditions and always writes an audit record.
 
    Parameters
    ----------
    tool_name
        MCP tool name, e.g. ``"oct_lint"``.
    raw_args
        Unvalidated dict from the MCP client.
 
    Returns
    -------
    Redacted tool output string (or a structured error message).
    """
    func = "_run_tool"
    state = get_server_state()
    request_id = str(uuid.uuid4())
    start_time = time.monotonic()
 
    _dbg("MCP", func, f"tool={tool_name} request_id={request_id}", 2)
 
    # Build a skeleton audit record filled in as we proceed.
    record = McpAuditRecord(
        timestamp=_utc_now_iso(),
        session_id=state.session_id,
        request_id=request_id,
        tool=tool_name,
        args={},             # filled after validation
        args_redacted=True,
        user=_safe_getuser(),
        project_root_hash=_hash_project_root(state.project_root),
    )
 
    try:
        # Layer 1: rate limiter
        if not state.rate_limiter.consume():
            record.decision = "denied"
            record.policy_rule = "rate_limit"
            record.duration_ms = _elapsed_ms(start_time)
            state.audit_logger.write(record)
            state.metrics.record_rate_limit_hit(tool_name)
            return _error_response(
                "Rate limit exceeded. Please wait before making more tool calls.",
                tool_name,
            )
 
        # Layer 3: policy check (before validation, so unknown tools are rejected fast)
        decision = state.policy.check(tool_name)
        record.decision = decision.decision
        record.policy_rule = decision.policy_rule
 
        if not decision.allowed:
            record.duration_ms = _elapsed_ms(start_time)
            state.audit_logger.write(record)
            state.metrics.record_call(tool_name, decision.decision,
                                      _elapsed_ms(start_time) / 1000, 0)
            return _error_response(decision.reason, tool_name)
 
        # Layer 2: input validation (Pydantic)
        try:
            validated_model = validate_tool_args(
                tool_name, raw_args, project_root=state.project_root
            )
        except Exception as exc:
            record.decision = "denied"
            record.policy_rule = "validation_failure"
            record.duration_ms = _elapsed_ms(start_time)
            state.audit_logger.write(record)
            state.metrics.record_validation_failure(tool_name)
            return _error_response(f"Input validation failed: {exc}", tool_name)
 
        # Store redacted args in audit record.
        record.args = validated_model.model_dump()
 
        # Layer 4: execution sandbox
        sandbox_result = state.executor.execute(
            tool_name=tool_name,
            validated_args=validated_model.model_dump(),
            project_root=state.project_root,
        )
 
        raw_output = sandbox_result.stdout
        if sandbox_result.stderr:
            if raw_output:
                raw_output += "\n--- stderr ---\n" + sandbox_result.stderr
            else:
                raw_output = sandbox_result.stderr
 
        # Compute output hash before redaction (integrity check).
        output_hash = hashlib.sha256(
            raw_output.encode("utf-8", errors="replace")
        ).hexdigest()
 
        # Layer 5: output redaction
        redactor_result = state.redactor.apply_all(raw_output, state.project_root)
 
        # Layer 6: audit record completion
        record.exit_code = sandbox_result.exit_code
        duration_ms = _elapsed_ms(start_time)
        record.duration_ms = duration_ms
        record.output_size_bytes = len(
            redactor_result.text.encode("utf-8", errors="replace")
        )
        record.output_truncated = (
            sandbox_result.output_truncated or redactor_result.truncated
        )
        record.output_hash_sha256 = output_hash
        record.redactions_applied = redactor_result.redactions_applied
 
        state.audit_logger.write(record)
        state.metrics.record_call(
            tool_name, "allowed", duration_ms / 1000, redactor_result.redactions_applied
        )
 
        _dbg(
            "MCP", func,
            f"tool={tool_name} exit_code={sandbox_result.exit_code} "
            f"redactions={redactor_result.redactions_applied}",
            2,
        )
        return redactor_result.text
 
    except Exception as exc:  # pragma: no cover - catch-all for unexpected errors
        _dbg("MCP", func, f"SYSTEM_ERROR: unexpected error in {tool_name}: {exc}", 1)
        record.decision = "denied"
        record.policy_rule = "internal_error"
        record.duration_ms = _elapsed_ms(start_time)
        try:
            state.audit_logger.write(record)
        except Exception:
            pass
        return _error_response(f"Internal server error: {exc}", tool_name)
 
 
def _run_tool_write(tool_name: str, raw_args: dict) -> str:
    """Execute one write tool through the HITL-extended pipeline.
 
    This is the single shared entry point called by every write tool
    function. It takes the same ``_TOOL_CALL_LOCK`` as :func:`_run_tool`
    - one lock covers both pipelines - then delegates to
    :func:`_run_tool_write_pipeline`.
 
    Parameters
    ----------
    tool_name
        One of the 11 Phase 5B write tool names.
    raw_args
        Unvalidated dict from the MCP client.
 
    Returns
    -------
    Redacted tool output string, a confirmation-required string, or a
    structured error message.
    """
    with _TOOL_CALL_LOCK:
        return _run_tool_write_pipeline(tool_name, raw_args)
 
 
def _run_tool_write_pipeline(tool_name: str, raw_args: dict) -> str:
    """Run the write pipeline; caller holds ``_TOOL_CALL_LOCK``.
 
    Pipeline: rate-limiter → policy → validator → safety gate →
    executor → redactor → audit.
 
    When the safety gate requires confirmation (``confirm=False`` in
    args), returns a structured ``[oct-mcp requires_confirmation] …``
    string without executing anything. The AI host must re-submit with
    ``confirm=True`` after the user approves.
 
    Parameters
    ----------
    tool_name
        One of the 11 Phase 5B write tool names.
    raw_args
        Unvalidated dict from the MCP client.
 
    Returns
    -------
    Redacted tool output string, a confirmation-required string, or a
    structured error message.
    """
    func = "_run_tool_write"
    state = get_server_state()
    request_id = str(uuid.uuid4())
    start_time = time.monotonic()
 
    _dbg("MCP", func, f"tool={tool_name} request_id={request_id}", 2)
 
    from oct.mcp.policy import TOOL_MANIFEST
 
    record = McpAuditRecord(
        timestamp=_utc_now_iso(),
        session_id=state.session_id,
        request_id=request_id,
        tool=tool_name,
        args={},
        args_redacted=True,
        user=_safe_getuser(),
        project_root_hash=_hash_project_root(state.project_root),
    )
 
    try:
        # Layer 1: rate limiter
        if not state.rate_limiter.consume():
            record.decision = "denied"
            record.policy_rule = "rate_limit"
            record.duration_ms = _elapsed_ms(start_time)
            state.audit_logger.write(record)
            state.metrics.record_rate_limit_hit(tool_name)
            return _error_response(
                "Rate limit exceeded. Please wait before making more tool calls.",
                tool_name,
            )
 
        # Layer 3: policy check
        decision = state.policy.check(tool_name)
        record.decision = decision.decision
        record.policy_rule = decision.policy_rule
 
        if not decision.allowed:
            record.duration_ms = _elapsed_ms(start_time)
            state.audit_logger.write(record)
            state.metrics.record_call(tool_name, decision.decision,
                                      _elapsed_ms(start_time) / 1000, 0)
            return _error_response(decision.reason, tool_name)
 
        # Layer 2: input validation (Pydantic)
        try:
            validated_model = validate_tool_args(
                tool_name, raw_args, project_root=state.project_root
            )
        except Exception as exc:
            record.decision = "denied"
            record.policy_rule = "validation_failure"
            record.duration_ms = _elapsed_ms(start_time)
            state.audit_logger.write(record)
            state.metrics.record_validation_failure(tool_name)
            return _error_response(f"Input validation failed: {exc}", tool_name)
 
        record.args = validated_model.model_dump()
 
        # Layer 2.5: HITL safety gate
        spec = TOOL_MANIFEST.get(tool_name)
        if spec is None:  # pragma: no cover - policy already checked this
            record.decision = "denied"
            record.policy_rule = "unknown_tool"
            record.duration_ms = _elapsed_ms(start_time)
            state.audit_logger.write(record)
            return _error_response(f"Unknown tool: {tool_name}", tool_name)
 
        safety_decision: SafetyDecision = state.safety.check(
            tool_name, spec, validated_model.model_dump()
        )
 
        if not safety_decision.proceed:
            record.decision = "requires_confirmation"
            record.policy_rule = "hitl_gate"
            duration_ms = _elapsed_ms(start_time)
            record.duration_ms = duration_ms
            state.audit_logger.write(record)
            state.metrics.record_call(tool_name, "requires_confirmation",
                                      duration_ms / 1000, 0)
            return _format_confirmation_request(
                tool_name, spec, validated_model.model_dump()
            )
 
        # Layer 4: execution sandbox
        sandbox_result = state.executor.execute_write(
            tool_name=tool_name,
            validated_args=validated_model.model_dump(),
            project_root=state.project_root,
            dry_run_override=safety_decision.dry_run,
        )
 
        raw_output = sandbox_result.stdout
        if sandbox_result.stderr:
            if raw_output:
                raw_output += "\n--- stderr ---\n" + sandbox_result.stderr
            else:
                raw_output = sandbox_result.stderr
 
        output_hash = hashlib.sha256(
            raw_output.encode("utf-8", errors="replace")
        ).hexdigest()
 
        # Layer 5: output redaction
        redactor_result = state.redactor.apply_all(raw_output, state.project_root)
 
        # Layer 6: audit record completion
        record.decision = "allowed"
        record.exit_code = sandbox_result.exit_code
        duration_ms = _elapsed_ms(start_time)
        record.duration_ms = duration_ms
        record.output_size_bytes = len(
            redactor_result.text.encode("utf-8", errors="replace")
        )
        record.output_truncated = (
            sandbox_result.output_truncated or redactor_result.truncated
        )
        record.output_hash_sha256 = output_hash
        record.redactions_applied = redactor_result.redactions_applied
 
        state.audit_logger.write(record)
        state.metrics.record_call(
            tool_name, "allowed", duration_ms / 1000, redactor_result.redactions_applied
        )
 
        _dbg(
            "MCP", func,
            f"tool={tool_name} exit_code={sandbox_result.exit_code} "
            f"redactions={redactor_result.redactions_applied}",
            2,
        )
        return redactor_result.text
 
    except Exception as exc:  # pragma: no cover - catch-all for unexpected errors
        _dbg("MCP", func, f"SYSTEM_ERROR: unexpected error in {tool_name}: {exc}", 1)
        record.decision = "denied"
        record.policy_rule = "internal_error"
        record.duration_ms = _elapsed_ms(start_time)
        try:
            state.audit_logger.write(record)
        except Exception:
            pass
        return _error_response(f"Internal server error: {exc}", tool_name)
 
 
def _elapsed_ms(start: float) -> int:
    return int((time.monotonic() - start) * 1000)
 
 
def _safe_getuser() -> str:
    try:
        return getpass.getuser()
    except Exception:
        return ""
 
 
def _error_response(message: str, tool_name: str) -> str:
    """Return a structured error string visible to the MCP client."""
    return f"[oct-mcp error] {tool_name}: {message}"
 
 
# ---------------------------------------------------------------------
# Tool registration
# ---------------------------------------------------------------------
 
 
def register_tools(mcp: MCPServer) -> None:
    """Register all 9 Phase 5A tools with the MCP server instance.
 
    Called once during :func:`~oct.mcp.server.build_mcp_server`.
    """
 
    @mcp.tool(
        name="oct_lint",
        description=(
            "Run the Option C linter on the project. Returns a JSON summary "
            "of lint violations including header checks, secret detection, "
            "and coding style enforcement. Use `changed=true` to lint only "
            "git-modified files."
        ),
    )
    def oct_lint(
        paths: list[str] | None = None,
        json_output: bool = True,
        jobs: int = 1,
        changed: bool = False,
        fix_headers: bool = False,
        profile: str | None = None,
    ) -> str:
        args: dict = {
            "paths": paths or [],
            "json_output": json_output,
            "jobs": jobs,
            "changed": changed,
            "fix_headers": fix_headers,
            "profile": profile,
        }
        return _run_tool("oct_lint", args)
 
    @mcp.tool(
        name="oct_health",
        description=(
            "Run the Option C project health check. Returns a summary of "
            "project structure integrity, configuration validity, and any "
            "detected issues."
        ),
    )
    def oct_health(
        json_output: bool = True,
        verbose: bool = False,
    ) -> str:
        return _run_tool("oct_health", {"json_output": json_output, "verbose": verbose})
 
    @mcp.tool(
        name="oct_skeleton",
        description=(
            "Export the project source skeleton (file tree with content "
            "stubs). Returns a JSON manifest of exported files. Sensitive "
            "files (.env, *.key, etc.) are automatically excluded."
        ),
    )
    def oct_skeleton(
        json_output: bool = True,
        output_dir: str | None = None,
    ) -> str:
        return _run_tool("oct_skeleton", {"json_output": json_output, "output_dir": output_dir})
 
    @mcp.tool(
        name="oct_deps",
        description=(
            "Inspect project dependencies. Returns a JSON summary of "
            "installed packages, missing dependencies, and version mismatches."
        ),
    )
    def oct_deps(
        json_output: bool = True,
        check: bool = False,
    ) -> str:
        return _run_tool("oct_deps", {"json_output": json_output, "check": check})
 
    @mcp.tool(
        name="oct_test",
        description=(
            "Run the project test suite using pytest. Returns a JSON summary "
            "of test results including pass/fail counts and short failure "
            "tracebacks. Use `changed=true` to run only tests related to "
            "git-modified files."
        ),
    )
    def oct_test(
        json_output: bool = True,
        paths: list[str] | None = None,
        changed: bool = False,
        jobs: int = 1,
    ) -> str:
        args: dict = {
            "json_output": json_output,
            "paths": paths or [],
            "changed": changed,
            "jobs": jobs,
        }
        return _run_tool("oct_test", args)
 
    @mcp.tool(
        name="oct_typecheck",
        description=(
            "Run type checking on the project using pyright (falls back to "
            "mypy if pyright is not installed). Returns a JSON object with "
            "the tool used, output, and exit code."
        ),
    )
    def oct_typecheck(
        json_output: bool = True,
        paths: list[str] | None = None,
    ) -> str:
        args: dict = {
            "json_output": json_output,
            "paths": paths or [],
        }
        return _run_tool("oct_typecheck", args)
 
    @mcp.tool(
        name="oct_diag",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Run Option C diagnostics. Available subcommands: "
            "'validate-config' (check .octrc.json), "
            "'show-config' (display resolved config), "
            "'list-checks' (list available lint checks)."
        ),
    )
    def oct_diag(
        json_output: bool = True,
        subcommand: str = "validate-config",
    ) -> str:
        return _run_tool("oct_diag", {"json_output": json_output, "subcommand": subcommand})
 
    @mcp.tool(
        name="oct_git_status",
        description=(
            "Show the git status of the project: staged files, modified "
            "files, branch name, and commit counts. Returns a JSON summary."
        ),
    )
    def oct_git_status(
        json_output: bool = True,
        verbose: bool = False,
    ) -> str:
        return _run_tool("oct_git_status", {"json_output": json_output, "verbose": verbose})
 
    @mcp.tool(
        name="oct_git_check",
        description=(
            "Run the Option C git pre-commit quality gate check. Verifies "
            "that staged changes pass lint, format, and secret-detection "
            "checks. Returns a JSON summary of results."
        ),
    )
    def oct_git_check(
        json_output: bool = True,
        verbose: bool = False,
    ) -> str:
        return _run_tool("oct_git_check", {"json_output": json_output, "verbose": verbose})
 
 
def register_write_tools(mcp: MCPServer) -> None:
    """Register all 11 Phase 5B write tools with the MCP server instance.
 
    All write tools require the user to confirm with ``confirm=True``
    before execution proceeds (HITL gate). Without confirmation the tool
    returns a ``[oct-mcp requires_confirmation]`` string describing what
    would happen.
 
    Called once during :func:`~oct.mcp.server.build_mcp_server`.
    """
 
    @mcp.tool(
        name="oct_format",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Auto-fix Option C compliance violations in Python files. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "Set apply=True AND confirm=True to write changes. "
            "With apply=False (default) runs in --dry-run mode even when confirmed."
        ),
    )
    def oct_format(
        paths: list[str] | None = None,
        apply: bool = False,
        confirm: bool = False,
        changed: bool = False,
        verbose: bool = False,
        jobs: int = 1,
        json_output: bool = True,
    ) -> str:
        return _run_tool_write("oct_format", {
            "paths": paths or [], "apply": apply, "confirm": confirm,
            "changed": changed, "verbose": verbose, "jobs": jobs,
            "json_output": json_output,
        })
 
    @mcp.tool(
        name="oct_docs",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Generate or clean Option C documentation. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "Use sphinx=True, doxygen=True, or all_docs=True to select output. "
            "dry_run=True (default) previews without generating files."
        ),
    )
    def oct_docs(
        sphinx: bool = False,
        doxygen: bool = False,
        all_docs: bool = False,
        clean: bool = False,
        dry_run: bool = True,
        confirm: bool = False,
    ) -> str:
        return _run_tool_write("oct_docs", {
            "sphinx": sphinx, "doxygen": doxygen, "all_docs": all_docs,
            "clean": clean, "dry_run": dry_run, "confirm": confirm,
        })
 
    @mcp.tool(
        name="oct_export_source",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Export consolidated source-code files for inspection. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "Writes _source_code-* export directories to the project root."
        ),
    )
    def oct_export_source(
        confirm: bool = False,
        verbose: bool = False,
        single_dir: bool = False,
        warn_secrets: bool = True,
        block_secrets: bool = False,
    ) -> str:
        return _run_tool_write("oct_export_source", {
            "confirm": confirm, "verbose": verbose, "single_dir": single_dir,
            "warn_secrets": warn_secrets, "block_secrets": block_secrets,
        })
 
    @mcp.tool(
        name="oct_new",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Create a new Option C-compliant Python source file. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "dry_run=True (default) shows what would be created without writing."
        ),
    )
    def oct_new(
        path: str = "",
        confirm: bool = False,
        force: bool = False,
        create_test: bool = False,
        dry_run: bool = True,
        verbose: bool = False,
    ) -> str:
        return _run_tool_write("oct_new", {
            "path": path, "confirm": confirm, "force": force,
            "create_test": create_test, "dry_run": dry_run, "verbose": verbose,
        })
 
    @mcp.tool(
        name="oct_scaffold",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Create a new Option C project directory tree. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "dry_run=True (default) shows what would be created without writing."
        ),
    )
    def oct_scaffold(
        name: str = "",
        confirm: bool = False,
        dry_run: bool = True,
    ) -> str:
        return _run_tool_write("oct_scaffold", {
            "name": name, "confirm": confirm, "dry_run": dry_run,
        })
 
    @mcp.tool(
        name="oct_clean",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Remove _source_code-*, _skeleton_code-*, and __pycache__ "
            "directories from the project. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "dry_run=True (default) shows what would be deleted without deleting."
        ),
    )
    def oct_clean(
        path: str | None = None,
        confirm: bool = False,
        dry_run: bool = True,
    ) -> str:
        return _run_tool_write("oct_clean", {
            "path": path, "confirm": confirm, "dry_run": dry_run,
        })
 
    @mcp.tool(
        name="oct_install_hooks",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Install Option C pre-commit hooks configuration. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "Writes .pre-commit-config.yaml to the project root."
        ),
    )
    def oct_install_hooks(
        confirm: bool = False,
        force: bool = False,
    ) -> str:
        return _run_tool_write("oct_install_hooks", {
            "confirm": confirm, "force": force,
        })
 
    @mcp.tool(
        name="oct_git_commit",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Commit staged changes with Option C quality gate enforcement. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "Validates Conventional Commits format and runs lint/format checks. "
            "Note: --no-verify is not available via MCP."
        ),
    )
    def oct_git_commit(
        message: str = "",
        confirm: bool = False,
        force: bool = False,
        json_output: bool = True,
    ) -> str:
        return _run_tool_write("oct_git_commit", {
            "message": message, "confirm": confirm, "force": force,
            "json_output": json_output,
        })
 
    @mcp.tool(
        name="oct_git_hooks_install",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Install Option C git pre-commit hooks via 'oct git hooks install'. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed."
        ),
    )
    def oct_git_hooks_install(
        confirm: bool = False,
        force: bool = False,
    ) -> str:
        return _run_tool_write("oct_git_hooks_install", {
            "confirm": confirm, "force": force,
        })
 
    @mcp.tool(
        name="oct_git_init",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Initialise or enhance a git repository with Option C conventions "
            "(gitignore, gitattributes, hooks). "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "dry_run=True (default) shows what would change without modifying."
        ),
    )
    def oct_git_init(
        confirm: bool = False,
        dry_run: bool = True,
        github: bool = False,
        no_hooks: bool = False,
    ) -> str:
        return _run_tool_write("oct_git_init", {
            "confirm": confirm, "dry_run": dry_run,
            "github": github, "no_hooks": no_hooks,
        })
 
    @mcp.tool(
        name="oct_git_changelog",
        description=(
            # OCT-LINT: disable=no_hardcoded_secrets reason="MCP tool description prose read by the AI host to choose a tool, not a credential"
            "Generate CHANGELOG.md from Conventional Commits history. "
            "REQUIRES CONFIRMATION: call with confirm=True to proceed. "
            "dry_run=True (default) prints to stdout without writing the file."
        ),
    )
    def oct_git_changelog(
        confirm: bool = False,
        dry_run: bool = True,
        since: str | None = None,
        version: str | None = None,
        json_output: bool = True,
    ) -> str:
        return _run_tool_write("oct_git_changelog", {
            "confirm": confirm, "dry_run": dry_run,
            "since": since, "version": version, "json_output": json_output,
        })