fix(runtime): preserve listener startup permission diagnostics

loopx/control_plane/effect_runtime.py

@@ -833,13 +833,12 @@ def _read_startup_stderr(capture: IO[bytes]) -> bytes:
 
 
 def _startup_diagnostic(raw: bytes) -> tuple[str, str] | None:
-    """Return the typed diagnostic a rejected managed runtime published.
+    """Return the typed diagnostic a managed runtime published before readiness.
 
-    A server that rejects its own startup configuration writes one JSON
-    envelope to stderr and exits before it listens, so a matching envelope is
-    the authoritative configuration error. Any other stderr content, such as a
-    Node.js stack trace, is not a typed diagnostic and must not be reported as
-    one.
+    Recognized startup failures publish a JSON envelope to stderr before exit,
+    so a matching envelope is the authoritative startup error. Any other stderr
+    content, such as a Node.js stack trace, is not a typed diagnostic and must
+    not be reported as one.
     """
 
     if not raw:
@@ -925,7 +924,7 @@ def _start_runtime(*, fingerprint: str, info_path: Path) -> dict[str, Any]:
         # Fixed daemon context, never a model-supplied executable argument.
         environment["LOOPX_EFFECT_RUNTIME_PYTHON"] = sys.executable
         # Capture stderr so a rejected startup can publish a typed
-        # configuration diagnostic instead of a bare exit status. The capture
+        # startup diagnostic instead of a bare exit status. The capture
         # is an unlinked temporary file, so it cannot deadlock the child on a
         # full pipe and it leaves no stale path behind.
         with tempfile.TemporaryFile() as startup_stderr:
@@ -976,6 +975,8 @@ def _start_runtime(*, fingerprint: str, info_path: Path) -> dict[str, Any]:
                     )
                     if diagnostic is not None:
                         code, message = diagnostic
+                        if code == "runtime_host_permission_denied":
+                            raise EffectRuntimeHostPermissionError()
                         raise EffectRuntimeStartupError(
                             message,
                             diagnostic_code=code,

loopx/control_plane/effect_runtime_server.ts

@@ -290,7 +290,17 @@ server.on("close", () => {
   })().finally(() => process.exit(0));
 });
 
+function failListenerStartup(error: Error): never {
+  const { code } = effectRuntimeErrorPayload(error);
+  const diagnosticCode = code === "io_permission_denied"
+    ? "runtime_host_permission_denied" : code;
+  failStartup(diagnosticCode,
+    `TypeScript Effect runtime could not start its loopback listener (${diagnosticCode})`);
+}
+
+server.once("error", failListenerStartup);
 server.listen(0, "127.0.0.1", async () => {
+  server.removeListener("error", failListenerStartup);
   try {
     const address = server.address();
     if (!address || typeof address === "string") throw new Error("invalid address");

tests/control_plane/test_effect_runtime_listener_startup.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from loopx.control_plane import effect_runtime
+
+
+@pytest.mark.parametrize(
+    ("node_code", "diagnostic_code", "launch_count"),
+    [
+        ("EPERM", "runtime_host_permission_denied", 1),
+        ("EACCES", "runtime_host_permission_denied", 1),
+        ("EMFILE", "io_process_descriptor_limit", 2),
+        ("EADDRINUSE", "io_failure", 2),
+    ],
+)
+def test_listener_failure_preserves_safe_diagnostic_and_retry_boundary(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    node_code: str,
+    diagnostic_code: str,
+    launch_count: int,
+) -> None:
+    for variable in ("TMPDIR", "TEMP", "TMP"):
+        monkeypatch.setenv(variable, str(tmp_path))
+    monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))
+    launches = tmp_path / "launches.txt"
+    hook = tmp_path / "denied-listener.mjs"
+    # Inject only the network failure; the real daemon and Python bridge run.
+    hook.write_text(
+        'import { appendFileSync } from "node:fs";\n'
+        'import { Server } from "node:net";\n'
+        'Server.prototype.listen = function () {\n'
+        f'  appendFileSync({json.dumps(str(launches))}, "attempt\\n");\n'
+        '  process.nextTick(() => this.emit("error", Object.assign(\n'
+        '    new Error("private-listener-fixture"),\n'
+        f'    {{ code: {json.dumps(node_code)}, syscall: "listen" }})));\n'
+        '  return this;\n'
+        '};\n',
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("NODE_OPTIONS", f"--import={hook.as_uri()}")
+
+    with pytest.raises(effect_runtime.EffectRuntimeStartupError) as raised:
+        effect_runtime.effect_runtime_result("runtime.ping", {})
+
+    assert raised.value.diagnostic_code == diagnostic_code
+    assert isinstance(raised.value, effect_runtime.EffectRuntimeHostPermissionError) == (
+        diagnostic_code == "runtime_host_permission_denied"
+    )
+    assert "private-listener-fixture" not in str(raised.value)
+    assert str(tmp_path) not in str(raised.value)
+    assert launches.read_text(encoding="utf-8").splitlines() == ["attempt"] * launch_count
+    runtime_dir = effect_runtime._runtime_dir()
+    assert not list(runtime_dir.glob("runtime-*.json"))
+    assert not list(runtime_dir.glob("start-*.lock"))