from __future__ import annotations

from pathlib import Path

import pytest

from crowley_diary.backends import BackendResult, FixtureBackend, TranscriptionRequest
from crowley_diary.cache import RunCache
from crowley_diary.builder import build_candidate_page
from crowley_diary.config import EvaluationProfile
from crowley_diary.corpus import (
    AdaptiveShardLimiter,
    Shard,
    _process_shard,
    finalize_corpus,
    page_lock,
    parse_worker_count,
    partition_physical_shards,
)
from crowley_diary.pipeline import transcribe_page
from crowley_diary.reconciliation import reconcile_adjacent_pages
from crowley_diary.source import SourcePackage
from crowley_diary.util import load_json, project_root, write_json, write_text


def profile(*, infrastructure_retries: int = 0) -> EvaluationProfile:
    return EvaluationProfile(
        backend="fixture",
        model="fixture",
        reasoning_effort="none",
        timeout_seconds=10,
        max_repair_attempts=1,
        previous_context_characters=100,
        next_context_characters=100,
        max_batch_pages=20,
        max_infrastructure_retries=infrastructure_retries,
        infrastructure_backoff_seconds=0,
    )


def test_worker_count_and_consecutive_shard_partitioning() -> None:
    assert parse_worker_count(1) == 1
    assert parse_worker_count(4) == 4
    with pytest.raises(ValueError, match="positive"):
        parse_worker_count(0)

    completed = {6, 26, *range(41, 61)}
    shards = partition_physical_shards(
        total_pages=295, completed_pages=completed, shard_size=20
    )
    assert all(shard.end_page - shard.start_page + 1 <= 20 for shard in shards)
    expected_ranges = [
        *[(start, start + 19) for start in range(1, 281, 20)],
        (281, 295),
    ]
    expected_ranges.remove((41, 60))
    assert [(shard.start_page, shard.end_page) for shard in shards] == expected_ranges
    assert 6 in shards[0].completed_anchors
    assert 6 not in shards[0].pending_pages
    assert [page for shard in shards for page in shard.pending_pages] == sorted(
        set(range(1, 296)) - completed
    )


def test_page_lock_prevents_duplicate_work(tmp_path: Path) -> None:
    path = tmp_path / "page-039.lock"
    with page_lock(path) as first:
        assert first is True
        with page_lock(path) as second:
            assert second is False
    with page_lock(path) as third:
        assert third is True


def test_workers_one_shard_execution_is_sequential_resumable_and_atomic(
    monkeypatch, source_root: Path, tmp_path: Path
) -> None:
    class CountingBackend:
        name = "fixture"

        def __init__(self) -> None:
            self.delegate = FixtureBackend(project_root() / "tests" / "fixtures" / "model")
            self.calls = 0

        def transcribe(self, request: TranscriptionRequest) -> BackendResult:
            self.calls += 1
            return self.delegate.transcribe(request)

    monkeypatch.setattr(
        "crowley_diary.pipeline.RunCache", lambda _path: RunCache(tmp_path / "cache")
    )
    source = SourcePackage(source_root)
    run_root = tmp_path / "run"
    shard = Shard(2, 39, 39, (39,), ())
    backend = CountingBackend()
    limiter = AdaptiveShardLimiter(1)
    first = _process_shard(
        shard=shard,
        source=source,
        run_id="corpus-test",
        run_root=run_root,
        profile=profile(),
        backend=backend,
        baseline={},
        resume=False,
        limiter=limiter,
    )
    assert first["status"] == "complete"
    assert backend.calls == 1
    assert limiter.max_observed == 1
    completion = run_root / "shards" / "shard-039-039.complete.json"
    assert completion.is_file()
    assert not list(completion.parent.glob(".*.tmp"))

    second = _process_shard(
        shard=shard,
        source=source,
        run_id="corpus-test",
        run_root=run_root,
        profile=profile(),
        backend=backend,
        baseline={},
        resume=True,
        limiter=AdaptiveShardLimiter(1),
    )
    assert second["results"][0]["resumed"] is True
    assert backend.calls == 1


def test_shared_content_cache_avoids_duplicate_backend_work(
    monkeypatch, source_root: Path, tmp_path: Path
) -> None:
    class CountingBackend:
        name = "fixture"

        def __init__(self) -> None:
            self.delegate = FixtureBackend(project_root() / "tests" / "fixtures" / "model")
            self.calls = 0

        def transcribe(self, request: TranscriptionRequest) -> BackendResult:
            self.calls += 1
            return self.delegate.transcribe(request)

    shared_cache = tmp_path / "cache"
    monkeypatch.setattr(
        "crowley_diary.pipeline.RunCache", lambda _path: RunCache(shared_cache)
    )
    backend = CountingBackend()
    source = SourcePackage(source_root)
    first = transcribe_page(
        source=source,
        page=39,
        run_id="cache-run-1",
        run_root=tmp_path / "run-1",
        profile=profile(),
        backend=backend,
    )
    second = transcribe_page(
        source=source,
        page=39,
        run_id="cache-run-2",
        run_root=tmp_path / "run-2",
        profile=profile(),
        backend=backend,
    )
    assert first.cache_hit is False
    assert second.cache_hit is True
    assert backend.calls == 1


def test_failed_page_is_quarantined_without_corrupting_later_shard(
    monkeypatch, source_root: Path, tmp_path: Path
) -> None:
    monkeypatch.setattr(
        "crowley_diary.pipeline.RunCache", lambda _path: RunCache(tmp_path / "cache")
    )
    source = SourcePackage(source_root)
    run_root = tmp_path / "run"
    missing = FixtureBackend(tmp_path / "missing-fixtures")
    failed = _process_shard(
        shard=Shard(1, 40, 40, (40,), ()),
        source=source,
        run_id="worker-failure-test",
        run_root=run_root,
        profile=profile(),
        backend=missing,
        baseline={},
        resume=False,
        limiter=AdaptiveShardLimiter(1),
    )
    assert failed["status"] == "complete_with_quarantine"
    assert (run_root / "quarantine" / "page-040.json").is_file()

    succeeded = _process_shard(
        shard=Shard(2, 39, 39, (39,), ()),
        source=source,
        run_id="worker-failure-test",
        run_root=run_root,
        profile=profile(),
        backend=FixtureBackend(project_root() / "tests" / "fixtures" / "model"),
        baseline={},
        resume=False,
        limiter=AdaptiveShardLimiter(1),
    )
    assert succeeded["status"] == "complete"
    assert load_json(run_root / "candidates" / "page-039.json")["page_id"] == "page-039"


def test_cross_shard_boundary_reconciles_like_an_ordinary_boundary() -> None:
    shards = partition_physical_shards(
        total_pages=40, completed_pages=set(), shard_size=20
    )
    assert [(shard.start_page, shard.end_page) for shard in shards] == [
        (1, 20),
        (21, 40),
    ]

    def page(number: int, from_previous: str, to_next: str) -> dict:
        return {
            "page_id": f"page-{number:03d}",
            "source": {"pdf_page": number},
            "blocks": [
                {
                    "id": f"p{number:03d}-b001",
                    "type": "diary_entry",
                    "continuation": {
                        "from_previous": from_previous,
                        "to_next": to_next,
                        "previous": None,
                        "next": None,
                    },
                }
            ],
            "provenance": {"steps": []},
        }

    left = page(20, "no", "yes")
    right = page(21, "yes", "no")
    result = reconcile_adjacent_pages(
        left, right, created_at="2026-08-08T00:00:00Z"
    )
    assert result.linked == [{"left": "p020-b001", "right": "p021-b001"}]
    assert left["blocks"][0]["continuation"]["next"] == {
        "page_id": "page-021",
        "block_id": "p021-b001",
    }
    assert right["blocks"][0]["continuation"]["previous"] == {
        "page_id": "page-020",
        "block_id": "p020-b001",
    }


def test_corpus_aggregation_is_stably_page_ordered(
    monkeypatch, source_root: Path, tmp_path: Path
) -> None:
    class ThreePageSource(SourcePackage):
        @property
        def page_count(self) -> int:
            return 3

    source = ThreePageSource(source_root)
    monkeypatch.setattr("crowley_diary.corpus.owner_approved_pages", lambda _source: {})
    run_root = tmp_path / "run"
    prompt = run_root / "input-transcription-specification.md"
    schema = run_root / "input-schema.json"
    write_text(prompt, "fixture prompt")
    write_json(schema, load_json(project_root() / "schema" / "model-output.schema.json"))
    blank = load_json(project_root() / "tests" / "fixtures" / "model" / "page-039.json")
    for number in (3, 1, 2):
        page = build_candidate_page(
            draft=blank,
            source=source,
            page=number,
            run_id="aggregate-test",
            backend="fixture",
            model="fixture",
            reasoning_effort="none",
            prompt_path=prompt,
            schema_path=schema,
            context_pages=[],
            created_at="2026-08-08T00:00:00Z",
        )
        write_json(run_root / "candidates" / f"page-{number:03d}.json", page)
        write_json(
            run_root / "pages" / f"page-{number:03d}" / "status.json",
            {
                "page": number,
                "status": "valid_candidate",
                "cache_hit": False,
                "repair_used": False,
                "attempt_count": 1,
                "elapsed_seconds": 1.0,
            },
        )
    integrity = {"sha256": "a" * 64}
    write_json(run_root / "source-integrity-before.json", integrity)
    write_json(run_root / "source-integrity-after-model.json", integrity)
    write_json(
        run_root / "run-manifest.json",
        {
            "run_id": "aggregate-test",
            "status": "model_execution_complete",
            "completed_at": "2026-08-08T00:00:00Z",
            "backend": "fixture",
            "model": "fixture",
            "reasoning_effort": "none",
            "configured_workers": 1,
            "maximum_observed_active_workers": 1,
            "final_effective_workers": 1,
            "concurrency_reductions": [],
            "shards": [
                {
                    "start_page": 1,
                    "end_page": 3,
                    "pending_pages": [1, 2, 3],
                    "completed_anchors": [],
                }
            ],
            "codex_version": None,
            "wall_clock_elapsed_seconds": 3.0,
        },
    )
    report = finalize_corpus(run_root=run_root, source=source, qc_seed="test-seed")
    assert [record["page"] for record in report["pages"]] == [1, 2, 3]
    assert report["coverage"]["valid_page_editions"] == 3
    assert (run_root / "review" / "index.html").is_file()
    assert len(list((run_root / "review" / "pages").glob("page-*.html"))) == 3
