orchestrata_ultimate.refactor.codemod.watcher

Reference › orchestrata_ultimate › refactor › codemod › watcher

An inotify file watcher that tells the daemon when Python files change so it can reload its workspace. _daemon_entry is its only importer.

FileWatcher.watch(repo_root, callback) loads libc's inotify functions through ctypes (_load_inotify), creates a non-blocking inotify descriptor (raising OSError if that fails), adds a watch on every directory that contains Python files (_walk_python_dirs, _add_watches), and starts a daemon thread running _watch_loop. The loop uses select with a one-second timeout so that stop is noticed promptly, reads events with _read_changed, and coalesces events that arrive in a short window so each changed path is reported once to the callback. stop ends the thread and closes the descriptor, and is_running reports the state. Without an inotify library the watcher holds a placeholder descriptor and reports no changes.

At a glance

Class diagram

Dependency diagram

API

Classes

FileWatcher

Decorators: traces(spec='110', story='US-110-013', reqs=['REQ-110-136']) · Line 54

Inotify-based file watcher for incremental workspace reload.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/watcher.py

"""SPEC-110 codemod engine — inotify file watcher for incremental reload.

Watches for file changes using inotify and triggers incremental re-parse
of affected files, updating the CIG and CSR index.
"""

from __future__ import annotations

import ctypes
import ctypes.util
import logging
import os
import select
import struct
import threading
import time
from collections.abc import Callable
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)

IN_MODIFY = 0x00000002
IN_CREATE = 0x00000100
IN_DELETE = 0x00000200
IN_MOVED_FROM = 0x00000040
IN_MOVED_TO = 0x00000080
IN_CLOSE_WRITE = 0x00000008
IN_MASK = (
    IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVED_FROM | IN_MOVED_TO | IN_CLOSE_WRITE
)

#: Quiet window that makes one logical save produce one callback: a single
#: write_text() is IN_MODIFY + IN_CLOSE_WRITE (+ IN_CREATE when new), and the
#: kernel may split those events across reads — without the drain, one save
#: fired several callbacks, the last carrying duplicate paths (measured
#: 2026-09-10: [mod.py, mod.py] 1.5s after the write).
_COALESCE_SEC = 0.1

EVENT_STRUCT = "iIII"
EVENT_SIZE = struct.calcsize(EVENT_STRUCT)


def _load_inotify() -> ctypes.CDLL | None:
    """Load libc for inotify syscalls."""
    libc_name = ctypes.util.find_library("c")
    if libc_name is None:
        return None
    return ctypes.CDLL(libc_name, use_errno=True)


@traces(spec="110", story="US-110-013", reqs=["REQ-110-136"])
class FileWatcher:
    """Inotify-based file watcher for incremental workspace reload.

    Watches a repository root for Python file changes and triggers
    incremental re-parse via a callback.
    """

    def __init__(self) -> None:
        """Initialize FileWatcher with no active watches."""
        self._fd: int = -1
        self._thread: threading.Thread | None = None
        self._stop_event = threading.Event()
        self._watches: dict[int, Path] = {}
        self._libc = _load_inotify()
        self._callback: Callable[[list[Path]], None] | None = None

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-136"])
    def watch(self, repo_root: Path, callback: Callable[[list[Path]], None]) -> None:
        """Start watching repo_root for Python file changes.

        Args:
            repo_root: Repository root directory to watch.
            callback: Called with list of changed file paths on change events.

        Raises:
            OSError: If inotify initialization fails.
        """
        self._callback = callback
        self._stop_event.clear()

        if self._libc is not None:
            self._fd = self._libc.inotify_init1(os.O_NONBLOCK)
            if self._fd < 0:
                # Older libc without inotify_init1: plain init + set_blocking.
                self._fd = self._libc.inotify_init()
                if self._fd < 0:
                    raise OSError("Failed to initialize inotify")
                os.set_blocking(self._fd, False)
        else:
            self._fd = os.open("/dev/null", os.O_RDONLY)

        self._add_watches(repo_root)

        self._thread = threading.Thread(
            target=self._watch_loop,
            name="codemod-file-watcher",
            daemon=True,
        )
        self._thread.start()
        logger.info("FileWatcher started for %s", repo_root)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-136"])
    def stop(self) -> None:
        """Stop watching for file changes.

        Sets the stop event and joins the watcher thread first — the
        non-blocking select loop wakes within ~1s — then removes watches
        and closes the fd.
        """
        self._stop_event.set()

        if self._thread is not None and self._thread.is_alive():
            self._thread.join(timeout=5.0)
        self._thread = None

        if self._fd >= 0:
            for wd in list(self._watches):
                if self._libc is not None:
                    self._libc.inotify_rm_watch(self._fd, wd)
                self._watches.pop(wd, None)
            try:
                os.close(self._fd)
            except OSError as e:
                logger.warning("Error closing inotify fd: %s", e)
            self._fd = -1

        logger.info("FileWatcher stopped")

    @property
    def is_running(self) -> bool:
        """Whether the watcher thread is active."""
        return self._thread is not None and self._thread.is_alive()

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-136"])
    def _add_watches(self, repo_root: Path) -> None:
        """Add inotify watches for all directories under repo_root."""
        for dirpath in _walk_python_dirs(repo_root):
            self._add_watch(dirpath)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-136"])
    def _add_watch(self, directory: Path) -> None:
        """Add a single inotify watch on a directory."""
        if self._libc is None or self._fd < 0:
            return
        dir_bytes = str(directory).encode("utf-8")
        wd = self._libc.inotify_add_watch(self._fd, dir_bytes, IN_MASK)
        if wd >= 0:
            self._watches[wd] = directory

    def _read_changed(self, buf_size: int) -> list[Path] | None:
        """One non-blocking read -> the new .py paths in it (deduped,
        order-stable). ``[]`` when nothing readable, ``None`` when the fd is
        done (closed under ``stop()`` or a read error)."""
        try:
            raw = os.read(self._fd, buf_size)
        except BlockingIOError:
            return []
        except OSError:
            if self._stop_event.is_set():
                return None
            logger.error("inotify read error", exc_info=True)
            return None
        if not raw:
            return None

        changed: list[Path] = []
        seen: set[Path] = set()
        offset = 0
        while offset < len(raw):
            if offset + EVENT_SIZE > len(raw):
                break
            wd, mask, _cookie, name_len = struct.unpack_from(EVENT_STRUCT, raw, offset)
            offset += EVENT_SIZE
            if name_len > 0:
                name_bytes = raw[offset : offset + name_len]
                offset += name_len
                name = name_bytes.rstrip(b"\x00").decode("utf-8", errors="replace")
            else:
                name = ""

            if not name.endswith(".py"):
                continue

            directory = self._watches.get(wd)
            if directory is not None:
                path = directory / name
                if path not in seen:
                    seen.add(path)
                    changed.append(path)
        return changed

    def _watch_loop(self) -> None:
        """Main inotify event loop running in a background thread.

        The fd is non-blocking; select with a 1s timeout lets the loop
        notice _stop_event promptly (a plain blocking read would not
        wake when stop() closes the fd — Linux close() does not
        interrupt a blocked read on another thread).
        """
        buf_size = 4096
        while not self._stop_event.is_set():
            try:
                readable, _, _ = select.select([self._fd], [], [], 1.0)
            except OSError, ValueError:
                if self._stop_event.is_set():
                    break
                logger.error("inotify select error", exc_info=True)
                break

            if not readable:
                continue

            paths = self._read_changed(buf_size)
            if paths is None:
                break
            if not paths:
                continue

            # Coalesce split events (see _COALESCE_SEC): drain until the fd
            # stays quiet for the window, then report each path once.
            deadline = time.monotonic() + _COALESCE_SEC
            while True:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
                try:
                    more, _, _ = select.select([self._fd], [], [], remaining)
                except OSError, ValueError:
                    break
                if not more:
                    break
                extra = self._read_changed(buf_size)
                if extra is None:
                    break
                known = set(paths)
                paths.extend(p for p in extra if p not in known)

            if self._callback is not None:
                try:
                    self._callback(paths)
                except Exception:
                    logger.error("FileWatcher callback error", exc_info=True)


@traces(spec="110", story="US-110-013", reqs=["REQ-110-136"])
def _walk_python_dirs(root: Path) -> list[Path]:
    """Collect all directories under root that contain .py files."""
    dirs: list[Path] = []
    skip = {".git", "__pycache__", ".venv", "node_modules", ".codemod_daemon"}
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in skip]
        dp = Path(dirpath)
        if any(f.endswith(".py") for f in filenames):
            dirs.append(dp)
    return dirs