Function build_execution_tools

src/harness/agentic_loop/tools.py:300–428

Build plan execution tool for reactive/interactive mode. In plan mode (--plan_mode), execution is driven programmatically by _run_plan_mode() and does NOT use this tool. This tool exists for cases where the LLM generates a plan in reactive mode and wants to execute it. The executio

(
    task_board: TaskBoard,
    plan_executor: Any,  # PlanExecutor (avoid circular import)
    summarize_fn: Any,  # Callable[[str, str], str] — summarizes raw result
)

def build_execution_tools(
    task_board: TaskBoard,
    plan_executor: Any,  # PlanExecutor (avoid circular import)
    summarize_fn: Any,  # Callable[[str, str], str] — summarizes raw result
) -> list[OrchestratorTool]:
    """Build plan execution tool for reactive/interactive mode.

    In plan mode (--plan_mode), execution is driven programmatically by
    _run_plan_mode() and does NOT use this tool. This tool exists for cases
    where the LLM generates a plan in reactive mode and wants to execute it.

    The execution result is summarized via summarize_fn before being returned,
    keeping the orchestrator's context compact. The full raw output stays
    inside the PlanExecutor's isolated scope.

    Args:
        task_board: TaskBoard for looking up plan paths and updating results.
        plan_executor: PlanExecutor instance for running YAML plans through
            the existing AgentSPEX Interpreter.
        summarize_fn: Callable(raw_result, task_description) -> compact summary.
            Typically AgenticLoop._summarize_execution_result, which uses a
            cheap LLM call to produce a structured summary.
    """

    def handle_execute_plan(args: dict) -> str:
        task_id = args["task_id"]
        task = task_board.get_task(task_id)
        if not task:
            return f"Error: Task {task_id} not found on the task board."

        # Guard: must have a generated and approved plan
        if not task.plan_file:
            return (
                f"Error: Task {task_id} has no plan file. "
                "Generate a plan with generate_plan first."
            )
        if not task.plan_verified:
            return (
                f"Error: Task {task_id} plan is not verified/approved. "
                "The plan must be generated and approved before execution."
            )

        plan_path = Path(task.plan_file)
        if not plan_path.exists():
            return f"Error: Plan file {task.plan_file} not found on disk."

        # Mark task as in_progress before execution
        task_board.update_task(task_id, status="in_progress")

        try:
            # Execute the plan in isolation (fresh Interpreter, no orchestrator context)
            execution_result = plan_executor.execute(
                plan_path=plan_path,
                model=args.get("model"),
            )

            # Summarize the raw result using a cheap LLM call —
            # only the compact summary enters the orchestrator's context,