"""nisse-review's scheduler (R1, R2): its one duty, and its items one at a time. Every round, it works out the repositories it reviews and checks it still can't write to any (``access``), then asks Gitea for their open PRs. A PR that is ready (not a draft, not marked work in progress) gets a work item. Each round takes in what changed on every item: a new push, and what was said on the PR since. The oldest item with a run due gets it: its first review, a later review of what changed, replies to what was said, or, at the round cap, a summary for the operator. An item closes when its PR merges or closes. Rounds (R2): each review, and each run that posted replies, is a round. After ``max_rounds`` of them, the next run sums up what is still open and the item goes quiet (``CAPPED``). A new push, or a comment from anyone but the reviewer and the PR's author, resets the count. An operator's `stop` comment quiets it until their `review` (R5), which asks for a fresh review. Crash safety (L8): an item found ``running`` was interrupted. Whether its review got posted is asked of Gitea, never assumed; until Gitea can say, the item waits rather than risk a second review. A reply carries the key of the point it answers, so a re-run never answers it twice. A stop (SIGTERM) ends a run at its next turn, or breaks off its model call, and the item goes back to ready, with nothing drafted posted and no failure counted. """ import re import tempfile import threading import time from collections.abc import Callable from dataclasses import replace from pathlib import Path from nisse_core.gitea import GiteaClient, GiteaError, PullRequest from nisse_core.models.interruptible import Stopped from nisse_core.oplog import OpLog from nisse_core.state.cancel import Cancel from nisse_core.state.store import WorkItem from nisse_review.access import check, reviewed from nisse_review.reader import Repository from nisse_review.review import Draft, Posted, Reviewer, Said, answered, said_on from nisse_review.state import ( CAPPED, FAILED, FIRST, LATER, POST_REVIEW, PR, READY, REPLIES, REVIEWED, RUNNING, STOPPED, SUMMARY, WAITING, StateStore, data, item_id, pr_of, ) PARTIAL = ( "A partial review: I ran out of turns before finishing, so these are the " "{n} point(s) I had drafted, without an overall view. A new push gets a " "full look." ) _WIP = re.compile(r"^\s*(\[wip\]|wip\b)", re.IGNORECASE) # The ending tools that post a review. _POSTING = {"post_review", "sign_off", "post_summary"} def ready(pr: PullRequest) -> bool: """R1: open, not a draft, and not marked work in progress.""" return not pr.draft and not _WIP.match(pr.title) def command(one: Said, operators: frozenset[str]) -> str | None: """R5: `stop` or `review`, as the whole of an operator's comment.""" if not one.key.startswith("comment-") or one.author not in operators: return None word = one.body.strip().strip(".!`").strip().lower() return word if word in ("stop", "review") else None class Scheduler: def __init__( self, store: StateStore, gitea: GiteaClient, reviewer: Reviewer, *, oplog: OpLog, scratch: Path, poll_s: float = 60.0, attempts: int = 2, patterns: frozenset[str] | None = None, ci_wait_s: float = 3600.0, ci_grace_s: float = 120.0, operators: frozenset[str] = frozenset(), max_rounds: int = 4, cancel: Cancel | None = None, clock: Callable[[], float] = time.time, ) -> None: self._store = store self._gitea = gitea self._reviewer = reviewer self._oplog = oplog self._scratch = scratch self._poll_s = poll_s self._attempts = attempts # The repositories reviewed, as patterns; None: every one it was given. self._patterns = patterns self._ci_wait_s = ci_wait_s self._ci_grace_s = ci_grace_s self._operators = operators self._max_rounds = max_rounds self._clock = clock self._said_waiting: set[str] = set() # "item@head", logged once self._stop = threading.Event() # The reviewer's driver's flag: a stop ends a run at its next turn. self._cancel = cancel or Cancel() self._repo_failing: dict[str, str] = {} # repo: why, logged on change # -- the loop --------------------------------------------------------------- def run_forever(self) -> None: while not self._stop.is_set(): if not self.round(): self._stop.wait(self._poll_s) def stop(self) -> None: """No more rounds, and a run in progress ends at its next turn, back to ready: a stop is no failure of the review.""" self._stop.set() self._cancel.set() def round(self) -> bool: """The wall, intake, then one run if one is due; True if one ran. An account that can write to a repository it reviews raises ``AccessRefused``: it stops, never reviews on (R3).""" try: repos = check( self._gitea, reviewed(self._gitea, self._patterns, self._oplog), self._oplog, ) self.recover() self.intake(repos) except GiteaError as exc: self._oplog.emit("intake.failed", cause=str(exc)[:300]) return False item = self._next() if item is None: return False self.review(item) return True # -- intake (R1) ------------------------------------------------------------ def intake(self, repos: list[str] | None = None) -> None: if repos is None: repos = reviewed(self._gitea, self._patterns, self._oplog) open_items = {i.id: i for i in self._store.work_items("open")} for repo in repos: try: prs = self._gitea.open_pull_requests(repo) except GiteaError as exc: # One repository Gitea won't list (an empty one, one without # pull requests) holds up none of the others; its items stay # as they are. Said when it starts, or the reason changes. why = str(exc)[:300] if self._repo_failing.get(repo) != why: self._repo_failing[repo] = why self._oplog.emit("intake.repo_failed", cause=why, repo=repo) continue if self._repo_failing.pop(repo, None) is not None: self._oplog.emit("intake.repo_ok", cause="listed again", repo=repo) seen: set[str] = set() for pr in prs: key = item_id(repo, pr.number) seen.add(key) try: self._see(open_items.get(key), repo, pr) except GiteaError as exc: # What was said couldn't be read: next round, then. self._oplog.emit( "intake.pr_failed", cause=str(exc)[:300], work_item=key ) for key, item in open_items.items(): if item.scope == repo and key not in seen: self._gone(item) def _see(self, item: WorkItem | None, repo: str, pr: PullRequest) -> None: """A PR as Gitea has it now: its head, its readiness, and what was said on it since the item last looked.""" if item is None: # Everything said so far is in its first review's brief. said = said_on(self._gitea, repo, pr.number) self._store.open_work_item( item_id(repo, pr.number), title=pr.title, scope=repo, position=READY if ready(pr) else WAITING, data=data( PR( repo, pr.number, head=pr.head_sha, seen_at=self._clock(), seen=tuple(one.key for one in said), ) ), cause="ready for review" if ready(pr) else "not ready", ) return if item.position == RUNNING: return # recovery's, not intake's known = pr_of(item) said = said_on(self._gitea, repo, pr.number) if "seen" not in item.data["pr"]: # Reviewed before M7.5: what was said then is taken in, unanswered. known = replace(known, seen=tuple(one.key for one in said)) moved = known.head != pr.head_sha stopped = item.position == STOPPED fresh = list(known.fresh) reset = moved asked = False why: list[str] = ["new push"] if moved else [] for one in said: if one.key in known.seen or one.author == self._reviewer.account: continue word = command(one, self._operators) if word is not None: self._acknowledge(item.id, repo, one, word) stopped = word == "stop" asked = word == "review" if asked: fresh = [] why.append(f"operator: {word}") continue # While stopped, this waits unanswered: `review` starts afresh. fresh.append(one.key) if one.author != pr.author: reset = True # someone else joined in (R2) if fresh != list(known.fresh): why.append("said since") new = replace( known, head=pr.head_sha, attempts=0 if moved or asked else known.attempts, seen_at=self._clock() if moved else known.seen_at, seen=tuple(dict.fromkeys([*known.seen, *(one.key for one in said)])), fresh=tuple(fresh), rounds=0 if reset or asked else known.rounds, full=known.full or asked, ) position = self._position( item.position, new, pr, moved=moved, stopped=stopped, reset=reset, asked=asked, ) if position == CAPPED: new = replace(new, fresh=()) # it stopped answering (R2) if new != pr_of(item) or position != item.position: self._store.update_work_item( item.id, position=position, data=data(new), cause=", ".join(why) or "PR changed", ) def _position( self, was: str, new: PR, pr: PullRequest, *, moved: bool, stopped: bool, reset: bool, asked: bool, ) -> str: if stopped: return STOPPED if was == CAPPED and not (reset or asked): return CAPPED if was == FAILED and not (moved or asked): return FAILED # failed at this head: new code or `review` retries if not ready(pr): return WAITING return READY if self._kind(new) is not None else REVIEWED def _acknowledge(self, item: str, repo: str, one: Said, word: str) -> None: """A command is answered with a reaction, not a comment: nothing for the PR's author to answer.""" self._oplog.emit(f"review.{word}", cause="operator", work_item=item) try: self._gitea.react(repo, int(one.key.split("-", 1)[1]), "+1") except GiteaError as exc: self._oplog.emit("review.unacknowledged", cause=str(exc)[:300]) def _gone(self, item: WorkItem) -> None: pr = pr_of(item) merged = self._gitea.pull_request(pr.repo, pr.number).merged self._store.close_work_item( item.id, outcome="completed" if merged else "abandoned", cause="PR merged" if merged else "PR closed", ) def _kind(self, known: PR) -> str | None: """The run an item has due, if any.""" if known.full or not known.reviewed: return FIRST if known.head != known.reviewed: return LATER if known.fresh: return SUMMARY if known.rounds >= self._max_rounds else REPLIES return None def _next(self) -> WorkItem | None: """The oldest ready item with a run due, a review's once its CI has settled; replies don't wait for CI.""" ready_items = [i for i in self._store.work_items("open") if i.position == READY] for item in sorted(ready_items, key=lambda i: i.opened_at): kind = self._kind(pr_of(item)) if kind is None: self._store.update_work_item( item.id, position=REVIEWED, data=item.data, cause="nothing due" ) continue if kind in (REPLIES, SUMMARY) or self._ci_settled(item): return item return None def _ci_settled(self, item: WorkItem) -> bool: """R1: a review judges the CI result, so a new head waits for its checks: reviewed once they've finished (passed or not), at once with no CI at all after a short grace for checks to appear, and anyway once they've been pending ``ci_wait_s``. Waiting also keeps reviews off a GPU that CI is using.""" known = pr_of(item) waited = self._clock() - known.seen_at try: state = self._gitea.checks(known.repo, known.head) except GiteaError as exc: self._oplog.emit("review.checks_unknown", cause=str(exc)[:300]) return False # asked again next round if state == "pending" and waited < self._ci_wait_s: key = f"{item.id}@{known.head}" if key not in self._said_waiting: self._said_waiting.add(key) self._oplog.emit( "review.waiting_for_ci", cause="checks pending", work_item=item.id ) return False return not (state == "none" and waited < self._ci_grace_s) def _seen(self, known: PR, head: str) -> float: """When ``head`` was first seen: kept for the head the item knew, now for one pushed during the review, so a retry still waits for CI. A retry at a head already past ``ci_wait_s`` doesn't wait again.""" return known.seen_at if head == known.head else self._clock() # -- a run ------------------------------------------------------------------ def review(self, item: WorkItem) -> None: known = pr_of(item) kind = self._kind(known) or FIRST self._store.update_work_item( item.id, position=RUNNING, data=data(replace(known, run=kind)), cause=f"{kind} run started", ) try: pr = self._gitea.pull_request(known.repo, known.number) said = said_on(self._gitea, known.repo, known.number) # Points a run before a crash already answered (their markers). done = answered(said, self._reviewer.account) known = replace(known, fresh=tuple(k for k in known.fresh if k not in done)) if kind == REPLIES and not known.fresh: self._settle(item, known, pr, "done", Draft(), posted=None) return scratch = Path(tempfile.mkdtemp(dir=self._scratch, prefix="review-")) repo = Repository( self._gitea, known.repo, {"head": pr.head_sha, "base": pr.base}, scratch, ) try: outcome, draft = self._reviewer.run( item.id, pr, repo, kind=kind, known=known, said=said ) finally: repo.close() if outcome.kind == "capped" and draft.comments: # Out of turns with points drafted: they're worth more posted # than lost (#156), as a review that says it's partial. self._oplog.emit( "review.partial", cause=outcome.note, work_item=item.id ) tool, summary = "post_review", PARTIAL.format(n=len(draft.comments)) elif outcome.kind == "stopped": self._interrupted(item, known, outcome.note) return elif outcome.kind != "wait": self._failed(item, known, pr, outcome.note or outcome.kind) return else: tool = outcome.tool summary = str((outcome.arguments or {}).get("summary", "")) posted = None if tool in _POSTING: kept = draft if tool == "post_review" else Draft() posted = self._reviewer.post( item.id, known.repo, pr, summary, kept, n=known.posts + 1 ) except Stopped as exc: # A stop broke off a model call: a stop all the same. self._interrupted(item, known, str(exc)) return except GiteaError as exc: self._failed(item, known, None, f"Gitea: {exc}") return except Exception as exc: # noqa: BLE001 -- counted, never a crash loop self._failed(item, known, None, repr(exc)) return self._settle(item, known, pr, tool, draft, posted) def _settle( self, item: WorkItem, known: PR, pr: PullRequest, tool: str, draft: Draft, posted: Posted | None, ) -> None: """A run ended on ``tool``: what it leaves the item with.""" reviewing = tool in ("post_review", "sign_off", "still_happy") counted = tool in ("post_review", "sign_off") or bool(draft.replied) if posted is not None: self._oplog.emit( "review.posted", cause=tool, work_item=item.id, review=posted.review, inline=posted.inline, linked=posted.linked, ) else: self._oplog.emit( f"review.{tool}", cause="run ended", work_item=item.id, replied=len(draft.replied), skipped=len(draft.skipped), ) happy = {"post_review": False, "sign_off": True, "still_happy": True} new = replace( known, head=pr.head_sha, reviewed=pr.head_sha if reviewing else known.reviewed, attempts=0, seen_at=self._seen(known, pr.head_sha), fresh=(), rounds=known.rounds + 1 if counted else known.rounds, posts=known.posts + (1 if posted is not None else 0), happy=happy.get(tool, known.happy), full=False, run="", ) self._store.update_work_item( item.id, position=CAPPED if tool == "post_summary" else REVIEWED, data=data(new), cause=f"{tool}", ) def _interrupted(self, item: WorkItem, known: PR, why: str) -> None: """A stop ended the run: nothing drafted is posted, no attempt is counted, and it runs again as it was. Replies it posted carry their markers, so the next run doesn't repeat them.""" self._oplog.emit("review.interrupted", cause=why[:300], work_item=item.id) self._store.update_work_item( item.id, position=READY, data=data(replace(known, run="")), cause="stopped mid-run", ) def _failed( self, item: WorkItem, known: PR, pr: PullRequest | None, why: str ) -> None: attempts = known.attempts + 1 head = pr.head_sha if pr else known.head new = replace( known, head=head, attempts=attempts, seen_at=self._seen(known, head), run="", ) self._oplog.emit("review.failed", cause=why[:300], work_item=item.id) if attempts < self._attempts: self._store.update_work_item( item.id, position=READY, data=data(new), cause="review failed" ) return # Said once on the PR, so its author isn't left waiting on nothing. try: self._gitea.post_comment( known.repo, known.number, f"nisse-review couldn't review this at {head[:12]} " f"({attempts} tries; the last: {why[:300]}). It tries again " "after the next push, or an operator's `review`.", ) except GiteaError as exc: self._oplog.emit("review.unannounced", cause=str(exc)[:300]) self._store.update_work_item( item.id, position=FAILED, data=data(new), cause="review kept failing" ) # -- recovery (L8) ---------------------------------------------------------- def recover(self) -> None: """Items left running by a crash: posted, or to run again, as Gitea says; a pending intent is resolved the same way.""" pending = { i.work_item: i for i in self._store.pending_intents() if i.kind == POST_REVIEW } for item in self._store.work_items("open"): if item.position != RUNNING: continue known = pr_of(item) intent = pending.get(item.id) if intent is None and known.run == REPLIES: # Replies carry their points' keys: the re-run skips those. self._store.update_work_item( item.id, position=READY, data=data(replace(known, run="")), cause="recovered after a restart", ) continue head = intent.payload["head"] if intent else known.head n = int(intent.payload.get("n", 1)) if intent else known.posts + 1 posted = self._reviewer.posted(known.repo, known.number, head, n) if intent is not None: if posted: self._store.commit_intent(intent.id, cause="review found posted") else: self._store.abort_intent(intent.id, cause="review not found") if posted: new = replace( known, head=head, reviewed=head if known.run != SUMMARY else known.reviewed, seen_at=self._seen(known, head), fresh=(), rounds=known.rounds + 1, posts=n, happy=False, # a review or a sign-off: it can't tell which full=False, run="", ) position = CAPPED if known.run == SUMMARY else REVIEWED else: new = replace(known, head=head, seen_at=self._seen(known, head), run="") position = READY self._store.update_work_item( item.id, position=position, data=data(new), cause="recovered after a restart", )