From 7e9beeecbe9c8bff29d0590b78ec131c4f5a86be Mon Sep 17 00:00:00 2001 From: Edwin Amirian Date: Wed, 12 Aug 2026 17:25:50 -0700 Subject: [PATCH] feat(intelligence): project decision outcome feedback loop --- ace/application/__init__.py | 2 + .../intelligence_resource_projection.py | 254 ++++++++++++ .../core/intelligence_resource_plane.py | 33 +- ...ce-resource-plane-v0.8.0-work-packet-v1.md | 15 +- ...on_outcome_feedback_resource_projection.py | 392 ++++++++++++++++++ 5 files changed, 683 insertions(+), 13 deletions(-) create mode 100644 tests/intelligence/test_decision_outcome_feedback_resource_projection.py diff --git a/ace/application/__init__.py b/ace/application/__init__.py index 3196c77..de32c51 100644 --- a/ace/application/__init__.py +++ b/ace/application/__init__.py @@ -346,6 +346,7 @@ ) from ace.application.intelligence_resource_projection import ( CompositeIntelligenceResourceProjectionReader, + DecisionOutcomeFeedbackResourceProjectionReader, IntelligenceLedgerProjectionError, IntelligenceLedgerResourceProjectionReader, IntelligenceResourceProjectionContributor, @@ -461,6 +462,7 @@ "IntelligenceLedgerResourceProjectionReader", "IntelligenceResourceProjectionContributor", "CompositeIntelligenceResourceProjectionReader", + "DecisionOutcomeFeedbackResourceProjectionReader", "MonitoringResourceProjectionReader", "ASSERTION_DECISION_RECORD_KIND", "EXTRACTION_RECEIPT_RECORD_KIND", diff --git a/ace/application/intelligence_resource_projection.py b/ace/application/intelligence_resource_projection.py index 3631091..6af93e4 100644 --- a/ace/application/intelligence_resource_projection.py +++ b/ace/application/intelligence_resource_projection.py @@ -15,7 +15,9 @@ ) from ace.application.monitoring import LIVE_MONITORING_RECORD_SPACE from ace.core.contracts import canonical_json +from ace.core.decisions import DecisionV1Alpha1, OutcomeV1Alpha1 from ace.core.records import ImmutableRecordStore, ImmutableRecordV1 +from ace.intelligence.contracts.feedback import FeedbackProposalV1Alpha1 from ace.intelligence.contracts.ledger import IntelligenceRecordKind from ace.intelligence.contracts.monitoring import ( MONITORING_LIFECYCLE_RECORD_KIND, @@ -66,6 +68,13 @@ _LEDGER_TO_PUBLIC = {value: key for key, value in _PUBLIC_TO_LEDGER.items()} LEDGER_RESOURCE_KINDS = frozenset(_PUBLIC_TO_LEDGER) MONITORING_RESOURCE_KINDS = frozenset({IntelligenceResourceKind.MONITOR, IntelligenceResourceKind.SUBSCRIPTION}) +DECISION_OUTCOME_FEEDBACK_RESOURCE_KINDS = frozenset( + { + IntelligenceResourceKind.DECISION, + IntelligenceResourceKind.OUTCOME, + IntelligenceResourceKind.FEEDBACK, + } +) _LINEAGE_TO_PUBLIC: dict[LineageResourceKind, IntelligenceResourceKind] = { LineageResourceKind.OBSERVATION: IntelligenceResourceKind.OBSERVATION, LineageResourceKind.ENTITY_SNAPSHOT: IntelligenceResourceKind.ENTITY, @@ -476,6 +485,250 @@ async def read( ) +_PREPARED_DECISION_MODELS: dict[ + IntelligenceResourceKind, + tuple[str, type[DecisionV1Alpha1] | type[OutcomeV1Alpha1] | type[FeedbackProposalV1Alpha1]], +] = { + IntelligenceResourceKind.DECISION: ("decision", DecisionV1Alpha1), + IntelligenceResourceKind.OUTCOME: ("outcome", OutcomeV1Alpha1), + IntelligenceResourceKind.FEEDBACK: ("feedback_proposal", FeedbackProposalV1Alpha1), +} + + +def _immutable_reference( + record, + *, + kind: IntelligenceResourceKind, +) -> IntelligenceResourceReferenceV1Alpha1: + return IntelligenceResourceReferenceV1Alpha1( + product_id=record.product_id, + resource_kind=kind, + resource_id=record.record_key, + resource_digest=record.material_hash, + resource_contract=record.payload_contract, + revision=1, + as_of=record.as_of, + available_at=record.available_at, + ) + + +def _record_reference( + reference, + *, + kind: IntelligenceResourceKind, +) -> IntelligenceResourceReferenceV1Alpha1: + return IntelligenceResourceReferenceV1Alpha1( + product_id=reference.product_id, + resource_kind=kind, + resource_id=reference.record_key, + resource_digest=reference.material_hash, + resource_contract=reference.payload_contract, + revision=1, + as_of=reference.as_of, + available_at=reference.available_at, + ) + + +def _prepared_decision_projection( + record: ImmutableRecordV1, + *, + kind: IntelligenceResourceKind, + value: DecisionV1Alpha1 | OutcomeV1Alpha1 | FeedbackProposalV1Alpha1, + decision_subject: IntelligenceResourceReferenceV1Alpha1 | None = None, +) -> IntelligenceResourceRecordV1Alpha1: + payload = CanonicalJsonValueV1Alpha1(value_json=canonical_json(value.model_dump(mode="json"))) + availability = IntelligenceResourceAvailability.AVAILABLE + degraded_reason_refs: tuple[str, ...] = () + if isinstance(value, DecisionV1Alpha1): + subject = value.intent.subject + if decision_subject is not None: + provenance = (decision_subject,) + elif subject.record_kind == "brief": + provenance = () + availability = IntelligenceResourceAvailability.DEGRADED + degraded_reason_refs = ("degraded_reason:unresolved-decision-subject",) + else: + provenance = () + availability = IntelligenceResourceAvailability.DEGRADED + degraded_reason_refs = ("degraded_reason:unsupported-decision-subject",) + title = f"Decision: {value.intent.decision_type}" + summary = ( + f"Disposition is {value.intent.disposition.value}; " + f"action disposition is {value.intent.action_disposition.value}." + ) + subject_refs = ( + value.intent.authenticated_context.actor_ref, + value.intent.actor_role_ref, + subject.record_key, + ) + elif isinstance(value, OutcomeV1Alpha1): + provenance = (_record_reference(value.intent.decision, kind=IntelligenceResourceKind.DECISION),) + title = f"Outcome: {value.intent.outcome_type}" + summary = f"Observed measure {value.intent.measure_id}." + subject_refs = ( + value.intent.authenticated_context.actor_ref, + value.intent.decision.record_key, + value.intent.measure_id, + ) + else: + provenance = ( + _record_reference(value.intent.decision, kind=IntelligenceResourceKind.DECISION), + _record_reference(value.intent.outcome, kind=IntelligenceResourceKind.OUTCOME), + ) + title = f"Feedback proposal: {value.intent.policy_id}" + summary = f"Proposes policy value {value.intent.prior_value} → {value.intent.proposed_value}." + subject_refs = ( + value.intent.decision.record_key, + value.intent.outcome.record_key, + value.intent.policy_id, + ) + return IntelligenceResourceRecordV1Alpha1( + reference=_immutable_reference(record, kind=kind), + availability=availability, + title=title, + summary=summary, + subject_refs=tuple(sorted(set(subject_refs))), + provenance=provenance, + payload=payload, + degraded_reason_refs=degraded_reason_refs, + ) + + +def _prepared_decision_envelope_is_exact( + record: ImmutableRecordV1, + *, + kind: IntelligenceResourceKind, + value: DecisionV1Alpha1 | OutcomeV1Alpha1 | FeedbackProposalV1Alpha1, +) -> bool: + if ( + record.product_id != value.intent.product_id + or record.record_space != "prepared" + or record.payload_contract != value.contract + or record.available_at != value.authorization.authorized_at + ): + return False + if isinstance(value, DecisionV1Alpha1): + return ( + kind is IntelligenceResourceKind.DECISION + and record.record_kind == "decision" + and record.record_key == value.decision_id + and record.as_of == value.intent.decided_at + ) + if isinstance(value, OutcomeV1Alpha1): + return ( + kind is IntelligenceResourceKind.OUTCOME + and record.record_kind == "outcome" + and record.record_key == value.outcome_id + and record.as_of == value.intent.observed_at + ) + return ( + kind is IntelligenceResourceKind.FEEDBACK + and record.record_kind == "feedback_proposal" + and record.record_key == value.proposal_id + and record.as_of == value.intent.outcome.as_of + ) + + +async def _decision_subject_reference( + store: ImmutableRecordStore, + value: DecisionV1Alpha1, +) -> IntelligenceResourceReferenceV1Alpha1 | None: + subject = value.intent.subject + if subject.record_kind != "brief" or subject.record_space not in { + IntelligenceResourceMode.PREPARED.value, + IntelligenceResourceMode.LIVE.value, + }: + return None + try: + stored = await store.load_record( + subject.storage_id, + product_id=subject.product_id, + record_space=subject.record_space, + record_kind=subject.record_kind, + ) + if stored is None or stored.reference() != subject: + return None + mode = IntelligenceResourceMode(subject.record_space) + return _project_record( + stored, + mode=mode, + ledger_kind=IntelligenceRecordKind.BRIEF, + ).reference + except Exception: + return None + + +class DecisionOutcomeFeedbackResourceProjectionReader(IntelligenceResourceProjectionReader): + """Project the immutable Decision → Outcome → governed-feedback proposal loop.""" + + def __init__(self, *, store: ImmutableRecordStore, degrade_unsupported: bool = True) -> None: + self.store = store + self.degrade_unsupported = degrade_unsupported + + @property + def supported_kinds(self) -> frozenset[IntelligenceResourceKind]: + return DECISION_OUTCOME_FEEDBACK_RESOURCE_KINDS + + async def read( + self, + *, + query: IntelligenceResourceQueryV1Alpha1, + after: IntelligenceResourceCursorV1Alpha1 | None, + limit: int, + ) -> IntelligenceResourceProjectionBatch: + requested = set(query.resource_kinds) + relevant = requested & DECISION_OUTCOME_FEEDBACK_RESOURCE_KINDS + degraded = { + f"degraded_reason:unsupported-{kind.value}" + for kind in requested - DECISION_OUTCOME_FEEDBACK_RESOURCE_KINDS + if self.degrade_unsupported + } + projected: list[IntelligenceResourceRecordV1Alpha1] = [] + for kind in sorted(relevant, key=lambda item: item.value): + record_kind, model = _PREPARED_DECISION_MODELS[kind] + try: + records = await self.store.read_as_of( + product_id=query.product_id, + record_space="prepared", + record_kind=record_kind, + available_at=query.available_at, + ) + except Exception: + degraded.add(f"degraded_reason:read-prepared-{record_kind}") + continue + for record in records: + try: + value = model.model_validate(record.payload) + if not _prepared_decision_envelope_is_exact(record, kind=kind, value=value): + raise ValueError("prepared decision-loop envelope mismatch") + if record.as_of > query.as_of: + continue + decision_subject = ( + await _decision_subject_reference(self.store, value) + if isinstance(value, DecisionV1Alpha1) + else None + ) + item = _prepared_decision_projection( + record, + kind=kind, + value=value, + decision_subject=decision_subject, + ) + if query.subject_refs and set(query.subject_refs).isdisjoint(item.subject_refs): + continue + degraded.update(item.degraded_reason_refs) + projected.append(item) + except Exception: + degraded.add(f"degraded_reason:invalid-prepared-{record_kind}") + visible = _after_cursor(projected, after)[:limit] + reasons = tuple(sorted(degraded)) + return IntelligenceResourceProjectionBatch( + records=tuple(visible), + state=(IntelligenceResourcePageState.DEGRADED if reasons else IntelligenceResourcePageState.COMPLETE), + degraded_reason_refs=reasons, + ) + + class CompositeIntelligenceResourceProjectionReader(IntelligenceResourceProjectionReader): """Merge disjoint rebuildable projection contributors into one stable page.""" @@ -529,6 +782,7 @@ async def read( __all__ = [ "CompositeIntelligenceResourceProjectionReader", + "DecisionOutcomeFeedbackResourceProjectionReader", "IntelligenceLedgerProjectionError", "IntelligenceLedgerResourceProjectionReader", "IntelligenceResourceProjectionContributor", diff --git a/core/engine/core/intelligence_resource_plane.py b/core/engine/core/intelligence_resource_plane.py index b141f3a..74d992d 100644 --- a/core/engine/core/intelligence_resource_plane.py +++ b/core/engine/core/intelligence_resource_plane.py @@ -15,6 +15,7 @@ from ace.application import ( RESOURCE_QUERY_AUTHORITY, CompositeIntelligenceResourceProjectionReader, + DecisionOutcomeFeedbackResourceProjectionReader, IntelligenceLedgerResourceProjectionReader, IntelligenceResourceCursorV1Alpha1, IntelligenceResourceKind, @@ -22,6 +23,7 @@ IntelligenceResourcePlaneAuthorizationPort, IntelligenceResourcePlaneError, IntelligenceResourcePlaneService, + IntelligenceResourceProjectionReader, IntelligenceResourceQueryV1Alpha1, MonitoringResourceProjectionReader, ) @@ -81,6 +83,25 @@ def intelligence_resource_runtime() -> IntelligenceResourceHttpRuntime: ) +def intelligence_resource_projection_reader(records: ImmutableRecordStore) -> IntelligenceResourceProjectionReader: + """Compose all disjoint rebuildable public projection contributors.""" + + return CompositeIntelligenceResourceProjectionReader( + IntelligenceLedgerResourceProjectionReader( + store=records, + degrade_unsupported=False, + ), + MonitoringResourceProjectionReader( + store=records, + degrade_unsupported=False, + ), + DecisionOutcomeFeedbackResourceProjectionReader( + store=records, + degrade_unsupported=False, + ), + ) + + def _verified_claims(user: dict) -> tuple[str, str]: actor_ref = user.get("sub") product_id = user.get("product") @@ -121,16 +142,7 @@ async def query_intelligence_resource_page( cursor=selector.cursor, ) return await IntelligenceResourcePlaneService( - reader=CompositeIntelligenceResourceProjectionReader( - IntelligenceLedgerResourceProjectionReader( - store=runtime.records, - degrade_unsupported=False, - ), - MonitoringResourceProjectionReader( - store=runtime.records, - degrade_unsupported=False, - ), - ), + reader=intelligence_resource_projection_reader(runtime.records), authority=runtime.authority, ).query(request, evaluated_at=evaluated_at) except GovernedCompositionAuthorityError as exc: @@ -149,6 +161,7 @@ async def query_intelligence_resource_page( "IntelligenceResourceHttpUnauthenticated", "IntelligenceResourceHttpUnavailable", "IntelligenceResourcePageV1Alpha1", + "intelligence_resource_projection_reader", "intelligence_resource_runtime", "query_intelligence_resource_page", ] diff --git a/docs/design/intelligence-resource-plane-v0.8.0-work-packet-v1.md b/docs/design/intelligence-resource-plane-v0.8.0-work-packet-v1.md index 5cb8011..5d1e756 100644 --- a/docs/design/intelligence-resource-plane-v0.8.0-work-packet-v1.md +++ b/docs/design/intelligence-resource-plane-v0.8.0-work-packet-v1.md @@ -1,9 +1,10 @@ # ACE 0.8.0 unified Intelligence resource plane work packet -Status: **active 0.8C packet; facade, ledger/monitoring projections, and governed HTTP query implemented** +Status: **active 0.8C packet; facade, intelligence/monitoring projections, governed HTTP query, +and Decision → Outcome → Feedback closure implemented** Public milestone: [issue #40](https://github.com/augmented-cognition-engine/core/issues/40) -Accepted base: `main@794183b` (0.8A architecture, AM4 lifecycle, completed 0.8B, facade, -ledger projection, and governed HTTP query) +Accepted base: `main@cf53360` (0.8A architecture, AM4 lifecycle, completed 0.8B, facade, +ledger/monitoring projection, and governed HTTP query) ## Outcome @@ -88,6 +89,14 @@ declares incomplete or divergent chains as degraded, and composes with ledger re disjoint family ownership. The host—not FastAPI—binds this composite reader to the same governed query service. +The next contributor projects immutable Decisions, Outcomes, and governed Feedback proposals from +the existing PREPARED loop. Public provenance resolves a Decision's exact Brief to the Brief's +content identity rather than substituting Core's storage-envelope hash; Outcomes point to their +exact Decision, and Feedback points to both exact Decision and Outcome. Unknown Decision subject +types remain visible only as explicitly degraded truth. The supported host composes all current +contributors through one named factory so future resource families cannot silently bypass the +same API path. + 0.8C must still add governed-state projections for the remaining canonical families and complete packaged schema/import integrity. 0.8D must prove Atrium consumes this interface rather than privileged internal state. diff --git a/tests/intelligence/test_decision_outcome_feedback_resource_projection.py b/tests/intelligence/test_decision_outcome_feedback_resource_projection.py new file mode 100644 index 0000000..b9325af --- /dev/null +++ b/tests/intelligence/test_decision_outcome_feedback_resource_projection.py @@ -0,0 +1,392 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from ace.application import DecisionOutcomeFeedbackResourceProjectionReader +from ace.core import ( + AuthenticatedRuntimeContextV1Alpha1, + DecisionActionDisposition, + DecisionDisposition, + DecisionIntentV1Alpha1, + DecisionV1Alpha1, + GovernedActionAuthorizationProjection, + GovernedStateHeadPreconditionV1Alpha1, + ImmutableRecordV1, + OutcomeIntentV1Alpha1, + OutcomeV1Alpha1, + ReceiptReferenceV1Alpha1, +) +from ace.core.contracts import canonical_hash +from ace.intelligence import ( + ActivationRevisionReferenceV1Alpha1, + BriefV1Alpha1, + CitationV1Alpha1, + CompiledPackRefV1, + EvidenceAcquisitionMode, + FeedbackProposalIntentV1Alpha1, + FeedbackProposalV1Alpha1, + GroundedClaimV1Alpha1, + IntelligenceResourceKind, + IntelligenceResourceMode, + IntelligenceResourcePageState, + IntelligenceResourceQueryV1Alpha1, +) +from ace.testing import InMemoryImmutableRecordStore +from core.engine.core.intelligence_resource_plane import intelligence_resource_projection_reader + +pytestmark = pytest.mark.unit + +PRODUCT = "product:decision-loop-projection" +NOW = datetime(2026, 8, 13, 1, 0, tzinfo=UTC) + + +def _context() -> AuthenticatedRuntimeContextV1Alpha1: + return AuthenticatedRuntimeContextV1Alpha1( + product_id=PRODUCT, + actor_ref="principal:executive", + authentication_receipt_ref="authentication:decision-loop", + authentication_receipt_digest="sha256:" + "a" * 64, + authenticated_at=NOW, + expires_at=NOW + timedelta(hours=2), + ) + + +def _authorization(minutes: int) -> GovernedActionAuthorizationProjection: + return GovernedActionAuthorizationProjection( + authorization_ref=ReceiptReferenceV1Alpha1( + receipt_id=f"authorization:decision-loop:{minutes}", + receipt_digest="sha256:" + f"{minutes % 10}" * 64, + ), + authorized_at=NOW + timedelta(minutes=minutes), + state_preconditions=( + GovernedStateHeadPreconditionV1Alpha1( + state_kind="capability_state", + product_id=PRODUCT, + state_id="capability_state:decision-loop", + sequence=1, + revision_id="revision:capability:1", + commit_receipt_id="commit:capability:1", + ), + GovernedStateHeadPreconditionV1Alpha1( + state_kind="authority_grant", + product_id=PRODUCT, + state_id="authority_grant:decision-loop", + sequence=1, + revision_id="revision:authority:1", + commit_receipt_id="commit:authority:1", + ), + ), + ) + + +def _brief() -> ImmutableRecordV1: + activation_key = "ai_command_center" + activation = ActivationRevisionReferenceV1Alpha1( + product_id=PRODUCT, + activation_key=activation_key, + activation_id=f"domain_activation:{canonical_hash([PRODUCT, activation_key])[:32]}", + revision=1, + revision_id="activation_revision:" + "b" * 32, + revision_digest="sha256:" + "b" * 64, + ) + citation = CitationV1Alpha1( + source_ref="source:public-ai-economics", + source_digest="sha256:" + "e" * 64, + acquisition_mode=EvidenceAcquisitionMode.PREPARED_FIXTURE, + acquisition_receipt_ref="source_acquisition:ai-economics", + acquisition_receipt_digest="sha256:" + "f" * 64, + source_as_of=NOW, + retrieved_at=NOW, + locator="section:token-economics", + excerpt="The observed economics changed.", + ) + brief = BriefV1Alpha1( + product_id=PRODUCT, + mode=IntelligenceResourceMode.PREPARED, + activation_revision=activation, + as_of=NOW, + brief_type_ref="brief_type:executive-intelligence", + title="AI economics changed", + executive_summary="A material AI economics change merits executive review.", + body_markdown="## What changed\n\nAI economics changed.", + generated_at=NOW, + citations=(citation,), + claims=( + GroundedClaimV1Alpha1( + statement="AI economics changed.", + citation_ids=(str(citation.citation_id),), + confidence=0.9, + ), + ), + ) + return ImmutableRecordV1( + product_id=PRODUCT, + record_space="prepared", + record_kind="brief", + record_key=str(brief.resource_id), + payload_contract=brief.contract, + payload=brief.model_dump(mode="python"), + as_of=brief.as_of, + available_at=brief.generated_at, + processing_order=0, + ) + + +def _decision() -> tuple[DecisionV1Alpha1, ImmutableRecordV1]: + value = DecisionV1Alpha1( + intent=DecisionIntentV1Alpha1( + product_id=PRODUCT, + authenticated_context=_context(), + subject=_brief().reference(), + actor_role_ref="persona:executive", + decision_type="investment_review", + disposition=DecisionDisposition.ACCEPT, + action_disposition=DecisionActionDisposition.NO_ACTION, + rationale="Use the evidence in the next investment review.", + decided_at=NOW + timedelta(minutes=1), + ), + authorization=_authorization(2), + ) + record = ImmutableRecordV1( + product_id=PRODUCT, + record_space="prepared", + record_kind="decision", + record_key=str(value.decision_id), + payload_contract=value.contract, + payload=value.model_dump(mode="python"), + as_of=value.intent.decided_at, + available_at=value.authorization.authorized_at, + processing_order=0, + ) + return value, record + + +def _outcome() -> tuple[OutcomeV1Alpha1, ImmutableRecordV1]: + _, decision_record = _decision() + value = OutcomeV1Alpha1( + intent=OutcomeIntentV1Alpha1( + product_id=PRODUCT, + authenticated_context=_context(), + decision=decision_record.reference(), + outcome_type="decision_usefulness", + measure_id="executive_usefulness", + value_json='"useful"', + observed_at=NOW + timedelta(minutes=3), + recorded_at=NOW + timedelta(minutes=4), + ), + authorization=_authorization(5), + ) + record = ImmutableRecordV1( + product_id=PRODUCT, + record_space="prepared", + record_kind="outcome", + record_key=str(value.outcome_id), + payload_contract=value.contract, + payload=value.model_dump(mode="python"), + as_of=value.intent.observed_at, + available_at=value.authorization.authorized_at, + processing_order=0, + ) + return value, record + + +def _feedback() -> tuple[FeedbackProposalV1Alpha1, ImmutableRecordV1]: + _, decision_record = _decision() + _, outcome_record = _outcome() + activation_key = "ai_command_center" + activation = ActivationRevisionReferenceV1Alpha1( + product_id=PRODUCT, + activation_key=activation_key, + activation_id=f"domain_activation:{canonical_hash([PRODUCT, activation_key])[:32]}", + revision=1, + revision_id="activation_revision:" + "b" * 32, + revision_digest="sha256:" + "b" * 64, + ) + value = FeedbackProposalV1Alpha1( + intent=FeedbackProposalIntentV1Alpha1( + product_id=PRODUCT, + activation_revision=activation, + pack=CompiledPackRefV1( + pack_id="ai_command_center", + pack_version="0.1.0", + compiled_pack_id="pack_ir:" + "c" * 32, + pack_digest="sha256:" + "c" * 64, + ), + policy_id="executive_usefulness", + policy_digest="sha256:" + "d" * 64, + decision=decision_record.reference(), + outcome=outcome_record.reference(), + prior_value=0.5, + outcome_value_json='"useful"', + adjustment=0.05, + proposed_value=0.55, + proposed_at=NOW + timedelta(minutes=6), + ), + authorization=_authorization(7), + ) + record = ImmutableRecordV1( + product_id=PRODUCT, + record_space="prepared", + record_kind="feedback_proposal", + record_key=str(value.proposal_id), + payload_contract=value.contract, + payload=value.model_dump(mode="python"), + as_of=outcome_record.as_of, + available_at=value.authorization.authorized_at, + processing_order=0, + ) + return value, record + + +def _store(*records: ImmutableRecordV1) -> InMemoryImmutableRecordStore: + store = InMemoryImmutableRecordStore() + store.records.update({str(record.storage_id): record for record in records}) + return store + + +def _query(*kinds: IntelligenceResourceKind, subject_refs: tuple[str, ...] = ()) -> IntelligenceResourceQueryV1Alpha1: + return IntelligenceResourceQueryV1Alpha1( + authenticated_context=_context(), + product_id=PRODUCT, + authority_grant_ref="authority_grant:decision-loop-read", + resource_kinds=kinds, + subject_refs=subject_refs, + as_of=NOW + timedelta(minutes=10), + available_at=NOW + timedelta(minutes=10), + page_size=20, + ) + + +@pytest.mark.asyncio +async def test_projects_exact_decision_outcome_feedback_chain_with_public_provenance() -> None: + _, decision_record = _decision() + _, outcome_record = _outcome() + _, feedback_record = _feedback() + batch = await DecisionOutcomeFeedbackResourceProjectionReader( + store=_store(_brief(), decision_record, outcome_record, feedback_record) + ).read( + query=_query( + IntelligenceResourceKind.DECISION, + IntelligenceResourceKind.OUTCOME, + IntelligenceResourceKind.FEEDBACK, + ), + after=None, + limit=20, + ) + + assert batch.state is IntelligenceResourcePageState.COMPLETE + assert [item.reference.resource_kind for item in batch.records] == [ + IntelligenceResourceKind.DECISION, + IntelligenceResourceKind.OUTCOME, + IntelligenceResourceKind.FEEDBACK, + ] + decision, outcome, feedback = batch.records + assert decision.provenance[0].resource_kind is IntelligenceResourceKind.BRIEF + assert decision.provenance[0].resource_digest == _brief().payload["resource_digest"] + assert decision.provenance[0].resource_digest != _brief().material_hash + assert outcome.provenance[0].resource_id == decision.reference.resource_id + assert [item.resource_kind for item in feedback.provenance] == [ + IntelligenceResourceKind.DECISION, + IntelligenceResourceKind.OUTCOME, + ] + assert feedback.payload is not None + assert feedback.payload.parsed_value()["intent"]["proposed_value"] == 0.55 + + +def test_supported_host_composition_includes_the_complete_current_resource_plane() -> None: + reader = intelligence_resource_projection_reader(InMemoryImmutableRecordStore()) + assert { + IntelligenceResourceKind.OBSERVATION, + IntelligenceResourceKind.MONITOR, + IntelligenceResourceKind.SUBSCRIPTION, + IntelligenceResourceKind.DECISION, + IntelligenceResourceKind.OUTCOME, + IntelligenceResourceKind.FEEDBACK, + } <= reader.supported_kinds + + +@pytest.mark.asyncio +async def test_subject_filter_and_historical_cutoff_preserve_exact_visibility() -> None: + _, decision_record = _decision() + _, outcome_record = _outcome() + reader = DecisionOutcomeFeedbackResourceProjectionReader(store=_store(decision_record, outcome_record)) + filtered = await reader.read( + query=_query(IntelligenceResourceKind.OUTCOME, subject_refs=("principal:executive",)), + after=None, + limit=20, + ) + assert len(filtered.records) == 1 + + query = _query(IntelligenceResourceKind.OUTCOME) + historical = query.model_copy(update={"as_of": NOW + timedelta(minutes=2)}) + hidden = await reader.read(query=historical, after=None, limit=20) + assert hidden.records == () + + +@pytest.mark.asyncio +async def test_invalid_envelope_degrades_without_exposing_payload() -> None: + value, record = _decision() + invalid = ImmutableRecordV1( + **record.model_dump(mode="python", exclude={"storage_id", "material_hash", "record_key"}), + record_key="decision:wrong-envelope", + ) + batch = await DecisionOutcomeFeedbackResourceProjectionReader(store=_store(invalid)).read( + query=_query(IntelligenceResourceKind.DECISION), + after=None, + limit=20, + ) + assert value.decision_id != invalid.record_key + assert batch.records == () + assert batch.state is IntelligenceResourcePageState.DEGRADED + assert batch.degraded_reason_refs == ("degraded_reason:invalid-prepared-decision",) + + +@pytest.mark.asyncio +async def test_unknown_decision_subject_is_visible_but_never_mislabeled_as_lineage() -> None: + subject = ImmutableRecordV1( + product_id=PRODUCT, + record_space="prepared", + record_kind="opaque_orientation", + record_key="orientation:one", + payload_contract="example.orientation/v1", + payload={"orientation": "opaque"}, + as_of=NOW, + available_at=NOW, + processing_order=0, + ) + value = DecisionV1Alpha1( + intent=DecisionIntentV1Alpha1( + product_id=PRODUCT, + authenticated_context=_context(), + subject=subject.reference(), + actor_role_ref="persona:executive", + decision_type="orientation_review", + disposition=DecisionDisposition.ACCEPT, + action_disposition=DecisionActionDisposition.NO_ACTION, + rationale="Retain the decision without inventing a public subject type.", + decided_at=NOW + timedelta(minutes=1), + ), + authorization=_authorization(2), + ) + record = ImmutableRecordV1( + product_id=PRODUCT, + record_space="prepared", + record_kind="decision", + record_key=str(value.decision_id), + payload_contract=value.contract, + payload=value.model_dump(mode="python"), + as_of=value.intent.decided_at, + available_at=value.authorization.authorized_at, + processing_order=0, + ) + batch = await DecisionOutcomeFeedbackResourceProjectionReader(store=_store(record)).read( + query=_query(IntelligenceResourceKind.DECISION), + after=None, + limit=20, + ) + assert len(batch.records) == 1 + assert batch.records[0].provenance == () + assert batch.records[0].availability.value == "degraded" + assert batch.degraded_reason_refs == ("degraded_reason:unsupported-decision-subject",)