dongarandClaude Opus 5.5
fe4548459f
A project's notes have their own folder, memory/nisse-dev/projects/<project>/, with its own index; remember and forget take scope (general, or project: always the current item's own, so no item writes another project's notes, and a conversation has none). At an item's start the transcript gets the general index and its project's, never another's. Index links are pages from the vault's root, as SilverBullet names them. The closing pass's memory_done can name repository documents the item left out of date; the harness says so in the thread and records it. The prompt: the repository is the authority, and a note that disagrees is corrected. With the operator: ARCHITECTURE.md isn't put into every item's context (its size); the infrastructure world model waits for nisse-ops. The walk writes project notes and checks they come only from that project's items. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
1086 lines
46 KiB
Python
1086 lines
46 KiB
Python
"""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 json
|
|
import os
|
|
import random
|
|
import re
|
|
import shutil
|
|
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 (
|
|
ActionVerdict,
|
|
AssistantText,
|
|
Fact,
|
|
PlanSubmitted,
|
|
ToolCall,
|
|
ToolResult,
|
|
UserMessage,
|
|
)
|
|
from nisse_dev.delegation import DELEGATION_INTENT, Caps
|
|
from nisse_dev.execution import HostUnavailable, SSHHost
|
|
from nisse_dev.gitea import Comment, Issue
|
|
from nisse_dev.llm import LLMError, RoleMap
|
|
from nisse_dev.loop.agent import LoopAgent, Project
|
|
from nisse_dev.loop.classifier import ActionClassifier
|
|
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.memory import Memory
|
|
from nisse_dev.oplog import OpLog
|
|
from nisse_dev.tools import Workspace
|
|
from nisse_dev.transport import InboundMessage, InMemoryTransport
|
|
from nisse_dev.vault import Vault, split_frontmatter
|
|
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
|
|
|
|
|
|
# Sends data out (dangerous), but nothing listens on the port.
|
|
DANGEROUS_COMMAND = "curl -sS -m 1 -X POST -d x http://127.0.0.1:9/"
|
|
# No rule knows it (inline Python), and it does nothing.
|
|
UNKNOWN_COMMAND = "python3 -c 'pass'"
|
|
|
|
|
|
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 _remember(self) -> Any:
|
|
scope = _scope(self.rng)
|
|
return call(
|
|
"remember", name=f"n{self.n % 4}", description="d", fact="f", scope=scope
|
|
)
|
|
|
|
def _forget(self) -> Any:
|
|
scope = _scope(self.rng)
|
|
return call("forget", name=f"n{self.n % 4}", why="superseded", scope=scope)
|
|
|
|
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"
|
|
if "memory_done" in offered:
|
|
# The memory pass at a close (M5.2): a note, a forgetting, done.
|
|
roll = self.rng.random()
|
|
if roll < 0.3:
|
|
return [self._remember()]
|
|
if roll < 0.55:
|
|
return [self._forget()]
|
|
return [call("memory_done", summary="done")]
|
|
if "remember" in offered and self.rng.random() < 0.06:
|
|
if self.rng.random() < 0.4:
|
|
return [self._forget()]
|
|
return [self._remember()]
|
|
if "toolchain_ready" in offered:
|
|
# A preflight (M4.1): usually reported, sometimes badly or not.
|
|
if self.rng.random() < 0.15:
|
|
return [AssistantText(text="setting up", model=model)]
|
|
command = self.rng.choice(["python3 --version"] * 4 + ["python3 -c x"])
|
|
return [
|
|
call(
|
|
"toolchain_ready",
|
|
summary=f"ready {self.n}",
|
|
toolchain=[{"name": "python", "version_command": command}], # type: ignore[arg-type]
|
|
)
|
|
]
|
|
if "request_reimage" in offered and self.rng.random() < 0.03:
|
|
# M4.3: the build VM seems broken past the usual fixes.
|
|
return [call("request_reimage", reason=f"broken {self.n}")]
|
|
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 "toolchain_step" in offered and self.rng.random() < 0.06:
|
|
# M4.2: the work asks for a toolchain change.
|
|
return [
|
|
call("start_milestone", name=f"m{self.n}"),
|
|
call("toolchain_step", reason=f"needs a tool {self.n}"),
|
|
]
|
|
if "delegate" in offered and self.rng.random() < 0.1:
|
|
return [call("delegate", task=f"help {self.n}")]
|
|
allowed = [t for t in facts[-3:] if "allowed this once" in t]
|
|
if allowed:
|
|
# Told to, as the driver is: run it again exactly as before.
|
|
said = allowed[-1].split("allowed this once: run_command ", 1)[-1]
|
|
command = json.loads(said.rsplit(". Run it again", 1)[0])["command"]
|
|
return [call("run_command", command=command)]
|
|
if self.rng.random() < 0.2:
|
|
# No rule knows it: the classifier judges (L4).
|
|
return [
|
|
call("start_milestone", name=f"m{self.n}"),
|
|
call("run_command", command=UNKNOWN_COMMAND),
|
|
]
|
|
if self.rng.random() < 0.12:
|
|
# Dangerous, always the same, so an allow can match it (P3, L3).
|
|
return [
|
|
call("start_milestone", name=f"m{self.n}"),
|
|
call("run_command", command=DANGEROUS_COMMAND),
|
|
]
|
|
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"]),
|
|
),
|
|
]
|
|
|
|
|
|
def _scope(rng: random.Random) -> str:
|
|
return rng.choice(["general", "project"])
|
|
|
|
|
|
class Watched:
|
|
"""The driver's model, watched: a run on a project item (not its
|
|
preflight) must find the project's marker on the VM, and no model runs
|
|
for an item whose build VM is blocked."""
|
|
|
|
def __init__(self, walk: "Walk", inner: Any) -> None:
|
|
self.walk = walk
|
|
self.inner = inner
|
|
|
|
def complete(self, events, tools, model, system=None) -> list[Any]:
|
|
offered = {t["function"]["name"] for t in tools}
|
|
item = self.walk.world["store"].active_item()
|
|
# Working on the VM: offered its tools (an answer given while the VM
|
|
# is blocked isn't, #137), and not setting it up.
|
|
working = (
|
|
"read_file" in offered
|
|
and "toolchain_ready" not in offered
|
|
and "propose_work" not in offered
|
|
)
|
|
marked = (
|
|
item is not None and (self.walk.markers / f"{item.scope}.ready").exists()
|
|
)
|
|
if item is not None and item.scope != "general" and working and not marked:
|
|
self.walk.unready.append(item.id)
|
|
if (
|
|
item is not None
|
|
and item.scope != "general"
|
|
and "read_file" in offered
|
|
and self.walk.down
|
|
):
|
|
self.walk.while_down.append(item.id)
|
|
if (
|
|
item is not None
|
|
and "read_file" in offered
|
|
and item.scope in self.walk.worlds
|
|
):
|
|
host = self.walk.worlds[item.scope]["host"].describe()
|
|
if host in self.walk.world["store"].vm_blocks():
|
|
self.walk.on_blocked.append(item.id)
|
|
return self.inner.complete(events, tools, model, system)
|
|
|
|
|
|
class ClassifierModel:
|
|
"""A classifier answering at random, junk and failures included: the
|
|
loop must stay safe whatever it says (L4 fails closed)."""
|
|
|
|
def __init__(self, rng: random.Random) -> None:
|
|
self.rng = rng
|
|
|
|
def complete(self, events, tools, model, system=None) -> list[Any]:
|
|
roll = self.rng.random()
|
|
if roll < 0.1:
|
|
raise LLMError("classifier down")
|
|
if roll < 0.2:
|
|
return [AssistantText(text="hmm", model=model)]
|
|
label = self.rng.choice(["safe", "in_scope", "needs_discussion", "maybe"])
|
|
return [
|
|
ToolCall(
|
|
call_id=f"v{roll}",
|
|
tool="verdict",
|
|
arguments={"verdict": label, "reason": f"judged {label}"},
|
|
)
|
|
]
|
|
|
|
|
|
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
|
|
if self.rng.random() < 0.4:
|
|
return [
|
|
call(
|
|
"run_command",
|
|
command=self.rng.choice([DANGEROUS_COMMAND, UNKNOWN_COMMAND]),
|
|
)
|
|
]
|
|
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,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> 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
|
|
# The trust dial (M3.3): half the walks have the agent merge approved
|
|
# PRs, behind protection that refuses anything else.
|
|
self.merge = "approved" if seed % 2 == 0 else "operator"
|
|
self.agent_merges: list[tuple[int, bool]] = [] # (PR, approved at head)
|
|
# M4.1: markers in the walk's own directory; project runs seen
|
|
# without one.
|
|
self.markers = Path(self.world["repo"].root).parent / "markers"
|
|
self.unready: list[str] = []
|
|
# #137: the build VM out of reach, now or after a few more commands
|
|
# (so mid-run); model calls made while it was.
|
|
self.down = False
|
|
self.drop_in = 0
|
|
self.while_down: list[str] = []
|
|
self.checking = False # the walk's own checks see the real VM
|
|
# M5.2: the agent's memory, in a vault that also holds the
|
|
# operator's notes; nothing may land outside its own folder.
|
|
vault_root = Path(self.world["repo"].root).parent / "vault"
|
|
(vault_root / "notes").mkdir(parents=True, exist_ok=True)
|
|
(vault_root / "notes" / "Ops.md").write_text("the operator's\n")
|
|
self.memory = Memory(
|
|
Vault(vault_root, {"memory/nisse-dev": "rw", "notes": "ro"}),
|
|
"memory/nisse-dev",
|
|
)
|
|
real = SSHHost.execute
|
|
walk = self
|
|
|
|
def execute(host: SSHHost, command: str, **kw: Any) -> Any:
|
|
if walk.checking:
|
|
return real(host, command, **kw)
|
|
if walk.drop_in:
|
|
walk.drop_in -= 1
|
|
walk.down = walk.down or walk.drop_in == 0
|
|
if walk.down:
|
|
raise HostUnavailable("ssh: connect to host: Connection timed out")
|
|
return real(host, command, **kw)
|
|
|
|
monkeypatch.setattr(SSHHost, "execute", execute)
|
|
self.wiped_at: dict[str, int] = {} # project -> last event before a wipe
|
|
self.on_blocked: list[str] = [] # items run while their VM was blocked
|
|
if self.merge == "approved":
|
|
for w in worlds.values():
|
|
self._watch_merges(w["forge"])
|
|
self.agent, self.transport = self._agent()
|
|
|
|
def _watch_merges(self, forge: Forge) -> None:
|
|
"""Protected, and each merge the agent asks for recorded with
|
|
whether an operator had approved the PR as it was."""
|
|
forge.protect()
|
|
merge = forge.merge_pull_request
|
|
|
|
def recorded(repo: str, index: int) -> None:
|
|
if not forge.prs[index]["head"].startswith("nisse-check-"):
|
|
head = forge.head_sha(index)
|
|
approved = any(
|
|
a.author == "don" and a.commit == head and not a.stale
|
|
for a in forge.approved.get(index, [])
|
|
)
|
|
self.agent_merges.append((index, approved))
|
|
merge(repo, index)
|
|
|
|
forge.merge_pull_request = recorded # type: ignore[method-assign]
|
|
|
|
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"],
|
|
merge=self.merge,
|
|
)
|
|
for name, w in self.worlds.items()
|
|
],
|
|
account="nisse-dev",
|
|
operators={DON},
|
|
gitea_operators={"don"},
|
|
driver=Driver(
|
|
Watched(self, 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(driver="m", helper="h", expert="x"),
|
|
caps=Caps(turns=4),
|
|
),
|
|
classifier=ActionClassifier(
|
|
ClassifierModel(self.rng), # type: ignore[arg-type]
|
|
"c",
|
|
),
|
|
bug_grace_s=600,
|
|
clock=lambda: self.now,
|
|
preflight=True,
|
|
preflight_turns=3,
|
|
markers=str(self.markers),
|
|
vault=self.memory._vault,
|
|
memory=self.memory,
|
|
memory_turns=3,
|
|
)
|
|
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,
|
|
"wipe": 1,
|
|
"issue": 1,
|
|
"chat": 1,
|
|
}
|
|
if self.world["store"].vm_blocks():
|
|
weights.update({"vm-ready": 3})
|
|
weights.update({"vm-up": 3} if self.down else {"vm-down": 1})
|
|
item = self.rng.choice(open_items) if open_items else None
|
|
loop = loop_of(item) if item else None
|
|
# The forge an item's project lives on; a new issue picks one.
|
|
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, "allow": 3 if loop.pending_action else 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}
|
|
)
|
|
if self.merge == "approved":
|
|
weights.update({"approve": 4})
|
|
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 == "wipe":
|
|
self._wipe(project)
|
|
elif action == "vm-down":
|
|
# #137: out of reach now, or partway through what comes next.
|
|
if self.rng.random() < 0.5:
|
|
self.down = True
|
|
else:
|
|
self.drop_in = self.rng.randint(1, 30)
|
|
elif action == "vm-up":
|
|
self.down, self.drop_in = False, 0
|
|
elif action == "vm-ready":
|
|
# M4.3: the operator recovers a blocked VM, reimaged or not; a
|
|
# reimage wipes every project on it.
|
|
blocks = self.world["store"].vm_blocks()
|
|
on_it = [
|
|
name
|
|
for name, w in self.worlds.items()
|
|
if w["host"].describe() in blocks
|
|
]
|
|
if self.rng.random() < 0.5:
|
|
for name in on_it:
|
|
self._wipe(name)
|
|
self.transport.receive(DON, f"vm ready {self.rng.choice(on_it)}")
|
|
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 == "allow" and loop:
|
|
self.transport.receive(DON, "allow", thread=loop.thread)
|
|
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 == "approve" and loop and forge.prs[loop.pr]["state"] == "open":
|
|
# An operator's approval, or someone else's; of the PR as it is,
|
|
# or of an earlier commit.
|
|
forge.approve(
|
|
loop.pr,
|
|
by=self.rng.choice(["don", "don", "mallory"]),
|
|
sha=self.rng.choice(["", "0" * 40]),
|
|
)
|
|
self.agent.watch()
|
|
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()
|
|
|
|
def _wipe(self, project: str) -> None:
|
|
"""The build VM is rebuilt (D3): the clone and its marker gone."""
|
|
w = self.worlds[project]
|
|
shutil.rmtree(w["repo"].root, ignore_errors=True)
|
|
events = self.world["store"].all_events()
|
|
self.wiped_at[project] = events[-1].id if events else 0
|
|
(self.markers / f"{project}.ready").unlink(missing_ok=True)
|
|
|
|
# -- the properties (specs/loop.qnt) -------------------------------------
|
|
|
|
def _dangerous_only_when_allowed(self, where: str) -> None:
|
|
"""P3, L3: a dangerous action runs only on an operator's allow, one
|
|
allow per run of it, and is otherwise never run."""
|
|
store = self.world["store"]
|
|
for item in store.work_items():
|
|
events = store.transcript(item.id, "orchestrator").read()
|
|
allows = 0
|
|
held: set[str] = set() # dangerous calls that weren't run
|
|
for e in events:
|
|
event = e.event
|
|
if isinstance(event, UserMessage) and event.text == "allow":
|
|
allows += 1
|
|
elif isinstance(event, ActionVerdict) and event.effect == "dangerous":
|
|
if event.source == "classifier" and event.rule == "in_scope":
|
|
assert event.decision == "ran", where
|
|
elif event.decision == "allowed":
|
|
allows -= 1
|
|
assert allows >= 0, (
|
|
f"dangerous_only_when_allowed: {item.id}: {where}"
|
|
)
|
|
else:
|
|
assert event.decision in ("asked", "refused"), where
|
|
held.add(event.call_id)
|
|
elif isinstance(event, ToolResult) and event.call_id in held:
|
|
assert event.result.startswith("not run"), (
|
|
f"dangerous_only_when_allowed: {item.id} ran "
|
|
f"{event.call_id} unallowed: {where}"
|
|
)
|
|
|
|
def check(self) -> None:
|
|
self.checking = True
|
|
try:
|
|
self._check()
|
|
finally:
|
|
self.checking = False
|
|
|
|
def _check(self) -> None:
|
|
where = f"seed {self.seed}, steps {self.steps}"
|
|
# M4.1: the driver never works on a project whose VM has no marker.
|
|
assert not self.unready, f"ready_before_work: {self.unready}: {where}"
|
|
# M5.2: the agent writes only its own memory folder.
|
|
root = self.memory._vault.root
|
|
strays = [
|
|
str(p.relative_to(root))
|
|
for p in root.rglob("*")
|
|
if p.is_file()
|
|
and not p.is_relative_to(root / "memory" / "nisse-dev")
|
|
and p != root / "notes" / "Ops.md"
|
|
]
|
|
assert not strays, f"memory_stays_home: {strays}: {where}"
|
|
# M5.3: a project's notes come only from that project's items.
|
|
projects = root / "memory" / "nisse-dev" / "projects"
|
|
for note in projects.rglob("*.md") if projects.exists() else []:
|
|
if note.name == "index.md":
|
|
continue
|
|
project = note.relative_to(projects).parts[0]
|
|
front, _ = split_frontmatter(note.read_text())
|
|
source = front["provenance"]["source"]
|
|
assert front["provenance"]["project"] == project, where
|
|
assert source.startswith(f"work item {project}-"), (
|
|
f"project_notes_stay_with_their_project: {note}: {source}: {where}"
|
|
)
|
|
assert (root / "notes" / "Ops.md").read_text() == "the operator's\n", where
|
|
# #137: no model works on a project whose build VM is out of reach.
|
|
assert not self.while_down, f"no_model_on_a_lost_vm: {self.while_down}: {where}"
|
|
# M4.3: nothing runs for a project whose build VM is blocked.
|
|
assert not self.on_blocked, (
|
|
f"blocked_vm_runs_nothing: {self.on_blocked}: {where}"
|
|
)
|
|
# D5: the agent asks Gitea to merge only what an operator approved
|
|
# as it is (Gitea would refuse the rest; the harness mustn't ask).
|
|
unapproved = [pr for pr, approved in self.agent_merges if not approved]
|
|
assert not unapproved, f"merges_only_approved: {unapproved}: {where}"
|
|
self._dangerous_only_when_allowed(where)
|
|
store = self.world["store"]
|
|
items = self.items()
|
|
# Open items: an issue assigned again after its item closed gets a new
|
|
# one (D1), which loop.qnt doesn't model.
|
|
issues = [
|
|
(i.scope, loop_of(i).issue)
|
|
for i in items
|
|
if loop_of(i).issue and i.status == "open"
|
|
]
|
|
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 [
|
|
# A wiped VM has no clone (and so no local branches).
|
|
*(w["repo"].local_branches() if w["repo"].exists() else []),
|
|
*_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"<!-- nisse-item: ([\w-]+) -->", 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"]
|
|
cloned = world["repo"].exists() # not while a wiped VM waits for a run
|
|
current = _git(root, "rev-parse", "--abbrev-ref", "HEAD") if cloned else ""
|
|
ids = [i.id for i in self.items() if i.status == "open" and i.scope == name]
|
|
heads = (
|
|
_git(root, "for-each-ref", "--format=%(refname:short)", "refs/heads")
|
|
if cloned
|
|
else ""
|
|
)
|
|
for line in 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
|
|
events = self.world["store"].transcript(item.id, "driver").read()
|
|
# Only writes that happened: one cut off by a lost VM (#137) didn't.
|
|
done = {
|
|
e.event.call_id
|
|
for e in events
|
|
if isinstance(e.event, ToolResult) and e.event.success
|
|
}
|
|
written = {
|
|
str(e.event.arguments.get("path"))
|
|
for e in events
|
|
# The driver's own: a failed helper's are discarded on purpose.
|
|
if isinstance(e.event, ToolCall)
|
|
and e.event.tool == "write_file"
|
|
and e.actor == "driver"
|
|
and e.event.call_id in done
|
|
# D3: a wiped VM loses what wasn't pushed, accepted.
|
|
and e.id > self.wiped_at.get(name, 0)
|
|
}
|
|
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, monkeypatch)
|
|
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, monkeypatch)
|
|
for _ in range(STEPS):
|
|
walk.step()
|
|
walk.check()
|