"""Exercise the G7 multimodal-alliance lifecycle over live HTTP.

The smoke uses a synthetic WAV only to derive deterministic metadata. It never
uploads or prints transcript/audio content. Durable fixtures are intentionally
left in the development database as auditable integration evidence.
"""

from __future__ import annotations

import argparse
import copy
import hashlib
import io
import json
import math
import secrets
import struct
import time
import urllib.error
import urllib.request
import wave
from dataclasses import dataclass
from datetime import UTC, datetime
from http.cookiejar import CookieJar
from pathlib import Path
from typing import Any
from uuid import uuid4


COHORT_ID = "e2e-hanshin"
OTHER_COHORT_ID = "e2e-other-cohort"
INTERNAL_HEADER = "X-Vignette-Multimodal-Token"


class SmokeError(RuntimeError):
    pass


@dataclass(frozen=True)
class ApiResponse:
    status: int
    body: Any


class ApiClient:
    def __init__(self, base_url: str, timeout: float) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self._opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(CookieJar())
        )

    def request(
        self,
        method: str,
        path: str,
        payload: dict[str, Any] | None = None,
        *,
        expected: set[int] | None = None,
        headers: dict[str, str] | None = None,
    ) -> ApiResponse:
        data = None
        request_headers = {"Accept": "application/json", **(headers or {})}
        if payload is not None:
            data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
            request_headers["Content-Type"] = "application/json"
        request = urllib.request.Request(
            f"{self.base_url}{path}",
            data=data,
            headers=request_headers,
            method=method,
        )
        try:
            with self._opener.open(request, timeout=self.timeout) as response:
                raw = response.read().decode("utf-8")
                result = ApiResponse(response.status, json.loads(raw) if raw else {})
        except urllib.error.HTTPError as exc:
            raw = exc.read().decode("utf-8", errors="replace")
            try:
                body = json.loads(raw) if raw else {}
            except json.JSONDecodeError:
                body = {"detail": raw[:500]}
            result = ApiResponse(exc.code, body)
        except urllib.error.URLError as exc:
            raise SmokeError(
                f"{method} {path} transport failed: {type(exc.reason).__name__}"
            ) from exc
        if result.status not in (expected or {200}):
            detail = result.body.get("detail") if isinstance(result.body, dict) else None
            raise SmokeError(
                f"{method} {path} returned HTTP {result.status}; detail={detail!r}"
            )
        return result


def _sign_in(
    client: ApiClient,
    *,
    suffix: str,
    identity: str,
    role: str,
    cohort_ids: list[str],
) -> str:
    login = client.request(
        "POST",
        "/auth/dev-login",
        {
            "email": f"dev.e2e.multimodal.{identity}.{suffix}@hs.ac.kr",
            "role": role,
            "display_name": f"Multimodal {identity.title()}",
            "cohort_ids": cohort_ids,
        },
    )
    client.request(
        "POST",
        "/users/me/onboarding",
        {
            "legal_name": f"Multimodal {identity.title()}",
            "affiliation": "한신대학교",
            "department": "상담심리학과",
            "grade_level": "통합검증",
            "phone": "010-0000-0000",
            "contact_address": "경기도 오산시 한신대학교",
            "nickname": f"Multimodal {identity.title()}",
            "self_introduction": "G7 멀티모달 동맹 API 검증 fixture입니다.",
            "avatar_url": "",
            "terms_accepted": True,
            "privacy_accepted": True,
        },
    )
    user_id = str(login.body.get("user_id") or "")
    if not user_id:
        raise SmokeError(f"dev-login omitted user_id for {identity}")
    return user_id


def _choose_persona(client: ApiClient) -> str:
    response = client.request("GET", "/personas")
    usable = [
        item
        for item in response.body
        if isinstance(item, dict)
        and item.get("source") == "database"
        and not item.get("degraded")
        and item.get("code")
    ]
    if not usable:
        raise SmokeError("persona catalog has no non-degraded database persona")
    preferred = next((item for item in usable if item.get("code") == "P1"), usable[0])
    return str(preferred["code"])


def _synthetic_wav() -> bytes:
    sample_rate = 16_000
    frames = sample_rate * 1200 // 1000
    output = io.BytesIO()
    with wave.open(output, "wb") as writer:
        writer.setnchannels(1)
        writer.setsampwidth(2)
        writer.setframerate(sample_rate)
        payload = bytearray()
        for index in range(frames):
            value = int(3200 * math.sin(2 * math.pi * 220 * index / sample_rate))
            payload.extend(struct.pack("<h", value))
        writer.writeframes(bytes(payload))
    return output.getvalue()


def _timeline(live_suffix: str) -> dict[str, Any]:
    return {
        "audio_duration_ms": 1200,
        "words": [
            {
                "word_index": 0,
                "start_ms": 40,
                "end_ms": 220,
                "speaker": "learner",
                "token_hash": "a" * 64,
            },
            {
                "word_index": 1,
                "start_ms": 500,
                "end_ms": 710,
                "speaker": "client",
                "token_hash": "b" * 64,
            },
        ],
        "events": [
            {
                "event_id": f"oas-g7-event-{live_suffix}-silence",
                "event_type": "silence",
                "start_ms": 220,
                "end_ms": 500,
                "actor": "both",
                "observed_feature": "280ms turn transition silence",
                "uncertainty": 0.05,
                "source": "stt_word_timestamps",
            },
            {
                "event_id": f"oas-g7-event-{live_suffix}-overlap",
                "event_type": "overlap",
                "start_ms": 710,
                "end_ms": 770,
                "actor": "both",
                "observed_feature": "60ms simultaneous speech segment",
                "uncertainty": 0.1,
                "source": "observed_audio_runtime",
            },
            {
                "event_id": f"oas-g7-event-{live_suffix}-interruption",
                "event_type": "interruption",
                "start_ms": 770,
                "end_ms": 840,
                "actor": "learner",
                "observed_feature": "learner segment began before client segment ended",
                "uncertainty": 0.12,
                "source": "observed_audio_runtime",
            },
            {
                "event_id": f"oas-g7-event-{live_suffix}-prosody",
                "event_type": "prosody",
                "start_ms": 840,
                "end_ms": 1080,
                "actor": "learner",
                "observed_feature": "median intensity decreased by 3dB",
                "uncertainty": 0.2,
                "source": "observed_audio_runtime",
            },
        ],
    }


def _stable_retry(
    client: ApiClient,
    path: str,
    body: dict[str, Any],
    *,
    id_field: str,
    headers: dict[str, str] | None = None,
) -> dict[str, Any]:
    created = client.request("POST", path, body, expected={201}, headers=headers)
    retried = client.request("POST", path, body, expected={201}, headers=headers)
    if created.body.get(id_field) != retried.body.get(id_field):
        raise SmokeError(f"same submission changed {id_field} for {path}")
    if retried.body.get("idempotent_replay") is not True:
        raise SmokeError(f"same submission retry was not stable for {path}")
    return created.body


def _assert_safe_metadata(value: Any, path: str = "response") -> None:
    if isinstance(value, dict):
        forbidden = {
            "raw_transcript",
            "transcript",
            "utterance_text",
            "diagnosis",
            "emotion_label",
            "total_score",
        } & set(value)
        if forbidden:
            raise SmokeError(f"{path} exposed forbidden keys: {sorted(forbidden)}")
        if value.get("clinical_claim_allowed") is True:
            raise SmokeError(f"{path} enabled a clinical claim")
        for key, child in value.items():
            _assert_safe_metadata(child, f"{path}.{key}")
    elif isinstance(value, list):
        for index, child in enumerate(value):
            _assert_safe_metadata(child, f"{path}[{index}]")


def _prepare_case(case: dict[str, Any], suffix: str) -> dict[str, Any]:
    result = copy.deepcopy(case)
    for key in ("text_measurement", "voice_measurement"):
        measurement = result[key]
        measurement["measurement_id"] = f"{measurement['measurement_id']}-{suffix}"
        if measurement["status"] == "ready":
            measurement["model_run_id"] = str(uuid4())
    result["calibration"]["calibration_id"] = (
        f"{result['calibration']['calibration_id']}-{suffix}"
    )
    return result


def run(args: argparse.Namespace) -> dict[str, Any]:
    if len(args.internal_token) < 32:
        raise SmokeError("--internal-token must contain at least 32 characters")
    root = ApiClient(args.api_base_url, args.request_timeout)
    health = root.request("GET", "/health")
    if not health.body.get("db") or not health.body.get("engine"):
        raise SmokeError("API health is not DB+engine ready")

    internal = ApiClient(args.api_base_url, args.request_timeout)
    internal.request(
        "POST",
        "/internal/sessions/00000000-0000-4000-8000-000000000000/multimodal-alliance/timelines",
        {},
        expected={401},
    )
    internal.request(
        "POST",
        "/internal/sessions/00000000-0000-4000-8000-000000000000/multimodal-alliance/timelines",
        {},
        expected={403},
        headers={INTERNAL_HEADER: "wrong-token"},
    )
    token_header = {INTERNAL_HEADER: args.internal_token}

    login_suffix = f"{int(time.time())}.{secrets.token_hex(3)}"
    live_suffix = secrets.token_hex(5)
    learner = ApiClient(args.api_base_url, args.request_timeout)
    teacher = ApiClient(args.api_base_url, args.request_timeout)
    outsider = ApiClient(args.api_base_url, args.request_timeout)
    learner_id = _sign_in(
        learner,
        suffix=login_suffix,
        identity="learner",
        role="learner",
        cohort_ids=[COHORT_ID],
    )
    _sign_in(
        teacher,
        suffix=login_suffix,
        identity="teacher",
        role="teacher",
        cohort_ids=[COHORT_ID],
    )
    _sign_in(
        outsider,
        suffix=login_suffix,
        identity="outsider",
        role="teacher",
        cohort_ids=[OTHER_COHORT_ID],
    )

    started = learner.request(
        "POST",
        "/sessions",
        {
            "persona_code": _choose_persona(learner),
            "theory_mode": "humanistic",
            "goal_stages": ["라포", "탐색"],
        },
        expected={201},
    )
    session_id = str(started.body.get("session_id") or "")
    if not session_id or started.body.get("degraded"):
        raise SmokeError("session start did not produce a durable non-degraded session")

    consent_path = f"/sessions/{session_id}/multimodal-alliance/consent"
    consent_body = {
        "submission_id": str(uuid4()),
        "consent_status": "granted",
        "retain_audio": True,
        "retain_derived_features": True,
        "transcript_retained": True,
        "retention_days": 30,
        "policy_version": "g7-live-v1",
        "reason_code": None,
    }
    consent = _stable_retry(
        learner,
        consent_path,
        consent_body,
        id_field="consent_snapshot_id",
    )
    changed_consent = copy.deepcopy(consent_body)
    changed_consent["retention_days"] = 31
    learner.request("POST", consent_path, changed_consent, expected={409})

    synthetic_audio = _synthetic_wav()
    timeline_body = {
        "submission_id": str(uuid4()),
        "timeline": _timeline(live_suffix),
        "audio_asset": {
            "audio_ref": f"g7-live://synthetic/{live_suffix}.wav",
            "audio_sha256": hashlib.sha256(synthetic_audio).hexdigest(),
            "media_type": "audio/wav",
            "byte_size": len(synthetic_audio),
        },
    }
    timeline_path = f"/internal/sessions/{session_id}/multimodal-alliance/timelines"
    timeline = _stable_retry(
        internal,
        timeline_path,
        timeline_body,
        id_field="timeline_id",
        headers=token_header,
    )
    changed_timeline = copy.deepcopy(timeline_body)
    changed_timeline["timeline"]["audio_duration_ms"] = 1300
    internal.request(
        "POST", timeline_path, changed_timeline, expected={409}, headers=token_header
    )

    pack = json.loads(Path(args.benchmark_path).read_text(encoding="utf-8"))
    measurement_path = (
        f"/internal/sessions/{session_id}/multimodal-alliance/measurements"
    )
    fusion_results: list[dict[str, Any]] = []
    first_body: dict[str, Any] | None = None
    provenance = {
        "instrument_id": "alliance-axis-observer",
        "instrument_version": "g7-live-v1",
        "model_name": "synthetic-observer",
        "prompt_version": "g7-live-v1",
    }
    for index, source_case in enumerate(pack["cases"]):
        case = _prepare_case(source_case, live_suffix)
        body = {
            "submission_id": str(uuid4()),
            "text_measurement": case["text_measurement"],
            "text_provenance": provenance,
            "voice_measurement": case["voice_measurement"],
            "voice_provenance": provenance,
            "calibration": case["calibration"],
        }
        if index == 0:
            result = _stable_retry(
                internal,
                measurement_path,
                body,
                id_field="fusion_record_id",
                headers=token_header,
            )
            first_body = body
        else:
            result = internal.request(
                "POST", measurement_path, body, expected={201}, headers=token_header
            ).body
        applied = bool(result["result"]["fusion_applied"])
        if applied is not bool(case["expected_fusion_applied"]):
            raise SmokeError(f"fusion gate mismatch for {case['case_id']}")
        fusion_results.append(result)
    if first_body is None:
        raise SmokeError("benchmark pack did not yield a measurement case")
    changed_measurement = copy.deepcopy(first_body)
    changed_measurement["text_measurement"]["value"] = 0.63
    internal.request(
        "POST",
        measurement_path,
        changed_measurement,
        expected={409},
        headers=token_header,
    )

    metadata_path = f"/sessions/{session_id}/multimodal-alliance"
    learner_view = learner.request("GET", metadata_path).body
    teacher_view = teacher.request("GET", metadata_path).body
    outsider.request("GET", metadata_path, expected={404})
    teacher.request(
        "GET", f"{metadata_path}/raw-audio", expected={403}
    )
    raw_before = learner.request("GET", f"{metadata_path}/raw-audio").body
    _assert_safe_metadata(learner_view)
    _assert_safe_metadata(teacher_view)
    if len(learner_view.get("word_timestamps") or []) != 2:
        raise SmokeError("single audio clock did not retain two word timestamps")
    if len(learner_view.get("voice_events") or []) != 4:
        raise SmokeError("single audio clock did not retain four interaction events")
    if len(learner_view.get("measurements") or []) != 6:
        raise SmokeError("text and voice measurements were not independently retained")
    if len(learner_view.get("fusion_decisions") or []) != 3:
        raise SmokeError("all three benchmark fusion decisions were not retained")
    if len(raw_before.get("items") or []) != 1:
        raise SmokeError("learner raw-audio access did not expose one retained asset")

    deletion_body = {
        "submission_id": str(uuid4()),
        "scopes": ["audio", "derived_features"],
    }
    deletion_path = f"{metadata_path}/deletion-requests"
    explicit_deletion = _stable_retry(
        learner,
        deletion_path,
        deletion_body,
        id_field="deletion_request_id",
    )

    withdrawal_body = {
        "submission_id": str(uuid4()),
        "policy_version": "g7-live-v1",
        "reason_code": "learner_withdrawal",
        "transcript_retained": True,
    }
    withdrawal = _stable_retry(
        learner,
        f"/sessions/{session_id}/multimodal-alliance/withdraw",
        withdrawal_body,
        id_field="consent_snapshot_id",
    )
    internal.request(
        "POST",
        timeline_path,
        {
            **timeline_body,
            "submission_id": str(uuid4()),
        },
        expected={409},
        headers=token_header,
    )

    deleted_at = datetime.now(UTC).isoformat().replace("+00:00", "Z")
    tombstones = [
        {
            "scope": "audio",
            "target_ref_hash": hashlib.sha256(
                timeline_body["audio_asset"]["audio_ref"].encode("utf-8")
            ).hexdigest(),
            "deletion_proof": "synthetic object-store delete acknowledged",
            "deleted_at": deleted_at,
        },
        {
            "scope": "derived_features",
            "target_ref_hash": hashlib.sha256(
                str(timeline["timeline_id"]).encode("utf-8")
            ).hexdigest(),
            "deletion_proof": "derived feature access tombstoned",
            "deleted_at": deleted_at,
        },
    ]
    completion_ids: list[str] = []
    for deletion_request_id in (
        explicit_deletion["deletion_request_id"],
        withdrawal["deletion_request_id"],
    ):
        completion = _stable_retry(
            internal,
            f"/internal/multimodal-alliance/deletion-requests/{deletion_request_id}/complete",
            {
                "submission_id": str(uuid4()),
                "actor_uid": None,
                "actor_kind": "retention_worker",
                "tombstones": tombstones,
            },
            id_field="deletion_request_id",
            headers=token_header,
        )
        completion_ids.extend(str(item) for item in completion["tombstone_ids"])

    after = learner.request("GET", metadata_path).body
    raw_after = learner.request("GET", f"{metadata_path}/raw-audio").body
    if raw_after.get("items"):
        raise SmokeError("audio tombstone did not remove raw-audio access")
    if after.get("word_timestamps") or after.get("voice_events"):
        raise SmokeError("derived-feature tombstone did not hide voice timeline detail")
    if any(item.get("modality") != "text" for item in after.get("measurements") or []):
        raise SmokeError("deleted derived features leaked voice measurements")
    if any(
        "voice" in (item.get("modalities_used") or [])
        for item in after.get("fusion_decisions") or []
    ):
        raise SmokeError("deleted derived features leaked voice fusion decisions")

    return {
        "schema": "vignette.multimodal-alliance-live-api-evidence.v1",
        "generated_at": datetime.now(UTC).isoformat(),
        "api_base_url": args.api_base_url.rstrip("/"),
        "health": {"db": True, "engine": True},
        "learner_id": learner_id,
        "session_id": session_id,
        "consent_snapshot_id": str(consent["consent_snapshot_id"]),
        "timeline_id": str(timeline["timeline_id"]),
        "audio_asset_id": str(timeline["audio_asset_id"]),
        "fusion_record_ids": [str(item["fusion_record_id"]) for item in fusion_results],
        "fusion_applied": [bool(item["result"]["fusion_applied"]) for item in fusion_results],
        "withdrawal_snapshot_id": str(withdrawal["consent_snapshot_id"]),
        "deletion_request_ids": [
            str(explicit_deletion["deletion_request_id"]),
            str(withdrawal["deletion_request_id"]),
        ],
        "tombstone_ids": completion_ids,
        "assertions": {
            "internal_missing_token_401": True,
            "internal_wrong_token_403": True,
            "stable_retries": True,
            "changed_payload_conflicts_409": True,
            "single_clock_word_count": 2,
            "interaction_event_count": 4,
            "independent_measurement_count": 6,
            "fusion_benchmark_cases": 3,
            "fusion_gain_gate": [True, False, False],
            "teacher_metadata_without_raw_audio": True,
            "cross_cohort_hidden_404": True,
            "clinical_claim_allowed": False,
            "withdrawal_blocks_voice_processing": True,
            "raw_audio_deleted": True,
            "derived_voice_features_deleted": True,
            "transcript_retained": True,
        },
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--api-base-url", default="http://127.0.0.1:8014")
    parser.add_argument("--internal-token", required=True)
    parser.add_argument(
        "--benchmark-path",
        default="apps/api/app/data/multimodal_alliance_benchmark_g7.v1.json",
    )
    parser.add_argument(
        "--output",
        default="docs/ops/evidence/multimodal-alliance-live-api-2026-08-06.json",
    )
    parser.add_argument("--request-timeout", type=float, default=30.0)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    evidence = run(args)
    output = Path(args.output)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(
        json.dumps(evidence, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )
    print(json.dumps(evidence, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()