#!/usr/bin/env python3
"""Observe HDF5 process exit without close; not power loss. Private files only."""
import argparse
from datetime import datetime, timezone
import hashlib
import json
import multiprocessing as mp
import os
from pathlib import Path
import platform
import subprocess
import tempfile

import h5py
import numpy as np

CASES = ("clean-published", "exit-unpublished", "exit-published", "exit-non-swmr")
TIMEOUT = 20
EXIT_WITHOUT_CLOSE = 73


def write_case(path, case):
    if case not in CASES:
        raise ValueError("Unknown experiment case")
    with h5py.File(path, "x", libver="latest", driver="sec2") as f:
        x = f.create_dataset("x", (0,), maxshape=(None,), chunks=(1,), dtype="i8")
        y = f.create_dataset("y", (0,), maxshape=(None,), chunks=(1,), dtype="i8")
        mark = f.create_dataset("published", data=np.array([0], dtype="u8"), chunks=(1,))
        if case != "exit-non-swmr":
            f.swmr_mode = True
        for dataset, value in [(x, 10), (y, 100)]:
            dataset.resize((1,))
            dataset[0] = value
            dataset.flush()
        if case != "exit-unpublished":
            mark[0] = 1
            mark.flush()
        f.flush()
        if case != "clean-published":
            # Intentionally bypass HDF5 close, Python cleanup and atexit.
            os._exit(EXIT_WITHOUT_CLOSE)


def read_case(path, swmr):
    try:
        with h5py.File(path, "r", libver="latest", driver="sec2", swmr=swmr) as f:
            x, y = f["x"][:].tolist(), f["y"][:].tolist()
            n = int(f["published"][0])
            if not 0 <= n <= min(len(x), len(y)):
                raise RuntimeError("Publication watermark exceeds paired payload")
            return {"opened": True, "x": x, "y": y, "watermark": n,
                    "selectedPrefix": [[int(a), int(b)] for a, b in zip(x[:n], y[:n])]}
    except OSError as error:
        return {"opened": False, "errorType": type(error).__name__,
                "error": str(error).replace(str(path), "<private-file>")}


def status_flags(header):
    # Only the no-user-block v3 files created by this harness are supported.
    if len(header) != 12 or header[:8] != b"\x89HDF\r\n\x1a\n" or header[8] != 3:
        raise RuntimeError("Expected a no-user-block HDF5 version-3 superblock")
    return header[11]


def validate_case(row):
    case = row["case"]
    if case not in CASES:
        raise RuntimeError("Unknown case")
    clean = case == "clean-published"
    swmr = case != "exit-non-swmr"
    expected_flags = 0 if clean else (5 if swmr else 1)
    if type(row["writerExitCode"]) is not int or row["writerExitCode"] != (0 if clean else EXIT_WITHOUT_CLOSE):
        raise RuntimeError("Writer did not reach the intended exit boundary")
    if type(row["statusFlags"]) is not int or row["statusFlags"] != expected_flags:
        raise RuntimeError("Unexpected superblock flags; review this environment")
    for name in ("ordinaryRead", "swmrRead"):
        result = row[name]
        expected_open = clean or (name == "swmrRead" and swmr)
        if result.get("opened") is not expected_open:
            raise RuntimeError("Unexpected " + name + " result")
        if expected_open:
            n = 0 if case == "exit-unpublished" else 1
            expected = {"opened": True, "x": [10], "y": [100], "watermark": n,
                        "selectedPrefix": [] if n == 0 else [[10, 100]]}
            if result != expected:
                raise RuntimeError("Read values differ from the complete oracle")
        elif result.get("errorType") != "OSError" or not result.get("error"):
            raise RuntimeError("Expected a recorded HDF5 open error")


def trial(directory, case):
    path = Path(directory) / "observation.h5"
    process = mp.get_context("spawn").Process(target=write_case, args=(str(path), case))
    process.start()
    try:
        process.join(TIMEOUT)
        if process.is_alive():
            raise TimeoutError("Writer did not finish within bounded wait")
        with path.open("rb") as stream:
            flags = status_flags(stream.read(12))
        row = {"case": case, "writerExitCode": process.exitcode, "statusFlags": flags,
               "ordinaryRead": read_case(path, False), "swmrRead": read_case(path, True)}
        validate_case(row)
        return row
    finally:
        if process.is_alive():
            process.terminate()
            process.join(TIMEOUT)
            if process.is_alive():
                process.kill()
                process.join(TIMEOUT)
        process.close()


def run(repetitions, temp_dir=None):
    if type(repetitions) is not int or not 1 <= repetitions <= 100:
        raise ValueError("Repetitions must be an integer from 1 to 100")
    rows = []
    with tempfile.TemporaryDirectory(prefix="hesela-hdf5-exit-", dir=temp_dir) as directory:
        filesystem = subprocess.check_output(["stat", "-f", "-c", "%T", directory], text=True).strip()
        for repetition in range(1, repetitions + 1):
            for case in CASES:
                with tempfile.TemporaryDirectory(dir=directory) as case_dir:
                    rows.append({"repetition": repetition, **trial(case_dir, case)})
    return {
        "schemaVersion": 1, "experiment": "hdf5-process-exit",
        "evidenceType": "measured-api-behavior",
        "generatedAt": datetime.now(timezone.utc).isoformat(),
        "repetitions": repetitions, "casesPerRepetition": len(CASES),
        "sourceSha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "requirementsSha256": hashlib.sha256(Path(__file__).with_name("hdf5-swmr-requirements.txt").read_bytes()).hexdigest(),
        "environment": {
            "python": platform.python_version(), "kernel": platform.release(),
            "machine": platform.machine(), "h5py": h5py.__version__,
            "hdf5": h5py.version.hdf5_version, "numpy": np.__version__,
            "filesystem": filesystem, "driver": "sec2", "libver": "latest",
            "multiprocessing": "spawn",
            "HDF5_USE_FILE_LOCKING": os.environ.get("HDF5_USE_FILE_LOCKING"),
        },
        "method": "Four fresh files per repetition. A spawned writer flushes two integer payloads and optionally a publication count, then closes normally or calls os._exit(73). After join, the parent reads the v3 superblock flag byte and attempts fresh ordinary and SWMR read-only opens. No writer is active during either read.",
        "limits": "One host and selected post-flush boundaries only. No power loss, reboot, kernel crash, torn write, fsync trace, concurrent writer, in-flight I/O interruption or recovery-tool execution. OS and device caches remain available. Synthetic inputs are not sensor measurements; repeats do not estimate failure probabilities. The header parser supports only this harness's no-user-block version-3 files.",
        "license": "CC0-1.0 for original harness and observations; upstream licenses unchanged.",
        "observations": rows,
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repetitions", type=int, default=3)
    parser.add_argument("--temp-dir")
    args = parser.parse_args()
    if not 1 <= args.repetitions <= 100:
        parser.error("--repetitions must be between 1 and 100")
    print(json.dumps(run(args.repetitions, args.temp_dir), indent=2))


if __name__ == "__main__":
    main()