fix(runtime): scope disposable cleanup and release departed readers

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

loopx/control_plane/effect_runtime_handlers.ts

@@ -54,6 +54,8 @@ function lazyHandler<Module>(
 
 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<string, EffectRuntimeHandler>([
     [
       "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",

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<void>((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();

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:

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,
             )

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)

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)