scripts.sync_architecture_to_graph

Reference › scripts › sync_architecture_to_graph

scripts/sync_architecture_to_graph.py syncs orchestrata-ultimate's own architecture, the ratified architecture ontology, into the knowledge graph. The architecture is declared as tables in the module (COMPONENTS, CONSTRAINTS, PATTERNS, plus the reference data WAF_PILLARS and CLOUD_SERVICES) and written with ArchitectureWriter; the ADR texts come from orchestrata_ultimate.knowledge_graph.adr_parser. A human runs it, and bootstrap_graphs invokes it with --repo-adrs. Every write is an idempotent MERGE. See orchestrata_ultimate.knowledge_graph.architecture_writer, check_adr_graph_sync (the read-only drift detector for the same ADRs) and the decisions page.

Options: --dry-run (print what would sync without connecting), --seed (also seed the reference-data catalogue first), --adrs [PATH] (also sync the knowledge-wiki ADR corpus, decisions.md, into ArchitectureDecision nodes with Artifact and DOCUMENTS edges; the default path is resolved at call time, and a missing file is a loud skip, never a failure) and --repo-adrs [DIR] (also sync the repository's one-ADR-per-file corpus under docs/adr, DECIDES-linked to the project architecture). sync_adrs and sync_repo_adrs do the two ADR syncs; is_pointer_entry and _is_docs_owned keep a wiki entry that points at a docs/adr record from being written twice; rebaseline_adr re-baselines one ADR's recorded body. A foreign register is refused up front with REFUSED: and exit 2. Body drift alone never fails a run; only errors do (_adr_exit_code).

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

AdrSyncOutcome (dataclass)

Decorators: dataclass(frozen=True) · Line 116

One ADR sync_adrs() could not write cleanly.

AdrSyncReport (dataclass)

Decorators: dataclass(frozen=True) · Line 132

Outcome of one sync_adrs() run.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/sync_architecture_to_graph.py

"""Sync orchestrata-ultimate's own architecture into the knowledge graph.

Thin CLI over ``ArchitectureWriter`` (``knowledge_graph/architecture_writer.py``)
-- the converged writer for the RATIFIED architecture ontology
(``plans/architecture-ontology-draft.md``). Re-expresses this project's own
self-documented architecture (5 infra components, 5 invariants, the
operational-protocol pattern catalog) in ontology terms:

* ``Architecture`` id ``"orchestrata-ultimate:orchestrata"`` (project-qualified,
  ruling 2) -- superseding this script's old global ``Architecture{name}`` node.
* Each ``Component`` project-qualified (``"orchestrata-ultimate:{name}"``).
* The old writer's architecture "rules" become ``Constraint{kind:"invariant"}``
  nodes linked ``Component -[:MUST_FOLLOW]-> Constraint -[:CONSTRAINS]->
  Architecture`` -- the ``Rule`` label is retired here in favour of the
  canonical Tier-A Prolog-rules-only meaning (ruling 3).
* Operational protocols (PID file, health check, ...) become ``Pattern``
  nodes (``category="operational"``) linked ``Component -[:APPLIES]->
  Pattern`` -- superseding the old ``USES_PATTERN`` synonym (§7).

This SUPERSEDES the script's pre-ontology node shapes (global Architecture/
Component/Rule/Pattern, HAS_RULE/USES_PATTERN edges) per §7 of the draft; the
old shapes are not preserved or migrated by this script (that migration is
``canonical_conformance.py``'s job, out of scope here).

Idempotent -- every write is a MERGE through ``ArchitectureWriter``.

``--adrs`` (wave ontology-phase2) additionally graduates the knowledge-wiki's
``decisions.md`` ADR corpus into first-class ``ArchitectureDecision`` nodes
(via ``adr_parser.parse_adrs`` + ``ArchitectureWriter.upsert_architecture_decision``),
links each ``-[:DECIDES]->`` this project's ``Architecture`` node, and records
the decisions page itself plus ``plans/architecture-ontology-draft.md`` as
``Artifact`` nodes ``-[:DOCUMENTS]->`` the same ``Architecture`` node. The
wiki checkout is optional infra (session-start hook machines may lack it) --
a missing ``decisions.md`` is a loud SKIP, never a failure.

``--repo-adrs`` (D5) graduates the OTHER register -- the one-ADR-per-file
factory register under ``docs/adr/NNNN-*.md`` -- into the same
``ArchitectureDecision`` nodes (``parse_adr_file`` + the identical writer
path), each ``-[:DECIDES]->`` the project ``Architecture`` which hangs off the
``Project`` hub via ``HAS_ARCHITECTURE`` (so the node is reachable from the
hub per ``check_kg_connectivity``, which walks relationships undirected).
This is the step ``scripts/bootstrap_graphs.py`` runs, so a full bootstrap
leaves one node per repo ADR. Ids are ``{project}:ADR-{padded_id}`` (ADR-0047:
the 4-digit factory ids never collide with the wiki's 1-3 digit project
register), MERGEd, so a re-run is a no-op. A malformed ADR file aborts the
run BEFORE any repo-ADR write (all files are parsed first).

An ADR that was Accepted in the graph and whose BODY later changed in the
document is NOT rewritten: the writer's immutability guard raises
``AdrImmutabilityError``, which is recorded as ``BODY_DRIFT`` (informational,
exit 0) and the body stays frozen -- evolve a decision via a new ADR +
``link_supersedes``. Title/status/date edits are not body edits and DO flow
through on every re-run.

Per-ADR ``AdrImmutabilityError`` (ruling 1's guard on
``upsert_architecture_decision``) is caught here, not left to abort the run:
the wiki's one-paragraph summary of an Accepted ADR is allowed to fall out of
sync with the graph's frozen body -- that body is owned by ``docs/adr``, not
the wiki page (ruling 1) -- so one drifted ADR must never stop every other
ADR in the corpus from syncing. Each drifted id is recorded as ``BODY_DRIFT``
in the returned :class:`AdrSyncReport` and the loop continues; the body
itself is never overwritten. See ``sync_adrs()`` for the full contract.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

# Ensure repo src is on path when run directly (mirrors kg_schema.py / run_ast_pipeline.py).
_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from orchestrata_ultimate.core.env_bootstrap import load_env_contract  # noqa: E402
from orchestrata_ultimate.knowledge_graph.adr_parser import (  # noqa: E402
    ParsedAdr,
    parse_adr_file,
    parse_adrs,
)
from orchestrata_ultimate.knowledge_graph.architecture_writer import (  # noqa: E402
    CLOUD_SERVICES,
    WAF_PILLARS,
    AdrImmutabilityError,
    ArchitectureWriter,
    _adr_body_sha,
)
from orchestrata_ultimate.knowledge_graph.canonical_schema import (  # noqa: E402
    CANONICAL_LABELS,
    LABEL_ARCHITECTURE,
    LABEL_COMPONENT,
)

# ── ADR corpus (wave ontology-phase2) ───────────────────────────────────────
# The decisions.md page + this project's own ontology design doc, recorded
# as Artifact nodes -[:DOCUMENTS]-> Architecture (§9 of the draft).
ADRS_ARTIFACT_PATH = "projects/orchestrata-ultimate/decisions.md"
DESIGN_DOC_ARTIFACT_PATH = "plans/architecture-ontology-draft.md"


def _default_adrs_path() -> str:
    """Resolve the knowledge-wiki ``decisions.md`` path AT CALL TIME (never
    at import time, per project doctrine) from ``WIKI_ROOT`` -- falls back to
    ``~/workspace/knowledge-wiki`` when unset."""
    root = os.environ.get("WIKI_ROOT", str(Path.home() / "workspace/knowledge-wiki"))
    return str(Path(root) / "projects" / "orchestrata-ultimate" / "decisions.md")


@dataclass(frozen=True)
class AdrSyncOutcome:
    """One ADR ``sync_adrs()`` could not write cleanly.

    ``state`` is ``"BODY_DRIFT"`` for the only case this loop produces today
    (an Accepted node whose frozen body the wiki summary no longer matches).
    """

    adr_id: str
    state: str
    detail: str

    def to_dict(self) -> dict[str, str]:
        return {"id": self.adr_id, "state": self.state, "detail": self.detail}


@dataclass(frozen=True)
class AdrSyncReport:
    """Outcome of one ``sync_adrs()`` run.

    ``body_drift`` counts ``ArchitectureDecision`` nodes left untouched
    because ``upsert_architecture_decision`` raised ``AdrImmutabilityError``
    -- informational, never a failure (ruling 1: the body of an Accepted ADR
    is owned by ``docs/adr``, not the wiki summary). ``errors`` is reserved
    for anything else that interrupts a single ADR's write without aborting
    the whole run; ``sync_adrs()`` never populates it today because its
    per-ADR ``try/except`` catches ONLY ``AdrImmutabilityError`` -- every
    other exception still propagates and fails the run loudly, unchanged.
    """

    synced: int
    created: int
    body_drift: int
    errors: int
    outcomes: list[AdrSyncOutcome] = field(default_factory=list)
    pointer_skipped: int = 0

    def summary_line(self) -> str:
        line = (
            f"synced {self.synced} · created {self.created} · "
            f"body_drift {self.body_drift} · errors {self.errors}"
        )
        if self.pointer_skipped:
            line += f" · pointer_skipped {self.pointer_skipped}"
        return line

    def to_dict(self) -> dict[str, Any]:
        return {
            "synced": self.synced,
            "created": self.created,
            "body_drift": self.body_drift,
            "errors": self.errors,
            "pointer_skipped": self.pointer_skipped,
            "outcomes": [o.to_dict() for o in self.outcomes],
        }


def _adr_exit_code(report: AdrSyncReport) -> int:
    """Body drift alone never fails the run -- only ``errors`` does."""
    return 1 if report.errors else 0


def _write_parsed_adrs(
    writer: ArchitectureWriter, parsed: list[ParsedAdr], arch_id: str
) -> tuple[int, list[AdrSyncOutcome]]:
    """MERGE each parsed ADR and link it ``-[:DECIDES]->`` ``arch_id``.

    Catches ONLY ``AdrImmutabilityError`` per ADR (recorded as
    ``BODY_DRIFT``); anything else propagates and aborts the run.
    """
    synced = 0
    outcomes: list[AdrSyncOutcome] = []
    for adr in parsed:
        node_id = f"{PROJECT}:ADR-{adr.padded_id}"
        try:
            adr_id = writer.upsert_architecture_decision(
                PROJECT,
                adr.number,
                adr.title,
                padded_id=adr.padded_id,
                register=adr.register,
                status=adr.status or "Proposed",
                date=adr.date,
                context=adr.context,
                decision=adr.decision,
                alternatives=adr.alternatives,
                consequences=adr.consequences,
            )
        except AdrImmutabilityError as exc:
            outcomes.append(AdrSyncOutcome(node_id, "BODY_DRIFT", str(exc)))
            continue
        writer.link_decides(adr_id, LABEL_ARCHITECTURE, arch_id)
        synced += 1
    return synced, outcomes


#: A wiki entry that points at the full record: ``Full record: docs/adr/<file>``
#: (with or without backticks / a trailing period). The parser folds a
#: section to one paragraph, so the marker can sit mid-body (ADR-0073's entry
#: carries an execution note after it).
_POINTER_RE = re.compile(r"Full record: `?docs/adr/[^\s`]+\.md`?\.?")


def is_pointer_entry(adr: ParsedAdr) -> bool:
    """True when the wiki entry is a pointer stub to its ``docs/adr`` record."""
    body = " ".join((adr.context, adr.decision, adr.alternatives, adr.consequences))
    return _POINTER_RE.search(body) is not None


def _is_docs_owned(adr: ParsedAdr, docs_dir: Path) -> bool:
    """A factory-register wiki entry whose ``docs/adr`` file exists, or that is
    a pointer stub, is owned by ``docs/adr`` (ADR-0096): the wiki pass never
    hashes or writes its body. Project-register entries are the wiki's own."""
    if adr.register != "factory":
        return False
    return is_pointer_entry(adr) or any(docs_dir.glob(f"{adr.padded_id}-*.md"))


def sync_adrs(
    writer: ArchitectureWriter,
    adrs_path: Path,
    arch_id: str,
    adr_dir: Path | None = None,
) -> AdrSyncReport:
    """Graduate the ADR corpus at ``adrs_path`` into ``ArchitectureDecision``
    nodes -[:DECIDES]-> ``arch_id``, and record the decisions page itself
    plus the ontology design doc as ``Artifact`` nodes -[:DOCUMENTS]-> the
    same architecture.

    Each ADR's ``upsert_architecture_decision`` call is wrapped in its own
    ``try/except`` catching ONLY ``AdrImmutabilityError`` -- the matched node
    is Accepted and this call would change its frozen body. That id is
    recorded as ``BODY_DRIFT`` in the returned report and the loop moves on
    to the next ADR; the body is never overwritten. Any other exception
    still propagates uncaught and aborts the run -- that is ruling 1's guard
    working, unchanged.

    Returns:
        An :class:`AdrSyncReport` (``synced``/``created``/``body_drift``/
        ``errors`` counts + the per-ADR ``outcomes`` for anything not
        cleanly synced).
    """
    all_parsed = parse_adrs(adrs_path.read_text(encoding="utf-8"))
    docs_dir = Path(adr_dir) if adr_dir is not None else Path(_default_repo_adr_dir())
    parsed = [a for a in all_parsed if not _is_docs_owned(a, docs_dir)]
    pointer_skipped = len(all_parsed) - len(parsed)
    synced, outcomes = _write_parsed_adrs(writer, parsed, arch_id)

    decisions_artifact_id = writer.upsert_artifact(
        PROJECT, "decisions.md", artifact_type="adr", path=ADRS_ARTIFACT_PATH
    )
    writer.link_documents(decisions_artifact_id, arch_id)

    design_doc_artifact_id = writer.upsert_artifact(
        PROJECT,
        "architecture-ontology-draft.md",
        artifact_type="design-doc",
        path=DESIGN_DOC_ARTIFACT_PATH,
    )
    writer.link_documents(design_doc_artifact_id, arch_id)

    body_drift = sum(1 for o in outcomes if o.state == "BODY_DRIFT")
    errors = sum(1 for o in outcomes if o.state == "ERROR")
    return AdrSyncReport(
        synced=synced,
        created=2,
        body_drift=body_drift,
        errors=errors,
        outcomes=outcomes,
        pointer_skipped=pointer_skipped,
    )


def _default_repo_adr_dir() -> str:
    """The repo's ``docs/adr`` directory (the factory register)."""
    return str(Path(__file__).resolve().parent.parent / "docs" / "adr")


def sync_repo_adrs(
    writer: ArchitectureWriter, adr_dir: Path, arch_id: str
) -> AdrSyncReport:
    """Graduate every ``adr_dir/NNNN-*.md`` into ``ArchitectureDecision``
    nodes -[:DECIDES]-> ``arch_id`` (``README.md`` excluded).

    All files are parsed BEFORE the first write, so a malformed ADR raises
    ``ValueError`` (from ``parse_adr_file``) without a partial sync. Writes go
    through the same per-ADR path as :func:`sync_adrs`, so the immutability
    guard behaves identically (``BODY_DRIFT``, body never overwritten).
    ``created`` is 0: this register records no Artifact nodes.
    """
    paths = sorted(p for p in adr_dir.glob("*.md") if p.name.lower() != "readme.md")
    parsed = [parse_adr_file(p.read_text(encoding="utf-8")) for p in paths]
    synced, outcomes = _write_parsed_adrs(writer, parsed, arch_id)
    return AdrSyncReport(
        synced=synced,
        created=0,
        body_drift=sum(1 for o in outcomes if o.state == "BODY_DRIFT"),
        errors=sum(1 for o in outcomes if o.state == "ERROR"),
        outcomes=outcomes,
    )


#: Matches a literal ``ADR-NNN``/``ADR-NNNN`` token in an authority ADR's
#: Decision section (ADR-0102's naming mechanism).
_AUTHORITY_ID_RE = re.compile(r"ADR-(\d{1,4})")


def _parse_authority(
    adr_dir: Path, authority: str
) -> tuple[frozenset[str], str] | None:
    """Parse the authority ADR's own docs record in ``adr_dir`` (ADR-0102).

    Returns ``(named_ids, status)`` -- the padded ``ADR-NNNN`` ids the
    authority's Decision section names, and the record's normalised status --
    or ``None`` when the authority has no parseable docs file there (the
    writer's hardcoded authority map is the fallback for those, ADR-0096's
    original path).
    """
    padded = authority.removeprefix("ADR-")
    matches = sorted(adr_dir.glob(f"{padded}-*.md"))
    if len(matches) != 1:
        return None
    try:
        record = parse_adr_file(matches[0].read_text(encoding="utf-8"))
    except ValueError:
        return None
    named = frozenset(
        f"ADR-{int(digits):04d}" for digits in _AUTHORITY_ID_RE.findall(record.decision)
    )
    return named, record.status


def rebaseline_adr(
    writer: ArchitectureWriter,
    adr_dir: Path,
    adr_label: str,
    authority: str,
    wiki_path: Path | None = None,
) -> str:
    """Re-baseline ONE ADR node to its record's body under ``authority``.

    ``adr_label`` is the exact ``ADR-NNNN`` id. When the authority ADR's own
    docs record parses (ADR-0102's generic mechanism), the ids its Decision
    section names govern: a named id re-baselines from its ``docs/adr``
    record at WHATEVER status that record carries (a Proposed record stays
    Proposed), or -- when the id has no docs record -- from its wiki register
    entry (``wiki_path``: the ``## ADR-NNNN:`` heading, its ``**Status:**``
    line and folded body). An unnamed id is refused with ``ValueError``
    (``"<authority> does not name <id>"``); the writer independently
    re-checks both the naming and the authority's Accepted status. When the
    authority's docs record cannot be parsed, the legacy path is unchanged:
    exactly one docs file, an Accepted record, and the writer's hardcoded
    authority map (ADR-0096). Logs the act; returns the node id.
    """
    named_ids: frozenset[str] | None = None
    authority_status: str | None = None
    parsed_authority = _parse_authority(adr_dir, authority)
    if parsed_authority is not None:
        named_ids, authority_status = parsed_authority
        if adr_label not in named_ids:
            raise ValueError(f"{authority} does not name {adr_label}")

    padded = adr_label.removeprefix("ADR-")
    matches = sorted(adr_dir.glob(f"{padded}-*.md"))
    if not matches and named_ids is not None:
        return _rebaseline_from_wiki(
            writer,
            wiki_path,
            adr_label,
            authority,
            named_ids=named_ids,
            authority_status=authority_status,
        )
    if len(matches) != 1:
        raise ValueError(f"expected exactly one docs/adr file for {adr_label}")
    adr = parse_adr_file(matches[0].read_text(encoding="utf-8"))
    if named_ids is None and adr.status != "Accepted":
        raise ValueError(f"{adr_label} is {adr.status!r} in docs, not Accepted")
    before = writer.read_architecture_decision_hashes(PROJECT).get(adr.padded_id, "")
    node_id = writer.rebaseline_architecture_decision(
        PROJECT,
        adr.padded_id,
        authority=authority,
        title=adr.title,
        status=adr.status,
        date=adr.date,
        context=adr.context,
        decision=adr.decision,
        alternatives=adr.alternatives,
        consequences=adr.consequences,
        named_ids=named_ids,
        authority_status=authority_status,
    )
    after = _adr_body_sha(adr.context, adr.decision, adr.alternatives, adr.consequences)
    print(f"REBASELINED: {node_id} {before[:10]} -> {after[:10]} under {authority}")
    return node_id


def _rebaseline_from_wiki(
    writer: ArchitectureWriter,
    wiki_path: Path | None,
    adr_label: str,
    authority: str,
    *,
    named_ids: frozenset[str],
    authority_status: str | None,
) -> str:
    """Re-baseline a wiki-only id from its register entry (ADR-0102 rule 1).

    The id has no ``docs/adr`` record; its ``## ADR-NNNN:`` heading's
    ``**Status:**`` line and folded body ARE the record. The writer still
    enforces the authority (Accepted, names the id) and node existence.
    """
    padded = adr_label.removeprefix("ADR-")
    if wiki_path is None or not wiki_path.exists():
        raise ValueError(
            f"{adr_label} has no docs/adr record and no wiki register to "
            f"re-baseline from (wiki path: {wiki_path})"
        )
    entries = {
        entry.padded_id: entry
        for entry in parse_adrs(wiki_path.read_text(encoding="utf-8"))
    }
    entry = entries.get(padded)
    if entry is None:
        raise ValueError(f"{adr_label} has no entry in the wiki register {wiki_path}")
    before = writer.read_architecture_decision_hashes(PROJECT).get(entry.padded_id, "")
    node_id = writer.rebaseline_architecture_decision(
        PROJECT,
        entry.padded_id,
        authority=authority,
        title=entry.title,
        status=entry.status,
        date=entry.date,
        context=entry.context,
        decision=entry.decision,
        alternatives=entry.alternatives,
        consequences=entry.consequences,
        named_ids=named_ids,
        authority_status=authority_status,
    )
    after = _adr_body_sha(
        entry.context, entry.decision, entry.alternatives, entry.consequences
    )
    print(f"REBASELINED: {node_id} {before[:10]} -> {after[:10]} under {authority}")
    return node_id


def compute_first_capture_set(
    writer: ArchitectureWriter, wiki_path: Path, adr_dir: Path
) -> tuple[list[tuple[ParsedAdr, str, str | None]], list[tuple[str, str]]]:
    """The ADR-0098 first-capture set -- computed, never listed.

    An id is in the set when a docs file exists for it, the graph node exists
    with a body that differs from docs, and that graph body hash is the
    empty-body hash or the hash of the wiki entry's parsed body (it was never
    captured from docs). A differing body that is neither is real drift and is
    returned as ``(padded_id, reason)`` in the refused list, never touched.

    Returns:
        ``(eligible, refused)``; each eligible item is ``(docs_adr,
        graph_sha, wiki_sha_or_None)``. Read-only.
    """
    graph = writer.read_architecture_decision_hashes(PROJECT)
    wiki_shas: dict[str, str] = {}
    if wiki_path.exists():
        for entry in parse_adrs(wiki_path.read_text(encoding="utf-8")):
            if entry.register == "factory":
                wiki_shas[entry.padded_id] = _adr_body_sha(
                    entry.context,
                    entry.decision,
                    entry.alternatives,
                    entry.consequences,
                )
    empty_sha = _adr_body_sha("", "", "", "")
    eligible: list[tuple[ParsedAdr, str, str | None]] = []
    refused: list[tuple[str, str]] = []
    for path in sorted(
        p for p in adr_dir.glob("*.md") if p.name.lower() != "readme.md"
    ):
        adr = parse_adr_file(path.read_text(encoding="utf-8"))
        graph_sha = graph.get(adr.padded_id)
        docs_sha = _adr_body_sha(
            adr.context, adr.decision, adr.alternatives, adr.consequences
        )
        if graph_sha is None or graph_sha == docs_sha:
            continue
        wiki_sha = wiki_shas.get(adr.padded_id)
        if graph_sha == empty_sha or graph_sha == wiki_sha:
            eligible.append((adr, graph_sha, wiki_sha))
        else:
            refused.append(
                (adr.padded_id, f"graph body {graph_sha[:10]} is not empty or wiki")
            )
    return eligible, refused


def rebaseline_first_capture_run(
    writer: ArchitectureWriter, wiki_path: Path, adr_dir: Path, authority: str
) -> tuple[int, int]:
    """Baseline every computed first-capture id from docs, once, logged.

    Prints one ``REBASELINED:`` line per id (before/after hashes) and one
    ``REFUSED:`` line per differing id that is not a first capture. Returns
    ``(rebaselined, refused)``.
    """
    if authority != "ADR-0098":
        raise ValueError(f"{authority!r} does not authorise a first-capture run")
    eligible, refused = compute_first_capture_set(writer, wiki_path, adr_dir)
    for adr, _graph_sha, wiki_sha in eligible:
        before, after = writer.rebaseline_first_capture(
            PROJECT,
            adr.padded_id,
            authority=authority,
            wiki_body_sha=wiki_sha,
            title=adr.title,
            status=adr.status or "Proposed",
            date=adr.date,
            context=adr.context,
            decision=adr.decision,
            alternatives=adr.alternatives,
            consequences=adr.consequences,
        )
        print(
            f"REBASELINED: {PROJECT}:ADR-{adr.padded_id} {before[:10]} -> "
            f"{after[:10]} under {authority}"
        )
    for padded_id, reason in refused:
        print(f"REFUSED: {PROJECT}:ADR-{padded_id}: {reason}")
    return len(eligible), len(refused)


def _allowed_register_roots() -> tuple[Path, Path]:
    """ADR-0102 rule 3's two register roots: this repository's root (from
    the script's own location) and the wiki's
    ``projects/orchestrata-ultimate/`` directory (``WIKI_ROOT`` read at call
    time, exactly as ``_default_adrs_path`` reads it). Both sides are
    symlink-resolved by the caller before comparison."""
    repo_root = Path(__file__).resolve().parents[1]
    wiki_root = (
        Path(os.environ.get("WIKI_ROOT", str(Path.home() / "workspace/knowledge-wiki")))
        / "projects"
        / "orchestrata-ultimate"
    )
    return repo_root, wiki_root.resolve()


def _foreign_register_refusal(args: argparse.Namespace) -> str | None:
    """ADR-0102 rule 3: refuse a register that is not this repository's.

    ``--adrs`` and ``--repo-adrs`` must each resolve under one of the two
    allowed roots until a project key makes a foreign path meaningful.
    Returns the refusal message, or ``None`` when every given path is
    admitted. Runs before any graph write, in every mode.
    """
    roots = _allowed_register_roots()
    for given in (args.adrs, args.repo_adrs):
        if given is None:
            continue
        resolved = Path(given).resolve()
        if not any(resolved.is_relative_to(root) for root in roots):
            return (
                f"foreign register {given}; ADR-0102 rule 3: no sync of "
                "another project's ADRs until a project key exists"
            )
    return None


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase environment contract.

    Delegates to ``core.env_bootstrap.load_env_contract`` — THE canonical
    loader — instead of parsing ``.env`` here. The hand-rolled copy this
    replaced (byte-identical in four scripts) stopped after phase 1, and
    ``.env`` is the prod-BASE layer: it carries ``CXP_GRPC_HOST=localhost``
    for the host user-units, so a script that loads it and stops resolves
    PRODUCTION endpoints while believing it is on the test fleet. The
    canonical loader then applies the ``ORCHESTRATA_ENV_PROFILE`` overlay,
    which is the half that points the process at the right infrastructure,
    and honours ``declare_owned`` so it cannot reset a test worker's
    per-slot isolation.
    """
    load_env_contract()


# ── Project + Architecture ──────────────────────────────────────────────────
PROJECT = "orchestrata-ultimate"
ARCHITECTURE_NAME = "orchestrata"

ARCHITECTURE = {
    "version": "0.7.0",
    "description": "Multi-agent software engineering system with browser UI (VizInt), WebSocket gateway, gRPC kernel, Redis queues, and knowledge graph.",
    "status": "active",
}

# ── Components ───────────────────────────────────────────────────────────
# Must match the canonical PATTERNS names exactly (below), or the
# link_applies() call silently binds nothing (MATCH on an unknown Pattern.name).
COMPONENTS: list[dict[str, Any]] = [
    {
        "name": "VizInt UI",
        "protocol": "HTTP",
        "compute_platform": "python -m http.server (port 5000)",
        "required": True,
        "responsibility": "Browser-based command center with 31 gadgets. Served independently by python -m http.server. WebSocket bridge connects to Gateway.",
        "invariants": [
            "Gateway never serves HTML/JS/CSS",
            "VizInt is independent of Gateway",
        ],
        "protocols": [],
    },
    {
        "name": "Gateway",
        "protocol": "WebSocket (JSON-RPC 2.0)",
        "compute_platform": "asyncio process (port 8765)",
        "required": True,
        "responsibility": "WebSocket JSON-RPC server. Authenticates clients, dispatches RPCs, manages pub/sub, broadcasts pipeline events. Connects to the out-of-process kernel via the cxpt direct transport (grpcio retired 2026-08-07).",
        "invariants": [
            "Gateway never serves HTML/JS/CSS",
            "kernel is out-of-process (cxpt direct transport)",
            "Never use except: pass",
            "Never hardcode service IPs",
        ],
        "protocols": [
            "PID file",
            "Health check",
            "Readiness probe",
            "Graceful shutdown",
            "Idempotent start",
            "Foreign process detection",
        ],
    },
    {
        "name": "Kernel",
        "protocol": "cxpt direct transport (port 50053)",
        "compute_platform": "asyncio process",
        "required": True,
        "responsibility": "Out-of-process agent runtime. Boots full PipelineRunner with all 9 volks. Gateway connects via the cxpt direct transport (grpcio retired 2026-08-07); falls back to in-process PipelineKernelClient when unreachable.",
        "invariants": [
            "kernel is out-of-process (cxpt direct transport)",
            "Never hardcode service IPs",
        ],
        "protocols": [],
    },
    {
        "name": "Redis",
        "protocol": "TCP (port 6379)",
        "compute_platform": "Incus container (redis-queue)",
        "required": True,
        "responsibility": "Message broker for CXP packets, session state cache, pipeline event stream (gw:ringbuf:pipeline.events). Runs in Incus container.",
        "invariants": ["Never hardcode service IPs"],
        "protocols": [],
    },
    {
        "name": "orchgraph",
        "protocol": "Bolt (port 7687)",
        "compute_platform": "Incus HA cluster (orchgraph — Memgraph 3.11 fork: coordinators + main/replica)",
        "required": True,
        "responsibility": "Operational knowledge graph (orchkg database): code structure, specs/requirements, sessions, audit trail, architecture ontology. The Memgraph semantic twin lives in the 'memgraph' database on the same cluster. Neo4j was retired 2026-07-27; the neo4j:// scheme and the Python neo4j driver are Bolt-client naming only.",
        "invariants": ["Never hardcode service IPs"],
        "protocols": [],
    },
]

# ── Constraints (formerly "Rules" -- ruling 3: Rule is Prolog-only now) ─────
CONSTRAINTS = [
    {
        "name": "Gateway never serves HTML/JS/CSS",
        "severity": "CRITICAL",
        "description": "The gateway is a WebSocket JSON-RPC server. Adding process_request for static files is an architectural violation. VizInt UI served independently on port 5000.",
    },
    {
        "name": "VizInt is independent of Gateway",
        "severity": "CRITICAL",
        "description": "Browser loads UI from port 5000. WebSocketBridge connects to gateway at ws://localhost:8765. E2E tests use localStorage.setItem('vizint.gateway.url', ...) for dynamic port configuration.",
    },
    {
        "name": "Never use except: pass",
        "severity": "CRITICAL",
        "description": "Every exception handler must log the actual error at the appropriate level (INFO/WARNING/ERROR). Silent failures are the hardest bugs.",
    },
    {
        "name": "kernel is out-of-process (cxpt direct transport)",
        "severity": "HIGH",
        "description": "Gateway connects to kernel via the cxpt direct transport (grpcio retired 2026-08-07) on port 50053. Three-tier logging: unconfigured (INFO), unreachable (WARNING with host:port+error), connected (INFO). Falls back to PipelineKernelClient only when unreachable.",
    },
    {
        "name": "Never hardcode service IPs",
        "severity": "CRITICAL",
        "description": "All infrastructure addresses come from .env. Both start_test_services.py and gateway CLI auto-load .env via _load_dotenv(). Never assume localhost for Redis/graph.",
    },
    {
        "name": "Never dismiss a test failure",
        "severity": "CRITICAL",
        "description": "Every test failure must be traced to root cause. 'Pre-existing', 'flaky', 'race condition' are not valid dismissals. Fix it or report you cannot.",
    },
]

# ── Patterns (operational protocols; category="operational" per §7) ────────
PATTERNS = [
    {
        "name": "PID file",
        "description": "Write PID on start, validate liveness with os.kill(pid, 0), remove on graceful stop.",
    },
    {
        "name": "Health check",
        "description": "Application-level RPC (e.g. gateway.ping), not just port-open. Port can be open while process is hung.",
    },
    {
        "name": "Readiness probe",
        "description": "Wait for health check response, not just TCP accept. Service may accept TCP but still be loading dependencies.",
    },
    {
        "name": "Graceful shutdown",
        "description": "SIGTERM -> drain connections (10s) -> SIGKILL as last resort -> cleanup PID file.",
    },
    {
        "name": "Idempotent start",
        "description": "Check PID + port + health before spawning. Safe to run start_test_services.py N times.",
    },
    {
        "name": "Foreign process detection",
        "description": "If port is in use but no matching PID file, refuse to start. Prevents hijacking another process's port.",
    },
    {
        "name": "Stale PID cleanup",
        "description": "If PID file exists but process is dead, auto-remove it so restart is not blocked.",
    },
    {
        "name": "Three-tier gRPC logging",
        "description": "Unconfigured (no env vars) -> INFO. Unreachable (server down) -> WARNING with host:port + error. Connected -> INFO.",
    },
    {
        "name": "Exponential backoff",
        "description": "Reply consumer uses 2s->4s->8s->...->60s max backoff. Logs actual error on first failure only. Stops after 10 consecutive failures.",
    },
]


def sync(writer: ArchitectureWriter) -> dict[str, int]:
    """Run the full idempotent MERGE sequence through ``writer``.

    Order matters: Constraints and Patterns are created BEFORE Components,
    same reasoning the pre-ontology writer documented -- a MATCH on a
    not-yet-created node yields no rows and silently drops the edge, so on a
    fresh graph the component->constraint/pattern links would never bind.

    Returns:
        A summary count dict (architecture/components/constraints/patterns).
    """
    arch_id = writer.upsert_architecture(
        PROJECT,
        ARCHITECTURE_NAME,
        version=ARCHITECTURE["version"],
        description=ARCHITECTURE["description"],
        status=ARCHITECTURE["status"],
    )

    constraint_ids: dict[str, str] = {}
    for con in CONSTRAINTS:
        cid = writer.upsert_constraint(
            PROJECT,
            con["name"],
            kind="invariant",
            severity=con["severity"],
            description=con["description"],
        )
        writer.link_constrains(cid, arch_id)
        constraint_ids[con["name"]] = cid

    for pat in PATTERNS:
        writer.upsert_pattern(
            pat["name"], category="operational", description=pat["description"]
        )

    for comp in COMPONENTS:
        comp_id = writer.upsert_component(
            PROJECT,
            comp["name"],
            responsibility=comp["responsibility"],
            compute_platform=comp["compute_platform"],
            protocol=comp["protocol"],
            required=comp["required"],
        )
        writer.link_has_component(arch_id, comp_id)
        for invariant_name in comp.get("invariants", []):
            linked_cid = constraint_ids.get(invariant_name)
            if linked_cid is not None:
                writer.link_must_follow(comp_id, linked_cid)
        for pat_name in comp.get("protocols", []):
            writer.link_applies(LABEL_COMPONENT, comp_id, pat_name)

    return {
        "architecture": 1,
        "components": len(COMPONENTS),
        "constraints": len(CONSTRAINTS),
        "patterns": len(PATTERNS),
    }


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Sync orchestrata-ultimate's own architecture into the "
        "knowledge graph (RATIFIED architecture ontology)."
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print what would be synced without connecting to the graph at all.",
    )
    parser.add_argument(
        "--seed",
        action="store_true",
        help="Also seed the reference-data catalog (WAFPillar + CloudService "
        "nodes, via ArchitectureWriter.seed_reference_data()) before the sync.",
    )
    parser.add_argument(
        "--adrs",
        nargs="?",
        const=_default_adrs_path(),
        default=None,
        help="Also sync the knowledge-wiki ADR corpus (decisions.md) into "
        "ArchitectureDecision nodes + Artifact/DOCUMENTS edges. Optional PATH "
        "overrides the default ($WIKI_ROOT/projects/orchestrata-ultimate/"
        "decisions.md, resolved at call time). A missing file is a loud SKIP, "
        "never a failure.",
    )
    parser.add_argument(
        "--repo-adrs",
        nargs="?",
        const=_default_repo_adr_dir(),
        default=None,
        metavar="DIR",
        help="Also sync the repo's one-ADR-per-file corpus (docs/adr/NNNN-*.md) "
        "into ArchitectureDecision nodes DECIDES-linked to the project "
        "Architecture. Optional DIR overrides docs/adr. A missing directory is "
        "a loud SKIP, never a failure. scripts/bootstrap_graphs.py passes this.",
    )
    parser.add_argument(
        "--report-json",
        default=None,
        metavar="PATH",
        help="With --adrs and/or --repo-adrs, write the per-ADR sync report "
        "(the synced/created/body_drift/errors counts plus every BODY_DRIFT row) "
        "as JSON to PATH. With both, the repo report is nested under "
        '"repo_adrs" in the --adrs report.',
    )
    parser.add_argument(
        "--rebaseline-id",
        default=None,
        metavar="ADR-NNNN",
        help="Re-baseline ONE ADR node's body to its docs/adr record (ADR-0096). "
        "Requires --authority; refuses any id the authority does not name.",
    )
    parser.add_argument(
        "--rebaseline-first-capture",
        action="store_true",
        help="ADR-0098: baseline once from docs every id whose graph body is "
        "empty or the wiki entry's body (computed set; refuses real drift). "
        "Requires --authority ADR-0098; uses --adrs / --repo-adrs paths.",
    )
    parser.add_argument(
        "--authority",
        default=None,
        metavar="ADR-NNNN",
        help="The Accepted ruling ADR that authorises --rebaseline-id.",
    )
    parser.add_argument(
        "--retire-legacy",
        action="append",
        default=None,
        metavar="LABEL:id",
        help="DETACH DELETE the node with the given label and canonical id "
        "(project-qualified: orchestrata-ultimate:...). Refuses ids that do "
        "not start with 'orchestrata-ultimate:'. Refuses labels that are not "
        "canonical schema labels. Prints the target and the pre-delete property "
        "snapshot. Exits non-zero if the node does not exist. This is the ONLY "
        "delete path; default runs NEVER delete anything. Coordinator-run, "
        "one-shot, per authorized correction.",
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    """Sync architecture metadata to the graph. Idempotent via MERGE."""
    args = _build_parser().parse_args(argv)
    _load_dotenv()

    refusal = _foreign_register_refusal(args)
    if refusal is not None:
        print(f"REFUSED: {refusal}")
        return 2

    if args.dry_run:
        if args.seed:
            print(
                f"Would seed reference data: {len(WAF_PILLARS)} WAFPillar(s), "
                f"{len(CLOUD_SERVICES)} CloudService(s)."
            )
        print(
            f"Would sync Architecture '{PROJECT}:{ARCHITECTURE_NAME}': "
            f"{len(COMPONENTS)} component(s), {len(CONSTRAINTS)} constraint(s), "
            f"{len(PATTERNS)} pattern(s)."
        )
        for comp in COMPONENTS:
            print(f"  - Component: {comp['name']}")
        for con in CONSTRAINTS:
            print(f"  - Constraint (invariant): {con['name']}")
        for pat in PATTERNS:
            print(f"  - Pattern (operational): {pat['name']}")
        if args.adrs is not None:
            adrs_path = Path(args.adrs)
            if not adrs_path.exists():
                print(f"SKIP adrs: {adrs_path} does not exist")
            else:
                parsed = parse_adrs(adrs_path.read_text(encoding="utf-8"))
                print(f"Would sync {len(parsed)} ADRs + 2 artifacts")
        if args.repo_adrs is not None:
            repo_dir = Path(args.repo_adrs)
            if not repo_dir.is_dir():
                print(f"SKIP repo-adrs: {repo_dir} does not exist")
            else:
                n = sum(
                    1 for p in repo_dir.glob("*.md") if p.name.lower() != "readme.md"
                )
                print(f"Would sync {n} repo ADRs from {repo_dir}")
        return 0

    if args.rebaseline_id or args.authority or args.rebaseline_first_capture:
        if bool(args.rebaseline_id) == bool(args.rebaseline_first_capture) or not (
            args.authority
        ):
            print(
                "give --authority with exactly one of --rebaseline-id or "
                "--rebaseline-first-capture"
            )
            return 2
        writer = ArchitectureWriter()
        try:
            adr_dir = Path(args.repo_adrs or _default_repo_adr_dir())
            if args.rebaseline_first_capture:
                done, refused = rebaseline_first_capture_run(
                    writer,
                    Path(args.adrs or _default_adrs_path()),
                    adr_dir,
                    args.authority,
                )
                print(f"first-capture: rebaselined {done} · refused {refused}")
            else:
                rebaseline_adr(
                    writer,
                    adr_dir,
                    args.rebaseline_id,
                    args.authority,
                    Path(args.adrs or _default_adrs_path()),
                )
        except ValueError as exc:
            print(f"REFUSED: {exc}")
            return 2
        finally:
            writer.close()
        return 0

    # ── --retire-legacy ─────────────────────────────────────────────────────
    if args.retire_legacy:
        writer = ArchitectureWriter()
        try:
            for item in args.retire_legacy:
                if ":" not in item:
                    print(
                        f"REFUSED: --retire-legacy expects LABEL:id format, got {item!r}"
                    )
                    return 2
                label, node_id = item.split(":", 1)
                if label not in CANONICAL_LABELS:
                    print(
                        f"REFUSED: --retire-legacy label must be a canonical schema label, got {label!r}"
                    )
                    return 2
                if not node_id.startswith("orchestrata-ultimate:"):
                    print(
                        f"REFUSED: --retire-legacy ids must start with 'orchestrata-ultimate:', got {node_id!r}"
                    )
                    return 2
                # Read the node first to print its snapshot
                props = writer._read(
                    f"MATCH (n:{label} {{id: $id}}) RETURN properties(n) AS props",
                    {"id": node_id},
                )
                if not props:
                    print(f"REFUSED: node {item} does not exist")
                    return 2
                snapshot = props[0]["props"]
                print(f"RETIRE LEGACY: {item}")
                print(
                    f"  Pre-delete snapshot: {json.dumps(snapshot, indent=2, default=str)}"
                )
                writer._write(
                    f"MATCH (n:{label} {{id: $id}}) DETACH DELETE n",
                    {"id": node_id},
                )
                print(f"  ✅ Deleted {item}")
        finally:
            writer.close()
        return 0

    writer = ArchitectureWriter()
    exit_code = 0
    try:
        if args.seed:
            seed_summary = writer.seed_reference_data(PROJECT)
            print(f"✅ Reference data seeded to knowledge graph: {seed_summary}")
        summary = sync(writer)
        wiki_report: AdrSyncReport | None = None
        if args.adrs is not None:
            adrs_path = Path(args.adrs)
            if not adrs_path.exists():
                print(f"SKIP adrs: {adrs_path} does not exist")
            else:
                arch_id = f"{PROJECT}:{ARCHITECTURE_NAME}"
                adr_report = sync_adrs(writer, adrs_path, arch_id)
                summary.update(
                    {"adrs": adr_report.synced, "artifacts": adr_report.created}
                )
                print(adr_report.summary_line())
                for outcome in adr_report.outcomes:
                    print(f"  {outcome.state}: {outcome.adr_id}: {outcome.detail}")
                wiki_report = adr_report
                exit_code = _adr_exit_code(adr_report)
        repo_report: AdrSyncReport | None = None
        if args.repo_adrs is not None:
            repo_dir = Path(args.repo_adrs)
            if not repo_dir.is_dir():
                print(f"SKIP repo-adrs: {repo_dir} does not exist")
            else:
                arch_id = f"{PROJECT}:{ARCHITECTURE_NAME}"
                repo_report = sync_repo_adrs(writer, repo_dir, arch_id)
                summary["repo_adrs"] = repo_report.synced
                print("repo-adrs " + repo_report.summary_line())
                for outcome in repo_report.outcomes:
                    print(f"  {outcome.state}: {outcome.adr_id}: {outcome.detail}")
                exit_code = max(exit_code, _adr_exit_code(repo_report))
        primary_report = wiki_report or repo_report
        if args.report_json and primary_report is not None:
            payload = primary_report.to_dict()
            if wiki_report and repo_report:
                payload["repo_adrs"] = repo_report.to_dict()
            Path(args.report_json).write_text(
                json.dumps(payload, indent=2) + "\n", encoding="utf-8"
            )
        print(f"✅ Architecture nodes synced to knowledge graph: {summary}")
    finally:
        writer.close()
    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())