feat(pm): contain child output interactively, stream it in CI

uv, npm ci, icon generation and the desktop build printed hundreds of
lines on every interactive install, update and `hermes desktop`.

pm/progress.py holds one policy. CI (CI / GITHUB_ACTIONS) or
HERMES_VERBOSE=1 streams child output unchanged. Otherwise each child is
one "→ label…" line, rewritten in place with its latest output on a
terminal, or just the start and finish lines when piped. A failure
always prints the last 80 lines. HERMES_VERBOSE=0 forces containment.

Applied to PM's uv runs (no uv --verbose outside CI; quick steps stay
silent unless they fail), the source build scripts (node-deps, TUI,
icons, web; npm warn lines stay out of the live line), the desktop
build and package steps, and the Windows ARM64 vcpkg/OpenSSL provider.
The policy reads the parent's environment, so the CI=1 the builders
force on their node children does not switch it to verbose.
This commit is contained in:
ethernet
2026-09-24 14:17:03 -04:00
parent df1b647b42
commit a564b9dc17
8 changed files with 261 additions and 16 deletions

View File

@@ -1199,8 +1199,9 @@ def _promote_staged_desktop_app(
def build_prepared_desktop(desktop_dir: Path, *, source_mode: bool, npm: str, env: dict,
icons: Path | None = None) -> Optional[Path]:
"""Build prepared desktop sources, then publish the verified staged app."""
from pm.progress import run_contained
build_label = "source build" if source_mode else "packaged app"
print(f"→ Building desktop {build_label}...")
build_env = dict(env)
if sys.platform == "win32":
# The installer stages pinned Git in its own PowerShell process. Product
@@ -1222,10 +1223,11 @@ def build_prepared_desktop(desktop_dir: Path, *, source_mode: bool, npm: str, en
if stopped:
print(f" ⚠ Stopped running desktop app to free the build output (pid {', '.join(map(str, stopped))})")
try:
subprocess.run(build_cmd, cwd=desktop_dir, env=build_env, check=True)
run_contained(build_cmd, f"Building desktop {build_label}", cwd=desktop_dir, env=build_env)
if staging_dir is not None:
subprocess.run([npm, "run", "builder", "--", "--dir", "--publish", "never",
f"-c.directories.output={staging_dir}"], cwd=desktop_dir, env=build_env, check=True)
run_contained([npm, "run", "builder", "--", "--dir", "--publish", "never",
f"-c.directories.output={staging_dir}"], "Packaging the desktop app",
cwd=desktop_dir, env=build_env)
packaged_executable = (
_promote_staged_desktop_app(desktop_dir, staging_dir) if staging_dir is not None else None
)

View File

@@ -46,10 +46,15 @@ def source_build_env(base_env: dict | None = None, *, explicit: bool = False) ->
return ensure("npm", base_env=env, explicit=explicit).env
def run_source_script(project_root: Path, script: str, *args: str, env: dict) -> None:
subprocess.run(
def run_source_script(project_root: Path, script: str, *args: str, env: dict, label: str) -> None:
from pm.progress import run_contained
# npm's deprecation warnings are the loudest lines and never actionable
# here; they still land in the failure tail.
run_contained(
[shutil.which("node", path=env["PATH"]), str(project_root / script), *args],
cwd=project_root, env=env, check=True,
label, hide=lambda line: line.lower().startswith("npm warn"), indent=" ",
cwd=project_root, env=env,
)
@@ -61,6 +66,7 @@ def prepare_source_dependencies(project_root: Path, workspaces: tuple[str, ...],
project_root, "scripts/build/node-deps.mjs", "--source", str(project_root), "--reuse",
*(() if explicit or lazy_installs_allowed() else ("--no-install",)),
*(arg for workspace in workspaces for arg in ("--workspace", workspace)), env=env,
label="Preparing Node dependencies",
)
@@ -75,15 +81,16 @@ def prepare_launch_dependencies(project_root: Path, *, env: dict) -> None:
def build_source_tui(project_root: Path, *, env: dict) -> None:
run_source_script(project_root, "scripts/build/tui.mjs", env=env)
run_source_script(project_root, "scripts/build/tui.mjs", env=env, label="Building the TUI")
def build_source_web(project_root: Path, *, env: dict, icons: Path | None = None) -> None:
if icons is None:
icons = project_root
run_source_script(project_root, "scripts/generate-icons.mjs", env=env)
run_source_script(project_root, "scripts/generate-icons.mjs", env=env, label="Generating icons")
run_source_script(project_root, "scripts/build/web.mjs", "--source", str(project_root),
"--icons", str(icons), "--out", str(project_root / "hermes_cli/web_dist"), env=env)
"--icons", str(icons), "--out", str(project_root / "hermes_cli/web_dist"), env=env,
label="Building the web UI")
def source_frontends(project_root: Path) -> tuple[str, ...]:

View File

@@ -20,6 +20,16 @@ import time
from typing import TextIO
from pm.package import InstallError
from pm.progress import LiveTail, TextSink, verbose_output
# Slow steps get a status line; unlisted quick ones (venv, export, pip check)
# stay silent unless they fail.
_UV_LABELS: dict[object, str] = {
"sync": "Installing Python dependencies",
"lock": "Resolving Python dependencies",
("pip", "install"): "Installing Python packages",
"cache": "Pruning the uv cache",
}
# Deliberately narrow: a fetch timeout or index outage must not be misread as a
# conflict — and regardless of classification, nothing here ever disables a
@@ -118,7 +128,7 @@ def _read_pipe(fd: int) -> bytes:
def _run_streaming(command: list[str], *, cwd: Path, env: dict[str, str],
timeout: int, output: TextIO) -> subprocess.CompletedProcess:
timeout: int, output: TextSink) -> subprocess.CompletedProcess:
"""Keep CI progress live, a bounded diagnostic tail, and a wall-clock timeout."""
deadline = time.monotonic() + timeout
proc = subprocess.Popen(command, cwd=str(cwd), env=env, stdout=subprocess.PIPE,
@@ -261,7 +271,7 @@ class PythonEnvironment:
if self.offline:
command.append("--offline")
try:
if self.output is not None:
if self.output is not None and verbose_output():
# uv hides build-backend output until failure without verbose mode,
# but --verbose alone is uv's DEBUG level: ~200 lines of interpreter
# and cache internals on every streamed run. RUST_LOG scopes it to the
@@ -269,6 +279,18 @@ class PythonEnvironment:
command.append("--verbose")
env.setdefault("RUST_LOG", "uv_build_frontend=debug")
return _run_streaming(command, cwd=cwd, env=env, timeout=timeout, output=self.output)
if self.output is not None:
# uv prints backend output on failure anyway; interactive users see
# one live status line and the captured tail only if it fails.
tail = LiveTail(_UV_LABELS.get(tuple(args[:2]), _UV_LABELS.get(args[0])),
self.output, indent=" ")
try:
result = _run_streaming(command, cwd=cwd, env=env, timeout=timeout, output=tail)
except BaseException:
tail.close(False)
raise
tail.close(result.returncode == 0)
return result
return subprocess.run(command, cwd=str(cwd), env=env, capture_output=True,
text=True, encoding="utf-8", errors="replace", timeout=timeout)
except subprocess.TimeoutExpired as exc:

View File

@@ -16,6 +16,8 @@ import shutil
import subprocess
import tempfile
from pm.progress import run_contained
_PROVIDER = Path("scripts/build/windows-deps.ps1")
@@ -27,11 +29,14 @@ def prepare_windows_environment(*, source: Path, state: Path, env: Mapping[str,
state.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="environment-", dir=state) as scratch:
output = Path(scratch) / "environment.json"
subprocess.run(
# A cold vcpkg clone and OpenSSL build print thousands of lines; the
# user needs the step and its failure, not the patch log.
run_contained(
[shell, "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File",
str(source / _PROVIDER), "-StateRoot", str(state),
"-EnvironmentFile", str(output)],
cwd=source, env=dict(env), check=True, stdin=subprocess.DEVNULL,
"Preparing Windows ARM64 build tools", indent=" ",
cwd=source, env=dict(env), stdin=subprocess.DEVNULL,
)
prepared = json.loads(output.read_text(encoding="utf-8-sig"))
if not isinstance(prepared, dict) or any(not isinstance(k, str) or not isinstance(v, str) for k, v in prepared.items()):

170
pm/progress.py Normal file
View File

@@ -0,0 +1,170 @@
"""One output policy for long child processes: verbose in CI, contained elsewhere.
CI logs are the only record of a remote failure, so CI (``CI`` /
``GITHUB_ACTIONS``) or ``HERMES_VERBOSE=1`` streams child output unchanged.
Everywhere else a child collapses into one status line: on a terminal it is
rewritten in place with the child's latest line, off a terminal only the
start and finish are printed. A failure always prints the captured tail, so
containment never hides an error. ``HERMES_VERBOSE=0`` forces containment.
Stdlib-only: the bootstrap runner imports this from a pre-3.11 system Python.
"""
from __future__ import annotations
from collections import deque
import os
import re
import shutil
import subprocess
import sys
import time
from typing import IO, Callable, Mapping, Optional, Protocol, Sequence
TAIL_LINES = 80
_TRUE = {"1", "true", "yes", "on"}
_FALSE = {"0", "false", "no", "off"}
_ANSI = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")
# A redraw per child line would dominate a fast install's runtime on slow terminals.
_REDRAW_INTERVAL = 0.05
class TextSink(Protocol):
def write(self, text: str, /) -> int: ...
def flush(self) -> None: ...
def verbose_output(env: Optional[Mapping[str, str]] = None) -> bool:
"""Whether child output streams unchanged instead of being contained."""
source = os.environ if env is None else env
explicit = source.get("HERMES_VERBOSE", "").strip().lower()
if explicit in _TRUE:
return True
if explicit in _FALSE:
return False
return any(source.get(key, "").strip().lower() not in ("", *_FALSE) for key in ("CI", "GITHUB_ACTIONS"))
def _is_terminal(stream: IO[str]) -> bool:
try:
return stream.isatty()
except (AttributeError, ValueError):
return False
class LiveTail:
"""A text sink that contains a child's output behind one status line.
``label=None`` runs silently and speaks only on failure, for steps too
quick to deserve a line of their own.
"""
def __init__(self, label: Optional[str], stream: Optional[IO[str]] = None, *,
hide: Optional[Callable[[str], bool]] = None, indent: str = "") -> None:
self.label = label
self.stream = sys.stdout if stream is None else stream
self.hide = hide
self.indent = indent
self.live = label is not None and _is_terminal(self.stream)
self.tail: deque[str] = deque(maxlen=TAIL_LINES)
self._partial = ""
self._drawn = 0
self._last_draw = 0.0
if label is not None:
if self.live:
self._draw(f"{indent}→ {label}…")
else:
self._emit(f"{indent}→ {label}…\n")
def write(self, text: str) -> int:
# npm and uv redraw with bare CRs; each redraw is a line of progress.
lines = (self._partial + text).replace("\r\n", "\n").replace("\r", "\n").split("\n")
self._partial = lines.pop()
for line in lines:
self._line(line)
return len(text)
def flush(self) -> None:
return None
def close(self, ok: bool) -> None:
if self._partial:
self._line(self._partial)
self._partial = ""
if self.live:
self._draw("")
self._emit("\r")
if ok:
if self.label is not None:
self._emit(f"{self.indent}✓ {self.label}\n")
return
name = self.label or "command"
if not self.tail:
self._emit(f"{self.indent}✗ {name} failed\n")
return
self._emit(f"{self.indent}✗ {name} failed; last {len(self.tail)} lines of output:\n")
self._emit("".join(f"{self.indent} {line}\n" for line in self.tail))
def _line(self, line: str) -> None:
line = _ANSI.sub("", line).rstrip()
if not line.strip():
return
self.tail.append(line)
if not self.live or (self.hide is not None and self.hide(line)):
return
now = time.monotonic()
if now - self._last_draw < _REDRAW_INTERVAL:
return
self._last_draw = now
self._draw(f"{self.indent}→ {self.label}… {line.strip()}")
def _draw(self, text: str) -> None:
# Pad instead of an ANSI erase: a bare CR works on every Windows console.
width = max(20, shutil.get_terminal_size().columns - 1)
text = text[:width]
self._emit("\r" + text + " " * max(0, self._drawn - len(text)))
self._drawn = len(text)
def _emit(self, text: str) -> None:
self.stream.write(text)
self.stream.flush()
def run_contained(command: Sequence[str], label: str, *, stream: Optional[IO[str]] = None,
hide: Optional[Callable[[str], bool]] = None, indent: str = "",
**kwargs) -> subprocess.CompletedProcess:
"""``subprocess.run(check=True)`` under the output policy."""
if verbose_output():
out = sys.stdout if stream is None else stream
out.write(f"{indent}→ {label}…\n")
out.flush()
return subprocess.run(command, check=True, **kwargs)
tail = LiveTail(label, stream, hide=hide, indent=indent)
captured = dict(stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
encoding="utf-8", errors="replace")
if not tail.live:
try:
result = subprocess.run(command, check=True, **captured, **kwargs)
except subprocess.CalledProcessError as exc:
tail.write(exc.output or "")
tail.close(False)
raise
except BaseException:
tail.close(False)
raise
tail.close(True)
return result
try:
with subprocess.Popen(command, **captured, **kwargs) as proc:
assert proc.stdout is not None # stdout=PIPE above.
for line in proc.stdout:
tail.write(line)
code = proc.wait()
except BaseException:
tail.close(False)
raise
tail.close(code == 0)
output = "\n".join(tail.tail)
if code:
raise subprocess.CalledProcessError(code, list(command), output=output)
return subprocess.CompletedProcess(list(command), code, output, None)

View File

@@ -389,10 +389,12 @@ def test_streaming_bounds_memory_without_losing_failure_class(tmp_path, diagnost
assert "final diagnostic" in str(actual)
def test_child_output_is_live_and_keeps_explicit_index_credentials(tmp_path):
def test_child_output_is_live_and_keeps_explicit_index_credentials(tmp_path, monkeypatch):
import io
from pm.environment import PythonEnvironment
monkeypatch.setenv("HERMES_VERBOSE", "1") # CI's streamed log, not the contained view
released = tmp_path / "release-child"
class AcknowledgingLog(io.StringIO):
@@ -433,12 +435,14 @@ def test_child_output_is_live_and_keeps_explicit_index_credentials(tmp_path):
@pytest.fixture(params=["environment", "cli"])
def streaming_runner(request, tmp_path):
def streaming_runner(request, tmp_path, monkeypatch):
import contextlib
import io
from pm.cli import _run_live
from pm.environment import PythonEnvironment
monkeypatch.setenv("HERMES_VERBOSE", "1") # CI's streamed log, not the contained view
output = io.StringIO()
environment = PythonEnvironment(
uv=Path(sys.executable), python=Path(sys.executable), destination=tmp_path / "venv",

View File

@@ -56,6 +56,8 @@ def test_streamed_runs_do_not_request_uv_debug_output(tmp_path, monkeypatch):
import io
from pm import environment
monkeypatch.setenv("HERMES_VERBOSE", "1")
seen: list[list[str]] = []
kwargs_seen: list[dict] = []

33
tests/pm/test_progress.py Normal file
View File

@@ -0,0 +1,33 @@
"""Contained child output: CI keeps every line, interactive runs never hide a failure."""
import io
import subprocess
import sys
import pytest
from pm.progress import run_contained, verbose_output
def test_ci_streams_and_an_explicit_choice_overrides_it():
assert verbose_output({"CI": "true"}) and verbose_output({"GITHUB_ACTIONS": "true"})
assert not verbose_output({}) and not verbose_output({"CI": "0"})
assert not verbose_output({"CI": "1", "HERMES_VERBOSE": "0"})
assert verbose_output({"HERMES_VERBOSE": "1"})
@pytest.mark.parametrize("fails", [False, True])
def test_contained_run_shows_output_only_when_the_child_fails(monkeypatch, fails):
monkeypatch.setenv("HERMES_VERBOSE", "0")
stream = io.StringIO()
script = "import sys\nfor i in range(200): print('noise', i)\nprint('the real error')\nsys.exit(%d)" % fails
command = [sys.executable, "-c", script]
if fails:
with pytest.raises(subprocess.CalledProcessError):
run_contained(command, "Installing things", stream=stream)
else:
run_contained(command, "Installing things", stream=stream)
text = stream.getvalue()
assert "→ Installing things" in text
assert ("the real error" in text) is fails
assert "noise 0" not in text, "the failure tail stays bounded"