NODEDC_MISSION_CORE/tests/test_mqtt_capture.py

import hashlib
import json
import stat
from collections.abc import Callable
from pathlib import Path
from typing import Any, cast

import paho.mqtt.client as mqtt
import pytest
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.reasoncodes import ReasonCode

import k1link.device_plugins.xgrids_k1.mqtt.capture as capture_module
from k1link.device_plugins.xgrids_k1.mqtt.capture import (
    FRAME_HEADER,
    RAW_MAGIC,
    REPORT_TOPICS,
    CaptureError,
    CaptureFormatError,
    capture_mqtt,
    iter_capture_frames,
    read_capture_clock_envelope,
    seal_capture_clock,
    validate_private_ipv4,
)


class FakeClient:
    def __init__(self, *, topic: str = "RealtimePath", payload: bytes = b"pose-data") -> None:
        self.on_connect: Callable[..., None] | None = None
        self.on_subscribe: Callable[..., None] | None = None
        self.on_message: Callable[..., None] | None = None
        self.on_disconnect: Callable[..., None] | None = None
        self.topic = topic
        self.payload = payload
        self.connect_calls: list[tuple[str, int, int]] = []
        self.subscribe_calls: list[Any] = []
        self.disconnect_count = 0
        self._step = 0

    def connect(self, host: str, port: int, keepalive: int) -> mqtt.MQTTErrorCode:
        self.connect_calls.append((host, port, keepalive))
        return mqtt.MQTT_ERR_SUCCESS

    def subscribe(self, topics: Any) -> tuple[mqtt.MQTTErrorCode, int]:
        self.subscribe_calls.append(topics)
        return mqtt.MQTT_ERR_SUCCESS, 7

    def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
        assert timeout > 0
        self._step += 1
        if self._step == 1:
            assert self.on_connect is not None
            self.on_connect(
                self,
                None,
                mqtt.ConnectFlags(session_present=False),
                ReasonCode(PacketTypes.CONNACK, "Success"),
                None,
            )
        elif self._step == 2:
            assert self.on_subscribe is not None
            self.on_subscribe(
                self,
                None,
                7,
                [ReasonCode(PacketTypes.SUBACK, identifier=0) for _ in REPORT_TOPICS],
                None,
            )
        elif self._step == 3:
            assert self.on_message is not None
            message = mqtt.MQTTMessage(topic=self.topic.encode())
            message.payload = self.payload
            message.qos = 0
            self.on_message(self, None, message)
        else:
            raise KeyboardInterrupt
        return mqtt.MQTT_ERR_SUCCESS

    def disconnect(self) -> mqtt.MQTTErrorCode:
        self.disconnect_count += 1
        return mqtt.MQTT_ERR_SUCCESS


class ConnectionLostAfterMessageClient(FakeClient):
    """Emit one complete subscription/message cycle, then lose the socket."""

    def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
        if self._step < 3:  # noqa: SLF001 - deterministic MQTT test double
            return super().loop(timeout)
        assert timeout > 0
        self._step += 1  # noqa: SLF001
        return mqtt.MQTT_ERR_CONN_LOST


class SubscriptionUnavailableClient(FakeClient):
    """Accept TCP/CONNACK, then fail the transient recovery subscription."""

    def subscribe(self, topics: Any) -> tuple[mqtt.MQTTErrorCode, int]:
        self.subscribe_calls.append(topics)
        return mqtt.MQTT_ERR_NO_CONN, 0


class SubscribedWithoutPointCloudClient(FakeClient):
    """Reach SUBACK, emit only non-PCL reports, then lose the socket."""

    def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
        if self._step < 2:  # noqa: SLF001 - deterministic MQTT test double
            return super().loop(timeout)
        assert timeout > 0
        self._step += 1  # noqa: SLF001
        if self._step == 3:  # noqa: SLF001
            assert self.on_message is not None
            message = mqtt.MQTTMessage(topic=b"lixel/application/report/heartbeat")
            message.payload = b"fresh-heartbeat"
            message.qos = 0
            self.on_message(self, None, message)
            return mqtt.MQTT_ERR_SUCCESS
        if self._step == 4:  # noqa: SLF001
            assert self.on_message is not None
            message = mqtt.MQTTMessage(topic=b"lixel/application/report/device_status")
            message.payload = b"fresh-status"
            message.qos = 0
            self.on_message(self, None, message)
            return mqtt.MQTT_ERR_SUCCESS
        if self._step == 5:  # noqa: SLF001
            assert self.on_message is not None
            message = mqtt.MQTTMessage(topic=b"lixel/application/report/lio_pose")
            message.payload = b"fresh-pose"
            message.qos = 0
            self.on_message(self, None, message)
            return mqtt.MQTT_ERR_SUCCESS
        return mqtt.MQTT_ERR_CONN_LOST


class SpatialSequenceClient(FakeClient):
    def __init__(self, messages: list[tuple[str, bytes, bool]]) -> None:
        super().__init__()
        self._messages = messages

    def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
        if self._step < 2:  # noqa: SLF001 - deterministic MQTT test double
            return super().loop(timeout)
        assert timeout > 0
        self._step += 1  # noqa: SLF001
        message_index = self._step - 3  # noqa: SLF001
        if message_index < len(self._messages):
            assert self.on_message is not None
            topic, payload, retain = self._messages[message_index]
            message = mqtt.MQTTMessage(topic=topic.encode())
            message.payload = payload
            message.qos = 0
            message.retain = retain
            self.on_message(self, None, message)
            return mqtt.MQTT_ERR_SUCCESS
        raise KeyboardInterrupt


class LateOldClientCallbackRecoveryClient(SpatialSequenceClient):
    """Replay one late callback from the retired client before fresh data."""

    def __init__(self, old_client: FakeClient) -> None:
        super().__init__([("RealtimePointcloud", b"fresh-new-client-pcl", False)])
        self._old_client = old_client

    def loop(self, timeout: float) -> mqtt.MQTTErrorCode:
        if self._step < 2:  # noqa: SLF001 - deterministic MQTT test double
            return super().loop(timeout)
        if self._step == 2:  # noqa: SLF001
            assert self._old_client.on_message is not None
            late = mqtt.MQTTMessage(topic=b"RealtimePointcloud")
            late.payload = b"late-old-client-pcl"
            late.qos = 0
            self._old_client.on_message(self._old_client, None, late)
        return super().loop(timeout)


@pytest.mark.parametrize("address", ["10.0.0.1", "172.16.0.1", "172.31.255.254", "192.168.4.2"])
def test_validate_private_ipv4_accepts_only_rfc1918(address: str) -> None:
    assert validate_private_ipv4(address) == address


@pytest.mark.parametrize(
    "address",
    ["8.8.8.8", "127.0.0.1", "169.254.1.2", "::1", "k1.local", " 192.168.1.2"],
)
def test_validate_private_ipv4_rejects_other_targets(address: str) -> None:
    with pytest.raises(ValueError, match="private IPv4|RFC1918"):
        validate_private_ipv4(address)


@pytest.mark.parametrize(
    ("attempt", "expected"),
    [
        (1, 0.5),
        (2, 1.0),
        (3, 2.0),
        (4, 4.0),
        (5, 5.0),
        (1025, 5.0),
        (10**100, 5.0),
    ],
)
def test_recovery_backoff_saturates_without_unbounded_exponentiation(
    attempt: int,
    expected: float,
) -> None:
    assert capture_module._recovery_backoff_seconds(attempt) == expected  # noqa: SLF001


def test_capture_writes_verifiable_frames_metadata_and_summary(tmp_path: Path) -> None:
    fake = FakeClient()
    observed = []
    clock_ready: list[bool] = []

    def on_clock_established() -> None:
        origin = tmp_path / "capture" / "mqtt.timeline.origin.json"
        clock_ready.append(origin.is_file() and bool(origin.read_bytes()))

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "capture",
        duration_seconds=30,
        on_clock_established=on_clock_established,
        on_message_recorded=observed.append,
        _client_factory=lambda: cast(mqtt.Client, fake),
    )

    assert fake.connect_calls == [("192.168.1.50", 1883, 30)]
    assert fake.subscribe_calls == [[(topic, 0) for topic in REPORT_TOPICS]]
    assert fake.disconnect_count == 1
    assert clock_ready == [True]
    assert summary["stop_reason"] == "keyboard_interrupt"
    assert summary["message_count"] == 1
    assert summary["payload_bytes"] == len(fake.payload)
    assert summary["subscriptions"] == list(REPORT_TOPICS)
    assert len(observed) == 1
    assert observed[0].topic == fake.topic
    assert observed[0].payload == fake.payload
    assert observed[0].received_at_epoch_ns > 0

    capture_dir = tmp_path / "capture"
    raw = (capture_dir / "mqtt.raw.k1mqtt").read_bytes()
    assert raw.startswith(RAW_MAGIC)
    topic_length, payload_length = FRAME_HEADER.unpack_from(raw, len(RAW_MAGIC))
    topic_start = len(RAW_MAGIC) + FRAME_HEADER.size
    payload_start = topic_start + topic_length
    assert raw[topic_start:payload_start].decode() == fake.topic
    assert payload_length == len(fake.payload)
    assert raw[payload_start:] == fake.payload

    records: list[dict[str, object]] = [
        json.loads(line) for line in (capture_dir / "mqtt.metadata.jsonl").read_text().splitlines()
    ]
    assert len(records) == 1
    record = records[0]
    assert record["record_type"] == "message"
    assert record["topic"] == fake.topic
    assert record["payload_bytes"] == len(fake.payload)
    assert isinstance(record["received_at_epoch_ns"], int)
    assert record["payload_sha256"] == hashlib.sha256(fake.payload).hexdigest()
    assert record["raw_frame_offset"] == len(RAW_MAGIC)
    assert record["raw_payload_offset"] == payload_start

    frames = list(iter_capture_frames(capture_dir / "mqtt.raw.k1mqtt"))
    assert len(frames) == 1
    assert frames[0].sequence == 1
    assert frames[0].topic == fake.topic
    assert frames[0].payload == fake.payload
    assert frames[0].raw_frame_offset == len(RAW_MAGIC)
    assert frames[0].raw_payload_offset == payload_start

    saved_summary = json.loads((capture_dir / "mqtt.summary.json").read_text())
    assert saved_summary == summary
    assert saved_summary["schema_version"] == 2
    assert saved_summary["artifact_hashes"]["raw_sha256"] == hashlib.sha256(raw).hexdigest()
    clock = read_capture_clock_envelope(
        capture_dir / "mqtt.timeline.json",
        expected_sha256=saved_summary["artifact_hashes"]["capture_clock_sha256"],
    )
    assert clock.started_monotonic_ns <= observed[0].received_monotonic_ns
    assert observed[0].received_monotonic_ns <= clock.completed_monotonic_ns
    assert clock.duration_ns > 0
    assert saved_summary["session_elapsed_seconds"] == pytest.approx(
        clock.duration_ns / 1_000_000_000
    )
    assert saved_summary["artifacts"]["capture_clock"] == "mqtt.timeline.json"
    for artifact_name in (
        "mqtt.raw.k1mqtt",
        "mqtt.metadata.jsonl",
        "mqtt.recovery.jsonl",
        "mqtt.timeline.origin.json",
        "mqtt.timeline.json",
        "mqtt.summary.json",
    ):
        assert stat.S_IMODE((capture_dir / artifact_name).stat().st_mode) == 0o600


def test_capture_without_duration_runs_until_explicit_stop(tmp_path: Path) -> None:
    fake = FakeClient()
    stop_checks = 0

    def should_stop() -> bool:
        nonlocal stop_checks
        stop_checks += 1
        return stop_checks >= 4

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "unbounded-capture",
        duration_seconds=None,
        should_stop=should_stop,
        _client_factory=lambda: cast(mqtt.Client, fake),
    )

    assert summary["requested_duration_seconds"] is None
    assert summary["stop_reason"] == "external_stop"


def test_oversize_payload_is_not_written_to_raw_capture(tmp_path: Path) -> None:
    fake = FakeClient(payload=b"oversize")
    capture_dir = tmp_path / "capture"

    with pytest.raises(CaptureError, match="limit is 4 bytes") as error:
        capture_mqtt(
            "10.1.2.3",
            capture_dir,
            max_message_bytes=4,
            _client_factory=lambda: cast(mqtt.Client, fake),
        )

    assert (capture_dir / "mqtt.raw.k1mqtt").read_bytes() == RAW_MAGIC
    record = json.loads((capture_dir / "mqtt.metadata.jsonl").read_text())
    assert record["record_type"] == "rejected_message"
    assert record["payload_bytes"] == len(fake.payload)
    assert error.value.summary is not None
    assert error.value.summary["stop_reason"] == "message_too_large"
    assert error.value.summary["message_count"] == 0
    assert error.value.summary["rejected_message_count"] == 1


def test_capture_refuses_to_overwrite_existing_artifacts(tmp_path: Path) -> None:
    capture_dir = tmp_path / "capture"
    capture_dir.mkdir()
    raw = capture_dir / "mqtt.raw.k1mqtt"
    raw.write_bytes(b"existing evidence")

    with pytest.raises(FileExistsError, match="refusing to overwrite"):
        capture_mqtt(
            "192.168.1.2",
            capture_dir,
            _client_factory=lambda: cast(mqtt.Client, FakeClient()),
        )

    assert raw.read_bytes() == b"existing evidence"


def test_capture_can_be_stopped_by_owner_without_losing_artifacts(tmp_path: Path) -> None:
    fake = FakeClient()
    stop_checks = 0

    def should_stop() -> bool:
        nonlocal stop_checks
        stop_checks += 1
        return stop_checks >= 4

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "capture",
        duration_seconds=30,
        should_stop=should_stop,
        _client_factory=lambda: cast(mqtt.Client, fake),
    )

    assert summary["stop_reason"] == "external_stop"
    assert summary["message_count"] == 1
    assert list(iter_capture_frames(tmp_path / "capture" / "mqtt.raw.k1mqtt"))[0].payload


def test_guarded_recovery_keeps_one_evidence_writer_and_never_publishes(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient(payload=b"before-loss")
    second = FakeClient(topic="RealtimePointcloud", payload=b"after-recovery")
    clients = iter((first, second))
    losses: list[str] = []
    attempts: list[int] = []
    candidates: list[tuple[int, int]] = []

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "recovering-capture",
        duration_seconds=30,
        on_connection_lost=losses.append,
        recover_connection=lambda attempt: attempts.append(attempt) or "resume",
        on_recovery_point_cloud_candidate=lambda attempt, sequence: candidates.append(
            (attempt, sequence)
        ),
        _client_factory=lambda: cast(mqtt.Client, next(clients)),
    )

    frames = list(iter_capture_frames(tmp_path / "recovering-capture" / "mqtt.raw.k1mqtt"))
    assert [frame.payload for frame in frames] == [b"before-loss", b"after-recovery"]
    assert losses and "network loop failed" in losses[0]
    assert attempts == [1]
    assert candidates == [(1, 2)]
    assert summary["reconnect_enabled"] is True
    assert summary["recovery_attempts"] == 1
    assert summary["successful_recoveries"] == 0
    assert summary["recovery_point_cloud_candidates"] == 1
    assert summary["recovery_blocked"] is False
    assert summary["publishing_enabled"] is False
    assert first.subscribe_calls and second.subscribe_calls


def test_recovery_success_requires_one_exact_post_publish_confirmation(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient(payload=b"before-loss")
    second = FakeClient(topic="RealtimePointcloud", payload=b"after-recovery")
    clients = iter((first, second))
    confirmers: list[Callable[[int], bool]] = []
    confirmation_results: list[bool] = []

    def on_message_recorded(message: object) -> None:
        if getattr(message, "payload", None) != b"after-recovery":
            return
        confirmer = confirmers[0]
        confirmation_results.extend(
            (
                confirmer(999),  # stale/future attempt
                confirmer(1),  # exact post-publication edge
                confirmer(1),  # duplicate edge
            )
        )

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "confirmed-recovery",
        duration_seconds=30,
        on_message_recorded=on_message_recorded,
        recover_connection=lambda _attempt: "resume",
        on_recovery_confirmer_ready=confirmers.append,
        _client_factory=lambda: cast(mqtt.Client, next(clients)),
    )

    assert confirmation_results == [False, True, False]
    assert summary["recovery_attempts"] == 1
    assert summary["recovery_point_cloud_candidates"] == 1
    assert summary["successful_recoveries"] == 1
    assert len(summary["recovery_gaps"]) == 1
    gap = summary["recovery_gaps"][0]
    assert gap["gap_index"] == 1
    assert gap["recovery_attempt"] == 1
    assert gap["outcome"] == "recovered"
    assert gap["ended_monotonic_ns"] >= gap["started_monotonic_ns"]
    assert gap["duration_seconds"] >= 0

    journal_path = tmp_path / "confirmed-recovery" / "mqtt.recovery.jsonl"
    journal = [json.loads(line) for line in journal_path.read_text().splitlines()]
    assert [record["record_type"] for record in journal] == [
        "recovery_gap_started",
        "recovery_gap_ended",
    ]
    assert journal[1]["gap_index"] == gap["gap_index"]
    assert journal[1]["recovery_attempt"] == gap["recovery_attempt"]
    assert journal[1]["outcome"] == gap["outcome"]
    assert summary["artifact_hashes"]["recovery_gaps_jsonl_sha256"] == hashlib.sha256(
        journal_path.read_bytes()
    ).hexdigest()


@pytest.mark.parametrize("wake_step", [3, 4])
def test_owner_wake_enters_same_single_recovery_loop_before_or_with_paho_loss(
    tmp_path: Path,
    wake_step: int,
) -> None:
    first: FakeClient = (
        FakeClient(payload=b"before-owner-wake")
        if wake_step == 3
        else ConnectionLostAfterMessageClient(payload=b"before-owner-wake")
    )
    second = FakeClient(topic="RealtimePointcloud", payload=b"after-owner-wake")
    clients = iter((first, second))
    wake_consumed = False
    losses: list[str] = []
    attempts: list[int] = []

    def consume_owner_wake() -> str | None:
        nonlocal wake_consumed
        if not wake_consumed and first._step >= wake_step:  # noqa: SLF001
            wake_consumed = True
            return "camera-source-ended"
        return None

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / f"owner-wake-{wake_step}",
        duration_seconds=30,
        on_connection_lost=losses.append,
        consume_connection_recovery_request=consume_owner_wake,
        recover_connection=lambda attempt: attempts.append(attempt) or "resume",
        _client_factory=lambda: cast(mqtt.Client, next(clients)),
    )

    frames = list(
        iter_capture_frames(tmp_path / f"owner-wake-{wake_step}" / "mqtt.raw.k1mqtt")
    )
    assert [frame.payload for frame in frames] == [
        b"before-owner-wake",
        b"after-owner-wake",
    ]
    assert wake_consumed is True
    assert len(losses) == 1
    assert attempts == [1]
    assert first.disconnect_count == 1
    assert second.disconnect_count == 1
    assert summary["recovery_attempts"] == 1
    assert summary["successful_recoveries"] == 0
    assert summary["recovery_point_cloud_candidates"] == 1
    assert summary["message_count"] == 2


def test_guarded_recovery_retries_transient_subscription_handshake_failure(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient(payload=b"before-loss")
    transient = SubscriptionUnavailableClient()
    recovered_client = FakeClient(topic="RealtimePointcloud", payload=b"after-recovery")
    clients = iter((first, transient, recovered_client))
    attempts: list[int] = []
    candidates: list[tuple[int, int]] = []

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "recovering-subscription",
        duration_seconds=30,
        recover_connection=lambda attempt: attempts.append(attempt) or "resume",
        on_recovery_point_cloud_candidate=lambda attempt, sequence: candidates.append(
            (attempt, sequence)
        ),
        _client_factory=lambda: cast(mqtt.Client, next(clients)),
    )

    frames = list(
        iter_capture_frames(tmp_path / "recovering-subscription" / "mqtt.raw.k1mqtt")
    )
    assert [frame.payload for frame in frames] == [b"before-loss", b"after-recovery"]
    assert attempts == [1, 2]
    assert candidates == [(2, 2)]
    assert summary["recovery_attempts"] == 2
    assert summary["successful_recoveries"] == 0
    assert summary["recovery_point_cloud_candidates"] == 1


def test_guarded_recovery_suback_and_pose_without_point_cloud_stay_reconnecting(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient(payload=b"before-loss")
    subscribed_only = SubscribedWithoutPointCloudClient()
    recovered_client = SpatialSequenceClient(
        [("lixel/application/report/lio_pcl", b"fresh-pcl", False)]
    )
    clients = iter((first, subscribed_only, recovered_client))
    attempts: list[int] = []
    candidates: list[tuple[int, int]] = []

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "suback-without-spatial-data",
        duration_seconds=30,
        recover_connection=lambda attempt: attempts.append(attempt) or "resume",
        on_recovery_point_cloud_candidate=lambda attempt, sequence: candidates.append(
            (attempt, sequence)
        ),
        _client_factory=lambda: cast(mqtt.Client, next(clients)),
    )

    assert attempts == [1, 2]
    assert candidates == [(2, 5)]
    assert summary["successful_recoveries"] == 0
    assert summary["recovery_point_cloud_candidates"] == 1
    assert summary["message_count"] == 5


def test_guarded_recovery_arms_candidate_before_enqueueing_first_fresh_point_cloud(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient(payload=b"before-loss")
    recovered_client = SpatialSequenceClient(
        [
            ("lixel/application/report/heartbeat", b"heartbeat", False),
            ("lixel/application/report/lio_pcl", b"retained-pcl", True),
            ("DeviceStatus", b"status", False),
            ("RealtimePath", b"fresh-pose", False),
            ("RealtimePointcloud", b"fresh-pcl", False),
            ("RealtimePath", b"later-pose", False),
        ]
    )
    clients = iter((first, recovered_client))
    candidates: list[tuple[int, int]] = []
    events: list[str] = []

    def record_candidate(attempt: int, sequence: int) -> None:
        candidates.append((attempt, sequence))
        events.append(f"candidate:{attempt}:{sequence}")

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "fresh-point-cloud-recovery",
        duration_seconds=30,
        on_message_recorded=lambda message: events.append(
            f"message:{message.payload.decode()}"
        ),
        recover_connection=lambda _attempt: "resume",
        on_recovery_point_cloud_candidate=record_candidate,
        _client_factory=lambda: cast(mqtt.Client, next(clients)),
    )

    assert candidates == [(1, 6)]
    assert events.index("message:fresh-pose") < events.index("message:fresh-pcl")
    assert events.index("candidate:1:6") < events.index("message:fresh-pcl")
    assert events.count("candidate:1:6") == 1
    assert summary["successful_recoveries"] == 0
    assert summary["recovery_point_cloud_candidates"] == 1
    assert summary["message_count"] == 7


def test_guarded_recovery_ignores_late_spatial_callback_from_retired_client(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient(payload=b"before-loss")
    recovered_client = LateOldClientCallbackRecoveryClient(first)
    clients = iter((first, recovered_client))
    recorded_payloads: list[bytes] = []
    candidates: list[tuple[int, int]] = []

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "late-old-client-callback",
        duration_seconds=30,
        on_message_recorded=lambda message: recorded_payloads.append(message.payload),
        recover_connection=lambda _attempt: "resume",
        on_recovery_point_cloud_candidate=lambda attempt, sequence: candidates.append(
            (attempt, sequence)
        ),
        _client_factory=lambda: cast(mqtt.Client, next(clients)),
    )

    assert candidates == [(1, 2)]
    assert recorded_payloads == [b"before-loss", b"fresh-new-client-pcl"]
    assert summary["message_count"] == 2
    assert summary["successful_recoveries"] == 0
    assert summary["recovery_point_cloud_candidates"] == 1


def test_failed_resubscribe_consumes_resume_before_later_standby(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient(payload=b"before-loss")
    transient = SubscriptionUnavailableClient()
    clients = iter((first, transient))
    factory_calls = 0

    def factory() -> mqtt.Client:
        nonlocal factory_calls
        factory_calls += 1
        return cast(mqtt.Client, next(clients))

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "recovery-standby-after-resubscribe-failure",
        duration_seconds=30,
        recover_connection=lambda attempt: "resume" if attempt == 1 else "standby",
        _client_factory=factory,
    )

    assert factory_calls == 2
    assert summary["stop_reason"] == "recovery_standby"
    assert summary["recovery_attempts"] == 2
    assert summary["successful_recoveries"] == 0


def test_guarded_recovery_can_stay_blocked_until_local_owner_finishes(
    tmp_path: Path,
) -> None:
    first = ConnectionLostAfterMessageClient()
    factory_calls = 0
    stop_checks = 0

    def factory() -> mqtt.Client:
        nonlocal factory_calls
        factory_calls += 1
        return cast(mqtt.Client, first)

    def should_stop() -> bool:
        nonlocal stop_checks
        stop_checks += 1
        return stop_checks >= 6

    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "blocked-capture",
        duration_seconds=30,
        should_stop=should_stop,
        recover_connection=lambda _attempt: "blocked",
        _client_factory=factory,
    )

    assert factory_calls == 1
    assert summary["stop_reason"] == "external_stop"
    assert summary["recovery_attempts"] == 1
    assert summary["successful_recoveries"] == 0
    assert summary["recovery_blocked"] is True


def test_guarded_recovery_standby_is_truthful_non_error_completion(
    tmp_path: Path,
) -> None:
    summary = capture_mqtt(
        "192.168.1.50",
        tmp_path / "standby-capture",
        duration_seconds=30,
        recover_connection=lambda _attempt: "standby",
        _client_factory=lambda: cast(mqtt.Client, ConnectionLostAfterMessageClient()),
    )

    assert summary["stop_reason"] == "recovery_standby"
    assert summary["message_count"] == 1
    assert summary["recovery_attempts"] == 1
    assert summary["successful_recoveries"] == 0


def test_owner_seals_session_clock_after_all_producers_stop(tmp_path: Path) -> None:
    capture_dir = tmp_path / "capture"
    summary = capture_mqtt(
        "192.168.1.50",
        capture_dir,
        duration_seconds=30,
        _client_factory=lambda: cast(mqtt.Client, FakeClient()),
    )
    provisional = read_capture_clock_envelope(capture_dir / "mqtt.timeline.json")
    assert summary["capture_clock_scope"] == "transport"

    sealed = seal_capture_clock(capture_dir)

    updated = json.loads((capture_dir / "mqtt.summary.json").read_text(encoding="utf-8"))
    assert updated["capture_clock_scope"] == "session"
    sealed_path = capture_dir / updated["artifacts"]["capture_clock"]
    assert sealed_path.name == f"mqtt.timeline.session-{sealed.artifact_sha256}.json"
    assert (capture_dir / "mqtt.timeline.json").read_bytes()
    assert sealed.started_monotonic_ns == provisional.started_monotonic_ns
    assert sealed.completed_monotonic_ns >= provisional.completed_monotonic_ns
    assert updated["artifact_hashes"]["capture_clock_sha256"] == sealed.artifact_sha256
    assert updated["session_elapsed_seconds"] == pytest.approx(sealed.duration_ns / 1_000_000_000)

    repeated = seal_capture_clock(capture_dir)
    repeated_summary = json.loads((capture_dir / "mqtt.summary.json").read_text(encoding="utf-8"))
    assert repeated == sealed
    assert repeated_summary == updated


def test_owner_seal_is_retryable_if_summary_pointer_switch_fails(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    capture_dir = tmp_path / "capture"
    capture_mqtt(
        "192.168.1.50",
        capture_dir,
        duration_seconds=30,
        _client_factory=lambda: cast(mqtt.Client, FakeClient()),
    )
    provisional_summary = (capture_dir / "mqtt.summary.json").read_bytes()

    with monkeypatch.context() as context:
        context.setattr(
            capture_module,
            "_write_json_atomic_replace",
            lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("injected switch failure")),
        )
        with pytest.raises(CaptureError, match="injected switch failure"):
            seal_capture_clock(capture_dir)

    assert (capture_dir / "mqtt.summary.json").read_bytes() == provisional_summary
    assert list(capture_dir.glob("mqtt.timeline.session-*.json"))
    sealed = seal_capture_clock(capture_dir)
    summary = json.loads((capture_dir / "mqtt.summary.json").read_text(encoding="utf-8"))
    assert summary["capture_clock_scope"] == "session"
    assert summary["artifact_hashes"]["capture_clock_sha256"] == sealed.artifact_sha256


def test_preview_failure_happens_after_raw_message_is_preserved(tmp_path: Path) -> None:
    fake = FakeClient()
    capture_dir = tmp_path / "capture"

    def fail_preview(_message: object) -> None:
        raise ValueError("synthetic preview failure")

    with pytest.raises(CaptureError, match="preview callback failed") as error:
        capture_mqtt(
            "192.168.1.50",
            capture_dir,
            on_message_recorded=fail_preview,
            _client_factory=lambda: cast(mqtt.Client, fake),
        )

    assert error.value.summary is not None
    assert error.value.summary["message_count"] == 1
    frames = list(iter_capture_frames(capture_dir / "mqtt.raw.k1mqtt"))
    assert frames[0].payload == fake.payload


@pytest.mark.parametrize(
    ("raw", "message"),
    [
        (b"not-mqtt", "magic/version"),
        (RAW_MAGIC + b"\x00", "truncated length header"),
        (RAW_MAGIC + FRAME_HEADER.pack(4, 1) + b"ab", "topic .* is truncated"),
        (RAW_MAGIC + FRAME_HEADER.pack(1, 4) + b"t" + b"ab", "payload .* is truncated"),
        (RAW_MAGIC + FRAME_HEADER.pack(1, 5) + b"t" + b"abcde", "exceeds 4"),
    ],
)
def test_capture_reader_rejects_invalid_or_unbounded_frames(
    tmp_path: Path,
    raw: bytes,
    message: str,
) -> None:
    path = tmp_path / "capture.raw"
    path.write_bytes(raw)

    with pytest.raises(CaptureFormatError, match=message):
        list(iter_capture_frames(path, max_payload_bytes=4))


def _mqtt_message(topic: str, payload: bytes) -> mqtt.MQTTMessage:
    message = mqtt.MQTTMessage(topic=topic.encode("utf-8"))
    message.payload = payload
    message.qos = 0
    message.retain = False
    message.dup = False
    return message


def test_group_commit_fsyncs_raw_before_publishing_metadata(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    writer = capture_module._CaptureWriter(tmp_path / "capture", 1024)  # noqa: SLF001
    writer.open()
    assert writer._raw is not None  # noqa: SLF001
    assert writer._metadata is not None  # noqa: SLF001
    raw_fd = writer._raw.fileno()  # noqa: SLF001
    metadata_fd = writer._metadata.fileno()  # noqa: SLF001
    fsync_calls: list[int] = []
    real_fsync = capture_module.os.fsync

    def observe_fsync(descriptor: int) -> None:
        fsync_calls.append(descriptor)
        real_fsync(descriptor)

    monkeypatch.setattr(capture_module.os, "fsync", observe_fsync)
    monkeypatch.setattr(capture_module, "GROUP_COMMIT_MAX_MESSAGES", 2)

    writer.record(_mqtt_message("RealtimePath", b"one"))
    assert writer.metadata_path.read_bytes() == b""
    writer.record(_mqtt_message("RealtimePath", b"two"))

    assert fsync_calls[:2] == [raw_fd, metadata_fd]
    assert len(writer.metadata_path.read_text(encoding="utf-8").splitlines()) == 2
    writer.close()


def test_group_commit_timer_bounds_quiet_stream_rpo(
    tmp_path: Path,
) -> None:
    writer = capture_module._CaptureWriter(tmp_path / "capture", 1024)  # noqa: SLF001
    writer.open()
    writer.record(_mqtt_message("RealtimePath", b"one"))
    assert writer.metadata_path.read_bytes() == b""

    writer.maybe_commit(
        writer._last_commit_monotonic  # noqa: SLF001
        + capture_module.GROUP_COMMIT_INTERVAL_SECONDS
    )

    assert len(writer.metadata_path.read_text(encoding="utf-8").splitlines()) == 1
    writer.close()


def test_raw_fsync_failure_never_publishes_metadata_ahead_of_raw(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    writer = capture_module._CaptureWriter(tmp_path / "capture", 1024)  # noqa: SLF001
    writer.open()
    writer.record(_mqtt_message("RealtimePath", b"one"))
    assert writer._raw is not None  # noqa: SLF001
    raw_fd = writer._raw.fileno()  # noqa: SLF001
    real_fsync = capture_module.os.fsync

    def fail_raw_fsync(descriptor: int) -> None:
        if descriptor == raw_fd:
            raise OSError("synthetic raw fsync failure")
        real_fsync(descriptor)

    monkeypatch.setattr(capture_module.os, "fsync", fail_raw_fsync)

    with pytest.raises(OSError, match="synthetic raw fsync failure"):
        writer._commit_pending()  # noqa: SLF001

    assert writer.metadata_path.read_bytes() == b""
    # Restore durability primitive so the fixture can close normally.
    monkeypatch.setattr(capture_module.os, "fsync", real_fsync)
    writer.close()