orchestrata_ultimate.agents.tool_registry

Reference › orchestrata_ultimate › agents › tool_registry

tool_registry defines the function-calling tools that volks hand to their language model and the sandboxed executor that runs them. It is used by base, whose tool-calling loop in BaseAgent takes the schemas from here and dispatches each call to the executor.

The schemas are OpenAI-style dictionaries: file tools (read_file, write_file, list_dir, search_code), run_shell, ast_query, check_coverage, validate_mermaid, check_doc_links, peer tools (message_volk, get_volk_status, publish_topic, subscribe_topic, unsubscribe_topic), web tools (web_search, web_fetch) and design_patterns. ALL_TOOLS lists them, and AGENT_TOOLS maps each volk alias to the tool names it may use; get_agent_tools returns the definitions for one alias.

ToolExecutor is built over a workspace path, an optional Postman handle, the calling volk's alias and thread id, and an optional volk registry for status probes. execute dispatches a tool name and arguments and returns a ToolResult. Paths are confined to the workspace by _safe_path, and run_shell rejects commands in a BLOCKED_COMMANDS set. validate_mermaid checks structure only, without rendering: the header must name a known diagram type, brackets must balance, and a code fence is stripped. web_search uses DuckDuckGo's HTML endpoint through _DDGResultParser, and web_fetch extracts visible text with _HTMLTextExtractor; both refuse loopback, private, link-local and reserved addresses (_is_blocked_address), with an environment-controlled escape hatch read at call time (_web_tools_allow_private). design_patterns reads a cached catalog from configs/cloud-design-patterns.yaml. build_tool_registry and register_native_tools expose the legacy tools and the native tools through the unified ToolRegistry, and _ApprovalOverrideTool forces approval for selected native tools.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

_DDGResultParser

Bases: HTMLParser · Line 810

Tolerant parser for DuckDuckGo's HTML search-results markup.

_HTMLTextExtractor

Bases: HTMLParser · Line 867

Strip scripts/styles/tags from an HTML document, keeping visible text.

ToolExecutor

Line 1033

Thread-safe executor that maps tool names to sandboxed Python functions.

_ApprovalOverrideTool

Bases: BaseTool · Line 2013

Wrap a native BaseTool, forcing requires_approval=True.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/tool_registry.py

"""Agent tool registry — OpenAI/DeepSeek function-calling tool definitions.

Provides tool schemas and a thread-safe executor that maps tool names to
sandboxed Python functions. Agents wire these tools via
``BaseAgent._call_llm_with_tools()``.

Spec: SPEC-103 (Tool-Calling Refactoring) FR-005
"""

from __future__ import annotations

import asyncio
import dataclasses
import ipaddress
import logging
import os
import re
import signal
import socket
import threading
import urllib.parse
from collections.abc import Mapping, Sequence
from html.parser import HTMLParser
from pathlib import Path
from typing import Any

import httpx
import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import VolkHealthProbe
from orchestrata_ultimate.tools.approval import PrologApprovalGate
from orchestrata_ultimate.tools.base import (
    BaseTool,
    FunctionToolAdapter,
    ToolCategory,
    ToolMetadata,
    ToolResult,
)
from orchestrata_ultimate.tools.registry import ToolRegistry

_logger = logging.getLogger(__name__)


# ── Tool Result ──────────────────────────────────────────────────────
# ``ToolResult`` is the canonical type from the unified tool framework
# (``orchestrata_ultimate.tools.base``); imported above and re-exported here so
# existing ``from ...agents.tool_registry import ToolResult`` imports keep working.


# ── Tool Definitions (OpenAI/DeepSeek function format) ────────────────


READ_FILE_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "read_file",
        "description": "Read contents of a file. Returns file content as text.",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Path to the file relative to the project workspace.",
                },
                "start_line": {
                    "type": "integer",
                    "description": "Starting line number (1-based, default 1).",
                },
                "max_lines": {
                    "type": "integer",
                    "description": "Maximum lines to return (default 200).",
                },
            },
            "required": ["path"],
        },
    },
}

WRITE_FILE_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "write_file",
        "description": "Create or overwrite a file with content. Creates parent directories if needed.",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Path to the file relative to the project workspace.",
                },
                "content": {
                    "type": "string",
                    "description": "Full file content to write.",
                },
            },
            "required": ["path", "content"],
        },
    },
}

LIST_DIR_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "list_dir",
        "description": "List files and directories in a directory.",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Directory path relative to the project workspace (default: root).",
                },
            },
            "required": [],
        },
    },
}

SEARCH_CODE_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "search_code",
        "description": "Search for a regex pattern in project source files. Returns matching lines with context.",
        "parameters": {
            "type": "object",
            "properties": {
                "pattern": {
                    "type": "string",
                    "description": "Regular expression pattern to search for.",
                },
                "path": {
                    "type": "string",
                    "description": "Directory or file to search (relative to workspace, default: src/).",
                },
                "include": {
                    "type": "string",
                    "description": "File pattern to include (e.g. '*.py').",
                },
            },
            "required": ["pattern"],
        },
    },
}

RUN_SHELL_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "run_shell",
        "description": "Execute a shell command and return stdout. Commands run in a sandboxed subprocess with a 30-second timeout. Only for safe operations (formatting, linting, testing, git status).",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "Shell command to execute (e.g. 'ruff check src/', 'git diff', 'pytest tests/ -x').",
                },
                "cwd": {
                    "type": "string",
                    "description": "Working directory for the command (default: workspace root).",
                },
            },
            "required": ["command"],
        },
    },
}

AST_QUERY_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "ast_query",
        "description": "Query the project's serialized AST for class structure, methods, inheritance, and dependencies.",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "One of: summary, find <class>, largest <n>, search <pattern>, inheritance <class>, protocols, abstract, pydantic, stubs, modules, stats.",
                },
                "class_name": {
                    "type": "string",
                    "description": "Class name for find/inheritance commands.",
                },
            },
            "required": ["command"],
        },
    },
}

CHECK_COVERAGE_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "check_coverage",
        "description": "Check test coverage for the project. Returns covered/untested classes and coverage percentage.",
        "parameters": {
            "type": "object",
            "properties": {},
            "required": [],
        },
    },
}

VALIDATE_MERMAID_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "validate_mermaid",
        "description": (
            "Validate a Mermaid diagram's structure without rendering it. "
            "Checks that the diagram declares a recognized type (classDiagram, "
            "flowchart, graph, sequenceDiagram, stateDiagram, erDiagram, etc.) "
            "and that its braces/brackets/parentheses are balanced. Use this to "
            "verify a diagram before writing it into a doc so downstream readers "
            "get a well-formed diagram."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "diagram": {
                    "type": "string",
                    "description": (
                        "The Mermaid diagram source (with or without a "
                        "```mermaid fence)."
                    ),
                },
            },
            "required": ["diagram"],
        },
    },
}

CHECK_DOC_LINKS_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "check_doc_links",
        "description": (
            "Check that the local file links in a Markdown document actually "
            "resolve on disk. Parses '[text](target)' links and, for every LOCAL "
            "target (not an http/https/mailto URL and not a bare '#anchor'), "
            "resolves it relative to 'base_dir' inside the workspace and reports "
            "any referenced files that are missing. Pure filesystem check, no "
            "network. Use this before publishing a doc so its cross-references "
            "don't dangle."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "content": {
                    "type": "string",
                    "description": "The Markdown document text to scan for links.",
                },
                "base_dir": {
                    "type": "string",
                    "description": (
                        "Directory the doc lives in, relative to the workspace; "
                        "local links resolve against it (default '.')."
                    ),
                },
            },
            "required": ["content"],
        },
    },
}

MESSAGE_VOLK_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "message_volk",
        "description": (
            "Send a message to another volk (agent) and get their reply. Use this "
            "to ask a peer for information, delegate a sub-task, or coordinate — "
            "e.g. the Architect asking the BA to clarify a requirement. Address the "
            "recipient by their alias (U:Cx, U:BA, U:PM, U:Ux, U:Ar, U:Cd, U:Qa, "
            "J:Junior, U:Do)."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "to_volk": {
                    "type": "string",
                    "description": "Alias of the recipient volk (e.g. 'U:BA').",
                },
                "message": {
                    "type": "string",
                    "description": "The question or instruction to send.",
                },
                "subject": {
                    "type": "string",
                    "description": "Short subject/intent for the message (default 'query').",
                },
            },
            "required": ["to_volk", "message"],
        },
    },
}

PUBLISH_TOPIC_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "publish_topic",
        "description": (
            "Publish an event to every volk subscribed to a topic (SPEC-002 "
            "FR-005/FR-006 pub/sub). Use this to broadcast a notification or "
            "status update to whichever peers care about it, instead of "
            "message_volk's one-to-one addressing — e.g. U:Cd announcing "
            "'build.failed' so any volk subscribed to 'build.*' hears about "
            "it. Delivery is filtered per-subscriber by their subscription's "
            "filters."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "topic": {
                    "type": "string",
                    "description": "Exact topic name to publish on (e.g. 'build.failed').",
                },
                "message": {
                    "type": "string",
                    "description": "The event text delivered to subscribers.",
                },
                "subject": {
                    "type": "string",
                    "description": (
                        "Optional short subject for the event, usable by "
                        "subscribers' 'subject' filters (glob-matched)."
                    ),
                },
            },
            "required": ["topic", "message"],
        },
    },
}

SUBSCRIBE_TOPIC_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "subscribe_topic",
        "description": (
            "Subscribe to a topic pattern (SPEC-002 FR-006) so future "
            "publish_topic events matching it are delivered to you. Supports "
            "fnmatch-style wildcards (e.g. 'build.*') and optional per-key "
            "filters (e.g. {'subject': 'release-*'}) to narrow delivery "
            "further."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "topic_pattern": {
                    "type": "string",
                    "description": "Topic pattern to subscribe to (wildcards allowed).",
                },
                "filters": {
                    "type": "object",
                    "description": (
                        "Optional key-value predicates a publish payload must "
                        "match (AND semantics) to be delivered."
                    ),
                },
                "ttl_seconds": {
                    "type": "integer",
                    "description": "Optional subscription time-to-live in seconds.",
                },
            },
            "required": ["topic_pattern"],
        },
    },
}

UNSUBSCRIBE_TOPIC_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "unsubscribe_topic",
        "description": (
            "Unsubscribe from a topic pattern you previously subscribed to "
            "with subscribe_topic (SPEC-002 FR-006), stopping further "
            "publish_topic deliveries for that pattern."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "topic_pattern": {
                    "type": "string",
                    "description": "The exact topic pattern used when subscribing.",
                },
            },
            "required": ["topic_pattern"],
        },
    },
}

GET_VOLK_STATUS_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "get_volk_status",
        "description": (
            "Query the registry for the current health/status of a peer volk. "
            "Returns registration info, heartbeat age, and whether the volk is "
            "considered healthy. Use before message_volk to decide if a peer is "
            "responsive, or to monitor orchestrator↔volk liveness."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "alias": {
                    "type": "string",
                    "description": "Alias of the volk to check (e.g. 'U:BA').",
                },
            },
            "required": ["alias"],
        },
    },
}

WEB_SEARCH_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "web_search",
        "description": (
            "Search the public web (keyless, via a DuckDuckGo HTML backend). Use "
            "this for current external facts, third-party API docs, or "
            "version-sensitive information that isn't in this codebase or your "
            "training data — then cite the URL in your answer. Returns a "
            "numbered list of title/url/snippet results; follow up with "
            "web_fetch to read a specific page in full."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query.",
                },
                "max_results": {
                    "type": "integer",
                    "description": "Maximum results to return (1-10, default 5).",
                },
            },
            "required": ["query"],
        },
    },
}

WEB_FETCH_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "web_fetch",
        "description": (
            "Fetch a web page or API resource by URL (http/https only) and "
            "return its readable text (HTML is stripped to plain text; JSON and "
            "other text content is returned as-is). Use this to read a page "
            "found via web_search for current external facts, third-party API "
            "docs, or version-sensitive information — then cite the URL in your "
            "answer."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "The http(s) URL to fetch.",
                },
            },
            "required": ["url"],
        },
    },
}

DESIGN_PATTERNS_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "design_patterns",
        "description": (
            "Look up canonical Azure/cloud design patterns, domain guides "
            "(HPC, IAM, RBAC, networking, web hosting, SaaS multitenancy, "
            "PaaS, storage), and performance antipatterns. Use when designing "
            "systems involving infrastructure, distributed reliability, "
            "messaging, storage selection, identity, or multitenancy — cite "
            "the returned URL for every pattern you apply."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": (
                        "Free-text search matched case-insensitively against "
                        "pattern/guide name, problem, solution, use_when, and "
                        "summary fields. Empty string with a category returns "
                        "every entry in that category."
                    ),
                },
                "category": {
                    "type": "string",
                    "description": (
                        "Optional exact category filter (e.g. 'reliability', "
                        "'messaging', 'identity', 'networking', 'storage', "
                        "'hpc', 'saas-multitenancy')."
                    ),
                },
            },
            "required": ["query"],
        },
    },
}

# All available tools
ALL_TOOLS: list[dict[str, Any]] = [
    READ_FILE_TOOL,
    WRITE_FILE_TOOL,
    LIST_DIR_TOOL,
    SEARCH_CODE_TOOL,
    RUN_SHELL_TOOL,
    AST_QUERY_TOOL,
    CHECK_COVERAGE_TOOL,
    VALIDATE_MERMAID_TOOL,
    CHECK_DOC_LINKS_TOOL,
    MESSAGE_VOLK_TOOL,
    PUBLISH_TOPIC_TOOL,
    SUBSCRIBE_TOPIC_TOOL,
    UNSUBSCRIBE_TOPIC_TOOL,
    GET_VOLK_STATUS_TOOL,
    WEB_SEARCH_TOOL,
    WEB_FETCH_TOOL,
    DESIGN_PATTERNS_TOOL,
]

# Names of the legacy ALL_TOOLS function-schemas — used to tell a native
# SPEC-062 BaseTool (repl/lsp/worktree/task/remote_trigger/schedule_cron/
# mcp-resource) apart from a legacy tool when building an alias's schema list.
_LEGACY_TOOL_NAMES: frozenset[str] = frozenset(t["function"]["name"] for t in ALL_TOOLS)

# Agent → tool mapping
AGENT_TOOLS: dict[str, list[str]] = {
    "U:Cx": [
        "search_code",
        "ast_query",
        "web_search",
        "web_fetch",
    ],  # Elicitation — needs context
    "U:BA": [
        "search_code",
        "ast_query",
        "read_file",
        "web_search",
        "web_fetch",
        "design_patterns",
        "list_mcp_resources",
        "read_mcp_resource",
        "search_tools",
        "codebase_impact",
        "codebase_module_deps",
    ],  # Analysis — needs source; blast radius + dependency shape for scoping
    # (cycles/dep-path are engineering-facing, not analysis-facing)
    "U:PM": [
        "ast_query",
        "read_file",
        "web_search",
        "web_fetch",
        "task_stop",
        "task_output",
        "remote_trigger",
        "schedule_cron",
    ],  # Scoping — reads plans
    "U:Ar": [
        "read_file",
        "search_code",
        "ast_query",
        "list_dir",
        "web_search",
        "web_fetch",
        "design_patterns",
        "lsp",
        "list_mcp_resources",
        "read_mcp_resource",
        "get_code",
        "get_tests",
        "get_requirements",
        "search_code_semantic",
        "search_tools",
        "codebase_impact",
        "codebase_module_deps",
        "codebase_find_cycles",
        "codebase_dep_path",
    ],  # Design — full codebase access; the impact-closure consumer par
    # excellence (assessing a design change's blast radius/cycles/dep chains)
    "U:Cd": [
        "read_file",
        "write_file",
        "list_dir",
        "search_code",
        "ast_query",
        "run_shell",
        "web_search",
        "web_fetch",
        "design_patterns",
        "enter_worktree",
        "exit_worktree",
        "task_stop",
        "task_output",
        "lsp",
        "schedule_cron",
        "get_code",
        "get_tests",
        "get_requirements",
        "search_code_semantic",
        "search_tools",
        "codebase_impact",
        "codebase_module_deps",
        "codebase_find_cycles",
        "codebase_dep_path",
    ],  # Implementation — full access; scoping a refactor's blast radius
    "U:Qa": [
        "read_file",
        "write_file",
        "run_shell",
        "search_code",
        "check_coverage",
        "web_search",
        "web_fetch",
        "repl",
        "lsp",
        "get_code",
        "get_tests",
        "get_requirements",
        "search_code_semantic",
    ],  # QA — test execution
    "U:Cp": [],  # Deterministic — no LLM
    "J:Junior": [
        "read_file",
        "write_file",
        "run_shell",
        "list_dir",
        "search_code",
        "web_search",
        "web_fetch",
        "design_patterns",
        "repl",
        "lsp",
    ],  # Junior — code generation
    "U:Ux": [
        "read_file",
        "search_code",
        "web_search",
        "web_fetch",
    ],  # UX — reads requirements
    "U:Do": [
        "read_file",
        "write_file",
        "list_dir",
        "search_code",
        "ast_query",
        "validate_mermaid",
        "check_doc_links",
        "web_search",
        "web_fetch",
        "design_patterns",
        "list_mcp_resources",
        "read_mcp_resource",
        "get_code",
        "get_requirements",
        "search_code_semantic",
    ],  # Documentor — docs gen (validates the Mermaid it emits)
    "U:SD": [
        "repl",
        "enter_worktree",
        "exit_worktree",
        "lsp",
        "web_search",
        "web_fetch",
        "codebase_impact",
        "codebase_module_deps",
        "codebase_find_cycles",
        "codebase_dep_path",
    ],  # Senior Dev — oscillation-recovery debugging (SPEC-062 tool wave); every
    # tool-capable volk carries the baseline web-research pair (dogfood F11),
    # matching every other non-empty entry in this map.
}

# Every LLM-driven volk can message its peers directly (message_volk), use
# the SPEC-002 pub/sub fabric (publish_topic/subscribe_topic/unsubscribe_topic),
# and query peer status (get_volk_status) before messaging; U:Cp is
# deterministic (no LLM, no tools). Applied here so a new volk added above
# automatically gains messaging + pub/sub + status awareness.
for _tools in AGENT_TOOLS.values():
    if not _tools:
        continue
    for _pubsub_tool in (
        "message_volk",
        "publish_topic",
        "subscribe_topic",
        "unsubscribe_topic",
        "get_volk_status",
    ):
        if _pubsub_tool not in _tools:
            _tools.append(_pubsub_tool)


# ── Mermaid diagram validation helpers ──────────────────────────────
# Extracted from ToolExecutor._validate_mermaid — pure functions, no state.


def _is_mermaid_content_line(ln: str) -> bool:
    """True for a non-empty, non-comment (``%%``) Mermaid source line."""
    s = ln.strip()
    return bool(s) and not s.startswith("%%")


def _strip_mermaid_fence(lines: Sequence[str]) -> Sequence[str]:
    """Strip an optional ```` ```mermaid ... ``` ```` code fence."""
    if lines and lines[0].lstrip().startswith("```"):
        lines = lines[1:]
        if lines and lines[-1].strip().startswith("```"):
            lines = lines[:-1]
    return lines


def _mermaid_header(lines: Sequence[str]) -> str:
    """First non-empty, non-comment line — expected to declare the diagram type."""
    return next((ln.strip() for ln in lines if _is_mermaid_content_line(ln)), "")


def _count_mermaid_content_lines(lines: Sequence[str]) -> int:
    return sum(1 for ln in lines if _is_mermaid_content_line(ln))


def _check_balanced_mermaid_brackets(lines: Sequence[str]) -> ToolResult | None:
    """Balanced brackets/braces/parens across the diagram body, or an error
    ``ToolResult`` describing the first mismatch found.
    """
    pairs = {")": "(", "]": "[", "}": "{"}
    openers = set(pairs.values())
    stack: list[str] = []
    for ch in "\n".join(lines):
        if ch in openers:
            stack.append(ch)
        elif ch in pairs:
            if not stack or stack[-1] != pairs[ch]:
                return ToolResult(
                    "validate_mermaid",
                    False,
                    "",
                    f"Unbalanced '{ch}' — no matching '{pairs[ch]}'.",
                )
            stack.pop()
    if stack:
        return ToolResult(
            "validate_mermaid",
            False,
            "",
            f"Unclosed '{stack[-1]}' — {len(stack)} bracket(s) never closed.",
        )
    return None


# ── Web tools helpers ────────────────────────────────────────────────
# Keyless web research for volks (dogfood finding F11): a DuckDuckGo HTML
# search backend plus a governed page fetcher, both parsed with the stdlib
# ``html.parser`` (no bs4/lxml dependency).

_DESKTOP_USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
)
_WEB_TIMEOUT_SECONDS = 15.0
_MAX_FETCH_BYTES = 500_000
_MAX_FETCH_CHARS = 20_000

# HTML void elements never carry a matching closing tag — excluded from depth
# tracking so a stray ``<br>``/``<img>`` inside a captured region doesn't leave
# the parser "stuck" thinking it's still inside an unclosed element.
_VOID_HTML_ELEMENTS: frozenset[str] = frozenset(
    {
        "area",
        "base",
        "br",
        "col",
        "embed",
        "hr",
        "img",
        "input",
        "link",
        "meta",
        "param",
        "source",
        "track",
        "wbr",
    }
)


def _resolve_ddg_href(href: str) -> str:
    """Resolve a DuckDuckGo result href to the real target URL.

    DuckDuckGo's HTML endpoint wraps result links in a redirect of the form
    ``//duckduckgo.com/l/?uddg=<urlencoded-real-url>&...``; decode the ``uddg``
    parameter when present, otherwise return the href unchanged (normalizing a
    protocol-relative ``//host/...`` href to ``https://``).
    """
    href = href.strip()
    if not href:
        return ""
    normalized = f"https:{href}" if href.startswith("//") else href
    parsed = urllib.parse.urlsplit(normalized)
    if "duckduckgo.com" in parsed.netloc and parsed.path.startswith("/l/"):
        real = urllib.parse.parse_qs(parsed.query).get("uddg", [""])[0]
        if real:
            return urllib.parse.unquote(real)
    return normalized


class _DDGResultParser(HTMLParser):
    """Tolerant parser for DuckDuckGo's HTML search-results markup.

    Extracts (title, url, snippet) triples from anchors carrying the
    ``result__a`` class (title text + href) and the sibling element carrying
    ``result__snippet`` (the snippet text, which may itself nest inline tags
    such as ``<b>`` around highlighted terms).
    """

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.results: list[dict[str, str]] = []
        self._current: dict[str, str] | None = None
        self._capture: str | None = None
        self._capture_stack: list[str] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        attrs_dict = dict(attrs)
        classes = (attrs_dict.get("class") or "").split()

        if self._capture is not None:
            if tag not in _VOID_HTML_ELEMENTS:
                self._capture_stack.append(tag)
            return

        if tag == "a" and "result__a" in classes:
            href = attrs_dict.get("href") or ""
            self._current = {
                "title": "",
                "url": _resolve_ddg_href(href),
                "snippet": "",
            }
            self._capture = "title"
            self._capture_stack = [tag]
        elif (
            "result__snippet" in classes
            and self._current is not None
            and tag not in _VOID_HTML_ELEMENTS
        ):
            self._capture = "snippet"
            self._capture_stack = [tag]

    def handle_endtag(self, tag: str) -> None:
        if self._capture is None or not self._capture_stack:
            return
        if self._capture_stack[-1] == tag:
            self._capture_stack.pop()
        if not self._capture_stack:
            if self._capture == "title" and self._current is not None:
                self.results.append(self._current)
            self._capture = None

    def handle_data(self, data: str) -> None:
        if self._capture and self._current is not None:
            self._current[self._capture] += data


class _HTMLTextExtractor(HTMLParser):
    """Strip scripts/styles/tags from an HTML document, keeping visible text."""

    _SKIP_TAGS = frozenset({"script", "style", "noscript"})

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self._chunks: list[str] = []
        self._skip_depth = 0

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag in self._SKIP_TAGS:
            self._skip_depth += 1

    def handle_endtag(self, tag: str) -> None:
        if tag in self._SKIP_TAGS and self._skip_depth > 0:
            self._skip_depth -= 1

    def handle_data(self, data: str) -> None:
        if self._skip_depth == 0:
            self._chunks.append(data)

    def get_text(self) -> str:
        return " ".join("".join(self._chunks).split())


def _extract_html_text(html_source: str) -> str:
    """Strip an HTML document down to its collapsed, visible text."""
    parser = _HTMLTextExtractor()
    parser.feed(html_source)
    return parser.get_text()


def _is_blocked_address(ip_str: str) -> bool:
    """True if ``ip_str`` is loopback/private/link-local/reserved (or unparsable)."""
    try:
        addr = ipaddress.ip_address(ip_str)
    except ValueError:
        return True  # can't parse -> treat as unsafe
    return addr.is_loopback or addr.is_private or addr.is_link_local or addr.is_reserved


def _web_tools_allow_private(env: Mapping[str, str] | None = None) -> bool:
    """Escape hatch for the SSRF guard, read at call time (project convention).

    Lets tests (and deliberately local-service deployments) target
    loopback/private addresses with ``web_fetch``; unset/falsy by default.
    An optional ``env`` mapping can be supplied by callers (e.g. tests) to
    avoid mutating the process environment.
    """
    env = os.environ if env is None else env
    return env.get("WEB_TOOLS_ALLOW_PRIVATE", "").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    )


# ── Design pattern catalog helpers ───────────────────────────────────
# Backs the ``design_patterns`` tool: a lazily-loaded, process-cached view of
# ``configs/cloud-design-patterns.yaml`` (43 Azure/cloud design patterns, 8
# domain guides, 10 performance antipatterns). Located by walking up from this
# file (not cwd) so the tool works regardless of the caller's workspace.

_DESIGN_PATTERNS_CATALOG: dict[str, list[dict[str, Any]]] | None = None
_DESIGN_PATTERNS_LOCK = threading.Lock()

# Fields checked for a case-insensitive free-text match. Every entry type
# (pattern / domain guide / antipattern) only populates the fields it has —
# missing fields contribute nothing to the match.
_DESIGN_PATTERNS_QUERY_FIELDS: tuple[str, ...] = (
    "name",
    "problem",
    "solution",
    "use_when",
    "summary",
)

_DESIGN_PATTERNS_MATCH_CAP = 8


def _find_cloud_design_patterns_yaml() -> Path | None:
    """Walk up from this file to find configs/cloud-design-patterns.yaml."""
    current = Path(__file__).resolve()
    for parent in current.parents:
        candidate = parent / "configs" / "cloud-design-patterns.yaml"
        if candidate.is_file():
            return candidate
        if (parent / ".git").exists():
            return candidate
    return None


def _load_design_patterns_catalog() -> dict[str, list[dict[str, Any]]]:
    """Load + cache ``configs/cloud-design-patterns.yaml`` (module-level, lazy)."""
    global _DESIGN_PATTERNS_CATALOG
    if _DESIGN_PATTERNS_CATALOG is not None:
        return _DESIGN_PATTERNS_CATALOG
    with _DESIGN_PATTERNS_LOCK:
        if _DESIGN_PATTERNS_CATALOG is not None:
            return _DESIGN_PATTERNS_CATALOG
        path = _find_cloud_design_patterns_yaml()
        data: dict[str, Any] = {}
        if path is not None and path.is_file():
            with open(path, encoding="utf-8") as fh:
                data = yaml.safe_load(fh) or {}
        _DESIGN_PATTERNS_CATALOG = {
            "patterns": list(data.get("patterns") or []),
            "domain_guides": list(data.get("domain_guides") or []),
            "antipatterns": list(data.get("antipatterns") or []),
        }
        return _DESIGN_PATTERNS_CATALOG


def _design_patterns_entry_matches(
    entry: dict[str, Any], query: str, category: str
) -> bool:
    """True if ``entry`` passes the (optional) category filter and free-text query.

    The query is split into whitespace terms and matches if ANY term appears
    in the entry's text — LLM callers send multi-word queries ("throttling
    noisy neighbor") that a whole-string substring test would never satisfy.
    """
    if category and str(entry.get("category") or "").lower() != category:
        return False
    if query:
        haystack = " ".join(
            str(entry.get(field) or "") for field in _DESIGN_PATTERNS_QUERY_FIELDS
        ).lower()
        if not any(term in haystack for term in query.split()):
            return False
    return True


def _format_design_pattern_block(pattern: dict[str, Any]) -> str:
    return (
        f"Pattern: {pattern.get('name', '')}\n"
        f"Category: {pattern.get('category', '')}\n"
        f"Problem: {pattern.get('problem', '')}\n"
        f"Solution: {pattern.get('solution', '')}\n"
        f"Use when: {pattern.get('use_when', '')}\n"
        f"URL: {pattern.get('url', '')}"
    )


def _format_domain_guide_block(guide: dict[str, Any]) -> str:
    urls = ", ".join(guide.get("urls") or [])
    return (
        f"Domain guide: {guide.get('name', '')}\n"
        f"Summary: {guide.get('summary', '')}\n"
        f"URLs: {urls}"
    )


def _format_antipattern_block(antipattern: dict[str, Any]) -> str:
    return (
        f"Antipattern: {antipattern.get('name', '')}\n"
        f"Problem: {antipattern.get('problem', '')}\n"
        f"URL: {antipattern.get('url', '')}"
    )


# ── Tool Executor ────────────────────────────────────────────────────


class ToolExecutor:
    """Thread-safe executor that maps tool names to sandboxed Python functions.

    Tool calls are validated for safety before execution. File operations
    are sandboxed to the workspace directory.
    """

    # Commands that are explicitly blocked in run_shell
    BLOCKED_COMMANDS: frozenset[str] = frozenset(
        {
            "rm -rf",
            "rm -r",
            "sudo",
            "chmod 777",
            "mkfs",
            "dd if=",
            ":(){ :|:& };:",
            "wget",
            "curl",
        }
    )

    def __init__(
        self,
        workspace: Path,
        postman: Any | None = None,
        from_alias: str | None = None,
        thread_id: str | None = None,
        volk_registry: VolkHealthProbe | None = None,
        codebase_kb: Any | None = None,
        codebase_kb_environ: dict[str, str] | None = None,
        coverage_timeout_s: float = 30.0,
        approval_gate: Any | None = None,
    ):
        self._workspace = workspace.resolve()
        # Per-call ceiling for the `check_coverage` subprocess. Default 30s is
        # production timing; a test can inject a small value to exercise the
        # timeout/killpg branch in ~2s instead of waiting the real 30s bound.
        self._coverage_timeout_s = coverage_timeout_s
        self._lock = threading.Lock()
        # Optional CXP messaging fabric so a volk can talk to its peers. When not
        # bound (e.g. a standalone tool run), message_volk returns a clear error.
        self._postman = postman
        self._from_alias = from_alias
        self._thread_id = thread_id
        # Optional registry for peer status/heartbeat queries.
        self._volk_registry = volk_registry
        # Test-injection seams for the codebase-intelligence query tools'
        # production KB lifecycle (see codebase_query_tools._build_production_kb)
        # -- both None in every real caller, so real callers get the real
        # env-resolved production KB on first codebase_* query.
        self._codebase_kb = codebase_kb
        self._codebase_kb_environ = codebase_kb_environ
        # Lazily built, this-executor-only registry of the SPEC-062 native
        # BaseTools (repl/worktree/lsp/task/remote_trigger/schedule_cron/
        # mcp-resource) — see ``_native_registry()`` below.
        self._native_registry: ToolRegistry | None = None
        # Approval gate for tools declaring `requires_approval`. None means
        # the production governance-Prolog gate, built lazily against this
        # executor's alias; tests inject one over a scratch rulebase.
        self._approval_gate = approval_gate

    def _safe_path(self, rel_path: str) -> Path:
        """Resolve a relative path and ensure it's within the workspace."""
        p = (self._workspace / rel_path).resolve()
        if not str(p).startswith(str(self._workspace)):
            raise ValueError(f"Path escapes workspace: {rel_path}")
        return p

    def _native_tools(self) -> ToolRegistry:
        """Lazily build (once, thread-safe) this executor's native-tool registry.

        Built from :func:`register_native_tools` so the dispatch bridge shares
        the exact same tool set (and approval semantics) as the SPEC-062
        unified registry (``build_tool_registry``), while keeping each
        executor's native tool instances (REPL sessions, TaskManager, ...)
        private to it rather than sharing mutable state process-wide.
        """
        if self._native_registry is not None:
            return self._native_registry
        with self._lock:
            if self._native_registry is None:
                # The gate is bound to THIS executor's volk alias, which is the
                # actor governance grants against. Without it the registry
                # fails closed and every approval-requiring native tool
                # (repl, remote_trigger) is refused.
                registry = ToolRegistry(
                    approval_gate=self._approval_gate
                    or PrologApprovalGate(self._from_alias)
                )
                register_native_tools(
                    registry,
                    codebase_kb=self._codebase_kb,
                    codebase_kb_environ=self._codebase_kb_environ,
                )
                self._native_registry = registry
            return self._native_registry

    async def execute(
        self,
        tool_name: str,
        arguments: dict[str, Any],
        env: Mapping[str, str] | None = None,
    ) -> ToolResult:
        """Execute a tool call and return the result."""
        # The legacy if/elif chain below dispatches 17 tools directly, never
        # through a ToolRegistry, so ToolRegistry's own approval enforcement
        # cannot see them. `run_shell` is one of them and is flagged
        # `requires_approval` (it takes a caller-supplied cwd and is a shell
        # regardless, so the workspace does not bound it) -- without this check
        # that flag would be decorative on the only path production uses.
        # Native tools take the fallthrough into `_native_tools()`, a real
        # registry, and are gated there.
        if tool_name in _LEGACY_APPROVAL_TOOLS:
            gate = self._approval_gate or PrologApprovalGate(self._from_alias)
            if not gate.approve(tool_name, dict(arguments)):
                return ToolResult(
                    tool_name,
                    False,
                    "",
                    f"{tool_name} requires approval and no "
                    f"may_execute_tool/2 fact grants it to "
                    f"{self._from_alias or '<unattributed caller>'}.",
                )
        try:
            if tool_name == "read_file":
                return await self._read_file(arguments)
            elif tool_name == "write_file":
                return await self._write_file(arguments)
            elif tool_name == "list_dir":
                return await self._list_dir(arguments)
            elif tool_name == "search_code":
                return await self._search_code(arguments)
            elif tool_name == "run_shell":
                return await self._run_shell(arguments)
            elif tool_name == "ast_query":
                return await self._ast_query(arguments)
            elif tool_name == "check_coverage":
                return await self._check_coverage(arguments)
            elif tool_name == "validate_mermaid":
                return await self._validate_mermaid(arguments)
            elif tool_name == "check_doc_links":
                return await self._check_doc_links(arguments)
            elif tool_name == "message_volk":
                return await self._message_volk(arguments)
            elif tool_name == "publish_topic":
                return await self._publish_topic(arguments)
            elif tool_name == "subscribe_topic":
                return await self._subscribe_topic(arguments)
            elif tool_name == "unsubscribe_topic":
                return await self._unsubscribe_topic(arguments)
            elif tool_name == "get_volk_status":
                return await self._get_volk_status(arguments)
            elif tool_name == "web_search":
                return await self._web_search(arguments, env=env)
            elif tool_name == "web_fetch":
                return await self._web_fetch(arguments, env=env)
            elif tool_name == "design_patterns":
                return await self._design_patterns(arguments)
            elif self._native_tools().has(tool_name):
                # Fallthrough: a SPEC-062 native BaseTool (tool wave) rather
                # than one of the legacy handlers above. ``ToolRegistry.execute``
                # already returns the identical ``ToolResult`` type these legacy
                # handlers do, so no adaptation is needed.
                return await self._native_tools().execute(tool_name, arguments)
            else:
                return ToolResult(tool_name, False, "", f"Unknown tool: {tool_name}")
        except Exception as e:
            _logger.warning("Tool %s failed: %s", tool_name, e)
            return ToolResult(tool_name, False, "", str(e))

    async def _message_volk(self, args: dict) -> ToolResult:
        """Send a CXP message to a peer volk and return their reply text."""
        if self._postman is None:
            return ToolResult(
                "message_volk",
                False,
                "",
                "Messaging unavailable: no postman bound to this run.",
            )
        to_volk = (args.get("to_volk") or args.get("to_agent") or "").strip()
        if not to_volk:
            return ToolResult(
                "message_volk", False, "", "message_volk requires 'to_volk'."
            )

        # Peer-status awareness: warn when the recipient is known-unhealthy,
        # but still attempt delivery (best-effort messaging).
        health_warning = ""
        if self._volk_registry is not None:
            try:
                health = self._volk_registry.get_volk_health(to_volk)
                if health is not None and not health.is_healthy:
                    health_warning = (
                        f" [WARNING: {to_volk} is currently unhealthy"
                        f" ({health.status.value}"
                        f"{f', {health.error_message}' if health.error_message else ''})]"
                    )
            except Exception as e:
                _logger.debug("Health pre-check failed for %s: %s", to_volk, e)

        from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus
        from orchestrata_ultimate.postman.delivery_outcome import DeliveryFailure

        packet = CXPPacket(
            from_agent=self._from_alias or "UNKNOWN",
            to_agent=to_volk,
            subject=args.get("subject") or "query",
            payload={"text": args.get("message", "")},
            thread_id=self._thread_id or "",
            work_status=WorkStatus.IN_PROGRESS,
        )
        outcome = await self._postman.aroute_outcome(packet)
        if not outcome.delivered:
            # G4: an unroutable/failed delivery must never be reported to
            # the sending volk as delivered (REQ-002-009, REQ-002-035,
            # REQ-002-038, REQ-002-US-001-026).
            if outcome.reason is DeliveryFailure.HANDLER_ERROR:
                return ToolResult(
                    "message_volk",
                    False,
                    "",
                    f"{to_volk}'s handler raised an error: {outcome.error}",
                )
            return ToolResult(
                "message_volk",
                False,
                "",
                f"No route to {to_volk}: {outcome.error or 'no handler registered'}.",
            )

        # G6: successful delivery through the recipient's handler counts as
        # a heartbeat for the RECIPIENT (REQ-001-018) — best-effort, a
        # heartbeat failure must never fail the delivery result.
        if self._volk_registry is not None:
            try:
                self._volk_registry.record_heartbeat(to_volk)
            except Exception as e:
                _logger.debug("Heartbeat recording failed for %s: %s", to_volk, e)

        reply = outcome.response
        if reply is None:
            return ToolResult(
                "message_volk",
                True,
                f"Message delivered to {to_volk} (no reply).{health_warning}",
            )
        payload = getattr(reply, "payload", None)
        if isinstance(payload, dict):
            text = payload.get("text") or payload.get("reply") or str(payload)
        else:
            text = str(payload) if payload is not None else ""
        return ToolResult(
            "message_volk", True, f"{to_volk} replied: {text}{health_warning}"
        )

    async def _get_volk_status(self, args: dict) -> ToolResult:
        """Return the registry health/status of a peer volk."""
        if self._volk_registry is None:
            return ToolResult(
                "get_volk_status",
                False,
                "",
                "Volk status unavailable: no registry bound to this run.",
            )
        alias = (args.get("alias") or "").strip()
        if not alias:
            return ToolResult(
                "get_volk_status", False, "", "get_volk_status requires 'alias'."
            )

        if not self._volk_registry.is_registered(alias):
            return ToolResult(
                "get_volk_status", True, f"Volk {alias} is not registered."
            )

        health = self._volk_registry.get_volk_health(alias)
        if health is None:
            return ToolResult(
                "get_volk_status", True, f"Volk {alias} has no health record."
            )

        status = getattr(health, "status", None)
        is_healthy = getattr(health, "is_healthy", False)
        age = getattr(health, "heartbeat_age_seconds", None)
        error = getattr(health, "error_message", None)
        lines = [
            f"Volk: {alias}",
            f"Status: {status.value if status is not None else 'unknown'}",
            f"Healthy: {is_healthy}",
        ]
        if age is not None:
            lines.append(f"Heartbeat age (s): {age:.1f}")
        if error:
            lines.append(f"Error: {error}")
        return ToolResult("get_volk_status", True, "\n".join(lines))

    @traces(
        spec="002",
        story="US-002",
        reqs=["REQ-002-US-002-007", "REQ-002-US-001-016"],
    )
    async def _publish_topic(self, args: dict) -> ToolResult:
        """Publish an event to every volk subscribed to a topic (SPEC-002)."""
        if self._postman is None:
            return ToolResult(
                "publish_topic",
                False,
                "",
                "Publishing unavailable: no postman bound to this run.",
            )
        topic = (args.get("topic") or "").strip()
        if not topic:
            return ToolResult(
                "publish_topic", False, "", "publish_topic requires 'topic'."
            )

        payload: dict[str, Any] = {"text": args.get("message", "")}
        if args.get("subject"):
            payload["subject"] = args["subject"]

        result = await self._postman.publish_topic(
            topic,
            payload,
            from_agent=self._from_alias or "UNKNOWN",
            thread_id=self._thread_id or "",
        )
        return ToolResult(
            "publish_topic",
            True,
            f"Published to '{topic}': {result['matched']} matched, "
            f"{len(result['delivered'])} delivered, {len(result['failed'])} failed.",
        )

    @traces(spec="002", story="US-002-001", reqs=["REQ-002-US-002-001"])
    async def _subscribe_topic(self, args: dict) -> ToolResult:
        """Subscribe this volk to a topic pattern (SPEC-002 FR-006)."""
        if self._postman is None:
            return ToolResult(
                "subscribe_topic",
                False,
                "",
                "Subscribing unavailable: no postman bound to this run.",
            )
        topic_pattern = (args.get("topic_pattern") or "").strip()
        if not topic_pattern:
            return ToolResult(
                "subscribe_topic",
                False,
                "",
                "subscribe_topic requires 'topic_pattern'.",
            )
        filters = args.get("filters")
        if not isinstance(filters, dict):
            filters = None
        subscription = await self._postman.subscription_manager.subscribe(
            self._from_alias or "UNKNOWN",
            topic_pattern,
            ttl_seconds=args.get("ttl_seconds"),
            filters=filters,
        )
        return ToolResult(
            "subscribe_topic",
            True,
            f"Subscribed to '{topic_pattern}' "
            f"(subscription_id={subscription.subscription_id}).",
        )

    @traces(spec="002", story="US-002-001", reqs=["REQ-002-US-002-002"])
    async def _unsubscribe_topic(self, args: dict) -> ToolResult:
        """Unsubscribe this volk from a topic pattern (SPEC-002 FR-006)."""
        if self._postman is None:
            return ToolResult(
                "unsubscribe_topic",
                False,
                "",
                "Unsubscribing unavailable: no postman bound to this run.",
            )
        topic_pattern = (args.get("topic_pattern") or "").strip()
        if not topic_pattern:
            return ToolResult(
                "unsubscribe_topic",
                False,
                "",
                "unsubscribe_topic requires 'topic_pattern'.",
            )

        from orchestrata_ultimate.postman.exceptions import SubscriptionNotFoundError

        try:
            await self._postman.subscription_manager.unsubscribe(
                self._from_alias or "UNKNOWN", topic_pattern
            )
        except SubscriptionNotFoundError as e:
            return ToolResult("unsubscribe_topic", False, "", str(e))
        return ToolResult(
            "unsubscribe_topic", True, f"Unsubscribed from '{topic_pattern}'."
        )

    async def _read_file(self, args: dict) -> ToolResult:
        path = self._safe_path(args["path"])
        start = args.get("start_line", 1)
        max_lines = args.get("max_lines", 200)
        if not path.exists():
            return ToolResult("read_file", False, "", f"File not found: {args['path']}")
        lines = path.read_text(encoding="utf-8").splitlines()
        selected = lines[start - 1 : start - 1 + max_lines]
        return ToolResult("read_file", True, "\n".join(selected))

    async def _write_file(self, args: dict) -> ToolResult:
        path = self._safe_path(args["path"])
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(args["content"], encoding="utf-8")
        return ToolResult(
            "write_file", True, f"Wrote {len(args['content'])} bytes to {args['path']}"
        )

    async def _list_dir(self, args: dict) -> ToolResult:
        rel = args.get("path", ".")
        p = self._safe_path(rel) if rel != "." else self._workspace
        # ``iterdir()`` yields absolute paths (since ``p`` is absolute), so
        # displaying the entry itself — instead of its bare ``.name`` —  would
        # leak the sandbox's real filesystem path to the calling agent. Every
        # sibling tool (read_file/write_file/search_code) is workspace-relative;
        # list_dir must be too.
        entries = sorted(
            [f"{'[D]' if e.is_dir() else '[F]'} {e.name}" for e in p.iterdir()],
            key=lambda x: (x[0] != "[D]", x),
        )
        return ToolResult("list_dir", True, "\n".join(entries[:100]))

    async def _search_code(self, args: dict) -> ToolResult:
        pattern = args["pattern"]
        search_path = self._safe_path(args.get("path", "src"))
        include = args.get("include", "*.py")

        results = []
        try:
            # Use grep for speed
            cmd = ["grep", "-rn", "--include", include, pattern, str(search_path)]
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, _ = await proc.communicate()
            results = stdout.decode("utf-8", errors="replace").splitlines()[:50]
        except Exception as e:
            return ToolResult("search_code", False, "", str(e))

        return ToolResult(
            "search_code", True, "\n".join(results) or "No matches found."
        )

    async def _run_shell(self, args: dict) -> ToolResult:
        command = args["command"]
        cwd = args.get("cwd", str(self._workspace))

        # Safety check
        lower = command.lower()
        for blocked in self.BLOCKED_COMMANDS:
            if blocked in lower:
                return ToolResult("run_shell", False, "", f"Blocked command: {blocked}")

        try:
            proc = await asyncio.create_subprocess_shell(
                command,
                cwd=cwd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
            out = stdout.decode("utf-8", errors="replace")
            err = stderr.decode("utf-8", errors="replace")
            if proc.returncode != 0:
                return ToolResult(
                    "run_shell",
                    False,
                    out[:2000],
                    err[:500] or f"Exit code: {proc.returncode}",
                )
            return ToolResult("run_shell", True, out[:2000])
        except TimeoutError:
            return ToolResult(
                "run_shell", False, "", "Command timed out after 30 seconds."
            )
        except Exception as e:
            return ToolResult("run_shell", False, "", str(e) or type(e).__name__)

    async def _ast_query(self, args: dict) -> ToolResult:
        command = args.get("command", "summary")
        class_name = args.get("class_name", "")

        cmd = ["uv", "--quiet", "run", "ast-query"]
        if (self._workspace / ".venv" / "bin" / "ast-query").exists():
            # `uv run` implicitly re-syncs the venv (rebuilding optional/
            # heavy deps like tree-sitter-* from source) before exec'ing the
            # console script. That sync is unrelated to what ast-query
            # itself needs and can fail on hosts where those builds are
            # broken (e.g. tree-sitter-typescript/tree-sitter-cpp vs. a
            # newer GCC). Skip it when the workspace's own venv already has
            # ast-query built — there's nothing to sync for this call. Leave
            # syncing on otherwise (e.g. a fresh ephemeral project whose own
            # ast-query script hasn't been built yet) so it still gets built.
            cmd.insert(3, "--no-sync")
        parts = command.split()
        cmd.extend(parts)
        if class_name and "find" in command:
            cmd.append(class_name)
        elif class_name and "inheritance" in command:
            cmd.append(class_name)

        proc = None
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                cwd=str(self._workspace),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                start_new_session=True,
            )
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=15)
            out = stdout.decode("utf-8", errors="replace")
            err = stderr.decode("utf-8", errors="replace")
            if proc.returncode != 0 and not out.strip():
                if not err.strip():
                    err = f"ast-query exited {proc.returncode} with no output"
                return ToolResult("ast_query", False, out[:2000], err[:500])
            return ToolResult("ast_query", True, out[:2000])
        except TimeoutError:
            # Kill the whole process group too — wait_for only cancels the
            # wait, and `proc.kill()` alone would leave any grandchildren
            # (e.g. `uv run` -> ast-query) holding the pipes open, blocking
            # the post-kill `communicate()` below on their runtime (same
            # leak class fixed in `_check_coverage` above, 2026-08-04).
            if proc is not None and proc.returncode is None:
                try:
                    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
                except ProcessLookupError, PermissionError, OSError:
                    proc.kill()
                await proc.communicate()
            return ToolResult("ast_query", False, "", "AST query timed out.")
        except Exception as e:
            return ToolResult("ast_query", False, "", str(e) or type(e).__name__)

    async def _check_coverage(self, _args: dict) -> ToolResult:
        proc = None
        try:
            cmd = ["uv", "--quiet", "run", "test-gen", "--coverage"]
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                cwd=str(self._workspace),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                start_new_session=True,
            )
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=self._coverage_timeout_s
            )
            out = stdout.decode("utf-8", errors="replace")
            if proc.returncode != 0 and not out.strip():
                err = stderr.decode("utf-8", errors="replace").strip()
                if not err:
                    err = f"test-gen exited {proc.returncode} with no output"
                return ToolResult("check_coverage", False, "", err[:500])
            return ToolResult("check_coverage", True, out[:2000])
        except TimeoutError:
            # wait_for cancels the wait but NOT the child — kill it or the
            # coverage subprocess keeps burning CPU long after the tool call
            # (observed as an orphaned 88%-CPU test-gen, 2026-07-06). Note
            # str(asyncio.TimeoutError()) is "" — never stringify it as the
            # error message. `proc.kill()` alone only signals the immediate
            # `uv` process — `test-gen`/pytest grandchildren it spawned
            # survive, still holding the stdout/stderr pipes open, so the
            # post-kill `communicate()` below then blocks on those pipes for
            # the grandchildren's full remaining runtime (observed as a 107s
            # test instead of the 30s bound, 2026-08-04). `start_new_session=
            # True` above puts the whole tree in its own process group so
            # killpg takes out every descendant at once.
            if proc is not None and proc.returncode is None:
                try:
                    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
                except ProcessLookupError, PermissionError, OSError:
                    proc.kill()
                await proc.communicate()
            return ToolResult(
                "check_coverage",
                False,
                "",
                f"check_coverage timed out after {self._coverage_timeout_s}s",
            )
        except Exception as e:
            return ToolResult("check_coverage", False, "", str(e) or type(e).__name__)

    # Mermaid diagram types the validator recognizes as a valid header.
    _MERMAID_TYPES: frozenset[str] = frozenset(
        {
            "classDiagram",
            "flowchart",
            "graph",
            "sequenceDiagram",
            "stateDiagram",
            "stateDiagram-v2",
            "erDiagram",
            "journey",
            "gantt",
            "pie",
            "mindmap",
            "timeline",
            "gitGraph",
            "quadrantChart",
            "requirementDiagram",
            "C4Context",
            "sankey-beta",
            "xychart-beta",
        }
    )

    def _check_mermaid_type(self, first_token: str) -> ToolResult | None:
        """The header's first token must be a recognized diagram type."""
        if first_token in self._MERMAID_TYPES:
            return None
        return ToolResult(
            "validate_mermaid",
            False,
            "",
            (
                f"Unrecognized diagram type '{first_token or '(empty)'}'. "
                f"Expected one of: {', '.join(sorted(self._MERMAID_TYPES))}."
            ),
        )

    async def _validate_mermaid(self, args: dict) -> ToolResult:
        """Validate a Mermaid diagram's structure (no rendering).

        Checks the diagram declares a recognized type and has balanced
        braces/brackets/parentheses — the two failure modes that most often make
        an LLM-authored diagram fail to render.
        """
        raw = (args.get("diagram") or "").strip()
        if not raw:
            return ToolResult(
                "validate_mermaid", False, "", "validate_mermaid requires 'diagram'."
            )

        lines = _strip_mermaid_fence(raw.splitlines())
        header = _mermaid_header(lines)
        first_token = header.split()[0] if header else ""

        type_error = self._check_mermaid_type(first_token)
        if type_error is not None:
            return type_error

        bracket_error = _check_balanced_mermaid_brackets(lines)
        if bracket_error is not None:
            return bracket_error

        body_lines = _count_mermaid_content_lines(lines)
        return ToolResult(
            "validate_mermaid",
            True,
            f"Valid {first_token} diagram ({body_lines} non-empty lines, "
            f"brackets balanced).",
        )

    # Markdown inline-link pattern: ``[text](target)``. The captured target may
    # carry a ``"title"`` suffix and/or a ``#fragment``; both are stripped before
    # the file is resolved.
    _MD_LINK_RE = re.compile(r"\[[^\]]*\]\(\s*([^)]+?)\s*\)")

    async def _check_doc_links(self, args: dict) -> ToolResult:
        """Check that local file links in a Markdown doc resolve on disk.

        Parses ``[text](target)`` links and, for every LOCAL target (not an
        http/https/mailto URL and not a bare ``#anchor``), resolves it relative
        to ``base_dir`` inside the workspace via ``_safe_path`` and reports any
        referenced files that are missing. Pure filesystem check — no network.

        Succeeds whenever the request is well-formed (listing any missing files
        in the message); fails only on a malformed request (empty ``content``).
        """
        content = args.get("content") or ""
        if not content.strip():
            return ToolResult(
                "check_doc_links", False, "", "check_doc_links requires 'content'."
            )
        base_dir = args.get("base_dir") or "."

        checked = 0
        missing: list[str] = []
        for raw_target in self._MD_LINK_RE.findall(content):
            target = raw_target.strip()
            # Drop an optional link title: ``[text](path "Title")``.
            if " " in target:
                target = target.split()[0]
            low = target.lower()
            if low.startswith(("http://", "https://", "mailto:")):
                continue  # external link — no filesystem check
            if not target or target.startswith("#"):
                continue  # in-page anchor, nothing to resolve
            # Strip a ``#fragment`` on a file link (``file.md#section``).
            local = target.split("#", 1)[0].strip()
            if not local:
                continue
            checked += 1
            try:
                resolved = self._safe_path(str(Path(base_dir) / local))
            except ValueError:
                missing.append(local)  # escapes the workspace → unresolvable
                continue
            if not resolved.exists():
                missing.append(local)

        if missing:
            return ToolResult(
                "check_doc_links",
                True,
                f"{len(missing)} of {checked} local file link(s) missing: "
                + ", ".join(missing),
            )
        if checked == 0:
            return ToolResult("check_doc_links", True, "No local file links to check.")
        return ToolResult(
            "check_doc_links", True, f"All {checked} local file link(s) resolve."
        )

    async def _web_search(
        self, args: dict, env: Mapping[str, str] | None = None
    ) -> ToolResult:
        """Keyless web search via the DuckDuckGo HTML endpoint (dogfood F11).

        The endpoint is read from ``WEB_SEARCH_ENDPOINT`` at call time (project
        convention: env read at call time, not import time) so tests can point
        it at a local fixture server; it defaults to the real DDG HTML backend.
        An optional ``env`` mapping lets callers override the environment without
        mutating ``os.environ``.
        """
        query = (args.get("query") or "").strip()
        if not query:
            return ToolResult("web_search", False, "", "web_search requires 'query'.")
        try:
            max_results = int(args.get("max_results", 5))
        except TypeError, ValueError:
            max_results = 5
        max_results = max(1, min(10, max_results))

        env = os.environ if env is None else env
        endpoint = env.get("WEB_SEARCH_ENDPOINT", "https://html.duckduckgo.com/html/")
        try:
            async with httpx.AsyncClient(
                follow_redirects=True, timeout=_WEB_TIMEOUT_SECONDS
            ) as client:
                resp = await client.get(
                    endpoint,
                    params={"q": query},
                    headers={"User-Agent": _DESKTOP_USER_AGENT},
                )
                resp.raise_for_status()
        except httpx.HTTPError as e:
            return ToolResult(
                "web_search", False, "", f"web_search request failed: {e}"
            )

        parser = _DDGResultParser()
        parser.feed(resp.text)
        results = parser.results[:max_results]
        if not results:
            return ToolResult("web_search", True, "No results found.")

        lines = []
        for i, r in enumerate(results, start=1):
            title = " ".join(r["title"].split())
            snippet = " ".join(r["snippet"].split())
            lines.append(f"{i}. {title}\n   {r['url']}\n   {snippet}")
        return ToolResult("web_search", True, "\n".join(lines))

    async def _web_fetch(
        self, args: dict, env: Mapping[str, str] | None = None
    ) -> ToolResult:
        """Fetch a URL and return its readable text, SSRF-guarded (dogfood F11).

        Only http/https are accepted. The hostname is resolved and refused if
        any address is loopback/private/link-local/reserved, unless
        ``WEB_TOOLS_ALLOW_PRIVATE`` is truthy (read at call time — exists so
        tests, and deliberately local deployments, can target a local server).
        An optional ``env`` mapping lets callers override the environment without
        mutating ``os.environ``.
        """
        url = (args.get("url") or "").strip()
        if not url:
            return ToolResult("web_fetch", False, "", "web_fetch requires 'url'.")

        parsed = urllib.parse.urlsplit(url)
        scheme = parsed.scheme.lower()
        if scheme not in ("http", "https"):
            return ToolResult(
                "web_fetch",
                False,
                "",
                f"Unsupported URL scheme: {scheme or '(none)'}",
            )
        hostname = parsed.hostname
        if not hostname:
            return ToolResult("web_fetch", False, "", "URL has no hostname.")

        if not _web_tools_allow_private(env):
            port = parsed.port or (443 if scheme == "https" else 80)
            try:
                infos = socket.getaddrinfo(hostname, port)
            except OSError as e:
                return ToolResult(
                    "web_fetch", False, "", f"Could not resolve host: {e}"
                )
            for info in infos:
                ip_str = str(info[4][0])
                if _is_blocked_address(ip_str):
                    return ToolResult(
                        "web_fetch",
                        False,
                        "",
                        f"Refused: {hostname} resolves to a disallowed address "
                        f"({ip_str}). Set WEB_TOOLS_ALLOW_PRIVATE to override.",
                    )

        content_type = ""
        try:
            async with httpx.AsyncClient(
                follow_redirects=True, timeout=_WEB_TIMEOUT_SECONDS
            ) as client:
                async with client.stream(
                    "GET", url, headers={"User-Agent": _DESKTOP_USER_AGENT}
                ) as resp:
                    resp.raise_for_status()
                    content_type = (
                        resp.headers.get("content-type", "")
                        .split(";")[0]
                        .strip()
                        .lower()
                    )
                    if not (
                        content_type.startswith("text/")
                        or content_type == "application/json"
                    ):
                        return ToolResult(
                            "web_fetch",
                            False,
                            "",
                            f"Unsupported content type: {content_type or '(unknown)'}",
                        )
                    body = bytearray()
                    async for chunk in resp.aiter_bytes():
                        body += chunk
                        if len(body) >= _MAX_FETCH_BYTES:
                            break
        except httpx.HTTPError as e:
            return ToolResult("web_fetch", False, "", f"web_fetch request failed: {e}")

        text = bytes(body[:_MAX_FETCH_BYTES]).decode("utf-8", errors="replace")
        if content_type == "text/html":
            text = _extract_html_text(text)
        return ToolResult("web_fetch", True, text[:_MAX_FETCH_CHARS])

    async def _design_patterns(self, args: dict) -> ToolResult:
        """Look up Azure/cloud design patterns, domain guides, and antipatterns.

        Backed by ``configs/cloud-design-patterns.yaml`` (loaded once, lazily,
        and cached at module scope — resolved relative to this file so it
        works regardless of the executor's workspace). Matches ``query``
        case-insensitively against name/problem/solution/use_when/summary
        fields, with an optional exact ``category`` filter. Pattern matches
        are capped at 8; matching domain guides and antipatterns are always
        appended on top of the cap.
        """
        query = (args.get("query") or "").strip().lower()
        category = (args.get("category") or "").strip().lower()

        catalog = _load_design_patterns_catalog()
        matched_patterns = [
            p
            for p in catalog["patterns"]
            if _design_patterns_entry_matches(p, query, category)
        ][:_DESIGN_PATTERNS_MATCH_CAP]
        matched_guides = [
            g
            for g in catalog["domain_guides"]
            if _design_patterns_entry_matches(g, query, category)
        ]
        matched_antipatterns = [
            a
            for a in catalog["antipatterns"]
            if _design_patterns_entry_matches(a, query, category)
        ]

        if not matched_patterns and not matched_guides and not matched_antipatterns:
            categories = sorted(
                {
                    str(p.get("category"))
                    for p in (*catalog["patterns"], *catalog["domain_guides"])
                    if p.get("category")
                }
            )
            return ToolResult(
                "design_patterns",
                True,
                "No matching patterns. Categories: " + ", ".join(categories),
            )

        blocks = (
            [_format_design_pattern_block(p) for p in matched_patterns]
            + [_format_domain_guide_block(g) for g in matched_guides]
            + [_format_antipattern_block(a) for a in matched_antipatterns]
        )
        return ToolResult("design_patterns", True, "\n\n".join(blocks))


def get_agent_tools(agent_alias: str) -> list[dict[str, Any]]:
    """Return the tool definitions available to a specific agent.

    Combines the legacy ``ALL_TOOLS`` function-schemas with OpenAI-style
    schemas for any SPEC-062 native tools (tool wave: repl/worktree/lsp/
    task/remote_trigger/schedule_cron/mcp-resource) allowed for this alias in
    ``AGENT_TOOLS``, built from each native tool's ``ToolMetadata`` via
    :meth:`~orchestrata_ultimate.tools.registry.ToolRegistry.to_function_schemas`.
    """
    allowed = AGENT_TOOLS.get(agent_alias, [])
    legacy = [t for t in ALL_TOOLS if t["function"]["name"] in allowed]
    native_names = [n for n in allowed if n not in _LEGACY_TOOL_NAMES]
    if not native_names:
        return legacy
    return legacy + _native_schema_registry().to_function_schemas(native_names)


# ── Bridge to the unified tool registry ──────────────────────────────
# Maps each legacy function-tool to its SPEC-062 category so the live tools can
# be exposed through the unified ``ToolRegistry`` (discoverable/searchable by
# SPEC-072) without changing the invocation path above.
_LEGACY_TOOL_CATEGORIES: dict[str, ToolCategory] = {
    "read_file": ToolCategory.FILE,
    "write_file": ToolCategory.FILE,
    "list_dir": ToolCategory.FILE,
    "search_code": ToolCategory.SEARCH,
    "run_shell": ToolCategory.SHELL,
    "ast_query": ToolCategory.SEARCH,
    "check_coverage": ToolCategory.UTILITY,
    "validate_mermaid": ToolCategory.UTILITY,
    "check_doc_links": ToolCategory.UTILITY,
    "message_volk": ToolCategory.AGENT,
    "publish_topic": ToolCategory.AGENT,
    "subscribe_topic": ToolCategory.AGENT,
    "unsubscribe_topic": ToolCategory.AGENT,
    "web_search": ToolCategory.WEB,
    "web_fetch": ToolCategory.WEB,
    "design_patterns": ToolCategory.SEARCH,
}

# Legacy tools that cannot be CONFINED, and so need a human before they run.
#
# `requires_approval` means one thing: this tool can reach beyond the
# workspace, so path checking alone cannot make it safe.
#
# `run_shell` qualifies -- it takes a caller-supplied `cwd`
# (`args.get("cwd", str(self._workspace))`) and is a shell regardless of where
# it starts, so the workspace is not a boundary for it.
#
# `write_file` deliberately does NOT qualify, though it did until 2026-08-09:
# `_safe_path` resolves the target and rejects anything not under the
# workspace, so its confinement is structural, not advisory. Once the flag
# became load-bearing (ToolRegistry.execute now REFUSES flagged tools rather
# than merely labelling them), keeping write_file in this set would have taken
# file writes away from U:Cd, U:Qa, J:Junior and U:Do -- the entire build
# pipeline -- to guard a path that already cannot escape.
_LEGACY_APPROVAL_TOOLS: frozenset[str] = frozenset({"run_shell"})

# Native SPEC-062 tools that run arbitrary code but don't declare
# ``requires_approval`` on their own (frozen) ``ToolMetadata`` — flagged here so
# they flow through the identical approval-surfacing mechanism as
# ``_LEGACY_APPROVAL_TOOLS`` (``remote_trigger`` already declares it itself).
_NATIVE_APPROVAL_OVERRIDES: frozenset[str] = frozenset({"repl"})


class _ApprovalOverrideTool(BaseTool):
    """Wrap a native ``BaseTool``, forcing ``requires_approval=True``.

    Composes the wrapped tool unchanged (``execute`` delegates straight
    through) and only overrides the approval flag on its metadata, so a tool
    like ``repl`` — which executes arbitrary code but doesn't mark itself
    approval-worthy — is gated the same way ``write_file``/``run_shell`` are,
    rather than silently bypassing that gate.
    """

    def __init__(self, tool: BaseTool) -> None:
        self._tool = tool
        self._metadata = dataclasses.replace(tool.metadata, requires_approval=True)

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        return await self._tool.execute(arguments)


@traces(spec="072", story="US-072-001", reqs=["REQ-072-011"])
def build_tool_registry(executor: ToolExecutor) -> ToolRegistry:
    """Expose the legacy function-tools (``ALL_TOOLS``) via a unified registry.

    Wraps each of the pre-existing tools as a
    :class:`~orchestrata_ultimate.tools.base.FunctionToolAdapter` backed by
    ``executor``, so the 8 production tools become part of the metadata-indexed
    registry (SPEC-062) that SPEC-072 ToolSearch can search — while the live
    ``get_agent_tools``/``ToolExecutor`` invocation path is left untouched.
    """
    registry = ToolRegistry()
    for schema in ALL_TOOLS:
        fn = schema["function"]
        name = fn["name"]
        metadata = ToolMetadata(
            name=name,
            description=fn.get("description", ""),
            category=_LEGACY_TOOL_CATEGORIES.get(name, ToolCategory.UTILITY),
            parameters=fn.get("parameters", {}),
            requires_approval=name in _LEGACY_APPROVAL_TOOLS,
        )
        registry.register(FunctionToolAdapter(metadata, executor.execute))
    register_native_tools(registry)
    return registry


def register_native_tools(
    registry: ToolRegistry,
    *,
    codebase_kb: Any | None = None,
    codebase_kb_environ: dict[str, str] | None = None,
) -> None:
    """Register the native ``BaseTool`` implementations from the tool wave.

    These are first-class SPEC-062 tools (not ``ALL_TOOLS`` function-schema
    wrappers): SPEC-063 REPL, SPEC-064 worktree lifecycle, SPEC-065 LSP,
    SPEC-067 remote trigger, SPEC-068 cron scheduling, SPEC-069 task
    management, SPEC-071 MCP resources, SPEC-072 tool search, SPEC-107
    on-demand context tools, and the janus-backed Prolog codebase-
    intelligence query tools (``codebase_query_tools.py`` — impact /
    module deps / find cycles / dep path). Volk-facing exposure through
    ``AGENT_TOOLS`` is a separate, governance-gated decision — registration
    here makes them discoverable and dispatchable via the unified registry
    (and searchable via SPEC-072 ToolSearch).

    ``codebase_kb``/``codebase_kb_environ`` are test-injection seams passed
    straight through to :func:`~orchestrata_ultimate.tools.codebase_query_tools.register_codebase_query_tools`
    (typed ``Any`` here to avoid importing ``CodebaseTools`` at module scope
    for a param real production callers never pass); both default to
    ``None``, so real callers get the unchanged production lifecycle
    (:func:`~orchestrata_ultimate.tools.codebase_query_tools._build_production_kb`
    resolving real env + the real project root on first query).
    """
    from orchestrata_ultimate.tools.codebase_query_tools import (
        register_codebase_query_tools,
    )
    from orchestrata_ultimate.tools.context_tools import register_context_tools
    from orchestrata_ultimate.tools.lsp_tool import LSPTool
    from orchestrata_ultimate.tools.mcp_resource_tools import (
        ListMcpResourcesTool,
        ReadMcpResourceTool,
    )
    from orchestrata_ultimate.tools.remote_trigger_tool import (
        register_remote_trigger_tool,
    )
    from orchestrata_ultimate.tools.repl_tool import REPLTool
    from orchestrata_ultimate.tools.schedule_cron_tool import (
        register_schedule_cron_tool,
    )
    from orchestrata_ultimate.tools.task_tools import register_task_tools
    from orchestrata_ultimate.tools.worktree_tools import (
        EnterWorktreeTool,
        ExitWorktreeTool,
    )

    registry.register(REPLTool(), replace=True)
    registry.register(EnterWorktreeTool(), replace=True)
    registry.register(ExitWorktreeTool(), replace=True)
    registry.register(LSPTool(), replace=True)
    registry.register(ListMcpResourcesTool(), replace=True)
    registry.register(ReadMcpResourceTool(), replace=True)
    register_task_tools(registry)
    register_remote_trigger_tool(registry)
    register_schedule_cron_tool(registry)
    register_context_tools(registry)
    register_codebase_query_tools(registry, kb=codebase_kb, environ=codebase_kb_environ)

    # Apply the approval overrides as a final pass over everything registered
    # above (including tools added via the register_* helpers), so membership
    # in _NATIVE_APPROVAL_OVERRIDES is the single mechanism — not a hardcoded
    # wrap on individual constructor calls.
    for name in _NATIVE_APPROVAL_OVERRIDES:
        tool = registry.get(name)
        if tool is not None and not tool.metadata.requires_approval:
            registry.register(_ApprovalOverrideTool(tool), replace=True)

    # SPEC-072: the search tool over this registry, registered LAST so its
    # rebuild-on-startup corpus covers everything above (in the
    # build_tool_registry context that includes the legacy ALL_TOOLS wrappers,
    # which are registered before this function runs).
    from orchestrata_ultimate.tools.search import ToolSearchTool

    registry.register(ToolSearchTool.create(registry), replace=True)


# Lazily built, process-cached registry of native tools' metadata (schema
# exposure only — never dispatched against, so it's fine to share process-wide
# unlike the per-``ToolExecutor`` dispatch registry in ``_native_tools()``).
_NATIVE_SCHEMA_REGISTRY: ToolRegistry | None = None
_NATIVE_SCHEMA_REGISTRY_LOCK = threading.Lock()


def _native_schema_registry() -> ToolRegistry:
    """The process-wide registry of native tools' ``ToolMetadata``, for schemas.

    ``get_agent_tools`` only needs each native tool's name/description/
    parameters to render an OpenAI-style schema — not a live, request-bound
    instance — so this is built once (lazily, thread-safe) via
    :func:`register_native_tools` and cached, distinct from each
    :class:`ToolExecutor`'s own per-instance dispatch registry.
    """
    global _NATIVE_SCHEMA_REGISTRY
    if _NATIVE_SCHEMA_REGISTRY is not None:
        return _NATIVE_SCHEMA_REGISTRY
    with _NATIVE_SCHEMA_REGISTRY_LOCK:
        if _NATIVE_SCHEMA_REGISTRY is None:
            registry = ToolRegistry()
            register_native_tools(registry)
            _NATIVE_SCHEMA_REGISTRY = registry
        return _NATIVE_SCHEMA_REGISTRY