"""Two-sided clock evidence and recoverable owner-preserving admission."""

import threading
from dataclasses import replace

import pytest
from test_perception_streaming_grpc import identity

from k1link.perception.streaming_clock import (
    ClockMappingError,
    ClockProbe,
    ClockReceipt,
    ClockWindow,
)
from k1link.perception.streaming_continuity import StreamSuspended
from k1link.perception.streaming_lifecycle import StreamingLifecycle
from k1link.perception.streaming_queue import StreamMailbox
from k1link.perception.streaming_source_clock import SourceAnchor, SourceClockMonitor


def test_responder_warms_without_source_admission_and_rejects_wide_start():
    monitor = SourceClockMonitor("worker", 10**12)
    wide = sample(outbound=20_000_000, inbound=20_000_000)
    now = wide.local_receive_ns + 10**12 + 1_000_000
    anchor = SourceAnchor(10**12, wide.local_receive_ns + 1_000_000_000)
    response = monitor.observe(wide, now, anchor, False)
    assert response.anchor is monitor.anchor is None and response.bounds.samples == 1
    with pytest.raises(StreamSuspended):
        monitor.observed(now)
    narrow = sample(2)
    now = narrow.local_receive_ns + 10**12 + 1_000_000
    response = monitor.observe(narrow, now, None, False)
    assert response.anchor is None and response.bounds.samples == 2
    response.bounds.require(now)
    with pytest.raises(StreamSuspended):
        monitor.observed(now)  # Valid clock is not permission to start without an anchor.
    third = sample(3)
    now = third.local_receive_ns + 10**12 + 1_000_000
    assert monitor.observe(third, now, anchor, False).anchor == anchor
    assert monitor.observed(now)[1] < 5_000_000


def test_end_before_start_is_terminal_without_inventing_a_source_timeline():
    monitor = SourceClockMonitor("worker", 10**12)
    first = sample()
    state = monitor.observe(first, first.local_receive_ns + 10**12 + 1_000_000, None, True)
    assert state.ended and state.anchor is None
    with pytest.raises(StreamSuspended):
        monitor.observed(state.bounds.measured_at_ns)


def sample(number=1, outbound=1_000_000, inbound=2_000_000, offset=10**12):
    t1 = 10**16 + number * 100_000_000
    return ClockProbe(
        "source",
        "worker",
        str(number),
        t1,
        t1 + outbound + offset,
        t1 + outbound + offset + 10_000,
        t1 + outbound + inbound + 10_000,
    )


@pytest.mark.parametrize("outbound,inbound", [(1, 9_000_000), (9_000_000, 1), (1, 1)])
@pytest.mark.parametrize("offset", [-(10**12), 0, 10**12])
def test_responder_interval_contains_true_reverse_offset(outbound, inbound, offset):
    probe = sample(outbound=outbound, inbound=inbound, offset=offset)
    receipt = ClockReceipt(probe, probe.local_receive_ns + offset + 1_000_000)
    window = ClockWindow("worker", "source", rate_ppm=500, timestamp_error_ns=50_000)
    bounds = window.add(receipt)
    lower, upper = bounds.offset_at(receipt.acknowledged_ns)
    assert lower <= -offset <= upper
    assert bounds.uncertainty_ns(receipt.acknowledged_ns + 1_000_000) > bounds.uncertainty_ns(
        receipt.acknowledged_ns
    )


def test_source_timeline_immutable_expiring_and_observation_bounds_conservative():
    first = sample()
    anchor = SourceAnchor(10**12, first.local_receive_ns + 2_000_000_000)
    monitor = SourceClockMonitor("worker", anchor.source_zero_ns)
    with pytest.raises(StreamSuspended):
        monitor.observed(first.remote_send_ns)
    now = first.local_receive_ns + 10**12 + 1_000_000
    monitor.observe(first, now, anchor, False)
    midpoint, uncertainty = monitor.observed(now)
    actual = anchor.source_zero_ns + (now - 10**12) - anchor.local_zero_ns
    assert midpoint - uncertainty <= actual <= monitor.source_now(now)
    with pytest.raises(StreamSuspended):
        monitor.observed(now + 2_000_000_000)
    second = sample(30)
    later = second.local_receive_ns + 10**12 + 1_000_000
    with pytest.raises(ValueError, match="immutable"):
        monitor.observe(
            second, later, replace(anchor, local_zero_ns=anchor.local_zero_ns + 1), False
        )
    monitor.observe(second, later, anchor, True)
    assert monitor.ended
    with pytest.raises(ValueError, match="ended"):
        monitor.observe(sample(31), later + 100_000_000, anchor, False)


def test_bad_ack_order_expired_jump_and_foreign_clock():
    first = sample()
    for stamp in (first.remote_send_ns - 1, first.remote_receive_ns + 500_000_001):
        with pytest.raises(ClockMappingError):
            ClockReceipt(first, stamp)
    monitor = SourceClockMonitor("worker", 10**12)
    anchor = SourceAnchor(10**12, first.local_receive_ns + 2_000_000_000)
    monitor.observe(first, first.local_receive_ns + 10**12 + 1_000_000, anchor, False)
    jumped = sample(2, offset=10**12 + 100_000_000)
    with pytest.raises(ClockMappingError, match="envelope"):
        monitor.observe(jumped, jumped.local_receive_ns + 10**12 + 101_000_000, anchor, False)
    with pytest.raises(StreamSuspended):
        monitor.observed(jumped.local_receive_ns + 10**12 + 101_000_000)


def test_clock_wait_does_not_stop_owner_or_reopen_mailbox_prematurely(tmp_path):
    healthy = [True]

    def source_now():
        if not healthy[0]:
            raise StreamSuspended("synthetic expired clock mapping")
        return 1_000_000_000

    run = StreamingLifecycle(
        identity(),
        tmp_path,
        StreamMailbox(),
        threading.Event(),
        clock_ns=lambda: 1_000_000_000,
        recover_input=True,
        source_clock_ns=source_now,
    )
    run.ready()
    try:
        healthy[0] = False
        with pytest.raises(StreamSuspended):
            run.check_input(identity())
        assert run.continuity.reason == "source-clock" and not run.stop_event.is_set()
        run.renew(identity())
        with pytest.raises(StreamSuspended):
            run.begin_input(identity())
        assert run.mailbox.epoch_drained and run.continuity.phase == "waiting"
        healthy[0] = True
        epoch = run.begin_input(identity())
        assert epoch != identity() and epoch.lease_generation == 1
        assert run.continuity.phase == "synchronizing"
        healthy[0] = False
        with pytest.raises(StreamSuspended, match="obsolete"):
            run.check_input(identity())
        assert run.continuity.phase == "synchronizing"  # Stale peer cannot pause new epoch.
    finally:
        assert run.close()