"""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", ]