Merge pull request 'The queue: parking, the operator's order, blocked items (M2.2)' (#101) from loop-queue into main

Reviewed-on: dongar/nisse#101
Reviewed-by: Don Garrett <dgarrett@acm.org>

docs/Development-Plan.md

@@ -23,7 +23,7 @@ Only `nisse-dev` has real code; `nisse-core` and `nisse-ops` are placeholders.
 
 | Section | Status | Notes |
 |---|---|---|
-| **L1** Work model | partial | Work items table with an active pointer (M1.1); no issue intake, queue, parking or blocked items |
+| **L1** Work model | mostly done | Work items with an active pointer (M1.1); issue intake (M2.1); queue, parking and blocked items (M2.2) on the new loop; no grace period yet |
 | **L2** Agent loop | partial | `control.py`, a fixed state machine with no-stranding, `stop`, `abandon`. Replaced by the new loop (Phase 2) |
 | **L3** Authority | partial | Revision binding and arrival stamping (`gates.py`); scope is still plan approval, not the issue and its conversation |
 | **L4** Classifier | none | |
@@ -144,9 +144,15 @@ later one widens it and extends the end-to-end test (M1.6) to cover it.
 - Acceptance: the end-to-end test runs that path on the new loop, plus a crash
   mid-publish; then a first manual live run.
 
-**M2.2 The queue.** *Pre-approved.* Several items: states, park-and-switch with
+**M2.2 The queue.** *Pre-approved.* **Done** (this PR). Several items: states,
-branches pushed, next-item order, blocked items from Gitea dependencies, a closed
+park-and-switch with branches pushed, next-item order, blocked items from Gitea
-PR handled (L1, D4). End-to-end: two issues and a dependency.
+dependencies, a closed PR handled (L1, D4). End-to-end: two issues and a dependency.
+- Switching happens as soon as the active item waits; the grace period (§6, L1)
+  is left until a watching operator misses it. Order is "do #N next"; priority
+  labels and driver-chosen order wait for a need.
+- The property walk covers several issues, dependencies and the operator's order,
+  and checks that parked work is never lost (every file an open item wrote is on
+  the workspace or in the forge).
 
 **M2.3 Threads and labels.** *Pre-approved.* Routing by thread, ambiguous commands
 asked about (L5); `bug` grace periods, `discuss`, `design`; questions and

nisse-dev/src/nisse_dev/gitea.py

@@ -237,6 +237,16 @@ class GiteaClient:
             if not item.get("pull_request")
         ]
 
+    def open_dependencies(self, repo: str, index: int) -> list[str]:
+        """The issues this one depends on that are still open, as
+        "owner/repo#N": what blocks it (L1)."""
+        data = self._request("GET", f"/repos/{repo}/issues/{index}/dependencies")
+        return [
+            f"{(item.get('repository') or {}).get('full_name', repo)}#{item['number']}"
+            for item in data or []
+            if item.get("state") == "open"
+        ]
+
     def assigned_by(self, repo: str, index: int, assignee: str) -> str | None:
         """Who last assigned the issue to ``assignee``, from its timeline;
         None if the timeline doesn't say."""

nisse-dev/src/nisse_dev/loop/agent.py

@@ -7,11 +7,15 @@ because by the time the worker reaches it a newer one may be up (L3).
 
 One work item at a time is active. An item runs until its driver calls a
 waiting tool; it then waits on the operator, the active pointer is cleared
-(a safe point, L1), and the next ready item runs. Conversation is not work:
+(a safe point, L1), and the next ready item runs: the operator's order when
+given ("do #14 next"), otherwise the oldest. Switching parks the item that was
+on the workspace: its work is committed and its branch pushed (D4). An issue
+with open Gitea dependencies is blocked until they close. Conversation is not work:
 a message is recorded and changes its item's state, but never interrupts a
 run (L1).
 """
 
+import re
 import threading
 import uuid
 from collections.abc import Callable, Iterable
@@ -24,6 +28,7 @@ from nisse_dev.gitea import GiteaClient, GiteaError, Issue
 from nisse_dev.loop.driver import Driver, RunOutcome
 from nisse_dev.loop.items import (
     ANSWER,
+    BLOCKED,
     CLOSED,
     GO_AHEAD,
     MERGE,
@@ -53,6 +58,7 @@ from nisse_dev.worker import Worker
 
 INTAKE_INTERVAL_S = 30.0
 _GO = frozenset({"go", "approve"})
+_NEXT = re.compile(r"(?:do|work on) #?(\d+) (?:next|first|now)")
 
 # Before the go-ahead the driver may look around (read, list, run a command
 # to reproduce a bug) but not change anything (D1); after it, everything.
@@ -235,8 +241,14 @@ class LoopAgent:
                     issue=issue.number,
                 )
                 continue
-            self._open(issue)
+            try:
+                blocked_by = self._gitea.open_dependencies(self._repo, issue.number)
+            except GiteaError as exc:
+                self._oplog.emit("loop.intake_unavailable", cause=str(exc))
+                continue  # taken on the next round
+            self._open(issue, tuple(blocked_by))
         self.watch()
+        self.check_blocked()
         self._schedule()
 
     # -- PR review and merge (D4, D5) -------------------------------------------
@@ -308,6 +320,40 @@ class LoopAgent:
                     loop = replace(loop, wait="", note="")
                 save(self._store, item.id, position, loop, cause="PR update")
 
+    def check_blocked(self) -> None:
+        """L1: an item that hasn't started running yet is blocked while its
+        Gitea dependencies are open, and ready once they have all closed.
+        Only ready and blocked items are checked: the others are mid-work or
+        waiting on the operator, and an agent that finds a blocker mid-work
+        asks before recording it."""
+        for item in self._store.work_items("open"):
+            if item.position not in (READY, BLOCKED):
+                continue
+            loop = loop_of(item)
+            issue = loop.issue
+            try:
+                blocked_by = tuple(self._gitea.open_dependencies(self._repo, issue))
+            except GiteaError as exc:
+                self._oplog.emit("loop.watch_unavailable", cause=str(exc))
+                continue  # as it was, until Gitea answers
+            position = BLOCKED if blocked_by else READY
+            if position == item.position and blocked_by == loop.blocked_by:
+                continue
+            if position == READY:
+                said = f"#{issue} is unblocked: its dependencies have closed."
+            else:
+                said = f"#{issue} is blocked by {', '.join(blocked_by)}."
+            self._fact(item, said)
+            if position != item.position:
+                self._transport.post(said, thread=loop.thread)
+            save(
+                self._store,
+                item.id,
+                position,
+                replace(loop, blocked_by=blocked_by),
+                cause=said,
+            )
+
     def _from_an_operator(self, issue: Issue) -> bool:
         """L11: only an operator's issue, or one an operator assigned, is
         an instruction."""
@@ -319,18 +365,22 @@ class LoopAgent:
             return False
         return who in self._gitea_operators
 
-    def _open(self, issue: Issue) -> None:
+    def _open(self, issue: Issue, blocked_by: tuple[str, ...]) -> None:
         item_id = f"{self._project}-{issue.number}-{uuid.uuid4().hex[:6]}"
         cause = f"assigned {self._repo}#{issue.number}"
+        ahead = len(self._store.work_items("open"))
+        where = f", {ahead} ahead" if ahead else ""
+        if blocked_by:
+            where += f"; blocked by {', '.join(blocked_by)}"
         root = self._transport.post(
-            f"Queued {self._repo}#{issue.number}: {issue.title}"
+            f"Queued {self._repo}#{issue.number}: {issue.title}{where}"
         )
-        loop = Loop(thread=root, issue=issue.number)
+        loop = Loop(thread=root, issue=issue.number, blocked_by=blocked_by)
         self._store.open_work_item(
             item_id,
             title=issue.title,
             scope=self._project,
-            position=READY,
+            position=BLOCKED if blocked_by else READY,
             data={"loop": loop.to_data(), "roles": self._roles},
             cause=cause,
         )
@@ -370,6 +420,16 @@ class LoopAgent:
         item = item_for_thread(self._store, msg.thread) if msg.thread else None
         text = (msg.original or msg.text).strip()
         command = command_form(msg.text)
+        order = _NEXT.fullmatch(command)
+        if order is not None:
+            self._store.transcript(item.id if item else None, "operator").append(
+                UserMessage(text=text, source=f"transport:{msg.sender}")
+            )
+            self._transport.post(
+                self._put_first(int(order.group(1))), thread=msg.thread
+            )
+            self._schedule()
+            return
         if item is None:
             self._store.transcript(None, "operator").append(
                 UserMessage(text=text, source=f"transport:{msg.sender}")
@@ -393,6 +453,12 @@ class LoopAgent:
             item = self._store.work_item(item.id) or item
         if command == "abandon":
             self._close(item, loop, "abandoned", f"abandoned by {msg.sender}")
+        elif item.position == BLOCKED:
+            self._transport.post(
+                f"#{loop.issue} is blocked by {', '.join(loop.blocked_by)}; "
+                "I'll start on it when they close, or when you remove them.",
+                thread=loop.thread,
+            )
         elif item.position == WAITING and loop.wait == GO_AHEAD and command in _GO:
             if about != loop.rev:
                 # L3: a go-ahead is about what was on screen when it arrived.
@@ -421,16 +487,48 @@ class LoopAgent:
             )
         self._schedule()
 
+    def _put_first(self, issue: int) -> str:
+        """The operator's "do #N next" (L1): it runs before every other
+        ready item, the latest such order first. It doesn't unblock."""
+        items = self._store.work_items("open")
+        item = next((i for i in items if loop_of(i).issue == issue), None)
+        if item is None:
+            return f"#{issue} isn't one of my open work items."
+        first = max(loop_of(i).first for i in items) + 1
+        loop = replace(loop_of(item), first=first)
+        save(self._store, item.id, item.position, loop, cause=f"#{issue} put first")
+        self._fact(item, "the operator put this item first in the queue")
+        if item.position == BLOCKED:
+            return (
+                f"#{issue} goes first once it's unblocked "
+                f"(blocked by {', '.join(loop.blocked_by)})."
+            )
+        if item.position == WAITING:
+            return f"#{issue} goes first, but it's waiting for {loop.wait}: {loop.note}"
+        return f"#{issue} is next."
+
+    def _queue(self) -> list[WorkItem]:
+        """Ready items in the order they run (L1): the operator's order
+        first, then the oldest."""
+        items = self._store.work_items("open")  # oldest first
+        ready = [i for i in items if i.position == READY]
+        return sorted(ready, key=lambda i: -loop_of(i).first)
+
     def status(self) -> str:
         items = self._store.work_items("open")
         if not items:
             return f"idle: no open work items. I take issues assigned to me in {self._repo}."
+        queue = [i.id for i in self._queue()]
         lines = []
         for item in items:
             loop = loop_of(item)
             what = item.position
             if item.position == WAITING:
                 what = f"waiting for {loop.wait} ({loop.note or f'rev-{loop.rev}'})"
+            elif item.position == BLOCKED:
+                what = f"blocked by {', '.join(loop.blocked_by)}"
+            elif item.position == READY:
+                what = f"ready, {queue.index(item.id) + 1} of {len(queue)} in line"
             lines.append(f"#{loop.issue} {item.title}: {what}")
         return "\n".join(lines)
 
@@ -441,24 +539,34 @@ class LoopAgent:
         if self._running:
             return
         while True:
-            ready = [i for i in self._store.work_items("open") if i.position == READY]
+            ready = self._queue()
             if not ready:
                 self._store.set_active(None, cause="nothing ready")
                 return
-            self._run(ready[0])  # oldest first (L1)
+            self._run(ready[0])
 
     def _run(self, item: WorkItem) -> None:
         loop = loop_of(item)
         self._store.set_active(item.id, cause="ready")
-        if not loop.prepared and self._milestones is not None:
+        if self._milestones is not None:
-            # D3: a fresh workspace when a work item starts.
+            # Branches of open items, parked or not, are never set aside.
+            keep = tuple(f"{i.id}-" for i in self._store.work_items("open"))
             try:
-                notes = self._milestones.prepare(f"{item.id}-work")
+                if not loop.prepared:
+                    # D3: a fresh workspace when a work item starts.
+                    notes = self._milestones.prepare(f"{item.id}-work", keep=keep)
+                else:
+                    # D4: back to its branch, parking whoever was there.
+                    notes = []
+                    self._milestones.resume(
+                        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
                 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")
                 return
+            if not loop.prepared:
                 self._fact(item, "workspace prepared from the latest default branch")
             if notes:
                 said = "Set aside from the workspace: " + "; ".join(notes)

nisse-dev/src/nisse_dev/loop/items.py

@@ -13,6 +13,7 @@ from nisse_dev.state import StateStore, WorkItem
 READY = "ready"
 RUNNING = "running"  # a driver run is under way; on boot it means interrupted
 WAITING = "waiting"
+BLOCKED = "blocked"  # waits for its recorded dependencies to close (L1)
 CLOSED = "closed"
 
 # What a waiting item waits for.
@@ -25,7 +26,7 @@ MERGE = "merge"  # the PR's review and merge (D4, D5)
 class Loop:
     thread: str = ""  # the root of the item's Matrix thread (L5)
     issue: int = 0  # its record issue (D1)
-    wait: str = ""  # GO_AHEAD / ANSWER, when waiting
+    wait: str = ""  # GO_AHEAD / ANSWER / MERGE, when waiting
     rev: int = 0  # the latest proposal put up (L3)
     started: bool = False  # scope agreed: a go-ahead was given (D1)
     note: str = ""  # what a waiting item said it is waiting on
@@ -37,6 +38,8 @@ class Loop:
     comments: int = 0  # the last PR conversation comment taken (L8)
     notes: tuple[str, ...] = ()  # review notes taken (#69)
     fixes: int = 0  # review fixes pushed, for follow-up branch names
+    blocked_by: tuple[str, ...] = ()  # open dependencies, "owner/repo#N" (L1)
+    first: int = 0  # the operator's "do #N next": higher runs sooner
     # Where the operator last spoke (D5: answer where asked): "" for the
     # Matrix thread, or the PR it came from, with the inline location and a
     # little of what they said, for quoting.
@@ -51,6 +54,7 @@ class Loop:
 def loop_of(item: WorkItem) -> Loop:
     data = dict(item.data.get("loop", {}))
     data["notes"] = tuple(data.get("notes", ()))
+    data["blocked_by"] = tuple(data.get("blocked_by", ()))
     return Loop(**data)
 
 

nisse-dev/src/nisse_dev/loop/milestones.py

@@ -57,11 +57,14 @@ class Milestones:
 
     # -- a work item's workspace (D3) ------------------------------------------
 
-    def prepare(self, item_branch: str) -> list[str]:
+    def prepare(self, item_branch: str, *, keep: tuple[str, ...] = ()) -> list[str]:
         """A fresh workspace for a work item: fetch, set aside (never
         discard) whatever was left, check out a clean default branch, and put
-        the item on a branch of its own. Returns what was set aside."""
+        the item on a branch of its own. Branches starting with a ``keep``
+        prefix belong to open work items, which parking has pushed: they are
+        left alone. Returns what was set aside."""
         git = self._git
+        self.park(keep=keep)
         git.fetch(credential=self._credential)
         stamp = time.strftime("%Y%m%d-%H%M%S")
         notes: list[str] = []
@@ -73,6 +76,8 @@ class Milestones:
         for branch in git.local_branches():
             if branch == self._base or branch.startswith(f"{SET_ASIDE}/"):
                 continue
+            if keep and branch.startswith(keep):
+                continue
             git.rename_branch(branch, f"{SET_ASIDE}/{stamp}/{branch}")
             notes.append(f"branch {branch}, as {SET_ASIDE}/{stamp}/{branch}")
         if self._base in git.local_branches() and self._ahead_of_origin():
@@ -83,6 +88,42 @@ class Milestones:
         git.checkout_new_branch(item_branch)
         return notes
 
+    # -- parking and resuming (L1, D4) -----------------------------------------
+
+    def park(self, *, keep: tuple[str, ...]) -> str | None:
+        """Before another item uses the workspace: commit whatever the item
+        on it left, and push its branch, so nothing of a parked item lives
+        only on the build VM. Only a harness branch of an open item (one
+        starting with a ``keep`` prefix) is parked; returns its name."""
+        git = self._git
+        branch = git.current_branch()
+        if not keep or not branch.startswith(keep):
+            return None
+        git.commit_all("WIP: parked while another work item runs")
+        if git.remote_sha(self._remote, branch, credential=self._credential) != (
+            git.head_sha()
+        ):
+            git.push(self._remote, branch, credential=self._credential)
+        return branch
+
+    def resume(self, branch: str, *, keep: tuple[str, ...], mine: str = "") -> None:
+        """Back to a parked item: park whatever is on the workspace now, then
+        check out the item's branch. The local branch wins when it exists (it
+        can only be ahead: parking pushed it); otherwise it comes from origin.
+        A workspace already on one of the item's own branches (``mine``, a
+        prefix) is where it left off, and stays."""
+        git = self._git
+        current = git.current_branch()
+        if current == branch or (mine and current.startswith(mine)):
+            return
+        self.park(keep=keep)
+        if branch in git.local_branches():
+            git.checkout_existing_branch(branch)
+            git.reset_to(branch, branch)
+        else:
+            git.fetch(credential=self._credential)
+            git.reset_to(branch, f"origin/{branch}")
+
     def _ahead_of_origin(self) -> bool:
         current = self._git.current_branch()
         try:

nisse-dev/tests/test_loop.py

@@ -48,6 +48,10 @@ class FakeGitea:
         self.assigner = assigner
         self.comments: list[tuple[int, str]] = []
         self.closed: list[int] = []
+        self.deps: dict[int, list[int]] = {}
+
+    def open_dependencies(self, repo: str, index: int) -> list[str]:
+        return [f"{repo}#{n}" for n in self.deps.get(index, []) if n not in self.closed]
 
     def assigned_issues(self, repo: str, assignee: str) -> list[Issue]:
         return [i for i in self.issues if i.number not in self.closed]

nisse-dev/tests/test_loop_milestones.py

@@ -58,11 +58,19 @@ class Forge:
         self.notes: dict[int, list[ReviewNote]] = {}
         self.posted: list[tuple[int, str]] = []
         self.closed_issues: list[int] = []
-        self.issue = Issue(14, "Add count_words", "count words", "don", ())
+        self.issues = [Issue(14, "Add count_words", "count words", "don", ())]
+        self.deps: dict[int, list[int]] = {}  # issue -> the issues it depends on
 
     # intake
     def assigned_issues(self, repo: str, assignee: str) -> list[Issue]:
-        return [] if self.issue.number in self.closed_issues else [self.issue]
+        return [i for i in self.issues if i.number not in self.closed_issues]
+
+    def open_dependencies(self, repo: str, index: int) -> list[str]:
+        return [
+            f"{repo}#{n}"
+            for n in self.deps.get(index, [])
+            if n not in self.closed_issues
+        ]
 
     def assigned_by(self, repo: str, index: int, assignee: str) -> str:
         return "don"
@@ -109,14 +117,31 @@ class Forge:
 
     # what the operator does in Gitea
     def merge(self, n: int) -> None:
-        head = self.prs[n]["head"]
+        head = f"refs/heads/{self.prs[n]['head']}"
-        git(
+        bare = ["--git-dir", str(self.bare)]
-            "--git-dir",
+        merged = head
-            str(self.bare),
+        if subprocess.run(
-            "update-ref",
+            ["git", *bare, "merge-base", "--is-ancestor", "refs/heads/main", head],
+            check=False,
+        ).returncode:
+            # Main moved on (another item's PR): a real merge commit.
+            tree = git(*bare, "merge-tree", "--write-tree", "refs/heads/main", head)
+            merged = git(
+                *bare,
+                "-c",
+                "user.name=g",
+                "-c",
+                "user.email=g@t",
+                "commit-tree",
+                tree,
+                "-p",
                 "refs/heads/main",
-            f"refs/heads/{head}",
+                "-p",
+                head,
+                "-m",
+                f"Merge PR {n}",
             )
+        git(*bare, "update-ref", "refs/heads/main", merged)
         self.prs[n].update(state="closed", merged=True)
 
     def files(self, n: int) -> list[str]:

nisse-dev/tests/test_loop_properties.py

@@ -1,9 +1,10 @@
 """specs/loop.qnt's properties, checked against the real loop (M2.1, PR D).
 
 The Quint model checks the design; this checks the code. A seeded random
-walk interleaves what the operator and Gitea can do (go-aheads, answers,
+walk interleaves what the operator and Gitea can do (new issues, some
-comments, merges, closes, abandon, a crash and restart) with an adaptive
+depending on others, go-aheads, answers, "do #N next", comments, merges,
-scripted model, and checks every property after every step. A failure prints
+closes, abandon, a crash and restart) with an adaptive scripted model, and
+checks every property after every step. A failure prints
 the seed and the steps, so it replays exactly.
 """
 
@@ -19,17 +20,25 @@ from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture
 
 from nisse_dev.cancel import Cancel
 from nisse_dev.canonical import Fact, ToolCall
-from nisse_dev.gitea import Comment
+from nisse_dev.gitea import Comment, Issue
 from nisse_dev.loop.agent import LoopAgent
 from nisse_dev.loop.driver import Driver
-from nisse_dev.loop.items import ANSWER, GO_AHEAD, MERGE, WAITING, loop_of
+from nisse_dev.loop.items import (
+    ANSWER,
+    BLOCKED,
+    GO_AHEAD,
+    MERGE,
+    RUNNING,
+    WAITING,
+    loop_of,
+)
 from nisse_dev.oplog import OpLog
 from nisse_dev.tools import Workspace
 from nisse_dev.transport import InboundMessage, InMemoryTransport
 from nisse_dev.worker import InlineWorker
 
 SEEDS = range(32)
-STEPS = 20
+STEPS = 30
 
 
 class PolicyModel:
@@ -55,6 +64,13 @@ class PolicyModel:
             return [call("restate", text=f"restatement {self.n}")]
         if "wait_for_review" in offered and self.rng.random() < 0.2:
             return [call("wait_for_review", note=f"waiting {self.n}")]
+        if self.rng.random() < 0.15:
+            # Half done, then a question: work a switch must park.
+            return [
+                call("start_milestone", name=f"m{self.n}"),
+                call("write_file", path=f"f{self.n}.py", content=f"x = {self.n}\n"),
+                call("ask_human", question=f"halfway {self.n}?"),
+            ]
         merged = any("merged" in t for t in facts[-6:])
         if merged and self.rng.random() < 0.8:
             return [call("close_item", summary=f"done {self.n}")]
@@ -71,7 +87,8 @@ class Walk:
         self.rng = random.Random(seed)
         self.seed = seed
         self.steps: list[str] = []
-        self.shown_at_go: list[int] = []  # the revision on screen at each go
+        self.shown_at_go: list[str] = []  # item:revision on screen at each go
+        self.next_issue = 15
         self.agent, self.transport = self._agent()
 
     def _agent(self) -> tuple[LoopAgent, InMemoryTransport]:
@@ -105,20 +122,30 @@ class Walk:
         open_items = [i for i in self.items() if i.status == "open"]
         # Weighted towards what an item can actually do, so review paths
         # are walked often, not drowned out by intake and crashes.
-        weights = {"intake": 2 if not open_items else 1, "crash": 1}
+        weights = {"intake": 2 if not open_items else 1, "crash": 1, "issue": 1}
-        for item in open_items:
+        item = self.rng.choice(open_items) if open_items else None
-            loop = loop_of(item)
+        loop = loop_of(item) if item else None
+        if item and loop:
             weights.update({"go": 3, "answer": 2, "abandon": 1, "go-stale": 1})
+            weights.update({"first": 1})
             if loop.pr:
                 weights.update(
                     {"comment": 4, "merge": 2, "close-pr": 1, "comment-merge": 2}
                 )
         action = self.rng.choices(list(weights), list(weights.values()))[0]
-        self.steps.append(action)
+        self.steps.append(f"{action}#{loop.issue}" if loop else action)
-        item = open_items[0] if open_items else None
-        loop = loop_of(item) if item else None
         if action == "intake":
             self.agent.intake()
+        elif action == "issue":
+            # A new issue, sometimes depending on one that is still open.
+            n, self.next_issue = self.next_issue, self.next_issue + 1
+            forge.issues.append(Issue(n, f"task {n}", f"do task {n}", "don", ()))
+            live = [i.number for i in forge.assigned_issues("", "") if i.number != n]
+            if live and self.rng.random() < 0.5:
+                forge.deps[n] = [self.rng.choice(live)]
+            self.agent.intake()
+        elif action == "first" and loop:
+            self.agent.handle(InboundMessage(DON, f"do #{loop.issue} next"), None)
         elif action == "crash":
             # Volatile state is lost: a new agent over the same store.
             self.agent, self.transport = self._agent()
@@ -127,7 +154,7 @@ class Walk:
         elif action in ("go", "answer", "abandon") and loop:
             text = {"go": "go", "answer": "an answer", "abandon": "abandon"}[action]
             if action == "go" and item.position == WAITING and loop.wait == GO_AHEAD:
-                self.shown_at_go.append(loop.rev)
+                self.shown_at_go.append(f"{item.id}:{loop.rev}")
             self.transport.receive(DON, text, thread=loop.thread)
         elif action == "go-stale" and loop and loop.rev > 1:
             # A go sent while an older revision was on screen.
@@ -185,9 +212,21 @@ class Walk:
                 if isinstance(f.event, Fact)
                 and f.event.text.startswith("go-ahead on rev-")
             ]
-            assert all(rev in self.shown_at_go for rev in gos), (
+            assert all(f"{item.id}:{rev}" in self.shown_at_go for rev in gos), (
                 f"go_binds: {gos}: {where}"
             )
+            # Blocked comes from external truth (L1): an item whose issue
+            # had an open dependency from the start never ran before it closed.
+            if forge.open_dependencies("don/sandbox", loop.issue):
+                assert item.position == BLOCKED or item.status == "closed", (
+                    f"blocked_never_runs: #{loop.issue} {item.position}: {where}"
+                )
+                ran = [
+                    e
+                    for e in store.transcript(item.id, "driver").read()
+                    if e.actor == "driver"
+                ]
+                assert not ran, f"blocked_never_runs (driver): #{loop.issue}: {where}"
             mine = [p for p in forge.prs.values() if p["head"].startswith(item.id)]
             assert sum(p["state"] == "open" for p in mine) <= 1, (
                 f"milestone_after_merge: {where}"
@@ -199,6 +238,10 @@ class Walk:
                 assert not [
                     i for i in store.pending_intents() if i.work_item == item.id
                 ], f"closed_items_leave_nothing (intent): {where}"
+        assert not [i for i in items if i.position == RUNNING], (
+            f"one_active (nothing left running): {where}"
+        )
+        self._parked_work_safe(where)
         heads = [p["head"] for p in forge.prs.values()]
         assert len(heads) == len(set(heads)), f"one_pr_per_milestone: {where}"
         bare = self.world["bare"]
@@ -216,6 +259,58 @@ class Walk:
             )
             assert landed, f"review_fixes_land (PR {n}): {where}"
 
+    def _parked_work_safe(self, where: str) -> None:
+        """L1, D4: nothing of a parked item lives only on the build VM: what
+        every branch of an open item that isn't on the workspace points at is
+        in the forge."""
+        root, bare = self.world["repo"].root, self.world["bare"]
+        current = _git(root, "rev-parse", "--abbrev-ref", "HEAD")
+        ids = [i.id for i in self.items() if i.status == "open"]
+        for line in _git(
+            root, "for-each-ref", "--format=%(refname:short)", "refs/heads"
+        ).splitlines():
+            if line == current or not any(line.startswith(f"{i}-") for i in ids):
+                continue
+            local = _git(root, "rev-parse", line)
+            pushed = subprocess.run(
+                [
+                    "git",
+                    "--git-dir",
+                    str(bare),
+                    "cat-file",
+                    "-e",
+                    f"{local}^{{commit}}",
+                ],
+                capture_output=True,
+                check=False,
+            )
+            assert pushed.returncode == 0, (
+                f"parked_work_safe: {line} isn't pushed: {where}"
+            )
+        # And no write is lost: every file an open item's driver wrote is on
+        # the workspace while it's that item's, or in the forge.
+        for item in self.items():
+            if item.status != "open":
+                continue
+            written = {
+                str(e.event.arguments.get("path"))
+                for e in self.world["store"].transcript(item.id, "driver").read()
+                if isinstance(e.event, ToolCall) and e.event.tool == "write_file"
+            }
+            here = current.startswith(f"{item.id}-")
+            for path in written:
+                if here and (root / path).exists():
+                    continue
+                history = _git(bare, "log", "--all", "--format=%H", "--", path)
+                assert history, f"parked_work_safe: {path} of #{item.id} lost: {where}"
+
+
+def _git(root: Path, *args: str) -> str:
+    where = ["--git-dir", str(root)] if root.suffix == ".git" else ["-C", str(root)]
+    return subprocess.run(
+        ["git", *where, *args], capture_output=True, text=True, check=True
+    ).stdout.strip()
+
 
 class _Null:
     def write(self, _: str) -> None: ...

nisse-dev/tests/test_loop_queue.py

@@ -0,0 +1,192 @@
+"""The queue in the new loop (M2.2): several items, parking and resuming on
+switch, the operator's order, and items blocked by Gitea dependencies (L1,
+D4). Real git on the throwaway build host; a fake Gitea."""
+
+from typing import Any
+
+from test_loop_milestones import DON, Forge, ScriptedModel, call, git, loop_agent
+from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture
+
+from nisse_dev.gitea import Comment, Issue
+from nisse_dev.loop.items import (
+    ANSWER,
+    BLOCKED,
+    GO_AHEAD,
+    MERGE,
+    READY,
+    WAITING,
+    item_for_issue,
+    loop_of,
+)
+from nisse_dev.transport import InboundMessage, InMemoryTransport
+
+
+def two_issues(forge: Forge) -> None:
+    forge.issues.append(Issue(15, "Add count_lines", "count lines", "don", ()))
+
+
+def item(world: dict[str, Any], issue: int) -> Any:
+    found = item_for_issue(world["store"], issue)
+    assert found is not None
+    return found
+
+
+def say(
+    transport: InMemoryTransport, world: dict[str, Any], issue: int, text: str
+) -> None:
+    transport.receive(DON, text, thread=loop_of(item(world, issue)).thread)
+
+
+def restated(transport: InMemoryTransport) -> list[str]:
+    """The threads restatements went up in, in order."""
+    return [m.thread for m in transport.outbound if m.text.startswith("rev-1:")]
+
+
+def test_switching_parks_the_work_and_resuming_finds_it(
+    world: dict[str, Any],
+) -> None:
+    forge: Forge = world["forge"]
+    two_issues(forge)
+    model = ScriptedModel(
+        [call("restate", text="count_words")],  # 14
+        [call("restate", text="count_lines")],  # 15
+        [  # 14, after its go: a PR
+            call("start_milestone", name="count_words"),
+            call("write_file", path="words.py", content="W = 1\n"),
+            call("finish_milestone", summary="count_words"),
+        ],
+        [  # 15, after its go: half done, then a question
+            call("start_milestone", name="count_lines"),
+            call("write_file", path="lines.py", content="L = 1\n"),
+            call("ask_human", question="one per line, or per paragraph?"),
+        ],
+        [  # 14, a review comment
+            call("write_file", path="words.py", content="W = 2\n"),
+            call("finish_milestone", summary="review fix"),
+        ],
+        [  # 15, answered: back where it was
+            call("run_command", command="cat lines.py"),
+            call("finish_milestone", summary="count_lines"),
+        ],
+    )
+    agent, transport = loop_agent(world, model)
+    agent.intake()
+    assert [loop_of(item(world, n)).wait for n in (14, 15)] == [GO_AHEAD, GO_AHEAD]
+    # The second item is acknowledged with its place in line.
+    assert any("#15" in m.text and "1 ahead" in m.text for m in transport.outbound)
+    say(transport, world, 14, "go")
+    assert loop_of(item(world, 14)).wait == MERGE
+    a_branch = loop_of(item(world, 14)).branch
+    say(transport, world, 15, "go")
+    assert loop_of(item(world, 15)).wait == ANSWER
+    # 15 was prepared while 14's branch was on the workspace: kept, not set aside.
+    assert not any("Set aside" in m.text for m in transport.outbound)
+    assert a_branch in world["repo"].local_branches()
+
+    [pr] = forge.prs
+    forge.comments[pr] = [Comment(7, "don", "make W 2", "t")]
+    agent.watch()
+    agent._schedule()
+
+    # Switching to 14 parked 15: its half-done work is committed and pushed.
+    b_branch = loop_of(item(world, 15)).branch
+    pushed = git(
+        "--git-dir", str(world["bare"]), "show", f"refs/heads/{b_branch}:lines.py"
+    )
+    assert pushed == "L = 1"
+    # 14's fix went to its own branch, without 15's work.
+    assert loop_of(item(world, 14)).wait == MERGE
+    files = git("--git-dir", str(world["bare"]), "ls-tree", "--name-only", a_branch)
+    assert "words.py" in files.split() and "lines.py" not in files.split()
+    assert "W = 2" in git(
+        "--git-dir", str(world["bare"]), "show", f"refs/heads/{a_branch}:words.py"
+    )
+
+    say(transport, world, 15, "one per line")
+    assert world["repo"].current_branch() == b_branch
+    assert (world["repo"].root / "lines.py").read_text() == "L = 1\n"
+    assert loop_of(item(world, 15)).wait == MERGE
+    assert len(forge.prs) == 2
+
+
+def test_a_blocked_item_waits_for_its_dependency_to_close(
+    world: dict[str, Any],
+) -> None:
+    forge: Forge = world["forge"]
+    two_issues(forge)
+    forge.deps[15] = [14]
+    model = ScriptedModel(
+        [call("restate", text="count_words")],
+        [call("restate", text="count_lines")],
+    )
+    agent, transport = loop_agent(world, model)
+    agent.intake()
+    assert item(world, 15).position == BLOCKED
+    assert loop_of(item(world, 15)).blocked_by == ("don/sandbox#14",)
+    assert any(
+        "#15" in m.text and "blocked by don/sandbox#14" in m.text
+        for m in transport.outbound
+    )
+    assert "#15 Add count_lines: blocked by don/sandbox#14" in agent.status()
+    # Talking to it doesn't start it.
+    say(transport, world, 15, "go")
+    assert item(world, 15).position == BLOCKED
+    assert "blocked" in transport.outbound[-1].text
+
+    forge.closed_issues.append(14)
+    agent.intake()
+    assert (item(world, 15).position, loop_of(item(world, 15)).wait) == (
+        WAITING,
+        GO_AHEAD,
+    )
+    assert loop_of(item(world, 15)).blocked_by == ()
+
+
+def test_an_item_that_gains_a_dependency_is_blocked(world: dict[str, Any]) -> None:
+    forge: Forge = world["forge"]
+    two_issues(forge)
+    model = ScriptedModel([call("restate", text="count_words")])
+    agent, _ = loop_agent(world, model)
+    agent._running = True  # hold the queue: both stay ready
+    agent.intake()
+    assert item(world, 15).position == READY
+    forge.deps[15] = [14]
+    agent.check_blocked()
+    assert item(world, 15).position == BLOCKED
+    agent._running = False
+    agent._schedule()  # 14 restates; 15 stays blocked
+    assert item(world, 15).position == BLOCKED
+    assert loop_of(item(world, 14)).wait == GO_AHEAD
+
+
+def test_the_operators_order_comes_before_the_oldest(world: dict[str, Any]) -> None:
+    forge: Forge = world["forge"]
+    two_issues(forge)
+    forge.issues.append(Issue(16, "Add count_chars", "count chars", "don", ()))
+    forge.issues.append(Issue(17, "Add count_vowels", "count vowels", "don", ()))
+    for n in (15, 16, 17):
+        forge.deps[n] = [14]
+    model = ScriptedModel(*[[call("restate", text=f"r{n}")] for n in range(4)])
+    agent, transport = loop_agent(world, model)
+    agent.intake()  # 14 runs; the rest wait on it
+    transport.receive(DON, "Do #17 next.")
+    assert "#17 goes first once it's unblocked" in transport.outbound[-1].text
+    transport.receive(DON, "do #99 next")
+    assert "#99 isn't one of my open work items" in transport.outbound[-1].text
+
+    forge.closed_issues.append(14)
+    agent.intake()
+    threads = {loop_of(item(world, n)).thread: n for n in (14, 15, 16, 17)}
+    assert [threads[t] for t in restated(transport)] == [14, 17, 15, 16]
+
+
+def test_status_shows_the_line(world: dict[str, Any]) -> None:
+    forge: Forge = world["forge"]
+    two_issues(forge)
+    agent, _ = loop_agent(world, ScriptedModel())
+    agent._running = True  # hold the queue
+    agent.intake()
+    assert "#14 Add count_words: ready, 1 of 2 in line" in agent.status()
+    assert "#15 Add count_lines: ready, 2 of 2 in line" in agent.status()
+    agent.handle(InboundMessage(DON, "do #15 next"), None)
+    assert "#15 Add count_lines: ready, 1 of 2 in line" in agent.status()

tests/integration/e2e_support.py

@@ -17,6 +17,15 @@ def task(run: str) -> str:
     )
 
 
+def dependent_task(run: str) -> str:
+    """A request that builds on task(run): it can only be done after it."""
+    return (
+        f"Add a function consonants_{run}(text) to the sandbox package that "
+        "returns how many letters in the text are not vowels, using "
+        f"vowels_{run}, with tests."
+    )
+
+
 def operator_gitea(url: str) -> GiteaClient:
     token = os.environ.get("NISSE_OPERATOR_TOKEN", "")
     assert token.strip(), (

tests/integration/test_end_to_end_loop.py

@@ -1,6 +1,7 @@
-"""The end-to-end test on the new loop (Development Plan M2.1, PR D): the
+"""The end-to-end tests on the new loop (Development Plan M2.1, M2.2): the
 whole slice, from an issue assigned in Gitea to a merged PR and a closed work
-item, with real models, the Gitea sandbox and a real build host.
+item, and the queue, with an issue blocked on another until it closes; with
+real models, the Gitea sandbox and a real build host.
 
 A scripted operator plays the operator's part: in Gitea through its own test
 account (NISSE_OPERATOR_TOKEN; an operator only in nisse-test.yaml), and in
@@ -19,13 +20,21 @@ from pathlib import Path
 from typing import Any
 
 import pytest
-from e2e_support import base_branch, operator_gitea, task
+from e2e_support import base_branch, dependent_task, operator_gitea, task
 
 from nisse_dev.config import AgentConfig
 from nisse_dev.execution import SSHHost
 from nisse_dev.gitea import GiteaClient, GiteaError
 from nisse_dev.loop.build import RunningLoop, build_loop_agent
-from nisse_dev.loop.items import ANSWER, GO_AHEAD, MERGE, WAITING, loop_of
+from nisse_dev.loop.items import (
+    ANSWER,
+    BLOCKED,
+    GO_AHEAD,
+    MERGE,
+    WAITING,
+    item_for_issue,
+    loop_of,
+)
 from nisse_dev.transport import InMemoryTransport
 
 pytestmark = [pytest.mark.integration, pytest.mark.e2e]
@@ -121,20 +130,25 @@ class Operator:
         """Intake and PR watching, as the agent's timer would do."""
         self.agent.worker.call(self.agent.loop.intake, timeout=DEADLINE_S)
 
-    def item(self) -> Any:
+    def item(self, issue: int = 0) -> Any:
+        """The work item for ``issue``; by default the first one."""
+        if issue:
+            store = self.agent.store
+            return self.agent.worker.call(lambda: item_for_issue(store, issue))
         items = self.agent.worker.call(self.agent.store.work_items)
         return items[0] if items else None
 
-    def say(self, text: str) -> None:
+    def say(self, text: str, issue: int = 0) -> None:
-        self.transport.receive(OPERATOR, text, thread=loop_of(self.item()).thread)
+        thread = loop_of(self.item(issue)).thread
+        self.transport.receive(OPERATOR, text, thread=thread)
         self.settle()
 
-    def drive_until(self, wait: str) -> Any:
+    def drive_until(self, wait: str, issue: int = 0) -> Any:
         """Answer whatever the agent asks until it waits for ``wait``."""
         deadline = time.monotonic() + DEADLINE_S
         answers = 0
         while time.monotonic() < deadline:
-            item = self.item()
+            item = self.item(issue)
             assert item is not None, "no work item"
             loop = loop_of(item)
             if item.position == WAITING and loop.wait == wait:
@@ -142,16 +156,56 @@ class Operator:
             if item.status == "closed":
                 raise AssertionError(f"closed early: {self.last()}")
             if item.position == WAITING and loop.wait == GO_AHEAD:
-                self.say("go")
+                self.say("go", issue)
             elif item.position == WAITING and loop.wait == ANSWER:
                 answers += 1
                 # Answering forever is how a stuck agent looks: give up.
                 assert answers <= MAX_ANSWERS, f"stuck asking: {self.last()}"
-                self.say("Use your best judgement and proceed.")
+                self.say("Use your best judgement and proceed.", issue)
             else:
                 self.settle()
         raise AssertionError(f"never waited for {wait}: {self.last()}")
 
+    def merge_and_close(self, issue: int) -> None:
+        """Approve and merge the item's PR, and see the agent close the item."""
+        gitea, repo = self.agent.gitea, self.agent.config.gitea_repo
+        as_operator = operator_gitea(self.agent.config.gitea_url)
+        pr = loop_of(self.item(issue)).pr
+        with contextlib.suppress(GiteaError):  # approval may not be required
+            as_operator._request(
+                "POST",
+                f"/repos/{repo}/pulls/{pr}/reviews",
+                json={"event": "APPROVED", "body": "looks good"},
+            )
+        merge(as_operator, repo, pr)
+        self.tick()
+        deadline = time.monotonic() + DEADLINE_S
+        nudges = 0
+        while self.item(issue).status != "closed" and time.monotonic() < deadline:
+            if loop_of(self.item(issue)).wait == ANSWER:
+                nudges += 1
+                assert nudges <= MAX_ANSWERS, f"never closed: {self.last()}"
+                self.say("It's merged; please close the item.", issue)
+            self.settle()
+        assert self.item(issue).outcome == "completed"
+        state = gitea._request("GET", f"/repos/{repo}/issues/{issue}")["state"]
+        assert state == "closed"
+
+    def tidy(self, issues: list[int], prs: list[int]) -> None:
+        """Leave the sandbox tidy: whatever is still open gets closed."""
+        gitea, repo = self.agent.gitea, self.agent.config.gitea_repo
+        as_operator = operator_gitea(self.agent.config.gitea_url)
+        for issue in issues:
+            item = self.item(issue)
+            if item is not None and item.status == "open":
+                self.say("abandon", issue)
+            with contextlib.suppress(GiteaError):
+                as_operator.close_issue(repo, issue)
+        for pr in prs:
+            with contextlib.suppress(GiteaError):
+                head = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["ref"]
+                as_operator._request("DELETE", f"/repos/{repo}/branches/{head}")
+
     def last(self) -> str:
         return self.transport.outbound[-1].text if self.transport.outbound else ""
 
@@ -201,34 +255,58 @@ def test_the_slice_end_to_end(
         assert marker in diff, "the pushed fix doesn't do what the comment asked"
 
         # --- the operator approves and merges; the agent closes the item ---
-        with contextlib.suppress(GiteaError):  # approval may not be required
+        operator.merge_and_close(issue)
+    finally:
+        operator.tidy([issue] if issue is not None else [], [pr] if pr else [])
+
+
+def test_an_issue_blocked_on_another_waits_for_it(
+    running: tuple[RunningLoop, InMemoryTransport, str],
+) -> None:
+    """M2.2: B depends on A in Gitea. B is blocked until A's PR merges and A
+    closes; then B starts from a workspace that has A's work."""
+    agent, transport, run = running
+    operator = Operator(agent, transport)
+    gitea, repo = agent.gitea, agent.config.gitea_repo
+    as_operator = operator_gitea(agent.config.gitea_url)
+    owner, name = repo.split("/")
+    issues: list[int] = []
+    prs: list[int] = []
+    try:
+        a = as_operator.create_issue(repo, f"vowels_{run}", task(run))
+        issues.append(a)
+        b = as_operator.create_issue(repo, f"consonants_{run}", dependent_task(run))
+        issues.append(b)
         as_operator._request(
             "POST",
-                f"/repos/{repo}/pulls/{pr}/reviews",
+            f"/repos/{repo}/issues/{b}/dependencies",
-                json={"event": "APPROVED", "body": "looks good"},
+            json={"owner": owner, "repo": name, "index": a},
+        )
+        for n in (b, a):  # B first: the dependency, not the order, decides
+            as_operator._request(
+                "PATCH",
+                f"/repos/{repo}/issues/{n}",
+                json={"assignees": [agent.account]},
             )
-        merge(as_operator, repo, pr)
         operator.tick()
-        deadline = time.monotonic() + DEADLINE_S
+        assert operator.item(b).position == BLOCKED
-        nudges = 0
+        assert loop_of(operator.item(b)).blocked_by == (f"{repo}#{a}",)
-        while operator.item().status != "closed" and time.monotonic() < deadline:
+
-            if loop_of(operator.item()).wait == ANSWER:
+        # --- A runs to a merged PR; B stays blocked meanwhile ---
-                nudges += 1
+        item = operator.drive_until(MERGE, a)
-                assert nudges <= MAX_ANSWERS, f"never closed: {operator.last()}"
+        prs.append(loop_of(item).pr)
-                operator.say("It's merged; please close the item.")
+        assert operator.item(b).position == BLOCKED
-            operator.settle()
+        operator.merge_and_close(a)
-        assert operator.item().outcome == "completed"
+
-        state = gitea._request("GET", f"/repos/{repo}/issues/{issue}")["state"]
+        # --- A closed: B is unblocked and starts from A's work ---
-        assert state == "closed"
+        operator.tick()
+        assert operator.item(b).position != BLOCKED
+        item = operator.drive_until(MERGE, b)
+        prs.append(loop_of(item).pr)
+        files = gitea._request("GET", f"/repos/{repo}/pulls/{prs[-1]}/files")
+        diff = gitea._http.get(f"/repos/{repo}/pulls/{prs[-1]}.diff").text
+        assert f"consonants_{run}" in diff, [f["filename"] for f in files]
+        # Built on A's merged work, so A's function isn't added again.
+        assert f"\n+def vowels_{run}" not in diff
     finally:
-        # Leave the sandbox tidy: whatever is still open gets closed.
+        operator.tidy(issues, prs)
-        item = operator.item()
-        if item is not None and item.status == "open":
-            operator.say("abandon")
-        if issue is not None:
-            with contextlib.suppress(GiteaError):
-                as_operator.close_issue(repo, issue)
-        if pr:
-            with contextlib.suppress(GiteaError):
-                head = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["ref"]
-                as_operator._request("DELETE", f"/repos/{repo}/branches/{head}")