mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 01:48:17 +00:00
feat(detection-core): preserve model handoff records
This commit is contained in:
Marius Ciepluch
@@ -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.
|
||||
|
||||
@@ -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}
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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})
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user