| """Single-scene success gate + scale-up manifest orchestrator (AC-8).""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| import pathlib |
| from dataclasses import dataclass |
|
|
| log = logging.getLogger(__name__) |
|
|
|
|
| @dataclass(slots=True) |
| class ScaleUpGateConfig: |
| count_acc_min: float = 0.9 |
| delay_mae_ns_max: float = 0.3 |
| peak_db_bias_abs_max: float = 3.0 |
|
|
| @classmethod |
| def from_shared(cls, shared) -> "ScaleUpGateConfig": |
| """Derive from `config.ScaleUpGateConfig` for single-source-of-truth.""" |
| return cls( |
| count_acc_min=shared.count_acc_min, |
| delay_mae_ns_max=shared.delay_mae_ns_max, |
| peak_db_bias_abs_max=shared.peak_db_bias_abs_max, |
| ) |
|
|
|
|
| def check_single_scene_gate(metrics: dict[str, float], cfg: ScaleUpGateConfig | None = None) -> bool: |
| cfg = cfg or ScaleUpGateConfig() |
| ok = ( |
| metrics.get("count_acc", 0.0) >= cfg.count_acc_min |
| and metrics.get("delay_mae_matched_ns", float("inf")) <= cfg.delay_mae_ns_max |
| and abs(metrics.get("peak_db_bias_matched", 1e9)) <= cfg.peak_db_bias_abs_max |
| ) |
| if not ok: |
| log.warning( |
| "single_scene_gate FAIL (count_acc=%.3f delay_mae_ns=%.3f peak_db_bias=%.3f)", |
| metrics.get("count_acc", 0.0), |
| metrics.get("delay_mae_matched_ns", float("nan")), |
| metrics.get("peak_db_bias_matched", float("nan")), |
| ) |
| return bool(ok) |
|
|
|
|
| def orchestrate_scale_up( |
| single_scene_metrics: dict[str, float], |
| branch_root: str | pathlib.Path, |
| cfg: ScaleUpGateConfig | None = None, |
| include_tiny10: bool = True, |
| ) -> list[pathlib.Path]: |
| """Return a list of manifests to launch, or raise if the gate failed. |
| |
| Round 7 fix (Codex R6 finding 2): |
| - The gate decision is always taken from SINGLE-SCENE metrics; tiny10 |
| metrics are out of scope for this helper and must not be passed in. |
| - Post-tiny10 callers pass `include_tiny10=False` so the orchestrator |
| returns only tiny30/50/100 (the scale-up sweep proper). Pre-tiny10 |
| callers keep the default and receive tiny10/30/50/100. |
| |
| Each returned path is checked against `reject_radiomap_artifacts`. |
| """ |
| from ..config import ManifestPaths, reject_radiomap_artifacts |
|
|
| branch_root = pathlib.Path(branch_root) |
| if not check_single_scene_gate(single_scene_metrics, cfg): |
| raise RuntimeError("single-scene gate failed; refusing to launch scale-up") |
| defaults = ManifestPaths() |
| names = [] |
| if include_tiny10: |
| names.append(defaults.tiny10) |
| names.extend([defaults.tiny30, defaults.tiny50, defaults.tiny100]) |
| out: list[pathlib.Path] = [] |
| for rel in names: |
| path = branch_root / rel |
| if path.exists(): |
| reject_radiomap_artifacts(str(path)) |
| out.append(path) |
| return out |
|
|
|
|
| def write_gate_report(path: str | pathlib.Path, metrics: dict[str, float], passed: bool) -> None: |
| report = {"passed": bool(passed), "metrics": dict(metrics)} |
| pathlib.Path(path).write_text(json.dumps(report, indent=2)) |
|
|
|
|
| __all__ = [ |
| "ScaleUpGateConfig", |
| "check_single_scene_gate", |
| "orchestrate_scale_up", |
| "write_gate_report", |
| ] |
|
|
