"""A build VM out of reach (D3, #137): it blocks every project on it, the run
that finds it ends there, and it is tried again in the background; once it
answers, and still answers a settle period later, work goes on by itself. A
new host key is the operator's to trust (`vm ready`), and a VM the driver
found broken is never unblocked by a probe."""

from typing import Any

import pytest
from test_loop_milestones import DON, RESTATE, ScriptedModel, call, git, say
from test_loop_milestones import item as the_item
from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture
from test_loop_toolchain import (
    agent_for,
    by_issue,
    ready,
    reimage,
    said,
)
from test_loop_toolchain import project as project  # noqa: PLC0414 -- the fixture

from nisse_core.execution.ssh import HostKeyChanged, HostUnavailable
from nisse_core.gitea import Issue
from nisse_core.log.canonical import ToolResult
from nisse_dev.loop.items import GO_AHEAD, READY, loop_of

RETRY, SETTLE = 300.0, 180.0


class Faults:
    """The build VM's SSH, broken on demand: ``down`` makes it unreachable,
    ``rekey`` makes it answer with a new host key, and ``drop_on`` takes it
    down when a command containing that text runs. Probes are counted."""

    def __init__(self, monkeypatch: pytest.MonkeyPatch, host: Any) -> None:
        self.down = False
        self.rekey = False
        self.drop_on = ""
        self.probes = 0
        real = type(host).execute
        faults = self

        def execute(self: Any, command: str, **kw: Any) -> Any:
            if faults.drop_on and faults.drop_on in command:
                faults.down = True
            if command == "true":
                faults.probes += 1
            if faults.rekey:
                raise HostKeyChanged("the host key changed")
            if faults.down:
                raise HostUnavailable("ssh: connect to host: Connection timed out")
            return real(self, command, **kw)

        monkeypatch.setattr(type(host), "execute", execute)


class Clock:
    def __init__(self) -> None:
        self.now = 1000.0

    def __call__(self) -> float:
        return self.now


def blocks(world: dict[str, Any]) -> dict[str, str]:
    return {h: b.cause for h, b in world["store"].vm_blocks().items()}


def test_an_unreachable_vm_blocks_and_goes_on_once_settled(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    world["forge"].issues.append(Issue(15, "Add count_lines", "count lines", "don", ()))
    faults = Faults(monkeypatch, world["host"])
    clock = Clock()
    faults.down = True
    agent, transport = agent_for(
        world,
        project,
        ScriptedModel(ready(), RESTATE, RESTATE),  # only once it's back
        clock=clock,
        vm_retry_s=RETRY,
        vm_settle_s=SETTLE,
    )
    agent.intake()
    assert list(blocks(world).values()) == ["unreachable"]
    [told] = [m for m in transport.outbound if "I can't reach the build VM" in m.text]
    assert told.thread is None
    assert "try again every 5 minutes" in told.text
    assert "answered for 3 minutes" in told.text
    assert f"`vm ready {project}`" in told.text  # if it's gone for good
    assert "is out of reach" in agent.status()
    assert loop_of(by_issue(world, 14)).wait == ""  # not waiting on anyone
    assert by_issue(world, 14).position == READY
    # Tried again only when due, and quietly.
    before = len(transport.outbound)
    agent.intake()  # the first round after it went: tried at once
    clock.now += 60
    agent.intake()  # not due yet
    assert faults.probes == 1
    clock.now += RETRY
    agent.intake()
    assert faults.probes == 2
    assert len(transport.outbound) == before
    # Back: it answers, but work waits for it to settle.
    faults.down = False
    clock.now += RETRY
    agent.intake()
    assert blocks(world)
    # Gone again while settling: the wait starts over.
    faults.down = True
    clock.now += SETTLE
    agent.intake()
    assert blocks(world)
    faults.down = False
    clock.now += RETRY
    agent.intake()  # answers
    clock.now += SETTLE - 10
    agent.intake()  # not settled yet
    assert blocks(world)
    clock.now += 10
    agent.intake()  # settled: unblocked, and the work runs
    assert blocks(world) == {}
    assert [t for t in said(transport) if "is back" in t] == [
        f"The build VM {world['host'].describe()} is back: {project} can go on."
    ]
    assert loop_of(by_issue(world, 14)).wait == GO_AHEAD
    assert loop_of(by_issue(world, 15)).wait == GO_AHEAD


def test_a_vm_lost_mid_run_ends_the_run_and_blocks(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    faults.drop_on = "GO-DOWN"
    model = ScriptedModel(
        ready(),
        RESTATE,
        [
            call("start_milestone", name="m"),
            call("run_command", command="echo GO-DOWN"),
            call("run_command", command="echo never"),
        ],
        # Nothing more: the model is never shown the failure.
    )
    agent, transport = agent_for(world, project, model, clock=Clock())
    agent.intake()
    say(transport, world, "go")
    assert list(blocks(world).values()) == ["unreachable"]
    item = the_item(world)
    assert item.position == READY
    results = [
        e.event.result
        for e in world["store"].transcript(item.id, "driver").read()
        if isinstance(e.event, ToolResult)
    ]
    assert results[-2].startswith("not finished: the build VM is unreachable")
    assert results[-1] == "not run: the turn had ended"
    assert not model.turns


def test_a_vm_lost_in_a_requested_step_ends_the_drivers_run(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    faults.drop_on = "GO-DOWN"
    model = ScriptedModel(
        ready(),
        RESTATE,
        [
            call("start_milestone", name="m"),
            call("toolchain_step", reason="it needs httpx"),
        ],
        [call("run_command", command="pip install httpx # GO-DOWN")],  # the step
    )
    agent, transport = agent_for(world, project, model, clock=Clock())
    agent.intake()
    say(transport, world, "go")
    assert list(blocks(world).values()) == ["unreachable"]
    assert the_item(world).position == READY
    assert not model.turns  # neither the step nor the driver went on


def test_a_vm_back_with_a_new_host_key_waits_for_the_operator(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    forgotten: list[str] = []
    monkeypatch.setattr(
        type(world["host"]), "forget_host_key", lambda self: forgotten.append("yes")
    )
    clock = Clock()
    faults.down = True
    agent, transport = agent_for(
        world, project, ScriptedModel(ready(), RESTATE), clock=clock
    )
    agent.intake()
    faults.down, faults.rekey = False, True
    clock.now += RETRY
    agent.intake()
    assert list(blocks(world).values()) == ["host key"]
    [asked] = [t for t in said(transport) if "different host key" in t]
    assert f"`vm ready {project}`" in asked
    clock.now += RETRY + SETTLE
    agent.intake()
    agent.intake()
    assert blocks(world)  # never cleared by a probe
    faults.rekey = False
    transport.receive(DON, f"vm ready {project}")
    assert forgotten == ["yes"]
    assert blocks(world) == {}
    assert loop_of(the_item(world)).wait == GO_AHEAD


def test_the_operator_can_clear_an_unreachable_block(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    faults.down = True
    agent, transport = agent_for(
        world, project, ScriptedModel(ready(), RESTATE), clock=Clock()
    )
    agent.intake()
    assert blocks(world)
    faults.down = False
    transport.receive(DON, "vm ready")
    assert blocks(world) == {}
    assert loop_of(the_item(world)).wait == GO_AHEAD


def test_a_broken_vm_is_never_unblocked_by_a_probe(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    clock = Clock()
    agent, _ = agent_for(world, project, ScriptedModel(ready(), reimage()), clock=clock)
    agent.intake()
    for _ in range(3):
        clock.now += RETRY + SETTLE
        agent.intake()
    assert list(blocks(world).values()) == ["broken"]
    assert faults.probes == 0


def test_an_unreachable_block_survives_a_restart_and_is_tried_at_once(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    faults.down = True
    clock = Clock()
    agent, _ = agent_for(world, project, ScriptedModel(), clock=clock)
    agent.intake()
    agent.intake()
    assert faults.probes == 1
    again, _ = agent_for(world, project, ScriptedModel(), clock=clock)
    again.boot()
    again.intake()  # a new process knows the block, not when it last tried
    assert faults.probes == 2
    assert list(blocks(world).values()) == ["unreachable"]


def test_a_question_is_answered_while_the_vm_is_blocked(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    model = ScriptedModel(
        ready(), RESTATE, [call("reply", text="Because counting words is asked.")]
    )
    agent, transport = agent_for(world, project, model, clock=Clock())
    agent.intake()
    faults.down = True
    say(transport, world, "why count words?")  # L5: a question
    assert list(blocks(world).values()) == ["unreachable"]
    assert "Because counting words is asked." in said(transport)
    loop = loop_of(the_item(world))
    assert loop.wait == GO_AHEAD  # back to its wait, restatement still up


def test_a_vm_lost_while_the_step_records_it_blocks_before_any_work(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    faults.drop_on = "base64 -d >"  # writing the marker
    model = ScriptedModel(ready())  # no driver run: there is no marker
    agent, _ = agent_for(world, project, model, clock=Clock())
    agent.intake()
    assert list(blocks(world).values()) == ["unreachable"]
    assert not model.turns


def test_a_lost_runs_work_is_parked_before_the_vm_is_used_again(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    clock = Clock()
    model = ScriptedModel(
        ready(),
        RESTATE,
        [
            call("start_milestone", name="m"),
            call("write_file", path="w.py", content="x = 1\n"),
            call("run_command", command="echo GO-DOWN"),
        ],
        [call("finish_milestone", summary="w.py, tested", final="yes")],
    )
    agent, transport = agent_for(world, project, model, clock=clock)
    agent.intake()
    faults.drop_on = "GO-DOWN"
    say(transport, world, "go")  # writes w.py, then loses its VM
    assert blocks(world)
    faults.drop_on, faults.down = "", False
    parked: list[str] = []
    real = agent._schedule

    def schedule() -> None:
        # What Gitea holds when the VM is first used again.
        refs = git("--git-dir", str(world["forge"].bare), "branch", "--list")
        parked.append(refs)
        real()

    monkeypatch.setattr(agent, "_schedule", schedule)
    clock.now += RETRY
    agent.intake()  # answers
    clock.now += SETTLE
    agent.intake()  # settled: parked, unblocked, and the item goes on
    assert blocks(world) == {}
    assert f"{the_item(world).id}-parked" in parked[-1]
    forge = world["forge"]
    [pr] = [n for n, p in forge.prs.items() if not p["head"].startswith("nisse-check")]
    assert "w.py" in forge.files(pr)


def test_a_vm_lost_just_before_the_step_blocks_before_any_model_call(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    faults = Faults(monkeypatch, world["host"])
    faults.drop_on = "git status --porcelain"  # what the step checks first
    model = ScriptedModel()  # no model call at all
    agent, _ = agent_for(world, project, model, clock=Clock())
    agent.intake()
    assert list(blocks(world).values()) == ["unreachable"]


def test_a_vm_lost_as_a_helper_hands_back_ends_the_drivers_run(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    # Found by the property walk (seed 468): taking the helper's work back
    # onto the milestone's branch failed for the lost VM, was noted as a git
    # error, and the driver's model was called with the VM gone.
    from test_loop_help import Spy, hand_back, helpers

    faults = Faults(monkeypatch, world["host"])

    class Leaving(Spy):
        def complete(self, events, tools, model, system=None) -> list[Any]:
            out = super().complete(events, tools, model, system)
            if not self.turns:  # gone as its work is taken back
                faults.drop_on = "checkout -q -f -B"
            return out

    helper = Leaving(
        [call("write_file", path="helped.py", content="H = 1\n")], [hand_back()]
    )
    model = ScriptedModel(
        ready(),
        RESTATE,
        [
            call("start_milestone", name="m"),
            call("delegate", task="write helped.py"),
        ],
        [call("run_command", command="echo never")],  # never asked for
    )
    agent, transport = agent_for(
        world, project, model, clock=Clock(), helpers=helpers(helper)
    )
    agent.intake()
    say(transport, world, "go")
    assert list(blocks(world).values()) == ["unreachable"]
    assert the_item(world).position == READY
    assert len(model.turns) == 1  # the driver's model wasn't called again