fix(runtime): prevent cross-namespace fixture shutdown

docs/development/testing-and-quality.md

@@ -444,18 +444,31 @@ Windows 文件锁只在已观察到争用后，对独占创建返回的 `EPERM`
 初次权限拒绝、创建后的错误和其它平台仍直接失败；持续拒绝保留原错误。
 故障注入验证重试边界，原生 Windows 运行仍须单独验收。
 
-Smoke cleanup owns every valid runtime locator in its newly created private
-temporary namespace, including runtimes loaded by temporary base checkouts.
-It uses the existing authenticated shutdown protocol, then observes actual
+Test teardown and smoke cleanup require the authenticated server to confirm its
+exact canonical serving locator. A valid locator copied into a disposable
+directory proves server identity, not ownership of its namespace. The internal
+`expected_info_path` request opts into this proof before shutdown and fences the
+shutdown again at the TypeScript owner. Ordinary ping, shutdown and operator
+restart requests retain their existing behavior. Missing proof, including from
+an older runtime that ignores the parameter, never authorizes shutdown. Pytest
+ignores borrowed or unproven locators; smoke cleanup reports failure and retains
+the fixture for inspection, rather than treating legacy cleanup as qualified.
+Owned source revisions, including copied packages, still use authenticated
+shutdown and observe actual
 process exit before deleting the namespace; serving-locator turnover alone
 does not establish that Node has finished its exit-time cache writes. Namespace
 and locator symlinks are not followed. A cleanup failure retains the fixture
 and reports a failed check with the original command output. Operator restart
 keeps its existing serving-handoff semantics. This fixture repair does not
 qualify native intermittent failures or sustained operation.
 
-Smoke 清理覆盖新建私有临时目录中所有合法 runtime locator，包括临时基线 checkout
-启动的进程。它复用既有认证 shutdown，等实际进程退出后再删目录；定位文件退休不代表
+测试 teardown 与 smoke 清理要求认证服务端确认精确的规范化服务定位路径。合法定位
+文件被复制到临时目录，只证明服务身份，不证明目录归属。内部 `expected_info_path`
+请求在 shutdown 前取得证明，并由 TypeScript owner 在关闭时再次校验；普通 ping、
+shutdown 和操作员 restart 沿用既有行为。缺失证明时，包括旧 runtime 忽略该参数，
+均不发送关闭请求。Pytest 忽略借用或未证明的定位记录；smoke 报告失败并保留目录，
+不把旧版本清理视为已验收。属于该目录的源码及复制包进程仍复用既有认证 shutdown，
+等实际进程退出后再删目录；定位文件退休不代表
 Node 的退出缓存写入已经结束。清理不跟随 namespace 或 locator 符号链接；无法证明
 退出时保留 fixture，以失败结果报告并保留原命令输出。操作员 restart 仍沿用既有服务
 交接语义；该 fixture 修复不认证原生间歇故障或持续运行。

loopx/canary/runner.py

@@ -278,11 +278,14 @@ def _retire_fixture_runtimes(root: Path, *, timeout: float = 15) -> None:
                 continue
             if path.name != f"runtime-{info['fingerprint'][:16]}.json":
                 raise RuntimeError(f"invalid managed runtime fingerprint: {path.name}")
-            remaining = (deadline - time.monotonic()) / 2
+            # Ownership probe, shutdown request and observed exit share the
+            # existing cleanup deadline; the new proof cannot extend it.
+            remaining = (deadline - time.monotonic()) / 3
             if remaining <= 0:
                 raise RuntimeError("managed runtime cleanup deadline exceeded")
             result = _stop_effect_runtime(
                 path, info, timeout=min(5, remaining), wait_for_process_exit=True,
+                expected_info_path=path,
             )
             if result["status"] != "stopped":
                 raise RuntimeError(f"managed runtime still exiting: {path.name}")

loopx/control_plane/effect_runtime.py

@@ -617,12 +617,15 @@ def _stop_effect_runtime(
     *,
     timeout: float,
     wait_for_process_exit: bool = False,
+    expected_info_path: Path | None = None,
 ) -> dict[str, Any]:
     """Transport shutdown for a validated locator, optionally awaiting OS exit.
 
     Operator restart keeps its existing serving-locator turnover semantics.
     An isolated fixture instead needs the original process to stop writing
-    before deleting its namespace. TS still owns admission, drain and retirement.
+    before deleting its namespace. An optional expected locator also fences the
+    shutdown at the serving TS owner; copied metadata is not namespace ownership.
+    TS still owns admission, drain and retirement.
     """
     identity = runtime_identity_from_info(info)
     if info is None:
@@ -633,14 +636,29 @@ def _stop_effect_runtime(
             "previous_runtime_identity": identity,
             "info_path": str(info_path),
         }
+    params = {} if expected_info_path is None else {
+        "expected_info_path": str(expected_info_path.resolve()),
+    }
+    if expected_info_path is not None:
+        response = _request_with_info(
+            info, request_id=uuid.uuid4().hex, method="runtime.ping", params=params,
+            timeout=min(1, timeout),
+        )
+        ping = response.get("result")
+        if (not isinstance(ping, dict) or ping.get("ready") is not True
+                or ping.get("pid") != info.get("pid") or ping.get("fingerprint") != info.get("fingerprint")
+                or ping.get("owns_locator") is not True):
+            # Older runtimes can ignore unknown parameters. Require a positive
+            # serving-owner proof before sending any shutdown, even to them.
+            raise EffectRuntimeRejected("Runtime did not confirm the expected serving locator")
     pid = info.get("pid")
     serving_token = info.get("token")
     try:
         _request_with_info(
             info,
             request_id=uuid.uuid4().hex,
             method="runtime.shutdown",
-            params={},
+            params=params,
             timeout=timeout,
         )
     except (

loopx/control_plane/effect_runtime_handlers.ts

@@ -54,9 +54,19 @@ function lazyHandler<Module>(
 
 export interface EffectRuntimeHandlerContext {
   fingerprint: string;
+  /** Canonical serving locator, captured before publication, never caller data. */
+  infoPath?: string;
   requestShutdown: () => void;
 }
 
+function runtimeLocatorMatch(
+  context: EffectRuntimeHandlerContext, params: JsonObject,
+): boolean | undefined {
+  if (!("expected_info_path" in params)) return undefined;
+  const expected = requiredString(params.expected_info_path, "expected_info_path");
+  return context.infoPath !== undefined && expected === context.infoPath;
+}
+
 function asObject(value: unknown): JsonObject {
   return typeof value === "object" && value !== null && !Array.isArray(value)
     ? (value as JsonObject)
@@ -248,11 +258,18 @@ export function createEffectRuntimeHandlers(
   return new Map<string, EffectRuntimeHandler>([
     [
       "runtime.ping",
-      () => ({ ready: true, pid: process.pid, fingerprint: context.fingerprint }),
+      (params) => {
+        const ownsLocator = runtimeLocatorMatch(context, params);
+        return { ready: true, pid: process.pid, fingerprint: context.fingerprint,
+          ...(ownsLocator === undefined ? {} : { owns_locator: ownsLocator }) };
+      },
     ],
     [
       "runtime.shutdown",
-      () => {
+      (params) => {
+        // A copied authentic locator can authenticate to a different namespace.
+        // Fence the effect too; an earlier ping is not shutdown authorization.
+        if (runtimeLocatorMatch(context, params) === false) return { stopped: false };
         context.requestShutdown();
         return { stopped: true };
       },

loopx/control_plane/effect_runtime_server.ts

@@ -1,11 +1,13 @@
 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 { basename, dirname, join } from "node:path";
 
 import type { JsonObject } from "./effect_program.ts";
 import {
   createEffectRuntimeHandlers,
   dispatchEffectRuntimeMethod,
+  type EffectRuntimeHandlerContext,
 } from "./effect_runtime_handlers.ts";
 import {
   EffectRuntimeRequestError,
@@ -117,12 +119,13 @@ let resolveDrain: (() => void) | undefined;
 let publicationComplete = false;
 let resolvePublication: () => void;
 const publicationReady = new Promise<void>((resolve) => { resolvePublication = resolve; });
-const handlers = createEffectRuntimeHandlers({
+const handlerContext: EffectRuntimeHandlerContext = {
   fingerprint,
   requestShutdown: () => {
     shutdownRequested = true;
   },
-});
+};
+const handlers = createEffectRuntimeHandlers(handlerContext);
 
 function resetIdleTimer(server: ReturnType<typeof createServer>): void {
   clearTimeout(idleTimer);
@@ -291,6 +294,9 @@ server.listen(0, "127.0.0.1", async () => {
   try {
     const address = server.address();
     if (!address || typeof address === "string") throw new Error("invalid address");
+    // Resolve aliases while the private directory still exists. Keep this
+    // serving identity after pytest retention removes the original directory.
+    handlerContext.infoPath = join(await realpath(dirname(infoPath)), basename(infoPath));
     await withFileMutationLock(infoPath, async () => {
       await atomicWriteJson(infoPath, {
         schema_version: INFO_SCHEMA,

tests/canary/test_telemetry_isolation.py

@@ -351,3 +351,42 @@ def test_fixture_retirement_does_not_follow_runtime_namespace_links(tmp_path, mo
     monkeypatch.setattr(effect_runtime, "_read_info", lambda *a, **kw: pytest.fail("followed external locator"))
     runner._retire_fixture_runtimes(root)
     assert locator.read_text() == "external runtime metadata"
+
+
+def test_smoke_refuses_a_copied_authentic_locator_and_retains_its_fixture(tmp_path, monkeypatch):
+    from loopx.control_plane import effect_runtime
+
+    foreign = tmp_path / "foreign"
+    foreign.mkdir()
+    marker = tmp_path / "owned-root.txt"
+    examples = tmp_path / "examples"
+    examples.mkdir()
+    retained = None
+    with monkeypatch.context() as isolated:
+        isolated.setattr(effect_runtime.tempfile, "tempdir", str(foreign))
+        original = 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"))
+        (examples / "borrow.py").write_text(
+            "import os,shutil\nfrom pathlib import Path\n"
+            f"source=Path({str(locator)!r})\n"
+            "root=Path(os.environ['TMPDIR'])\n"
+            "copy=root/source.parent.name/source.name\n"
+            "copy.parent.mkdir()\nshutil.copyfile(source,copy)\n"
+            f"Path({str(marker)!r}).write_text(str(root))\n"
+            "print('original smoke output')\n", encoding="utf-8",
+        )
+        isolated.setattr(runner, "REPO_ROOT", tmp_path)
+        try:
+            result = runner._run_check({"command": "python examples/borrow.py"}, timeout_seconds=5)
+            retained = Path(marker.read_text())
+            assert result["status"] == "failed" and not result["ok"]
+            assert result["returncode"] == 0 and "original smoke output" in result["stdout_tail"]
+            assert "expected serving locator" in result["reason"] and retained.is_dir()
+            response = effect_runtime._request_with_info(info, request_id="foreign-still-serving",
+                method="runtime.ping", params={}, timeout=2)
+            assert response["result"] == original
+        finally:
+            effect_runtime._stop_effect_runtime(locator, info, timeout=5, wait_for_process_exit=True)
+            if retained is not None and retained.exists():
+                shutil.rmtree(retained)

tests/control_plane_ts/effect_runtime_handlers.test.ts

@@ -11,6 +11,55 @@ const handlers = createEffectRuntimeHandlers({
   requestShutdown: () => undefined,
 });
 
+test("runtime locator proof is opt-in and comes from the serving context", async () => {
+  const infoPath = "/disposable/serving/runtime.json";
+  const scoped = createEffectRuntimeHandlers({fingerprint: "fixture", infoPath, requestShutdown() {}});
+  assert.deepEqual(await dispatchEffectRuntimeMethod(scoped, "runtime.ping", {}),
+    {ready: true, pid: process.pid, fingerprint: "fixture"});
+  for (const expected of [infoPath, "/borrowed/runtime.json", `${infoPath}.copy`]) {
+    assert.deepEqual(await dispatchEffectRuntimeMethod(scoped, "runtime.ping", {
+      expected_info_path: expected, info_path: expected,
+    }), {ready: true, pid: process.pid, fingerprint: "fixture", owns_locator: expected === infoPath});
+  }
+  assert.equal((await dispatchEffectRuntimeMethod(handlers, "runtime.ping", {
+    expected_info_path: infoPath,
+  }) as {owns_locator: boolean}).owns_locator, false);
+});
+
+test("guarded shutdown checks ownership again and keeps ordinary shutdown unchanged", async () => {
+  let shutdowns = 0;
+  const infoPath = "/disposable/serving/runtime.json";
+  const scoped = createEffectRuntimeHandlers({fingerprint: "fixture", infoPath,
+    requestShutdown() { shutdowns++; }});
+  await dispatchEffectRuntimeMethod(scoped, "runtime.ping", {expected_info_path: infoPath});
+  assert.deepEqual(await dispatchEffectRuntimeMethod(scoped, "runtime.shutdown", {
+    expected_info_path: "/borrowed/runtime.json",
+  }), {stopped: false});
+  assert.equal(shutdowns, 0);
+  assert.deepEqual(await dispatchEffectRuntimeMethod(scoped, "runtime.shutdown", {
+    expected_info_path: infoPath,
+  }), {stopped: true});
+  assert.equal(shutdowns, 1);
+  assert.deepEqual(await dispatchEffectRuntimeMethod(scoped, "runtime.shutdown", {}), {stopped: true});
+  assert.equal(shutdowns, 2);
+  assert.deepEqual(await dispatchEffectRuntimeMethod(handlers, "runtime.shutdown", {
+    expected_info_path: infoPath,
+  }), {stopped: false});
+});
+
+test("an explicitly malformed locator guard never falls back to unguarded shutdown", async () => {
+  let shutdowns = 0;
+  const scoped = createEffectRuntimeHandlers({fingerprint: "fixture", infoPath: "/runtime.json",
+    requestShutdown() { shutdowns++; }});
+  for (const expected of [null, "", 12, {}, []]) {
+    for (const method of ["runtime.ping", "runtime.shutdown"]) {
+      await assert.rejects(dispatchEffectRuntimeMethod(scoped, method, {expected_info_path: expected}),
+        /expected_info_path must be a non-empty string/);
+    }
+  }
+  assert.equal(shutdowns, 0);
+});
+
 test("runtime boundary rejects incomplete settlement identity", async () => {
   await assert.rejects(
     dispatchEffectRuntimeMethod(handlers, "settlement.identity", {

tests/managed_effect_runtime_fixture.py

@@ -3,7 +3,6 @@
 
 from pathlib import Path
 import tempfile
-from uuid import uuid4
 
 from loopx.control_plane import effect_runtime
 
@@ -33,7 +32,7 @@ def stop_disposable_effect_runtimes(
         return
     # pytest may remove passed tmp_path directories during fixture teardown.
     # Validate the captured bytes with the existing reader after monkeypatches
-    # are restored; the authenticated ping still verifies the exact live owner.
+    # are restored; the authenticated server must also confirm its own locator.
     with tempfile.TemporaryDirectory(prefix="loopx-test-runtime-cleanup-") as scratch:
         for locator, raw in records:
             validation_file = Path(scratch) / locator.name
@@ -42,17 +41,11 @@ def stop_disposable_effect_runtimes(
             if info is None or locator.name != f"runtime-{info['fingerprint'][:16]}.json":
                 continue
             try:
-                response = effect_runtime._request_with_info(
-                    info, request_id=uuid4().hex, method="runtime.ping", params={}, timeout=1,
+                result = effect_runtime._stop_effect_runtime(
+                    locator, info, timeout=5, wait_for_process_exit=True,
+                    expected_info_path=locator,
                 )
             except (OSError, RuntimeError):
                 # Corrupt, stale or rejected metadata cannot authorize shutdown.
                 continue
-            ping = response.get("result")
-            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
-            result = effect_runtime._stop_effect_runtime(
-                locator, info, timeout=5, wait_for_process_exit=True,
-            )
             assert result["stopped"], f"Disposable runtime did not exit: {locator}"