M4.4: an unreachable build VM blocks its projects until it's back (#137)

M4.3's block, with new triggers and a way out. A VM out of reach at the
check before a run, in a run (which ends there: the driver, a toolchain
step and a requested step), or while preparing the workspace blocks every
project on it; the item stays ready, waiting on no one. On the intake round
it is tried again every tuning.build_vm.retry_s (300), and once it answers
it must still answer settle_s (180) later; then the work a lost run left
there is parked (pushed), the VM unblocked, and that said once.

A changed host key, at the check or on a retry, blocks the VM too, until
the operator's `vm ready`, which trusts the new key: `trust new host key`
is retired, and the message says it's a key problem that may be an
impersonation. Questions are answered while a VM is blocked, without it.
Starting the agent with a VM down no longer fails.

The walk breaks the VM's SSH, now or partway through a run, and checks that
no model works on a VM that is down; loop.qnt models the VM going and
coming back (unblocked_only_when_up). Together they found: a lost run's
work set aside by the next item there; a question left behind a block; a
lost VM swallowed while the toolchain step checked or recorded it; and a
remark waiting to be answered that outlived its PR's merge. All fixed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk

nisse-dev/tests/test_config.py

@@ -167,6 +167,7 @@ def test_tuning_is_read() -> None:
             "delegation": {"max_turns": 20, "max_tokens": 100000, "wall_clock_s": 600},
             "tools": {"command_timeout_s": 300, "max_output_chars": 2000},
             "loop": {"bug_grace_s": 120},
+            "build_vm": {"retry_s": 60},
         }
     )
     tuning = config.tuning
@@ -176,6 +177,7 @@ def test_tuning_is_read() -> None:
     assert tuning.delegation.wall_clock_s == 600.0
     assert tuning.tools.max_output_chars == 2000
     assert tuning.loop.bug_grace_s == 120.0
+    assert (tuning.build_vm.retry_s, tuning.build_vm.settle_s) == (60.0, 180.0)
 
 
 def test_a_partly_given_tuning_section_keeps_the_other_defaults() -> None:

nisse-dev/tests/test_loop.py

@@ -24,6 +24,7 @@ from nisse_dev.loop.items import CLOSED, GO_AHEAD, READY, RUNNING, WAITING, loop
 from nisse_dev.loop.reader import command_form
 from nisse_dev.oplog import OpLog
 from nisse_dev.state import StateStore
+from nisse_dev.tools import ToolOutcome
 from nisse_dev.transport import InMemoryTransport
 from nisse_dev.worker import InlineWorker
 
@@ -430,7 +431,7 @@ def test_a_command_that_works_resets_the_failure_count(tmp_path: Path) -> None:
     class Commands:
         def dispatch(self, tool: str, arguments: dict[str, Any]) -> Any:
             ok = arguments["command"] == "good"
-            return type("Outcome", (), {"result": "out", "success": ok})()
+            return ToolOutcome("out", ok)
 
     def run(command: str) -> list[Any]:
         return [call("run_command", command=command)]

nisse-dev/tests/test_loop_live_fixes.py

@@ -156,6 +156,34 @@ def test_a_review_comment_can_be_answered_without_a_fix(
     assert "It's a constant on purpose" in forge.posted[-1][1]
 
 
+def test_a_remark_waiting_to_be_answered_doesnt_outlive_the_merge(
+    world: dict[str, Any],
+) -> None:
+    """#137's walk: a review comment readies the item to answer it and go back
+    to waiting for the merge; if the merge comes before that run (a blocked
+    VM, a queue), there is no merge left to go back to."""
+    forge: Forge = world["forge"]
+    model = ScriptedModel(
+        RESTATE,
+        [
+            call("start_milestone", name="words"),
+            call("write_file", path="w.py", content="W = 1\n"),
+            call("finish_milestone", summary="words", final="yes"),
+        ],
+    )
+    agent, transport = loop_agent(world, model)
+    agent.intake()
+    say(transport, world, "go")
+    [pr] = forge.prs
+    forge.comments[pr] = [Comment(7, "don", "why a constant", "t")]
+    agent.watch()  # not run yet
+    assert loop_of(the_item(world)).back_to == MERGE
+    forge.merge(pr)
+    agent.watch()
+    loop = loop_of(the_item(world))
+    assert (loop.pr, loop.back_to, loop.back_note) == (0, "", "")
+
+
 # --- the session's later findings ----------------------------------------------
 
 

nisse-dev/tests/test_loop_properties.py

@@ -35,7 +35,7 @@ from nisse_dev.canonical import (
     UserMessage,
 )
 from nisse_dev.delegation import DELEGATION_INTENT, Caps
-from nisse_dev.execution import SSHHost
+from nisse_dev.execution import HostUnavailable, SSHHost
 from nisse_dev.gitea import Comment, Issue
 from nisse_dev.llm import LLMError, RoleMap
 from nisse_dev.loop.agent import LoopAgent, Project
@@ -209,13 +209,30 @@ class Watched:
     def complete(self, events, tools, model, system=None) -> list[Any]:
         offered = {t["function"]["name"] for t in tools}
         item = self.walk.world["store"].active_item()
-        working = "toolchain_ready" not in offered and "propose_work" not in offered
+        # Working on the VM: offered its tools (an answer given while the VM
+        # is blocked isn't, #137), and not setting it up.
+        working = (
+            "read_file" in offered
+            and "toolchain_ready" not in offered
+            and "propose_work" not in offered
+        )
         marked = (
             item is not None and (self.walk.markers / f"{item.scope}.ready").exists()
         )
         if item is not None and item.scope != "general" and working and not marked:
             self.walk.unready.append(item.id)
-        if item is not None and item.scope in self.walk.worlds:
+        if (
+            item is not None
+            and item.scope != "general"
+            and "read_file" in offered
+            and self.walk.down
+        ):
+            self.walk.while_down.append(item.id)
+        if (
+            item is not None
+            and "read_file" in offered
+            and item.scope in self.walk.worlds
+        ):
             host = self.walk.worlds[item.scope]["host"].describe()
             if host in self.walk.world["store"].vm_blocks():
                 self.walk.on_blocked.append(item.id)
@@ -282,7 +299,12 @@ class Walk:
     """``worlds``: project name -> its world (forge, bare repo, clone,
     milestones), all over one store."""
 
-    def __init__(self, worlds: dict[str, dict[str, Any]], seed: int) -> None:
+    def __init__(
+        self,
+        worlds: dict[str, dict[str, Any]],
+        seed: int,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
         self.worlds = worlds
         self.world = next(iter(worlds.values()))  # the store's home
         self.rng = random.Random(seed)
@@ -298,6 +320,26 @@ class Walk:
         # without one.
         self.markers = Path(self.world["repo"].root).parent / "markers"
         self.unready: list[str] = []
+        # #137: the build VM out of reach, now or after a few more commands
+        # (so mid-run); model calls made while it was.
+        self.down = False
+        self.drop_in = 0
+        self.while_down: list[str] = []
+        self.checking = False  # the walk's own checks see the real VM
+        real = SSHHost.execute
+        walk = self
+
+        def execute(host: SSHHost, command: str, **kw: Any) -> Any:
+            if walk.checking:
+                return real(host, command, **kw)
+            if walk.drop_in:
+                walk.drop_in -= 1
+                walk.down = walk.down or walk.drop_in == 0
+            if walk.down:
+                raise HostUnavailable("ssh: connect to host: Connection timed out")
+            return real(host, command, **kw)
+
+        monkeypatch.setattr(SSHHost, "execute", execute)
         self.wiped_at: dict[str, int] = {}  # project -> last event before a wipe
         self.on_blocked: list[str] = []  # items run while their VM was blocked
         if self.merge == "approved":
@@ -406,6 +448,7 @@ class Walk:
         }
         if self.world["store"].vm_blocks():
             weights.update({"vm-ready": 3})
+        weights.update({"vm-up": 3} if self.down else {"vm-down": 1})
         item = self.rng.choice(open_items) if open_items else None
         loop = loop_of(item) if item else None
         # The forge an item's project lives on; a new issue picks one.
@@ -435,6 +478,14 @@ class Walk:
             self.agent.intake()
         elif action == "wipe":
             self._wipe(project)
+        elif action == "vm-down":
+            # #137: out of reach now, or partway through what comes next.
+            if self.rng.random() < 0.5:
+                self.down = True
+            else:
+                self.drop_in = self.rng.randint(1, 30)
+        elif action == "vm-up":
+            self.down, self.drop_in = False, 0
         elif action == "vm-ready":
             # M4.3: the operator recovers a blocked VM, reimaged or not; a
             # reimage wipes every project on it.
@@ -616,9 +667,18 @@ class Walk:
                     )
 
     def check(self) -> None:
+        self.checking = True
+        try:
+            self._check()
+        finally:
+            self.checking = False
+
+    def _check(self) -> None:
         where = f"seed {self.seed}, steps {self.steps}"
         # M4.1: the driver never works on a project whose VM has no marker.
         assert not self.unready, f"ready_before_work: {self.unready}: {where}"
+        # #137: no model works on a project whose build VM is out of reach.
+        assert not self.while_down, f"no_model_on_a_lost_vm: {self.while_down}: {where}"
         # M4.3: nothing runs for a project whose build VM is blocked.
         assert not self.on_blocked, (
             f"blocked_vm_runs_nothing: {self.on_blocked}: {where}"
@@ -859,13 +919,21 @@ class Walk:
         for item in self.items():
             if item.status != "open" or item.scope != name:
                 continue
+            events = self.world["store"].transcript(item.id, "driver").read()
+            # Only writes that happened: one cut off by a lost VM (#137) didn't.
+            done = {
+                e.event.call_id
+                for e in events
+                if isinstance(e.event, ToolResult) and e.event.success
+            }
             written = {
                 str(e.event.arguments.get("path"))
-                for e in self.world["store"].transcript(item.id, "driver").read()
+                for e in events
                 # The driver's own: a failed helper's are discarded on purpose.
                 if isinstance(e.event, ToolCall)
                 and e.event.tool == "write_file"
                 and e.actor == "driver"
+                and e.event.call_id in done
                 # D3: a wiped VM loses what wasn't pushed, accepted.
                 and e.id > self.wiped_at.get(name, 0)
             }
@@ -932,7 +1000,7 @@ def test_the_loop_keeps_its_properties(
     world: dict[str, Any], seed: int, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     _seeded_ids(monkeypatch, seed)
-    walk = Walk({"sandbox": world}, seed)
+    walk = Walk({"sandbox": world}, seed, monkeypatch)
     for _ in range(STEPS):
         walk.step()
         walk.check()
@@ -948,7 +1016,7 @@ def test_two_projects_keep_the_properties(
     other = make_world(
         ssh_host, tmp_path / "other", store=sandbox["store"], repo_name="don/other"
     )
-    walk = Walk({"sandbox": sandbox, "other": other}, seed)
+    walk = Walk({"sandbox": sandbox, "other": other}, seed, monkeypatch)
     for _ in range(STEPS):
         walk.step()
         walk.check()

nisse-dev/tests/test_loop_toolchain.py

@@ -66,7 +66,7 @@ def markers(world: dict[str, Any]) -> str:
 
 
 def agent_for(
-    world: dict[str, Any], project: str, model: ScriptedModel
+    world: dict[str, Any], project: str, model: ScriptedModel, **more: Any
 ) -> tuple[LoopAgent, InMemoryTransport]:
     transport = InMemoryTransport()
     cancel = Cancel()
@@ -92,6 +92,7 @@ def agent_for(
         preflight=True,
         preflight_turns=5,
         markers=markers(world),
+        **more,
     )
     transport.on_message(loop.on_message)
     return loop, transport
@@ -236,7 +237,7 @@ def test_a_missing_baseline_waits_on_the_operator(
     assert loop.wait == ANSWER and "lacks no-such-tool-xyz" in loop.note
 
 
-def test_a_changed_host_key_waits_for_the_operators_word(
+def test_a_changed_host_key_blocks_the_vm_until_the_operators_word(
     world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     real = preflight.check
@@ -254,13 +255,15 @@ def test_a_changed_host_key_waits_for_the_operators_word(
     )
     agent, transport = agent_for(world, project, ScriptedModel(ready(), RESTATE))
     agent.intake()
-    loop = loop_of(the_item(world))
-    assert loop.host_key_changed and "trust new host key" in loop.note
+    [block] = world["store"].vm_blocks().values()
+    assert block.cause == "host key"
+    [asked] = [t for t in said(transport) if "different host key" in t]
+    assert f"`vm ready {project}`" in asked and "impersonating" in asked
+    assert "has a new host key" in agent.status()
     say(transport, world, "it's fine")  # an answer isn't the operator's word
-    still = loop_of(the_item(world))
-    assert (still.wait, still.host_key_changed, forgotten) == (ANSWER, True, [])
+    assert world["store"].vm_blocks() and forgotten == []
     changed["on"] = False
-    say(transport, world, "trust new host key")
+    transport.receive(DON, f"vm ready {project}")
     assert forgotten == ["yes"]
     assert loop_of(the_item(world)).wait == GO_AHEAD  # set up, then planned
 
@@ -386,7 +389,7 @@ def test_a_broken_vm_blocks_its_project_until_the_operator_says_ready(
     # Neither runs while the VM is blocked, even after a restart.
     again, _ = agent_for(world, project, ScriptedModel())
     assert again._queue() == []
-    assert "its build VM needs recovery" in again.status()
+    assert "its build VM needs reimaging" in again.status()
     # The operator reimages it: the clone and the marker are gone.
     shutil.rmtree(world["repo"].root)
     world["host"].execute(f'rm -f "{preflight.marker_path(project, markers(world))}"')

nisse-dev/tests/test_loop_vm.py

@@ -0,0 +1,340 @@
+"""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_dev.canonical import ToolResult
+from nisse_dev.execution import HostKeyChanged, HostUnavailable
+from nisse_dev.gitea import Issue
+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"]