From 30fea4b9987df253314de0254e2c903512b814a8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:29:13 -0400 Subject: [PATCH 1/2] P264: rebind checked bus copies without rebuilding their unchanged index --- src/agent_comms/checkpoint_seals.py | 14 ++++++- src/agent_comms/private_bus_checkpoint.py | 49 ++++++++++++++--------- tests/test_declared_family.py | 2 +- 3 files changed, 42 insertions(+), 23 deletions(-) diff --git a/src/agent_comms/checkpoint_seals.py b/src/agent_comms/checkpoint_seals.py index 30327b35a3..6616a76bec 100644 --- a/src/agent_comms/checkpoint_seals.py +++ b/src/agent_comms/checkpoint_seals.py @@ -10,7 +10,7 @@ from .declared_family import DeclaredFamily from .errors import CheckpointNeedsRepairError, RelationViolationError from .field_codec import Id128Text, Sha256Text -from .private_path import FileRevision +from .private_path import FileIdentity, FileRevision if TYPE_CHECKING: import os @@ -100,7 +100,15 @@ def recover( path: Path, saved: PrefixWitness, info: os.stat_result, - ) -> None: + ) -> PrefixWitness | None: + if saved.revision.identity != FileIdentity.from_stat(info): + self.require_rebuilt(saved, saved) + copied = FileRevision.from_stat(path.stat()) + if (copied.size, copied.mtime_ns) != (self.db_revision.size, self.db_revision.mtime_ns): + raise RelationViolationError("Private bus copied index content revision changed.") + if info.st_size == saved.offset: + return saved.rebind_source(bus, marker, db, path, info) + return None self.check_final(saved, path) def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None: @@ -137,6 +145,8 @@ def recover( or self.prior.root_id != marker.root_id ): raise RelationViolationError("Private bus checkpoint pending intent is inconsistent.") + if replace(self.prior, revision=self.expected.revision) == self.expected: + return saved.rebind_source(bus, marker, db, path, info) return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self) def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None: diff --git a/src/agent_comms/private_bus_checkpoint.py b/src/agent_comms/private_bus_checkpoint.py index 7687dd4816..d0fb511ccb 100644 --- a/src/agent_comms/private_bus_checkpoint.py +++ b/src/agent_comms/private_bus_checkpoint.py @@ -92,15 +92,28 @@ def require_marker(self, marker: WireMetadata) -> None: raise RelationViolationError("Checked bus prefix differs from its stored marker.") def require_source(self, marker: WireMetadata, info: os.stat_result, stream: BinaryIO) -> None: - """Reject changed bytes; a copied file needs rebuilding under the write lock.""" + """Validate the logical prefix independently of its physical file identity.""" if self.root_id != marker.root_id or info.st_size < self.offset: raise RelationViolationError("Private bus checkpoint root/size changed.") if _tail(stream, self.offset) != self.tail: raise RelationViolationError("Private bus checkpoint prefix tail changed.") if self.through_seq > marker.last_seq: raise RelationViolationError("Private bus checkpoint exceeds the durable sequence marker.") - if self.revision.identity != FileIdentity.from_stat(info): - raise CheckpointNeedsRepairError("Private bus source was relocated.") + + def rebind_source(self, bus: WireLog, marker: WireMetadata, db: sqlite3.Connection, + path: Path, info: os.stat_result) -> PrefixWitness: + """Publish the unchanged prefix's new revision under the exclusive bus lock.""" + rebound = replace(self, revision=replace(FileRevision.from_stat(info), size=self.offset)) + marker.seal.require_rebuilt(self, rebound) + marker.seal_with(PendingSeal.capture(self, rebound, path)) + bus.write_metadata_unlocked(marker) + with db: + CheckedPrefix.capture(rebound.root_id, rebound.revision, rebound.through_seq, + bytes.fromhex(rebound.digest), rebound.tail).upsert(db) + _directory_sync(path) + marker.seal_with(FinalSeal.capture(rebound, path)) + bus.write_metadata_unlocked(marker) + return rebound def require_append(self, marker: WireMetadata, info: os.stat_result, offset: int, sequence: int) -> None: @@ -1086,7 +1099,7 @@ def collect(offset, raw, record): != FileRevision.from_stat(info) ): raise RelationViolationError("Private bus changed during checkpoint installation.") - cls.capture(marker.root_id, info, through_seq, digest, + cls.capture(marker.root_id, FileRevision.from_stat(info), through_seq, digest, _tail(stream, info.st_size)).insert(db) _directory_sync(staged) with closing(_connect(staged, readonly=True)) as db: @@ -1135,20 +1148,20 @@ def witness(self) -> PrefixWitness: @classmethod def capture( - cls, root_id: str, info: os.stat_result, seq: int, digest: bytes, tail: str + cls, root_id: str, revision: FileRevision, seq: int, digest: bytes, tail: str ) -> CheckedPrefix: return cls( 1, 2, root_id, - info.st_dev, - info.st_ino, - info.st_size, + revision.identity.device, + revision.identity.inode, + revision.size, seq, digest.hex(), tail, - info.st_mtime_ns, - info.st_ctime_ns, + revision.mtime_ns, + revision.ctime_ns, ) @@ -1309,7 +1322,7 @@ def collect(offset, raw, record): seal.require_rebuilt(saved, expected) marker.seal_with(PendingSeal.capture(saved, expected, db_path)) bus.write_metadata_unlocked(marker) - CheckedPrefix.capture(expected.root_id, info, last_seq, digest, expected.tail).upsert( + CheckedPrefix.capture(expected.root_id, expected.revision, last_seq, digest, expected.tail).upsert( db ) _directory_sync(db_path) @@ -1323,9 +1336,9 @@ def _check_final_checkpoint_read_only(bus, marker, db, stream, path) -> PrefixWi saved = _saved(db) info = os.fstat(stream.fileno()) saved.require_source(marker, info, stream) - marker.seal.require_final(saved, path) if FileRevision.from_stat(info) != saved.revision: - raise CheckpointNeedsRepairError("Bus has rows the checkpoint does not cover yet.") + raise CheckpointNeedsRepairError("Bus revision requires exclusive checkpoint validation.") + marker.seal.require_final(saved, path) return saved @@ -1358,11 +1371,7 @@ def _verify_open_checkpoint_unlocked(bus, marker, db, stream, path) -> PrefixWit _directory_sync(bus.path) saved = _saved(db) info = os.fstat(stream.fileno()) - try: - saved.require_source(marker, info, stream) - except CheckpointNeedsRepairError: - # The copied index rows point at the old file; rebuild them. - return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, marker.seal) + saved.require_source(marker, info, stream) recovered = marker.seal.recover(bus, marker, db, path, saved, info) if recovered is not None: return recovered @@ -1430,7 +1439,7 @@ def collect( for offset, raw, record in additions: indexed_sequence = _index_row(db, offset, raw, record, indexed_sequence) CheckedPrefix.capture( - saved.root_id, info, last_seq, digest, expected.tail + saved.root_id, expected.revision, last_seq, digest, expected.tail ).upsert(db) _directory_sync(path) marker.seal_with(FinalSeal.capture(expected, path)) @@ -1502,7 +1511,7 @@ def append_private_bus_checkpoint_unlocked( with db: _index_row(db, offset, raw, record, saved.through_seq) CheckedPrefix.capture( - saved.root_id, info, sequence, digest, expected.tail + saved.root_id, expected.revision, sequence, digest, expected.tail ).upsert(db) _directory_sync(path) marker.seal_with(FinalSeal.capture(expected, path)) diff --git a/tests/test_declared_family.py b/tests/test_declared_family.py index 373a1a3e88..110b206679 100644 --- a/tests/test_declared_family.py +++ b/tests/test_declared_family.py @@ -151,7 +151,7 @@ def test_checkpoint_schema_query_stays_local_but_external_names_still_discover() for table in tables: table.create(db) checked_prefix = checkpoint.CheckedPrefix.capture( - '0' * 32, os.fstat(source.fileno()), 0, bytes(32), '0' * 64) + '0' * 32, checkpoint.FileRevision.from_file(source), 0, bytes(32), '0' * 64) checked_prefix.insert(db) assert checkpoint._saved(db) == checked_prefix.witness() assert 'agent_comms.acp' not in sys.modules From 787a99bfb20e2c0b87edafb5eb7cab989126f972 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:43:30 -0400 Subject: [PATCH 2/2] P264: recover concrete prefixes through chunked tail commits Incorporates chunked commit_tail and removal of the cold recovery bound from desktop PR #1976 (core-read-locks-2), head 1c7cd836697d79c2f0bd9b790455d2bd05636773. PrefixWitness owns tail continuation; the seal family owns each recovery cut. Delete nullable recovery and both full-prefix rescans. --- src/agent_comms/checkpoint_seals.py | 48 ++-- src/agent_comms/private_bus_checkpoint.py | 260 +++++++--------------- src/agent_comms/wire_log.py | 11 +- 3 files changed, 126 insertions(+), 193 deletions(-) diff --git a/src/agent_comms/checkpoint_seals.py b/src/agent_comms/checkpoint_seals.py index 6616a76bec..b20046bd45 100644 --- a/src/agent_comms/checkpoint_seals.py +++ b/src/agent_comms/checkpoint_seals.py @@ -66,7 +66,14 @@ def recover( path: Path, saved: PrefixWitness, info: os.stat_result, - ) -> PrefixWitness | None: ... + ) -> PrefixWitness: + return saved.index_tail(bus, marker, db, path, info) + + @abstractmethod + def recovery_end(self, info: os.stat_result) -> int: ... + + @abstractmethod + def chunk_complete(self, offset: int, size: int, rows: int) -> bool: ... def check_final(self, saved: PrefixWitness, db_path: Path) -> None: raise RelationViolationError("Private bus checkpoint index seal changed or is pending.") @@ -100,16 +107,24 @@ def recover( path: Path, saved: PrefixWitness, info: os.stat_result, - ) -> PrefixWitness | None: + ) -> PrefixWitness: if saved.revision.identity != FileIdentity.from_stat(info): self.require_rebuilt(saved, saved) copied = FileRevision.from_stat(path.stat()) if (copied.size, copied.mtime_ns) != (self.db_revision.size, self.db_revision.mtime_ns): raise RelationViolationError("Private bus copied index content revision changed.") - if info.st_size == saved.offset: - return saved.rebind_source(bus, marker, db, path, info) - return None - self.check_final(saved, path) + saved = saved.rebind_source(bus, marker, db, info) + else: + self.check_final(saved, path) + if saved.revision == FileRevision.from_stat(info): + return saved + return super().recover(bus, marker, db, path, saved, info) + + def recovery_end(self, info: os.stat_result) -> int: + return info.st_size + + def chunk_complete(self, offset: int, size: int, rows: int) -> bool: + return size >= 16 * 1024 * 1024 or rows >= 10_000 def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None: if self.witness != saved.seal(): @@ -137,17 +152,18 @@ def recover( saved: PrefixWitness, info: os.stat_result, ) -> PrefixWitness: - from .private_bus_checkpoint import _rebuild_checkpoint_unlocked - - if ( - saved.seal() not in (self.prior, self.expected) - or self.expected.revision != FileRevision.from_stat(info) - or self.prior.root_id != marker.root_id - ): + if (saved.seal() not in (self.prior, self.expected) + or not self.expected.revision.holds_cut( + FileRevision.from_stat(info), self.expected.revision.size) + or self.prior.root_id != marker.root_id): raise RelationViolationError("Private bus checkpoint pending intent is inconsistent.") - if replace(self.prior, revision=self.expected.revision) == self.expected: - return saved.rebind_source(bus, marker, db, path, info) - return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self) + return super().recover(bus, marker, db, path, saved, info) + + def recovery_end(self, info: os.stat_result) -> int: + return self.expected.revision.size + + def chunk_complete(self, offset: int, size: int, rows: int) -> bool: + return offset == self.expected.revision.size def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None: if ( diff --git a/src/agent_comms/private_bus_checkpoint.py b/src/agent_comms/private_bus_checkpoint.py index d0fb511ccb..5d43cc2048 100644 --- a/src/agent_comms/private_bus_checkpoint.py +++ b/src/agent_comms/private_bus_checkpoint.py @@ -26,7 +26,7 @@ from typing import TYPE_CHECKING, BinaryIO, ClassVar, Generic, Literal, TypeVar from .bus_source_page import AddressedPage -from .checkpoint_seals import CheckpointSeal, FinalSeal, PendingSeal, PrefixSeal, PrefixSource +from .checkpoint_seals import FinalSeal, PendingSeal, PrefixSeal, PrefixSource from .coordination_database import CoordinationStore from .coordination_errors import IdentityConflict from .envelope_claim_transitions import ( @@ -101,19 +101,59 @@ def require_source(self, marker: WireMetadata, info: os.stat_result, stream: Bin raise RelationViolationError("Private bus checkpoint exceeds the durable sequence marker.") def rebind_source(self, bus: WireLog, marker: WireMetadata, db: sqlite3.Connection, - path: Path, info: os.stat_result) -> PrefixWitness: + info: os.stat_result) -> PrefixWitness: """Publish the unchanged prefix's new revision under the exclusive bus lock.""" rebound = replace(self, revision=replace(FileRevision.from_stat(info), size=self.offset)) - marker.seal.require_rebuilt(self, rebound) - marker.seal_with(PendingSeal.capture(self, rebound, path)) - bus.write_metadata_unlocked(marker) - with db: - CheckedPrefix.capture(rebound.root_id, rebound.revision, rebound.through_seq, - bytes.fromhex(rebound.digest), rebound.tail).upsert(db) - _directory_sync(path) - marker.seal_with(FinalSeal.capture(rebound, path)) - bus.write_metadata_unlocked(marker) - return rebound + return CheckedPrefix.commit_tail(bus, marker, db, self, rebound, []) + + def index_tail(self, bus: WireLog, marker: WireMetadata, db: sqlite3.Connection, + path: Path, info: os.stat_result) -> PrefixWitness: + """Continue the proved cut, committing bounded chunks without a prefix rescan.""" + saved = self + if saved.offset == 0 and saved.digest != _SEED.hex(): + raise RelationViolationError("Private bus has no proved empty prefix.") + digest = bytes.fromhex(saved.digest) + last_seq = saved.through_seq + additions: list[tuple[int, bytes, WireRecord]] = [] + size = 0 + end = marker.seal.recovery_end(info) + revision = FileRevision.from_stat(info) + with bus.path.open("rb") as stream: + def commit(cut: int) -> None: + nonlocal saved, additions, size + if (FileRevision.from_file(stream) != revision + or FileRevision.from_stat(bus.path.stat()) != revision): + raise RelationViolationError("Private bus changed during tail recovery.") + expected = replace( + saved, revision=replace(revision, size=cut), through_seq=last_seq, + digest=digest.hex(), tail=_tail(stream, cut)) + saved = CheckedPrefix.commit_tail(bus, marker, db, saved, expected, additions) + additions, size = [], 0 + + def collect(offset: int, raw: bytes, record: WireRecord) -> None: + nonlocal digest, last_seq, size + digest = _chain(digest, raw) + last_seq = record.sequence_after(last_seq) + if last_seq > marker.last_seq: + raise RelationViolationError("Private bus tail exceeds its reserved sequence.") + additions.append((offset, raw, record)) + size += len(raw) + cut = offset + len(raw) + if marker.seal.chunk_complete(cut, size, len(additions)): + commit(cut) + + for _ in bus.verified_records_unlocked( + marker, on_row=collect, offset=saved.offset, + previous_sequence=saved.through_seq, end=end, + ): + pass + if additions or saved.revision != replace(revision, size=end): + commit(end) + else: + marker.seal.require_rebuilt(saved, saved) + marker.seal_with(FinalSeal.capture(saved, path)) + bus.write_metadata_unlocked(marker) + return marker.seal.recover(bus, marker, db, path, saved, info) def require_append(self, marker: WireMetadata, info: os.stat_result, offset: int, sequence: int) -> None: @@ -1099,8 +1139,9 @@ def collect(offset, raw, record): != FileRevision.from_stat(info) ): raise RelationViolationError("Private bus changed during checkpoint installation.") - cls.capture(marker.root_id, FileRevision.from_stat(info), through_seq, digest, - _tail(stream, info.st_size)).insert(db) + cls.capture( + marker.root_id, FileRevision.from_stat(info), through_seq, + digest, _tail(stream, info.st_size)).insert(db) _directory_sync(staged) with closing(_connect(staged, readonly=True)) as db: witness = _saved(db) @@ -1150,19 +1191,35 @@ def witness(self) -> PrefixWitness: def capture( cls, root_id: str, revision: FileRevision, seq: int, digest: bytes, tail: str ) -> CheckedPrefix: - return cls( - 1, - 2, - root_id, - revision.identity.device, - revision.identity.inode, - revision.size, - seq, - digest.hex(), - tail, - revision.mtime_ns, - revision.ctime_ns, - ) + return cls.from_witness(PrefixWitness(root_id, revision, seq, digest.hex(), tail)) + + @classmethod + def from_witness(cls, witness: PrefixWitness) -> CheckedPrefix: + revision = witness.revision + return cls(1, 2, witness.root_id, revision.identity.device, revision.identity.inode, + witness.offset, witness.through_seq, witness.digest, witness.tail, + revision.mtime_ns, revision.ctime_ns) + + @classmethod + def commit_tail(cls, bus: WireLog, marker: WireMetadata, db: sqlite3.Connection, + prior: PrefixWitness, expected: PrefixWitness, + additions: list[tuple[int, bytes, WireRecord]]) -> PrefixWitness: + """Commit one chunk through the same intent as an ordinary append.""" + path = _path(bus.path) + marker.seal.require_rebuilt(prior, expected) + marker.seal_with(PendingSeal.capture(prior, expected, path)) + bus.write_metadata_unlocked(marker) + with db: + sequence = prior.through_seq + for offset, raw, record in additions: + sequence = _index_row(db, offset, raw, record, sequence) + if sequence != expected.through_seq: + raise RelationViolationError("Indexed tail differs from its sealed sequence.") + cls.from_witness(expected).upsert(db) + _directory_sync(path) + marker.seal_with(FinalSeal.capture(expected, path)) + bus.write_metadata_unlocked(marker) + return expected def _path(bus_path: Path) -> Path: @@ -1268,69 +1325,6 @@ def checked_prefix_exists(bus_path: Path) -> bool: return path.exists() or path.is_symlink() -def _rebuild_checkpoint_unlocked( - bus: WireLog, - marker: WireMetadata, - db: sqlite3.Connection, - db_path: Path, - info: os.stat_result, - seal: CheckpointSeal, -) -> PrefixWitness: - """Read the whole bus and rebuild its index while holding the write lock.""" - saved = _saved(db) - digest = _SEED - prior_seen = saved.offset == 0 - last_seq = 0 - count = 0 - size = 0 - - with db: - for owner in TypedTable.members_with(CheckpointTable): - owner.delete(db) - - def collect(offset, raw, record): - nonlocal digest, prior_seen, last_seq, count, size - count += 1 - size += len(raw) - if count > 100_000 or size > 128 * 1024 * 1024: - raise RelationViolationError("Private bus pending recovery exceeds cold bound.") - digest = _chain(digest, raw) - last_seq = _index_row(db, offset, raw, record, last_seq) - if offset + len(raw) == saved.offset: - prior_seen = True - if digest.hex() != saved.digest or last_seq != saved.through_seq: - raise RelationViolationError( - "Private bus checkpoint old prefix changed during recovery." - ) - - for _ in bus.verified_records_unlocked(marker, on_row=collect): - pass - if not prior_seen or (saved.offset == 0 and saved.digest != _SEED.hex()): - raise RelationViolationError("Private bus checkpoint old prefix is unavailable.") - if FileRevision.from_stat(bus.path.stat()) != FileRevision.from_stat(info): - raise RelationViolationError("Private bus changed during pending recovery.") - with bus.path.open("rb") as stream: - expected = PrefixWitness( - marker.root_id, - FileRevision.from_stat(info), - last_seq, - digest.hex(), - _tail(stream, info.st_size), - ) - if last_seq > marker.last_seq: - raise RelationViolationError("Private bus checkpoint exceeds the durable sequence marker.") - seal.require_rebuilt(saved, expected) - marker.seal_with(PendingSeal.capture(saved, expected, db_path)) - bus.write_metadata_unlocked(marker) - CheckedPrefix.capture(expected.root_id, expected.revision, last_seq, digest, expected.tail).upsert( - db - ) - _directory_sync(db_path) - marker.seal_with(FinalSeal.capture(expected, db_path)) - bus.write_metadata_unlocked(marker) - return expected - - def _check_final_checkpoint_read_only(bus, marker, db, stream, path) -> PrefixWitness: """Check a final, up-to-date checkpoint under the shared bus lock; never writes.""" saved = _saved(db) @@ -1372,79 +1366,7 @@ def _verify_open_checkpoint_unlocked(bus, marker, db, stream, path) -> PrefixWit saved = _saved(db) info = os.fstat(stream.fileno()) saved.require_source(marker, info, stream) - recovered = marker.seal.recover(bus, marker, db, path, saved, info) - if recovered is not None: - return recovered - if FileRevision.from_stat(info) == saved.revision: - return saved - # Changed revision: a suffix alone cannot rule out an earlier in-place edit. - # Validate complete canonical history, compare digest at the saved offset, - # and collect only newly appended rows for one atomic index transaction. - digest = _SEED - observed_prefix = saved.offset == 0 - prefix_seq = 0 - additions: list[tuple[int, bytes, WireRecord]] = [] - suffix_bytes = 0 - last_seq = 0 - - def collect( - offset: int, - raw: bytes, - record: WireRecord, - ) -> None: - nonlocal digest, observed_prefix, prefix_seq, suffix_bytes, last_seq - digest = _chain(digest, raw) - end = offset + len(raw) - last_seq = record.sequence_after(last_seq) - if end == saved.offset: - observed_prefix = True - prefix_seq = last_seq - if digest.hex() != saved.digest or prefix_seq != saved.through_seq: - raise RelationViolationError( - "Private bus checkpoint checked prefix changed." - ) - elif end > saved.offset: - if not observed_prefix or offset < saved.offset: - raise RelationViolationError( - "Private bus checkpoint offset is not a row boundary." - ) - suffix_bytes += len(raw) - if suffix_bytes > 16 * 1024 * 1024 or len(additions) >= 10_000: - raise RelationViolationError( - "Private bus checkpoint crash suffix exceeds recovery bound." - ) - additions.append((offset, raw, record)) - - for _ in bus.verified_records_unlocked(marker, on_row=collect): - pass - if not observed_prefix or (saved.offset == 0 and saved.digest != _SEED.hex()): - raise RelationViolationError("Private bus checkpoint prefix is unavailable.") - if ( - FileRevision.from_stat(os.fstat(stream.fileno())) != FileRevision.from_stat(info) - or FileRevision.from_stat(bus.path.stat()) != FileRevision.from_stat(info) - ): - raise RelationViolationError("Private bus changed during checkpoint validation.") - # Full parser above checked all cross-prefix response key duplicates. - expected = PrefixWitness( - saved.root_id, - FileRevision.from_stat(info), - last_seq, - digest.hex(), - _tail(stream, info.st_size), - ) - marker.seal_with(PendingSeal.capture(saved, expected, path)) - bus.write_metadata_unlocked(marker) - with db: - indexed_sequence = saved.through_seq - for offset, raw, record in additions: - indexed_sequence = _index_row(db, offset, raw, record, indexed_sequence) - CheckedPrefix.capture( - saved.root_id, expected.revision, last_seq, digest, expected.tail - ).upsert(db) - _directory_sync(path) - marker.seal_with(FinalSeal.capture(expected, path)) - bus.write_metadata_unlocked(marker) - return _saved(db) + return marker.seal.recover(bus, marker, db, path, saved, info) @contextmanager @@ -1506,17 +1428,7 @@ def append_private_bus_checkpoint_unlocked( digest.hex(), _tail(stream, info.st_size), ) - marker.seal_with(PendingSeal.capture(saved, expected, path)) - bus.write_metadata_unlocked(marker) - with db: - _index_row(db, offset, raw, record, saved.through_seq) - CheckedPrefix.capture( - saved.root_id, expected.revision, sequence, digest, expected.tail - ).upsert(db) - _directory_sync(path) - marker.seal_with(FinalSeal.capture(expected, path)) - bus.write_metadata_unlocked(marker) - return _saved(db) + return CheckedPrefix.commit_tail(bus, marker, db, saved, expected, [(offset, raw, record)]) diff --git a/src/agent_comms/wire_log.py b/src/agent_comms/wire_log.py index 4aaa135198..4129ba0cc5 100644 --- a/src/agent_comms/wire_log.py +++ b/src/agent_comms/wire_log.py @@ -8,6 +8,7 @@ import os import sqlite3 import stat +import sys import time from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping @@ -772,13 +773,17 @@ def verified_records_unlocked( metadata: WireMetadata, *, on_row: Callable[[int, bytes, WireRecord], None] | None = None, + offset: int = 0, + previous_sequence: int = 0, + end: int = sys.maxsize, ) -> Iterator[WireRecord]: - """Read the whole file strictly; reject corruption even after the requested append.""" - scan = WireScan(metadata) + """Decode the selected range, continuing its already proved sequence.""" + scan = WireScan(metadata, previous_sequence=previous_sequence) if not self.path.exists(): return with self.path.open("rb") as stream: - for offset, line in WireRecord.rows(stream): + stream.seek(offset) + for offset, line in WireRecord.rows(stream, end, end): try: record = scan.read(line) except StoredHistoryUnreadable as error: