Function watch_helper

src/codegraphcontext/cli/cli_helpers.py:1088–1209

Watch a directory for changes and auto-update the graph (blocking mode).

(
    path: str,
    context: Optional[str] = None,
    use_polling: Optional[bool] = None,
    sync_on_start: bool = False,
)

def watch_helper(
    path: str,
    context: Optional[str] = None,
    use_polling: Optional[bool] = None,
    sync_on_start: bool = False,
):
    """Watch a directory for changes and auto-update the graph (blocking mode)."""
    import logging
    from ..core.watcher import CodeWatcher
    
    # Suppress verbose watchdog DEBUG logs
    logging.getLogger('watchdog').setLevel(logging.WARNING)
    logging.getLogger('watchdog.observers').setLevel(logging.WARNING)
    logging.getLogger('watchdog.observers.inotify_buffer').setLevel(logging.WARNING)
    
    services = _initialize_services(context)
    if not all(services[:3]):
        _fail_services_init()

    db_manager, graph_builder, code_finder, ctx = services
    path_obj = Path(path).resolve()

    if not path_obj.exists():
        console.print(f"[red]Error: Path does not exist: {path_obj}[/red]")
        db_manager.close_driver()
        raise typer.Exit(code=1)
    
    if not path_obj.is_dir():
        console.print(f"[red]Error: Path must be a directory: {path_obj}[/red]")
        db_manager.close_driver()
        raise typer.Exit(code=1)

    console.print(f"[bold cyan]🔍 Watching {path_obj} for changes...[/bold cyan]")
    
    # Check if already indexed — use File node count as a robust fallback so a
    # transient empty result from list_indexed_repositories never triggers a
    # destructive full rescan of an already-populated graph.
    indexed_repos = code_finder.list_indexed_repositories()
    is_indexed = any_repo_matches_path(indexed_repos, path_obj)
    if not is_indexed:
        # Fallback: count File nodes whose path starts with this repo's path.
        # If > 100 exist, the repo is clearly already indexed — skip the scan.
        try:
            with code_finder.driver.session() as _s:
                _r = _s.run(
                    "MATCH (n:File) WHERE n.path STARTS WITH $p RETURN count(n) AS c",
                    p=path_obj.as_posix() + "/"
                )
                _count = _r.single()["c"]
            if _count > 100:
                info_logger(
                    f"[watch] list_indexed_repositories returned no match for {path_obj} "
                    f"but {_count} File nodes exist — treating as already indexed."
                )
                is_indexed = True
        except Exception as _e:
            warning_logger(f"[watch] Fallback indexed check failed: {_e}")