diff --git a/ace/application/intelligence_build_execution.py b/ace/application/intelligence_build_execution.py index ddefb1d..7500462 100644 --- a/ace/application/intelligence_build_execution.py +++ b/ace/application/intelligence_build_execution.py @@ -34,6 +34,7 @@ PreparedShiftSignalDerivationRequestV1Alpha1, ) from ace.application.recorded_source_admission import RecordedSourceAdmission, RecordedSourceMaterialV1Alpha1 + from ace.intelligence.contracts.source_mapping import ResolvedSubjectBindingV1Alpha1 IntelligenceBuildEffect = Literal[ "connect_sources", @@ -214,6 +215,14 @@ async def query( class IntelligenceBuildRecordedSourcePort(Protocol): """Narrow host capability for the exact recorded material set in one build.""" + def bind_subject( + self, + *, + subject_binding_id: str, + entity_type_id: str, + entity_ref: str, + ) -> "ResolvedSubjectBindingV1Alpha1": ... + async def admit( self, materials: tuple["RecordedSourceMaterialV1Alpha1", ...], diff --git a/ace/application/intelligence_build_host.py b/ace/application/intelligence_build_host.py new file mode 100644 index 0000000..8e19399 --- /dev/null +++ b/ace/application/intelligence_build_host.py @@ -0,0 +1,366 @@ +"""Durable invocation-scoped host capabilities for one authorized Intelligence build.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime + +from ace.application.domain_activation import ( + CommittedDomainActivation, + DomainActivationAdmissionService, + bind_committed_activation, +) +from ace.application.installed_pack_artifacts import InstalledPackArtifactError +from ace.application.intelligence_build_execution import ( + AuthorizedIntelligenceBuild, + IntelligenceBuildHostServices, + IntelligenceBuildResourcePagePort, +) +from ace.application.intelligence_builder import ( + INTELLIGENCE_BUILDER_RECORD_SPACE, + ONBOARDING_ARTIFACT_RECORD_KIND, + IntelligenceBuilderSessionError, + IntelligenceBuilderSessionService, +) +from ace.application.intelligence_builder_activation import ExactCompiledPackResolver +from ace.application.intelligence_builder_activation_contracts import ( + BUILDER_ACTIVATION_PLAN_ARTIFACT_VERSION, + BUILDER_ACTIVATION_RECEIPT_ARTIFACT_VERSION, + BuilderActivationPlanArtifactV1, + BuilderActivationReceiptArtifactV1, +) +from ace.application.intelligence_builder_contracts import ( + OnboardingArtifactKind, + OnboardingArtifactReferenceV1, + OnboardingStage, +) +from ace.application.intelligence_ledger import PreparedIntelligenceLedgerService +from ace.application.prepared_shift_signal import CorePreparedShiftSignalDerivationService +from ace.application.recorded_source_admission import CoreRecordedSourceAdmissionService +from ace.core import CoreAuthorityResolver, ImmutableRecordStore, RuntimeUseResolver +from ace.core.state import GovernedStateStore +from ace.intelligence.contracts.resources import ActivationRevisionReferenceV1Alpha1 + + +class IntelligenceBuildHostCompositionError(RuntimeError): + """Durable activation/bootstrap material could not compose safe build ports.""" + + +@dataclass(frozen=True, slots=True) +class _BootstrapCandidate: + plan_reference: OnboardingArtifactReferenceV1 + receipt_reference: OnboardingArtifactReferenceV1 + plan: BuilderActivationPlanArtifactV1 + receipt: BuilderActivationReceiptArtifactV1 + + +def _artifact_reference( + kind: OnboardingArtifactKind, + artifact_id: str | None, + artifact_digest: str | None, +) -> OnboardingArtifactReferenceV1: + if artifact_id is None or artifact_digest is None: + raise IntelligenceBuildHostCompositionError("Builder activation artifact is missing exact identity") + return OnboardingArtifactReferenceV1( + artifact_kind=kind, + artifact_id=artifact_id, + artifact_digest=artifact_digest, + ) + + +def _record_matches_artifact(record, artifact, *, timestamp: datetime) -> bool: + return ( + record.record_key == artifact.artifact_id + and record.payload_contract == artifact.contract + and record.as_of == timestamp + and record.available_at == timestamp + ) + + +class DurableIntelligenceBuildHostComposer: + """Recover exact Builder activation material and grant only bounded build ports.""" + + def __init__( + self, + *, + governed_state: GovernedStateStore, + runtime_use: RuntimeUseResolver, + packs: ExactCompiledPackResolver, + ) -> None: + self.governed_state = governed_state + self.runtime_use = runtime_use + self.packs = packs + + async def _matching_candidates( + self, + *, + build: AuthorizedIntelligenceBuild, + records: ImmutableRecordStore, + ) -> tuple[_BootstrapCandidate, ...]: + evaluated_at = build.authority_use.evaluated_at + try: + artifacts = await records.read_as_of( + product_id=build.product_id, + record_space=INTELLIGENCE_BUILDER_RECORD_SPACE, + record_kind=ONBOARDING_ARTIFACT_RECORD_KIND, + available_at=evaluated_at, + ) + except Exception: + raise IntelligenceBuildHostCompositionError( + "durable Builder activation artifacts are unavailable" + ) from None + + plans: list[tuple[OnboardingArtifactReferenceV1, BuilderActivationPlanArtifactV1]] = [] + correlated_plan_ids: set[str] = set() + for record in artifacts: + if record.payload_contract != BUILDER_ACTIVATION_PLAN_ARTIFACT_VERSION: + continue + raw_spec = record.payload.get("spec_id") if isinstance(record.payload, dict) else None + try: + plan = BuilderActivationPlanArtifactV1.model_validate(record.payload, strict=False) + except Exception: + if raw_spec == build.request.activation_approval_subject_ref: + raise IntelligenceBuildHostCompositionError( + "correlated Builder activation plan failed exact revalidation" + ) from None + continue + if plan.spec_id != build.request.activation_approval_subject_ref: + continue + if plan.source_commit.product_id != build.product_id or not _record_matches_artifact( + record, + plan, + timestamp=plan.created_at, + ): + raise IntelligenceBuildHostCompositionError( + "correlated Builder activation plan crossed exact product or persistence material" + ) + reference = _artifact_reference( + OnboardingArtifactKind.ACTIVATION_PLAN, + plan.artifact_id, + plan.artifact_digest, + ) + plans.append((reference, plan)) + correlated_plan_ids.add(reference.artifact_id) + if not plans: + return () + + receipts: list[tuple[OnboardingArtifactReferenceV1, BuilderActivationReceiptArtifactV1]] = [] + for record in artifacts: + if record.payload_contract != BUILDER_ACTIVATION_RECEIPT_ARTIFACT_VERSION: + continue + raw_plan_id = ( + record.payload.get("activation_plan_artifact_id") if isinstance(record.payload, dict) else None + ) + try: + receipt = BuilderActivationReceiptArtifactV1.model_validate(record.payload, strict=False) + except Exception: + if raw_plan_id in correlated_plan_ids: + raise IntelligenceBuildHostCompositionError( + "correlated Builder activation receipt failed exact revalidation" + ) from None + continue + if receipt.activation_plan_artifact_id not in correlated_plan_ids: + continue + if receipt.canonical_revision.product_id != build.product_id or not _record_matches_artifact( + record, + receipt, + timestamp=receipt.activated_at, + ): + raise IntelligenceBuildHostCompositionError( + "correlated Builder activation receipt crossed exact product or persistence material" + ) + receipts.append( + ( + _artifact_reference( + OnboardingArtifactKind.ACTIVATION_RECEIPT, + receipt.artifact_id, + receipt.artifact_digest, + ), + receipt, + ) + ) + + sessions = IntelligenceBuilderSessionService(store=records) + candidates: list[_BootstrapCandidate] = [] + for plan_reference, plan in plans: + matching_receipts = [ + (reference, receipt) + for reference, receipt in receipts + if receipt.activation_plan_artifact_id == plan_reference.artifact_id + and receipt.activation_plan_artifact_digest == plan_reference.artifact_digest + and receipt.source_commit == plan.source_commit + and receipt.session_id == plan.session_id + ] + for receipt_reference, receipt in matching_receipts: + try: + session = await sessions.load_latest( + product_id=build.product_id, + session_id=receipt.session_id, + available_at=evaluated_at, + ) + except IntelligenceBuilderSessionError: + raise IntelligenceBuildHostCompositionError( + "correlated active Builder session failed exact durable replay" + ) from None + if session is None or session.stage is not OnboardingStage.ACTIVE: + continue + if ( + session.approval_receipt_ref != build.request.activation_approval_receipt_ref + or plan_reference not in session.artifacts + or receipt_reference not in session.artifacts + ): + raise IntelligenceBuildHostCompositionError( + "correlated active Builder session changed approval or activation artifacts" + ) + candidates.append( + _BootstrapCandidate( + plan_reference=plan_reference, + receipt_reference=receipt_reference, + plan=plan, + receipt=receipt, + ) + ) + return tuple(candidates) + + async def _binding( + self, + *, + build: AuthorizedIntelligenceBuild, + records: ImmutableRecordStore, + activation_authority: CoreAuthorityResolver, + ): + candidates = await self._matching_candidates(build=build, records=records) + if not candidates: + return None + if len(candidates) != 1: + raise IntelligenceBuildHostCompositionError( + "authorized build resolves more than one exact active Builder bootstrap" + ) + candidate = candidates[0] + sessions = IntelligenceBuilderSessionService(store=records) + try: + plan = await sessions.load_artifact( + product_id=build.product_id, + reference=candidate.plan_reference, + artifact_type=BuilderActivationPlanArtifactV1, + available_at=build.authority_use.evaluated_at, + ) + receipt = await sessions.load_artifact( + product_id=build.product_id, + reference=candidate.receipt_reference, + artifact_type=BuilderActivationReceiptArtifactV1, + available_at=build.authority_use.evaluated_at, + ) + except IntelligenceBuilderSessionError: + raise IntelligenceBuildHostCompositionError( + "authorized Builder bootstrap artifacts failed exact reload" + ) from None + if plan != candidate.plan or receipt != candidate.receipt: + raise IntelligenceBuildHostCompositionError("Builder bootstrap changed during exact reload") + + canonical = DomainActivationAdmissionService( + store=self.governed_state, + authority=activation_authority, + ) + try: + committed = await canonical.load_exact( + product_id=build.product_id, + revision_id=receipt.canonical_revision.revision_id, + commit_receipt_id=receipt.canonical_commit_receipt_id, + ) + except Exception: + raise IntelligenceBuildHostCompositionError( + "Builder bootstrap canonical activation failed exact durable reload" + ) from None + if committed is None: + return None + current = await canonical.reload( + product_id=build.product_id, + activation_key=receipt.canonical_revision.activation_key, + ) + if current is None or current != committed: + return None + revision = committed.revision + canonical_reference = ActivationRevisionReferenceV1Alpha1( + product_id=build.product_id, + activation_key=revision.spec.activation_key, + activation_id=str(revision.activation_id), + revision=revision.revision, + revision_id=str(revision.revision_id), + revision_digest=f"sha256:{revision.revision_hash}", + ) + if ( + receipt.canonical_revision != canonical_reference + or receipt.canonical_state_kind != committed.commit_receipt.state_kind + or receipt.canonical_commit_receipt_digest != f"sha256:{committed.commit_receipt.receipt_hash}" + or committed.commit_receipt.approval != build.activation_approval + or committed.commit_receipt.actor_ref != build.actor_ref + or revision.spec.spec_id != build.request.activation_approval_subject_ref + or revision.spec.pack != plan.pack + ): + raise IntelligenceBuildHostCompositionError( + "authorized build approval, plan, receipt, and canonical activation do not exactly agree" + ) + try: + pack = await self.packs.load_exact(reference=plan.pack) + except InstalledPackArtifactError: + raise IntelligenceBuildHostCompositionError("installed Pack artifact failed exact resolution") from None + if pack is None: + return None + try: + return bind_committed_activation( + pack=pack, + committed=CommittedDomainActivation( + revision=revision, + commit_receipt=committed.commit_receipt, + ), + ) + except Exception: + raise IntelligenceBuildHostCompositionError( + "installed Pack and canonical activation failed exact binding" + ) from None + + async def compose( + self, + *, + build: AuthorizedIntelligenceBuild, + records: ImmutableRecordStore, + resources: IntelligenceBuildResourcePagePort, + activation_authority: CoreAuthorityResolver, + ) -> IntelligenceBuildHostServices: + """Create per-invocation ports, or keep both unavailable without exact bootstrap.""" + + binding = await self._binding( + build=build, + records=records, + activation_authority=activation_authority, + ) + if binding is None: + return IntelligenceBuildHostServices( + records=records, + resources=resources, + activation_authority=activation_authority, + ) + return IntelligenceBuildHostServices( + records=records, + resources=resources, + activation_authority=activation_authority, + recorded_sources=CoreRecordedSourceAdmissionService( + build=build, + binding=binding, + store=records, + ), + prepared_derivations=CorePreparedShiftSignalDerivationService( + build=build, + binding=binding, + ledger=PreparedIntelligenceLedgerService(binding=binding, store=records), + governed_state=self.governed_state, + runtime_use=self.runtime_use, + ), + ) + + +__all__ = [ + "DurableIntelligenceBuildHostComposer", + "IntelligenceBuildHostCompositionError", +] diff --git a/ace/application/recorded_source_admission.py b/ace/application/recorded_source_admission.py index 5b338f5..8bb2dc6 100644 --- a/ace/application/recorded_source_admission.py +++ b/ace/application/recorded_source_admission.py @@ -52,7 +52,12 @@ IntelligenceResourceMode, ObservationV1Alpha1, ) -from ace.intelligence.contracts.source_mapping import ResolvedSubjectBindingV1Alpha1 +from ace.intelligence.contracts.source_mapping import ( + SOURCE_MAPPING_MODULE_VERSION, + ResolvedSubjectBindingV1Alpha1, + SourceMappingModuleV1, +) +from ace.intelligence.packs.runtime import resolve_entity_type_declaration from ace.intelligence.source_mapping import interpret_prepared_source_mapping RECORDED_SOURCE_MATERIAL_VERSION = "ace.application.recorded-source-material/v1alpha1" @@ -380,6 +385,46 @@ def _validate_build(self) -> None: if self.binding.prepared_binding.reference.product_id != self.build.product_id: raise RecordedSourceAdmissionError("committed activation crossed the authorized build product") + def bind_subject( + self, + *, + subject_binding_id: str, + entity_type_id: str, + entity_ref: str, + ) -> ResolvedSubjectBindingV1Alpha1: + """Resolve one declared subject identity without exposing activation inputs.""" + + try: + declared_types = { + mapping.entity_type_id + for module_ir in self.binding.prepared_binding.pack.modules + if module_ir.contract == SOURCE_MAPPING_MODULE_VERSION + for mapping in SourceMappingModuleV1.model_validate_json(module_ir.canonical_payload).mappings + if mapping.subject_binding_id == subject_binding_id + } + if len(declared_types) != 1 or entity_type_id not in declared_types: + raise RecordedSourceAdmissionError( + "subject binding and entity type must resolve exactly once in the committed Pack" + ) + resolved_entity = resolve_entity_type_declaration( + self.binding.prepared_binding, + entity_type_id=entity_type_id, + ) + if resolved_entity.entity_type_id != entity_type_id: + raise RecordedSourceAdmissionError("subject binding resolved a different Pack entity type") + return ResolvedSubjectBindingV1Alpha1( + product_id=self.build.product_id, + mode=IntelligenceResourceMode.PREPARED, + activation_revision=self.binding.prepared_binding.reference, + subject_binding_id=subject_binding_id, + entity_type_id=entity_type_id, + entity_ref=entity_ref, + ) + except RecordedSourceAdmissionError: + raise + except Exception: + raise RecordedSourceAdmissionError("subject binding failed exact committed Pack resolution") from None + def _materials( self, materials: tuple[RecordedSourceMaterialV1Alpha1, ...], diff --git a/core/engine/core/intelligence_build.py b/core/engine/core/intelligence_build.py index d3a8f47..677dec8 100644 --- a/core/engine/core/intelligence_build.py +++ b/core/engine/core/intelligence_build.py @@ -13,14 +13,16 @@ from pydantic import BaseModel, ConfigDict, ValidationError -from ace.application import IntelligenceResourcePageV1Alpha1 +from ace.application import InstalledCompiledPackArtifactResolver, IntelligenceResourcePageV1Alpha1 from ace.application.intelligence_build_execution import ( AuthorizedIntelligenceBuild, IntelligenceBuildExecutor, IntelligenceBuildHostServices, + IntelligenceBuildResourcePagePort, IntelligenceBuildStartV1, ProductScopedImmutableRecordStore, ) +from ace.application.intelligence_build_host import DurableIntelligenceBuildHostComposer from ace.core import CoreAuthorityResolver, ImmutableRecordStore, ResolvedApprovalReceiptV1 from ace.core.contracts import canonical_hash from ace.core.runtime_use import AuthorityUseReceiptV1Alpha1 @@ -72,12 +74,24 @@ async def resolve_authority_use( ) -> AuthorityUseReceiptV1Alpha1: ... +class IntelligenceBuildHostCompositionPort(Protocol): + async def compose( + self, + *, + build: AuthorizedIntelligenceBuild, + records: ImmutableRecordStore, + resources: IntelligenceBuildResourcePagePort, + activation_authority: CoreAuthorityResolver, + ) -> IntelligenceBuildHostServices: ... + + @dataclass(frozen=True, slots=True) class IntelligenceBuildHttpRuntime: records: ImmutableRecordStore authority: IntelligenceBuildAuthorizationPort activation_authority: CoreAuthorityResolver executor: IntelligenceBuildExecutor + host_composer: IntelligenceBuildHostCompositionPort | None = None class IntelligenceBuildError(RuntimeError): @@ -126,11 +140,17 @@ async def start( def intelligence_build_runtime() -> IntelligenceBuildHttpRuntime: records = SurrealImmutableRecordStore(pool) governed_state = SurrealGovernedStateStore(pool) + authority = GovernedStateRuntimeUseResolver(governed_state=governed_state) return IntelligenceBuildHttpRuntime( records=records, - authority=GovernedStateRuntimeUseResolver(governed_state=governed_state), + authority=authority, activation_authority=_UnavailableActivationAuthority(), executor=_InstalledIntelligenceBuildExecutor(), + host_composer=DurableIntelligenceBuildHostComposer( + governed_state=governed_state, + runtime_use=authority, + packs=InstalledCompiledPackArtifactResolver.discover(), + ), ) @@ -241,15 +261,23 @@ async def start_intelligence_build( activation_approval=activation_approval, ) scoped_records = ProductScopedImmutableRecordStore(product_id=product_id, store=runtime.records) + resources = CoreIntelligenceBuildResourcePagePort( + build=authorized_build, + records=scoped_records, + authority=runtime.authority, + ) host_services = IntelligenceBuildHostServices( records=scoped_records, - resources=CoreIntelligenceBuildResourcePagePort( - build=authorized_build, - records=scoped_records, - authority=runtime.authority, - ), + resources=resources, activation_authority=runtime.activation_authority, ) + if runtime.host_composer is not None: + host_services = await runtime.host_composer.compose( + build=authorized_build, + records=scoped_records, + resources=resources, + activation_authority=runtime.activation_authority, + ) page = IntelligenceResourcePageV1Alpha1.model_validate( (await runtime.executor.start(authorized_build, host_services)).model_dump(mode="python") ) @@ -283,6 +311,7 @@ async def start_intelligence_build( "IntelligenceBuildDenied", "IntelligenceBuildExecutor", "IntelligenceBuildHttpRuntime", + "IntelligenceBuildHostCompositionPort", "IntelligenceBuildResultV1", "IntelligenceBuildStartV1", "IntelligenceBuildUnauthenticated", diff --git a/docs/evidence/v1-durable-intelligence-build-host.md b/docs/evidence/v1-durable-intelligence-build-host.md new file mode 100644 index 0000000..be1e79d --- /dev/null +++ b/docs/evidence/v1-durable-intelligence-build-host.md @@ -0,0 +1,58 @@ +# Durable Intelligence build host — v1 acceptance evidence + +## Scope + +This packet closes one bounded Core composition gap on base +`7e75feea1c1f757a8a32f6c729e486ec2e933e2f`. An authorized Intelligence +build can now recover its exact active Builder bootstrap from the same +product-scoped durable record store, reload the exact canonical activation, +resolve the exact installed compiled Pack, and receive fresh invocation-scoped +recorded-source and prepared-derivation ports. + +The recorded-source port also resolves a requested subject binding only when +the binding and entity type are declared by the exact committed Pack. The +caller supplies no activation coordinates and receives no raw activation or +binding object. + +## Fail-closed boundary + +- No matching active Builder bootstrap or no exact installed Pack keeps both + optional ports unavailable. +- Malformed correlated activation material, more than one exact candidate, + changed approval/session material, product drift, or Pack/activation drift + fails the composition request explicitly. +- The host reuses the authorized build's product-fenced immutable-record port, + existing activation authority, and current runtime-use resolver. It does not + create a global setter, cached authority decision, alternate store, or + fixture identity. +- Composition performs no new approval or grant resolution. Prepared + derivation retains its existing requirement to resolve the current build + grant when derivation is invoked. + +## Durable restart proof + +`tests/intelligence/test_intelligence_build_host_restart.py` starts a +disposable SurrealKV service, applies the supported schema, persists the full +Watch/Brief-to-active-Builder chain and canonical activation, and discovers the +matching compiled Pack from an inert installed distribution. It then composes +both ports, stops the database process, reopens the same database in a fresh +runtime, rediscovers the Pack, and recomposes distinct port instances with the +same exact activation binding and a fresh runtime-use resolver. The proof also +checks same-store reuse, product fencing, and absence of additional approval or +grant calls. + +## Verification + +- Focused build-host, subject binding, prepared derivation, HTTP boundary, and + host-service tests: **24 passed**. +- Public Core, naked-kernel, exact-eleven, installed-Pack, executor registry, + package identity/policy, and Intelligence schema boundaries: **68 passed**. +- Disposable-Surreal process restart/reopen proof: **1 passed**. +- Ruff and `git diff --check`: passed. + +## Exclusions + +This packet does not implement Brief or cognition behavior, token scopes, +owner bootstrap, UI, domain logic, MCP, connectors, providers, collaboration, +release, tag, or publication work. It does not widen the eleven-tool public +surface and does not make external connector secrets durable or portable. diff --git a/tests/intelligence/test_intelligence_build_host_restart.py b/tests/intelligence/test_intelligence_build_host_restart.py new file mode 100644 index 0000000..676006c --- /dev/null +++ b/tests/intelligence/test_intelligence_build_host_restart.py @@ -0,0 +1,161 @@ +"""Real SurrealKV restart proof for durable Intelligence build host composition.""" + +from __future__ import annotations + +import asyncio +import json +import os +import shutil +import subprocess +import sys + +import pytest + +from ace.application.installed_pack_artifacts import InstalledCompiledPackArtifactResolver +from ace.application.intelligence_build_execution import ( + ImmutableRecordScopeError, + ProductScopedImmutableRecordStore, +) +from ace.application.intelligence_build_host import DurableIntelligenceBuildHostComposer +from core.engine.core.governed_state import SurrealGovernedStateStore +from core.engine.core.immutable_records import SurrealImmutableRecordStore +from tests.intelligence.test_builder_database_recovery import ( + ROOT, + _port, + _SingleConnectionPool, + _stop, + _surreal_process, + _wait_port, +) +from tests.intelligence.test_domain_activation_plan_admission import _pack_material +from tests.test_installed_pack_artifacts import _Distribution +from tests.test_intelligence_build_host_composition import ( + _activated_stack, + _build, + _Resources, + _RuntimeUse, +) + +pytestmark = pytest.mark.e2e + + +def _installed_packs(tmp_path): + manifest_document, modules, fixture = _pack_material() + manifest = json.loads(manifest_document) + pack_id = manifest["metadata"]["pack_id"] + root = f"domain_packs/{pack_id}" + resources = { + f"{root}/manifest.json": manifest_document, + **{f"{root}/{path}": payload for path, payload in modules.items()}, + f"{root}/conformance/activation_golden_fixture.json": fixture, + } + return InstalledCompiledPackArtifactResolver.discover( + [_Distribution(tmp_path / "installed-pack", "ace-test-domain", resources)] + ) + + +@pytest.mark.asyncio +async def test_durable_host_recomposes_exact_invocation_ports_after_service_restart(tmp_path): + surreal = os.environ.get("ACE_SURREAL_BIN") or shutil.which("surreal") + if not surreal: + pytest.skip("surreal binary is unavailable") + port = _port() + endpoint = f"ws://127.0.0.1:{port}" + database_store = tmp_path / "surrealkv" + log = (tmp_path / "surreal.log").open("wb") + process = _surreal_process(surreal, port, database_store, log) + pool: _SingleConnectionPool | None = None + try: + await _wait_port(port, process) + target = type("Target", (), {})() + target.endpoint = endpoint + target.namespace = "ace_build_host_restart" + target.database = "ace_build_host_restart" + target.username = "root" + target.password = "root" + env = os.environ | { + "SURREAL_URL": endpoint, + "SURREAL_NS": target.namespace, + "SURREAL_DB": target.database, + "SURREAL_USER": target.username, + "SURREAL_PASS": target.password, + "JWT_SECRET": "build-host-restart-fixture-secret-at-least-32-bytes", + "LLM_API_KEY": "sk-test-placeholder", + } + await asyncio.to_thread( + subprocess.run, + [sys.executable, "scripts/schema_apply.py"], + cwd=ROOT, + env=env, + capture_output=True, + text=True, + check=True, + ) + + pool = _SingleConnectionPool(target) + await pool.open() + records = SurrealImmutableRecordStore(pool) + governed = SurrealGovernedStateStore(pool) + _, _, _, authority, active = await _activated_stack(records=records, governed=governed) + build = _build(active) + runtime_use = _RuntimeUse() + installed = _installed_packs(tmp_path) + first = await DurableIntelligenceBuildHostComposer( + governed_state=governed, + runtime_use=runtime_use, + packs=installed, + ).compose( + build=build, + records=ProductScopedImmutableRecordStore(product_id=build.product_id, store=records), + resources=_Resources(), + activation_authority=authority, + ) + assert first.recorded_sources is not None + assert first.prepared_derivations is not None + first_binding = first.recorded_sources.binding + authority_calls = (len(authority.approvals), len(authority.grants)) + await pool.close() + pool = None + + await _stop(process) + process = _surreal_process(surreal, port, database_store, log) + await _wait_port(port, process) + + pool = _SingleConnectionPool(target) + await pool.open() + reopened_records = SurrealImmutableRecordStore(pool) + reopened_governed = SurrealGovernedStateStore(pool) + reopened_runtime_use = _RuntimeUse() + reopened = await DurableIntelligenceBuildHostComposer( + governed_state=reopened_governed, + runtime_use=reopened_runtime_use, + packs=_installed_packs(tmp_path), + ).compose( + build=build, + records=ProductScopedImmutableRecordStore( + product_id=build.product_id, + store=reopened_records, + ), + resources=_Resources(), + activation_authority=authority, + ) + + assert reopened.recorded_sources is not None + assert reopened.prepared_derivations is not None + assert reopened.recorded_sources is not first.recorded_sources + assert reopened.prepared_derivations is not first.prepared_derivations + assert reopened.recorded_sources.binding == first_binding + assert reopened.recorded_sources.store is reopened.records + assert reopened.prepared_derivations.ledger.store is reopened.records + assert reopened.prepared_derivations.runtime_use is reopened_runtime_use + assert reopened.activation_authority is authority + assert (len(authority.approvals), len(authority.grants)) == authority_calls + assert reopened_runtime_use.calls == [] + assert not hasattr(reopened, "binding") + with pytest.raises(ImmutableRecordScopeError): + await reopened.records.scan_product_records(product_id="product:other") + finally: + if pool is not None: + await pool.close() + await _stop(process) + log.close() diff --git a/tests/intelligence/test_recorded_source_admission.py b/tests/intelligence/test_recorded_source_admission.py index 3723ad7..07b84ad 100644 --- a/tests/intelligence/test_recorded_source_admission.py +++ b/tests/intelligence/test_recorded_source_admission.py @@ -341,6 +341,44 @@ async def test_recorded_observation_is_visible_from_fresh_canonical_resource_pro assert entity_page.records[0].reference.resource_kind is IntelligenceResourceKind.ENTITY +@pytest.mark.asyncio +async def test_subject_binding_is_resolved_only_from_the_exact_committed_pack() -> None: + binding, build, records, material = await _stack() + service = CoreRecordedSourceAdmissionService(build=build, binding=binding, store=records) + + subject = service.bind_subject( + subject_binding_id=material.subject_binding.subject_binding_id, + entity_type_id=material.subject_binding.entity_type_id, + entity_ref="entity:recorded-subject", + ) + + assert subject.product_id == build.product_id + assert subject.mode is IntelligenceResourceMode.PREPARED + assert subject.activation_revision == binding.prepared_binding.reference + assert subject.subject_binding_id == material.subject_binding.subject_binding_id + assert subject.entity_type_id == material.subject_binding.entity_type_id + assert subject.entity_ref == "entity:recorded-subject" + + with pytest.raises(RecordedSourceAdmissionError, match="resolve exactly once"): + service.bind_subject( + subject_binding_id="undeclared_subject", + entity_type_id=material.subject_binding.entity_type_id, + entity_ref="entity:recorded-subject", + ) + with pytest.raises(RecordedSourceAdmissionError, match="resolve exactly once"): + service.bind_subject( + subject_binding_id=material.subject_binding.subject_binding_id, + entity_type_id="undeclared_entity_type", + entity_ref="entity:recorded-subject", + ) + with pytest.raises(RecordedSourceAdmissionError, match="exact committed Pack resolution"): + service.bind_subject( + subject_binding_id=material.subject_binding.subject_binding_id, + entity_type_id=material.subject_binding.entity_type_id, + entity_ref="not a valid entity reference", + ) + + @pytest.mark.asyncio async def test_substituted_recorded_material_fails_before_any_write() -> None: binding, build, records, material = await _stack() diff --git a/tests/test_intelligence_build_host_composition.py b/tests/test_intelligence_build_host_composition.py new file mode 100644 index 0000000..db81365 --- /dev/null +++ b/tests/test_intelligence_build_host_composition.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +from datetime import timedelta + +import pytest + +from ace.application.domain_activation import DomainActivationAdmissionService +from ace.application.domain_activation_compatibility import DomainActivationCompatibilityService +from ace.application.domain_activation_plan import ( + DomainActivationPlanAdmissionService, + prepare_activation_onboarding_handoff, +) +from ace.application.domain_activation_plan_contracts import ActivationPlanAction +from ace.application.intelligence_build_execution import ( + REQUIRED_INTELLIGENCE_BUILD_EFFECTS, + AuthorizedIntelligenceBuild, + IntelligenceBuildStartV1, + ProductScopedImmutableRecordStore, +) +from ace.application.intelligence_build_host import ( + DurableIntelligenceBuildHostComposer, + IntelligenceBuildHostCompositionError, +) +from ace.application.intelligence_builder import IntelligenceBuilderSessionService +from ace.application.intelligence_builder_activation import IntelligenceBuilderActivationService +from ace.application.recorded_source_admission import CoreRecordedSourceAdmissionService +from ace.core import ( + AuthenticatedRuntimeContextV1Alpha1, + AuthorityUseReceiptV1Alpha1, + GovernedStateHeadPreconditionV1Alpha1, +) +from ace.testing import InMemoryImmutableRecordStore +from ace.testing.watch_brief import exercise_watch_brief_restart +from tests.intelligence.test_domain_activation_plan_admission import ( + _activation_material, + _Authority, + _MemoryStore, + _PackResolver, + _plan, + _revision, +) + +pytestmark = pytest.mark.unit + + +class _Resources: + async def query(self, **_kwargs): + raise AssertionError("host composition must not query resource projections") + + +class _RuntimeUse: + def __init__(self) -> None: + self.calls = [] + + async def resolve_authority_use(self, **kwargs): + self.calls.append(kwargs) + raise AssertionError("composition alone must not spend a second build authority use") + + +async def _activated_stack(*, records=None, governed=None): + records = records or InMemoryImmutableRecordStore() + watch = await exercise_watch_brief_restart(store=records) + session = watch.briefing.session.revision + handoff = prepare_activation_onboarding_handoff( + session=session, + observations=watch.observations.observation_set, + intelligence_model=watch.approved.proposal, + intelligence_disposition=watch.approved.disposition, + first_briefing=watch.briefing.brief, + ) + pack, conformance, spec = _activation_material(product_id=session.product_id) + created = session.occurred_at + timedelta(seconds=1) + plan = _plan( + spec=spec, + action=ActivationPlanAction.INITIAL_ACTIVATION, + created_at=created, + handoff=handoff, + ) + revision = _revision(plan=plan, revision=1, occurred_at=created + timedelta(seconds=2)) + authority = _Authority(approved_at=created + timedelta(seconds=1)) + governed = governed or _MemoryStore() + plans = DomainActivationPlanAdmissionService(store=governed, authority=authority) + committed_plan = await plans.admit( + revision, + pack=pack, + conformance_receipts=(conformance,), + committed_at=revision.occurred_at + timedelta(seconds=1), + session=session, + observations=watch.observations.observation_set, + intelligence_model=watch.approved.proposal, + intelligence_disposition=watch.approved.disposition, + first_briefing=watch.briefing.brief, + ) + activation = IntelligenceBuilderActivationService( + sessions=IntelligenceBuilderSessionService(store=records), + plans=plans, + compatibility=DomainActivationCompatibilityService(authority=authority), + canonical=DomainActivationAdmissionService(store=governed, authority=authority), + packs=_PackResolver(pack), + ) + recorded = await activation.record_current_plan( + product_id=session.product_id, + session_id=session.session_id, + committed=committed_plan, + pack=spec.pack, + recorded_at=revision.occurred_at + timedelta(seconds=2), + ) + active = await activation.activate( + product_id=session.product_id, + session_id=recorded.session.revision.session_id, + activation_approval_receipt_ref="approval:canonical-spec", + evaluated_at=revision.occurred_at + timedelta(seconds=3), + ) + return records, governed, pack, authority, active + + +def _build(active) -> AuthorizedIntelligenceBuild: + evaluated_at = active.receipt_artifact.activated_at + timedelta(minutes=1) + product_id = active.binding.prepared_binding.reference.product_id + actor_ref = active.binding.commit_receipt.actor_ref + context = AuthenticatedRuntimeContextV1Alpha1( + product_id=product_id, + actor_ref=actor_ref, + authentication_receipt_ref="authentication_receipt:build-host", + authentication_receipt_digest="sha256:" + "1" * 64, + authenticated_at=evaluated_at - timedelta(minutes=2), + expires_at=evaluated_at + timedelta(hours=1), + ) + request = IntelligenceBuildStartV1( + authority_grant_ref="authority_grant:build-host", + resource_authority_grant_ref="authority_grant:build-host-read", + activation_approval_receipt_ref=str(active.binding.commit_receipt.approval.receipt_ref), + activation_approval_subject_ref=str(active.binding.prepared_binding.revision.spec.spec_id), + client_request_id="atrium_request:build-host", + profile_id="intelligence_onboarding_profile:build-host", + subject="Track this exact activated subject for meaningful material changes.", + outcome_id="decision_readiness", + source_group_ids=(), + recorded_source_refs=(), + cadence_id="daily_pulse", + approved_effects=REQUIRED_INTELLIGENCE_BUILD_EFFECTS, + requested_at=evaluated_at - timedelta(seconds=1), + ) + build_id = "intelligence_build:durable-host" + request_digest = "sha256:" + "2" * 64 + authority_use = AuthorityUseReceiptV1Alpha1( + product_id=product_id, + actor_ref=actor_ref, + authenticated_context=context, + use_subject_ref=build_id, + use_subject_digest=request_digest, + operation="start_intelligence_build", + authority="intelligence_build", + grant_ref=request.authority_grant_ref, + grant_hash="3" * 64, + evaluated_at=evaluated_at, + expires_at=evaluated_at + timedelta(hours=1), + state_head_precondition=GovernedStateHeadPreconditionV1Alpha1( + state_kind="authority_grant", + product_id=product_id, + state_id=request.authority_grant_ref, + sequence=1, + revision_id="authority_grant_revision:build-host", + commit_receipt_id="governed_state_commit:build-host", + ), + ) + return AuthorizedIntelligenceBuild( + build_id=build_id, + request_digest=request_digest, + product_id=product_id, + actor_ref=actor_ref, + request=request, + authority_use=authority_use, + activation_approval=active.binding.commit_receipt.approval, + ) + + +@pytest.mark.asyncio +async def test_exact_durable_bootstrap_composes_fresh_product_fenced_ports() -> None: + records, governed, pack, authority, active = await _activated_stack() + build = _build(active) + runtime_use = _RuntimeUse() + composer = DurableIntelligenceBuildHostComposer( + governed_state=governed, + runtime_use=runtime_use, + packs=_PackResolver(pack), + ) + scoped = ProductScopedImmutableRecordStore(product_id=build.product_id, store=records) + + first = await composer.compose( + build=build, + records=scoped, + resources=_Resources(), + activation_authority=authority, + ) + reopened = await DurableIntelligenceBuildHostComposer( + governed_state=governed, + runtime_use=runtime_use, + packs=_PackResolver(pack), + ).compose( + build=build, + records=ProductScopedImmutableRecordStore(product_id=build.product_id, store=records), + resources=_Resources(), + activation_authority=authority, + ) + + assert isinstance(first.recorded_sources, CoreRecordedSourceAdmissionService) + assert first.prepared_derivations is not None + assert first.records.product_id == build.product_id + assert first.recorded_sources.store is first.records + assert first.prepared_derivations.ledger.store is first.records + assert first.prepared_derivations.runtime_use is runtime_use + assert first.activation_authority is authority + assert reopened.recorded_sources is not first.recorded_sources + assert reopened.prepared_derivations is not first.prepared_derivations + assert reopened.recorded_sources.binding == first.recorded_sources.binding == active.binding + assert runtime_use.calls == [] + + +@pytest.mark.asyncio +async def test_missing_or_unavailable_exact_bootstrap_keeps_ports_unavailable() -> None: + records, governed, pack, authority, active = await _activated_stack() + build = _build(active) + empty = ProductScopedImmutableRecordStore( + product_id=build.product_id, + store=InMemoryImmutableRecordStore(), + ) + missing_bootstrap = await DurableIntelligenceBuildHostComposer( + governed_state=governed, + runtime_use=_RuntimeUse(), + packs=_PackResolver(pack), + ).compose( + build=build, + records=empty, + resources=_Resources(), + activation_authority=authority, + ) + + class _MissingPack: + async def load_exact(self, *, reference): + return None + + missing_pack = await DurableIntelligenceBuildHostComposer( + governed_state=governed, + runtime_use=_RuntimeUse(), + packs=_MissingPack(), + ).compose( + build=build, + records=ProductScopedImmutableRecordStore(product_id=build.product_id, store=records), + resources=_Resources(), + activation_authority=authority, + ) + + for services in (missing_bootstrap, missing_pack): + assert services.recorded_sources is None + assert services.prepared_derivations is None + + +@pytest.mark.asyncio +async def test_correlated_malformed_or_ambiguous_bootstrap_fails_closed() -> None: + records, governed, pack, authority, active = await _activated_stack() + build = _build(active) + + class _DuplicateArtifacts: + def __init__(self, store): + self.store = store + + async def read_as_of(self, **kwargs): + result = await self.store.read_as_of(**kwargs) + if kwargs["record_kind"] == "onboarding_artifact": + return (*result, *result) + return result + + def __getattr__(self, name): + return getattr(self.store, name) + + composer = DurableIntelligenceBuildHostComposer( + governed_state=governed, + runtime_use=_RuntimeUse(), + packs=_PackResolver(pack), + ) + with pytest.raises(IntelligenceBuildHostCompositionError, match="more than one"): + await composer.compose( + build=build, + records=ProductScopedImmutableRecordStore( + product_id=build.product_id, + store=_DuplicateArtifacts(records), + ), + resources=_Resources(), + activation_authority=authority, + ) + + class _CorruptPlan: + def __init__(self, store): + self.store = store + + async def read_as_of(self, **kwargs): + result = await self.store.read_as_of(**kwargs) + if kwargs["record_kind"] != "onboarding_artifact": + return result + return tuple( + item.model_copy(update={"payload": {**item.payload, "artifact_digest": "sha256:" + "f" * 64}}) + if item.payload_contract == "ace.application.builder-activation-plan-artifact/v1alpha1" + else item + for item in result + ) + + def __getattr__(self, name): + return getattr(self.store, name) + + with pytest.raises(IntelligenceBuildHostCompositionError, match="correlated Builder activation plan"): + await composer.compose( + build=build, + records=ProductScopedImmutableRecordStore( + product_id=build.product_id, + store=_CorruptPlan(records), + ), + resources=_Resources(), + activation_authority=authority, + )