Own durable file publication in AtomicFile and Platform

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()

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)

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

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]

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
 

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)

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:

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:

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."""