orchestrata_ultimate.knowledge_graph.runtime_writer

Reference › orchestrata_ultimate › knowledge_graph › runtime_writer

RuntimeSignalWriter persists operational signals as Tier-H nodes in the operational graph (orchkg on orchgraph) (Spec 106, REQ-106-066). The signals are messages, gateway events, decisions and tool calls, with timeouts and escalations described in the module docstring. Each is merged under a deterministic id, attached to the root Project whose slug _project_slug resolves, and linked by MENTIONS edges to the code entities it concerns, so that Prolog relevance rules can boost recently touched or failing nodes for debugging tasks. The writer is constructed with a synchronous graph repository, or None.

The public methods are write_message, write_gateway_event, write_decision (volk alias, rationale, requirements, files, entities) and write_tool_call (tool name, arguments, result summary, outcome). Each accepts mentioned files and entity ids, which _link_entities turns into MENTIONS edges by canonical id; signal_extraction is the helper that finds those mentions in a payload. All writes go through _run, which swallows and logs failures: a transient graph outage never breaks the pipeline. The volk base class and the kernel's pipeline_kg import it. Labels and keys follow canonical_schema.

At a glance

Class diagram

Dependency diagram

API

Classes

RuntimeSignalWriter

Decorators: traces(spec='106', reqs=['REQ-106-066']) · Line 42

Write runtime Tier-H signals to the operational graph.

Functions

Source

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

"""Runtime Tier-H signal writer.

Persists operational signals (messages, gateway events, decisions, tool calls,
timeouts, escalations) into the knowledge graph as Tier-H nodes. Each
signal is MERGEd by a deterministic id and linked to the code entities it
concerns via ``MENTIONS`` edges so that Prolog relevance rules can boost
recently touched or failing nodes for DEBUG tasks.

All writes are best-effort: failures are logged and swallowed so a transient
graph outage does not break the pipeline.
"""

from __future__ import annotations

import json
import logging
import os
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph.interfaces import SyncGraphRepository

_logger = logging.getLogger(__name__)

_DEFAULT_PROJECT_SLUG = "orchestrata-ultimate"


def _project_slug() -> str:
    """Resolve the root ``:Project`` slug this writer's hub edge MERGEs onto.

    Read at call time, never at import (Key Rule / RUL-003): the test suite and
    the CLI both override ``ORCHESTRATA_PROJECT_ID`` per process.
    """
    return os.environ.get("ORCHESTRATA_PROJECT_ID", _DEFAULT_PROJECT_SLUG)


@traces(spec="106", reqs=["REQ-106-066"])
class RuntimeSignalWriter:
    """Write runtime Tier-H signals to the operational graph.

    Args:
        repo: ``SyncGraphRepository`` for the operational graph. If None,
            every write degrades to a no-op and logs a warning once.
    """

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

    def _connected_repo(self) -> SyncGraphRepository:
        if self._repo is None:
            raise RuntimeError("RuntimeSignalWriter not connected")
        return self._repo

    def _run(self, cypher: str, params: dict[str, Any]) -> list[dict[str, Any]]:
        """Execute one write, swallowing failures so the pipeline survives."""
        try:
            return self._connected_repo().execute_write(cypher, params)
        except Exception as exc:
            if not self._warned:
                _logger.warning(
                    "Runtime signal write failed (further failures at DEBUG): %s", exc
                )
                self._warned = True
            else:
                _logger.debug("Runtime signal write failed: %s", exc)
            return []

    def write_message(
        self,
        message_id: str,
        *,
        thread_id: str = "",
        sender: str = "",
        recipient: str = "",
        subject: str = "",
        summary: str = "",
        mentioned_files: list[str] | None = None,
        mentioned_entities: list[str] | None = None,
        timestamp: datetime | None = None,
        project_id: str | None = None,
    ) -> None:
        """Persist an inter-agent message node.

        Args:
            message_id: Deterministic id for the message.
            thread_id: Pipeline thread / session id.
            sender: Volk alias or role that sent the message.
            recipient: Volk alias or role that received the message.
            subject: CXP subject, if any.
            summary: Short content summary (already truncated by caller).
            mentioned_files: File paths the message references.
            mentioned_entities: Code entity ids the message references.
            timestamp: UTC datetime; defaults to now.
            project_id: Optional project scope.
        """
        ts = (timestamp or datetime.now(UTC)).isoformat()
        project = project_id or _project_slug()
        # Message's canonical merge key is `message_id` (canonical_schema
        # CANONICAL_MERGE_KEYS -- Message is a Tier H natural-id label, never
        # the shapeless-event derived `id`). MERGEing on `id` here diverged
        # from the `message_message_id` uniqueness constraint --
        # ConstraintValidationFailed once a canonically-keyed Message node
        # with the same message_id already existed. `id` is kept (never drop
        # data; `_link_entities` MATCHes MENTIONS sources by `id`).
        params: dict[str, Any] = {
            "id": message_id,
            "message_id": message_id,
            "thread_id": thread_id,
            "sender": sender,
            "recipient": recipient,
            "subject": subject,
            "summary": summary[:1000],
            "timestamp": ts,
            "project": project,
            "mentioned_files": mentioned_files or [],
        }
        if thread_id:
            params["thread_qid"] = cs.thread_entity_id(project, thread_id)
            self._run(
                f"MERGE (m:{cs.LABEL_MESSAGE} {{message_id: $message_id}}) "
                "SET m.id = $id, "
                "    m.thread_id = $thread_id, "
                "    m.sender = $sender, "
                "    m.recipient = $recipient, "
                "    m.subject = $subject, "
                "    m.summary = $summary, "
                "    m.timestamp = datetime($timestamp), "
                "    m.project_id = $project "
                "WITH m "
                f"MERGE (t:{cs.LABEL_THREAD} {{id: $thread_qid}}) "
                "  ON CREATE SET t.thread_id = $thread_id, t.project = $project "
                f"MERGE (m)-[:{cs.REL_IN_THREAD}]->(t) "
                "WITH m, t "
                f"MERGE (proj:{cs.LABEL_PROJECT} {{slug: $project}}) "
                f"MERGE (proj)-[:{cs.REL_HAS_THREAD}]->(t) "
                "WITH m "
                "UNWIND $mentioned_files AS path "
                "MATCH (f:File {path: path}) "
                "MERGE (m)-[:MENTIONS]->(f)",
                params,
            )
        else:
            _logger.warning(
                "Message %s has no thread_id; writing without thread anchor", message_id
            )
            self._run(
                f"MERGE (m:{cs.LABEL_MESSAGE} {{message_id: $message_id}}) "
                "SET m.id = $id, "
                "    m.thread_id = $thread_id, "
                "    m.sender = $sender, "
                "    m.recipient = $recipient, "
                "    m.subject = $subject, "
                "    m.summary = $summary, "
                "    m.timestamp = datetime($timestamp), "
                "    m.project_id = $project "
                "WITH m "
                "UNWIND $mentioned_files AS path "
                "MATCH (f:File {path: path}) "
                "MERGE (m)-[:MENTIONS]->(f)",
                params,
            )
        self._link_entities(message_id, cs.LABEL_MESSAGE, mentioned_entities or [])

    def write_gateway_event(
        self,
        entry_id: str,
        *,
        thread_id: str = "",
        event_type: str = "",
        outcome: str = "",
        summary: str = "",
        mentioned_files: list[str] | None = None,
        mentioned_entities: list[str] | None = None,
        timestamp: datetime | None = None,
        project_id: str | None = None,
    ) -> None:
        """Persist a gateway/audit event node."""
        ts = (timestamp or datetime.now(UTC)).isoformat()
        project = project_id or _project_slug()
        # GatewayEvent's canonical merge key is `entry_id` (same reasoning as
        # write_message's `message_id` fix above -- see that comment).
        params: dict[str, Any] = {
            "id": entry_id,
            "entry_id": entry_id,
            "thread_id": thread_id,
            "event_type": event_type,
            "outcome": outcome,
            "summary": summary[:1000],
            "timestamp": ts,
            "project": project,
            "mentioned_files": mentioned_files or [],
        }
        if thread_id:
            params["thread_qid"] = cs.thread_entity_id(project, thread_id)
            self._run(
                f"MERGE (e:{cs.LABEL_GATEWAY_EVENT} {{entry_id: $entry_id}}) "
                "SET e.id = $id, "
                "    e.thread_id = $thread_id, "
                "    e.event_type = $event_type, "
                "    e.outcome = $outcome, "
                "    e.summary = $summary, "
                "    e.timestamp = datetime($timestamp), "
                "    e.project_id = $project "
                "WITH e "
                f"MERGE (t:{cs.LABEL_THREAD} {{id: $thread_qid}}) "
                "  ON CREATE SET t.thread_id = $thread_id, t.project = $project "
                f"MERGE (e)-[:{cs.REL_IN_THREAD}]->(t) "
                "WITH e, t "
                f"MERGE (proj:{cs.LABEL_PROJECT} {{slug: $project}}) "
                f"MERGE (proj)-[:{cs.REL_HAS_THREAD}]->(t) "
                "WITH e "
                "UNWIND $mentioned_files AS path "
                "MATCH (f:File {path: path}) "
                "MERGE (e)-[:MENTIONS]->(f)",
                params,
            )
        else:
            _logger.warning(
                "GatewayEvent %s has no thread_id; writing without thread anchor",
                entry_id,
            )
            self._run(
                f"MERGE (e:{cs.LABEL_GATEWAY_EVENT} {{entry_id: $entry_id}}) "
                "SET e.id = $id, "
                "    e.thread_id = $thread_id, "
                "    e.event_type = $event_type, "
                "    e.outcome = $outcome, "
                "    e.summary = $summary, "
                "    e.timestamp = datetime($timestamp), "
                "    e.project_id = $project "
                "WITH e "
                "UNWIND $mentioned_files AS path "
                "MATCH (f:File {path: path}) "
                "MERGE (e)-[:MENTIONS]->(f)",
                params,
            )
        self._link_entities(entry_id, cs.LABEL_GATEWAY_EVENT, mentioned_entities or [])

    def write_decision(
        self,
        decision_id: str,
        *,
        volk_alias: str = "",
        rationale: str = "",
        requirements: list[str] | None = None,
        files: list[str] | None = None,
        entities: list[str] | None = None,
        timestamp: datetime | None = None,
        project_id: str | None = None,
    ) -> None:
        """Persist a governance decision node.

        This is the runtime counterpart to ``PipelineKGMixin._sync_decisions_to_kg``:
        decisions made during a run are surfaced as Tier-H signals in addition to
        any durable decision nodes the pipeline already writes.
        """
        ts = (timestamp or datetime.now(UTC)).isoformat()
        self._run(
            "MERGE (d:Decision {id: $id}) "
            "SET d.decision_id = $decision_id, "
            "    d.rationale = $rationale, "
            "    d.timestamp = datetime($timestamp), "
            "    d.project_id = $project_id "
            "WITH d "
            "UNWIND $files AS path "
            "MATCH (f:File {path: path}) "
            "MERGE (d)-[:APPLIES_TO]->(f)",
            {
                "id": decision_id,
                "decision_id": decision_id,
                "rationale": rationale[:2000],
                "timestamp": ts,
                "project_id": project_id,
                "files": files or [],
            },
        )
        if volk_alias:
            # MADE_BY is the FR-002-named reverse edge; MADE is what the
            # spec's own example query traverses (query_engine.volk_work_context)
            # — keep both so either traversal direction resolves.
            self._run(
                "MATCH (d:Decision {id: $id}), (v:Volk {alias: $alias}) "
                "MERGE (v)-[:MADE]->(d) "
                "MERGE (d)-[:MADE_BY]->(v)",
                {"id": decision_id, "alias": volk_alias},
            )
        for req in requirements or []:
            self._run(
                "MATCH (d:Decision {id: $id}) "
                "MERGE (r:Requirement {id: $req_id}) "
                "MERGE (d)-[:JUSTIFIED_BY]->(r)",
                {"id": decision_id, "req_id": req},
            )
        self._link_entities(decision_id, cs.LABEL_DECISION, entities or [])

    def write_tool_call(
        self,
        tool_call_id: str,
        *,
        thread_id: str = "",
        tool_name: str = "",
        arguments: dict[str, Any] | None = None,
        result_summary: str = "",
        outcome: str = "",
        mentioned_files: list[str] | None = None,
        mentioned_entities: list[str] | None = None,
        timestamp: datetime | None = None,
        project_id: str | None = None,
    ) -> None:
        """Persist a tool invocation node."""
        ts = (timestamp or datetime.now(UTC)).isoformat()
        project = project_id or _project_slug()
        args = arguments or {}
        params: dict[str, Any] = {
            "id": tool_call_id,
            "tool_call_id": tool_call_id,
            "thread_id": thread_id,
            "tool_name": tool_name,
            "arguments_json": json.dumps(args, sort_keys=True),
            "result_summary": result_summary[:1000],
            "outcome": outcome,
            "timestamp": ts,
            "project": project,
            "mentioned_files": mentioned_files or [],
        }
        if thread_id:
            params["thread_qid"] = cs.thread_entity_id(project, thread_id)
            self._run(
                f"MERGE (tc:{cs.LABEL_TOOL_CALL} {{id: $id}}) "
                "SET tc.tool_call_id = $tool_call_id, "
                "    tc.thread_id = $thread_id, "
                "    tc.tool_name = $tool_name, "
                "    tc.arguments_json = $arguments_json, "
                "    tc.result_summary = $result_summary, "
                "    tc.outcome = $outcome, "
                "    tc.timestamp = datetime($timestamp), "
                "    tc.project_id = $project "
                "WITH tc "
                f"MERGE (t:{cs.LABEL_THREAD} {{id: $thread_qid}}) "
                "  ON CREATE SET t.thread_id = $thread_id, t.project = $project "
                f"MERGE (tc)-[:{cs.REL_IN_THREAD}]->(t) "
                "WITH tc, t "
                f"MERGE (proj:{cs.LABEL_PROJECT} {{slug: $project}}) "
                f"MERGE (proj)-[:{cs.REL_HAS_THREAD}]->(t) "
                "WITH tc "
                "UNWIND $mentioned_files AS path "
                "MATCH (f:File {path: path}) "
                "MERGE (tc)-[:MENTIONS]->(f)",
                params,
            )
        else:
            _logger.warning(
                "ToolCall %s has no thread_id; writing without thread anchor",
                tool_call_id,
            )
            self._run(
                f"MERGE (tc:{cs.LABEL_TOOL_CALL} {{id: $id}}) "
                "SET tc.tool_call_id = $tool_call_id, "
                "    tc.thread_id = $thread_id, "
                "    tc.tool_name = $tool_name, "
                "    tc.arguments_json = $arguments_json, "
                "    tc.result_summary = $result_summary, "
                "    tc.outcome = $outcome, "
                "    tc.timestamp = datetime($timestamp), "
                "    tc.project_id = $project "
                "WITH tc "
                "UNWIND $mentioned_files AS path "
                "MATCH (f:File {path: path}) "
                "MERGE (tc)-[:MENTIONS]->(f)",
                params,
            )
        self._link_entities(tool_call_id, cs.LABEL_TOOL_CALL, mentioned_entities or [])

    def _link_entities(
        self, source_id: str, source_label: str, entity_ids: list[str]
    ) -> None:
        """Link a Tier-H signal to code entities by canonical id via MENTIONS."""
        if not entity_ids:
            return
        self._run(
            f"MATCH (s:{source_label} {{id: $source_id}}) "
            "UNWIND $entity_ids AS entity_id "
            "MATCH (n) WHERE n.id = entity_id "
            "MERGE (s)-[:MENTIONS]->(n)",
            {"source_id": source_id, "entity_ids": entity_ids},
        )