diff --git a/gitea_saas/gitea_client.py b/gitea_saas/gitea_client.py index bad6643..65b8b90 100644 --- a/gitea_saas/gitea_client.py +++ b/gitea_saas/gitea_client.py @@ -153,8 +153,11 @@ class GiteaClient: """Map a raw Gitea PR record to the candidate shape. Every candidate supplies at least ``number``, ``body``, ``head_sha``, - ``base_ref``, ``labels`` (label NAMES), and ``repo`` — the shape - ``ReviewPolicy`` and the review gate consume. + ``base_ref``, ``labels`` (label NAMES), ``repo``, ``merged``, and + ``state`` — the shape ``ReviewPolicy`` and the review gate consume. + ``merged``/``state`` are the raw Gitea record's values (``merged`` is + a boolean; only ``merged is True`` proves a merge — ``state == + "closed"`` alone can also mean closed without merging). """ head = pr.get("head") or {} base = pr.get("base") or {} @@ -167,6 +170,8 @@ class GiteaClient: "base_ref": base.get("ref"), "labels": labels, "repo": repo, + "merged": bool(pr.get("merged")), + "state": pr.get("state"), } # -- API surface used by the factory front-end (issue #4) -------------- @@ -257,9 +262,11 @@ class GiteaClient: def get_pr(self, repo: str, number: int) -> dict: """Fetch a PR in the normalized candidate shape. - ``{number, body, head_sha, base_ref, labels: [names], repo, - changed_files: [filename, ...]}`` — the review gate consumes - ``head_sha`` and ``changed_files``; policy consumes the rest. + ``{number, body, head_sha, base_ref, labels: [names], repo, merged, + state, changed_files: [filename, ...]}`` — the review gate consumes + ``head_sha`` and ``changed_files``; policy consumes the rest + (``merged``/``state`` let recovery positively distinguish a merged + PR from a merely closed one). """ pr = self._request("GET", f"{repo}/pulls/{number}") or {} files = self._request("GET", f"{repo}/pulls/{number}/files") or [] diff --git a/gitea_saas/review_policy.py b/gitea_saas/review_policy.py index fb6ddbe..73656ac 100644 --- a/gitea_saas/review_policy.py +++ b/gitea_saas/review_policy.py @@ -9,8 +9,10 @@ Seams (all duck-typed, faked in tests): (see ``billing_core.BillingCore``). * ``gitea`` — Gitea client: ``list_open_prs(repo=None)``, ``get_pr(repo, n)``, ``get_issue(repo, n)``, ``post_comment``, - ``add_label``, ``remove_label``, ``merge_pr(repo, n) -> bool``, - ``close_issue(repo, n)``. + ``add_label``, ``remove_label``, ``merge_pr(repo, n, + expected_head_sha=None) -> bool``, ``close_issue(repo, n)``. The + normalized PR shape carries ``merged`` (bool) and ``state`` so the + policy can positively distinguish a merged PR from a merely closed one. * ``reviewer`` — the review gate: ``review_pr(repo, pr, linked_issue, repo_dir=...) -> Verdict`` and ``post_verdict(repo, pr, verdict)``. @@ -31,6 +33,42 @@ head moves. The table is bootstrapped here (``CREATE TABLE IF NOT EXISTS``) because ``schema.sql`` belongs to issue #1's PR and is out of this issue's allowed file set — same schema intent, idempotent. +Transactional outcome application (issue #29): a head is recorded into +``reviews`` (i.e. considered COMPLETED) only after its required policy +action has fully succeeded. Failure/incomplete attempts are persisted in +the companion ``review_state`` table — one row per head carrying the +attempt count, the last error, and idempotent step checkpoints +(``merged_step``, ``closed_step``, ``debit_step``, ``label_step``, +``comment_step``) — so a transient failure at any boundary (verdict +generation, label, comment, merge, close, debit) leaves the same +unchanged head eligible for retry on the next poll instead of stranding it. + +Crash-window recovery is NOT driven by the open-PR scan. A merge is the +point of no return: once the PR is truly merged it disappears from +``list_open_prs()``, so any later failure (debit, label removal, issue +close, comment) would strand the persisted row. Therefore the outcome +plan (verdict, tier, linked issue) is persisted into ``review_state`` +BEFORE the merge, every post-merge side effect is checkpointed, and +``recover_incomplete()`` resumes those rows directly from the retry table +(via ``get_pr`` + checkpoints) — independently of any open-PR listing — +before the regular poll scan runs. A merge is only ever adopted when the +client reports positive proof (``merged is True``, now part of the +production normalized shape); ``state == "closed"`` without a merge is +never treated as a merge, so a closed-but-unmerged PR cannot trigger a +debit, label removal, issue close, or success comment. Retries re-apply +only the steps that have not already been checkpointed, so a partial side +effect can never double-merge a PR, double-close an issue, double-remove +a label, or double-post the success comment; the delivery debit is +additionally protected by its existing billing-core idempotency key +``#@``. The merge request itself is pinned to the +reviewed head: ``_safe_merge`` passes ``expected_head_sha`` through to the +client so a branch that moved after review is refused before any POST +(issue #28 client contract). Legacy ``reviews`` rows (pre-#29 schema) are +preserved untouched and keep their completed meaning; the retry state +lives entirely in the new table, so no migration of existing records is +needed (new step columns are ALTER-added idempotently for databases +created by earlier builds). + Delivery debit: implemented inline here behind the ``debit_for_merge`` signature shared with the delivery issue (#7); when delivery lands it can refactor to call it. FAIL-CLOSED: this module never fake-succeeds a debit — @@ -63,6 +101,45 @@ CREATE TABLE IF NOT EXISTS reviews ( ) """ +# Retry state for heads whose outcome application is in flight or failed +# (issue #29). One row per head; ``attempts`` counts failed/incomplete +# attempts, ``last_error`` keeps the most recent failure, and the step +# checkpoints record which idempotent side effects already happened so a +# retry resumes instead of re-applying them. ``tier``/``linked_issue`` +# snapshot the outcome plan BEFORE the merge so ``recover_incomplete()`` +# can finish the post-merge steps even after the PR leaves the open-PR +# scan. Rows are deleted once the head completes (a ``reviews`` row is +# written instead). +_REVIEW_STATE_DDL = """ +CREATE TABLE IF NOT EXISTS review_state ( + repo TEXT NOT NULL, + pr INTEGER NOT NULL, + head_sha TEXT NOT NULL, + verdict TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + last_attempt_at TEXT, + merged_step INTEGER NOT NULL DEFAULT 0, + closed_step INTEGER NOT NULL DEFAULT 0, + debit_step INTEGER NOT NULL DEFAULT 0, + label_step INTEGER NOT NULL DEFAULT 0, + comment_step INTEGER NOT NULL DEFAULT 0, + tier TEXT, + linked_issue INTEGER, + PRIMARY KEY (repo, pr, head_sha) +) +""" + +# Columns added after the initial #29 schema shipped (compat: databases +# created by earlier builds get these via idempotent ALTERs). +_REVIEW_STATE_EXTRA_COLUMNS = { + "debit_step": "debit_step INTEGER NOT NULL DEFAULT 0", + "label_step": "label_step INTEGER NOT NULL DEFAULT 0", + "comment_step": "comment_step INTEGER NOT NULL DEFAULT 0", + "tier": "tier TEXT", + "linked_issue": "linked_issue INTEGER", +} + _LINK_RE = re.compile(r"(?im)\b(?:closes|fixes|resolves|close|fix|resolve)\s+#(\d+)\b") # Markdown-tolerant: "**Tier:** gold" / "- Tier: gold" / "**Client:** acme" all parse # (the value can sit after formatting chars like "*" / "**" following the colon). @@ -98,26 +175,148 @@ class ReviewPolicy: self._reviews_conn = sqlite3.connect(str(self._db_path)) self._reviews_conn.execute("PRAGMA journal_mode=WAL") self._reviews_conn.execute(_REVIEWS_DDL) + self._reviews_conn.execute(_REVIEW_STATE_DDL) + self._ensure_review_state_columns() self._reviews_conn.commit() + def _ensure_review_state_columns(self) -> None: + """Idempotently add columns introduced after the first #29 build.""" + existing = {row[1] for row in self._reviews_conn.execute( + "PRAGMA table_info(review_state)")} + for column, ddl in _REVIEW_STATE_EXTRA_COLUMNS.items(): + if column not in existing: + self._reviews_conn.execute( + f"ALTER TABLE review_state ADD COLUMN {ddl}") + # ------------------------------------------------------------------ - # Reviews table (re-review safety) + # Reviews table (completed-history) + review_state (retry metadata) # ------------------------------------------------------------------ def _already_reviewed(self, repo: str, pr: int, head_sha: str) -> bool: + """True once a head's outcome application COMPLETED successfully. + + A failed/incomplete attempt lives only in ``review_state`` and does + not make a head "reviewed": the same unchanged head must stay + eligible for retry. + """ row = self._reviews_conn.execute( "SELECT 1 FROM reviews WHERE repo = ? AND pr = ? AND head_sha = ?", (repo, pr, head_sha)).fetchone() return row is not None - def _record_review(self, repo: str, pr: int, head_sha: str, verdict: str) -> None: + def _record_completed(self, repo: str, pr: int, head_sha: str, verdict: str) -> None: + """Persist a fully applied outcome; drop the retry state for the head.""" self._reviews_conn.execute( "INSERT OR IGNORE INTO reviews (repo, pr, head_sha, verdict, reviewed_at)" " VALUES (?, ?, ?, ?, ?)", (repo, pr, head_sha, verdict, datetime.now(timezone.utc).isoformat())) + self._reviews_conn.execute( + "DELETE FROM review_state WHERE repo = ? AND pr = ? AND head_sha = ?", + (repo, pr, head_sha)) self._reviews_conn.commit() + def _record_failure(self, repo: str, pr: int, head_sha: str, + verdict: Optional[str], error: str) -> None: + """Persist a failed/incomplete attempt (upserts per head). + + Step checkpoints on an existing row are preserved, so a retry of a + partially applied outcome resumes from where it left off. + """ + now = datetime.now(timezone.utc).isoformat() + self._reviews_conn.execute( + "INSERT INTO review_state (repo, pr, head_sha, verdict, attempts," + " last_error, last_attempt_at)" + " VALUES (?, ?, ?, ?, 1, ?, ?)" + " ON CONFLICT(repo, pr, head_sha) DO UPDATE SET" + " verdict = excluded.verdict," + " attempts = review_state.attempts + 1," + " last_error = excluded.last_error," + " last_attempt_at = excluded.last_attempt_at", + (repo, pr, head_sha, verdict, error, now)) + self._reviews_conn.commit() + + _STEP_COLUMNS = {"merged": "merged_step", "closed": "closed_step", + "debit": "debit_step", "label": "label_step", + "comment": "comment_step"} + + _POST_MERGE_STEPS = ("debit", "label", "closed", "comment") + + def _ensure_plan(self, repo: Optional[str], pr: int, head_sha: Optional[str], + verdict: str, tier: Optional[str], + issue_num: Optional[int]) -> None: + """Persist the outcome plan BEFORE the first side effect (#29). + + The merge is the point of no return: once merged, the PR leaves + the open-PR scan, so every remaining step (debit, label removal, + issue close, success comment) must be recoverable from this row + alone via :meth:`recover_incomplete`. Upserts; preserves attempt + counts and step checkpoints from earlier attempts. + """ + self._reviews_conn.execute( + "INSERT INTO review_state (repo, pr, head_sha, verdict, tier," + " linked_issue)" + " VALUES (?, ?, ?, ?, ?, ?)" + " ON CONFLICT(repo, pr, head_sha) DO UPDATE SET" + " verdict = excluded.verdict," + " tier = excluded.tier," + " linked_issue = excluded.linked_issue", + (repo, pr, head_sha, verdict, tier, issue_num)) + self._reviews_conn.commit() + + def _linked_issue(self, repo: Optional[str], pr: int, + head_sha: Optional[str]) -> Optional[int]: + row = self._reviews_conn.execute( + "SELECT linked_issue FROM review_state" + " WHERE repo = ? AND pr = ? AND head_sha = ?", + (repo, pr, head_sha)).fetchone() + return row[0] if row else None + + def _step_done(self, repo: Optional[str], pr: int, + head_sha: Optional[str], step: str) -> bool: + """True when the idempotent step already succeeded for this head.""" + column = self._STEP_COLUMNS.get(step) + if column is None: + raise ValueError(f"unknown retry step {step!r}") + row = self._reviews_conn.execute( + f"SELECT {column} FROM review_state" + " WHERE repo = ? AND pr = ? AND head_sha = ?", + (repo, pr, head_sha)).fetchone() + return bool(row and row[0]) + + def _mark_step(self, repo: Optional[str], pr: int, head_sha: Optional[str], + step: str) -> None: + """Checkpoint a succeeded side effect so a retry never re-applies it.""" + column = self._STEP_COLUMNS.get(step) + if column is None: + raise ValueError(f"unknown retry step {step!r}") + self._reviews_conn.execute( + f"INSERT INTO review_state (repo, pr, head_sha, {column})" + " VALUES (?, ?, ?, 1)" + f" ON CONFLICT(repo, pr, head_sha) DO UPDATE SET {column} = 1", + (repo, pr, head_sha)) + self._reviews_conn.commit() + + def _attempts(self, repo: str, pr: int, head_sha: str) -> int: + row = self._reviews_conn.execute( + "SELECT attempts FROM review_state" + " WHERE repo = ? AND pr = ? AND head_sha = ?", + (repo, pr, head_sha)).fetchone() + return row[0] if row else 0 + + @staticmethod + def _error_text(exc: Exception) -> str: + return f"{type(exc).__name__}: {exc}" + + def _failure_result(self, repo: Optional[str], pr: int, head_sha: str, + verdict: Optional[str], exc: Exception) -> dict: + return { + "repo": repo, "pr": pr, "head_sha": head_sha, "verdict": verdict, + "outcome": "failed", + "error": {"type": type(exc).__name__, "message": str(exc)}, + "attempts": self._attempts(repo or "", pr, head_sha), + } + # ------------------------------------------------------------------ # Polling # ------------------------------------------------------------------ @@ -138,24 +337,183 @@ class ReviewPolicy: return str(issue or "") def poll_and_review(self, repo: Optional[str] = None) -> list[dict]: - """Review every in-scope PR at its current head; apply tier policy.""" - results = [] + """Review every in-scope PR at its current head; apply tier policy. + + Transactional + retryable (issue #29): a head is only recorded as + completed AFTER its required policy action succeeded; any failure — + verdict generation or a policy action — is persisted as a retryable + attempt and reported as a structured failure result, leaving the + same unchanged head eligible for the next poll. + + Stranded post-merge applications are recovered FIRST, directly + from the ``review_state`` table: a truly merged PR is no longer + returned by ``list_open_prs()``, so the open-PR scan can never + reach those rows again (see :meth:`recover_incomplete`). + """ + results = list(self.recover_incomplete(repo)) for pr in self.list_candidate_prs(repo): + repo_key = repo or (pr.get("repo") or "") pr_number = pr["number"] head_sha = pr.get("head_sha") or "unknown" - if self._already_reviewed(repo or (pr.get("repo") or ""), pr_number, head_sha): + if self._already_reviewed(repo_key, pr_number, head_sha): continue - pr_body = pr.get("body") or "" - issue_body = self._fetch_issue_body(repo, linked_issue_number(pr_body)) - verdict = self.reviewer.review_pr( - repo, pr_number, issue_body, - repo_dir=self.config.get("worktrees_dir")) - self._record_review(repo or "", pr_number, head_sha, verdict.verdict) - action = self._apply_policy(repo, pr, pr_body, issue_body, verdict) + try: + pr_body = pr.get("body") or "" + issue_body = self._fetch_issue_body(repo, linked_issue_number(pr_body)) + verdict = self.reviewer.review_pr( + repo, pr_number, issue_body, + repo_dir=self.config.get("worktrees_dir")) + except Exception as exc: # noqa: BLE001 - any failure must be retryable + self._record_failure(repo_key, pr_number, head_sha, None, + self._error_text(exc)) + results.append(self._failure_result(repo, pr_number, head_sha, + None, exc)) + continue + try: + action = self._apply_policy(repo, pr, pr_body, issue_body, verdict) + except Exception as exc: # noqa: BLE001 - any failure must be retryable + self._record_failure(repo_key, pr_number, head_sha, verdict.verdict, + self._error_text(exc)) + results.append(self._failure_result(repo, pr_number, head_sha, + verdict.verdict, exc)) + continue + if action.get("action") in ("merge_failed", "merge_but_debit_failed"): + # The required action did NOT succeed: keep the head retryable + # and persist the incomplete attempt (fail-closed branches + # report a structured non-completion instead of raising). + self._record_failure(repo_key, pr_number, head_sha, verdict.verdict, + f"code={action.get('action')}") + results.append({ + "repo": repo, "pr": pr_number, "head_sha": head_sha, + "verdict": verdict.verdict, "outcome": "failed", + "error": {"code": action.get("action")}, + "attempts": self._attempts(repo_key, pr_number, head_sha), + **action}) + continue + self._record_completed(repo_key, pr_number, head_sha, verdict.verdict) results.append({"repo": repo, "pr": pr_number, "head_sha": head_sha, "verdict": verdict.verdict, **action}) return results + def recover_incomplete(self, repo: Optional[str] = None) -> list[dict]: + """Resume stranded outcome applications directly from retry state. + + Independent of the open-PR scan: ``list_open_prs()`` cannot reach + a head whose PR was truly merged (it is no longer open), so any + post-merge step that failed — debit, label removal, issue close, + success comment — would otherwise be lost forever. This method + walks ``review_state`` rows and finishes each one from its + checkpoints, re-fetching the PR by number via ``get_pr`` (which + works for closed/merged PRs) rather than from any open-PR listing. + + Only rows whose merge is already confirmed (checkpointed, or the + live PR positively reports ``merged is True`` at the EXACT + planned/reviewed head — the production normalized shape includes + the raw ``merged`` boolean and ``head_sha``) are handled here; rows + for still-open PRs are left to the regular poll scan, which can + reach them. ``state == "closed"`` alone never counts as merged: a + PR closed without merging must not trigger a debit, label removal, + issue close, or success comment. A PR merged at a DIFFERENT head + than the plan reviewed is never adopted — it fails closed with a + structured ``HEAD_MISMATCH`` and no side effects (the debit, + label, close and comment steps all belong to the reviewed head, + not to whatever head actually landed). + Completed heads are recorded into ``reviews`` and their retry row + deleted; a step that fails again is persisted with an incremented + attempt count for the next recovery pass. + """ + results = [] + where, params = (" WHERE repo = ?", [repo]) if repo else ("", []) + rows = self._reviews_conn.execute( + f"SELECT repo, pr, head_sha, verdict FROM review_state{where}", + params).fetchall() + for row in rows: + row_repo, pr_number, head_sha, verdict = row + if verdict is None: + continue # no policy action started yet; open scan owns it + try: + live = None + try: + live = self.gitea.get_pr(row_repo, pr_number) or {} + except Exception: # noqa: BLE001 - PR may be gone; checkpoints decide + live = None + pr_merged = bool(live and live.get("merged") is True) + merged_done = self._step_done(row_repo, pr_number, head_sha, + "merged") + if pr_merged and not merged_done: + # Crash window: the merge applied server-side but the + # checkpoint write never happened — adopt it ONLY when + # the live merged head is the EXACT planned/reviewed + # head (head-integrity gate, governor 10539). A merge + # of a DIFFERENT head must never be attributed to this + # plan's debit/label/close/comment steps. + live_head = (live or {}).get("head_sha") + if live_head != head_sha: + message = ( + f"merged head mismatch: live head {live_head!r} " + f"!= planned/reviewed head {head_sha!r} — the " + f"outcome application for the reviewed head is " + f"refused (no debit, no label removal, no issue " + f"close, no success comment)") + self._record_failure(row_repo, pr_number, head_sha, + verdict, message) + results.append({ + "repo": row_repo, "pr": pr_number, + "head_sha": head_sha, "verdict": verdict, + "outcome": "failed", + "error": {"code": "HEAD_MISMATCH", + "message": message}, + "attempts": self._attempts( + row_repo, pr_number, head_sha), + }) + continue + # Same-head crash window: adopt and finish the plan. + self._mark_step(row_repo, pr_number, head_sha, "merged") + merged_done = True + if not merged_done: + continue # still reachable via the open scan; leave it + + issue_body = self._fetch_issue_body( + row_repo, self._linked_issue(row_repo, pr_number, head_sha)) + if not self._step_done(row_repo, pr_number, head_sha, "debit"): + ok, debit = self.debit_for_merge( + row_repo, pr_number, head_sha, issue_body) + if not ok: + raise RuntimeError( + f"recover debit failed: {debit.get('error') or debit}") + self._mark_step(row_repo, pr_number, head_sha, "debit") + else: + debit = {"debit": "recorded", "replayed": True} + + if not self._step_done(row_repo, pr_number, head_sha, "label"): + self.gitea.remove_label(row_repo, pr_number, "in-review") + self._mark_step(row_repo, pr_number, head_sha, "label") + + self._safe_close_issue(row_repo, pr_number, head_sha, + self._linked_issue(row_repo, pr_number, + head_sha)) + + if not self._step_done(row_repo, pr_number, head_sha, "comment"): + summary = ("delivery debit recorded" + if debit.get("debit") == "recorded" + else "no delivery debit (tier rate 0)") + self.gitea.post_comment( + row_repo, pr_number, + f"Merged by review-gate — {summary}.") + self._mark_step(row_repo, pr_number, head_sha, "comment") + + self._record_completed(row_repo, pr_number, head_sha, verdict) + results.append({"repo": row_repo, "pr": pr_number, + "head_sha": head_sha, "verdict": verdict, + "outcome": "merged", "action": "merged", + "debit": debit, "recovered": True}) + except Exception as exc: # noqa: BLE001 - any failure must be retryable + self._record_failure(row_repo, pr_number, head_sha, verdict, + self._error_text(exc)) + results.append(self._failure_result(row_repo, pr_number, + head_sha, verdict, exc)) + return results + # ------------------------------------------------------------------ # Tier resolution + delivery debit # ------------------------------------------------------------------ @@ -216,6 +574,77 @@ class ReviewPolicy: # Tier policy application # ------------------------------------------------------------------ + def _safe_merge(self, repo: Optional[str], pr: dict) -> bool: + """Merge guarded for idempotent retry (issue #29). + + Never merges twice for the same head: the ``merged_step`` checkpoint + makes a retry skip an already-applied merge, and when the fetched PR + shape already positively reports ``merged is True`` (crash window + between the API side effect and the checkpoint write) the merge is + skipped as well. ``state == "closed"`` alone is NOT adopted — a PR + can be closed without merging, and adopting it would debit the + client, strip labels, and close the linked issue for a merge that + never happened. + + The merge request is pinned to the REVIEWED head: the exact + ``head_sha`` this policy reviewed is passed through to + ``gitea.merge_pr(..., expected_head_sha=...)``, whose production + implementation (issue #28) fails closed before any POST when the + live branch moved after review — an unreviewed SHA can never be + merged. A candidate that carries NO reviewed head_sha is rejected + BEFORE the client is called (head-integrity gate, governor 10539): + ``expected_head_sha=None`` would arm the client's backward- + compatible live-head fallback and could merge a head this policy + never reviewed, so the merge is refused up front and the failure + is surfaced as a structured retryable result with zero side + effects. + """ + repo_key = repo or "" + pr_number = pr["number"] + reviewed_head = pr.get("head_sha") or "" + if not reviewed_head: + raise ValueError( + "merge refused: candidate PR carries no reviewed head_sha — " + "expected_head_sha must be pinned; refusing to fall back to " + "the live head") + head_sha = reviewed_head + if self._step_done(repo_key, pr_number, head_sha, "merged"): + return True + if pr.get("merged") is True: + self._mark_step(repo_key, pr_number, head_sha, "merged") + return True + merged = self.gitea.merge_pr( + repo, pr_number, expected_head_sha=reviewed_head) + if merged: + self._mark_step(repo_key, pr_number, head_sha, "merged") + return merged + + def _safe_close_issue(self, repo: Optional[str], pr_number: int, + head_sha: Optional[str], issue_num: Optional[int]) -> None: + """Close guarded for idempotent retry (issue #29). + + Skips when the ``closed_step`` checkpoint says this head already + closed the issue, when the head already completed its outcome + application (poll path), or when the live issue state is already + closed. ``head_sha`` may be None (webhook path): the checkpoint is + keyed on the head only when one is known. + """ + repo_key = repo or "" + head = head_sha or "unknown" + if issue_num is None or self._step_done(repo_key, pr_number, head, "closed"): + return + if self._already_reviewed(repo_key, pr_number, head): + return + try: + issue = self.gitea.get_issue(repo, issue_num) + except Exception: # noqa: BLE001 - fail-closed: let close_issue decide + issue = None + if isinstance(issue, dict) and issue.get("state") == "closed": + self._mark_step(repo_key, pr_number, head, "closed") + return + self.gitea.close_issue(repo, issue_num) + self._mark_step(repo_key, pr_number, head, "closed") + def _apply_policy(self, repo: Optional[str], pr: dict, pr_body: str, issue_body: str, verdict) -> dict: pr_number = pr["number"] @@ -230,22 +659,43 @@ class ReviewPolicy: "Approved — awaiting operator merge " f"(tier `{tier}`, auto_merge off).") return {"action": "approved_awaiting_merge", "tier": tier} - merged = self.gitea.merge_pr(repo, pr_number) + # Persist the outcome plan BEFORE the merge: once the PR is + # truly merged it leaves the open-PR scan, so every following + # step must be recoverable from the row alone (issue #29 / + # review bounce: merged PRs vanish from list_open_prs()). + head_sha = pr.get("head_sha") or "" + if not head_sha: + # Head-integrity gate (governor 10539): a candidate without + # a reviewed head_sha is rejected BEFORE any plan write or + # client call — expected_head_sha must be pinned, never the + # client's live-head fallback. + raise ValueError( + "merge refused: candidate PR carries no reviewed head_sha — " + "expected_head_sha must be pinned; refusing to fall back to " + "the live head") + issue_num = linked_issue_number(pr_body) + self._ensure_plan(repo, pr_number, head_sha, verdict.verdict, + tier, issue_num) + merged = self._safe_merge(repo, pr) if not merged: self.gitea.post_comment( repo, pr_number, "[review-gate] merge failed — PR left open for manual review.") return {"action": "merge_failed", "tier": tier} - ok, debit = self.debit_for_merge(repo, pr_number, pr.get("head_sha"), issue_body) - self.gitea.remove_label(repo, pr_number, "in-review") - issue_num = linked_issue_number(pr_body) + ok, debit = self.debit_for_merge(repo, pr_number, head_sha, issue_body) if ok: - if issue_num is not None: - self.gitea.close_issue(repo, issue_num) + self._mark_step(repo, pr_number, head_sha, "debit") + if not self._step_done(repo, pr_number, head_sha, "label"): + self.gitea.remove_label(repo, pr_number, "in-review") + self._mark_step(repo, pr_number, head_sha, "label") + if ok: + self._safe_close_issue(repo, pr_number, head_sha, issue_num) summary = "delivery debit recorded" if debit.get("debit") == "recorded" \ else "no delivery debit (tier rate 0)" - self.gitea.post_comment(repo, pr_number, - f"Merged by review-gate — {summary}.") + if not self._step_done(repo, pr_number, head_sha, "comment"): + self.gitea.post_comment(repo, pr_number, + f"Merged by review-gate — {summary}.") + self._mark_step(repo, pr_number, head_sha, "comment") return {"action": "merged", "tier": tier, "debit": debit} # fail-closed: merge happened but the debit failed — do NOT close # the issue, tell the operator. @@ -290,8 +740,7 @@ class ReviewPolicy: else "no debit for this tier.") ) issue_num = linked_issue_number(pr.get("body") or "") - if issue_num is not None: - self.gitea.close_issue(repo, issue_num) + self._safe_close_issue(repo, pr_number, head_sha, issue_num) return ok, {"pr": pr_number, "head_sha": head_sha, "debit": debit} self.gitea.post_comment( repo, pr_number, diff --git a/tests/test_review_policy.py b/tests/test_review_policy.py index 93017aa..8714b30 100644 --- a/tests/test_review_policy.py +++ b/tests/test_review_policy.py @@ -27,10 +27,19 @@ from gitea_saas.reviewer import Verdict # noqa: E402 class FakeGitea: - """In-memory Gitea seam recording every mutation.""" + """In-memory Gitea seam recording every mutation. + + Mirrors the production client's NORMALIZED shape (issue #28/#31): + PRs carry ``merged`` (bool) and ``state`` exactly like + ``GiteaClient._normalize_pr`` output, ``list_open_prs`` returns only + open-unmerged PRs, ``get_pr`` returns the live record, and a + successful ``merge_pr`` flips the live record to ``merged=True`` / + ``state="closed"`` — so tests exercise the production-shaped dict, not + a richer test-only one. + """ def __init__(self, prs=None, issues=None): - self.prs = list(prs or []) # [{number, head_sha, labels, body}] + self.prs = list(prs or []) # [{number, head_sha, labels, body, merged, state}] self.issues = dict(issues or {}) # number -> {"body": ...} self.comments = [] self.labels_added = [] @@ -38,12 +47,21 @@ class FakeGitea: self.merged = [] self.closed = [] self.merge_results = {} # pr -> bool (default True) + self.fail_next = {} # method -> one-shot Exception to raise + + def _maybe_fail(self, name): + """Raise a one-shot injected failure (issue #29 retry tests).""" + exc = self.fail_next.pop(name, None) + if exc is not None: + raise exc def _pr(self, number): return next(p for p in self.prs if p["number"] == number) def list_open_prs(self, repo=None): - return [dict(p) for p in self.prs] + # Production behavior: ?state=open returns open, unmerged PRs only. + return [dict(p) for p in self.prs + if p.get("state") == "open" and not p.get("merged")] def get_pr(self, repo, number): return dict(self._pr(number)) @@ -52,19 +70,34 @@ class FakeGitea: return dict(self.issues[number]) def post_comment(self, repo, number, body): + self._maybe_fail("post_comment") self.comments.append((repo, number, body)) def add_label(self, repo, number, label): + self._maybe_fail("add_label") self.labels_added.append((repo, number, label)) def remove_label(self, repo, number, label): + self._maybe_fail("remove_label") self.labels_removed.append((repo, number, label)) - def merge_pr(self, repo, number): + def merge_pr(self, repo, number, expected_head_sha=None): + self._maybe_fail("merge_pr") + live = self._pr(number) + if expected_head_sha is not None and live.get("head_sha") != expected_head_sha: + raise RuntimeError( + f"merge_pr: PR {number} head is {live.get('head_sha')!r} but the " + f"reviewed head is {expected_head_sha!r}; refusing to merge " + "a branch that moved after review") self.merged.append((repo, number)) - return self.merge_results.get(number, True) + merged = self.merge_results.get(number, True) + if merged: + live["merged"] = True + live["state"] = "closed" + return merged def close_issue(self, repo, number): + self._maybe_fail("close_issue") self.closed.append((repo, number)) @@ -96,7 +129,10 @@ def make_verdict(pr_number, verdict, head_sha="sha-abc"): def pr(number, head_sha="sha-abc", labels=("in-review",), body="Closes #1\nSome PR"): - return {"number": number, "head_sha": head_sha, "labels": list(labels), "body": body} + # Production-normalized shape (as GiteaClient._normalize_pr returns): + # merged bool + state, defaults open / not merged. + return {"number": number, "head_sha": head_sha, "labels": list(labels), + "body": body, "merged": False, "state": "open"} def tier_issue(tier, number=1): @@ -154,8 +190,12 @@ def test_review_once_per_head_and_skip_same_head(policy_factory): def test_head_move_triggers_rereview(policy_factory): + # Standard tier keeps the PR OPEN after the first poll (comment only), + # so a head move is observable by the next scan and must re-review. + # (A gold-tier poll merges, and a merged PR leaves the open scan — the + # production-shaped fake reflects that.) policy, gitea, reviewer, _, _ = policy_factory( - gitea_prs=[pr(10, head_sha="aaa")], gitea_issues=tier_issue("gold"), + gitea_prs=[pr(10, head_sha="aaa")], gitea_issues=tier_issue("standard"), verdicts={10: make_verdict(10, APPROVE)}) policy.poll_and_review("gitea-saas-core") @@ -386,3 +426,526 @@ def test_handle_merge_event_records_receipt_and_closes(policy_factory): debits = [e for e in ledger["entries"] if e["kind"] == "debit"] assert ok and len(debits) == 1 assert debits[0]["idempotency_key"] == "gitea-saas-core#70@sha-i" + + +# --------------------------------------------------------------------------- +# Transactional outcome application + retry (issue #29) +# --------------------------------------------------------------------------- + +def _ledger_debits(core, client_id): + ok, ledger = core.list_ledger(client_id) + assert ok + return [e for e in ledger["entries"] if e["kind"] == "debit"] + + +def test_verdict_failure_reported_and_retried(policy_factory): + # Verdict generation raises once: structured failure + persisted retry + # metadata; the same unchanged head completes exactly once next poll. + calls = {"n": 0} + + def flaky(repo, pr): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("verdict generation boom") + return make_verdict(pr, APPROVE) + + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(80, head_sha="sha-v", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={80: flaky}) + + first = policy.poll_and_review("gitea-saas-core") + assert len(first) == 1 and first[0]["pr"] == 80 + assert first[0]["outcome"] == "failed" + assert first[0]["error"] == {"type": "RuntimeError", + "message": "verdict generation boom"} + assert first[0]["attempts"] == 1 + assert gitea.merged == [] and gitea.closed == [] and gitea.comments == [] + + # Persisted: a failed attempt exists with its error, and the head is + # NOT completed — the same unchanged head stays eligible for retry. + rows = policy._reviews_conn.execute( + "SELECT attempts, last_error FROM review_state" + " WHERE repo = 'gitea-saas-core' AND pr = 80 AND head_sha = 'sha-v'").fetchall() + assert len(rows) == 1 and rows[0][0] == 1 + assert "verdict generation boom" in rows[0][1] + assert policy._already_reviewed("gitea-saas-core", 80, "sha-v") is False + + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "merged" + assert gitea.merged == [("gitea-saas-core", 80)] + assert gitea.closed == [("gitea-saas-core", 1)] + assert len(_ledger_debits(core, c["id"])) == 1 + # Completed now: an exact third poll skips the head. + assert policy.poll_and_review("gitea-saas-core") == [] + assert calls["n"] == 2 + + +def test_merge_boundary_failure_retried_merged_once(policy_factory): + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(81, head_sha="sha-m", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={81: make_verdict(81, APPROVE)}) + gitea.fail_next["merge_pr"] = RuntimeError("merge boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert first[0]["error"]["message"] == "merge boom" + assert gitea.merged == [] and gitea.closed == [] + + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "merged" + assert gitea.merged == [("gitea-saas-core", 81)] # merged exactly once + assert gitea.closed == [("gitea-saas-core", 1)] + assert len(_ledger_debits(core, c["id"])) == 1 + assert policy.poll_and_review("gitea-saas-core") == [] + + +def test_close_boundary_failure_no_double_merge_or_close(policy_factory): + # Failure AFTER the merge side effect: the retry must skip the already + # applied merge (checkpoint) and close the issue exactly once. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(82, head_sha="sha-c", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={82: make_verdict(82, APPROVE)}) + gitea.fail_next["close_issue"] = RuntimeError("close boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert first[0]["error"]["message"] == "close boom" + assert gitea.merged == [("gitea-saas-core", 82)] # merge already applied + + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "merged" + assert gitea.merged == [("gitea-saas-core", 82)] # NO second merge + assert gitea.closed == [("gitea-saas-core", 1)] # exactly one close + assert len(_ledger_debits(core, c["id"])) == 1 + assert policy.poll_and_review("gitea-saas-core") == [] + + +def test_debit_boundary_failure_never_double_debits(policy_factory): + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(83, head_sha="sha-d", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={83: make_verdict(83, APPROVE)}) + real_debit = core.debit + calls = {"n": 0} + + def flaky_debit(*args, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("debit boom") + return real_debit(*args, **kwargs) + + core.debit = flaky_debit + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert first[0]["error"]["message"] == "debit boom" + + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "merged" + assert gitea.merged == [("gitea-saas-core", 83)] # NO second merge + assert gitea.closed == [("gitea-saas-core", 1)] + assert len(_ledger_debits(core, c["id"])) == 1 # exactly one debit + assert calls["n"] == 2 + + +def test_label_boundary_failure_bounce_applied_once(policy_factory): + policy, gitea, reviewer, _, _ = policy_factory( + gitea_prs=[pr(84, head_sha="sha-l", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={84: make_verdict(84, BOUNCE)}) + gitea.fail_next["add_label"] = RuntimeError("label boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert first[0]["error"]["message"] == "label boom" + assert gitea.labels_added == [] + + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "bounced" + assert gitea.labels_added == [("gitea-saas-core", 84, "bounced")] + assert len(reviewer.posted) == 1 + assert policy.poll_and_review("gitea-saas-core") == [] + + +def test_comment_boundary_failure_retried(policy_factory): + policy, gitea, reviewer, _, _ = policy_factory( + gitea_prs=[pr(85, head_sha="sha-co", labels=("in-review",))], + gitea_issues=tier_issue("standard", number=1), + verdicts={85: make_verdict(85, APPROVE)}) + gitea.fail_next["post_comment"] = RuntimeError("comment boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert gitea.comments == [] + + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "approved_awaiting_merge" + assert any("awaiting operator merge" in body for _, _, body in gitea.comments) + assert policy.poll_and_review("gitea-saas-core") == [] + + +def test_remove_label_boundary_failure_retried(policy_factory): + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(86, head_sha="sha-rl", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={86: make_verdict(86, APPROVE)}) + gitea.fail_next["remove_label"] = RuntimeError("label boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert gitea.merged == [("gitea-saas-core", 86)] # merge already applied + + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "merged" + assert gitea.merged == [("gitea-saas-core", 86)] # NO second merge + assert gitea.closed == [("gitea-saas-core", 1)] # close exactly once + assert len(_ledger_debits(core, c["id"])) == 1 + + +def test_merge_refused_is_retryable_not_completed(policy_factory): + # A merge returning False is a non-completion, not a completed outcome: + # the head must stay eligible for retry until the action succeeds. + policy, gitea, reviewer, _, _ = policy_factory( + gitea_prs=[pr(87, head_sha="sha-mf", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={87: make_verdict(87, APPROVE)}) + gitea.merge_results[87] = False + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["action"] == "merge_failed" + assert first[0]["outcome"] == "failed" + assert first[0]["error"] == {"code": "merge_failed"} + assert gitea.closed == [] + assert policy._already_reviewed("gitea-saas-core", 87, "sha-mf") is False + + gitea.merge_results[87] = True + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "merged" + # The refused attempt had no side effect, so the retry re-attempted and + # succeeded. The fake records every merge ATTEMPT (like the API call), + # so two attempts are visible; the action itself completed exactly once. + # (The no-double-merge guarantee after a SUCCEEDED merge + later failure + # is asserted by the close/debit/remove-label boundary tests above.) + assert gitea.merged == [("gitea-saas-core", 87), ("gitea-saas-core", 87)] + assert gitea.closed == [("gitea-saas-core", 1)] + + +def test_webhook_after_completed_poll_does_not_double_close(policy_factory): + # The poll path already applied the outcome; the merge-event webhook + # redelivery must replay the debit (idempotency key) but not close the + # linked issue a second time. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(88, head_sha="sha-wh", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={88: make_verdict(88, APPROVE)}) + + policy.poll_and_review("gitea-saas-core") + assert gitea.closed == [("gitea-saas-core", 1)] + + ok, details = policy.handle_merge_event("gitea-saas-core", 88) + assert ok and details["debit"]["debit"] == "recorded" + assert details["debit"].get("replayed") is True + assert gitea.closed == [("gitea-saas-core", 1)] # closed exactly once + assert len(_ledger_debits(core, c["id"])) == 1 # debited exactly once + + +def test_legacy_review_row_still_skips_head(policy_factory): + # Rows written under the pre-#29 schema (no retry metadata) keep their + # completed meaning: the head is skipped and never re-reviewed. + policy, gitea, reviewer, _, _ = policy_factory( + gitea_prs=[pr(89, head_sha="sha-legacy", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={89: make_verdict(89, APPROVE)}) + policy._reviews_conn.execute( + "INSERT INTO reviews (repo, pr, head_sha, verdict, reviewed_at)" + " VALUES (?, ?, ?, ?, ?)", + ("gitea-saas-core", 89, "sha-legacy", APPROVE, "2026-01-01T00:00:00+00:00")) + policy._reviews_conn.commit() + + assert policy.poll_and_review("gitea-saas-core") == [] + assert reviewer.calls == [] and gitea.merged == [] + + +def test_failed_attempt_then_head_move_reviews_new_head(policy_factory): + policy, gitea, reviewer, _, _ = policy_factory( + gitea_prs=[pr(90, head_sha="sha-a", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={90: make_verdict(90, APPROVE)}) + gitea.fail_next["merge_pr"] = RuntimeError("merge boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + + gitea.prs[0]["head_sha"] = "sha-b" # head moved before the retry + second = policy.poll_and_review("gitea-saas-core") + assert second[0]["action"] == "merged" + assert policy._already_reviewed("gitea-saas-core", 90, "sha-a") is False + assert policy._already_reviewed("gitea-saas-core", 90, "sha-b") is True + + +# --------------------------------------------------------------------------- +# Crash-window recovery independent of the open-PR scan (issue #29 review +# bounce): once a merge truly succeeds the PR is no longer returned by +# list_open_prs(), so the persisted retry row must be finished from the +# review_state table alone, and replay must never duplicate an +# externally-visible side effect (label removal, close, success comment). +# --------------------------------------------------------------------------- + +def test_merged_pr_gone_from_open_scan_recovered_after_post_merge_failure(policy_factory): + # The reviewer's exact scenario: merge succeeds, a post-merge step + # (label removal) fails, and the merged PR disappears from the open-PR + # scan before the next retry. The persisted row must still be finished + # — exactly one merge, one debit, one label removal, one close, one + # success comment. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(91, head_sha="sha-91", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={91: make_verdict(91, APPROVE)}) + gitea.fail_next["remove_label"] = RuntimeError("label boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert first[0]["error"]["message"] == "label boom" + assert gitea.merged == [("gitea-saas-core", 91)] # merge applied + # The merged PR is gone from list_open_prs() — exactly what real Gitea + # does once a merge succeeds — so the open scan can never reach it. + gitea.prs = [p for p in gitea.prs if p["number"] != 91] + assert gitea.list_open_prs() == [] + + second = policy.poll_and_review("gitea-saas-core") + assert second and second[0]["action"] == "merged" + assert second[0].get("recovered") is True # via recovery + assert gitea.merged == [("gitea-saas-core", 91)] # NO second merge + assert gitea.labels_removed.count(("gitea-saas-core", 91, "in-review")) == 1 + assert gitea.closed == [("gitea-saas-core", 1)] # closed exactly once + comments = [b for _, _, b in gitea.comments] + assert sum("Merged by review-gate" in b for b in comments) == 1 + assert len(_ledger_debits(core, c["id"])) == 1 # debited exactly once + assert policy._already_reviewed("gitea-saas-core", 91, "sha-91") is True + assert policy.poll_and_review("gitea-saas-core") == [] # completed: skipped + + +def test_crash_after_label_removed_no_duplicate_visible_actions(policy_factory): + # remove_label succeeded, then a later step (the success comment) + # failed: the retry must not remove the label again, close the issue + # again, or re-post the comment — even though the PR is gone from the + # open scan. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(92, head_sha="sha-92", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={92: make_verdict(92, APPROVE)}) + gitea.fail_next["post_comment"] = RuntimeError("comment boom") + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert first[0]["error"]["message"] == "comment boom" + # merge, debit, label removal and close all succeeded before the crash + assert gitea.merged == [("gitea-saas-core", 92)] + assert gitea.labels_removed.count(("gitea-saas-core", 92, "in-review")) == 1 + assert gitea.closed == [("gitea-saas-core", 1)] + + gitea.prs = [p for p in gitea.prs if p["number"] != 92] + second = policy.poll_and_review("gitea-saas-core") + assert second[0].get("recovered") is True + comments = [b for _, _, b in gitea.comments] + assert sum("Merged by review-gate" in b for b in comments) == 1 # once + assert gitea.labels_removed.count(("gitea-saas-core", 92, "in-review")) == 1 + assert gitea.closed == [("gitea-saas-core", 1)] + assert len(_ledger_debits(core, c["id"])) == 1 + assert policy.poll_and_review("gitea-saas-core") == [] + + +def test_crash_after_success_comment_before_completion_record(policy_factory): + # The success comment posted, then the process crashed before the + # completion row was written: replay must finish the head WITHOUT + # re-posting the comment or repeating any other side effect. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(93, head_sha="sha-93", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={93: make_verdict(93, APPROVE)}) + + real_record = policy._record_completed + + def crash_after_comment(*args, **kwargs): + raise RuntimeError("process killed after success comment") + + policy._record_completed = crash_after_comment + with pytest.raises(RuntimeError): # the simulated crash + policy.poll_and_review("gitea-saas-core") + policy._record_completed = real_record + + comments = [b for _, _, b in gitea.comments] + assert sum("Merged by review-gate" in b for b in comments) == 1 + assert gitea.merged == [("gitea-saas-core", 93)] + assert gitea.closed == [("gitea-saas-core", 1)] + assert gitea.labels_removed.count(("gitea-saas-core", 93, "in-review")) == 1 + + gitea.prs = [p for p in gitea.prs if p["number"] != 93] + second = policy.poll_and_review("gitea-saas-core") + assert second and second[0].get("recovered") is True + comments = [b for _, _, b in gitea.comments] + assert sum("Merged by review-gate" in b for b in comments) == 1 # not duplicated + assert gitea.labels_removed.count(("gitea-saas-core", 93, "in-review")) == 1 + assert gitea.closed == [("gitea-saas-core", 1)] + assert len(_ledger_debits(core, c["id"])) == 1 + assert policy.poll_and_review("gitea-saas-core") == [] + + +# --------------------------------------------------------------------------- +# Governor review #2 regressions: +# 1. the merge is pinned to the REVIEWED head (expected_head_sha) — a branch +# that moved after review is refused before any merge POST; +# 2. recovery distinguishes a merged PR from a merely closed one using the +# PRODUCTION normalized shape (positive merged is True only); +# 3. a closed-but-unmerged PR is never adopted as merged (no debit, no label +# removal, no issue close, no success comment). +# --------------------------------------------------------------------------- + +def test_reviewed_head_pinned_so_moved_branch_is_never_merged(policy_factory): + # Integration regression: review scans SHA A, the branch moves to SHA B + # before the merge attempt. The policy must hand the REVIEWED head to + # merge_pr, and the production-shaped client seam fails closed: no + # merge POST, no debit, no close, no success comment. The moved head is + # then a NEW head, re-reviewed and merged normally on the next poll. + def review_then_move_head(repo, pr): + gitea.prs[0]["head_sha"] = "sha-B" # branch moved after review + return make_verdict(pr, APPROVE) + + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(94, head_sha="sha-A", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={94: review_then_move_head}) + + first = policy.poll_and_review("gitea-saas-core") + assert first[0]["outcome"] == "failed" + assert "refusing to merge" in first[0]["error"]["message"] + assert "sha-A" in first[0]["error"]["message"] and "sha-B" in first[0]["error"]["message"] + assert gitea.merged == [] # NO merge POST happened + assert gitea.closed == [] + assert not any("Merged by review-gate" in b for _, _, b in gitea.comments) + assert _ledger_debits(core, c["id"]) == [] # no billing for the refused head + + second = policy.poll_and_review("gitea-saas-core") + assert second and second[0]["action"] == "merged" # sha-B reviewed + merged + assert gitea.merged == [("gitea-saas-core", 94)] # exactly one merge, ever + assert gitea.closed == [("gitea-saas-core", 1)] + assert len(_ledger_debits(core, c["id"])) == 1 + assert policy._already_reviewed("gitea-saas-core", 94, "sha-A") is False + assert policy._already_reviewed("gitea-saas-core", 94, "sha-B") is True + + +def test_crash_window_adopts_merged_pr_from_production_shape(policy_factory): + # The merge applied server-side but the process died before the + # merged_step checkpoint write: with the production normalized shape + # (merged/state now included), recovery positively adopts the merged PR + # and finishes debit/label/close/comment — without ever calling + # merge_pr again. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(96, head_sha="sha-96", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={}) + policy._ensure_plan("gitea-saas-core", 96, "sha-96", APPROVE, "gold", 1) + # The live PR record already reports the server-side merge (raw Gitea + # record normalized by the production client); the checkpoint is the + # only thing missing. + gitea.prs[0]["merged"] = True + gitea.prs[0]["state"] = "closed" + + results = policy.poll_and_review("gitea-saas-core") + assert results and results[0]["action"] == "merged" + assert results[0].get("recovered") is True + assert gitea.merged == [] # crash window: no merge request needed + assert gitea.labels_removed.count(("gitea-saas-core", 96, "in-review")) == 1 + assert gitea.closed == [("gitea-saas-core", 1)] + assert any("Merged by review-gate" in b for _, _, b in gitea.comments) + assert len(_ledger_debits(core, c["id"])) == 1 + assert policy._already_reviewed("gitea-saas-core", 96, "sha-96") is True + assert policy.poll_and_review("gitea-saas-core") == [] # completed: skipped + + +def test_closed_unmerged_pr_recovery_never_debits_or_closes(policy_factory): + # Negative regression: state == "closed" is NOT proof of a merge. The + # outcome plan exists for the head, the merge never happened, and the + # PR is closed without merging — recovery must not adopt it as merged: + # no debit, no label removal, no issue close, no success comment. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(95, head_sha="sha-95", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={}) + policy._ensure_plan("gitea-saas-core", 95, "sha-95", APPROVE, "gold", 1) + gitea.prs[0]["state"] = "closed" # closed WITHOUT merging + gitea.prs[0]["merged"] = False + + results = policy.poll_and_review("gitea-saas-core") + assert results == [] # nothing adopted, nothing applied + assert gitea.merged == [] and gitea.closed == [] + assert gitea.labels_removed == [] + assert gitea.comments == [] + assert _ledger_debits(core, c["id"]) == [] + assert policy._step_done("gitea-saas-core", 95, "sha-95", "merged") is False + assert policy._already_reviewed("gitea-saas-core", 95, "sha-95") is False + + +# --------------------------------------------------------------------------- +# Governor review #3 (10539) regressions — head-integrity gates: +# 1. recovery adopts a merged PR ONLY when the live merged head equals the +# persisted planned/reviewed head; a DIFFERENT merged head fails closed +# with a structured HEAD_MISMATCH and no debit/label/close/comment; +# 2. a candidate with NO reviewed head_sha is rejected BEFORE the merge +# client is called (expected_head_sha=None would arm the client's +# live-head fallback and could merge an unreviewed head). +# --------------------------------------------------------------------------- + +def test_recovery_merged_different_head_fails_closed(policy_factory): + # Negative regression: the plan was persisted for reviewed head sha-A, + # but the branch moved and head sha-B was merged externally before the + # merged_step checkpoint write. Recovery must NOT attribute the sha-B + # merge to the sha-A plan: zero side effects (no debit, no label + # removal, no issue close, no success comment), a structured + # HEAD_MISMATCH result, and the row left unadopted for the operator. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(97, head_sha="sha-A", labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={}) + policy._ensure_plan("gitea-saas-core", 97, "sha-A", APPROVE, "gold", 1) + gitea.prs[0]["head_sha"] = "sha-B" # branch moved AND merged externally + gitea.prs[0]["merged"] = True + gitea.prs[0]["state"] = "closed" + + results = policy.poll_and_review("gitea-saas-core") + assert results and results[0]["outcome"] == "failed" + assert results[0]["error"]["code"] == "HEAD_MISMATCH" + assert "sha-A" in results[0]["error"]["message"] + assert "sha-B" in results[0]["error"]["message"] + assert gitea.merged == [] and gitea.closed == [] + assert gitea.labels_removed == [] + assert gitea.comments == [] + assert _ledger_debits(core, c["id"]) == [] + assert policy._step_done("gitea-saas-core", 97, "sha-A", "merged") is False + assert policy._already_reviewed("gitea-saas-core", 97, "sha-A") is False + + +def test_missing_reviewed_head_never_reaches_merge_client(policy_factory): + # Negative regression: a candidate PR with no head_sha must be refused + # BEFORE the merge client is called — passing expected_head_sha=None + # would arm the production client's live-head fallback and could merge + # a head the policy never reviewed. Assert zero merge attempts and zero + # billing/label/close/comment side effects, with a structured failure. + policy, gitea, reviewer, core, c = policy_factory( + gitea_prs=[pr(98, labels=("in-review",))], + gitea_issues=tier_issue("gold", number=1), + verdicts={98: make_verdict(98, APPROVE)}) + gitea.prs[0]["head_sha"] = None # no reviewed head on the candidate + + results = policy.poll_and_review("gitea-saas-core") + assert results and results[0]["outcome"] == "failed" + assert "no reviewed head_sha" in results[0]["error"]["message"] + assert "fall back to the live head" in results[0]["error"]["message"] + assert gitea.merged == [] and gitea.closed == [] + assert gitea.labels_removed == [] + assert gitea.comments == [] + assert _ledger_debits(core, c["id"]) == [] + assert policy._already_reviewed("gitea-saas-core", 98, "unknown") is False