Spaces:
Sleeping
Sleeping
File size: 5,569 Bytes
31226fd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """Engineering-only v0.6 performance and pacing acceptance contract.
The contract deliberately separates reproducible software budgets from
scientific evidence. It validates benchmark/telemetry payloads and never
promotes experimental results.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Mapping, cast
class PerformanceContractError(ValueError):
"""Raised when a performance payload violates the v0.6 contract."""
def _number(value: object, *, default: float) -> float:
if value is None:
return default
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise PerformanceContractError(
f"expected numeric performance value, got {value!r}"
)
return float(value)
def _string_mapping(value: object, *, label: str) -> Mapping[str, object]:
if not isinstance(value, dict):
raise PerformanceContractError(f"{label} must be an object")
mapping = cast(dict[object, object], value)
if any(not isinstance(key, str) for key in mapping):
raise PerformanceContractError(f"{label} keys must be strings")
return cast(Mapping[str, object], mapping)
def load_performance_budgets(path: Path) -> dict[str, object]:
payload: object = json.loads(path.read_text(encoding="utf-8"))
data = dict(_string_mapping(payload, label="performance budget"))
schema_version = data.get("schema_version", 0)
if isinstance(schema_version, bool) or not isinstance(schema_version, int):
raise PerformanceContractError(
"performance budget schema_version must be an integer"
)
if schema_version != 1:
raise PerformanceContractError("unsupported performance budget schema")
return data
def evaluate_scaling_tier(
tier: Mapping[str, object], budgets: Mapping[str, object]
) -> tuple[str, ...]:
"""Return deterministic budget violations for one scaling tier."""
scaling = _string_mapping(budgets.get("scaling", {}), label="scaling budget")
violations: list[str] = []
tick_cost = _number(tier.get("mean_tick_cost_ms"), default=0.0)
peak_per_neuron = _number(tier.get("python_peak_bytes_per_neuron"), default=0.0)
max_tick = _number(scaling.get("max_mean_tick_cost_ms"), default=float("inf"))
max_peak = _number(
scaling.get("max_python_peak_bytes_per_neuron"), default=float("inf")
)
if tick_cost > max_tick:
violations.append(f"mean_tick_cost_ms {tick_cost:.6f} > {max_tick:.6f}")
if peak_per_neuron > max_peak:
violations.append(
f"python_peak_bytes_per_neuron {peak_per_neuron:.3f} > {max_peak:.3f}"
)
return tuple(violations)
def evaluate_runtime_profile(
*,
tick_latency_ms: float,
phases_ms: Mapping[str, float],
budgets: Mapping[str, object],
) -> tuple[str, ...]:
"""Validate RuntimeController and named subsystem phase thresholds."""
runtime = _string_mapping(budgets.get("runtime", {}), label="runtime budget")
phase_budget = _string_mapping(
budgets.get("phases_ms_per_tick", {}), label="phase budget"
)
violations: list[str] = []
max_latency = _number(runtime.get("max_tick_latency_ms"), default=float("inf"))
if tick_latency_ms > max_latency:
violations.append(f"tick_latency_ms {tick_latency_ms:.6f} > {max_latency:.6f}")
required = {
"learning",
"homeostasis",
"structural",
"embodiment",
"neural_symbiosis_msba",
"dashboard_telemetry",
"storage",
}
missing = sorted(required - set(phase_budget.keys()))
if missing:
raise PerformanceContractError(
"missing subsystem phase budgets: " + ", ".join(missing)
)
for phase in sorted(required):
elapsed = phases_ms.get(phase, 0.0)
maximum = _number(phase_budget[phase], default=float("inf"))
if elapsed > maximum:
violations.append(f"{phase} {elapsed:.6f}ms > {maximum:.6f}ms")
return tuple(violations)
def evaluate_pacing(
*,
target_hz: float | None,
achieved_hz: float,
realtime_ratio: float,
dt_seconds: float,
runtime_mode: str,
budgets: Mapping[str, object],
) -> tuple[str, ...]:
"""Apply explicit acceptance criteria to TARGETED and MAX runtime modes."""
pacing = _string_mapping(budgets.get("pacing", {}), label="pacing budget")
violations: list[str] = []
if achieved_hz < 0.0 or realtime_ratio < 0.0 or dt_seconds <= 0.0:
return ("pacing telemetry contains invalid values",)
expected_ratio = achieved_hz * dt_seconds
tolerance = _number(pacing.get("realtime_ratio_absolute_tolerance"), default=1e-9)
if abs(realtime_ratio - expected_ratio) > tolerance:
violations.append("realtime_ratio is inconsistent with achieved_hz * dt")
if target_hz is None:
if runtime_mode != "MAX":
violations.append("unlimited target requires runtime_mode=MAX")
return tuple(violations)
if target_hz <= 0.0:
return (*violations, "target_hz must be positive or null")
minimum_fraction = _number(pacing.get("minimum_target_fraction"), default=0.75)
if achieved_hz < target_hz * minimum_fraction:
violations.append(
f"achieved_hz {achieved_hz:.6f} below "
f"{minimum_fraction:.3f} of target {target_hz:.6f}"
)
if runtime_mode not in {"TARGETED", "COMPUTE LIMITED"}:
violations.append("finite target requires TARGETED or COMPUTE LIMITED mode")
return tuple(violations)
|