"""Interleaved, A/A-controlled execution of the scale op catalog on one tier.

Schedule: for each run, for each unit (an op or a short dependent op

sequence, see ops.UNIT_ORDER), subjects execute in a freshly shuffled

(seeded) order, so every subject sees the same drift and host noise. A/A

control subjects (``git_aa``, ``oak_installed_aa``: the same binary under a

second name) share their kind's workspace and are interleaved exactly like

the real arms; every cell gets its own A/A floor (oakbench.cellstats).

Measurement is serialized with the cross-process measurement lock around

the whole timed schedule (preparation runs outside it).

Every row is development evidence: ``evidence_class`` says so and

``claim_eligible`` is false. The calibrated campaign re-runs the same

machinery on a quiet host.

"""

from __future__ import annotations

import json

import platform

import random

import statistics

import sys

import time

from pathlib import Path

from typing import Any, Dict, List, Optional, Sequence, Tuple

from . import ops as ops_mod

from .workspace import TRANSPORT_GIT, TRANSPORT_OAK, Prepared

_scripts = str(Path(__file__).resolve().parents[3] / "scripts")

if _scripts not in sys.path:

    sys.path.insert(0, _scripts)

from oakbench import scale_calibration  # noqa: E402

SCALE_ROW_VERSION = "scale-2026-09-28.1"

EVIDENCE_CLASS = "development_only_loaded_host"

AA_PAIRS = {"git": ("git", "git_aa"), "oak": ("oak_installed", "oak_installed_aa")}

NETWORK_OPS = frozenset({"push.branch", "pull.ff", "pull.diverged", "clone.full"})

def _say(message: str) -> None:

    print(message, file=sys.stderr, flush=True)

def _subject_meta(subject: Any, details: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:

    item = details.get(subject.name, {})

    return {"subject_bin": item.get("bin"), "subject_version": item.get("version"),

            "binary_sha256": item.get("binary_sha256"), "subject_role": item.get("role"),

            "git_shim_measured": item.get("git_shim_measured"), "git_config_profile": item.get("git_config_profile")}

def build_row(*, bench_id: str, prepared: Prepared, subject: Any, op: str, run: int, outcome: ops_mod.Outcome,

              details: Dict[str, Dict[str, Any]], env_sample: Dict[str, Any], lock_wait_ms: Optional[float],

              order_position: int, runner_fields: Dict[str, Any],

              calibration: Optional[Tuple[Dict[str, Any], Dict[str, Any], List[str]]] = None

              ) -> Dict[str, Any]:

    info = ops_mod.OP_INFO[op]

    captures = [item.capture for item in outcome.timed]

    def total(attr: str) -> Optional[float]:

        values = [getattr(c, attr) for c in captures]

        if not values or any(v is None for v in values):

            return None

        return round(sum(values), 3)

    row: Dict[str, Any] = {

        "bench_id": bench_id,

        "profile": "scale",

        "scale_row_version": SCALE_ROW_VERSION,

        "scenario": prepared.tier.tier_id,

        "operation": op,

        "run": run,

        "subject": subject.name,

        "subject_kind": subject.kind,

        "command": [item.command for item in outcome.timed],

        "component_elapsed_ms": {item.label: round(item.capture.elapsed_ms, 3) for item in outcome.timed},

        "elapsed_ms": round(sum(c.elapsed_ms for c in captures), 3) if captures else None,

        "cpu_user_ms": total("cpu_user_ms"),

        "cpu_sys_ms": total("cpu_sys_ms"),

        "io_read_blocks": total("io_read_blocks"),

        "io_write_blocks": total("io_write_blocks"),

        "peak_rss_bytes": max((c.peak_rss_bytes or 0 for c in captures), default=None) if captures else None,

        "stdout_bytes": sum(c.stdout_bytes for c in captures) if captures else None,

        "timed_out": any(c.timed_out for c in captures),

        "measurement_source": "direct_cli_timed_subprocess",

        "rusage_source": "os_wait4_child_rusage",

        "page_cache_state": "warm",

        "cache_mode_requested": "warm",

        "tperf_refs": list(info.tperf),

        "core_yaml_ops": list(info.core_yaml),

        "oracle_ok": outcome.oracle_ok,

        "oracle_method": outcome.oracle_method,

        "oracle_reader": outcome.oracle_reader,

        "oracle_timed": False,

        "oracle_detail": outcome.oracle_detail,

        "semantic_class": outcome.semantic_class,

        "transport": (TRANSPORT_GIT if subject.kind == "git" else TRANSPORT_OAK) if op in NETWORK_OPS else None,

        "order_position": order_position,

        "environment_sample": env_sample,

        "measurement_lock_wait_ms": lock_wait_ms,

        **_subject_meta(subject, details),

        **prepared.fixture_fields(),

        **runner_fields,

        **outcome.extra,

    }

    if outcome.skip_reason is not None:

        # Skip rows follow the core lane: elapsed 0.0 with skipped: true (never a measurement).

        row.update(returncode=77, skipped=True, skip_reason=outcome.skip_reason, elapsed_ms=0.0)

        row.pop("measurement_source", None)

    elif outcome.failed is not None:

        process_rc = captures[-1].returncode if captures and captures[-1].returncode != 0 else None

        row.update(returncode=1, process_returncode=process_rc, failure=outcome.failed,

                   elapsed_ms=row["elapsed_ms"] if row["elapsed_ms"] is not None else 0.0)

    elif outcome.oracle_ok is False:

        row.update(returncode=1, process_returncode=captures[-1].returncode if captures else None,

                   failure="oracle_failed")

    else:

        row.update(returncode=0)

    # evidence_class / claim_eligible are derived from a validated quiet-host

    # calibration record, never set freely (oakbench.scale_calibration).

    record, facts, reasons = calibration if calibration else (None, None, [])

    return scale_calibration.stamp(row, record, facts, reasons)

def abort_row(base: Dict[str, Any], op: str, after: str) -> Dict[str, Any]:

    row = dict(base)

    info = ops_mod.OP_INFO[op]

    for key in ("oracle_ok", "oracle_method", "oracle_reader", "semantic_class", "failure", "process_returncode",

                "component_elapsed_ms", "cpu_user_ms", "cpu_sys_ms", "io_read_blocks", "io_write_blocks",

                "peak_rss_bytes", "stdout_bytes", "dirty_applied", "harness_error"):

        row.pop(key, None)

    row.pop("measurement_source", None)

    for key in ("calibration_record", "calibration_record_sha256"):

        row.pop(key, None)

    row.update(evidence_class=EVIDENCE_CLASS, claim_eligible=False, claim_ineligible_reasons=["skip row"])

    row.update(operation=op, command=[], elapsed_ms=0.0, returncode=77, skipped=True, oracle_detail={},

               skip_reason=f"aborted_after_failed_prerequisite:{after}", tperf_refs=list(info.tperf),

               core_yaml_ops=list(info.core_yaml))

    return row

def run_schedule(prepared: Prepared, subjects: Sequence[Any], *, runs: int, ops: Sequence[str], bench_id: str,

                 seed: int, scratch: Path, details: Dict[str, Dict[str, Any]], runner_fields: Dict[str, Any],

                 timeout_s: float, lock_wait_ms: Optional[float], log=None,

                 calibration: Optional[Tuple[Dict[str, Any], Dict[str, Any], List[str]]] = None

                 ) -> List[Dict[str, Any]]:

    scripts = str(Path(__file__).resolve().parents[3] / "scripts")

    if scripts not in sys.path:

        sys.path.insert(0, scripts)

    from oakbench.envwatch import sample_environment

    from oakbench.execution import run_timed

    scratch.mkdir(parents=True, exist_ok=True)

    rng = random.Random(seed)

    wanted = set(ops)

    units = [tuple(op for op in unit if op in wanted) for unit in ops_mod.UNIT_ORDER]

    units = [unit for unit in units if unit]

    rows: List[Dict[str, Any]] = []

    skip_by_tier = set(prepared.tier.skip_ops)

    for run in range(1, runs + 1):

        started = time.perf_counter()

        for unit in units:

            order = list(subjects)

            rng.shuffle(order)

            env_sample = sample_environment()

            for position, subject in enumerate(order):

                failed_at: Optional[str] = None

                for op in unit:

                    ctx = ops_mod.Ctx(prepared, subject, run, run_timed, scratch, timeout_s)

                    if failed_at is not None:

                        base = build_row(bench_id=bench_id, prepared=prepared, subject=subject, op=op, run=run,

                                         outcome=ops_mod.Outcome(), details=details, env_sample=env_sample,

                                         lock_wait_ms=lock_wait_ms, order_position=position,

                                         runner_fields=runner_fields)

                        rows.append(abort_row(base, op, failed_at))

                        continue

                    if op in skip_by_tier:

                        outcome = ops_mod.Outcome(skip_reason=f"tier_excludes_op_per_run_cost:{prepared.tier.tier_id}")

                    elif subject.kind not in prepared.kinds:

                        outcome = ops_mod.Outcome(skip_reason=f"kind_not_prepared:{subject.kind}")

                    else:

                        outcome = ops_mod.execute(ctx, op)

                    row = build_row(bench_id=bench_id, prepared=prepared, subject=subject, op=op, run=run,

                                    outcome=outcome, details=details, env_sample=env_sample,

                                    lock_wait_ms=lock_wait_ms, order_position=position,

                                    runner_fields=runner_fields, calibration=calibration)

                    rows.append(row)

                    if row["returncode"] == 1:

                        if len(unit) > 1:

                            failed_at = op

                        # Untimed repair so one failure cannot poison later units.

                        problem = ops_mod.recover(ctx)

                        if problem:

                            row["recover_error"] = problem

        (log or _say)(f"[scale] {prepared.tier.tier_id}: run {run}/{runs} done in {time.perf_counter() - started:.1f}s")

    return rows

# ---------------------------------------------------------------------------

# Statistics: per-cell A/A floors, dev-only oak/git ratio, MDE per cell

# ---------------------------------------------------------------------------

COMPARABLE = "comparable"

def cell_comparability(rows: List[Dict[str, Any]], key: Tuple[str, str]) -> str:

    """Whether Oak and Git measured the same thing in a cell.

    An Oak server seeded with a one-commit snapshot (push admission caps)

    serves a different repository than Git's full-history server, so

    push/pull/clone Oak-vs-Git deltas are meaningless there; only the Oak A/A

    (oak-vs-oak) evidence stays valid.

    """

    seeds = {r.get("oak_server_seed") for r in rows if (r.get("scenario"), r.get("operation")) == key

             and r.get("subject_kind") == "oak" and r.get("returncode") != 77}

    seeds.discard(None)

    seeds.discard("full_history")

    if seeds:

        return "not_comparable:oak_server_seed_" + ",".join(sorted(seeds))

    return COMPARABLE

def cell_report(rows: List[Dict[str, Any]], subjects: Sequence[str]) -> List[Dict[str, Any]]:

    from oakbench import cellstats

    measured = [r for r in rows if r.get("returncode") == 0 and isinstance(r.get("elapsed_ms"), (int, float))]

    floors: Dict[str, Dict[Tuple[str, str], float]] = {}

    for kind, (arm, control) in AA_PAIRS.items():

        if arm in subjects and control in subjects:

            floors[kind] = cellstats.cell_floors(measured, arm, control, lambda r: r.get("elapsed_ms"))

    comparability = {key: cell_comparability(rows, key) for key in {(r["scenario"], r["operation"]) for r in rows}}

    cross: Dict[Tuple[str, str], Dict[str, Any]] = {}

    if "git" in subjects and "oak_installed" in subjects:

        keys = set(floors.get("git", {})) | set(floors.get("oak", {}))

        combined = {key: max(floors.get("git", {}).get(key, 0.0), floors.get("oak", {}).get(key, 0.0))

                    for key in keys}

        # Only comparable cells enter the Oak-vs-Git family (and its BH).

        comparable = [key for key, label in comparability.items() if label == COMPARABLE]

        cross = cellstats.classify_cells(measured, "oak_installed", "git", combined, cells=comparable)

    # Oak-vs-Oak resolution: what an oak changeset comparison can detect here.

    self_mde: Dict[Tuple[str, str], Optional[float]] = {}

    if "oak" in floors:

        aa = cellstats.classify_cells(measured, "oak_installed_aa", "oak_installed", floors["oak"])

        self_mde = {key: item.get("mde_pct") for key, item in aa.items()}

    cells = sorted({(r["scenario"], r["operation"]) for r in rows})

    report = []

    for key in cells:

        per_subject: Dict[str, Any] = {}

        for name in subjects:

            values = [r["elapsed_ms"] for r in measured if r["subject"] == name and (r["scenario"], r["operation"]) == key]

            fails = [r for r in rows if r["subject"] == name and (r["scenario"], r["operation"]) == key

                     and r.get("returncode") == 1]

            skips = [r for r in rows if r["subject"] == name and (r["scenario"], r["operation"]) == key

                     and r.get("returncode") == 77]

            per_subject[name] = {

                "n": len(values), "median_ms": round(statistics.median(values), 3) if values else None,

                "failures": len(fails), "skips": len(skips),

                "skip_reason": skips[0].get("skip_reason") if skips else None,

                "failure": (fails[0].get("failure") or "")[:200] if fails else None,

                "teardown_errors": sum(1 for r in rows if r["subject"] == name

                                       and (r["scenario"], r["operation"]) == key and r.get("teardown_error")),

            }

        item = cross.get(key, {})

        cell_rows = [r for r in rows if (r["scenario"], r["operation"]) == key and r.get("returncode") == 0]

        report.append({

            "scenario": key[0], "operation": key[1], "subjects": per_subject,

            "oak_vs_git_comparability": comparability[key],

            "git_aa_floor_pct": floors.get("git", {}).get(key), "oak_aa_floor_pct": floors.get("oak", {}).get(key),

            "oak_vs_git_delta_pct_dev_only": item.get("delta_pct"),

            "oak_vs_git_ci_pct_dev_only": [item.get("ci_low_pct"), item.get("ci_high_pct")] if item else None,

            "oak_vs_git_mde_pct_80": item.get("mde_pct"),

            "oak_vs_oak_mde_pct_80": self_mde.get(key),

            "verdict_suppressed": item.get("verdict"),

            "claim_eligible": bool(cell_rows) and all(r.get("claim_eligible") is True for r in cell_rows),

        })

    return report

def _fmt(value: Any, suffix: str = "") -> str:

    if value is None:

        return "-"

    if isinstance(value, float):

        return f"{value:.1f}{suffix}"

    return f"{value}{suffix}"

def summary_markdown(prepared: Prepared, report: List[Dict[str, Any]], meta: Dict[str, Any]) -> str:

    lines = [

        f"# Scale lane: {prepared.tier.tier_id} "

        f"({'calibrated quiet host' if meta.get('calibrated') else 'DEVELOPMENT ONLY - not a claim'})",

        "",

        f"- bench_id `{meta['bench_id']}`; runs {meta['runs']}; subjects {', '.join(meta['subjects'])}",

        f"- fixture `{prepared.tier.fixture_id}` refs_sha256 `{prepared.manifest['refs_sha256'][:16]}` "

        f"spec_sha256 `{prepared.manifest['spec_sha256'][:16]}`",

        f"- host load at start {meta.get('loadavg_start')} / end {meta.get('loadavg_end')}. Rows are claimable only "

        "with a validated quiet-host calibration record (oakbench.scale_calibration). MDE = smallest change "

        "detectable at 80% power here.",

        "",

        "| op | git ms | oak ms | oak/git (dev) | git A/A floor | oak A/A floor | oak-vs-git MDE | oak-vs-oak MDE | notes |",

        "|---|---:|---:|---:|---:|---:|---:|---:|---|",

    ]

    diagnostic_ops = {phase for phase in prepared.phases}

    for item in report:

        if item["operation"] in diagnostic_ops:

            continue

        g = item["subjects"].get("git", {})

        o = item["subjects"].get("oak_installed", {})

        ratio = None

        comparable = item.get("oak_vs_git_comparability", COMPARABLE) == COMPARABLE

        if comparable and g.get("median_ms") and o.get("median_ms"):

            ratio = o["median_ms"] / g["median_ms"]

        notes = [] if comparable else [f"oak-vs-git suppressed ({item['oak_vs_git_comparability']}); oak-vs-oak only"]

        for name, data in item["subjects"].items():

            if data.get("skip_reason"):

                notes.append(f"{name}: skip {data['skip_reason']}")

            elif data.get("failures"):

                notes.append(f"{name}: {data['failures']} fail ({data.get('failure')})")

            if data.get("teardown_errors"):

                notes.append(f"{name}: {data['teardown_errors']} untimed-teardown defect(s)")

        lines.append(

            f"| {item['operation']} | {_fmt(g.get('median_ms'))} | {_fmt(o.get('median_ms'))} | "

            f"{('n/c' if not comparable else _fmt(ratio)) if ratio is None else f'{ratio:.2f}x'} | {_fmt(item['git_aa_floor_pct'], '%')} | "

            f"{_fmt(item['oak_aa_floor_pct'], '%')} | {_fmt(item['oak_vs_git_mde_pct_80'], '%')} | "

            f"{_fmt(item['oak_vs_oak_mde_pct_80'], '%')} | {'; '.join(notes)[:300]} |")

    lines += ["", "## Fixture preparation (untimed, seconds)", ""]

    for key, value in prepared.phases.items():

        lines.append(f"- {key}: {value}")

    if prepared.notes:

        lines += ["", "## Notes", ""] + [f"- {note}" for note in prepared.notes]

    return "\n".join(lines) + "\n"

def ingest_rows(prepared: Prepared, bench_id: str, runner_fields: Dict[str, Any]) -> List[Dict[str, Any]]:

    """One diagnostic row per preparation phase (setup cost is scale evidence)."""

    rows = []

    oak_identity = getattr(prepared, "oak_identity", None) or {}

    for phase_name, seconds in prepared.phases.items():

        if phase_name.endswith("_reused"):

            continue

        kind = "oak" if ".oak_" in phase_name else ("git" if ".git_" in phase_name else "harness")

        rows.append({

            "bench_id": bench_id, "profile": "scale", "scale_row_version": SCALE_ROW_VERSION,

            "scenario": prepared.tier.tier_id, "operation": phase_name, "run": 0,

            "subject": {"oak": "oak_installed", "git": "git"}.get(kind, "harness"), "subject_kind": kind,

            "command": [], "elapsed_ms": round(seconds * 1000.0, 3), "returncode": 0, "diagnostic": True,

            "claim_ineligible_reasons": ["untimed preparation phase (diagnostic)"],

            "measurement_source": "harness_phase_wall_clock", "evidence_class": EVIDENCE_CLASS,

            "claim_eligible": False, **prepared.fixture_fields(), **runner_fields,

            # Oak rows carry binary identity (publish_gate conformance check).

            **(oak_identity if kind == "oak" else {}),

        })

    return rows