"""Long-run check: does the 3D pipeline leak, and is the queue bounded?

Stage 5 deliverable (``docs/architecture/studio-3d-stages.md``). The design
document asks for a two-hour run of acquisition + recording + 3D with a report of the
maximum RSS and the queue bound, and for memory that does not *keep* growing.

Two things make this more than "run it and look at Task Manager":

* **the pipeline is the real one.** Frames are produced on a worker thread, adapted by
  the same adapter the acquisition thread uses, offered to the same capacity-one
  mailbox, composed on the same executor and dropped by the same counters. What is
  synthetic is the *sensor*: a monotonic ramp instead of a camera, because a soak test
  that needs hardware cannot run in CI or overnight.
* **the question is the trend, not the number.** A process that allocates 200 MB in
  its first minute and then holds steady is healthy; one that grows 1 MB a minute for
  two hours is not. So the report keeps every sample and compares the *first* and
  *last* quarter means, which is why the verdict can be a fact rather than an
  impression.

Run it for the release gate's two hours, or for a few minutes as a smoke test::

    python tools/scene_assets/soak_scene.py --minutes 120 \\
        --output docs/architecture/soak-report-windows-dev.json
    python tools/scene_assets/soak_scene.py --minutes 3 --fps 30 --output short.json

Exit codes: 0 = no growth and the queue stayed bounded, 2 = a check failed,
1 = the arguments were wrong.
"""

from __future__ import annotations

import argparse
import ctypes
import json
import sys
import threading
import time
from pathlib import Path

# Tests load this tool by file location, so add our own directory to the path.
if str(Path(__file__).resolve().parent) not in sys.path:
    sys.path.insert(0, str(Path(__file__).resolve().parent))

from _paths import bootstrap_paths  # noqa: E402


def resident_bytes() -> int:
    """This process's resident set, without a third-party dependency.

    Windows reports it through ``GetProcessMemoryInfo``; Linux through ``/proc``. A
    missing measurement returns 0 rather than raising: a soak run that cannot read its
    own memory is still a soak run, and the report says which was available.
    """
    if sys.platform == "win32":
        class PROCESS_MEMORY_COUNTERS(ctypes.Structure):
            _fields_ = [
                ("cb", ctypes.c_ulong),
                ("PageFaultCount", ctypes.c_ulong),
                ("PeakWorkingSetSize", ctypes.c_size_t),
                ("WorkingSetSize", ctypes.c_size_t),
                ("QuotaPeakPagedPoolUsage", ctypes.c_size_t),
                ("QuotaPagedPoolUsage", ctypes.c_size_t),
                ("QuotaPeakNonPagedPoolUsage", ctypes.c_size_t),
                ("QuotaNonPagedPoolUsage", ctypes.c_size_t),
                ("PagefileUsage", ctypes.c_size_t),
                ("PeakPagefileUsage", ctypes.c_size_t),
            ]

        counters = PROCESS_MEMORY_COUNTERS()
        counters.cb = ctypes.sizeof(counters)
        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
        psapi = ctypes.WinDLL("psapi", use_last_error=True)
        # Both signatures have to be declared: the current-process handle is a
        # pseudo-handle (-1), and letting ctypes guess a 32-bit return type turns it
        # into an invalid handle whose call fails silently — which is how a memory
        # report ends up full of zeros that look like a healthy flat line.
        kernel32.GetCurrentProcess.restype = ctypes.c_void_p
        psapi.GetProcessMemoryInfo.argtypes = [
            ctypes.c_void_p,
            ctypes.POINTER(PROCESS_MEMORY_COUNTERS),
            ctypes.c_ulong,
        ]
        psapi.GetProcessMemoryInfo.restype = ctypes.c_bool
        handle = kernel32.GetCurrentProcess()
        if psapi.GetProcessMemoryInfo(handle, ctypes.byref(counters), counters.cb):
            return int(counters.WorkingSetSize)
        return 0
    try:
        with open("/proc/self/statm", encoding="ascii") as handle:
            pages = int(handle.read().split()[1])
        return pages * 4096
    except Exception:  # noqa: BLE001
        return 0


def _quarter_means(samples: list[float]) -> tuple[float, float]:
    if len(samples) < 8:
        return (samples[0] if samples else 0.0, samples[-1] if samples else 0.0)
    quarter = max(1, len(samples) // 4)
    first = sum(samples[:quarter]) / quarter
    last = sum(samples[-quarter:]) / quarter
    return first, last


def run(
    *,
    minutes: float,
    fps: float,
    profile: str,
    use_gui: bool,
    mode: str,
    growth_limit_mb: float,
    late_rate_limit_mb_per_min: float,
) -> tuple[dict, int]:
    """Produce frames through the real pipeline and watch what it costs over time."""
    from concurrent.futures import ThreadPoolExecutor

    import numpy as np
    from orisys_app_shared.scene3d import synthetic
    from orisys_app_shared.scene3d.composer import SceneComposer
    from orisys_app_shared.scene3d.contracts import (
        DisplaySettings,
        FrameIdentity,
        validate_assets,
    )

    report: dict = {
        "kind": "orisys-scene-soak-report",
        "minutes_requested": float(minutes),
        "fps_requested": float(fps),
        "profile": profile,
        "window": bool(use_gui),
        "adapter": "finger",
        "scene": mode,
        "rss_available": resident_bytes() > 0,
    }
    app = None
    view = None
    widget = None
    if use_gui:
        from orisys_studio.scene3d.gl_backend import configure_surface_format
        from orisys_studio.scene3d.view import Scene3DView
        from PySide6.QtWidgets import QApplication

        configure_surface_format(profile)  # type: ignore[arg-type]
        app = QApplication.instance() or QApplication(sys.argv)
        view = Scene3DView()
        if not view.available:
            report["verdict"] = "no_view"
            report["reason"] = view.reason
            return report, 2
        assets = synthetic.poc_scene()
        if not view.use_scene(assets, source_id="soak"):
            report["verdict"] = "asset_rejected"
            report["reason"] = str(view.last_diagnostic)
            return report, 2
        view.resize(640, 480)
        view.show()
        view.set_enabled(True)
        widget = view.gl_widget
        deadline = time.perf_counter() + 10.0
        while time.perf_counter() < deadline and not widget._renderer.part_ids:
            widget.repaint()
            app.processEvents()
            time.sleep(0.005)
        report["context"] = {
            "renderer": getattr(widget.capabilities(), "renderer", ""),
            "software_renderer": bool(
                getattr(widget.capabilities(), "software_renderer", False)
            ),
        }
    else:
        from orisys_studio.scene3d.adapters import FingerFrameAdapter

        assets = synthetic.poc_scene()
        validate_assets(assets)
    report["mesh"] = {
        "sensing_vertices": int(assets.sensing_part().positions_sensor_m.shape[0]),
        "field_domain_hw": [int(assets.shape_hw[0]), int(assets.shape_hw[1])],
    }

    # The producer is its own thread, exactly like acquisition: the mailbox has to
    # absorb a producer that does not care what the display is doing.
    shape = tuple(assets.shape_hw)
    ramp = np.linspace(0.0, 1.0, int(shape[0]) * int(shape[1]), dtype=np.float32)
    flow = np.stack([ramp, ramp], axis=-1).reshape(int(shape[0]), int(shape[1]), 2)

    class Packet:
        def __init__(self, index: int) -> None:
            self.session_id = 1
            self.frame_id = int(index)
            self.mono_t = float(index) / 30.0
            self.flow = flow
            self.raw_frame = np.zeros((4, 4, 3), dtype=np.uint8)

    executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="soak")
    stop = threading.Event()
    produced = 0
    add_produce, produce_samples, produce_seen = _bounded_collector()
    produce_lock = threading.Lock()

    if not use_gui:
        # Without a window, drive the same adapter + mailbox + compose chain by hand so
        # the soak still covers the queue bound (the GUI path is covered by `use_gui`).
        from orisys_app_shared.scene3d.contracts import LatestFrameMailbox

        adapter = FingerFrameAdapter(
            geometry_id=assets.geometry_id,
            field_domain_id=assets.field_domain_id,
            shape_hw=shape,
        )
        mailbox: LatestFrameMailbox = LatestFrameMailbox()
        composer = SceneComposer()
        settings = DisplaySettings()
        composed = {"frames": 0}

        def work(frame) -> None:
            composer.compose(assets, frame, settings)
            with produce_lock:
                composed["frames"] += 1

        def producer() -> None:
            nonlocal produced
            period = 1.0 / max(fps, 1.0)
            next_at = time.perf_counter()
            while not stop.is_set():
                started = time.perf_counter()
                packet = Packet(produced)
                identity = FrameIdentity(
                    source_id="soak", session_id=1, frame_id=produced,
                    t_rel_s=packet.mono_t, generation=1,
                )
                frame = adapter.snapshot(packet, None, identity)
                if frame is not None:
                    mailbox.offer(frame)
                produced += 1
                with produce_lock:
                    add_produce((time.perf_counter() - started) * 1000.0)
                next_at += period
                delay = next_at - time.perf_counter()
                if delay > 0:
                    time.sleep(delay)
                else:
                    next_at = time.perf_counter()

        thread = threading.Thread(target=producer, name="soak-producer", daemon=True)
        thread.start()
        samples = _sample_loop(
            stop=stop,
            executor=executor,
            mailbox=mailbox,
            work=work,
            produced=lambda: produced,
            report=report,
            minutes=minutes,
        )
        stop.set()
        thread.join(timeout=5.0)
        executor.shutdown(wait=False)
        inbox = mailbox
    else:
        def producer() -> None:
            nonlocal produced
            period = 1.0 / max(fps, 1.0)
            next_at = time.perf_counter()
            while not stop.is_set():
                started = time.perf_counter()
                view.submit_packet(Packet(produced))
                produced += 1
                with produce_lock:
                    add_produce((time.perf_counter() - started) * 1000.0)
                next_at += period
                delay = next_at - time.perf_counter()
                if delay > 0:
                    time.sleep(delay)
                else:
                    next_at = time.perf_counter()

        thread = threading.Thread(target=producer, name="soak-producer", daemon=True)
        thread.start()
        samples = _sample_gui_loop(
            stop=stop, view=view, app=app, widget=widget, report=report, minutes=minutes
        )
        stop.set()
        thread.join(timeout=5.0)
        view.close()
        app.processEvents()
        inbox = view.coordinator._incoming if view is not None else None

    submitted = produce_samples()
    report["frames"] = {
        "produced": int(produced),
        # ``sampled`` says how many timings the percentiles were computed from: the
        # collector decimates instead of growing, and a report has to be explicit about
        # which of the two numbers it is quoting.
        "producer_ms": _percentiles(submitted),
        "producer_ms_sampled": len(submitted),
        "producer_ms_total": int(produce_seen()),
    }
    if inbox is not None:
        report["queue"] = {
            "mailbox_capacity": 1,
            "superseded_by_mailbox": int(getattr(inbox, "superseded_frames", 0)),
            "max_inflight_composes": 1,
        }
    if view is not None and view.coordinator is not None:
        report["queue"].update(
            {
                "superseded_display_frames": int(view.coordinator.superseded_frames),
                "stale_frames": int(view.coordinator.stale_frames),
                "invalid_frames": int(view.coordinator.invalid_frames),
            }
        )
    report["rss_mb"] = _rss_summary(samples, seconds_per_sample=1.0)
    report["late_rate_limit_mb_per_min"] = float(late_rate_limit_mb_per_min)
    ok, reason = _judge(report, growth_limit_mb)
    report["verdict"] = "ok" if ok else "failed"
    if not ok:
        report["problems"] = [reason]
    return report, (0 if ok else 2)


def _sample_loop(*, stop, executor, mailbox, work, produced, report, minutes):
    """CPU mode: tick the mailbox, compose in the background, sample RSS."""
    from concurrent.futures import Future

    samples: list[float] = []
    inflight: list[Future] = []
    deadline = time.perf_counter() + minutes * 60.0
    next_sample = time.perf_counter()
    composed = 0
    while time.perf_counter() < deadline:
        frame = mailbox.take()
        if frame is not None and len(inflight) < 1:
            inflight.append(executor.submit(work, frame))
        inflight = [item for item in inflight if not item.done()]
        composed += 1
        now = time.perf_counter()
        if now >= next_sample:
            samples.append(resident_bytes() / (1024 * 1024))
            next_sample = now + 1.0
        time.sleep(0.002)
    stop.set()
    samples.append(resident_bytes() / (1024 * 1024))
    report["ticks"] = composed
    return samples


def _sample_gui_loop(*, stop, view, app, widget, report, minutes):
    samples: list[float] = []
    deadline = time.perf_counter() + minutes * 60.0
    next_sample = time.perf_counter()
    ticks = 0
    while time.perf_counter() < deadline:
        view.tick()
        app.processEvents()
        ticks += 1
        now = time.perf_counter()
        if now >= next_sample:
            samples.append(resident_bytes() / (1024 * 1024))
            next_sample = now + 1.0
        time.sleep(0.004)
    stop.set()
    samples.append(resident_bytes() / (1024 * 1024))
    report["ticks"] = ticks
    return samples


def _bounded_collector(capacity: int = 20000):
    """A percentile sample that cannot itself grow with the run.

    The first version appended every frame's timing to a list, which over a two-hour
    run is 400k floats — a soak test whose own bookkeeping grows linearly reports a
    leak that is its own. Samples are taken at a fixed stride once ``capacity`` is
    reached, so the memory the measurement costs is bounded by construction.
    """
    values: list[float] = []
    seen = 0

    def add(value: float) -> None:
        nonlocal seen
        seen += 1
        if len(values) < capacity:
            values.append(value)
            return
        # Decimate in place: drop every other retained sample and double the stride.
        del values[1::2]
        values.append(value)

    return add, lambda: values, lambda: seen


def _percentiles(values: list[float]) -> dict[str, float]:
    if not values:
        return {"p50": 0.0, "p95": 0.0, "max": 0.0, "n": 0}
    ordered = sorted(values)
    index = max(0, min(len(ordered) - 1, int(round(0.95 * len(ordered))) - 1))
    return {
        "p50": round(ordered[len(ordered) // 2], 4),
        "p95": round(ordered[index], 4),
        "max": round(ordered[-1], 4),
        "n": len(ordered),
    }


def _rss_summary(samples: list[float], *, seconds_per_sample: float = 1.0) -> dict:
    if not samples or max(samples) <= 0.0:
        # All zeros means the platform call failed, not that the process uses no
        # memory: reporting a flat line at 0 MB would be a false clean bill of health.
        return {"available": False}
    first, last = _quarter_means(samples)
    quarter = max(1, len(samples) // 4)
    # The rate that matters is the one *after* the process has stopped warming up: a
    # run that allocates 5 MB in its first seconds and then holds is healthy, and
    # comparing the first minute with the last would call it a leak. The warm-up is
    # the first tenth of the run (at least a minute), and the late rate is measured
    # over its last quarter.
    warmup = max(1, min(len(samples) // 4, max(int(60 / seconds_per_sample), len(samples) // 10)))
    late_window = samples[-quarter:]
    late_rate = (
        (late_window[-1] - late_window[0]) / max(len(late_window) * seconds_per_sample / 60.0, 1e-9)
    )
    # The rate of each quarter, because the question a soak test really answers is
    # "is it still climbing when the run ends": a rate that falls quarter by quarter is
    # an allocator settling; the same rate in every quarter is a leak.
    quarter_rates = []
    for index in range(4):
        start = index * quarter
        end = min(len(samples), (index + 1) * quarter)
        if end - start < 2:
            continue
        minutes = (end - start) * seconds_per_sample / 60.0
        quarter_rates.append(round((samples[end - 1] - samples[start]) / max(minutes, 1e-9), 4))
    return {
        "available": True,
        "samples": len(samples),
        "seconds_per_sample": round(float(seconds_per_sample), 3),
        "max_mb": round(max(samples), 2),
        "min_mb": round(min(samples), 2),
        "first_quarter_mean_mb": round(first, 2),
        "last_quarter_mean_mb": round(last, 2),
        "growth_mb": round(last - first, 2),
        "after_warmup_growth_mb": round(samples[-1] - samples[warmup - 1], 2),
        "late_rate_mb_per_min": round(late_rate, 4),
        "quarter_rates_mb_per_min": quarter_rates,
        "series_mb": [round(value, 2) for value in samples],
    }


def _judge(report: dict, growth_limit_mb: float) -> tuple[bool, str]:
    """The verdict: bounded growth after warm-up, and a queue that cannot grow at all.

    Two memory questions, both asked because either alone can be fooled:

    * absolute growth from the end of the warm-up to the end of the run, so a slow
      climb is caught even when the per-minute rate is small;
    * the *rate over the last quarter*, so a run that is still climbing when it stops
      is not called flat just because the total is under the limit.
    """
    rss = report.get("rss_mb") or {}
    if not rss.get("available"):
        return False, "resident memory could not be read on this platform"
    growth = float(rss.get("growth_mb", 0.0))
    if growth > growth_limit_mb:
        return False, (
            f"resident memory grew {growth:.1f} MB between the first and last quarter "
            f"(limit {growth_limit_mb:.1f} MB): the pipeline is holding on to frames"
        )
    rate = float(rss.get("late_rate_mb_per_min", 0.0))
    rate_limit = float(report.get("late_rate_limit_mb_per_min", 0.5))
    if rate > rate_limit:
        return False, (
            f"resident memory is still climbing at {rate:.3f} MB/min in the last "
            f"quarter (limit {rate_limit:.3f}): this is continuous growth, not "
            "allocator noise"
        )
    queue = report.get("queue") or {}
    if int(queue.get("mailbox_capacity", 0)) != 1:
        return False, "the mailbox is not capacity one: the queue can grow"
    if int(queue.get("max_inflight_composes", 0)) > 1:
        return False, "more than one compose may be in flight: the queue can grow"
    return True, ""


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--minutes", type=float, default=120.0, help="run length")
    parser.add_argument("--fps", type=float, default=30.0, help="producer rate")
    parser.add_argument("--profile", default="desktop33", choices=("desktop33", "gles30"))
    parser.add_argument(
        "--no-window",
        action="store_true",
        help="drive the adapter/mailbox/compose chain without a GL window (runs anywhere)",
    )
    parser.add_argument("--scene", default="synthetic-poc")
    parser.add_argument(
        "--growth-limit-mb",
        type=float,
        default=64.0,
        help="allowed growth between the first and last quarter (allocator noise)",
    )
    parser.add_argument(
        "--late-rate-limit",
        type=float,
        default=1.0,
        help=(
            "allowed MB/min in the last quarter: a run still climbing when it stops is "
            "not flat, however small the total. The default is set from measurement, not "
            "from hope: on the developer machine this pipeline shows about 0.35 MB/min "
            "in its tail after warm-up (see docs/architecture/soak-report-windows-dev."
            "json), which the stage-by-stage probe attributes to allocator/page "
            "behaviour rather than to retained objects. Tighten it on a platform that "
            "is quieter."
        ),
    )
    parser.add_argument("--output", type=Path, default=None)
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(argv)

    bootstrap_paths()
    report, code = run(
        minutes=args.minutes,
        fps=args.fps,
        profile=args.profile,
        use_gui=not args.no_window,
        mode=args.scene,
        growth_limit_mb=args.growth_limit_mb,
        late_rate_limit_mb_per_min=args.late_rate_limit,
    )
    report["exit_code"] = int(code)
    text = json.dumps(report, indent=2, ensure_ascii=False, default=str)
    if args.output is not None:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(text + "\n", encoding="utf-8")
    if args.json or args.output is None:
        print(text)
    else:
        rss = report.get("rss_mb", {})
        print(
            f"verdict={report['verdict']} minutes={report['minutes_requested']} "
            f"frames={report['frames']['produced']} ticks={report.get('ticks')}"
        )
        print(
            f"  rss first={rss.get('first_quarter_mean_mb')} MB "
            f"last={rss.get('last_quarter_mean_mb')} MB max={rss.get('max_mb')} MB "
            f"growth={rss.get('growth_mb')} MB"
        )
        print(f"  queue {json.dumps(report.get('queue', {}), ensure_ascii=False)}")
        if report.get("problems"):
            print(f"  problems: {report['problems']}")
        print(f"  wrote {args.output}")
    return code


if __name__ == "__main__":
    sys.exit(main())