P1086: qualify selected original-input retention from 3829

src/agent_comms/assignment_store.py

@@ -102,14 +102,8 @@ def settle_harness_inputs(
         Every step runs in the caller's coordinator write, so the harness attempt
         and its claim consequence commit together or not at all.
         """
-        claims = WakeAssignment.select(
-            db,
-            where=f"recipient_lookup=? AND wire_seq IN ({','.join('?' * len(sources))})",
-            parameters=(recipient_lookup, *(source.seq for source in sources)),
-        )
+        claims = WakeAssignment.for_harness_sources(db, recipient_lookup, sources)
         for claim in claims:
-            if claim.source not in sources:
-                raise IdentityConflict("harness delivery names another message at this sequence")
             for disposition in steps(claim.lifecycle):
                 # The revision fence refuses a step if the native owner moved the claim meanwhile.
                 claim = self.transition_in(

src/agent_comms/bus.py

@@ -51,7 +51,9 @@
 )
 from agent_comms.errors import CommittedMessageError, RelationViolationError, UnclaimedError
 from agent_comms.nominal.field_codec import ClaimId, FieldCodec, MessageId, ThreadLookup
-from agent_comms.locked_store import BusFile, FileStore, SQLiteReader, StoreCommittedError
+from agent_comms.locked_store import (
+    BusFile, CompleteInputReader, FileStore, SQLiteReader, StoreCommittedError,
+)
 from agent_comms.message_reference import MessageReference
 from agent_comms.private_path import FileIdentity, PresentPath
 from agent_comms.read_basis import CursorRow
@@ -523,7 +525,10 @@ def observing(self) -> Iterator[sqlite3.Connection]:
             yield db
 
 
-class Bus(FileStore, StoredBusReader, ReconciledFile, WalStore, CarriedWriter, BusFile, Watched):
+class Bus(
+    FileStore, StoredBusReader, ReconciledFile, WalStore, CarriedWriter, BusFile,
+    CompleteInputReader, Watched,
+):
     """One SQLite transaction owns the sequence and all of a record's fields."""
 
     filename = "bus.sqlite3"

src/agent_comms/bus_tables.py

@@ -20,6 +20,7 @@
     ProjectionProducerTables,
     RowNotFoundError,
     TableReference,
+    TableSource,
     TypedRow,
 )
 
@@ -112,10 +113,10 @@ class MessageIdentitySource(MessageReference, messages.BusProjectionTables):
     """A record's existing reference, searchable without recomputing its content hash."""
 
     @classmethod
-    def practice_sequence(cls, db: sqlite3.Connection, source: sqlite3.Connection) -> int:
-        through = messages.BusRow.last_seq(source)
+    def practice_sequence(cls, db: sqlite3.Connection, source: TableSource) -> int:
+        through = messages.BusRow.last_seq(db, table_source=source)
         if through:
-            expected = cls.require(source, seq=through)
+            expected = cls.require(db, table_source=source, seq=through)
             if cls.require(db, seq=through) != expected:
                 raise RelationViolationError("Practice bus identity differs at its last sequence")
         return through

src/agent_comms/cohort_schema.py

@@ -32,6 +32,7 @@
     Index,
     InitialOneWayColumn,
     NoDelete,
+    RetainsChildren,
     SharedFileTables,
     SQLiteJournalMode,
     TextStorage,
@@ -60,6 +61,7 @@ class AwarenessTable(SharedFileTables, DurableTables):
 
 @dataclass(frozen=True, kw_only=True)
 class ClaimBatchAcknowledgements(
+    RetainsChildren,
     InitialOneWayColumn,
     GuardedColumns,
     NoDelete,

src/agent_comms/context_store.py

@@ -14,6 +14,7 @@
 from typing import Annotated, ClassVar, Self
 
 from .errors import RelationViolationError, StoredHistoryUnreadable
+from .locked_store import InputCaptureFile
 from .nominal.field_codec import ContextStoreId, FieldCodec, MalformedRecord, NullMember
 from .native_session_reopen import NativeSessionIdentity
 from .registration import Registration
@@ -31,7 +32,10 @@
     UnrecordedSegment,
 )
 
-from .typed_table import StoreRows, Column, DurableStore, Generated, Index, RowidSequenced, SingletonRow, TypedTable
+from .typed_table import (
+    AttachedPracticeRebuild, StoreRows, Column, DurableStore, Generated, Index,
+    RowidSequenced, SingletonRow, TableSchema, TypedTable,
+)
 from .turn_item import NativeItemId
 from .native_session_reopen import PiSessionId
 
@@ -242,6 +246,35 @@ def recorded(self, db: sqlite3.Connection) -> RecordedContextManifest:
             parts.append(ContextSegmentVersion.require(db, id=version).segment)
         return self.identity.with_segments(tuple(parts))
 
+    @classmethod
+    def retain_required_rows(
+        cls, original: AttachedPracticeRebuild, schema: type[TableSchema],
+    ) -> int:
+        """The stored request owns its serialized version, entry and predecessor edges."""
+        changed = super().retain_required_rows(original, schema)
+        db = original.destination
+        for request in cls.select(db):
+            changed += original.retain_rows(ContextStoreInfo)
+            ContextStoreInfo.require(db, singleton=1)
+            changed += original.retain_keys(
+                ContextSegmentVersion, tuple({"id": version} for version in request.versions),
+            )
+            record = request.recorded(db)  # Missing original versions refuse, never disappear.
+            for part in record.segments:
+                predecessors = part.predecessors(request.id)
+                changed += original.retain_keys(cls, tuple({"id": key} for key in predecessors))
+                for key in predecessors:
+                    if cls.require(db, id=key).identity.thread != request.identity.thread:
+                        raise RelationViolationError("Context predecessor is missing or foreign")
+                changed += original.retain_keys(ContextEntry, tuple(
+                    {"session_id": ref.identity.session_id,
+                     "session_file": ref.identity.session_file, "entry_id": ref.entry_id}
+                    for ref in part.entries
+                ))
+                for reference in part.entries:
+                    ContextEntry.original(db, reference)
+        return changed
+
 
 class ContextPriorSupersededError(RelationViolationError):
     """Another writer recorded this thread's context after the prior this writer holds."""
@@ -447,7 +480,7 @@ def reference(self, manifest: ContextManifest) -> ContextManifestReference:
 
 
 @dataclass(frozen=True)
-class ContextStore(DurableStore):
+class ContextStore(DurableStore, InputCaptureFile):
     root: Path
     filename: ClassVar[str] = "turn_context.sqlite3"
     tables = ContextStoreRows
@@ -484,6 +517,51 @@ def entries(
                                                    session_file=identity.session_file,
                                                    entry_id=entry_id)}
 
+    def original_segments(self) -> Iterator[SegmentManifest]:
+        """Original segments actually present in this selected entry/request store.
+
+        Requests restore through the same original predecessor/entry reader as
+        the browser. Entry-only stores need no invented request/store identity.
+        """
+        with self.snapshot() as db:
+            for entry in ContextEntry.stream(db):
+                yield from entry.manifest.original_values()
+            for request in ContextRequest.stream(db):
+                source = ContextStoreInfo.require(db, singleton=1).source
+                for root in self.resolve(db, source, request).value.segments:
+                    yield from root.original_values()
+
+    def retain_contexts(
+        self, destination: ContextStore, identities: Iterable[NativeSessionIdentity], *,
+        maximum_bytes: int, requests: Iterable[ContextManifestReference] = (),
+    ) -> None:
+        """Retain original entry manifests and explicitly selected saved requests.
+
+        Raw writable values, source identity, predecessor/version/entry closure
+        and sequence high-water marks travel unchanged. Session paths are keys,
+        not relocated storage or permission to launch or replay a native session.
+        Entry-only selection retains no request or fabricated store identity.
+        """
+        with type(self).retaining_rows(
+            self.path, destination.path, maximum_bytes=maximum_bytes,
+        ) as original:
+            original.retain_keys(ContextEntry, tuple(
+                {"session_id": identity.session_id, "session_file": identity.session_file}
+                for identity in dict.fromkeys(identities)
+            ))
+            selected = tuple(dict.fromkeys(requests))
+            if selected:
+                original.retain_rows(ContextStoreInfo)
+                source = ContextStoreInfo.require(original.destination, singleton=1).source
+                for reference in selected:
+                    if reference.source != source:
+                        raise RelationViolationError("Selected context belongs to another store")
+                original.retain_keys(ContextRequest, tuple(
+                    {"id": reference.record_id} for reference in selected
+                ))
+                for reference in selected:
+                    ContextRequest.require(original.destination, id=reference.record_id)
+
     def prior(self, owner: ThreadIncarnation) -> RecordedContextState:
         """Read the thread's previous request from a snapshot, outside any write lock."""
         with self.snapshot() as db:

src/agent_comms/coordination_database.py

@@ -1,3 +1,3 @@
 """Private coordinator storage and declaration-owned schema installation."""
 
 from __future__ import annotations
@@ -15,7 +15,7 @@
 
 from agent_comms.coordination_errors import CoordinationReadUnavailable, IntegrityViolationError
 from agent_comms.coordination_schema import CoordinatorTable
-from agent_comms.locked_store import FileStore, SQLiteReader
+from agent_comms.locked_store import FileStore, InputCaptureFile, SQLiteReader
 from agent_comms.typed_table import (
     CarriedWriter,
     ReconciledFile,
@@ -27,7 +27,6 @@
 from .nominal.declared_family import DeclaredFamily
 from .store_files import DEFAULT_STORE_CONTENTION, StoreContention
 
-
 class StoreFile(DeclaredFamily):
     """One decoded file state, with the consumer's result inside its read lifetime."""
 
@@ -67,7 +66,7 @@
             yield consumer.capture(connection, *arguments)
 
 
-class CoordinationStore(FileStore, CarriedWriter, ReconciledFile):
+class CoordinationStore(FileStore, CarriedWriter, ReconciledFile, InputCaptureFile):
     """The private coordinator database and its declared schema."""
 
     filename = "coordination.sqlite3"
@@ -119,7 +118,7 @@
         with CoordinationReadUnavailable.checking():
             self._connection = self._open_connection(contention)
             try:
-                self._configure_connection()
+                self.configure_connection(self._connection)
                 self._initialize_schema()
                 self._enforce_private_modes()
             except BaseException:
@@ -133,13 +132,15 @@
         """The root's own composition of this file decides its schema and its strays."""
         self.schema_at(self.path).install_schema(self._connection, self.schema_change(self.path))
 
-    def _configure_connection(self) -> None:
+    @classmethod
+    def configure_connection(cls, connection: sqlite3.Connection) -> None:
+        """Ordinary opening and selected loading use the same row-owned SQL behavior."""
         for row_type in CoordinatorTable.declarations():
-            row_type.configure_connection(self._connection)
-        self._connection.row_factory = sqlite3.Row
-        self._connection.execute("PRAGMA foreign_keys = ON")
+            row_type.configure_connection(connection)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys = ON")
         # Rollback journal avoids a post-schema WAL-mode race among fresh openers.
-        self._connection.execute("PRAGMA synchronous = FULL")
+        connection.execute("PRAGMA synchronous = FULL")
 
     def _prepare_private_file(self) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)

src/agent_comms/coordination_session.py

@@ -206,7 +206,8 @@ def _prepare_private_file(self) -> None:
     def _open_connection(self, contention: StoreContention) -> sqlite3.Connection:
         return self._open_reader(self.path, contention)
 
-    def _configure_connection(self) -> None:
+    @classmethod
+    def configure_connection(cls, connection: sqlite3.Connection) -> None:
         pass  # _open_reader configured query-only behavior before publication.
 
     def _initialize_schema(self) -> None:

src/agent_comms/coordination_tables/annotations.py

@@ -1,16 +1,25 @@
 """Durable label/request declarations; the existing coordinator owns storage."""
 from __future__ import annotations
 
+import sqlite3
+from abc import abstractmethod
+from collections.abc import Iterable, Iterator
 from dataclasses import dataclass, field
-from typing import Annotated, ClassVar
+from typing import Annotated, ClassVar, Self
 
 from agent_comms.coordination_schema import CoordinatorRow
 from agent_comms.message_reference import MessageReference
+from agent_comms.turn_context import ContextSpan, SegmentManifest
 from agent_comms.typed_table import (
+    LOCAL_TABLE_SOURCE,
+    AttachedPracticeRebuild,
     Column,
     Generated,
     Index,
     RowId,
+    TableSchema,
+    TableSource,
+    TypedTable,
     UnallocatedRowId,
     sql_literal,
 )
@@ -19,8 +28,46 @@
 from agent_comms.working_memory_requests import AnnotationOutcome, DisclosureRequest
 
 
+class ContextSpanRows(TypedTable, declared_name=None):
+    """Stored answers and request dispositions share original-span eligibility."""
+
+    __slots__ = ()
+
+    @property
+    @abstractmethod
+    def context_span(self) -> ContextSpan: ...
+
+    @classmethod
+    def for_segments(
+        cls, db: sqlite3.Connection, segments: Iterable[SegmentManifest], *,
+        table_source: TableSource = LOCAL_TABLE_SOURCE, **key: object,
+    ) -> Iterator[Self]:
+        originals: dict[str, list[SegmentManifest]] = {}
+        for segment in segments:
+            originals.setdefault(segment.sha256, []).append(segment)
+        for row in cls.stream_keys(db, tuple(
+            dict(key, segment_digest=digest) for digest in originals
+        ), table_source=table_source):
+            span = row.context_span
+            if any(segment.contains_span(span) for segment in originals[span.segment_sha256]):
+                yield row
+
+    @classmethod
+    def retain_segments(
+        cls, original: AttachedPracticeRebuild, segments: Iterable[SegmentManifest],
+    ) -> int:
+        # Restored requests may repeat large original prefixes. Consume one
+        # segment at a time, not a second materialized context corpus.
+        members = cls.members_with(cls)
+        return sum(row.retain_original(original)
+                   for segment in segments
+                   for member in members
+                   for row in member.for_segments(original.destination, (segment,),
+                                                  table_source=original.source))
+
+
 @dataclass(frozen=True, kw_only=True)
-class SpanAnnotationsRow(CoordinatorRow):
+class SpanAnnotationsRow(ContextSpanRows, CoordinatorRow):
     label: ModelLabel
     created_at_ms: int
     id: Annotated[RowId, Column(primary_key=True, auto_increment=True)] = field(
@@ -78,22 +125,17 @@ def effective(cls, rows: tuple[SpanAnnotationsRow, ...]):
         # Row ids order successive corrections from the same authority. Nominal
         # inheritance, not sequence, decides human versus model authority.
         latest = {}
-        for row in sorted(rows, key=lambda value: value.id):
+        for row in sorted(rows, key=lambda value: int(value.id)):
             latest[type(row.label)] = row.label
         return SpanLabel.effective(tuple(latest.values()))
 
-    @classmethod
-    def for_digests(cls, db, digests, classifier):
-        if not digests:
-            return ()
-        return cls.select(db,
-            where='segment_digest IN (' + ','.join('?' for _ in digests) + ')',
-            parameters=tuple(digests), order_by=("id",),
-            classifier=classifier.classifier, classifier_pin=classifier.pin)
+    @property
+    def context_span(self) -> ContextSpan:
+        return self.label.span
 
 
 @dataclass(frozen=True, kw_only=True)
-class AnnotationRequestsRow(CoordinatorRow):
+class AnnotationRequestsRow(ContextSpanRows, CoordinatorRow):
     request: DisclosureRequest
     outcome: AnnotationOutcome
     created_at_ms: int
@@ -137,6 +179,33 @@ class AnnotationRequestsRow(CoordinatorRow):
     unique = (SpanAnnotationsRow.address,)
     indexes = (Index(("grant_reference", "created_at_ms")),)
 
+    @property
+    def context_span(self) -> ContextSpan:
+        return self.request.span
+
+    @classmethod
+    def retain_required_rows(
+        cls, original: AttachedPracticeRebuild, schema: type[TableSchema],
+    ) -> int:
+        """Keep original grant usage and each request's complete answer/correction group."""
+        changes = super().retain_required_rows(original, schema)
+        changes += original.retain_keys(cls, tuple(
+            {"grant_reference": row.request.grant} for row in cls.stream(original.destination)
+        ))
+        # One natural address can name multiple correction rows; it is not a
+        # foreign key. Compare the actual declared generated columns, without
+        # making storage import a reader to reconstruct their domain values.
+        related = " AND ".join(
+            f"{SpanAnnotationsRow.qualified(name)} IS {cls.qualified(name)}"
+            for name in SpanAnnotationsRow.address
+        )
+        changes += original.retain_rows(
+            SpanAnnotationsRow,
+            where=f"EXISTS (SELECT 1 FROM main.{LOCAL_TABLE_SOURCE.qualified(cls.declared_name)} "
+                  f"WHERE {related})",
+        )
+        return changes
+
     @classmethod
     def requests_since(cls, db, reference: MessageReference, since: int) -> int:
         return cls.count(db, where='created_at_ms>?', parameters=(since,), grant_reference=reference)