mia-data: re-trigger of a failed develop round should resume the existing snippet PR

docs/mia-data-workflow.md

@@ -111,8 +111,11 @@ the draft binding and what failed. A second autodev deployment, `snippets`, turn
 `snippets/<name>/` 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
 

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)

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}'}"

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)
 

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("<!-- mia-data: issue=1 round=i1-e100 -->")
+    other = (await t.snip.create_issue("other", "<!-- mia-data: issue=2 round=i2-e5 -->", []))["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()

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 -------------------------------------------------------------------------