scripts.sync_ast_to_graph

Reference › scripts › sync_ast_to_graph

scripts/sync_ast_to_graph.py is the command-line wrapper for the AST sync engine: it loads an AST JSON file (produced by serialize_codebase_ast) into the knowledge graph as code nodes and relationships. A human runs it; for continuous sync see ast_live_sync. _make_syncer builds the GraphSyncer from the command-line flags.

Global flags are --uri (default KG_ROUTING_URI, then KG_URI), --user, --password (with no environment fallback, so omitting them means empty credentials), --database (default KG_DATABASE; needed for orchgraph, which serves orchkg and the per-worker orchkg_w<N> from one endpoint) and --dialect (neo4j or memgraph, default from KG_ENGINE; Memgraph has no IF NOT EXISTS forms and no full-text index, so the canonical statements are translated and the full-text ones dropped). Subcommands, dispatched by main to _cmd_* handlers: schema (create the schema), full (a full sync), map, incremental, stats and verify. An exception is printed as ERROR: and exits 1.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

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

#!/usr/bin/env python3
"""CLI wrapper for the Neo4j AST sync engine.

Reads a serialised AST JSON file produced by serialize_codebase_ast.py and
syncs the contents into a knowledge graph via GraphSyncer.

Subcommands
-----------
schema      Create constraints and indexes (idempotent, safe to re-run).
full        Full sync: clear existing project data and re-sync from AST JSON.
map         Create cross-project MAPS_TO edges from class-mapping.json.
incremental Re-sync specific modules (delete + recreate + repair cross edges).
stats       Print node/relationship counts per label/type.
verify      Run verification queries and report pass/fail.

Connection URI defaults to the ``KG_ROUTING_URI`` environment variable
(falling back to ``KG_URI``); there is no username/password env fallback —
pass ``--user``/``--password`` explicitly, or ``--no-auth`` for the live
orchgraph cluster, which negotiates no authentication at all.  CLI flags
override env vars.  Global connection flags (``--uri``/``--user``/
``--password``/``--database``/``--dialect``/``--no-auth``) go BEFORE the
subcommand.

Examples
--------
    python scripts/sync_ast_to_graph.py --uri "$KG_ROUTING_URI" \\
        --database "$KG_DATABASE" --no-auth schema

    python scripts/sync_ast_to_graph.py full \\
        --ast plans/ast-ultimate.json --project orchestrata-ultimate

    python scripts/sync_ast_to_graph.py map \\
        --mapping plans/class-mapping.json

    python scripts/sync_ast_to_graph.py incremental \\
        --ast plans/ast-ultimate.json --project orchestrata-ultimate \\
        --modules agents.base_agent,kernel.event_bus

    python scripts/sync_ast_to_graph.py stats

    python scripts/sync_ast_to_graph.py verify --project orchestrata-ultimate
"""

import argparse

# Allow running from the repo root without package installation.
# Use importlib to load sync.py directly, bypassing orchestrata_ultimate.__init__
# which has a broken import chain in core/traceability/__init__.py.
import importlib.util
import json
import sys
from pathlib import Path

_sync_path = (
    Path(__file__).resolve().parent.parent
    / "src"
    / "orchestrata_ultimate"
    / "knowledge_graph"
    / "sync.py"
)
_spec = importlib.util.spec_from_file_location(
    "orchestrata_ultimate.knowledge_graph.sync", _sync_path
)
if _spec is None or _spec.loader is None:
    raise ImportError(f"Cannot load sync module from {_sync_path}")
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
GraphSyncer = _mod.GraphSyncer


def _make_syncer(args: argparse.Namespace) -> GraphSyncer:
    """Instantiate GraphSyncer from CLI args (override env vars if supplied)."""
    # An explicitly-passed-but-EMPTY --uri must not silently fall back to the
    # env default. `--uri "$KG_ROUTING_URI"` with that key blank expands to
    # `--uri ""`, and the old `or None` turned that into "use the default
    # graph" -- so `--uri "" --database orchkg schema` aimed orchgraph's DDL
    # at the wrong server. Caught 2026-07-26 only because --no-auth made the
    # wrong target reject the connection; without it the write would have landed.
    uri_arg = getattr(args, "uri", None)
    if uri_arg is not None and not uri_arg.strip():
        raise SystemExit(
            "--uri was passed but is empty; omit it to use $KG_ROUTING_URI "
            "(falling back to $KG_URI), or pass a real address"
        )
    uri = uri_arg or None
    user = getattr(args, "user", None) or None
    password = getattr(args, "password", None) or None
    database = getattr(args, "database", None) or None
    no_auth = bool(getattr(args, "no_auth", False))
    # No literal default: an unset --dialect must fall through to
    # GraphSyncer -> resolve_kg_dialect(), which reads KG_ENGINE at CALL
    # time. Hardcoding "neo4j" here overrode that resolver and sent Neo4j
    # DDL ("CREATE TEXT INDEX ... IF NOT EXISTS") to an engine that had
    # been orchgraph/Memgraph for hours, failing `schema` outright.
    dialect = getattr(args, "dialect", None)
    return GraphSyncer(
        uri=uri,
        username=user,
        password=password,
        database=database,
        no_auth=no_auth,
        dialect=dialect,
    )


def _cmd_schema(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        syncer.create_schema()
        print("Schema created (constraints + indexes).")
        return 0
    finally:
        syncer.close()


def _cmd_full(args: argparse.Namespace) -> int:
    ast_path = Path(args.ast)
    if not ast_path.exists():
        print(f"ERROR: AST file not found: {ast_path}", file=sys.stderr)
        return 1
    with ast_path.open(encoding="utf-8") as fh:
        ast_data = json.load(fh)

    packages_meta: dict | None = None
    if getattr(args, "packages_from", None):
        mapping_path = Path(args.packages_from)
        if mapping_path.exists():
            with mapping_path.open(encoding="utf-8") as fh:
                mapping = json.load(fh)
            packages_meta = mapping.get("tier_assignments", {})

    syncer = _make_syncer(args)
    syncer.connect()
    try:
        syncer.full_sync(ast_data, args.project, packages_meta=packages_meta)
        print(f"Full sync complete for project '{args.project}'.")
        return 0
    finally:
        syncer.close()


def _cmd_map(args: argparse.Namespace) -> int:
    mapping_path = Path(args.mapping)
    if not mapping_path.exists():
        print(f"ERROR: Mapping file not found: {mapping_path}", file=sys.stderr)
        return 1
    with mapping_path.open(encoding="utf-8") as fh:
        mapping = json.load(fh)

    source = getattr(args, "source_project", "orchestrata")
    target = getattr(args, "target_project", "orchestrata-ultimate")

    syncer = _make_syncer(args)
    syncer.connect()
    try:
        syncer.map_classes(mapping, source, target)
        print(f"MAPS_TO edges created from '{source}' -> '{target}'.")
        return 0
    finally:
        syncer.close()


def _cmd_incremental(args: argparse.Namespace) -> int:
    ast_path = Path(args.ast)
    if not ast_path.exists():
        print(f"ERROR: AST file not found: {ast_path}", file=sys.stderr)
        return 1
    with ast_path.open(encoding="utf-8") as fh:
        ast_data = json.load(fh)

    modules = [m.strip() for m in args.modules.split(",") if m.strip()]
    if not modules:
        print(
            "ERROR: --modules must be a comma-separated list of module paths.",
            file=sys.stderr,
        )
        return 1

    syncer = _make_syncer(args)
    syncer.connect()
    try:
        syncer.incremental_sync(ast_data, args.project, modules)
        print(
            f"Incremental sync complete for {len(modules)} module(s) "
            f"in project '{args.project}'."
        )
        return 0
    finally:
        syncer.close()


def _cmd_stats(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        stats = syncer.get_stats()
        print("\n=== Node Counts ===")
        for label, count in stats["nodes"].items():
            print(f"  {label:<14} {count:>8}")
        print("\n=== Relationship Counts ===")
        for rel_type, count in stats["relationships"].items():
            print(f"  {rel_type:<20} {count:>8}")
        total_nodes = sum(stats["nodes"].values())
        total_rels = sum(stats["relationships"].values())
        print(f"\nTotal nodes: {total_nodes}  |  Total relationships: {total_rels}")
        return 0
    finally:
        syncer.close()


def _cmd_verify(args: argparse.Namespace) -> int:
    syncer = _make_syncer(args)
    syncer.connect()
    try:
        result = syncer.verify(args.project)
        print(f"\n=== Verification: {args.project} ===")
        ok = result.pop("ok", False)
        for key, value in result.items():
            status = "PASS" if value >= 1 else "WARN"
            print(f"  {status}  {key:<20} = {value}")
        if ok:
            print("\nResult: PASS")
            return 0
        else:
            print("\nResult: FAIL — hierarchy incomplete")
            return 2
    finally:
        syncer.close()


def _build_parser() -> argparse.ArgumentParser:
    """Build the top-level argument parser with all subcommands."""
    parser = argparse.ArgumentParser(
        prog="sync_ast_to_graph",
        description="Sync an AST JSON file into a knowledge graph.",
    )
    # Global connection flags
    parser.add_argument(
        "--uri",
        default=None,
        help="Graph Bolt URI (default: KG_ROUTING_URI, falling back to KG_URI).",
    )
    parser.add_argument(
        "--user",
        default=None,
        help="Graph username. No env fallback: omitted means empty credentials.",
    )
    parser.add_argument(
        "--password",
        default=None,
        help="Graph password. No env fallback: omitted means empty credentials.",
    )
    parser.add_argument(
        "--database",
        default=None,
        help="Target database (default: KG_DATABASE, else the server's "
        "default). Needed for orchgraph, which serves orchkg and the "
        "per-worker orchkg_w<N> from one endpoint.",
    )
    parser.add_argument(
        "--dialect",
        default=None,
        choices=["neo4j", "memgraph"],
        help="DDL dialect for the `schema` subcommand (default: resolved "
        "from KG_ENGINE at call time). Memgraph has no "
        "IF NOT EXISTS on any index/constraint form and no FULLTEXT index at "
        "all, so the canonical Neo4j statements are translated (and the "
        "fulltext ones dropped) rather than sent verbatim.",
    )
    parser.add_argument(
        "--no-auth",
        action="store_true",
        help="Connect with NO authentication. Required for orchgraph, which "
        "negotiates none; without it the syncer presents whatever "
        "--user/--password supplied (empty strings by default) to a server "
        "that wants no credentials at all.",
    )

    sub = parser.add_subparsers(dest="command", required=True)

    # schema
    sub.add_parser("schema", help="Create Neo4j constraints and indexes.")

    # full
    full_p = sub.add_parser("full", help="Full sync: clear and re-sync a project.")
    full_p.add_argument("--ast", required=True, help="Path to the AST JSON file.")
    full_p.add_argument("--project", required=True, help="Project name (graph root).")
    full_p.add_argument(
        "--packages-from",
        default=None,
        help="Path to class-mapping.json for package tier/loc metadata.",
    )

    # map
    map_p = sub.add_parser("map", help="Create MAPS_TO edges from class-mapping.json.")
    map_p.add_argument("--mapping", required=True, help="Path to class-mapping.json.")
    map_p.add_argument(
        "--source-project",
        default="orchestrata",
        help="Source project name (default: orchestrata).",
    )
    map_p.add_argument(
        "--target-project",
        default="orchestrata-ultimate",
        help="Target project name (default: orchestrata-ultimate).",
    )

    # incremental
    inc_p = sub.add_parser(
        "incremental", help="Incrementally re-sync specific modules."
    )
    inc_p.add_argument("--ast", required=True, help="Path to the AST JSON file.")
    inc_p.add_argument("--project", required=True, help="Project name.")
    inc_p.add_argument(
        "--modules",
        required=True,
        help="Comma-separated list of module paths to re-sync.",
    )

    # stats
    sub.add_parser("stats", help="Print node/relationship counts.")

    # verify
    verify_p = sub.add_parser(
        "verify", help="Run verification queries against a synced project."
    )
    verify_p.add_argument("--project", required=True, help="Project name to verify.")

    return parser


def main(argv: list[str] | None = None) -> None:
    """Entry point for the CLI wrapper."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "schema": _cmd_schema,
        "full": _cmd_full,
        "map": _cmd_map,
        "incremental": _cmd_incremental,
        "stats": _cmd_stats,
        "verify": _cmd_verify,
    }
    handler = dispatch.get(args.command)
    if handler is None:
        parser.print_help()
        sys.exit(1)

    try:
        exit_code = handler(args)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)

    sys.exit(exit_code)


if __name__ == "__main__":
    main()