"""Read-only environment preflight with actionable, never automatic repairs.""" from __future__ import annotations from dataclasses import replace import os from pathlib import Path import socket from .common import now from .environment import execution_environment, resolve_tools, tool_version from .workspace import inspect_repository, load_project, selected_repositories from .process import require_capture def _required_tools(project, workspace_root, selected, *, filtered, profile=None): """Use the catalog's selected commands, including dependencies, without running them.""" from .catalog import _custom_stages if profile is None: # Doctor covers all declared profiles by default, or all declared checks # when the project only uses the context/doctor commands so far. profiles = project.config.get("profiles", {}) records = project.config.get("checks", []) if not isinstance(profiles, dict) or not isinstance(records, list): raise ValueError( "Project profiles and checks must have their declared shapes" ) identities = [] if profiles: for values in profiles.values(): if not isinstance(values, list) or any( not isinstance(value, str) for value in values ): raise ValueError("A project profile must be a list of check IDs") identities.extend(values) else: if any(not isinstance(record, dict) for record in records): raise ValueError("Project checks must be objects") identities = [record.get("id") for record in records] if any(not isinstance(identity, str) for identity in identities): raise ValueError("Project checks must have string IDs") project = replace( project, config={ **project.config, "profiles": {"quick": list(dict.fromkeys(identities))}, }, ) profile = "quick" stages = _custom_stages(project, workspace_root, profile, selected, filtered) # Python is required by the runner and its environment fingerprint even if # all selected check commands use another interpreter. Explicitly configured # tools also declare requirements for commands hidden inside project scripts. required = {"python", *project.config.get("tools", {})} for stage in stages: argv = stage["argv"] for name in ("node", "npm", "python"): if any("{" + name + "}" in value for value in argv): required.add(name) executable = Path(argv[0]).name if executable in {"node", "nodejs"}: required.add("node") elif executable in {"npm", "npx"}: required.update({"npm", "node"}) if "npm" in required: required.add("node") return required def diagnose(args) -> dict: project = load_project(args.workspace_root, args.project) tools = resolve_tools(args.workspace_root, project) env = execution_environment(args.workspace_root, project, tools) selected = selected_repositories(project, args.repo) required = ( _required_tools( project, args.workspace_root, selected, filtered=bool(args.repo), profile=getattr(args, "profile", None), ) if args.project else set(tools) ) checks = [] for name, executable in tools.items(): if name not in required: checks.append( { "id": name, "status": "not_required", "detail": "Not required by the selected declared project checks; not probed.", "path": executable, "repair": "For indirect dependencies inside scripts, declare the tool under project tools.", } ) continue version = tool_version(executable, env) checks.append( { "id": name, "status": "passed" if version != "unavailable" else "blocked", "detail": version, "path": executable, "repair": f"Install/configure {name}; set {name.upper()} or project tools.{name}. No installation was attempted.", } ) for repo in selected: snapshot = inspect_repository(repo) checks.append( { "id": repo.name, "status": "blocked" if snapshot["errors"] else "passed", "detail": "; ".join(snapshot["errors"]) or "Git checkout readable (dirty work is allowed).", "repair": "Use tools/repo/bootstrap-repositories.py --check; review missing checkouts before cloning.", } ) for package_dir in (repo.path, repo.path / "webui"): if (package_dir / "package.json").is_file(): present = (package_dir / "node_modules").is_dir() checks.append( { "id": f"dependencies:{repo.name}:{package_dir.name}", "status": "passed" if present else "warning", "detail": "Dependency directory exists; availability is not a full dependency audit." if present else "No local node_modules directory; workspace-hoisted packages may still resolve.", "repair": "Inspect the owning package lock and install instructions before running npm ci.", } ) if not args.project: script = args.workspace_root / "govoplan/tools/repo/sync-python-environment.py" if script.is_file(): result = require_capture( [ tools["python"], str(script), "--check", "--requirements", str(args.workspace_root / "govoplan/requirements-dev.txt"), "--python", tools["python"], ], timeout=60, max_stdout=65536, env=env, ) checks.append( { "id": "python-environment-sync", "status": "passed" if result.returncode == 0 else "warning", "detail": "Environment synchronization fingerprint is current." if result.returncode == 0 else "Environment sync check did not pass; inspect its --dry-run output.", "repair": "./.venv/bin/python tools/repo/sync-python-environment.py --dry-run --requirements requirements-dev.txt --python ./.venv/bin/python", } ) browser_roots = [ Path.home() / ".cache/ms-playwright", Path.home() / ".var/app/com.vscodium.codium/cache/ms-playwright", ] explicit_browser = os.environ.get("PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH") browser_available = ( (Path(explicit_browser).is_file() and os.access(explicit_browser, os.X_OK)) if explicit_browser else any( root.is_dir() and any(root.glob("chromium-*/chrome-linux*/chrome")) for root in browser_roots ) ) checks.append( { "id": "browser", "status": "passed" if browser_available else "warning", "detail": "Configured/cached Chromium found." if browser_available else "Chromium executable not found; check PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH or install the pinned Playwright browser.", "repair": "Follow Core WebUI conformance setup; do not start a development server to repair this.", } ) with socket.socket() as probe: probe.settimeout(0.25) busy = probe.connect_ex(("127.0.0.1", 4174)) == 0 checks.append( { "id": "browser-test-port", "status": "warning" if busy else "passed", "detail": "Port 4174 is occupied; do not kill another run." if busy else "Port 4174 is free now (not a reservation).", "repair": "Wait for the owning check run; use devkit resource coordination.", } ) summary = [ f"Environment preflight: {sum(item['status'] == 'blocked' for item in checks)} blockers, {sum(item['status'] == 'warning' for item in checks)} warnings." ] summary.extend( f"{item['id']}: {item['status']} — {item['detail']}" for item in checks if item["status"] != "passed" ) summary.append( "Read-only: no installations, configuration repairs or servers started." ) if args.project: summary.append( "Required tools follow declared checks and explicit project tools; indirect script dependencies must be declared explicitly." ) return { "schema_version": 1, "generated_at": now(), "checks": checks, "required_tools": sorted(required), "summary": summary, "_exit_code": 1 if any(item["status"] == "blocked" for item in checks) else 0, } def register(subparsers): parser = subparsers.add_parser( "doctor", help="Read-only environment preflight and exact repair guidance" ) parser.add_argument("--repo", action="append", default=[]) parser.add_argument( "--profile", choices=("quick", "ui", "backend", "full"), help="For portable projects, limit tool requirements to this check profile (default: all declared profiles)", ) parser.set_defaults(handler=diagnose)