From 3e4951fd3004f3fbccd178b4eef282b82bebdcf9 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 15:00:39 +0100 Subject: [PATCH] feat(detection-core): derive measured budgets and enforce scope --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/budget.py | 456 ++++++++++++++++++++++++++-- src/openultrasast/core/evidence.py | 79 ++++- tests/test_core_budget_v2.py | 348 +++++++++++++++++++++ 4 files changed, 855 insertions(+), 30 deletions(-) create mode 100644 tests/test_core_budget_v2.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 451aa27..5b9fb66 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -102,7 +102,7 @@ - _Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 15.2, 15.4_ - _Boundary: Coverage Funnel_ - _Depends: 2.1, 2.5_ -- [ ] 4.2 (P) Derive and enforce measured workload budgets and change scope +- [x] 4.2 (P) Derive and enforce measured workload budgets and change scope - Record changed regions, semantic-transition count, changed bytes, and total reply time. - Derive workload-class budgets with sample count, percentile, reference workload, reference hardware, and canonical identity. - Resolve the changed-byte measurement and refuse missing, zero, non-finite, or incompatible complexity data. diff --git a/src/openultrasast/core/budget.py b/src/openultrasast/core/budget.py index a3240a6..3f80099 100644 --- a/src/openultrasast/core/budget.py +++ b/src/openultrasast/core/budget.py @@ -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", +] diff --git a/src/openultrasast/core/evidence.py b/src/openultrasast/core/evidence.py index 9a8a159..1819692 100644 --- a/src/openultrasast/core/evidence.py +++ b/src/openultrasast/core/evidence.py @@ -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", ] diff --git a/tests/test_core_budget_v2.py b/tests/test_core_budget_v2.py new file mode 100644 index 0000000..cbb7f9e --- /dev/null +++ b/tests/test_core_budget_v2.py @@ -0,0 +1,348 @@ +"""Measured workload budgets and affected-scope enforcement.""" + +from __future__ import annotations + +import math +from dataclasses import replace +from pathlib import Path +from typing import cast + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import budget as budget_v2 +from openultrasast.core import canonical, evidence, provenance, semantics + + +def _digest(value: str) -> str: + return canonical.content_digest(value) + + +def _measurement(changed_bytes: int) -> tuple[canonical.ContentRef, InMemoryContentResolver]: + content = canonical.canonical_bytes({"schema_version": budget_v2.CHANGED_BYTES_MEASUREMENT_SCHEMA, "changed_bytes": changed_bytes}) + return stored_artifact(content, artifact_id=f"changed-bytes-{changed_bytes}") + + +def _complexity( + *, + workload_class: str = "small", + changed_regions: int = 2, + transition_count: int = 4, + changed_bytes: int = 120, +) -> tuple[budget_v2.ChangeComplexity, InMemoryContentResolver]: + reference, resolver = _measurement(changed_bytes) + return ( + budget_v2.ChangeComplexity( + change_id=_digest("change-1"), + workload_class=workload_class, + changed_regions=changed_regions, + transition_count=transition_count, + changed_bytes=changed_bytes, + measurement=reference, + ), + resolver, + ) + + +def _reference_workload() -> budget_v2.ReferenceWorkload: + artifact, _ = stored_artifact(b"samples", artifact_id="sample-artifact") + value = budget_v2.ReferenceWorkload( + name="small-reference", + description="small change workload", + changed_region_range=(1, 3), + transition_count_range=(0, 10), + changed_bytes_range=(1, 500), + sample_artifact=artifact, + record_digest="", + ) + return replace(value, record_digest=budget_v2.derive_reference_workload_digest(value)) + + +def _hardware() -> budget_v2.ReferenceHardware: + value = budget_v2.ReferenceHardware( + cpu_model="Example CPU", + logical_cpu_count=8, + memory_bytes=16_000_000_000, + operating_system="linux", + runtime_version="python-3.11", + record_digest="", + ) + return replace(value, record_digest=budget_v2.derive_reference_hardware_digest(value)) + + +def _samples() -> tuple[budget_v2.MeasuredSample, ...]: + values: list[budget_v2.MeasuredSample] = [] + for elapsed in (80, 100, 120, 140, 160): + complexity, _ = _complexity(changed_bytes=100 + elapsed) + values.append(budget_v2.MeasuredSample(complexity, elapsed)) + return tuple(values) + + +def _derivation() -> budget_v2.BudgetDerivation: + return budget_v2.derive_budget( + _samples(), + workload_class="small", + percentile=0.8, + reference_workload=_reference_workload(), + reference_hardware=_hardware(), + ) + + +def _scope(*regions: evidence.SourceRegion) -> evidence.AnalyzedScope: + selected = regions or (evidence.SourceRegion("src/app.py", 1, 30),) + return evidence.AnalyzedScope("base", "head", _digest("change"), selected) + + +def _provenance() -> semantics.EvidenceProvenance: + return semantics.EvidenceProvenance(semantics.ProvenanceKind.PRODUCT_CODE, _digest("source"), None, None, None) + + +def _packet( + *, + line: int = 10, + scope: evidence.AnalyzedScope | None = None, + observation_fields: tuple[semantics.FieldValue, ...] = (), + family_fields: tuple[semantics.FieldValue, ...] = (), +) -> evidence.EvidencePacket: + location = semantics.SourceLocation("src/app.py", line, 0, "handle") + relation = semantics.make_relation( + kind=semantics.RelationKind.CALL, + role="operation", + source=semantics.SemanticNode("source", "source", location), + target=semantics.SemanticNode("target", "target", location), + location=location, + status=semantics.RelationStatus.COMPLETE, + conditions=(), + reasons=(), + provenance=_provenance(), + ) + stamp = provenance.ProvenanceStamp(*(_digest(name) for name in ("a", "f", "s", "q", "c", "p", "i"))) + observations: tuple[semantics.Observation, ...] = () + if observation_fields: + observations = ( + semantics.make_observation( + role="operation", + location=location, + fields=observation_fields, + status=semantics.RelationStatus.COMPLETE, + provenance=_provenance(), + ), + ) + return evidence.EvidencePacket( + evidence.EVIDENCE_PACKET_SCHEMA, + "reference-family", + stamp.family_contract_digest, + "object-1", + "python", + _scope() if scope is None else scope, + (relation,), + observations, + family_fields, + (), + stamp, + ) + + +def test_budget_v2_test_instrument_is_nonzero() -> None: + assert __file__ + assert len(Path(__file__).read_bytes()) > 0 + + +def test_derivation_is_deterministic_and_discloses_measurement_context() -> None: + first = _derivation() + second = budget_v2.derive_budget( + tuple(reversed(_samples())), + workload_class="small", + percentile=0.8, + reference_workload=_reference_workload(), + reference_hardware=_hardware(), + ) + assert first == second + assert first.sample_count == 5 + assert first.percentile == 0.8 + assert first.budget_ms == 144 + assert len(first.derivation_digest) == 64 + + +def test_v2_design_signature_accepts_positional_derivation_context() -> None: + derived = budget_v2.derive_budget(_samples(), "small", 0.8, _reference_workload(), _hardware()) + assert isinstance(derived, budget_v2.BudgetDerivation) + assert derived.budget_ms == 144 + + +@pytest.mark.parametrize( + ("samples", "workload_class", "percentile"), + [ + ((), "small", 0.8), + (_samples(), "", 0.8), + (_samples(), "small", 0.0), + (_samples(), "small", 1.1), + (_samples(), "small", math.nan), + ], +) +def test_derivation_refuses_missing_or_nonfinite_inputs( + samples: tuple[budget_v2.MeasuredSample, ...], workload_class: str, percentile: float +) -> None: + with pytest.raises((TypeError, ValueError)): + budget_v2.derive_budget( + samples, + workload_class=workload_class, + percentile=percentile, + reference_workload=_reference_workload(), + reference_hardware=_hardware(), + ) + + +def test_derivation_refuses_mixed_classes_and_incompatible_complexity() -> None: + wrong_class, _ = _complexity(workload_class="large") + too_large, _ = _complexity(changed_bytes=501) + for candidate in (wrong_class, too_large): + with pytest.raises(ValueError): + budget_v2.derive_budget( + (budget_v2.MeasuredSample(candidate, 100),), + workload_class="small", + percentile=0.8, + reference_workload=_reference_workload(), + reference_hardware=_hardware(), + ) + + +def test_derivation_revalidates_tampered_sample_complexity() -> None: + complexity, _ = _complexity() + sample = budget_v2.MeasuredSample(complexity, 100) + object.__setattr__(complexity, "changed_bytes", math.nan) + with pytest.raises(ValueError): + budget_v2.derive_budget( + (sample,), + workload_class="small", + percentile=0.8, + reference_workload=_reference_workload(), + reference_hardware=_hardware(), + ) + + +def test_reference_workload_identity_covers_complete_sample_reference() -> None: + first = _reference_workload() + other_reference, _ = stored_artifact(b"samples", artifact_id="other-sample-artifact") + second = replace(first, sample_artifact=other_reference, record_digest="") + assert first.record_digest != second.record_digest + + +@pytest.mark.parametrize("elapsed", [0, -1, math.inf, math.nan]) +def test_measured_sample_refuses_nonpositive_or_nonfinite_reply_time(elapsed: float) -> None: + complexity, _ = _complexity() + with pytest.raises((TypeError, ValueError)): + budget_v2.MeasuredSample(complexity, elapsed) # type: ignore[arg-type] + + +def test_budget_gate_resolves_exact_changed_byte_measurement() -> None: + complexity, resolver = _complexity(changed_bytes=120) + budget_v2.check_budget(complexity, 100, _derivation(), resolver) + + +def test_budget_gate_refuses_unreadable_zero_or_substituted_measurement() -> None: + complexity, resolver = _complexity(changed_bytes=120) + missing = InMemoryContentResolver({}) + zero_reference, zero_resolver = _measurement(0) + with pytest.raises(ValueError): + replace(complexity, changed_bytes=0, measurement=zero_reference) + substituted = replace(complexity, changed_bytes=121) + assert zero_resolver.resolve(zero_reference).content + for change, selected in ((complexity, missing), (substituted, resolver)): + with pytest.raises((canonical.ArtifactResolutionError, canonical.ArtifactValidationError, ValueError)): + budget_v2.check_budget(change, 100, _derivation(), selected) + + +def test_budget_gate_names_change_elapsed_class_and_derivation() -> None: + complexity, resolver = _complexity() + derivation = _derivation() + with pytest.raises(budget_v2.BudgetGateFailure) as caught: + budget_v2.check_budget(complexity, derivation.budget_ms + 1, derivation, resolver) + message = str(caught.value) + assert complexity.change_id in message + assert str(derivation.budget_ms + 1) in message + assert derivation.workload_class in message + assert derivation.derivation_digest in message + + +def test_budget_gate_refuses_tampered_derivation_and_incompatible_change() -> None: + complexity, resolver = _complexity() + derivation = _derivation() + object.__setattr__(derivation, "sample_count", 99) + with pytest.raises(ValueError): + budget_v2.check_budget(complexity, 100, derivation, resolver) + incompatible, incompatible_resolver = _complexity(changed_regions=4) + with pytest.raises(ValueError): + budget_v2.check_budget(incompatible, 100, _derivation(), incompatible_resolver) + + +def test_scope_accepts_contained_regions_and_semantic_anchors() -> None: + allowed = _scope(evidence.SourceRegion("src/app.py", 1, 30), evidence.SourceRegion("src/other.py", 1, 5)) + packet = _packet(scope=_scope(evidence.SourceRegion("src/app.py", 5, 20))) + evidence.validate_scope(packet, allowed) + + +def test_scope_refuses_revision_change_or_region_escape() -> None: + allowed = _scope(evidence.SourceRegion("src/app.py", 1, 30)) + wrong_change = evidence.AnalyzedScope("base", "head", _digest("different"), allowed.regions) + escaped = _packet(scope=_scope(evidence.SourceRegion("src/app.py", 1, 31))) + for packet, boundary in ((_packet(), wrong_change), (escaped, allowed)): + with pytest.raises(evidence.ScopeValidationError): + evidence.validate_scope(packet, boundary) + + +def test_scope_refuses_relation_anchor_outside_declared_packet_region() -> None: + packet = _packet(line=25, scope=_scope(evidence.SourceRegion("src/app.py", 1, 20))) + with pytest.raises(evidence.ScopeValidationError): + evidence.validate_scope(packet, _scope()) + + +def _location_field(name: str, *, line: int) -> semantics.FieldValue: + return semantics.FieldValue( + name, + semantics.FieldKind.SOURCE_LOCATION, + {"path": "src/app.py", "line": line, "column": 0, "symbol": "handle"}, + ) + + +def test_scope_refuses_out_of_scope_observation_embedded_location() -> None: + packet = _packet(observation_fields=(_location_field("related_site", line=31),)) + with pytest.raises(evidence.ScopeValidationError): + evidence.validate_scope(packet, _scope()) + + +def test_scope_refuses_out_of_scope_family_field_location() -> None: + packet = _packet( + scope=_scope(evidence.SourceRegion("src/app.py", 1, 20)), + family_fields=(_location_field("claim_site", line=25),), + ) + with pytest.raises(evidence.ScopeValidationError): + evidence.validate_scope(packet, _scope()) + + +def test_scope_revalidates_object_setattr_field_tampering() -> None: + field = _location_field("claim_site", line=10) + packet = _packet(family_fields=(field,)) + object.__setattr__(field, "value", {"path": "src/other.py", "line": 1, "column": 0, "symbol": None}) + with pytest.raises(evidence.ScopeValidationError): + evidence.validate_scope(packet, _scope()) + + +def test_source_location_field_mapping_is_deeply_immutable() -> None: + field = _location_field("claim_site", line=10) + with pytest.raises(TypeError): + cast(dict[str, object], field.value)["line"] = 31 + + +def test_scope_revalidates_post_construction_tampering() -> None: + allowed = _scope() + object.__setattr__(allowed.regions[0], "end_line", 0) + with pytest.raises((evidence.EvidenceValidationError, evidence.ScopeValidationError)): + evidence.validate_scope(_packet(), allowed) + + +def test_legacy_budget_api_remains_available() -> None: + samples = ((budget_v2.Complexity(1, 1, 10), 1.0), (budget_v2.Complexity(2, 2, 20), 2.0)) + legacy = budget_v2.derive_budget(samples, workload_class="legacy", percentile=0.5, hardware="runner") + assert isinstance(legacy, budget_v2.Budget) + budget_v2.check_budget(change_id="change", seconds=1.0, budget=legacy)