"""Bounded reads, artifact retention, and recovery cadence."""

from __future__ import annotations

from testing.tests.test_hermes_cli_support import (
    Path,
    SimpleNamespace,
    _completed_result,
    json,
    lanes,
    nullcontext,
    os,
    pytest,
    sys,
)


def test_invalid_utf8_journal_quarantines_and_does_not_stop_later_replay(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    bad = lanes._terminal_path(lanes.state_path("cassandra", "a_bad_utf8"), 31)
    bad.parent.mkdir(parents=True)
    bad.write_bytes(b"\xff\xfe\x80not-json")
    good, _record = lanes._write_terminal_record(
        lanes.state_path("cassandra", "z_good"),
        board="cassandra",
        task_id="z_good",
        run_id=32,
        structured=_completed_result("valid after invalid UTF-8"),
        summary="valid after invalid UTF-8",
        metadata={},
    )
    tasks = {
        "a_bad_utf8": SimpleNamespace(
            id="a_bad_utf8",
            status="running",
            current_run_id=31,
            result=None,
            assignee="cli-auto",
        ),
        "z_good": SimpleNamespace(
            id="z_good",
            status="running",
            current_run_id=32,
            result=None,
            assignee="cli-auto",
        ),
    }

    class Connection:
        def close(self):
            return None

    def reclaim_task(_conn, task_id, **_kwargs):
        tasks[task_id].status = "ready"
        tasks[task_id].current_run_id = None
        return True

    def complete_task(_conn, task_id, **kwargs):
        tasks[task_id].status = "done"
        tasks[task_id].result = kwargs["result"]
        tasks[task_id].current_run_id = None
        return True

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, task_id: tasks[task_id],
        reclaim_task=reclaim_task,
        complete_task=complete_task,
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))

    assert lanes.recover_pending_finalizations() == 1
    assert not bad.exists()
    assert not good.exists()
    assert tasks["a_bad_utf8"].status == "ready"
    assert tasks["z_good"].status == "done"
    diagnostic_path = next((bad.parent / "quarantine").glob("*.quarantine"))
    diagnostic = json.loads(diagnostic_path.read_text(encoding="utf-8"))
    assert diagnostic["size"] == len(b"\xff\xfe\x80not-json")
    assert diagnostic_path.stat().st_size < 4096


@pytest.mark.parametrize(
    "size",
    [lanes.MAX_TERMINAL_RECORD_BYTES + 1, 16 * 1024 * 1024],
)
def test_oversized_sparse_journal_has_bounded_read_and_quarantine(
    tmp_path: Path,
    monkeypatch,
    size: int,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    huge = lanes._terminal_path(lanes.state_path("cassandra", "a_huge"), 41)
    huge.parent.mkdir(parents=True)
    with huge.open("wb") as stream:
        stream.seek(size - 1)
        stream.write(b"\0")
    good, _record = lanes._write_terminal_record(
        lanes.state_path("cassandra", "z_after_huge"),
        board="cassandra",
        task_id="z_after_huge",
        run_id=42,
        structured=_completed_result("valid after oversized journal"),
        summary="valid after oversized journal",
        metadata={},
    )
    tasks = {
        "a_huge": SimpleNamespace(
            id="a_huge",
            status="running",
            current_run_id=41,
            result=None,
            assignee="cli-auto",
        ),
        "z_after_huge": SimpleNamespace(
            id="z_after_huge",
            status="running",
            current_run_id=42,
            result=None,
            assignee="cli-auto",
        ),
    }
    reads = []
    real_read_bounded = lanes._read_bounded

    def observe_read_limit(descriptor, limit):
        reads.append((os.fstat(descriptor).st_size, limit))
        return real_read_bounded(descriptor, limit)

    monkeypatch.setattr(lanes, "_read_bounded", observe_read_limit)

    class Connection:
        def close(self):
            return None

    def reclaim_task(_conn, task_id, **_kwargs):
        tasks[task_id].status = "ready"
        tasks[task_id].current_run_id = None
        return True

    def complete_task(_conn, task_id, **kwargs):
        tasks[task_id].status = "done"
        tasks[task_id].result = kwargs["result"]
        tasks[task_id].current_run_id = None
        return True

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, task_id: tasks[task_id],
        reclaim_task=reclaim_task,
        complete_task=complete_task,
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))

    assert lanes.recover_pending_finalizations() == 1
    assert not huge.exists()
    assert not good.exists()
    assert tasks["a_huge"].status == "ready"
    assert tasks["z_after_huge"].status == "done"
    diagnostics = list((huge.parent / "quarantine").glob("*.quarantine"))
    assert len(diagnostics) == 1
    assert diagnostics[0].stat().st_size < 4096
    diagnostic = json.loads(diagnostics[0].read_text(encoding="utf-8"))
    assert diagnostic["size"] == size
    assert diagnostic["hashed_bytes"] == lanes.QUARANTINE_HASH_BYTES
    assert diagnostic["hash_complete"] is False
    assert (size, lanes.QUARANTINE_HASH_BYTES) in reads
    assert (size, lanes.MAX_TERMINAL_RECORD_BYTES + 1) not in reads
    assert max(limit for _file_size, limit in reads) <= (
        lanes.MAX_TERMINAL_RECORD_BYTES + 1
    )


def test_recovery_does_not_parse_committed_journals_every_tick(
    tmp_path: Path,
    monkeypatch,
):
    state_root = tmp_path / "cli-lanes"
    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
    committed = lanes._terminal_path(
        lanes.state_path("cassandra", "t_done"), 1, "committed"
    )
    lanes.atomic_json(committed, {"kanban_state": "committed"})
    reads = []
    monkeypatch.setattr(lanes, "load_json", lambda path: (reads.append(path) or {}))
    monkeypatch.setitem(
        sys.modules,
        "hermes_cli",
        SimpleNamespace(kanban_db=SimpleNamespace()),
    )

    assert lanes.recover_pending_finalizations() == 0
    assert reads == []


def test_atomic_json_fsyncs_file_and_directory_and_uses_private_mode(
    tmp_path: Path,
    monkeypatch,
):
    calls = []
    real_fsync = lanes.os.fsync
    monkeypatch.setattr(
        lanes.os,
        "fsync",
        lambda descriptor: (calls.append(descriptor), real_fsync(descriptor))[1],
    )
    path = tmp_path / "state.json"

    lanes.atomic_json(path, {"safe": True})

    assert json.loads(path.read_text()) == {"safe": True}
    assert path.stat().st_mode & 0o777 == 0o600
    assert len(calls) >= 2


def test_atomic_json_refuses_a_precreated_temp_symlink(
    tmp_path: Path,
    monkeypatch,
):
    path = tmp_path / "state.json"
    victim = tmp_path / "credential"
    victim.write_text("do not overwrite", encoding="utf-8")
    temporary = tmp_path / ".state.json.fixed.tmp"
    temporary.symlink_to(victim)
    monkeypatch.setattr(lanes.uuid, "uuid4", lambda: SimpleNamespace(hex="fixed"))

    with pytest.raises(FileExistsError):
        lanes.atomic_json(path, {"unsafe": True})

    assert victim.read_text(encoding="utf-8") == "do not overwrite"
    assert not path.exists()


def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    board = lanes.STATE_ROOT / "cassandra"
    board.mkdir(parents=True)
    quarantine = board / "quarantine"
    quarantine.mkdir()
    old_artifacts = [
        board / "t.run-1.candidate-1.json",
        board / "t.run-1.provider-1.result.json",
        board / "t.run-1.terminal.committed.json",
        board / f"t.run-1.terminal.conflict-{'a' * 32}.json",
        quarantine / "t.invalid.1234.quarantine",
    ]
    for artifact in old_artifacts:
        artifact.write_text("{}", encoding="utf-8")
    quarantine_symlink = quarantine / "attacker.quarantine"
    quarantine_symlink.symlink_to(tmp_path / "missing-target")
    pending = board / "t.run-1.terminal.pending.json"
    pending.write_text("{}", encoding="utf-8")
    prepared = board / f"t.run-1.terminal.prepared-{'b' * 32}.json"
    prepared.write_text("{}", encoding="utf-8")
    old_time = 100.0
    for artifact in old_artifacts:
        os.utime(artifact, (old_time, old_time))
    os.utime(pending, (old_time, old_time))
    os.utime(prepared, (old_time, old_time))

    assert lanes.gc_lane_artifacts(
        now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000
    ) == len(old_artifacts) + 1
    assert not any(artifact.exists() for artifact in old_artifacts)
    assert not quarantine_symlink.is_symlink()
    assert pending.exists()
    assert prepared.exists()


def test_artifact_gc_prunes_oldest_by_count_and_total_bytes(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    board = lanes.STATE_ROOT / "cassandra"
    board.mkdir(parents=True)
    files = []
    for sequence in range(3):
        path = board / f"t.run-1.provider-{sequence}.result.json"
        path.write_text("x" * 40, encoding="utf-8")
        os.utime(path, (100 + sequence, 100 + sequence))
        files.append(path)

    assert lanes.gc_lane_artifacts(
        now=200.0, max_age_seconds=1000, max_count=2, max_bytes=45
    ) == 2
    assert [path.exists() for path in files] == [False, False, True]


def test_artifact_gc_preserves_a_concurrent_replacement(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    board = lanes.STATE_ROOT / "cassandra"
    board.mkdir(parents=True)
    path = board / "t.run-1.provider-1.result.json"
    path.write_text("old", encoding="utf-8")
    os.utime(path, (100, 100))
    replacement = board / "replacement.tmp"
    replacement.write_text("new", encoding="utf-8")
    real_unlink = lanes._unlink_artifact_if_same
    swapped = {"value": False}

    def swap_before_identity_checked(candidate, observed):
        if not swapped["value"]:
            os.replace(replacement, candidate)
            swapped["value"] = True
        return real_unlink(candidate, observed)

    monkeypatch.setattr(lanes, "_unlink_artifact_if_same", swap_before_identity_checked)

    assert lanes.gc_lane_artifacts(
        now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000
    ) == 0
    assert swapped["value"] is True
    assert path.read_text(encoding="utf-8") == "new"