From 3f9927a615b1ace1a2f0d014a3082f83e8f4a283 Mon Sep 17 00:00:00 2001 From: "mia-autodev[bot]" <341115074+mia-autodev[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 14:56:35 -0400 Subject: [PATCH] mia-data: a go after needs-human resumes the open snippet PR A develop round that ends in needs-human for a reason of our own (a lost job, a failed status write) could not be resumed: the next go reset the state, matched again and opened a second mia-snippets issue, so autodev wrote the same snippet twice. MiaDataDataset._round now remembers the snippets issue and PR of a needs-human round. When the PR is still open, the new round (fresh id and pins, tests cleared) resumes it: the issue goes to testing with a comment saying so and the current head gets test crops again. A PR closed or merged meanwhile falls through to the normal match, and the remembered snippets issue is handed to open_snippet_issue, which reuses it while it is open and ours instead of opening another. Guarded with workflow.patched("resume-snippet-pr"); the recorded histories replay unchanged. Tests: the resume (same PR re-tested, no second issue, new round's job keys), a closed and a merged PR (new match), and the activity reusing a remembered issue only while open, ours and this dataset's. Docs and the spec's re-trigger rule updated. Closes #38 Co-Authored-By: Claude Fable 5.1 --- docs/mia-data-workflow.md | 10 +++++-- mia_data/activities.py | 24 +++++++++++----- mia_data/workflows.py | 53 ++++++++++++++++++++++++++++++------ notes/mia-crops-spec.md | 8 ++++-- tests/test_mia_activities.py | 16 +++++++++++ tests/test_mia_workflows.py | 51 ++++++++++++++++++++++++++++++++-- 6 files changed, 140 insertions(+), 22 deletions(-) diff --git a/docs/mia-data-workflow.md b/docs/mia-data-workflow.md index 835d4ba..1e65e1a 100644 --- a/docs/mia-data-workflow.md +++ b/docs/mia-data-workflow.md @@ -111,8 +111,11 @@ the draft binding and what failed. A second autodev deployment, `snippets`, turn `snippets//` and an example binding. The data issue sits in `data:developing` and wakes up whenever the PR changes. Every new PR head gets test crops (step 5) and a required `mia-data/crops` commit status on that exact commit. A human reviews and merges the PR; after the merge, a tree-hash comparison confirms the merged -snippet is the one that was tested, or the merge commit is tested again. Components: autodev (its own -namespace; the two workflows interact only through GitHub), mia-snippets CI, the GitHub client. +snippet is the one that was tested, or the merge commit is tested again. If the round ends in `data:needs-human` +while the PR is still open (a lost job, a failed status write), a new `data:go` resumes that PR: the current head +is tested again under a new round, and no second snippets issue is opened. A PR closed or merged by then means a +new match. Components: autodev (its own namespace; the two workflows interact only through GitHub), mia-snippets +CI, the GitHub client. ### 5. Test crops and review @@ -144,7 +147,8 @@ is `data:needs-human` with the reason. Every stage change posts a new comment, never an edit: the stage, what it produced, and the folders to look at. Labels show the current stage, exactly one `data:*` at a time, set only by the worker. Anything the system cannot handle ends in `data:needs-human` with a comment saying why and what to change; a `data:go` afterwards -retries from the start with a new seed. +retries from the start with a new seed, or, when the round was following a snippet PR that is still open, picks +that PR up again. ## Trust boundaries in one paragraph diff --git a/mia_data/activities.py b/mia_data/activities.py index 06251c9..c4328bd 100644 --- a/mia_data/activities.py +++ b/mia_data/activities.py @@ -711,16 +711,26 @@ async def publish_crops( @activity.defn(name="open_snippet_issue") @mia_permanent_errors - async def open_snippet_issue(self, issue: int, round_id: str, draft: dict) -> int: - """Open the mia-snippets issue asking autodev for a new snippet, or find it again by its first-line - marker on a retry. Returns the issue number.""" + async def open_snippet_issue(self, issue: int, round_id: str, draft: dict, snippet_issue: int | None = None) -> int: + """Open the mia-snippets issue asking autodev for a new snippet, or find it again: the one the workflow + remembers as `snippet_issue` from an earlier round, while it is still open, else by this round's + first-line marker on a retry. Returns the issue number.""" gh, github_cfg, pins = self.gh_snippets, self.cfg.github, self._pins(round_id) marker = self.cfg.marker(f"issue={int(issue)} round={round_id}") - # ponytail: lists every issue; one call per round. Filter by author here: an outsider can copy the marker. + + def ours(existing: dict, prefix: str) -> bool: # filter by author: an outsider can copy the marker + return (existing.get("user") or {}).get("login") == self.bot and (existing.get("body") or "").startswith( + prefix + ) + + if snippet_issue: # any round's marker for this data issue will do: the round id is new on a re-trigger + remembered = await gh.get_issue(int(snippet_issue)) + any_round = self.cfg.marker(f"issue={int(issue)} round=").removesuffix(" -->") + if remembered.get("state") != "closed" and ours(remembered, any_round): + return remembered["number"] + # ponytail: lists every issue; one call per round. for existing in await gh.list_issues(state="all"): - if (existing.get("user") or {}).get("login") == self.bot and (existing.get("body") or "").startswith( - marker - ): + if ours(existing, marker): return existing["number"] record = self._record(round_id) urls = record_urls(record) diff --git a/mia_data/workflows.py b/mia_data/workflows.py index 0027852..4068a91 100644 --- a/mia_data/workflows.py +++ b/mia_data/workflows.py @@ -301,13 +301,21 @@ async def _test(self, sha: str, seed: str, trusted: bool = True, notes: tuple = # --- the round ------------------------------------------------------------------------- async def _round(self, event: LabelEvent) -> None: - """One round from a `go`: admit, match, smoke, test crops, then wait for review.""" + """One round from a `go`: admit, match, smoke, test crops, then wait for review. + + After a develop round ended in needs-human, the `go` resumes its snippet PR instead when that PR is still + open (see _resume_pr); its snippets issue is remembered either way, so a retry never opens a second one.""" state = self.state + snippet_issue, snippet_pr = None, None # what the previous round's develop path left behind, if anything + if state.stage == "needs-human" and state.snippet_issue and workflow.patched("resume-snippet-pr"): + snippet_issue, snippet_pr = state.snippet_issue, state.pr state.round_id, state.seed = f"i{state.issue}-e{event.event_id}", str(workflow.random().getrandbits(64)) state.pins = state.binding = state.binding_hash = state.snippet_name = state.snippet_issue = state.pr = None state.tests, state.attempts, state.pending = [], {}, [] state.pins = await self._act("admit", state.issue, event.event_id, seconds=300, result_type=Pins) self.limits = await self._act("limits", result_type=RoundLimits) + if snippet_issue and snippet_pr and await self._resume_pr(snippet_issue, snippet_pr): + return await self._stage("matching") # Match: the agent picks a snippet and writes a binding, then the worker smoke-tests that binding itself. @@ -318,7 +326,7 @@ async def _round(self, event: LabelEvent) -> None: return await self._finish("needs-human", describe_failure("match", job)) match = await self._act("read_match", workspace, seconds=300, result_type=MatchResult) if match.outcome == "needs_snippet": - return await self._develop(match) + return await self._develop(match, snippet_issue) if match.outcome not in ("matched", "ineligible"): return await self._finish("needs-human", f"match blocked: {match.reason}") state.binding, state.binding_hash = match.binding_path, match.binding_hash @@ -434,11 +442,20 @@ async def _review(self) -> bool: # --- the development path --------------------------------------------------------------- - async def _develop(self, match: MatchResult) -> None: - """No snippet fits: open the mia-snippets issue (autodev writes the snippet), then follow its PR.""" + async def _develop(self, match: MatchResult, snippet_issue: int | None = None) -> None: + """No snippet fits: open the mia-snippets issue (autodev writes the snippet), then follow its PR. + + `snippet_issue` is the issue an earlier round opened for this dataset, when the workflow still knows it: + the activity reuses it while it is open rather than asking autodev for the same snippet twice.""" state = self.state state.snippet_issue = await self._act( - "open_snippet_issue", state.issue, state.round_id, match.draft or {}, seconds=300, result_type=int + "open_snippet_issue", + state.issue, + state.round_id, + match.draft or {}, + *([snippet_issue] if snippet_issue else []), # the extra argument only when it carries something + seconds=300, + result_type=int, ) await self._stage( "developing", @@ -447,6 +464,25 @@ async def _develop(self, match: MatchResult) -> None: ) await self._follow_pr() + async def _resume_pr(self, snippet_issue: int, pr_number: int) -> bool: + """A `go` after a develop round ended in needs-human: pick its snippet PR up again if it is still open, + and test the current head afresh under the new round and pins (spec §7.4, "comment and re-trigger"). + + Returns False when the PR was merged or closed meanwhile, or is not this snippets issue's: the round + then matches as usual (a merged snippet is on main now).""" + state = self.state + pr = await self._act("pr_state", pr_number, result_type=PrUpdate) + if pr.state != "open" or pr.snippet_issue != snippet_issue: + return False + state.snippet_issue, state.pr = snippet_issue, pr_number + await self._stage( + "testing", + f"resuming snippet PR {pr.url or f'#{pr.pr}'} from the previous round; its current head gets test crops again", + ) + if not await self._on_pr(pr_number, pr): + await self._follow_pr() + return True + async def _follow_pr(self) -> None: """Wake on pr_update until the round ends, refusing labels meanwhile (busy). Rolls over only here, idle.""" state = self.state @@ -462,12 +498,13 @@ async def _follow_pr(self) -> None: if await self._on_pr(state.pending.pop(0).pr): return - async def _on_pr(self, pr_number: int) -> bool: + async def _on_pr(self, pr_number: int, pr: PrUpdate | None = None) -> bool: """Act on the PR as it is now (never on the signal's copy) until nothing is left to do. - Returns True when the round ended (merged, closed, needs-human), False when waiting for a new head.""" + `pr` is the PR as the caller just read it; otherwise it is read here. Returns True when the round ended + (merged, closed, needs-human), False when waiting for a new head.""" state = self.state - pr = await self._act("pr_state", pr_number, result_type=PrUpdate) + pr = pr or await self._act("pr_state", pr_number, result_type=PrUpdate) while pr.snippet_issue == state.snippet_issue: # else another round's PR state.pr = pr.pr where = f"snippet PR {pr.url or f'#{pr.pr}'}" diff --git a/notes/mia-crops-spec.md b/notes/mia-crops-spec.md index ad6f094..be46141 100644 --- a/notes/mia-crops-spec.md +++ b/notes/mia-crops-spec.md @@ -142,7 +142,9 @@ any stage ──> data:needs-human (comment says why) · data:ineligible (no pla the round that is under review: its label event must be newer than the round's test result. Anywhere else it is removed with an "ignored" comment. - **Re-trigger.** A new `data:go` on a `done`, `needs-human` or `ineligible` issue starts a new round from - matching, with a new seed. Rounds never overlap: + matching, with a new seed. One exception: a `needs-human` round that was following a snippet PR (§7.3) which + is still open resumes that PR instead, under the new round id and pins, and re-tests its current head; its + snippets issue is remembered so no second one is opened. Rounds never overlap: - a `go` that arrives while a round is running is acknowledged with a "busy, ignored" comment and dropped - a new round starts only after every job from the previous one is terminal (§8.2) @@ -583,7 +585,9 @@ evidence for any axis override or transform. snippet and its example binding, so no `data:approve` is needed. - **Different:** §7.4 runs again on the merge SHA, and only a pass admits `cropping`. 6. If the PR is closed without merging, or the issue gets `snippets:needs-human`, the data issue moves to - `needs-human` with a link. + `needs-human` with a link. A `needs-human` for a reason of our own (a lost job, a failed status write) is + recovered with `data:go`: while the PR is open, the new round resumes it from step 3 (§4, re-trigger); once + it is closed or merged, the new round matches as usual, and a `needs_snippet` reuses the open snippets issue. ### 7.4 Test crops (plain jobs) diff --git a/tests/test_mia_activities.py b/tests/test_mia_activities.py index e1622fc..1aca604 100644 --- a/tests/test_mia_activities.py +++ b/tests/test_mia_activities.py @@ -220,6 +220,22 @@ async def test_open_snippet_issue_twice_gives_one_issue(tmp_path): ) +async def test_open_snippet_issue_reuses_the_remembered_issue_in_a_new_round(tmp_path): + t = _setup(tmp_path) + for event_id in ("99", "100", "101"): + await t.a.admit(1, event_id) + draft = {"draft_binding": BINDING, "what_fails": "no reader", "tried": []} + n = await t.a.open_snippet_issue(1, ROUND, draft) + assert await t.a.open_snippet_issue(1, "i1-e100", draft, n) == n and len(t.snip.created) == 1 # a re-trigger + t.snip.issues[n]["state"] = "closed" # autodev is done with it: the new round gets a new one + m = await t.a.open_snippet_issue(1, "i1-e100", draft, n) + assert m != n and t.snip.created[1][1].startswith("") + other = (await t.snip.create_issue("other", "", []))["number"] + assert await t.a.open_snippet_issue(1, "i1-e101", draft, other) not in (n, m, other) # another dataset's + t.snip.issues[m]["user"]["login"] = "mallory" + assert await t.a.open_snippet_issue(1, "i1-e101", draft, m) not in (n, m, other) # not ours + + def _pr_binding(t, b, files=("snippets/new-snip/snippet.py", "snippets/new-snip/examples/rec-1.yaml")): d = t.cfg.paths.work_root / "_snapshots" / SNIP_SHA / "snippets" / "new-snip" # the PR head's snapshot d.mkdir() diff --git a/tests/test_mia_workflows.py b/tests/test_mia_workflows.py index 65a9858..b21b17e 100644 --- a/tests/test_mia_workflows.py +++ b/tests/test_mia_workflows.py @@ -170,8 +170,8 @@ async def publish_crops(issue: int, round_id: str, crop_ws: str, verified: list[ return {"published": [c["crop_id"] for c in verified], "already": [], "total": len(verified)} @activity.defn(name="open_snippet_issue") - async def open_snippet_issue(issue: int, round_id: str, draft: dict) -> int: - rec("open_snippet_issue", round_id) + async def open_snippet_issue(issue: int, round_id: str, draft: dict, snippet_issue: int | None = None) -> int: + rec("open_snippet_issue", round_id, snippet_issue) return 100 @activity.defn(name="pr_state") @@ -638,6 +638,53 @@ async def test_merge_signal_before_test_finishes(env): assert m.names().index("tree_sha") > m.names().index("read_verify") +async def to_needs_human_in_develop(h, m): + """needs_snippet → snippets issue → h1's crop job is lost (an infrastructure failure) → needs-human.""" + await idle(h, m, "developing") + await wake(h, m, head(H1)) + await idle(h, m, "needs-human") + + +async def test_go_after_needs_human_resumes_the_snippet_pr(env): + """A develop round that failed for an infrastructure reason: `go` re-tests the same PR under a new round and + pins, and opens no second snippets issue (spec §7.4, comment and re-trigger).""" + m = Mocks(outcome="needs_snippet", states={f"crop-{H1}": ("UNKNOWN", None)}, trees={H1: "T", MC: "T"}) + async with started(env, m) as h: + await to_needs_human_in_develop(h, m) + m.states = {} # the infrastructure is fixed + await h.signal(MiaDataDataset.go, ev("go", "12")) + await idle(h, m, "pr-ready") + await wake(h, m, head(H1, "merged", MC)) + await idle(h, m, "done") + st = await h.query(MiaDataDataset.status) + assert m.stages == ["matching", "developing", "testing", "needs-human", "testing", "pr-ready", "cropping", "done"] + assert m.of("admit") == [("11",), ("12",)] and m.of("open_snippet_issue") == [("i7-e11", None)] + assert [a[1] for a in m.of("submit")] == ["match", "crop", "crop", "verify", "crop", "verify"] # no second match + keys = [a[0] for a in m.of("submit")] + assert len(keys) == len(set(keys)) and keys[2].startswith("mia-7-i7-e12-crop-") # the retest is the new round's + assert statuses(m) == [(H1, "pending"), (H1, "error"), (H1, "pending"), (H1, "success")] + resumed = m.of("set_stage")[4] + assert resumed[0] == "testing" and "resuming" in resumed[1] and PR_URL in resumed[1] + assert (st.round_id, st.snippet_issue, st.pr) == ("i7-e12", 100, 5) and not m.of("refuse_label") + assert [t.test_id for t in st.tests] == [f"i7-e12-{H1[:12]}"] + + +@pytest.mark.parametrize("now", [head(H1, "closed"), head(H1, "merged", MC)], ids=["closed", "merged"]) +async def test_go_after_needs_human_with_a_finished_pr_matches_again(env, now): + """The PR is gone by the time `go` arrives: a new match, which hands the snippets issue back for reuse.""" + m = Mocks(outcome="needs_snippet", states={f"crop-{H1}": ("UNKNOWN", None)}) + async with started(env, m) as h: + await to_needs_human_in_develop(h, m) + m.pr_now = now + await h.signal(MiaDataDataset.go, ev("go", "12")) + await idle(h, m, "developing") + st = await h.query(MiaDataDataset.status) + assert m.stages == ["matching", "developing", "testing", "needs-human", "matching", "developing"] + assert m.of("pr_state") == [(5,), (5,)] and [a[1] for a in m.of("submit")] == ["match", "crop", "match"] + assert m.of("open_snippet_issue") == [("i7-e11", None), ("i7-e12", 100)] + assert (st.round_id, st.snippet_issue, st.pr) == ("i7-e12", 100, None) and not m.of("refuse_label") + + # --- the poll activity -------------------------------------------------------------------------