diff --git a/ace/intelligence/contracts/resources.py b/ace/intelligence/contracts/resources.py index 81aad27..8f460a5 100644 --- a/ace/intelligence/contracts/resources.py +++ b/ace/intelligence/contracts/resources.py @@ -524,8 +524,8 @@ def prevent_fixture_as_live(self) -> Self: raise ValueError("a live Observation requires live acquisition") if self.observed_at > self.ingested_at: raise ValueError("Observation ingested_at cannot precede observed_at") - if self.ingested_at > self.as_of: - raise ValueError("Observation as_of cannot precede ingested_at") + if self.as_of > self.ingested_at: + raise ValueError("Observation as_of cannot follow ingested_at") if self.source_published_at is not None and self.source_published_at > self.observed_at: raise ValueError("Observation source_published_at cannot follow observed_at") if self.source_mapping is not None and self.source_mapping.activation_revision != self.activation_revision: diff --git a/ace/intelligence/detection/categorical_transition.py b/ace/intelligence/detection/categorical_transition.py index 1d4e3bf..abada17 100644 --- a/ace/intelligence/detection/categorical_transition.py +++ b/ace/intelligence/detection/categorical_transition.py @@ -209,9 +209,9 @@ def _validate_pair( ) if baseline.as_of >= current.as_of: raise CategoricalTransitionDetectionError("categorical transition baseline must precede current state") - if baseline.as_of < binding.revision.occurred_at: + if min(baseline.projected_at, current.projected_at) < binding.revision.occurred_at: raise CategoricalTransitionDetectionError( - "categorical transition baseline predates the prepared activation revision" + "categorical transition snapshot projection predates the prepared activation revision" ) available_at = max(baseline.projected_at, current.projected_at) if _aware_utc(detected_at, label="detected_at") < available_at: @@ -388,8 +388,8 @@ def _route_categorical_shift_as_signal( raise CategoricalTransitionDetectionError("the Shift does not use the exact bound activation revision") if validated_shift.product_id != validated_binding.revision.spec.product_id: raise CategoricalTransitionDetectionError("the Shift is outside the bound product scope") - if validated_shift.as_of < validated_binding.revision.occurred_at: - raise CategoricalTransitionDetectionError("the Shift predates the prepared activation revision") + if validated_shift.detected_at < validated_binding.revision.occurred_at: + raise CategoricalTransitionDetectionError("the Shift detection predates the prepared activation revision") if validated_shift.shift_type_ref != rule.shift_type: raise CategoricalTransitionDetectionError("the Shift does not match the bound detector rule") delta = validated_shift.delta.parsed_value() diff --git a/ace/intelligence/detection/numeric_delta.py b/ace/intelligence/detection/numeric_delta.py index 41f24dc..016ad58 100644 --- a/ace/intelligence/detection/numeric_delta.py +++ b/ace/intelligence/detection/numeric_delta.py @@ -208,8 +208,8 @@ def _validate_pair( raise NumericDeltaDetectionError("numeric delta rule does not target the snapshot entity type") if baseline.as_of >= current.as_of: raise NumericDeltaDetectionError("numeric delta baseline must precede current state") - if baseline.as_of < binding.revision.occurred_at: - raise NumericDeltaDetectionError("numeric delta baseline predates the prepared activation revision") + if min(baseline.projected_at, current.projected_at) < binding.revision.occurred_at: + raise NumericDeltaDetectionError("numeric delta snapshot projection predates the prepared activation revision") available_at = max(baseline.projected_at, current.projected_at) if _aware_utc(detected_at, label="detected_at") < available_at: raise NumericDeltaDetectionError("numeric delta cannot be detected before both snapshots were projected") @@ -423,8 +423,8 @@ def _route_shift_as_signal( raise NumericDeltaDetectionError("the Shift does not use the exact bound activation revision") if validated_shift.product_id != validated_binding.revision.spec.product_id: raise NumericDeltaDetectionError("the Shift is outside the bound product scope") - if validated_shift.as_of < validated_binding.revision.occurred_at: - raise NumericDeltaDetectionError("the Shift predates the prepared activation revision") + if validated_shift.detected_at < validated_binding.revision.occurred_at: + raise NumericDeltaDetectionError("the Shift detection predates the prepared activation revision") if validated_shift.shift_type_ref != rule.shift_type: raise NumericDeltaDetectionError("the Shift does not match the bound detector rule") delta = validated_shift.delta.parsed_value() diff --git a/ace/intelligence/routing.py b/ace/intelligence/routing.py index 82fd4ab..336d576 100644 --- a/ace/intelligence/routing.py +++ b/ace/intelligence/routing.py @@ -58,8 +58,8 @@ def _eligible_signal_routes( raise SignalRoutingError("Signal does not use the exact bound activation revision") if validated_signal.product_id != validated_binding.revision.spec.product_id: raise SignalRoutingError("Signal is outside the bound product scope") - if validated_signal.as_of < validated_binding.revision.occurred_at: - raise SignalRoutingError("Signal predates the prepared activation revision") + if validated_signal.detected_at < validated_binding.revision.occurred_at: + raise SignalRoutingError("Signal detection predates the prepared activation revision") routes = ( route diff --git a/ace/intelligence/source_mapping.py b/ace/intelligence/source_mapping.py index d08b483..4aef59d 100644 --- a/ace/intelligence/source_mapping.py +++ b/ace/intelligence/source_mapping.py @@ -368,11 +368,12 @@ def _interpret_source_mapping( mapping_id=mapping.mapping_id, mapping_digest=f"sha256:{canonical_hash(mapping)}", ) + state_as_of = snapshot.event_effective_at or snapshot.source_published_at or snapshot.observed_at observation = ObservationV1Alpha1( product_id=product_id, mode=mode, activation_revision=validated_binding.reference, - as_of=snapshot.ingested_at, + as_of=state_as_of, source_ref=snapshot.source_snapshot_ref, source_digest=snapshot.source_snapshot_digest, acquisition_mode=acquisition_mode, @@ -393,7 +394,7 @@ def _interpret_source_mapping( product_id=product_id, mode=mode, activation_revision=validated_binding.reference, - as_of=snapshot.ingested_at, + as_of=state_as_of, lineage=( LineageReferenceV1Alpha1( resource_kind=LineageResourceKind.OBSERVATION, diff --git a/ace/intelligence/synthesis.py b/ace/intelligence/synthesis.py index 9716b8b..1eb6608 100644 --- a/ace/intelligence/synthesis.py +++ b/ace/intelligence/synthesis.py @@ -176,7 +176,7 @@ def _lineage(resource) -> LineageReferenceV1Alpha1: def _citation(observation: ObservationV1Alpha1) -> CitationV1Alpha1: - source_as_of = observation.source_published_at or observation.event_effective_at or observation.observed_at + source_as_of = observation.event_effective_at or observation.source_published_at or observation.observed_at return CitationV1Alpha1( source_ref=observation.source_ref, source_digest=observation.source_digest, diff --git a/docs/evidence/platform-p1c1-declarative-source-mapping-v1.md b/docs/evidence/platform-p1c1-declarative-source-mapping-v1.md index e273dd9..d5af167 100644 --- a/docs/evidence/platform-p1c1-declarative-source-mapping-v1.md +++ b/docs/evidence/platform-p1c1-declarative-source-mapping-v1.md @@ -24,6 +24,13 @@ Observation and one exact-lineage Entity Snapshot. executable. - Product, activation, mode, source and receipt identities, and all times come from the host envelope and resolved binding. Source payload labels cannot override them. +- Mapped intelligence state time is the first available semantic timestamp in the strict order + event-effective, source-published, then observed. Ingestion and projection retain actual + availability time. This bitemporal correction was reverified on 2026-08-13 and intentionally + changes mapped Observation and Entity Snapshot identities. +- Detection and routing compare semantic state in `as_of`, but activation admissibility is checked + against projection and detection availability. Historical source state may predate installation; + projecting, detecting, or routing it before the exact activation remains fail-closed. - Observation provenance durably pins the activation revision, compiled Pack ID and digest, source-mapping module ID and digest, and mapping ID and digest. The Entity Snapshot's single lineage edge pins that exact Observation without adding domain data to its attributes. @@ -58,8 +65,8 @@ branches. Their pinned outputs are: | Fixture | Observation | Entity Snapshot | |---|---|---| -| numeric | `observation:b5f4394738f8b5e50049d251a86c57e6` / `sha256:b5f4394738f8b5e50049d251a86c57e6ae2ef8df3adb2e620739d3c80c5b0244` | `entity_snapshot:34e55f3de2964b0ead40b4f3139069a7` / `sha256:34e55f3de2964b0ead40b4f3139069a73d70837eff877c8961b0f337f5b21f0f` | -| categorical | `observation:66426d907e83a36f6fe9eb8b619fade2` / `sha256:66426d907e83a36f6fe9eb8b619fade207c0a256f1eeec0089556f023e756912` | `entity_snapshot:c6103b731c5dd2e02694650c0d213e77` / `sha256:c6103b731c5dd2e02694650c0d213e77f136108562fdf9319647042f67a0905c` | +| numeric | `observation:1a12528799c8192eef0f450608da203b` / `sha256:1a12528799c8192eef0f450608da203b228c5dc220a19c267aaa14fe75efc035` | `entity_snapshot:935bb7668ce0cce0655877877793db66` / `sha256:935bb7668ce0cce0655877877793db663ee7aa3e7c3b6dcfc58264fed40f197e` | +| categorical | `observation:db365d16a6ed9a399d36f90a96a67621` / `sha256:db365d16a6ed9a399d36f90a96a676213f0aa37677f6b31be48324787addafe1` | `entity_snapshot:377c44ca271b576f3768768b47a23446` / `sha256:377c44ca271b576f3768768b47a234468b22288eb7e56da9c762f5923487bb84` | Semantically reordered declarations and `0.0`/`-0.0` confidence compile to identical Pack IR and identity; material changes alter Pack identity. Regression coverage also proves exact source and @@ -70,6 +77,18 @@ normalization, and rejection of every LIVE attempt. ## Verification record +- 2026-08-13 bitemporal correction: **54 passed** across source mapping, resource contracts, and + recorded admission; **160 passed** across LIVE ingestion, numeric/categorical detection, + synthesis, resource projections, and public boundaries. The full supported gate reached + **7,976 passed, 50 skipped, 262 deselected** with one unrelated persistent-store transaction + failure; that exact failing test passed immediately in isolation. One recorded two-snapshot + transaction proves distinct semantic state times survive a shared admission time and are + accepted by the unchanged numeric detector. +- Follow-up activation-time audit: **109 passed, 2 skipped** across numeric and categorical + PREPARED/LIVE detection, routing, source mapping, recorded admission, solution Pack behavior, + and public boundaries. Positive and negative fixtures distinguish historical state from + post-activation projection/detection without weakening exact activation binding. + - Focused source-mapping, compiler, resource, ledger, boundary, artifact-contract, and build-backend gate: **89 passed, 1 deselected**. - Complete Intelligence suite: **132 passed**. diff --git a/tests/intelligence/test_categorical_transition_detection.py b/tests/intelligence/test_categorical_transition_detection.py index 8421bd8..97f9577 100644 --- a/tests/intelligence/test_categorical_transition_detection.py +++ b/tests/intelligence/test_categorical_transition_detection.py @@ -133,7 +133,7 @@ def _compiled_pack( return compile_pack_document(_encoded(manifest), resources) -def _binding(**pack_changes: Any) -> PreparedActivationBinding: +def _binding(*, activated_at: datetime | None = None, **pack_changes: Any) -> PreparedActivationBinding: pack = _compiled_pack(**pack_changes) overlay = compile_overlay( pack, @@ -158,7 +158,7 @@ def _binding(**pack_changes: Any) -> PreparedActivationBinding: state=ActivationState.ACTIVE, actor_ref="principal:test-author", approval_receipt_ref="receipt:prepared-approval", - occurred_at=AS_OF - timedelta(days=100), + occurred_at=activated_at or AS_OF - timedelta(days=100), ) return bind_prepared_activation(pack=pack, revision=revision) @@ -621,6 +621,58 @@ def test_snapshot_pair_time_and_scope_discipline_is_enforced() -> None: ) +def test_historical_categorical_state_requires_post_activation_processing() -> None: + from ace.intelligence import route_categorical_shift_as_signal + from ace.intelligence.detection import CategoricalTransitionDetectionError + + activated_at = AS_OF - timedelta(hours=1) + binding = _binding(activated_at=activated_at) + baseline = _snapshot( + binding, + "draft", + as_of=AS_OF - timedelta(days=10), + projected_at=AS_OF, + ) + current = _snapshot( + binding, + "active", + as_of=AS_OF - timedelta(days=5), + projected_at=AS_OF, + ) + + shift = _detect(binding, baseline, current, detected_at=AS_OF) + assert shift is not None + assert shift.as_of < activated_at < shift.detected_at + signal = route_categorical_shift_as_signal( + binding=binding, + detector_id=DETECTOR_ID, + shift=shift, + detected_at=AS_OF, + ) + assert signal.as_of < activated_at <= signal.detected_at + + preactivation_current = _snapshot( + binding, + "active", + as_of=AS_OF - timedelta(days=5), + projected_at=activated_at - timedelta(seconds=1), + ) + with pytest.raises(CategoricalTransitionDetectionError, match="projection predates"): + _detect(binding, baseline, preactivation_current, detected_at=AS_OF) + + shift_material = shift.model_dump(mode="python", exclude={"resource_id", "resource_digest"}) + shift_material["detected_at"] = activated_at - timedelta(seconds=1) + shift_material["lineage"] = () + preactivation_shift = shift.__class__.model_validate(shift_material) + with pytest.raises(CategoricalTransitionDetectionError, match="detection predates"): + route_categorical_shift_as_signal( + binding=binding, + detector_id=DETECTOR_ID, + shift=preactivation_shift, + detected_at=activated_at, + ) + + def test_cross_pack_policy_cannot_relabel_or_route_a_shift() -> None: from ace.intelligence import route_categorical_shift_as_signal from ace.intelligence.detection import CategoricalTransitionDetectionError diff --git a/tests/intelligence/test_intelligence_resource_contracts.py b/tests/intelligence/test_intelligence_resource_contracts.py index 9a4e604..c882746 100644 --- a/tests/intelligence/test_intelligence_resource_contracts.py +++ b/tests/intelligence/test_intelligence_resource_contracts.py @@ -441,6 +441,22 @@ def test_observation_cannot_precede_its_claimed_source_publication() -> None: ) +def test_observation_state_time_cannot_follow_its_ingestion_availability() -> None: + with pytest.raises(ValidationError, match="as_of cannot follow ingested_at"): + ObservationV1Alpha1( + **_common(), + source_ref="evidence:future-state", + source_digest="sha256:" + "4" * 64, + acquisition_mode=EvidenceAcquisitionMode.PREPARED_FIXTURE, + acquisition_receipt_ref="receipt:future-state-acquisition", + acquisition_receipt_digest="sha256:" + "4" * 64, + observed_at=AS_OF - timedelta(seconds=2), + ingested_at=AS_OF - timedelta(seconds=1), + payload=_json(), + confidence=0.5, + ) + + def test_explicit_inference_requires_basis_and_uncertainty_but_not_a_fake_citation() -> None: basis_digest = "sha256:" + "3" * 64 basis_ref = "shift:" + "3" * 32 diff --git a/tests/intelligence/test_numeric_delta_detection.py b/tests/intelligence/test_numeric_delta_detection.py index 7dcfb00..05d3573 100644 --- a/tests/intelligence/test_numeric_delta_detection.py +++ b/tests/intelligence/test_numeric_delta_detection.py @@ -132,7 +132,7 @@ def _compiled_pack( ) -def _binding(**pack_changes) -> PreparedActivationBinding: +def _binding(*, activated_at: datetime | None = None, **pack_changes) -> PreparedActivationBinding: pack = _compiled_pack(**pack_changes) overlay = compile_overlay( pack, @@ -157,7 +157,7 @@ def _binding(**pack_changes) -> PreparedActivationBinding: state=ActivationState.ACTIVE, actor_ref="principal:test-author", approval_receipt_ref="receipt:prepared-approval", - occurred_at=AS_OF - timedelta(days=100), + occurred_at=activated_at or AS_OF - timedelta(days=100), ) return bind_prepared_activation(pack=pack, revision=revision) @@ -345,6 +345,71 @@ def test_detection_and_signal_timestamps_follow_input_availability() -> None: ) +def test_semantic_state_may_predate_activation_but_processing_may_not() -> None: + activated_at = AS_OF - timedelta(hours=1) + binding = _binding(activated_at=activated_at) + baseline = _snapshot( + binding, + 100.0, + as_of=AS_OF - timedelta(days=10), + projected_at=AS_OF, + ) + current = _snapshot( + binding, + 90.0, + as_of=AS_OF - timedelta(days=5), + projected_at=AS_OF, + ) + + shift = _detect(binding, baseline, current, detected_at=AS_OF) + assert shift is not None + assert shift.as_of < activated_at < shift.detected_at + signal = route_shift_as_signal( + binding=binding, + detector_id=DETECTOR_ID, + shift=shift, + detected_at=AS_OF, + ) + assert signal.as_of < activated_at <= signal.detected_at + assert eligible_signal_routes(binding=binding, signal=signal) == () + + preactivation_baseline = _snapshot( + binding, + 100.0, + as_of=AS_OF - timedelta(days=10), + projected_at=activated_at - timedelta(seconds=1), + ) + with pytest.raises(NumericDeltaDetectionError, match="projection predates"): + _detect(binding, preactivation_baseline, current, detected_at=AS_OF) + + shift_material = shift.model_dump(mode="python", exclude={"resource_id", "resource_digest"}) + shift_material["detected_at"] = activated_at - timedelta(seconds=1) + shift_material["lineage"] = () + preactivation_shift = shift.__class__.model_validate(shift_material) + with pytest.raises(NumericDeltaDetectionError, match="detection predates"): + route_shift_as_signal( + binding=binding, + detector_id=DETECTOR_ID, + shift=preactivation_shift, + detected_at=activated_at, + ) + + preactivation_signal = SignalV1Alpha1( + product_id=PRODUCT_ID, + mode=IntelligenceResourceMode.PREPARED, + activation_revision=binding.reference, + as_of=AS_OF - timedelta(days=5), + signal_type_ref="measure_attention", + title="Pre-activation processing is invalid", + summary="Historical state cannot be routed before the activation exists.", + details=CanonicalJsonValueV1Alpha1(value_json="{}"), + detected_at=activated_at - timedelta(seconds=1), + confidence=0.9, + ) + with pytest.raises(SignalRoutingError, match="detection predates"): + eligible_signal_routes(binding=binding, signal=preactivation_signal) + + def test_comparison_context_prevents_cross_unit_delta() -> None: binding = _binding(context_attribute_ids=["unit"]) with pytest.raises(NumericDeltaDetectionError, match="context attribute unit changed"): diff --git a/tests/intelligence/test_recorded_source_admission.py b/tests/intelligence/test_recorded_source_admission.py index d3ca4bc..3723ad7 100644 --- a/tests/intelligence/test_recorded_source_admission.py +++ b/tests/intelligence/test_recorded_source_admission.py @@ -1,6 +1,7 @@ from __future__ import annotations import hashlib +from dataclasses import replace from datetime import UTC, datetime, timedelta import pytest @@ -26,14 +27,26 @@ canonical_json, ) from ace.core.runtime_use import AUTHORITY_GRANT_STATE_KIND -from ace.intelligence import EvidenceAcquisitionMode, IntelligenceResourceMode +from ace.intelligence import ( + EvidenceAcquisitionMode, + IntelligenceResourceMode, + detect_numeric_shift, +) +from ace.intelligence.contracts.pack import DomainPackManifestV1 from ace.intelligence.contracts.resource_plane import ( IntelligenceResourceKind, IntelligenceResourceQueryV1Alpha1, ) +from ace.intelligence.packs.compiler import compile_pack from ace.testing import InMemoryImmutableRecordStore +from tests.intelligence.conftest import digest_bytes, encode_json from tests.intelligence.test_domain_activation_admission import _Authority, _MemoryStore -from tests.intelligence.test_source_mapping import _binding, _compiled, _fixture_documents, _subject +from tests.intelligence.test_source_mapping import ( + _binding, + _fixture_documents, + _manifest_and_resources, + _subject, +) pytestmark = pytest.mark.unit @@ -43,8 +56,52 @@ ADMITTED_AT = datetime(2026, 8, 13, 18, tzinfo=UTC) +def _compiled_recorded_numeric_pack(): + ontology, mapping, _ = _fixture_documents("numeric") + manifest, resources = _manifest_and_resources(ontology, mapping) + detection = { + "contract": "ace.intelligence.detection/v1alpha1", + "module_id": "detection", + "numeric_delta_rules": [ + { + "detector_id": "material_reading_change", + "entity_type_id": "reading", + "attribute_id": "value", + "baseline": "prior_snapshot", + "context_attribute_ids": ["code"], + "metric": "percent_change", + "threshold": 5.0, + "direction": "any", + "shift_type": "material_reading_change", + "signal_type": "reading_attention", + } + ], + } + detection_bytes = encode_json(detection) + resources["modules/detection.json"] = detection_bytes + material = manifest.model_dump(mode="python") + material["resources"] = ( + *material["resources"], + { + "resource_id": "detection_resource", + "path": "modules/detection.json", + "digest": digest_bytes(detection_bytes), + }, + ) + material["modules"] = ( + *material["modules"], + { + "module_id": "detection", + "contract": "ace.intelligence.detection/v1alpha1", + "resource_id": "detection_resource", + "depends_on": ("ontology",), + }, + ) + return compile_pack(DomainPackManifestV1.model_validate(material), resources) + + async def _stack(): - pack = _compiled("numeric") + pack = _compiled_recorded_numeric_pack() prepared = _binding(pack, product_id=PRODUCT) activation_store = _MemoryStore() committed = await DomainActivationAdmissionService( @@ -186,6 +243,65 @@ async def test_recorded_replay_admits_canonical_observation_entity_and_reopens_e assert replay.transaction_receipt == first.transaction_receipt +@pytest.mark.asyncio +async def test_one_recorded_batch_preserves_distinct_state_times_for_detection() -> None: + binding, build, records, baseline_material = await _stack() + current_payload = canonical_json({"reading": {"value": "90.000"}, "subject": {"code": "AX"}}) + current_material = RecordedSourceMaterialV1Alpha1( + **baseline_material.model_dump( + mode="python", + exclude={ + "source_uri", + "captured_payload_json", + "captured_payload_digest", + "source_published_at", + "event_effective_at", + "observed_at", + "locator", + "material_id", + "material_digest", + }, + ), + source_uri="https://example.invalid/recorded/reading-2", + captured_payload_json=current_payload, + captured_payload_digest="sha256:" + hashlib.sha256(current_payload.encode()).hexdigest(), + source_published_at=OBSERVED_AT + timedelta(minutes=10), + event_effective_at=OBSERVED_AT + timedelta(minutes=15), + observed_at=OBSERVED_AT + timedelta(minutes=30), + locator="record:2", + ) + refs = tuple( + RecordedSourceReferenceV1( + source_group_id=item.source_group_id, + material_id=str(item.material_id), + material_digest=str(item.material_digest), + ) + for item in (baseline_material, current_material) + ) + request_material = build.request.model_dump(mode="python") + request_material["recorded_source_refs"] = refs + build = replace(build, request=IntelligenceBuildStartV1.model_validate(request_material)) + + admitted = await CoreRecordedSourceAdmissionService(build=build, binding=binding, store=records).admit( + (baseline_material, current_material) + ) + baseline, current = sorted(admitted.entity_snapshots, key=lambda item: item.as_of) + shift = detect_numeric_shift( + binding=binding.prepared_binding, + detector_id="material_reading_change", + baseline=baseline, + current=current, + detected_at=ADMITTED_AT, + ) + + assert baseline.as_of == baseline_material.event_effective_at + assert current.as_of == current_material.event_effective_at + assert baseline.as_of < current.as_of < baseline.projected_at == current.projected_at == ADMITTED_AT + assert shift is not None + assert shift.baseline_as_of == baseline.as_of + assert shift.as_of == current.as_of + + @pytest.mark.asyncio async def test_recorded_observation_is_visible_from_fresh_canonical_resource_projection() -> None: binding, build, records, material = await _stack() diff --git a/tests/intelligence/test_source_mapping.py b/tests/intelligence/test_source_mapping.py index f15add2..4add30b 100644 --- a/tests/intelligence/test_source_mapping.py +++ b/tests/intelligence/test_source_mapping.py @@ -319,6 +319,49 @@ def test_numeric_and_categorical_fixtures_use_one_branch_free_public_interpreter signature = set(inspect.signature(interpret_prepared_source_mapping).parameters) assert signature == {"binding", "mapping_id", "source_snapshot", "subject_binding"} + +def test_mapping_uses_semantic_state_time_and_keeps_actual_availability() -> None: + compiled = _compiled("numeric") + binding = _binding(compiled) + subject = _subject(binding, "numeric") + snapshot = _snapshot("numeric") + + def rematerialize(**updates): + material = snapshot.model_dump( + mode="python", + exclude={"source_snapshot_ref", "source_snapshot_digest"}, + ) + material.update(updates) + return CanonicalSourceSnapshotV1Alpha1.model_validate(material) + + event = interpret_prepared_source_mapping( + binding=binding, + mapping_id="reading_snapshot", + source_snapshot=snapshot, + subject_binding=subject, + ) + publication = interpret_prepared_source_mapping( + binding=binding, + mapping_id="reading_snapshot", + source_snapshot=rematerialize(event_effective_at=None), + subject_binding=subject, + ) + observation = interpret_prepared_source_mapping( + binding=binding, + mapping_id="reading_snapshot", + source_snapshot=rematerialize(event_effective_at=None, source_published_at=None), + subject_binding=subject, + ) + + assert event.observation.as_of == event.entity_snapshot.as_of == snapshot.event_effective_at + assert publication.observation.as_of == publication.entity_snapshot.as_of == snapshot.source_published_at + assert observation.observation.as_of == observation.entity_snapshot.as_of == snapshot.observed_at + for mapped in (event, publication, observation): + assert mapped.observation.ingested_at == INGESTED_AT + assert mapped.entity_snapshot.projected_at == INGESTED_AT + assert mapped.entity_snapshot.lineage[0].resource_as_of == mapped.observation.as_of + assert mapped.entity_snapshot.lineage[0].resource_available_at == INGESTED_AT + _, _, numeric = _interpret("numeric") _, _, categorical = _interpret("categorical") @@ -357,7 +400,7 @@ def test_numeric_and_categorical_fixtures_use_one_branch_free_public_interpreter assert conformance.lineage_relation is LineageRelation.DERIVED_FROM assert conformance.lineage_resource_id == conformance.observation_id assert conformance.lineage_resource_digest == conformance.observation_digest - assert conformance.lineage_resource_as_of == INGESTED_AT + assert conformance.lineage_resource_as_of == OBSERVED_AT - timedelta(minutes=30) assert conformance.lineage_resource_available_at == INGESTED_AT @@ -497,10 +540,10 @@ def test_exact_prepared_outputs_are_pinned() -> None: numeric.entity_snapshot.resource_id, numeric.entity_snapshot.resource_digest, ) == ( - "observation:b5f4394738f8b5e50049d251a86c57e6", - "sha256:b5f4394738f8b5e50049d251a86c57e6ae2ef8df3adb2e620739d3c80c5b0244", - "entity_snapshot:34e55f3de2964b0ead40b4f3139069a7", - "sha256:34e55f3de2964b0ead40b4f3139069a73d70837eff877c8961b0f337f5b21f0f", + "observation:1a12528799c8192eef0f450608da203b", + "sha256:1a12528799c8192eef0f450608da203b228c5dc220a19c267aaa14fe75efc035", + "entity_snapshot:935bb7668ce0cce0655877877793db66", + "sha256:935bb7668ce0cce0655877877793db663ee7aa3e7c3b6dcfc58264fed40f197e", ) assert ( categorical.observation.resource_id, @@ -508,10 +551,10 @@ def test_exact_prepared_outputs_are_pinned() -> None: categorical.entity_snapshot.resource_id, categorical.entity_snapshot.resource_digest, ) == ( - "observation:66426d907e83a36f6fe9eb8b619fade2", - "sha256:66426d907e83a36f6fe9eb8b619fade207c0a256f1eeec0089556f023e756912", - "entity_snapshot:c6103b731c5dd2e02694650c0d213e77", - "sha256:c6103b731c5dd2e02694650c0d213e77f136108562fdf9319647042f67a0905c", + "observation:db365d16a6ed9a399d36f90a96a67621", + "sha256:db365d16a6ed9a399d36f90a96a676213f0aa37677f6b31be48324787addafe1", + "entity_snapshot:377c44ca271b576f3768768b47a23446", + "sha256:377c44ca271b576f3768768b47a234468b22288eb7e56da9c762f5923487bb84", )