orchestrata_ultimate.tools.context_tools

Reference › orchestrata_ultimate › tools › context_tools

This module provides four on-demand context tools (SPEC-107 FR-007) that let a volk pull structured context from the operational knowledge graph and the semantic graph when it needs it, instead of relying only on the context assembled at injection time (see assembler). agents.tool_registry imports it; tests are tests/tools/test_context_tools.py and tests/tools/test_context_tools_coverage.py.

All four are BaseTool classes traced to REQ-107-050 to REQ-107-054:

• GetCodeTool returns content or the signature of a canonical code entity. If the exact id misses, it falls back to a lookup by path and name.
• GetTestsTool returns the tests that cover an entity.
• GetRequirementsTool returns the requirement items the entity traces to, collected directly and by a path-and-name fallback.
• SearchCodeTool performs semantic search over the semantic graph by reusing semantic_context, so there is one implementation of the mapping.

_LazyRepo holds the graph repository and creates it on first use. The project id comes from the environment variable ORCHESTRATA_PROJECT_ID at call time. Entity ids are split into project, path and qualified name by _split_entity_id. When a backend is unreachable, each tool returns a failed but clear ToolResult. register_context_tools registers all four.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

_LazyRepo

Line 91

Lazy, instance-scoped SyncGraphRepository.

GetCodeTool

Bases: BaseTool · Decorators: traces(spec='107', reqs=['REQ-107-050', 'REQ-107-054']) · Line 205

Return content or signature for a canonical code entity from the graph.

GetTestsTool

Bases: BaseTool · Decorators: traces(spec='107', reqs=['REQ-107-051', 'REQ-107-054']) · Line 349

Return the tests that cover a canonical code entity.

GetRequirementsTool

Bases: BaseTool · Decorators: traces(spec='107', reqs=['REQ-107-052', 'REQ-107-054']) · Line 462

Return requirements traced by a canonical code entity.

SearchCodeTool

Bases: BaseTool · Decorators: traces(spec='107', reqs=['REQ-107-053', 'REQ-107-054']) · Line 617

Semantic search across the codebase via the Memgraph semantic graph.

Functions

Source

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

"""On-demand context tools for volks — get_code, get_tests, get_requirements, search_code.

Spec: SPEC-107 FR-007. These are first-class SPEC-062 native tools that let a
volk pull structured context from the operational knowledge graph and the
Memgraph semantic graph on demand, rather than relying solely on the context
assembled at packet injection time.

Architecture:
    - Each tool is a :class:`~orchestrata_ultimate.tools.base.BaseTool`.
    - Tools compose a lazy :class:`~orchestrata_ultimate.core.graph.SyncGraphRepository`
      and an optional project id (read from ``ORCHESTRATA_PROJECT_ID`` at call
      time with a sensible default, per project env-injection convention).
    - ``search_code`` reuses the existing semantic-search bridge
      (:func:`~orchestrata_ultimate.context.providers.semantic_context.render_semantic_context`)
      so there is one implementation of the Memgraph→Neo4j mapping, not two.
    - Every execute path degrades to a failed-but-clear :class:`ToolResult` when
      its backend is unreachable, matching SPEC-107 ruling #3 (degrade, don't
      crash).
"""

from __future__ import annotations

import json
import logging
import os
from typing import Any

from orchestrata_ultimate.context.providers.semantic_context import (
    render_semantic_context,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import create_graph_repository
from orchestrata_ultimate.core.graph.interfaces import SyncGraphRepository
from orchestrata_ultimate.core.graph_factory import NO_AUTH, create_graph
from orchestrata_ultimate.knowledge_graph.requirement_resolution import (
    resolve_requirement_chain,
)
from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)
from orchestrata_ultimate.tools.registry import ToolRegistry

_logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Entity-id parsing
# ---------------------------------------------------------------------------


@traces(
    spec="107",
    story="FR-007",
    reqs=["REQ-107-050", "REQ-107-051", "REQ-107-052"],
)
def _split_entity_id(entity_id: str) -> tuple[str | None, str, str | None]:
    """Split an entity id into (project, path, qualname).

    Supported shapes:
        - ``{project}:{relpath}::{qualname}`` (canonical code entity)
        - ``{relpath}::{qualname}``                    (code entity, default project)
        - ``{project}:{path}``                         (file entity)
        - ``{path}``                                   (file entity, default project)

    Returns:
        ``(project_or_none, path, qualname_or_none)``.
    """
    if "::" in entity_id:
        head, _sep, qualname = entity_id.partition("::")
        if ":" in head:
            project, _sep2, path = head.partition(":")
            return project, path, qualname
        return None, head, qualname

    head = entity_id
    if ":" in head:
        project, _sep, path = head.partition(":")
        return project, path, None
    return None, head, None


# ---------------------------------------------------------------------------
# Repository helper
# ---------------------------------------------------------------------------


class _LazyRepo:
    """Lazy, instance-scoped SyncGraphRepository.

    Lets tools be registered with ``register_native_tools()`` (which calls
    constructors with no arguments) while still sharing one repository across
    multiple calls within the lifetime of a tool instance.
    """

    def __init__(self, repo: SyncGraphRepository | None = None) -> None:
        self._repo = repo
        self._failed = False

    @traces(spec="107", story="FR-007", reqs=["REQ-107-054"])
    def get(self) -> SyncGraphRepository | None:
        if self._repo is not None:
            return self._repo
        if self._failed:
            return None
        try:
            self._repo = create_graph_repository()
            return self._repo
        except Exception as exc:  # noqa: BLE001 — degrade, don't crash
            _logger.warning("Context tool cannot create graph repository: %s", exc)
            self._failed = True
            return None


# ---------------------------------------------------------------------------
# Shared result formatting
# ---------------------------------------------------------------------------


@traces(spec="107", story="FR-007", reqs=["REQ-107-054"])
def _ok(tool_name: str, payload: dict[str, Any]) -> ToolResult:
    return ToolResult(tool_name, True, json.dumps(payload, indent=2))


@traces(spec="107", story="FR-007", reqs=["REQ-107-054"])
def _error(tool_name: str, message: str) -> ToolResult:
    return ToolResult(
        tool_name,
        False,
        json.dumps({"error": message}),
        message,
    )


# ---------------------------------------------------------------------------
# Tool parameters
# ---------------------------------------------------------------------------

_GET_CODE_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "entity_id": {
            "type": "string",
            "description": (
                "Canonical entity id. Forms: '{project}:{relpath}::{qualname}' "
                "for code entities, '{project}:{path}' or '{path}' for files."
            ),
        },
    },
    "required": ["entity_id"],
}

_GET_TESTS_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "entity_id": {
            "type": "string",
            "description": (
                "Canonical code entity id (e.g. '{project}:src/pkg/mod.py::Class.method')."
            ),
        },
    },
    "required": ["entity_id"],
}

_GET_REQUIREMENTS_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "entity_id": {
            "type": "string",
            "description": (
                "Canonical code entity id (e.g. '{project}:src/pkg/mod.py::Class.method')."
            ),
        },
    },
    "required": ["entity_id"],
}

_SEARCH_CODE_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "description": "Natural-language query to search the semantic graph.",
        },
        "target_files": {
            "type": "array",
            "items": {"type": "string"},
            "description": "Optional file paths to bias the search.",
        },
    },
    "required": ["query"],
}


# ---------------------------------------------------------------------------
# GetCodeTool
# ---------------------------------------------------------------------------


@traces(spec="107", reqs=["REQ-107-050", "REQ-107-054"])
class GetCodeTool(BaseTool):
    """Return content or signature for a canonical code entity from the graph."""

    def __init__(self, repo: SyncGraphRepository | None = None) -> None:
        self._metadata = ToolMetadata(
            name="get_code",
            description=(
                "Fetch the stored source content or signature for a canonical "
                "code entity (File, Class, Function, or Method) from the "
                "knowledge graph."
            ),
            category=ToolCategory.SEARCH,
            parameters=_GET_CODE_PARAMETERS,
            priority=ToolPriority.P1,
            tags=("kg", "code", "context"),
        )
        self._lazy_repo = _LazyRepo(repo)

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

    @traces(spec="107", reqs=["REQ-107-050", "REQ-107-054"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        entity_id = (arguments.get("entity_id") or "").strip()
        if not entity_id:
            return _error(self.name, "entity_id is required")

        repo = self._lazy_repo.get()
        if repo is None:
            return _error(self.name, "Knowledge graph unavailable")

        project, path, qualname = _split_entity_id(entity_id)
        if project is None:
            project = os.environ.get("ORCHESTRATA_PROJECT_ID", "orchestrata-ultimate")

        try:
            # First try the exact canonical id.
            rows = repo.execute_read(
                "MATCH (n) WHERE n.id = $id RETURN labels(n)[0] AS label, "
                "       properties(n) AS props LIMIT 1",
                {"id": entity_id},
            )
            if not rows:
                # Fallback: resolve file by project-qualified or bare path,
                # then the named symbol inside it.
                rows = self._resolve_by_path_and_name(repo, project, path, qualname)

            if not rows:
                return _error(self.name, f"Entity not found: {entity_id}")

            record = rows[0]
            label = record.get("label", "Unknown")
            props: dict[str, Any] = record.get("props") or {}
            content = props.get("content")
            if content:
                return _ok(
                    self.name,
                    {
                        "entity_id": entity_id,
                        "label": label,
                        "path": props.get("path"),
                        "name": props.get("name"),
                        "content": content,
                        "line_count": props.get("line_count"),
                    },
                )

            # No stored content: render a signature summary.
            signature = props.get("signature") or props.get("name") or entity_id
            docstring = props.get("docstring") or ""
            return _ok(
                self.name,
                {
                    "entity_id": entity_id,
                    "label": label,
                    "path": props.get("path"),
                    "name": props.get("name"),
                    "signature": signature,
                    "docstring": docstring.split("\n")[0] if docstring else "",
                    "note": "Content not stored; returning signature.",
                },
            )
        except Exception as exc:  # noqa: BLE001 — degrade, don't crash
            _logger.warning("get_code failed for %r: %s", entity_id, exc)
            return _error(self.name, f"Knowledge graph query failed: {exc}")

    @traces(spec="107", story="FR-007", reqs=["REQ-107-050"])
    def _resolve_by_path_and_name(
        self,
        repo: SyncGraphRepository,
        project: str,
        path: str,
        qualname: str | None,
    ) -> list[dict[str, Any]]:
        """Fallback resolution when the exact canonical id misses.

        Looks for a File by project-qualified id or bare path, then for a
        Class/Function/Method by name inside that file.
        """
        try:
            file_rows = repo.execute_read(
                "MATCH (f:File) WHERE f.id = $qualified_id OR f.path = $path "
                "RETURN f.path AS path LIMIT 1",
                {"qualified_id": f"{project}:{path}", "path": path},
            )
            if not file_rows:
                return []
            file_path = file_rows[0]["path"]

            if qualname is None:
                # File-only lookup.
                return repo.execute_read(
                    "MATCH (f:File) WHERE f.path = $path "
                    "RETURN labels(f)[0] AS label, properties(f) AS props LIMIT 1",
                    {"path": file_path},
                )

            # Named symbol: try the qualname as-is, then the final dotted part.
            names = [qualname]
            if "." in qualname:
                names.append(qualname.split(".")[-1])

            for name in names:
                rows = repo.execute_read(
                    "MATCH (f:File {path: $path})-[:DEFINES]->(n) "
                    "WHERE n.name = $name "
                    "RETURN labels(n)[0] AS label, properties(n) AS props LIMIT 1",
                    {"path": file_path, "name": name},
                )
                if rows:
                    return rows
            return []
        except Exception as exc:  # noqa: BLE001
            _logger.debug("get_code fallback resolution failed: %s", exc)
            return []


# ---------------------------------------------------------------------------
# GetTestsTool
# ---------------------------------------------------------------------------


@traces(spec="107", reqs=["REQ-107-051", "REQ-107-054"])
class GetTestsTool(BaseTool):
    """Return the tests that cover a canonical code entity."""

    def __init__(self, repo: SyncGraphRepository | None = None) -> None:
        self._metadata = ToolMetadata(
            name="get_tests",
            description=(
                "Return the test names and ids that cover a canonical code "
                "entity in the knowledge graph."
            ),
            category=ToolCategory.SEARCH,
            parameters=_GET_TESTS_PARAMETERS,
            priority=ToolPriority.P1,
            tags=("kg", "tests", "coverage"),
        )
        self._lazy_repo = _LazyRepo(repo)

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

    @traces(spec="107", reqs=["REQ-107-051", "REQ-107-054"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        entity_id = (arguments.get("entity_id") or "").strip()
        if not entity_id:
            return _error(self.name, "entity_id is required")

        repo = self._lazy_repo.get()
        if repo is None:
            return _error(self.name, "Knowledge graph unavailable")

        try:
            # COVERS is the legacy test-linkage name; TESTED_BY is canonical.
            # count(DISTINCT n) detects the no-such-entity case: a RETURN of
            # only aggregations yields exactly one row even when MATCH (n)
            # matched nothing, which made the path+name fallback below
            # unreachable before the count was added.
            rows = repo.execute_read(
                "MATCH (n {id: $id}) "
                "OPTIONAL MATCH (n)<-[:COVERS]-(t:Test) "
                "OPTIONAL MATCH (n)<-[:TESTED_BY]-(tr:TestResult) "
                "RETURN count(DISTINCT n) AS matched, "
                "       collect(DISTINCT {id: t.id, name: t.name}) AS tests, "
                "       collect(DISTINCT {id: tr.id, name: tr.name}) AS test_results",
                {"id": entity_id},
            )
            if not rows or not rows[0].get("matched"):
                # Fallback: find by path+name.
                rows = self._resolve_tests_fallback(repo, entity_id)

            if not rows:
                return _ok(self.name, {"entity_id": entity_id, "tests": []})

            record = rows[0]
            tests = [
                {"id": t.get("id"), "name": t.get("name")}
                for t in record.get("tests") or []
                if t.get("id") or t.get("name")
            ]
            test_results = [
                {"id": t.get("id"), "name": t.get("name")}
                for t in record.get("test_results") or []
                if t.get("id") or t.get("name")
            ]
            return _ok(
                self.name,
                {
                    "entity_id": entity_id,
                    "tests": tests,
                    "test_results": test_results,
                },
            )
        except Exception as exc:  # noqa: BLE001 — degrade, don't crash
            _logger.warning("get_tests failed for %r: %s", entity_id, exc)
            return _error(self.name, f"Knowledge graph query failed: {exc}")

    @traces(spec="107", story="FR-007", reqs=["REQ-107-051"])
    def _resolve_tests_fallback(
        self, repo: SyncGraphRepository, entity_id: str
    ) -> list[dict[str, Any]]:
        project, path, qualname = _split_entity_id(entity_id)
        if project is None:
            project = os.environ.get("ORCHESTRATA_PROJECT_ID", "orchestrata-ultimate")
        if qualname is None:
            return []
        names = [qualname]
        if "." in qualname:
            names.append(qualname.split(".")[-1])
        try:
            return repo.execute_read(
                "MATCH (f:File) WHERE f.id = $qualified_id OR f.path = $path "
                "MATCH (f)-[:DEFINES]->(n) WHERE n.name IN $names "
                "OPTIONAL MATCH (n)<-[:COVERS]-(t:Test) "
                "OPTIONAL MATCH (n)<-[:TESTED_BY]-(tr:TestResult) "
                "RETURN collect(DISTINCT {id: t.id, name: t.name}) AS tests, "
                "       collect(DISTINCT {id: tr.id, name: tr.name}) AS test_results",
                {
                    "qualified_id": f"{project}:{path}",
                    "path": path,
                    "names": names,
                },
            )
        except Exception as exc:  # noqa: BLE001
            _logger.debug("get_tests fallback failed: %s", exc)
            return []


# ---------------------------------------------------------------------------
# GetRequirementsTool
# ---------------------------------------------------------------------------


@traces(spec="107", reqs=["REQ-107-052", "REQ-107-054"])
class GetRequirementsTool(BaseTool):
    """Return requirements traced by a canonical code entity."""

    def __init__(self, repo: SyncGraphRepository | None = None) -> None:
        self._metadata = ToolMetadata(
            name="get_requirements",
            description=(
                "Return the RequirementItem and Requirement chain that a "
                "canonical code entity traces to in the knowledge graph."
            ),
            category=ToolCategory.SEARCH,
            parameters=_GET_REQUIREMENTS_PARAMETERS,
            priority=ToolPriority.P1,
            tags=("kg", "requirements", "traceability"),
        )
        self._lazy_repo = _LazyRepo(repo)

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

    @traces(spec="107", reqs=["REQ-107-052", "REQ-107-054"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        entity_id = (arguments.get("entity_id") or "").strip()
        if not entity_id:
            return _error(self.name, "entity_id is required")

        repo = self._lazy_repo.get()
        if repo is None:
            return _error(self.name, "Knowledge graph unavailable")

        try:
            # The entity id is project-QUALIFIED ("{project}:src/pkg/mod.py::X"),
            # so a caller that passes a canonical id has already said which
            # project it means. Prefer that over the ambient
            # ORCHESTRATA_PROJECT_ID: an explicit argument losing to a
            # process-global environment variable means this tool can only ever
            # answer about one project, and answers the WRONG one — silently,
            # as an empty result — for every entity outside it.
            #
            # Found when ORCHESTRATA_PROJECT_ID was declared in the env
            # profiles (2026-08-10). Until then the variable was unset here
            # (this call site, unlike its two neighbours in this module, passes
            # no default), so the queries were unscoped and happened to work.
            embedded_project, _path, _qualname = _split_entity_id(entity_id)
            project_id = embedded_project or os.environ.get("ORCHESTRATA_PROJECT_ID")
            item_ids = self._collect_traced_item_ids(repo, entity_id, project_id)
            if not item_ids:
                item_ids = self._collect_fallback_item_ids(repo, entity_id, project_id)

            chains = resolve_requirement_chain(item_ids, repo, project_id=project_id)
            items: list[dict[str, str]] = []
            requirements: list[dict[str, str]] = []
            specs: list[str] = []
            seen_specs: set[str] = set()
            for chain in chains:
                requirements.append(
                    {"id": chain.id, "text": chain.description or chain.id}
                )
                items.extend(
                    {"id": item_id, "text": chain.description or item_id}
                    for item_id in chain.item_ids
                )
                for spec in chain.specs:
                    if spec not in seen_specs:
                        seen_specs.add(spec)
                        specs.append(spec)

            return _ok(
                self.name,
                {
                    "entity_id": entity_id,
                    "requirement_items": items,
                    "requirements": requirements,
                    "specs": specs,
                },
            )
        except Exception as exc:  # noqa: BLE001 — degrade, don't crash
            _logger.warning("get_requirements failed for %r: %s", entity_id, exc)
            return _error(self.name, f"Knowledge graph query failed: {exc}")

    @traces(spec="107", story="FR-007", reqs=["REQ-107-052"])
    def _collect_traced_item_ids(
        self,
        repo: SyncGraphRepository,
        entity_id: str,
        project_id: str | None,
    ) -> list[str]:
        """Collect RequirementItem ids directly traced by the entity."""
        params: dict[str, Any] = {"id": entity_id}
        project_filter = ""
        if project_id is not None:
            project_filter = " AND ri.project_id = $project_id"
            params["project_id"] = project_id
        try:
            rows = repo.execute_read(
                "MATCH (n {id: $id})-[:TRACES_TO]->(ri:RequirementItem) "
                "WHERE true " + project_filter + " "
                "RETURN collect(DISTINCT ri.id) AS item_ids",
                params,
            )
            if rows:
                return [i for i in rows[0].get("item_ids", []) if i]
        except Exception as exc:  # noqa: BLE001
            _logger.debug("get_requirements direct lookup failed: %s", exc)
        return []

    @traces(spec="107", story="FR-007", reqs=["REQ-107-052"])
    def _collect_fallback_item_ids(
        self,
        repo: SyncGraphRepository,
        entity_id: str,
        project_id: str | None,
    ) -> list[str]:
        """Fallback: resolve by file path + symbol name when id has no TRACES_TO."""
        project, path, qualname = _split_entity_id(entity_id)
        if project is None:
            project = os.environ.get("ORCHESTRATA_PROJECT_ID", "orchestrata-ultimate")
        if qualname is None:
            return []
        names = [qualname]
        if "." in qualname:
            names.append(qualname.split(".")[-1])
        params: dict[str, Any] = {
            "qualified_id": f"{project}:{path}",
            "path": path,
            "names": names,
        }
        project_filter = ""
        if project_id is not None:
            project_filter = " AND f.project_id = $project_id"
            params["project_id"] = project_id
        try:
            rows = repo.execute_read(
                "MATCH (f:File) WHERE (f.id = $qualified_id OR f.path = $path) "
                + project_filter
                + " "
                "MATCH (f)-[:DEFINES]->(n) WHERE n.name IN $names "
                "OPTIONAL MATCH (n)-[:TRACES_TO]->(ri:RequirementItem) "
                "RETURN collect(DISTINCT ri.id) AS item_ids",
                params,
            )
            if rows:
                return [i for i in rows[0].get("item_ids", []) if i]
        except Exception as exc:  # noqa: BLE001
            _logger.debug("get_requirements fallback failed: %s", exc)
        return []


# ---------------------------------------------------------------------------
# SearchCodeTool
# ---------------------------------------------------------------------------


@traces(spec="107", reqs=["REQ-107-053", "REQ-107-054"])
class SearchCodeTool(BaseTool):
    """Semantic search across the codebase via the Memgraph semantic graph."""

    def __init__(
        self,
        repo: SyncGraphRepository | None = None,
        config: SemanticGraphConfig | None = None,
    ) -> None:
        self._metadata = ToolMetadata(
            name="search_code_semantic",
            description=(
                "Search the codebase for semantically relevant chunks using "
                "the Memgraph semantic graph. Returns a markdown block with "
                "matched chunk snippets and root file references."
            ),
            category=ToolCategory.SEARCH,
            parameters=_SEARCH_CODE_PARAMETERS,
            priority=ToolPriority.P1,
            tags=("semantic", "search", "memgraph"),
        )
        self._lazy_repo = _LazyRepo(repo)
        self._config = config

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

    @traces(spec="107", reqs=["REQ-107-053", "REQ-107-054"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        query = (arguments.get("query") or "").strip()
        if not query:
            return _error(self.name, "query is required")

        target_files = arguments.get("target_files") or []
        if not isinstance(target_files, list) or not all(
            isinstance(f, str) for f in target_files
        ):
            return _error(self.name, "target_files must be a list of strings")

        repo = self._lazy_repo.get()
        project_id = os.environ.get("ORCHESTRATA_PROJECT_ID", "orchestrata-ultimate")
        config = self._config or SemanticGraphConfig.from_env()

        memgraph_driver = None
        try:
            auth = (
                (config.username, config.password)
                if config.username or config.password
                else NO_AUTH
            )
            memgraph_driver = create_graph(uri=config.uri, auth=auth)
            memgraph_driver.verify_connectivity()
        except Exception as exc:  # noqa: BLE001 — degrade, don't crash
            _logger.debug("Memgraph unreachable for search_code_semantic: %s", exc)
            return _error(self.name, f"Memgraph semantic graph unavailable: {exc}")
        finally:
            if memgraph_driver is not None:
                try:
                    memgraph_driver.close()
                except Exception as close_exc:  # pragma: no cover - defensive
                    _logger.debug("Failed to close Memgraph driver: %s", close_exc)

        try:
            context = render_semantic_context(
                user_message=query,
                target_files=target_files,
                subject="implement",
                project_id=project_id,
                config=config,
                graph_repo=repo,
                max_terms=6,
                limit_per_term=5,
                max_lines=30,
            )
            if not context:
                return _ok(
                    self.name,
                    {
                        "query": query,
                        "target_files": target_files,
                        "results": "No relevant semantic chunks found.",
                    },
                )
            return _ok(
                self.name,
                {
                    "query": query,
                    "target_files": target_files,
                    "results": context,
                },
            )
        except Exception as exc:  # noqa: BLE001 — degrade, don't crash
            _logger.warning("search_code_semantic failed for %r: %s", query, exc)
            return _error(self.name, f"Semantic search failed: {exc}")


# ---------------------------------------------------------------------------
# Registration
# ---------------------------------------------------------------------------


@traces(spec="107", reqs=["REQ-107-055"])
def register_context_tools(registry: ToolRegistry) -> None:
    """Register the four on-demand context tools with ``registry``."""
    registry.register(GetCodeTool(), replace=True)
    registry.register(GetTestsTool(), replace=True)
    registry.register(GetRequirementsTool(), replace=True)
    registry.register(SearchCodeTool(), replace=True)