5e4a2a3d24
Competing installers and checkout-local venv assumptions bypassed PM selection, install consent, and generation lifetimes. Route consumers through PM and installation-bound launchers. Refresh source launchers before obsolete Python entries can be collected. Remove Node, browser, and CUA acquisition engines, obsolete venv-holder handling, detached sync, and unused PM APIs. Keep historical updater exports inert and preserve external tool ownership and native integration. Share product freshness and prepared inputs across builders. Align plugin admission, Docker provisioning, setup instructions, and behavioral tests. Verified targeted Python and JavaScript tests, desktop and web typechecks, scoped lint, real product builds, and the Docker frontend smoke test. The missed post-setup test cleanup is included and verified. Native Windows/macOS execution, full Rust compilation, and the complete repository suite remain unverified. Historical compatibility requirements were preserved and extended, not fully rescanned.
87 lines
4.2 KiB
Python
87 lines
4.2 KiB
Python
"""Source import-integrity checks for update and stash restoration."""
|
|
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from hermes_cli._launchers import runtime_command
|
|
|
|
# Modules imported on every startup. Unlike _UPDATE_CRITICAL_FILES (only parsed) these are
|
|
# *imported*, catching cross-module breakage (a name pulled from a sibling no longer exists).
|
|
_UPDATE_CRITICAL_MODULES = "hermes_cli.main", "run_agent", "model_tools", "toolsets"
|
|
|
|
|
|
def _critical_module_import_failures(
|
|
root, *, report_runtime_errors: bool = False) -> dict[str, tuple[str, str]]:
|
|
"""Import each ``_UPDATE_CRITICAL_MODULES`` entry in a subprocess; return failures in probe order.
|
|
|
|
Syntax validation only *parses*: a partially-updated tree (Windows ZIP copy loop) parses yet
|
|
dies with ``ImportError: cannot import name``. The boot-selected subprocess
|
|
keeps import side effects out of the updater's ``sys.modules``.
|
|
Generic import-time exceptions are tolerated unless ``report_runtime_errors=True``.
|
|
"""
|
|
from hermes_cli.update_cmd import _UPDATE_CRITICAL_MODULES
|
|
from hermes_constants import FIRST_PARTY_MODULE_ROOTS
|
|
import secrets
|
|
marker = f"__HERMES_IMPORT_HEALTH_{secrets.token_hex(16)}__"
|
|
probe = (
|
|
"import importlib, json, sys\n"
|
|
"failures = []\n"
|
|
"for name in %r:\n"
|
|
" try:\n"
|
|
" importlib.import_module(name)\n"
|
|
" except ModuleNotFoundError as exc:\n"
|
|
# A missing *third-party* module means deps aren't installed, not a skewed checkout;
|
|
# only our own packages count. Roots come from hermes_constants so the user hint can't drift.
|
|
" missing = (getattr(exc, 'name', '') or '').split('.')[0]\n"
|
|
" if missing in %r or missing.startswith('hermes_') or %r:\n"
|
|
" failures.append((name, type(exc).__name__, str(exc)))\n"
|
|
" except ImportError as exc:\n"
|
|
" failures.append((name, type(exc).__name__, str(exc)))\n"
|
|
" except Exception as exc:\n"
|
|
" if %r:\n"
|
|
" failures.append((name, type(exc).__name__, str(exc)))\n"
|
|
" except BaseException as exc:\n"
|
|
" failures.append((name, type(exc).__name__, str(exc)))\n"
|
|
"sys.stdout.write('\\n%s' + json.dumps(failures))\n"
|
|
% (_UPDATE_CRITICAL_MODULES, tuple(sorted(FIRST_PARTY_MODULE_ROOTS)), report_runtime_errors,
|
|
report_runtime_errors, marker))
|
|
try:
|
|
result = subprocess.run(
|
|
runtime_command(Path(root), code=probe), cwd=str(root), capture_output=True, text=True,
|
|
encoding="utf-8", errors="replace", timeout=120)
|
|
except subprocess.TimeoutExpired:
|
|
return _probe_failure("TimeoutExpired", "timed out before reporting import health")
|
|
except (OSError, subprocess.SubprocessError):
|
|
# Can't run the probe — don't block the update on our own tooling.
|
|
return {}
|
|
output = result.stdout or ""
|
|
if marker not in output:
|
|
return _probe_failure(
|
|
"ProbeTerminated",
|
|
f"terminated before reporting import health (exit code {result.returncode})")
|
|
try:
|
|
failures = json.loads(output.rsplit(marker, 1)[1])
|
|
if not isinstance(failures, list) or any(
|
|
not isinstance(item, list) or len(item) != 3 or not all(isinstance(v, str) for v in item)
|
|
for item in failures):
|
|
raise ValueError("invalid import-health payload")
|
|
return {str(module): (str(kind), str(detail)) for module, kind, detail in failures}
|
|
except (TypeError, ValueError):
|
|
return _probe_failure("MalformedPayload", "reported malformed import health data")
|
|
|
|
|
|
def _probe_failure(kind: str, detail: str) -> dict[str, tuple[str, str]]:
|
|
"""Failure row for the probe itself (as opposed to a module it imported)."""
|
|
return {"critical-module probe": (kind, detail)}
|
|
|
|
|
|
def _validate_critical_modules_import(
|
|
root, *, report_runtime_errors: bool = False) -> tuple[bool, str | None, str | None]:
|
|
"""Return the first critical-module import failure, if any."""
|
|
failures = _critical_module_import_failures(root, report_runtime_errors=report_runtime_errors)
|
|
if failures:
|
|
module = next(iter(failures))
|
|
return False, module, failures[module][1]
|
|
return True, None, None
|
|
|