Skip to content

fix(fleet): stop the dispatcher from re-publishing an already-open sweep PR #691

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
swackhamer merged 1 commit into from
Aug 11, 2026
Merged

fix(fleet): stop the dispatcher from re-publishing an already-open sweep PR #691

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 69 additions & 6 deletions scripts/overlord_sweep.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1406,15 +1406,17 @@ def run_sweep(*, repo_root, home, cache_dir, comparison_fn, checkout_fn, lint_fn
config_path=DEFAULT_CONFIG_PATH, cargo_test_workspace_fn=None, create_pr_fn=None,
push_branch_fn=None, fmt_fn=None, run_git=None, now_fn=time.time, log_fn=print,
sweep_state_path=None, quarantine_path=None, sweep_review_log_path=None,
origin_ref=ORIGIN_MAIN, dispatcher_lock_path=None):
origin_ref=ORIGIN_MAIN, dispatcher_lock_path=None, open_sweep_prs_fn=None):
"""One full overlord sweep pass (spec M4). See module docstring for
the step-by-step breakdown. Returns a summary dict whose "status"
is one of: "no_news", "branch_cut_failed", "nothing_merged",
"sweep_aborted" (bisection quarantined every candidate, could not
revert a candidate, or ended without a passing recheck -- see
bisect_sweep_failure), "reattach_failed", "zero_delta" (the assembled
branch is tree-identical to origin_ref: nothing to publish),
"workspace_tests_failed", "push_failed", "pr_create_failed", "ok".
"duplicate_of_open_pr" (tree-identical to an ALREADY-OPEN sweep PR --
see the check below), "workspace_tests_failed", "push_failed",
"pr_create_failed", "ok".
"""
run_git = run_git or default_run_git
cargo_test_workspace_fn = cargo_test_workspace_fn or _real_cargo_test_workspace
Expand Down Expand Up @@ -1669,6 +1671,66 @@ def clear_parks(squads_to_clear):
"failed_squads": failed_squads, "bisection": bisection_result, "preflight": health,
}

# A second idempotency gate, same placement and same reasoning as the
# origin_ref check just above, for the content this repo's own
# "the fleet publishes, it never merges" design leaves sitting on
# origin: as long as nobody merges a sweep PR, origin_ref never
# advances, so the NEXT round's gap detection (which only ever
# compares against origin_ref) rediscovers the exact same gap,
# re-fixes it, and produces a genuinely NEW stamp (new sha, new ts --
# collect_green_stamps correctly treats it as news) whose tree is
# nonetheless byte-identical to a PR still sitting open from an
# earlier round. Measured 2026-08-11: nine open PRs
# (sweep/tags-2026-08-10-1 through sweep/tags-2026-08-11-8), the last
# seven of them with IDENTICAL diffs -- the same three tags
# (Composite:Duration, DjVu:Note, XMP:ComponentsConfiguration)
# re-solved and re-published every round for hours, each one costing
# a full workspace suite, a push and a CI cycle for a reviewer to
# eventually discover duplicates each other.
#
# open_sweep_prs_fn is optional (None in every existing test and any
# caller that predates this check) so this is additive: skip
# entirely rather than fail a sweep over a `gh` hiccup.
if open_sweep_prs_fn is not None:
# origin_ref is "<remote>/<branch>" in production (ORIGIN_MAIN =
# "origin/main") and a bare local branch name ("main") in tests
# that want no real remote at all -- same split every other
# remote-touching test in this suite avoids by construction. Mirror
# that here: only fetch when origin_ref actually names a remote,
# otherwise compare directly against a same-named local ref (what a
# test can set up with a plain `git branch <head> <sha>`).
remote = origin_ref.split("/", 1)[0] if "/" in origin_ref else None
open_prs = open_sweep_prs_fn() or []
for pr in open_prs:
head = pr.get("headRefName") if isinstance(pr, dict) else None
if not head or head == branch:
continue
candidate_ref = head
if remote:
fetch_rc, _out, _err = run_git(
["fetch", remote, f"{head}:refs/remotes/{remote}/{head}"], repo_root)
if fetch_rc != 0:
# The PR's branch may have been deleted, renamed, or this
# worktree's remote may be unreachable this round --
# either way, one unfetchable candidate must cost
# checking that candidate, never the whole duplicate scan.
continue
candidate_ref = f"refs/remotes/{remote}/{head}"
cmp_rc, _out2, _err2 = run_git(["diff", "--quiet", f"{candidate_ref}..HEAD"], repo_root)
if cmp_rc == 0:
pr_ref = pr.get("url") or pr.get("number") or head
log_fn(f"{branch} is tree-identical to already-open {pr_ref} ({head}) -- skipping "
"the workspace suite, the fmt commit, the push and a duplicate PR; advancing "
"the cursor so these stamps do not resurface next round")
durable_squads.update(merge_infos.keys())
persist_cursor(durable_squads)
clear_parks(merge_infos.keys())
return {
"status": "duplicate_of_open_pr", "branch": branch, "duplicate_of": pr_ref,
"merged_squads": sorted(merge_infos), "failed_squads": failed_squads,
"bisection": bisection_result, "preflight": health,
}

all_shas = []
for squad, info in merge_infos.items():
all_shas.extend(commits_contributed(repo_root, info, run_git))
Expand Down Expand Up @@ -1902,10 +1964,11 @@ def comparison_fn(repo, cache_dir, fmt, suffix):
print(f" retry with: gh pr create --head {result.get('branch')} --base main")
elif result.get("pr") is not None:
print(f"PR: {result['pr']}")
# "zero_delta" joins the success set: the branch was tree-identical to
# origin/main, so there was genuinely nothing to publish -- the same
# kind of legitimate no-op as "no_news", not a failure to report.
return 0 if result.get("status") in ("ok", "no_news", "zero_delta") else 1
# "zero_delta" and "duplicate_of_open_pr" join the success set: both
# mean the branch had genuinely nothing NEW to publish (tree-identical
# to origin/main, or to an already-open sweep PR) -- the same kind of
# legitimate no-op as "no_news", not a failure to report.
return 0 if result.get("status") in ("ok", "no_news", "zero_delta", "duplicate_of_open_pr") else 1


if __name__ == "__main__":
Expand Down
22 changes: 16 additions & 6 deletions scripts/parallel_model_fix_loop.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2330,8 +2330,12 @@ def default_sweep_fn(**kwargs):
def comparison_fn(repo, cache_dir, fmt, suffix):
return squad_merge_loop.real_format_match(repo, cache_dir, fmt, suffix)

def open_sweep_prs_fn():
return list_open_sweep_prs(kwargs.get("repo_root"))

return overlord_sweep.run_sweep(
comparison_fn=comparison_fn, checkout_fn=overlord_sweep.real_checkout, **kwargs,
comparison_fn=comparison_fn, checkout_fn=overlord_sweep.real_checkout,
open_sweep_prs_fn=open_sweep_prs_fn, **kwargs,
)


Expand Down Expand Up @@ -2784,8 +2788,9 @@ def auto_publish_round(*, repo_root=REPO_ROOT, cache_dir, home=None, config_path
Returns a summary dict whose "status" is either one of run_sweep's
own statuses passed straight through ("no_news",
"branch_cut_failed", "nothing_merged", "sweep_aborted",
"reattach_failed", "zero_delta", "workspace_tests_failed",
"push_failed", "pr_create_failed"), or one of this function's own:
"reattach_failed", "zero_delta", "duplicate_of_open_pr",
"workspace_tests_failed", "push_failed", "pr_create_failed"), or one
of this function's own:
"no_worktree", "bisection_unverified", "zero_delta", "checks_red",
"checks_timeout", "checks_unknown", "reviews_<state>",
"published_awaiting_review". Every status also carries "adopted":
Expand Down Expand Up @@ -2946,13 +2951,18 @@ def auto_publish_round(*, repo_root=REPO_ROOT, cache_dir, home=None, config_path
# A publish that either landed something or had nothing to land. Every
# other status is a round that did NOT publish, which is what the one-shot
# exit code and the --infinite stall counter both key off.
PUBLISH_OK_STATUSES = frozenset({"published_awaiting_review", "no_news", "zero_delta"})
PUBLISH_OK_STATUSES = frozenset({
"published_awaiting_review", "no_news", "zero_delta", "duplicate_of_open_pr",
})

# Sweep statuses meaning "there was nothing to publish", as opposed to
# "publishing was attempted and failed". Only these earn the idle backoff:
# a failing round should keep its configured cadence so a transient fault
# is retried promptly.
IDLE_STATUSES = frozenset({"no_news", "nothing_merged", "zero_delta"})
# is retried promptly. "duplicate_of_open_pr" belongs here for the same
# reason as "zero_delta": the round correctly found nothing NEW to publish
# (this round's content already sits on an earlier round's open PR), so
# hammering at full cadence just re-discovers the same duplicate.
IDLE_STATUSES = frozenset({"no_news", "nothing_merged", "zero_delta", "duplicate_of_open_pr"})
IDLE_ROUND_DELAY_SECONDS = 60.0

# How many consecutive non-publishing rounds before the loop says so out
Expand Down
110 changes: 110 additions & 0 deletions scripts/test_overlord_sweep.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1634,6 +1634,116 @@ def test_a_stamp_already_an_ancestor_of_origin_main_is_zero_delta_too(self):
self.assertEqual(tested, [])
self.assertEqual(pushed, [])

def test_content_identical_to_an_already_open_pr_is_a_duplicate_not_a_second_pr(self):
"""Measured 2026-08-11: nine open sweep PRs, the last seven of them
with byte-identical diffs. As long as nobody merges a sweep PR,
origin_ref never advances, so the next round's gap detection --
which only ever compares against origin_ref -- rediscovers the
exact same gap, re-fixes it under a fresh sha, and stamps it as
genuine news (it IS a new commit). The origin_ref zero-delta check
just above cannot catch this: the branch really is different from
origin_ref, just not from a PR still open from an earlier round.
This is the second gate: compare against every already-open sweep
PR's branch content too, before paying for the workspace suite, a
push, and a PR that would just be one more duplicate.
"""
repo = self.make_repo()
# An earlier round already published this exact fix as an open PR.
git(repo, "branch", "sweep/tags-earlier", "main")
git(repo, "checkout", "-q", "sweep/tags-earlier")
self.commit_file(repo, "src/a.rs", "fn a() {}\n", "sweep: fix JPEG:Foo")
git(repo, "checkout", "-q", "main")

# THIS round's squad branch produces the identical fix under a
# fresh sha, still never merged to main -- exactly what a worker
# produces when gap detection rediscovers a gap only main's own
# (unchanged) content still has.
git(repo, "branch", "squad/canon", "main")
git(repo, "checkout", "-q", "squad/canon")
canon_sha = self.commit_file(
repo, "src/a.rs", "fn a() {}\n", "fix JPEG:Foo",
trailers=[("Format", "JPEG"), ("Tag", "MakerNotes:Foo")],
)
git(repo, "checkout", "-q", "main")

with tempfile.TemporaryDirectory() as tmpdir:
home = Path(tmpdir) / "home"
config_toml = self._config_toml(Path(tmpdir), ["canon"])
squad_merge_loop.record_head(
squad_merge_loop.squad_status_file(home, "canon"), "workerhead", status="consumed",
patch_id="p1", format_name="JPEG", squad_sha=canon_sha, now_fn=lambda: 100,
)
sweep_state_path = home / "sweep-state.json"
tested, pushed, prs = [], [], []

result = overlord_sweep.run_sweep(
repo_root=repo, home=home, cache_dir="/unused",
comparison_fn=self._passing_comparison_fn, checkout_fn=self._checkout_fn,
config_path=config_toml, sweep_state_path=sweep_state_path, origin_ref="main",
dispatcher_lock_path=home / "logs" / "dispatcher.lock",
cargo_test_workspace_fn=lambda repo_root: tested.append(1) or (True, "ok"),
push_branch_fn=lambda repo_root, branch: pushed.append(branch) or (True, "pushed"),
create_pr_fn=lambda *a, **kw: prs.append(a) or {"ok": True, "url": "u"},
fmt_fn=self._reformatting_fmt_fn, lint_fn=lambda repo_root: (True, ""), log_fn=lambda *a: None,
open_sweep_prs_fn=lambda: [
{"headRefName": "sweep/tags-earlier", "number": 42, "url": "https://example/pull/42"},
],
)
cursor = overlord_sweep.load_sweep_state(sweep_state_path)

self.assertEqual(result["status"], "duplicate_of_open_pr")
self.assertEqual(result["duplicate_of"], "https://example/pull/42")
self.assertEqual(tested, [])
self.assertEqual(pushed, [])
self.assertEqual(prs, [])
# Advancing here is correct and required, same as zero_delta: the
# content IS already published (under someone else's open PR), so
# re-collecting the stamp forever would spin every round.
self.assertIn("canon", cursor["squads"])

def test_a_content_DIFFERENT_open_pr_does_not_veto_a_genuinely_new_fix(self):
"""The duplicate gate must not become a second, accidental
zero_delta check: an open sweep PR fixing a DIFFERENT tag must
never block this round's own, different content from publishing."""
repo = self.make_repo()
git(repo, "branch", "sweep/tags-earlier", "main")
git(repo, "checkout", "-q", "sweep/tags-earlier")
self.commit_file(repo, "src/b.rs", "fn b() {}\n", "sweep: fix JPEG:Bar")
git(repo, "checkout", "-q", "main")

git(repo, "branch", "squad/canon", "main")
git(repo, "checkout", "-q", "squad/canon")
canon_sha = self.commit_file(
repo, "src/a.rs", "fn a() {}\n", "fix JPEG:Foo",
trailers=[("Format", "JPEG"), ("Tag", "MakerNotes:Foo")],
)
git(repo, "checkout", "-q", "main")

with tempfile.TemporaryDirectory() as tmpdir:
home = Path(tmpdir) / "home"
config_toml = self._config_toml(Path(tmpdir), ["canon"])
squad_merge_loop.record_head(
squad_merge_loop.squad_status_file(home, "canon"), "workerhead", status="consumed",
patch_id="p1", format_name="JPEG", squad_sha=canon_sha, now_fn=lambda: 100,
)
pushed, prs = [], []
result = overlord_sweep.run_sweep(
repo_root=repo, home=home, cache_dir="/unused",
comparison_fn=self._passing_comparison_fn, checkout_fn=self._checkout_fn,
config_path=config_toml, sweep_state_path=home / "sweep-state.json", origin_ref="main",
dispatcher_lock_path=home / "logs" / "dispatcher.lock",
cargo_test_workspace_fn=lambda repo_root: (True, "ok"),
push_branch_fn=lambda repo_root, branch: pushed.append(branch) or (True, "pushed"),
create_pr_fn=lambda *a, **kw: prs.append(a) or {"ok": True, "url": "u"},
fmt_fn=self._reformatting_fmt_fn, lint_fn=lambda repo_root: (True, ""), log_fn=lambda *a: None,
open_sweep_prs_fn=lambda: [
{"headRefName": "sweep/tags-earlier", "number": 42, "url": "https://example/pull/42"},
],
)
self.assertEqual(result["status"], "ok")
self.assertEqual(len(pushed), 1)
self.assertEqual(len(prs), 1)


class EmptyRevertIsNotAFailureTests(unittest.TestCase):
""""Nothing to revert" and "cannot revert" are opposites.
Expand Down
Loading