From 00fc99d1ccfbc04c99972738aa07adcf69c133d4 Mon Sep 17 00:00:00 2001 From: Edwin Amirian Date: Wed, 12 Aug 2026 16:27:11 -0700 Subject: [PATCH] feat(intelligence): project immutable ledgers --- ace/application/__init__.py | 6 + .../intelligence_resource_plane.py | 2 +- .../intelligence_resource_projection.py | 293 ++++++++++++++++++ ace/intelligence/contracts/resource_plane.py | 2 +- ...ce-resource-plane-v0.8.0-work-packet-v1.md | 20 +- .../test_intelligence_resource_plane.py | 2 + .../test_intelligence_resource_projection.py | 275 ++++++++++++++++ 7 files changed, 591 insertions(+), 9 deletions(-) create mode 100644 ace/application/intelligence_resource_projection.py create mode 100644 tests/intelligence/test_intelligence_resource_projection.py diff --git a/ace/application/__init__.py b/ace/application/__init__.py index 88b35b0..eb43f59 100644 --- a/ace/application/__init__.py +++ b/ace/application/__init__.py @@ -340,6 +340,10 @@ IntelligenceResourceProjectionBatch, IntelligenceResourceProjectionReader, ) +from ace.application.intelligence_resource_projection import ( + IntelligenceLedgerProjectionError, + IntelligenceLedgerResourceProjectionReader, +) from ace.application.live_intelligence_bridge import ( LiveBriefAdmission, LiveBriefSynthesisError, @@ -442,6 +446,8 @@ "IntelligenceResourcePlaneService", "IntelligenceResourceProjectionBatch", "IntelligenceResourceProjectionReader", + "IntelligenceLedgerProjectionError", + "IntelligenceLedgerResourceProjectionReader", "ASSERTION_DECISION_RECORD_KIND", "EXTRACTION_RECEIPT_RECORD_KIND", "GRAPH_PROJECTION_RECORD_KIND", diff --git a/ace/application/intelligence_resource_plane.py b/ace/application/intelligence_resource_plane.py index 76cb8f0..97fc80d 100644 --- a/ace/application/intelligence_resource_plane.py +++ b/ace/application/intelligence_resource_plane.py @@ -16,7 +16,7 @@ ) RESOURCE_QUERY_OPERATION = "query_intelligence_resources" -RESOURCE_QUERY_AUTHORITY = "read_intelligence_resources" +RESOURCE_QUERY_AUTHORITY = "observe_read" class IntelligenceResourcePlaneError(RuntimeError): diff --git a/ace/application/intelligence_resource_projection.py b/ace/application/intelligence_resource_projection.py new file mode 100644 index 0000000..260054f --- /dev/null +++ b/ace/application/intelligence_resource_projection.py @@ -0,0 +1,293 @@ +"""Rebuildable resource-plane projections over the immutable Intelligence ledgers.""" + +from __future__ import annotations + +from collections.abc import Iterable +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, TypeAdapter + +from ace.application.intelligence_resource_plane import ( + IntelligenceResourceProjectionBatch, + IntelligenceResourceProjectionReader, +) +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.resource_plane import ( + IntelligenceResourceAvailability, + IntelligenceResourceCursorV1Alpha1, + IntelligenceResourceKind, + IntelligenceResourcePageState, + IntelligenceResourceQueryV1Alpha1, + IntelligenceResourceRecordV1Alpha1, + IntelligenceResourceReferenceV1Alpha1, +) +from ace.intelligence.contracts.resources import ( + BriefV1Alpha1, + CanonicalJsonValueV1Alpha1, + CaseV1Alpha1, + EntitySnapshotV1Alpha1, + IntelligenceResourceMode, + LineageReferenceV1Alpha1, + LineageResourceKind, + ObservationV1Alpha1, + ShiftV1Alpha1, + SignalV1Alpha1, +) + +_JSON_OBJECT = TypeAdapter(dict[str, Any]) + +_RESOURCE_MODELS: dict[IntelligenceRecordKind, type[BaseModel]] = { + IntelligenceRecordKind.OBSERVATION: ObservationV1Alpha1, + IntelligenceRecordKind.ENTITY_SNAPSHOT: EntitySnapshotV1Alpha1, + IntelligenceRecordKind.SHIFT: ShiftV1Alpha1, + IntelligenceRecordKind.SIGNAL: SignalV1Alpha1, + IntelligenceRecordKind.CASE: CaseV1Alpha1, + IntelligenceRecordKind.BRIEF: BriefV1Alpha1, +} +_PUBLIC_TO_LEDGER: dict[IntelligenceResourceKind, IntelligenceRecordKind] = { + IntelligenceResourceKind.OBSERVATION: IntelligenceRecordKind.OBSERVATION, + IntelligenceResourceKind.ENTITY: IntelligenceRecordKind.ENTITY_SNAPSHOT, + IntelligenceResourceKind.SHIFT: IntelligenceRecordKind.SHIFT, + IntelligenceResourceKind.SIGNAL: IntelligenceRecordKind.SIGNAL, + IntelligenceResourceKind.CASE: IntelligenceRecordKind.CASE, + IntelligenceResourceKind.BRIEF: IntelligenceRecordKind.BRIEF, +} +_LEDGER_TO_PUBLIC = {value: key for key, value in _PUBLIC_TO_LEDGER.items()} +_LINEAGE_TO_PUBLIC: dict[LineageResourceKind, IntelligenceResourceKind] = { + LineageResourceKind.OBSERVATION: IntelligenceResourceKind.OBSERVATION, + LineageResourceKind.ENTITY_SNAPSHOT: IntelligenceResourceKind.ENTITY, + LineageResourceKind.SIGNAL: IntelligenceResourceKind.SIGNAL, + LineageResourceKind.SHIFT: IntelligenceResourceKind.SHIFT, + LineageResourceKind.CASE: IntelligenceResourceKind.CASE, + LineageResourceKind.BRIEF: IntelligenceResourceKind.BRIEF, +} +_LINEAGE_CONTRACTS: dict[LineageResourceKind, str] = { + LineageResourceKind.OBSERVATION: "ace.intelligence.observation/v1alpha1", + LineageResourceKind.ENTITY_SNAPSHOT: "ace.intelligence.entity-snapshot/v1alpha1", + LineageResourceKind.SIGNAL: "ace.intelligence.signal/v1alpha1", + LineageResourceKind.SHIFT: "ace.intelligence.shift/v1alpha1", + LineageResourceKind.CASE: "ace.intelligence.case/v1alpha1", + LineageResourceKind.BRIEF: "ace.intelligence.brief/v1alpha1", +} + + +class IntelligenceLedgerProjectionError(RuntimeError): + """An immutable ledger record could not be projected exactly.""" + + +def _resource_times(resource: BaseModel) -> tuple[datetime, datetime]: + as_of = resource.as_of + if isinstance(resource, ObservationV1Alpha1): + return as_of, resource.ingested_at + if isinstance(resource, EntitySnapshotV1Alpha1): + return as_of, resource.projected_at + if isinstance(resource, (ShiftV1Alpha1, SignalV1Alpha1)): + return as_of, resource.detected_at + if isinstance(resource, CaseV1Alpha1): + return as_of, resource.assembled_at + if isinstance(resource, BriefV1Alpha1): + return as_of, resource.generated_at + raise IntelligenceLedgerProjectionError("unsupported immutable Intelligence resource") + + +def _resource_title(resource: BaseModel) -> str: + if isinstance(resource, ObservationV1Alpha1): + return f"Observation from {resource.source_ref}" + if isinstance(resource, EntitySnapshotV1Alpha1): + return resource.entity_ref + return str(resource.title) + + +def _resource_summary(resource: BaseModel) -> str | None: + if isinstance(resource, (ShiftV1Alpha1, SignalV1Alpha1)): + return resource.summary + if isinstance(resource, CaseV1Alpha1): + return resource.purpose + if isinstance(resource, BriefV1Alpha1): + return resource.executive_summary + return None + + +def _resource_subjects(resource: BaseModel) -> tuple[str, ...]: + if isinstance(resource, EntitySnapshotV1Alpha1): + return (resource.entity_ref,) + if isinstance(resource, (ObservationV1Alpha1, ShiftV1Alpha1, SignalV1Alpha1, CaseV1Alpha1)): + return resource.subject_refs + return () + + +def _lineage_reference( + lineage: LineageReferenceV1Alpha1, + *, + product_id: str, +) -> IntelligenceResourceReferenceV1Alpha1: + if lineage.resource_kind not in _LINEAGE_TO_PUBLIC: + raise IntelligenceLedgerProjectionError( + "lineage kind lacks an exact resource contract in this projection slice" + ) + return IntelligenceResourceReferenceV1Alpha1( + product_id=product_id, + resource_kind=_LINEAGE_TO_PUBLIC[lineage.resource_kind], + resource_id=lineage.resource_id, + resource_digest=lineage.resource_digest, + resource_contract=_LINEAGE_CONTRACTS[lineage.resource_kind], + revision=1, + as_of=lineage.resource_as_of, + available_at=lineage.resource_available_at, + ) + + +def _decode_record( + record: ImmutableRecordV1, + *, + mode: IntelligenceResourceMode, + ledger_kind: IntelligenceRecordKind, +) -> BaseModel: + model = _RESOURCE_MODELS[ledger_kind] + if record.record_space != mode.value or record.record_kind != ledger_kind.value: + raise IntelligenceLedgerProjectionError("immutable record crossed its ledger bucket") + try: + resource = model.model_validate_json(_JSON_OBJECT.dump_json(record.payload)) + except (TypeError, ValueError) as exc: + raise IntelligenceLedgerProjectionError("immutable Intelligence payload failed exact replay") from exc + as_of, available_at = _resource_times(resource) + if ( + resource.product_id != record.product_id + or resource.mode is not mode + or resource.resource_id != record.record_key + or resource.contract != record.payload_contract + or as_of != record.as_of + or available_at != record.available_at + ): + raise IntelligenceLedgerProjectionError("immutable record envelope does not match its Intelligence payload") + return resource + + +def _project_record( + record: ImmutableRecordV1, + *, + mode: IntelligenceResourceMode, + ledger_kind: IntelligenceRecordKind, +) -> IntelligenceResourceRecordV1Alpha1: + resource = _decode_record(record, mode=mode, ledger_kind=ledger_kind) + as_of, available_at = _resource_times(resource) + return IntelligenceResourceRecordV1Alpha1( + reference=IntelligenceResourceReferenceV1Alpha1( + product_id=record.product_id, + resource_kind=_LEDGER_TO_PUBLIC[ledger_kind], + resource_id=str(resource.resource_id), + resource_digest=str(resource.resource_digest), + resource_contract=str(resource.contract), + revision=1, + as_of=as_of, + available_at=available_at, + ), + availability=IntelligenceResourceAvailability.AVAILABLE, + title=_resource_title(resource), + summary=_resource_summary(resource), + subject_refs=_resource_subjects(resource), + provenance=tuple(_lineage_reference(item, product_id=record.product_id) for item in resource.lineage), + payload=CanonicalJsonValueV1Alpha1(value_json=canonical_json(resource.model_dump(mode="json"))), + ) + + +def _after_cursor( + records: Iterable[IntelligenceResourceRecordV1Alpha1], + cursor: IntelligenceResourceCursorV1Alpha1 | None, +) -> list[IntelligenceResourceRecordV1Alpha1]: + ordered = sorted( + records, + key=lambda item: ( + item.reference.available_at, + item.reference.resource_kind.value, + item.reference.resource_id, + item.reference.revision, + ), + ) + if cursor is None: + return ordered + after = ( + cursor.after_available_at, + cursor.after_resource_kind.value, + cursor.after_resource_id, + cursor.after_revision, + ) + return [ + item + for item in ordered + if ( + item.reference.available_at, + item.reference.resource_kind.value, + item.reference.resource_id, + item.reference.revision, + ) + > after + ] + + +class IntelligenceLedgerResourceProjectionReader(IntelligenceResourceProjectionReader): + """Rebuild the supported public resource slice from PREPARED and LIVE records.""" + + def __init__( + self, + *, + store: ImmutableRecordStore, + modes: tuple[IntelligenceResourceMode, ...] = ( + IntelligenceResourceMode.PREPARED, + IntelligenceResourceMode.LIVE, + ), + ) -> 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 + + async def read( + self, + *, + query: IntelligenceResourceQueryV1Alpha1, + after: IntelligenceResourceCursorV1Alpha1 | None, + limit: int, + ) -> IntelligenceResourceProjectionBatch: + projected: list[IntelligenceResourceRecordV1Alpha1] = [] + degraded: set[str] = set() + 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}") + continue + for mode in self.modes: + try: + records = await self.store.read_as_of( + product_id=query.product_id, + record_space=mode.value, + record_kind=ledger_kind.value, + available_at=query.available_at, + ) + projected.extend( + _project_record(record, mode=mode, ledger_kind=ledger_kind) + for record in records + if record.as_of <= query.as_of + ) + except Exception: + degraded.add(f"degraded_reason:read-{mode.value}-{ledger_kind.value}") + + if query.subject_refs: + requested = set(query.subject_refs) + projected = [item for item in projected if not requested.isdisjoint(item.subject_refs)] + 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, + ) + + +__all__ = [ + "IntelligenceLedgerProjectionError", + "IntelligenceLedgerResourceProjectionReader", +] diff --git a/ace/intelligence/contracts/resource_plane.py b/ace/intelligence/contracts/resource_plane.py index 4cb0570..9fc5eae 100644 --- a/ace/intelligence/contracts/resource_plane.py +++ b/ace/intelligence/contracts/resource_plane.py @@ -402,7 +402,7 @@ def validate_page(self) -> Self: or authority.use_subject_ref != self.query_id or authority.use_subject_digest != self.query_digest or authority.operation != "query_intelligence_resources" - or authority.authority != "read_intelligence_resources" + or authority.authority != "observe_read" or authority.evaluated_at != self.evaluated_at ): raise ValueError("page does not preserve the exact resource-query authority evaluation") 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 f597979..5725afa 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,8 @@ # ACE 0.8.0 unified Intelligence resource plane work packet -Status: **active 0.8C packet; C1 public contracts and authorized query service implemented** +Status: **active 0.8C packet; C1 public facade and C2 immutable-ledger projection implemented** Public milestone: [issue #40](https://github.com/augmented-cognition-engine/core/issues/40) -Accepted base: `main@18d0aef` (0.8A architecture, AM4 lifecycle, and completed 0.8B boundary realignment) +Accepted base: `main@bf4a75a` (0.8A architecture, AM4 lifecycle, completed 0.8B, and C1 facade) ## Outcome @@ -43,7 +43,8 @@ The C1 seam provides: Queries and cursors never become bearer authority. Every page preserves the exact authenticated principal, product, query digest, grant, operation, evaluation time, and Core authority-use receipt. -A cursor changes the read position but not the query identity. +A cursor changes the read position but not the query identity. Resource queries use Core's existing +`observe_read` authority class; the resource plane does not invent a parallel grant vocabulary. Readers fail closed if they widen product, resource-kind, subject, temporal, pagination, or result size boundaries. Degraded results require explicit reason references. Tombstoned resources cannot @@ -66,10 +67,15 @@ the requested point-in-time view. ## Remaining 0.8C sequence -C2 must bind the projection-reader port to supported PREPARED/LIVE Intelligence and governed-state -records, preserving exact provenance and restart behavior. C3 must expose the same contracts through -one supported machine interface and verify packaged schema/import integrity. C4 must prove Atrium -consumes that interface rather than privileged internal state. +C2 binds the six existing immutable PREPARED/LIVE resource families—Observations, Entity Snapshots, +Signals, Shifts, Cases, and Briefs—to the same public plane. The adapter merges record spaces, +projects Entity Snapshots as public Entities, preserves exact lineage and payloads, filters by +subject and cursor, and remains reproducible when reconstructed over the same store. Unsupported or +unavailable buckets return explicit degradation while preserving available truth. + +C3 must add governed-state projections for the remaining canonical families, expose the contracts +through one supported machine interface, and verify packaged schema/import integrity. C4 must prove +Atrium consumes that interface rather than privileged internal state. The 0.8C exit gate is one authorized query path that can reproduce the evidence-to-outcome resource chain after restart, report partial truth honestly, and remain identical for World and Market diff --git a/tests/intelligence/test_intelligence_resource_plane.py b/tests/intelligence/test_intelligence_resource_plane.py index 63192eb..63a5705 100644 --- a/tests/intelligence/test_intelligence_resource_plane.py +++ b/tests/intelligence/test_intelligence_resource_plane.py @@ -12,6 +12,7 @@ IntelligenceResourcePlaneService, IntelligenceResourceProjectionBatch, ) +from ace.core.agent_composition import AuthorityClass from ace.core.runtime_use import AuthenticatedRuntimeContextV1Alpha1, AuthorityUseReceiptV1Alpha1 from ace.core.state import GovernedStateHeadPreconditionV1Alpha1 from ace.intelligence import IntelligenceResourceKind as PublicIntelligenceResourceKind @@ -190,6 +191,7 @@ def test_resource_plane_is_exported_through_the_supported_public_packages() -> N assert PublicIntelligenceResourcePlaneService is IntelligenceResourcePlaneService assert PublicIntelligenceResourceKind is IntelligenceResourceKind assert IntelligenceResourceQueryV1Alpha1.model_json_schema()["type"] == "object" + assert RESOURCE_QUERY_AUTHORITY == AuthorityClass.OBSERVE_READ.value def test_query_identity_excludes_cursor_but_cursor_is_bound_to_the_exact_query() -> None: diff --git a/tests/intelligence/test_intelligence_resource_projection.py b/tests/intelligence/test_intelligence_resource_projection.py new file mode 100644 index 0000000..421d9ec --- /dev/null +++ b/tests/intelligence/test_intelligence_resource_projection.py @@ -0,0 +1,275 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from ace.application import IntelligenceLedgerResourceProjectionReader +from ace.core import ImmutableRecordV1, canonical_hash +from ace.core.runtime_use import AuthenticatedRuntimeContextV1Alpha1 +from ace.intelligence import ( + ActivationRevisionReferenceV1Alpha1, + CanonicalJsonValueV1Alpha1, + EntitySnapshotV1Alpha1, + IntelligenceRecordKind, + IntelligenceResourceCursorV1Alpha1, + IntelligenceResourceKind, + IntelligenceResourceMode, + IntelligenceResourcePageState, + IntelligenceResourceQueryV1Alpha1, + LineageReferenceV1Alpha1, + LineageRelation, + LineageResourceKind, + SignalV1Alpha1, +) +from ace.testing import InMemoryImmutableRecordStore + +pytestmark = pytest.mark.unit + +PRODUCT = "product:resource-projection" +NOW = datetime(2026, 8, 12, 20, 0, tzinfo=UTC) + + +def _activation() -> ActivationRevisionReferenceV1Alpha1: + digest = "sha256:" + "a" * 64 + return ActivationRevisionReferenceV1Alpha1( + product_id=PRODUCT, + activation_key="generic_intelligence", + activation_id=f"domain_activation:{canonical_hash([PRODUCT, 'generic_intelligence'])[:32]}", + revision=1, + revision_id="activation_revision:" + "a" * 32, + revision_digest=digest, + ) + + +def _entity(*, mode: IntelligenceResourceMode) -> EntitySnapshotV1Alpha1: + return EntitySnapshotV1Alpha1( + product_id=PRODUCT, + mode=mode, + activation_revision=_activation(), + as_of=NOW, + entity_ref="entity:ace", + entity_type_ref="entity_type:system", + attributes=CanonicalJsonValueV1Alpha1(value_json='{"name":"ACE"}'), + projected_at=NOW, + confidence=0.95, + ) + + +def _signal( + entity: EntitySnapshotV1Alpha1, + *, + mode: IntelligenceResourceMode, +) -> SignalV1Alpha1: + return SignalV1Alpha1( + product_id=PRODUCT, + mode=mode, + activation_revision=_activation(), + as_of=NOW + timedelta(minutes=1), + lineage=( + LineageReferenceV1Alpha1( + resource_kind=LineageResourceKind.ENTITY_SNAPSHOT, + relation=LineageRelation.SUPPORTS, + resource_id=str(entity.resource_id), + resource_digest=str(entity.resource_digest), + resource_as_of=entity.as_of, + resource_available_at=entity.projected_at, + ), + ), + signal_type_ref="signal_type:material-change", + title="ACE changed", + summary="A material change was detected.", + subject_refs=("entity:ace",), + details=CanonicalJsonValueV1Alpha1(value_json='{"change":"material"}'), + detected_at=NOW + timedelta(minutes=1), + confidence=0.9, + ) + + +def _record(resource, *, kind: IntelligenceRecordKind) -> ImmutableRecordV1: + if isinstance(resource, EntitySnapshotV1Alpha1): + available_at = resource.projected_at + else: + available_at = resource.detected_at + return ImmutableRecordV1( + product_id=PRODUCT, + record_space=resource.mode.value, + record_kind=kind.value, + record_key=str(resource.resource_id), + payload_contract=resource.contract, + payload=resource.model_dump(mode="python"), + as_of=resource.as_of, + available_at=available_at, + processing_order=0, + ) + + +def _store(*records: ImmutableRecordV1) -> InMemoryImmutableRecordStore: + store = InMemoryImmutableRecordStore() + store.records.update({str(item.storage_id): item for item in records}) + return store + + +def _query( + *, + kinds: tuple[IntelligenceResourceKind, ...], + subject_refs: tuple[str, ...] = (), + cursor: IntelligenceResourceCursorV1Alpha1 | None = None, +) -> IntelligenceResourceQueryV1Alpha1: + context = AuthenticatedRuntimeContextV1Alpha1( + product_id=PRODUCT, + actor_ref="principal:analyst", + authentication_receipt_ref="authentication_receipt:projection", + authentication_receipt_digest="sha256:" + "b" * 64, + authenticated_at=NOW - timedelta(minutes=5), + expires_at=NOW + timedelta(minutes=30), + ) + return IntelligenceResourceQueryV1Alpha1( + authenticated_context=context, + product_id=PRODUCT, + authority_grant_ref="authority_grant:resource-read", + resource_kinds=kinds, + subject_refs=subject_refs, + as_of=NOW + timedelta(minutes=1), + available_at=NOW + timedelta(minutes=1), + page_size=10, + cursor=cursor, + ) + + +@pytest.mark.asyncio +async def test_immutable_ledgers_project_through_one_public_resource_plane() -> None: + prepared_entity = _entity(mode=IntelligenceResourceMode.PREPARED) + live_entity = _entity(mode=IntelligenceResourceMode.LIVE) + live_signal = _signal(live_entity, mode=IntelligenceResourceMode.LIVE) + store = _store( + _record(prepared_entity, kind=IntelligenceRecordKind.ENTITY_SNAPSHOT), + _record(live_entity, kind=IntelligenceRecordKind.ENTITY_SNAPSHOT), + _record(live_signal, kind=IntelligenceRecordKind.SIGNAL), + ) + query = _query(kinds=(IntelligenceResourceKind.ENTITY, IntelligenceResourceKind.SIGNAL)) + + batch = await IntelligenceLedgerResourceProjectionReader(store=store).read( + query=query, + after=None, + limit=20, + ) + + assert batch.state is IntelligenceResourcePageState.COMPLETE + assert [item.reference.resource_kind for item in batch.records] == [ + IntelligenceResourceKind.ENTITY, + IntelligenceResourceKind.ENTITY, + IntelligenceResourceKind.SIGNAL, + ] + projected_signal = batch.records[-1] + assert projected_signal.subject_refs == ("entity:ace",) + assert projected_signal.provenance[0].resource_kind is IntelligenceResourceKind.ENTITY + assert projected_signal.provenance[0].resource_id == live_entity.resource_id + assert projected_signal.payload is not None + assert projected_signal.payload.parsed_value()["mode"] == "live" + + +@pytest.mark.asyncio +async def test_projection_honors_subject_cursor_limit_and_restart() -> None: + entity = _entity(mode=IntelligenceResourceMode.LIVE) + signal = _signal(entity, mode=IntelligenceResourceMode.LIVE) + store = _store( + _record(entity, kind=IntelligenceRecordKind.ENTITY_SNAPSHOT), + _record(signal, kind=IntelligenceRecordKind.SIGNAL), + ) + first_query = _query( + kinds=(IntelligenceResourceKind.ENTITY, IntelligenceResourceKind.SIGNAL), + subject_refs=("entity:ace",), + ) + first_reader = IntelligenceLedgerResourceProjectionReader(store=store) + first = await first_reader.read(query=first_query, after=None, limit=1) + assert len(first.records) == 1 + + reference = first.records[0].reference + cursor = IntelligenceResourceCursorV1Alpha1( + query_id=str(first_query.query_id), + after_available_at=reference.available_at, + after_resource_kind=reference.resource_kind, + after_resource_id=reference.resource_id, + after_revision=reference.revision, + ) + restarted_reader = IntelligenceLedgerResourceProjectionReader(store=store) + second = await restarted_reader.read(query=first_query, after=cursor, limit=10) + + assert len(second.records) == 1 + assert second.records[0].reference.resource_kind is IntelligenceResourceKind.SIGNAL + + +@pytest.mark.asyncio +async def test_unsupported_resource_family_is_honestly_degraded() -> None: + query = _query(kinds=(IntelligenceResourceKind.ACTION,)) + batch = await IntelligenceLedgerResourceProjectionReader(store=_store()).read( + query=query, + after=None, + limit=10, + ) + assert batch.state is IntelligenceResourcePageState.DEGRADED + assert batch.records == () + assert batch.degraded_reason_refs == ("degraded_reason:unsupported-action",) + + +@pytest.mark.asyncio +async def test_unavailable_ledger_bucket_does_not_hide_available_truth() -> None: + entity = _entity(mode=IntelligenceResourceMode.PREPARED) + + class _PartiallyUnavailableStore(InMemoryImmutableRecordStore): + async def read_as_of(self, **kwargs): + if kwargs["record_space"] == IntelligenceResourceMode.LIVE.value: + raise RuntimeError("live ledger is unavailable") + return await super().read_as_of(**kwargs) + + store = _PartiallyUnavailableStore() + record = _record(entity, kind=IntelligenceRecordKind.ENTITY_SNAPSHOT) + store.records[str(record.storage_id)] = record + + batch = await IntelligenceLedgerResourceProjectionReader(store=store).read( + query=_query(kinds=(IntelligenceResourceKind.ENTITY,)), + after=None, + limit=10, + ) + assert len(batch.records) == 1 + assert batch.state is IntelligenceResourcePageState.DEGRADED + assert batch.degraded_reason_refs == ("degraded_reason:read-live-entity_snapshot",) + + +@pytest.mark.asyncio +async def test_projection_does_not_invent_contracts_for_external_lineage() -> None: + signal = SignalV1Alpha1( + product_id=PRODUCT, + mode=IntelligenceResourceMode.LIVE, + activation_revision=_activation(), + as_of=NOW + timedelta(minutes=1), + lineage=( + LineageReferenceV1Alpha1( + resource_kind=LineageResourceKind.EVIDENCE, + relation=LineageRelation.SUPPORTS, + resource_id="evidence:external", + resource_digest="sha256:" + "c" * 64, + resource_as_of=NOW, + resource_available_at=NOW, + ), + ), + signal_type_ref="signal_type:material-change", + title="Externally grounded change", + summary="The evidence contract is not available in this ledger slice.", + subject_refs=("entity:ace",), + details=CanonicalJsonValueV1Alpha1(value_json='{"change":"material"}'), + detected_at=NOW + timedelta(minutes=1), + confidence=0.9, + ) + batch = await IntelligenceLedgerResourceProjectionReader( + store=_store(_record(signal, kind=IntelligenceRecordKind.SIGNAL)) + ).read( + query=_query(kinds=(IntelligenceResourceKind.SIGNAL,)), + after=None, + limit=10, + ) + + assert batch.records == () + assert batch.state is IntelligenceResourcePageState.DEGRADED + assert batch.degraded_reason_refs == ("degraded_reason:read-live-signal",)