"""Bounded functional tests of display/registration separation and delta fences."""

from dataclasses import replace
from types import SimpleNamespace

import numpy as np
from fastapi import FastAPI
from fastapi.testclient import TestClient
from test_planning_live import event, fixture_service
from test_planning_stabilization import fit, sample

from k1link.missions.live_buffer import LiveCloudBuffer
from k1link.missions.live_display_buffer import LiveDisplayBuffer
from k1link.missions.live_scene_delta import decode_cursor, scene_delta
from k1link.web.planning_live_api import build_planning_live_router


def ingest(display, numeric, t, sequence, points=None):
    pose = event("pose", t=t, sequence=sequence, p=(t / 100, 0, 0))
    cloud = event(
        "points",
        t=t + 0.001,
        sequence=sequence + 1,
        points=np.array([[t / 100, 1, 0]]) if points is None else points,
    )
    for e in (pose, cloud):
        numeric.ingest(e)
        display.ingest(e, numeric)
    return cloud


def test_display_preserves_high_points_independently_from_tracking_filter():
    numeric, display = LiveCloudBuffer([[0,0,0],[10,0,0]]), LiveDisplayBuffer()
    points = np.array([[1,1,1],[1,0,40],[1,0,79.9],[1,0,80.1]])
    ingest(display, numeric, 100, 1, points)
    np.testing.assert_allclose(display.snapshot()["points"], points[:3])
    np.testing.assert_allclose(numeric.snapshot()["points"], points[:1])


def test_native_packets_are_not_numerical_sampling_and_snapshots_are_immutable():
    numeric, display = LiveCloudBuffer([[0, 0, 0], [10, 0, 0]]), LiveDisplayBuffer()
    ingest(display, numeric, 100, 1)
    first = display.snapshot()
    for i in range(1, 10):
        ingest(display, numeric, 100 + i / 10, i * 2 + 1)
    assert display.frames == 10
    assert len(numeric.chunks) == 2  # registration's 500 ms policy is unchanged
    assert display.snapshot()["sequence"] == 20
    assert first["sequence"] == 2 and first["cloud_revision"] == 1
    np.testing.assert_equal(first["points"], [[1, 1, 0]])
    assert len(display.chunks) == 1  # The current half-second receipt stays live.
    np.testing.assert_allclose(display.snapshot()["current_points"], [[1.009, 1, 0]])


def test_budgets_eviction_tombstones_old_receipts_and_segment_reset():
    numeric, display = LiveCloudBuffer([[0, 0, 0], [10, 0, 0]]), LiveDisplayBuffer()
    points = np.array([[x * 0.3, y * 0.3, 0] for x in range(35) for y in range(35)])
    for i in range(60):
        ingest(display, numeric, 100 + i * 0.5, 2 * i, points)
        assert len(display.chunks) <= 40
        assert sum(map(len, display.chunks.values())) <= 40_000
    before = display.frames
    display.ingest(event("points", t=100, points=points), numeric)
    assert display.frames == before
    assert any(p is None and revision > 0 for _, revision, p in display.snapshot()["chunks"])
    ingest(display, numeric, 133, 200, points)
    assert numeric.segment == 1 and len(display.chunks) == 0
    assert len(display.snapshot()["current_points"]) == len(points)


def test_delta_sends_only_changed_chunks_pose_and_transform(monkeypatch):
    import k1link.missions.live_scene_delta as module

    logs = []

    class Recording:
        def __init__(self, *args, **kwargs):
            pass

        def binary_stream(self):
            return SimpleNamespace(read=lambda: b"RRF2")

        def log(self, name, value, **kwargs):
            logs.append((name, value, kwargs))

        def set_time(self, *args, **kwargs):
            logs.append(("time", args, kwargs))

        def flush(self):
            pass

        def disconnect(self):
            pass

    monkeypatch.setattr(module.rr, "RecordingStream", Recording)
    monkeypatch.setattr(module, "log_base", lambda *args: None)
    monkeypatch.setattr(module, "log_view", lambda *args: None)
    monkeypatch.setattr(module.rr, "Points3D", lambda points, **kwargs: np.asarray(points))
    monkeypatch.setattr(module.rr, "Transform3D", lambda **kwargs: kwargs)
    numeric, display = LiveCloudBuffer([[0, 0, 0], [10, 0, 0]]), LiveDisplayBuffer()
    ingest(display, numeric, 100, 1)

    def render(cursor="", result=None, live=True, evidence=None, options=None):
        return scene_delta(
            "run",
            display.epoch,
            np.zeros((3, 3)),
            np.zeros((2, 3)),
            display.snapshot(),
            result or fit(),
            evidence,
            live,
            cursor=cursor,
            options=options,
        )

    _, cursor = render()
    assert all(
        kwargs.get("static")
        for name, _, kwargs in logs
        if name == "world/query" or "/cloud/" in name
    )
    np.testing.assert_equal(dict((n, v) for n, v, _ in logs)["world/query/live"], [[1, 1, 0]])
    logs.clear()
    render(options={"ceiling_m": -0.1})
    assert len(dict((n, v) for n, v, _ in logs)["world/query/live"]) == 0
    logs.clear()
    assert render(cursor)[0] == b""
    assert logs == []
    ingest(display, numeric, 100.1, 3)
    _, cursor = render(cursor)
    assert [n for n, _, _ in logs if "/cloud/" in n] == []
    assert "world/query/live" in [n for n, _, _ in logs]
    assert "world/query" not in [n for n, _, _ in logs]
    logs.clear()
    adjusted = fit()
    adjusted["T_reference_query"][0][3] = 9
    _, cursor = render(cursor, result=adjusted)
    assert [n for n, _, _ in logs] == ["world/query", "time", "world/query/live"]
    assert logs[0][1]["translation"][0] == 9
    logs.clear()
    _, cursor = render(cursor, result=adjusted, evidence=(sample(), fit()))
    assert "world/validated_query" in [n for n, _, _ in logs]
    logs.clear()
    render(cursor, result=adjusted, live=False)
    assert "world/validated_query" in [n for n, _, _ in logs]
    assert not any("/cloud/" in n for n, _, _ in logs)


def test_presentation_changes_and_geometry_repairs_never_reset_camera(monkeypatch):
    import k1link.missions.live_scene_delta as module

    views = []
    monkeypatch.setattr(module, "log_view", lambda *args: views.append(args[-1].copy()))
    reference = np.array([[0., 0., -1.], [20., 1., 40.]])
    cursor = ""

    def render(options=None, **kwargs):
        nonlocal cursor
        payload, cursor = scene_delta(
            "camera-test", kwargs.pop("epoch", "first"), reference, reference,
            None, None, None, False, cursor=cursor, options=options, **kwargs,
        )
        return payload

    assert render().startswith(b"RRF2")
    assert len(views) == 1
    for options in ({"ceiling_m": 3}, {"reference": False}, {"query": False},
                    {"trajectory": False}, {"point_size": 4}, {"grid": False}, {}):
        assert render(options).startswith(b"RRF2")
        assert len(views) == 1
    assert render(base=True).startswith(b"RRF2")
    assert render(epoch="reconnected").startswith(b"RRF2")
    assert len(views) == 1
    render({"mode": "top"})
    render({"mode": "top", "reset": 1})
    assert len(views) == 3
    assert render({"mode": "top", "reset": 1}) == b""
    assert len(views) == 3


def test_grid_is_display_geometry_and_never_a_camera_command():
    from k1link.missions.live_scene import grid_lines, log_base

    reference = np.array([[0., 0., -1.], [20., 1., 40.]])
    logs = {}
    recording = SimpleNamespace(log=lambda path, value, **kw: logs.update({path: value}))
    log_base(recording, reference, reference, {"ceiling_m": 3})
    assert "world/grid" in logs
    assert len(logs["world/reference"].positions.as_arrow_array()) == 1
    assert len(grid_lines(reference)) > 0
    log_base(recording, reference, reference, {"grid": False})
    assert len(logs["world/grid"].strips.as_arrow_array()) == 0


def test_service_freezes_after_loss_fences_identity_and_expires_without_packets(
    tmp_path, monkeypatch
):
    import k1link.missions.live_tests as module

    service, source, _, _ = fixture_service(tmp_path, monkeypatch)
    run_id = "cedf3261-a703-453d-a1c8-aac71c6ce5e3"
    service.run = dict(
        id=run_id,
        state="running",
        query_session_id="B",
        query_generation=2,
        vehicle_control=False,
    )
    service.source = source
    source.state.update(active=True, session_id="B", session_generation=2)
    service.reference, service.reference_path = np.zeros((3, 3)), np.zeros((2, 3))
    summary = service.get()
    assert summary["scene_height_min_m"] == 0.0
    assert summary["scene_height_max_m"] == 80.0
    clock = [int(100.1e9)]
    monkeypatch.setattr(module, "time", SimpleNamespace(monotonic_ns=lambda: clock[0]))
    monkeypatch.setattr(service, "persist", lambda: None)
    numeric = LiveCloudBuffer(service.reference_path)
    for e in (event("pose", t=100), event("points", t=100.01, points=np.ones((2, 3)))):
        numeric.ingest(e)
        service.observe_display(e, numeric, clock[0])
    assert service.presentation.sample is None  # no hint fallback
    service.commit_result(
        fit(),
        sample(),
        {"accepted": True},
        "tracking",
        phase="tracking",
        message="tracking",
        tracking_established=True,
    )
    app = FastAPI()
    app.include_router(build_planning_live_router(service))
    with TestClient(app) as client:
        url = f"/api/v1/mission-planner/live-tests/{run_id}/scene-delta.rrd"
        first = client.get(url)
        assert first.status_code == 200 and first.content.startswith(b"RRF2")
        cursor = first.headers["X-Planning-Scene-Cursor"]
        assert first.headers["X-Planning-Cloud-Revision"] == "1"
        assert first.headers["X-Planning-Cloud-Sequence"] == "1"
        assert first.headers["X-Planning-Height-Min"] == "0.0"
        assert first.headers["X-Planning-Height-Max"] == "80.0"
        observation = {
            "schema_version": "missioncore.planning-browser-presentation/v1",
            "samples": [
                {
                    "cloud_revision": 1,
                    "cloud_sequence": 1,
                    "display_epoch": service.display.epoch,
                    "request_ms": 40.0,
                    "rerun_admission_ms": 2.0,
                    "first_animation_frame_ms": 12.0,
                    "second_animation_frame_ms": 28.0,
                    "frame_timeout": False,
                    "source_to_second_animation_frame_upper_bound_ms": 268.0,
                }
            ],
        }
        report_url = (
            f"/api/v1/mission-planner/live-tests/{run_id}/presentation-observations"
        )
        assert client.post(report_url, json=observation).status_code == 204
        presentation = service.get()["browser_presentation"]
        assert presentation["reported_sample_count"] == 1
        assert presentation["second_animation_frame_ms"]["p50"] == 28.0
        assert service.get()["vehicle_control"] is False
        observation["samples"][0]["cloud_revision"] = 2
        assert client.post(report_url, json=observation).status_code == 204
        assert service.get()["browser_presentation"]["rejected_sample_count"] == 1
        assert client.get(url, params={"cursor": cursor}).status_code == 204
        assert client.get(url, params={"cursor": "corrupt"}).status_code == 200
        assert client.get(url, params={"cursor": "a" * 2049}).status_code == 422
        clock[0] = int(102.2e9)
        expired = client.get(url, params={"cursor": cursor})
        assert expired.status_code == 200
        assert decode_cursor(expired.headers["X-Planning-Scene-Cursor"])["evidence"] is None
    frozen = service.presentation.sample
    service.accepted_sample = None
    fresh = event("points", t=102.3, points=np.ones((2, 3)))
    service.observe_display(replace(fresh, session_id="other"), numeric, int(102.3e9))
    assert service.display.frames == 1
    numeric.ingest(event("pose", t=102.2))
    service.observe_display(event("pose", t=102.2), numeric, int(102.2e9))
    service.observe_display(fresh, numeric, int(102.3e9))
    assert service.presentation.sample is frozen
    source.state["session_generation"] = 3
    assert not service._view_live(int(102.3e9))


def test_reinitialize_endpoint_discards_only_a_failed_initialization(tmp_path, monkeypatch):
    service, _, _, _ = fixture_service(tmp_path, monkeypatch)
    run_id = "cedf3261-a703-453d-a1c8-aac71c6ce5e4"
    service.run = dict(
        id=run_id,
        state="running",
        planning_phase="lost",
        tracking_established=False,
        initialization_attempt=1,
        reinitialization_count=0,
    )
    service.persist = lambda: None
    app = FastAPI()
    app.include_router(build_planning_live_router(service))
    with TestClient(app) as client:
        response = client.post(f"/api/v1/mission-planner/live-tests/{run_id}/reinitialize")
    assert response.status_code == 200
    body = response.json()
    assert body["state"] == "running"
    assert body["planning_phase"] == "waiting-cloud"
    assert body["initialization_attempt"] == 2
    assert body["reinitialization_count"] == 1