nodes/tool_executor.py

from __future__ import annotations

import subprocess
import sys
from pathlib import Path
from typing import Any

from config import Settings
from graph.state import AgentState, PlanStep
from models.openrouter_model import OpenRouterModels
from tools.audio_transcriber import transcribe_audio
from tools.image_analyzer import analyze_image
from tools.pdf_reader import read_supported_file
from tools.python_tool import calculate_expression
from tools.web_search import search_web
from tools.webpage_reader import read_webpage


def execute_next_tool(
    state: AgentState,
    settings: Settings,
    models: OpenRouterModels,
) -> dict[str, Any]:
    plan = [
        dict(step)
        for step in state.get("plan", [])
    ]

    index = state.get(
        "step_index",
        0,
    )

    if index >= len(plan):
        return {
            "step_index": index,
        }

    step: PlanStep = plan[index]
    action = step.get(
        "action",
        "",
    )

    result: dict[str, Any]

    try:
        if action == "web_search":
            query = (
                step.get("query")
                or step.get("instruction")
                or state["question"]
            )

            result = search_web(
                query,
                settings,
            )

        elif action == "read_webpage":
            url = (
                step.get("query")
                or _best_url(state)
            )

            result = read_webpage(
                url,
                settings,
            )

        elif action == "read_file":
            result = read_supported_file(
                state.get("file_path"),
                state["question"],
            )

        elif action == "analyze_image":
            result = analyze_image(
                state.get("file_path"),
                state["question"],
                models.vision,
            )

        elif action == "transcribe_audio":
            file_path = state.get(
                "file_path"
            )

            if not file_path:
                raise ValueError(
                    "No audio file is available."
                )

            result = transcribe_audio(
                file_path
            )

        elif action == "execute_python_file":
            file_path = state.get(
                "file_path"
            )

            if not file_path:
                raise ValueError(
                    "No Python file is available."
                )

            result = _execute_python_file(
                file_path
            )

        elif action == "calculate":
            expression = (
                step.get("expression")
                or step.get("query")
                or step.get(
                    "instruction",
                    "",
                )
            )

            result = calculate_expression(
                expression
            )

        else:
            result = {
                "ok": False,
                "error": (
                    f"Unsupported action: {action}"
                ),
            }

    except Exception as exc:
        result = {
            "ok": False,
            "error": (
                f"{type(exc).__name__}: {exc}"
            ),
        }

    step["status"] = (
        "done"
        if result.get("ok")
        else "failed"
    )

    step["result"] = str(
        result.get("content")
        or result.get("error")
        or ""
    )[:6000]

    plan[index] = step

    tool_record = {
        "action": action,
        "instruction": step.get(
            "instruction",
            "",
        ),
        **result,
    }

    evidence = list(
        state.get("evidence", [])
    )

    if result.get("ok"):
        evidence.append(
            {
                "source": result.get(
                    "source",
                    action,
                ),
                "content": str(
                    result.get(
                        "content",
                        "",
                    )
                )[:10000],
                "metadata": result.get(
                    "metadata",
                    {},
                ),
            }
        )

    return {
        "plan": plan,
        "step_index": index + 1,
        "iteration": (
            state.get("iteration", 0)
            + 1
        ),
        "tool_results": [
            *state.get(
                "tool_results",
                [],
            ),
            tool_record,
        ],
        "evidence": evidence,
        "error": "",
    }


def _execute_python_file(
    file_path: str,
    timeout: int = 45,
) -> dict[str, Any]:
    path = Path(file_path)

    if not path.exists():
        return {
            "ok": False,
            "error": (
                f"Python file not found: {path}"
            ),
        }

    if path.suffix.lower() != ".py":
        return {
            "ok": False,
            "error": (
                "Attachment is not a Python file."
            ),
        }

    try:
        completed = subprocess.run(
            [
                sys.executable,
                str(path),
            ],
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )

    except subprocess.TimeoutExpired:
        return {
            "ok": False,
            "error": (
                "Python execution timed out."
            ),
        }

    stdout = completed.stdout.strip()
    stderr = completed.stderr.strip()

    output_lines = [
        line.strip()
        for line in stdout.splitlines()
        if line.strip()
    ]

    final_output = (
        output_lines[-1]
        if output_lines
        else ""
    )

    if completed.returncode != 0:
        return {
            "ok": False,
            "error": (
                stderr
                or "Python execution failed."
            ),
            "metadata": {
                "stdout": stdout,
                "stderr": stderr,
                "returncode": (
                    completed.returncode
                ),
            },
        }

    return {
        "ok": bool(final_output),
        "source": str(path),
        "content": final_output,
        "metadata": {
            "stdout": stdout,
            "stderr": stderr,
            "returncode": (
                completed.returncode
            ),
        },
    }


def _best_url(
    state: AgentState,
) -> str:
    for record in reversed(
        state.get("tool_results", [])
    ):
        for item in record.get(
            "results",
            [],
        ):
            if (
                isinstance(item, dict)
                and item.get("url")
            ):
                return str(item["url"])

    raise ValueError(
        "No URL is available. Add a web_search "
        "step before read_webpage."
    )