From f6878457286c69e32f49e4c6946a401d2eed5e05 Mon Sep 17 00:00:00 2001 From: Konrad Rokicki Date: Tue, 13 Oct 2026 14:40:53 -0400 Subject: [PATCH] Poll the PR through observe_pr's retry policy instead of a workflow timer A quiet PR used to add an observe_pr activity and a timer to the history every poll_minutes, drawing a staircase on the Temporal UI timeline that hid the real work. Now observe_pr fails (retryably) while decide() would WAIT and the workflow calls it once with a fixed-interval retry policy, so the whole wait is one activity row and retries add no events (Temporal's "infrequent polling" pattern). - ObserveInput carries the decision context (handled, revisions, max_revisions, escalated) so the activity can run decide() itself. - decide() treats an escalation already raised for the same reason and head as WAIT; IssueInput.escalated records it, so an escalated PR polls quietly instead of re-commenting every round. - The old loop stays behind workflow.patched("observe-by-retry") so in-flight runs replay; drop it once they have rolled over. Co-Authored-By: Claude Fable 5.1 --- autodev/activities.py | 24 ++++++++++++++------ autodev/decide.py | 18 +++++++++++---- autodev/models.py | 8 +++++++ autodev/workflows.py | 38 +++++++++++++++++++++++++------- tests/test_autodev_activities.py | 21 ++++++++++-------- tests/test_autodev_decide.py | 10 ++++++++- tests/test_autodev_workflows.py | 17 ++++++++++++-- 7 files changed, 105 insertions(+), 31 deletions(-) diff --git a/autodev/activities.py b/autodev/activities.py index ae71d3c..a275306 100644 --- a/autodev/activities.py +++ b/autodev/activities.py @@ -21,6 +21,7 @@ from .agent import log_tail, read_result, render_prompt from .config import STAGES +from .decide import WAIT, decide from .github import GitHubError from .gitops import RejectedBundleError from .models import ( @@ -290,14 +291,23 @@ async def publish_round(self, request: PublishInput) -> PrRef: @activity.defn(name="observe_pr") @permanent_errors async def observe_pr(self, request: ObserveInput) -> PrSnapshot: - """Read the PR's state, approvals, pending change requests and CI status into one snapshot.""" - pr = await self.gh.get_pr(request.pr) + """Read the PR's state, approvals, pending change requests and CI status into one snapshot. + + Raises a retryable error while there is nothing for the workflow to act on, so Temporal's retry + policy (not a workflow timer) spaces the polls and the history stays flat.""" + snapshot = await self._snapshot(request.pr) + if decide(snapshot, request.handled, request.revisions, request.max_revisions, request.escalated)[0] == WAIT: + raise ApplicationError(f"PR #{request.pr}: nothing to do yet") + return snapshot + + async def _snapshot(self, number: int) -> PrSnapshot: + pr = await self.gh.get_pr(number) head_sha = pr["head"]["sha"] if pr.get("merged") or pr["state"] != "open": - return PrSnapshot(request.pr, "merged" if pr.get("merged") else pr["state"], head_sha, False, "none") - reviews = await self.gh.pr_reviews(request.pr) - inline_comments = await self.gh.pr_review_comments(request.pr) - comments = await self.gh.issue_comments(request.pr) + return PrSnapshot(number, "merged" if pr.get("merged") else pr["state"], head_sha, False, "none") + reviews = await self.gh.pr_reviews(number) + inline_comments = await self.gh.pr_review_comments(number) + comments = await self.gh.issue_comments(number) logins = {(item.get("user") or {}).get("login") for item in [*reviews, *comments]} - {None} # ponytail: one permission lookup per participant per poll; cache if the rate limit ever matters. permissions = {login: await self.gh.permission(login) for login in logins} @@ -322,7 +332,7 @@ async def observe_pr(self, request: ObserveInput) -> PrSnapshot: latest_round_at(comments, marker_prefix, self.bot), ) checks = await self.gh.check_state(head_sha, self.cfg.github.required_checks) - return PrSnapshot(request.pr, "open", head_sha, is_approved, checks, pending) + return PrSnapshot(number, "open", head_sha, is_approved, checks, pending) @activity.defn(name="merge_pr") @permanent_errors diff --git a/autodev/decide.py b/autodev/decide.py index f43e73b..6bfe39f 100644 --- a/autodev/decide.py +++ b/autodev/decide.py @@ -10,13 +10,23 @@ DONE, CLOSED, MERGE, REVISE, ESCALATE, WAIT = "DONE", "CLOSED", "MERGE", "REVISE", "ESCALATE", "WAIT" +def escalation(reason: str, snapshot: PrSnapshot) -> str: + """The identity of an escalation: the same reason for the same head is raised with a human only once.""" + return f"{reason} {snapshot.head_sha}" + + def decide( - snapshot: PrSnapshot, handled: list[str], revisions: int, max_revisions: int + snapshot: PrSnapshot, handled: list[str], revisions: int, max_revisions: int, escalated: str | None = None ) -> tuple[str, Feedback | str | None]: """Return (action, detail): the Feedback to act on for REVISE, the reason for ESCALATE, else None. Pending change requests win over an approval while revisions remain; once the limit is hit, a pending - request escalates to a human instead.""" + request escalates to a human instead. An escalation already raised (`escalated`) is a WAIT: the PR + is in a human's hands until something about it changes.""" + + def escalate(reason: str) -> tuple[str, str | None]: + return (WAIT, None) if escalation(reason, snapshot) == escalated else (ESCALATE, reason) + if snapshot.state == "merged": return DONE, None if snapshot.state == "closed": @@ -28,7 +38,7 @@ def decide( if snapshot.checks in ("success", "none"): return MERGE, None if snapshot.checks == "failure": - return ESCALATE, "approved, but checks are failing" + return escalate("approved, but checks are failing") if pending: - return ESCALATE, f"revision limit ({max_revisions}) reached" + return escalate(f"revision limit ({max_revisions}) reached") return WAIT, None diff --git a/autodev/models.py b/autodev/models.py index 1a85484..10ee9e9 100644 --- a/autodev/models.py +++ b/autodev/models.py @@ -50,6 +50,7 @@ class IssueInput: pr: int | None = None revisions: int = 0 handled: list[str] = field(default_factory=list) # feedback ids already addressed by a round + escalated: str | None = None # decide.escalation() of the escalation already raised with a human @dataclass @@ -128,7 +129,14 @@ class PrRef: @dataclass class ObserveInput: + """What to observe, plus the workflow's decision context: the activity fails (retryably) while the + snapshot would decide WAIT, so one activity with a fixed-interval retry policy is the poll loop.""" + pr: int + handled: list[str] = field(default_factory=list) + revisions: int = 0 + max_revisions: int = 3 + escalated: str | None = None @dataclass diff --git a/autodev/workflows.py b/autodev/workflows.py index b9a2b9f..8186c17 100644 --- a/autodev/workflows.py +++ b/autodev/workflows.py @@ -15,7 +15,7 @@ from temporalio.exceptions import ActivityError with workflow.unsafe.imports_passed_through(): - from .decide import CLOSED, DONE, ESCALATE, MERGE, REVISE, decide + from .decide import CLOSED, DONE, ESCALATE, MERGE, REVISE, WAIT, decide, escalation from .models import ( KEY_RE, AwaitInput, @@ -78,7 +78,7 @@ async def _act( name: str, arg: Any, seconds: int = 60, - budget: timedelta = timedelta(hours=6), + budget: timedelta | None = timedelta(hours=6), result_type: type | None = None, ) -> Any: """Run an activity with a per-attempt timeout and an overall retry budget. @@ -170,14 +170,19 @@ async def _run_round(self, feedback: Feedback | None) -> PrRef: return result async def _watch_pr(self) -> str: - """Poll the PR until it is merged or closed, running revision rounds and escalations as needed. + """Watch the PR until it is merged or closed, running revision rounds and escalations as needed. - Rolls the history over with continue_as_new when Temporal suggests it or the poll cap is hit.""" + The wait between polls is observe_pr's retry interval (the activity fails while there is nothing to + do), so a quiet PR is one activity in the history instead of a poll/timer staircase. The history + rolls over with continue_as_new when Temporal suggests it or the poll cap is hit.""" state = self.state + # ponytail: by_retry is False only when replaying runs started before this change; drop the old + # branches (workflow.deprecate_patch) once those have all rolled over. + by_retry = workflow.patched("observe-by-retry") for _ in range(MAX_POLLS_PER_EXECUTION): self.phase = "REVIEW" - snapshot = await self._act("observe_pr", ObserveInput(state.pr), result_type=PrSnapshot) - action, detail = decide(snapshot, state.handled, state.revisions, state.max_revisions) + snapshot = await self._observe(by_retry) + action, detail = decide(snapshot, state.handled, state.revisions, state.max_revisions, state.escalated) if action == DONE: await self._act("set_stage", StageInput(state.issue, "done")) return "merged" @@ -202,12 +207,29 @@ async def _watch_pr(self) -> str: await self._act("set_stage", StageInput(state.issue, "review")) continue elif action == ESCALATE: - await self._comment(state.pr, f"Needs a human: {detail}.", f"escalate {detail} {snapshot.head_sha}") - await workflow.sleep(timedelta(minutes=state.poll_minutes)) + if by_retry: + state.escalated = escalation(detail, snapshot) + await self._comment(state.pr, f"Needs a human: {detail}.", f"escalate {escalation(detail, snapshot)}") + if action in (MERGE, WAIT) or not by_retry: # a failed merge (or an idle snapshot): look again later + await workflow.sleep(timedelta(minutes=state.poll_minutes)) if workflow.info().is_continue_as_new_suggested(): break workflow.continue_as_new(state) + async def _observe(self, by_retry: bool) -> PrSnapshot: + state = self.state + inp = ObserveInput(state.pr, state.handled, state.revisions, state.max_revisions, state.escalated) + if not by_retry: # pre-change runs: same commands as before, but the "nothing to do" retries never expire + return await self._act("observe_pr", inp, budget=None, result_type=PrSnapshot) + poll = timedelta(minutes=state.poll_minutes) + return await workflow.execute_activity( + "observe_pr", + inp, + start_to_close_timeout=timedelta(seconds=60), + retry_policy=RetryPolicy(initial_interval=poll, backoff_coefficient=1.0), # retries forever + result_type=PrSnapshot, + ) + @workflow.defn class PollWorkflow: diff --git a/tests/test_autodev_activities.py b/tests/test_autodev_activities.py index 18466b3..03c5888 100644 --- a/tests/test_autodev_activities.py +++ b/tests/test_autodev_activities.py @@ -10,7 +10,6 @@ from temporalio.exceptions import ApplicationError from autodev.activities import Activities, neutralize -from autodev.decide import WAIT, decide from autodev.github import GitHub from autodev.gitops import GitError, Mirror from autodev.models import FailureInput, Feedback, ObserveInput, PublishInput, RoundInput, StageInput @@ -122,11 +121,10 @@ async def _observe_required(tmp_path, gh, runs, statuses): @respx.mock async def test_required_check_missing_means_pending(tmp_path, gh): - snap = await _observe_required( - tmp_path, gh, [{"name": "check", "status": "completed", "conclusion": "success"}], [] - ) - assert snap.approved and snap.checks == "pending" - assert decide(snap, [], 0, 3)[0] == WAIT + """Approved but waiting on a check is nothing to act on: the activity fails so its retry becomes the next poll.""" + with pytest.raises(ApplicationError, match="nothing to do") as info: + await _observe_required(tmp_path, gh, [{"name": "check", "status": "completed", "conclusion": "success"}], []) + assert not info.value.non_retryable @respx.mock @@ -270,7 +268,12 @@ async def test_observe_counts_approve_comments_only_after_latest_round(tmp_path, ) respx.get(f"{API}{R}/commits/{head}/check-runs").mock(return_value=httpx.Response(200, json={"check_runs": []})) respx.get(f"{API}{R}/commits/{head}/status").mock(return_value=httpx.Response(200, json={"statuses": []})) - assert (await acts(tmp_path, gh).observe_pr(ObserveInput(5))).approved is want + observe = acts(tmp_path, gh).observe_pr(ObserveInput(5)) + if want: + assert (await observe).approved + else: + with pytest.raises(ApplicationError, match="nothing to do"): + await observe async def _prepared(tmp_path, gh, issue=7): @@ -462,8 +465,8 @@ async def test_observe_unchanged_pr_sends_etags_and_reuses_cache(tmp_path, gh): for path, body in bodies.items() } a = acts(tmp_path, gh) - first = await a.observe_pr(ObserveInput(5)) - assert await a.observe_pr(ObserveInput(5)) == first + first = await a._snapshot(5) + assert await a._snapshot(5) == first for path, route in routes.items(): assert route.call_count == 2 and "if-none-match" not in route.calls[0].request.headers assert route.calls[1].request.headers["if-none-match"] == f'"{path}"' diff --git a/tests/test_autodev_decide.py b/tests/test_autodev_decide.py index b1af68c..6fac2a1 100644 --- a/tests/test_autodev_decide.py +++ b/tests/test_autodev_decide.py @@ -1,4 +1,4 @@ -from autodev.decide import CLOSED, DONE, ESCALATE, MERGE, REVISE, WAIT, decide +from autodev.decide import CLOSED, DONE, ESCALATE, MERGE, REVISE, WAIT, decide, escalation from autodev.models import Feedback, PrSnapshot F1, F2 = Feedback("review-1", "a"), Feedback("comment-2", "b") @@ -35,3 +35,11 @@ def test_approved_but_pending_waits_and_failing_escalates(): def test_revision_limit_escalates_unless_approved(): assert decide(snap(feedback=[F1]), [], 3, 3) == (ESCALATE, "revision limit (3) reached") assert decide(snap(approved=True, feedback=[F1]), [], 3, 3) == (MERGE, None) + + +def test_raised_escalation_waits_until_the_head_changes(): + reason = "revision limit (3) reached" + raised = escalation(reason, snap()) + assert decide(snap(feedback=[F1]), [], 3, 3, raised) == (WAIT, None) + assert decide(snap(approved=True, checks="failure"), [], 3, 3, raised)[0] == ESCALATE # a different reason + assert decide(PrSnapshot(5, "open", "h2", False, "none", [F1]), [], 3, 3, raised) == (ESCALATE, reason) diff --git a/tests/test_autodev_workflows.py b/tests/test_autodev_workflows.py index 5f2b20c..e9836d9 100644 --- a/tests/test_autodev_workflows.py +++ b/tests/test_autodev_workflows.py @@ -14,6 +14,7 @@ from temporalio.worker import Replayer, UnsandboxedWorkflowRunner, Worker from autodev import workflows +from autodev.decide import WAIT, decide from autodev.models import ( AwaitInput, CommentInput, @@ -107,7 +108,10 @@ async def observe(a: ObserveInput) -> PrSnapshot: if self.observe_failures: self.observe_failures -= 1 raise ApplicationError("github 502") - return self.snaps.pop(0) if len(self.snaps) > 1 else self.snaps[0] + s = self.snaps.pop(0) if len(self.snaps) > 1 else self.snaps[0] + if decide(s, a.handled, a.revisions, a.max_revisions, a.escalated)[0] == WAIT: # as the real one does + raise ApplicationError("nothing to do yet") + return s @activity.defn(name="merge_pr") async def merge(a: MergeInput) -> MergeResult: @@ -212,6 +216,15 @@ async def test_revision_limit_escalates_but_approval_still_merges(env): assert any("revision limit" in c.body for c in m.of("comment")) +async def test_escalation_is_raised_once_then_waits(env): + f1, f2 = Feedback("review-1", "a"), Feedback("comment-2", "b") + stuck = snap(feedback=[f1, f2]) + m = Mocks([snap(feedback=[f1]), stuck, stuck, stuck, snap(feedback=[f1, f2], approved=True)]) + assert await run(env, m, IssueInput(7, max_revisions=1)) == "merged" + assert m.names().count("observe") == 5 + assert sum("revision limit" in c.body for c in m.of("comment")) == 1 + + async def test_closed_pr_stops(env): m = Mocks([snap(state="closed")]) assert await run(env, m) == "closed" @@ -225,7 +238,7 @@ async def test_merge_failure_comments_then_external_merge_finishes(env): async def test_continue_as_new_keeps_handled_feedback(env, monkeypatch): - monkeypatch.setattr(workflows, "MAX_POLLS_PER_EXECUTION", 2) + monkeypatch.setattr(workflows, "MAX_POLLS_PER_EXECUTION", 1) fb = Feedback("review-1", "x") m = Mocks([snap(feedback=[fb])] + [snap(sha="h2", feedback=[fb])] * 3 + [snap(sha="h2", approved=True)]) assert await run(env, m, runner=UnsandboxedWorkflowRunner()) == "merged"