"""Run provenance and first-writer conflict behavior."""

from __future__ import annotations

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


def test_legacy_valid_retirement_staging_is_promoted_without_reexecution(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    board = lanes.STATE_ROOT / "cassandra"
    board.mkdir(parents=True)
    staged = board / f".retire.{'d' * 32}.0"
    record = _pending_terminal_record(
        "cassandra", "t_legacy_staged", 56, "legacy staged result"
    )
    lanes.atomic_json(staged, record)
    task = SimpleNamespace(
        id="t_legacy_staged",
        status="running",
        current_run_id=56,
        completed_run_id=None,
        result=None,
        assignee="cli-auto",
    )
    completions = []
    reclaims = []
    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)

    assert lanes._has_pending_finalization(
        "cassandra", "t_legacy_staged", 56
    ) is True
    assert lanes.recover_pending_finalizations() == 1
    assert completions == [record["result"]]
    assert reclaims == []
    assert not staged.exists()
    assert lanes.recover_pending_finalizations() == 0
    assert completions == [record["result"]]


def test_done_result_bytes_do_not_authorize_a_different_run_journal(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path, stale = lanes._write_terminal_record(
        lanes.state_path("cassandra", "t_provenance"),
        board="cassandra",
        task_id="t_provenance",
        run_id=1,
        structured=_completed_result("same bytes"),
        summary="same bytes",
        metadata={},
    )
    task = SimpleNamespace(
        id="t_provenance",
        status="done",
        current_run_id=None,
        completed_run_id=2,
        result=stale["result"],
        assignee="cli-auto",
    )

    class Connection:
        def close(self):
            return None

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, _task_id: task,
        complete_task=lambda *_args, **_kwargs: pytest.fail(
            "a done task with different run provenance must not be completed"
        ),
    )
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))

    assert lanes.recover_pending_finalizations() == 0
    assert not path.exists()
    assert not list(path.parent.glob("*.terminal.committed.json"))
    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
    assert len(conflicts) == 1
    conflict = json.loads(conflicts[0].read_text())
    assert conflict["expected_run_id"] == 1
    assert conflict["result"] == task.result


@pytest.mark.parametrize(
    "swap_point",
    [
        "during-complete",
        "after-db-before-promote",
        "before-committed-create",
        "before-pending-retire",
    ],
)
def test_terminal_first_writer_preserves_valid_replacement_conflicts(
    tmp_path: Path,
    monkeypatch,
    swap_point: str,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path, old_record = lanes._write_terminal_record(
        lanes.state_path("cassandra", "t_first_writer"),
        board="cassandra",
        task_id="t_first_writer",
        run_id=27,
        structured=_completed_result("first result"),
        summary="first result",
        metadata={"writer": "first"},
    )
    replacement_structured = _completed_result("replacement result")
    replacement_record = {
        "board": "cassandra",
        "task_id": "t_first_writer",
        "expected_run_id": 27,
        "result": json.dumps(replacement_structured, sort_keys=True),
        "summary": "replacement result",
        "metadata": {"writer": "replacement"},
        "kanban_state": "pending",
        "recorded_at": lanes.utc_now(),
    }
    replacement = path.with_name(f"replacement-{swap_point}.tmp")
    lanes.atomic_json(replacement, replacement_record)
    swapped = {"value": False}

    def swap_pending():
        if not swapped["value"]:
            os.replace(replacement, path)
            swapped["value"] = True

    task = SimpleNamespace(
        id="t_first_writer",
        status="running",
        result=None,
        current_run_id=27,
        assignee="cli-auto",
    )

    class Connection:
        def close(self):
            return None

    def complete_task(_conn, _task_id, **kwargs):
        task.status = "done"
        task.result = kwargs["result"]
        task.current_run_id = None
        if swap_point == "during-complete":
            swap_pending()
        return True

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, _task_id: task,
        complete_task=complete_task,
    )
    if swap_point == "after-db-before-promote":
        real_promote = lanes._promote_prepared_evidence

        def swap_then_promote(*args, **kwargs):
            swap_pending()
            return real_promote(*args, **kwargs)

        monkeypatch.setattr(lanes, "_promote_prepared_evidence", swap_then_promote)
    elif swap_point == "before-committed-create":
        real_create = lanes._write_json_noreplace

        def swap_before_committed(path_arg, value):
            if path_arg.name.endswith(".terminal.committed.json"):
                swap_pending()
            return real_create(path_arg, value)

        monkeypatch.setattr(lanes, "_write_json_noreplace", swap_before_committed)
    elif swap_point == "before-pending-retire":
        real_retire = lanes._retire_snapshot

        def swap_before_retire(path_arg, snapshot):
            if path_arg == path:
                swap_pending()
            return real_retire(path_arg, snapshot)

        monkeypatch.setattr(lanes, "_retire_snapshot", swap_before_retire)

    assert lanes._finalize_terminal_record(fake_db, path, old_record) == "committed"

    assert swapped["value"] is True
    assert task.result == old_record["result"]
    assert not path.exists()
    committed = list(path.parent.glob("*.terminal.committed.json"))
    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
    prepared = list(path.parent.glob("*.terminal.prepared-*.json"))
    assert len(committed) == 1
    assert len(conflicts) == 1
    assert prepared == []
    assert committed[0].stat().st_mode & 0o777 == 0o600
    assert conflicts[0].stat().st_mode & 0o777 == 0o600
    committed_document = json.loads(committed[0].read_text(encoding="utf-8"))
    conflict_document = json.loads(conflicts[0].read_text(encoding="utf-8"))
    assert committed_document["result"] == old_record["result"]
    assert committed_document["kanban_state"] == "committed"
    assert conflict_document["result"] == replacement_record["result"]
    assert conflict_document["metadata"] == {"writer": "replacement"}
    assert conflict_document["kanban_state"] == "conflict"


def test_terminal_commit_directory_fsync_failure_recovers_from_prepared_evidence(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    path, record = lanes._write_terminal_record(
        lanes.state_path("cassandra", "t_post_db_fsync"),
        board="cassandra",
        task_id="t_post_db_fsync",
        run_id=28,
        structured=_completed_result("durable DB winner"),
        summary="durable DB winner",
        metadata={},
    )
    task = SimpleNamespace(
        id="t_post_db_fsync",
        status="running",
        result=None,
        current_run_id=28,
        completed_run_id=None,
        assignee="cli-auto",
    )
    replacement_structured = _completed_result("replacement after DB commit")
    replacement_record = {
        "board": "cassandra",
        "task_id": "t_post_db_fsync",
        "expected_run_id": 28,
        "result": json.dumps(replacement_structured, sort_keys=True),
        "summary": "replacement after DB commit",
        "metadata": {"writer": "replacement"},
        "kanban_state": "pending",
        "recorded_at": lanes.utc_now(),
    }
    replacement = path.with_name("post-db-fsync-replacement.tmp")
    lanes.atomic_json(replacement, replacement_record)

    class Connection:
        def close(self):
            return None

    def complete_task(_conn, _task_id, **kwargs):
        task.status = "done"
        task.result = kwargs["result"]
        task.completed_run_id = kwargs["expected_run_id"]
        task.current_run_id = None
        os.replace(replacement, path)
        return True

    fake_db = SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: Connection(),
        get_task=lambda _conn, _task_id: task,
        complete_task=complete_task,
    )
    real_fsync = lanes.os.fsync
    failed = {"value": False}

    def fail_committed_directory_fsync(descriptor):
        if (
            stat.S_ISDIR(os.fstat(descriptor).st_mode)
            and list(path.parent.glob("*.terminal.committed.json"))
            and not failed["value"]
        ):
            failed["value"] = True
            raise OSError(errno.ENOSPC, "post-DB committed directory fsync failed")
        real_fsync(descriptor)

    monkeypatch.setattr(lanes.os, "fsync", fail_committed_directory_fsync)

    with pytest.raises(OSError, match="post-DB committed"):
        lanes._finalize_terminal_record(fake_db, path, record)

    assert task.status == "done"
    assert task.result == record["result"]
    assert path.exists()
    assert json.loads(path.read_text(encoding="utf-8"))["result"] == (
        replacement_record["result"]
    )
    assert len(list(path.parent.glob("*.terminal.prepared-*.json"))) == 1

    monkeypatch.setattr(lanes.os, "fsync", real_fsync)
    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
    assert lanes.recover_pending_finalizations() == 1
    assert not path.exists()
    assert list(path.parent.glob("*.terminal.prepared-*.json")) == []
    committed = list(path.parent.glob("*.terminal.committed.json"))
    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
    assert len(committed) == 1
    assert len(conflicts) == 1
    assert json.loads(committed[0].read_text())["result"] == record["result"]
    assert json.loads(conflicts[0].read_text())["result"] == replacement_record["result"]