orchestrata_ultimate.knowledge_graph.architecture_writer

Reference › orchestrata_ultimate › knowledge_graph › architecture_writer

ArchitectureWriter is the single, converged writer for the architecture ontology (Tier J, ratified in the ontology phase 1 wave). It supersedes the ad-hoc global-name writes of the sync_architecture_to_graph script and writes to the operational graph (orchkg on orchgraph) through a synchronous graph repository, using the label and relationship constants of canonical_schema. Scripts such as check_adr_graph_sync and kg_schema import it; the architecture narrative is in the architecture ontology page.

Project-scoped labels merge on an id of the form project, colon, natural id; the genuinely global reference labels (WAFPillar, CloudService, and the global Pattern) merge on their bare natural id. The private helpers _merge_node and _edge implement every write as a MERGE, so repeated runs are idempotent. The public surface is a family of upsert_* methods, one per node kind: architecture, component, API contract, data entity, quality attribute, constraint, capacity model, cost model and line item, environment, deployment, trade-off, risk, failure mode, technology choice, SLO, alert, pattern, artifact (limited to the types in ARTIFACT_TYPES), WAF pillar and cloud service. A matching set of link_* methods creates the edges, for example link_has_component, link_exposes, link_data_flows_to, link_depends_on, link_persists_to, link_runs_on and link_satisfies. seed_reference_data idempotently loads the WAF_PILLARS and CLOUD_SERVICES tables and returns counts.

ADR immutability is enforced here. upsert_architecture_decision computes a body hash over the four body fields in ADR_BODY_FIELDS and, once the stored decision has status Accepted, refuses to change them, raising AdrImmutabilityError (REQ-106-006). Evolution is a new ADR joined by link_supersedes; set_architecture_decision_status may always change status. rebaseline_architecture_decision and rebaseline_first_capture are the audited exceptions, gated by a table of authorising ADRs. Records arrive from adr_parser.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

AdrImmutabilityError

Bases: Exception · Decorators: traces(spec='106', reqs=['REQ-106-006']) · Line 60

Raised when a write would mutate an Accepted ArchitectureDecision's body.

ArchitectureWriter

Line 390

The converged writer over the RATIFIED architecture ontology (Tier J).

Functions

Source

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

"""The converged architecture-ontology writer (RATIFIED, wave ontology-phase1).

Supersedes ``scripts/sync_architecture_to_graph.py``'s ad-hoc global-name
writes (``plans/architecture-ontology-draft.md`` §7, ruling 8): the single
writer for every Tier J node/relationship family the ratified ontology
defines (``canonical_schema.py`` LABEL_*/REL_* constants), MERGEing on the
canonical identity for each label:

* project-scoped labels key on ``id = "{project}:{natural}"`` (ruling 2, the
  same File/Directory/Package precedent as every other project-scoped tier);
* the genuinely global reference-data labels — ``WAFPillar``, ``CloudService``
  (``Region`` deferred, ruling 6) — key on their bare natural id instead.

**ADR immutability (ruling 1) is enforced at BOTH layers**:

1. Writer-side guard — :meth:`ArchitectureWriter.upsert_architecture_decision`
   computes a ``body_sha`` over the four immutable body fields (context/
   decision/alternatives/consequences) and refuses to change them once the
   matched node's ``status`` is ``"Accepted"`` (raises
   :class:`AdrImmutabilityError`). Evolution is a **new** ADR linked by
   :meth:`ArchitectureWriter.link_supersedes` — never an in-place body edit.
   ``status`` alone (and non-body metadata) may still change on an Accepted
   ADR — :meth:`ArchitectureWriter.set_architecture_decision_status` is the
   dedicated seam for that.
2. Conformance gate — :func:`check_adr_integrity` is the read-only half: it
   recomputes ``body_sha`` for every live ``Accepted`` ADR and flags any node
   whose stored hash no longer matches (a bypassed writer, hand-tampering, or
   an out-of-band Cypher write). ``scripts/kg_schema.py check`` wires this in
   alongside :func:`canonical_schema.check`.

Follows the driver-ownership + ``execute_read``/``execute_write`` convention
already live in ``knowledge_graph/kg_design.py`` (``SyncGraphRepository``),
but resolves the connection via ``core.graph.repository.create_graph_repository``
+ ``GraphRepositoryConfig.from_env()`` rather than ``kg_design.py``'s direct
``NEO4J_URI``-constant construction — the newer, call-time env-resolution
seam (``core/graph/config.py`` docstring; project doctrine "read env at call
time", not at import time).
"""

from __future__ import annotations

import hashlib
import re
from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import (
    GraphRepositoryConfig,
    create_graph_repository,
)
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository


@traces(spec="106", reqs=["REQ-106-006"])
class AdrImmutabilityError(Exception):
    """Raised when a write would mutate an Accepted ArchitectureDecision's body.

    An ADR's body (context/decision/alternatives/consequences) is frozen once
    ``status = "Accepted"`` (ruling 1, ``plans/architecture-ontology-draft.md``
    §2 rule 5). Evolve via a new ADR + :meth:`ArchitectureWriter.link_supersedes`
    instead of editing this one in place.
    """


#: Accepted ruling ADR -> the exact ADR ids it authorises
#: ``rebaseline_architecture_decision`` to overwrite (ADR-0096: ADR-0073's graph
#: body was captured from the wiki summary while Proposed and is stale).
_REBASELINE_AUTHORITIES: dict[str, frozenset[str]] = {
    "ADR-0096": frozenset({"ADR-0073"}),
}


#: Accepted ruling ADR(s) that authorise ``rebaseline_first_capture`` (ADR-0098:
#: a node whose body never came from docs is baselined once from docs).
_FIRST_CAPTURE_AUTHORITIES = frozenset({"ADR-0098"})


@traces(spec="106", reqs=["REQ-106-003"])
def _require_project_slug(project: str, caller: str) -> None:
    """Guard every project-scoped write against an empty/whitespace/qualified
    slug -- same fail-fast doctrine as ``kg_design.py``'s
    ``_require_project_slug`` (2026-07 audit: an anonymous/throwaway slug
    MERGEs its own junk ``:Project`` hub every project-scoped label anchors
    onto)."""
    if not project or not project.strip():
        raise ValueError(
            f"ArchitectureWriter.{caller}: project must be a non-empty, "
            "non-whitespace string -- refusing to write an anonymous "
            "Project hub node."
        )
    if ":" in project:
        raise ValueError(
            f"ArchitectureWriter.{caller}: project must not contain ':': {project!r}"
        )


def _pid(project: str, natural: str, caller: str) -> str:
    """Project-qualified id builder: ``"{project}:{natural}"`` (ruling 2)."""
    _require_project_slug(project, caller)
    if not natural or not str(natural).strip():
        raise ValueError(f"ArchitectureWriter.{caller}: natural id must be non-empty")
    return f"{project}:{natural}"


_SLUG_RE = re.compile(r"[^a-z0-9]+")


def slugify(text: str) -> str:
    """Deterministic slug for the ``id = "{project}:{slug}"`` labels
    (Constraint/QualityAttribute/TradeOff/Risk/FailureMode/TechnologyChoice/
    Alert) -- lowercase, non-alphanumeric runs collapsed to a single hyphen,
    leading/trailing hyphens trimmed."""
    slug = _SLUG_RE.sub("-", text.strip().lower()).strip("-")
    return slug or "unnamed"


ADR_BODY_FIELDS: tuple[str, ...] = (
    "context",
    "decision",
    "alternatives",
    "consequences",
)

# Ratified Artifact.artifact_type enum (plans/architecture-ontology-draft.md
# §3 Phase-6/18/24 rows: design-doc/adr Phase 6, iac-module Phase 24,
# helm-chart/image Phase 18/24, diagram Phase 4/6).
ARTIFACT_TYPES: frozenset[str] = frozenset(
    {"design-doc", "adr", "iac-module", "helm-chart", "image", "diagram"}
)


@traces(spec="106", reqs=["REQ-106-006"])
def _adr_body_sha(
    context: str, decision: str, alternatives: str, consequences: str
) -> str:
    """Content hash over the four immutable ADR body fields (ruling 1)."""
    payload = "\x1f".join(
        (context or "", decision or "", alternatives or "", consequences or "")
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _now_iso() -> str:
    return datetime.now(UTC).isoformat()


@traces(spec="106", reqs=["REQ-106-006"])
def check_adr_integrity(repo: SyncGraphRepository) -> list[str]:
    """Read-only conformance gate (ruling 1, gate half).

    Recomputes ``body_sha`` for every live ``ArchitectureDecision`` whose
    ``status = "Accepted"`` and reports the ids of any whose current body no
    longer matches its stamped hash -- a bypassed writer, hand-tampering, or
    an out-of-band Cypher write. Never mutates the graph.

    Args:
        repo: A ``SyncGraphRepository`` connected to the graph to check.

    Returns:
        A sorted list of offending ``ArchitectureDecision.id`` values (empty
        when every Accepted ADR's body is intact).
    """
    rows = repo.execute_read(
        f"MATCH (a:{cs.LABEL_ARCHITECTURE_DECISION} {{status: 'Accepted'}}) "
        "RETURN a.id AS id, a.body_sha AS body_sha, a.context AS context, "
        "a.decision AS decision, a.alternatives AS alternatives, "
        "a.consequences AS consequences"
    )
    findings: list[str] = []
    for row in rows:
        recomputed = _adr_body_sha(
            row.get("context") or "",
            row.get("decision") or "",
            row.get("alternatives") or "",
            row.get("consequences") or "",
        )
        if row.get("body_sha") != recomputed:
            findings.append(str(row["id"]))
    return sorted(findings)


# ---------------------------------------------------------------------------
# Ruling 6 -- curated CloudService seed + the 5 fixed WAFPillar nodes.
#
# The ~30 services actually named across the system-design skill's 26
# reference files (messaging-servicebus-eventhubs.md, databases-postgres-
# mysql-redis.md, databricks-fabric-analytics.md, azure-storage.md,
# kubernetes-aks-helm.md, api-management-apim.md, key-vault-secrets.md,
# azure-networking.md, cicd-github-actions-argocd.md, observability.md) --
# NOT the full ~714-icon Azure catalog (full-catalog ingestion was explicitly
# rejected, ruling 6). Region seeding is lazy, deferred (not this wave).
# ---------------------------------------------------------------------------
WAF_PILLARS: tuple[dict[str, str], ...] = (
    {"name": "Reliability", "concern": "Resiliency, availability, recovery"},
    {"name": "Security", "concern": "Confidentiality, integrity, threat mitigation"},
    {"name": "CostOptimization", "concern": "ROI, reduce waste"},
    {"name": "OperationalExcellence", "concern": "Observability, DevOps, safe deploys"},
    {"name": "PerformanceEfficiency", "concern": "Scale to demand efficiently"},
)

CLOUD_SERVICES: tuple[dict[str, str], ...] = (
    {
        "name": "Azure Service Bus",
        "vendor": "azure",
        "neutral_kind": "message-broker",
        "category": "messaging",
    },
    {
        "name": "Azure Event Hubs",
        "vendor": "azure",
        "neutral_kind": "event-stream",
        "category": "messaging",
    },
    {
        "name": "Azure Event Grid",
        "vendor": "azure",
        "neutral_kind": "event-router",
        "category": "messaging",
    },
    {
        "name": "Azure Database for PostgreSQL - Flexible Server",
        "vendor": "azure",
        "neutral_kind": "relational-db",
        "category": "database",
    },
    {
        "name": "Azure Database for MySQL - Flexible Server",
        "vendor": "azure",
        "neutral_kind": "relational-db",
        "category": "database",
    },
    {
        "name": "Azure Cache for Redis",
        "vendor": "azure",
        "neutral_kind": "cache",
        "category": "database",
    },
    {
        "name": "Azure Managed Redis",
        "vendor": "azure",
        "neutral_kind": "cache",
        "category": "database",
    },
    {
        "name": "Azure Databricks",
        "vendor": "azure",
        "neutral_kind": "lakehouse-platform",
        "category": "analytics",
    },
    {
        "name": "Microsoft Fabric",
        "vendor": "azure",
        "neutral_kind": "lakehouse-platform",
        "category": "analytics",
    },
    {
        "name": "Azure SQL Database",
        "vendor": "azure",
        "neutral_kind": "relational-db",
        "category": "database",
    },
    {
        "name": "Azure Blob Storage",
        "vendor": "azure",
        "neutral_kind": "object-storage",
        "category": "storage",
    },
    {
        "name": "Azure Data Lake Storage Gen2",
        "vendor": "azure",
        "neutral_kind": "data-lake",
        "category": "storage",
    },
    {
        "name": "Azure Files",
        "vendor": "azure",
        "neutral_kind": "file-storage",
        "category": "storage",
    },
    {
        "name": "Azure NetApp Files",
        "vendor": "azure",
        "neutral_kind": "file-storage",
        "category": "storage",
    },
    {
        "name": "Azure Managed Lustre",
        "vendor": "azure",
        "neutral_kind": "hpc-file-storage",
        "category": "storage",
    },
    {
        "name": "Azure Managed Disks",
        "vendor": "azure",
        "neutral_kind": "block-storage",
        "category": "storage",
    },
    {
        "name": "Azure Kubernetes Service",
        "vendor": "azure",
        "neutral_kind": "container-orchestration",
        "category": "compute",
    },
    {
        "name": "Azure Container Apps",
        "vendor": "azure",
        "neutral_kind": "serverless-containers",
        "category": "compute",
    },
    {
        "name": "Azure Batch",
        "vendor": "azure",
        "neutral_kind": "batch-scheduler",
        "category": "compute",
    },
    {
        "name": "Azure API Management",
        "vendor": "azure",
        "neutral_kind": "api-gateway",
        "category": "integration",
    },
    {
        "name": "Azure Key Vault",
        "vendor": "azure",
        "neutral_kind": "secrets-manager",
        "category": "security",
    },
    {
        "name": "Azure Virtual Network",
        "vendor": "azure",
        "neutral_kind": "virtual-network",
        "category": "networking",
    },
    {
        "name": "Azure Firewall",
        "vendor": "azure",
        "neutral_kind": "network-firewall",
        "category": "networking",
    },
    {
        "name": "Azure Bastion",
        "vendor": "azure",
        "neutral_kind": "bastion-host",
        "category": "networking",
    },
    {
        "name": "Azure Load Balancer",
        "vendor": "azure",
        "neutral_kind": "load-balancer",
        "category": "networking",
    },
    {
        "name": "Azure Container Registry",
        "vendor": "azure",
        "neutral_kind": "container-registry",
        "category": "devops",
    },
    {
        "name": "GitHub Actions",
        "vendor": "github",
        "neutral_kind": "ci-pipeline",
        "category": "devops",
    },
    {
        "name": "Argo CD",
        "vendor": "argoproj",
        "neutral_kind": "gitops-cd",
        "category": "devops",
    },
    {
        "name": "Azure DevOps",
        "vendor": "azure",
        "neutral_kind": "alm-platform",
        "category": "devops",
    },
    {
        "name": "Azure Monitor",
        "vendor": "azure",
        "neutral_kind": "monitoring",
        "category": "observability",
    },
)


class ArchitectureWriter:
    """The converged writer over the RATIFIED architecture ontology (Tier J).

    Ownership: when ``repo`` is None, the writer lazily builds and owns its
    own repository (via ``create_graph_repository`` / ``GraphRepositoryConfig
    .from_env()``, resolved at first write, not at import time) --
    ``close()`` on it really closes the underlying driver. When a ``repo`` is
    injected, ownership belongs to whoever built it (``BoltSyncRepository
    .close()`` is already a no-op for a shared/injected driver).
    """

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

    @traces(spec="106", reqs=["REQ-106-008"])
    def _get_repo(self) -> SyncGraphRepository:
        if self._repo is None:
            self._repo = create_graph_repository(GraphRepositoryConfig.from_env())
        return self._repo

    def close(self) -> None:
        """Close the writer's repository (no-op for an injected/shared one)."""
        if self._repo is not None:
            self._repo.close()
        self._repo = None

    @traces(spec="106", reqs=["REQ-106-008"])
    def _write(
        self, query: str, params: Mapping[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        return self._get_repo().execute_write(query, dict(params or {}))

    @traces(spec="106", reqs=["REQ-106-008"])
    def _read(
        self, query: str, params: Mapping[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        return self._get_repo().execute_read(query, dict(params or {}))

    # -- generic node/edge primitives -----------------------------------

    @traces(spec="106", reqs=["REQ-106-008"])
    def _merge_node(
        self, label: str, key_prop: str, key_value: str, properties: Mapping[str, Any]
    ) -> str:
        """MERGE one node on ``(label, key_prop=key_value)`` and SET the rest.

        ``label`` is always one of this module's own ``cs.LABEL_*`` constants
        (never external input), so f-string label interpolation is safe --
        the same trusted-fixed-set stance as the writer this module
        supersedes.
        """
        props = {k: v for k, v in properties.items() if k != key_prop}
        set_clause = ", ".join(f"n.{k} = ${k}" for k in props)
        query = f"MERGE (n:{label} {{{key_prop}: $__key}})"
        if set_clause:
            query += f" SET {set_clause}"
        params: dict[str, Any] = dict(props)
        params["__key"] = key_value
        self._write(query, params)
        return key_value

    @traces(spec="106", reqs=["REQ-106-008"])
    def _edge(
        self,
        from_label: str,
        from_key_prop: str,
        from_key_value: str,
        rel_type: str,
        to_label: str,
        to_key_prop: str,
        to_key_value: str,
        rel_properties: Mapping[str, Any] | None = None,
    ) -> None:
        """MERGE one edge between two already-MERGEd nodes.

        A MATCH on a not-yet-created endpoint yields no rows and silently
        drops the edge (same ordering caveat the superseded writer's own
        docstring called out) -- callers create both endpoints first.
        """
        rel_properties = dict(rel_properties or {})
        set_clause = ""
        if rel_properties:
            set_clause = " SET " + ", ".join(f"r.{k} = ${k}" for k in rel_properties)
        query = (
            f"MATCH (a:{from_label} {{{from_key_prop}: $__from}}) "
            f"MATCH (b:{to_label} {{{to_key_prop}: $__to}}) "
            f"MERGE (a)-[r:{rel_type}]->(b)" + set_clause
        )
        params: dict[str, Any] = dict(rel_properties)
        params["__from"] = from_key_value
        params["__to"] = to_key_value
        self._write(query, params)

    @staticmethod
    def _key_prop(label: str) -> str:
        return cs.CANONICAL_MERGE_KEYS[label]

    # -- Architecture -----------------------------------------------------

    def upsert_architecture(
        self,
        project: str,
        name: str,
        *,
        version: str = "",
        description: str = "",
        status: str = "draft",
    ) -> str:
        """MERGE an ``Architecture`` root and its owning ``Project`` (HAS_ARCHITECTURE)."""
        arch_id = _pid(project, name, "upsert_architecture")
        self._merge_node(
            cs.LABEL_ARCHITECTURE,
            "id",
            arch_id,
            {
                "name": name,
                "version": version,
                "description": description,
                "status": status,
                "project": project,
            },
        )
        self._merge_node(cs.LABEL_PROJECT, "slug", project, {"slug": project})
        self._edge(
            cs.LABEL_PROJECT,
            "slug",
            project,
            cs.REL_HAS_ARCHITECTURE,
            cs.LABEL_ARCHITECTURE,
            "id",
            arch_id,
        )
        return arch_id

    def upsert_component(
        self,
        project: str,
        name: str,
        *,
        responsibility: str = "",
        compute_platform: str = "",
        protocol: str = "",
        required: bool = True,
        boundary: str = "",
    ) -> str:
        comp_id = _pid(project, name, "upsert_component")
        self._merge_node(
            cs.LABEL_COMPONENT,
            "id",
            comp_id,
            {
                "name": name,
                "responsibility": responsibility,
                "compute_platform": compute_platform,
                "protocol": protocol,
                "required": required,
                "boundary": boundary,
                "project": project,
            },
        )
        return comp_id

    def upsert_api_contract(
        self,
        project: str,
        component_natural: str,
        name: str,
        *,
        kind: str = "",
        shape: str = "",
        sync: bool | None = None,
        version: str = "",
    ) -> str:
        _require_project_slug(project, "upsert_api_contract")
        api_id = f"{project}:{component_natural}:{name}"
        self._merge_node(
            cs.LABEL_API_CONTRACT,
            "id",
            api_id,
            {
                "name": name,
                "kind": kind,
                "shape": shape,
                "sync": sync,
                "version": version,
                "project": project,
            },
        )
        self.link_exposes(
            _pid(project, component_natural, "upsert_api_contract"), api_id
        )
        return api_id

    def upsert_data_entity(
        self,
        project: str,
        name: str,
        *,
        access_patterns: Sequence[str] | None = None,
        consistency: str = "",
        store_kind: str = "",
        pii: bool | None = None,
    ) -> str:
        de_id = _pid(project, name, "upsert_data_entity")
        self._merge_node(
            cs.LABEL_DATA_ENTITY,
            "id",
            de_id,
            {
                "name": name,
                "access_patterns": list(access_patterns or []),
                "consistency": consistency,
                "store_kind": store_kind,
                "pii": pii,
                "project": project,
            },
        )
        return de_id

    def upsert_quality_attribute(
        self,
        project: str,
        name: str,
        *,
        metric: str = "",
        target: str = "",
        measured_as: str = "",
        pillar: str = "",
    ) -> str:
        qa_id = _pid(project, slugify(name), "upsert_quality_attribute")
        self._merge_node(
            cs.LABEL_QUALITY_ATTRIBUTE,
            "id",
            qa_id,
            {
                "name": name,
                "metric": metric,
                "target": target,
                "measured_as": measured_as,
                "pillar": pillar,
                "project": project,
            },
        )
        return qa_id

    def upsert_constraint(
        self,
        project: str,
        name: str,
        *,
        kind: str = "",
        severity: str = "",
        description: str = "",
    ) -> str:
        con_id = _pid(project, slugify(name), "upsert_constraint")
        self._merge_node(
            cs.LABEL_CONSTRAINT,
            "id",
            con_id,
            {
                "name": name,
                "kind": kind,
                "severity": severity,
                "description": description,
                "project": project,
            },
        )
        return con_id

    def upsert_capacity_model(
        self,
        project: str,
        name: str,
        *,
        peak_rps: int | None = None,
        avg_rps: int | None = None,
        read_write_ratio: str = "",
        storage_gb_yr: float | None = None,
        working_set_fits_memory: bool | None = None,
        binding_constraint: str = "",
    ) -> str:
        cap_id = _pid(project, name, "upsert_capacity_model")
        self._merge_node(
            cs.LABEL_CAPACITY_MODEL,
            "id",
            cap_id,
            {
                "name": name,
                "peak_rps": peak_rps,
                "avg_rps": avg_rps,
                "read_write_ratio": read_write_ratio,
                "storage_gb_yr": storage_gb_yr,
                "working_set_fits_memory": working_set_fits_memory,
                "binding_constraint": binding_constraint,
                "project": project,
            },
        )
        return cap_id

    def upsert_cost_line_item(
        self,
        project: str,
        cost_model_natural: str,
        index: int,
        *,
        component: str = "",
        meter: str = "",
        quantity: float | None = None,
        unit_price: float | None = None,
        monthly: float | None = None,
        billing_model: str = "",
    ) -> str:
        _require_project_slug(project, "upsert_cost_line_item")
        li_id = f"{project}:{cost_model_natural}:{index}"
        self._merge_node(
            cs.LABEL_COST_LINE_ITEM,
            "id",
            li_id,
            {
                "component": component,
                "meter": meter,
                "quantity": quantity,
                "unit_price": unit_price,
                "monthly": monthly,
                "billing_model": billing_model,
                "project": project,
            },
        )
        return li_id

    def upsert_cost_model(
        self,
        project: str,
        name: str,
        *,
        run_rate_monthly: float | None = None,
        currency: str = "",
        at_10x_dominant_meter: str = "",
        top_drivers: Sequence[str] | None = None,
        priced_at: str | None = None,
        region: str = "",
        line_items: Sequence[Mapping[str, Any]] | None = None,
    ) -> str:
        """MERGE a ``CostModel``, and (per REQ) its ``CostLineItem`` children.

        ``line_items`` is a list of kwargs for :meth:`upsert_cost_line_item`
        (``component``/``meter``/``quantity``/``unit_price``/``monthly``/
        ``billing_model``); each is created and wired via ``HAS_LINE_ITEM``
        in the order given (the id's ``{n}`` is the list index).
        """
        cm_id = _pid(project, name, "upsert_cost_model")
        self._merge_node(
            cs.LABEL_COST_MODEL,
            "id",
            cm_id,
            {
                "name": name,
                "run_rate_monthly": run_rate_monthly,
                "currency": currency,
                "at_10x_dominant_meter": at_10x_dominant_meter,
                "top_drivers": list(top_drivers or []),
                "priced_at": priced_at,
                "region": region,
                "project": project,
            },
        )
        for i, item in enumerate(line_items or []):
            li_id = self.upsert_cost_line_item(project, name, i, **item)
            self.link_has_line_item(cm_id, li_id)
        return cm_id

    def upsert_environment(self, project: str, name: str, *, tier: str = "") -> str:
        env_id = _pid(project, name, "upsert_environment")
        self._merge_node(
            cs.LABEL_ENVIRONMENT,
            "id",
            env_id,
            {"name": name, "tier": tier, "project": project},
        )
        return env_id

    def upsert_deployment(
        self,
        project: str,
        component_natural: str,
        environment_natural: str,
        *,
        delivery_model: str = "",
        resiliency: str = "",
        replicas: int | None = None,
        network_placement: str = "",
    ) -> str:
        """MERGE a ``Deployment`` and wire ``DEPLOYED_AS``/``IN_ENVIRONMENT``.

        Requires the ``Component`` and ``Environment`` nodes to already exist
        (a MATCH on either not-yet-created endpoint silently drops the edge --
        create them first, same caveat as the superseded writer).
        """
        _require_project_slug(project, "upsert_deployment")
        dep_id = f"{project}:{component_natural}:{environment_natural}"
        self._merge_node(
            cs.LABEL_DEPLOYMENT,
            "id",
            dep_id,
            {
                "delivery_model": delivery_model,
                "resiliency": resiliency,
                "replicas": replicas,
                "network_placement": network_placement,
                "project": project,
            },
        )
        comp_id = f"{project}:{component_natural}"
        env_id = f"{project}:{environment_natural}"
        self.link_deployed_as(comp_id, dep_id)
        self.link_in_environment(dep_id, env_id)
        return dep_id

    # -- ArchitectureDecision (ADR) — RULING 1 immutability guard ----------

    @traces(spec="106", reqs=["REQ-106-002", "REQ-106-003"])
    def upsert_architecture_decision(
        self,
        project: str,
        adr_number: int,
        title: str,
        *,
        padded_id: str,
        register: str,
        status: str = "Proposed",
        date: str | None = None,
        deciders: Sequence[str] | None = None,
        context: str = "",
        decision: str = "",
        alternatives: str = "",
        consequences: str = "",
        one_way_door: bool | None = None,
    ) -> str:
        """MERGE an ``ArchitectureDecision``, enforcing ruling-1 immutability.

        ADR-0047: this project has TWO independent ADR registers -- the
        factory register (``docs/adr/NNNN-*.md``, always 4-digit) and the
        project register (the wiki's own 1-3 digit ``ADR-NNN`` sections) --
        and they are NOT the same id space: ``ADR-001`` (project) and
        ``ADR-0001`` (factory) are different decisions that happen to share
        an integer. The node id is therefore keyed on the caller-supplied
        ``padded_id`` verbatim (``f"{project}:ADR-{padded_id}"``), never on
        a re-padded ``adr_number`` -- that re-padding is exactly what used
        to collapse the two registers onto one node. ``register`` and
        ``padded_id`` are stamped as node properties alongside ``adr_number``
        (which stays, for callers that only need the integer).

        The body (``context``/``decision``/``alternatives``/``consequences``)
        is hashed into ``body_sha`` and frozen once the matched node's
        ``status`` is ``"Accepted"``: a call that would change the hash
        raises :class:`AdrImmutabilityError` instead of writing. Everything
        else (``title``/``status``/``deciders``/``date``/``one_way_door``)
        remains freely mutable, including on an Accepted node -- evolve the
        body via a *new* ADR + :meth:`link_supersedes` instead.

        Args:
            padded_id: the heading's digit string verbatim (e.g. ``"0001"``
                or ``"001"``) -- forms the node id together with ``project``.
            register: ``"factory"`` or ``"project"`` (ADR-0047) -- which
                register this decision belongs to.

        Raises:
            AdrImmutabilityError: the matched node is Accepted and this call
                would change its body.
        """
        _require_project_slug(project, "upsert_architecture_decision")
        adr_id = f"{project}:ADR-{padded_id}"
        body_sha = _adr_body_sha(context, decision, alternatives, consequences)

        existing = self._read(
            f"MATCH (a:{cs.LABEL_ARCHITECTURE_DECISION} {{id: $id}}) "
            "RETURN a.status AS status, a.body_sha AS body_sha",
            {"id": adr_id},
        )
        if (
            existing
            and existing[0]["status"] == "Accepted"
            and existing[0]["body_sha"] != body_sha
        ):
            raise AdrImmutabilityError(
                f"ArchitectureDecision {adr_id!r} is Accepted; its body "
                "(context/decision/alternatives/consequences) is immutable. "
                "Evolve via a new ADR + link_supersedes() instead of editing "
                "this one in place."
            )

        self._merge_node(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            {
                "adr_number": int(adr_number),
                "register": register,
                "padded_id": padded_id,
                "title": title,
                "status": status,
                "date": date,
                "deciders": list(deciders or []),
                "context": context,
                "decision": decision,
                "alternatives": alternatives,
                "consequences": consequences,
                "one_way_door": one_way_door,
                "body_sha": body_sha,
                "project": project,
            },
        )
        return adr_id

    @traces(spec="106", reqs=["REQ-106-006"])
    def rebaseline_architecture_decision(
        self,
        project: str,
        padded_id: str,
        *,
        authority: str,
        title: str,
        status: str,
        date: str | None,
        context: str,
        decision: str,
        alternatives: str,
        consequences: str,
        named_ids: frozenset[str] | None = None,
        authority_status: str | None = None,
    ) -> str:
        """Single-id, ruling-authorised body re-baseline of an existing node.

        The ONE sanctioned bypass of the ruling-1 immutability guard: an
        Accepted ADR ruling (``authority``) names the exact ADR ids whose
        graph body is stale and may be overwritten once with the docs (or
        wiki-register) body. Two authority-verification modes:

        * ``named_ids`` given (ADR-0102's generic mechanism) -- the caller
          parsed the authority ADR's own docs record; ``named_ids`` is the
          set of ``ADR-NNNN`` ids its Decision section names and
          ``authority_status`` that record's status. Refused unless the
          authority record is Accepted and the target id is named.
        * ``named_ids=None`` -- the authority's docs record could not be
          parsed, so the hardcoded ``_REBASELINE_AUTHORITIES`` map is the
          fallback (ADR-0096's original path).

        Any other ``(authority, id)`` pair is refused with ``ValueError``
        before a read or write. The node must already exist (a re-baseline
        is never a create). Writes the body, its ``body_sha``, ``status``
        and the audit properties ``rebaselined_by`` / ``rebaselined_at``.
        ``upsert_architecture_decision`` and its guard are untouched.
        """
        _require_project_slug(project, "rebaseline_architecture_decision")
        label = f"ADR-{padded_id}"
        if named_ids is not None:
            if authority_status != "Accepted":
                raise ValueError(
                    f"authority {authority} is {authority_status!r} in docs, "
                    "not Accepted"
                )
            if label not in named_ids:
                raise ValueError(f"{authority} does not name {label}")
        elif label not in _REBASELINE_AUTHORITIES.get(authority, frozenset()):
            raise ValueError(
                f"{authority!r} does not authorise re-baselining {label!r}; "
                f"it names {sorted(_REBASELINE_AUTHORITIES.get(authority, ()))}"
            )
        adr_id = f"{project}:{label}"
        existing = self._read(
            f"MATCH (a:{cs.LABEL_ARCHITECTURE_DECISION} {{id: $id}}) RETURN a.id AS id",
            {"id": adr_id},
        )
        if not existing:
            raise ValueError(f"cannot re-baseline {adr_id!r}: no such node")
        return self._merge_node(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            {
                "title": title,
                "status": status,
                "date": date,
                "context": context,
                "decision": decision,
                "alternatives": alternatives,
                "consequences": consequences,
                "body_sha": _adr_body_sha(
                    context, decision, alternatives, consequences
                ),
                "rebaselined_by": authority,
                "rebaselined_at": datetime.now(UTC).isoformat(),
            },
        )

    @traces(spec="106", reqs=["REQ-106-006"])
    def read_architecture_decision_hashes(self, project: str) -> dict[str, str]:
        """``{padded_id: body_sha}`` for every factory-register decision node
        of ``project`` (read-only; a node with no stored hash maps to ``""``)."""
        _require_project_slug(project, "read_architecture_decision_hashes")
        rows = self._read(
            f"MATCH (a:{cs.LABEL_ARCHITECTURE_DECISION} "
            "{project: $project, register: 'factory'}) "
            "RETURN a.padded_id AS padded_id, a.body_sha AS body_sha",
            {"project": project},
        )
        return {str(r["padded_id"]): str(r.get("body_sha") or "") for r in rows}

    @traces(spec="106", reqs=["REQ-106-006"])
    def rebaseline_first_capture(
        self,
        project: str,
        padded_id: str,
        *,
        authority: str,
        wiki_body_sha: str | None,
        title: str,
        status: str,
        date: str | None,
        context: str,
        decision: str,
        alternatives: str,
        consequences: str,
    ) -> tuple[str, str]:
        """First docs capture of a node whose body never came from docs.

        ADR-0098 rules 1-3: the ONLY bodies immutability protects are docs
        bodies, so a node holding the empty body or the wiki entry's parsed
        body (``wiki_body_sha``) is baselined once from docs. Refused with
        ``ValueError`` unless ``authority`` is in ``_FIRST_CAPTURE_AUTHORITIES``,
        the node exists, and its stored ``body_sha`` is the empty-body hash or
        ``wiki_body_sha`` -- a node whose body already differs from both is real
        drift and is never touched. Writes the body, ``body_sha``, status, the
        audit pair ``rebaselined_by``/``rebaselined_at`` and the docs-capture
        marker ``docs_captured``.

        Returns:
            ``(before_sha, after_sha)`` of the node's body hash.
        """
        _require_project_slug(project, "rebaseline_first_capture")
        if authority not in _FIRST_CAPTURE_AUTHORITIES:
            raise ValueError(
                f"{authority!r} does not authorise a first-capture re-baseline; "
                f"authorities: {sorted(_FIRST_CAPTURE_AUTHORITIES)}"
            )
        adr_id = f"{project}:ADR-{padded_id}"
        existing = self._read(
            f"MATCH (a:{cs.LABEL_ARCHITECTURE_DECISION} {{id: $id}}) "
            "RETURN a.body_sha AS body_sha",
            {"id": adr_id},
        )
        if not existing:
            raise ValueError(f"cannot re-baseline {adr_id!r}: no such node")
        before = str(existing[0].get("body_sha") or "")
        first_capture = {_adr_body_sha("", "", "", "")}
        if wiki_body_sha:
            first_capture.add(wiki_body_sha)
        if before not in first_capture:
            raise ValueError(
                f"{adr_id!r} body {before[:10]!r} is neither empty nor the wiki "
                "body: not a first capture (real drift needs its own ADR)"
            )
        after = _adr_body_sha(context, decision, alternatives, consequences)
        self._merge_node(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            {
                "title": title,
                "status": status,
                "date": date,
                "context": context,
                "decision": decision,
                "alternatives": alternatives,
                "consequences": consequences,
                "body_sha": after,
                "docs_captured": True,
                "rebaselined_by": authority,
                "rebaselined_at": datetime.now(UTC).isoformat(),
            },
        )
        return before, after

    def set_architecture_decision_status(self, adr_id: str, status: str) -> None:
        """Status-only mutation -- always allowed, even on an Accepted ADR.

        The dedicated seam for a status transition (``Proposed``
        -> ``Accepted``, ``Accepted`` -> ``Superseded``/``Deprecated``) that
        never touches the immutable body fields, so it never raises
        :class:`AdrImmutabilityError`.
        """
        self._write(
            f"MATCH (a:{cs.LABEL_ARCHITECTURE_DECISION} {{id: $id}}) "
            "SET a.status = $status, a.updated_at = $ts",
            {"id": adr_id, "status": status, "ts": _now_iso()},
        )

    def upsert_tradeoff(
        self,
        project: str,
        name: str,
        *,
        improves: str = "",
        at_cost_of: str = "",
        rationale: str = "",
    ) -> str:
        to_id = _pid(project, slugify(name), "upsert_tradeoff")
        self._merge_node(
            cs.LABEL_TRADE_OFF,
            "id",
            to_id,
            {
                "name": name,
                "improves": improves,
                "at_cost_of": at_cost_of,
                "rationale": rationale,
                "project": project,
            },
        )
        return to_id

    def upsert_risk(
        self,
        project: str,
        name: str,
        *,
        description: str = "",
        likelihood: str = "",
        impact: str = "",
        mitigation: str = "",
    ) -> str:
        risk_id = _pid(project, slugify(name), "upsert_risk")
        self._merge_node(
            cs.LABEL_RISK,
            "id",
            risk_id,
            {
                "name": name,
                "description": description,
                "likelihood": likelihood,
                "impact": impact,
                "mitigation": mitigation,
                "project": project,
            },
        )
        return risk_id

    def upsert_failure_mode(
        self,
        project: str,
        name: str,
        *,
        trigger: str = "",
        blast_radius: str = "",
        degradation: str = "",
    ) -> str:
        fm_id = _pid(project, slugify(name), "upsert_failure_mode")
        self._merge_node(
            cs.LABEL_FAILURE_MODE,
            "id",
            fm_id,
            {
                "name": name,
                "trigger": trigger,
                "blast_radius": blast_radius,
                "degradation": degradation,
                "project": project,
            },
        )
        return fm_id

    def upsert_technology_choice(
        self,
        project: str,
        name: str,
        *,
        category: str = "",
        rationale: str = "",
        reversibility: str = "",
    ) -> str:
        tc_id = _pid(project, slugify(name), "upsert_technology_choice")
        self._merge_node(
            cs.LABEL_TECHNOLOGY_CHOICE,
            "id",
            tc_id,
            {
                "name": name,
                "category": category,
                "rationale": rationale,
                "reversibility": reversibility,
                "project": project,
            },
        )
        return tc_id

    def upsert_slo(
        self,
        project: str,
        component_natural: str,
        name: str,
        *,
        sli: str = "",
        target: str = "",
        window: str = "",
        error_budget: str = "",
        burn_alert_windows: Sequence[str] | None = None,
    ) -> str:
        _require_project_slug(project, "upsert_slo")
        slo_id = f"{project}:{component_natural}:{name}"
        self._merge_node(
            cs.LABEL_SERVICE_LEVEL_OBJECTIVE,
            "id",
            slo_id,
            {
                "name": name,
                "sli": sli,
                "target": target,
                "window": window,
                "error_budget": error_budget,
                "burn_alert_windows": list(burn_alert_windows or []),
                "project": project,
            },
        )
        self.link_has_slo(_pid(project, component_natural, "upsert_slo"), slo_id)
        return slo_id

    def upsert_alert(
        self,
        project: str,
        name: str,
        *,
        alert_on: str = "",
        tier: str = "",
        burn_rate: str = "",
        runbook_url: str = "",
    ) -> str:
        alert_id = _pid(project, slugify(name), "upsert_alert")
        self._merge_node(
            cs.LABEL_ALERT,
            "id",
            alert_id,
            {
                "name": name,
                "alert_on": alert_on,
                "tier": tier,
                "burn_rate": burn_rate,
                "runbook_url": runbook_url,
                "project": project,
            },
        )
        return alert_id

    def upsert_pattern(
        self,
        name: str,
        *,
        category: str = "",
        problem_area: str = "",
        pillars: Sequence[str] | None = None,
        description: str = "",
        source: str = "",
    ) -> str:
        """MERGE a (global) ``Pattern`` -- REUSE+EXT of the existing canon label."""
        self._merge_node(
            cs.LABEL_PATTERN,
            "name",
            name,
            {
                "category": category,
                "problem_area": problem_area,
                "pillars": list(pillars or []),
                "description": description,
                "source": source,
            },
        )
        return name

    def upsert_artifact(
        self,
        project: str,
        name: str,
        *,
        artifact_type: str,
        path: str = "",
    ) -> str:
        """MERGE an ``Artifact`` -- a deliverable file (design doc, ADR page,
        IaC module, Helm chart, golden image, diagram) -- project-qualified
        per ruling 2, exactly like every other Tier-J label.

        Raises:
            ValueError: ``artifact_type`` is not one of the ratified enum
                values (:data:`ARTIFACT_TYPES`).
        """
        if artifact_type not in ARTIFACT_TYPES:
            raise ValueError(
                "ArchitectureWriter.upsert_artifact: artifact_type must be "
                f"one of {sorted(ARTIFACT_TYPES)}, got {artifact_type!r}"
            )
        art_id = _pid(project, name, "upsert_artifact")
        self._merge_node(
            cs.LABEL_ARTIFACT,
            "id",
            art_id,
            {
                "name": name,
                "artifact_type": artifact_type,
                "path": path,
                "project": project,
            },
        )
        return art_id

    # -- Ruling 6 reference data (global) ----------------------------------

    @traces(spec="106", reqs=["REQ-106-008"])
    def upsert_waf_pillar(self, name: str, *, concern: str = "") -> str:
        self._merge_node(cs.LABEL_WAF_PILLAR, "name", name, {"concern": concern})
        return name

    @traces(spec="106", reqs=["REQ-106-008"])
    def upsert_cloud_service(
        self,
        name: str,
        *,
        vendor: str = "",
        neutral_kind: str = "",
        category: str = "",
        docs_url: str = "",
    ) -> str:
        self._merge_node(
            cs.LABEL_CLOUD_SERVICE,
            "name",
            name,
            {
                "vendor": vendor,
                "neutral_kind": neutral_kind,
                "category": category,
                "docs_url": docs_url,
            },
        )
        return name

    @traces(spec="106", reqs=["REQ-106-005"])
    def seed_reference_data(
        self,
        project: str,
        *,
        waf_pillars: Sequence[Mapping[str, str]] | None = None,
        cloud_services: Sequence[Mapping[str, str]] | None = None,
    ) -> dict[str, int]:
        """Idempotent MERGE of the 5 WAFPillar nodes + the curated CloudService
        seed (ruling 6). Defaults to the real, ratified :data:`WAF_PILLARS` /
        :data:`CLOUD_SERVICES` catalogs; ``waf_pillars``/``cloud_services``
        override for callers (e.g. tests) that want a smaller/different set
        without touching the module constants.

        ``project`` is REQUIRED: the session-254 root-connectivity invariant
        says every node must reach the root ``Project``, so each seeded
        reference node is anchored via ``(:Project)-[:REFERENCES]->(node)``.
        The nodes themselves stay global (natural ``name`` identity, shared
        across projects); only the hub edge is per-project.

        Returns:
            ``{"waf_pillars": N, "cloud_services": M}`` counts written.
        """
        _require_project_slug(project, "seed_reference_data")
        pillars = list(waf_pillars if waf_pillars is not None else WAF_PILLARS)
        services = list(
            cloud_services if cloud_services is not None else CLOUD_SERVICES
        )
        self._merge_node(cs.LABEL_PROJECT, "slug", project, {"slug": project})
        for pillar in pillars:
            self.upsert_waf_pillar(pillar["name"], concern=pillar.get("concern", ""))
            self._edge(
                cs.LABEL_PROJECT,
                "slug",
                project,
                cs.REL_REFERENCES,
                cs.LABEL_WAF_PILLAR,
                "name",
                pillar["name"],
            )
        for service in services:
            self.upsert_cloud_service(
                service["name"],
                vendor=service.get("vendor", ""),
                neutral_kind=service.get("neutral_kind", ""),
                category=service.get("category", ""),
                docs_url=service.get("docs_url", ""),
            )
            self._edge(
                cs.LABEL_PROJECT,
                "slug",
                project,
                cs.REL_REFERENCES,
                cs.LABEL_CLOUD_SERVICE,
                "name",
                service["name"],
            )
        return {"waf_pillars": len(pillars), "cloud_services": len(services)}

    # -- §4 edge helpers ----------------------------------------------------

    def link_has_component(self, architecture_id: str, component_id: str) -> None:
        self._edge(
            cs.LABEL_ARCHITECTURE,
            "id",
            architecture_id,
            cs.REL_HAS_COMPONENT,
            cs.LABEL_COMPONENT,
            "id",
            component_id,
        )

    def link_exposes(self, component_id: str, api_contract_id: str) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_EXPOSES,
            cs.LABEL_API_CONTRACT,
            "id",
            api_contract_id,
        )

    def link_data_flows_to(
        self,
        from_component_id: str,
        to_component_id: str,
        *,
        protocol: str = "",
        sync: bool | None = None,
        critical_path: bool | None = None,
        ordering: str = "",
        via: str = "",
    ) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            from_component_id,
            cs.REL_DATA_FLOWS_TO,
            cs.LABEL_COMPONENT,
            "id",
            to_component_id,
            {
                "protocol": protocol,
                "sync": sync,
                "critical_path": critical_path,
                "ordering": ordering,
                "via": via,
            },
        )

    def link_depends_on(
        self, from_component_id: str, to_component_id: str, *, kind: str = ""
    ) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            from_component_id,
            cs.REL_DEPENDS_ON,
            cs.LABEL_COMPONENT,
            "id",
            to_component_id,
            {"kind": kind},
        )

    def link_persists_to(
        self, component_id: str, data_entity_id: str, *, mode: str = ""
    ) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_PERSISTS_TO,
            cs.LABEL_DATA_ENTITY,
            "id",
            data_entity_id,
            {"mode": mode},
        )

    def link_relates_to(
        self,
        from_data_entity_id: str,
        to_data_entity_id: str,
        *,
        cardinality: str = "",
        via: str = "",
    ) -> None:
        self._edge(
            cs.LABEL_DATA_ENTITY,
            "id",
            from_data_entity_id,
            cs.REL_RELATES_TO,
            cs.LABEL_DATA_ENTITY,
            "id",
            to_data_entity_id,
            {"cardinality": cardinality, "via": via},
        )

    def link_runs_on(self, component_id: str, technology_choice_id: str) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_RUNS_ON,
            cs.LABEL_TECHNOLOGY_CHOICE,
            "id",
            technology_choice_id,
        )

    def link_has_scenario(self, requirement_item_id: str, scenario_id: str) -> None:
        self._edge(
            cs.LABEL_REQUIREMENT_ITEM,
            "id",
            requirement_item_id,
            cs.REL_HAS_SCENARIO,
            cs.LABEL_SCENARIO,
            "id",
            scenario_id,
        )

    def link_satisfies(self, component_id: str, requirement_item_id: str) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_SATISFIES,
            cs.LABEL_REQUIREMENT_ITEM,
            "id",
            requirement_item_id,
        )

    def link_has_quality_attribute(
        self, requirement_item_id: str, quality_attribute_id: str
    ) -> None:
        self._edge(
            cs.LABEL_REQUIREMENT_ITEM,
            "id",
            requirement_item_id,
            cs.REL_HAS_QUALITY_ATTRIBUTE,
            cs.LABEL_QUALITY_ATTRIBUTE,
            "id",
            quality_attribute_id,
        )

    def link_measured_on(self, quality_attribute_id: str, component_id: str) -> None:
        self._edge(
            cs.LABEL_QUALITY_ATTRIBUTE,
            "id",
            quality_attribute_id,
            cs.REL_MEASURED_ON,
            cs.LABEL_COMPONENT,
            "id",
            component_id,
        )

    def link_addresses_pillar(
        self, quality_attribute_id: str, waf_pillar_name: str
    ) -> None:
        self._edge(
            cs.LABEL_QUALITY_ATTRIBUTE,
            "id",
            quality_attribute_id,
            cs.REL_ADDRESSES_PILLAR,
            cs.LABEL_WAF_PILLAR,
            "name",
            waf_pillar_name,
        )

    def link_must_follow(self, component_id: str, constraint_id: str) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_MUST_FOLLOW,
            cs.LABEL_CONSTRAINT,
            "id",
            constraint_id,
        )

    def link_constrains(self, constraint_id: str, architecture_id: str) -> None:
        self._edge(
            cs.LABEL_CONSTRAINT,
            "id",
            constraint_id,
            cs.REL_CONSTRAINS,
            cs.LABEL_ARCHITECTURE,
            "id",
            architecture_id,
        )

    def link_modeled_by(self, architecture_id: str, capacity_model_id: str) -> None:
        self._edge(
            cs.LABEL_ARCHITECTURE,
            "id",
            architecture_id,
            cs.REL_MODELED_BY,
            cs.LABEL_CAPACITY_MODEL,
            "id",
            capacity_model_id,
        )

    def link_informs(self, capacity_model_id: str, cost_model_id: str) -> None:
        self._edge(
            cs.LABEL_CAPACITY_MODEL,
            "id",
            capacity_model_id,
            cs.REL_INFORMS,
            cs.LABEL_COST_MODEL,
            "id",
            cost_model_id,
        )

    def link_has_line_item(self, cost_model_id: str, cost_line_item_id: str) -> None:
        self._edge(
            cs.LABEL_COST_MODEL,
            "id",
            cost_model_id,
            cs.REL_HAS_LINE_ITEM,
            cs.LABEL_COST_LINE_ITEM,
            "id",
            cost_line_item_id,
        )

    def link_meters(self, cost_line_item_id: str, cloud_service_name: str) -> None:
        self._edge(
            cs.LABEL_COST_LINE_ITEM,
            "id",
            cost_line_item_id,
            cs.REL_METERS,
            cs.LABEL_CLOUD_SERVICE,
            "name",
            cloud_service_name,
        )

    def link_decides(self, adr_id: str, target_label: str, target_id: str) -> None:
        self._edge(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            cs.REL_DECIDES,
            target_label,
            self._key_prop(target_label),
            target_id,
        )

    def link_documents(self, artifact_id: str, architecture_id: str) -> None:
        """``(Artifact)-[:DOCUMENTS]->(Architecture)`` -- same pattern as
        :meth:`link_decides`."""
        self._edge(
            cs.LABEL_ARTIFACT,
            "id",
            artifact_id,
            cs.REL_DOCUMENTS,
            cs.LABEL_ARCHITECTURE,
            "id",
            architecture_id,
        )

    def link_chose(self, adr_id: str, technology_choice_id: str) -> None:
        self._edge(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            cs.REL_CHOSE,
            cs.LABEL_TECHNOLOGY_CHOICE,
            "id",
            technology_choice_id,
        )

    def link_rejected(
        self, adr_id: str, target_label: str, target_id: str, *, reason: str = ""
    ) -> None:
        self._edge(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            cs.REL_REJECTED,
            target_label,
            self._key_prop(target_label),
            target_id,
            {"reason": reason},
        )

    def link_justified_by(self, adr_id: str, requirement_item_id: str) -> None:
        self._edge(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            cs.REL_JUSTIFIED_BY,
            cs.LABEL_REQUIREMENT_ITEM,
            "id",
            requirement_item_id,
        )

    def link_weighs(self, adr_id: str, tradeoff_id: str) -> None:
        self._edge(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            adr_id,
            cs.REL_WEIGHS,
            cs.LABEL_TRADE_OFF,
            "id",
            tradeoff_id,
        )

    def link_supersedes(
        self, new_adr_id: str, old_adr_id: str, *, date: str | None = None
    ) -> None:
        """The evolution path for an Accepted ADR (ruling 1): a NEW node
        supersedes the old one, rather than editing the old one's body."""
        self._edge(
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            new_adr_id,
            cs.REL_SUPERSEDES,
            cs.LABEL_ARCHITECTURE_DECISION,
            "id",
            old_adr_id,
            {"date": date or _now_iso()},
        )

    def link_selects(self, technology_choice_id: str, cloud_service_name: str) -> None:
        self._edge(
            cs.LABEL_TECHNOLOGY_CHOICE,
            "id",
            technology_choice_id,
            cs.REL_SELECTS,
            cs.LABEL_CLOUD_SERVICE,
            "name",
            cloud_service_name,
        )

    def link_forced_by(
        self, technology_choice_id: str, requirement_item_id: str
    ) -> None:
        """The "name the requirement that forces it" gate (§2 rule 6):
        a TechnologyChoice with no FORCED_BY edge is unjustified."""
        self._edge(
            cs.LABEL_TECHNOLOGY_CHOICE,
            "id",
            technology_choice_id,
            cs.REL_FORCED_BY,
            cs.LABEL_REQUIREMENT_ITEM,
            "id",
            requirement_item_id,
        )

    def link_applies(
        self, source_label: str, source_id: str, pattern_name: str
    ) -> None:
        """``(ArchitectureDecision|Component)-[:APPLIES]->(Pattern)``."""
        self._edge(
            source_label,
            self._key_prop(source_label),
            source_id,
            cs.REL_APPLIES,
            cs.LABEL_PATTERN,
            "name",
            pattern_name,
        )

    def link_implements(self, component_id: str, pattern_name: str) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_IMPLEMENTS,
            cs.LABEL_PATTERN,
            "name",
            pattern_name,
        )

    def link_threatens(self, risk_id: str, target_label: str, target_id: str) -> None:
        self._edge(
            cs.LABEL_RISK,
            "id",
            risk_id,
            cs.REL_THREATENS,
            target_label,
            self._key_prop(target_label),
            target_id,
        )

    def link_affects(self, failure_mode_id: str, component_id: str) -> None:
        self._edge(
            cs.LABEL_FAILURE_MODE,
            "id",
            failure_mode_id,
            cs.REL_AFFECTS,
            cs.LABEL_COMPONENT,
            "id",
            component_id,
        )

    def link_mitigated_by(
        self, source_label: str, source_id: str, target_label: str, target_id: str
    ) -> None:
        """``(Risk|FailureMode)-[:MITIGATED_BY]->(Pattern|Constraint)``."""
        self._edge(
            source_label,
            self._key_prop(source_label),
            source_id,
            cs.REL_MITIGATED_BY,
            target_label,
            self._key_prop(target_label),
            target_id,
        )

    def link_has_slo(self, component_id: str, slo_id: str) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_HAS_SLO,
            cs.LABEL_SERVICE_LEVEL_OBJECTIVE,
            "id",
            slo_id,
        )

    def link_measures(self, slo_id: str, quality_attribute_id: str) -> None:
        self._edge(
            cs.LABEL_SERVICE_LEVEL_OBJECTIVE,
            "id",
            slo_id,
            cs.REL_MEASURES,
            cs.LABEL_QUALITY_ATTRIBUTE,
            "id",
            quality_attribute_id,
        )

    def link_alerts_on(self, alert_id: str, slo_id: str) -> None:
        self._edge(
            cs.LABEL_ALERT,
            "id",
            alert_id,
            cs.REL_ALERTS_ON,
            cs.LABEL_SERVICE_LEVEL_OBJECTIVE,
            "id",
            slo_id,
        )

    def link_deployed_as(self, component_id: str, deployment_id: str) -> None:
        self._edge(
            cs.LABEL_COMPONENT,
            "id",
            component_id,
            cs.REL_DEPLOYED_AS,
            cs.LABEL_DEPLOYMENT,
            "id",
            deployment_id,
        )

    def link_in_environment(self, deployment_id: str, environment_id: str) -> None:
        self._edge(
            cs.LABEL_DEPLOYMENT,
            "id",
            deployment_id,
            cs.REL_IN_ENVIRONMENT,
            cs.LABEL_ENVIRONMENT,
            "id",
            environment_id,
        )