hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity

refactor(pm): split cmd_install and cmd_update into their phases

Browse Source 
cmd_install: flag validation and the venv/test-environment phase move
to helpers (CC 40 -> 18). cmd_update: resolution, per-row report, pin
application, uv and npm refreshes move to helpers (CC 51 -> 22). The
messages, order and exit codes are unchanged; a dead `target` local in
cmd_update is dropped.
This commit is contained in:
ethernet
2026-09-24 13:59:57 -04:00
parent 581bb440fd
commit 3e256363a4
1 changed files with 154 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

278
pm/cli.py
View File

@@ -175,58 +175,29 @@ def _install_names(names: list[str], target: str | None = None, *, verify: bool
def cmd_install(args) -> int:
cross_target = getattr(args, "target", None)
def _install_flag_error(args, *, extras: list[str], cross_target, tools_only: bool,
trust_recorded: bool, test_environment) -> str | None:
"""The first incompatible flag combination, as its user-facing message."""
if cross_target:
if cross_target not in ALL_TARGETS:
print(f"✗ unknown target {cross_target!r}; known: {', '.join(ALL_TARGETS)}")
return 1
return f"unknown target {cross_target!r}; known: {', '.join(ALL_TARGETS)}"
if not args.names:
print("✗ --target requires explicit package names")
return 1
# Source-install launchers require the store interpreter, even though
# Python remains optional when provisioning individual tools.
extras = list(dict.fromkeys(getattr(args, "extra", None) or ()))
tools_only = bool(getattr(args, "tools_only", False))
trust_recorded = bool(getattr(args, "trust_recorded", False))
test_environment = getattr(args, "test_environment", None)
return "--target requires explicit package names"
if test_environment is not None and (extras or cross_target or args.names or tools_only):
print("✗ --test-environment builds beside the default closure; it does not take names, --extra, --target, or --tools-only")
return 1
return "--test-environment builds beside the default closure; it does not take names, --extra, --target, or --tools-only"
if tools_only and (extras or cross_target or args.names):
print("✗ --tools-only installs the tool closure and then stops; it does not take names, --extra, or --target")
return 1
return "--tools-only installs the tool closure and then stops; it does not take names, --extra, or --target"
if trust_recorded and (extras or cross_target or args.names):
print("✗ --trust-recorded installs the default closure and then stops; it does not take names, --extra, or --target")
return 1
return "--trust-recorded installs the default closure and then stops; it does not take names, --extra, or --target"
if extras and cross_target:
print("✗ --extra syncs this install's venv and cannot combine with --target")
return 1
names = args.names if args.names or extras else source_install_packages(_lockfile().names())
# Only the whole default closure verifies everything an activated shell
# composes from, so only it advances the prologue's input stamps.
full_closure = not (args.names or tools_only or cross_target)
from pm.environments import activation_input_mtimes
return "--extra syncs this install's venv and cannot combine with --target"
return None
input_mtimes = activation_input_mtimes(repo_root()) if full_closure else {}
# Tools before the venv. A bare `pm install` used to install tools and
# sync the venv in one breath, so a native build (Windows ARM64 source
# wheels) resolved compilers and git from the host PATH. Publish every
# tool first and put it on PATH; sync only after that.
tool_names = names if args.names else tool_roots(names)
failed = _install_names(tool_names, target=cross_target, verify=not trust_recorded)
if failed:
return 1
if not cross_target and (not args.names or tools_only):
from pm.install import activate
problems = activate(allow_incomplete=True)
if problems:
print(f"✗ tools not on PATH before venv sync: {'; '.join(problems)}", flush=True)
return 1
if tools_only:
return 0
if extras or not args.names:
def _install_python_environments(extras: list[str], *, sync: bool, test_environment) -> int:
"""Sync the venv and build the side test environment; return the failure count."""
failed = 0
if sync:
from pm.install import sync_venv
try:
@@ -253,6 +224,47 @@ def cmd_install(args) -> int:
except InstallError as e:
print(f"✗ {e}")
failed += 1
return failed
def cmd_install(args) -> int:
cross_target = getattr(args, "target", None)
# Source-install launchers require the store interpreter, even though
# Python remains optional when provisioning individual tools.
extras = list(dict.fromkeys(getattr(args, "extra", None) or ()))
tools_only = bool(getattr(args, "tools_only", False))
trust_recorded = bool(getattr(args, "trust_recorded", False))
test_environment = getattr(args, "test_environment", None)
error = _install_flag_error(args, extras=extras, cross_target=cross_target, tools_only=tools_only,
trust_recorded=trust_recorded, test_environment=test_environment)
if error:
print(f"✗ {error}")
return 1
names = args.names if args.names or extras else source_install_packages(_lockfile().names())
# Only the whole default closure verifies everything an activated shell
# composes from, so only it advances the prologue's input stamps.
full_closure = not (args.names or tools_only or cross_target)
from pm.environments import activation_input_mtimes
input_mtimes = activation_input_mtimes(repo_root()) if full_closure else {}
# Tools before the venv. A bare `pm install` used to install tools and
# sync the venv in one breath, so a native build (Windows ARM64 source
# wheels) resolved compilers and git from the host PATH. Publish every
# tool first and put it on PATH; sync only after that.
tool_names = names if args.names else tool_roots(names)
if _install_names(tool_names, target=cross_target, verify=not trust_recorded):
return 1
if not cross_target and (not args.names or tools_only):
from pm.install import activate
problems = activate(allow_incomplete=True)
if problems:
print(f"✗ tools not on PATH before venv sync: {'; '.join(problems)}", flush=True)
return 1
if tools_only:
return 0
failed = _install_python_environments(extras, sync=bool(extras or not args.names),
test_environment=test_environment)
if full_closure and not failed:
from pm.environments import activation_inputs_dir, record_activation_inputs
@@ -389,6 +401,98 @@ def _run_live(cmd: list[str], *, cwd, env, timeout: int = 3600) -> tuple[int, st
return result.returncode, result.stderr
def _resolve_updates(names: list[str], lockfile, only_target: str | None) -> tuple[list, list[str]]:
"""Resolve each package's latest; an upstream outage records a failure instead of aborting."""
resolved = []
failures = []
for name in names:
package = get_package(name)
targets = [t for t in ALL_TARGETS if package.missing_reason(t) is None]
if only_target: # cross-target check: only the requested target matters
targets = [t for t in targets if t == only_target]
if not targets:
continue
try:
decision = resolve_package(package, targets, lockfile.version(name),
artifacts=lockfile.pinned_artifacts(name))
except Exception as e: # an upstream index outage must not kill the whole check
decision = Resolved(name, lockfile.version(name), package.version_style, reason=f"resolve failed: {e}")
failures.append(name)
resolved.append(decision)
return resolved, failures
def _report_update(decision, width: int) -> None:
d = decision
if d.version is None:
print(f"{d.name:<{width}} {d.reason or 'up to date'}")
elif not d.changed:
print(f"{d.name:<{width}} {d.locked} up to date")
elif d.version == d.locked and d.artifact_updates:
print(f"{d.name:<{width}} {d.version}: newer artifacts for {', '.join(sorted(d.artifact_updates))}")
else:
per = ""
if d.per_target and len(set(d.per_target.values())) > 1:
per = " (" + ", ".join(f"{t}={v}" for t, v in sorted(d.per_target.items())) + ")"
print(f"{d.name:<{width}} {d.locked or '—'} → {d.version}{per}")
def _sync_venv_step() -> bool:
from pm.install import sync_venv
try:
sync_venv(explicit=True)
except InstallError as e:
print(f"✗ {e}")
return False
print("✓ venv")
return True
def _apply_pins(changed: list, lockfile) -> int:
if not changed:
print("pm update: nothing to update")
return 0
for d in changed:
package = get_package(d.name)
artifacts = _pin_artifacts(package, d, lockfile.pinned_artifacts(d.name))
lockfile.set_pin(d.name, d.version, artifacts)
print(f"✓ {d.name} pinned {d.locked or '—'} → {d.version}")
lockfile.save()
if _install_names([d.name for d in changed]):
return 1
return 0 if _sync_venv_step() else 1
def _refresh_uv_lock() -> int:
try:
lock_project(repo_root(), upgrade=True, explicit=True)
except InstallError as exc:
print(f"✗ Python lock refresh failed: {exc}")
return 1
print("✓ uv.lock refreshed")
return 0 if _sync_venv_step() else 1
def _refresh_npm_lock() -> int:
from pm.install import env_for, installed_package
from pm.packages import npm_env
from pm.paths import writable_store_root
npm = installed_package("npm")
node = installed_package("node")
if npm is None or npm.binary is None or node is None or node.binary is None:
print("✗ npm or Node: not installed; run `hermes pm install`")
return 1
env = npm_env(writable_store_root() / ".npm-cache", env_for("npm"))
code, tail = _run_live([str(npm.binary), "update"], cwd=str(repo_root()), env=env)
if code != 0:
print(f"✗ npm update failed:\n{tail}")
return 1
print("✓ package-lock.json refreshed")
return 0
def cmd_update(args) -> int:
"""`hermes pm update [names...] [--check] [--target T] [--uv] [--npm] [--termux]`.
@@ -410,44 +514,15 @@ def cmd_update(args) -> int:
if args.target and not args.check:
print("::warning::--target is a CHECK-only cross-resolution flag; ignoring it for apply (the lockfile pins every target)")
args.target = None
target = args.target or current_target()
resolved = []
failures = []
for name in names:
package = get_package(name)
targets = [t for t in ALL_TARGETS if package.missing_reason(t) is None]
if args.target: # cross-target check: only the requested target matters
targets = [t for t in targets if t == args.target]
if not targets:
continue
try:
decision = resolve_package(package, targets, lockfile.version(name),
artifacts=lockfile.pinned_artifacts(name))
except Exception as e: # an upstream index outage must not kill the whole check
decision = Resolved(name, lockfile.version(name), package.version_style, reason=f"resolve failed: {e}")
failures.append(name)
resolved.append(decision)
# ── report ────────────────────────────────────────────────────────────
resolved, failures = _resolve_updates(names, lockfile, args.target)
changed = [d for d in resolved if d.changed]
if not resolved:
print("pm update: nothing to check (no resolvable packages)")
return 0
width = max(len(d.name) for d in resolved)
for d in resolved:
if d.version is None:
print(f"{d.name:<{width}} {d.reason or 'up to date'}")
elif d.changed:
per = ""
if d.per_target and len(set(d.per_target.values())) > 1:
per = " (" + ", ".join(f"{t}={v}" for t, v in sorted(d.per_target.items())) + ")"
if d.version == d.locked and d.artifact_updates:
print(f"{d.name:<{width}} {d.version}: newer artifacts for {', '.join(sorted(d.artifact_updates))}")
else:
print(f"{d.name:<{width}} {d.locked or '—'} → {d.version}{per}")
else:
print(f"{d.name:<{width}} {d.locked} up to date")
_report_update(d, width)
if failures:
print(f"pm update: resolution failed for {', '.join(failures)}; no changes applied")
return 1
@@ -458,57 +533,12 @@ def cmd_update(args) -> int:
print("npm deps: would run `npm update`")
return 1 if changed else 0
# ── apply ─────────────────────────────────────────────────────────────
if changed:
for d in changed:
package = get_package(d.name)
artifacts = _pin_artifacts(package, d, lockfile.pinned_artifacts(d.name))
lockfile.set_pin(d.name, d.version, artifacts)
print(f"✓ {d.name} pinned {d.locked or '—'} → {d.version}")
lockfile.save()
failed = _install_names([d.name for d in changed])
if failed:
return 1
try:
from pm.install import sync_venv
sync_venv(explicit=True)
print("✓ venv")
except InstallError as e:
print(f"✗ {e}")
return 1
else:
print("pm update: nothing to update")
if args.uv:
try:
lock_project(repo_root(), upgrade=True, explicit=True)
except InstallError as exc:
print(f"✗ Python lock refresh failed: {exc}")
return 1
print("✓ uv.lock refreshed")
try:
from pm.install import sync_venv
sync_venv(explicit=True)
print("✓ venv")
except InstallError as e:
print(f"✗ {e}")
return 1
if _apply_pins(changed, lockfile):
return 1
if args.uv and _refresh_uv_lock():
return 1
if args.npm:
from pm.install import env_for, installed_package
from pm.packages import npm_env
from pm.paths import writable_store_root
npm = installed_package("npm")
node = installed_package("node")
if npm is None or npm.binary is None or node is None or node.binary is None:
print("✗ npm or Node: not installed; run `hermes pm install`")
return 1
env = npm_env(writable_store_root() / ".npm-cache", env_for("npm"))
code, tail = _run_live([str(npm.binary), "update"], cwd=str(repo_root()), env=env)
if code != 0:
print(f"✗ npm update failed:\n{tail}")
return 1
print("✓ package-lock.json refreshed")
return _refresh_npm_lock()
return 0