"""Small synthetic memory/clock/pipe checks; no real decode or models on the Mac."""

import json
import socket
import struct
from contextlib import contextmanager
from pathlib import Path
from types import SimpleNamespace

import numpy as np
import pytest

from k1link.perception.streaming_pipe_rpc import BoundedPipeRpc, PipeRpcError
from k1link.perception.streaming_queue import StreamMailbox
from k1link.perception.streaming_sensors import CausalSensorWindow, normalized_sensor


def bundle(seq=0, size=10):
    return {"sequence": seq, "payload_bytes": size}


def test_reservation_handoff_is_atomic_and_does_not_double_charge():
    mailbox = StreamMailbox(byte_limit=10)
    ticket = mailbox.reserve_ingress(10)
    value = bundle()
    assert mailbox.put_reserved(value, ticket)
    assert mailbox.bytes == mailbox.peak_bytes == 10
    with pytest.raises(ValueError, match="ownership"):
        ticket.release()
    assert mailbox.take() is value
    mailbox.cancel()
    assert mailbox.bytes == 10 and not mailbox.quiescent
    mailbox.release(value)
    assert mailbox.quiescent


@pytest.mark.parametrize("case", ["closed", "size", "foreign", "sequence"])
def test_rejected_handoff_keeps_caller_ownership(case):
    mailbox = StreamMailbox(byte_limit=40)
    ticket = mailbox.reserve_ingress(10)
    value = bundle()
    if case == "closed":
        mailbox.cancel()
        assert not mailbox.put_reserved(value, ticket)
    else:
        if case == "size":
            value["payload_bytes"] = 11
        if case == "sequence":
            mailbox.put(bundle(1))
        with pytest.raises(ValueError):
            (StreamMailbox() if case == "foreign" else mailbox).put_reserved(value, ticket)
    assert mailbox.bytes >= 10
    ticket.size = 99999  # Only the stored reservation size is authoritative.
    ticket.release()
    mailbox.cancel()
    assert mailbox.bytes == 0 and mailbox.quiescent


def test_handoff_overflow_drops_pending_only():
    mailbox = StreamMailbox(byte_limit=40)
    mailbox.put(bundle(0))
    active = mailbox.take()
    mailbox.put(bundle(1, 5))
    mailbox.put(bundle(2, 5))
    ticket = mailbox.reserve_ingress(20)
    assert mailbox.put_reserved(bundle(3, 20), ticket)
    assert mailbox.dropped == [{"sequence": 1, "reason": "pending-overflow"}]
    assert mailbox.bytes == 35
    mailbox.cancel()
    assert mailbox.bytes == 10
    mailbox.release(active)
    assert mailbox.quiescent


@contextmanager
def pipe(response=None, check=lambda: None):
    client, server = socket.socketpair()
    try:
        if response:
            server.sendall(response)
        yield BoundedPipeRpc(client.fileno(), client.fileno(), check), server
    finally:
        client.close()
        server.close()


def message(header, payload=b""):
    raw = json.dumps({**header, "payload_bytes": len(payload)}).encode()
    return struct.pack("<I", len(raw)) + raw + payload


def test_rpc_fills_only_the_callers_preallocated_buffer():
    with pipe(message({"frame_index": 0}, b"1234")) as (rpc, server):
        target = bytearray(4)
        assert rpc.exchange({"op": "decode"}, b"fragment", target) == {"frame_index": 0}
        assert target == b"1234"
        raw = server.recv(1024)
        size = struct.unpack("<I", raw[:4])[0]
        assert json.loads(raw[4 : 4 + size]) == {"op": "decode", "payload_bytes": 8}
        assert raw[4 + size :] == b"fragment"


@pytest.mark.parametrize("case", ["timeout", "size", "header", "duplicate", "truncated", "guard"])
def test_rpc_failures_are_bounded_and_cannot_be_retried(case):
    raw = message({}, b"x") if case == "size" else None
    if case == "header":
        raw = struct.pack("<I", 4097)
    if case == "duplicate":
        body = b'{"payload_bytes":0,"payload_bytes":0}'
        raw = struct.pack("<I", len(body)) + body

    def guard():
        if case == "guard":
            raise ValueError("lost epoch")

    with pipe(raw, guard) as (rpc, server):
        if case == "truncated":
            server.shutdown(socket.SHUT_WR)
        with pytest.raises(ValueError):
            rpc.exchange(None, b"", bytearray(), timeout=0.02)
        assert rpc.failed
        with pytest.raises(PipeRpcError, match="poisoned"):
            rpc.exchange(None, b"", bytearray(), timeout=0.02)


def point(stamp, seq=1):
    raw = struct.pack("<I", 1) + np.array([[1, 2, 3]], "<f8").tobytes() + b"\xff"
    return normalized_sensor("lidar", seq, stamp, raw)


def pose(stamp):
    return normalized_sensor("pose", 1, stamp, np.array([0, 0, 0, 0, 0, 0, 1], "<f8").tobytes())


def test_causal_preroll_is_preserved_without_future_sensor_substitution():
    mailbox = StreamMailbox()
    window = CausalSensorWindow(mailbox)
    window.append(point(500_000_000))
    window.append(pose(990_000_000))
    window.append(point(999_000_000, 2))
    cut = window.bind(1_000_000_000)
    assert cut.available and len(cut.increments) == len(cut.history_only) == 1
    assert len(window.rolling) == 2 and cut.increments[0].sequence == 2
    window.finish_camera(1_000_000_000)
    window.append(pose(1_001_000_000))
    with pytest.raises(ValueError, match="backwards"):
        window.bind(1_000_000_000)
    assert cut.pose_age_ns == 10_000_000  # New pose cannot mutate the existing cut.
    mailbox.cancel()
    assert mailbox.bytes == window.CACHE_BYTES and not mailbox.quiescent
    window.close()
    assert mailbox.quiescent


def test_sensor_cache_cannot_grow_when_camera_stalls():
    mailbox = StreamMailbox()
    window = CausalSensorWindow(mailbox)
    for sequence in range(64):
        window.append(point(sequence, sequence))
    with pytest.raises(ValueError, match="cache exceeds"):
        window.append(point(65, 65))
    window.close()
    mailbox.finish()
    assert mailbox.quiescent


@pytest.mark.parametrize(
    "modality,raw",
    [
        ("lidar", b""),
        ("lidar", struct.pack("<I", 50001)),
        ("pose", bytes(55)),
        ("pose", bytes(56)),
        ("unknown", bytes(56)),
        ("lidar", struct.pack("<I", 1) + np.array([np.nan, 0, 0], "<f8").tobytes() + b"x"),
    ],
)
def test_normalized_sensor_layout_is_explicit_and_validated(modality, raw):
    with pytest.raises(ValueError):
        normalized_sensor(modality, 1, 100, raw)


def test_cloud_validation_checks_later_bounded_chunks():
    xyz = np.ones((2000, 3), "<f8")
    xyz[-1, -1] = np.inf
    with pytest.raises(ValueError, match="nonfinite"):
        normalized_sensor("lidar", 1, 100, struct.pack("<I", 2000) + xyz.tobytes() + bytes(2000))


def test_wire_pose_owns_aligned_immutable_quaternion_with_exact_values():
    from k1link.perception.geometry_math import quaternion_xyzw_to_rotation_matrix

    quaternion = np.array([0.1, -0.2, 0.3, 0.9273618495495703], "<f8")
    raw = np.array([1, 2, 3], "<f8").tobytes() + quaternion.tobytes()
    position, decoded = normalized_sensor("pose", 1, 100, raw).value
    assert decoded.flags.owndata and decoded.ctypes.data % 16 == 0
    assert not decoded.flags.writeable and not position.flags.writeable
    assert decoded.tobytes() == quaternion.tobytes()
    assert np.array_equal(
        quaternion_xyzw_to_rotation_matrix(decoded),
        quaternion_xyzw_to_rotation_matrix(quaternion),
    )


@pytest.mark.parametrize("failure", ["socket", "receiver-registration", "source-registration"])
def test_bridge_initialization_failure_releases_unstarted_resources(monkeypatch, failure):
    monkeypatch.syspath_prepend(
        str(
            Path(__file__).resolve().parents[1]
            / "experiments/perception/worker/streaming_profile_stage1"
        )
    )
    import pilot_binary_bridge as bridge

    mailbox = StreamMailbox()
    pair = socket.socketpair()
    calls = []

    def sockets():
        if failure == "socket":
            raise ValueError("injected socket failure")
        return pair

    def register(thread):
        calls.append(thread)
        if failure == "receiver-registration" or len(calls) == 2:
            raise ValueError("injected registration failure")

    monkeypatch.setattr(bridge.socket, "socketpair", sockets)
    # Only constructor ownership is under test, without a live lease/thread.
    monkeypatch.setattr("k1link.perception.streaming_ingress.wire.binding", lambda start: {})
    runtime = SimpleNamespace(
        mailbox=mailbox,
        start=None,
        track_thread=register,
        register_ingress=lambda thread, epoch: register(thread),
    )
    source = SimpleNamespace(source_zero=0, wall_zero=0, run=lambda sock: None)
    try:
        with pytest.raises(ValueError, match="injected"):
            bridge.BinaryGraphBridge(runtime, None, source, {})
        mailbox.finish()
        assert mailbox.bytes == 0 and mailbox.quiescent
        if failure != "socket":
            assert all(s.fileno() == -1 for s in pair)
        assert all(t.ident is None for t in calls)
    finally:
        for connection in pair:
            connection.close()