"""ensure(): make the installed state match the lockfile for a package, and hand back its composed environment.""" from __future__ import annotations import json import logging import os import shutil import threading from contextlib import ExitStack, contextmanager, nullcontext from dataclasses import dataclass from pathlib import Path from typing import Optional from pm import paths from pm.downloader import DownloadPaused, ProgressFn from pm.lock import Facts, Lockfile from pm.package import InstallError, Package, Runner, StatePackage, compose_env from pm.plugin_inputs import Candidates, Members, PluginInput, Selection, StagedUpdate from pm.registry import get_package, walk from pm.store import Store, current_target, merge_tree, tree_digest LOG = logging.getLogger(__name__) # ``progress(stage, done, total, label)`` reports download/unpack/verify; # multi-archive labels follow lockfile order. def _prepare_artifacts(package, store, scratch, artifacts, version, target, *, progress=None, pause_event=None, download_progress=None): def tick(done, total, ranges): if progress is not None: active = next(reversed(ranges)) index = next(i for i, artifact in enumerate(artifacts) if artifact["url"] == active) label = f"{index + 1}/{len(artifacts)}" if len(artifacts) > 1 else "" progress("download", done, total, label) if download_progress is not None: download_progress(done, total, ranges) staged = scratch / "tree" archives = store.fetch_many(artifacts, scratch, progress=tick, pause_event=pause_event) for index, archive in enumerate(archives): if pause_event is not None and pause_event.is_set(): raise DownloadPaused("install paused") if progress is not None: label = f"{index + 1}/{len(artifacts)}" if len(artifacts) > 1 else "" progress("unpack", 0, 0, label) # unpack empties its destination; extract additional archives separately. destination = staged if index == 0 else scratch / f"extra-{index}" package.unpack(archive, destination, target) if index: merge_tree(destination, staged) package.stage(store, staged, version, target) return staged def _lockfile() -> Lockfile: return Lockfile(paths.lockfile_path()) def _facts() -> Facts: return Facts(paths.facts_path()) def _store() -> Store: return Store(paths.store_root()) def _heal_exec_bit(binary: Path) -> bool: """agent-browser entries staged before stage() set the exec bit sit at 0644 and fail every launch with PermissionError. Modes are not part of the pinned digest, so restore it in place; an entry we cannot chmod (sealed store) is treated as not installed.""" if os.name == "nt" or os.access(binary, os.X_OK): return True try: binary.chmod(binary.stat().st_mode | 0o111) except OSError: return False return os.access(binary, os.X_OK) def _installed_location(package: Package, lockfile: Lockfile, target: str, *, verify: bool = False, allow_outdated: bool = False, roots: tuple[Path, ...] | None = None): """Prefer the current pin. An explicit read may retain a prior PM install.""" search_roots = dict.fromkeys(roots if roots is not None else (paths.store_root(), paths.writable_store_root())) fallback = None for root in search_roots: store = Store(root) facts = _facts() if root == paths.store_root() else Facts(root / "facts.json") fact = facts.get(package.name) if not fact or not facts.installed(package.name, None, root): continue binary = package.binary(store.entry(fact["entry"]), target) if binary is not None and not binary.is_file(): continue if binary is not None and target == current_target() and not _heal_exec_bit(binary): continue if verify and not _entry_verified(package, fact, store, target): continue if facts.installed(package.name, lockfile.version(package.name), root, _identity(lockfile, package.name, target)): return facts, store if (allow_outdated and fact.get("target") == target and fact.get("artifacts") and fact.get("digest")): fallback = facts, store return fallback @dataclass(frozen=True) class InstalledPackage: path: Path version: str binary: Path | None def installed_package(name: str, *, allow_outdated: bool = False) -> InstalledPackage | None: """Read the selected PM entry without installing or changing its facts.""" package = get_package(name) if package.internal: raise ValueError(f"{name} is internal PM tooling, not an application package") target = current_target() location = _installed_location(package, _lockfile(), target, allow_outdated=allow_outdated) if location is None: return None facts, store = location fact = facts.get(name) entry = store.entry(fact["entry"]) return InstalledPackage(entry, fact["version"], package.binary(entry, target)) def uv_launcher(name: str) -> Path | None: """PM's installed ``uv``/``uvx`` for a user-declared MCP stdio ``command:``, so a bare ``uvx`` server runs the packaged uv, never the user's. Read-only; Hermes's own Python work still goes through PM operations, not this executable.""" if name not in ("uv", "uvx"): raise ValueError(f"{name!r} is not a uv launcher") package = get_package("uv") target = current_target() location = _installed_location(package, _lockfile(), target) if location is None: return None facts, store = location binary = package.binary(store.entry(facts.get("uv")["entry"]), target) launcher = binary.with_name(name + binary.suffix) if binary is not None else None return launcher if launcher is not None and launcher.is_file() else None def _identity(lockfile: Lockfile, name: str, target: str): """The identity the lock currently pins for `name` on `target`: (target, tuple(artifact sha256s)) — or None when the lock pins no artifacts (nothing digest-bound to compare).""" artifacts = lockfile.artifacts(name, target) if not artifacts: return None return (target, tuple(a["sha256"] for a in artifacts)) def lazy_installs_allowed() -> bool: """Policy: may pm install things on demand right now? HERMES_DISABLE_LAZY_INSTALLS is an internal bridge var set by the official Docker image and the hermetic test harness. The user-facing setting is security.allow_lazy_installs in config.yaml; a config system that fails to load counts as ALLOWED only when hermes_cli is genuinely absent (bootstrap) — config errors fail closed. """ import os if os.environ.get("HERMES_DISABLE_LAZY_INSTALLS", "").strip().lower() in ( "1", "true", "yes", ): return False try: from hermes_cli.config import cfg_get, load_config_readonly, require_readable_config_before_write except ModuleNotFoundError as exc: return exc.name in {"hermes_cli", "hermes_cli.config"} except ImportError: return False try: # The normal loader falls back to defaults on invalid YAML. A broken # security policy must not grant permission to acquire dependencies. require_readable_config_before_write() return cfg_get(load_config_readonly(), "security", "allow_lazy_installs", default=True) is True except Exception: return False def enabled_extras() -> list[str]: """The venv extras recorded in the installed state.""" fact = Facts(paths.runtime_facts_path()).get("venv") or _facts().get("venv") or {} return list(fact.get("extras", [])) def is_installed(name: str) -> bool: return _installed_location(get_package(name), _lockfile(), current_target()) is not None def sealed() -> bool: """Sealed = bundled LAYOUT only: the store sits beside the bundle manifest and is read-only as shipped.""" return (paths.store_root().parent / "manifest.json").is_file() def _refuse_lazy(name: str, what: str) -> InstallError: from pm import receipt error = InstallError( name, f"not installed and lazy installs are disabled: {what}", "enable security.allow_lazy_installs or run `hermes pm install`", ) receipt.record_refusal("lazy-install", str(error)) return error def _remove_entry(store: Store, entry_name: str) -> None: """Remove a replaced or failed entry, retrying transient Windows holds. Corruption may leave a file where the directory belonged. Failure must propagate so recovery never claims to have removed surviving bytes. """ import time entry = store.entry(entry_name) for attempt in range(5): try: if entry.is_symlink() or not entry.is_dir(): entry.unlink(missing_ok=True) else: shutil.rmtree(entry) return except FileNotFoundError: return except OSError as e: if attempt == 4: raise time.sleep(0.2 * (attempt + 1)) def _remove_downloads(store: Store, artifacts: list[dict]) -> None: """Release this package's archives after publication, under its store lock.""" for artifact in artifacts: _remove_entry(store, f"fetch-{artifact['sha256']}") def _entry_verified(package: Package, fact: dict, store: Store, target: str) -> bool: """Explicit installs re-check realized bytes; startup keeps its cheap facts check.""" entry = store.entry(fact["entry"]) try: return not package.verify(entry, target) and fact.get("digest") == tree_digest(entry) except OSError: return False def _restore_previous_entry(store: Store, entry, previous) -> None: """Keep both versions recoverable until the restore rename succeeds.""" import uuid displaced = store.entry(f".displaced-{uuid.uuid4().hex}") had_entry = entry.exists() or entry.is_symlink() if had_entry: entry.rename(displaced) try: previous.rename(entry) except BaseException: if had_entry: displaced.rename(entry) raise if had_entry: _remove_entry(store, displaced.name) @contextmanager def _publish_entry(package, store, staged, entry, previous_entry, target): """Keep rollback live through the caller's native facts commit, if any.""" if entry.exists() or entry.is_symlink(): entry.rename(previous_entry) try: store.publish(staged, entry.name) reason = package.verify(entry, target) if reason: raise InstallError(package.name, f"published entry failed verification: {reason}") yield except BaseException: if previous_entry.exists(): _restore_previous_entry(store, entry, previous_entry) raise if previous_entry.exists(): _remove_entry(store, previous_entry.name) def _settle_previous_entry(package, store, entry, previous_entry, previous, target) -> None: """Finish or undo a publication an earlier install left behind.""" if not previous_entry.exists(): return # Facts commit last. Stages have no host-side commit record, so # an interrupted stage always restores its prior usable bytes. if (previous and previous.get("entry") == entry.name and _entry_verified(package, previous, store, target)): _remove_entry(store, previous_entry.name) else: _restore_previous_entry(store, entry, previous_entry) def _entry_current(package, lockfile, facts, store, entry, previous, version, pin, target) -> bool: if facts is not None: return previous is not None and facts.installed( package.name, version, store.root, _identity(lockfile, package.name, target) ) and _entry_verified(package, previous, store, target) try: recorded = (entry / ".pm-stage-pin.json").read_text(encoding="utf-8-sig") except OSError: recorded = None return recorded == pin and not package.verify(entry, target) def _copy_verified_source(package, lockfile, copy_from, staged, version, target) -> None: source_facts, source_store = copy_from source = source_facts.get(package.name) if (not source_facts.installed(package.name, version, source_store.root, _identity(lockfile, package.name, target)) or not _entry_verified(package, source, source_store, target)): raise InstallError(package.name, "bundled copy source failed verification") shutil.copytree(source_store.entry(source["entry"]), staged, symlinks=True) if tree_digest(staged) != source["digest"]: raise InstallError(package.name, "copied bytes do not match the bundled source") def _log_repair(package, previous, version, artifacts) -> None: """Work item 6: replacing an ESTABLISHED fact is a repair — log it, no transaction system, no receipt file.""" if not previous or "entry" not in previous: return old_artifact = (previous.get("artifacts") or ["?"])[0] old = f"{previous.get('version', '?')}/{str(old_artifact)[:12]}" new = f"{version}/{artifacts[0]['sha256'][:12]}" if artifacts else version LOG.info("repair: %s re-realized %s -> %s", package.name, old, new) def _install( package: Package, lockfile: Lockfile, facts: Facts | None, store: Store, target: str, progress=None, pause_event: threading.Event | None = None, download_progress: ProgressFn | None = None, *, copy_from: tuple[Facts, Store] | None = None, _lock_held: bool = False, _fresh_copy: bool = False, ) -> Path: """Realize one pin. Host installs commit facts; cross-target stages carry a marker.""" version = lockfile.version(package.name) if version is None: raise InstallError( package.name, "not in the lockfile", "add it with `hermes pm lock --bump`" ) reason = package.missing_reason(target) if reason is not None: raise InstallError(package.name, f"unavailable on {target}: {reason}", "none") entry_name = package.store_entry(version, target) entry = store.entry(entry_name) if getattr(package, "pin_only", False): return entry artifacts = lockfile.artifacts(package.name, target) pin = json.dumps({"target": target, "sha256": [a["sha256"] for a in artifacts]}) with nullcontext() if _lock_held else store.install_lock(): if pause_event is not None and pause_event.is_set(): raise DownloadPaused("install paused") if facts is not None: facts.reload() previous = facts.get(package.name) if facts is not None else None previous_entry = store.entry(f".previous-{'stage-' if facts is None else ''}{entry_name}") _settle_previous_entry(package, store, entry, previous_entry, previous, target) if (_entry_current(package, lockfile, facts, store, entry, previous, version, pin, target) and not _fresh_copy): _remove_downloads(store, artifacts) return entry if not artifacts: raise InstallError( package.name, f"no artifact for {target} in the lockfile", "run `hermes pm lock --bump` for this package", ) with store.scratch() as scratch: staged = scratch / "tree" try: if copy_from is not None: _copy_verified_source(package, lockfile, copy_from, staged, version, target) else: staged = _prepare_artifacts(package, store, scratch, artifacts, version, target, progress=progress, pause_event=pause_event, download_progress=download_progress) if pause_event is not None and pause_event.is_set(): raise DownloadPaused("install paused") if progress is not None: progress("verify", 0, 0, "") reason = package.verify(staged, target) if reason: raise InstallError(package.name, f"staged entry failed verification: {reason}") if facts is None: (staged / ".pm-stage-pin.json").write_text(pin, encoding="utf-8") with _publish_entry(package, store, staged, entry, previous_entry, target): if facts is not None: facts.record( package.name, version, entry_name, package.env(entry, target), store.root, target=target, artifacts=[a["sha256"] for a in artifacts], digest=tree_digest(entry), ) _remove_downloads(store, artifacts) except (InstallError, DownloadPaused): raise except Exception as e: raise InstallError(package.name, f"install failed: {e}") from e _log_repair(package, previous, version, artifacts) return entry def stage_only( name: str, target: str, progress=None, *, pause_event: threading.Event | None = None, download_progress: ProgressFn | None = None, ) -> Path: """Realize a cross-target pin without host facts or an executable Runner.""" return _install(get_package(name), _lockfile(), None, _store(), target, progress=progress, pause_event=pause_event, download_progress=download_progress) class _InstallOperation: """Validity lasts only while this operation holds the publication lock.""" def __init__(self) -> None: self.stack = ExitStack() self.store: Store | None = None self.checked: set[tuple[str, str | None, str, str]] = set() def lock(self) -> Store: if self.store is None: self.store = Store(paths.writable_store_root()) self.stack.enter_context(self.store.install_lock()) return self.store def close(self) -> None: self.checked.clear() self.store = None self.stack.close() @contextmanager def _install_operation(): operation = _InstallOperation() try: yield operation finally: operation.close() def ensure( name: str, *, base_env: Optional[dict] = None, explicit: bool = False, verify: bool = True, progress=None, pause_event: threading.Event | None = None, download_progress: ProgressFn | None = None, _operation: _InstallOperation | None = None, ) -> Runner: """``explicit`` marks a deliberate install command (`hermes pm install`, `hermes pm bundle`) — those ARE the remedy the lazy-install policy names, so the policy does not apply to them. ``verify`` re-hashes an already-recorded entry and repairs it when the bytes moved. A deliberate install keeps that check. Shell activation passes ``False``. It trusts the recorded digest, the same check startup uses, because hashing every tool tree costs seconds per shell. ``progress(stage, done, total, label)`` reports the slow parts of an install to a UI, including ordered multi-archive labels. """ if isinstance(get_package(name), StatePackage): if _operation is not None: # Python construction can provision tools itself. Drop both the # lock and its validity before entering that independent operation. _operation.close() sync_venv(explicit=explicit) return Runner(name, compose_env([], base=base_env)) lockfile = _lockfile() target = current_target() chain = walk([name]) checked = _operation.checked if _operation is not None else set() if _operation is not None: _operation.lock() missing = [] for package in chain: identity = (package.name, lockfile.version(package.name), target, json.dumps(_identity(lockfile, package.name, target), sort_keys=True)) if identity in checked: continue if _installed_location(package, lockfile, target, verify=explicit and verify) is None: missing.append(package) else: checked.add(identity) if missing and not explicit and not lazy_installs_allowed(): raise _refuse_lazy(name, ", ".join(p.name for p in missing)) if missing: store = _operation.lock() if _operation is not None else Store(paths.writable_store_root()) facts = _facts() if store.root == paths.store_root() else Facts(store.root / "facts.json") for package in missing: # Publication may change entries; do not carry observations across it. checked.clear() _install(package, lockfile, facts, store, target, progress=progress, pause_event=pause_event, download_progress=download_progress, _lock_held=_operation is not None) return Runner(name, env_for(name, base_env=base_env)) def env_for(*names: str, base_env: Optional[dict] = None) -> dict[str, str]: """Composed env of already-installed packages only. Never installs, never raises on missing packages — they contribute nothing.""" lockfile = _lockfile() target = current_target() diffs: list[dict] = [] for name in names: try: chain = walk([name]) except KeyError: continue for package in chain: if package.internal: continue location = _installed_location(package, lockfile, target) if location: facts, store = location diffs.append(facts.env_for(package.name, store.root)) return compose_env(diffs, base=base_env) def _runtime_state_matches(fact: dict, stamp: str, *, project_root: Path | None = None) -> bool: if not isinstance(fact, dict) or fact.get("stamp") != stamp: return False from pm.environments import selected_venv try: environment = selected_venv(paths.repo_root() if project_root is None else project_root) except (OSError, RuntimeError, ValueError): return False recorded = fact.get("environment") if recorded is not None and (not isinstance(recorded, str) or Path(recorded).resolve() != environment): return False return (environment / "pyvenv.cfg").is_file() def _member_inputs(plugins: PluginInput | None) -> dict: """The ``plugin_dirs`` argument of the venv package; empty means config discovery.""" from pm.publication import candidate_members if isinstance(plugins, Candidates): return {"plugin_dirs": candidate_members(plugins.dirs)} if isinstance(plugins, Members): return {"plugin_dirs": plugins.dirs} if plugins is None: return {} raise TypeError(f"{type(plugins).__name__} changes plugin state; only a sync may carry it") def _still_declared(package, recorded: list[str]) -> list[str]: """The recorded extras this tree still declares. An extra the source removed (``hindsight``) would otherwise ride the ledger into every later ``uv sync`` and fail it with "Extra is not defined". Only recorded extras are pruned; an explicitly requested unknown extra still fails. Membership uses PEP 685 names (uv matches ``foo_bar`` to ``foo-bar``); the recorded spelling is what reaches uv. """ import re from pm.features import declared_extras def normalized(name: str) -> str: return re.sub(r"[-_.]+", "-", name).lower() root = package.project_root() if not (root / "pyproject.toml").is_file(): return list(recorded) declared = {normalized(extra) for extra in declared_extras(root)} return [extra for extra in recorded if normalized(extra) in declared] def venv_is_current(*, extras: list[str] | None = None, plugins: Members | Candidates | None = None, project_root: Path | None = None) -> bool: """Probe the requested union without changing recorded dependency state.""" from pm.environments import runtime_facts_path from pm.packages import Venv root = paths.repo_root() if project_root is None else Path(project_root).absolute() package = get_package("venv") if project_root is None else Venv(root) fact = Facts(runtime_facts_path(root), strict=True).get("venv") if fact is None: fact = Facts(paths.facts_path(), strict=True).get("venv") if fact is None: return False if (not isinstance(fact, dict) or not isinstance(fact.get("stamp"), str) or not fact["stamp"] or not isinstance(fact.get("extras"), list) or any(not isinstance(extra, str) for extra in fact["extras"])): raise ValueError("invalid recorded dependency state") enabled = sorted(set(_still_declared(package, fact["extras"])) | set(extras or [])) stamp = package.expected_stamp(enabled, **_member_inputs(plugins)) return _runtime_state_matches(fact, stamp, project_root=root) def _feature_policy(extras: Optional[list[str]], *, repair: bool) -> tuple[list[str] | None, list[str] | None]: """Refuse extras this platform or a frozen bundle cannot carry; return (shipped, frozen).""" from pm.features import read_features if extras: from pm.extras import extra_supported unsupported = [extra for extra in extras if not extra_supported(extra, importable=lambda _: False)] if unsupported: raise InstallError("venv", f"extras {unsupported} are not supported by this Python/platform", "choose a supported provider; no dependency environment was changed") shipped = read_features() frozen = shipped # Without a frozen declaration, explicit source setup needs no policy # read: the config loader initializes/chmods unrelated user state. if frozen is not None and not repair and lazy_installs_allowed(): frozen = None if frozen is not None and extras: outside = sorted(set(extras) - set(frozen)) if outside: raise _refuse_lazy( "venv", f"extras {outside} are outside this bundle's frozen feature " "set (security.allow_lazy_installs is false)", ) return shipped, frozen @contextmanager def _venv_install_lock(*, patient: bool): """Hold the dependency lock, or refuse when an impatient caller would queue.""" from pm import receipt from hermes_cli.runtime_state import INSTALL_LOCK_TIMEOUT_SECONDS, runtime_lock # Holding this lock means rebuilding the whole dependency environment, which takes tens of # seconds on a bundle. Only an install the user asked for may queue for it; an opportunistic # one (a lazy extra at first use, the only non-explicit caller) refuses instead of holding # a sibling profile's backend off its port behind a rebuild it did not request. with runtime_lock(paths.repo_root(), timeout=None if patient else INSTALL_LOCK_TIMEOUT_SECONDS) as held: if not held: error = InstallError( "venv", f"another Hermes process is installing dependencies (waited {INSTALL_LOCK_TIMEOUT_SECONDS:.0f}s)", "retry in a moment, or run `hermes pm install` to install explicitly", ) receipt.record_refusal("install-busy", str(error)) raise error yield def _publication(plugins: PluginInput | None): """Snapshot a plugin state change under the held lock, or None for member-only inputs.""" from pm.publication import PluginSelection, StagedPlugin if isinstance(plugins, Selection): return PluginSelection(dict(plugins.data)) if isinstance(plugins, StagedUpdate): return StagedPlugin(dict(plugins.data)) return None def _publish_inactive(change) -> None: """A disabled plugin's code changes without touching the dependency environment.""" from pm import receipt from hermes_cli.runtime_state import finish_publication, recover_publication try: change.publish(paths.repo_root()) finish_publication(paths.repo_root()) except BaseException: recover_publication(paths.repo_root()) raise receipt.record_venv_rebuild(False, "inactive plugin") def _target_selection(package, fact: dict, *, extras, inputs: dict, repair: bool, shipped, frozen): """Return (enabled extras, expected stamp, package inputs) this sync must reach.""" if repair: if fact and (not isinstance(fact.get("extras"), list) or any(not isinstance(extra, str) for extra in fact["extras"]) or not isinstance(fact.get("stamp"), str) or not fact["stamp"]): raise InstallError("venv", "recorded dependency selection is incomplete; refusing to change its graph") enabled = list(fact.get("extras", frozen if frozen is not None else ["all"])) stamp = fact.get("stamp") or package.expected_stamp(enabled, plugin_dirs=[]) return enabled, stamp, {"repair": True} # The first writable generation replaces, rather than layers on, # the payload. Retain its extras until a recorded selection owns them. enabled = sorted(set(_still_declared(package, fact.get("extras", shipped or []))) | set(extras or [])) return enabled, package.expected_stamp(enabled, **inputs), inputs def _commit_selection(package, facts: Facts, change, *, enabled: list[str], stamp: str, inputs: dict, current: bool, repair: bool, explicit: bool, skip_invalid_secondary: bool = False) -> None: """Build (unless current), publish the plugin change, then record the selection.""" from pm import receipt from hermes_cli.runtime_state import finish_publication, recover_publication try: result = {} if current else (package.apply(enabled, explicit=explicit, skip_invalid_secondary=skip_invalid_secondary, **inputs) or {}) if not repair and package.expected_stamp(enabled, **inputs) != stamp: raise ValueError("Dependency inputs changed while preparing publication; retry.") if change is not None: change.publish(paths.repo_root()) if not current: if result.get("environment") is not None: from pm.environments import flush_before_selecting flush_before_selecting() facts.record_state("venv", stamp, enabled, **result) if change is not None: finish_publication(paths.repo_root()) receipt.record_venv_rebuild(not current, "already in sync" if current else "") except BaseException: recover_publication(paths.repo_root()) raise def sync_venv(extras: Optional[list[str]] = None, *, explicit: bool = False, plugins: PluginInput | None = None, repair: bool = False, evict_incompatible_plugins: bool = False) -> None: """Make the venv match uv.lock + the enabled extras. Extras union into the installed state (one ledger); no-op when the stamp already matches. ``repair`` restores the recorded dependency graph into a fresh generation, bypassing both that shortcut and config discovery. It cannot add features. ``explicit`` marks a deliberate install command (`hermes pm install`, `hermes update`) — those are the remedy the lazy-install policy points at, so the policy does not apply to them. ``plugins`` names the one source of plugin members (see pm.plugin_inputs); None discovers them from config. ``evict_incompatible_plugins`` is the update's contract: a discovered plugin that keeps the environment from building is disabled instead of failing the sync (see pm.plugin_eviction). Lazy installs OFF = the frozen feature set: when security.allow_lazy_installs is false AND the bundle's enabled-features.json exists, the feature list is FROZEN to that file — requested extras outside it are refused, and plugin members are never installed (the bundle IS the install). Receipt contract: EVERY outcome writes a receipt. ``begin`` fires BEFORE the frozen/lazy refusals (a refusal is a recorded ``failed`` outcome, not a silent raise); finalize runs in FINALLY — no-op syncs ("ok" with ``venv_rebuild`` false) and refusals ("failed") both get a receipt. Plugin selection data is discovered and published by the worker under this lock; no executable transaction phases cross the process boundary.""" from pm import receipt token = receipt.begin("sync") outcome = "failed" try: if repair and (extras is not None or plugins is not None): raise ValueError("repair restores the recorded environment; it cannot change features or plugins") if evict_incompatible_plugins and (repair or plugins is not None or not explicit): raise ValueError("only an explicit sync of the discovered plugin selection may disable plugins") shipped, frozen = _feature_policy(extras, repair=repair) package = get_package("venv") from hermes_cli.runtime_state import recover_publication from pm.publication import StagedPlugin with _venv_install_lock(patient=explicit or repair): recover_publication(paths.repo_root()) change = _publication(plugins) if isinstance(change, StagedPlugin) and not change.active: _publish_inactive(change) elif evict_incompatible_plugins: from pm.plugin_eviction import sync_evicting facts = Facts(paths.runtime_facts_path()) fact = facts.get("venv") or _facts().get("venv") or {} sync_evicting(package, facts, fact, extras=extras, shipped=shipped, frozen=frozen, explicit=explicit) else: inputs = {"plugin_dirs": change.members} if change is not None else _member_inputs(plugins) facts = Facts(paths.runtime_facts_path(), strict=repair) fact = facts.get("venv") or _facts().get("venv") or {} enabled, stamp, inputs = _target_selection(package, fact, extras=extras, inputs=inputs, repair=repair, shipped=shipped, frozen=frozen) current = not repair and _runtime_state_matches(fact, stamp) if not current and not repair and not explicit and not lazy_installs_allowed(): raise _refuse_lazy("venv", str(extras) if extras else "venv out of sync") receipt.record_feature_list(enabled) _commit_selection(package, facts, change, enabled=enabled, stamp=stamp, inputs=inputs, current=current, repair=repair, explicit=explicit) outcome = "ok" except BaseException as exc: receipt.record_step("dependency-sync", False, f"{type(exc).__name__}: {exc}") raise finally: receipt.finalize(outcome, 0 if outcome == "ok" else 1, token=token) def drift(*, include_venv: bool = True) -> dict[str, str]: """Cheap stamp comparisons of the installed state against the lockfile. Maps package names to reasons; empty means healthy. Never installs, never touches the network. An install pm has never touched (no installed-state file) reports nothing — pm only vouches for what it installed. Lockfile packages this build doesn't know (version skew during a partial update) are skipped, not fatal.""" if not paths.facts_path().is_file() and not paths.runtime_facts_path().is_file(): return {} problems: dict[str, str] = {} lockfile = _lockfile() facts = _facts() store = _store() target = current_target() for name in lockfile.names(): try: package = get_package(name) except KeyError: continue if package.optional or package.internal: continue if package.missing_reason(target) is not None: continue if _installed_location(package, lockfile, target) is None: problems[name] = "not installed or outdated" try: venv = get_package("venv") except KeyError: venv = None if include_venv and venv is not None and (paths.runtime_facts_path().is_file() or facts.get("venv") is not None): try: if not venv_is_current(): problems["venv"] = "out of sync with uv.lock" except (OSError, RuntimeError, ValueError) as exc: problems["venv"] = str(exc) return problems def check(*, include_venv: bool = True) -> list[str]: """Human-readable startup diagnostics. Use drift() for package identities.""" return [f"{name}: {reason}" for name, reason in drift(include_venv=include_venv).items()] def _store_path_dirs() -> list[str]: """Composed PATH dirs of all installed (non-internal, on_path) store packages, deps-first, deduped. Includes optional packages that are *installed* (facts say so) — an installed git/gh must be on PATH even though it's not in the root closure. A package whose dependency chain is not fully installed contributes nothing: store npm over a missing store node would run the user's node, a partial toolchain. Never installs.""" lockfile = _lockfile() target = current_target() locations: dict[str, object] = {} def located(package): if package.name not in locations: locations[package.name] = _installed_location(package, lockfile, target) return locations[package.name] dirs: list[str] = [] for name in lockfile.names(): try: package = get_package(name) chain = walk([name]) except KeyError: continue if package.internal: continue if not getattr(package, "on_path", True): continue if package.missing_reason(target) is not None: continue location = located(package) if location is None or any(located(dep) is None for dep in chain if dep.missing_reason(target) is None): continue facts, store = location env = facts.env_for(name, store.root) path_dirs = env.get("PATH") or [] if isinstance(path_dirs, str): path_dirs = [path_dirs] for directory in path_dirs: if directory and directory not in dirs: dirs.append(str(directory)) return dirs def activate(*, allow_incomplete: bool = False) -> list[str]: """Make the installed store usable: put its tool dirs at the FRONT of os.environ['PATH'] so reactive `shutil.which('git'|'node'|'ffmpeg'|...)` resolves the bundled binaries, never a user copy that sorts earlier. Return `check()`'s problems, or an empty list on success, so startup callers can report the verdict without checking the store twice. Drift still activates every package whose whole chain is installed (`_store_path_dirs`): refusing all of them handed every tool to the user's PATH copies. A damaged package, and anything depending on it, stays off PATH so no toolchain is served half from the store. ``allow_incomplete`` is the install-time exception: tools are published before the venv sync, so the venv verdict is not computed at all. This is the ONE sanctioned global PATH write: PATH is the discovery contract every `which` reads, not a tool-specific env leak. Store-first unconditionally — pinned bundled versions win on dev machines too. """ import os # The venv verdict is discarded here, and computing it imports application # config readers (ruamel) that the bare update interpreter does not carry. problems = check(include_venv=not allow_incomplete) path = os.environ.get("PATH", "") first = store_first_path(path) if first != path: os.environ["PATH"] = first return problems def store_first_path(path: str) -> str: """``path`` with the installed store's tool dirs moved to the front, for Hermes's own children whose PATH gets other dirs prepended after activate.""" import os dirs = _store_path_dirs() if not dirs: return path store = {d.lower() for d in dirs} rest = [p for p in path.split(os.pathsep) if p and p.lower() not in store] return os.pathsep.join([*dirs, *rest])