orchestrata_ultimate.tools.ast_pipeline

Reference › orchestrata_ultimate › tools › ast_pipeline

This module is the automated AST serialization and comparison pipeline behind the uv run ast-sync command and the script scripts.run_ast_pipeline. It is also imported by scripts.trace_intel. Tests are in tests/tools/test_ast_pipeline.py, tests/tools/test_ast_pipeline_coverage.py and tests/scripts/test_run_ast_pipeline_cli.py. It produces the snapshot (plans/ast-ultimate.json) that the code-intelligence knowledge base and the ast_query engine read. The snapshot format and analyzer come from orchestrata_ast.

main runs the steps by calling the serializer script as a subprocess through run. Step 1 serializes the local package plus the extra source trees returned by source_trees, always with --incremental so only changed files are reparsed. source_trees is read at call time and not frozen at import, so tests can rebind the roots. Step 1b derives the tests snapshot from the snapshot just written with derive_subtree from analyzer, so the two files cannot disagree. Steps 2 and 3 serialize the frozen monorepo and deep-compare the two.

--snapshot-only stops after step 1b. --output-dir writes to another directory instead of plans/, implies snapshot-only, and seeds a cold target from the committed snapshot so the first run is a warm incremental pass. Paths are rooted at get_project_root. main returns 1 with an error on standard error when the serializer script is missing.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

#!/usr/bin/env python3
"""Automated AST serialization and deep-comparison pipeline.

Runs the three-step pipeline that is the source of truth for porting progress:
1. Serialize orchestrata-ultimate (local package)
2. Serialize orchestrata monorepo (frozen source)
3. Deep-compare with semantic class mapping

Usage:
    uv run ast-sync                  # all three steps
    uv run ast-sync --snapshot-only  # step 1/1b only — the KB's input
    # or
    python scripts/run_ast_pipeline.py

``--snapshot-only`` exists because the code-intelligence KB is rebuilt from
``plans/ast-ultimate.json`` alone, while steps 2-3 re-serialize a FROZEN monorepo
and re-derive a comparison against it.

Measured 2026-08-13, on a run where nothing had changed: 60.7s full pipeline,
54.8s with ``--snapshot-only``. The 6s gap says steps 2-3 were never the problem —
step 1 was re-parsing all 2819 modules every time, because the serializer's
``--incremental`` warm start existed but was never passed from here. It is now.
Freshness that costs a minute per edit does not get run, and a KB nobody refreshes
answers confidently about code that no longer exists.
"""

from __future__ import annotations

import argparse
import json
import shutil
import subprocess
import sys
from pathlib import Path

from orchestrata_ast.analyzer import derive_subtree
from orchestrata_ultimate.core.paths import get_project_root

#: Resolved through ``get_project_root`` (``ORCHESTRATA_ROOT``, else the nearest
#: ``pyproject.toml`` above this file), never a fixed ``__file__`` parent count.
#: Since the proper install (dd919d55c, 2026-10-20) this module lives in
#: ``.venv/lib/python3.14t/site-packages/``, where four parents up is
#: ``.venv/lib/python3.14t`` — every tree below was rooted there.
REPO_ROOT = get_project_root()
PLANS_DIR = REPO_ROOT / "plans"
SOURCE_ULTIMATE = REPO_ROOT / "src" / "orchestrata_ultimate"
#: The trees ``ast-ultimate.json`` covers, in order. The FIRST is the primary
#: root that dotted module ids are numbered from; the rest contribute modules
#: prefixed with their own directory name.
#:
#: Mohamed, 2026-08-08: "the ast must see tests and scripts too." Before this,
#: every consumer of the snapshot — the Prolog gate, code_intel, the RTM — was
#: structurally unable to answer a question about a test or a script, and the
#: gate demonstrated the cost by reporting PASS over a fact base in which
#: ``file_class`` was 850/850 ``src``.
#:
#: THIS list is the single source of truth for the snapshot's scope. A second
#: copy in the shell wrapper is why that wrapper now delegates here instead of
#: rebuilding the command: whichever producer ran last would otherwise decide
#: the scope, silently.
#: 2026-08-13: ``src/orchestrata_ast`` and ``src/orchestrata_tools`` joined the list.
#: They are sibling packages under ``src/`` that the snapshot did not cover, so the
#: affected-tests gate could not bound a change to either one and fell back to the
#: smoke floor — reported on a real push as "src/orchestrata_ast/analyzer.py is under
#: src/ but outside the serialized import closure ... cannot bound its blast radius".
#: Both HAVE tests (tests/test_orchestrata_ast_*.py, tests/test_orchestrata_tools_*.py);
#: nothing could map a source change onto them.
#: 2026-08-19: ``claude`` joined, for the same reason and with a sharper edge.
#: It holds 143 tracked files — the hooks, tools, agents and skills deployed to
#: ~/.claude — i.e. the context-injection machinery itself (SPEC-107 FR-013).
#: Being outside the snapshot made that machinery structurally unable to answer
#: about its own source: ``code_intel module-api claude/tools/module_context.py``
#: returned ``unknown_module``, and the per-module injection reported
#: "[AST no KB for this project]" on every file under ``claude/`` — the AST layer
#: only, no dependency contracts and no blast radius, for the one tree that runs
#: on every prompt. The affected-tests gate had the matching gap and papered over
#: it with hand-written ``claude/`` rules in plans/affected-tests-manifest.json,
#: which is coverage asserted by a human rather than derived from the closure.
#: Measured while fixing refresh_snapshot: editing it without contracts is how
#: the ``ast-sync`` name collision went unseen.
#: Ids are unambiguous by construction — extra roots lead with their own
#: directory name (analyzer._file_to_module_path), so these are ``claude.tools.*``
#: and ``claude.hooks.*``, never colliding with the primary root's ``tools.*``
#: from ``src/orchestrata_ultimate/tools``.
EXTRA_SOURCE_TREES = (
    REPO_ROOT / "tests",
    REPO_ROOT / "scripts",
    REPO_ROOT / "src" / "orchestrata_ast",
    REPO_ROOT / "src" / "orchestrata_tools",
    REPO_ROOT / "claude",
)


def source_trees() -> tuple[Path, ...]:
    """The trees to serialize, read at CALL time.

    Not a module constant, deliberately. This module's callers rebind
    ``SOURCE_ULTIMATE``/``SOURCE_ORCHESTRATA`` to a tiny ``tmp_path`` tree to
    drive :func:`main` against real files without serializing the repo. A tuple
    frozen at import time silently ignores that rebinding — measured: the
    pipeline test kept passing while serializing the REAL 2,722-module tree,
    taking one test module from ~1s to 204s and asserting nothing it claimed to.
    """
    return (SOURCE_ULTIMATE, *EXTRA_SOURCE_TREES)


SOURCE_ORCHESTRATA = Path("/home/mohamed/workspace/orchestrata/packages")
SERIALIZER = REPO_ROOT / "scripts" / "serialize_codebase_ast.py"
MAPPING = PLANS_DIR / "class-mapping.json"


def run(cmd: list[str], description: str) -> None:
    print(f"▶ {description} ...", flush=True)
    result = subprocess.run(
        [sys.executable, str(SERIALIZER), *cmd],
        cwd=REPO_ROOT,
        capture_output=False,
        text=True,
    )
    if result.returncode != 0:
        sys.exit(result.returncode)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Serialize the AST snapshot and (by default) deep-compare it "
        "against the frozen monorepo."
    )
    parser.add_argument(
        "--snapshot-only",
        action="store_true",
        help=(
            "Run only step 1/1b — write plans/ast-ultimate.json and derive "
            "plans/ast-tests.json — and skip serializing the frozen monorepo and "
            "the deep comparison. This is the stage the Prolog code-intelligence "
            "KB is built from, so it is the only stage a freshness refresh needs. "
            "MEASURED 2026-08-13 on a no-change run: 60.7s full, 54.8s with this "
            "flag — so steps 2-3 are only ~6s and were NOT the cost; step 1 was, "
            "and the fix for that was passing --incremental to the serializer "
            "(see step 1). This flag still earns its keep by not re-deriving a "
            "comparison against a frozen monorepo on every refresh, but it is the "
            "smaller of the two savings, not the main one."
        ),
    )
    parser.add_argument(
        "--output-dir",
        default=None,
        help=(
            "Write the snapshot (and the derived tests projection) into this "
            "directory instead of plans/. Implies --snapshot-only. This exists "
            "so an automatic, on-demand refresh has somewhere to write that is "
            "not a committed artifact: plans/ast-ultimate.json is tracked, and "
            "a refresh that rewrites it leaves the tree dirty and churns it "
            "into unrelated commits (DESIGN_WAIVER 2026-08-09, backlog #284) — "
            "measured again 2026-08-22, one refresh modified two tracked "
            "files. A cold target is SEEDED from the committed snapshot first, "
            "so the first live refresh is a warm incremental pass (~13s, "
            "reusing ~2970 modules) and not a full cold parse."
        ),
    )
    args = parser.parse_args(argv)

    if not SERIALIZER.exists():
        print(f"ERROR: AST serializer not found at {SERIALIZER}", file=sys.stderr)
        return 1

    PLANS_DIR.mkdir(exist_ok=True)

    # --output-dir redirects steps 1/1b only, and forces snapshot-only: steps
    # 2-3 compare against the frozen monorepo via PLANS_DIR, so honouring a
    # redirect for step 1 while step 3 still read plans/ would deep-compare a
    # snapshot against a DIFFERENT snapshot's comparison inputs and report
    # drift that is an artifact of the redirect.
    out_dir = PLANS_DIR if args.output_dir is None else Path(args.output_dir)
    snapshot_only = args.snapshot_only or args.output_dir is not None
    committed_snapshot = PLANS_DIR / "ast-ultimate.json"
    live_snapshot = out_dir / "ast-ultimate.json"
    if out_dir != PLANS_DIR:
        out_dir.mkdir(parents=True, exist_ok=True)
        # Seed a cold target so --incremental has a freshness map to diff
        # against. Without this the first refresh is a full cold parse of
        # ~2977 modules; with it, it is a warm pass over the handful that
        # actually moved since the committed snapshot was cut.
        if not live_snapshot.exists() and committed_snapshot.exists():
            shutil.copy2(committed_snapshot, live_snapshot)
            print(f"▶ Seeded {live_snapshot} from {committed_snapshot.name}")

    # Step 1 — serialize ultimate
    sources: list[str] = []
    for tree in source_trees():
        sources.extend(["--source", str(tree)])
    # --incremental warm-starts from the snapshot already on disk and reparses
    # only files whose content changed, falling back to a full serialize on any
    # mismatch (same name/root, file_meta present) — the serializer guarantees an
    # identical output either way, so this is a cost change, not a result change.
    #
    # It was NEVER passed here until 2026-08-13, which made every "incremental
    # refresh" a full re-serialize of 2819 modules. Measured on a no-change run:
    # 60.7s for the whole pipeline, 54.8s with steps 2-3 skipped — i.e. the frozen
    # monorepo and the deep compare were never the cost, step 1 was, and the flag
    # that fixes it had existed on the serializer the whole time. A refresh this
    # expensive is a refresh that gets skipped, which is how a code-intelligence
    # KB ends up answering about code that no longer exists.
    run(
        [
            "serialize",
            *sources,
            "--name",
            "orchestrata-ultimate",
            "--output",
            str(live_snapshot),
            "--incremental",
        ],
        "Serializing orchestrata-ultimate",
    )

    # Step 1b — project the tests tree out of the snapshot just written.
    #
    # DERIVED, never re-parsed. `plans/ast-tests.json` used to be a second
    # `serialize --source tests` run, and two snapshots of the same files made
    # by different invocations drift: the checked-in one predated four fields
    # the wide snapshot already carried, and it answered `is_pydantic: False`
    # for a class whose base lives in src/ — a question it was structurally
    # unable to get right, because it could not see the base at all.
    tests_tree = REPO_ROOT / "tests"
    if tests_tree in EXTRA_SOURCE_TREES:
        ultimate = live_snapshot
        derived = derive_subtree(
            json.loads(ultimate.read_text(encoding="utf-8")),
            tree=tests_tree.name,
            name="orchestrata-ultimate-tests",
            repo_root=REPO_ROOT,
        )
        out = out_dir / "ast-tests.json"
        out.write_text(json.dumps(derived, indent=2, default=str), encoding="utf-8")
        print(
            f"▶ Derived {out} from {ultimate.name} ({len(derived['modules'])} modules)"
        )

    if snapshot_only:
        why = "--snapshot-only" if args.snapshot_only else "--output-dir"
        print(f"\n✅ AST snapshot complete (steps 2-3 skipped: {why})")
        return 0

    # Step 2 — serialize orchestrata (frozen monorepo)
    if SOURCE_ORCHESTRATA.exists():
        run(
            [
                "serialize",
                "--source",
                str(SOURCE_ORCHESTRATA),
                "--name",
                "orchestrata",
                "--output",
                str(PLANS_DIR / "ast-orchestrata.json"),
                "--monorepo",
            ],
            "Serializing orchestrata monorepo",
        )
    else:
        print(
            f"⚠ Skipping orchestrata serialization — source not found at {SOURCE_ORCHESTRATA}"
        )

    # Step 3 — deep compare
    if (PLANS_DIR / "ast-ultimate.json").exists() and (
        PLANS_DIR / "ast-orchestrata.json"
    ).exists():
        run(
            [
                "deep-compare",
                str(PLANS_DIR / "ast-ultimate.json"),
                str(PLANS_DIR / "ast-orchestrata.json"),
                "--mapping",
                str(MAPPING),
                "--output",
                str(PLANS_DIR / "ast-deep-comparison.json"),
            ],
            "Running deep comparison",
        )
    else:
        print("⚠ Skipping deep compare — missing AST inputs")

    print("\n✅ AST pipeline complete")
    return 0


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