From 762ff45dcda329c1bb256863a39eb7ca0ea0ef06 Mon Sep 17 00:00:00 2001 From: Ziqi Lin <760448737@qq.com> Date: Sat, 10 Oct 2026 11:29:38 -0400 Subject: [PATCH 1/2] fix(mcp): run post-processing during auto-watch updates --- code_review_graph/incremental.py | 9 +- code_review_graph/main.py | 6 +- tests/test_mcp_auto_watch.py | 266 +++++++++++++++++++++++++++++++ 3 files changed, 278 insertions(+), 3 deletions(-) create mode 100644 tests/test_mcp_auto_watch.py diff --git a/code_review_graph/incremental.py b/code_review_graph/incremental.py index 7ad0c0315..b0951d5e4 100644 --- a/code_review_graph/incremental.py +++ b/code_review_graph/incremental.py @@ -3199,7 +3199,7 @@ def watch( store: Graph database to update. on_files_updated: Optional callback invoked after each debounced batch of file updates completes. Receives the store as its - only argument. Used by the CLI to run post-processing + only argument. Used by the CLI and MCP auto-watch to run post-processing (FTS, flows, communities) after watch updates. stop_event: Optional event that ends the loop cleanly, for callers that run ``watch`` on a thread they need to shut down. @@ -3326,9 +3326,14 @@ def start_watch_thread( repo_root: Path, store: GraphStore, daemon: bool = True, + *, + on_files_updated: Optional[Callable] = None, ) -> threading.Thread | None: """Start watch mode in a background thread. + ``on_files_updated`` is forwarded to ``watch`` for post-processing + after changed batches, including startup reconciliation. + Returns the started thread, or None if watchdog is unavailable. """ try: @@ -3341,7 +3346,7 @@ def _run() -> None: # A thread cannot take the process down, so the one thing it must not # do is die quietly: the server would keep serving a frozen graph. try: - watch(repo_root, store) + watch(repo_root, store, on_files_updated=on_files_updated) except RuntimeError as exc: logger.error("Auto-watch for %s stopped: %s", repo_root, exc) diff --git a/code_review_graph/main.py b/code_review_graph/main.py index be58e8d91..1ea3defbd 100644 --- a/code_review_graph/main.py +++ b/code_review_graph/main.py @@ -1462,8 +1462,12 @@ def main( watch_store: GraphStore | None = None try: if auto_watch: + from .postprocessing import run_post_processing + watch_store = GraphStore(get_db_path(root)) - thread = start_watch_thread(root, watch_store, daemon=True) + thread = start_watch_thread( + root, watch_store, daemon=True, on_files_updated=run_post_processing, + ) if thread is None: logger.warning("Auto-watch was requested but could not be started") diff --git a/tests/test_mcp_auto_watch.py b/tests/test_mcp_auto_watch.py new file mode 100644 index 000000000..4c9f218d2 --- /dev/null +++ b/tests/test_mcp_auto_watch.py @@ -0,0 +1,266 @@ +"""MCP auto-watch must run the shared post-processing pipeline (issue #1103).""" + +from __future__ import annotations + +import subprocess +import threading +from unittest.mock import Mock, patch + +import pytest + +from code_review_graph import incremental +from code_review_graph import main as crg_main +from code_review_graph.graph import GraphStore +from code_review_graph.postprocessing import run_post_processing + + +@pytest.fixture(autouse=True) +def isolate_mcp_state(monkeypatch): + """Keep main() calls from leaking their default root or changing tool exposure.""" + monkeypatch.setattr(crg_main, "_default_repo_root", None) + monkeypatch.delenv("CRG_TOOLS", raising=False) + + +def test_main_auto_watch_supplies_shared_postprocessing(tmp_path, monkeypatch): + store = Mock(spec=GraphStore) + start_watch = Mock(return_value=Mock(spec=threading.Thread)) + monkeypatch.setattr(crg_main, "get_db_path", Mock(return_value=tmp_path / "graph.db")) + monkeypatch.setattr(crg_main, "GraphStore", Mock(return_value=store)) + monkeypatch.setattr(crg_main, "start_watch_thread", start_watch) + monkeypatch.setattr(crg_main.mcp, "run", Mock()) + + crg_main.main(repo_root=str(tmp_path), auto_watch=True) + + start_watch.assert_called_once_with( + tmp_path, store, daemon=True, on_files_updated=run_post_processing, + ) + store.close.assert_called_once_with() + + +def test_main_without_auto_watch_does_not_start_watcher(tmp_path, monkeypatch): + open_store = Mock() + start_watch = Mock() + monkeypatch.setattr(crg_main, "GraphStore", open_store) + monkeypatch.setattr(crg_main, "start_watch_thread", start_watch) + monkeypatch.setattr(crg_main.mcp, "run", Mock()) + + crg_main.main(repo_root=str(tmp_path), auto_watch=False) + + start_watch.assert_not_called() + open_store.assert_not_called() + + +def test_start_watch_thread_forwards_callback_to_watch(tmp_path, monkeypatch): + callback = Mock(return_value={}) + watch = Mock() + monkeypatch.setattr(incremental, "watch", watch) + + with GraphStore(tmp_path / "graph.db") as store: + thread = incremental.start_watch_thread( + tmp_path, store, daemon=True, on_files_updated=callback, + ) + assert thread is not None + try: + thread.join(timeout=5) + assert not thread.is_alive(), "the mocked watch call did not finish" + watch.assert_called_once_with(tmp_path, store, on_files_updated=callback) + finally: + thread.join(timeout=5) + + +def test_main_auto_watch_reconciliation_keeps_fts_consistent(tmp_path, monkeypatch): + """Use real MCP wiring and startup reconciliation, with no native OS observer.""" + repo = tmp_path / "repo" + repo.mkdir() + source = repo / "mod.py" + source.write_text( + "def alphaHelper():\n return 1\n\n" + "def betaCaller():\n return alphaHelper()\n", + encoding="utf-8", + ) + subprocess.run(["git", "init", "-q"], cwd=repo, check=True, capture_output=True) + subprocess.run(["git", "add", "mod.py"], cwd=repo, check=True, capture_output=True) + db_path = incremental.get_db_path(repo) + + with GraphStore(db_path) as store: + built = incremental.full_build(repo, store) + assert built["errors"] == [] + processed = run_post_processing(store) + assert not processed.get("warnings"), processed + node_ids = {row[0] for row in store._conn.execute("SELECT id FROM nodes")} + indexed_ids = { + row[0] for row in store._conn.execute("SELECT id FROM nodes_fts_docsize") + } + assert indexed_ids == node_ids + assert store._conn.execute( + "SELECT COUNT(*) FROM nodes_fts f JOIN nodes n ON n.id = f.rowid " + "WHERE nodes_fts MATCH ? AND n.name = ?", + ("alphaHelper", "alphaHelper"), + ).fetchone()[0] == 1 + + source.write_text( + source.read_text(encoding="utf-8") + + "\ndef gammaHelper():\n return alphaHelper() + 2\n", + encoding="utf-8", + ) + start_watch = incremental.start_watch_thread + + def start_and_join(*args, **kwargs): + # Forward only what main supplies: never inject the missing callback here. + thread = start_watch(*args, **kwargs) + assert thread is not None + thread.join(timeout=5) + assert not thread.is_alive(), "startup reconciliation did not finish" + return thread + + monkeypatch.setattr(crg_main, "start_watch_thread", start_and_join) + monkeypatch.setattr(crg_main.mcp, "run", Mock()) + with ( + patch("watchdog.observers.Observer") as observer, + patch("time.sleep", side_effect=KeyboardInterrupt), + ): + crg_main.main(repo_root=str(repo), auto_watch=True) + observer.return_value.start.assert_called_once_with() + + with GraphStore(db_path) as store: + assert "gammaHelper" in {node.name for node in store.get_nodes_by_file(str(source))} + matches = { + name: store._conn.execute( + "SELECT COUNT(*) FROM nodes_fts f JOIN nodes n ON n.id = f.rowid " + "WHERE nodes_fts MATCH ? AND n.name = ?", + (name, name), + ).fetchone()[0] + for name in ("gammaHelper", "alphaHelper") + } + node_ids = {row[0] for row in store._conn.execute("SELECT id FROM nodes")} + indexed_ids = { + row[0] for row in store._conn.execute("SELECT id FROM nodes_fts_docsize") + } + assert matches == {"gammaHelper": 1, "alphaHelper": 1} and indexed_ids == node_ids, ( + f"matches={matches}; missing FTS IDs={sorted(node_ids - indexed_ids)}; " + f"stale FTS IDs={sorted(indexed_ids - node_ids)}" + ) + + +def test_mcp_auto_watch_event_batches_keep_fts_consistent(tmp_path, monkeypatch): + """MCP's callback must sync real FTS after each post-startup event batch.""" + from watchdog.events import FileCreatedEvent, FileDeletedEvent, FileModifiedEvent + + from tests.test_watch_robustness import FakeObserver + + repo = tmp_path / "repo" + repo.mkdir() + source = repo / "mod.py" + source.write_text( + "def alphaHelper():\n return 1\n\n" + "def betaCaller():\n return 2\n", + encoding="utf-8", + ) + subprocess.run(["git", "init", "-q"], cwd=repo, check=True, capture_output=True) + subprocess.run(["git", "add", "mod.py"], cwd=repo, check=True, capture_output=True) + db_path = incremental.get_db_path(repo) + + def assert_fts_consistent(store, present, absent=()): + node_ids = {row[0] for row in store._conn.execute("SELECT id FROM nodes")} + # FTS is an external-content table: its docsize rows reveal the actual + # indexed IDs, whereas a plain SELECT from nodes_fts reads live nodes. + indexed_ids = { + row[0] for row in store._conn.execute("SELECT id FROM nodes_fts_docsize") + } + assert not node_ids - indexed_ids, f"missing FTS IDs: {node_ids - indexed_ids}" + assert not indexed_ids - node_ids, f"stale FTS IDs: {indexed_ids - node_ids}" + for name in (*present, *absent): + graph_ids = { + row[0] for row in store._conn.execute( + "SELECT id FROM nodes WHERE name = ?", (name,), + ) + } + assert len(graph_ids) == (1 if name in present else 0), name + matched_ids = { + row[0] for row in store._conn.execute( + "SELECT rowid FROM nodes_fts WHERE nodes_fts MATCH ?", (name,), + ) + } + assert matched_ids == graph_ids, f"{name}: FTS={matched_ids}, graph={graph_ids}" + + with GraphStore(db_path) as store: + assert incremental.full_build(repo, store)["errors"] == [] + # Establish a healthy baseline only; subsequent batches must repair + # FTS through the callback supplied by main(), with no test injection. + processed = run_post_processing(store) + assert not processed.get("warnings"), processed + assert_fts_consistent(store, ("alphaHelper", "betaCaller")) + + observer = FakeObserver() + ready = threading.Event() + stop = threading.Event() + threads = [] + start_watch = incremental.start_watch_thread + + def capture_thread(*args, **kwargs): + thread = start_watch(*args, **kwargs) + assert thread is not None + threads.append(thread) + return thread + + def wait_for_batches(_seconds): + # The first loop tick occurs after reconciliation, handler.start(), + # and observer.start(). Keep the watcher alive while MCP handles tests. + ready.set() + stop.wait(timeout=10) + raise KeyboardInterrupt + + def exercise_batches(**kwargs): + try: + assert ready.wait(timeout=5), "watcher did not reach its post-startup loop" + assert observer.started and observer.handler is not None + handler = observer.handler + assert handler.events_seen == 0 + with GraphStore(db_path) as store: + assert_fts_consistent(store, ("alphaHelper", "betaCaller")) + + created = repo / "extra.py" + created.write_text("def gammaHelper():\n return 3\n", encoding="utf-8") + # Reuse the synchronous batch boundary used by existing watcher + # tests: process() returns after real updates and post-processing. + handler.process([FileCreatedEvent(str(created))]) + handler.raise_if_failed() + assert handler.events_seen == 1 + assert_fts_consistent(store, ("alphaHelper", "betaCaller", "gammaHelper")) + + source.write_text( + "def deltaHelper():\n return 4\n\n" + "def betaCaller():\n return 2\n", + encoding="utf-8", + ) + handler.process([FileModifiedEvent(str(source))]) + handler.raise_if_failed() + assert handler.events_seen == 2 + assert_fts_consistent( + store, ("deltaHelper", "betaCaller", "gammaHelper"), ("alphaHelper",), + ) + + created.unlink() + handler.process([FileDeletedEvent(str(created))]) + handler.raise_if_failed() + assert handler.events_seen == 3 + assert not store.get_nodes_by_file(str(created)) + assert_fts_consistent( + store, ("deltaHelper", "betaCaller"), ("alphaHelper", "gammaHelper"), + ) + finally: + stop.set() + # main() closes its GraphStore as soon as the mocked transport exits. + # Join first so neither watch nor its debouncer outlives that store. + for thread in threads: + thread.join(timeout=5) + assert not thread.is_alive(), "watcher did not stop after event batches" + + monkeypatch.setattr(crg_main, "start_watch_thread", capture_thread) + monkeypatch.setattr(crg_main.mcp, "run", exercise_batches) + with ( + patch("watchdog.observers.Observer", return_value=observer), + patch("time.sleep", side_effect=wait_for_batches), + ): + crg_main.main(repo_root=str(repo), auto_watch=True) + assert observer.stopped From 09dca915946fb03b48e3b809fa1bd1e69498f799 Mon Sep 17 00:00:00 2001 From: Ziqi Lin <760448737@qq.com> Date: Thu, 15 Oct 2026 22:49:41 -0400 Subject: [PATCH 2/2] fix(mcp): preserve startup file events during auto-watch Start the filesystem observer before startup reconciliation so edits made during post-processing are captured. Serialize startup work and event batches through the same processor lock, and prevent the initial debouncer wake-up from being lost. Add regression coverage for startup edits, event serialization, and first-event delivery. --- code_review_graph/incremental.py | 130 ++++++-- tests/test_incremental.py | 10 +- tests/test_mcp_auto_watch_startup_window.py | 123 ++++++++ tests/test_watch_debouncer_wakeup.py | 162 ++++++++++ tests/test_watch_robustness.py | 5 +- tests/test_watch_startup_serialization.py | 319 ++++++++++++++++++++ 6 files changed, 718 insertions(+), 31 deletions(-) create mode 100644 tests/test_mcp_auto_watch_startup_window.py create mode 100644 tests/test_watch_debouncer_wakeup.py create mode 100644 tests/test_watch_startup_serialization.py diff --git a/code_review_graph/incremental.py b/code_review_graph/incremental.py index b0951d5e4..63bd651b6 100644 --- a/code_review_graph/incremental.py +++ b/code_review_graph/incremental.py @@ -2172,6 +2172,54 @@ def incremental_update( _DEBOUNCE_SECONDS = 1 +class _WatchDebouncer(threading.Thread): + """Batch events until a quiet interval passes, then hand them on. + + The algorithm of watchdog 6.0.0's ``EventDebouncer``, with one change: + the wait for a first event checks for queued events and for a stop + under the condition lock before it sleeps. watchdog's waits + unconditionally, so an event or a stop that arrives between ``start()`` + and that first wait is notified to nobody and then slept on. The + observer starts before startup work, so that event can be the only one. + """ + + def __init__(self, interval: float, callback: Callable[[list[Any]], None]) -> None: + super().__init__(name="crg-watch-debouncer", daemon=True) + self._interval = interval + self._callback = callback + self._events: list[Any] = [] + self._stopping = False + self._cond = threading.Condition() + + def handle_event(self, event: Any) -> None: + with self._cond: + self._events.append(event) + self._cond.notify() + + def stop(self) -> None: + """Ask the thread to exit; events still queued are dropped, as in watchdog.""" + with self._cond: + self._stopping = True + self._cond.notify() + + def run(self) -> None: + with self._cond: + while True: + while not self._events and not self._stopping: + self._cond.wait() + if self._interval: + # Every further event restarts the quiet interval. + while not self._stopping: + if not self._cond.wait(timeout=self._interval): + break + if self._stopping: + break + events, self._events = self._events, [] + # Called with the condition held, as watchdog does: events + # arriving meanwhile wait in handle_event for the next batch. + self._callback(events) + + def _raise_watch_update_errors(result: dict, context: str) -> None: """Fail the watch boundary when an incremental update reports errors.""" errors = result.get("errors") or [] @@ -2456,7 +2504,7 @@ def unwatched_paths(self) -> list[str]: return sorted(self._unwatched) def attach(self, observer: Any) -> None: - """Bind the observer, once the initial build has earned one.""" + """Bind the observer that the watches are scheduled on.""" self._observer = observer def schedule_initial(self, handler: Any) -> None: @@ -2893,7 +2941,6 @@ def _create_watch_handler( ): """Create the debounced watchdog handler for one repository.""" from watchdog.events import FileSystemEvent, FileSystemEventHandler - from watchdog.utils.event_debouncer import EventDebouncer ignore_patterns = _load_ignore_patterns(repo_root) parser = CodeParser(repo_root) @@ -2906,6 +2953,10 @@ def __init__(self) -> None: self.failure: BaseException | None = None self.last_event_at: float | None = None self.events_seen: int = 0 + # Every graph write this watcher makes holds this lock: event + # batches here, and startup reconciliation and post-processing in + # watch(). The store's one connection cannot take two writers. + self.lock = threading.Lock() def _relative_path(self, path: str, *, apply_ignores: bool = True) -> str | None: candidate = Path(os.path.abspath(path)) @@ -3072,32 +3123,34 @@ def process(self, events: list[FileSystemEvent]) -> None: # from a watcher whose repository is simply quiet. self.last_event_at = time.time() self.events_seen += len(events) - try: - files_updated, newly_included = self._refresh_ignore_patterns(events) - changed_files = sorted( - newly_included | {path for event in events for path in self._event_paths(event)} - ) - if changed_files: - result = incremental_update( - repo_root, - store, - changed_files=changed_files, - reconcile_stale=False, + with self.lock: + try: + files_updated, newly_included = self._refresh_ignore_patterns(events) + changed_files = sorted( + newly_included + | {path for event in events for path in self._event_paths(event)} ) - _raise_watch_update_errors(result, "incremental update") - files_updated += result["files_updated"] - if files_updated > 0 and on_files_updated is not None: - postprocess_result = on_files_updated(store) - _raise_watch_postprocess_warnings(postprocess_result) - except BaseException as exc: - self.failure = exc + if changed_files: + result = incremental_update( + repo_root, + store, + changed_files=changed_files, + reconcile_stale=False, + ) + _raise_watch_update_errors(result, "incremental update") + files_updated += result["files_updated"] + if files_updated > 0 and on_files_updated is not None: + postprocess_result = on_files_updated(store) + _raise_watch_postprocess_warnings(postprocess_result) + except BaseException as exc: + self.failure = exc def raise_if_failed(self) -> None: if self.failure is not None: raise RuntimeError("watch update failed") from self.failure processor = WatchBatchProcessor() - debouncer = EventDebouncer(_DEBOUNCE_SECONDS, processor.process) + debouncer = _WatchDebouncer(_DEBOUNCE_SECONDS, processor.process) class GraphUpdateHandler(FileSystemEventHandler): def dispatch(self, event: FileSystemEvent) -> None: @@ -3117,6 +3170,10 @@ def stop(self) -> None: def process(self, events: list[FileSystemEvent]) -> None: processor.process(events) + def serialized(self) -> threading.Lock: + """The lock event batches hold while they write to the graph.""" + return processor.lock + def raise_if_failed(self) -> None: processor.raise_if_failed() @@ -3226,17 +3283,36 @@ def watch( # stalled for the whole build. supervisor.report_health(observer_alive=True, phase="initial-build", force=True) - initial = incremental_update(repo_root, store, changed_files=[]) - _raise_watch_update_errors(initial, "initial watch reconciliation") - if initial["files_updated"] > 0 and on_files_updated is not None: - postprocess_result = on_files_updated(store) - _raise_watch_postprocess_warnings(postprocess_result) observer = Observer() supervisor.attach(observer) handler = _create_watch_handler(repo_root, store, on_files_updated) supervisor.schedule_initial(handler) handler.start() - observer.start() + observer_started = False + try: + # The observer starts before reconciliation reads anything, so a file + # saved during startup — post-processing a large graph takes seconds — + # is either read by reconciliation or captured as an event. The batch + # lock is taken first: captured events wait in the debouncer and are + # written to the graph only after startup, never alongside it. Nothing + # here may dispatch to the handler while holding the lock: the + # debouncer holds its own lock while it waits for this one. + with handler.serialized(): + observer.start() + observer_started = True + initial = incremental_update(repo_root, store, changed_files=[]) + _raise_watch_update_errors(initial, "initial watch reconciliation") + if initial["files_updated"] > 0 and on_files_updated is not None: + postprocess_result = on_files_updated(store) + _raise_watch_postprocess_warnings(postprocess_result) + except BaseException: + # Outside the lock, so a batch already waiting can finish; stopping the + # debouncer joins it, and no writer outlives this call. + _run_time_boxed(observer.stop, "observer stop") + if observer_started: + observer.join(timeout=_WATCH_STOP_TIMEOUT) + handler.stop() + raise supervisor.report_health(observer_alive=True, force=True) logger.info("Watching %s for changes... (Ctrl+C to stop)", repo_root) diff --git a/tests/test_incremental.py b/tests/test_incremental.py index 98b99ac74..5ff915115 100644 --- a/tests/test_incremental.py +++ b/tests/test_incremental.py @@ -1364,7 +1364,7 @@ def test_watch_dispatch_ignores_irrelevant_events(self, tmp_path, event_factory) store = GraphStore(tmp_path / "graph.db") debouncer = MagicMock() with patch( - "watchdog.utils.event_debouncer.EventDebouncer", + "code_review_graph.incremental._WatchDebouncer", return_value=debouncer, ): handler = _create_watch_handler(tmp_path, store, None) @@ -1419,7 +1419,7 @@ def on_files_updated(_store): finally: store.close() - def test_watch_startup_postprocess_warning_prevents_observer_start(self, tmp_path): + def test_watch_startup_postprocess_warning_stops_observer(self, tmp_path): import sqlite3 from code_review_graph.postprocessing import run_post_processing @@ -1448,7 +1448,11 @@ def test_watch_startup_postprocess_warning_prevents_observer_start(self, tmp_pat ): watch(tmp_path, store, on_files_updated=run_post_processing) - observer.assert_not_called() + # The observer starts before startup work (PR #1108), so failing + # closed means it is stopped and joined before watch() raises. + observer.return_value.start.assert_called_once_with() + observer.return_value.stop.assert_called() + observer.return_value.join.assert_called() finally: store.close() diff --git a/tests/test_mcp_auto_watch_startup_window.py b/tests/test_mcp_auto_watch_startup_window.py new file mode 100644 index 000000000..718229d59 --- /dev/null +++ b/tests/test_mcp_auto_watch_startup_window.py @@ -0,0 +1,123 @@ +"""MCP auto-watch must not go blind while its startup post-processing runs. + +PR #1108 runs the shared post-processing pipeline from ``watch()`` right after +startup reconciliation, and ``watch()`` only creates and starts the filesystem +observer after that call returns. Any edit saved while post-processing runs is +therefore never observed: the reconciliation scan has already read the file and +the observer snapshot/stream starts after the edit. The file stays stale in the +graph until it is edited again or the server restarts. +""" + +from __future__ import annotations + +import functools +import subprocess +import threading +import time + +from code_review_graph import incremental, postprocessing +from code_review_graph import main as crg_main +from code_review_graph.graph import GraphStore + +# Startup post-processing (FTS sync, full flow trace, community detection) over +# a large graph. Measured at ~1.3-1.7 s for a 17k-node graph (CPython stdlib). +SLOW_POSTPROCESS_SECONDS = 3.0 +EDIT_AFTER_RECONCILIATION_SECONDS = 1.0 + + +def test_edit_saved_during_startup_postprocessing_reaches_graph(tmp_path, monkeypatch): + import watchdog.observers + from watchdog.observers.polling import PollingObserver + + monkeypatch.setattr(crg_main, "_default_repo_root", None) + monkeypatch.delenv("CRG_TOOLS", raising=False) + # Portable, deterministic observer: its snapshot is taken when it starts. + monkeypatch.setattr( + watchdog.observers, "Observer", lambda: PollingObserver(timeout=0.2), + ) + + repo = tmp_path / "repo" + repo.mkdir() + module = repo / "mod.py" + leaf = repo / "leaf.py" + module.write_text("def alphaHelper():\n return 1\n", encoding="utf-8") + leaf.write_text("def leafHelper():\n return 1\n", encoding="utf-8") + subprocess.run(["git", "init", "-q"], cwd=repo, check=True, capture_output=True) + subprocess.run(["git", "add", "-A"], cwd=repo, check=True, capture_output=True) + db_path = incremental.get_db_path(repo) + with GraphStore(db_path) as store: + assert incremental.full_build(repo, store)["errors"] == [] + assert not postprocessing.run_post_processing(store).get("warnings") + + # Offline change (e.g. a `git pull` while the editor was closed): startup + # reconciliation has something to update. + module.write_text( + "def alphaHelper():\n return 1\n\ndef betaHelper():\n return 2\n", + encoding="utf-8", + ) + + real_postprocess = postprocessing.run_post_processing + + def slow_postprocess(store, **kwargs): + time.sleep(SLOW_POSTPROCESS_SECONDS) # a large graph + return real_postprocess(store, **kwargs) + + monkeypatch.setattr(postprocessing, "run_post_processing", slow_postprocess) + + reconciled = threading.Event() + real_update = incremental.incremental_update + + def tracking_update(*args, **kwargs): + result = real_update(*args, **kwargs) + reconciled.set() + return result + + monkeypatch.setattr(incremental, "incremental_update", tracking_update) + + stop = threading.Event() + monkeypatch.setattr( + incremental, "watch", functools.partial(incremental.watch, stop_event=stop), + ) + threads: list[threading.Thread] = [] + real_start = crg_main.start_watch_thread + + def capture(*args, **kwargs): + thread = real_start(*args, **kwargs) + threads.append(thread) + return thread + + monkeypatch.setattr(crg_main, "start_watch_thread", capture) + outcome: dict[str, bool] = {} + + def serve(**_kwargs): + try: + assert reconciled.wait(30), "startup reconciliation never finished" + time.sleep(EDIT_AFTER_RECONCILIATION_SECONDS) + # The user (or agent) saves a file while the server is starting. + leaf.write_text( + "def leafHelper():\n return 1\n\ndef lateEditHelper():\n return 2\n", + encoding="utf-8", + ) + deadline = time.time() + SLOW_POSTPROCESS_SECONDS + 10 + found = False + while time.time() < deadline and not found: + with GraphStore(db_path) as store: + found = "lateEditHelper" in { + node.name for node in store.get_nodes_by_file(str(leaf)) + } + time.sleep(0.2) + outcome["found"] = found + finally: + stop.set() + for thread in threads: + thread.join(timeout=SLOW_POSTPROCESS_SECONDS + 15) + + monkeypatch.setattr(crg_main.mcp, "run", serve) + crg_main.main(repo_root=str(repo), auto_watch=True) + + with GraphStore(db_path) as store: + assert "betaHelper" in {n.name for n in store.get_nodes_by_file(str(module))} + assert outcome.get("found"), ( + "an edit saved while MCP auto-watch was running its startup post-processing " + "never reached the graph" + ) diff --git a/tests/test_watch_debouncer_wakeup.py b/tests/test_watch_debouncer_wakeup.py new file mode 100644 index 000000000..2937b3edb --- /dev/null +++ b/tests/test_watch_debouncer_wakeup.py @@ -0,0 +1,162 @@ +"""The watch debouncer must not lose a wake-up sent before it first waits. + +PR #1108 review: watchdog 6.0.0's ``EventDebouncer.run`` starts with an +unconditional ``Condition.wait()``. ``start()`` returns before the consumer +thread gets there, so an event (or a stop) delivered in between appends and +notifies nobody; the consumer then sleeps on a queued event until some other +event arrives, or never wakes to stop at all. With the observer started +before startup work, that first event can be the only edit the user makes. + +These tests hold the consumer thread *before* its run body, deliver exactly +one event (or the stop) through the production handler, then release it. +No second event, notification or polling tick is ever supplied. The graph +update is stubbed, so no SQLite connection is involved. +""" + +from __future__ import annotations + +import threading +import time + +import pytest +from watchdog.events import FileModifiedEvent + +from code_review_graph import incremental + +_BOUND = incremental._DEBOUNCE_SECONDS + 5.0 + + +@pytest.fixture +def held_debouncers(monkeypatch): + """Start every debouncer thread held before its run body, until released.""" + release = threading.Event() + stop_sent = threading.Event() + held: list[threading.Thread] = [] + real_start = threading.Thread.start + + def start(self): + if hasattr(self, "handle_event"): + run, stop = self.run, self.stop + + def held_run(): + release.wait(_BOUND) + run() + + def signalled_stop(): + stop() + stop_sent.set() + + self.run, self.stop = held_run, signalled_stop + held.append(self) + real_start(self) + + monkeypatch.setattr(threading.Thread, "start", start) + return release, held, stop_sent + + +@pytest.fixture +def updates(monkeypatch): + """Record graph updates instead of performing them.""" + calls: list[list[str]] = [] + done = threading.Event() + + def fake_update(repo_root, store, *, changed_files, **_kwargs): + calls.append(list(changed_files)) + done.set() + return {"files_updated": len(changed_files), "errors": []} + + monkeypatch.setattr(incremental, "incremental_update", fake_update) + return calls, done + + +@pytest.fixture +def source(tmp_path): + path = tmp_path / "mod.py" + path.write_text("def alphaHelper():\n return 1\n", encoding="utf-8") + return path + + +def _stop_bounded(handler) -> threading.Thread: + """Stop the handler on a helper thread; a hang must fail the test, not wedge it.""" + stopper = threading.Thread(target=handler.stop, name="debouncer-stopper", daemon=True) + stopper.start() + stopper.join(_BOUND) + return stopper + + +def test_event_queued_before_the_consumer_first_waits_is_processed( + tmp_path, source, held_debouncers, updates, +): + release, held, _stop_sent = held_debouncers + calls, done = updates + # No store is touched on this path; anything that tried would fail loudly. + handler = incremental._create_watch_handler(tmp_path, object(), None) + handler.start() + try: + assert len(held) == 1, "the debouncer thread was not intercepted" + assert held[0].is_alive() + # Queued and notified while the consumer cannot yet be waiting. + handler.dispatch(FileModifiedEvent(str(source))) + release.set() + + processed = done.wait(_BOUND) + pending = len(getattr(held[0], "_events", [])) + assert processed, ( + f"single event processed: False; events still pending: {pending}" + ) + assert calls == [["mod.py"]] + handler.raise_if_failed() + finally: + release.set() + stopper = _stop_bounded(handler) + assert not stopper.is_alive(), "stopping the debouncer hung" + assert not held[0].is_alive() + + +def test_stop_before_the_consumer_first_waits_does_not_hang( + tmp_path, held_debouncers, updates, +): + release, held, stop_sent = held_debouncers + calls, _done = updates + handler = incremental._create_watch_handler(tmp_path, object(), None) + handler.start() + assert len(held) == 1 + stopper = threading.Thread(target=handler.stop, name="debouncer-stopper", daemon=True) + stopper.start() + # The stop is signalled while the consumer cannot yet be waiting. + assert stop_sent.wait(_BOUND), "stop() itself never returned" + release.set() + stopper.join(_BOUND) + assert not stopper.is_alive(), "a stop sent before the first wait was lost" + assert not held[0].is_alive() + assert calls == [] + + +def test_debouncer_coalesces_a_burst_in_arrival_order_and_restarts_its_timer(): + batches: list[list[str]] = [] + batch_seen = threading.Event() + + def callback(events): + batches.append(list(events)) + batch_seen.set() + + debouncer = incremental._WatchDebouncer(0.3, callback) + debouncer.start() + try: + started = time.monotonic() + for name in ("c", "a", "b"): + debouncer.handle_event(name) + time.sleep(0.1) # each event restarts the quiet interval + assert batch_seen.wait(5) + elapsed = time.monotonic() - started + assert batches == [["c", "a", "b"]] + assert elapsed >= 0.2 + 0.3, "the batch fired before a full quiet interval" + + batch_seen.clear() + debouncer.handle_event("d") + assert batch_seen.wait(5) + assert batches == [["c", "a", "b"], ["d"]] + finally: + debouncer.stop() + debouncer.join(5) + assert not debouncer.is_alive() diff --git a/tests/test_watch_robustness.py b/tests/test_watch_robustness.py index 133c5b51e..68da5dd3d 100644 --- a/tests/test_watch_robustness.py +++ b/tests/test_watch_robustness.py @@ -973,7 +973,10 @@ def test_a_graph_from_another_root_is_refused_not_emptied(self, tmp_path): ): watch(repo, store) assert store.get_all_files() == ["/somewhere/else/mod.py"], "nothing was deleted" - observer.assert_not_called() + # Started before startup reconciliation (PR #1108), and torn down + # again when it refuses the graph. + observer.return_value.stop.assert_called() + observer.return_value.join.assert_called() finally: store.close() diff --git a/tests/test_watch_startup_serialization.py b/tests/test_watch_startup_serialization.py new file mode 100644 index 000000000..5adbd9e07 --- /dev/null +++ b/tests/test_watch_startup_serialization.py @@ -0,0 +1,319 @@ +"""Startup work and event batches share one serialized processing path. + +PR #1108 runs post-processing after the watcher's startup reconciliation. +``watch()`` therefore starts the filesystem observer *before* that startup work +and holds the batch processor's lock from before the observer starts until the +startup work is done. An event captured meanwhile waits in the debouncer and +becomes an ordinary batch once startup releases the lock; the two never write +to the shared ``GraphStore`` at the same time, and a startup failure still +stops the observer and the debouncer before ``watch()`` raises. +""" + +from __future__ import annotations + +import subprocess +import threading +import time + +import pytest +from watchdog.events import FileModifiedEvent + +from code_review_graph import incremental +from code_review_graph.graph import GraphStore +from code_review_graph.postprocessing import run_post_processing +from tests.test_watch_robustness import FakeObserver + +_WATCH_THREAD = "crg-watch-under-test" +_LEAF_EDIT = "def leafHelper():\n return 1\n\ndef lateEditHelper():\n return 2\n" +# Long enough for the debouncer to close a batch and block on the processing lock. +_BATCH_READY_SECONDS = incremental._DEBOUNCE_SECONDS + 1.0 + + +class _RecordingObserver(FakeObserver): + def __init__(self, log: list) -> None: + super().__init__() + self.log = log + self.joined = False + + def start(self) -> None: + self.log.append("observer-start") + super().start() + + def join(self, timeout=None) -> None: + self.joined = True + + +@pytest.fixture +def debouncers(monkeypatch): + """Every debouncer ``watch()`` creates, with every event handed to it.""" + created = [] + + class RecordingDebouncer(incremental._WatchDebouncer): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.received = [] + created.append(self) + + def handle_event(self, event): + self.received.append(event) + super().handle_event(event) + + monkeypatch.setattr(incremental, "_WatchDebouncer", RecordingDebouncer) + return created + + +@pytest.fixture +def repo(tmp_path, monkeypatch): + """A built repository with one offline change for startup to reconcile.""" + monkeypatch.delenv("CRG_DATA_DIR", raising=False) + root = tmp_path / "repo" + root.mkdir() + (root / "mod.py").write_text("def alphaHelper():\n return 1\n", encoding="utf-8") + (root / "leaf.py").write_text("def leafHelper():\n return 1\n", encoding="utf-8") + subprocess.run(["git", "init", "-q"], cwd=root, check=True, capture_output=True) + subprocess.run(["git", "add", "-A"], cwd=root, check=True, capture_output=True) + with GraphStore(incremental.get_db_path(root)) as store: + assert incremental.full_build(root, store)["errors"] == [] + assert not run_post_processing(store).get("warnings") + (root / "mod.py").write_text( + "def alphaHelper():\n return 1\n\ndef betaHelper():\n return 2\n", + encoding="utf-8", + ) + return root + + +@pytest.fixture +def update_log(monkeypatch): + """Log each graph update as (begin|end, startup|batch), in the order they happen.""" + log: list = [] + real_update = incremental.incremental_update + + def tracking_update(*args, **kwargs): + role = "startup" if threading.current_thread().name == _WATCH_THREAD else "batch" + log.append(("update-begin", role)) + try: + return real_update(*args, **kwargs) + finally: + log.append(("update-end", role)) + + monkeypatch.setattr(incremental, "incremental_update", tracking_update) + return log + + +def _run_watch(monkeypatch, repo, observer, on_files_updated): + monkeypatch.setattr("watchdog.observers.Observer", lambda: observer) + store = GraphStore(incremental.get_db_path(repo)) + stop = threading.Event() + outcome: dict = {} + + def run() -> None: + try: + incremental.watch(repo, store, on_files_updated=on_files_updated, stop_event=stop) + except BaseException as exc: # noqa: BLE001 - the test inspects it + outcome["error"] = exc + + thread = threading.Thread(target=run, name=_WATCH_THREAD, daemon=True) + thread.start() + return thread, stop, store, outcome + + +def _graph_has(repo, path, name, timeout: float) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + with GraphStore(incremental.get_db_path(repo)) as store: + if name in {node.name for node in store.get_nodes_by_file(str(path))}: + return True + time.sleep(0.1) + return False + + +def _finish(thread, stop, store, *workers) -> None: + """Stop the watcher, and close the store only once nothing can write to it. + + Every thread that may use the store gets a bounded join. If one is still + alive the store is left open on purpose — closing a connection under a + live writer can crash the interpreter — and the test fails instead. + """ + stop.set() + owners = (*workers, thread) + for owner in owners: + owner.join(timeout=30) + alive = [owner.name for owner in owners if owner.is_alive()] + if alive: + pytest.fail(f"still running, so the store was left open: {alive}") + store.close() + + +def _assert_fts_consistent(repo, present: tuple[str, ...]) -> None: + with GraphStore(incremental.get_db_path(repo)) as store: + node_ids = {row[0] for row in store._conn.execute("SELECT id FROM nodes")} + indexed = {row[0] for row in store._conn.execute("SELECT id FROM nodes_fts_docsize")} + assert indexed == node_ids, ( + f"missing FTS ids {node_ids - indexed}; stale FTS ids {indexed - node_ids}" + ) + for name in present: + graph_ids = { + row[0] for row in store._conn.execute( + "SELECT id FROM nodes WHERE name = ?", (name,), + ) + } + matched = { + row[0] for row in store._conn.execute( + "SELECT rowid FROM nodes_fts WHERE nodes_fts MATCH ?", (name,), + ) + } + assert len(graph_ids) == 1 and matched == graph_ids, (name, graph_ids, matched) + + +def test_single_event_captured_during_startup_reaches_graph_after_startup( + repo, monkeypatch, debouncers, update_log, +): + leaf = repo / "leaf.py" + observer = _RecordingObserver(update_log) + startups = [] + + def on_files_updated(store): + if not startups: + startups.append(True) + update_log.append("startup-postprocess-begin") + # Saved after reconciliation read leaf.py: only the observer sees it. + leaf.write_text(_LEAF_EDIT, encoding="utf-8") + observer.handler.dispatch(FileModifiedEvent(str(leaf))) + time.sleep(_BATCH_READY_SECONDS) + result = run_post_processing(store) + update_log.append("startup-postprocess-end") + return result + return run_post_processing(store) + + thread, stop, store, outcome = _run_watch(monkeypatch, repo, observer, on_files_updated) + try: + # No further event is ever dispatched: the one captured event must + # become a batch on its own once startup lets go. + found = _graph_has(repo, leaf, "lateEditHelper", timeout=20) + finally: + _finish(thread, stop, store, *debouncers) + + assert "error" not in outcome, outcome + assert found, "an event captured during startup never reached the graph" + _assert_fts_consistent(repo, ("alphaHelper", "betaHelper", "leafHelper", "lateEditHelper")) + assert [event.src_path for d in debouncers for event in d.received] == [str(leaf)] + assert update_log == [ + "observer-start", + ("update-begin", "startup"), + ("update-end", "startup"), + "startup-postprocess-begin", + "startup-postprocess-end", + ("update-begin", "batch"), + ("update-end", "batch"), + ] + + +def test_batch_ready_the_moment_the_observer_starts_waits_for_startup( + repo, monkeypatch, debouncers, update_log, +): + """The lock is already held when the observer starts: there is no gap to race.""" + leaf = repo / "leaf.py" + batches: list[threading.Thread] = [] + batch_started = threading.Event() + + class BatchAtStartObserver(_RecordingObserver): + def start(self) -> None: + super().start() + leaf.write_text(_LEAF_EDIT, encoding="utf-8") + batch = threading.Thread( + target=self.handler.process, + args=([FileModifiedEvent(str(leaf))],), + name="instant-batch", + ) + batches.append(batch) + batch.start() + # Give the batch every chance to get in ahead of startup. + batch.join(timeout=0.5) + batch_started.set() + + observer = BatchAtStartObserver(update_log) + thread, stop, store, outcome = _run_watch(monkeypatch, repo, observer, run_post_processing) + try: + assert batch_started.wait(timeout=20), "the observer never started" + batches[0].join(timeout=20) + assert not batches[0].is_alive(), "the batch never finished" + observer.handler.raise_if_failed() + finally: + # This batch bypasses the debouncer, so watch() does not join it. + _finish(thread, stop, store, *batches, *debouncers) + + assert "error" not in outcome, outcome + assert update_log == [ + "observer-start", + ("update-begin", "startup"), + ("update-end", "startup"), + ("update-begin", "batch"), + ("update-end", "batch"), + ] + with GraphStore(incremental.get_db_path(repo)) as check: + assert "betaHelper" in {n.name for n in check.get_nodes_by_file(str(repo / "mod.py"))} + assert "lateEditHelper" in {n.name for n in check.get_nodes_by_file(str(leaf))} + + +@pytest.mark.parametrize( + ("failure", "error_type", "message"), + [ + ("reconciliation", RuntimeError, "initial watch reconciliation reported errors"), + ("postprocess-warning", RuntimeError, "post-processing reported warnings"), + ("postprocess-crash", ValueError, "forced startup crash"), + ], +) +def test_startup_failure_stops_observer_and_debouncer_and_keeps_its_error( + repo, monkeypatch, debouncers, update_log, failure, error_type, message, +): + leaf = repo / "leaf.py" + observer = _RecordingObserver(update_log) + + def queue_one_batch() -> None: + # A batch is waiting on the processing lock when startup fails. + leaf.write_text(_LEAF_EDIT, encoding="utf-8") + observer.handler.dispatch(FileModifiedEvent(str(leaf))) + time.sleep(_BATCH_READY_SECONDS) + update_log.append("startup-failed") + + tracking_update = incremental.incremental_update + + def failing_reconciliation(*args, **kwargs): + result = tracking_update(*args, **kwargs) + if threading.current_thread().name != _WATCH_THREAD: + return result + queue_one_batch() + return {**result, "errors": [{"file": "mod.py", "error": "forced"}]} + + if failure == "reconciliation": + monkeypatch.setattr(incremental, "incremental_update", failing_reconciliation) + + def on_files_updated(store): + if threading.current_thread().name != _WATCH_THREAD: + return run_post_processing(store) + queue_one_batch() + if failure == "postprocess-crash": + raise ValueError("forced startup crash") + return {"warnings": ["forced startup warning"]} + + thread, stop, store, outcome = _run_watch(monkeypatch, repo, observer, on_files_updated) + thread.join(timeout=30) + returned_on_its_own = not thread.is_alive() + _finish(thread, stop, store, *debouncers) + assert returned_on_its_own, "watch() did not return after a startup failure" + + # The startup error is what surfaces, unmasked by teardown or the queued batch. + assert isinstance(outcome.get("error"), error_type), outcome + assert message in str(outcome["error"]) + assert observer.started and observer.stopped and observer.joined + assert debouncers and not any(d.is_alive() for d in debouncers) + # The queued batch ran only after startup let go, and had finished (no + # writer left on the store) by the time watch() returned. + assert update_log == [ + "observer-start", + ("update-begin", "startup"), + ("update-end", "startup"), + "startup-failed", + ("update-begin", "batch"), + ("update-end", "batch"), + ]