From 67b3ae62a969ef885cd52aea0aa6504aadbf5b2c Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 1 Nov 2026 09:03:44 +0100 Subject: [PATCH] feat(detection-core): validate threshold feasibility --- .kiro/specs/detection-core/tasks.md | 2 +- .../core/admission_feasibility.py | 220 +++++++++++++ tests/test_core_admission_feasibility.py | 308 ++++++++++++++++++ 3 files changed, 529 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/admission_feasibility.py create mode 100644 tests/test_core_admission_feasibility.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index a0b8975..2cf135c 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -250,7 +250,7 @@ - Reject label exposure, acquisition drift, context substitution, stale readiness, scope mismatch, or prior spend. - Done when each refusal branch fails before feasibility and no digest-only set or lineage claim is accepted. - _Requirements: 12.1, 12.2, 13.1_ -- [ ] 8.5 Refuse arithmetically impossible threshold configurations +- [x] 8.5 Refuse arithmetically impossible threshold configurations - Enumerate reachable counts from label-blind pair, case, semantic-site, and model-call capacities. - Evaluate count, rate, per-100, and Wilson-bound thresholds with the fixed comparator direction for each metric. - Require every grouped threshold to have at least one jointly attainable count assignment. diff --git a/src/openultrasast/core/admission_feasibility.py b/src/openultrasast/core/admission_feasibility.py new file mode 100644 index 0000000..0f8b318 --- /dev/null +++ b/src/openultrasast/core/admission_feasibility.py @@ -0,0 +1,220 @@ +"""Label-blind arithmetic feasibility for admission thresholds.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from openultrasast.core import canonical +from openultrasast.core.admission_review import ( + AdmissionManifestV2, + AdmissionThreshold, + PopulationCensus, + ScopeCensus, + validate_reviewed_admission_payloads, +) +from openultrasast.core.factpack import ReviewIndependencePolicy +from openultrasast.core.measurement import ( + DenominatorPolicy, + MeasurementContractError, + MetricId, + MetricRule, + ScopeKey, + evaluate_metric, + metric_passes, + validate_denominator_policy, +) + +FEASIBILITY_SCHEMA = "openultrasast.detection-core.threshold-feasibility.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class AdmissionFeasibilityError(ValueError): + """An admission threshold has no reachable count assignment.""" + + +@dataclass(frozen=True) +class FeasibleCountAssignment: + metric_id: MetricId + numerator: int + denominator: int + + +@dataclass(frozen=True) +class ThresholdFeasibility: + census_digest: str + denominator_policy_digest: str + threshold_set_digest: str + witnesses: tuple[FeasibleCountAssignment, ...] + feasibility_digest: str + + +@dataclass(frozen=True) +class _Capacities: + pairs: int + vulnerable: int + fixed: int + ordinary: int + sites: int + model_calls: int + + +def _count(value: object, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise AdmissionFeasibilityError(f"{field} must be a non-negative integer") + return value + + +def _validate_row(row: object) -> ScopeCensus: + if type(row) is not ScopeCensus: + raise AdmissionFeasibilityError("census rows must be exact ScopeCensus values") + counts = tuple( + _count(getattr(row, field), field) + for field in ( + "total_cases", + "pair_count", + "vulnerable_side_count", + "fixed_side_count", + "ordinary_case_count", + "semantic_site_capacity", + "model_call_capacity", + ) + ) + total, pairs, vulnerable, fixed, ordinary, _, _ = counts + if total != vulnerable + fixed + ordinary: + raise AdmissionFeasibilityError("census total does not reconcile with aggregate roles") + if pairs > min(vulnerable, fixed): + raise AdmissionFeasibilityError("census pair count exceeds side capacity") + return row + + +def _capacities(census: PopulationCensus) -> tuple[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: + raise AdmissionFeasibilityError("feasibility requires a label-blind census") + if _DIGEST.fullmatch(census.census_digest) is None: + raise AdmissionFeasibilityError("census digest must be a full lowercase SHA-256 digest") + rows = tuple(_validate_row(row) for row in census.rows) + 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), + ) + + +def _validated_inputs( + manifest: AdmissionManifestV2, + policy: DenominatorPolicy, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> tuple[_Capacities, str]: + try: + validate_reviewed_admission_payloads(manifest.census, manifest, review_policy, resolver) + validate_denominator_policy(policy) + except (AttributeError, LookupError, TypeError, ValueError) as exc: + 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") + return capacities, manifest.threshold_set_digest + + +def _best_counts(metric_id: MetricId, capacities: _Capacities) -> tuple[int, int]: + """Return the jointly optimal reachable point under the fixed comparator.""" + + fixed = { + MetricId.PAIRED_CORRECTNESS: capacities.pairs, + MetricId.VULNERABLE_RECALL: capacities.vulnerable, + MetricId.FIXED_FALSE_ALARM_RATE: capacities.fixed, + MetricId.ORDINARY_FALSE_ALARMS_PER_100: capacities.ordinary, + MetricId.SEMANTIC_STAGE_COVERAGE: capacities.sites, + MetricId.MODEL_SUCCESS_RATE: capacities.model_calls, + } + if metric_id is MetricId.PRECISION: + if capacities.vulnerable: + return capacities.vulnerable, capacities.vulnerable + false_positive_capacity = capacities.fixed + capacities.ordinary + return (0, 1) if false_positive_capacity else (0, 0) + denominator = fixed[metric_id] + false_alarm = metric_id in { + MetricId.FIXED_FALSE_ALARM_RATE, + MetricId.ORDINARY_FALSE_ALARMS_PER_100, + } + return (0 if false_alarm else denominator), denominator + + +def _grouped_thresholds( + thresholds: tuple[AdmissionThreshold, ...], +) -> tuple[tuple[MetricId, tuple[AdmissionThreshold, ...]], ...]: + return tuple( + (metric_id, tuple(item for item in thresholds if item.metric_id is metric_id)) + for metric_id in MetricId + if any(item.metric_id is metric_id for item in thresholds) + ) + + +def _witness( + rule: MetricRule, + thresholds: tuple[AdmissionThreshold, ...], + capacities: _Capacities, +) -> FeasibleCountAssignment: + numerator, denominator = _best_counts(rule.metric_id, capacities) + try: + result = evaluate_metric(rule, numerator=numerator, denominator=denominator) + outcomes = tuple(metric_passes(result, item.statistic, item.comparator, item.target) for item in thresholds) + except MeasurementContractError as exc: + raise AdmissionFeasibilityError(str(exc)) from exc + if not all(outcomes): + raise AdmissionFeasibilityError(f"{rule.metric_id.value} thresholds are arithmetically impossible") + return FeasibleCountAssignment(rule.metric_id, numerator, denominator) + + +def _feasibility_value( + census: PopulationCensus, + policy: DenominatorPolicy, + threshold_digest: str, + witnesses: tuple[FeasibleCountAssignment, ...], +) -> canonical.JsonValue: + return { + "schema_version": FEASIBILITY_SCHEMA, + "census_digest": census.census_digest, + "denominator_policy_digest": policy.policy_digest, + "threshold_set_digest": threshold_digest, + "witnesses": tuple( + { + "metric_id": item.metric_id.value, + "numerator": item.numerator, + "denominator": item.denominator, + } + for item in witnesses + ), + } + + +def evaluate_threshold_feasibility( + manifest: AdmissionManifestV2, + policy: DenominatorPolicy, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> ThresholdFeasibility: + capacities, threshold_digest = _validated_inputs(manifest, policy, review_policy, resolver) + rules = {rule.metric_id: rule for rule in policy.rules} + witnesses = tuple(_witness(rules[metric_id], group, capacities) for metric_id, group in _grouped_thresholds(manifest.thresholds)) + census = manifest.census + digest = canonical.content_digest(_feasibility_value(census, policy, threshold_digest, witnesses)) + return ThresholdFeasibility( + census.census_digest, + policy.policy_digest, + threshold_digest, + witnesses, + digest, + ) diff --git a/tests/test_core_admission_feasibility.py b/tests/test_core_admission_feasibility.py new file mode 100644 index 0000000..c4542c8 --- /dev/null +++ b/tests/test_core_admission_feasibility.py @@ -0,0 +1,308 @@ +"""Label-blind arithmetic admission feasibility (Requirements 12.3 and 12.4).""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical +from openultrasast.core.admission_feasibility import ( + AdmissionFeasibilityError, + ThresholdFeasibility, + evaluate_threshold_feasibility, +) +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.factpack import ( + IndependenceAttestation, + ReviewIndependencePolicy, + derive_independence_attestation_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.measurement import ( + BaselineIdentity, + Comparator, + DenominatorPolicy, + IntervalMethod, + MetricId, + ScopeKey, + Statistic, + standard_metric_rules, +) + + +def _scope() -> ScopeKey: + return ScopeKey("access-control", "python") + + +def _census( + *, + pairs: int = 12, + vulnerable: int = 12, + fixed: int = 12, + ordinary: int = 20, + sites: int = 8, + calls: int = 5, +) -> PopulationCensus: + row = ScopeCensus( + _scope(), + vulnerable + fixed + ordinary, + pairs, + vulnerable, + fixed, + ordinary, + sites, + calls, + ) + disclosure, _ = stored_artifact(b'{"conflicts":[]}', artifact_id="feasibility-conflicts") + producer = canonical.content_digest("benchmark-producer") + reviewer = canonical.content_digest("independent-reviewer") + provisional_review = _attestation("0" * 64, producer, reviewer, disclosure) + provisional = PopulationCensus(canonical.content_digest("population"), (row,), False, "0" * 64, provisional_review, "0" * 64) + payload = derive_census_payload_digest(provisional) + reviewed = replace(provisional, census_payload_digest=payload, review=_attestation(payload, producer, reviewer, disclosure)) + return replace(reviewed, census_digest=derive_census_digest(reviewed)) + + +def _policy(scope: ScopeKey | None = None) -> DenominatorPolicy: + rules = standard_metric_rules((scope or _scope(),)) + return DenominatorPolicy(rules, IntervalMethod.WILSON_95, "") + + +def _thresholds(*values: AdmissionThreshold) -> tuple[AdmissionThreshold, ...]: + return tuple(sorted(values, key=lambda item: (item.metric_id.value, item.statistic.value))) + + +def _threshold( + metric: MetricId, + statistic: Statistic, + comparator: Comparator, + target: float, +) -> AdmissionThreshold: + return AdmissionThreshold(metric, statistic, comparator, target) + + +def _attestation( + subject: str, + producer: str, + reviewer: str, + disclosure: canonical.ContentRef, +) -> IndependenceAttestation: + provisional = IndependenceAttestation(subject, producer, reviewer, disclosure, "0" * 64) + return replace(provisional, attestation_digest=derive_independence_attestation_digest(provisional)) + + +def _review_policy(reviewer: str) -> ReviewIndependencePolicy: + stores = ("trusted-store",) + digest = derive_review_independence_policy_digest((reviewer,), (), stores, 4096) + return ReviewIndependencePolicy((reviewer,), (), stores, 4096, digest) + + +def _baseline( + census: PopulationCensus, + denominator: DenominatorPolicy, + threshold_digest: str, +) -> BaselineIdentity: + return BaselineIdentity( + "arm-a", + census.population_digest, + canonical.content_digest("freeze"), + denominator.rules[0].scopes, + (canonical.content_digest("contract"),), + canonical.content_digest("catalog"), + canonical.content_digest("fact-schema"), + canonical.content_digest("fact-set"), + canonical.content_digest("acquisition"), + "analyzer-v2", + canonical.content_digest("configuration"), + (), + None, + denominator.policy_digest, + threshold_digest, + canonical.content_digest("outcomes"), + ) + + +def _reviewed_manifest( + census: PopulationCensus, + denominator: DenominatorPolicy, + thresholds: tuple[AdmissionThreshold, ...], +) -> tuple[AdmissionManifestV2, ReviewIndependencePolicy, InMemoryContentResolver]: + disclosure, resolver = stored_artifact(b'{"conflicts":[]}', artifact_id="feasibility-conflicts") + producer = canonical.content_digest("benchmark-producer") + reviewer = canonical.content_digest("independent-reviewer") + threshold_digest = derive_threshold_set_digest(thresholds) + baseline = _baseline(census, denominator, threshold_digest) + provisional_review = _attestation("0" * 64, producer, reviewer, disclosure) + provisional = AdmissionManifestV2( + "0" * 64, + provisional_review, + True, + False, + baseline, + census, + (canonical.content_digest("readiness"),), + canonical.content_digest("custody"), + denominator.policy_digest, + thresholds, + threshold_digest, + canonical.content_digest("threshold-lineage"), + canonical.content_digest("validation-context"), + "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)), _review_policy(reviewer), resolver + + +def _evaluate( + census: PopulationCensus, + denominator: DenominatorPolicy, + thresholds: tuple[AdmissionThreshold, ...], +) -> ThresholdFeasibility: + manifest, review_policy, resolver = _reviewed_manifest(census, denominator, thresholds) + return evaluate_threshold_feasibility(manifest, denominator, review_policy, resolver) + + +def test_stale_reviewed_census_identity_is_refused_before_capacity_use() -> None: + thresholds = (_threshold(MetricId.SEMANTIC_STAGE_COVERAGE, Statistic.COUNT, Comparator.AT_LEAST, 50.0),) + census = _census(sites=12) + denominator = _policy() + manifest, review_policy, resolver = _reviewed_manifest(census, denominator, thresholds) + changed_row = replace(census.rows[0], semantic_site_capacity=99) + stale_census = replace(census, rows=(changed_row,)) + stale_manifest = replace(manifest, census=stale_census) + + with pytest.raises(AdmissionFeasibilityError, match="census payload"): + evaluate_threshold_feasibility(stale_manifest, denominator, review_policy, resolver) + + +def test_enumerates_capacity_witnesses_for_every_metric_family() -> None: + thresholds = _thresholds( + _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 6.0), + _threshold(MetricId.VULNERABLE_RECALL, Statistic.RATE, Comparator.AT_LEAST, 0.5), + _threshold(MetricId.FIXED_FALSE_ALARM_RATE, Statistic.RATE, Comparator.AT_MOST, 0.1), + _threshold(MetricId.FIXED_FALSE_ALARM_RATE, Statistic.WILSON_UPPER, Comparator.AT_MOST, 0.25), + _threshold(MetricId.ORDINARY_FALSE_ALARMS_PER_100, Statistic.PER_100, Comparator.AT_MOST, 5.0), + _threshold(MetricId.ORDINARY_FALSE_ALARMS_PER_100, Statistic.WILSON_LOWER, Comparator.AT_MOST, 0.0), + _threshold(MetricId.PRECISION, Statistic.WILSON_LOWER, Comparator.AT_LEAST, 0.7), + _threshold(MetricId.SEMANTIC_STAGE_COVERAGE, Statistic.COUNT, Comparator.AT_LEAST, 4.0), + _threshold(MetricId.MODEL_SUCCESS_RATE, Statistic.WILSON_UPPER, Comparator.AT_LEAST, 0.9), + ) + + result = _evaluate(_census(), _policy(), thresholds) + + witnesses = {item.metric_id: (item.numerator, item.denominator) for item in result.witnesses} + assert witnesses == { + MetricId.PAIRED_CORRECTNESS: (12, 12), + MetricId.VULNERABLE_RECALL: (12, 12), + MetricId.FIXED_FALSE_ALARM_RATE: (0, 12), + MetricId.ORDINARY_FALSE_ALARMS_PER_100: (0, 20), + MetricId.PRECISION: (12, 12), + MetricId.SEMANTIC_STAGE_COVERAGE: (8, 8), + MetricId.MODEL_SUCCESS_RATE: (5, 5), + } + assert result.census_digest == _census().census_digest + assert result.denominator_policy_digest == _policy().policy_digest + assert len(result.threshold_set_digest) == 64 + assert len(result.feasibility_digest) == 64 + + +def test_one_assignment_must_jointly_satisfy_grouped_statistics() -> None: + thresholds = _thresholds( + _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 10.0), + _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.RATE, Comparator.AT_LEAST, 0.8), + _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.WILSON_LOWER, Comparator.AT_LEAST, 0.5), + _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.WILSON_UPPER, Comparator.AT_LEAST, 0.8), + ) + + result = _evaluate(_census(), _policy(), thresholds) + + assert [(item.metric_id, item.numerator, item.denominator) for item in result.witnesses] == [(MetricId.PAIRED_CORRECTNESS, 12, 12)] + + +def test_historical_impossible_denominator_is_configuration_refusal() -> None: + threshold = _threshold( + MetricId.VULNERABLE_RECALL, + Statistic.WILSON_LOWER, + Comparator.AT_LEAST, + 0.5, + ) + + with pytest.raises(AdmissionFeasibilityError, match="vulnerable_recall.*arithmetically impossible"): + _evaluate( + _census(pairs=2, vulnerable=2, fixed=2, ordinary=4), + _policy(), + (threshold,), + ) + + +@pytest.mark.parametrize( + ("census", "threshold"), + ( + ( + _census(pairs=0, vulnerable=0, fixed=0, ordinary=4), + _threshold(MetricId.VULNERABLE_RECALL, Statistic.COUNT, Comparator.AT_LEAST, 0.0), + ), + ( + _census(calls=0), + _threshold(MetricId.MODEL_SUCCESS_RATE, Statistic.RATE, Comparator.AT_LEAST, 0.0), + ), + ( + _census(pairs=0, vulnerable=0), + _threshold(MetricId.PRECISION, Statistic.RATE, Comparator.AT_LEAST, 1.0), + ), + ), +) +def test_zero_or_absent_success_capacity_never_becomes_a_clean_zero( + census: PopulationCensus, + threshold: AdmissionThreshold, +) -> None: + with pytest.raises(AdmissionFeasibilityError, match="arithmetically impossible"): + _evaluate(census, _policy(), (threshold,)) + + +def test_precision_enumerates_a_false_positive_denominator_when_no_true_case_exists() -> None: + threshold = _threshold(MetricId.PRECISION, Statistic.RATE, Comparator.AT_LEAST, 0.0) + + result = _evaluate(_census(pairs=0, vulnerable=0), _policy(), (threshold,)) + + assert [(item.numerator, item.denominator) for item in result.witnesses] == [(0, 1)] + + +def test_fixed_metric_algebra_rejects_wrong_comparator_or_statistic() -> None: + reversed_comparator = _threshold( + MetricId.VULNERABLE_RECALL, + Statistic.RATE, + Comparator.AT_MOST, + 1.0, + ) + wrong_statistic = _threshold( + MetricId.ORDINARY_FALSE_ALARMS_PER_100, + Statistic.RATE, + Comparator.AT_MOST, + 5.0, + ) + + with pytest.raises(AdmissionFeasibilityError, match="comparator direction"): + _evaluate(_census(), _policy(), (reversed_comparator,)) + with pytest.raises(AdmissionFeasibilityError, match="per_100"): + _evaluate(_census(), _policy(), (wrong_statistic,)) + + +def test_policy_scope_must_match_the_label_blind_census() -> None: + threshold = _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 1.0) + + with pytest.raises(AdmissionFeasibilityError, match="scope"): + _evaluate(_census(), _policy(ScopeKey("other", "python")), (threshold,))