diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index b0d82cb..79c7ea4 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -422,7 +422,7 @@ - Mutate a semantic identity and verify transitive staleness through all dependent readiness and measurement records. - Done when every injected failure is diagnosed at its source and no late admission-only surprise remains. - _Requirements: 5.3, 5.4, 5.5, 8.3, 8.4, 8.5, 15.4, 15.5_ -- [ ] 11.3 Verify bounded validation cost and total consumer deadline accounting +- [x] 11.3 Verify bounded validation cost and total consumer deadline accounting - Measure canonicalization and validation across increasing relation, observation, receipt, and case counts. - Exercise large transition sets for linear behavior and bounded memory without weakening semantic checks. - Include optional model and proof processing in the same measured consumer budget. diff --git a/src/openultrasast/core/budget.py b/src/openultrasast/core/budget.py index c319a32..c8dd785 100644 --- a/src/openultrasast/core/budget.py +++ b/src/openultrasast/core/budget.py @@ -7,6 +7,7 @@ import math import re from collections.abc import Sequence from dataclasses import dataclass +from fractions import Fraction from openultrasast.core import canonical @@ -38,13 +39,16 @@ def _nonnegative_int(value: object, field: str) -> None: 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 +def _percentile(value: object) -> tuple[float, Fraction]: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError("percentile must be finite") + if isinstance(value, float) and not math.isfinite(value): + raise ValueError("percentile must be finite") + if value <= 0: + raise ValueError("percentile must be positive") + if value > 1: + raise ValueError("percentile must not exceed 1") + return float(value), Fraction(str(value)) @dataclass(frozen=True) @@ -241,9 +245,7 @@ def derive_budget_derivation_digest(value: BudgetDerivation) -> str: 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") + _percentile(value.percentile) if not isinstance(value.reference_workload, ReferenceWorkload): raise TypeError("reference_workload must be ReferenceWorkload") if not isinstance(value.reference_hardware, ReferenceHardware): @@ -257,12 +259,12 @@ def _validate_derivation(value: BudgetDerivation, *, allow_empty_digest: bool = raise ValueError("budget derivation digest does not match fields") -def _quantile(sorted_values: Sequence[float], percentile: float) -> float: +def _quantile(sorted_values: Sequence[int], percentile: Fraction) -> Fraction: position = percentile * (len(sorted_values) - 1) - lower = math.floor(position) + lower = position.numerator // position.denominator upper = math.ceil(position) if lower == upper: - return sorted_values[lower] + return Fraction(sorted_values[lower]) weight = position - lower return sorted_values[lower] * (1 - weight) + sorted_values[upper] * weight @@ -286,8 +288,8 @@ def derive_budget( reference_hardware: ReferenceHardware, ) -> BudgetDerivation: _text(workload_class, "workload_class") - value = _finite_number(percentile, "percentile", positive=True) - if value > 1 or not samples: + value, exact_percentile = _percentile(percentile) + if not samples: raise ValueError("v2 derivation requires samples and percentile in (0, 1]") _validate_reference_workload(reference_workload) _validate_reference_hardware(reference_hardware) @@ -301,7 +303,7 @@ def derive_budget( 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)) + budget_ms = math.ceil(_quantile(sorted(item.total_reply_ms for item in measured), exact_percentile)) return BudgetDerivation(workload_class, len(measured), value, reference_workload, reference_hardware, budget_ms, "") diff --git a/tests/detection_core_consumer_performance.py b/tests/detection_core_consumer_performance.py new file mode 100644 index 0000000..d74ce0a --- /dev/null +++ b/tests/detection_core_consumer_performance.py @@ -0,0 +1,211 @@ +"""Complete consumer measurements used by the detection-core load contract.""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from detection_core_fixtures import ( + InMemoryContentResolver, + InMemoryPopulationCustody, + custody_contract, + stored_artifact, +) +from detection_core_performance_reporting import nanoseconds_to_milliseconds +from test_core_consumer_v2 import _proof +from test_core_validation_v2 import _catalog, _model_validation_case, _review_policy, _trust_policy, _with_fact_set + +from openultrasast.core import budget, canonical, family, provenance, validation +from openultrasast.core.consumer import PushRequest, consume_records, make_push_request +from openultrasast.core.proof import ProofAttachment, ProofTrustPolicy +from openultrasast.core.validation import EvidenceValidationContext, EvidenceValidationInput + +if TYPE_CHECKING: + from detection_core_performance import Workload + + from openultrasast.core.evidence import EvidenceRecord + + +@dataclass(frozen=True) +class PreparedConsumer: + value: EvidenceValidationInput + context: EvidenceValidationContext + custody: provenance.PopulationCustodyPort + resolver: InMemoryContentResolver + request: PushRequest + attachments: tuple[ProofAttachment, ...] + trust: ProofTrustPolicy | None + + +@dataclass(frozen=True) +class CompleteRun: + mode: str + size: int + measured: budget.MeasuredSample + resolver: InMemoryContentResolver + + +def merge_resolvers(values: tuple[InMemoryContentResolver, ...]) -> InMemoryContentResolver: + artifacts: dict[str, canonical.ResolvedContent] = {} + for value in values: + artifacts.update(value.artifacts) + return InMemoryContentResolver(artifacts) + + +def measurement(changed_bytes: int, artifact_id: str) -> tuple[bytes, canonical.ContentRef, InMemoryContentResolver]: + content = canonical.canonical_bytes({"schema_version": budget.CHANGED_BYTES_MEASUREMENT_SCHEMA, "changed_bytes": changed_bytes}) + reference, resolver = stored_artifact(content, artifact_id=artifact_id) + if not content or reference.byte_length != len(content): + raise AssertionError("changed-byte measurement input was not opened") + return content, reference, resolver + + +def _named_inputs( + record: EvidenceRecord, + contract: family.FamilyContract, + change: bytes, +) -> tuple[provenance.NamedInputs, InMemoryContentResolver]: + stamp = record.packet.provenance + contents = { + provenance.ProvenanceInputRole.ANALYZER: canonical.canonical_bytes("analyzer"), + provenance.ProvenanceInputRole.FAMILY_CONTRACT: canonical.canonical_bytes(family._contract_value(contract)), + provenance.ProvenanceInputRole.FACT_SET: validation.fact_set_bytes(()), + provenance.ProvenanceInputRole.ACQUISITION: canonical.canonical_bytes("acquisition"), + provenance.ProvenanceInputRole.CONFIGURATION: canonical.canonical_bytes("configuration"), + provenance.ProvenanceInputRole.POPULATION: canonical.canonical_bytes("population"), + provenance.ProvenanceInputRole.CHANGE_INPUT: change, + } + digests = { + provenance.ProvenanceInputRole.ANALYZER: stamp.analyzer_revision, + provenance.ProvenanceInputRole.FAMILY_CONTRACT: stamp.family_contract_digest, + provenance.ProvenanceInputRole.FACT_SET: stamp.fact_set_digest, + provenance.ProvenanceInputRole.ACQUISITION: stamp.acquisition_record_digest, + provenance.ProvenanceInputRole.CONFIGURATION: stamp.configuration_digest, + provenance.ProvenanceInputRole.POPULATION: stamp.population_digest, + provenance.ProvenanceInputRole.CHANGE_INPUT: stamp.change_input_digest, + } + items: list[provenance.NamedInput] = [] + resolvers: list[InMemoryContentResolver] = [] + for role in provenance.ProvenanceInputRole: + reference, resolver = stored_artifact(contents[role], artifact_id=f"performance-{role.value}-{stamp.change_input_digest}") + if reference.content_digest != digests[role] or reference.byte_length <= 0: + raise AssertionError(f"performance input {role.value} did not bind its exact bytes") + items.append(provenance.NamedInput(role, digests[role], reference)) + resolvers.append(resolver) + selected = tuple(items) + named = provenance.NamedInputs(selected, provenance.derive_named_inputs_digest(selected)) + return named, merge_resolvers(tuple(resolvers)) + + +def deterministic_prepared( + workload: Workload, + size: int, + derivation: budget.BudgetDerivation, +) -> PreparedConsumer: + from detection_core_performance import _record + + contract, raw, change = _record(workload, size) + record = _with_fact_set(raw, ()) + named, named_resolver = _named_inputs(record, contract, change) + initial, custody_policy, _ = custody_contract() + context = validation.make_evidence_validation_context( + family_catalog=_catalog(contract), + fact_packs=(), + model_programs=(), + lineages=(), + fact_review_policy=_review_policy(), + provenance_trust_policy=_trust_policy(), + custody_policy=custody_policy, + ) + measured, reference, measure_resolver = measurement(len(change), f"consumer-changed-bytes-{size}") + if measured != canonical.canonical_bytes({"schema_version": budget.CHANGED_BYTES_MEASUREMENT_SCHEMA, "changed_bytes": len(change)}): + raise AssertionError("changed-byte measurement serialization is unstable") + complexity = budget.ChangeComplexity( + record.packet.scope.change_digest, + workload.workload_class, + len(record.packet.scope.regions), + len(record.packet.transitions), + len(change), + reference, + ) + request = make_push_request( + scope=record.packet.scope, + workload_class=workload.workload_class, + complexity=complexity, + budget=derivation, + ) + value = validation.make_evidence_validation_input(record=record, named_inputs=named) + resolver = merge_resolvers((named_resolver, measure_resolver)) + custody = InMemoryPopulationCustody(initial, custody_policy) + return PreparedConsumer(value, context, custody, resolver, request, (), None) + + +def optional_prepared( + workload: Workload, + derivation: budget.BudgetDerivation, + artifact_id: str, +) -> PreparedConsumer: + value, context, custody, resolver, _ = _model_validation_case() + attachment, trust, proof_resolver = _proof(value.record) + selected = next(item for item in value.named_inputs.inputs if item.role is provenance.ProvenanceInputRole.CHANGE_INPUT) + change = resolver.resolve(selected.content).content + if ( + not change + or canonical.bytes_digest(change) != selected.content.content_digest + or selected.content.content_digest != value.record.packet.provenance.change_input_digest + ): + raise AssertionError("optional consumer did not open its exact change input") + _, reference, measure_resolver = measurement(len(change), artifact_id) + complexity = budget.ChangeComplexity( + value.record.packet.scope.change_digest, + workload.workload_class, + len(value.record.packet.scope.regions), + len(value.record.packet.transitions), + len(change), + reference, + ) + request = make_push_request( + scope=value.record.packet.scope, + workload_class=workload.workload_class, + complexity=complexity, + budget=derivation, + ) + combined = merge_resolvers((resolver, proof_resolver, measure_resolver)) + return PreparedConsumer(value, context, custody, combined, request, (attachment,), trust) + + +def measure_complete( + setup: Callable[[], PreparedConsumer], + mode: str, + size: int, +) -> CompleteRun: + started = time.monotonic_ns() + prepared = setup() + preparation_ms = nanoseconds_to_milliseconds(max(1, time.monotonic_ns() - started)) + verdict = consume_records( + prepared.request, + (prepared.value,), + prepared.context, + prepared.custody, + preparation_ms, + prepared.resolver, + attachments=prepared.attachments, + proof_trust=prepared.trust, + ) + wall_ms = nanoseconds_to_milliseconds(max(1, time.monotonic_ns() - started)) + elapsed = max(wall_ms, verdict.total_reply_ms) + sample = budget.MeasuredSample(prepared.request.complexity, elapsed) + return CompleteRun(mode, size, sample, prepared.resolver) + + +__all__ = [ + "CompleteRun", + "PreparedConsumer", + "deterministic_prepared", + "measure_complete", + "measurement", + "merge_resolvers", + "optional_prepared", +] diff --git a/tests/detection_core_performance.py b/tests/detection_core_performance.py new file mode 100644 index 0000000..5c1da65 --- /dev/null +++ b/tests/detection_core_performance.py @@ -0,0 +1,354 @@ +"""Test-owned measured workload composition over production detection-core validators.""" + +from __future__ import annotations + +import gc +import json +import time +import tracemalloc +from dataclasses import dataclass, replace +from functools import lru_cache, partial +from pathlib import Path + +import detection_core_performance_reporting as reporting +from detection_core_fixtures import InMemoryContentResolver, stored_artifact +from detection_core_scaling import _integer_median +from detection_core_work import ValidationWork, count_validation_work +from test_core_admission_metrics import _setup as admission_setup +from test_core_family import _contract as family_contract +from test_core_family import _stamp as family_stamp +from test_core_readiness import _chain as readiness_chain + +from openultrasast.core import admission_case, budget, canonical, evidence, family, readiness, semantics +from openultrasast.core.consumer import PushRequest +from openultrasast.core.proof import ProofAttachment, ProofTrustPolicy +from openultrasast.core.provenance import DecisionProvenance, PopulationCustodyPort +from openultrasast.core.validation import EvidenceValidationContext, EvidenceValidationInput + +PERFORMANCE_FIXTURE = Path(__file__).parent / "fixtures/detection_core/performance_workload.json" +_SCHEMA = "detection-core-performance-workload-v1" + + +@dataclass(frozen=True) +class Workload: + workload_class: str + sizes: tuple[int, ...] + repeats: int + percentile: float + changed_regions: int + changed_bytes: int + bytes_read: int + content_digest: str + + +@dataclass(frozen=True) +class PerformancePoint: + relation_count: int + observation_count: int + receipt_count: int + case_count: int + work: ValidationWork + elapsed_ns: int + elapsed_samples_ns: tuple[int, ...] + peak_bytes: int + peak_samples_bytes: tuple[int, ...] + serialized_bytes: int + change_digest: str + changed_bytes: int + changed_bytes_measurement: bytes + change_content: bytes + + @property + def elapsed_ms(self) -> int: + return reporting.nanoseconds_to_milliseconds(self.elapsed_ns) + + +@dataclass(frozen=True) +class ConsumerMeasurement: + mode: str + size: int + measured: budget.MeasuredSample + + @property + def total_reply_ms(self) -> int: + return self.measured.total_reply_ms + + +@dataclass(frozen=True) +class PerformanceProfile: + workload: Workload + points: tuple[PerformancePoint, ...] + consumer_samples: tuple[ConsumerMeasurement, ...] + samples: tuple[budget.MeasuredSample, ...] + reference_workload: budget.ReferenceWorkload + reference_hardware: budget.ReferenceHardware + budget: budget.BudgetDerivation + resolver: InMemoryContentResolver + sample_bytes: int + + +@dataclass(frozen=True) +class OptionalConsumerCase: + value: EvidenceValidationInput + context: EvidenceValidationContext + custody: PopulationCustodyPort + resolver: InMemoryContentResolver + request: PushRequest + attachment: ProofAttachment + trust: ProofTrustPolicy + + +def _strict_object(value: object) -> dict[str, object]: + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + raise ValueError("performance workload must be an object") + return value + + +def _positive_int(value: object, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ValueError(f"{field} must be a positive integer") + return value + + +def read_workload(path: Path) -> Workload: + content = path.read_bytes() + if not content: + raise ValueError("performance workload input is empty") + raw = _strict_object(json.loads(content)) + expected = {"schema_version", "workload_class", "sizes", "repeats", "percentile", "changed_regions", "changed_bytes"} + if set(raw) != expected or raw["schema_version"] != _SCHEMA: + raise ValueError("performance workload schema is not the reviewed closed schema") + sizes_raw = raw["sizes"] + if not isinstance(sizes_raw, list): + raise ValueError("performance workload sizes must be a list") + sizes = tuple(_positive_int(item, "workload size") for item in sizes_raw) + if len(sizes) < 3 or tuple(sorted(set(sizes))) != sizes: + raise ValueError("performance workload sizes must be unique and increasing") + workload_class = raw["workload_class"] + percentile = raw["percentile"] + if not isinstance(workload_class, str) or not workload_class: + raise ValueError("workload class must be nonempty") + if isinstance(percentile, bool) or not isinstance(percentile, (int, float)) or not 0 < percentile <= 1: + raise ValueError("percentile must be in (0, 1]") + return Workload( + workload_class, + sizes, + _positive_int(raw["repeats"], "repeats"), + float(percentile), + _positive_int(raw["changed_regions"], "changed regions"), + _positive_int(raw["changed_bytes"], "changed bytes"), + len(content), + canonical.bytes_digest(content), + ) + + +def _location(line: int) -> semantics.SourceLocation: + return semantics.SourceLocation("src/app.py", line, 0, "handle") + + +def _provenance() -> semantics.EvidenceProvenance: + return semantics.EvidenceProvenance( + semantics.ProvenanceKind.PRODUCT_CODE, + canonical.content_digest("performance-source"), + None, + None, + None, + ) + + +def _relation(index: int) -> semantics.SemanticRelation: + line = index * 2 + 2 + return semantics.make_relation( + kind=semantics.RelationKind.CONSTRAINT_EFFECT, + role="constraint", + source=semantics.SemanticNode(f"operation-{index}", "operation", _location(line)), + target=semantics.SemanticNode(f"policy-{index}", "policy", _location(line + 1)), + location=_location(line), + status=semantics.RelationStatus.COMPLETE, + conditions=(f"policy-{index}",), + reasons=(), + provenance=_provenance(), + ) + + +def _observation(index: int) -> semantics.Observation: + return semantics.make_observation( + role="security_object", + location=_location(index * 2 + 2), + fields=(semantics.FieldValue("resource", semantics.FieldKind.TEXT, f"document-{index}"),), + status=semantics.RelationStatus.COMPLETE, + provenance=_provenance(), + ) + + +def _change_content(workload: Workload, size: int) -> bytes: + return canonical.canonical_bytes( + {"workload_input_digest": workload.content_digest, "size": size, "source": "x" * (workload.changed_bytes + size)} + ) + + +def _record(workload: Workload, size: int) -> tuple[family.FamilyContract, evidence.EvidenceRecord, bytes]: + contract = family_contract() + content = _change_content(workload, size) + change_digest = canonical.bytes_digest(content) + fields = (semantics.FieldValue("operation", semantics.FieldKind.TEXT, "read"),) + scope = evidence.AnalyzedScope( + "base", + "head", + change_digest, + (evidence.SourceRegion("src/app.py", 1, size * 2 + 4),), + ) + packet = evidence.EvidencePacket( + evidence.EVIDENCE_PACKET_SCHEMA, + contract.family_id, + family.family_contract_digest(contract), + family.derive_security_object_id(contract.security_object, fields), + "python", + scope, + tuple(_relation(index) for index in range(size)), + tuple(_observation(index) for index in range(size)), + fields, + (), + replace(family_stamp(contract), change_input_digest=change_digest), + ) + decision = evidence.Decision( + "blocked", + evidence.NeutralOutcome.VIOLATION, + evidence.EvidenceBasis.DETERMINISTIC, + scope, + (), + evidence.derive_packet_digest(packet), + DecisionProvenance(None, None, None, None, None, None), + ) + return contract, evidence.EvidenceRecord(packet, decision, ""), content + + +def _validate_work( + record: evidence.EvidenceRecord, + contract: family.FamilyContract, + size: int, + receipts: tuple[readiness.ReadinessResult, ...], + cases: tuple[admission_case.AdmissionCaseResult, ...], +) -> int: + content = evidence.evidence_record_bytes(record) + family.validate_record(contract, record) + for index in range(size): + readiness.validate_readiness_result(receipts[index % len(receipts)]) + for index in range(size): + admission_case.validate_admission_case_shape(cases[index % len(cases)]) + return len(content) + + +def _measure_point( + workload: Workload, + size: int, + receipts: tuple[readiness.ReadinessResult, ...], + cases: tuple[admission_case.AdmissionCaseResult, ...], +) -> PerformancePoint: + contract, record, content = _record(workload, size) + elapsed: list[int] = [] + peaks: list[int] = [] + _validate_work(record, contract, size, receipts, cases) + for _ in range(workload.repeats): + started = time.perf_counter_ns() + _validate_work(record, contract, size, receipts, cases) + elapsed.append(max(1, time.perf_counter_ns() - started)) + tracemalloc.start() + _validate_work(record, contract, size, receipts, cases) + tracemalloc.stop() + for _ in range(workload.repeats): + gc.collect() + tracemalloc.start() + _validate_work(record, contract, size, receipts, cases) + _, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + peaks.append(peak) + serialized, work = count_validation_work(lambda: _validate_work(record, contract, size, receipts, cases)) + measurement, _, _ = reporting.measurement(size, len(content)) + return PerformancePoint( + size, + size, + size, + size, + work, + _integer_median(tuple(elapsed)), + tuple(elapsed), + _integer_median(tuple(peaks)), + tuple(peaks), + serialized, + record.packet.scope.change_digest, + len(content), + measurement, + content, + ) + + +@lru_cache(maxsize=1) +def measured_profile(workload: Workload) -> PerformanceProfile: + import detection_core_consumer_performance as consumer_performance + + cases = admission_setup().cases + receipts = readiness_chain() + points = tuple(_measure_point(workload, size, receipts, cases) for size in workload.sizes) + hardware = reporting.hardware() + seed = reporting.seed_budget(workload, points, hardware) + complete = [] + for point in points: + for _ in range(workload.repeats): + setup = partial(consumer_performance.deterministic_prepared, workload, point.relation_count, seed) + complete.append(consumer_performance.measure_complete(setup, "deterministic", point.relation_count)) + for repeat in range(workload.repeats): + setup = partial(consumer_performance.optional_prepared, workload, seed, f"optional-measurement-{repeat}") + complete.append(consumer_performance.measure_complete(setup, "model_and_proof", 1)) + consumers = tuple(ConsumerMeasurement(item.mode, item.size, item.measured) for item in complete) + report = reporting.sample_report(workload, points, consumers) + report_ref, report_resolver = stored_artifact(report, artifact_id="performance-complete-samples") + changed = tuple(item.measured.complexity.changed_bytes for item in complete) + reference = reporting.reference(workload, report_ref, (min(changed), max(changed))) + samples = tuple(item.measured for item in complete) + derived = budget.derive_budget(samples, workload.workload_class, workload.percentile, reference, hardware) + return PerformanceProfile( + workload, + points, + consumers, + samples, + reference, + hardware, + derived, + consumer_performance.merge_resolvers((report_resolver, *(item.resolver for item in complete))), + len(report), + ) + + +def assert_large_set_envelope(profile: PerformanceProfile) -> None: + from detection_core_scaling import assert_incremental_envelope + + assert_incremental_envelope(profile.points) + + +def reject_invalid_large_set(workload: Workload, size: int) -> None: + contract, record, _ = _record(workload, size) + invalid = replace(record.packet.transitions[-1], role="undeclared-role", relation_id="") + packet = replace(record.packet, transitions=(*record.packet.transitions[:-1], invalid)) + decision = replace(record.decision, packet_digest=evidence.derive_packet_digest(packet)) + candidate = evidence.EvidenceRecord(packet, decision, "") + try: + family.validate_record(contract, candidate) + except family.FamilyContractError: + return + raise AssertionError("large invalid semantic set was accepted") + + +def optional_consumer_case(profile: PerformanceProfile) -> OptionalConsumerCase: + from detection_core_consumer_performance import optional_prepared + + prepared = optional_prepared(profile.workload, profile.budget, "optional-final-measurement") + return OptionalConsumerCase( + prepared.value, + prepared.context, + prepared.custody, + prepared.resolver, + prepared.request, + prepared.attachments[0], + prepared.trust, + ) diff --git a/tests/detection_core_performance_reporting.py b/tests/detection_core_performance_reporting.py new file mode 100644 index 0000000..7efa3b8 --- /dev/null +++ b/tests/detection_core_performance_reporting.py @@ -0,0 +1,139 @@ +"""Measurement artifact and budget composition for detection-core load tests.""" + +from __future__ import annotations + +import os +import platform +import sys +from dataclasses import replace +from typing import TYPE_CHECKING + +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import budget, canonical + +if TYPE_CHECKING: + from detection_core_performance import ConsumerMeasurement, PerformancePoint, Workload + +_SAMPLE_SCHEMA = "detection-core-performance-samples-v1" + + +def nanoseconds_to_milliseconds(value: object) -> int: + if type(value) is not int or value <= 0: + raise ValueError("elapsed nanoseconds must be an exact positive integer") + return (value + 999_999) // 1_000_000 + + +def hardware() -> budget.ReferenceHardware: + pages = os.sysconf("SC_PHYS_PAGES") + page_size = os.sysconf("SC_PAGE_SIZE") + value = budget.ReferenceHardware( + platform.processor() or platform.machine() or "unreported-cpu", + os.cpu_count() or 1, + int(pages * page_size), + platform.platform(), + sys.version.split()[0], + "", + ) + return replace(value, record_digest=budget.derive_reference_hardware_digest(value)) + + +def sample_report( + workload: Workload, + points: tuple[PerformancePoint, ...], + consumers: tuple[ConsumerMeasurement, ...], +) -> bytes: + return canonical.canonical_bytes( + { + "schema_version": _SAMPLE_SCHEMA, + "workload_class": workload.workload_class, + "workload_input_digest": workload.content_digest, + "repeats": workload.repeats, + "points": tuple(_point_value(point) for point in points), + "consumer_samples": tuple(_consumer_value(item) for item in consumers), + } + ) + + +def _point_value(point: PerformancePoint) -> dict[str, canonical.JsonValue]: + return { + "relations": point.relation_count, + "observations": point.observation_count, + "receipts": point.receipt_count, + "cases": point.case_count, + "work": { + "canonical_relations": point.work.canonical_relations, + "canonical_observations": point.work.canonical_observations, + "validated_relations": point.work.validated_relations, + "validated_observations": point.work.validated_observations, + "readiness_receipts": point.work.readiness_receipts, + "admission_cases": point.work.admission_cases, + "total_operations": point.work.total_operations, + }, + "elapsed_samples_ns": point.elapsed_samples_ns, + "peak_samples_bytes": point.peak_samples_bytes, + "serialized_bytes": point.serialized_bytes, + "change_digest": point.change_digest, + "changed_bytes": point.changed_bytes, + } + + +def _consumer_value(item: ConsumerMeasurement) -> dict[str, canonical.JsonValue]: + return { + "mode": item.mode, + "size": item.size, + "change_id": item.measured.complexity.change_id, + "changed_regions": item.measured.complexity.changed_regions, + "transition_count": item.measured.complexity.transition_count, + "changed_bytes": item.measured.complexity.changed_bytes, + "total_reply_ms": item.total_reply_ms, + } + + +def measurement(size: int, changed_bytes: int) -> tuple[bytes, canonical.ContentRef, InMemoryContentResolver]: + content = canonical.canonical_bytes({"schema_version": budget.CHANGED_BYTES_MEASUREMENT_SCHEMA, "changed_bytes": changed_bytes}) + reference, resolver = stored_artifact(content, artifact_id=f"performance-changed-bytes-{size}") + return content, reference, resolver + + +def reference( + workload: Workload, + report_ref: canonical.ContentRef, + changed_range: tuple[int, int], +) -> budget.ReferenceWorkload: + value = budget.ReferenceWorkload( + "measured-detection-core-consumer", + "complete deterministic and model+proof consumer runs; scaled relations, observations, receipts, and cases", + (workload.changed_regions, workload.changed_regions), + (1, workload.sizes[-1]), + changed_range, + report_ref, + "", + ) + return replace(value, record_digest=budget.derive_reference_workload_digest(value)) + + +def seed_budget( + workload: Workload, + points: tuple[PerformancePoint, ...], + reference_hardware: budget.ReferenceHardware, +) -> budget.BudgetDerivation: + report = sample_report(workload, points, ()) + report_ref, _ = stored_artifact(report, artifact_id="performance-calibration-seed") + reference_workload = reference(workload, report_ref, (1, max(point.changed_bytes for point in points))) + samples = [] + for point in points: + _, measured_bytes, _ = measurement(point.relation_count, point.changed_bytes) + complexity = budget.ChangeComplexity( + point.change_digest, + workload.workload_class, + workload.changed_regions, + point.relation_count, + point.changed_bytes, + measured_bytes, + ) + samples.extend(budget.MeasuredSample(complexity, nanoseconds_to_milliseconds(value)) for value in point.elapsed_samples_ns) + return budget.derive_budget(tuple(samples), workload.workload_class, 1.0, reference_workload, reference_hardware) + + +__all__ = ["hardware", "measurement", "reference", "sample_report", "seed_budget"] diff --git a/tests/detection_core_scaling.py b/tests/detection_core_scaling.py new file mode 100644 index 0000000..4e1c264 --- /dev/null +++ b/tests/detection_core_scaling.py @@ -0,0 +1,127 @@ +"""Deterministic work and isolated-memory scaling checks.""" + +from __future__ import annotations + +from dataclasses import dataclass +from fractions import Fraction +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from detection_core_performance import PerformancePoint + + +@dataclass(frozen=True) +class _SlopeInterval: + lower: Fraction + upper: Fraction + + +def _work_units(point: PerformancePoint) -> int: + return point.relation_count + point.observation_count + point.receipt_count + point.case_count + + +def _positive_integer(value: object, name: str) -> int: + if type(value) is not int or value <= 0: + raise AssertionError(f"incremental validation {name} requires exact positive integer values") + return value + + +def _integer_median(values: tuple[int, ...]) -> int: + ordered = tuple(sorted(values)) + midpoint = len(ordered) // 2 + if len(ordered) % 2: + return ordered[midpoint] + return (ordered[midpoint - 1] + ordered[midpoint]) // 2 + + +def _series( + points: tuple[PerformancePoint, ...], + sample_field: str, + summary_field: str, + name: str, +) -> tuple[tuple[int, ...], tuple[int, ...]]: + samples = tuple(tuple(getattr(point, sample_field)) for point in points) + if len({len(values) for values in samples}) != 1: + raise AssertionError("incremental validation measurements require equal repeat counts") + if any(len(values) < 2 for values in samples): + raise AssertionError("incremental validation measurements require repeated positive integer samples") + for values in samples: + for value in values: + _positive_integer(value, name) + medians = tuple(_integer_median(values) for values in samples) + summaries = tuple(_positive_integer(getattr(point, summary_field), name) for point in points) + if medians != summaries: + raise AssertionError(f"incremental validation {name} summary does not match repeated median") + spreads = tuple(max(values) - min(values) for values in samples) + return medians, spreads + + +def _slope_intervals( + units: tuple[int, ...], + medians: tuple[int, ...], + spreads: tuple[int, ...], +) -> tuple[_SlopeInterval, ...]: + result = [] + pairs = zip(units[:-1], units[1:], medians[:-1], medians[1:], spreads[:-1], spreads[1:], strict=True) + for old_units, new_units, old_value, new_value, old_spread, new_spread in pairs: + width = new_units - old_units + increment = new_value - old_value + uncertainty = Fraction(old_spread + new_spread, 2) + lower = (increment - uncertainty) / width + upper = (increment + uncertainty) / width + result.append(_SlopeInterval(lower, upper)) + return tuple(result) + + +def _assert_memory(points: tuple[PerformancePoint, ...], units: tuple[int, ...]) -> None: + medians, spreads = _series(points, "peak_samples_bytes", "peak_bytes", "memory") + for median, spread in zip(medians, spreads, strict=True): + if spread > max(1_024, median // 10): + raise AssertionError("incremental validation memory is obscured by point-local variability") + increments = tuple(new - old for old, new in zip(medians[:-1], medians[1:], strict=True)) + if any(increment <= 0 for increment in increments): + raise AssertionError("incremental validation memory requires monotonic measured peaks") + slopes = _slope_intervals(units, medians, spreads) + for previous, current in zip(slopes[:-1], slopes[1:], strict=True): + if current.lower > previous.upper * Fraction(51, 50): + raise AssertionError( + f"incremental validation memory slope increased beyond point-local uncertainty: {current.lower}>{previous.upper}" + ) + + +def _assert_work(points: tuple[PerformancePoint, ...]) -> None: + pairs = ( + ("canonical_relations", "relation_count"), + ("validated_relations", "relation_count"), + ("canonical_observations", "observation_count"), + ("validated_observations", "observation_count"), + ("readiness_receipts", "receipt_count"), + ("admission_cases", "case_count"), + ) + for point in points: + expected_total = 0 + for work_field, point_field in pairs: + measured = _positive_integer(getattr(point.work, work_field), "work") + expected = _positive_integer(getattr(point, point_field), "work") + if measured != expected: + raise AssertionError(f"incremental validation work missed {work_field}: {measured}!={expected}") + expected_total += expected + total = _positive_integer(point.work.total_operations, "work") + if total != expected_total: + raise AssertionError(f"incremental validation work total disagrees: {total}!={expected_total}") + + +def assert_incremental_envelope(points: tuple[PerformancePoint, ...]) -> None: + units = tuple(_work_units(point) for point in points) + dimensions = tuple( + tuple(getattr(point, field) for point in points) for field in ("relation_count", "observation_count", "receipt_count", "case_count") + ) + increasing = all(all(new > old for old, new in zip(values[:-1], values[1:], strict=True)) for values in dimensions) + if len(units) < 3 or not increasing: + raise AssertionError("incremental validation requires increasing workload points") + _series(points, "elapsed_samples_ns", "elapsed_ns", "elapsed") + _assert_work(points) + _assert_memory(points, units) + + +__all__ = ["assert_incremental_envelope"] diff --git a/tests/detection_core_work.py b/tests/detection_core_work.py new file mode 100644 index 0000000..59d6a9e --- /dev/null +++ b/tests/detection_core_work.py @@ -0,0 +1,65 @@ +"""Test-owned counters around actual detection-core validation boundaries.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from types import ModuleType +from typing import Any +from unittest.mock import patch + +from openultrasast.core import admission_case, evidence, family, readiness + + +@dataclass(frozen=True) +class ValidationWork: + canonical_relations: int + canonical_observations: int + validated_relations: int + validated_observations: int + readiness_receipts: int + admission_cases: int + total_operations: int + + +def _counted( + module: ModuleType, + function_name: str, + counter: dict[str, int], + counter_name: str, +) -> Any: + original = getattr(module, function_name) + + def wrapped(*args: object, **kwargs: object) -> Any: + result = original(*args, **kwargs) + counter[counter_name] += 1 + return result + + return patch.object(module, function_name, wrapped) + + +def count_validation_work(run: Callable[[], int]) -> tuple[int, ValidationWork]: + """Run real production validators and count each successful scaled boundary.""" + counts = { + "canonical_relations": 0, + "canonical_observations": 0, + "validated_relations": 0, + "validated_observations": 0, + "readiness_receipts": 0, + "admission_cases": 0, + } + hooks = ( + _counted(evidence, "_relation_value", counts, "canonical_relations"), + _counted(evidence, "_observation_value", counts, "canonical_observations"), + _counted(family, "_validate_relation", counts, "validated_relations"), + _counted(family, "_validate_observation", counts, "validated_observations"), + _counted(readiness, "validate_readiness_result", counts, "readiness_receipts"), + _counted(admission_case, "validate_admission_case_shape", counts, "admission_cases"), + ) + with hooks[0], hooks[1], hooks[2], hooks[3], hooks[4], hooks[5]: + serialized_bytes = run() + values = tuple(counts.values()) + return serialized_bytes, ValidationWork(*values, sum(values)) + + +__all__ = ["ValidationWork", "count_validation_work"] diff --git a/tests/fixtures/detection_core/performance_workload.json b/tests/fixtures/detection_core/performance_workload.json new file mode 100644 index 0000000..454fbb8 --- /dev/null +++ b/tests/fixtures/detection_core/performance_workload.json @@ -0,0 +1,9 @@ +{ + "schema_version": "detection-core-performance-workload-v1", + "workload_class": "small", + "sizes": [8, 32, 128, 256], + "repeats": 3, + "percentile": 1.0, + "changed_regions": 1, + "changed_bytes": 120 +} diff --git a/tests/test_core_budget_v2.py b/tests/test_core_budget_v2.py index 583731e..e276d65 100644 --- a/tests/test_core_budget_v2.py +++ b/tests/test_core_budget_v2.py @@ -161,6 +161,7 @@ def test_derivation_is_deterministic_and_discloses_measurement_context() -> None assert first.sample_count == 5 assert first.percentile == 0.8 assert first.budget_ms == 144 + assert first.derivation_digest == "89149c1b78b413f543cc3c5950f9ee1b55379c11feb7499d9d49e972f5427b93" assert len(first.derivation_digest) == 64 @@ -193,6 +194,58 @@ def test_derivation_refuses_missing_or_nonfinite_inputs( ) +@pytest.mark.parametrize( + "percentile", + (10**400, -(10**400), True, "0.8", object(), math.inf, -math.inf, math.nan, 0, -1, 1.1), +) +def test_derivation_refuses_oversized_or_malformed_percentile_without_overflow(percentile: object) -> None: + with pytest.raises((TypeError, ValueError)) as caught: + budget_v2.derive_budget( + _samples(), + workload_class="small", + percentile=percentile, # type: ignore[arg-type] + reference_workload=_reference_workload(), + reference_hardware=_hardware(), + ) + assert not isinstance(caught.value, OverflowError) + + +def test_huge_reply_times_derive_check_and_excess_remain_exact() -> None: + huge = 10**400 + complexity, resolver = _complexity() + samples = ( + budget_v2.MeasuredSample(complexity, huge), + budget_v2.MeasuredSample(complexity, huge + 2), + ) + derived = budget_v2.derive_budget( + samples, + workload_class="small", + percentile=0.5, + reference_workload=_reference_workload(), + reference_hardware=_hardware(), + ) + + assert derived.budget_ms == huge + 1 + assert derived == replace(derived, derivation_digest="") + encoded = canonical.canonical_bytes({"budget_ms": derived.budget_ms}) + assert encoded == canonical.canonical_bytes({"budget_ms": huge + 1}) + assert str(huge + 1).encode() in encoded + budget_v2.check_budget(complexity, huge + 1, derived, resolver) + with pytest.raises(budget_v2.BudgetGateFailure, match="budget:") as caught: + budget_v2.check_budget(complexity, huge + 2, derived, resolver) + assert f"{huge + 2}ms>{huge + 1}ms" in str(caught.value) + + +def test_budget_derivation_refuses_oversized_percentile_after_tampering() -> None: + derived = _derivation() + object.__setattr__(derived, "percentile", 10**400) + complexity, resolver = _complexity() + + with pytest.raises(ValueError, match="percentile must not exceed 1") as caught: + budget_v2.check_budget(complexity, 100, derived, resolver) + assert not isinstance(caught.value, OverflowError) + + def test_derivation_refuses_mixed_classes_and_incompatible_complexity() -> None: wrong_class, _ = _complexity(workload_class="large") too_large, _ = _complexity(changed_bytes=501) diff --git a/tests/test_detection_core_performance.py b/tests/test_detection_core_performance.py new file mode 100644 index 0000000..666df91 --- /dev/null +++ b/tests/test_detection_core_performance.py @@ -0,0 +1,379 @@ +"""Measured performance and total consumer deadline validation for detection-core.""" + +from __future__ import annotations + +import json +import math +from dataclasses import replace + +import pytest +from detection_core_performance import ( + PERFORMANCE_FIXTURE, + PerformanceProfile, + assert_large_set_envelope, + measured_profile, + optional_consumer_case, + read_workload, + reject_invalid_large_set, +) +from detection_core_scaling import _series + +import openultrasast.core.consumer as consumer_module +from openultrasast.core import budget, canonical +from openultrasast.core.consumer import ConsumerContractError, consume_records, make_push_request + + +def _instrument_optional_path( + monkeypatch: pytest.MonkeyPatch, + profile: PerformanceProfile, +) -> tuple[list[str], list[int]]: + calls: list[str] = [] + clock = [0] + original_validate = consumer_module.validate_evidence_input + original_attach = consumer_module.attach_differential_proof + original_outputs = consumer_module._derive_outputs + + def measured_validate(*args: object, **kwargs: object) -> object: + result = original_validate(*args, **kwargs) # type: ignore[arg-type] + calls.append("model") + return result + + def measured_attach(*args: object, **kwargs: object) -> object: + result = original_attach(*args, **kwargs) # type: ignore[arg-type] + calls.append("proof") + return result + + def delayed_outputs(*args: object, **kwargs: object) -> object: + result = original_outputs(*args, **kwargs) # type: ignore[arg-type] + clock[0] += (profile.budget.budget_ms + 1) * 1_000_000 + return result + + monkeypatch.setattr(consumer_module, "validate_evidence_input", measured_validate) + monkeypatch.setattr(consumer_module, "attach_differential_proof", measured_attach) + monkeypatch.setattr(consumer_module, "_derive_outputs", delayed_outputs) + monkeypatch.setattr(consumer_module, "_monotonic_ns", lambda: clock[0]) + return calls, clock + + +def test_measured_profile_discloses_increasing_workload_and_fixed_budget_contract() -> None: + workload = read_workload(PERFORMANCE_FIXTURE) + profile = measured_profile(workload) + + assert workload.bytes_read > 0 + assert len(workload.content_digest) == 64 + assert tuple(point.relation_count for point in profile.points) == workload.sizes + assert tuple(point.observation_count for point in profile.points) == workload.sizes + assert tuple(point.receipt_count for point in profile.points) == workload.sizes + assert tuple(point.case_count for point in profile.points) == workload.sizes + assert all( + type(value) is int and value > 0 + for point in profile.points + for value in (point.elapsed_ns, *point.elapsed_samples_ns, point.peak_bytes, *point.peak_samples_bytes) + ) + assert profile.sample_bytes > 0 + expected_samples = (len(workload.sizes) + 1) * workload.repeats + assert profile.budget.sample_count == len(profile.samples) == expected_samples + assert profile.budget.percentile == workload.percentile + assert profile.budget.reference_workload == profile.reference_workload + assert profile.budget.reference_hardware == profile.reference_hardware + assert profile.budget.budget_ms == max(sample.total_reply_ms for sample in profile.samples) + assert len(profile.budget.derivation_digest) == 64 + + +def test_large_transition_validation_is_proportional_bounded_and_semantically_strict() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + + assert_large_set_envelope(profile) + reject_invalid_large_set(profile.workload, profile.workload.sizes[-1]) + + +def _adversarial_values(profile: PerformanceProfile, shape: str) -> tuple[int, ...]: + sizes = tuple(point.relation_count for point in profile.points) + if shape == "quadratic": + return tuple(size**2 for size in sizes) + if shape == "fixed-overhead-quadratic": + return tuple(100_000 + size**2 for size in sizes) + if shape == "n^1.1": + return tuple(round(10_000 * size**1.1) for size in sizes) + if shape == "n^1.25": + return tuple(round(10_000 * size**1.25) for size in sizes) + if shape == "mild-superlinear": + return tuple(round(10_000 * size**1.05) for size in sizes) + return (1_000, 4_000, 500_000, 502_000) + + +def _valid_memory_points(profile: PerformanceProfile) -> tuple[object, ...]: + return tuple( + replace( + point, + peak_bytes=100_000 + point.relation_count * 1_000, + peak_samples_bytes=(100_000 + point.relation_count * 1_000,) * profile.workload.repeats, + ) + for point in profile.points + ) + + +def _adversarial_points( + profile: PerformanceProfile, + target: str, + values: tuple[int, ...], +) -> tuple[object, ...]: + source = _valid_memory_points(profile) if target == "work" else profile.points + if target == "work": + return tuple(replace(point, work=replace(point.work, total_operations=value)) for point, value in zip(source, values, strict=True)) + return tuple( + replace(point, peak_bytes=value, peak_samples_bytes=(value,) * profile.workload.repeats) + for point, value in zip(source, values, strict=True) + ) + + +@pytest.mark.parametrize("target", ("work", "memory")) +@pytest.mark.parametrize( + "shape", + ("quadratic", "fixed-overhead-quadratic", "n^1.1", "n^1.25", "mild-superlinear", "penultimate-spike"), +) +def test_adversarial_superlinear_profiles_are_rejected(target: str, shape: str) -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + values = _adversarial_values(profile, shape) + points = _adversarial_points(profile, target, values) + with pytest.raises(AssertionError, match=f"incremental validation {target}"): + assert_large_set_envelope(replace(profile, points=points)) + + +def test_elapsed_samples_are_published_but_not_used_as_deterministic_scaling_oracle() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + values = (1_000, 4_000, 500_000, 502_000) + points = tuple( + replace(point, elapsed_ns=value, elapsed_samples_ns=(value,) * profile.workload.repeats) + for point, value in zip(_valid_memory_points(profile), values, strict=True) + ) + + assert_large_set_envelope(replace(profile, points=points)) + + +@pytest.mark.parametrize("metric", ("elapsed", "memory")) +@pytest.mark.parametrize("invalid", (math.inf, math.nan, 1.5, True, 0, -1)) +def test_measurement_series_refuse_non_integer_nonpositive_or_nonfinite_values(metric: str, invalid: object) -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + point = profile.points[-1] + if metric == "elapsed": + changed = replace(point, elapsed_samples_ns=(*point.elapsed_samples_ns[:-1], invalid)) + else: + changed = replace(point, peak_samples_bytes=(*point.peak_samples_bytes[:-1], invalid)) + with pytest.raises(AssertionError, match="positive integer"): + assert_large_set_envelope(replace(profile, points=(*profile.points[:-1], changed))) + + +@pytest.mark.parametrize("metric", ("elapsed", "memory")) +def test_measurement_series_refuse_unequal_repeats_and_mismatched_summaries(metric: str) -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + point = profile.points[-1] + if metric == "elapsed": + unequal = replace(point, elapsed_samples_ns=point.elapsed_samples_ns[:-1]) + mismatch = replace(point, elapsed_ns=point.elapsed_ns + 1) + else: + unequal = replace(point, peak_samples_bytes=point.peak_samples_bytes[:-1]) + mismatch = replace(point, peak_bytes=point.peak_bytes + 1) + with pytest.raises(AssertionError, match="equal repeat counts"): + assert_large_set_envelope(replace(profile, points=(*profile.points[:-1], unequal))) + with pytest.raises(AssertionError, match="summary does not match repeated median"): + assert_large_set_envelope(replace(profile, points=(*profile.points[:-1], mismatch))) + + +def test_inflated_repeat_variance_cannot_hide_superlinear_memory() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + values = _adversarial_values(profile, "n^1.25") + points = tuple( + replace( + point, + peak_bytes=value, + peak_samples_bytes=(max(1, value // 100), value, value * 100), + ) + for point, value in zip(profile.points, values, strict=True) + ) + with pytest.raises(AssertionError, match="incremental validation memory"): + assert_large_set_envelope(replace(profile, points=points)) + + +def test_huge_work_disagreement_is_a_named_domain_failure() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + point = profile.points[-1] + changed = replace(point, work=replace(point.work, total_operations=10**400)) + + with pytest.raises(AssertionError, match="incremental validation work total disagrees"): + assert_large_set_envelope(replace(profile, points=(*profile.points[:-1], changed))) + + +def test_huge_superlinear_memory_is_a_named_domain_failure() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + huge = 10**400 + values = (huge, huge * 2, huge * 8, huge * 64) + points = tuple( + replace(point, peak_bytes=value, peak_samples_bytes=(value,) * profile.workload.repeats) + for point, value in zip(profile.points, values, strict=True) + ) + + with pytest.raises(AssertionError, match="incremental validation memory slope increased"): + assert_large_set_envelope(replace(profile, points=points)) + + +def test_huge_even_repeat_median_and_elapsed_conversion_remain_exact() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + huge = 10**400 + point = replace( + profile.points[-1], + elapsed_ns=huge + 1, + elapsed_samples_ns=(huge, huge + 2, huge, huge + 2), + ) + medians, _ = _series((point,), "elapsed_samples_ns", "elapsed_ns", "elapsed") + assert medians == (huge + 1,) + + point = replace(point, elapsed_ns=huge + 999_999) + assert point.elapsed_ms == (huge + 1_999_998) // 1_000_000 + + +def test_huge_elapsed_samples_publish_and_derive_without_float_narrowing() -> None: + from detection_core_performance_reporting import hardware, sample_report, seed_budget + + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + huge = 10**400 + points = tuple( + replace( + point, + elapsed_ns=huge + index, + elapsed_samples_ns=(huge + index,) * profile.workload.repeats, + ) + for index, point in enumerate(profile.points) + ) + + report = sample_report(profile.workload, points, ()) + assert str(huge).encode() in report + derived = seed_budget(profile.workload, points, hardware()) + assert derived.budget_ms == (huge + len(points) - 1 + 999_999) // 1_000_000 + + +@pytest.mark.parametrize("invalid", (math.inf, math.nan, 1.5, True, 0, -1, "1")) +def test_elapsed_conversion_refuses_non_exact_or_nonpositive_values(invalid: object) -> None: + from detection_core_performance_reporting import nanoseconds_to_milliseconds + + with pytest.raises(ValueError, match="exact positive integer") as caught: + nanoseconds_to_milliseconds(invalid) + assert not isinstance(caught.value, OverflowError) + + +def test_deterministic_work_counts_cover_every_production_validation_boundary() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + + for point in profile.points: + assert point.work.canonical_relations == point.relation_count + assert point.work.validated_relations == point.relation_count + assert point.work.canonical_observations == point.observation_count + assert point.work.validated_observations == point.observation_count + assert point.work.readiness_receipts == point.receipt_count + assert point.work.admission_cases == point.case_count + assert point.work.total_operations == ( + 2 * point.relation_count + 2 * point.observation_count + point.receipt_count + point.case_count + ) + + +@pytest.mark.parametrize( + "field", + ( + "canonical_relations", + "validated_relations", + "canonical_observations", + "validated_observations", + "readiness_receipts", + "admission_cases", + ), +) +def test_deterministic_work_refuses_disagreement_or_a_missing_dimension(field: str) -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + point = profile.points[-1] + bad_work = replace(point.work, **{field: 0}) + changed = replace(point, work=bad_work) + + with pytest.raises(AssertionError, match="incremental validation work"): + assert_large_set_envelope(replace(profile, points=(*profile.points[:-1], changed))) + + +def test_every_sample_binds_exact_scope_and_measured_change_bytes() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + by_size = {point.relation_count: point for point in profile.points} + deterministic = (item for item in profile.consumer_samples if item.mode == "deterministic") + for item in deterministic: + sample = item.measured + point = by_size[item.size] + assert sample.complexity.change_id == point.change_digest == canonical.bytes_digest(point.change_content) + assert sample.complexity.changed_bytes == point.changed_bytes == len(point.change_content) + resolved = profile.resolver.resolve(sample.complexity.measurement) + assert resolved.content == point.changed_bytes_measurement + assert point.change_content == canonical.canonical_bytes(json.loads(point.change_content)) + for sample in profile.samples: + parsed = json.loads(profile.resolver.resolve(sample.complexity.measurement).content) + assert parsed["changed_bytes"] == sample.complexity.changed_bytes + + +def test_budget_samples_are_complete_consumer_runs_with_optional_work() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + + assert {sample.mode for sample in profile.consumer_samples} == {"deterministic", "model_and_proof"} + assert all(sample.total_reply_ms > 0 for sample in profile.consumer_samples) + assert tuple(item.measured for item in profile.consumer_samples) == profile.samples + + +def test_measured_samples_fit_derived_envelope_and_excess_names_gate() -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + for sample in profile.samples: + budget.check_budget(sample.complexity, sample.total_reply_ms, profile.budget, profile.resolver) + + selected = profile.samples[-1].complexity + with pytest.raises(budget.BudgetGateFailure) as caught: + budget.check_budget(selected, profile.budget.budget_ms + 1, profile.budget, profile.resolver) + message = str(caught.value) + assert selected.change_id in message + assert f"{profile.budget.budget_ms + 1}ms>{profile.budget.budget_ms}ms" in message + assert profile.budget.workload_class in message + assert profile.budget.derivation_digest in message + + +def test_optional_model_and_proof_processing_share_total_consumer_budget( + monkeypatch: pytest.MonkeyPatch, +) -> None: + profile = measured_profile(read_workload(PERFORMANCE_FIXTURE)) + case = optional_consumer_case(profile) + accepted = consume_records( + case.request, + (case.value,), + case.context, + case.custody, + 0, + case.resolver, + attachments=(case.attachment,), + proof_trust=case.trust, + ) + assert 0 < accepted.total_reply_ms <= profile.budget.budget_ms + calls, _ = _instrument_optional_path(monkeypatch, profile) + request = make_push_request( + scope=case.request.scope, + workload_class=profile.budget.workload_class, + complexity=case.request.complexity, + budget=profile.budget, + ) + + with pytest.raises(ConsumerContractError, match="measured budget gate failed") as caught: + consume_records( + request, + (case.value,), + case.context, + case.custody, + 0, + case.resolver, + attachments=(case.attachment,), + proof_trust=case.trust, + ) + + assert calls == ["model", "proof"] + assert request.complexity.change_id in str(caught.value) + assert profile.budget.derivation_digest in str(caught.value) + assert replace(request.budget, derivation_digest="").derivation_digest == request.budget.derivation_digest