From 2100bbf51a28676bca9d03be7e8c08fd41624b22 Mon Sep 17 00:00:00 2001 From: Edwin Amirian Date: Thu, 13 Aug 2026 17:18:25 -0700 Subject: [PATCH] Resolve exact installed domain packs --- ace/application/__init__.py | 10 + ace/application/installed_pack_artifacts.py | 345 ++++++++++++++++++ ace/intelligence/__init__.py | 3 + ace/intelligence/conformance.py | 298 +++++++++++++++ ace/testing/domain_pack.py | 297 +-------------- .../intelligence/test_contract_boundaries.py | 1 + tests/test_installed_pack_artifacts.py | 344 +++++++++++++++++ 7 files changed, 1003 insertions(+), 295 deletions(-) create mode 100644 ace/application/installed_pack_artifacts.py create mode 100644 ace/intelligence/conformance.py create mode 100644 tests/test_installed_pack_artifacts.py diff --git a/ace/application/__init__.py b/ace/application/__init__.py index 21396e8..1214dd6 100644 --- a/ace/application/__init__.py +++ b/ace/application/__init__.py @@ -272,6 +272,12 @@ delivery_payload_digest, export_manifest_checksum, ) +from ace.application.installed_pack_artifacts import ( + ACTIVATION_GOLDEN_FIXTURE_PATH, + InstalledCompiledPackArtifact, + InstalledCompiledPackArtifactResolver, + InstalledPackArtifactError, +) from ace.application.intelligence_agent import ( AuthorizedObservationSetAdmission, IntelligenceAgent, @@ -496,6 +502,10 @@ ) __all__ = [ + "ACTIVATION_GOLDEN_FIXTURE_PATH", + "InstalledCompiledPackArtifact", + "InstalledCompiledPackArtifactResolver", + "InstalledPackArtifactError", "PERSONAL_INTELLIGENCE_DELETION_PROOF_RECORD_KIND", "PERSONAL_INTELLIGENCE_OWNERSHIP_RECORD_SPACE", "PersonalIntelligenceDeletePreviewStale", diff --git a/ace/application/installed_pack_artifacts.py b/ace/application/installed_pack_artifacts.py new file mode 100644 index 0000000..52b82e2 --- /dev/null +++ b/ace/application/installed_pack_artifacts.py @@ -0,0 +1,345 @@ +"""Discover and verify installed inert Domain Pack artifacts without imports. + +The resolver reads only distribution-declared package data. It never imports a +Domain Pack, loads an entry point, executes pack code, or treats a wheel version +as the Pack version. Every result is recompiled and re-conformed from bytes. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from importlib import metadata +from pathlib import Path, PurePosixPath +from typing import Iterable, Protocol + +from pydantic import ValidationError + +from ace.intelligence.conformance import run_domain_pack_conformance +from ace.intelligence.contracts.activation import CompiledPackRefV1 +from ace.intelligence.contracts.conformance import DomainPackConformanceReceiptV1 +from ace.intelligence.contracts.pack import CompiledDomainPackV1, DomainPackManifestV1 +from ace.intelligence.packs.compiler import ( + CompiledPackResultV1, + PackCompilationError, + compile_pack_document_with_report, + validate_compiled_pack_set, +) +from ace.intelligence.packs.diagnostics import StablePackCompilationResultV1 + +ACTIVATION_GOLDEN_FIXTURE_PATH = PurePosixPath("conformance/activation_golden_fixture.json") +MAX_MANIFEST_BYTES = 1_000_000 + + +class InstalledPackArtifactError(RuntimeError): + """Installed Pack bytes or exact provenance failed closed.""" + + +class InstalledDistribution(Protocol): + @property + def files(self): ... + + @property + def metadata(self): ... + + @property + def version(self) -> str: ... + + def locate_file(self, path) -> Path: ... + + +@dataclass(frozen=True, slots=True) +class InstalledCompiledPackArtifact: + """One recompiled, passing, immutable installed Pack artifact.""" + + distribution: str + distribution_version: str + manifest_resource_path: str + pack: CompiledDomainPackV1 + compilation: StablePackCompilationResultV1 + conformance_receipts: tuple[DomainPackConformanceReceiptV1, ...] + + +@dataclass(frozen=True, slots=True) +class _InstalledPackRoot: + """Validated manifest identity whose activation material remains unloaded.""" + + distribution: InstalledDistribution + distribution_name: str + distribution_version: str + declared: dict[str, object] + manifest_path: PurePosixPath + manifest_document: bytes + manifest: DomainPackManifestV1 + + +def _distribution_name(distribution: InstalledDistribution) -> str: + value = distribution.metadata.get("Name") + if not isinstance(value, str) or not value.strip(): + raise InstalledPackArtifactError("installed distribution omitted its canonical name") + return value.strip() + + +def _declared_paths(distribution: InstalledDistribution) -> dict[str, object]: + result: dict[str, object] = {} + for value in distribution.files or (): + normalized = str(value).replace("\\", "/") + if normalized in result: + raise InstalledPackArtifactError("installed distribution declares one resource path more than once") + result[normalized] = value + return result + + +def _manifest_paths(declared: dict[str, object]) -> tuple[object, ...]: + matches = [] + for normalized, original in declared.items(): + path = PurePosixPath(normalized) + if len(path.parts) == 3 and path.parts[0] == "domain_packs" and path.name == "manifest.json": + matches.append(original) + return tuple(sorted(matches, key=str)) + + +def _safe_bytes( + *, + distribution: InstalledDistribution, + declared: dict[str, object], + resource_path: PurePosixPath, + label: str, +) -> bytes: + original = declared.get(resource_path.as_posix()) + if original is None: + raise InstalledPackArtifactError(f"installed Pack omitted declared {label}: {resource_path}") + try: + distribution_root = Path(distribution.locate_file("")) + path = Path(distribution.locate_file(original)) + root = distribution_root.resolve(strict=True) + resolved = path.resolve(strict=True) + if not resolved.is_relative_to(root): + raise InstalledPackArtifactError(f"installed Pack {label} escaped its distribution root") + current = path + while current != distribution_root and current != current.parent: + if current.is_symlink(): + raise InstalledPackArtifactError(f"installed Pack {label} uses a symbolic link") + current = current.parent + if not path.is_file(): + raise InstalledPackArtifactError(f"installed Pack {label} is not a regular file") + return path.read_bytes() + except InstalledPackArtifactError: + raise + except OSError as exc: + raise InstalledPackArtifactError(f"installed Pack {label} is unreadable") from exc + + +def _manifest(bytes_value: bytes) -> DomainPackManifestV1: + if not bytes_value or len(bytes_value) > MAX_MANIFEST_BYTES: + raise InstalledPackArtifactError("installed Pack manifest exceeded its bounded size") + try: + return DomainPackManifestV1.model_validate(json.loads(bytes_value)) + except (UnicodeDecodeError, json.JSONDecodeError, ValidationError, TypeError, ValueError) as exc: + raise InstalledPackArtifactError("installed Pack manifest failed exact validation") from exc + + +def _validate_provenance( + *, + compiled: CompiledPackResultV1, + receipt: DomainPackConformanceReceiptV1, +) -> None: + pack = compiled.pack + exact = DomainPackConformanceReceiptV1.model_validate(receipt.model_dump(mode="python")) + if not exact.passed: + raise InstalledPackArtifactError("installed Pack activation conformance did not pass") + if ( + exact.pack_id != pack.metadata.pack_id + or exact.pack_version != pack.metadata.version + or exact.compiled_pack_id != pack.compiled_pack_id + or exact.pack_digest != pack.pack_digest + or exact.manifest_contract != pack.manifest_contract + or exact.compiler_contract != pack.compiler_contract + or exact.intelligence_contract != pack.intelligence_contract + or exact.compatibility_status != compiled.compatibility.status + or exact.compilation_result_id != compiled.compilation.result_id + or exact.compilation_result_digest != compiled.compilation.result_digest + ): + raise InstalledPackArtifactError("installed Pack conformance provenance crossed exact compilation material") + + +def _load_artifact( + *, + root: _InstalledPackRoot, +) -> InstalledCompiledPackArtifact: + pack_root = root.manifest_path.parent + resources = { + item.path: _safe_bytes( + distribution=root.distribution, + declared=root.declared, + resource_path=pack_root / item.path, + label=f"resource {item.resource_id}", + ) + for item in root.manifest.resources + } + fixture_document = _safe_bytes( + distribution=root.distribution, + declared=root.declared, + resource_path=pack_root / ACTIVATION_GOLDEN_FIXTURE_PATH, + label="activation golden fixture", + ) + try: + compiled = compile_pack_document_with_report(root.manifest_document, resources) + receipt = run_domain_pack_conformance( + manifest_document=root.manifest_document, + resources=resources, + fixture_document=fixture_document, + ) + _validate_provenance(compiled=compiled, receipt=receipt) + except InstalledPackArtifactError: + raise + except (PackCompilationError, ValidationError, TypeError, ValueError) as exc: + raise InstalledPackArtifactError("installed Pack compilation or activation conformance failed") from exc + return InstalledCompiledPackArtifact( + distribution=root.distribution_name, + distribution_version=root.distribution_version, + manifest_resource_path=root.manifest_path.as_posix(), + pack=compiled.pack, + compilation=compiled.compilation, + conformance_receipts=(receipt,), + ) + + +def _index_installed_pack_roots( + distributions: Iterable[InstalledDistribution] | None = None, +) -> tuple[_InstalledPackRoot, ...]: + """Index manifest identities without activating unrelated installed Packs.""" + + installed = metadata.distributions() if distributions is None else distributions + roots: list[_InstalledPackRoot] = [] + seen_roots: set[tuple[str, str]] = set() + seen_pack_ids: set[str] = set() + for distribution in sorted(installed, key=lambda item: _distribution_name(item).lower()): + name = _distribution_name(distribution) + declared = _declared_paths(distribution) + for manifest_path_value in _manifest_paths(declared): + manifest_path = PurePosixPath(str(manifest_path_value).replace("\\", "/")) + root_key = (name.lower(), manifest_path.parent.as_posix()) + if root_key in seen_roots: + raise InstalledPackArtifactError("installed Pack root is declared more than once") + seen_roots.add(root_key) + manifest_document = _safe_bytes( + distribution=distribution, + declared=declared, + resource_path=manifest_path, + label="manifest", + ) + manifest = _manifest(manifest_document) + if manifest.metadata.pack_id in seen_pack_ids: + raise InstalledPackArtifactError("installed Pack identifiers are ambiguous") + seen_pack_ids.add(manifest.metadata.pack_id) + roots.append( + _InstalledPackRoot( + distribution=distribution, + distribution_name=name, + distribution_version=str(distribution.version), + declared=declared, + manifest_path=manifest_path, + manifest_document=manifest_document, + manifest=manifest, + ) + ) + return tuple(sorted(roots, key=lambda item: item.manifest.metadata.pack_id)) + + +class InstalledCompiledPackArtifactResolver: + """Exact immutable resolver over one freshly discovered installed set.""" + + def __init__( + self, + artifacts: Iterable[InstalledCompiledPackArtifact] = (), + *, + roots: Iterable[_InstalledPackRoot] = (), + ) -> None: + exact = tuple(artifacts) + indexed = tuple(roots) + try: + validate_compiled_pack_set([item.pack for item in exact]) + for artifact in exact: + pack = CompiledDomainPackV1.model_validate(artifact.pack.model_dump(mode="python")) + compilation = StablePackCompilationResultV1.model_validate( + artifact.compilation.model_dump(mode="python") + ) + if ( + compilation.compiled_pack_id != pack.compiled_pack_id + or compilation.pack_digest != pack.pack_digest + or compilation.manifest_contract != pack.manifest_contract + or compilation.compiler_contract != pack.compiler_contract + or compilation.intelligence_contract != pack.intelligence_contract + or len(artifact.conformance_receipts) != 1 + ): + raise InstalledPackArtifactError("installed Pack resolver artifact crossed exact compilation") + receipt = DomainPackConformanceReceiptV1.model_validate( + artifact.conformance_receipts[0].model_dump(mode="python") + ) + if ( + not receipt.passed + or receipt.pack_id != pack.metadata.pack_id + or receipt.pack_version != pack.metadata.version + or receipt.compiled_pack_id != pack.compiled_pack_id + or receipt.pack_digest != pack.pack_digest + or receipt.compilation_result_id != compilation.result_id + or receipt.compilation_result_digest != compilation.result_digest + ): + raise InstalledPackArtifactError("installed Pack resolver artifact crossed exact conformance") + except InstalledPackArtifactError: + raise + except (PackCompilationError, ValidationError, AttributeError, TypeError, ValueError) as exc: + raise InstalledPackArtifactError("installed Pack resolver artifacts failed exact validation") from exc + root_ids = [item.manifest.metadata.pack_id for item in indexed] + if len(root_ids) != len(set(root_ids)) or set(root_ids).intersection( + item.pack.metadata.pack_id for item in exact + ): + raise InstalledPackArtifactError("installed Pack identifiers are ambiguous") + self._artifacts = {item.pack.metadata.pack_id: item for item in exact} + self._roots = {item.manifest.metadata.pack_id: item for item in indexed} + + @classmethod + def discover( + cls, + distributions: Iterable[InstalledDistribution] | None = None, + ) -> "InstalledCompiledPackArtifactResolver": + return cls(roots=_index_installed_pack_roots(distributions)) + + async def resolve_exact(self, *, reference: CompiledPackRefV1) -> InstalledCompiledPackArtifact | None: + artifact = self._artifacts.get(reference.pack_id) + if artifact is None: + root = self._roots.get(reference.pack_id) + if root is None or root.manifest.metadata.version != reference.pack_version: + return None + artifact = _load_artifact(root=root) + pack = artifact.pack + if ( + pack.metadata.version != reference.pack_version + or pack.compiled_pack_id != reference.compiled_pack_id + or pack.pack_digest != reference.pack_digest + ): + return None + return InstalledCompiledPackArtifact( + distribution=artifact.distribution, + distribution_version=artifact.distribution_version, + manifest_resource_path=artifact.manifest_resource_path, + pack=CompiledDomainPackV1.model_validate(pack.model_dump(mode="python")), + compilation=StablePackCompilationResultV1.model_validate(artifact.compilation.model_dump(mode="python")), + conformance_receipts=tuple( + DomainPackConformanceReceiptV1.model_validate(item.model_dump(mode="python")) + for item in artifact.conformance_receipts + ), + ) + + async def load_exact(self, *, reference: CompiledPackRefV1) -> CompiledDomainPackV1 | None: + artifact = await self.resolve_exact(reference=reference) + return None if artifact is None else artifact.pack + + +__all__ = [ + "ACTIVATION_GOLDEN_FIXTURE_PATH", + "InstalledCompiledPackArtifact", + "InstalledCompiledPackArtifactResolver", + "InstalledPackArtifactError", +] diff --git a/ace/intelligence/__init__.py b/ace/intelligence/__init__.py index 51a6c8a..c8c1b6d 100644 --- a/ace/intelligence/__init__.py +++ b/ace/intelligence/__init__.py @@ -7,6 +7,7 @@ deterministic functions over contract values only. """ +from ace.intelligence.conformance import conformance_receipt_json, run_domain_pack_conformance from ace.intelligence.contracts import * # noqa: F403 from ace.intelligence.contracts import __all__ as _CONTRACTS_ALL from ace.intelligence.derivation import ( @@ -123,6 +124,7 @@ "eligible_signal_routes", "evaluate_measured_impact", "compare_measured_composition", + "conformance_receipt_json", "interpret_live_source_mapping", "interpret_prepared_source_mapping", "render_canonical_brief_body", @@ -130,5 +132,6 @@ "route_live_categorical_shift_as_signal", "route_live_shift_as_signal", "route_shift_as_signal", + "run_domain_pack_conformance", "validate_brief_synthesis_draft", ] diff --git a/ace/intelligence/conformance.py b/ace/intelligence/conformance.py new file mode 100644 index 0000000..fc7306d --- /dev/null +++ b/ace/intelligence/conformance.py @@ -0,0 +1,298 @@ +"""Provider-free golden-fixture conformance for installed inert Domain Packs.""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import UTC +from typing import Any + +from pydantic import ValidationError + +from ace.core.contracts import canonical_hash, canonical_json +from ace.intelligence.contracts.activation import ( + ActivationState, + CompiledPackRefV1, + DomainActivationRevisionV1, + DomainActivationSpecV1, + OrganizationOverlayV1, +) +from ace.intelligence.contracts.common import MAX_RESOURCE_BYTES, parse_json_strict +from ace.intelligence.contracts.conformance import ( + DomainPackConformanceReceiptV1, + DomainPackGoldenFixtureV1, + GoldenDetectorOutcomeV1, +) +from ace.intelligence.contracts.detection import ( + DETECTION_MODULE_V1ALPHA2_VERSION, + DETECTION_MODULE_VERSION, + DetectionModuleV1, + DetectionModuleV1Alpha2, + NumericDeltaRuleV1, +) +from ace.intelligence.contracts.resources import ( + CanonicalJsonValueV1Alpha1, + EntitySnapshotV1Alpha1, + IntelligenceResourceMode, +) +from ace.intelligence.detection.categorical_transition import ( + detect_categorical_shift, + route_categorical_shift_as_signal, +) +from ace.intelligence.detection.numeric_delta import detect_numeric_shift, route_shift_as_signal +from ace.intelligence.packs.activation import compile_overlay +from ace.intelligence.packs.compiler import compile_pack_document_with_report +from ace.intelligence.packs.diagnostics import PackDiagnosticV1 +from ace.intelligence.packs.runtime import bind_prepared_activation, resolve_entity_type_declaration +from ace.intelligence.routing import eligible_signal_routes + + +def _digest(value: Any) -> str: + return f"sha256:{canonical_hash(value)}" + + +def _fixture(document: bytes) -> DomainPackGoldenFixtureV1: + if not isinstance(document, bytes): + raise TypeError("fixture document must be bytes") + if len(document) > MAX_RESOURCE_BYTES: + raise ValueError(f"fixture exceeds the {MAX_RESOURCE_BYTES}-byte bound") + try: + payload = parse_json_strict(document.decode("utf-8")) + return DomainPackGoldenFixtureV1.model_validate(payload) + except (UnicodeDecodeError, ValueError, ValidationError, RecursionError) as exc: + raise ValueError("fixture must be strict UTF-8 JSON matching the stable golden-fixture schema") from exc + + +def _binding(pack, fixture: DomainPackGoldenFixtureV1): + overlay = compile_overlay( + pack, + OrganizationOverlayV1( + overlay_id="pack_conformance", + version=fixture.fixture_version, + pack_id=pack.metadata.pack_id, + pack_version=pack.metadata.version, + pack_digest=pack.pack_digest, + values=fixture.overlay_values, + ), + ) + pack_ref = CompiledPackRefV1( + pack_id=pack.metadata.pack_id, + pack_version=pack.metadata.version, + compiled_pack_id=pack.compiled_pack_id, + pack_digest=pack.pack_digest, + ) + occurred_at = min(item.baseline_as_of for item in fixture.observations).astimezone(UTC) + spec = DomainActivationSpecV1( + product_id="product:pack-conformance", + activation_key=pack.metadata.pack_id, + pack=pack_ref, + overlay=overlay, + compilation_receipt_ref=f"compilation:{pack.pack_digest.removeprefix('sha256:')[:32]}", + conformance_receipt_refs=("conformance:pending",), + ) + revision = DomainActivationRevisionV1( + revision=1, + spec=spec, + state=ActivationState.ACTIVE, + actor_ref="principal:pack-conformance", + approval_receipt_ref="approval:pack-conformance-ephemeral", + occurred_at=occurred_at, + ) + return bind_prepared_activation(pack=pack, revision=revision) + + +def _detector_rules(pack) -> tuple[Any, ...]: + models = { + DETECTION_MODULE_VERSION: DetectionModuleV1, + DETECTION_MODULE_V1ALPHA2_VERSION: DetectionModuleV1Alpha2, + } + rules: list[Any] = [] + for module in pack.modules: + model = models.get(module.contract) + if model is None: + continue + parsed = model.model_validate_json(module.canonical_payload) + rules.extend(parsed.numeric_delta_rules) + rules.extend(getattr(parsed, "categorical_transition_rules", ())) + return tuple(sorted(rules, key=lambda item: item.detector_id)) + + +def _outcomes(pack, fixture: DomainPackGoldenFixtureV1) -> tuple[dict[str, Any], ...]: + binding = _binding(pack, fixture) + rules = _detector_rules(pack) + results: list[dict[str, Any]] = [] + for case in fixture.observations: + resolve_entity_type_declaration(binding, entity_type_id=case.entity_type_id) + baseline = EntitySnapshotV1Alpha1( + product_id=binding.reference.product_id, + mode=IntelligenceResourceMode.PREPARED, + activation_revision=binding.reference, + as_of=case.baseline_as_of, + entity_ref=case.entity_ref, + entity_type_ref=case.entity_type_id, + attributes=CanonicalJsonValueV1Alpha1(value_json=case.baseline_attributes_json), + projected_at=case.baseline_as_of, + confidence=case.confidence, + ) + current = EntitySnapshotV1Alpha1( + product_id=binding.reference.product_id, + mode=IntelligenceResourceMode.PREPARED, + activation_revision=binding.reference, + as_of=case.current_as_of, + entity_ref=case.entity_ref, + entity_type_ref=case.entity_type_id, + attributes=CanonicalJsonValueV1Alpha1(value_json=case.current_attributes_json), + projected_at=case.current_as_of, + confidence=case.confidence, + ) + case_outcomes: list[GoldenDetectorOutcomeV1] = [] + for rule in rules: + if rule.entity_type_id != case.entity_type_id: + continue + if isinstance(rule, NumericDeltaRuleV1): + shift = detect_numeric_shift( + binding=binding, + detector_id=rule.detector_id, + baseline=baseline, + current=current, + detected_at=case.current_as_of, + ) + signal = ( + None + if shift is None + else route_shift_as_signal( + binding=binding, + detector_id=rule.detector_id, + shift=shift, + detected_at=case.current_as_of, + ) + ) + else: + shift = detect_categorical_shift( + binding=binding, + detector_id=rule.detector_id, + baseline=baseline, + current=current, + detected_at=case.current_as_of, + ) + signal = ( + None + if shift is None + else route_categorical_shift_as_signal( + binding=binding, + detector_id=rule.detector_id, + shift=shift, + detected_at=case.current_as_of, + ) + ) + routes = () if signal is None else eligible_signal_routes(binding=binding, signal=signal) + case_outcomes.append( + GoldenDetectorOutcomeV1( + detector_id=rule.detector_id, + entity_ref=case.entity_ref, + material=shift is not None, + shift_type=None if shift is None else shift.shift_type_ref, + signal_type=None if signal is None else signal.signal_type_ref, + routing_rule_ids=tuple(route.routing_rule_id for route in routes), + persona_ids=tuple(sorted({persona for route in routes for persona in route.persona_ids})), + template_ids=tuple( + sorted({route.brief_template_id for route in routes if route.brief_template_id is not None}) + ), + ) + ) + results.append( + { + "case_id": case.case_id, + "outcomes": [ + item.model_dump(mode="json") for item in sorted(case_outcomes, key=lambda x: x.detector_id) + ], + } + ) + return tuple(results) + + +def run_domain_pack_conformance( + *, + manifest_document: bytes, + resources: Mapping[str, bytes], + fixture_document: bytes, + prior_receipt: DomainPackConformanceReceiptV1 | None = None, +) -> DomainPackConformanceReceiptV1: + """Compile and exercise one pack without providers, network, persistence, or a database.""" + + compiled = compile_pack_document_with_report(manifest_document, resources) + pack = compiled.pack + fixture = _fixture(fixture_document) + fixture_material = fixture.model_dump(mode="json") + fixture_digest = _digest(fixture_material) + expected = tuple( + { + "case_id": case.case_id, + "outcomes": [item.model_dump(mode="json") for item in case.expected], + } + for case in fixture.observations + ) + diagnostics: list[PackDiagnosticV1] = [] + try: + actual = _outcomes(pack, fixture) + except (TypeError, ValueError) as exc: + actual = () + diagnostics.append( + PackDiagnosticV1( + severity="error", + code="conformance_runtime_error", + path="fixture.observations", + message=str(exc)[:1_000] or "golden-fixture evaluation failed", + ) + ) + if ( + prior_receipt is not None + and prior_receipt.fixture_id == fixture.fixture_id + and (prior_receipt.fixture_digest != fixture_digest or prior_receipt.expected_digest != _digest(expected)) + ): + diagnostics.append( + PackDiagnosticV1( + severity="error", + code="divergent_conformance_identity", + path="fixture.fixture_id", + message="the same fixture identity cannot name changed fixture or expected-result material", + ) + ) + if actual != expected: + diagnostics.append( + PackDiagnosticV1( + severity="error", + code="golden_result_mismatch", + path="fixture.observations", + message="derived entity, Shift, Signal, routing, persona, or synthesis selection differs from the golden result", + ) + ) + expected_digest = _digest(expected) + actual_digest = _digest(actual) + return DomainPackConformanceReceiptV1( + pack_id=pack.metadata.pack_id, + pack_version=pack.metadata.version, + compiled_pack_id=pack.compiled_pack_id, + pack_digest=pack.pack_digest, + manifest_contract=pack.manifest_contract, + compiler_contract=pack.compiler_contract, + intelligence_contract=pack.intelligence_contract, + compatibility_status=compiled.compatibility.status, + compilation_result_id=compiled.compilation.result_id, + compilation_result_digest=compiled.compilation.result_digest, + fixture_id=fixture.fixture_id, + fixture_version=fixture.fixture_version, + fixture_digest=fixture_digest, + expected_digest=expected_digest, + actual_digest=actual_digest, + passed=not diagnostics, + diagnostics=tuple(diagnostics), + ) + + +def conformance_receipt_json(receipt: DomainPackConformanceReceiptV1) -> str: + """Return the exact portable receipt representation used for byte comparisons.""" + + return canonical_json(receipt) + + +__all__ = ["conformance_receipt_json", "run_domain_pack_conformance"] diff --git a/ace/testing/domain_pack.py b/ace/testing/domain_pack.py index a42fb9f..7942783 100644 --- a/ace/testing/domain_pack.py +++ b/ace/testing/domain_pack.py @@ -1,298 +1,5 @@ -"""Public provider-free golden-fixture conformance for third-party Domain Packs.""" - -from __future__ import annotations - -from collections.abc import Mapping -from datetime import UTC -from typing import Any - -from pydantic import ValidationError - -from ace.core.contracts import canonical_hash, canonical_json -from ace.intelligence.contracts.activation import ( - ActivationState, - CompiledPackRefV1, - DomainActivationRevisionV1, - DomainActivationSpecV1, - OrganizationOverlayV1, -) -from ace.intelligence.contracts.common import MAX_RESOURCE_BYTES, parse_json_strict -from ace.intelligence.contracts.conformance import ( - DomainPackConformanceReceiptV1, - DomainPackGoldenFixtureV1, - GoldenDetectorOutcomeV1, -) -from ace.intelligence.contracts.detection import ( - DETECTION_MODULE_V1ALPHA2_VERSION, - DETECTION_MODULE_VERSION, - DetectionModuleV1, - DetectionModuleV1Alpha2, - NumericDeltaRuleV1, -) -from ace.intelligence.contracts.resources import ( - CanonicalJsonValueV1Alpha1, - EntitySnapshotV1Alpha1, - IntelligenceResourceMode, -) -from ace.intelligence.detection.categorical_transition import ( - detect_categorical_shift, - route_categorical_shift_as_signal, -) -from ace.intelligence.detection.numeric_delta import detect_numeric_shift, route_shift_as_signal -from ace.intelligence.packs.activation import compile_overlay -from ace.intelligence.packs.compiler import compile_pack_document_with_report -from ace.intelligence.packs.diagnostics import PackDiagnosticV1 -from ace.intelligence.packs.runtime import bind_prepared_activation, resolve_entity_type_declaration -from ace.intelligence.routing import eligible_signal_routes - - -def _digest(value: Any) -> str: - return f"sha256:{canonical_hash(value)}" - - -def _fixture(document: bytes) -> DomainPackGoldenFixtureV1: - if not isinstance(document, bytes): - raise TypeError("fixture document must be bytes") - if len(document) > MAX_RESOURCE_BYTES: - raise ValueError(f"fixture exceeds the {MAX_RESOURCE_BYTES}-byte bound") - try: - payload = parse_json_strict(document.decode("utf-8")) - return DomainPackGoldenFixtureV1.model_validate(payload) - except (UnicodeDecodeError, ValueError, ValidationError, RecursionError) as exc: - raise ValueError("fixture must be strict UTF-8 JSON matching the stable golden-fixture schema") from exc - - -def _binding(pack, fixture: DomainPackGoldenFixtureV1): - overlay = compile_overlay( - pack, - OrganizationOverlayV1( - overlay_id="pack_conformance", - version=fixture.fixture_version, - pack_id=pack.metadata.pack_id, - pack_version=pack.metadata.version, - pack_digest=pack.pack_digest, - values=fixture.overlay_values, - ), - ) - pack_ref = CompiledPackRefV1( - pack_id=pack.metadata.pack_id, - pack_version=pack.metadata.version, - compiled_pack_id=pack.compiled_pack_id, - pack_digest=pack.pack_digest, - ) - occurred_at = min(item.baseline_as_of for item in fixture.observations).astimezone(UTC) - spec = DomainActivationSpecV1( - product_id="product:pack-conformance", - activation_key=pack.metadata.pack_id, - pack=pack_ref, - overlay=overlay, - compilation_receipt_ref=f"compilation:{pack.pack_digest.removeprefix('sha256:')[:32]}", - conformance_receipt_refs=("conformance:pending",), - ) - revision = DomainActivationRevisionV1( - revision=1, - spec=spec, - state=ActivationState.ACTIVE, - actor_ref="principal:pack-conformance", - approval_receipt_ref="approval:pack-conformance-ephemeral", - occurred_at=occurred_at, - ) - return bind_prepared_activation(pack=pack, revision=revision) - - -def _detector_rules(pack) -> tuple[Any, ...]: - models = { - DETECTION_MODULE_VERSION: DetectionModuleV1, - DETECTION_MODULE_V1ALPHA2_VERSION: DetectionModuleV1Alpha2, - } - rules: list[Any] = [] - for module in pack.modules: - model = models.get(module.contract) - if model is None: - continue - parsed = model.model_validate_json(module.canonical_payload) - rules.extend(parsed.numeric_delta_rules) - rules.extend(getattr(parsed, "categorical_transition_rules", ())) - return tuple(sorted(rules, key=lambda item: item.detector_id)) - - -def _outcomes(pack, fixture: DomainPackGoldenFixtureV1) -> tuple[dict[str, Any], ...]: - binding = _binding(pack, fixture) - rules = _detector_rules(pack) - results: list[dict[str, Any]] = [] - for case in fixture.observations: - resolve_entity_type_declaration(binding, entity_type_id=case.entity_type_id) - baseline = EntitySnapshotV1Alpha1( - product_id=binding.reference.product_id, - mode=IntelligenceResourceMode.PREPARED, - activation_revision=binding.reference, - as_of=case.baseline_as_of, - entity_ref=case.entity_ref, - entity_type_ref=case.entity_type_id, - attributes=CanonicalJsonValueV1Alpha1(value_json=case.baseline_attributes_json), - projected_at=case.baseline_as_of, - confidence=case.confidence, - ) - current = EntitySnapshotV1Alpha1( - product_id=binding.reference.product_id, - mode=IntelligenceResourceMode.PREPARED, - activation_revision=binding.reference, - as_of=case.current_as_of, - entity_ref=case.entity_ref, - entity_type_ref=case.entity_type_id, - attributes=CanonicalJsonValueV1Alpha1(value_json=case.current_attributes_json), - projected_at=case.current_as_of, - confidence=case.confidence, - ) - case_outcomes: list[GoldenDetectorOutcomeV1] = [] - for rule in rules: - if rule.entity_type_id != case.entity_type_id: - continue - if isinstance(rule, NumericDeltaRuleV1): - shift = detect_numeric_shift( - binding=binding, - detector_id=rule.detector_id, - baseline=baseline, - current=current, - detected_at=case.current_as_of, - ) - signal = ( - None - if shift is None - else route_shift_as_signal( - binding=binding, - detector_id=rule.detector_id, - shift=shift, - detected_at=case.current_as_of, - ) - ) - else: - shift = detect_categorical_shift( - binding=binding, - detector_id=rule.detector_id, - baseline=baseline, - current=current, - detected_at=case.current_as_of, - ) - signal = ( - None - if shift is None - else route_categorical_shift_as_signal( - binding=binding, - detector_id=rule.detector_id, - shift=shift, - detected_at=case.current_as_of, - ) - ) - routes = () if signal is None else eligible_signal_routes(binding=binding, signal=signal) - case_outcomes.append( - GoldenDetectorOutcomeV1( - detector_id=rule.detector_id, - entity_ref=case.entity_ref, - material=shift is not None, - shift_type=None if shift is None else shift.shift_type_ref, - signal_type=None if signal is None else signal.signal_type_ref, - routing_rule_ids=tuple(route.routing_rule_id for route in routes), - persona_ids=tuple(sorted({persona for route in routes for persona in route.persona_ids})), - template_ids=tuple( - sorted({route.brief_template_id for route in routes if route.brief_template_id is not None}) - ), - ) - ) - results.append( - { - "case_id": case.case_id, - "outcomes": [ - item.model_dump(mode="json") for item in sorted(case_outcomes, key=lambda x: x.detector_id) - ], - } - ) - return tuple(results) - - -def run_domain_pack_conformance( - *, - manifest_document: bytes, - resources: Mapping[str, bytes], - fixture_document: bytes, - prior_receipt: DomainPackConformanceReceiptV1 | None = None, -) -> DomainPackConformanceReceiptV1: - """Compile and exercise one pack without providers, network, persistence, or a database.""" - - compiled = compile_pack_document_with_report(manifest_document, resources) - pack = compiled.pack - fixture = _fixture(fixture_document) - fixture_material = fixture.model_dump(mode="json") - fixture_digest = _digest(fixture_material) - expected = tuple( - { - "case_id": case.case_id, - "outcomes": [item.model_dump(mode="json") for item in case.expected], - } - for case in fixture.observations - ) - diagnostics: list[PackDiagnosticV1] = [] - try: - actual = _outcomes(pack, fixture) - except (TypeError, ValueError) as exc: - actual = () - diagnostics.append( - PackDiagnosticV1( - severity="error", - code="conformance_runtime_error", - path="fixture.observations", - message=str(exc)[:1_000] or "golden-fixture evaluation failed", - ) - ) - if ( - prior_receipt is not None - and prior_receipt.fixture_id == fixture.fixture_id - and (prior_receipt.fixture_digest != fixture_digest or prior_receipt.expected_digest != _digest(expected)) - ): - diagnostics.append( - PackDiagnosticV1( - severity="error", - code="divergent_conformance_identity", - path="fixture.fixture_id", - message="the same fixture identity cannot name changed fixture or expected-result material", - ) - ) - if actual != expected: - diagnostics.append( - PackDiagnosticV1( - severity="error", - code="golden_result_mismatch", - path="fixture.observations", - message="derived entity, Shift, Signal, routing, persona, or synthesis selection differs from the golden result", - ) - ) - expected_digest = _digest(expected) - actual_digest = _digest(actual) - return DomainPackConformanceReceiptV1( - pack_id=pack.metadata.pack_id, - pack_version=pack.metadata.version, - compiled_pack_id=pack.compiled_pack_id, - pack_digest=pack.pack_digest, - manifest_contract=pack.manifest_contract, - compiler_contract=pack.compiler_contract, - intelligence_contract=pack.intelligence_contract, - compatibility_status=compiled.compatibility.status, - compilation_result_id=compiled.compilation.result_id, - compilation_result_digest=compiled.compilation.result_digest, - fixture_id=fixture.fixture_id, - fixture_version=fixture.fixture_version, - fixture_digest=fixture_digest, - expected_digest=expected_digest, - actual_digest=actual_digest, - passed=not diagnostics, - diagnostics=tuple(diagnostics), - ) - - -def conformance_receipt_json(receipt: DomainPackConformanceReceiptV1) -> str: - """Return the exact portable receipt representation used for byte comparisons.""" - - return canonical_json(receipt) +"""Compatibility exports for the production-safe Domain Pack conformance runner.""" +from ace.intelligence.conformance import conformance_receipt_json, run_domain_pack_conformance __all__ = ["conformance_receipt_json", "run_domain_pack_conformance"] diff --git a/tests/intelligence/test_contract_boundaries.py b/tests/intelligence/test_contract_boundaries.py index 018ca03..6d6f209 100644 --- a/tests/intelligence/test_contract_boundaries.py +++ b/tests/intelligence/test_contract_boundaries.py @@ -54,6 +54,7 @@ def test_intelligence_initializer_exports_contracts_and_pure_interpreters_only() imports = _imports(INTELLIGENCE / "__init__.py") assert imports == { "ace.intelligence.contracts", + "ace.intelligence.conformance", "ace.intelligence.derivation", "ace.intelligence.detection", "ace.intelligence.epistemic", diff --git a/tests/test_installed_pack_artifacts.py b/tests/test_installed_pack_artifacts.py new file mode 100644 index 0000000..dfd9ba7 --- /dev/null +++ b/tests/test_installed_pack_artifacts.py @@ -0,0 +1,344 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path, PurePosixPath + +import pytest + +from ace.application.installed_pack_artifacts import ( + InstalledCompiledPackArtifactResolver, + InstalledPackArtifactError, +) +from ace.intelligence.contracts.activation import CompiledPackRefV1 +from ace.intelligence.packs.compiler import compile_pack_document_with_report + +pytestmark = pytest.mark.unit + + +def _encoded(value) -> bytes: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode() + + +def _digest(value: bytes) -> str: + return f"sha256:{hashlib.sha256(value).hexdigest()}" + + +def _pack(*, pack_id: str = "neutral_measurement", expected_material: bool = True) -> dict[str, bytes]: + modules = { + "modules/ontology.json": { + "contract": "ace.intelligence.ontology/v1alpha1", + "module_id": "ontology", + "entity_types": [ + { + "entity_type_id": "measurement", + "attributes": [{"attribute_id": "value", "value_type": "number", "required": True}], + } + ], + "relation_types": [], + }, + "modules/detection.json": { + "contract": "ace.intelligence.detection/v1alpha1", + "module_id": "detection", + "numeric_delta_rules": [ + { + "detector_id": "value_change", + "entity_type_id": "measurement", + "attribute_id": "value", + "metric": "absolute_change", + "threshold": 5, + "direction": "any", + "shift_type": "value_changed", + "signal_type": "value_attention", + } + ], + }, + "modules/synthesis.json": { + "contract": "ace.intelligence.synthesis/v1alpha1", + "module_id": "synthesis", + "brief_templates": [ + { + "template_id": "measurement_brief", + "brief_type": "measurement_update", + "display_name": "Measurement update", + "objective": "Explain the material change.", + "required_sections": ["summary"], + } + ], + }, + "modules/personas.json": { + "contract": "ace.intelligence.personas/v1alpha1", + "module_id": "personas", + "personas": [{"persona_id": "analyst", "display_name": "Analyst", "description": "Reviews."}], + "signal_routing_rules": [ + { + "routing_rule_id": "measurement_route", + "signal_type": "value_attention", + "persona_ids": ["analyst"], + "minimum_confidence": 0.5, + "brief_template_id": "measurement_brief", + } + ], + }, + } + refs = ( + ("ontology", "ace.intelligence.ontology/v1alpha1", "ontology_resource", ()), + ("detection", "ace.intelligence.detection/v1alpha1", "detection_resource", ("ontology",)), + ("synthesis", "ace.intelligence.synthesis/v1alpha1", "synthesis_resource", ()), + ("personas", "ace.intelligence.personas/v1alpha1", "personas_resource", ("detection", "synthesis")), + ) + manifest = { + "contract": "ace.intelligence.domain-pack-manifest/v1", + "metadata": {"pack_id": pack_id, "version": "1.0.0", "display_name": "Neutral Measurement"}, + "compatibility": { + "compiler_minimum": "ace.intelligence.pack-compiler/v1alpha1", + "compiler_maximum_exclusive": "ace.intelligence.pack-compiler/v2", + "intelligence_minimum": "ace.intelligence.runtime/v1alpha1", + "intelligence_maximum_exclusive": "ace.intelligence.runtime/v2", + }, + "resources": [ + { + "resource_id": resource_id, + "path": path, + "digest": _digest(_encoded(modules[path])), + } + for path, (_, _, resource_id, _) in zip(modules, refs, strict=True) + ], + "modules": [ + { + "module_id": module_id, + "contract": contract, + "resource_id": resource_id, + "depends_on": depends_on, + } + for module_id, contract, resource_id, depends_on in refs + ], + } + expected = [ + { + "detector_id": "value_change", + "entity_ref": "entity:measurement-one", + "material": expected_material, + **( + { + "shift_type": "value_changed", + "signal_type": "value_attention", + "routing_rule_ids": ["measurement_route"], + "persona_ids": ["analyst"], + "template_ids": ["measurement_brief"], + } + if expected_material + else {} + ), + } + ] + fixture = { + "contract": "ace.intelligence.domain-pack-golden-fixture/v1", + "fixture_id": f"{pack_id}_activation", + "fixture_version": "1.0.0", + "observations": [ + { + "case_id": "material_change", + "entity_type_id": "measurement", + "entity_ref": "entity:measurement-one", + "baseline_attributes_json": '{"value":10}', + "current_attributes_json": '{"value":20}', + "baseline_as_of": "2026-08-11T00:00:00Z", + "current_as_of": "2026-08-11T01:00:00Z", + "confidence": 0.9, + "expected": expected, + } + ], + } + root = f"domain_packs/{pack_id}" + return { + f"{root}/manifest.json": _encoded(manifest), + **{f"{root}/{path}": _encoded(value) for path, value in modules.items()}, + f"{root}/conformance/activation_golden_fixture.json": _encoded(fixture), + f"{root}/conformance/manifest.json": b"not an installed Pack root", + f"{root}/releases/v2/manifest.json": b"not an installed Pack root", + f"{root}/__init__.py": b"raise RuntimeError('must never import inert pack')", + } + + +@dataclass +class _Metadata: + name: str + + def get(self, key: str): + return self.name if key == "Name" else None + + +class _Distribution: + def __init__(self, root: Path, name: str, resources: dict[str, bytes], *, version: str = "99.4.1") -> None: + self.root = root + self.metadata = _Metadata(name) + self.version = version + self.files = tuple(PurePosixPath(path) for path in resources) + for relative, payload in resources.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(payload) + + def locate_file(self, path) -> Path: + return self.root / str(path) + + @property + def entry_points(self): + raise AssertionError("installed Pack discovery must never inspect or execute entry points") + + +def _reference(resources: dict[str, bytes], *, pack_id: str = "neutral_measurement") -> CompiledPackRefV1: + root = f"domain_packs/{pack_id}" + manifest_document = resources[f"{root}/manifest.json"] + manifest = json.loads(manifest_document) + pack = compile_pack_document_with_report( + manifest_document, + {item["path"]: resources[f"{root}/{item['path']}"] for item in manifest["resources"]}, + ).pack + return CompiledPackRefV1( + pack_id=pack.metadata.pack_id, + pack_version=pack.metadata.version, + compiled_pack_id=pack.compiled_pack_id, + pack_digest=pack.pack_digest, + ) + + +@pytest.mark.asyncio +async def test_discovers_inert_pack_by_shape_and_resolves_all_exact_coordinates(tmp_path: Path) -> None: + resources = _pack() + distribution = _Distribution(tmp_path / "renamed", "anything-at-all", resources) + + first = InstalledCompiledPackArtifactResolver.discover([distribution]) + second = InstalledCompiledPackArtifactResolver.discover([distribution]) + exact = _reference(resources) + artifact = await first.resolve_exact(reference=exact) + rediscovered = await second.resolve_exact(reference=exact) + + assert artifact == rediscovered + assert artifact is not None + assert artifact.distribution == "anything-at-all" + assert artifact.distribution_version == "99.4.1" + assert artifact.pack.metadata.version == "1.0.0" + assert artifact.conformance_receipts[0].passed is True + assert artifact.conformance_receipts[0].compilation_result_id == artifact.compilation.result_id + assert await first.load_exact(reference=exact) == artifact.pack + assert (await first.resolve_exact(reference=exact)).conformance_receipts == artifact.conformance_receipts + for field, value in ( + ("pack_id", "other_pack"), + ("pack_version", "2.0.0"), + ("compiled_pack_id", "pack_ir:" + "0" * 32), + ("pack_digest", "sha256:" + "0" * 64), + ): + changed = exact.model_copy(update={field: value}) + assert await first.load_exact(reference=changed) is None + + +@pytest.mark.asyncio +async def test_exact_requested_pack_rejects_missing_malformed_or_failed_fixed_fixture(tmp_path: Path) -> None: + missing = _pack() + missing.pop("domain_packs/neutral_measurement/conformance/activation_golden_fixture.json") + missing_resolver = InstalledCompiledPackArtifactResolver.discover( + [_Distribution(tmp_path / "missing", "missing", missing)] + ) + with pytest.raises(InstalledPackArtifactError, match="activation golden fixture"): + await missing_resolver.resolve_exact(reference=_reference(missing)) + + malformed = _pack() + malformed["domain_packs/neutral_measurement/conformance/activation_golden_fixture.json"] = b"not-json" + malformed_resolver = InstalledCompiledPackArtifactResolver.discover( + [_Distribution(tmp_path / "malformed", "malformed", malformed)] + ) + with pytest.raises(InstalledPackArtifactError, match="validation|conformance"): + await malformed_resolver.resolve_exact(reference=_reference(malformed)) + + failed = _pack(expected_material=False) + failed_resolver = InstalledCompiledPackArtifactResolver.discover( + [_Distribution(tmp_path / "failed", "failed", failed)] + ) + with pytest.raises(InstalledPackArtifactError, match="conformance"): + await failed_resolver.resolve_exact(reference=_reference(failed)) + + +@pytest.mark.asyncio +async def test_unrelated_unactivatable_pack_does_not_block_exact_valid_resolution(tmp_path: Path) -> None: + requested = _pack() + unrelated = _pack(pack_id="unrelated_unready") + unrelated.pop("domain_packs/unrelated_unready/conformance/activation_golden_fixture.json") + distribution = _Distribution(tmp_path / "combined", "combined", {**requested, **unrelated}) + + resolver = InstalledCompiledPackArtifactResolver.discover([distribution]) + artifact = await resolver.resolve_exact(reference=_reference(requested)) + + assert artifact is not None + assert artifact.pack.metadata.pack_id == "neutral_measurement" + with pytest.raises(InstalledPackArtifactError, match="activation golden fixture"): + await resolver.resolve_exact(reference=_reference(unrelated, pack_id="unrelated_unready")) + + +@pytest.mark.asyncio +async def test_rejects_requested_resource_substitution_and_indexes_duplicates_fail_closed(tmp_path: Path) -> None: + substituted = _pack() + substituted["domain_packs/neutral_measurement/modules/ontology.json"] += b" " + substituted_resolver = InstalledCompiledPackArtifactResolver.discover( + [_Distribution(tmp_path / "changed", "changed", substituted)] + ) + with pytest.raises(InstalledPackArtifactError, match="compilation"): + await substituted_resolver.resolve_exact(reference=_reference(_pack())) + + one = _Distribution(tmp_path / "one", "one", _pack()) + two = _Distribution(tmp_path / "two", "two", _pack()) + with pytest.raises(InstalledPackArtifactError, match="ambiguous"): + InstalledCompiledPackArtifactResolver.discover([one, two]) + + +@pytest.mark.asyncio +async def test_rejects_symlinked_requested_installed_material(tmp_path: Path) -> None: + resources = _pack() + distribution = _Distribution(tmp_path / "linked", "linked", resources) + module = distribution.root / "domain_packs/neutral_measurement/modules/ontology.json" + target = tmp_path / "outside.json" + target.write_bytes(module.read_bytes()) + module.unlink() + module.symlink_to(target) + resolver = InstalledCompiledPackArtifactResolver.discover([distribution]) + with pytest.raises(InstalledPackArtifactError, match="symbolic link|escaped"): + await resolver.resolve_exact(reference=_reference(resources)) + + +@pytest.mark.asyncio +async def test_resolver_constructor_rejects_changed_compilation_or_conformance(tmp_path: Path) -> None: + resources = _pack() + discovered = InstalledCompiledPackArtifactResolver.discover([_Distribution(tmp_path / "exact", "exact", resources)]) + artifact = await discovered.resolve_exact(reference=_reference(resources)) + assert artifact is not None + changed_compilation = artifact.compilation.model_copy(update={"pack_digest": "sha256:" + "0" * 64}) + with pytest.raises(InstalledPackArtifactError, match="validation|compilation"): + InstalledCompiledPackArtifactResolver( + ( + artifact.__class__( + distribution=artifact.distribution, + distribution_version=artifact.distribution_version, + manifest_resource_path=artifact.manifest_resource_path, + pack=artifact.pack, + compilation=changed_compilation, + conformance_receipts=artifact.conformance_receipts, + ), + ) + ) + + changed_conformance = artifact.conformance_receipts[0].model_copy(update={"passed": False}) + with pytest.raises(InstalledPackArtifactError, match="validation|conformance"): + InstalledCompiledPackArtifactResolver( + ( + artifact.__class__( + distribution=artifact.distribution, + distribution_version=artifact.distribution_version, + manifest_resource_path=artifact.manifest_resource_path, + pack=artifact.pack, + compilation=artifact.compilation, + conformance_receipts=(changed_conformance,), + ), + ) + )