From a2788d11b0e1cdbf8f3d708c698e024f5aa1285e Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:30:40 +0100 Subject: [PATCH] feat(detection-core): replay model derivation in validation --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/_validation_model.py | 79 ++++++++++++++ src/openultrasast/core/validation.py | 28 ++--- tests/test_core_validation_v2.py | 114 +++++++++++++++++++- 4 files changed, 207 insertions(+), 16 deletions(-) create mode 100644 src/openultrasast/core/_validation_model.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 94f5b3c..6f05abe 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -187,7 +187,7 @@ - Re-run family and fact validation and reject embedded pack, lineage, or named-input substitutes. - Done when raw, forged, and context-mismatched records cannot reach admission or consumer decision code. - _Requirements: 4.4, 10.1, 10.2, 10.4, 12.6, 14.1_ -- [ ] 7.2 Replay model derivation at the shared validation seam +- [x] 7.2 Replay model derivation at the shared validation seam - Require deterministic records to omit model derivation and model-corroborated records to include both request and result. - Re-run model application from the pre-model record and compare the derived record canonically. - Refuse constructible “validated” markers or model programs outside the context. diff --git a/src/openultrasast/core/_validation_model.py b/src/openultrasast/core/_validation_model.py new file mode 100644 index 0000000..e24b606 --- /dev/null +++ b/src/openultrasast/core/_validation_model.py @@ -0,0 +1,79 @@ +"""Model-derivation replay for the closed evidence validation seam.""" + +from __future__ import annotations + +from dataclasses import fields + +from openultrasast.core import canonical +from openultrasast.core.custody import PopulationCustodyPolicy, PopulationCustodyPort +from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, evidence_record_bytes +from openultrasast.core.family import FamilyContract +from openultrasast.core.lineage import LineageManifest +from openultrasast.core.model import ModelProgramDeclaration, ModelRequest, ModelResult +from openultrasast.core.model_apply import apply_model_result + + +class ModelReplayValidationError(ValueError): + """A model-derived record cannot be reproduced from its closed context.""" + + +def _require_exact_program( + request: ModelRequest, + programs: tuple[ModelProgramDeclaration, ...], +) -> None: + matches = tuple(item for item in programs if item.declaration_digest == request.program.declaration_digest) + if len(matches) != 1 or matches[0] != request.program: + raise ModelReplayValidationError("model declaration is not an exact validation-context member") + + +def _require_exact_lineage(request: ModelRequest, lineages: tuple[LineageManifest, ...]) -> None: + matches = tuple(item for item in lineages if item.manifest_digest == request.lineage.manifest_digest) + if len(matches) != 1 or matches[0] != request.lineage: + raise ModelReplayValidationError("model lineage is not an exact validation-context member") + + +def _require_deterministic_shape(record: EvidenceRecord) -> None: + if record.decision.basis is not EvidenceBasis.DETERMINISTIC: + raise ModelReplayValidationError("model request must start from a deterministic record") + provenance = record.decision.provenance + if any(getattr(provenance, item.name) is not None for item in fields(provenance)): + raise ModelReplayValidationError("deterministic record contains model, proof, or lineage provenance") + + +def _same_canonical_record(expected: EvidenceRecord, supplied: EvidenceRecord) -> bool: + return expected.record_id == supplied.record_id and evidence_record_bytes(expected) == evidence_record_bytes(supplied) + + +def validate_and_replay_model_derivation( + supplied: EvidenceRecord, + request: ModelRequest | None, + result: ModelResult | None, + contract: FamilyContract, + programs: tuple[ModelProgramDeclaration, ...], + lineages: tuple[LineageManifest, ...], + resolver: canonical.ContentResolver, + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, +) -> EvidenceRecord: + """Validate the envelope shape and replay model corroboration when present.""" + if supplied.decision.basis is EvidenceBasis.DETERMINISTIC: + if request is not None or result is not None: + raise ModelReplayValidationError("deterministic records must omit model request and result") + _require_deterministic_shape(supplied) + return supplied + if supplied.decision.basis is not EvidenceBasis.MODEL_CORROBORATED: + raise ModelReplayValidationError("execution-backed records require later proof replay") + if request is None or result is None: + raise ModelReplayValidationError("model-corroborated records require both request and result") + _require_deterministic_shape(request.record) + _require_exact_program(request, programs) + _require_exact_lineage(request, lineages) + derived = apply_model_result(request.record, contract, request, result, resolver, custody, custody_policy) + if derived.decision.basis is not EvidenceBasis.MODEL_CORROBORATED: + raise ModelReplayValidationError("model result did not derive a model-corroborated record") + if not _same_canonical_record(derived, supplied): + raise ModelReplayValidationError("model-derived record does not match canonical replay") + return supplied + + +__all__ = ["ModelReplayValidationError", "validate_and_replay_model_derivation"] diff --git a/src/openultrasast/core/validation.py b/src/openultrasast/core/validation.py index 868ba67..3e7939a 100644 --- a/src/openultrasast/core/validation.py +++ b/src/openultrasast/core/validation.py @@ -2,16 +2,17 @@ from __future__ import annotations -from dataclasses import dataclass, fields +from dataclasses import dataclass from openultrasast.core import canonical +from openultrasast.core._validation_model import validate_and_replay_model_derivation 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.evidence import EvidenceRecord, derive_record_id from openultrasast.core.factpack import ( FactPack, ReviewIndependencePolicy, @@ -276,17 +277,6 @@ def _validate_record_identity(record: EvidenceRecord) -> None: 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") @@ -337,7 +327,6 @@ def validate_evidence_input( 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, @@ -345,6 +334,17 @@ def validate_evidence_input( record.packet.family_contract_digest, record.packet.ecosystem, ) + record = validate_and_replay_model_derivation( + record, + value.model_request, + value.model_result, + contract, + context.model_programs, + context.lineages, + resolver, + custody, + context.custody_policy, + ) 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): diff --git a/tests/test_core_validation_v2.py b/tests/test_core_validation_v2.py index 2235f25..2224e54 100644 --- a/tests/test_core_validation_v2.py +++ b/tests/test_core_validation_v2.py @@ -13,9 +13,12 @@ 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 _application_case as model_application_case from test_core_model import _case as model_case +from test_core_model import _contract as model_contract +from test_core_model import _record as model_record -from openultrasast.core import canonical, evidence, family, provenance, validation +from openultrasast.core import canonical, evidence, family, model_apply, provenance, validation from openultrasast.core.factpack import ( FactPack, ReviewIndependencePolicy, @@ -131,6 +134,45 @@ def _deterministic_case() -> tuple[ ) +def _model_validation_case() -> tuple[ + validation.EvidenceValidationInput, + validation.EvidenceValidationContext, + InMemoryPopulationCustody, + InMemoryContentResolver, + evidence.EvidenceRecord, +]: + contract = model_contract() + deterministic = _with_fact_set(model_record(contract), ()) + handoff = model_application_case(contract=contract, record=deterministic) + named, named_resolver = _named_inputs(deterministic, contract, ()) + resolver = InMemoryContentResolver({**handoff.resolver.artifacts, **named_resolver.artifacts}) + context = validation.make_evidence_validation_context( + family_catalog=_catalog(contract), + fact_packs=(), + model_programs=(handoff.request.program,), + lineages=(handoff.request.lineage,), + fact_review_policy=_review_policy(), + provenance_trust_policy=_trust_policy(), + custody_policy=handoff.custody_policy, + ) + derived = model_apply.apply_model_result( + deterministic, + contract, + handoff.request, + handoff.result, + resolver, + handoff.custody, + handoff.custody_policy, + ) + value = validation.make_evidence_validation_input( + record=derived, + named_inputs=named, + model_request=handoff.request, + model_result=handoff.result, + ) + return value, context, handoff.custody, resolver, deterministic + + def test_shared_context_revalidates_record_and_all_seven_named_inputs() -> None: value, context, custody, resolver = _deterministic_case() counting = CountingResolver(resolver) @@ -297,3 +339,73 @@ def test_context_digest_binds_pack_review_evidence_to_its_exact_lineage() -> Non 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) + + +def test_shared_validator_replays_model_derivation_from_the_deterministic_record() -> None: + value, context, custody, resolver, deterministic = _model_validation_case() + + accepted = validation.validate_evidence_input(value, context, custody, resolver) + + assert value.model_request is not None + assert accepted is value.record + assert accepted.packet is value.model_request.record.packet + assert value.model_request.record is deterministic + assert accepted.decision.basis is evidence.EvidenceBasis.MODEL_CORROBORATED + + +def test_deterministic_and_model_records_require_their_exact_handoff_shape() -> None: + value, context, custody, resolver, _ = _model_validation_case() + for request, result in ((value.model_request, None), (None, value.model_result)): + incomplete_handoff = validation.make_evidence_validation_input( + record=value.record, + named_inputs=value.named_inputs, + model_request=request, + model_result=result, + ) + with pytest.raises(validation.EvidenceInputValidationError, match="request and result|both"): + validation.validate_evidence_input(incomplete_handoff, context, custody, resolver) + + deterministic, det_context, det_custody, det_resolver = _deterministic_case() + attached = validation.make_evidence_validation_input( + record=deterministic.record, + named_inputs=deterministic.named_inputs, + model_request=value.model_request, + model_result=value.model_result, + ) + with pytest.raises(validation.EvidenceInputValidationError, match="deterministic.*omit|forbids model"): + validation.validate_evidence_input(attached, det_context, det_custody, det_resolver) + + +def test_model_replay_refuses_programs_outside_the_closed_context() -> None: + value, context, custody, resolver, _ = _model_validation_case() + without_program = validation.make_evidence_validation_context( + family_catalog=context.family_catalog, + fact_packs=context.fact_packs, + 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|program.*context"): + validation.validate_evidence_input(value, without_program, custody, resolver) + + +def test_model_replay_compares_the_canonical_derived_record() -> None: + value, context, custody, resolver, _ = _model_validation_case() + changed_decision = replace( + value.record.decision, + family_code="safe", + outcome=evidence.NeutralOutcome.SATISFIED, + ) + substituted = evidence.EvidenceRecord(value.record.packet, changed_decision, "") + forged = validation.make_evidence_validation_input( + record=substituted, + named_inputs=value.named_inputs, + model_request=value.model_request, + model_result=value.model_result, + ) + + with pytest.raises(validation.EvidenceInputValidationError, match="derived record|replay"): + validation.validate_evidence_input(forged, context, custody, resolver)