STT-Assistant/src/stt_daemon/assistant/executor.py

"""Executor seam: turns a command string into an action result.

Phase 3 ships `NullExecutor`; Phase 4 replaces it with the llama.cpp tool-calling
executor. Executors emit their own `tool_call` event (via the supplied `emit`)
before running, so a confirmation flow can exist later.
"""

from dataclasses import dataclass, field
from typing import Callable

from .llm import LlmError

READ_ONLY = "read_only"
MUTATING = "mutating"


@dataclass
class ExecutionResult:
    """Outcome of one command execution."""

    ok: bool
    result: str
    tool: str | None = None
    args: dict = field(default_factory=dict)
    side_effect: str = READ_ONLY
    requires_confirmation: bool = False
    confirmed: bool | None = None


class NullExecutor:
    """Placeholder executor used until the LLM/tool layer is wired."""

    def run(self, command_text: str, emit: Callable[..., None]) -> ExecutionResult:
        """Always declines: no action backend is configured yet."""
        return ExecutionResult(ok=False, result="no executor configured")


class LlmExecutor:
    """Runs a command through the model's tool-calling loop."""

    def __init__(self, cfg, client, registry, slot_id: int | None = None):
        """Create the executor.

        Args:
            cfg: An `LlmConfig`.
            client: An `LlmClient`.
            registry: A `ToolRegistry`.
            slot_id: Optional llama.cpp slot id for command calls; omitted
                from the request when None.
        """
        self.cfg = cfg
        self.client = client
        self.registry = registry
        self.slot_id = slot_id

    def run(self, command_text: str, emit: Callable[..., None]) -> ExecutionResult:
        """Ask the model, validate/execute any tool calls, and return the answer."""
        messages: list[dict] = [
            {"role": "system", "content": self.cfg.system_prompt_command},
            {"role": "user", "content": command_text},
        ]
        tools = self.registry.specs()
        chat_kwargs: dict = {}
        if self.slot_id is not None:
            chat_kwargs["slot_id"] = self.slot_id
        last_tool: str | None = None
        last_args: dict = {}

        for _ in range(max(1, self.cfg.max_tool_iterations)):
            try:
                response = self.client.chat(messages, tools=tools, **chat_kwargs)
            except LlmError as e:
                return ExecutionResult(ok=False, result=f"llm error: {e}")

            if not response.tool_calls:
                if response.finish_reason == "length":
                    emit("error", code="truncated", message="response truncated")
                    return ExecutionResult(
                        ok=False,
                        result=response.content or "response truncated",
                        tool=last_tool,
                        args=last_args,
                    )
                return ExecutionResult(
                    ok=True, result=response.content, tool=last_tool, args=last_args
                )

            messages.append(
                {
                    "role": "assistant",
                    "content": response.content or "",
                    "tool_calls": [call.message() for call in response.tool_calls],
                }
            )

            for call in response.tool_calls:
                args, error = self.registry.validate(call.name, call.arguments)
                tool = self.registry.get(call.name)
                if error is not None or tool is None:
                    message = error or f"unknown tool: {call.name}"
                    emit("error", code="invalid_tool_call", message=message)
                    return ExecutionResult(ok=False, result=message, tool=call.name)
                emit(
                    "tool_call",
                    tool=call.name,
                    args=args or {},
                    requires_confirmation=tool.requires_confirmation,
                )
                if tool.requires_confirmation:
                    message = f"{call.name} requires confirmation"
                    emit("error", code="confirmation_required", message=message)
                    return ExecutionResult(
                        ok=False,
                        result="confirmation required",
                        tool=call.name,
                        args=args,
                        side_effect=tool.side_effect,
                        requires_confirmation=True,
                    )
                try:
                    output = tool.run(args)
                except Exception as e:  # noqa: BLE001 - a tool crash must not kill the loop
                    emit("error", code="tool_error", message=str(e))
                    return ExecutionResult(
                        ok=False,
                        result=f"tool error: {e}",
                        tool=call.name,
                        args=args,
                        side_effect=tool.side_effect,
                    )
                messages.append(
                    {"role": "tool", "tool_call_id": call.id, "content": output}
                )
                last_tool, last_args = call.name, args

        emit("error", code="tool_loop", message="tool loop limit exceeded")
        return ExecutionResult(
            ok=False, result="tool loop limit exceeded", tool=last_tool, args=last_args
        )