From a1800c0c74eaa2644dde6a00725ae94a81dda102 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 16:57:11 +0100 Subject: [PATCH] feat(detection-core): apply bounded model corroboration --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/model_apply.py | 197 ++++++++++++++++ tests/test_core_model.py | 321 +++++++++++++++++++++++++- 3 files changed, 516 insertions(+), 4 deletions(-) create mode 100644 src/openultrasast/core/model_apply.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 0adced5..d088c91 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -163,7 +163,7 @@ - _Requirements: 9.1, 9.2, 10.3_ - _Boundary: Model Handoff_ - _Depends: 2.4, 3.1_ -- [ ] 6.2 Apply model results without rewriting deterministic evidence +- [x] 6.2 Apply model results without rewriting deterministic evidence - Validate current prompt/example/fitted-model/mapping lineage against the evaluated population and declared artifacts. - Reject proposer/adjudicator role collision and results for another packet, family, residual, or mapping. - Allow model-corroborated output only when deterministic residual prerequisites are complete. diff --git a/src/openultrasast/core/model_apply.py b/src/openultrasast/core/model_apply.py new file mode 100644 index 0000000..e7f6c58 --- /dev/null +++ b/src/openultrasast/core/model_apply.py @@ -0,0 +1,197 @@ +"""Apply validated model results without mutating deterministic evidence.""" + +from __future__ import annotations + +from openultrasast.core._model_common import ModelContractError +from openultrasast.core._model_result import ModelResult, ModelStatus +from openultrasast.core.canonical import ContentResolver +from openultrasast.core.custody import ( + LineagePurpose, + PopulationCustodyError, + PopulationCustodyPolicy, + PopulationCustodyPort, + PopulationUseReceipt, + validate_population_custody_log, +) +from openultrasast.core.evidence import ( + Decision, + EvidenceBasis, + EvidenceRecord, + NeutralOutcome, + derive_packet_digest, +) +from openultrasast.core.family import FamilyContract, family_contract_digest, validate_record +from openultrasast.core.lineage import LineageExpectation, LineageValidationError, validate_lineage +from openultrasast.core.model import ModelProgramDeclaration, ModelRequest, validate_model_result +from openultrasast.core.model import completeness_snapshot as _completeness_snapshot +from openultrasast.core.provenance import DecisionProvenance +from openultrasast.core.semantics import RelationStatus + + +def _expectations(program: ModelProgramDeclaration) -> tuple[LineageExpectation, ...]: + expected = [ + LineageExpectation(LineagePurpose.PROMPT, program.prompt_digest, 1, 1), + LineageExpectation(LineagePurpose.DECISION_MAPPING, program.decision_mapping_digest, 1, 1), + ] + if program.example_set_digest is not None: + expected.append(LineageExpectation(LineagePurpose.EXAMPLE, program.example_set_digest, 1, 1)) + if program.fitted_component_digest is not None: + expected.append(LineageExpectation(LineagePurpose.FITTED_MODEL, program.fitted_component_digest, 1, 1)) + return tuple(expected) + + +def _expected_artifacts(program: ModelProgramDeclaration) -> set[tuple[LineagePurpose, str]]: + return {(item.purpose, item.influenced_artifact_digest) for item in _expectations(program)} + + +def _authoritative_receipts( + request: ModelRequest, + protected_population_digest: str, + custody: PopulationCustodyPort, + policy: PopulationCustodyPolicy, +) -> tuple[PopulationUseReceipt, ...]: + populations = {protected_population_digest, *(use.source_population_digest for use in request.lineage.uses)} + expected = _expected_artifacts(request.program) + receipts: list[PopulationUseReceipt] = [] + for population_digest in sorted(populations): + log = custody.current_log(population_digest) + validate_population_custody_log(log, policy, current=custody) + receipts.extend( + receipt + for receipt in log.receipts + if (receipt.request.purpose, receipt.request.influenced_artifact_digest) in expected + and (receipt.request.arm_id == request.lineage.arm_id or receipt.request.labels_accessed) + ) + return tuple(receipts) + + +def _refuse_omitted_uses( + request: ModelRequest, + receipts: tuple[PopulationUseReceipt, ...], + protected_population_digest: str, +) -> None: + declared = {use.population_use_receipt_digest for use in request.lineage.uses} + for receipt in receipts: + if receipt.receipt_digest in declared: + continue + population = receipt.request.population_digest + if receipt.request.labels_accessed: + reason = f"label_access_forbidden:{population}" + if population == protected_population_digest: + reason += f",protected_population_labels:{population}" + raise ModelContractError(f"model lineage is contaminated: {reason}") + raise ModelContractError(f"model lineage omitted current use: {receipt.receipt_digest}") + + +def _validate_program_lineage( + request: ModelRequest, + record: EvidenceRecord, + custody: PopulationCustodyPort, + policy: PopulationCustodyPolicy, +) -> None: + protected = record.packet.provenance.population_digest + try: + decision = validate_lineage( + request.lineage, + request.program.input_bundle_digest, + _expectations(request.program), + protected, + True, + custody, + policy, + ) + receipts = _authoritative_receipts(request, protected, custody, policy) + except (LineageValidationError, PopulationCustodyError, TypeError, ValueError) as exc: + raise ModelContractError(f"model lineage cannot be validated: {exc}") from exc + if not decision.allowed: + raise ModelContractError("model lineage is not allowed: " + ",".join(decision.reasons)) + _refuse_omitted_uses(request, receipts, protected) + + +def _validate_application_inputs( + record: EvidenceRecord, + contract: FamilyContract, + request: ModelRequest, + result: ModelResult, + resolver: ContentResolver, + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, +) -> None: + if not isinstance(record, EvidenceRecord) or not isinstance(contract, FamilyContract): + raise ModelContractError("record and contract must use detection-core contract types") + if record != request.record or contract != request.contract: + raise ModelContractError("record or family contract does not match the frozen model request") + if family_contract_digest(contract) != request.family_contract_digest: + raise ModelContractError("family contract identity does not match the frozen model request") + if request.custody_policy_digest != custody_policy.policy_digest: + raise ModelContractError("custody policy does not match the frozen model request") + if request.program.program_digest in request.proposal_program_digests: + raise ModelContractError("one program cannot propose and adjudicate the same evidence path") + try: + validate_record(contract, record) + validate_model_result(request, result, resolver) + except (TypeError, ValueError) as exc: + raise ModelContractError(f"model application input is invalid: {exc}") from exc + _validate_program_lineage(request, record, custody, custody_policy) + + +def _residual_complete(request: ModelRequest) -> bool: + statuses = {item.role: item.status for item in _completeness_snapshot(request.record, request.contract)} + return all(statuses.get(role) is RelationStatus.COMPLETE for role in request.residual_question.required_roles) + + +def _derived_record(record: EvidenceRecord, contract: FamilyContract, request: ModelRequest, result: ModelResult) -> EvidenceRecord: + if result.family_decision is None: + raise ModelContractError("successful model result requires a family decision") + rules = tuple(item for item in contract.decisions if item.family_code == result.family_decision) + if len(rules) != 1: + raise ModelContractError("model family decision does not resolve exactly once") + rule = rules[0] + if rule.outcome is NeutralOutcome.INDETERMINATE: + raise ModelContractError("model output cannot synthesize an indeterminate reason") + provenance = DecisionProvenance( + result.identity, + result.result_digest, + request.program.declaration_digest, + None, + None, + request.lineage.manifest_digest, + ) + decision = Decision( + result.family_decision, + rule.outcome, + EvidenceBasis.MODEL_CORROBORATED, + record.decision.bounded_scope, + (), + derive_packet_digest(record.packet), + provenance, + ) + derived = EvidenceRecord(record.packet, decision, "") + try: + validate_record(contract, derived) + except (TypeError, ValueError) as exc: + raise ModelContractError(f"model-derived record is invalid: {exc}") from exc + return derived + + +def apply_model_result( + record: EvidenceRecord, + contract: FamilyContract, + request: ModelRequest, + result: ModelResult, + resolver: ContentResolver, + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, +) -> EvidenceRecord: + """Return a corroborated decision over the exact immutable deterministic packet.""" + _validate_application_inputs(record, contract, request, result, resolver, custody, custody_policy) + if result.status is ModelStatus.FAILED: + return record + if not _residual_complete(request): + if record.decision.outcome is not NeutralOutcome.INDETERMINATE: + raise ModelContractError("incomplete model prerequisites require an indeterminate deterministic record") + return record + return _derived_record(record, contract, request, result) + + +__all__ = ["apply_model_result"] diff --git a/tests/test_core_model.py b/tests/test_core_model.py index dc6ca90..78c4b90 100644 --- a/tests/test_core_model.py +++ b/tests/test_core_model.py @@ -3,12 +3,17 @@ from __future__ import annotations import json -from dataclasses import dataclass +from dataclasses import dataclass, replace import pytest -from detection_core_fixtures import InMemoryContentResolver, stored_artifact +from detection_core_fixtures import ( + InMemoryContentResolver, + InMemoryPopulationCustody, + custody_contract, + stored_artifact, +) -from openultrasast.core import canonical, evidence, family, model, provenance, semantics +from openultrasast.core import canonical, evidence, family, model, model_apply, provenance, semantics def _digest(value: str) -> str: @@ -326,3 +331,313 @@ def test_core_model_contract_has_no_runtime_dependency_imports() -> None: assert "import dspy" not in source assert "litellm" not in source assert "openai" not in source + + +@dataclass(frozen=True) +class ApplicationCase: + contract: family.FamilyContract + record: evidence.EvidenceRecord + request: model.ModelRequest + result: model.ModelResult + resolver: InMemoryContentResolver + custody: InMemoryPopulationCustody + custody_policy: provenance.PopulationCustodyPolicy + custodian: str + + +def _append_model_use( + custody: InMemoryPopulationCustody, + policy: provenance.PopulationCustodyPolicy, + custodian: str, + *, + arm_id: str, + purpose: provenance.LineagePurpose, + artifact_digest: str, + labels_accessed: bool = False, +) -> provenance.LineageUse: + population = custody.current_log(_digest("population")) + request = provenance.PopulationUseRequest( + population.population_digest, + arm_id, + purpose, + labels_accessed, + artifact_digest, + population.log_digest, + ) + receipt = custody.append(request, custodian) + provenance.validate_population_custody_log(custody.current_log(population.population_digest), policy) + return provenance.LineageUse( + purpose, + population.population_digest, + labels_accessed, + artifact_digest, + receipt.receipt_digest, + ) + + +def _application_case( + *, + contract: family.FamilyContract | None = None, + record: evidence.EvidenceRecord | None = None, + failed: bool = False, +) -> ApplicationCase: + contract = contract or _contract() + record = record or _record(contract) + question = contract.residual_questions[0] + completeness = model.completeness_snapshot(record, contract) + bundle = model.derive_model_input_bundle_digest(record, contract, question, completeness) + initial, policy, custodian = custody_contract() + custody = InMemoryPopulationCustody(initial, policy) + artifacts = ( + (provenance.LineagePurpose.PROMPT, _digest("prompt")), + (provenance.LineagePurpose.EXAMPLE, _digest("examples")), + (provenance.LineagePurpose.FITTED_MODEL, _digest("fitted")), + (provenance.LineagePurpose.DECISION_MAPPING, _digest("mapping")), + ) + uses = tuple( + _append_model_use( + custody, + policy, + custodian, + arm_id="model-arm", + purpose=purpose, + artifact_digest=artifact, + ) + for purpose, artifact in artifacts + ) + lineage = provenance.LineageManifest( + "model-arm", + bundle, + uses, + provenance.derive_lineage_manifest_digest("model-arm", bundle, uses), + ) + program = model.make_model_program_declaration( + family_contract_digest=family.family_contract_digest(contract), + provider="provider", + model_name="model-v1", + program_digest=_digest("program"), + prompt_digest=_digest("prompt"), + example_set_digest=_digest("examples"), + fitted_component_digest=_digest("fitted"), + configuration_digest=_digest("model-configuration"), + adapter_revision="adapter-v1", + decision_mapping_digest=_digest("mapping"), + input_bundle_digest=bundle, + lineage_manifest_digest=lineage.manifest_digest, + ) + request = model.make_model_request( + record=record, + contract=contract, + residual_question_id=question.question_id, + program=program, + lineage=lineage, + custody_policy_digest=policy.policy_digest, + proposal_program_digests=(_digest("proposer"),), + trusted_response_store_identities=("trusted-store",), + maximum_response_bytes=256, + ) + response = model.make_preserved_model_response( + normalized_fields=(semantics.FieldValue("decision", semantics.FieldKind.TEXT, "unsafe"),), + inline_redacted_text="[redacted] unsafe", + artifact=None, + redaction_policy_digest=_digest("redaction-policy"), + ) + result = model.make_model_result( + request=request, + status=model.ModelStatus.FAILED if failed else model.ModelStatus.SUCCEEDED, + response=response, + family_decision=None if failed else "unsafe", + failure_reason="provider_timeout" if failed else None, + ) + return ApplicationCase(contract, record, request, result, InMemoryContentResolver({}), custody, policy, custodian) + + +def test_apply_model_result_preserves_packet_and_adds_only_model_corroboration() -> None: + case = _application_case() + packet_digest = evidence.derive_packet_digest(case.record.packet) + + derived = model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + ) + + assert isinstance(derived, evidence.EvidenceRecord) + assert derived.packet is case.record.packet + assert evidence.derive_packet_digest(derived.packet) == packet_digest + assert derived.decision.basis is evidence.EvidenceBasis.MODEL_CORROBORATED + assert derived.decision.family_code == "unsafe" + assert derived.decision.provenance.model_result_digest == case.result.result_digest + assert derived.decision.provenance.proof_record_digest is None + + +def test_failed_model_result_preserves_the_deterministic_record() -> None: + case = _application_case(failed=True) + + derived = model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + ) + + assert derived is case.record + + +def _incomplete_application_case() -> ApplicationCase: + base_contract = _contract() + unknown = family.DecisionRule( + "unknown", + evidence.NeutralOutcome.INDETERMINATE, + ("flow",), + (semantics.RelationStatus.INCOMPLETE,), + ) + contract = replace(base_contract, decisions=(*base_contract.decisions, unknown)) + complete = _record(contract) + relation = complete.packet.transitions[0] + incomplete_relation = semantics.make_relation( + kind=relation.kind, + role=relation.role, + source=relation.source, + target=relation.target, + location=relation.location, + status=semantics.RelationStatus.INCOMPLETE, + conditions=relation.conditions, + reasons=("missing",), + provenance=relation.provenance, + ) + packet = replace(complete.packet, transitions=(incomplete_relation,), completeness_reasons=("missing",)) + decision = evidence.Decision( + "unknown", + evidence.NeutralOutcome.INDETERMINATE, + evidence.EvidenceBasis.DETERMINISTIC, + packet.scope, + ("missing",), + evidence.derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + return _application_case(contract=contract, record=evidence.EvidenceRecord(packet, decision, "")) + + +def test_incomplete_residual_prerequisite_remains_indeterminate() -> None: + case = _incomplete_application_case() + + derived = model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + ) + + assert derived is case.record + assert derived.decision.outcome is evidence.NeutralOutcome.INDETERMINATE + assert derived.decision.basis is evidence.EvidenceBasis.DETERMINISTIC + + +def test_model_application_refuses_proposer_adjudicator_collision_and_family_substitution() -> None: + case = _application_case() + object.__setattr__(case.request, "proposal_program_digests", (case.request.program.program_digest,)) + with pytest.raises(model.ModelContractError, match="propose|adjudicat"): + model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + ) + + case = _application_case() + other_contract = replace(case.contract, family_id="other-family") + with pytest.raises(model.ModelContractError, match="family|contract"): + model_apply.apply_model_result( + case.record, + other_contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + ) + + +def test_model_application_refuses_stale_or_label_contaminated_lineage() -> None: + case = _application_case() + _append_model_use( + case.custody, + case.custody_policy, + case.custodian, + arm_id="model-arm", + purpose=provenance.LineagePurpose.PROMPT, + artifact_digest=case.request.program.prompt_digest, + ) + with pytest.raises(model.ModelContractError, match="omitted|current"): + model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + ) + + case = _application_case() + _append_model_use( + case.custody, + case.custody_policy, + case.custodian, + arm_id="other-arm", + purpose=provenance.LineagePurpose.DECISION_MAPPING, + artifact_digest=case.request.program.decision_mapping_digest, + labels_accessed=True, + ) + with pytest.raises(model.ModelContractError, match="label|contaminat|spent"): + model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + ) + + +def test_model_application_refuses_custody_policy_and_result_substitution() -> None: + case = _application_case() + other_policy = replace(case.custody_policy, policy_digest=_digest("other-policy")) + with pytest.raises(model.ModelContractError, match="custody policy"): + model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + other_policy, + ) + + case = _application_case() + object.__setattr__(case.result, "packet_digest", _digest("other-packet")) + with pytest.raises(model.ModelContractError, match="packet"): + model_apply.apply_model_result( + case.record, + case.contract, + case.request, + case.result, + case.resolver, + case.custody, + case.custody_policy, + )