orchestrata_ultimate.refactor.codemod._daemon_entry

Reference › orchestrata_ultimate › refactor › codemod › _daemon_entry

The process entry point of the codemod daemon, started by daemon through the deployed launcher. main(repo_root, socket_path, pid_file) configures logging, loads a CodemodWorkspace over the src, scripts and tests roots in parallel with a parse cache under .codemod_daemon/parse_cache, creates a CodemodServer with an identity and starts a FileWatcher whose callback reloads the workspace and swaps it into the server. The swap is dropped if a commit wrote to disk during the reload. A SIGTERM handler requests a clean stop.

build_identity returns the fields reported by the server's status reply (ADR-0067: identity, not health): the repo root, the code root, the deployed commit, the process id and whether the native codemod_fast extension loaded, so a deployed daemon can be told apart from one running from a checkout. read_deployed_commit and find_deployed_commit read the commit from the deployed VERSION file.

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/refactor/codemod/_daemon_entry.py

"""SPEC-110 codemod daemon entry point.

Launched by CodemodDaemon.start() as a background process.
Parses the workspace, starts the server and file watcher, and runs
until SIGTERM is received.
"""

from __future__ import annotations

import logging
import os
import re
import signal
import sys
from pathlib import Path

logger = logging.getLogger(__name__)

_COMMIT_LINE = re.compile(r"^commit ([0-9a-f]{7,40})$")


def read_deployed_commit(version_file: Path) -> str | None:
    """Return the sha on the ``commit <sha>`` line of a deploy VERSION file.

    Args:
        version_file: Path to ``<code_root>/VERSION`` (written by the deploy).

    Returns:
        The sha, or None when the file is absent, unreadable, or carries no
        ``commit <sha>`` line (a checkout, not a deployed copy).
    """
    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 build_identity(repo_root: Path, code_root: Path | None = None) -> dict:
    """Identity fields reported in the ``status`` reply (ADR-0067: identity, not health).

    ``repo_root`` is the data the daemon serves; ``code_root`` is the directory
    holding the running ``orchestrata_ultimate`` package, so a deployed copy
    under /opt/orchestrata-tools is told apart from a checkout.
    ``codemod_fast_native`` is True when the C extension loaded and False when
    the pure-Python fallback is serving, so the deploy smoke can refuse a
    silently degraded install.
    """
    if code_root is None:
        import orchestrata_ultimate

        code_root = Path(orchestrata_ultimate.__file__).resolve().parents[1]
    import codemod_fast

    return {
        "repo_root": str(repo_root.resolve()),
        "code_root": str(code_root),
        "deployed_commit": find_deployed_commit(code_root),
        "pid": os.getpid(),
        "codemod_fast_native": bool(getattr(codemod_fast, "_HAS_NATIVE", False)),
    }


def main(repo_root: Path, socket_path: Path, pid_file: Path) -> None:
    """Run the daemon process.

    Args:
        repo_root: Repository root to parse.
        socket_path: Unix socket path for JSON-RPC.
        pid_file: PID file path (written by parent, validated here).
    """
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(name)s %(levelname)s %(message)s",
    )

    from orchestrata_ultimate.refactor.codemod.server import CodemodServer
    from orchestrata_ultimate.refactor.codemod.watcher import FileWatcher
    from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

    # Content-addressed parse cache lives next to the daemon state so
    # restarts and incremental reloads skip re-parsing unchanged files.
    cache_dir = repo_root / ".codemod_daemon" / "parse_cache"

    logger.info("Daemon starting: parsing workspace at %s", repo_root)
    workspace = CodemodWorkspace.load(
        repo_root,
        include=("src", "scripts", "tests"),
        parallel=True,
        cache_dir=cache_dir,
    )
    logger.info(
        "Workspace loaded: %d files, %.1fs parse time",
        workspace.stats().file_count,
        workspace.stats().parse_time_seconds,
    )

    server = CodemodServer(identity=build_identity(repo_root))
    watcher = FileWatcher()

    def _on_file_change(changed_paths: list[Path]) -> None:
        """Incremental reload triggered by inotify."""
        logger.info("File change detected: %d files", len(changed_paths))
        try:
            # Read before the load: a commit that writes during the load makes
            # the loaded tree older than the disk, and the swap is dropped.
            generation = server.commit_generation
            new_ws = CodemodWorkspace.load(
                repo_root,
                include=("src", "scripts", "tests"),
                parallel=True,
                cache_dir=cache_dir,
            )
            if server.replace_workspace(new_ws, generation):
                logger.info("Incremental reload complete")
            else:
                logger.info("Incremental reload dropped: a commit wrote meanwhile")
        except Exception:
            logger.error("Incremental reload failed", exc_info=True)

    shutdown_requested = False

    def _handle_sigterm(signum: int, frame: object) -> None:
        nonlocal shutdown_requested
        shutdown_requested = True
        # Signal-handler-safe only: no logging, no locks, no thread joins —
        # a logging call here can deadlock if the signal interrupted a
        # thread holding the logging lock. watcher.stop() runs in the
        # finally below after serve() unwinds.
        server.request_stop()

    signal.signal(signal.SIGTERM, _handle_sigterm)

    watcher.watch(repo_root, _on_file_change)

    logger.info("Daemon ready, serving on %s", socket_path)
    try:
        server.serve(socket_path, workspace)
    except Exception:
        logger.error("Server error", exc_info=True)
    finally:
        watcher.stop()
        server.stop()
        logger.info("Daemon exiting")


if __name__ == "__main__":
    if len(sys.argv) != 4:
        print(
            "Usage: _daemon_entry <repo_root> <socket_path> <pid_file>", file=sys.stderr
        )
        sys.exit(1)

    main(
        repo_root=Path(sys.argv[1]),
        socket_path=Path(sys.argv[2]),
        pid_file=Path(sys.argv[3]),
    )