| 1 | """Typed source health: classify a source/tool outcome honestly. |
| 2 | |
| 3 | The pipeline historically collapsed every failure into "returned nothing" or a |
| 4 | flat ``errors_by_source`` entry, which hides the difference between a tool that |
| 5 | is *absent*, one that is *present but broken* (the classic stale-venv-shim after |
| 6 | a Python upgrade), one that *timed out*, and one that merely *degraded* (fewer |
| 7 | results than expected). This module gives callers a small typed vocabulary so |
| 8 | warnings can say what actually happened and prescribe the right fix. |
| 9 | |
| 10 | It complements ``preflight.py`` (which gates doomed *queries*); this gates |
| 11 | doomed *sources/tools*. |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import os |
| 17 | import shutil |
| 18 | import subprocess |
| 19 | from dataclasses import dataclass |
| 20 | from pathlib import Path |
| 21 | from typing import Dict, Iterable, List, Optional, Tuple |
| 22 | |
| 23 | # Health states, best to worst. |
| 24 | OK = "ok" |
| 25 | DEGRADED = "degraded" # ran, but returned less than expected |
| 26 | MISSING = "missing" # tool/binary/credential absent |
| 27 | BROKEN = "broken" # present but won't execute (stale shim, bad perms) |
| 28 | TIMEOUT = "timeout" # exceeded the probe deadline |
| 29 | ERROR = "error" # ran and failed for another reason |
| 30 | |
| 31 | # Per-run outcomes. Doctor does not emit these: it predicts source readiness |
| 32 | # before retrieval, while Report.source_status records what happened in one run. |
| 33 | NO_RESULTS = "no-results" |
| 34 | PARTIAL = "partial" |
| 35 | RATE_LIMITED = "rate-limited" |
| 36 | AUTH_FAILED = "auth-failed" |
| 37 | PAYMENT_REQUIRED = "payment-required" # HTTP 402 / credits exhausted: top up, not re-login |
| 38 | UNREACHABLE = "unreachable" |
| 39 | SCHEMA_DRIFT = "schema-drift" |
| 40 | SKIPPED_UNCONFIGURED = "skipped-unconfigured" |
| 41 | |
| 42 | |
| 43 | def credits_exhausted_label(source: str) -> str: |
| 44 | """Human label for a ``PAYMENT_REQUIRED`` outcome on ``source``. |
| 45 | |
| 46 | The X source names the API whose credits ran out; every other source |
| 47 | gets the generic phrasing so the same label serves render summaries and |
| 48 | the doctor post-mortem without each surface hand-writing its own. |
| 49 | """ |
| 50 | return "X API credits exhausted" if source == "x" else "credits exhausted" |
| 51 | |
| 52 | |
| 53 | @dataclass |
| 54 | class SourceHealth: |
| 55 | """Typed outcome for a source or the tool backing it. |
| 56 | |
| 57 | ``state`` is one of the module-level constants. ``reason`` is a short, |
| 58 | human-readable explanation suitable for a run warning. |
| 59 | """ |
| 60 | |
| 61 | name: str |
| 62 | state: str |
| 63 | reason: str = "" |
| 64 | |
| 65 | @property |
| 66 | def ok(self) -> bool: |
| 67 | return self.state == OK |
| 68 | |
| 69 | @property |
| 70 | def usable(self) -> bool: |
| 71 | """True when the source produced something worth keeping (ok/degraded).""" |
| 72 | return self.state in (OK, DEGRADED) |
| 73 | |
| 74 | |
| 75 | def probe_command( |
| 76 | command: list[str], |
| 77 | timeout: float = 5.0, |
| 78 | ) -> SourceHealth: |
| 79 | """Probe an external command, distinguishing missing/broken/timeout/ok. |
| 80 | |
| 81 | Separating these is what lets the caller emit a correct repair prescription |
| 82 | instead of a generic "failed": |
| 83 | - ``missing``: the executable is not on PATH. |
| 84 | - ``broken``: on PATH but won't run — FileNotFoundError/OSError on exec, or |
| 85 | shell exit 126/127 (not-executable / not-found-after-resolution), the |
| 86 | signature of a stale interpreter shim after an upgrade. |
| 87 | - ``timeout``: exceeded ``timeout`` seconds. |
| 88 | - ``ok``: exited 0. |
| 89 | - ``error``: ran but exited non-zero for another reason. |
| 90 | |
| 91 | The command should be side-effect-free (e.g. ``["gh", "auth", "status"]``); |
| 92 | callers pass a status/version subcommand, not a mutating one. |
| 93 | """ |
| 94 | name = command[0] if command else "" |
| 95 | if not name or shutil.which(name) is None: |
| 96 | return SourceHealth(name=name, state=MISSING, reason=f"{name or 'command'} not found on PATH") |
| 97 | |
| 98 | try: |
| 99 | proc = subprocess.run( |
| 100 | command, |
| 101 | capture_output=True, |
| 102 | text=True, |
| 103 | timeout=timeout, |
| 104 | ) |
| 105 | except (FileNotFoundError, OSError) as exc: |
| 106 | return SourceHealth(name=name, state=BROKEN, reason=f"{name} present but won't execute: {exc}") |
| 107 | except subprocess.TimeoutExpired: |
| 108 | return SourceHealth(name=name, state=TIMEOUT, reason=f"{name} timed out after {timeout:g}s") |
| 109 | |
| 110 | if proc.returncode == 0: |
| 111 | return SourceHealth(name=name, state=OK) |
| 112 | if proc.returncode in (126, 127): |
| 113 | return SourceHealth(name=name, state=BROKEN, reason=f"{name} not executable (exit {proc.returncode})") |
| 114 | detail = (proc.stderr or proc.stdout or "").strip().splitlines() |
| 115 | first = detail[0] if detail else f"exit {proc.returncode}" |
| 116 | return SourceHealth(name=name, state=ERROR, reason=f"{name}: {first}") |
| 117 | |
| 118 | |
| 119 | # --------------------------------------------------------------------------- |
| 120 | # Dependency probes (doctor command, issue #692). |
| 121 | # |
| 122 | # ``probe_dependency`` generalizes ``probe_command`` for the skill's external |
| 123 | # binaries (yt-dlp, Printing Press CLIs, node for the vendored bird client, |
| 124 | # ffmpeg). It answers three questions the bare shutil.which gate cannot: |
| 125 | # - Is the binary genuinely runnable (a stale shim that resolves on PATH but |
| 126 | # cannot exec is BROKEN, not available)? |
| 127 | # - If not, WHICH fix applies (install vs reinstall vs a PATH edit), keyed to |
| 128 | # the package manager that owns the binary on this machine? |
| 129 | # - Is an on-disk binary merely off the agent-subprocess PATH (the Digg |
| 130 | # ~/.local/bin case) — MISSING with a PATH-fix, never "installed"? |
| 131 | # |
| 132 | # Semantics follow the engine gate: availability means PATH-resolvable in THIS |
| 133 | # process, not present-on-disk. Probes are one short-timeout version exec each |
| 134 | # and memoized per process, so doctor and setup can consult them freely. |
| 135 | # --------------------------------------------------------------------------- |
| 136 | |
| 137 | # Per-probe budget in seconds: a healthy --version exec is near-instant, so a |
| 138 | # slow probe is itself a diagnostic (network-mounted shim, hung interpreter). |
| 139 | PROBE_TIMEOUT = 5.0 |
| 140 | |
| 141 | _PP_CLI_SUFFIX = "-pp-cli" |
| 142 | # Matches setup_wizard.PRINTING_PRESS_NPM (pinned catalog installer). |
| 143 | _PRINTING_PRESS_NPM = "@mvanhorn/printing-press-library@0.1.16" |
| 144 | |
| 145 | # Dependencies the doctor probes by default. |
| 146 | KNOWN_DEPENDENCIES: Tuple[str, ...] = ("yt-dlp", "digg-pp-cli", "node", "ffmpeg", "grok") |
| 147 | |
| 148 | # Cheap side-effect-free version invocation per dependency (default --version). |
| 149 | _VERSION_ARGS: Dict[str, List[str]] = { |
| 150 | "ffmpeg": ["-version"], |
| 151 | } |
| 152 | |
| 153 | # Package managers each dependency may be owned by, in preference order, and |
| 154 | # the (install, reinstall) prescription for each. "reinstall" wording matters: |
| 155 | # a BROKEN binary is present, so telling the user to "install" it reads as a |
| 156 | # no-op ("it's already installed") — the stale-shim trap this module exists |
| 157 | # to name. |
| 158 | _MANAGER_PRESCRIPTIONS: Dict[str, Dict[str, Tuple[str, str]]] = { |
| 159 | "yt-dlp": { |
| 160 | "brew": ("brew install yt-dlp", "brew reinstall yt-dlp"), |
| 161 | "pipx": ("pipx install yt-dlp", "pipx reinstall yt-dlp"), |
| 162 | }, |
| 163 | "node": { |
| 164 | "brew": ("brew install node", "brew reinstall node"), |
| 165 | "nvm": ("nvm install --lts", "reinstall node via nvm: nvm install --lts && nvm use --lts"), |
| 166 | }, |
| 167 | "ffmpeg": { |
| 168 | "brew": ("brew install ffmpeg", "brew reinstall ffmpeg"), |
| 169 | "apt": ("sudo apt-get install -y ffmpeg", "sudo apt-get install -y --reinstall ffmpeg"), |
| 170 | }, |
| 171 | # The official installer is the primary path; npm is a real alternative |
| 172 | # (the package is published as @xai-official/grok) and fits the existing |
| 173 | # manager-preference machinery. |
| 174 | "grok": { |
| 175 | "npm": ( |
| 176 | "npm install -g @xai-official/grok", |
| 177 | "reinstall the Grok CLI: npm install -g @xai-official/grok@latest", |
| 178 | ), |
| 179 | }, |
| 180 | } |
| 181 | |
| 182 | # Last-resort prescriptions when no known package manager is detected. |
| 183 | _FALLBACK_PRESCRIPTIONS: Dict[str, Tuple[str, str]] = { |
| 184 | "yt-dlp": ( |
| 185 | "install yt-dlp (https://github.com/yt-dlp/yt-dlp#installation) and ensure it is on PATH", |
| 186 | "reinstall yt-dlp (https://github.com/yt-dlp/yt-dlp#installation); the current binary won't run", |
| 187 | ), |
| 188 | "node": ( |
| 189 | "install Node.js 22+ (https://nodejs.org) and ensure `node` is on PATH", |
| 190 | "reinstall Node.js 22+ (https://nodejs.org); the current binary won't run", |
| 191 | ), |
| 192 | "ffmpeg": ( |
| 193 | "install ffmpeg (https://ffmpeg.org/download.html) and ensure it is on PATH", |
| 194 | "reinstall ffmpeg (https://ffmpeg.org/download.html); the current binary won't run", |
| 195 | ), |
| 196 | "grok": ( |
| 197 | "install the Grok CLI: curl -fsSL https://x.ai/cli/install.sh | bash, then run `grok login`", |
| 198 | "reinstall the Grok CLI: curl -fsSL https://x.ai/cli/install.sh | bash; the current binary won't run", |
| 199 | ), |
| 200 | } |
| 201 | |
| 202 | |
| 203 | @dataclass |
| 204 | class DependencyProbe: |
| 205 | """Uniform probe result for one external dependency. |
| 206 | |
| 207 | ``status`` is one of the module-level constants (OK/MISSING/BROKEN/TIMEOUT). |
| 208 | ``detail`` says what was observed (version string, exec error, off-PATH |
| 209 | location). ``prescription`` is the copy-pasteable fix, empty when OK. |
| 210 | ``owner_pkg_manager`` names the manager the prescription targets |
| 211 | ("brew", "pipx", "apt", "nvm", "npx"), or "" for PATH fixes / fallbacks. |
| 212 | """ |
| 213 | |
| 214 | name: str |
| 215 | status: str |
| 216 | detail: str = "" |
| 217 | prescription: str = "" |
| 218 | owner_pkg_manager: str = "" |
| 219 | # True for the on-disk-but-off-PATH case: MISSING (the engine gate would |
| 220 | # not pass) but the fix is a PATH edit, not an install. |
| 221 | off_path: bool = False |
| 222 | |
| 223 | @property |
| 224 | def ok(self) -> bool: |
| 225 | return self.status == OK |
| 226 | |
| 227 | |
| 228 | # Safe under the GIL (dict get/set are atomic) and each dependency name is |
| 229 | # probed from a single builder today; worst case is one redundant probe. |
| 230 | _dependency_probe_cache: Dict[str, DependencyProbe] = {} |
| 231 | |
| 232 | |
| 233 | def clear_dependency_probe_cache() -> None: |
| 234 | """Reset memoized probes (tests, or a doctor re-run after a fix).""" |
| 235 | _dependency_probe_cache.clear() |
| 236 | |
| 237 | |
| 238 | def _nvm_present() -> bool: |
| 239 | return bool(os.environ.get("NVM_DIR")) or (Path.home() / ".nvm").is_dir() |
| 240 | |
| 241 | |
| 242 | def _manager_available(manager: str) -> bool: |
| 243 | if manager == "nvm": |
| 244 | return _nvm_present() |
| 245 | if manager == "apt": |
| 246 | return shutil.which("apt-get") is not None |
| 247 | return shutil.which(manager) is not None |
| 248 | |
| 249 | |
| 250 | def _is_pp_cli(name: str) -> bool: |
| 251 | return name.endswith(_PP_CLI_SUFFIX) and len(name) > len(_PP_CLI_SUFFIX) |
| 252 | |
| 253 | |
| 254 | def _pp_install_cmd(name: str) -> str: |
| 255 | slug = name[: -len(_PP_CLI_SUFFIX)] |
| 256 | return f"npx -y {_PRINTING_PRESS_NPM} install {slug} --cli-only" |
| 257 | |
| 258 | |
| 259 | def pp_install_cmd(slug: str) -> str: |
| 260 | """Public catalog-install command for the Printing Press CLI ``<slug>-pp-cli``.""" |
| 261 | return _pp_install_cmd(f"{slug}{_PP_CLI_SUFFIX}") |
| 262 | |
| 263 | |
| 264 | def static_prescription(name: str, manager: str) -> Tuple[str, str]: |
| 265 | """Public ``(install, reinstall)`` strings for one dependency/manager pair. |
| 266 | |
| 267 | Reads the static table without probing manager availability; raises |
| 268 | KeyError for unknown pairs so consumers fail loudly at import time. |
| 269 | """ |
| 270 | return _MANAGER_PRESCRIPTIONS[name][manager] |
| 271 | |
| 272 | |
| 273 | def _prescription(name: str, kind: str) -> Tuple[str, str]: |
| 274 | """Return ``(prescription, owner_pkg_manager)`` for install/reinstall. |
| 275 | |
| 276 | ``kind`` is "install" (MISSING) or "reinstall" (BROKEN). Printing Press |
| 277 | CLIs always re-run the catalog installer; other deps pick the first |
| 278 | detected manager from their preference table, falling back to a generic |
| 279 | but still actionable instruction. |
| 280 | """ |
| 281 | idx = 0 if kind == "install" else 1 |
| 282 | if _is_pp_cli(name): |
| 283 | cmd = _pp_install_cmd(name) |
| 284 | if kind == "reinstall": |
| 285 | return f"re-run the Printing Press install: {cmd}", "npx" |
| 286 | return cmd, "npx" |
| 287 | for manager, prescriptions in _MANAGER_PRESCRIPTIONS.get(name, {}).items(): |
| 288 | if _manager_available(manager): |
| 289 | return prescriptions[idx], manager |
| 290 | fallback = _FALLBACK_PRESCRIPTIONS.get(name) |
| 291 | if fallback: |
| 292 | return fallback[idx], "" |
| 293 | verb = "install" if kind == "install" else "reinstall" |
| 294 | return f"{verb} {name} and ensure it is on PATH", "" |
| 295 | |
| 296 | |
| 297 | def windows_printing_press_bin_dir() -> Optional[Path]: |
| 298 | """Windows managed install dir for Printing Press CLIs, when applicable. |
| 299 | |
| 300 | Returns ``%LOCALAPPDATA%/Programs/PrintingPress/bin`` on Windows when |
| 301 | LOCALAPPDATA is set; ``None`` otherwise. |
| 302 | """ |
| 303 | if os.name != "nt": |
| 304 | return None |
| 305 | local_app = os.environ.get("LOCALAPPDATA") or os.environ.get("LocalAppData") |
| 306 | if not local_app: |
| 307 | return None |
| 308 | return Path(local_app) / "Programs" / "PrintingPress" / "bin" |
| 309 | |
| 310 | |
| 311 | def installer_bin_dirs() -> List[Path]: |
| 312 | """Installer-managed bin dirs shared with setup_wizard's Digg candidates. |
| 313 | |
| 314 | Single source of truth for where installers drop binaries: the Printing |
| 315 | Press library default (~/.local/bin), Go bins, and — on Windows — the |
| 316 | managed %LOCALAPPDATA%/Programs/PrintingPress/bin dir. |
| 317 | ``setup_wizard._digg_bin_candidate_paths`` derives its Digg-specific |
| 318 | paths from this list; keep the two in lockstep by editing only here. |
| 319 | """ |
| 320 | home = Path.home() |
| 321 | dirs = [home / ".local" / "bin"] |
| 322 | gopath = os.environ.get("GOPATH") |
| 323 | if gopath: |
| 324 | dirs.append(Path(gopath) / "bin") |
| 325 | dirs.append(home / "go" / "bin") |
| 326 | win_dir = windows_printing_press_bin_dir() |
| 327 | if win_dir is not None: |
| 328 | dirs.append(win_dir) |
| 329 | return dirs |
| 330 | |
| 331 | |
| 332 | def _off_path_candidate_dirs() -> List[Path]: |
| 333 | """Directories where installers drop binaries that PATH may not cover. |
| 334 | |
| 335 | The shared installer dirs (``installer_bin_dirs``, which also backs |
| 336 | setup_wizard's Digg candidates) plus the Homebrew prefixes (an agent |
| 337 | subprocess PATH sometimes omits even those). |
| 338 | """ |
| 339 | dirs = installer_bin_dirs() |
| 340 | dirs.extend([Path("/opt/homebrew/bin"), Path("/usr/local/bin")]) |
| 341 | return dirs |
| 342 | |
| 343 | |
| 344 | def _off_path_binary(name: str) -> Optional[Path]: |
| 345 | """Return an executable for ``name`` in a known dir that PATH misses.""" |
| 346 | names = [name, f"{name}.exe"] if os.name == "nt" else [name] |
| 347 | for directory in _off_path_candidate_dirs(): |
| 348 | for candidate_name in names: |
| 349 | candidate = directory / candidate_name |
| 350 | if candidate.is_file() and os.access(candidate, os.X_OK): |
| 351 | return candidate |
| 352 | return None |
| 353 | |
| 354 | |
| 355 | def _path_hint(directory: Path) -> str: |
| 356 | """Render a bin dir with $HOME substituted for copy-pasteable PATH edits.""" |
| 357 | raw = str(directory) |
| 358 | if os.name == "nt": |
| 359 | return raw |
| 360 | home = str(Path.home()) |
| 361 | if raw == home: |
| 362 | return "$HOME" |
| 363 | if raw.startswith(home + os.sep): |
| 364 | return "$HOME/" + raw[len(home) + 1:].replace(os.sep, "/") |
| 365 | return raw |
| 366 | |
| 367 | |
| 368 | def probe_dependency(name: str, timeout: float = PROBE_TIMEOUT) -> DependencyProbe: |
| 369 | """Probe one external dependency: OK | MISSING | BROKEN | TIMEOUT. |
| 370 | |
| 371 | - MISSING: not resolvable on this process's PATH. If the binary exists in |
| 372 | a known install dir, the prescription is a PATH edit, not an install — |
| 373 | installing again would not fix anything. |
| 374 | - BROKEN: shutil.which resolves it but a cheap version exec fails |
| 375 | (OSError/exec-format, or any non-zero exit). Prescription says |
| 376 | *reinstall* — the #692 stale-shim class must never read as available. |
| 377 | - TIMEOUT: the version exec exceeded the per-probe budget. |
| 378 | - OK: version exec exited 0; ``detail`` carries the version line. |
| 379 | |
| 380 | Memoized per process; ``clear_dependency_probe_cache()`` resets. |
| 381 | """ |
| 382 | cached = _dependency_probe_cache.get(name) |
| 383 | if cached is not None: |
| 384 | return cached |
| 385 | probe = _probe_dependency_uncached(name, timeout) |
| 386 | _dependency_probe_cache[name] = probe |
| 387 | return probe |
| 388 | |
| 389 | |
| 390 | def _probe_dependency_uncached(name: str, timeout: float) -> DependencyProbe: |
| 391 | resolved = shutil.which(name) |
| 392 | if resolved is None: |
| 393 | off_path = _off_path_binary(name) |
| 394 | if off_path is not None: |
| 395 | hint = _path_hint(off_path.parent) |
| 396 | return DependencyProbe( |
| 397 | name=name, |
| 398 | status=MISSING, |
| 399 | detail=f"{name} is installed at {off_path} but that directory is not on this process's PATH", |
| 400 | prescription=f'add {hint} to PATH (e.g. export PATH="{hint}:$PATH") so {name} resolves', |
| 401 | owner_pkg_manager="", |
| 402 | off_path=True, |
| 403 | ) |
| 404 | prescription, manager = _prescription(name, "install") |
| 405 | return DependencyProbe( |
| 406 | name=name, |
| 407 | status=MISSING, |
| 408 | detail=f"{name} not found on PATH", |
| 409 | prescription=prescription, |
| 410 | owner_pkg_manager=manager, |
| 411 | ) |
| 412 | |
| 413 | command = [name] + _VERSION_ARGS.get(name, ["--version"]) |
| 414 | try: |
| 415 | proc = subprocess.run( |
| 416 | command, |
| 417 | capture_output=True, |
| 418 | text=True, |
| 419 | timeout=timeout, |
| 420 | ) |
| 421 | except (FileNotFoundError, OSError) as exc: |
| 422 | prescription, manager = _prescription(name, "reinstall") |
| 423 | return DependencyProbe( |
| 424 | name=name, |
| 425 | status=BROKEN, |
| 426 | detail=f"{name} resolves to {resolved} but won't execute: {exc}", |
| 427 | prescription=prescription, |
| 428 | owner_pkg_manager=manager, |
| 429 | ) |
| 430 | except subprocess.TimeoutExpired: |
| 431 | prescription, manager = _prescription(name, "reinstall") |
| 432 | return DependencyProbe( |
| 433 | name=name, |
| 434 | status=TIMEOUT, |
| 435 | detail=f"{name} version probe timed out after {timeout:g}s", |
| 436 | prescription=f"re-run doctor; if the timeout persists: {prescription}", |
| 437 | owner_pkg_manager=manager, |
| 438 | ) |
| 439 | |
| 440 | if proc.returncode == 0: |
| 441 | lines = (proc.stdout or proc.stderr or "").strip().splitlines() |
| 442 | version = lines[0].strip() if lines else "" |
| 443 | return DependencyProbe(name=name, status=OK, detail=version) |
| 444 | |
| 445 | lines = (proc.stderr or proc.stdout or "").strip().splitlines() |
| 446 | why = lines[0].strip() if lines else f"exit {proc.returncode}" |
| 447 | prescription, manager = _prescription(name, "reinstall") |
| 448 | return DependencyProbe( |
| 449 | name=name, |
| 450 | status=BROKEN, |
| 451 | detail=f"{name} resolves to {resolved} but the version probe failed: {why}", |
| 452 | prescription=prescription, |
| 453 | owner_pkg_manager=manager, |
| 454 | ) |
| 455 | |
| 456 | |
| 457 | def probe_dependencies(names: Optional[Iterable[str]] = None) -> Dict[str, DependencyProbe]: |
| 458 | """Probe every known dependency (or ``names``), memoized per process.""" |
| 459 | return {name: probe_dependency(name) for name in (names or KNOWN_DEPENDENCIES)} |
| 460 |