from __future__ import annotations

import ctypes
import os
import pathlib
import platform
import signal
import subprocess
import threading
import time
import uuid
from typing import Any, BinaryIO, Callable, Iterable, Mapping, Sequence


PROCESS_MARKER_ENV = "WRLAB_PROCESS_RUN_TOKEN"
DEFAULT_MAX_OUTPUT_BYTES = 1024 * 1024


def cleanup_reserve_seconds(total_timeout_seconds: float) -> float:
    if total_timeout_seconds <= 0:
        raise ValueError("total_timeout_seconds must be positive")
    return min(1.0, max(0.15, float(total_timeout_seconds) * 0.35))


def total_timeout_for_operation(operation_seconds: float) -> float:
    if operation_seconds <= 0:
        raise ValueError("operation_seconds must be positive")
    total = float(operation_seconds)
    for _ in range(16):
        candidate = float(operation_seconds) + cleanup_reserve_seconds(total)
        if abs(candidate - total) <= time.get_clock_info("monotonic").resolution:
            return candidate
        total = candidate
    return total


class _BoundedOutput:
    def __init__(self, limit: int) -> None:
        if limit < 4096:
            raise ValueError("max_output_bytes must be at least 4096")
        self.limit = int(limit)
        self.head_limit = self.limit // 2
        self.tail_limit = self.limit - self.head_limit
        self.head = bytearray()
        self.tail = bytearray()
        self.total = 0
        self.error: str | None = None

    def add(self, data: bytes) -> None:
        self.total += len(data)
        if len(self.head) < self.head_limit:
            take = min(self.head_limit - len(self.head), len(data))
            self.head.extend(data[:take])
            data = data[take:]
        if data:
            self.tail.extend(data)
            if len(self.tail) > self.tail_limit:
                del self.tail[: len(self.tail) - self.tail_limit]

    @property
    def truncated(self) -> bool:
        return self.total > self.limit

    def text(self) -> str:
        if not self.truncated:
            payload = bytes(self.head + self.tail)
        else:
            marker = f"\n[OUTPUT_TRUNCATED total_bytes={self.total} retained_bytes={self.limit}]\n".encode("ascii")
            payload = bytes(self.head) + marker + bytes(self.tail)
        return payload.decode("utf-8", errors="replace")


def _drain_pipe(stream: BinaryIO, output: _BoundedOutput, callback: Callable[[bytes], None] | None = None) -> None:
    try:
        while True:
            chunk = stream.read(65536)
            if not chunk:
                return
            output.add(chunk)
            if callback is not None:
                callback(chunk)
    except (OSError, ValueError) as error:
        output.error = f"{type(error).__name__}:{error}"
    finally:
        try:
            stream.close()
        except OSError:
            pass


def _windows_process_snapshot() -> dict[int, dict[str, Any]]:
    if os.name != "nt":
        return {}
    from ctypes import wintypes

    class PROCESSENTRY32W(ctypes.Structure):
        _fields_ = [
            ("dwSize", wintypes.DWORD), ("cntUsage", wintypes.DWORD),
            ("th32ProcessID", wintypes.DWORD), ("th32DefaultHeapID", ctypes.c_size_t),
            ("th32ModuleID", wintypes.DWORD), ("cntThreads", wintypes.DWORD),
            ("th32ParentProcessID", wintypes.DWORD), ("pcPriClassBase", wintypes.LONG),
            ("dwFlags", wintypes.DWORD), ("szExeFile", wintypes.WCHAR * 260),
        ]

    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
    kernel.CreateToolhelp32Snapshot.restype = wintypes.HANDLE
    kernel.Process32FirstW.argtypes = [wintypes.HANDLE, ctypes.POINTER(PROCESSENTRY32W)]
    kernel.Process32NextW.argtypes = [wintypes.HANDLE, ctypes.POINTER(PROCESSENTRY32W)]
    kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
    kernel.OpenProcess.restype = wintypes.HANDLE
    kernel.GetProcessTimes.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.FILETIME), ctypes.POINTER(wintypes.FILETIME), ctypes.POINTER(wintypes.FILETIME), ctypes.POINTER(wintypes.FILETIME)]
    kernel.CloseHandle.argtypes = [wintypes.HANDLE]
    handle = kernel.CreateToolhelp32Snapshot(0x00000002, 0)
    if not handle or handle == ctypes.c_void_p(-1).value:
        return {}
    rows: dict[int, dict[str, Any]] = {}
    try:
        entry = PROCESSENTRY32W()
        entry.dwSize = ctypes.sizeof(entry)
        more = bool(kernel.Process32FirstW(handle, ctypes.byref(entry)))
        while more:
            pid = int(entry.th32ProcessID)
            process_handle = kernel.OpenProcess(0x1000, False, pid)
            if process_handle:
                created = wintypes.FILETIME()
                exited = wintypes.FILETIME()
                kernel_time = wintypes.FILETIME()
                user_time = wintypes.FILETIME()
                try:
                    if kernel.GetProcessTimes(process_handle, ctypes.byref(created), ctypes.byref(exited), ctypes.byref(kernel_time), ctypes.byref(user_time)):
                        marker = str((int(created.dwHighDateTime) << 32) | int(created.dwLowDateTime))
                        rows[pid] = {"pid": pid, "ppid": int(entry.th32ParentProcessID), "started": marker, "state": "RUNNING"}
                finally:
                    kernel.CloseHandle(process_handle)
            more = bool(kernel.Process32NextW(handle, ctypes.byref(entry)))
    finally:
        kernel.CloseHandle(handle)
    return rows


def process_snapshot() -> dict[int, dict[str, Any]]:
    if os.name == "nt":
        return _windows_process_snapshot()
    proc = pathlib.Path("/proc")
    if proc.is_dir():
        rows: dict[int, dict[str, Any]] = {}
        for entry in proc.iterdir():
            if not entry.name.isdigit():
                continue
            try:
                fields = (entry / "stat").read_text(encoding="utf-8").rsplit(")", 1)[1].strip().split()
                pid = int(entry.name)
                rows[pid] = {"pid": pid, "ppid": int(fields[1]), "started": fields[19], "state": fields[0]}
            except (OSError, IndexError, ValueError):
                continue
        return rows
    completed = subprocess.run(["ps", "-eo", "pid=,ppid=,state=,lstart="], capture_output=True, text=True, timeout=5, check=False)
    rows = {}
    for line in completed.stdout.splitlines():
        fields = line.split(None, 3)
        if len(fields) != 4:
            continue
        try:
            pid, ppid = int(fields[0]), int(fields[1])
        except ValueError:
            continue
        rows[pid] = {"pid": pid, "ppid": ppid, "state": fields[2], "started": fields[3]}
    return rows


def _identity_key(identity: Mapping[str, Any]) -> str:
    return f"{int(identity['pid'])}:{identity['started']}"


def _same_process(identity: Mapping[str, Any], current: Mapping[str, Any] | None, *, include_zombie: bool = False) -> bool:
    return bool(
        current
        and str(current.get("started")) == str(identity.get("started"))
        and (include_zombie or str(current.get("state")) != "Z")
    )


def collect_descendant_identities(root_pid: int, known: Iterable[Mapping[str, Any]] = ()) -> list[dict[str, Any]]:
    snapshot = process_snapshot()
    root_identity = snapshot.get(int(root_pid))
    parents: dict[int, str] = {int(root_pid): str((root_identity or {}).get("started", ""))}
    for identity in known:
        if _same_process(identity, snapshot.get(int(identity["pid"]))):
            parents[int(identity["pid"])] = str(identity.get("started", ""))
    selected: dict[int, dict[str, Any]] = {}
    changed = True
    while changed:
        changed = False
        for pid, identity in snapshot.items():
            parent_pid = int(identity["ppid"])
            if pid == root_pid or pid in selected or parent_pid not in parents:
                continue
            try:
                if parents[parent_pid] and int(str(identity.get("started", "0"))) < int(parents[parent_pid]):
                    continue
            except ValueError:
                pass
            selected[pid] = identity
            parents[pid] = str(identity.get("started", ""))
            changed = True
    return [selected[pid] for pid in sorted(selected)]


def _marker_identities(marker: str) -> list[dict[str, Any]]:
    proc = pathlib.Path("/proc")
    if os.name == "nt" or not proc.is_dir():
        return []
    needle = f"{PROCESS_MARKER_ENV}={marker}".encode("utf-8")
    snapshot = process_snapshot()
    rows = []
    for pid, identity in snapshot.items():
        try:
            values = (proc / str(pid) / "environ").read_bytes().split(b"\0")
        except (OSError, PermissionError):
            continue
        if needle in values:
            rows.append(identity)
    return rows


def surviving_processes(identities: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]:
    snapshot = process_snapshot()
    return [dict(item) for item in identities if _same_process(item, snapshot.get(int(item["pid"])))]


def zombie_processes(identities: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]:
    snapshot = process_snapshot()
    return [
        dict(item)
        for item in identities
        if _same_process(item, snapshot.get(int(item["pid"])), include_zombie=True)
        and str(snapshot[int(item["pid"])].get("state")) == "Z"
    ]


def _enable_linux_subreaper() -> bool:
    if os.name == "nt" or not pathlib.Path("/proc").is_dir():
        return False
    try:
        return ctypes.CDLL(None, use_errno=True).prctl(36, 1, 0, 0, 0) == 0
    except Exception:
        return False


def _assign_windows_kill_job(process: subprocess.Popen[bytes]) -> Any:
    if os.name != "nt":
        return None

    class IO_COUNTERS(ctypes.Structure):
        _fields_ = [(name, ctypes.c_uint64) for name in (
            "ReadOperationCount", "WriteOperationCount", "OtherOperationCount",
            "ReadTransferCount", "WriteTransferCount", "OtherTransferCount",
        )]

    class BASIC_LIMIT(ctypes.Structure):
        _fields_ = [
            ("PerProcessUserTimeLimit", ctypes.c_int64), ("PerJobUserTimeLimit", ctypes.c_int64),
            ("LimitFlags", ctypes.c_uint32), ("MinimumWorkingSetSize", ctypes.c_size_t),
            ("MaximumWorkingSetSize", ctypes.c_size_t), ("ActiveProcessLimit", ctypes.c_uint32),
            ("Affinity", ctypes.c_size_t), ("PriorityClass", ctypes.c_uint32), ("SchedulingClass", ctypes.c_uint32),
        ]

    class EXTENDED_LIMIT(ctypes.Structure):
        _fields_ = [
            ("BasicLimitInformation", BASIC_LIMIT), ("IoInfo", IO_COUNTERS),
            ("ProcessMemoryLimit", ctypes.c_size_t), ("JobMemoryLimit", ctypes.c_size_t),
            ("PeakProcessMemoryUsed", ctypes.c_size_t), ("PeakJobMemoryUsed", ctypes.c_size_t),
        ]

    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
    kernel.CreateJobObjectW.restype = ctypes.c_void_p
    kernel.SetInformationJobObject.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p, ctypes.c_uint32]
    kernel.AssignProcessToJobObject.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
    kernel.CloseHandle.argtypes = [ctypes.c_void_p]
    handle = kernel.CreateJobObjectW(None, None)
    if not handle:
        return None
    information = EXTENDED_LIMIT()
    information.BasicLimitInformation.LimitFlags = 0x00002000
    assigned = kernel.SetInformationJobObject(handle, 9, ctypes.byref(information), ctypes.sizeof(information))
    assigned = assigned and kernel.AssignProcessToJobObject(handle, ctypes.c_void_p(int(process._handle)))
    if not assigned:
        kernel.CloseHandle(handle)
        return None
    return handle


def _close_windows_job(handle: Any) -> None:
    if handle and os.name == "nt":
        ctypes.WinDLL("kernel32", use_last_error=True).CloseHandle(handle)


def _terminate_identity(identity: Mapping[str, Any], force: bool) -> None:
    if not surviving_processes([identity]):
        return
    pid = int(identity["pid"])
    if os.name != "nt":
        try:
            os.kill(pid, signal.SIGKILL if force else signal.SIGTERM)
        except (ProcessLookupError, PermissionError):
            pass
        return
    from ctypes import wintypes
    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
    kernel.OpenProcess.restype = wintypes.HANDLE
    handle = kernel.OpenProcess(0x0001 | 0x1000, False, pid)
    if handle:
        try:
            kernel.TerminateProcess(handle, 137 if force else 143)
        finally:
            kernel.CloseHandle(handle)


def _reap_tracked(identities: Iterable[Mapping[str, Any]], deadline: float) -> int:
    if os.name == "nt":
        return 0
    pending = {int(item["pid"]) for item in identities}
    reaped: set[int] = set()
    while pending and time.monotonic() < deadline:
        for pid in list(pending):
            try:
                observed, _ = os.waitpid(pid, os.WNOHANG)
            except (ChildProcessError, ProcessLookupError):
                pending.remove(pid)
                continue
            if observed == pid:
                reaped.add(pid)
                pending.remove(pid)
        if pending:
            time.sleep(min(0.01, max(0.0, deadline - time.monotonic())))
    return len(reaped)


def _handshake(path: pathlib.Path, field: str, status: str, process: subprocess.Popen[bytes], deadline: float) -> dict[str, Any]:
    started = time.monotonic()
    last_error = "not_created"
    while time.monotonic() < deadline:
        if path.is_file():
            try:
                import json
                payload = json.loads(path.read_text(encoding="utf-8"))
                if payload.get(field) is True and payload.get("status") == status:
                    return {"status": status, field: True, "reason": f"explicit_{status.lower()}_payload", "elapsed_ms": round((time.monotonic() - started) * 1000, 3), "payload": payload}
                last_error = "status_or_flag_mismatch"
            except (OSError, ValueError) as error:
                last_error = type(error).__name__
        if process.poll() is not None:
            return {"status": "FAIL", field: False, "reason": f"process_exited_before_{status.lower()}", "elapsed_ms": round((time.monotonic() - started) * 1000, 3), "payload": None}
        time.sleep(min(0.01, max(0.0, deadline - time.monotonic())))
    return {"status": "FAIL", field: False, "reason": f"{status.lower()}_timeout:{last_error}", "elapsed_ms": round((time.monotonic() - started) * 1000, 3), "payload": None}


def run_supervised_process(
    command: Sequence[str],
    cwd: pathlib.Path,
    *,
    environment: Mapping[str, str] | None = None,
    total_timeout_seconds: float = 120.0,
    execution_timeout_seconds: float | None = None,
    bootstrapped_path: pathlib.Path | None = None,
    ready_path: pathlib.Path | None = None,
    handshake_timeout_seconds: float | None = None,
    bootstrap_timeout_seconds: float | None = None,
    ready_timeout_seconds: float | None = None,
    max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES,
    stdout_callback: Callable[[bytes], None] | None = None,
    stderr_callback: Callable[[bytes], None] | None = None,
) -> dict[str, Any]:
    if total_timeout_seconds <= 0:
        raise ValueError("total_timeout_seconds must be positive")
    for name, value in (("bootstrap_timeout_seconds", bootstrap_timeout_seconds), ("ready_timeout_seconds", ready_timeout_seconds), ("execution_timeout_seconds", execution_timeout_seconds)):
        if value is not None and value <= 0:
            raise ValueError(f"{name} must be positive")
    if bootstrapped_path is not None and ready_path is not None and pathlib.Path(bootstrapped_path).resolve() == pathlib.Path(ready_path).resolve():
        raise ValueError("BOOTSTRAPPED and READY must use distinct paths")
    started = time.monotonic()
    started_epoch = time.time()
    total_deadline = started + float(total_timeout_seconds)
    cleanup_reserve = cleanup_reserve_seconds(float(total_timeout_seconds))
    operation_deadline = max(started, total_deadline - cleanup_reserve)
    final_verification_reserve = min(0.12, max(0.04, float(total_timeout_seconds) * 0.15))
    cleanup_action_deadline = max(operation_deadline, total_deadline - final_verification_reserve)
    marker = uuid.uuid4().hex
    env = dict(environment or os.environ)
    env[PROCESS_MARKER_ENV] = marker
    env["PYTHONDONTWRITEBYTECODE"] = "1"
    popen_kwargs: dict[str, Any] = {
        "cwd": pathlib.Path(cwd), "stdin": subprocess.DEVNULL, "stdout": subprocess.PIPE,
        "stderr": subprocess.PIPE, "env": env, "text": False,
    }
    if os.name == "nt":
        popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
    else:
        popen_kwargs["start_new_session"] = True
    subreaper_enabled = _enable_linux_subreaper()
    process = subprocess.Popen(list(command), **popen_kwargs)
    if process.stdout is None or process.stderr is None:
        raise RuntimeError("pipe creation failed")
    job_handle = _assign_windows_kill_job(process)
    job_assigned = job_handle is not None
    root_identity = process_snapshot().get(process.pid, {"pid": process.pid, "ppid": os.getpid(), "started": f"popen-{started_epoch}", "state": "RUNNING"})
    stdout_buffer = _BoundedOutput(max_output_bytes)
    stderr_buffer = _BoundedOutput(max_output_bytes)
    stdout_thread = threading.Thread(target=_drain_pipe, args=(process.stdout, stdout_buffer, stdout_callback), name=f"stdout-drain-{process.pid}", daemon=True)
    stderr_thread = threading.Thread(target=_drain_pipe, args=(process.stderr, stderr_buffer, stderr_callback), name=f"stderr-drain-{process.pid}", daemon=True)
    stdout_thread.start()
    stderr_thread.start()
    tracked: dict[str, dict[str, Any]] = {}
    tracked_lock = threading.Lock()
    stop_tracking = threading.Event()

    def sample() -> None:
        with tracked_lock:
            known = list(tracked.values())
        candidates = collect_descendant_identities(process.pid, known)
        candidates.extend(_marker_identities(marker))
        for identity in candidates:
            if int(identity["pid"]) == process.pid:
                continue
            with tracked_lock:
                tracked.setdefault(_identity_key(identity), dict(identity))

    def tracker() -> None:
        while not stop_tracking.is_set():
            sample()
            stop_tracking.wait(0.01)

    sample()
    tracker_thread = threading.Thread(target=tracker, name=f"process-tree-{process.pid}", daemon=True)
    tracker_thread.start()
    not_requested = {"status": "NOT_REQUESTED", "reason": "not_requested", "elapsed_ms": 0.0, "payload": None}
    bootstrap = {**not_requested, "bootstrapped": None}
    ready = {**not_requested, "ready": None}
    legacy_handshake_limit = float(handshake_timeout_seconds) if handshake_timeout_seconds is not None else float(total_timeout_seconds)
    bootstrap_limit = float(bootstrap_timeout_seconds) if bootstrap_timeout_seconds is not None else legacy_handshake_limit
    ready_limit = float(ready_timeout_seconds) if ready_timeout_seconds is not None else legacy_handshake_limit
    handshake_failed = False
    if bootstrapped_path is not None:
        bootstrap = _handshake(pathlib.Path(bootstrapped_path), "bootstrapped", "BOOTSTRAPPED", process, min(operation_deadline, time.monotonic() + bootstrap_limit))
        handshake_failed = bootstrap["bootstrapped"] is not True
    if ready_path is not None and not handshake_failed:
        ready = _handshake(pathlib.Path(ready_path), "ready", "READY", process, min(operation_deadline, time.monotonic() + ready_limit))
        handshake_failed = ready["ready"] is not True
    sample()
    timed_out = False
    if not handshake_failed:
        execution_deadline = operation_deadline
        if execution_timeout_seconds is not None:
            execution_deadline = min(execution_deadline, time.monotonic() + float(execution_timeout_seconds))
        while process.poll() is None and time.monotonic() < execution_deadline:
            time.sleep(min(0.01, max(0.0, execution_deadline - time.monotonic())))
        timed_out = process.poll() is None

    sample()
    with tracked_lock:
        captured = list(tracked.values())
    termination_strategy = "normal_exit_with_marker_and_identity_verification"
    if handshake_failed or timed_out or process.poll() is None or surviving_processes(captured):
        if os.name == "nt" and job_handle:
            _close_windows_job(job_handle)
            job_handle = None
            termination_strategy = "windows_job_object_close_plus_identity_verify"
        else:
            termination_strategy = "posix_group_plus_marker_identity_term_kill" if os.name != "nt" else "windows_identity_terminate"
            if os.name != "nt":
                try:
                    os.killpg(process.pid, signal.SIGTERM)
                except (ProcessLookupError, PermissionError):
                    pass
            elif process.poll() is None:
                try:
                    process.terminate()
                except OSError:
                    pass
            for identity in surviving_processes(captured):
                _terminate_identity(identity, False)
    cleanup_started = time.monotonic()
    force_deadline = min(cleanup_action_deadline, cleanup_started + max(0.01, (cleanup_action_deadline - cleanup_started) * 0.45))
    while time.monotonic() < force_deadline:
        sample()
        with tracked_lock:
            captured = list(tracked.values())
        if process.poll() is not None and not surviving_processes(captured):
            break
        time.sleep(min(0.01, max(0.0, force_deadline - time.monotonic())))
    with tracked_lock:
        captured = list(tracked.values())
    survivors = surviving_processes(captured)
    if process.poll() is None and os.name != "nt":
        try:
            os.killpg(process.pid, signal.SIGKILL)
        except (ProcessLookupError, PermissionError):
            pass
    elif process.poll() is None:
        try:
            process.kill()
        except OSError:
            pass
    for identity in survivors:
        _terminate_identity(identity, True)
    observed_zombies: dict[str, dict[str, Any]] = {}
    reaped_count = 0
    while time.monotonic() < cleanup_action_deadline:
        sample()
        with tracked_lock:
            captured = list(tracked.values())
        for identity in zombie_processes(captured):
            observed_zombies[_identity_key(identity)] = identity
        reaped_count += _reap_tracked(captured, min(cleanup_action_deadline, time.monotonic() + 0.02))
        current_survivors = surviving_processes(captured)
        current_zombies = zombie_processes(captured)
        if process.poll() is not None and not current_survivors and not current_zombies:
            break
        time.sleep(min(0.005, max(0.0, cleanup_action_deadline - time.monotonic())))
    stop_tracking.set()
    remaining = max(0.0, total_deadline - time.monotonic())
    tracker_thread.join(timeout=remaining)
    with tracked_lock:
        captured = list(tracked.values())
    zombies_before = list(observed_zombies.values())
    remaining = max(0.0, total_deadline - time.monotonic())
    stdout_thread.join(timeout=remaining)
    remaining = max(0.0, total_deadline - time.monotonic())
    stderr_thread.join(timeout=remaining)
    _close_windows_job(job_handle)
    process.poll()
    final_snapshot = process_snapshot()
    final_survivors = [dict(item) for item in captured if _same_process(item, final_snapshot.get(int(item["pid"])))]
    zombies_after = [dict(item) for item in captured if _same_process(item, final_snapshot.get(int(item["pid"])), include_zombie=True) and str(final_snapshot[int(item["pid"])].get("state")) == "Z"]
    elapsed = time.monotonic() - started
    deadline_overrun_ms = max(0.0, (time.monotonic() - total_deadline) * 1000)
    ready_failed = ready_path is not None and ready["ready"] is not True
    bootstrap_failed = bootstrapped_path is not None and bootstrap["bootstrapped"] is not True
    exit_code = 125 if bootstrap_failed or ready_failed else 124 if timed_out else process.returncode
    return {
        "command": list(command),
        "exit_code": exit_code,
        "elapsed_ms": round(elapsed * 1000, 3),
        "stdout": stdout_buffer.text(),
        "stderr": stderr_buffer.text(),
        "stdout_total_bytes": stdout_buffer.total,
        "stderr_total_bytes": stderr_buffer.total,
        "stdout_truncated": stdout_buffer.truncated,
        "stderr_truncated": stderr_buffer.truncated,
        "output_limit_bytes": int(max_output_bytes),
        "output_drain_threads_stopped": not stdout_thread.is_alive() and not stderr_thread.is_alive(),
        "output_drain_errors": [error for error in (stdout_buffer.error, stderr_buffer.error) if error],
        "pid": process.pid,
        "total_timeout_seconds": float(total_timeout_seconds),
        "bootstrap_timeout_seconds": bootstrap_limit if bootstrapped_path is not None else None,
        "ready_timeout_seconds": ready_limit if ready_path is not None else None,
        "execution_timeout_seconds": execution_timeout_seconds,
        "timed_out": timed_out,
        "total_deadline_exceeded": deadline_overrun_ms > 50.0,
        "deadline_overrun_ms": round(deadline_overrun_ms, 3),
        "orphan_process_count": len(final_survivors),
        "tracked_descendant_count": len(captured),
        "tracked_descendants": [{"pid": int(item["pid"]), "started": str(item["started"]), "state": str(item.get("state", "UNKNOWN"))} for item in captured],
        "surviving_descendant_pids": [int(item["pid"]) for item in final_survivors],
        "zombie_process_count": len(zombies_after),
        "zombies_before_reap_count": len(zombies_before),
        "reaped_descendant_count": reaped_count,
        "tracker_thread_stopped": not tracker_thread.is_alive(),
        "linux_subreaper_enabled": subreaper_enabled,
        "bootstrap_handshake": bootstrap,
        "ready_handshake": ready,
        "stage_timings": {
            "python_start_to_bootstrapped_ms": bootstrap.get("elapsed_ms") if bootstrapped_path is not None else None,
            "bootstrapped_to_ready_ms": ready.get("elapsed_ms") if ready_path is not None and bootstrap.get("bootstrapped") is True else None,
            "bootstrapped_event_epoch": (bootstrap.get("payload") or {}).get("event_epoch"),
            "ready_event_epoch": (ready.get("payload") or {}).get("event_epoch"),
        },
        "started_epoch": started_epoch,
        "ended_epoch": time.time(),
        "platform": platform.platform(),
        "root_process_identity": {"pid": int(root_identity["pid"]), "started": str(root_identity["started"]), "state": str(root_identity.get("state", "UNKNOWN"))},
        "process_tree_strategy": termination_strategy,
        "windows_job_assigned": job_assigned,
        "process_marker": hashlib_marker(marker),
    }


def hashlib_marker(marker: str) -> str:
    import hashlib
    return hashlib.sha256(marker.encode("ascii")).hexdigest()
