DRU-1044 -- Wait for CI before the review, and keep a PR that is not ready in draft

backend/druks/contrib/software_factory/constants.py

@@ -13,3 +13,11 @@
 )
 # A review takes minutes. A build still waiting after this long was never answered.
 REVIEW_TTL_SECONDS = 2 * 60 * 60
+# A repo with CI registers a check within this long of a push. With none by then, the
+# repo has no CI, and the build goes on as if CI were green.
+CI_REGISTER_SECONDS = 3 * 60
+# CI still running after this long goes to the review unfinished.
+CI_TTL_SECONDS = 60 * 60
+# A webhook wakes the CI wait early. The poll catches a check that finished before the
+# wait began, and a delivery that never came.
+CI_POLL_SECONDS = 60

backend/druks/contrib/software_factory/contracts.py

@@ -3,6 +3,7 @@
 from pydantic import BaseModel, Field, model_validator
 
 from druks.agents import AgentOutput
+from druks.contrib.software_factory.constants import CI_REGISTER_SECONDS, CI_TTL_SECONDS
 from druks.contrib.software_factory.enums import HumanFeedbackAction, ReviewDecision
 from druks.workflows import Gate, OperatorReply, Workflow
 
@@ -22,7 +23,10 @@ class ReviewWork(Gate):
     @classmethod
     async def on_wait(cls, workflow: Workflow) -> None:
         build = cast("Build", workflow)
-        await build.set_pr_draft(draft=False)
+        # A round parks for a person even when its CI is not green or its review requested
+        # changes, but its PR is not ready to merge.
+        if build.journal.is_ready_for_review:
+            await build.set_pr_draft(draft=False)
         await build.request_assignee_review()
 
 
@@ -300,3 +304,53 @@ def to_artifact(self) -> dict[str, str]:
 
     def to_event(self) -> dict[str, str]:
         return {"topic": "review.completed", "summary": self.summary}
+
+
+# A suite or status that passed or did not apply.
+_GREEN_CONCLUSIONS = frozenset({"success", "neutral", "skipped"})
+
+
+class CiReport(BaseModel):
+    """CI's verdict on one delivered head. ``red`` sends the round back to implement.
+    ``unfinished`` is CI that did not finish in time."""
+
+    head_sha: str
+    conclusion: Literal["green", "red", "unfinished"]
+
+    @classmethod
+    def judge(
+        cls,
+        head_sha: str,
+        suites: list[dict[str, Any]],
+        statuses: list[dict[str, Any]],
+        *,
+        waited: float,
+    ) -> "CiReport | None":
+        """The verdict on ``head_sha`` from its check suites and commit statuses after
+        ``waited`` seconds, or None while CI can still change."""
+        registering = waited < CI_REGISTER_SECONDS
+        # Some apps leave a queued suite with no check runs for good.
+        suites = [suite for suite in suites if suite["latest_check_runs_count"] or registering]
+        if not suites and not statuses:
+            return None if registering else cls(head_sha=head_sha, conclusion="green")
+        # A suite stays open until its workflow run completes, while the check run of a
+        # ``needs:`` job appears only once the jobs it needs finish.
+        finished = all(
+            suite["latest_check_runs_count"] and suite["status"] == "completed" for suite in suites
+        ) and all(status["state"] != "pending" for status in statuses)
+        if not finished:
+            if waited < CI_TTL_SECONDS:
+                return None
+            return cls(head_sha=head_sha, conclusion="unfinished")
+        green = all(suite["conclusion"] in _GREEN_CONCLUSIONS for suite in suites) and all(
+            status["state"] == "success" for status in statuses
+        )
+        return cls(head_sha=head_sha, conclusion="green" if green else "red")
+
+    @property
+    def revision_request(self) -> str:
+        return (
+            f"CI on `{self.head_sha}` is red. Read it with `gh pr checks` and "
+            "`gh run view --log-failed`, fix what your diff caused, and report the rest "
+            "in `checks`."
+        )

backend/druks/contrib/software_factory/journal.py

@@ -2,6 +2,7 @@
 from typing import Any
 
 from druks.contrib.software_factory.contracts import (
+    CiReport,
     ImplementationOutput,
     PlanData,
     ReviewOutput,
@@ -56,8 +57,26 @@ def assignee_github_login(self) -> str | None:
                 return plan.assignee_github_login
         return
 
+    @property
+    def is_ready_for_review(self) -> bool:
+        """The round's review of a head with green CI requested no changes."""
+        ci_report = self.latest(CiReport)
+        if not ci_report or ci_report.conclusion != "green":
+            return False
+        reviews = self.filter(ReviewReport, after=ci_report)
+        return bool(reviews) and reviews[-1].decision != "request_changes"
+
     @property
     def revision_request(self) -> str | None:
+        # Red CI sends the round back before its review, so a review recorded after it
+        # means the round moved on.
+        ci_report = self.latest(CiReport)
+        if (
+            ci_report
+            and ci_report.conclusion == "red"
+            and not self.filter(ReviewReport, after=ci_report)
+        ):
+            return ci_report.revision_request
         # The newest triage's instructions drive exactly one implement round.
         # A review recorded after that triage means the round already ran —
         # the review's findings are the current request, not these.

backend/druks/contrib/software_factory/subscribers.py

@@ -106,6 +106,15 @@ async def pr_review_answers_the_gate(*, repo: str, pr_number: int, payload: dict
             )
 
 
+@subscribe("ci.updated")
+async def ci_update_wakes_the_build(*, repo: str, payload: dict) -> None:
+    """A build waiting on CI reads its checks again. A commit no build delivered wakes
+    no one."""
+    for branch in payload["branches"]:
+        if item := await WorkItem.get_for_branch(repo=repo, branch=branch):
+            await Build.answer(item, "ci.updated")
+
+
 @subscribe("pr.closed")
 async def pr_close_settles_the_item(*, repo: str, pr_number: int, payload: dict) -> None:
     """GitHub announcing the verdict on a PR druks owns — one path for every merge,

backend/druks/contrib/software_factory/templates/build/implement.md

@@ -45,11 +45,11 @@ acceptance criterion — nothing more, nothing less.
 
 {% endif %}When the checkout has `.druks/software_factory/review.md`, read it before you write: it is the repo's standing rules for code and prose, and the reviewer holds your diff to it.
 
-Implement the approved plan (rendered above as **Current plan**) on the PR branch (see "Delivering your work" below). When a **Revision request** section is rendered above, implement it on top of the plan — it is the reviewer's feedback, already triaged into instructions. Verification is yours: before pushing, run the commands in the **Verification profile** above. Fix a failure your diff causes. A failure you cannot attribute to your own diff is environment noise to report, not a gate to chase. Do not run ad hoc install, build, or smoke commands beyond that unless the plan explicitly requires changing those commands or generated outputs. Never leave dependency lockfile, generated, or cache changes unless they are part of the requested implementation. Return structured evidence for every acceptance criterion, every check you ran or intentionally did not run, changed files, and known risks. If a check was not run, include status not_run and a reason.
+Implement the approved plan (rendered above as **Current plan**) on the PR branch (see "Delivering your work" below). When a **Revision request** section is rendered above, implement it on top of the plan — it is the reviewer's feedback, already triaged into instructions, or a note that CI on your last push is red. Verification is yours: before every push, run the lint and typecheck commands in the **Verification profile** above, and its test commands narrowed to the files you changed. Report each result in `checks`. Fix a failure your diff causes. A failure you cannot attribute to your own diff is environment noise to report, not a gate to chase. Do not run ad hoc install, build, or smoke commands beyond that unless the plan explicitly requires changing those commands or generated outputs. Never leave dependency lockfile, generated, or cache changes unless they are part of the requested implementation. Return structured evidence for every acceptance criterion, every check you ran or intentionally did not run, changed files, and known risks. If a check was not run, include status not_run and a reason.
 
 ## Delivering your work
 
-When the implementation is complete you MUST commit and push it to the PR branch — the pushed commit is the deliverable. The repo is checked out on the PR branch `{{ build.branch }}` already. Stay on it: Druks finds the PR by this name. Run, from the repo root:
+When the implementation is complete you MUST commit and push it to the PR branch — the pushed commit is the deliverable. The one exception is a revision request for red CI that your diff did not cause: then push nothing, report each red check in `checks` with the reason it is not yours, and report the branch's current head as `base_sha`, `head_sha`, and `commit_sha`. The repo is checked out on the PR branch `{{ build.branch }}` already. Stay on it: Druks finds the PR by this name. Run, from the repo root:
 
 ```
 git status --short                      # review what changed

backend/druks/contrib/software_factory/workflows.py

@@ -1,10 +1,13 @@
 import logging
+import time
+from contextlib import suppress
 from typing import Any
 
 from pydantic import BaseModel, Field
 
 from druks.accounts import Account
 from druks.contrib.software_factory.contracts import (
+    CiReport,
     ImplementationOutput,
     ReviewReport,
     ReviewWork,
@@ -20,11 +23,18 @@
 from druks.services.exceptions import ServiceNotConnectedError
 from druks.settings import get_settings
 from druks.skills.models import Skill
-from druks.workflows import FatalError, Workflow, step
+from druks.workflows import FatalError, WaitTimeout, Workflow, step
 from druks.workspaces import RepoWorkspace
 
 from .app import SoftwareFactory
-from .constants import GITHUB_MCP_NAME, GITHUB_MCP_URL, REVIEW_TTL_SECONDS, TICKET_TOOLS
+from .constants import (
+    CI_POLL_SECONDS,
+    CI_TTL_SECONDS,
+    GITHUB_MCP_NAME,
+    GITHUB_MCP_URL,
+    REVIEW_TTL_SECONDS,
+    TICKET_TOOLS,
+)
 from .datastructures import PullRequest
 from .exceptions import TicketAlreadyMerged, TicketNotRoutable
 from .github import get_review_actor
@@ -288,15 +298,54 @@ async def _plan_phase(self) -> bool:
     async def _implement_phase(self) -> None:
         await self._create_branch()
         while True:
-            await self.implement()
-            review = await self._review_work()
-            if review.decision == "request_changes" and (
+            previous = self.journal.last_implementation
+            delivery = await self.implement()
+            can_revise = (
                 self.journal.implementation_revision < self._settings.max_implementation_revisions
-            ):
+            )
+            # The implementer pushes nothing when no red check is its diff's, and CI on
+            # that head has already been read.
+            if not previous or delivery.head_sha != previous.head_sha:
+                ci_report = await self._wait_for_ci()
+                if ci_report.conclusion == "red" and can_revise:
+                    continue
+            review = await self._review_work()
+            if review.decision == "request_changes" and can_revise:
                 continue
             if await self._work_gate():
                 return
 
+    async def _wait_for_ci(self) -> CiReport:
+        """Wait until CI has finished on the delivered head, and return its verdict."""
+        head_sha = self.journal.last_implementation.head_sha
+        started_at = await self._now()
+        while not (ci_report := await self._read_ci(head_sha, started_at)):
+            with suppress(WaitTimeout):
+                await self.wait(
+                    "ci.updated",
+                    label="Waiting for CI",
+                    ttl_seconds=CI_POLL_SECONDS,
+                    release_sandbox=True,
+                )
+        self.journal.add(ci_report)
+        return ci_report
+
+    @step
+    async def _now(self) -> float:
+        return time.time()
+
+    @step
+    async def _read_ci(self, head_sha: str, started_at: float) -> CiReport | None:
+        """CI's verdict on ``head_sha``, or None while it can still change."""
+        repo = (await self.subject).repo
+        github = await Github.get_client(repo)
+        return CiReport.judge(
+            head_sha,
+            await github.list_check_suites(repo, head_sha),
+            await github.list_commit_statuses(repo, head_sha),
+            waited=time.time() - started_at,
+        )
+
     async def _review_work(self) -> ReviewReport:
         """Hand the pull request to the standalone review and wait for its decision."""
         # The body starts the child run, so a replay gets back the review it started.
@@ -315,16 +364,28 @@ async def _work_gate(self) -> bool:
         goes back to implementation."""
         if self._policy.implementation_approval_gate() == "none":
             return await self._approved_work()
+        label = f"Review implementation on GitHub: PR #{self.pr_number}"
+        ci_report = self.journal.latest(CiReport)
+        if ci_report.conclusion == "red":
+            label = f"{label}. CI on {ci_report.head_sha[:7]} is red"
+        if ci_report.conclusion == "unfinished":
+            label = (
+                f"{label}. CI on {ci_report.head_sha[:7]} did not finish within "
+                f"{CI_TTL_SECONDS // 60} minutes"
+            )
         decision = await ReviewWork.wait(
             input_request={
                 "presentation": "external",
-                "label": f"Review implementation on GitHub: PR #{self.pr_number}",
+                "label": label,
                 "url": f"https://github.com/{(await self.subject).repo}/pull/{self.pr_number}",
             }
         )
         if decision.action == "merged":
             return True
         if decision.action == "approve":
+            # A person's approval stands over CI, and GitHub merges no draft.
+            if self._policy.on_approval == "merge":
+                await self._clear_draft()
             return await self._approved_work()
         if decision.action == "request_changes":
             return await self._triage()

backend/druks/core/apis/github.py

@@ -550,6 +550,27 @@ async def post_check_run(
             body["output"] = {**output, "annotations": list(annotations[start : start + limit])}
             await checks.async_update(owner, repo_name, check_run_id, data=body)
 
+    async def list_check_suites(self, repo: str, ref: str) -> list[dict[str, Any]]:
+        """The check suites on ``ref``: one per Actions workflow run, or per other app."""
+        owner, name = repo.split("/", 1)
+        suites = self._github.rest.paginate(
+            self._github.rest.checks.async_list_suites_for_ref,
+            map_func=lambda response: response.parsed_data.check_suites,
+            owner=owner,
+            repo=name,
+            ref=ref,
+            per_page=100,
+        )
+        return [suite.model_dump() async for suite in suites]
+
+    async def list_commit_statuses(self, repo: str, ref: str) -> list[dict[str, Any]]:
+        """The latest status of each context on ``ref``."""
+        owner, name = repo.split("/", 1)
+        response = await self._github.rest.repos.async_get_combined_status_for_ref(
+            owner, name, ref, per_page=100
+        )
+        return [status.model_dump() for status in response.parsed_data.statuses]
+
     async def get_file_content(
         self,
         repo: str,

backend/druks/core/webhooks/github.py

@@ -234,6 +234,20 @@ async def on_pull_request_closed(self) -> Response:
         )
         return _accepted()
 
+    async def on_check_run_completed(self) -> Response:
+        # A check on a fork's commit or a tag has no head branch.
+        branch = self.data["check_run"]["check_suite"]["head_branch"]
+        return await self._publish_ci_update([branch] if branch else [])
+
+    async def on_status(self) -> Response:
+        return await self._publish_ci_update([branch["name"] for branch in self.data["branches"]])
+
+    async def _publish_ci_update(self, branches: list[str]) -> Response:
+        """Publish that a check changed on the commit these branches head. A subscriber
+        reads the checks themselves."""
+        await publish("ci.updated", repo=_repo_name(self.data), payload={"branches": branches})
+        return _accepted()
+
     async def on_push(self) -> Response:
         repository = self.data["repository"]
         repo = repository["full_name"]

backend/druks/workflows.py

@@ -962,10 +962,15 @@ async def review(
         self.journal.add(reply)
         return reply
 
-    async def wait(self, topic: str, *, label: str, ttl_seconds: float) -> dict[str, Any]:
+    async def wait(
+        self, topic: str, *, label: str, ttl_seconds: float, release_sandbox: bool = False
+    ) -> dict[str, Any]:
         """Wait for the facts that ``answer`` delivers on ``topic``, and return them. The run
-        stays running, shows ``label``, and keeps its warm sandbox. A failure answered
-        instead raises ``WaitFailed``."""
+        stays running and shows ``label``. It keeps its warm sandbox, or with
+        ``release_sandbox`` releases it as a gate park does. A failure answered instead
+        raises ``WaitFailed``."""
+        if release_sandbox:
+            await self._reap_run()
         # The wait takes answers before its label shows, so an answer sent on seeing it arrives.
         await DBOS.set_event_async("waiting_on", topic)
         await set_run_phase(label)