diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index ff4f607..0adced5 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -155,7 +155,7 @@ - _Requirements: 2.4, 4.4, 10.4, 11.4, 11.5_ - [ ] 6. Add bounded model handoff and semantic shortcut probes -- [ ] 6.1 (P) Preserve reproducible model requests, results, and responses +- [x] 6.1 (P) Preserve reproducible model requests, results, and responses - Present the canonical evidence, completeness snapshot, family declaration, and declared residual question without importing a model runtime. - Freeze provider, model, program, prompt, examples, fitted component, configuration, adapter, decision mapping, and input-bundle identities before execution. - Preserve normalized responses plus one bounded redacted inline value or trusted immutable artifact, including failures. diff --git a/src/openultrasast/core/_model_common.py b/src/openultrasast/core/_model_common.py new file mode 100644 index 0000000..61a6e31 --- /dev/null +++ b/src/openultrasast/core/_model_common.py @@ -0,0 +1,39 @@ +"""Shared validation primitives for dependency-free model contracts.""" + +import json +import re +from typing import cast + +from openultrasast.core import canonical +from openultrasast.core.evidence import EvidenceRecord, evidence_record_bytes + +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class ModelContractError(ValueError): + """A model request, response, or result cannot be reproduced.""" + + +def require_text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise ModelContractError(f"{field} must be nonempty text") + + +def require_digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise ModelContractError(f"{field} must be a full lowercase SHA-256 digest") + + +def _frozen_json(value: object) -> canonical.JsonValue: + if isinstance(value, list): + return tuple(_frozen_json(item) for item in value) + if isinstance(value, dict): + return {str(key): _frozen_json(item) for key, item in value.items()} + return cast(canonical.JsonValue, value) + + +def record_value(record: EvidenceRecord) -> canonical.JsonValue: + body = _frozen_json(json.loads(evidence_record_bytes(record))) + if not isinstance(body, dict): + raise ModelContractError("canonical evidence record must be an object") + return body | {"record_id": record.record_id} diff --git a/src/openultrasast/core/_model_result.py b/src/openultrasast/core/_model_result.py new file mode 100644 index 0000000..f53b2fd --- /dev/null +++ b/src/openultrasast/core/_model_result.py @@ -0,0 +1,285 @@ +"""Preserved model response and result validation internals.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import TYPE_CHECKING + +from openultrasast.core import canonical +from openultrasast.core._model_common import ModelContractError +from openultrasast.core._model_common import require_digest as _digest +from openultrasast.core._model_common import require_text as _text +from openultrasast.core.provenance import ModelCallIdentity +from openultrasast.core.semantics import FieldValue + +if TYPE_CHECKING: + from openultrasast.core.model import ModelProgramDeclaration, ModelRequest + +MODEL_RESPONSE_SCHEMA = "openultrasast.detection-core.model-response.v2" +MODEL_RESULT_SCHEMA = "openultrasast.detection-core.model-result.v2" + + +class ModelStatus(Enum): + SUCCEEDED = "succeeded" + FAILED = "failed" + + +@dataclass(frozen=True) +class PreservedModelResponse: + schema_version: str + normalized_fields: tuple[FieldValue, ...] + inline_redacted_text: str | None + artifact: canonical.ContentRef | None + content_digest: str + redaction_policy_digest: str + + def __post_init__(self) -> None: + _validate_response(self) + + +def _content_ref_value(reference: canonical.ContentRef) -> canonical.JsonValue: + return { + "artifact_id": reference.artifact_id, + "content_digest": reference.content_digest, + "media_type": reference.media_type, + "byte_length": reference.byte_length, + "retention_policy_digest": reference.retention_policy_digest, + "storage_receipt_digest": reference.storage_receipt_digest, + } + + +def _response_value(response: PreservedModelResponse) -> canonical.JsonValue: + artifact: canonical.JsonValue = None if response.artifact is None else _content_ref_value(response.artifact) + return { + "schema_version": response.schema_version, + "normalized_fields": tuple( + {"name": item.name, "kind": item.kind.value, "value": item.value} for item in response.normalized_fields + ), + "inline_redacted_text": response.inline_redacted_text, + "artifact": artifact, + "content_digest": response.content_digest, + "redaction_policy_digest": response.redaction_policy_digest, + } + + +def _validate_response(response: PreservedModelResponse) -> None: + if response.schema_version != MODEL_RESPONSE_SCHEMA: + raise ModelContractError("unknown preserved response schema") + if not isinstance(response.normalized_fields, tuple) or not all(isinstance(item, FieldValue) for item in response.normalized_fields): + raise ModelContractError("normalized response fields must be a tuple of FieldValue values") + for item in response.normalized_fields: + item.__post_init__() + names = tuple(item.name for item in response.normalized_fields) + if len(set(names)) != len(names): + raise ModelContractError("normalized response field names must be unique") + inline = response.inline_redacted_text + artifact = response.artifact + if (inline is None) == (artifact is None): + raise ModelContractError("response requires exactly one inline or artifact preservation form") + if inline is not None: + _text(inline, "inline redacted response") + expected = canonical.bytes_digest(inline.encode("utf-8")) + else: + if not isinstance(artifact, canonical.ContentRef): + raise ModelContractError("response artifact must be a ContentRef") + artifact.__post_init__() + expected = artifact.content_digest + _digest(response.redaction_policy_digest, "redaction_policy_digest") + if not response.content_digest: + object.__setattr__(response, "content_digest", expected) + elif response.content_digest != expected: + raise ModelContractError("response content digest cannot be re-derived") + + +def make_preserved_model_response( + *, + normalized_fields: tuple[FieldValue, ...], + inline_redacted_text: str | None, + artifact: canonical.ContentRef | None, + redaction_policy_digest: str, +) -> PreservedModelResponse: + return PreservedModelResponse( + MODEL_RESPONSE_SCHEMA, + normalized_fields, + inline_redacted_text, + artifact, + "", + redaction_policy_digest, + ) + + +@dataclass(frozen=True) +class ModelResult: + request_digest: str + packet_digest: str + identity: ModelCallIdentity + status: ModelStatus + response: PreservedModelResponse | None + family_decision: str | None + decision_mapping_digest: str | None + failure_reason: str | None + result_digest: str + + def __post_init__(self) -> None: + _validate_result_shape(self) + + +def _identity_value(identity: ModelCallIdentity) -> canonical.JsonValue: + return { + "provider": identity.provider, + "model": identity.model, + "program_digest": identity.program_digest, + "prompt_digest": identity.prompt_digest, + "configuration_digest": identity.configuration_digest, + "adapter_revision": identity.adapter_revision, + } + + +def _result_value(result: ModelResult) -> dict[str, canonical.JsonValue]: + response: canonical.JsonValue = None if result.response is None else _response_value(result.response) + return { + "schema_version": MODEL_RESULT_SCHEMA, + "request_digest": result.request_digest, + "packet_digest": result.packet_digest, + "identity": _identity_value(result.identity), + "status": result.status.value, + "response": response, + "family_decision": result.family_decision, + "decision_mapping_digest": result.decision_mapping_digest, + "failure_reason": result.failure_reason, + } + + +def derive_model_result_digest(result: ModelResult) -> str: + return canonical.content_digest(_result_value(result)) + + +def _validate_result_outcome(result: ModelResult) -> None: + if result.status is ModelStatus.SUCCEEDED: + if result.response is None: + raise ModelContractError("successful result requires a preserved response") + _text(result.family_decision, "family_decision") + _digest(result.decision_mapping_digest, "decision_mapping_digest") + if result.failure_reason is not None: + raise ModelContractError("successful result forbids a failure reason") + else: + _text(result.failure_reason, "failure_reason") + if result.family_decision is not None or result.decision_mapping_digest is not None: + raise ModelContractError("failed result forbids a family decision and mapping") + + +def _validate_result_shape(result: ModelResult) -> None: + _digest(result.request_digest, "request_digest") + _digest(result.packet_digest, "packet_digest") + if not isinstance(result.identity, ModelCallIdentity): + raise ModelContractError("result identity must be a ModelCallIdentity") + result.identity.__post_init__() + if not isinstance(result.status, ModelStatus): + raise ModelContractError("result status must be a ModelStatus") + if result.response is not None: + if not isinstance(result.response, PreservedModelResponse): + raise ModelContractError("result response must be a PreservedModelResponse") + result.response.__post_init__() + _validate_result_outcome(result) + expected = derive_model_result_digest(result) + if not result.result_digest: + object.__setattr__(result, "result_digest", expected) + elif result.result_digest != expected: + raise ModelContractError("model result digest cannot be re-derived") + + +def _program_identity(program: ModelProgramDeclaration) -> ModelCallIdentity: + return ModelCallIdentity( + program.provider, + program.model, + program.program_digest, + program.prompt_digest, + program.configuration_digest, + program.adapter_revision, + ) + + +def make_model_result( + *, + request: ModelRequest, + status: ModelStatus, + response: PreservedModelResponse | None, + family_decision: str | None, + failure_reason: str | None, +) -> ModelResult: + mapping = request.program.decision_mapping_digest if status is ModelStatus.SUCCEEDED else None + return ModelResult( + request.request_id, + request.packet_digest, + _program_identity(request.program), + status, + response, + family_decision, + mapping, + failure_reason, + "", + ) + + +def _validate_response_bytes( + request: ModelRequest, + response: PreservedModelResponse, + resolver: canonical.ContentResolver, +) -> None: + response.__post_init__() + if response.inline_redacted_text is not None: + if len(response.inline_redacted_text.encode("utf-8")) > request.maximum_response_bytes: + raise ModelContractError("inline response exceeds request maximum_response_bytes") + return + assert response.artifact is not None + try: + canonical.resolve_content( + response.artifact, + resolver, + trusted_store_identities=request.trusted_response_store_identities, + maximum_bytes=request.maximum_response_bytes, + ) + except (canonical.ArtifactResolutionError, canonical.ArtifactValidationError) as exc: + raise ModelContractError(f"response artifact cannot be validated: {exc}") from exc + + +def validate_model_result(request: ModelRequest, result: ModelResult, resolver: canonical.ContentResolver) -> None: + """Revalidate a preserved result against its exact frozen request identities.""" + from openultrasast.core.model import ModelRequest, _validate_request + + if not isinstance(request, ModelRequest) or not isinstance(result, ModelResult): + raise ModelContractError("request and result must use model handoff contract types") + if result.request_digest != request.request_id: + raise ModelContractError("result request digest does not match request") + if result.packet_digest != request.packet_digest: + raise ModelContractError("result packet digest does not match request") + if result.identity != _program_identity(request.program): + raise ModelContractError("result program identity does not match frozen declaration") + if result.status is ModelStatus.SUCCEEDED: + if result.decision_mapping_digest != request.program.decision_mapping_digest: + raise ModelContractError("result decision mapping does not match frozen declaration") + if result.family_decision not in request.residual_question.allowed_decisions: + raise ModelContractError("result family decision is not allowed by the residual question") + if result.response is not None: + _validate_response_bytes(request, result.response, resolver) + _validate_request(request) + _validate_result_shape(result) + + +def model_result_bytes(result: ModelResult) -> bytes: + if not isinstance(result, ModelResult): + raise ModelContractError("result must be a ModelResult") + _validate_result_shape(result) + return canonical.canonical_bytes(_result_value(result) | {"result_digest": result.result_digest}) + + +__all__ = [ + "ModelResult", + "PreservedModelResponse", + "derive_model_result_digest", + "make_model_result", + "make_preserved_model_response", + "model_result_bytes", + "validate_model_result", +] diff --git a/src/openultrasast/core/model.py b/src/openultrasast/core/model.py new file mode 100644 index 0000000..45a27d7 --- /dev/null +++ b/src/openultrasast/core/model.py @@ -0,0 +1,455 @@ +"""Dependency-free, reproducible contracts for bounded model handoff.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from openultrasast.core import canonical +from openultrasast.core._model_common import ModelContractError +from openultrasast.core._model_common import record_value as _record_value +from openultrasast.core._model_common import require_digest as _digest +from openultrasast.core._model_common import require_text as _text +from openultrasast.core._model_result import ( + MODEL_RESPONSE_SCHEMA as MODEL_RESPONSE_SCHEMA, +) +from openultrasast.core._model_result import ( + MODEL_RESULT_SCHEMA as MODEL_RESULT_SCHEMA, +) +from openultrasast.core._model_result import ( + ModelResult as ModelResult, +) +from openultrasast.core._model_result import ( + ModelStatus as ModelStatus, +) +from openultrasast.core._model_result import ( + PreservedModelResponse as PreservedModelResponse, +) +from openultrasast.core._model_result import ( + derive_model_result_digest as derive_model_result_digest, +) +from openultrasast.core._model_result import ( + make_model_result as make_model_result, +) +from openultrasast.core._model_result import ( + make_preserved_model_response as make_preserved_model_response, +) +from openultrasast.core._model_result import ( + model_result_bytes as model_result_bytes, +) +from openultrasast.core._model_result import ( + validate_model_result as validate_model_result, +) +from openultrasast.core.evidence import EvidenceRecord, derive_packet_digest, derive_record_id +from openultrasast.core.family import FamilyContract, FieldSpec, ResidualQuestion, family_contract_digest, validate_record +from openultrasast.core.lineage import LineageManifest, derive_lineage_manifest_digest +from openultrasast.core.semantics import RelationStatus + +MODEL_REQUEST_SCHEMA = "openultrasast.detection-core.model-request.v2" + + +def _digest_tuple(values: object, field: str) -> None: + if not isinstance(values, tuple): + raise ModelContractError(f"{field} must be a tuple") + for value in values: + _digest(value, field) + if len(set(values)) != len(values): + raise ModelContractError(f"{field} must be unique") + + +def _text_tuple(values: object, field: str) -> None: + if not isinstance(values, tuple) or not values: + raise ModelContractError(f"{field} must be a nonempty tuple") + for value in values: + _text(value, field) + if len(set(values)) != len(values): + raise ModelContractError(f"{field} must be unique") + + +def _field_spec_value(value: FieldSpec) -> canonical.JsonValue: + return {"name": value.name, "kind": value.kind.value, "required": value.required} + + +def _question_value(question: ResidualQuestion) -> canonical.JsonValue: + return { + "question_id": question.question_id, + "required_roles": question.required_roles, + "allowed_decisions": question.allowed_decisions, + } + + +def _contract_value(contract: FamilyContract) -> canonical.JsonValue: + return { + "family_id": contract.family_id, + "family_version": contract.family_version, + "schema_version": contract.schema_version, + "ecosystems": contract.ecosystems, + "security_object": { + "kind": contract.security_object.kind, + "fields": tuple(_field_spec_value(item) for item in contract.security_object.fields), + "identity_fields": contract.security_object.identity_fields, + }, + "roles": tuple( + { + "role": role.role, + "fields": tuple(_field_spec_value(item) for item in role.fields), + "allowed_relation_kinds": tuple(item.value for item in role.allowed_relation_kinds), + "origin_policy": role.origin_policy.value, + } + for role in contract.roles + ), + "decisions": tuple( + { + "family_code": rule.family_code, + "outcome": rule.outcome.value, + "required_roles": rule.required_roles, + "allowed_statuses": tuple(item.value for item in rule.allowed_statuses), + } + for rule in contract.decisions + ), + "residual_questions": tuple(_question_value(item) for item in contract.residual_questions), + "indeterminate_reasons": contract.indeterminate_reasons, + } + + +@dataclass(frozen=True) +class CompletenessItem: + role: str + status: RelationStatus + reasons: tuple[str, ...] + + def __post_init__(self) -> None: + _text(self.role, "completeness role") + if not isinstance(self.status, RelationStatus): + raise ModelContractError("completeness status must be a RelationStatus") + if not isinstance(self.reasons, tuple) or not all(isinstance(item, str) and item for item in self.reasons): + raise ModelContractError("completeness reasons must be a tuple of nonempty text") + if tuple(sorted(set(self.reasons))) != self.reasons: + raise ModelContractError("completeness reasons must be sorted and unique") + if self.status is RelationStatus.COMPLETE and self.reasons: + raise ModelContractError("complete role cannot carry completeness reasons") + if self.status is not RelationStatus.COMPLETE and not self.reasons: + raise ModelContractError("non-complete role requires a named reason") + + +def _completeness_value(item: CompletenessItem) -> canonical.JsonValue: + return {"role": item.role, "status": item.status.value, "reasons": item.reasons} + + +def _role_completeness(record: EvidenceRecord, role: str) -> CompletenessItem: + relations = tuple(item for item in record.packet.transitions if item.role == role) + observations = tuple(item for item in record.packet.observations if item.role == role) + statuses = tuple(item.status for item in relations) + tuple(item.status for item in observations) + if not statuses: + return CompletenessItem(role, RelationStatus.INCOMPLETE, (f"missing_role:{role}",)) + if RelationStatus.UNRESOLVED in statuses: + status = RelationStatus.UNRESOLVED + elif RelationStatus.INCOMPLETE in statuses: + status = RelationStatus.INCOMPLETE + else: + status = RelationStatus.COMPLETE + reasons = {reason for relation in relations for reason in relation.reasons} + if status is not RelationStatus.COMPLETE and not reasons: + reasons.add(f"role_status:{role}:{status.value}") + return CompletenessItem(role, status, tuple(sorted(reasons))) + + +def completeness_snapshot(record: EvidenceRecord, contract: FamilyContract) -> tuple[CompletenessItem, ...]: + try: + validate_record(contract, record) + except (TypeError, ValueError) as exc: + raise ModelContractError(f"record or family contract is invalid: {exc}") from exc + return tuple(_role_completeness(record, role.role) for role in contract.roles) + + +def derive_completeness_digest(completeness: tuple[CompletenessItem, ...]) -> str: + if not isinstance(completeness, tuple) or not all(isinstance(item, CompletenessItem) for item in completeness): + raise ModelContractError("completeness must be a tuple of CompletenessItem values") + return canonical.content_digest(tuple(_completeness_value(item) for item in completeness)) + + +def derive_model_input_bundle_digest( + record: EvidenceRecord, + contract: FamilyContract, + question: ResidualQuestion, + completeness: tuple[CompletenessItem, ...], +) -> str: + return canonical.content_digest( + { + "record": _record_value(record), + "contract": _contract_value(contract), + "residual_question": _question_value(question), + "completeness": tuple(_completeness_value(item) for item in completeness), + } + ) + + +@dataclass(frozen=True) +class ModelProgramDeclaration: + family_contract_digest: str + provider: str + model: str + program_digest: str + prompt_digest: str + example_set_digest: str | None + fitted_component_digest: str | None + configuration_digest: str + adapter_revision: str + decision_mapping_digest: str + input_bundle_digest: str + lineage_manifest_digest: str + declaration_digest: str + + def __post_init__(self) -> None: + _validate_program(self) + + +def _program_value(program: ModelProgramDeclaration) -> dict[str, canonical.JsonValue]: + return { + "family_contract_digest": program.family_contract_digest, + "provider": program.provider, + "model": program.model, + "program_digest": program.program_digest, + "prompt_digest": program.prompt_digest, + "example_set_digest": program.example_set_digest, + "fitted_component_digest": program.fitted_component_digest, + "configuration_digest": program.configuration_digest, + "adapter_revision": program.adapter_revision, + "decision_mapping_digest": program.decision_mapping_digest, + "input_bundle_digest": program.input_bundle_digest, + "lineage_manifest_digest": program.lineage_manifest_digest, + } + + +def derive_model_program_declaration_digest(program: ModelProgramDeclaration) -> str: + return canonical.content_digest(_program_value(program)) + + +def _validate_program(program: ModelProgramDeclaration) -> None: + for field in ("provider", "model", "adapter_revision"): + _text(getattr(program, field), field) + for field in ( + "family_contract_digest", + "program_digest", + "prompt_digest", + "configuration_digest", + "decision_mapping_digest", + "input_bundle_digest", + "lineage_manifest_digest", + ): + _digest(getattr(program, field), field) + for field in ("example_set_digest", "fitted_component_digest"): + value = getattr(program, field) + if value is not None: + _digest(value, field) + expected = derive_model_program_declaration_digest(program) + if not program.declaration_digest: + object.__setattr__(program, "declaration_digest", expected) + elif program.declaration_digest != expected: + raise ModelContractError("model program declaration digest cannot be re-derived") + + +def make_model_program_declaration( + *, + family_contract_digest: str, + provider: str, + model_name: str, + program_digest: str, + prompt_digest: str, + example_set_digest: str | None, + fitted_component_digest: str | None, + configuration_digest: str, + adapter_revision: str, + decision_mapping_digest: str, + input_bundle_digest: str, + lineage_manifest_digest: str, +) -> ModelProgramDeclaration: + return ModelProgramDeclaration( + family_contract_digest, + provider, + model_name, + program_digest, + prompt_digest, + example_set_digest, + fitted_component_digest, + configuration_digest, + adapter_revision, + decision_mapping_digest, + input_bundle_digest, + lineage_manifest_digest, + "", + ) + + +def _lineage_value(lineage: LineageManifest) -> canonical.JsonValue: + return { + "arm_id": lineage.arm_id, + "subject_digest": lineage.subject_digest, + "uses": tuple( + { + "purpose": use.purpose.value, + "source_population_digest": use.source_population_digest, + "labels_accessed": use.labels_accessed, + "influenced_artifact_digest": use.influenced_artifact_digest, + "population_use_receipt_digest": use.population_use_receipt_digest, + } + for use in lineage.uses + ), + "manifest_digest": lineage.manifest_digest, + } + + +@dataclass(frozen=True) +class ModelRequest: + request_id: str + record: EvidenceRecord + contract: FamilyContract + residual_question: ResidualQuestion + completeness: tuple[CompletenessItem, ...] + packet_digest: str + family_contract_digest: str + residual_question_id: str + completeness_digest: str + program: ModelProgramDeclaration + lineage: LineageManifest + custody_policy_digest: str + proposal_program_digests: tuple[str, ...] + trusted_response_store_identities: tuple[str, ...] + maximum_response_bytes: int + + def __post_init__(self) -> None: + _validate_request(self) + + +def _request_value(request: ModelRequest) -> dict[str, canonical.JsonValue]: + return { + "schema_version": MODEL_REQUEST_SCHEMA, + "record": _record_value(request.record), + "contract": _contract_value(request.contract), + "residual_question": _question_value(request.residual_question), + "completeness": tuple(_completeness_value(item) for item in request.completeness), + "packet_digest": request.packet_digest, + "family_contract_digest": request.family_contract_digest, + "residual_question_id": request.residual_question_id, + "completeness_digest": request.completeness_digest, + "program": _program_value(request.program) | {"declaration_digest": request.program.declaration_digest}, + "lineage": _lineage_value(request.lineage), + "custody_policy_digest": request.custody_policy_digest, + "proposal_program_digests": request.proposal_program_digests, + "trusted_response_store_identities": request.trusted_response_store_identities, + "maximum_response_bytes": request.maximum_response_bytes, + } + + +def derive_model_request_digest(request: ModelRequest) -> str: + return canonical.content_digest(_request_value(request)) + + +def _validate_lineage(lineage: LineageManifest, subject: str) -> None: + if not isinstance(lineage, LineageManifest): + raise ModelContractError("lineage must be a LineageManifest") + for use in lineage.uses: + use.__post_init__() + expected = derive_lineage_manifest_digest(lineage.arm_id, lineage.subject_digest, lineage.uses) + if lineage.manifest_digest != expected: + raise ModelContractError("lineage manifest digest cannot be re-derived") + if lineage.subject_digest != subject: + raise ModelContractError("lineage subject does not match model input bundle") + + +def _validate_request_evidence(request: ModelRequest) -> str: + expected_record_id = derive_record_id(request.record.packet, request.record.decision) + if request.record.record_id != expected_record_id: + raise ModelContractError("evidence record identity cannot be re-derived") + try: + request.record.__post_init__() + validate_record(request.contract, request.record) + except (TypeError, ValueError) as exc: + raise ModelContractError(f"record or family contract is invalid: {exc}") from exc + contract_digest = family_contract_digest(request.contract) + if request.family_contract_digest != contract_digest or request.record.packet.family_contract_digest != contract_digest: + raise ModelContractError("family contract digest does not match request content") + if request.packet_digest != derive_packet_digest(request.record.packet): + raise ModelContractError("packet digest does not match request record") + questions = tuple(item for item in request.contract.residual_questions if item.question_id == request.residual_question_id) + if len(questions) != 1 or request.residual_question != questions[0]: + raise ModelContractError("residual question is not the exact declared family question") + expected_completeness = completeness_snapshot(request.record, request.contract) + if request.completeness != expected_completeness: + raise ModelContractError("completeness snapshot does not match the evidence record") + if request.completeness_digest != derive_completeness_digest(request.completeness): + raise ModelContractError("completeness digest cannot be re-derived") + return contract_digest + + +def _validate_request_program(request: ModelRequest, contract_digest: str) -> None: + request.program.__post_init__() + expected_bundle = derive_model_input_bundle_digest(request.record, request.contract, request.residual_question, request.completeness) + if request.program.input_bundle_digest != expected_bundle: + raise ModelContractError("program input bundle does not match request content") + if request.program.family_contract_digest != contract_digest: + raise ModelContractError("program family contract does not match request") + _validate_lineage(request.lineage, expected_bundle) + if request.program.lineage_manifest_digest != request.lineage.manifest_digest: + raise ModelContractError("program lineage manifest does not match request lineage") + + +def _validate_request_policy(request: ModelRequest) -> None: + _digest(request.custody_policy_digest, "custody_policy_digest") + _digest_tuple(request.proposal_program_digests, "proposal_program_digests") + _text_tuple(request.trusted_response_store_identities, "trusted_response_store_identities") + invalid_maximum = isinstance(request.maximum_response_bytes, bool) or not isinstance(request.maximum_response_bytes, int) + if invalid_maximum or request.maximum_response_bytes <= 0: + raise ModelContractError("maximum_response_bytes must be a positive integer") + expected = derive_model_request_digest(request) + if not request.request_id: + object.__setattr__(request, "request_id", expected) + elif request.request_id != expected: + raise ModelContractError("model request digest cannot be re-derived") + + +def _validate_request(request: ModelRequest) -> None: + contract_digest = _validate_request_evidence(request) + _validate_request_program(request, contract_digest) + _validate_request_policy(request) + + +def make_model_request( + *, + record: EvidenceRecord, + contract: FamilyContract, + residual_question_id: str, + program: ModelProgramDeclaration, + lineage: LineageManifest, + custody_policy_digest: str, + proposal_program_digests: tuple[str, ...], + trusted_response_store_identities: tuple[str, ...], + maximum_response_bytes: int, +) -> ModelRequest: + questions = tuple(item for item in contract.residual_questions if item.question_id == residual_question_id) + if len(questions) != 1: + raise ModelContractError("residual question is not declared exactly once") + completeness = completeness_snapshot(record, contract) + return ModelRequest( + "", + record, + contract, + questions[0], + completeness, + derive_packet_digest(record.packet), + family_contract_digest(contract), + residual_question_id, + derive_completeness_digest(completeness), + program, + lineage, + custody_policy_digest, + proposal_program_digests, + trusted_response_store_identities, + maximum_response_bytes, + ) + + +def model_request_bytes(request: ModelRequest) -> bytes: + if not isinstance(request, ModelRequest): + raise ModelContractError("request must be a ModelRequest") + _validate_request(request) + return canonical.canonical_bytes(_request_value(request) | {"request_id": request.request_id}) diff --git a/tests/test_core_model.py b/tests/test_core_model.py new file mode 100644 index 0000000..dc6ca90 --- /dev/null +++ b/tests/test_core_model.py @@ -0,0 +1,328 @@ +"""Dependency-free, reproducible model handoff contracts.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical, evidence, family, model, provenance, semantics + + +def _digest(value: str) -> str: + return canonical.content_digest(value) + + +def _contract() -> family.FamilyContract: + return family.FamilyContract( + "reference-family", + "1", + evidence.EVIDENCE_PACKET_SCHEMA, + ("python",), + family.ObjectSpec( + "resource", + (family.FieldSpec("name", semantics.FieldKind.TEXT, True),), + ("name",), + ), + ( + family.RoleSpec( + "flow", + (), + (semantics.RelationKind.CALL,), + family.RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT, + ), + family.RoleSpec( + "security_object", + (family.FieldSpec("kind", semantics.FieldKind.TEXT, True),), + (), + family.RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT, + ), + ), + ( + family.DecisionRule( + "unsafe", + evidence.NeutralOutcome.VIOLATION, + ("flow", "security_object"), + (semantics.RelationStatus.COMPLETE,), + ), + family.DecisionRule( + "safe", + evidence.NeutralOutcome.SATISFIED, + ("flow", "security_object"), + (semantics.RelationStatus.COMPLETE,), + ), + ), + (family.ResidualQuestion("classify", ("flow", "security_object"), ("unsafe", "safe")),), + ("missing",), + ) + + +def _record(contract: family.FamilyContract) -> evidence.EvidenceRecord: + location = semantics.SourceLocation("src/product.py", 7, 2, "handle") + origin = semantics.EvidenceProvenance(semantics.ProvenanceKind.PRODUCT_CODE, _digest("source"), None, None, None) + relation = semantics.make_relation( + kind=semantics.RelationKind.CALL, + role="flow", + source=semantics.SemanticNode("source", "source", location), + target=semantics.SemanticNode("target", "target", location), + location=location, + status=semantics.RelationStatus.COMPLETE, + conditions=(), + reasons=(), + provenance=origin, + ) + observation = semantics.make_observation( + role="security_object", + location=location, + fields=(semantics.FieldValue("kind", semantics.FieldKind.TEXT, "resource"),), + status=semantics.RelationStatus.COMPLETE, + provenance=origin, + ) + object_fields = (semantics.FieldValue("name", semantics.FieldKind.TEXT, "document"),) + contract_digest = family.family_contract_digest(contract) + scope = evidence.AnalyzedScope("base", "head", _digest("change"), (evidence.SourceRegion("src/product.py", 1, 20),)) + packet = evidence.EvidencePacket( + evidence.EVIDENCE_PACKET_SCHEMA, + contract.family_id, + contract_digest, + family.derive_security_object_id(contract.security_object, object_fields), + "python", + scope, + (relation,), + (observation,), + object_fields, + (), + provenance.ProvenanceStamp( + _digest("analyzer"), + contract_digest, + _digest("facts"), + _digest("acquisition"), + _digest("configuration"), + _digest("population"), + _digest("change-input"), + ), + ) + decision = evidence.Decision( + "unsafe", + evidence.NeutralOutcome.VIOLATION, + evidence.EvidenceBasis.DETERMINISTIC, + scope, + (), + evidence.derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + return evidence.EvidenceRecord(packet, decision, "") + + +@dataclass(frozen=True) +class HandoffCase: + contract: family.FamilyContract + record: evidence.EvidenceRecord + request: model.ModelRequest + response: model.PreservedModelResponse + result: model.ModelResult + resolver: InMemoryContentResolver + + +def _case(*, artifact_response: bool = False, failed: bool = False) -> HandoffCase: + contract = _contract() + record = _record(contract) + question = contract.residual_questions[0] + completeness = model.completeness_snapshot(record, contract) + bundle = model.derive_model_input_bundle_digest(record, contract, question, completeness) + lineage_digest = provenance.derive_lineage_manifest_digest("model-arm", bundle, ()) + lineage = provenance.LineageManifest("model-arm", bundle, (), lineage_digest) + 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=_digest("custody-policy"), + proposal_program_digests=(_digest("proposer"),), + trusted_response_store_identities=("trusted-store",), + maximum_response_bytes=256, + ) + raw_response = b'{"decision":"unsafe"}' + reference, resolver = stored_artifact(raw_response, artifact_id="model-response") + response = model.make_preserved_model_response( + normalized_fields=(semantics.FieldValue("decision", semantics.FieldKind.TEXT, "unsafe"),), + inline_redacted_text=None if artifact_response else "[redacted] unsafe", + artifact=reference if artifact_response else 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 HandoffCase(contract, record, request, response, result, resolver) + + +@pytest.mark.parametrize("artifact_response", [False, True]) +def test_success_request_and_result_round_trip_canonically(artifact_response: bool) -> None: + case = _case(artifact_response=artifact_response) + request_bytes = model.model_request_bytes(case.request) + result_bytes = model.model_result_bytes(case.result) + + assert len(request_bytes) > 0 and len(result_bytes) > 0 + assert json.dumps(json.loads(request_bytes), sort_keys=True, separators=(",", ":")).encode() == request_bytes + assert json.dumps(json.loads(result_bytes), sort_keys=True, separators=(",", ":")).encode() == result_bytes + model.validate_model_result(case.request, case.result, case.resolver) + + +def test_failed_result_preserves_bounded_response_and_named_failure() -> None: + case = _case(failed=True) + + model.validate_model_result(case.request, case.result, case.resolver) + + assert case.result.response == case.response + assert case.result.failure_reason == "provider_timeout" + assert case.result.family_decision is None + + +def test_request_contains_exact_record_contract_question_and_completeness() -> None: + case = _case() + + assert case.request.record is case.record + assert case.request.contract is case.contract + assert case.request.residual_question == case.contract.residual_questions[0] + assert case.request.packet_digest == evidence.derive_packet_digest(case.record.packet) + assert case.request.completeness == ( + model.CompletenessItem("flow", semantics.RelationStatus.COMPLETE, ()), + model.CompletenessItem("security_object", semantics.RelationStatus.COMPLETE, ()), + ) + + +@pytest.mark.parametrize( + ("field", "replacement", "message"), + [ + ("packet_digest", _digest("other-packet"), "packet"), + ("family_contract_digest", _digest("other-contract"), "family contract"), + ("residual_question_id", "other-question", "residual"), + ("completeness_digest", _digest("other-completeness"), "completeness"), + ], +) +def test_request_refuses_substituted_redundant_identity(field: str, replacement: str, message: str) -> None: + case = _case() + object.__setattr__(case.request, field, replacement) + + with pytest.raises(model.ModelContractError, match=message): + model.model_request_bytes(case.request) + + +def test_request_refuses_substituted_record_program_and_mapping() -> None: + case = _case() + other_record = _record(case.contract) + object.__setattr__(other_record.packet, "security_object_id", "substituted") + object.__setattr__(case.request, "record", other_record) + + with pytest.raises(model.ModelContractError): + model.model_request_bytes(case.request) + + case = _case() + object.__setattr__(case.request.program, "program_digest", _digest("substituted-program")) + with pytest.raises(model.ModelContractError, match="declaration"): + model.model_request_bytes(case.request) + + case = _case() + object.__setattr__(case.request.program, "decision_mapping_digest", _digest("substituted-mapping")) + with pytest.raises(model.ModelContractError, match="declaration"): + model.model_request_bytes(case.request) + + +def test_request_and_result_refuse_substituted_record_identity() -> None: + case = _case() + object.__setattr__(case.request.record, "record_id", _digest("substituted-record-identity")) + + with pytest.raises(model.ModelContractError, match="record identity"): + model.model_request_bytes(case.request) + with pytest.raises(model.ModelContractError, match="record identity"): + model.validate_model_result(case.request, case.result, case.resolver) + + +def test_result_refuses_substituted_request_packet_program_mapping_and_response() -> None: + mutations = ( + ("request_digest", _digest("other-request"), "request"), + ("packet_digest", _digest("other-packet"), "packet"), + ("decision_mapping_digest", _digest("other-mapping"), "mapping"), + ) + for field, replacement, message in mutations: + case = _case() + object.__setattr__(case.result, field, replacement) + with pytest.raises(model.ModelContractError, match=message): + model.validate_model_result(case.request, case.result, case.resolver) + + case = _case() + object.__setattr__(case.result.identity, "program_digest", _digest("other-program")) + with pytest.raises(model.ModelContractError, match="program"): + model.validate_model_result(case.request, case.result, case.resolver) + + case = _case() + object.__setattr__(case.result.response, "inline_redacted_text", "substituted response") + with pytest.raises(model.ModelContractError, match="response"): + model.validate_model_result(case.request, case.result, case.resolver) + + +def test_artifact_response_requires_trusted_resolved_bytes() -> None: + case = _case(artifact_response=True) + model.validate_model_result(case.request, case.result, case.resolver) + + with pytest.raises(model.ModelContractError, match="response artifact"): + model.validate_model_result(case.request, case.result, InMemoryContentResolver({})) + + +def test_inline_response_must_be_bounded_and_exactly_one_preservation_form() -> None: + case = _case() + object.__setattr__(case.request, "maximum_response_bytes", 4) + with pytest.raises(model.ModelContractError, match="maximum"): + model.validate_model_result(case.request, case.result, case.resolver) + + with pytest.raises(model.ModelContractError, match="exactly one"): + model.make_preserved_model_response( + normalized_fields=case.response.normalized_fields, + inline_redacted_text="inline", + artifact=stored_artifact(b"artifact")[0], + redaction_policy_digest=_digest("redaction-policy"), + ) + + +def test_failure_without_received_response_is_preserved() -> None: + case = _case() + result = model.make_model_result( + request=case.request, + status=model.ModelStatus.FAILED, + response=None, + family_decision=None, + failure_reason="no_response", + ) + + model.validate_model_result(case.request, result, case.resolver) + assert result.response is None + + +def test_core_model_contract_has_no_runtime_dependency_imports() -> None: + with open(model.__file__, encoding="utf-8") as source_file: + source = source_file.read() + + assert "import dspy" not in source + assert "litellm" not in source + assert "openai" not in source