"""

This module implements the live file-watching functionality using the `watchdog` library.

It observes directories for changes and triggers updates to the code graph.

"""

import os

import threading

from pathlib import Path

import typing

from watchdog.observers import Observer

from watchdog.observers.polling import PollingObserver

from watchdog.events import FileSystemEventHandler

if typing.TYPE_CHECKING:

from pathspec import PathSpec

from ..tools.graph_builder import GraphBuilder

from ..core.jobs import JobManager

from .cgcignore import build_ignore_spec

from ..tools.indexing.constants import DEFAULT_IGNORE_PATTERNS

from ..cli.config_manager import get_config_value

from ..utils.debug_log import debug_log, info_logger, error_logger, warning_logger

POLLING_ENV_VAR = "CGC_WATCH_POLLING"

TRUE_ENV_VALUES = {"1", "true", "yes", "on"}

def should_use_polling_observer(use_polling: typing.Optional[bool] = None) -> bool:

if use_polling is not None:

return use_polling

return os.getenv(POLLING_ENV_VAR, "").strip().lower() in TRUE_ENV_VALUES

class RepositoryEventHandler(FileSystemEventHandler):

def __init__(

self,

graph_builder: "GraphBuilder",

repo_path: Path,

debounce_interval=2.0,

perform_initial_scan: bool = True,

cgcignore_path: str = None,

ignore_spec: "PathSpec" = None,

sync_on_start: bool = False,

):

super().__init__()

self.graph_builder = graph_builder

self.repo_path = repo_path.resolve()

self.debounce_interval = debounce_interval

self.timers = {}

# Guards self.timers, and serialises the graph updates that the

# per-path debounce timers would otherwise run concurrently.

self._timers_lock = threading.Lock()

self._update_lock = threading.RLock()

self.ignore_root = self.repo_path

self.ignore_spec = ignore_spec

self.cgcignore_path = cgcignore_path

self._load_ignore_spec(cgcignore_path)

self.all_file_data = []

self.imports_map = {}

if sync_on_start:

self.synchronize_with_disk()

elif perform_initial_scan:

self._initial_scan()

def _load_ignore_spec(self, cgcignore_path: str = None) -> None:

if self.ignore_spec is not None:

return

try:

self.ignore_spec, resolved = build_ignore_spec(

ignore_root=self.ignore_root,

default_patterns=DEFAULT_IGNORE_PATTERNS,

explicit_path=cgcignore_path,

)

if resolved:

debug_log(f"Watcher using ignore file: {resolved}")

except OSError as e:

self.ignore_spec = None

warning_logger(f"Could not load ignore rules: {e}")

def _should_ignore(self, path: str | Path) -> bool:

path_obj = Path(path).resolve()

ignore_root = getattr(self, "ignore_root", self.repo_path)

ignore_dirs_str = get_config_value("IGNORE_DIRS") or ""

if ignore_dirs_str:

ignore_dirs = {d.strip().lower() for d in ignore_dirs_str.split(",") if d.strip()}

try:

parts = {p.lower() for p in path_obj.relative_to(ignore_root).parent.parts}

if parts.intersection(ignore_dirs):

return True

except ValueError:

pass

ignore_spec = getattr(self, "ignore_spec", None)

if not ignore_spec:

return False

try:

rel = path_obj.relative_to(ignore_root).as_posix()

except ValueError:

return False

return ignore_spec.match_file(rel)

def _is_supported_code_file(self, path: str | Path) -> bool:

path_obj = Path(path)

return (

path_obj.is_file()

and path_obj.suffix in self.graph_builder.parsers

and not self._should_ignore(path_obj)

)

def _iter_supported_files(self) -> list[Path]:

from ..tools.indexing.discovery import discover_files_to_index

supported = self.graph_builder.parsers.keys()

# Forward the explicit ignore file and re-apply this handler's own spec.

# Without them, discovery fell back to the repo-local .cgcignore, so the

# initial scan and disk sync indexed files that later change events

# would skip — the graph kept entries the user asked to exclude, and

# they went stale on edit.

files, _ = discover_files_to_index(

self.repo_path,

cgcignore_path=getattr(self, "cgcignore_path", None),

supported_extensions=set(supported),

)

return [f for f in files if not self._should_ignore(f)]

def _initial_scan(self):

info_logger(f"Initial scan: {self.repo_path}")

all_files = self._iter_supported_files()

self.imports_map = self.graph_builder.pre_scan_imports(all_files)

for f in all_files:

parsed = self.graph_builder.parse_file(self.repo_path, f)

if "error" not in parsed:

self.all_file_data.append(parsed)

repo_name = self.repo_path.name

repo_path_str = self.repo_path.resolve().as_posix()

self.graph_builder.add_repository_to_graph(self.repo_path, is_dependency=False)

for fd in self.all_file_data:

self.graph_builder.add_file_to_graph(

fd, repo_name, self.imports_map, repo_path_str=repo_path_str

)

self.graph_builder.link_function_calls(self.all_file_data, self.imports_map)

self.graph_builder.link_inheritance(self.all_file_data, self.imports_map)

self.all_file_data.clear()

info_logger("Initial scan complete")

def synchronize_with_disk(self) -> None:

info_logger(f"Syncing: {self.repo_path}")

current_files = self._iter_supported_files()

current_paths = {p.resolve().as_posix() for p in current_files}

indexed = self.graph_builder.get_repo_file_paths(self.repo_path)

self.imports_map = self.graph_builder.pre_scan_imports(current_files)

for stale in indexed - current_paths:

self.graph_builder.delete_file_from_graph(stale)

refreshed = []

refreshed_paths: list[str] = []

for p in current_files:

fd = self.graph_builder.update_file_in_graph(

p, self.repo_path, self.imports_map

)

if fd and "error" not in fd:

refreshed.append(fd)

refreshed_paths.append(p.resolve().as_posix())

if refreshed_paths:

# Only clear edges originating from the files we touched — do not

# wipe the entire repo call graph like delete_relationship_links().

self.graph_builder.delete_outgoing_calls_from_files(refreshed_paths)

self.graph_builder.delete_inherits_for_files(refreshed_paths)

self.graph_builder.link_function_calls(refreshed, self.imports_map)

self.graph_builder.link_inheritance(refreshed, self.imports_map)

info_logger("Sync complete")

def _debounce(self, event_path, action):

# Timers are keyed per path, so N files changed inside the debounce

# window fire N handler threads concurrently. Those handlers do

# read-modify-write on the shared imports_map and interleave

# delete/add/delete_outgoing_calls for overlapping caller sets, so one

# can delete edges another just created. A branch switch or `git pull`

# is the normal trigger. _handle_modification now takes _update_lock.

def _run():

# Drop the fired timer: entries were never removed, so self.timers

# grew without bound for the life of the watcher.

with self._timers_lock:

if self.timers.get(event_path) is timer:

del self.timers[event_path]

action()

with self._timers_lock:

existing = self.timers.get(event_path)

if existing is not None:

existing.cancel()

timer = threading.Timer(self.debounce_interval, _run)

self.timers[event_path] = timer

timer.start()

def cancel_timers(self):

with self._timers_lock:

for t in self.timers.values():

t.cancel()

self.timers.clear()

def _update_imports_map_for_file(self, changed_path: Path):

"""Re-scan a single file and merge its contributions into self.imports_map."""

changed_str = str(changed_path.resolve())

for symbol in list(self.imports_map.keys()):

old_list = self.imports_map[symbol]

if changed_str in old_list:

new_list = [p for p in old_list if p != changed_str]

if new_list:

self.imports_map[symbol] = new_list

else:

del self.imports_map[symbol]

if changed_path.exists():

new_map = self.graph_builder.pre_scan_imports([changed_path])

for symbol, paths in new_map.items():

if symbol not in self.imports_map:

self.imports_map[symbol] = []

self.imports_map[symbol].extend(paths)

def _handle_modification(self, event_path_str: str):

"""Incremental update: re-parse and re-link only the changed file and its neighbours."""

# Serialised: concurrent handlers previously did read-modify-write on

# the shared imports_map (lost updates) and interleaved

# delete_file_from_graph / add_file_to_graph /

# delete_outgoing_calls_from_files for overlapping caller sets, so one

# could delete edges another had just created.

with self._update_lock:

self._handle_modification_locked(event_path_str)

def _handle_modification_locked(self, event_path_str: str):

info_logger(f"File change detected (incremental update): {event_path_str}")

changed_path = Path(event_path_str)

if self._should_ignore(changed_path):

debug_log(f"Ignored watcher update based on .cgcignore: {changed_path}")

return

changed_path_str = changed_path.resolve().as_posix()

supported_extensions = self.graph_builder.parsers.keys()

caller_paths = {

p

for p in self.graph_builder.get_caller_file_paths(changed_path_str)

if p and not self._should_ignore(p)

}

inheritor_paths = {

p

for p in self.graph_builder.get_inheritance_neighbor_paths(changed_path_str)

if p and not self._should_ignore(p)

}

affected_paths = {changed_path_str} | caller_paths | inheritor_paths

info_logger(

f"[INCREMENTAL] affected={len(affected_paths)} files "

f"(callers={len(caller_paths)}, inheritors={len(inheritor_paths)})"

)

self._update_imports_map_for_file(changed_path)

self.graph_builder.update_file_in_graph(changed_path, self.repo_path, self.imports_map)

# Every file in affected_paths is re-parsed below and fed back into

# link_function_calls, so every one of them needs its outgoing CALLS

# cleared first. Clearing only caller_paths left the inheritance-only

# neighbours to have their edges re-created on top of the existing ones

# — and on Neo4j/Nornic the writer uses CREATE, not MERGE, so duplicate

# CALLS multiplied on every save. (FalkorDB and Kùzu use MERGE, which is

# why this never showed up there.) The changed file itself is excluded:

# update_file_in_graph above already deleted and rebuilt it.

other_callers = list(affected_paths - {changed_path_str})

other_inheritors = list(inheritor_paths)

if other_callers:

self.graph_builder.delete_outgoing_calls_from_files(other_callers)

if other_inheritors:

self.graph_builder.delete_inherits_for_files(other_inheritors)

subset_file_data = []

for path_str in affected_paths:

p = Path(path_str)

if p.exists() and p.suffix in supported_extensions and not self._should_ignore(p):

parsed = self.graph_builder.parse_file(self.repo_path, p)

if "error" not in parsed:

subset_file_data.append(parsed)

file_class_lookup = self.graph_builder.get_repo_class_lookup(self.repo_path)

info_logger(f"[INCREMENTAL] Re-linking {len(subset_file_data)} files...")

self.graph_builder.link_function_calls(

subset_file_data, self.imports_map, file_class_lookup

)

self.graph_builder.link_inheritance(subset_file_data, self.imports_map)

try:

from codegraphcontext.cli.config_manager import get_config_value as _gcv

_vector_enabled = (_gcv("ENABLE_VECTOR_RESOLVE") or "false").lower() == "true"

_inherit_enabled = (_gcv("ENABLE_INHERIT_RESOLVE") or "false").lower() == "true"

except Exception as _cfg_e:

warning_logger(f"[PHASE4/5] Could not read config flags: {_cfg_e}")

_vector_enabled = False

_inherit_enabled = False

if _vector_enabled:

try:

from codegraphcontext.tools.indexing.embeddings import EmbeddingPipeline

embed_pipeline = EmbeddingPipeline(self.graph_builder.driver)

embed_pipeline.invalidate_for_file(changed_path_str)

embed_pipeline.run(str(self.repo_path))

info_logger(f"[EMBED] Incremental embedding complete for {changed_path_str}")

except Exception as _e:

warning_logger(f"[EMBED] Incremental embedding failed: {_e}")

if _inherit_enabled:

try:

from codegraphcontext.tools.indexing.resolution.post_resolution import run_inheritance_reresolve

_vector_resolver = None

if _vector_enabled:

try:

from codegraphcontext.tools.indexing.vector_resolver import VectorResolver

_vector_resolver = VectorResolver(self.graph_builder.driver)

except Exception as _ve:

warning_logger(f"[VECTOR] Resolver unavailable for watcher: {_ve}")

n_improved = run_inheritance_reresolve(

self.graph_builder.driver, str(self.repo_path), _vector_resolver

)

info_logger(f"[INHERIT-RESOLVE] Incremental: {n_improved} edges improved")

except Exception as _e:

warning_logger(f"[INHERIT-RESOLVE] Incremental failed: {_e}")

info_logger(f"[INCREMENTAL] Done. Graph refresh for {event_path_str} complete! ✅")

def on_created(self, event):

if not event.is_directory and self._is_supported_code_file(event.src_path):

self._debounce(event.src_path, lambda: self._handle_modification(event.src_path))

def on_modified(self, event):

if not event.is_directory and self._is_supported_code_file(event.src_path):

self._debounce(event.src_path, lambda: self._handle_modification(event.src_path))

def on_deleted(self, event):

if not event.is_directory:

self._debounce(event.src_path, lambda: self._handle_modification(event.src_path))

def on_moved(self, event):

if event.is_directory:

return

# Both endpoints matter. Only dest_path was handled, so the node for

# the *old* path and all of its symbols stayed in the graph forever:

# every rename duplicated every symbol in the file, and a normal

# refactoring session or a `git checkout` between branches accumulated

# them indefinitely until find_callers started returning dead paths.

src_path = getattr(event, "src_path", None)

if src_path:

self._debounce(src_path, lambda: self._handle_removal(src_path))

self._debounce(event.dest_path, lambda: self._handle_modification(event.dest_path))

def _handle_removal(self, path_str):

"""Drop a path that no longer exists (the source side of a rename)."""

with self._update_lock:

try:

self.graph_builder.delete_file_from_graph(str(Path(path_str).resolve()))

info_logger(f"[WATCH] removed stale node for moved file: {path_str}")

except Exception as exc: # noqa: BLE001 - a watcher must not die on one file

error_logger(f"[WATCH] failed to remove {path_str}: {exc}")

class CodeWatcher:

def __init__(

self,

graph_builder: "GraphBuilder",

job_manager="JobManager",

use_polling: typing.Optional[bool] = None,

):

self.graph_builder = graph_builder

observer_cls = PollingObserver if should_use_polling_observer(use_polling) else Observer

self.observer = observer_cls()

self.watched_paths = set()

self.watches = {}

self.handlers = {}

def watch_directory(

self,

path: str,

perform_initial_scan: bool = True,

cgcignore_path: str = None,

sync_on_start: bool = False,

):

path_obj = Path(path).resolve()

path_str = str(path_obj)

if path_str in self.watched_paths:

return {"message": "Already watching"}

handler = RepositoryEventHandler(

self.graph_builder,

path_obj,

perform_initial_scan=perform_initial_scan,

sync_on_start=sync_on_start,

cgcignore_path=cgcignore_path,

)

watch = self.observer.schedule(handler, path_str, recursive=True)

self.watches[path_str] = watch

self.handlers[path_str] = handler

self.watched_paths.add(path_str)

return {"message": f"Watching {path_str}"}

def unwatch_directory(self, path: str):

path_str = str(Path(path).resolve())

handler = self.handlers.pop(path_str, None)

if handler:

handler.cancel_timers()

watch = self.watches.pop(path_str, None)

if watch:

self.observer.unschedule(watch)

self.watched_paths.discard(path_str)

return {"message": f"Stopped watching {path_str}"}

def list_watched_paths(self):

return list(self.watched_paths)

def start(self):

if not self.observer.is_alive():

self.observer.start()

def stop(self):

for h in self.handlers.values():

h.cancel_timers()

self.handlers.clear()

if self.observer.is_alive():

self.observer.stop()

self.observer.join()