"""Gate B1 measurement: Discovery execution and readiness.

Does not overwrite 13A-01 or 13A-02. Does not score or change Planner matching.
Timeout policy is recorded separately and is not a reliability win by itself.
"""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import json
import sys
import time
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
from uuid import UUID

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from src.core.provider_capacity import watchdog_seconds  # noqa: E402
from src.domain.projects.discovery_b1 import (  # noqa: E402
    B1_COLUMNS,
    capacity_trace_fields,
    evaluate_trial_fidelity,
    replay_turn,
    score_b1,
    trial_payload,
)
from src.domain.projects.discovery_catalog import (  # noqa: E402
    build_discovery_payload,
    model_facing_discovery_payload,
)

FROZEN_13A02 = ROOT / "docs/audits/fixtures/13a_run_02"
PROTECTED_OUTPUTS = {
    ROOT / "docs/audits/fixtures/13a_run_01",
    ROOT / "docs/audits/fixtures/13a_run_02",
}


def write_json(path: Path, value) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def load_trials(source: Path) -> list[dict]:
    trials_dir = source / "trials"
    if not trials_dir.is_dir():
        raise ValueError(f"No trials directory at {source}")
    rows = []
    for path in sorted(trials_dir.glob("13A-*.json")):
        rows.append(json.loads(path.read_text(encoding="utf-8")))
    if len(rows) != 120:
        raise ValueError(f"Expected 120 frozen oracle trials, found {len(rows)}")
    return rows


def payload_instrumentation() -> dict:
    templates = [
        type(
            "T",
            (),
            {
                "ref": f"demo-{index}",
                "name": f"Demo {index}",
                "description": "x" * 800,
                "framework": "nuxt_app",
                "tags": ["demo"],
            },
        )
        for index in range(8)
    ]
    payload = build_discovery_payload(
        "nuxt_app",
        "generated",
        {"messages": [{"role": "user", "content": "Staff-only customer list"}]},
        templates,
    )
    model_payload = model_facing_discovery_payload(payload)
    full = len(json.dumps(payload, ensure_ascii=False))
    model = len(json.dumps(model_payload, ensure_ascii=False))
    return {
        "full_payload_bytes": full,
        "model_payload_bytes": model,
        "bytes_removed": full - model,
        "aliases_omitted": sorted(set(payload) - set(model_payload)),
    }


def replay_returned(trials: list[dict]) -> list[dict]:
    replayed = []
    for trial in trials:
        row = json.loads(json.dumps(trial))
        discovery = row.get("discovery") or {}
        if discovery.get("status") == "success" and discovery.get("raw_turn"):
            payload = trial_payload(row)
            result = replay_turn(discovery["raw_turn"], payload)
            fidelity = evaluate_trial_fidelity(
                {
                    **row,
                    "discovery": {
                        **discovery,
                        "result": result.model_dump(mode="json"),
                    },
                }
            )
            row["discovery"] = {
                **discovery,
                "result": result.model_dump(mode="json"),
                "b1_replay": True,
            }
            row["b1_fidelity"] = fidelity
        replayed.append(row)
    return replayed


async def live_context(baseline: dict):
    """Read the same workspace catalog without modifying the evaluation project."""
    from sqlalchemy import text
    from src.core.database import async_session_factory, engine
    from src.domain.projects.discovery_service import available_templates

    engine.echo = False
    try:
        async with async_session_factory() as db:
            await db.execute(text("SET TRANSACTION READ ONLY"))
            project = (
                (
                    await db.execute(
                        text(
                            "SELECT type, source_kind FROM projects "
                            "WHERE id=:pid AND workspace_id=:wid"
                        ),
                        {
                            "pid": UUID(baseline["project_id"]),
                            "wid": UUID(baseline["workspace_id"]),
                        },
                    )
                )
                .mappings()
                .first()
            )
            if not project:
                raise ValueError("Project does not belong to the evaluation workspace")
            templates = [
                SimpleNamespace(
                    **{
                        key: getattr(item, key)
                        for key in ("ref", "name", "description", "framework", "tags")
                    }
                )
                for item in await available_templates(db, UUID(baseline["workspace_id"]))
            ]
            revisions = list(
                (await db.execute(text("SELECT version_num FROM alembic_version"))).scalars()
            )
            await db.rollback()
            return dict(project), templates, revisions
    finally:
        await engine.dispose()


async def live_measure(
    trials: list[dict],
    output: Path,
    metadata: dict,
    runs: int,
    start_interval: float = 1.1,
    *,
    admission_gated: bool = False,
) -> dict:
    from src.config import get_settings
    from src.integrations.project_intake_client import discover_project

    settings = get_settings()
    baseline = json.loads((FROZEN_13A02 / "metadata.json").read_text(encoding="utf-8"))
    if runs < 2:
        raise ValueError("Live B1 requires at least two runs for reproducibility")
    if start_interval < 1.1:
        raise ValueError("Live B1 must not exceed the baseline request start rate")
    if settings.project_intake_timeout_seconds != baseline["timeout_seconds"]:
        raise ValueError("Live B1 requires the frozen 8s timeout policy")
    if settings.project_intake_model != baseline["discovery_model"]:
        raise ValueError("Model differs from the frozen baseline; record a separate experiment")
    if not settings.openai_api_key:
        raise ValueError("Live B1 requires the configured provider credential")
    project, templates, revisions = await live_context(baseline)
    if revisions != baseline["database_alembic_revisions"]:
        raise ValueError("Catalog revision differs from the frozen baseline")
    if project["type"] != baseline["project_type"] or len(templates) != baseline["template_count"]:
        raise ValueError("Project type or template count differs from the frozen baseline")
    metadata.update(
        {
            "discovery_model": settings.project_intake_model,
            "repeat_runs": runs,
            "concurrency": 4,
            "start_interval_seconds": start_interval,
            "load_policy_changed": start_interval != 1.1,
            "database_alembic_revisions": revisions,
            "template_count": len(templates),
            "workspace_verified": True,
            "status": "running",
            "project_id": baseline["project_id"],
            "workspace_id": baseline["workspace_id"],
            # M1: whether this run is gated by production's own admission policy
            # (reserve_discovery_call) or is raw_unthrottled_stress against the provider
            # directly. Never compare these two conditions' scorecards as equivalent.
            "measurement_kind": "production_admission_gated"
            if admission_gated
            else "raw_unthrottled_stress",
            "admission_policy_enabled": admission_gated,
            "configured_calls_per_minute": settings.project_intake_calls_per_minute,
            "provider_capacity_calibration": "unverified",
        }
    )
    metadata["implementation_hashes"] = {
        path: hashlib.sha256((ROOT / path).read_bytes()).hexdigest()
        for path in (
            "src/integrations/project_intake_client.py",
            "src/domain/projects/discovery_fidelity.py",
            "src/domain/projects/discovery_readiness.py",
            "src/domain/projects/discovery_catalog.py",
            "src/domain/projects/discovery_b1.py",
            "scripts/measure_discovery_readiness_13b.py",
            "src/domain/app_blocks/planner.py",
        )
    }
    write_json(output / "metadata.json", metadata)
    semaphore, start_lock = asyncio.Semaphore(4), asyncio.Lock()
    last_start = 0.0

    async def one(trial, run):
        nonlocal last_start
        async with semaphore:
            async with start_lock:
                await asyncio.sleep(max(0, last_start + start_interval - time.monotonic()))
                last_start = time.monotonic()
            row = {key: trial[key] for key in ("id", "prompt", "oracle_kind", "stratum")}
            payload = build_discovery_payload(
                project["type"], project["source_kind"], trial_payload(trial), templates
            )
            trace = {}
            end_to_end_started = time.monotonic()
            admission = None
            if admission_gated:
                # Reuse the exact production admission gate rather than reimplementing its
                # Redis/budget logic here (src/workers/tasks/project_discovery.py). It is a
                # reject-immediately check, not a queue: it never waits, so there is no
                # separate "admission wait" duration to report, only the Redis round trip.
                from src.workers.tasks.project_discovery import reserve_discovery_call

                admission_started = time.monotonic()
                try:
                    await reserve_discovery_call(settings)
                    admission = {
                        "attempted": True,
                        "granted": True,
                        "check_seconds": round(time.monotonic() - admission_started, 4),
                    }
                except ValueError as exc:
                    # The real production denial path: budget_exhausted, raised intentionally.
                    admission = {
                        "attempted": True,
                        "granted": False,
                        "denial_kind": "policy",
                        "reason": str(exc),
                        "check_seconds": round(time.monotonic() - admission_started, 4),
                    }
                except Exception as exc:
                    # Redis connection/timeout failure, distinct from an intentional denial.
                    # Must not crash the whole asyncio.gather() batch over one trial — every
                    # other admission/provider failure in this harness is recorded per-trial.
                    admission = {
                        "attempted": True,
                        "granted": False,
                        "denial_kind": "infrastructure",
                        "reason": f"{type(exc).__name__}: {exc}",
                        "check_seconds": round(time.monotonic() - admission_started, 4),
                    }
            if admission is not None and not admission["granted"]:
                row["discovery"] = {
                    "status": "error",
                    "error_type": (
                        "AdmissionDenied"
                        if admission["denial_kind"] == "policy"
                        else f"AdmissionCheckFailed:{admission['reason'].split(':')[0]}"
                    ),
                    "http_status": None,
                    "rate_limit_retries": [],
                }
            else:
                provider_started = time.monotonic()
                try:
                    turn = await asyncio.wait_for(
                        discover_project(payload, settings, trace=trace),
                        timeout=watchdog_seconds(
                            settings, settings.project_intake_timeout_seconds
                        ),
                    )
                    result = replay_turn(turn.model_dump(mode="json"), payload)
                    row["discovery"] = {
                        "status": "success",
                        "raw_turn": turn.model_dump(mode="json"),
                        "result": result.model_dump(mode="json"),
                        "usage": (trace.get("response") or {}).get("usage"),
                        "rate_limit_retries": trace.get("rate_limit_retries", []),
                        **capacity_trace_fields(trace),
                    }
                except Exception as exc:
                    row["discovery"] = {
                        "status": "error",
                        "error_type": type(exc).__name__,
                        "http_status": getattr(exc, "status_code", None),
                        "rate_limit_retries": trace.get("rate_limit_retries", []),
                        **capacity_trace_fields(trace),
                    }
                row["discovery"]["provider_elapsed_seconds"] = round(
                    time.monotonic() - provider_started, 3
                )
                if (row["discovery"].get("provider_execution_ms")) is not None:
                    row["discovery"]["provider_elapsed_seconds"] = round(
                        row["discovery"]["provider_execution_ms"] / 1000, 3
                    )
            if admission is not None:
                row["discovery"]["admission"] = admission
            row["discovery"]["elapsed_seconds"] = round(time.monotonic() - end_to_end_started, 3)
            row["payload_sha256"] = hashlib.sha256(
                json.dumps(model_facing_discovery_payload(payload), sort_keys=True).encode()
            ).hexdigest()
            write_json(output / f"run_{run:02}" / "trials" / f"{row['id']}.json", row)
            print(
                json.dumps({"run": run, "id": row["id"], "status": row["discovery"]["status"]}),
                flush=True,
            )
            return row

    repeated = []
    for run in range(1, runs + 1):
        rows = await asyncio.gather(*(one(trial, run) for trial in trials))
        repeated.append(rows)
        write_json(output / f"run_{run:02}" / "scorecard.json", score_b1(rows))
    scored = score_b1(repeated[0], repeats=repeated)
    scored["per_run"] = [score_b1(rows) for rows in repeated]
    metadata.update(status="complete", finished_at=datetime.now(UTC).isoformat())
    write_json(output / "metadata.json", metadata)
    write_json(output / "scorecard.json", scored)
    return scored


def measure(args) -> dict:
    output = Path(args.out).resolve()
    if output in PROTECTED_OUTPUTS or any(
        output == path or path in output.parents or output in path.parents
        for path in PROTECTED_OUTPUTS
    ):
        raise ValueError("Refusing to write into frozen 13A-01 or 13A-02 evidence")
    if output.exists() and (output / "scorecard.json").exists() and not args.allow_existing:
        raise ValueError("Existing measurement must not be overwritten; select a new output folder")
    source = Path(args.source).resolve()
    if args.mode == "after-live-replay":
        return asyncio.run(replay_live(source, output))
    trials = load_trials(source)
    if args.mode == "live":
        frozen = load_trials(FROZEN_13A02)
        keys = ("id", "prompt", "oracle_kind", "expected_product_blocks")
        if [[row[key] for key in keys] for row in trials] != [
            [row[key] for key in keys] for row in frozen
        ]:
            raise ValueError("Live B1 requires the unchanged frozen corpus and oracle")
        if output.exists() and any(output.iterdir()):
            raise ValueError("Live evidence requires an empty new output directory")
    metadata = {
        "phase": "13B",
        "gate": "B1",
        "started_at": datetime.now(UTC).isoformat(),
        "source": str(source.relative_to(ROOT)) if source.is_relative_to(ROOT) else str(source),
        "mode": args.mode,
        "timeout_policy_changed": False,
        "timeout_seconds": 8.0,
        "required_columns": list(B1_COLUMNS),
        "payload_instrumentation": payload_instrumentation(),
        "source_hashes": {
            "13a_run_02_freeze": hashlib.sha256(
                (ROOT / "docs/audits/fixtures/13a_run_02_freeze.json").read_bytes()
            ).hexdigest(),
            "13a_run_02_scorecard": hashlib.sha256(
                (ROOT / "docs/audits/fixtures/13a_run_02/scorecard.json").read_bytes()
            ).hexdigest(),
            "phase_13a_corpus": hashlib.sha256(
                (ROOT / "scripts/phase_13a_corpus.txt").read_bytes()
            ).hexdigest(),
        },
        "do_not": [
            "overwrite 13A-01 or 13A-02",
            "begin Gate B2",
            "treat timeout increase as success",
            "change Planner or Phase 12 contracts",
        ],
    }
    if args.mode == "live":
        return asyncio.run(
            live_measure(
                trials,
                output,
                metadata,
                args.runs,
                args.start_interval,
                admission_gated=args.admission_gated,
            )
        )
    if args.mode == "before":
        scored = score_b1(trials, timeout_policy_changed=False)
        write_json(output / "metadata.json", metadata)
        write_json(output / "scorecard.json", scored)
        write_json(
            output / "fidelity.json",
            [evaluate_trial_fidelity(row) | {"id": row["id"]} for row in trials],
        )
        return scored
    if args.mode == "after-replay":
        replayed = replay_returned(trials)
        scored = score_b1(replayed, timeout_policy_changed=False)
        write_json(output / "metadata.json", metadata)
        write_json(output / "scorecard.json", scored)
        write_json(
            output / "fidelity.json",
            [evaluate_trial_fidelity(row) | {"id": row["id"]} for row in replayed],
        )
        before_by_id = {row["id"]: row for row in trials}
        promoted = [
            row["id"]
            for row in replayed
            if (row.get("discovery") or {}).get("status") == "success"
            and (row.get("discovery") or {}).get("result", {}).get("decision") == "ready"
            and (before_by_id[row["id"]].get("discovery") or {}).get("result", {}).get("decision")
            != "ready"
        ]
        write_json(output / "promoted.json", promoted)
        return scored
    raise ValueError(f"Unsupported mode {args.mode}")


async def replay_live(source: Path, output: Path) -> dict:
    """Replay the finalizer against live raw turns; never claim new provider observations."""
    if output.exists() and any(output.iterdir()):
        raise ValueError("Replay evidence requires an empty new output directory")
    metadata = json.loads((source / "metadata.json").read_text(encoding="utf-8"))
    if metadata.get("mode") != "live" or metadata.get("status") != "complete":
        raise ValueError("Replay requires a completed live measurement")
    project, templates, _ = await live_context(metadata)
    repeated = []
    for run in range(1, metadata["repeat_runs"] + 1):
        rows = load_trials(source / f"run_{run:02}")
        for row in rows:
            payload = build_discovery_payload(
                project["type"], project["source_kind"], trial_payload(row), templates
            )
            digest = hashlib.sha256(
                json.dumps(model_facing_discovery_payload(payload), sort_keys=True).encode()
            ).hexdigest()
            if digest != row["payload_sha256"]:
                raise ValueError("Live payload changed; refusing a non-equivalent replay")
            discovery = row["discovery"]
            if discovery["status"] == "success":
                discovery["result"] = replay_turn(discovery["raw_turn"], payload).model_dump(
                    mode="json"
                )
                discovery["b1_replay"] = True
            write_json(output / f"run_{run:02}" / "trials" / f"{row['id']}.json", row)
        repeated.append(rows)
    scored = score_b1(repeated[0], repeats=repeated)
    scored["per_run"] = [score_b1(rows) for rows in repeated]
    scored["execution_metrics_source"] = "Inherited live timings and errors; finalizer replay only."
    write_json(output / "scorecard.json", scored)
    write_json(
        output / "metadata.json",
        {
            "phase": "13B",
            "gate": "B1",
            "mode": "after-live-replay",
            "source": str(source),
            "provider_calls": 0,
            "source_scorecard_sha256": hashlib.sha256(
                (source / "scorecard.json").read_bytes()
            ).hexdigest(),
            "payload_hashes_verified": True,
            "implementation_hashes": {
                path: hashlib.sha256((ROOT / path).read_bytes()).hexdigest()
                for path in (
                    "src/domain/projects/discovery_fidelity.py",
                    "src/domain/projects/discovery_readiness.py",
                )
            },
        },
    )
    return scored


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out", required=True)
    parser.add_argument("--source", default=str(FROZEN_13A02))
    parser.add_argument(
        "--mode", choices=("before", "after-replay", "live", "after-live-replay"), default="before"
    )
    parser.add_argument("--runs", type=int, default=2)
    parser.add_argument("--start-interval", type=float, default=1.1)
    parser.add_argument("--allow-existing", action="store_true")
    parser.add_argument(
        "--admission-gated",
        action="store_true",
        help=(
            "Route each call through the real production admission policy "
            "(reserve_discovery_call, shared Redis budget) instead of calling "
            "discover_project directly. Without this flag, --mode live is "
            "raw_unthrottled_stress, not production-representative."
        ),
    )
    scored = measure(parser.parse_args())
    print(json.dumps(scored, indent=2))