P1086: preserve original identity in retained native readers

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
@@ -63,6 +67,12 @@ class RuntimeLocation(DeclaredFamily, affix="Runtime"):
     def backend(self):
         return self._backend
 
+    @classmethod
+    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")
+
     def require_forkable(self):
         pass
 
@@ -80,8 +90,31 @@ 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) -> 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: ReadLedger, viewer: str, through: int) -> None:
+        for location in self.read_locations():
+            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)
@@ -304,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:
@@ -325,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

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:

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)
 

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]
 

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