"""Overnight Oak perf-improvement loop primitives.

The loop is intentionally propose-only: scripts can select a target, verify a

candidate, and append the ledger, but they never edit Oak and never merge.

"""

from __future__ import annotations

import json

import os

import statistics

import subprocess

import time

from dataclasses import dataclass

from datetime import datetime, timezone

from pathlib import Path

from typing import Any, Iterable, Sequence

from oakbench.reporting import median

from oakbench.rows import row_returncode

SCORECARD_SCHEMA_VERSION = 1

LEDGER_MARKER_START = "<!-- oak-perf-loop-entry"

LEDGER_MARKER_END = "-->"

DIAGNOSTIC_SUFFIXES = (".determinism", ".inforecall")

SKIP_RETURNCODE = 77

SCORECARD_EXIT_BETTER = 0

SCORECARD_EXIT_NO_CHANGE = 10

SCORECARD_EXIT_REGRESSED = 20

def load_jsonl(paths: Sequence[Path]) -> list[dict[str, Any]]:

    rows: list[dict[str, Any]] = []

    for path in paths:

        with path.open() as fh:

            for line_number, raw in enumerate(fh, start=1):

                line = raw.strip()

                if not line:

                    continue

                try:

                    row = json.loads(line)

                except json.JSONDecodeError as exc:

                    raise SystemExit(f"{path}:{line_number}: invalid JSON: {exc}") from exc

                if not isinstance(row, dict):

                    raise SystemExit(f"{path}:{line_number}: expected JSON object")

                rows.append(row)

    return rows

def write_json(path: Path, payload: dict[str, Any]) -> None:

    path.parent.mkdir(parents=True, exist_ok=True)

    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")

def scorecard_exit_code(verdict: str) -> int:

    if verdict == "better":

        return SCORECARD_EXIT_BETTER

    if verdict == "no-change":

        return SCORECARD_EXIT_NO_CHANGE

    if verdict == "regressed":

        return SCORECARD_EXIT_REGRESSED

    raise ValueError(f"unknown scorecard verdict: {verdict}")

def row_lane(row: dict[str, Any]) -> str:

    for key in ("lane", "benchmark_lane", "measurement_lane"):

        if row.get(key):

            return str(row[key])

    if "workflow" in row or "phase" in row:

        return "workflow"

    if "contention_mode" in row or "workers" in row:

        return "contention"

    if all(key in row for key in ("profile", "scenario", "operation", "subject", "subject_kind")):

        return "core"

    return "unknown"

def scorecard_key(row: dict[str, Any]) -> tuple[str, str | None, str, str]:

    profile = row.get("profile")

    return (

        row_lane(row),

        str(profile) if profile is not None else None,

        str(row.get("scenario") or ""),

        str(row.get("operation") or ""),

    )

def _successful_latency_rows(rows: Iterable[dict[str, Any]], subject: str) -> list[dict[str, Any]]:

    return [

        row

        for row in rows

        if str(row.get("subject")) == subject

        and row_returncode(row) == 0

        and not bool(row.get("skipped"))

        and isinstance(row.get("elapsed_ms"), (int, float))

    ]

def _transport_signature(rows: list[dict[str, Any]]) -> tuple[str, ...]:

    values = {

        str(row.get("remote_transport") or row.get("transport") or "local")

        for row in rows

        if row.get("remote_transport") is not None or row.get("transport") is not None

    }

    if not values:

        return ("local",)

    return tuple(sorted(values))

def _entry_id(entry: dict[str, Any]) -> str:

    profile = entry.get("profile")

    profile_part = "" if profile is None else f"{profile}:"

    return f"{entry['lane']}:{profile_part}{entry['scenario']}/{entry['operation']}"

def _baseline_by_id(baseline: dict[str, Any] | None) -> dict[str, dict[str, Any]]:

    if not baseline:

        return {}

    return {_entry_id(entry): entry for entry in baseline.get("entries", []) if isinstance(entry, dict)}

def build_scorecard(

    rows: list[dict[str, Any]],

    *,

    oak_subject: str,

    git_subject: str = "git",

    baseline: dict[str, Any] | None = None,

    change_threshold_pct: float = 3.0,

) -> dict[str, Any]:

    grouped: dict[tuple[str, str | None, str, str], list[dict[str, Any]]] = {}

    for row in rows:

        grouped.setdefault(scorecard_key(row), []).append(row)

    baseline_entries = _baseline_by_id(baseline)

    entries: list[dict[str, Any]] = []

    skipped: list[dict[str, Any]] = []

    any_better = False

    any_regressed = False

    for key, group in sorted(grouped.items()):

        lane, profile, scenario, operation = key

        if not scenario or not operation:

            continue

        oak_rows = _successful_latency_rows(group, oak_subject)

        git_rows = _successful_latency_rows(group, git_subject)

        if not oak_rows or not git_rows:

            continue

        oak_transport = _transport_signature(oak_rows)

        git_transport = _transport_signature(git_rows)

        if oak_transport != git_transport:

            skipped.append(

                {

                    "lane": lane,

                    "profile": profile,

                    "scenario": scenario,

                    "operation": operation,

                    "reason": "transport_mismatch",

                    "oak_transport": list(oak_transport),

                    "git_transport": list(git_transport),

                }

            )

            continue

        oak_values = [float(row["elapsed_ms"]) for row in oak_rows]

        git_values = [float(row["elapsed_ms"]) for row in git_rows]

        oak_median = median(oak_values)

        git_median = median(git_values)

        if oak_median is None or git_median is None or git_median <= 0:

            continue

        ratio = oak_median / git_median

        entry: dict[str, Any] = {

            "lane": lane,

            "profile": profile,

            "scenario": scenario,

            "operation": operation,

            "metric": "elapsed_ms",

            "oak_subject": oak_subject,

            "git_subject": git_subject,

            "oak_median_ms": round(oak_median, 3),

            "git_median_ms": round(git_median, 3),

            "ratio": ratio,

            "n_oak": len(oak_values),

            "n_git": len(git_values),

            "transport": list(oak_transport),

            "verdict": "oak_slower" if ratio > 1.0 else ("oak_faster" if ratio < 1.0 else "parity"),

        }

        previous = baseline_entries.get(_entry_id(entry))

        if previous and isinstance(previous.get("ratio"), (int, float)):

            baseline_ratio = float(previous["ratio"])

            ratio_delta_pct = ((ratio - baseline_ratio) / baseline_ratio) * 100.0 if baseline_ratio else 0.0

            entry["baseline"] = {

                "ratio": round(baseline_ratio, 6),

                "ratio_delta_pct": round(ratio_delta_pct, 3),

            }

            if ratio_delta_pct <= -change_threshold_pct:

                entry["baseline"]["verdict"] = "better"

                any_better = True

            elif ratio_delta_pct >= change_threshold_pct:

                entry["baseline"]["verdict"] = "regressed"

                any_regressed = True

            else:

                entry["baseline"]["verdict"] = "no-change"

        entries.append(entry)

    gate_verdict = "regressed" if any_regressed else ("better" if any_better else "no-change")

    worst = max(

        (entry for entry in entries if entry.get("ratio") is not None),

        key=lambda entry: float(entry["ratio"]),

        default=None,

    )

    return {

        "schema_version": SCORECARD_SCHEMA_VERSION,

        "generated_at": datetime.now(timezone.utc).isoformat(),

        "oak_subject": oak_subject,

        "git_subject": git_subject,

        "metric": "elapsed_ms",

        "change_threshold_pct": change_threshold_pct,

        "summary": {

            "gate_verdict": gate_verdict,

            "entry_count": len(entries),

            "skipped_comparison_count": len(skipped),

            "worst": worst,

        },

        "entries": entries,

        "skipped_comparisons": skipped,

    }

def load_scorecard(path: Path) -> dict[str, Any]:

    payload = json.loads(path.read_text())

    if not isinstance(payload, dict):

        raise SystemExit(f"{path}: expected scorecard JSON object")

    return payload

def target_key(target: dict[str, Any]) -> str:

    profile = target.get("profile")

    profile_part = "" if profile is None else f"{profile}:"

    return f"{target.get('lane', 'core')}:{profile_part}{target['scenario']}/{target['operation']}"

def is_diagnostic_operation(operation: str) -> bool:

    return operation.endswith(DIAGNOSTIC_SUFFIXES)

def parse_ledger(path: Path) -> list[dict[str, Any]]:

    if not path.exists():

        return []

    lines = path.read_text().splitlines()

    entries: list[dict[str, Any]] = []

    index = 0

    while index < len(lines):

        if lines[index].strip() != LEDGER_MARKER_START:

            index += 1

            continue

        index += 1

        payload_lines: list[str] = []

        while index < len(lines) and lines[index].strip() != LEDGER_MARKER_END:

            payload_lines.append(lines[index])

            index += 1

        if index >= len(lines):

            break

        index += 1

        try:

            payload = json.loads("\n".join(payload_lines))

        except json.JSONDecodeError:

            continue

        if isinstance(payload, dict):

            entries.append(payload)

    return entries

def no_payoff_targets(ledger_entries: list[dict[str, Any]]) -> set[str]:

    blocked: set[str] = set()

    for entry in ledger_entries:

        target = entry.get("target")

        if not isinstance(target, dict):

            continue

        verdict = str(entry.get("verdict") or entry.get("landed_or_parked") or "")

        reason = str(entry.get("reason") or "")

        if verdict in {"no-payoff", "parked_no_payoff"} or "did_not_pay" in reason or "no_payoff" in reason:

            blocked.add(target_key(target))

    return blocked

def select_target(scorecard: dict[str, Any], ledger_entries: list[dict[str, Any]]) -> dict[str, Any] | None:

    blocked = no_payoff_targets(ledger_entries)

    candidates = []

    for entry in scorecard.get("entries", []):

        if not isinstance(entry, dict):

            continue

        if entry.get("metric") != "elapsed_ms":

            continue

        if is_diagnostic_operation(str(entry.get("operation") or "")):

            continue

        if float(entry.get("ratio") or 0.0) <= 1.0:

            continue

        if target_key(entry) in blocked:

            continue

        candidates.append(entry)

    if not candidates:

        return None

    candidates.sort(key=lambda entry: (-float(entry["ratio"]), str(entry["scenario"]), str(entry["operation"])))

    return candidates[0]

def append_ledger(path: Path, entry: dict[str, Any]) -> None:

    path.parent.mkdir(parents=True, exist_ok=True)

    if not path.exists():

        path.write_text("# Oak Perf Loop Ledger\n\n")

    target = entry.get("target") if isinstance(entry.get("target"), dict) else {}

    title = target_key(target) if target else "unknown-target"

    verdict = entry.get("landed_or_parked") or entry.get("verdict") or "unknown"

    reason = entry.get("reason") or ""

    text = path.read_text()

    with path.open("a") as fh:

        if text and not text.endswith("\n"):

            fh.write("\n")

        fh.write(f"## {entry.get('timestamp', datetime.now(timezone.utc).isoformat())} — {title}\n\n")

        fh.write(f"- Verdict: {verdict}\n")

        if reason:

            fh.write(f"- Reason: {reason}\n")

        before = entry.get("before_after_medians") or {}

        if before.get("before_ms") is not None and before.get("after_ms") is not None:

            fh.write(

                "- Medians: "

                f"before {before.get('before_ms')} ms, after {before.get('after_ms')} ms"

                f", delta {before.get('delta_pct')}%\n"

            )

        elif before.get("before_ms") is not None:

            fh.write(f"- Current median: {before.get('before_ms')} ms\n")

        if entry.get("change_summary"):

            fh.write(f"- Change: {entry['change_summary']}\n")

        if entry.get("diagnosis"):

            fh.write(f"- Diagnosis: {entry['diagnosis']}\n")

        rejected = entry.get("levers_rejected") or []

        if rejected:

            fh.write("- Levers rejected:\n")

            for item in rejected:

                fh.write(f"  - {item}\n")

        fh.write("\n")

        fh.write(LEDGER_MARKER_START + "\n")

        fh.write(json.dumps(entry, indent=2, sort_keys=True) + "\n")

        fh.write(LEDGER_MARKER_END + "\n\n")

@dataclass(frozen=True)

class ABSample:

    label: str

    elapsed_ms: float

    returncode: int = 0

    stderr: str = ""

def run_interleaved_commands(

    before_command: Sequence[str],

    after_command: Sequence[str],

    *,

    cwd: Path,

    pairs: int,

    env: dict[str, str] | None = None,

) -> list[ABSample]:

    if pairs < 20:

        raise ValueError("interleaved A/B gate requires at least 20 before/after pairs")

    samples: list[ABSample] = []

    run_env = os.environ.copy()

    if env:

        run_env.update(env)

    for _ in range(pairs):

        for label, command in (("before", before_command), ("after", after_command)):

            start = time.perf_counter()

            proc = subprocess.run(

                list(command),

                cwd=cwd,

                env=run_env,

                stdout=subprocess.DEVNULL,

                stderr=subprocess.PIPE,

                text=True,

                check=False,

            )

            samples.append(

                ABSample(

                    label=label,

                    elapsed_ms=(time.perf_counter() - start) * 1000.0,

                    returncode=proc.returncode,

                    stderr=(proc.stderr or "")[-1000:],

                )

            )

    return samples

def evaluate_interleaved_ab(

    samples: Sequence[ABSample],

    *,

    min_pairs: int = 20,

    min_improvement_pct: float = 3.0,

) -> dict[str, Any]:

    if len(samples) < min_pairs * 2:

        return {

            "passed": False,

            "reason": f"need at least {min_pairs} before/after pairs",

            "sample_count": len(samples),

        }

    for index, sample in enumerate(samples):

        expected = "before" if index % 2 == 0 else "after"

        if sample.label != expected:

            return {"passed": False, "reason": "samples are not strictly alternating before/after"}

        if sample.returncode != 0:

            return {

                "passed": False,

                "reason": f"{sample.label} command failed at sample {index}: {sample.stderr}",

            }

    before_values = [sample.elapsed_ms for sample in samples if sample.label == "before"]

    after_values = [sample.elapsed_ms for sample in samples if sample.label == "after"]

    if len(before_values) < min_pairs or len(after_values) < min_pairs:

        return {"passed": False, "reason": f"need {min_pairs} samples per side"}

    before_median = float(statistics.median(before_values))

    after_median = float(statistics.median(after_values))

    before_p25 = percentile(before_values, 0.25)

    after_p25 = percentile(after_values, 0.25)

    delta_pct = ((after_median - before_median) / before_median) * 100.0 if before_median else 0.0

    median_improved = delta_pct <= -min_improvement_pct

    consistency_improved = after_p25 < before_p25

    if median_improved and consistency_improved:

        reason = "target_metric_improved"

    elif median_improved:

        reason = "target_metric_improvement_not_consistent"

    else:

        reason = "target_metric_did_not_improve"

    return {

        "passed": median_improved and consistency_improved,

        "reason": reason,

        "before_ms": round(before_median, 3),

        "after_ms": round(after_median, 3),

        "delta_pct": round(delta_pct, 3),

        "before_p25_ms": round(before_p25, 3),

        "after_p25_ms": round(after_p25, 3),

        "before_n": len(before_values),

        "after_n": len(after_values),

    }

def run_correctness_gate(oak_repo: Path, *, skip: bool = False) -> list[dict[str, Any]]:

    if skip:

        return [{"command": "skip", "returncode": 0, "skipped": True}]

    commands = [

        ["cargo", "build", "-p", "oakvcs-cli"],

        ["cargo", "test", "-p", "oakvcs-cli"],

        ["cargo", "clippy", "-p", "oakvcs-cli", "--all-targets", "--", "-D", "warnings"],

    ]

    results: list[dict[str, Any]] = []

    for command in commands:

        proc = subprocess.run(

            command,

            cwd=oak_repo,

            stdout=subprocess.PIPE,

            stderr=subprocess.STDOUT,

            text=True,

            check=False,

        )

        results.append(

            {

                "command": command,

                "returncode": proc.returncode,

                "output_tail": (proc.stdout or "")[-4000:],

            }

        )

        if proc.returncode != 0:

            break

    return results

def correctness_passed(results: Sequence[dict[str, Any]]) -> bool:

    return all(int(result.get("returncode", 1)) == 0 for result in results)

def percentile(values: Sequence[float], fraction: float) -> float:

    if not values:

        raise ValueError("percentile requires at least one value")

    if not 0.0 <= fraction <= 1.0:

        raise ValueError("percentile fraction must be between 0 and 1")

    ordered = sorted(values)

    if len(ordered) == 1:

        return float(ordered[0])

    position = (len(ordered) - 1) * fraction

    lower = int(position)

    upper = min(lower + 1, len(ordered) - 1)

    weight = position - lower

    return float((ordered[lower] * (1.0 - weight)) + (ordered[upper] * weight))

def verification_entry(

    *,

    target: dict[str, Any],

    diagnosis: str,

    change_summary: str,

    correctness_results: Sequence[dict[str, Any]],

    ab_result: dict[str, Any],

    oak_gate_verdict: str = "MISSING",

    fleet_gate_verdict: str = "MISSING",

    levers_rejected: Sequence[str] = (),

    propose_mode: bool = True,

) -> dict[str, Any]:

    if not diagnosis.strip():

        status = "parked"

        reason = "missing_diagnosis_measurements"

    elif not correctness_passed(correctness_results):

        status = "parked"

        reason = "correctness_gate_failed"

    elif not ab_result.get("passed"):

        status = "parked"

        reason = str(ab_result.get("reason") or "interleaved_ab_failed")

    elif oak_gate_verdict != "PASS":

        status = "parked"

        reason = "oak_efficiency_gate_regressed"

    elif fleet_gate_verdict != "PASS":

        status = "parked"

        reason = "fleet_reliability_gate_regressed"

    elif propose_mode:

        status = "parked"

        reason = "verified_propose_mode_human_review_required"

    else:

        status = "landed"

        reason = "verified"

    return {

        "schema_version": 1,

        "timestamp": datetime.now(timezone.utc).isoformat(),

        "target": {

            "lane": target.get("lane"),

            "profile": target.get("profile"),

            "scenario": target.get("scenario"),

            "operation": target.get("operation"),

            "ratio": target.get("ratio"),

            "oak_median_ms": target.get("oak_median_ms"),

            "git_median_ms": target.get("git_median_ms"),

        },

        "diagnosis": diagnosis,

        "change_summary": change_summary,

        "before_after_medians": {

            "before_ms": ab_result.get("before_ms"),

            "after_ms": ab_result.get("after_ms"),

            "delta_pct": ab_result.get("delta_pct"),

        },

        "correctness_gate": list(correctness_results),

        "oak_efficiency_gate": oak_gate_verdict,

        "fleet_reliability_gate": fleet_gate_verdict,

        "levers_rejected": list(levers_rejected),

        "landed_or_parked": status,

        "reason": reason,

    }

def dry_run_entry(target: dict[str, Any]) -> dict[str, Any]:

    return verification_entry(

        target=target,

        diagnosis="dry-run: target selection and ledger plumbing exercised; no Rust edit attempted",

        change_summary="dry-run parked candidate",

        correctness_results=[{"command": "dry-run", "returncode": 0}],

        ab_result={

            "passed": False,

            "reason": "dry_run_no_candidate_branch",

            "before_ms": target.get("oak_median_ms"),

            "after_ms": None,

            "delta_pct": None,

        },

        levers_rejected=[],

    )