scripts.ast_live_sync

Reference › scripts › ast_live_sync

Live code-intelligence sync: when a source file is saved, the code graph is updated within seconds. It runs the AST analyzer from analyzer over the declared roots from roots, serializes the result with serialize and pushes the difference to the graph through a GraphSyncer. A long-running service or a human runs it. tests/scripts/test_ast_live_sync.py and tests/scripts/test_ast_live_sync_main.py import it.

Invocation: main accepts --uri (default taken from KG_ROUTING_URI, else MEMGRAPH_URI), --database (default orchkg), --project (default orchestrata-ultimate), --repo-root, --source-dir, --extra-root, --cache, --watch, --debounce-ms (1000), --once for a single pass, and --measure with --slo-seconds to time a sync against a budget. It exits 2 when the repo root or source directory is missing, and it refuses to run when no source roots are declared, so the graph is never narrowed to a single tree.

LiveCodeSync holds the cycle. sync_once analyses, compares with a cache file and writes the changes to the graph, returning a SyncOutcome. watch is an async loop that debounces file events, using is_watchable and filter_changes. assert_cache_root_matches and assert_destination_populated guard against syncing against the wrong root or an empty destination, and _preserve_runtime_calls keeps runtime call edges across passes.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

SyncOutcome (dataclass)

Decorators: dataclass(frozen=True) · Line 152

What one pass actually changed, and how long the whole leg took.

LiveCodeSync

Line 211

Keeps the graph's code-intelligence subgraph current with the tree.

Functions

Source

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

"""Live code-intelligence sync: file save -> graph updated in seconds.

Step 3.4 of ``plans/neo4j-to-orchgraph-migration-plan.md``. Wires three
components that already exist and are tested, rather than growing a fourth:

    watchfiles.awatch (debounced)
        -> orchestrata_ast.load_or_analyze  (warm: re-parses ONLY the files
           whose content hash changed, reusing every other module's cached
           dict verbatim)
        -> GraphSyncer.incremental_sync     (deletes + recreates just those
           modules and repairs their cross-edges, failing loudly on a module
           path the AST does not contain)

The AST cache doubles as the freshness map: ``load_or_analyze`` stat+hashes
every live file against the ``file_meta`` block of the previous serialize and
returns ``stats["reparsed"]`` -- exactly the module list ``incremental_sync``
wants. That is why this does not re-implement change detection: the watcher
says "something under src/ moved", and the hash map says which modules
actually differ. An editor that rewrites a file with identical content
produces a watch event and zero graph writes.

**SLO:** file save -> graph updated, end to end, < 5 s. ``--measure`` runs
that leg once against a real file and exits non-zero if it misses, which is
what ``benchmark_code_intel.py`` deliberately does not cover (that harness
measures query latency; this is pipeline latency).

Deployed as the ``orchestrata-ast-live-sync`` user systemd unit -- always on,
not a remembered command.

Usage::

    python scripts/ast_live_sync.py                      # run the service
    python scripts/ast_live_sync.py --once               # one pass, then exit
    python scripts/ast_live_sync.py --measure            # SLO check
"""

from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import sys
import time
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any

_REPO_ROOT = Path(__file__).resolve().parent.parent
_REPO_SRC = _REPO_ROOT / "src"
if str(_REPO_SRC) not in sys.path:
    sys.path.insert(0, str(_REPO_SRC))
# Repo root on the path so the sibling ``scripts.*`` modules import whether
# this file is run as ``python scripts/ast_live_sync.py`` (sys.path[0] is
# ``scripts/``) or imported as ``scripts.ast_live_sync`` (test context).
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from orchestrata_ast.analyzer import (  # noqa: E402
    compute_file_meta,
    load_or_analyze,
)
from orchestrata_ast.roots import (  # noqa: E402
    declared_roots,
    existing_roots,
    split_roots,
    undeclared_message,
)
from orchestrata_ast.serialize import serialize  # noqa: E402
from orchestrata_ultimate.knowledge_graph.sync import GraphSyncer  # noqa: E402
from scripts.build_runtime_callgraph import (  # noqa: E402
    augment_ast,
    collect_ast_nodes,
)

# The committed runtime call-graph summary (SPEC-023). ``serialize()`` knows
# nothing about the ``runtime_calls`` field -- only build_runtime_callgraph.py
# writes it -- so a fresh serialize DROPS it. This daemon re-serializes its
# own untracked freshness-map cache (``--cache``, default
# ``_DEFAULT_CACHE_PATH`` below -- never the committed ``plans/ast-ultimate
# .json`` artifact) on every source change, which without a preservation
# step silently strips the merged static+runtime call graph that
# scripts/derive_rtm.py, scripts/code_intel.py's merged queries, and
# ``trace-path`` all depend on. We re-apply ``runtime_calls`` from this
# committed summary after every serialize (the same edges reaugment uses),
# deterministically and idempotently.
_RUNTIME_CALLGRAPH_SUMMARY = _REPO_ROOT / "plans" / "runtime-callgraph.json"

# Untracked sidecar (see .gitignore) that doubles as the live-sync
# freshness map. plans/ast-ultimate.json stays CLI-only: it is the
# snapshot scripts/gates.py's ast-refresh step regenerates and the
# Prolog fact base reads, and this always-on service rewriting it kept
# it permanently dirty and churned it in unrelated commits (context.md
# backlog #284, DESIGN_WAIVER 2026-08-09).
_DEFAULT_CACHE_PATH = ".ast_daemon/ast-live-cache.json"

# The committed snapshot this service seeds ITS sidecar from on a cold
# start (sidecar absent, artifact present) -- never written back to.
_COMMITTED_AST_CACHE = _REPO_ROOT / "plans" / "ast-ultimate.json"

_logger = logging.getLogger("ast_live_sync")


# The analyzer roots the graph's module ids are RELATIVE to. They must match
# the roots `plans/ast-ultimate.json` was serialized from, because a module's
# canonical id is its dotted path from its root: parsing `src/` instead names
# the same file `orchestrata_ultimate.agents.communication` where the graph
# holds `agents.communication`, so an "incremental" sync inserts a second,
# parallel copy of the entire codebase rather than updating it. Measured
# 2026-07-26: one pass against `src/` took orchkg from 783 Modules to 1621 and
# had to be repaired with a full re-derive.
#
# NO LITERAL HERE (2026-10-09). This file used to pin
# `DEFAULT_SOURCE_DIR = "src/orchestrata_ultimate"` while the serializer read
# `$ORCHESTRATA_AST_ROOTS`, so the snapshot covered six roots and the graph
# one: 3,171 modules in the AST, 905 in the graph, every scripts/, claude/
# and tests/ module silently absent from code intelligence. Both tools now
# read the SAME declaration (env/ast-roots.env via orchestrata_ast.roots):
# the first root is primary, the rest are `extra_roots`, and the watch set is
# all of them. An undeclared set is a refusal, not a fallback.
def default_roots(
    env: Mapping[str, str] | None = None,
) -> tuple[str | None, tuple[str, ...], tuple[str, ...]]:
    """``(source_dir, extra_roots, watch_dirs)`` from the declared root set.

    ``(None, (), ())`` when nothing is declared -- the caller decides how loud
    to be; a service must refuse rather than narrow the graph.
    """
    primary, extra = split_roots(declared_roots(env))
    if primary is None:
        return None, (), ()
    return primary, extra, (primary, *extra)


# Paths that change constantly and carry no code intelligence. Without these
# the watcher wakes on its own output (the AST cache) and on every pytest run.
IGNORE_FRAGMENTS: tuple[str, ...] = (
    "/.git/",
    "/.venv/",
    "/__pycache__/",
    "/.pytest_cache/",
    "/node_modules/",
    "/reports/",
    "/reviews/",
    "/backups/",
)


@dataclass(frozen=True)
class SyncOutcome:
    """What one pass actually changed, and how long the whole leg took."""

    reparsed: tuple[str, ...]
    removed: tuple[str, ...]
    nodes: int
    relationships: int
    duration_s: float
    fallback: bool
    # How many modules the LOCAL AST holds for this project. Carried so the
    # destination check can compare "what the tree has" against "what the graph
    # has"; a pass that writes nothing is only correct if the graph is already
    # populated, and nothing else in this outcome can tell those apart.
    total_modules: int = 0

    @property
    def changed(self) -> bool:
        return bool(self.reparsed or self.removed)


def is_watchable(path: str, watch_dirs: Sequence[str], repo_root: Path) -> bool:
    """Whether a changed path belongs to the code-intelligence graph.

    Filtering here rather than inside the sync keeps a ``reports/*.json``
    write from costing a hash sweep of the whole tree. Watch dirs are matched
    as path PREFIXES, not first components, because the analyzer root is
    nested (``src/orchestrata_ultimate``) -- a first-component match would
    accept everything under ``src/``, which is the mis-scoping that inserted
    a duplicate codebase on 2026-07-26.
    """
    if not path.endswith(".py"):
        return False
    normalized = f"/{Path(path).as_posix().lstrip('/')}"
    if any(fragment in normalized for fragment in IGNORE_FRAGMENTS):
        return False
    try:
        relative = Path(path).resolve().relative_to(repo_root.resolve())
    except ValueError:
        return False
    return any(
        relative == Path(watched) or Path(watched) in relative.parents
        for watched in watch_dirs
    )


def filter_changes(
    paths: Iterable[str], watch_dirs: Sequence[str], repo_root: Path
) -> list[str]:
    """The watchable subset of one debounced batch, order-stable."""
    seen: set[str] = set()
    kept: list[str] = []
    for path in paths:
        if path in seen or not is_watchable(path, watch_dirs, repo_root):
            continue
        seen.add(path)
        kept.append(path)
    return kept


class LiveCodeSync:
    """Keeps the graph's code-intelligence subgraph current with the tree.

    Args:
        repo_root: Repository root; also the analyzer's source root.
        cache_path: The serialize output that doubles as the freshness map
            (an untracked sidecar by default -- ``_DEFAULT_CACHE_PATH``,
            never the committed ``plans/ast-ultimate.json`` artifact).
            Rewritten after every successful pass so the next one only
            re-parses what changed since.
        project: Project id the graph nodes are namespaced under.
        syncer: A CONNECTED ``GraphSyncer`` aimed at the target graph.
        source_dir: Subdirectory the analyzer parses (``src``).
    """

    def __init__(
        self,
        repo_root: Path,
        cache_path: Path,
        project: str,
        syncer: GraphSyncer,
        *,
        source_dir: str,
        extra_roots: Sequence[str] = (),
    ) -> None:
        self._repo_root = repo_root
        self._cache_path = cache_path
        self._project = project
        self._syncer = syncer
        self._source = repo_root / source_dir
        # Secondary trees, analyzed alongside the primary root and prefixed
        # with their own directory name (tests.x, scripts.x) -- the same
        # contract as the serializer's repeated --source.
        self._extra_roots: tuple[Path, ...] = tuple(repo_root / r for r in extra_roots)
        self._passes = 0

    @property
    def source_root(self) -> Path:
        """The directory module ids are relative to (the analyzer root)."""
        return self._source

    @property
    def extra_roots(self) -> tuple[Path, ...]:
        """The secondary trees this sync analyzes (declared, not assumed)."""
        return self._extra_roots

    @property
    def passes(self) -> int:
        """Completed sync passes — the liveness signal for the unit."""
        return self._passes

    def assert_cache_root_matches(self) -> None:
        """Refuse to sync when the cache was built from a different root.

        ``load_or_analyze`` treats a root mismatch as a cache miss and
        silently full-reparses -- correct for a one-shot CLI, catastrophic
        for a service: module ids are dotted paths RELATIVE to the root, so
        parsing a different root renames every module and
        ``incremental_sync`` inserts a parallel copy of the whole codebase
        instead of updating it. That is exactly what happened on 2026-07-26
        (783 Modules -> 1621). A loud refusal costs one restart; the silent
        fallback cost a full re-derive.
        """
        try:
            cached_root = json.loads(self._cache_path.read_text(encoding="utf-8")).get(
                "root_path"
            )
        except OSError, ValueError:
            return  # no usable cache: a full parse is the correct outcome
        if not cached_root:
            return
        if Path(str(cached_root)).resolve() != self._source.resolve():
            raise SystemExit(
                f"AST cache {self._cache_path} was built from "
                f"{cached_root!r}, but this sync parses "
                f"{str(self._source)!r}. Module ids are relative to the root, "
                "so syncing would insert a duplicate codebase rather than "
                "update it. Regenerate the cache from the same root, or pass "
                "--cache for the matching one."
            )
        # The set of trees is part of the cache's identity (load_or_analyze
        # checks it too, but falls back to a SILENT full reparse). A service
        # must refuse instead: a cache built for six roots warm-starting a
        # sync declared for one would drop five trees from the graph on the
        # next pass without a word.
        try:
            cached_extra = json.loads(self._cache_path.read_text(encoding="utf-8")).get(
                "extra_roots"
            )
        except OSError, ValueError:
            return
        if cached_extra is None:
            return
        want = [str(p.resolve()) for p in self._extra_roots]
        have = [str(Path(str(p)).resolve()) for p in cached_extra]
        if have != want:
            raise SystemExit(
                f"AST cache {self._cache_path} covers extra roots {have}, but "
                f"this sync declares {want}. The root set is part of the "
                "cache's identity; regenerate the cache for the declared set "
                "(env/ast-roots.env) or pass --cache for the matching one."
            )

    def sync_once(self) -> SyncOutcome:
        """Re-parse what changed and push exactly those modules to the graph.

        Returns a :class:`SyncOutcome` even when nothing changed (``nodes``
        and ``relationships`` zero), because "the watcher fired and the graph
        needed no write" is a real, reportable result -- not an error.
        """
        self.assert_cache_root_matches()
        started = time.perf_counter()
        result, stats = load_or_analyze(
            self._project,
            self._source,
            self._cache_path,
            extra_roots=self._extra_roots,
        )
        reparsed: list[str] = list(stats.get("reparsed", []))
        removed: list[str] = list(stats.get("removed", []))
        fallback = bool(stats.get("fallback", False))

        ast_data = _as_ast_dict(result)
        nodes = relationships = 0
        if removed:
            # A deleted file has no AST entry left, so incremental_sync
            # cannot express this at all (it raises on module paths the AST
            # does not contain). Without this the sync reported "1 removed"
            # while the module's whole subtree stayed in the graph forever —
            # observed live 2026-07-26, Modules stuck at 785 after a delete.
            self._syncer.delete_modules(self._project, removed)
        if reparsed:
            # incremental_sync deletes + recreates only these modules and
            # repairs their cross-edges; it raises on a module path the AST
            # does not contain rather than silently syncing nothing.
            counts = self._syncer.incremental_sync(ast_data, self._project, reparsed)
            nodes = int(counts.get("nodes_created", 0))
            relationships = int(counts.get("relationships_created", 0))

        if reparsed or removed:
            # Persist AFTER the graph write: if the write raises, the cache
            # still describes the graph's actual contents, so the next pass
            # retries the same modules instead of believing them synced.
            self._write_cache(ast_data)

        self._passes += 1
        return SyncOutcome(
            reparsed=tuple(reparsed),
            removed=tuple(removed),
            nodes=nodes,
            relationships=relationships,
            total_modules=len(ast_data.get("modules", ()) or ()),
            duration_s=time.perf_counter() - started,
            fallback=fallback,
        )

    def _write_cache(self, ast_data: dict[str, Any]) -> None:
        """Rewrite the freshness map atomically.

        A torn cache would send the next pass down ``load_or_analyze``'s
        full-fallback path -- correct, but a multi-minute reparse of the whole
        tree in a service meant to react in seconds.
        """
        tmp = self._cache_path.with_suffix(self._cache_path.suffix + ".tmp")
        tmp.parent.mkdir(parents=True, exist_ok=True)
        with open(tmp, "w", encoding="utf-8") as handle:
            json.dump(ast_data, handle, indent=2, default=str)
        tmp.replace(self._cache_path)

    async def watch(
        self,
        watch_dirs: Sequence[str],
        *,
        debounce_ms: int = 1000,
        stop: asyncio.Event | None = None,
    ) -> None:
        """Run until *stop* is set, syncing after each debounced batch."""
        from watchfiles import awatch

        targets = [str(self._repo_root / d) for d in watch_dirs]
        _logger.info("watching %s (debounce %d ms)", ", ".join(targets), debounce_ms)
        async for changes in awatch(*targets, debounce=debounce_ms, stop_event=stop):
            paths = filter_changes(
                (path for _, path in changes), watch_dirs, self._repo_root
            )
            if not paths:
                continue
            try:
                outcome = await asyncio.to_thread(self.sync_once)
            except Exception as exc:
                # A watcher that dies on one bad parse stops being live. Log
                # loudly and keep watching; the next save retries.
                _logger.error("sync pass failed: %s", exc)
                continue
            _log_outcome(outcome)


def _as_ast_dict(result: Any) -> dict[str, Any]:
    """Serialized AST plus the ``file_meta`` freshness map the cache needs."""
    data = serialize(result)
    if not isinstance(data, dict):  # defensive: serialize contract is a dict
        raise TypeError(f"serialize() returned {type(data).__name__}, expected dict")
    data["file_meta"] = compute_file_meta(result.modules)
    _preserve_runtime_calls(data)
    return data


def _preserve_runtime_calls(
    ast_data: dict[str, Any], summary_path: Path = _RUNTIME_CALLGRAPH_SUMMARY
) -> None:
    """Re-apply the ``runtime_calls`` field that ``serialize()`` drops.

    ``serialize()`` reconstructs the AST from source and has no notion of the
    SPEC-023 runtime call-graph, so it emits functions/methods WITHOUT a
    ``runtime_calls`` field. Persisting that as ``plans/ast-ultimate.json``
    strips the merged static+runtime graph that ``codebase_db.pl`` asserts
    ``runtime_call/2`` from -- breaking ``scripts/derive_rtm.py``,
    ``scripts/code_intel.py``'s merged reachability, and ``trace-path``.

    We deterministically restore it from the committed
    ``plans/runtime-callgraph.json`` summary (the same edge set
    ``build_runtime_callgraph.py reaugment`` uses), mutating ``ast_data`` in
    place. If the summary is absent (no runtime capture committed) we leave
    the AST as serialized rather than fail the live sync -- callers that need
    the field will hit the loud guard in derive_rtm.py, not a corrupt cache.
    """
    if not summary_path.exists():
        return
    with open(summary_path, encoding="utf-8") as fh:
        summary = json.load(fh)
    edges = [(pair[0], pair[1]) for pair in summary.get("edges", [])]
    _, func_index = collect_ast_nodes(ast_data)
    augment_ast(func_index, edges)


def _log_outcome(outcome: SyncOutcome) -> None:
    if not outcome.changed:
        _logger.info("no code change after hashing (%.2fs)", outcome.duration_s)
        return
    _logger.info(
        "synced %d module(s) (%d removed) -> %d nodes / %d rels in %.2fs%s",
        len(outcome.reparsed),
        len(outcome.removed),
        outcome.nodes,
        outcome.relationships,
        outcome.duration_s,
        " [full reparse fallback]" if outcome.fallback else "",
    )


def resolve_target_uri(env: Mapping[str, str], explicit: str | None) -> str:
    """Address of the graph to keep current, from .env at call time (RUL-003).

    Prefers ``KG_ROUTING_URI`` and falls back to ``MEMGRAPH_URI``, matching
    ``benchmark_code_intel.py``: the former is blank until the Phase 4.2
    cutover fills it, the latter is the orchgraph routing address today.
    """
    if explicit:
        return explicit
    for key in ("KG_ROUTING_URI", "MEMGRAPH_URI"):
        value = env.get(key, "").strip()
        if value:
            return value
    raise SystemExit(
        "neither KG_ROUTING_URI nor MEMGRAPH_URI is set, and no --uri was passed"
    )


def _build_parser(env: Mapping[str, str] | None = None) -> argparse.ArgumentParser:
    """The CLI. Root defaults come from the declared set (``env`` is the seam)."""
    source_dir, extra_roots, watch_dirs = default_roots(env)
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--uri", default=None, help="default: $KG_ROUTING_URI, else $MEMGRAPH_URI"
    )
    parser.add_argument("--database", default="orchkg")
    parser.add_argument("--project", default="orchestrata-ultimate")
    # Repo root and source dir were pinned to THIS checkout (`_REPO_ROOT` from
    # __file__, `src/orchestrata_ultimate`), which made a tool that is otherwise
    # fully project-generic — nodes are already namespaced by `--project` — usable
    # in exactly one repository. dr-hannah was absent from the AST graph for that
    # reason alone.
    parser.add_argument(
        "--repo-root",
        default=None,
        help="project root to analyze (default: this repository)",
    )
    parser.add_argument(
        "--source-dir",
        default=source_dir,
        help=(
            "primary analyzer root relative to --repo-root (default: the first "
            f"declared root, currently {source_dir!r})"
        ),
    )
    parser.add_argument(
        "--extra-root",
        action="append",
        default=None,
        help=(
            "secondary tree relative to --repo-root, repeatable; its modules are "
            "prefixed with the directory name (default: the declared roots after "
            f"the first, currently {' '.join(extra_roots) or '<none>'})"
        ),
    )
    parser.add_argument(
        "--cache",
        default=_DEFAULT_CACHE_PATH,
        help=(
            "serialize output used as the freshness map (default: an "
            "untracked sidecar, never the committed plans/ast-ultimate.json "
            "artifact)"
        ),
    )
    parser.add_argument(
        "--watch",
        nargs="+",
        default=None,
        help=(
            "directories to watch (default: every analyzed root, currently "
            f"{' '.join(watch_dirs) or '<none declared>'})"
        ),
    )
    parser.add_argument("--debounce-ms", type=int, default=1000)
    parser.add_argument(
        "--once", action="store_true", help="run one sync pass and exit"
    )
    parser.add_argument(
        "--measure",
        action="store_true",
        help="measure the file-save -> graph SLO once and exit non-zero on a miss",
    )
    parser.add_argument(
        "--slo-seconds",
        type=float,
        default=5.0,
        help="file-save -> graph budget (default: 5.0)",
    )
    return parser


def _make_syncer(uri: str, database: str) -> GraphSyncer:
    """A connected syncer aimed at the unauthenticated orchgraph cluster."""
    syncer = GraphSyncer(uri=uri, database=database, no_auth=True, dialect="memgraph")
    syncer.connect()
    return syncer


# A pass may legitimately write nothing: the tree is unchanged AND the graph
# already holds it. The failure this guard exists for is the OTHER shape --
# unchanged tree, EMPTY graph -- which is indistinguishable in the outcome.
# Below this fraction of the local module count, the destination is treated as
# unpopulated rather than merely stale.
_DESTINATION_MIN_COVERAGE = 0.5


def assert_destination_populated(
    syncer: GraphSyncer, project: str, total_modules: int
) -> int:
    """Refuse to report success for a sync whose target did not receive the tree.

    Freshness here is decided by the AST CACHE, never by the destination. A warm
    cache reports "0 modules changed" and writes nothing -- and does so just as
    happily when the target graph holds ZERO modules for the project. That
    combination printed ``RESULT: PASS`` having populated nothing (measured
    2026-08-23 against a graph with 0 Module nodes for ``orchestrata-ultimate``;
    the run reported ``modules reparsed: 1, nodes/rels written: 3/2503`` and
    exited 0).

    The cache can only say what changed LOCALLY. Only the graph can say what it
    actually holds, so success has to be asserted against the graph.

    Returns 0 when the destination is populated, 1 when it is not. Never raises
    on a merely-empty result: the caller turns this into an exit code, and a
    swallowed exception here would restore the exact false green it removes.
    """
    rows = syncer._connected_repo.execute_read(
        "MATCH (m:Module {project: $project}) RETURN count(m) AS cnt",
        {"project": project},
    )
    actual = int(rows[0]["cnt"]) if rows else 0
    floor = int(total_modules * _DESTINATION_MIN_COVERAGE)

    if actual == 0:
        print(
            f"\nRESULT: FAIL — the target graph holds 0 Module nodes for project "
            f"{project!r} while the local AST holds {total_modules}. The AST cache "
            f"reported the tree as unchanged, so this pass wrote nothing into an "
            f"EMPTY destination. Populate it with a full sync:\n"
            f"  scripts/sync_ast_to_graph.py full "
            f"--ast plans/ast-ultimate.json --project {project}",
            file=sys.stderr,
        )
        return 1

    if total_modules and actual < floor:
        print(
            f"\nRESULT: FAIL — the target graph holds {actual} Module nodes for "
            f"project {project!r} but the local AST holds {total_modules} "
            f"({actual / total_modules:.0%} coverage, floor {_DESTINATION_MIN_COVERAGE:.0%}). "
            f"An incremental pass cannot close a gap this large because the cache "
            f"believes those modules are already synced. Run a full sync.",
            file=sys.stderr,
        )
        return 1

    print(
        f"  destination        : {actual} modules in graph (local AST {total_modules})"
    )
    return 0


def measure_slo(sync: LiveCodeSync, budget_s: float) -> int:
    """Time one real save -> graph leg; return 0 only if it fits the budget.

    Writes a NEW module into the source root, measures the save -> graph
    leg, then deletes it and syncs the removal. A new file is used rather
    than an edit to an existing one because:

    * ``touch()`` is not enough. Freshness is decided by CONTENT hash, so
      an mtime-only change reparses nothing and writes nothing -- the first
      version of this function reported PASS at 0.552 s having produced 0
      nodes, timing a hash sweep and calling it a pipeline measurement.
    * Editing a real module would have to be reverted, and a crash between
      the edit and the revert would leave the tree modified.

    The probe file is uuid-named so a crashed run leaves something obviously
    disposable rather than a plausible-looking module.
    """
    sync.sync_once()  # settle: make the cache describe the tree exactly

    import uuid

    probe = sync.source_root / f"_slo_probe_{uuid.uuid4().hex[:10]}.py"
    probe.write_text(
        '"""Transient probe written by ast_live_sync --measure."""\n\n\n'
        "def _slo_probe() -> None:\n    return None\n",
        encoding="utf-8",
    )
    try:
        started = time.perf_counter()
        outcome = sync.sync_once()
        elapsed = time.perf_counter() - started
    finally:
        probe.unlink(missing_ok=True)
        sync.sync_once()  # propagate the removal to the graph and the cache

    if not outcome.reparsed:
        # The probe was written but nothing reparsed: the pipeline is not
        # detecting new files at all. Reporting the elapsed time as a PASS
        # here is precisely the false green this guard exists to stop.
        print(
            "\nRESULT: FAIL — the probe module was written but no module was "
            "reparsed; the change-detection leg is broken"
        )
        return 1

    print(f"  file-save -> graph : {elapsed:.3f} s  budget={budget_s:.3f} s")
    print(f"  modules reparsed   : {len(outcome.reparsed)}")
    print(f"  nodes/rels written : {outcome.nodes} / {outcome.relationships}")
    if elapsed >= budget_s:
        print(
            f"\nRESULT: FAIL — {elapsed:.3f} s >= {budget_s:.3f} s budget",
        )
        return 1
    # Timing a probe module through the pipeline proves the LEG works; it says
    # nothing about whether the destination holds the rest of the tree. Both
    # must hold before this prints PASS.
    if assert_destination_populated(sync._syncer, sync._project, outcome.total_modules):
        return 1
    print("\nRESULT: PASS")
    return 0


def _seed_cache_from_committed_artifact(cache_path: Path) -> bool:
    """Populate an absent sidecar from the committed AST snapshot.

    ``load_or_analyze`` treats a missing freshness map as a full cold
    reparse of the whole tree -- correct, but multi-minute on this
    codebase (~122K LOC), which a service meant to react to a single
    file save within 5s cannot pay on every fresh checkout / deploy.
    ``plans/ast-ultimate.json`` already carries an equivalent snapshot
    (including the ``file_meta`` hash map ``load_or_analyze`` diffs
    against), so a byte-for-byte copy into the sidecar location makes
    the very next pass incremental instead of full. A no-op when the
    sidecar already exists -- including an explicit
    ``--cache plans/ast-ultimate.json`` override, since then *cache_path*
    IS the committed artifact and therefore already exists -- or the
    committed artifact itself is absent.

    Returns True when a seed copy was written.
    """
    if cache_path.exists() or not _COMMITTED_AST_CACHE.exists():
        return False
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    tmp = cache_path.with_suffix(cache_path.suffix + ".seed.tmp")
    with open(_COMMITTED_AST_CACHE, "rb") as src, open(tmp, "wb") as dst:
        dst.write(src.read())
    tmp.replace(cache_path)
    return True


def main(argv: list[str] | None = None, env: Mapping[str, str] | None = None) -> int:
    """Run, measure, or single-pass the live code-intelligence sync."""
    environ = env if env is not None else os.environ
    args = _build_parser(environ).parse_args(argv)
    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s"
    )

    uri = resolve_target_uri(environ, args.uri)
    repo_root = (
        Path(args.repo_root).expanduser().resolve() if args.repo_root else _REPO_ROOT
    )
    if not repo_root.is_dir():
        _logger.error("repo root does not exist: %s", repo_root)
        return 2
    if args.source_dir is None:
        # Refuse, never narrow: an undeclared set defaulting to one tree is
        # how the graph lost five of six roots (see default_roots).
        _logger.error("%s Or pass --source-dir explicitly.", undeclared_message())
        return 2
    if not (repo_root / args.source_dir).is_dir():
        _logger.error(
            "source dir %s does not exist under %s — pass --source-dir",
            args.source_dir,
            repo_root,
        )
        return 2
    # An explicit --extra-root list wins; otherwise the declared roots after
    # the first. Declared roots absent from THIS checkout (vizint-ui, infra on
    # a trimmed deploy) are reported and skipped -- a service that refused to
    # start over an optional tree would be down more often than it is wrong.
    if args.extra_root is not None:
        extra_roots = list(args.extra_root)
        _, absent = existing_roots(extra_roots, repo_root)
        if absent:
            _logger.error("extra root(s) do not exist under %s: %s", repo_root, absent)
            return 2
    else:
        _, declared_extra, _ = default_roots(environ)
        extra_roots, absent = existing_roots(list(declared_extra), repo_root)
        if absent:
            _logger.info("declared roots absent here, skipped: %s", " ".join(absent))
    cache_path = Path(args.cache)
    if not cache_path.is_absolute():
        cache_path = repo_root / cache_path
    # Only for THIS repository: the committed snapshot is orchestrata's own AST, and
    # its cached root would not match another project's source root. LiveCodeSync
    # raises on that mismatch rather than corrupting the graph, but seeding a foreign
    # project's sidecar with it is still wrong, so do not do it at all.
    if repo_root == _REPO_ROOT and _seed_cache_from_committed_artifact(cache_path):
        _logger.info(
            "seeded %s from the committed AST snapshot (cold start)", cache_path
        )

    syncer = _make_syncer(uri, args.database)
    try:
        sync = LiveCodeSync(
            repo_root,
            cache_path,
            args.project,
            syncer,
            source_dir=args.source_dir,
            extra_roots=extra_roots,
        )
        _logger.info("target %s db=%s project=%s", uri, args.database, args.project)
        _logger.info(
            "roots: %s%s",
            args.source_dir,
            "".join(f" {r}" for r in extra_roots),
        )

        if args.measure:
            rc = measure_slo(sync, args.slo_seconds)
            return rc
        if args.once:
            outcome = sync.sync_once()
            _log_outcome(outcome)
            # A single pass is NOT a success just because it raised no error.
            # Verify the destination actually holds the tree — the cache cannot
            # see the graph, and "nothing to do" over an empty graph is the
            # false green this exit path used to print by returning 0 blindly.
            return assert_destination_populated(
                syncer, args.project, outcome.total_modules
            )
        # An unspecified --watch means "watch what we analyze": the primary
        # root plus every extra root actually present. A save under scripts/
        # or tests/ now produces a module, so waking for it is no longer a
        # wasted hash sweep -- it is the point.
        watch_dirs = (
            args.watch if args.watch is not None else [args.source_dir, *extra_roots]
        )
        asyncio.run(sync.watch(watch_dirs, debounce_ms=args.debounce_ms))
        return 0
    finally:
        syncer.close()


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