Poll the PR through observe_pr's retry policy instead of a workflow timer

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 (
@@ -297,14 +298,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 = {login for item in [*reviews, *comments] if (login := (item.get("user") or {}).get("login"))}
         # 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}
@@ -329,7 +339,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

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

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

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,15 +170,20 @@ 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
         assert state.pr is not None  # a round opened the PR before _watch_pr runs
+        # 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"
@@ -204,12 +209,31 @@ 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))
+                assert isinstance(detail, str)  # decide() returns the reason for ESCALATE
+                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
+        assert state.pr is not None  # see _watch_pr
+        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:

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

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)

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"