orchestrata_ultimate.tools.search

Reference › orchestrata_ultimate › tools › search

ToolSearchTool (SPEC-072) searches all registered tools by semantic similarity, keywords and fuzzy matching, and is itself a BaseTool, so volks find tools the same way they use them. It was the first tool built by the build-factory pipeline from a contract skeleton. agents.tool_registry imports it and tests/tools/test_search.py covers it. The corpus comes from registry and the contract from base. The narrative is in the tools package page.

The design is composition. ToolSearchTool holds a ToolSearchEngine, which composes:

• ToolIndex, built from the registry as ToolIndexEntry records.
• QueryValidator and QueryParser, which validate and sanitise input and parse AND, OR and wildcard expressions into a ParsedQuery.
• Three MatchStrategy implementations: SemanticMatchStrategy (embedding similarity), KeywordMatchStrategy (TF-IDF) and FuzzyMatchStrategy (difflib).
• RelevanceRanker, which blends the scores with the documented weights: 0.40 semantic, 0.25 keyword or fuzzy, 0.15 popularity, 0.10 success rate and 0.10 recency.
• SearchCache, a result cache with time and size limits.
• An injected Tracer for runtime tracing.

Request and response shapes are frozen dataclasses: SearchFilters, SearchOptions, SearchQuery, MatchScore, SearchResultItem, FacetCount and SearchResponse. Failures use ToolSearchError with a SearchErrorCode and are returned as failed ToolResult values and not raised. When embeddings are unavailable, the semantic strategy contributes nothing and the other strategies carry the match.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

SearchErrorCode

Decorators: traces(spec='072', story='US-072-003', reqs=['REQ-072-101', 'REQ-072-102', 'REQ-072-103', 'REQ-072-104', 'REQ-072-105',… · Line 138

Error codes for the ToolSearchTool pipeline (spec.md Error Conditions).

ToolSearchError

Bases: Exception · Decorators: traces(spec='072', story='US-072-003', reqs=['REQ-072-101', 'REQ-072-111']) · Line 161

Raised by the search pipeline on any spec.md ERR-00N condition.

SearchFilters (dataclass)

Decorators: traces(spec='072', story='US-072-003', reqs=['REQ-072-003']), dataclass(slots=True, frozen=True) · Line 191

Filter constraints for a search (spec.md SearchRequest.filters).

SearchOptions (dataclass)

Decorators: traces(spec='072', story='US-072-003', reqs=['REQ-072-004', 'REQ-072-002']), dataclass(slots=True, frozen=True) · Line 203

Search behaviour options (spec.md SearchRequest.options).

SearchQuery (dataclass)

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-001']), dataclass(slots=True, frozen=True) · Line 215

A validated-shape search request (spec.md ToolSearchRequest).

ToolIndexEntry (dataclass)

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-011']), dataclass(slots=True, frozen=True) · Line 269

A single indexed tool (spec.md 4.1 Tool Index Entry).

MatchScore (dataclass)

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-005']), dataclass(slots=True, frozen=True) · Line 321

A single strategy's (or the combined) match quality for one entry.

SearchResultItem (dataclass)

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-005']), dataclass(slots=True, frozen=True) · Line 331

One ranked, final search result (spec.md SearchResponse.results[]).

FacetCount (dataclass)

Decorators: traces(spec='072', story='US-072-002', reqs=['REQ-072-007']), dataclass(slots=True, frozen=True) · Line 342

A single facet value + its count (spec.md SearchFacets).

SearchResponse (dataclass)

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-004', 'REQ-072-005', 'REQ-072-006', 'REQ-072-007']), dataclass(slots=True) · Line 355

The full search response (spec.md ToolSearchResponse).

ParsedQuery (dataclass)

Decorators: traces(spec='072', story='US-072-004', reqs=['REQ-072-010', 'REQ-072-024']), dataclass(slots=True, frozen=True) · Line 402

A tokenized query: OR-of-AND groups + any wildcards seen.

QueryParser

Decorators: traces(spec='072', story='US-072-004', reqs=['REQ-072-010', 'REQ-072-024']) · Line 411

Parses a sanitized query string into OR-of-AND term groups.

QueryValidator

Decorators: traces(spec='072', story='US-072-003', reqs=['REQ-072-003', 'REQ-072-004', 'REQ-072-101', 'REQ-072-102', 'REQ-072-103',… · Line 476

Validates a SearchQuery and sanitizes its raw query text.

MatchStrategy (protocol)

Bases: Protocol · Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-002', 'REQ-072-008', 'REQ-072-009']), runtime_checkable · Line 586

Structural interface every scoring strategy implements.

SemanticMatchStrategy

Bases: MatchStrategy · Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-008', 'REQ-072-108']) · Line 645

Embedding-based cosine-similarity match (REQ-072-008).

KeywordMatchStrategy

Bases: MatchStrategy · Decorators: traces(spec='072', story='US-072-002', reqs=['REQ-072-009', 'REQ-072-024']) · Line 681

TF-IDF weighted exact/prefix keyword match (REQ-072-009/REQ-072-024).

FuzzyMatchStrategy

Bases: MatchStrategy · Decorators: traces(spec='072', story='US-072-004', reqs=['REQ-072-002']) · Line 767

Typo-tolerant match via string-similarity ratio (REQ-072-002).

RelevanceRanker

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-005']) · Line 832

Blends match quality with popularity/success/recency (spec.md formula).

ToolIndex

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-011', 'REQ-072-012', 'REQ-072-013', 'REQ-072-014', 'REQ-072-015',… · Line 929

The in-memory, thread-safe searchable corpus (spec.md Index Management).

SearchCache

Decorators: traces(spec='072', story='US-072-005', reqs=['REQ-072-016', 'REQ-072-017', 'REQ-072-018', 'REQ-072-019', 'REQ-072-020']) · Line 1085

TTL/LRU cache of search results (spec.md Caching Strategy).

ToolSearchEngine

Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-001', 'REQ-072-003', 'REQ-072-004', 'REQ-072-006', 'REQ-072-007',… · Line 1210

The search pipeline: validate -> cache -> filter -> score -> rank -> page.

ToolSearchTool

Bases: BaseTool · Decorators: traces(spec='072', story='US-072-001', reqs=['REQ-072-001', 'REQ-072-013', 'REQ-072-019']) · Line 1480

The registrable BaseTool (Command) callers/LLMs actually invoke.

Functions

Source

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

"""ToolSearchTool — semantic/keyword/fuzzy search across all registered tools.

Spec: SPEC-072 (ToolSearchTool). The first tool built by the build factory
pipeline (see ``plans/build-factory-architecture.md``): a Mermaid contract
(``plans/contracts/SPEC-072.reqmapped.mmd`` / ``.contract.json``) was turned
into a typed skeleton via ``gen-skeleton`` and this module fills its leaves.

Architecture (composition over inheritance, one ABC/Protocol boundary):
    - :class:`ToolSearchTool` is the single :class:`~orchestrata_ultimate.tools.base.BaseTool`
      (Command) every caller sees. It composes a :class:`ToolSearchEngine`.
    - :class:`ToolSearchEngine` composes the pipeline: :class:`ToolIndex`
      (the searchable corpus, built from :class:`~orchestrata_ultimate.tools.registry.ToolRegistry`),
      :class:`QueryValidator` + :class:`QueryParser` (input handling),
      a list of :class:`MatchStrategy` implementations (Strategy —
      :class:`SemanticMatchStrategy` / :class:`KeywordMatchStrategy` /
      :class:`FuzzyMatchStrategy`), :class:`RelevanceRanker` (the weighted
      blend), and :class:`SearchCache` (a TTL/LRU result cache).
    - Runtime tracing is injected (:class:`~orchestrata_ultimate.core.tracing.tracer.Tracer`,
      REUSED from ``core.tracing.tracer`` — not recreated here per the build
      factory's cross-cutting-class rule).

Ranking (spec.md "Search Ranking Algorithm"):
    ``final = 0.40*semantic + 0.25*keyword(or fuzzy) + 0.15*popularity +
    0.10*success_rate + 0.10*recency``. The semantic/keyword sub-weights are
    folded into the combined :class:`MatchScore` handed to
    :class:`RelevanceRanker` (see :meth:`ToolSearchEngine._combine`);
    :class:`RelevanceRanker` owns all five weight constants and applies the
    remaining three (popularity/success_rate/recency) from
    :class:`ToolIndexEntry` fields.
"""

from __future__ import annotations

import difflib
import hashlib
import json
import math
import re
import threading
import time
from collections import Counter
from collections.abc import Callable
from dataclasses import asdict, dataclass, field, replace
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, ClassVar, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.lru_cache import LRUCache
from orchestrata_ultimate.core.tracing.tracer import TraceConfig, Tracer
from orchestrata_ultimate.memory.embedding import EmbeddingService
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolResult,
)
from orchestrata_ultimate.tools.registry import ToolRegistry

#: Soft wall-clock budget for a single ``ToolSearchEngine.search`` call
#: (REQ-072-021/REQ-072-112). Exceeding it while entries remain unscored
#: degrades to partial results (or raises TIMEOUT if nothing scored at all)
#: rather than blocking indefinitely — mutate this module attribute in tests
#: to exercise the ERR-012 path deterministically.
_SEARCH_TIMEOUT_SECONDS = 5.0

#: Minimum combined match score (post 0.40/0.25-weighting, pre-ranking) for an
#: entry to be considered a real match at all. The hash-bucket embedding used
#: by :class:`~orchestrata_ultimate.memory.embedding.EmbeddingService` can
#: produce small nonzero cosine similarity between entirely unrelated terms
#: on random bucket collisions (``hash()`` is process-salted) — this floor
#: keeps such noise out of "relevant tools" results (FR-001) rather than
#: surfacing a technically-nonzero-but-meaningless match.
_MIN_RELEVANCE_SCORE = 0.05

_VALID_RANK_BY = frozenset({"relevance", "popularity", "recency", "success_rate"})
_VALID_RETURN_TYPES = frozenset(
    {"string", "number", "boolean", "object", "array", "binary", "stream", "void"}
)
_ALLOWED_QUERY_CHARS = re.compile(r"[^A-Za-z0-9_\-\s*]")

_SEARCH_TOOL_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "description": "Natural-language, keyword, or AND/OR/wildcard search query.",
        },
        "filters": {
            "type": "object",
            "properties": {
                "categories": {"type": "array", "items": {"type": "string"}},
                "tags": {"type": "array", "items": {"type": "string"}},
                "return_types": {"type": "array", "items": {"type": "string"}},
                "authors": {"type": "array", "items": {"type": "string"}},
                "min_success_rate": {"type": "number"},
            },
        },
        "options": {
            "type": "object",
            "properties": {
                "limit": {"type": "integer"},
                "offset": {"type": "integer"},
                "fuzzy_threshold": {"type": "number"},
                "include_disabled": {"type": "boolean"},
                "rank_by": {
                    "type": "string",
                    "enum": sorted(_VALID_RANK_BY),
                },
            },
        },
    },
    "required": ["query"],
}


# ── errors ─────────────────────────────────────────────────────────────


@traces(
    spec="072",
    story="US-072-003",
    reqs=[
        "REQ-072-101",
        "REQ-072-102",
        "REQ-072-103",
        "REQ-072-104",
        "REQ-072-105",
        "REQ-072-106",
        "REQ-072-107",
        "REQ-072-108",
        "REQ-072-109",
        "REQ-072-110",
        "REQ-072-111",
        "REQ-072-112",
    ],
)
class SearchErrorCode:
    """Error codes for the ToolSearchTool pipeline (spec.md Error Conditions).

    A plain namespace of string constants (rather than ``StrEnum``) so
    ``ToolSearchError`` stays a trivial, easily-constructed value in tests —
    the *codes* are what spec.md traces (ERR-001..ERR-012), not an enum type.
    """

    INVALID_QUERY = "invalid_query"
    INVALID_FILTER = "invalid_filter"
    INVALID_PAGINATION = "invalid_pagination"
    INDEX_NOT_READY = "index_not_ready"
    INDEX_CORRUPTED = "index_corrupted"
    TOOL_NOT_FOUND = "tool_not_found"
    CACHE_ERROR = "cache_error"
    EMBEDDING_SERVICE_ERROR = "embedding_service_error"
    RATE_LIMIT_EXCEEDED = "rate_limit_exceeded"
    PERMISSION_DENIED = "permission_denied"
    INVALID_SORT_OPTION = "invalid_sort_option"
    TIMEOUT = "timeout"


@traces(spec="072", story="US-072-003", reqs=["REQ-072-101", "REQ-072-111"])
class ToolSearchError(Exception):
    """Raised by the search pipeline on any spec.md ERR-00N condition.

    Carries a :data:`SearchErrorCode` + message and knows how to render
    itself as a failed :class:`~orchestrata_ultimate.tools.base.ToolResult`
    (:meth:`to_result`) so :class:`ToolSearchTool.execute` never has to
    translate error codes itself.
    """

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message

    @traces(spec="072", story="US-072-003", reqs=["REQ-072-101", "REQ-072-111"])
    def to_result(self, tool_name: str) -> ToolResult:
        """Render this error as a failed ``ToolResult`` for ``tool_name``."""
        return ToolResult(
            tool_name=tool_name,
            success=False,
            output="",
            error=f"{self.code}: {self.message}",
        )


# ── request / options dataclasses ───────────────────────────────────────


@traces(spec="072", story="US-072-003", reqs=["REQ-072-003"])
@dataclass(slots=True, frozen=True)
class SearchFilters:
    """Filter constraints for a search (spec.md SearchRequest.filters)."""

    categories: tuple[str, ...] = ()
    tags: tuple[str, ...] = ()
    return_types: tuple[str, ...] = ()
    authors: tuple[str, ...] = ()
    min_success_rate: float = 0.0


@traces(spec="072", story="US-072-003", reqs=["REQ-072-004", "REQ-072-002"])
@dataclass(slots=True, frozen=True)
class SearchOptions:
    """Search behaviour options (spec.md SearchRequest.options)."""

    limit: int = 10
    offset: int = 0
    fuzzy_threshold: float = 0.7
    include_disabled: bool = False
    rank_by: str = "relevance"


@traces(spec="072", story="US-072-001", reqs=["REQ-072-001"])
@dataclass(slots=True, frozen=True)
class SearchQuery:
    """A validated-shape search request (spec.md ToolSearchRequest)."""

    query: str
    filters: SearchFilters = field(default_factory=SearchFilters)
    options: SearchOptions = field(default_factory=SearchOptions)

    @traces(
        spec="072",
        story="US-072-001",
        reqs=["REQ-072-001", "REQ-072-101", "REQ-072-102"],
    )
    @classmethod
    def from_arguments(cls, arguments: dict[str, Any]) -> SearchQuery:
        """Build a ``SearchQuery`` from a tool-call ``arguments`` mapping.

        Raises:
            ToolSearchError: ``INVALID_QUERY`` when ``arguments['query']`` is
                missing, blank, or not a string (ERR-001).
        """
        raw_query = arguments.get("query")
        if not isinstance(raw_query, str) or not raw_query.strip():
            raise ToolSearchError(
                SearchErrorCode.INVALID_QUERY, "arguments['query'] is required"
            )
        filters_raw = arguments.get("filters") or {}
        options_raw = arguments.get("options") or {}
        try:
            filters = SearchFilters(
                categories=tuple(filters_raw.get("categories") or ()),
                tags=tuple(filters_raw.get("tags") or ()),
                return_types=tuple(filters_raw.get("return_types") or ()),
                authors=tuple(filters_raw.get("authors") or ()),
                min_success_rate=float(filters_raw.get("min_success_rate") or 0.0),
            )
            options = SearchOptions(
                limit=int(options_raw.get("limit", 10)),
                offset=int(options_raw.get("offset", 0)),
                fuzzy_threshold=float(options_raw.get("fuzzy_threshold", 0.7)),
                include_disabled=bool(options_raw.get("include_disabled", False)),
                rank_by=str(options_raw.get("rank_by", "relevance")),
            )
        except (TypeError, ValueError) as exc:
            raise ToolSearchError(
                SearchErrorCode.INVALID_FILTER, f"malformed filters/options: {exc}"
            ) from exc
        return cls(query=raw_query, filters=filters, options=options)


# ── index entry + match/result dataclasses ──────────────────────────────


@traces(spec="072", story="US-072-001", reqs=["REQ-072-011"])
@dataclass(slots=True, frozen=True)
class ToolIndexEntry:
    """A single indexed tool (spec.md 4.1 Tool Index Entry)."""

    tool_id: str
    name: str
    description: str
    category: str
    tags: tuple[str, ...] = ()
    parameters: dict[str, Any] = field(default_factory=dict)
    return_type: str = "string"
    version: str = "1.0.0"
    author: str = ""
    usage_count: int = 0
    success_rate: float = 1.0
    updated_at: str = ""
    embedding: tuple[float, ...] = ()

    @traces(
        spec="072",
        story="US-072-001",
        reqs=["REQ-072-011", "REQ-072-012"],
    )
    @classmethod
    def from_metadata(cls, meta: ToolMetadata) -> ToolIndexEntry:
        """Derive an index entry from a registered tool's metadata.

        ``ToolMetadata`` carries no ``author``/``usage_count``/``success_rate``
        telemetry (those are operational fields spec.md expects a live
        deployment to track over time) — they default to neutral values
        (empty author, zero usage, a healthy 1.0 success rate, "now" as the
        update time) so a freshly-registered tool ranks neither favoured nor
        penalised until real telemetry arrives via :meth:`ToolIndex.sync`.
        """
        return cls(
            tool_id=meta.name,
            name=meta.name,
            description=meta.description,
            category=str(meta.category),
            tags=tuple(meta.tags),
            parameters=dict(meta.parameters),
            return_type="string",
            version=meta.version,
            author="",
            usage_count=0,
            success_rate=1.0,
            updated_at=datetime.now(UTC).isoformat(),
            embedding=(),
        )


@traces(spec="072", story="US-072-001", reqs=["REQ-072-005"])
@dataclass(slots=True, frozen=True)
class MatchScore:
    """A single strategy's (or the combined) match quality for one entry."""

    score: float = 0.0
    matched_fields: tuple[str, ...] = ()
    fuzzy_applied: bool = False


@traces(spec="072", story="US-072-001", reqs=["REQ-072-005"])
@dataclass(slots=True, frozen=True)
class SearchResultItem:
    """One ranked, final search result (spec.md SearchResponse.results[])."""

    entry: ToolIndexEntry
    score: float
    matched_fields: tuple[str, ...] = ()
    highlight: str = ""


@traces(spec="072", story="US-072-002", reqs=["REQ-072-007"])
@dataclass(slots=True, frozen=True)
class FacetCount:
    """A single facet value + its count (spec.md SearchFacets)."""

    value: str
    count: int


@traces(
    spec="072",
    story="US-072-001",
    reqs=["REQ-072-004", "REQ-072-005", "REQ-072-006", "REQ-072-007"],
)
@dataclass(slots=True)
class SearchResponse:
    """The full search response (spec.md ToolSearchResponse)."""

    results: list[SearchResultItem]
    total_count: int
    query: str
    limit: int
    offset: int
    has_more: bool
    suggestions: list[str] = field(default_factory=list)
    facets: dict[str, list[FacetCount]] = field(default_factory=dict)

    @traces(
        spec="072",
        story="US-072-001",
        reqs=["REQ-072-004", "REQ-072-005", "REQ-072-006", "REQ-072-007"],
    )
    def to_json(self) -> str:
        """Render as the spec.md ``ToolSearchResponse`` JSON shape."""
        payload = {
            "results": [
                {
                    "tool": asdict(item.entry),
                    "score": item.score,
                    "matched_fields": list(item.matched_fields),
                    "highlight": item.highlight,
                }
                for item in self.results
            ],
            "total_count": self.total_count,
            "query": self.query,
            "pagination": {
                "limit": self.limit,
                "offset": self.offset,
                "has_more": self.has_more,
            },
            "suggestions": list(self.suggestions),
            "facets": {k: [asdict(fc) for fc in v] for k, v in self.facets.items()},
        }
        return json.dumps(payload)


# ── query parsing / validation ───────────────────────────────────────────


@traces(spec="072", story="US-072-004", reqs=["REQ-072-010", "REQ-072-024"])
@dataclass(slots=True, frozen=True)
class ParsedQuery:
    """A tokenized query: OR-of-AND ``groups`` + any ``wildcards`` seen."""

    raw: str
    groups: tuple[tuple[str, ...], ...] = ()
    wildcards: tuple[str, ...] = ()


@traces(spec="072", story="US-072-004", reqs=["REQ-072-010", "REQ-072-024"])
class QueryParser:
    """Parses a sanitized query string into OR-of-AND term groups.

    ``"file AND write OR shell"`` -> ``groups=(("file", "write"), ("shell",))``
    (REQ-072-010); a term ending in ``*`` is a wildcard prefix (REQ-072-024).
    """

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(spec="072", story="US-072-004", reqs=["REQ-072-010", "REQ-072-024"])
    def parse(self, raw: str) -> ParsedQuery:
        """Split ``raw`` into OR-of-AND term groups.

        An explicit ``" AND "`` between two chunks produces one *strict*
        multi-term group (every term must match — REQ-072-010's compound
        semantics). A bare multi-word phrase with no explicit ``AND``/``OR``
        (the common natural-language case, e.g. ``"search the codebase"``)
        is instead split into independent single-word *soft* groups, each
        scored on its own and accumulated by :class:`KeywordMatchStrategy` —
        so a free-text query still surfaces partial matches instead of
        requiring every word to literally appear.
        """
        with self._tracer.span("QueryParser.parse"):
            text = raw.strip()
            or_parts = re.split(r"\s+OR\s+", text) if text else [""]
            groups: list[tuple[str, ...]] = []
            wildcards: list[str] = []
            for part in or_parts:
                part = part.strip()
                if not part:
                    continue
                and_chunks = [
                    c.strip() for c in re.split(r"\s+AND\s+", part) if c.strip()
                ]
                if len(and_chunks) > 1:
                    terms = tuple(w for chunk in and_chunks for w in chunk.split())
                    groups.append(terms)
                    wildcards.extend(t for t in terms if t.endswith("*"))
                else:
                    for word in part.split():
                        groups.append((word,))
                        if word.endswith("*"):
                            wildcards.append(word)
            return ParsedQuery(
                raw=text, groups=tuple(groups), wildcards=tuple(wildcards)
            )


@traces(
    spec="072",
    story="US-072-003",
    reqs=[
        "REQ-072-003",
        "REQ-072-004",
        "REQ-072-101",
        "REQ-072-102",
        "REQ-072-103",
        "REQ-072-111",
        "REQ-072-201",
        "REQ-072-202",
        "REQ-072-203",
        "REQ-072-204",
    ],
)
class QueryValidator:
    """Validates a ``SearchQuery`` and sanitizes its raw query text.

    All spec.md ERR-001/002/003/011 and SR-007..SR-010 input-validation
    requirements live here — the single gate ``ToolSearchEngine.search``
    calls before touching the index.
    """

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(
        spec="072",
        story="US-072-003",
        reqs=[
            "REQ-072-101",
            "REQ-072-102",
            "REQ-072-103",
            "REQ-072-111",
            "REQ-072-202",
            "REQ-072-204",
        ],
    )
    def validate(self, query: SearchQuery) -> None:
        """Raise :class:`ToolSearchError` on any malformed field.

        Raises:
            ToolSearchError: ``INVALID_QUERY`` (blank/too-long/wildcard-only
                query, REQ-072-101/REQ-072-202/SR-012), ``INVALID_PAGINATION``
                (limit/offset out of range, REQ-072-103), ``INVALID_FILTER``
                (malformed filter values, REQ-072-102/REQ-072-204), or
                ``INVALID_SORT_OPTION`` (unrecognized ``rank_by``, REQ-072-111).
        """
        with self._tracer.span("QueryValidator.validate"):
            text = query.query
            if not text or not text.strip():
                raise ToolSearchError(
                    SearchErrorCode.INVALID_QUERY, "query must not be blank"
                )
            if len(text) > 500:
                raise ToolSearchError(
                    SearchErrorCode.INVALID_QUERY,
                    "query exceeds the maximum length of 500 characters",
                )
            if text.strip() == "*":
                raise ToolSearchError(
                    SearchErrorCode.INVALID_QUERY,
                    "a bare wildcard query ('*') is not allowed (SR-012)",
                )

            opts = query.options
            if not (1 <= opts.limit <= 100):
                raise ToolSearchError(
                    SearchErrorCode.INVALID_PAGINATION,
                    "limit must be between 1 and 100",
                )
            if opts.offset < 0:
                raise ToolSearchError(
                    SearchErrorCode.INVALID_PAGINATION, "offset must be >= 0"
                )
            if not (0.0 <= opts.fuzzy_threshold <= 1.0):
                raise ToolSearchError(
                    SearchErrorCode.INVALID_FILTER,
                    "fuzzy_threshold must be within [0, 1]",
                )
            if opts.rank_by not in _VALID_RANK_BY:
                raise ToolSearchError(
                    SearchErrorCode.INVALID_SORT_OPTION,
                    f"invalid rank_by {opts.rank_by!r}; valid options: "
                    f"{sorted(_VALID_RANK_BY)}",
                )

            filters = query.filters
            if filters.return_types and any(
                rt not in _VALID_RETURN_TYPES for rt in filters.return_types
            ):
                raise ToolSearchError(
                    SearchErrorCode.INVALID_FILTER,
                    f"invalid return_type in filters; valid values: "
                    f"{sorted(_VALID_RETURN_TYPES)}",
                )
            if not (0.0 <= filters.min_success_rate <= 1.0):
                raise ToolSearchError(
                    SearchErrorCode.INVALID_FILTER,
                    "min_success_rate must be within [0, 1]",
                )

    @traces(spec="072", story="US-072-003", reqs=["REQ-072-201", "REQ-072-203"])
    def sanitize(self, text: str) -> str:
        """Strip anything outside ``[A-Za-z0-9_\\-\\s*]`` and collapse whitespace.

        SR-007 (injection prevention) / SR-009 (escape special characters):
        the search domain is a keyword/AND/OR/wildcard index, not a shell or
        SQL context, so sanitizing to a safe alphanumeric+wildcard charset is
        both sufficient and preserves query semantics (``AND``/``OR``/``*``).
        """
        with self._tracer.span("QueryValidator.sanitize"):
            cleaned = _ALLOWED_QUERY_CHARS.sub(" ", text)
            return re.sub(r"\s+", " ", cleaned).strip()


# ── match strategies (Strategy pattern) ──────────────────────────────────


@traces(
    spec="072",
    story="US-072-001",
    reqs=["REQ-072-002", "REQ-072-008", "REQ-072-009"],
)
@runtime_checkable
class MatchStrategy(Protocol):
    """Structural interface every scoring strategy implements."""

    @traces(
        spec="072",
        story="US-072-001",
        reqs=["REQ-072-002", "REQ-072-008", "REQ-072-009"],
    )
    def score(
        self, parsed: ParsedQuery, entry: ToolIndexEntry, options: SearchOptions
    ) -> MatchScore:
        """Score ``entry`` against ``parsed`` and return a :class:`MatchScore`."""
        ...


@traces(spec="072", story="US-072-002", reqs=["REQ-072-009", "REQ-072-024"])
@traces(spec="072", story="US-072-004", reqs=["REQ-072-002"])
def _tokenize(text: str) -> list[str]:
    """Lowercase word tokens (shared by keyword/fuzzy strategies)."""
    return re.findall(r"[a-z0-9]+", text.lower())


@traces(spec="072", story="US-072-002", reqs=["REQ-072-009", "REQ-072-024"])
def _entry_full_tokens(entry: ToolIndexEntry) -> list[str]:
    """All searchable tokens for ``entry`` (name + description + tags)."""
    text = " ".join(
        [
            entry.name.replace("_", " "),
            entry.description,
            " ".join(t.replace("_", " ") for t in entry.tags),
        ]
    )
    return _tokenize(text)


@traces(spec="072", story="US-072-002", reqs=["REQ-072-009", "REQ-072-024"])
@traces(spec="072", story="US-072-004", reqs=["REQ-072-002"])
def _entry_field_tokens(entry: ToolIndexEntry) -> dict[str, set[str]]:
    """Per-field token sets, used to attribute a match to name/description/tags."""
    return {
        "name": set(_tokenize(entry.name.replace("_", " "))),
        "description": set(_tokenize(entry.description)),
        "tags": set(_tokenize(" ".join(t.replace("_", " ") for t in entry.tags))),
    }


@traces(spec="072", story="US-072-002", reqs=["REQ-072-009", "REQ-072-024"])
def _term_matches(term: str, tokens: set[str]) -> bool:
    """Whether ``term`` (possibly a ``prefix*`` wildcard) appears in ``tokens``."""
    clean = term.lower()
    if clean.endswith("*"):
        prefix = clean[:-1]
        if not prefix:
            return False
        return any(tok.startswith(prefix) for tok in tokens)
    return clean in tokens


@traces(spec="072", story="US-072-001", reqs=["REQ-072-008", "REQ-072-108"])
class SemanticMatchStrategy(MatchStrategy):
    """Embedding-based cosine-similarity match (REQ-072-008).

    ``available()`` is ``False`` when composed without an embedding service
    (``embeddings=None``) — the composition-based ERR-008 fallback seam:
    :class:`ToolSearchEngine` skips this strategy's contribution entirely
    when it is unavailable, degrading gracefully to keyword search.
    """

    def __init__(
        self, embeddings: EmbeddingService | None, tracer: Tracer | None = None
    ) -> None:
        self._embeddings = embeddings
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-008", "REQ-072-108"])
    def available(self) -> bool:
        return self._embeddings is not None

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-008", "REQ-072-108"])
    def score(
        self, parsed: ParsedQuery, entry: ToolIndexEntry, options: SearchOptions
    ) -> MatchScore:
        with self._tracer.span("SemanticMatchStrategy.score"):
            if self._embeddings is None or not entry.embedding:
                return MatchScore()
            query_embedding = self._embeddings.compute_embedding(parsed.raw)
            similarity = self._embeddings.cosine_similarity(
                query_embedding, list(entry.embedding)
            )
            similarity = max(0.0, min(1.0, similarity))
            matched_fields = ("description",) if similarity > 0.0 else ()
            return MatchScore(score=similarity, matched_fields=matched_fields)


@traces(spec="072", story="US-072-002", reqs=["REQ-072-009", "REQ-072-024"])
class KeywordMatchStrategy(MatchStrategy):
    """TF-IDF weighted exact/prefix keyword match (REQ-072-009/REQ-072-024).

    The always-available fallback: no external dependency, so this strategy
    is what keeps search working when :class:`SemanticMatchStrategy` is
    unavailable (ERR-008 graceful degradation).
    """

    def __init__(self, idf: dict[str, float], tracer: Tracer | None = None) -> None:
        self._idf = idf
        self._doc_count = 0
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(spec="072", story="US-072-002", reqs=["REQ-072-009", "REQ-072-024"])
    def fit(self, entries: list[ToolIndexEntry]) -> None:
        """(Re)compute IDF weights over the current corpus."""
        with self._tracer.span("KeywordMatchStrategy.fit"):
            self._doc_count = len(entries)
            df: dict[str, int] = {}
            for entry in entries:
                for term in set(_entry_full_tokens(entry)):
                    df[term] = df.get(term, 0) + 1
            self._idf.clear()
            for term, count in df.items():
                self._idf[term] = math.log((self._doc_count + 1) / (count + 1)) + 1.0

    @traces(spec="072", story="US-072-002", reqs=["REQ-072-009", "REQ-072-024"])
    def _idf_of(self, term: str) -> float:
        return self._idf.get(term, math.log(self._doc_count + 1) + 1.0)

    @traces(
        spec="072",
        story="US-072-002",
        reqs=["REQ-072-009", "REQ-072-010", "REQ-072-024"],
    )
    def score(
        self, parsed: ParsedQuery, entry: ToolIndexEntry, options: SearchOptions
    ) -> MatchScore:
        """Sum TF-IDF weight across every fully-matching group.

        Each independent (soft, single-word) group contributes its own
        weight, so a free-text query rewards matching MORE words rather than
        only the single best one; a strict (explicit-AND) multi-term group
        contributes only when ALL of its terms are present (REQ-072-010).
        """
        with self._tracer.span("KeywordMatchStrategy.score"):
            full_tokens = _entry_full_tokens(entry)
            token_set = set(full_tokens)
            field_tokens = _entry_field_tokens(entry)
            total_weight = 0.0
            fields: set[str] = set()
            matched_any = False
            for group in parsed.groups:
                if not group or not all(_term_matches(t, token_set) for t in group):
                    continue
                matched_any = True
                group_weight = 0.0
                for term in group:
                    clean = term.lower().rstrip("*")
                    if term.lower().endswith("*") and clean:
                        matches = [tok for tok in full_tokens if tok.startswith(clean)]
                        tf = len(matches)
                        idf = max(
                            (self._idf_of(m) for m in set(matches)),
                            default=self._idf_of(clean),
                        )
                    else:
                        tf = full_tokens.count(clean)
                        idf = self._idf_of(clean)
                    group_weight += tf * idf
                    for field_name, toks in field_tokens.items():
                        hit = (
                            any(tok.startswith(clean) for tok in toks)
                            if term.lower().endswith("*")
                            else clean in toks
                        )
                        if hit:
                            fields.add(field_name)
                total_weight += group_weight / len(group)
            if not matched_any:
                return MatchScore()
            score = total_weight / (total_weight + 1.0)
            return MatchScore(score=score, matched_fields=tuple(sorted(fields)))


@traces(spec="072", story="US-072-004", reqs=["REQ-072-002"])
class FuzzyMatchStrategy(MatchStrategy):
    """Typo-tolerant match via string-similarity ratio (REQ-072-002).

    Each query term is compared against the entry's name and every
    name/description/tag token using :class:`difflib.SequenceMatcher`;
    ``options.fuzzy_threshold`` gates how permissive the match is (``1.0`` ==
    exact only, lower == more typo tolerance).
    """

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(spec="072", story="US-072-004", reqs=["REQ-072-002"])
    def score(
        self, parsed: ParsedQuery, entry: ToolIndexEntry, options: SearchOptions
    ) -> MatchScore:
        with self._tracer.span("FuzzyMatchStrategy.score"):
            threshold = options.fuzzy_threshold
            field_tokens = _entry_field_tokens(entry)
            candidates: list[tuple[str, str]] = [("name", entry.name)]
            for field_name, toks in field_tokens.items():
                candidates.extend((field_name, tok) for tok in toks)

            best_score = 0.0
            best_fields: set[str] = set()
            any_group_matched = False
            for group in parsed.groups:
                if not group:
                    continue
                term_best: list[float] = []
                group_fields: set[str] = set()
                for term in group:
                    clean = term.lower().rstrip("*")
                    if not clean:
                        term_best.append(0.0)
                        continue
                    ratio_best = 0.0
                    field_best: str | None = None
                    for field_name, candidate in candidates:
                        ratio = difflib.SequenceMatcher(
                            None, clean, candidate.lower()
                        ).ratio()
                        if ratio > ratio_best:
                            ratio_best = ratio
                            field_best = field_name
                    term_best.append(ratio_best)
                    if ratio_best >= threshold and field_best:
                        group_fields.add(field_best)
                if term_best and all(r >= threshold for r in term_best):
                    any_group_matched = True
                    group_score = sum(term_best) / len(term_best)
                    if group_score > best_score:
                        best_score = group_score
                        best_fields = group_fields
            return MatchScore(
                score=best_score if any_group_matched else 0.0,
                matched_fields=tuple(sorted(best_fields)),
                fuzzy_applied=any_group_matched and best_score > 0.0,
            )


# ── ranking ───────────────────────────────────────────────────────────


@traces(spec="072", story="US-072-001", reqs=["REQ-072-005"])
class RelevanceRanker:
    """Blends match quality with popularity/success/recency (spec.md formula).

    Owns all five ranking weight constants:
    ``final = 0.40*semantic + 0.25*keyword + 0.15*popularity +
    0.10*success_rate + 0.10*recency``. The semantic+keyword portion arrives
    pre-combined in each entry's :class:`MatchScore` (see
    :meth:`ToolSearchEngine._combine`, which reads
    :data:`WEIGHT_SEMANTIC`/:data:`WEIGHT_KEYWORD` from here); this class
    adds the remaining three signals, derived from :class:`ToolIndexEntry`.
    """

    WEIGHT_SEMANTIC: ClassVar[float] = 0.40
    WEIGHT_KEYWORD: ClassVar[float] = 0.25
    WEIGHT_POPULARITY: ClassVar[float] = 0.15
    WEIGHT_SUCCESS_RATE: ClassVar[float] = 0.10
    WEIGHT_RECENCY: ClassVar[float] = 0.10

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-005"])
    def rank(
        self,
        scored: list[tuple[ToolIndexEntry, MatchScore]],
        options: SearchOptions,
    ) -> list[SearchResultItem]:
        with self._tracer.span("RelevanceRanker.rank"):
            if not scored:
                return []
            max_usage = max((entry.usage_count for entry, _ in scored), default=0)
            now = datetime.now(UTC)
            items: list[SearchResultItem] = []
            for entry, match in scored:
                popularity = (entry.usage_count / max_usage) if max_usage > 0 else 0.0
                success = max(0.0, min(1.0, entry.success_rate))
                recency = self._recency_score(entry.updated_at, now)
                final_score = (
                    match.score
                    + self.WEIGHT_POPULARITY * popularity
                    + self.WEIGHT_SUCCESS_RATE * success
                    + self.WEIGHT_RECENCY * recency
                )
                final_score = max(0.0, min(1.0, final_score))
                items.append(
                    SearchResultItem(
                        entry=entry,
                        score=final_score,
                        matched_fields=match.matched_fields,
                        highlight=self._highlight(entry),
                    )
                )
            items.sort(key=lambda item: item.score, reverse=True)
            if options.rank_by == "popularity":
                items.sort(key=lambda item: item.entry.usage_count, reverse=True)
            elif options.rank_by == "recency":
                items.sort(key=lambda item: item.entry.updated_at, reverse=True)
            elif options.rank_by == "success_rate":
                items.sort(key=lambda item: item.entry.success_rate, reverse=True)
            return items

    @staticmethod
    @traces(spec="072", story="US-072-001", reqs=["REQ-072-005"])
    def _recency_score(updated_at: str, now: datetime) -> float:
        try:
            parsed = datetime.fromisoformat(updated_at)
        except ValueError, TypeError:
            return 0.5
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=UTC)
        age_days = max(0.0, (now - parsed).total_seconds() / 86400.0)
        return max(0.0, 1.0 - age_days / 365.0)

    @staticmethod
    def _highlight(entry: ToolIndexEntry) -> str:
        text = entry.description or entry.name
        if len(text) > 160:
            text = text[:157] + "..."
        return text


# ── index ─────────────────────────────────────────────────────────────


@traces(
    spec="072",
    story="US-072-001",
    reqs=[
        "REQ-072-011",
        "REQ-072-012",
        "REQ-072-013",
        "REQ-072-014",
        "REQ-072-015",
        "REQ-072-022",
        "REQ-072-023",
    ],
)
class ToolIndex:
    """The in-memory, thread-safe searchable corpus (spec.md Index Management).

    Sourced from :meth:`~orchestrata_ultimate.tools.registry.ToolRegistry.metadata`
    (FR-011); ``entries()`` snapshots under a lock so concurrent
    :meth:`sync`/:meth:`rebuild` calls never produce torn reads (REQ-072-022).
    """

    def __init__(
        self,
        registry: ToolRegistry,
        embeddings: EmbeddingService | None,
        entries: dict[str, ToolIndexEntry],
        tracer: Tracer | None = None,
    ) -> None:
        self._registry = registry
        self._embeddings = embeddings
        self._entries = entries
        self._tracer = tracer if tracer is not None else Tracer.default()
        self._lock = threading.Lock()
        self._ready = bool(entries)

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-013"])
    def rebuild(self) -> int:
        """Clear and fully repopulate the index from the registry (FR-013)."""
        with self._tracer.span("ToolIndex.rebuild"):
            with self._lock:
                self._entries.clear()
                self._ready = False
            return self.sync()

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-012", "REQ-072-015"])
    def sync(self) -> int:
        """Add/update entries whose registry metadata changed (FR-012/015).

        Returns the number of entries touched (added or updated) — unchanged
        entries are left object-identical (no embedding recompute).
        """
        with self._tracer.span("ToolIndex.sync"):
            touched = 0
            with self._lock:
                for meta in self._registry.metadata():
                    tool_id = meta.name
                    fingerprint = (
                        meta.description,
                        str(meta.category),
                        tuple(meta.tags),
                        meta.version,
                    )
                    current = self._entries.get(tool_id)
                    current_fp = (
                        (
                            current.description,
                            current.category,
                            current.tags,
                            current.version,
                        )
                        if current is not None
                        else None
                    )
                    if current is None or current_fp != fingerprint:
                        entry = ToolIndexEntry.from_metadata(meta)
                        if self._embeddings is not None:
                            text = (
                                f"{meta.name} {meta.description} {' '.join(meta.tags)}"
                            )
                            entry = replace(
                                entry,
                                embedding=tuple(
                                    self._embeddings.compute_embedding(text)
                                ),
                            )
                        self._entries[tool_id] = entry
                        touched += 1
                self._ready = True
            return touched

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-012"])
    def remove(self, tool_id: str) -> bool:
        """Remove ``tool_id`` from the index (FR-012).

        Returns ``False`` when the tool was never indexed — the in-process
        analogue of ERR-006 TOOL_NOT_FOUND for this lookup surface.
        """
        with self._tracer.span("ToolIndex.remove"):
            with self._lock:
                return self._entries.pop(tool_id, None) is not None

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-011", "REQ-072-022"])
    def entries(self) -> list[ToolIndexEntry]:
        """A lock-protected snapshot of all indexed entries (FR-011/FR-022)."""
        with self._lock:
            return list(self._entries.values())

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-104"])
    def ready(self) -> bool:
        """Whether at least one successful build/sync has completed (ERR-004)."""
        with self._lock:
            return self._ready

    def stats(self) -> dict[str, Any]:
        """Basic index statistics (count + readiness).

        Note: not traced to REQ-072-023 — that requirement is a capacity
        target ("index size SHALL support at least 10,000 tools"), not a
        statistics-reporting feature; this accessor doesn't implement or
        verify capacity. No indexed requirement describes the
        ``GET /index/stats`` reporting surface itself (spec.md documents the
        endpoint but the requirements checklist has no distinct REQ id for
        it), so this method is intentionally left untraced.
        """
        with self._lock:
            return {"count": len(self._entries), "ready": self._ready}

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-014"])
    def save(self, path: Path) -> None:
        """Persist the index to ``path`` as JSON (FR-014)."""
        with self._tracer.span("ToolIndex.save"):
            with self._lock:
                payload = [asdict(e) for e in self._entries.values()]
            path.write_text(json.dumps(payload))

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-014", "REQ-072-105"])
    def load(self, path: Path) -> bool:
        """Load a previously :meth:`save`-d index from ``path`` (FR-014).

        Returns ``False`` (never raises) on a missing/corrupted file
        (ERR-005 INDEX_CORRUPTED) — callers should :meth:`rebuild` instead.
        """
        with self._tracer.span("ToolIndex.load"):
            try:
                raw = json.loads(path.read_text())
                loaded: dict[str, ToolIndexEntry] = {}
                for item in raw:
                    record = dict(item)
                    record["tags"] = tuple(record.get("tags", ()))
                    record["embedding"] = tuple(record.get("embedding", ()))
                    record["parameters"] = dict(record.get("parameters", {}))
                    entry = ToolIndexEntry(**record)
                    loaded[entry.tool_id] = entry
            except OSError, json.JSONDecodeError, TypeError, KeyError:
                return False
            with self._lock:
                self._entries = loaded
                self._ready = True
            return True


# ── cache ─────────────────────────────────────────────────────────────


@traces(
    spec="072",
    story="US-072-005",
    reqs=["REQ-072-016", "REQ-072-017", "REQ-072-018", "REQ-072-019", "REQ-072-020"],
)
class SearchCache:
    """TTL/LRU cache of search results (spec.md Caching Strategy).

    Composes the shared :class:`~orchestrata_ultimate.core.lru_cache.LRUCache`
    (the exact 5-minute/1000-entry policy spec.md calls for) and tracks which
    cache keys reference which ``tool_id`` so :meth:`invalidate_tool` can
    evict precisely (FR-017/FR-018) without a full :meth:`clear`.
    """

    def __init__(
        self,
        cache: LRUCache[SearchResponse],
        tool_keys: dict[str, set[str]],
        tracer: Tracer | None = None,
    ) -> None:
        self._cache = cache
        self._tool_keys = tool_keys
        self._tracer = tracer if tracer is not None else Tracer.default()

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-016"])
    def key_for(self, query: SearchQuery) -> str:
        """A deterministic hash of query + normalized filters + options (FR-016)."""
        payload = {
            "query": query.query.strip().lower(),
            "filters": {
                "categories": sorted(query.filters.categories),
                "tags": sorted(query.filters.tags),
                "return_types": sorted(query.filters.return_types),
                "authors": sorted(query.filters.authors),
                "min_success_rate": query.filters.min_success_rate,
            },
            "options": {
                "limit": query.options.limit,
                "offset": query.options.offset,
                "fuzzy_threshold": query.options.fuzzy_threshold,
                "include_disabled": query.options.include_disabled,
                "rank_by": query.options.rank_by,
            },
        }
        raw = json.dumps(payload, sort_keys=True)
        return hashlib.sha256(raw.encode()).hexdigest()

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-107"])
    def get(self, key: str) -> SearchResponse | None:
        """Cache lookup; any cache failure degrades to a miss (ERR-007)."""
        with self._tracer.span("SearchCache.get"):
            try:
                return self._cache.get(key)
            except Exception:  # noqa: BLE001 - ERR-007: continue without cache
                return None

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-016"])
    def put(self, key: str, response: SearchResponse, tool_ids: list[str]) -> None:
        """Cache ``response`` under ``key`` and index it by each of ``tool_ids``."""
        with self._tracer.span("SearchCache.put"):
            try:
                self._cache.put(key, response)
            except Exception:  # noqa: BLE001 - ERR-007: continue without cache
                return
            for tool_id in tool_ids:
                self._tool_keys.setdefault(tool_id, set()).add(key)

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-017", "REQ-072-018"])
    def invalidate_tool(self, tool_id: str) -> int:
        """Evict every cached response referencing ``tool_id`` (FR-017/018)."""
        with self._tracer.span("SearchCache.invalidate_tool"):
            keys = self._tool_keys.pop(tool_id, set())
            return sum(1 for key in keys if self._cache.invalidate(key))

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-017"])
    def clear(self) -> None:
        """Evict everything (e.g. after an index rebuild).

        REQ-072-017's title is narrowly "invalidate cache entries when tools
        are updated", but spec.md's Caching Strategy narrative explicitly
        widens invalidation triggers to "Tool updates, index rebuild, manual
        invalidation" — a full clear (as opposed to REQ-072-016's ongoing
        TTL-based caching behaviour) falls under that same invalidation
        requirement, not the TTL-caching one.
        """
        with self._tracer.span("SearchCache.clear"):
            self._cache.clear()
            self._tool_keys.clear()

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-019"])
    def warm(
        self, queries: list[str], search: Callable[[SearchQuery], SearchResponse]
    ) -> None:
        """Pre-populate the cache by running ``search`` for each of ``queries`` (FR-019)."""
        with self._tracer.span("SearchCache.warm"):
            for raw_query in queries:
                if not raw_query or not raw_query.strip():
                    continue
                query = SearchQuery(query=raw_query)
                try:
                    response = search(query)
                except ToolSearchError:
                    continue
                key = self.key_for(query)
                tool_ids = [item.entry.tool_id for item in response.results]
                self.put(key, response, tool_ids)

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-020"])
    def stats(self) -> dict[str, Any]:
        """Hit/miss/size statistics (FR-020), delegated to the core LRU cache."""
        return self._cache.get_stats()


# ── engine (Command pipeline) ────────────────────────────────────────────


@traces(
    spec="072",
    story="US-072-001",
    reqs=[
        "REQ-072-001",
        "REQ-072-003",
        "REQ-072-004",
        "REQ-072-006",
        "REQ-072-007",
        "REQ-072-009",
        "REQ-072-021",
        "REQ-072-022",
    ],
)
class ToolSearchEngine:
    """The search pipeline: validate -> cache -> filter -> score -> rank -> page."""

    def __init__(
        self,
        index: ToolIndex,
        validator: QueryValidator,
        parser: QueryParser,
        strategies: list[MatchStrategy],
        ranker: RelevanceRanker,
        cache: SearchCache,
        history: list[str],
        tracer: Tracer | None = None,
    ) -> None:
        self._index = index
        self._validator = validator
        self._parser = parser
        self._strategies = list(strategies)
        self._ranker = ranker
        self._cache = cache
        self._history = history
        self._tracer = tracer if tracer is not None else Tracer.default()

    def _strategy_of(self, kind: type) -> Any:
        return next((s for s in self._strategies if isinstance(s, kind)), None)

    @traces(
        spec="072",
        story="US-072-001",
        reqs=[
            "REQ-072-001",
            "REQ-072-003",
            "REQ-072-004",
            "REQ-072-006",
            "REQ-072-007",
            "REQ-072-009",
            "REQ-072-021",
            "REQ-072-022",
            "REQ-072-104",
            "REQ-072-112",
        ],
    )
    def search(
        self,
        query: SearchQuery,
        *,
        record_history: bool = True,
        timeout_seconds: float | None = None,
    ) -> SearchResponse:
        """Run the full pipeline for ``query`` (FR-001), honouring the cache.

        Args:
            query: The search request.
            record_history: Whether this call counts toward :meth:`suggest`'s
                history (US-072-005). ``False`` for internal/cache-warming
                calls (:meth:`ToolSearchTool.create`) so warming the cache
                with each tool's own name doesn't masquerade as user history.
            timeout_seconds: Optional per-call wall-clock budget. When omitted,
                the module default :data:`_SEARCH_TIMEOUT_SECONDS` is used.
                Tests pass ``0.0`` here to exercise ERR-012 deterministically
                without patching module state.

        Raises:
            ToolSearchError: propagated from :class:`QueryValidator`
                (ERR-001/002/003/011), ``INDEX_NOT_READY`` (ERR-004) if the
                index has never been built, or ``TIMEOUT`` (ERR-012) if the
                soft time budget elapses before anything was scored.
        """
        with self._tracer.span("ToolSearchEngine.search"):
            self._validator.validate(query)
            if record_history:
                self._record_history(query.query)

            cache_key = self._cache.key_for(query)
            cached = self._cache.get(cache_key)
            if cached is not None:
                return cached

            if not self._index.ready():
                raise ToolSearchError(
                    SearchErrorCode.INDEX_NOT_READY, "search index is not initialized"
                )

            sanitized = self._validator.sanitize(query.query)
            parsed = self._parser.parse(sanitized)

            semantic = self._strategy_of(SemanticMatchStrategy)
            keyword = self._strategy_of(KeywordMatchStrategy)
            fuzzy = self._strategy_of(FuzzyMatchStrategy)

            entries = self._apply_filters(self._index.entries(), query.filters)

            scored: list[tuple[ToolIndexEntry, MatchScore]] = []
            budget = (
                timeout_seconds
                if timeout_seconds is not None
                else _SEARCH_TIMEOUT_SECONDS
            )
            deadline = time.monotonic() + budget
            timed_out = False
            for entry in entries:
                if time.monotonic() > deadline:
                    timed_out = True
                    break
                # REQ-072-108/ERR-008: an unavailable semantic strategy simply
                # contributes nothing — keyword/fuzzy carry the match alone.
                semantic_ms = (
                    semantic.score(parsed, entry, query.options)
                    if semantic is not None and semantic.available()
                    else MatchScore()
                )
                keyword_ms = (
                    keyword.score(parsed, entry, query.options)
                    if keyword is not None
                    else MatchScore()
                )
                fuzzy_ms = (
                    fuzzy.score(parsed, entry, query.options)
                    if fuzzy is not None
                    else MatchScore()
                )
                combined = self._combine(semantic_ms, keyword_ms, fuzzy_ms)
                if combined.score >= _MIN_RELEVANCE_SCORE:
                    scored.append((entry, combined))

            if timed_out and not scored:
                raise ToolSearchError(
                    SearchErrorCode.TIMEOUT,
                    "search timed out before any results could be scored",
                )

            ranked = self._ranker.rank(scored, query.options)
            total_count = len(ranked)
            page = self._paginate(ranked, query.options)
            facets = self._facets(ranked)
            suggestions = self.suggest(query.query)

            response = SearchResponse(
                results=page,
                total_count=total_count,
                query=query.query,
                limit=query.options.limit,
                offset=query.options.offset,
                has_more=(query.options.offset + query.options.limit) < total_count,
                suggestions=suggestions,
                facets=facets,
            )
            self._cache.put(cache_key, response, [item.entry.tool_id for item in page])
            return response

    @staticmethod
    @traces(spec="072", story="US-072-001", reqs=["REQ-072-005"])
    def _combine(
        semantic_ms: MatchScore, keyword_ms: MatchScore, fuzzy_ms: MatchScore
    ) -> MatchScore:
        """Fold semantic + (keyword-or-fuzzy) into one weighted MatchScore.

        Applies :data:`RelevanceRanker.WEIGHT_SEMANTIC`/``WEIGHT_KEYWORD`` —
        :class:`RelevanceRanker` applies the remaining three weights.
        """
        keyword_component = keyword_ms.score
        fuzzy_applied = False
        if fuzzy_ms.score > keyword_component:
            keyword_component = fuzzy_ms.score
            fuzzy_applied = fuzzy_ms.fuzzy_applied
        combined_score = (
            RelevanceRanker.WEIGHT_SEMANTIC * semantic_ms.score
            + RelevanceRanker.WEIGHT_KEYWORD * keyword_component
        )
        matched_fields = tuple(
            sorted(
                set(semantic_ms.matched_fields)
                | set(keyword_ms.matched_fields)
                | set(fuzzy_ms.matched_fields)
            )
        )
        return MatchScore(
            score=combined_score,
            matched_fields=matched_fields,
            fuzzy_applied=fuzzy_applied,
        )

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-006"])
    def suggest(self, partial: str) -> list[str]:
        """Recent-history suggestions matching ``partial`` (FR-006).

        Per US-072-005's acceptance criteria, suggestions only appear once
        3+ historical queries have been recorded.
        """
        with self._tracer.span("ToolSearchEngine.suggest"):
            if len(self._history) < 3:
                return []
            needle = partial.strip().lower()
            seen: list[str] = []
            for past in reversed(self._history):
                if needle and needle not in past.lower():
                    continue
                if past not in seen:
                    seen.append(past)
                if len(seen) >= 5:
                    break
            return seen

    @traces(spec="072", story="US-072-003", reqs=["REQ-072-003"])
    def _apply_filters(
        self, entries: list[ToolIndexEntry], filters: SearchFilters
    ) -> list[ToolIndexEntry]:
        """Category/tag/return_type/author/min_success_rate filtering (FR-003)."""
        result = entries
        if filters.categories:
            result = [e for e in result if e.category in filters.categories]
        if filters.tags:
            result = [e for e in result if any(t in e.tags for t in filters.tags)]
        if filters.return_types:
            result = [e for e in result if e.return_type in filters.return_types]
        if filters.authors:
            result = [e for e in result if e.author in filters.authors]
        if filters.min_success_rate > 0.0:
            result = [e for e in result if e.success_rate >= filters.min_success_rate]
        return result

    @traces(spec="072", story="US-072-003", reqs=["REQ-072-004"])
    def _paginate(
        self, items: list[SearchResultItem], options: SearchOptions
    ) -> list[SearchResultItem]:
        """Slice ``items`` by ``options.offset``/``options.limit`` (FR-004)."""
        return items[options.offset : options.offset + options.limit]

    @traces(spec="072", story="US-072-002", reqs=["REQ-072-007"])
    def _facets(self, items: list[SearchResultItem]) -> dict[str, list[FacetCount]]:
        """Category/tag/return_type distribution over the full result set (FR-007)."""
        categories: Counter[str] = Counter()
        tags: Counter[str] = Counter()
        return_types: Counter[str] = Counter()
        for item in items:
            categories[item.entry.category] += 1
            return_types[item.entry.return_type] += 1
            for tag in item.entry.tags:
                tags[tag] += 1

        def _to_facets(counter: Counter[str]) -> list[FacetCount]:
            return [
                FacetCount(value=value, count=count)
                for value, count in sorted(
                    counter.items(), key=lambda kv: (-kv[1], kv[0])
                )
            ]

        return {
            "categories": _to_facets(categories),
            "tags": _to_facets(tags),
            "return_types": _to_facets(return_types),
        }

    @traces(spec="072", story="US-072-005", reqs=["REQ-072-006"])
    def _record_history(self, query: str) -> None:
        """Append ``query`` to the search history used by :meth:`suggest`."""
        cleaned = query.strip()
        if cleaned:
            self._history.append(cleaned)


# ── the tool itself (Command + Factory + Singleton registration) ────────


@traces(
    spec="072",
    story="US-072-001",
    reqs=["REQ-072-001", "REQ-072-013", "REQ-072-019"],
)
class ToolSearchTool(BaseTool):
    """The registrable ``BaseTool`` (Command) callers/LLMs actually invoke.

    Composes a :class:`ToolSearchEngine`; :meth:`create` is the Factory that
    wires the whole pipeline (index/strategies/cache/ranker) from a live
    :class:`~orchestrata_ultimate.tools.registry.ToolRegistry`.
    """

    def __init__(
        self,
        engine: ToolSearchEngine,
        metadata: ToolMetadata,
        tracer: Tracer | None = None,
    ) -> None:
        self._engine = engine
        self._metadata = metadata
        self._tracer = tracer if tracer is not None else Tracer.default()

    @property
    def metadata(self) -> ToolMetadata:
        """The tool's own registration metadata (BaseTool Protocol contract).

        Not traced to REQ-072-001 — that requirement is about the search
        behaviour ("accept a query, return relevant tools"), not this tool
        exposing its own metadata for registration/discovery. This accessor
        implements the same generic ``BaseTool.metadata`` contract as every
        other tool (SPEC-062 framework code), for which no REQ id is indexed.
        """
        return self._metadata

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-001"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Run a search from raw tool-call ``arguments`` (FR-001).

        Never raises: any :class:`ToolSearchError` (or unforeseen exception)
        becomes a failed :class:`ToolResult` instead (ERR-001/ERR-004/...).
        """
        with self._tracer.span("ToolSearchTool.execute"):
            try:
                query = SearchQuery.from_arguments(arguments)
                response = self._engine.search(query)
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=True,
                    output=response.to_json(),
                )
            except ToolSearchError as exc:
                return exc.to_result(self.metadata.name)
            except (
                Exception
            ) as exc:  # noqa: BLE001 - never propagate; report as ToolResult
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=False,
                    output="",
                    error=str(exc),
                )

    @traces(spec="072", story="US-072-001", reqs=["REQ-072-013", "REQ-072-019"])
    @classmethod
    def create(
        cls, registry: ToolRegistry, trace_config: TraceConfig | None = None
    ) -> ToolSearchTool:
        """Build a fully-wired ``ToolSearchTool`` over ``registry`` (FR-013/FR-019).

        Rebuilds the index immediately (rebuild-on-startup) and warms the
        cache with each indexed tool's own name — the most likely first
        lookup for a freshly booted search tool.
        """
        tracer = Tracer(trace_config) if trace_config is not None else Tracer.default()
        embeddings = EmbeddingService()
        index = ToolIndex(
            registry=registry, embeddings=embeddings, entries={}, tracer=tracer
        )
        index.rebuild()

        keyword = KeywordMatchStrategy(idf={}, tracer=tracer)
        keyword.fit(index.entries())
        semantic = SemanticMatchStrategy(embeddings=embeddings, tracer=tracer)
        fuzzy = FuzzyMatchStrategy(tracer=tracer)
        ranker = RelevanceRanker(tracer=tracer)
        cache = SearchCache(
            cache=LRUCache(max_size=1000, default_ttl=300.0),
            tool_keys={},
            tracer=tracer,
        )
        engine = ToolSearchEngine(
            index=index,
            validator=QueryValidator(tracer=tracer),
            parser=QueryParser(tracer=tracer),
            strategies=[semantic, keyword, fuzzy],
            ranker=ranker,
            cache=cache,
            history=[],
            tracer=tracer,
        )
        warm_queries = [entry.name for entry in index.entries()]
        cache.warm(
            warm_queries,
            lambda q: engine.search(q, record_history=False),
        )

        metadata = ToolMetadata(
            name="search_tools",
            description=(
                "Search for registered tools by natural-language query, "
                "keyword, AND/OR/wildcard expression, or fuzzy tool name."
            ),
            category=ToolCategory.SEARCH,
            parameters=_SEARCH_TOOL_PARAMETERS,
            tags=("search", "tools", "discovery"),
        )
        return cls(engine=engine, metadata=metadata, tracer=tracer)


@traces(spec="072", story="US-072-001", reqs=["REQ-072-013"])
def register_search_tool(
    registry: ToolRegistry, trace_config: TraceConfig | None = None
) -> ToolSearchTool:
    """Build a :class:`ToolSearchTool` over ``registry`` and register it.

    The Factory+Singleton wiring seam: constructs the tool via
    :meth:`ToolSearchTool.create` and registers it into ``registry`` (e.g.
    :func:`orchestrata_ultimate.tools.registry.default_registry`) so it is
    itself discoverable/searchable/dispatchable like any other tool.
    """
    tool = ToolSearchTool.create(registry, trace_config)
    registry.register(tool, replace=True)
    return tool