"""Bounded actual scene/mask codec; no models or real recordings."""

import importlib
import json
from hashlib import sha256
from pathlib import Path

import pytest

from k1link.perception.realtime_contract import StreamStart
from k1link.perception.streaming_scene_payload import (
    MAX_SCENE_JSON,
    decode_scene_payload,
    encode_scene_payload,
)


@pytest.fixture
def sample(monkeypatch):
    monkeypatch.syspath_prepend(
        str(
            Path(__file__).resolve().parents[1]
            / "experiments/perception/worker/streaming_profile_stage1"
        )
    )
    pilot = importlib.import_module("pilot_freshness")
    epoch = StreamStart(
        "run", "source", "worker", "epoch", 1, *(["a" * 64] * 4), pilot.CLOCK_DOMAIN, "live"
    )
    mask = b"\x01\x02\x03\x04"
    stamp = 9_007_199_254_740_993
    scene = dict(
        segmentation_sha256=sha256(mask).hexdigest(),
        proposals=[],
        observations=[],
        tracks=[],
        threats=[],
        surface_state="valid",
        range_estimator={},
        costmap_states=[1],
        costmap_material=[1],
        policy_actions=[0],
        costmap_grid=[[0.0, 0.0, 0.45, 0.45]],
        policy_counts={"ALLOW_candidate": 1, "HIGH_COST": 0, "NO_GO": 0},
        tgs_counts={"oldest_permissive_cell_source_ns": stamp},
        commands_enabled=False,
        actuation_allowed=False,
        sequence=2,
        original_source_ns=stamp,
        runtime_binding=epoch.to_dict(),
    )
    bundle = dict(
        sequence=2,
        time_ns=stamp,
        due_ns=1_000_000_000,
        available=True,
        lineage={
            "pose_host_monotonic_ns": stamp,
            "point_increments": [{"host_monotonic_ns": stamp}],
        },
    )
    ddr = dict(state="current", source_sequence=2, source_host_monotonic_ns=stamp)
    pilot.prepare_publication(scene, bundle, ddr, epoch_id="epoch", now_ns=bundle["due_ns"])
    return scene, mask, epoch, bundle, pilot


def test_actual_mask_grid_and_int64_source_identity_roundtrip(sample):
    scene, mask, epoch, bundle, pilot = sample
    raw = encode_scene_payload(json.dumps(scene).encode(), mask, (2, 2))
    received, plane, shape = decode_scene_payload(raw, epoch, 2)
    assert received == scene and bytes(plane) == mask and shape == (2, 2)
    assert plane.obj is raw and plane.readonly  # no second retained mask allocation
    freshness = pilot.validate_receipt(received, bundle, epoch_id="epoch")
    view, assessed = pilot.assess_receipt(
        received, freshness, bundle=bundle, now_ns=bundle["due_ns"] + 251_000_000
    )
    assert not assessed.fresh_complete and view["policy_actions"] == [2]
    assert received["policy_actions"] == [0]  # transit age never rewrites published evidence


@pytest.mark.parametrize(
    "mutation",
    [
        "mask",
        "shape",
        "truncated",
        "extra",
        "magic",
        "sequence",
        "epoch",
        "authority",
        "digest",
        "nan",
        "duplicate",
    ],
)
def test_payload_rejects_bad_framing_binding_and_integrity(sample, mutation):
    scene, mask, epoch, _, _ = sample
    if mutation == "sequence":
        scene["sequence"] = 3
    if mutation == "epoch":
        scene["runtime_binding"]["epoch_id"] = "old"
    if mutation == "authority":
        scene["actuation_allowed"] = True
    if mutation == "digest":
        scene["segmentation_sha256"] = "0" * 64
    if mutation == "nan":
        scene["range_estimator"] = {"range": float("nan")}
    encoded = json.dumps(scene).encode()
    if mutation == "duplicate":
        encoded = b'{"sequence":2,' + encoded[1:]
    raw = encode_scene_payload(encoded, mask, (2, 2))
    if mutation == "mask":
        raw = raw[:-1] + b"z"
    if mutation == "shape":
        raw = raw[:10] + b"\x00\x03" + raw[12:]
    if mutation == "truncated":
        raw = raw[:-1]
    if mutation == "extra":
        raw += b"z"
    if mutation == "magic":
        raw = b"NOPE" + raw[4:]
    with pytest.raises(ValueError):
        decode_scene_payload(raw, epoch, 2)


def test_grid_is_covered_by_the_costmap_layer_digest(sample):
    scene, mask, epoch, bundle, pilot = sample
    scene["costmap_grid"][0][0] = 500
    raw = encode_scene_payload(json.dumps(scene).encode(), mask, (2, 2))
    received, _, _ = decode_scene_payload(raw, epoch, 2)
    with pytest.raises(ValueError, match="digest"):
        pilot.validate_receipt(received, bundle, epoch_id="epoch")


@pytest.mark.parametrize(
    "shape,mask,scene",
    [
        ((0, 2), b"", b"{}"),
        ((True, 1), b"x", b"{}"),
        ((2, 2), b"abc", b"{}"),
        ((1, 1), b"x", b" " * (MAX_SCENE_JSON + 1)),
        ((1, 1), b"x", b""),
    ],
)
def test_encoder_bounds(shape, mask, scene):
    with pytest.raises(ValueError):
        encode_scene_payload(scene, mask, shape)