fix(runtime): enforce private IO and preserve Windows long paths

loopx/control_plane/effect_runtime.py

@@ -21,6 +21,7 @@
 
 from ..file_lock import process_is_alive
 from .runtime.file_reads import iter_binary_file_reads
+from .runtime.private_permissions import protect_private_path, require_private_path
 from .runtime.node_probe import (
     HOST_PERMISSION_RECOMMENDATION,
     MINIMUM_NODE_VERSION as MINIMUM_NODE_VERSION,
@@ -835,9 +836,13 @@ def _request_with_info(
                     diagnostic_code="invalid_request",
                 )
             directory = Path(stack.enter_context(tempfile.TemporaryDirectory(prefix="loopx-effect-")))
+            protect_private_path(directory, directory=True)
             response_sink = directory / "response.json"
             descriptor = os.open(response_sink, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
-            os.close(descriptor)
+            try:
+                protect_private_path(response_sink)
+            finally:
+                os.close(descriptor)
             request["response_sink"] = str(response_sink)
             encoded = (json.dumps(request, separators=(",", ":")) + "\n").encode()
             if len(encoded) > MAX_REQUEST_BYTES:
@@ -850,6 +855,7 @@ def _request_with_info(
                 params_path = directory / "params.json"
                 descriptor = os.open(params_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
                 with os.fdopen(descriptor, "wb") as file:
+                    protect_private_path(params_path)
                     file.write(params_bytes)
                 request.pop("params")
                 request["params_ref"] = {
@@ -1037,8 +1043,10 @@ def _read_local_snapshot_response(
                 or size > MAX_LOCAL_SNAPSHOT_BYTES or not isinstance(digest, str)
                 or BARE_SHA256_PATTERN.fullmatch(digest) is None):
             raise ValueError("invalid local snapshot reference")
+        require_private_path(sink.parent, directory=True)
         descriptor = os.open(sink, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
         with os.fdopen(descriptor, "rb") as file:
+            require_private_path(sink, descriptor=file.fileno())
             metadata = os.fstat(file.fileno())
             if (not os.path.isfile(sink) or metadata.st_size != size
                     or (hasattr(os, "getuid") and
@@ -1197,6 +1205,8 @@ def _start_runtime(*, fingerprint: str, info_path: Path) -> dict[str, Any]:
         environment["LOOPX_EFFECT_RUNTIME_TOKEN"] = token
         # Fixed daemon context, never a model-supplied executable argument.
         environment["LOOPX_EFFECT_RUNTIME_PYTHON"] = sys.executable
+        # Windows snapshot ACL facts use this parent-selected interpreter only.
+        environment["LOOPX_EFFECT_RUNTIME_PYTHON_BIN"] = sys.executable
         # Capture stderr so a rejected startup can publish a typed
         # startup diagnostic instead of a bare exit status. The capture
         # is an unlinked temporary file, so it cannot deadlock the child on a

loopx/control_plane/effect_runtime_snapshot.ts

@@ -1,6 +1,8 @@
 /** Private same-UID files for complete local facts that exceed the RPC wire.
  * The digest covers the bytes used by the typed owner; paths are never evidence. */
 import {createHash} from "node:crypto";
+import {execFile} from "node:child_process";
+import {fileURLToPath} from "node:url";
 import {constants} from "node:fs";
 import {lstat, open, type FileHandle} from "node:fs/promises";
 import {basename, dirname, isAbsolute} from "node:path";
@@ -19,11 +21,23 @@ async function privateFile(path: string, flags: number): Promise<FileHandle> {
       (parent.uid !== process.getuid() || (parent.mode & 0o077) !== 0))) {
     throw new Error("invalid private snapshot directory");
   }
-  if ((await lstat(path)).isSymbolicLink()) throw new Error("private snapshot is a symlink");
+  const entry = await lstat(path);
+  if (entry.isSymbolicLink()) throw new Error("private snapshot is a symlink");
+  if (process.platform === "win32") {
+    const python = process.env.LOOPX_EFFECT_RUNTIME_PYTHON_BIN;
+    if (!python || !isAbsolute(python)) throw new Error("private snapshot ACL helper unavailable");
+    // Fixed package-owned Win32 IO, not a caller-supplied executable or decision owner.
+    const helper = fileURLToPath(new URL("./runtime/private_permissions.py", import.meta.url));
+    await new Promise<void>((resolve, reject) => {
+      execFile(python, [helper, "--verify-snapshot", path],
+        {timeout: 5000, windowsHide: true, maxBuffer: 4096},
+        error => error ? reject(error) : resolve());
+    });
+  }
   const file = await open(path, flags | (constants.O_NOFOLLOW ?? 0));
   try {
     const stat = await file.stat();
-    if (!stat.isFile() || (process.getuid &&
+    if (!stat.isFile() || stat.dev !== entry.dev || stat.ino !== entry.ino || (process.getuid &&
         (stat.uid !== process.getuid() || (stat.mode & 0o077) !== 0))) {
       throw new Error("invalid private snapshot file");
     }

loopx/control_plane/runtime/document_io.py

@@ -7,23 +7,36 @@
 from __future__ import annotations
 
 import os
+import json
+from typing import Any
+
+from .file_paths import windows_extended_path
+from .private_permissions import protect_private_path
 import stat
 import tempfile
 from pathlib import Path
 
 
-def atomic_write_state_text(path: Path, text: str, *, create_only: bool = False) -> None:
+def atomic_write_state_text(
+    path: Path, text: str, *, create_only: bool = False, private: bool = False,
+) -> None:
     """Persist complete UTF-8 state while the caller holds its sibling lock."""
 
+    path = windows_extended_path(path)
     path.parent.mkdir(parents=True, exist_ok=True)
+    if private:
+        protect_private_path(path.parent, directory=True)
     mode = stat.S_IMODE(path.stat().st_mode) if not create_only and path.exists() else 0o600
     descriptor, temporary = tempfile.mkstemp(
         prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent)
     )
     temporary_path = Path(temporary)
     try:
         with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as handle:
-            os.chmod(temporary_path, mode)
+            if private:
+                protect_private_path(temporary_path)
+            else:
+                os.chmod(temporary_path, mode)
             handle.write(text)
             handle.flush()
             os.fsync(handle.fileno())
@@ -51,8 +64,15 @@ def verify_state_text_durable(path: Path, text: str) -> None:
     Equality of visible bytes alone does not prove the directory entry persisted.
     The caller holds the same document lock as the state writer.
     """
+    path = windows_extended_path(path)
     with path.open("r+" if os.name == "nt" else "r", encoding="utf-8", newline="") as handle:
         if handle.read() != text:
             raise RuntimeError("Todo Markdown projection readback mismatch")
         os.fsync(handle.fileno())
     fsync_state_directory(path)
+
+
+def atomic_write_private_json(path: Path, payload: dict[str, Any]) -> None:
+    """Publish private complete JSON after enforcing platform access controls."""
+    text = json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
+    atomic_write_state_text(path, text, private=True)

loopx/control_plane/runtime/local_state_migration.py

@@ -28,6 +28,7 @@
     _is_redirected_path,
 )
 from ...runtime import validate_goal_id_path_segment
+from .file_paths import windows_extended_path
 
 
 LOCAL_STATE_MIGRATION_SCHEMA = "loopx_local_state_migration_v1"
@@ -105,14 +106,14 @@ def _write_registry(path: Path, payload: dict[str, Any]) -> None:
     _require_unlinked_directory_chain(path.parent, label="migration write")
     if _is_redirected_path(path):
         raise ValueError(f"migration write is a symlink or junction: {path}")
-    descriptor, name = tempfile.mkstemp(prefix=".loopx-migration-", dir=path.parent.parent)
+    descriptor, name = tempfile.mkstemp(prefix=".loopx-migration-", dir=windows_extended_path(path.parent.parent))
     temporary = Path(name)
     try:
         with os.fdopen(descriptor, "wb") as stream:
             stream.write(_registry_bytes(payload))
             stream.flush()
             os.fsync(stream.fileno())
-        temporary.replace(path)
+        temporary.replace(windows_extended_path(path))
     finally:
         temporary.unlink(missing_ok=True)
 
@@ -127,19 +128,20 @@ def _write_project_registry(path: Path, payload: dict[str, Any]) -> None:
 def _digest(path: Path) -> str:
     """Hash source bytes without traversing redirected files or directories."""
 
+    address = windows_extended_path(path)
     digest = hashlib.sha256()
-    if _is_redirected_path(path):
+    if _is_redirected_path(path) or (address != path and _is_redirected_path(address)):
         raise ValueError(f"migration source contains a symlink or junction: {path}")
-    if path.is_file():
+    if address.is_file():
         digest.update(b"file\0")
-        with path.open("rb") as stream:
+        with address.open("rb") as stream:
             for chunk in iter(lambda: stream.read(1024 * 1024), b""):
                 digest.update(chunk)
-    elif path.is_dir():
+    elif address.is_dir():
         digest.update(b"dir\0")
-        for child in sorted(path.iterdir(), key=lambda item: item.name):
+        for child in sorted(address.iterdir(), key=lambda item: item.name):
             digest.update(child.name.encode("utf-8") + b"\0")
-            digest.update(_digest(child).encode("ascii"))
+            digest.update(_digest(path / child.name).encode("ascii"))
     else:
         raise FileNotFoundError(path)
     return digest.hexdigest()
@@ -219,7 +221,7 @@ def _rewrite_registry(
         source, target = route
         declared = Path(str(goal["state_file"])).expanduser()
         goal["state_file"] = (
-            str(target) if declared.is_absolute() else str(DEFAULT_PROJECT_GOALS / target.parent.name / target.name)
+            str(target) if declared.is_absolute() else (DEFAULT_PROJECT_GOALS / target.parent.name / target.name).as_posix()
         )
     return updated
 
@@ -316,7 +318,15 @@ def _copy(source: Path, target: Path) -> None:
     if _is_redirected_path(source):
         raise ValueError(f"migration source is a symlink or junction: {source}")
     _require_backup_path(target.parent)
-    target.parent.mkdir(parents=True, exist_ok=True)
+    windows_extended_path(target.parent).mkdir(parents=True, exist_ok=True)
+    _require_backup_path(target.parent)
+    if target.exists() or _is_redirected_path(target):
+        raise FileExistsError(f"backup snapshot target already exists: {target}")
+    source, target = windows_extended_path(source), windows_extended_path(target)
+    # Check the supported native address too, including long reparse ancestors.
+    _require_unlinked_directory_chain(source.parent, label="migration source")
+    if _is_redirected_path(source):
+        raise ValueError(f"migration source is a symlink or junction: {source}")
     _require_backup_path(target.parent)
     if target.exists() or _is_redirected_path(target):
         raise FileExistsError(f"backup snapshot target already exists: {target}")
@@ -338,7 +348,7 @@ def _write_verified_backup(plan: dict[str, Any]) -> None:
     backup = Path(plan["backup_dir"])
     entries = plan["entries"]
     _require_backup_path(backup, must_be_absent=True)
-    backup.mkdir(mode=0o700, parents=True, exist_ok=False)
+    windows_extended_path(backup).mkdir(mode=0o700, parents=True, exist_ok=False)
     _require_backup_path(backup)
     try:
         for index, entry in enumerate(entries):
@@ -355,7 +365,7 @@ def _write_verified_backup(plan: dict[str, Any]) -> None:
             if _digest(copied) != entry["digest"]:
                 raise ValueError(f"backup verification failed for {original}")
         _require_backup_path(backup)
-        (backup / "plan.json").write_text(json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+        windows_extended_path(backup / "plan.json").write_text(json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     except Exception:
         # Keep any partial backup for inspection. No authoritative state moved.
         raise
@@ -384,7 +394,7 @@ def _verify_migrated_state(plan: dict[str, Any], writes: _MigrationWrites) -> No
             if legacy.exists() or _is_redirected_path(legacy):
                 raise ValueError(f"legacy authority reappeared during migration: {legacy}")
     before_runtime = backup / "snapshot" / "0"
-    before_children = {child.name for child in before_runtime.iterdir()}
+    before_children = {child.name for child in windows_extended_path(before_runtime).iterdir()}
     after_children = {child.name for child in target.iterdir()}
     if before_children != after_children:
         raise ValueError("runtime files changed during migration")
@@ -541,7 +551,7 @@ def _read_recovery_receipt(receipt_path: Path) -> dict[str, Any]:
     _require_backup_path(receipt_path.parent)
     if _is_redirected_path(receipt_path):
         raise ValueError(f"backup receipt is a symlink or junction: {receipt_path}")
-    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    receipt = json.loads(windows_extended_path(receipt_path).read_text(encoding="utf-8"))
     if (
         not isinstance(receipt, dict)
         or receipt.get("schema_version") != LOCAL_STATE_MIGRATION_SCHEMA
@@ -553,7 +563,7 @@ def _read_recovery_receipt(receipt_path: Path) -> dict[str, Any]:
     plan_path = backup / "plan.json"
     if _is_redirected_path(plan_path):
         raise ValueError("migration backup plan is redirected")
-    plan = json.loads(plan_path.read_text(encoding="utf-8"))
+    plan = json.loads(windows_extended_path(plan_path).read_text(encoding="utf-8"))
     entries = receipt.get("entries")
     if not isinstance(entries, list) or not isinstance(plan, dict):
         raise ValueError("migration receipt has no entries")
@@ -578,9 +588,9 @@ def _read_recovery_receipt(receipt_path: Path) -> dict[str, Any]:
 
 def _require_unchanged_runtime(receipt: dict[str, Any], active: Path) -> None:
     before = Path(receipt["backup_dir"]) / "snapshot" / "0"
-    if {child.name for child in before.iterdir()} != {child.name for child in active.iterdir()}:
+    if {child.name for child in windows_extended_path(before).iterdir()} != {child.name for child in active.iterdir()}:
         raise ValueError(f"migrated state changed; automatic rollback is unsafe: {active}")
-    for child in before.iterdir():
+    for child in windows_extended_path(before).iterdir():
         current = active / child.name
         if child.name != GLOBAL_REGISTRY_FILENAME:
             if _digest(current) != _digest(child):
@@ -649,11 +659,11 @@ def _restore_snapshot_file(snapshot: Path, destination: Path, *, expected_digest
     if _digest(snapshot) != expected_digest:
         raise ValueError(f"migration backup changed: {snapshot}")
     before = _digest(destination)
-    descriptor, name = tempfile.mkstemp(prefix=".loopx-rollback-", dir=destination.parent.parent)
+    descriptor, name = tempfile.mkstemp(prefix=".loopx-rollback-", dir=windows_extended_path(destination.parent.parent))
     os.close(descriptor)
     temporary = Path(name)
     try:
-        shutil.copy2(snapshot, temporary)
+        shutil.copy2(windows_extended_path(snapshot), temporary)
         if _digest(temporary) != expected_digest or _digest(snapshot) != expected_digest:
             raise ValueError(f"registry recovery copy changed: {snapshot}")
         with temporary.open("r+b") as stream:
@@ -663,7 +673,7 @@ def _restore_snapshot_file(snapshot: Path, destination: Path, *, expected_digest
             raise ValueError("registry recovery destination is redirected")
         if _digest(destination) != before:
             raise ValueError(f"registry changed during recovery: {destination}")
-        temporary.replace(destination)
+        temporary.replace(windows_extended_path(destination))
     finally:
         temporary.unlink(missing_ok=True)