scripts.check_adr_graph_sync

Reference › scripts › check_adr_graph_sync

The ADR graph-sync gate: a read-only, three-way drift detector. It compares the per-file ADRs under docs/adr/, the wiki's decisions.md register and the ArchitectureDecision nodes in the live knowledge graph, and never writes to any of them. A human or a gate runs it; no test imports it. It loads the environment contract through env_bootstrap and reads the graph through the interface in core.graph.interfaces.

Invocation: main takes --adr-dir (default docs/adr), --wiki-path (default $WIKI_ROOT/projects/orchestrata-ultimate/decisions.md), --project and --json. A missing or unreadable ADR directory prints an INVALID line and returns the invalid exit code, which is distinct from a drift failure.

load_repo_adrs, load_wiki_adrs and load_graph_adrs each return ADRs keyed by identity, and the graph loader also returns rows it could not key. compute_adr_sync_report compares the three sources for each ADR and produces an AdrSyncRow per ADR, including title agreement (_titles_agree) and body drift. AdrSyncReport collects the rows, counts body drift, lists the violations and answers ok. The text report is printed by _print_report.

At a glance

Class diagram

Dependency diagram

API

Classes

AdrSyncRow (dataclass)

Decorators: dataclass(frozen=True) · Line 202

The three-way state for one (register, padded_id) ADR identity.

AdrSyncReport (dataclass)

Decorators: dataclass(frozen=True) · Line 238

Functions

Source

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

#!/usr/bin/env python3
"""ADR graph-sync gate — a READ-ONLY three-way drift detector.

WHY THIS EXISTS
----------------
This project has **two independent ADR registers that number their entries
the same way**:

* ``docs/adr/NNNN-kebab-title.md`` — one file per decision, committed to
  git, first landing 2026-07-23 (``docs/adr/README.md``'s own "Why this log
  exists" note: three review rounds' worth of decisions that had been
  folded into ``spec.md`` prose and lost).
* ``knowledge-wiki/projects/orchestrata-ultimate/decisions.md`` — a single
  wiki page, ``## ADR-NNN: Title`` sections, deployed into the live
  ``orchkg`` graph as ``ArchitectureDecision`` nodes on 2026-07-19 (all
  ``Accepted``, ruling-1 immutability guard ACTIVE) via
  ``scripts/sync_architecture_to_graph.py --adrs``. ``plans/
  master-execution-plan.md:13`` names THIS page the "Ratified-decision
  registry."

Both registers write into the SAME id space
(``{project}:ADR-{number:04d}``, ``architecture_writer.upsert_architecture_
decision``), and ``docs/adr/README.md`` never mentions the wiki register at
all. An audit (2026-08-19) found the two corpora disagree on the DECISION
ITSELF for all 25 overlapping ids (ADR-0001..ADR-0025) — e.g. graph/wiki
``ADR-0001`` = "Memgraph as Semantic Graph Backend", repo ``ADR-0001`` =
"Enforcement boundary is committed worktree state, not tool interception".
The live ``ArchitectureDecision`` nodes carry no property that names which
register produced them (verified: ``keys(n)`` is exactly ``id/status/
project/body_sha/context/decision/alternatives/consequences/adr_number/
date/deciders/title`` — nothing else).

Given that, a script that "syncs docs/adr into the graph" cannot safely run
unattended for ids 1-25: it would either hit ``AdrImmutabilityError`` (the
existing node is ``Accepted`` with a different body) or, if the guard were
ever bypassed, silently overwrite a ratified wiki decision with an
unrelated repo one. **This gate does not resolve that.** It reports the
three-way state — present in the repo corpus / the wiki corpus / the graph,
and whether the titles agree where more than one source claims an id — and
FAILS loudly on any ambiguity or gap. Resolving the id collision (a
namespaced id, a ``register`` property, retiring one corpus, ...) is a
decision for a human, not this script.

WHAT IT CHECKS, per ADR number (union of every id seen in any of the three
sources):

* ``in_repo`` / ``in_wiki`` / ``in_graph`` — presence.
* If both ``in_repo`` and ``in_wiki`` and their titles disagree: an
  **AMBIGUOUS_COLLISION** — the same id names two different decisions, and
  the gate does not attempt to guess which one (if either) the graph
  should represent. VIOLATION regardless of what the graph currently holds.
* If ``in_repo`` (or ``in_wiki``) and NOT ``in_graph``: **MISSING_FROM_
  GRAPH** — a corpus decision the graph has never seen. VIOLATION.
* If ``in_repo`` XOR-only (no wiki collision) and ``in_graph`` but the
  titles differ: **DRIFT** — the graph disagrees with the one register
  that claims this id. VIOLATION.
* If ``in_graph`` only (neither corpus claims the id): **ORPHAN_IN_GRAPH**
  — reported, not a violation (this gate's job is "did the corpus fall
  behind the graph", not graph cleanup; no deletions, ever).

Read-only: never calls any ``ArchitectureWriter`` write method, never issues
a Cypher write. Exit codes match the project convention: 0 pass / 1
violations found / 2 could not determine the answer (unreadable corpus,
graph unreachable) — an inconclusive run is never reported as a pass.

Usage::

    python scripts/check_adr_graph_sync.py                # human-readable
    python scripts/check_adr_graph_sync.py --json          # machine-readable
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any

_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

import os  # noqa: E402

from orchestrata_ultimate.core.env_bootstrap import load_env_contract  # noqa: E402
from orchestrata_ultimate.core.graph.interfaces import (  # noqa: E402
    SyncGraphRepository,
)
from orchestrata_ultimate.core.graph.repository import (  # noqa: E402
    create_graph_repository,
)
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
    _adr_body_sha,
)

REPO_ROOT = Path(__file__).resolve().parent.parent

EXIT_PASS = 0
EXIT_VIOLATIONS = 1
EXIT_INVALID = 2

DEFAULT_PROJECT = "orchestrata-ultimate"
DEFAULT_ADR_DIR = REPO_ROOT / "docs" / "adr"


def _default_wiki_path() -> Path:
    """Resolve the knowledge-wiki ``decisions.md`` path at CALL TIME (never
    at import time, per project doctrine), mirroring ``sync_architecture_
    to_graph._default_adrs_path``."""
    root = os.environ.get("WIKI_ROOT", str(Path.home() / "workspace/knowledge-wiki"))
    return Path(root) / "projects" / DEFAULT_PROJECT / "decisions.md"


def load_repo_adrs(adr_dir: Path) -> dict[tuple[str, str], ParsedAdr]:
    """Parse every ``docs/adr/NNNN-*.md`` file (``README.md`` excluded).

    Keyed by ``(register, padded_id)`` (ADR-0047) rather than the bare
    integer -- ``parse_adr_file`` always reports ``register="factory"``, so
    every key here has that register, but the tuple form keeps this dict
    directly comparable/unionable with ``load_wiki_adrs``' (which spans both
    registers) without an int collapsing two different ids together.

    Raises:
        ValueError: a file's heading is malformed (propagated from
            ``parse_adr_file`` — an unparseable corpus is not a "0 found"
            pass, it is a reason to return EXIT_INVALID).
    """
    adrs: dict[tuple[str, str], ParsedAdr] = {}
    for path in sorted(adr_dir.glob("*.md")):
        if path.name.lower() == "readme.md":
            continue
        parsed = parse_adr_file(path.read_text(encoding="utf-8"))
        adrs[(parsed.register, parsed.padded_id)] = parsed
    return adrs


def load_wiki_adrs(wiki_path: Path) -> dict[tuple[str, str], ParsedAdr]:
    """Parse the wiki's ``decisions.md`` ADR register.

    Keyed by ``(register, padded_id)`` (ADR-0047): the wiki page carries
    BOTH registers (1-3 digit project-register headings and 4-digit
    factory-register mirrors) under the same numbering, e.g. ``## ADR-001:``
    and ``## ADR-0001:`` are different decisions that happen to share the
    integer 1 -- an int-keyed dict would silently let the later heading win.

    Returns an empty dict (never raises) if the file is absent — a missing
    wiki checkout is infra a session-start hook machine may lack, matching
    ``sync_architecture_to_graph.sync_adrs``'s existing "loud SKIP" policy.
    Callers must check ``wiki_path.exists()`` themselves to tell "empty
    corpus" apart from "corpus unavailable".
    """
    if not wiki_path.exists():
        return {}
    return {
        (adr.register, adr.padded_id): adr
        for adr in parse_adrs(wiki_path.read_text(encoding="utf-8"))
    }


def load_graph_adrs(
    repo: SyncGraphRepository, project: str
) -> tuple[dict[tuple[str, str], dict[str, Any]], list[dict[str, Any]]]:
    """Read every live ``ArchitectureDecision`` node for ``project``.

    ADR-0047: a node written by the current writer carries ``register`` and
    ``padded_id`` properties and is returned keyed by ``(register,
    padded_id)`` in the first dict. A node written BEFORE ADR-0047 (or by
    any out-of-band Cypher) has neither property and is reported in the
    second list instead -- never silently folded into the keyed dict under
    a guessed identity, since there is no way to know which register it
    belongs to. Callers surface that list as ``LEGACY_UNKEYED`` violations.

    Read-only: a single ``MATCH``/``RETURN``, no write query.
    """
    rows = repo.execute_read(
        "MATCH (n:ArchitectureDecision {project: $project}) "
        "RETURN n.id AS id, n.adr_number AS number, n.title AS title, "
        "n.status AS status, n.register AS register, n.padded_id AS padded_id, "
        "n.body_sha AS body_sha",
        {"project": project},
    )
    keyed: dict[tuple[str, str], dict[str, Any]] = {}
    legacy: list[dict[str, Any]] = []
    for row in rows:
        register = row.get("register")
        padded_id = row.get("padded_id")
        if not register or not padded_id:
            legacy.append(row)
            continue
        keyed[(register, padded_id)] = row
    return keyed, legacy


@dataclass(frozen=True)
class AdrSyncRow:
    """The three-way state for one ``(register, padded_id)`` ADR identity.

    ADR-0047: ``register``/``padded_id`` are both ``None`` for a
    ``LEGACY_UNKEYED`` row -- a live graph node with no register/padded_id
    property, so it has no ``(register, padded_id)`` identity to key by at
    all. Its node id/title are named in ``detail`` instead.
    """

    register: str | None
    padded_id: str | None
    in_repo: bool
    in_wiki: bool
    in_graph: bool
    repo_title: str | None
    wiki_title: str | None
    graph_title: str | None
    state: str
    detail: str

    def to_dict(self) -> dict[str, Any]:
        return {
            "register": self.register,
            "padded_id": self.padded_id,
            "in_repo": self.in_repo,
            "in_wiki": self.in_wiki,
            "in_graph": self.in_graph,
            "repo_title": self.repo_title,
            "wiki_title": self.wiki_title,
            "graph_title": self.graph_title,
            "state": self.state,
            "detail": self.detail,
        }


@dataclass(frozen=True)
class AdrSyncReport:
    rows: list[AdrSyncRow]
    wiki_available: bool
    repo_count: int
    wiki_count: int
    graph_count: int
    body_compared: int = 0

    @property
    def body_drift_count(self) -> int:
        return sum(1 for r in self.rows if r.state == _STATE_BODY_DRIFT)

    @property
    def violations(self) -> list[AdrSyncRow]:
        return [r for r in self.rows if r.state in _VIOLATION_STATES]

    @property
    def ok(self) -> bool:
        return not self.violations

    def to_dict(self) -> dict[str, Any]:
        return {
            "wiki_available": self.wiki_available,
            "repo_count": self.repo_count,
            "wiki_count": self.wiki_count,
            "graph_count": self.graph_count,
            "body_compared": self.body_compared,
            "body_drift_count": self.body_drift_count,
            "rows": [r.to_dict() for r in self.rows],
            "violation_count": len(self.violations),
        }


_STATE_AMBIGUOUS = "AMBIGUOUS_COLLISION"
_STATE_MISSING = "MISSING_FROM_GRAPH"
_STATE_DRIFT = "DRIFT"
_STATE_ORPHAN = "ORPHAN_IN_GRAPH"
_STATE_LEGACY_UNKEYED = "LEGACY_UNKEYED"
_STATE_BODY_DRIFT = "BODY_DRIFT"
_STATE_OK = "OK"
_VIOLATION_STATES = frozenset(
    {
        _STATE_AMBIGUOUS,
        _STATE_MISSING,
        _STATE_DRIFT,
        _STATE_LEGACY_UNKEYED,
        _STATE_BODY_DRIFT,
    }
)


def _titles_agree(a: str | None, b: str | None) -> bool:
    if a is None or b is None:
        return False
    return a.strip().lower() == b.strip().lower()


def compute_adr_sync_report(
    *,
    adr_dir: Path,
    wiki_path: Path,
    graph_repo: SyncGraphRepository,
    project: str = DEFAULT_PROJECT,
) -> AdrSyncReport:
    """Build the full three-way report. Pure read: no write anywhere."""
    repo_adrs = load_repo_adrs(adr_dir)
    wiki_available = wiki_path.exists()
    wiki_adrs = load_wiki_adrs(wiki_path)
    graph_adrs, legacy_nodes = load_graph_adrs(graph_repo, project)

    all_keys = sorted(set(repo_adrs) | set(wiki_adrs) | set(graph_adrs))
    rows: list[AdrSyncRow] = []
    body_compared = 0
    for key in all_keys:
        register, padded_id = key
        repo_entry = repo_adrs.get(key)
        wiki_entry = wiki_adrs.get(key)
        graph_entry = graph_adrs.get(key)
        repo_title = repo_entry.title if repo_entry else None
        wiki_title = wiki_entry.title if wiki_entry else None
        graph_title = graph_entry["title"] if graph_entry else None

        if repo_entry and wiki_entry and not _titles_agree(repo_title, wiki_title):
            if graph_entry is None:
                detail = (
                    f"repo title {repo_title!r} and wiki title {wiki_title!r} "
                    "name DIFFERENT decisions under the same id, and neither "
                    "is in the graph yet."
                )
            elif _titles_agree(graph_title, wiki_title):
                detail = (
                    f"repo title {repo_title!r} and wiki title {wiki_title!r} "
                    "name DIFFERENT decisions under the same id; the graph "
                    f"currently holds the WIKI one ({graph_title!r})."
                )
            elif _titles_agree(graph_title, repo_title):
                detail = (
                    f"repo title {repo_title!r} and wiki title {wiki_title!r} "
                    "name DIFFERENT decisions under the same id; the graph "
                    f"currently holds the REPO one ({graph_title!r})."
                )
            else:
                detail = (
                    f"repo title {repo_title!r} and wiki title {wiki_title!r} "
                    f"name DIFFERENT decisions under the same id; the graph "
                    f"title ({graph_title!r}) matches NEITHER."
                )
            rows.append(
                AdrSyncRow(
                    register=register,
                    padded_id=padded_id,
                    in_repo=True,
                    in_wiki=True,
                    in_graph=graph_entry is not None,
                    repo_title=repo_title,
                    wiki_title=wiki_title,
                    graph_title=graph_title,
                    state=_STATE_AMBIGUOUS,
                    detail=detail,
                )
            )
            continue

        # No repo/wiki collision on this id: at most one of them claims it
        # (or both, in agreement) -- compare the graph against whichever
        # corpus source is present, preferring the repo (this gate's named
        # subject) when both are present and agree.
        source_title = repo_title if repo_entry else wiki_title
        source_name = "repo" if repo_entry else ("wiki" if wiki_entry else None)

        if source_name is None:
            rows.append(
                AdrSyncRow(
                    register=register,
                    padded_id=padded_id,
                    in_repo=False,
                    in_wiki=False,
                    in_graph=True,
                    repo_title=None,
                    wiki_title=None,
                    graph_title=graph_title,
                    state=_STATE_ORPHAN,
                    detail=(
                        f"graph node {graph_title!r} has no matching entry in "
                        "either corpus."
                    ),
                )
            )
            continue

        if graph_entry is None:
            rows.append(
                AdrSyncRow(
                    register=register,
                    padded_id=padded_id,
                    in_repo=bool(repo_entry),
                    in_wiki=bool(wiki_entry),
                    in_graph=False,
                    repo_title=repo_title,
                    wiki_title=wiki_title,
                    graph_title=None,
                    state=_STATE_MISSING,
                    detail=(
                        f"{source_name} title {source_title!r} has no "
                        "ArchitectureDecision node in the graph."
                    ),
                )
            )
            continue

        if not _titles_agree(source_title, graph_title):
            rows.append(
                AdrSyncRow(
                    register=register,
                    padded_id=padded_id,
                    in_repo=bool(repo_entry),
                    in_wiki=bool(wiki_entry),
                    in_graph=True,
                    repo_title=repo_title,
                    wiki_title=wiki_title,
                    graph_title=graph_title,
                    state=_STATE_DRIFT,
                    detail=(
                        f"{source_name} title {source_title!r} disagrees with "
                        f"the graph's title {graph_title!r} for the same id."
                    ),
                )
            )
            continue

        # Body comparison (ADR-0096): for a factory id the docs record is the
        # body's owner, so its body_sha must equal the graph's. A wiki entry
        # is never compared -- it is a pointer to the docs record.
        state, detail = _STATE_OK, "in sync."
        if repo_entry is not None and register == "factory":
            body_compared += 1
            docs_sha = _adr_body_sha(
                repo_entry.context,
                repo_entry.decision,
                repo_entry.alternatives,
                repo_entry.consequences,
            )
            graph_sha = graph_entry.get("body_sha")
            if graph_sha != docs_sha:
                state = _STATE_BODY_DRIFT
                detail = (
                    f"docs body_sha {docs_sha[:10]} != graph body_sha "
                    f"{str(graph_sha)[:10]}: the graph holds a different body "
                    "than docs/adr (re-baseline under a ruling, or sync)."
                )
        rows.append(
            AdrSyncRow(
                register=register,
                padded_id=padded_id,
                in_repo=bool(repo_entry),
                in_wiki=bool(wiki_entry),
                in_graph=True,
                repo_title=repo_title,
                wiki_title=wiki_title,
                graph_title=graph_title,
                state=state,
                detail=detail,
            )
        )

    for node in legacy_nodes:
        node_id = node.get("id")
        node_title = node.get("title")
        rows.append(
            AdrSyncRow(
                register=None,
                padded_id=None,
                in_repo=False,
                in_wiki=False,
                in_graph=True,
                repo_title=None,
                wiki_title=None,
                graph_title=node_title,
                state=_STATE_LEGACY_UNKEYED,
                detail=(
                    f"graph node id={node_id!r} title={node_title!r} has no "
                    "register/padded_id property -- a pre-ADR-0047 write (or "
                    "an out-of-band Cypher write). Not matched against any "
                    "corpus id; re-key it explicitly rather than guessing "
                    "which register it belongs to."
                ),
            )
        )

    return AdrSyncReport(
        rows=rows,
        wiki_available=wiki_available,
        repo_count=len(repo_adrs),
        wiki_count=len(wiki_adrs),
        graph_count=len(graph_adrs) + len(legacy_nodes),
        body_compared=body_compared,
    )


def _row_label(row: AdrSyncRow) -> str:
    """``ADR-<padded_id>``, prefixed by the register when it is ``project``
    (ADR-0047) -- a factory-register row prints bare (it is the majority/
    default case); an unkeyed legacy row has no id to print at all."""
    if row.register is None or row.padded_id is None:
        return "(unkeyed)"
    prefix = "" if row.register == "factory" else f"{row.register} "
    return f"{prefix}ADR-{row.padded_id}"


def _print_report(report: AdrSyncReport) -> None:
    print("ADR Graph-Sync Report (read-only, three-way: repo / wiki / graph)")
    print("=" * 72)
    print(
        f"repo corpus  : {report.repo_count} ADR(s) under docs/adr/\n"
        f"wiki corpus  : {report.wiki_count} ADR(s)"
        + ("" if report.wiki_available else " (decisions.md NOT FOUND — degraded)")
        + f"\ngraph        : {report.graph_count} ArchitectureDecision node(s)"
        + f"\nbody compare : {report.body_compared} docs id(s) compared by "
        f"body_sha, {report.body_drift_count} BODY_DRIFT"
    )
    print()
    for row in report.rows:
        marker = "OK  " if row.state == _STATE_OK else f"{row.state:<20}"
        print(f"  {_row_label(row):<18}{marker}  {row.detail}")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="check_adr_graph_sync",
        description=(
            "ADR graph-sync gate: read-only three-way drift detector across "
            "docs/adr/, the knowledge-wiki decisions.md register, and the "
            "live orchkg ArchitectureDecision nodes. Never writes."
        ),
    )
    parser.add_argument(
        "--adr-dir",
        default=str(DEFAULT_ADR_DIR),
        help="Directory of per-file ADRs (default: docs/adr).",
    )
    parser.add_argument(
        "--wiki-path",
        default=None,
        help="Path to the wiki decisions.md (default: $WIKI_ROOT/projects/"
        "orchestrata-ultimate/decisions.md).",
    )
    parser.add_argument("--project", default=DEFAULT_PROJECT)
    parser.add_argument(
        "--json", action="store_true", help="Emit a machine-readable JSON report."
    )
    args = parser.parse_args(argv)

    load_env_contract()

    adr_dir = Path(args.adr_dir)
    if not adr_dir.is_dir():
        print(
            f"[adr-graph-sync] INVALID — ADR directory not found: {adr_dir}",
            file=sys.stderr,
        )
        return EXIT_INVALID

    try:
        repo_adrs = load_repo_adrs(adr_dir)
    except ValueError as exc:
        print(
            f"[adr-graph-sync] INVALID — {adr_dir} unreadable: {exc}", file=sys.stderr
        )
        return EXIT_INVALID

    if not repo_adrs:
        print(
            f"[adr-graph-sync] INVALID — {adr_dir} contains no parseable ADR "
            "files; a PASS here would report 'nothing wrong' when the truth "
            "is 'nothing looked at'",
            file=sys.stderr,
        )
        return EXIT_INVALID

    wiki_path = Path(args.wiki_path) if args.wiki_path else _default_wiki_path()

    try:
        graph_repo = create_graph_repository()
        graph_repo.verify_connectivity()
    except Exception as exc:  # noqa: BLE001 -- any connectivity failure is INVALID
        print(
            f"[adr-graph-sync] INVALID — could not reach the graph: {exc}",
            file=sys.stderr,
        )
        return EXIT_INVALID

    try:
        report = compute_adr_sync_report(
            adr_dir=adr_dir,
            wiki_path=wiki_path,
            graph_repo=graph_repo,
            project=args.project,
        )
    finally:
        graph_repo.close()

    if args.json:
        print(json.dumps(report.to_dict(), indent=2))
    else:
        _print_report(report)
        if not report.wiki_available:
            print(
                f"\nNOTE: wiki register not found at {wiki_path} — collision "
                "detection against the wiki corpus is DISABLED for this run; "
                "only repo<->graph agreement was checked.",
                file=sys.stderr,
            )
        if report.violations:
            print(
                f"\n[adr-graph-sync] FAIL — {len(report.violations)} "
                f"violation(s) across {len(report.rows)} id(s) examined. "
                "This gate never writes or deletes anything; resolving an "
                "AMBIGUOUS_COLLISION or MISSING_FROM_GRAPH row is a human "
                "decision (id-collision reconciliation or an explicit sync "
                "run), not something this script will do for you.",
                file=sys.stderr,
            )
        else:
            print(
                f"\n[adr-graph-sync] OK — {len(report.rows)} id(s) examined, "
                "no drift or collisions detected."
            )

    return EXIT_PASS if report.ok else EXIT_VIOLATIONS


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