from __future__ import annotations

import importlib.util
import io
import json
import sys
from argparse import Namespace
from pathlib import Path

import pytest


def _module() -> object:
    perception = Path(__file__).resolve().parents[1] / "experiments" / "perception"
    worker = perception / "worker"
    sys.path.insert(0, str(perception))
    sys.path.insert(0, str(worker))
    try:
        spec = importlib.util.spec_from_file_location(
            "e15_shadow_inference_test",
            worker / "run_e15_shadow_inference.py",
        )
        assert spec is not None and spec.loader is not None
        module = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = module
        spec.loader.exec_module(module)
        return module
    finally:
        sys.path.pop(0)
        sys.path.pop(0)


def test_e15_profile_pins_replay_shadow_authority_and_bounded_runtime() -> None:
    module = _module()
    profile_path = (
        Path(__file__).resolve().parents[1]
        / "experiments"
        / "perception"
        / "worker"
        / "e15_shadow_inference_profile.json"
    )

    profile, digest = module.read_live_profile(profile_path)

    assert len(digest) == 64
    assert profile["mode"] == "replay-shadow-gate"
    assert profile["authority"] == {
        "commands_enabled": False,
        "navigation_or_safety_accepted": False,
    }
    assert profile["transport"] == {
        "wire_schema": "missioncore.live-perception-wire/v1",
        "camera_media": "persistent-fmp4-pyav",
        "pyav_version": "18.0.0",
        "maximum_media_buffer_bytes": 8 * 1024 * 1024,
        "camera_metadata_capacity": 16,
    }
    assert profile["scheduling"] == {
        "detector_queue_capacity": 2,
        "semantic_queue_capacity": 1,
        "semantic_sample_every_frames": 5,
        "semantic_ttl_ms": 750.0,
        "sensor_wait_ms": 90.0,
    }
    assert profile["acceptance"]["minimum_fused_fraction"] == 0.85
    assert profile["acceptance"]["maximum_p95_world_state_age_ms"] == 200.0


def test_e15_profile_rejects_command_authority(tmp_path: Path) -> None:
    module = _module()
    source = (
        Path(__file__).resolve().parents[1]
        / "experiments"
        / "perception"
        / "worker"
        / "e15_shadow_inference_profile.json"
    )
    value = json.loads(source.read_text())
    value["authority"]["commands_enabled"] = True
    changed = tmp_path / "unsafe-profile.json"
    changed.write_text(json.dumps(value))

    with pytest.raises(RuntimeError, match="profile contract"):
        module.read_live_profile(changed)


def test_e28_profile_pins_physical_k1_local_surface_gate() -> None:
    module = _module()
    profile_path = (
        Path(__file__).resolve().parents[1]
        / "experiments"
        / "perception"
        / "worker"
        / "e28_physical_k1_local_surface_profile.json"
    )

    profile, digest = module.read_live_profile(profile_path)

    assert len(digest) == 64
    assert profile["mode"] == "physical-shadow-gate"
    assert profile["local_surface"] == {
        "enabled": True,
        "profile_id": "k1-vendor-map-dynamic-local-surface/v1",
        "profile_sha256": (
            "7a59edc8404d0177a39175578743589bfb6b7822837170cded38ca2b6698cc26"
        ),
        "point_queue_capacity": 2,
        "pose_buffer_capacity": 16,
        "future_pose_wait_ms": 25.0,
        "retention_seconds": 3.0,
        "result_capacity": 8,
        "acceptance": {
            "minimum_bound_frames": 100,
            "maximum_pose_miss_fraction": 0.05,
            "maximum_point_drop_fraction": 0.01,
            "maximum_runtime_drop_fraction": 0.01,
            "maximum_p95_result_age_ms": 80.0,
        },
    }
    assert profile["authority"] == {
        "commands_enabled": False,
        "navigation_or_safety_accepted": False,
    }


def test_e28_profile_rejects_unpinned_local_surface(tmp_path: Path) -> None:
    module = _module()
    source = (
        Path(__file__).resolve().parents[1]
        / "experiments"
        / "perception"
        / "worker"
        / "e28_physical_k1_local_surface_profile.json"
    )
    value = json.loads(source.read_text())
    value["local_surface"]["profile_sha256"] = "0" * 64
    changed = tmp_path / "unpinned-local-surface.json"
    changed.write_text(json.dumps(value))

    with pytest.raises(RuntimeError, match="E28 physical local-surface"):
        module.read_live_profile(changed)


def test_e28_local_surface_acceptance_requires_exact_bounded_accounting() -> None:
    module = _module()
    config = {
        "profile_id": "k1-vendor-map-dynamic-local-surface/v1",
        "point_queue_capacity": 2,
        "pose_buffer_capacity": 16,
        "acceptance": {
            "minimum_bound_frames": 100,
            "maximum_pose_miss_fraction": 0.05,
            "maximum_point_drop_fraction": 0.01,
            "maximum_runtime_drop_fraction": 0.01,
            "maximum_p95_result_age_ms": 80.0,
        },
    }
    authority = {
        "commands_enabled": False,
        "navigation_or_safety_accepted": False,
    }
    snapshot = {
        "closed": True,
        "authority": authority,
        "binder": {
            "points": {
                "capacity": 2,
                "depth": 0,
                "maximum_depth": 2,
                "published": 102,
                "bound": 100,
                "missed": 1,
                "dropped_overflow": 1,
            },
            "poses": {
                "capacity": 16,
                "maximum_depth": 12,
            },
        },
        "runtime": {
            "closed": True,
            "profile": {
                "profile_id": "k1-vendor-map-dynamic-local-surface/v1",
            },
            "queue": {
                "capacity": 2,
                "depth": 0,
                "maximum_depth": 2,
                "published": 100,
                "consumed": 100,
                "dropped_overflow": 0,
            },
            "results": {
                "published": 100,
                "failed": 0,
                "result_age_ms": {"p95": 40.0},
            },
            "occupancy_policy": {
                "absence_of_points_means_free": False,
                "unknown_is_traversable": False,
            },
            "authority": authority,
        },
    }

    checks = module._local_surface_acceptance_checks(snapshot, config)

    assert checks
    assert all(checks.values())
    snapshot["runtime"]["results"]["result_age_ms"]["p95"] = 90.0
    assert (
        module._local_surface_acceptance_checks(snapshot, config)[
            "local_surface_maximum_p95_result_age_ms"
        ]
        is False
    )


def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded(
    tmp_path: Path,
) -> None:
    module = _module()
    service = Namespace(
        output_root=tmp_path,
        listen_host="127.0.0.1",
        listen_port=18020,
        command="serve",
        max_duration_seconds=180.0,
        marker="preserved",
    )

    run = module._persistent_run_arguments(
        service,
        {
            "request_id": "physical-k1-shadow-001",
            "output_name": "physical-k1-shadow-001",
            "token": "a" * 64,
            "max_duration_seconds": 90,
        },
    )

    assert run.command == "run"
    assert run.output == tmp_path / "physical-k1-shadow-001"
    assert run.token == "a" * 64
    assert run.token_stdin is False
    assert run.max_duration_seconds == 90.0
    assert run.marker == "preserved"
    assert not hasattr(run, "listen_host")
    assert not hasattr(run, "output_root")

    with pytest.raises(RuntimeError, match="request contract"):
        module._persistent_run_arguments(
            service,
            {
                "request_id": "physical-k1-shadow-002",
                "output_name": "physical-k1-shadow-002",
                "token": "short",
            },
        )
    with pytest.raises(RuntimeError, match="request contract"):
        module._persistent_run_arguments(
            service,
            {
                "request_id": "physical-k1-shadow-004",
                "output_name": "physical-k1-shadow-004",
                "token": "b" * 64,
                "max_duration_seconds": 181,
            },
        )
    with pytest.raises(RuntimeError, match="request contract"):
        module._persistent_run_arguments(
            service,
            {
                "request_id": "physical-k1-shadow-003",
                "output_name": "../escape",
                "token": "b" * 64,
            },
        )


def test_runtime_telemetry_captures_bounded_queue_snapshots() -> None:
    module = _module()
    stream = io.StringIO()
    telemetry = module._RuntimeTelemetry(
        stream,
        interval_seconds=1.0,
        snapshotters={
            "detector": lambda: {
                "capacity": 2,
                "maximum_depth": 1,
                "final_depth": 0,
            }
        },
    )

    telemetry._sample()
    summary = telemetry.summary()

    assert summary["sample_count"] == 1
    assert summary["final_queues"]["detector"]["capacity"] == 2
    row = json.loads(stream.getvalue())
    assert row["schema_version"] == "missioncore.worker-runtime-telemetry/v1"
    assert row["queues"]["detector"]["maximum_depth"] == 1