diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index bfc284e0d4..e4f7206de1 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -389,6 +389,14 @@ it differs from the live server before dispatching expensive work. Stop only that isolated runtime after its requests settle, retaining the same temporary environment for shutdown; never restart a live server as test cleanup. +Automatic test cleanup must verify namespace ownership as well as PID/token +identity. An ordinary copy of a valid locator inside a test directory is not +permission to stop its original server. The private authenticated `runtime.ping` +reports the physical locator path frozen at publication; compare it with the +exact disposable locator before requesting shutdown, including when pytest has +already removed the fixture directory. Missing provenance remains unknown and +cannot authorize cleanup. This observation is not a Goal or execution grant. + Process isolation still shares machine resources. Run matched timing arms sequentially without overlapping builds or recovery work. If interference is discovered, retain failures and durable-receipt evidence, mark latency samples diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 57bd22f70b..64b388427f 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -54,6 +54,8 @@ function lazyHandler( export interface EffectRuntimeHandlerContext { fingerprint: string; + /** Physical publication path observed by the managed server, not its caller. */ + readonly infoPath?: string; requestShutdown: () => void; } @@ -248,7 +250,8 @@ export function createEffectRuntimeHandlers( return new Map([ [ "runtime.ping", - () => ({ ready: true, pid: process.pid, fingerprint: context.fingerprint }), + () => ({ ready: true, pid: process.pid, fingerprint: context.fingerprint, + ...(context.infoPath === undefined ? {} : {info_path: context.infoPath}) }), ], [ "runtime.shutdown", diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index 2ef72aac66..7ddbff9586 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -1,6 +1,6 @@ import { writeSync } from "node:fs"; import { createServer, type Socket } from "node:net"; -import { type FileHandle } from "node:fs/promises"; +import { realpath, type FileHandle } from "node:fs/promises"; import type { JsonObject } from "./effect_program.ts"; import { @@ -110,6 +110,7 @@ function parseIdleMs(raw: string | undefined): number { } const infoPath = parseArg("--info"); +let publishedInfoPath: string | undefined; const fingerprint = parseArg("--fingerprint"); const token = requiredString(process.env.LOOPX_EFFECT_RUNTIME_TOKEN, "runtime token"); const idleMs = parseIdleMs(process.env.LOOPX_EFFECT_RUNTIME_IDLE_MS); @@ -121,6 +122,7 @@ let resolvePublication: () => void; const publicationReady = new Promise((resolve) => { resolvePublication = resolve; }); const handlers = createEffectRuntimeHandlers({ fingerprint, + get infoPath() { return publishedInfoPath; }, requestShutdown: () => { shutdownRequested = true; }, @@ -305,6 +307,10 @@ server.listen(0, "127.0.0.1", async () => { // serving a goal is not necessarily the one the caller resolves from PATH. runtime_identity: sqliteRuntimeIdentity(), }); + // Freeze the native publication namespace while it exists. Test cleanup + // may observe it after the fixture directory was removed; a copied + // locator's contents cannot rewrite this authenticated boot observation. + publishedInfoPath = await realpath(infoPath); }); publicationComplete = true; resolvePublication(); diff --git a/loopx/presentation/codex_attached_api.py b/loopx/presentation/codex_attached_api.py index 6f82a78206..922916f21e 100644 --- a/loopx/presentation/codex_attached_api.py +++ b/loopx/presentation/codex_attached_api.py @@ -169,16 +169,19 @@ def emit(packet: dict[str, Any]) -> None: current_binding(self._codex_goal(goal_id), agent_id, expected) emit(pending) previous, pending, emitted_at = pending["records"], None, now - elif now - emitted_at >= 1: + elif now - emitted_at >= 0.5: + # An unchanged but busy native stream still needs HTTP + # writes to notice a departed reader. Two probes per + # second bound TCP's possible first successful close write. http.wfile.write(b": observing\n\n") http.wfile.flush() emitted_at = now - except DesktopProtocolError: - emit({"kind": "unavailable", "retry": False}) - except DesktopUnavailable: - emit({"kind": "unavailable", "retry": True}) - except (ValueError, KeyError, TypeError): - emit({"kind": "unavailable", "retry": False}) + except (DesktopUnavailable, ValueError, KeyError, TypeError) as error: + try: + emit({"kind": "unavailable", "retry": isinstance(error, DesktopUnavailable) + and not isinstance(error, DesktopProtocolError)}) + except OSError: + pass # A departed HTTP reader cannot receive an error packet. except (BrokenPipeError, ConnectionResetError, OSError): return finally: diff --git a/tests/managed_effect_runtime_fixture.py b/tests/managed_effect_runtime_fixture.py index bdd9b5aeb9..ce580c9bc2 100644 --- a/tests/managed_effect_runtime_fixture.py +++ b/tests/managed_effect_runtime_fixture.py @@ -28,6 +28,9 @@ def capture_disposable_effect_runtimes(root: Path) -> list[tuple[Path, str]]: def stop_disposable_effect_runtimes( root: Path, captured: list[tuple[Path, str]] | None = None, ) -> None: + if root.is_symlink(): + return + owned_root = root.resolve() records = dict.fromkeys([*(captured or []), *capture_disposable_effect_runtimes(root)]) if not records: return @@ -36,6 +39,8 @@ def stop_disposable_effect_runtimes( # are restored; the authenticated ping still verifies the exact live owner. with tempfile.TemporaryDirectory(prefix="loopx-test-runtime-cleanup-") as scratch: for locator, raw in records: + if locator.is_symlink() or not locator.resolve().is_relative_to(owned_root): + continue validation_file = Path(scratch) / locator.name validation_file.write_text(raw, encoding="utf-8") info = effect_runtime._read_info(validation_file, fingerprint=None) @@ -52,6 +57,12 @@ def stop_disposable_effect_runtimes( if (not isinstance(ping, dict) or ping.get("ready") is not True or ping.get("pid") != info["pid"] or ping.get("fingerprint") != info["fingerprint"]): continue + # Authentication proves identity, not namespace ownership. Require + # the server's frozen physical publication path, never a path + # claimed by copied JSON. Unknown/older servers remain untouched. + published = ping.get("info_path") + if not isinstance(published, str) or Path(published) != locator.resolve(): + continue result = effect_runtime._stop_effect_runtime( locator, info, timeout=5, wait_for_process_exit=True, ) diff --git a/tests/test_managed_effect_runtime_fixture.py b/tests/test_managed_effect_runtime_fixture.py index 7ef8c10023..1be72ed3c9 100644 --- a/tests/test_managed_effect_runtime_fixture.py +++ b/tests/test_managed_effect_runtime_fixture.py @@ -10,7 +10,7 @@ import pytest -from managed_effect_runtime_fixture import stop_disposable_effect_runtimes +from managed_effect_runtime_fixture import capture_disposable_effect_runtimes, stop_disposable_effect_runtimes from loopx.control_plane import effect_runtime @@ -133,8 +133,41 @@ def test_cleanup_does_not_hide_an_unfinished_shutdown(tmp_path, monkeypatch): "pid": 123, "fingerprint": fingerprint, }) monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *_a, **_k: { - "result": {"ready": True, "pid": 123, "fingerprint": fingerprint}, + "result": {"ready": True, "pid": 123, "fingerprint": fingerprint, + "info_path": str(locator.resolve())}, }) monkeypatch.setattr(effect_runtime, "_stop_effect_runtime", lambda *_a, **_k: {"stopped": False}) with pytest.raises(AssertionError, match="Disposable runtime did not exit"): stop_disposable_effect_runtimes(tmp_path) + + +@pytest.mark.parametrize("removed", [False, True]) +def test_regular_locator_copy_cannot_authorize_foreign_runtime_shutdown(tmp_path, monkeypatch, removed): + foreign, owned = tmp_path / "foreign", tmp_path / "owned" + foreign.mkdir() + owned.mkdir() + with monkeypatch.context() as isolated: + isolated.setattr(effect_runtime.tempfile, "tempdir", str(foreign)) + effect_runtime.effect_runtime_result("runtime.ping", {}) + locator = effect_runtime._runtime_info_path(effect_runtime._runtime_fingerprint()) + info = json.loads(locator.read_text(encoding="utf-8")) + try: + copied = owned / locator.parent.name / locator.name + copied.parent.mkdir() + # Even a claimed publication path in the untrusted JSON is not + # proof of where the authenticated server originally published. + copied.write_text(json.dumps({**info, "info_path": str(copied)}), encoding="utf-8") + captured = capture_disposable_effect_runtimes(owned) + assert len(captured) == 1 and not copied.is_symlink() + if removed: + copied.unlink() + copied.parent.rmdir() + owned.rmdir() + stop_disposable_effect_runtimes(owned, captured if removed else None) + # Do not use normal ping: it could restart a killed server and + # obscure the original runtime's loss. + response = effect_runtime._request_with_info(info, request_id="foreign-survives", + method="runtime.ping", params={}, timeout=2) + assert response["result"]["pid"] == info["pid"] + finally: + effect_runtime._stop_effect_runtime(locator, info, timeout=5, wait_for_process_exit=True) diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py index 6d474a3769..fedad177c5 100644 --- a/tests/test_projection_envelope.py +++ b/tests/test_projection_envelope.py @@ -602,6 +602,9 @@ def test_receipt_case_collection_does_not_depend_on_the_collecting_process_clock capture_output=True, text=True, timeout=60, ) assert result.returncode == 0, result.stdout + result.stderr + # Isolation must not hide a collector that actually writes a default. + assert not (home / ".loopx").exists() + assert not (home / ".codex" / "loopx").exists() nodeids = [line for line in result.stdout.splitlines() if line.startswith(target + "[")] assert len(nodeids) == len(set(nodeids)) == 6 collections.append(nodeids)