Watch watch_path for new or modified files and auto-update the graph. For code-only changes: re-runs AST extraction + rebuild immediately (no LLM). For doc/paper/image changes: writes a needs_update flag and notifies the user to run /graphify --update (LLM extraction required).
(watch_path: Path, debounce: float = 3.0)
| 1105 | |
| 1106 | |
| 1107 | def watch(watch_path: Path, debounce: float = 3.0) -> None: |
| 1108 | """ |
| 1109 | Watch watch_path for new or modified files and auto-update the graph. |
| 1110 | |
| 1111 | For code-only changes: re-runs AST extraction + rebuild immediately (no LLM). |
| 1112 | For doc/paper/image changes: writes a needs_update flag and notifies the user |
| 1113 | to run /graphify --update (LLM extraction required). |
| 1114 | |
| 1115 | debounce: seconds to wait after the last change before triggering (avoids |
| 1116 | running on every keystroke when many files are saved at once). |
| 1117 | """ |
| 1118 | try: |
| 1119 | from watchdog.observers import Observer |
| 1120 | from watchdog.observers.polling import PollingObserver |
| 1121 | from watchdog.events import FileSystemEventHandler |
| 1122 | except ImportError as e: |
| 1123 | raise ImportError("watchdog not installed. Run: pip install watchdog") from e |
| 1124 | |
| 1125 | last_trigger: float = 0.0 |
| 1126 | pending: bool = False |
| 1127 | changed: set[Path] = set() |
| 1128 | |
| 1129 | # Load .graphifyignore patterns ONCE at startup so the handler does not |
| 1130 | # re-parse the file on every filesystem event. Watchdog's handler runs on |
| 1131 | # the observer thread and is invoked for every event the OS delivers |
| 1132 | # (Time Machine writes, Docker/Colima VM I/O, Spotlight indexing, …) — |
| 1133 | # without this short-circuit a busy volume can saturate a CPU core |
| 1134 | # discarding events one extension at a time. (gh-928) |
| 1135 | watch_root_for_ignore = watch_path.resolve() |
| 1136 | ignore_patterns = _load_graphifyignore(watch_root_for_ignore) |
| 1137 | |
| 1138 | class Handler(FileSystemEventHandler): |
| 1139 | def on_any_event(self, event): |
| 1140 | nonlocal last_trigger, pending |
| 1141 | if event.is_directory: |
| 1142 | return |
| 1143 | path = Path(os.fsdecode(event.src_path)) |
| 1144 | # Check .graphifyignore BEFORE the extension/dotfile/out filters so |
| 1145 | # the cheapest short-circuit for users with broad ignore patterns |
| 1146 | # (node_modules/, .venv/, build/, …) fires first. _is_ignored |
| 1147 | # tolerates absolute paths outside watch_root via its internal |
| 1148 | # relative_to guard, so a stray symlinked event won't raise. |
| 1149 | if ignore_patterns and _is_ignored(path, watch_root_for_ignore, ignore_patterns): |
| 1150 | return |
| 1151 | if path.suffix.lower() not in _WATCHED_EXTENSIONS: |
| 1152 | return |
| 1153 | try: |
| 1154 | filter_parts = path.relative_to(watch_root_for_ignore).parts |
| 1155 | except ValueError: |
| 1156 | filter_parts = path.parts |
| 1157 | if any(part.startswith(".") for part in filter_parts): |
| 1158 | return |
| 1159 | if _GRAPHIFY_OUT in filter_parts: |
| 1160 | return |
| 1161 | last_trigger = time.monotonic() |
| 1162 | pending = True |
| 1163 | changed.add(path) |
| 1164 |