mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 17:58:18 +00:00
feat(detection-core): derive measured budgets and enforce scope
This commit is contained in:
Marius Ciepluch
@@ -1,21 +1,56 @@
|
||||
"""Measured budget contract for the detection core (Req 4).
|
||||
|
||||
The per-change time budget is derived from measurement, never a static constant:
|
||||
every analyzed change records its complexity and reply time, and each workload
|
||||
class gets a budget computed as a percentile of its measured samples with the
|
||||
sample count, percentile, and reference hardware disclosed (Req 12 schema).
|
||||
"""
|
||||
"""Measured workload budgets for legacy and v2 detection-core contracts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import overload
|
||||
|
||||
from openultrasast.core import canonical
|
||||
|
||||
CHANGED_BYTES_MEASUREMENT_SCHEMA = "openultrasast.detection-core.changed-bytes.v2"
|
||||
_DIGEST = re.compile(r"[0-9a-f]{64}")
|
||||
|
||||
|
||||
class BudgetGateFailure(ValueError):
|
||||
"""A change exceeded its class's measured budget."""
|
||||
|
||||
|
||||
def _text(value: object, field: str) -> None:
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ValueError(f"{field} must be nonempty text")
|
||||
|
||||
|
||||
def _digest(value: object, field: str) -> None:
|
||||
if not isinstance(value, str) or _DIGEST.fullmatch(value) is None:
|
||||
raise ValueError(f"{field} must be a full lowercase SHA-256 digest")
|
||||
|
||||
|
||||
def _positive_int(value: object, field: str) -> None:
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
|
||||
raise ValueError(f"{field} must be a positive integer")
|
||||
|
||||
|
||||
def _nonnegative_int(value: object, field: str) -> None:
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
||||
raise ValueError(f"{field} must be a non-negative integer")
|
||||
|
||||
|
||||
def _finite_number(value: object, field: str, *, positive: bool = False) -> float:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
|
||||
raise ValueError(f"{field} must be finite")
|
||||
number = float(value)
|
||||
if positive and number <= 0:
|
||||
raise ValueError(f"{field} must be positive")
|
||||
return number
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Complexity:
|
||||
"""Measured change complexity recorded per change (Req 4.1)."""
|
||||
"""Legacy measured change complexity."""
|
||||
|
||||
regions: int
|
||||
calls: int
|
||||
@@ -24,7 +59,7 @@ class Complexity:
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Budget:
|
||||
"""A derived per-class budget with its published derivation."""
|
||||
"""Legacy derived per-class budget."""
|
||||
|
||||
workload_class: str
|
||||
seconds: float
|
||||
@@ -33,16 +68,220 @@ class Budget:
|
||||
hardware: str
|
||||
|
||||
def describe(self) -> str:
|
||||
"""The derivation string gate failures must name (Req 4.3)."""
|
||||
return f"class={self.workload_class} n={self.n} p{int(self.percentile * 100)} hardware={self.hardware} budget={self.seconds:g}s"
|
||||
|
||||
|
||||
class BudgetGateFailure(ValueError):
|
||||
"""A change exceeded its class's measured budget."""
|
||||
@dataclass(frozen=True)
|
||||
class ChangeComplexity:
|
||||
change_id: str
|
||||
workload_class: str
|
||||
changed_regions: int
|
||||
transition_count: int
|
||||
changed_bytes: int
|
||||
measurement: canonical.ContentRef
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_validate_complexity(self)
|
||||
|
||||
|
||||
def _validate_complexity(value: ChangeComplexity) -> None:
|
||||
_digest(value.change_id, "change_id")
|
||||
_text(value.workload_class, "workload_class")
|
||||
_positive_int(value.changed_regions, "changed_regions")
|
||||
_nonnegative_int(value.transition_count, "transition_count")
|
||||
_positive_int(value.changed_bytes, "changed_bytes")
|
||||
if not isinstance(value.measurement, canonical.ContentRef) or value.measurement.byte_length <= 0:
|
||||
raise ValueError("measurement must be a nonempty ContentRef")
|
||||
value.measurement.__post_init__()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MeasuredSample:
|
||||
complexity: ChangeComplexity
|
||||
total_reply_ms: int
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not isinstance(self.complexity, ChangeComplexity):
|
||||
raise TypeError("complexity must be ChangeComplexity")
|
||||
_positive_int(self.total_reply_ms, "total_reply_ms")
|
||||
|
||||
|
||||
def _range(value: object, field: str, *, allow_zero: bool) -> tuple[int, int]:
|
||||
if not isinstance(value, tuple) or len(value) != 2:
|
||||
raise ValueError(f"{field} must be a two-item tuple")
|
||||
lower, upper = value
|
||||
validator = _nonnegative_int if allow_zero else _positive_int
|
||||
validator(lower, f"{field} lower")
|
||||
validator(upper, f"{field} upper")
|
||||
if lower > upper:
|
||||
raise ValueError(f"{field} lower must not exceed upper")
|
||||
return lower, upper
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReferenceWorkload:
|
||||
name: str
|
||||
description: str
|
||||
changed_region_range: tuple[int, int]
|
||||
transition_count_range: tuple[int, int]
|
||||
changed_bytes_range: tuple[int, int]
|
||||
sample_artifact: canonical.ContentRef
|
||||
record_digest: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_validate_reference_workload(self, allow_empty_digest=True)
|
||||
expected = derive_reference_workload_digest(self)
|
||||
if not self.record_digest:
|
||||
object.__setattr__(self, "record_digest", expected)
|
||||
elif self.record_digest != expected:
|
||||
raise ValueError("reference workload digest does not match fields")
|
||||
|
||||
|
||||
def _workload_value(value: ReferenceWorkload) -> canonical.JsonValue:
|
||||
return {
|
||||
"schema_version": "openultrasast.detection-core.reference-workload.v2",
|
||||
"name": value.name,
|
||||
"description": value.description,
|
||||
"changed_region_range": value.changed_region_range,
|
||||
"transition_count_range": value.transition_count_range,
|
||||
"changed_bytes_range": value.changed_bytes_range,
|
||||
"sample_artifact": _content_ref_value(value.sample_artifact),
|
||||
}
|
||||
|
||||
|
||||
def _content_ref_value(value: canonical.ContentRef) -> canonical.JsonValue:
|
||||
return {
|
||||
"artifact_id": value.artifact_id,
|
||||
"content_digest": value.content_digest,
|
||||
"media_type": value.media_type,
|
||||
"byte_length": value.byte_length,
|
||||
"retention_policy_digest": value.retention_policy_digest,
|
||||
"storage_receipt_digest": value.storage_receipt_digest,
|
||||
}
|
||||
|
||||
|
||||
def derive_reference_workload_digest(value: ReferenceWorkload) -> str:
|
||||
return canonical.content_digest(_workload_value(value))
|
||||
|
||||
|
||||
def _validate_reference_workload(value: ReferenceWorkload, *, allow_empty_digest: bool = False) -> None:
|
||||
_text(value.name, "reference workload name")
|
||||
_text(value.description, "reference workload description")
|
||||
_range(value.changed_region_range, "changed_region_range", allow_zero=False)
|
||||
_range(value.transition_count_range, "transition_count_range", allow_zero=True)
|
||||
_range(value.changed_bytes_range, "changed_bytes_range", allow_zero=False)
|
||||
if not isinstance(value.sample_artifact, canonical.ContentRef) or value.sample_artifact.byte_length <= 0:
|
||||
raise ValueError("sample_artifact must be a nonempty ContentRef")
|
||||
value.sample_artifact.__post_init__()
|
||||
if value.record_digest or not allow_empty_digest:
|
||||
_digest(value.record_digest, "reference workload digest")
|
||||
if value.record_digest and value.record_digest != derive_reference_workload_digest(value):
|
||||
raise ValueError("reference workload digest does not match fields")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReferenceHardware:
|
||||
cpu_model: str
|
||||
logical_cpu_count: int
|
||||
memory_bytes: int
|
||||
operating_system: str
|
||||
runtime_version: str
|
||||
record_digest: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_validate_reference_hardware(self, allow_empty_digest=True)
|
||||
expected = derive_reference_hardware_digest(self)
|
||||
if not self.record_digest:
|
||||
object.__setattr__(self, "record_digest", expected)
|
||||
elif self.record_digest != expected:
|
||||
raise ValueError("reference hardware digest does not match fields")
|
||||
|
||||
|
||||
def _hardware_value(value: ReferenceHardware) -> canonical.JsonValue:
|
||||
return {
|
||||
"schema_version": "openultrasast.detection-core.reference-hardware.v2",
|
||||
"cpu_model": value.cpu_model,
|
||||
"logical_cpu_count": value.logical_cpu_count,
|
||||
"memory_bytes": value.memory_bytes,
|
||||
"operating_system": value.operating_system,
|
||||
"runtime_version": value.runtime_version,
|
||||
}
|
||||
|
||||
|
||||
def derive_reference_hardware_digest(value: ReferenceHardware) -> str:
|
||||
return canonical.content_digest(_hardware_value(value))
|
||||
|
||||
|
||||
def _validate_reference_hardware(value: ReferenceHardware, *, allow_empty_digest: bool = False) -> None:
|
||||
for field, item in (
|
||||
("cpu_model", value.cpu_model),
|
||||
("operating_system", value.operating_system),
|
||||
("runtime_version", value.runtime_version),
|
||||
):
|
||||
_text(item, field)
|
||||
_positive_int(value.logical_cpu_count, "logical_cpu_count")
|
||||
_positive_int(value.memory_bytes, "memory_bytes")
|
||||
if value.record_digest or not allow_empty_digest:
|
||||
_digest(value.record_digest, "reference hardware digest")
|
||||
if value.record_digest and value.record_digest != derive_reference_hardware_digest(value):
|
||||
raise ValueError("reference hardware digest does not match fields")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BudgetDerivation:
|
||||
workload_class: str
|
||||
sample_count: int
|
||||
percentile: float
|
||||
reference_workload: ReferenceWorkload
|
||||
reference_hardware: ReferenceHardware
|
||||
budget_ms: int
|
||||
derivation_digest: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_validate_derivation(self, allow_empty_digest=True)
|
||||
expected = derive_budget_derivation_digest(self)
|
||||
if not self.derivation_digest:
|
||||
object.__setattr__(self, "derivation_digest", expected)
|
||||
elif self.derivation_digest != expected:
|
||||
raise ValueError("budget derivation digest does not match fields")
|
||||
|
||||
|
||||
def _derivation_value(value: BudgetDerivation) -> canonical.JsonValue:
|
||||
return {
|
||||
"schema_version": "openultrasast.detection-core.budget-derivation.v2",
|
||||
"workload_class": value.workload_class,
|
||||
"sample_count": value.sample_count,
|
||||
"percentile": value.percentile,
|
||||
"reference_workload_digest": value.reference_workload.record_digest,
|
||||
"reference_hardware_digest": value.reference_hardware.record_digest,
|
||||
"budget_ms": value.budget_ms,
|
||||
}
|
||||
|
||||
|
||||
def derive_budget_derivation_digest(value: BudgetDerivation) -> str:
|
||||
return canonical.content_digest(_derivation_value(value))
|
||||
|
||||
|
||||
def _validate_derivation(value: BudgetDerivation, *, allow_empty_digest: bool = False) -> None:
|
||||
_text(value.workload_class, "workload_class")
|
||||
_positive_int(value.sample_count, "sample_count")
|
||||
percentile = _finite_number(value.percentile, "percentile", positive=True)
|
||||
if percentile > 1:
|
||||
raise ValueError("percentile must not exceed 1")
|
||||
if not isinstance(value.reference_workload, ReferenceWorkload):
|
||||
raise TypeError("reference_workload must be ReferenceWorkload")
|
||||
if not isinstance(value.reference_hardware, ReferenceHardware):
|
||||
raise TypeError("reference_hardware must be ReferenceHardware")
|
||||
_validate_reference_workload(value.reference_workload)
|
||||
_validate_reference_hardware(value.reference_hardware)
|
||||
_positive_int(value.budget_ms, "budget_ms")
|
||||
if value.derivation_digest or not allow_empty_digest:
|
||||
_digest(value.derivation_digest, "derivation_digest")
|
||||
if value.derivation_digest and value.derivation_digest != derive_budget_derivation_digest(value):
|
||||
raise ValueError("budget derivation digest does not match fields")
|
||||
|
||||
|
||||
def _quantile(sorted_values: Sequence[float], percentile: float) -> float:
|
||||
"""Linear-interpolated quantile (numpy 'linear' convention)."""
|
||||
position = percentile * (len(sorted_values) - 1)
|
||||
lower = math.floor(position)
|
||||
upper = math.ceil(position)
|
||||
@@ -52,32 +291,193 @@ def _quantile(sorted_values: Sequence[float], percentile: float) -> float:
|
||||
return sorted_values[lower] * (1 - weight) + sorted_values[upper] * weight
|
||||
|
||||
|
||||
def _compatible(value: ChangeComplexity, reference: ReferenceWorkload) -> bool:
|
||||
return all(
|
||||
lower <= current <= upper
|
||||
for current, (lower, upper) in (
|
||||
(value.changed_regions, reference.changed_region_range),
|
||||
(value.transition_count, reference.transition_count_range),
|
||||
(value.changed_bytes, reference.changed_bytes_range),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@overload
|
||||
def derive_budget(samples: Sequence[tuple[Complexity, float]], *, workload_class: str, percentile: float, hardware: str) -> Budget: ...
|
||||
|
||||
|
||||
@overload
|
||||
def derive_budget(
|
||||
samples: Sequence[tuple[Complexity, float]],
|
||||
*,
|
||||
samples: Sequence[MeasuredSample],
|
||||
workload_class: str,
|
||||
percentile: float,
|
||||
reference_workload: ReferenceWorkload,
|
||||
reference_hardware: ReferenceHardware,
|
||||
) -> BudgetDerivation: ...
|
||||
|
||||
|
||||
def derive_budget(
|
||||
samples: Sequence[tuple[Complexity, float]] | Sequence[MeasuredSample],
|
||||
workload_class: str | None = None,
|
||||
percentile: float | None = None,
|
||||
*v2_context: object,
|
||||
hardware: str | None = None,
|
||||
reference_workload: ReferenceWorkload | None = None,
|
||||
reference_hardware: ReferenceHardware | None = None,
|
||||
) -> Budget | BudgetDerivation:
|
||||
if v2_context:
|
||||
if len(v2_context) != 2 or reference_workload is not None or reference_hardware is not None:
|
||||
raise TypeError("v2 positional context must be reference workload and hardware")
|
||||
positional_workload, positional_hardware = v2_context
|
||||
if not isinstance(positional_workload, ReferenceWorkload) or not isinstance(positional_hardware, ReferenceHardware):
|
||||
raise TypeError("v2 positional context must be reference workload and hardware")
|
||||
reference_workload, reference_hardware = positional_workload, positional_hardware
|
||||
if workload_class is None or percentile is None:
|
||||
raise TypeError("workload_class and percentile are required")
|
||||
if reference_workload is not None or reference_hardware is not None:
|
||||
if (
|
||||
not isinstance(reference_workload, ReferenceWorkload)
|
||||
or not isinstance(reference_hardware, ReferenceHardware)
|
||||
or hardware is not None
|
||||
):
|
||||
raise ValueError("v2 derivation requires reference workload and hardware only")
|
||||
return _derive_v2(samples, workload_class, percentile, reference_workload, reference_hardware)
|
||||
if hardware is None:
|
||||
raise ValueError("legacy derivation requires hardware")
|
||||
return _derive_legacy(samples, workload_class, percentile, hardware)
|
||||
|
||||
|
||||
def _derive_legacy(
|
||||
samples: Sequence[tuple[Complexity, float]] | Sequence[MeasuredSample],
|
||||
workload_class: str,
|
||||
percentile: float,
|
||||
hardware: str,
|
||||
) -> Budget:
|
||||
"""Derive a class budget as a percentile of measured (complexity, seconds) samples."""
|
||||
if not samples:
|
||||
raise ValueError("budget derivation needs at least one measured sample")
|
||||
if not 0.0 < percentile < 1.0:
|
||||
raise ValueError(f"percentile {percentile} outside (0, 1)")
|
||||
seconds = sorted(float(value) for _, value in samples)
|
||||
return Budget(
|
||||
workload_class=workload_class,
|
||||
seconds=_quantile(seconds, percentile),
|
||||
n=len(samples),
|
||||
percentile=percentile,
|
||||
hardware=hardware,
|
||||
seconds = sorted(float(item[1]) for item in samples if isinstance(item, tuple))
|
||||
if len(seconds) != len(samples):
|
||||
raise TypeError("legacy samples must be (Complexity, seconds) tuples")
|
||||
return Budget(workload_class, _quantile(seconds, percentile), len(samples), percentile, hardware)
|
||||
|
||||
|
||||
def _derive_v2(
|
||||
samples: Sequence[tuple[Complexity, float]] | Sequence[MeasuredSample],
|
||||
workload_class: str,
|
||||
percentile: float,
|
||||
reference_workload: ReferenceWorkload,
|
||||
reference_hardware: ReferenceHardware,
|
||||
) -> BudgetDerivation:
|
||||
_text(workload_class, "workload_class")
|
||||
value = _finite_number(percentile, "percentile", positive=True)
|
||||
if value > 1 or not samples:
|
||||
raise ValueError("v2 derivation requires samples and percentile in (0, 1]")
|
||||
_validate_reference_workload(reference_workload)
|
||||
_validate_reference_hardware(reference_hardware)
|
||||
measured = tuple(item for item in samples if isinstance(item, MeasuredSample))
|
||||
if len(measured) != len(samples):
|
||||
raise TypeError("v2 samples must be MeasuredSample values")
|
||||
for sample in measured:
|
||||
sample.__post_init__()
|
||||
_validate_complexity(sample.complexity)
|
||||
if sample.complexity.workload_class != workload_class:
|
||||
raise ValueError("sample workload class does not match derivation")
|
||||
if not _compatible(sample.complexity, reference_workload):
|
||||
raise ValueError("sample complexity is outside reference workload")
|
||||
budget_ms = math.ceil(_quantile(sorted(float(item.total_reply_ms) for item in measured), value))
|
||||
return BudgetDerivation(workload_class, len(measured), value, reference_workload, reference_hardware, budget_ms, "")
|
||||
|
||||
|
||||
class _ResolvedOnce:
|
||||
def __init__(self, resolved: canonical.ResolvedContent) -> None:
|
||||
self._resolved = resolved
|
||||
|
||||
def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent:
|
||||
return self._resolved
|
||||
|
||||
|
||||
def _validate_measurement(change: ChangeComplexity, resolver: canonical.ContentResolver) -> None:
|
||||
resolved = resolver.resolve(change.measurement)
|
||||
content = canonical.resolve_content(
|
||||
change.measurement,
|
||||
_ResolvedOnce(resolved),
|
||||
trusted_store_identities=(resolved.receipt.store_identity,),
|
||||
maximum_bytes=4096,
|
||||
)
|
||||
try:
|
||||
parsed = json.loads(content)
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise ValueError("changed-byte measurement is not canonical JSON") from exc
|
||||
expected = {"schema_version": CHANGED_BYTES_MEASUREMENT_SCHEMA, "changed_bytes": change.changed_bytes}
|
||||
if parsed != expected or canonical.canonical_bytes(parsed) != content:
|
||||
raise ValueError("changed-byte measurement does not match complexity")
|
||||
|
||||
|
||||
def check_budget(change_id: str, seconds: float, budget: Budget) -> None:
|
||||
"""Fail the gate naming the change, the elapsed time, and the derivation (Req 4.3)."""
|
||||
if seconds > budget.seconds:
|
||||
raise BudgetGateFailure(f"budget:{change_id}:{seconds:g}s>{budget.seconds:g}s ({budget.describe()})")
|
||||
@overload
|
||||
def check_budget(change_id: str, seconds: float, budget: Budget) -> None: ...
|
||||
|
||||
|
||||
__all__ = ["Budget", "BudgetGateFailure", "Complexity", "check_budget", "derive_budget"]
|
||||
@overload
|
||||
def check_budget(
|
||||
change: ChangeComplexity,
|
||||
total_reply_ms: int,
|
||||
budget: BudgetDerivation,
|
||||
resolver: canonical.ContentResolver,
|
||||
/,
|
||||
) -> None: ...
|
||||
|
||||
|
||||
def check_budget(
|
||||
change_id: str | ChangeComplexity | None = None,
|
||||
seconds: float | int | None = None,
|
||||
budget: Budget | BudgetDerivation | None = None,
|
||||
resolver: canonical.ContentResolver | None = None,
|
||||
**v2_names: object,
|
||||
) -> None:
|
||||
if v2_names:
|
||||
if set(v2_names) != {"change", "total_reply_ms"} or change_id is not None or seconds is not None:
|
||||
raise TypeError("use either legacy or v2 budget parameter names")
|
||||
candidate_change = v2_names["change"]
|
||||
candidate_elapsed = v2_names["total_reply_ms"]
|
||||
if not isinstance(candidate_change, ChangeComplexity) or not isinstance(candidate_elapsed, int):
|
||||
raise TypeError("v2 budget parameter values have invalid types")
|
||||
change_id, seconds = candidate_change, candidate_elapsed
|
||||
if change_id is None or seconds is None or budget is None:
|
||||
raise TypeError("change, elapsed time, and budget are required")
|
||||
if isinstance(change_id, str) and isinstance(budget, Budget):
|
||||
if seconds > budget.seconds:
|
||||
raise BudgetGateFailure(f"budget:{change_id}:{seconds:g}s>{budget.seconds:g}s ({budget.describe()})")
|
||||
return
|
||||
if not isinstance(change_id, ChangeComplexity) or not isinstance(budget, BudgetDerivation) or resolver is None:
|
||||
raise TypeError("v2 budget check requires complexity, derivation, and resolver")
|
||||
_validate_complexity(change_id)
|
||||
_validate_derivation(budget)
|
||||
_positive_int(seconds, "total_reply_ms")
|
||||
if change_id.workload_class != budget.workload_class or not _compatible(change_id, budget.reference_workload):
|
||||
raise ValueError("change complexity is incompatible with budget derivation")
|
||||
_validate_measurement(change_id, resolver)
|
||||
if seconds > budget.budget_ms:
|
||||
raise BudgetGateFailure(
|
||||
f"budget:{change_id.change_id}:{seconds}ms>{budget.budget_ms}ms "
|
||||
f"class={budget.workload_class} derivation={budget.derivation_digest}"
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"Budget",
|
||||
"BudgetDerivation",
|
||||
"BudgetGateFailure",
|
||||
"CHANGED_BYTES_MEASUREMENT_SCHEMA",
|
||||
"ChangeComplexity",
|
||||
"Complexity",
|
||||
"MeasuredSample",
|
||||
"ReferenceHardware",
|
||||
"ReferenceWorkload",
|
||||
"check_budget",
|
||||
"derive_budget",
|
||||
"derive_budget_derivation_digest",
|
||||
"derive_reference_hardware_digest",
|
||||
"derive_reference_workload_digest",
|
||||
]
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
@@ -10,6 +11,7 @@ from openultrasast.core import canonical
|
||||
from openultrasast.core.provenance import DecisionProvenance, ModelCallIdentity, ProvenanceStamp
|
||||
from openultrasast.core.semantics import (
|
||||
EvidenceProvenance,
|
||||
FieldKind,
|
||||
FieldValue,
|
||||
Observation,
|
||||
RelationStatus,
|
||||
@@ -27,6 +29,10 @@ class EvidenceValidationError(ValueError):
|
||||
"""An evidence packet or decision violates the neutral core contract."""
|
||||
|
||||
|
||||
class ScopeValidationError(EvidenceValidationError):
|
||||
"""Evidence or one of its semantic anchors escapes the affected change."""
|
||||
|
||||
|
||||
class NeutralOutcome(Enum):
|
||||
VIOLATION = "violation"
|
||||
SATISFIED = "satisfied"
|
||||
@@ -340,7 +346,76 @@ def _scope_within(bounded: AnalyzedScope, analyzed: AnalyzedScope) -> bool:
|
||||
analyzed.change_digest,
|
||||
):
|
||||
return False
|
||||
return all(region in analyzed.regions for region in bounded.regions)
|
||||
return all(any(_region_contains(allowed, region) for allowed in analyzed.regions) for region in bounded.regions)
|
||||
|
||||
|
||||
def _region_contains(allowed: SourceRegion, candidate: SourceRegion) -> bool:
|
||||
return candidate.path == allowed.path and candidate.start_line >= allowed.start_line and candidate.end_line <= allowed.end_line
|
||||
|
||||
|
||||
def _scope_location(scope: AnalyzedScope, location: SourceLocation) -> bool:
|
||||
return any(location.path == region.path and region.start_line <= location.line <= region.end_line for region in scope.regions)
|
||||
|
||||
|
||||
def _field_anchor(field: FieldValue) -> SourceLocation | None:
|
||||
field.__post_init__()
|
||||
if field.kind is not FieldKind.SOURCE_LOCATION:
|
||||
return None
|
||||
value = field.value
|
||||
if not isinstance(value, Mapping) or set(value) != {"path", "line", "column", "symbol"}:
|
||||
raise ScopeValidationError("source-location field must be a canonical mapping")
|
||||
path, line, column, symbol = value["path"], value["line"], value["column"], value["symbol"]
|
||||
if not isinstance(path, str) or not isinstance(line, int) or isinstance(line, bool):
|
||||
raise ScopeValidationError("source-location field path or line has an invalid type")
|
||||
if (column is not None and (not isinstance(column, int) or isinstance(column, bool))) or (
|
||||
symbol is not None and not isinstance(symbol, str)
|
||||
):
|
||||
raise ScopeValidationError("source-location field column or symbol has an invalid type")
|
||||
return SourceLocation(path, line, column, symbol)
|
||||
|
||||
|
||||
def _revalidate_scope(scope: AnalyzedScope, field: str) -> None:
|
||||
if not isinstance(scope, AnalyzedScope):
|
||||
raise ScopeValidationError(f"{field} must be an AnalyzedScope")
|
||||
try:
|
||||
for region in scope.regions:
|
||||
if not isinstance(region, SourceRegion):
|
||||
raise EvidenceValidationError("every region must be a SourceRegion")
|
||||
region.__post_init__()
|
||||
scope.__post_init__()
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ScopeValidationError(f"invalid {field}: {exc}") from exc
|
||||
|
||||
|
||||
def validate_scope(packet: EvidencePacket, allowed: AnalyzedScope) -> None:
|
||||
"""Require packet regions and every semantic anchor to stay in the affected change."""
|
||||
if not isinstance(packet, EvidencePacket):
|
||||
raise ScopeValidationError("packet must be an EvidencePacket")
|
||||
_revalidate_scope(allowed, "allowed scope")
|
||||
_revalidate_scope(packet.scope, "packet scope")
|
||||
try:
|
||||
_validate_packet(packet)
|
||||
for relation in packet.transitions:
|
||||
relation.__post_init__()
|
||||
for observation in packet.observations:
|
||||
observation.__post_init__()
|
||||
fields = tuple(field for observation in packet.observations for field in observation.fields) + packet.family_fields
|
||||
field_anchors = tuple(anchor for field in fields if (anchor := _field_anchor(field)) is not None)
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ScopeValidationError(f"invalid packet: {exc}") from exc
|
||||
if not _scope_within(packet.scope, allowed):
|
||||
raise ScopeValidationError("packet scope is outside affected change scope")
|
||||
anchors = (
|
||||
tuple(
|
||||
location
|
||||
for relation in packet.transitions
|
||||
for location in (relation.source.location, relation.target.location, relation.location)
|
||||
)
|
||||
+ tuple(observation.location for observation in packet.observations)
|
||||
+ field_anchors
|
||||
)
|
||||
if any(not _scope_location(packet.scope, location) for location in anchors):
|
||||
raise ScopeValidationError("semantic anchor is outside packet scope")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -407,8 +482,10 @@ __all__ = [
|
||||
"EvidenceRecord",
|
||||
"EvidenceValidationError",
|
||||
"NeutralOutcome",
|
||||
"ScopeValidationError",
|
||||
"SourceRegion",
|
||||
"derive_packet_digest",
|
||||
"derive_record_id",
|
||||
"evidence_record_bytes",
|
||||
"validate_scope",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user