diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py
index b5c78ed674..1eef5eb23f 100644
--- a/loopx/control_plane/effect_runtime.py
+++ b/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
diff --git a/loopx/control_plane/effect_runtime_snapshot.ts b/loopx/control_plane/effect_runtime_snapshot.ts
index b3bd40438a..6bf287a57b 100644
--- a/loopx/control_plane/effect_runtime_snapshot.ts
+++ b/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");
     }
diff --git a/loopx/control_plane/runtime/document_io.py b/loopx/control_plane/runtime/document_io.py
index 66f94cc959..cbfa0077ad 100644
--- a/loopx/control_plane/runtime/document_io.py
+++ b/loopx/control_plane/runtime/document_io.py
@@ -7,15 +7,25 @@
 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)
@@ -23,7 +33,10 @@ def atomic_write_state_text(path: Path, text: str, *, create_only: bool = False)
     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)
diff --git a/loopx/control_plane/runtime/local_state_migration.py b/loopx/control_plane/runtime/local_state_migration.py
index 4ac75cdb52..af0498a282 100644
--- a/loopx/control_plane/runtime/local_state_migration.py
+++ b/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)
 
diff --git a/loopx/control_plane/runtime/private_permissions.py b/loopx/control_plane/runtime/private_permissions.py
new file mode 100644
index 0000000000..e7b4f57c1d
--- /dev/null
+++ b/loopx/control_plane/runtime/private_permissions.py
@@ -0,0 +1,198 @@
+"""Platform permissions for caller-owned private local IO; no authority decisions."""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import stat
+import sys
+
+
+def protect_private_path(path: Path, *, directory: bool = False) -> None:
+    """Restrict an existing private leaf to its current account and OS admins.
+
+    The caller owns the namespace/locking. Reparse points are never repaired.
+    Windows uses a protected DACL, not chmod's emulated POSIX write flag.
+    """
+    info = path.lstat()
+    if (stat.S_ISLNK(info.st_mode)
+            or getattr(info, "st_file_attributes", 0) & 0x400
+            or (not stat.S_ISDIR(info.st_mode) if directory else not stat.S_ISREG(info.st_mode))):
+        raise OSError("private IO path is not a regular owned leaf")
+    if os.name != "nt":
+        os.chmod(path, 0o700 if directory else 0o600)
+        return
+    _protect_windows_path(path, directory=directory)
+
+
+def _protect_windows_path(path: Path, *, directory: bool) -> None:
+    # Native APIs are loaded only on Windows. No subprocess, external package,
+    # ambient inherited ACE or caller-supplied principal is trusted here.
+    if sys.platform != "win32":
+        raise OSError("Windows private IO requires a Windows host")
+    import ctypes as ct
+    from ctypes import wintypes as w
+
+    adv = ct.WinDLL("advapi32", use_last_error=True)
+    kernel = ct.WinDLL("kernel32", use_last_error=True)
+    pointer = ct.c_void_p
+    kernel.GetCurrentProcess.restype = w.HANDLE
+    kernel.CloseHandle.argtypes = [w.HANDLE]
+    kernel.LocalFree.argtypes = [pointer]
+    kernel.LocalFree.restype = pointer
+    adv.OpenProcessToken.argtypes = [w.HANDLE, w.DWORD, ct.POINTER(w.HANDLE)]
+    adv.GetTokenInformation.argtypes = [w.HANDLE, ct.c_int, pointer, w.DWORD, ct.POINTER(w.DWORD)]
+    adv.ConvertSidToStringSidW.argtypes = [pointer, ct.POINTER(w.LPWSTR)]
+    adv.ConvertStringSecurityDescriptorToSecurityDescriptorW.argtypes = [w.LPCWSTR, w.DWORD, ct.POINTER(pointer), ct.POINTER(w.DWORD)]
+    adv.GetSecurityDescriptorDacl.argtypes = [pointer, ct.POINTER(w.BOOL), ct.POINTER(pointer), ct.POINTER(w.BOOL)]
+    adv.SetNamedSecurityInfoW.argtypes = [w.LPWSTR, ct.c_int, w.DWORD, pointer, pointer, pointer, pointer]
+    adv.SetNamedSecurityInfoW.restype = w.DWORD
+    token = w.HANDLE()
+    sd = pointer()
+    sid_text = w.LPWSTR()
+    try:
+        if not adv.OpenProcessToken(kernel.GetCurrentProcess(), 0x8, ct.byref(token)):
+            raise ct.WinError(ct.get_last_error())
+        size = w.DWORD()
+        adv.GetTokenInformation(token, 1, None, 0, ct.byref(size))  # TokenUser
+        if not size.value:
+            raise ct.WinError(ct.get_last_error())
+        user = ct.create_string_buffer(size.value)
+        if not adv.GetTokenInformation(token, 1, user, size, ct.byref(size)):
+            raise ct.WinError(ct.get_last_error())
+        sid = ct.cast(user, ct.POINTER(pointer))[0]
+        if not adv.ConvertSidToStringSidW(sid, ct.byref(sid_text)):
+            raise ct.WinError(ct.get_last_error())
+        flags = "OICI" if directory else ""
+        descriptor = f"D:P(A;{flags};FA;;;{sid_text.value})(A;{flags};FA;;;SY)(A;{flags};FA;;;BA)"
+        if not adv.ConvertStringSecurityDescriptorToSecurityDescriptorW(descriptor, 1, ct.byref(sd), None):
+            raise ct.WinError(ct.get_last_error())
+        present, defaulted = w.BOOL(), w.BOOL()
+        dacl = pointer()
+        if not adv.GetSecurityDescriptorDacl(sd, ct.byref(present), ct.byref(dacl), ct.byref(defaulted)) or not present.value or not dacl.value:
+            raise OSError("private IO DACL unavailable")
+        # SE_FILE_OBJECT; DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION.
+        error = adv.SetNamedSecurityInfoW(str(path), 1, 0x80000004, None, None, dacl, None)
+        if error:
+            raise ct.WinError(error)
+    finally:
+        if sid_text:
+            kernel.LocalFree(ct.cast(sid_text, pointer))
+        if sd:
+            kernel.LocalFree(sd)
+        if token:
+            kernel.CloseHandle(token)
+
+
+def require_private_path(path: Path, *, directory: bool = False, descriptor: int | None = None) -> None:
+    """Read OS access facts without repairing an untrusted caller's file."""
+    info = path.lstat()
+    if (stat.S_ISLNK(info.st_mode)
+            or getattr(info, "st_file_attributes", 0) & 0x400
+            or (not stat.S_ISDIR(info.st_mode) if directory else not stat.S_ISREG(info.st_mode))):
+        raise OSError("private IO path is not a regular owned leaf")
+    if descriptor is not None:
+        opened = os.fstat(descriptor)
+        if (opened.st_dev, opened.st_ino) != (info.st_dev, info.st_ino):
+            raise OSError("private IO path changed after open")
+        info = opened
+    if sys.platform != "win32":
+        if info.st_uid != os.getuid() or info.st_mode & 0o077:
+            raise OSError("private IO permissions are invalid")
+        return
+    _require_windows_private_path(path, descriptor=descriptor)
+
+
+def _require_windows_private_path(path: Path, *, descriptor: int | None) -> None:
+    if sys.platform != "win32":
+        raise OSError("Windows private IO requires a Windows host")
+    import ctypes as ct
+    from ctypes import wintypes as w
+    import msvcrt
+
+    adv = ct.WinDLL("advapi32", use_last_error=True)
+    kernel = ct.WinDLL("kernel32", use_last_error=True)
+    pointer = ct.c_void_p
+    out = ct.POINTER(pointer)
+    kernel.GetCurrentProcess.restype = w.HANDLE
+    kernel.CloseHandle.argtypes = [w.HANDLE]
+    kernel.LocalFree.argtypes = [pointer]
+    kernel.LocalFree.restype = pointer
+    adv.OpenProcessToken.argtypes = [w.HANDLE, w.DWORD, ct.POINTER(w.HANDLE)]
+    adv.GetTokenInformation.argtypes = [w.HANDLE, ct.c_int, pointer, w.DWORD, ct.POINTER(w.DWORD)]
+    adv.ConvertSidToStringSidW.argtypes = [pointer, ct.POINTER(w.LPWSTR)]
+    adv.GetNamedSecurityInfoW.argtypes = [w.LPCWSTR, ct.c_int, w.DWORD, out, out, out, out, out]
+    adv.GetNamedSecurityInfoW.restype = w.DWORD
+    adv.GetSecurityInfo.argtypes = [w.HANDLE, ct.c_int, w.DWORD, out, out, out, out, out]
+    adv.GetSecurityInfo.restype = w.DWORD
+    adv.GetSecurityDescriptorControl.argtypes = [pointer, ct.POINTER(w.WORD), ct.POINTER(w.DWORD)]
+    adv.GetAclInformation.argtypes = [pointer, pointer, w.DWORD, ct.c_int]
+    adv.GetAce.argtypes = [pointer, w.DWORD, out]
+    token, sd, owner, dacl = w.HANDLE(), pointer(), pointer(), pointer()
+
+    def sid_text(sid: ct.c_void_p | int | None) -> str:
+        value = w.LPWSTR()
+        if not sid or not adv.ConvertSidToStringSidW(sid, ct.byref(value)):
+            raise OSError("private IO SID unavailable")
+        try:
+            return str(value.value)
+        finally:
+            kernel.LocalFree(ct.cast(value, pointer))
+
+    try:
+        if not adv.OpenProcessToken(kernel.GetCurrentProcess(), 0x8, ct.byref(token)):
+            raise ct.WinError(ct.get_last_error())
+        size = w.DWORD()
+        adv.GetTokenInformation(token, 1, None, 0, ct.byref(size))
+        if not size.value:
+            raise OSError("private IO account unavailable")
+        user = ct.create_string_buffer(size.value)
+        if not adv.GetTokenInformation(token, 1, user, size, ct.byref(size)):
+            raise ct.WinError(ct.get_last_error())
+        allowed = {sid_text(ct.cast(user, ct.POINTER(pointer))[0]), "S-1-5-18", "S-1-5-32-544"}
+        args = (1, 0x5, ct.byref(owner), None, ct.byref(dacl), None, ct.byref(sd))
+        error = (adv.GetNamedSecurityInfoW(str(path), *args) if descriptor is None else
+                 adv.GetSecurityInfo(w.HANDLE(msvcrt.get_osfhandle(descriptor)), *args))
+        if error:
+            raise ct.WinError(error)
+        control, revision = w.WORD(), w.DWORD()
+        if (not sd or not dacl or sid_text(owner) not in allowed
+                or not adv.GetSecurityDescriptorControl(sd, ct.byref(control), ct.byref(revision))
+                or not control.value & 0x1000):  # SE_DACL_PROTECTED
+            raise OSError("private IO owner or protected DACL is invalid")
+        info = (w.DWORD * 3)()
+        if not adv.GetAclInformation(dacl, info, ct.sizeof(info), 2) or info[0] != 3:
+            raise OSError("private IO ACE set is invalid")
+        seen = set()
+        for index in range(info[0]):
+            ace = pointer()
+            if not adv.GetAce(dacl, index, ct.byref(ace)) or not ace.value:
+                raise OSError("private IO ACE unavailable")
+            header = (ct.c_ubyte * 4).from_address(ace.value)
+            # Only explicit ACCESS_ALLOWED_ACE, full control, current account/OS admins.
+            if header[0] != 0 or header[1] & 0x10 or ct.c_uint32.from_address(ace.value + 4).value != 0x1F01FF:
+                raise OSError("private IO ACE grants unexpected access")
+            principal = sid_text(pointer(ace.value + 8))
+            if principal not in allowed:
+                raise OSError("private IO principal is invalid")
+            seen.add(principal)
+        if seen != allowed:
+            raise OSError("private IO account access is incomplete")
+    finally:
+        if sd:
+            kernel.LocalFree(sd)
+        if token:
+            kernel.CloseHandle(token)
+
+
+if __name__ == "__main__":
+    # Fixed, package-owned OS IO helper for Node on Windows; no Goal decisions.
+    import sys
+    if len(sys.argv) != 3 or sys.argv[1] != "--verify-snapshot":
+        raise SystemExit(2)
+    try:
+        leaf = Path(sys.argv[2])
+        require_private_path(leaf.parent, directory=True)
+        with leaf.open("rb") as handle:
+            require_private_path(leaf, descriptor=handle.fileno())
+    except OSError:
+        raise SystemExit(1) from None
diff --git a/loopx/control_plane/todos/completion_validation_store.py b/loopx/control_plane/todos/completion_validation_store.py
index a90b384672..1454ac5bf9 100644
--- a/loopx/control_plane/todos/completion_validation_store.py
+++ b/loopx/control_plane/todos/completion_validation_store.py
@@ -2,15 +2,16 @@
 
 from __future__ import annotations
 
-import os
 import re
 from collections.abc import Iterable, Mapping
 from pathlib import Path
 from typing import Any
 
-from ...registry import atomic_write_json, read_json
+from ...registry import read_json
 from ...file_lock import exclusive_cross_runtime_file_lock
-from ..runtime.document_io import fsync_state_directory
+from ..runtime.document_io import atomic_write_private_json as atomic_write_json, fsync_state_directory
+from ..runtime.file_paths import windows_extended_path
+from ..runtime.private_permissions import protect_private_path
 from .completion_validation_projection import (
     completion_validation_declaration,
     completion_validation_declaration_sha256,
@@ -31,7 +32,7 @@ def _require_public_id(value: str, label: str) -> str:
 def completion_validation_declaration_path(
     *, runtime_root: Path, goal_id: str, todo_id: str
 ) -> Path:
-    return (
+    return windows_extended_path(
         runtime_root.expanduser().resolve(strict=False)
         / "goals"
         / _require_public_id(goal_id, "goal_id")
@@ -67,9 +68,7 @@ def persist_completion_validation_declaration(
             "declaration_sha256": digest,
             "declaration": normalized,
         },
-        preserve_mode=True,
     )
-    os.chmod(path, 0o600)
     return digest
 
 
@@ -96,6 +95,8 @@ def prepare_completion_validation_declaration(
                 raise ValueError("prepared validation declaration digest mismatch")
         else:
             atomic_write_json(path, payload)
+        protect_private_path(path.parent, directory=True)
+        protect_private_path(path)
         # Also re-establish directory durability on an idempotent retry.
         fsync_state_directory(path)
     return digest
diff --git a/loopx/control_plane/turn_driver/codex_sessions.py b/loopx/control_plane/turn_driver/codex_sessions.py
index 741a2df4a2..c845e96966 100644
--- a/loopx/control_plane/turn_driver/codex_sessions.py
+++ b/loopx/control_plane/turn_driver/codex_sessions.py
@@ -13,6 +13,7 @@
 
 from ...boundary_authority import normalize_checkpointed_boundary_authority_entries
 from ...file_lock import exclusive_file_lock
+from ..runtime.private_permissions import protect_private_path
 from ...runtime import validate_goal_id_path_segment
 from ..content_digest import BARE_SHA256_PATTERN
 from ..projects.registry_codec import (
@@ -409,13 +410,13 @@ def _write_codex_cli_session(
         raise ValueError("Codex CLI returned an invalid session id")
     path = _session_path(runtime_root, lineage, session_scope)
     path.parent.mkdir(parents=True, exist_ok=True)
+    protect_private_path(path.parent, directory=True)
     descriptor, temporary_name = tempfile.mkstemp(
         prefix=f".{path.name}.", dir=path.parent
     )
     temporary = Path(temporary_name)
     try:
-        if hasattr(os, "fchmod"):
-            os.fchmod(descriptor, 0o600)
+        protect_private_path(temporary)
         handle = os.fdopen(descriptor, "w", encoding="utf-8")
         descriptor = -1
         with handle:
@@ -459,7 +460,6 @@ def _write_codex_cli_session(
                 handle.flush()
                 os.fsync(handle.fileno())
         os.replace(temporary, path)
-        path.chmod(0o600)
     finally:
         if descriptor >= 0:
             os.close(descriptor)
diff --git a/loopx/file_lock.py b/loopx/file_lock.py
index 09959b2b80..8b021c0bfb 100644
--- a/loopx/file_lock.py
+++ b/loopx/file_lock.py
@@ -40,6 +40,7 @@ def write_bytes_atomic(
     preserve_mode: bool = False,
     directory_mode: int | None = None,
     fsync_directory: bool = False,
+    replace_retry_seconds: float = 0,
 ) -> None:
     """Replace one file with `payload` through a same-directory temp file.
 
@@ -52,7 +53,7 @@ def write_bytes_atomic(
     untouched; an explicit `mode` is applied with chmod. `preserve_mode` keeps
     an existing destination's mode and wins over both.
     """
-    destination = Path(path).expanduser()
+    destination = windows_extended_path(Path(path).expanduser())
     destination.parent.mkdir(parents=True, exist_ok=True, mode=directory_mode or 0o777)
     if preserve_mode and destination.exists():
         mode = destination.stat().st_mode & 0o777
@@ -69,7 +70,17 @@ def write_bytes_atomic(
             os.fsync(handle.fileno())
         if mode is not None:
             os.chmod(temporary, mode)
-        os.replace(temporary, destination)
+        deadline = time.monotonic() + replace_retry_seconds
+        while True:
+            try:
+                os.replace(temporary, destination)
+                break
+            except OSError as error:
+                remaining = deadline - time.monotonic()
+                if (os.name != "nt" or getattr(error, "winerror", None) != 5
+                        or remaining <= 0):
+                    raise
+                time.sleep(min(0.05, remaining))
         if fsync_directory and os.name == "posix":
             directory_fd = os.open(
                 destination.parent,
@@ -91,6 +102,7 @@ def write_json_atomic(
     preserve_mode: bool = False,
     directory_mode: int | None = None,
     fsync_directory: bool = False,
+    replace_retry_seconds: float = 0,
     indent: int | None = 2,
     sort_keys: bool = False,
     separators: tuple[str, str] | None = None,
@@ -116,6 +128,7 @@ def write_json_atomic(
         preserve_mode=preserve_mode,
         directory_mode=directory_mode,
         fsync_directory=fsync_directory,
+        replace_retry_seconds=replace_retry_seconds,
     )
 
 
@@ -345,7 +358,20 @@ def _write_holder_sidecar(holder_path: Path, record: dict[str, object]) -> None:
             holder_file.write("\n")
             holder_file.flush()
             os.fsync(holder_file.fileno())
-        os.replace(temporary_path, holder_path)
+        # A Windows reader may briefly deny DELETE sharing. Keep the old
+        # complete diagnostic until atomic replacement succeeds; persistent
+        # access denial still propagates within this bounded IO window.
+        deadline = time.monotonic() + 0.5
+        while True:
+            try:
+                os.replace(temporary_path, holder_path)
+                break
+            except OSError as error:
+                remaining = deadline - time.monotonic()
+                if (os.name != "nt" or getattr(error, "winerror", None) != 5
+                        or remaining <= 0):
+                    raise
+                time.sleep(min(0.05, remaining))
     finally:
         temporary_path.unlink(missing_ok=True)
 
diff --git a/loopx/project_prompt.py b/loopx/project_prompt.py
index 026f00b135..a96edde69b 100644
--- a/loopx/project_prompt.py
+++ b/loopx/project_prompt.py
@@ -446,10 +446,10 @@ def build_new_project_prompt(
         if isinstance(existing_registry, dict) and existing_registry.get("common_runtime_root")
         else None
     )
-    active_state_path = rel_or_abs(
+    active_state_path = Path(rel_or_abs(
         registered_goal_state_file(project.expanduser(), resolved_goal_id, existing_registry),
         project.expanduser(),
-    )
+    )).as_posix()
     allowed_domains = allowed_domains or []
     write_scope = write_scope or []
     connect_command = render_connect_command(
diff --git a/tests/control_plane/test_document_io.py b/tests/control_plane/test_document_io.py
index 8a4f8e590d..f809534d66 100644
--- a/tests/control_plane/test_document_io.py
+++ b/tests/control_plane/test_document_io.py
@@ -63,6 +63,33 @@ def test_durable_readback_never_repairs_mismatched_bytes(tmp_path):
     assert path.read_bytes() == b"original\r\n"
 
 
+@pytest.mark.skipif(os.name != "nt", reason="Windows ordinary MAX_PATH boundary")
+@pytest.mark.parametrize("private", [False, True])
+def test_ordinary_windows_long_path_publish_preserves_bytes_and_create_only(tmp_path, private):
+    # Include the real .json extension and eight random mkstemp characters.
+    name = "a" * 64 + ".json"
+    temporary_name = "." + name + ".12345678.tmp"
+    padding = 293 - len(str(tmp_path / temporary_name)) - 1
+    assert 1 <= padding <= 200
+    path = tmp_path / ("g" * padding) / name
+    assert len(str(path.parent / temporary_name)) == 293
+    assert not str(path).startswith(chr(92) * 2 + "?" + chr(92))
+    # Inspect the same physical file through Win32's supported long address;
+    # the writer and durability reader still receive the ordinary caller path.
+    address = Path(chr(92) * 2 + "?" + chr(92) + str(path))
+    original = "私有\r\nfirst\n"
+    io.atomic_write_state_text(path, original, create_only=True, private=private)
+    assert address.read_bytes() == original.encode("utf-8")
+    with pytest.raises(FileExistsError):
+        io.atomic_write_state_text(path, "loser", create_only=True, private=private)
+    assert address.read_bytes() == original.encode("utf-8")
+    updated = original + "tail\r\n"
+    io.atomic_write_state_text(path, updated, private=private)
+    io.verify_state_text_durable(path, updated)
+    assert address.read_bytes() == updated.encode("utf-8")
+    assert list(address.parent.iterdir()) == [address]
+
+
 @pytest.mark.skipif(os.name != "posix", reason="POSIX directory fsync boundary")
 def test_directory_sync_closes_descriptor_on_failure(tmp_path, monkeypatch):
     closed = []
diff --git a/tests/control_plane/test_todo_completion_validation.py b/tests/control_plane/test_todo_completion_validation.py
index eb9b685b45..c2d97779a4 100644
--- a/tests/control_plane/test_todo_completion_validation.py
+++ b/tests/control_plane/test_todo_completion_validation.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import os
 import re
 import shlex
 import subprocess
@@ -901,15 +902,16 @@ def test_private_validation_store_is_owner_only_and_detects_tampering(
         todo_id="todo_private_validation",
     )
 
-    assert path.stat().st_mode & 0o777 == 0o600
-    assert (
-        read_completion_validation_declaration(
-            runtime_root=tmp_path,
-            goal_id=GOAL_ID,
-            todo_id="todo_private_validation",
-        )
-        == declaration
-    )
+    if os.name == "posix":
+        assert path.stat().st_mode & 0o777 == 0o600
+    else:
+        from tests.control_plane.windows_private_acl import assert_private_acl
+        assert_private_acl(path)
+    assert read_completion_validation_declaration(
+        runtime_root=tmp_path,
+        goal_id=GOAL_ID,
+        todo_id="todo_private_validation",
+    ) == declaration
     assert digest == completion_validation_declaration_sha256(declaration)
 
     stored = json.loads(path.read_text(encoding="utf-8"))
diff --git a/tests/control_plane/test_windows_snapshot_privacy.py b/tests/control_plane/test_windows_snapshot_privacy.py
new file mode 100644
index 0000000000..0a811bb903
--- /dev/null
+++ b/tests/control_plane/test_windows_snapshot_privacy.py
@@ -0,0 +1,171 @@
+"""Native Windows private IO behavior; all state is disposable."""
+from pathlib import Path
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import pytest
+from loopx.control_plane import effect_runtime as er
+from loopx.control_plane.runtime import private_permissions as permissions
+from loopx.control_plane.turn_driver import codex_sessions as sessions
+from tests.control_plane.windows_private_acl import assert_private_acl
+
+pytestmark=pytest.mark.skipif(os.name!='nt',reason='Windows native DACL behavior')
+R=Path.cwd()
+NODE='''import {readPrivateJsonSnapshot,openPrivateResponseSink} from "MODULE";
+const a=JSON.parse(process.argv[1]);let effects=0;
+try{if(a.op==='read') await readPrivateJsonSnapshot(a.ref);
+else {const h=await openPrivateResponseSink(a.ref.path);await h.close();}
+effects++;process.stdout.write(JSON.stringify({accepted:true,effects}));}
+catch(e){process.stdout.write(JSON.stringify({accepted:false,effects,error:e.message}));}
+'''.replace('MODULE',(R/'loopx/control_plane/effect_runtime_snapshot.ts').as_uri())
+
+def broaden(path):
+    rights='*S-1-1-0:(OI)(CI)F' if path.is_dir() else '*S-1-1-0:F'
+    r=subprocess.run(['icacls.exe',str(path),'/grant',rights],capture_output=True,text=True,timeout=30)
+    assert r.returncode==0,r.stdout+r.stderr
+
+def typed(path,*,op='read',bad_digest=False):
+    raw = path.read_bytes()
+    ref = {
+        "path": str(path),
+        "byte_count": len(raw),
+        "sha256": "0" * 64 if bad_digest else hashlib.sha256(raw).hexdigest(),
+    }
+    r=subprocess.run([er._node_executable(),'--no-warnings','--experimental-strip-types','--input-type=module','-e',NODE,json.dumps({'op':op,'ref':ref})],env=dict(os.environ,LOOPX_EFFECT_RUNTIME_PYTHON_BIN=sys.executable),capture_output=True,text=True,timeout=30)
+    assert r.returncode==0,r.stdout+r.stderr
+    return json.loads(r.stdout)
+
+@pytest.fixture
+def snapshot(tmp_path):
+    parent=Path(tempfile.mkdtemp(prefix='loopx-effect-',dir=tmp_path))
+    permissions.protect_private_path(parent,directory=True)
+    path=parent/'response.json'
+    payload={'schema_version':er.EFFECT_RUNTIME_RESPONSE_SCHEMA_VERSION,'request_id':'owned-request','ok':True,'result':{'blob':'x'*2200000}}
+    path.write_text(json.dumps(payload,separators=(',',':')),encoding='utf8')
+    permissions.protect_private_path(path)
+    return path,payload
+
+def envelope(path):
+    raw=path.read_bytes()
+    return {'schema_version':er.EFFECT_RUNTIME_RESPONSE_SCHEMA_VERSION,'request_id':'owned-request','ok':True,'result_ref':{'byte_count':len(raw),'sha256':hashlib.sha256(raw).hexdigest()}}
+
+def pyread(path,env=None):
+    return er._read_local_snapshot_response(env or envelope(path),path,method='goal.checkpoint_read_context.commit',request_id='owned-request',timeout=10)
+
+def test_actual_producer_protects_before_socket_dispatch(monkeypatch):
+    seen=[]
+    class Connection:
+        def __enter__(self):return self
+        def __exit__(self,*args):return None
+        def settimeout(self,value):pass
+        def sendall(self,raw):
+            request = json.loads(raw)
+            sink = Path(request["response_sink"])
+            params = Path(request["params_ref"]["path"])
+            for p in (sink.parent, sink, params):
+                assert_private_acl(p)
+            assert len(params.read_bytes())>er.MAX_REQUEST_BYTES
+            seen.append(request)
+        def recv(self,size):return json.dumps({'schema_version':er.EFFECT_RUNTIME_RESPONSE_SCHEMA_VERSION,'request_id':'producer','ok':True,'result':{}}).encode()+b'\n'
+    monkeypatch.setattr(er.socket,'create_connection',lambda *a,**k:Connection())
+    er._request_with_info({'token':'synthetic','host':'127.0.0.1','port':1},request_id='producer',method='goal.checkpoint_read_context.resolve',params={'large':'x'*2200000},timeout=2,large_local_snapshot=True)
+    assert len(seen)==1 and not Path(seen[0]['response_sink']).exists()
+
+def test_large_protected_bytes_and_digest(snapshot):
+    path, payload = snapshot
+    assert_private_acl(path.parent)
+    assert_private_acl(path)
+    assert typed(path)=={'accepted':True,'effects':1}
+    assert pyread(path)==payload
+    assert typed(path,bad_digest=True)['accepted'] is False
+    bad = envelope(path)
+    bad["result_ref"]["sha256"] = "0" * 64
+    with pytest.raises(er.EffectRuntimeResponseAmbiguous):
+        pyread(path, bad)
+
+
+@pytest.mark.parametrize('op',['read','sink'])
+@pytest.mark.parametrize('target',['file','parent'])
+def test_typed_rejects_broad_acl_before_effect(snapshot,op,target):
+    path, _ = snapshot
+    broaden(path if target == "file" else path.parent)
+    result=typed(path,op=op)
+    assert result['accepted'] is False and result['effects']==0
+
+@pytest.mark.parametrize('target',['file','parent'])
+def test_python_postdispatch_acl_failure_is_ambiguous(snapshot,target):
+    path, _ = snapshot
+    broaden(path if target == "file" else path.parent)
+    with pytest.raises(er.EffectRuntimeResponseAmbiguous) as error:
+        pyread(path)
+    assert error.value.diagnostic_code=='runtime_response_ambiguous'
+
+def test_session_writer_protects_and_keeps_original_identity(tmp_path):
+    lineage={'goal_id':'isolated-privacy','agent_id':'fixture','todo_id':'original'}
+    path = sessions._session_path(tmp_path, lineage, "agent")
+    path.parent.mkdir(parents=True)
+    permissions.protect_private_path(path.parent, directory=True)
+    broaden(path.parent)
+    sessions._store_codex_cli_session(tmp_path,lineage=lineage,session_scope='agent',session_id='original-session-id')
+    assert_private_acl(path.parent)
+    assert_private_acl(path)
+    value=sessions.select_codex_cli_session(tmp_path,lineage=lineage,session_scope='agent')
+    assert value['session_id']=='original-session-id' and value['agent_id']=='fixture'
+    assert value['todo_id']=='original' and value['session_scope']=='agent'
+
+@pytest.mark.parametrize('failure',['protect','replace'])
+def test_session_publication_failure_keeps_prior_session(tmp_path,monkeypatch,failure):
+    lineage={'goal_id':'isolated-privacy','agent_id':'fixture','todo_id':'original'}
+    sessions._store_codex_cli_session(tmp_path,lineage=lineage,session_scope='agent',session_id='original-session-id')
+    path = sessions._session_path(tmp_path, lineage, "agent")
+    old = path.read_bytes()
+
+    def deny(*a,**k):raise PermissionError('injected native IO refusal')
+    if failure=='protect':
+        if not hasattr(sessions, "protect_private_path"):
+            pytest.fail("Original session writer has no protection admission")
+        monkeypatch.setattr(sessions,'protect_private_path',deny)
+    else:
+        monkeypatch.setattr(sessions.os, "replace", deny)
+    with pytest.raises(PermissionError):
+        sessions._store_codex_cli_session(
+            tmp_path, lineage=lineage, session_scope="agent", session_id="replacement"
+        )
+    assert path.read_bytes()==old
+    assert not [x for x in path.parent.iterdir() if x.name.startswith('.'+path.name+'.')]
+
+def test_python_rejects_junction_parent(snapshot,tmp_path):
+    path, _ = snapshot
+    link = tmp_path / "loopx-effect-junction"
+    cmd=['pwsh','-NoLogo','-NoProfile','-Command',"New-Item -ItemType Junction -Path $env:LX_LINK -Target $env:LX_TARGET | Out-Null"]
+    r=subprocess.run(cmd,env=dict(os.environ,LX_LINK=str(link),LX_TARGET=str(path.parent)),capture_output=True,text=True,timeout=30)
+    assert r.returncode==0,r.stderr
+    try:
+        other=link/path.name
+        assert typed(other)['accepted'] is False
+        with pytest.raises(er.EffectRuntimeResponseAmbiguous):
+            pyread(other)
+    finally:
+        os.rmdir(link)
+
+
+@pytest.mark.parametrize('helper',['missing','invalid'])
+def test_typed_missing_os_fact_helper_rejects(snapshot,helper):
+    path, _ = snapshot
+    raw = path.read_bytes()
+    ref = {
+        "path": str(path),
+        "byte_count": len(raw),
+        "sha256": hashlib.sha256(raw).hexdigest(),
+    }
+    env = dict(os.environ)
+    env.pop("LOOPX_EFFECT_RUNTIME_PYTHON_BIN", None)
+    if helper == "invalid":
+        env["LOOPX_EFFECT_RUNTIME_PYTHON_BIN"] = str(path.parent / "absent-python.exe")
+    r=subprocess.run([er._node_executable(),'--no-warnings','--experimental-strip-types','--input-type=module','-e',NODE,json.dumps({'op':'read','ref':ref})],env=env,capture_output=True,text=True,timeout=30)
+    assert r.returncode==0,r.stderr
+    result = json.loads(r.stdout)
+    assert result["accepted"] is False and result["effects"] == 0
diff --git a/tests/control_plane/windows_private_acl.py b/tests/control_plane/windows_private_acl.py
new file mode 100644
index 0000000000..43596f2a36
--- /dev/null
+++ b/tests/control_plane/windows_private_acl.py
@@ -0,0 +1,26 @@
+"""Independent OS readback for the Windows private IO tests."""
+from pathlib import Path
+import subprocess
+import os
+
+
+def assert_private_acl(path: Path) -> None:
+    # PowerShell uses .NET's independent ACL reader; product IO uses Win32.
+    script = '''$ErrorActionPreference='Stop'
+    $p=$env:LOOPX_PRIVATE_IO_TEST_PATH
+    $acl=Get-Acl -LiteralPath $p
+    if(-not $acl.AreAccessRulesProtected){throw 'DACL is inherited'}
+    $owner=[System.Security.Principal.WindowsIdentity]::GetCurrent().User.Value
+    $allowed=@($owner,'S-1-5-18','S-1-5-32-544')
+    foreach($ace in $acl.Access){
+      $sid=$ace.IdentityReference.Translate([System.Security.Principal.SecurityIdentifier]).Value
+      if($sid -notin $allowed -or $ace.IsInherited -or $ace.AccessControlType -ne 'Allow'){throw 'Unexpected private ACE'}
+    }
+    if($acl.Access.Count -ne 3){throw 'Private ACE set incomplete'}
+    '''
+    result = subprocess.run(
+        ["pwsh", "-NoLogo", "-NoProfile", "-Command", script],
+        env=dict(os.environ, LOOPX_PRIVATE_IO_TEST_PATH=str(path)),
+        capture_output=True, text=True, timeout=30,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
diff --git a/tests/control_plane_ts/effect_runtime_snapshot.test.ts b/tests/control_plane_ts/effect_runtime_snapshot.test.ts
index 75aeb75fcf..50583f3b95 100644
--- a/tests/control_plane_ts/effect_runtime_snapshot.test.ts
+++ b/tests/control_plane_ts/effect_runtime_snapshot.test.ts
@@ -1,20 +1,49 @@
 import assert from "node:assert/strict";
+import {execFile} from "node:child_process";
 import {createHash} from "node:crypto";
-import {mkdtemp, writeFile, chmod, symlink, rm} from "node:fs/promises";
-import {tmpdir} from "node:os";
+import {mkdir, mkdtemp, writeFile, chmod, symlink, rm} from "node:fs/promises";
 import {join} from "node:path";
 import {test} from "node:test";
+import {promisify} from "node:util";
 import {
   openPrivateResponseSink, readPrivateJsonSnapshot, writePrivateResponse,
   MAX_LOCAL_SNAPSHOT_BYTES,
 } from "../../loopx/control_plane/effect_runtime_snapshot.ts";
+import {resolveTestPython} from "../../scripts/test-python.mjs";
+
+const python = process.platform === "win32" ? resolveTestPython() : null;
+if (python) process.env.LOOPX_EFFECT_RUNTIME_PYTHON_BIN = python;
+const execute = promisify(execFile);
+
+async function createFixtureDirectory() {
+  const local = join(process.cwd(), ".local");
+  await mkdir(local, {recursive: true});
+  return mkdtemp(join(local, "loopx-effect-"));
+}
+
+async function protectWindowsFixture(directory: string, path: string) {
+  if (!python) return;
+  const source = [
+    "import sys",
+    "from pathlib import Path",
+    "from loopx.control_plane.runtime.private_permissions import protect_private_path",
+    "protect_private_path(Path(sys.argv[1]), directory=True)",
+    "protect_private_path(Path(sys.argv[2]))",
+  ].join("\n");
+  await execute(python, ["-c", source, directory, path], {timeout: 10_000});
+}
+
+async function removeFixture(directory: string) {
+  await rm(directory, {recursive: true, force: true, maxRetries: 10, retryDelay: 100});
+}
 
 test("local snapshot accepts exact private bytes and rejects altered or unsafe references", async () => {
-  const directory = await mkdtemp(join(tmpdir(), "loopx-effect-"));
+  const directory = await createFixtureDirectory();
   try {
     const path = join(directory, "params.json");
     const bytes = Buffer.from(JSON.stringify({context: "x".repeat(2_200_000)}));
     await writeFile(path, bytes, {mode: 0o600});
+    await protectWindowsFixture(directory, path);
     const ref = {path, byte_count: bytes.length,
       sha256: createHash("sha256").update(bytes).digest("hex")};
     assert.deepEqual(await readPrivateJsonSnapshot(ref), JSON.parse(bytes.toString()));
@@ -38,15 +67,16 @@ test("local snapshot accepts exact private bytes and rejects altered or unsafe r
         /snapshot unavailable or invalid/);
     }
   } finally {
-    await rm(directory, {recursive: true, force: true});
+    await removeFixture(directory);
   }
 });
 
 test("large response sink is private and digest-bound", async () => {
-  const directory = await mkdtemp(join(tmpdir(), "loopx-effect-"));
+  const directory = await createFixtureDirectory();
   try {
     const path = join(directory, "response.json");
     await writeFile(path, "", {mode: 0o600});
+    await protectWindowsFixture(directory, path);
     const file = await openPrivateResponseSink(path);
     const bytes = Buffer.from(JSON.stringify({result: "a".repeat(2_200_000)}));
     try {
@@ -58,6 +88,6 @@ test("large response sink is private and digest-bound", async () => {
       await file.close();
     }
   } finally {
-    await rm(directory, {recursive: true, force: true});
+    await removeFixture(directory);
   }
 });
diff --git a/tests/test_file_lock_windows_atomic.py b/tests/test_file_lock_windows_atomic.py
new file mode 100644
index 0000000000..39e5105dba
--- /dev/null
+++ b/tests/test_file_lock_windows_atomic.py
@@ -0,0 +1,146 @@
+"""Windows atomic holder publication with real file-sharing boundaries."""
+from contextlib import contextmanager
+import ctypes as ct
+from ctypes import wintypes as w
+import errno
+import json
+import os
+from pathlib import Path
+import threading
+import time
+from types import SimpleNamespace
+
+import pytest
+import loopx.file_lock as locks
+
+pytestmark = pytest.mark.skipif(os.name != 'nt', reason='Windows file sharing boundary')
+
+@contextmanager
+def reader_without_delete_share(path):
+    kernel = ct.WinDLL('kernel32', use_last_error=True)
+    kernel.CreateFileW.argtypes = [w.LPCWSTR,w.DWORD,w.DWORD,ct.c_void_p,w.DWORD,w.DWORD,w.HANDLE]
+    kernel.CreateFileW.restype = w.HANDLE
+    kernel.CloseHandle.argtypes = [w.HANDLE]
+    handle = kernel.CreateFileW(str(path),0x80000000,0x3,None,3,0,None)
+    assert handle != ct.c_void_p(-1).value
+    closed = False
+    def close():
+        nonlocal closed
+        if not closed:
+            closed = True
+            assert kernel.CloseHandle(handle)
+    try:
+        yield close
+    finally:
+        close()
+
+@contextmanager
+def transient_reader(path, monkeypatch):
+    attempts = []
+    first_attempt = threading.Event()
+    replace = os.replace
+    def observed_replace(source,destination):
+        if Path(destination) == path:
+            attempts.append(time.monotonic())
+            first_attempt.set()
+        return replace(source,destination)
+    with reader_without_delete_share(path) as close:
+        def release():
+            first_attempt.wait(2)
+            time.sleep(0.1)
+            close()
+        worker = threading.Thread(target=release)
+        worker.start()
+        monkeypatch.setattr(locks.os,'replace',observed_replace)
+        try:
+            yield attempts
+        finally:
+            first_attempt.set()
+            worker.join(3)
+            assert not worker.is_alive()
+
+def test_transient_reader_allows_atomic_holder_replace(tmp_path,monkeypatch):
+    path = tmp_path / "holder.json"
+    path.write_text(json.dumps({"sequence": 0}))
+    with transient_reader(path,monkeypatch) as attempts:
+        locks._write_holder_sidecar(path,{'sequence':1})
+    assert len(attempts)>=2
+    assert json.loads(path.read_text())=={'sequence':1}
+    assert not list(tmp_path.glob('*.tmp'))
+
+def test_transient_reader_preserves_real_kernel_lock_body(tmp_path,monkeypatch):
+    target=tmp_path/'state.json'
+    with locks.exclusive_file_lock(target):
+        pass
+    holder = locks.lock_holder_path(target)
+    effects = []
+    with transient_reader(holder,monkeypatch):
+        with locks.exclusive_file_lock(target,agent_id='writer',operation='atomic-publish'):
+            effects.append('business')
+    assert effects==['business']
+    assert json.loads(holder.read_text())['released_at']
+    with locks.try_exclusive_file_lock(target) as acquired:
+        assert acquired is not None
+
+
+def test_persistent_reader_keeps_old_bytes_and_propagates_denial(tmp_path):
+    path = tmp_path / "holder.json"
+    path.write_text(json.dumps({"sequence": 0}))
+    before = path.read_bytes()
+    with reader_without_delete_share(path):
+        start=time.monotonic()
+        with pytest.raises(OSError) as failure:
+            locks._write_holder_sidecar(path, {"sequence": 1})
+        assert failure.value.winerror==5 and time.monotonic()-start<1.2
+        assert path.read_bytes()==before
+        assert not list(tmp_path.glob('*.tmp'))
+
+def test_failed_holder_publication_releases_kernel_without_business(tmp_path):
+    target=tmp_path/'state.json'
+    with locks.exclusive_file_lock(target):
+        pass
+    holder = locks.lock_holder_path(target)
+    effects = []
+    with reader_without_delete_share(holder):
+        start=time.monotonic()
+        with pytest.raises(OSError) as failure:
+            with locks.exclusive_file_lock(target):
+                effects.append("business")
+        assert failure.value.winerror==5 and time.monotonic()-start<2
+        assert effects==[]
+    with locks.try_exclusive_file_lock(target) as acquired:
+        assert acquired is not None
+    assert not list(tmp_path.glob('*.tmp'))
+
+def test_unrelated_error_has_no_retry_or_fallback(tmp_path,monkeypatch):
+    attempts=[]
+    def fail(source,destination):
+        attempts.append(1)
+        raise FileNotFoundError(errno.ENOENT,'unrelated source error')
+    monkeypatch.setattr(locks.os,'replace',fail)
+    monkeypatch.setattr(locks.time,'sleep',lambda _:pytest.fail('unrelated error slept'))
+    with pytest.raises(FileNotFoundError):
+        locks._write_holder_sidecar(tmp_path / "holder.json", {"sequence": 1})
+    assert attempts==[1] and not list(tmp_path.glob('*.tmp'))
+
+def test_successful_holder_publication_does_not_sleep(tmp_path,monkeypatch):
+    path = tmp_path / "holder.json"
+    path.write_text(json.dumps({"sequence": 0}))
+    monkeypatch.setattr(locks.time,'sleep',lambda _:pytest.fail('success slept'))
+    locks._write_holder_sidecar(path,{'sequence':1})
+    assert json.loads(path.read_text())=={'sequence':1}
+
+def test_posix_adapter_permission_error_remains_immediate(tmp_path,monkeypatch):
+    attempts=[]
+    def fail(source,destination):
+        attempts.append(1)
+        error = PermissionError(errno.EACCES, "denied")
+        error.winerror = 5
+        raise error
+
+    adapter=SimpleNamespace(**{name:getattr(os,name) for name in ('fdopen','fsync')},name='posix',replace=fail)
+    monkeypatch.setattr(locks,'os',adapter)
+    monkeypatch.setattr(locks.time,'sleep',lambda _:pytest.fail('POSIX denial slept'))
+    with pytest.raises(PermissionError):
+        locks._write_holder_sidecar(tmp_path / "holder.json", {"sequence": 1})
+    assert attempts==[1] and not list(tmp_path.glob('*.tmp'))
diff --git a/tests/test_local_state_migration.py b/tests/test_local_state_migration.py
index a110f904a0..ea7d7018fc 100644
--- a/tests/test_local_state_migration.py
+++ b/tests/test_local_state_migration.py
@@ -4,6 +4,7 @@
 import hashlib
 import json
 import os
+import shlex
 import stat
 import subprocess
 import sys
@@ -24,6 +25,13 @@
     rollback_local_state_migration,
 )
 from loopx.project_prompt import build_new_project_prompt
+from loopx.control_plane.runtime.file_paths import windows_extended_path
+
+
+def _runtime_root_argument(command: str) -> str:
+    argv = shlex.split(command)
+    assert argv.count("--runtime-root") == 1
+    return argv[argv.index("--runtime-root") + 1]
 
 
 def _write_json(path: Path, payload: dict[str, object]) -> None:
@@ -386,7 +394,7 @@ def test_doctor_reads_legacy_capture_hosts_from_selected_runtime(
     tmp_path: Path,
 ) -> None:
     source, _target, _projects = _fixture(tmp_path, projects=1)
-    env = dict(os.environ, HOME=str(source.parents[1]), LOOPX_USAGE_PING="0")
+    env = dict(os.environ, HOME=str(source.parents[1]), USERPROFILE=str(source.parents[1]), LOOPX_USAGE_PING="0")
     env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1])
     completed = subprocess.run(
         [sys.executable, "-m", "loopx.cli", "--format", "json", "doctor"],
@@ -460,9 +468,9 @@ def test_existing_project_prompt_keeps_registered_goal_and_runtime_routes(tmp_pa
         allowed_domains=None,
         write_scope=None,
     )
-    assert f"--runtime-root {source}" in payload["quota_guard_command"]
-    assert f"--runtime-root {source}" in payload["quota_spend_command"]
-    assert f"--runtime-root {source}" in payload["connect_command"]
+    assert _runtime_root_argument(payload["quota_guard_command"]) == str(source)
+    assert _runtime_root_argument(payload["quota_spend_command"]) == str(source)
+    assert _runtime_root_argument(payload["connect_command"]) == str(source)
     assert ".codex/goals/goal-0/ACTIVE_GOAL_STATE.md" in payload["prompt"]
 
 
@@ -545,7 +553,7 @@ def test_strict_project_registry_keeps_legacy_route_and_wire_format(tmp_path: Pa
         adapter_kind="read_only_project_map_v0", adapter_status="connected-read-only",
         next_probe=None, spawn_allowed=False, allowed_domains=None, write_scope=None,
     )
-    assert f"--runtime-root {source}" in prompt["quota_guard_command"]
+    assert _runtime_root_argument(prompt["quota_guard_command"]) == str(source)
     assert ".codex/goals/goal-0/ACTIVE_GOAL_STATE.md" in prompt["prompt"]
 
     preview = migrate_local_state(source_runtime_root=source, target_runtime_root=target)
@@ -849,7 +857,7 @@ def test_preview_execute_and_verified_rollback_cover_all_registered_projects(tmp
         execute=True,
     )
     receipt_path = Path(receipt["backup_dir"]) / RECEIPT_NAME
-    assert receipt_path.exists()
+    assert windows_extended_path(receipt_path).is_file()
     assert not source.exists()
     assert _read(target / "registry.global.json")["common_runtime_root"] == str(target)
     for index, project in enumerate(projects):
@@ -932,7 +940,7 @@ def fail_rename_once(self: Path, destination: Path) -> Path:
     def fail_copy_once(original: Path, destination: Path, **kwargs: object) -> Path:
         nonlocal failed
         selected = backup / "snapshot" / ("0/registry.global.json" if failed_kind == "global-registry" else "1")
-        if not failed and failed_kind in {"registry", "global-registry"} and Path(original) == selected:
+        if not failed and failed_kind in {"registry", "global-registry"} and migration.windows_extended_path(Path(original)) == migration.windows_extended_path(selected):
             failed = True
             raise OSError(5, "synthetic rollback I/O failure")
         return original_copy(original, destination, **kwargs)
@@ -989,17 +997,17 @@ def test_hard_interruption_has_a_recoverable_receipt(
 interrupt = Path(sys.argv.pop(1))
 def rename(self, destination):
     result = original(self, destination)
-    if self == interrupt:
+    if migration.windows_extended_path(self) == migration.windows_extended_path(interrupt):
         os._exit(137)
     return result
 def replace(self, destination):
     result = original_replace(self, destination)
-    if Path(destination) == interrupt:
+    if migration.windows_extended_path(Path(destination)) == migration.windows_extended_path(interrupt):
         os._exit(137)
     return result
 def project_write(path, payload):
     original_project_write(path, payload)
-    if path == interrupt:
+    if migration.windows_extended_path(path) == migration.windows_extended_path(interrupt):
         os._exit(137)
 Path.rename = rename
 Path.replace = replace
@@ -1079,7 +1087,7 @@ def test_registry_recovery_rechecks_backup_before_replacement(
     original_copy = migration.shutil.copy2
 
     def corrupt_before_copy(original: Path, destination: Path, **kwargs: object) -> Path:
-        if Path(original) == snapshot:
+        if migration.windows_extended_path(Path(original)) == migration.windows_extended_path(snapshot):
             snapshot.write_text("corrupted backup\n", encoding="utf-8")
         return original_copy(original, destination, **kwargs)
 
@@ -1263,7 +1271,7 @@ def inject_after_backup_dir(
     ) -> None:
         nonlocal redirected
         original_mkdir(self, mode=mode, parents=parents, exist_ok=exist_ok)
-        if self == backup:
+        if migration.windows_extended_path(self) == migration.windows_extended_path(backup):
             redirected = True
 
     monkeypatch.setattr(migration, "_is_redirected_path", classify)
@@ -1613,6 +1621,7 @@ def test_custom_state_keeps_its_declared_file_through_migration_and_rollback(
     source, target, project, custom = _custom_state_fixture(tmp_path)
     local_registry = project / ".loopx" / "registry.json"
     before = local_registry.read_bytes()
+    declared = load_project_registry(local_registry)["goals"][0]["state_file"]
     preview = migrate_local_state(source_runtime_root=source, target_runtime_root=target)
     assert preview["goal_directory_count"] == 0
     receipt = migrate_local_state(
@@ -1621,7 +1630,7 @@ def test_custom_state_keeps_its_declared_file_through_migration_and_rollback(
     )
     migrated = load_project_registry(local_registry)
     assert migrated["common_runtime_root"] == str(target)
-    assert migrated["goals"][0]["state_file"] == "custom/STATE.md"
+    assert migrated["goals"][0]["state_file"] == declared
     assert custom.read_text(encoding="utf-8") == "custom state\n"
     rollback_local_state_migration(Path(receipt["backup_dir"]) / RECEIPT_NAME, execute=True)
     assert local_registry.read_bytes() == before
diff --git a/tests/test_local_state_migration_path_text.py b/tests/test_local_state_migration_path_text.py
new file mode 100644
index 0000000000..c1f278e2f0
--- /dev/null
+++ b/tests/test_local_state_migration_path_text.py
@@ -0,0 +1,73 @@
+"""Migration changes owned routes while preserving declared path text."""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from tests import test_local_state_migration as migration
+
+
+def _set_declaration(source, project, text, strict):
+    local = project / ".loopx" / "registry.json"
+    for registry in (local, source / "registry.global.json"):
+        payload = migration._read(registry)
+        payload["goals"][0]["state_file"] = text
+        if registry == local and strict:
+            migration._write_strict_project_registry(registry, payload)
+        else:
+            migration._write_json(registry, payload)
+    return local
+
+
+@pytest.mark.parametrize("strict", [False, True])
+@pytest.mark.parametrize("declaration", ["portable-relative", "native-relative", "absolute"])
+def test_owned_migration_path_text_and_exact_rollback(tmp_path, strict, declaration):
+    source, target, projects = migration._fixture(tmp_path, projects=1)
+    project = projects[0]
+    relative = Path(".codex") / "goals" / "goal-0" / "ACTIVE_GOAL_STATE.md"
+    text = {
+        "portable-relative": ".codex/goals/goal-0/ACTIVE_GOAL_STATE.md",
+        "native-relative": str(relative),
+        "absolute": str(project / relative),
+    }[declaration]
+    local = _set_declaration(source, project, text, strict)
+    original_local, original_global = local.read_bytes(), (source / "registry.global.json").read_bytes()
+    preview = migration.migrate_local_state(source_runtime_root=source, target_runtime_root=target)
+    assert local.read_bytes() == original_local and (source / "registry.global.json").read_bytes() == original_global
+    receipt = migration.migrate_local_state(
+        source_runtime_root=source, target_runtime_root=target,
+        expected_plan_id=preview["plan_id"], execute=True,
+    )
+    expected = str(project / ".loopx" / "goals" / "goal-0" / "ACTIVE_GOAL_STATE.md") if declaration == "absolute" else ".loopx/goals/goal-0/ACTIVE_GOAL_STATE.md"
+    for registry in (local, target / "registry.global.json"):
+        payload = migration.load_project_registry(registry)
+        assert payload["goals"][0]["state_file"] == expected
+        assert payload["common_runtime_root"] == str(target)
+    assert isinstance(json.loads(local.read_text()), list) == strict
+    assert (project / ".loopx" / "goals" / "goal-0" / "ACTIVE_GOAL_STATE.md").read_text() == "# goal-0\n"
+    migration.rollback_local_state_migration(Path(receipt["backup_dir"]) / migration.RECEIPT_NAME, execute=True)
+    assert local.read_bytes() == original_local
+    assert (source / "registry.global.json").read_bytes() == original_global
+    assert (project / relative).read_text() == "# goal-0\n" and not target.exists()
+
+
+@pytest.mark.parametrize("strict", [False, True])
+@pytest.mark.parametrize("native_text", [False, True])
+def test_custom_path_keeps_exact_declared_text(tmp_path, strict, native_text):
+    source, target, project, custom = migration._custom_state_fixture(tmp_path)
+    text = str(Path("custom") / "STATE.md") if native_text else "custom/STATE.md"
+    local = _set_declaration(source, project, text, strict)
+    original_local, original_global = local.read_bytes(), (source / "registry.global.json").read_bytes()
+    preview = migration.migrate_local_state(source_runtime_root=source, target_runtime_root=target)
+    assert preview["goal_directory_count"] == 0
+    receipt = migration.migrate_local_state(
+        source_runtime_root=source, target_runtime_root=target,
+        expected_plan_id=preview["plan_id"], execute=True,
+    )
+    for registry in (local, target / "registry.global.json"):
+        assert migration.load_project_registry(registry)["goals"][0]["state_file"] == text
+    assert custom.read_text() == "custom state\n"
+    migration.rollback_local_state_migration(Path(receipt["backup_dir"]) / migration.RECEIPT_NAME, execute=True)
+    assert local.read_bytes() == original_local and (source / "registry.global.json").read_bytes() == original_global
+    assert custom.read_text() == "custom state\n" and not target.exists()
diff --git a/tests/test_windows_atomic_writes.py b/tests/test_windows_atomic_writes.py
index 1cfc8454ad..192bafa35e 100644
--- a/tests/test_windows_atomic_writes.py
+++ b/tests/test_windows_atomic_writes.py
@@ -59,7 +59,7 @@ def _write_botmux_binding(path: Path) -> None:
 
 
 def _write_native_isolation(path: Path) -> None:
-    native_codex_isolation._atomic_write_text(path, "prompt = \"ready\"\n")
+    native_codex_isolation._atomic_write_text(path, 'prompt = "ready"\n')
 
 
 @pytest.mark.parametrize(
@@ -122,3 +122,63 @@ def test_atomic_writers_skip_unsupported_windows_directory_fsync(
     content = target.read_text(encoding="utf-8")
     actual = json.loads(content) if isinstance(expected, dict) else content
     assert actual == expected
+
+
+def _long_private_target(tmp_path: Path, *, parent_length: int) -> Path:
+    parent = tmp_path
+    while len(str(parent)) < parent_length:
+        remaining = parent_length - len(str(parent)) - 1
+        assert remaining > 0
+        component_length = min(40, remaining)
+        if remaining == 41:
+            component_length = 39
+        parent /= "p" * component_length
+    assert len(str(parent)) == parent_length
+    return parent / ("a" * 64 + ".json")
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Native Windows extended-path IO")
+@pytest.mark.parametrize("parent_length", [180, 280], ids=["long-temp", "long-target"])
+def test_lark_private_json_long_paths(tmp_path: Path, parent_length: int) -> None:
+    from loopx.control_plane.runtime.file_paths import windows_extended_path
+
+    target = _long_private_target(tmp_path, parent_length=parent_length)
+    assert len(str(target)) + 14 > 260
+    if parent_length == 180:
+        assert len(str(target)) < 260
+    else:
+        assert len(str(target)) > 260
+    address = windows_extended_path(target)
+    private_json.write_private_json_atomic(target, {"status": "初次", "count": 1})
+    assert json.loads(address.read_text(encoding="utf-8")) == {
+        "status": "初次",
+        "count": 1,
+    }
+    private_json.write_private_json_atomic(target, {"status": "更新", "count": 2})
+    assert json.loads(address.read_text(encoding="utf-8")) == {
+        "status": "更新",
+        "count": 2,
+    }
+    assert not list(address.parent.glob("*.tmp"))
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Native Windows extended-path IO")
+def test_lark_private_json_long_path_interrupted_replace(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from loopx.control_plane.runtime.file_paths import windows_extended_path
+
+    target = _long_private_target(tmp_path, parent_length=180)
+    address = windows_extended_path(target)
+    address.parent.mkdir(parents=True, exist_ok=True)
+    address.write_text('{"status": "original"}\n', encoding="utf-8")
+    before = address.read_bytes()
+
+    def interrupted_replace(*args: object) -> None:
+        raise OSError("interrupted replacement")
+
+    monkeypatch.setattr(file_lock.os, "replace", interrupted_replace)
+    with pytest.raises(OSError, match="interrupted replacement"):
+        private_json.write_private_json_atomic(target, {"status": "replacement"})
+    assert address.read_bytes() == before
+    assert not list(address.parent.glob("*.tmp"))