gitea-saas-core#29: transactional + retryable review outcome application #32
@ -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 []
|
||||
|
||||
@ -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
|
||||
``<repo>#<pr>@<head_sha>``. 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,
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user