feat(detection-core): replay model derivation in validation

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

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