feat(intelligence): project monitors and subscriptions

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",

ace/application/intelligence_resource_projection.py

@@ -2,19 +2,27 @@
 
 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
 
 from ace.application.intelligence_resource_plane import (
     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",
 ]

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:

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