scripts.bootstrap_graphs

Reference › scripts › bootstrap_graphs

An idempotent bootstrap for both graphs, the knowledge graph and Memgraph. It derives graph state from source, never by hand. A human or a provisioning step runs it; no module or test imports it.

Invocation: main requires at least one of --kg, --memgraph, --all, --check or --plan, and exits 2 otherwise. --neo4j is a deprecated alias of --kg and prints a warning. --plan prints the steps without running anything (_print_plan), and --check verifies the current state (_check_kg, _check_memgraph) and reports what is missing. _resolve_targets decides which graphs are in play.

Each graph has an ordered chain of Step records (_kg_steps, _memgraph_steps), and _run_steps executes them, aborting when a required step fails and skipping optional ones. The environment is the process environment merged with a dotenv file (_effective_env). The step lists can be passed as keyword arguments to main so tests can exercise the dispatch without touching shared infrastructure.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

Step (dataclass)

Decorators: dataclass(frozen=True) · Line 84

One orchestrated building-block invocation.

Functions

Source

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

#!/usr/bin/env python3
"""Idempotent bootstrap for both graphs — derive graph state from source, never hand-build.

Orchestrating CLI for plans/reproducibility-plan.md Phase R4. It does NOT
reimplement any graph-writing logic itself — it composes the EXISTING
building-block scripts as subprocess steps, so both graphs can be rebuilt
from source + seeds through ONE entry point:

KG (operational knowledge graph) chain:
    1. ``kg_schema.py apply``               — canonical constraints/indexes (SPEC-106).
    2. ``serialize_codebase_ast.py serialize`` — regenerate the AST JSON, ONLY if
       ``plans/ast-ultimate.json`` is missing (best-effort; a failure here just
       skips the code-intelligence step below with a warning, it never aborts).
    3. ``sync_ast_to_graph.py schema`` + ``... full --ast <path>`` — code
       intelligence (CodeFunction/CodeClass/Method); skipped with a logged
       warning if no AST JSON is available.
    4. ``sync_architecture_to_graph.py --repo-adrs docs/adr`` — architecture/
       component/rule nodes plus one ``ArchitectureDecision`` per
       ``docs/adr/*.md`` (D5).
    5. ``enrich_semantic_graph_specs.py`` — specs/RTM (spec side).

Memgraph (semantic/RAG) chain:
    1. ``ingest_orchestrata_ultimate.py`` — source -> chunk/embed -> Memgraph.
       Respects shared-cluster discipline (label-scoped, project_id-tagged;
       never touches ``:Entity`` nodes this project didn't create).

Modes
-----
``--kg`` / ``--memgraph`` / ``--all``
    Scope which graph(s) to target. Combine with ``--check`` or ``--plan``.
    With neither ``--check`` nor ``--plan``, actually executes the chain(s)
    (mutates the live graph(s)) — each step is logged before it runs, and any
    non-optional step failure aborts with a clear message and non-zero exit.
    ``--neo4j`` is accepted as a deprecated alias for ``--kg`` (the operational
    KG no longer runs on Neo4j at all — same orchgraph/Memgraph cluster as the
    semantic graph, different database — the flag is graph-role-named now).
``--check``
    READ-ONLY. Reports drift instead of mutating: ``kg_schema.py check``
    (live KG schema vs. ``canonical_schema.py``) and/or a light Memgraph
    reachability + project-scoped-node check. Exits 0 iff no drift was found.
    With no scope flag, checks BOTH graphs (same as ``--all --check``).
``--plan``
    READ-ONLY. Prints the ordered step list (name + exact argv) that the
    matching ``--kg``/``--memgraph``/``--all`` run would execute, without
    running or connecting to anything. Also the runbook: paste the printed
    commands to run steps by hand.

Examples
--------
    python scripts/bootstrap_graphs.py --check
    python scripts/bootstrap_graphs.py --all --check
    python scripts/bootstrap_graphs.py --all --plan
    python scripts/bootstrap_graphs.py --kg
    python scripts/bootstrap_graphs.py --memgraph
"""

from __future__ import annotations

import argparse
import logging
import os
import subprocess
import sys
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path

from dotenv import dotenv_values
from neo4j import GraphDatabase

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

REPO_ROOT = Path(__file__).resolve().parent.parent
SCRIPTS_DIR = REPO_ROOT / "scripts"
AST_JSON_PATH = REPO_ROOT / "plans" / "ast-ultimate.json"
PROJECT_ID = "orchestrata-ultimate"


@dataclass(frozen=True)
class Step:
    """One orchestrated building-block invocation."""

    name: str
    argv: list[str]
    optional: bool = False
    """If True, a non-zero exit logs a warning and the run continues."""
    requires: Path | None = field(default=None)
    """If set and missing at run time, the step is skipped with a warning
    (not counted as a failure) instead of being executed."""


def _effective_env(
    env: Mapping[str, str] | None,
    dotenv_path: Path | None = None,
) -> dict[str, str]:
    """Build the env mapping passed to every subprocess step.

    Starts from ``env`` (the injectable seam — defaults to a copy of the
    real ``os.environ``) and fills in any keys missing from it using the
    project ``.env`` file, mirroring the ``os.environ.get(key, default)``
    convention: explicit/injected values always win over the file default.
    Never mutates the real ``os.environ``.

    Args:
        env: Base environment mapping. Defaults to a copy of ``os.environ``.
        dotenv_path: Optional path to a ``.env`` file. Defaults to the
            project root ``.env``.
    """
    base: dict[str, str] = dict(env) if env is not None else dict(os.environ)
    dotenv_path = dotenv_path if dotenv_path is not None else REPO_ROOT / ".env"
    if dotenv_path.exists():
        for key, value in dotenv_values(dotenv_path).items():
            if value is not None and key not in base:
                base[key] = value
    return base


def _kg_steps() -> list[Step]:
    """Build the ordered operational-KG bootstrap chain (SPEC-106 building blocks)."""
    py = sys.executable
    steps: list[Step] = [
        Step(
            "kg_schema.apply",
            [py, str(SCRIPTS_DIR / "kg_schema.py"), "apply"],
        )
    ]
    if not AST_JSON_PATH.exists():
        steps.append(
            Step(
                "serialize_codebase_ast.generate",
                [
                    py,
                    str(SCRIPTS_DIR / "serialize_codebase_ast.py"),
                    "serialize",
                    "--source",
                    "src/orchestrata_ultimate",
                    "--name",
                    PROJECT_ID,
                    "--output",
                    str(AST_JSON_PATH),
                ],
                optional=True,
            )
        )
    steps.append(
        Step(
            "sync_ast_to_graph.schema",
            [py, str(SCRIPTS_DIR / "sync_ast_to_graph.py"), "schema"],
            requires=AST_JSON_PATH,
        )
    )
    steps.append(
        Step(
            "sync_ast_to_graph.full",
            [
                py,
                str(SCRIPTS_DIR / "sync_ast_to_graph.py"),
                "full",
                "--ast",
                str(AST_JSON_PATH),
                "--project",
                PROJECT_ID,
            ],
            requires=AST_JSON_PATH,
        )
    )
    steps.append(
        Step(
            "sync_architecture_to_graph",
            [
                py,
                str(SCRIPTS_DIR / "sync_architecture_to_graph.py"),
                "--repo-adrs",
                str(REPO_ROOT / "docs" / "adr"),
            ],
        )
    )
    steps.append(
        Step(
            "enrich_semantic_graph_specs",
            [py, str(SCRIPTS_DIR / "enrich_semantic_graph_specs.py")],
        )
    )
    return steps


def _memgraph_steps() -> list[Step]:
    """Build the ordered Memgraph bootstrap chain."""
    py = sys.executable
    return [
        Step(
            "ingest_orchestrata_ultimate",
            [py, str(SCRIPTS_DIR / "ingest_orchestrata_ultimate.py")],
        )
    ]


def _run_steps(steps: list[Step], env: Mapping[str, str]) -> int:
    """Execute ``steps`` in order against ``env``; abort on the first hard failure."""
    for step in steps:
        if step.requires is not None and not step.requires.exists():
            logger.warning(
                "Skipping step %s — required input not found: %s",
                step.name,
                step.requires,
            )
            continue
        logger.info("Running step: %s", step.name)
        logger.info("  $ %s", " ".join(step.argv))
        result = subprocess.run(step.argv, cwd=REPO_ROOT, env=dict(env))
        if result.returncode != 0:
            if step.optional:
                logger.warning(
                    "Optional step %s failed (exit %s) — continuing",
                    step.name,
                    result.returncode,
                )
                continue
            logger.error(
                "Step %s failed (exit %s) — aborting bootstrap",
                step.name,
                result.returncode,
            )
            return result.returncode
    return 0


def _print_plan(steps: list[Step]) -> None:
    """Print the ordered step list without executing anything (the runbook)."""
    print(f"Planned {len(steps)} step(s):")
    for idx, step in enumerate(steps, start=1):
        tags = []
        if step.optional:
            tags.append("optional")
        if step.requires is not None:
            tags.append(f"requires={step.requires}")
        suffix = f"  [{', '.join(tags)}]" if tags else ""
        print(f"  {idx}. {step.name}{suffix}")
        print(f"       $ {' '.join(step.argv)}")


def _check_kg(env: Mapping[str, str]) -> tuple[bool, list[str]]:
    """Report operational-KG schema drift via ``kg_schema.py check`` (read-only)."""
    py = sys.executable
    result = subprocess.run(
        [py, str(SCRIPTS_DIR / "kg_schema.py"), "check"],
        cwd=REPO_ROOT,
        env=dict(env),
        capture_output=True,
        text=True,
    )
    lines = [
        line for line in (result.stdout + result.stderr).splitlines() if line.strip()
    ]
    if not lines:
        lines = [f"kg_schema.py check exited {result.returncode} with no output."]
    return result.returncode == 0, lines


def _check_memgraph(env: Mapping[str, str]) -> tuple[bool, list[str]]:
    """Light Memgraph reachability + project-scoped-node check (read-only).

    Never queries by label (the cluster is shared with knowledge-wiki) —
    only by the ``project_id`` property every node this project writes
    carries (see ``semantic_graph/ingestion.py``), so this can never touch
    or even see another project's ``:Entity``/``:Chunk`` nodes.
    """
    uri = env.get("MEMGRAPH_URI")
    if not uri:
        raise RuntimeError(
            "No graph URI configured — set MEMGRAPH_URI (see .env); this "
            "code has no default address (RUL-003). It used to fall back to "
            "bolt://localhost:7687, which nothing in this infrastructure "
            "listens on."
        )
    user = env.get("MEMGRAPH_USERNAME") or None
    password = env.get("MEMGRAPH_PASSWORD", "")
    auth = (user, password) if user else None
    driver = GraphDatabase.driver(uri, auth=auth)
    try:
        driver.verify_connectivity()
    except Exception as exc:  # noqa: BLE001
        driver.close()
        return False, [f"Memgraph unreachable at {uri}: {exc}"]
    try:
        with driver.session() as session:
            record = session.run(
                "MATCH (n) WHERE n.project_id = $pid RETURN count(n) AS c",
                {"pid": PROJECT_ID},
            ).single()
        count = record["c"] if record else 0
    finally:
        driver.close()
    if count == 0:
        return False, [
            f"No {PROJECT_ID}-tagged nodes found in Memgraph ({uri}) — "
            "semantic graph has not been ingested from source."
        ]
    return True, [
        f"OK — {count} {PROJECT_ID}-tagged node(s) found in Memgraph ({uri})."
    ]


def _resolve_targets(args: argparse.Namespace) -> tuple[bool, bool]:
    """Resolve which graph(s) a flag combination targets.

    ``--all`` always means both. A bare scope flag (``--kg``/``--memgraph``)
    scopes to just that graph. With no scope flag at all (bare ``--check`` or
    bare ``--plan``), both graphs are targeted.
    """
    if args.all:
        return True, True
    if args.kg or args.memgraph:
        return args.kg, args.memgraph
    return True, True


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="bootstrap_graphs",
        description=(
            "Idempotent bootstrap for both graphs — derive graph state from "
            "source, never hand-build (reproducibility-plan.md Phase R4)."
        ),
    )
    parser.add_argument(
        "--kg", action="store_true", help="Target the operational knowledge graph (KG)."
    )
    parser.add_argument(
        "--neo4j",
        action="store_true",
        dest="neo4j_legacy",
        help=argparse.SUPPRESS,  # deprecated alias for --kg (graph-rename wave)
    )
    parser.add_argument(
        "--memgraph",
        action="store_true",
        help="Target the Memgraph semantic/RAG graph.",
    )
    parser.add_argument("--all", action="store_true", help="Target both graphs.")
    parser.add_argument(
        "--check",
        action="store_true",
        help="Read-only: report schema/ingest drift instead of mutating.",
    )
    parser.add_argument(
        "--plan",
        action="store_true",
        help="Read-only: print the ordered step list without executing.",
    )
    return parser


def main(
    argv: list[str] | None = None,
    env: Mapping[str, str] | None = None,
    *,
    kg_steps: list[Step] | None = None,
    memgraph_steps: list[Step] | None = None,
) -> int:
    """CLI entry point.

    ``kg_steps``/``memgraph_steps`` default to the real building-block
    chains (:func:`_kg_steps`/:func:`_memgraph_steps`) — every real CLI
    invocation runs them. The keyword-only overrides exist purely so tests
    can exercise the full ``--plan``/``--check``/execute dispatch (including
    the abort-on-failure and optional-skip control flow in
    :func:`_run_steps`) end-to-end without ever running the real chains
    against shared live infra — mirroring the same pattern used by
    ``kg_schema.py``'s ``_cmd_apply(..., *, statements=None)``.
    """
    parser = _build_parser()
    args = parser.parse_args(argv)

    if args.neo4j_legacy:
        print(
            "WARNING: --neo4j is deprecated, use --kg instead.",
            file=sys.stderr,
        )
        args.kg = True

    if not (args.kg or args.memgraph or args.all or args.check or args.plan):
        parser.print_usage(sys.stderr)
        print(
            "ERROR: specify at least one of --kg/--memgraph/--all/--check/--plan",
            file=sys.stderr,
        )
        return 2

    do_kg, do_memgraph = _resolve_targets(args)
    effective_env = _effective_env(env)
    resolved_kg_steps = kg_steps if kg_steps is not None else _kg_steps()
    resolved_memgraph_steps = (
        memgraph_steps if memgraph_steps is not None else _memgraph_steps()
    )

    if args.plan:
        steps: list[Step] = []
        if do_kg:
            steps.extend(resolved_kg_steps)
        if do_memgraph:
            steps.extend(resolved_memgraph_steps)
        _print_plan(steps)
        return 0

    if args.check:
        ok = True
        if do_kg:
            print("=== KG schema check (kg_schema.py check) ===")
            kg_ok, lines = _check_kg(effective_env)
            for line in lines:
                print(line)
            ok = ok and kg_ok
        if do_memgraph:
            print("=== Memgraph reachability/label check ===")
            mg_ok, lines = _check_memgraph(effective_env)
            for line in lines:
                print(line)
            ok = ok and mg_ok
        return 0 if ok else 1

    exit_code = 0
    if do_kg:
        logger.info("=== KG bootstrap chain ===")
        exit_code = _run_steps(resolved_kg_steps, effective_env)
        if exit_code != 0:
            return exit_code
    if do_memgraph:
        logger.info("=== Memgraph bootstrap ===")
        exit_code = _run_steps(resolved_memgraph_steps, effective_env)
    return exit_code


if __name__ == "__main__":
    sys.exit(main())