diff --git a/ace/application/__init__.py b/ace/application/__init__.py index 14f07c7..88b35b0 100644 --- a/ace/application/__init__.py +++ b/ace/application/__init__.py @@ -331,6 +331,15 @@ PreparedIntelligenceLedgerService, PreparedResourceSetAdmission, ) +from ace.application.intelligence_resource_plane import ( + RESOURCE_QUERY_AUTHORITY, + RESOURCE_QUERY_OPERATION, + IntelligenceResourcePlaneAuthorizationPort, + IntelligenceResourcePlaneError, + IntelligenceResourcePlaneService, + IntelligenceResourceProjectionBatch, + IntelligenceResourceProjectionReader, +) from ace.application.live_intelligence_bridge import ( LiveBriefAdmission, LiveBriefSynthesisError, @@ -426,6 +435,13 @@ ) __all__ = [ + "RESOURCE_QUERY_AUTHORITY", + "RESOURCE_QUERY_OPERATION", + "IntelligenceResourcePlaneAuthorizationPort", + "IntelligenceResourcePlaneError", + "IntelligenceResourcePlaneService", + "IntelligenceResourceProjectionBatch", + "IntelligenceResourceProjectionReader", "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 new file mode 100644 index 0000000..76cb8f0 --- /dev/null +++ b/ace/application/intelligence_resource_plane.py @@ -0,0 +1,222 @@ +"""Authorized application service for the unified Intelligence resource plane.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Protocol + +from ace.core.runtime_use import AuthorityUseReceiptV1Alpha1 +from ace.intelligence.contracts.resource_plane import ( + IntelligenceResourceCursorV1Alpha1, + IntelligenceResourcePageState, + IntelligenceResourcePageV1Alpha1, + IntelligenceResourceQueryV1Alpha1, + IntelligenceResourceRecordV1Alpha1, +) + +RESOURCE_QUERY_OPERATION = "query_intelligence_resources" +RESOURCE_QUERY_AUTHORITY = "read_intelligence_resources" + + +class IntelligenceResourcePlaneError(RuntimeError): + """A resource query failed closed before exposing a projection.""" + + +class IntelligenceResourcePlaneAuthorizationPort(Protocol): + async def resolve_authority_use( + self, + *, + context, + use_subject_ref: str, + use_subject_digest: str, + operation: str, + authority: str, + grant_ref: str, + evaluated_at: datetime, + ) -> AuthorityUseReceiptV1Alpha1: ... + + +@dataclass(frozen=True, slots=True) +class IntelligenceResourceProjectionBatch: + """One rebuildable adapter result; never authoritative state.""" + + records: tuple[IntelligenceResourceRecordV1Alpha1, ...] + state: IntelligenceResourcePageState = IntelligenceResourcePageState.COMPLETE + degraded_reason_refs: tuple[str, ...] = () + + +class IntelligenceResourceProjectionReader(Protocol): + async def read( + self, + *, + query: IntelligenceResourceQueryV1Alpha1, + after: IntelligenceResourceCursorV1Alpha1 | None, + limit: int, + ) -> IntelligenceResourceProjectionBatch: ... + + +def _aware(value: datetime, *, name: str) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise IntelligenceResourcePlaneError(f"{name} must include a timezone") + return value.astimezone(UTC) + + +def _sort_key(record: IntelligenceResourceRecordV1Alpha1) -> tuple[datetime, str, str, int]: + reference = record.reference + return ( + reference.available_at, + reference.resource_kind.value, + reference.resource_id, + reference.revision, + ) + + +def _revalidate_query(value: IntelligenceResourceQueryV1Alpha1) -> IntelligenceResourceQueryV1Alpha1: + try: + return IntelligenceResourceQueryV1Alpha1.model_validate(value.model_dump(mode="python")) + except (AttributeError, TypeError, ValueError) as exc: + raise IntelligenceResourcePlaneError("resource query failed exact revalidation") from exc + + +def _revalidate_batch(value: IntelligenceResourceProjectionBatch) -> IntelligenceResourceProjectionBatch: + if not isinstance(value, IntelligenceResourceProjectionBatch): + raise IntelligenceResourcePlaneError("resource projection reader returned an unsupported batch") + try: + records = tuple( + IntelligenceResourceRecordV1Alpha1.model_validate(record.model_dump(mode="python")) + for record in value.records + ) + state = IntelligenceResourcePageState(value.state) + reasons = tuple(value.degraded_reason_refs) + except (AttributeError, TypeError, ValueError) as exc: + raise IntelligenceResourcePlaneError("resource projection batch failed exact revalidation") from exc + if state is IntelligenceResourcePageState.COMPLETE and reasons: + raise IntelligenceResourcePlaneError("complete projection batch cannot declare degraded reasons") + if state is IntelligenceResourcePageState.DEGRADED and not reasons: + raise IntelligenceResourcePlaneError("degraded projection batch requires explicit reason references") + return IntelligenceResourceProjectionBatch(records=records, state=state, degraded_reason_refs=reasons) + + +class IntelligenceResourcePlaneService: + """Query one authorized resource view without owning persistence or authority.""" + + def __init__( + self, + *, + reader: IntelligenceResourceProjectionReader, + authority: IntelligenceResourcePlaneAuthorizationPort, + ) -> None: + self.reader = reader + self.authority = authority + + async def query( + self, + request: IntelligenceResourceQueryV1Alpha1, + *, + evaluated_at: datetime, + ) -> IntelligenceResourcePageV1Alpha1: + exact = _revalidate_query(request) + evaluated = _aware(evaluated_at, name="evaluated_at") + context = exact.authenticated_context + if not (exact.available_at <= evaluated < context.expires_at): + raise IntelligenceResourcePlaneError("resource query evaluation fell outside its temporal window") + + authority_use = await self.authority.resolve_authority_use( + context=context, + use_subject_ref=str(exact.query_id), + use_subject_digest=str(exact.query_digest), + operation=RESOURCE_QUERY_OPERATION, + authority=RESOURCE_QUERY_AUTHORITY, + grant_ref=exact.authority_grant_ref, + evaluated_at=evaluated, + ) + try: + resolved = AuthorityUseReceiptV1Alpha1.model_validate(authority_use.model_dump(mode="python")) + except (AttributeError, TypeError, ValueError) as exc: + raise IntelligenceResourcePlaneError("resource query authority receipt failed revalidation") from exc + if ( + resolved.product_id != exact.product_id + or resolved.actor_ref != context.actor_ref + or resolved.authenticated_context != context + or resolved.use_subject_ref != exact.query_id + or resolved.use_subject_digest != exact.query_digest + or resolved.operation != RESOURCE_QUERY_OPERATION + or resolved.authority != RESOURCE_QUERY_AUTHORITY + or resolved.grant_ref != exact.authority_grant_ref + or resolved.evaluated_at != evaluated + ): + raise IntelligenceResourcePlaneError("authority resolver did not preserve the exact resource query") + + batch = _revalidate_batch( + await self.reader.read( + query=exact, + after=exact.cursor, + limit=exact.page_size + 1, + ) + ) + if len(batch.records) > exact.page_size + 1: + raise IntelligenceResourcePlaneError("resource reader exceeded the bounded page request") + ordering = [_sort_key(item) for item in batch.records] + if ordering != sorted(ordering) or len(ordering) != len(set(ordering)): + raise IntelligenceResourcePlaneError("resource reader returned unstable or duplicate ordering") + + requested_kinds = set(exact.resource_kinds) + requested_subjects = set(exact.subject_refs) + for item in batch.records: + reference = item.reference + if reference.product_id != exact.product_id: + raise IntelligenceResourcePlaneError("resource reader crossed product scope") + if reference.resource_kind not in requested_kinds: + raise IntelligenceResourcePlaneError("resource reader returned an unrequested resource kind") + if reference.as_of > exact.as_of or reference.available_at > exact.available_at: + raise IntelligenceResourcePlaneError("resource reader crossed the query temporal cutoff") + if requested_subjects and requested_subjects.isdisjoint(item.subject_refs): + raise IntelligenceResourcePlaneError("resource reader returned an item outside the subject filter") + if exact.cursor is not None: + cursor_key = ( + exact.cursor.after_available_at, + exact.cursor.after_resource_kind.value, + exact.cursor.after_resource_id, + exact.cursor.after_revision, + ) + if _sort_key(item) <= cursor_key: + raise IntelligenceResourcePlaneError("resource reader did not advance beyond the cursor") + + visible = batch.records[: exact.page_size] + next_cursor = None + if len(batch.records) > exact.page_size: + last = visible[-1].reference + next_cursor = IntelligenceResourceCursorV1Alpha1( + query_id=str(exact.query_id), + after_available_at=last.available_at, + after_resource_kind=last.resource_kind, + after_resource_id=last.resource_id, + after_revision=last.revision, + ) + + return IntelligenceResourcePageV1Alpha1( + query_id=str(exact.query_id), + query_digest=str(exact.query_digest), + product_id=exact.product_id, + actor_ref=context.actor_ref, + as_of=exact.as_of, + available_at=exact.available_at, + evaluated_at=evaluated, + state=batch.state, + items=visible, + next_cursor=next_cursor, + degraded_reason_refs=batch.degraded_reason_refs, + authority_use=resolved, + ) + + +__all__ = [ + "RESOURCE_QUERY_AUTHORITY", + "RESOURCE_QUERY_OPERATION", + "IntelligenceResourcePlaneAuthorizationPort", + "IntelligenceResourcePlaneError", + "IntelligenceResourcePlaneService", + "IntelligenceResourceProjectionBatch", + "IntelligenceResourceProjectionReader", +] diff --git a/ace/intelligence/contracts/__init__.py b/ace/intelligence/contracts/__init__.py index 8463cec..2fada0a 100644 --- a/ace/intelligence/contracts/__init__.py +++ b/ace/intelligence/contracts/__init__.py @@ -363,6 +363,22 @@ PersonasModuleV1, SignalRoutingRuleV1, ) +from ace.intelligence.contracts.resource_plane import ( + MAX_RESOURCE_PLANE_PAGE_SIZE, + RESOURCE_PLANE_CURSOR_VERSION, + RESOURCE_PLANE_PAGE_VERSION, + RESOURCE_PLANE_QUERY_VERSION, + RESOURCE_PLANE_RECORD_VERSION, + RESOURCE_PLANE_REFERENCE_VERSION, + IntelligenceResourceAvailability, + IntelligenceResourceCursorV1Alpha1, + IntelligenceResourceKind, + IntelligenceResourcePageState, + IntelligenceResourcePageV1Alpha1, + IntelligenceResourceQueryV1Alpha1, + IntelligenceResourceRecordV1Alpha1, + IntelligenceResourceReferenceV1Alpha1, +) from ace.intelligence.contracts.resources import ( ActivationRevisionReferenceV1Alpha1, BriefV1Alpha1, @@ -451,6 +467,20 @@ ) __all__ = [ + "MAX_RESOURCE_PLANE_PAGE_SIZE", + "RESOURCE_PLANE_CURSOR_VERSION", + "RESOURCE_PLANE_PAGE_VERSION", + "RESOURCE_PLANE_QUERY_VERSION", + "RESOURCE_PLANE_RECORD_VERSION", + "RESOURCE_PLANE_REFERENCE_VERSION", + "IntelligenceResourceAvailability", + "IntelligenceResourceCursorV1Alpha1", + "IntelligenceResourceKind", + "IntelligenceResourcePageState", + "IntelligenceResourcePageV1Alpha1", + "IntelligenceResourceQueryV1Alpha1", + "IntelligenceResourceRecordV1Alpha1", + "IntelligenceResourceReferenceV1Alpha1", "ActivatedMemoryConstraintsV1Alpha1", "AssertionFamilyV1Alpha1", "AssertionLifecycle", diff --git a/ace/intelligence/contracts/resource_plane.py b/ace/intelligence/contracts/resource_plane.py new file mode 100644 index 0000000..4cb0570 --- /dev/null +++ b/ace/intelligence/contracts/resource_plane.py @@ -0,0 +1,432 @@ +"""Unified, domain-neutral query contracts for the ACE Intelligence resource plane. + +The resource plane is a read model over authoritative Core state and rebuildable +Intelligence projections. It does not grant authority, acquire sources, execute +effects, or become another persistence engine. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from enum import StrEnum +from typing import Literal, Self + +from pydantic import ConfigDict, Field, StrictInt, field_validator, model_validator + +from ace.core.contracts import FrozenContract, canonical_hash +from ace.core.runtime_use import AuthenticatedRuntimeContextV1Alpha1, AuthorityUseReceiptV1Alpha1 +from ace.intelligence.contracts.common import ( + validate_contract, + validate_digest, + validate_product_id, + validate_reference, +) +from ace.intelligence.contracts.resources import CanonicalJsonValueV1Alpha1 + +RESOURCE_PLANE_QUERY_VERSION = "ace.intelligence.resource-plane-query/v1alpha1" +RESOURCE_PLANE_CURSOR_VERSION = "ace.intelligence.resource-plane-cursor/v1alpha1" +RESOURCE_PLANE_REFERENCE_VERSION = "ace.intelligence.resource-plane-reference/v1alpha1" +RESOURCE_PLANE_RECORD_VERSION = "ace.intelligence.resource-plane-record/v1alpha1" +RESOURCE_PLANE_PAGE_VERSION = "ace.intelligence.resource-plane-page/v1alpha1" + +MAX_RESOURCE_PLANE_KINDS = 32 +MAX_RESOURCE_PLANE_SUBJECTS = 256 +MAX_RESOURCE_PLANE_PROVENANCE = 256 +MAX_RESOURCE_PLANE_PAGE_SIZE = 200 + + +class _StrictFrozenContract(FrozenContract): + model_config = ConfigDict( + extra="forbid", + frozen=True, + strict=True, + revalidate_instances="always", + validate_default=True, + allow_inf_nan=False, + ) + + +def _aware(value: datetime, *, name: str) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"{name} must include a timezone") + return value.astimezone(UTC) + + +def _unique_references(values: tuple[str, ...], *, name: str) -> tuple[str, ...]: + normalized = tuple(sorted(validate_reference(value, name=name) for value in values)) + if len(normalized) != len(set(normalized)): + raise ValueError(f"{name} must be unique") + return normalized + + +def _derive_identity(instance: _StrictFrozenContract, *, prefix: str, id_field: str, digest_field: str) -> None: + material = instance.model_dump(mode="json", exclude={id_field, digest_field}) + digest = canonical_hash(material) + expected_id = f"{prefix}:{digest[:32]}" + expected_digest = f"sha256:{digest}" + if getattr(instance, id_field) not in {None, expected_id}: + raise ValueError(f"{id_field} does not match exact contract material") + if getattr(instance, digest_field) not in {None, expected_digest}: + raise ValueError(f"{digest_field} does not match exact contract material") + object.__setattr__(instance, id_field, expected_id) + object.__setattr__(instance, digest_field, expected_digest) + + +class IntelligenceResourceKind(StrEnum): + """Stable domain-neutral families rendered by Atrium and machine consumers.""" + + CONNECTION = "connection" + SOURCE = "source" + SOURCE_HEALTH = "source_health" + ENTITY = "entity" + OBSERVATION = "observation" + SIGNAL = "signal" + SHIFT = "shift" + CASE = "case" + BRIEF = "brief" + MONITOR = "monitor" + SUBSCRIPTION = "subscription" + AGENT = "agent" + DECISION = "decision" + ACTION = "action" + OUTCOME = "outcome" + FEEDBACK = "feedback" + EVIDENCE_LINEAGE = "evidence_lineage" + UNCERTAINTY = "uncertainty" + CONFLICT = "conflict" + SEMANTIC_REVISION = "semantic_revision" + CONTEXT_MANIFEST = "context_manifest" + MEMORY_USE = "memory_use" + + +class IntelligenceResourceAvailability(StrEnum): + AVAILABLE = "available" + DEGRADED = "degraded" + TOMBSTONED = "tombstoned" + + +class IntelligenceResourcePageState(StrEnum): + COMPLETE = "complete" + DEGRADED = "degraded" + + +class IntelligenceResourceReferenceV1Alpha1(_StrictFrozenContract): + """Exact reference to one projected resource revision.""" + + contract: Literal["ace.intelligence.resource-plane-reference/v1alpha1"] = RESOURCE_PLANE_REFERENCE_VERSION + product_id: str + resource_kind: IntelligenceResourceKind + resource_id: str + resource_digest: str + resource_contract: str + revision: StrictInt = Field(ge=1) + as_of: datetime + available_at: datetime + + @field_validator("product_id") + @classmethod + def validate_scope(cls, value: str) -> str: + return validate_product_id(value) + + @field_validator("resource_id") + @classmethod + def validate_id(cls, value: str) -> str: + return validate_reference(value, name="resource_id") + + @field_validator("resource_digest") + @classmethod + def validate_resource_digest(cls, value: str) -> str: + return validate_digest(value) + + @field_validator("resource_contract") + @classmethod + def validate_resource_contract(cls, value: str) -> str: + return validate_contract(value) + + @field_validator("as_of", "available_at") + @classmethod + def normalize_times(cls, value: datetime, info) -> datetime: + return _aware(value, name=info.field_name) + + @model_validator(mode="after") + def validate_availability(self) -> Self: + if self.available_at < self.as_of: + raise ValueError("resource available_at cannot precede as_of") + return self + + +class IntelligenceResourceRecordV1Alpha1(_StrictFrozenContract): + """One rebuildable public projection with exact provenance and revision lineage.""" + + contract: Literal["ace.intelligence.resource-plane-record/v1alpha1"] = RESOURCE_PLANE_RECORD_VERSION + reference: IntelligenceResourceReferenceV1Alpha1 + availability: IntelligenceResourceAvailability + title: str = Field(min_length=1, max_length=300) + summary: str | None = Field(default=None, min_length=1, max_length=4_000) + subject_refs: tuple[str, ...] = Field(default_factory=tuple, max_length=MAX_RESOURCE_PLANE_SUBJECTS) + provenance: tuple[IntelligenceResourceReferenceV1Alpha1, ...] = Field( + default_factory=tuple, + max_length=MAX_RESOURCE_PLANE_PROVENANCE, + ) + supersedes: IntelligenceResourceReferenceV1Alpha1 | None = None + payload: CanonicalJsonValueV1Alpha1 | None = None + degraded_reason_refs: tuple[str, ...] = Field(default_factory=tuple, max_length=64) + + @field_validator("subject_refs", "degraded_reason_refs") + @classmethod + def normalize_refs(cls, value: tuple[str, ...], info) -> tuple[str, ...]: + return _unique_references(value, name=info.field_name) + + @field_validator("provenance") + @classmethod + def normalize_provenance( + cls, + value: tuple[IntelligenceResourceReferenceV1Alpha1, ...], + ) -> tuple[IntelligenceResourceReferenceV1Alpha1, ...]: + keys = [(item.resource_kind.value, item.resource_id, item.revision, item.resource_digest) for item in value] + if len(keys) != len(set(keys)): + raise ValueError("provenance references must be unique") + return tuple(sorted(value, key=lambda item: (item.resource_kind.value, item.resource_id, item.revision))) + + @model_validator(mode="after") + def validate_projection(self) -> Self: + for item in self.provenance: + if item.product_id != self.reference.product_id: + raise ValueError("resource provenance crossed product scope") + if item.as_of > self.reference.as_of or item.available_at > self.reference.available_at: + raise ValueError("resource provenance was not available at the projected revision") + if self.supersedes is not None: + if ( + self.supersedes.product_id != self.reference.product_id + or self.supersedes.resource_kind is not self.reference.resource_kind + or self.supersedes.resource_id != self.reference.resource_id + or self.supersedes.revision != self.reference.revision - 1 + ): + raise ValueError("supersedes must identify the immediately previous revision of the same resource") + if self.availability is IntelligenceResourceAvailability.AVAILABLE and self.degraded_reason_refs: + raise ValueError("an available resource cannot declare degraded reasons") + if self.availability is IntelligenceResourceAvailability.DEGRADED and not self.degraded_reason_refs: + raise ValueError("a degraded resource requires explicit reason references") + if self.availability is IntelligenceResourceAvailability.TOMBSTONED and self.payload is not None: + raise ValueError("a tombstoned projection cannot expose payload material") + return self + + +class IntelligenceResourceCursorV1Alpha1(_StrictFrozenContract): + """Content-addressed pagination position; explicitly not reusable authority.""" + + contract: Literal["ace.intelligence.resource-plane-cursor/v1alpha1"] = RESOURCE_PLANE_CURSOR_VERSION + query_id: str + after_available_at: datetime + after_resource_kind: IntelligenceResourceKind + after_resource_id: str + after_revision: StrictInt = Field(ge=1) + cursor_id: str | None = None + cursor_digest: str | None = None + + @field_validator("query_id", "after_resource_id") + @classmethod + def validate_refs(cls, value: str, info) -> str: + return validate_reference(value, name=info.field_name) + + @field_validator("after_available_at") + @classmethod + def normalize_after(cls, value: datetime) -> datetime: + return _aware(value, name="after_available_at") + + @field_validator("cursor_digest") + @classmethod + def validate_optional_digest(cls, value: str | None) -> str | None: + return validate_digest(value) if value is not None else None + + @model_validator(mode="after") + def derive_identity(self) -> Self: + _derive_identity(self, prefix="resource_cursor", id_field="cursor_id", digest_field="cursor_digest") + return self + + @property + def reusable_authority(self) -> Literal[False]: + return False + + +class IntelligenceResourceQueryV1Alpha1(_StrictFrozenContract): + """One authenticated, product-scoped, point-in-time resource query.""" + + contract: Literal["ace.intelligence.resource-plane-query/v1alpha1"] = RESOURCE_PLANE_QUERY_VERSION + authenticated_context: AuthenticatedRuntimeContextV1Alpha1 + product_id: str + authority_grant_ref: str + resource_kinds: tuple[IntelligenceResourceKind, ...] = Field( + min_length=1, + max_length=MAX_RESOURCE_PLANE_KINDS, + ) + subject_refs: tuple[str, ...] = Field(default_factory=tuple, max_length=MAX_RESOURCE_PLANE_SUBJECTS) + as_of: datetime + available_at: datetime + page_size: StrictInt = Field(ge=1, le=MAX_RESOURCE_PLANE_PAGE_SIZE) + cursor: IntelligenceResourceCursorV1Alpha1 | None = None + query_id: str | None = None + query_digest: str | None = None + + @field_validator("product_id") + @classmethod + def validate_scope(cls, value: str) -> str: + return validate_product_id(value) + + @field_validator("authority_grant_ref") + @classmethod + def validate_grant(cls, value: str) -> str: + return validate_reference(value, name="authority_grant_ref") + + @field_validator("resource_kinds") + @classmethod + def normalize_kinds(cls, value: tuple[IntelligenceResourceKind, ...]) -> tuple[IntelligenceResourceKind, ...]: + normalized = tuple(sorted(value, key=lambda item: item.value)) + if len(normalized) != len(set(normalized)): + raise ValueError("resource_kinds must be unique") + return normalized + + @field_validator("subject_refs") + @classmethod + def normalize_subjects(cls, value: tuple[str, ...]) -> tuple[str, ...]: + return _unique_references(value, name="subject_refs") + + @field_validator("as_of", "available_at") + @classmethod + def normalize_times(cls, value: datetime, info) -> datetime: + return _aware(value, name=info.field_name) + + @field_validator("query_digest") + @classmethod + def validate_optional_digest(cls, value: str | None) -> str | None: + return validate_digest(value) if value is not None else None + + @model_validator(mode="after") + def validate_scope_time_and_identity(self) -> Self: + if self.authenticated_context.product_id != self.product_id: + raise ValueError("resource query crossed authenticated product scope") + if self.available_at < self.as_of: + raise ValueError("query available_at cannot precede as_of") + if not ( + self.authenticated_context.authenticated_at <= self.available_at < self.authenticated_context.expires_at + ): + raise ValueError("query availability cutoff must be inside the authenticated window") + material = self.model_dump(mode="json", exclude={"cursor", "query_id", "query_digest"}) + digest = canonical_hash(material) + expected_id = f"resource_query:{digest[:32]}" + expected_digest = f"sha256:{digest}" + if self.query_id not in {None, expected_id} or self.query_digest not in {None, expected_digest}: + raise ValueError("query identity does not match exact selector material") + object.__setattr__(self, "query_id", expected_id) + object.__setattr__(self, "query_digest", expected_digest) + if self.cursor is not None and self.cursor.query_id != expected_id: + raise ValueError("pagination cursor belongs to a different query") + return self + + +class IntelligenceResourcePageV1Alpha1(_StrictFrozenContract): + """Authorized page returned from the unified resource plane.""" + + contract: Literal["ace.intelligence.resource-plane-page/v1alpha1"] = RESOURCE_PLANE_PAGE_VERSION + query_id: str + query_digest: str + product_id: str + actor_ref: str + as_of: datetime + available_at: datetime + evaluated_at: datetime + state: IntelligenceResourcePageState + items: tuple[IntelligenceResourceRecordV1Alpha1, ...] = Field(max_length=MAX_RESOURCE_PLANE_PAGE_SIZE) + next_cursor: IntelligenceResourceCursorV1Alpha1 | None = None + degraded_reason_refs: tuple[str, ...] = Field(default_factory=tuple, max_length=64) + authority_use: AuthorityUseReceiptV1Alpha1 + page_id: str | None = None + page_digest: str | None = None + + @field_validator("query_id", "actor_ref") + @classmethod + def validate_refs(cls, value: str, info) -> str: + return validate_reference(value, name=info.field_name) + + @field_validator("query_digest", "page_digest") + @classmethod + def validate_digests(cls, value: str | None) -> str | None: + return validate_digest(value) if value is not None else None + + @field_validator("product_id") + @classmethod + def validate_scope(cls, value: str) -> str: + return validate_product_id(value) + + @field_validator("as_of", "available_at", "evaluated_at") + @classmethod + def normalize_times(cls, value: datetime, info) -> datetime: + return _aware(value, name=info.field_name) + + @field_validator("degraded_reason_refs") + @classmethod + def normalize_reasons(cls, value: tuple[str, ...]) -> tuple[str, ...]: + return _unique_references(value, name="degraded_reason_refs") + + @model_validator(mode="after") + def validate_page(self) -> Self: + if self.available_at < self.as_of or self.evaluated_at < self.available_at: + raise ValueError("page time cutoffs must be monotonic") + if self.state is IntelligenceResourcePageState.COMPLETE and self.degraded_reason_refs: + raise ValueError("a complete page cannot declare degraded reasons") + if self.state is IntelligenceResourcePageState.DEGRADED and not self.degraded_reason_refs: + raise ValueError("a degraded page requires explicit reason references") + if any(item.reference.product_id != self.product_id for item in self.items): + raise ValueError("page items crossed product scope") + if any( + item.reference.as_of > self.as_of or item.reference.available_at > self.available_at for item in self.items + ): + raise ValueError("page contains resources outside its temporal cutoff") + ordering = [ + ( + item.reference.available_at, + item.reference.resource_kind.value, + item.reference.resource_id, + item.reference.revision, + ) + for item in self.items + ] + if ordering != sorted(ordering): + raise ValueError("page items must use stable ascending resource order") + if self.next_cursor is not None and self.next_cursor.query_id != self.query_id: + raise ValueError("next cursor belongs to a different query") + authority = self.authority_use + if ( + authority.product_id != self.product_id + or authority.actor_ref != self.actor_ref + 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.evaluated_at != self.evaluated_at + ): + raise ValueError("page does not preserve the exact resource-query authority evaluation") + _derive_identity(self, prefix="resource_page", id_field="page_id", digest_field="page_digest") + return self + + @property + def reusable_authority(self) -> Literal[False]: + return False + + +__all__ = [ + "MAX_RESOURCE_PLANE_PAGE_SIZE", + "RESOURCE_PLANE_CURSOR_VERSION", + "RESOURCE_PLANE_PAGE_VERSION", + "RESOURCE_PLANE_QUERY_VERSION", + "RESOURCE_PLANE_RECORD_VERSION", + "RESOURCE_PLANE_REFERENCE_VERSION", + "IntelligenceResourceAvailability", + "IntelligenceResourceCursorV1Alpha1", + "IntelligenceResourceKind", + "IntelligenceResourcePageState", + "IntelligenceResourcePageV1Alpha1", + "IntelligenceResourceQueryV1Alpha1", + "IntelligenceResourceRecordV1Alpha1", + "IntelligenceResourceReferenceV1Alpha1", +] diff --git a/docs/README.md b/docs/README.md index 30e5b50..4f56c1d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -78,6 +78,10 @@ support the public roadmap but do not compete with it for outcome state or dispa - [Intelligence OS runtime-boundary realignment](design/intelligence-os-runtime-boundary-v0.8.0-work-packet-v1.md) — the active 0.8B packet, accepted AM4 input, default isolation of embedded product-intelligence engines, compatibility switch, remaining runtime convergence, and rollback. +- [Unified Intelligence resource plane](design/intelligence-resource-plane-v0.8.0-work-packet-v1.md) + — the active 0.8C contract for domain-neutral resources, authorized point-in-time queries, + exact provenance, degradation, revision lineage, and rebuildable projections shared by Atrium, + World Intelligence, and Market Intelligence. - [Legacy host compatibility disposition](design/core-engine-compatibility-disposition-v0.8.0.json) — the machine-checked 0.8 owner and migration treatment for every top-level `core.engine` package; the canonical public roots remain `ace.core`, `ace.intelligence`, and `ace.application`. 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 new file mode 100644 index 0000000..f597979 --- /dev/null +++ b/docs/design/intelligence-resource-plane-v0.8.0-work-packet-v1.md @@ -0,0 +1,96 @@ +# ACE 0.8.0 unified Intelligence resource plane work packet + +Status: **active 0.8C packet; C1 public contracts and authorized query service 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) + +## Outcome + +ACE exposes one domain-neutral, authorized resource plane for people, applications, and Atrium. +Consumers can inspect the same Intelligence OS resources without reaching into legacy hosts, +inventing a second authority system, or learning Core's storage layout. + +The plane is a rebuildable read model over authoritative Core state and Intelligence-owned +projections. It does not acquire sources, execute effects, grant authority, or become a new +persistence engine. + +## Canonical resource families + +The first public contract covers: + +- Connections, Sources, and Source Health; +- Entities, Observations, Signals, Shifts, and Cases; +- Briefs, Monitors, Subscriptions, and Agents; +- Decisions, Actions, Outcomes, and Feedback; and +- Evidence Lineage, Uncertainty, Conflicts, Semantic Revisions, Context Manifests, and Memory Use. + +Atrium may present product language over these families. In particular, **Opportunities** is a +filtered and ranked Case experience, not a second kernel concept. Domain Packs provide vocabulary, +policy, projections, and presentation hints without adding public resource kinds to Core. + +## C1 — public query contract and application service + +The C1 seam provides: + +1. strict, frozen, versioned references, records, selectors, cursors, and pages; +2. stable product-scoped identities and immediately adjacent revision lineage; +3. exact provenance available no later than the projected resource revision; +4. explicit available, degraded, and tombstoned states; +5. point-in-time `as_of` and `available_at` cutoffs; +6. deterministic ascending pagination with content-addressed cursors; +7. current Core authority resolution for every page request; and +8. a rebuildable projection-reader port owned by adapters rather than the application service. + +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. + +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 +return payload material. + +## Ownership + +| Concern | Owner | +| --- | --- | +| Authentication, authority, governed state, durable receipts | Core | +| Resource meaning, evidence semantics, revision and projection rules | Intelligence | +| Authorized query composition and fail-closed boundary checks | Application | +| Source acquisition and read-model materialization | Adapters | +| Domain vocabulary, mappings, policy, templates, presentation hints | Domain Pack | +| Product navigation and human experience | Atrium | + +Projections may be cached, but they must be fully derivable from governed state and source receipts. +They are never authoritative and must declare degradation when their dependencies cannot reproduce +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. + +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 +Intelligence. + +## Acceptance + +C1 passes only when: + +1. all canonical families are represented without domain nouns; +2. public `ace.intelligence` and `ace.application` imports expose the contracts and service; +3. exact schema generation succeeds; +4. product, subject, kind, time, and pagination widening fail closed; +5. provenance, degradation, tombstone, and immediate-supersession rules are tested; +6. current Core authority is re-resolved on every request; +7. the resource plane imports no host, transport, connector, or extension implementation; +8. the exact eleven-tool MCP surface remains unchanged; and +9. no World, Market, UI, or provider code enters Core. + +## Rollback + +C1 is additive. Reverting its contracts, service, exports, tests, and this packet removes the public +facade without rewriting governed state or projections. No existing runtime path depends on it until +C2 installs an explicit adapter. diff --git a/tests/intelligence/test_intelligence_resource_plane.py b/tests/intelligence/test_intelligence_resource_plane.py new file mode 100644 index 0000000..63192eb --- /dev/null +++ b/tests/intelligence/test_intelligence_resource_plane.py @@ -0,0 +1,363 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from ace.application import IntelligenceResourcePlaneService as PublicIntelligenceResourcePlaneService +from ace.application.intelligence_resource_plane import ( + RESOURCE_QUERY_AUTHORITY, + RESOURCE_QUERY_OPERATION, + IntelligenceResourcePlaneError, + IntelligenceResourcePlaneService, + IntelligenceResourceProjectionBatch, +) +from ace.core.runtime_use import AuthenticatedRuntimeContextV1Alpha1, AuthorityUseReceiptV1Alpha1 +from ace.core.state import GovernedStateHeadPreconditionV1Alpha1 +from ace.intelligence import IntelligenceResourceKind as PublicIntelligenceResourceKind +from ace.intelligence.contracts.resource_plane import ( + IntelligenceResourceAvailability, + IntelligenceResourceCursorV1Alpha1, + IntelligenceResourceKind, + IntelligenceResourcePageState, + IntelligenceResourceQueryV1Alpha1, + IntelligenceResourceRecordV1Alpha1, + IntelligenceResourceReferenceV1Alpha1, +) +from ace.intelligence.contracts.resources import CanonicalJsonValueV1Alpha1 + +pytestmark = pytest.mark.unit + +PRODUCT = "product:resource-plane" +NOW = datetime(2026, 8, 12, 20, 0, tzinfo=UTC) + + +def _context(*, product_id: str = PRODUCT) -> AuthenticatedRuntimeContextV1Alpha1: + return AuthenticatedRuntimeContextV1Alpha1( + product_id=product_id, + actor_ref="principal:analyst", + authentication_receipt_ref="authentication_receipt:resource-plane", + authentication_receipt_digest="sha256:" + "a" * 64, + authenticated_at=NOW - timedelta(minutes=10), + expires_at=NOW + timedelta(minutes=30), + ) + + +def _reference( + kind: IntelligenceResourceKind, + suffix: str, + *, + product_id: str = PRODUCT, + revision: int = 1, + available_at: datetime = NOW, +) -> IntelligenceResourceReferenceV1Alpha1: + return IntelligenceResourceReferenceV1Alpha1( + product_id=product_id, + resource_kind=kind, + resource_id=f"{kind.value}:{suffix}", + resource_digest="sha256:" + (suffix[0] if suffix[0] in "abcdef" else "b") * 64, + resource_contract=f"ace.intelligence.{kind.value.replace('_', '-')}/v1alpha1", + revision=revision, + as_of=available_at - timedelta(minutes=1), + available_at=available_at, + ) + + +def _record( + kind: IntelligenceResourceKind, + suffix: str, + *, + product_id: str = PRODUCT, + revision: int = 1, + available_at: datetime = NOW, + subject_refs: tuple[str, ...] = ("entity:ai-model",), +) -> IntelligenceResourceRecordV1Alpha1: + return IntelligenceResourceRecordV1Alpha1( + reference=_reference( + kind, + suffix, + product_id=product_id, + revision=revision, + available_at=available_at, + ), + availability=IntelligenceResourceAvailability.AVAILABLE, + title=f"{kind.value} {suffix}", + subject_refs=subject_refs, + payload=CanonicalJsonValueV1Alpha1(value_json='{"status":"current"}'), + ) + + +def _query( + *, + kinds: tuple[IntelligenceResourceKind, ...] = (IntelligenceResourceKind.SIGNAL,), + page_size: int = 2, + cursor: IntelligenceResourceCursorV1Alpha1 | None = None, + subject_refs: tuple[str, ...] = (), +) -> IntelligenceResourceQueryV1Alpha1: + 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, + available_at=NOW, + page_size=page_size, + cursor=cursor, + ) + + +class _Authority: + def __init__(self, *, mutate_operation: bool = False) -> None: + self.calls: list[dict] = [] + self.mutate_operation = mutate_operation + + async def resolve_authority_use(self, **kwargs) -> AuthorityUseReceiptV1Alpha1: + self.calls.append(kwargs) + return AuthorityUseReceiptV1Alpha1( + product_id=kwargs["context"].product_id, + actor_ref=kwargs["context"].actor_ref, + authenticated_context=kwargs["context"], + use_subject_ref=kwargs["use_subject_ref"], + use_subject_digest=kwargs["use_subject_digest"], + operation="inspect" if self.mutate_operation else kwargs["operation"], + authority=kwargs["authority"], + grant_ref=kwargs["grant_ref"], + grant_hash="d" * 64, + evaluated_at=kwargs["evaluated_at"], + expires_at=NOW + timedelta(minutes=20), + state_head_precondition=GovernedStateHeadPreconditionV1Alpha1( + state_kind="authority_grant", + product_id=kwargs["context"].product_id, + state_id=kwargs["grant_ref"], + sequence=1, + revision_id="authority_revision:resource-read", + commit_receipt_id="authority_receipt:resource-read", + ), + ) + + +class _Reader: + def __init__( + self, + records: tuple[IntelligenceResourceRecordV1Alpha1, ...], + *, + state: IntelligenceResourcePageState = IntelligenceResourcePageState.COMPLETE, + reasons: tuple[str, ...] = (), + ) -> None: + self.records = records + self.state = state + self.reasons = reasons + self.calls: list[dict] = [] + + async def read(self, **kwargs) -> IntelligenceResourceProjectionBatch: + self.calls.append(kwargs) + return IntelligenceResourceProjectionBatch( + records=self.records, + state=self.state, + degraded_reason_refs=self.reasons, + ) + + +def test_resource_plane_covers_the_complete_0_8_public_family() -> None: + assert {item.value for item in IntelligenceResourceKind} == { + "connection", + "source", + "source_health", + "entity", + "observation", + "signal", + "shift", + "case", + "brief", + "monitor", + "subscription", + "agent", + "decision", + "action", + "outcome", + "feedback", + "evidence_lineage", + "uncertainty", + "conflict", + "semantic_revision", + "context_manifest", + "memory_use", + } + + +def test_resource_plane_is_exported_through_the_supported_public_packages() -> None: + assert PublicIntelligenceResourcePlaneService is IntelligenceResourcePlaneService + assert PublicIntelligenceResourceKind is IntelligenceResourceKind + assert IntelligenceResourceQueryV1Alpha1.model_json_schema()["type"] == "object" + + +def test_query_identity_excludes_cursor_but_cursor_is_bound_to_the_exact_query() -> None: + first = _query() + cursor = IntelligenceResourceCursorV1Alpha1( + query_id=str(first.query_id), + after_available_at=NOW, + after_resource_kind=IntelligenceResourceKind.SIGNAL, + after_resource_id="signal:b", + after_revision=1, + ) + second = _query(cursor=cursor) + assert second.query_id == first.query_id + assert second.query_digest == first.query_digest + assert cursor.reusable_authority is False + + with pytest.raises(ValueError, match="different query"): + _query( + cursor=cursor, + kinds=(IntelligenceResourceKind.BRIEF,), + ) + + +def test_record_requires_exact_provenance_scope_and_revision_lineage() -> None: + prior = _reference(IntelligenceResourceKind.ENTITY, "a", revision=1) + current = _reference(IntelligenceResourceKind.ENTITY, "a", revision=2, available_at=NOW + timedelta(minutes=1)) + record = IntelligenceResourceRecordV1Alpha1( + reference=current, + availability=IntelligenceResourceAvailability.AVAILABLE, + title="Entity revision", + provenance=(prior,), + supersedes=prior, + payload=CanonicalJsonValueV1Alpha1(value_json='{"name":"ACE"}'), + ) + assert record.supersedes == prior + + with pytest.raises(ValueError, match="crossed product scope"): + IntelligenceResourceRecordV1Alpha1( + reference=current, + availability=IntelligenceResourceAvailability.AVAILABLE, + title="Invalid entity revision", + provenance=(_reference(IntelligenceResourceKind.SOURCE, "b", product_id="product:other"),), + ) + + with pytest.raises(ValueError, match="immediately previous revision"): + IntelligenceResourceRecordV1Alpha1( + reference=_reference( + IntelligenceResourceKind.ENTITY, + "a", + revision=3, + available_at=NOW + timedelta(minutes=2), + ), + availability=IntelligenceResourceAvailability.AVAILABLE, + title="Skipped entity lineage", + supersedes=prior, + ) + + +def test_degraded_and_tombstoned_records_fail_closed_on_missing_truth() -> None: + with pytest.raises(ValueError, match="requires explicit reason"): + IntelligenceResourceRecordV1Alpha1( + reference=_reference(IntelligenceResourceKind.SOURCE_HEALTH, "a"), + availability=IntelligenceResourceAvailability.DEGRADED, + title="Source health unavailable", + ) + with pytest.raises(ValueError, match="cannot expose payload"): + IntelligenceResourceRecordV1Alpha1( + reference=_reference(IntelligenceResourceKind.MEMORY_USE, "a"), + availability=IntelligenceResourceAvailability.TOMBSTONED, + title="Erased memory use", + payload=CanonicalJsonValueV1Alpha1(value_json="{}"), + ) + + +@pytest.mark.asyncio +async def test_authorized_query_returns_stable_page_and_next_cursor() -> None: + records = ( + _record(IntelligenceResourceKind.SIGNAL, "a", available_at=NOW - timedelta(seconds=2)), + _record(IntelligenceResourceKind.SIGNAL, "b", available_at=NOW - timedelta(seconds=1)), + _record(IntelligenceResourceKind.SIGNAL, "c", available_at=NOW), + ) + reader = _Reader(records) + authority = _Authority() + service = IntelligenceResourcePlaneService(reader=reader, authority=authority) + request = _query(page_size=2) + + page = await service.query(request, evaluated_at=NOW + timedelta(minutes=1)) + + assert page.items == records[:2] + assert page.next_cursor is not None + assert page.next_cursor.after_resource_id == "signal:b" + assert page.next_cursor.query_id == request.query_id + assert page.authority_use.operation == RESOURCE_QUERY_OPERATION + assert page.authority_use.authority == RESOURCE_QUERY_AUTHORITY + assert page.reusable_authority is False + assert reader.calls[0]["limit"] == 3 + assert authority.calls[0]["use_subject_ref"] == request.query_id + + +@pytest.mark.asyncio +async def test_subject_filter_and_degraded_state_are_preserved() -> None: + record = _record( + IntelligenceResourceKind.BRIEF, + "a", + subject_refs=("entity:ai-model", "entity:provider"), + ) + reader = _Reader( + (record,), + state=IntelligenceResourcePageState.DEGRADED, + reasons=("degraded_reason:stale-secondary-source",), + ) + page = await IntelligenceResourcePlaneService(reader=reader, authority=_Authority()).query( + _query( + kinds=(IntelligenceResourceKind.BRIEF,), + subject_refs=("entity:provider",), + ), + evaluated_at=NOW + timedelta(minutes=1), + ) + assert page.state is IntelligenceResourcePageState.DEGRADED + assert page.degraded_reason_refs == ("degraded_reason:stale-secondary-source",) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("record", "message"), + [ + (_record(IntelligenceResourceKind.SIGNAL, "a", product_id="product:other"), "crossed product scope"), + (_record(IntelligenceResourceKind.BRIEF, "a"), "unrequested resource kind"), + ( + _record(IntelligenceResourceKind.SIGNAL, "a", available_at=NOW + timedelta(minutes=1)), + "temporal cutoff", + ), + ( + _record(IntelligenceResourceKind.SIGNAL, "a", subject_refs=("entity:other",)), + "outside the subject filter", + ), + ], +) +async def test_reader_cannot_widen_query_scope(record, message: str) -> None: + request = _query(subject_refs=("entity:ai-model",)) + with pytest.raises(IntelligenceResourcePlaneError, match=message): + await IntelligenceResourcePlaneService(reader=_Reader((record,)), authority=_Authority()).query( + request, + evaluated_at=NOW + timedelta(minutes=1), + ) + + +@pytest.mark.asyncio +async def test_reader_must_advance_stably_beyond_cursor() -> None: + first = _query(page_size=1) + cursor = IntelligenceResourceCursorV1Alpha1( + query_id=str(first.query_id), + after_available_at=NOW, + after_resource_kind=IntelligenceResourceKind.SIGNAL, + after_resource_id="signal:b", + after_revision=1, + ) + with pytest.raises(IntelligenceResourcePlaneError, match="did not advance"): + await IntelligenceResourcePlaneService( + reader=_Reader((_record(IntelligenceResourceKind.SIGNAL, "b"),)), + authority=_Authority(), + ).query(_query(page_size=1, cursor=cursor), evaluated_at=NOW + timedelta(minutes=1)) + + +@pytest.mark.asyncio +async def test_authority_receipt_cannot_substitute_operation() -> None: + with pytest.raises(IntelligenceResourcePlaneError, match="did not preserve"): + await IntelligenceResourcePlaneService( + reader=_Reader(()), + authority=_Authority(mutate_operation=True), + ).query(_query(), evaluated_at=NOW + timedelta(minutes=1))