Adopt an already-published PR instead of publishing again
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m12s
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m12s
Fixes counterexample 5 from specs/orchestrator.qnt. A crash after the PR was opened but before PR_OPEN was recorded left the agent in EXECUTING; the restart question's 'continue' and 'start' then published again, Gitea refused a second PR, and there was no way back to PR_OPEN. On the live run the executor model happened to notice the existing work; recovery must not depend on that. - Boot (A3 step 4): in EXECUTING, if Gitea shows the milestone's PR open, adopt it, watch it from the start, and resume in PR_OPEN with a notice. Gitea unreachable or the PR closed: the restart question, as before. - MilestonePublisher.publish is idempotent: after pushing, an open PR for the branch is adopted (and so updated) rather than a second one opened. This also covers a PR opened whose response was lost. - The model adopts too, gains the lost-response outcome, and now counts PRs opened per milestone (one_pr_per_milestone was a flag the fixed model could never set). A mutant without adoption fails the new scenario; the properties held over 200,000 traces each. 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:
1 parent
6fdec6922b
commit
8bd1ac0fa2
6 files changed
+165
-24
No files matched your search
@@ -130,6 +130,8 @@ class MilestonePublisher(Protocol):
|
||||
|
||||
def close_pull_request(self, index: int, note: str) -> None: ...
|
||||
|
||||
def published_pull_request(self, task_id: str, milestone: int) -> Any: ...
|
||||
|
||||
|
||||
class PlanBoard(Protocol):
|
||||
"""The Gitea side of plan gating; the concrete one is in
|
||||
@@ -323,6 +325,27 @@ class Orchestrator:
|
||||
"restarted while waiting on you: "
|
||||
+ pending.get("prompt", "my last question still stands")
|
||||
)
|
||||
elif state.state == EXECUTING and (published := self._published_pr()):
|
||||
# A3 step 4: Gitea shows this milestone's PR open, so it was
|
||||
# published and only PR_OPEN went unrecorded. Adopt the PR rather
|
||||
# than ask, and never publish it again. Nothing watched it yet,
|
||||
# so every comment on it is new.
|
||||
milestone = int(self._data.get("milestone", 1))
|
||||
self._data["pr"] = {
|
||||
"number": published.number,
|
||||
"url": published.url,
|
||||
"milestone": milestone,
|
||||
"last_comment": self._watch(
|
||||
"review", published.number, self.handle_review_comment, after=0
|
||||
),
|
||||
}
|
||||
self._unwatch("plan")
|
||||
self._transition(PR_OPEN, cause=f"boot: milestone {milestone} published")
|
||||
self._transport.send_message(
|
||||
f"restarted after milestone {milestone} was published: PR "
|
||||
f"#{published.number} is open ({published.url}); watching it for "
|
||||
"review comments"
|
||||
)
|
||||
elif state.state == PR_OPEN:
|
||||
# Waiting on review is a committed state like a gate (A3 step
|
||||
# 8): resume, and watch the PR again from its cursor.
|
||||
@@ -980,6 +1003,20 @@ class Orchestrator:
|
||||
f"task on {project} abandoned{closed}. Say 'use <project>' to start another"
|
||||
)
|
||||
|
||||
def _published_pr(self) -> Any:
|
||||
"""The open PR for the current milestone, if Gitea shows one; None
|
||||
when there is none or Gitea cannot say (then the restart question
|
||||
is asked as before)."""
|
||||
task_id = self._data.get("task_id")
|
||||
if self._publisher is None or not task_id:
|
||||
return None
|
||||
try:
|
||||
return self._publisher.published_pull_request(
|
||||
str(task_id), int(self._data.get("milestone", 1))
|
||||
)
|
||||
except GiteaError:
|
||||
return None
|
||||
|
||||
def _return_to_pr_open(self, cause: str) -> None:
|
||||
self._transition(PR_OPEN, cause=cause)
|
||||
pr = self._data.get("pr") or {}
|
||||
|
||||
@@ -84,6 +84,19 @@ class MilestonePublisher:
|
||||
cause=f"publishing milestone {milestone}",
|
||||
)
|
||||
self._git.push(self._remote, branch, token=self._token)
|
||||
# Idempotent: if a PR for this branch is already open (a crash after
|
||||
# opening it, or a response lost on the way back), the push above
|
||||
# has updated it; adopt it rather than fail on a second.
|
||||
existing = self.published_pull_request(task_id, milestone)
|
||||
if existing is not None:
|
||||
self._store.commit_intent(
|
||||
intent_id, cause=f"adopted open PR #{existing.number}"
|
||||
)
|
||||
return PublishResult(
|
||||
existing,
|
||||
f"PR #{existing.number} was already open for this milestone; "
|
||||
f"pushed to it: {existing.url}",
|
||||
)
|
||||
pr = self._client.open_pull_request(
|
||||
self._repo,
|
||||
head=branch,
|
||||
@@ -94,6 +107,17 @@ class MilestonePublisher:
|
||||
self._store.commit_intent(intent_id, cause=f"opened PR #{pr.number}")
|
||||
return PublishResult(pr, f"opened PR #{pr.number}: {pr.url}")
|
||||
|
||||
def published_pull_request(
|
||||
self, task_id: str, milestone: int
|
||||
) -> PullRequest | None:
|
||||
"""The open PR for a milestone's branch, if there is one. A closed one
|
||||
does not count: the operator closed it, so a new one may be opened.
|
||||
Raises GiteaError when Gitea cannot say."""
|
||||
pr = self._client.find_pull_request(
|
||||
self._repo, self.branch_for(task_id, milestone)
|
||||
)
|
||||
return pr if pr is not None and pr.state == "open" else None
|
||||
|
||||
def comment(self, index: int, body: str) -> None:
|
||||
"""Reply on the PR. Courtesy only — the pushed commit is the real
|
||||
answer — so a failure here must not derail the review loop."""
|
||||
|
||||
@@ -16,6 +16,7 @@ class FakeGitea:
|
||||
def __init__(self, *, fail: bool = False) -> None:
|
||||
self.opened: list[dict[str, Any]] = []
|
||||
self.existing: list[str] = []
|
||||
self.closed: set[str] = set()
|
||||
self._fail = fail
|
||||
|
||||
def open_pull_request(
|
||||
@@ -33,7 +34,8 @@ class FakeGitea:
|
||||
|
||||
def find_pull_request(self, repo: str, head: str) -> PullRequest | None:
|
||||
if head in self.existing:
|
||||
return PullRequest(number=1, state="open", head=head, url="u")
|
||||
state = "closed" if head in self.closed else "open"
|
||||
return PullRequest(number=1, state=state, head=head, url="u")
|
||||
return None
|
||||
|
||||
|
||||
@@ -212,3 +214,32 @@ def test_a_followup_the_executor_committed_itself_is_pushed(
|
||||
# Nothing new since that push: no second push, and it says so.
|
||||
assert publisher.push_followup(task_id="t", milestone=1, message="m") is None
|
||||
assert len(git.pushed) == 2
|
||||
|
||||
|
||||
def test_publishing_again_adopts_the_open_pr_instead_of_opening_another(
|
||||
store: StateStore, tmp_path: Path
|
||||
) -> None:
|
||||
# E.g. Gitea opened the PR but the response was lost, and the milestone
|
||||
# was published again: the PR is adopted and updated, never duplicated.
|
||||
git = FakeGit(tmp_path / "work")
|
||||
client = FakeGitea()
|
||||
publisher = build(store, git, client)
|
||||
publisher.publish(task_id="t", milestone=1, project="demo", plan="p")
|
||||
(tmp_path / "work" / "more.txt").write_text("more")
|
||||
result = publisher.publish(task_id="t", milestone=1, project="demo", plan="p")
|
||||
assert len(client.opened) == 1
|
||||
assert result.pull_request is not None and result.pull_request.number == 1
|
||||
assert "already open" in result.note
|
||||
assert git.pushed == ["t/milestone-1", "t/milestone-1"] # updated it
|
||||
assert store.pending_intents() == []
|
||||
|
||||
|
||||
def test_a_closed_pr_is_not_adopted(store: StateStore, tmp_path: Path) -> None:
|
||||
git = FakeGit(tmp_path / "work")
|
||||
client = FakeGitea()
|
||||
client.existing.append("t/milestone-1")
|
||||
client.closed.add("t/milestone-1")
|
||||
publisher = build(store, git, client)
|
||||
assert publisher.published_pull_request("t", 1) is None
|
||||
publisher.publish(task_id="t", milestone=1, project="demo", plan="p")
|
||||
assert len(client.opened) == 1 # the operator closed it; a new one opens
|
||||
@@ -134,6 +134,12 @@ class Publisher:
|
||||
PullRequest(number=20, state="open", head=branch, url="u"), "opened PR #20"
|
||||
)
|
||||
|
||||
def published_pull_request(self, task_id: str, milestone: int) -> Any:
|
||||
branch = f"{task_id}/milestone-{milestone}"
|
||||
if branch in self._gitea.prs:
|
||||
return PullRequest(number=20, state="open", head=branch, url="u")
|
||||
return None
|
||||
|
||||
def push_followup(self, **kwargs: Any) -> str | None:
|
||||
return "abc12345"
|
||||
|
||||
@@ -249,7 +255,6 @@ def test_an_afterthought_rejects_the_next_plan(tmp_path: Path) -> None:
|
||||
assert agent.gate().get("rev") == 2 # arch was rev-1; not re-planned
|
||||
|
||||
|
||||
@XFAIL
|
||||
def test_crash_after_opening_the_pr_publishes_again(tmp_path: Path) -> None:
|
||||
"""crashAfterOpeningThePrPublishesAgainTest"""
|
||||
gitea = FakeGitea()
|
||||
@@ -261,9 +266,11 @@ def test_crash_after_opening_the_pr_publishes_again(tmp_path: Path) -> None:
|
||||
assert agent.state == EXECUTING # the PR exists; PR_OPEN was never recorded
|
||||
|
||||
agent = Agent(tmp_path, gitea) # restart: the intent is verified
|
||||
agent.say("continue", "start milestone 1")
|
||||
# There is a PR for this milestone: the agent must end up watching it,
|
||||
# not trying to open another.
|
||||
# There is a PR for this milestone: the agent adopts it on boot and ends
|
||||
# up watching it, never trying to open another.
|
||||
assert agent.state == PR_OPEN
|
||||
assert "PR #20 is open" in agent.transport.outbound[-1].text
|
||||
agent.say("continue", "start milestone 1") # no longer means anything
|
||||
assert agent.publisher.attempts == 0
|
||||
assert agent.state == PR_OPEN
|
||||
|
||||
@@ -321,3 +328,22 @@ def test_the_harness_reaches_pr_open_without_faults(tmp_path: Path) -> None:
|
||||
assert agent.state == PR_OPEN
|
||||
assert agent.publisher.attempts == 1
|
||||
_ = (AWAIT_INPUT, MILESTONE_PLANNING)
|
||||
|
||||
|
||||
def test_boot_asks_as_before_when_gitea_cannot_say(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
gitea = FakeGitea()
|
||||
agent = Agent(tmp_path, gitea)
|
||||
agent.say("use demo", "approve", "approve")
|
||||
agent.publisher.crash_after_opening = True
|
||||
with pytest.raises(SystemExit):
|
||||
agent.say("start milestone 1")
|
||||
|
||||
def unreachable(self: Publisher, task_id: str, milestone: int) -> Any:
|
||||
raise GiteaError("gitea is down")
|
||||
|
||||
monkeypatch.setattr(Publisher, "published_pull_request", unreachable)
|
||||
agent = Agent(tmp_path, gitea)
|
||||
assert agent.state == AWAIT_INPUT
|
||||
assert "restarted mid-EXECUTING" in agent.transport.outbound[-1].text
|
||||
+23
-11
@@ -67,6 +67,7 @@ module orchestrator {
|
||||
prThread: List[PrComment],
|
||||
nextId: int,
|
||||
giteaPr: int -> str, // PR key -> "open" / "closed"
|
||||
opened: int -> int, // PR key -> how many PRs Gitea ever opened for it
|
||||
openIssues: Set[int], // plan issues still open, by task
|
||||
// -- volatile: lost in a crash --
|
||||
queue: List[Job],
|
||||
@@ -83,7 +84,6 @@ module orchestrator {
|
||||
answered: Set[int], // ... and answered (pushed, asked, or reported)
|
||||
badApproval: bool,
|
||||
staleFeedback: bool,
|
||||
dupPublish: bool,
|
||||
}
|
||||
|
||||
// ---- helpers ------------------------------------------------------------
|
||||
@@ -97,6 +97,11 @@ module orchestrator {
|
||||
pure def prIds(s: Sys, key: int): int =
|
||||
s.prThread.foldl(0, (acc, c) => if (c.key == key and c.id > acc) c.id else acc)
|
||||
pure def max(a: int, b: int): int = if (a > b) a else b
|
||||
// Gitea opens a PR for a branch.
|
||||
pure def openPr(s: Sys, key: int): Sys =
|
||||
{ ...s, giteaPr: s.giteaPr.put(key, "open"),
|
||||
opened: s.opened.put(key, (if (s.opened.keys().contains(key)) s.opened.get(key) else 0) + 1) }
|
||||
|
||||
pure def queued(s: Sys, src: str): bool = s.queue.foldl(false, (acc, j) => acc or j.src == src)
|
||||
|
||||
// _await_input
|
||||
@@ -260,6 +265,11 @@ module orchestrator {
|
||||
awaitInput(s0, { kind: "boot", g: "", interrupted: s.st })
|
||||
else if (s.st == "AWAIT_INPUT" and s.pending.kind != "")
|
||||
s0
|
||||
else if (s.st == "EXECUTING" and giteaUp and prState(s, prKeyOf(s)) == "open")
|
||||
// A3 step 4: the milestone's PR is open, so it was published and only
|
||||
// PR_OPEN went unrecorded. Adopt it; nothing watched it, so from 0.
|
||||
{ ...s0, st: "PR_OPEN", prOn: true, prKey: prKeyOf(s), prLast: 0,
|
||||
prWatch: { on: true, floor: 0 }, planWatch: OFF }
|
||||
else if (s.st == "PR_OPEN")
|
||||
{ ...s0, prWatch: if (s.prOn) { on: true, floor: s.prLast } else OFF }
|
||||
else if (s.st == "IDLE" or (SIGNOFF.contains(s.st) and s.gate.on))
|
||||
@@ -278,11 +288,11 @@ module orchestrator {
|
||||
boardOn: false, boardPosted: false, boardLast: 0,
|
||||
intents: Set(),
|
||||
planThread: List(), prThread: List(), nextId: 1,
|
||||
giteaPr: Map(), openIssues: Set(),
|
||||
giteaPr: Map(), opened: Map(), openIssues: Set(),
|
||||
queue: List(), work: "none", workGate: "", workCid: 0, pubStage: 0, cancel: false,
|
||||
planWatch: OFF, prWatch: OFF,
|
||||
shown: NO_PLAN, taken: Set(), answered: Set(),
|
||||
badApproval: false, staleFeedback: false, dupPublish: false,
|
||||
badApproval: false, staleFeedback: false,
|
||||
}
|
||||
|
||||
// Each action has a deterministic form taking its choices as parameters,
|
||||
@@ -372,8 +382,9 @@ module orchestrator {
|
||||
|
||||
// Publishing, one step at a time, so a crash can land between them
|
||||
// (MilestonePublisher.publish and control._publish_milestone):
|
||||
// 0 commit, begin_intent; 1 push, open the PR; 2 commit_intent;
|
||||
// 3 record the PR, watch it, PR_OPEN.
|
||||
// 0 commit, begin_intent; 1 push, then adopt an open PR for the branch or
|
||||
// open one ("lost": Gitea opened it but the response never came back);
|
||||
// 2 commit_intent; 3 record the PR, watch it, PR_OPEN.
|
||||
action publishStep(outcome: str): bool = {
|
||||
val key = prKeyOf(s)
|
||||
all {
|
||||
@@ -389,10 +400,11 @@ module orchestrator {
|
||||
else if (outcome == "fail" or outcome == "nochanges")
|
||||
awaitInput(endWork(s), { kind: "milestone_stopped", g: "", interrupted: "" })
|
||||
else if (prState(s, key) == "open")
|
||||
// Gitea refuses a second PR for the branch: a GiteaError.
|
||||
awaitInput(endWork({ ...s, dupPublish: true }),
|
||||
{ ...s, pubStage: 2 } // adopted, never a second
|
||||
else if (outcome == "lost")
|
||||
awaitInput(endWork(openPr(s, key)),
|
||||
{ kind: "milestone_stopped", g: "", interrupted: "" })
|
||||
else { ...s, giteaPr: s.giteaPr.put(key, "open"), pubStage: 2 }
|
||||
else { ...openPr(s, key), pubStage: 2 }
|
||||
else if (s.pubStage == 2)
|
||||
{ ...s, intents: s.intents.exclude(Set(key)), pubStage: 3 }
|
||||
else
|
||||
@@ -423,7 +435,7 @@ module orchestrator {
|
||||
nondet up = Set(true, false).oneOf()
|
||||
nondet planOutcome = Set("submit", "ask", "error", "stopped", "exception").oneOf()
|
||||
nondet workOutcome = Set("completed", "stopped", "escalate", "exception").oneOf()
|
||||
nondet pubOutcome = Set("ok", "nochanges", "fail", "exception").oneOf()
|
||||
nondet pubOutcome = Set("ok", "nochanges", "fail", "lost", "exception").oneOf()
|
||||
any {
|
||||
send(kind), commentPlan(ckind), commentPr, pollPlan, pollPr, runJob(up),
|
||||
finishPlanning(planOutcome, up), finishExec(workOutcome), publishStep(pubOutcome),
|
||||
@@ -449,8 +461,8 @@ module orchestrator {
|
||||
// Rejection feedback only ever lands on the plan it was written about.
|
||||
val feedback_only_on_what_was_seen = not(s.staleFeedback)
|
||||
|
||||
// Never a second attempt to open a PR for a milestone that has one.
|
||||
val one_pr_per_milestone = not(s.dupPublish)
|
||||
// Never more than one PR opened for a milestone.
|
||||
val one_pr_per_milestone = s.opened.keys().forall(k => s.opened.get(k) <= 1)
|
||||
|
||||
// A review comment the agent took up is answered, or is being worked on.
|
||||
val review_comments_answered =
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
// Counterexamples found in orchestrator.qnt, written out as short traces.
|
||||
// Each ends by expecting the property it breaks to be false; once the design
|
||||
// is fixed, the expectation is flipped and the trace kept as a regression.
|
||||
// Fixed: 1-4 (replies bind to the revision up when they arrived).
|
||||
// Fixed: 1-4 (replies bind to the revision up when they arrived), 5 (boot
|
||||
// and publishing adopt an open PR).
|
||||
//
|
||||
// Run: quint test specs/orchestrator_scenarios.qnt (runs every *Test)
|
||||
module orchestrator_scenarios {
|
||||
@@ -77,22 +78,32 @@ module orchestrator_scenarios {
|
||||
.expect(s.st == "AWAIT_MILESTONE_SIGNOFF" and feedback_only_on_what_was_seen)
|
||||
|
||||
// 5. A crash after the PR is opened but before PR_OPEN is recorded.
|
||||
// Boot finds the intent's PR and marks it done, but the state still
|
||||
// says EXECUTING, so it asks the restart question. 'continue' returns
|
||||
// to the approved plan, and starting it publishes again: Gitea refuses
|
||||
// a second PR for the branch, and there is no path back to PR_OPEN.
|
||||
// Boot found the intent's PR and marked it done, but the state still
|
||||
// said EXECUTING, so it asked the restart question; 'continue' and
|
||||
// 'start' then published again, which Gitea refused, with no path back
|
||||
// to PR_OPEN. Fixed: boot sees the open PR and adopts it.
|
||||
run crashAfterOpeningThePrPublishesAgainTest =
|
||||
toMilestoneApproved
|
||||
.then(send("start")).then(runJob(true))
|
||||
.then(finishExec("completed"))
|
||||
.then(publishStep("ok")) // begin_intent
|
||||
.then(publishStep("ok")) // push, open the PR
|
||||
.then(crash(true)) // intent verified and committed; state EXECUTING
|
||||
.then(send("continue")).then(runJob(true))
|
||||
.then(crash(true)) // adopted on boot
|
||||
.expect(s.st == "PR_OPEN" and s.prOn and one_pr_per_milestone)
|
||||
|
||||
// 5b. Gitea opens the PR but the response is lost; the operator starts
|
||||
// the milestone again. Publishing adopts the open PR.
|
||||
run lostResponseThenRetryAdoptsThePrTest =
|
||||
toMilestoneApproved
|
||||
.then(send("start")).then(runJob(true))
|
||||
.then(finishExec("completed"))
|
||||
.then(publishStep("ok")).then(publishStep("lost"))
|
||||
.then(send("continue")).then(runJob(true)) // back to the approved plan
|
||||
.then(send("start")).then(runJob(true))
|
||||
.then(finishExec("completed"))
|
||||
.then(publishStep("ok")).then(publishStep("ok"))
|
||||
.expect(s.st == "AWAIT_INPUT" and not(one_pr_per_milestone))
|
||||
.then(publishStep("ok")).then(publishStep("ok"))
|
||||
.expect(s.st == "PR_OPEN" and one_pr_per_milestone)
|
||||
|
||||
// 6. A crash while the executor answers a review comment. The comment's
|
||||
// cursor was saved before the work started (at most once), and a
|
||||
|
||||
Reference in new issue
Block a user