from __future__ import annotations

import hashlib
import json
import os
import pathlib
import sys
import time
from typing import Any

from .process_control import run_supervised_process


ROOT = pathlib.Path(__file__).resolve().parents[2]
PORTABLE_RUN_FIELDS = {
    "run_id", "platform", "python_version", "source_tree_sha256", "command",
    "started_epoch", "ended_epoch", "exit_code", "timed_out",
    "orphan_process_count", "zombie_process_count", "test_count",
    "stdout_sha256", "stderr_sha256", "run_ordinal", "cold_start", "source_origin",
}


def _digest(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def deadlock_payload_bytes(pipe_capacity_bytes: int = 65536) -> int:
    """A semantic pipe-pressure payload, deliberately not a throughput benchmark."""
    if pipe_capacity_bytes <= 0:
        raise ValueError("pipe_capacity_bytes must be positive")
    return min(512 * 1024, max(256 * 1024, 4 * int(pipe_capacity_bytes)))


def validate_portable_run(record: dict[str, Any], expected_source_hash: str, expected_origin: str) -> dict[str, Any]:
    missing = sorted(PORTABLE_RUN_FIELDS - set(record))
    checks = {
        "required_fields": not missing,
        "source_hash": record.get("source_tree_sha256") == expected_source_hash and len(expected_source_hash) == 64,
        "source_origin": record.get("source_origin") == expected_origin,
        "cold_process": record.get("cold_start") is True,
        "bounded_and_clean": record.get("timed_out") is False and record.get("orphan_process_count") == 0 and record.get("zombie_process_count") == 0,
        "raw_digests": all(isinstance(record.get(field), str) and len(record[field]) == 64 for field in ("stdout_sha256", "stderr_sha256")),
        "time_order": isinstance(record.get("started_epoch"), (int, float)) and isinstance(record.get("ended_epoch"), (int, float)) and record["started_epoch"] <= record["ended_epoch"],
    }
    return {"passed": all(checks.values()), "checks": checks, "missing_fields": missing}


def candidate_final_source_binding(candidate_hash: str, final_hash: str) -> dict[str, Any]:
    valid = all(isinstance(value, str) and len(value) == 64 for value in (candidate_hash, final_hash))
    return {"passed": valid and candidate_hash == final_hash, "candidate_source_tree_sha256": candidate_hash, "final_source_tree_sha256": final_hash}


def run_sacrificial_batch(
    count: int = 10,
    *,
    per_run_timeout_seconds: float = 20.0,
    batch_timeout_seconds: float = 240.0,
    injected_modes: dict[int, str] | None = None,
) -> dict[str, Any]:
    """Run every dangerous repetition in a distinct supervised interpreter.

    A failed or hung victim is recorded and cleaned; later victims still run until
    the single batch deadline prevents starting another complete watchdog window.
    """
    if count < 1:
        raise ValueError("count must be positive")
    started = time.monotonic()
    deadline = started + batch_timeout_seconds
    rows: list[dict[str, Any]] = []
    modes = dict(injected_modes or {})
    for iteration in range(1, count + 1):
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            rows.append({"iteration": iteration, "started": False, "reason": "batch_deadline_exhausted"})
            continue
        watchdog = min(per_run_timeout_seconds, remaining)
        mode = modes.get(iteration, "pass")
        command = [
            sys.executable,
            str(ROOT / "scripts" / "r8_sacrificial_worker.py"),
            "--test-id", "R4-RG-05",
            "--iteration", str(iteration),
            "--mode", mode,
        ]
        record = run_supervised_process(
            command,
            ROOT,
            environment={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
            total_timeout_seconds=watchdog,
            max_output_bytes=256 * 1024,
        )
        parsed: dict[str, Any] | None = None
        try:
            parsed = json.loads(record["stdout"].strip().splitlines()[-1])
        except (IndexError, ValueError, TypeError):
            parsed = None
        rows.append({
            "iteration": iteration,
            "started": True,
            "mode": mode,
            "pid": record["pid"],
            "watchdog_seconds": watchdog,
            "exit_code": record["exit_code"],
            "timed_out": record["timed_out"],
            "orphan_process_count": record["orphan_process_count"],
            "zombie_process_count": record["zombie_process_count"],
            "stdout_sha256": _digest(record["stdout"]),
            "stderr_sha256": _digest(record["stderr"]),
            "parsed_result": parsed,
            "passed": bool(record["exit_code"] == 0 and not record["timed_out"] and record["orphan_process_count"] == 0 and record["zombie_process_count"] == 0 and parsed and parsed.get("result", {}).get("passed") is True),
        })
    completed = [row for row in rows if row.get("started")]
    return {
        "requested_count": count,
        "started_count": len(completed),
        "completed_count": len(rows),
        "pass_count": sum(row.get("passed") is True for row in rows),
        "failure_count": sum(row.get("started") is True and row.get("passed") is not True for row in rows),
        "unique_victim_pids": len({row.get("pid") for row in completed}),
        "continued_after_failure": any(row.get("passed") is not True and any(next_row.get("started") for next_row in rows[index + 1:]) for index, row in enumerate(rows)),
        "batch_deadline_seconds": batch_timeout_seconds,
        "batch_elapsed_ms": round((time.monotonic() - started) * 1000, 3),
        "rows": rows,
    }
