Check loop.qnt's properties against the real loop (M2.1, PR D part 1)
CI / checks (push) Skipped

test_loop_properties.py walks the new loop with a seeded mix of operator
and Gitea actions (go-aheads, stale go-aheads, answers, comments, merges,
the comment-then-merge race, closes, abandon, a crash and restart) and an
adaptive scripted model, checking every loop.qnt property after every
step. It found two bugs, both fixed with regression tests: abandoning an
item left its open PR open (the model's fix 1, now in code), and after a
review fix became a follow-up PR, the next fix went to the merged branch.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
This commit is contained in:
dongarandClaude Opus 5.5 committed 2026-10-03 14:19:19 +00:00
1 parent 21744efac1
commit 266b5dbbd7
5 files changed
+318 -4

No files matched your search

+45
View File
@@ -96,6 +96,9 @@ class Forge:
def close_issue(self, repo: str, index: int) -> None:
self.closed_issues.append(index)
def close_pull_request(self, repo: str, index: int) -> None:
self.prs[index]["state"] = "closed"
# what the operator does in Gitea
def merge(self, n: int) -> None:
head = self.prs[n]["head"]
@@ -429,3 +432,45 @@ def test_boot_asks_when_a_publish_never_reached_gitea(world: dict[str, Any]) ->
assert store.pending_intents() == []
assert loop_of(item(world)).wait == ANSWER
assert "publishing was cut off by a restart" in transport.outbound[-1].text
def test_a_fix_after_a_follow_up_pr_goes_to_the_follow_up(
world: dict[str, Any],
) -> None:
# Found by the property walk (seed 7): after a fix became a follow-up PR,
# the next fix was pushed to the original, merged branch and lost.
forge: Forge = world["forge"]
model = ScriptedModel(
RESTATE,
WORK + FINISH,
[call("write_file", path="fix1.py", content="a = 1\n"), *FINISH],
[call("write_file", path="fix2.py", content="b = 2\n"), *FINISH],
)
agent, transport = loop_agent(world, model)
agent.intake()
say(transport, world, "go")
[first] = forge.prs
forge.comments[first] = [Comment(1, "don", "add fix1", "t")]
agent.watch()
forge.merge(first) # merged before the fix lands: a follow-up PR
agent._schedule()
follow = loop_of(item(world)).pr
assert follow != first
forge.comments[follow] = [Comment(2, "don", "add fix2", "t")]
agent.watch()
agent._schedule()
assert len(forge.prs) == 2 # pushed to the follow-up, not a third PR
assert "fix2.py" in forge.files(follow)
def test_abandoning_an_item_closes_its_open_pr(world: dict[str, Any]) -> None:
# Found by the property walk (seed 4); the model's fix 1 (L2).
forge: Forge = world["forge"]
agent, transport = loop_agent(world, ScriptedModel(RESTATE, WORK + FINISH))
agent.intake()
say(transport, world, "go")
[pr] = forge.prs
say(transport, world, "abandon")
assert item(world).outcome == "abandoned"
assert forge.prs[pr]["state"] == "closed"
assert (pr, "abandoned: abandoned by @don:test") in forge.posted