From cceb30c729e0d957c380fffd56ea8caf57ad4ef4 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Sun, 4 Oct 2026 13:34:09 +0000 Subject: [PATCH] The queue: parking, the operator's order, blocked items (M2.2) Several work items on the new loop (L1, D4): - Switching parks the item on the workspace: its work is committed and its branch pushed; resuming checks the item's branch back out. Branches of open items are never set aside by a fresh workspace. - Next item: the operator's "do #N next" first, otherwise the oldest. - Issues with open Gitea dependencies are blocked until they close, checked at intake and on every round; status shows blocked items and the line. Tests: queue unit tests, the property walk over several issues with dependencies and a new parked_work_safe property (it catches parking disabled), and an end-to-end test with two issues where B depends on A. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE --- docs/Development-Plan.md | 14 +- nisse-dev/src/nisse_dev/gitea.py | 10 ++ nisse-dev/src/nisse_dev/loop/agent.py | 132 ++++++++++++-- nisse-dev/src/nisse_dev/loop/items.py | 6 +- nisse-dev/src/nisse_dev/loop/milestones.py | 45 ++++- nisse-dev/tests/test_loop.py | 4 + nisse-dev/tests/test_loop_milestones.py | 45 +++-- nisse-dev/tests/test_loop_properties.py | 125 ++++++++++++-- nisse-dev/tests/test_loop_queue.py | 192 +++++++++++++++++++++ tests/integration/e2e_support.py | 9 + tests/integration/test_end_to_end_loop.py | 160 ++++++++++++----- 11 files changed, 657 insertions(+), 85 deletions(-) create mode 100644 nisse-dev/tests/test_loop_queue.py diff --git a/docs/Development-Plan.md b/docs/Development-Plan.md index 86544af..3f1e282 100644 --- a/docs/Development-Plan.md +++ b/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 -branches pushed, next-item order, blocked items from Gitea dependencies, a closed -PR handled (L1, D4). End-to-end: two issues and a dependency. +**M2.2 The queue.** *Pre-approved.* **Done** (this PR). Several items: states, +park-and-switch with branches pushed, next-item order, blocked items from Gitea +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 diff --git a/nisse-dev/src/nisse_dev/gitea.py b/nisse-dev/src/nisse_dev/gitea.py index 7c586f0..a84228b 100644 --- a/nisse-dev/src/nisse_dev/gitea.py +++ b/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.""" diff --git a/nisse-dev/src/nisse_dev/loop/agent.py b/nisse-dev/src/nisse_dev/loop/agent.py index 02530b9..cb02eb9 100644 --- a/nisse-dev/src/nisse_dev/loop/agent.py +++ b/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,25 +539,35 @@ 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: - # D3: a fresh workspace when a work item starts. + if self._milestones is not None: + # 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 - self._fact(item, "workspace prepared from the latest default branch") + if not loop.prepared: + self._fact(item, "workspace prepared from the latest default branch") if notes: said = "Set aside from the workspace: " + "; ".join(notes) self._fact(item, said) diff --git a/nisse-dev/src/nisse_dev/loop/items.py b/nisse-dev/src/nisse_dev/loop/items.py index d902a48..70e476e 100644 --- a/nisse-dev/src/nisse_dev/loop/items.py +++ b/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) diff --git a/nisse-dev/src/nisse_dev/loop/milestones.py b/nisse-dev/src/nisse_dev/loop/milestones.py index efb6cea..67d1a44 100644 --- a/nisse-dev/src/nisse_dev/loop/milestones.py +++ b/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: diff --git a/nisse-dev/tests/test_loop.py b/nisse-dev/tests/test_loop.py index 5a88a13..5e9ceeb 100644 --- a/nisse-dev/tests/test_loop.py +++ b/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] diff --git a/nisse-dev/tests/test_loop_milestones.py b/nisse-dev/tests/test_loop_milestones.py index f2030be..4e32e0a 100644 --- a/nisse-dev/tests/test_loop_milestones.py +++ b/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"] - git( - "--git-dir", - str(self.bare), - "update-ref", - "refs/heads/main", - f"refs/heads/{head}", - ) + head = f"refs/heads/{self.prs[n]['head']}" + bare = ["--git-dir", str(self.bare)] + merged = head + if subprocess.run( + ["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", + "-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]: diff --git a/nisse-dev/tests/test_loop_properties.py b/nisse-dev/tests/test_loop_properties.py index bf8c227..e1a851f 100644 --- a/nisse-dev/tests/test_loop_properties.py +++ b/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, -comments, merges, closes, abandon, a crash and restart) with an adaptive -scripted model, and checks every property after every step. A failure prints +walk interleaves what the operator and Gitea can do (new issues, some +depending on others, go-aheads, answers, "do #N next", comments, merges, +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} - for item in open_items: - loop = loop_of(item) + weights = {"intake": 2 if not open_items else 1, "crash": 1, "issue": 1} + item = self.rng.choice(open_items) if open_items else None + 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) - item = open_items[0] if open_items else None - loop = loop_of(item) if item else None + self.steps.append(f"{action}#{loop.issue}" if loop else action) 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: ... diff --git a/nisse-dev/tests/test_loop_queue.py b/nisse-dev/tests/test_loop_queue.py new file mode 100644 index 0000000..a398b5a --- /dev/null +++ b/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() diff --git a/tests/integration/e2e_support.py b/tests/integration/e2e_support.py index 91f0752..7a6e70d 100644 --- a/tests/integration/e2e_support.py +++ b/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(), ( diff --git a/tests/integration/test_end_to_end_loop.py b/tests/integration/test_end_to_end_loop.py index d8d9a01..a466af9 100644 --- a/tests/integration/test_end_to_end_loop.py +++ b/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: - self.transport.receive(OPERATOR, text, thread=loop_of(self.item()).thread) + def say(self, text: str, issue: int = 0) -> None: + 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 - as_operator._request( - "POST", - f"/repos/{repo}/pulls/{pr}/reviews", - json={"event": "APPROVED", "body": "looks good"}, - ) - merge(as_operator, repo, pr) - operator.tick() - deadline = time.monotonic() + DEADLINE_S - nudges = 0 - while operator.item().status != "closed" and time.monotonic() < deadline: - if loop_of(operator.item()).wait == ANSWER: - nudges += 1 - assert nudges <= MAX_ANSWERS, f"never closed: {operator.last()}" - operator.say("It's merged; please close the item.") - operator.settle() - assert operator.item().outcome == "completed" - state = gitea._request("GET", f"/repos/{repo}/issues/{issue}")["state"] - assert state == "closed" + operator.merge_and_close(issue) finally: - # Leave the sandbox tidy: whatever is still open gets closed. - 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}") + 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}/issues/{b}/dependencies", + 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]}, + ) + operator.tick() + assert operator.item(b).position == BLOCKED + assert loop_of(operator.item(b)).blocked_by == (f"{repo}#{a}",) + + # --- A runs to a merged PR; B stays blocked meanwhile --- + item = operator.drive_until(MERGE, a) + prs.append(loop_of(item).pr) + assert operator.item(b).position == BLOCKED + operator.merge_and_close(a) + + # --- A closed: B is unblocked and starts from A's work --- + 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: + operator.tidy(issues, prs) -- 2.54.0