Class PlanRunner

src/mcp_cli/planning/executor.py:97–813

Orchestrates LLM-driven plan execution with mcp-cli integration. Each step is executed by the LLM: given the step description and tool schemas, the LLM generates the actual tool call (with correct parameter names and values). If a tool fails, the error is fed back to the LLM which c

class PlanRunner:
    """Orchestrates LLM-driven plan execution with mcp-cli integration.

    Each step is executed by the LLM: given the step description and tool
    schemas, the LLM generates the actual tool call (with correct parameter
    names and values). If a tool fails, the error is fed back to the LLM
    which can retry with corrected arguments.

    Features:
    - LLM-driven tool call generation with tool schemas
    - Automatic retry on failure with LLM error correction
    - Parallel batch execution for independent steps (topological batching)
    - Progress callbacks for terminal/dashboard display
    - Dry-run mode (trace without executing)
    - Execution checkpointing and resume
    - DAG visualization
    """

    def __init__(
        self,
        context: PlanningContext,
        *,
        model_manager: ModelManagerProtocol | None = None,
        on_step_start: Callable[[str, str, str], None] | None = None,
        on_step_complete: Callable[[StepResult], None] | None = None,
        on_tool_start: ToolStartCallback | None = None,
        on_tool_complete: ToolCompleteCallback | None = None,
        enable_guards: bool = False,
        max_concurrency: int = DEFAULT_PLAN_MAX_CONCURRENCY,
        max_step_retries: int = DEFAULT_PLAN_MAX_STEP_RETRIES,
    ) -> None:
        """Initialize the plan runner.

        Args:
            context: PlanningContext with tool_manager and graph_store.
            model_manager: ModelManager for LLM-driven step execution.
                When provided, the LLM generates tool calls from step
                descriptions and can retry on failure with error feedback.
            on_step_start: Callback(step_index, step_title, tool_name) before each step.
            on_step_complete: Callback(StepResult) after each step.
            on_tool_start: Async callback(tool_name, arguments) before each tool call
                within a step. Called for each agentic loop tool invocation.
            on_tool_complete: Async callback(tool_name, result_str, success, elapsed)
                after each tool call. Called for each agentic loop tool invocation.
            enable_guards: If True, enforce guard checks during execution.
            max_concurrency: Maximum concurrent steps within a batch.
            max_step_retries: Maximum LLM retry attempts per step on failure.
        """
        self.context = context
        self._model_manager = model_manager
        self._on_step_start = on_step_start
        self._on_step_complete = on_step_complete
        self._on_tool_start = on_tool_start
        self._on_tool_complete = on_tool_complete
        self._max_concurrency = max_concurrency
        self._max_step_retries = max_step_retries

        # Create the MCP tool backend with guard integration