From 8e077e25765e8b4c53eb49b04d13087fda0e65ae Mon Sep 17 00:00:00 2001 From: Edwin Amirian Date: Thu, 13 Aug 2026 20:37:12 -0700 Subject: [PATCH] Correct Brief analysis cutoff semantics --- ace/application/brief_synthesis.py | 4 +- .../intelligence_build_first_brief.py | 3 +- ace/application/live_intelligence_bridge.py | 2 +- ace/intelligence/contracts/synthesis.py | 18 +++++--- tests/intelligence/test_brief_synthesis.py | 46 +++++++++---------- .../intelligence/test_case_brief_synthesis.py | 6 ++- 6 files changed, 40 insertions(+), 39 deletions(-) diff --git a/ace/application/brief_synthesis.py b/ace/application/brief_synthesis.py index 28de178..2e80560 100644 --- a/ace/application/brief_synthesis.py +++ b/ace/application/brief_synthesis.py @@ -679,8 +679,8 @@ async def _exact_closure_unchecked( loaded_signal = await ledger.load_exact(attention.signal) if not isinstance(loaded_signal, SignalV1Alpha1): raise BriefSynthesisError("routed attention Signal is missing from exact PREPARED scope") - if loaded_signal.as_of != request.brief_as_of: - raise BriefSynthesisError("Brief cutoff must equal the routed Signal as_of time") + if loaded_signal.as_of > request.brief_as_of: + raise BriefSynthesisError("routed Signal semantic as_of cannot follow the Brief cutoff") if len(loaded_signal.lineage) != 1 or any( item.resource_kind is not LineageResourceKind.SHIFT or item.relation is not LineageRelation.DERIVED_FROM for item in loaded_signal.lineage diff --git a/ace/application/intelligence_build_first_brief.py b/ace/application/intelligence_build_first_brief.py index c6bb26a..d527895 100644 --- a/ace/application/intelligence_build_first_brief.py +++ b/ace/application/intelligence_build_first_brief.py @@ -384,7 +384,6 @@ async def create_first_brief( ) from None except Exception: raise IntelligenceBuildFirstBriefError("routed PREPARED material failed exact replay") from None - signal = signals[0] synthesis_material = { "build_id": self.build.build_id, "request_id": exact.request_id, @@ -405,7 +404,7 @@ async def create_first_brief( pack=binding.prepared_binding.revision.spec.pack, attention_receipt_id=exact.attention_receipt_id, attention_receipt_digest=exact.attention_receipt_digest, - brief_as_of=signal.as_of, + brief_as_of=attention.evaluated_at, context_cutoff_at=attention.evaluated_at, requested_at=exact.requested_at, ) diff --git a/ace/application/live_intelligence_bridge.py b/ace/application/live_intelligence_bridge.py index cdb88ab..3dfdd07 100644 --- a/ace/application/live_intelligence_bridge.py +++ b/ace/application/live_intelligence_bridge.py @@ -811,7 +811,7 @@ async def _derivation( or receipt.attention != attention.record_reference() or receipt.signal != resource_reference(signal) or receipt.shift != resource_reference(shift) - or signal.as_of != request.brief_as_of + or signal.as_of > request.brief_as_of ): raise LiveBriefSynthesisError("LIVE route receipt is missing or cross-wired") try: diff --git a/ace/intelligence/contracts/synthesis.py b/ace/intelligence/contracts/synthesis.py index 40c1459..33590a2 100644 --- a/ace/intelligence/contracts/synthesis.py +++ b/ace/intelligence/contracts/synthesis.py @@ -283,8 +283,10 @@ def validate_scope_time_and_identity(self) -> Self: or self.activation_revision.product_id != self.product_id ): raise ValueError("Brief synthesis request crossed exact product scope") - if not self.brief_as_of <= self.context_cutoff_at <= self.requested_at: - raise ValueError("Brief semantic as_of, context cutoff, and request time must be ordered") + if self.brief_as_of != self.context_cutoff_at: + raise ValueError("Brief as_of must equal its frozen analysis and evidence cutoff") + if self.context_cutoff_at > self.requested_at: + raise ValueError("Brief analysis and evidence cutoff cannot follow request time") if not ( self.authenticated_context.authenticated_at <= self.requested_at < self.authenticated_context.expires_at ): @@ -1092,12 +1094,14 @@ def validate_scope_time_and_identity(self) -> Self: raise ValueError("Case Brief synthesis request crossed exact product scope") if self.case.resource_kind is not IntelligenceRecordKind.CASE or self.case.mode is not self.mode: raise ValueError("Case Brief synthesis request must bind one exact PREPARED Case") - if self.case.as_of != self.brief_as_of: - raise ValueError("Brief cutoff must equal the bound Case as_of closure") - if self.case.available_at > self.context_cutoff_at: + if self.case.as_of > self.brief_as_of: + raise ValueError("the bound Case semantic as_of cannot follow the Brief cutoff") + if self.case.available_at > self.brief_as_of: raise ValueError("the bound Case must be available by the context cutoff") - if not self.brief_as_of <= self.context_cutoff_at <= self.requested_at: - raise ValueError("Brief semantic as_of, context cutoff, and request time must be ordered") + if self.brief_as_of != self.context_cutoff_at: + raise ValueError("Brief as_of must equal its frozen analysis and evidence cutoff") + if self.context_cutoff_at > self.requested_at: + raise ValueError("Brief analysis and evidence cutoff cannot follow request time") if not ( self.authenticated_context.authenticated_at <= self.requested_at < self.authenticated_context.expires_at ): diff --git a/tests/intelligence/test_brief_synthesis.py b/tests/intelligence/test_brief_synthesis.py index 4b8444a..52d00d1 100644 --- a/tests/intelligence/test_brief_synthesis.py +++ b/tests/intelligence/test_brief_synthesis.py @@ -72,11 +72,12 @@ PRODUCT = "product:prepared-brief" ACTIVATED_AT = datetime(2026, 8, 6, 11, 55, tzinfo=UTC) BASELINE_AS_OF = datetime(2026, 8, 6, 12, 0, tzinfo=UTC) -BRIEF_AS_OF = datetime(2026, 8, 6, 12, 1, tzinfo=UTC) +SIGNAL_AS_OF = datetime(2026, 8, 6, 12, 1, tzinfo=UTC) PROJECTED_AT = datetime(2026, 8, 6, 12, 1, 30, tzinfo=UTC) SHIFT_DETECTED_AT = datetime(2026, 8, 6, 12, 2, tzinfo=UTC) SIGNAL_DETECTED_AT = datetime(2026, 8, 6, 12, 2, 30, tzinfo=UTC) ROUTED_AT = datetime(2026, 8, 6, 12, 4, tzinfo=UTC) +BRIEF_AS_OF = ROUTED_AT REQUESTED_AT = datetime(2026, 8, 6, 12, 5, tzinfo=UTC) GENERATED_AT = datetime(2026, 8, 6, 12, 5, 15, tzinfo=UTC) @@ -325,16 +326,16 @@ def _batch(binding, *, same_source: bool = False) -> PreparedResourceAdmissionV1 "source_digest": "sha256:" + "3" * 64, "acquisition_receipt_ref": "acquisition:edge-x1-current", "acquisition_receipt_digest": "sha256:" + "4" * 64, - "source_published_at": BRIEF_AS_OF, + "source_published_at": SIGNAL_AS_OF, } second = ObservationV1Alpha1( product_id=PRODUCT, mode=IntelligenceResourceMode.PREPARED, activation_revision=reference, - as_of=BRIEF_AS_OF, + as_of=SIGNAL_AS_OF, **second_source, - observed_at=BRIEF_AS_OF, - ingested_at=BRIEF_AS_OF, + observed_at=SIGNAL_AS_OF, + ingested_at=SIGNAL_AS_OF, subject_refs=("entity:edge-x1",), payload=CanonicalJsonValueV1Alpha1( value_json=canonical_json({"name": "Edge X1", "price": {"amount": 1080, "currency": "USD"}}) @@ -346,10 +347,10 @@ def _batch(binding, *, same_source: bool = False) -> PreparedResourceAdmissionV1 product_id=PRODUCT, mode=IntelligenceResourceMode.PREPARED, activation_revision=reference, - as_of=BRIEF_AS_OF, + as_of=SIGNAL_AS_OF, **second_source, - observed_at=BRIEF_AS_OF, - ingested_at=BRIEF_AS_OF, + observed_at=SIGNAL_AS_OF, + ingested_at=SIGNAL_AS_OF, subject_refs=("entity:edge-x1",), payload=CanonicalJsonValueV1Alpha1( value_json=canonical_json( @@ -381,7 +382,7 @@ def _batch(binding, *, same_source: bool = False) -> PreparedResourceAdmissionV1 product_id=PRODUCT, mode=IntelligenceResourceMode.PREPARED, activation_revision=reference, - as_of=BRIEF_AS_OF, + as_of=SIGNAL_AS_OF, lineage=tuple( _lineage(item, LineageResourceKind.OBSERVATION) for item in (second, convergent) if item is not None ), @@ -866,15 +867,15 @@ async def test_live_then_fresh_service_replay_is_exact_authorized_and_provider_o assert env.request.activation_revision.revision_digest == ( "sha256:caeb8fcafd17a6ba50741d44837d9980f755549d82cabcdb64d127efab72fff0" ) - assert first.brief.resource_id == "brief:52d3d753b9b2ee30d1a8faaa316e1652" - assert first.brief.resource_digest == ("sha256:52d3d753b9b2ee30d1a8faaa316e16526b3a6e5e4cf793d8417df8b91fe6a206") - assert first.synthesis_receipt.receipt_id == ("brief_synthesis_receipt:13083446bbc36acc4ae97e2a96a22700") + assert first.brief.resource_id == "brief:b5974bba19f9cbd4fa854db254c063db" + assert first.brief.resource_digest == ("sha256:b5974bba19f9cbd4fa854db254c063db4a111204fe4fa6c3a5650a447e6dfe95") + assert first.synthesis_receipt.receipt_id == ("brief_synthesis_receipt:da04b6b9c195d0cf98fe7d5ad969df4a") assert first.synthesis_receipt.receipt_digest == ( - "sha256:13083446bbc36acc4ae97e2a96a22700087d98c9b4e5b3145a68c8b94439dfc1" + "sha256:da04b6b9c195d0cf98fe7d5ad969df4a5f0c226b64e34c011bc10720d07b6de9" ) assert first.transaction_receipt.receipt_id == ("append_only_receipt:a0f17f23345df62697e317b927484ef1") assert first.transaction_receipt.request_hash == ( - "sha256:ce69f1018e1ea95d8b262626bbfe15580563cb999cb0b0b51b05f6960aa0d35b" + "sha256:42f2dcdc73c34a9b09b4ab371445b0c73f6bdb4e8314dec03e5178f39789598b" ) assert env.provider.calls == 1 assert env.runtime.capability_calls == env.runtime.authority_calls == 4 @@ -1318,28 +1319,23 @@ async def test_divergent_synthesis_replay_and_cross_wired_attempt_fail_before_pr @pytest.mark.asyncio -async def test_semantic_as_of_and_context_availability_are_distinct_and_fail_closed(): +async def test_signal_semantic_time_precedes_brief_analysis_cutoff_and_future_evidence_fails_closed(): env = await _environment() - wrong_as_of = BriefSynthesisRequestV1Alpha1.model_validate( + assert SIGNAL_AS_OF < env.request.brief_as_of == env.request.context_cutoff_at + stale_cutoff = BriefSynthesisRequestV1Alpha1.model_validate( { **env.request.model_dump(mode="python", exclude={"request_id", "request_digest"}), "brief_as_of": BRIEF_AS_OF - timedelta(seconds=1), + "context_cutoff_at": BRIEF_AS_OF - timedelta(seconds=1), } ) with pytest.raises(BriefSynthesisError): - await env.service.synthesize(wrong_as_of) - unavailable = BriefSynthesisRequestV1Alpha1.model_validate( - { - **env.request.model_dump(mode="python", exclude={"request_id", "request_digest"}), - "context_cutoff_at": ROUTED_AT - timedelta(seconds=1), - } - ) - with pytest.raises(BriefSynthesisError): - await env.service.synthesize(unavailable) + await env.service.synthesize(stale_cutoff) with pytest.raises(ValidationError): BriefSynthesisRequestV1Alpha1.model_validate( { **env.request.model_dump(mode="python", exclude={"request_id", "request_digest"}), + "brief_as_of": REQUESTED_AT + timedelta(seconds=1), "context_cutoff_at": REQUESTED_AT + timedelta(seconds=1), } ) diff --git a/tests/intelligence/test_case_brief_synthesis.py b/tests/intelligence/test_case_brief_synthesis.py index 203eb37..5892ffb 100644 --- a/tests/intelligence/test_case_brief_synthesis.py +++ b/tests/intelligence/test_case_brief_synthesis.py @@ -61,6 +61,7 @@ REQUESTED_AT, ROUTED_AT, SHIFT_DETECTED_AT, + SIGNAL_AS_OF, SIGNAL_DETECTED_AT, _ActivationAuthority, _ActivationStore, @@ -301,7 +302,7 @@ def _derivation( reference, entity=entity, suffix=f"{key}-current", - as_of=BRIEF_AS_OF, + as_of=SIGNAL_AS_OF, attributes=current, ) baseline_snapshot = _snapshot( @@ -315,7 +316,7 @@ def _derivation( reference, entity=entity, observations=(second,), - as_of=BRIEF_AS_OF, + as_of=SIGNAL_AS_OF, attributes=current, ) shift = detect_numeric_shift( @@ -791,6 +792,7 @@ async def test_stale_context_cutoff_excludes_a_case_member_and_fails_closed(): CaseBriefSynthesisRequestV1Alpha1.model_validate( env.request.model_copy( update={ + "brief_as_of": stale_cutoff, "context_cutoff_at": stale_cutoff, "request_id": None, "request_digest": None,