fix(update): test runs never mutate the live checkout — pytest-guard the marker and repair paths

Guard the .lazy-refresh-incomplete marker writer (update_cmd), launch-time
recovery (main.py), and _early_recovery repair paths behind a two-condition
check: running under pytest AND the target is this live checkout. Sandboxed
tmp_path tests still exercise the real code paths.

Salvaged from PR #72002 by @fcavalcantirj. Fixes #72000.

Co-authored-by: fcavalcantirj <felipe.cavalcanti.rj@gmail.com>
This commit is contained in:
fcavalcantirj 2026-07-29 20:12:33 -07:00 committed by Teknium
parent 3233de9a21
commit f04fd1e7ad
15 changed files with 398 additions and 63 deletions

View file

@ -1,2 +0,0 @@
started=1785343166.9711895
pid=2093896

View file

@ -0,0 +1,2 @@
fcavalcantirj
# PR #72002 salvage

View file

@ -171,6 +171,22 @@ def _run_repair_install(specs: list[str], project_root: Path) -> bool:
return True
def _pytest_owns_live_checkout(root: Path) -> bool:
"""True when running under pytest AND ``root`` is this module's own
checkout the one whose venv is executing the suite right now.
Lifecycle tests spawn real subprocesses that import ``hermes_cli.main``
with recovery armed; ``PYTEST_CURRENT_TEST`` rides the inherited env into
those children. Without this guard, a genuinely-broken dev venv gets a
REAL ``ensurepip`` + ``pip install --force-reinstall`` from inside a
running test suite. Tests that sandbox ``project_root`` to a tmp_path are
unaffected (same posture as ``managed_scope._under_pytest``)."""
return (
"PYTEST_CURRENT_TEST" in os.environ
and root == Path(__file__).resolve().parent.parent
)
def recover_if_needed(
project_root: Path | None = None,
argv: list[str] | None = None,
@ -193,6 +209,8 @@ def recover_if_needed(
if "update" in args:
return
root = _project_root() if project_root is None else project_root
if _pytest_owns_live_checkout(root):
return
core_marker = root / ".update-incomplete"
lazy_marker = root / ".lazy-refresh-incomplete"
if not core_marker.exists() and not lazy_marker.exists():

View file

@ -7540,6 +7540,22 @@ def _lazy_refresh_marker_path() -> Path:
return PROJECT_ROOT / ".lazy-refresh-incomplete"
def _pytest_owns_live_checkout(root: Path) -> bool:
"""True when running under pytest AND ``root`` is this checkout itself.
Tests that drive update/recovery without sandboxing ``PROJECT_ROOT``
must neither litter the live repo root with recovery breadcrumbs
(a leftover ``.lazy-refresh-incomplete`` / ``.update-incomplete``
false-arms recovery on the developer's next real launch) nor run a real
reinstall against the executing venv. Sandboxed tests point at a
tmp_path and are unaffected (same posture as
``managed_scope._under_pytest``)."""
return (
"PYTEST_CURRENT_TEST" in os.environ
and root == Path(__file__).resolve().parent.parent
)
def _clear_marker_file(path: Path, *, label: str) -> None:
"""Remove an update-recovery breadcrumb. Never raises."""
try:
@ -7586,6 +7602,8 @@ def _recover_from_interrupted_install() -> None:
protocol stream (``hermes acp`` speaks JSON-RPC on stdout) must never get
install noise on stdout.
"""
if _pytest_owns_live_checkout(PROJECT_ROOT):
return
core_marker = _update_marker_path().exists()
lazy_marker = _lazy_refresh_marker_path().exists()
if not core_marker and not lazy_marker:

View file

@ -1392,6 +1392,9 @@ def _invalidate_update_cache():
def _write_marker_file(path: Path, *, label: str) -> None:
"""Drop an update-recovery breadcrumb. Never raises."""
if _m()._pytest_owns_live_checkout(path.parent):
logger.debug("Skipping %s marker under pytest (live checkout)", label)
return
try:
path.write_text(
f"started={_time.time()}\npid={os.getpid()}\n", encoding="utf-8"

View file

@ -440,6 +440,20 @@ def _hermetic_environment(tmp_path, monkeypatch):
(fake_hermes_home / "skills").mkdir()
monkeypatch.setenv("HERMES_HOME", str(fake_hermes_home))
# 3b. hermes_state computes ``DEFAULT_DB_PATH = get_hermes_home() / "state.db"``
# at import time. When the module is first imported at collection (any
# test file with a top-level ``from hermes_state import ...``) that
# happens BEFORE this fixture ever runs, so every argless
# ``SessionDB()`` in every test opens the developer's REAL state.db —
# reading real sessions into assertions and writing test rows into the
# real profile. Re-pin the constant to this test's home. (Several test
# files already do this locally; this makes it an invariant.)
hermes_state_mod = sys.modules.get("hermes_state")
if hermes_state_mod is not None and hasattr(hermes_state_mod, "DEFAULT_DB_PATH"):
monkeypatch.setattr(
hermes_state_mod, "DEFAULT_DB_PATH", fake_hermes_home / "state.db"
)
# 4. Deterministic locale / timezone / hashseed. CI runs in UTC with
# C.UTF-8 locale; local dev often doesn't. Pin everything.
monkeypatch.setenv("TZ", "UTC")
@ -648,6 +662,105 @@ def _kanban_write_guard(_hermetic_environment, monkeypatch):
# approvals from one test's session into another's.
# ── tui_gateway.server shared-module state isolation ───────────────────────
#
# ``tui_gateway.server`` registers its RPC handlers in a module-level
# ``_methods`` dict at import time and keeps per-session state in module
# globals (sessions, child-run registry, config cache, DB handle). The
# canonical per-file process isolation above hides any leakage, but a direct
# multi-file invocation (``pytest tests/tui_gateway/ tests/test_tui_gateway_server.py``,
# or plain ``pytest tests/``) shares one interpreter: a test that stubs
# ``_methods["slash.exec"]`` or leaves an active-session lease behind breaks
# unrelated tests in later files. This fixture snapshots the cheap-to-copy
# globals before each test and restores them after, so any file combination
# is order-independent. It is a near no-op (one sys.modules lookup) while
# the module has not been imported.
#
# The case this cannot cover — the module is first imported *during* a test
# that also mutates ``_methods`` — is handled by the importing files' own
# ``server`` fixtures (tests/tui_gateway/test_protocol.py and friends), which
# snapshot immediately after the import.
_TUI_SERVER_MODULE = "tui_gateway.server"
def _teardown_tui_server_sessions(mod) -> None:
"""Close leftover sessions through the production teardown boundary.
Besides returning active-session leases, this finalizes the session,
unregisters notification state, and closes its agent and slash worker.
"""
sessions = getattr(mod, "_sessions", None)
if not isinstance(sessions, dict):
return
for sid in list(sessions):
mod._close_session_by_id(sid, end_reason="test_cleanup")
@pytest.fixture(autouse=True)
def _reset_tui_gateway_server_state():
mod = sys.modules.get(_TUI_SERVER_MODULE)
snapshot = None
if mod is not None:
snapshot = {
"methods": dict(mod._methods),
"cfg": (mod._cfg_cache, mod._cfg_mtime, mod._cfg_path),
"db": (mod._db, mod._db_error),
"real_stdout": mod._real_stdout,
}
yield
mod = sys.modules.get(_TUI_SERVER_MODULE)
if mod is None:
return
# This finalizer can run before the test's own monkeypatch undo, so a
# global may still be replaced with a non-dict test double — skip those
# (monkeypatch restores the real, pre-test object afterwards anyway).
sessions = mod._sessions
if isinstance(sessions, dict):
_teardown_tui_server_sessions(mod)
for name in (
"_pending",
"_pending_prompt_payloads",
"_answers",
"_child_mirrors",
"_active_child_runs",
):
obj = getattr(mod, name, None)
if isinstance(obj, dict):
obj.clear()
if snapshot is not None:
mod._methods.clear()
mod._methods.update(snapshot["methods"])
mod._cfg_cache, mod._cfg_mtime, mod._cfg_path = snapshot["cfg"]
mod._db, mod._db_error = snapshot["db"]
mod._real_stdout = snapshot["real_stdout"]
else:
# First imported during this test — reset to import-time defaults
# for the globals we could not snapshot (``_methods`` is left to
# the importing file's fixture, see block comment above).
mod._cfg_cache = None
mod._cfg_mtime = None
mod._cfg_path = None
mod._db = None
mod._db_error = None
# A leaked context-local Hermes home override redirects every later
# ``get_hermes_home()`` call (active-session registry, config paths)
# to a stale per-test tmpdir. Force the main-thread ContextVar back
# to its default.
try:
from hermes_constants import get_hermes_home_override, set_hermes_home_override
if get_hermes_home_override() is not None:
set_hermes_home_override(None)
except Exception:
pass
@pytest.fixture()
def tmp_dir(tmp_path):
"""Provide a temporary directory that is cleaned up automatically."""

View file

@ -0,0 +1,108 @@
"""The test suite must never mutate the LIVE checkout or its venv.
Regression tests for the pytest-guards on the checkout-root mutation paths.
Before the guards, tests that drove ``cmd_update``/recovery without
sandboxing ``PROJECT_ROOT`` left ``.lazy-refresh-incomplete`` at the real
repo root (observed after full-suite runs), and on a venv with genuinely
broken packages test-spawned subprocesses importing ``hermes_cli.main``
ran a REAL ``ensurepip`` + ``pip install --force-reinstall`` against the
developer's executing environment mid-suite.
The guard predicate requires BOTH conditions (under pytest AND the target is
this checkout itself), so every tmp_path-sandboxed test keeps exercising the
real code paths unchanged.
"""
from __future__ import annotations
from pathlib import Path
import hermes_cli.main as main_mod
from hermes_cli import _early_recovery as er
CHECKOUT_ROOT = Path(er.__file__).resolve().parent.parent
class TestPredicate:
def test_true_for_live_checkout_under_pytest(self):
# PYTEST_CURRENT_TEST is set by pytest itself right now.
assert er._pytest_owns_live_checkout(CHECKOUT_ROOT) is True
assert main_mod._pytest_owns_live_checkout(CHECKOUT_ROOT) is True
def test_false_for_sandboxed_root(self, tmp_path):
assert er._pytest_owns_live_checkout(tmp_path) is False
assert main_mod._pytest_owns_live_checkout(tmp_path) is False
def test_false_outside_pytest(self, monkeypatch):
monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False)
assert er._pytest_owns_live_checkout(CHECKOUT_ROOT) is False
assert main_mod._pytest_owns_live_checkout(CHECKOUT_ROOT) is False
class TestMarkerWrites:
def test_refuses_breadcrumb_at_live_repo_root(self):
target = CHECKOUT_ROOT / ".lazy-refresh-incomplete"
# The marker may legitimately pre-exist: upstream currently TRACKS a
# littered copy in git (the exact pollution this guard prevents), so
# the contract is content-unchanged, not never-exists.
before = target.read_text(encoding="utf-8") if target.exists() else None
try:
main_mod._write_marker_file(target, label="lazy-refresh-incomplete")
after = (
target.read_text(encoding="utf-8") if target.exists() else None
)
assert after == before, (
"marker breadcrumb written into the LIVE checkout from a test"
)
finally:
# If the guard is broken (RED state), restore the pre-test state —
# leaving pollution behind is exactly the bug being pinned.
if before is None:
target.unlink(missing_ok=True)
else:
target.write_text(before, encoding="utf-8")
def test_still_writes_sandboxed(self, tmp_path):
target = tmp_path / ".lazy-refresh-incomplete"
main_mod._write_marker_file(target, label="lazy-refresh-incomplete")
assert target.exists()
assert "pid=" in target.read_text(encoding="utf-8")
class TestEarlyRecovery:
def test_skips_live_checkout_before_any_probe_or_lock(self, monkeypatch):
# A probe call would mean recovery is proceeding against the live
# checkout; the guard must return before ANY side-effectful step.
def _boom():
raise AssertionError("probe ran against the live checkout")
monkeypatch.setattr(er, "_probe_broken_packages", _boom)
monkeypatch.setattr(er, "_run_repair_install", lambda *a, **k: _boom())
er.recover_if_needed(project_root=CHECKOUT_ROOT, argv=[])
def test_sandboxed_root_still_recovers(self, tmp_path, monkeypatch):
# The guard must not disable recovery for sandboxed roots: with a
# marker present and a broken probe, the repair path still runs.
(tmp_path / ".lazy-refresh-incomplete").write_text("started=1\npid=1\n")
(tmp_path / "pyproject.toml").write_text("[project]\nname='x'\n")
monkeypatch.setattr(er, "_probe_broken_packages", lambda: ["PyYAML"])
monkeypatch.setattr(er, "_pinned_specs", lambda broken, root: broken)
installs = []
monkeypatch.setattr(
er, "_run_repair_install", lambda specs, root: installs.append(specs) or True
)
er.recover_if_needed(project_root=tmp_path, argv=[])
assert installs, "sandboxed recovery was wrongly disabled by the guard"
class TestLaunchRecovery:
def test_recover_from_interrupted_install_noops_on_live_checkout(
self, monkeypatch
):
# PROJECT_ROOT is the live checkout in-suite; the launch-time
# recovery must return before touching markers or spawning installs.
def _boom(*a, **k):
raise AssertionError("launch recovery ran against the live checkout")
monkeypatch.setattr(main_mod, "_update_marker_path", _boom)
main_mod._recover_from_interrupted_install()

View file

@ -17,6 +17,8 @@ import pytest
@pytest.fixture()
def server():
# Mocks are scoped to the initial import only (see
# tests/tui_gateway/test_protocol.py for the rationale).
with patch.dict(
"sys.modules",
{
@ -28,7 +30,8 @@ def server():
"hermes_state": MagicMock(),
},
):
yield importlib.import_module("tui_gateway.server")
mod = importlib.import_module("tui_gateway.server")
yield mod
def _capture(server, monkeypatch):

View file

@ -35,6 +35,8 @@ def hermes_home(tmp_path, monkeypatch):
@pytest.fixture()
def server(hermes_home):
# Mocks are scoped to the initial import only (see
# tests/tui_gateway/test_protocol.py for the rationale).
with patch.dict(
"sys.modules",
{
@ -43,18 +45,17 @@ def server(hermes_home):
},
):
mod = importlib.import_module("tui_gateway.server")
yield mod
# Reset module-level session state without re-importing. importlib.reload
# would re-register the module's atexit hooks (ThreadPoolExecutor
# shutdown, _shutdown_sessions); the duplicates race the stderr
# buffer at interpreter shutdown and surface as Fatal Python error:
# _enter_buffered_busy. Clearing the per-session dicts gives the
# next test a clean slate; _methods is NOT cleared because it's
# populated at module import time and re-registration only happens
# via reload (which we don't do).
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
yield mod
# Reset module-level session state without re-importing. importlib.reload
# would re-register the module's atexit hooks (ThreadPoolExecutor
# shutdown, _shutdown_sessions); the duplicates race the stderr
# buffer at interpreter shutdown and surface as Fatal Python error:
# _enter_buffered_busy. Clearing the per-session dicts gives the
# next test a clean slate.
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
@pytest.fixture()

View file

@ -34,6 +34,9 @@ def _restore_stdout():
@pytest.fixture()
def server():
# Mocks are scoped to the initial import only — keeping them active for
# the whole test would poison modules first imported inside test bodies
# (see tests/tui_gateway/test_protocol.py for the full rationale).
with patch.dict("sys.modules", {
"hermes_constants": MagicMock(get_hermes_home=MagicMock(return_value="/tmp/hermes_test")),
"hermes_cli.env_loader": MagicMock(),
@ -42,10 +45,19 @@ def server():
}):
import importlib
mod = importlib.import_module("tui_gateway.server")
yield mod
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
# Tests below stub handlers ("session.list", "prompt.submit", ...) in
# the module-level _methods dict shared with every other test file in
# the process — snapshot and restore it around each test.
methods = dict(mod._methods)
real_stdout = mod._real_stdout
yield mod
mod._methods.clear()
mod._methods.update(methods)
mod._real_stdout = real_stdout
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
@pytest.fixture()

View file

@ -16,6 +16,8 @@ import pytest
@pytest.fixture()
def server():
# Mocks are scoped to the initial import only (see
# tests/tui_gateway/test_protocol.py for the rationale).
with patch.dict(
"sys.modules",
{
@ -30,8 +32,9 @@ def server():
import importlib
mod = importlib.import_module("tui_gateway.server")
yield mod
mod._sessions.clear()
yield mod
mod._sessions.clear()
@pytest.fixture()

View file

@ -21,6 +21,12 @@ def _restore_stdout():
@pytest.fixture()
def server():
# The sys.modules mocks only need to cover the *initial* import — once
# tui_gateway.server is cached, they are inert. Keeping them active for
# the whole test poisons any module first imported inside a test body:
# e.g. hermes_cli.active_sessions would bind the mocked get_hermes_home
# (a fixed shared path) forever, leaking active-session registry entries
# across every later test in the process. Scope the patch to the import.
with patch.dict("sys.modules", {
"hermes_constants": MagicMock(get_hermes_home=MagicMock(return_value="/tmp/hermes_test")),
"hermes_cli.env_loader": MagicMock(),
@ -29,19 +35,60 @@ def server():
}):
import importlib
mod = importlib.import_module("tui_gateway.server")
yield mod
# Reset module-level session state without re-importing. importlib.reload
# would re-register the module's atexit hooks (ThreadPoolExecutor
# shutdown, _shutdown_sessions); the duplicates race the stderr
# buffer at interpreter shutdown and surface as Fatal Python error:
# _enter_buffered_busy. Clearing the per-session dicts gives the
# next test a clean slate; _methods is NOT cleared because it's
# populated at module import time and re-registration only happens
# via reload (which we don't do).
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
mod._live_transports.clear()
# Snapshot the RPC registry: several tests below stub handlers
# ("slash.exec", "fast.ping", ...) directly in the module-level dict,
# which is shared with every other test file in the process.
methods = dict(mod._methods)
real_stdout = mod._real_stdout
yield mod
# Reset module-level state without re-importing. importlib.reload
# would re-register the module's atexit hooks (ThreadPoolExecutor
# shutdown, _shutdown_sessions); the duplicates race the stderr
# buffer at interpreter shutdown and surface as Fatal Python error:
# _enter_buffered_busy. Restoring the dicts in place gives the next
# test a clean slate.
mod._methods.clear()
mod._methods.update(methods)
mod._real_stdout = real_stdout
for sid in list(mod._sessions):
mod._close_session_by_id(sid, end_reason="test_cleanup")
mod._pending.clear()
mod._answers.clear()
mod._live_transports.clear()
def test_shared_fixture_cleanup_uses_full_session_teardown(server, monkeypatch):
"""The cross-file autouse cleanup must close every retained resource."""
from tests import conftest
closed = {"worker": 0, "agent": 0, "lease": 0}
class _Closable:
def __init__(self, key):
self.key = key
def close(self):
closed[self.key] += 1
class _Lease:
def release(self):
closed["lease"] += 1
monkeypatch.setattr(server, "_get_db", lambda: None)
server._sessions["leaked"] = {
"session_key": "leaked",
"agent": _Closable("agent"),
"slash_worker": _Closable("worker"),
"active_session_lease": _Lease(),
"history": [],
}
conftest._teardown_tui_server_sessions(server)
assert server._sessions == {}
assert closed == {"worker": 1, "agent": 1, "lease": 1}
>>>>>>> theirs
@pytest.fixture()

View file

@ -18,6 +18,8 @@ import pytest
@pytest.fixture()
def server():
# Mocks are scoped to the initial import only (see
# tests/tui_gateway/test_protocol.py for the rationale).
with patch.dict(
"sys.modules",
{
@ -32,18 +34,17 @@ def server():
import importlib
mod = importlib.import_module("tui_gateway.server")
yield mod
# Reset module-level session state without re-importing. importlib.reload
# would re-register the module's atexit hooks (ThreadPoolExecutor
# shutdown, _shutdown_sessions); the duplicates race the stderr
# buffer at interpreter shutdown and surface as Fatal Python error:
# _enter_buffered_busy. Clearing the per-session dicts gives the
# next test a clean slate; _methods is NOT cleared because it's
# populated at module import time and re-registration only happens
# via reload (which we don't do).
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
yield mod
# Reset module-level session state without re-importing. importlib.reload
# would re-register the module's atexit hooks (ThreadPoolExecutor
# shutdown, _shutdown_sessions); the duplicates race the stderr
# buffer at interpreter shutdown and surface as Fatal Python error:
# _enter_buffered_busy. Clearing the per-session dicts gives the
# next test a clean slate.
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
def test_init_session_attaches_background_review_callback(server, monkeypatch):

View file

@ -17,6 +17,8 @@ import pytest
@pytest.fixture()
def server():
# Mocks are scoped to the initial import only (see
# tests/tui_gateway/test_protocol.py for the rationale).
with patch.dict(
"sys.modules",
{
@ -31,12 +33,13 @@ def server():
import importlib
mod = importlib.import_module("tui_gateway.server")
yield mod
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
mod._child_mirrors.clear()
mod._active_child_runs.clear()
yield mod
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
mod._child_mirrors.clear()
mod._active_child_runs.clear()
@pytest.fixture()

View file

@ -34,6 +34,8 @@ def hermes_home(tmp_path, monkeypatch):
@pytest.fixture()
def server(hermes_home):
# Mocks are scoped to the initial import only (see
# tests/tui_gateway/test_protocol.py for the rationale).
with patch.dict(
"sys.modules",
{
@ -42,17 +44,20 @@ def server(hermes_home):
},
):
mod = importlib.import_module("tui_gateway.server")
yield mod
# Reset module-level session state without re-importing. importlib.reload
# would re-register the module's atexit hooks; duplicated hooks race the
# stderr buffer at interpreter shutdown (Fatal Python error:
# _enter_buffered_busy) — same class as PR #34217.
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
# NOTE: _methods is intentionally NOT cleared — it's populated at import
# time and would only repopulate via reload.
mod._db = None
methods = dict(mod._methods)
yield mod
# Restore in place instead of clear+reload: importlib.reload
# re-registers atexit hooks (duplicate ThreadPoolExecutor shutdowns
# race the stderr buffer at interpreter exit — same class as PR #34217)
# and re-captures module-level paths like _hermes_home against this
# test's soon-deleted tmpdir, breaking later files in the same process.
mod._methods.clear()
mod._methods.update(methods)
mod._sessions.clear()
mod._pending.clear()
mod._answers.clear()
mod._db = None
@pytest.fixture()