diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 91101da..336380d 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -272,7 +272,7 @@ - Done when the reproduced vulnerable/fixed payload swap and every release-field mutation fail before any admission metric is returned. - _Requirements: 12.2, 12.5, 12.6, 13.4, 15.1_ - _Boundary: Admission Case Release_ -- [ ] 8.8 Derive scoped admission metrics from canonical case evidence +- [x] 8.8 Derive scoped admission metrics from canonical case evidence - Re-run shared evidence validation, instrument, budget, verdict, and proof validation for every case. - Consume only cases that exactly match the authenticated post-spend release; use its roles and scoped pairs while reconciling aggregate counts to the reviewed census. - Score quality only over pre-registered family/ecosystem scopes while retaining whole-population coverage. diff --git a/src/openultrasast/core/_admission_funnel.py b/src/openultrasast/core/_admission_funnel.py new file mode 100644 index 0000000..e74465e --- /dev/null +++ b/src/openultrasast/core/_admission_funnel.py @@ -0,0 +1,65 @@ +"""Canonical admission funnel derivation from validated case evidence.""" + +from __future__ import annotations + +from collections import Counter + +from openultrasast.core.admission_case import AdmissionCaseResult +from openultrasast.core.coverage import CoverageFunnel, ReasonCount, RelationStatusCount +from openultrasast.core.evidence import EvidenceRecord, NeutralOutcome +from openultrasast.core.semantics import RelationStatus, SemanticRelation + + +def _relation_reason(relation: SemanticRelation) -> str: + return "|".join(relation.reasons) + + +def derive_case_funnel( + case: AdmissionCaseResult, + records: tuple[EvidenceRecord, ...], + finding_count: int, +) -> CoverageFunnel: + transitions = tuple(item for record in records for item in record.packet.transitions) + observations = tuple(item for record in records for item in record.packet.observations) + statuses = Counter(item.status for item in transitions) + reasons = Counter(_relation_reason(item) for item in transitions if item.status is not RelationStatus.COMPLETE) + return CoverageFunnel( + inputs_read=len(case.input_receipts), + questions_formed=len(records), + sites_considered=len(transitions) + len(observations), + sites_recognized=sum(item.status is RelationStatus.COMPLETE for item in transitions) + + sum(item.status is RelationStatus.COMPLETE for item in observations), + transitions_materialized=len(transitions), + security_objects_constructed=len({record.packet.security_object_id for record in records}), + constraints_evaluated=len(records), + determinate_outcomes=sum(record.decision.outcome is not NeutralOutcome.INDETERMINATE for record in records), + findings_emitted=finding_count, + relation_status_counts=tuple(RelationStatusCount(status, statuses[status]) for status in RelationStatus), + unresolved_by_reason=tuple(ReasonCount(reason, count) for reason, count in sorted(reasons.items())), + ) + + +def aggregate_funnels(funnels: tuple[CoverageFunnel, ...]) -> CoverageFunnel: + statuses = tuple( + RelationStatusCount( + status, + sum(next(item.count for item in funnel.relation_status_counts if item.status is status) for funnel in funnels), + ) + for status in RelationStatus + ) + reasons: Counter[str] = Counter() + for funnel in funnels: + reasons.update({item.reason: item.count for item in funnel.unresolved_by_reason}) + return CoverageFunnel( + inputs_read=sum(item.inputs_read for item in funnels), + questions_formed=sum(item.questions_formed for item in funnels), + sites_considered=sum(item.sites_considered for item in funnels), + sites_recognized=sum(item.sites_recognized for item in funnels), + transitions_materialized=sum(item.transitions_materialized for item in funnels), + security_objects_constructed=sum(item.security_objects_constructed for item in funnels), + constraints_evaluated=sum(item.constraints_evaluated for item in funnels), + determinate_outcomes=sum(item.determinate_outcomes for item in funnels), + findings_emitted=sum(item.findings_emitted for item in funnels), + relation_status_counts=statuses, + unresolved_by_reason=tuple(ReasonCount(reason, count) for reason, count in sorted(reasons.items())), + ) diff --git a/src/openultrasast/core/admission_context.py b/src/openultrasast/core/admission_context.py index 0f3fbeb..a37c306 100644 --- a/src/openultrasast/core/admission_context.py +++ b/src/openultrasast/core/admission_context.py @@ -193,9 +193,9 @@ def _validate_scopes_and_sets(manifest: AdmissionManifestV2, denominator: Denomi 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: + if any(scope not in census_scopes for scope in baseline.claimed_scopes): + raise AdmissionContextError("every claimed scope must have one reviewed census row") + for scope in census_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: diff --git a/src/openultrasast/core/admission_feasibility.py b/src/openultrasast/core/admission_feasibility.py index 0f8b318..fe734af 100644 --- a/src/openultrasast/core/admission_feasibility.py +++ b/src/openultrasast/core/admission_feasibility.py @@ -88,7 +88,7 @@ def _validate_row(row: object) -> ScopeCensus: return row -def _capacities(census: PopulationCensus) -> tuple[tuple[ScopeKey, ...], _Capacities]: +def _capacities(census: PopulationCensus, selected_scopes: tuple[ScopeKey, ...]) -> _Capacities: if type(census) is not PopulationCensus or not census.rows: raise AdmissionFeasibilityError("census must be a nonempty PopulationCensus") if census.case_labels_exposed: @@ -99,13 +99,16 @@ def _capacities(census: PopulationCensus) -> tuple[tuple[ScopeKey, ...], _Capaci scopes = tuple(row.scope for row in rows) if scopes != tuple(sorted(scopes)) or len(scopes) != len(set(scopes)): raise AdmissionFeasibilityError("census scopes must be unique and canonically ordered") - return scopes, _Capacities( - sum(row.pair_count for row in rows), - sum(row.vulnerable_side_count for row in rows), - sum(row.fixed_side_count for row in rows), - sum(row.ordinary_case_count for row in rows), - sum(row.semantic_site_capacity for row in rows), - sum(row.model_call_capacity for row in rows), + if any(scope not in scopes for scope in selected_scopes): + raise AdmissionFeasibilityError("every claimed scope must have one reviewed census row") + selected = tuple(row for row in rows if row.scope in selected_scopes) + return _Capacities( + sum(row.pair_count for row in selected), + sum(row.vulnerable_side_count for row in selected), + sum(row.fixed_side_count for row in selected), + sum(row.ordinary_case_count for row in selected), + sum(row.semantic_site_capacity for row in selected), + sum(row.model_call_capacity for row in selected), ) @@ -122,9 +125,10 @@ def _validated_inputs( raise AdmissionFeasibilityError(str(exc)) from exc if policy.policy_digest != manifest.denominator_policy_digest: raise AdmissionFeasibilityError("denominator policy does not match the reviewed manifest") - scopes, capacities = _capacities(manifest.census) - if any(rule.scopes != scopes for rule in policy.rules): - raise AdmissionFeasibilityError("denominator policy scope does not match the census") + claimed_scopes = manifest.identity.claimed_scopes + if any(rule.scopes != claimed_scopes for rule in policy.rules): + raise AdmissionFeasibilityError("denominator policy scope does not match the manifest claimed scopes") + capacities = _capacities(manifest.census, claimed_scopes) return capacities, manifest.threshold_set_digest diff --git a/src/openultrasast/core/admission_metrics.py b/src/openultrasast/core/admission_metrics.py new file mode 100644 index 0000000..f04d079 --- /dev/null +++ b/src/openultrasast/core/admission_metrics.py @@ -0,0 +1,397 @@ +"""Canonical, scoped admission metric derivation from revalidated cases.""" + +from __future__ import annotations + +from collections import Counter +from collections.abc import Sequence +from dataclasses import dataclass + +from openultrasast.core import canonical +from openultrasast.core._admission_funnel import aggregate_funnels, derive_case_funnel +from openultrasast.core.admission_case import ( + AdmissionCaseRelease, + AdmissionCaseResult, + AdmissionCaseRole, + ReleasedCaseBinding, + derive_admission_case_result_digest, + derive_case_coverage_digest, + validate_admission_case_release, + validate_admission_case_shape, +) +from openultrasast.core.admission_case import ( + create_admission_case_result as create_admission_case_result, +) +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_review import AdmissionManifestV2, ScopeCensus, validate_reviewed_admission_payloads +from openultrasast.core.admission_spend import AdmissionPreparation +from openultrasast.core.budget import check_budget +from openultrasast.core.coverage import CoverageFunnel, ReasonCount, validate_instrument +from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt +from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, NeutralOutcome, derive_packet_digest +from openultrasast.core.family import FamilyContract +from openultrasast.core.family_catalog import resolve_contract +from openultrasast.core.measurement import DenominatorPolicy, MetricId, MetricResult, ScopeKey, evaluate_metric, validate_denominator_policy +from openultrasast.core.measurement_record import BasisCount, OutcomeCount +from openultrasast.core.model import ModelStatus +from openultrasast.core.proof import ProofAttachment, attach_differential_proof +from openultrasast.core.readiness import ReadinessResult +from openultrasast.core.validation import validate_evidence_input +from openultrasast.core.verdict_v2 import DefectClaim, derive_verdict_outputs + + +class AdmissionMetricsError(ValueError): + """Admission cases cannot produce trustworthy metrics.""" + + +@dataclass(frozen=True) +class AdmissionMetricEvidence: + family_decision_vocabulary: tuple[str, ...] + family_decision_counts: tuple[ReasonCount, ...] + neutral_outcome_counts: tuple[OutcomeCount, ...] + evidence_basis_counts: tuple[BasisCount, ...] + metrics: tuple[MetricResult, ...] + stage_funnel: CoverageFunnel + model_outcome_counts: tuple[ReasonCount, ...] + model_failures: tuple[str, ...] + runtime_ms: int + whole_population_funnel: CoverageFunnel + whole_population_runtime_ms: int + in_scope_case_count: int + whole_population_case_count: int + case_result_digests: tuple[str, ...] + + +@dataclass(frozen=True) +class _ValidatedCase: + supplied: AdmissionCaseResult + binding: ReleasedCaseBinding + records: tuple[EvidenceRecord, ...] + claims: tuple[DefectClaim, ...] + funnel: CoverageFunnel + + +@dataclass(frozen=True) +class _DerivedCounts: + vocabulary: tuple[str, ...] + decisions: tuple[ReasonCount, ...] + outcomes: tuple[OutcomeCount, ...] + bases: tuple[BasisCount, ...] + metrics: tuple[MetricResult, ...] + model_outcomes: tuple[ReasonCount, ...] + model_failures: tuple[str, ...] + + +def _case_shape(case: AdmissionCaseResult) -> None: + try: + validate_admission_case_shape(case) + except (TypeError, ValueError) as exc: + raise AdmissionMetricsError(str(exc)) from exc + if not case.evidence_inputs: + raise AdmissionMetricsError("case requires evidence inputs") + + +def _contract(case: AdmissionCaseResult, records: tuple[EvidenceRecord, ...], context: AdmissionValidationContext) -> FamilyContract: + digests = {item.packet.family_contract_digest for item in records} + if len(digests) != 1: + raise AdmissionMetricsError("case records must use one exact family contract") + contract = resolve_contract(context.evidence.family_catalog, case.scope.family_id, next(iter(digests)), case.scope.ecosystem) + for record in records: + packet = record.packet + if (packet.family_id, packet.ecosystem) != (case.scope.family_id, case.scope.ecosystem): + raise AdmissionMetricsError("case scope does not match validated evidence") + return contract + + +def _apply_proofs( + claims: tuple[DefectClaim, ...], + attachments: tuple[ProofAttachment, ...], + context: AdmissionValidationContext, + resolver: canonical.ContentResolver, +) -> tuple[DefectClaim, ...]: + if not attachments: + return claims + if context.proof_trust_policy is None: + raise AdmissionMetricsError("proof attachments require the context-bound trust policy") + keys = { + (item.record.packet.family_id, derive_packet_digest(item.record.packet), item.location): index for index, item in enumerate(claims) + } + if len(keys) != len(claims): + raise AdmissionMetricsError("proof target is ambiguous") + result = list(claims) + consumed: set[int] = set() + for attachment in attachments: + key = (attachment.family_id, attachment.packet_digest, attachment.location) + if key not in keys or keys[key] in consumed: + raise AdmissionMetricsError("proof attachment does not target one exact claim") + index = keys[key] + result[index] = attach_differential_proof(result[index], attachment, context.proof_trust_policy, resolver) + consumed.add(index) + return tuple(result) + + +def _validated_records( + case: AdmissionCaseResult, context: AdmissionValidationContext, custody: PopulationCustodyPort, resolver: canonical.ContentResolver +) -> tuple[EvidenceRecord, ...]: + _case_shape(case) + try: + validate_instrument(case.input_manifest, case.input_receipts) + except (TypeError, ValueError) as exc: + raise AdmissionMetricsError(f"instrument validation failed: {exc}") from exc + try: + check_budget(case.complexity, case.total_reply_ms, case.budget, resolver) + except (TypeError, ValueError) as exc: + raise AdmissionMetricsError(f"budget validation failed: {exc}") from exc + try: + records = tuple(validate_evidence_input(item, context.evidence, custody, resolver) for item in case.evidence_inputs) + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise AdmissionMetricsError(f"evidence validation failed: {exc}") from exc + if len({item.record_id for item in records}) != len(records): + raise AdmissionMetricsError("evidence inputs contain duplicate record identities") + scopes = {record.packet.scope for record in records} + if len(scopes) != 1: + raise AdmissionMetricsError("case records must share one exact analyzed scope") + analyzed_scope = next(iter(scopes)) + expected_complexity = ( + analyzed_scope.change_digest, + len(analyzed_scope.regions), + sum(len(record.packet.transitions) for record in records), + ) + actual_complexity = (case.complexity.change_id, case.complexity.changed_regions, case.complexity.transition_count) + if actual_complexity != expected_complexity: + raise AdmissionMetricsError("case complexity does not match revalidated evidence") + return records + + +def _validated_verdict( + case: AdmissionCaseResult, + records: tuple[EvidenceRecord, ...], + context: AdmissionValidationContext, + resolver: canonical.ContentResolver, +) -> tuple[tuple[DefectClaim, ...], CoverageFunnel]: + contract = _contract(case, records, context) + base_outputs = derive_verdict_outputs(records, contract, canonical.content_digest("admission-coverage-pending")) + funnel = derive_case_funnel(case, records, len(base_outputs.defects)) + coverage_digest = derive_case_coverage_digest(case.case_id_digest, case.scope, funnel) + outputs = derive_verdict_outputs(records, contract, coverage_digest) + claims = _apply_proofs(outputs.defects, case.proof_attachments, context, resolver) + if (claims, outputs.satisfactions, outputs.silence) != (case.defect_claims, case.satisfactions, case.silence): + raise AdmissionMetricsError("supplied verdict differs from the revalidated canonical verdict") + funnel = derive_case_funnel(case, records, len(claims)) + if case.coverage != funnel: + raise AdmissionMetricsError("coverage aggregate does not match revalidated case evidence") + if case.result_digest != derive_admission_case_result_digest(case): + raise AdmissionMetricsError("case result digest cannot be re-derived") + return claims, funnel + + +def _validate_case( + case: AdmissionCaseResult, + binding: ReleasedCaseBinding, + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> _ValidatedCase: + records = _validated_records(case, context, custody, resolver) + claims, funnel = _validated_verdict(case, records, context, resolver) + return _ValidatedCase(case, binding, records, claims, funnel) + + +def _validate_population(cases: tuple[_ValidatedCase, ...], rows: tuple[ScopeCensus, ...]) -> None: + by_scope = {row.scope: row for row in rows} + missing_scopes = tuple(sorted({item.binding.scope for item in cases if item.binding.scope not in by_scope})) + if missing_scopes: + raise AdmissionMetricsError("case scope has no reviewed census row") + if len(cases) != sum(row.total_cases for row in rows): + raise AdmissionMetricsError("case population count does not reconcile with the reviewed census") + for row in rows: + selected = tuple(item for item in cases if item.binding.scope == row.scope) + roles = Counter(item.binding.role for item in selected) + model_calls = sum( + value.model_result is not None for item in selected for value in item.supplied.evidence_inputs + ) + actual = ( + len(selected), + roles[AdmissionCaseRole.VULNERABLE], + roles[AdmissionCaseRole.FIXED], + roles[AdmissionCaseRole.ORDINARY], + sum(item.funnel.sites_considered for item in selected), + model_calls, + ) + expected = ( + row.total_cases, + row.vulnerable_side_count, + row.fixed_side_count, + row.ordinary_case_count, + row.semantic_site_capacity, + row.model_call_capacity, + ) + if actual != expected: + raise AdmissionMetricsError("case counts do not reconcile with the reviewed census") + paired: dict[str, list[AdmissionCaseRole]] = {} + for item in selected: + binding = item.binding + if binding.role is AdmissionCaseRole.ORDINARY and binding.pair_id_digest is not None: + raise AdmissionMetricsError("ordinary case cannot carry a pair identity") + if binding.pair_id_digest is not None: + paired.setdefault(binding.pair_id_digest, []).append(binding.role) + valid = {AdmissionCaseRole.VULNERABLE, AdmissionCaseRole.FIXED} + if len(paired) != row.pair_count or any(len(values) != 2 or set(values) != valid for values in paired.values()): + raise AdmissionMetricsError("case pairs do not reconcile with the reviewed census") + + +def _vocabulary(context: AdmissionValidationContext, scopes: tuple[ScopeKey, ...]) -> tuple[str, ...]: + values: list[str] = [] + for entry in context.evidence.family_catalog.entries: + if any(scope.family_id == entry.contract.family_id and scope.ecosystem in entry.contract.ecosystems for scope in scopes): + values.extend(rule.family_code for rule in entry.contract.decisions) + if not values or len(values) != len(set(values)): + raise AdmissionMetricsError("exact catalog decision vocabulary is empty or ambiguous") + return tuple(values) + + +def derive_scoped_pair_key(scope: ScopeKey, pair_id_digest: str) -> tuple[str, str, str]: + return scope.family_id, scope.ecosystem, pair_id_digest + + +def _metric_counts(cases: tuple[_ValidatedCase, ...], funnel: CoverageFunnel) -> dict[MetricId, tuple[int, int]]: + by_pair: dict[tuple[str, str, str], dict[AdmissionCaseRole, _ValidatedCase]] = {} + for item in cases: + if item.binding.pair_id_digest is not None: + key = derive_scoped_pair_key(item.binding.scope, item.binding.pair_id_digest) + by_pair.setdefault(key, {})[item.binding.role] = item + correct = sum(bool(pair[AdmissionCaseRole.VULNERABLE].claims) and not pair[AdmissionCaseRole.FIXED].claims for pair in by_pair.values()) + vulnerable = tuple(item for item in cases if item.binding.role is AdmissionCaseRole.VULNERABLE) + fixed = tuple(item for item in cases if item.binding.role is AdmissionCaseRole.FIXED) + ordinary = tuple(item for item in cases if item.binding.role is AdmissionCaseRole.ORDINARY) + defects = tuple(item for item in cases if item.claims) + model_results = tuple(value.model_result for item in cases for value in item.supplied.evidence_inputs if value.model_result is not None) + return { + MetricId.PAIRED_CORRECTNESS: (correct, len(by_pair)), + MetricId.VULNERABLE_RECALL: (sum(bool(item.claims) for item in vulnerable), len(vulnerable)), + MetricId.FIXED_FALSE_ALARM_RATE: (sum(bool(item.claims) for item in fixed), len(fixed)), + MetricId.ORDINARY_FALSE_ALARMS_PER_100: (sum(bool(item.claims) for item in ordinary), len(ordinary)), + MetricId.PRECISION: (sum(bool(item.claims) for item in vulnerable), len(defects)), + MetricId.SEMANTIC_STAGE_COVERAGE: (funnel.sites_recognized, funnel.sites_considered), + MetricId.MODEL_SUCCESS_RATE: (sum(item.status is ModelStatus.SUCCEEDED for item in model_results), len(model_results)), + } + + +def _validated_cases( + manifest: AdmissionManifestV2, + release: AdmissionCaseRelease, + cases: tuple[AdmissionCaseResult, ...], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> tuple[_ValidatedCase, ...]: + selected = cases + if not selected or len({item.case_id_digest for item in selected}) != len(selected): + raise AdmissionMetricsError("case population must be nonempty with unique identities") + bindings = {item.case_id_digest: item for item in release.bindings} + validated = tuple(_validate_case(item, bindings[item.case_id_digest], context, custody, resolver) for item in selected) + input_ids = tuple(value.input_digest for item in validated for value in item.supplied.evidence_inputs) + record_ids = tuple(record.record_id for item in validated for record in item.records) + if len(input_ids) != len(set(input_ids)) or len(record_ids) != len(set(record_ids)): + raise AdmissionMetricsError("case population repeats canonical evidence identities") + _validate_population(validated, manifest.census.rows) + return validated + + +def _derive_counts( + cases: tuple[_ValidatedCase, ...], + funnel: CoverageFunnel, + denominator: DenominatorPolicy, + context: AdmissionValidationContext, + scopes: tuple[ScopeKey, ...], +) -> _DerivedCounts: + counts = _metric_counts(cases, funnel) + metrics = tuple( + evaluate_metric(rule, numerator=counts[rule.metric_id][0], denominator=counts[rule.metric_id][1]) for rule in denominator.rules + ) + records = tuple(record for item in cases for record in item.records) + vocabulary = _vocabulary(context, scopes) + by_decision = Counter(record.decision.family_code for record in records) + decisions = tuple(ReasonCount(name, by_decision[name]) for name in vocabulary) + by_outcome = Counter(record.decision.outcome for record in records) + outcomes = tuple(OutcomeCount(outcome, by_outcome[outcome]) for outcome in NeutralOutcome) + by_basis = Counter(claim.basis for item in cases for claim in item.claims) + bases = tuple(BasisCount(basis, by_basis[basis]) for basis in EvidenceBasis) + model_results = tuple(value.model_result for item in cases for value in item.supplied.evidence_inputs if value.model_result is not None) + model_counts = Counter("success" if item.status is ModelStatus.SUCCEEDED else "failure" for item in model_results) + model_outcomes = tuple(ReasonCount(name, model_counts[name]) for name in ("failure", "success") if model_counts[name]) + failures = tuple(sorted({item.failure_reason for item in model_results if item.failure_reason is not None})) + return _DerivedCounts(vocabulary, decisions, outcomes, bases, metrics, model_outcomes, failures) + + +def _derive_validated_metrics( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + release: AdmissionCaseRelease, + cases: tuple[AdmissionCaseResult, ...], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionMetricEvidence: + validate_reviewed_admission_payloads(manifest.census, manifest, context.admission_review_policy, resolver) + validate_denominator_policy(denominator) + if manifest.validation_context_digest != context.context_digest: + raise AdmissionMetricsError("validation context does not match the reviewed manifest") + if denominator.policy_digest != manifest.denominator_policy_digest: + raise AdmissionMetricsError("denominator does not match the reviewed manifest") + validated = _validated_cases(manifest, release, cases, context, custody, resolver) + scopes = manifest.identity.claimed_scopes + in_scope = tuple(item for item in validated if item.binding.scope in scopes) + funnel = aggregate_funnels(tuple(item.funnel for item in in_scope)) + whole_funnel = aggregate_funnels(tuple(item.funnel for item in validated)) + counts = _derive_counts(in_scope, funnel, denominator, context, scopes) + return AdmissionMetricEvidence( + counts.vocabulary, + counts.decisions, + counts.outcomes, + counts.bases, + counts.metrics, + funnel, + counts.model_outcomes, + counts.model_failures, + sum(item.supplied.total_reply_ms for item in in_scope), + whole_funnel, + sum(item.supplied.total_reply_ms for item in validated), + len(in_scope), + len(validated), + tuple(sorted(item.supplied.result_digest for item in validated)), + ) + + +def derive_admission_metrics( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + preparation: AdmissionPreparation, + label_release_receipt: PopulationUseReceipt, + case_release: AdmissionCaseRelease, + cases: Sequence[AdmissionCaseResult], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionMetricEvidence: + """Revalidate immutable case evidence and derive only pre-registered quality metrics.""" + try: + frozen_cases = tuple(cases) + frozen_readiness = tuple(readiness) + validate_admission_case_release( + manifest, + denominator, + frozen_readiness, + context, + preparation, + label_release_receipt, + case_release, + frozen_cases, + custody, + resolver, + ) + return _derive_validated_metrics(manifest, denominator, case_release, frozen_cases, context, custody, resolver) + except AdmissionMetricsError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise AdmissionMetricsError(f"admission metric derivation failed before metric production: {exc}") from exc diff --git a/src/openultrasast/core/coverage.py b/src/openultrasast/core/coverage.py index d64266d..469f3d0 100644 --- a/src/openultrasast/core/coverage.py +++ b/src/openultrasast/core/coverage.py @@ -217,8 +217,6 @@ def _validate_relation_status_count(value: RelationStatusCount) -> None: def _validate_reason_count(value: ReasonCount) -> None: _require_text(value.reason, "reason") _require_count(value.count, "reason count") - if value.count == 0: - raise ValueError("reason count must be positive") def _validate_stage_expectation(value: StageExpectation) -> None: @@ -268,6 +266,8 @@ def _validate_reason_counts(funnel: CoverageFunnel) -> None: raise ValueError("unresolved_by_reason must contain ReasonCount values") for item in funnel.unresolved_by_reason: _validate_reason_count(item) + if item.count == 0: + raise ValueError("unresolved reason count must be positive") reasons = tuple(item.reason for item in funnel.unresolved_by_reason) if len(reasons) != len(set(reasons)): raise ValueError("duplicate unresolved reason") diff --git a/src/openultrasast/core/measurement_record.py b/src/openultrasast/core/measurement_record.py index 23c15dc..a844d3c 100644 --- a/src/openultrasast/core/measurement_record.py +++ b/src/openultrasast/core/measurement_record.py @@ -267,8 +267,8 @@ def _validate_record_totals(value: MeasurementRecordV2, decisions: tuple[ReasonC raise MeasurementRecordError("family decision counts must equal the question count") if sum(item.count for item in value.neutral_outcome_counts) != question_count: raise MeasurementRecordError("neutral outcome counts must equal the question count") - if sum(item.count for item in value.evidence_basis_counts) != question_count: - raise MeasurementRecordError("evidence basis counts must equal the question count") + if sum(item.count for item in value.evidence_basis_counts) != value.stage_funnel.findings_emitted: + raise MeasurementRecordError("evidence basis counts must equal emitted findings") determinate = sum(item.count for item in value.neutral_outcome_counts if item.outcome is not NeutralOutcome.INDETERMINATE) if determinate != value.stage_funnel.determinate_outcomes: raise MeasurementRecordError("determinate outcome counts must equal the funnel determinate count") diff --git a/tests/test_core_admission_feasibility.py b/tests/test_core_admission_feasibility.py index c4542c8..6889cf5 100644 --- a/tests/test_core_admission_feasibility.py +++ b/tests/test_core_admission_feasibility.py @@ -10,6 +10,7 @@ from detection_core_fixtures import InMemoryContentResolver, stored_artifact from openultrasast.core import canonical from openultrasast.core.admission_feasibility import ( AdmissionFeasibilityError, + FeasibleCountAssignment, ThresholdFeasibility, evaluate_threshold_feasibility, ) @@ -80,6 +81,17 @@ def _policy(scope: ScopeKey | None = None) -> DenominatorPolicy: return DenominatorPolicy(rules, IntervalMethod.WILSON_95, "") +def _with_extra_census_row(census: PopulationCensus) -> PopulationCensus: + extra = ScopeCensus(ScopeKey("out-of-scope", "javascript"), 300, 100, 100, 100, 100, 500, 200) + rows = tuple(sorted((*census.rows, extra), key=lambda item: item.scope)) + draft = replace(census, rows=rows, census_payload_digest="0" * 64, census_digest="0" * 64) + payload = derive_census_payload_digest(draft) + old = census.review + review = _attestation(payload, old.producer_identity, old.reviewer_identity, old.conflict_disclosure) + reviewed = replace(draft, census_payload_digest=payload, review=review) + return replace(reviewed, census_digest=derive_census_digest(reviewed)) + + def _thresholds(*values: AdmissionThreshold) -> tuple[AdmissionThreshold, ...]: return tuple(sorted(values, key=lambda item: (item.metric_id.value, item.statistic.value))) @@ -219,6 +231,14 @@ def test_enumerates_capacity_witnesses_for_every_metric_family() -> None: assert len(result.feasibility_digest) == 64 +def test_extra_census_scope_does_not_change_claimed_scope_feasibility() -> None: + threshold = _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 12.0) + + result = _evaluate(_with_extra_census_row(_census()), _policy(), (threshold,)) + + assert result.witnesses == (FeasibleCountAssignment(MetricId.PAIRED_CORRECTNESS, 12, 12),) + + def test_one_assignment_must_jointly_satisfy_grouped_statistics() -> None: thresholds = _thresholds( _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 10.0), diff --git a/tests/test_core_admission_funnel.py b/tests/test_core_admission_funnel.py new file mode 100644 index 0000000..032a653 --- /dev/null +++ b/tests/test_core_admission_funnel.py @@ -0,0 +1,62 @@ +"""Admission funnel fields are derived from canonical evidence, never runner counts.""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest +from test_core_admission_metrics import _derive, _release_for, _run, _setup + +from openultrasast.core.admission_metrics import AdmissionMetricsError, derive_admission_case_result_digest +from openultrasast.core.coverage import CoverageFunnel, ReasonCount, RelationStatusCount +from openultrasast.core.semantics import RelationStatus + + +def _mutate(value: CoverageFunnel, field: str) -> CoverageFunnel: + if field == "inputs_read": + return replace(value, inputs_read=2) + if field == "questions_formed": + return replace(value, questions_formed=2) + if field == "sites_considered": + return replace(value, sites_considered=3) + if field == "sites_recognized": + return replace(value, sites_recognized=1) + if field == "transitions_materialized": + counts = tuple(RelationStatusCount(status, 2 if status is RelationStatus.COMPLETE else 0) for status in RelationStatus) + return replace(value, transitions_materialized=2, relation_status_counts=counts) + if field == "security_objects_constructed": + return replace(value, security_objects_constructed=2) + if field == "constraints_evaluated": + return replace(value, constraints_evaluated=2) + if field == "determinate_outcomes": + return replace(value, determinate_outcomes=0) + if field == "findings_emitted": + return replace(value, findings_emitted=1) + counts = tuple(RelationStatusCount(status, int(status is RelationStatus.INCOMPLETE)) for status in RelationStatus) + return replace(value, relation_status_counts=counts, unresolved_by_reason=(ReasonCount("fabricated_incomplete_relation", 1),)) + + +@pytest.mark.parametrize( + "field", + ( + "inputs_read", + "questions_formed", + "sites_considered", + "sites_recognized", + "transitions_materialized", + "security_objects_constructed", + "constraints_evaluated", + "determinate_outcomes", + "findings_emitted", + "relation_accounting", + ), +) +def test_every_case_funnel_field_is_rederived_from_canonical_evidence(field: str) -> None: + run = _run(_setup()) + cases = run.setup.cases + changed = replace(cases[1], coverage=_mutate(cases[1].coverage, field)) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + changed_cases = (cases[0], changed, cases[2]) + + with pytest.raises(AdmissionMetricsError, match="coverage aggregate"): + _derive(run, cases=changed_cases, release=_release_for(run, changed_cases)) diff --git a/tests/test_core_admission_metrics.py b/tests/test_core_admission_metrics.py new file mode 100644 index 0000000..5a1d83f --- /dev/null +++ b/tests/test_core_admission_metrics.py @@ -0,0 +1,596 @@ +"""Canonical admission cases derive scoped metrics without runner aggregates.""" + +from __future__ import annotations + +from collections.abc import Iterator, Sequence +from dataclasses import replace +from typing import NamedTuple, TypeVar + +import pytest +from detection_core_fixtures import InMemoryContentResolver +from test_core_admission_context import AdmissionCase, MultiPopulationCustody, _attestation, _resign_manifest +from test_core_admission_release import _binding, _resign_release +from test_core_admission_spend import _feasible_case, _prepare, _spend +from test_core_consumer_v2 import _proof, _request +from test_core_model import _record as model_record +from test_core_readiness import _chain +from test_core_validation_v2 import _named_inputs, _with_fact_set + +from openultrasast.core import admission_context, canonical, provenance, validation +from openultrasast.core.admission_case import ADMISSION_CASE_RELEASE_SCHEMA, AdmissionCaseRelease +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_metrics import ( + AdmissionCaseResult, + AdmissionCaseRole, + AdmissionMetricsError, + create_admission_case_result, + derive_admission_case_result_digest, + derive_admission_metrics, + derive_scoped_pair_key, +) +from openultrasast.core.admission_review import ( + AdmissionManifestV2, + PopulationCensus, + ScopeCensus, + derive_census_digest, + derive_census_payload_digest, +) +from openultrasast.core.admission_spend import AdmissionPreparation +from openultrasast.core.coverage import ( + CoverageFunnel, + InputManifest, + InputReadReceipt, + RelationStatusCount, + derive_input_manifest_digest, +) +from openultrasast.core.custody import PopulationUseReceipt +from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, NeutralOutcome, derive_packet_digest +from openultrasast.core.family import FamilyContract, family_contract_digest +from openultrasast.core.family_catalog import FamilyCatalog, FamilyCatalogEntry, catalog_entry_key, derive_catalog_digest +from openultrasast.core.measurement import DenominatorPolicy, IntervalMethod, MetricId, ScopeKey, standard_metric_rules +from openultrasast.core.readiness import ReadinessResult +from openultrasast.core.semantics import RelationStatus +from openultrasast.core.verdict_v2 import derive_verdict_outputs + + +class _InputManifestSubclass(InputManifest): + pass + + +_T = TypeVar("_T") + + +class _TwoPassSequence(Sequence[_T]): + def __init__(self, first: tuple[_T, ...], second: tuple[_T, ...]) -> None: + self.first = first + self.second = second + self.traversals = 0 + + def __iter__(self) -> Iterator[_T]: + self.traversals += 1 + return iter(self.first if self.traversals == 1 else self.second) + + def __len__(self) -> int: + return len(self.first) + + def __getitem__(self, index: int | slice) -> _T | tuple[_T, ...]: + return self.first[index] + + +class _MetricSetup(NamedTuple): + manifest: AdmissionManifestV2 + denominator: DenominatorPolicy + cases: tuple[AdmissionCaseResult, ...] + context: AdmissionValidationContext + custody: MultiPopulationCustody + resolver: InMemoryContentResolver + readiness: tuple[ReadinessResult, ...] + + +class _MetricRun(NamedTuple): + setup: _MetricSetup + preparation: AdmissionPreparation + receipt: PopulationUseReceipt + release: AdmissionCaseRelease + + +def _run(setup: _MetricSetup) -> _MetricRun: + admission = AdmissionCase( + setup.manifest, + setup.denominator, + setup.readiness, + setup.context, + setup.custody, + setup.resolver, + ) + preparation = _prepare(admission) + receipt = _spend(admission, preparation) + bindings = tuple(sorted((_binding(case) for case in setup.cases), key=lambda item: item.case_id_digest)) + draft = AdmissionCaseRelease( + ADMISSION_CASE_RELEASE_SCHEMA, + setup.manifest.identity.population_digest, + setup.manifest.manifest_digest, + preparation.authorization.authorization_digest, + preparation.preparation_digest, + receipt.receipt_digest, + receipt.custodian_identity, + bindings, + "0" * 64, + setup.manifest.review, + "0" * 64, + ) + release = _resign_release(draft, admission, receipt) + return _MetricRun(setup, preparation, receipt, release) + + +def _derive( + run: _MetricRun, + *, + cases: tuple[AdmissionCaseResult, ...] | None = None, + release: AdmissionCaseRelease | None = None, +) -> object: + setup = run.setup + selected_cases = setup.cases if cases is None else cases + selected_release = run.release if release is None else release + return derive_admission_metrics( + setup.manifest, + setup.denominator, + setup.readiness, + run.preparation, + run.receipt, + selected_release, + selected_cases, + setup.context, + setup.custody, + setup.resolver, + ) + + +def _release_for(run: _MetricRun, cases: tuple[AdmissionCaseResult, ...]) -> AdmissionCaseRelease: + setup = run.setup + bindings = tuple(sorted((_binding(case) for case in cases), key=lambda item: item.case_id_digest)) + draft = replace(run.release, bindings=bindings, release_payload_digest="0" * 64, release_digest="0" * 64) + admission = AdmissionCase( + setup.manifest, + setup.denominator, + setup.readiness, + setup.context, + setup.custody, + setup.resolver, + ) + return _resign_release(draft, admission, run.receipt) + + +def _resolver(*values: InMemoryContentResolver) -> InMemoryContentResolver: + artifacts: dict[str, canonical.ResolvedContent] = {} + for value in values: + artifacts.update(value.artifacts) + return InMemoryContentResolver(artifacts) + + +def _rename_named_inputs( + named: provenance.NamedInputs, + resolver: InMemoryContentResolver, + prefix: str, +) -> tuple[provenance.NamedInputs, InMemoryContentResolver]: + artifacts: dict[str, canonical.ResolvedContent] = {} + items: list[provenance.NamedInput] = [] + for item in named.inputs: + artifact_id = f"{prefix}-{item.content.artifact_id}" + reference = replace(item.content, artifact_id=artifact_id) + artifacts[artifact_id] = resolver.artifacts[item.content.artifact_id] + items.append(replace(item, content=reference)) + values = tuple(items) + return provenance.NamedInputs(values, provenance.derive_named_inputs_digest(values)), InMemoryContentResolver(artifacts) + + +def _coverage(*, violation: bool) -> CoverageFunnel: + return CoverageFunnel( + 1, + 1, + 2, + 2, + 1, + 1, + 1, + 1, + int(violation), + ( + RelationStatusCount(RelationStatus.COMPLETE, 1), + RelationStatusCount(RelationStatus.INCOMPLETE, 0), + RelationStatusCount(RelationStatus.UNRESOLVED, 0), + ), + (), + ) + + +def _instrument(name: str) -> tuple[InputManifest, tuple[InputReadReceipt, ...]]: + manifest = InputManifest((name,), derive_input_manifest_digest((name,))) + receipt = InputReadReceipt(name, f"/{name}.py", True, 10, canonical.content_digest(name)) + return manifest, (receipt,) + + +def _case_result( + value: validation.EvidenceValidationInput, + role: AdmissionCaseRole, + pair: str | None, + resolver: InMemoryContentResolver, + contract: FamilyContract, +) -> tuple[AdmissionCaseResult, InMemoryContentResolver]: + request, combined = _request(value, resolver) + coverage = _coverage(violation=value.record.decision.outcome is NeutralOutcome.VIOLATION) + outputs = derive_verdict_outputs((value.record,), contract, canonical.content_digest("case-coverage")) + manifest, receipts = _instrument(role.value + (pair or "")) + result = create_admission_case_result( + case_id_digest=canonical.content_digest(role.value + (pair or "ordinary")), + scope=ScopeKey(value.record.packet.family_id, value.record.packet.ecosystem), + role=role, + pair_id_digest=pair, + evidence_inputs=(value,), + defect_claims=outputs.defects, + satisfactions=outputs.satisfactions, + silence=outputs.silence, + proof_attachments=(), + coverage=coverage, + input_manifest=manifest, + input_receipts=receipts, + complexity=request.complexity, + budget=request.budget, + total_reply_ms=10, + ) + return result, combined + + +def _resign_census(census: PopulationCensus, row: ScopeCensus, population_digest: str) -> PopulationCensus: + return _resign_census_rows(census, (row,), population_digest) + + +def _resign_census_rows( + census: PopulationCensus, + rows: tuple[ScopeCensus, ...], + population_digest: str, +) -> PopulationCensus: + changed = replace( + census, + population_digest=population_digest, + rows=rows, + census_payload_digest="0" * 64, + census_digest="0" * 64, + ) + payload = derive_census_payload_digest(changed) + old = census.review + review = _attestation(payload, old.producer_identity, old.reviewer_identity, old.conflict_disclosure) + changed = replace(changed, census_payload_digest=payload, review=review) + return replace(changed, census_digest=derive_census_digest(changed)) + + +def _setup() -> _MetricSetup: + admission = _feasible_case() + contract = admission.context.evidence.family_catalog.entries[0].contract + vulnerable_record = _with_fact_set(model_record(contract), ()) + vulnerable_named, vulnerable_resolver = _named_inputs(vulnerable_record, contract, ()) + vulnerable = validation.make_evidence_validation_input(record=vulnerable_record, named_inputs=vulnerable_named) + safe_decision = replace(vulnerable_record.decision, family_code="safe", outcome=NeutralOutcome.SATISFIED) + safe_record = EvidenceRecord(vulnerable_record.packet, safe_decision, "") + safe_named, safe_resolver = _named_inputs(safe_record, contract, ()) + safe = validation.make_evidence_validation_input(record=safe_record, named_inputs=safe_named) + ordinary_region = replace(safe_record.packet.scope.regions[0], end_line=19) + ordinary_scope = replace(safe_record.packet.scope, regions=(ordinary_region,)) + ordinary_packet = replace(safe_record.packet, scope=ordinary_scope) + ordinary_decision = replace( + safe_record.decision, + bounded_scope=ordinary_scope, + packet_digest=derive_packet_digest(ordinary_packet), + ) + ordinary_record = EvidenceRecord(ordinary_packet, ordinary_decision, "") + ordinary_named, ordinary_resolver = _named_inputs(ordinary_record, contract, ()) + ordinary = validation.make_evidence_validation_input(record=ordinary_record, named_inputs=ordinary_named) + resolver = _resolver(admission.resolver, vulnerable_resolver, safe_resolver, ordinary_resolver) + + pair = canonical.content_digest("pair-1") + vulnerable_case, resolver = _case_result(vulnerable, AdmissionCaseRole.VULNERABLE, pair, resolver, contract) + fixed_case, resolver = _case_result(safe, AdmissionCaseRole.FIXED, pair, resolver, contract) + ordinary_case, resolver = _case_result(ordinary, AdmissionCaseRole.ORDINARY, None, resolver, contract) + + scope = vulnerable_case.scope + policy = DenominatorPolicy(standard_metric_rules((scope,)), IntervalMethod.WILSON_95, "") + row = ScopeCensus(scope, 3, 1, 1, 1, 1, 6, 0) + cases = (vulnerable_case, fixed_case, ordinary_case) + population_digest = admission.manifest.census.population_digest + census = _resign_census(admission.manifest.census, row, population_digest) + identity = replace(admission.manifest.identity, population_digest=population_digest) + manifest = _resign_manifest( + replace(admission.manifest, identity=identity, census=census), + census.review.conflict_disclosure, + ) + return _MetricSetup(manifest, policy, cases, admission.context, admission.port, resolver, admission.readiness) + + +def test_metrics_are_derived_from_validated_cases_and_reviewed_counts() -> None: + run = _run(_setup()) + + result = _derive(run) + + by_metric = {item.metric_id: item for item in result.metrics} + assert (result.in_scope_case_count, result.whole_population_case_count) == (3, 3) + assert by_metric[MetricId.PAIRED_CORRECTNESS].numerator == 1 + assert by_metric[MetricId.VULNERABLE_RECALL].numerator == 1 + assert by_metric[MetricId.FIXED_FALSE_ALARM_RATE].numerator == 0 + assert by_metric[MetricId.ORDINARY_FALSE_ALARMS_PER_100].value == 0.0 + assert by_metric[MetricId.PRECISION].value == 1.0 + assert by_metric[MetricId.SEMANTIC_STAGE_COVERAGE].numerator == 6 + assert by_metric[MetricId.MODEL_SUCCESS_RATE].value is None + assert result.family_decision_vocabulary == ("unsafe", "safe") + assert tuple((item.reason, item.count) for item in result.family_decision_counts) == (("unsafe", 1), ("safe", 2)) + assert result.neutral_outcome_counts[0].count == 1 + assert result.evidence_basis_counts[tuple(EvidenceBasis).index(EvidenceBasis.DETERMINISTIC)].count == 1 + + +def test_caller_case_sequence_is_snapshotted_before_release_validation() -> None: + run = _run(_setup()) + original = run.setup.cases + substituted = replace(original[0], total_reply_ms=original[0].total_reply_ms + 1) + substituted = replace(substituted, result_digest=derive_admission_case_result_digest(substituted)) + switching = _TwoPassSequence(original, (substituted, *original[1:])) + + result = derive_admission_metrics( + run.setup.manifest, + run.setup.denominator, + run.setup.readiness, + run.preparation, + run.receipt, + run.release, + switching, + run.setup.context, + run.setup.custody, + run.setup.resolver, + ) + + assert switching.traversals == 1 + assert result.runtime_ms == sum(item.total_reply_ms for item in original) + assert result.case_result_digests == tuple(sorted(item.result_digest for item in original)) + + +def test_caller_readiness_sequence_is_snapshotted_once() -> None: + run = _run(_setup()) + switching = _TwoPassSequence(run.setup.readiness, run.setup.readiness[:-1]) + + result = derive_admission_metrics( + run.setup.manifest, + run.setup.denominator, + switching, + run.preparation, + run.receipt, + run.release, + run.setup.cases, + run.setup.context, + run.setup.custody, + run.setup.resolver, + ) + + assert switching.traversals == 1 + assert result.in_scope_case_count == len(run.setup.cases) + + +def test_case_count_and_pair_drift_fail_before_metric_production() -> None: + run = _run(_setup()) + cases = run.setup.cases + with pytest.raises(AdmissionMetricsError, match="release|binding|case population"): + _derive(run, cases=cases[:-1]) + + run = _run(_setup()) + cases = run.setup.cases + changed = replace(cases[0], pair_id_digest=None) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + with pytest.raises(AdmissionMetricsError, match="release|binding|pair"): + _derive(run, cases=(changed, *cases[1:])) + + +def test_supplied_verdict_and_instrument_are_not_trusted() -> None: + run = _run(_setup()) + cases = run.setup.cases + forged = replace(cases[0], defect_claims=()) + forged = replace(forged, result_digest=derive_admission_case_result_digest(forged)) + forged_cases = (forged, *cases[1:]) + with pytest.raises(AdmissionMetricsError, match="verdict"): + _derive(run, cases=forged_cases, release=_release_for(run, forged_cases)) + + run = _run(_setup()) + cases = run.setup.cases + unread = replace(cases[0].input_receipts[0], opened=False) + broken = replace(cases[0], input_receipts=(unread,)) + broken = replace(broken, result_digest=derive_admission_case_result_digest(broken)) + broken_cases = (broken, *cases[1:]) + with pytest.raises(AdmissionMetricsError, match="instrument|tracked input"): + _derive(run, cases=broken_cases, release=_release_for(run, broken_cases)) + + +def test_forged_evidence_and_budget_fail_before_metric_production() -> None: + run = _run(_setup()) + cases = run.setup.cases + forged_input = replace(cases[0].evidence_inputs[0], input_digest="0" * 64) + forged = replace(cases[0], evidence_inputs=(forged_input,)) + forged = replace(forged, result_digest=derive_admission_case_result_digest(forged)) + forged_cases = (forged, *cases[1:]) + with pytest.raises(AdmissionMetricsError, match="evidence"): + _derive(run, cases=forged_cases, release=_release_for(run, forged_cases)) + + run = _run(_setup()) + cases = run.setup.cases + over_budget = replace(cases[0], total_reply_ms=cases[0].budget.budget_ms + 1) + over_budget = replace(over_budget, result_digest=derive_admission_case_result_digest(over_budget)) + budget_cases = (over_budget, *cases[1:]) + with pytest.raises(AdmissionMetricsError, match="budget"): + _derive(run, cases=budget_cases, release=_release_for(run, budget_cases)) + + +def test_runner_case_digest_is_rederived() -> None: + run = _run(_setup()) + cases = run.setup.cases + forged = replace(cases[0], result_digest="0" * 64) + with pytest.raises(AdmissionMetricsError, match="case result digest"): + _derive(run, cases=(forged, *cases[1:])) + + +def test_proof_attachments_are_revalidated_with_the_context_policy() -> None: + run = _run(_setup()) + cases = run.setup.cases + attachment, _, proof_resolver = _proof(cases[0].evidence_inputs[0].record) + changed = replace(cases[0], proof_attachments=(attachment,)) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + changed_cases = (changed, *cases[1:]) + proof_run = run._replace(setup=run.setup._replace(resolver=_resolver(run.setup.resolver, proof_resolver))) + + with pytest.raises(AdmissionMetricsError, match="proof.*trust policy"): + _derive(proof_run, cases=changed_cases, release=_release_for(proof_run, changed_cases)) + + +def test_semantic_site_coverage_is_rederived_from_case_evidence() -> None: + run = _run(_setup()) + cases = run.setup.cases + changed_coverage = replace(cases[0].coverage, sites_recognized=1) + changed = replace(cases[0], coverage=changed_coverage) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + changed_cases = (changed, *cases[1:]) + + with pytest.raises(AdmissionMetricsError, match="coverage aggregate"): + _derive(run, cases=changed_cases, release=_release_for(run, changed_cases)) + + +def test_nested_case_graph_refuses_substituted_contract_types() -> None: + run = _run(_setup()) + cases = run.setup.cases + original = cases[0].input_manifest + substituted = _InputManifestSubclass(original.tracked_input_ids, original.manifest_digest) + changed = replace(cases[0], input_manifest=substituted) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + + with pytest.raises(AdmissionMetricsError, match="closed admission case"): + _derive(run, cases=(changed, *cases[1:])) + + +def test_pair_identity_is_namespaced_by_family_and_ecosystem() -> None: + digest = canonical.content_digest("shared-pair") + python = derive_scoped_pair_key(ScopeKey("family", "python"), digest) + javascript = derive_scoped_pair_key(ScopeKey("family", "javascript"), digest) + + assert python != javascript + + +def _with_out_of_scope_case(setup: _MetricSetup) -> _MetricSetup: + contract = setup.context.evidence.family_catalog.entries[0].contract + expanded_contract = replace(contract, family_version="1-js", ecosystems=("javascript",)) + javascript_entry = FamilyCatalogEntry(expanded_contract, family_contract_digest(expanded_contract)) + entries = tuple(sorted((*setup.context.evidence.family_catalog.entries, javascript_entry), key=catalog_entry_key)) + catalog = FamilyCatalog(entries, derive_catalog_digest(entries)) + evidence_context = validation.make_evidence_validation_context( + family_catalog=catalog, + fact_packs=setup.context.evidence.fact_packs, + model_programs=setup.context.evidence.model_programs, + lineages=setup.context.evidence.lineages, + fact_review_policy=setup.context.evidence.fact_review_policy, + provenance_trust_policy=setup.context.evidence.provenance_trust_policy, + custody_policy=setup.context.evidence.custody_policy, + ) + context = admission_context.make_admission_validation_context( + evidence=evidence_context, + admission_review_policy=setup.context.admission_review_policy, + proof_trust_policy=None, + ) + outside_record = _with_fact_set(model_record(expanded_contract), ()) + outside_packet = replace(outside_record.packet, ecosystem="javascript") + outside_decision = replace(outside_record.decision, packet_digest=derive_packet_digest(outside_packet)) + outside_record = EvidenceRecord(outside_packet, outside_decision, "") + outside_named, outside_resolver = _named_inputs(outside_record, expanded_contract, ()) + outside_named, outside_resolver = _rename_named_inputs(outside_named, outside_resolver, "javascript") + outside_input = validation.make_evidence_validation_input(record=outside_record, named_inputs=outside_named) + resolver = _resolver(setup.resolver, outside_resolver) + outside, resolver = _case_result(outside_input, AdmissionCaseRole.ORDINARY, None, resolver, expanded_contract) + outside = replace(outside, case_id_digest=canonical.content_digest("outside-javascript-case")) + outside = replace(outside, result_digest=derive_admission_case_result_digest(outside)) + cases = (*setup.cases, outside) + population_digest = setup.manifest.census.population_digest + row = setup.manifest.census.rows[0] + outside_row = ScopeCensus(outside.scope, 1, 0, 0, 0, 1, 2, 0) + rows = tuple(sorted((row, outside_row), key=lambda item: item.scope)) + census = _resign_census_rows(setup.manifest.census, rows, population_digest) + contract_digests = tuple(sorted(entry.contract_digest for entry in catalog.entries)) + identity = replace( + setup.manifest.identity, + population_digest=population_digest, + family_contract_digests=contract_digests, + family_contract_set_digest=catalog.catalog_digest, + outcome_mapping_digest=admission_context.derive_outcome_mapping_digest(catalog), + ) + program_digest = context.evidence.model_programs[0].declaration_digest + current_identity = admission_context.derive_analysis_identity(identity, catalog, identity.claimed_scopes[0], program_digest) + readiness = _chain(identity=current_identity)[:6] + manifest = _resign_manifest( + replace( + setup.manifest, + identity=identity, + census=census, + readiness_digests=tuple(sorted(item.result_digest for item in readiness)), + validation_context_digest=context.context_digest, + ), + census.review.conflict_disclosure, + ) + return _MetricSetup(manifest, setup.denominator, cases, context, setup.custody, resolver, readiness) + + +def test_whole_population_coverage_and_runtime_retain_out_of_scope_cases() -> None: + run = _run(_with_out_of_scope_case(_setup())) + + result = _derive(run) + + assert (result.in_scope_case_count, result.whole_population_case_count) == (3, 4) + assert (result.stage_funnel.questions_formed, result.whole_population_funnel.questions_formed) == (3, 4) + assert (result.runtime_ms, result.whole_population_runtime_ms) == (30, 40) + ordinary = next(item for item in result.metrics if item.metric_id is MetricId.ORDINARY_FALSE_ALARMS_PER_100) + assert ordinary.denominator == 1 + + +def _with_census_rows(setup: _MetricSetup, rows: tuple[ScopeCensus, ...]) -> _MetricSetup: + census = _resign_census_rows(setup.manifest.census, rows, setup.manifest.census.population_digest) + manifest = _resign_manifest( + replace(setup.manifest, census=census), + census.review.conflict_disclosure, + ) + return setup._replace(manifest=manifest) + + +def test_claimed_scope_must_have_a_reviewed_census_row() -> None: + setup = _setup() + unknown = ScopeCensus(ScopeKey("unknown-family", "ruby"), 1, 0, 0, 0, 1, 1, 0) + + with pytest.raises(ValueError, match="claimed scope.*census row"): + _run(_with_census_rows(setup, (unknown,))) + + +def test_every_additional_census_scope_must_resolve_in_the_catalog() -> None: + setup = _setup() + unknown = ScopeCensus(ScopeKey("unknown-family", "ruby"), 1, 0, 0, 0, 1, 1, 0) + rows = tuple(sorted((*setup.manifest.census.rows, unknown), key=lambda item: item.scope)) + + with pytest.raises(ValueError, match="resolve one exact family contract"): + _run(_with_census_rows(setup, rows)) + + +def test_released_case_without_a_census_row_is_refused_before_metrics() -> None: + setup = _with_out_of_scope_case(_setup()) + claimed = tuple(row for row in setup.manifest.census.rows if row.scope in setup.manifest.identity.claimed_scopes) + run = _run(_with_census_rows(setup, claimed)) + + with pytest.raises(AdmissionMetricsError, match="case scope.*census row"): + _derive(run) + + +@pytest.mark.parametrize("field", ("semantic_site_capacity", "model_call_capacity")) +def test_released_case_aggregates_must_match_every_census_capacity(field: str) -> None: + setup = _setup() + row = setup.manifest.census.rows[0] + changed = replace(row, **{field: getattr(row, field) + 1}) + run = _run(_with_census_rows(setup, (changed,))) + + with pytest.raises(AdmissionMetricsError, match="case counts.*census"): + _derive(run) diff --git a/tests/test_core_coverage_v2.py b/tests/test_core_coverage_v2.py index 50014fc..2234e58 100644 --- a/tests/test_core_coverage_v2.py +++ b/tests/test_core_coverage_v2.py @@ -74,6 +74,16 @@ def test_input_manifest_and_receipts_validate_exactly() -> None: validate_instrument(_manifest(), _receipts()) +def test_zero_named_counts_are_allowed_for_measurements_but_not_unresolved_coverage() -> None: + with pytest.raises(ValueError, match="unresolved reason count must be positive"): + _funnel( + unresolved_by_reason=( + ReasonCount("missing_constraint", 2), + ReasonCount("unused_measurement_bucket", 0), + ) + ) + + @pytest.mark.parametrize( ("opened", "byte_count", "digest", "match"), [ diff --git a/tests/test_core_measurement_record.py b/tests/test_core_measurement_record.py index c296ac4..000b2b2 100644 --- a/tests/test_core_measurement_record.py +++ b/tests/test_core_measurement_record.py @@ -80,7 +80,7 @@ def _record(**overrides: object) -> MeasurementRecordV2: OutcomeCount(NeutralOutcome.INDETERMINATE, 0), ), "evidence_basis_counts": ( - BasisCount(EvidenceBasis.DETERMINISTIC, 2), + BasisCount(EvidenceBasis.DETERMINISTIC, 1), BasisCount(EvidenceBasis.MODEL_CORROBORATED, 0), BasisCount(EvidenceBasis.EXECUTION_BACKED, 0), ), @@ -182,6 +182,7 @@ def test_clean_or_zero_looking_results_require_read_inputs_and_passed_stages() - OutcomeCount(NeutralOutcome.SATISFIED, 2), OutcomeCount(NeutralOutcome.INDETERMINATE, 0), ), + evidence_basis_counts=tuple(BasisCount(basis, 0) for basis in EvidenceBasis), metrics=_metrics(model_numerator=2, model_denominator=2), stage_funnel=_funnel(findings=0), model_outcome_counts=(ReasonCount("success", 2),), @@ -359,6 +360,19 @@ def test_family_decision_counts_follow_declared_not_lexical_order() -> None: assert tuple(item.reason for item in record.family_decision_counts) == record.family_decision_vocabulary +def test_family_decision_counts_retain_zero_entries_for_exact_vocabulary() -> None: + record = _record( + family_decision_vocabulary=("allow", "deny", "unknown"), + family_decision_counts=(ReasonCount("allow", 1), ReasonCount("deny", 1), ReasonCount("unknown", 0)), + ) + + assert tuple((item.reason, item.count) for item in record.family_decision_counts) == ( + ("allow", 1), + ("deny", 1), + ("unknown", 0), + ) + + def test_holdout_report_refuses_wrong_role_overlap_or_empty_manifests() -> None: report = _report() with pytest.raises(MeasurementRecordError, match="repository_holdout"):