Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 38 additions & 12 deletions src/agent_comms/checkpoint_seals.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.")
Expand Down Expand Up @@ -100,8 +107,24 @@ def recover(
path: Path,
saved: PrefixWitness,
info: os.stat_result,
) -> None:
self.check_final(saved, path)
) -> 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.")
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():
Expand Down Expand Up @@ -129,15 +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.")
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 (
Expand Down
Loading