orchestrata_ultimate.knowledge_graph.context_injector

Reference › orchestrata_ultimate › knowledge_graph › context_injector

ContextInjector enriches CXP packets with graph context so a volk knows the existing code structure before it changes or integrates with it (Spec 008 REQ-008-013, Spec 106 REQ-106-007). The context assembler and the kernel's pipeline_kg import it. It reads the operational graph (orchkg on orchgraph) through a synchronous repository and, via semantic_bridge, can add file hits from the shared semantic graph (memgraph).

There are two layers. The packet layer: should_inject checks whether a subject (implement, fix issues, review and test, design review) triggers injection, and inject extracts target files (_extract_targets), builds a GraphContext with _query_context and attaches it to the payload. Query steps are exact file lookup by id and path, a multi-hop traversal capped at 50 entities and roughly 2000 tokens, recent changes, similar patterns (files sharing base classes), the semantic bridge, and the semantic layer of requirements and prior decisions the code traces to (resolved with requirement_resolution). GraphContext.to_prompt renders the sections within a token budget. Called without a task type, inject keeps its original behaviour.

The task-type layer was merged in from a former separate injector: TaskType (build, refactor, debug, document, test, review) drives inject_for_task, which returns a VolkContext of ContextNode entries with to_dict and to_markdown. That path runs the deterministic pipeline in context_pack, and _apply_context_pack adapts the packed result into the node-oriented shape. ContextExpansion carries the broader parameters that context_validator_loop requests on a retry. Project ids come from project_resolver.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

TaskType (enum)

Bases: Enum · Line 87

Types of tasks a Volk can perform (drives the task-type query layer).

ContextExpansion (dataclass)

Decorators: traces(spec='008', story='US-003', reqs=['REQ-008-122']), dataclass · Line 100

Parameters for expanding KG context on a validator-loop retry.

GraphContext (dataclass)

Decorators: traces(spec='008', story='US-008-001', reqs=['REQ-008-102']), dataclass · Line 174

Context extracted from the knowledge graph for a task.

ContextNode (dataclass)

Decorators: dataclass · Line 393

A node injected into Volk context (task-type-aware layer).

VolkContext (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-042', 'REQ-106-043']), dataclass · Line 406

Complete task-type-aware context for a Volk task.

ContextInjector

Decorators: traces(spec='008', story='US-003', reqs=['REQ-008-013']), traces(spec='106', reqs=['REQ-106-007']) · Line 501

Queries the graph and injects graph context into CXP packets.

Functions

Source

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

# mypy: ignore-errors
"""Knowledge Graph Context Injector — enriches CXP packets with graph context.

Queries the knowledge graph for entities related to the target files
in a CXP packet and attaches them as graph_context in the payload. Volks
use this context for awareness of existing code structure when modifying
or integrating with existing code.

Triggers on: implement, fix_issues, review_and_test, design_review subjects.
Query strategy: multi-hop traversal capped at 50 entities and ~2000 tokens.

W4.4: also hosts the task-type-aware query layer (BUILD/REFACTOR/DEBUG/
DOCUMENT/TEST/REVIEW), merged in from the formerly-orphaned
``volk_context_injector.VolkContextInjector`` (ruled D9,
plans/decisions-2026-07-06.md — wire the handlers into the live path rather
than leave them as dead unit-tested code). ``ContextInjector.inject()``
keeps its original behavior byte-for-byte when called without ``task_type``;
passing ``task_type`` additionally runs :meth:`ContextInjector.inject_for_task`
and adds a ``task_context`` key to the payload.

W8.5 (SPEC-106 FR-005, REQ-106-040): :meth:`ContextInjector.inject_for_task`
no longer dispatches to the six ad-hoc ``_inject_*_context`` handlers
(``_inject_build_context`` .. ``_inject_review_context``) or their
``_build_summary`` helper — it now builds a ``RequestFrame``
(``context_pack.analyze_request``), selects a ranked subgraph
(``context_pack.select``), and greedily packs it under a token budget
(``context_pack.pack``), then adapts the result into the same
:class:`VolkContext` shape this method has always returned. Those six
handlers had zero remaining references anywhere in the codebase (confirmed
via ``serialize_codebase_ast.py refs`` before removal) and were deleted
outright rather than kept as dead code (session 2026-07-09).
"""

import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces

# Runtime import (not TYPE_CHECKING): the ``repo:`` annotations must be
# evaluable — PEP 649 lazy evaluation runs them when inspect.signature()
# is called (test_knowledge_signatures), and a guard-only import NameErrors.
from orchestrata_ultimate.core.graph import SyncGraphRepository
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs
from orchestrata_ultimate.memory.token_counter import WordTokenCounter

if TYPE_CHECKING:
    # KgRelevanceRetriever's own module imports context_pack.selector, which
    # imports TaskType from THIS module — a real (non-TYPE_CHECKING) import
    # here would be circular. Kept type-only; the ``relevance:`` parameter
    # below uses a quoted forward reference so PEP 649's lazy annotation
    # evaluation (see the ``repo:`` note above) never needs this name bound
    # at runtime — inspect.signature() only reads parameter names/defaults
    # in this codebase's signature tests, never .annotation.
    from orchestrata_ultimate.knowledge_graph.context_pack.cache import (
        ContextPackCache,
    )
    from orchestrata_ultimate.knowledge_graph.context_pack.relevance import (
        KgRelevanceRetriever,
    )
    from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig

_logger = logging.getLogger(__name__)

# Subjects that trigger context injection
INJECT_SUBJECTS = frozenset(
    {
        "implement",
        "fix_issues",
        "review_and_test",
        "design_review",
        "scope_review",
        "ux_design",
    }
)

MAX_ENTITIES = 50
MAX_CONTEXT_CHARS = 8000

# Default token budget for inject_for_task's selector/packer path (SPEC-106
# FR-005, REQ-106-040). Matches GraphContext.to_prompt's existing default so
# the two context-injection surfaces stay budget-consistent.
DEFAULT_TASK_CONTEXT_BUDGET = 4000


class TaskType(Enum):
    """Types of tasks a Volk can perform (drives the task-type query layer)."""

    BUILD = "build"  # Build from scratch
    REFACTOR = "refactor"  # Refactor existing code
    DEBUG = "debug"  # Debug/repair
    DOCUMENT = "document"  # Generate documentation
    TEST = "test"  # Write tests
    REVIEW = "review"  # Code review


@traces(spec="008", story="US-003", reqs=["REQ-008-122"])
@dataclass
class ContextExpansion:
    """Parameters for expanding KG context on a validator-loop retry.

    A ``None`` expansion keeps the default query shape; any non-default field
    broadens the context injected into the next LLM call.
    """

    extra_hops: int = 0
    include_tested_by: bool = False
    include_requirement_closure: bool = False
    limit_multiplier: int = 1


# Maps a CXP packet ``subject`` (see INJECT_SUBJECTS) to the TaskType the
# task-type-aware layer understands. Wired into the live pipeline call site:
# ``PipelineKGMixin.inject_for_subject`` (kernel/pipeline_kg.py) passes
# ``task_type=SUBJECT_TASK_TYPE.get(subject)`` into ``inject()``, so every
# INJECT_SUBJECTS member (all of which have a TaskType entry here, kept in
# lockstep on purpose) reaches ``inject_for_task``; a subject absent from
# this table keeps the pre-existing graph_context-only path.
SUBJECT_TASK_TYPE: dict[str, TaskType] = {
    "implement": TaskType.BUILD,
    "fix_issues": TaskType.DEBUG,
    "review_and_test": TaskType.TEST,
    "design_review": TaskType.REVIEW,
    "scope_review": TaskType.REVIEW,
    "ux_design": TaskType.BUILD,
}


@traces(spec="008", story="US-003", reqs=["REQ-008-103"])
def _format_change_ts(ts: Any) -> str:
    """Normalize a Neo4j timestamp value to an ISO-ish string.

    Handles ``datetime`` objects, numeric epochs (ms or s), and strings.
    Unknown/invalid values return the empty string so the caller can decide
    whether to keep the record.
    """
    if ts is None:
        return ""
    if hasattr(ts, "isoformat"):
        return ts.isoformat()
    if isinstance(ts, int | float):
        # Neo4j datetime is the common case; raw numbers are treated as ms.
        from datetime import UTC, datetime

        try:
            return datetime.fromtimestamp(ts / 1000, tz=UTC).isoformat()
        except ValueError, OSError, OverflowError:
            return str(ts)
    return str(ts)


def _task_params_text(task_params: dict[str, Any]) -> str:
    """Flatten ``task_params`` values into one entity-linking text blob.

    ``context_pack.analyze_request`` links free TEXT to graph entities, but
    the legacy task-type handlers took structured keys instead (``specs``,
    ``file``, ``package``, ``feature``, ``error``, ``bug``, ...). Rather than
    special-case every legacy key, every value is joined into one blob so
    any embedded canonical id/qualname token is linkable regardless of which
    key carried it (REQ-106-040).
    """
    parts: list[str] = []
    for value in task_params.values():
        if isinstance(value, list | tuple | set):
            parts.extend(str(v) for v in value if v)
        elif value:
            parts.append(str(value))
    return " ".join(parts)


@traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
@dataclass
class GraphContext:
    """Context extracted from the knowledge graph for a task.

    Attributes:
        files: File nodes with path, classes, functions.
        relationships: Import/call/inherit relationships.
        tests: Test functions covering the target entities.
        recent_changes: Recent modifications to target files.
        similar_patterns: Files with similar interfaces.
        semantic_files: Files matched by semantic chunk search.
    """

    files: list[dict[str, Any]] = field(default_factory=list)
    relationships: list[dict[str, Any]] = field(default_factory=list)
    tests: list[str] = field(default_factory=list)
    recent_changes: list[dict[str, Any]] = field(default_factory=list)
    similar_patterns: list[dict[str, Any]] = field(default_factory=list)
    semantic_files: list[dict[str, Any]] = field(default_factory=list)
    # Semantic layer: the "why" behind the code — requirements the target code
    # traces to, and decisions that apply to it. Populated from the KG's
    # TRACES_TO / (:Decision)-[:APPLIES_TO] edges, not just file structure.
    requirements: list[dict[str, Any]] = field(default_factory=list)
    decisions: list[dict[str, Any]] = field(default_factory=list)

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def to_prompt(self, max_context_tokens: int = 4000) -> str:
        """Format context as a text block for volk prompts.

        The output is structured to be unambiguous and force the LLM
        to acknowledge existing code before generating new code.

        Returns:
            Formatted context string capped at MAX_CONTEXT_CHARS.
        """
        parts: list[str] = [
            "# KNOWLEDGE GRAPH CONTEXT — READ BEFORE WRITING CODE\n"
            "The following entities and relationships are ALREADY in the "
            "codebase. You MUST integrate with them, not duplicate them. "
            "Reference specific class/function names from this context "
            "in your response.\n"
        ]

        parts.extend(self._format_files_section())
        parts.extend(self._format_relationships_section())
        parts.extend(self._format_tests_section())
        parts.extend(self._format_recent_changes_section())
        parts.extend(self._format_similar_patterns_section())
        parts.extend(self._format_semantic_files_section())
        parts.extend(self._format_requirements_section())
        parts.extend(self._format_decisions_section())

        result = "\n".join(parts)
        if WordTokenCounter().count(result) > max_context_tokens:
            result = result[: max_context_tokens * 4]  # rough char estimate
        return result

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_files_section(self) -> list[str]:
        """Format the existing-code-structure section (files → classes/funcs)."""
        if not self.files:
            return []
        parts = ["## Existing Code Structure"]
        for f in self.files[:10]:
            path = f.get("path", "?")
            parts.append(f"\n### {path}")
            for cls in f.get("classes", []):
                methods = ", ".join(cls.get("methods", [])[:8])
                bases = cls.get("bases", [])
                base_str = f"({', '.join(bases)})" if bases else ""
                parts.append(f"  class {cls['name']}{base_str}: {methods}")
            for fn in f.get("functions", []):
                parts.append(f"  def {fn['name']}({fn.get('args', '')})")
        return parts

    # Ordered render config for typed relationship groups:
    # (rel_type, heading, limit, line_formatter).
    _REL_GROUPS = (
        (
            cs.REL_INHERITS,
            "\n## Inheritance (class → base class)",
            10,
            lambda rel: f"  {rel['from']} inherits from {rel['to']}",
        ),
        (
            cs.REL_CALLS,
            "\n## Call Graph (caller → callee)",
            15,
            lambda rel: f"  {rel['from']}() calls {rel['to']}()",
        ),
        (
            cs.REL_IMPORTS,
            "\n## Import Dependencies (file → file)",
            10,
            lambda rel: f"  {rel['from']} imports from {rel['to']}",
        ),
        (
            cs.REL_DEFINES,
            "\n## Class Methods (class → method)",
            15,
            lambda rel: f"  {rel['from']}.{rel['to']}()",
        ),
    )

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_relationships_section(self) -> list[str]:
        """Format relationships grouped by type (inherits/calls/imports/...)."""
        if not self.relationships:
            return []

        known = {t for t, _, _, _ in self._REL_GROUPS}
        parts: list[str] = []
        for rel_type, heading, limit, fmt in self._REL_GROUPS:
            group = [r for r in self.relationships if r.get("type") == rel_type]
            if group:
                parts.append(heading)
                parts.extend(fmt(rel) for rel in group[:limit])

        others = [r for r in self.relationships if r.get("type") not in known]
        if others:
            parts.append("\n## Other Relationships")
            parts.extend(
                f"  {rel.get('from', '?')} "
                f"--[{rel.get('type', '?')}]--> "
                f"{rel.get('to', '?')}"
                for rel in others[:10]
            )
        return parts

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_tests_section(self) -> list[str]:
        """Format the covering-tests section."""
        if not self.tests:
            return []
        parts = ["\n## Covering Tests (do NOT break these)"]
        parts.extend(f"  {t}" for t in self.tests[:10])
        return parts

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_recent_changes_section(self) -> list[str]:
        """Format the recent-changes section."""
        if not self.recent_changes:
            return []
        parts = ["\n## Recent Changes"]
        for ch in self.recent_changes[:5]:
            ts = ch.get("timestamp") or ""
            author = ch.get("author") or ""
            tail = ""
            if ts and author:
                tail = f" ({ts} by {author})"
            elif ts:
                tail = f" ({ts})"
            elif author:
                tail = f" (by {author})"
            parts.append(f"  {ch.get('file', '?')}: {ch.get('change', '?')}{tail}")
        return parts

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_similar_patterns_section(self) -> list[str]:
        """Format the similar-patterns section (files sharing bases/protocols)."""
        if not self.similar_patterns:
            return []
        parts = ["\n## Similar Patterns (analogous code to reference)"]
        for sp in self.similar_patterns[:5]:
            path = sp.get("path", "?")
            shared = sp.get("shared_bases", [])
            if shared:
                parts.append(f"  {path} — shares {', '.join(shared[:3])}")
            else:
                parts.append(f"  {path}")
        return parts

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_semantic_files_section(self) -> list[str]:
        """Format the semantic-search files section."""
        if not self.semantic_files:
            return []
        parts = ["\n## Semantically related files"]
        for sf in self.semantic_files[:5]:
            path = sf.get("path", "?")
            snippet = sf.get("snippet", "")
            if snippet:
                parts.append(f"  {path}: {snippet}")
            else:
                parts.append(f"  {path}")
        return parts

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_requirements_section(self) -> list[str]:
        """Format the requirements the target code traces to (the 'why')."""
        if not self.requirements:
            return []
        parts = ["\n## Requirements this code traces to (satisfy them)"]
        for r in self.requirements[:20]:
            line = f"  {r.get('id', '?')}: {r.get('description', '')}"
            specs = r.get("specs")
            if specs:
                line += f" [{', '.join(specs)}]"
            distance = r.get("distance")
            if distance is not None and distance > 0:
                line += f" (distance {distance})"
            parts.append(line.rstrip())
        return parts

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-102"])
    def _format_decisions_section(self) -> list[str]:
        """Format prior decisions that apply to the target code (respect them)."""
        if not self.decisions:
            return []
        parts = ["\n## Prior decisions affecting this code (respect them)"]
        for d in self.decisions[:10]:
            reqs = d.get("requirements") or []
            tail = f" [{', '.join(reqs)}]" if reqs else ""
            parts.append(
                f"  {d.get('id', '?')}: {d.get('rationale', '')}{tail}".rstrip()
            )
        return parts


@dataclass
class ContextNode:
    """A node injected into Volk context (task-type-aware layer)."""

    node_type: str
    label: str
    properties: dict[str, Any]
    relationships: list[dict[str, str]] = field(default_factory=list)
    source_file: str | None = None
    line_number: int | None = None


@traces(spec="106", reqs=["REQ-106-042", "REQ-106-043"])
@dataclass
class VolkContext:
    """Complete task-type-aware context for a Volk task.

    Built by :meth:`ContextInjector.inject_for_task` — the node-oriented
    counterpart to :class:`GraphContext` (which drives the generic
    ``inject()`` prompt). Distinct from GraphContext because it's keyed by
    a ``task_type`` and renders per-node markdown/dict output rather than a
    single prompt block.

    Attributes:
        metadata: SPEC-106 FR-005 selector/packer measurement + degradation
            info (REQ-106-042/REQ-106-043) — e.g. ``{"budget": ..., "spent":
            ..., "section_count": ..., "node_ids": [...]}`` on a normal
            ``inject_for_task`` call, or ``{"reason": ...}`` when degraded
            (e.g. the KG is unavailable). Empty when neither applies.
    """

    volk_alias: str
    task_type: TaskType
    nodes: list[ContextNode] = field(default_factory=list)
    files: list[str] = field(default_factory=list)
    specs: list[str] = field(default_factory=list)
    requirements: list[str] = field(default_factory=list)
    relationships: list[str] = field(default_factory=list)
    summary: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for injection."""
        return {
            "volk": self.volk_alias,
            "task_type": self.task_type.value,
            "files": self.files,
            "specs": self.specs,
            "requirements": self.requirements,
            "relationships": self.relationships,
            "node_count": len(self.nodes),
            "summary": self.summary,
            "nodes": [
                {
                    "type": n.node_type,
                    "label": n.label,
                    "properties": n.properties,
                    "file": n.source_file,
                    "line": n.line_number,
                }
                for n in self.nodes
            ],
        }

    def to_markdown(self) -> str:
        """Convert to markdown for human-readable context."""
        lines = [
            f"# Context for {self.volk_alias}",
            f"## Task: {self.task_type.value}\n",
        ]
        lines.extend(self._markdown_bullet_section("Specifications", self.specs))
        lines.extend(self._markdown_bullet_section("Requirements", self.requirements))
        lines.extend(
            self._markdown_bullet_section("Files", self.files, code_quoted=True)
        )
        lines.extend(self._markdown_nodes_section())
        return "\n".join(lines)

    @staticmethod
    def _markdown_bullet_section(
        heading: str, items: list[str], code_quoted: bool = False
    ) -> list[str]:
        """Render a simple bulleted markdown section, or nothing if empty."""
        if not items:
            return []
        lines = [f"## {heading}"]
        for item in items:
            lines.append(f"- `{item}`" if code_quoted else f"- {item}")
        lines.append("")
        return lines

    def _markdown_nodes_section(self) -> list[str]:
        """Render the Nodes section with per-node properties and relationships."""
        lines = ["## Nodes"]
        for node in self.nodes:
            lines.append(f"### {node.label} ({node.node_type})")
            for key, value in node.properties.items():
                lines.append(f"- **{key}**: `{value}`")
            if node.relationships:
                lines.append("**Relationships:**")
                for rel in node.relationships:
                    lines.append(f"  - `{rel['type']}` → {rel['target']}")
            lines.append("")
        return lines


@traces(spec="008", story="US-003", reqs=["REQ-008-013"])
@traces(spec="106", reqs=["REQ-106-007"])
class ContextInjector:
    """Queries the graph and injects graph context into CXP packets.

    Used by PipelineKernel before routing escalation packets.
    Queries multi-hop relationships around target files and attaches
    the result as payload['graph_context'].

    Also hosts the task-type-aware layer (:meth:`inject_for_task`, merged
    in W4.4 from the former standalone ``VolkContextInjector``) — pass
    ``task_type=`` to :meth:`inject` to additionally populate
    ``payload['task_context']`` with BUILD/REFACTOR/DEBUG/DOCUMENT/TEST/
    REVIEW-specific graph queries.

    Attributes:
        repo: ``SyncGraphRepository`` backing this injector (backlog #163
            phase 5b — was a raw ``neo4j.Driver`` before).
    """

    @traces(spec="106", reqs=["REQ-106-063"])
    def __init__(
        self,
        repo: SyncGraphRepository | None,
        *,
        relevance: "KgRelevanceRetriever | None" = None,  # noqa: UP037
        project_id: str | None = None,
        pack_cache: "ContextPackCache | None" = None,  # noqa: UP037
        semantic_config: "SemanticGraphConfig | None" = None,  # noqa: UP037
    ) -> None:
        """Initialize with a graph repository.

        Args:
            repo: ``SyncGraphRepository`` (backlog #163). If None, injection
                is disabled.
            relevance: Optional SPEC-106 FR-006 symbolic relevance retriever.
                When ``None`` (the default — every pre-existing call site),
                :meth:`inject_for_task` is byte-identical to the FR-003-only
                path (REQ-106-063). When wired, it refines the selected
                subgraph's ranking with Prolog-deduced relevance tiers.
            project_id: Optional project scope for canonical ``File`` lookups.
                When set, :meth:`_query_context` resolves target paths against
                ``f.path`` exactly and filters by ``f.project_id``; when
                omitted, lookups are project-agnostic.
            pack_cache: Optional cache for rendered context packs. When set,
                repeated calls with the same snapshot_id/task/volk/subject/
                payload-fingerprint reuse the rendered markdown.
            semantic_config: Optional Memgraph semantic-graph config for the
                semantic-search bridge. When omitted, the bridge uses
                ``SemanticGraphConfig.from_env()``.
        """
        self._repo = repo
        self._available = repo is not None
        self._relevance = relevance
        self._project_id = project_id
        self._pack_cache = pack_cache
        self._semantic_config = semantic_config

    @property
    def _connected_repo(self) -> SyncGraphRepository:
        """Type-narrowing accessor for the query helpers below.

        Only reached after a ``self._available`` check already confirmed
        ``self._repo`` is not None, so this never actually raises in
        practice — it exists so pyright can see a non-Optional repository
        without re-guarding every query method (mirrors
        ``GraphSyncer._connected_repo``).
        """
        if self._repo is None:
            raise RuntimeError("Not connected — no graph repository.")
        return self._repo

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def should_inject(self, subject: str) -> bool:
        """Check if a CXP packet subject triggers injection.

        Args:
            subject: CXP packet subject string.

        Returns:
            True if context is injected.
        """
        return self._available and subject in INJECT_SUBJECTS

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def inject(
        self,
        payload: dict[str, Any],
        *,
        task_type: TaskType | None = None,
        task_params: dict[str, Any] | None = None,
        volk_alias: str = "",
        expansion: ContextExpansion | None = None,
    ) -> dict[str, Any]:
        """Query the graph and add context to a CXP packet payload.

        Extracts target files from the payload, queries the graph for
        related entities, and adds a 'graph_context' key. When ``task_type``
        is supplied, ALSO runs the task-type-aware layer (see
        :meth:`inject_for_task`) and adds a 'task_context' key. Callers that
        never pass ``task_type`` (the pre-W4.4 default) see zero behavioral
        change.

        Args:
            payload: The CXP packet payload dict.
            task_type: Optional task category (BUILD/REFACTOR/DEBUG/
                DOCUMENT/TEST/REVIEW) driving the task-type-specific query
                layer. ``None`` (default) skips it entirely.
            task_params: Parameters for the task-type layer (e.g. ``specs``,
                ``file``, ``package``, ``feature``, ``error``); defaults to
                ``payload`` itself when omitted.
            volk_alias: The volk this task context is being built for
                (cosmetic — rendered into the task context output).
            expansion: Optional expansion controls for the validator feedback
                loop (larger limits, extra hops, TESTED_BY neighbours).

        Returns:
            The payload with graph_context (and, when ``task_type`` is
            given, task_context) added.
        """
        if not self._available:
            return payload

        target_files = self._extract_targets(payload)
        if target_files:
            try:
                query = payload.get("text") or payload.get("intent") or ""
                context = self._query_context(
                    target_files, query=query, expansion=expansion
                )
                payload["graph_context"] = context.to_prompt()
                _logger.debug(
                    "Injected graph context: %d files, %d rels, %d tests",
                    len(context.files),
                    len(context.relationships),
                    len(context.tests),
                )
            except Exception as exc:
                _logger.warning("Context injection failed: %s", exc)
                payload["graph_context"] = ""
        elif task_type is None:
            return payload

        if task_type is not None:
            try:
                volk_context = self.inject_for_task(
                    volk_alias=volk_alias,
                    task_type=task_type,
                    task_params=task_params if task_params is not None else payload,
                )
                payload["task_context"] = volk_context.to_dict()
            except Exception as exc:
                _logger.warning("Task-type context injection failed: %s", exc)

        return payload

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def _extract_targets(
        self,
        payload: dict[str, Any],
    ) -> list[str]:
        """Extract target file paths from a packet payload.

        Args:
            payload: CXP packet payload.

        Returns:
            List of file path strings to query context for.
        """
        targets: list[str] = []

        # From files_created (QA, CodeDirector payloads)
        files_created = payload.get("files_created", [])
        targets.extend(files_created)

        # From project_dir (scan for .py files)
        project_dir = payload.get("project_dir", "")
        if project_dir:
            import os

            for root, dirs, files in os.walk(project_dir):
                dirs[:] = [d for d in dirs if d not in {".venv", "__pycache__", ".git"}]
                for f in files:
                    if f.endswith(".py") and not f.startswith("test_"):
                        targets.append(
                            os.path.relpath(os.path.join(root, f), project_dir)
                        )
                if len(targets) > 30:
                    break

        # From file_tree (Architect payloads)
        file_tree = payload.get("file_tree", "")
        if file_tree:
            import re

            for m in re.finditer(r"(\S+\.py)", file_tree):
                fname = m.group(1).strip("`'\"")
                if fname not in targets:
                    targets.append(fname)

        return targets[:30]

    @traces(spec="008", story="US-003", reqs=["REQ-008-005"])
    def _query_context(
        self,
        target_files: list[str],
        query: str = "",
        expansion: ContextExpansion | None = None,
    ) -> GraphContext:
        """Query the graph for context around target files.

        Performs multi-hop traversal:
        1. Find File nodes matching targets
        2. Get classes and functions defined in those files
        3. Get relationships (imports, calls, inherits)
        4. Get tests covering those functions
        5. Get recent changes
        6. Get similar patterns
        7. Get semantically related files (when ``query`` is non-empty)

        Args:
            target_files: File paths to query context for.
            query: Optional user/intent text for semantic chunk search.
            expansion: Optional expansion controls for broader context.

        Returns:
            GraphContext with aggregated results.
        """
        context = GraphContext()
        repo = self._connected_repo
        expansion = expansion or ContextExpansion()
        limit_mult = max(1, expansion.limit_multiplier)

        # 1. File entities with classes (including methods and bases) and functions.
        #    Resolve by canonical File.id first (project-qualified), then by
        #    exact path for legacy nodes. Project scoping is applied whenever
        #    the injector was configured with a project_id.
        candidate_paths = target_files[: 10 * limit_mult]
        self._query_exact_files(candidate_paths, context, limit_mult=limit_mult)

        if not context.files:
            return context

        # 2. Relationships (imports, calls, inherits, has_method)
        file_paths = [f["path"] for f in context.files if f.get("path")]
        if file_paths:
            result = repo.execute_read(
                "MATCH (f:File)-[:DEFINES]->(src) "
                "WHERE f.path IN $paths "
                "OPTIONAL MATCH (src)-[r:IMPORTS|CALLS|INHERITS|DEFINES]->(tgt) "
                "WHERE r IS NOT NULL "
                "RETURN src.name AS from_name, "
                "       type(r) AS rel_type, "
                "       tgt.name AS to_name, "
                "       labels(tgt)[0] AS to_type "
                f"LIMIT {50 * limit_mult}",
                {"paths": file_paths},
            )
            for record in result:
                if record["rel_type"]:
                    context.relationships.append(
                        {
                            "from": record["from_name"],
                            "type": record["rel_type"],
                            "to": record["to_name"],
                            "to_type": record.get("to_type", ""),
                        }
                    )

            # 2b. Extra relationship hops for validator-loop expansion.
            if expansion.extra_hops > 0:
                result = repo.execute_read(
                    "MATCH (f:File)-[:DEFINES]->(src) "
                    "WHERE f.path IN $paths "
                    "OPTIONAL MATCH (src)-[:IMPORTS|CALLS|INHERITS|DEFINES*1.."
                    f"{expansion.extra_hops + 1}]->(tgt) "
                    "RETURN DISTINCT src.name AS from_name, "
                    "       'reaches' AS rel_type, "
                    "       tgt.name AS to_name, "
                    "       labels(tgt)[0] AS to_type "
                    f"LIMIT {30 * limit_mult}",
                    {"paths": file_paths},
                )
                for record in result:
                    if record["to_name"] and record["to_name"] not in {
                        r["to"] for r in context.relationships
                    }:
                        context.relationships.append(
                            {
                                "from": record["from_name"],
                                "type": record["rel_type"],
                                "to": record["to_name"],
                                "to_type": record.get("to_type", ""),
                            }
                        )

        # 3. Covering tests
        if file_paths:
            result = repo.execute_read(
                "MATCH (f:File)-[:DEFINES]->(fn:Function) "
                "WHERE f.path IN $paths "
                "OPTIONAL MATCH (t:TestResult)-[:COVERS]->(fn) "
                "WHERE t IS NOT NULL "
                "RETURN t.name AS test_name "
                f"LIMIT {20 * limit_mult}",
                {"paths": file_paths},
            )
            for record in result:
                if record["test_name"]:
                    context.tests.append(record["test_name"])

            # 3b. TESTED_BY neighbours for validator-loop expansion.
            if expansion.include_tested_by:
                result = repo.execute_read(
                    "MATCH (f:File)-[:DEFINES]->(fn:Function) "
                    "WHERE f.path IN $paths "
                    "OPTIONAL MATCH (fn)<-[:TESTED_BY]-(t:Test) "
                    "WHERE t IS NOT NULL "
                    "RETURN t.name AS test_name "
                    f"LIMIT {20 * limit_mult}",
                    {"paths": file_paths},
                )
                for record in result:
                    if record["test_name"]:
                        context.tests.append(record["test_name"])

        # 4. Recent changes — target files and direct neighbours.
        self._query_recent_changes(file_paths, context)

        # 5. Similar patterns — files whose classes share bases/protocols.
        self._query_similar_patterns(file_paths, context)

        # 6. Semantic files — Memgraph chunks mapped to operational-KG File roots.
        if query:
            self._query_semantic_bridge(file_paths, query, context)

        # 7. Semantic layer — the "why": requirements the target code traces
        #    to, and decisions applying to it (not just code structure).
        self._query_semantic_context(
            file_paths,
            context,
            max_depth=3 if expansion.include_requirement_closure else 2,
        )

        return context

    @traces(spec="008", story="US-003", reqs=["REQ-008-005"])
    def _query_exact_files(
        self,
        target_files: list[str],
        context: GraphContext,
        limit_mult: int = 1,
    ) -> set[str]:
        """Query ``File`` nodes by canonical ``f.id`` and exact ``f.path``.

        When ``project_id`` is set, the canonical id has the shape
        ``"{project_id}:{path}"`` (per :data:`CANONICAL_MERGE_KEYS`). The query
        also falls back to exact ``f.path`` matches within the same project so
        legacy File nodes remain resolvable. Returns the set of matched file
        paths; found files are appended to ``context.files``.
        """
        repo = self._connected_repo
        or_clauses: list[str] = []
        params: dict[str, Any] = {}
        if self._project_id is not None:
            or_clauses.append("f.id IN $ids")
            params["ids"] = [f"{self._project_id}:{p}" for p in target_files]
        or_clauses.append("f.path IN $paths")
        params["paths"] = target_files
        where_clause = "(" + " OR ".join(or_clauses) + ")"
        if self._project_id is not None:
            where_clause += " AND f.project_id = $project_id"
            params["project_id"] = self._project_id
        result = repo.execute_read(
            "MATCH (f:File) "
            f"WHERE {where_clause} "
            "OPTIONAL MATCH (f)-[:DEFINES]->(c:Class) "
            "OPTIONAL MATCH (f)-[:DEFINES]->(fn:Function) "
            "WHERE NOT EXISTS { MATCH (:Class)-[:DEFINES]->(fn) } "
            "RETURN f.path AS path, f.line_count AS lines, "
            "       collect(DISTINCT {name: c.name, methods: c.methods, "
            "                         bases: c.base_classes}) AS classes, "
            "       collect(DISTINCT {name: fn.name, "
            "                         args: fn.signature}) AS functions "
            f"LIMIT {50 * limit_mult}",
            params,
        )
        matched: set[str] = set()
        for record in result:
            path = record["path"]
            if path:
                matched.add(path)
            self._append_file_record(context, record)
        return matched

    @traces(spec="008", story="US-003", reqs=["REQ-008-005"])
    def _append_file_record(
        self, context: GraphContext, record: dict[str, Any]
    ) -> None:
        """Append a single Cypher file record to ``context.files``.

        Avoids duplicating a file already present in the context.
        """
        path = record.get("path")
        if any(f.get("path") == path for f in context.files):
            return
        context.files.append(
            {
                "path": path,
                "lines": record["lines"],
                "classes": [
                    {
                        "name": c.get("name", ""),
                        "methods": c.get("methods", []) or [],
                        "bases": c.get("bases", []) or [],
                    }
                    for c in (record["classes"] or [])
                    if c.get("name")
                ],
                "functions": [
                    {
                        "name": fn.get("name", ""),
                        "args": fn.get("args", ""),
                    }
                    for fn in (record["functions"] or [])
                    if fn.get("name")
                ],
            }
        )

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def _query_recent_changes(
        self,
        file_paths: list[str],
        context: GraphContext,
    ) -> None:
        """Populate ``context.recent_changes`` for target files and neighbours.

        Tries three sources in order and stops at the first one that yields
        results:

        1. ``(:File)-[:CHANGED_IN]->(:Change)`` — explicit change records.
        2. ``(:File)-[:NEXT_VERSION]->(:File)`` — version-chain diff metadata.
        3. File node properties ``updated_at`` / ``synced_at`` — last sync.

        Direct neighbours (files linked via imports/calls/inherits through a
        shared symbol) are included so volks see changes in code they depend
        on. Project scoping is applied when the injector was configured with a
        ``project_id``. Empty results leave the section omitted.
        """
        if not file_paths:
            return

        from orchestrata_ultimate.knowledge_graph.canonical_schema import (
            REL_CHANGED_IN,
            REL_NEXT_VERSION,
        )

        repo = self._connected_repo
        paths = file_paths[:10]

        def _project_filter(prefix: str = "f") -> tuple[str, dict[str, Any]]:
            extra: dict[str, Any] = {}
            if self._project_id is None:
                return "", extra
            return f" AND {prefix}.project_id = $project_id", {
                "project_id": self._project_id
            }

        def _neighbour_clause() -> str:
            return (
                "OPTIONAL MATCH (f)-[:DEFINES]->(sym) "
                "OPTIONAL MATCH (sym)-[:IMPORTS|CALLS|INHERITS]->(other) "
                "OPTIONAL MATCH (other)<-[:DEFINES]-(nf:File) "
                "WITH collect(DISTINCT f) + collect(DISTINCT nf) AS files "
                "UNWIND files AS f "
            )

        def _run_query(cypher: str, params: dict[str, Any]) -> list[dict[str, Any]]:
            try:
                return repo.execute_read(cypher, params)
            except Exception as exc:
                _logger.warning("Recent-changes query failed: %s", exc)
                return []

        # ── Source 1: explicit Change nodes ─────────────────────────────────
        pf, pf_params = _project_filter()
        params = {"paths": paths, **pf_params}
        result = _run_query(
            "MATCH (f:File) WHERE f.path IN $paths "
            + pf
            + " "
            + _neighbour_clause()
            + f"OPTIONAL MATCH (f)-[:{REL_CHANGED_IN}]->(ch:Change) "
            "WITH f, ch WHERE ch IS NOT NULL "
            "RETURN f.path AS path, "
            "       coalesce(ch.description, ch.change_type, 'modified') AS change, "
            "       coalesce(ch.timestamp, ch.created_at) AS ts, "
            "       coalesce(ch.author, 'unknown') AS author "
            "ORDER BY ts DESC LIMIT 5",
            params,
        )
        if result:
            self._append_recent_changes(result, context)
            return

        # ── Source 2: NEXT_VERSION chain ────────────────────────────────────
        result = _run_query(
            "MATCH (f:File) WHERE f.path IN $paths "
            + pf
            + " "
            + _neighbour_clause()
            + f"OPTIONAL MATCH (f)-[:{REL_NEXT_VERSION}]->(next:File) "
            "WITH f, next WHERE next IS NOT NULL "
            "RETURN f.path AS path, "
            "       coalesce(next.change_summary, 'newer version available') AS change, "
            "       coalesce(next.created_at, next.updated_at, next.synced_at) AS ts, "
            "       coalesce(next.author, 'unknown') AS author "
            "ORDER BY ts DESC LIMIT 5",
            dict(params),
        )
        if result:
            self._append_recent_changes(result, context)
            return

        # ── Source 3: file timestamp properties ─────────────────────────────
        result = _run_query(
            "MATCH (f:File) WHERE f.path IN $paths "
            + pf
            + " "
            + _neighbour_clause()
            + "RETURN f.path AS path, "
            "       'file updated' AS change, "
            "       coalesce(f.updated_at, f.synced_at) AS ts, "
            "       coalesce(f.author, 'unknown') AS author "
            "ORDER BY ts DESC LIMIT 5",
            dict(params),
        )
        if result:
            self._append_recent_changes(result, context)

    @traces(spec="008", story="US-003", reqs=["REQ-008-103"])
    def _append_recent_changes(
        self,
        records: list[dict[str, Any]],
        context: GraphContext,
    ) -> None:
        """Append non-empty recent-change records to ``context``.

        Filters out records with no usable timestamp and deduplicates by
        file path, keeping the most recent entry per file.
        """
        seen: set[str] = set()
        for record in records:
            path = record.get("path")
            ts = record.get("ts")
            if not path or ts is None:
                continue
            if path in seen:
                continue
            seen.add(path)
            context.recent_changes.append(
                {
                    "file": path,
                    "change": record.get("change") or "modified",
                    "timestamp": _format_change_ts(ts),
                    "author": record.get("author") or "unknown",
                }
            )

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def _query_similar_patterns(
        self,
        file_paths: list[str],
        context: GraphContext,
    ) -> None:
        """Populate ``context.similar_patterns`` with files whose classes share
        base classes or implemented protocols with the target files.

        The query gathers base/protocol names from the target files' classes
        (via ``base_classes`` property and ``INHERITS``/``IMPLEMENTS``
        relationships), then finds other ``File`` nodes whose classes share at
        least one of those names. Results are ranked by the number of shared
        bases/protocols and then by path similarity, capped at 5. Target files
        themselves are excluded. Project scoping is applied when configured.
        """
        if not file_paths:
            return

        from orchestrata_ultimate.knowledge_graph.canonical_schema import (
            REL_IMPLEMENTS,
            REL_INHERITS,
        )

        repo = self._connected_repo
        paths = file_paths[:10]

        pf: str = ""
        params: dict[str, Any] = {"paths": paths}
        if self._project_id is not None:
            pf = " AND f.project_id = $project_id"
            params["project_id"] = self._project_id

        # Collect base/protocol names from target files.
        result = repo.execute_read(
            "MATCH (f:File)-[:DEFINES]->(c:Class) "
            "WHERE f.path IN $paths " + pf + " "
            "RETURN collect(DISTINCT c.base_classes) AS base_lists, "
            "       collect(DISTINCT c.name) AS class_names",
            dict(params),
        )
        if not result:
            return

        bases: set[str] = set()
        for lst in result[0].get("base_lists", []) or []:
            if lst:
                bases.update(str(b) for b in lst if b)

        # Also gather explicit INHERITS/IMPLEMENTS targets.
        result = repo.execute_read(
            "MATCH (f:File)-[:DEFINES]->(c:Class)-[r:"
            + REL_INHERITS
            + "|"
            + REL_IMPLEMENTS
            + "]->(base) "
            "WHERE f.path IN $paths " + pf + " "
            "RETURN collect(DISTINCT base.name) AS explicit_bases",
            dict(params),
        )
        if result:
            for name in result[0].get("explicit_bases", []) or []:
                if name:
                    bases.add(name)

        if not bases:
            return

        # Find other files sharing those bases/protocols via explicit edges.
        sim_params = {"bases": sorted(bases), **params}
        other_pf = ""
        if self._project_id is not None:
            other_pf = " AND other.project_id = $project_id"
        edge_query = (
            "MATCH (other:File)-[:DEFINES]->(c:Class)-[:"
            + REL_INHERITS
            + "|"
            + REL_IMPLEMENTS
            + "]->(base) "
            "WHERE NOT other.path IN $paths " + other_pf + " AND base.name IN $bases "
            "RETURN other.path AS path, collect(DISTINCT base.name) AS shared_bases, "
            "       count(DISTINCT base) AS shared_count "
            # ORDER BY the RETURNED ALIAS, not `other.path`: after an
            # aggregation orchgraph/Memgraph only keeps the projected names
            # in scope and answers "Unbound variable: other" (verified live
            # 2026-07-26). Neo4j tolerates either, so the alias is portable.
            "ORDER BY shared_count DESC, path "
            "LIMIT 5"
        )
        edge_result = repo.execute_read(edge_query, sim_params)

        # Also include files whose Class.base_classes property overlaps.
        prop_query = (
            "MATCH (other:File)-[:DEFINES]->(c:Class) "
            "WHERE NOT other.path IN $paths " + other_pf + " "
            # coalesce, because a Class with no `base_classes` property makes
            # this a `UNWIND null`: Neo4j quietly yields zero rows, orchgraph/
            # Memgraph raises "Argument of UNWIND must be a list, but 'null'
            # was provided" and kills the whole similar-patterns section
            # (verified live 2026-07-26). An empty list is the same answer on
            # both engines.
            "UNWIND coalesce(c.base_classes, []) AS base "
            "WITH other, base "
            "WHERE base IN $bases "
            "RETURN other.path AS path, collect(DISTINCT base) AS shared_bases, "
            "       count(DISTINCT base) AS shared_count "
            # ORDER BY the RETURNED ALIAS, not `other.path`: after an
            # aggregation orchgraph/Memgraph only keeps the projected names
            # in scope and answers "Unbound variable: other" (verified live
            # 2026-07-26). Neo4j tolerates either, so the alias is portable.
            "ORDER BY shared_count DESC, path "
            "LIMIT 5"
        )
        prop_result = repo.execute_read(prop_query, sim_params)

        # Merge and dedupe by path, preferring higher shared_count.
        merged: dict[str, dict[str, Any]] = {}
        for record in list(edge_result) + list(prop_result):
            path = record.get("path")
            if not path or path in paths:
                continue
            shared = set(record.get("shared_bases", []) or [])
            if path in merged:
                merged[path]["shared_bases"].update(shared)
            else:
                merged[path] = {"path": path, "shared_bases": shared}

        if not merged:
            return

        sorted_patterns = sorted(
            merged.values(),
            key=lambda x: (-len(x["shared_bases"]), x["path"]),
        )[:5]
        for sp in sorted_patterns:
            context.similar_patterns.append(
                {
                    "path": sp["path"],
                    "shared_bases": sorted(sp["shared_bases"])[:3],
                }
            )

    @traces(spec="107", reqs=["REQ-107-015"])
    def _query_semantic_bridge(
        self,
        file_paths: list[str],
        query: str,
        context: GraphContext,
    ) -> None:
        """Populate ``context.semantic_files`` via the semantic-search bridge.

        Queries Memgraph for chunks matching the user intent, maps their
        ``file_path`` values back to operational-KG ``File`` roots, and
        appends the matched files (with a snippet) to ``context.semantic_files``.
        Empty results or unreachable backends leave the section omitted.
        """
        if not file_paths or not query:
            return

        from orchestrata_ultimate.knowledge_graph.semantic_bridge import (
            search_semantic_roots,
        )

        try:
            roots = search_semantic_roots(
                query=query,
                target_files=file_paths,
                project_id=self._project_id,
                config=self._semantic_config,
                graph_repo=self._repo,
                max_terms=6,
                limit_per_term=5,
                max_chunks_per_file=1,
            )
            for root in roots[:5]:
                context.semantic_files.append(
                    {
                        "path": root.path,
                        "snippet": root.snippet(),
                        "line_count": root.line_count,
                    }
                )
        except Exception as exc:
            _logger.debug("Semantic bridge query failed: %s", exc)

    @traces(spec="008", story="US-003", reqs=["REQ-008-106", "REQ-008-107"])
    def _query_semantic_context(
        self,
        file_paths: list[str],
        context: GraphContext,
        max_depth: int = 2,
    ) -> None:
        """Add the KG semantic layer to ``context``: requirements the target
        code traces to (``(sym)-[:TRACES_TO]->(:RequirementItem)`` — the
        as-built label the KG sync writer actually MERGEs, per
        ``session/kg_sync/_stage_specs_traces.py``) and decisions that apply
        to it (``(:Decision)-[:APPLIES_TO]->(file|symbol)``).

        When ``self._project_id`` is set, all queries are scoped to that
        project so requirements/decisions from other projects do not leak
        into the volk context.

        Matched by symbol *name* (robust to the Class/CodeClass label
        divergence) and file path. Best-effort — divergent id/description keys
        are coalesced, and an empty result simply omits the section.

        Args:
            max_depth: Maximum depth for requirement closure traversal.
        """
        repo = self._connected_repo
        symbol_names = [
            name
            for f in context.files
            for name in (
                [c.get("name") for c in f.get("classes", [])]
                + [fn.get("name") for fn in f.get("functions", [])]
            )
            if name
        ]

        if symbol_names:
            from orchestrata_ultimate.knowledge_graph.canonical_schema import (
                REL_TRACES_TO,
            )
            from orchestrata_ultimate.knowledge_graph.requirement_resolution import (
                closure_requirement_chain,
            )

            project_filter = ""
            params: dict[str, Any] = {"names": symbol_names}
            if self._project_id is not None:
                project_filter = " AND ri.project_id = $project_id"
                params["project_id"] = self._project_id

            result = repo.execute_read(
                f"MATCH (sym)-[:{REL_TRACES_TO}]->(ri:RequirementItem) "
                "WHERE sym.name IN $names " + project_filter + " "
                "RETURN collect(DISTINCT ri.id) AS item_ids",
                params,
            )
            item_ids = (
                [i for i in (result[0].get("item_ids") or []) if i] if result else []
            )
            for chain in closure_requirement_chain(
                item_ids, repo, project_id=self._project_id, max_depth=max_depth
            ):
                req: dict[str, Any] = {
                    "id": chain.id,
                    "description": chain.description,
                    "distance": chain.distance,
                }
                if chain.specs:
                    req["specs"] = chain.specs
                context.requirements.append(req)

        if file_paths or symbol_names:
            dec_params: dict[str, Any] = {"paths": file_paths, "names": symbol_names}
            dec_filter = ""
            if self._project_id is not None:
                # Decisions apply to files/symbols that carry the project scope.
                dec_filter = " AND x.project_id = $project_id"
                dec_params["project_id"] = self._project_id
            result = repo.execute_read(
                "MATCH (d:Decision)-[:APPLIES_TO]->(x) "
                "WHERE (x.path IN $paths OR x.name IN $names) " + dec_filter + " "
                "OPTIONAL MATCH (d)-[:JUSTIFIED_BY]->(req:Requirement) "
                "RETURN DISTINCT coalesce(d.id, d.decision_id) AS id, "
                "       coalesce(d.rationale, d.summary, d.title) AS rationale, "
                "       collect(DISTINCT coalesce(req.req_id, req.id)) AS reqs "
                "LIMIT 10",
                dec_params,
            )
            for record in result:
                if record["id"]:
                    context.decisions.append(
                        {
                            "id": record["id"],
                            "rationale": record["rationale"] or "",
                            "requirements": [r for r in (record["reqs"] or []) if r],
                        }
                    )

    # ── Task-type-aware layer (W4.4 merge of former VolkContextInjector;
    # W8.5 re-wire onto the SPEC-106 selector/packer) ──────────────────────
    # The six _inject_*_context handlers below are no longer called from
    # inject_for_task (superseded by the canonical selector/packer) but are
    # left in place, untouched — other tests/coverage still reference them
    # directly, and retiring them is a later cleanup decision, not this
    # wire-up's.

    @traces(
        spec="106",
        reqs=[
            "REQ-106-040",
            "REQ-106-042",
            "REQ-106-043",
            "REQ-106-063",
            "REQ-106-065",
        ],
    )
    def inject_for_task(
        self,
        volk_alias: str,
        task_type: TaskType,
        task_params: dict[str, Any],
    ) -> VolkContext:
        """Build task-type-aware context for a Volk task.

        SPEC-106 FR-005 (REQ-106-040): builds a deterministic ``RequestFrame``
        from ``task_params`` (``context_pack.analyze_request``), selects a
        ranked, task-type-templated subgraph (``context_pack.select``), and
        greedily packs it under a token budget (``context_pack.pack``) — no
        LLM/embedding anywhere in selection (D11). The pack is then adapted
        into this method's original return shape by node label: Module/
        Class/Function sections also contribute their source file to
        ``context.files``; Spec sections to ``context.specs``; Requirement/
        RequirementItem sections to ``context.requirements``; every section
        becomes a :class:`ContextNode`. The rendered pack markdown
        (``context_pack.render``) becomes ``context.summary``.

        Also records one measurement (REQ-106-042) — task type, budget,
        spent tokens, section count, and selected node ids — as both a
        structured log line and ``context.metadata``.

        Args:
            volk_alias: The volk this context is being built for (cosmetic —
                used only in the rendered markdown/dict output and as the
                frame's ``requester``).
            task_type: Which of the six task categories to select for.
            task_params: Task-type-specific parameters (e.g. ``specs``,
                ``file``, ``package``, ``feature``, ``error``) — every value
                is flattened into the frame's entity-linking text
                (:func:`_task_params_text`).

        Returns:
            VolkContext with nodes/files/specs/requirements/relationships
            populated from the packed subgraph, plus the rendered pack as
            ``summary``. Empty (but valid), with a reason in ``metadata``,
            when the graph is unavailable (REQ-106-043).
        """
        from orchestrata_ultimate.knowledge_graph.context_pack import (
            analyze_request,
            render,
            select,
        )
        from orchestrata_ultimate.knowledge_graph.context_pack import (
            pack as pack_subgraph,
        )
        from orchestrata_ultimate.knowledge_graph.context_pack.relevance import (
            TIER_CORE,
            TIER_PERIPHERAL,
            TIER_RELATED,
            rescore_by_tier,
        )

        context = VolkContext(volk_alias=volk_alias, task_type=task_type)
        if not self._available:
            context.metadata = {"reason": "Graph unavailable"}
            return context

        repo = self._connected_repo
        frame = analyze_request(
            {"text": _task_params_text(task_params)},
            repo=repo,
            task_type=task_type,
            budget=DEFAULT_TASK_CONTEXT_BUDGET,
            requester=volk_alias or "agent",
        )
        subgraph = select(frame, repo)

        # SPEC-106 FR-006 (REQ-106-063): strictly opt-in. With no retriever
        # wired, this branch never runs and `subgraph` is untouched — the
        # FR-003-only ranking stays byte-identical to the pre-wiring path.
        relevance_tiers: dict[str, str] | None = None
        if self._relevance is not None:
            relevance_tiers = self._relevance.refine(subgraph)
            subgraph = rescore_by_tier(subgraph, relevance_tiers)

        packed = pack_subgraph(subgraph, frame)

        self._apply_context_pack(context, subgraph, packed)
        cached_summary = (
            self._pack_cache.get(packed, task_type=task_type, volk_alias=volk_alias)
            if self._pack_cache is not None
            else None
        )
        if cached_summary is not None:
            context.summary = cached_summary
            _logger.debug("Context pack cache hit: %s", packed.snapshot_id)
        else:
            rendered = render(packed)
            context.summary = rendered
            if self._pack_cache is not None:
                self._pack_cache.put(
                    rendered,
                    packed,
                    task_type=task_type,
                    volk_alias=volk_alias,
                )
        context.metadata = {
            "task_type": task_type.value,
            "budget": packed.budget,
            "spent": packed.spent,
            "section_count": len(packed.sections),
            "node_ids": [section.node_id for section in packed.sections],
        }
        if relevance_tiers is not None:
            # REQ-106-065: per-tier node counts observable alongside the
            # FR-005 measurement record.
            context.metadata["relevance_tiers"] = {
                TIER_CORE: sum(1 for t in relevance_tiers.values() if t == TIER_CORE),
                TIER_RELATED: sum(
                    1 for t in relevance_tiers.values() if t == TIER_RELATED
                ),
                TIER_PERIPHERAL: sum(
                    1 for t in relevance_tiers.values() if t == TIER_PERIPHERAL
                ),
            }
        _logger.info(
            "inject_for_task measurement: task_type=%s budget=%d spent=%d "
            "sections=%d node_ids=%s",
            task_type.value,
            packed.budget,
            packed.spent,
            len(packed.sections),
            context.metadata["node_ids"],
        )
        return context

    @traces(spec="106", reqs=["REQ-106-040"])
    def _apply_context_pack(
        self,
        context: VolkContext,
        subgraph: Any,
        packed: Any,
    ) -> None:
        """Adapt a packed ``ContextPack`` into ``context``'s node-oriented shape.

        Routes each packed section into VolkContext's fields BY THE SELECTED
        NODE'S CANONICAL LABEL: Module/Class/Function sections also
        contribute their source file (the relpath half of the canonical
        ``<relpath>::<qualname>`` id) to ``context.files``; Spec sections to
        ``context.specs``; Requirement/RequirementItem sections to
        ``context.requirements``. ``context.relationships`` records the two
        per-node graph SIGNALS the selector surfaces (REQ-106-022) —
        TESTED_BY presence and TRACES_TO degree — since the canonical
        selector (unlike the six retired ad-hoc handlers) never returns Test
        nodes themselves: TESTED_BY is ranking-only, not an expansion edge.

        Args:
            context: The VolkContext being built; mutated in place.
            subgraph: The ``RankedSubgraph`` :func:`select` produced (for the
                per-node ranking signals — properties, tested_by, trace
                degree — that a packed ``PackSection`` alone doesn't carry).
            packed: The ``ContextPack`` :func:`pack` produced.
        """
        from orchestrata_ultimate.knowledge_graph.canonical_schema import (
            LABEL_CLASS,
            LABEL_FUNCTION,
            LABEL_MODULE,
            LABEL_REQUIREMENT,
            LABEL_REQUIREMENT_ITEM,
            LABEL_SPEC,
            parse_code_entity_id,
        )

        code_labels = {LABEL_MODULE, LABEL_CLASS, LABEL_FUNCTION}
        ranked_by_id = {node.node_id: node for node in subgraph.nodes}
        seen_files: set[str] = set()

        for section in packed.sections:
            ranked = ranked_by_id.get(section.node_id)
            properties = dict(ranked.properties) if ranked is not None else {}
            context.nodes.append(
                ContextNode(
                    node_type=section.label,
                    label=section.node_id,
                    properties=properties,
                )
            )
            if section.label in code_labels:
                try:
                    relpath, _ = parse_code_entity_id(section.node_id)
                except ValueError:
                    relpath = section.node_id
                if relpath not in seen_files:
                    seen_files.add(relpath)
                    context.files.append(relpath)
            elif section.label == LABEL_SPEC:
                context.specs.append(section.node_id)
            elif section.label in (LABEL_REQUIREMENT, LABEL_REQUIREMENT_ITEM):
                context.requirements.append(section.node_id)

            if ranked is not None:
                if ranked.tested_by:
                    context.relationships.append(f"{section.node_id} TESTED_BY")
                if ranked.trace_degree > 0:
                    context.relationships.append(
                        f"{section.node_id} TRACES_TO x{ranked.trace_degree}"
                    )