"""Small synthetic pause/resume tests: never model/GPU load on the Mac."""

import socket
import threading
from dataclasses import replace

import pytest

from k1link.compute.live_perception import LiveIngressEvent
from k1link.perception import streaming_wire as wire
from k1link.perception.graph_contracts import GraphState
from k1link.perception.realtime_contract import StreamStart
from k1link.perception.streaming_continuity import ResumeEvidence, StreamSuspended
from k1link.perception.streaming_ingress import StreamingIngress
from k1link.perception.streaming_lifecycle import StreamingLifecycle
from k1link.perception.streaming_queue import StreamMailbox
from k1link.perception.streaming_scheduler import SerialGpuStage
from k1link.perception.worker_lease import WorkerLeaseError
from k1link.perception.worker_operating_envelope import WorkerOperatingEnvelope, WorkerSnapshot
from k1link.perception.worker_readiness import WorkerReadinessMonitor


def binding():
    return StreamStart(
        "run",
        "source",
        "worker",
        "epoch",
        1,
        "a" * 64,
        "b" * 64,
        "c" * 64,
        "d" * 64,
        "source-clock",
        "live",
    )


def snapshot(now, **changes):
    return replace(
        WorkerSnapshot(
            "worker",
            "worker-clock",
            now,
            "GPU",
            "driver",
            "b" * 64,
            "c" * 64,
            8000,
            8192,
            2610,
            10251,
            "run",
            1,
            (),
            True,
        ),
        **changes,
    )


@pytest.fixture
def harness(tmp_path):
    runs = []

    def create(*, telemetry=False):
        clock = [1_000_000_000]
        monitor = (
            WorkerReadinessMonitor(
                binding(),
                WorkerOperatingEnvelope("envelope", "GPU", "driver", 8000, 8192, 2610, 10251),
                snapshot(clock[0]),
                mode="strict-envelope",
                clock_domain_id="worker-clock",
                now_monotonic_ns=clock[0],
                recoverable=True,
            )
            if telemetry
            else None
        )
        run = StreamingLifecycle(
            binding(),
            tmp_path / "lease",
            StreamMailbox(),
            threading.Event(),
            clock_ns=lambda: clock[0],
            recover_input=True,
            source_clock_ns=lambda: clock[0],
            readiness=monitor,
        )
        run.ready()
        runs.append(run)
        return run, clock

    yield create
    for run in runs:
        assert run.close()


def proof(now):
    return ResumeEvidence(now, now, now, now, True)


def test_pause_keeps_lease_mailbox_and_warm_models_then_new_epoch(harness):
    run, clock = harness()
    run.admit(run.start, {"sequence": 100, "payload_bytes": 8})
    run.pause_input(run.start, "input-disconnected")
    assert run.state == GraphState.RUNNING and not run.stop_event.is_set()
    assert not run.mailbox.done and not run.mailbox.bytes and not run.lease.released
    assert run.mailbox.drop_counts["input-gap"] == 1
    for _ in range(4):
        clock[0] += 900_000_000
        run.renew(run.start)
    with pytest.raises(StreamSuspended):
        run.admit(run.start, {"sequence": 101, "payload_bytes": 8})
    epoch = run.begin_input(run.start)
    assert epoch.epoch_id != run.start.epoch_id and epoch.lease_generation == 1
    reset = []
    with pytest.raises(StreamSuspended):
        run.admit(epoch, {"sequence": 0, "payload_bytes": 8})
    run.resume_input(epoch, proof(clock[0]), lambda: reset.append("temporal-reset"))
    assert reset == ["temporal-reset"] and run.continuity.phase == "active"
    assert run.admit(epoch, {"sequence": 0, "payload_bytes": 8})
    run.mailbox.release(run.mailbox.take())
    with pytest.raises(StreamSuspended):
        run.validate_result_binding(run.start.to_dict())
    run.validate_result_binding(epoch.to_dict())


def test_active_old_callback_cannot_publish_or_be_freed_early(harness):
    run, clock = harness()
    packet = {"sequence": 0, "payload_bytes": 8}
    run.admit(run.start, packet)
    assert run.mailbox.take() is packet
    with pytest.raises(StreamSuspended), run.work(run.start, "gpu"):
        run.pause_input(run.start, "input-timeout")
        run.check_current(run.start)  # In-flight IPC can drain without poisoning the child.
        with pytest.raises(StreamSuspended):
            run.begin_input(run.start)
    with pytest.raises(StreamSuspended, match="still active"):
        run.begin_input(run.start)
    run.mailbox.release(packet, discard_reason="input-gap")
    epoch = run.begin_input(run.start)
    run.resume_input(epoch, proof(clock[0]), lambda: None)
    with pytest.raises(StreamSuspended):
        run.pause_input(run.start, "input-timeout")
    assert run.continuity.phase == "active"


@pytest.mark.parametrize(
    "field,value",
    [
        ("decoded_keyframe", False),
        ("pose_ns", 0),
        ("newest_points_ns", 0),
        ("oldest_points_ns", 0),
        ("camera_ns", 2_000_000_000),
    ],
)
def test_resume_requires_fresh_synchronized_evidence(harness, field, value):
    run, clock = harness()
    run.pause_input(run.start, "source-gap")
    epoch = run.begin_input(run.start)
    called = []
    with pytest.raises(StreamSuspended):
        run.resume_input(
            epoch, replace(proof(clock[0]), **{field: value}), lambda: called.append(True)
        )
    assert not called and run.continuity.phase == "synchronizing" and not run.stop_event.is_set()


def test_pause_during_reset_never_reactivates_input(harness):
    run, clock = harness()
    run.pause_input(run.start, "source-gap")
    epoch = run.begin_input(run.start)
    with pytest.raises(StreamSuspended):
        run.resume_input(
            epoch, proof(clock[0]), lambda: run.pause_input(epoch, "input-disconnected")
        )
    assert run.continuity.phase == "waiting"


def test_reset_error_is_fatal_not_hidden_as_a_network_pause(harness):
    run, clock = harness()
    run.pause_input(run.start, "source-gap")
    epoch = run.begin_input(run.start)

    def broken():
        raise ValueError("reset failed")

    with pytest.raises(ValueError, match="reset failed"):
        run.resume_input(epoch, proof(clock[0]), broken)
    assert run.stop_event.is_set()


@pytest.mark.parametrize(
    "changes",
    [
        {"competing_gpu_clients": None},
        {"gpu_owner_run_id": None},
        {"memory_clock_mhz": 405},
        {"image_sha256": None},
    ],
)
def test_telemetry_unknown_or_low_clocks_pause_and_fresh_facts_require_resync(harness, changes):
    run, clock = harness(telemetry=True)
    clock[0] += 100_000_000
    run.observe_worker(run.start, snapshot(clock[0], **changes))
    assert run.continuity.phase == "waiting" and not run.stop_event.is_set()
    run.renew(run.start)
    with pytest.raises(StreamSuspended):
        run.begin_input(run.start)
    clock[0] += 100_000_000
    run.observe_worker(run.start, snapshot(clock[0]))
    assert run.continuity.phase == "waiting"
    epoch = run.begin_input(run.start)
    run.resume_input(epoch, proof(clock[0]), lambda: None)
    assert run.continuity.phase == "active" and run.lease.renewals == 1


def test_expired_inventory_can_recover_but_expired_local_lease_cannot(harness):
    run, clock = harness(telemetry=True)
    for _ in range(4):
        clock[0] += 900_000_000
        run.renew(run.start)
    assert run.continuity.phase == "waiting" and not run.stop_event.is_set()
    run.observe_worker(run.start, snapshot(clock[0]))
    epoch = run.begin_input(run.start)
    run.resume_input(epoch, proof(clock[0]), lambda: None)
    clock[0] += 2_000_000_000
    with pytest.raises(WorkerLeaseError):
        run.observe_worker(run.start, snapshot(clock[0]))
    assert run.reason == "lease-lost"


@pytest.mark.parametrize(
    "changes",
    [
        {"competing_gpu_clients": ("other-model",)},
        {"gpu_owner_run_id": "other"},
        {"lease_generation": 2},
        {"image_sha256": "f" * 64},
    ],
)
def test_confirmed_owner_or_identity_conflict_still_stops(harness, changes):
    run, clock = harness(telemetry=True)
    clock[0] += 100_000_000
    with pytest.raises(WorkerLeaseError):
        run.observe_worker(run.start, snapshot(clock[0], **changes))
    assert run.stop_event.is_set() and not run.lease.released


def test_serial_gpu_lane_survives_a_discarded_epoch(harness):
    run, _ = harness()
    seen = threading.Event()

    def compute(packet):
        if packet["sequence"] == 0:
            seen.set()
            raise StreamSuspended("discard this frame")
        return packet["sequence"]

    stage = SerialGpuStage(run.mailbox, compute, run.stop_event)
    try:
        run.admit(run.start, {"sequence": 0, "payload_bytes": 8})
        assert seen.wait(1)
        run.admit(run.start, {"sequence": 1, "payload_bytes": 8})
        packet, result = stage.take()
        assert result == 1 and stage.error is None
        run.mailbox.release(packet)
    finally:
        assert stage.close()


@pytest.mark.parametrize("fault", ["eof", "idle", "partial", "gap"])
def test_connection_loss_keeps_runtime_and_reconnects_without_thread_growth(harness, fault):
    run, clock = harness()
    epoch = run.start
    for index in range(10):
        left, right = socket.socketpair()
        seen = []
        recv = StreamingIngress(
            right,
            run,
            "capture",
            1,
            seen.append,
            lambda _: None,
            input_epoch=epoch,
            idle_timeout=0.05,
            fragment_timeout=0.05,
        )
        recv.start()
        try:
            left.sendall(wire.open_packet(epoch, "capture", 1))
            if fault == "eof":
                left.shutdown(socket.SHUT_WR)
            if fault == "partial":
                event = LiveIngressEvent(
                    1, "capture", 1, "lidar", "points", index, 0, clock[0], b"points"
                )
                header, _ = next(wire.event_packets(epoch, event))
                left.sendall(header + b"p")
                left.shutdown(socket.SHUT_WR)
            if fault == "gap":
                left.sendall(
                    wire.gap_packet(epoch, modality="camera-frame", reason="source-gap", count=1)
                )
            assert recv.join(1)
            assert recv.terminal == "paused" and not seen and not run.stop_event.is_set()
            assert (
                run.continuity.phase == "waiting"
                and run.mailbox.bytes == 0
                and not run.mailbox.done
            )
            clock[0] += 1_000_000
            epoch = run.begin_input(run.start)
            run.resume_input(epoch, proof(clock[0]), lambda: None)
        finally:
            left.close()
    assert len(run._threads) <= 1 and run.continuity.generation == 10


def test_old_transport_binding_and_backlog_never_reach_new_consumer(harness):
    run, clock = harness()
    run.pause_input(run.start, "input-timeout")
    clock[0] += 100_000_000
    epoch = run.begin_input(run.start)
    with pytest.raises(StreamSuspended):
        run.check_input(epoch, synchronizing=True, event_ns=clock[0] - 1)
    with pytest.raises(StreamSuspended):
        run.check_input(run.start, synchronizing=True)
    assert not run.stop_event.is_set()


def test_late_old_packet_on_new_connection_does_not_kill_models(harness):
    run, clock = harness()
    run.pause_input(run.start, "input-disconnected")
    epoch = run.begin_input(run.start)
    run.resume_input(epoch, proof(clock[0]), lambda: None)
    left, right = socket.socketpair()
    seen = []
    receiver = StreamingIngress(
        right, run, "capture", 1, seen.append, lambda _: None, input_epoch=epoch
    )
    receiver.start()
    try:
        left.sendall(wire.open_packet(epoch, "capture", 1))
        left.sendall(wire.terminal_packet(run.start))
        assert receiver.join() and receiver.terminal == "paused"
        assert not seen and not run.stop_event.is_set() and not run.mailbox.done
    finally:
        left.close()