M4.3: a broken build VM blocks its projects until it's ready

The driver can request_reimage(reason) when its build VM is broken past the
usual fixes (before the go-ahead too: it asks, it doesn't act). The harness
parks the work in progress (pushed to Gitea), then blocks the VM in the state
store (schema 5, vm_blocks): no item on any project on it is taken up, across
restarts. The operator hears once at top level, with the command to clear it,
and each blocked item's thread gets a line. `vm ready <project>` lifts the
block and trusts the VM's new host key; a reimaged VM is set up again by the
toolchain step, and the item that found it goes first.

loop.qnt gains the block (blocked_vm_runs_nothing) and a scenario; it found
that a reimage lost work in progress, hence parking first. The walk requests
reimages and recovers them, reimaged or not. The unreachable case reuses the
block in #137. Architecture P5 is softer, and D3 describes the block.

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

nisse-dev/src/nisse_dev/state.py

@@ -29,7 +29,7 @@ from nisse_dev.canonical import (
 from nisse_dev.oplog import OpLog
 from nisse_dev.views import project
 
-CURRENT_SCHEMA_VERSION = 4
+CURRENT_SCHEMA_VERSION = 5
 
 # Stores older than this predate work items (Development Plan M1.1). There was
 # no production data to carry over, so there is no migration from them: start
@@ -38,8 +38,23 @@ OLDEST_SUPPORTED_VERSION = 4
 
 Migration = Callable[[sqlite3.Connection], None]
 
+_VM_BLOCKS = """
+    CREATE TABLE vm_blocks (
+        host TEXT PRIMARY KEY,
+        cause TEXT NOT NULL,
+        reason TEXT NOT NULL,
+        work_item TEXT,
+        since TEXT NOT NULL
+    )
+"""
+
+
+def _add_vm_blocks(conn: sqlite3.Connection) -> None:
+    conn.execute(_VM_BLOCKS)
+
+
 # Maps version N to the migration that takes the schema from N to N+1.
-MIGRATIONS: Mapping[int, Migration] = {}
+MIGRATIONS: Mapping[int, Migration] = {4: _add_vm_blocks}
 
 # Who produced an event (L7): the operator, the orchestrator's own code, or a
 # model role (L9). planner and executor are the old loop's (until M2.8), kept
@@ -77,6 +92,18 @@ class WorkItem:
     closed_at: str | None
 
 
+@dataclass(frozen=True)
+class VMBlock:
+    """A build VM nothing may run on until it is recovered (D3): every
+    project on it waits."""
+
+    host: str
+    cause: str  # broken (the driver said so); unreachable (#137)
+    reason: str
+    work_item: str | None  # the item that found it
+    since: str
+
+
 @dataclass(frozen=True)
 class Intent:
     id: int
@@ -142,6 +169,7 @@ _SCHEMA = [
         marked_at TEXT NOT NULL
     )
     """,
+    _VM_BLOCKS,
 ]
 
 
@@ -369,6 +397,34 @@ class StateStore:
             "work_item.close", cause=cause, work_item=item_id, outcome=outcome
         )
 
+    # --- build VMs (D3) ----------------------------------------------------
+
+    def block_vm(
+        self, host: str, *, cause: str, reason: str, work_item: str | None
+    ) -> None:
+        """Block a build VM; a second block on it keeps the first."""
+        row = self._conn.execute(
+            "INSERT INTO vm_blocks (host, cause, reason, work_item, since) "
+            "VALUES (?, ?, ?, ?, ?) ON CONFLICT(host) DO NOTHING RETURNING host",
+            (host, cause, reason, work_item, _now()),
+        ).fetchone()
+        if row is not None:
+            self.oplog.emit("vm.block", cause=cause, host=host, work_item=work_item)
+
+    def unblock_vm(self, host: str, *, cause: str) -> VMBlock | None:
+        """Lift a build VM's block; returns it, or None if there was none."""
+        row = self._conn.execute(
+            "DELETE FROM vm_blocks WHERE host = ? RETURNING *", (host,)
+        ).fetchone()
+        if row is None:
+            return None
+        self.oplog.emit("vm.unblock", cause=cause, host=host)
+        return VMBlock(**dict(row))
+
+    def vm_blocks(self) -> dict[str, VMBlock]:
+        rows = self._conn.execute("SELECT * FROM vm_blocks").fetchall()
+        return {row["host"]: VMBlock(**dict(row)) for row in rows}
+
     def carry_idle_messages(self, item_id: str) -> int:
         """Copy what the operator said while the agent was idle into a work
         item that has just opened (D1: messages sent before `use` are kept