#!/usr/bin/env python3
"""Resumable 34/34-capability maximum-scale stress harness."""

import hashlib
import json
import os
import platform
import signal
import subprocess
import sys
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
RUNNER = HERE / "universal_vrp_all34_stress_case.js"
RESULTS = HERE / "universal_vrp_all34_max_stress_results.json"
REPO = Path("/Users/ratiomoralis/routing_engine")
SOURCE_FILES = [
    REPO / "vrp_saas/lib/universal-constrained-vrp/ils.js",
    REPO / "vrp_saas/lib/universal-constrained-vrp-production/index.js",
    REPO / "vrp_saas/lib/universal-constrained-vrp-verifier/verifier.js",
    REPO / "vrp_saas/lib/production-ils-runtime/index.js",
    REPO / "vrp_saas/lib/production-ils-runtime/adapter.js",
    # Il caso di prova decide che cosa significhi "famiglia attiva": senza la
    # sua impronta la misura non sarebbe legata al criterio con cui e' stata
    # ottenuta, e un criterio piu' permissivo passerebbe inosservato.
    REPO / "universal_vrp_all34_stress_case.js",
]
TIMEOUT_S = 300.0
ITERATION_LIMIT = 100
SEEDS = (11, 29)

AXES = {
    "all34_combined": {
        "mode": "combined", "coarse": (1, 2, 4, 8, 12, 16, 20, 24, 28, 32, 36, 40),
        "resolution": 2,
    },
    "all34_customer_fixed10": {
        "mode": "customer", "coarse": (8, 16, 32, 48, 64, 80, 96, 112, 128, 144),
        "resolution": 8,
    },
    "all34_vehicle_frontier": {
        "mode": "vehicle", "coarse": (8, 16, 32, 64, 128, 256, 384, 512, 640, 768, 896),
        "resolution": 32,
    },
    # Quarto asse: pochi mezzi, molte fermate ciascuno.
    # I tre assi sopra delimitano il motore ma nessuno ha la forma di un giro di
    # distribuzione reale: 288 clienti su 72 mezzi fanno quattro fermate a mezzo,
    # 88 su 10 ne fanno nove. Un furgone vero ne fa venti o quaranta. Qui i mezzi
    # generici restano tre — cinque col nucleo completo — e crescono solo i clienti,
    # quindi cresce il numero di fermate per mezzo. I livelli sono multipli di 24
    # perche' devono essere divisibili sia per 8 (vincolo della fixture) sia per 3.
    "all34_real_round_3v": {
        "mode": "customer", "fixedVehicles": 3,
        "coarse": (24, 72, 120, 168, 216, 264),
        "resolution": 24,
    },
}


def sha256(path):
    value = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            value.update(chunk)
    return value.hexdigest()


def initial_data():
    return {
        "completed": False,
        "metadata": {
            "machine": platform.platform(), "python": sys.version,
            "node": subprocess.check_output(["node", "--version"], text=True).strip(),
            "cpuCount": os.cpu_count(), "technicalTimeoutS": TIMEOUT_S,
            "iterationLimit": ITERATION_LIMIT, "solverSeeds": list(SEEDS),
            "providerCalls": 0, "matrix": "deterministic synthetic offline",
            "requiredActiveConstraintFamilies": 34,
            "sourceSha256": {str(path): sha256(path) for path in SOURCE_FILES},
        },
        "runs": [], "axes": {},
    }


def load():
    return json.loads(RESULTS.read_text()) if RESULTS.exists() else initial_data()


def save(data):
    temporary = RESULTS.with_suffix(".tmp")
    temporary.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
    os.replace(temporary, RESULTS)


def spec_for(mode, level, seed, repetition, fixed_vehicles=None):
    spec = {"mode": mode, "solverSeed": seed, "repetition": repetition,
            "iterationLimit": ITERATION_LIMIT}
    if mode == "combined": spec["cells"] = level
    elif mode == "customer":
        spec["freeJobs"] = level
        # Assente per gli assi storici: il runner mantiene il valore di sempre,
        # otto mezzi generici, quindi le misure gia' eseguite non cambiano.
        if fixed_vehicles is not None: spec["fixedVehicles"] = fixed_vehicles
    elif mode == "vehicle": spec["genericVehicles"] = level
    else: raise ValueError(mode)
    return spec


def case_key(axis, level, seed, repetition):
    return f"{axis}:{level}:{seed}:{repetition}"


def run_case(axis, mode, level, seed, repetition, data, fixed_vehicles=None):
    key = case_key(axis, level, seed, repetition)
    for old in data["runs"]:
        if old.get("caseKey") == key:
            print(json.dumps({"resumed": key, "status": old["status"]}), flush=True)
            return old
    spec = spec_for(mode, level, seed, repetition, fixed_vehicles)
    started = time.monotonic()
    process = subprocess.Popen(
        ["node", str(RUNNER), json.dumps(spec, separators=(",", ":"))],
        cwd=str(HERE), stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        text=True, start_new_session=True,
    )
    terminal = None
    while process.poll() is None:
        if time.monotonic() - started >= TIMEOUT_S:
            terminal = "TIMEOUT"
            os.killpg(process.pid, signal.SIGTERM)
            try: process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                os.killpg(process.pid, signal.SIGKILL)
                process.wait()
            break
        time.sleep(0.25)
    stdout, stderr = process.communicate()
    wall = time.monotonic() - started
    if terminal:
        row = {"caseKey": key, "axis": axis, "level": level, "mode": mode,
               "solverSeed": seed, "repetition": repetition, "status": terminal,
               "wallS": wall, "iterationLimit": ITERATION_LIMIT}
    else:
        lines = [line for line in stdout.splitlines() if line.strip()]
        try: payload = json.loads(lines[-1])
        except Exception:
            payload = {"status": "HARNESS_ERROR", "stdoutTail": stdout[-2000:],
                       "stderrTail": stderr[-2000:]}
        row = {"caseKey": key, "axis": axis, "level": level, **payload,
               "wallS": wall, "exitCode": process.returncode}
        if stderr.strip(): row["stderrTail"] = stderr[-2000:]
    data["runs"].append(row)
    save(data)
    print(json.dumps(row, sort_keys=True), flush=True)
    return row


def status_at(data, axis, level, repetitions=(0,)):
    rows = [row for row in data["runs"] if row.get("axis") == axis and
            row.get("level") == level and row.get("repetition") in repetitions]
    return rows


def explore_axis(axis, config, data):
    mode, coarse, resolution = config["mode"], config["coarse"], config["resolution"]
    fixed_vehicles = config.get("fixedVehicles")
    seed11_pass = []
    first_seed11_fail = None
    for level in coarse:
        row = run_case(axis, mode, level, SEEDS[0], 0, data, fixed_vehicles)
        if row["status"] == "PASS": seed11_pass.append(level)
        else:
            first_seed11_fail = level
            break
    if not seed11_pass:
        raise RuntimeError(f"{axis}: no passing level")
    if first_seed11_fail is None:
        raise RuntimeError(f"{axis}: frontier not reached by declared coarse levels")

    low = max(seed11_pass)
    for level in range(low + resolution, first_seed11_fail, resolution):
        row = run_case(axis, mode, level, SEEDS[0], 0, data, fixed_vehicles)
        if row["status"] == "PASS": low = level
        else:
            first_seed11_fail = level
            break

    # CORREZIONE 14/08/2026 — che cosa vuol dire "stabile".
    #
    # Fino a questa data il commento diceva 4/4 e il codice faceva 2/2: le due
    # ripetizioni di conferma venivano eseguite e il loro esito non veniva mai
    # letto, quindi "stabile" significava soltanto "i due semi passano alla
    # prima ripetizione". Il quarto asse ha reso visibile il difetto — il
    # livello 72 usciva PASS, PASS, PASS, TIMEOUT e veniva riportato come
    # massimo stabile — e la pagina "capacita' e limiti" dichiara invece,
    # testualmente, quattro esecuzioni riuscite su quattro.
    #
    # La regola e' asimmetrica, e volutamente: per DICHIARARE stabile un
    # livello servono quattro esecuzioni riuscite su quattro; per SCARTARLO
    # bastano i due fallimenti della prima ripetizione. Affermare costa piu'
    # che negare.
    def prova_due(level):
        """Le due prime esecuzioni, un seme ciascuna. Bastano a scartare."""
        return [run_case(axis, mode, level, seed, 0, data, fixed_vehicles)["status"]
                for seed in SEEDS].count("PASS")

    def prova_quattro(level):
        """Tutte e quattro le esecuzioni: due semi per due ripetizioni. Vengono
           eseguite tutte anche dopo il primo fallimento, perche' "tre su
           quattro" e' un risultato da scrivere, non da dedurre da un'assenza."""
        return [run_case(axis, mode, level, seed, repetition, data, fixed_vehicles)["status"]
                for seed in SEEDS for repetition in (0, 1)].count("PASS")

    # Search downward for the highest level with four completions out of four.
    stable = None
    marginal = []
    for level in range(low, 0, -resolution):
        if prova_due(level) < 2: continue
        if prova_quattro(level) == 4:
            stable = level
            break
        marginal.append(level)
    if stable is None:
        raise RuntimeError(f"{axis}: nessun livello con quattro esecuzioni riuscite su quattro")

    # Scan upward at declared resolution until both seeds timeout/fail.
    maximum_completed = max([stable] + marginal)
    first_stable_failure = None
    level = stable + resolution
    hard_ceiling = max(coarse) + resolution * 4
    while level <= hard_ceiling:
        if prova_due(level) == 0:
            first_stable_failure = level
            break
        passati = prova_quattro(level)
        maximum_completed = max(maximum_completed, level)
        if passati == 4: stable = level
        elif level not in marginal: marginal.append(level)
        level += resolution
    marginal = sorted(set(marginal))
    if first_stable_failure is None:
        raise RuntimeError(f"{axis}: stable failure not reached")

    result = {
        "mode": mode, "resolution": resolution, "fixedVehicles": fixed_vehicles,
        "maximumStableLevel": stable,
        "maximumCompletedAtLeastOnceLevel": maximum_completed,
        "marginalLevels": marginal, "firstStableFailureLevel": first_stable_failure,
    }
    data["axes"][axis] = result
    save(data)
    return result


def main():
    data = load()
    data["completed"] = False
    save(data)
    for axis, config in AXES.items():
        explore_axis(axis, config, data)
    data["completed"] = True
    save(data)
    print(json.dumps({"completed": True, "axes": data["axes"]}, sort_keys=True), flush=True)


if __name__ == "__main__":
    main()