From f3adbfe333dac4c3113d59fd0f650a45bba860d8 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 19:38:01 +0100 Subject: [PATCH] feat(detection-core): validate exact admission context --- .kiro/specs/detection-core/tasks.md | 2 +- .../core/_admission_review_shape.py | 109 +-- src/openultrasast/core/_closed_graph.py | 148 ++++ src/openultrasast/core/_validation_shape.py | 83 ++ src/openultrasast/core/admission_context.py | 405 ++++++++++ src/openultrasast/core/measurement.py | 13 + src/openultrasast/core/readiness.py | 8 + src/openultrasast/core/validation.py | 15 + tests/test_core_admission_context.py | 731 ++++++++++++++++++ 9 files changed, 1406 insertions(+), 108 deletions(-) create mode 100644 src/openultrasast/core/_closed_graph.py create mode 100644 src/openultrasast/core/_validation_shape.py create mode 100644 src/openultrasast/core/admission_context.py create mode 100644 tests/test_core_admission_context.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index cd88b44..a0b8975 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -242,7 +242,7 @@ - Preserve aggregate census cardinalities without exposing case identities or per-case labels. - Done when only independently attested label-blind payloads can proceed to manifest validation. - _Requirements: 12.1, 12.2_ -- [ ] 8.4 Validate exact admission context, readiness, scope, sets, and threshold lineage +- [x] 8.4 Validate exact admission context, readiness, scope, sets, and threshold lineage - Bind the exact evidence-validation context, reviewed census, denominator, thresholds, readiness results, untouched population, and current custody head. - Re-derive claimed scopes and exact contract, fact, model, threshold, and outcome-mapping sets from canonical objects. - Require current instrument-through-development readiness with matching analysis identity and evidence-use roles. diff --git a/src/openultrasast/core/_admission_review_shape.py b/src/openultrasast/core/_admission_review_shape.py index 1842c5e..dee2c0b 100644 --- a/src/openultrasast/core/_admission_review_shape.py +++ b/src/openultrasast/core/_admission_review_shape.py @@ -1,110 +1,5 @@ -"""Recursive closed-graph checks for reviewed admission payloads.""" - -from __future__ import annotations - -from dataclasses import fields -from typing import Any, cast - -_SCALAR_TYPES: tuple[type[object], ...] = (str, int, float, bool, type(None)) -_ENUM_STATE_FIELDS = frozenset(("_value_", "_name_", "__objclass__", "_sort_order_")) - - -def closed_graph_problem( - value: object, - *, - dataclass_types: tuple[type[object], ...], - enum_types: tuple[type[object], ...], - path: str = "payload", -) -> str | None: - """Return the first non-closed node in a whitelisted payload graph.""" - value_type = type(value) - if value_type in dataclass_types: - return _dataclass_problem(value, value_type, dataclass_types, enum_types, path) - if value_type is tuple: - for index, item in enumerate(cast(tuple[object, ...], value)): - problem = closed_graph_problem( - item, - dataclass_types=dataclass_types, - enum_types=enum_types, - path=f"{path}[{index}]", - ) - if problem is not None: - return problem - return None - if value_type in enum_types: - return _enum_problem(value, value_type, path) - if value_type in _SCALAR_TYPES: - return None - return f"{path} uses non-closed type {value_type.__name__}" - - -def _dataclass_problem( - value: object, - value_type: type[object], - dataclass_types: tuple[type[object], ...], - enum_types: tuple[type[object], ...], - path: str, -) -> str | None: - declared_fields = fields(cast(Any, value_type)) - declared_names = {field.name for field in declared_fields} - try: - actual_names = set(vars(value)) - except TypeError: - actual_names = declared_names - if actual_names != declared_names: - return f"{path} must have exactly its declared fields" - for field in declared_fields: - try: - field_value = getattr(value, field.name) - except AttributeError: - return f"{path}.{field.name} is missing" - problem = closed_graph_problem( - field_value, - dataclass_types=dataclass_types, - enum_types=enum_types, - path=f"{path}.{field.name}", - ) - if problem is not None: - return problem - return None - - -def _enum_problem(value: object, enum_type: type[object], path: str) -> str | None: - state = vars(value) - if set(state) != _ENUM_STATE_FIELDS: - return f"{path} enum member must have exactly its canonical state fields" - expected = _canonical_enum_state(value, enum_type) - if expected is None: - return f"{path} enum type has aliases or nonunique member mappings" - expected_name, expected_value, expected_order = expected - if ( - state["_name_"] is not expected_name - or state["_value_"] is not expected_value - or state["__objclass__"] is not enum_type - or type(state["_sort_order_"]) is not int - or state["_sort_order_"] != expected_order - ): - return f"{path} enum member state is not canonical" - return None - - -def _canonical_enum_state(value: object, enum_type: type[object]) -> tuple[object, object, int] | None: - enum_class = cast(Any, enum_type) - canonical_members = tuple(enum_class) - name_entries = tuple(enum_class.__members__.items()) - value_entries = tuple(enum_class._value2member_map_.items()) - if len(name_entries) != len(canonical_members) or len(value_entries) != len(canonical_members): - return None - if any(sum(candidate is member for _, candidate in name_entries) != 1 for member in canonical_members): - return None - if any(sum(candidate is member for _, candidate in value_entries) != 1 for member in canonical_members): - return None - names = tuple(name for name, member in name_entries if member is value) - values = tuple(item for item, member in value_entries if member is value) - orders = tuple(index for index, member in enumerate(canonical_members) if member is value) - if len(names) != 1 or len(values) != 1 or len(orders) != 1: - return None - return names[0], values[0], orders[0] +"""Compatibility import for the neutral closed-graph validator.""" +from openultrasast.core._closed_graph import closed_graph_problem as closed_graph_problem __all__ = ["closed_graph_problem"] diff --git a/src/openultrasast/core/_closed_graph.py b/src/openultrasast/core/_closed_graph.py new file mode 100644 index 0000000..2e4a772 --- /dev/null +++ b/src/openultrasast/core/_closed_graph.py @@ -0,0 +1,148 @@ +"""Reusable recursive checks for canonical closed object graphs.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import fields +from typing import Any, cast + +_SCALAR_TYPES: tuple[type[object], ...] = (str, int, float, bool, type(None)) +_ENUM_STATE_FIELDS = frozenset(("_value_", "_name_", "__objclass__", "_sort_order_")) + + +def closed_graph_problem( + value: object, + *, + dataclass_types: tuple[type[object], ...], + enum_types: tuple[type[object], ...], + mapping_types: tuple[type[object], ...] = (), + path: str = "payload", +) -> str | None: + """Return the first non-closed node in a whitelisted payload graph.""" + value_type = type(value) + if value_type in dataclass_types: + return _dataclass_problem(value, value_type, dataclass_types, enum_types, mapping_types, path) + if value_type is tuple: + for index, item in enumerate(cast(tuple[object, ...], value)): + problem = closed_graph_problem( + item, + dataclass_types=dataclass_types, + enum_types=enum_types, + mapping_types=mapping_types, + path=f"{path}[{index}]", + ) + if problem is not None: + return problem + return None + if value_type in enum_types: + return _enum_problem(value, value_type, path) + if value_type in mapping_types: + return _mapping_problem(value, dataclass_types, enum_types, mapping_types, path) + if value_type in _SCALAR_TYPES: + return None + return f"{path} uses non-closed type {value_type.__name__}" + + +def _dataclass_problem( + value: object, + value_type: type[object], + dataclass_types: tuple[type[object], ...], + enum_types: tuple[type[object], ...], + mapping_types: tuple[type[object], ...], + path: str, +) -> str | None: + declared_fields = fields(cast(Any, value_type)) + declared_names = {field.name for field in declared_fields} + try: + actual_names = set(vars(value)) + except TypeError: + actual_names = declared_names + if actual_names != declared_names: + return f"{path} must have exactly its declared fields" + for field in declared_fields: + try: + field_value = getattr(value, field.name) + except AttributeError: + return f"{path}.{field.name} is missing" + problem = closed_graph_problem( + field_value, + dataclass_types=dataclass_types, + enum_types=enum_types, + mapping_types=mapping_types, + path=f"{path}.{field.name}", + ) + if problem is not None: + return problem + return None + + +def _mapping_problem( + value: object, + dataclass_types: tuple[type[object], ...], + enum_types: tuple[type[object], ...], + mapping_types: tuple[type[object], ...], + path: str, +) -> str | None: + for key, item in cast(Mapping[object, object], value).items(): + for name, candidate in (("key", key), ("value", item)): + problem = closed_graph_problem( + candidate, + dataclass_types=dataclass_types, + enum_types=enum_types, + mapping_types=mapping_types, + path=f"{path}.{name}", + ) + if problem is not None: + return problem + return None + + +def exact_dataclass_problem(value: object, expected_type: type[object], path: str) -> str | None: + """Check only an aggregate root's exact class and declared instance fields.""" + if type(value) is not expected_type: + return f"{path} must use exact type {expected_type.__name__}" + declared = {field.name for field in fields(cast(Any, expected_type))} + if set(vars(value)) != declared: + return f"{path} must have exactly its declared fields" + return None + + +def _enum_problem(value: object, enum_type: type[object], path: str) -> str | None: + state = vars(value) + if set(state) != _ENUM_STATE_FIELDS: + return f"{path} enum member must have exactly its canonical state fields" + expected = _canonical_enum_state(value, enum_type) + if expected is None: + return f"{path} enum type has aliases or nonunique member mappings" + expected_name, expected_value, expected_order = expected + if ( + state["_name_"] is not expected_name + or state["_value_"] is not expected_value + or state["__objclass__"] is not enum_type + or type(state["_sort_order_"]) is not int + or state["_sort_order_"] != expected_order + ): + return f"{path} enum member state is not canonical" + return None + + +def _canonical_enum_state(value: object, enum_type: type[object]) -> tuple[object, object, int] | None: + enum_class = cast(Any, enum_type) + canonical_members = tuple(enum_class) + name_entries = tuple(enum_class.__members__.items()) + value_entries = tuple(enum_class._value2member_map_.items()) + if len(name_entries) != len(canonical_members) or len(value_entries) != len(canonical_members): + return None + if any(sum(candidate is member for _, candidate in name_entries) != 1 for member in canonical_members): + return None + if any(sum(candidate is member for _, candidate in value_entries) != 1 for member in canonical_members): + return None + names = tuple(name for name, member in name_entries if member is value) + values = tuple(item for item, member in value_entries if member is value) + orders = tuple(index for index, member in enumerate(canonical_members) if member is value) + if len(names) != 1 or len(values) != 1 or len(orders) != 1: + return None + return names[0], values[0], orders[0] + + +__all__ = ["closed_graph_problem", "exact_dataclass_problem"] diff --git a/src/openultrasast/core/_validation_shape.py b/src/openultrasast/core/_validation_shape.py new file mode 100644 index 0000000..e527d3c --- /dev/null +++ b/src/openultrasast/core/_validation_shape.py @@ -0,0 +1,83 @@ +"""Closed object-graph whitelist for evidence validation contexts.""" + +from __future__ import annotations + +from types import MappingProxyType + +from openultrasast.core._closed_graph import closed_graph_problem +from openultrasast.core.canonical import ContentRef +from openultrasast.core.custody import LineagePurpose, PopulationCustodyPolicy +from openultrasast.core.evidence import NeutralOutcome +from openultrasast.core.factpack import ( + FactPack, + IndependenceAttestation, + KnowledgeEvidence, + KnowledgeEvidenceKind, + RelationEffect, + ReviewIndependencePolicy, + SemanticFact, + StructuralRequirement, + StructuralRequirementKind, +) +from openultrasast.core.family import ( + DecisionRule, + FamilyContract, + FieldSpec, + ObjectSpec, + RelationOriginPolicy, + ResidualQuestion, + RoleSpec, +) +from openultrasast.core.family_catalog import FamilyCatalog, FamilyCatalogEntry +from openultrasast.core.lineage import LineageManifest, LineageUse +from openultrasast.core.model import ModelProgramDeclaration +from openultrasast.core.provenance import ProvenanceTrustPolicy +from openultrasast.core.semantics import FieldKind, RelationKind, RelationStatus + +_DATACLASSES: tuple[type[object], ...] = ( + FamilyCatalog, + FamilyCatalogEntry, + FamilyContract, + FieldSpec, + ObjectSpec, + RoleSpec, + DecisionRule, + ResidualQuestion, + FactPack, + IndependenceAttestation, + ReviewIndependencePolicy, + KnowledgeEvidence, + StructuralRequirement, + RelationEffect, + SemanticFact, + ContentRef, + ModelProgramDeclaration, + LineageManifest, + LineageUse, + ProvenanceTrustPolicy, + PopulationCustodyPolicy, +) +_ENUMS: tuple[type[object], ...] = ( + NeutralOutcome, + RelationOriginPolicy, + FieldKind, + RelationKind, + RelationStatus, + KnowledgeEvidenceKind, + StructuralRequirementKind, + LineagePurpose, +) +_MAPPINGS: tuple[type[object], ...] = (type(MappingProxyType({})),) + + +def evidence_context_problem(value: object, context_type: type[object]) -> str | None: + return closed_graph_problem( + value, + dataclass_types=(context_type, *_DATACLASSES), + enum_types=_ENUMS, + mapping_types=_MAPPINGS, + path="evidence validation context", + ) + + +__all__ = ["evidence_context_problem"] diff --git a/src/openultrasast/core/admission_context.py b/src/openultrasast/core/admission_context.py new file mode 100644 index 0000000..0f3fbeb --- /dev/null +++ b/src/openultrasast/core/admission_context.py @@ -0,0 +1,405 @@ +"""Exact, label-blind admission context validation before feasibility.""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass + +from openultrasast.core import canonical +from openultrasast.core._closed_graph import closed_graph_problem, exact_dataclass_problem +from openultrasast.core.admission_review import AdmissionManifestV2, validate_reviewed_admission_payloads +from openultrasast.core.custody import ( + LineagePurpose, + PopulationCustodyPort, + PopulationStatus, + validate_population_custody_log, +) +from openultrasast.core.factpack import FACT_PACK_SCHEMA, ReviewIndependencePolicy +from openultrasast.core.family_catalog import FamilyCatalog, FamilyCatalogEntry, validate_catalog +from openultrasast.core.lineage import LineageExpectation, LineageManifest, validate_lineage +from openultrasast.core.measurement import ( + BaselineIdentity, + DenominatorPolicy, + ScopeKey, + derive_denominator_policy_digest, + derive_identity_set_digest, + validate_denominator_policy, +) +from openultrasast.core.model import ModelProgramDeclaration +from openultrasast.core.proof import ProofTrustPolicy, derive_proof_trust_policy_digest +from openultrasast.core.readiness import ( + AnalysisIdentity, + ReadinessResult, + ReadinessStage, + authorize_stage, +) +from openultrasast.core.semantics import ( + OBSERVATION_SCHEMA, + SEMANTIC_RELATION_SCHEMA, + FieldKind, + ProvenanceKind, + RelationKind, + RelationStatus, +) +from openultrasast.core.validation import ( + EvidenceValidationContext, + derive_fact_set_digest, + validate_evidence_validation_context, +) + +ADMISSION_VALIDATION_CONTEXT_SCHEMA = "openultrasast.detection-core.admission-validation-context.v2" +OUTCOME_MAPPING_SCHEMA = "openultrasast.detection-core.outcome-mapping.v2" +SEMANTIC_VOCABULARY_SCHEMA = "openultrasast.detection-core.semantic-vocabulary.v2" + + +class AdmissionContextError(ValueError): + """Admission inputs disagree before any feasibility work or label access.""" + + +@dataclass(frozen=True) +class AdmissionValidationContext: + evidence: EvidenceValidationContext + admission_review_policy: ReviewIndependencePolicy + proof_trust_policy: ProofTrustPolicy | None + context_digest: str + + +def derive_fact_schema_digest() -> str: + return canonical.content_digest(FACT_PACK_SCHEMA) + + +def derive_semantic_vocabulary_digest() -> str: + return canonical.content_digest( + { + "schema_version": SEMANTIC_VOCABULARY_SCHEMA, + "relation_schema": SEMANTIC_RELATION_SCHEMA, + "observation_schema": OBSERVATION_SCHEMA, + "field_kinds": tuple(item.value for item in FieldKind), + "relation_kinds": tuple(item.value for item in RelationKind), + "relation_statuses": tuple(item.value for item in RelationStatus), + "provenance_kinds": tuple(item.value for item in ProvenanceKind), + } + ) + + +def _decision_value(entry: FamilyCatalogEntry) -> canonical.JsonValue: + contract = entry.contract + return { + "contract_digest": entry.contract_digest, + "decisions": tuple( + { + "family_code": rule.family_code, + "outcome": rule.outcome.value, + "required_roles": rule.required_roles, + "allowed_statuses": tuple(status.value for status in rule.allowed_statuses), + } + for rule in contract.decisions + ), + } + + +def derive_outcome_mapping_digest(catalog: FamilyCatalog) -> str: + validate_catalog(catalog) + return canonical.content_digest( + {"schema_version": OUTCOME_MAPPING_SCHEMA, "contracts": tuple(_decision_value(entry) for entry in catalog.entries)} + ) + + +def derive_admission_validation_context_digest(context: AdmissionValidationContext) -> str: + proof_digest = None + if context.proof_trust_policy is not None: + proof_digest = derive_proof_trust_policy_digest(context.proof_trust_policy) + return canonical.content_digest( + { + "schema_version": ADMISSION_VALIDATION_CONTEXT_SCHEMA, + "evidence_context_digest": context.evidence.context_digest, + "admission_review_policy_digest": context.admission_review_policy.policy_digest, + "proof_trust_policy_digest": proof_digest, + } + ) + + +def make_admission_validation_context( + *, + evidence: EvidenceValidationContext, + admission_review_policy: ReviewIndependencePolicy, + proof_trust_policy: ProofTrustPolicy | None, +) -> AdmissionValidationContext: + provisional = AdmissionValidationContext(evidence, admission_review_policy, proof_trust_policy, "0" * 64) + return AdmissionValidationContext( + evidence, + admission_review_policy, + proof_trust_policy, + derive_admission_validation_context_digest(provisional), + ) + + +def _contract_for_scope(catalog: FamilyCatalog, scope: ScopeKey) -> FamilyCatalogEntry: + matches = tuple( + entry for entry in catalog.entries if entry.contract.family_id == scope.family_id and scope.ecosystem in entry.contract.ecosystems + ) + if len(matches) != 1: + raise AdmissionContextError("each claimed scope must resolve one exact family contract") + return matches[0] + + +def derive_analysis_identity( + baseline: BaselineIdentity, + catalog: FamilyCatalog, + scope: ScopeKey, + model_program_digest: str | None, +) -> AnalysisIdentity: + entry = _contract_for_scope(catalog, scope) + return AnalysisIdentity( + entry.contract.schema_version, + derive_semantic_vocabulary_digest(), + baseline.analyzer_revision, + entry.contract_digest, + baseline.fact_set_digest, + baseline.acquisition_record_digest, + baseline.configuration_digest, + model_program_digest, + baseline.outcome_mapping_digest, + ) + + +def _validate_context(context: AdmissionValidationContext) -> None: + problem = exact_dataclass_problem(context, AdmissionValidationContext, "admission validation context") + if problem is not None: + raise AdmissionContextError(f"closed admission context violation: {problem}") + validate_evidence_validation_context(context.evidence) + context.admission_review_policy.__post_init__() + if context.proof_trust_policy is not None: + problem = closed_graph_problem( + context.proof_trust_policy, + dataclass_types=(ProofTrustPolicy,), + enum_types=(), + path="proof trust policy", + ) + if problem is not None: + raise AdmissionContextError(f"closed proof policy violation: {problem}") + context.proof_trust_policy.__post_init__() + expected = derive_admission_validation_context_digest(context) + if context.context_digest != expected: + raise AdmissionContextError("admission validation context digest cannot be re-derived") + + +def _validate_scopes_and_sets(manifest: AdmissionManifestV2, denominator: DenominatorPolicy, context: AdmissionValidationContext) -> None: + baseline = manifest.identity + evidence = context.evidence + catalog = evidence.family_catalog + validate_denominator_policy(denominator) + entries = tuple(entry.contract_digest for entry in catalog.entries) + if baseline.family_contract_digests != tuple(sorted(entries)) or baseline.family_contract_set_digest != catalog.catalog_digest: + raise AdmissionContextError("baseline family catalog set does not match canonical catalog objects") + census_scopes = tuple(row.scope for row in manifest.census.rows) + if baseline.claimed_scopes != census_scopes: + raise AdmissionContextError("claimed scopes do not match the reviewed census") + for scope in baseline.claimed_scopes: + _contract_for_scope(catalog, scope) + expected_fact_set = derive_fact_set_digest(evidence.fact_packs) + if baseline.fact_schema_digest != derive_fact_schema_digest() or baseline.fact_set_digest != expected_fact_set: + raise AdmissionContextError("baseline fact set does not match canonical fact objects") + programs = tuple(sorted(item.declaration_digest for item in evidence.model_programs)) + expected_program_set = derive_identity_set_digest("model_program", programs) if programs else None + if baseline.model_program_digests != programs or baseline.model_program_set_digest != expected_program_set: + raise AdmissionContextError("baseline model program set does not match canonical model objects") + if baseline.outcome_mapping_digest != derive_outcome_mapping_digest(catalog): + raise AdmissionContextError("baseline outcome mapping does not match ordered contract decisions") + if denominator.policy_digest != derive_denominator_policy_digest(denominator): + raise AdmissionContextError("denominator policy digest cannot be re-derived") + if denominator.policy_digest != baseline.denominator_policy_digest or denominator.policy_digest != manifest.denominator_policy_digest: + raise AdmissionContextError("denominator policy does not match the reviewed manifest") + if any(rule.scopes != baseline.claimed_scopes for rule in denominator.rules): + raise AdmissionContextError("denominator scope does not exactly match claimed scopes") + + +def _expectations_for_program(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 _register( + expected: dict[str, tuple[str, tuple[LineageExpectation, ...]]], + digest: str, + subject: str, + expectations: tuple[LineageExpectation, ...], +) -> None: + value = (subject, expectations) + if digest in expected and expected[digest] != value: + raise AdmissionContextError("lineage manifest is ambiguously owned by canonical objects") + expected[digest] = value + + +def _lineage_expectations( + manifest: AdmissionManifestV2, + context: AdmissionValidationContext, +) -> dict[str, tuple[str, tuple[LineageExpectation, ...]]]: + expected: dict[str, tuple[str, tuple[LineageExpectation, ...]]] = {} + _register( + expected, + manifest.threshold_lineage_manifest_digest, + manifest.threshold_set_digest, + (LineageExpectation(LineagePurpose.THRESHOLD, manifest.threshold_set_digest, 1, 1),), + ) + for program in context.evidence.model_programs: + _register(expected, program.lineage_manifest_digest, program.input_bundle_digest, _expectations_for_program(program)) + for pack in context.evidence.fact_packs: + expectations = ( + LineageExpectation(LineagePurpose.SEMANTIC_FACT, pack.normalized_content_digest, 1, 1), + LineageExpectation(LineagePurpose.PROMOTION, pack.normalized_content_digest, 0, 1), + ) + for review in pack.review_evidence: + _register(expected, review.lineage_manifest_digest, pack.normalized_content_digest, expectations) + return expected + + +def _orphan_expectations(lineage: LineageManifest) -> tuple[LineageExpectation, ...]: + if not lineage.uses: + raise AdmissionContextError("unowned lineage must not be empty") + allowed = {LineagePurpose.MATCHING_BEHAVIOR, LineagePurpose.PROMOTION} + if any(use.purpose not in allowed or use.influenced_artifact_digest != lineage.subject_digest for use in lineage.uses): + raise AdmissionContextError("lineage is not owned by a canonical matcher or promotion object") + purposes = tuple(sorted({use.purpose for use in lineage.uses}, key=lambda item: item.value)) + return tuple( + LineageExpectation( + purpose, + lineage.subject_digest, + sum(use.purpose is purpose for use in lineage.uses), + sum(use.purpose is purpose for use in lineage.uses), + ) + for purpose in purposes + ) + + +def _refuse_omitted_lineage_uses( + lineage: LineageManifest, + expectations: tuple[LineageExpectation, ...], + populations: set[str], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, +) -> None: + expected = {(item.purpose, item.influenced_artifact_digest) for item in expectations} + declared = {use.population_use_receipt_digest for use in lineage.uses} + for population in sorted(populations): + log = custody.current_log(population) + validate_population_custody_log(log, context.evidence.custody_policy, current=custody) + for receipt in log.receipts: + request = receipt.request + key = (request.purpose, request.influenced_artifact_digest) + relevant = key in expected and (request.arm_id == lineage.arm_id or request.labels_accessed) + if relevant and receipt.receipt_digest not in declared: + label = "label-influenced " if request.labels_accessed else "" + raise AdmissionContextError(f"lineage omitted current {label}use: {receipt.receipt_digest}") + + +def _validate_lineages(manifest: AdmissionManifestV2, context: AdmissionValidationContext, custody: PopulationCustodyPort) -> None: + lineages = context.evidence.lineages + indexed = {item.manifest_digest: item for item in lineages} + if len(indexed) != len(lineages): + raise AdmissionContextError("duplicate lineage manifest identity") + expected = _lineage_expectations(manifest, context) + if any(digest not in indexed for digest in expected): + raise AdmissionContextError("required threshold, model, or fact lineage is absent") + populations = {manifest.identity.population_digest, *(use.source_population_digest for item in lineages for use in item.uses)} + for lineage in lineages: + binding = expected.get(lineage.manifest_digest) + subject, expectations = binding if binding is not None else (lineage.subject_digest, _orphan_expectations(lineage)) + decision = validate_lineage( + lineage, + subject, + expectations, + manifest.identity.population_digest, + True, + custody, + context.evidence.custody_policy, + ) + if not decision.allowed: + raise AdmissionContextError("lineage validation refused: " + ",".join(decision.reasons)) + _refuse_omitted_lineage_uses(lineage, expectations, populations, context, custody) + + +def _model_for_scope(context: AdmissionValidationContext, contract_digest: str) -> str | None: + programs = tuple(item.declaration_digest for item in context.evidence.model_programs if item.family_contract_digest == contract_digest) + if len(programs) > 1: + raise AdmissionContextError("a readiness scope cannot select multiple model programs") + return programs[0] if programs else None + + +def _validate_readiness(manifest: AdmissionManifestV2, readiness: Sequence[ReadinessResult], context: AdmissionValidationContext) -> None: + if type(readiness) is not tuple: + raise AdmissionContextError("readiness container must be an exact tuple") + values = tuple(readiness) + digests = tuple(sorted(item.result_digest for item in values)) + if digests != manifest.readiness_digests or len(set(digests)) != len(digests): + raise AdmissionContextError("readiness result set does not match the reviewed manifest") + consumed: set[str] = set() + for scope in manifest.identity.claimed_scopes: + receipts = tuple( + item + for item in values + if item.arm_id == manifest.identity.arm_id and item.family_id == scope.family_id and item.ecosystem == scope.ecosystem + ) + entry = _contract_for_scope(context.evidence.family_catalog, scope) + model = _model_for_scope(context, entry.contract_digest) + current = derive_analysis_identity(manifest.identity, context.evidence.family_catalog, scope, model) + authorization = authorize_stage(ReadinessStage.ADMISSION, receipts, current) + if not authorization.authorized: + raise AdmissionContextError("current instrument-through-development readiness refused: " + ",".join(authorization.reasons)) + consumed.update(item.result_digest for item in receipts) + if consumed != set(digests): + raise AdmissionContextError("readiness contains a scope, arm, or identity outside the reviewed manifest") + + +def _validate_custody(manifest: AdmissionManifestV2, context: AdmissionValidationContext, custody: PopulationCustodyPort) -> None: + log = custody.current_log(manifest.identity.population_digest) + validate_population_custody_log(log, context.evidence.custody_policy, current=custody) + if log.log_digest != manifest.custody_log_digest: + raise AdmissionContextError("manifest custody head is not the current complete custody log") + if log.status is not PopulationStatus.AVAILABLE: + raise AdmissionContextError("admission population is spent and no longer available") + + +def validate_admission_context( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> None: + """Validate every pre-opening identity and lineage; feasibility is deliberately absent.""" + try: + _validate_context(context) + if manifest.validation_context_digest != context.context_digest: + raise AdmissionContextError("manifest validation context has been substituted") + validate_reviewed_admission_payloads(manifest.census, manifest, context.admission_review_policy, resolver) + _validate_scopes_and_sets(manifest, denominator, context) + _validate_custody(manifest, context, custody) + _validate_lineages(manifest, context, custody) + _validate_readiness(manifest, readiness, context) + except AdmissionContextError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise AdmissionContextError(f"admission context validation failed: {exc}") from exc + + +__all__ = [ + "ADMISSION_VALIDATION_CONTEXT_SCHEMA", + "AdmissionContextError", + "AdmissionValidationContext", + "derive_admission_validation_context_digest", + "derive_analysis_identity", + "derive_fact_schema_digest", + "derive_outcome_mapping_digest", + "derive_semantic_vocabulary_digest", + "make_admission_validation_context", + "validate_admission_context", +] diff --git a/src/openultrasast/core/measurement.py b/src/openultrasast/core/measurement.py index 8d4bbac..e17c147 100644 --- a/src/openultrasast/core/measurement.py +++ b/src/openultrasast/core/measurement.py @@ -9,6 +9,7 @@ from dataclasses import dataclass from enum import Enum from openultrasast.core import canonical +from openultrasast.core._closed_graph import closed_graph_problem BASELINE_IDENTITY_SCHEMA = "openultrasast.detection-core.baseline-identity.v2" DENOMINATOR_POLICY_SCHEMA = "openultrasast.detection-core.denominator-policy.v2" @@ -300,12 +301,20 @@ class DenominatorPolicy: _validate_policy(self) +_POLICY_TYPES: tuple[type[object], ...] = (DenominatorPolicy, MetricRule, ScopeKey) +_POLICY_ENUMS: tuple[type[object], ...] = (MetricId, NumeratorKind, DenominatorKind, IntervalMethod) + + def _validate_policy(value: DenominatorPolicy) -> None: + problem = closed_graph_problem(value, dataclass_types=_POLICY_TYPES, enum_types=_POLICY_ENUMS, path="denominator policy") + if problem is not None: + raise MeasurementContractError(f"closed denominator graph violation: {problem}") if not isinstance(value.rules, tuple) or not all(isinstance(item, MetricRule) for item in value.rules): raise MeasurementContractError("rules must be a tuple of MetricRule values") if tuple(rule.metric_id for rule in value.rules) != tuple(MetricId): raise MeasurementContractError("each standard metric must occur exactly once in canonical order") for rule in value.rules: + rule.__post_init__() expected_numerator, expected_denominator, _ = _ALGEBRA[rule.metric_id] if (rule.numerator, rule.denominator) != (expected_numerator, expected_denominator): raise MeasurementContractError("metric rule violates the fixed algebra") @@ -320,6 +329,9 @@ def _validate_policy(value: DenominatorPolicy) -> None: raise MeasurementContractError("denominator policy digest does not match policy") +validate_denominator_policy = _validate_policy + + def _policy_digest(rules: tuple[MetricRule, ...], method: IntervalMethod) -> str: return canonical.content_digest( { @@ -478,4 +490,5 @@ __all__ = [ "metric_result_value", "require_comparable_baselines", "standard_metric_rules", + "validate_denominator_policy", ] diff --git a/src/openultrasast/core/readiness.py b/src/openultrasast/core/readiness.py index eaf0faf..824768b 100644 --- a/src/openultrasast/core/readiness.py +++ b/src/openultrasast/core/readiness.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from enum import Enum from openultrasast.core import canonical +from openultrasast.core._closed_graph import closed_graph_problem READINESS_RESULT_SCHEMA = "openultrasast.detection-core.readiness-result.v2" _DIGEST = re.compile(r"[0-9a-f]{64}") @@ -197,6 +198,10 @@ class StageAuthorization: validate_stage_authorization(self) +_RESULT_TYPES: tuple[type[object], ...] = (ReadinessResult, AnalysisIdentity, FixtureResult) +_RESULT_ENUMS: tuple[type[object], ...] = (ReadinessStage, ReadinessEvidenceUse, StageResult, FixtureRole) + + def _identity_value(value: AnalysisIdentity) -> canonical.JsonValue: return { "core_schema_version": value.core_schema_version, @@ -332,6 +337,9 @@ def derive_readiness_result_digest( def validate_readiness_result(value: ReadinessResult) -> None: + problem = closed_graph_problem(value, dataclass_types=_RESULT_TYPES, enum_types=_RESULT_ENUMS, path="readiness result") + if problem is not None: + raise ReadinessValidationError(f"closed readiness graph violation: {problem}; exact tuple fields required") if not isinstance(value, ReadinessResult): raise ReadinessValidationError("readiness result must be a ReadinessResult") _validate_result_fields(value) diff --git a/src/openultrasast/core/validation.py b/src/openultrasast/core/validation.py index 3e7939a..1e7f4a7 100644 --- a/src/openultrasast/core/validation.py +++ b/src/openultrasast/core/validation.py @@ -6,6 +6,7 @@ from dataclasses import dataclass from openultrasast.core import canonical from openultrasast.core._validation_model import validate_and_replay_model_derivation +from openultrasast.core._validation_shape import evidence_context_problem from openultrasast.core.custody import ( PopulationCustodyPolicy, PopulationCustodyPort, @@ -255,6 +256,9 @@ def _validate_context_members(context: EvidenceValidationContext) -> None: def _validate_context(context: EvidenceValidationContext) -> None: + problem = evidence_context_problem(context, EvidenceValidationContext) + if problem is not None: + raise EvidenceInputValidationError(f"closed evidence context violation: {problem}") if not isinstance(context, EvidenceValidationContext): raise EvidenceInputValidationError("context must be an EvidenceValidationContext") _validate_context_types(context) @@ -266,6 +270,16 @@ def _validate_context(context: EvidenceValidationContext) -> None: _validate_context_members(context) +def validate_evidence_validation_context(context: EvidenceValidationContext) -> None: + """Re-derive a closed evidence context without requiring a case record.""" + try: + _validate_context(context) + except EvidenceInputValidationError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise EvidenceInputValidationError(f"evidence validation context is invalid: {exc}") from exc + + def _validate_record_identity(record: EvidenceRecord) -> None: if not isinstance(record, EvidenceRecord): raise EvidenceInputValidationError("record must be an EvidenceRecord") @@ -386,5 +400,6 @@ __all__ = [ "fact_set_bytes", "make_evidence_validation_context", "make_evidence_validation_input", + "validate_evidence_validation_context", "validate_evidence_input", ] diff --git a/tests/test_core_admission_context.py b/tests/test_core_admission_context.py new file mode 100644 index 0000000..55db320 --- /dev/null +++ b/tests/test_core_admission_context.py @@ -0,0 +1,731 @@ +"""Exact pre-feasibility admission context validation (Reqs 12.1, 12.2, 13.1).""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from typing import TypeVar + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact +from test_core_admission_review import _attestation, _policy +from test_core_model import ApplicationCase, _application_case +from test_core_readiness import _chain +from test_core_validation_v2 import _catalog, _review_policy, _trust_policy + +from openultrasast.core import canonical, custody, validation +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_review import ( + AdmissionManifestV2, + AdmissionThreshold, + PopulationCensus, + ScopeCensus, + derive_census_digest, + derive_census_payload_digest, + derive_manifest_digest, + derive_manifest_payload_digest, + derive_threshold_set_digest, +) +from openultrasast.core.family_catalog import FamilyCatalogEntry +from openultrasast.core.lineage import LineageManifest, LineageUse, derive_lineage_manifest_digest +from openultrasast.core.measurement import ( + BaselineIdentity, + Comparator, + DenominatorKind, + DenominatorPolicy, + IntervalMethod, + MetricId, + MetricRule, + NumeratorKind, + ScopeKey, + Statistic, + derive_denominator_policy_digest, + derive_identity_set_digest, + standard_metric_rules, +) +from openultrasast.core.model import ModelProgramDeclaration +from openultrasast.core.readiness import ( + AnalysisIdentity, + FixtureResult, + FixtureRole, + ReadinessEvidenceUse, + ReadinessResult, + ReadinessStage, + StageResult, +) +from openultrasast.core.validation import EvidenceValidationContext + +_T = TypeVar("_T") + + +class MultiPopulationCustody: + def __init__(self, logs: tuple[custody.PopulationCustodyLog, ...], policy: custody.PopulationCustodyPolicy) -> None: + self.logs = {item.population_digest: item for item in logs} + self.policy = policy + + def current_log(self, population_digest: str) -> custody.PopulationCustodyLog: + try: + return self.logs[population_digest] + except KeyError as exc: + raise custody.PopulationCustodyError("population custody log not found") from exc + + def append(self, request: custody.PopulationUseRequest, custodian: str) -> custody.PopulationUseReceipt: + prior = self.current_log(request.population_digest) + receipt = custody.record_population_use(request, prior, custodian, self.policy) + self.logs[request.population_digest] = custody.append_population_use(prior, receipt, self.policy) + return receipt + + +class _PolicySubclass(DenominatorPolicy): + pass + + +class _RuleSubclass(MetricRule): + pass + + +class _TupleSubclass(tuple[MetricRule, ...]): + pass + + +class _ScopeSubclass(ScopeKey): + pass + + +class _ScalarSubclass(str): + pass + + +class _AdmissionContextSubclass(AdmissionValidationContext): + pass + + +class _EvidenceContextSubclass(EvidenceValidationContext): + pass + + +class _CatalogEntrySubclass(FamilyCatalogEntry): + pass + + +class _ModelProgramSubclass(ModelProgramDeclaration): + pass + + +class _LineageSubclass(LineageManifest): + pass + + +class _ReadinessResultSubclass(ReadinessResult): + pass + + +class _AnalysisIdentitySubclass(AnalysisIdentity): + pass + + +class _FixtureResultSubclass(FixtureResult): + pass + + +class _ReadinessTupleSubclass(tuple[ReadinessResult, ...]): + pass + + +def _as_subclass(value: object, subtype: type[_T]) -> _T: + changed = object.__new__(subtype) + for field, item in vars(value).items(): + object.__setattr__(changed, field, item) + return changed + + +@dataclass(frozen=True) +class AdmissionCase: + manifest: AdmissionManifestV2 + denominator: DenominatorPolicy + readiness: tuple[ReadinessResult, ...] + context: AdmissionValidationContext + port: MultiPopulationCustody + resolver: InMemoryContentResolver + + +@dataclass(frozen=True) +class ContextSetup: + model_case: ApplicationCase + port: MultiPopulationCustody + context: AdmissionValidationContext + threshold: AdmissionThreshold + threshold_lineage: LineageManifest + admission_population: str + + +def _append_lineage( + port: MultiPopulationCustody, + population: str, + arm: str, + purpose: custody.LineagePurpose, + artifact: str, + custodian: str, + *, + labels: bool = False, +) -> LineageManifest: + current = port.current_log(population) + request = custody.PopulationUseRequest(population, arm, purpose, labels, artifact, current.log_digest) + receipt = port.append(request, custodian) + use = LineageUse(purpose, population, labels, artifact, receipt.receipt_digest) + return LineageManifest(arm, artifact, (use,), derive_lineage_manifest_digest(arm, artifact, (use,))) + + +def _reviewed_census(population: str, scope: ScopeKey, disclosure: canonical.ContentRef) -> PopulationCensus: + producer = canonical.content_digest("benchmark-producer") + reviewer = canonical.content_digest("independent-reviewer") + row = ScopeCensus(scope, 8, 2, 2, 2, 4, 12, 3) + provisional_review = _attestation("0" * 64, producer, reviewer, disclosure) + provisional = PopulationCensus(population, (row,), False, "0" * 64, provisional_review, "0" * 64) + payload = derive_census_payload_digest(provisional) + census = replace(provisional, census_payload_digest=payload, review=_attestation(payload, producer, reviewer, disclosure)) + return replace(census, census_digest=derive_census_digest(census)) + + +def _resign_manifest( + manifest: AdmissionManifestV2, + disclosure: canonical.ContentRef, +) -> AdmissionManifestV2: + producer = canonical.content_digest("benchmark-producer") + reviewer = canonical.content_digest("independent-reviewer") + provisional = replace( + manifest, + manifest_payload_digest="0" * 64, + review=_attestation("0" * 64, producer, reviewer, disclosure), + manifest_digest="0" * 64, + ) + payload = derive_manifest_payload_digest(provisional) + reviewed = replace(provisional, manifest_payload_digest=payload, review=_attestation(payload, producer, reviewer, disclosure)) + return replace(reviewed, manifest_digest=derive_manifest_digest(reviewed)) + + +def _support_lineages( + port: MultiPopulationCustody, + source_population: str, + custodian: str, +) -> tuple[LineageManifest, LineageManifest]: + matcher = _append_lineage( + port, + source_population, + "matcher-arm", + custody.LineagePurpose.MATCHING_BEHAVIOR, + canonical.content_digest("matcher"), + custodian, + ) + promotion = _append_lineage( + port, + source_population, + "promotion-arm", + custody.LineagePurpose.PROMOTION, + canonical.content_digest("promotion"), + custodian, + ) + return matcher, promotion + + +def _context_setup() -> ContextSetup: + from openultrasast.core import admission_context + + model_case = _application_case() + contract = model_case.contract + catalog = _catalog(contract) + admission_population = canonical.content_digest("admission-population") + custodian = model_case.custodian + admission_log = custody.empty_population_custody_log(admission_population, custodian) + source_log = model_case.custody.current_log(canonical.content_digest("population")) + port = MultiPopulationCustody((source_log, admission_log), model_case.custody_policy) + matcher, promotion = _support_lineages(port, source_log.population_digest, custodian) + threshold = AdmissionThreshold(MetricId.VULNERABLE_RECALL, Statistic.WILSON_LOWER, Comparator.AT_LEAST, 0.5) + threshold_set = derive_threshold_set_digest((threshold,)) + threshold_lineage = _append_lineage( + port, + source_log.population_digest, + "reference-arm", + custody.LineagePurpose.THRESHOLD, + threshold_set, + custodian, + ) + evidence_context = validation.make_evidence_validation_context( + family_catalog=catalog, + fact_packs=(), + model_programs=(model_case.request.program,), + lineages=(model_case.request.lineage, matcher, promotion, threshold_lineage), + fact_review_policy=_review_policy(), + provenance_trust_policy=_trust_policy(), + custody_policy=model_case.custody_policy, + ) + review_policy = _policy(canonical.content_digest("independent-reviewer")) + context = admission_context.make_admission_validation_context( + evidence=evidence_context, + admission_review_policy=review_policy, + proof_trust_policy=None, + ) + return ContextSetup(model_case, port, context, threshold, threshold_lineage, admission_population) + + +def _baseline(setup: ContextSetup, denominator: DenominatorPolicy, scope: ScopeKey) -> BaselineIdentity: + from openultrasast.core import admission_context + + catalog = setup.context.evidence.family_catalog + program_digest = setup.model_case.request.program.declaration_digest + threshold_set = derive_threshold_set_digest((setup.threshold,)) + return BaselineIdentity( + "reference-arm", + setup.admission_population, + canonical.content_digest("admission-freeze"), + (scope,), + (catalog.entries[0].contract_digest,), + catalog.catalog_digest, + admission_context.derive_fact_schema_digest(), + validation.derive_fact_set_digest(()), + canonical.content_digest("acquisition"), + "analyzer@abc123", + canonical.content_digest("configuration"), + (program_digest,), + derive_identity_set_digest("model_program", (program_digest,)), + denominator.policy_digest, + threshold_set, + admission_context.derive_outcome_mapping_digest(catalog), + ) + + +def _manifest( + setup: ContextSetup, + baseline: BaselineIdentity, + census: PopulationCensus, + readiness: tuple[ReadinessResult, ...], + disclosure: canonical.ContentRef, +) -> AdmissionManifestV2: + provisional_review = _attestation( + "0" * 64, + canonical.content_digest("benchmark-producer"), + canonical.content_digest("independent-reviewer"), + disclosure, + ) + provisional = AdmissionManifestV2( + "0" * 64, + provisional_review, + True, + False, + baseline, + census, + tuple(sorted(item.result_digest for item in readiness)), + setup.port.current_log(setup.admission_population).log_digest, + baseline.denominator_policy_digest, + (setup.threshold,), + baseline.threshold_set_digest, + setup.threshold_lineage.manifest_digest, + setup.context.context_digest, + "0" * 64, + ) + return _resign_manifest(provisional, disclosure) + + +def _case() -> AdmissionCase: + from openultrasast.core import admission_context + + setup = _context_setup() + contract = setup.model_case.contract + scope = ScopeKey(contract.family_id, "python") + denominator = DenominatorPolicy(standard_metric_rules((scope,)), IntervalMethod.WILSON_95, "") + baseline = _baseline(setup, denominator, scope) + program_digest = setup.model_case.request.program.declaration_digest + catalog = setup.context.evidence.family_catalog + current_identity = admission_context.derive_analysis_identity(baseline, catalog, scope, program_digest) + readiness = _chain(identity=current_identity)[:6] + disclosure, resolver = stored_artifact(b'{"conflicts":[]}', artifact_id="admission-context-conflicts") + census = _reviewed_census(setup.admission_population, scope, disclosure) + manifest = _manifest(setup, baseline, census, readiness, disclosure) + return AdmissionCase(manifest, denominator, readiness, setup.context, setup.port, resolver) + + +def _validate(case: AdmissionCase) -> None: + from openultrasast.core.admission_context import validate_admission_context + + validate_admission_context( + case.manifest, + case.denominator, + case.readiness, + case.context, + case.port, + case.resolver, + ) + + +def _rebind_denominator(case: AdmissionCase, denominator: DenominatorPolicy) -> AdmissionCase: + object.__setattr__(denominator, "policy_digest", derive_denominator_policy_digest(denominator)) + identity = replace(case.manifest.identity, denominator_policy_digest=denominator.policy_digest) + manifest = _resign_manifest( + replace(case.manifest, identity=identity, denominator_policy_digest=denominator.policy_digest), + case.manifest.review.conflict_disclosure, + ) + return replace(case, manifest=manifest, denominator=denominator) + + +def _forged_policy(template: DenominatorPolicy, rules: tuple[MetricRule, ...]) -> DenominatorPolicy: + changed = object.__new__(DenominatorPolicy) + object.__setattr__(changed, "rules", rules) + object.__setattr__(changed, "interval_method", template.interval_method) + object.__setattr__(changed, "policy_digest", template.policy_digest) + return changed + + +def _smuggled_denominator(case: AdmissionCase, kind: str) -> DenominatorPolicy: + policy = case.denominator + first = policy.rules[0] + if kind == "policy_subclass": + changed_policy = object.__new__(_PolicySubclass) + object.__setattr__(changed_policy, "rules", policy.rules) + object.__setattr__(changed_policy, "interval_method", policy.interval_method) + object.__setattr__(changed_policy, "policy_digest", policy.policy_digest) + return changed_policy + if kind == "policy_state": + object.__setattr__(policy, "hidden", "state") + return policy + changed: MetricRule + if kind == "rule_subclass": + changed = _RuleSubclass(first.metric_id, first.scopes, first.numerator, first.denominator) + elif kind == "rule_state": + changed = replace(first) + object.__setattr__(changed, "hidden", "state") + elif kind == "scope_subclass": + scope = first.scopes[0] + changed = replace(first, scopes=(_ScopeSubclass(scope.family_id, scope.ecosystem),)) + elif kind == "scope_state": + scope = replace(first.scopes[0]) + object.__setattr__(scope, "hidden", "state") + changed = replace(first, scopes=(scope,)) + elif kind == "scalar_subclass": + scope = first.scopes[0] + changed = replace(first, scopes=(ScopeKey(_ScalarSubclass(scope.family_id), scope.ecosystem),)) + else: + return _forged_policy(policy, _TupleSubclass(policy.rules)) + return _forged_policy(policy, (changed, *policy.rules[1:])) + + +def test_exact_reviewed_context_is_accepted_before_feasibility() -> None: + case = _case() + _validate(case) + assert case.port.current_log(case.manifest.identity.population_digest).status is custody.PopulationStatus.AVAILABLE + + +@pytest.mark.parametrize( + "field", + ("family_contract_set_digest", "fact_set_digest", "model_program_digests", "outcome_mapping_digest"), +) +def test_claimed_digest_sets_are_rederived_from_canonical_objects(field: str) -> None: + case = _case() + if field == "model_program_digests": + models = (canonical.content_digest("other-model"),) + identity = replace( + case.manifest.identity, + model_program_digests=models, + model_program_set_digest=derive_identity_set_digest("model_program", models), + ) + elif field == "family_contract_set_digest": + identity = replace(case.manifest.identity, family_contract_set_digest=canonical.content_digest("other-catalog")) + elif field == "fact_set_digest": + identity = replace(case.manifest.identity, fact_set_digest=canonical.content_digest("other-facts")) + else: + identity = replace(case.manifest.identity, outcome_mapping_digest=canonical.content_digest("other-mapping")) + changed = _resign_manifest( + replace(case.manifest, identity=identity), + case.manifest.review.conflict_disclosure, + ) + with pytest.raises(ValueError, match="catalog|fact set|model program|outcome mapping"): + _validate(replace(case, manifest=changed)) + + +def test_scope_and_denominator_are_exact_not_digest_only() -> None: + case = _case() + other_scope = ScopeKey("other-family", "python") + denominator = DenominatorPolicy(standard_metric_rules((other_scope,)), IntervalMethod.WILSON_95, "") + identity = replace(case.manifest.identity, denominator_policy_digest=denominator.policy_digest) + changed = _resign_manifest( + replace( + case.manifest, + identity=identity, + denominator_policy_digest=denominator.policy_digest, + ), + case.manifest.review.conflict_disclosure, + ) + with pytest.raises(ValueError, match="denominator scope"): + _validate(replace(case, manifest=changed, denominator=denominator)) + + +def test_rederived_denominator_digest_cannot_hide_fixed_algebra_mutation() -> None: + case = _case() + first = case.denominator.rules[0] + changed_rule = replace(first, numerator=NumeratorKind.FIXED_DEFECT_CASES) + object.__setattr__(case.denominator, "rules", (changed_rule, *case.denominator.rules[1:])) + object.__setattr__(case.denominator, "policy_digest", derive_denominator_policy_digest(case.denominator)) + identity = replace(case.manifest.identity, denominator_policy_digest=case.denominator.policy_digest) + changed = _resign_manifest( + replace( + case.manifest, + identity=identity, + denominator_policy_digest=case.denominator.policy_digest, + ), + case.manifest.review.conflict_disclosure, + ) + + with pytest.raises(ValueError, match="fixed algebra"): + _validate(replace(case, manifest=changed)) + + +@pytest.mark.parametrize( + "kind", + ( + "policy_subclass", + "policy_state", + "rule_subclass", + "rule_state", + "tuple_subclass", + "scope_subclass", + "scope_state", + "scalar_subclass", + ), +) +def test_denominator_graph_rejects_subclasses_and_injected_state(kind: str) -> None: + case = _case() + changed = _rebind_denominator(case, _smuggled_denominator(case, kind)) + + with pytest.raises(ValueError, match="closed"): + _validate(changed) + + +@pytest.mark.parametrize( + "member", + ( + MetricId.PAIRED_CORRECTNESS, + NumeratorKind.CORRECT_PAIRS, + DenominatorKind.PAIRS, + IntervalMethod.WILSON_95, + Statistic.WILSON_LOWER, + Comparator.AT_LEAST, + ), +) +def test_admission_graph_rejects_injected_enum_singleton_state(member: object) -> None: + case = _case() + object.__setattr__(member, "hidden", "state") + try: + with pytest.raises(ValueError, match="closed"): + _validate(case) + finally: + object.__delattr__(member, "hidden") + + +@pytest.mark.parametrize("kind", ("subclass", "state")) +def test_admission_context_root_is_exact(kind: str) -> None: + case = _case() + context: AdmissionValidationContext + if kind == "subclass": + context = _as_subclass(case.context, _AdmissionContextSubclass) + else: + context = case.context + object.__setattr__(context, "hidden", "state") + + with pytest.raises(ValueError, match="closed|exact"): + _validate(replace(case, context=context)) + + +@pytest.mark.parametrize( + "kind", + ( + "root_subclass", + "root_state", + "entry_subclass", + "entry_state", + "program_subclass", + "program_state", + "lineage_subclass", + "lineage_state", + ), +) +def test_evidence_context_graph_is_closed_before_digest_validation(kind: str) -> None: + case = _case() + evidence = case.context.evidence + changed: EvidenceValidationContext + if kind == "root_subclass": + changed = _as_subclass(evidence, _EvidenceContextSubclass) + elif kind == "root_state": + changed = evidence + object.__setattr__(changed, "hidden", "state") + elif kind.startswith("entry"): + entry = evidence.family_catalog.entries[0] + changed_entry = _as_subclass(entry, _CatalogEntrySubclass) if kind.endswith("subclass") else entry + if kind.endswith("state"): + object.__setattr__(changed_entry, "hidden", "state") + catalog = replace(evidence.family_catalog, entries=(changed_entry,)) + changed = replace(evidence, family_catalog=catalog) + elif kind.startswith("program"): + program = evidence.model_programs[0] + changed_program = _as_subclass(program, _ModelProgramSubclass) if kind.endswith("subclass") else program + if kind.endswith("state"): + object.__setattr__(changed_program, "hidden", "state") + changed = replace(evidence, model_programs=(changed_program,)) + else: + lineage = evidence.lineages[0] + changed_lineage = _as_subclass(lineage, _LineageSubclass) if kind.endswith("subclass") else lineage + if kind.endswith("state"): + object.__setattr__(changed_lineage, "hidden", "state") + changed = replace(evidence, lineages=(changed_lineage, *evidence.lineages[1:])) + context = replace(case.context, evidence=changed) + + with pytest.raises(ValueError, match="closed"): + _validate(replace(case, context=context)) + + +@pytest.mark.parametrize( + "kind", + ( + "root_subclass", + "root_state", + "identity_subclass", + "identity_state", + "fixture_subclass", + "fixture_state", + "tuple_subclass", + ), +) +def test_readiness_graph_and_container_are_closed(kind: str) -> None: + case = _case() + first = case.readiness[0] + changed_first: ReadinessResult + if kind == "root_subclass": + changed_first = _as_subclass(first, _ReadinessResultSubclass) + elif kind == "root_state": + changed_first = first + object.__setattr__(changed_first, "hidden", "state") + elif kind.startswith("identity"): + identity = _as_subclass(first.identity, _AnalysisIdentitySubclass) if kind.endswith("subclass") else first.identity + if kind.endswith("state"): + object.__setattr__(identity, "hidden", "state") + changed_first = _as_subclass(first, ReadinessResult) + object.__setattr__(changed_first, "identity", identity) + elif kind.startswith("fixture"): + fixture = first.fixture_results[0] + changed_fixture = _as_subclass(fixture, _FixtureResultSubclass) if kind.endswith("subclass") else fixture + if kind.endswith("state"): + object.__setattr__(changed_fixture, "hidden", "state") + changed_first = _as_subclass(first, ReadinessResult) + object.__setattr__(changed_first, "fixture_results", (changed_fixture, *first.fixture_results[1:])) + else: + readiness: tuple[ReadinessResult, ...] = _ReadinessTupleSubclass(case.readiness) + with pytest.raises(ValueError, match="tuple"): + _validate(replace(case, readiness=readiness)) + return + readiness = (changed_first, *case.readiness[1:]) + + with pytest.raises(ValueError, match="closed"): + _validate(replace(case, readiness=readiness)) + + +@pytest.mark.parametrize( + "member", + ( + ReadinessStage.INSTRUMENT, + ReadinessEvidenceUse.REGRESSION_FIXTURE, + StageResult.PASSED, + FixtureRole.POSITIVE, + ), +) +def test_readiness_graph_rejects_injected_enum_singleton_state(member: object) -> None: + case = _case() + object.__setattr__(member, "hidden", "state") + try: + with pytest.raises(ValueError, match="closed readiness graph"): + _validate(case) + finally: + object.__delattr__(member, "hidden") + + +def test_missing_or_stale_readiness_is_refused() -> None: + case = _case() + shortened = case.readiness[:-1] + missing = _resign_manifest( + replace(case.manifest, readiness_digests=tuple(sorted(item.result_digest for item in shortened))), + case.manifest.review.conflict_disclosure, + ) + with pytest.raises(ValueError, match="development readiness"): + _validate(replace(case, manifest=missing, readiness=shortened)) + + stale_identity = replace(case.readiness[0].identity, acquisition_record_digest=canonical.content_digest("drift")) + stale = _chain(identity=stale_identity)[:6] + stale_manifest = _resign_manifest( + replace(case.manifest, readiness_digests=tuple(sorted(item.result_digest for item in stale))), + case.manifest.review.conflict_disclosure, + ) + with pytest.raises(ValueError, match="stale|analysis identity|acquisition"): + _validate(replace(case, manifest=stale_manifest, readiness=stale)) + + +def test_context_substitution_and_lineage_object_tampering_are_refused() -> None: + case = _case() + evidence_context = case.context.evidence + substituted_evidence = validation.make_evidence_validation_context( + family_catalog=evidence_context.family_catalog, + fact_packs=evidence_context.fact_packs, + model_programs=evidence_context.model_programs, + lineages=evidence_context.lineages[:-1], + fact_review_policy=evidence_context.fact_review_policy, + provenance_trust_policy=evidence_context.provenance_trust_policy, + custody_policy=evidence_context.custody_policy, + ) + from openultrasast.core import admission_context + + substituted = admission_context.make_admission_validation_context( + evidence=substituted_evidence, + admission_review_policy=case.context.admission_review_policy, + proof_trust_policy=None, + ) + with pytest.raises(ValueError, match="validation context"): + _validate(replace(case, context=substituted)) + + object.__setattr__(evidence_context.lineages[0].uses[0], "labels_accessed", True) + with pytest.raises(ValueError, match="lineage|label"): + _validate(case) + + +def test_label_influenced_promotion_is_refused_while_admission_population_is_available() -> None: + case = _case() + promotion = next(item for item in case.context.evidence.lineages if item.uses[0].purpose is custody.LineagePurpose.PROMOTION) + source = promotion.uses[0].source_population_digest + current = case.port.current_log(source) + request = custody.PopulationUseRequest( + source, + promotion.arm_id, + custody.LineagePurpose.PROMOTION, + True, + promotion.subject_digest, + current.log_digest, + ) + case.port.append(request, current.custodian_identity) + assert case.port.current_log(case.manifest.identity.population_digest).status is custody.PopulationStatus.AVAILABLE + with pytest.raises(ValueError, match="omitted current label-influenced use"): + _validate(case) + + +def test_current_custody_head_must_remain_available() -> None: + case = _case() + population = case.manifest.identity.population_digest + current = case.port.current_log(population) + request = custody.PopulationUseRequest( + population, + case.manifest.identity.arm_id, + custody.LineagePurpose.ADMISSION_LABEL_OPEN, + True, + case.manifest.manifest_digest, + current.log_digest, + ) + case.port.append(request, current.custodian_identity) + spent = _resign_manifest( + replace(case.manifest, custody_log_digest=case.port.current_log(population).log_digest), + case.manifest.review.conflict_disclosure, + ) + with pytest.raises(ValueError, match="spent|available"): + _validate(replace(case, manifest=spent))