diff --git a/src/agent_comms/diagnostics.py b/src/agent_comms/diagnostics.py
index 7b46814031..4ac6633ffa 100644
--- a/src/agent_comms/diagnostics.py
+++ b/src/agent_comms/diagnostics.py
@@ -14,10 +14,11 @@
 from typing import TYPE_CHECKING, Literal
 from uuid import uuid4
 
+
 from .child_process import Platform
 from .field_codec import FieldCodec, Id128Text, MalformedRecord
 from .locked_store import DurableFile, PatternFile
-from .store_files import _atomic_write_text
+from .store_files import AtomicFile
 
 if TYPE_CHECKING:
     from .activity import DrainDiagnostic
@@ -277,5 +278,5 @@ def write(cls, root: Path, name: str, document: dict) -> Path:
         path = cls.path_in(root, name=name)
         path.parent.mkdir(mode=0o700, exist_ok=True)
         Platform.current().sync_directory(root)
-        _atomic_write_text(path, json.dumps(document, indent=2) + "\n")
+        AtomicFile(path).write_text(json.dumps(document, indent=2) + "\n")
         return path.resolve()
diff --git a/src/agent_comms/exporting.py b/src/agent_comms/exporting.py
index 28c32d44ec..a4960abea6 100644
--- a/src/agent_comms/exporting.py
+++ b/src/agent_comms/exporting.py
@@ -6,7 +6,6 @@
 import json
 import math
 import os
-import tempfile
 from abc import abstractmethod
 from collections import deque
 from collections.abc import Callable, Iterable, Mapping
@@ -15,14 +14,14 @@
 from pathlib import Path
 from typing import BinaryIO, ClassVar
 
-from .child_process import Platform
 from .channels import Channel
 from .cli_result import CliResult
 from .declared_family import DeclaredFamily
 from .field_codec import FieldCodec, NullMember, SingleFieldValue, SubclassUnion
-from .messages import Message
 from .message_reference import MessageReference
+from .messages import Message
 from .registration import Registration
+from .store_files import AtomicFile, ExclusiveAtomicFile
 
 
 class WireExportFormat(DeclaredFamily, affix="Format"):
@@ -445,35 +444,25 @@ def export(
         if not overwrite and os.path.lexists(destination):
             raise FileExistsError(f"Wire-export destination {str(destination)!r} already exists.")
 
-        descriptor, temporary = tempfile.mkstemp(
-            prefix=f".{destination.name}.", suffix=".tmp", dir=destination.parent
-        )
-        temporary_path = Path(temporary)
+        artifact = AtomicFile(destination) if overwrite else ExclusiveAtomicFile(destination)
         stats = ExportSelectionStats()
         exported = 0
         first_selected: Message | None = None
         last_selected: Message | None = None
-        try:
-            with os.fdopen(descriptor, "wb") as output:
-                writer = _ArtifactWriter(output)
-                writer.write(header)
-                rows = self._rows(messages, stats)
-                for row in self.limit.select(
-                    rows, len(header), self.boundary.export_started_at, stats
-                ):
-                    writer.write(row.data)
-                    exported += 1
-                    if first_selected is None:
-                        first_selected = row.message
-                    last_selected = row.message
-                output.flush()
-                os.fsync(output.fileno())
-                bytes_written = writer.size
-                output_sha256 = writer.digest.hexdigest()
-
-            self._publish(temporary_path, destination, overwrite=overwrite)
-        finally:
-            temporary_path.unlink(missing_ok=True)
+        with artifact.staged() as output:
+            writer = _ArtifactWriter(output)
+            writer.write(header)
+            rows = self._rows(messages, stats)
+            for row in self.limit.select(
+                rows, len(header), self.boundary.export_started_at, stats
+            ):
+                writer.write(row.data)
+                exported += 1
+                if first_selected is None:
+                    first_selected = row.message
+                last_selected = row.message
+            bytes_written = writer.size
+            output_sha256 = writer.digest.hexdigest()
 
         omitted = stats.source_messages - exported
         return WireExportProvenance(
@@ -530,16 +519,3 @@ def _validate_sequence(message: Message, previous: int) -> int:
         if message.seq <= previous:
             raise ValueError("Wire-export envelopes must have unique ascending sequences.")
         return message.seq
-
-    @staticmethod
-    def _publish(temporary: Path, destination: Path, *, overwrite: bool) -> None:
-        if overwrite:
-            os.replace(temporary, destination)
-        else:
-            try:
-                os.link(temporary, destination)
-            except FileExistsError:
-                raise FileExistsError(
-                    f"Wire-export destination {str(destination)!r} already exists."
-                ) from None
-        Platform.current().sync_directory(destination.parent)
diff --git a/src/agent_comms/fresh_private_session.py b/src/agent_comms/fresh_private_session.py
index f24fb4bae1..3f0d8ab332 100644
--- a/src/agent_comms/fresh_private_session.py
+++ b/src/agent_comms/fresh_private_session.py
@@ -8,7 +8,7 @@
 from pathlib import Path
 from uuid import uuid4
 
-from .child_process import Platform
+from agent_comms.child_process import Platform
 from .field_codec import RecordedTimestamp
 from .native_entries import SessionEntry
 from .native_failures import NativePiUnavailable
diff --git a/src/agent_comms/goal_attempts.py b/src/agent_comms/goal_attempts.py
index 58773669dc..aff5467abc 100644
--- a/src/agent_comms/goal_attempts.py
+++ b/src/agent_comms/goal_attempts.py
@@ -62,6 +62,7 @@
 from .locked_store import DurableFile, PublishedSQLiteFile, WireWakeFile
 from .pi_counts import PiCount, UnreportedCount
 from .pi_payloads import PiUsage
+from .store_files import AtomicFile
 from .typed_table import Column, Generated, SQLiteJournalMode, TypedTable
 
 if TYPE_CHECKING:
@@ -708,14 +709,12 @@ def initialize(cls, root: str | Path) -> GoalAttemptStore:
                 conn.execute("PRAGMA foreign_keys=ON")
                 with cls.write_transaction(conn):
                     GoalLedgerTable.create_schema(conn)
-            platform = Platform.current()
-            platform.sync_file(staging)
-            platform.sync_directory(directory)
+            AtomicFile(staging).sync()
             # Publish only the complete schema; a concurrent initializer may win.
             # The staging descriptor is closed before its inode becomes shared.
             with suppress(FileExistsError):
                 os.link(staging, path, follow_symlinks=False)
-            platform.sync_directory(directory)
+            Platform.current().sync_directory(directory)
         except (sqlite3.Error, OSError) as error:
             raise StorageUncertainError("Goal attempt initialization is uncertain.") from error
         finally:
@@ -751,9 +750,7 @@ def database(cls, path: Path) -> sqlite3.Connection:
             raise StorageUncertainError("Goal attempt database unavailable.") from error
 
     def _sync(self) -> None:
-        platform = Platform.current()
-        platform.sync_file(self.path)
-        platform.sync_directory(self.root)
+        AtomicFile(self.path).sync()
 
     def _change(
         self, operation: Callable[[sqlite3.Connection], _T], verify: Callable[[_T], bool]
diff --git a/src/agent_comms/goal_history.py b/src/agent_comms/goal_history.py
index ff6f4bdbcf..8cca5cf096 100644
--- a/src/agent_comms/goal_history.py
+++ b/src/agent_comms/goal_history.py
@@ -23,11 +23,17 @@
 from agent_comms.owner_requests import NoRequestRevision, RequestAtRevision, RequestRevision
 
 from .child_process import Platform
-from .field_codec import FieldCodec, MalformedRecord, HistoryExclude
+from .field_codec import FieldCodec, HistoryExclude, MalformedRecord
 from .goals import GoalDeclaration, NoGoal
 from .locked_store import DurableFile
+from .store_files import AtomicFile
 from .typed_table import (
-    Column, DurableTables, Index, JsonStorage, ReconciledFile, TypedTable,
+    Column,
+    DurableTables,
+    Index,
+    JsonStorage,
+    ReconciledFile,
+    TypedTable,
 )
 
 
@@ -169,9 +175,7 @@ def _connect(self) -> sqlite3.Connection:
             raise GoalHistoryError("Goal history database is unavailable.") from error
 
     def _sync(self) -> None:
-        platform = Platform.current()
-        platform.sync_file(self.path)
-        platform.sync_directory(self.path.parent)
+        AtomicFile(self.path).sync()
 
     def _initialize(self) -> None:
         try:
@@ -243,10 +247,8 @@ def _sync_visible_registry(self) -> None:
         if guard.exists():
             targets.append(guard)
         try:
-            platform = Platform.current()
             for target in targets:
-                platform.sync_file(target)
-            platform.sync_directory(self.registry_path.parent)
+                AtomicFile(target).sync()
         except OSError as error:
             raise GoalHistoryError("Cannot sync visible goal authority.") from error
 
diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py
index 8c175b58cd..26261f9da9 100644
--- a/src/agent_comms/historical_views.py
+++ b/src/agent_comms/historical_views.py
@@ -47,7 +47,7 @@
 from .registration import Registration
 from .registry_provenance import RegistryProvenance
 from .revision_views import BusSourceView
-from .store_files import _atomic_write_text, _store_lock
+from .store_files import AtomicFile, _store_lock
 from .thread_identity import UnrecordedThreadIncarnation
 from .thread_provenance import ThreadProvenance
 
@@ -505,10 +505,7 @@ def attach(self, source_root: Path, *, source_read: HistorySource | None = None)
                 )
                 if not source.provenance.threads:
                     raise ValueError("Historical source has no identity declarations")
-                _atomic_write_text(
-                    self.path,
-                    json.dumps([FieldCodec.encode(item) for item in (*sources, source)]),
-                )
+                AtomicFile(self.path).write_text(json.dumps([FieldCodec.encode(item) for item in (*sources, source)]))
                 return source
             except BaseException:
                 shutil.rmtree(stage)
diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py
index 1d29f7eee1..d16d2e3908 100644
--- a/src/agent_comms/installer/backup.py
+++ b/src/agent_comms/installer/backup.py
@@ -25,7 +25,7 @@
 from agent_comms.installer.interpreter_operation import RouteSelection
 from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile
 from agent_comms.private_path import FileObservation, FileRevision, MissingFile
-from agent_comms.store_files import _atomic_write_text
+from agent_comms.store_files import AtomicFile
 
 
 class UpgradeRefused(CoordinationError):
@@ -192,7 +192,7 @@ def restore_route(self) -> None:
                 return
             saved = self.path / 'active-route.json'
             if saved.exists():
-                _atomic_write_text(self.route, saved.read_text())
+                AtomicFile(self.route).write_text(saved.read_text())
             else:
                 os.unlink(self.route.name, dir_fd=directory)
                 os.fsync(directory)
@@ -239,13 +239,13 @@ def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path)
                 backup.route_revision = FileRevision.capture(route)
             return backup
         path.mkdir(mode=0o700, parents=True)
-        _atomic_write_text(path / 'selected.json', json.dumps(selected))
+        AtomicFile(path / 'selected.json').write_text(json.dumps(selected))
         contents = backup.capture(files)
         with route_directory(route):
             if route.exists():
                 shutil.copy2(route, path / 'active-route.json')
             backup.route_revision = FileRevision.capture(route)
-        _atomic_write_text(path / 'files.json', json.dumps(FieldCodec.encode(contents), indent=1))
+        AtomicFile(path / 'files.json').write_text(json.dumps(FieldCodec.encode(contents), indent=1))
         return backup
 
     def retain(self, spill_dir: Path) -> tuple[Path, ...]:
@@ -348,8 +348,7 @@ def restore(self) -> None:
                 stage = Path(temporary)
                 try:
                     shutil.copy2(self.data_root / name, stage)
-                    Platform.current().sync_file(stage)
-                    os.replace(stage, target)
+                    AtomicFile(target).replace(stage)
                 finally:
                     stage.unlink(missing_ok=True)
             else:
diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py
index 8a8746b420..08a8ba74c2 100644
--- a/src/agent_comms/locked_store.py
+++ b/src/agent_comms/locked_store.py
@@ -22,8 +22,15 @@
 from .child_process import Platform
 from .coordination_errors import CoordinationReadUnavailable
 from .field_codec import FieldCodec, MalformedRecord
+from .private_path import PrivateFileRole
 from .store_files import (
-    SharedLock, StoreLock, StoreLockOwner, StoreOwner, _store_lock, store_lock_path,
+    AtomicFile,
+    SharedLock,
+    StoreLock,
+    StoreLockOwner,
+    StoreOwner,
+    _store_lock,
+    store_lock_path,
 )
 
 T = TypeVar("T")
@@ -86,49 +93,32 @@ def _write_unlocked(self, text: str) -> None:
         old-or-new semantics, not a multi-file transaction or recovery journal.
         """
         fd, staged = StoreOwner.stage_file(self.path)
+        os.close(fd)
+        staged.unlink()
         backup = StoreOwner.rollback_file(staged)
-        published = False
-        had_original = False
-        keep_backup = False
         try:
-            with os.fdopen(fd, "w", encoding="utf-8") as output:
-                try:
-                    mode = stat.S_IMODE(self.path.stat().st_mode)
-                except FileNotFoundError:
-                    mode = 0o600
-                output.write(text)
-                output.flush()
-                os.chmod(staged, mode)
-                os.fsync(output.fileno())
+            mode = stat.S_IMODE(self.path.stat().st_mode)
+            os.link(self.path, backup)
+        except FileNotFoundError:
+            mode = PrivateFileRole.permissions
+        try:
             try:
-                os.link(self.path, backup)
-                had_original = True
-            except FileNotFoundError:
-                pass
-            os.replace(staged, self.path)
-            published = True
-            Platform.current().sync_directory(self.path.parent)
-            # A failed backup cleanup is still a failed update: retain the old
-            # inode until unlink succeeds so the exception path can restore it.
-            backup.unlink(missing_ok=True)
-        except BaseException:
-            if published:
-                if had_original:
-                    try:
+                AtomicFile(self.path, mode).write_text(text)
+                backup.unlink(missing_ok=True)
+            except BaseException:
+                if backup.exists():
+                    if not self.path.exists() or not self.path.samefile(backup):
                         os.replace(backup, self.path)
-                    except BaseException:
-                        # Keep the recovery bytes if the filesystem also
-                        # refuses rollback; never erase the remaining copy.
-                        keep_backup = True
-                        raise
+                        Platform.current().sync_directory(self.path.parent)
                 else:
-                    self.path.unlink()
-                Platform.current().sync_directory(self.path.parent)
-            raise
+                    self.path.unlink(missing_ok=True)
+                    Platform.current().sync_directory(self.path.parent)
+                raise
         finally:
-            staged.unlink(missing_ok=True)
-            if not keep_backup:
-                backup.unlink(missing_ok=True)
+            # A failed restore leaves the last original inode for recovery.
+            if backup.exists() and self.path.exists() and self.path.samefile(backup):
+                backup.unlink()
+
 
 
 class RootFile:
diff --git a/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py
index 208d6fc839..cf7ed02035 100644
--- a/src/agent_comms/maintenance_barrier.py
+++ b/src/agent_comms/maintenance_barrier.py
@@ -26,9 +26,9 @@
 from .locked_store import DurableFile, LockedStore, RegistryFile
 from .send_outcome import GrantedSend, HeldSend
 from .store_files import (
+    AtomicFile,
     SharedLock,
     _async_store_lock,
-    _atomic_write_text,
     _store_lock,
 )
 
@@ -406,8 +406,8 @@ def _write_unlocked(self, hold: MaintenanceState) -> None:
         marker_text = json.dumps(FieldCodec.encode(marker), sort_keys=True)
         state_text = json.dumps(FieldCodec.encode(hold), sort_keys=True)
         # The marker leads so an interrupted replacement still holds new turns.
-        _atomic_write_text(self.marker_path, marker_text)
-        _atomic_write_text(self.state_path, state_text)
+        AtomicFile(self.marker_path).write_text(marker_text)
+        AtomicFile(self.state_path).write_text(state_text)
 
     def begin(self, operator: MaintenanceOperator) -> MaintenanceState:
         """Hold new turns before fleet drain; an uncertain write stays closed."""
diff --git a/src/agent_comms/native_package.py b/src/agent_comms/native_package.py
index a7c56cdb25..927c95ce81 100644
--- a/src/agent_comms/native_package.py
+++ b/src/agent_comms/native_package.py
@@ -25,6 +25,8 @@
 from agent_comms.native_build_inputs import INPUTS_PREFIX, NativeBuildInputs
 from agent_comms.private_path import PrivateDirectoryRole, TrustedAncestorRole
 
+from .store_files import AtomicFile
+
 TREE_PREFIX = "# agent-comms-native-tree-v1 "
 MAX_ENTRIES = 30_000
 MAX_BYTES = 512 * 1024**2
@@ -311,8 +313,7 @@ def stage(self) -> Path:
         try:
             with os.fdopen(descriptor, "wb") as stream:
                 stream.write(self.content)
-                stream.flush()
-                os.fsync(stream.fileno())
+
             os.chmod(temporary, 0o600)
             if temporary.stat().st_nlink != 1:
                 raise NativePackageError("Native extension source staging file is shared")
@@ -357,7 +358,7 @@ def install_declared_sources(root: Path, snapshot_root: Path) -> None:
             if not source.is_installed():
                 replacements.append((source.stage(), source))
         for temporary, source in replacements:
-            os.replace(temporary, source.destination)
+            AtomicFile(source.destination).replace(temporary)
         for source in sources:
             source.verify_installed()
         for parent in {source.destination.parent for source in sources}:
diff --git a/src/agent_comms/native_session_files.py b/src/agent_comms/native_session_files.py
index daf24b3401..ec75c274b0 100644
--- a/src/agent_comms/native_session_files.py
+++ b/src/agent_comms/native_session_files.py
@@ -8,15 +8,16 @@
 from pathlib import Path
 from typing import Self
 
-from .child_process import Platform
-from .errors import RelationViolationError
-from .native_failures import NativePiUnavailable
-from agent_comms.locked_store import DurableFile, PatternFile, SQLiteFile
+from agent_comms.child_process import Platform
 from agent_comms.field_codec import Id128Text
-from agent_comms.participant_identity import ParticipantIdentity
+from agent_comms.locked_store import DurableFile, PatternFile, SQLiteFile
 from agent_comms.native_session_reopen import NativeSessionIdentity
+from agent_comms.participant_identity import ParticipantIdentity
 from agent_comms.private_path import PrivateDirectoryRole, TrustedAncestorRole
 
+from .errors import RelationViolationError
+from .native_failures import NativePiUnavailable
+
 
 class SessionFiles:
     """Declared session members share reference checks and deletion lifecycle."""
diff --git a/src/agent_comms/private_registry_guard.py b/src/agent_comms/private_registry_guard.py
index afcaacad41..c8d6c2f05f 100644
--- a/src/agent_comms/private_registry_guard.py
+++ b/src/agent_comms/private_registry_guard.py
@@ -28,8 +28,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, NoReturn
 
-from . import store_files
 from .child_process import Platform
+from . import store_files
 from .coordination_database import CoordinationStore
 from .coordinator import Coordination
 from .errors import RelationViolationError
@@ -210,7 +210,7 @@ def namespace(self) -> str:
         _reject("native namespace requires the original registry guard")
 
     def write(self, registry_path: Path, text: str) -> None:
-        store_files._atomic_write_text(registry_path, text)
+        store_files.AtomicFile(registry_path).write_text(text)
 
     def register_restored(self, threads: Iterable[Thread]) -> None:
         """A public bus has no coordinator identities for restored threads."""
@@ -446,7 +446,7 @@ def write(self, registry_path: Path, text: str) -> None:
         # A pending guard after any failed write MUST remain fail-closed.
         digest = hashlib.sha256(b"present\0" + text.encode("utf-8")).digest()
         sequence, slot = self.prepare(digest)
-        store_files._atomic_write_text(registry_path, text)
+        store_files.AtomicFile(registry_path).write_text(text)
         self.commit(sequence, slot, digest)
 
     def register_restored(self, threads: Iterable[Thread]) -> None:
@@ -470,29 +470,22 @@ def create_pending(self) -> None:
                 os.fsync(registry_fd)
             finally:
                 os.close(registry_fd)
-        platform = Platform.current()
-        platform.sync_directory(self.registry_path.parent)  # baseline registry name and existing root
+        Platform.current().sync_directory(self.registry_path.parent)
         fd = os.open(
             self.path,
             os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
             0o600,
         )
         try:
-            if (
-                os.pwrite(
-                    fd,
-                    _record(self.root_id, 0, _COMMITTED, digest) * 2,
-                    0,
-                )
-                != 2 * _SIZE
-            ):
+            if os.pwrite(fd, _record(self.root_id, 0, _COMMITTED, digest) * 2, 0) != 2 * _SIZE:
                 _reject("initial slots were incomplete")
             os.fsync(fd)
-            platform.sync_directory(self.registry_path.parent)
+            Platform.current().sync_directory(self.registry_path.parent)
             self._write(fd, 1, _record(self.root_id, 1, _PENDING, digest))
         finally:
             os.close(fd)
 
+
     def commit_initial(self) -> None:
         fd = self._open()
         try:
@@ -524,8 +517,7 @@ class PrivateOwnerRenameIntent(DurableFile):
     completed: bool = False
 
     def write(self, root: Path) -> None:
-        store_files._atomic_write_text(self.path_in(root),
-                           json.dumps(FieldCodec.encode(self), sort_keys=True))
+        store_files.AtomicFile(self.path_in(root)).write_text(json.dumps(FieldCodec.encode(self), sort_keys=True))
 
     @classmethod
     def complete(cls, root: Path) -> None:
diff --git a/src/agent_comms/restart_queue.py b/src/agent_comms/restart_queue.py
index 8cf98a78b6..7262f76786 100644
--- a/src/agent_comms/restart_queue.py
+++ b/src/agent_comms/restart_queue.py
@@ -25,10 +25,10 @@
 from .coordination_errors import CoordinationError
 from .coordinator import Coordination
 from .declared_family import DeclaredFamily
-from .lifecycle_capability import Active, LifecycleCapability, Pending
 from .errors import RelationViolationError
 from .field_codec import FieldCodec, WireOmitDefault
 from .launch_runtime import LaunchRuntime, RestartEnvironment
+from .lifecycle_capability import Active, LifecycleCapability, Pending
 from .native_arguments import GivenBinary, NativeArguments
 from .owner_cutover import PreserveOwnerRuntime
 from .owner_launch import RetainedOwnerLaunch
@@ -37,7 +37,7 @@
 from .private_path import PrivateDirectoryRole
 from .restart_refusals import RestartRefusal
 from .restart_scope import OwnersNamed
-from .store_files import _atomic_write_text, _store_lock
+from .store_files import AtomicFile, _store_lock
 from .thread_identity import RecordedThreadIncarnation
 from .watched import ChangeSignal, DirectorySource, FileSource
 
@@ -245,7 +245,7 @@ def move(self, path: Path, record: QueuedRestart) -> None:
 
     def save(self, record: QueuedRestart) -> None:
         path = self.active_path(record.selection.identity.incarnation.require_recorded())
-        _atomic_write_text(path, json.dumps(FieldCodec.encode(record), sort_keys=True))
+        AtomicFile(path).write_text(json.dumps(FieldCodec.encode(record), sort_keys=True))
         if not record.state.items_with(Active, record):
             self.move(path, record)
 
diff --git a/src/agent_comms/selected_tool_broker.py b/src/agent_comms/selected_tool_broker.py
index 5c5a63706b..e8fc10b09b 100644
--- a/src/agent_comms/selected_tool_broker.py
+++ b/src/agent_comms/selected_tool_broker.py
@@ -18,6 +18,7 @@
 from pathlib import Path
 from typing import ClassVar
 
+from agent_comms.child_process import Platform
 from agent_comms.coordinator import Coordination
 
 from .comms import Comms
diff --git a/src/agent_comms/selected_write_plan.py b/src/agent_comms/selected_write_plan.py
index 3a0bd02930..229c35f005 100644
--- a/src/agent_comms/selected_write_plan.py
+++ b/src/agent_comms/selected_write_plan.py
@@ -12,14 +12,15 @@
 import hashlib
 import json
 import os
-import secrets
 from contextlib import suppress
 from dataclasses import dataclass, fields
 from pathlib import Path
-from typing import TYPE_CHECKING, Literal, Annotated
+from typing import TYPE_CHECKING, Annotated, Literal
 
+from agent_comms.child_process import Platform
 from .coordination_database import CoordinationStore
 from .locked_store import DurableFile, PatternFile
+from .store_files import AtomicFile
 
 if TYPE_CHECKING:
     from .acp_extension import SelectedWriteAcceptedUpdate
@@ -269,16 +270,6 @@ def applied(self, assignment: WakeAssignment, owner: Thread, operation_id: str)
         if accepted.operation_id != operation_id:
             raise IdentityConflict("Selected write intent is no longer unique")
         completed = accepted.applied()
-        temporary = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp")
-        fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
-        try:
-            with os.fdopen(fd, "wb") as stream:
-                stream.write(json.dumps(FieldCodec.encode(completed), sort_keys=True, separators=(",", ":")).encode())
-                stream.flush()
-                os.fsync(stream.fileno())
-            os.replace(temporary, path)
-            Platform.current().sync_directory(self.directory)
-        except BaseException:
-            # A post-write failure is UNKNOWN. The bus claim and attempt
-            # remain visible; no path automatically reissues the write.
-            raise
+        AtomicFile(path).write_bytes(json.dumps(
+            FieldCodec.encode(completed), sort_keys=True, separators=(",", ":")
+        ).encode())
diff --git a/src/agent_comms/store_files.py b/src/agent_comms/store_files.py
index ff6f076de9..ef5dd4d00b 100644
--- a/src/agent_comms/store_files.py
+++ b/src/agent_comms/store_files.py
@@ -432,23 +432,58 @@ def _async_store_lock(
     )
 
 
-def _atomic_write_text(
-    path: Path, text: str, *,
-    mode: int = PrivateFileRole.permissions,
-) -> None:
-    """Replace a snapshot and durably sync its name."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    fd, temporary_path = StoreOwner.stage_file(path)
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as output:
-            os.chmod(temporary_path, mode)
-            output.write(text)
-            output.flush()
-            os.fsync(output.fileno())
-        os.replace(temporary_path, path)
-        Platform.current().sync_directory(path.parent)
-    finally:
-        temporary_path.unlink(missing_ok=True)
+@dataclass(frozen=True)
+class AtomicFile:
+    """A staged replacement always syncs its contents and directory entry."""
+
+    path: Path
+    mode: int = PrivateFileRole.permissions
+
+    @contextmanager
+    def staged(self) -> Iterator[BinaryIO]:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        descriptor, temporary = StoreOwner.stage_file(self.path)
+        try:
+            with os.fdopen(descriptor, "wb") as output:
+                os.chmod(temporary, self.mode)
+                yield output
+                output.flush()
+                os.fsync(output.fileno())
+            self._publish(temporary)
+        finally:
+            temporary.unlink(missing_ok=True)
+
+    def write_text(self, text: str) -> None:
+        self.write_bytes(text.encode("utf-8"))
+
+    def write_bytes(self, data: bytes) -> None:
+        with self.staged() as output:
+            output.write(data)
+
+    def replace(self, staged: Path) -> None:
+        """Publish an independently prepared file after syncing its data."""
+        Platform.current().sync_file(staged)
+        self._publish(staged)
+
+    def _publish(self, staged: Path) -> None:
+        self._install(staged)
+        Platform.current().sync_directory(self.path.parent)
+
+    def _install(self, staged: Path) -> None:
+        os.replace(staged, self.path)
+
+    def sync(self) -> None:
+        """Sync an existing file without replacing an open database's inode."""
+        platform = Platform.current()
+        platform.sync_file(self.path)
+        platform.sync_directory(self.path.parent)
+
+
+class ExclusiveAtomicFile(AtomicFile):
+    """Create a new name atomically, refusing an existing destination."""
+
+    def _install(self, staged: Path) -> None:
+        os.link(staged, self.path)
 
 
 def _repair_trailing_jsonl(path: Path) -> None:
diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py
index d1d18b589d..a7c7424309 100644
--- a/src/agent_comms/thread_management.py
+++ b/src/agent_comms/thread_management.py
@@ -30,8 +30,8 @@
 from .catalog_store import ChannelCatalog
 from .collaboration_ledger import CollaborationLedger
 from .errors import RelationViolationError
-from .importing import ImportLimits, ImportSnapshot, ThreadImported
 from .field_codec import FieldCodec, NullMember, RecordMember, SingleFieldValue, SubclassUnion
+from .importing import ImportLimits, ImportSnapshot, ThreadImported
 from .launch_runtime import LaunchRuntime, TagListText
 from .message_bus import MessageBus
 from .native_arguments import ConfiguredBinary
@@ -43,13 +43,13 @@
 )
 from .registry_document import RegistrySnapshot
 from .registry_provenance import RegistryNames
-from .store_files import _atomic_write_text
-from .thread_identity import AgentThreadRole, ThreadRole
-from .thread_text import ForkedParent, GivenThreadText, NoThreadText, ThreadText
 from .session_file import SavedSessionFile
+from .store_files import AtomicFile
+from .thread_identity import AgentThreadRole, ThreadRole
 from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel
 from .thread_owned_state import ThreadOwnedState
 from .thread_rename import RenameThreadResult, ThreadRename
+from .thread_text import ForkedParent, GivenThreadText, NoThreadText, ThreadText
 from .threads import KeptModel, ModelChoice, NextTurnConfiguration, Thread, current_thread
 
 
@@ -258,7 +258,7 @@ def import_thread(
                 or stat.S_IMODE(info.st_mode) != 0o700
             ):
                 raise ValueError("Imported session directory is not owner-controlled.")
-            _atomic_write_text(session_path, snapshot.pi_session(project))
+            AtomicFile(session_path).write_text(snapshot.pi_session(project))
             try:
                 self.registry._declare_unlocked(thread, StoppedThreadStatus())
                 self.bus.mark_delivered_through(name, self.bus.bus.revision())
diff --git a/src/agent_comms/threads.py b/src/agent_comms/threads.py
index e01e1f3499..7ed9c60095 100644
--- a/src/agent_comms/threads.py
+++ b/src/agent_comms/threads.py
@@ -8,8 +8,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field, replace
 from pathlib import Path
-from typing import TYPE_CHECKING, ClassVar, Annotated
-
+from typing import TYPE_CHECKING, Annotated, ClassVar
 
 from .agent_backend import AgentBackend, NewThreadBackend
 from .child_process import NoProcess, ProcessIdentity, ProcessState
@@ -17,8 +16,15 @@
 from .context_segments.wake_values import WakeWorkContext
 from .errors import RelationViolationError
 from .field_codec import (
-    Argument, Description, FieldCodec, NullMember, RegistrationPolicy,
-    SingleFieldValue, SubclassUnion, WireOmitDefault, WireRequired,
+    Argument,
+    Description,
+    FieldCodec,
+    NullMember,
+    RegistrationPolicy,
+    SingleFieldValue,
+    SubclassUnion,
+    WireOmitDefault,
+    WireRequired,
 )
 from .goals import (
     GoalDeclaration,
@@ -40,14 +46,11 @@
     InheritEmpty,
     InheritPrevious,
 )
+from .session_file import NoSessionFile, SessionFile
 from .storage_rules import StoragePath, StorageRoots, StorageRule
-from .store_files import _atomic_write_text
+from .store_files import AtomicFile
 from .task_sources import TaskScope
 from .thread_execution import NativeThreadExecution, ThreadExecution
-from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel
-from .thread_text import NoParent, NoThreadText, ThreadParent, ThreadText
-from .session_file import NoSessionFile, SessionFile
-from .thread_provenance import ThreadProvenance
 from .thread_identity import (
     AgentThreadRole,
     ThreadRole,
@@ -55,6 +58,9 @@
     TurnId,
     TurnIdentity,
 )
+from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel
+from .thread_provenance import ThreadProvenance
+from .thread_text import NoParent, NoThreadText, ThreadParent, ThreadText
 from .turn_lease import NoTurnRecord, TurnLease, TurnLeaseFence, TurnRecord, TurnState
 
 if TYPE_CHECKING:
@@ -284,7 +290,7 @@ def native_launch_arguments(
         catalog = ThreadToolCatalogResult(
             tuple(NativeRequestTools.catalog()), self.service_tier, launch.arguments(),
         )
-        _atomic_write_text(path, json.dumps(catalog.to_wire(), separators=(",", ":")))
+        AtomicFile(path).write_text(json.dumps(catalog.to_wire(), separators=(",", ":")))
         return (*arguments.argv, "--comms-catalog", str(path))
 
     def launch(self) -> ThreadLaunch:
diff --git a/tests/test_atomic_file.py b/tests/test_atomic_file.py
new file mode 100644
index 0000000000..f47760cbdc
--- /dev/null
+++ b/tests/test_atomic_file.py
@@ -0,0 +1,88 @@
+"""File contents and directory names become durable through the same writer."""
+
+import os
+import stat
+
+import pytest
+
+from agent_comms.child_process import Platform, Win32Platform
+from agent_comms.store_files import AtomicFile, ExclusiveAtomicFile
+
+
+@pytest.mark.skipif(os.name != "posix", reason="POSIX directory flush")
+def test_text_and_binary_writes_sync_contents_then_the_directory(tmp_path, monkeypatch):
+    target = tmp_path / "value"
+    synced = []
+    fsync = os.fsync
+
+    def observed(descriptor):
+        synced.append(stat.S_ISDIR(os.fstat(descriptor).st_mode))
+        fsync(descriptor)
+
+    monkeypatch.setattr(os, "fsync", observed)
+    AtomicFile(target, mode=0o640).write_text("first α")
+    assert target.read_text() == "first α"
+    assert stat.S_IMODE(target.stat().st_mode) == 0o640
+    AtomicFile(target).write_bytes(b"second")
+    assert target.read_bytes() == b"second"
+    assert synced == [False, True, False, True]
+
+
+def test_failed_stream_preparation_keeps_the_old_file(tmp_path):
+    target = tmp_path / "value"
+    target.write_bytes(b"original")
+    with pytest.raises(ValueError, match="stream failed"):
+        with AtomicFile(target).staged() as output:
+            output.write(b"partial")
+            raise ValueError("stream failed")
+    assert target.read_bytes() == b"original"
+    assert set(tmp_path.iterdir()) == {target}
+
+
+@pytest.mark.skipif(os.name != "posix", reason="POSIX directory flush")
+def test_parent_sync_failure_is_reported_after_the_visible_replace(tmp_path, monkeypatch):
+    target = tmp_path / "value"
+    fsync = os.fsync
+
+    def failed_directory(descriptor):
+        if stat.S_ISDIR(os.fstat(descriptor).st_mode):
+            raise OSError("directory sync failed")
+        fsync(descriptor)
+
+    monkeypatch.setattr(os, "fsync", failed_directory)
+    with pytest.raises(OSError, match="directory sync failed"):
+        AtomicFile(target).write_bytes(b"visible but not attested durable")
+    assert target.read_bytes() == b"visible but not attested durable"
+
+
+def test_exclusive_publication_never_replaces_an_existing_destination(tmp_path):
+    target = tmp_path / "value"
+    ExclusiveAtomicFile(target).write_text("original")
+    with pytest.raises(FileExistsError):
+        ExclusiveAtomicFile(target).write_text("replacement")
+    assert target.read_text() == "original"
+    assert set(tmp_path.iterdir()) == {target}
+
+
+def test_windows_publication_and_existing_file_sync_use_writable_handles(tmp_path, monkeypatch):
+    # These file operations need no kernel32 process handles, so exercise the
+    # actual Windows implementation on either host without initializing them.
+    platform = object.__new__(Win32Platform)
+    monkeypatch.setattr(Platform, "current", classmethod(lambda cls: platform))
+    fsync = os.fsync
+
+    def writable_sync(descriptor):
+        os.write(descriptor, b"")
+        fsync(descriptor)
+
+    monkeypatch.setattr(os, "fsync", writable_sync)
+    target = tmp_path / "value"
+    AtomicFile(target).write_bytes(b"first")
+    staged = tmp_path / "prepared"
+    staged.write_bytes(b"second")
+    AtomicFile(target).replace(staged)
+    AtomicFile(target).sync()
+    assert target.read_bytes() == b"second"
+    # A Windows directory flush does not open the directory, including a path
+    # that does not exist; it must not ask os for POSIX-only flags.
+    platform.sync_directory(tmp_path / "absent")
diff --git a/tests/test_registry_revision.py b/tests/test_registry_revision.py
index edc74245ac..66226d079d 100644
--- a/tests/test_registry_revision.py
+++ b/tests/test_registry_revision.py
@@ -6,7 +6,7 @@
 
 from agent_comms.field_codec import MalformedRecord
 from agent_comms.registration import Registration
-from agent_comms.store_files import _atomic_write_text
+from agent_comms.store_files import AtomicFile
 from agent_comms.thread_status import StoppedThreadStatus
 
 
@@ -31,11 +31,11 @@ def test_malformed_replacement_never_serves_stale_success(tmp_path):
     registry.register(registry.new_thread(name='worker', tags=frozenset(), worktree=str(tmp_path)))
     saved = path.read_text()
     registry.require("worker")
-    _atomic_write_text(path, "{")
+    AtomicFile(path).write_text("{")
     for _ in range(2):
         with pytest.raises(MalformedRecord):
             registry.require("worker")
-    _atomic_write_text(path, saved)
+    AtomicFile(path).write_text(saved)
     assert registry.require("worker").name == "worker"
 
 
@@ -56,7 +56,7 @@ def test_a_new_revision_decodes_only_the_threads_that_changed(tmp_path):
     stopped.register(stopped.new_thread(name='second', tags=frozenset(), worktree=str(tmp_path)))
     stopped.unregister("second")
     raw["statuses"]["second"] = json.loads(stopped.store.path.read_text())["statuses"]["second"]
-    _atomic_write_text(path, json.dumps(raw, indent=2))
+    AtomicFile(path).write_text(json.dumps(raw, indent=2))
 
     after = reader.all_threads()
     # Unchanged threads keep their decoded declarations and statuses.
@@ -80,12 +80,12 @@ def test_a_changed_revision_is_checked_as_a_whole(tmp_path):
     # A declaration without its status: the unchanged rows do not excuse it.
     raw = json.loads(path.read_text())
     del raw["statuses"]["second"]
-    _atomic_write_text(path, json.dumps(raw, indent=2))
+    AtomicFile(path).write_text(json.dumps(raw, indent=2))
     with pytest.raises(RelationViolationError):
         registry.require("first")
     # A changed row that does not decode is rejected too.
     raw = json.loads(path.read_text())
     raw["statuses"]["second"] = {"kind": "no-such-status"}
-    _atomic_write_text(path, json.dumps(raw, indent=2))
+    AtomicFile(path).write_text(json.dumps(raw, indent=2))
     with pytest.raises(RelationViolationError):
         registry.require("first")
diff --git a/tools/cutover/carry_bus_history.py b/tools/cutover/carry_bus_history.py
index 19ec5a9327..e22e44881d 100644
--- a/tools/cutover/carry_bus_history.py
+++ b/tools/cutover/carry_bus_history.py
@@ -55,7 +55,7 @@
 from agent_comms.registry_provenance import RegistryProvenance
 from agent_comms.registry_store import RegistryStore
 from agent_comms.response_acknowledgement import KeyedResponseAcknowledgement
-from agent_comms.store_files import SharedLock, _atomic_write_text, _store_lock
+from agent_comms.store_files import AtomicFile, SharedLock, _store_lock
 from agent_comms.typed_table import Column, DerivedTables, SharedFileTables, TypedTable
 from agent_comms.wake_policy import (
     MessageAudience,
@@ -149,7 +149,7 @@ def publish(
         payload = json.dumps(FieldCodec.encode(converted))
         if path.read_bytes() == payload.encode():
             return Carried(changed, rewritten=rewritten)
-        _atomic_write_text(path, payload)
+        AtomicFile(path).write_text(payload)
         return Carried(changed + len(converted), rewritten=(*rewritten, HistoryArchive.filename))
 
     def mapped_root(self, root: Path, original_root: Path) -> Path:
diff --git a/tools/cutover/carry_owner_requests.py b/tools/cutover/carry_owner_requests.py
index 386d24c179..4f583b76e1 100644
--- a/tools/cutover/carry_owner_requests.py
+++ b/tools/cutover/carry_owner_requests.py
@@ -29,7 +29,7 @@
     RequestDirective,
 )
 from agent_comms.registry_store import RegistryStore
-from agent_comms.store_files import _atomic_write_text, _store_lock
+from agent_comms.store_files import AtomicFile, _store_lock
 from agent_comms.thread_identity import RecordedThreadIncarnation
 from agent_comms.todos import OwnerRequest, Todo, TodoStore
 from agent_comms.typed_table import (
@@ -163,9 +163,7 @@ def carry(self, *, apply: bool) -> Carried:
                 for name in result.rewritten
             ]
             if resealed:
-                _atomic_write_text(
-                    HistoryArchive.path_in(self.root), json.dumps(FieldCodec.encode(tuple(rebound)))
-                )
+                AtomicFile(HistoryArchive.path_in(self.root)).write_text(json.dumps(FieldCodec.encode(tuple(rebound))))
                 rewritten.append(HistoryArchive.filename)
             detail: dict[str, int] = {}
             for _, result in results:
diff --git a/tools/cutover/carry_turn_owner.py b/tools/cutover/carry_turn_owner.py
index a1dd94f0b3..986d282109 100644
--- a/tools/cutover/carry_turn_owner.py
+++ b/tools/cutover/carry_turn_owner.py
@@ -18,7 +18,7 @@
 from agent_comms.field_codec import FieldCodec
 from agent_comms.installer.carry import Carried, Carry
 from agent_comms.registry_store import RegistryStore
-from agent_comms.store_files import _atomic_write_text
+from agent_comms.store_files import AtomicFile
 from agent_comms.thread_identity import TurnIdentity
 from agent_comms.turn_lease import ActiveTurn, StoredTurnRecord
 
@@ -47,6 +47,6 @@ def carry(self, *, apply: bool) -> Carried:
             carried = stored | {"owner": FieldCodec.encode(owner, ProcessIdentity)}
             FieldCodec.decode(ActiveTurn, carried)
             if apply:
-                _atomic_write_text(path, json.dumps(carried))
+                AtomicFile(path).write_text(json.dumps(carried))
             rewritten.append(str(path.relative_to(self.root)))
         return Carried(len(rewritten), rewritten=tuple(rewritten) if apply else ())