"""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 (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. """ import os import random import re import subprocess import uuid from pathlib import Path from typing import Any import pytest from model_contract import assert_renderable from test_loop_milestones import DON, Forge, call, make_world from test_loop_milestones import world as world # noqa: PLC0414 -- the fixture from test_loop_projects import Router from nisse_dev.cancel import Cancel from nisse_dev.canonical import Fact, PlanSubmitted, ToolCall, UserMessage from nisse_dev.delegation import DELEGATION_INTENT, Caps from nisse_dev.execution import SSHHost from nisse_dev.gitea import Comment, Issue from nisse_dev.llm import RoleMap from nisse_dev.loop.agent import LoopAgent, Project from nisse_dev.loop.driver import Driver from nisse_dev.loop.help import Helpers from nisse_dev.loop.items import ( ANSWER, BLOCKED, CHAT, GO_AHEAD, MERGE, PROVISION, 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 # How many walks, and from which seed: all 48 by default; CI runs fewer on a # PR, from a seed that moves with the run number, so PRs still cover new # ground (NISSE_WALK_SEEDS, NISSE_WALK_FIRST). Two thirds walk one project, # a third two. _WALKS = int(os.environ.get("NISSE_WALK_SEEDS") or 48) _FIRST = int(os.environ.get("NISSE_WALK_FIRST") or 0) SEEDS = range(_FIRST, _FIRST + _WALKS * 2 // 3) TWO_PROJECT_SEEDS = range(100 + _FIRST, 100 + _FIRST + _WALKS - _WALKS * 2 // 3) STEPS = 30 class PolicyModel: """Plays a plausible driver: restate first, then a milestone, fixes after comments, close after a merge; sometimes asks a question or replies in prose. Seeded, so a walk replays exactly.""" def __init__(self, rng: random.Random, projects: list[str]) -> None: self.rng = rng self.projects = projects self.n = 0 def complete(self, events, tools, model, system=None) -> list[Any]: assert_renderable(events) # as the real API would self.n += 1 offered = {t["function"]["name"] for t in tools} facts = [getattr(e.event, "text", "") for e in events] # A `design` item writes documents until its design has merged. design = "labelled design" in (system or "") and not any( "the design has merged" in t for t in facts ) path = f"docs/d{self.n}.md" if design else f"f{self.n}.py" roll = self.rng.random() if roll < 0.08: return [call("ask_human", question=f"question {self.n}?")] if roll < 0.12: return [ToolCall(call_id=f"p{self.n}", tool="nonsense", arguments={})] if "reply" in offered and not offered & { "propose_work", "restate", "ask_human", "start_milestone", }: return [call("reply", text=f"because {self.n}")] # a question run if "reply" in offered and "propose_work" not in offered and roll < 0.2: return [call("reply", text=f"noted {self.n}")] # a remark answered if "reply" in offered: # a conversation: talk, or propose work if self.rng.random() < 0.1: return [ call( "propose_project", name=f"proj{self.n}", purpose=f"a project {self.n}.", decisions="Python.", ) ] if self.rng.random() < 0.4: return [ call( "propose_work", project=self.rng.choice(self.projects), title=f"filed {self.n}", summary=f"do thing {self.n}.", ) ] return [call("reply", text=f"an answer {self.n}")] if "restate" not in offered and "finish_milestone" not in offered: return [call("ask_human", question=f"talk first {self.n}?")] # discuss if "finish_milestone" not in offered: return [call("restate", text=f"restatement {self.n}", changed=f"c{self.n}")] if "delegate" in offered and self.rng.random() < 0.1: return [call("delegate", task=f"help {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=path, 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}")] return [ call("start_milestone", name=f"m{self.n}"), call("write_file", path=path, content=f"x = {self.n}\n"), call( "finish_milestone", summary=f"work {self.n}", final=self.rng.choice(["yes", "no"]), ), ] class HelperModel: """A scripted helper: does some work, then hands back, succeeding or not as the seed says.""" def __init__(self, rng: random.Random) -> None: self.rng = rng self.n = 0 def complete(self, events, tools, model, system=None) -> list[Any]: assert_renderable(events) self.n += 1 return [ call("write_file", path=f"h{self.n}.py", content=f"h = {self.n}\n"), ToolCall( call_id=f"hb{self.n}", tool="hand_back", arguments={ "changed": f"h{self.n}.py", "now_true": "helped", "next": "go on", "succeeded": self.rng.random() < 0.6, }, ), ] class Walk: """``worlds``: project name -> its world (forge, bare repo, clone, milestones), all over one store.""" def __init__(self, worlds: dict[str, dict[str, Any]], seed: int) -> None: self.worlds = worlds self.world = next(iter(worlds.values())) # the store's home self.rng = random.Random(seed) self.seed = seed self.steps: list[str] = [] self.shown_at_go: list[str] = [] # item:revision on screen at each go self.now = 1000.0 self.agent, self.transport = self._agent() def reads_yes(self, proposal: str, reply: str) -> bool: """The scripted go-ahead reader: a plain yes usually reads as one, and nothing else ever does.""" return reply == "yep, that's right" and self.rng.random() < 0.7 def _agent(self) -> tuple[LoopAgent, InMemoryTransport]: transport = InMemoryTransport() # Event IDs are unique across restarts, as a server's are: a new # transport goes on numbering where the old one stopped. old = getattr(self, "transport", None) transport._next_id = old._next_id if old is not None else 0 cancel = Cancel() router = Router({w["repo_name"]: w["forge"] for w in self.worlds.values()}) for w in self.worlds.values(): w["milestones"]._gitea = router agent = LoopAgent( self.world["store"], transport, router, # type: ignore[arg-type] projects=[ Project( name=name, repo=w["repo_name"], workspace=Workspace(w["host"], w["repo"].root, timeout_s=30), milestones=w["milestones"], ) for name, w in self.worlds.items() ], account="nisse-dev", operators={DON}, gitea_operators={"don"}, driver=Driver( PolicyModel(self.rng, list(self.worlds)), # type: ignore[arg-type] "m", cancel=cancel, ), worker=InlineWorker(), cancel=cancel, oplog=OpLog("nisse-dev", _Null()), reader=self, # type: ignore[arg-type] helpers=Helpers( client=HelperModel(self.rng), # type: ignore[arg-type] roles=RoleMap(executor="", planner="m", helper="h", expert="x"), caps=Caps(turns=4), ), bug_grace_s=600, clock=lambda: self.now, ) transport.on_message(agent.on_message) return agent, transport def agrees(self, proposal: str, reply: str, model: str = "") -> bool: return self.reads_yes(proposal, reply) def items(self) -> list[Any]: return self.world["store"].work_items() def name(self, item: Any) -> str: """How the operator names an item: by project when there are two.""" issue = loop_of(item).issue return f"#{issue}" if len(self.worlds) == 1 else f"{item.scope}#{issue}" def step(self) -> None: 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, "issue": 1, "chat": 1, } item = self.rng.choice(open_items) if open_items else None loop = loop_of(item) if item else None # The forge an item's project lives on; a new issue picks one. project = ( item.scope if item and item.scope != "general" else self.rng.choice(list(self.worlds)) ) forge: Forge = self.worlds[project]["forge"] if item and loop: weights.update({"go": 3, "answer": 2, "abandon": 1, "go-stale": 1}) weights.update({"plain-yes": 2, "wait": 2, "question": 2}) weights.update({"status-here": 1}) if loop.new_project: weights.update({"done": 2}) if item.scope != "general": # conversations aren't named by number weights.update({"first": 1, "named-go": 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(f"{action}{self.name(item)}" if item else action) if action == "intake": self.agent.intake() elif action == "done" and loop: # The operator says a new project is ready; no entry was added. self.transport.receive(DON, "done", thread=loop.thread) elif action == "status-here" and loop: # L5: a command about the item, not a reply: nothing changes, # not even a countdown. before = (item.position, loop.wait, loop.rev, loop.grace_until) self.transport.receive(DON, "status", thread=loop.thread) after_item = self.world["store"].work_item(item.id) after = loop_of(after_item) now = (after_item.position, after.wait, after.rev, after.grace_until) assert now == before, ( f"status_changes_nothing: {before} -> {now}: " f"seed {self.seed}, steps {self.steps}" ) elif action == "question" and loop: # L5: a question never changes the item it's asked about. before = (item.position, loop.wait, loop.rev, loop.pr, loop.branch) self.transport.receive(DON, f"why {len(self.steps)}?", thread=loop.thread) after_item = self.world["store"].work_item(item.id) after = loop_of(after_item) if ( item.scope != "general" and item.position == WAITING and loop.wait in (GO_AHEAD, ANSWER, MERGE) ): now = ( after_item.position, after.wait, after.rev, after.pr, after.branch, ) assert now == before, ( f"questions_change_nothing: {before} -> {now}: " f"seed {self.seed}, steps {self.steps}" ) elif action == "chat": # A top-level message: a conversation of its own (L1, L5). self.transport.receive(DON, f"thinking aloud {len(self.steps)}") elif action == "issue": # A new issue, sometimes depending on one that is still open. project = self.rng.choice(list(self.worlds)) forge = self.worlds[project]["forge"] # Numbered by the forge, as filed records are: never reused. n = max([i.number for i in forge.issues] + [14]) + 1 labels = self.rng.choice([(), (), ("bug",), ("discuss",), ("design",)]) forge.issues.append(Issue(n, f"task {n}", f"do task {n}", "don", labels)) 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 == "plain-yes" and loop: # Read as a go-ahead or not; either way bound to what's up now. if item.position == WAITING and loop.wait == GO_AHEAD: self.shown_at_go.append(f"{item.id}:{loop.rev}") self.transport.receive(DON, "yep, that's right", thread=loop.thread) elif action == "named-go" and loop: if item.position == WAITING and loop.wait == GO_AHEAD: self.shown_at_go.append(f"{item.id}:{loop.rev}") self.transport.receive(DON, f"go {self.name(item)}") elif action == "wait": # Time passes: bug countdowns may run out, conversations idle. self.now += self.rng.choice([60, 700, 90000]) for other in open_items: o = loop_of(other) if ( other.position == WAITING and o.wait == GO_AHEAD and 0 < o.grace_until <= self.now ): self.shown_at_go.append(f"{other.id}:{o.rev}") self.agent.intake() elif action == "first" and loop: self.agent.handle(InboundMessage(DON, f"do {self.name(item)} next"), None) elif action == "crash": # Volatile state is lost: a new agent over the same store. self.agent, self.transport = self._agent() self.agent.boot() self.agent._schedule() 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(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. self.agent.handle( InboundMessage(DON, "go", thread=loop.thread), about=loop.rev - 1 ) elif action == "comment" and loop: forge.comments.setdefault(loop.pr, []).append( Comment(len(self.steps) * 10, "don", f"comment {len(self.steps)}", "t") ) self.agent.watch() self.agent._schedule() elif ( action == "comment-merge" and loop and forge.prs[loop.pr]["state"] == "open" ): # The race: a comment is seen, then the PR merges before the # driver's fix lands. forge.comments.setdefault(loop.pr, []).append( Comment(len(self.steps) * 10, "don", f"comment {len(self.steps)}", "t") ) self.agent.watch() forge.merge(loop.pr) self.agent._schedule() elif action == "merge" and loop and forge.prs[loop.pr]["state"] == "open": forge.merge(loop.pr) self.agent.watch() self.agent._schedule() elif action == "close-pr" and loop and forge.prs[loop.pr]["state"] == "open": forge.prs[loop.pr]["state"] = "closed" self.agent.watch() self.agent._schedule() # -- the properties (specs/loop.qnt) ------------------------------------- def check(self) -> None: where = f"seed {self.seed}, steps {self.steps}" store = self.world["store"] items = self.items() issues = [(i.scope, loop_of(i).issue) for i in items if loop_of(i).issue] assert len(issues) == len(set(issues)), f"one_item_per_issue: {where}" for item in items: loop = loop_of(item) if item.status == "open" and item.position == WAITING: explained = ( (loop.wait == CHAT and item.scope == "general") or (loop.wait == PROVISION and bool(loop.new_project)) or (loop.wait == GO_AHEAD and loop.rev > 0) or (loop.wait == ANSWER and bool(loop.note)) or (loop.wait == MERGE and loop.pr > 0) ) assert explained, ( f"waits_are_explainable: {item.position} {loop}: {where}" ) gos = [ int(f.event.text.split("rev-")[1].split()[0]) for f in store.transcript(item.id, "orchestrator").read() if isinstance(f.event, Fact) and f.event.text.startswith("go-ahead on rev-") ] assert all(f"{item.id}:{rev}" in self.shown_at_go for rev in gos), ( f"go_binds: {item.id} {gos} not in {self.shown_at_go}: {where}" ) if item.scope == "general": continue # no forge: nothing below applies forge = self.worlds[item.scope]["forge"] repo = self.worlds[item.scope]["repo_name"] # 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(repo, 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}" ) if item.status == "closed": assert not any(p["state"] == "open" for p in mine), ( f"closed_items_leave_nothing (PR): {where}" ) assert not [ i for i in store.pending_intents() if i.work_item == item.id ], f"closed_items_leave_nothing (intent): {where}" for item in items: # D1: a bug's grace period runs out only on an unanswered # restatement: any reply since it went up stops the countdown. replied = False for e in store.transcript(item.id, "orchestrator").read(): text = getattr(e.event, "text", "") if isinstance(e.event, Fact) and text.startswith("restatement rev-"): replied = False elif isinstance(e.event, UserMessage) and e.actor == "operator": replied = replied or e.event.source.startswith("transport:") elif text.endswith("from the bug grace period"): assert not replied, ( f"grace_only_unanswered: #{loop_of(item).issue}: {where}" ) for item in items: # D1: a `discuss` item talks before it restates. if "discuss" not in loop_of(item).labels: continue events = store.transcript(item.id, "driver").read() plans = [e.id for e in events if isinstance(e.event, PlanSubmitted)] spoke = [ e.id for e in events if e.actor == "operator" and isinstance(e.event, UserMessage) and e.event.source.startswith("transport:") ] assert not plans or (spoke and spoke[0] < plans[0]), ( f"discuss_talks_first: #{loop_of(item).issue}: {where}" ) assert not [i for i in items if i.position == RUNNING], ( f"one_active (nothing left running): {where}" ) for item in items: if item.scope == "general": loop = loop_of(item) # L1: a conversation has no workspace, milestone or PR. assert not (loop.prepared or loop.milestone or loop.pr), ( f"conversations_do_no_work: {item.id}: {where}" ) # L9: a delegation never outlives its step: merged back or discarded. assert not [ i for i in store.pending_intents() if i.kind == DELEGATION_INTENT ], f"delegations_settle (intent): {where}" for name, w in self.worlds.items(): helping = [ b for b in [ *w["repo"].local_branches(), *_git( w["bare"], "for-each-ref", "--format=%(refname:short)" ).split(), ] if "-help-" in b ] assert not helping, f"delegations_settle (branch {helping}): {where}" self._records_have_one_item(name, w, where) self._own_repository_only(name, w, where) self._parked_work_safe(w, name, where) self._milestones_start_fresh(w, where) self._review_fixes_land(w, where) def _records_have_one_item(self, name: str, world: Any, where: str) -> None: """D1: a record issue filed from a conversation is its item's, from the moment it exists, and intake never takes it for another.""" store = self.world["store"] for issue in world["forge"].issues: found = re.search(r"", issue.body or "") if not found: continue item = store.work_item(found.group(1)) assert item is not None and item.scope == name, ( f"records_have_one_item: {name}#{issue.number} has no item: {where}" ) assert loop_of(item).issue == issue.number, ( f"records_have_one_item: {item.id} isn't #{issue.number}: {where}" ) assert loop_of(item).started, f"filed work starts agreed: {where}" def _own_repository_only(self, name: str, world: Any, where: str) -> None: """D8: an item's branches and PRs reach only its own project's repository.""" others = [i.id for i in self.items() if i.scope != name] heads = [p["head"] for p in world["forge"].prs.values()] branches = _git( world["bare"], "for-each-ref", "--format=%(refname:short)", "refs/heads" ).split() for ref in [*heads, *branches]: assert not any(ref.startswith(f"{i}-") for i in others), ( f"own_repository_only: {ref} in {name}: {where}" ) def _review_fixes_land(self, world: Any, where: str) -> None: forge: Forge = world["forge"] heads = [p["head"] for p in forge.prs.values()] assert len(heads) == len(set(heads)), f"one_pr_per_milestone: {where}" bare = world["bare"] for n, pr in forge.prs.items(): if not pr["merged"]: continue head = _rev(bare, f"refs/heads/{pr['head']}") # A fix pushed after the merge must reach the operator as a PR # (loop.qnt's lostFix): in main, or in a later PR, whatever they # then decide about it. landed = _is_ancestor(bare, head, "refs/heads/main") or any( _is_ancestor(bare, head, f"refs/heads/{other['head']}") for m, other in forge.prs.items() if m > n ) assert landed, f"review_fixes_land (PR {n}): {where}" def _milestones_start_fresh(self, world: Any, where: str) -> None: """D4: a milestone starts from the latest default branch, so one redone after a closed PR never carries the rejected work.""" forge: Forge = world["forge"] bare = world["bare"] for item in self.items(): mine = sorted( (n, p) for n, p in forge.prs.items() if p["head"].startswith(f"{item.id}-milestone-") and "-fix-" not in p["head"] ) for i, (n, earlier) in enumerate(mine): if earlier["merged"]: continue rejected = _rev(bare, f"refs/heads/{earlier['head']}") if _is_ancestor(bare, rejected, "refs/heads/main"): continue for m, later in mine[i + 1 :]: assert not _is_ancestor( bare, rejected, f"refs/heads/{later['head']}" ), f"milestones_start_fresh: PR {m} carries PR {n}: {where}" def _parked_work_safe(self, world: Any, name: str, 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 its project's forge.""" root, bare = world["repo"].root, world["bare"] current = _git(root, "rev-parse", "--abbrev-ref", "HEAD") ids = [i.id for i in self.items() if i.status == "open" and i.scope == name] 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" or item.scope != name: continue written = { str(e.event.arguments.get("path")) for e in self.world["store"].transcript(item.id, "driver").read() # The driver's own: a failed helper's are discarded on purpose. if isinstance(e.event, ToolCall) and e.event.tool == "write_file" and e.actor == "driver" } here = current.startswith(f"{item.id}-") for path in written: if here and (root / path).exists(): continue history = _git( # Full history: a parked stash keeps new files in its # third parent, which plain log simplifies away. bare, "log", "--all", "--full-history", "--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: ... def flush(self) -> None: ... def _rev(bare: Path, ref: str) -> str: return subprocess.run( ["git", "--git-dir", str(bare), "rev-parse", ref], capture_output=True, text=True, check=True, ).stdout.strip() def _is_ancestor(bare: Path, commit: str, ref: str) -> bool: return ( subprocess.run( ["git", "--git-dir", str(bare), "merge-base", "--is-ancestor", commit, ref], capture_output=True, check=False, ).returncode == 0 ) def _seeded_ids(monkeypatch: pytest.MonkeyPatch, seed: int) -> None: """Item IDs and branch names come from uuid4: seeded too, so a walk replays exactly from its seed.""" ids = random.Random(seed) monkeypatch.setattr(uuid, "uuid4", lambda: uuid.UUID(int=ids.getrandbits(128))) @pytest.mark.parametrize("seed", SEEDS) def test_the_loop_keeps_its_properties( world: dict[str, Any], seed: int, monkeypatch: pytest.MonkeyPatch ) -> None: _seeded_ids(monkeypatch, seed) walk = Walk({"sandbox": world}, seed) for _ in range(STEPS): walk.step() walk.check() @pytest.mark.parametrize("seed", TWO_PROJECT_SEEDS) def test_two_projects_keep_the_properties( ssh_host: SSHHost, tmp_path: Path, seed: int, monkeypatch: pytest.MonkeyPatch ) -> None: """D8: the same walk over two projects, whose issue numbers collide.""" _seeded_ids(monkeypatch, seed) sandbox = make_world(ssh_host, tmp_path / "sandbox") other = make_world( ssh_host, tmp_path / "other", store=sandbox["store"], repo_name="don/other" ) walk = Walk({"sandbox": sandbox, "other": other}, seed) for _ in range(STEPS): walk.step() walk.check()