test(e2e/windows): stamped transcripts, claims log, observed-only gates, standard-user control dir

tests/e2e/core/windows_update/_machine.py

@@ -44,6 +44,7 @@ import os
 import shutil
 import subprocess
 import tempfile
+import threading
 import time
 import uuid
 from dataclasses import dataclass, field
@@ -268,18 +269,33 @@ class Machine:
         self._seq += 1
         log = self.logs / f"{self._seq:02d}-{label}.log"
         started = time.monotonic()
-        with log.open("wb") as fh:
-            proc = subprocess.Popen(argv, cwd=cwd or self.profile, env=self.env(env_extra),
-                                    stdin=subprocess.DEVNULL, stdout=fh, stderr=subprocess.STDOUT)
-            try:
-                code = proc.wait(timeout=timeout)
-                self.timings.append((label, round(time.monotonic() - started, 1)))
-            except subprocess.TimeoutExpired:
-                subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"], capture_output=True, timeout=60)
-                proc.wait(timeout=60)
-                raise AssertionError(
-                    f"{label} did not finish within {timeout:.0f}s\n{self._tail(log)}\n{self.evidence()}")
-        return Run(code, _decode(log.read_bytes()), "")
+        chunks: list[bytes] = []
+        proc = subprocess.Popen(argv, cwd=cwd or self.profile, env=self.env(env_extra),
+                                stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+
+        def pump() -> None:
+            # Every transcript line carries its offset: install.ps1 prints no timestamps, and
+            # a stall is only visible as a gap. The pump outlives the command (daemon) so a
+            # detached grandchild that inherited the pipe can never block on a full buffer.
+            with log.open("w", encoding="utf-8") as fh:
+                for raw in iter(proc.stdout.readline, b""):
+                    chunks.append(raw)
+                    fh.write(f"[{time.monotonic() - started:7.1f}s] {_decode(raw).rstrip()}\n")
+                    fh.flush()
+
+        reader = threading.Thread(target=pump, name=f"pump-{label}", daemon=True)
+        reader.start()
+        try:
+            code = proc.wait(timeout=timeout)
+            self.timings.append((label, round(time.monotonic() - started, 1)))
+        except subprocess.TimeoutExpired:
+            subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"], capture_output=True, timeout=60)
+            proc.wait(timeout=60)
+            reader.join(timeout=10)
+            raise AssertionError(
+                f"{label} did not finish within {timeout:.0f}s\n{self._tail(log)}\n{self.evidence()}") from None
+        reader.join(timeout=20)
+        return Run(code, _decode(b"".join(chunks)), "")
 
     @staticmethod
     def _tail(path: Path, n: int = 6000) -> str:
@@ -526,6 +542,8 @@ def failure_line(run: Run) -> str:
 
 def fail_with(machine: Machine, message: str, run: Run | None = None) -> str:
     """Assertion text: the claim first (gates match on it), then the transcript and evidence."""
+    with contextlib.suppress(OSError), (machine.logs / "claims.txt").open("a", encoding="utf-8") as fh:
+        fh.write(message.splitlines()[0] + "\n")  # CI prints no reason for a passing file's xfails
     tail = f"\n--- transcript rc={run.returncode} ---\n{run.stdout[-6000:]}" if run is not None else ""
     return f"{message}{tail}\n{machine.evidence()}"
 

tests/e2e/core/windows_update/test_interpreter.py

@@ -37,12 +37,9 @@ from tests.fakes.fake_llm_provider import FakeLLMServer
 pytestmark = [pytest.mark.platforms("windows"), pytest.mark.integration,
               pytest.mark.live_system_guard_bypass, REQUIRES_OPT_IN]
 
-KNOWN = {
-    "path_python": (r"^the gateway runs on a Python outside the managed runtime",
-                    "gated on #123185: the Windows gateway boots under a newer system Python on PATH"),
-    "stale_venv": (r"^the gateway loaded the stale pre-PM in-tree venv",
-                   "gated on #123965/#123972: the Windows gateway adds the stale in-tree venv to sys.path"),
-}
+# Gates cover observed failures only. On main these cells pass: the gateway boots on the
+# managed Python and skips the stale venv (#123185, #123965 and #123972 do not reproduce).
+KNOWN: dict[str, tuple[str, str]] = {}
 _WORKER_DEATH = re.compile(r"^.*Supervised task \S+ died.*$", re.M)
 
 
@@ -65,11 +62,14 @@ def _seed_stale_venv(machine: Machine) -> Path:
     (venv / "Scripts").mkdir(exist_ok=True)
     (venv / "pyvenv.cfg").write_text(
         "home = C:\\Python311\ninclude-system-site-packages = false\nversion = 3.11.15\n", encoding="utf-8")
-    (site / "zz_e2e_stale_venv_probe.pth").write_text(
+    (machine.install_dir / _PROBE).write_text(
         f"import os; open(os.path.join({str(markers)!r}, str(os.getpid())), 'w').close()\n", encoding="utf-8")
     return markers
 
 
+_PROBE = Path("venv", "Lib", "site-packages", "zz_e2e_stale_venv_probe.pth")
+
+
 def _loaded_by(markers: Path) -> set[int]:
     return {int(p.name) for p in markers.iterdir() if p.name.isdigit()} if markers.is_dir() else set()
 
@@ -104,11 +104,13 @@ def journey(tmp_path_factory):
                                                            f"hermes update exited {update.returncode}", update))
                 if j.ok("update_ok"):
                     with machine.gateway_phase():
+                        j.results["probe_present"] = (machine.install_dir / _PROBE).is_file()
                         j.step("spawn", machine.spawn_gateway)
                         state = j.step("state", machine.wait_gateway_running)
                         if j.ok("state"):
                             time.sleep(15)  # supervised workers start ~2 s after boot; give them room to die
                             j.step("gateway_proc", lambda: _inspect(int(state["pid"])))
+                            j.results["loaded_by_gateway"] = _loaded_by(markers)
                         machine.kill_owned()  # nothing of this machine outlives its gateway phase
                     j.step("turn", lambda: one_shot_turn(machine, srv, "turn-stale-venv"))
             yield j
@@ -133,24 +135,25 @@ def test_gateway_runs_on_managed_python(journey: Journey) -> None:
 
 
 def test_gateway_does_not_load_stale_in_tree_venv(journey: Journey) -> None:
-    m, info, markers = journey.machine, journey["gateway_proc"], journey.results["markers"]
+    m, info = journey.machine, journey["gateway_proc"]
     stale = str(m.install_dir / "venv")
     published = [k for k in ("VIRTUAL_ENV", "PYTHONPATH")
                  if os.path.normcase(stale) in os.path.normcase(info[k])]
-    loaded = info["pid"] in _loaded_by(markers)
+    assert journey.results["probe_present"], fail_with(
+        m, f"harness: the stale-venv probe {_PROBE} was gone before the gateway started")
+    loaded = sorted(journey.results["loaded_by_gateway"])
     log = m.hermes_home / "logs" / "gateway.log"
     deaths = _WORKER_DEATH.findall(log.read_text(encoding="utf-8", errors="replace")) if log.is_file() else []
     with known_gate(KNOWN, "stale_venv"):
         assert not loaded and not published, fail_with(
-            m, f"the gateway loaded the stale pre-PM in-tree venv {stale} (site dir added in the gateway "
-               f"pid {info['pid']}={loaded}; published via {published or 'nothing'}; worker deaths: {deaths[:3]})")
+            m, f"the gateway loaded the stale pre-PM in-tree venv {stale} (site dir added by pids "
+               f"{loaded}, gateway pid {info['pid']}; published via {published or 'nothing'}; worker deaths: {deaths[:3]})")
     assert not deaths, fail_with(m, f"gateway supervised workers died: {deaths[:5]}")
 
 
 def test_cli_turn_ignores_stale_venv_and_path_python(journey: Journey) -> None:
     m, turn, markers = journey.machine, journey["turn"], journey.results["markers"]
-    gateway_pid = journey["gateway_proc"]["pid"] if journey.ok("gateway_proc") else None
-    others = sorted(_loaded_by(markers) - {gateway_pid})
+    others = sorted(_loaded_by(markers) - set(journey.results.get("loaded_by_gateway", ())))
     assert turn.ok, fail_with(
         m, f"a turn with a stale in-tree venv and a system Python on PATH failed (reply printed="
            f"{turn.reply_id in turn.run.stdout}, prompt reached provider={turn.reached_wire})", turn.run)

tests/e2e/core/windows_update/test_paths_and_acls.py

@@ -34,9 +34,6 @@ pytestmark = [pytest.mark.platforms("windows"), pytest.mark.integration,
 
 PERSON = "Jörg Ñúñez"  # the profile is "Jörg Ñúñez hermes-e2e-<id>"
 KNOWN = {
-    "install": (r"^install\.ps1 failed for a profile path with non-ASCII characters and spaces: .*"
-                r"is not recognized as the name of a cmdlet",
-                "gated on #124526: a non-ASCII profile path breaks uv Python path resolution in python-deps"),
     "acl": (r"^managed tools are not executable by a non-elevated process after update: .*WinError 5",
             "gated on #122935: tools\\* keep a hardened DACL a standard-user token cannot execute"),
 }
@@ -112,7 +109,11 @@ def run_as_standard_user(argv: list[str], timeout: float = 120.0) -> tuple[int |
 
 def _standard_user_token_is_really_restricted(scratch: Path) -> str:
     """Harness control: the Basic User token runs system binaries and is not an admin."""
-    out = scratch / "whoami-groups.txt"
+    # A dir the restricted token may write: the machine root is Administrators-owned.
+    box = scratch / "basic-user"
+    box.mkdir(parents=True, exist_ok=True)
+    subprocess.run(["icacls", str(box), "/grant", "*S-1-1-0:(OI)(CI)F"], capture_output=True, check=True, timeout=60)
+    out = box / "whoami-groups.txt"
     code, err = run_as_standard_user(
         [r"C:\Windows\System32\cmd.exe", "/d", "/c", f'whoami /groups /fo csv > "{out}"'])
     assert code == 0, f"harness: cannot launch cmd.exe with a Basic User token (code={code}, {err})"