From 3a0157154afa8fd690fd6bf25d596881c928989b Mon Sep 17 00:00:00 2001 From: Edwin Amirian Date: Wed, 12 Aug 2026 17:12:13 -0700 Subject: [PATCH] feat(intelligence): project monitors and subscriptions --- ace/application/__init__.py | 6 + .../intelligence_resource_projection.py | 247 ++++++++++++++++- .../core/intelligence_resource_plane.py | 13 +- ...ce-resource-plane-v0.8.0-work-packet-v1.md | 16 +- .../test_monitoring_resource_projection.py | 256 ++++++++++++++++++ 5 files changed, 531 insertions(+), 7 deletions(-) create mode 100644 tests/intelligence/test_monitoring_resource_projection.py diff --git a/ace/application/__init__.py b/ace/application/__init__.py index 8bac5d9..3196c77 100644 --- a/ace/application/__init__.py +++ b/ace/application/__init__.py @@ -345,8 +345,11 @@ IntelligenceResourceQueryV1Alpha1, ) from ace.application.intelligence_resource_projection import ( + CompositeIntelligenceResourceProjectionReader, IntelligenceLedgerProjectionError, IntelligenceLedgerResourceProjectionReader, + IntelligenceResourceProjectionContributor, + MonitoringResourceProjectionReader, ) from ace.application.live_intelligence_bridge import ( LiveBriefAdmission, @@ -456,6 +459,9 @@ "IntelligenceResourceQueryV1Alpha1", "IntelligenceLedgerProjectionError", "IntelligenceLedgerResourceProjectionReader", + "IntelligenceResourceProjectionContributor", + "CompositeIntelligenceResourceProjectionReader", + "MonitoringResourceProjectionReader", "ASSERTION_DECISION_RECORD_KIND", "EXTRACTION_RECEIPT_RECORD_KIND", "GRAPH_PROJECTION_RECORD_KIND", diff --git a/ace/application/intelligence_resource_projection.py b/ace/application/intelligence_resource_projection.py index 260054f..3631091 100644 --- a/ace/application/intelligence_resource_projection.py +++ b/ace/application/intelligence_resource_projection.py @@ -2,9 +2,10 @@ from __future__ import annotations +from collections import defaultdict from collections.abc import Iterable from datetime import datetime -from typing import Any +from typing import Any, Protocol from pydantic import BaseModel, TypeAdapter @@ -12,9 +13,16 @@ IntelligenceResourceProjectionBatch, IntelligenceResourceProjectionReader, ) +from ace.application.monitoring import LIVE_MONITORING_RECORD_SPACE from ace.core.contracts import canonical_json from ace.core.records import ImmutableRecordStore, ImmutableRecordV1 from ace.intelligence.contracts.ledger import IntelligenceRecordKind +from ace.intelligence.contracts.monitoring import ( + MONITORING_LIFECYCLE_RECORD_KIND, + MonitoringLifecycleReceiptV1Alpha1, + MonitoringLifecycleState, + MonitoringTargetKind, +) from ace.intelligence.contracts.resource_plane import ( IntelligenceResourceAvailability, IntelligenceResourceCursorV1Alpha1, @@ -56,6 +64,8 @@ IntelligenceResourceKind.BRIEF: IntelligenceRecordKind.BRIEF, } _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}) _LINEAGE_TO_PUBLIC: dict[LineageResourceKind, IntelligenceResourceKind] = { LineageResourceKind.OBSERVATION: IntelligenceResourceKind.OBSERVATION, LineageResourceKind.ENTITY_SNAPSHOT: IntelligenceResourceKind.ENTITY, @@ -78,6 +88,21 @@ class IntelligenceLedgerProjectionError(RuntimeError): """An immutable ledger record could not be projected exactly.""" +class IntelligenceResourceProjectionContributor(Protocol): + """A disjoint owner of one or more public resource projection families.""" + + @property + def supported_kinds(self) -> frozenset[IntelligenceResourceKind]: ... + + async def read( + self, + *, + query: IntelligenceResourceQueryV1Alpha1, + after: IntelligenceResourceCursorV1Alpha1 | None, + limit: int, + ) -> IntelligenceResourceProjectionBatch: ... + + def _resource_times(resource: BaseModel) -> tuple[datetime, datetime]: as_of = resource.as_of if isinstance(resource, ObservationV1Alpha1): @@ -239,11 +264,17 @@ def __init__( IntelligenceResourceMode.PREPARED, IntelligenceResourceMode.LIVE, ), + degrade_unsupported: bool = True, ) -> None: if not modes or len(modes) != len(set(modes)): raise ValueError("projection modes must be a non-empty unique sequence") self.store = store self.modes = modes + self.degrade_unsupported = degrade_unsupported + + @property + def supported_kinds(self) -> frozenset[IntelligenceResourceKind]: + return LEDGER_RESOURCE_KINDS async def read( self, @@ -257,7 +288,8 @@ async def read( for public_kind in query.resource_kinds: ledger_kind = _PUBLIC_TO_LEDGER.get(public_kind) if ledger_kind is None: - degraded.add(f"degraded_reason:unsupported-{public_kind.value}") + if self.degrade_unsupported: + degraded.add(f"degraded_reason:unsupported-{public_kind.value}") continue for mode in self.modes: try: @@ -287,7 +319,218 @@ async def read( ) +def _monitoring_kind(receipt: MonitoringLifecycleReceiptV1Alpha1) -> IntelligenceResourceKind: + if receipt.target_kind is MonitoringTargetKind.MONITOR: + return IntelligenceResourceKind.MONITOR + return IntelligenceResourceKind.SUBSCRIPTION + + +def _monitoring_reference( + receipt: MonitoringLifecycleReceiptV1Alpha1, +) -> IntelligenceResourceReferenceV1Alpha1: + return IntelligenceResourceReferenceV1Alpha1( + product_id=receipt.product_id, + resource_kind=_monitoring_kind(receipt), + resource_id=receipt.lifecycle.reference, + resource_digest=str(receipt.receipt_digest), + resource_contract=receipt.contract, + revision=receipt.sequence, + as_of=receipt.applied_at, + available_at=receipt.applied_at, + ) + + +def _monitoring_projection( + receipt: MonitoringLifecycleReceiptV1Alpha1, + *, + previous: MonitoringLifecycleReceiptV1Alpha1 | None, +) -> IntelligenceResourceRecordV1Alpha1: + tombstoned = receipt.state_after is MonitoringLifecycleState.REVOKED + label = "Monitor" if receipt.target_kind is MonitoringTargetKind.MONITOR else "Subscription" + return IntelligenceResourceRecordV1Alpha1( + reference=_monitoring_reference(receipt), + availability=( + IntelligenceResourceAvailability.TOMBSTONED if tombstoned else IntelligenceResourceAvailability.AVAILABLE + ), + title=f"{label}: {receipt.target.reference}", + summary=f"{label} lifecycle is {receipt.state_after.value}.", + subject_refs=tuple( + sorted( + { + receipt.owner_ref, + receipt.persona_binding.reference, + receipt.target.reference, + } + ) + ), + supersedes=_monitoring_reference(previous) if previous is not None else None, + payload=( + None + if tombstoned + else CanonicalJsonValueV1Alpha1(value_json=canonical_json(receipt.model_dump(mode="json"))) + ), + ) + + +class MonitoringResourceProjectionReader(IntelligenceResourceProjectionReader): + """Project current Monitor and Subscription lifecycle revisions.""" + + 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 MONITORING_RESOURCE_KINDS + + async def read( + self, + *, + query: IntelligenceResourceQueryV1Alpha1, + after: IntelligenceResourceCursorV1Alpha1 | None, + limit: int, + ) -> IntelligenceResourceProjectionBatch: + requested = set(query.resource_kinds) + relevant = requested & MONITORING_RESOURCE_KINDS + degraded = { + f"degraded_reason:unsupported-{kind.value}" + for kind in requested - MONITORING_RESOURCE_KINDS + if self.degrade_unsupported + } + if not relevant: + return IntelligenceResourceProjectionBatch( + records=(), + state=(IntelligenceResourcePageState.DEGRADED if degraded else IntelligenceResourcePageState.COMPLETE), + degraded_reason_refs=tuple(sorted(degraded)), + ) + try: + records = await self.store.read_as_of( + product_id=query.product_id, + record_space=LIVE_MONITORING_RECORD_SPACE, + record_kind=MONITORING_LIFECYCLE_RECORD_KIND, + available_at=query.available_at, + ) + except Exception: + return IntelligenceResourceProjectionBatch( + records=(), + state=IntelligenceResourcePageState.DEGRADED, + degraded_reason_refs=("degraded_reason:read-monitoring-lifecycle",), + ) + + chains: dict[str, list[MonitoringLifecycleReceiptV1Alpha1]] = defaultdict(list) + for record in records: + try: + receipt = MonitoringLifecycleReceiptV1Alpha1.model_validate(record.payload) + if ( + record.product_id != query.product_id + or record.record_space != LIVE_MONITORING_RECORD_SPACE + or record.record_kind != MONITORING_LIFECYCLE_RECORD_KIND + or record.record_key != receipt.receipt_id + or record.payload_contract != receipt.contract + or record.as_of != receipt.applied_at + or record.available_at != receipt.applied_at + ): + raise ValueError("monitoring envelope mismatch") + if _monitoring_kind(receipt) not in relevant: + continue + if receipt.applied_at <= query.as_of: + chains[receipt.lifecycle.reference].append(receipt) + except Exception: + degraded.add("degraded_reason:invalid-monitoring-lifecycle") + + projected: list[IntelligenceResourceRecordV1Alpha1] = [] + for lifecycle_id, chain in chains.items(): + ordered = sorted(chain, key=lambda item: item.sequence) + if [item.sequence for item in ordered] != list(range(1, len(ordered) + 1)): + degraded.add(f"degraded_reason:incomplete-{lifecycle_id}") + continue + if any( + current.prior_receipt != previous.reference() or current.state_before is not previous.state_after + for previous, current in zip(ordered, ordered[1:]) + ): + degraded.add(f"degraded_reason:divergent-{lifecycle_id}") + continue + current = ordered[-1] + public_kind = _monitoring_kind(current) + if public_kind not in relevant: + continue + item = _monitoring_projection( + current, + previous=ordered[-2] if len(ordered) > 1 else None, + ) + if query.subject_refs and set(query.subject_refs).isdisjoint(item.subject_refs): + continue + projected.append(item) + + 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.""" + + def __init__(self, *contributors: IntelligenceResourceProjectionContributor) -> None: + if not contributors: + raise ValueError("at least one resource projection contributor is required") + supported: set[IntelligenceResourceKind] = set() + for contributor in contributors: + overlap = supported & set(contributor.supported_kinds) + if overlap: + raise ValueError(f"resource projection contributors overlap: {sorted(item.value for item in overlap)}") + supported.update(contributor.supported_kinds) + self.contributors = contributors + self.supported_kinds = frozenset(supported) + + async def read( + self, + *, + query: IntelligenceResourceQueryV1Alpha1, + after: IntelligenceResourceCursorV1Alpha1 | None, + limit: int, + ) -> IntelligenceResourceProjectionBatch: + records: list[IntelligenceResourceRecordV1Alpha1] = [] + degraded = { + f"degraded_reason:unsupported-{kind.value}" for kind in set(query.resource_kinds) - self.supported_kinds + } + for contributor in self.contributors: + if not (set(query.resource_kinds) & contributor.supported_kinds): + continue + batch = await contributor.read(query=query, after=after, limit=limit) + records.extend(batch.records) + degraded.update(batch.degraded_reason_refs) + visible = _after_cursor(records, after)[:limit] + keys = [ + ( + item.reference.resource_kind, + item.reference.resource_id, + item.reference.revision, + ) + for item in visible + ] + if len(keys) != len(set(keys)): + raise IntelligenceLedgerProjectionError("resource projection contributors returned duplicate revisions") + reasons = tuple(sorted(degraded)) + return IntelligenceResourceProjectionBatch( + records=tuple(visible), + state=(IntelligenceResourcePageState.DEGRADED if reasons else IntelligenceResourcePageState.COMPLETE), + degraded_reason_refs=reasons, + ) + + __all__ = [ + "CompositeIntelligenceResourceProjectionReader", "IntelligenceLedgerProjectionError", "IntelligenceLedgerResourceProjectionReader", + "IntelligenceResourceProjectionContributor", + "MonitoringResourceProjectionReader", ] diff --git a/core/engine/core/intelligence_resource_plane.py b/core/engine/core/intelligence_resource_plane.py index 9da4524..b141f3a 100644 --- a/core/engine/core/intelligence_resource_plane.py +++ b/core/engine/core/intelligence_resource_plane.py @@ -14,6 +14,7 @@ from ace.application import ( RESOURCE_QUERY_AUTHORITY, + CompositeIntelligenceResourceProjectionReader, IntelligenceLedgerResourceProjectionReader, IntelligenceResourceCursorV1Alpha1, IntelligenceResourceKind, @@ -22,6 +23,7 @@ IntelligenceResourcePlaneError, IntelligenceResourcePlaneService, IntelligenceResourceQueryV1Alpha1, + MonitoringResourceProjectionReader, ) from ace.core import ImmutableRecordPersistenceError, ImmutableRecordStore from core.engine.core.agent_composition_runtime import ( @@ -119,7 +121,16 @@ async def query_intelligence_resource_page( cursor=selector.cursor, ) return await IntelligenceResourcePlaneService( - reader=IntelligenceLedgerResourceProjectionReader(store=runtime.records), + reader=CompositeIntelligenceResourceProjectionReader( + IntelligenceLedgerResourceProjectionReader( + store=runtime.records, + degrade_unsupported=False, + ), + MonitoringResourceProjectionReader( + store=runtime.records, + degrade_unsupported=False, + ), + ), authority=runtime.authority, ).query(request, evaluated_at=evaluated_at) except GovernedCompositionAuthorityError as exc: 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 3fbd297..5cb8011 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,8 +1,9 @@ # ACE 0.8.0 unified Intelligence resource plane work packet -Status: **active 0.8C packet; C1 facade, C2 ledger projection, and C3 governed HTTP query implemented** +Status: **active 0.8C packet; facade, ledger/monitoring projections, and governed HTTP query implemented** Public milestone: [issue #40](https://github.com/augmented-cognition-engine/core/issues/40) -Accepted base: `main@6b4d6b2` (0.8A architecture, AM4 lifecycle, completed 0.8B, C1 facade, and C2 projection) +Accepted base: `main@794183b` (0.8A architecture, AM4 lifecycle, completed 0.8B, facade, +ledger projection, and governed HTTP query) ## Outcome @@ -80,8 +81,15 @@ public page contract. Historical data cutoffs are independent from login time; e reauthenticated and reauthorized, while query identity remains stable across authentication receipt refreshes for the same actor and exact selector. -C3 must still add governed-state projections for the remaining canonical families and complete -packaged schema/import integrity. C4 must prove Atrium consumes this interface rather than +The next additive projection contributor exposes the current Monitor and Subscription lifecycle +revision from the existing append-only monitoring ledger. It validates the complete contiguous +receipt chain, projects revoke as a payload-free tombstone, preserves immediate supersession, +declares incomplete or divergent chains as degraded, and composes with ledger resources through +disjoint family ownership. The host—not FastAPI—binds this composite reader to the same governed +query service. + +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. The 0.8C exit gate is one authorized query path that can reproduce the evidence-to-outcome resource diff --git a/tests/intelligence/test_monitoring_resource_projection.py b/tests/intelligence/test_monitoring_resource_projection.py new file mode 100644 index 0000000..d1ba115 --- /dev/null +++ b/tests/intelligence/test_monitoring_resource_projection.py @@ -0,0 +1,256 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from ace.application import ( + CompositeIntelligenceResourceProjectionReader, + IntelligenceLedgerResourceProjectionReader, + MonitoringResourceProjectionReader, +) +from ace.application.monitoring import LIVE_MONITORING_RECORD_SPACE +from ace.core import ImmutableRecordV1 +from ace.core.runtime_use import AuthenticatedRuntimeContextV1Alpha1 +from ace.intelligence import ( + ExactMaterialReferenceV1Alpha1, + IntelligenceResourceAvailability, + IntelligenceResourceKind, + IntelligenceResourcePageState, + IntelligenceResourceQueryV1Alpha1, + MonitoringLifecycleAction, + MonitoringLifecycleReceiptV1Alpha1, + MonitoringLifecycleState, + MonitoringTargetKind, + monitoring_lifecycle_identity, +) +from ace.intelligence.contracts.monitoring import MONITORING_LIFECYCLE_RECORD_KIND +from ace.testing import InMemoryImmutableRecordStore + +pytestmark = pytest.mark.unit + +PRODUCT = "product:monitoring-projection" +NOW = datetime(2026, 8, 12, 20, 0, tzinfo=UTC) +TARGET = ExactMaterialReferenceV1Alpha1( + reference="monitor:ai-releases", + digest="sha256:" + "a" * 64, +) +PERSONA = ExactMaterialReferenceV1Alpha1( + reference="persona_binding:executive", + digest="sha256:" + "b" * 64, +) + + +def _receipt( + *, + sequence: int, + state_before: MonitoringLifecycleState | None, + state_after: MonitoringLifecycleState, + prior: MonitoringLifecycleReceiptV1Alpha1 | None = None, + target_kind: MonitoringTargetKind = MonitoringTargetKind.MONITOR, +) -> MonitoringLifecycleReceiptV1Alpha1: + lifecycle = monitoring_lifecycle_identity( + product_id=PRODUCT, + target_kind=target_kind, + target=TARGET, + persona_binding=PERSONA, + ) + action = { + 1: MonitoringLifecycleAction.CREATE, + 2: MonitoringLifecycleAction.PAUSE, + 3: MonitoringLifecycleAction.REVOKE, + }[sequence] + return MonitoringLifecycleReceiptV1Alpha1( + product_id=PRODUCT, + owner_ref="principal:executive", + target_kind=target_kind, + target=TARGET, + persona_binding=PERSONA, + lifecycle=lifecycle, + request=ExactMaterialReferenceV1Alpha1( + reference=f"monitoring_request:{sequence}", + digest="sha256:" + f"{sequence}" * 64, + ), + action=action, + sequence=sequence, + state_before=state_before, + state_after=state_after, + prior_receipt=prior.reference() if prior is not None else None, + applied_at=NOW + timedelta(minutes=sequence), + ) + + +def _record(receipt: MonitoringLifecycleReceiptV1Alpha1) -> ImmutableRecordV1: + return ImmutableRecordV1( + product_id=PRODUCT, + record_space=LIVE_MONITORING_RECORD_SPACE, + record_kind=MONITORING_LIFECYCLE_RECORD_KIND, + record_key=str(receipt.receipt_id), + payload_contract=receipt.contract, + payload=receipt.model_dump(mode="python"), + as_of=receipt.applied_at, + available_at=receipt.applied_at, + processing_order=0, + ) + + +def _store(*receipts: MonitoringLifecycleReceiptV1Alpha1) -> InMemoryImmutableRecordStore: + store = InMemoryImmutableRecordStore() + records = [_record(receipt) for receipt in receipts] + 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=AuthenticatedRuntimeContextV1Alpha1( + product_id=PRODUCT, + actor_ref="principal:executive", + authentication_receipt_ref="authentication_receipt:monitoring", + authentication_receipt_digest="sha256:" + "c" * 64, + authenticated_at=NOW, + expires_at=NOW + timedelta(hours=1), + ), + product_id=PRODUCT, + authority_grant_ref="authority_grant:monitoring-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_monitoring_projection_returns_only_the_current_exact_revision() -> None: + created = _receipt( + sequence=1, + state_before=None, + state_after=MonitoringLifecycleState.ACTIVE, + ) + paused = _receipt( + sequence=2, + state_before=MonitoringLifecycleState.ACTIVE, + state_after=MonitoringLifecycleState.PAUSED, + prior=created, + ) + batch = await MonitoringResourceProjectionReader(store=_store(created, paused)).read( + query=_query(IntelligenceResourceKind.MONITOR, subject_refs=("principal:executive",)), + after=None, + limit=20, + ) + + assert batch.state is IntelligenceResourcePageState.COMPLETE + assert len(batch.records) == 1 + current = batch.records[0] + assert current.reference.resource_kind is IntelligenceResourceKind.MONITOR + assert current.reference.resource_id == paused.lifecycle.reference + assert current.reference.revision == 2 + assert current.supersedes is not None + assert current.supersedes.revision == 1 + assert current.payload is not None + assert current.payload.parsed_value()["state_after"] == "paused" + + +@pytest.mark.asyncio +async def test_revoked_monitor_is_tombstoned_without_payload() -> None: + created = _receipt( + sequence=1, + state_before=None, + state_after=MonitoringLifecycleState.ACTIVE, + ) + paused = _receipt( + sequence=2, + state_before=MonitoringLifecycleState.ACTIVE, + state_after=MonitoringLifecycleState.PAUSED, + prior=created, + ) + revoked = _receipt( + sequence=3, + state_before=MonitoringLifecycleState.PAUSED, + state_after=MonitoringLifecycleState.REVOKED, + prior=paused, + ) + batch = await MonitoringResourceProjectionReader(store=_store(created, paused, revoked)).read( + query=_query(IntelligenceResourceKind.MONITOR), + after=None, + limit=20, + ) + assert batch.records[0].availability is IntelligenceResourceAvailability.TOMBSTONED + assert batch.records[0].payload is None + assert batch.records[0].reference.revision == 3 + + +@pytest.mark.asyncio +async def test_subscription_lifecycle_projects_as_a_distinct_resource_family() -> None: + created = _receipt( + sequence=1, + state_before=None, + state_after=MonitoringLifecycleState.ACTIVE, + target_kind=MonitoringTargetKind.SUBSCRIPTION, + ) + batch = await MonitoringResourceProjectionReader(store=_store(created)).read( + query=_query(IntelligenceResourceKind.SUBSCRIPTION), + after=None, + limit=20, + ) + assert len(batch.records) == 1 + assert batch.records[0].reference.resource_kind is IntelligenceResourceKind.SUBSCRIPTION + + +@pytest.mark.asyncio +async def test_incomplete_lifecycle_degrades_instead_of_inventing_current_state() -> None: + created = _receipt( + sequence=1, + state_before=None, + state_after=MonitoringLifecycleState.ACTIVE, + ) + paused = _receipt( + sequence=2, + state_before=MonitoringLifecycleState.ACTIVE, + state_after=MonitoringLifecycleState.PAUSED, + prior=created, + ) + batch = await MonitoringResourceProjectionReader(store=_store(paused)).read( + query=_query(IntelligenceResourceKind.MONITOR), + after=None, + limit=20, + ) + assert batch.records == () + assert batch.state is IntelligenceResourcePageState.DEGRADED + assert batch.degraded_reason_refs[0].startswith("degraded_reason:incomplete-") + + +@pytest.mark.asyncio +async def test_composite_reader_merges_contributors_and_owns_unsupported_degradation() -> None: + created = _receipt( + sequence=1, + state_before=None, + state_after=MonitoringLifecycleState.ACTIVE, + ) + store = _store(created) + reader = CompositeIntelligenceResourceProjectionReader( + IntelligenceLedgerResourceProjectionReader(store=store, degrade_unsupported=False), + MonitoringResourceProjectionReader(store=store, degrade_unsupported=False), + ) + batch = await reader.read( + query=_query(IntelligenceResourceKind.MONITOR, IntelligenceResourceKind.ACTION), + after=None, + limit=20, + ) + assert len(batch.records) == 1 + assert batch.records[0].reference.resource_kind is IntelligenceResourceKind.MONITOR + assert batch.state is IntelligenceResourcePageState.DEGRADED + assert batch.degraded_reason_refs == ("degraded_reason:unsupported-action",) + + +def test_composite_reader_rejects_overlapping_resource_owners() -> None: + store = InMemoryImmutableRecordStore() + with pytest.raises(ValueError, match="overlap"): + CompositeIntelligenceResourceProjectionReader( + MonitoringResourceProjectionReader(store=store), + MonitoringResourceProjectionReader(store=store), + )