From da8363979ebecca50b80dd9cb779ae1d8cb643b6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:49:45 -0400 Subject: [PATCH 1/2] Centralize durable file publication in AtomicFile --- src/agent_comms/diagnostics.py | 12 ++-- src/agent_comms/exporting.py | 62 +++++------------- src/agent_comms/goal_attempts.py | 27 ++------ src/agent_comms/goal_history.py | 36 +++-------- src/agent_comms/harness_install.py | 7 +- src/agent_comms/historical_views.py | 7 +- src/agent_comms/installer/backup.py | 33 +++------- src/agent_comms/locked_store.py | 75 ++++++++-------------- src/agent_comms/maintenance_barrier.py | 6 +- src/agent_comms/native_package.py | 15 ++--- src/agent_comms/native_session_files.py | 15 ++--- src/agent_comms/private_registry_guard.py | 43 +++++-------- src/agent_comms/restart_queue.py | 13 ++-- src/agent_comms/selected_write_plan.py | 20 ++---- src/agent_comms/store_files.py | 78 ++++++++++++++++------- src/agent_comms/thread_management.py | 10 +-- src/agent_comms/threads.py | 26 +++++--- tests/test_atomic_file.py | 61 ++++++++++++++++++ tests/test_goal_history.py | 18 +++--- tests/test_registry_revision.py | 12 ++-- tools/cutover/carry_bus_history.py | 4 +- tools/cutover/carry_owner_requests.py | 6 +- 22 files changed, 274 insertions(+), 312 deletions(-) create mode 100644 tests/test_atomic_file.py diff --git a/src/agent_comms/diagnostics.py b/src/agent_comms/diagnostics.py index ef91cf1817..8aa0b78f3a 100644 --- a/src/agent_comms/diagnostics.py +++ b/src/agent_comms/diagnostics.py @@ -14,9 +14,11 @@ from typing import TYPE_CHECKING, Literal from uuid import uuid4 +from agent_comms.private_path import fsync_directory + 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 @@ -276,10 +278,6 @@ 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) if os.name == "posix": - fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(fd) - finally: - os.close(fd) - _atomic_write_text(path, json.dumps(document, indent=2) + "\n", fsync_parent=True) + fsync_directory(root) + 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 95296734ee..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 @@ -19,9 +18,10 @@ 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"): @@ -444,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( @@ -529,21 +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 - if os.name != "nt": - directory = os.open(destination.parent, os.O_RDONLY) - try: - os.fsync(directory) - finally: - os.close(directory) diff --git a/src/agent_comms/goal_attempts.py b/src/agent_comms/goal_attempts.py index 7a93d00ab2..728c880246 100644 --- a/src/agent_comms/goal_attempts.py +++ b/src/agent_comms/goal_attempts.py @@ -29,6 +29,7 @@ from uuid import uuid4 from agent_comms.goal_turn_permit import GoalTurnPermit +from agent_comms.private_path import fsync_directory from .declared_family import DeclaredFamily from .field_codec import MalformedRecord @@ -62,6 +63,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: @@ -710,12 +712,12 @@ def initialize(cls, root: str | Path) -> GoalAttemptStore: conn.execute("PRAGMA foreign_keys=ON") with cls.write_transaction(conn): GoalLedgerTable.create_schema(conn) - cls._sync_paths(staging, 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) - cls._sync_directory(directory) + fsync_directory(directory) except (sqlite3.Error, OSError) as error: raise StorageUncertainError("Goal attempt initialization is uncertain.") from error finally: @@ -753,27 +755,8 @@ def database(cls, path: Path) -> sqlite3.Connection: except sqlite3.Error as error: raise StorageUncertainError("Goal attempt database unavailable.") from error - @classmethod - def _sync_paths(cls, path: Path, directory: Path) -> None: - # Windows' CRT _commit (used by os.fsync) needs a writable handle. - file_fd = os.open(path, os.O_RDWR if os.name == "nt" else os.O_RDONLY) - try: - os.fsync(file_fd) - finally: - os.close(file_fd) - cls._sync_directory(directory) - - @staticmethod - def _sync_directory(directory: Path) -> None: - if os.name == "posix": - dir_fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(dir_fd) - finally: - os.close(dir_fd) - def _sync(self) -> None: - self._sync_paths(self.path, 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 ddffceb111..ba46507c79 100644 --- a/src/agent_comms/goal_history.py +++ b/src/agent_comms/goal_history.py @@ -22,11 +22,17 @@ from agent_comms.field_codec import WireOmitDefault from agent_comms.owner_requests import NoRequestRevision, RequestAtRevision, RequestRevision -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,27 +175,7 @@ def _connect(self) -> sqlite3.Connection: raise GoalHistoryError("Goal history database is unavailable.") from error def _sync(self) -> None: - self._sync_file(self.path) - if os.name == "posix": - self._sync_directory(self.path.parent) - - @staticmethod - def _sync_file(path: Path) -> None: - # Windows' CRT _commit (used by os.fsync) rejects a read-only fd. - # Retain the existing read-only POSIX reopen semantics. - descriptor = os.open(path, os.O_RDWR if os.name == "nt" else os.O_RDONLY) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) - - @staticmethod - def _sync_directory(path: Path) -> None: - descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) + AtomicFile(self.path).sync() def _initialize(self) -> None: try: @@ -262,9 +248,7 @@ def _sync_visible_registry(self) -> None: targets.append(guard) try: for target in targets: - self._sync_file(target) - if os.name == "posix": - self._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/harness_install.py b/src/agent_comms/harness_install.py index 420cbe2353..fedb4a1bd8 100644 --- a/src/agent_comms/harness_install.py +++ b/src/agent_comms/harness_install.py @@ -16,6 +16,7 @@ from pathlib import Path from typing import TYPE_CHECKING, ClassVar +from agent_comms.private_path import fsync_directory from agent_comms_hooks import ClaudeHookProgram from .declared_family import DeclaredFamily @@ -34,11 +35,7 @@ def replace_link(link: Path, target: Path) -> None: try: temporary.symlink_to(target) os.replace(temporary, link) - directory = os.open(link.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) - try: - os.fsync(directory) - finally: - os.close(directory) + fsync_directory(link.parent) finally: temporary.unlink(missing_ok=True) 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 05ead906f3..e4d8be0822 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -26,8 +26,8 @@ from agent_comms.field_codec import FieldCodec from agent_comms.installer.interpreter_operation import HoldTranscriptsProbe, 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.private_path import FileObservation, FileRevision, MissingFile, fsync_directory +from agent_comms.store_files import AtomicFile class UpgradeRefused(CoordinationError): @@ -205,11 +205,7 @@ def files(self) -> dict[str, str]: def retire(self) -> Path: archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}') self.path.rename(archived) - directory = os.open(archived.parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(directory) - finally: - os.close(directory) + fsync_directory(archived.parent) print(f'Restored data backup kept at {archived}', flush=True) return archived @@ -222,7 +218,7 @@ def restore_route(self) -> None: return saved = self.path / 'active-route.json' if saved.exists(): - _atomic_write_text(self.route, saved.read_text(), fsync_parent=True) + AtomicFile(self.route).write_text(saved.read_text()) else: os.unlink(self.route.name, dir_fd=directory) os.fsync(directory) @@ -269,14 +265,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), fsync_parent=True) + 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), - fsync_parent=True) + AtomicFile(path / 'files.json').write_text(json.dumps(FieldCodec.encode(contents), indent=1)) return backup def retain(self, spill_dir: Path) -> tuple[Path, ...]: @@ -348,11 +343,7 @@ def capture(self, files: tuple[Path, ...]) -> dict[str, str]: with target.open('rb') as ready: os.fsync(ready.fileno()) contents[name] = self.digest(target) - directory = os.open(target.parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(directory) - finally: - os.close(directory) + fsync_directory(target.parent) return contents def file_original(self, name: str, saved: dict[str, str]) -> bool: @@ -384,20 +375,14 @@ def restore(self) -> None: stage = Path(temporary) try: shutil.copy2(self.data_root / name, stage) - with stage.open('rb') as ready: - os.fsync(ready.fileno()) - os.replace(stage, target) + AtomicFile(target).replace(stage) finally: stage.unlink(missing_ok=True) else: target.unlink(missing_ok=True) for companion in SQLiteFile.files(target)[1:]: companion.unlink(missing_ok=True) - directory = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(directory) - finally: - os.close(directory) + fsync_directory(self.root) self.restore_route() if not self.stores_original(): raise UpgradeRefused('Selected stores differ after backup restoration') diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index a99805b2bf..2107edead5 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -21,8 +21,15 @@ from .coordination_errors import CoordinationReadUnavailable from .field_codec import FieldCodec, MalformedRecord +from .private_path import PrivateFileRole, fsync_directory 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") @@ -85,58 +92,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) - directory_fd = None - 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()) - if os.name == "posix": - directory_fd = os.open( - self.path.parent, os.O_RDONLY | os.O_DIRECTORY - ) + 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 - if directory_fd is not None: - os.fsync(directory_fd) - # 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 + fsync_directory(self.path.parent) else: - self.path.unlink() - if directory_fd is not None: - os.fsync(directory_fd) - raise + self.path.unlink(missing_ok=True) + fsync_directory(self.path.parent) + raise finally: - if directory_fd is not None: - os.close(directory_fd) - 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 767c8b4663..12d0ac506c 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, ) @@ -407,8 +407,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, fsync_parent=True) - _atomic_write_text(self.state_path, state_text, fsync_parent=True) + 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 79938518ab..591b61ebcb 100644 --- a/src/agent_comms/native_package.py +++ b/src/agent_comms/native_package.py @@ -22,7 +22,9 @@ from agent_comms.field_codec import FieldCodec, MalformedRecord, PathText, Sha256Text, WireName from agent_comms.native_build_inputs import INPUTS_PREFIX, NativeBuildInputs -from agent_comms.private_path import PrivateDirectoryRole, TrustedAncestorRole +from agent_comms.private_path import PrivateDirectoryRole, TrustedAncestorRole, fsync_directory + +from .store_files import AtomicFile TREE_PREFIX = "# agent-comms-native-tree-v1 " MAX_ENTRIES = 30_000 @@ -310,8 +312,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") @@ -356,15 +357,11 @@ 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}: - descriptor = os.open(parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) + fsync_directory(parent) except OSError as error: raise NativePackageError("Native extension source could not be installed") from error finally: diff --git a/src/agent_comms/native_session_files.py b/src/agent_comms/native_session_files.py index 940cad4730..fac9a7ff84 100644 --- a/src/agent_comms/native_session_files.py +++ b/src/agent_comms/native_session_files.py @@ -8,14 +8,15 @@ from pathlib import Path from typing import Self -from .errors import RelationViolationError -from .native_failures import NativePiUnavailable -from agent_comms.locked_store import DurableFile, PatternFile, SQLiteFile 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, fsync_directory +from .errors import RelationViolationError +from .native_failures import NativePiUnavailable + class SessionFiles: """Declared session members share reference checks and deletion lifecycle.""" @@ -94,11 +95,7 @@ def delete(self) -> int: freed += path.lstat().st_size path.unlink() if self.session.parent.is_dir(): - directory = os.open(self.session.parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(directory) - finally: - os.close(directory) + fsync_directory(self.session.parent) return freed diff --git a/src/agent_comms/private_registry_guard.py b/src/agent_comms/private_registry_guard.py index ff5bf8733c..7db93224ca 100644 --- a/src/agent_comms/private_registry_guard.py +++ b/src/agent_comms/private_registry_guard.py @@ -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, fsync_parent=True) + 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, fsync_parent=True) + store_files.AtomicFile(registry_path).write_text(text) self.commit(sequence, slot, digest) def register_restored(self, threads: Iterable[Thread]) -> None: @@ -470,33 +470,21 @@ def create_pending(self) -> None: os.fsync(registry_fd) finally: os.close(registry_fd) - directory_fd = os.open( - self.registry_path.parent, os.O_RDONLY | os.O_DIRECTORY + fsync_directory(self.registry_path.parent) + fd = os.open( + self.path, + os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, ) try: - os.fsync(directory_fd) # baseline registry name and existing root - 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 - ): - _reject("initial slots were incomplete") - os.fsync(fd) - os.fsync(directory_fd) # guard name is durable before native preparation returns - self._write(fd, 1, _record(self.root_id, 1, _PENDING, digest)) - finally: - os.close(fd) + if os.pwrite(fd, _record(self.root_id, 0, _COMMITTED, digest) * 2, 0) != 2 * _SIZE: + _reject("initial slots were incomplete") + os.fsync(fd) + fsync_directory(self.registry_path.parent) + self._write(fd, 1, _record(self.root_id, 1, _PENDING, digest)) finally: - os.close(directory_fd) + os.close(fd) + def commit_initial(self) -> None: fd = self._open() @@ -529,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), fsync_parent=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 8a61fef9f5..9d0c9e3d6e 100644 --- a/src/agent_comms/restart_queue.py +++ b/src/agent_comms/restart_queue.py @@ -19,16 +19,17 @@ from pathlib import Path from typing import TYPE_CHECKING, Annotated +from agent_comms.private_path import fsync_directory from .child_process import OutlivesLauncherLifetime, ProcessIdentity, SpawnedProcess from .comms import Comms, wire 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 +38,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 @@ -241,15 +242,11 @@ def move(self, path: Path, record: QueuedRestart) -> None: raise FileExistsError(destination) os.rename(path, destination) for parent in (path.parent, destination.parent): - descriptor = os.open(parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) + fsync_directory(parent) 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), fsync_parent=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_write_plan.py b/src/agent_comms/selected_write_plan.py index 5e6d2ee2b3..ea4eaf62b6 100644 --- a/src/agent_comms/selected_write_plan.py +++ b/src/agent_comms/selected_write_plan.py @@ -12,14 +12,14 @@ 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 .coordination_database import CoordinationStore from .locked_store import DurableFile, PatternFile +from .store_files import AtomicFile if TYPE_CHECKING: from .acp_extension import SelectedWriteAcceptedUpdate @@ -268,16 +268,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) - fsync_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 042eb4cb50..909e267a6b 100644 --- a/src/agent_comms/store_files.py +++ b/src/agent_comms/store_files.py @@ -29,7 +29,7 @@ from .child_process import Platform from .declared_family import DeclaredFamily from .errors import RelationViolationError -from .private_path import FileIdentity, PrivateFileRole +from .private_path import FileIdentity, PrivateFileRole, fsync_directory with suppress(ModuleNotFoundError): # Windows; Win32Platform never asks a mode for its flock. import fcntl @@ -432,28 +432,60 @@ def _async_store_lock( ) -def _atomic_write_text( - path: Path, text: str, *, fsync_parent: bool = False, - mode: int = PrivateFileRole.permissions, -) -> None: - """Replace a snapshot; private guarded writes also 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) - if fsync_parent: - directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(directory_fd) - finally: - os.close(directory_fd) - 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.""" + with staged.open("rb") as source: + os.fsync(source.fileno()) + self._publish(staged) + + def _publish(self, staged: Path) -> None: + self._install(staged) + fsync_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.""" + # Windows' CRT needs a writable handle; these files are owned writes. + with self.path.open("rb+") as source: + os.fsync(source.fileno()) + fsync_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 7e1d4aec59..ad643618ac 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..1e21825326 --- /dev/null +++ b/tests/test_atomic_file.py @@ -0,0 +1,61 @@ +"""File contents and directory names become durable through the same writer.""" + +import os +import stat + +import pytest + +from agent_comms.store_files import AtomicFile, ExclusiveAtomicFile + + +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} + + +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} diff --git a/tests/test_goal_history.py b/tests/test_goal_history.py index 0b74baa703..3326c09830 100644 --- a/tests/test_goal_history.py +++ b/tests/test_goal_history.py @@ -9,7 +9,6 @@ import agent_comms.store_files as store_files from agent_comms.comms import Comms -from conftest import installed_comms from agent_comms.goal_actions import ( ClearGoalAction, CompletedGoalAction, @@ -18,10 +17,11 @@ SetGoalAction, ) from agent_comms.goal_history import GoalHistoryError, GoalHistoryStore -from agent_comms.goals import NoGoal from agent_comms.goal_states import ActiveGoal -from agent_comms.tools import ToolRequest +from agent_comms.goals import NoGoal from agent_comms.tool_caller import ThreadCaller +from agent_comms.tools import ToolRequest +from conftest import installed_comms def _wire(tmp_path) -> Comms: @@ -34,7 +34,7 @@ def test_history_file_fsync_reopens_with_supported_platform_flags(tmp_path): path = tmp_path / "history.sqlite3" path.write_bytes(b"journal bytes") # Windows' os.fsync delegates to the CRT, which rejects O_RDONLY fds. - GoalHistoryStore._sync_file(path) + store_files.AtomicFile(path).sync() assert path.read_bytes() == b"journal bytes" @@ -105,15 +105,15 @@ def test_existing_goal_seeds_only_current_observed_baseline(tmp_path): def test_crash_before_registry_write_does_not_expose_history_intent(tmp_path, monkeypatch): comms = _wire(tmp_path) current = comms.goals.update_goal("worker", SetGoalAction(text="Keep current")) - original_write = store_files._atomic_write_text + original_replace = store_files.os.replace - def failed_registry_write(path, text, *, fsync_parent=False): - if path == comms.registry.store.path: + def failed_registry_write(source, target, *args, **kwargs): + if target == comms.registry.store.path: raise OSError("simulated registry write failure") - return original_write(path, text, fsync_parent=fsync_parent) + return original_replace(source, target, *args, **kwargs) with monkeypatch.context() as patch: - patch.setattr(store_files, "_atomic_write_text", failed_registry_write) + patch.setattr(store_files.os, "replace", failed_registry_write) with pytest.raises(OSError): comms.goals.update_goal( "worker", PausedGoalAction(expect=GoalPrecondition(goal_id=current.id)) diff --git a/tests/test_registry_revision.py b/tests/test_registry_revision.py index 58959eaddd..d90726c31d 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 @@ -16,11 +16,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" @@ -41,7 +41,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. @@ -65,12 +65,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: From 5f7d21cc3fffebe406ff4c7626a28c435665e5d0 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:53:21 -0400 Subject: [PATCH 2/2] Own file and directory sync in the Platform family --- src/agent_comms/child_process.py | 31 ++++++++++++++++++++++- src/agent_comms/diagnostics.py | 4 +-- src/agent_comms/fresh_private_session.py | 5 ++-- src/agent_comms/goal_attempts.py | 4 +-- src/agent_comms/harness_install.py | 4 +-- src/agent_comms/installer/backup.py | 9 ++++--- src/agent_comms/locked_store.py | 7 ++--- src/agent_comms/native_package.py | 5 ++-- src/agent_comms/native_session_files.py | 9 ++++--- src/agent_comms/private_path.py | 9 ------- src/agent_comms/private_registry_guard.py | 8 +++--- src/agent_comms/restart_queue.py | 5 ++-- src/agent_comms/selected_tool_broker.py | 9 ++++--- src/agent_comms/selected_write_plan.py | 7 ++--- src/agent_comms/store_files.py | 14 +++++----- tests/test_atomic_file.py | 22 ++++++++++++++++ 16 files changed, 99 insertions(+), 53 deletions(-) diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py index 0ce62d27d0..93368dea8e 100644 --- a/src/agent_comms/child_process.py +++ b/src/agent_comms/child_process.py @@ -36,7 +36,7 @@ from dataclasses import dataclass from functools import partial from pathlib import Path -from typing import TYPE_CHECKING, Any, ClassVar, NoReturn +from typing import TYPE_CHECKING, Any, BinaryIO, ClassVar, NoReturn from .declared_family import DeclaredFamily from .errors import RelationViolationError @@ -413,6 +413,18 @@ class Platform(DeclaredFamily, affix="Platform"): def current(cls) -> Platform: return cls.member_named(sys.platform)() + @abstractmethod + def fsync_directory(self, path: Path) -> None: + """Finish publication using this platform's directory durability contract.""" + + @abstractmethod + def open_for_sync(self, path: Path) -> BinaryIO: + """Open an existing file with the access required by the native flush.""" + + def fsync_file(self, path: Path) -> None: + with self.open_for_sync(path) as source: + os.fsync(source.fileno()) + @abstractmethod def try_store_lock(self, descriptor: int, *, mode: LockMode) -> None: """One nonblocking native attempt; busy is BlockingIOError.""" @@ -481,6 +493,16 @@ def signal_group(self, leader: ProcessIdentity, signum: int) -> None: class PosixPlatform(Platform): + def fsync_directory(self, path: Path) -> None: + descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + def open_for_sync(self, path: Path) -> BinaryIO: + return path.open("rb") + def acquire_store_lock(self, descriptor: int, *, mode: LockMode, blocking: bool) -> None: import fcntl @@ -685,6 +707,13 @@ class Win32Platform(Platform): store_lock_interval = 0.01 + def fsync_directory(self, path: Path) -> None: + """NTFS journals the rename; there is no POSIX directory flush.""" + + def open_for_sync(self, path: Path) -> BinaryIO: + # The Windows CRT's fsync requires a writable descriptor. + return path.open("rb+") + def try_store_lock(self, descriptor: int, *, mode: LockMode) -> None: import errno import msvcrt diff --git a/src/agent_comms/diagnostics.py b/src/agent_comms/diagnostics.py index 8aa0b78f3a..66d4903121 100644 --- a/src/agent_comms/diagnostics.py +++ b/src/agent_comms/diagnostics.py @@ -14,8 +14,8 @@ from typing import TYPE_CHECKING, Literal from uuid import uuid4 -from agent_comms.private_path import fsync_directory +from .child_process import Platform from .field_codec import FieldCodec, Id128Text, MalformedRecord from .locked_store import DurableFile, PatternFile from .store_files import AtomicFile @@ -278,6 +278,6 @@ 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) if os.name == "posix": - fsync_directory(root) + Platform.current().fsync_directory(root) AtomicFile(path).write_text(json.dumps(document, indent=2) + "\n") return path.resolve() diff --git a/src/agent_comms/fresh_private_session.py b/src/agent_comms/fresh_private_session.py index 13fe73e5f0..f636402ef9 100644 --- a/src/agent_comms/fresh_private_session.py +++ b/src/agent_comms/fresh_private_session.py @@ -8,13 +8,14 @@ from pathlib import Path from uuid import uuid4 +from agent_comms.child_process import Platform from .field_codec import RecordedTimestamp from .native_entries import SessionEntry from .native_failures import NativePiUnavailable from agent_comms.native_session_files import NativeSessionFiles from .native_session_files import PrivateSession, SavedSessionFile from .native_session_reopen import NativeSessionIdentity -from .private_path import NativeSessionFileRole, fsync_directory +from .private_path import NativeSessionFileRole def create_fresh_private_session( @@ -47,7 +48,7 @@ def create_fresh_private_session( os.fsync(descriptor) finally: os.close(descriptor) - fsync_directory(session_dir) + Platform.current().fsync_directory(session_dir) return NativeSessionIdentity(session_id, str(path)) except OSError as error: # A failed sync can leave a committed file. Preserve it, never infer success. diff --git a/src/agent_comms/goal_attempts.py b/src/agent_comms/goal_attempts.py index 728c880246..fd4c53b000 100644 --- a/src/agent_comms/goal_attempts.py +++ b/src/agent_comms/goal_attempts.py @@ -28,8 +28,8 @@ from typing import TYPE_CHECKING, Annotated, ClassVar, TypeVar from uuid import uuid4 +from agent_comms.child_process import Platform from agent_comms.goal_turn_permit import GoalTurnPermit -from agent_comms.private_path import fsync_directory from .declared_family import DeclaredFamily from .field_codec import MalformedRecord @@ -717,7 +717,7 @@ def initialize(cls, root: str | Path) -> GoalAttemptStore: # The staging descriptor is closed before its inode becomes shared. with suppress(FileExistsError): os.link(staging, path, follow_symlinks=False) - fsync_directory(directory) + Platform.current().fsync_directory(directory) except (sqlite3.Error, OSError) as error: raise StorageUncertainError("Goal attempt initialization is uncertain.") from error finally: diff --git a/src/agent_comms/harness_install.py b/src/agent_comms/harness_install.py index fedb4a1bd8..28e88178a3 100644 --- a/src/agent_comms/harness_install.py +++ b/src/agent_comms/harness_install.py @@ -16,9 +16,9 @@ from pathlib import Path from typing import TYPE_CHECKING, ClassVar -from agent_comms.private_path import fsync_directory from agent_comms_hooks import ClaudeHookProgram +from .child_process import Platform from .declared_family import DeclaredFamily from .thread_execution import ExternalThreadExecution @@ -35,7 +35,7 @@ def replace_link(link: Path, target: Path) -> None: try: temporary.symlink_to(target) os.replace(temporary, link) - fsync_directory(link.parent) + Platform.current().fsync_directory(link.parent) finally: temporary.unlink(missing_ok=True) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index e4d8be0822..0e744535b4 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -21,12 +21,13 @@ from typing import ClassVar, Iterator, Self from agent_comms.active_route import ActiveRoute, _route_directory, route_directory +from agent_comms.child_process import Platform from agent_comms.coordination_errors import CoordinationError from agent_comms.external_transcript import ExternalTranscript from agent_comms.field_codec import FieldCodec from agent_comms.installer.interpreter_operation import HoldTranscriptsProbe, RouteSelection from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile -from agent_comms.private_path import FileObservation, FileRevision, MissingFile, fsync_directory +from agent_comms.private_path import FileObservation, FileRevision, MissingFile from agent_comms.store_files import AtomicFile @@ -205,7 +206,7 @@ def files(self) -> dict[str, str]: def retire(self) -> Path: archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}') self.path.rename(archived) - fsync_directory(archived.parent) + Platform.current().fsync_directory(archived.parent) print(f'Restored data backup kept at {archived}', flush=True) return archived @@ -343,7 +344,7 @@ def capture(self, files: tuple[Path, ...]) -> dict[str, str]: with target.open('rb') as ready: os.fsync(ready.fileno()) contents[name] = self.digest(target) - fsync_directory(target.parent) + Platform.current().fsync_directory(target.parent) return contents def file_original(self, name: str, saved: dict[str, str]) -> bool: @@ -382,7 +383,7 @@ def restore(self) -> None: target.unlink(missing_ok=True) for companion in SQLiteFile.files(target)[1:]: companion.unlink(missing_ok=True) - fsync_directory(self.root) + Platform.current().fsync_directory(self.root) self.restore_route() if not self.stores_original(): raise UpgradeRefused('Selected stores differ after backup restoration') diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index 2107edead5..2f89169006 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -19,9 +19,10 @@ from pkgutil import walk_packages from typing import Any, ClassVar, Generic, Self, TypeVar +from .child_process import Platform from .coordination_errors import CoordinationReadUnavailable from .field_codec import FieldCodec, MalformedRecord -from .private_path import PrivateFileRole, fsync_directory +from .private_path import PrivateFileRole from .store_files import ( AtomicFile, SharedLock, @@ -108,10 +109,10 @@ def _write_unlocked(self, text: str) -> None: if backup.exists(): if not self.path.exists() or not self.path.samefile(backup): os.replace(backup, self.path) - fsync_directory(self.path.parent) + Platform.current().fsync_directory(self.path.parent) else: self.path.unlink(missing_ok=True) - fsync_directory(self.path.parent) + Platform.current().fsync_directory(self.path.parent) raise finally: # A failed restore leaves the last original inode for recovery. diff --git a/src/agent_comms/native_package.py b/src/agent_comms/native_package.py index 591b61ebcb..1c3c5663ba 100644 --- a/src/agent_comms/native_package.py +++ b/src/agent_comms/native_package.py @@ -20,9 +20,10 @@ from pathlib import Path from typing import Annotated, Literal +from agent_comms.child_process import Platform from agent_comms.field_codec import FieldCodec, MalformedRecord, PathText, Sha256Text, WireName from agent_comms.native_build_inputs import INPUTS_PREFIX, NativeBuildInputs -from agent_comms.private_path import PrivateDirectoryRole, TrustedAncestorRole, fsync_directory +from agent_comms.private_path import PrivateDirectoryRole, TrustedAncestorRole from .store_files import AtomicFile @@ -361,7 +362,7 @@ def install_declared_sources(root: Path, snapshot_root: Path) -> None: for source in sources: source.verify_installed() for parent in {source.destination.parent for source in sources}: - fsync_directory(parent) + Platform.current().fsync_directory(parent) except OSError as error: raise NativePackageError("Native extension source could not be installed") from error finally: diff --git a/src/agent_comms/native_session_files.py b/src/agent_comms/native_session_files.py index fac9a7ff84..f056b44f9f 100644 --- a/src/agent_comms/native_session_files.py +++ b/src/agent_comms/native_session_files.py @@ -8,11 +8,12 @@ from pathlib import Path from typing import Self +from agent_comms.child_process import Platform from agent_comms.field_codec import Id128Text 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, fsync_directory +from agent_comms.private_path import PrivateDirectoryRole, TrustedAncestorRole from .errors import RelationViolationError from .native_failures import NativePiUnavailable @@ -95,7 +96,7 @@ def delete(self) -> int: freed += path.lstat().st_size path.unlink() if self.session.parent.is_dir(): - fsync_directory(self.session.parent) + Platform.current().fsync_directory(self.session.parent) return freed @@ -172,11 +173,11 @@ def prepare_private_directory(cls, directory: Path) -> None: for path in reversed(missing): path.mkdir(mode=0o700, exist_ok=True) cls.require_private_directory(path) - fsync_directory(path.parent) + Platform.current().fsync_directory(path.parent) cls.require_private_directory(directory) parent = directory.parent while True: - fsync_directory(parent) + Platform.current().fsync_directory(parent) if parent == parent.parent: break try: diff --git a/src/agent_comms/private_path.py b/src/agent_comms/private_path.py index 3381337db4..e14127e3f0 100644 --- a/src/agent_comms/private_path.py +++ b/src/agent_comms/private_path.py @@ -22,15 +22,6 @@ from .field_codec import NullMember, RecordMember, SubclassUnion -def fsync_directory(path: Path) -> None: - """Make a directory's entries durable without following a final symlink.""" - descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) - try: - os.fsync(descriptor) - finally: - os.close(descriptor) - - @dataclass(frozen=True, slots=True) class FileIdentity: device: int diff --git a/src/agent_comms/private_registry_guard.py b/src/agent_comms/private_registry_guard.py index 7db93224ca..8fada2fd64 100644 --- a/src/agent_comms/private_registry_guard.py +++ b/src/agent_comms/private_registry_guard.py @@ -28,6 +28,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Annotated, NoReturn +from .child_process import Platform from . import store_files from .coordination_database import CoordinationStore from .coordinator import Coordination @@ -35,7 +36,6 @@ from .field_codec import FieldCodec, Id128Text, MalformedRecord, WireName from .locked_store import DurableFile, WireWakeFile from .native_runtime_input import NativeRuntimeInput, NativeRuntimeTable -from .private_path import fsync_directory if TYPE_CHECKING: from .acp_extension import CursorAdvancedUpdate, CursorScope @@ -470,7 +470,7 @@ def create_pending(self) -> None: os.fsync(registry_fd) finally: os.close(registry_fd) - fsync_directory(self.registry_path.parent) + Platform.current().fsync_directory(self.registry_path.parent) fd = os.open( self.path, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, @@ -480,7 +480,7 @@ def create_pending(self) -> None: if os.pwrite(fd, _record(self.root_id, 0, _COMMITTED, digest) * 2, 0) != 2 * _SIZE: _reject("initial slots were incomplete") os.fsync(fd) - fsync_directory(self.registry_path.parent) + Platform.current().fsync_directory(self.registry_path.parent) self._write(fd, 1, _record(self.root_id, 1, _PENDING, digest)) finally: os.close(fd) @@ -525,7 +525,7 @@ def complete(cls, root: Path) -> None: intent = cls.path_in(root) replace(FieldCodec.decode(cls, json.loads(intent.read_text())), completed=True).write(root) intent.unlink() - fsync_directory(root) + Platform.current().fsync_directory(root) def _require_no_private_owner_rename(root: Path) -> None: diff --git a/src/agent_comms/restart_queue.py b/src/agent_comms/restart_queue.py index 9d0c9e3d6e..fff9f63f07 100644 --- a/src/agent_comms/restart_queue.py +++ b/src/agent_comms/restart_queue.py @@ -19,9 +19,8 @@ from pathlib import Path from typing import TYPE_CHECKING, Annotated -from agent_comms.private_path import fsync_directory -from .child_process import OutlivesLauncherLifetime, ProcessIdentity, SpawnedProcess +from .child_process import OutlivesLauncherLifetime, Platform, ProcessIdentity, SpawnedProcess from .comms import Comms, wire from .coordination_errors import CoordinationError from .coordinator import Coordination @@ -242,7 +241,7 @@ def move(self, path: Path, record: QueuedRestart) -> None: raise FileExistsError(destination) os.rename(path, destination) for parent in (path.parent, destination.parent): - fsync_directory(parent) + Platform.current().fsync_directory(parent) def save(self, record: QueuedRestart) -> None: path = self.active_path(record.selection.identity.incarnation.require_recorded()) diff --git a/src/agent_comms/selected_tool_broker.py b/src/agent_comms/selected_tool_broker.py index 7b0b4cf3d8..1e6e02acde 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 @@ -40,7 +41,7 @@ ToolTransport, UnboundToolTransport, ) -from .private_path import PrivateDirectoryRole, PrivateFileRole, fsync_directory +from .private_path import PrivateDirectoryRole, PrivateFileRole from .selected_actions import SelectedAction from .selected_claim import publish_selected_resource_claim, write_selected_claimed_file from .turn_context import InstructionFile @@ -194,7 +195,7 @@ def consume_selected_slot(directory: Path, input_id: str, call_id: str) -> None: info = ledger.lstat() if PrivateDirectoryRole.violation(info) is not None: raise SelectedToolDenied("Selected tool ledger is not private") - fsync_directory(directory) + Platform.current().fsync_directory(directory) fd = os.open(ledger / input_id, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) except FileExistsError as error: raise SelectedToolDenied( @@ -207,7 +208,7 @@ def consume_selected_slot(directory: Path, input_id: str, call_id: str) -> None: stream.write((call_id + "\n").encode("ascii")) stream.flush() os.fsync(stream.fileno()) - fsync_directory(ledger) + Platform.current().fsync_directory(ledger) except OSError as error: # Do NOT remove the visible slot; a failed fsync or a crash may have # committed it. No automatic re-append, retry, or write follows. @@ -223,7 +224,7 @@ def record_selected_terminal(directory: Path, input_id: str, call_id: str) -> No output.write((call_id + "\n").encode("ascii")) output.flush() os.fsync(output.fileno()) - fsync_directory(result.parent) + Platform.current().fsync_directory(result.parent) except OSError as error: raise SelectedToolDenied("Selected tool terminal acknowledgement UNKNOWN; no retry") from error diff --git a/src/agent_comms/selected_write_plan.py b/src/agent_comms/selected_write_plan.py index ea4eaf62b6..3929a25499 100644 --- a/src/agent_comms/selected_write_plan.py +++ b/src/agent_comms/selected_write_plan.py @@ -17,6 +17,7 @@ from pathlib import Path 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 @@ -34,7 +35,7 @@ from .envelope_claim_transitions import ExistingFileClaim from .field_codec import FieldCodec, Id128Text, MalformedRecord, WireName from .message_reference import MessageReference -from .private_path import PrivateDirectoryRole, PrivateFileRole, fsync_directory +from .private_path import PrivateDirectoryRole, PrivateFileRole from .thread_identity import ClaimedIdentity, RecordedThreadIncarnation from .threads import Thread @@ -128,7 +129,7 @@ def _prepare_dir(self) -> None: # Also on an existing directory: a previous mkdir may have survived # while its parent fsync failed. Do not acknowledge a later intent # until the directory entry itself is known durable. - fsync_directory(self.comms.root) + Platform.current().fsync_directory(self.comms.root) def _validate_dir(self) -> bool: try: @@ -232,7 +233,7 @@ def submit( stream.write(data) stream.flush() os.fsync(stream.fileno()) - fsync_directory(self.directory) + Platform.current().fsync_directory(self.directory) except BaseException: # The visible file may have committed despite the error. Keep # it as an UNKNOWN barrier; never unlink/retry the operation. diff --git a/src/agent_comms/store_files.py b/src/agent_comms/store_files.py index 909e267a6b..6f42bdc15f 100644 --- a/src/agent_comms/store_files.py +++ b/src/agent_comms/store_files.py @@ -29,7 +29,7 @@ from .child_process import Platform from .declared_family import DeclaredFamily from .errors import RelationViolationError -from .private_path import FileIdentity, PrivateFileRole, fsync_directory +from .private_path import FileIdentity, PrivateFileRole with suppress(ModuleNotFoundError): # Windows; Win32Platform never asks a mode for its flock. import fcntl @@ -462,23 +462,21 @@ def write_bytes(self, data: bytes) -> None: def replace(self, staged: Path) -> None: """Publish an independently prepared file after syncing its data.""" - with staged.open("rb") as source: - os.fsync(source.fileno()) + Platform.current().fsync_file(staged) self._publish(staged) def _publish(self, staged: Path) -> None: self._install(staged) - fsync_directory(self.path.parent) + Platform.current().fsync_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.""" - # Windows' CRT needs a writable handle; these files are owned writes. - with self.path.open("rb+") as source: - os.fsync(source.fileno()) - fsync_directory(self.path.parent) + platform = Platform.current() + platform.fsync_file(self.path) + platform.fsync_directory(self.path.parent) class ExclusiveAtomicFile(AtomicFile): diff --git a/tests/test_atomic_file.py b/tests/test_atomic_file.py index 1e21825326..fc6b452480 100644 --- a/tests/test_atomic_file.py +++ b/tests/test_atomic_file.py @@ -5,9 +5,11 @@ 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 = [] @@ -37,6 +39,7 @@ def test_failed_stream_preparation_keeps_the_old_file(tmp_path): 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 @@ -59,3 +62,22 @@ def test_exclusive_publication_never_replaces_an_existing_destination(tmp_path): 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)) + 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() + with platform.open_for_sync(target) as source: + assert source.writable() + assert source.read() == 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.fsync_directory(tmp_path / "absent")