diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index f30c756..94f5b3c 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -181,7 +181,7 @@ - _Depends: 3.3, 5.1_ - [ ] 7. Compose validated evidence into bounded decisions and consumer output -- [ ] 7.1 Revalidate deterministic and fact-derived records through one shared context +- [x] 7.1 Revalidate deterministic and fact-derived records through one shared context - Bind the exact catalog, packs, model declarations, lineage set, review policies, provenance trust, and custody policy. - Resolve the family contract and all seven named provenance inputs for each raw record. - Re-run family and fact validation and reject embedded pack, lineage, or named-input substitutes. diff --git a/src/openultrasast/core/validation.py b/src/openultrasast/core/validation.py new file mode 100644 index 0000000..868ba67 --- /dev/null +++ b/src/openultrasast/core/validation.py @@ -0,0 +1,390 @@ +"""Closed validation context for raw detection-core evidence records.""" + +from __future__ import annotations + +from dataclasses import dataclass, fields + +from openultrasast.core import canonical +from openultrasast.core.custody import ( + PopulationCustodyPolicy, + PopulationCustodyPort, + derive_population_custody_policy_digest, + validate_population_custody_log, +) +from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, derive_record_id +from openultrasast.core.factpack import ( + FactPack, + ReviewIndependencePolicy, + derive_fact_pack_content_digest, + derive_review_evidence_digest, + derive_review_independence_policy_digest, + validate_fact_pack, +) +from openultrasast.core.factvalidate import validate_fact_relations +from openultrasast.core.family import validate_record +from openultrasast.core.family_catalog import FamilyCatalog, resolve_contract, validate_catalog +from openultrasast.core.lineage import LineageManifest, derive_lineage_manifest_digest +from openultrasast.core.model import ( + ModelProgramDeclaration, + ModelRequest, + ModelResult, + derive_model_program_declaration_digest, +) +from openultrasast.core.provenance import ( + NamedInputs, + ProvenanceTrustPolicy, + derive_trust_policy_digest, + validate_provenance, +) + +EVIDENCE_VALIDATION_CONTEXT_SCHEMA = "openultrasast.detection-core.evidence-validation-context.v2" +EVIDENCE_VALIDATION_INPUT_SCHEMA = "openultrasast.detection-core.evidence-validation-input.v2" +FACT_SET_SCHEMA = "openultrasast.detection-core.fact-set.v2" + + +class EvidenceInputValidationError(ValueError): + """A raw evidence input cannot be reproduced in its closed context.""" + + +@dataclass(frozen=True) +class EvidenceValidationContext: + family_catalog: FamilyCatalog + fact_packs: tuple[FactPack, ...] + model_programs: tuple[ModelProgramDeclaration, ...] + lineages: tuple[LineageManifest, ...] + fact_review_policy: ReviewIndependencePolicy + provenance_trust_policy: ProvenanceTrustPolicy + custody_policy: PopulationCustodyPolicy + context_digest: str + + +@dataclass(frozen=True) +class EvidenceValidationInput: + record: EvidenceRecord + named_inputs: NamedInputs + model_request: ModelRequest | None + model_result: ModelResult | None + input_digest: str + + +def fact_set_bytes(packs: tuple[FactPack, ...]) -> bytes: + if not isinstance(packs, tuple) or not all(isinstance(pack, FactPack) for pack in packs): + raise EvidenceInputValidationError("fact packs must be a tuple of FactPack values") + return canonical.canonical_bytes( + { + "schema_version": FACT_SET_SCHEMA, + "fact_pack_digests": tuple(sorted(pack.normalized_content_digest for pack in packs)), + } + ) + + +def derive_fact_set_digest(packs: tuple[FactPack, ...]) -> str: + return canonical.bytes_digest(fact_set_bytes(packs)) + + +def derive_evidence_validation_context_digest(context: EvidenceValidationContext) -> str: + packs = tuple( + sorted( + context.fact_packs, + key=lambda pack: ( + pack.normalized_content_digest, + pack.review_evidence_digest, + pack.review_policy_digest, + pack.review_identity, + ), + ) + ) + return canonical.content_digest( + { + "schema_version": EVIDENCE_VALIDATION_CONTEXT_SCHEMA, + "family_catalog_digest": context.family_catalog.catalog_digest, + "fact_pack_review_bindings": tuple( + { + "normalized_content_digest": pack.normalized_content_digest, + "review_evidence_digest": pack.review_evidence_digest, + "review_policy_digest": pack.review_policy_digest, + "review_identity": pack.review_identity, + } + for pack in packs + ), + "model_program_digests": tuple(sorted(program.declaration_digest for program in context.model_programs)), + "lineage_manifest_digests": tuple(sorted(lineage.manifest_digest for lineage in context.lineages)), + "fact_review_policy_digest": context.fact_review_policy.policy_digest, + "provenance_trust_policy_digest": context.provenance_trust_policy.policy_digest, + "custody_policy_digest": context.custody_policy.policy_digest, + } + ) + + +def make_evidence_validation_context( + *, + family_catalog: FamilyCatalog, + fact_packs: tuple[FactPack, ...], + model_programs: tuple[ModelProgramDeclaration, ...], + lineages: tuple[LineageManifest, ...], + fact_review_policy: ReviewIndependencePolicy, + provenance_trust_policy: ProvenanceTrustPolicy, + custody_policy: PopulationCustodyPolicy, +) -> EvidenceValidationContext: + provisional = EvidenceValidationContext( + family_catalog, + fact_packs, + model_programs, + lineages, + fact_review_policy, + provenance_trust_policy, + custody_policy, + "0" * 64, + ) + return EvidenceValidationContext( + family_catalog, + fact_packs, + model_programs, + lineages, + fact_review_policy, + provenance_trust_policy, + custody_policy, + derive_evidence_validation_context_digest(provisional), + ) + + +def derive_evidence_validation_input_digest(value: EvidenceValidationInput) -> str: + return canonical.content_digest( + { + "schema_version": EVIDENCE_VALIDATION_INPUT_SCHEMA, + "record_id": value.record.record_id, + "named_inputs_digest": value.named_inputs.manifest_digest, + "model_request_digest": None if value.model_request is None else value.model_request.request_id, + "model_result_digest": None if value.model_result is None else value.model_result.result_digest, + } + ) + + +def make_evidence_validation_input( + *, + record: EvidenceRecord, + named_inputs: NamedInputs, + model_request: ModelRequest | None = None, + model_result: ModelResult | None = None, +) -> EvidenceValidationInput: + provisional = EvidenceValidationInput(record, named_inputs, model_request, model_result, "0" * 64) + return EvidenceValidationInput( + record, + named_inputs, + model_request, + model_result, + derive_evidence_validation_input_digest(provisional), + ) + + +def _unique(values: tuple[str, ...], field: str) -> None: + if len(set(values)) != len(values): + raise EvidenceInputValidationError(f"duplicate {field} identity") + + +def _validate_context_collections(context: EvidenceValidationContext) -> None: + collections = ( + (context.fact_packs, FactPack, "fact_packs"), + (context.model_programs, ModelProgramDeclaration, "model_programs"), + (context.lineages, LineageManifest, "lineages"), + ) + for values, expected, field in collections: + if not isinstance(values, tuple) or not all(isinstance(item, expected) for item in values): + raise EvidenceInputValidationError(f"{field} must be a tuple of {expected.__name__} values") + _unique(tuple(pack.normalized_content_digest for pack in context.fact_packs), "fact pack") + _unique(tuple(program.declaration_digest for program in context.model_programs), "model program") + _unique(tuple(lineage.manifest_digest for lineage in context.lineages), "lineage manifest") + + +def _validate_context_types(context: EvidenceValidationContext) -> None: + fields = ( + (context.family_catalog, FamilyCatalog, "family_catalog"), + (context.fact_review_policy, ReviewIndependencePolicy, "fact_review_policy"), + (context.provenance_trust_policy, ProvenanceTrustPolicy, "provenance_trust_policy"), + (context.custody_policy, PopulationCustodyPolicy, "custody_policy"), + ) + for value, expected, field in fields: + if not isinstance(value, expected): + raise EvidenceInputValidationError(f"{field} must be a {expected.__name__}") + _validate_context_collections(context) + + +def _validate_context_policies(context: EvidenceValidationContext) -> None: + review = context.fact_review_policy + review.__post_init__() + expected_review = derive_review_independence_policy_digest( + review.reviewed_reviewer_digests, + review.forbidden_identity_pairs, + review.trusted_store_identities, + review.maximum_evidence_bytes, + ) + if review.policy_digest != expected_review: + raise EvidenceInputValidationError("fact review policy digest cannot be re-derived") + trust = context.provenance_trust_policy + trust.__post_init__() + if trust.policy_digest != derive_trust_policy_digest(trust.trusted_store_identities, trust.maximum_input_bytes): + raise EvidenceInputValidationError("provenance trust policy digest cannot be re-derived") + custody = context.custody_policy + custody.__post_init__() + if custody.policy_digest != derive_population_custody_policy_digest(custody.reviewed_custodian_digests): + raise EvidenceInputValidationError("custody policy digest cannot be re-derived") + + +def _validate_context_members(context: EvidenceValidationContext) -> None: + catalog_digests = {entry.contract_digest for entry in context.family_catalog.entries} + lineage_digests = {lineage.manifest_digest for lineage in context.lineages} + for lineage in context.lineages: + expected = derive_lineage_manifest_digest(lineage.arm_id, lineage.subject_digest, lineage.uses) + if lineage.manifest_digest != expected: + raise EvidenceInputValidationError("lineage manifest digest cannot be re-derived") + for pack in context.fact_packs: + pack.__post_init__() + if pack.normalized_content_digest != derive_fact_pack_content_digest(pack): + raise EvidenceInputValidationError("fact pack content digest cannot be re-derived") + if pack.review_evidence_digest != derive_review_evidence_digest(pack.review_evidence): + raise EvidenceInputValidationError("fact pack review evidence digest cannot be re-derived") + for program in context.model_programs: + program.__post_init__() + if program.declaration_digest != derive_model_program_declaration_digest(program): + raise EvidenceInputValidationError("model declaration digest cannot be re-derived") + if program.family_contract_digest not in catalog_digests: + raise EvidenceInputValidationError("model declaration family contract is absent from the catalog") + if program.lineage_manifest_digest not in lineage_digests: + raise EvidenceInputValidationError("model declaration lineage is absent from the context") + + +def _validate_context(context: EvidenceValidationContext) -> None: + if not isinstance(context, EvidenceValidationContext): + raise EvidenceInputValidationError("context must be an EvidenceValidationContext") + _validate_context_types(context) + expected = derive_evidence_validation_context_digest(context) + if context.context_digest != expected: + raise EvidenceInputValidationError("context digest cannot be re-derived") + validate_catalog(context.family_catalog) + _validate_context_policies(context) + _validate_context_members(context) + + +def _validate_record_identity(record: EvidenceRecord) -> None: + if not isinstance(record, EvidenceRecord): + raise EvidenceInputValidationError("record must be an EvidenceRecord") + record.packet.__post_init__() + record.decision.__post_init__() + expected = derive_record_id(record.packet, record.decision) + if record.record_id != expected: + raise EvidenceInputValidationError("record identity cannot be re-derived") + record.__post_init__() + + +def _validate_raw_deterministic_input(value: EvidenceValidationInput) -> None: + if value.model_request is not None or value.model_result is not None: + raise EvidenceInputValidationError("model handoff replay belongs to task 7.2") + decision = value.record.decision + if decision.basis is not EvidenceBasis.DETERMINISTIC: + raise EvidenceInputValidationError("raw validation accepts deterministic records only before model or proof replay") + provenance_values = tuple(getattr(decision.provenance, item.name) for item in fields(decision.provenance)) + if any(item is not None for item in provenance_values): + raise EvidenceInputValidationError("deterministic record contains model, proof, or lineage provenance") + + +def _validate_input_types(value: EvidenceValidationInput) -> None: + if not isinstance(value.record, EvidenceRecord): + raise EvidenceInputValidationError("record must be an EvidenceRecord") + if not isinstance(value.named_inputs, NamedInputs): + raise EvidenceInputValidationError("named_inputs must be NamedInputs") + if value.model_request is not None and not isinstance(value.model_request, ModelRequest): + raise EvidenceInputValidationError("model_request must be a ModelRequest or None") + if value.model_result is not None and not isinstance(value.model_result, ModelResult): + raise EvidenceInputValidationError("model_result must be a ModelResult or None") + + +def _validate_packs( + context: EvidenceValidationContext, + population_digest: str, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> None: + for pack in context.fact_packs: + contract = resolve_contract( + context.family_catalog, + pack.family_id, + pack.family_contract_digest, + pack.ecosystem, + ) + validate_fact_pack( + pack, + contract, + population_digest, + context.lineages, + custody, + context.custody_policy, + context.fact_review_policy, + resolver, + ) + + +def validate_evidence_input( + value: EvidenceValidationInput, + context: EvidenceValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> EvidenceRecord: + try: + if not isinstance(value, EvidenceValidationInput): + raise EvidenceInputValidationError("value must be an EvidenceValidationInput") + _validate_context(context) + _validate_input_types(value) + if value.input_digest != derive_evidence_validation_input_digest(value): + raise EvidenceInputValidationError("input digest cannot be re-derived") + _validate_record_identity(value.record) + _validate_raw_deterministic_input(value) + record = value.record + contract = resolve_contract( + context.family_catalog, + record.packet.family_id, + record.packet.family_contract_digest, + record.packet.ecosystem, + ) + validate_record(contract, record) + validate_provenance(record.packet.provenance, value.named_inputs, resolver, context.provenance_trust_policy) + if record.packet.provenance.fact_set_digest != derive_fact_set_digest(context.fact_packs): + raise EvidenceInputValidationError("record fact set does not match the validation context") + population = record.packet.provenance.population_digest + validate_population_custody_log(custody.current_log(population), context.custody_policy, current=custody) + _validate_packs(context, population, custody, resolver) + validate_fact_relations( + record, + contract, + context.fact_packs, + population, + context.lineages, + custody, + context.custody_policy, + context.fact_review_policy, + resolver, + ) + return record + except EvidenceInputValidationError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + message = str(exc) + if "provenance" not in message and exc.__class__.__module__.endswith("provenance"): + message = f"provenance validation failed: {message}" + raise EvidenceInputValidationError(f"evidence input validation failed: {message}") from exc + + +__all__ = [ + "EVIDENCE_VALIDATION_CONTEXT_SCHEMA", + "EVIDENCE_VALIDATION_INPUT_SCHEMA", + "FACT_SET_SCHEMA", + "EvidenceInputValidationError", + "EvidenceValidationContext", + "EvidenceValidationInput", + "derive_evidence_validation_context_digest", + "derive_evidence_validation_input_digest", + "derive_fact_set_digest", + "fact_set_bytes", + "make_evidence_validation_context", + "make_evidence_validation_input", + "validate_evidence_input", +] diff --git a/tests/test_core_validation_v2.py b/tests/test_core_validation_v2.py new file mode 100644 index 0000000..2235f25 --- /dev/null +++ b/tests/test_core_validation_v2.py @@ -0,0 +1,299 @@ +"""Closed evidence revalidation seam (Reqs 4.4, 10.1-10.4, 12.6, 14.1).""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest +from detection_core_fixtures import InMemoryContentResolver, InMemoryPopulationCustody, custody_contract, stored_artifact +from test_core_factapply import _append_use as append_fact_use +from test_core_factapply import _apply as apply_fact_case +from test_core_factapply import _case as fact_case +from test_core_factapply import _manifest as fact_manifest +from test_core_factapply import _record as fact_record +from test_core_family import _contract as family_contract +from test_core_family import _record as family_record +from test_core_model import _case as model_case + +from openultrasast.core import canonical, evidence, family, provenance, validation +from openultrasast.core.factpack import ( + FactPack, + ReviewIndependencePolicy, + derive_review_evidence_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.family_catalog import FamilyCatalog, FamilyCatalogEntry, derive_catalog_digest + + +class CountingResolver: + def __init__(self, delegate: InMemoryContentResolver) -> None: + self.delegate = delegate + self.resolved: list[str] = [] + + def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent: + self.resolved.append(reference.artifact_id) + return self.delegate.resolve(reference) + + +def _review_policy() -> ReviewIndependencePolicy: + reviewer = canonical.content_digest("reviewer") + stores = ("trusted-store",) + digest = derive_review_independence_policy_digest((reviewer,), (), stores, 4096) + return ReviewIndependencePolicy((reviewer,), (), stores, 4096, digest) + + +def _trust_policy() -> provenance.ProvenanceTrustPolicy: + stores = ("trusted-store",) + maximum = 4096 + return provenance.ProvenanceTrustPolicy(stores, maximum, provenance.derive_trust_policy_digest(stores, maximum)) + + +def _catalog(contract: family.FamilyContract) -> FamilyCatalog: + entry = FamilyCatalogEntry(contract, family.family_contract_digest(contract)) + entries = (entry,) + return FamilyCatalog(entries, derive_catalog_digest(entries)) + + +def _with_fact_set(record: evidence.EvidenceRecord, packs: tuple[FactPack, ...]) -> evidence.EvidenceRecord: + digest = validation.derive_fact_set_digest(packs) + stamp = replace(record.packet.provenance, fact_set_digest=digest) + packet = replace(record.packet, provenance=stamp) + decision = replace(record.decision, packet_digest=evidence.derive_packet_digest(packet)) + return evidence.EvidenceRecord(packet, decision, "") + + +def _named_inputs( + record: evidence.EvidenceRecord, + contract: family.FamilyContract, + packs: tuple[FactPack, ...], + *, + fact_set_content: bytes | None = None, +) -> tuple[provenance.NamedInputs, InMemoryContentResolver]: + def matching_preimage(identity: str, *values: str) -> bytes: + matches = tuple(canonical.canonical_bytes(value) for value in values if canonical.content_digest(value) == identity) + assert len(matches) == 1 + return matches[0] + + stamp = record.packet.provenance + preimages: dict[provenance.ProvenanceInputRole, bytes] = { + provenance.ProvenanceInputRole.ANALYZER: matching_preimage(stamp.analyzer_revision, "analyzer"), + provenance.ProvenanceInputRole.FAMILY_CONTRACT: canonical.canonical_bytes(family._contract_value(contract)), + provenance.ProvenanceInputRole.FACT_SET: fact_set_content or validation.fact_set_bytes(packs), + provenance.ProvenanceInputRole.ACQUISITION: matching_preimage(stamp.acquisition_record_digest, "acquisition"), + provenance.ProvenanceInputRole.CONFIGURATION: matching_preimage(stamp.configuration_digest, "configuration"), + provenance.ProvenanceInputRole.POPULATION: matching_preimage(stamp.population_digest, "population"), + provenance.ProvenanceInputRole.CHANGE_INPUT: matching_preimage(stamp.change_input_digest, "change", "change-input"), + } + items: list[provenance.NamedInput] = [] + artifacts: dict[str, canonical.ResolvedContent] = {} + fields = { + provenance.ProvenanceInputRole.ANALYZER: stamp.analyzer_revision, + provenance.ProvenanceInputRole.FAMILY_CONTRACT: stamp.family_contract_digest, + provenance.ProvenanceInputRole.FACT_SET: stamp.fact_set_digest, + provenance.ProvenanceInputRole.ACQUISITION: stamp.acquisition_record_digest, + provenance.ProvenanceInputRole.CONFIGURATION: stamp.configuration_digest, + provenance.ProvenanceInputRole.POPULATION: stamp.population_digest, + provenance.ProvenanceInputRole.CHANGE_INPUT: stamp.change_input_digest, + } + for role in provenance.ProvenanceInputRole: + reference, resolver = stored_artifact(preimages[role], artifact_id=f"named-{role.value}") + assert reference.byte_length > 0 and reference.content_digest == fields[role] + items.append(provenance.NamedInput(role, fields[role], reference)) + artifacts.update(resolver.artifacts) + values = tuple(items) + return provenance.NamedInputs(values, provenance.derive_named_inputs_digest(values)), InMemoryContentResolver(artifacts) + + +def _deterministic_case() -> tuple[ + validation.EvidenceValidationInput, + validation.EvidenceValidationContext, + InMemoryPopulationCustody, + InMemoryContentResolver, +]: + contract = family_contract() + record = _with_fact_set(family_record(contract), ()) + named, resolver = _named_inputs(record, contract, ()) + initial, custody_policy, _ = custody_contract() + context = validation.make_evidence_validation_context( + family_catalog=_catalog(contract), + fact_packs=(), + model_programs=(), + lineages=(), + fact_review_policy=_review_policy(), + provenance_trust_policy=_trust_policy(), + custody_policy=custody_policy, + ) + return ( + validation.make_evidence_validation_input(record=record, named_inputs=named), + context, + InMemoryPopulationCustody(initial, custody_policy), + resolver, + ) + + +def test_shared_context_revalidates_record_and_all_seven_named_inputs() -> None: + value, context, custody, resolver = _deterministic_case() + counting = CountingResolver(resolver) + + accepted = validation.validate_evidence_input(value, context, custody, counting) + + assert accepted is value.record + assert set(counting.resolved) == {item.content.artifact_id for item in value.named_inputs.inputs} + assert len(counting.resolved) == len(provenance.ProvenanceInputRole) == 7 + + +def test_context_input_and_record_identities_are_rederived() -> None: + value, context, custody, resolver = _deterministic_case() + with pytest.raises(validation.EvidenceInputValidationError, match="context digest"): + validation.validate_evidence_input(value, replace(context, context_digest="0" * 64), custody, resolver) + with pytest.raises(validation.EvidenceInputValidationError, match="input digest"): + validation.validate_evidence_input(replace(value, input_digest="0" * 64), context, custody, resolver) + + forged = object.__new__(evidence.EvidenceRecord) + object.__setattr__(forged, "packet", value.record.packet) + object.__setattr__(forged, "decision", value.record.decision) + object.__setattr__(forged, "record_id", "0" * 64) + forged_value = validation.make_evidence_validation_input(record=forged, named_inputs=value.named_inputs) + with pytest.raises(validation.EvidenceInputValidationError, match="record"): + validation.validate_evidence_input(forged_value, context, custody, resolver) + + changed_contract = replace(family_contract(), family_version="2.0.0") + mismatched = validation.make_evidence_validation_context( + family_catalog=_catalog(changed_contract), + fact_packs=context.fact_packs, + model_programs=context.model_programs, + lineages=context.lineages, + fact_review_policy=context.fact_review_policy, + provenance_trust_policy=context.provenance_trust_policy, + custody_policy=context.custody_policy, + ) + with pytest.raises(validation.EvidenceInputValidationError, match="family, digest, and ecosystem"): + validation.validate_evidence_input(value, mismatched, custody, resolver) + + +def test_context_fact_set_and_named_inputs_cannot_be_substituted() -> None: + value, context, custody, resolver = _deterministic_case() + changed = replace(value.record.packet.provenance, fact_set_digest=canonical.content_digest("other-facts")) + packet = replace(value.record.packet, provenance=changed) + decision = replace(value.record.decision, packet_digest=evidence.derive_packet_digest(packet)) + record = evidence.EvidenceRecord(packet, decision, "") + named, changed_resolver = _named_inputs( + record, + family_contract(), + (), + fact_set_content=canonical.canonical_bytes("other-facts"), + ) + substituted = validation.make_evidence_validation_input(record=record, named_inputs=named) + with pytest.raises(validation.EvidenceInputValidationError, match="fact set"): + validation.validate_evidence_input(substituted, context, custody, changed_resolver) + + first = value.named_inputs.inputs[0] + forged_named = replace(first, identity="0" * 64) + inputs = (forged_named, *value.named_inputs.inputs[1:]) + bad_named = provenance.NamedInputs(inputs, provenance.derive_named_inputs_digest(inputs)) + bad_value = validation.make_evidence_validation_input(record=value.record, named_inputs=bad_named) + with pytest.raises(validation.EvidenceInputValidationError, match="provenance"): + validation.validate_evidence_input(bad_value, context, custody, resolver) + + +def test_fact_derived_record_is_replayed_from_context_pack_and_match_bytes() -> None: + case = fact_case(fact_only=True) + (relation,) = apply_fact_case(case) + record = _with_fact_set(fact_record(case, relation), (case.pack,)) + named, named_resolver = _named_inputs(record, case.contract, (case.pack,)) + resolver = InMemoryContentResolver({**case.resolver.artifacts, **named_resolver.artifacts}) + context = validation.make_evidence_validation_context( + family_catalog=_catalog(case.contract), + fact_packs=(case.pack,), + model_programs=(), + lineages=case.lineages, + fact_review_policy=case.review_policy, + provenance_trust_policy=_trust_policy(), + custody_policy=case.custody_policy, + ) + value = validation.make_evidence_validation_input(record=record, named_inputs=named) + + assert validation.validate_evidence_input(value, context, case.custody, resolver) is record + + missing_pack = validation.make_evidence_validation_context( + family_catalog=context.family_catalog, + fact_packs=(), + model_programs=(), + lineages=context.lineages, + fact_review_policy=context.fact_review_policy, + provenance_trust_policy=context.provenance_trust_policy, + custody_policy=context.custody_policy, + ) + with pytest.raises(validation.EvidenceInputValidationError, match="fact set|semantic pack"): + validation.validate_evidence_input(value, missing_pack, case.custody, resolver) + + +def test_context_binds_exact_model_declarations_and_their_lineages() -> None: + handoff = model_case() + record = _with_fact_set(handoff.record, ()) + named, resolver = _named_inputs(record, handoff.contract, ()) + initial, custody_policy, _ = custody_contract() + context = validation.make_evidence_validation_context( + family_catalog=_catalog(handoff.contract), + fact_packs=(), + model_programs=(handoff.request.program,), + lineages=(handoff.request.lineage,), + fact_review_policy=_review_policy(), + provenance_trust_policy=_trust_policy(), + custody_policy=custody_policy, + ) + value = validation.make_evidence_validation_input(record=record, named_inputs=named) + custody = InMemoryPopulationCustody(initial, custody_policy) + + assert validation.validate_evidence_input(value, context, custody, resolver) is record + + missing_lineage = validation.make_evidence_validation_context( + family_catalog=context.family_catalog, + fact_packs=(), + model_programs=context.model_programs, + lineages=(), + fact_review_policy=context.fact_review_policy, + provenance_trust_policy=context.provenance_trust_policy, + custody_policy=context.custody_policy, + ) + with pytest.raises(validation.EvidenceInputValidationError, match="model declaration lineage"): + validation.validate_evidence_input(value, missing_lineage, custody, resolver) + + +def test_context_digest_binds_pack_review_evidence_to_its_exact_lineage() -> None: + case = fact_case(fact_only=True) + alternate_use = append_fact_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + provenance.LineagePurpose.SEMANTIC_FACT, + case.pack.normalized_content_digest, + labels_accessed=False, + arm_id="alternate-fact-arm", + ) + alternate = fact_manifest("alternate-fact-arm", case.pack.normalized_content_digest, (alternate_use,)) + lineages = (*case.lineages, alternate) + context = validation.make_evidence_validation_context( + family_catalog=_catalog(case.contract), + fact_packs=(case.pack,), + model_programs=(), + lineages=lineages, + fact_review_policy=case.review_policy, + provenance_trust_policy=_trust_policy(), + custody_policy=case.custody_policy, + ) + changed_evidence = replace(case.pack.review_evidence[0], lineage_manifest_digest=alternate.manifest_digest) + swapped = replace( + case.pack, + review_evidence=(changed_evidence,), + review_evidence_digest=derive_review_evidence_digest((changed_evidence,)), + ) + substituted = replace(context, fact_packs=(swapped,)) + (relation,) = apply_fact_case(case) + record = _with_fact_set(fact_record(case, relation), (case.pack,)) + named, named_resolver = _named_inputs(record, case.contract, (case.pack,)) + resolver = InMemoryContentResolver({**case.resolver.artifacts, **named_resolver.artifacts}) + value = validation.make_evidence_validation_input(record=record, named_inputs=named) + + assert validation.derive_evidence_validation_context_digest(substituted) != context.context_digest + with pytest.raises(validation.EvidenceInputValidationError, match="context digest"): + validation.validate_evidence_input(value, substituted, case.custody, resolver)