Discover trusted Intelligence build executors

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

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

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