mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 01:48:17 +00:00
feat(detection-core): add semantic shortcut probes
This commit is contained in:
Marius Ciepluch
@@ -18,12 +18,14 @@ from openultrasast.core.coverage import Coverage, InstrumentFailure, ResolutionO
|
||||
from openultrasast.core.factpack import FactPackError, admit_fact_pack, verify_fact_pack
|
||||
from openultrasast.core.probes import (
|
||||
FittedComponentBlocked,
|
||||
ProbeResult,
|
||||
ShortcutDetected,
|
||||
assert_no_shortcut,
|
||||
require_admission_evidence,
|
||||
reversed_pair_probe,
|
||||
)
|
||||
from openultrasast.core.probes import (
|
||||
LegacyProbeResult as ProbeResult,
|
||||
)
|
||||
from openultrasast.core.provenance import ProvenanceMismatch, stamp, verify_stamp
|
||||
from openultrasast.core.trace import Guard, Hop, TraceRecord, Witness
|
||||
from openultrasast.core.verdict import BoundedSilence, Judgment, ProvenClaim, as_clean
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
"""Validation and evaluation helpers for canonical reversed-surface probes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
from openultrasast.core import canonical
|
||||
from openultrasast.core.factpack import (
|
||||
ReviewIndependencePolicy,
|
||||
derive_independence_attestation_digest,
|
||||
derive_review_independence_policy_digest,
|
||||
)
|
||||
from openultrasast.core.family import FamilyContract, family_contract_digest
|
||||
from openultrasast.core.probes import (
|
||||
ProbeObservation,
|
||||
ProbeResult,
|
||||
ProbeValidationError,
|
||||
ReversedPairProbe,
|
||||
SemanticProjection,
|
||||
_probe_digest,
|
||||
derive_equivalence_subject_digest,
|
||||
project_semantics,
|
||||
)
|
||||
|
||||
|
||||
def _validate_policy(policy: ReviewIndependencePolicy) -> None:
|
||||
if not isinstance(policy, ReviewIndependencePolicy):
|
||||
raise ProbeValidationError("review policy has the wrong type")
|
||||
policy.__post_init__()
|
||||
expected = derive_review_independence_policy_digest(
|
||||
policy.reviewed_reviewer_digests,
|
||||
policy.forbidden_identity_pairs,
|
||||
policy.trusted_store_identities,
|
||||
policy.maximum_evidence_bytes,
|
||||
)
|
||||
if policy.policy_digest != expected:
|
||||
raise ProbeValidationError("review policy digest cannot be re-derived")
|
||||
|
||||
|
||||
def _resolve_inputs(
|
||||
probe: ReversedPairProbe,
|
||||
policy: ReviewIndependencePolicy,
|
||||
resolver: canonical.ContentResolver,
|
||||
) -> None:
|
||||
surfaces = tuple(
|
||||
canonical.resolve_content(
|
||||
reference,
|
||||
resolver,
|
||||
trusted_store_identities=policy.trusted_store_identities,
|
||||
maximum_bytes=policy.maximum_evidence_bytes,
|
||||
)
|
||||
for reference in (probe.forward_surface, probe.reversed_surface)
|
||||
)
|
||||
if not all(surfaces):
|
||||
raise ProbeValidationError("surface artifacts must be nonempty")
|
||||
if surfaces[0] == surfaces[1]:
|
||||
raise ProbeValidationError("probe requires distinct surface artifact bytes")
|
||||
disclosure = canonical.resolve_content(
|
||||
probe.equivalence.independence.conflict_disclosure,
|
||||
resolver,
|
||||
trusted_store_identities=policy.trusted_store_identities,
|
||||
maximum_bytes=policy.maximum_evidence_bytes,
|
||||
)
|
||||
if not disclosure:
|
||||
raise ProbeValidationError("equivalence conflict disclosure must be nonempty")
|
||||
|
||||
|
||||
def _validate_review(
|
||||
probe: ReversedPairProbe,
|
||||
forward_digest: str,
|
||||
reversed_digest: str,
|
||||
policy: ReviewIndependencePolicy,
|
||||
) -> None:
|
||||
review = probe.equivalence
|
||||
review.__post_init__()
|
||||
attestation = review.independence
|
||||
attestation.__post_init__()
|
||||
if (review.forward_projection_digest, review.reversed_projection_digest) != (forward_digest, reversed_digest):
|
||||
raise ProbeValidationError("review projection digests do not match derived semantics")
|
||||
expected_subject = derive_equivalence_subject_digest(
|
||||
forward_digest,
|
||||
reversed_digest,
|
||||
probe.family_contract_digest,
|
||||
probe.forward_surface,
|
||||
probe.reversed_surface,
|
||||
)
|
||||
if attestation.subject_digest != expected_subject:
|
||||
raise ProbeValidationError("equivalence attestation subject does not match canonical probe inputs")
|
||||
if attestation.attestation_digest != derive_independence_attestation_digest(attestation):
|
||||
raise ProbeValidationError("equivalence attestation digest cannot be re-derived")
|
||||
if attestation.reviewer_identity not in policy.reviewed_reviewer_digests:
|
||||
raise ProbeValidationError("equivalence reviewer is not trusted")
|
||||
if attestation.producer_identity == attestation.reviewer_identity:
|
||||
raise ProbeValidationError("equivalence review must be independent")
|
||||
if (attestation.producer_identity, attestation.reviewer_identity) in policy.forbidden_identity_pairs:
|
||||
raise ProbeValidationError("equivalence producer/reviewer pair is forbidden")
|
||||
|
||||
|
||||
def _validate_inputs(
|
||||
probe: ReversedPairProbe,
|
||||
observation: ProbeObservation,
|
||||
contract: FamilyContract,
|
||||
policy: ReviewIndependencePolicy,
|
||||
resolver: canonical.ContentResolver,
|
||||
) -> tuple[SemanticProjection, SemanticProjection]:
|
||||
if not isinstance(probe, ReversedPairProbe) or not isinstance(observation, ProbeObservation):
|
||||
raise ProbeValidationError("probe and observation must use v2 probe types")
|
||||
probe.__post_init__()
|
||||
observation.__post_init__()
|
||||
_validate_policy(policy)
|
||||
expected_contract = family_contract_digest(contract)
|
||||
if probe.family_contract_digest != expected_contract:
|
||||
raise ProbeValidationError("probe family contract does not match supplied contract")
|
||||
if probe.probe_id != _probe_digest(probe) or observation.probe_digest != probe.probe_id:
|
||||
raise ProbeValidationError("observation probe identity does not match canonical probe")
|
||||
decisions = {item.family_code for item in contract.decisions}
|
||||
if observation.forward_decision not in decisions or observation.reversed_decision not in decisions:
|
||||
raise ProbeValidationError("observation decision is not declared by the family")
|
||||
forward = project_semantics(probe.forward_record, contract)
|
||||
reversed_projection = project_semantics(probe.reversed_record, contract)
|
||||
_validate_review(probe, forward.projection_digest, reversed_projection.projection_digest, policy)
|
||||
_resolve_inputs(probe, policy, resolver)
|
||||
return forward, reversed_projection
|
||||
|
||||
|
||||
def _result_reasons(
|
||||
probe: ReversedPairProbe,
|
||||
observation: ProbeObservation,
|
||||
expected_decision: str,
|
||||
) -> tuple[str, ...]:
|
||||
reasons: list[str] = []
|
||||
if observation.forward_decision != observation.reversed_decision:
|
||||
reasons.append("decision_changed_under_reversed_surface")
|
||||
if expected_decision not in (observation.forward_decision, observation.reversed_decision) or (
|
||||
observation.forward_decision == observation.reversed_decision != expected_decision
|
||||
):
|
||||
reasons.append("component_decision_mismatch")
|
||||
if (
|
||||
observation.forward_score is not None
|
||||
and observation.reversed_score is not None
|
||||
and abs(Decimal(str(observation.forward_score)) - Decimal(str(observation.reversed_score)))
|
||||
> Decimal(str(probe.maximum_score_delta))
|
||||
):
|
||||
reasons.append("score_delta_exceeds_tolerance")
|
||||
return tuple(reasons)
|
||||
|
||||
|
||||
def evaluate_probe(
|
||||
probe: ReversedPairProbe,
|
||||
observation: ProbeObservation,
|
||||
contract: FamilyContract,
|
||||
review_policy: ReviewIndependencePolicy,
|
||||
resolver: canonical.ContentResolver,
|
||||
) -> ProbeResult:
|
||||
"""Re-derive semantic equivalence and evaluate one frozen component observation."""
|
||||
try:
|
||||
forward, reversed_projection = _validate_inputs(probe, observation, contract, review_policy, resolver)
|
||||
except ProbeValidationError:
|
||||
raise
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ProbeValidationError(f"probe input cannot be validated: {exc}") from exc
|
||||
semantic_equal = forward == reversed_projection
|
||||
if not semantic_equal:
|
||||
invalid_reasons = ["semantic_projection_mismatch"]
|
||||
if forward.family_decision != reversed_projection.family_decision:
|
||||
invalid_reasons.append("family_decision_mismatch")
|
||||
return ProbeResult(probe.probe_id, observation.component_digest, False, False, tuple(invalid_reasons), "")
|
||||
reasons = _result_reasons(probe, observation, forward.family_decision)
|
||||
return ProbeResult(probe.probe_id, observation.component_digest, True, not reasons, reasons, "")
|
||||
|
||||
|
||||
__all__ = ["evaluate_probe"]
|
||||
@@ -1,56 +1,408 @@
|
||||
"""Anti-shortcut probes for fitted components (Req 8, 9).
|
||||
|
||||
A fitted prompt, threshold, or model must not separate a fix from its reversed fix
|
||||
using shape features (diff size, deletion count, line counts): that is pool-specific
|
||||
shortcut learning, the 2026-10-14 deletion-count 0.90 finding. A component making any
|
||||
detection claim must also carry per-repository holdout evidence and a GO admission on
|
||||
an untouched population.
|
||||
"""
|
||||
"""Canonical reversed-surface probes for fitted-component shortcut resistance."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import re
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, replace
|
||||
from typing import Any
|
||||
|
||||
from openultrasast.core import canonical
|
||||
from openultrasast.core.evidence import EvidenceRecord
|
||||
from openultrasast.core.factpack import IndependenceAttestation, ReviewIndependencePolicy
|
||||
from openultrasast.core.family import FamilyContract, family_contract_digest, validate_record
|
||||
from openultrasast.core.semantics import FieldValue, RelationKind, RelationStatus
|
||||
|
||||
_DIGEST = re.compile(r"[0-9a-f]{64}")
|
||||
|
||||
|
||||
class ProbeValidationError(ValueError):
|
||||
"""A probe input or claimed result cannot be reproduced."""
|
||||
|
||||
|
||||
class ShortcutDetected(ValueError):
|
||||
"""A component separates a fix from its reversed fix by its decision features."""
|
||||
"""A legacy component separates a fix from its reversed surface."""
|
||||
|
||||
|
||||
class FittedComponentBlocked(ValueError):
|
||||
"""A fitted component claims detection without holdout + admission evidence."""
|
||||
"""A legacy fitted component lacks its required release evidence."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProbeResult:
|
||||
"""Outcome of one reversed-pair probe."""
|
||||
class LegacyProbeResult:
|
||||
"""Frozen v1 shortcut result retained until the migration cutover."""
|
||||
|
||||
separated: bool
|
||||
fix_score: float
|
||||
reversed_score: float
|
||||
|
||||
|
||||
def reversed_pair_probe(
|
||||
score_fn: Callable[[Mapping[str, Any]], float], fix_change: Mapping[str, Any], reversed_change: Mapping[str, Any]
|
||||
def _digest(value: object, field: str) -> None:
|
||||
if not isinstance(value, str) or _DIGEST.fullmatch(value) is None:
|
||||
raise ProbeValidationError(f"{field} must be a full lowercase SHA-256 digest")
|
||||
|
||||
|
||||
def _text(value: object, field: str) -> None:
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ProbeValidationError(f"{field} must be nonempty text")
|
||||
|
||||
|
||||
def _field_value(field: FieldValue) -> canonical.JsonValue:
|
||||
return {"name": field.name, "kind": field.kind.value, "value": field.value}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RoleProjection:
|
||||
role: str
|
||||
fields: tuple[FieldValue, ...]
|
||||
relation_kinds: tuple[RelationKind, ...]
|
||||
statuses: tuple[RelationStatus, ...]
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_text(self.role, "projection role")
|
||||
if not isinstance(self.fields, tuple) or not all(isinstance(item, FieldValue) for item in self.fields):
|
||||
raise ProbeValidationError("projection fields must be FieldValue values")
|
||||
if not isinstance(self.relation_kinds, tuple) or not all(isinstance(item, RelationKind) for item in self.relation_kinds):
|
||||
raise ProbeValidationError("projection relation kinds are invalid")
|
||||
if not isinstance(self.statuses, tuple) or not all(isinstance(item, RelationStatus) for item in self.statuses):
|
||||
raise ProbeValidationError("projection statuses are invalid")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SemanticProjection:
|
||||
family_contract_digest: str
|
||||
security_object_fields: tuple[FieldValue, ...]
|
||||
roles: tuple[RoleProjection, ...]
|
||||
completeness_reasons: tuple[str, ...]
|
||||
family_decision: str
|
||||
projection_digest: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_digest(self.family_contract_digest, "projection family contract")
|
||||
_text(self.family_decision, "projection family decision")
|
||||
if not isinstance(self.security_object_fields, tuple) or not all(
|
||||
isinstance(item, FieldValue) for item in self.security_object_fields
|
||||
):
|
||||
raise ProbeValidationError("projection security-object fields are invalid")
|
||||
if not isinstance(self.roles, tuple) or not all(isinstance(item, RoleProjection) for item in self.roles):
|
||||
raise ProbeValidationError("projection roles are invalid")
|
||||
if not isinstance(self.completeness_reasons, tuple) or not all(
|
||||
isinstance(item, str) and item for item in self.completeness_reasons
|
||||
):
|
||||
raise ProbeValidationError("projection completeness reasons are invalid")
|
||||
expected = _projection_digest(self)
|
||||
if not self.projection_digest:
|
||||
object.__setattr__(self, "projection_digest", expected)
|
||||
elif self.projection_digest != expected:
|
||||
raise ProbeValidationError("projection digest cannot be re-derived")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SemanticEquivalenceReview:
|
||||
forward_projection_digest: str
|
||||
reversed_projection_digest: str
|
||||
independence: IndependenceAttestation
|
||||
review_digest: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_digest(self.forward_projection_digest, "forward projection")
|
||||
_digest(self.reversed_projection_digest, "reversed projection")
|
||||
if not isinstance(self.independence, IndependenceAttestation):
|
||||
raise ProbeValidationError("equivalence review requires an independence attestation")
|
||||
expected = _review_digest(self)
|
||||
if not self.review_digest:
|
||||
object.__setattr__(self, "review_digest", expected)
|
||||
elif self.review_digest != expected:
|
||||
raise ProbeValidationError("equivalence review digest cannot be re-derived")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReversedPairProbe:
|
||||
probe_id: str
|
||||
family_contract_digest: str
|
||||
forward_record: EvidenceRecord
|
||||
reversed_record: EvidenceRecord
|
||||
forward_surface: canonical.ContentRef
|
||||
reversed_surface: canonical.ContentRef
|
||||
equivalence: SemanticEquivalenceReview
|
||||
maximum_score_delta: float
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_digest(self.family_contract_digest, "probe family contract")
|
||||
if not isinstance(self.forward_record, EvidenceRecord) or not isinstance(self.reversed_record, EvidenceRecord):
|
||||
raise ProbeValidationError("probe records must be EvidenceRecord values")
|
||||
if not isinstance(self.forward_surface, canonical.ContentRef) or not isinstance(self.reversed_surface, canonical.ContentRef):
|
||||
raise ProbeValidationError("probe surfaces must be ContentRef values")
|
||||
if not isinstance(self.equivalence, SemanticEquivalenceReview):
|
||||
raise ProbeValidationError("probe equivalence review is invalid")
|
||||
if isinstance(self.maximum_score_delta, bool) or not isinstance(self.maximum_score_delta, (int, float)):
|
||||
raise ProbeValidationError("maximum score delta must be a finite non-negative number")
|
||||
if not math.isfinite(self.maximum_score_delta) or self.maximum_score_delta < 0:
|
||||
raise ProbeValidationError("maximum score delta must be a finite non-negative number")
|
||||
expected = _probe_digest(self)
|
||||
if not self.probe_id:
|
||||
object.__setattr__(self, "probe_id", expected)
|
||||
elif self.probe_id != expected:
|
||||
raise ProbeValidationError("probe identity cannot be re-derived")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProbeObservation:
|
||||
probe_digest: str
|
||||
component_digest: str
|
||||
forward_decision: str
|
||||
reversed_decision: str
|
||||
forward_score: float | None
|
||||
reversed_score: float | None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_digest(self.probe_digest, "observation probe")
|
||||
_digest(self.component_digest, "observation component")
|
||||
_text(self.forward_decision, "forward decision")
|
||||
_text(self.reversed_decision, "reversed decision")
|
||||
scores = (self.forward_score, self.reversed_score)
|
||||
if (scores[0] is None) != (scores[1] is None):
|
||||
raise ProbeValidationError("probe scores must both be present or both be absent")
|
||||
if any(
|
||||
value is not None and (isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value))
|
||||
for value in scores
|
||||
):
|
||||
raise ProbeValidationError("probe scores must be finite numbers")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProbeEvaluationInput:
|
||||
probe: ReversedPairProbe
|
||||
observation: ProbeObservation
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not isinstance(self.probe, ReversedPairProbe) or not isinstance(self.observation, ProbeObservation):
|
||||
raise ProbeValidationError("probe evaluation input has invalid members")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProbeResult:
|
||||
probe_digest: str
|
||||
component_digest: str
|
||||
valid: bool
|
||||
passed: bool
|
||||
reasons: tuple[str, ...]
|
||||
result_digest: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_digest(self.probe_digest, "result probe")
|
||||
_digest(self.component_digest, "result component")
|
||||
if not isinstance(self.valid, bool) or not isinstance(self.passed, bool):
|
||||
raise ProbeValidationError("probe result flags must be booleans")
|
||||
if self.passed and not self.valid:
|
||||
raise ProbeValidationError("an invalid probe cannot pass")
|
||||
if not isinstance(self.reasons, tuple) or not all(isinstance(item, str) and item for item in self.reasons):
|
||||
raise ProbeValidationError("probe result reasons are invalid")
|
||||
if self.passed and self.reasons:
|
||||
raise ProbeValidationError("a passing probe cannot carry failure reasons")
|
||||
if not self.passed and not self.reasons:
|
||||
raise ProbeValidationError("a non-passing probe requires reasons")
|
||||
expected = _result_digest(self)
|
||||
if not self.result_digest:
|
||||
object.__setattr__(self, "result_digest", expected)
|
||||
elif self.result_digest != expected:
|
||||
raise ProbeValidationError("probe result digest cannot be re-derived")
|
||||
|
||||
|
||||
def _projection_value(projection: SemanticProjection) -> canonical.JsonValue:
|
||||
return {
|
||||
"family_contract_digest": projection.family_contract_digest,
|
||||
"security_object_fields": tuple(_field_value(item) for item in projection.security_object_fields),
|
||||
"roles": tuple(
|
||||
{
|
||||
"role": role.role,
|
||||
"fields": tuple(_field_value(item) for item in role.fields),
|
||||
"relation_kinds": tuple(item.value for item in role.relation_kinds),
|
||||
"statuses": tuple(item.value for item in role.statuses),
|
||||
}
|
||||
for role in projection.roles
|
||||
),
|
||||
"completeness_reasons": projection.completeness_reasons,
|
||||
"family_decision": projection.family_decision,
|
||||
}
|
||||
|
||||
|
||||
def _projection_digest(projection: SemanticProjection) -> str:
|
||||
return canonical.content_digest(_projection_value(projection))
|
||||
|
||||
|
||||
def _content_ref_value(reference: canonical.ContentRef) -> canonical.JsonValue:
|
||||
return {
|
||||
"artifact_id": reference.artifact_id,
|
||||
"content_digest": reference.content_digest,
|
||||
"media_type": reference.media_type,
|
||||
"byte_length": reference.byte_length,
|
||||
"retention_policy_digest": reference.retention_policy_digest,
|
||||
"storage_receipt_digest": reference.storage_receipt_digest,
|
||||
}
|
||||
|
||||
|
||||
def _attestation_value(attestation: IndependenceAttestation) -> canonical.JsonValue:
|
||||
return {
|
||||
"subject_digest": attestation.subject_digest,
|
||||
"producer_identity": attestation.producer_identity,
|
||||
"reviewer_identity": attestation.reviewer_identity,
|
||||
"conflict_disclosure": _content_ref_value(attestation.conflict_disclosure),
|
||||
"attestation_digest": attestation.attestation_digest,
|
||||
}
|
||||
|
||||
|
||||
def _review_digest(review: SemanticEquivalenceReview) -> str:
|
||||
return canonical.content_digest(
|
||||
{
|
||||
"forward_projection_digest": review.forward_projection_digest,
|
||||
"reversed_projection_digest": review.reversed_projection_digest,
|
||||
"independence": _attestation_value(review.independence),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _probe_digest(probe: ReversedPairProbe) -> str:
|
||||
return canonical.content_digest(
|
||||
{
|
||||
"family_contract_digest": probe.family_contract_digest,
|
||||
"forward_record_id": probe.forward_record.record_id,
|
||||
"reversed_record_id": probe.reversed_record.record_id,
|
||||
"forward_surface": _content_ref_value(probe.forward_surface),
|
||||
"reversed_surface": _content_ref_value(probe.reversed_surface),
|
||||
"equivalence_review_digest": probe.equivalence.review_digest,
|
||||
"maximum_score_delta": probe.maximum_score_delta,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _result_digest(result: ProbeResult) -> str:
|
||||
return canonical.content_digest(
|
||||
{
|
||||
"probe_digest": result.probe_digest,
|
||||
"component_digest": result.component_digest,
|
||||
"valid": result.valid,
|
||||
"passed": result.passed,
|
||||
"reasons": result.reasons,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _sort_fields(values: tuple[FieldValue, ...]) -> tuple[FieldValue, ...]:
|
||||
return tuple(sorted(values, key=lambda item: canonical.canonical_bytes(_field_value(item))))
|
||||
|
||||
|
||||
def project_semantics(record: EvidenceRecord, contract: FamilyContract) -> SemanticProjection:
|
||||
"""Derive only family-declared security semantics, excluding presentation shape."""
|
||||
for relation in record.packet.transitions:
|
||||
relation.__post_init__()
|
||||
for observation in record.packet.observations:
|
||||
observation.__post_init__()
|
||||
for field in observation.fields:
|
||||
field.__post_init__()
|
||||
for field in record.packet.family_fields:
|
||||
field.__post_init__()
|
||||
record.packet.__post_init__()
|
||||
record.decision.__post_init__()
|
||||
record.__post_init__()
|
||||
validate_record(contract, record)
|
||||
rule = next(item for item in contract.decisions if item.family_code == record.decision.family_code)
|
||||
actual_fields = {item.name: item for item in record.packet.family_fields}
|
||||
object_fields = tuple(actual_fields[item.name] for item in contract.security_object.fields if item.name in actual_fields)
|
||||
roles: list[RoleProjection] = []
|
||||
for role in rule.required_roles:
|
||||
observations = tuple(item for item in record.packet.observations if item.role == role)
|
||||
relations = tuple(item for item in record.packet.transitions if item.role == role)
|
||||
fields = _sort_fields(tuple(field for item in observations for field in item.fields))
|
||||
kinds = tuple(sorted((item.kind for item in relations), key=lambda item: item.value))
|
||||
relation_statuses = tuple(item.status for item in relations)
|
||||
observation_statuses = tuple(item.status for item in observations)
|
||||
statuses = tuple(sorted((*relation_statuses, *observation_statuses), key=lambda item: item.value))
|
||||
roles.append(RoleProjection(role, fields, kinds, statuses))
|
||||
provisional = SemanticProjection(
|
||||
family_contract_digest(contract),
|
||||
object_fields,
|
||||
tuple(roles),
|
||||
tuple(sorted(record.packet.completeness_reasons)),
|
||||
record.decision.family_code,
|
||||
"",
|
||||
)
|
||||
return replace(provisional, projection_digest=_projection_digest(provisional))
|
||||
|
||||
|
||||
def derive_equivalence_subject_digest(
|
||||
forward_projection_digest: str,
|
||||
reversed_projection_digest: str,
|
||||
contract_digest: str,
|
||||
forward_surface: canonical.ContentRef,
|
||||
reversed_surface: canonical.ContentRef,
|
||||
) -> str:
|
||||
for value, field in (
|
||||
(forward_projection_digest, "forward projection"),
|
||||
(reversed_projection_digest, "reversed projection"),
|
||||
(contract_digest, "family contract"),
|
||||
):
|
||||
_digest(value, field)
|
||||
return canonical.content_digest(
|
||||
{
|
||||
"forward_projection_digest": forward_projection_digest,
|
||||
"reversed_projection_digest": reversed_projection_digest,
|
||||
"family_contract_digest": contract_digest,
|
||||
"forward_surface": _content_ref_value(forward_surface),
|
||||
"reversed_surface": _content_ref_value(reversed_surface),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def evaluate_probe(
|
||||
probe: ReversedPairProbe,
|
||||
observation: ProbeObservation,
|
||||
contract: FamilyContract,
|
||||
review_policy: ReviewIndependencePolicy,
|
||||
resolver: canonical.ContentResolver,
|
||||
) -> ProbeResult:
|
||||
"""Score a fix and its reversed fix; separated=True means the features shortcut."""
|
||||
from openultrasast.core._probe_eval import evaluate_probe as evaluate
|
||||
|
||||
return evaluate(probe, observation, contract, review_policy, resolver)
|
||||
|
||||
|
||||
def validate_probe_result(
|
||||
evaluation: ProbeEvaluationInput,
|
||||
claimed: ProbeResult,
|
||||
contract: FamilyContract,
|
||||
review_policy: ReviewIndependencePolicy,
|
||||
resolver: canonical.ContentResolver,
|
||||
) -> None:
|
||||
actual = evaluate_probe(evaluation.probe, evaluation.observation, contract, review_policy, resolver)
|
||||
if claimed != actual:
|
||||
raise ProbeValidationError("claimed probe result does not match canonical evaluation")
|
||||
|
||||
|
||||
def reversed_pair_probe(
|
||||
score_fn: Callable[[Mapping[str, Any]], float],
|
||||
fix_change: Mapping[str, Any],
|
||||
reversed_change: Mapping[str, Any],
|
||||
) -> LegacyProbeResult:
|
||||
"""Run the frozen v1 shape probe without granting it v2 authority."""
|
||||
fix_score = float(score_fn(fix_change))
|
||||
reversed_score = float(score_fn(reversed_change))
|
||||
return ProbeResult(separated=fix_score != reversed_score, fix_score=fix_score, reversed_score=reversed_score)
|
||||
return LegacyProbeResult(fix_score != reversed_score, fix_score, reversed_score)
|
||||
|
||||
|
||||
def assert_no_shortcut(
|
||||
score_fn: Callable[[Mapping[str, Any]], float], fix_change: Mapping[str, Any], reversed_change: Mapping[str, Any]
|
||||
score_fn: Callable[[Mapping[str, Any]], float],
|
||||
fix_change: Mapping[str, Any],
|
||||
reversed_change: Mapping[str, Any],
|
||||
) -> None:
|
||||
"""Raise ShortcutDetected when the component separates fix from reversed fix."""
|
||||
"""Preserve the v1 exception behavior during migration."""
|
||||
result = reversed_pair_probe(score_fn, fix_change, reversed_change)
|
||||
if result.separated:
|
||||
raise ShortcutDetected(f"shape shortcut: fix={result.fix_score} vs reversed={result.reversed_score}")
|
||||
|
||||
|
||||
def require_admission_evidence(*, holdout_passed: bool, admission_decision: str) -> None:
|
||||
"""Block any fitted-component claim without holdout and an admission GO (Req 8.2)."""
|
||||
"""Preserve the v1 fitted-component release guard during migration."""
|
||||
if not holdout_passed:
|
||||
raise FittedComponentBlocked("fitted component lacks per-repository holdout evidence")
|
||||
if admission_decision != "GO":
|
||||
@@ -59,9 +411,21 @@ def require_admission_evidence(*, holdout_passed: bool, admission_decision: str)
|
||||
|
||||
__all__ = [
|
||||
"FittedComponentBlocked",
|
||||
"LegacyProbeResult",
|
||||
"ProbeEvaluationInput",
|
||||
"ProbeObservation",
|
||||
"ProbeResult",
|
||||
"ProbeValidationError",
|
||||
"ReversedPairProbe",
|
||||
"RoleProjection",
|
||||
"SemanticEquivalenceReview",
|
||||
"SemanticProjection",
|
||||
"ShortcutDetected",
|
||||
"assert_no_shortcut",
|
||||
"derive_equivalence_subject_digest",
|
||||
"evaluate_probe",
|
||||
"project_semantics",
|
||||
"require_admission_evidence",
|
||||
"reversed_pair_probe",
|
||||
"validate_probe_result",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user