From 8d1bb04882501d5b04a66662c169c4b1e8ecbc0e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 3 Nov 2026 14:16:18 -0500 Subject: [PATCH 1/7] P1086: separate retained native bytes from original lookup identity --- src/agent_comms/agent_backend.py | 8 ++ src/agent_comms/native_runtime_input.py | 4 +- src/agent_comms/native_transcript.py | 97 ++++++++++++++++++------- src/agent_comms/pi_native_backend.py | 57 ++++++++++++++- src/agent_comms/session_file.py | 37 ++++++++++ tests/test_native_transcript.py | 85 +++++++++++++++++++++- 6 files changed, 255 insertions(+), 33 deletions(-) diff --git a/src/agent_comms/agent_backend.py b/src/agent_comms/agent_backend.py index 13211cc8d9..78ce966d0c 100644 --- a/src/agent_comms/agent_backend.py +++ b/src/agent_comms/agent_backend.py @@ -44,6 +44,8 @@ from .launch_runtime import LaunchRuntime from .owned_turn import TurnDelivery from .threads import Thread + from .native_session_reopen import NativeSessionIdentity + from .private_path import FileRevision from .transcript_sources import TranscriptSource from .turn_runner import TurnRunner @@ -63,6 +65,12 @@ class RuntimeLocation(DeclaredFamily, affix="Runtime"): def backend(self): return self._backend + @classmethod + def retained(cls, path: Path, identity: NativeSessionIdentity, + revision: FileRevision) -> RuntimeLocation: + """Only a backend with an original-coordinate reader can bind retained bytes.""" + raise BackendDeclarationError(f"{cls.__name__} cannot read retained native history") + def require_forkable(self): pass diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py index 2d5943fb06..7ce36bb6ca 100644 --- a/src/agent_comms/native_runtime_input.py +++ b/src/agent_comms/native_runtime_input.py @@ -255,7 +255,7 @@ def for_native_user(cls, db, reader, user, *, session_id): """ rows = [row for input_id in user.input_id.present() for row in cls.select( db, where="input_id=? AND session_file=? AND session_id=?", - parameters=(input_id, str(reader.path), session_id), + parameters=(input_id, str(reader.source_path), session_id), )] for row in rows: user.require_tracked_user() @@ -307,7 +307,7 @@ def publication_revision(cls, root, reader, owner_lookup) -> PublishedReplyRevis "WHERE n.owner_lookup=? AND n.session_file=? " "AND n.session_id=? AND n.session_entry_id IS NOT NULL " "AND o.receipt_seq IS NOT NULL", - (owner_lookup, str(reader.path), reader.session_id), + (owner_lookup, str(reader.source_path), reader.session_id), ) )[0] diff --git a/src/agent_comms/native_transcript.py b/src/agent_comms/native_transcript.py index e0c25697eb..bba8237022 100644 --- a/src/agent_comms/native_transcript.py +++ b/src/agent_comms/native_transcript.py @@ -234,6 +234,11 @@ class NativeTranscript(TranscriptSource): def __init__(self, path: Path): self.path = path + @property + def source_path(self) -> Path: + """Original lookup spelling, independent of the physical byte location.""" + return self.path + def completed_input(self, context: NativeContextProof) -> bool: """Start at the proved input byte offset; never borrow the following input's final.""" cut = self.cut() @@ -282,18 +287,19 @@ def imported_sources(self): producer may also prepend model/settings records. Stop at conversation; never scan the remaining history or claim this was a model request. """ - revision = FileRevision.from_stat(self.path.stat()) - with self.path.open("rb") as stream: - SessionEntry.read(stream.readline()) - sources = [] - for raw in stream: - entry = NativeEntry.read(raw) - if entry.is_message: - break - sources.extend(entry.imported_sources()) - if FileRevision.from_stat(self.path.stat()) != revision: - raise ValueError("Imported context saved source changed while reading") - return tuple(sources) + with self.reading(): + revision = FileRevision.from_stat(self.path.stat()) + with self.path.open("rb") as stream: + SessionEntry.read(stream.readline()) + sources = [] + for raw in stream: + entry = NativeEntry.read(raw) + if entry.is_message: + break + sources.extend(entry.imported_sources()) + if FileRevision.from_stat(self.path.stat()) != revision: + raise ValueError("Imported context saved source changed while reading") + return tuple(sources) def cut(self) -> SessionCut: try: @@ -307,9 +313,9 @@ def cut(self) -> SessionCut: def committed_output_cut(self, location: NativeRecordLocation) -> SessionCut: """Corroborate one indexed output against its original header and row.""" - if self.path.absolute() != location.path: + if self.source_path.absolute() != location.path: raise TranscriptChanged() - with self.path.open("rb") as stream: + with self.reading(), self.path.open("rb") as stream: revision = FileRevision.from_file(stream) header = SessionEntry.read(stream.readline()) if header.id != location.session_id or revision.size < location.session_entry_end: @@ -330,7 +336,7 @@ def publication_revision(self, root: Path, recipient: RecordedThreadIncarnation) @contextmanager def projection(self, sources, routes): """Project records through their native input and the session's routes.""" - with routes.for_session(str(self.path)) as session_routes: + with routes.for_session(str(self.source_path)) as session_routes: def project(records, *, max_records, max_bytes): return sources.native_records( self.fragments(records, max_records=max_records, max_bytes=max_bytes), @@ -358,7 +364,7 @@ def fragments(self, records, *, max_records, max_bytes): yield tuple(fragment) def recorded_contexts(self, contexts, entry_ids): - return contexts.entries(NativeSessionIdentity(self.session_id, str(self.path)), + return contexts.entries(NativeSessionIdentity(self.session_id, str(self.source_path)), tuple(NativeItemId(identity) for identity in entry_ids)) @property @@ -475,15 +481,16 @@ def voice(self, deal: Speaking) -> Voice: return deal def tail(self, *, max_bytes: int | None = None): - end = self.path.stat().st_size - floor = max(0, end - max_bytes) if max_bytes is not None else 0 - for _, _, raw in reverse_records(self.path, end, floor=floor): - # Skip only the record still being appended; corrupt history raises. - if raw.endswith(b"\n"): - yield NativeEntry.read(raw) + with self.reading(): + end = self.path.stat().st_size + floor = max(0, end - max_bytes) if max_bytes is not None else 0 + for _, _, raw in reverse_records(self.path, end, floor=floor): + # Skip only the record still being appended; corrupt history raises. + if raw.endswith(b"\n"): + yield NativeEntry.read(raw) def forward(self, after: int, through: int): - with self.path.open("rb") as stream: + with self.reading(), self.path.open("rb") as stream: stream.seek(after) while stream.tell() < through: start = stream.tell() @@ -493,5 +500,45 @@ def forward(self, after: int, through: int): yield NativeRecord.read(start, stream.tell(), raw) def reverse(self, before: int, *, floor: int = 0) -> Generator[NativeRecord]: - for start, end, raw in reverse_records(self.path, before, floor=floor): - yield NativeRecord.read(start, end, raw) + with self.reading(): + for start, end, raw in reverse_records(self.path, before, floor=floor): + yield NativeRecord.read(start, end, raw) + + +class RetainedNativeTranscript(NativeTranscript): + """Read an existing immutable journal under its original lookup coordinates. + + This binding copies nothing and grants no native launch or publication authority. + The original journal header and the retained file revision must both still match. + """ + + def __init__(self, path: Path, identity: NativeSessionIdentity, revision: FileRevision) -> None: + super().__init__(path) + self.original = identity + self.revision = revision + self.require_retained() + + @property + def source_path(self) -> Path: + return self.original.path + + @property + def session_id(self): + return self.original.session_id + + def require_retained(self) -> None: + if (FileRevision.capture(self.path) != self.revision + or super().session_id != self.original.session_id): + raise TranscriptChanged("Retained native journal differs from its binding") + + @contextmanager + def reading(self): + self.require_retained() + try: + yield self + finally: + self.require_retained() + + def cut(self) -> SessionCut: + with self.reading(): + return super().cut() diff --git a/src/agent_comms/pi_native_backend.py b/src/agent_comms/pi_native_backend.py index 24380cc25b..ff85b38792 100644 --- a/src/agent_comms/pi_native_backend.py +++ b/src/agent_comms/pi_native_backend.py @@ -37,11 +37,13 @@ from .native_package import SYSTEM_PROMPT from .native_pi import NativePiRpcLaunch from .native_session_reopen import NativeSessionIdentity -from .native_transcript import NativeTranscript +from .native_transcript import NativeTranscript, RetainedNativeTranscript +from .private_path import FileRevision from .pi_commands import Compact, NativeQuery from .rule_deal import SystemPromptDelivery from .selected_session import SavedSelectedSession from .session_revision import ( + NoSessionObservation, SessionObservation, SessionRevision, SessionRevisionUnavailable, @@ -90,6 +92,15 @@ class PiSessionFile(RuntimeLocation): path: str + @property + def transcript(self) -> NativeTranscript: + return NativeTranscript(Path(self.path)) + + @classmethod + def retained(cls, path: Path, identity: NativeSessionIdentity, + revision: FileRevision) -> RetainedPiSessionFile: + return RetainedPiSessionFile(str(path), identity, revision) + def select_session(self, directory, execution, source_reads, *, participant): path = Path(self.path).absolute() retained = execution.retained_source(path) @@ -101,14 +112,14 @@ def select_session(self, directory, execution, source_reads, *, participant): owner=participant.owner.thread.incarnation.require_recorded()) def history_source(self, root, incarnation): - return NativeTranscript(Path(self.path)) + return self.transcript def deal_delivery(self) -> SystemPromptDelivery: """Native Pi always launches with the prepared prompt (NativePiRpcLaunch.bootstrap).""" return SystemPromptDelivery(SYSTEM_PROMPT) def imported_sources(self): - return NativeTranscript(Path(self.path)).imported_sources() + return self.transcript.imported_sources() def storage_files(self): return (NativeSessionLocks(Path(self.path)),) @@ -121,10 +132,48 @@ def session_observation(self, root: Path, thread: Thread) -> SessionObservation: @property def saved_model(self) -> ThreadModel: """The newest model the saved session selected, read from its tail.""" - return next((model for entry in NativeTranscript(Path(self.path)).tail() + return next((model for entry in self.transcript.tail() for model in entry.saved_models), UndeclaredModel()) +@dataclass(frozen=True) +class RetainedPiSessionFile(PiSessionFile): + """Original-coordinate history, with no selected native session or writable sidecars.""" + + identity: NativeSessionIdentity + revision: FileRevision + resumable: ClassVar[bool] = False + + @property + def read_key(self) -> str: + return self.identity.session_file + + @property + def transcript(self) -> NativeTranscript: + return RetainedNativeTranscript(Path(self.path), self.identity, self.revision) + + def require_forkable(self): + raise NativePiUnavailable("Retained native history cannot fork a session") + + def arguments(self): + raise NativePiUnavailable("Retained native history has no launch arguments") + + def new_location(self, root, worktree): + raise NativePiUnavailable("Retained native history cannot start a session") + + def select_session(self, directory, execution, source_reads, *, participant): + raise NativePiUnavailable("Retained native history cannot select a native input") + + async def compact(self, operation): + return FailedCompaction("Retained native history cannot be compacted.") + + def storage_files(self): + return () + + def session_observation(self, root: Path, thread: Thread) -> SessionObservation: + return NoSessionObservation() + + class PersistentPiSession: """Serialize borrowing and retirement of one native session's actual child.""" diff --git a/src/agent_comms/session_file.py b/src/agent_comms/session_file.py index 228cf018a6..5e05c1c665 100644 --- a/src/agent_comms/session_file.py +++ b/src/agent_comms/session_file.py @@ -10,6 +10,8 @@ from .errors import RelationViolationError from .nominal.field_codec import NullMember, PathText, RecordMember, SubclassUnion, WireTransient from .unstarted_runtime import UnstartedRuntime +from .native_session_reopen import NativeSessionIdentity +from .private_path import FileRevision if TYPE_CHECKING: from .agent_backend import AgentBackend, RuntimeLocation @@ -171,6 +173,41 @@ class BoundExternalSession(BoundSessionLocation, ExternalSessionLocation, declar pass +@dataclass(frozen=True, slots=True) +class RetainedSessionLocation(RecordMember, SessionFile): + """An already retained journal: physical storage and original native identity. + + The storage root supplies bytes, never registry, publication or launch authority. + This declaration performs no capture and never substitutes newly imported entry IDs. + """ + + storage: BusSessionLocation + identity: NativeSessionIdentity + revision: FileRevision + + def bound_to(self, root: Path) -> BoundRetainedSession: + return BoundRetainedSession(self.storage, self.identity, self.revision, root.resolve()) + + def location(self, backend: type[AgentBackend]) -> RuntimeLocation: + raise RelationViolationError("Retained history must be bound to its storage root") + + def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation: + raise RelationViolationError("Retained native history has no writable session") + + def applied(self, previous: SessionFile) -> SessionFile: + return self + + +@dataclass(frozen=True, slots=True) +class BoundRetainedSession(RetainedSessionLocation, declared_name=None): + root: Annotated[Path, WireTransient()] = field(repr=False, compare=False) + + def location(self, backend: type[AgentBackend]) -> RuntimeLocation: + return backend.location_type.retained( + self.storage.resolve(self.root), self.identity, self.revision, + ) + + @dataclass(frozen=True, slots=True) class NoSessionFile(NullMember, SessionFile): def bound_to(self, root: Path) -> NoSessionFile: diff --git a/tests/test_native_transcript.py b/tests/test_native_transcript.py index e6edc07818..da1a2dc37e 100644 --- a/tests/test_native_transcript.py +++ b/tests/test_native_transcript.py @@ -13,9 +13,17 @@ ToolCallId, ) from agent_comms.native_entries import MessageEntry, NativeEntry, TranscriptProjection, UnknownEntry -from agent_comms.native_transcript import NativeTranscript +from agent_comms.native_transcript import NativeTranscript, RetainedNativeTranscript +from agent_comms.coordination_errors import TranscriptChanged +from agent_comms.errors import RelationViolationError +from agent_comms.native_failures import NativePiUnavailable +from agent_comms.pi_native_backend import PiNativeBackend +from agent_comms.private_path import FileRevision +from agent_comms.repo import Checkout +from agent_comms.threads import Thread +from agent_comms.session_file import BusSessionLocation, RetainedSessionLocation, SessionFile from agent_comms.native_input_proof import NativeContextProof -from agent_comms.native_session_reopen import PiSessionId +from agent_comms.native_session_reopen import NativeSessionIdentity, PiSessionId from agent_comms.pi_payloads import UserMessage from agent_comms.transcript_events import ( AssistantTranscript, @@ -36,6 +44,79 @@ def encoded(value): return json.dumps(value).encode() + b"\n" +@pytest.fixture +def retained_native_location(tmp_path: Path) -> tuple[Path, RetainedSessionLocation]: + session = PiSessionId.fresh() + identity = NativeSessionIdentity(session, str(tmp_path / "original" / "session.jsonl")) + storage = tmp_path / "retained" / "session.jsonl" + storage.parent.mkdir() + storage.write_bytes( + encoded(dict(type="session", version=3, id=session, cwd=str(tmp_path), + timestamp="2026-10-01T00:00:00.000Z")) + + encoded(dict(type="message", id="original-entry", parentId=None, + timestamp="2026-10-01T00:00:00.000Z", + message=dict(role="assistant", content=[dict(type="text", text="saved")], + stopReason="stop"))) + ) + revision = FileRevision.capture(storage).require(ValueError("Missing retained fixture")) + location = RetainedSessionLocation( + BusSessionLocation(storage.relative_to(tmp_path)), identity, revision, + ) + return tmp_path, location + + +def test_retained_location_roundtrip_keeps_original_keys_and_reads_retained_bytes( + retained_native_location: tuple[Path, RetainedSessionLocation], +) -> None: + root, original = retained_native_location + restored = FieldCodec.decode(SessionFile, FieldCodec.encode(original)).bound_to(root) + runtime = restored.location(PiNativeBackend) + assert not original.identity.path.exists() + assert runtime.read_key == original.identity.session_file + assert Path(runtime.path) == original.storage.resolve(root) + reader = runtime.history_source(root, Thread("reader", frozenset(), Checkout(root)).incarnation) + with reader.reading(): + cut = reader.cut() + records = tuple(reader.forward(0, cut.offset)) + assert reader.source_path == original.identity.path + assert reader.session_id == original.identity.session_id + assert records[-1].require_entry().require_entry_id() == FreshNativeId("original-entry") + assert records[-1].events[0].text == "saved" + assert not runtime.resumable + assert runtime.storage_files() == () + with pytest.raises(RelationViolationError): + restored.writable_location(PiNativeBackend) + with pytest.raises(NativePiUnavailable): + runtime.require_forkable() + with pytest.raises(NativePiUnavailable): + runtime.arguments() + with pytest.raises(NativePiUnavailable): + runtime.new_location(root, root) + + +def test_retained_reader_rejects_another_header_identity( + retained_native_location: tuple[Path, RetainedSessionLocation], +) -> None: + root, original = retained_native_location + changed = replace(original.identity, session_id=PiSessionId.fresh()) + with pytest.raises(TranscriptChanged): + RetainedNativeTranscript(original.storage.resolve(root), changed, original.revision) + + +def test_retained_reader_rejects_changed_bytes_during_its_read( + retained_native_location: tuple[Path, RetainedSessionLocation], +) -> None: + root, original = retained_native_location + path = original.storage.resolve(root) + reader = RetainedNativeTranscript(path, original.identity, original.revision) + with pytest.raises(TranscriptChanged): + with reader.reading(): + with path.open("ab") as output: + output.write(b"\n") + with pytest.raises(TranscriptChanged): + reader.cut() + + def test_native_discriminator_is_separate_from_normalized_family_codec(): entry = NativeEntry.read( encoded( From 16d81e23ce41b1f40fbbd6a70592f2cced312432 Mon Sep 17 00:00:00 2001 From: pi-publication-carry Date: Tue, 3 Nov 2026 15:25:12 -0500 Subject: [PATCH 2/7] Bind retained read marks and unread workers to original source identity --- src/agent_comms/agent_backend.py | 35 ++++++++- src/agent_comms/coordinated_runtime.py | 3 +- src/agent_comms/native_evidence_read.py | 6 ++ src/agent_comms/native_session_reopen.py | 18 +++++ src/agent_comms/native_transcript.py | 17 ++++- src/agent_comms/pi_native_backend.py | 28 ++++++- src/agent_comms/read_ledger.py | 4 +- src/agent_comms/session_file.py | 4 +- src/agent_comms/team/tool_watch.py | 6 +- src/agent_comms/transcript_routes.py | 11 +-- src/agent_comms/transcripts.py | 5 +- src/agent_comms/unstarted_runtime.py | 2 +- src/agent_comms/view_unread.py | 60 ++++++++++----- tests/test_native_transcript.py | 93 +++++++++++++++++++++--- tests/test_thread_unread.py | 12 +-- tests/test_watched.py | 3 +- 16 files changed, 244 insertions(+), 63 deletions(-) diff --git a/src/agent_comms/agent_backend.py b/src/agent_comms/agent_backend.py index 78ce966d0c..9bf3481323 100644 --- a/src/agent_comms/agent_backend.py +++ b/src/agent_comms/agent_backend.py @@ -44,8 +44,8 @@ from .launch_runtime import LaunchRuntime from .owned_turn import TurnDelivery from .threads import Thread - from .native_session_reopen import NativeSessionIdentity - from .private_path import FileRevision + from .native_session_reopen import NativeSessionSnapshot + from .private_path import FileObservation, FileRevision from .transcript_sources import TranscriptSource from .turn_runner import TurnRunner @@ -66,7 +66,7 @@ def backend(self): return self._backend @classmethod - def retained(cls, path: Path, identity: NativeSessionIdentity, + def retained(cls, path: Path, identity: NativeSessionSnapshot, revision: FileRevision) -> RuntimeLocation: """Only a backend with an original-coordinate reader can bind retained bytes.""" raise BackendDeclarationError(f"{cls.__name__} cannot read retained native history") @@ -88,8 +88,35 @@ def new_location(self, root, worktree): def arguments(self): return {"session_file": self.read_key} + def read_locations(self) -> tuple[RuntimeLocation, ...]: + """The bound byte locations; a logical read key is never a filesystem path.""" + return (self,) + def read_keys(self): - return (self.read_key,) + return tuple(location.read_key for location in self.read_locations()) + + def tool_watch_locations(self) -> tuple[RuntimeLocation, ...]: + return self.read_locations() + + def read_revision(self, observed: FileObservation) -> FileObservation: + return observed + + def read_inode(self, revision: FileRevision) -> int: + return revision.identity.inode + + def read_cut(self): + # Like transcript_cursor, this boundary loads its implementation only after + # threads has finished declaring the runtime family. + from .native_transcript import NativeTranscript + + return NativeTranscript(Path(self.path)).cut() + + def read_position(self, mark: int, revision: FileRevision) -> int: + return mark if mark <= revision.size else 0 + + def mark_read(self, reads, viewer: str, through: int) -> None: + for location in self.read_locations(): + reads.mark_read(viewer, location.read_key, through, location.read_cut()) def transcript_cursor(self, frontier, offset, *, source): return frontier.cursor(source.read_key, offset) diff --git a/src/agent_comms/coordinated_runtime.py b/src/agent_comms/coordinated_runtime.py index 702b257215..1b03839a5c 100644 --- a/src/agent_comms/coordinated_runtime.py +++ b/src/agent_comms/coordinated_runtime.py @@ -100,13 +100,14 @@ def validate(self) -> None: try: NativeSessionFiles.require_private_directory(self.root) owner = wire(self.root).registry.require(self.owner_name) + runtime = owner.writable_runtime if not owner.worktree.path.is_dir(): raise ValueError("registered participant worktree is unavailable") directory = owner.native_session_directory(self.root) directory.mkdir(parents=True, mode=0o700, exist_ok=True) NativeSessionFiles.require_private_directory(directory) paths = ((self.session_file,) if self.session_file is not None - else owner.runtime.read_keys()) + else runtime.read_keys()) for path in paths: NativeSessionFiles.require_private_directory(Path(path).absolute().parent) except NativePiUnavailable as error: diff --git a/src/agent_comms/native_evidence_read.py b/src/agent_comms/native_evidence_read.py index 90c8e4130f..92ff4f5400 100644 --- a/src/agent_comms/native_evidence_read.py +++ b/src/agent_comms/native_evidence_read.py @@ -34,6 +34,9 @@ class UnverifiedNativeEvidencePrefix: size = 0 + def observed_revision(self) -> FileRevision: + raise NativePiUnavailable("Native evidence has no acquired source revision") + @staticmethod def prepare(source: PrivateEvidenceRead, revision: FileRevision) -> FileRevision: return revision @@ -50,6 +53,9 @@ class VerifiedNativeEvidencePrefix(UnverifiedNativeEvidencePrefix): size: int = field() revision: FileRevision + def observed_revision(self) -> FileRevision: + return self.revision + def prepare(self, source: PrivateEvidenceRead, revision: FileRevision) -> FileRevision: return source.verify_snapshot(self.size, self.revision) diff --git a/src/agent_comms/native_session_reopen.py b/src/agent_comms/native_session_reopen.py index 26580782d7..fdb405c921 100644 --- a/src/agent_comms/native_session_reopen.py +++ b/src/agent_comms/native_session_reopen.py @@ -166,3 +166,21 @@ def locate(cls, session_file: str) -> NativeSessionIdentity: return cls.from_evidence(scope.for_source(Path(session_file).absolute())) except NativePiUnavailable as error: raise NativeReopenError("Saved native session source cannot be opened") from error + + +@dataclass(frozen=True) +class NativeSessionSnapshot(NativeSessionIdentity): + """The original journal's identity and observed revision, not a new observation of it.""" + + revision: FileRevision + + @classmethod + def from_evidence(cls, evidence: NativeEvidenceRead) -> NativeSessionSnapshot: + identity = NativeSessionIdentity.from_evidence(evidence) + return cls(identity.session_id, identity.session_file, + evidence.source.prefix.observed_revision()) + + def require_retained(self, retained: FileRevision) -> None: + """A full retained image preserves the original offsets and read-mark extent.""" + if self.revision.size != retained.size: + raise NativeReopenError("Retained journal does not contain the complete original image") diff --git a/src/agent_comms/native_transcript.py b/src/agent_comms/native_transcript.py index bba8237022..a6b947a69d 100644 --- a/src/agent_comms/native_transcript.py +++ b/src/agent_comms/native_transcript.py @@ -37,7 +37,7 @@ TranscriptProjection, ) from .native_runtime_input import NativeRuntimeInput, PublishedReplyRevision -from .native_session_reopen import NativeSessionIdentity +from .native_session_reopen import NativeSessionIdentity, NativeSessionSnapshot from .private_path import FileRevision from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime from .transcript_sources import TranscriptRecord, TranscriptSource @@ -301,6 +301,9 @@ def imported_sources(self): raise ValueError("Imported context saved source changed while reading") return tuple(sources) + def session_cut(self, revision: FileRevision, offset: int) -> PresentSessionCut: + return PresentSessionCut(revision, offset) + def cut(self) -> SessionCut: try: stream = self.path.open("rb") @@ -309,7 +312,7 @@ def cut(self) -> SessionCut: # One open file supplies both the revision and its last record boundary. with stream: revision = FileRevision.from_file(stream) - return PresentSessionCut(revision, record_end(stream, revision.size)) + return self.session_cut(revision, record_end(stream, revision.size)) def committed_output_cut(self, location: NativeRecordLocation) -> SessionCut: """Corroborate one indexed output against its original header and row.""" @@ -328,7 +331,7 @@ def committed_output_cut(self, location: NativeRecordLocation) -> SessionCut: or not raw.endswith(b"\n")): raise TranscriptChanged() location.require_entry(NativeEntry.read(raw)) - return PresentSessionCut(revision, location.session_entry_end) + return self.session_cut(revision, location.session_entry_end) def publication_revision(self, root: Path, recipient: RecordedThreadIncarnation): return self.cut().publication_revision(root, self, recipient) @@ -512,10 +515,11 @@ class RetainedNativeTranscript(NativeTranscript): The original journal header and the retained file revision must both still match. """ - def __init__(self, path: Path, identity: NativeSessionIdentity, revision: FileRevision) -> None: + def __init__(self, path: Path, identity: NativeSessionSnapshot, revision: FileRevision) -> None: super().__init__(path) self.original = identity self.revision = revision + identity.require_retained(revision) self.require_retained() @property @@ -539,6 +543,11 @@ def reading(self): finally: self.require_retained() + def session_cut(self, revision: FileRevision, offset: int) -> PresentSessionCut: + # The physical revision is fenced by reading(); positions and marks still + # belong to the original acquired inode, including committed-output cuts. + return PresentSessionCut(self.original.revision, offset) + def cut(self) -> SessionCut: with self.reading(): return super().cut() diff --git a/src/agent_comms/pi_native_backend.py b/src/agent_comms/pi_native_backend.py index ff85b38792..ce3b19c5b0 100644 --- a/src/agent_comms/pi_native_backend.py +++ b/src/agent_comms/pi_native_backend.py @@ -36,9 +36,10 @@ ) from .native_package import SYSTEM_PROMPT from .native_pi import NativePiRpcLaunch -from .native_session_reopen import NativeSessionIdentity +from .native_session_reopen import NativeSessionIdentity, NativeSessionSnapshot +from .coordination_errors import TranscriptChanged from .native_transcript import NativeTranscript, RetainedNativeTranscript -from .private_path import FileRevision +from .private_path import FileObservation, FileRevision from .pi_commands import Compact, NativeQuery from .rule_deal import SystemPromptDelivery from .selected_session import SavedSelectedSession @@ -97,7 +98,7 @@ def transcript(self) -> NativeTranscript: return NativeTranscript(Path(self.path)) @classmethod - def retained(cls, path: Path, identity: NativeSessionIdentity, + def retained(cls, path: Path, identity: NativeSessionSnapshot, revision: FileRevision) -> RetainedPiSessionFile: return RetainedPiSessionFile(str(path), identity, revision) @@ -114,6 +115,9 @@ def select_session(self, directory, execution, source_reads, *, participant): def history_source(self, root, incarnation): return self.transcript + def read_cut(self): + return self.transcript.cut() + def deal_delivery(self) -> SystemPromptDelivery: """Native Pi always launches with the prepared prompt (NativePiRpcLaunch.bootstrap).""" return SystemPromptDelivery(SYSTEM_PROMPT) @@ -140,7 +144,7 @@ def saved_model(self) -> ThreadModel: class RetainedPiSessionFile(PiSessionFile): """Original-coordinate history, with no selected native session or writable sidecars.""" - identity: NativeSessionIdentity + identity: NativeSessionSnapshot revision: FileRevision resumable: ClassVar[bool] = False @@ -152,6 +156,22 @@ def read_key(self) -> str: def transcript(self) -> NativeTranscript: return RetainedNativeTranscript(Path(self.path), self.identity, self.revision) + def read_revision(self, observed: FileObservation) -> FileObservation: + with self.transcript.reading(): + if observed != self.revision: + raise TranscriptChanged("Retained journal revision changed") + return observed + + def read_inode(self, revision: FileRevision) -> int: + return self.identity.revision.identity.inode + + def read_position(self, mark: int, revision: FileRevision) -> int: + # An original mark beyond this older image has already read every retained byte. + return min(mark, revision.size) + + def tool_watch_locations(self) -> tuple[RuntimeLocation, ...]: + return () + def require_forkable(self): raise NativePiUnavailable("Retained native history cannot fork a session") diff --git a/src/agent_comms/read_ledger.py b/src/agent_comms/read_ledger.py index 2bd28a7bea..cdcc04118d 100644 --- a/src/agent_comms/read_ledger.py +++ b/src/agent_comms/read_ledger.py @@ -256,7 +256,9 @@ def remove(document: ReadDocument) -> DocumentChange[ReadDocument, ReadDocument] @staticmethod def _transcript_key(viewer: str, source: str, inode: int) -> tuple[str, str, int]: - return viewer, str(Path(source).resolve()), inode + # RuntimeLocation already supplies the logical key. Resolving an archived + # name against today's filesystem can silently move an original read mark. + return viewer, source, inode def transcript_seen( self, viewer: str, source: str, inode: int, *, document: ReadDocument | None = None diff --git a/src/agent_comms/session_file.py b/src/agent_comms/session_file.py index 5e05c1c665..c93e17d4aa 100644 --- a/src/agent_comms/session_file.py +++ b/src/agent_comms/session_file.py @@ -10,7 +10,7 @@ from .errors import RelationViolationError from .nominal.field_codec import NullMember, PathText, RecordMember, SubclassUnion, WireTransient from .unstarted_runtime import UnstartedRuntime -from .native_session_reopen import NativeSessionIdentity +from .native_session_reopen import NativeSessionSnapshot from .private_path import FileRevision if TYPE_CHECKING: @@ -182,7 +182,7 @@ class RetainedSessionLocation(RecordMember, SessionFile): """ storage: BusSessionLocation - identity: NativeSessionIdentity + identity: NativeSessionSnapshot revision: FileRevision def bound_to(self, root: Path) -> BoundRetainedSession: diff --git a/src/agent_comms/team/tool_watch.py b/src/agent_comms/team/tool_watch.py index c9b8dc28d5..bfadae95ce 100644 --- a/src/agent_comms/team/tool_watch.py +++ b/src/agent_comms/team/tool_watch.py @@ -11,9 +11,9 @@ from .order import Team def thread_names(registry: Registration) -> dict[str, str]: - """Session file -> thread name, from the registry snapshot's threads.""" - return {key: name for name, thread in registry.snapshot().threads.items() - for key in thread.runtime.read_keys()} + """Live byte location -> thread name; retained history cannot become a live feed.""" + return {location.path: name for name, thread in registry.snapshot().threads.items() + for location in thread.runtime.tool_watch_locations()} def enforce_storage(ctx: Team, thread: str, call_id: str, diff --git a/src/agent_comms/transcript_routes.py b/src/agent_comms/transcript_routes.py index 29a749db49..a7191c39d1 100644 --- a/src/agent_comms/transcript_routes.py +++ b/src/agent_comms/transcript_routes.py @@ -259,11 +259,12 @@ def record_turn_publication(self, registry, log, lease, checkpoint, routing, pub thread = RegistryOwner.capture_local(initial, incarnation.name).thread if thread.turn_lease != lease: raise RelationViolationError("Native publication has no current original turn") - path = Path(thread.runtime.path) + runtime = thread.writable_runtime + path = Path(runtime.path) revision = FileRevision.capture(path).require( RelationViolationError("Native publication has no retained session")) reader = NativeTranscript(path) - floor = checkpoint.offset if thread.runtime.read_key == checkpoint.session_file else 0 + floor = checkpoint.offset if runtime.read_key == checkpoint.session_file else 0 try: final, user = reader.publication_input( after=floor, through=revision.size, @@ -272,7 +273,7 @@ def record_turn_publication(self, registry, log, lease, checkpoint, routing, pub ) except MalformedRecord as error: raise RelationViolationError("Native publication evidence is invalid") from error - ids = routing.annotation_ids(reader, checkpoint, thread.runtime, final.end) + ids = routing.annotation_ids(reader, checkpoint, runtime, final.end) inputs = InputDispositions(log.path.parent) lookup = incarnation.lookup with log.reading() as source, registry.store.current_snapshot() as cut: @@ -282,7 +283,7 @@ def record_turn_publication(self, registry, log, lease, checkpoint, routing, pub if current.turn_lease != lease: raise RelationViolationError("Native publication turn lease changed") if ( - current.runtime.read_key != thread.runtime.read_key + current.writable_runtime != runtime or FileRevision.capture(path) != revision ): raise RelationViolationError("Native publication session changed") @@ -309,7 +310,7 @@ def record_turn_publication(self, registry, log, lease, checkpoint, routing, pub tuple(item for item in originals if item.message.reference in published), lookup ) self._record_publication( - thread.runtime.read_key, ids, final.entry.id, routing, published + runtime.read_key, ids, final.entry.id, routing, published ) def _record_publication(self, session_file, entry_ids, final_id, routing, published): diff --git a/src/agent_comms/transcripts.py b/src/agent_comms/transcripts.py index 38e41ce5c7..ce1f06fec8 100644 --- a/src/agent_comms/transcripts.py +++ b/src/agent_comms/transcripts.py @@ -26,7 +26,7 @@ from .message_bus import MessageBus from .message_reference import MessageReference from .native_runtime_input import PublishedReplyRevision -from .native_transcript import NativeTranscript, RecordsCut, SessionCut +from .native_transcript import RecordsCut, SessionCut from .native_output import NativeRecordLocation from .notification_assignment import NotificationAssignment from agent_comms.message_notification import NotificationWindow @@ -238,8 +238,7 @@ def with_frontier(self, frontier: WireFrontier) -> TranscriptCursor: def mark_read(self, reads: TranscriptReadState, viewer: str, runtime: RuntimeLocation) -> None: if self.session_file != runtime.read_key: raise TranscriptChanged() - for key in runtime.read_keys(): - reads.mark_read(viewer, key, self.offset, NativeTranscript(Path(key)).cut()) + runtime.mark_read(reads, viewer, self.offset) @dataclass(frozen=True) diff --git a/src/agent_comms/unstarted_runtime.py b/src/agent_comms/unstarted_runtime.py index ac3332a593..5c5530cff6 100644 --- a/src/agent_comms/unstarted_runtime.py +++ b/src/agent_comms/unstarted_runtime.py @@ -37,7 +37,7 @@ def new_location(self, root, worktree): def arguments(self): return {} - def read_keys(self): + def read_locations(self): return () def transcript_cursor(self, frontier, offset, *, source): diff --git a/src/agent_comms/view_unread.py b/src/agent_comms/view_unread.py index ef8d9cc32b..5350c205c8 100644 --- a/src/agent_comms/view_unread.py +++ b/src/agent_comms/view_unread.py @@ -16,7 +16,7 @@ from weakref import WeakValueDictionary, finalize from .agent_backend import RuntimeLocation -from .coordination_errors import CoordinationReadUnavailable +from .coordination_errors import CoordinationReadUnavailable, TranscriptChanged from .locked_store import PublishedSQLiteFile, RuntimeFile, SQLiteFile from .store_files import StoreBusy from .bus import Bus @@ -98,31 +98,35 @@ def counted(self, total: int, seen: int) -> Counted: """Coverage, not the number found, determines whether this is a lower bound.""" return CappedCount(total) if self.start > seen else Counted(total) - def prepare(self, revision: FileRevision, seen: int, cancelled: Event) -> ReplyIndexResult: + def prepare(self, location: RuntimeLocation, revision: FileRevision, + seen: int, cancelled: Event) -> ReplyIndexResult: """Visit at most limit newest records, stopping earlier at the read mark. An append visits only the appended records and keeps the indexed prefix, so a growing transcript never re-reads records it already indexed.""" - with Path(self.source).open("rb") as stream: + with Path(location.path).open("rb") as stream: if not revision.opened_by(stream): return RevisionMoved() if self.continues(revision): - return self._scan(revision, seen, self.start, self.records, cancelled, ContinuedReplyIndex) - fresh = partial(self._scan, revision, seen, revision.size, 0, cancelled, ReplyIndexUpdate) + return self._scan(location, revision, seen, self.start, self.records, cancelled, ContinuedReplyIndex) + fresh = partial(self._scan, location, revision, seen, revision.size, 0, cancelled, ReplyIndexUpdate) for cut in self.appended_cut(revision): # Only appended: visit the new records, keep the indexed prefix and replies. boundary = max(seen, cut) return self._scan( - revision, boundary, revision.size, 0, cancelled, ContinuedReplyIndex, + location, revision, boundary, revision.size, 0, cancelled, ContinuedReplyIndex, ).extending(self, boundary, fresh) return fresh() - def _scan(self, revision: FileRevision, seen: int, start: int, visited: int, - cancelled: Event, result: type[ReplyIndexUpdate]) -> ReplyIndexResult: + def _scan(self, location: RuntimeLocation, revision: FileRevision, seen: int, + start: int, visited: int, cancelled: Event, + result: type[ReplyIndexUpdate]) -> ReplyIndexResult: replies: list[TranscriptReply] = [] - source = Path(self.source) + source = Path(location.path) with source.open("rb") as stream, closing( reverse_records(source, start, floor=max(0, seen - 1), stream=stream) ) as records: + if not revision.opened_by(stream): + return RevisionMoved() complete = record_end(stream, revision.size) for begin, end, raw in records: if visited >= self.limit or start <= seen: @@ -146,11 +150,16 @@ class ReplyRead: index: ReplyIndex revision: FileRevision seen: int + location: RuntimeLocation + + def current_revision(self) -> FileObservation: + return self.location.read_revision(FileRevision.capture(Path(self.location.path))) def prepare(self, cancelled: Event) -> ReplyIndexResult: """One independently scheduled source read; no loop over other people's files.""" try: - return self.index.prepare(self.revision, self.seen, cancelled) + self.location.read_revision(self.revision) + return self.index.prepare(self.location, self.revision, self.seen, cancelled) except FileNotFoundError: return RevisionMoved() @@ -367,7 +376,7 @@ def commit(self, database: sqlite3.Connection) -> None: if self.future.cancelled(): return expected = self.read.revision - revision = FileRevision.capture(Path(self.read.index.source)) + revision = self.read.current_revision() if revision.contains_cut(expected, expected.size): self.future.result().commit(database) @@ -445,10 +454,11 @@ def capture(cls, thread: Thread): return cls(thread.incarnation, thread.runtime, {}, {}).refresh({}) def refresh(self, observed: Mapping[str, FileObservation]) -> TranscriptReadKeys: - paths = {key: self.paths[key] if key in self.paths else Path(key).absolute() - for key in self.runtime.read_keys()} + locations = {location.read_key: location for location in self.runtime.read_locations()} + paths = {key: Path(location.path).absolute() for key, location in locations.items()} return replace(self, paths=paths, revisions={ - key: observed[key] if key in observed else FileRevision.capture(path) + key: locations[key].read_revision( + observed[key] if key in observed else FileRevision.capture(path)) for key, path in paths.items() }) @@ -575,7 +585,8 @@ def remove_threads(self, threads: Sequence[Thread]) -> None: @contextmanager def _indexed_sources( - self, revisions: dict[str, FileObservation], seen: Mapping[str, int] + self, revisions: dict[str, FileObservation], seen: Mapping[str, int], + locations: Mapping[str, RuntimeLocation], ) -> Iterator[tuple[sqlite3.Connection, dict[str, ReplyIndex]]]: database = self._database() # Publish before reading committed counts, if the preparation has finished. @@ -584,7 +595,7 @@ def _indexed_sources( self.begin_snapshot(database) indexes = ReplyIndex.for_sources(database, tuple(revisions)) needed = { - source: ReplyRead(indexes[source], revision, seen[source]) + source: ReplyRead(indexes[source], revision, seen[source], locations[source]) for source, observed in revisions.items() for revision in observed.revisions() if not indexes[source].counts(revision, seen[source]) @@ -604,7 +615,9 @@ def _refresh_keys(self, snapshot: RegistrySnapshot) -> None: prior = self._thread_keys.get(thread.name) keys = (prior if prior is not None and prior.matches(thread) else TranscriptReadKeys.capture(thread)) - rebuild |= prior is None or prior.revisions.keys() != keys.revisions.keys() + rebuild |= prior is None or prior.paths != keys.paths + if prior is not None and prior.runtime != keys.runtime: + self._preparation.retire(prior.revisions.keys()) current[thread.name] = keys rebuild |= bool(self._thread_keys.keys() - current.keys()) self._thread_keys = current @@ -665,6 +678,13 @@ def _counts_locked(self, viewer: str, document: ReadDocument, threads: Mapping[str, TranscriptReadKeys]) -> TranscriptUnread: revisions = {source: revision for keys in threads.values() for source, revision in keys.revisions.items()} + locations: dict[str, RuntimeLocation] = {} + for keys in threads.values(): + for location in keys.runtime.read_locations(): + key = location.read_key + if key in locations and locations[key] != location: + raise TranscriptChanged("Logical source has conflicting read bindings") + locations[key] = location if not revisions: return TranscriptUnread({name: ReplyUnread(0) for name in threads}) # A missing transcript has no replies. @@ -674,10 +694,10 @@ def _counts_locked(self, viewer: str, document: ReadDocument, seen = {} for source, revision in present.items(): mark = self.reads.transcript_seen( - viewer, source, revision.identity.inode, document=document + viewer, source, locations[source].read_inode(revision), document=document ) - seen[source] = mark if mark <= revision.size else 0 - with self._indexed_sources(revisions, seen) as (database, indexes): + seen[source] = locations[source].read_position(mark, revision) + with self._indexed_sources(revisions, seen, locations) as (database, indexes): ends: dict[str, int] = {} for source, revision in present.items(): if seen[source] == revision.size: diff --git a/tests/test_native_transcript.py b/tests/test_native_transcript.py index da1a2dc37e..e93f24ca6a 100644 --- a/tests/test_native_transcript.py +++ b/tests/test_native_transcript.py @@ -3,6 +3,7 @@ import json from dataclasses import fields, replace from pathlib import Path +from shutil import copyfile import pytest @@ -12,7 +13,7 @@ MalformedRecord, ToolCallId, ) -from agent_comms.native_entries import MessageEntry, NativeEntry, TranscriptProjection, UnknownEntry +from agent_comms.native_entries import MessageEntry, NativeEntry, NativeEvidenceScope, TranscriptProjection, UnknownEntry from agent_comms.native_transcript import NativeTranscript, RetainedNativeTranscript from agent_comms.coordination_errors import TranscriptChanged from agent_comms.errors import RelationViolationError @@ -23,7 +24,7 @@ from agent_comms.threads import Thread from agent_comms.session_file import BusSessionLocation, RetainedSessionLocation, SessionFile from agent_comms.native_input_proof import NativeContextProof -from agent_comms.native_session_reopen import NativeSessionIdentity, PiSessionId +from agent_comms.native_session_reopen import NativeSessionIdentity, NativeSessionSnapshot, PiSessionId from agent_comms.pi_payloads import UserMessage from agent_comms.transcript_events import ( AssistantTranscript, @@ -35,7 +36,12 @@ UserTranscript, ) from agent_comms.turn_item import FreshNativeId, NativeItemId -from conftest import transcript_records +from conftest import installed_comms, transcript_records +from agent_comms.comms import wire +from agent_comms.transcripts import NativeTranscriptCursor +from agent_comms.ui_model.unread import ReplyUnread +from agent_comms.view_unread import TranscriptReadKeys +from native_unread_fixture import completed_counts STAMPED = 1790812800.0 # "2026-10-01T00:00:00.000Z", the time every fixture row records @@ -47,10 +53,10 @@ def encoded(value): @pytest.fixture def retained_native_location(tmp_path: Path) -> tuple[Path, RetainedSessionLocation]: session = PiSessionId.fresh() - identity = NativeSessionIdentity(session, str(tmp_path / "original" / "session.jsonl")) - storage = tmp_path / "retained" / "session.jsonl" - storage.parent.mkdir() - storage.write_bytes( + root = installed_comms(tmp_path / "wire").root + original = tmp_path / "original" / "session.jsonl" + original.parent.mkdir(mode=0o700) + original.write_bytes( encoded(dict(type="session", version=3, id=session, cwd=str(tmp_path), timestamp="2026-10-01T00:00:00.000Z")) + encoded(dict(type="message", id="original-entry", parentId=None, @@ -58,11 +64,20 @@ def retained_native_location(tmp_path: Path) -> tuple[Path, RetainedSessionLocat message=dict(role="assistant", content=[dict(type="text", text="saved")], stopReason="stop"))) ) + original.chmod(0o600) + with NativeEvidenceScope() as scope: + identity = NativeSessionSnapshot.from_evidence(scope.for_source(original)) + storage = root / "retained" / "session.jsonl" + storage.parent.mkdir() + copyfile(original, storage) + storage.chmod(0o600) revision = FileRevision.capture(storage).require(ValueError("Missing retained fixture")) + assert revision.identity != identity.revision.identity + original.unlink() location = RetainedSessionLocation( - BusSessionLocation(storage.relative_to(tmp_path)), identity, revision, + BusSessionLocation(storage.relative_to(root)), identity, revision, ) - return tmp_path, location + return root, location def test_retained_location_roundtrip_keeps_original_keys_and_reads_retained_bytes( @@ -84,6 +99,7 @@ def test_retained_location_roundtrip_keeps_original_keys_and_reads_retained_byte assert records[-1].events[0].text == "saved" assert not runtime.resumable assert runtime.storage_files() == () + assert runtime.tool_watch_locations() == () with pytest.raises(RelationViolationError): restored.writable_location(PiNativeBackend) with pytest.raises(NativePiUnavailable): @@ -117,6 +133,65 @@ def test_retained_reader_rejects_changed_bytes_during_its_read( reader.cut() +@pytest.mark.parametrize("already_read", [False, True]) +def test_retained_marks_and_unread_index_keep_the_original_inode( + retained_native_location: tuple[Path, RetainedSessionLocation], already_read: bool, +) -> None: + root, location = retained_native_location + comms = wire(root) + comms.registry.declare(Thread("reader", frozenset(), Checkout(root), session_file=location)) + runtime = comms.registry.require("reader").runtime + state = comms.views.transcript_reads + key = location.identity.session_file + original_inode = location.identity.revision.identity.inode + size = location.identity.revision.size + if already_read: + state.reads.mark_transcript("viewer", key, original_inode, size, size) + assert completed_counts(state, "viewer").counts["reader"] == ReplyUnread(0 if already_read else 1) + cut = runtime.read_cut() + assert cut.revision == location.identity.revision + NativeTranscriptCursor(key, cut.offset).mark_read(state, "viewer", runtime) + assert state.reads.transcript_seen("viewer", key, original_inode) == cut.offset + assert state.reads.transcript_seen("viewer", key, location.revision.identity.inode) == 0 + assert completed_counts(state, "viewer").counts["reader"] == ReplyUnread(0) + assert runtime.read_position(size + 1, location.revision) == size + + +def test_original_read_key_does_not_follow_a_recreated_path_alias( + retained_native_location: tuple[Path, RetainedSessionLocation], +) -> None: + root, location = retained_native_location + state = wire(root).views.transcript_reads + runtime = location.bound_to(root).location(PiNativeBackend) + key = runtime.read_key + cut = runtime.read_cut() + NativeTranscriptCursor(key, cut.offset).mark_read(state, "viewer", runtime) + directory = location.identity.path.parent + directory.rmdir() + replacement = root / "different-original" + replacement.mkdir() + directory.symlink_to(replacement, target_is_directory=True) + assert str(location.identity.path.resolve()) != key + assert state.reads.transcript_seen("viewer", key, location.identity.revision.identity.inode) == cut.offset + assert runtime.read_cut() == cut + + +def test_retained_unread_observation_watches_physical_storage_and_rejects_replacement( + retained_native_location: tuple[Path, RetainedSessionLocation], +) -> None: + root, location = retained_native_location + comms = wire(root) + comms.registry.declare(Thread("reader", frozenset(), Checkout(root), session_file=location)) + keys = TranscriptReadKeys.capture(comms.registry.require("reader")) + path = location.storage.resolve(root) + assert keys.paths == {location.identity.session_file: path} + replacement = root / "replacement.jsonl" + copyfile(path, replacement) + replacement.replace(path) + with pytest.raises(TranscriptChanged): + keys.refresh({}) + + def test_native_discriminator_is_separate_from_normalized_family_codec(): entry = NativeEntry.read( encoded( diff --git a/tests/test_thread_unread.py b/tests/test_thread_unread.py index 7012f0d8c3..3a9642e20c 100644 --- a/tests/test_thread_unread.py +++ b/tests/test_thread_unread.py @@ -11,6 +11,7 @@ from agent_comms.comms import wire from conftest import installed_comms from agent_comms.private_path import FileRevision +from agent_comms.pi_native_backend import PiSessionFile from agent_comms.registry_document import DEFAULT_ROSTER from agent_comms.ui_model.unread import CappedReplyUnread, EmptyCappedReplyUnread, ReplyUnread from agent_comms.view_unread import ( @@ -79,7 +80,7 @@ def test_prepare_caps_records_not_replies_and_detects_replacement(tmp_path): append(source, 'toolResult', 'not an assistant reply', toolCallId='tool') revision = FileRevision.capture(source) index = ReplyIndex(source=str(source)) - update = index.prepare(revision, 0, get_context('spawn').Event()) + update = index.prepare(PiSessionFile(str(source)), revision, 0, get_context('spawn').Event()) assert isinstance(update, ReplyIndexUpdate) assert update.after.records == ReplyIndex.limit assert update.after.start > 0 @@ -91,7 +92,7 @@ def test_prepare_caps_records_not_replies_and_detects_replacement(tmp_path): replacement = tmp_path / 'replacement' session(replacement, 'replacement') replacement.replace(source) - assert isinstance(index.prepare(revision, 0, get_context('spawn').Event()), RevisionMoved) + assert isinstance(index.prepare(PiSessionFile(str(source)), revision, 0, get_context('spawn').Event()), RevisionMoved) def test_reply_counts_are_persistent_human_read_positions(tmp_path): @@ -147,7 +148,8 @@ def test_completed_source_publishes_while_another_source_is_still_pending(tmp_pa append(second) state = comms.views.transcript_reads database = state._database() - reads = {str(path): ReplyRead(ReplyIndex(source=str(path)), FileRevision.capture(path), 0) + reads = {str(path): ReplyRead(ReplyIndex(source=str(path)), FileRevision.capture(path), 0, + PiSessionFile(str(path))) for path in (source, second)} futures = {name: Future() for name in reads} preparation = NativeReplyPreparation({name: ReplyReading(read, futures[name]) for name, read in reads.items()}) @@ -156,13 +158,13 @@ def test_completed_source_publishes_while_another_source_is_still_pending(tmp_pa subscribed = state.watch(frozenset(), observed.append) assert subscribed == frozenset(futures.values()) first = reads[str(source)] - futures[str(source)].set_result(first.index.prepare(first.revision, 0, get_context('spawn').Event())) + futures[str(source)].set_result(first.prepare(get_context('spawn').Event())) state._preparation = preparation.finished(state, database, reads) assert not futures[str(second)].done() assert observed == [futures[str(source)]] assert TranscriptReply.after(database, {str(source): 0, str(second): 0}) == {str(source): 1, str(second): 0} last = reads[str(second)] - futures[str(second)].set_result(last.index.prepare(last.revision, 0, get_context('spawn').Event())) + futures[str(second)].set_result(last.prepare(get_context('spawn').Event())) state._preparation = preparation.finished(state, database, reads) assert state.preparation is state._idle assert observed == list(futures.values()) diff --git a/tests/test_watched.py b/tests/test_watched.py index 824938f392..ed02e1ddee 100644 --- a/tests/test_watched.py +++ b/tests/test_watched.py @@ -22,6 +22,7 @@ from agent_comms.messages import InfoKind, Message from agent_comms.watched import ChangeSignal, FileLifetimeSource, FileSource from agent_comms.private_path import FileRevision +from agent_comms.pi_native_backend import PiSessionFile from agent_comms.presentation import WireRevision from agent_comms.view_unread import ( NativeReplyPreparation, ReplyIndex, ReplyIndexUpdate, ReplyRead, ReplyReading, @@ -259,7 +260,7 @@ def test_native_reply_publication_uses_the_index_commit_source(tmp_path): database = state._database() before = ReplyIndex(source=str(transcript)) revision = FileRevision.capture(transcript).require(ValueError("Missing transcript")) - read = ReplyRead(before, revision, 0) + read = ReplyRead(before, revision, 0, PiSessionFile(str(transcript))) prepared = Future() prepared.set_result(ReplyIndexUpdate(before, replace(before, revision=revision), ())) preparation = NativeReplyPreparation({str(transcript): ReplyReading(read, prepared)}) From e56fe2ec910342114176895d097b5cd2ad93bfe8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 3 Nov 2026 17:07:09 -0500 Subject: [PATCH 3/7] Bind retained transcript reads to their opened descriptors --- src/agent_comms/native_transcript.py | 93 +++++++++++++++++----------- tests/test_native_transcript.py | 71 +++++++++++++++++++++ 2 files changed, 129 insertions(+), 35 deletions(-) diff --git a/src/agent_comms/native_transcript.py b/src/agent_comms/native_transcript.py index a6b947a69d..716ef4d7b5 100644 --- a/src/agent_comms/native_transcript.py +++ b/src/agent_comms/native_transcript.py @@ -38,7 +38,7 @@ ) from .native_runtime_input import NativeRuntimeInput, PublishedReplyRevision from .native_session_reopen import NativeSessionIdentity, NativeSessionSnapshot -from .private_path import FileRevision +from .private_path import FileObservation, FileRevision from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime from .transcript_sources import TranscriptRecord, TranscriptSource from .turn_item import FreshNativeId, NativeItemId @@ -280,6 +280,20 @@ def reading(self): """Keep the native record owner bound across one page projection.""" yield self + def require_opened(self, stream: BinaryIO) -> None: + """Ordinary journals may append; retained members bind the actual descriptor.""" + pass + + @contextmanager + def opened(self) -> Generator[BinaryIO]: + """Own every native file open and fence the descriptor consumers actually read.""" + with self.reading(), self.path.open("rb") as stream: + self.require_opened(stream) + try: + yield stream + finally: + self.require_opened(stream) + def imported_sources(self): """Read original metadata before conversation, including native configuration. @@ -287,16 +301,15 @@ def imported_sources(self): producer may also prepend model/settings records. Stop at conversation; never scan the remaining history or claim this was a model request. """ - with self.reading(): - revision = FileRevision.from_stat(self.path.stat()) - with self.path.open("rb") as stream: - SessionEntry.read(stream.readline()) - sources = [] - for raw in stream: - entry = NativeEntry.read(raw) - if entry.is_message: - break - sources.extend(entry.imported_sources()) + with self.opened() as stream: + revision = FileRevision.from_file(stream) + SessionEntry.read(stream.readline()) + sources = [] + for raw in stream: + entry = NativeEntry.read(raw) + if entry.is_message: + break + sources.extend(entry.imported_sources()) if FileRevision.from_stat(self.path.stat()) != revision: raise ValueError("Imported context saved source changed while reading") return tuple(sources) @@ -306,19 +319,18 @@ def session_cut(self, revision: FileRevision, offset: int) -> PresentSessionCut: def cut(self) -> SessionCut: try: - stream = self.path.open("rb") + # One bound descriptor supplies the revision and last record boundary. + with self.opened() as stream: + revision = FileRevision.from_file(stream) + return self.session_cut(revision, record_end(stream, revision.size)) except FileNotFoundError: return MissingSessionCut() - # One open file supplies both the revision and its last record boundary. - with stream: - revision = FileRevision.from_file(stream) - return self.session_cut(revision, record_end(stream, revision.size)) def committed_output_cut(self, location: NativeRecordLocation) -> SessionCut: """Corroborate one indexed output against its original header and row.""" if self.source_path.absolute() != location.path: raise TranscriptChanged() - with self.reading(), self.path.open("rb") as stream: + with self.opened() as stream: revision = FileRevision.from_file(stream) header = SessionEntry.read(stream.readline()) if header.id != location.session_id or revision.size < location.session_entry_end: @@ -372,7 +384,7 @@ def recorded_contexts(self, contexts, entry_ids): @property def session_id(self): - with self.path.open("rb") as stream: + with self.opened() as stream: try: entry = SessionEntry.read(stream.readline()) except MalformedRecord: @@ -398,7 +410,7 @@ def require_publication_context( this bounded indexed lookup does not guess a request kind or options. """ files.require_private_directory(self.path.parent) - with self.path.open("rb") as stream: + with self.opened() as stream: header = SessionEntry.read(stream.readline()) with NativeContextJournal.open_evidence(files.input_proof) as db: try: @@ -484,16 +496,16 @@ def voice(self, deal: Speaking) -> Voice: return deal def tail(self, *, max_bytes: int | None = None): - with self.reading(): - end = self.path.stat().st_size + with self.opened() as stream: + end = FileRevision.from_file(stream).size floor = max(0, end - max_bytes) if max_bytes is not None else 0 - for _, _, raw in reverse_records(self.path, end, floor=floor): + for _, _, raw in reverse_records(self.path, end, floor=floor, stream=stream): # Skip only the record still being appended; corrupt history raises. if raw.endswith(b"\n"): yield NativeEntry.read(raw) def forward(self, after: int, through: int): - with self.reading(), self.path.open("rb") as stream: + with self.opened() as stream: stream.seek(after) while stream.tell() < through: start = stream.tell() @@ -503,8 +515,8 @@ def forward(self, after: int, through: int): yield NativeRecord.read(start, stream.tell(), raw) def reverse(self, before: int, *, floor: int = 0) -> Generator[NativeRecord]: - with self.reading(): - for start, end, raw in reverse_records(self.path, before, floor=floor): + with self.opened() as stream: + for start, end, raw in reverse_records(self.path, before, floor=floor, stream=stream): yield NativeRecord.read(start, end, raw) @@ -520,7 +532,8 @@ def __init__(self, path: Path, identity: NativeSessionSnapshot, revision: FileRe self.original = identity self.revision = revision identity.require_retained(revision) - self.require_retained() + with self.opened(): + pass @property def source_path(self) -> Path: @@ -530,11 +543,25 @@ def source_path(self) -> Path: def session_id(self): return self.original.session_id - def require_retained(self) -> None: - if (FileRevision.capture(self.path) != self.revision - or super().session_id != self.original.session_id): + def require_revision(self, revision: FileObservation) -> None: + if revision != self.revision: raise TranscriptChanged("Retained native journal differs from its binding") + def require_retained(self) -> None: + self.require_revision(FileRevision.capture(self.path)) + + def require_opened(self, stream: BinaryIO) -> None: + self.require_revision(FileRevision.from_file(stream)) + position = stream.tell() + try: + stream.seek(0) + if SessionEntry.read(stream.readline()).id != self.original.session_id: + raise TranscriptChanged("Retained native journal header differs from its binding") + except MalformedRecord as error: + raise TranscriptChanged("Retained native journal header differs from its binding") from error + finally: + stream.seek(position) + @contextmanager def reading(self): self.require_retained() @@ -544,10 +571,6 @@ def reading(self): self.require_retained() def session_cut(self, revision: FileRevision, offset: int) -> PresentSessionCut: - # The physical revision is fenced by reading(); positions and marks still - # belong to the original acquired inode, including committed-output cuts. + # Translation is legal only after the actual opened descriptor was checked. + self.require_revision(revision) return PresentSessionCut(self.original.revision, offset) - - def cut(self) -> SessionCut: - with self.reading(): - return super().cut() diff --git a/tests/test_native_transcript.py b/tests/test_native_transcript.py index e93f24ca6a..ca4b2dea27 100644 --- a/tests/test_native_transcript.py +++ b/tests/test_native_transcript.py @@ -1,9 +1,11 @@ """Saved Pi boundary, bounded readers and presentation roundtrips.""" import json +from collections.abc import Callable from dataclasses import fields, replace from pathlib import Path from shutil import copyfile +from typing import BinaryIO, Literal import pytest @@ -15,6 +17,7 @@ ) from agent_comms.native_entries import MessageEntry, NativeEntry, NativeEvidenceScope, TranscriptProjection, UnknownEntry from agent_comms.native_transcript import NativeTranscript, RetainedNativeTranscript +from agent_comms.native_output import NativeRecordLocation from agent_comms.coordination_errors import TranscriptChanged from agent_comms.errors import RelationViolationError from agent_comms.native_failures import NativePiUnavailable @@ -133,6 +136,74 @@ def test_retained_reader_rejects_changed_bytes_during_its_read( reader.cut() +@pytest.mark.parametrize("read", [ + pytest.param(lambda reader, size, output: reader.cut(), id="cut"), + pytest.param(lambda reader, size, output: reader.committed_output_cut(output), id="committed-cut"), + pytest.param(lambda reader, size, output: reader.imported_sources(), id="imported-sources"), + pytest.param(lambda reader, size, output: tuple(reader.tail()), id="tail"), + pytest.param(lambda reader, size, output: tuple(reader.forward(0, size)), id="forward"), + pytest.param(lambda reader, size, output: tuple(reader.reverse(size)), id="reverse"), +]) +def test_retained_reader_rejects_descriptor_from_directory_aba( + retained_native_location: tuple[Path, RetainedSessionLocation], + monkeypatch: pytest.MonkeyPatch, + read: Callable[[NativeTranscript, int, NativeRecordLocation], object], +) -> None: + root, location = retained_native_location + path = location.storage.resolve(root) + reader = RetainedNativeTranscript(path, location.identity, location.revision) + size = reader.cut().offset + last = tuple(reader.forward(0, size))[-1] + output = NativeRecordLocation( + session_id=location.identity.session_id, + session_file=location.identity.session_file, + session_entry_id=NativeItemId("original-entry"), + session_entry_start=last.start, + session_entry_end=last.end, + ) + replacement = root / "other-retained" + replacement.mkdir() + other = replacement / path.name + other.write_bytes(path.read_bytes().replace(b'"saved"', b'"other"')) + parked = root / "parked-retained" + physical_open = Path.open + opened: list[FileRevision] = [] + + def open_after_directory_swap(candidate: Path, mode: Literal["rb"]) -> BinaryIO: + if candidate != path: + return physical_open(candidate, mode) + # The foreign IO boundary opens real B, then restores A before returning + # the descriptor. No product reader or revision method is replaced. + path.parent.rename(parked) + replacement.rename(path.parent) + try: + stream = physical_open(candidate, mode) + opened.append(FileRevision.from_file(stream)) + finally: + path.parent.rename(replacement) + parked.rename(path.parent) + return stream + + monkeypatch.setattr(Path, "open", open_after_directory_swap) + with pytest.raises(TranscriptChanged): + read(reader, size, output) + assert len(opened) == 1 + assert opened[0].identity != location.revision.identity + assert FileRevision.capture(path) == location.revision + + +def test_retained_cut_validates_descriptor_before_translating_original_coordinates( + retained_native_location: tuple[Path, RetainedSessionLocation], +) -> None: + root, location = retained_native_location + path = location.storage.resolve(root) + reader = RetainedNativeTranscript(path, location.identity, location.revision) + other = root / "same-header.jsonl" + copyfile(path, other) + with other.open("rb") as stream, pytest.raises(TranscriptChanged): + reader.session_cut(FileRevision.from_file(stream), location.revision.size) + + @pytest.mark.parametrize("already_read", [False, True]) def test_retained_marks_and_unread_index_keep_the_original_inode( retained_native_location: tuple[Path, RetainedSessionLocation], already_read: bool, From a5ba12271b8f2044e5a098ac8f15a2144f044303 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 3 Nov 2026 17:30:43 -0500 Subject: [PATCH 4/7] Refuse missing acquisitions of bound retained journals --- src/agent_comms/native_transcript.py | 11 +++++- tests/test_native_transcript.py | 56 +++++++++++++++++++++++++++- 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/native_transcript.py b/src/agent_comms/native_transcript.py index 716ef4d7b5..2f9191ce9d 100644 --- a/src/agent_comms/native_transcript.py +++ b/src/agent_comms/native_transcript.py @@ -317,14 +317,18 @@ def imported_sources(self): def session_cut(self, revision: FileRevision, offset: int) -> PresentSessionCut: return PresentSessionCut(revision, offset) + def missing_cut(self, error: FileNotFoundError) -> SessionCut: + """An ordinary journal can belong to a session that has not written it yet.""" + return MissingSessionCut() + def cut(self) -> SessionCut: try: # One bound descriptor supplies the revision and last record boundary. with self.opened() as stream: revision = FileRevision.from_file(stream) return self.session_cut(revision, record_end(stream, revision.size)) - except FileNotFoundError: - return MissingSessionCut() + except FileNotFoundError as error: + return self.missing_cut(error) def committed_output_cut(self, location: NativeRecordLocation) -> SessionCut: """Corroborate one indexed output against its original header and row.""" @@ -550,6 +554,9 @@ def require_revision(self, revision: FileObservation) -> None: def require_retained(self) -> None: self.require_revision(FileRevision.capture(self.path)) + def missing_cut(self, error: FileNotFoundError) -> SessionCut: + raise TranscriptChanged("Bound retained native journal could not be opened") from error + def require_opened(self, stream: BinaryIO) -> None: self.require_revision(FileRevision.from_file(stream)) position = stream.tell() diff --git a/tests/test_native_transcript.py b/tests/test_native_transcript.py index ca4b2dea27..e699f3aa26 100644 --- a/tests/test_native_transcript.py +++ b/tests/test_native_transcript.py @@ -18,6 +18,7 @@ from agent_comms.native_entries import MessageEntry, NativeEntry, NativeEvidenceScope, TranscriptProjection, UnknownEntry from agent_comms.native_transcript import NativeTranscript, RetainedNativeTranscript from agent_comms.native_output import NativeRecordLocation +from agent_comms.agent_backend import RuntimeLocation from agent_comms.coordination_errors import TranscriptChanged from agent_comms.errors import RelationViolationError from agent_comms.native_failures import NativePiUnavailable @@ -43,7 +44,7 @@ from agent_comms.comms import wire from agent_comms.transcripts import NativeTranscriptCursor from agent_comms.ui_model.unread import ReplyUnread -from agent_comms.view_unread import TranscriptReadKeys +from agent_comms.view_unread import TranscriptReadKeys, TranscriptReadState from native_unread_fixture import completed_counts STAMPED = 1790812800.0 # "2026-10-01T00:00:00.000Z", the time every fixture row records @@ -204,6 +205,59 @@ def test_retained_cut_validates_descriptor_before_translating_original_coordinat reader.session_cut(FileRevision.from_file(stream), location.revision.size) +@pytest.mark.parametrize("read", [ + pytest.param(lambda runtime, reads, size: runtime.read_cut(), id="cut"), + pytest.param( + lambda runtime, reads, size: NativeTranscriptCursor(runtime.read_key, size).mark_read( + reads, "viewer", runtime), + id="nonzero-acknowledgement", + ), +]) +def test_retained_failed_cut_acquisition_refuses_after_parent_returns( + retained_native_location: tuple[Path, RetainedSessionLocation], + monkeypatch: pytest.MonkeyPatch, + read: Callable[[RuntimeLocation, TranscriptReadState, int], object], +) -> None: + root, location = retained_native_location + runtime = location.bound_to(root).location(PiNativeBackend) + reads = wire(root).views.transcript_reads + size = runtime.read_cut().offset + assert size > 0 + path = location.storage.resolve(root) + parked = root / "temporarily-unavailable" + physical_open = Path.open + attempts: list[Path] = [] + + def failed_second_open(candidate: Path, mode: Literal["rb"]) -> BinaryIO: + if candidate != path: + return physical_open(candidate, mode) + attempts.append(candidate) + # Construction opens A successfully. Its subsequent cut cannot acquire + # any descriptor; A returns before the reader's final pathname fence. + if len(attempts) == 2: + path.parent.rename(parked) + try: + return physical_open(candidate, mode) + finally: + parked.rename(path.parent) + return physical_open(candidate, mode) + + monkeypatch.setattr(Path, "open", failed_second_open) + with pytest.raises(TranscriptChanged) as refused: + read(runtime, reads, size) + assert isinstance(refused.value.__cause__, FileNotFoundError) + assert len(attempts) == 2 + assert FileRevision.capture(path) == location.revision + assert reads.reads.transcript_seen( + "viewer", runtime.read_key, location.identity.revision.identity.inode) == 0 + + +def test_unstarted_native_journal_may_have_an_empty_cut(tmp_path: Path) -> None: + cut = NativeTranscript(tmp_path / "not-written-yet.jsonl").cut() + assert cut.offset == 0 + assert cut.holds(cut, 0) + + @pytest.mark.parametrize("already_read", [False, True]) def test_retained_marks_and_unread_index_keep_the_original_inode( retained_native_location: tuple[Path, RetainedSessionLocation], already_read: bool, From a287ed0c01e5e18c6565ff859f251814fe2ca8d3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 3 Nov 2026 20:44:50 -0500 Subject: [PATCH 5/7] Use the declared transcript reload exception at every refusal --- src/agent_comms/native_transcript.py | 12 ++++++------ src/agent_comms/pi_native_backend.py | 2 +- src/agent_comms/view_unread.py | 2 +- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/agent_comms/native_transcript.py b/src/agent_comms/native_transcript.py index 2f9191ce9d..3502d5503d 100644 --- a/src/agent_comms/native_transcript.py +++ b/src/agent_comms/native_transcript.py @@ -245,7 +245,7 @@ def completed_input(self, context: NativeContextProof) -> bool: with closing(self.forward(context.session_entry_start, cut.offset)) as records: anchor = next(records) if not anchor.project(lambda original: original.entry.require_entry_id() == context.session_entry_id): - raise TranscriptChanged("Native answer input changed after its proof was read") + raise TranscriptChanged() own_turn = takewhile( lambda record: not record.project(lambda original: original.entry.input_boundary), records) finals = tuple(record for record in own_turn if record.project(lambda original: @@ -253,7 +253,7 @@ def completed_input(self, context: NativeContextProof) -> bool: text for event in original.events for text in event.assistant_parts()))) answered = anchor in self.inputs_for(finals, floor=context.session_entry_start) if not cut.holds(self.cut(), cut.offset): - raise TranscriptChanged("Native answer source changed during its read") + raise TranscriptChanged() return answered def _tool_call_times(self, since: float = -inf) -> Generator[float]: @@ -549,13 +549,13 @@ def session_id(self): def require_revision(self, revision: FileObservation) -> None: if revision != self.revision: - raise TranscriptChanged("Retained native journal differs from its binding") + raise TranscriptChanged() def require_retained(self) -> None: self.require_revision(FileRevision.capture(self.path)) def missing_cut(self, error: FileNotFoundError) -> SessionCut: - raise TranscriptChanged("Bound retained native journal could not be opened") from error + raise TranscriptChanged() from error def require_opened(self, stream: BinaryIO) -> None: self.require_revision(FileRevision.from_file(stream)) @@ -563,9 +563,9 @@ def require_opened(self, stream: BinaryIO) -> None: try: stream.seek(0) if SessionEntry.read(stream.readline()).id != self.original.session_id: - raise TranscriptChanged("Retained native journal header differs from its binding") + raise TranscriptChanged() except MalformedRecord as error: - raise TranscriptChanged("Retained native journal header differs from its binding") from error + raise TranscriptChanged() from error finally: stream.seek(position) diff --git a/src/agent_comms/pi_native_backend.py b/src/agent_comms/pi_native_backend.py index ce3b19c5b0..27199b8678 100644 --- a/src/agent_comms/pi_native_backend.py +++ b/src/agent_comms/pi_native_backend.py @@ -159,7 +159,7 @@ def transcript(self) -> NativeTranscript: def read_revision(self, observed: FileObservation) -> FileObservation: with self.transcript.reading(): if observed != self.revision: - raise TranscriptChanged("Retained journal revision changed") + raise TranscriptChanged() return observed def read_inode(self, revision: FileRevision) -> int: diff --git a/src/agent_comms/view_unread.py b/src/agent_comms/view_unread.py index 5350c205c8..077e8620b4 100644 --- a/src/agent_comms/view_unread.py +++ b/src/agent_comms/view_unread.py @@ -683,7 +683,7 @@ def _counts_locked(self, viewer: str, document: ReadDocument, for location in keys.runtime.read_locations(): key = location.read_key if key in locations and locations[key] != location: - raise TranscriptChanged("Logical source has conflicting read bindings") + raise TranscriptChanged() locations[key] = location if not revisions: return TranscriptUnread({name: ReplyUnread(0) for name in threads}) From 85c6f60fd7add6b6d1c1a23aac3b1a4fec0b24ec Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 3 Nov 2026 21:02:41 -0500 Subject: [PATCH 6/7] Own retained acquisition failures at the shared native open boundary --- src/agent_comms/native_transcript.py | 29 ++++++++++++++++------------ tests/test_native_transcript.py | 23 +++++++++++++++------- 2 files changed, 33 insertions(+), 19 deletions(-) diff --git a/src/agent_comms/native_transcript.py b/src/agent_comms/native_transcript.py index 3502d5503d..c24bd1b8fe 100644 --- a/src/agent_comms/native_transcript.py +++ b/src/agent_comms/native_transcript.py @@ -284,15 +284,24 @@ def require_opened(self, stream: BinaryIO) -> None: """Ordinary journals may append; retained members bind the actual descriptor.""" pass + def opening_failed(self, error: FileNotFoundError) -> Never: + """An ordinary journal may not exist yet; its consumer owns that absence.""" + raise error + @contextmanager def opened(self) -> Generator[BinaryIO]: - """Own every native file open and fence the descriptor consumers actually read.""" - with self.reading(), self.path.open("rb") as stream: - self.require_opened(stream) + """Own acquisition failures and fence the descriptor consumers actually read.""" + with self.reading(): try: - yield stream - finally: + stream = self.path.open("rb") + except FileNotFoundError as error: + self.opening_failed(error) + with stream: self.require_opened(stream) + try: + yield stream + finally: + self.require_opened(stream) def imported_sources(self): """Read original metadata before conversation, including native configuration. @@ -317,18 +326,14 @@ def imported_sources(self): def session_cut(self, revision: FileRevision, offset: int) -> PresentSessionCut: return PresentSessionCut(revision, offset) - def missing_cut(self, error: FileNotFoundError) -> SessionCut: - """An ordinary journal can belong to a session that has not written it yet.""" - return MissingSessionCut() - def cut(self) -> SessionCut: try: # One bound descriptor supplies the revision and last record boundary. with self.opened() as stream: revision = FileRevision.from_file(stream) return self.session_cut(revision, record_end(stream, revision.size)) - except FileNotFoundError as error: - return self.missing_cut(error) + except FileNotFoundError: + return MissingSessionCut() def committed_output_cut(self, location: NativeRecordLocation) -> SessionCut: """Corroborate one indexed output against its original header and row.""" @@ -554,7 +559,7 @@ def require_revision(self, revision: FileObservation) -> None: def require_retained(self) -> None: self.require_revision(FileRevision.capture(self.path)) - def missing_cut(self, error: FileNotFoundError) -> SessionCut: + def opening_failed(self, error: FileNotFoundError) -> Never: raise TranscriptChanged() from error def require_opened(self, stream: BinaryIO) -> None: diff --git a/tests/test_native_transcript.py b/tests/test_native_transcript.py index e699f3aa26..76bc58e612 100644 --- a/tests/test_native_transcript.py +++ b/tests/test_native_transcript.py @@ -24,6 +24,7 @@ from agent_comms.native_failures import NativePiUnavailable from agent_comms.pi_native_backend import PiNativeBackend from agent_comms.private_path import FileRevision +from agent_comms.recorded_time import UnrecordedEventTime from agent_comms.repo import Checkout from agent_comms.threads import Thread from agent_comms.session_file import BusSessionLocation, RetainedSessionLocation, SessionFile @@ -206,17 +207,23 @@ def test_retained_cut_validates_descriptor_before_translating_original_coordinat @pytest.mark.parametrize("read", [ - pytest.param(lambda runtime, reads, size: runtime.read_cut(), id="cut"), + pytest.param(lambda runtime, reads, size, root: runtime.read_cut(), id="cut"), pytest.param( - lambda runtime, reads, size: NativeTranscriptCursor(runtime.read_key, size).mark_read( + lambda runtime, reads, size, root: NativeTranscriptCursor(runtime.read_key, size).mark_read( reads, "viewer", runtime), id="nonzero-acknowledgement", ), + pytest.param( + lambda runtime, reads, size, root: runtime.history_source( + root, Thread("reader", frozenset(), Checkout(root)).incarnation).last_tool_call(), + id="last-tool-call", + ), + pytest.param(lambda runtime, reads, size, root: runtime.saved_model, id="saved-model"), ]) -def test_retained_failed_cut_acquisition_refuses_after_parent_returns( +def test_retained_failed_acquisition_refuses_after_parent_returns( retained_native_location: tuple[Path, RetainedSessionLocation], monkeypatch: pytest.MonkeyPatch, - read: Callable[[RuntimeLocation, TranscriptReadState, int], object], + read: Callable[[RuntimeLocation, TranscriptReadState, int, Path], object], ) -> None: root, location = retained_native_location runtime = location.bound_to(root).location(PiNativeBackend) @@ -232,7 +239,7 @@ def failed_second_open(candidate: Path, mode: Literal["rb"]) -> BinaryIO: if candidate != path: return physical_open(candidate, mode) attempts.append(candidate) - # Construction opens A successfully. Its subsequent cut cannot acquire + # Construction opens A successfully. Its subsequent read cannot acquire # any descriptor; A returns before the reader's final pathname fence. if len(attempts) == 2: path.parent.rename(parked) @@ -244,7 +251,7 @@ def failed_second_open(candidate: Path, mode: Literal["rb"]) -> BinaryIO: monkeypatch.setattr(Path, "open", failed_second_open) with pytest.raises(TranscriptChanged) as refused: - read(runtime, reads, size) + read(runtime, reads, size, root) assert isinstance(refused.value.__cause__, FileNotFoundError) assert len(attempts) == 2 assert FileRevision.capture(path) == location.revision @@ -253,9 +260,11 @@ def failed_second_open(candidate: Path, mode: Literal["rb"]) -> BinaryIO: def test_unstarted_native_journal_may_have_an_empty_cut(tmp_path: Path) -> None: - cut = NativeTranscript(tmp_path / "not-written-yet.jsonl").cut() + reader = NativeTranscript(tmp_path / "not-written-yet.jsonl") + cut = reader.cut() assert cut.offset == 0 assert cut.holds(cut, 0) + assert reader.last_tool_call() == UnrecordedEventTime() @pytest.mark.parametrize("already_read", [False, True]) From e229515b50ca1eed9bdc1763de45d6fd724444c2 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Tue, 3 Nov 2026 21:33:23 -0500 Subject: [PATCH 7/7] Type retained read boundaries and mark the ledger through its cut --- src/agent_comms/agent_backend.py | 22 +++++++++++----------- src/agent_comms/pi_native_backend.py | 22 +++++++++++++--------- src/agent_comms/read_ledger.py | 1 - src/agent_comms/transcripts.py | 2 +- src/agent_comms/view_unread.py | 11 +++++------ 5 files changed, 30 insertions(+), 28 deletions(-) diff --git a/src/agent_comms/agent_backend.py b/src/agent_comms/agent_backend.py index 9bf3481323..d7e627224b 100644 --- a/src/agent_comms/agent_backend.py +++ b/src/agent_comms/agent_backend.py @@ -28,6 +28,10 @@ from .nominal.declared_family import DeclaredFamily from .nominal.field_codec import AcpSessionId, InputId from .native_input_proof import NativeContextProof +from .native_session_reopen import NativeSessionSnapshot +from .native_transcript import NativeTranscript, SessionCut +from .private_path import FileObservation, FileRevision +from .read_ledger import ReadLedger from .pi_counts import PiCount from .thread_model import ThreadModel, UndeclaredModel from .pi_commands import NativeQuery @@ -44,8 +48,6 @@ from .launch_runtime import LaunchRuntime from .owned_turn import TurnDelivery from .threads import Thread - from .native_session_reopen import NativeSessionSnapshot - from .private_path import FileObservation, FileRevision from .transcript_sources import TranscriptSource from .turn_runner import TurnRunner @@ -104,19 +106,15 @@ def read_revision(self, observed: FileObservation) -> FileObservation: def read_inode(self, revision: FileRevision) -> int: return revision.identity.inode - def read_cut(self): - # Like transcript_cursor, this boundary loads its implementation only after - # threads has finished declaring the runtime family. - from .native_transcript import NativeTranscript - + def read_cut(self) -> SessionCut: return NativeTranscript(Path(self.path)).cut() def read_position(self, mark: int, revision: FileRevision) -> int: return mark if mark <= revision.size else 0 - def mark_read(self, reads, viewer: str, through: int) -> None: + def mark_read(self, reads: ReadLedger, viewer: str, through: int) -> None: for location in self.read_locations(): - reads.mark_read(viewer, location.read_key, through, location.read_cut()) + location.read_cut().mark_read(reads, viewer, location.read_key, through) def transcript_cursor(self, frontier, offset, *, source): return frontier.cursor(source.read_key, offset) @@ -339,7 +337,7 @@ def for_worker(cls, runner: TurnRunner, session_id: AcpSessionId) -> Self: @classmethod @abstractmethod - def new_location(cls, root, worktree) -> RuntimeLocation: ... + def new_location(cls, root: Path, worktree: Path) -> RuntimeLocation: ... @classmethod def seal(cls) -> None: @@ -360,7 +358,9 @@ def seal(cls) -> None: AgentBackend.__registry__ = MappingProxyType(dict(AgentBackend.__registry__)) # type: ignore[assignment] AgentBackend._sealed = True - def run_turn(self, thread: Thread, content: InputContent, delivery: TurnDelivery): + def run_turn( + self, thread: Thread, content: InputContent, delivery: TurnDelivery, + ) -> AsyncIterator[AgentEvent]: raise BackendDeclarationError(f"{type(self).__name__} does not run turns") @classmethod diff --git a/src/agent_comms/pi_native_backend.py b/src/agent_comms/pi_native_backend.py index 27199b8678..c1c356c5a8 100644 --- a/src/agent_comms/pi_native_backend.py +++ b/src/agent_comms/pi_native_backend.py @@ -16,7 +16,7 @@ from dataclasses import dataclass from functools import partial from pathlib import Path -from typing import TYPE_CHECKING, ClassVar, Self +from typing import TYPE_CHECKING, ClassVar, Never, Self from agent_comms.native_session_files import ForkedSession, NativeSessionFiles, NativeSessionLocks from agent_comms.pi_wire import PiOmitted @@ -38,7 +38,7 @@ from .native_pi import NativePiRpcLaunch from .native_session_reopen import NativeSessionIdentity, NativeSessionSnapshot from .coordination_errors import TranscriptChanged -from .native_transcript import NativeTranscript, RetainedNativeTranscript +from .native_transcript import NativeTranscript, RetainedNativeTranscript, SessionCut from .private_path import FileObservation, FileRevision from .pi_commands import Compact, NativeQuery from .rule_deal import SystemPromptDelivery @@ -115,7 +115,7 @@ def select_session(self, directory, execution, source_reads, *, participant): def history_source(self, root, incarnation): return self.transcript - def read_cut(self): + def read_cut(self) -> SessionCut: return self.transcript.cut() def deal_delivery(self) -> SystemPromptDelivery: @@ -172,22 +172,26 @@ def read_position(self, mark: int, revision: FileRevision) -> int: def tool_watch_locations(self) -> tuple[RuntimeLocation, ...]: return () - def require_forkable(self): + def require_forkable(self) -> Never: raise NativePiUnavailable("Retained native history cannot fork a session") - def arguments(self): + def arguments(self) -> Never: raise NativePiUnavailable("Retained native history has no launch arguments") - def new_location(self, root, worktree): + def new_location(self, root: Path, worktree: Path) -> Never: raise NativePiUnavailable("Retained native history cannot start a session") - def select_session(self, directory, execution, source_reads, *, participant): + def select_session( + self, directory: object, execution: object, source_reads: object, *, participant: object, + ) -> Never: raise NativePiUnavailable("Retained native history cannot select a native input") - async def compact(self, operation): + async def compact( + self, operation: Callable[[], Awaitable[CompactionOutcome]], + ) -> CompactionOutcome: return FailedCompaction("Retained native history cannot be compacted.") - def storage_files(self): + def storage_files(self) -> tuple[()]: return () def session_observation(self, root: Path, thread: Thread) -> SessionObservation: diff --git a/src/agent_comms/read_ledger.py b/src/agent_comms/read_ledger.py index cdcc04118d..306d3bea58 100644 --- a/src/agent_comms/read_ledger.py +++ b/src/agent_comms/read_ledger.py @@ -11,7 +11,6 @@ from abc import abstractmethod from collections.abc import Callable, Iterator, Sequence from dataclasses import dataclass, field, replace -from pathlib import Path from typing import TYPE_CHECKING from .bus import Bus, BusReader diff --git a/src/agent_comms/transcripts.py b/src/agent_comms/transcripts.py index ce1f06fec8..3941dc47e0 100644 --- a/src/agent_comms/transcripts.py +++ b/src/agent_comms/transcripts.py @@ -238,7 +238,7 @@ def with_frontier(self, frontier: WireFrontier) -> TranscriptCursor: def mark_read(self, reads: TranscriptReadState, viewer: str, runtime: RuntimeLocation) -> None: if self.session_file != runtime.read_key: raise TranscriptChanged() - runtime.mark_read(reads, viewer, self.offset) + runtime.mark_read(reads.reads, viewer, self.offset) @dataclass(frozen=True) diff --git a/src/agent_comms/view_unread.py b/src/agent_comms/view_unread.py index 077e8620b4..1ca1f6682b 100644 --- a/src/agent_comms/view_unread.py +++ b/src/agent_comms/view_unread.py @@ -21,7 +21,7 @@ from .store_files import StoreBusy from .bus import Bus from .native_entries import NativeEntry -from .native_transcript import SessionCut, record_end, reverse_records +from .native_transcript import record_end, reverse_records from .private_path import FileObservation, FileRevision, MissingFile from .read_ledger import ReadDocument, ReadLedger from .registry_document import RosterVisibility @@ -616,9 +616,11 @@ def _refresh_keys(self, snapshot: RegistrySnapshot) -> None: keys = (prior if prior is not None and prior.matches(thread) else TranscriptReadKeys.capture(thread)) rebuild |= prior is None or prior.paths != keys.paths - if prior is not None and prior.runtime != keys.runtime: - self._preparation.retire(prior.revisions.keys()) current[thread.name] = keys + for name in current.keys() & self._thread_keys.keys(): + prior = self._thread_keys[name] + if prior.runtime != current[name].runtime: + self._preparation.retire(prior.revisions.keys()) rebuild |= bool(self._thread_keys.keys() - current.keys()) self._thread_keys = current self._key_snapshot = snapshot @@ -714,9 +716,6 @@ def _counts_locked(self, viewer: str, document: ReadDocument, total.record(name, result) return TranscriptUnread(result) - def mark_read(self, viewer: str, source: str, through: int, cut: SessionCut) -> None: - cut.mark_read(self.reads, viewer, source, through) - _shared_lock = RLock() _shared_states: WeakValueDictionary[str, TranscriptReadState] = WeakValueDictionary()