orchestrata_ultimate.kernel.pipeline_kg

Reference › orchestrata_ultimate › kernel › pipeline_kg

The knowledge-graph mixin of PipelineRunner. It does two jobs: it writes what the pipeline does into the operational graph, and it reads the graph back to enrich the payloads that travel between volks. pipeline mixes it in, and the dynamic router in the orchestrator package (orchestrator.pipeline_runner) also imports it. It serves SPEC-008 (REQ-008-006, -013, -023, -024, -041). Every graph operation is best-effort: when the graph is unreachable the mixin returns 0 or False and never raises into the pipeline.

Writing side:

• sync_intent_to_kg merges an :Intent node and the :PipelineRun it triggered, keyed by thread id and anchored to a :Thread node under the :Project, so the graph can be traversed from a human intent to its run.
• sync_volks_to_kg merges one :Volk node per registered volk with an edge from the project.
• sync_components_to_kg does the same for components.
• sync_artifacts_to_kg merges an :Artifact node per pipeline artifact with a generated-by edge to the producing volk, plus ownership edges when the artifact's file resolves to a component in the source tree.
• sync_decisions_to_kg records volk decisions.
• _write_message_signal and _write_tool_call_signal write message and tool-call nodes through the runtime signal writer. The message signal is written by pipeline_escalation around each hop.

Reading side:

• _init_context_injector connects to the graph, resolves the project id (from ORCHESTRATA_PROJECT_ID when set, otherwise from the graph, falling back to the directory name) and builds a ContextInjector, a RuntimeSignalWriter and a UnifiedContextAssembler. If the connection fails, it builds injectors without a graph and prints that injection is unavailable. _kg_available_now redials opportunistically.
• inject_for_subject takes a payload and a CXP subject; when the injector says the subject warrants it, it runs _pre_task_sync (a file sync, skippable with ORCHESTRATA_KG_PRESTAGE_SYNC), then assembles memory, artifacts, mission and task-type context under one budget, and adds graph_context to the payload. If assembly fails, it falls back to the legacy injector. _maybe_inject_context is the packet-shaped wrapper that the escalation loop calls before each hop.

At a glance

Class diagram

Dependency diagram

API

Classes

PipelineKGMixin

Line 78

Knowledge graph helpers for PipelineRunner.

Functions

Source

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

"""PipelineRunner knowledge graph mixin.

Extracted from pipeline.py to keep PipelineRunner under the
300-line class limit. Contains graph context injection helpers.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, cast

from orchestrata_ultimate.context.assembler import (
    UnifiedContextAssembler,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import SyncGraphRepository
from orchestrata_ultimate.core.infra_config import GRAPH_PASSWORD, GRAPH_URI, GRAPH_USER
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    LABEL_PROJECT,
    LABEL_THREAD,
    LABEL_VOLK,
    REL_HAS_THREAD,
    REL_HAS_VOLK,
    REL_IN_THREAD,
    thread_entity_id,
)
from orchestrata_ultimate.knowledge_graph.context_injector import (
    SUBJECT_TASK_TYPE,
    ContextInjector,
)
from orchestrata_ultimate.knowledge_graph.context_pack.relevance import (
    KgRelevanceRetriever,
)
from orchestrata_ultimate.knowledge_graph.project_resolver import (
    derive_project_id,
    resolve_project_id,
)
from orchestrata_ultimate.knowledge_graph.runtime_writer import RuntimeSignalWriter

if TYPE_CHECKING:
    from orchestrata_ultimate.session.models import ContextData

_logger = logging.getLogger(__name__)

# Marker that anchors a file path to the project's own source tree so
# ownership derivation never guesses at paths outside it (used by
# sync_artifacts_to_kg's Component-ownership rows, gap #2).
_SRC_PACKAGE_MARKER = "src/orchestrata_ultimate/"


@traces(spec="008", story="US-005", reqs=["REQ-008-006"])
def _component_from_source_path(file_path: str) -> str | None:
    """Derive the top-level package (Component) name from a source file path.

    Only resolves paths under this project's own source tree
    (``src/orchestrata_ultimate/<package>/...``) — returns ``None`` for
    anything else (including a bare top-level module with no package
    directory, e.g. ``src/orchestrata_ultimate/cli.py``) so callers never
    fabricate a Component for an unresolvable or out-of-tree path.
    """
    normalized = file_path.replace("\\", "/")
    marker_idx = normalized.find(_SRC_PACKAGE_MARKER)
    if marker_idx == -1:
        return None
    rest = normalized[marker_idx + len(_SRC_PACKAGE_MARKER) :]
    parts = [p for p in rest.split("/") if p]
    if not parts or parts[0].endswith(".py"):
        return None
    return parts[0]


class PipelineKGMixin:
    """Knowledge graph helpers for PipelineRunner.

    Expects ``self.context_injector`` to be set by ``_init_context_injector``.

    Attributes:
        _graph_repo: The ``SyncGraphRepository`` this mixin owns (backlog
            #163 phase 5) — constructed once by ``_init_context_injector``
            and shared with ``self.context_injector`` and, per call, with
            ``FileSynchronizer``. ``None`` when the graph is unavailable.
    """

    context_injector: Any
    _graph_repo: SyncGraphRepository | None = None
    # Resolved project identifier, set by _init_context_injector. Prefer this
    # over re-deriving from _project_dir everywhere project scope is needed.
    _project_id: str | None = None
    # Declared here (not assigned) as a structural contract: PipelineRunner
    # sets this in its own __init__. It is this mixin's only project source
    # (2026-07-12 multi-project identity migration) — see
    # sync_decisions_to_kg's project-qualification of Requirement ids.
    _project_dir: str | None = None
    # Throttled lazy re-dial state (backlog #219): a transient graph blip at
    # boot (`_init_context_injector` failing) used to permanently disable
    # every KG write for the process lifetime. `_last_kg_dial` starts at
    # +inf so hosts that construct `_graph_repo`/`context_injector` directly
    # (test harnesses bypassing `_init_context_injector`) never trigger an
    # unsolicited redial; it is only set to a real timestamp once
    # `_init_context_injector` actually runs (success or failure), after
    # which `_kg_available_now()` retries at most once per interval.
    _last_kg_dial: float = float("inf")
    _kg_redial_interval: float = 30.0

    @traces(spec="008", story="US-005", reqs=["REQ-008-024"])
    def _get_graph_client(self) -> Any:
        """Return a minimal config object for FileSynchronizer.

        The synchronizer expects client._config.uri/username/password.

        Returns:
            Minimal graph client config object.
        """

        class _Cfg:
            uri = GRAPH_URI
            username = GRAPH_USER
            password = GRAPH_PASSWORD

        class _Client:
            _config = _Cfg()

        return _Client()

    @traces(spec="008", story="US-003", reqs=["REQ-008-013", "REQ-008-041"])
    def _init_context_injector(self, env: Mapping[str, str] | None = None) -> None:
        """Initialize the knowledge graph context injector and assembler.

        Connects to the graph for graph-aware context injection into
        CXP packets. Non-fatal — injection disabled if the graph is unavailable.
        Also wires the ``UnifiedContextAssembler`` so production runners
        feed memory, artifacts, mission, and task-type context into volk
        prompts through one budgeted bundle.

        The project identifier is resolved against the KG first; only when
        the graph is unreachable do we fall back to the directory basename.
        """
        try:
            from orchestrata_ultimate.core.graph import (
                GraphRepositoryConfig,
                create_graph_repository,
            )

            repo = create_graph_repository(
                GraphRepositoryConfig(uri=GRAPH_URI, auth=(GRAPH_USER, GRAPH_PASSWORD))
            )
            repo.verify_connectivity()
            self._graph_repo = repo
            # ORCHESTRATA_PROJECT_ID is the highest-precedence source in
            # resolve_project_id's documented order, and until now nothing
            # passed it — `explicit_project_id` had zero callers, so step 1 of
            # that order was unreachable. It is needed here: this checkout's
            # basename ("orchestrata-ultimate-architecture") is not the
            # project's slug ("orchestrata-ultimate"), and the node lookup
            # cannot bridge them because it matches p.path/p.slug/p.id and the
            # live node carries a null path, a null id, and a slug that
            # differs from the basename. Unset leaves resolution exactly as it
            # was for every other project.
            project_id = resolve_project_id(
                self._project_dir,
                repo,
                explicit_project_id=(env if env is not None else os.environ).get(
                    "ORCHESTRATA_PROJECT_ID"
                ),
            )
            self._project_id = project_id
            relevance = self._try_create_relevance_retriever(repo)
            self.context_injector = ContextInjector(
                repo, relevance=relevance, project_id=project_id
            )
            self._runtime_writer = RuntimeSignalWriter(repo)
            self._context_assembler = UnifiedContextAssembler(
                injector=self.context_injector, project_id=project_id
            )
            print("  [KG Context] Injector connected", flush=True)
        except Exception as exc:
            self._graph_repo = None
            # Same source as the connected branch: an explicitly configured id
            # is a statement about identity, not about graph reachability, so
            # a graph-down boot must not silently fall back to the basename.
            project_id = (env if env is not None else os.environ).get(
                "ORCHESTRATA_PROJECT_ID"
            ) or derive_project_id(self._project_dir)
            self._project_id = project_id
            self.context_injector = ContextInjector(None, project_id=project_id)
            self._runtime_writer = RuntimeSignalWriter(None)
            self._context_assembler = UnifiedContextAssembler(
                injector=self.context_injector, project_id=project_id
            )
            print(
                f"  [KG Context] Unavailable ({exc}) — no injection",
                flush=True,
            )
        self._last_kg_dial = time.monotonic()

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def _kg_available_now(self) -> bool:
        """KG availability, with a throttled lazy re-dial so a transient
        graph blip at boot is recovered on next use (backlog #219).

        Re-attempting a full ``_init_context_injector()`` (rather than just
        re-verifying ``_graph_repo``) is deliberate: the injector, runtime
        writer, and context assembler are all constructed from the same
        repository and would otherwise keep pointing at the dead one even
        after a successful redial.
        """
        if self._graph_repo is not None:
            return True
        now = time.monotonic()
        if now - self._last_kg_dial < self._kg_redial_interval:
            return False
        self._init_context_injector()
        return self._graph_repo is not None

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def _try_create_relevance_retriever(
        self, repo: SyncGraphRepository
    ) -> KgRelevanceRetriever | None:
        """Create a Prolog-backed relevance retriever when Prolog is available.

        Falls back to None if the Prolog worker is unreachable so that KG
        context injection still works without symbolic relevance refinement.
        """
        try:
            from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

            bridge = PrologBridge()
            # Health-check the Prolog worker before wiring the retriever.
            health = bridge.query("true")
            if health.error or not health.success:
                return None
            return KgRelevanceRetriever(bridge, repo)
        except Exception as exc:
            _logger.debug("Prolog relevance retriever unavailable: %s", exc)
            return None

    @traces(spec="008", story="US-005", reqs=["REQ-008-024"])
    def _sync_to_graph(self, project_dir: str) -> None:
        """Sync generated code to the graph knowledge base.

        Parses all Python files in the project and creates/updates
        nodes (File, Class, Function) and relationships (IMPORTS,
        DEFINES) in the graph.

        Args:
            project_dir: Path to the generated project directory.
        """
        try:
            from orchestrata_ultimate.knowledge_graph.synchronizer import (
                FileSynchronizer,
            )

            # self._graph_repo is the ONE repository this mixin owns (backlog
            # #163 phase 5) -> pass it to FileSynchronizer's `repo=` param.
            # Passing it as `graph_client=` leaves self._repo=None, so
            # _acquire_sync_repo falls through to `getattr(client, "_config")`,
            # raises "graph client has no _config", the bare except swallows
            # it, and the sync writes ZERO nodes.
            self._kg_available_now()  # opportunistic redial (backlog #219)
            project_id = self._project_id or derive_project_id(project_dir)
            syncer = FileSynchronizer(repo=self._graph_repo, project_id=project_id)
            results = syncer.sync_project(project_dir)
            synced = sum(1 for r in results if r.status.value == "completed")
            total_nodes = sum(r.nodes_created for r in results)
            print(
                f"  [KG Sync] {synced} files → {total_nodes} nodes synced",
                flush=True,
            )
        except Exception as exc:
            _logger.warning("KG sync failed: %s", exc)
            print(f"  [KG Sync] Failed (non-fatal): {exc}", flush=True)

    def sync_volks_to_kg(self) -> int:
        """MERGE a ``:Volk`` node per registered volk into the graph (alias
        identity), so the W5 semantic layer is queryable with live roster data
        (``QueryEngine.volk_work_context``) instead of an empty graph.

        Also MERGEs the ``(:Project)-[:HAS_VOLK]->(:Volk)`` hub edge (session-254
        connectivity ruling) so every synced Volk node hangs off the root
        Project node instead of forming its own disconnected island.

        Best-effort: returns 0 and never raises when the graph is unavailable,
        the runner exposes no status report, or no volks are registered.
        """
        if not self._kg_available_now():
            return 0
        repo = self._graph_repo
        if repo is None:
            return 0
        status_report = getattr(self, "status_report", None)
        if not callable(status_report):
            return 0
        try:
            # getattr + callable() types the call result as bare `object`;
            # the roster contract is a list of per-volk dicts.
            report = cast("list[dict[str, Any]] | None", status_report())
        except Exception:  # roster read is auxiliary — never break the caller
            return 0
        rows = [
            {
                "alias": r.get("alias", ""),
                "role": r.get("role", "") or "",
                "status": str(r.get("status", "") or ""),
                "summary": r.get("summary", "") or "",
            }
            for r in (report or [])
            if r.get("alias")
        ]
        if not rows:
            return 0
        # Same project-slug resolution as sync_components_to_kg, falling back
        # to the ORCHESTRATA_PROJECT_ID / "orchestrata-ultimate" default the
        # runtime writers use (tools/context_tools.py) when no project
        # directory has been resolved (e.g. a bare boot() in-process).
        project_id = (
            self._project_id
            or derive_project_id(self._project_dir)
            or os.environ.get("ORCHESTRATA_PROJECT_ID", "orchestrata-ultimate")
        )
        cypher = (
            "UNWIND $rows AS row "
            f"MERGE (v:{LABEL_VOLK} {{alias: row.alias}}) "
            "SET v.role = row.role, v.status = row.status, "
            "v.summary = row.summary, v.updated_at = timestamp() "
            "WITH v "
            f"MERGE (p:{LABEL_PROJECT} {{slug: $project_id}}) "
            f"MERGE (p)-[:{REL_HAS_VOLK}]->(v)"
        )
        try:
            repo.execute_write(cypher, {"rows": rows, "project_id": project_id})
        except Exception as exc:
            _logger.warning("Volk KG sync failed (non-fatal): %s", exc)
            return 0
        return len(rows)

    @traces(spec="008", story="US-005", reqs=["REQ-008-006"])
    def sync_components_to_kg(self) -> int:
        """MERGE a ``:Component`` node per top-level source package, derived
        from the code-structure ``:Package`` nodes the AST synchronizer already
        wrote (``depth: 0``), plus ``(:Component)-[:PART_OF]->(:Project)``.

        This is what makes the ``(:Volk)-[:OWNS]->(:Component)`` traversal
        (``QueryEngine.volk_work_context``/``component_work_context``) resolve
        against real component identities instead of an always-empty graph
        (SPEC-008 FR-002 gap #2).

        Component identity is the canonical, project-qualified ``id =
        "{project}:{name}"`` (ratified 2026-07-19, plans/architecture-
        ontology-draft.md §9.2 ruling 2; ``canonical_schema.CANONICAL_MERGE_KEYS
        [LABEL_COMPONENT] == "id"``) — Package/File already follow this
        multi-project invariant, and a bare-``name`` key would otherwise
        collide the same top-level package name across projects. At depth 0,
        a Package's own ``id`` is already exactly ``"{project}:{name}"``, so
        it is reused verbatim as the Component id; ``c.name``/``c.project``
        are kept as the bare display properties.

        Best-effort: returns 0 and never raises when the graph is unavailable,
        no project is resolved, or no top-level packages are synced yet.
        """
        if not self._kg_available_now():
            return 0
        repo = self._graph_repo
        if repo is None:
            return 0
        project_id = self._project_id or derive_project_id(self._project_dir)
        if not project_id:
            return 0
        cypher = (
            "MATCH (pkg:Package {project: $project_id, depth: 0}) "
            "MERGE (c:Component {id: pkg.id}) "
            "SET c.name = pkg.name, c.project = $project_id, c.path = pkg.path, "
            "c.type = 'package', c.language = 'python', c.updated_at = timestamp() "
            "WITH c "
            # slug is the canonical Project key (CANONICAL_MERGE_KEYS) and the
            # only key BOTH writer families set: the AST-sync writer merges on
            # id but also sets slug, while the session/design writers are
            # slug-keyed with no id at all.
            "MATCH (proj:Project {slug: $project_id}) "
            "MERGE (c)-[:PART_OF]->(proj) "
            "RETURN count(c) AS cnt"
        )
        try:
            result = repo.execute_write(cypher, {"project_id": project_id})
        except Exception as exc:
            _logger.warning("Component KG sync failed (non-fatal): %s", exc)
            return 0
        return result[0]["cnt"] if result else 0

    @traces(spec="008", story="US-005", reqs=["REQ-008-006", "REQ-008-023"])
    def sync_artifacts_to_kg(self, artifacts: list[Any] | None = None) -> int:
        """MERGE an ``:Artifact`` node per pipeline artifact plus a
        ``(:Artifact)-[:GENERATED_BY]->(:Volk)`` provenance edge to its producing
        volk (the artifact's ``stage`` alias).

        This is what makes ``QueryEngine.volk_work_context`` return real generated
        artifacts: the semantic labels/edges are declared in the schema but only
        become queryable once the runtime writes them here.

        Also writes authorship-derived ownership — ``(:Volk)-[:OWNS]->(:Component)``
        + ``(:Component)-[:OWNED_BY]->(:Volk)`` — for artifacts whose ``file_path``
        resolves under this project's source tree (gap #2), so a volk that has
        touched a package accrues real ``OWNS`` provenance over time.

        Args:
            artifacts: The artifacts to sync; when omitted, the whole
                ``ArtifactStore`` (``self.artifacts.all()``) is replayed.

        Best-effort: returns 0 and never raises when the graph is unavailable or
        there are no artifacts. The ``GENERATED_BY``/``OWNS`` edges are written only when
        ``stage`` is a registered volk alias, so non-volk stage names never
        fabricate a ``:Volk`` node; ownership is written only when ``file_path``
        resolves to a component under the source tree — an out-of-tree or empty
        path never fabricates a ``:Component``.

        Session-254 root-connectivity ruling: the ``GENERATED_BY`` edge above
        is guarded on ``stage`` being a registered volk alias, so an Artifact
        whose stage is NOT a volk alias was left with no edge at all (2 live
        orphans, both carrying a ``thread_id``). The Thread anchor below is
        unconditional — every Artifact with a non-empty ``thread_id`` gets
        ``IN_THREAD``/``HAS_THREAD`` regardless of whether its stage
        resolves to a volk.
        """
        if not self._kg_available_now():
            return 0
        repo = self._graph_repo
        if repo is None:
            return 0
        if artifacts is None:
            store = getattr(self, "artifacts", None)
            artifacts = store.all() if store is not None else []
        known = set(getattr(self, "volks", {}) or {})
        project_id = (
            self._project_id or derive_project_id(self._project_dir) or "unscoped"
        )
        rows = []
        ownership_rows = []
        for art in artifacts or []:
            stage = getattr(art, "stage", "") or ""
            content = getattr(art, "content", "") or ""
            art_type = getattr(art, "type", "") or ""
            thread_id = getattr(art, "thread_id", "") or ""
            file_path = getattr(art, "file_path", None) or ""
            digest = hashlib.sha1(
                f"{thread_id}|{stage}|{art_type}|{content}".encode()
            ).hexdigest()[:16]
            alias = stage if stage in known else ""
            rows.append(
                {
                    "id": f"{thread_id or 'no-thread'}:{stage or 'stage'}:{art_type}:{digest}",
                    "type": art_type,
                    "stage": stage,
                    "format": getattr(art, "format", "") or "",
                    "thread_id": thread_id,
                    "thread_qid": (
                        thread_entity_id(project_id, thread_id) if thread_id else ""
                    ),
                    "file_path": file_path,
                    "alias": alias,
                    "run_id": f"run:{thread_id}" if thread_id else "",
                }
            )
            component = _component_from_source_path(file_path) if file_path else None
            if alias and component:
                ownership_rows.append(
                    {
                        "alias": alias,
                        "component_id": f"{project_id}:{component}",
                        "component_name": component,
                        "project_id": project_id,
                    }
                )
        if not rows:
            return 0
        cypher = (
            "UNWIND $rows AS row "
            "MERGE (a:Artifact {id: row.id}) "
            "SET a.type = row.type, a.stage = row.stage, a.format = row.format, "
            "a.thread_id = row.thread_id, a.file_path = row.file_path, "
            "a.updated_at = timestamp() "
            "FOREACH (_ignore IN CASE WHEN row.alias <> '' THEN [1] ELSE [] END | "
            "  MERGE (v:Volk {alias: row.alias}) "
            "  MERGE (a)-[:GENERATED_BY]->(v) "
            "  FOREACH (_r IN CASE WHEN row.run_id <> '' THEN [1] ELSE [] END | "
            "    MERGE (r:PipelineRun {id: row.run_id}) "
            "    MERGE (r)-[:EXECUTED_BY]->(v))) "
            "FOREACH (_t IN CASE WHEN row.thread_qid <> '' THEN [1] ELSE [] END | "
            f"  MERGE (th:{LABEL_THREAD} {{id: row.thread_qid}}) "
            "  ON CREATE SET th.thread_id = row.thread_id, th.project = $project_id "
            f"  MERGE (a)-[:{REL_IN_THREAD}]->(th) "
            f"  MERGE (proj:{LABEL_PROJECT} {{slug: $project_id}}) "
            f"  MERGE (proj)-[:{REL_HAS_THREAD}]->(th))"
        )
        # Component id is the canonical, project-qualified key (ratified
        # 2026-07-19, plans/architecture-ontology-draft.md §9.2 ruling 2) —
        # id/name/project mirror sync_components_to_kg's shape so a component
        # first created here (before sync_components_to_kg ever runs) still
        # conforms.
        ownership_cypher = (
            "UNWIND $rows AS row "
            "MERGE (v:Volk {alias: row.alias}) "
            "MERGE (c:Component {id: row.component_id}) "
            "SET c.name = row.component_name, c.project = row.project_id "
            "MERGE (v)-[:OWNS]->(c) "
            "MERGE (c)-[:OWNED_BY]->(v)"
        )
        try:
            with repo.unit_of_work() as uow:
                uow.run(cypher, {"rows": rows, "project_id": project_id})
                if ownership_rows:
                    uow.run(ownership_cypher, {"rows": ownership_rows})
        except Exception as exc:
            _logger.warning("Artifact KG sync failed (non-fatal): %s", exc)
            return 0
        return len(rows)

    @traces(spec="008", story="US-001", reqs=["REQ-008-006", "REQ-008-002"])
    def sync_intent_to_kg(self, thread_id: str, text: str) -> bool:
        """MERGE the ``:Intent`` node and the ``:PipelineRun`` it triggered
        (``(:Intent)-[:TRIGGERED]->(:PipelineRun)``), both keyed by ``thread_id``.

        This is the entry node of the provenance chain: once the run's stages
        write artifacts (``sync_artifacts_to_kg`` also adds
        ``(:PipelineRun)-[:EXECUTED_BY]->(:Volk)``), the graph is traversable from
        a human intent → its run → the volks that executed → their artifacts.

        Best-effort: returns False and never raises when the graph is unavailable
        or ``thread_id`` is empty. Idempotent — re-submitting the same thread updates
        text in place and preserves the original ``created_at``/``started_at``.

        Session-254 root-connectivity ruling: the Intent/PipelineRun pair used
        to link only to each other, with no path back to the ``:Project``
        hub (187 of each found unreachable live). Every one carries a real
        ``thread_id``, so this now also MERGEs the ``:Thread`` correlation
        anchor (``sync_volks_to_kg``'s ``HAS_VOLK`` hub-MERGE is the template
        this mirrors for ``HAS_THREAD``) and anchors both nodes onto it via
        ``IN_THREAD``.
        """
        if not self._kg_available_now() or not thread_id:
            return False
        repo = self._graph_repo
        if repo is None:
            return False
        project_id = (
            self._project_id
            or derive_project_id(self._project_dir)
            or os.environ.get("ORCHESTRATA_PROJECT_ID", "orchestrata-ultimate")
        )
        thread_qid = thread_entity_id(project_id, thread_id)
        cypher = (
            "MERGE (i:Intent {id: $intent_id}) "
            "SET i.text = $text, i.thread_id = $thread_id, "
            "i.created_at = coalesce(i.created_at, timestamp()) "
            "MERGE (r:PipelineRun {id: $run_id}) "
            "SET r.thread_id = $thread_id, "
            "r.started_at = coalesce(r.started_at, timestamp()) "
            "MERGE (i)-[:TRIGGERED]->(r) "
            "WITH i, r "
            f"MERGE (t:{LABEL_THREAD} {{id: $thread_qid}}) "
            "ON CREATE SET t.thread_id = $thread_id, t.project = $project_id "
            f"MERGE (i)-[:{REL_IN_THREAD}]->(t) "
            f"MERGE (r)-[:{REL_IN_THREAD}]->(t) "
            "WITH t "
            f"MERGE (p:{LABEL_PROJECT} {{slug: $project_id}}) "
            f"MERGE (p)-[:{REL_HAS_THREAD}]->(t)"
        )
        try:
            repo.execute_write(
                cypher,
                {
                    "intent_id": f"intent:{thread_id}",
                    "run_id": f"run:{thread_id}",
                    "text": text or "",
                    "thread_id": thread_id,
                    "thread_qid": thread_qid,
                    "project_id": project_id,
                },
            )
        except Exception as exc:
            _logger.warning("Intent KG sync failed (non-fatal): %s", exc)
            return False
        return True

    @traces(spec="008", story="US-005", reqs=["REQ-008-006", "REQ-008-023"])
    def sync_decisions_to_kg(
        self, volk_alias: str, decisions: list[dict[str, Any]]
    ) -> int:
        """MERGE ``(:Volk)-[:MADE]->(:Decision)`` for each decision a volk made,
        linking the decision to the files it applies to
        (``(:Decision)-[:APPLIES_TO]->(:File)``) and the requirements that justify
        it (``(:Decision)-[:JUSTIFIED_BY]->(:Requirement)``).

        This is what makes ``volk_work_context.decisions`` and the context
        injector's decision section real. ``APPLIES_TO`` links only to *existing*
        ``:File`` nodes (matched by path suffix, since a volk reports relative
        paths while the graph holds absolute ones) — it never fabricates files.

        Args:
            volk_alias: The deciding volk (the ``MADE`` source).
            decisions: dicts of ``{rationale, files, requirements}``.

        Best-effort: returns 0 and never raises when the graph is unavailable,
        the alias is empty, or no decision carries a rationale.
        """
        if not self._kg_available_now() or not volk_alias or not decisions:
            return 0
        repo = self._graph_repo
        if repo is None:
            return 0
        # Project source: prefer the resolved project_id set by
        # _init_context_injector; fall back to deriving from _project_dir or
        # the documented "unscoped" bucket so Requirement ids stay
        # project-qualified (2026-07-12 multi-project ruling).
        project_id = (
            self._project_id or derive_project_id(self._project_dir) or "unscoped"
        )
        rows: list[dict[str, Any]] = []
        links: list[dict[str, Any]] = []
        for dec in decisions:
            rationale = (dec.get("rationale") or "").strip()
            if not rationale:
                continue
            files = [f for f in (dec.get("files") or []) if f]
            # Requirement id is project-qualified ("{project_id}:{tag}") to
            # converge with sync.py's own @traces-derived Requirement nodes
            # (same bare-tag identity space, e.g. "REQ-008-023") rather than
            # colliding across projects on the bare tag (2026-07-12
            # multi-project ruling). req_id keeps the bare tag for display —
            # matching context_injector's coalesce(req.req_id, req.id) and
            # sync.py's own Requirement writer's property name.
            reqs = [
                {"id": f"{project_id}:{r}", "req_id": r}
                for r in (dec.get("requirements") or [])
                if r
            ]
            digest = hashlib.sha1(
                f"{volk_alias}|{rationale}|{'|'.join(files)}".encode()
            ).hexdigest()[:16]
            dec_id = f"dec:{volk_alias}:{digest}"
            rows.append(
                {
                    "id": dec_id,
                    "rationale": rationale,
                    "volk": volk_alias,
                    "files": files,
                    "requirements": reqs,
                }
            )
            links.extend({"id": dec_id, "fp": fp} for fp in files)
        if not rows:
            return 0
        node_cypher = (
            "UNWIND $rows AS row "
            "MERGE (d:Decision {id: row.id}) "
            "SET d.rationale = row.rationale, d.volk = row.volk, "
            "d.files = row.files, d.updated_at = timestamp() "
            "MERGE (v:Volk {alias: row.volk}) "
            "MERGE (v)-[:MADE]->(d) "
            "FOREACH (rq IN row.requirements | "
            "  MERGE (r:Requirement {id: rq.id}) "
            "  SET r.req_id = rq.req_id "
            "  MERGE (d)-[:JUSTIFIED_BY]->(r))"
        )
        # This File match is scoped by project. It used to be deliberately
        # unscoped, on the grounds that FileSynchronizer "MERGEs File by bare
        # `path` (no `project_id` property)" so scoping would zero out every
        # APPLIES_TO edge. Both halves of that are false in the current tree:
        # `FileSynchronizer._persist_to_graph` MERGEs `File {id: $id}` with
        # `_file_node_id` -> f"{project_id}:{file_path}" (the canonical
        # qualified id), and it populates `FileNode.project_id`. The W1 wave
        # that comment deferred to has landed on the writer side.
        #
        # Without the scope this is a suffix match over every File in the
        # graph, so a Decision in one project attaches to another project's
        # identically-named file — `src/cli.py` exists nearly everywhere.
        # `_sync_to_graph` builds the synchronizer from the same project_id
        # expression used here, so the written and matched values share a
        # source.
        link_cypher = (
            "UNWIND $links AS link "
            "MATCH (d:Decision {id: link.id}) "
            "MATCH (f:File {project_id: $project_id}) "
            "WHERE f.path = link.fp OR f.path ENDS WITH ('/' + link.fp) "
            "MERGE (d)-[:APPLIES_TO]->(f)"
        )
        try:
            # Both statements share one transaction (unit_of_work) — same
            # one-transaction guarantee as the pre-seam single session block.
            with repo.unit_of_work() as uow:
                uow.run(node_cypher, {"rows": rows})
                if links:
                    uow.run(link_cypher, {"links": links, "project_id": project_id})
        except Exception as exc:
            _logger.warning("Decision KG sync failed (non-fatal): %s", exc)
            return 0

        # Tier-H signal: also emit each decision through RuntimeSignalWriter so
        # Prolog relevance rules can boost recently decided files/entities.
        writer = getattr(self, "_runtime_writer", None)
        if writer is not None:
            for row in rows:
                try:
                    writer.write_decision(
                        decision_id=row["id"],
                        volk_alias=row["volk"],
                        rationale=row["rationale"],
                        requirements=[r["req_id"] for r in row.get("requirements", [])],
                        files=row.get("files", []),
                        entities=[],
                        project_id=self._project_id,
                    )
                except Exception as exc:
                    _logger.debug("Runtime decision signal write failed: %s", exc)

        return len(rows)

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def inject_for_subject(
        self,
        payload: dict[str, Any],
        subject: str,
        *,
        thread_id: str = "",
        artifact_store: Any | None = None,
    ) -> dict[str, Any]:
        """Payload-level context injection (W3.1 extraction).

        Shared by both call shapes that need it: the CXP-packet escalation
        path (:meth:`_maybe_inject_context`, below) and the orchestrator's
        dynamic-router StageSpec/StageResult path
        (``orchestrator.pipeline_runner.PipelineRunner._run_stage``, which
        builds its outbound packet directly and has no separate escalation
        hop to hook into). Pulled out so both engines share one
        implementation instead of forking the sync + injection logic.

        When a ``UnifiedContextAssembler`` is wired (the production default),
        injection runs through the assembler so memory, artifacts, mission,
        and task-type context are folded into the prompt under a single
        token budget. On any assembly failure it degrades to the legacy
        ``ContextInjector.inject()`` path.

        Args:
            payload: The stage payload to potentially enrich.
            subject: The CXP subject the payload is being routed under.
            thread_id: Pipeline thread id for memory recall and artifact
                filtering.
            artifact_store: Optional artifact store for prior-stage digest;
                defaults to ``self.artifacts`` when available.

        Returns:
            ``payload`` unchanged when the subject doesn't warrant injection
            (or the injector is unavailable); otherwise the same dict with
            ``graph_context`` (and ``task_context`` when the subject maps to
            a TaskType) added.
        """
        self._kg_available_now()  # opportunistic redial (backlog #219)
        if not self.context_injector.should_inject(subject):
            return payload
        project_dir = payload.get("project_dir", "")
        if project_dir and os.path.isdir(project_dir):
            self._pre_task_sync(project_dir)

        assembler = getattr(self, "_context_assembler", None)
        if assembler is not None:
            try:
                result = self._assemble_context_for_subject(
                    payload,
                    subject,
                    assembler,
                    thread_id=thread_id,
                    artifact_store=artifact_store,
                )
                self._write_context_injection_event(payload, subject, thread_id)
                return result
            except Exception as exc:
                _logger.warning(
                    "Unified context assembly failed for subject %r: %s — "
                    "falling back to ContextInjector",
                    subject,
                    exc,
                )

        result = cast(
            "dict[str, Any]",
            self.context_injector.inject(
                payload,
                task_type=SUBJECT_TASK_TYPE.get(subject),
            ),
        )
        self._write_context_injection_event(payload, subject, thread_id)
        return result

    @traces(spec="008", story="US-003", reqs=["REQ-008-115"])
    def _write_context_injection_event(
        self,
        payload: dict[str, Any],
        subject: str,
        thread_id: str,
    ) -> None:
        """Emit a Tier-H GatewayEvent so DEBUG relevance can boost these nodes.

        Best-effort: if the graph is unavailable the writer degrades silently.
        """
        self._kg_available_now()  # opportunistic redial (backlog #219)
        writer = getattr(self, "_runtime_writer", None)
        if writer is None:
            return
        target_files = payload.get("files_created", []) or []
        if not target_files:
            return
        try:
            writer.write_gateway_event(
                entry_id=f"ctx:{thread_id or 'unknown'}:{subject}:{hashlib.sha256(','.join(target_files).encode()).hexdigest()[:16]}",
                thread_id=thread_id,
                event_type="context_injection",
                outcome="success",
                summary=f"Context injected for subject {subject}",
                mentioned_files=target_files[:10],
                project_id=getattr(self, "_project_id", None),
            )
        except Exception as exc:
            _logger.debug("Context injection event write failed: %s", exc)

    def _write_message_signal(
        self,
        packet: dict[str, Any],
        *,
        sender: str = "",
        recipient: str = "",
        subject: str = "",
        thread_id: str = "",
        direction: str = "",
    ) -> None:
        """Emit a Tier-H Message node for a dispatched or received CXP packet.

        Best-effort: if the graph is unavailable the writer degrades silently.
        """
        self._kg_available_now()  # opportunistic redial (backlog #219)
        writer = getattr(self, "_runtime_writer", None)
        if writer is None:
            return
        try:
            from orchestrata_ultimate.knowledge_graph.signal_extraction import (
                extract_entity_ids,
                extract_file_paths,
            )

            message_id = f"msg:{thread_id or 'unknown'}:{direction}:{hashlib.sha256(json.dumps(packet, sort_keys=True, default=str).encode()).hexdigest()[:16]}"
            summary = packet.get("text") or packet.get("intent") or ""
            if len(summary) > 200:
                summary = summary[:197] + "..."
            writer.write_message(
                message_id=message_id,
                thread_id=thread_id,
                sender=sender,
                recipient=recipient,
                subject=subject,
                summary=summary,
                mentioned_files=extract_file_paths(packet),
                mentioned_entities=extract_entity_ids(packet),
                project_id=getattr(self, "_project_id", None),
            )
        except Exception as exc:
            _logger.debug("Message signal write failed: %s", exc)

    def _write_tool_call_signal(
        self,
        *,
        tool_name: str,
        arguments: dict[str, Any],
        result_summary: str = "",
        outcome: str = "",
        thread_id: str = "",
    ) -> None:
        """Emit a Tier-H ToolCall node.

        Best-effort: if the graph is unavailable the writer degrades silently.
        """
        self._kg_available_now()  # opportunistic redial (backlog #219)
        writer = getattr(self, "_runtime_writer", None)
        if writer is None:
            return
        try:
            from orchestrata_ultimate.knowledge_graph.signal_extraction import (
                extract_entity_ids,
                extract_file_paths,
            )

            tool_call_id = f"tc:{thread_id or 'unknown'}:{tool_name}:{hashlib.sha256(json.dumps(arguments, sort_keys=True, default=str).encode()).hexdigest()[:16]}"
            writer.write_tool_call(
                tool_call_id=tool_call_id,
                thread_id=thread_id,
                tool_name=tool_name,
                arguments=arguments,
                result_summary=result_summary,
                outcome=outcome,
                mentioned_files=extract_file_paths(arguments),
                mentioned_entities=extract_entity_ids(arguments),
                project_id=getattr(self, "_project_id", None),
            )
        except Exception as exc:
            _logger.debug("Tool call signal write failed: %s", exc)

    @traces(spec="008", story="US-003", reqs=["REQ-008-043"])
    def _session_context_data(self) -> ContextData | None:
        """Load session context from whichever runner's controller is available.

        The orchestrator runner stores its controller in ``self._session_ctrl``
        while the kernel runner stores its controller in ``self.session``.
        Returns ``None`` when no controller is wired or reading fails, so
        context injection degrades gracefully outside a managed session.
        """
        controller = getattr(self, "_session_ctrl", None) or getattr(
            self, "session", None
        )
        if controller is None:
            return None
        try:
            return cast("ContextData | None", controller.ctx.read())
        except Exception as exc:
            _logger.debug("Failed to read session context: %s", exc)
            return None

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def _assemble_context_for_subject(
        self,
        payload: dict[str, Any],
        subject: str,
        assembler: UnifiedContextAssembler,
        *,
        thread_id: str = "",
        artifact_store: Any | None = None,
    ) -> dict[str, Any]:
        """Apply the unified assembler and write its parts into ``payload``.

        The assembler returns budget-separated parts; this helper collapses
        the non-user layers into ``payload["graph_context"]`` so existing
        volks (which read only that key) receive memory/artifacts/mission/
        graph/task context without changing the CXP contract. The structured
        ``task_context`` dict is also preserved for volks that look for it.
        """
        if artifact_store is None:
            artifact_store = getattr(self, "artifacts", None)

        assembly = assembler.assemble(
            payload=payload,
            subject=subject,
            user_message=payload.get("text")
            or payload.get("requirements")
            or payload.get("intent")
            or "",
            task_type=SUBJECT_TASK_TYPE.get(subject),
            task_params=payload,
            volk_alias="",
            context_data=self._session_context_data(),
            artifact_store=artifact_store,
            thread_id=thread_id,
        )
        parts = assembly.parts
        blocks: list[str] = []
        for key in ("mission", "plan", "memory", "artifacts", "graph"):
            text = parts.get(key, "")
            if text:
                blocks.append(text)

        task_text = parts.get("task", "")
        if task_text:
            try:
                task_dict = json.loads(task_text)
            except Exception:
                task_dict = None
            if task_dict:
                payload["task_context"] = task_dict
                summary = task_dict.get("summary") or task_text
                blocks.append(f"# TASK CONTEXT\n{summary}")
            else:
                blocks.append(f"# TASK CONTEXT\n{task_text}")

        if blocks:
            payload["graph_context"] = "\n\n---\n\n".join(blocks)

        return payload

    @traces(spec="008", story="US-003", reqs=["REQ-008-013"])
    def _maybe_inject_context(self, packet: CXPPacket) -> CXPPacket:
        """Sync project files to KG and inject context if the subject warrants it.

        Thin CXPPacket-shaped wrapper over :meth:`inject_for_subject` —
        mutates and returns the packet in place so escalation routing
        (``_route_escalation``) can chain it.

        Args:
            packet: The CXP packet to potentially enrich.

        Returns:
            The packet with payload possibly enriched by KG context.
        """
        packet.payload = self.inject_for_subject(
            packet.payload,
            packet.subject,
            thread_id=packet.thread_id,
            artifact_store=getattr(self, "artifacts", None),
        )
        return packet

    @traces(spec="008", story="US-005", reqs=["REQ-008-024"])
    def _pre_task_sync(
        self, project_dir: str, *, env: Mapping[str, str] | None = None
    ) -> None:
        """Sync changed project files to the graph before task injection.

        Gated by ``ORCHESTRATA_KG_PRESTAGE_SYNC``, read at call time (RUL-003
        — never a module-level constant, since this runs once per stage):
        unset or any of "1"/"true"/"yes"/"on" preserves today's behaviour
        (the sync runs); "0"/"false"/"no"/"off" skips it outright and logs
        one line saying so. ``FileSynchronizer.ensure_fresh`` already
        carries its own file-count and wall-clock budget so a slow tree
        degrades instead of hanging (see its docstring) — this switch is a
        coarser escape hatch for callers who want to bypass the sync
        entirely.

        Args:
            project_dir: Path to the project directory to sync.
            env: Optional env mapping (test injection seam, mirrors
                ``_init_context_injector``); defaults to ``os.environ``.
        """
        env_map = env if env is not None else os.environ
        if env_map.get("ORCHESTRATA_KG_PRESTAGE_SYNC", "1").strip().lower() in (
            "0",
            "false",
            "no",
            "off",
        ):
            print(
                "  [KG Sync] Pre-task sync disabled via "
                "ORCHESTRATA_KG_PRESTAGE_SYNC",
                flush=True,
            )
            return
        try:
            from orchestrata_ultimate.knowledge_graph.synchronizer import (
                FileSynchronizer,
            )

            project_id = self._project_id or derive_project_id(project_dir)
            syncer = FileSynchronizer(
                graph_client=self._get_graph_client(), project_id=project_id
            )
            resynced = syncer.ensure_fresh(project_dir)
            if resynced:
                print(
                    f"  [KG Sync] Pre-task: {resynced} files re-synced",
                    flush=True,
                )
        except Exception as exc:
            _logger.warning(
                "Pre-task KG sync failed for %r: %s", project_dir, exc, exc_info=True
            )
            print(f"  [KG Sync] Pre-task sync skipped: {exc}", flush=True)