scripts.ast_daemon

Reference › scripts › ast_daemon

A per-repository daemon that keeps the AST toolkit warm in memory and answers read-only queries over a Unix domain socket, so that repeated toolkit calls skip the cold parse. It builds on the HTTP routes in ast_api and optionally runs the snapshot publisher in ast_daemon_publish. The toolkit client calls the daemon first and falls back to a cold run when it is absent. A human or a service manager starts it; nothing imports it.

Invocation: main parses arguments and calls serve. Options include --repo-root, --socket (relative to the repo root), --source (what to watch and warm), --cache and --no-cache for the sha1 warm-cache sidecar, --idle-timeout and --idle-poll-interval (default 30 seconds), --watch-debounce-ms, --watch-rust-timeout-ms, --publish, --snapshot-root (repeatable; implies publish), --snapshot-name (default orchestrata-ultimate) and --snapshot-output (default plans/ast-ultimate.json).

_prepare validates the repo root, cleans up a stale socket and pid file (_pid_alive, _cleanup_stale), takes a lock file and builds a DaemonState. It records which deployed commit the daemon runs (_capture_identity). The server exposes GET /health, which returns counters, and POST /query, which dispatches one command to the warm tree and returns the result with a generation number. An unknown command gives 400 and a handler failure gives 500, so it fails loudly rather than returning a wrong answer. A watch loop invalidates changed files and bumps the generation; an idle loop shuts the server down after the idle timeout. A /publish route is installed when publishing is enabled. A lock serializes the toolkit routes.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

PublishRequest (pydantic)

Bases: BaseModel · Line 98

One /publish request (ADR-0097).

QueryRequest (pydantic)

Bases: BaseModel · Line 197

One /query request: a dispatch command name + its keyword params.

DaemonState (dataclass)

Decorators: dataclasses.dataclass · Line 230

All per-daemon-instance mutable state, threaded explicitly rather than kept as module globals — so multiple instances (one per test, all in one pytest process) never bleed into each other.

_Daemon (dataclass)

Decorators: dataclasses.dataclass · Line 491

A fully-built, not-yet-running daemon: state + FastAPI routes + a uvicorn Server + its two background threads (constructed but not started).

Functions

Source

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

#!/usr/bin/env python3
"""Per-repo AST daemon — a warm, in-memory process for the read-only AST
toolkit subcommands (symbols/refs/source/who-imports/find-cycles/call-graph/
complexity).

WHY: scripts/serialize_codebase_ast.py's per-invocation warm cache (the
content-sha1-keyed pickle sidecar — see its "Warm-cache read path" docstring
block, added alongside this daemon) only bought ~1.25-1.6x: interpreter
startup + the Python-level tree walk dominate a COLD process regardless of
how little re-parsing it does (measured; see docs/ast-toolkit.md). The
order-of-magnitude win requires that walk to happen ONCE, in a process that
stays warm across queries. That's this daemon: a resident FastAPI app
(scripts/ast_api.py, imported UNMODIFIED — this is a thin wrapper around it,
not a fork) served over a per-repo Unix domain socket, with its own file
watcher for freshness.

CORRECTNESS NEVER DEPENDS ON THIS PROCESS. Every query this daemon answers
calls the exact same functions scripts/serialize_codebase_ast.py would call
locally (via scripts/ast_api.py's route functions) — the daemon adds no
logic of its own beyond dispatch, freshness, and lifecycle. The CLI
(``query_ast_daemon`` in serialize_codebase_ast.py) falls back to a local,
authoritative computation silently whenever this daemon is absent,
unhealthy, or errors mid-query. This is a deliberate departure from the
janus-swi/free-threading doctrine ("never silently fall back") — THAT patch
is load-bearing because there is no fallback path; THIS daemon is pure
acceleration with one, so silent fallback is correct here.

LIFECYCLE: spawned by .claude/hooks/start-ast-daemon.sh on SessionStart
(detached, never blocking session start), self-exits after an idle period
with no queries (default 2h), and shuts down cleanly on SIGTERM/SIGINT. An
``flock`` on "<socket>.lock" makes concurrent spawns from multiple Claude
sessions safe — exactly one daemon instance ever ends up serving a given
socket; a loser exits(0) within milliseconds.

Run directly (normally the SessionStart hook does this for you):
    uv run python scripts/ast_daemon.py --socket .claude/run/ast-daemon.sock
"""

from __future__ import annotations

import argparse
import dataclasses
import fcntl
import functools
import importlib.util
import os
import re
import signal
import sys
import threading
import time
from collections.abc import Callable
from pathlib import Path
from typing import Any

import uvicorn
import watchfiles
from fastapi import HTTPException
from fastapi.routing import APIRoute
from pydantic import BaseModel

_HERE = Path(__file__).resolve().parent
_AST_API_PATH = _HERE / "ast_api.py"
_PUBLISH_PATH = _HERE / "ast_daemon_publish.py"

_DEFAULT_SOURCE = "src/orchestrata_ultimate"
_DEFAULT_SOCKET = ".claude/run/ast-daemon.sock"
# Untracked sidecar (see .gitignore) -- never the committed
# plans/ast-ultimate.json artifact. That file is a CLI-only snapshot
# read by scripts/gates.py's ast-refresh step and the Prolog fact
# base; a live, always-on daemon writing derived cache state onto it
# kept it permanently dirty and churned it in unrelated commits
# (context.md backlog #284, DESIGN_WAIVER 2026-08-09).
_DEFAULT_CACHE = ".ast_daemon/ast-daemon-cache.json"
_CODE_ROOT = Path(__file__).resolve().parents[1]  # the directory containing scripts/
_COMMIT_LINE = re.compile(r"^commit ([0-9a-f]{7,40})$")
_UNLOCKED_PATHS = frozenset({"/health", "/healthz"})
_DEFAULT_IDLE_TIMEOUT = 7200.0  # 2h — brief item 5
_DEFAULT_IDLE_POLL_INTERVAL = 30.0
_DEFAULT_WATCH_DEBOUNCE_MS = 300
_DEFAULT_WATCH_RUST_TIMEOUT_MS = 500


def _load_publish_module() -> Any:
    """The snapshot-publisher module (scripts/ast_daemon_publish.py), loaded
    by path like ast_api so the daemon works whether it runs as a script, from
    the deployed lib/scripts package, or under a test's spec load."""
    spec = importlib.util.spec_from_file_location(
        "ast_daemon_publish_loaded", _PUBLISH_PATH
    )
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load ast_daemon_publish from {_PUBLISH_PATH}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


class PublishRequest(BaseModel):
    """One ``/publish`` request (ADR-0097)."""

    since: float | None = None  # edited file's mtime: no-op when already covered
    wait: bool = False  # block until the covering publish finished
    timeout: float = 120.0


def _load_ast_api() -> Any:
    """Load a FRESH scripts/ast_api.py module instance.

    Deliberately NOT registered in ``sys.modules`` — mirrors ast_api.py's own
    ``_load_tool()``. Every call gets an isolated module object with its own
    ``_tool`` (and that tool's own ``_PARSE_CACHE``/``_CACHE_STATE``
    globals), so several daemon instances spawned in one process — as the
    tests do, one per function-scoped fixture — never bleed into each other.
    """
    spec = importlib.util.spec_from_file_location("ast_api", _AST_API_PATH)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load ast_api from {_AST_API_PATH}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _read_deployed_commit(version_file: Path) -> str | None:
    """The sha on the ``commit <sha>`` line of *version_file* (the code
    root's ``VERSION``), or ``None`` when the file is absent (a checkout
    run, not a deployment)."""
    try:
        text = version_file.read_text(encoding="utf-8")
    except OSError:
        return None
    for line in text.splitlines():
        match = _COMMIT_LINE.match(line.strip())
        if match:
            return match.group(1)
    return None


def _find_deployed_commit(code_root: Path) -> str | None:
    """The deployed commit for *code_root*: ``<code_root>/VERSION``, then
    ``<code_root>/../VERSION`` (install-tools.sh stamps the install prefix,
    one level above the ``lib`` code root); the first file that yields a commit
    wins, so a stray unstamped VERSION falls through to the next location."""
    for version_file in (code_root / "VERSION", code_root.parent / "VERSION"):
        commit = _read_deployed_commit(version_file)
        if commit is not None:
            return commit
    return None


def _capture_identity(repo_root: Path, code_root: Path) -> dict[str, Any]:
    """The daemon's identity, read ONCE at start and reported verbatim by
    ``/health`` for the life of the process (a VERSION rewritten under a
    running daemon does not change what that daemon is)."""
    return {
        "repo_root": str(repo_root),
        "code_root": str(code_root),
        "deployed_commit": _find_deployed_commit(code_root),
        "pid": os.getpid(),
    }


def _check_repo_root(raw: str | Path | None) -> Path:
    """Resolve ``--repo-root`` (the cwd when the flag is absent) and refuse a
    path that is not an existing directory, before anything is created."""
    try:
        root = Path(raw).resolve() if raw is not None else Path.cwd().resolve()
    except FileNotFoundError:
        print(
            "ast-daemon: --repo-root not given and the current directory no "
            "longer exists — exiting.",
            file=sys.stderr,
        )
        raise SystemExit(2) from None
    if not root.is_dir():
        print(
            f"ast-daemon: --repo-root {root} is not an existing directory — exiting.",
            file=sys.stderr,
        )
        raise SystemExit(2)
    return root


def _under(repo_root: Path, path: str | Path) -> Path:
    """*path* anchored on *repo_root* when relative, untouched when absolute."""
    p = Path(path)
    return p if p.is_absolute() else repo_root / p


def _sources(source: str) -> list[str]:
    return [s.strip() for s in source.split(",") if s.strip()]


def _param_source(params: dict[str, Any]) -> str:
    return params.get("source") or _DEFAULT_SOURCE


class QueryRequest(BaseModel):
    """One ``/query`` request: a dispatch command name + its keyword params."""

    cmd: str
    params: dict[str, Any] = {}


# Dispatch table for the /query passthrough — every handler calls the SAME
# scripts/ast_api.py route function the plain HTTP GET endpoints call, so a
# delegated result is byte-identical (same functions, same inputs) to what
# the CLI would compute locally.
_QUERY_DISPATCH: dict[str, Callable[[dict, Any], Any]] = {
    "symbols": lambda p, api: api.symbols(
        source=_param_source(p), pattern=p.get("pattern"), kind=p.get("kind")
    ),
    "refs": lambda p, api: api.refs(name=p["name"], source=_param_source(p)),
    "source": lambda p, api: api.source(
        name=p["name"], source=_param_source(p), residue=p.get("residue", False)
    ),
    "who-imports": lambda p, api: api.who_imports(
        target=p["target"], source=_param_source(p)
    ),
    "find-cycles": lambda p, api: api.cycles(source=_param_source(p)),
    "call-graph": lambda p, api: api.call_graph(
        func_name=p["name"], source=_param_source(p)
    ),
    "complexity": lambda p, api: api.complexity(
        source=_param_source(p), threshold=p.get("threshold", 10)
    ),
}


@dataclasses.dataclass
class DaemonState:
    """All per-daemon-instance mutable state, threaded explicitly rather
    than kept as module globals — so multiple instances (one per test, all
    in one pytest process) never bleed into each other."""

    api_module: Any
    sources: list[str]
    repo_root: Path | None = None
    # ADR-0097: the single-writer snapshot publisher; None when publishing is off.
    publisher: Any = None
    # Captured once at start (see _capture_identity); never re-read.
    identity: dict[str, Any] = dataclasses.field(default_factory=dict)
    generation: int = 0
    queries: int = 0
    start_time: float = dataclasses.field(default_factory=time.monotonic)
    last_activity: float = dataclasses.field(default_factory=time.monotonic)
    counters_lock: threading.Lock = dataclasses.field(default_factory=threading.Lock)
    # Serializes ALL cache-mutating access — both /query dispatch and the
    # watcher's invalidation — because the shared parse/derived-extraction
    # caches in serialize_codebase_ast.py (_PARSE_CACHE / _CACHE_STATE) have
    # no locking of their own (out of scope for this brief; that module is
    # owned by the warm-cache wave). Trades a little request parallelism for
    # correctness under free-threading — the right call for a single local
    # CLI client, not a meaningful bottleneck.
    cache_lock: threading.Lock = dataclasses.field(default_factory=threading.Lock)
    stop_event: threading.Event = dataclasses.field(default_factory=threading.Event)

    @property
    def tool(self) -> Any:
        """The loaded ``serialize_codebase_ast`` module behind the API."""
        return self.api_module._tool

    def touch(self) -> None:
        """Record query activity (idle-timeout clock + query counter)."""
        with self.counters_lock:
            self.last_activity = time.monotonic()
            self.queries += 1

    def bump_generation(self) -> None:
        """Increment the cache generation (a watched file changed)."""
        with self.counters_lock:
            self.generation += 1

    def snapshot(self) -> dict[str, Any]:
        """The ``/health`` payload: generation, query count, cache size,
        uptime, and the daemon's identity (which checkout it serves, which
        code it runs, the deployed commit, its pid)."""
        with self.counters_lock:
            generation, queries = self.generation, self.queries
        return {
            "generation": generation,
            "queries": queries,
            "modules": len(self.tool._PARSE_CACHE),
            "uptime": time.monotonic() - self.start_time,
            "publish": (
                self.publisher.status() if self.publisher is not None else None
            ),
            **self.identity,
        }

    def idle_seconds(self) -> float:
        """Seconds since the last query (drives the idle self-exit)."""
        with self.counters_lock:
            return time.monotonic() - self.last_activity


def _invalidate_file(tool_mod: Any, path_str: str) -> None:
    """Drop *path_str*'s cache entries and eagerly re-derive them.

    Correctness never required this — ``_parse_cached``/``_cached_extract``
    in serialize_codebase_ast.py already content-sha1-validate on every
    access and self-heal on a mismatch — but re-deriving eagerly (instead of
    waiting for the next query to hit the resulting miss) is what keeps the
    daemon actually WARM immediately after an edit, which is the point of
    running it.
    """
    path = Path(path_str)
    key = str(path.resolve())
    tool_mod._PARSE_CACHE.pop(key, None)
    if tool_mod._CACHE_STATE.get("enabled"):
        try:
            tool_mod._load_sidecar().pop(key, None)
        except Exception:
            pass
    if path.suffix == ".py" and path.exists():
        try:
            tool_mod._parse_cached(path)
            tool_mod.list_symbols([str(path)])
        except Exception:
            pass


def _watch_loop(state: DaemonState, *, debounce_ms: int, rust_timeout_ms: int) -> None:
    """Background freshness watcher.

    Uses ``watchfiles`` (already an ``orchestrata_ultimate`` dependency,
    Rust-backed change detection — see ``knowledge_graph/file_watcher.py``
    for the project's other consumer) directly via its sync ``watch()``
    generator, rather than a hand-rolled mtime-poll loop: event-driven
    (near-instant on Linux inotify) beats a fixed interval, and it reuses a
    library the codebase already depends on. Deliberately NOT the
    knowledge_graph.file_watcher.FileWatcher CLASS: that one is hard-wired
    to FileSynchronizer/GraphClient for graph-sync semantics this daemon has
    no use for, and knowledge_graph/** is out of scope for this brief — this
    calls the same underlying primitive standalone instead.
    """
    try:
        roots = [
            str(_under(state.repo_root, s)) if state.repo_root else s
            for s in state.sources
        ]
        if state.publisher is not None:
            roots += [p for p in state.publisher.watch_paths() if p not in roots]
        for changes in watchfiles.watch(
            # Absolute, so the watch never depends on the process cwd.
            *roots,
            debounce=debounce_ms,
            rust_timeout=rust_timeout_ms,
            yield_on_timeout=True,
            stop_event=state.stop_event,
            raise_interrupt=False,
        ):
            if state.stop_event.is_set():
                return
            py_paths = {path for _change, path in changes if path.endswith(".py")}
            if not py_paths:
                continue
            with state.cache_lock:
                for path in py_paths:
                    _invalidate_file(state.tool, path)
            state.bump_generation()
            if state.publisher is not None:
                state.publisher.request()
    except Exception:
        # Best-effort acceleration thread: a crash here must never take the
        # daemon down. Queries keep working correctly regardless (the
        # sha1-based staleness check is the actual correctness backstop) —
        # just without this thread's eager pre-warm.
        return


def _idle_loop(
    state: DaemonState,
    server: uvicorn.Server,
    *,
    idle_timeout: float,
    poll_interval: float,
) -> None:
    while not state.stop_event.is_set():
        if state.idle_seconds() >= idle_timeout:
            server.should_exit = True
            return
        state.stop_event.wait(poll_interval)


def _lock_toolkit_routes(app: Any, lock: threading.Lock) -> None:
    """Run every mounted ast_api GET handler under *lock* — the same lock
    ``/query`` and the watcher's invalidation take — so no handler observes a
    half-invalidated cache. The handlers are synchronous (threadpool), so the
    wait never blocks the event loop and ``/health`` stays reachable."""
    for route in app.routes:
        if not isinstance(route, APIRoute) or "GET" not in route.methods:
            continue
        if route.path in _UNLOCKED_PATHS:  # liveness probes never wait
            continue
        original = route.dependant.call
        if original is None:
            continue

        @functools.wraps(original)
        def locked(*args: Any, __original: Any = original, **kwargs: Any) -> Any:
            with lock:
                return __original(*args, **kwargs)

        route.dependant.call = locked


def _install_routes(app: Any, state: DaemonState) -> None:
    _lock_toolkit_routes(app, state.cache_lock)

    @app.get("/health", tags=["daemon"])
    def health() -> dict[str, Any]:
        """Liveness probe + counters — the client's 50ms pre-flight check."""
        return state.snapshot()

    @app.post("/query", tags=["daemon"])
    def query(body: QueryRequest) -> dict[str, Any]:
        """Dispatch one read-only AST query against the warm in-memory tree."""
        handler = _QUERY_DISPATCH.get(body.cmd)
        if handler is None:
            raise HTTPException(status_code=400, detail=f"unknown cmd {body.cmd!r}")
        state.touch()
        try:
            with state.cache_lock:
                result = handler(body.params, state.api_module)
        except Exception as exc:  # fail LOUDLY (500) — never silently wrong
            raise HTTPException(status_code=500, detail=str(exc)) from exc
        with state.counters_lock:
            generation = state.generation
        return {"result": result, "generation": generation}


def _install_publish_route(app: Any, state: DaemonState) -> None:
    def publish(body: PublishRequest) -> dict[str, Any]:
        """ADR-0097: have the daemon (re)publish the snapshot from its warm
        cache. Coalesces with the watcher's own publishes; with ``since`` set it
        is a no-op when the last publish already read the tree after that mtime."""
        publisher = state.publisher
        if publisher is None:
            raise HTTPException(status_code=409, detail="snapshot publishing is off")
        ticket = publisher.request(body.since)
        done = publisher.wait(ticket, body.timeout) if body.wait else ticket is None
        return {"ticket": ticket, "done": done, "publish": publisher.status()}

    app.add_api_route("/publish", publish, methods=["POST"], tags=["daemon"])


def _pid_alive(pid: int) -> bool:
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True  # exists, owned by someone else — treat as alive
    except OSError:
        return False
    return True


def _cleanup_stale(socket_path: Path, pid_path: Path) -> None:
    """Remove a leftover socket/pidfile from a daemon that died without
    cleaning up after itself (e.g. SIGKILL)."""
    stale = True
    if pid_path.exists():
        try:
            pid = int(pid_path.read_text(encoding="utf-8").strip())
            stale = not _pid_alive(pid)
        except ValueError, OSError:
            stale = True
    if stale:
        socket_path.unlink(missing_ok=True)
        pid_path.unlink(missing_ok=True)


def _cleanup(socket_path: Path, pid_path: Path, lock_file: Any) -> None:
    for p in (socket_path, pid_path):
        try:
            p.unlink(missing_ok=True)
        except OSError:
            pass
    try:
        fcntl.flock(lock_file, fcntl.LOCK_UN)
    except OSError:
        pass
    try:
        lock_file.close()
    except OSError:
        pass


@dataclasses.dataclass
class _Daemon:
    """A fully-built, not-yet-running daemon: state + FastAPI routes + a
    uvicorn ``Server`` + its two background threads (constructed but not
    started). Split out of ``serve()`` so tests can drive the pieces
    individually — start the server in a background thread, assert against
    it, then stop it deterministically — instead of only being able to run
    the whole blocking lifecycle end to end."""

    state: DaemonState
    server: uvicorn.Server
    watch_thread: threading.Thread
    idle_thread: threading.Thread
    lock_file: Any
    socket_path: Path
    pid_path: Path

    def start_background_threads(self) -> None:
        """Start the file-watcher and idle-timeout threads."""
        self.watch_thread.start()
        self.idle_thread.start()
        if self.state.publisher is not None:
            self.state.publisher.start()

    def cleanup(self) -> None:
        """Stop threads and remove the socket/pid/lock files on shutdown."""
        self.state.stop_event.set()
        if self.state.publisher is not None:
            self.state.publisher.stop()
        # ``ident is None`` => never started (e.g. a caller that never
        # reached start_background_threads(), such as a losing _prepare()
        # racing another instance's lock) — nothing to join.
        if self.watch_thread.ident is not None:
            self.watch_thread.join(timeout=2.0)
        if self.idle_thread.ident is not None:
            self.idle_thread.join(timeout=2.0)
        _cleanup(self.socket_path, self.pid_path, self.lock_file)


def _build_publisher(args: argparse.Namespace, repo_root: Path, tool: Any) -> Any:
    """The snapshot publisher (ADR-0097), or ``None`` when publishing is off:
    it is opt-in (``--publish`` or an explicit ``--snapshot-root``) so a bare
    query daemon -- every test fixture, any non-orchestrata checkout -- never
    writes a snapshot it was not asked to own."""
    if not (args.publish or args.snapshot_root):
        return None
    publish = _load_publish_module()
    roots = list(args.snapshot_root) or publish.declared_roots(repo_root)
    if not roots:
        print(
            "ast-daemon: --publish given but no snapshot roots declared "
            "($ORCHESTRATA_AST_ROOTS / env/ast-roots.env) — publishing off.",
            file=sys.stderr,
        )
        return None
    return publish.SnapshotPublisher(
        tool,
        repo_root,
        name=args.snapshot_name,
        roots=roots,
        output=args.snapshot_output,
    )


def _prepare(args: argparse.Namespace) -> _Daemon | None:
    """Acquire the lock, warm-start, and build everything ``serve()`` needs
    — but start nothing that blocks. Returns ``None`` when another instance
    already holds the lock on this socket: a clean, successful no-op (the
    goal — a warm daemon serving this socket — is already satisfied by
    whichever instance got there first)."""
    repo_root = _check_repo_root(args.repo_root)
    socket_path = _under(repo_root, args.socket).resolve()
    run_dir = socket_path.parent
    run_dir.mkdir(parents=True, exist_ok=True)
    try:
        os.chmod(run_dir, 0o700)
    except OSError:
        pass

    pid_path = socket_path.with_name(socket_path.name + ".pid")
    lock_path = socket_path.with_name(socket_path.name + ".lock")

    _cleanup_stale(socket_path, pid_path)

    lock_file = open(lock_path, "w", encoding="utf-8")
    try:
        fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        print(
            f"ast-daemon: another instance holds the lock on {lock_path} — exiting.",
            file=sys.stderr,
        )
        lock_file.close()
        return None

    pid_path.write_text(str(os.getpid()), encoding="utf-8")

    api_module = _load_ast_api()
    # The served checkout is data (ADR-0067): relative sources resolve
    # against repo_root, never the process cwd.
    api_module._tool.configure_repo_root(repo_root)
    api_module._tool.configure_parse_cache(
        _under(repo_root, args.cache), enabled=not args.no_cache
    )

    sources = _sources(args.source)
    try:
        api_module._tool.list_symbols(sources)  # warm start
        # Warm the refs_index derived cache too: find_references computes it
        # for EVERY file as a side effect, so one call here means the first
        # refs query never pays the cold walk inside a request (measured
        # 2026-08-11: 17s landing inside an innocent test instead).
        api_module._tool.find_references("__warm__", sources)
    except Exception as exc:
        print(
            f"ast-daemon: warm-start failed (continuing cold): {exc}", file=sys.stderr
        )

    state = DaemonState(
        api_module=api_module,
        sources=sources,
        repo_root=repo_root,
        identity=_capture_identity(repo_root, _CODE_ROOT),
    )
    state.publisher = _build_publisher(args, repo_root, api_module._tool)
    _install_routes(api_module.app, state)
    _install_publish_route(api_module.app, state)

    watch_thread = threading.Thread(
        target=_watch_loop,
        args=(state,),
        kwargs={
            "debounce_ms": args.watch_debounce_ms,
            "rust_timeout_ms": args.watch_rust_timeout_ms,
        },
        daemon=True,
        name="ast-daemon-watch",
    )

    config = uvicorn.Config(
        api_module.app,
        uds=str(socket_path),
        log_level="warning",
        # SPEC-023: httptools/uvloop are C extensions that silently
        # re-enable the GIL process-wide the moment they're imported (the
        # same failure mode the project vendors free-threaded grpcio to
        # avoid). Pin the pure-Python h11 protocol + stdlib asyncio loop so
        # this daemon's own background threads (watcher, idle check, query
        # handling) stay genuinely parallel. WebSockets are never used here.
        http="h11",
        loop="asyncio",
        ws="none",
        lifespan="off",
    )
    server = uvicorn.Server(config)

    idle_thread = threading.Thread(
        target=_idle_loop,
        args=(state, server),
        kwargs={
            "idle_timeout": args.idle_timeout,
            "poll_interval": args.idle_poll_interval,
        },
        daemon=True,
        name="ast-daemon-idle",
    )

    return _Daemon(
        state=state,
        server=server,
        watch_thread=watch_thread,
        idle_thread=idle_thread,
        lock_file=lock_file,
        socket_path=socket_path,
        pid_path=pid_path,
    )


def serve(args: argparse.Namespace) -> int:
    """Run the daemon to completion (blocks until shutdown). Returns the
    process exit code — 0 in every case, including "another instance is
    already running" (that is success, not failure: the goal — a warm
    daemon on this socket — is satisfied either way)."""
    daemon = _prepare(args)
    if daemon is None:
        return 0
    daemon.start_background_threads()

    # uvicorn's own SIGTERM/SIGINT handling (Server.capture_signals) does a
    # graceful async shutdown and then RE-RAISES the terminating signal with
    # its ORIGINAL (default) disposition once done — so systemd/process
    # supervisors see a proper signal-based exit code. That re-raise kills
    # the process immediately, which would skip the `finally` below
    # entirely (pidfile/socket/lock never get removed). Installing a no-op
    # handler BEFORE server.run() makes IT the "original" handler uvicorn
    # captures and restores just before that final re-raise, so the
    # re-raised signal is swallowed harmlessly and control returns here as
    # normal. Main-thread only, exactly like uvicorn's own guard — tests
    # that run the server in a background thread need no such handling
    # (SIGTERM there would hit the whole test process, never done in tests;
    # see tests/scripts/test_ast_daemon.py's real-subprocess SIGTERM test).
    on_main_thread = threading.current_thread() is threading.main_thread()
    prev_term = prev_int = None
    if on_main_thread:

        def _noop(signum: int, frame: object) -> None:
            pass

        prev_term = signal.signal(signal.SIGTERM, _noop)
        prev_int = signal.signal(signal.SIGINT, _noop)
    try:
        daemon.server.run()  # blocks; returns on should_exit (idle) or SIGTERM/SIGINT
    finally:
        if on_main_thread:
            signal.signal(signal.SIGTERM, prev_term)
            signal.signal(signal.SIGINT, prev_int)
        daemon.cleanup()
    return 0


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Warm, resident AST-toolkit daemon over a Unix domain socket."
    )
    parser.add_argument(
        "--repo-root",
        type=lambda raw: Path(raw).resolve(),
        default=None,
        help=(
            "Checkout to serve (must exist); relative --source/--cache/--socket "
            "resolve against it (default: current working directory)"
        ),
    )
    parser.add_argument(
        "--socket",
        default=_DEFAULT_SOCKET,
        help=f"UDS path, relative to --repo-root (default: {_DEFAULT_SOCKET})",
    )
    parser.add_argument(
        "--source",
        default=_DEFAULT_SOURCE,
        help="Source dir(s)/file(s) to watch + warm",
    )
    parser.add_argument(
        "--cache", default=_DEFAULT_CACHE, help="Warm-cache JSON sidecar path"
    )
    parser.add_argument(
        "--no-cache",
        action="store_true",
        help="Disable the sha1 warm cache (always cold)",
    )
    parser.add_argument(
        "--idle-timeout",
        type=float,
        default=_DEFAULT_IDLE_TIMEOUT,
        help=(
            "Self-exit after this many idle seconds with no queries "
            "(default: 7200 = 2h)"
        ),
    )
    parser.add_argument(
        "--idle-poll-interval",
        type=float,
        default=_DEFAULT_IDLE_POLL_INTERVAL,
        help="How often to check idle time, in seconds (default: 30)",
    )
    parser.add_argument(
        "--watch-debounce-ms", type=int, default=_DEFAULT_WATCH_DEBOUNCE_MS
    )
    parser.add_argument(
        "--watch-rust-timeout-ms", type=int, default=_DEFAULT_WATCH_RUST_TIMEOUT_MS
    )
    parser.add_argument(
        "--publish",
        action="store_true",
        help=(
            "ADR-0097: publish the snapshot (--snapshot-output) from the warm "
            "cache after every debounced source change and on POST /publish; "
            "roots come from $ORCHESTRATA_AST_ROOTS or env/ast-roots.env"
        ),
    )
    parser.add_argument(
        "--snapshot-root",
        action="append",
        default=[],
        help="Snapshot source root (repeatable; first is primary); implies --publish",
    )
    parser.add_argument("--snapshot-name", default="orchestrata-ultimate")
    parser.add_argument("--snapshot-output", default="plans/ast-ultimate.json")
    return parser


def main() -> None:
    """CLI entry point: parse args and run the daemon until exit."""
    args = _build_parser().parse_args()
    sys.exit(serve(args))


if __name__ == "__main__":
    main()