orchestrata_ultimate.knowledge_graph.schema

Reference › orchestrata_ultimate › knowledge_graph › schema

The legacy lookup layer of schema DDL for the knowledge graph (Spec 008, REQ-008-037 and REQ-008-052). Identity and merge keys now live in canonical_schema, whose label constants this module reuses. Here three lists hold Neo4j-syntax statements: CONSTRAINTS (uniqueness on node ids), INDEXES (for example on file path) and FULLTEXT_INDEXES (the search_names index). get_all_ddl returns all of them for a dialect. For "neo4j" it returns the strings as written; for "memgraph", the dialect of the orchgraph engine behind the orchkg database, it translates them with to_memgraph_ddl instead of keeping a second hand-written list, so there is one source. The translator is total over the statement shapes it knows and raises on any it does not, so a new DDL form cannot be dropped silently; it returns None for a statement that has no Memgraph equivalent. An unknown dialect raises ValueError. get_unified_ddl puts the canonical identity DDL first, then this module's statements, deduplicated, and get_node_label maps a node type string to its label. sync, graph_client and cxp_auditor import it.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

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

"""Knowledge Graph schema definition.

Defines constraints, indexes, and schema DDL for the knowledge graph.
"""

import re

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

# Spec: 008 - Knowledge Graph

# Node labels
#
# (L-D4, reports/research-dsl-benefit.md §3.C / §5 D4): every label below
# that canonical_schema.py ALSO declares is a re-export, not a duplicate
# literal — canonical_schema is the single source of truth for the label
# vocabulary. LABEL_UNRESOLVED_CALL/LABEL_UNRESOLVED_CLASS/LABEL_DOCUMENT
# stay hand-authored: canonical_schema.py's own docstring declares Document/
# UnresolvedCall/UnresolvedClass out of its canon (deferred generalization),
# so there is nothing to re-export them from. LABEL_METHOD graduated into
# the canon 2026-08-06 (measured live: Method already carried a real
# UNIQUE constraint) — it is now a re-export like every other canonical
# label here, not hand-authored.
LABEL_PORTFOLIO = canonical_schema.LABEL_PORTFOLIO
LABEL_PROJECT = canonical_schema.LABEL_PROJECT
LABEL_SESSION = canonical_schema.LABEL_SESSION
LABEL_TASK = (
    canonical_schema.LABEL_TASK
)  # FR-001 node type (gap #1); see graph-schema.cql:7,18
LABEL_FILE = canonical_schema.LABEL_FILE
LABEL_MODULE = canonical_schema.LABEL_MODULE
LABEL_PACKAGE = canonical_schema.LABEL_PACKAGE
LABEL_CLASS = canonical_schema.LABEL_CLASS
LABEL_FUNCTION = canonical_schema.LABEL_FUNCTION
LABEL_METHOD = (
    canonical_schema.LABEL_METHOD
)  # written by the AST/code-intel sync; indexed here
LABEL_PREDICATE = canonical_schema.LABEL_PREDICATE
LABEL_RULE = canonical_schema.LABEL_RULE
LABEL_DECISION = canonical_schema.LABEL_DECISION
LABEL_ARTIFACT = canonical_schema.LABEL_ARTIFACT
LABEL_SPEC = canonical_schema.LABEL_SPEC
LABEL_REQUIREMENT = canonical_schema.LABEL_REQUIREMENT
LABEL_REQUIREMENT_ITEM = canonical_schema.LABEL_REQUIREMENT_ITEM
LABEL_DIRECTORY = canonical_schema.LABEL_DIRECTORY
LABEL_ACTION = canonical_schema.LABEL_ACTION
LABEL_UNRESOLVED_CALL = (
    "UnresolvedCall"  # no canonical counterpart (canon defers UnresolvedCall)
)
LABEL_UNRESOLVED_CLASS = (
    "UnresolvedClass"  # no canonical counterpart (canon defers UnresolvedClass)
)
LABEL_DOCUMENT = "Document"  # no canonical counterpart (canon defers Document)
LABEL_CHANGE = canonical_schema.LABEL_CHANGE

# Semantic / runtime domain labels (W5) — the agent + provenance layer the
# code-structure graph lacked. See docs/knowledge-graph.md.
LABEL_VOLK = canonical_schema.LABEL_VOLK
LABEL_INTENT = canonical_schema.LABEL_INTENT
LABEL_PIPELINE_RUN = canonical_schema.LABEL_PIPELINE_RUN
LABEL_COMPONENT = canonical_schema.LABEL_COMPONENT

ALL_LABELS = [
    LABEL_PORTFOLIO,
    LABEL_PROJECT,
    LABEL_SESSION,
    LABEL_TASK,
    LABEL_FILE,
    LABEL_CLASS,
    LABEL_FUNCTION,
    LABEL_PREDICATE,
    LABEL_RULE,
    LABEL_DECISION,
    LABEL_ARTIFACT,
    LABEL_SPEC,
    LABEL_REQUIREMENT,
    LABEL_REQUIREMENT_ITEM,
    LABEL_DIRECTORY,
    LABEL_ACTION,
    LABEL_DOCUMENT,
    LABEL_CHANGE,
    LABEL_VOLK,
    LABEL_INTENT,
    LABEL_PIPELINE_RUN,
    LABEL_COMPONENT,
]

# Relationship types
#
# (L-D4): same re-export discipline as the labels above — REL_* constants
# that canonical_schema.py also declares are re-exports of it, not
# duplicate literals. CONTAINS/CALLS/IMPORTS graduated into the canon
# 2026-09-17 (the code-intelligence relationship vocabulary — see
# canonical_schema.py). The rest (BELONGS_TO, HAS_FILE, OVERRIDES, ...)
# have no canonical counterpart and stay hand-authored.
REL_CONTAINS = canonical_schema.REL_CONTAINS
REL_BELONGS_TO = "BELONGS_TO"
REL_HAS_FILE = "HAS_FILE"
REL_DEFINES = canonical_schema.REL_DEFINES
REL_CALLS = canonical_schema.REL_CALLS
REL_INHERITS = canonical_schema.REL_INHERITS
REL_IMPORTS = canonical_schema.REL_IMPORTS
REL_DEPENDS_ON = canonical_schema.REL_DEPENDS_ON
REL_OVERRIDES = "OVERRIDES"
REL_NEXT_VERSION = canonical_schema.REL_NEXT_VERSION
REL_CHANGED_IN = canonical_schema.REL_CHANGED_IN
REL_HAS_SPEC = "HAS_SPEC"
REL_HAS_ROOT_DIR = "HAS_ROOT_DIR"
REL_HAS_SESSION = "HAS_SESSION"
REL_DEFINES_REQUIREMENT = canonical_schema.REL_DEFINES_REQUIREMENT
REL_HAS_ITEM = canonical_schema.REL_HAS_ITEM
REL_TRACES_TO = canonical_schema.REL_TRACES_TO
REL_IMPLEMENTS = canonical_schema.REL_IMPLEMENTS
REL_MODIFIED = "MODIFIED"
REL_PRODUCED = "PRODUCED"
# Wave 4 retired the writer-specific containment synonyms (HAS_CLASS,
# HAS_FUNCTION, HAS_METHOD). All code-structure containment now uses the
# canonical DEFINES relationship (canonical_schema REL_DEFINES).
REL_CALLS_UNRESOLVED = "CALLS_UNRESOLVED"
REL_INHERITS_UNRESOLVED = "INHERITS_UNRESOLVED"
REL_LOGGED_ACTION = "LOGGED_ACTION"
REL_MADE_DECISION = "MADE_DECISION"
REL_TARGETS = "TARGETS"
REL_HAS_DOC = "HAS_DOC"

# Provenance / ownership / decision edges (W5) — the agent↔work and
# artifact-provenance layer SPEC-008 specified but that had zero occurrences.
REL_ASSIGNED_TO = "ASSIGNED_TO"  # (:Task)-[:ASSIGNED_TO]->(:Volk)
REL_OWNS = "OWNS"  # (:Volk)-[:OWNS]->(:Component|:File)
REL_EXECUTED_BY = "EXECUTED_BY"  # (:PipelineRun)-[:EXECUTED_BY]->(:Volk)
REL_TRIGGERED = "TRIGGERED"  # (:Intent)-[:TRIGGERED]->(:PipelineRun)
REL_GENERATED_BY = "GENERATED_BY"  # (:Artifact)-[:GENERATED_BY]->(:Volk)
REL_DERIVED_FROM = "DERIVED_FROM"  # (:Artifact)-[:DERIVED_FROM]->(:Spec|:Artifact)
REL_MADE = "MADE"  # (:Volk)-[:MADE]->(:Decision)
REL_MADE_BY = "MADE_BY"  # (:Decision)-[:MADE_BY]->(:Volk) — inverse of MADE
REL_APPLIES_TO = "APPLIES_TO"  # (:Decision)-[:APPLIES_TO]->(:Component|:File)
REL_SUPERSEDES = (
    canonical_schema.REL_SUPERSEDES
)  # (:Decision)-[:SUPERSEDES]->(:Decision)
REL_JUSTIFIED_BY = (
    canonical_schema.REL_JUSTIFIED_BY
)  # (:Decision)-[:JUSTIFIED_BY]->(:Requirement)
REL_PART_OF = "PART_OF"  # (:Component)-[:PART_OF]->(:Project)

# FR-002 relationship types missing under any name (gap #3) — vocabulary only,
# writers are tracked separately (gap #2).
REL_WORKED_ON_BY = "WORKED_ON_BY"  # (:Session)-[:WORKED_ON_BY]->(:Volk)
REL_REQUIRES = canonical_schema.REL_REQUIRES  # (:Task)-[:REQUIRES]->(:Component)
REL_OWNED_BY = "OWNED_BY"  # (:Component)-[:OWNED_BY]->(:Volk)
REL_WORKED_ON = "WORKED_ON"  # (:Volk)-[:WORKED_ON]->(:Session)
# Extension beyond the FR-002 list: the implementation's own task-tracking
# edge, written by gateway/handlers/projects.py.
REL_HAS_TASK = "HAS_TASK"  # (:Project)-[:HAS_TASK]->(:Task)

# Constraint DDL
#
# NOTE (2026-07-12): file_path/directory_path UNIQUE constraints were REMOVED
# from here. File.path and Directory.path are project-RELATIVE and
# legitimately collide across projects (worst case: Directory "." is every
# project's root) — a global UNIQUE on path is unsound and collapses distinct
# projects' File/Directory nodes into one, letting the second sync overwrite
# the first's project_id. The canonical identity for File/Directory is now
# the writer's project-qualified `id` ("{project_id}:{path}", same shape as
# Package) — see canonical_schema.py's file_id/directory_id constraints,
# applied via get_unified_ddl() below. Non-unique lookup indexes on path
# remain (see INDEXES) since plenty of reads still filter by path.
CONSTRAINTS: list[str] = [
    # Portfolio/Project identity is owned by the canonical schema
    # (portfolio_name_unique on Portfolio.name, project_slug on Project.slug).
    # Their legacy id constraints were removed here to stop the two-source
    # split-brain where ensure_schema() recreated duplicate constraints.
    f"CREATE CONSTRAINT class_id IF NOT EXISTS FOR (c:{LABEL_CLASS}) REQUIRE c.id IS UNIQUE",
    f"CREATE CONSTRAINT function_id IF NOT EXISTS FOR (f:{LABEL_FUNCTION}) REQUIRE f.id IS UNIQUE",
    f"CREATE CONSTRAINT predicate_id IF NOT EXISTS FOR (p:{LABEL_PREDICATE}) REQUIRE p.id IS UNIQUE",
    f"CREATE CONSTRAINT rule_id IF NOT EXISTS FOR (r:{LABEL_RULE}) REQUIRE r.id IS UNIQUE",
    f"CREATE CONSTRAINT spec_id IF NOT EXISTS FOR (s:{LABEL_SPEC}) REQUIRE s.id IS UNIQUE",
    f"CREATE CONSTRAINT requirement_id IF NOT EXISTS FOR (r:{LABEL_REQUIREMENT}) REQUIRE r.id IS UNIQUE",
    f"CREATE CONSTRAINT req_item_id IF NOT EXISTS FOR (ri:{LABEL_REQUIREMENT_ITEM}) REQUIRE ri.id IS UNIQUE",
    f"CREATE CONSTRAINT action_id IF NOT EXISTS FOR (a:{LABEL_ACTION}) REQUIRE a.id IS UNIQUE",
    f"CREATE CONSTRAINT document_id IF NOT EXISTS FOR (d:{LABEL_DOCUMENT}) REQUIRE d.id IS UNIQUE",
    # FR-001 node type (Task, gap #1) — contract: graph-schema.cql:7 (task_id_unique).
    f"CREATE CONSTRAINT task_id IF NOT EXISTS FOR (t:{LABEL_TASK}) REQUIRE t.id IS UNIQUE",
    # W5 — realize the declared-but-never-created provenance labels and the new
    # semantic labels with identity constraints. Safe to add: no nodes with
    # conflicting keys exist for these labels (unlike Session/Project/Requirement,
    # whose divergent writer keys must be canonicalised first — W5b).
    f"CREATE CONSTRAINT decision_id IF NOT EXISTS FOR (d:{LABEL_DECISION}) REQUIRE d.id IS UNIQUE",
    f"CREATE CONSTRAINT artifact_id IF NOT EXISTS FOR (a:{LABEL_ARTIFACT}) REQUIRE a.id IS UNIQUE",
    f"CREATE CONSTRAINT volk_alias IF NOT EXISTS FOR (v:{LABEL_VOLK}) REQUIRE v.alias IS UNIQUE",
    f"CREATE CONSTRAINT intent_id IF NOT EXISTS FOR (i:{LABEL_INTENT}) REQUIRE i.id IS UNIQUE",
    f"CREATE CONSTRAINT pipeline_run_id IF NOT EXISTS FOR (r:{LABEL_PIPELINE_RUN}) REQUIRE r.id IS UNIQUE",
    # Component canon RE-SCOPED 2026-07-19 (architecture-ontology ruling 2):
    # key = id "{project}:{name}"; the old `component_name` (bare-name)
    # constraint is RETIRED — canonical_conformance drops it, so shipping
    # it here re-created the very drift the kg_schema check flags.
    f"CREATE CONSTRAINT component_id IF NOT EXISTS FOR (c:{LABEL_COMPONENT}) REQUIRE c.id IS UNIQUE",
    f"CREATE CONSTRAINT change_id IF NOT EXISTS FOR (c:{LABEL_CHANGE}) REQUIRE c.id IS UNIQUE",
]

# Index DDL
INDEXES: list[str] = [
    # Non-unique lookup indexes on path (2026-07-12): path is no longer the
    # unique identity for File/Directory (see the CONSTRAINTS note above),
    # but plenty of reads still filter/MATCH by path, so keep it indexed.
    f"CREATE INDEX file_path_idx IF NOT EXISTS FOR (f:{LABEL_FILE}) ON (f.path)",
    f"CREATE INDEX directory_path_idx IF NOT EXISTS FOR (d:{LABEL_DIRECTORY}) ON (d.path)",
    f"CREATE INDEX file_name IF NOT EXISTS FOR (f:{LABEL_FILE}) ON (f.name)",
    f"CREATE INDEX class_name IF NOT EXISTS FOR (c:{LABEL_CLASS}) ON (c.name)",
    f"CREATE INDEX function_name IF NOT EXISTS FOR (f:{LABEL_FUNCTION}) ON (f.name)",
    f"CREATE INDEX predicate_functor IF NOT EXISTS FOR (p:{LABEL_PREDICATE}) ON (p.functor)",
    f"CREATE INDEX spec_title IF NOT EXISTS FOR (s:{LABEL_SPEC}) ON (s.title)",
    f"CREATE INDEX directory_name IF NOT EXISTS FOR (d:{LABEL_DIRECTORY}) ON (d.name)",
    f"CREATE INDEX action_type IF NOT EXISTS FOR (a:{LABEL_ACTION}) ON (a.type)",
    f"CREATE INDEX document_type IF NOT EXISTS FOR (d:{LABEL_DOCUMENT}) ON (d.type)",
    # FR-001 node type (Task, gap #1) — contract: graph-schema.cql:18 (task_status).
    f"CREATE INDEX task_status IF NOT EXISTS FOR (t:{LABEL_TASK}) ON (t.status)",
    "CREATE INDEX node_created IF NOT EXISTS FOR (n:Node) ON (n.created_at)",
    # W5 — temporal/lookup indexes on the semantic labels actually written.
    f"CREATE INDEX intent_created IF NOT EXISTS FOR (i:{LABEL_INTENT}) ON (i.created_at)",
    f"CREATE INDEX pipeline_run_started IF NOT EXISTS FOR (r:{LABEL_PIPELINE_RUN}) ON (r.started_at)",
    f"CREATE INDEX decision_created IF NOT EXISTS FOR (d:{LABEL_DECISION}) ON (d.created_at)",
    f"CREATE INDEX artifact_type_idx IF NOT EXISTS FOR (a:{LABEL_ARTIFACT}) ON (a.type)",
    f"CREATE INDEX volk_status IF NOT EXISTS FOR (v:{LABEL_VOLK}) ON (v.status)",
    f"CREATE INDEX change_timestamp IF NOT EXISTS FOR (c:{LABEL_CHANGE}) ON (c.timestamp)",
    # TEXT indexes back SearchEngine._contains_search's per-label CONTAINS
    # branches (RANGE indexes cannot serve CONTAINS; TEXT can — graph-DB
    # research 2026-07-21, reports/research-graph-db-best-practices.md #2).
    # They coexist with the RANGE name indexes above (one index per type
    # per label/property pair).
    f"CREATE TEXT INDEX class_name_text IF NOT EXISTS FOR (c:{LABEL_CLASS}) ON (c.name)",
    f"CREATE TEXT INDEX function_name_text IF NOT EXISTS FOR (f:{LABEL_FUNCTION}) ON (f.name)",
    f"CREATE TEXT INDEX predicate_name_text IF NOT EXISTS FOR (p:{LABEL_PREDICATE}) ON (p.name)",
    # qualified_name lookups (2026-07-21): the ruling-2 property-based joins —
    # frame.py entity linking and sync_traces_to_graph.py's TRACES_TO writer —
    # match code nodes by qualified_name; without these, per-row MATCHes in
    # UNWIND batches degrade to label scans.
    f"CREATE INDEX class_qualified_name IF NOT EXISTS FOR (c:{LABEL_CLASS}) ON (c.qualified_name)",
    f"CREATE INDEX function_qualified_name IF NOT EXISTS FOR (f:{LABEL_FUNCTION}) ON (f.qualified_name)",
    f"CREATE INDEX method_qualified_name IF NOT EXISTS FOR (m:{LABEL_METHOD}) ON (m.qualified_name)",
    # Method canonical-id + name lookups (Step 3.4, 2026-07-26). Method is the
    # single largest code label (5,561 nodes) and `id` is its canonical,
    # project-qualified identity — at the time this index was added it had no
    # `id` index because Method was out of canonical_schema.py's canon and so
    # got no CONSTRAINT (which is what supplies the implicit index for every
    # other code label); a lookup by canonical id therefore planned
    # `ScanAll (m)` + `Filter`, the exact miss `benchmark_code_intel.py`'s
    # plan gate then failed on. Method graduated into the canon 2026-08-06
    # (see canonical_schema.CANONICAL_MERGE_KEYS), so `method_id`'s
    # constraint now supplies this index too — the explicit one below is kept
    # as a harmless, redundant belt-and-braces index rather than removed
    # mid-wave. `name` mirrors class_name/function_name so symbol search by
    # bare method name is indexed the same way.
    f"CREATE INDEX method_id IF NOT EXISTS FOR (m:{LABEL_METHOD}) ON (m.id)",
    f"CREATE INDEX method_name IF NOT EXISTS FOR (m:{LABEL_METHOD}) ON (m.name)",
]

# Full-text index (requires separate creation syntax)
FULLTEXT_INDEXES: list[str] = [
    "CREATE FULLTEXT INDEX search_names IF NOT EXISTS "
    f"FOR (n:{LABEL_CLASS}|{LABEL_FUNCTION}|{LABEL_PREDICATE}) ON EACH [n.name, n.docstring]",
    # W5 — semantic full-text over the domain layer (decisions, requirements,
    # intents). Neo4j only indexes nodes that actually have the listed properties.
    "CREATE FULLTEXT INDEX search_semantic IF NOT EXISTS "
    f"FOR (n:{LABEL_DECISION}|{LABEL_REQUIREMENT}|{LABEL_INTENT}|{LABEL_SPEC}) "
    "ON EACH [n.rationale, n.title, n.description, n.statement]",
]


@traces(spec="008", story="US-001", reqs=["REQ-008-037", "REQ-008-052"])
@traces(spec="015", story="US-001", reqs=["REQ-015-168"])
def get_all_ddl(dialect: str = "neo4j") -> list[str]:
    """Get all schema DDL statements, rendered for ``dialect``.

    ``"memgraph"`` TRANSLATES the Neo4j statements above rather than keeping
    a second hand-written list: two lists drift, and the drift is silent
    until a re-derive fails on the cluster months later. The Neo4j strings
    stay the single source; :func:`to_memgraph_ddl` is a total function over
    their shapes and RAISES on anything it does not recognise, so a new DDL
    form cannot be quietly dropped from the Memgraph side.
    """
    statements = CONSTRAINTS + INDEXES + FULLTEXT_INDEXES
    if dialect == "neo4j":
        return list(statements)
    if dialect == "memgraph":
        return [
            translated
            for stmt in statements
            if (translated := to_memgraph_ddl(stmt)) is not None
        ]
    raise ValueError(f"unknown DDL dialect: {dialect!r} (neo4j|memgraph)")


# Shapes of the DDL above. Anchored and exhaustive on purpose — see
# `to_memgraph_ddl`.
# The constraint/index NAME and the `IF NOT EXISTS` guard are both OPTIONAL in
# Neo4j's grammar, so they are optional here too: a translator that only
# accepted the fully-spelled form raised `cannot translate DDL to the Memgraph
# dialect` on legitimate Neo4j DDL. That is not hypothetical — it is how
# KGDesignStore._ensure_indexes ("CREATE INDEX IF NOT EXISTS FOR (c:CodeClass)
# ON (c.name)", no name) lost every one of its indexes on the new engine.
_OPT_NAME = r"(?:\s+(?P<name>\w+))?(?:\s+IF NOT EXISTS)?"
_DDL_CONSTRAINT_RE = re.compile(
    r"^CREATE CONSTRAINT" + _OPT_NAME + r" FOR \(\w+:(?P<label>\w+)\) "
    r"REQUIRE \w+\.(?P<prop>\w+) IS UNIQUE$"
)
_DDL_INDEX_RE = re.compile(
    r"^CREATE INDEX" + _OPT_NAME + r" FOR \(\w+:(?P<label>\w+)\) "
    r"ON \(\w+\.(?P<prop>\w+)\)$"
)
_DDL_TEXT_INDEX_RE = re.compile(
    r"^CREATE TEXT INDEX" + _OPT_NAME + r" FOR \(\w+:(?P<label>\w+)\) "
    r"ON \(\w+\.(?P<prop>\w+)\)$"
)
_DDL_FULLTEXT_RE = re.compile(
    r"^CREATE FULLTEXT INDEX(?:\s+\w+)?(?:\s+IF NOT EXISTS)?\s+FOR "
)


def to_memgraph_ddl(statement: str) -> str | None:
    """Translate one Neo4j DDL statement to Memgraph, or None if it has no
    equivalent.

    Grammar verified live against the deployed orchgraph build (2026-07-26),
    not inferred from documentation:

    * ``CREATE TEXT INDEX <name> ON :Label(prop)`` — accepted.
    * the same with ``IF NOT EXISTS`` — rejected, "mismatched input 'IF'".
      Memgraph has no ``IF NOT EXISTS`` on any index/constraint form;
      re-application is tolerated by the appliers' already-exists handling.
    * ``CREATE FULLTEXT INDEX ... ON EACH [...]`` — rejected outright. The
      engine has no full-text index; its text search goes through the text
      index above plus the ``text_search`` query module. Returning None
      (rather than raising) is correct here: the capability is genuinely
      absent, and the TEXT indexes cover the CONTAINS-style lookups
      SearchEngine actually issues.

    Anything else raises, so an unrecognised statement fails loudly at the
    call site instead of silently vanishing from the Memgraph schema.
    """
    if (m := _DDL_CONSTRAINT_RE.match(statement)) is not None:
        return f"CREATE CONSTRAINT ON (n:{m['label']}) ASSERT n.{m['prop']} IS UNIQUE"
    if (m := _DDL_TEXT_INDEX_RE.match(statement)) is not None:
        # Memgraph's TEXT index grammar REQUIRES a name where Neo4j's makes it
        # optional, so an unnamed source statement gets a deterministic one.
        name = m["name"] or f"{m['label'].lower()}_{m['prop']}_text"
        return f"CREATE TEXT INDEX {name} ON :{m['label']}({m['prop']})"
    if (m := _DDL_INDEX_RE.match(statement)) is not None:
        return f"CREATE INDEX ON :{m['label']}({m['prop']})"
    if _DDL_FULLTEXT_RE.match(statement):
        return None
    raise ValueError(f"cannot translate DDL to the Memgraph dialect: {statement!r}")


@traces(spec="106", reqs=["REQ-106-005"])
def get_unified_ddl(
    extra: list[str] | None = None, dialect: str = "neo4j"
) -> list[str]:
    """Canonical identity DDL first, then this module's lookup layer, deduped.

    THE one combined DDL list every schema applier uses
    (``GraphClient.ensure_schema``, ``GraphSyncer.create_schema``): the
    SPEC-106 canonical constraints (generated from ``CANONICAL_MERGE_KEYS``,
    the single identity source) followed by this module's lookup layer
    (legacy extra constraints, lookup indexes, fulltext indexes). Appliers
    with surface-specific needs pass them via ``extra``.

    Args:
        extra: Additional DDL appended after the unified list (deduped too).
            Rendered for ``dialect`` on the same terms as everything else —
            a caller's ``extra`` is Neo4j-shaped DDL like the rest.
        dialect: ``"neo4j"`` (default) or ``"memgraph"`` (orchgraph).

    Returns:
        Ordered, duplicate-free DDL statements.
    """
    # canonical_schema is imported at module scope (L-D4): verified no import
    # cycle (canonical_schema.py only imports core.base.traces; nothing in
    # its 14 other importers is this module) via
    # `serialize_codebase_ast.py who-imports`, 2026-07-21.
    if dialect == "memgraph":
        rendered_extra = [
            translated
            for stmt in (extra or [])
            if (translated := to_memgraph_ddl(stmt)) is not None
        ]
    elif dialect == "neo4j":
        rendered_extra = list(extra or [])
    else:
        raise ValueError(f"unknown DDL dialect: {dialect!r} (neo4j|memgraph)")

    seen: set[str] = set()
    unified: list[str] = []
    for ddl in (
        canonical_schema.get_all_ddl(dialect=dialect)
        + get_all_ddl(dialect=dialect)
        + rendered_extra
    ):
        if ddl not in seen:
            seen.add(ddl)
            unified.append(ddl)
    return unified


@traces(spec="008", story="US-001", reqs=["REQ-008-052"])
@traces(spec="015", story="US-001", reqs=["REQ-015-169"])
def get_node_label(node_type: str) -> str:
    """Map node type string to graph label."""
    mapping = {
        "portfolio": LABEL_PORTFOLIO,
        "project": LABEL_PROJECT,
        "session": LABEL_SESSION,
        "task": LABEL_TASK,
        "file": LABEL_FILE,
        "module": LABEL_MODULE,
        "package": LABEL_PACKAGE,
        "class_def": LABEL_CLASS,
        "function": LABEL_FUNCTION,
        "predicate": LABEL_PREDICATE,
        "rule": LABEL_RULE,
        "decision": LABEL_DECISION,
        "artifact": LABEL_ARTIFACT,
        "spec": LABEL_SPEC,
        "requirement": LABEL_REQUIREMENT,
        "requirement_item": LABEL_REQUIREMENT_ITEM,
        "directory": LABEL_DIRECTORY,
        "action": LABEL_ACTION,
        "document": LABEL_DOCUMENT,
        "change": LABEL_CHANGE,
        "volk": LABEL_VOLK,
        "intent": LABEL_INTENT,
        "pipeline_run": LABEL_PIPELINE_RUN,
        "component": LABEL_COMPONENT,
    }
    return mapping.get(node_type, "Node")