#!/usr/bin/env python3
"""G7 외부 종료 증거 3종을 하나의 겹치는 시간창에서 동시에 잡는 오케스트레이터.

`check-g7-external-proof.py` 는 네 artifact 를 요구하고, 그 중 세 개는 **같은 public host 의
겹치는 시간창**이어야 한다. 지금까지는 운영자가 세 명령을 따로 띄우고 시계를 손으로 맞춰야 했다.
50분짜리 실행에서 그건 한 번 어긋나면 처음부터 다시 해야 하는 종류의 취약점이다.

이 스크립트가 하는 일:

1. soak / runtime / topology 캡처를 **동시에** 시작하고 전부 끝날 때까지 기다린다.
2. WSS / admin / topology transport 가 같은 public API host 를 보는지 시작 전에
   확인한다(fail-closed). 브라우저 ``Origin`` 은 API host 와 섞지 않고 별도
   allowlist 로 검증한다.
3. 끝나면 human voice-gain pack 을 더해 `check-g7-external-proof.py` 를 그대로 돌린다.

이 스크립트는 게이트를 **약화시키지 않는다**. 물리 마이크 캡처는 `--confirm-physical-capture` 와
`--microphone-device` 가 둘 다 있어야 시작하고, human pack 은 만들어 주지 않는다. 네 번째 artifact 는
실제 참가자와 독립 평가자가 있어야 한다.

`--rehearse` 는 마이크·사람 없이 배관만 확인하는 모드다. 짧은 시간창으로 세 러너를 실제로 띄워
인자·경로·산출 파일까지 검증하되, soak 은 `--preflight-only` 로 돌려 장치를 열지 않는다. 실제 실행은
52분을 캡처해 시작 skew 를 제외한 유효 교집합 50분을 증명한다. 그 전에 이걸로 먼저 실패를 뽑아내라.

공개 기본 계약은 브라우저 ``Origin=https://vignette.chanpaca.net`` 과 transport
host ``api-vignette.chanpaca.net`` 이 의도적으로 다른 구성이다. 인증 WSS 의
``ready`` 는 기본적으로 ``local_whisper/small`` 과 ``melotts/melotts-korean`` 을
정확히 반환해야 한다.
"""

from __future__ import annotations

import argparse
import json
import math
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
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"
TOPOLOGY_SCRIPT = SCRIPTS / "capture-g7-topology-evidence.py"
CHECKER_SCRIPT = SCRIPTS / "check-g7-external-proof.py"

DEFAULT_WSS_URL = "wss://api-vignette.chanpaca.net/voice/ws"
DEFAULT_BROWSER_ORIGIN = "https://vignette.chanpaca.net"
DEFAULT_ADMIN_RUNTIME_URL = (
    "https://api-vignette.chanpaca.net/admin/voice-runtime"
)
DEFAULT_ALLOWED_BROWSER_ORIGINS = (DEFAULT_BROWSER_ORIGIN,)

# canonical checker 가 요구하는 실제 artifact 교집합은 3,000초다. 세 캡처의
# 시작 skew 와 마지막 sampler tick 때문에 실제 캡처는 최소 120초 더 길게 잡는다.
MIN_REQUIRED_OVERLAP_SECONDS = 3_000.0
CAPTURE_START_SKEW_MARGIN_SECONDS = 120.0
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):
    """Fail-closed 오케스트레이션 오류."""


@dataclass(frozen=True)
class Leg:
    name: str
    argv: list[str]
    output: Path


@dataclass(frozen=True)
class LegResult:
    name: str
    returncode: int
    output: Path

    @property
    def ok(self) -> bool:
        return self.returncode == 0


def host_of(url: str) -> str:
    from urllib.parse import urlsplit

    return (urlsplit(url).hostname or "").lower()


def assert_single_host(*urls: str) -> str:
    """transport URL 들이 같은 public host 를 보는지 확인한다."""

    hosts = {host_of(url) for url in urls if url}
    hosts.discard("")
    if len(hosts) != 1:
        # URL, query, tenant 이름을 오류에 반사하지 않는다.
        raise WindowError("transport_hosts_must_match")
    return hosts.pop()


def assert_transport_schemes(wss_url: str, admin_runtime_url: str) -> None:
    """공개 proof transport 를 암호화된 WSS/HTTPS 로만 고정한다."""

    from urllib.parse import urlsplit

    try:
        wss = urlsplit(wss_url)
        admin = urlsplit(admin_runtime_url)
    except ValueError as exc:
        raise WindowError("transport_url_invalid") from exc
    if wss.scheme.lower() != "wss" or not wss.hostname:
        raise WindowError("wss_url_scheme_invalid")
    if admin.scheme.lower() != "https" or not admin.hostname:
        raise WindowError("admin_runtime_url_scheme_invalid")


def normalize_browser_origin(value: str) -> str:
    """브라우저 Origin 직렬화만 허용하고 안전하게 정규화한다."""

    from urllib.parse import urlsplit

    try:
        parsed = urlsplit(value)
        port = parsed.port
    except ValueError as exc:
        raise WindowError("browser_origin_invalid") from exc
    if (
        parsed.scheme.lower() != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in ("", "/")
        or parsed.query
        or parsed.fragment
    ):
        raise WindowError("browser_origin_invalid")
    host = parsed.hostname.lower()
    authority = host if port in (None, 443) else f"{host}:{port}"
    return f"https://{authority}"


def assert_browser_origin_allowed(
    origin: str, allowed_origins: Sequence[str] | None
) -> str:
    """실제 브라우저 Origin 을 transport host 와 독립된 allowlist 로 검증한다."""

    # CLI 옵션은 기본 공개 frontend 를 대체하지 않고 추가한다.
    configured = (*DEFAULT_ALLOWED_BROWSER_ORIGINS, *(allowed_origins or ()))
    try:
        normalized_allowed = {normalize_browser_origin(item) for item in configured}
    except WindowError as exc:
        raise WindowError("browser_origin_allowlist_invalid") from exc
    normalized = normalize_browser_origin(origin)
    if normalized not in normalized_allowed:
        # 입력 Origin 을 반사하면 query/PII 가 로그에 남을 수 있으므로 코드만 낸다.
        raise WindowError("browser_origin_not_allowed")
    return normalized


def sample_plan(duration_seconds: float, interval_seconds: float) -> int:
    """시간창을 덮기에 충분한 샘플 수. 모자라면 겹침 검증이 깨진다."""

    if duration_seconds <= 0 or interval_seconds <= 0:
        raise WindowError("invalid_sampling_plan")
    # t=0 샘플과 duration 끝을 덮는 종료 샘플을 모두 포함한다.
    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,
        "-X",
        "utf8",
        str(SOAK_SCRIPT),
        "--wss-url",
        args.wss_url,
        "--origin",
        args.origin,
        "--expected-stt-provider",
        args.expected_stt_provider,
        "--expected-stt-model",
        args.expected_stt_model,
        "--expected-tts-provider",
        args.expected_tts_provider,
        "--expected-tts-model",
        args.expected_tts_model,
        "--evidence-output",
        str(output),
    ]
    if args.rehearse:
        # 배관 확인 전용. 장치를 열지 않는다.
        argv.append("--preflight-only")
        return Leg("voice_soak_preflight", argv, output)
    if not args.microphone_device or not args.confirm_physical_capture:
        raise WindowError("physical_microphone_consent_required")
    argv += [
        "--microphone-device",
        args.microphone_device,
        "--confirm-physical-capture",
        "--duration-seconds",
        str(args.duration_seconds),
    ]
    return Leg("voice_soak", argv, output)


def build_runtime_leg(args: argparse.Namespace, output: Path) -> Leg:
    samples = sample_plan(args.duration_seconds, args.runtime_interval_seconds)
    argv = [
        sys.executable,
        "-X",
        "utf8",
        str(RUNTIME_SCRIPT),
        "--url",
        args.admin_runtime_url,
        "--samples",
        str(samples),
        "--interval-seconds",
        str(args.runtime_interval_seconds),
        "--evidence-output",
        str(output),
    ]
    return Leg("runtime", argv, output)


def build_topology_leg(args: argparse.Namespace, output: Path) -> Leg:
    samples = sample_plan(args.duration_seconds, args.topology_interval_seconds)
    argv = [
        sys.executable,
        "-X",
        "utf8",
        str(TOPOLOGY_SCRIPT),
        "--topology-mode",
        args.topology_mode,
        "--public-host",
        host_of(args.wss_url),
        "--samples",
        str(samples),
        "--interval-seconds",
        str(args.topology_interval_seconds),
        "--evidence-output",
        str(output),
    ]
    if args.topology_mode == "linux-compose":
        argv += [
            "--compose-project",
            args.compose_project,
            "--api-container",
            args.api_container,
            "--api-image-digest",
            args.api_image_digest,
            "--caddy-container",
            args.caddy_container,
            "--caddy-image-digest",
            args.caddy_image_digest,
        ]
    else:
        argv += [
            "--repo-root",
            args.repo_root,
            "--git-sha",
            args.git_sha,
            "--git-tree-sha",
            args.git_tree_sha,
            "--runner-script-sha256",
            args.runner_script_sha256,
            "--collector-script-sha256",
            args.collector_script_sha256,
            "--checker-script-sha256",
            args.checker_script_sha256,
            "--psutil-version",
            args.psutil_version,
            "--api-pid",
            str(args.api_pid),
            "--api-executable-name",
            args.api_executable_name,
            "--api-executable-sha256",
            args.api_executable_sha256,
            "--api-cwd",
            args.api_cwd,
            "--api-listen-port",
            str(args.api_listen_port),
            "--cloudflared-pid",
            str(args.cloudflared_pid),
            "--cloudflared-executable-name",
            args.cloudflared_executable_name,
            "--cloudflared-executable-sha256",
            args.cloudflared_executable_sha256,
            "--cloudflared-cwd",
            args.cloudflared_cwd,
        ]
    return Leg("topology", argv, output)


ADMIN_RUNTIME_PATH = "/admin/voice-runtime"
# Cloudflare 앞단은 자동화 클라이언트에 error code 1010 으로 **모든 경로**에 403 을 준다.
# 그래서 미인증 상태 코드로는 "endpoint 없음"과 "edge 차단"을 구분할 수 없다.
# 배포 여부는 OpenAPI spec 의 paths 로만 판정한다.
_BROWSER_UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/140.0 Safari/537.36"
)


def assert_admin_endpoint_deployed(
    url: str, *, paths_probe: Callable[[str], Sequence[str]] | None = None
) -> str:
    """관리자 runtime endpoint 가 그 배포에 실제로 존재하는지 먼저 본다.

    2026-08-08 확인: 공개 런타임은 119 paths 구버전이라 `/admin/voice-runtime` 이
    아예 없었다. 이 상태로 50분 창을 돌리면 마이크와 사람을 다 준비하고도 artifact 2
    에서 실패한다. 그래서 시작 전에 막는다.

    미인증 HTTP 상태로 판정하지 않는다. Cloudflare 가 자동화 클라이언트에 모든 경로를
    403(error code 1010)으로 돌려주므로 그 신호로는 배포 누락을 알 수 없다. OpenAPI
    spec 의 `paths` 만 authoritative 하게 쓴다.
    """

    from urllib.parse import urlsplit

    parsed = urlsplit(url)
    if parsed.path.rstrip("/") != ADMIN_RUNTIME_PATH:
        raise WindowError("admin_runtime_url_unexpected_path")
    origin = f"{parsed.scheme}://{parsed.netloc}"
    paths = (paths_probe or _openapi_paths)(origin)
    if ADMIN_RUNTIME_PATH not in set(paths):
        raise WindowError("admin_voice_runtime_not_deployed")
    return origin


def _openapi_paths(origin: str) -> Sequence[str]:  # pragma: no cover - 네트워크 I/O
    import urllib.error
    import urllib.request

    request = urllib.request.Request(
        f"{origin}/openapi.json",
        headers={"User-Agent": _BROWSER_UA, "Accept": "application/json"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            document = json.loads(response.read(8 * 1024 * 1024))
    except (urllib.error.URLError, OSError, ValueError) as exc:
        raise WindowError("openapi_unreachable") from exc
    paths = document.get("paths") if isinstance(document, dict) else None
    if not isinstance(paths, dict):
        raise WindowError("openapi_shape_invalid")
    return list(paths)


def plan_legs(
    args: argparse.Namespace,
    out_dir: Path,
    *,
    admin_probe: Callable[[str], Sequence[str]] | None = None,
) -> list[Leg]:
    assert_transport_schemes(args.wss_url, args.admin_runtime_url)
    assert_single_host(args.wss_url, args.admin_runtime_url)
    assert_browser_origin_allowed(args.origin, args.allowed_browser_origins)
    assert_admin_endpoint_deployed(args.admin_runtime_url, paths_probe=admin_probe)
    return [
        build_soak_leg(args, out_dir / "voice-soak.json"),
        build_runtime_leg(args, out_dir / "runtime.json"),
        build_topology_leg(args, out_dir / "topology.json"),
    ]


def _run(leg: Leg, *, timeout: float, runner: Callable[..., Any]) -> LegResult:
    completed = runner(
        leg.argv,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=timeout,
        shell=False,
    )
    return LegResult(leg.name, int(completed.returncode), leg.output)


def run_window(
    legs: Sequence[Leg],
    *,
    timeout: float,
    runner: Callable[..., Any] | None = None,
) -> list[LegResult]:
    """세 캡처를 동시에 시작한다. 하나가 죽어도 나머지를 끝까지 돌린다."""

    call = runner or subprocess.run
    with ThreadPoolExecutor(max_workers=len(legs)) as pool:
        futures = [
            pool.submit(_run, leg, timeout=timeout, runner=call) for leg in legs
        ]
        return [future.result() for future in futures]


def build_checker_argv(
    results: Sequence[LegResult], human_pack: Path
) -> list[str]:
    by_name = {result.name: result.output for result in results}
    soak = by_name.get("voice_soak") or by_name.get("voice_soak_preflight")
    if soak is None:
        raise WindowError("soak_evidence_missing")
    return [
        sys.executable,
        "-X",
        "utf8",
        str(CHECKER_SCRIPT),
        "--voice-soak",
        str(soak),
        "--runtime",
        str(by_name["runtime"]),
        "--topology",
        str(by_name["topology"]),
        "--human-voice-gain",
        str(human_pack),
        "--json",
    ]


def summarize(
    results: Sequence[LegResult],
    *,
    rehearse: bool,
    checker_returncode: int | None,
) -> dict[str, Any]:
    all_legs_passed = all(r.ok for r in results)
    gate_closed = (
        all_legs_passed and checker_returncode == 0 and not rehearse
    )
    return {
        "schema_version": "vignette.g7-external-proof-window.v1",
        "mode": "rehearse" if rehearse else "production",
        "legs": [
            {"name": r.name, "returncode": r.returncode, "evidence": str(r.output)}
            for r in results
        ],
        "all_legs_passed": all_legs_passed,
        "checker_returncode": checker_returncode,
        "gate_closed": gate_closed,
        "note": (
            "rehearse 는 배관만 확인한다. 게이트는 실제 물리 마이크 50분 실행과 "
            "독립 human voice-gain pack 이 있어야 닫힌다."
            if rehearse
            else "네 artifact 가 모두 통과해야 게이트가 닫힌다."
        ),
    }


def process_exit_code(report: dict[str, Any]) -> int:
    """프로세스 성공을 실제 gate closure 에 fail-closed 결속한다."""

    if report.get("mode") == "production":
        return 0 if (
            report.get("checker_returncode") == 0
            and report.get("gate_closed") is True
        ) else 1
    return 0 if report.get("all_legs_passed") is True else 1


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "공개 기본값:\n"
            "  browser Origin  https://vignette.chanpaca.net\n"
            "  WSS transport   wss://api-vignette.chanpaca.net/voice/ws\n"
            "  admin transport https://api-vignette.chanpaca.net/admin/voice-runtime\n"
            "인증 WSS ready 기대값:\n"
            "  stt_provider=local_whisper stt_model=small\n"
            "  tts_provider=melotts tts_model=melotts-korean\n"
            "실제 캡처 기본값: 3120초(유효 교집합 최소 3000초 + skew margin 120초)\n"
            "추가 frontend Origin 은 --allowed-browser-origin 을 반복 지정한다."
        ),
    )
    parser.add_argument("--wss-url", default=DEFAULT_WSS_URL)
    parser.add_argument(
        "--origin",
        default=DEFAULT_BROWSER_ORIGIN,
        help="WebSocket Origin 헤더에 넣을 실제 브라우저 HTTPS origin",
    )
    parser.add_argument(
        "--allowed-browser-origin",
        dest="allowed_browser_origins",
        action="append",
        help=(
            "허용할 브라우저 HTTPS origin. 반복 가능하며 생략 시 공개 frontend만 허용"
        ),
    )
    parser.add_argument("--admin-runtime-url", default=DEFAULT_ADMIN_RUNTIME_URL)
    parser.add_argument(
        "--topology-mode",
        choices=("linux-compose", "windows-host"),
        default="linux-compose",
    )
    parser.add_argument("--compose-project")
    parser.add_argument("--api-container")
    parser.add_argument("--api-image-digest")
    parser.add_argument("--caddy-container")
    parser.add_argument("--caddy-image-digest")
    parser.add_argument("--repo-root")
    parser.add_argument("--git-sha")
    parser.add_argument("--git-tree-sha")
    parser.add_argument("--runner-script-sha256")
    parser.add_argument("--collector-script-sha256")
    parser.add_argument("--checker-script-sha256")
    parser.add_argument("--psutil-version")
    parser.add_argument("--api-pid", type=int)
    parser.add_argument("--api-executable-name")
    parser.add_argument("--api-executable-sha256")
    parser.add_argument("--api-cwd")
    parser.add_argument("--api-listen-port", type=int)
    parser.add_argument("--cloudflared-pid", type=int)
    parser.add_argument("--cloudflared-executable-name")
    parser.add_argument("--cloudflared-executable-sha256")
    parser.add_argument("--cloudflared-cwd")
    parser.add_argument("--expected-stt-provider", default="local_whisper")
    parser.add_argument("--expected-stt-model", default="small")
    parser.add_argument("--expected-tts-provider", default="melotts")
    parser.add_argument("--expected-tts-model", default="melotts-korean")
    parser.add_argument("--microphone-device", default="")
    parser.add_argument("--confirm-physical-capture", action="store_true")
    parser.add_argument(
        "--duration-seconds",
        type=float,
        default=MIN_PRODUCTION_SECONDS,
        help="실제 캡처는 최소 3120초; 유효 artifact 교집합 기준은 3000초",
    )
    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(
        "--rehearse",
        action="store_true",
        help="마이크 없이 배관만 확인한다. 게이트를 닫지 않는다.",
    )
    return parser


def validate(args: argparse.Namespace) -> None:
    if not args.rehearse and args.duration_seconds < MIN_PRODUCTION_SECONDS:
        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":
        required = (
            "compose_project",
            "api_container",
            "api_image_digest",
            "caddy_container",
            "caddy_image_digest",
        )
    else:
        required = (
            "repo_root",
            "git_sha",
            "git_tree_sha",
            "runner_script_sha256",
            "collector_script_sha256",
            "checker_script_sha256",
            "psutil_version",
            "api_pid",
            "api_executable_name",
            "api_executable_sha256",
            "api_cwd",
            "api_listen_port",
            "cloudflared_pid",
            "cloudflared_executable_name",
            "cloudflared_executable_sha256",
            "cloudflared_cwd",
        )
    for field in required:
        value = getattr(args, field)
        if value is None or (isinstance(value, str) and not value.strip()):
            raise WindowError(f"topology_argument_required:{field.replace('_', '-')}")


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:
        print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False), file=sys.stderr)
        return 2

    results = run_window(legs, timeout=args.duration_seconds + 900)

    checker_returncode: int | None = None
    if all(result.ok for result in results) and args.human_voice_gain is not None:
        checker = subprocess.run(
            build_checker_argv(results, args.human_voice_gain),
            check=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            shell=False,
        )
        checker_returncode = checker.returncode

    report = summarize(
        results, rehearse=args.rehearse, checker_returncode=checker_returncode
    )
    print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
    return process_exit_code(report)


if __name__ == "__main__":  # pragma: no cover - CLI
    raise SystemExit(main())