# SPDX-License-Identifier: AGPL-3.0-only
# Copyright (c) 2026 sol pbc

from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path

import pytest

from solstone.think import catchup_state, media

DAY = "20260101"
CMD_DAILY = ["journal", "think", "-v", "--day", DAY]
CMD_FROM_SCRATCH = ["journal", "think", "-v", "--day", DAY, "--from-scratch"]
CMD_SEGMENT = ["journal", "think", "-v", "--day", DAY, "--segment", "120000_300"]


@pytest.fixture
def journal(tmp_path, monkeypatch):
    monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
    return tmp_path


def _day_dir(journal: Path, day: str = DAY) -> Path:
    path = journal / "chronicle" / day
    path.mkdir(parents=True, exist_ok=True)
    return path


def _segment(
    journal: Path,
    day: str = DAY,
    stream: str = "default",
    segment: str = "120000_300",
) -> Path:
    path = _day_dir(journal, day) / stream / segment
    path.mkdir(parents=True, exist_ok=True)
    return path


def _touch_marker(journal: Path, day: str, name: str, mtime: float) -> Path:
    path = _day_dir(journal, day) / "health" / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.touch()
    os.utime(path, (mtime, mtime))
    return path


def _state_file(journal: Path) -> Path:
    path = journal / "health" / "catchup-state.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    return path


def _write_state(journal: Path, entries: dict[str, dict]) -> None:
    _state_file(journal).write_text(
        json.dumps({"version": catchup_state.STATE_VERSION, "entries": entries}),
        encoding="utf-8",
    )


def _read_entries(journal: Path) -> dict:
    return json.loads(_state_file(journal).read_text(encoding="utf-8"))["entries"]


def _record(
    day: str,
    kind: str,
    *,
    attempts: int = 1,
    consecutive: int = 0,
    last_outcome: str = "",
    next_retry_at: float = 0,
    entered_backoff_at: float | None = None,
    active: dict | None = None,
    fingerprint: str | None = None,
    reason_code: str | None = None,
    timeout_seconds: float | None = None,
    bounded: bool | None = None,
    cleared: int | None = None,
    remaining: int | None = None,
    exit_reason: str | None = None,
    daily_progress: dict | None = None,
) -> dict:
    return {
        "day": day,
        "command_kind": kind,
        "attempts": attempts,
        "consecutive_non_completion": consecutive,
        "last_attempt_at": 0,
        "last_outcome": last_outcome,
        "next_retry_at": next_retry_at,
        "entered_backoff_at": entered_backoff_at,
        "notified_at": entered_backoff_at,
        "fingerprint": fingerprint,
        "active": active,
        "reason_code": reason_code,
        "timeout_seconds": timeout_seconds,
        "bounded": bounded,
        "cleared": cleared,
        "remaining": remaining,
        "exit_reason": exit_reason,
        "daily_progress": daily_progress,
    }


def test_derive_command_kind_and_extract_day():
    assert (
        catchup_state.derive_command_kind(CMD_DAILY) == catchup_state.KIND_DAILY_CATCHUP
    )
    assert (
        catchup_state.derive_command_kind(CMD_FROM_SCRATCH)
        == catchup_state.KIND_DAILY_FROM_SCRATCH
    )
    assert catchup_state.derive_command_kind(CMD_SEGMENT) == catchup_state.KIND_SEGMENT
    assert (
        catchup_state.derive_command_kind(
            ["journal", "think", "--day", DAY, "--segments"]
        )
        == catchup_state.KIND_SEGMENT
    )
    for flag in ("--flush", "--activity", "--weekly", "--cadence"):
        assert (
            catchup_state.derive_command_kind(["journal", "think", "--day", DAY, flag])
            is None
        )
    assert catchup_state.derive_command_kind(["journal", "think"]) is None
    assert catchup_state.extract_day(CMD_DAILY) == DAY
    assert catchup_state.extract_day(["journal", "think", "--day"]) is None


def test_raw_input_fingerprint_stability_and_allowlist(journal):
    first = _segment(journal)
    (first / "audio.jsonl").write_text('{"raw":"a.flac"}\n', encoding="utf-8")
    (first / "screen.webm").write_bytes(b"aa")
    (first / "stream.json").write_text("{}\n", encoding="utf-8")
    talents = first / "talents"
    talents.mkdir()
    (talents / "x.md").write_text("derived\n", encoding="utf-8")
    (first / ".analyzing_screen").write_text("busy\n", encoding="utf-8")
    (_day_dir(journal) / "health").mkdir(exist_ok=True)
    (_day_dir(journal) / "health" / "segment.jsonl").write_text(
        "derived\n", encoding="utf-8"
    )

    original = catchup_state.read_raw_input_fingerprint(DAY)
    assert catchup_state.read_raw_input_fingerprint(DAY) == original

    (first / "stream.json").write_text('{"stream":"default"}\n', encoding="utf-8")
    (talents / "x.md").write_text("changed\n", encoding="utf-8")
    (first / ".analyzing_screen").write_text("changed\n", encoding="utf-8")
    (_day_dir(journal) / "health" / "segment.jsonl").write_text(
        "changed\n", encoding="utf-8"
    )
    assert catchup_state.read_raw_input_fingerprint(DAY) == original

    (first / "audio.jsonl").write_text('{"raw":"a.flac"}\n', encoding="utf-8")
    assert catchup_state.read_raw_input_fingerprint(DAY) == original

    (first / "screen.webm").write_bytes(b"bb")
    assert catchup_state.read_raw_input_fingerprint(DAY) == original

    (first / "screen.webm").write_bytes(b"bbb")
    media_changed = catchup_state.read_raw_input_fingerprint(DAY)
    assert media_changed != original

    (first / "photo.jpg").write_bytes(b"image")
    image_changed = catchup_state.read_raw_input_fingerprint(DAY)
    assert image_changed != media_changed

    (first / "audio.jsonl").write_text('{"raw":"different.flac"}\n', encoding="utf-8")
    raw_changed = catchup_state.read_raw_input_fingerprint(DAY)
    assert raw_changed != image_changed

    second = _segment(journal, segment="120500_300")
    (second / "conversation_transcript.jsonl").write_text("{}\n", encoding="utf-8")
    assert catchup_state.read_raw_input_fingerprint(DAY) != raw_changed


@pytest.mark.parametrize(
    "extension",
    (".jpg", ".jpeg", ".heic", ".heif", ".gif", ".webp", ".tiff"),
)
def test_raw_input_fingerprint_includes_image_extensions(journal, extension):
    segment = _segment(journal)
    baseline = catchup_state.read_raw_input_fingerprint(DAY)

    (segment / f"image{extension}").write_bytes(b"raw-image")

    assert catchup_state.read_raw_input_fingerprint(DAY) != baseline


def test_media_extensions_sourced_from_registry(journal):
    assert catchup_state.MEDIA_EXTENSIONS is media.MEDIA_EXTENSIONS

    segment = _segment(journal)
    for extension in media.MEDIA_EXTENSIONS:
        baseline = catchup_state.read_raw_input_fingerprint(DAY)

        (segment / f"media_{extension.removeprefix('.')}{extension}").write_bytes(
            b"raw-media"
        )

        assert catchup_state.read_raw_input_fingerprint(DAY) != baseline


def test_raw_input_fingerprint_pdf_only_day_differs_from_empty_day(journal):
    empty_digest = catchup_state.read_raw_input_fingerprint(DAY)

    segment = _segment(journal)
    (segment / "original.pdf").write_bytes(b"%PDF-1.4 one")

    assert catchup_state.PDF_EXTENSIONS is media.PDF_EXTENSIONS
    assert catchup_state.read_raw_input_fingerprint(DAY) != empty_digest


def test_raw_input_fingerprint_pdf_size_changes_digest(journal):
    segment = _segment(journal)
    pdf = segment / "original.pdf"
    pdf.write_bytes(b"%PDF-1.4 one")
    first_digest = catchup_state.read_raw_input_fingerprint(DAY)

    pdf.write_bytes(b"%PDF-1.4 one plus more bytes")

    assert catchup_state.read_raw_input_fingerprint(DAY) != first_digest


def test_raw_input_fingerprint_skips_file_deleted_mid_scan(journal, monkeypatch):
    first = _segment(journal)
    deleted = first / "audio.jsonl"
    survivor = first / "screen.jsonl"
    deleted.write_text("deleted\n", encoding="utf-8")
    survivor.write_text("survivor\n", encoding="utf-8")
    real_sha256 = catchup_state._file_sha256

    def flaky_sha256(path: Path) -> str:
        if path == deleted:
            raise OSError("deleted")
        return real_sha256(path)

    monkeypatch.setattr(catchup_state, "_file_sha256", flaky_sha256)

    digest = catchup_state.read_raw_input_fingerprint(DAY)

    expected_entries = [
        [
            survivor.relative_to(_day_dir(journal)).as_posix(),
            hashlib.sha256(survivor.read_bytes()).hexdigest(),
        ]
    ]
    payload = json.dumps(expected_entries, separators=(",", ":"), ensure_ascii=True)
    assert digest == hashlib.sha256(payload.encode()).hexdigest()
    assert catchup_state.read_raw_input_fingerprint(DAY) == digest


def test_record_attempt_sets_active_and_resets_backoff_on_fingerprint_change(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("{}\n", encoding="utf-8")
    _touch_marker(journal, DAY, "stream.updated", 100)
    marker = _touch_marker(journal, DAY, "daily.updated", 123)
    key = f"{DAY}:{catchup_state.KIND_DAILY_CATCHUP}"
    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_DAILY_CATCHUP,
                attempts=5,
                consecutive=3,
                next_retry_at=9999,
                entered_backoff_at=10,
                fingerprint="old",
            )
        },
    )

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-1", started_at=200)

    record = catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)
    assert record["attempts"] == 6
    assert record["consecutive_non_completion"] == 0
    assert record["entered_backoff_at"] is None
    assert record["notified_at"] is None
    assert record["next_retry_at"] == 0
    assert record["fingerprint"] != "old"
    assert record["active"] == {
        "ref": "ref-1",
        "started_at": 200,
        "marker_mtime_at_start": marker.stat().st_mtime,
    }


def test_record_outcome_uses_daily_marker_delta(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("{}\n", encoding="utf-8")

    catchup_state.record_attempt(CMD_DAILY, DAY, "complete-new", started_at=10)
    _touch_marker(journal, DAY, "stream.updated", 100)
    _touch_marker(journal, DAY, "daily.updated", 200)
    result = catchup_state.record_outcome(
        CMD_DAILY, DAY, "complete-new", exit_status="ok", ended_at=300
    )
    assert result.completed is True
    assert catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP) is None

    day2 = "20260102"
    cmd2 = ["journal", "think", "-v", "--day", day2]
    _segment(journal, day=day2).joinpath("audio.jsonl").write_text(
        "{}\n", encoding="utf-8"
    )
    _touch_marker(journal, day2, "stream.updated", 100)
    catchup_state.record_attempt(cmd2, day2, "no-marker", started_at=10)
    result = catchup_state.record_outcome(
        cmd2, day2, "no-marker", exit_status="ok", ended_at=300
    )
    record = catchup_state.read_day_record(day2, catchup_state.KIND_DAILY_CATCHUP)
    assert result.completed is False
    assert record["last_outcome"] == "ran-not-completed"
    assert record["consecutive_non_completion"] == 1
    assert record["next_retry_at"] == 900

    day3 = "20260103"
    cmd3 = ["journal", "think", "-v", "--day", day3]
    _segment(journal, day=day3).joinpath("audio.jsonl").write_text(
        "{}\n", encoding="utf-8"
    )
    _touch_marker(journal, day3, "stream.updated", 1000)
    _touch_marker(journal, day3, "daily.updated", 1000)
    catchup_state.record_attempt(cmd3, day3, "error-complete", started_at=10)
    _touch_marker(journal, day3, "stream.updated", 2000)
    _touch_marker(journal, day3, "daily.updated", 3000)
    result = catchup_state.record_outcome(
        cmd3, day3, "error-complete", exit_status="error", ended_at=4000
    )
    assert result.completed is True
    assert catchup_state.read_day_record(day3, catchup_state.KIND_DAILY_CATCHUP) is None


def test_daily_backoff_schedule_and_single_transition(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("{}\n", encoding="utf-8")
    _touch_marker(journal, DAY, "stream.updated", 100)
    entered_at = None

    for index, expected_delta in enumerate((600, 1200, 2400), start=1):
        catchup_state.record_attempt(CMD_DAILY, DAY, f"ref-{index}", started_at=index)
        ended_at = 1000 * index
        result = catchup_state.record_outcome(
            CMD_DAILY, DAY, f"ref-{index}", exit_status="ok", ended_at=ended_at
        )
        assert result.next_retry_at - ended_at == expected_delta
        assert result.entered_backoff is (index == 3)
        if index == 3:
            entered_at = catchup_state.read_day_record(
                DAY, catchup_state.KIND_DAILY_CATCHUP
            )["entered_backoff_at"]

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-4", started_at=4)
    result = catchup_state.record_outcome(
        CMD_DAILY, DAY, "ref-4", exit_status="ok", ended_at=4000
    )
    assert result.entered_backoff is False
    assert (
        catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)[
            "entered_backoff_at"
        ]
        == entered_at
    )


def test_daily_catchup_progressing_handoff_and_zero_cleared_failure(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("{}\n", encoding="utf-8")

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-progress", started_at=10)
    catchup_state.record_daily_catchup_progress(DAY, cleared=2, remaining=5)
    result = catchup_state.record_outcome(
        CMD_DAILY, DAY, "ref-progress", exit_status="timeout", ended_at=1000
    )

    record = catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)
    assert result.completed is False
    assert result.entered_backoff is False
    assert result.last_outcome == catchup_state.PROGRESSING_OUTCOME
    assert result.consecutive_non_completion == 0
    assert result.next_retry_at == 1600
    assert record["last_outcome"] == catchup_state.PROGRESSING_OUTCOME
    assert record["consecutive_non_completion"] == 0
    assert record["entered_backoff_at"] is None
    assert record["notified_at"] is None
    assert record["next_retry_at"] == 1600
    assert record["cleared"] == 2
    assert record["remaining"] == 5
    assert record["exit_reason"] == "timeout"
    assert record["daily_progress"] is None

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-failed", started_at=20)
    catchup_state.record_daily_catchup_progress(DAY, cleared=0, remaining=6)
    failed_result = catchup_state.record_outcome(
        CMD_DAILY, DAY, "ref-failed", exit_status="timeout", ended_at=2000
    )

    failed_record = catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)
    assert failed_result.last_outcome == "timeout"
    assert failed_result.consecutive_non_completion == 1
    assert failed_result.next_retry_at == 2600
    assert failed_record["last_outcome"] == "timeout"
    assert failed_record["consecutive_non_completion"] == 1
    assert failed_record["next_retry_at"] == 2600
    assert failed_record["daily_progress"] is None
    assert failed_record["cleared"] is None
    assert failed_record["remaining"] is None
    assert failed_record["exit_reason"] is None


def test_daily_progress_is_attempt_scoped_and_reconciled(journal, monkeypatch):
    _segment(journal).joinpath("audio.jsonl").write_text("{}\n", encoding="utf-8")
    _touch_marker(journal, DAY, "stream.updated", 100)

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-1", started_at=10)
    catchup_state.record_daily_catchup_progress(DAY, cleared=1, remaining=4)
    assert catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)[
        "daily_progress"
    ] == {"cleared": 1, "remaining": 4}

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-2", started_at=20)
    assert (
        catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)[
            "daily_progress"
        ]
        is None
    )

    catchup_state.record_daily_catchup_progress(DAY, cleared=3, remaining=2)
    monkeypatch.setattr(catchup_state.time, "time", lambda: 1000)
    catchup_state.reconcile_interrupted_attempts()

    record = catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)
    assert record["last_outcome"] == "interrupted"
    assert record["active"] is None
    assert record["daily_progress"] is None


def test_day_eligible_to_drain(journal, monkeypatch):
    _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8")
    fingerprint = catchup_state.read_raw_input_fingerprint(DAY)
    key = f"{DAY}:{catchup_state.KIND_DAILY_CATCHUP}"
    monkeypatch.setattr(catchup_state.time, "time", lambda: 100)

    assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_DAILY_CATCHUP)

    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_DAILY_CATCHUP,
                next_retry_at=1000,
                fingerprint=fingerprint,
                active={
                    "ref": "active",
                    "started_at": 1,
                    "marker_mtime_at_start": None,
                },
            )
        },
    )
    assert not catchup_state.day_eligible_to_drain(
        DAY, catchup_state.KIND_DAILY_CATCHUP
    )

    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_DAILY_CATCHUP,
                next_retry_at=1000,
                fingerprint=fingerprint,
            )
        },
    )
    assert not catchup_state.day_eligible_to_drain(
        DAY, catchup_state.KIND_DAILY_CATCHUP
    )

    _segment(journal).joinpath("audio.jsonl").write_text("two\n", encoding="utf-8")
    assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_DAILY_CATCHUP)

    monkeypatch.setattr(catchup_state.time, "time", lambda: 2000)
    assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_DAILY_CATCHUP)


def test_screen_describe_output_rearms_daily_catchup_progress(journal, monkeypatch):
    segment = _segment(journal)
    (segment / "center_DP-3_screen.webm").write_bytes(b"raw-video")
    monkeypatch.setattr(catchup_state.time, "time", lambda: 30)

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-1", started_at=10)
    started_record = catchup_state.read_day_record(
        DAY, catchup_state.KIND_DAILY_CATCHUP
    )
    started_fingerprint = started_record["fingerprint"]

    (segment / "center_DP-3_screen.jsonl").write_text(
        '{"timestamp":0,"text":"frame"}\n',
        encoding="utf-8",
    )
    progressed_fingerprint = catchup_state.read_raw_input_fingerprint(DAY)
    assert progressed_fingerprint != started_fingerprint

    result = catchup_state.record_outcome(
        CMD_DAILY, DAY, "ref-1", exit_status="timeout", ended_at=20
    )
    record = catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)
    assert result.completed is False
    assert record["consecutive_non_completion"] == 1
    assert record["next_retry_at"] > catchup_state.time.time()
    assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_DAILY_CATCHUP)

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-2", started_at=40)
    reset_record = catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP)
    assert reset_record["fingerprint"] == progressed_fingerprint
    assert reset_record["consecutive_non_completion"] == 0
    assert reset_record["entered_backoff_at"] is None
    assert reset_record["notified_at"] is None
    assert reset_record["next_retry_at"] == 0


def test_unchanged_daily_noncompletion_stays_gated_until_retry(journal, monkeypatch):
    _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8")

    catchup_state.record_attempt(CMD_DAILY, DAY, "ref-1", started_at=10)
    result = catchup_state.record_outcome(
        CMD_DAILY, DAY, "ref-1", exit_status="ok", ended_at=100
    )
    assert result.completed is False
    assert result.next_retry_at == 700

    monkeypatch.setattr(catchup_state.time, "time", lambda: 200)
    assert not catchup_state.day_eligible_to_drain(
        DAY, catchup_state.KIND_DAILY_CATCHUP
    )

    monkeypatch.setattr(catchup_state.time, "time", lambda: 701)
    assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_DAILY_CATCHUP)


def test_record_segment_repair_attempt_sets_active_and_resets_on_fingerprint_change(
    journal,
):
    raw = _segment(journal).joinpath("audio.jsonl")
    raw.write_text("one\n", encoding="utf-8")
    key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}"

    catchup_state.record_segment_repair_attempt(DAY, started_at=10)

    record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR)
    fingerprint = catchup_state.read_raw_input_fingerprint(DAY)
    assert record["attempts"] == 1
    assert record["fingerprint"] == fingerprint
    assert record["active"] == {"ref": "segment-repair", "started_at": 10}

    record.update(
        consecutive_non_completion=2,
        entered_backoff_at=100,
        notified_at=100,
        next_retry_at=999,
        reason_code=catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON,
        timeout_seconds=600,
        bounded=True,
        cleared=2,
        remaining=1,
        exit_reason=catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON,
        active=None,
    )
    _write_state(journal, {key: record})
    raw.write_text("two\n", encoding="utf-8")

    catchup_state.record_segment_repair_attempt(DAY, started_at=20)

    reset_record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR)
    assert reset_record["attempts"] == 2
    assert reset_record["fingerprint"] != fingerprint
    assert reset_record["consecutive_non_completion"] == 0
    assert reset_record["entered_backoff_at"] is None
    assert reset_record["notified_at"] is None
    assert reset_record["next_retry_at"] == 0
    assert reset_record["reason_code"] is None
    assert reset_record["timeout_seconds"] is None
    assert reset_record["bounded"] is None
    assert reset_record["cleared"] is None
    assert reset_record["remaining"] is None
    assert reset_record["exit_reason"] is None
    assert reset_record["active"] == {"ref": "segment-repair", "started_at": 20}


def test_segment_repair_attempt_clears_stale_progressing_after_fingerprint_change(
    journal,
):
    raw = _segment(journal).joinpath("audio.jsonl")
    raw.write_text("one\n", encoding="utf-8")
    first_fingerprint = catchup_state.read_raw_input_fingerprint(DAY)

    catchup_state.record_segment_repair_attempt(DAY, started_at=10)
    catchup_state.record_segment_repair_outcome(
        DAY,
        success=False,
        timed_out=True,
        timeout_seconds=300,
        ended_at=1000,
        cleared=2,
        remaining=10,
    )

    key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}"
    progressing_record = _read_entries(journal)[key]
    assert progressing_record["fingerprint"] == first_fingerprint
    assert progressing_record["last_outcome"] == catchup_state.PROGRESSING_OUTCOME

    raw.write_text("two\n", encoding="utf-8")
    assert catchup_state.read_raw_input_fingerprint(DAY) != first_fingerprint

    catchup_state.record_segment_repair_attempt(DAY, started_at=20)

    assert catchup_state.read_segment_repair_summary(DAY) is None
    reset_record = _read_entries(journal)[key]
    assert reset_record["fingerprint"] != first_fingerprint
    assert reset_record["last_outcome"] == ""
    assert reset_record["cleared"] is None
    assert reset_record["remaining"] is None
    assert reset_record["exit_reason"] is None
    assert reset_record["active"] == {"ref": "segment-repair", "started_at": 20}


def test_record_segment_repair_outcome_records_failure_metadata_and_backoff(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8")

    catchup_state.record_segment_repair_attempt(DAY, started_at=10)
    catchup_state.record_segment_repair_outcome(
        DAY,
        success=False,
        timed_out=True,
        timeout_seconds=300,
        ended_at=1000,
        cleared=None,
        remaining=None,
    )

    timeout_record = catchup_state.read_day_record(
        DAY, catchup_state.KIND_SEGMENT_REPAIR
    )
    assert timeout_record["active"] is None
    assert timeout_record["last_outcome"] == "timeout"
    assert timeout_record["reason_code"] == catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON
    assert timeout_record["timeout_seconds"] == 300
    assert timeout_record["bounded"] is True
    assert timeout_record["consecutive_non_completion"] == 1
    assert timeout_record["next_retry_at"] == 1600
    assert timeout_record["entered_backoff_at"] is None
    assert timeout_record["cleared"] is None
    assert timeout_record["remaining"] is None
    assert timeout_record["exit_reason"] is None

    catchup_state.record_segment_repair_attempt(DAY, started_at=20)
    catchup_state.record_segment_repair_outcome(
        DAY,
        success=False,
        timed_out=False,
        timeout_seconds=300,
        ended_at=2000,
        cleared=None,
        remaining=None,
    )
    error_record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR)
    assert error_record["last_outcome"] == "error"
    assert error_record["reason_code"] == catchup_state.SEGMENT_REPAIR_FAILED_REASON
    assert error_record["timeout_seconds"] is None
    assert error_record["bounded"] is False
    assert error_record["consecutive_non_completion"] == 2

    catchup_state.record_segment_repair_attempt(DAY, started_at=30)
    catchup_state.record_segment_repair_outcome(
        DAY,
        success=False,
        timed_out=False,
        timeout_seconds=300,
        ended_at=3000,
        cleared=None,
        remaining=None,
    )
    stuck_record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR)
    assert stuck_record["consecutive_non_completion"] == catchup_state.STUCK_THRESHOLD
    assert stuck_record["entered_backoff_at"] == 3000
    assert stuck_record["notified_at"] == 3000


def test_record_segment_repair_outcome_progressing_resets_backoff(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8")

    catchup_state.record_segment_repair_attempt(DAY, started_at=10)
    catchup_state.record_segment_repair_outcome(
        DAY,
        success=False,
        timed_out=True,
        timeout_seconds=300,
        ended_at=1000,
        cleared=2,
        remaining=3,
    )

    record = _read_entries(journal)[f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}"]
    assert record["active"] is None
    assert record["last_outcome"] == catchup_state.PROGRESSING_OUTCOME
    assert record["consecutive_non_completion"] == 0
    assert record["entered_backoff_at"] is None
    assert record["notified_at"] is None
    assert record["next_retry_at"] == 1600
    assert record["reason_code"] == catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON
    assert record["cleared"] == 2
    assert record["remaining"] == 3
    assert record["exit_reason"] == catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON

    summary = catchup_state.read_segment_repair_summary(DAY)
    assert summary is not None
    assert summary["status"] == catchup_state.PROGRESSING_OUTCOME
    assert summary["cleared"] == 2
    assert summary["remaining"] == 3
    assert summary["consecutive_non_completion"] == 0


def test_segment_repair_progressing_and_failed_persist_different_records(journal):
    progressing_day = "20260101"
    failed_day = "20260102"
    _segment(journal, day=progressing_day).joinpath("audio.jsonl").write_text(
        "one\n", encoding="utf-8"
    )
    _segment(journal, day=failed_day).joinpath("audio.jsonl").write_text(
        "one\n", encoding="utf-8"
    )

    catchup_state.record_segment_repair_attempt(progressing_day, started_at=10)
    catchup_state.record_segment_repair_outcome(
        progressing_day,
        success=False,
        timed_out=False,
        timeout_seconds=None,
        ended_at=1000,
        cleared=1,
        remaining=4,
    )
    catchup_state.record_segment_repair_attempt(failed_day, started_at=20)
    catchup_state.record_segment_repair_outcome(
        failed_day,
        success=False,
        timed_out=False,
        timeout_seconds=None,
        ended_at=2000,
        cleared=0,
        remaining=5,
    )

    entries = _read_entries(journal)
    progressing = entries[f"{progressing_day}:{catchup_state.KIND_SEGMENT_REPAIR}"]
    failed = entries[f"{failed_day}:{catchup_state.KIND_SEGMENT_REPAIR}"]
    assert progressing["last_outcome"] == catchup_state.PROGRESSING_OUTCOME
    assert progressing["consecutive_non_completion"] == 0
    assert progressing["cleared"] == 1
    assert progressing["remaining"] == 4
    assert failed["last_outcome"] == "error"
    assert failed["consecutive_non_completion"] == 1
    assert failed["cleared"] is None
    assert failed["remaining"] is None
    assert failed["exit_reason"] is None


def test_segment_repair_progressing_survives_prune(journal):
    old_day = "20260101"
    new_day = "20260215"
    _segment(journal, day=old_day).joinpath("audio.jsonl").write_text(
        "one\n", encoding="utf-8"
    )
    _segment(journal, day=new_day).joinpath("audio.jsonl").write_text(
        "one\n", encoding="utf-8"
    )

    catchup_state.record_segment_repair_attempt(old_day, started_at=10)
    catchup_state.record_segment_repair_outcome(
        old_day,
        success=False,
        timed_out=True,
        timeout_seconds=300,
        ended_at=1000,
        cleared=1,
        remaining=0,
    )
    catchup_state.record_segment_repair_attempt(new_day, started_at=20)

    entries = _read_entries(journal)
    old_key = f"{old_day}:{catchup_state.KIND_SEGMENT_REPAIR}"
    assert old_key in entries
    assert entries[old_key]["last_outcome"] == catchup_state.PROGRESSING_OUTCOME


def test_record_segment_repair_outcome_success_pops_record(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8")
    catchup_state.record_segment_repair_attempt(DAY, started_at=10)
    catchup_state.record_segment_repair_outcome(
        DAY,
        success=False,
        timed_out=False,
        timeout_seconds=None,
        ended_at=1000,
        cleared=None,
        remaining=None,
    )
    assert catchup_state.read_segment_repair_summary(DAY) is not None

    catchup_state.record_segment_repair_outcome(
        DAY,
        success=True,
        timed_out=False,
        timeout_seconds=None,
        ended_at=2000,
        cleared=None,
        remaining=None,
    )

    assert catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR) is None
    assert catchup_state.read_segment_repair_summary(DAY) is None


def test_read_segment_repair_attempted_tracks_attempt_record_lifetime(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8")

    assert not catchup_state.read_segment_repair_attempted(DAY)

    catchup_state.record_segment_repair_attempt(DAY, started_at=10)
    assert catchup_state.read_segment_repair_attempted(DAY)

    catchup_state.record_segment_repair_outcome(
        DAY,
        success=True,
        timed_out=False,
        timeout_seconds=None,
        ended_at=20,
        cleared=None,
        remaining=None,
    )
    assert not catchup_state.read_segment_repair_attempted(DAY)


def test_read_segment_repair_summary_states(journal):
    raw = _segment(journal).joinpath("audio.jsonl")
    raw.write_text("one\n", encoding="utf-8")
    fingerprint = catchup_state.read_raw_input_fingerprint(DAY)
    key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}"

    assert catchup_state.read_segment_repair_summary(DAY) is None

    _write_state(
        journal,
        {key: _record(DAY, catchup_state.KIND_SEGMENT_REPAIR, fingerprint=fingerprint)},
    )
    assert catchup_state.read_segment_repair_summary(DAY) is None

    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_SEGMENT_REPAIR,
                consecutive=1,
                fingerprint="stale",
            )
        },
    )
    assert catchup_state.read_segment_repair_summary(DAY) is None

    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_SEGMENT_REPAIR,
                attempts=2,
                consecutive=1,
                last_outcome="error",
                next_retry_at=999,
                fingerprint=fingerprint,
                reason_code=catchup_state.SEGMENT_REPAIR_FAILED_REASON,
                bounded=False,
            )
        },
    )
    degraded = catchup_state.read_segment_repair_summary(DAY)
    assert degraded == {
        "status": "degraded",
        "attempts": 2,
        "consecutive_non_completion": 1,
        "last_outcome": "error",
        "next_retry_at": 999,
        "repair_reason_code": catchup_state.SEGMENT_REPAIR_FAILED_REASON,
        "timeout_seconds": None,
        "bounded": False,
    }

    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_SEGMENT_REPAIR,
                attempts=3,
                consecutive=3,
                last_outcome="timeout",
                next_retry_at=1200,
                entered_backoff_at=600,
                fingerprint=fingerprint,
                reason_code=catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON,
                timeout_seconds=300,
                bounded=True,
            )
        },
    )
    stuck = catchup_state.read_segment_repair_summary(DAY)
    assert stuck["status"] == "stuck"
    assert stuck["repair_reason_code"] == catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON
    assert stuck["timeout_seconds"] == 300
    assert stuck["bounded"] is True

    _state_file(journal).write_text("{not json", encoding="utf-8")
    assert catchup_state.read_segment_repair_summary(DAY) == {"status": "unknown"}


def test_day_eligible_to_drain_segment_repair(journal, monkeypatch):
    raw = _segment(journal).joinpath("audio.jsonl")
    raw.write_text("one\n", encoding="utf-8")
    fingerprint = catchup_state.read_raw_input_fingerprint(DAY)
    key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}"
    monkeypatch.setattr(catchup_state.time, "time", lambda: 100)

    assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_SEGMENT_REPAIR)

    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_SEGMENT_REPAIR,
                next_retry_at=1000,
                fingerprint=fingerprint,
                active={"ref": "segment-repair", "started_at": 1},
            )
        },
    )
    assert not catchup_state.day_eligible_to_drain(
        DAY, catchup_state.KIND_SEGMENT_REPAIR
    )

    _write_state(
        journal,
        {
            key: _record(
                DAY,
                catchup_state.KIND_SEGMENT_REPAIR,
                next_retry_at=1000,
                fingerprint=fingerprint,
            )
        },
    )
    assert not catchup_state.day_eligible_to_drain(
        DAY, catchup_state.KIND_SEGMENT_REPAIR
    )

    raw.write_text("two\n", encoding="utf-8")
    assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_SEGMENT_REPAIR)


def test_reconcile_interrupted_attempts(journal, monkeypatch):
    incomplete = "20260101"
    complete = "20260102"
    _segment(journal, day=incomplete).joinpath("audio.jsonl").write_text(
        "{}\n", encoding="utf-8"
    )
    _touch_marker(journal, incomplete, "stream.updated", 100)
    _segment(journal, day=complete).joinpath("audio.jsonl").write_text(
        "{}\n", encoding="utf-8"
    )
    _touch_marker(journal, complete, "stream.updated", 100)
    _touch_marker(journal, complete, "daily.updated", 200)
    incomplete_key = f"{incomplete}:{catchup_state.KIND_DAILY_CATCHUP}"
    complete_key = f"{complete}:{catchup_state.KIND_DAILY_CATCHUP}"
    _write_state(
        journal,
        {
            incomplete_key: _record(
                incomplete,
                catchup_state.KIND_DAILY_CATCHUP,
                attempts=3,
                consecutive=2,
                active={
                    "ref": "active-1",
                    "started_at": 1,
                    "marker_mtime_at_start": None,
                },
            ),
            complete_key: _record(
                complete,
                catchup_state.KIND_DAILY_CATCHUP,
                attempts=1,
                active={
                    "ref": "active-2",
                    "started_at": 1,
                    "marker_mtime_at_start": None,
                },
            ),
        },
    )
    monkeypatch.setattr(catchup_state.time, "time", lambda: 1000)

    transitions = catchup_state.reconcile_interrupted_attempts()

    entries = _read_entries(journal)
    assert len(transitions) == 1
    assert transitions[0].day == incomplete
    assert entries[incomplete_key]["last_outcome"] == "interrupted"
    assert entries[incomplete_key]["consecutive_non_completion"] == 3
    assert entries[incomplete_key]["active"] is None
    assert complete_key not in entries


def test_reconcile_interrupted_segment_repair_on_complete_day_is_not_popped(
    journal, monkeypatch
):
    incomplete_daily = "20260101"
    complete_day = "20260102"
    _segment(journal, day=incomplete_daily).joinpath("audio.jsonl").write_text(
        "{}\n", encoding="utf-8"
    )
    _touch_marker(journal, incomplete_daily, "stream.updated", 100)
    _segment(journal, day=complete_day).joinpath("audio.jsonl").write_text(
        "{}\n", encoding="utf-8"
    )
    _touch_marker(journal, complete_day, "stream.updated", 100)
    _touch_marker(journal, complete_day, "daily.updated", 200)
    daily_incomplete_key = f"{incomplete_daily}:{catchup_state.KIND_DAILY_CATCHUP}"
    daily_complete_key = f"{complete_day}:{catchup_state.KIND_DAILY_CATCHUP}"
    repair_key = f"{complete_day}:{catchup_state.KIND_SEGMENT_REPAIR}"
    _write_state(
        journal,
        {
            daily_incomplete_key: _record(
                incomplete_daily,
                catchup_state.KIND_DAILY_CATCHUP,
                attempts=3,
                consecutive=2,
                active={
                    "ref": "daily-incomplete",
                    "started_at": 1,
                    "marker_mtime_at_start": None,
                },
            ),
            daily_complete_key: _record(
                complete_day,
                catchup_state.KIND_DAILY_CATCHUP,
                attempts=1,
                active={
                    "ref": "daily-complete",
                    "started_at": 1,
                    "marker_mtime_at_start": None,
                },
            ),
            repair_key: _record(
                complete_day,
                catchup_state.KIND_SEGMENT_REPAIR,
                attempts=2,
                consecutive=2,
                active={"ref": "segment-repair", "started_at": 1},
            ),
        },
    )
    monkeypatch.setattr(catchup_state.time, "time", lambda: 1000)

    transitions = catchup_state.reconcile_interrupted_attempts()

    entries = _read_entries(journal)
    assert daily_complete_key not in entries
    assert entries[repair_key]["last_outcome"] == "interrupted"
    assert entries[repair_key]["consecutive_non_completion"] == 3
    assert entries[repair_key]["entered_backoff_at"] == 1000
    assert entries[repair_key]["active"] is None
    assert len(transitions) == 1
    assert transitions[0].day == incomplete_daily
    assert transitions[0].command_kind == catchup_state.KIND_DAILY_CATCHUP


def test_daily_completion_and_clear_day_backoff_leave_segment_repair_record(journal):
    raw = _segment(journal).joinpath("audio.jsonl")
    raw.write_text("{}\n", encoding="utf-8")
    _touch_marker(journal, DAY, "stream.updated", 100)
    _touch_marker(journal, DAY, "daily.updated", 100)
    fingerprint = catchup_state.read_raw_input_fingerprint(DAY)
    daily_key = f"{DAY}:{catchup_state.KIND_DAILY_CATCHUP}"
    scratch_key = f"{DAY}:{catchup_state.KIND_DAILY_FROM_SCRATCH}"
    repair_key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}"
    repair_record = _record(
        DAY,
        catchup_state.KIND_SEGMENT_REPAIR,
        consecutive=1,
        last_outcome="error",
        next_retry_at=999,
        fingerprint=fingerprint,
        reason_code=catchup_state.SEGMENT_REPAIR_FAILED_REASON,
    )
    _write_state(
        journal,
        {
            daily_key: _record(
                DAY,
                catchup_state.KIND_DAILY_CATCHUP,
                active={
                    "ref": "daily",
                    "started_at": 1,
                    "marker_mtime_at_start": 100,
                },
            ),
            repair_key: repair_record,
        },
    )
    _touch_marker(journal, DAY, "daily.updated", 200)

    result = catchup_state.record_outcome(
        CMD_DAILY, DAY, "daily", exit_status="ok", ended_at=300
    )

    entries = _read_entries(journal)
    assert result.completed is True
    assert daily_key not in entries
    assert entries[repair_key]["command_kind"] == catchup_state.KIND_SEGMENT_REPAIR

    _write_state(
        journal,
        {
            daily_key: _record(DAY, catchup_state.KIND_DAILY_CATCHUP),
            scratch_key: _record(DAY, catchup_state.KIND_DAILY_FROM_SCRATCH),
            repair_key: entries[repair_key],
        },
    )

    catchup_state.clear_day_backoff(DAY)

    entries = _read_entries(journal)
    assert daily_key not in entries
    assert scratch_key not in entries
    assert repair_key in entries


def test_prune_removes_old_cleared_entries_but_keeps_old_stuck(journal):
    old_day = "20260101"
    recent_day = "20260214"
    newest_day = "20260215"
    for day in (old_day, recent_day, newest_day):
        _day_dir(journal, day)
    old_completed = f"{old_day}:{catchup_state.KIND_SEGMENT}"
    old_stuck = f"{old_day}:{catchup_state.KIND_DAILY_CATCHUP}"
    recent = f"{recent_day}:{catchup_state.KIND_SEGMENT}"
    _write_state(
        journal,
        {
            old_completed: _record(
                old_day, catchup_state.KIND_SEGMENT, last_outcome="completed"
            ),
            old_stuck: _record(
                old_day,
                catchup_state.KIND_DAILY_CATCHUP,
                consecutive=3,
                next_retry_at=9999,
                entered_backoff_at=1,
                last_outcome="timeout",
            ),
            recent: _record(
                recent_day, catchup_state.KIND_SEGMENT, last_outcome="completed"
            ),
        },
    )

    catchup_state.clear_day_backoff("20990101")

    entries = _read_entries(journal)
    assert old_completed not in entries
    assert old_stuck in entries
    assert recent in entries


def test_from_scratch_completion_clears_daily_catchup_record(journal):
    _segment(journal).joinpath("audio.jsonl").write_text("{}\n", encoding="utf-8")
    catchup_key = f"{DAY}:{catchup_state.KIND_DAILY_CATCHUP}"
    _write_state(
        journal,
        {
            catchup_key: _record(
                DAY,
                catchup_state.KIND_DAILY_CATCHUP,
                consecutive=3,
                next_retry_at=9999,
                entered_backoff_at=1,
                last_outcome="timeout",
            )
        },
    )

    catchup_state.record_attempt(CMD_FROM_SCRATCH, DAY, "scratch", started_at=10)
    _touch_marker(journal, DAY, "stream.updated", 100)
    _touch_marker(journal, DAY, "daily.updated", 200)
    result = catchup_state.record_outcome(
        CMD_FROM_SCRATCH, DAY, "scratch", exit_status="error", ended_at=300
    )

    assert result.completed is True
    assert catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_CATCHUP) is None
    assert (
        catchup_state.read_day_record(DAY, catchup_state.KIND_DAILY_FROM_SCRATCH)
        is None
    )
