orchestrata_ultimate.knowledge_graph.canonical_schema

Reference › orchestrata_ultimate › knowledge_graph › canonical_schema

canonical_schema is the single source for identity, merge keys and DDL of the operational knowledge graph, the orchkg database on orchgraph (SPEC-106 FR-001). It exists because an inventory found six uncoordinated writers using different merge keys for the same labels (for instance Session keyed three ways), so a query saw different disjoint node sets depending on which writer ran last. The module fixes one key per label and exports it as CANONICAL_MERGE_KEYS, with the LABEL_* and REL_* constants that nearly every other module in the package imports, including the architecture_writer, ontology_mapping, cxp_auditor and the promotion store.

Identity is project-qualified, because the graph holds many projects whose relative paths and qualified names recur: code entities use <project>:<relpath>::<qualname>, built by code_entity_id and project_qualified_id and split by parse_code_entity_id; conversation_entity_id, conversation_turn_entity_id and thread_entity_id do the same for conversation and thread nodes, and test results key on run id plus name. The bare natural id stays as a separate display property. The label set covers code structure, specs, runtime and provenance, testing, semantic-graph-specific labels, design, and knowledge and memory tiers.

Operations are get_all_ddl (constraints and indexes for a neo4j or memgraph dialect; the Memgraph forms have no names and no IF NOT EXISTS), apply (idempotent application to a live repository, dropping plain indexes that a constraint owns), live_unique_constraints and live_index_pairs (read the live schema) and check, a read-only conformance gate returning a ConformanceReport of missing constraints, extra constraints and non-canonical nodes with samples; its ok property is true only when the live graph conforms exactly. The older schema keeps legacy lookup DDL.

At a glance

Class diagram

Dependency diagram

Truncated at 25 nodes: showing 3 of 3 imports and 21 of 35 importers.

API

Constants

Classes

ExtraConstraint (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-006']), dataclass(frozen=True) · Line 1211

A live constraint on a canonical label that uses a non-canonical key.

ConformanceReport (dataclass)

Decorators: traces(spec='106', reqs=['REQ-106-006']), dataclass(frozen=True) · Line 1226

The result of :func:check — drift between canon and the live graph.

Functions

Source

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

"""Canonical Knowledge-Graph schema — the single DDL source (SPEC-106 FR-001).

W8.1's ground-truth inventory (plans/evidence/w8-context-layer-inventory.md)
found SIX uncoordinated KG writers with divergent merge keys for the same
labels: Session keyed on ``session_id`` / ``id`` / ``seq`` (3 ways), Volk on
``alias`` / ``volk_id`` / ``name`` (3 ways), Requirement on ``id`` / ``req_id``
(2 ways) — plus a 3-way constraint split-brain across ``sync.py`` (hardcoded),
``schema.py`` (``LABEL_*`` + ``get_all_ddl()``, zero callers) and
``types_enums.py`` (``NodeType``/``RelationshipType`` StrEnums). A selector
query built on top of that graph sees 0/1/3 disjoint nodes depending on which
writer last ran.

This module is THE canonical identity + DDL source D12 (unified persistence
layer, plans/decisions-2026-07-06.md) references for its P2 phase (single
canonical KG writer). It settles, once, per FR-001:

* the identity scheme for code entities — originally ``<relpath>::<qualname>``,
  now project-qualified: ``<project>:<relpath>::<qualname>`` (plans/
  kg-project-scoped-identity.md, 2026-07-12). The KG holds 100+ projects and
  every project-derived label's natural id (relative paths, qualified names,
  spec/requirement ids, test names) recurs across projects, so a bare natural
  id is unsound as a global MERGE key — File/Directory (W1), Class/Function/
  Method/Parameter/Requirement/Test/TestResult and the Spec/Requirement/
  RequirementItem tier (W2) all key on ``"{project}:{natural id}"`` (or, for
  TestResult, ``"{run_id}:{name}"`` — run-scoped is strictly finer than
  project-scoped and ``run_id`` is already a uuid). The bare natural id is
  kept as a separate display/lookup property (``qualified_name``, ``path``,
  ``natural_id``, ``req_tag``, ``name`` depending on label) alongside the
  qualified ``id``, mirroring the File/Directory precedent (``f.path`` stayed
  bare while ``f.id`` became qualified);
* the ONE merge key per label for the runtime/semantic nodes whose keys had
  diverged (Session, Volk, Project, Requirement/RequirementItem, Intent,
  PipelineRun, Decision, Artifact);
* the canonical containment/test-linkage relationship names (DEFINES,
  TESTED_BY, TRACES_TO);
* an idempotent ``apply()`` and a read-only ``check()`` conformance gate that
  reports drift between the live graph and this canon.

Scope note: this canon has grown from FR-001's original 13 labels (Module/
Class/Function, Spec/Requirement/RequirementItem, Volk/Session/Project/
Intent/PipelineRun/Decision/Artifact) to the full converged label set ratified
by ``plans/semantic-graph-ontology-design.md`` §3.2/§3.4 — Tier A code
structure (Package/File/Directory/Parameter/Predicate/Rule), Tier B specs
(UserStory/Scenario), Tier C runtime/provenance (Portfolio/Task/Component/
Action), Tier D testing (Test/TestRun/TestResult), Tier E semantic-graph-
specific (Tool/SemanticChunk), Tier F design (DesignModule/DesignClass/
Architecture/Pattern), and Tier G knowledge/memory (Fact/Memory). This module
is now THE single label/key/constraint vocabulary source; ``schema.py``'s
``LABEL_*``/``CONSTRAINTS`` remain the live writer surface until a later wave
migrates the writers onto this canon (REQ-106-008). Document/ImportNode/
DecoratorNode/UnresolvedCall/UnresolvedClass stay out of this canon — they
are not semantic-graph node types per the ontology design.

Migrating the six existing writer call sites (sync.py, FileSynchronizer,
pipeline_kg.py, session/kg_sync, kg_design.py) onto this canon, and fixing
ContextInjector's label mismatches (REQ-106-007), are explicitly OUT of scope
for this module — this is the foundation those migrations converge on.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_factory import resolve_kg_dialect

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository

# ---------------------------------------------------------------------------
# Canonical labels — FR-001's original 13 (see module docstring).
# ---------------------------------------------------------------------------

LABEL_MODULE = "Module"
LABEL_CLASS = "Class"
LABEL_FUNCTION = "Function"
LABEL_SPEC = "Spec"
LABEL_REQUIREMENT = "Requirement"
LABEL_REQUIREMENT_ITEM = "RequirementItem"
LABEL_VOLK = "Volk"
LABEL_SESSION = "Session"
LABEL_PROJECT = "Project"
LABEL_INTENT = "Intent"
LABEL_PIPELINE_RUN = "PipelineRun"
LABEL_DECISION = "Decision"
LABEL_ARTIFACT = "Artifact"
LABEL_CHANGE = "Change"

# ---------------------------------------------------------------------------
# Codemod engine labels (SPEC-110 FR-010, REQ-110-030)
# ---------------------------------------------------------------------------

LABEL_CODEMOD_RUN = "CodemodRun"
LABEL_CODEMOD_PASS = "CodemodPass"
LABEL_CODEMOD_FINDING = "CodemodFinding"

# ---------------------------------------------------------------------------
# Canonical labels — ontology-design extension (§3.2, §3.4 of
# plans/semantic-graph-ontology-design.md). Tiers per that doc.
# ---------------------------------------------------------------------------

# Tier A — code structure (from AST sync).
LABEL_PACKAGE = "Package"
LABEL_FILE = "File"
LABEL_DIRECTORY = "Directory"
LABEL_PARAMETER = "Parameter"
LABEL_PREDICATE = "Predicate"
LABEL_RULE = "Rule"

# Method graduated into this canon 2026-08-06: measured live against the
# shared test cluster, Method already carries a real UNIQUE constraint on
# `id` (26 nodes) — hand-rolled by sync.py's SYNC_DDL (``method_id``), not
# generated here, contradicting schema.py's LABEL_METHOD comment ("no
# CONSTRAINT" — stale; SYNC_DDL created one anyway). This constant + its
# CANONICAL_MERGE_KEYS/`_CONSTRAINT_NAMES` entries below REUSE that exact
# live name so this module's generator becomes the SINGLE definer; sync.py's
# hand-rolled copy is removed in the same change. schema.py's own
# ``LABEL_METHOD`` becomes a re-export of this constant (L-D4).
LABEL_METHOD = "Method"

# Tier B — specs & requirements.
LABEL_USER_STORY = "UserStory"
LABEL_SCENARIO = "Scenario"

# Tier C — runtime & provenance.
LABEL_PORTFOLIO = "Portfolio"
LABEL_TASK = "Task"
LABEL_COMPONENT = "Component"
LABEL_ACTION = "Action"

# Tier D — testing.
LABEL_TEST = "Test"
LABEL_TEST_RUN = "TestRun"
LABEL_TEST_RESULT = "TestResult"

# Tier E — semantic-graph-specific.
LABEL_TOOL = "Tool"
LABEL_SEMANTIC_CHUNK = "SemanticChunk"

# Tier F — design (own layer, distinct from Tier A as-built structure).
LABEL_DESIGN_MODULE = "DesignModule"
LABEL_DESIGN_CLASS = "DesignClass"
LABEL_ARCHITECTURE = "Architecture"
LABEL_PATTERN = "Pattern"

# Tier G — knowledge & memory.
LABEL_FACT = "Fact"
LABEL_MEMORY = "Memory"

# Tier K — factory fact-namespace control (SPEC-033 §4, B4a, ADR-0004). The
# ``PromotionStore`` control node for a `(worktree, commit)` fact namespace --
# distinct from ``Fact``/``Memory`` (Tier G, generic knowledge/memory) and
# from the fact NODES it groups (arbitrary caller-supplied labels; see
# ``factory.promotion.store.write_fact``). Session-254 root-connectivity
# ruling: every node must lead to ``(:Project)`` -- ``FactNamespace`` and
# every fact node it owns previously had no edge at all (permanently
# unreachable islands, swept by hand twice in one session before this fix).
# ``REL_HAS_FACT_NAMESPACE``/``REL_HAS_FACT`` below hub-link it the same way
# ``REL_HAS_VOLK``/``REL_HAS_THREAD`` already do for their tiers.
LABEL_FACT_NAMESPACE = "FactNamespace"

# memory/cold_store.py's Memory-authorship node: `MERGE (v:Agent {alias:
# $volk_id})` keys on the same alias vocabulary as `:Volk.alias` but is
# deliberately a DISTINCT label (not a Volk rename) -- this fix scopes to
# "add the missing hub edge to nodes the writer already creates" per the
# session-254 KG-connectivity gate, not a label migration. Session-254
# root-connectivity ruling: every node must lead to ``(:Project)`` --
# ``Agent`` previously had no hub edge at all (same floating-island failure
# mode HAS_VOLK/HAS_FACT_NAMESPACE already fixed for their own tiers).
# REL_HAS_AGENT below closes that gap.
LABEL_AGENT = "Agent"

# Tier H — operational events & messages. Full-fidelity import: these are real,
# identifiable operational records (inter-agent messages, gateway/audit events,
# governance decisions, session lifecycle), NOT churn to discard. Each keys on
# its natural id where it has one, else a derived <session_id>::<timestamp> id.
# Thread is the correlation anchor for this tier (session-254 Option A,
# plans/thread-ontology-design.md): every Tier-H record carries a producer
# thread_id, and (record)-[:IN_THREAD]->(Thread) is how the whole tier
# becomes reachable from the Project hub.
LABEL_MESSAGE = "Message"  # CXPPacket — inter-agent messages
# CXPPacket graduated into this canon 2026-08-06: LABEL_MESSAGE above is the
# ontology_mapping.py CONVERGENCE TARGET for a future, not-yet-executed
# migration of CXPPacket-labeled nodes (that migration is explicitly OUT of
# scope for this module per the docstring) — until it runs, the graph's REAL,
# currently-written label for this concept is CXPPacket (95,176 nodes on the
# shared test cluster, measured 2026-08-06), not Message (0 nodes so far).
# This constant declares that live reality as its own canonical entry,
# keyed the same way (`message_id`) and REUSING cxp_auditor.py's existing
# live constraint name (`cxp_message_id`) so this module's generator becomes
# the single definer; cxp_auditor.py's hand-rolled copy is removed in the
# same change. The resulting Message/CXPPacket duplication (two canonical
# labels for one concept) is a known, pre-existing defect this wave does not
# resolve — it predates this change and requires the writer migration above.
LABEL_CXP_PACKET = "CXPPacket"
LABEL_GATEWAY_EVENT = "GatewayEvent"  # GatewayAuditEvent — gateway audit log
LABEL_AUDIT_ENTRY = "AuditEntry"  # correlated audit entries
LABEL_POLICY_DECISION = "PolicyDecision"  # governance policy decisions
LABEL_EVENT = "Event"  # generic runtime events
LABEL_RESOLUTION = "Resolution"  # conflict resolutions
LABEL_TOOL_CALL = "ToolCall"  # tool invocations
LABEL_VIOLATION = "Violation"  # governance violations
LABEL_TIMEOUT = "Timeout"  # timeout events
LABEL_ESCALATION = "Escalation"  # escalations
LABEL_SESSION_START = "SessionStart"  # session lifecycle — start
LABEL_SESSION_END = "SessionEnd"  # session lifecycle — end
LABEL_THREAD = "Thread"  # correlation anchor for Tier-H operational records

# Tier I — conversation capture (SPEC-021, US-021-007). Conversation/
# ConversationTurn key on a project-qualified id (multi-project KG ruling,
# plans/kg-project-scoped-identity.md, 2026-07-12): the spec's bare
# session_id ("opencode-{seq}") and the producer-generated turn UUID both
# recur across the KG's 100+ projects, so the canonical key is
# "{project}:{natural_id}" / "{project}:{turn_id}" — the bare value is kept
# as a separate display property (natural_id / turn_id), mirroring the
# File/Directory precedent. See plans/contracts/spec-021-ontology.contract.json
# deviations D1-D4 for the full resolution.
LABEL_CONVERSATION = "Conversation"  # conversation-capture session (opencode/...)
LABEL_CONVERSATION_TURN = "ConversationTurn"  # one turn (user/assistant) within it

# Tier J — architecture ontology (RATIFIED 2026-07-19,
# plans/architecture-ontology-draft.md §3). Extends the canon with
# system-design knowledge: components/interfaces/dataflow, requirements-
# driven decisions (ADRs), capacity/cost models, deployment topology, and
# SLOs/risk. §3's REUSE-vs-NEW ruling: Architecture, Component, Pattern,
# Requirement, RequirementItem, Scenario, UserStory and Artifact are
# pre-existing canonical labels above. Component's key was re-scoped to the
# project-qualified id in phase 1b (ruling 2, §9.2 — migration EXECUTED on
# primary + all worker replicas 2026-07-19); Architecture's re-scoping
# remains a separate gated migration, not applied here. Change
# (LABEL_CHANGE, already declared as a FR-001-core label) gains its first
# canonical merge key below (ruling 7, §9.6) — it previously had none. The
# 19 labels here are genuinely new (§3: 21 NEW claimed vs. 19 counted —
# Architecture/Component were miscounted as NEW in that tally; both are
# REUSE+EXT and already existed before this wave).
LABEL_API_CONTRACT = "ApiContract"
LABEL_DATA_ENTITY = "DataEntity"
LABEL_QUALITY_ATTRIBUTE = "QualityAttribute"
LABEL_WAF_PILLAR = "WAFPillar"
LABEL_CONSTRAINT = "Constraint"
LABEL_CAPACITY_MODEL = "CapacityModel"
LABEL_COST_MODEL = "CostModel"
LABEL_COST_LINE_ITEM = "CostLineItem"
LABEL_ENVIRONMENT = "Environment"
LABEL_REGION = "Region"
LABEL_DEPLOYMENT = "Deployment"
LABEL_ARCHITECTURE_DECISION = "ArchitectureDecision"
LABEL_TRADE_OFF = "TradeOff"
LABEL_RISK = "Risk"
LABEL_FAILURE_MODE = "FailureMode"
LABEL_TECHNOLOGY_CHOICE = "TechnologyChoice"
LABEL_CLOUD_SERVICE = "CloudService"
LABEL_SERVICE_LEVEL_OBJECTIVE = "ServiceLevelObjective"
LABEL_ALERT = "Alert"

# Canonical relationship names (REQ-106-004). Containment uses DEFINES
# (never the writer-specific synonyms HAS_CLASS/HAS_FUNCTION/HAS_METHOD seen
# in session/kg_sync); test linkage uses TESTED_BY (never TESTS); TRACES_TO
# always targets RequirementItem, never Requirement directly.
REL_DEFINES = "DEFINES"
REL_TESTED_BY = "TESTED_BY"
REL_TRACES_TO = "TRACES_TO"
REL_CHANGED_IN = "CHANGED_IN"
REL_NEXT_VERSION = "NEXT_VERSION"
REL_INHERITS = "INHERITS"
REL_IMPLEMENTS = "IMPLEMENTS"

# Code-intelligence relationship vocabulary (AST toolkit → graph). These three
# edges were hand-authored literals in knowledge_graph/schema.py and
# .types_enums.py until 2026-09-17, when they graduated into the canon (L-D4
# re-export discipline). CONTAINS is NAMESPACE containment — Project→Package→
# subpackage, a different semantic from DEFINES (module→symbol containment, the
# relationship the containment-synonym wave consolidated HAS_CLASS/HAS_FUNCTION/
# HAS_METHOD into). CALLS/IMPORTS are the call-graph and import-graph edges the
# AST/code-intelligence sync writes; they have no synonym in the DEFINES family
# and so stay first-class here.
REL_CONTAINS = "CONTAINS"
REL_CALLS = "CALLS"
REL_IMPORTS = "IMPORTS"

# Requirement hierarchy: a Spec defines Requirements, each Requirement has
# RequirementItems, and code symbols trace to RequirementItems.
REL_DEFINES_REQUIREMENT = "DEFINES_REQUIREMENT"
REL_HAS_ITEM = "HAS_ITEM"
REL_REQUIRES = "REQUIRES"

# Tier H runtime-signal relationships. A runtime signal (message, event,
# decision, tool call) is linked to the code entities it concerns so that
# Prolog relevance rules can boost recently touched or failing nodes for
# DEBUG tasks.
REL_MENTIONS = "MENTIONS"  # (signal)-[:MENTIONS]->(File|Class|Function)
REL_IN_THREAD = "IN_THREAD"  # (signal)-[:IN_THREAD]->(Thread|Session)
REL_TRIGGERED_BY = "TRIGGERED_BY"  # (signal)-[:TRIGGERED_BY]->(Event)
REL_RESULTED_IN = "RESULTED_IN"  # (signal)-[:RESULTED_IN]->(Decision|Resolution)
REL_FAILED = "FAILED"  # (:TestRun)-[:FAILED]->(node) or (:Event)-[:FAILED]->(node)

# Thread anchor (session-254 Option A, plans/thread-ontology-design.md).
# IN_THREAD above already names Thread as its target — this wave declares
# the label it was always pointing at. HAS_THREAD mirrors the canonical
# Project-[:HAS_SESSION]->Session hub edge. BY_VOLK formalizes the bare
# literal kernel/audit/writers.py:311 already writes (AuditEntry->Volk, 202
# live edges); TO_VOLK is its addressee-side mirror — deliberately NOT
# "DELIVERED_TO", since delivery_status says many were not.
REL_HAS_THREAD = "HAS_THREAD"  # (:Project)-[:HAS_THREAD]->(:Thread)
REL_BY_VOLK = (
    "BY_VOLK"  # (:CXPPacket|:Message|:AuditEntry|:Decision)-[:BY_VOLK]->(:Volk)
)
REL_TO_VOLK = "TO_VOLK"  # (:CXPPacket|:Message)-[:TO_VOLK]->(:Volk)

# Session hub (mirrors HAS_THREAD/HAS_VOLK below): formalizes the
# Project-[:HAS_SESSION]->Session edge already referenced by the HAS_THREAD
# comment above and written as a bare literal by session/kg_sync/
# _stage_session.py, knowledge_graph/kg_design.py, and knowledge_graph/
# schema.py's own (separate) REL_HAS_SESSION -- this is its first
# canonical_schema constant. Session-254 root-connectivity ruling:
# gateway/audit.py, kernel/session_manager.py, and memory/cold_store.py each
# independently MERGE a `:Session` node with no Project hub edge at all
# (permanently unreachable islands -- 25 live Sessions + 60 dependent
# GatewayAuditEvent nodes found by scripts/check_kg_connectivity.py); this
# constant is what closes that gap in all three writers.
REL_HAS_SESSION = "HAS_SESSION"  # (:Project)-[:HAS_SESSION]->(:Session)

# Code-intelligence ownership edge. ``knowledge_graph/sync.py`` has written this
# as a bare literal at 16 sites since the AST sync was ported; it is the last hop
# of the Project->Package->Module->Function|Method->Parameter reachability chain
# that ``_clear_project``'s traversal already assumes exists. Session-254
# root-connectivity ruling: 5044 live ``:Parameter`` nodes are unreachable
# islands because this edge was never written for them.
REL_HAS_PARAMETER = (
    "HAS_PARAMETER"  # (:Function|:Method)-[:HAS_PARAMETER]->(:Parameter)
)

# Spec tier hub edges. HAS_SPEC exists in the separate, NON-canonical
# knowledge_graph/schema.py:99 and types_enums.py:96 -- this is its first
# canonical_schema constant. HAS_STORY/HAS_REQUIREMENT are written as bare
# literals by kg_design.py and semantic_graph/spec_trace_enricher.py. Together
# they are the (:Project)-[:HAS_SPEC]->(:Spec)-[:HAS_STORY]->(:UserStory)
# -[:HAS_REQUIREMENT]->(:Requirement) chain that 183 live nodes hang off with no
# Project hub at its root.
REL_HAS_SPEC = "HAS_SPEC"  # (:Project)-[:HAS_SPEC]->(:Spec)
REL_HAS_STORY = "HAS_STORY"  # (:Spec)-[:HAS_STORY]->(:UserStory)
REL_HAS_REQUIREMENT = (
    "HAS_REQUIREMENT"  # (:UserStory)-[:HAS_REQUIREMENT]->(:Requirement)
)

# Root-connectivity invariant (session-254 ruling: EVERY node must lead to
# the root Project node — no floating subgraphs). HAS_VOLK is the hub edge
# the kernel boot-time Volk sync MERGEs so the volk roster hangs off the
# project like Sessions/Threads do. A volk serving multiple projects gets a
# HAS_VOLK edge from each — reachability, not exclusive ownership.
REL_HAS_VOLK = "HAS_VOLK"  # (:Project)-[:HAS_VOLK]->(:Volk)

# Agent hub (memory/cold_store.py's Memory-authorship node -- see LABEL_AGENT
# above for why this is a distinct label from Volk, not a rename). Mirrors
# HAS_VOLK's convention exactly; closes the same connectivity gap for the
# `:Agent` nodes cold_store.py already creates.
REL_HAS_AGENT = "HAS_AGENT"  # (:Project)-[:HAS_AGENT]->(:Agent)

# Tier K -- factory fact-namespace control (SPEC-033 §4, B4a, ADR-0004).
# HAS_FACT_NAMESPACE mirrors the HAS_VOLK/HAS_THREAD hub-edge convention;
# HAS_FACT hangs each fact node off its owning control node so both are
# reachable from the root Project node (session-254 invariant).
REL_HAS_FACT_NAMESPACE = (
    "HAS_FACT_NAMESPACE"  # (:Project)-[:HAS_FACT_NAMESPACE]->(:FactNamespace)
)
REL_HAS_FACT = "HAS_FACT"  # (:FactNamespace)-[:HAS_FACT]->(fact node)

# Tier I — conversation capture (SPEC-021). HAS_CONVERSATION mirrors the
# existing Project-[:HAS_SESSION]->Session convention; HAS_TURN is REQ-021-067
# verbatim. IN_SESSION / TRIGGERED reuse existing Tier-H vocabulary already in
# use as literal strings elsewhere (orchestrator/graph_audit.py,
# ontology_mapping.py) and are deliberately not re-declared here.
REL_HAS_CONVERSATION = "HAS_CONVERSATION"
REL_HAS_TURN = "HAS_TURN"

# Tier I.5 — codemod engine provenance (SPEC-110 FR-010, REQ-110-030).
# HAS_CODEMOD_RUN mirrors the Project-[:HAS_SESSION]->Session hub convention;
# ABOUT links a finding to the code entity it concerns; PRODUCED_BY traces a
# change back to the pass that created it; BREAKING_CHANGE marks incompatible
# API mutations. CHANGED_IN (declared above, line ~232) is reused as-is.
REL_HAS_CODEMOD_RUN = "HAS_CODEMOD_RUN"  # (:Project)-[:HAS_CODEMOD_RUN]->(:CodemodRun)
REL_HAS_PASS = "HAS_PASS"  # (:CodemodRun)-[:HAS_PASS]->(:CodemodPass)
REL_HAS_FINDING = "HAS_FINDING"  # (:CodemodRun)-[:HAS_FINDING]->(:CodemodFinding)
REL_ABOUT = "ABOUT"  # (:CodemodFinding)-[:ABOUT]->(:Module|:Class|:Function)
REL_PRODUCED_BY = "PRODUCED_BY"  # (:CodeChange)-[:PRODUCED_BY]->(:CodemodPass)
REL_BREAKING_CHANGE = (
    "BREAKING_CHANGE"  # (:CodeChange)-[:BREAKING_CHANGE]->(:Module|:Class|:Function)
)

# Tier J — architecture ontology (RATIFIED 2026-07-19,
# plans/architecture-ontology-draft.md §4). REL_IMPLEMENTS and REL_TRACES_TO
# above already cover two of the ontology's six REUSE rows (IMPLEMENTS,
# TRACES_TO) verbatim — no new constant needed for those. The other four
# REUSE rows (HAS_COMPONENT, MUST_FOLLOW, JUSTIFIED_BY, SUPERSEDES)
# formalize an edge name a legacy writer (sync_architecture_to_graph.py) or
# an existing provenance convention already used as a bare string literal —
# they get their first REL_* constant here. REL_USES_PATTERN is the writer's
# legacy synonym for IMPLEMENTS/APPLIES, kept only until writers converge
# (§7) — new code should prefer REL_APPLIES/REL_IMPLEMENTS.
REL_HAS_ARCHITECTURE = "HAS_ARCHITECTURE"
REL_HAS_COMPONENT = "HAS_COMPONENT"
REL_EXPOSES = "EXPOSES"
REL_DATA_FLOWS_TO = "DATA_FLOWS_TO"
REL_DEPENDS_ON = "DEPENDS_ON"
REL_PERSISTS_TO = "PERSISTS_TO"
REL_RELATES_TO = "RELATES_TO"
REL_RUNS_ON = "RUNS_ON"
REL_HAS_SCENARIO = "HAS_SCENARIO"
REL_SATISFIES = "SATISFIES"
REL_HAS_QUALITY_ATTRIBUTE = "HAS_QUALITY_ATTRIBUTE"
REL_MEASURED_ON = "MEASURED_ON"
REL_ADDRESSES_PILLAR = "ADDRESSES_PILLAR"
REL_MUST_FOLLOW = "MUST_FOLLOW"
REL_CONSTRAINS = "CONSTRAINS"
REL_MODELED_BY = "MODELED_BY"
REL_INFORMS = "INFORMS"
REL_HAS_LINE_ITEM = "HAS_LINE_ITEM"
REL_METERS = "METERS"
REL_DECIDES = "DECIDES"
REL_CHOSE = "CHOSE"
REL_REJECTED = "REJECTED"
REL_JUSTIFIED_BY = "JUSTIFIED_BY"
REL_WEIGHS = "WEIGHS"
REL_SUPERSEDES = "SUPERSEDES"
REL_SELECTS = "SELECTS"
REL_FORCED_BY = "FORCED_BY"
REL_APPLIES = "APPLIES"
REL_USES_PATTERN = (
    "USES_PATTERN"  # deprecated synonym — see §7, converge onto APPLIES/IMPLEMENTS
)
REL_THREATENS = "THREATENS"
REL_AFFECTS = "AFFECTS"
REL_MITIGATED_BY = "MITIGATED_BY"
REL_HAS_SLO = "HAS_SLO"
REL_MEASURES = "MEASURES"
REL_ALERTS_ON = "ALERTS_ON"
REL_DEPLOYED_AS = "DEPLOYED_AS"
REL_IN_ENVIRONMENT = "IN_ENVIRONMENT"
REL_IN_REGION = "IN_REGION"
REL_PROPOSES = "PROPOSES"
REL_MODIFIES = "MODIFIES"
REL_DOCUMENTS = "DOCUMENTS"
REL_REFERENCES = "REFERENCES"

# ---------------------------------------------------------------------------
# DERIVED labels — re-generated from source, never hand-copied.
# ---------------------------------------------------------------------------
#
# Ratified 2026-07-26 (plans/neo4j-to-orchgraph-migration-plan.md Step 3.3).
# These labels are PRODUCED by parsing the repository — the AST sync for code
# structure, `code_parser.py` for the Prolog sources (Predicate/Rule). Nothing
# that reads them is a record of fact; re-running the parse reproduces them
# exactly.
#
# Two consumers, one definition, on purpose:
#   * the Neo4j -> orchgraph migration EXCLUDES them (copying derived data
#     violates bootstrap doctrine — Step 3.3 re-derives them instead), and
#   * `knowledge_graph.sync._clear_project` DELETES them before a full sync
#     (they are exactly what that sync rebuilds).
# Those two sets must be identical or the migration copies something the sync
# then wipes, or the sync wipes something only the migration could restore.
# Restating the list in either place is how they drift.
DERIVED_LABELS: frozenset[str] = frozenset(
    {
        LABEL_PACKAGE,
        LABEL_MODULE,
        LABEL_CLASS,
        LABEL_FUNCTION,
        LABEL_METHOD,  # schema.py LABEL_METHOD re-exports this (graduated 2026-08-06)
        LABEL_PARAMETER,
        LABEL_FILE,
        LABEL_DIRECTORY,
        LABEL_PREDICATE,
        LABEL_RULE,
        LABEL_TEST,
        "CodeFunction",  # legacy kg_design.py RTM-linking scheme
        "CodeClass",  # legacy kg_design.py RTM-linking scheme
    }
)


# ---------------------------------------------------------------------------
# Canonical merge keys (REQ-106-001..003) — ONE key per label, everywhere.
# ---------------------------------------------------------------------------
#
# | Label           | Canonical key | Divergent keys this replaces           |
# |------------------|--------------|-----------------------------------------|
# | Module           | id           | id = <project>:module:<path> (2026-07-12 project-qualified) |
# | Class            | id           | id = <project>:<relpath>::<qualname> (2026-07-12 project-qualified; was bare) |
# | Function         | id           | methods too, +class_id; id = <project>:<relpath>::<qualname> (2026-07-12 project-qualified; was bare) |
# | Spec             | id           | id = <project>:<spec id> (2026-07-12 project-qualified; was bare, e.g. "023") |
# | Requirement      | id           | req_id (kernel/pipeline_kg.py, knowledge_graph/kg_design.py); id = <project>:<natural id> (2026-07-12 project-qualified — shape varies by writer, see writer docstrings) |
# | RequirementItem  | id           | id = <project>:<item id> (2026-07-12 project-qualified; was bare) |
# | Volk             | alias        | volk_id (kernel/audit/writers.py), name (harness/history.py) |
# | Session          | session_id   | id (memory/cold_store.py), seq (session/kg_sync/_stage_session.py) |
# | Project          | slug         | id (session/kg_sync/_stage_session.py)  |
# | Intent           | id           | (uncontested)                           |
# | PipelineRun      | id           | (uncontested)                           |
# | Decision         | id           | (uncontested)                           |
# | Artifact         | id           | (uncontested)                           |
CANONICAL_MERGE_KEYS: dict[str, str] = {
    LABEL_MODULE: "id",
    LABEL_CLASS: "id",
    LABEL_FUNCTION: "id",
    LABEL_SPEC: "id",
    LABEL_REQUIREMENT: "id",
    LABEL_REQUIREMENT_ITEM: "id",
    LABEL_VOLK: "alias",
    LABEL_SESSION: "session_id",
    LABEL_PROJECT: "slug",
    LABEL_INTENT: "id",
    LABEL_PIPELINE_RUN: "id",
    LABEL_DECISION: "id",
    LABEL_ARTIFACT: "id",
}

# Ontology-design extension (plans/semantic-graph-ontology-design.md §3.4) —
# one merge key per label, same discipline as the FR-001 core above.
#
# | Label        | Canonical key | Note                                     |
# |--------------|---------------|-------------------------------------------|
# | Package      | id            | writer key "{project}:{path}" — bare path collides across projects (2026-07-12) |
# | File         | id            | writer key "{project_id}:{path}" — bare relative path collides across projects, same shape as Package (2026-07-12) |
# | Directory    | id            | writer key "{project_id}:{path}" — Directory "." (every project's ROOT) is the worst-case bare-path collision (2026-07-12) |
# | Parameter    | id            | writer key "{project}:{parent_qualname}:param:{i}" — project-qualified (2026-07-12), no prior writer to diverge from |
# | Predicate    | id            | matches the live schema.py predicate_id   |
# | Rule         | id            | matches the live schema.py rule_id        |
# | UserStory    | id            | writer key "{project}:{spec id}:{story id}" where the class has one (knowledge_graph/kg_design.py, 2026-07-12 project-qualified); no other writer |
# | Scenario     | id            | (uncontested; no prior writer)            |
# | Portfolio    | name          | RECONCILIATION (resolved to SPEC): specs/008.../graph-schema.cql declares portfolio_name_unique (Portfolio.name) — the spec contract is authoritative, so canon = name and REUSES that contract constraint name. schema.py's live portfolio_id (Portfolio.id) is the legacy outlier check() will flag for migration (0 live nodes, clean). Design doc §3.4's earlier `slug` was corrected to match the spec. |
# | Task         | id            | matches the live schema.py task_id (REUSED verbatim — do not invent a new name/key) |
# | Component    | id            | RE-SCOPED (ratified 2026-07-19, plans/architecture-ontology-draft.md §9.2, ruling 2): UNIFIES the design-time Architecture Component with the runtime-ownership Component (`(volk)-[:OWNS]->(:Component)`) as one node; canon = id = "{project}:{name}", the same Package/File/Directory project-qualification precedent — bare `name` (the prior canon) is unsound once a second project's Component can collide on the same top-level module name. `name` is kept as the bare display property. The 5 live Component nodes (Gateway/Kernel/Neo4j/Redis/VizInt UI, all `orchestrata-ultimate`'s own — verified unambiguous by `scripts/canonical_conformance.py`) migrate via that script, which backfills id/project and drops the retired `component_name` constraint. |
# | Action       | id            | matches the live schema.py action_id      |
# | Test         | id            | converges TestNode/DLTest; writer key "{project}:{qualname}" (2026-07-12 project-qualified; was bare) |
# | TestRun      | id            | (uncontested; no prior writer)            |
# | TestResult   | id            | RECONCILIATION: knowledge_graph/test_results_store.py's live writer MERGEd on bare `name` (COLLIDES across runs AND projects) — canon = id = "{run_id}:{name}" (run-scoped is strictly finer than project-scoped and TestRun.run_id is already a uuid), 2026-07-12. `name` kept as a plain property for the COVERS/get_tests_covering lookups. |
# | Tool         | name          | (new — sourced from tools.registry, not the graph) |
# | SemanticChunk| id            | (new — unifies SemanticChunk/CodestralChunk/QwenChunk/Chunk) |
# | DesignModule | id            | (new — Tier F design layer)               |
# | DesignClass  | id            | (new — Tier F design layer)               |
# | Architecture | id            | (new — Tier F design layer)               |
# | Pattern      | name          | (new — Tier F design layer)               |
# | Fact         | id            | (new — Tier G knowledge/memory)           |
# | Memory       | id            | (new — Tier G knowledge/memory)           |
CANONICAL_MERGE_KEYS.update(
    {
        # Package key is the WRITER's project-qualified id ("{project}:{path}"),
        # NOT the bare dotted path: relative paths ("pkg", "core", "utils")
        # legitimately collide across projects in the shared graph, so a global
        # UNIQUE on path is unsound — proven by parallel-suite constraint
        # failures 2026-07-12 (two tmp projects both syncing a package "pkg").
        # This also makes Package consistent with its Module/Class/Function
        # siblings, whose canonical keys are qualified ids.
        LABEL_PACKAGE: "id",
        # File/Directory keys are the WRITER's project-qualified id
        # ("{project_id}:{path}"), NOT the bare relative path: two projects
        # legitimately share relative paths (README.md, src/__init__.py,
        # and — worst case — Directory "." IS every project's root), so a
        # global UNIQUE on path collapsed every project's root Directory
        # into one node and the second sync silently overwrote the first's
        # project_id (multi-project KG ruling, 2026-07-12). Mirrors the
        # Package fix (commit 7eee8a46) — same shape, same reasoning.
        LABEL_FILE: "id",
        LABEL_DIRECTORY: "id",
        LABEL_PARAMETER: "id",
        LABEL_PREDICATE: "id",
        LABEL_RULE: "id",
        LABEL_USER_STORY: "id",
        LABEL_SCENARIO: "id",
        LABEL_PORTFOLIO: "name",
        LABEL_TASK: "id",
        LABEL_COMPONENT: "id",
        LABEL_ACTION: "id",
        LABEL_TEST: "id",
        LABEL_TEST_RUN: "id",
        LABEL_TEST_RESULT: "id",
        LABEL_TOOL: "name",
        LABEL_SEMANTIC_CHUNK: "id",
        LABEL_DESIGN_MODULE: "id",
        LABEL_DESIGN_CLASS: "id",
        LABEL_ARCHITECTURE: "id",
        LABEL_PATTERN: "name",
        LABEL_FACT: "id",
        LABEL_MEMORY: "id",
    }
)

# Tier H — operational events & messages. Natural ids where present (message_id,
# entry_id, resolution_id); the shapeless typed-events ({data, session_id?,
# timestamp}) key on ``id``, which the importer derives as <session_id>::<timestamp>
# (or <timestamp> when there is no session) and writes onto the node.
CANONICAL_MERGE_KEYS.update(
    {
        LABEL_MESSAGE: "message_id",
        LABEL_GATEWAY_EVENT: "entry_id",
        LABEL_AUDIT_ENTRY: "entry_id",
        LABEL_RESOLUTION: "resolution_id",
        LABEL_POLICY_DECISION: "id",
        LABEL_EVENT: "id",
        LABEL_TOOL_CALL: "id",
        LABEL_VIOLATION: "id",
        LABEL_TIMEOUT: "id",
        LABEL_ESCALATION: "id",
        LABEL_SESSION_START: "id",
        LABEL_SESSION_END: "id",
        LABEL_THREAD: "id",
    }
)

# Tier I — conversation capture (SPEC-021). Both new labels key on the
# project-qualified ``id`` produced by conversation_entity_id() /
# conversation_turn_entity_id() below — never the bare natural id.
CANONICAL_MERGE_KEYS.update(
    {
        LABEL_CONVERSATION: "id",
        LABEL_CONVERSATION_TURN: "id",
    }
)

# Tier J — architecture ontology (RATIFIED 2026-07-19,
# plans/architecture-ontology-draft.md §2/§3). Project-scoped labels key on
# ``id`` (ruling 2: ``"{project}:{natural}"``, the same File/Directory/
# Package precedent as every other project-scoped tier above); the three
# genuinely global reference-data labels (ruling 3: shared vocabulary that
# means the same thing in every one of the KG's 100+ projects) key on their
# natural global identifier instead: WAFPillar on its fixed pillar ``name``
# (5 nodes total), Region on its cloud ``code``, CloudService on its vendor
# ``name``. LABEL_CHANGE gains its first canonical key here (ruling 7,
# §9.6): ``LABEL_CHANGE`` was declared with the FR-001 core labels above but
# had NO merge key until now.
CANONICAL_MERGE_KEYS.update(
    {
        LABEL_CHANGE: "id",
        LABEL_API_CONTRACT: "id",
        LABEL_DATA_ENTITY: "id",
        LABEL_QUALITY_ATTRIBUTE: "id",
        LABEL_WAF_PILLAR: "name",
        LABEL_CONSTRAINT: "id",
        LABEL_CAPACITY_MODEL: "id",
        LABEL_COST_MODEL: "id",
        LABEL_COST_LINE_ITEM: "id",
        LABEL_ENVIRONMENT: "id",
        LABEL_REGION: "code",
        LABEL_DEPLOYMENT: "id",
        LABEL_ARCHITECTURE_DECISION: "id",
        LABEL_TRADE_OFF: "id",
        LABEL_RISK: "id",
        LABEL_FAILURE_MODE: "id",
        LABEL_TECHNOLOGY_CHOICE: "id",
        LABEL_CLOUD_SERVICE: "name",
        LABEL_SERVICE_LEVEL_OBJECTIVE: "id",
        LABEL_ALERT: "id",
    }
)

# Tier K -- factory fact-namespace control (SPEC-033 §4, B4a, ADR-0004). The
# control node's identity is the pair (worktree, commit); PromotionStore
# already derives and stores the unique `key` property ("{worktree}@{commit}",
# see FactNamespace.key) on every control node it MERGEs, so that -- not the
# composite pair -- is the canonical single-property merge key.
CANONICAL_MERGE_KEYS.update(
    {
        LABEL_FACT_NAMESPACE: "key",
    }
)

# Agent (memory/cold_store.py's Memory-authorship node -- see LABEL_AGENT
# above). Canonical key mirrors Volk's own `alias` key: cold_store.py's live
# writer already MERGEs on `alias`, so this is a REUSE of the existing write
# pattern, not a new convention.
CANONICAL_MERGE_KEYS.update(
    {
        LABEL_AGENT: "alias",
    }
)

# CXPPacket + Method (measured live 2026-08-06 against the shared test
# cluster, per LABEL_CXP_PACKET's and LABEL_METHOD's comments above): both
# already carry a real UNIQUE constraint, hand-rolled outside this module
# (cxp_auditor.py / sync.py's SYNC_DDL respectively). REUSING those exact
# live constraint names in `_CONSTRAINT_NAMES` below converges each onto
# this canon's single generator instead of a second, independent definer of
# the same fact — the hand-rolled copies are removed in the same change.
CANONICAL_MERGE_KEYS.update(
    {
        LABEL_CXP_PACKET: "message_id",
        LABEL_METHOD: "id",
    }
)

CANONICAL_LABELS: frozenset[str] = frozenset(CANONICAL_MERGE_KEYS)

# Explicit constraint names — matched to the names already live on the shared
# graph for labels that were already canonical (module_id, class_id,
# function_id, spec_id, requirement_id, req_item_id, volk_alias, intent_id,
# pipeline_run_id, decision_id, artifact_id all pre-exist via schema.py/
# sync.py) so apply() does not create a second, redundant constraint under a
# new name. Session and Project get new names since no unique constraint on
# their canonical key exists yet (only bare, non-unique indexes do).
_CONSTRAINT_NAMES: dict[str, str] = {
    LABEL_MODULE: "module_id",
    LABEL_CLASS: "class_id",
    LABEL_FUNCTION: "function_id",
    LABEL_SPEC: "spec_id",
    LABEL_REQUIREMENT: "requirement_id",
    LABEL_REQUIREMENT_ITEM: "req_item_id",
    LABEL_VOLK: "volk_alias",
    LABEL_SESSION: "session_session_id",
    LABEL_PROJECT: "project_slug",
    LABEL_INTENT: "intent_id",
    LABEL_PIPELINE_RUN: "pipeline_run_id",
    LABEL_DECISION: "decision_id",
    LABEL_ARTIFACT: "artifact_id",
}

# Ontology-design extension — REUSE the existing constraint name whenever the
# canonical key matches an already-live/contract constraint.
# predicate_id/rule_id/action_id/task_id are REUSED verbatim (identical key
# already live in schema.py's CONSTRAINTS); portfolio_name_unique is REUSED
# from the SPEC-008 contract (graph-schema.cql) — the authoritative Portfolio
# key. The legacy schema.py portfolio_id (Portfolio.id) is an outlier check()
# will flag for migration. file_id/directory_id/component_id are NEW names:
# file_id/directory_id retire the old file_path/directory_path names (unsound
# bare-path key; see the File/Directory comment above and schema.py's
# CONSTRAINTS, which drops those two statements outright); component_id
# retires the live component_name constraint (ratified 2026-07-19, re-scoping
# Component to a project-qualified id, plans/architecture-ontology-draft.md
# §9.2 ruling 2 — scripts/canonical_conformance.py drops component_name).
# All other labels below have no live constraint at all (first constraint for
# that label).
_CONSTRAINT_NAMES.update(
    {
        # REUSED from the live writer's SYNC_DDL (sync.py MERGEs Package{id}
        # and already creates this constraint under this name).
        LABEL_PACKAGE: "package_id",
        # NEW names — retiring file_path/directory_path (unsound bare-path
        # key). Do NOT reuse those names for anything else.
        LABEL_FILE: "file_id",
        LABEL_DIRECTORY: "directory_id",
        LABEL_PARAMETER: "parameter_id",
        LABEL_PREDICATE: "predicate_id",
        LABEL_RULE: "rule_id",
        LABEL_USER_STORY: "user_story_id",
        LABEL_SCENARIO: "scenario_id",
        LABEL_PORTFOLIO: "portfolio_name_unique",
        LABEL_TASK: "task_id",
        LABEL_COMPONENT: "component_id",
        LABEL_ACTION: "action_id",
        LABEL_TEST: "test_id",
        LABEL_TEST_RUN: "test_run_id",
        LABEL_TEST_RESULT: "test_result_id",
        LABEL_TOOL: "tool_name",
        LABEL_SEMANTIC_CHUNK: "semantic_chunk_id",
        LABEL_DESIGN_MODULE: "design_module_id",
        LABEL_DESIGN_CLASS: "design_class_id",
        LABEL_ARCHITECTURE: "architecture_id",
        LABEL_PATTERN: "pattern_name",
        LABEL_FACT: "fact_id",
        LABEL_MEMORY: "memory_id",
    }
)

# Tier H — operational events & messages (all new; no live constraint yet).
_CONSTRAINT_NAMES.update(
    {
        LABEL_MESSAGE: "message_message_id",
        LABEL_GATEWAY_EVENT: "gateway_event_entry_id",
        LABEL_AUDIT_ENTRY: "audit_entry_entry_id",
        LABEL_RESOLUTION: "resolution_resolution_id",
        LABEL_POLICY_DECISION: "policy_decision_id",
        LABEL_EVENT: "event_id",
        LABEL_TOOL_CALL: "tool_call_id",
        LABEL_VIOLATION: "violation_id",
        LABEL_TIMEOUT: "timeout_id",
        LABEL_ESCALATION: "escalation_id",
        LABEL_SESSION_START: "session_start_id",
        LABEL_SESSION_END: "session_end_id",
        LABEL_THREAD: "thread_id",
    }
)

# Tier I — conversation capture (SPEC-021, all new; no live constraint yet).
_CONSTRAINT_NAMES.update(
    {
        LABEL_CONVERSATION: "conversation_id",
        LABEL_CONVERSATION_TURN: "conversation_turn_id",
    }
)

# Tier J — architecture ontology (RATIFIED 2026-07-19,
# plans/architecture-ontology-draft.md §5, all new; no live constraint
# yet). Names match the DDL block in §5 verbatim so this canon and the
# ratified draft can never silently drift apart.
_CONSTRAINT_NAMES.update(
    {
        LABEL_CHANGE: "change_id",  # fills the LABEL_CHANGE gap (ruling 7, §9.6)
        LABEL_API_CONTRACT: "api_contract_id",
        LABEL_DATA_ENTITY: "data_entity_id",
        LABEL_QUALITY_ATTRIBUTE: "quality_attribute_id",
        LABEL_WAF_PILLAR: "waf_pillar_name",
        LABEL_CONSTRAINT: "constraint_id",
        LABEL_CAPACITY_MODEL: "capacity_model_id",
        LABEL_COST_MODEL: "cost_model_id",
        LABEL_COST_LINE_ITEM: "cost_line_item_id",
        LABEL_ENVIRONMENT: "environment_id",
        LABEL_REGION: "region_code",
        LABEL_DEPLOYMENT: "deployment_id",
        LABEL_ARCHITECTURE_DECISION: "architecture_decision_id",
        LABEL_TRADE_OFF: "tradeoff_id",
        LABEL_RISK: "risk_id",
        LABEL_FAILURE_MODE: "failure_mode_id",
        LABEL_TECHNOLOGY_CHOICE: "technology_choice_id",
        LABEL_CLOUD_SERVICE: "cloud_service_name",
        LABEL_SERVICE_LEVEL_OBJECTIVE: "slo_id",
        LABEL_ALERT: "alert_id",
    }
)

# Tier K -- factory fact-namespace control (SPEC-033 §4, B4a; new; no live
# constraint yet).
_CONSTRAINT_NAMES.update(
    {
        LABEL_FACT_NAMESPACE: "fact_namespace_key",
    }
)

# Agent -- new; no live constraint yet (cold_store.py's write path has never
# had a uniqueness constraint backing its Agent{alias} MERGE key).
_CONSTRAINT_NAMES.update(
    {
        LABEL_AGENT: "agent_alias",
    }
)

# CXPPacket + Method -- both REUSE the exact name their pre-existing,
# now-retired hand-rolled definer already used live, so apply() does not
# create a second, redundant constraint under a new name (same discipline as
# the FR-001-core REUSE block above).
_CONSTRAINT_NAMES.update(
    {
        # REUSED from cxp_auditor.py's SpikeAuditor._ensure_indexes.
        LABEL_CXP_PACKET: "cxp_message_id",
        # REUSED from sync.py's GraphSyncer.SYNC_DDL.
        LABEL_METHOD: "method_id",
    }
)


@traces(spec="106", reqs=["REQ-106-001"])
def code_entity_id(relpath: str, qualname: str) -> str:
    """Canonical identity for Module/Class/Function nodes.

    Format: ``<relpath>::<qualname>``. Methods are Function nodes too — they
    carry a separate ``class_id`` property pointing at their owning Class's
    id; they do NOT get a distinct id scheme.

    Args:
        relpath: Path to the source file, relative to the project root.
        qualname: Fully-qualified name of the module/class/function within
            that file (e.g. ``"MyClass.my_method"``).

    Returns:
        The canonical node id.

    Raises:
        ValueError: If either part is empty or contains the ``::`` separator.
    """
    if not relpath or not qualname:
        raise ValueError("relpath and qualname must both be non-empty")
    if "::" in relpath or "::" in qualname:
        raise ValueError(
            f"relpath/qualname must not contain the '::' separator: "
            f"{relpath!r}, {qualname!r}"
        )
    return f"{relpath}::{qualname}"


@traces(spec="106", reqs=["REQ-106-001"])
def parse_code_entity_id(entity_id: str) -> tuple[str, str]:
    """Inverse of :func:`code_entity_id` — split a canonical id back apart.

    Args:
        entity_id: A canonical ``<relpath>::<qualname>`` id.

    Returns:
        ``(relpath, qualname)``.

    Raises:
        ValueError: If ``entity_id`` does not contain the ``::`` separator.
    """
    relpath, sep, qualname = entity_id.partition("::")
    if not sep:
        raise ValueError(
            f"not a canonical code-entity id (missing '::'): {entity_id!r}"
        )
    return relpath, qualname


@traces(spec="106", reqs=["REQ-106-001"])
def project_qualified_id(project: str, qualified_name: str) -> str:
    """Canonical project-qualified id: ``<project>:<qualified_name>``.

    This is the SHARED scheme behind every ``knowledge_graph.sync.GraphSyncer``
    Class/Function/Method node id (``_sync_classes``/``_sync_functions``/
    ``_sync_methods``: ``f"{project}:{qname}"`` where ``qname`` is the AST's
    ``qualified_name`` or a ``f"{module_path}.{name}"`` fallback) -- extracted
    here so every writer that needs a Function/Class id in this shape calls
    the SAME function instead of re-deriving the ``f"{project}:{...}"``
    concatenation inline, which is what let ``test_results_store.py``'s
    COVERS writer drift onto a bare, non-project-qualified key in the first
    place (2026-07-22 MERGE-key audit).

    Args:
        project: ``Project.slug`` this entity belongs to.
        qualified_name: The already-qualified name (e.g. a dotted
            ``pkg.module.Class.method`` path, or -- for callers with no AST
            dict to derive one from -- any pre-existing qualified string).

    Returns:
        The canonical node id.

    Raises:
        ValueError: If either part is empty, or ``project`` contains ``:``.
    """
    if not project or not qualified_name:
        raise ValueError("project and qualified_name must both be non-empty")
    if ":" in project:
        raise ValueError(f"project must not contain ':': {project!r}")
    return f"{project}:{qualified_name}"


@traces(spec="021", story="US-021-007", reqs=["REQ-021-065"])
def conversation_entity_id(project: str, natural_id: str) -> str:
    """Canonical identity for ``Conversation`` nodes: ``<project>:<natural_id>``.

    ``natural_id`` is the producer's bare id (e.g. ``"opencode-227"``), kept
    separately as the ``natural_id`` display property — the same validation
    discipline as :func:`code_entity_id`.

    Args:
        project: ``Project.slug`` this conversation belongs to.
        natural_id: The producer's bare conversation id.

    Returns:
        The canonical node id.

    Raises:
        ValueError: If either part is empty, or ``project`` contains ``:``.
    """
    if not project or not natural_id:
        raise ValueError("project and natural_id must both be non-empty")
    if ":" in project:
        raise ValueError(f"project must not contain ':': {project!r}")
    return f"{project}:{natural_id}"


@traces(spec="106", reqs=["REQ-106-003"])
def thread_entity_id(project: str, thread_id: str) -> str:
    """Canonical identity for ``Thread`` nodes: ``<project>:<thread_id>``.

    ``thread_id`` is the producer's bare correlation token (e.g. ``"t1"``,
    ``"resume-dispatch"``, ``"pipeline_aae21620"``), kept separately as the
    ``thread_id`` display property — the same validation discipline as
    :func:`conversation_entity_id`. Thread is project-qualified, not global
    (session-254 Option A, ``plans/thread-ontology-design.md``): a thread id
    is a per-run correlation token, and 34% of live values are non-UUID
    tokens that will recur once a second project's kernel emits its own
    ``t1`` or ``pipeline_<8hex>``.

    Args:
        project: ``Project.slug`` this thread belongs to.
        thread_id: The producer's bare thread/correlation id.

    Returns:
        The canonical node id.

    Raises:
        ValueError: If either part is empty, or ``project`` contains ``:``.
    """
    if not project or not thread_id:
        raise ValueError("project and thread_id must both be non-empty")
    if ":" in project:
        raise ValueError(f"project must not contain ':': {project!r}")
    return f"{project}:{thread_id}"


@traces(spec="021", story="US-021-007", reqs=["REQ-021-065"])
def conversation_turn_entity_id(project: str, turn_id: str) -> str:
    """Canonical identity for ``ConversationTurn`` nodes: ``<project>:<turn_id>``.

    ``turn_id`` is the producer-generated UUID (deterministic per queue
    message), which is what makes re-consuming the same message MERGE onto
    the same node — the idempotency requirement REQ-021-065 relies on.

    Args:
        project: ``Project.slug`` this turn belongs to.
        turn_id: The producer-generated turn UUID.

    Returns:
        The canonical node id.

    Raises:
        ValueError: If either part is empty, or ``project`` contains ``:``.
    """
    if not project or not turn_id:
        raise ValueError("project and turn_id must both be non-empty")
    if ":" in project:
        raise ValueError(f"project must not contain ':': {project!r}")
    return f"{project}:{turn_id}"


@traces(spec="106", reqs=["REQ-106-002", "REQ-106-003"])
def get_canonical_merge_keys() -> dict[str, str]:
    """Return the canonical ``{label: key_property}`` map (a defensive copy).

    Settles the merge-key divergence the W8.1 inventory documented: Requirement
    keys on ``id`` (not ``req_id``), Volk on ``alias``, Session on
    ``session_id``, Project on ``slug`` — one key per label, everywhere.
    """
    return dict(CANONICAL_MERGE_KEYS)


def _constraint_ddl(label: str, prop: str, name: str) -> str:
    """Render one ``CREATE CONSTRAINT ... IF NOT EXISTS`` statement."""
    return f"CREATE CONSTRAINT {name} IF NOT EXISTS FOR (n:{label}) REQUIRE n.{prop} IS UNIQUE"


def _index_ddl(label: str, prop: str, name: str) -> str:
    """Render one ``CREATE INDEX ... IF NOT EXISTS`` statement.

    Unlike :func:`_constraint_ddl`, this is a plain (non-unique) lookup
    index — for a query-shaped property that is not part of any label's
    identity (e.g. ``ArchitectureDecision.status``).
    """
    return f"CREATE INDEX {name} IF NOT EXISTS FOR (n:{label}) ON (n.{prop})"


# Uniqueness constraint per canonical (label, key) pair — generated from the
# single CANONICAL_MERGE_KEYS source of truth so the DDL can never drift from
# the identity table above.
CONSTRAINTS: list[str] = [
    _constraint_ddl(label, prop, _CONSTRAINT_NAMES[label])
    for label, prop in CANONICAL_MERGE_KEYS.items()
]

# Non-identity lookup indexes (architecture ontology, RATIFIED 2026-07-19,
# plans/architecture-ontology-draft.md §5) — every canonical *key* above
# already gets a backing index for free from its uniqueness constraint;
# these five back the ontology's most common non-identity lookups. Broader
# non-identity indexes for the pre-ontology labels remain the concern of
# knowledge_graph/schema.py until a later wave folds them into this canon.
INDEXES: list[str] = [
    _index_ddl(LABEL_ARCHITECTURE_DECISION, "status", "architecture_decision_status"),
    _index_ddl(LABEL_COMPONENT, "project", "component_project"),
    _index_ddl(LABEL_TECHNOLOGY_CHOICE, "category", "technology_choice_category"),
    _index_ddl(LABEL_CLOUD_SERVICE, "neutral_kind", "cloud_service_neutral_kind"),
    _index_ddl(LABEL_RISK, "impact", "risk_impact"),
]


@traces(spec="106", reqs=["REQ-106-005"])
def _memgraph_constraint_ddl(label: str, prop: str) -> str:
    """Memgraph-dialect uniqueness constraint (no name, no IF NOT EXISTS —
    Memgraph's grammar; re-application is tolerated by :func:`apply`'s
    already-exists handling on that engine)."""
    return f"CREATE CONSTRAINT ON (n:{label}) ASSERT n.{prop} IS UNIQUE"


@traces(spec="106", reqs=["REQ-106-005"])
def _memgraph_index_ddl(label: str, prop: str) -> str:
    """Memgraph-dialect label-property index (no name in the grammar)."""
    return f"CREATE INDEX ON :{label}({prop})"


@traces(spec="106", reqs=["REQ-106-005"])
def get_all_ddl(dialect: str = "neo4j") -> list[str]:
    """Return the full canonical DDL (constraints + indexes) as a flat list.

    This is THE single DDL source FR-001 requires: one list, one place,
    applied idempotently by :func:`apply`. ``dialect`` renders the SAME
    canonical identity table for a specific engine — ``"neo4j"`` (default,
    named constraints + IF NOT EXISTS) or ``"memgraph"`` (orchgraph; the
    ASSERT/ON grammar, migration plan Step 1.2). Never hand-write DDL for
    either engine.
    """
    if dialect == "neo4j":
        return list(CONSTRAINTS) + list(INDEXES)
    if dialect == "memgraph":
        constraints = [
            _memgraph_constraint_ddl(label, prop)
            for label, prop in CANONICAL_MERGE_KEYS.items()
        ]
        # Memgraph: a uniqueness constraint does NOT create a backing
        # label-property index (unlike Neo4j), so every canonical key gets
        # an explicit index too — that is what makes the sub-ms indexed
        # lookups of migration plan Step 3.4 real on the KG cluster.
        key_indexes = [
            _memgraph_index_ddl(label, prop)
            for label, prop in CANONICAL_MERGE_KEYS.items()
        ]
        extra_indexes = [
            _memgraph_index_ddl(LABEL_ARCHITECTURE_DECISION, "status"),
            _memgraph_index_ddl(LABEL_COMPONENT, "project"),
            _memgraph_index_ddl(LABEL_TECHNOLOGY_CHOICE, "category"),
            _memgraph_index_ddl(LABEL_CLOUD_SERVICE, "neutral_kind"),
            _memgraph_index_ddl(LABEL_RISK, "impact"),
        ]
        return constraints + key_indexes + extra_indexes
    raise ValueError(f"unknown DDL dialect: {dialect!r} (neo4j|memgraph)")


_CONSTRAINT_STMT_RE = re.compile(
    r"^CREATE CONSTRAINT \S+ IF NOT EXISTS "
    r"FOR \(n:(?P<label>[A-Za-z0-9_]+)\) REQUIRE n\.(?P<prop>[A-Za-z0-9_]+) IS UNIQUE$"
)


@traces(spec="106", story="FR-001", reqs=["REQ-106-005"])
def _drop_orphan_indexes(repo: SyncGraphRepository, label: str, prop: str) -> list[str]:
    """Drop plain (non-constraint-owned) indexes on ``(label, prop)``.

    Neo4j refuses to create a uniqueness constraint while a plain index
    exists on the same label/property (``IndexAlreadyExists``), and
    ``IF NOT EXISTS`` does not soften that. Such orphans are environmental
    drift (auto-created ``index_<hex>`` from pre-canonical code paths); the
    constraint's backing index subsumes them, so dropping is always safe.
    Returns the executed ``DROP INDEX`` statements.

    Neo4j-only by construction, and a no-op elsewhere: the refusal this works
    around is Neo4j's, Memgraph indexes are unnamed (so there is no
    ``DROP INDEX <name>`` to issue), and ``SHOW INDEXES YIELD ...`` is not
    Memgraph syntax at all — calling it there raises instead of finding
    nothing.
    """
    if resolve_kg_dialect() == "memgraph":
        return []
    rows = repo.execute_read(
        "SHOW INDEXES YIELD name, labelsOrTypes, properties, owningConstraint "
        "WHERE owningConstraint IS NULL "
        "AND labelsOrTypes = [$label] AND properties = [$prop] "
        "RETURN name",
        {"label": label, "prop": prop},
    )
    dropped: list[str] = []
    for row in rows:
        stmt = f"DROP INDEX `{row['name']}` IF EXISTS"
        repo.execute_write(stmt)
        dropped.append(stmt)
    return dropped


@traces(spec="106", reqs=["REQ-106-005"])
def apply(repo: SyncGraphRepository, statements: list[str] | None = None) -> list[str]:
    """Idempotently apply canonical DDL to a live operational-KG graph.

    Every statement uses ``IF NOT EXISTS``, so running this twice (or a
    hundred times) against the same graph is a no-op after the first call —
    the required property for a schema-migration entry point.

    Args:
        repo: A ``SyncGraphRepository`` (backlog #163 phase 5b). Each
            statement runs as its own managed write — the original
            implementation ran every statement in one raw session but never
            shared a transaction across them (plain ``session.run`` calls
            are independent auto-commit transactions), so independent
            ``execute_write`` calls are the faithful mapping.
        statements: DDL statements to run; defaults to :func:`get_all_ddl`.
            Callers (tests, the CLI's ``--uri`` test-fleet path) may pass a
            narrower list to scope the effect, e.g. to a single test-only
            label.

    Returns:
        The list of statements that were executed (in order), including any
        ``DROP INDEX`` reconciliations for orphan plain indexes that would
        otherwise block a uniqueness constraint (``IndexAlreadyExists``).
    """
    stmts = list(statements) if statements is not None else get_all_ddl()
    executed: list[str] = []
    for stmt in stmts:
        match = _CONSTRAINT_STMT_RE.match(stmt)
        if match is not None:
            executed.extend(
                _drop_orphan_indexes(repo, match.group("label"), match.group("prop"))
            )
        try:
            # execute_autocommit, NOT execute_write: Memgraph rejects
            # constraint/index DDL inside a managed/explicit transaction
            # ("Constraint manipulation is not allowed in multicommand
            # transactions") — measured live 2026-07-26 applying this exact
            # DDL to a fresh orchgraph database via the Neo4j -> orchgraph
            # migration's per-worker test databases (plans/neo4j-to-
            # orchgraph-migration-plan.md Phase 2 Step 2.3). Neo4j accepts
            # DDL either way, so this is portable across both engines.
            repo.execute_autocommit(stmt)
        except Exception as exc:
            # Memgraph-dialect statements have no IF NOT EXISTS grammar —
            # re-application on that engine surfaces as an "already exists"
            # error, which is exactly the no-op the Neo4j dialect gets for
            # free. Anything else is a real failure and propagates.
            if "already exists" not in str(exc).lower():
                raise
            continue
        executed.append(stmt)
    return executed


@traces(spec="106", reqs=["REQ-106-006"])
@dataclass(frozen=True)
class ExtraConstraint:
    """A live constraint on a canonical label that uses a non-canonical key.

    E.g. the legacy ``project_id`` constraint (``Project.id``) once
    ``Project.slug`` becomes canonical — a real signal that a writer (or an
    older schema generation) still targets the wrong merge key.
    """

    name: str
    label: str
    property: str


@traces(spec="106", reqs=["REQ-106-006"])
@dataclass(frozen=True)
class ConformanceReport:
    """The result of :func:`check` — drift between canon and the live graph."""

    missing_constraints: list[str] = field(default_factory=list)
    extra_constraints: list[ExtraConstraint] = field(default_factory=list)
    non_canonical_nodes: dict[str, int] = field(default_factory=dict)
    # Per drifted label, up to _SAMPLE_LIMIT property dicts from offending
    # nodes. The count alone cannot name the non-conforming writer (the
    # residue is often transient and swept before anyone can inspect it —
    # 16 batch-gate flake recoveries carried only "Class: 7 node(s)"); the
    # write SHAPE (which properties the node carries) identifies the writer
    # from the report alone.
    non_canonical_samples: dict[str, list[dict[str, str]]] = field(default_factory=dict)

    @property
    @traces(spec="106", reqs=["REQ-106-006"])
    def ok(self) -> bool:
        """True iff the live graph conforms exactly to this canon."""
        return not (
            self.missing_constraints
            or self.extra_constraints
            or self.non_canonical_nodes
        )

    @traces(spec="106", reqs=["REQ-106-006"])
    def to_lines(self) -> list[str]:
        """Render a short human-readable summary (used by the CLI)."""
        if self.ok:
            return ["OK — live graph conforms to the canonical schema."]
        lines: list[str] = []
        if self.missing_constraints:
            lines.append("Missing constraints:")
            lines.extend(f"  - {m}" for m in self.missing_constraints)
        if self.extra_constraints:
            lines.append("Extra/unknown constraints (canonical label, wrong key):")
            lines.extend(
                f"  - {e.name}: {e.label}.{e.property}" for e in self.extra_constraints
            )
        if self.non_canonical_nodes:
            lines.append("Labels with nodes lacking their canonical key:")
            for label, count in self.non_canonical_nodes.items():
                lines.append(f"  - {label}: {count} node(s)")
                for props in self.non_canonical_samples.get(label, []):
                    rendered = ", ".join(f"{k}={v}" for k, v in sorted(props.items()))
                    lines.append(f"      sample: {{{rendered}}}")
        return lines


@traces(spec="106", story="FR-001", reqs=["REQ-106-006"])
def _live_unique_pairs(repo: SyncGraphRepository) -> dict[tuple[str, str], str]:
    """Map ``(label, property) -> constraint name`` for live uniqueness constraints.

    Only single-label NODE uniqueness/key constraints are considered —
    composite and relationship constraints are outside this canon's scope.

    Engine-aware (Neo4j -> orchgraph migration, plans/neo4j-to-orchgraph-
    migration-plan.md dialect gate): Neo4j's admin surface is
    ``SHOW CONSTRAINTS`` (columns ``entityType``/``type``/
    ``labelsOrTypes``/``properties``/``name``); Memgraph's is
    ``SHOW CONSTRAINT INFO`` (columns ``constraint type``/``label``/
    ``properties`` — no entity-type split, no constraint name, and no
    composite NODE_KEY concept, so every row IS a single-label uniqueness
    constraint by construction). Memgraph's ``SHOW CONSTRAINT INFO`` is
    also, like DDL, rejected inside a managed/explicit transaction
    ("Storage information query is not allowed in multicommand
    transactions" — measured live 2026-07-26), so the Memgraph branch uses
    ``execute_autocommit``, not ``execute_read``.
    """
    pairs: dict[tuple[str, str], str] = {}
    if resolve_kg_dialect() == "memgraph":
        rows = repo.execute_autocommit("SHOW CONSTRAINT INFO")
        for record in rows:
            if (record.get("constraint type") or "").lower() != "unique":
                continue
            label = record.get("label")
            props = record.get("properties") or []
            if not label or len(props) != 1:
                continue
            pairs[(label, props[0])] = ""  # Memgraph constraints are unnamed
        return pairs
    rows = repo.execute_read("SHOW CONSTRAINTS")
    for record in rows:
        entity_type = record.get("entityType")
        ctype = record.get("type")
        labels = record.get("labelsOrTypes") or []
        props = record.get("properties") or []
        if entity_type != "NODE" or ctype not in ("UNIQUENESS", "NODE_KEY"):
            continue
        if len(labels) != 1 or len(props) != 1:
            continue
        pairs[(labels[0], props[0])] = record.get("name", "")
    return pairs


def live_unique_constraints(repo: SyncGraphRepository) -> set[tuple[str, str]]:
    """Return every live single-label uniqueness constraint as ``(label, prop)``.

    Dialect-aware, and deliberately keyed on the PAIR rather than the
    constraint name: Memgraph constraints are unnamed, so a caller asserting
    on a name (``"module_id" in names``) is asking a question only Neo4j can
    answer. The pair is the invariant that actually matters.
    """
    return set(_live_unique_pairs(repo))


def live_index_pairs(repo: SyncGraphRepository) -> set[tuple[str, str]]:
    """Return every live single-property index as ``(label, prop)``.

    The Neo4j surface is ``SHOW INDEXES`` (``labelsOrTypes``/``properties``);
    Memgraph's is ``SHOW INDEX INFO`` (``label``/``property``), unnamed, and —
    like DDL and ``SHOW CONSTRAINT INFO`` — rejected inside a managed
    transaction, so it goes through ``execute_autocommit``.

    Memgraph's ``property`` column is a LIST for ``label+property`` indexes
    (and absent for bare ``label`` indexes), despite the singular name; both
    shapes are normalized here.
    """
    pairs: set[tuple[str, str]] = set()
    if resolve_kg_dialect() == "memgraph":
        for record in repo.execute_autocommit("SHOW INDEX INFO"):
            label = record.get("label")
            prop = record.get("property")
            if not label or not prop:
                continue
            props = prop if isinstance(prop, list) else [prop]
            pairs.update((label, one) for one in props)
        return pairs
    for record in repo.execute_read("SHOW INDEXES"):
        labels = record.get("labelsOrTypes") or []
        props = record.get("properties") or []
        if len(labels) == 1 and len(props) == 1:
            pairs.add((labels[0], props[0]))
    return pairs


_SAMPLE_LIMIT = 3
_SAMPLE_VALUE_WIDTH = 60


@traces(spec="106", story="FR-001", reqs=["REQ-106-006"])
def _scan_non_canonical_nodes(
    repo: SyncGraphRepository, merge_keys: dict[str, str]
) -> tuple[dict[str, int], dict[str, list[dict[str, str]]]]:
    """Count AND sample, per canonical label, nodes missing their canonical key.

    This is the direct real-data check for "labels written by known writers
    that lack canonical keys" — e.g. Requirement nodes MERGEd by
    ``kernel/pipeline_kg.py`` on ``req_id`` alone have no ``id`` property and
    show up here. Each label is an independent read (the original ran every
    label's query in one raw session without a shared transaction), so
    independent ``execute_read`` calls are the faithful mapping.

    Count and samples come from ONE Cypher round-trip per label —
    ``RETURN count(n) AS c, collect(properties(n))[0.._SAMPLE_LIMIT] AS
    samples`` — not two separate queries. The keyless residue that trips the
    conformance gate under batch load is TRANSIENT — written and swept
    within one batch window (live scan 2026-08-05: zero keyless nodes at
    rest on every band and the shared DB, yet 16 recorded flake recoveries)
    — so a second query issued after the first frequently sampled nothing
    even though the first query's count was nonzero (count/sample TOCTOU).
    One round-trip means the sample can never diverge from the count it is
    reported alongside. Values are truncated to ``_SAMPLE_VALUE_WIDTH``
    chars so a blob-carrying node cannot balloon the report.

    Nodes carrying a ``test_id`` property are EXEMPT (Mohamed ruling
    2026-08-05): ``test_id`` marks ephemeral test-fixture data that the
    session sweep deletes at session end — it is not production drift, and
    counting it made the batch-mode conformance gate (test_kg_schema)
    report phantom drift whenever a fixture-seeding test shared its batch
    (Class/Function/File/Test/Project fixtures legitimately seed minimal
    keyless nodes mid-session). Production nodes never carry ``test_id``,
    so the exemption cannot mask a real non-conforming writer.
    """
    counts: dict[str, int] = {}
    samples: dict[str, list[dict[str, str]]] = {}
    for label, key in merge_keys.items():
        rows = repo.execute_read(
            f"MATCH (n:`{label}`) WHERE n.`{key}` IS NULL AND n.test_id IS NULL "
            "RETURN count(n) AS c, "
            f"collect(properties(n))[0..{_SAMPLE_LIMIT}] AS samples"
        )
        count = rows[0]["c"] if rows else 0
        if not count:
            continue
        counts[label] = count
        rendered = [
            {k: str(v)[:_SAMPLE_VALUE_WIDTH] for k, v in (props or {}).items()}
            for props in (rows[0]["samples"] or [])
        ]
        if rendered:
            samples[label] = rendered
    return counts, samples


@traces(spec="106", reqs=["REQ-106-006"])
def check(
    repo: SyncGraphRepository,
    merge_keys: dict[str, str] | None = None,
) -> ConformanceReport:
    """Conformance gate: diff the live graph against this canon.

    Read-only (never mutates the graph). Reports three kinds of drift:

    1. ``missing_constraints`` — a canonical (label, key) pair with no live
       uniqueness constraint.
    2. ``extra_constraints`` — a live uniqueness constraint on a canonical
       label that targets a DIFFERENT property than the canonical key (e.g.
       the legacy ``Project.id`` constraint once ``Project.slug`` is canon).
    3. ``non_canonical_nodes`` — nodes under a canonical label that lack the
       canonical key property entirely (real data written by a non-conforming
       writer).

    Args:
        repo: A ``SyncGraphRepository`` (backlog #163 phase 5b).
        merge_keys: Canonical ``{label: key_property}`` map to check against;
            defaults to :data:`CANONICAL_MERGE_KEYS`. Tests may pass a
            narrower map to scope the check to a test-only label.

    Returns:
        A :class:`ConformanceReport`. ``report.ok`` is False (and callers —
        e.g. the ``kg_schema.py check`` CLI — should exit non-zero) whenever
        any drift was found.
    """
    keys = merge_keys if merge_keys is not None else CANONICAL_MERGE_KEYS
    canonical_pairs = set(keys.items())
    canonical_labels = frozenset(keys)

    live_pairs = _live_unique_pairs(repo)

    missing = sorted(
        f"{label}.{prop}"
        for label, prop in canonical_pairs
        if (label, prop) not in live_pairs
    )
    extra = sorted(
        (
            ExtraConstraint(name=name, label=label, property=prop)
            for (label, prop), name in live_pairs.items()
            if label in canonical_labels and (label, prop) not in canonical_pairs
        ),
        key=lambda e: (e.label, e.property),
    )
    non_canonical, samples = _scan_non_canonical_nodes(repo, keys)

    return ConformanceReport(
        missing_constraints=missing,
        extra_constraints=extra,
        non_canonical_nodes=non_canonical,
        non_canonical_samples=samples,
    )