fix(update): record SQLite runtime repair outcomes in receipts

This commit is contained in:
wangtao
2026-09-15 11:04:34 +08:00
committed by Teknium
parent 882c59ee93
commit 2f3d0b5cc2
3 changed files with 113 additions and 12 deletions
+22 -7
View File
@@ -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(
+7 -5
View File
@@ -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"
@@ -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-*"))