Persist recorded entities for first-Brief derivation

ace/application/recorded_source_admission.py

@@ -44,9 +44,10 @@
     validate_reference,
     validate_slug,
 )
-from ace.intelligence.contracts.ledger import IntelligenceRecordKind
+from ace.intelligence.contracts.ledger import IntelligenceRecordKind, resource_available_at
 from ace.intelligence.contracts.resources import (
     ActivationRevisionReferenceV1Alpha1,
+    EntitySnapshotV1Alpha1,
     EvidenceAcquisitionMode,
     IntelligenceResourceMode,
     ObservationV1Alpha1,
@@ -269,11 +270,12 @@ def reusable_authority(self) -> Literal[False]:
 
 @dataclass(frozen=True, slots=True)
 class RecordedSourceAdmission:
-    """Exact reopened recorded receipts, snapshots, Observations, and append receipt."""
+    """Exact reopened recorded evidence, mapped entities, and append receipt."""
 
     acquisition_receipts: tuple[RecordedSourceAcquisitionReceiptV1Alpha1, ...]
     source_snapshots: tuple[CanonicalSourceSnapshotV1Alpha1, ...]
     observations: tuple[ObservationV1Alpha1, ...]
+    entity_snapshots: tuple[EntitySnapshotV1Alpha1, ...]
     transaction_receipt: AppendOnlyTransactionReceiptV1
     replayed: bool
 
@@ -370,9 +372,11 @@ def _validate_build(self) -> None:
             or authority.operation != INTELLIGENCE_BUILD_OPERATION
             or authority.authority != INTELLIGENCE_BUILD_AUTHORITY
             or authority.grant_ref != self.build.request.authority_grant_ref
-            or CONNECT_SOURCES_EFFECT not in self.build.request.approved_effects
+            or not {CONNECT_SOURCES_EFFECT, "map_concepts"}.issubset(self.build.request.approved_effects)
         ):
-            raise RecordedSourceAdmissionError("authorized build does not cover exact recorded source admission")
+            raise RecordedSourceAdmissionError(
+                "authorized build does not cover exact recorded source admission and mapping"
+            )
         if self.binding.prepared_binding.reference.product_id != self.build.product_id:
             raise RecordedSourceAdmissionError("committed activation crossed the authorized build product")
 
@@ -426,6 +430,7 @@ async def admit(self, materials: tuple[RecordedSourceMaterialV1Alpha1, ...]) ->
         acquisitions: list[RecordedSourceAcquisitionReceiptV1Alpha1] = []
         snapshots: list[CanonicalSourceSnapshotV1Alpha1] = []
         observations: list[ObservationV1Alpha1] = []
+        entities: list[EntitySnapshotV1Alpha1] = []
         for material in exact:
             acquisition = RecordedSourceAcquisitionReceiptV1Alpha1(
                 product_id=self.build.product_id,
@@ -472,19 +477,36 @@ async def admit(self, materials: tuple[RecordedSourceMaterialV1Alpha1, ...]) ->
             except Exception:
                 raise RecordedSourceAdmissionError("recorded material failed activation-bound source mapping") from None
             observation = mapped.observation
+            entity = mapped.entity_snapshot
             if (
                 observation.mode is not IntelligenceResourceMode.PREPARED
                 or observation.acquisition_mode is not EvidenceAcquisitionMode.RECORDED_REPLAY
                 or observation.acquisition_receipt_ref != acquisition.receipt_id
                 or observation.acquisition_receipt_digest != acquisition.receipt_digest
+                or entity.mode is not IntelligenceResourceMode.PREPARED
+                or entity.activation_revision != observation.activation_revision
+                or len(entity.lineage) != 1
+                or entity.lineage[0].resource_id != observation.resource_id
+                or entity.lineage[0].resource_digest != observation.resource_digest
+                or entity.lineage[0].resource_as_of != observation.as_of
+                or entity.lineage[0].resource_available_at != resource_available_at(observation)
             ):
-                raise RecordedSourceAdmissionError("prepared mapping changed recorded acquisition truth")
+                raise RecordedSourceAdmissionError(
+                    "prepared mapping changed recorded acquisition truth or entity lineage"
+                )
             acquisitions.append(acquisition)
             snapshots.append(snapshot)
             observations.append(observation)
+            entities.append(entity)
 
         records_list: list[ImmutableRecordV1] = []
-        for acquisition, snapshot, observation in zip(acquisitions, snapshots, observations, strict=True):
+        for acquisition, snapshot, observation, entity in zip(
+            acquisitions,
+            snapshots,
+            observations,
+            entities,
+            strict=True,
+        ):
             base = len(records_list)
             records_list.extend(
                 (
@@ -515,6 +537,15 @@ async def admit(self, materials: tuple[RecordedSourceMaterialV1Alpha1, ...]) ->
                         available_at=admitted_at,
                         order=base + 2,
                     ),
+                    _record(
+                        entity,
+                        product_id=self.build.product_id,
+                        kind=IntelligenceRecordKind.ENTITY_SNAPSHOT.value,
+                        key=str(entity.resource_id),
+                        as_of=entity.as_of,
+                        available_at=resource_available_at(entity),
+                        order=base + 3,
+                    ),
                 )
             )
         request = AppendOnlyTransactionRequestV1(
@@ -550,7 +581,7 @@ async def _replay(
         )
         if receipt is None:
             return None
-        if len(receipt.records) != len(expected) * 3:
+        if len(receipt.records) != len(expected) * 4:
             raise RecordedSourceAdmissionError("recorded admission receipt lost exact material-set shape")
         loaded: list[ImmutableRecordV1] = []
         for reference in receipt.records:
@@ -567,18 +598,21 @@ async def _replay(
             RECORDED_SOURCE_RECORD_KIND,
             SOURCE_SNAPSHOT_RECORD_KIND,
             IntelligenceRecordKind.OBSERVATION.value,
+            IntelligenceRecordKind.ENTITY_SNAPSHOT.value,
         ) * len(expected)
         if tuple(item.record_kind for item in loaded) != expected_kinds:
             raise RecordedSourceAdmissionError("recorded admission record kinds changed")
         acquisitions: list[RecordedSourceAcquisitionReceiptV1Alpha1] = []
         snapshots: list[CanonicalSourceSnapshotV1Alpha1] = []
         observations: list[ObservationV1Alpha1] = []
+        entities: list[EntitySnapshotV1Alpha1] = []
         for index, material in enumerate(expected):
-            offset = index * 3
+            offset = index * 4
             try:
                 acquisition = RecordedSourceAcquisitionReceiptV1Alpha1.model_validate(loaded[offset].payload)
                 snapshot = CanonicalSourceSnapshotV1Alpha1.model_validate(loaded[offset + 1].payload)
                 observation = ObservationV1Alpha1.model_validate(loaded[offset + 2].payload)
+                entity = EntitySnapshotV1Alpha1.model_validate(loaded[offset + 3].payload)
             except (TypeError, ValueError) as exc:
                 raise RecordedSourceAdmissionError("recorded admission payload failed exact replay") from exc
             if (
@@ -596,11 +630,21 @@ async def _replay(
                 or observation.activation_revision != self.binding.prepared_binding.reference
                 or observation.mode is not IntelligenceResourceMode.PREPARED
                 or observation.acquisition_mode is not EvidenceAcquisitionMode.RECORDED_REPLAY
+                or entity.product_id != self.build.product_id
+                or entity.mode is not IntelligenceResourceMode.PREPARED
+                or entity.activation_revision != self.binding.prepared_binding.reference
+                or len(entity.lineage) != 1
+                or entity.lineage[0].resource_id != observation.resource_id
+                or entity.lineage[0].resource_digest != observation.resource_digest
+                or entity.lineage[0].resource_as_of != observation.as_of
+                or entity.lineage[0].resource_available_at != resource_available_at(observation)
+                or loaded[offset + 3].available_at != resource_available_at(entity)
             ):
                 raise RecordedSourceAdmissionError("recorded admission chain crossed exact governed material")
             acquisitions.append(acquisition)
             snapshots.append(snapshot)
             observations.append(observation)
+            entities.append(entity)
         first_acquisition = acquisitions[0]
         expected_preconditions = tuple(
             sorted(
@@ -617,6 +661,7 @@ async def _replay(
             acquisition_receipts=tuple(acquisitions),
             source_snapshots=tuple(snapshots),
             observations=tuple(observations),
+            entity_snapshots=tuple(entities),
             transaction_receipt=receipt,
             replayed=replayed,
         )

tests/intelligence/test_recorded_source_admission.py

@@ -153,7 +153,7 @@ async def _stack():
 
 
 @pytest.mark.asyncio
-async def test_recorded_replay_admits_canonical_observation_and_reopens_exactly() -> None:
+async def test_recorded_replay_admits_canonical_observation_entity_and_reopens_exactly() -> None:
     binding, build, records, material = await _stack()
     service = CoreRecordedSourceAdmissionService(build=build, binding=binding, store=records)
 
@@ -173,11 +173,16 @@ async def test_recorded_replay_admits_canonical_observation_and_reopens_exactly(
     assert first.observations[0].mode is IntelligenceResourceMode.PREPARED
     assert first.observations[0].acquisition_mode is EvidenceAcquisitionMode.RECORDED_REPLAY
     assert first.observations[0].acquisition_receipt_ref == acquisition.receipt_id
+    assert first.entity_snapshots[0].mode is IntelligenceResourceMode.PREPARED
+    assert first.entity_snapshots[0].entity_ref == material.subject_binding.entity_ref
+    assert first.entity_snapshots[0].lineage[0].resource_id == first.observations[0].resource_id
+    assert len(first.transaction_receipt.records) == 4
     assert len(first.transaction_receipt.governed_state_preconditions) == 2
     assert replay.replayed is True
     assert replay.acquisition_receipts == first.acquisition_receipts
     assert replay.source_snapshots == first.source_snapshots
     assert replay.observations == first.observations
+    assert replay.entity_snapshots == first.entity_snapshots
     assert replay.transaction_receipt == first.transaction_receipt
 
 
@@ -209,6 +214,16 @@ async def test_recorded_observation_is_visible_from_fresh_canonical_resource_pro
     assert page.records[0].reference.resource_id == admitted.observations[0].resource_id
     assert page.records[0].reference.resource_kind is IntelligenceResourceKind.OBSERVATION
 
+    entity_query = query.model_copy(update={"resource_kinds": (IntelligenceResourceKind.ENTITY,)})
+    entity_page = await IntelligenceLedgerResourceProjectionReader(store=records).read(
+        query=entity_query,
+        after=None,
+        limit=10,
+    )
+    assert len(entity_page.records) == 1
+    assert entity_page.records[0].reference.resource_id == admitted.entity_snapshots[0].resource_id
+    assert entity_page.records[0].reference.resource_kind is IntelligenceResourceKind.ENTITY
+
 
 @pytest.mark.asyncio
 async def test_substituted_recorded_material_fails_before_any_write() -> None: