"""Quiet-host calibration for the scale lane: evidence verified by recomputation.

A scale row's ``claim_eligible`` / ``evidence_class`` are DERIVED, never a

free flag, and nothing a calibration record *states* is trusted:

1. ``run_scale_suite.py calibrate`` runs an identical-binary A/A campaign

   (git, git_aa, oak_installed, oak_installed_aa) plus host probes (load

   average, small-file creation rate, F_FULLFSYNC latency) and writes its rows

   to a JSONL file. The record names that file's content sha256.

2. Publishing requires that calibration JSONL among publish_gate's inputs,

   matched by content sha256. The gate RECOMPUTES the A/A cell count, per-cell

   floors (oakbench.cellstats, the P1 machinery), identical-binary false

   claims, host load and disk probes from those rows, and checks that the

   calibration rows and the published rows share runner identity, host

   cpu_count and physical memory, the Oak and Git binary sha256, and

   registry-pinned fixtures. The published run must carry its own quiet disk

   probes too.

This detects edited, resealed, copied or invented records (the sealed digest

is only an integrity check of the record's own content). It cannot detect a

fabricated calibration JSONL with invented timings; that would need a signed

runner, which the lane does not have. Do not describe it as forgery-proof.

"""

from __future__ import annotations

import hashlib

import json

import os

import statistics

import time

from pathlib import Path

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

SCHEMA_VERSION = 2

KIND = "oakbench_quiet_host_calibration"

CALIBRATED_EVIDENCE_CLASS = "calibrated_quiet_host"

DEVELOPMENT_EVIDENCE_CLASS = "development_only_loaded_host"

POLICY_VERSION = "2026-09-29.2"

# Quiet-host policy (provisional; re-derive on the dedicated runner and bump

# POLICY_VERSION). Load/core 0.1 is ~1.4 runnable threads on a 14-core host.

MAX_LOAD_PER_CORE = 0.10

MIN_AA_CELLS = 20

MIN_RUNS = 5

MAX_AA_FLOOR_MEDIAN_PCT = 10.0

# Disk contention (load average does not see it; Oak clone/switch are

# fsync-bound, fb-549). Measured on the loaded dev host: F_FULLFSYNC median

# 3.7 ms, file creation as low as 170 files/s.

MAX_FULLFSYNC_MEDIAN_MS = 2.5

MIN_CREATE_FILES_PER_S = 2000.0

PROBE_OPERATIONS = ("host.disk_fullfsync", "host.file_create_rate")

AA_PAIRS = (("git", "git_aa"), ("oak_installed", "oak_installed_aa"))

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

# Host facts and probes

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

def host_facts() -> Dict[str, Any]:

    try:

        mem = int(os.sysconf("SC_PAGE_SIZE")) * int(os.sysconf("SC_PHYS_PAGES"))

    except (ValueError, OSError, AttributeError):

        mem = None

    return {"host_cpu_count": os.cpu_count(), "host_mem_bytes": mem}

def disk_probe(directory: Path, samples: int = 20, files: int = 500) -> Dict[str, Any]:

    """F_FULLFSYNC (fsync elsewhere) latency of a 4 KiB write, and small-file creation rate."""

    import fcntl

    directory = Path(directory)

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

    full = getattr(fcntl, "F_FULLFSYNC", None)

    latencies = []

    path = directory / ".oakbench-fsync-probe"

    for _ in range(samples):

        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)

        try:

            os.write(fd, os.urandom(4096))

            start = time.perf_counter()

            if full is not None:

                fcntl.fcntl(fd, full)

            else:

                os.fsync(fd)

            latencies.append((time.perf_counter() - start) * 1000.0)

        finally:

            os.close(fd)

    path.unlink()

    tree = directory / ".oakbench-create-probe"

    tree.mkdir(exist_ok=True)

    start = time.perf_counter()

    for index in range(files):

        with open(tree / f"f{index}", "wb") as handle:

            handle.write(b"x" * 64)

    elapsed = time.perf_counter() - start

    for index in range(files):

        (tree / f"f{index}").unlink()

    tree.rmdir()

    return {"fullfsync_ms": [round(v, 4) for v in latencies],

            "fullfsync_kind": "F_FULLFSYNC" if full is not None else "fsync",

            "create_files": files, "create_elapsed_s": round(elapsed, 6)}

def probe_rows(probe: Dict[str, Any], base: Dict[str, Any], phase: str) -> List[Dict[str, Any]]:

    """Diagnostic rows carrying probe samples (so the gate can recompute them)."""

    common = {**base, "subject": "harness", "subject_kind": "harness", "command": [], "returncode": 0,

              "diagnostic": True, "probe_phase": phase, "measurement_source": "harness_phase_wall_clock",

              "evidence_class": DEVELOPMENT_EVIDENCE_CLASS, "claim_eligible": False,

              "claim_ineligible_reasons": ["host probe (diagnostic)"], **host_facts()}

    rows = [dict(common, operation="host.disk_fullfsync", run=index, elapsed_ms=value,

                 fsync_kind=probe["fullfsync_kind"]) for index, value in enumerate(probe["fullfsync_ms"])]

    rate = probe["create_files"] / probe["create_elapsed_s"] if probe["create_elapsed_s"] > 0 else 0.0

    rows.append(dict(common, operation="host.file_create_rate", run=0,

                     elapsed_ms=round(probe["create_elapsed_s"] * 1000.0, 3), files=probe["create_files"],

                     files_per_s=round(rate, 1)))

    return rows

def probe_verdict(rows: Iterable[Dict[str, Any]]) -> Tuple[List[str], Dict[str, Any]]:

    rows = list(rows)

    fsyncs = [r["elapsed_ms"] for r in rows if r.get("operation") == "host.disk_fullfsync"

              and isinstance(r.get("elapsed_ms"), (int, float))]

    rates = [r.get("files_per_s") for r in rows if r.get("operation") == "host.file_create_rate"]

    reasons, facts = [], {}

    if len(fsyncs) < 10 or not rates:

        return ["disk probes missing"], facts

    facts["fullfsync_median_ms"] = statistics.median(fsyncs)

    facts["create_files_per_s_min"] = min(r for r in rates if isinstance(r, (int, float)))

    if facts["fullfsync_median_ms"] > MAX_FULLFSYNC_MEDIAN_MS:

        reasons.append(f"disk contended: F_FULLFSYNC median {facts['fullfsync_median_ms']:.2f} ms > "

                       f"{MAX_FULLFSYNC_MEDIAN_MS} ms")

    if facts["create_files_per_s_min"] < MIN_CREATE_FILES_PER_S:

        reasons.append(f"disk contended: {facts['create_files_per_s_min']:.0f} files/s < {MIN_CREATE_FILES_PER_S:.0f}")

    return reasons, facts

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

# Records

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

def file_sha256(path: Path) -> str:

    digest = hashlib.sha256()

    with Path(path).open("rb") as handle:

        for chunk in iter(lambda: handle.read(1 << 20), b""):

            digest.update(chunk)

    return digest.hexdigest()

def record_digest(record: Dict[str, Any]) -> str:

    body = {key: value for key, value in record.items() if key != "record_sha256"}

    return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest()

def seal(record: Dict[str, Any]) -> Dict[str, Any]:

    sealed = dict(record)

    sealed.update(schema_version=SCHEMA_VERSION, kind=KIND, policy_version=POLICY_VERSION)

    sealed["record_sha256"] = record_digest(sealed)

    return sealed

def load_rows(path: Path) -> List[Dict[str, Any]]:

    return [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()]

def _pin_problems(rows: Iterable[Dict[str, Any]], registry: Optional[Dict[str, Any]]) -> List[str]:

    if registry is None:

        return []

    pins = {entry.get("spec_sha256"): entry.get("refs_sha256") for entry in registry.values()

            if isinstance(entry, dict) and entry.get("spec_sha256")}

    bad = set()

    for row in rows:

        if row.get("subject_kind") == "harness" or row.get("diagnostic"):

            continue

        spec, refs = row.get("fixture_spec_sha256"), row.get("fixture_manifest_sha256")

        if not spec or pins.get(spec) is None or pins.get(spec) != refs:

            bad.add(str(row.get("fixture_id")))

    return [f"fixture not registry-pinned or pin mismatch: {sorted(bad)[:3]}"] if bad else []

def campaign_identity(rows: Sequence[Dict[str, Any]]) -> Tuple[Dict[str, Any], List[str]]:

    """Runner/host/binary identity shared by every row of a run (or problems)."""

    reasons: List[str] = []

    identity: Dict[str, Any] = {}

    for key in ("runner_id", "host_cpu_count", "host_mem_bytes"):

        values = {row.get(key) for row in rows}

        if len(values) != 1 or None in values:

            reasons.append(f"rows disagree on or lack {key}: {sorted(map(str, values))[:3]}")

        else:

            identity[key] = values.pop()

    for kind in ("git", "oak"):

        shas = {row.get("binary_sha256") for row in rows if row.get("subject_kind") == kind

                and not row.get("diagnostic")}

        shas.discard(None)

        if len(shas) > 1:

            reasons.append(f"{kind} rows name several binaries")

        identity[f"{kind}_binary_sha256"] = shas.pop() if len(shas) == 1 else None

    return identity, reasons

def recompute(rows: Sequence[Dict[str, Any]], registry: Optional[Dict[str, Any]] = None

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

    """Recompute every calibration number from the campaign's own rows."""

    from oakbench import cellstats

    identity, reasons = campaign_identity(rows)

    measured = [r for r in rows if r.get("profile") == "scale" and r.get("returncode") == 0

                and not r.get("diagnostic") and isinstance(r.get("elapsed_ms"), (int, float))]

    value = lambda r: r.get("elapsed_ms")  # noqa: E731

    floors: List[float] = []

    false_claims = 0

    for arm, control in AA_PAIRS:

        runs = {r.get("run") for r in measured if r.get("subject") == arm}

        if len(runs) < MIN_RUNS:

            reasons.append(f"{arm}: fewer than {MIN_RUNS} runs")

            continue

        cell = cellstats.cell_floors(measured, arm, control, value)

        floors.extend(cell.values())

        verdicts = cellstats.classify_cells(measured, control, arm, cell, value_of=value)

        false_claims += sum(1 for item in verdicts.values() if item.get("verdict") in ("regression", "improvement"))

    cpus = identity.get("host_cpu_count")

    loads = [r["environment_sample"]["loadavg"][0] for r in rows

             if isinstance((r.get("environment_sample") or {}).get("loadavg"), list)]

    load_per_core = max(loads) / cpus if loads and isinstance(cpus, int) and cpus > 0 else None

    disk_reasons, disk = probe_verdict(rows)

    facts = {**identity, "aa_cells": len(floors),

             "aa_floor_median_pct": round(statistics.median(floors), 3) if floors else None,

             "aa_false_claims": false_claims,

             "loadavg_per_core_max": round(load_per_core, 4) if load_per_core is not None else None, **disk}

    if len(floors) < MIN_AA_CELLS:

        reasons.append(f"too few A/A cells ({len(floors)} < {MIN_AA_CELLS})")

    if facts["aa_floor_median_pct"] is None or facts["aa_floor_median_pct"] > MAX_AA_FLOOR_MEDIAN_PCT:

        reasons.append(f"A/A floor median {facts['aa_floor_median_pct']}% > {MAX_AA_FLOOR_MEDIAN_PCT}%")

    if false_claims:

        reasons.append(f"identical-binary A/A produced {false_claims} claim(s)")

    if load_per_core is None or load_per_core > MAX_LOAD_PER_CORE:

        reasons.append(f"host not quiet: load/core {load_per_core} > {MAX_LOAD_PER_CORE}")

    reasons += disk_reasons

    reasons += _pin_problems(rows, registry)

    return facts, reasons

def verify_campaign(record: Any, calibration_rows: Optional[Sequence[Dict[str, Any]]],

                    calibration_sha256: Optional[str], registry: Optional[Dict[str, Any]] = None

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

    """Validate a record against its recomputed campaign; returns (facts, reasons)."""

    if not isinstance(record, dict):

        return {}, ["calibration record missing or not an object"]

    reasons = []

    if record.get("schema_version") != SCHEMA_VERSION or record.get("kind") != KIND:

        reasons.append("wrong schema_version/kind")

    if record.get("record_sha256") != record_digest(record):

        reasons.append("record_sha256 does not match content (edited record)")

    if calibration_rows is None:

        return {}, reasons + ["calibration campaign rows not provided (need the JSONL named by the record)"]

    if not calibration_sha256 or calibration_sha256 != record.get("calibration_rows_sha256"):

        return {}, reasons + ["calibration JSONL content sha256 does not match the record"]

    facts, problems = recompute(calibration_rows, registry)

    reasons += problems

    for key in ("runner_id", "host_cpu_count", "host_mem_bytes", "oak_binary_sha256", "git_binary_sha256",

                "aa_cells", "aa_false_claims"):

        if record.get(key) != facts.get(key):

            reasons.append(f"record {key} {record.get(key)!r} != recomputed {facts.get(key)!r}")

    return facts, reasons

def row_problems(row: Dict[str, Any], facts: Dict[str, Any]) -> List[str]:

    """A published row must come from the calibrated runner, host and binaries, on a quiet host."""

    reasons = []

    for key in ("runner_id", "host_cpu_count", "host_mem_bytes"):

        if row.get(key) != facts.get(key):

            reasons.append(f"row {key} {row.get(key)!r} != calibration {facts.get(key)!r}")

    kind = row.get("subject_kind")

    if kind in ("git", "oak") and row.get("binary_sha256") != facts.get(f"{kind}_binary_sha256"):

        reasons.append(f"row {kind} binary differs from the calibrated binary")

    sample = row.get("environment_sample") or {}

    load = sample.get("loadavg") if isinstance(sample, dict) else None

    cpus = facts.get("host_cpu_count")

    if not isinstance(load, list) or not load or not isinstance(cpus, int) or load[0] / cpus > MAX_LOAD_PER_CORE:

        reasons.append("row host load/core not quiet")

    if row.get("fixture_pin_status") != "verified":

        reasons.append("row fixture not verified against its registry pin")

    if row.get("returncode") != 0:

        reasons.append("row is not a successful measurement")

    return reasons

def stamp(row: Dict[str, Any], record: Optional[Dict[str, Any]], facts: Optional[Dict[str, Any]],

          campaign_reasons: Sequence[str]) -> Dict[str, Any]:

    """Derive evidence_class / claim_eligible for one row of a run."""

    if record is None:

        reasons = ["no calibration record"]

    else:

        reasons = list(campaign_reasons) or row_problems(row, facts or {})

    if reasons:

        row["evidence_class"] = DEVELOPMENT_EVIDENCE_CLASS

        row["claim_eligible"] = False

        row["claim_ineligible_reasons"] = reasons[:5]

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

            row.pop(key, None)

    else:

        row["evidence_class"] = CALIBRATED_EVIDENCE_CLASS

        row["claim_eligible"] = True

        row["claim_ineligible_reasons"] = []

        row["calibration_record"] = record

        row["calibration_record_sha256"] = record["record_sha256"]

    return row

def gate(scale_rows: Sequence[Tuple[str, Dict[str, Any]]], inputs: Dict[str, List[Dict[str, Any]]],

         input_sha256: Dict[str, str], registry: Optional[Dict[str, Any]]) -> List[Tuple[str, Any, Any, Any, str]]:

    """publish_gate: every publishable scale row is re-verified by recomputation.

    ``inputs``/``input_sha256`` map each input path to its rows and content

    sha256; the calibration JSONL must be one of them. Returns refusals.

    """

    by_sha = {sha: path for path, sha in input_sha256.items()}

    # Inputs that are the calibration campaigns referenced by rows are evidence

    # FOR the gate, not measurements being published.

    campaign_paths = {by_sha.get((row.get("calibration_record") or {}).get("calibration_rows_sha256"))

                      for _, row in scale_rows if isinstance(row.get("calibration_record"), dict)}

    campaign_paths.discard(None)

    verified: Dict[str, Tuple[Dict[str, Any], List[str]]] = {}

    run_probe_cache: Dict[Tuple[str, str], List[str]] = {}

    refused = []

    for path, row in scale_rows:

        if path in campaign_paths:

            continue

        if row.get("diagnostic") or row.get("returncode") == 77 or row.get("subject_kind") == "harness":

            continue

        where = (path, row.get("scenario"), row.get("operation"), row.get("subject"))

        if row.get("evidence_class") != CALIBRATED_EVIDENCE_CLASS or row.get("claim_eligible") is not True:

            refused.append((*where, f"evidence_class {row.get('evidence_class')!r} is not calibrated"))

            continue

        record = row.get("calibration_record")

        if not isinstance(record, dict) or row.get("calibration_record_sha256") != record.get("record_sha256"):

            refused.append((*where, "calibration record missing or its digest does not match the row"))

            continue

        key = record.get("record_sha256")

        if key not in verified:

            cal_sha = record.get("calibration_rows_sha256")

            cal_path = by_sha.get(cal_sha)

            verified[key] = verify_campaign(record, inputs.get(cal_path) if cal_path else None,

                                            cal_sha if cal_path else None, registry)

        facts, reasons = verified[key]

        if reasons:

            refused.append((*where, reasons[0]))

            continue

        problems = row_problems(row, facts)

        probe_key = (path, str(row.get("bench_id")))

        if probe_key not in run_probe_cache:

            run_rows = [r for r in inputs.get(path, []) if r.get("bench_id") == row.get("bench_id")]

            run_probe_cache[probe_key] = probe_verdict(run_rows)[0]

        problems += [f"published run: {reason}" for reason in run_probe_cache[probe_key]]

        if problems:

            refused.append((*where, problems[0]))

    return refused