Compare selected inputs before releasing their original capture cut (P1086)

docs/retained-practice-inputs.md

@@ -34,6 +34,26 @@ probe or create the selected destination. The first check incorrectly compared
 the CLI subclass identity with the decoded base request; that failed epoch is
 retained beside the corrected declared-value check.
 
+## Readback belongs to the captured cut
+
+`PracticeCopy.capture_inputs` now reads the selected stores back before leaving
+`capturing_inputs`, while the existing installer hold still excludes source
+writers. It compares the ordinary unresolved-input projection, native inputs by
+both session and original path, selected participant presence, and every retained
+participant's ordinary aggregate. The held original's registry selects the
+coordinator schema once; that declaration validates both databases. Selected data
+does not gain a registry or runnable-root authority merely to pass readback.
+`ContextStore.require_retained` owns context comparison: complete selected entry
+groups and all retained requests, including predecessors, resolve through the same
+reader as saved-context browsing.
+
+A mismatch refuses the selected operation. The checks run under the existing
+source-integrity and registry checks, without restoring holder write permission,
+a callback, another hold, or a post-rehearsal comparison against newer originals.
+A final byte measurement still does not establish peak resources, native launch
+permission, or the real Toad journey. The two additional controls are authored,
+not executed in this source change.
+
 ## Original recipient birth participates in owner matching
 
 Source tracing found `StoredInput.matches_owner` consulted only its origin,

src/agent_comms/context_store.py

@@ -1,3 +1,3 @@
 """Append-only original turn contexts, independent of message delivery bytes."""
 
 from __future__ import annotations
@@ -562,6 +562,40 @@
                 for reference in selected:
                     ContextRequest.require(original.destination, id=reference.record_id)
 
+    def require_retained(
+        self, destination: ContextStore, identities: tuple[NativeSessionIdentity, ...],
+        requests: tuple[ContextManifestReference, ...],
+    ) -> None:
+        """Compare selected originals through the browser's entry and request readers.
+
+        The caller owns the cross-store capture lifetime. Resolve every retained
+        request, including predecessors, rather than comparing only its final
+        displayed value or recreating the serialized version/entry relationship.
+        """
+        with self.snapshot() as source, destination.snapshot() as retained:
+            for identity in identities:
+                source_entries, retained_entries = ({row.entry_id: row for row in ContextEntry.select(
+                    db, session_id=identity.session_id, session_file=identity.session_file,
+                )} for db in (source, retained))
+                if source_entries != retained_entries:
+                    raise RelationViolationError("Selected context entries differ from their originals")
+            for reference in requests:
+                stored = ContextStoreInfo.require(retained, singleton=1).source
+                if stored != reference.source:
+                    raise RelationViolationError("Retained context reference names another source")
+                ContextRequest.require(retained, id=reference.record_id)
+            for row in ContextRequest.stream(retained):
+                original_source = ContextStoreInfo.require(source, singleton=1).source
+                retained_source = ContextStoreInfo.require(retained, singleton=1).source
+                original_row = ContextRequest.require(source, id=row.id)
+                expected = self.resolve(source, original_source, original_row)
+                actual = destination.resolve(retained, retained_source, row)
+                if row != original_row or actual != expected:
+                    raise RelationViolationError("Retained context reader differs from its original")
+            for version in ContextSegmentVersion.stream(retained):
+                if version != ContextSegmentVersion.require(source, id=version.id):
+                    raise RelationViolationError("Retained context version differs from its original")
+
     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/installer/practice.py

@@ -1,3 +1,3 @@
 """The practice bus a rehearsal carries: its copy, private home, size and space."""
 
 from __future__ import annotations
@@ -11,9 +11,10 @@
 from collections.abc import Iterable, Iterator
 from contextlib import contextmanager
 from agent_comms.context_store import ContextManifestReference, ContextStore
-from agent_comms.coordination_database import CoordinationStore
+from agent_comms.coordination_database import CoordinationStore, ReadableStore
 from agent_comms.coordination_tables.annotations import ContextSpanRows
-from agent_comms.input_disposition import InputDocument, SnapshotInputDispositions
+from agent_comms.coordination_tables.participants import Participants
+from agent_comms.input_disposition import InputDispositions, InputDocument, SnapshotInputDispositions
 from agent_comms.native_runtime_input import NativeRuntimeInput
 from agent_comms.participant_store import ParticipantRows, ParticipantStore
 from agent_comms.registry_document import RegistryEntry
@@ -249,10 +250,50 @@
             self.remaining_bytes(maximum_bytes)
             self.include_contexts(identities, maximum_bytes=maximum_bytes, requests=requests)
             self.include_inputs(entries, maximum_bytes=maximum_bytes, identities=identities)
+            self.require_inputs(entries, identities=identities, requests=requests)
             measured = PracticeMeasurement.capture(self.path)
             measured.remaining(maximum_bytes)
             return measured
 
+    def require_inputs(
+        self, entries: tuple[RegistryEntry, ...], *,
+        identities: tuple[NativeSessionIdentity, ...],
+        requests: tuple[ContextManifestReference, ...],
+    ) -> None:
+        """Read back the selected copy before its original capture cut is released.
+
+        Use the ordinary unresolved projection, context resolver and participant
+        aggregate reader. No lifecycle reset, input replay, callback or second
+        hold can stand in for equality while these originals are still frozen.
+        """
+        ContextStore(self.root).require_retained(ContextStore(self.path), identities, requests)
+        if InputDispositions(self.root).unresolved(entries) != InputDispositions(self.path).unresolved(entries):
+            raise UpgradeRefused("Selected unresolved input reader differs from its original")
+        with (
+            ReadableStore(CoordinationStore.path_in(self.root)).reader_snapshot() as source,
+            ReadableStore(CoordinationStore.path_in(self.path)).reader_snapshot() as retained,
+        ):
+            schema = CoordinationStore.schema_at(CoordinationStore.path_in(self.root))
+            schema.require_schema(source)
+            schema.require_schema(retained)
+            for identity in identities:
+                source_inputs, retained_inputs = ({row.input_id: row for row in NativeRuntimeInput.select(
+                    db, session_id=identity.session_id, session_file=identity.session_file,
+                )} for db in (source, retained))
+                if source_inputs != retained_inputs:
+                    raise UpgradeRefused("Selected native session/path inputs differ from their originals")
+            for entry in entries:
+                lookup = entry.thread.incarnation.require_recorded().lookup
+                if Participants.count(source, participant_lookup=lookup) != Participants.count(
+                    retained, participant_lookup=lookup,
+                ):
+                    raise UpgradeRefused("Selected participant presence differs from its original")
+            for person in Participants.stream(retained):
+                if ParticipantStore.read_at(source, person.participant_lookup) != ParticipantStore.read_at(
+                    retained, person.participant_lookup,
+                ):
+                    raise UpgradeRefused("Selected participant aggregate differs from its original")
+
     def include_contexts(
         self, identities: Iterable[NativeSessionIdentity], *, maximum_bytes: int,
         requests: Iterable[ContextManifestReference] = (),

tests/test_context_manifest_deltas.py

@@ -237,6 +237,7 @@ def test_selected_context_copy_preserves_closure_and_original_sequences(tmp_path
     copy = PracticeCopy(comms.root, tmp_path / 'retained')
     copy.include_contexts((), requests=(reference, reference), maximum_bytes=2**20)
     retained = ContextStore(copy.path)
+    comms.contexts.require_retained(retained, (), (reference,))
 
     with retained.snapshot() as db:
         info = ContextStoreInfo.require(db, singleton=1)
@@ -261,6 +262,17 @@ def test_selected_context_copy_preserves_closure_and_original_sequences(tmp_path
     assert copy.request.root == copy.path
 
 
+def test_selected_context_readback_refuses_omitted_native_entries(tmp_path: Path) -> None:
+    comms = context_comms(tmp_path / 'source')
+    owner = comms.registry.require('Alice')
+    selected = comms.contexts.record_after(UNREAD_CONTEXT, context(owner, tmp_path, 1, 1))
+    identity = selected.recorded.segments[2].entries[0].identity
+    destination = ContextStore(tmp_path / 'retained')
+    comms.contexts.retain_contexts(destination, (), maximum_bytes=2**20)
+    with pytest.raises(RelationViolationError, match='context entries differ'):
+        comms.contexts.require_retained(destination, (identity,), ())
+
+
 def foreign_context_reference(
     store: ContextStore, previous: CapturedContext, reference: ContextManifestReference,
 ) -> ContextManifestReference:

tests/test_retained_capture.py

@@ -61,6 +61,31 @@ def test_selected_capture_request_survives_cli_and_launcher_wire(tmp_path: Path)
     assert not probe.destination.exists()
 
 
+def test_selected_capture_reads_original_inputs_before_releasing_its_cut(
+    capture_source: Comms, tmp_path: Path,
+) -> None:
+    source = capture_source
+    entry = source.registry.snapshot().entry('capture-owner')
+    lookup = entry.thread.incarnation.require_recorded().lookup
+    key = TurnInputKey(TurnId('b' * 32))
+    InputDispositions(source.root).record(
+        key, seq=None, owner=entry.thread.name, claim_scope=1,
+        target=entry.thread.name, text='original unresolved input',
+    )
+    gate = MaintenanceBarrier(source.registry.store.path)
+    held = gate.retarget(gate.begin(MaintenanceOperator.current()), LaunchRuntime.running())
+    destination = tmp_path / 'read-back'
+    try:
+        with held.holding():
+            measured = SelectedInputsProbe(source.root, destination, 3 * 1024**3, (lookup,)).run()
+            assert measured.bytes > 0
+            assert InputDispositions(destination).unresolved((entry,)) == InputDispositions(
+                source.root,
+            ).unresolved((entry,))
+    finally:
+        gate.clear(held)
+
+
 def test_capture_refuses_a_destination_inside_its_source(
     capture_source: Comms,
 ) -> None: