orchestrata_ultimate.knowledge_graph.file_watcher

Reference › orchestrata_ultimate › knowledge_graph › file_watcher

FileWatcher keeps the operational graph fresh by watching a project directory and handing changes to a synchronizer FileSynchronizer (Spec 008, REQ-008-026). It uses watchfiles.awatch, which has a Rust backend and debounces events, and reads its extensions, ignore patterns, watch interval and debounce from SyncConfig in config. start runs _watch_loop as a background task after a lightweight startup reconciliation (_reconcile_hashes), which walks the project and syncs any file whose hash cache says it is new or changed; stop ends the task. Each batch of events is filtered by extension and ignore pattern in _filter_changes. _process_batch runs a full project sync in a worker thread when the batch exceeds an internal cap, and otherwise _process_individual_changes syncs each file, with _handle_delete cascading a deletion into the graph. The properties running and processed_batches expose its state. The synchronizer is its only importer.

At a glance

Class diagram

Dependency diagram

API

Classes

FileWatcher

Decorators: traces(spec='008', story='US-005', reqs=['REQ-008-026']) · Line 31

Watches a directory for file changes and syncs them.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/file_watcher.py

"""File watcher with debouncing for the knowledge graph.

Uses watchfiles (Rust backend) to detect filesystem changes and
bridges to FileSynchronizer for incremental graph updates.
"""

from __future__ import annotations

import asyncio
import fnmatch
import logging
from pathlib import Path
from typing import TYPE_CHECKING

from watchfiles import Change, awatch

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.config import SyncConfig

if TYPE_CHECKING:
    from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient
    from orchestrata_ultimate.knowledge_graph.synchronizer import FileSynchronizer

logger = logging.getLogger(__name__)

# Maximum batch size before falling back to full project sync
_MAX_BATCH_SIZE = 200


@traces(spec="008", story="US-005", reqs=["REQ-008-026"])
class FileWatcher:
    """Watches a directory for file changes and syncs them.

    Uses watchfiles.awatch() with configurable debounce. Changes
    are batched and dispatched to FileSynchronizer. If a batch
    exceeds _MAX_BATCH_SIZE, a full project sync is triggered
    instead of individual file syncs.

    Args:
        project_path: Root directory to watch.
        synchronizer: FileSynchronizer instance for graph updates.
        graph_client: GraphClient for cascade deletes.
        config: SyncConfig with debounce and ignore settings.
    """

    def __init__(
        self,
        project_path: str,
        synchronizer: FileSynchronizer,
        graph_client: GraphClient | None = None,
        config: SyncConfig | None = None,
    ) -> None:
        self._project_path = project_path
        self._synchronizer = synchronizer
        self._graph_client = graph_client
        self._config = config or SyncConfig()
        self._running = False
        self._task: asyncio.Task | None = None
        self._processed_batches = 0
        self._shutdown_event = asyncio.Event()

    @property
    def running(self) -> bool:
        """Whether the watcher is currently active."""
        return self._running

    @property
    def processed_batches(self) -> int:
        """Number of change batches processed so far."""
        return self._processed_batches

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    async def start(self) -> None:
        """Start the file watcher in a background task.

        Performs a lightweight hash reconciliation on startup,
        then begins watching for filesystem changes.
        """
        if self._running:
            return
        self._running = True
        self._shutdown_event.clear()
        await self._reconcile_hashes()
        self._task = asyncio.create_task(self._watch_loop())
        logger.info("FileWatcher started for %s", self._project_path)

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    async def stop(self) -> None:
        """Gracefully stop the file watcher.

        Sets the shutdown flag and waits for in-flight batches
        to complete before returning.
        """
        if not self._running:
            return
        self._running = False
        self._shutdown_event.set()
        if self._task and not self._task.done():
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                logger.debug("file watcher task cancelled")
        self._task = None
        logger.info("FileWatcher stopped for %s", self._project_path)

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    async def _watch_loop(self) -> None:
        """Main watch loop using watchfiles.awatch().

        Iterates over change sets with debouncing, filters by
        extension and ignore patterns, then dispatches batches.
        """
        debounce_ms = int(self._config.debounce_seconds * 1000)
        try:
            async for changes in awatch(
                self._project_path,
                debounce=debounce_ms,
                stop_event=self._shutdown_event,
            ):
                if not self._running:
                    break
                filtered = self._filter_changes(changes)
                if filtered:
                    await self._process_batch(filtered)
        except asyncio.CancelledError:
            return

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    def _filter_changes(
        self, changes: set[tuple[Change, str]]
    ) -> list[tuple[Change, str]]:
        """Filter changes by extension and ignore patterns.

        Args:
            changes: Raw change set from watchfiles.

        Returns:
            List of (change_type, path) tuples that pass filters.
        """
        result: list[tuple[Change, str]] = []
        extensions = self._config.watch_extensions
        ignore = self._config.ignore_patterns

        for change_type, path_str in changes:
            path = Path(path_str)
            if path.suffix not in extensions:
                continue
            if self._matches_ignore(path, ignore):
                continue
            result.append((change_type, path_str))
        return result

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    def _matches_ignore(self, path: Path, patterns: list[str]) -> bool:
        """Check if a path matches any ignore pattern.

        Args:
            path: File path to check.
            patterns: List of ignore patterns (dir names or globs).

        Returns:
            True if the path is ignored.
        """
        for pattern in patterns:
            if pattern in path.parts:
                return True
            if fnmatch.fnmatch(path.name, pattern):
                return True
        return False

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    async def _process_batch(self, changes: list[tuple[Change, str]]) -> None:
        """Process a batch of file changes.

        If the batch exceeds _MAX_BATCH_SIZE, triggers a full
        project sync. Otherwise processes each change individually.

        Args:
            changes: Filtered list of (change_type, path) tuples.
        """
        if len(changes) > _MAX_BATCH_SIZE:
            logger.info(
                "Batch size %d exceeds cap, running full sync",
                len(changes),
            )
            await asyncio.to_thread(
                self._synchronizer.sync_project,
                self._project_path,
                self._config.watch_extensions,
            )
        else:
            await self._process_individual_changes(changes)
        self._processed_batches += 1

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    async def _process_individual_changes(
        self, changes: list[tuple[Change, str]]
    ) -> None:
        """Process each change individually via synchronizer.

        Handles create/modify via sync_file and delete via
        cascade deletion in the graph.

        Args:
            changes: List of (change_type, path) tuples.
        """
        for change_type, path_str in changes:
            if change_type == Change.deleted:
                await self._handle_delete(path_str)
            else:
                await asyncio.to_thread(self._synchronizer.sync_file, path_str)

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    async def _handle_delete(self, file_path: str) -> None:
        """Handle a file deletion by cascading in the graph.

        Args:
            file_path: Path of the deleted file.
        """
        if self._graph_client is not None:
            # File.path is project-relative and collides across projects
            # (2026-07-12); derive project_id the same way _SyncerBase does
            # (basename of the watched root) so the cascade delete cannot
            # touch another project's File node.
            project_id = Path(self._project_path.rstrip("/")).name
            deleted = await self._graph_client.delete_file_cascade(
                file_path, project_id
            )
            logger.info(
                "Cascade-deleted %d nodes for %s",
                deleted,
                file_path,
            )
        else:
            logger.warning(
                "No graph client — cannot cascade-delete %s",
                file_path,
            )

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    async def _reconcile_hashes(self) -> None:
        """Lightweight startup reconciliation.

        Walks the project directory and checks each file against
        the synchronizer's hash cache. Files that differ or are
        new get synced.
        """
        project = Path(self._project_path)
        if not project.is_dir():
            return

        extensions = self._config.watch_extensions
        ignore = self._config.ignore_patterns
        files_to_sync: list[str] = []

        for source_file in project.rglob("*"):
            if source_file.suffix not in extensions:
                continue
            if self._matches_ignore(source_file, ignore):
                continue
            if self._synchronizer.needs_sync(str(source_file)):
                files_to_sync.append(str(source_file))

        if not files_to_sync:
            return

        logger.info("Reconciliation: %d files need sync", len(files_to_sync))
        for file_path in files_to_sync:
            await asyncio.to_thread(self._synchronizer.sync_file, file_path)