"""Multi-repo watch daemon for code-review-graph. Reads ``~/.code-review-graph/watch.toml`` to configure which repositories to watch, then spawns one ``code-review-graph watch`` child process per repo. Monitors the config file for live changes (adding/removing repos) and health-checks child processes, restarting any that die. No external dependencies beyond Python stdlib — no tmux required. """ from __future__ import annotations import json import logging import os import shutil import signal import subprocess import sys import threading import time from collections.abc import Callable from dataclasses import dataclass, field from pathlib import Path from typing import Any if sys.version_info >= (3, 11): import tomllib else: try: import tomli as tomllib # type: ignore[no-redef] except ImportError: tomllib = None # type: ignore[assignment] logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Config file location # --------------------------------------------------------------------------- CONFIG_PATH: Path = Path.home() / ".code-review-graph" / "watch.toml" PID_PATH: Path = Path.home() / ".code-review-graph" / "daemon.pid" STATE_PATH: Path = Path.home() / ".code-review-graph" / "daemon-state.json" _HEALTH_CHECK_INTERVAL = 30 # --------------------------------------------------------------------------- # Dataclasses # --------------------------------------------------------------------------- @dataclass class WatchRepo: """A single repository to watch.""" path: str """Resolved absolute path to the repository root.""" alias: str """Short name for this repo (derived from directory name when not specified).""" @dataclass class DaemonConfig: """Top-level daemon configuration.""" session_name: str = "crg-watch" """Logical daemon name (used in log messages and status output).""" log_dir: Path = field(default_factory=lambda: Path.home() / ".code-review-graph" / "logs") """Directory for per-repo log files.""" poll_interval: int = 2 """Seconds between file-system polls for config changes.""" repos: list[WatchRepo] = field(default_factory=list) """Repositories the daemon watches.""" # --------------------------------------------------------------------------- # Loading # --------------------------------------------------------------------------- def load_config(path: Path | None = None) -> DaemonConfig: """Load daemon configuration from a TOML file. Args: path: Explicit config path. Falls back to :data:`CONFIG_PATH`. Returns: A fully-validated :class:`DaemonConfig`. Raises: RuntimeError: If ``tomllib`` / ``tomli`` is unavailable on Python < 3.11. """ if tomllib is None: raise RuntimeError( "TOML parsing requires the 'tomli' package on Python < 3.11. " "Install it with: pip install tomli" ) config_path = path or CONFIG_PATH if not config_path.exists(): logger.info("Config file not found at %s — using defaults", config_path) return DaemonConfig() with open(config_path, "rb") as fh: raw: dict[str, Any] = tomllib.load(fh) # -- [daemon] section --------------------------------------------------- daemon_section: dict[str, Any] = raw.get("daemon", {}) session_name: str = daemon_section.get("session_name", "crg-watch") log_dir = Path(daemon_section.get("log_dir", str(DaemonConfig().log_dir))) poll_interval: int = int(daemon_section.get("poll_interval", 2)) # -- [[repos]] array ---------------------------------------------------- repos: list[WatchRepo] = [] seen_aliases: set[str] = set() for entry in raw.get("repos", []): repo_path_str: str = entry.get("path", "") if not repo_path_str: logger.warning("Skipping repo entry with empty path") continue repo_path = Path(repo_path_str).expanduser().resolve() if not repo_path.is_dir(): logger.warning("Skipping repo %s — directory does not exist", repo_path) continue has_repo_marker = ( (repo_path / ".git").exists() or (repo_path / ".svn").exists() or (repo_path / ".code-review-graph").exists() ) if not has_repo_marker: logger.warning( "Skipping repo %s — no .git, .svn, or .code-review-graph directory found", repo_path, ) continue alias: str = entry.get("alias", "") or repo_path.name if alias in seen_aliases: logger.warning("Skipping duplicate alias '%s' for repo %s", alias, repo_path) continue seen_aliases.add(alias) repos.append(WatchRepo(path=str(repo_path), alias=alias)) return DaemonConfig( session_name=session_name, log_dir=log_dir, poll_interval=poll_interval, repos=repos, ) # --------------------------------------------------------------------------- # Saving # --------------------------------------------------------------------------- def _serialize_toml(config: DaemonConfig) -> str: """Serialize a :class:`DaemonConfig` to TOML text. ``tomllib`` is read-only, so we build the TOML manually. """ lines: list[str] = [ "[daemon]", f'session_name = "{config.session_name}"', f'log_dir = "{config.log_dir}"', f"poll_interval = {config.poll_interval}", ] for repo in config.repos: lines.append("") lines.append("[[repos]]") lines.append(f'path = "{repo.path}"') lines.append(f'alias = "{repo.alias}"') lines.append("") # trailing newline return "\n".join(lines) def save_config(config: DaemonConfig, path: Path | None = None) -> None: """Write *config* back to a TOML file. Creates parent directories if they do not exist. Args: config: The daemon configuration to persist. path: Explicit config path. Falls back to :data:`CONFIG_PATH`. """ config_path = path or CONFIG_PATH config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text(_serialize_toml(config), encoding="utf-8") logger.info("Config saved to %s", config_path) # --------------------------------------------------------------------------- # Convenience helpers (used by CLI commands) # --------------------------------------------------------------------------- def add_repo_to_config( repo_path: str, alias: str | None = None, config_path: Path | None = None, ) -> DaemonConfig: """Add a repository to the daemon config and persist the change. Args: repo_path: Path to the repository (will be resolved to absolute). alias: Optional short name. Derived from dirname if *None*. config_path: Explicit config file path. Falls back to :data:`CONFIG_PATH`. Returns: The updated :class:`DaemonConfig`. Raises: ValueError: If the path is not a valid repository directory. """ resolved = Path(repo_path).expanduser().resolve() if not resolved.is_dir(): raise ValueError(f"Not a directory: {resolved}") has_repo_marker = ( (resolved / ".git").exists() or (resolved / ".svn").exists() or (resolved / ".code-review-graph").exists() ) if not has_repo_marker: raise ValueError(f"No .git, .svn, or .code-review-graph directory in {resolved}") effective_alias = alias or resolved.name config = load_config(config_path) # Check for duplicate path or alias for existing in config.repos: if existing.path == str(resolved): logger.warning("Repo %s is already configured — skipping", resolved) return config if existing.alias == effective_alias: raise ValueError(f"Alias '{effective_alias}' is already in use by {existing.path}") config.repos.append(WatchRepo(path=str(resolved), alias=effective_alias)) save_config(config, config_path) return config def remove_repo_from_config( path_or_alias: str, config_path: Path | None = None, ) -> DaemonConfig: """Remove a repository from the daemon config by path or alias. Args: path_or_alias: Either the absolute/relative repo path or its alias. config_path: Explicit config file path. Falls back to :data:`CONFIG_PATH`. Returns: The updated :class:`DaemonConfig`. """ config = load_config(config_path) resolved = str(Path(path_or_alias).expanduser().resolve()) original_count = len(config.repos) config.repos = [r for r in config.repos if r.path != resolved and r.alias != path_or_alias] if len(config.repos) == original_count: logger.warning( "No repo matching '%s' found in config — nothing removed", path_or_alias, ) else: save_config(config, config_path) return config # --------------------------------------------------------------------------- # PID file management # --------------------------------------------------------------------------- def write_pid(pid: int | None = None, path: Path | None = None) -> None: """Write the current (or given) PID to the PID file.""" pid_path = path or PID_PATH pid_path.parent.mkdir(parents=True, exist_ok=True) pid_path.write_text(str(pid or os.getpid()), encoding="utf-8") def read_pid(path: Path | None = None) -> int | None: """Read the daemon PID from disk. Returns None if missing/invalid.""" pid_path = path or PID_PATH if not pid_path.exists(): return None try: return int(pid_path.read_text(encoding="utf-8").strip()) except (ValueError, OSError): return None def clear_pid(path: Path | None = None) -> None: """Remove the PID file.""" pid_path = path or PID_PATH try: pid_path.unlink(missing_ok=True) except OSError: pass # Win32 constants for the OpenProcess-based liveness check (#511). _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 _ERROR_ACCESS_DENIED = 5 _WAIT_OBJECT_0 = 0x0 def _pid_alive_windows( pid: int, kernel32: Any, get_last_error: Callable[[], int] | None = None, ) -> bool: """Win32 PID liveness check via OpenProcess/WaitForSingleObject. The kernel32 interface is injected so tests can drive handle/wait outcomes on non-Windows platforms. *get_last_error* defaults to ``kernel32.GetLastError``; the real caller passes ``ctypes.get_last_error`` (reliable with ``use_last_error=True``). """ if get_last_error is None: get_last_error = kernel32.GetLastError handle = kernel32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid) if not handle: # NULL handle: process is dead, or we lack access. ACCESS_DENIED # means it exists but is owned by another user — treat as alive. return get_last_error() == _ERROR_ACCESS_DENIED try: # WAIT_OBJECT_0 means the process handle is signaled (it exited). return kernel32.WaitForSingleObject(handle, 0) != _WAIT_OBJECT_0 finally: kernel32.CloseHandle(handle) def pid_alive(pid: int) -> bool: """Cross-platform check whether a process with *pid* is running. On Windows ``os.kill(pid, 0)`` routes to GenerateConsoleCtrlEvent and raises ``OSError`` (WinError 87) for alive PIDs outside the caller's console process group (#511), so the Win32 API is used instead. """ if sys.platform == "win32": import ctypes kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) return _pid_alive_windows(pid, kernel32, ctypes.get_last_error) try: os.kill(pid, 0) # signal 0 = existence check return True except ProcessLookupError: return False except PermissionError: return True # process exists but owned by another user except OSError as exc: # Unexpected platform quirk — treat as not alive rather than crash. logger.debug("PID %d liveness check failed: %s", pid, exc) return False def is_daemon_running(path: Path | None = None) -> bool: """Check whether a daemon process is alive.""" pid = read_pid(path) if pid is None: return False if pid_alive(pid): return True # Stale PID file — clean up clear_pid(path) return False # --------------------------------------------------------------------------- # Child state persistence (for cross-process status queries) # --------------------------------------------------------------------------- def load_state(path: Path | None = None) -> dict[str, Any]: """Load persisted child process state from disk. Returns a dict mapping alias to ``{"pid": int, "path": str}``. Returns an empty dict if the file is missing or corrupt. """ state_path = path or STATE_PATH if not state_path.exists(): return {} try: return json.loads(state_path.read_text(encoding="utf-8")) # type: ignore[no-any-return] except (json.JSONDecodeError, OSError): return {} def _is_pid_alive(pid: int) -> bool: """Check whether a process with the given PID is running.""" return pid_alive(pid) # --------------------------------------------------------------------------- # ConfigWatcher — monitors config file for live changes # --------------------------------------------------------------------------- class ConfigWatcher: """Watches the daemon config file for changes and triggers reconciliation.""" def __init__( self, config_path: Path, callback: Callable[[], None], poll_interval: int = 2, ) -> None: self._config_path = config_path self._callback = callback self._poll_interval = poll_interval self._observer: Any = None # watchdog Observer when available self._last_mtime: float = 0.0 self._poll_thread: threading.Thread | None = None self._stop_event: threading.Event = threading.Event() # ------------------------------------------------------------------ # Public # ------------------------------------------------------------------ def start(self) -> None: """Begin watching the config file for modifications.""" try: from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer watcher = self class _Handler(FileSystemEventHandler): # type: ignore[misc] def on_modified(self, event: Any) -> None: if Path(event.src_path).resolve() == watcher._config_path.resolve(): watcher._on_config_changed() handler = _Handler() self._observer = Observer() self._observer.schedule( handler, str(self._config_path.parent), recursive=False, ) self._observer.daemon = True self._observer.start() logger.info( "Config watcher started (watchdog) for %s", self._config_path, ) except ImportError: # Fallback to polling when watchdog is unavailable logger.info( "watchdog not available — falling back to polling for %s", self._config_path, ) self._start_polling() def stop(self) -> None: """Stop watching the config file.""" self._stop_event.set() if self._observer is not None: self._observer.stop() self._observer.join(timeout=5) self._observer = None if self._poll_thread is not None: self._poll_thread.join(timeout=5) self._poll_thread = None # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ def _start_polling(self) -> None: """Poll the config file mtime in a background thread.""" if self._config_path.exists(): self._last_mtime = self._config_path.stat().st_mtime def _poll() -> None: while not self._stop_event.is_set(): self._stop_event.wait(self._poll_interval) if self._stop_event.is_set(): break try: if not self._config_path.exists(): continue mtime = self._config_path.stat().st_mtime if mtime != self._last_mtime: self._last_mtime = mtime self._on_config_changed() except OSError: pass self._poll_thread = threading.Thread( target=_poll, daemon=True, name="config-poller", ) self._poll_thread.start() def _on_config_changed(self) -> None: """Handle a detected config file modification.""" logger.info("Config file changed, triggering reconciliation") try: self._callback() except Exception: logger.exception("Error during config-change reconciliation") # --------------------------------------------------------------------------- # WatchDaemon — manages child processes for multi-repo watching # --------------------------------------------------------------------------- class WatchDaemon: """Manages child processes for multi-repo file watching. Each watched repository gets a ``code-review-graph watch`` child process managed via :mod:`subprocess`. No external tools (tmux, screen, etc.) are required. """ def __init__( self, config: DaemonConfig | None = None, config_path: Path | None = None, ) -> None: self._config: DaemonConfig = config or load_config(config_path) self._config_path: Path = config_path or CONFIG_PATH self._state_path: Path = STATE_PATH self._children: dict[str, subprocess.Popen[bytes]] = {} self._current_repos: dict[str, WatchRepo] = {} self._config_watcher: ConfigWatcher | None = None self._health_thread: threading.Thread | None = None self._health_stop: threading.Event = threading.Event() self._lock: threading.Lock = threading.Lock() # ------------------------------------------------------------------ # Public interface # ------------------------------------------------------------------ def start(self) -> None: """Spawn a watcher child process for each configured repo.""" logger.info("Starting daemon '%s'", self._config.session_name) # Auto-register repos in the central registry from .registry import Registry registry = Registry() for repo in self._config.repos: registry.register(repo.path, alias=repo.alias) # Build initial graph for repos that lack a database for repo in self._config.repos: db_path = Path(repo.path) / ".code-review-graph" / "graph.db" if not db_path.exists(): self._initial_build(repo) # Spawn a watcher child for every repo for repo in self._config.repos: self._start_watcher(repo) # Track current state self._current_repos = {r.alias: r for r in self._config.repos} # Persist child PIDs to disk for cross-process status queries self._save_state() # Start watching the config file for live changes self.start_config_watcher() # Start health checker to auto-restart dead watchers self.start_health_checker() msg = f"Daemon started — watching {len(self._config.repos)} repo(s)" logger.info(msg) print(msg) # noqa: T201 def stop(self) -> None: """Tear down the daemon: stop watchers, terminate children.""" self.stop_config_watcher() self.stop_health_checker() with self._lock: for alias, proc in list(self._children.items()): self._terminate_child(alias, proc) self._children.clear() self._current_repos.clear() self._clear_state() clear_pid() logger.info("Daemon stopped") def reconcile(self, new_config: DaemonConfig | None = None) -> None: """Reconcile running watchers with the (possibly updated) config. Child processes are started, stopped, or restarted to match the desired state. New repos are registered in the central registry and their graphs are built automatically (mirroring ``start()``). """ if new_config is not None: self._config = new_config desired: dict[str, WatchRepo] = {r.alias: r for r in self._config.repos} current: set[str] = set(self._current_repos.keys()) to_add: set[str] = desired.keys() - current to_remove: set[str] = current - desired.keys() to_update: set[str] = { alias for alias in desired.keys() & current if desired[alias].path != self._current_repos[alias].path } # Register new/updated repos and build graphs *before* acquiring # the lock so that long-running builds don't block health checks. if to_add or to_update: from .registry import Registry registry = Registry() repos_needing_build: list[WatchRepo] = [] for alias in to_add | to_update: repo = desired[alias] registry.register(repo.path, alias=repo.alias) db_path = Path(repo.path) / ".code-review-graph" / "graph.db" if not db_path.exists(): repos_needing_build.append(repo) for repo in repos_needing_build: self._initial_build(repo) with self._lock: # Remove stale watchers for alias in to_remove: proc = self._children.pop(alias, None) if proc is not None: self._terminate_child(alias, proc) del self._current_repos[alias] # Add new watchers for alias in to_add: repo = desired[alias] self._start_watcher(repo) self._current_repos[alias] = repo # Update changed watchers (path changed for same alias) for alias in to_update: proc = self._children.pop(alias, None) if proc is not None: self._terminate_child(alias, proc) repo = desired[alias] self._start_watcher(repo) self._current_repos[alias] = repo # Persist updated state self._save_state() logger.info( "Reconcile complete — added: %d, removed: %d, updated: %d", len(to_add), len(to_remove), len(to_update), ) def status(self) -> dict[str, Any]: """Return a summary of daemon state. When called from the daemon process itself, uses the in-memory ``_children`` dict. When called from a separate process (e.g. the CLI ``status`` command), falls back to the persisted state file and checks liveness via ``os.kill(pid, 0)``. """ repos: list[dict[str, Any]] = [] with self._lock: if self._children: # In-process: we have live Popen handles for alias, repo in self._current_repos.items(): proc = self._children.get(alias) alive = proc is not None and proc.poll() is None repos.append( { "alias": alias, "path": repo.path, "alive": alive, "pid": proc.pid if proc is not None else None, } ) else: # Cross-process: read persisted state from disk state = load_state(self._state_path) for repo in self._config.repos: entry = state.get(repo.alias, {}) pid: int | None = entry.get("pid") alive = pid is not None and _is_pid_alive(pid) repos.append( { "alias": repo.alias, "path": repo.path, "alive": alive, "pid": pid, } ) return { "session_name": self._config.session_name, "running": True, "repos": repos, } # ------------------------------------------------------------------ # Config watching # ------------------------------------------------------------------ def start_config_watcher(self) -> None: """Begin watching the config file for live edits.""" self._config_watcher = ConfigWatcher( config_path=self._config_path, callback=self._on_config_change, poll_interval=self._config.poll_interval, ) self._config_watcher.start() def _on_config_change(self) -> None: """Reload configuration and reconcile running watchers.""" try: new_config = load_config(self._config_path) except Exception: logger.warning( "Failed to parse config file — keeping last good config", exc_info=True, ) return self.reconcile(new_config) def stop_config_watcher(self) -> None: """Stop the config file watcher if running.""" if self._config_watcher is not None: self._config_watcher.stop() self._config_watcher = None # ------------------------------------------------------------------ # Health checking # ------------------------------------------------------------------ def start_health_checker(self) -> None: """Start the background health-check thread.""" self._health_stop = threading.Event() self._health_thread = threading.Thread( target=self._health_loop, daemon=True, name="health-checker", ) self._health_thread.start() logger.info( "Health checker started (interval=%ds)", _HEALTH_CHECK_INTERVAL, ) def stop_health_checker(self) -> None: """Stop the health-check thread.""" if hasattr(self, "_health_stop"): self._health_stop.set() if hasattr(self, "_health_thread") and self._health_thread is not None: self._health_thread.join(timeout=5) self._health_thread = None def _health_loop(self) -> None: """Periodically check child processes and restart dead ones.""" while not self._health_stop.is_set(): self._health_stop.wait(_HEALTH_CHECK_INTERVAL) if self._health_stop.is_set(): break self._check_health() def _check_health(self) -> None: """Check each watcher child and restart if dead.""" restarted = False with self._lock: for alias, repo in list(self._current_repos.items()): proc = self._children.get(alias) if proc is None or proc.poll() is not None: logger.warning("Watcher for '%s' is dead — restarting", alias) # Clean up dead process entry self._children.pop(alias, None) self._start_watcher(repo) restarted = True if restarted: self._save_state() # ------------------------------------------------------------------ # Daemonization # ------------------------------------------------------------------ def daemonize(self) -> None: """Fork to background using the double-fork pattern. Redirects stdout/stderr to the daemon log file. Writes PID file. Sets up SIGTERM handler for graceful shutdown. On Windows, forking is not supported — the daemon runs in the foreground and a warning is logged. """ if sys.platform == "win32": logger.warning("Forking is not supported on Windows — running in foreground") write_pid() self._setup_signal_handlers() return # First fork pid = os.fork() if pid > 0: # Parent exits sys.exit(0) # Become session leader os.setsid() # Second fork (prevent acquiring a controlling terminal) pid = os.fork() if pid > 0: sys.exit(0) # Redirect file descriptors sys.stdout.flush() sys.stderr.flush() self._config.log_dir.mkdir(parents=True, exist_ok=True) log_file = self._config.log_dir / "daemon.log" # Open log file for stdout/stderr fd = os.open( str(log_file), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644, ) os.dup2(fd, sys.stdout.fileno()) os.dup2(fd, sys.stderr.fileno()) # Redirect stdin from /dev/null devnull = os.open(os.devnull, os.O_RDONLY) os.dup2(devnull, sys.stdin.fileno()) os.close(devnull) if fd > 2: os.close(fd) # Write PID file write_pid() # Set up signal handlers self._setup_signal_handlers() logger.info("Daemonized (PID %d)", os.getpid()) def _setup_signal_handlers(self) -> None: """Install SIGTERM/SIGHUP handlers for graceful shutdown.""" def _handle_sigterm(signum: int, frame: Any) -> None: logger.info("Received signal %d — shutting down", signum) self.stop() sys.exit(0) signal.signal(signal.SIGTERM, _handle_sigterm) if sys.platform != "win32": signal.signal(signal.SIGHUP, _handle_sigterm) def run_forever(self) -> None: """Block forever, keeping the daemon alive. The config watcher and health checker run in background threads. This method sleeps in the main thread until interrupted. """ try: while True: time.sleep(1) except KeyboardInterrupt: logger.info("Keyboard interrupt — stopping daemon") self.stop() # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _save_state(self) -> None: """Persist child PIDs and repo paths to disk for cross-process queries. Called after any mutation of ``_children`` so that ``status`` commands running in a separate process can determine which watchers are alive. """ state: dict[str, dict[str, Any]] = {} for alias, proc in self._children.items(): repo = self._current_repos.get(alias) state[alias] = { "pid": proc.pid, "path": repo.path if repo else "", } try: self._state_path.parent.mkdir(parents=True, exist_ok=True) self._state_path.write_text(json.dumps(state), encoding="utf-8") except OSError: logger.warning("Failed to persist daemon state to %s", self._state_path) def _clear_state(self) -> None: """Remove the state file from disk.""" try: self._state_path.unlink(missing_ok=True) except OSError: pass def _start_watcher(self, repo: WatchRepo) -> None: """Spawn a child process running ``code-review-graph watch`` for *repo*.""" self._config.log_dir.mkdir(parents=True, exist_ok=True) log_path = self._config.log_dir / f"{repo.alias}.log" crg_bin = shutil.which("code-review-graph") if crg_bin: cmd: list[str] = [crg_bin, "watch", "--repo", repo.path] else: cmd = [ sys.executable, "-m", "code_review_graph", "watch", "--repo", repo.path, ] log_fd = open(log_path, "ab") # noqa: SIM115 try: proc = subprocess.Popen( cmd, cwd=repo.path, stdout=log_fd, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, ) except Exception: log_fd.close() logger.exception("Failed to start watcher for '%s'", repo.alias) return # The log fd is inherited by the child; we can close our copy. # The child keeps the fd open via its own reference. log_fd.close() self._children[repo.alias] = proc logger.info( "Started watcher for '%s' (PID %d) — log: %s", repo.alias, proc.pid, log_path, ) @staticmethod def _terminate_child(alias: str, proc: subprocess.Popen[bytes]) -> None: """Gracefully terminate a child process (SIGTERM, then SIGKILL).""" if proc.poll() is not None: return # already dead logger.info("Terminating watcher '%s' (PID %d)", alias, proc.pid) proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: logger.warning("Watcher '%s' did not stop — sending SIGKILL", alias) proc.kill() proc.wait(timeout=5) def _initial_build(self, repo: WatchRepo) -> None: """Run a one-off graph build for a repo that has no database yet.""" logger.info("Building initial graph for %s...", repo.alias) crg_bin = shutil.which("code-review-graph") if crg_bin: cmd: list[str] = [crg_bin, "build", "--repo", repo.path] else: cmd = [ sys.executable, "-m", "code_review_graph", "build", "--repo", repo.path, ] result = subprocess.run( cmd, capture_output=True, text=True, check=False, ) if result.returncode != 0: logger.warning( "Initial build for '%s' failed (rc=%d): %s", repo.alias, result.returncode, result.stderr.strip(), )