From e2e6993c732ddcfc34a6eea93cfdbeeec4eb3fb4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:00:44 +0800 Subject: [PATCH] test(runtime): retain safe IO causes for admitted drain failures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../runtime_retirement_diagnostics.py | 54 ++++++- .../test_effect_runtime_integration.py | 134 ++++++++++++++---- 2 files changed, 156 insertions(+), 32 deletions(-) diff --git a/tests/control_plane/runtime_retirement_diagnostics.py b/tests/control_plane/runtime_retirement_diagnostics.py index 300fdd5358..096a869575 100644 --- a/tests/control_plane/runtime_retirement_diagnostics.py +++ b/tests/control_plane/runtime_retirement_diagnostics.py @@ -1,9 +1,10 @@ -"""Failure-only observations for the copied runtime retirement fixture.""" +"""Failure-only observations for copied runtime drain and retirement fixtures.""" from __future__ import annotations import json import os from pathlib import Path +import shutil from loopx.control_plane import effect_runtime @@ -11,9 +12,12 @@ EVENTS = frozenset({"shutdown_requested", "close", "flush_enter", "flush_done", "locator_read_done", "locator_read_error", "locator_unlink_enter", "locator_unlink_done", "locator_unlink_error", "cleanup_finally", "lock_attempt", "lock_acquired", "lock_acquire_error", "lock_release_error", - "lock_release_result"}) + "lock_release_result", "request_error"}) CODES = frozenset({"EACCES", "EPERM", "ENOENT", "EEXIST", "EIO", "EBUSY", "ENOTEMPTY", "EBADF", - "ENOSPC", "EMFILE", "ENFILE", "ETIMEDOUT", "EINTR", "ENOTDIR", "EROFS", "mutation_lock_timeout"}) + "ENOSPC", "EMFILE", "ENFILE", "ETIMEDOUT", "EINTR", "ENOTDIR", "EROFS", "EDQUOT", "EINVAL", + "EISDIR", "ENAMETOOLONG", "mutation_lock_timeout"}) +SYSCALLS = frozenset({"access", "chmod", "close", "fstat", "fsync", "mkdir", "open", "read", + "readdir", "rename", "rmdir", "stat", "unlink", "write"}) TRACE = ''' function fixtureTrace(event: string, details: Record = {}): void { try { console.error("loopx_fixture_retirement:" + JSON.stringify({ @@ -64,6 +68,46 @@ def instrument_copied_runtime(root: Path) -> None: 'const released = await releaseFileMutationLock(lock.targetPath, lock.token, null, true);\n' ' fixtureTrace("lock_release_result", {released});') io.write_text(text) + errors = root / "effect_runtime_errors.ts" + text = _replace(errors.read_text(), 'export const EFFECT_RUNTIME_ERROR_KINDS =', + TRACE + '\nexport const EFFECT_RUNTIME_ERROR_KINDS =') + text = _replace(text, ' if (error instanceof EffectRuntimeBoundaryError) {', + ' try {\n' + ' const system = error as {syscall?: unknown; errno?: unknown};\n' + ' fixtureTrace("request_error", {code: fixtureCode(error), syscall: system?.syscall, errno: system?.errno});\n' + ' } catch {}\n' + ' if (error instanceof EffectRuntimeBoundaryError) {') + errors.write_text(text) + + +def observe_copied_runtime(tmp_path: Path, monkeypatch, request) -> tuple[Path, list[int]]: + """Reuse the isolated source copy and retain stderr until failure readback.""" + source_root = effect_runtime._control_plane_root() + root = tmp_path / "control_plane" + for relative in effect_runtime._runtime_source_files(source_root): + target = root / relative + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source_root / relative, target) + instrument_copied_runtime(root) + captures: list[int] = [] + popen = effect_runtime.subprocess.Popen + + def close_captures(): + for descriptor in captures: + try: + os.close(descriptor) + except OSError: + pass + + request.addfinalizer(close_captures) + + def capture_stderr(args, *positional, **options): + retained_stderr(args, options, root, captures) + return popen(args, *positional, **options) + + monkeypatch.setattr(effect_runtime.subprocess, "Popen", capture_stderr) + monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: root) + return root, captures def trace_events(raw: bytes) -> list[dict]: @@ -87,6 +131,10 @@ def trace_events(raw: bytes) -> list[dict]: item["released"] = value["released"] if "code" in value: item["code"] = value["code"] if isinstance(value["code"], str) and value["code"] in CODES else None + if "syscall" in value: + item["syscall"] = value["syscall"] if isinstance(value["syscall"], str) and value["syscall"] in SYSCALLS else None + if type(value.get("errno")) is int: + item["errno"] = value["errno"] result.append(item) return result[-64:] diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index f64766f917..4a8ec22b06 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -12,7 +12,7 @@ import pytest from runtime_retirement_diagnostics import ( - failure_trace, file_state, instrument_copied_runtime, retained_stderr, trace_events, + failure_trace, file_state, observe_copied_runtime, trace_events, ) from loopx.control_plane import effect_runtime @@ -603,34 +603,8 @@ def test_runtime_decode_change_rotates_identity_and_starts_replacement( monkeypatch, request, ) -> None: - source_root = Path(effect_runtime.__file__).resolve().parent - staged_root = tmp_path / "control_plane" - for relative in effect_runtime._runtime_source_files(source_root): - source = source_root / relative - target = staged_root / relative - target.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source, target) - instrument_copied_runtime(staged_root) - captures: list[int] = [] - popen = effect_runtime.subprocess.Popen - - def close_captures(): - for descriptor in captures: - try: - os.close(descriptor) - except OSError: - pass - - request.addfinalizer(close_captures) - - def capture_runtime_stderr(args, *positional, **options): - retained_stderr(args, options, staged_root, captures) - return popen(args, *positional, **options) - - monkeypatch.setattr(effect_runtime.subprocess, "Popen", capture_runtime_stderr) - + staged_root, captures = observe_copied_runtime(tmp_path, monkeypatch, request) runtime_dir = tmp_path / "runtime" - monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: staged_root) monkeypatch.setattr(effect_runtime, "_runtime_dir", lambda: runtime_dir) monkeypatch.setenv("LOOPX_EFFECT_RUNTIME_IDLE_MS", "250") @@ -687,6 +661,53 @@ def test_runtime_retirement_trace_filters_private_stderr() -> None: assert trace_events(raw) == [{"event": "lock_release_error", "pid": 12, "ms": 1, "code": "EPERM"}] +@pytest.mark.parametrize("details,expected", [ + ({"code": "EISDIR", "syscall": "read", "errno": -21}, + {"code": "EISDIR", "syscall": "read", "errno": -21}), + ({"code": "EINVAL", "syscall": "rename", "errno": -22}, + {"code": "EINVAL", "syscall": "rename", "errno": -22}), + ({"code": "private-code", "syscall": "private/path", "errno": "private"}, + {"code": None, "syscall": None}), + ({"code": "EPERM", "syscall": "open", "errno": True}, + {"code": "EPERM", "syscall": "open"}), +]) +def test_runtime_request_trace_keeps_only_safe_system_error_fields(details, expected) -> None: + payload = {"event": "request_error", **details, "path": "private", "token": "private", + "message": "private", "stack": "private"} + raw = ("loopx_fixture_retirement:" + json.dumps(payload) + "\n").encode() + assert trace_events(raw) == [{"event": "request_error", **expected}] + + +def test_runtime_request_observation_preserves_error_payloads(tmp_path, monkeypatch, request) -> None: + source = effect_runtime._control_plane_root() / "effect_runtime_errors.ts" + root, _ = observe_copied_runtime(tmp_path, monkeypatch, request) + script = ''' +const rows = []; +for (const url of process.argv.slice(1)) { + const m = await import(url); + const boundary = new m.EffectRuntimeRequestError("safe detail"); + Object.defineProperty(boundary, "syscall", {get() {throw new Error("private getter");}}); + rows.push([ + m.effectRuntimeErrorPayload(boundary), + m.effectRuntimeErrorPayload({code: "EISDIR", syscall: "read", errno: -21, path: "private"}), + m.effectRuntimeErrorPayload(null), + ]); +} +console.log(JSON.stringify(rows)); +''' + result = effect_runtime.subprocess.run([ + effect_runtime._node_executable(), "--input-type=module", "--eval", script, + source.as_uri(), (root / "effect_runtime_errors.ts").as_uri(), + ], capture_output=True, text=True, timeout=10) + assert result.returncode == 0, result.stderr + expected = [ + {"kind": "request_rejected", "code": "invalid_request", "message": "safe detail"}, + {"kind": "io_permanent", "code": "io_is_directory", "message": "Effect runtime filesystem operation failed"}, + {"kind": "internal_failure", "code": "unexpected_handler_error", "message": "Effect runtime handler failed unexpectedly"}, + ] + assert json.loads(result.stdout) == [expected, expected] + + def test_typed_write_rejects_cross_effect_overwrite( tmp_path: Path, monkeypatch, @@ -1312,9 +1333,10 @@ def test_managed_runtime_releases_memory_after_idle_timeout( @pytest.mark.parametrize("shutdown", [False, True], ids=["idle", "shutdown"]) @pytest.mark.parametrize("disconnect", [False, True], ids=["connected", "timed-out"]) def test_runtime_drains_admitted_write_before_exit( - tmp_path: Path, monkeypatch, shutdown: bool, disconnect: bool, + tmp_path: Path, monkeypatch, request, capsys, shutdown: bool, disconnect: bool, ) -> None: """Socket lifetime must not define the lifetime of an accepted effect.""" + _, captures = observe_copied_runtime(tmp_path, monkeypatch, request) runtime_dir = tmp_path / "runtime" monkeypatch.setattr(effect_runtime, "_runtime_dir", lambda: runtime_dir) monkeypatch.setenv("LOOPX_EFFECT_RUNTIME_IDLE_MS", "250") @@ -1385,6 +1407,21 @@ def write() -> dict[str, object]: time.sleep(0.025) assert not info_path.exists(), "settled runtime must still retire when idle or stopped" assert json.loads(journal_path.read_text(encoding="utf-8")) == journal + assert capsys.readouterr().out == "", "healthy drains must stay quiet" + except Exception as error: + try: + print("runtime drain diagnostic: " + json.dumps({ + "shutdown": shutdown, + "disconnected": disconnect, + "permanent_io_error": isinstance(error, effect_runtime.EffectRuntimePermanentIOError), + "runtime": file_state(info_path, int(original["pid"])), + "journal": file_state(journal_path, int(original["pid"])), + "events": failure_trace(captures), + "trace_limit": "first 16 KiB / last 64 events; shared-offset reads may race writes", + })) + except Exception: + pass # Preserve the original failure even when observation fails. + raise finally: lock_path.unlink(missing_ok=True) cleanup_deadline = time.monotonic() + 3 @@ -1397,6 +1434,45 @@ def write() -> dict[str, object]: os.kill(int(original["pid"]), signal.SIGTERM) +@pytest.mark.parametrize("diagnostics_available", [True, False]) +def test_runtime_drain_io_failure_reports_cause_without_suppressing_error( + tmp_path: Path, monkeypatch, request, capsys, diagnostics_available: bool, +) -> None: + """A real directory substitution must fail the original accepted-write test.""" + journal = tmp_path / "admitted-turn.json" + lock = Path(f"{journal}.ts-effect.lock") + unlink = Path.unlink + injected = False + + def substitute_directory(path, *args, **kwargs): + nonlocal injected + if path == lock and not injected: + injected = True + journal.mkdir() + return unlink(path, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", substitute_directory) + if not diagnostics_available: + def failed_observation(_): + raise OSError("synthetic unavailable observation") + monkeypatch.setitem(globals(), "failure_trace", failed_observation) + with pytest.raises(effect_runtime.EffectRuntimePermanentIOError): + test_runtime_drains_admitted_write_before_exit( + tmp_path, monkeypatch, request, capsys, shutdown=True, disconnect=False, + ) + output = capsys.readouterr().out + prefix = "runtime drain diagnostic: " + reports = [json.loads(line[len(prefix):]) for line in output.splitlines() if line.startswith(prefix)] + assert len(reports) == (1 if diagnostics_available else 0) + if diagnostics_available: + assert reports[0]["permanent_io_error"] is True + errors = [row for row in reports[0]["events"] if row["event"] == "request_error"] + assert any(row.get("code") == "EISDIR" and row.get("syscall") in {"read", "open"} + and type(row.get("errno")) is int for row in errors) + assert journal.is_dir(), "diagnostics cannot pretend the write committed" + assert str(tmp_path) not in output and "fixture-holder" not in output + + def _envelope(code: str, received: str) -> bytes: """Frame one startup rejection the way the managed server publishes it.