/core/execution.py

"""Tool execution helpers for the shared LLM tool-calling runtime."""

from __future__ import annotations

import json
import logging
from collections.abc import Callable, Sequence
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field, replace
from typing import Any

from core.llm.types import ToolCall
from core.tool_framework.utils.integration_sources import availability_view
from core.types import AgentTool, AgentToolContext, RuntimeTool
from platform.observability.trace.redaction import redact_sensitive
from platform.observability.trace.spans import mark_span_outcome, tool_span

logger = logging.getLogger(__name__)

_TOOL_EXECUTOR_WORKERS = 10
_UNSET: object = object()
_INJECTED_CREDENTIAL_KEYS = frozenset(
    {
        "github_url",
        "github_mode",
        "github_token",
        "github_command",
        "github_args",
    }
)


@dataclass(frozen=True)
class ToolExecutionResult:
    """Structured result from one tool call."""

    content: str | list[dict[str, Any]]
    details: Any = None
    is_error: bool = False
    terminate: bool = False
    metadata: dict[str, Any] = field(default_factory=dict)

    def provider_content(self) -> str | list[dict[str, Any]]:
        """Return the content that should be sent back to the LLM provider."""
        return self.content

    def compat_payload(self) -> Any:
        """Return the historical raw payload shape used by old call sites."""
        if self.is_error:
            return {"error": self.content}
        return self.details if self.details is not None else self.content


@dataclass(frozen=True)
class ToolExecutionRequest:
    """Validated tool-call data passed to execution hooks."""

    tool_call: ToolCall
    tool: RuntimeTool
    arguments: dict[str, Any]
    source: str
    resolved_integrations: dict[str, Any]


@dataclass(frozen=True)
class ToolExecutionPatch:
    """Patch object returned by ``after_tool_call`` hooks."""

    content: str | list[dict[str, Any]] | None = None
    details: Any = _UNSET
    is_error: bool | None = None
    terminate: bool | None = None
    metadata: dict[str, Any] | None = None


@dataclass(frozen=True)
class BeforeToolCallResult:
    """Decision object returned by ``before_tool_call`` hooks."""

    approved: bool = False
    blocked: bool = False
    reason: str = ""
    details: Any = None
    terminate: bool = False
    metadata: dict[str, Any] = field(default_factory=dict)


BeforeToolCallHook = Callable[[ToolExecutionRequest], BeforeToolCallResult | None]
AfterToolCallHook = Callable[[ToolExecutionRequest, ToolExecutionResult], ToolExecutionPatch | None]
ToolUpdateHook = Callable[[ToolExecutionRequest, Any], None]


@dataclass(frozen=True)
class ToolExecutionHooks:
    """Lifecycle hooks around validated runtime tool execution."""

    before_tool_call: BeforeToolCallHook | None = None
    after_tool_call: AfterToolCallHook | None = None
    on_tool_update: ToolUpdateHook | None = None


def execute_tool_calls(
    tool_calls: list[ToolCall],
    tools: Sequence[RuntimeTool],
    resolved_integrations: dict[str, Any],
    *,
    hooks: ToolExecutionHooks | None = None,
    tool_resources: dict[str, Any] | None = None,
) -> list[ToolExecutionResult]:
    """Execute provider-requested tools and return structured results.

    Arguments are validated before execution. A single sequential tool in the
    batch forces the whole batch to run sequentially; otherwise calls run in
    parallel while preserving provider order in the returned list.
    """

    hooks = hooks or ToolExecutionHooks()
    tool_sources = availability_view(resolved_integrations)
    tool_map = {t.name: t for t in tools}
    runtime_resources = dict(tool_resources or {})

    def _call(tc: ToolCall) -> ToolExecutionResult:
        with tool_span(tc.name, tool_call_id=tc.id) as span_attrs:
            return _execute_one_tool_call(
                tc,
                tool_map=tool_map,
                tools=tools,
                tool_sources=tool_sources,
                resolved_integrations=resolved_integrations,
                runtime_resources=runtime_resources,
                hooks=hooks,
                span_attrs=span_attrs,
            )

    if len(tool_calls) == 1 or _requires_sequential_execution(tool_calls, tool_map):
        return [_call(tc) for tc in tool_calls]

    results: list[ToolExecutionResult | object] = [_UNSET] * len(tool_calls)
    submitted: dict[Future[ToolExecutionResult], int] = {}
    try:
        with ThreadPoolExecutor(max_workers=min(_TOOL_EXECUTOR_WORKERS, len(tool_calls))) as pool:
            for i, tc in enumerate(tool_calls):
                submitted[pool.submit(_call, tc)] = i
            for fut in as_completed(submitted):
                try:
                    results[submitted[fut]] = fut.result()
                except Exception as fut_exc:  # noqa: BLE001  # lgtm[py/catch-base-exception]
                    results[submitted[fut]] = _error_result(str(fut_exc))
    except RuntimeError as exc:
        logger.warning("[execute_tools] RuntimeError – falling back to sequential: %s", exc)
        for fut, i in submitted.items():
            if results[i] is _UNSET and fut.done():
                try:
                    results[i] = fut.result()
                except Exception as fut_exc:  # noqa: BLE001  # lgtm[py/catch-base-exception]
                    results[i] = _error_result(str(fut_exc))
        for i, tc in enumerate(tool_calls):
            if results[i] is _UNSET:
                results[i] = _call(tc)
    return [
        r if isinstance(r, ToolExecutionResult) else _error_result("tool did not run")
        for r in results
    ]


def execute_tools(
    tool_calls: list[ToolCall],
    tools: Sequence[RuntimeTool],
    resolved_integrations: dict[str, Any],
    *,
    on_tool_update: Callable[[ToolCall, Any], None] | None = None,
) -> list[Any]:
    """Compatibility wrapper returning historical raw payloads."""

    hooks: ToolExecutionHooks | None = None
    if on_tool_update is not None:

        def _on_update(request: ToolExecutionRequest, update: Any) -> None:
            on_tool_update(request.tool_call, update)

        hooks = ToolExecutionHooks(on_tool_update=_on_update)
    return [
        result.compat_payload()
        for result in execute_tool_calls(
            tool_calls,
            tools,
            resolved_integrations,
            hooks=hooks,
            tool_resources=None,
        )
    ]


def _execute_one_tool_call(
    tc: ToolCall,
    *,
    tool_map: dict[str, RuntimeTool],
    tools: Sequence[RuntimeTool],
    tool_sources: dict[str, Any],
    resolved_integrations: dict[str, Any],
    runtime_resources: dict[str, Any],
    hooks: ToolExecutionHooks,
    span_attrs: dict[str, Any],
) -> ToolExecutionResult:
    """Run one validated tool call; record outcome on ``span_attrs``."""
    tool = tool_map.get(tc.name)
    if tool is None:
        mark_span_outcome(span_attrs, "unknown_tool", error=True)
        logger.debug("tool_call unknown name=%s id=%s", tc.name, tc.id)
        return _error_result(f"unknown tool: {tc.name}", metadata={"tool_name": tc.name})

    try:
        validation_error = tool.validate_public_input(tc.input)
        if validation_error:
            mark_span_outcome(span_attrs, "validation_error", error=True)
            logger.debug("tool_call validation_error name=%s id=%s", tc.name, tc.id)
            return _error_result(validation_error, metadata={"tool_name": tc.name})

        source = tool_source(tools, tc.name)
        span_attrs["source"] = source
        request = ToolExecutionRequest(
            tool_call=tc,
            tool=tool,
            arguments=dict(tc.input),
            source=source,
            resolved_integrations=resolved_integrations,
        )
        before = _run_before_hook(hooks, request)
        if before is not None and before.blocked:
            mark_span_outcome(span_attrs, "blocked", error=True)
            logger.debug("tool_call blocked name=%s id=%s", tc.name, tc.id)
            return ToolExecutionResult(
                content=before.reason or f"{tc.name} blocked by before_tool_call hook.",
                details=before.details,
                is_error=True,
                terminate=before.terminate,
                metadata={"tool_name": tc.name, **before.metadata},
            )

        logger.debug("tool_call start name=%s id=%s source=%s", tc.name, tc.id, source)
        raw = _invoke_runtime_tool(
            tool,
            tc,
            request=request,
            tool_sources=tool_sources,
            resolved_integrations=resolved_integrations,
            runtime_resources=runtime_resources,
            hooks=hooks,
        )
        result = _normalize_result(raw, tool_name=tc.name)
        patch = _run_after_hook(hooks, request, result)
        if patch is not None:
            result = _apply_patch(result, patch)
        mark_span_outcome(
            span_attrs,
            "tool_error" if result.is_error else "ok",
            error=result.is_error,
            is_error=result.is_error,
            terminate=result.terminate,
        )
        logger.debug(
            "tool_call done name=%s id=%s outcome=%s",
            tc.name,
            tc.id,
            span_attrs["outcome"],
        )
        return result
    except Exception as exc:
        mark_span_outcome(span_attrs, "exception", error=True)
        logger.warning("[tool:%s] failed: %s", tc.name, exc)
        return _error_result(str(exc), metadata={"tool_name": tc.name})


def _invoke_runtime_tool(
    tool: RuntimeTool,
    tc: ToolCall,
    *,
    request: ToolExecutionRequest,
    tool_sources: dict[str, Any],
    resolved_integrations: dict[str, Any],
    runtime_resources: dict[str, Any],
    hooks: ToolExecutionHooks,
) -> Any:
    """Dispatch to AgentTool.execute or RegisteredTool.run."""
    if isinstance(tool, AgentTool):
        context = AgentToolContext(
            resolved_integrations=resolved_integrations,
            resources=runtime_resources,
            _emit_update=lambda update: _run_update_hook(hooks, request, update),
        )
        return tool.execute(tc.input, context)

    injected = tool.extract_params(tool_sources)
    kwargs = {**injected, **tc.input}
    for key, value in injected.items():
        if key in _INJECTED_CREDENTIAL_KEYS and value not in (None, "", []):
            kwargs[key] = value
    if getattr(tool, "accepts_runtime_context", False):
        context = AgentToolContext(
            resolved_integrations=resolved_integrations,
            resources=runtime_resources,
            _emit_update=lambda update: _run_update_hook(hooks, request, update),
        )
        return tool.run(**kwargs, context=context)
    return tool.run(**kwargs)


def _requires_sequential_execution(
    tool_calls: list[ToolCall],
    tool_map: dict[str, RuntimeTool],
) -> bool:
    for tc in tool_calls:
        tool = tool_map.get(tc.name)
        if isinstance(tool, AgentTool) and tool.effective_execution_mode == "sequential":
            return True
        if (
            not isinstance(tool, AgentTool)
            and tool is not None
            and not getattr(tool, "parallel_safe", True)
        ):
            return True
    return False


def _normalize_result(raw: Any, *, tool_name: str) -> ToolExecutionResult:
    if isinstance(raw, ToolExecutionResult):
        return raw
    # Flag failure on a truthy "error", not the mere presence of the key: a
    # success payload carrying "error": None must reach the agent, not be
    # replaced with {"error": "None"}. Matches bedrock_converse's convention.
    is_error = isinstance(raw, dict) and bool(raw.get("error"))
    content = _content_from_payload(raw)
    if is_error:
        content = str(raw.get("error", content))
    return ToolExecutionResult(
        content=content,
        details=raw,
        is_error=is_error,
        metadata={"tool_name": tool_name},
    )


def _content_from_payload(raw: Any) -> str | list[dict[str, Any]]:
    if isinstance(raw, str):
        return raw
    if isinstance(raw, list) and all(isinstance(item, dict) for item in raw):
        return raw
    return json.dumps(raw, default=str)


def _error_result(message: str, *, metadata: dict[str, Any] | None = None) -> ToolExecutionResult:
    return ToolExecutionResult(
        content=message,
        details={"error": message},
        is_error=True,
        metadata=dict(metadata or {}),
    )


def _run_before_hook(
    hooks: ToolExecutionHooks,
    request: ToolExecutionRequest,
) -> BeforeToolCallResult | None:
    if hooks.before_tool_call is None:
        return None
    try:
        return hooks.before_tool_call(request)
    except Exception as exc:  # noqa: BLE001 - lifecycle hooks should fail closed for the call
        logger.warning("[tool:%s] before_tool_call failed: %s", request.tool_call.name, exc)
        return BeforeToolCallResult(blocked=True, reason=str(exc))


def _run_after_hook(
    hooks: ToolExecutionHooks,
    request: ToolExecutionRequest,
    result: ToolExecutionResult,
) -> ToolExecutionPatch | None:
    if hooks.after_tool_call is None:
        return None
    try:
        return hooks.after_tool_call(request, result)
    except Exception:  # noqa: BLE001 - observer failures must not corrupt the transcript
        logger.debug(
            "[tool:%s] after_tool_call raised; ignoring",
            request.tool_call.name,
            exc_info=True,
        )
        return None


def _run_update_hook(
    hooks: ToolExecutionHooks,
    request: ToolExecutionRequest,
    update: Any,
) -> None:
    if hooks.on_tool_update is None:
        return
    try:
        hooks.on_tool_update(request, update)
    except Exception:  # noqa: BLE001 - partial rendering must not break tool execution
        logger.debug(
            "[tool:%s] on_tool_update raised; ignoring",
            request.tool_call.name,
            exc_info=True,
        )


def _apply_patch(result: ToolExecutionResult, patch: ToolExecutionPatch) -> ToolExecutionResult:
    metadata = dict(result.metadata)
    if patch.metadata:
        metadata.update(patch.metadata)
    kwargs: dict[str, Any] = {"metadata": metadata}
    if patch.content is not None:
        kwargs["content"] = patch.content
    if patch.details is not _UNSET:
        kwargs["details"] = patch.details
    if patch.is_error is not None:
        kwargs["is_error"] = patch.is_error
    if patch.terminate is not None:
        kwargs["terminate"] = patch.terminate
    return replace(result, **kwargs)


def public_tool_input(value: dict[str, Any]) -> dict[str, Any]:
    redacted = redact_sensitive(value)
    return {
        key: item
        for key, item in redacted.items()
        if item != "[runtime object]" and item != "[redacted]"
    }


def tool_source(tools: Sequence[RuntimeTool], tool_name: str) -> str:
    for tool in tools:
        if tool.name == tool_name:
            return str(getattr(tool, "source", "unknown"))
    return "unknown"


def summarise(output: Any) -> str:
    if isinstance(output, ToolExecutionResult):
        output = output.compat_payload()
    if isinstance(output, dict) and "error" in output:
        return f"error: {output['error']}"
    text = json.dumps(output, default=str)
    return text[:120] + "..." if len(text) > 120 else text