tool_registry.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/tool_registry.py
 
"""
Build the tool registry from an explicit module allowlist + ``oc_chat.tools`` entry points.
 
Purpose
-------
Bootstrap helper that builds the ``(registry, schemas)`` pair used by
the ReAct inner loop. Two discovery passes feed the registry:
 
1. **First-party pass** — every module in the explicit
   :data:`_TOOL_MODULE_ALLOWLIST` that exports ``TOOLS``
   (``dict[str, Callable[..., dict]]``) and ``SCHEMAS`` (``list[dict]``).
   This is a deterministic allowlist, **not** a ``tools/*.py`` glob
   (M8.0 WU-6 / G-2): an agent-written ``oc_chat/tools/*.py`` is never
   auto-imported, closing the primary F57 in-process-ACE vector. Adding a
   new first-party tool module requires adding its stem to the allowlist.
2. **Entry-point pass** (Phase K.1b) — every entry point in the
   ``oc_chat.tools`` group, loaded **only under the permissive profile**
   (M8.0 WU-6 / G-3: enforcing profiles deny untrusted plugin ``ep.load()``
   at startup). Third-party packages declare:
   ``[project.entry-points."oc_chat.tools"] my = "my_pkg.tools"`` and
   the named module is loaded and merged the same way.
 
This is logically distinct from tool *dispatch* (which executes a
single tool call inside the loop, lives in
:mod:`oc_chat.core.react_loop`) so a sub-agent can construct a
*subset* registry without touching dispatch code. That separation is
the reason this lives in its own module rather than next to
``run_turn``.
 
Responsibilities
----------------
- :func:`build_tool_registry` — orchestrate both discovery passes,
  detect name collisions, surface failed plugins via L1 ``_dbg``.
- :func:`failed_plugins` — return ``[(entry_point_name, error_msg), ...]``
  for the last build, so the startup REPL can surface a notice.
 
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``ImportError`` on a filesystem tool module is swallowed silently
  (kept for backwards-compat: a missing optional dep should not break
  the registry).
- ``ImportError`` on an entry-point module is logged at L1 with
  ``override=True``, the plugin is skipped, and the failure is recorded
  in :func:`failed_plugins` so a presenter can surface a notice.
- A name collision between any two sources (filesystem ↔ filesystem,
  filesystem ↔ entry-point, entry-point ↔ entry-point) raises
  :class:`ValueError` with both source identifiers in the message.
- Return shape ``(dict, list)`` is the public contract used by
  :func:`oc_chat.main.main` and any future :class:`Agent` builder.
- Filesystem-pass order follows ``sorted(...)`` and runs first, so
  entry-point tools always lose deterministically if a collision is
  somehow legal in future (currently they raise instead).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- importlib.metadata (stdlib, entry-point discovery)
- oc_chat.core.risk_profile (lazy; resolve_risk_profile gates the entry-point pass to the permissive profile)
- oc_chat.tools.ask_user (not imported; allowlisted tool module loaded by name for its TOOLS and SCHEMAS)
- oc_chat.tools.filesystem (not imported; allowlisted tool module loaded by name for its TOOLS and SCHEMAS)
- oc_chat.tools.memory (not imported; allowlisted tool module loaded by name for its TOOLS and SCHEMAS)
- oc_chat.tools.oct_tools (not imported; allowlisted tool module loaded by name for its TOOLS and SCHEMAS)
- oc_chat.tools.planner (not imported; allowlisted tool module loaded by name for its TOOLS and SCHEMAS)
- oc_chat.tools.shell (not imported; allowlisted tool module loaded by name for its TOOLS and SCHEMAS)
- oc_chat.tools.web (not imported; allowlisted tool module loaded by name for its TOOLS and SCHEMAS)
 
See Also
--------
- oc_chat/docs/conventions/TOOL_CONTRACT.md — full authoring contract.
- oc_chat/docs/conventions/CONTRACTS.md §3 — the ``TOOLS`` / ``SCHEMAS`` shape.
"""
from __future__ import annotations
 
import importlib
import importlib.metadata
from pathlib import Path
from types import ModuleType
from typing import Any
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.TOOLS", "module", "loaded", 4)
 
ENTRY_POINT_GROUP = "oc_chat.tools"
 
_failed_plugins: list[tuple[str, str]] = []
"""Entry-point name + error message for each failed plugin import in the last build."""
 
 
def failed_plugins() -> list[tuple[str, str]]:
    """Return ``[(entry_point_name, error_message), ...]`` from the last build.
 
    Empty when every plugin loaded cleanly. Callers (typically the startup
    flow) surface a notice via ``presenter.show_info`` when this is non-empty.
    """
    func = "failed_plugins"
    _dbg("OC_CHAT.TOOLS", func, f"count={len(_failed_plugins)}", 4)
    return list(_failed_plugins)
 
 
def _merge_module_tools(
    mod: ModuleType,
    source: str,
    registry: dict[str, Any],
    schemas: list[dict],
    tool_sources: dict[str, str],
) -> None:
    """Merge ``mod.TOOLS`` + ``mod.SCHEMAS`` into the accumulators.
 
    Raises ``ValueError`` on any name collision, with both source
    identifiers in the message so the offending plugin is unambiguous.
    """
    func = "_merge_module_tools"
    mod_tools = getattr(mod, "TOOLS", None)
    mod_schemas = getattr(mod, "SCHEMAS", None)
    if mod_tools:
        for name in mod_tools:
            if name in registry:
                raise ValueError(
                    f"Tool name collision: '{name}' is registered by "
                    f"{tool_sources[name]} and also by {source}. "
                    f"Tool names must be globally unique across the "
                    f"filesystem scan and every entry point."
                )
            registry[name] = mod_tools[name]
            tool_sources[name] = source
    if mod_schemas:
        schemas.extend(mod_schemas)
    _dbg(
        "OC_CHAT.TOOLS",
        func,
        f"merged {source}: tools={len(mod_tools or {})} schemas={len(mod_schemas or [])}",
        3,
    )
 
 
#: M8.0 WU-6 / G-2 — the explicit first-party tool-module allowlist (replaces the
#: legacy ``tools/*.py`` glob). Every module here declares ``TOOLS``/``SCHEMAS``;
#: an agent-written ``oc_chat/tools/*.py`` is NOT on the list, so it is never
#: auto-imported. Adding a new first-party tool module requires adding its stem here.
_TOOL_MODULE_ALLOWLIST: tuple[str, ...] = (
    "ask_user", "filesystem", "memory", "oct_tools", "planner", "shell", "web",
)
 
 
def build_tool_registry() -> tuple[dict, list]:
    """Build the tool registry from the explicit module allowlist and entry points.
 
    First-party pass imports the explicit :data:`_TOOL_MODULE_ALLOWLIST` (M8.0
    WU-6 / G-2 — no ``tools/*.py`` glob, so an agent-written tool module is not
    auto-imported); entry-point pass scans ``[project.entry-points."oc_chat.tools"]``
    (Phase K.1b) **only under the permissive profile** (G-3: enforcing profiles
    deny untrusted plugin code at startup). Both must declare ``TOOLS`` (dict) and
    ``SCHEMAS`` (list) at module level.
 
    Returns ``(registry, schemas)``. Raises :class:`ValueError` on any
    tool-name collision. Entry-point import failures are logged at L1
    and exposed via :func:`failed_plugins`.
    """
    func = "build_tool_registry"
    registry: dict = {}
    schemas: list = []
    tool_sources: dict[str, str] = {}
 
    _failed_plugins.clear()
 
    # ── First-party pass: an EXPLICIT module allowlist (M8.0 WU-6 / G-2) ──────
    # Deterministic registry: NOT a ``tools/*.py`` glob. An agent that writes a new
    # ``oc_chat/tools/<x>.py`` does not get it imported or registered — that auto-glob
    # was the primary F57 in-process-ACE vector, live even inside the sandbox / under
    # permissive. A NEW legitimate tool module must be ADDED to
    # :data:`_TOOL_MODULE_ALLOWLIST` (the deliberate deterministic-registry discipline).
    for stem in _TOOL_MODULE_ALLOWLIST:
        mod_name = f"oc_chat.tools.{stem}"
        try:
            mod = importlib.import_module(mod_name)
        except ImportError as e:
            # VALIDATION_ERROR — tool's optional dep missing; skip and continue.
            _dbg("OC_CHAT.TOOLS", func, f"VALIDATION_ERROR: {mod_name}: {e}", 3)
            continue
        _merge_module_tools(
            mod,
            source=f"allowlist:{mod_name}",
            registry=registry,
            schemas=schemas,
            tool_sources=tool_sources,
        )
 
    # ── Entry-point pass (Phase K.1b) — profile-gated (M8.0 WU-6 / G-3) ──────
    # Under guarded/isolated, do NOT ``ep.load()`` untrusted third-party plugin code
    # at startup (deny-by-default; an operator-trust custody record is M8.3 / G-6).
    # permissive loads plugins as before (dogfooding accepts the in-process residual).
    from oc_chat.core.risk_profile import resolve_risk_profile
    if resolve_risk_profile() != "permissive":
        _dbg("OC_CHAT.TOOLS", func,
             "entry-point plugin pass skipped under an enforcing profile (G-3)", 2)
    else:
        try:
            eps = importlib.metadata.entry_points(group=ENTRY_POINT_GROUP)
        except TypeError:
            # Python < 3.10 returned dict-of-list; we require 3.10+ but be defensive
            eps = importlib.metadata.entry_points().get(ENTRY_POINT_GROUP, [])
 
        for ep in eps:
            try:
                mod = ep.load()
            except Exception as e:
                # Plugin failed to import — log L1 (visible at default verbosity),
                # record for startup surfacing, and skip. Never crash OC Chat
                # because a third-party plugin is broken.
                err = f"{type(e).__name__}: {e}"
                _failed_plugins.append((ep.name, err))
                _dbg(
                    "OC_CHAT.TOOLS",
                    func,
                    f"VALIDATION_ERROR: entry-point '{ep.name}' failed to load: {err}",
                    1,
                    override=True,
                )
                continue
            _merge_module_tools(
                mod,
                source=f"entry-point:{ep.name}",
                registry=registry,
                schemas=schemas,
                tool_sources=tool_sources,
            )
            _dbg("OC_CHAT.TOOLS", func, f"loaded entry-point {ep.name}", 2)
 
    _dbg(
        "OC_CHAT.TOOLS",
        func,
        f"discovered tools={len(registry)} schemas={len(schemas)} failed_plugins={len(_failed_plugins)}",
        2,
    )
    return registry, schemas