/omnigent/tools/base.py

"""Abstract base class for agent tools."""

from __future__ import annotations

import abc
import re
from dataclasses import dataclass
from pathlib import Path  # used by ToolContext.workspace type hint
from typing import Any

# Tool name constraint: alphanumeric plus ``_`` and ``-``, up to
# 256 characters. OpenAI enforces 1–64 but other providers allow
# longer names, and client-side tools come from the user.
TOOL_NAME_RE: re.Pattern[str] = re.compile(r"^[a-zA-Z0-9_-]{1,256}$")


def is_valid_tool_name(name: str) -> bool:
    """
    Check whether a tool name is valid: 1–256 characters,
    alphanumeric plus ``_`` and ``-``.

    :param name: The tool name to validate, e.g. ``"get_weather"``.
    :returns: ``True`` if the name is valid, ``False`` otherwise.
    """
    return TOOL_NAME_RE.match(name) is not None


@dataclass(frozen=True)
class ToolContext:
    """
    Execution context passed to every tool invocation.

    Provides server-side metadata that tools may need but
    which the LLM does not supply (task identity, agent
    identity, workspace path). Individual tools read the
    fields they need and ignore the rest.

    :param task_id: The current task/workflow ID,
        e.g. ``"task_abc123"``.
    :param agent_id: The registered agent ID,
        e.g. ``"ag_xyz789"``.
    :param workspace: Per-conversation persistent working
        directory. ``upload_file`` resolves paths against it;
        ``sys_terminal_launch`` uses it as the cwd default
        (per ``designs/OMNIGENT_TERMINAL_BRIDGE.md`` §4.6 cwd-resolution
        precedence). ``None`` when no workspace is available (e.g.
        tests).
    :param conversation_id: The current conversation's ID,
        e.g. ``"conv_abc123"``. Used by conversation-scoped
        tools (the ``sys_terminal_*`` family looks up its
        :class:`TerminalRegistry` entries by this id; the
        ``sys_session_*`` family uses it for sub-agent
        addressing). ``None`` when not available (older workflow
        paths, unit tests) — tools that require it should fail
        loud.
    """

    task_id: str
    agent_id: str
    workspace: Path | None = None
    conversation_id: str | None = None


class Tool(abc.ABC):
    """
    Abstract base class for all tools available to the agent.

    Each tool has a unique name, an OpenAI-format schema for the
    LLM, and an ``invoke`` method that executes the tool and
    returns a string result.

    Subclasses must implement ``name()`` as a ``@classmethod``
    (for tools with a fixed name, e.g. ``SpawnTool.name()``)
    or as a regular method (for tools whose name depends on
    instance state, e.g. ``McpTool``).
    """

    @classmethod
    @abc.abstractmethod
    def name(cls) -> str:
        """
        Unique tool name used for dispatch and schema registration.

        :returns: The tool name, e.g. ``"load_skill"``.
        """

    @classmethod
    @abc.abstractmethod
    def description(cls) -> str:
        """
        Human-readable description of what the tool does.

        Must be readable without instantiation — used for tool
        discovery (e.g. the onboarding assistant's
        ``list_builtin_tools``) and should match the description
        in :meth:`get_schema`.

        :returns: The tool's description string.
        """

    @abc.abstractmethod
    def get_schema(self) -> dict[str, Any]:
        """
        Return the OpenAI Chat Completions tool schema.

        :returns: A dict with ``"type": "function"`` and a
            ``"function"`` sub-dict describing the tool's name,
            description, and parameters.
        """

    def invoke(self, arguments: str, ctx: ToolContext) -> str:
        """
        Execute the tool with the given arguments.

        Optional override. Tools that ship as schema-only
        (because the runner dispatches them out-of-band — e.g.
        ``sys_cancel_task``, ``sys_session_send``, async-inbox
        family) inherit this default, which fails loud if the
        AP-side path ever reaches them.

        :param arguments: JSON-encoded arguments string from the
            LLM, e.g. ``'{"name": "summarize"}'``.
        :param ctx: Server-side execution context with task and
            agent identity.
        :returns: The tool's string result.
        :raises NotImplementedError: When the subclass is
            runner-dispatched and Omnigent misroutes here.
        """
        del arguments, ctx
        raise NotImplementedError(
            f"{type(self).__name__}.invoke is runner-dispatched; the "
            f"AP-side path should not reach this method. The runner "
            f"handles dispatch via omnigent/runner/tool_dispatch.py."
        )

    def cancel(self) -> None:  # noqa: B027 — optional override hook; default is a no-op
        """
        Cancel an in-progress invocation.

        Called by ``call_tool_with_timeout`` when the deadline
        expires. Subprocess-based tools override this to kill
        the child process. Default is a no-op.
        """

    def shutdown(self) -> None:  # noqa: B027 — optional override hook; default is a no-op
        """
        Release resources held by this tool instance.

        Called by :meth:`ToolManager.shutdown` during teardown.
        Tools that hold subprocesses, file handles, or other
        long-lived resources override this to clean up.
        Default is a no-op.
        """

    def is_async(self, arguments: str | None = None) -> bool:
        """
        Return ``True`` if this invocation runs in a background workflow.

        Tools that NEVER run async (the common case) leave the
        default. ``arguments`` is passed in case a future tool's
        async-ness depends on call-time choice — current tools either
        ignore the parameter (fixed at the class level, e.g.
        ``SysCallAsyncTool``) or never go async at all.

        Async invocations bypass the inline ``invoke()`` path. Instead,
        the runtime calls :meth:`dispatch_async` which starts a child
        workflow and returns a handle immediately; the real result
        arrives later via the async-work drain.

        :param arguments: JSON-encoded argument string from the LLM,
            same shape as what ``invoke`` would receive. ``None``
            means the caller only wants the tool-level default (used
            by tool manifest generation that doesn't have call-time
            arguments). Tools that ignore per-call semantics can
            safely ignore this parameter.
        :returns: ``True`` iff this invocation should dispatch as a
            background workflow.
        """
        return False

    async def dispatch_async(
        self,
        *,
        parent_task_id: str,
        parent_conversation_id: str,
        agent_id: str,
        agent_name: str,
        arguments: str,
        workspace_path: str | None,
    ) -> Any:
        """
        Run an async invocation — either by spawning a child workflow
        or by producing the result inline.

        Called by the runtime (from an async workflow body) when
        :meth:`is_async` returned True. Two return shapes are
        supported, and the runtime's ``_execute_tools`` dispatcher
        accepts both:

        - **Background dispatch (the common case).** The tool creates
          a child ``task_store`` row, starts the runner-side async task pinned to
          that task_id, and returns an ``_AsyncToolHandle`` (defined
          in ``omnigent.runtime.workflow``). The handle is
          serialized via ``to_handle_json`` and surfaced to the LLM
          as the tool-call output; the real result arrives later
          through the async-work drain.
        - **Inline async work.** The tool returns a ``str`` directly.
          The runtime treats it as a normal sync tool result without
          spawning any child. Used by tools whose work needs the
          parent's async loop (e.g., draining a runner inbox via
          ``the async-work drain``) but produces a result immediately —
          ``sys_read_inbox`` is the canonical example. Also used by
          tools like ``sys_call_async`` to surface argument-validation
          errors that don't justify a child workflow.

        The default raises ``NotImplementedError`` — subclasses that
        override :meth:`is_async` to return True MUST override this
        as well. The split lets synchronous tools ignore the async
        path entirely without import-cycle headaches around
        ``_AsyncToolHandle``.

        :param parent_task_id: The currently-executing parent
            workflow's task_id. The new child task points at it via
            ``root_task_id``; the background workflow signals it via
            the ``async_work_complete`` topic. Inline-result tools
            may ignore this.
        :param parent_conversation_id: The owning conversation's id.
            Recorded on the child task row for conversation-scoped
            queries. Inline-result tools may ignore this.
        :param agent_id: The owning agent's id. Recorded on the
            child task row. Inline-result tools may ignore this.
        :param agent_name: The tool's name (same as ``self.name()``
            in most cases) — recorded as ``agent_name`` on the task
            so ``list_tasks`` results show what produced the work.
            Inline-result tools may ignore this.
        :param arguments: JSON-encoded argument string from the LLM.
        :param workspace_path: Per-conversation workspace directory
            (or ``None`` if this tool doesn't need it).
        :returns: Either an ``_AsyncToolHandle`` instance (the
            runtime's shared handle shape, for the background path)
            or a ``str`` (for inline-result tools). Typed as ``Any``
            to avoid importing the runtime module from the tool
            base.
        :raises NotImplementedError: When called on a tool that
            didn't override this method.
        """
        raise NotImplementedError(
            f"{type(self).__name__} declared is_async() true but did not override dispatch_async()"
        )