주기 실회기 검증과 G7 종료계약 보강

scripts/run-g7-external-proof-window.py

@@ -42,6 +42,16 @@ from typing import Any, Callable, Iterable, Sequence
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 SCRIPTS = REPO_ROOT / "scripts"
+API_ROOT = REPO_ROOT / "apps/api"
+if str(API_ROOT) not in sys.path:
+    sys.path.insert(0, str(API_ROOT))
+
+from app.contracts.g7_external_evidence import (  # noqa: E402
+    G7HumanVoiceGainEvidencePack,
+)
+from app.services.g7_voice_gain_evidence import (  # noqa: E402
+    evaluate_human_voice_gain,
+)
 
 SOAK_SCRIPT = SCRIPTS / "soak-public-voice-websocket.py"
 RUNTIME_SCRIPT = SCRIPTS / "capture-g7-runtime-evidence.py"
@@ -63,6 +73,8 @@ MIN_PRODUCTION_SECONDS = (
     MIN_REQUIRED_OVERLAP_SECONDS + CAPTURE_START_SKEW_MARGIN_SECONDS
 )
 RUNTIME_SAMPLE_MARGIN = 1
+MAX_CHILD_INTERVAL_SECONDS = 60.0
+DEFAULT_SAMPLE_INTERVAL_SECONDS = 60.0
 
 
 class WindowError(RuntimeError):
@@ -172,6 +184,23 @@ def sample_plan(duration_seconds: float, interval_seconds: float) -> int:
     return math.ceil(duration_seconds / interval_seconds) + RUNTIME_SAMPLE_MARGIN
 
 
+def validate_human_voice_gain_pack(path: Path) -> None:
+    """52분 캡처를 열기 전에 사람 pack의 production gate를 완전히 계산한다."""
+
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        pack = G7HumanVoiceGainEvidencePack.model_validate(payload)
+        result = evaluate_human_voice_gain(pack)
+    except (OSError, UnicodeError, json.JSONDecodeError):
+        raise WindowError("human_voice_gain_pack_unreadable") from None
+    except Exception:
+        # validation 원문에는 입력값이 포함될 수 있으므로 오류 code만 낸다.
+        raise WindowError("human_voice_gain_pack_invalid") from None
+    if not result.passed:
+        reasons = ",".join(result.failure_reasons)
+        raise WindowError(f"human_voice_gain_pack_failed:{reasons}")
+
+
 def build_soak_leg(args: argparse.Namespace, output: Path) -> Leg:
     argv = [
         sys.executable,
@@ -533,8 +562,16 @@ def build_parser() -> argparse.ArgumentParser:
         default=MIN_PRODUCTION_SECONDS,
         help="실제 캡처는 최소 3120초; 유효 artifact 교집합 기준은 3000초",
     )
-    parser.add_argument("--runtime-interval-seconds", type=float, default=100.0)
-    parser.add_argument("--topology-interval-seconds", type=float, default=100.0)
+    parser.add_argument(
+        "--runtime-interval-seconds",
+        type=float,
+        default=DEFAULT_SAMPLE_INTERVAL_SECONDS,
+    )
+    parser.add_argument(
+        "--topology-interval-seconds",
+        type=float,
+        default=DEFAULT_SAMPLE_INTERVAL_SECONDS,
+    )
     parser.add_argument("--human-voice-gain", type=Path)
     parser.add_argument("--out-dir", type=Path, required=True)
     parser.add_argument(
@@ -550,6 +587,12 @@ def validate(args: argparse.Namespace) -> None:
         raise WindowError("production_window_too_short")
     if not args.rehearse and args.human_voice_gain is None:
         raise WindowError("human_voice_gain_pack_required")
+    for value, code in (
+        (args.runtime_interval_seconds, "runtime_interval_out_of_bounds"),
+        (args.topology_interval_seconds, "topology_interval_out_of_bounds"),
+    ):
+        if not math.isfinite(value) or not 0.05 <= value <= MAX_CHILD_INTERVAL_SECONDS:
+            raise WindowError(code)
     if args.topology_mode not in ("linux-compose", "windows-host"):
         raise WindowError("topology_mode_invalid")
     if args.topology_mode == "linux-compose":
@@ -589,6 +632,9 @@ def main(argv: Iterable[str] | None = None) -> int:
     args = build_parser().parse_args(list(argv) if argv is not None else None)
     try:
         validate(args)
+        if not args.rehearse:
+            assert args.human_voice_gain is not None
+            validate_human_voice_gain_pack(args.human_voice_gain)
         args.out_dir.mkdir(parents=True, exist_ok=True)
         legs = plan_legs(args, args.out_dir)
     except WindowError as exc: