from __future__ import annotations

import hashlib
import json
import shutil
import sqlite3
from pathlib import Path

import pytest

from k1link.device_plugins.xgrids_k1 import xgrids_k1_archive_source
from k1link.device_plugins.xgrids_k1.mqtt.capture import (
    FRAME_HEADER,
    RAW_MAGIC,
    iter_capture_frames,
)
from k1link.sessions import (
    LayoutConflictError,
    SessionIntegrityError,
    SessionNotFoundError,
    SessionStore,
    resolve_missioncore_evidence_dir,
)


def make_legacy_session(
    sessions_root: Path,
    session_id: str,
    *,
    created_at: str = "2026-07-16T20:56:32.699Z",
) -> Path:
    session = sessions_root / session_id
    capture = session / "captures" / "mqtt_live"
    capture.mkdir(parents=True)
    frames = [
        ("lixel/application/report/lio_pcl", b"point-frame"),
        ("lixel/application/report/lio_pose", b"pose-frame"),
    ]
    raw = bytearray(RAW_MAGIC)
    metadata: list[dict[str, object]] = []
    for sequence, (topic, payload) in enumerate(frames, start=1):
        topic_bytes = topic.encode()
        frame_offset = len(raw)
        raw.extend(FRAME_HEADER.pack(len(topic_bytes), len(payload)))
        raw.extend(topic_bytes)
        raw.extend(payload)
        metadata.append(
            {
                "schema_version": 1,
                "record_type": "message",
                "sequence": sequence,
                "received_at_utc": f"2026-07-16T20:56:{31 + sequence:02d}.699Z",
                "received_at_epoch_ns": 1_784_235_391_699_000_000 + sequence * 1_000_000_000,
                "received_monotonic_ns": 9_000_000_000 + sequence * 1_000_000_000,
                "topic": topic,
                "payload_bytes": len(payload),
                "raw_frame_offset": frame_offset,
                "raw_payload_offset": frame_offset + FRAME_HEADER.size + len(topic_bytes),
                "raw_frame_bytes": FRAME_HEADER.size + len(topic_bytes) + len(payload),
            }
        )
    raw_path = capture / "mqtt.raw.k1mqtt"
    raw_path.write_bytes(raw)
    raw_hash = hashlib.sha256(raw).hexdigest()
    metadata_path = capture / "mqtt.metadata.jsonl"
    metadata_path.write_text(
        "".join(json.dumps(record, separators=(",", ":")) + "\n" for record in metadata),
        encoding="utf-8",
    )
    metadata_hash = hashlib.sha256(metadata_path.read_bytes()).hexdigest()
    (capture / "mqtt.summary.json").write_text(
        json.dumps(
            {
                "created_at_utc": created_at,
                "completed_at_utc": "2026-07-16T21:20:43.018Z",
                "capture_elapsed_seconds": 1440.1,
                "stop_reason": "external_stop",
                "error": None,
                "message_count": 2,
                "raw_bytes": len(raw),
                "payload_bytes": sum(len(payload) for _, payload in frames),
                "topic_counts": {
                    "lixel/application/report/lio_pcl": 1,
                    "lixel/application/report/lio_pose": 1,
                },
                "artifact_hashes": {
                    "raw_sha256": raw_hash,
                    "metadata_jsonl_sha256": metadata_hash,
                },
                # This is deliberately sensitive and must not enter API DTOs.
                "target_ipv4": "192.168.99.77",
            }
        ),
        encoding="utf-8",
    )
    (session / "manifest.redacted.json").write_text(
        json.dumps({"started_at_utc": created_at, "target": "redacted"}),
        encoding="utf-8",
    )
    return session


def add_capture_clock(session: Path, *, scope: str = "session") -> Path:
    capture = session / "captures" / "mqtt_live"
    origin_document = {
        "schema_version": 1,
        "started_at_epoch_ns": 1_784_235_391_699_000_000,
        "started_monotonic_ns": 9_000_000_000,
    }
    origin_path = capture / "mqtt.timeline.origin.json"
    origin_path.write_text(
        json.dumps(origin_document, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )
    clock_document = {
        **origin_document,
        "completed_at_epoch_ns": 1_784_235_394_699_000_000,
        "completed_monotonic_ns": 12_000_000_000,
    }
    clock_payload = (json.dumps(clock_document, separators=(",", ":")) + "\n").encode()
    clock_digest = hashlib.sha256(clock_payload).hexdigest()
    provisional_path = capture / "mqtt.timeline.json"
    provisional_path.write_bytes(clock_payload)
    clock_path = (
        capture / f"mqtt.timeline.session-{clock_digest}.json"
        if scope == "session"
        else provisional_path
    )
    if clock_path != provisional_path:
        clock_path.write_bytes(clock_payload)
    summary_path = capture / "mqtt.summary.json"
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["schema_version"] = 2
    summary["capture_clock_scope"] = scope
    summary["session_elapsed_seconds"] = 3.0
    summary["artifacts"] = {
        "raw": "mqtt.raw.k1mqtt",
        "metadata_jsonl": "mqtt.metadata.jsonl",
        "capture_clock_origin": origin_path.name,
        "capture_clock": clock_path.name,
        "summary": "mqtt.summary.json",
    }
    summary["artifact_hashes"]["capture_clock_origin_sha256"] = hashlib.sha256(
        origin_path.read_bytes()
    ).hexdigest()
    summary["artifact_hashes"]["capture_clock_sha256"] = hashlib.sha256(
        clock_path.read_bytes()
    ).hexdigest()
    summary_path.write_text(json.dumps(summary), encoding="utf-8")
    return clock_path


def test_evidence_root_is_private_and_configurable(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    repository = tmp_path / "repo"
    assert (
        resolve_missioncore_evidence_dir(repository)
        == (repository / ".runtime" / "mission-core" / "evidence" / "sessions").resolve()
    )

    configured = tmp_path / "external-evidence"
    monkeypatch.setenv("MISSIONCORE_EVIDENCE_DIR", str(configured))
    assert resolve_missioncore_evidence_dir(repository) == configured.resolve()


def test_catalog_reconciles_sessions_removed_from_one_evidence_root(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")

    assert store.reconcile_archive(xgrids_k1_archive_source(sessions)) == (session.name,)
    shutil.rmtree(session)

    assert store.reconcile_archive(xgrids_k1_archive_source(sessions)) == ()
    assert store.list_recent().items == ()


def test_corrupt_candidate_does_not_starve_later_valid_session(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    corrupt = sessions / "20260716T125502Z_viewer_live"
    corrupt.mkdir(parents=True)
    (corrupt / "manifest.redacted.json").write_text("{}", encoding="utf-8")
    valid = make_legacy_session(sessions, "20260718T201659Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")

    assert store.reconcile_archive(xgrids_k1_archive_source(sessions)) == (valid.name,)
    assert [item.session_id for item in store.list_recent().items] == [valid.name]


def test_present_corrupt_candidate_does_not_delete_its_last_valid_catalog_row(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    retained = make_legacy_session(sessions, "20260716T125502Z_viewer_live")
    later = make_legacy_session(sessions, "20260718T201659Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")
    source = xgrids_k1_archive_source(sessions)
    assert store.reconcile_archive(source) == (retained.name, later.name)

    shutil.rmtree(retained / "captures")

    assert store.reconcile_archive(source) == (later.name,)
    assert {item.session_id for item in store.list_recent().items} == {
        retained.name,
        later.name,
    }


def test_catalog_uses_valid_project_name_as_session_display_name(tmp_path: Path) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    manifest_path = session / "manifest.redacted.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    manifest["project_name"] = "  Ｋ１ Route Alpha  "
    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    assert store.list_recent().items[0].display_name == "K1 Route Alpha"


def test_catalog_rejects_non_utf8_project_display_name(tmp_path: Path) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    manifest_path = session / "manifest.redacted.json"
    manifest_path.write_text(
        json.dumps({"project_name": "unsafe-\ud800"}),
        encoding="utf-8",
    )
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    assert store.list_recent().items[0].display_name == session.name


def make_recorded_camera_source(
    session: Path,
    source_id: str = "sensor.camera.left",
    *,
    complete: bool = True,
) -> Path:
    epoch = session / "media" / source_id / "epoch-1"
    segments = epoch / "segments"
    segments.mkdir(parents=True)
    (epoch / "init.mp4").write_bytes(b"ftyp-mission-core")
    (segments / "1.m4s").write_bytes(b"moof-camera-frame")
    (epoch / "index.jsonl").write_text(
        json.dumps({"sequence": 1, "started_at_seconds": 0.0}) + "\n",
        encoding="utf-8",
    )
    if complete:
        (epoch / "summary.json").write_text(
            json.dumps(
                {
                    "schema_version": "missioncore.camera-recording/v1",
                    "source_id": source_id,
                    "segment_count": 1,
                }
            ),
            encoding="utf-8",
        )
    return epoch


def replace_summary_with_recovery_metadata(
    session: Path,
    *,
    corrupt_trailing_line: bool = False,
) -> None:
    capture = session / "captures" / "mqtt_live"
    raw_path = capture / "mqtt.raw.k1mqtt"
    timestamps = (
        ("2026-07-16T20:56:32.699Z", 1_784_235_392_699_000_000, 10_000_000_000),
        ("2026-07-16T20:56:35.199Z", 1_784_235_395_199_000_000, 12_500_000_000),
    )
    records = []
    for frame, (timestamp, epoch_ns, monotonic_ns) in zip(
        iter_capture_frames(raw_path),
        timestamps,
        strict=True,
    ):
        records.append(
            {
                "schema_version": 1,
                "record_type": "message",
                "sequence": frame.sequence,
                "received_at_utc": timestamp,
                "received_at_epoch_ns": epoch_ns,
                "received_monotonic_ns": monotonic_ns,
                "topic": frame.topic,
                "payload_bytes": frame.raw_frame_bytes
                - FRAME_HEADER.size
                - len(frame.topic.encode("utf-8")),
                "raw_frame_offset": frame.raw_frame_offset,
                "raw_payload_offset": frame.raw_payload_offset,
                "raw_frame_bytes": frame.raw_frame_bytes,
            }
        )
    metadata = b"".join(
        (json.dumps(record, separators=(",", ":")) + "\n").encode("utf-8") for record in records
    )
    if corrupt_trailing_line:
        metadata += b'{"record_type":"message"'
    (capture / "mqtt.metadata.jsonl").write_bytes(metadata)
    (capture / "mqtt.summary.json").write_text("{corrupt", encoding="utf-8")


def serialized(value: object) -> str:
    return json.dumps(value, ensure_ascii=False, default=str)


def test_store_uses_private_wal_database_and_idempotently_imports_legacy_session(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")

    first = store.reconcile_archive(xgrids_k1_archive_source(sessions))
    second = store.reconcile_archive(xgrids_k1_archive_source(sessions))

    assert first == second == ("20260716T205632Z_viewer_live",)
    page = store.list_recent()
    assert len(page.items) == 1
    summary = page.items[0]
    assert summary.status == "ready"
    assert summary.modalities == ("point-cloud", "trajectory")
    assert summary.replayable is True
    assert summary.source_count == 2
    assert "192.168.99.77" not in serialized(page.as_dict())
    assert str(repository) not in serialized(page.as_dict())

    detail = store.get_session(summary.session_id)
    assert [source.source_id for source in detail.sources] == [
        "sensor.lidar.primary",
        "spatial.trajectory",
    ]
    assert all(source.seekable for source in detail.sources)
    assert str(repository) not in serialized(detail.as_dict())
    command = store.prepare_replay(summary.session_id, speed=2.0, loop=True)
    assert command.primary_artifact.path.name == "mqtt.raw.k1mqtt"
    assert command.speed == 2.0
    assert command.loop is True

    with sqlite3.connect(store.database_path) as connection:
        assert connection.execute("PRAGMA journal_mode").fetchone()[0] == "wal"
    assert store.database_path.stat().st_mode & 0o777 == 0o600


def test_completed_capture_clock_is_a_digest_bound_replay_artifact(tmp_path: Path) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    clock_path = add_capture_clock(session)
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    assert store.get_session(session.name).summary.duration_seconds == 3.0
    command = store.prepare_replay(session.name)
    clock_artifact = next(
        artifact for artifact in command.artifacts if artifact.artifact_id == "raw-transport-clock"
    )
    assert command.timeline_origin_monotonic_ns == 9_000_000_000
    assert command.timeline_origin_epoch_ns == 1_784_235_391_699_000_000
    assert clock_artifact.path == clock_path
    assert clock_artifact.replay_byte_length == clock_path.stat().st_size
    assert clock_artifact.expected_sha256 == hashlib.sha256(clock_path.read_bytes()).hexdigest()


def test_completed_capture_rejects_corrupt_declared_clock(tmp_path: Path) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    clock_path = add_capture_clock(session)
    clock_path.write_text('{"schema_version":1,"tampered":true}\n', encoding="utf-8")
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "failed"
    assert detail.summary.replayable is False


def test_completed_capture_rejects_clock_filename_with_wrong_digest_suffix(
    tmp_path: Path,
) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    clock_path = add_capture_clock(session)
    wrong_path = clock_path.with_name(f"mqtt.timeline.session-{'0' * 64}.json")
    wrong_path.write_bytes(clock_path.read_bytes())
    summary_path = session / "captures" / "mqtt_live" / "mqtt.summary.json"
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["artifacts"]["capture_clock"] = wrong_path.name
    summary_path.write_text(json.dumps(summary), encoding="utf-8")
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "failed"
    assert detail.summary.replayable is False


def test_camera_session_does_not_advertise_provisional_transport_clock(
    tmp_path: Path,
) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    add_capture_clock(session, scope="transport")
    make_recorded_camera_source(session)
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.replayable is False


def test_interrupted_camera_session_uses_durable_origin_and_fails_closed(
    tmp_path: Path,
) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    add_capture_clock(session, scope="transport")
    replace_summary_with_recovery_metadata(session)
    make_recorded_camera_source(session)
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "interrupted"
    assert detail.summary.replayable is False


def test_interrupted_raw_replay_preserves_durable_pre_message_origin(tmp_path: Path) -> None:
    sessions = tmp_path / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    add_capture_clock(session, scope="transport")
    replace_summary_with_recovery_metadata(session)
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    command = store.prepare_replay(session.name)
    assert command.timeline_origin_monotonic_ns == 9_000_000_000
    assert any(
        artifact.artifact_id == "raw-transport-clock-origin" for artifact in command.artifacts
    )


def test_reconcile_claims_pre_plugin_catalog_row_without_changing_session_identity(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")
    source = xgrids_k1_archive_source(sessions)
    store.reconcile_archive(source)
    with sqlite3.connect(store.database_path) as connection:
        connection.execute(
            "UPDATE observation_sessions SET plugin_id = '', archive_id = '', "
            "primary_replay_artifact_id = NULL, timeline_origin_epoch_ns = NULL, "
            "timeline_origin_monotonic_ns = NULL WHERE session_id = ?",
            (session.name,),
        )
        connection.execute(
            "UPDATE observation_session_artifacts SET replay_byte_length = 0 WHERE session_id = ?",
            (session.name,),
        )
        connection.commit()

    assert store.reconcile_archive(source) == (session.name,)
    command = store.prepare_replay(session.name)
    assert command.plugin_id == source.plugin_id
    assert command.primary_artifact.replay_byte_length > 0


def test_recent_sessions_are_sorted_and_cursor_paginated(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    make_legacy_session(
        sessions,
        "20260716T191025Z_viewer_live",
        created_at="2026-07-16T19:10:25.352Z",
    )
    make_legacy_session(
        sessions,
        "20260716T205632Z_viewer_live",
        created_at="2026-07-16T20:56:32.699Z",
    )
    store = SessionStore(repository, data_dir=tmp_path / "data")
    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    first = store.list_recent(limit=1)
    assert first.items[0].session_id == "20260716T205632Z_viewer_live"
    assert first.next_cursor == "20260716T205632Z_viewer_live"
    second = store.list_recent(limit=1, cursor=first.next_cursor)
    assert second.items[0].session_id == "20260716T191025Z_viewer_live"
    assert second.next_cursor is None


def test_legacy_import_adds_video_only_for_validated_recording_tree(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    complete = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    incomplete = make_legacy_session(sessions, "20260716T205632Z_viewer_live_2")
    make_recorded_camera_source(complete)
    make_recorded_camera_source(incomplete, complete=False)
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    complete_detail = store.get_session(complete.name)
    assert complete_detail.summary.modalities == ("point-cloud", "trajectory", "video")
    assert complete_detail.summary.source_count == 3
    camera = next(
        source for source in complete_detail.sources if source.source_id == "sensor.camera.left"
    )
    assert camera.modality == "video"
    assert camera.semantic_channel_id == "camera.video.recorded"
    assert camera.seekable is True
    video_artifact = next(
        artifact
        for artifact in complete_detail.artifacts
        if artifact.artifact_id == camera.artifact_id
    )
    assert video_artifact.kind == "recorded-video"
    assert video_artifact.integrity_status == "validated-structure"
    assert str(repository) not in serialized(complete_detail.as_dict())

    incomplete_detail = store.get_session(incomplete.name)
    assert incomplete_detail.summary.modalities == ("point-cloud", "trajectory")
    assert all(source.modality != "video" for source in incomplete_detail.sources)


def test_interrupted_capture_is_recovered_from_aligned_raw_and_metadata_prefix(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    replace_summary_with_recovery_metadata(session, corrupt_trailing_line=True)
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "interrupted"
    assert detail.summary.replayable is True
    assert detail.summary.modalities == ("point-cloud", "trajectory")
    assert detail.summary.started_at_utc == "2026-07-16T20:56:32.699Z"
    assert detail.summary.completed_at_utc == "2026-07-16T20:56:35.199Z"
    assert detail.summary.duration_seconds == 2.5
    assert detail.artifacts[0].integrity_status == "validated-prefix"
    assert store.prepare_replay(session.name).primary_artifact.path.name == "mqtt.raw.k1mqtt"


def test_interrupted_capture_streams_metadata_beyond_legacy_total_limit(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    capture = session / "captures" / "mqtt_live"
    raw = bytearray(RAW_MAGIC)
    records: list[bytes] = []
    topic = "lixel/application/report/lio_pose"
    topic_bytes = topic.encode()
    padding = "x" * (60 * 1024)
    for sequence in range(1, 1_101):
        payload = b"p"
        frame_offset = len(raw)
        raw.extend(FRAME_HEADER.pack(len(topic_bytes), len(payload)))
        raw.extend(topic_bytes)
        raw.extend(payload)
        record = {
            "schema_version": 1,
            "record_type": "message",
            "sequence": sequence,
            "received_at_utc": "2026-07-16T20:56:32.699Z",
            "received_at_epoch_ns": 1_784_235_391_699_000_000 + sequence,
            "received_monotonic_ns": 9_000_000_000 + sequence,
            "topic": topic,
            "payload_bytes": len(payload),
            "raw_frame_offset": frame_offset,
            "raw_payload_offset": frame_offset + FRAME_HEADER.size + len(topic_bytes),
            "raw_frame_bytes": FRAME_HEADER.size + len(topic_bytes) + len(payload),
            "padding": padding,
        }
        records.append((json.dumps(record, separators=(",", ":")) + "\n").encode())
    (capture / "mqtt.raw.k1mqtt").write_bytes(raw)
    metadata_path = capture / "mqtt.metadata.jsonl"
    metadata_path.write_bytes(b"".join(records))
    assert metadata_path.stat().st_size > 64 * 1024 * 1024
    (capture / "mqtt.summary.json").write_text("{corrupt", encoding="utf-8")
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "interrupted"
    assert detail.summary.replayable is True
    command = store.prepare_replay(session.name)
    assert command.primary_artifact.replay_byte_length == len(raw)
    assert next(
        artifact.replay_byte_length
        for artifact in command.artifacts
        if artifact.artifact_id == "raw-transport-index"
    ) == metadata_path.stat().st_size


def test_catalog_upsert_promotes_recovered_session_after_summary_is_completed(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    summary_path = session / "captures" / "mqtt_live" / "mqtt.summary.json"
    metadata_path = session / "captures" / "mqtt_live" / "mqtt.metadata.jsonl"
    completed_summary = summary_path.read_text(encoding="utf-8")
    completed_metadata = metadata_path.read_text(encoding="utf-8")
    replace_summary_with_recovery_metadata(session)
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))
    assert store.get_session(session.name).summary.status == "interrupted"

    summary_path.write_text(completed_summary, encoding="utf-8")
    metadata_path.write_text(completed_metadata, encoding="utf-8")
    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    promoted = store.get_session(session.name).summary
    assert promoted.status == "ready"
    assert promoted.duration_seconds == 1440.1
    assert promoted.replayable is True


def test_interrupted_capture_does_not_trust_metadata_outside_session(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    replace_summary_with_recovery_metadata(session)
    metadata = session / "captures" / "mqtt_live" / "mqtt.metadata.jsonl"
    outside = tmp_path / "outside.metadata.jsonl"
    outside.write_bytes(metadata.read_bytes())
    metadata.unlink()
    metadata.symlink_to(outside)
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "failed"
    assert detail.summary.replayable is False
    assert detail.summary.modalities == ()


def test_interrupted_capture_rejects_newline_terminated_metadata_corruption(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    replace_summary_with_recovery_metadata(session)
    metadata = session / "captures" / "mqtt_live" / "mqtt.metadata.jsonl"
    with metadata.open("ab") as stream:
        stream.write(b"{corrupt\n")
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "failed"
    assert detail.summary.replayable is False


def test_replay_resolution_rejects_artifact_symlink_escape(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")
    store.reconcile_archive(xgrids_k1_archive_source(sessions))
    raw = session / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
    outside = tmp_path / "outside.k1mqtt"
    outside.write_bytes(raw.read_bytes())
    raw.unlink()
    raw.symlink_to(outside)

    with pytest.raises(SessionIntegrityError, match="escapes"):
        store.prepare_replay(session.name)


def test_completed_capture_is_not_replayable_when_declared_integrity_fails(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    raw = session / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
    corrupted = bytearray(raw.read_bytes())
    corrupted[-1] ^= 0x01
    raw.write_bytes(corrupted)
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "failed"
    assert detail.summary.replayable is False
    assert detail.summary.modalities == ()


def test_completed_capture_is_not_replayable_when_declared_count_is_wrong(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    summary_path = session / "captures" / "mqtt_live" / "mqtt.summary.json"
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["message_count"] = 3
    summary_path.write_text(json.dumps(summary), encoding="utf-8")
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    assert store.get_session(session.name).summary.replayable is False


def test_current_session_marker_keeps_active_capture_out_of_replay(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    (sessions / ".current_session").write_text(
        f"sessions/{session.name}\n",
        encoding="utf-8",
    )
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    with pytest.raises(SessionNotFoundError):
        store.get_session(session.name)

    (sessions / ".current_session").unlink()
    store.reconcile_archive(xgrids_k1_archive_source(sessions))
    assert store.get_session(session.name).summary.replayable is True


def test_delete_session_removes_only_the_exact_evidence_tree_and_catalog_row(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    deleted = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    retained = make_legacy_session(sessions, "20260716T205633Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")
    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    store.delete_session(deleted.name)

    assert not deleted.exists()
    assert retained.is_dir()
    with pytest.raises(SessionNotFoundError):
        store.get_session(deleted.name)
    assert store.get_session(retained.name).summary.session_id == retained.name


def test_delete_session_rejects_a_catalog_target_outside_its_allowed_root(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    store = SessionStore(repository, data_dir=tmp_path / "data")
    store.reconcile_archive(xgrids_k1_archive_source(sessions))
    with sqlite3.connect(store.database_path) as connection:
        connection.execute(
            "UPDATE observation_sessions SET session_root = ? WHERE session_id = ?",
            (str(repository), session.name),
        )

    with pytest.raises(SessionIntegrityError, match="escapes"):
        store.delete_session(session.name)

    assert session.is_dir()


def test_interrupted_capture_replays_only_committed_prefix_before_partial_raw_tail(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    replace_summary_with_recovery_metadata(session)
    raw = session / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
    committed_bytes = raw.stat().st_size
    with raw.open("ab") as stream:
        stream.write(FRAME_HEADER.pack(12, 100)[:7])
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    command = store.prepare_replay(session.name)
    assert detail.summary.status == "interrupted"
    assert detail.summary.replayable is True
    assert command.primary_artifact.replay_byte_length == committed_bytes
    assert (
        command.primary_artifact.replay_byte_length < command.primary_artifact.path.stat().st_size
    )


def test_interrupted_capture_rejects_non_frame_raw_tail(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    replace_summary_with_recovery_metadata(session)
    raw = session / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
    with raw.open("ab") as stream:
        stream.write(FRAME_HEADER.pack(0, 0))
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    assert store.get_session(session.name).summary.replayable is False


def test_interrupted_capture_tolerates_bounded_group_commit_raw_tail(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    replace_summary_with_recovery_metadata(session)
    raw = session / "captures" / "mqtt_live" / "mqtt.raw.k1mqtt"
    committed_bytes = raw.stat().st_size
    with raw.open("ab") as stream:
        for payload in (b"pending-one", b"pending-two"):
            topic = b"RealtimePath"
            stream.write(FRAME_HEADER.pack(len(topic), len(payload)))
            stream.write(topic)
            stream.write(payload)
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    command = store.prepare_replay(session.name)
    assert store.get_session(session.name).summary.status == "interrupted"
    assert command.primary_artifact.replay_byte_length == committed_bytes


def test_failed_final_summary_falls_back_to_last_committed_prefix(tmp_path: Path) -> None:
    repository = tmp_path / "repo"
    sessions = repository / "sessions"
    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
    capture = session / "captures" / "mqtt_live"
    metadata = capture / "mqtt.metadata.jsonl"
    first_record = metadata.read_text(encoding="utf-8").splitlines(keepends=True)[0]
    metadata.write_text(first_record, encoding="utf-8")
    summary_path = capture / "mqtt.summary.json"
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    summary["error"] = "synthetic metadata fsync failure"
    summary_path.write_text(json.dumps(summary), encoding="utf-8")
    store = SessionStore(repository, data_dir=tmp_path / "data")

    store.reconcile_archive(xgrids_k1_archive_source(sessions))

    detail = store.get_session(session.name)
    assert detail.summary.status == "interrupted"
    assert detail.summary.replayable is True
    assert detail.summary.modalities == ("point-cloud",)


def test_layout_save_is_atomic_and_revision_checked(tmp_path: Path) -> None:
    store = SessionStore(tmp_path / "repo", data_dir=tmp_path / "data")

    first = store.save_layout(
        "observation.spatial",
        schema_version=1,
        expected_revision=0,
        name="Операторская сцена",
        layout={"visible_source_ids": ["sensor.lidar.primary"], "windows": []},
    )
    assert first.revision == 1
    assert store.get_layout("observation.spatial") == first

    with pytest.raises(LayoutConflictError, match="revision changed"):
        store.save_layout(
            "observation.spatial",
            schema_version=1,
            expected_revision=0,
            name="Устаревшая запись",
            layout={},
        )

    second = store.save_layout(
        "observation.spatial",
        schema_version=1,
        expected_revision=1,
        name="Операторская сцена",
        layout={"visible_source_ids": [], "windows": []},
    )
    assert second.revision == 2
    assert store.get_layout("observation.spatial").layout["visible_source_ids"] == []


def test_layout_startup_migration_removes_transient_tool_window_state(
    tmp_path: Path,
) -> None:
    repository = tmp_path / "repo"
    data_dir = tmp_path / "data"
    initial = SessionStore(repository, data_dir=data_dir)
    initial.save_layout(
        "observation.spatial",
        schema_version=1,
        expected_revision=0,
        name="Пространственная сцена",
        layout={
            "tool_windows": {
                "sources_open": True,
                "display_open": True,
                "layers_open": True,
                "order": ["sources", "display", "layers"],
            },
            "visible_source_ids": [],
        },
    )

    migrated = SessionStore(repository, data_dir=data_dir).get_layout("observation.spatial")

    assert migrated.schema_version == 2
    assert "tool_windows" not in migrated.layout