diff --git a/docs/retained-practice-inputs.md b/docs/retained-practice-inputs.md index dd6e0fe657..ef82996a34 100644 --- a/docs/retained-practice-inputs.md +++ b/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, diff --git a/src/agent_comms/context_store.py b/src/agent_comms/context_store.py index 8c7fb8574d..8aa26590c7 100644 --- a/src/agent_comms/context_store.py +++ b/src/agent_comms/context_store.py @@ -562,6 +562,40 @@ def retain_contexts( 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: diff --git a/src/agent_comms/installer/practice.py b/src/agent_comms/installer/practice.py index db36562689..94b7fd68fe 100644 --- a/src/agent_comms/installer/practice.py +++ b/src/agent_comms/installer/practice.py @@ -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 @@ def capture_inputs( 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] = (), diff --git a/tests/test_context_manifest_deltas.py b/tests/test_context_manifest_deltas.py index 3ba487a366..b4d2d827b1 100644 --- a/tests/test_context_manifest_deltas.py +++ b/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: diff --git a/tests/test_retained_capture.py b/tests/test_retained_capture.py index ddad32c432..4039cfa42b 100644 --- a/tests/test_retained_capture.py +++ b/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: