diff --git a/ace/application/__init__.py b/ace/application/__init__.py index 1953563..9fe846d 100644 --- a/ace/application/__init__.py +++ b/ace/application/__init__.py @@ -297,6 +297,16 @@ WatchTargetV1, intelligence_model_semantic_diff, ) +from ace.application.intelligence_build_execution import ( + REQUIRED_INTELLIGENCE_BUILD_EFFECTS, + AuthorizedIntelligenceBuild, + IntelligenceBuildEffect, + IntelligenceBuildExecutor, + IntelligenceBuildHostServices, + IntelligenceBuildResourcePagePort, + IntelligenceBuildStartV1, + ProductScopedImmutableRecordStore, +) from ace.application.intelligence_builder import ( ConnectionAgent, ConnectionAgentError, @@ -760,6 +770,14 @@ "IntelligenceBuilderSessionReplayConflict", "IntelligenceBuilderSessionRevisionV1", "IntelligenceBuilderSessionService", + "IntelligenceBuildEffect", + "IntelligenceBuildExecutor", + "IntelligenceBuildHostServices", + "IntelligenceBuildResourcePagePort", + "IntelligenceBuildStartV1", + "AuthorizedIntelligenceBuild", + "ProductScopedImmutableRecordStore", + "REQUIRED_INTELLIGENCE_BUILD_EFFECTS", "IntelligenceActivationPlanV1Alpha2", "IntelligenceAgent", "IntelligenceAgentAttributionError", diff --git a/ace/application/intelligence_build_execution.py b/ace/application/intelligence_build_execution.py new file mode 100644 index 0000000..c491c61 --- /dev/null +++ b/ace/application/intelligence_build_execution.py @@ -0,0 +1,183 @@ +"""Public, domain-neutral contract for trusted Intelligence build executors. + +Core authorizes an exact onboarding request before handing this material to an +installed executor. Executors may interpret only the profile they declare and +must return a product- and actor-scoped Intelligence resource page. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Literal, Protocol + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from ace.application.intelligence_resource_plane import ( + IntelligenceResourceKind, + IntelligenceResourcePageV1Alpha1, +) +from ace.core.records import ( + AppendOnlyTransactionReceiptV1, + AppendOnlyTransactionRequestV1, + ImmutableRecordScopeError, + ImmutableRecordStore, + ImmutableRecordV1, +) +from ace.core.runtime_use import AuthorityUseReceiptV1Alpha1 + +IntelligenceBuildEffect = Literal[ + "connect_sources", + "map_concepts", + "activate_watch", + "create_first_brief", +] +REQUIRED_INTELLIGENCE_BUILD_EFFECTS: tuple[IntelligenceBuildEffect, ...] = ( + "connect_sources", + "map_concepts", + "activate_watch", + "create_first_brief", +) + + +class IntelligenceBuildStartV1(BaseModel): + """One reviewed Atrium plan submitted for governed execution.""" + + model_config = ConfigDict(extra="forbid") + + authority_grant_ref: str = Field(min_length=1, max_length=240) + resource_authority_grant_ref: str = Field(min_length=1, max_length=240) + client_request_id: str = Field(min_length=1, max_length=240) + profile_id: str = Field(min_length=1, max_length=240) + subject: str = Field(min_length=8, max_length=2_000) + outcome_id: str = Field(min_length=1, max_length=240) + source_group_ids: tuple[str, ...] = Field(default_factory=tuple, max_length=64) + cadence_id: str = Field(min_length=1, max_length=240) + approved_effects: tuple[IntelligenceBuildEffect, ...] + requested_at: datetime + + @field_validator("source_group_ids") + @classmethod + def _unique_source_groups(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if len(value) != len(set(value)): + raise ValueError("source_group_ids must be unique") + return value + + @field_validator("approved_effects") + @classmethod + def _exact_bounded_effects(cls, value: tuple[IntelligenceBuildEffect, ...]) -> tuple[IntelligenceBuildEffect, ...]: + if value != REQUIRED_INTELLIGENCE_BUILD_EFFECTS: + raise ValueError("approved_effects must preserve the exact bounded onboarding effect sequence") + return value + + +@dataclass(frozen=True, slots=True) +class AuthorizedIntelligenceBuild: + """Exact identity, scope, request, and authority admitted by Core.""" + + build_id: str + request_digest: str + product_id: str + actor_ref: str + request: IntelligenceBuildStartV1 + authority_use: AuthorityUseReceiptV1Alpha1 + + +class ProductScopedImmutableRecordStore: + """Restrict every public immutable-record operation to one product fence.""" + + def __init__(self, *, product_id: str, store: ImmutableRecordStore) -> None: + self.product_id = product_id + self._store = store + + def _check(self, product_id: str) -> None: + if product_id != self.product_id: + raise ImmutableRecordScopeError("Intelligence build store crossed its authorized product scope") + + async def append(self, request: AppendOnlyTransactionRequestV1) -> AppendOnlyTransactionReceiptV1: + self._check(request.product_id) + return await self._store.append(request) + + async def load_record(self, storage_id: str, *, product_id: str, record_space: str, record_kind: str): + self._check(product_id) + return await self._store.load_record( + storage_id, + product_id=product_id, + record_space=record_space, + record_kind=record_kind, + ) + + async def load_transaction_receipt(self, *, product_id: str, record_space: str, transaction_key: str): + self._check(product_id) + return await self._store.load_transaction_receipt( + product_id=product_id, + record_space=record_space, + transaction_key=transaction_key, + ) + + async def read_as_of( + self, *, product_id: str, record_space: str, record_kind: str, available_at: datetime + ) -> tuple[ImmutableRecordV1, ...]: + self._check(product_id) + return await self._store.read_as_of( + product_id=product_id, + record_space=record_space, + record_kind=record_kind, + available_at=available_at, + ) + + async def count_as_of(self, *, product_id: str, record_space: str, record_kind: str, available_at: datetime) -> int: + self._check(product_id) + return await self._store.count_as_of( + product_id=product_id, + record_space=record_space, + record_kind=record_kind, + available_at=available_at, + ) + + async def scan_product_records(self, *, product_id: str) -> tuple[ImmutableRecordV1, ...]: + self._check(product_id) + return await self._store.scan_product_records(product_id=product_id) + + +class IntelligenceBuildResourcePagePort(Protocol): + """Core-owned projection and read-authority boundary for one authorized build.""" + + async def query( + self, + *, + resource_kinds: tuple[IntelligenceResourceKind, ...], + subject_refs: tuple[str, ...], + as_of: datetime, + available_at: datetime, + evaluated_at: datetime, + page_size: int = 200, + ) -> IntelligenceResourcePageV1Alpha1: ... + + +@dataclass(frozen=True, slots=True) +class IntelligenceBuildHostServices: + """Invocation-scoped capabilities Core grants to one trusted executor.""" + + records: ImmutableRecordStore + resources: IntelligenceBuildResourcePagePort + + +class IntelligenceBuildExecutor(Protocol): + """Trusted executable adapter for one or more exact onboarding profiles.""" + + async def start( + self, build: AuthorizedIntelligenceBuild, host_services: IntelligenceBuildHostServices + ) -> IntelligenceResourcePageV1Alpha1: ... + + +__all__ = [ + "AuthorizedIntelligenceBuild", + "IntelligenceBuildEffect", + "IntelligenceBuildExecutor", + "IntelligenceBuildHostServices", + "IntelligenceBuildResourcePagePort", + "IntelligenceBuildStartV1", + "ProductScopedImmutableRecordStore", + "REQUIRED_INTELLIGENCE_BUILD_EFFECTS", +] diff --git a/core/engine/core/intelligence_build.py b/core/engine/core/intelligence_build.py index 4c86ca9..628471f 100644 --- a/core/engine/core/intelligence_build.py +++ b/core/engine/core/intelligence_build.py @@ -11,9 +11,16 @@ from datetime import UTC, datetime, timedelta from typing import Literal, Protocol -from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator +from pydantic import BaseModel, ConfigDict, ValidationError from ace.application import IntelligenceResourcePageV1Alpha1 +from ace.application.intelligence_build_execution import ( + AuthorizedIntelligenceBuild, + IntelligenceBuildExecutor, + IntelligenceBuildHostServices, + IntelligenceBuildStartV1, + ProductScopedImmutableRecordStore, +) from ace.core import ImmutableRecordStore from ace.core.contracts import canonical_hash from ace.core.runtime_use import AuthorityUseReceiptV1Alpha1 @@ -25,34 +32,17 @@ from core.engine.core.db import pool from core.engine.core.governed_state import SurrealGovernedStateStore from core.engine.core.immutable_records import SurrealImmutableRecordStore +from core.engine.core.intelligence_build_executor_registry import ( + IntelligenceBuildExecutorRegistryError, + resolve_intelligence_build_executor, +) +from core.engine.core.intelligence_resource_plane import CoreIntelligenceBuildResourcePagePort INTELLIGENCE_BUILD_AUTHORITY = "intelligence_build" INTELLIGENCE_BUILD_OPERATION = "start_intelligence_build" INTELLIGENCE_BUILD_RESULT_VERSION = "ace.http.intelligence-build-result/v1alpha1" -class IntelligenceBuildStartV1(BaseModel): - """One reviewed Atrium plan submitted for governed execution.""" - - model_config = ConfigDict(extra="forbid") - - authority_grant_ref: str = Field(min_length=1, max_length=240) - client_request_id: str = Field(min_length=1, max_length=240) - profile_id: str = Field(min_length=1, max_length=240) - subject: str = Field(min_length=8, max_length=2_000) - outcome_id: str = Field(min_length=1, max_length=240) - source_group_ids: tuple[str, ...] = Field(default_factory=tuple, max_length=64) - cadence_id: str = Field(min_length=1, max_length=240) - requested_at: datetime - - @field_validator("source_group_ids") - @classmethod - def _unique_source_groups(cls, value: tuple[str, ...]) -> tuple[str, ...]: - if len(value) != len(set(value)): - raise ValueError("source_group_ids must be unique") - return value - - class IntelligenceBuildResultV1(BaseModel): """Authorized result returned after a host completes or safely blocks a build.""" @@ -68,20 +58,6 @@ class IntelligenceBuildResultV1(BaseModel): resource_page: IntelligenceResourcePageV1Alpha1 -@dataclass(frozen=True, slots=True) -class AuthorizedIntelligenceBuild: - build_id: str - request_digest: str - product_id: str - actor_ref: str - request: IntelligenceBuildStartV1 - authority_use: AuthorityUseReceiptV1Alpha1 - - -class IntelligenceBuildExecutor(Protocol): - async def start(self, build: AuthorizedIntelligenceBuild) -> IntelligenceResourcePageV1Alpha1: ... - - class IntelligenceBuildAuthorizationPort(Protocol): async def resolve_authority_use( self, @@ -123,10 +99,19 @@ class IntelligenceBuildContractConflict(IntelligenceBuildError): """A host result did not preserve the authorized request.""" -class _UnavailableIntelligenceBuildExecutor: - async def start(self, build: AuthorizedIntelligenceBuild) -> IntelligenceResourcePageV1Alpha1: - del build - raise IntelligenceBuildUnavailable("no Intelligence build executor is registered") +class _InstalledIntelligenceBuildExecutor: + async def start( + self, build: AuthorizedIntelligenceBuild, host_services: IntelligenceBuildHostServices + ) -> IntelligenceResourcePageV1Alpha1: + try: + executor = resolve_intelligence_build_executor(build.request.profile_id) + except IntelligenceBuildExecutorRegistryError as exc: + raise IntelligenceBuildUnavailable("installed Intelligence build executors are ambiguous") from exc + if executor is None: + raise IntelligenceBuildUnavailable( + f"no Intelligence build executor is registered for profile: {build.request.profile_id}" + ) + return await executor.start(build, host_services) def intelligence_build_runtime() -> IntelligenceBuildHttpRuntime: @@ -135,7 +120,7 @@ def intelligence_build_runtime() -> IntelligenceBuildHttpRuntime: return IntelligenceBuildHttpRuntime( records=records, authority=GovernedStateRuntimeUseResolver(governed_state=governed_state), - executor=_UnavailableIntelligenceBuildExecutor(), + executor=_InstalledIntelligenceBuildExecutor(), ) @@ -145,13 +130,16 @@ def _verified_claims(user: dict) -> tuple[str, str]: authorities = user.get("authorities") if not isinstance(actor_ref, str) or not actor_ref or not isinstance(product_id, str) or not product_id: raise IntelligenceBuildUnauthenticated("verified token lacks product scope") - if not isinstance(authorities, list) or INTELLIGENCE_BUILD_AUTHORITY not in authorities: - raise IntelligenceBuildDenied("Intelligence build authority is required") + if not isinstance(authorities, list) or not { + INTELLIGENCE_BUILD_AUTHORITY, + "observe_read", + }.issubset(authorities): + raise IntelligenceBuildDenied("Intelligence build and read authorities are required") return actor_ref, product_id def _request_identity(*, request: IntelligenceBuildStartV1, product_id: str, actor_ref: str) -> tuple[str, str]: - material = request.model_dump(mode="json", exclude={"authority_grant_ref"}) + material = request.model_dump(mode="json", exclude={"authority_grant_ref", "resource_authority_grant_ref"}) raw_digest = canonical_hash([product_id, actor_ref, material]) return f"intelligence_build:{raw_digest[:32]}", f"sha256:{raw_digest}" @@ -204,19 +192,25 @@ async def start_intelligence_build( ): raise IntelligenceBuildContractConflict("authority resolver changed the Intelligence build request") + authorized_build = AuthorizedIntelligenceBuild( + build_id=build_id, + request_digest=request_digest, + product_id=product_id, + actor_ref=actor_ref, + request=request, + authority_use=exact_authority, + ) + scoped_records = ProductScopedImmutableRecordStore(product_id=product_id, store=runtime.records) + host_services = IntelligenceBuildHostServices( + records=scoped_records, + resources=CoreIntelligenceBuildResourcePagePort( + build=authorized_build, + records=scoped_records, + authority=runtime.authority, + ), + ) page = IntelligenceResourcePageV1Alpha1.model_validate( - ( - await runtime.executor.start( - AuthorizedIntelligenceBuild( - build_id=build_id, - request_digest=request_digest, - product_id=product_id, - actor_ref=actor_ref, - request=request, - authority_use=exact_authority, - ) - ) - ).model_dump(mode="python") + (await runtime.executor.start(authorized_build, host_services)).model_dump(mode="python") ) if page.product_id != product_id or page.actor_ref != actor_ref: raise IntelligenceBuildContractConflict("build executor crossed the authenticated product scope") diff --git a/core/engine/core/intelligence_build_executor_registry.py b/core/engine/core/intelligence_build_executor_registry.py new file mode 100644 index 0000000..086c14a --- /dev/null +++ b/core/engine/core/intelligence_build_executor_registry.py @@ -0,0 +1,106 @@ +"""Fail-closed discovery of trusted Intelligence build executors. + +Executable adapters use the dedicated ``ace.intelligence_builders`` entry-point +group. Domain Packs remain inert resources and are never imported here. +""" + +from __future__ import annotations + +import os +import re +from importlib import metadata +from inspect import iscoroutinefunction +from typing import Iterable + +from ace.application.intelligence_build_execution import IntelligenceBuildExecutor + +INTELLIGENCE_BUILDER_ENTRY_POINT_GROUP = "ace.intelligence_builders" +_PROFILE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,239}$") + + +class IntelligenceBuildExecutorRegistryError(RuntimeError): + """Installed executor material is invalid or ambiguous.""" + + +_executors: dict[str, IntelligenceBuildExecutor] = {} +_loaded = False +_load_error: IntelligenceBuildExecutorRegistryError | None = None + + +def register_intelligence_build_executor( + *, profile_id: str, executor: IntelligenceBuildExecutor +) -> IntelligenceBuildExecutor: + """Register one trusted executor for one exact profile, rejecting ambiguity.""" + + if not isinstance(profile_id, str) or not _PROFILE_ID.fullmatch(profile_id): + raise IntelligenceBuildExecutorRegistryError("invalid Intelligence build profile id") + if not iscoroutinefunction(getattr(executor, "start", None)): + raise IntelligenceBuildExecutorRegistryError("Intelligence build executor omitted async start") + if profile_id in _executors: + raise IntelligenceBuildExecutorRegistryError( + f"multiple Intelligence build executors claim profile: {profile_id}" + ) + _executors[profile_id] = executor + return executor + + +def load_installed_intelligence_build_executors(entry_points: Iterable | None = None) -> tuple[str, ...]: + """Load the dedicated installed executor group exactly once. + + A bad or duplicate executable package fails the registry closed. The naked + kernel switch disables executable discovery for the process lifetime. + """ + + global _load_error, _loaded + if _loaded: + if _load_error is not None: + raise _load_error + return tuple(sorted(_executors)) + _loaded = True + if os.environ.get("ACE_DISABLE_EXTENSIONS") == "1": + return () + + installed = ( + metadata.entry_points(group=INTELLIGENCE_BUILDER_ENTRY_POINT_GROUP) if entry_points is None else entry_points + ) + prior = dict(_executors) + try: + for entry_point in sorted(installed, key=lambda item: item.name): + loaded = entry_point.load() + executor = loaded() if isinstance(loaded, type) else loaded + profile_id = getattr(executor, "profile_id", None) + register_intelligence_build_executor(profile_id=profile_id, executor=executor) + except IntelligenceBuildExecutorRegistryError as exc: + _executors.clear() + _executors.update(prior) + _load_error = exc + raise + except Exception as exc: + _executors.clear() + _executors.update(prior) + _load_error = IntelligenceBuildExecutorRegistryError( + f"Intelligence build executor failed to load: {entry_point.name}" + ) + raise _load_error from exc + return tuple(sorted(_executors)) + + +def resolve_intelligence_build_executor(profile_id: str) -> IntelligenceBuildExecutor | None: + load_installed_intelligence_build_executors() + return _executors.get(profile_id) + + +def _reset_intelligence_build_executor_registry_for_tests() -> None: + global _load_error, _loaded + _executors.clear() + _loaded = False + _load_error = None + + +__all__ = [ + "INTELLIGENCE_BUILDER_ENTRY_POINT_GROUP", + "IntelligenceBuildExecutorRegistryError", + "load_installed_intelligence_build_executors", + "register_intelligence_build_executor", + "resolve_intelligence_build_executor", +] diff --git a/core/engine/core/intelligence_resource_plane.py b/core/engine/core/intelligence_resource_plane.py index c4db010..f40c239 100644 --- a/core/engine/core/intelligence_resource_plane.py +++ b/core/engine/core/intelligence_resource_plane.py @@ -32,6 +32,7 @@ LiveSourceResourceProjectionReader, MonitoringResourceProjectionReader, ) +from ace.application.intelligence_build_execution import AuthorizedIntelligenceBuild from ace.core import ImmutableRecordPersistenceError, ImmutableRecordStore from core.engine.core.agent_composition_runtime import ( GovernedCompositionAuthorityError, @@ -127,6 +128,49 @@ def intelligence_resource_projection_reader(records: ImmutableRecordStore) -> In ) +class CoreIntelligenceBuildResourcePagePort: + """Resolve one exact read page for an already-authorized build invocation.""" + + def __init__( + self, + *, + build: AuthorizedIntelligenceBuild, + records: ImmutableRecordStore, + authority: IntelligenceResourcePlaneAuthorizationPort, + ) -> None: + self.build = build + self.records = records + self.authority = authority + + async def query( + self, + *, + resource_kinds: tuple[IntelligenceResourceKind, ...], + subject_refs: tuple[str, ...], + as_of: datetime, + available_at: datetime, + evaluated_at: datetime, + page_size: int = 200, + ) -> IntelligenceResourcePageV1Alpha1: + context = self.build.authority_use.authenticated_context + if context.product_id != self.build.product_id or context.actor_ref != self.build.actor_ref: + raise IntelligenceResourceHttpContractConflict("build context crossed its authorized scope") + request = IntelligenceResourceQueryV1Alpha1( + authenticated_context=context, + product_id=self.build.product_id, + authority_grant_ref=self.build.request.resource_authority_grant_ref, + resource_kinds=resource_kinds, + subject_refs=subject_refs, + as_of=as_of, + available_at=available_at, + page_size=page_size, + ) + return await IntelligenceResourcePlaneService( + reader=intelligence_resource_projection_reader(self.records), + authority=self.authority, + ).query(request, evaluated_at=evaluated_at) + + def _verified_claims(user: dict) -> tuple[str, str]: actor_ref = user.get("sub") product_id = user.get("product") @@ -185,6 +229,7 @@ async def query_intelligence_resource_page( "IntelligenceResourceHttpRuntime", "IntelligenceResourceHttpUnauthenticated", "IntelligenceResourceHttpUnavailable", + "CoreIntelligenceBuildResourcePagePort", "IntelligenceResourcePageV1Alpha1", "intelligence_resource_projection_reader", "intelligence_resource_runtime", diff --git a/core/ui/canvas/src/api/intelligenceBuildsApi.test.ts b/core/ui/canvas/src/api/intelligenceBuildsApi.test.ts index cb5ebae..00c6995 100644 --- a/core/ui/canvas/src/api/intelligenceBuildsApi.test.ts +++ b/core/ui/canvas/src/api/intelligenceBuildsApi.test.ts @@ -47,6 +47,13 @@ describe('startIntelligenceBuild', () => { outcome_id: 'outcome:decision-readiness', source_group_ids: ['sources:official', 'sources:independent'], cadence_id: 'cadence:daily', + resource_authority_grant_ref: 'authority_grant:atrium-observe-read', + approved_effects: [ + 'connect_sources', + 'map_concepts', + 'activate_watch', + 'create_first_brief', + ], })) expect(body.client_request_id).toMatch(/^atrium-request:/) expect(body.requested_at).toEqual(expect.any(String)) diff --git a/core/ui/canvas/src/api/intelligenceBuildsApi.ts b/core/ui/canvas/src/api/intelligenceBuildsApi.ts index 507b9dd..369e5df 100644 --- a/core/ui/canvas/src/api/intelligenceBuildsApi.ts +++ b/core/ui/canvas/src/api/intelligenceBuildsApi.ts @@ -5,6 +5,15 @@ const BASE = import.meta.env.VITE_API_BASE_URL ?? '' const BUILD_AUTHORITY_GRANT_REF = import.meta.env.VITE_INTELLIGENCE_BUILD_AUTHORITY_GRANT_REF ?? 'authority_grant:atrium-intelligence-build' +const RESOURCE_AUTHORITY_GRANT_REF = + import.meta.env.VITE_INTELLIGENCE_RESOURCE_AUTHORITY_GRANT_REF ?? + 'authority_grant:atrium-observe-read' +const APPROVED_ONBOARDING_EFFECTS = [ + 'connect_sources', + 'map_concepts', + 'activate_watch', + 'create_first_brief', +] as const export interface IntelligenceBuildStartInput { readonly profile_id: string @@ -48,6 +57,8 @@ async function postBuild(token: string, input: IntelligenceBuildStartInput): Pro }, body: JSON.stringify({ authority_grant_ref: BUILD_AUTHORITY_GRANT_REF, + resource_authority_grant_ref: RESOURCE_AUTHORITY_GRANT_REF, + approved_effects: APPROVED_ONBOARDING_EFFECTS, client_request_id: requestId(), ...input, requested_at: new Date().toISOString(), diff --git a/tests/test_api_intelligence_builds.py b/tests/test_api_intelligence_builds.py index 8656fb8..ae87ec3 100644 --- a/tests/test_api_intelligence_builds.py +++ b/tests/test_api_intelligence_builds.py @@ -9,7 +9,7 @@ from ace.core import GovernedStateHeadPreconditionV1Alpha1 from ace.core.runtime_use import AuthorityUseReceiptV1Alpha1 from ace.intelligence.contracts.resource_plane import ( - IntelligenceResourcePageState, + IntelligenceResourceKind, IntelligenceResourcePageV1Alpha1, ) from ace.testing import InMemoryImmutableRecordStore @@ -86,22 +86,19 @@ async def resolve_authority_use(self, **kwargs) -> AuthorityUseReceiptV1Alpha1: class _Executor: def __init__(self) -> None: self.builds: list[AuthorizedIntelligenceBuild] = [] + self.host_services = [] - async def start(self, build: AuthorizedIntelligenceBuild) -> IntelligenceResourcePageV1Alpha1: + async def start(self, build: AuthorizedIntelligenceBuild, host_services) -> IntelligenceResourcePageV1Alpha1: self.builds.append(build) - read_authority = _receipt(build=build, kwargs={}) - evaluated_at = read_authority.evaluated_at - return IntelligenceResourcePageV1Alpha1( - query_id=read_authority.use_subject_ref, - query_digest=read_authority.use_subject_digest, - product_id=PRODUCT, - actor_ref=ACTOR, + self.host_services.append(host_services) + + evaluated_at = build.authority_use.evaluated_at + return await host_services.resources.query( + resource_kinds=(IntelligenceResourceKind.BRIEF,), + subject_refs=(), as_of=evaluated_at, available_at=evaluated_at, evaluated_at=evaluated_at, - state=IntelligenceResourcePageState.COMPLETE, - items=(), - authority_use=read_authority, ) @@ -109,7 +106,7 @@ def _claims(*, authorities: list[str] | None = None) -> dict: return { "sub": ACTOR, "product": PRODUCT, - "authorities": ["intelligence_build"] if authorities is None else authorities, + "authorities": ["intelligence_build", "observe_read"] if authorities is None else authorities, "exp": (NOW + timedelta(hours=1)).timestamp(), } @@ -117,12 +114,19 @@ def _claims(*, authorities: list[str] | None = None) -> dict: def _body() -> dict: return { "authority_grant_ref": GRANT, + "resource_authority_grant_ref": "authority_grant:personal-intelligence-read", "client_request_id": "atrium-request:first-picture", "profile_id": "profile:world-ai", "subject": "Keep me ahead of meaningful changes in artificial intelligence.", "outcome_id": "outcome:decision-readiness", "source_group_ids": ["sources:official", "sources:independent"], "cadence_id": "cadence:daily", + "approved_effects": [ + "connect_sources", + "map_concepts", + "activate_watch", + "create_first_brief", + ], "requested_at": NOW.isoformat(), } @@ -158,6 +162,9 @@ async def test_start_build_authorizes_exact_reviewed_plan_and_returns_resource_p assert executor.builds[0].request.source_group_ids == ("sources:official", "sources:independent") assert authority.calls[0]["authority"] == "intelligence_build" assert authority.calls[0]["operation"] == "start_intelligence_build" + assert authority.calls[1]["authority"] == "observe_read" + assert authority.calls[1]["grant_ref"] == _body()["resource_authority_grant_ref"] + assert executor.host_services[0].records.product_id == PRODUCT assert any(record.record_kind == "task_authentication" for record in records.records.values()) diff --git a/tests/test_intelligence_build_executor_registry.py b/tests/test_intelligence_build_executor_registry.py new file mode 100644 index 0000000..7ea88f1 --- /dev/null +++ b/tests/test_intelligence_build_executor_registry.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +from dataclasses import dataclass +from types import SimpleNamespace + +import pytest + +from core.engine.core.intelligence_build import _InstalledIntelligenceBuildExecutor +from core.engine.core.intelligence_build_executor_registry import ( + IntelligenceBuildExecutorRegistryError, + _reset_intelligence_build_executor_registry_for_tests, + load_installed_intelligence_build_executors, + register_intelligence_build_executor, + resolve_intelligence_build_executor, +) + +pytestmark = pytest.mark.unit + + +class _Executor: + profile_id = "intelligence_onboarding_profile:fixture" + + async def start(self, build, host_services): + del host_services + raise AssertionError("not executed by registry tests") + + +@dataclass(frozen=True) +class _EntryPoint: + name: str + value: object + + def load(self): + return self.value + + +@pytest.fixture(autouse=True) +def _clean_registry(monkeypatch): + monkeypatch.delenv("ACE_DISABLE_EXTENSIONS", raising=False) + _reset_intelligence_build_executor_registry_for_tests() + yield + _reset_intelligence_build_executor_registry_for_tests() + + +def test_dedicated_entry_point_loads_and_resolves_exact_profile() -> None: + assert load_installed_intelligence_build_executors((_EntryPoint("fixture", _Executor),)) == (_Executor.profile_id,) + assert isinstance(resolve_intelligence_build_executor(_Executor.profile_id), _Executor) + assert resolve_intelligence_build_executor("intelligence_onboarding_profile:unknown") is None + + +def test_duplicate_profile_claim_fails_closed() -> None: + register_intelligence_build_executor(profile_id=_Executor.profile_id, executor=_Executor()) + with pytest.raises(IntelligenceBuildExecutorRegistryError, match="multiple"): + register_intelligence_build_executor(profile_id=_Executor.profile_id, executor=_Executor()) + + +def test_duplicate_installed_claim_remains_failed_closed() -> None: + entries = (_EntryPoint("one", _Executor), _EntryPoint("two", _Executor)) + with pytest.raises(IntelligenceBuildExecutorRegistryError, match="multiple"): + load_installed_intelligence_build_executors(entries) + with pytest.raises(IntelligenceBuildExecutorRegistryError, match="multiple"): + resolve_intelligence_build_executor(_Executor.profile_id) + + +@pytest.mark.parametrize("profile_id", ["", "has spaces", "../escape", "x" * 241]) +def test_invalid_profile_identity_is_rejected(profile_id: str) -> None: + with pytest.raises(IntelligenceBuildExecutorRegistryError, match="invalid"): + register_intelligence_build_executor(profile_id=profile_id, executor=_Executor()) + + +def test_naked_kernel_does_not_load_installed_executors(monkeypatch) -> None: + monkeypatch.setenv("ACE_DISABLE_EXTENSIONS", "1") + assert load_installed_intelligence_build_executors((_EntryPoint("fixture", _Executor),)) == () + assert resolve_intelligence_build_executor(_Executor.profile_id) is None + + +def test_sync_executor_is_rejected() -> None: + class _SyncExecutor: + def start(self, build): + return build + + with pytest.raises(IntelligenceBuildExecutorRegistryError, match="async start"): + register_intelligence_build_executor(profile_id=_Executor.profile_id, executor=_SyncExecutor()) + + +@pytest.mark.anyio +async def test_installed_dispatcher_selects_only_the_authorized_profile(monkeypatch) -> None: + marker = object() + + class _ReturningExecutor: + async def start(self, build, host_services): + assert build.request.profile_id == _Executor.profile_id + assert host_services is marker + return marker + + monkeypatch.setattr( + "core.engine.core.intelligence_build.resolve_intelligence_build_executor", + lambda profile_id: _ReturningExecutor() if profile_id == _Executor.profile_id else None, + ) + build = SimpleNamespace(request=SimpleNamespace(profile_id=_Executor.profile_id)) + assert await _InstalledIntelligenceBuildExecutor().start(build, marker) is marker diff --git a/tests/test_intelligence_build_host_services.py b/tests/test_intelligence_build_host_services.py new file mode 100644 index 0000000..ef65a73 --- /dev/null +++ b/tests/test_intelligence_build_host_services.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from ace.application.intelligence_build_execution import ( + REQUIRED_INTELLIGENCE_BUILD_EFFECTS, + IntelligenceBuildStartV1, + ProductScopedImmutableRecordStore, +) +from ace.core import AppendOnlyTransactionRequestV1, ImmutableRecordScopeError, ImmutableRecordV1 +from ace.testing import InMemoryImmutableRecordStore + +pytestmark = pytest.mark.unit + +NOW = datetime(2026, 8, 13, 22, 0, tzinfo=UTC) +PRODUCT = "product:personal" + + +def _request(**changes) -> IntelligenceBuildStartV1: + material = { + "authority_grant_ref": "authority_grant:atrium-intelligence-build", + "resource_authority_grant_ref": "authority_grant:atrium-observe-read", + "client_request_id": "atrium-request:host-services", + "profile_id": "intelligence_onboarding_profile:fixture", + "subject": "Track the reviewed subject for meaningful material change.", + "outcome_id": "decision_readiness", + "source_group_ids": ("official_records",), + "cadence_id": "daily_pulse", + "approved_effects": REQUIRED_INTELLIGENCE_BUILD_EFFECTS, + "requested_at": NOW, + } + material.update(changes) + return IntelligenceBuildStartV1(**material) + + +def test_reviewed_build_requires_the_exact_bounded_internal_effects() -> None: + assert _request().approved_effects == REQUIRED_INTELLIGENCE_BUILD_EFFECTS + with pytest.raises(ValidationError, match="exact bounded onboarding effect sequence"): + _request(approved_effects=("connect_sources",)) + with pytest.raises(ValidationError): + _request(approved_effects=(*REQUIRED_INTELLIGENCE_BUILD_EFFECTS, "publish_content")) + + +@pytest.mark.asyncio +async def test_invocation_store_enforces_the_authorized_product_fence() -> None: + backing = InMemoryImmutableRecordStore() + scoped = ProductScopedImmutableRecordStore(product_id=PRODUCT, store=backing) + record = ImmutableRecordV1( + product_id=PRODUCT, + record_space="fixture", + record_kind="brief", + record_key="brief:one", + payload_contract="fixture.brief/v1", + payload={"title": "One"}, + as_of=NOW, + available_at=NOW, + processing_order=0, + ) + request = AppendOnlyTransactionRequestV1( + product_id=PRODUCT, + record_space="fixture", + transaction_key="fixture:one", + records=(record,), + submitted_at=NOW, + ) + assert await scoped.append(request) == request.receipt() + assert await scoped.scan_product_records(product_id=PRODUCT) == (record,) + + with pytest.raises(ImmutableRecordScopeError, match="authorized product"): + await scoped.scan_product_records(product_id="product:other") diff --git a/tests/test_public_core_boundaries.py b/tests/test_public_core_boundaries.py index 56bf801..06f7f02 100644 --- a/tests/test_public_core_boundaries.py +++ b/tests/test_public_core_boundaries.py @@ -109,6 +109,7 @@ def test_host_adapters_are_the_only_core_engine_edge_into_public_ace() -> None: "core/engine/core/intelligence_resource_plane.py", "core/engine/core/intelligence_build.py", "core/engine/core/installed_intelligence_catalog.py", + "core/engine/core/intelligence_build_executor_registry.py", "core/engine/core/live_cognition.py", "core/engine/core/action_execution.py", "core/engine/core/agent_composition_runtime.py",