diff --git a/hermes_cli/managed_uv.py b/hermes_cli/managed_uv.py index 7b5ad596c9..f924601140 100644 --- a/hermes_cli/managed_uv.py +++ b/hermes_cli/managed_uv.py @@ -203,6 +203,15 @@ def _run_runtime_repair( """Run the vulnerable-runtime repair hook; never raises (repair is non-fatal).""" try: repair = repair_vulnerable_runtime(uv_bin) + from hermes_cli.update_receipt import record_skip, record_step + + detail = ( + f"{repair.status}: {repair.detail} " + f"(sqlite {repair.sqlite_before or 'unknown'} → {repair.sqlite_after or 'unknown'})" + ) + record_step("sqlite_runtime_repair", repair.status in {"safe", "repaired"}, detail) + if repair.status in {"skipped", "not-applicable"}: + record_skip("sqlite_runtime_repair", detail) if repair_observer is not None: repair_observer(repair) if repair.status == "failed": @@ -612,8 +621,12 @@ def _stream_sync(argv: list[str], *, cwd: Path, env: dict[str, str]) -> tuple[in return status, _sync_reason(tail) +class _CandidateStageError(Exception): + """A rejected candidate, already cleaned up, with its diagnostic reason.""" + + def _stage_candidate_venv( - uv_bin: str, *, project_root: Path, generation: Path, python: Path) -> Path | None: + uv_bin: str, *, project_root: Path, generation: Path, python: Path) -> Path: runtime_root = project_root / _RUNTIME_DIR_NAME candidate = runtime_root / f"venv-candidate-{_token()}" env = managed_python_env(project_root, install_dir=generation) @@ -621,7 +634,9 @@ def _stage_candidate_venv( "UV_PROJECT_ENVIRONMENT": str(candidate), "UV_PYTHON": str(python), "UV_PYTHON_DOWNLOADS": "never", "VIRTUAL_ENV": str(candidate)}) - reject = partial(_reject, candidate, runtime_root) + def reject(message: str, *args) -> None: + _reject(candidate, runtime_root, message, *args) + raise _CandidateStageError(message % args if args else message) print(" → Building a relocatable replacement environment...") created = subprocess.run( [ @@ -964,13 +979,13 @@ def _repair_under_lock( return _result("failed", current, "could not provision a fixed private Python runtime") generation, python, candidate_info = provisioned - candidate = _stage_candidate_venv( - uv_bin, project_root=root, generation=generation, python=python) - if candidate is None: + try: + candidate = _stage_candidate_venv( + uv_bin, project_root=root, generation=generation, python=python) + except _CandidateStageError as exc: _remove_tree(generation, boundary=managed_python_install_dir(root)) return _result( - "failed", current, - "replacement environment did not pass dependency and import smoke tests", + "failed", current, str(exc), sqlite_after=candidate_info.sqlite_version_string) cut_over, backup, final_info, cutover_detail = _cut_over_candidate( diff --git a/tests/hermes_cli/test_managed_uv.py b/tests/hermes_cli/test_managed_uv.py index 6693d32498..fc42fe92f4 100644 --- a/tests/hermes_cli/test_managed_uv.py +++ b/tests/hermes_cli/test_managed_uv.py @@ -11,6 +11,8 @@ from unittest.mock import MagicMock, patch import pytest +from hermes_cli.managed_uv import _CandidateStageError + # --------------------------------------------------------------------------- # Helpers @@ -509,7 +511,7 @@ class TestRuntimeRepair: ), \ patch( "hermes_cli.managed_uv._stage_candidate_venv", - return_value=None, + side_effect=_CandidateStageError("replacement environment rejected"), ): result = repair_vulnerable_runtime("uv", project_root=root) @@ -1289,7 +1291,7 @@ class TestRepairRetriesAfterUvRefresh: ) as mock_refresh, \ patch( "hermes_cli.managed_uv._stage_candidate_venv", - return_value=None, + side_effect=_CandidateStageError("candidate dependency sync failed (rc=1)"), ): result = repair_vulnerable_runtime("uv", project_root=root) return result, attempts, mock_refresh, sentinel @@ -1332,10 +1334,10 @@ class TestRepairRetriesAfterUvRefresh: refresh_result=True, second_attempt=second_attempt, ) - # Provisioning succeeded on retry; staging (mocked to None) is what - # failed — proving the retry result flows into the normal pipeline. + # Provisioning succeeded on retry; staging rejects the candidate, + # proving the retry result flows into the normal pipeline. assert result.status == "failed" - assert "replacement environment" in result.detail + assert result.detail == "candidate dependency sync failed (rc=1)" assert len(attempts) == 2 assert sentinel.read_text(encoding="utf-8") == "live" diff --git a/tests/hermes_cli/test_runtime_repair_receipt.py b/tests/hermes_cli/test_runtime_repair_receipt.py new file mode 100644 index 0000000000..b17423fc55 --- /dev/null +++ b/tests/hermes_cli/test_runtime_repair_receipt.py @@ -0,0 +1,84 @@ +"""Runtime repair diagnostics must reach persisted update receipts (#111497).""" + +import json +from types import SimpleNamespace + +import pytest + +from hermes_cli import managed_uv as uv +from hermes_cli import update_receipt as receipts + + +@pytest.mark.parametrize("status", ["safe", "repaired", "failed", "skipped", "not-applicable"]) +@pytest.mark.parametrize("entry", ["update", "bootstrap"]) +def test_runtime_result_reaches_persisted_receipt(tmp_path, monkeypatch, status, entry): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr(receipts, "_current", None) + result = uv.RuntimeRepairResult(status, "repair diagnostic", "3.50.4", "3.53.1") + monkeypatch.setattr(uv, "repair_vulnerable_runtime", lambda _: result) + if entry == "update": + monkeypatch.setattr(uv, "resolve_uv", lambda: "uv") + monkeypatch.setattr(uv, "_uv_self_update_is_fresh", lambda: True) + invoke = uv.update_managed_uv + else: + paths = iter([None, "uv"]) + monkeypatch.setattr(uv, "resolve_uv", lambda: next(paths)) + monkeypatch.setattr(uv, "_install_uv", lambda _: None) + monkeypatch.setattr(uv, "_uv_version", lambda _: "test") + invoke = uv.ensure_uv + observed = [] + receipts.begin_update_receipt() + invoke(repair_observer=observed.append) + path = receipts.finalize_update_receipt("partial") + data = json.loads(path.read_text()) + step, = data["steps"] + assert step["name"] == "sqlite_runtime_repair" + assert step["ok"] == (status in {"safe", "repaired"}) + assert all(value in step["detail"] for value in ( + result.status, result.detail, result.sqlite_before, result.sqlite_after)) + assert bool(data["skips"]) == (status in {"skipped", "not-applicable"}) + assert observed == [result] + # The same repair hook is also used outside an update, without an active receipt. + uv._run_runtime_repair("uv", observed.append) + assert receipts._current is None + assert observed == [result, result] + + +@pytest.mark.parametrize("stage, reason", [ + ("create", "candidate venv creation failed (rc=1): permission denied"), + ("lock", "candidate dependency sync refused: uv.lock is missing"), + ("sync", "candidate dependency sync failed (rc=1)"), + ("smoke", "candidate venv smoke failed: missing module"), +]) +def test_stage_rejection_preserves_reason_and_live_environment(tmp_path, monkeypatch, stage, reason): + root = tmp_path / "checkout" + live = root / "venv" + live.mkdir(parents=True) + sentinel = live / "sentinel" + sentinel.write_text("unchanged") + generation = root / ".hermes-runtime" / "python" / "generation" + generation.mkdir(parents=True) + if stage != "lock": + (root / "uv.lock").write_text("lock") + current = SimpleNamespace(wal_reset_vulnerable=True, sqlite_version_string="3.50.4") + fixed = SimpleNamespace(sqlite_version_string="3.53.1") + monkeypatch.setattr(uv, "probe_sqlite_runtime", lambda _: current) + monkeypatch.setattr(uv, "_install_safe_python_generation", lambda *a, **kw: ( + generation, generation / "python", fixed)) + + def run(argv, **kwargs): + if argv[1] == "venv": + from pathlib import Path + Path(argv[2]).mkdir(parents=True) + failed = (argv[1] == "venv" and stage == "create") or (argv[1] == "sync" and stage == "sync") + return SimpleNamespace(returncode=int(failed), stderr="permission denied", stdout="") + + monkeypatch.setattr(uv.subprocess, "run", run) + monkeypatch.setattr(uv, "_smoke_candidate_venv", lambda _: (False, "missing module", None)) + result = uv._repair_under_lock("uv", root=root, live=live, live_python=live / "python", + runtime_root=root / ".hermes-runtime") + assert result.status == "failed" + assert result.detail == reason + assert sentinel.read_text() == "unchanged" + assert not generation.exists() + assert not list((root / ".hermes-runtime").glob("venv-candidate-*"))