From a921b73d34e9c79971820fa797e40a7b09ff91ad Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Thu, 8 Oct 2026 21:06:05 +0000 Subject: [PATCH] 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) Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk --- docs/Architecture.md | 8 +- docs/Development-Plan.md | 38 ++- docs/Tuning-Guide.md | 22 ++ nisse-dev.example.yaml | 3 + nisse-dev/src/nisse_dev/config.py | 10 + nisse-dev/src/nisse_dev/loop/agent.py | 372 +++++++++++++++++------- nisse-dev/src/nisse_dev/loop/build.py | 2 + nisse-dev/src/nisse_dev/loop/driver.py | 35 ++- nisse-dev/src/nisse_dev/loop/items.py | 6 +- nisse-dev/src/nisse_dev/state.py | 11 +- nisse-dev/src/nisse_dev/tools.py | 16 +- nisse-dev/tests/test_config.py | 2 + nisse-dev/tests/test_loop.py | 3 +- nisse-dev/tests/test_loop_live_fixes.py | 28 ++ nisse-dev/tests/test_loop_properties.py | 82 +++++- nisse-dev/tests/test_loop_toolchain.py | 19 +- nisse-dev/tests/test_loop_vm.py | 340 ++++++++++++++++++++++ specs/loop.qnt | 82 +++++- specs/loop_scenarios.qnt | 14 +- 19 files changed, 935 insertions(+), 158 deletions(-) create mode 100644 nisse-dev/tests/test_loop_vm.py diff --git a/docs/Architecture.md b/docs/Architecture.md index eb92ba0..d54a88c 100644 --- a/docs/Architecture.md +++ b/docs/Architecture.md @@ -709,8 +709,12 @@ driver asks for the VM to be reimaged from its template. The harness pushes the work in progress first, then blocks the VM: nothing runs on any project on it, across restarts, until the operator says `vm ready` (which also trusts its new host key), and a reimaged VM is set up again by the toolchain step. A VM that -can't be reached blocks the same way (#137). The operator recovers it for now; -`nisse-ops` later (O5). +can't be reached blocks the same way, whether the check before a run finds it +or a run loses it (which ends that run there); it is tried again in the +background, and work goes on by itself once it has answered for a while. A VM +that answers with a new host key blocks until the operator's `vm ready`. +Questions are still answered while a VM is blocked, without it. The operator +recovers a VM for now; `nisse-ops` later (O5). ### D4. Git, branches and PRs diff --git a/docs/Development-Plan.md b/docs/Development-Plan.md index 7080c4e..e61aa37 100644 --- a/docs/Development-Plan.md +++ b/docs/Development-Plan.md @@ -377,7 +377,8 @@ agent merging behind protection it can't satisfy, verified by a test (D4, D5, #6 ### Phase 4 — The environment -**M4.1 Toolchain preflight, M4.2 the toolchain step, M4.3 environment repair.** +**M4.1 Toolchain preflight, M4.2 the toolchain step, M4.3 environment repair, +M4.4 an unreachable build VM.** *Pre-approved.* All on the build VM (D3). **M4.1** is **done** (this PR). A model works out and sets up what a project @@ -397,7 +398,7 @@ needs; code makes sure that happens, bounds it and checks the result. - Later runs check the versions again; a change is a fact and a thread line, and the PR body says what the work was tested with. - A changed host key is its own failure: the item waits for the operator's - `trust new host key`, then the VM is set up again. + `trust new host key`, then the VM is set up again (since M4.4, `vm ready`). - Decided with the operator: the model judges toolchains (the harness can't without it), preflight runs on the driver's model, gaps don't block. Turning off unattended upgrades and the template's baseline are template choices. @@ -434,9 +435,34 @@ past the usual fixes (D3). - `loop.qnt` gained the block (`blocked_vm_runs_nothing`); it found that work in progress was lost on a reimage, hence parking first. The walk requests reimages and recovers them, reimaged or not. -- Next: an unreachable VM blocks through the same mechanism, unblocked by a slow - background retry once it is back (#137). Later, nisse-ops reimages a VM itself - (M6.2, M7.6); a changed address means editing `build_host` and restarting. +- Later, nisse-ops reimages a VM itself (M6.2, M7.6); a changed address means + editing `build_host` and restarting. + +**M4.4** *(#137, with the operator).* **Done** (this PR). A VM that can't be +reached blocks through M4.3's mechanism; only the triggers and the way out +differ. +- Triggers: the check before a run, a run that loses the VM (it ends there, so + no model works against a missing VM: the toolchain step and a requested step + too), and preparing the workspace. The item stays ready, waiting on no one. +- Tried again every `tuning.build_vm.retry_s` (5 min) on the intake round, said + only in the operations log; once it answers, it must still answer + `settle_s` (3 min) later, so a VM that has just booted is left to finish. + Then the work a lost run left on it is parked (pushed), before any item can + set it aside, and it is unblocked and said once. `vm ready` clears it too; + the message warns that a reimage loses what isn't pushed. +- A changed host key, at the check or on a retry, blocks the VM as well, and is + cleared only by `vm ready`, which trusts the new key: `trust new host key` is + retired, and the message says it's a key problem and that it may be an + impersonation. +- Questions are answered while a VM is blocked, without it (L5: a question + changes nothing). 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. With `loop.qnt`, it found: work a lost + run left behind set aside by the next item there (hence parking first); a + question left unanswered behind a block (hence answering without the VM); + a lost VM swallowed while the toolchain step wrote its marker; and a remark + waiting to be answered that outlived its PR's merge, sending the item back + to wait for a merge that had happened. All fixed. ### Phase 5 — Memory @@ -496,7 +522,7 @@ vision; one local model for every agent; several forges (#26); a room per projec | #69 inline review comments | M2.1 | | #70 watch stall | M2.1 | | #77 conversational agent | M2.3 (questions), M2.7 | -| #137 block on an unreachable build VM | After M4.3, which built the block | +| #137 block on an unreachable build VM | M4.4 | ## 6. Implementation notes diff --git a/docs/Tuning-Guide.md b/docs/Tuning-Guide.md index dd0de1c..7715f66 100644 --- a/docs/Tuning-Guide.md +++ b/docs/Tuning-Guide.md @@ -565,3 +565,25 @@ reply in its thread starts a new conversation linked to it. - **Too low:** a conversation you come back to after lunch has to be picked up again from its summary. - **Too high:** idle conversations pile up in `status` and in the queue. + +### `tuning.build_vm.retry_s` — default 300 + +How often a build VM that can't be reached is tried again (architecture D3). +While it's out of reach, every project on it is blocked; each try is one SSH +connection that gives up after 15 seconds. + +- **Too low:** a VM that is down for long is tried constantly, and each try + holds up the agent's other duties for up to 15 seconds. +- **Too high:** work waits longer than it needs to after the VM is back. +- **Deciding:** about how long a reboot of the VM's host takes. A VM you know + is back can be cleared at once with `vm ready `. + +### `tuning.build_vm.settle_s` — default 180 + +Once an unreachable VM answers again, how long it must keep answering before +work goes on: SSH comes up before a VM has finished booting. + +- **Too low:** work starts on a VM still running cloud-init or starting + services, and fails in confusing ways. +- **Too high:** work waits after the VM is ready. +- **Deciding:** how long the template takes from SSH answering to fully up. diff --git a/nisse-dev.example.yaml b/nisse-dev.example.yaml index ba612e7..efc11a7 100644 --- a/nisse-dev.example.yaml +++ b/nisse-dev.example.yaml @@ -108,3 +108,6 @@ tuning: loop: # the new loop (loop: new) bug_grace_s: 600 # a `bug` restatement proceeds after this long unanswered general_idle_s: 86400 # a conversation closes after this long without a message + build_vm: # an unreachable build VM (D3, #137) + retry_s: 300 # how often it is tried again + settle_s: 180 # once it answers, how long until work goes on diff --git a/nisse-dev/src/nisse_dev/config.py b/nisse-dev/src/nisse_dev/config.py index cf52e59..62a3173 100644 --- a/nisse-dev/src/nisse_dev/config.py +++ b/nisse-dev/src/nisse_dev/config.py @@ -137,6 +137,15 @@ class LoopTuning: general_idle_s: float = 86400.0 +@dataclass(frozen=True) +class BuildVMTuning: + # An unreachable build VM is tried again this often (#137). + retry_s: float = 300.0 + # Once it answers, it must still answer this much later before work + # goes on, so it has finished booting. + settle_s: float = 180.0 + + @dataclass(frozen=True) class Tuning: driver: DriverTuning = field(default_factory=DriverTuning) @@ -146,6 +155,7 @@ class Tuning: tools: ToolsTuning = field(default_factory=ToolsTuning) context: ContextTuning = field(default_factory=ContextTuning) loop: LoopTuning = field(default_factory=LoopTuning) + build_vm: BuildVMTuning = field(default_factory=BuildVMTuning) _TUNING_SECTIONS: dict[str, type] = { diff --git a/nisse-dev/src/nisse_dev/loop/agent.py b/nisse-dev/src/nisse_dev/loop/agent.py index 63b1a06..cf44f8e 100644 --- a/nisse-dev/src/nisse_dev/loop/agent.py +++ b/nisse-dev/src/nisse_dev/loop/agent.py @@ -56,7 +56,13 @@ from nisse_dev.llm import LLMError, RoleMap from nisse_dev.loop import preflight from nisse_dev.loop.actions import classify, commits from nisse_dev.loop.classifier import IN_SCOPE, ActionClassifier, Scope -from nisse_dev.loop.driver import ASK_ALLOW, Driver, RunOutcome, Screening +from nisse_dev.loop.driver import ( + ASK_ALLOW, + Driver, + RunOutcome, + Screening, + VMUnreachable, +) from nisse_dev.loop.help import ASK_EXPERT, DELEGATE, Helpers from nisse_dev.loop.items import ( ANSWER, @@ -112,7 +118,7 @@ _COMMITTED = ( ) # The operator's word that a blocked build VM is ready again (D3). _VM_READY = re.compile(r"vm ready(?: ([a-z0-9][a-z0-9-]*))?") -_NEEDS_ITEM = frozenset({"go", "approve", "abandon", "allow", "trust new host key"}) +_NEEDS_ITEM = frozenset({"go", "approve", "abandon", "allow"}) # Near-misses for abandon: asked about, never taken for it or for a reply. _NEAR_ABANDON = frozenset( { @@ -129,7 +135,19 @@ _NEAR_ABANDON = frozenset( BUG_GRACE_S = 600.0 GENERAL_IDLE_S = 86400.0 _SAVED = "The work in progress is pushed; anything else on it will be lost." -BROKEN = "broken" # a VM block's cause: the driver found it broken (D3) +# A build VM block's causes (D3): the driver found it broken (M4.3); it +# can't be reached, and is tried again; it answered with a new host key. +BROKEN = "broken" +UNREACHABLE = "unreachable" +HOST_KEY = "host key" +# How each reads in a sentence about the VM. +_BLOCKED_BY = { + BROKEN: "needs reimaging", + UNREACHABLE: "is out of reach", + HOST_KEY: "has a new host key", +} +VM_RETRY_S = 300.0 +VM_SETTLE_S = 180.0 PROGRESS_EVERY_S = 60.0 # progress in prose at a readable rate (L5, #42) PROGRESS_CHARS = 400 DETAILS = 10 # tool calls shown by `details` @@ -213,6 +231,9 @@ The operator has asked a question (the last message). Answer it with reply, from what you know and what you can look at; don't change anything. The item then goes back to waiting exactly as it was, so a restatement up stays up.""" +OFFLINE = """ The build VM is out of reach just now, so you can't look at +anything: answer from what you know, and say what you'd check.""" + CONVERSATION = """You are nisse-dev, a software agent, in a conversation with the operator in a Matrix thread. It isn't about a project yet: it may be a question, thoughts on feasibility or hardware, or the start of some work. @@ -263,6 +284,11 @@ class Project: merge: str = "operator" # the trust dial (L3): who merges its PRs +def _minutes(seconds: float) -> str: + m = max(1, round(seconds / 60)) + return f"{m} minute{'s' if m != 1 else ''}" + + def _restatement_problem(args: dict[str, object]) -> str | None: """L5, D1: a restatement is agreed to as it stands, so it asks nothing. One that ends in a question is sent back: questions go by ask_human.""" @@ -313,6 +339,8 @@ class LoopAgent: markers: str = preflight.MARKERS, clock: Callable[[], float] = time.time, intake_interval_s: float = INTAKE_INTERVAL_S, + vm_retry_s: float = VM_RETRY_S, + vm_settle_s: float = VM_SETTLE_S, ) -> None: self._store = store self._transport = transport @@ -348,6 +376,11 @@ class LoopAgent: self._judged: dict[tuple[str, int, str], tuple[Verdict, tuple[str, str]]] = {} self._clock = clock self._interval = intake_interval_s + # Unreachable build VMs (#137): when each is next tried, and since + # when it has answered. Volatile: after a boot, tried at once. + self._vm_retry = vm_retry_s + self._vm_settle = vm_settle_s + self._probes: dict[str, tuple[float, float | None]] = {} # thread root -> revision on screen there, read on the transport's # thread to stamp replies on arrival (L3); (project, issue) -> thread # root, for a command that names its item. @@ -471,21 +504,12 @@ class LoopAgent: root = str(project.workspace.root) try: ready = preflight.check(host, project.name, root, self._markers) - except HostKeyChanged: - note = ( - f"the build VM for {project.name} ({host.describe()}) answered " - "with a different host key than the one I pinned. If you " - "rebuilt it, say `trust new host key` and I'll set it up " - "again; if you didn't, something is impersonating it." - ) - self._wait( - item, replace(loop, wait=ANSWER, note=note, host_key_changed=True), note - ) + except HostKeyChanged as exc: + self._block_vm(host.describe(), HOST_KEY, str(exc), item) self._store.set_active(None, cause="host key changed") return None except HostUnavailable as exc: - note = f"I can't reach the build VM: {exc}. Reply to retry." - self._wait(item, replace(loop, wait=ANSWER, note=note), note) + self._block_vm(host.describe(), UNREACHABLE, str(exc), item) self._store.set_active(None, cause="build VM unreachable") return None if ready.missing: @@ -540,8 +564,9 @@ class LoopAgent: ) try: now = preflight.versions(host, before) - except HostUnavailable: - return loop # the run will find out, and say so + except HostUnavailable as exc: + self._settle(item.id, RunOutcome("unreachable", note=str(exc))) + return None changed = preflight.drift(before, now) if changed: said = "the toolchain changed since it was checked: " + "; ".join(changed) @@ -590,9 +615,13 @@ class LoopAgent: if requested else "toolchain step: setting the build VM up from the repository", ) - before = self._changed_files(project) - roles = self._roles_of(item) outer = self._running # a requested step runs inside the driver's run + try: + before = self._changed_files(project) + except HostUnavailable as exc: + self._lost_in_step(item, outer, exc) + return None + roles = self._roles_of(item) if not outer: self._active_thread = loop.thread self._active_name = self._label(item) @@ -620,6 +649,11 @@ class LoopAgent: self._cancel.clear() self._settle(item.id, outcome) return None + if outcome.kind == "unreachable": + if outer: + raise VMUnreachable(outcome.note) # ends the driver's run too + self._settle(item.id, outcome) + return None host = workspace.host now = loop_of(self._store.work_item(item.id) or item) if outcome.kind == "wait" and outcome.tool == "toolchain_ready": @@ -646,8 +680,10 @@ class LoopAgent: "at": datetime.now(UTC).isoformat(timespec="seconds"), }, ) - except HostUnavailable: - tools = report.tools + except HostUnavailable as exc: + # #137: no marker, so no work: the VM is blocked instead. + self._lost_in_step(item, outer, exc) + return None missing = [f"{name} ({why})" for name, why in report.missing] parts = [f"toolchain: {preflight.said(tools)}"] if installed: @@ -658,7 +694,11 @@ class LoopAgent: parts.append(report.summary) if not requested: # D3: the harness's step edits nothing in the repository. - edited = sorted(set(self._changed_files(project)) - set(before)) + try: + edited = sorted(set(self._changed_files(project)) - set(before)) + except HostUnavailable as exc: + self._lost_in_step(item, outer, exc) + return None if edited: parts.append( "it changed files it shouldn't have, left for the work: " @@ -671,16 +711,20 @@ class LoopAgent: loop = replace(now, toolchain=tuple(tuple(t.to_data()) for t in tools)) return loop, said + def _lost_in_step(self, item: WorkItem, outer: bool, exc: Exception) -> None: + """A toolchain step lost its VM (#137): a requested one ends the + driver's run with it; the harness's own blocks the VM, settled.""" + if outer: + raise VMUnreachable(str(exc)) from exc + self._settle(item.id, RunOutcome("unreachable", note=str(exc))) + def _changed_files(self, project: Project) -> list[str]: """The workspace's changes, as git sees them (for the setup step's promise to edit nothing).""" assert project.workspace is not None - try: - result = project.workspace.host.execute( - "git status --porcelain", cwd=str(project.workspace.root), timeout_s=30 - ) - except HostUnavailable: - return [] + result = project.workspace.host.execute( + "git status --porcelain", cwd=str(project.workspace.root), timeout_s=30 + ) # HostUnavailable: the step's caller handles a lost VM (#137) return [line[3:] for line in result.stdout.splitlines() if line.strip()] def _requested_step(self, item_id: str, reason: str) -> tuple[str, bool]: @@ -943,6 +987,7 @@ class LoopAgent: for project in self._projects.values(): self._intake(project) self.watch() + self.check_vms() self.check_blocked() self.check_grace() self.check_idle() @@ -1042,6 +1087,10 @@ class LoopAgent: and self._at(item).merge == "approved" ): pr, loop = self._merge_if_approved(item, loop, repo, pr) + if pr.state != "open" and loop.back_to == MERGE: + # A remark on it waited to be answered (a later run): the + # merge it would go back to waiting for is over. + loop = replace(loop, back_to="", back_note="") if pr.merged: self._fact(item, f"PR #{loop.pr} merged") loop, position = replace(loop, pr=0, pr_url="", branch=""), READY @@ -1392,27 +1441,6 @@ class LoopAgent: ), cause="the operator asked a question", ) - elif command == "trust new host key": - project = self._at(item) - if loop.host_key_changed and project.workspace is not None: - project.workspace.host.forget_host_key() - self._fact( - item, - f"the operator ({msg.sender}) trusted the build VM's new " - "host key: it will be set up again", - ) - save( - self._store, - item.id, - READY, - replace(loop, wait="", note="", host_key_changed=False), - cause=f"new host key trusted by {msg.sender}", - ) - else: - self._transport.post( - f"Nothing on {self._label(item)} is waiting for that.", - thread=loop.thread, - ) elif command == "allow": if item.position == WAITING and loop.pending_action: action = json.loads(loop.pending_action) @@ -1838,7 +1866,9 @@ class LoopAgent: ready = [ i for i in items - if i.position == READY and self._host_of(i.scope) not in blocks + if i.position == READY + # L5: a question is answered even while its VM is blocked. + and (self._host_of(i.scope) not in blocks or loop_of(i).question) ] return sorted(ready, key=lambda i: -loop_of(i).first) @@ -1873,8 +1903,11 @@ class LoopAgent: what = f"blocked by {', '.join(loop.blocked_by)}" elif item.position == READY and item.id not in queue: block = self._store.vm_blocks().get(self._host_of(item.scope) or "") - why = block.reason if block else "unknown" - what = f"blocked: its build VM needs recovery ({why})" + what = ( + f"blocked: its build VM {_BLOCKED_BY[block.cause]} ({block.reason})" + if block + else "blocked: its build VM is blocked" + ) elif item.position == READY: what = f"ready, {queue.index(item.id) + 1} of {len(queue)} in line" name = ( @@ -2415,13 +2448,20 @@ class LoopAgent: checked = self._preflight and project.workspace is not None marker: dict[str, Any] | None = None starting = not loop.prepared - if checked and milestones is not None: + # L5: a question doesn't wait for a blocked build VM: it is answered + # without one (#137). + offline = loop.question and self._vm_blocked(project.name) + if checked and milestones is not None and not offline: ready = self._readiness(item, project, loop) - if ready is None: - return # waiting on the operator - loop, marker = ready - starting = not loop.prepared - if milestones is not None: + if ready is None and loop.question and self._vm_blocked(project.name): + offline = True + self._store.set_active(item.id, cause="a question, without the VM") + elif ready is None: + return # waiting on the operator, or on the VM + else: + loop, marker = ready + starting = not loop.prepared + if milestones is not None and not offline: keep = self._keep(project.name) try: if not loop.prepared: @@ -2434,6 +2474,12 @@ class LoopAgent: loop.branch or f"{item.id}-work", keep=keep, mine=f"{item.id}-" ) except Exception as exc: # noqa: BLE001 -- the item waits, not the loop + host = self._host_of(project.name) + if isinstance(exc.__cause__, HostUnavailable) and host is not None: + # #137: the VM went between the check and this. + self._block_vm(host, UNREACHABLE, str(exc.__cause__), item) + self._store.set_active(None, cause="build VM unreachable") + return note = f"I couldn't prepare the workspace: {exc}. Reply to retry." self._wait(item, replace(loop, wait=ANSWER, note=note), note) self._store.set_active(None, cause="workspace failed") @@ -2446,10 +2492,10 @@ class LoopAgent: self._transport.post(said, thread=loop.thread) loop = replace(loop, prepared=True) save(self._store, item.id, RUNNING, loop, cause="driver run") - if checked and milestones is not None: + if checked and milestones is not None and not offline: toolchain = self._toolchain(item, project, loop, marker, starting) if toolchain is None: - return # preflight was stopped: settled + return # stopped, or the VM went: settled loop = toolchain save(self._store, item.id, RUNNING, loop, cause="toolchain checked") # D1: a `discuss` item talks before it restates. @@ -2478,23 +2524,27 @@ class LoopAgent: ), ] if loop.question: - tools = [REPLY, *_LOOK] + tools = [REPLY] if offline else [REPLY, *_LOOK] elif loop.back_to and not loop.started: tools = [*tools, REPLY] # before the go-ahead, a remark answered roles = self._roles_of(item) - workspace = project.workspace - system = SYSTEM.format( - repo=project.repo, - issue=loop.issue, - title=item.title, - labels="".join(f"{LABELS[x]}\n\n" for x in loop.labels if x in LABELS), - helpers=( - f"{self._helpers.briefing(roles)}\n\n" - if self._helpers and loop.started - else "" - ), - workspace=workspace.describe() if workspace else "", - ) + (QUESTION if loop.question else "") + workspace = None if offline else project.workspace + system = ( + SYSTEM.format( + repo=project.repo, + issue=loop.issue, + title=item.title, + labels="".join(f"{LABELS[x]}\n\n" for x in loop.labels if x in LABELS), + helpers=( + f"{self._helpers.briefing(roles)}\n\n" + if self._helpers and loop.started + else "" + ), + workspace=workspace.describe() if workspace else "", + ) + + (QUESTION if loop.question else "") + + (OFFLINE if offline else "") + ) self._active_thread = loop.thread self._active_name = self._label(item) self._running = True @@ -2728,6 +2778,13 @@ class LoopAgent: summary = str(args.get("summary", "")).strip() self._transport.post(f"Done: {summary}", thread=loop.thread) self._close(item, loop, "completed", summary) + elif outcome.kind == "unreachable": + # #137: not the work's fault. The item goes on when the VM is back. + save(self._store, item.id, READY, loop, cause="build VM unreachable") + project = self._at(item) + host = self._host_of(project.name) + if host is not None: + self._block_vm(host, UNREACHABLE, outcome.note, item) elif outcome.kind == "stopped": note = f"stopped ({outcome.note}). Why did you stop it? Reply to continue." self._wait(item, replace(loop, wait=ANSWER, note=note), note) @@ -2772,36 +2829,138 @@ class LoopAgent: self._block_vm(host, BROKEN, reason, item, risk) def _block_vm( - self, host: str, cause: str, reason: str, by: WorkItem, risk: str = "" + self, + host: str, + cause: str, + reason: str, + by: WorkItem | None, + risk: str = "", ) -> None: """Block a build VM (D3): every project on it waits until it is recovered. Said once at top level, and in each waiting item's thread. - The recovery is the operator's for now; nisse-ops' later (M7.6).""" - self._store.block_vm(host, cause=cause, reason=reason, work_item=by.id) - names = sorted(p for p in self._projects if self._host_of(p) == host) - blocked = [ - i - for i in self._store.work_items("open") - if i.scope in names and i.scope != GENERAL - ] - for other in blocked: - line = ( - f"Blocked: the build VM ({host}) needs reimaging. " - f"Waiting for `vm ready {other.scope}`." + The recovery is the operator's for now, or the VM's coming back; + nisse-ops' later (M7.6).""" + if host in self._store.vm_blocks(): + return # already said + self._store.block_vm( + host, cause=cause, reason=reason, work_item=by.id if by else None + ) + self._probes.pop(host, None) + names = self._on_host(host) + for other in self._store.work_items("open"): + if other.scope not in names: + continue + line = f"Blocked: the build VM ({host}) {_BLOCKED_BY[cause]}. " + ( + "I'll go on once it's back." + if cause == UNREACHABLE + else f"Waiting for `vm ready {other.scope}`." ) self._fact(other, line) self._transport.post(line, thread=loop_of(other).thread) - self._transport.post( - f"The build VM {host} seems broken, found working on " - f"{self._label(by)}: {reason}\n\nThe projects on it are blocked " - f"until it's fixed: {', '.join(names)}. Could you reimage it from " - f"its template? {risk or _SAVED} If its " + self._transport.post(self._blocked_message(host, cause, reason, by, risk)) + + def _blocked_message( + self, host: str, cause: str, reason: str, by: WorkItem | None, risk: str + ) -> str: + names = self._on_host(host) + ready = " or ".join(f"`vm ready {n}`" for n in names) + found = f", found working on {self._label(by)}" if by else "" + blocked = f"The projects on it are blocked: {', '.join(names)}." + if cause == UNREACHABLE: + return ( + f"I can't reach the build VM {host}{found}: {reason}\n\n" + f"{blocked} I'll try again every {_minutes(self._vm_retry)} and " + f"go on once it has answered for {_minutes(self._vm_settle)}. If " + f"it's gone for good, reimage it and say {ready}; anything not " + "pushed on it will be lost." + ) + if cause == HOST_KEY: + return ( + f"The build VM {host} answered with a different host key than " + f"the one I pinned{found}. {blocked}\n\nIf you rebuilt it, say " + f"{ready}: that trusts its new key, and I'll set it up again. " + "If you didn't, something may be impersonating it, so don't." + ) + return ( + f"The build VM {host} seems broken{found}: {reason}\n\n{blocked} " + f"Could you reimage it from its template? {risk or _SAVED} If its " "address changes, update `build_host` in my configuration and " - "restart me first.\n\nOnce it's ready, say " - + " or ".join(f"`vm ready {n}`" for n in names) - + "; that also trusts its new host key." + f"restart me first.\n\nOnce it's ready, say {ready}; that also " + "trusts its new host key." ) + def _vm_blocked(self, project: str) -> bool: + return self._host_of(project) in self._store.vm_blocks() + + def _on_host(self, host: str) -> list[str]: + return sorted(p for p in self._projects if self._host_of(p) == host) + + def _park_on(self, host: str) -> None: + """A run cut off by its VM going away couldn't park its work: park + each project's workspace on the VM before anything else is taken up + there, which would set that work aside (#137). Raises when the VM + can't be reached.""" + for name in self._on_host(host): + milestones = self._projects[name].milestones + if milestones is not None: + milestones.park(keep=self._keep(name)) + + def _unblock_vm(self, host: str, cause: str, said: str) -> None: + self._store.unblock_vm(host, cause=cause) + self._probes.pop(host, None) + for other in self._store.work_items("open"): + if other.scope in self._on_host(host): + self._fact(other, said) + + def check_vms(self) -> None: + """#137: each unreachable build VM, tried again when due. Once it + answers, it must still answer a settle period later before work goes + on (SSH is up before the VM has finished booting). A new host key + is the operator's to trust; a broken VM, theirs to reimage.""" + now = self._clock() + for host, block in self._store.vm_blocks().items(): + if block.cause != UNREACHABLE: + continue + due, answered = self._probes.get(host, (now, None)) + if now < due: + continue + names = self._on_host(host) + if not names: + continue # no project on it any more + workspace = self._projects[names[0]].workspace + assert workspace is not None + try: + workspace.host.execute("true", timeout_s=30) + except HostKeyChanged as exc: + self._store.reblock_vm(host, cause=HOST_KEY, reason=str(exc)) + self._probes.pop(host, None) + self._transport.post( + self._blocked_message(host, HOST_KEY, str(exc), None, "") + ) + continue + except HostUnavailable as exc: + self._oplog.emit("vm.unreachable", cause=str(exc), host=host) + self._probes[host] = (now + self._vm_retry, None) + continue + if answered is None or now - answered < self._vm_settle: + since = now if answered is None else answered + self._probes[host] = (since + self._vm_settle, since) + continue + try: + self._park_on(host) + except Exception as exc: # noqa: BLE001 -- not back after all + self._oplog.emit("vm.unreachable", cause=repr(exc), host=host) + self._probes[host] = (now + self._vm_retry, None) + continue + self._unblock_vm( + host, + "it answered again", + f"the build VM ({host}) is back: work goes on", + ) + self._transport.post( + f"The build VM {host} is back: {', '.join(names)} can go on." + ) + def _vm_ready(self, name: str | None, sender: str) -> str: """The operator's `vm ready`: the build VM's block lifts and its new host key is trusted (D3). A reimaged VM has no marker, so its first @@ -2824,18 +2983,17 @@ class LoopAgent: if host is None or workspace is None or host not in blocks: return f"Nothing is blocking {name}'s build VM." workspace.host.forget_host_key() - block = self._store.unblock_vm(host, cause=f"vm ready from {sender}") - names = sorted(p for p in self._projects if self._host_of(p) == host) - for other in self._store.work_items("open"): - if other.scope in names: - self._fact( - other, - f"the operator ({sender}) said the build VM ({host}) is " - "ready: its new host key is trusted, and it is set up again " - "if it was reimaged", - ) - assert block is not None # checked above, on the worker - return f"{host} is unblocked: {', '.join(names)} can go on." + try: + self._park_on(host) + except Exception as exc: # noqa: BLE001 -- reimaged: nothing left there + self._oplog.emit("vm.park_failed", cause=repr(exc), host=host) + self._unblock_vm( + host, + f"vm ready from {sender}", + f"the operator ({sender}) said the build VM ({host}) is ready: its " + "new host key is trusted, and it is set up again if it was reimaged", + ) + return f"{host} is unblocked: {', '.join(self._on_host(host))} can go on." # -- milestones (D4) ----------------------------------------------------- diff --git a/nisse-dev/src/nisse_dev/loop/build.py b/nisse-dev/src/nisse_dev/loop/build.py index c01696a..fdd937f 100644 --- a/nisse-dev/src/nisse_dev/loop/build.py +++ b/nisse-dev/src/nisse_dev/loop/build.py @@ -331,6 +331,8 @@ def build_loop_agent( preflight_timeout_s=config.tuning.preflight.command_timeout_s, bug_grace_s=config.tuning.loop.bug_grace_s, general_idle_s=config.tuning.loop.general_idle_s, + vm_retry_s=config.tuning.build_vm.retry_s, + vm_settle_s=config.tuning.build_vm.settle_s, ) return RunningLoop( config=config, diff --git a/nisse-dev/src/nisse_dev/loop/driver.py b/nisse-dev/src/nisse_dev/loop/driver.py index 82ae6d5..b9ebc72 100644 --- a/nisse-dev/src/nisse_dev/loop/driver.py +++ b/nisse-dev/src/nisse_dev/loop/driver.py @@ -29,13 +29,19 @@ _PROSE_NUDGE = ( @dataclass(frozen=True) class RunOutcome: - # stuck: failed commands in a row hit the failure limit (L9's trigger) - kind: Literal["wait", "capped", "stuck", "stopped", "failed"] + # stuck: failed commands in a row hit the failure limit (L9's trigger); + # unreachable: the build VM went out of reach (#137), noted in ``note`` + kind: Literal["wait", "capped", "stuck", "stopped", "failed", "unreachable"] tool: str = "" # the waiting tool called arguments: dict[str, Any] | None = None note: str = "" +class VMUnreachable(Exception): + """Raised by a handler whose work found the build VM out of reach: the + run ends there, as for an execution call (#137).""" + + # The waiting outcome a dangerous action ends a run with (P3): the harness # puts it to the operator, rendered from the action itself. ASK_ALLOW = "ask_allow" @@ -157,7 +163,10 @@ class Driver: ) return RunOutcome("wait", call.tool, dict(call.arguments)) if call.tool in handlers and call.tool in offered: - result, ok = handlers[call.tool](dict(call.arguments)) + try: + result, ok = handlers[call.tool](dict(call.arguments)) + except VMUnreachable as exc: + return self._unreachable(transcript, calls, index, str(exc)) elif call.tool not in offered or workspace is None: result, ok = ( f"no such tool: {call.tool}; the tools are " @@ -200,6 +209,12 @@ class Driver: ) break outcome = workspace.dispatch(call.tool, call.arguments) + if outcome.unreachable: + # Not the work failing: no model call works on a VM + # that isn't there (P6). + return self._unreachable( + transcript, calls, index, outcome.result + ) result, ok = outcome.result, outcome.success if screening.note: result = f"{result}\n\n[harness] {screening.note}" @@ -219,6 +234,20 @@ class Driver: ) return RunOutcome("capped", note=f"hit the {cap}-turn cap") + @staticmethod + def _unreachable( + transcript: Transcript, calls: list[ToolCall], index: int, said: str + ) -> RunOutcome: + transcript.append( + ToolResult( + call_id=calls[index].call_id, + result=f"not finished: {said}; your turn has ended", + success=False, + ) + ) + skip_calls(transcript, calls[index + 1 :], "not run: the turn had ended") + return RunOutcome("unreachable", note=said) + def _brief(call: ToolCall, result: str) -> str: """A failed command and the end of what it said, for a helper's brief.""" diff --git a/nisse-dev/src/nisse_dev/loop/items.py b/nisse-dev/src/nisse_dev/loop/items.py index 162ce60..c93b3e3 100644 --- a/nisse-dev/src/nisse_dev/loop/items.py +++ b/nisse-dev/src/nisse_dev/loop/items.py @@ -78,10 +78,8 @@ class Loop: # it's said once per commit and reason. merge_held: str = "" # The toolchain as the harness last checked it (M4.1): name, version - # command, version; and a build VM whose host key changed, waiting on - # the operator's word. + # command, version. toolchain: tuple[tuple[str, str, str], ...] = () - host_key_changed: bool = False def to_data(self) -> dict[str, Any]: return asdict(self) @@ -89,6 +87,8 @@ class Loop: def loop_of(item: WorkItem) -> Loop: data = dict(item.data.get("loop", {})) + # Retired: a changed host key now blocks the build VM (#137). + data.pop("host_key_changed", None) data["notes"] = tuple(data.get("notes", ())) data["blocked_by"] = tuple(data.get("blocked_by", ())) data["labels"] = tuple(data.get("labels", ())) diff --git a/nisse-dev/src/nisse_dev/state.py b/nisse-dev/src/nisse_dev/state.py index 4a68897..e588d04 100644 --- a/nisse-dev/src/nisse_dev/state.py +++ b/nisse-dev/src/nisse_dev/state.py @@ -98,7 +98,7 @@ class VMBlock: project on it waits.""" host: str - cause: str # broken (the driver said so); unreachable (#137) + cause: str # broken (the driver said so), unreachable, host key (#137) reason: str work_item: str | None # the item that found it since: str @@ -411,6 +411,15 @@ class StateStore: if row is not None: self.oplog.emit("vm.block", cause=cause, host=host, work_item=work_item) + def reblock_vm(self, host: str, *, cause: str, reason: str) -> None: + """A blocked VM's block, for another cause (an unreachable VM that + answers with a new host key).""" + self._conn.execute( + "UPDATE vm_blocks SET cause = ?, reason = ? WHERE host = ?", + (cause, reason, host), + ) + self.oplog.emit("vm.block", cause=cause, host=host) + 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( diff --git a/nisse-dev/src/nisse_dev/tools.py b/nisse-dev/src/nisse_dev/tools.py index 1ec8a68..acce705 100644 --- a/nisse-dev/src/nisse_dev/tools.py +++ b/nisse-dev/src/nisse_dev/tools.py @@ -93,6 +93,7 @@ EXECUTION_TOOLS: list[dict[str, Any]] = [ class ToolOutcome: result: str success: bool + unreachable: bool = False # the build VM couldn't be reached (#137) class Workspace: @@ -118,7 +119,12 @@ class Workspace: # The project secrets the model's commands may see (D7); redacted from # whatever comes back. self._secrets = dict(secrets or {}) - self._run(f"mkdir -p -- {shlex.quote(str(self.root))}") + try: + self._run(f"mkdir -p -- {shlex.quote(str(self.root))}") + except HostUnavailable: + # A VM down at startup mustn't stop the agent: the check before + # each run finds it, and blocks it (#137); cloning makes the root. + pass def with_timeout(self, timeout_s: int) -> "Workspace": """The same workspace with a longer limit per command (installs).""" @@ -173,7 +179,9 @@ class Workspace: return ToolOutcome(f"missing argument {exc} for {tool}", success=False) except HostUnavailable as exc: # Not the work failing: the build VM is out of reach. - return ToolOutcome(f"the build VM is unreachable: {exc}", success=False) + return ToolOutcome( + f"the build VM is unreachable: {exc}", success=False, unreachable=True + ) def read_file(self, path: str) -> ToolOutcome: target = self._resolve(path) @@ -244,7 +252,9 @@ class Workspace: timeout_s=self._timeout_s, ) except HostUnavailable as exc: - return ToolOutcome(f"the build VM is unreachable: {exc}", success=False) + return ToolOutcome( + f"the build VM is unreachable: {exc}", success=False, unreachable=True + ) if result.timed_out: return ToolOutcome( f"command timed out after {self._timeout_s}s: {command}", diff --git a/nisse-dev/tests/test_config.py b/nisse-dev/tests/test_config.py index b414ae2..030ef66 100644 --- a/nisse-dev/tests/test_config.py +++ b/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: diff --git a/nisse-dev/tests/test_loop.py b/nisse-dev/tests/test_loop.py index 6c844dc..5fa73f2 100644 --- a/nisse-dev/tests/test_loop.py +++ b/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)] diff --git a/nisse-dev/tests/test_loop_live_fixes.py b/nisse-dev/tests/test_loop_live_fixes.py index 19bce0b..af63356 100644 --- a/nisse-dev/tests/test_loop_live_fixes.py +++ b/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 ---------------------------------------------- diff --git a/nisse-dev/tests/test_loop_properties.py b/nisse-dev/tests/test_loop_properties.py index 217fe2f..d101f38 100644 --- a/nisse-dev/tests/test_loop_properties.py +++ b/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() diff --git a/nisse-dev/tests/test_loop_toolchain.py b/nisse-dev/tests/test_loop_toolchain.py index d20b907..4f4bc66 100644 --- a/nisse-dev/tests/test_loop_toolchain.py +++ b/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))}"') diff --git a/nisse-dev/tests/test_loop_vm.py b/nisse-dev/tests/test_loop_vm.py new file mode 100644 index 0000000..0dc6665 --- /dev/null +++ b/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"] diff --git a/specs/loop.qnt b/specs/loop.qnt index 825869d..892fc6e 100644 --- a/specs/loop.qnt +++ b/specs/loop.qnt @@ -63,6 +63,9 @@ module loop { inbox: List[Msg], // -- durable: the state store's VM blocks (D3) -- vmBlocked: Set[str], // projects whose build VM waits for recovery + vmUnreachable: Set[str], // ...because it was out of reach: probes may clear it (#137) + // -- the world: build VMs out of reach -- + vmDown: Set[str], // -- volatile -- work: str, // none / publish pubStage: int, @@ -73,6 +76,7 @@ module loop { lostFix: bool, // a review fix pushed to a PR no longer open lostWork: bool, // a parked item's unpushed work was set aside or wiped ranBlocked: bool, // an item was taken up on a blocked build VM + badUnblock: bool, // a probe unblocked a VM that was still down } var s: Sys @@ -102,9 +106,10 @@ module loop { issues: ISSUES.mapBy(_ => { open: false, assigned: false, linked: 0 }), depOpen: DEP_ISSUES.mapBy(_ => true), prs: Map(), opened: Map(), inbox: List(), vmBlocked: Set(), + vmUnreachable: Set(), vmDown: Set(), work: "none", pubStage: 0, pending: 0, badGo: false, earlyMs: false, lostFix: false, lostWork: false, - ranBlocked: false, + ranBlocked: false, badUnblock: false, } // ---- the operator ---------------------------------------------------------- @@ -237,15 +242,21 @@ module loop { } // Taking up a ready item: a fresh workspace on its project's build VM (D3) - // sets aside whatever is left there. + // sets aside whatever is left there. The check before it finds a VM out + // of reach, and blocks it instead (#137). action activate(i: int): bool = all { s.active == 0, s.work == "none", live(s).contains(i), itm(s, i).status == "ready", not(s.vmBlocked.contains(itm(s, i).project)), // D3: nothing runs on a blocked VM s' = { - val leftovers = s.items.keys().exists(j => j != i and itm(s, j).project == itm(s, i).project + val p = itm(s, i).project + val leftovers = s.items.keys().exists(j => j != i and itm(s, j).project == p and itm(s, j).unpushed) - { ...s, active: i, lostWork: s.lostWork or leftovers, - ranBlocked: s.ranBlocked or s.vmBlocked.contains(itm(s, i).project) } + if (s.vmDown.contains(p)) + { ...s, vmBlocked: s.vmBlocked.union(Set(p)), + vmUnreachable: s.vmUnreachable.union(Set(p)) } + else + { ...s, active: i, lostWork: s.lostWork or leftovers, + ranBlocked: s.ranBlocked or s.vmBlocked.contains(p) } }, } @@ -389,13 +400,15 @@ module loop { } // The build VM is wiped (D3): unpushed work on it is lost, accepted for - // the active item, never for a parked one. + // the active item, never for a parked one. Nor for an item cut off by its + // VM going out of reach (#137): that work couldn't be pushed, and the + // operator was told a reimage loses it. action wipe(p: str): bool = all { s' = { ...s, items: s.items.keys().mapBy(j => if (itm(s, j).project == p) { ...itm(s, j), unpushed: false } else itm(s, j)), - lostWork: s.lostWork or s.items.keys().exists(j => - j != s.active and itm(s, j).project == p and itm(s, j).unpushed) }, + lostWork: s.lostWork or (not(s.vmUnreachable.contains(p)) and s.items.keys().exists(j => + j != s.active and itm(s, j).project == p and itm(s, j).unpushed)) }, } // The driver finds its build VM broken (request_reimage, D3): its work is @@ -411,15 +424,47 @@ module loop { } } - // The operator's `vm ready`: the block lifts. A reimage loses what wasn't - // pushed on the VM; the operator was told so, and agreed by doing it. - action vmReady(p: str, reimaged: bool): bool = all { + // The build VM goes out of reach, or comes back (the world, #137). + action goDown(p: str): bool = s' = { ...s, vmDown: s.vmDown.union(Set(p)) } + action comeBack(p: str): bool = s' = { ...s, vmDown: s.vmDown.exclude(Set(p)) } + + // A run finds its VM gone: it ends there, the item stays ready, and the + // VM is blocked. Its work stays on the VM, which isn't wiped. + action lostMidRun: bool = { + val it = itm(s, s.active) + all { + s.active != 0, s.work == "none", s.intents.size() == 0, it.status == "ready", + s.vmDown.contains(it.project), + s' = { ...s, active: 0, + vmBlocked: s.vmBlocked.union(Set(it.project)), + vmUnreachable: s.vmUnreachable.union(Set(it.project)) }, + } + } + + // A background probe (#137): a VM blocked for being out of reach is + // unblocked once it answers (and has settled, which the model leaves out). + // First the work a lost run left on it is parked (pushed). + action probe(p: str): bool = all { + s.vmUnreachable.contains(p), + not(s.vmDown.contains(p)), // it answered + s' = { ...s, + items: s.items.keys().mapBy(j => + if (itm(s, j).project == p) { ...itm(s, j), unpushed: false } else itm(s, j)), + vmBlocked: s.vmBlocked.exclude(Set(p)), + vmUnreachable: s.vmUnreachable.exclude(Set(p)), + badUnblock: s.badUnblock or s.vmDown.contains(p) }, + } + + // The operator's `vm ready`: the block lifts, after parking what a lost + // run left there. A reimage loses what wasn't pushed on the VM; the + // operator was told so, and agreed by doing it. + action vmReady(p: str): bool = all { s.vmBlocked.contains(p), s' = { ...s, vmBlocked: s.vmBlocked.exclude(Set(p)), - items: if (reimaged) s.items.keys().mapBy(j => - if (itm(s, j).project == p) { ...itm(s, j), unpushed: false } else itm(s, j)) - else s.items }, + vmUnreachable: s.vmUnreachable.exclude(Set(p)), + items: s.items.keys().mapBy(j => + if (itm(s, j).project == p) { ...itm(s, j), unpushed: false } else itm(s, j)) }, } // ---- crash and boot (L8) -------------------------------------------------- @@ -457,7 +502,8 @@ module loop { intake(anyIssue, p, b), observe(i), handleMsg, park, activate(i), restate(b), graceOver(i), createRecord, linkRecord, startMilestone, turn(outcome), finishMilestone, publishStep(b), answerReview, - finish, wipe(p), crash, requestReimage, vmReady(p, b), + finish, wipe(p), crash, requestReimage, vmReady(p), + goDown(p), comeBack(p), lostMidRun, probe(p), } } @@ -506,10 +552,13 @@ module loop { // Nothing is taken up on a build VM that is blocked (D3). val blocked_vm_runs_nothing = not(s.ranBlocked) + // A VM blocked for being out of reach is unblocked only once it answers. + val unblocked_only_when_up = not(s.badUnblock) + val all_properties = and { one_item_per_issue, waits_are_explainable, go_binds, milestone_after_merge, parked_work_safe, one_pr_per_milestone, review_fixes_land, closed_items_leave_nothing, - blocked_vm_runs_nothing, + blocked_vm_runs_nothing, unblocked_only_when_up, } // ---- reachability witnesses (each should be violated) -------------------------- @@ -520,5 +569,6 @@ module loop { val w_review = not(s.items.keys().exists(i => itm(s, i).review)) val w_record = not(s.items.keys().exists(i => itm(s, i).src == "conv" and itm(s, i).issue != 0)) val w_vm_blocked = s.vmBlocked.size() == 0 + val w_vm_unreachable = s.vmUnreachable.size() == 0 val w_finished = not(s.items.keys().exists(i => itm(s, i).status == "closed" and itm(s, i).ms > 0)) } diff --git a/specs/loop_scenarios.qnt b/specs/loop_scenarios.qnt index d031526..899c4b1 100644 --- a/specs/loop_scenarios.qnt +++ b/specs/loop_scenarios.qnt @@ -82,7 +82,19 @@ module loop_scenarios { .then(assignIssue(1)).then(intake(1, "A", false)) .then(activate(1)).then(requestReimage) .expect(s.vmBlocked.contains("A") and s.active == 0) - .then(vmReady("A", true)) + .then(vmReady("A")) .then(activate(1)) .expect(blocked_vm_runs_nothing and s.active == 1) + + // #137: an out-of-reach VM found at the check blocks its project; a probe + // can't unblock it while it's down, and does once it's back. + run anUnreachableVmBlocksUntilItAnswersTest = + init + .then(assignIssue(1)).then(intake(1, "A", false)) + .then(goDown("A")) + .then(activate(1)) + .expect(s.vmUnreachable.contains("A") and s.active == 0) + .then(comeBack("A")).then(probe("A")) + .then(activate(1)) + .expect(all_properties and s.active == 1) }