orchestrata_ultimate.knowledge_graph.ontology_mapping

Reference › orchestrata_ultimate › knowledge_graph › ontology_mapping

A pure mapping from the labels found in a live graph to the canonical ontology defined in canonical_schema (Spec 106, REQ-106-001 to -003). It performs no I/O; an importer applies it node by node to the operational graph (orchkg on orchgraph). Duplicate label trees converge: for example CodeClass maps to Class, CodeFunction and Method to Function, Specification to Spec, Agent to Volk, Run to TestRun and WorkItem to Task. Operational events and messages (CXPPacket to Message, GatewayAuditEvent to GatewayEvent, governance and lifecycle events) are imported at full fidelity as Tier-H records rather than discarded.

The data is in SOURCE_LABEL_MAP, REL_MAP (containment synonyms fold into DEFINES), KEY_DERIVATION (per-label functions such as _function_class_key, _session_key, _memory_key and _event_key that derive an identity when a node has no canonical key; code entities use file_path::qualname) and the label sets ADDED_ONLY_LABELS and FOLDED_CANONICAL_LABELS. resolve_canonical_label picks one canonical label from a node's label set, map_node returns a frozen Mapped (label, key name, key value) or a Skipped with a machine-readable reason, map_relationship renames an edge type, and run_id_for_thread gives the pipeline-run id for a conversation thread. A node that cannot produce a key is reported, never fabricated. Only tests import the module.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

Mapped (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-001', 'REQ-106-002', 'REQ-106-003']), dataclass(frozen=True) · Line 191

A source node resolved to its canonical shape, ready to MERGE.

Skipped (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-001', 'REQ-106-002', 'REQ-106-003']), dataclass(frozen=True) · Line 202

A source node that is not imported, with a machine-readable reason.

Functions

Source

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

"""Graph → canonical-ontology mapping — full-fidelity, nothing discarded.

Maps every live graph label to its canonical label (:mod:`canonical_schema`).
Duplicate label trees *converge* (``CodeClass``→``Class``, ``CodeFunction``/
``Method``→``Function``, ``Specification``→``Spec``, ``Agent``→``Volk``,
``Run``→``TestRun``, ``WorkItem``→``Task``, ``TestNode``/``DLTest``→``Test``); the
operational events & messages (``CXPPacket``→``Message``, ``GatewayAuditEvent``→
``GatewayEvent``, governance/lifecycle events) are Tier H — NOT churn to discard,
but real, identifiable records imported at full fidelity.

Every node is identified: by its canonical key, an entity-specific id fallback
(``fact_id``/``spec_id``/…), or a derived id (code ``file_path::qualname``, event
``<session_id>::<timestamp>``). Pure — no I/O. The importer applies
:func:`map_node`; a node that genuinely cannot produce a key is reported (never
fabricated), matching ``canonical_schema.check``'s conformance stance.
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

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

# Live graph label → canonical label, or ``None`` to EXCLUDE (operational churn
# that is NOT semantic knowledge and stays queryable in the graph). Every entry is
# grounded in the live 48-label census + the ratified ontology §4.
SOURCE_LABEL_MAP: dict[str, str | None] = {
    # ---- adopt: already the canonical label ---------------------------------
    "Module": cs.LABEL_MODULE,
    "Package": cs.LABEL_PACKAGE,
    "File": cs.LABEL_FILE,
    "Parameter": cs.LABEL_PARAMETER,
    "Rule": cs.LABEL_RULE,
    "Directory": cs.LABEL_DIRECTORY,
    "Predicate": cs.LABEL_PREDICATE,
    "Action": cs.LABEL_ACTION,
    "Portfolio": cs.LABEL_PORTFOLIO,
    "RequirementItem": cs.LABEL_REQUIREMENT_ITEM,
    "Class": cs.LABEL_CLASS,
    "Function": cs.LABEL_FUNCTION,
    "Project": cs.LABEL_PROJECT,
    "Intent": cs.LABEL_INTENT,
    "PipelineRun": cs.LABEL_PIPELINE_RUN,
    "Task": cs.LABEL_TASK,
    "Component": cs.LABEL_COMPONENT,
    "Artifact": cs.LABEL_ARTIFACT,
    "Requirement": cs.LABEL_REQUIREMENT,
    "Scenario": cs.LABEL_SCENARIO,
    "UserStory": cs.LABEL_USER_STORY,
    "Volk": cs.LABEL_VOLK,
    "Session": cs.LABEL_SESSION,
    # Thread (session-254 Option A, plans/thread-ontology-design.md): the
    # Tier-H correlation anchor every producer writes directly onto its
    # canonical `id` (`thread_entity_id`) -- adopt path, no KEY_DERIVATION.
    # Pre-existing gap: LABEL_THREAD was declared in canonical_schema.py
    # without ever being wired into this map, so it silently failed the
    # every-canonical-label-is-reachable completeness invariant below.
    "Thread": cs.LABEL_THREAD,
    "Decision": cs.LABEL_DECISION,
    "Pattern": cs.LABEL_PATTERN,
    "Fact": cs.LABEL_FACT,
    "Memory": cs.LABEL_MEMORY,
    "DesignModule": cs.LABEL_DESIGN_MODULE,
    "DesignClass": cs.LABEL_DESIGN_CLASS,
    "Architecture": cs.LABEL_ARCHITECTURE,
    "Test": cs.LABEL_TEST,
    "TestRun": cs.LABEL_TEST_RUN,
    "TestResult": cs.LABEL_TEST_RESULT,
    # Tier I — conversation capture (SPEC-021). The consumer writes the
    # canonical "id" directly (adopt path) — no KEY_DERIVATION needed.
    "Conversation": cs.LABEL_CONVERSATION,
    "ConversationTurn": cs.LABEL_CONVERSATION_TURN,
    # Tier J — architecture ontology (RATIFIED 2026-07-19, wave
    # ontology-phase1). ``architecture_writer.py`` (the converged writer that
    # supersedes ``sync_architecture_to_graph.py``'s pre-ontology shape) MERGEs
    # these 18 labels directly onto their canonical key (``id`` for the
    # project-scoped ones, ``name`` for the two it actually seeds globally —
    # WAFPillar/CloudService) — an adopt path, same as Module/Package/... above,
    # no KEY_DERIVATION needed. ``Change``/``Region`` are NOT writer outputs
    # this wave (Change has no upsert method yet; Region seeding is lazy/
    # deferred per ruling 6) and stay in ADDED_ONLY_LABELS below.
    "ApiContract": cs.LABEL_API_CONTRACT,
    "DataEntity": cs.LABEL_DATA_ENTITY,
    "QualityAttribute": cs.LABEL_QUALITY_ATTRIBUTE,
    "WAFPillar": cs.LABEL_WAF_PILLAR,
    "Constraint": cs.LABEL_CONSTRAINT,
    "CapacityModel": cs.LABEL_CAPACITY_MODEL,
    "CostModel": cs.LABEL_COST_MODEL,
    "CostLineItem": cs.LABEL_COST_LINE_ITEM,
    "Environment": cs.LABEL_ENVIRONMENT,
    "Deployment": cs.LABEL_DEPLOYMENT,
    "ArchitectureDecision": cs.LABEL_ARCHITECTURE_DECISION,
    "TradeOff": cs.LABEL_TRADE_OFF,
    "Risk": cs.LABEL_RISK,
    "FailureMode": cs.LABEL_FAILURE_MODE,
    "TechnologyChoice": cs.LABEL_TECHNOLOGY_CHOICE,
    "CloudService": cs.LABEL_CLOUD_SERVICE,
    "ServiceLevelObjective": cs.LABEL_SERVICE_LEVEL_OBJECTIVE,
    "Alert": cs.LABEL_ALERT,
    # ---- converge: duplicate label trees → the single canonical label -------
    "CodeClass": cs.LABEL_CLASS,
    "CodeFunction": cs.LABEL_FUNCTION,
    "Method": cs.LABEL_FUNCTION,  # methods are Function nodes (+ class_id prop)
    "Specification": cs.LABEL_SPEC,
    "Agent": cs.LABEL_VOLK,
    "Run": cs.LABEL_TEST_RUN,  # per decision: test-run telemetry → TestRun (derived key)
    "WorkItem": cs.LABEL_TASK,
    "TestNode": cs.LABEL_TEST,
    "DLTest": cs.LABEL_TEST,
    # ---- Tier H: operational events & messages (full-fidelity — real records,
    #      each identifiable by a natural id or a derived <session_id>::<timestamp>)
    "GatewayAuditEvent": cs.LABEL_GATEWAY_EVENT,
    "CXPPacket": cs.LABEL_MESSAGE,
    "AuditEntry": cs.LABEL_AUDIT_ENTRY,
    "Event": cs.LABEL_EVENT,
    "PolicyDecision": cs.LABEL_POLICY_DECISION,
    "Resolution": cs.LABEL_RESOLUTION,
    "Violation": cs.LABEL_VIOLATION,
    "Escalation": cs.LABEL_ESCALATION,
    "Timeout": cs.LABEL_TIMEOUT,
    "SessionStart": cs.LABEL_SESSION_START,
    "SessionEnd": cs.LABEL_SESSION_END,
    "ToolCall": cs.LABEL_TOOL_CALL,
}

# Canonical labels that are NOT sourced from the graph — they are ADDED by the
# importer from elsewhere: ``Tool`` from ``tools.registry`` and ``SemanticChunk``
# by unifying Memgraph's existing chunk labels. Every OTHER canonical label must
# be reachable as a target in SOURCE_LABEL_MAP (enforced by a completeness test).
#
# Tier J — architecture ontology (RATIFIED 2026-07-19,
# plans/architecture-ontology-draft.md, wave ontology-phase1): 18 of the 20
# Tier J labels are now ADOPTED above — ``architecture_writer.py`` (the
# converged writer, wave ontology-phase1) MERGEs them directly onto their
# canonical key, so they moved out of this set once that writer landed (this
# comment previously said "once that writer lands ... should be pulled out
# of this set then" — done). The two still here have no writer output yet:
# ``Change`` (no ``upsert_change`` method this wave — OpenSpec integration is
# a separate follow-up) and ``Region`` (seeding is explicitly lazy/deferred
# per ruling 6, not this wave).
#
# Tier K — factory fact-namespace control (SPEC-033 §4, B4a, ADR-0004):
# ``FactNamespace`` is never imported FROM Neo4j by this module -- it is
# written directly by ``factory.promotion.store.PromotionStore`` (the
# session-254 root-connectivity fix), so it has no ``SOURCE_LABEL_MAP`` entry
# and belongs here alongside Tool/SemanticChunk.
ADDED_ONLY_LABELS: frozenset[str] = frozenset(
    {
        cs.LABEL_TOOL,
        cs.LABEL_SEMANTIC_CHUNK,
        cs.LABEL_CHANGE,
        cs.LABEL_REGION,
        cs.LABEL_FACT_NAMESPACE,
        # ``Agent`` (session-254): written directly by
        # ``memory.cold_store``'s ``MERGE (v:Agent {alias: $volk_id})`` and
        # hub-linked ``(:Project)-[:HAS_AGENT]->``. UNLIKE ``FactNamespace``,
        # it DOES have a ``SOURCE_LABEL_MAP`` entry (``Agent`` -> ``Volk``), so
        # it is also a member of ``FOLDED_CANONICAL_LABELS`` below. It stays in
        # this set because the direct-write path is real; the earlier claim here
        # that it "has no SOURCE_LABEL_MAP entry" was false and is corrected.
        cs.LABEL_AGENT,
    }
)

# Canonical labels that are deliberately FOLDED AWAY on import: they appear in
# ``SOURCE_LABEL_MAP`` as KEYS pointing at a different canonical label, so they
# are never import TARGETS -- yet they are canonical, because they exist live
# with their own UNIQUE constraint and node population (measured 2026-08-06:
# 95,176 ``CXPPacket`` nodes). This is transitional duplication: the converged
# twin (``Message``/``Function``) is the write target for new data, while the
# folded label stays canonical until its nodes are migrated. Distinct from
# ADDED_ONLY_LABELS, which is "no SOURCE_LABEL_MAP entry at all because another
# writer owns the label" -- these DO have an entry, pointing elsewhere.
#
# Derived, never hand-listed: an entry can only appear here by actually being a
# canonical label that the map folds into a DIFFERENT canonical label. That
# makes the exemption self-verifying -- it cannot be widened to hide a label
# that was simply forgotten, which is how a "reachability" gate goes false-green.
FOLDED_CANONICAL_LABELS: frozenset[str] = frozenset(
    source
    for source, target in SOURCE_LABEL_MAP.items()
    if source in cs.CANONICAL_LABELS and target is not None and target != source
)


@traces(spec="106", reqs=["REQ-106-001", "REQ-106-002", "REQ-106-003"])
@dataclass(frozen=True)
class Mapped:
    """A source node resolved to its canonical shape, ready to MERGE."""

    label: str
    key: str
    key_value: str
    properties: dict[str, Any] = field(default_factory=dict)


@traces(spec="106", reqs=["REQ-106-001", "REQ-106-002", "REQ-106-003"])
@dataclass(frozen=True)
class Skipped:
    """A source node that is not imported, with a machine-readable reason.

    ``reason`` is one of: ``"unmapped:<labels>"`` (no rule for any label),
    ``"excluded:<label>"`` (a deliberate churn exclusion), ``"ambiguous:<a|b>"``
    (labels map to two different canonical types — a real modelling smell), or
    ``"missing-key:<key>"`` (the canonical merge key is absent/underivable — a
    data-quality finding the importer surfaces).
    """

    reason: str


def _code_entity_key(props: dict[str, Any]) -> str | None:
    """Derive the canonical ``<relpath>::<qualname>`` id for the CodeFunction/
    CodeClass trees, which carry ``file_path`` + ``name`` instead of an ``id``."""
    relpath = props.get("file_path") or props.get("module_path") or props.get("module")
    name = props.get("name")
    if not relpath or not name:
        return None
    class_name = props.get("class_name")
    qualname = f"{class_name}.{name}" if class_name else str(name)
    try:
        return cs.code_entity_id(str(relpath), qualname)
    except ValueError:
        return None


def _memory_key(props: dict[str, Any]) -> str | None:
    """Session-scoped Memory identity ``<session_id>::<key>`` (ratified decision)."""
    session_id = props.get("session_id")
    key = props.get("key")
    if not session_id or not key:
        return None
    return f"{session_id}::{key}"


def _run_testrun_key(props: dict[str, Any]) -> str | None:
    """Derived TestRun id for Run telemetry: ``<git_commit>::<timestamp>``."""
    commit = props.get("git_commit")
    timestamp = props.get("timestamp")
    if not commit or timestamp is None:
        return None
    return f"{commit}::{timestamp}"


def _name_key(props: dict[str, Any]) -> str | None:
    """Identity from a node's ``name`` (DesignModule/DesignClass carry no id)."""
    name = props.get("name")
    return str(name) if name else None


@traces(spec="106", story="FR-001", reqs=["REQ-106-003"])
def _session_key(props: dict[str, Any]) -> str | None:
    """Session identity across its two real shapes: gateway/runtime sessions key
    on ``session_id``; session-CLI work-sessions key on ``id``. EVERY real session
    imports — the bare ``{session_id}`` gateway sessions are NOT stubs; they are
    real connections (spanning weeks of audit-stamped activity), just recorded
    with only their id."""
    value = props.get("session_id") or props.get("id")
    return str(value) if value else None


def _function_class_key(props: dict[str, Any]) -> str | None:
    """Unified id for the whole Function/Class tree: native ``Function``/``Class``/
    ``Method`` carry ``id``; ``CodeFunction``/``CodeClass`` derive it from
    ``file_path::qualname``; orphan ``{name, tag}`` nodes fall back to ``name``."""
    existing = props.get("id")
    if existing is not None and str(existing).strip():
        return str(existing)
    derived = _code_entity_key(props)
    if derived is not None:
        return derived
    name = props.get("name")
    return str(name) if name else None


def _candidates(*names: str) -> Callable[[dict[str, Any]], str | None]:
    """Build a deriver that returns the first present, non-blank property among
    ``names`` — for labels whose identity lives under an entity-specific id
    (``fact_id``, ``spec_id``, ``req_id`` …) rather than the canonical ``id``.
    No node with a real identity is dropped just because it uses its own id name.
    """

    @traces(spec="106", reqs=["REQ-106-002", "REQ-106-003"])
    def derive(props: dict[str, Any]) -> str | None:
        for name in names:
            value = props.get(name)
            if value is not None and str(value).strip():
                return str(value)
        return None

    return derive


def _event_key(props: dict[str, Any]) -> str | None:
    """Derived id for shapeless typed-events ({data, session_id?, timestamp}):
    ``<session_id>::<timestamp>``, or just ``<timestamp>`` when there is no
    session. The (session, microsecond-timestamp) pair is unique per event."""
    timestamp = props.get("timestamp")
    if timestamp is None or not str(timestamp).strip():
        return None
    session_id = props.get("session_id")
    return f"{session_id}::{timestamp}" if session_id else str(timestamp)


# SOURCE label -> function deriving the canonical key value, for labels whose
# identity lives under a different/derived property than the canonical merge key.
# (Absent -> read the canonical key straight from the node's properties.)
KEY_DERIVATION: dict[str, Callable[[dict[str, Any]], str | None]] = {
    # Whole Function/Class tree: id | file_path::qualname | name (orphans).
    "Function": _function_class_key,
    "Class": _function_class_key,
    "CodeFunction": _function_class_key,
    "CodeClass": _function_class_key,
    "Method": _function_class_key,
    "Session": _session_key,
    "Memory": _memory_key,
    "Run": _run_testrun_key,
    "DesignModule": _name_key,
    "DesignClass": _name_key,
    # Entity-specific id fallbacks (the keyless nodes are identifiable):
    "Fact": _candidates("id", "fact_id"),
    "Specification": _candidates("id", "spec_id"),
    "Requirement": _candidates("id", "req_id"),
    "UserStory": _candidates("id", "story_id"),
    "TestRun": _candidates("id", "run_id"),
    "TestResult": _candidates("id", "result_id", "name"),
    "Project": _candidates("slug", "name"),
    "Volk": _candidates("alias", "name"),
    "Rule": _candidates("id", "rule_id", "name"),
    "Architecture": _candidates("id", "name"),
    # Tier H shapeless typed-events -> derived <session_id>::<timestamp> id.
    # (Message/GatewayAuditEvent/AuditEntry/Resolution carry a natural key —
    # message_id/entry_id/resolution_id — so they need no derivation.)
    "PolicyDecision": _event_key,
    "Event": _event_key,
    "ToolCall": _event_key,
    "Violation": _event_key,
    "Timeout": _event_key,
    "Escalation": _event_key,
    "SessionStart": _event_key,
    "SessionEnd": _event_key,
}


@traces(spec="106", story="FR-001", reqs=["REQ-106-001"])
def resolve_canonical_label(source_labels: list[str]) -> str | Skipped:
    """Pick the single canonical label for a node's set of source labels.

    Excluded labels are ignored in favour of any canonical mapping; a node whose
    labels resolve to two *different* canonical types is ``ambiguous`` (reported,
    not guessed).
    """
    known = [lbl for lbl in source_labels if lbl in SOURCE_LABEL_MAP]
    target_set: set[str] = set()
    for lbl in known:
        target = SOURCE_LABEL_MAP[lbl]
        if target is not None:
            target_set.add(target)
    targets = sorted(target_set)
    if len(targets) == 1:
        return targets[0]
    if len(targets) > 1:
        return Skipped(reason="ambiguous:" + "|".join(targets))
    if not known:
        return Skipped(reason="unmapped:" + "|".join(sorted(source_labels)))
    # Every known label mapped to None → a deliberate churn exclusion.
    return Skipped(reason="excluded:" + "|".join(sorted(known)))


@traces(spec="106", reqs=["REQ-106-001", "REQ-106-002", "REQ-106-003"])
def map_node(source_labels: list[str], properties: dict[str, Any]) -> Mapped | Skipped:
    """Map one Neo4j node to its canonical form, or explain why it is skipped.

    The canonical merge key comes from ``canonical_schema.CANONICAL_MERGE_KEYS``.
    Its value is produced by the source label's :data:`KEY_DERIVATION` function
    when the identity lives under a different/derived property (code entities,
    Memory, Run, design nodes), otherwise read straight from ``properties``. A
    :data:`NODE_FILTER` may reject a node first (e.g. bare Session stubs). A
    missing/blank/underivable key yields a ``missing-key`` :class:`Skipped` — the
    node is never given a fabricated id.
    """
    resolved = resolve_canonical_label(source_labels)
    if isinstance(resolved, Skipped):
        return resolved
    label = resolved
    # The source label(s) that drove the resolution carry any key derivation.
    drivers = sorted(lbl for lbl in source_labels if SOURCE_LABEL_MAP.get(lbl) == label)
    key = cs.CANONICAL_MERGE_KEYS[label]
    value: str | None = None
    for driver in drivers:
        derive = KEY_DERIVATION.get(driver)
        if derive is not None:
            value = derive(properties)
            break
    if value is None:
        raw = properties.get(key)
        if raw is not None and not (isinstance(raw, str) and not raw.strip()):
            value = str(raw)
    if value is None or not value.strip():
        return Skipped(reason=f"missing-key:{key}")
    # The mapped properties must carry the CANONICAL key value (stringified /
    # derived), not the raw source one: importers MERGE on ``key_value`` and
    # then ``SET n += properties`` — a raw non-string ``id`` in properties
    # would clobber the merge key with a different type (two integer-id :Test
    # nodes reached the live graph this way and poisoned every STARTS WITH
    # scan, 2026-07-08).
    normalized = dict(properties)
    normalized[key] = value
    return Mapped(label=label, key=key, key_value=value, properties=normalized)


# ---------------------------------------------------------------------------
# Relationship-type ontology.
# ---------------------------------------------------------------------------
# The live graph's 33 edge types are mostly already distinct and meaningful —
# DURING, ROUTED_TO, ABOUT, BELONGS_TO, IN_SESSION, TRIGGERED, CALLS, etc. all
# carry different semantics and are PRESERVED, never collapsed. The only mandated
# convergence is the writer-specific containment synonyms (HAS_CLASS/HAS_FUNCTION/
# HAS_METHOD) into the single canonical DEFINES (canonical_schema convention).
# Endpoint labels converge automatically via map_node (Method->Function, Agent->
# Volk, CodeClass->Class, ...), so e.g. (Class)-[:HAS_METHOD]->(Method) imports
# as the canonical (Class)-[:DEFINES]->(Function).
REL_MAP: dict[str, str] = {
    "HAS_CLASS": cs.REL_DEFINES,
    "HAS_FUNCTION": cs.REL_DEFINES,
    "HAS_METHOD": cs.REL_DEFINES,
}


@traces(spec="106", reqs=["REQ-106-004"])
def map_relationship(rel_type: str) -> str:
    """Canonical relationship name: fold containment synonyms into DEFINES,
    preserve every other edge type as-is."""
    return REL_MAP.get(rel_type, rel_type)


# ---------------------------------------------------------------------------
# Derived provenance — thread_id -> pipeline run.
# ---------------------------------------------------------------------------
# A node carrying a ``thread_id`` belongs to the pipeline run whose id is
# ``run:<thread_id>`` (Intents already assert this via TRIGGERED; this extends the
# SAME faithful link to that run's Messages and Artifacts). This is not
# fabrication: the run's id and the node's thread_id encode the identical thread.
REL_IN_RUN = "IN_RUN"

# Source labels whose thread_id grounds them in their run (Intent is excluded — it
# already carries a native TRIGGERED edge to its PipelineRun).
THREAD_PROVENANCE_LABELS: tuple[str, ...] = ("CXPPacket", "Artifact")


def run_id_for_thread(thread_id: str) -> str:
    """The canonical PipelineRun id for a conversation thread."""
    return f"run:{thread_id}"