scripts.ast_daemon_publish

Reference › scripts › ast_daemon_publish

The snapshot publisher of the resident AST daemon (ADR-0097). It is a library module, not a command; ast_daemon builds a SnapshotPublisher and exposes it through its publish route when publishing is enabled.

SnapshotPublisher is constructed with the toolkit object, the repo root, a snapshot name, the source roots and an output path (for example plans/ast-ultimate.json). It runs a worker thread (start, stop). Callers ask for a publish with request, which returns a ticket, and may block on wait(ticket, timeout); status reports its state. watch_paths lists the files the daemon should watch for it. The worker loop _run calls _publish_once, which analyses the roots through the warm tool, merges runtime call-graph edges when available (_runtime_edges, _callgraph_module), assembles the snapshot document in _build and writes it to the output file. declared_roots reads the roots a repository declares, and _sha1_file hashes files to detect change.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

SnapshotPublisher

Line 68

Publishes output for the checkout at repo_root from a warm daemon.

Functions

Source

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

"""The resident AST daemon's snapshot publisher (ADR-0097).

``scripts/ast_daemon.py`` holds the parsed tree warm and serves queries; this
module is the half that PUBLISHES the on-disk snapshot (``plans/ast-<name>.json``)
from that process, so the post-edit hooks stop re-serializing the whole tree on
every write.

SINGLE WRITER. Exactly one thread (``_run``) ever writes the artifact, so a
burst of edits, ``/publish`` requests and watcher events coalesce into the
fewest possible writes: while one publish runs, any number of new requests
collapse into ONE follow-up publish. The write itself is the serializer's own
``_atomic_write_json`` (PID-unique temp + ``os.replace``), the same writer
``serialize_codebase_ast.py serialize`` uses, over the same
``load_or_analyze`` / ``serialize`` / ``compute_file_meta`` calls
``_cmd_serialize`` makes -- so the artifact is byte-compatible with a CLI
``serialize --incremental`` of the same tree.

RUNTIME CALLS. A bare serialize drops every function's ``runtime_calls`` (the
key is written by ``build_runtime_callgraph.py``, not the AST serializer). When
``plans/runtime-callgraph.json`` exists, this publisher re-applies its edges to
the document BEFORE the atomic write -- exactly what ``gates.py``'s reaugment
step does afterwards -- so a daemon publish never zeroes runtime_calls. With no
capture file the key is not written, matching the CLI byte for byte.
"""

from __future__ import annotations

import hashlib
import importlib.util
import json
import os
import sys
import threading
import time
from pathlib import Path
from typing import Any

_HERE = Path(__file__).resolve().parent
_ROOTS_ENV = "ORCHESTRATA_AST_ROOTS"
_ROOTS_FILE = "env/ast-roots.env"
_RUNTIME_EDGES = "plans/runtime-callgraph.json"


def declared_roots(repo_root: Path) -> list[str]:
    """The snapshot's declared root set: ``$ORCHESTRATA_AST_ROOTS``, else the
    value in ``<repo>/env/ast-roots.env`` (the file ``.envrc`` loads; a systemd
    unit has no direnv), else ``[]`` (publishing stays off)."""
    raw = os.environ.get(_ROOTS_ENV, "")
    if not raw.strip():
        try:
            text = (repo_root / _ROOTS_FILE).read_text(encoding="utf-8")
        except OSError:
            text = ""
        for line in text.splitlines():
            key, sep, value = line.strip().partition("=")
            if sep and key.strip() == _ROOTS_ENV:
                raw = value.strip().strip("'\"")
    return [tok for tok in raw.replace(":", " ").split() if tok]


def _sha1_file(path: Path) -> str | None:
    try:
        return hashlib.sha1(path.read_bytes()).hexdigest()
    except OSError:
        return None


class SnapshotPublisher:
    """Publishes ``output`` for the checkout at ``repo_root`` from a warm daemon."""

    def __init__(
        self,
        tool: Any,
        repo_root: Path,
        *,
        name: str,
        roots: list[str],
        output: str,
    ) -> None:
        self._tool = tool
        self._repo_root = repo_root
        self._name = name
        self._roots = list(roots)
        self._output = output
        self._cond = threading.Condition()
        self._stop = threading.Event()
        self._thread = threading.Thread(
            target=self._run, daemon=True, name="ast-daemon-publish"
        )
        self._requested = 0
        self._done = 0
        self._started_at: float | None = None
        self._status: dict[str, Any] = {
            "enabled": True,
            "output": str(self.output_path),
            "published_at": None,
            "finished_at": None,
            "content_hash": None,
            "publish_count": 0,
            "writes": 0,
            "last_ms": None,
            "last_error": None,
        }
        self._edges_cache: tuple[int, list[tuple[str, str]]] | None = None
        self._callgraph: Any = None

    @property
    def output_path(self) -> Path:
        out = Path(self._output)
        return out if out.is_absolute() else self._repo_root / out

    def watch_paths(self) -> list[str]:
        """Absolute snapshot roots that exist (the watcher covers them all)."""
        paths = (
            Path(r) if Path(r).is_absolute() else self._repo_root / r
            for r in self._roots
        )
        return [str(p) for p in paths if p.exists()]

    def start(self) -> None:
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        with self._cond:
            self._cond.notify_all()
        if self._thread.ident is not None:
            self._thread.join(timeout=5.0)

    def status(self) -> dict[str, Any]:
        with self._cond:
            return {**self._status, "pending": self._requested > self._done}

    def request(self, since: float | None = None) -> int | None:
        """Ask for a publish; returns a ticket for :meth:`wait`, or ``None``
        when the last publish already read the tree after ``since`` (the edited
        file's mtime) and nothing is pending -- nothing to do."""
        with self._cond:
            if (
                since is not None
                and self._started_at is not None
                and self._started_at >= since
                and self._requested == self._done
            ):
                return None
            self._requested += 1
            self._cond.notify_all()
            return self._requested

    def wait(self, ticket: int | None, timeout: float) -> bool:
        """Block until the publish that covers ``ticket`` finished."""
        if ticket is None:
            return True
        with self._cond:
            return self._cond.wait_for(lambda: self._done >= ticket, timeout=timeout)

    # -- the single writer ---------------------------------------------------

    def _run(self) -> None:
        seed = _sha1_file(self.output_path)
        with self._cond:
            self._status["content_hash"] = seed
        while not self._stop.is_set():
            with self._cond:
                self._cond.wait_for(
                    lambda: self._requested > self._done or self._stop.is_set(),
                    timeout=1.0,
                )
                if self._stop.is_set():
                    return
                if self._requested == self._done:
                    continue
                target = self._requested
            error: str | None = None
            try:
                self._publish_once()
            except Exception as exc:  # recorded, never fatal: the daemon keeps serving
                error = f"{type(exc).__name__}: {exc}"
                print(f"ast-daemon: publish failed: {error}", file=sys.stderr)
            with self._cond:
                self._status["last_error"] = error
                self._done = target
                self._cond.notify_all()

    def _runtime_edges(self) -> list[tuple[str, str]] | None:
        path = self._repo_root / _RUNTIME_EDGES
        try:
            mtime = path.stat().st_mtime_ns
        except OSError:
            return None
        if self._edges_cache is not None and self._edges_cache[0] == mtime:
            return self._edges_cache[1]
        summary = json.loads(path.read_text(encoding="utf-8"))
        edges = [(pair[0], pair[1]) for pair in summary["edges"]]
        self._edges_cache = (mtime, edges)
        return edges

    def _callgraph_module(self) -> Any:
        if self._callgraph is None:
            spec = importlib.util.spec_from_file_location(
                "ast_daemon_runtime_callgraph", _HERE / "build_runtime_callgraph.py"
            )
            if spec is None or spec.loader is None:
                raise ImportError("cannot load build_runtime_callgraph")
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            self._callgraph = module
        return self._callgraph

    def _publish_once(self) -> None:
        tool = self._tool
        started = time.time()
        t0 = time.monotonic()
        output = self.output_path
        wrote = False
        # Never touch the process working directory: it is process-global,
        # and a publish that moves it overlapping any other thread moving it
        # leaves the whole daemon in the wrong directory for good —
        # verify_deployment_identity.py then failed the instance mid-publish
        # and install-orchestrata.sh rolled back (review findings D9/Q10,
        # observed 2026-10-13). load_or_analyze takes the repo root
        # explicitly and anchors every path against it, so the artifact
        # stays byte-identical to a CLI run from the repository root.
        sources = [
            Path(root) if Path(root).is_absolute() else self._repo_root / root
            for root in self._roots
        ]
        source, extra_roots = sources[0], tuple(sources[1:])
        result, stats = tool.load_or_analyze(
            self._name,
            source,
            output,
            monorepo=False,
            extra_roots=extra_roots,
            root=self._repo_root,
        )
        unchanged = (
            not stats["fallback"]
            and not stats["reparsed"]
            and not stats["removed"]
            and output.exists()
        )
        if unchanged:
            os.utime(output)  # content identical; the mtime is the freshness pin
        else:
            serialized = self._build(result, stats)
            output.parent.mkdir(parents=True, exist_ok=True)
            tool._atomic_write_json(output, serialized, default=str)
            wrote = True
        digest = _sha1_file(output) if wrote else self._status["content_hash"]
        if digest is None:
            digest = _sha1_file(output)
        with self._cond:
            self._started_at = started
            self._status.update(
                published_at=started,
                finished_at=time.time(),
                content_hash=digest,
                last_ms=round((time.monotonic() - t0) * 1000),
                publish_count=self._status["publish_count"] + 1,
                writes=self._status["writes"] + (1 if wrote else 0),
            )

    def _build(self, result: Any, stats: dict[str, Any]) -> dict[str, Any]:
        """Mirror ``_cmd_serialize``'s assembly step, then re-apply runtime edges."""
        tool = self._tool
        serialized: dict[str, Any]
        trees = stats.get("trees")
        if not stats["fallback"]:
            base = tool.serialize(result)
            cached_modules = stats["cached_modules"]
            reparsed = set(stats["reparsed"])
            out_modules = []
            for mod_dict in base["modules"]:
                mpath = mod_dict["path"]
                if mpath not in reparsed and mpath in cached_modules:
                    out_modules.append(cached_modules[mpath])
                else:
                    single: dict[str, Any] = {"modules": [mod_dict]}
                    tool._augment_end_line_and_complexity(single, trees)
                    out_modules.append(single["modules"][0])
            base["modules"] = out_modules
            serialized = base
        else:
            serialized = tool._augment_end_line_and_complexity(
                tool.serialize(result), trees
            )
        serialized["file_meta"] = tool.compute_file_meta(
            result.modules, root=self._repo_root
        )
        edges = self._runtime_edges()
        if edges is not None:
            cg = self._callgraph_module()
            _, func_index = cg.collect_ast_nodes(serialized)
            cg.augment_ast(func_index, edges)
        return serialized